diff --git a/lib/orchpy/orchpy/services.py b/lib/orchpy/orchpy/services.py index 17cb368ca..c71d326f0 100644 --- a/lib/orchpy/orchpy/services.py +++ b/lib/orchpy/orchpy/services.py @@ -18,15 +18,21 @@ def cleanup(): p_sec += 1 if p_sec >= timeout_sec: p.kill() # supported from python 2.6 - print 'helper processes shut down!' + print "helper processes shut down!" all_processes = [] atexit.register(cleanup) -def start_scheduler(scheduler_address): +def start_scheduler(host, port): + scheduler_address = host + ":" + str(port) p = subprocess.Popen([os.path.join(_services_path, "scheduler"), str(scheduler_address)]) all_processes.append(p) -def start_objstore(objstore_address): +def start_objstore(host, port): + objstore_address = host + ":" + str(port) p = subprocess.Popen([os.path.join(_services_path, "objstore"), str(objstore_address)]) all_processes.append(p) + +def start_worker(test_path, host, scheduler_port, worker_port, objstore_port): + p = subprocess.Popen(["python", os.path.join(test_path, "testrecv.py"), host, str(scheduler_port), str(worker_port), str(objstore_port)]) + all_processes.append(p) diff --git a/test/runtest.py b/test/runtest.py index 56cf0fab0..e36653a07 100644 --- a/test/runtest.py +++ b/test/runtest.py @@ -13,31 +13,7 @@ from grpc.beta import implementations import orchestra_pb2 import types_pb2 -""" -class UnisonTest(unittest.TestCase): - - def testSerialize(self): - d = [1, 2L, "hello", 3.0] - res = unison.serialize_args(d) - c = unison.deserialize_args(res) - self.assertEqual(c, d) - - d = [{'hello': 'world'}] - res = unison.serialize_args(d) - c = unison.deserialize_args(res) - self.assertEqual(c, d) - - a = np.zeros((100, 100)) - res = unison.serialize_args(a) - b = unison.deserialize_args(res) - self.assertTrue((a == b).all()) - - a = [unison.ObjRef(42, int)] - res = unison.serialize_args(a) - b = unison.deserialize_args(res) - self.assertEqual(a, b) -""" - +IP_ADDRESS = "127.0.0.1" TIMEOUT_SECONDS = 5 def produce_data(num_chunks): @@ -52,30 +28,58 @@ def connect_to_objstore(host, port): channel = implementations.insecure_channel(host, port) return orchestra_pb2.beta_create_ObjStore_stub(channel) +def address(host, port): + return host + ":" + str(port) + +scheduler_port_counter = 0 +def new_scheduler_port(): + global scheduler_port_counter + scheduler_port_counter += 1 + return 10000 + scheduler_port_counter + +worker_port_counter = 0 +def new_worker_port(): + global worker_port_counter + worker_port_counter += 1 + return 40000 + worker_port_counter + +objstore_port_counter = 0 +def new_objstore_port(): + global objstore_port_counter + objstore_port_counter += 1 + return 20000 + objstore_port_counter + + class ObjStoreTest(unittest.TestCase): """Test setting up object stores, transfering data between them and retrieving data to a client""" def testObjStore(self): - services.start_scheduler("0.0.0.0:22221") - services.start_objstore("0.0.0.0:22222") - services.start_objstore("0.0.0.0:22223") + scheduler_port = new_scheduler_port() + objstore1_port = new_objstore_port() + objstore2_port = new_objstore_port() + worker1_port = new_worker_port() + worker2_port = new_worker_port() + + services.start_scheduler(IP_ADDRESS, scheduler_port) + services.start_objstore(IP_ADDRESS, objstore1_port) + services.start_objstore(IP_ADDRESS, objstore2_port) time.sleep(0.2) - scheduler_stub = connect_to_scheduler('localhost', 22221) - objstore1_stub = connect_to_objstore('localhost', 22222) - objstore2_stub = connect_to_objstore('localhost', 22223) + scheduler_stub = connect_to_scheduler(IP_ADDRESS, scheduler_port) + objstore1_stub = connect_to_objstore(IP_ADDRESS, objstore1_port) + objstore2_stub = connect_to_objstore(IP_ADDRESS, objstore2_port) worker1 = worker.Worker() - worker1.connect("127.0.0.1:22221", "127.0.0.1:40000", "127.0.0.1:22222") + worker1.connect(address(IP_ADDRESS, scheduler_port), address(IP_ADDRESS, worker1_port), address(IP_ADDRESS, objstore1_port)) worker2 = worker.Worker() - worker2.connect("127.0.0.1:22221", "127.0.0.1:40001", "127.0.0.1:22223") + worker2.connect(address(IP_ADDRESS, scheduler_port), address(IP_ADDRESS, worker2_port), address(IP_ADDRESS, objstore2_port)) for i in range(1, 100): l = i * 100 * "h" objref = worker1.push(l) - response = objstore1_stub.DeliverObj(orchestra_pb2.DeliverObjRequest(objref=objref.get_id(), objstore_address="0.0.0.0:22223"), TIMEOUT_SECONDS) + response = objstore1_stub.DeliverObj(orchestra_pb2.DeliverObjRequest(objref=objref.get_id(), objstore_address=address(IP_ADDRESS, objstore2_port)), TIMEOUT_SECONDS) s = worker2.get_serialized(objref.get_id()) result = worker.unison.deserialize_from_string(s) # result = worker1.pull(objref) @@ -86,31 +90,37 @@ class ObjStoreTest(unittest.TestCase): class SchedulerTest(unittest.TestCase): def testCall(self): - services.start_scheduler("0.0.0.0:22221") - services.start_objstore("0.0.0.0:22222") + scheduler_port = new_scheduler_port() + objstore_port = new_objstore_port() + worker1_port = new_worker_port() + worker2_port = new_worker_port() + worker3_port = new_worker_port() + + services.start_scheduler(IP_ADDRESS, scheduler_port) + services.start_objstore(IP_ADDRESS, objstore_port) time.sleep(0.2) - scheduler_stub = connect_to_scheduler('localhost', 22221) - objstore_stub = connect_to_objstore('localhost', 22222) + scheduler_stub = connect_to_scheduler(IP_ADDRESS, scheduler_port) + objstore_stub = connect_to_objstore(IP_ADDRESS, objstore_port) time.sleep(0.2) - w = worker.Worker() - w.connect("127.0.0.1:22221", "127.0.0.1:40003", "127.0.0.1:22222") - w.start_worker_service() - w2 = worker.Worker() - w2.connect("127.0.0.1:22221", "127.0.0.1:40004", "127.0.0.1:22222") - w2.start_worker_service() + worker1 = worker.Worker() + worker1.connect(address(IP_ADDRESS, scheduler_port), address(IP_ADDRESS, worker1_port), address(IP_ADDRESS, objstore_port)) + worker1.start_worker_service() + worker2 = worker.Worker() + worker2.connect(address(IP_ADDRESS, scheduler_port), address(IP_ADDRESS, worker2_port), address(IP_ADDRESS, objstore_port)) + worker2.start_worker_service() time.sleep(0.2) - w.register_function("hello_world", None, 2) - w2.register_function("hello_world", None, 2) + worker1.register_function("hello_world", None, 2) + worker2.register_function("hello_world", None, 2) time.sleep(0.1) - w.call("hello_world", ["hi"]) + worker1.call("hello_world", ["hi"]) time.sleep(0.1) @@ -119,29 +129,17 @@ class SchedulerTest(unittest.TestCase): # self.assertEqual(reply.task[0].name, u'hello_world') # doesn't currently work, because scheduler is now invoked on every interaction test_path = os.path.dirname(os.path.abspath(__file__)) - - p = subprocess.Popen(["python", os.path.join(test_path, "testrecv.py")]) + services.start_worker(test_path, IP_ADDRESS, scheduler_port, worker3_port, objstore_port) time.sleep(0.2) - w.call("hello_world", ["hi"]) - w.call("hello_world", ["hi"]) - w.call("hello_world", ["hi"]) + worker1.call("hello_world", ["hi"]) + worker1.call("hello_world", ["hi"]) + worker1.call("hello_world", ["hi"]) scheduler_stub.PushObj(orchestra_pb2.PushObjRequest(workerid=0), TIMEOUT_SECONDS) # self.assertEqual(p.wait(), 0, "argument was not received by the test program") # todo: reactivate - # w.main_loop() - # w2.main_loop() - # - # reply = scheduler_stub.GetDebugInfo(orchestra_pb2.GetDebugInfoRequest(do_scheduling=True), TIMEOUT_SECONDS) - # time.sleep(0.1) - # reply = scheduler_stub.GetDebugInfo(orchestra_pb2.GetDebugInfoRequest(), TIMEOUT_SECONDS) - # - # self.assertEqual(list(reply.task), []) - # - # services.cleanup() - if __name__ == '__main__': unittest.main() diff --git a/test/testprogram.py b/test/testprogram.py index 5d35e0f16..c97953c60 100644 --- a/test/testprogram.py +++ b/test/testprogram.py @@ -1,3 +1,5 @@ +import sys + import orchpy.unison as unison import orchpy.services as services import orchpy.worker as worker @@ -11,8 +13,16 @@ def print_string(string): def handle_int(a, b): return a+1, b+1 +def address(host, port): + return host + ":" + str(port) + if __name__ == '__main__': - worker.global_worker.connect("127.0.0.1:22221", "127.0.0.1:40000", "127.0.0.1:22222") + ip_address = sys.argv[1] + scheduler_port = sys.argv[2] + worker_port = sys.argv[3] + objstore_port = sys.argv[4] + + worker.global_worker.connect(address(ip_address, scheduler_port), address(ip_address, worker_port), address(ip_address, objstore_port)) worker.global_worker.register_function("hello_world", print_string, 1) worker.global_worker.register_function("handle_int", handle_int, 2) diff --git a/test/testrecv.py b/test/testrecv.py index 68295916e..e58984ea5 100644 --- a/test/testrecv.py +++ b/test/testrecv.py @@ -1,11 +1,20 @@ +import sys + import orchpy.unison as unison import orchpy.services as services import orchpy.worker as worker -import sys if __name__ == '__main__': + ip_address = sys.argv[1] + scheduler_port = sys.argv[2] + worker_port = sys.argv[3] + objstore_port = sys.argv[4] + + def address(host, port): + return host + ":" + str(port) + worker = worker.Worker() - worker.connect("127.0.0.1:22221", "127.0.0.1:50000", "127.0.0.1:22222") + worker.connect(address(ip_address, scheduler_port), address(ip_address, worker_port), address(ip_address, objstore_port)) worker.start_worker_service() worker.register_function("hello_world", None, 0) name, args, returnref = worker.wait_for_next_task()