distributed -> remote (#82)

lib/python/arrays/dist/core.py

@@ -69,12 +69,12 @@ class DistArray(object):
     a = self.assemble()
     return a[sliced]
 
-@halo.distributed([DistArray], [np.ndarray])
+@halo.remote([DistArray], [np.ndarray])
 def assemble(a):
   return a.assemble()
 
 # TODO(rkn): what should we call this method
-@halo.distributed([np.ndarray], [DistArray])
+@halo.remote([np.ndarray], [DistArray])
 def numpy_to_dist(a):
   result = DistArray(a.shape)
   for index in np.ndindex(*result.num_blocks):
@@ -83,28 +83,28 @@ def numpy_to_dist(a):
     result.objrefs[index] = halo.push(a[[slice(l, u) for (l, u) in zip(lower, upper)]])
   return result
 
-@halo.distributed([List[int], str], [DistArray])
+@halo.remote([List[int], str], [DistArray])
 def zeros(shape, dtype_name="float"):
   result = DistArray(shape)
   for index in np.ndindex(*result.num_blocks):
     result.objrefs[index] = single.zeros(DistArray.compute_block_shape(index, shape), dtype_name=dtype_name)
   return result
 
-@halo.distributed([List[int], str], [DistArray])
+@halo.remote([List[int], str], [DistArray])
 def ones(shape, dtype_name="float"):
   result = DistArray(shape)
   for index in np.ndindex(*result.num_blocks):
     result.objrefs[index] = single.ones(DistArray.compute_block_shape(index, shape), dtype_name=dtype_name)
   return result
 
-@halo.distributed([DistArray], [DistArray])
+@halo.remote([DistArray], [DistArray])
 def copy(a):
   result = DistArray(a.shape)
   for index in np.ndindex(*result.num_blocks):
     result.objrefs[index] = a.objrefs[index] # We don't need to actually copy the objects because cluster-level objects are assumed to be immutable.
   return result
 
-@halo.distributed([int, int, str], [DistArray])
+@halo.remote([int, int, str], [DistArray])
 def eye(dim1, dim2=-1, dtype_name="float"):
   dim2 = dim1 if dim2 == -1 else dim2
   shape = [dim1, dim2]
@@ -117,7 +117,7 @@ def eye(dim1, dim2=-1, dtype_name="float"):
       result.objrefs[i, j] = single.zeros(block_shape, dtype_name=dtype_name)
   return result
 
-@halo.distributed([DistArray], [DistArray])
+@halo.remote([DistArray], [DistArray])
 def triu(a):
   if a.ndim != 2:
     raise Exception("Input must have 2 dimensions, but a.ndim is " + str(a.ndim))
@@ -131,7 +131,7 @@ def triu(a):
       result.objrefs[i, j] = single.zeros_like(a.objrefs[i, j])
   return result
 
-@halo.distributed([DistArray], [DistArray])
+@halo.remote([DistArray], [DistArray])
 def tril(a):
   if a.ndim != 2:
     raise Exception("Input must have 2 dimensions, but a.ndim is " + str(a.ndim))
@@ -145,7 +145,7 @@ def tril(a):
       result.objrefs[i, j] = single.zeros_like(a.objrefs[i, j])
   return result
 
-@halo.distributed([np.ndarray, None], [np.ndarray])
+@halo.remote([np.ndarray, None], [np.ndarray])
 def blockwise_dot(*matrices):
   n = len(matrices)
   if n % 2 != 0:
@@ -156,7 +156,7 @@ def blockwise_dot(*matrices):
     result += np.dot(matrices[i], matrices[n / 2 + i])
   return result
 
-@halo.distributed([DistArray, DistArray], [DistArray])
+@halo.remote([DistArray, DistArray], [DistArray])
 def dot(a, b):
   if a.ndim != 2:
     raise Exception("dot expects its arguments to be 2-dimensional, but a.ndim = {}.".format(a.ndim))
@@ -172,7 +172,7 @@ def dot(a, b):
   return result
 
 # This is not in numpy, should we expose this?
-@halo.distributed([DistArray, List[int], None], [DistArray])
+@halo.remote([DistArray, List[int], None], [DistArray])
 def subblocks(a, *ranges):
   """
   This function produces a distributed array from a subset of the blocks in the `a`. The result and `a` will have the same number of dimensions.For example,
@@ -203,7 +203,7 @@ def subblocks(a, *ranges):
     result.objrefs[index] = a.objrefs[tuple([ranges[i][index[i]] for i in range(a.ndim)])]
   return result
 
-@halo.distributed([DistArray], [DistArray])
+@halo.remote([DistArray], [DistArray])
 def transpose(a):
   if a.ndim != 2:
     raise Exception("transpose expects its argument to be 2-dimensional, but a.ndim = {}, a.shape = {}.".format(a.ndim, a.shape))
@@ -214,7 +214,7 @@ def transpose(a):
   return result
 
 # TODO(rkn): support broadcasting?
-@halo.distributed([DistArray, DistArray], [DistArray])
+@halo.remote([DistArray, DistArray], [DistArray])
 def add(x1, x2):
   if x1.shape != x2.shape:
     raise Exception("add expects arguments `x1` and `x2` to have the same shape, but x1.shape = {}, and x2.shape = {}.".format(x1.shape, x2.shape))
@@ -224,7 +224,7 @@ def add(x1, x2):
   return result
 
 # TODO(rkn): support broadcasting?
-@halo.distributed([DistArray, DistArray], [DistArray])
+@halo.remote([DistArray, DistArray], [DistArray])
 def subtract(x1, x2):
   if x1.shape != x2.shape:
     raise Exception("subtract expects arguments `x1` and `x2` to have the same shape, but x1.shape = {}, and x2.shape = {}.".format(x1.shape, x2.shape))

lib/python/arrays/dist/linalg.py

@@ -8,7 +8,7 @@ from core import *
 
 __all__ = ["tsqr", "modified_lu", "tsqr_hr", "qr"]
 
-@halo.distributed([DistArray], [DistArray, np.ndarray])
+@halo.remote([DistArray], [DistArray, np.ndarray])
 def tsqr(a):
   """
   arguments:
@@ -80,7 +80,7 @@ def tsqr(a):
   return q_result, r
 
 # TODO(rkn): This is unoptimized, we really want a block version of this.
-@halo.distributed([DistArray], [DistArray, np.ndarray, np.ndarray])
+@halo.remote([DistArray], [DistArray, np.ndarray, np.ndarray])
 def modified_lu(q):
   """
   Algorithm 5 from http://www.eecs.berkeley.edu/Pubs/TechRpts/2013/EECS-2013-175.pdf
@@ -110,19 +110,19 @@ def modified_lu(q):
   U = np.triu(q_work)[:b, :]
   return numpy_to_dist(halo.push(L)), U, S # TODO(rkn): get rid of push and pull
 
-@halo.distributed([np.ndarray, np.ndarray, np.ndarray, int], [np.ndarray, np.ndarray])
+@halo.remote([np.ndarray, np.ndarray, np.ndarray, int], [np.ndarray, np.ndarray])
 def tsqr_hr_helper1(u, s, y_top_block, b):
   y_top = y_top_block[:b, :b]
   s_full = np.diag(s)
   t = -1 * np.dot(u, np.dot(s_full, np.linalg.inv(y_top).T))
   return t, y_top
 
-@halo.distributed([np.ndarray, np.ndarray], [np.ndarray])
+@halo.remote([np.ndarray, np.ndarray], [np.ndarray])
 def tsqr_hr_helper2(s, r_temp):
   s_full = np.diag(s)
   return np.dot(s_full, r_temp)
 
-@halo.distributed([DistArray], [DistArray, np.ndarray, np.ndarray, np.ndarray])
+@halo.remote([DistArray], [DistArray, np.ndarray, np.ndarray, np.ndarray])
 def tsqr_hr(a):
   """Algorithm 6 from http://www.eecs.berkeley.edu/Pubs/TechRpts/2013/EECS-2013-175.pdf"""
   q, r_temp = tsqr(a)
@@ -132,15 +132,15 @@ def tsqr_hr(a):
   r = tsqr_hr_helper2(s, r_temp)
   return y, t, y_top, r
 
-@halo.distributed([np.ndarray, np.ndarray, np.ndarray, np.ndarray], [np.ndarray])
+@halo.remote([np.ndarray, np.ndarray, np.ndarray, np.ndarray], [np.ndarray])
 def qr_helper1(a_rc, y_ri, t, W_c):
   return a_rc - np.dot(y_ri, np.dot(t.T, W_c))
 
-@halo.distributed([np.ndarray, np.ndarray], [np.ndarray])
+@halo.remote([np.ndarray, np.ndarray], [np.ndarray])
 def qr_helper2(y_ri, a_rc):
   return np.dot(y_ri.T, a_rc)
 
-@halo.distributed([DistArray], [DistArray, DistArray])
+@halo.remote([DistArray], [DistArray, DistArray])
 def qr(a):
   """Algorithm 7 from http://www.eecs.berkeley.edu/Pubs/TechRpts/2013/EECS-2013-175.pdf"""
   m, n = a.shape[0], a.shape[1]

lib/python/arrays/dist/random.py

@@ -6,7 +6,7 @@ import halo
 
 from core import *
 
-@halo.distributed([List[int]], [DistArray])
+@halo.remote([List[int]], [DistArray])
 def normal(shape):
   num_blocks = DistArray.compute_num_blocks(shape)
   objrefs = np.empty(num_blocks, dtype=object)

lib/python/arrays/single/core.py

@@ -4,77 +4,77 @@ import halo
 
 __all__ = ["zeros", "zeros_like", "ones", "eye", "dot", "vstack", "hstack", "subarray", "copy", "tril", "triu", "diag", "transpose", "add", "subtract", "sum", "shape"]
 
-@halo.distributed([List[int], str, str], [np.ndarray])
+@halo.remote([List[int], str, str], [np.ndarray])
 def zeros(shape, dtype_name="float", order="C"):
   return np.zeros(shape, dtype=np.dtype(dtype_name), order=order)
 
-@halo.distributed([np.ndarray, str, str, bool], [np.ndarray])
+@halo.remote([np.ndarray, str, str, bool], [np.ndarray])
 def zeros_like(a, dtype_name="None", order="K", subok=True):
   dtype_val = None if dtype_name == "None" else np.dtype(dtype_name)
   return np.zeros_like(a, dtype=dtype_val, order=order, subok=subok)
 
-@halo.distributed([List[int], str, str], [np.ndarray])
+@halo.remote([List[int], str, str], [np.ndarray])
 def ones(shape, dtype_name="float", order="C"):
   return np.ones(shape, dtype=np.dtype(dtype_name), order=order)
 
-@halo.distributed([int, int, int, str], [np.ndarray])
+@halo.remote([int, int, int, str], [np.ndarray])
 def eye(N, M=-1, k=0, dtype_name="float"):
   M = N if M == -1 else M
   return np.eye(N, M=M, k=k, dtype=np.dtype(dtype_name))
 
-@halo.distributed([np.ndarray, np.ndarray], [np.ndarray])
+@halo.remote([np.ndarray, np.ndarray], [np.ndarray])
 def dot(a, b):
   return np.dot(a, b)
 
 # TODO(rkn): My preferred signature would have been
-# @halo.distributed([List[np.ndarray]], [np.ndarray]) but that currently doesn't
+# @halo.remote([List[np.ndarray]], [np.ndarray]) but that currently doesn't
 # work because that would expect a list of ndarrays not a list of ObjRefs
-@halo.distributed([np.ndarray, None], [np.ndarray])
+@halo.remote([np.ndarray, None], [np.ndarray])
 def vstack(*xs):
   return np.vstack(xs)
 
-@halo.distributed([np.ndarray, None], [np.ndarray])
+@halo.remote([np.ndarray, None], [np.ndarray])
 def hstack(*xs):
   return np.hstack(xs)
 
 # TODO(rkn): this doesn't parallel the numpy API, but we can't really slice an ObjRef, think about this
-@halo.distributed([np.ndarray, List[int], List[int]], [np.ndarray])
+@halo.remote([np.ndarray, List[int], List[int]], [np.ndarray])
 def subarray(a, lower_indices, upper_indices): # TODO(rkn): be consistent about using "index" versus "indices"
   return a[[slice(l, u) for (l, u) in zip(lower_indices, upper_indices)]]
 
-@halo.distributed([np.ndarray, str], [np.ndarray])
+@halo.remote([np.ndarray, str], [np.ndarray])
 def copy(a, order="K"):
   return np.copy(a, order=order)
 
-@halo.distributed([np.ndarray, int], [np.ndarray])
+@halo.remote([np.ndarray, int], [np.ndarray])
 def tril(m, k=0):
   return np.tril(m, k=k)
 
-@halo.distributed([np.ndarray, int], [np.ndarray])
+@halo.remote([np.ndarray, int], [np.ndarray])
 def triu(m, k=0):
   return np.triu(m, k=k)
 
-@halo.distributed([np.ndarray, int], [np.ndarray])
+@halo.remote([np.ndarray, int], [np.ndarray])
 def diag(v, k=0):
   return np.diag(v, k=k)
 
-@halo.distributed([np.ndarray, List[int]], [np.ndarray])
+@halo.remote([np.ndarray, List[int]], [np.ndarray])
 def transpose(a, axes=[]):
   axes = None if axes == [] else axes
   return np.transpose(a, axes=axes)
 
-@halo.distributed([np.ndarray, np.ndarray], [np.ndarray])
+@halo.remote([np.ndarray, np.ndarray], [np.ndarray])
 def add(x1, x2):
   return np.add(x1, x2)
 
-@halo.distributed([np.ndarray, np.ndarray], [np.ndarray])
+@halo.remote([np.ndarray, np.ndarray], [np.ndarray])
 def subtract(x1, x2):
   return np.subtract(x1, x2)
 
-@halo.distributed([int, np.ndarray, None], [np.ndarray])
+@halo.remote([int, np.ndarray, None], [np.ndarray])
 def sum(axis, *xs):
   return np.sum(xs, axis=axis)
 
-@halo.distributed([np.ndarray], [tuple])
+@halo.remote([np.ndarray], [tuple])
 def shape(a):
   return np.shape(a)

lib/python/arrays/single/linalg.py

@@ -7,82 +7,82 @@ __all__ = ["matrix_power", "solve", "tensorsolve", "tensorinv", "inv",
            "svd", "eig", "eigh", "lstsq", "norm", "qr", "cond", "matrix_rank",
            "LinAlgError", "multi_dot"]
 
-@halo.distributed([np.ndarray, int], [np.ndarray])
+@halo.remote([np.ndarray, int], [np.ndarray])
 def matrix_power(M, n):
   return np.linalg.matrix_power(M, n)
 
-@halo.distributed([np.ndarray, np.ndarray], [np.ndarray])
+@halo.remote([np.ndarray, np.ndarray], [np.ndarray])
 def solve(a, b):
   return np.linalg.solve(a, b)
 
-@halo.distributed([np.ndarray], [np.ndarray, np.ndarray])
+@halo.remote([np.ndarray], [np.ndarray, np.ndarray])
 def tensorsolve(a):
   raise NotImplementedError
 
-@halo.distributed([np.ndarray], [np.ndarray, np.ndarray])
+@halo.remote([np.ndarray], [np.ndarray, np.ndarray])
 def tensorinv(a):
   raise NotImplementedError
 
-@halo.distributed([np.ndarray], [np.ndarray])
+@halo.remote([np.ndarray], [np.ndarray])
 def inv(a):
   return np.linalg.inv(a)
 
-@halo.distributed([np.ndarray], [np.ndarray])
+@halo.remote([np.ndarray], [np.ndarray])
 def cholesky(a):
   return np.linalg.cholesky(a)
 
-@halo.distributed([np.ndarray], [np.ndarray])
+@halo.remote([np.ndarray], [np.ndarray])
 def eigvals(a):
   return np.linalg.eigvals(a)
 
-@halo.distributed([np.ndarray], [np.ndarray])
+@halo.remote([np.ndarray], [np.ndarray])
 def eigvalsh(a):
   raise NotImplementedError
 
-@halo.distributed([np.ndarray], [np.ndarray])
+@halo.remote([np.ndarray], [np.ndarray])
 def pinv(a):
   return np.linalg.pinv(a)
 
-@halo.distributed([np.ndarray], [int])
+@halo.remote([np.ndarray], [int])
 def slogdet(a):
   raise NotImplementedError
 
-@halo.distributed([np.ndarray], [float])
+@halo.remote([np.ndarray], [float])
 def det(a):
   return np.linalg.det(a)
 
-@halo.distributed([np.ndarray], [np.ndarray, np.ndarray, np.ndarray])
+@halo.remote([np.ndarray], [np.ndarray, np.ndarray, np.ndarray])
 def svd(a):
   return np.linalg.svd(a)
 
-@halo.distributed([np.ndarray], [np.ndarray, np.ndarray])
+@halo.remote([np.ndarray], [np.ndarray, np.ndarray])
 def eig(a):
   return np.linalg.eig(a)
 
-@halo.distributed([np.ndarray], [np.ndarray, np.ndarray])
+@halo.remote([np.ndarray], [np.ndarray, np.ndarray])
 def eigh(a):
   return np.linalg.eigh(a)
 
-@halo.distributed([np.ndarray], [np.ndarray, np.ndarray, int, np.ndarray])
+@halo.remote([np.ndarray], [np.ndarray, np.ndarray, int, np.ndarray])
 def lstsq(a, b):
   return np.linalg.lstsq(a)
 
-@halo.distributed([np.ndarray], [float])
+@halo.remote([np.ndarray], [float])
 def norm(x):
   return np.linalg.norm(x)
 
-@halo.distributed([np.ndarray], [np.ndarray, np.ndarray])
+@halo.remote([np.ndarray], [np.ndarray, np.ndarray])
 def qr(a):
   return np.linalg.qr(a)
 
-@halo.distributed([np.ndarray], [float])
+@halo.remote([np.ndarray], [float])
 def cond(x):
   return np.linalg.cond(x)
 
-@halo.distributed([np.ndarray], [int])
+@halo.remote([np.ndarray], [int])
 def matrix_rank(M):
   return np.linalg.matrix_rank(M)
 
-@halo.distributed([np.ndarray, None], [np.ndarray])
+@halo.remote([np.ndarray, None], [np.ndarray])
 def multi_dot(a):
   raise NotImplementedError

lib/python/arrays/single/random.py

@@ -2,6 +2,6 @@ from typing import List
 import numpy as np
 import halo
 
-@halo.distributed([List[int]], [np.ndarray])
+@halo.remote([List[int]], [np.ndarray])
 def normal(shape):
   return np.random.normal(size=shape)

lib/python/halo/__init__.py

@@ -1,3 +1,3 @@
 import libhalolib as lib
 import serialization
-from worker import scheduler_info, register_module, connect, disconnect, pull, push, distributed
+from worker import scheduler_info, register_module, connect, disconnect, pull, push, remote

lib/python/halo/worker.py

@@ -60,7 +60,7 @@ def register_module(module, recursive=False, worker=global_worker):
   print "registering functions in module {}.".format(module.__name__)
   for name in dir(module):
     val = getattr(module, name)
-    if hasattr(val, "is_distributed") and val.is_distributed:
+    if hasattr(val, "is_remote") and val.is_remote:
       print "registering {}.".format(val.func_name)
       worker.register_function(val)
     # elif recursive and isinstance(val, ModuleType):
@@ -97,17 +97,17 @@ def main_loop(worker=global_worker):
     task = halo.lib.wait_for_next_task(worker.handle)
     process_task(task)
 
-def distributed(arg_types, return_types, worker=global_worker):
-  def distributed_decorator(func):
+def remote(arg_types, return_types, worker=global_worker):
+  def remote_decorator(func):
     def func_executor(arguments):
-      """This is what gets executed remotely on a worker after a distributed function is scheduled by the scheduler."""
+      """This is what gets executed remotely on a worker after a remote function is scheduled by the scheduler."""
       print "Calling function {}".format(func.__name__)
       result = func(*arguments)
       check_return_values(func_call, result) # throws an exception if result is invalid
       print "Finished executing function {}".format(func.__name__)
       return result
     def func_call(*args, **kwargs):
-      """This is what gets run immediately when a worker calls a distributed function."""
+      """This is what gets run immediately when a worker calls a remote function."""
       args = list(args)
       args.extend([kwargs[keyword] if kwargs.has_key(keyword) else default for keyword, default in func_call.keyword_defaults[len(args):]]) # fill in the remaining arguments
       check_arguments(func_call, args) # throws an exception if args are invalid
@@ -117,23 +117,23 @@ def distributed(arg_types, return_types, worker=global_worker):
     func_call.executor = func_executor
     func_call.arg_types = arg_types
     func_call.return_types = return_types
-    func_call.is_distributed = True
+    func_call.is_remote = True
     func_call.keyword_defaults = [(k, v.default) for k, v in funcsigs.signature(func).parameters.iteritems()]
     return func_call
-  return distributed_decorator
+  return remote_decorator
 
 # helper method, this should not be called by the user
 def check_return_values(function, result):
   if len(function.return_types) == 1:
     result = (result,)
     # if not isinstance(result, function.return_types[0]):
-    #   raise Exception("The @distributed decorator for function {} expects one return value with type {}, but {} returned a {}.".format(function.__name__, function.return_types[0], function.__name__, type(result)))
+    #   raise Exception("The @remote decorator for function {} expects one return value with type {}, but {} returned a {}.".format(function.__name__, function.return_types[0], function.__name__, type(result)))
   else:
     if len(result) != len(function.return_types):
-      raise Exception("The @distributed decorator for function {} has {} return values with types {}, but {} returned {} values.".format(function.__name__, len(function.return_types), function.return_types, function.__name__, len(result)))
+      raise Exception("The @remote decorator for function {} has {} return values with types {}, but {} returned {} values.".format(function.__name__, len(function.return_types), function.return_types, function.__name__, len(result)))
     for i in range(len(result)):
       if (not isinstance(result[i], function.return_types[i])) and (not isinstance(result[i], halo.lib.ObjRef)):
-        raise Exception("The {}th return value for function {} has type {}, but the @distributed decorator expected a return value of type {} or an ObjRef.".format(i, function.__name__, type(result[i]), function.return_types[i]))
+        raise Exception("The {}th return value for function {} has type {}, but the @remote decorator expected a return value of type {} or an ObjRef.".format(i, function.__name__, type(result[i]), function.return_types[i]))
 
 # helper method, this should not be called by the user
 def check_arguments(function, args):