updating addresses and adding tests

lib/orchpy/orchpy/services.py

@@ -9,42 +9,39 @@ all_processes = []
 
 def cleanup():
   global all_processes
-  for p, port in all_processes:
+  for p, address in all_processes:
     if p.poll() is not None: # process has already terminated
-      print "Process at port " + str(port) + " has already terminated."
+      print "Process at address " + address + " has already terminated."
       continue
-    print "Attempting to kill process at port " + str(port) + "."
+    print "Attempting to kill process at address " + address + "."
     p.kill()
     time.sleep(0.05) # is this necessary?
     if p.poll() is not None:
-      print "Successfully killed process at port " + str(port) + "."
+      print "Successfully killed process at address " + address + "."
       continue
-    print "Kill attempt failed, attempting to terminate process at port " + str(port) + "."
+    print "Kill attempt failed, attempting to terminate process at address " + address + "."
     p.terminate()
     time.sleep(0.05) # is this necessary?
     if p.poll is not None:
-      print "Successfully terminated process at port " + str(port) + "."
+      print "Successfully terminated process at address " + address + "."
       continue
     print "Termination attempt failed, giving up."
   all_processes = []
 
 atexit.register(cleanup)
 
-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, port))
+def start_scheduler(scheduler_address):
+  p = subprocess.Popen([os.path.join(_services_path, "scheduler"), scheduler_address])
+  all_processes.append((p, scheduler_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, port))
+def start_objstore(objstore_address):
+  p = subprocess.Popen([os.path.join(_services_path, "objstore"), objstore_address])
+  all_processes.append((p, objstore_address))
 
-def start_worker(test_path, host, scheduler_port, worker_port, objstore_port):
+def start_worker(test_path, scheduler_address, objstore_address, worker_address):
   p = subprocess.Popen(["python",
                         test_path,
-                        "--ip_address=" + host,
-                        "--scheduler_port=" + str(scheduler_port),
-                        "--objstore_port=" + str(objstore_port),
-                        "--worker_port=" + str(worker_port)])
-  all_processes.append((p, worker_port))
+                        "--scheduler-address=" + scheduler_address,
+                        "--objstore-address=" + objstore_address,
+                        "--worker-address=" + worker_address])
+  all_processes.append((p, worker_address))

lib/orchpy/orchpy/worker.py

@@ -83,24 +83,24 @@ def distributed(arg_types, return_types, worker=global_worker):
 def get_arguments_for_execution(function, args, worker=global_worker):
   arguments = []
   # check the number of args
-  if len(args) != len(function.types) and function.types[-1] is not None:
-    raise Exception("Function {} expects {} arguments, but received {}.".format(function.__name__, len(function.types), len(args)))
-  elif len(args) < len(function.types) - 1 and function.types[-1] is None:
-    raise Exception("Function {} expects at least {} arguments, but received {}.".format(function.__name__, len(function.types) - 1, len(args)))
+  if len(args) != len(function.arg_types) and function.arg_types[-1] is not None:
+    raise Exception("Function {} expects {} arguments, but received {}.".format(function.__name__, len(function.arg_types), len(args)))
+  elif len(args) < len(function.arg_types) - 1 and function.arg_types[-1] is None:
+    raise Exception("Function {} expects at least {} arguments, but received {}.".format(function.__name__, len(function.arg_types) - 1, len(args)))
 
   for (i, arg) in enumerate(args):
     print "Pulling argument {} for function {}.".format(i, function.__name__)
-    if i < len(function.types) - 1:
-      expected_type = function.types[i]
-    elif i == len(function.types) - 1 and function.types[-1] is not None:
-      expected_type = function.types[-1]
-    elif function.types[-1] is None and len(function.types > 1):
-      expected_type = function.types[-2]
+    if i < len(function.arg_types) - 1:
+      expected_type = function.arg_types[i]
+    elif i == len(function.arg_types) - 1 and function.arg_types[-1] is not None:
+      expected_type = function.arg_types[-1]
+    elif function.arg_types[-1] is None and len(function.arg_types > 1):
+      expected_type = function.arg_types[-2]
     else:
       assert False, "This code should be unreachable."
 
-    argument = worker.get_object(arg) if type(arg) == orchpy.ObjRef else arg
-    if type(arg) == orchpy.ObjRef:
+    argument = worker.get_object(arg) if type(arg) == orchpy.lib.ObjRef else arg
+    if type(arg) == orchpy.lib.ObjRef:
       # get the object from the local object store
       # TODO(rkn): Do we know that it is already there? Maybe we should call pull(arg, worker).
       argument = worker.get_object(arg)