Switch build system to use CMake completely. (#200)

* switch to CMake completely

...

* cleanup

* Run C tests, update installation instructions.

python/ray/__init__.py

@@ -0,0 +1,20 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+# Ray version string
+__version__ = "0.01"
+
+import ctypes
+# Windows only
+if hasattr(ctypes, "windll"):
+  # Makes sure that all child processes die when we die
+  # Also makes sure that fatal crashes result in process termination rather than an error dialog (the latter is annoying since we have a lot of processes)
+  # This is done by associating all child processes with a "job" object that imposes this behavior.
+  (lambda kernel32: (lambda job: (lambda n: kernel32.SetInformationJobObject(job, 9, "\0" * 17 + chr(0x8 | 0x4 | 0x20) + "\0" * (n - 18), n))(0x90 if ctypes.sizeof(ctypes.c_void_p) > ctypes.sizeof(ctypes.c_int) else 0x70) and kernel32.AssignProcessToJobObject(job, ctypes.c_void_p(kernel32.GetCurrentProcess())))(ctypes.c_void_p(kernel32.CreateJobObjectW(None, None))) if kernel32 is not None else None)(ctypes.windll.kernel32)
+
+import ray.experimental
+import ray.serialization
+from ray.worker import register_class, error_info, init, connect, disconnect, get, put, wait, remote, log_event, log_span, flush_log
+from ray.worker import EnvironmentVariable, env
+from ray.worker import SCRIPT_MODE, WORKER_MODE, PYTHON_MODE, SILENT_MODE

python/ray/experimental/__init__.py

@@ -0,0 +1,6 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+from .utils import copy_directory
+from .tfutils import TensorFlowVariables

python/ray/experimental/array/distributed/__init__.py

@@ -0,0 +1,7 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+from . import random
+from . import linalg
+from .core import *

python/ray/experimental/array/distributed/core.py

@@ -0,0 +1,227 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import numpy as np
+import ray.experimental.array.remote as ra
+import ray
+
+__all__ = ["BLOCK_SIZE", "DistArray", "assemble", "zeros", "ones", "copy",
+           "eye", "triu", "tril", "blockwise_dot", "dot", "transpose", "add", "subtract", "numpy_to_dist", "subblocks"]
+
+BLOCK_SIZE = 10
+
+class DistArray(object):
+  def __init__(self, shape, objectids=None):
+    self.shape = shape
+    self.ndim = len(shape)
+    self.num_blocks = [int(np.ceil(1.0 * a / BLOCK_SIZE)) for a in self.shape]
+    self.objectids = objectids if objectids is not None else np.empty(self.num_blocks, dtype=object)
+    if self.num_blocks != list(self.objectids.shape):
+      raise Exception("The fields `num_blocks` and `objectids` are inconsistent, `num_blocks` is {} and `objectids` has shape {}".format(self.num_blocks, list(self.objectids.shape)))
+
+  @staticmethod
+  def compute_block_lower(index, shape):
+    if len(index) != len(shape):
+      raise Exception("The fields `index` and `shape` must have the same length, but `index` is {} and `shape` is {}.".format(index, shape))
+    return [elem * BLOCK_SIZE for elem in index]
+
+  @staticmethod
+  def compute_block_upper(index, shape):
+    if len(index) != len(shape):
+      raise Exception("The fields `index` and `shape` must have the same length, but `index` is {} and `shape` is {}.".format(index, shape))
+    upper = []
+    for i in range(len(shape)):
+      upper.append(min((index[i] + 1) * BLOCK_SIZE, shape[i]))
+    return upper
+
+  @staticmethod
+  def compute_block_shape(index, shape):
+    lower = DistArray.compute_block_lower(index, shape)
+    upper = DistArray.compute_block_upper(index, shape)
+    return [u - l for (l, u) in zip(lower, upper)]
+
+  @staticmethod
+  def compute_num_blocks(shape):
+    return [int(np.ceil(1.0 * a / BLOCK_SIZE)) for a in shape]
+
+  def assemble(self):
+    """Assemble an array on this node from a distributed array of object IDs."""
+    first_block = ray.get(self.objectids[(0,) * self.ndim])
+    dtype = first_block.dtype
+    result = np.zeros(self.shape, dtype=dtype)
+    for index in np.ndindex(*self.num_blocks):
+      lower = DistArray.compute_block_lower(index, self.shape)
+      upper = DistArray.compute_block_upper(index, self.shape)
+      result[[slice(l, u) for (l, u) in zip(lower, upper)]] = ray.get(self.objectids[index])
+    return result
+
+  def __getitem__(self, sliced):
+    # TODO(rkn): fix this, this is just a placeholder that should work but is inefficient
+    a = self.assemble()
+    return a[sliced]
+
+# Register the DistArray class with Ray so that it knows how to serialize it.
+ray.register_class(DistArray)
+
+@ray.remote
+def assemble(a):
+  return a.assemble()
+
+# TODO(rkn): what should we call this method
+@ray.remote
+def numpy_to_dist(a):
+  result = DistArray(a.shape)
+  for index in np.ndindex(*result.num_blocks):
+    lower = DistArray.compute_block_lower(index, a.shape)
+    upper = DistArray.compute_block_upper(index, a.shape)
+    result.objectids[index] = ray.put(a[[slice(l, u) for (l, u) in zip(lower, upper)]])
+  return result
+
+@ray.remote
+def zeros(shape, dtype_name="float"):
+  result = DistArray(shape)
+  for index in np.ndindex(*result.num_blocks):
+    result.objectids[index] = ra.zeros.remote(DistArray.compute_block_shape(index, shape), dtype_name=dtype_name)
+  return result
+
+@ray.remote
+def ones(shape, dtype_name="float"):
+  result = DistArray(shape)
+  for index in np.ndindex(*result.num_blocks):
+    result.objectids[index] = ra.ones.remote(DistArray.compute_block_shape(index, shape), dtype_name=dtype_name)
+  return result
+
+@ray.remote
+def copy(a):
+  result = DistArray(a.shape)
+  for index in np.ndindex(*result.num_blocks):
+    result.objectids[index] = a.objectids[index] # We don't need to actually copy the objects because cluster-level objects are assumed to be immutable.
+  return result
+
+@ray.remote
+def eye(dim1, dim2=-1, dtype_name="float"):
+  dim2 = dim1 if dim2 == -1 else dim2
+  shape = [dim1, dim2]
+  result = DistArray(shape)
+  for (i, j) in np.ndindex(*result.num_blocks):
+    block_shape = DistArray.compute_block_shape([i, j], shape)
+    if i == j:
+      result.objectids[i, j] = ra.eye.remote(block_shape[0], block_shape[1], dtype_name=dtype_name)
+    else:
+      result.objectids[i, j] = ra.zeros.remote(block_shape, dtype_name=dtype_name)
+  return result
+
+@ray.remote
+def triu(a):
+  if a.ndim != 2:
+    raise Exception("Input must have 2 dimensions, but a.ndim is " + str(a.ndim))
+  result = DistArray(a.shape)
+  for (i, j) in np.ndindex(*result.num_blocks):
+    if i < j:
+      result.objectids[i, j] = ra.copy.remote(a.objectids[i, j])
+    elif i == j:
+      result.objectids[i, j] = ra.triu.remote(a.objectids[i, j])
+    else:
+      result.objectids[i, j] = ra.zeros_like.remote(a.objectids[i, j])
+  return result
+
+@ray.remote
+def tril(a):
+  if a.ndim != 2:
+    raise Exception("Input must have 2 dimensions, but a.ndim is " + str(a.ndim))
+  result = DistArray(a.shape)
+  for (i, j) in np.ndindex(*result.num_blocks):
+    if i > j:
+      result.objectids[i, j] = ra.copy.remote(a.objectids[i, j])
+    elif i == j:
+      result.objectids[i, j] = ra.tril.remote(a.objectids[i, j])
+    else:
+      result.objectids[i, j] = ra.zeros_like.remote(a.objectids[i, j])
+  return result
+
+@ray.remote
+def blockwise_dot(*matrices):
+  n = len(matrices)
+  if n % 2 != 0:
+    raise Exception("blockwise_dot expects an even number of arguments, but len(matrices) is {}.".format(n))
+  shape = (matrices[0].shape[0], matrices[n // 2].shape[1])
+  result = np.zeros(shape)
+  for i in range(n // 2):
+    result += np.dot(matrices[i], matrices[n // 2 + i])
+  return result
+
+@ray.remote
+def dot(a, b):
+  if a.ndim != 2:
+    raise Exception("dot expects its arguments to be 2-dimensional, but a.ndim = {}.".format(a.ndim))
+  if b.ndim != 2:
+    raise Exception("dot expects its arguments to be 2-dimensional, but b.ndim = {}.".format(b.ndim))
+  if a.shape[1] != b.shape[0]:
+    raise Exception("dot expects a.shape[1] to equal b.shape[0], but a.shape = {} and b.shape = {}.".format(a.shape, b.shape))
+  shape = [a.shape[0], b.shape[1]]
+  result = DistArray(shape)
+  for (i, j) in np.ndindex(*result.num_blocks):
+    args = list(a.objectids[i, :]) + list(b.objectids[:, j])
+    result.objectids[i, j] = blockwise_dot.remote(*args)
+  return result
+
+@ray.remote
+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,
+      subblocks(a, [0, 1], [2, 4])
+  will produce a DistArray whose objectids are
+      [[a.objectids[0, 2], a.objectids[0, 4]],
+       [a.objectids[1, 2], a.objectids[1, 4]]]
+  We allow the user to pass in an empty list [] to indicate the full range.
+  """
+  ranges = list(ranges)
+  if len(ranges) != a.ndim:
+    raise Exception("sub_blocks expects to receive a number of ranges equal to a.ndim, but it received {} ranges and a.ndim = {}.".format(len(ranges), a.ndim))
+  for i in range(len(ranges)):
+    if ranges[i] == []: # We allow the user to pass in an empty list to indicate the full range
+      ranges[i] = range(a.num_blocks[i])
+    if not np.alltrue(ranges[i] == np.sort(ranges[i])):
+      raise Exception("Ranges passed to sub_blocks must be sorted, but the {}th range is {}.".format(i, ranges[i]))
+    if ranges[i][0] < 0:
+      raise Exception("Values in the ranges passed to sub_blocks must be at least 0, but the {}th range is {}.".format(i, ranges[i]))
+    if ranges[i][-1] >= a.num_blocks[i]:
+      raise Exception("Values in the ranges passed to sub_blocks must be less than the relevant number of blocks, but the {}th range is {}, and a.num_blocks = {}.".format(i, ranges[i], a.num_blocks))
+  last_index = [r[-1] for r in ranges]
+  last_block_shape = DistArray.compute_block_shape(last_index, a.shape)
+  shape = [(len(ranges[i]) - 1) * BLOCK_SIZE + last_block_shape[i] for i in range(a.ndim)]
+  result = DistArray(shape)
+  for index in np.ndindex(*result.num_blocks):
+    result.objectids[index] = a.objectids[tuple([ranges[i][index[i]] for i in range(a.ndim)])]
+  return result
+
+@ray.remote
+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))
+  result = DistArray([a.shape[1], a.shape[0]])
+  for i in range(result.num_blocks[0]):
+    for j in range(result.num_blocks[1]):
+      result.objectids[i, j] = ra.transpose.remote(a.objectids[j, i])
+  return result
+
+# TODO(rkn): support broadcasting?
+@ray.remote
+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))
+  result = DistArray(x1.shape)
+  for index in np.ndindex(*result.num_blocks):
+    result.objectids[index] = ra.add.remote(x1.objectids[index], x2.objectids[index])
+  return result
+
+# TODO(rkn): support broadcasting?
+@ray.remote
+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))
+  result = DistArray(x1.shape)
+  for index in np.ndindex(*result.num_blocks):
+    result.objectids[index] = ra.subtract.remote(x1.objectids[index], x2.objectids[index])
+  return result

python/ray/experimental/array/distributed/linalg.py

@@ -0,0 +1,190 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import numpy as np
+import ray.experimental.array.remote as ra
+import ray
+
+from .core import *
+
+__all__ = ["tsqr", "modified_lu", "tsqr_hr", "qr"]
+
+@ray.remote(num_return_vals=2)
+def tsqr(a):
+  """
+  arguments:
+    a: a distributed matrix
+  Suppose that
+    a.shape == (M, N)
+    K == min(M, N)
+  return values:
+    q: DistArray, if q_full = ray.get(DistArray, q).assemble(), then
+      q_full.shape == (M, K)
+      np.allclose(np.dot(q_full.T, q_full), np.eye(K)) == True
+    r: np.ndarray, if r_val = ray.get(np.ndarray, r), then
+      r_val.shape == (K, N)
+      np.allclose(r, np.triu(r)) == True
+  """
+  if len(a.shape) != 2:
+    raise Exception("tsqr requires len(a.shape) == 2, but a.shape is {}".format(a.shape))
+  if a.num_blocks[1] != 1:
+    raise Exception("tsqr requires a.num_blocks[1] == 1, but a.num_blocks is {}".format(a.num_blocks))
+
+  num_blocks = a.num_blocks[0]
+  K = int(np.ceil(np.log2(num_blocks))) + 1
+  q_tree = np.empty((num_blocks, K), dtype=object)
+  current_rs = []
+  for i in range(num_blocks):
+    block = a.objectids[i, 0]
+    q, r = ra.linalg.qr.remote(block)
+    q_tree[i, 0] = q
+    current_rs.append(r)
+  for j in range(1, K):
+    new_rs = []
+    for i in range(int(np.ceil(1.0 * len(current_rs) / 2))):
+      stacked_rs = ra.vstack.remote(*current_rs[(2 * i):(2 * i + 2)])
+      q, r = ra.linalg.qr.remote(stacked_rs)
+      q_tree[i, j] = q
+      new_rs.append(r)
+    current_rs = new_rs
+  assert len(current_rs) == 1, "len(current_rs) = " + str(len(current_rs))
+
+  # handle the special case in which the whole DistArray "a" fits in one block
+  # and has fewer rows than columns, this is a bit ugly so think about how to
+  # remove it
+  if a.shape[0] >= a.shape[1]:
+    q_shape = a.shape
+  else:
+    q_shape = [a.shape[0], a.shape[0]]
+  q_num_blocks = DistArray.compute_num_blocks(q_shape)
+  q_objectids = np.empty(q_num_blocks, dtype=object)
+  q_result = DistArray(q_shape, q_objectids)
+
+  # reconstruct output
+  for i in range(num_blocks):
+    q_block_current = q_tree[i, 0]
+    ith_index = i
+    for j in range(1, K):
+      if np.mod(ith_index, 2) == 0:
+        lower = [0, 0]
+        upper = [a.shape[1], BLOCK_SIZE]
+      else:
+        lower = [a.shape[1], 0]
+        upper = [2 * a.shape[1], BLOCK_SIZE]
+      ith_index //= 2
+      q_block_current = ra.dot.remote(q_block_current, ra.subarray.remote(q_tree[ith_index, j], lower, upper))
+    q_result.objectids[i] = q_block_current
+  r = current_rs[0]
+  return q_result, ray.get(r)
+
+# TODO(rkn): This is unoptimized, we really want a block version of this.
+@ray.remote(num_return_vals=3)
+def modified_lu(q):
+  """
+  Algorithm 5 from http://www.eecs.berkeley.edu/Pubs/TechRpts/2013/EECS-2013-175.pdf
+  takes a matrix q with orthonormal columns, returns l, u, s such that q - s = l * u
+  arguments:
+    q: a two dimensional orthonormal q
+  return values:
+    l: lower triangular
+    u: upper triangular
+    s: a diagonal matrix represented by its diagonal
+  """
+  q = q.assemble()
+  m, b = q.shape[0], q.shape[1]
+  S = np.zeros(b)
+
+  q_work = np.copy(q)
+
+  for i in range(b):
+    S[i] = -1 * np.sign(q_work[i, i])
+    q_work[i, i] -= S[i]
+    q_work[(i + 1):m, i] /= q_work[i, i] # scale ith column of L by diagonal element
+    q_work[(i + 1):m, (i + 1):b] -= np.outer(q_work[(i + 1):m, i], q_work[i, (i + 1):b]) # perform Schur complement update
+
+  L = np.tril(q_work)
+  for i in range(b):
+    L[i, i] = 1
+  U = np.triu(q_work)[:b, :]
+  return ray.get(numpy_to_dist.remote(ray.put(L))), U, S # TODO(rkn): get rid of put
+
+@ray.remote(num_return_vals=2)
+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
+
+@ray.remote
+def tsqr_hr_helper2(s, r_temp):
+  s_full = np.diag(s)
+  return np.dot(s_full, r_temp)
+
+@ray.remote(num_return_vals=4)
+def tsqr_hr(a):
+  """Algorithm 6 from http://www.eecs.berkeley.edu/Pubs/TechRpts/2013/EECS-2013-175.pdf"""
+  q, r_temp = tsqr.remote(a)
+  y, u, s = modified_lu.remote(q)
+  y_blocked = ray.get(y)
+  t, y_top = tsqr_hr_helper1.remote(u, s, y_blocked.objectids[0, 0], a.shape[1])
+  r = tsqr_hr_helper2.remote(s, r_temp)
+  return ray.get(y), ray.get(t), ray.get(y_top), ray.get(r)
+
+@ray.remote
+def qr_helper1(a_rc, y_ri, t, W_c):
+  return a_rc - np.dot(y_ri, np.dot(t.T, W_c))
+
+@ray.remote
+def qr_helper2(y_ri, a_rc):
+  return np.dot(y_ri.T, a_rc)
+
+@ray.remote(num_return_vals=2)
+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]
+  k = min(m, n)
+
+  # we will store our scratch work in a_work
+  a_work = DistArray(a.shape, np.copy(a.objectids))
+
+  result_dtype = np.linalg.qr(ray.get(a.objectids[0, 0]))[0].dtype.name
+  r_res = ray.get(zeros.remote([k, n], result_dtype)) # TODO(rkn): It would be preferable not to get this right after creating it.
+  y_res = ray.get(zeros.remote([m, k], result_dtype)) # TODO(rkn): It would be preferable not to get this right after creating it.
+  Ts = []
+
+  for i in range(min(a.num_blocks[0], a.num_blocks[1])): # this differs from the paper, which says "for i in range(a.num_blocks[1])", but that doesn't seem to make any sense when a.num_blocks[1] > a.num_blocks[0]
+    sub_dist_array = subblocks.remote(a_work, list(range(i, a_work.num_blocks[0])), [i])
+    y, t, _, R = tsqr_hr.remote(sub_dist_array)
+    y_val = ray.get(y)
+
+    for j in range(i, a.num_blocks[0]):
+      y_res.objectids[j, i] = y_val.objectids[j - i, 0]
+    if a.shape[0] > a.shape[1]:
+      # in this case, R needs to be square
+      R_shape = ray.get(ra.shape.remote(R))
+      eye_temp = ra.eye.remote(R_shape[1], R_shape[0], dtype_name=result_dtype)
+      r_res.objectids[i, i] = ra.dot.remote(eye_temp, R)
+    else:
+      r_res.objectids[i, i] = R
+    Ts.append(numpy_to_dist.remote(t))
+
+    for c in range(i + 1, a.num_blocks[1]):
+      W_rcs = []
+      for r in range(i, a.num_blocks[0]):
+        y_ri = y_val.objectids[r - i, 0]
+        W_rcs.append(qr_helper2.remote(y_ri, a_work.objectids[r, c]))
+      W_c = ra.sum_list.remote(*W_rcs)
+      for r in range(i, a.num_blocks[0]):
+        y_ri = y_val.objectids[r - i, 0]
+        A_rc = qr_helper1.remote(a_work.objectids[r, c], y_ri, t, W_c)
+        a_work.objectids[r, c] = A_rc
+      r_res.objectids[i, c] = a_work.objectids[i, c]
+
+  # construct q_res from Ys and Ts
+  q = eye.remote(m, k, dtype_name=result_dtype)
+  for i in range(len(Ts))[::-1]:
+    y_col_block = subblocks.remote(y_res, [], [i])
+    q = subtract.remote(q, dot.remote(y_col_block, dot.remote(Ts[i], dot.remote(transpose.remote(y_col_block), q))))
+
+  return ray.get(q), r_res

python/ray/experimental/array/distributed/random.py

@@ -0,0 +1,18 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import numpy as np
+import ray.experimental.array.remote as ra
+import ray
+
+from .core import *
+
+@ray.remote
+def normal(shape):
+  num_blocks = DistArray.compute_num_blocks(shape)
+  objectids = np.empty(num_blocks, dtype=object)
+  for index in np.ndindex(*num_blocks):
+    objectids[index] = ra.random.normal.remote(DistArray.compute_block_shape(index, shape))
+  result = DistArray(shape, objectids)
+  return result

python/ray/experimental/array/remote/__init__.py

@@ -0,0 +1,7 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+from . import random
+from . import linalg
+from .core import *

python/ray/experimental/array/remote/core.py

@@ -0,0 +1,86 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import numpy as np
+import ray
+
+__all__ = ["zeros", "zeros_like", "ones", "eye", "dot", "vstack", "hstack", "subarray", "copy", "tril", "triu", "diag", "transpose", "add", "subtract", "sum", "shape", "sum_list"]
+
+@ray.remote
+def zeros(shape, dtype_name="float", order="C"):
+  return np.zeros(shape, dtype=np.dtype(dtype_name), order=order)
+
+@ray.remote
+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)
+
+@ray.remote
+def ones(shape, dtype_name="float", order="C"):
+  return np.ones(shape, dtype=np.dtype(dtype_name), order=order)
+
+@ray.remote
+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))
+
+@ray.remote
+def dot(a, b):
+  return np.dot(a, b)
+
+@ray.remote
+def vstack(*xs):
+  return np.vstack(xs)
+
+@ray.remote
+def hstack(*xs):
+  return np.hstack(xs)
+
+# TODO(rkn): instead of this, consider implementing slicing
+@ray.remote
+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)]]
+
+@ray.remote
+def copy(a, order="K"):
+  return np.copy(a, order=order)
+
+@ray.remote
+def tril(m, k=0):
+  return np.tril(m, k=k)
+
+@ray.remote
+def triu(m, k=0):
+  return np.triu(m, k=k)
+
+@ray.remote
+def diag(v, k=0):
+  return np.diag(v, k=k)
+
+@ray.remote
+def transpose(a, axes=[]):
+  axes = None if axes == [] else axes
+  return np.transpose(a, axes=axes)
+
+@ray.remote
+def add(x1, x2):
+  return np.add(x1, x2)
+
+@ray.remote
+def subtract(x1, x2):
+  return np.subtract(x1, x2)
+
+@ray.remote
+def sum(x, axis=-1):
+  return np.sum(x, axis=axis if axis != -1 else None)
+
+@ray.remote
+def shape(a):
+  return np.shape(a)
+
+# We use Any to allow different numerical types as well as numpy arrays.
+# TODO(rkn):this isn't in the numpy API, so be careful about exposing this.
+@ray.remote
+def sum_list(*xs):
+  return np.sum(xs, axis=0)

python/ray/experimental/array/remote/linalg.py

@@ -0,0 +1,91 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import numpy as np
+import ray
+
+__all__ = ["matrix_power", "solve", "tensorsolve", "tensorinv", "inv",
+           "cholesky", "eigvals", "eigvalsh", "pinv", "slogdet", "det",
+           "svd", "eig", "eigh", "lstsq", "norm", "qr", "cond", "matrix_rank",
+           "LinAlgError", "multi_dot"]
+
+@ray.remote
+def matrix_power(M, n):
+  return np.linalg.matrix_power(M, n)
+
+@ray.remote
+def solve(a, b):
+  return np.linalg.solve(a, b)
+
+@ray.remote(num_return_vals=2)
+def tensorsolve(a):
+  raise NotImplementedError
+
+@ray.remote(num_return_vals=2)
+def tensorinv(a):
+  raise NotImplementedError
+
+@ray.remote
+def inv(a):
+  return np.linalg.inv(a)
+
+@ray.remote
+def cholesky(a):
+  return np.linalg.cholesky(a)
+
+@ray.remote
+def eigvals(a):
+  return np.linalg.eigvals(a)
+
+@ray.remote
+def eigvalsh(a):
+  raise NotImplementedError
+
+@ray.remote
+def pinv(a):
+  return np.linalg.pinv(a)
+
+@ray.remote
+def slogdet(a):
+  raise NotImplementedError
+
+@ray.remote
+def det(a):
+  return np.linalg.det(a)
+
+@ray.remote(num_return_vals=3)
+def svd(a):
+  return np.linalg.svd(a)
+
+@ray.remote(num_return_vals=2)
+def eig(a):
+  return np.linalg.eig(a)
+
+@ray.remote(num_return_vals=2)
+def eigh(a):
+  return np.linalg.eigh(a)
+
+@ray.remote(num_return_vals=4)
+def lstsq(a, b):
+  return np.linalg.lstsq(a)
+
+@ray.remote
+def norm(x):
+  return np.linalg.norm(x)
+
+@ray.remote(num_return_vals=2)
+def qr(a):
+  return np.linalg.qr(a)
+
+@ray.remote
+def cond(x):
+  return np.linalg.cond(x)
+
+@ray.remote
+def matrix_rank(M):
+  return np.linalg.matrix_rank(M)
+
+@ray.remote
+def multi_dot(*a):
+  raise NotImplementedError

python/ray/experimental/array/remote/random.py

@@ -0,0 +1,10 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import numpy as np
+import ray
+
+@ray.remote
+def normal(shape):
+  return np.random.normal(size=shape)

python/ray/experimental/tfutils.py

@@ -0,0 +1,78 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+import numpy as np
+
+def unflatten(vector, shapes):
+  i = 0
+  arrays = []
+  for shape in shapes:
+    size = np.prod(shape)
+    array = vector[i:(i + size)].reshape(shape)
+    arrays.append(array)
+    i += size
+  assert len(vector) == i, "Passed weight does not have the correct shape."
+  return arrays
+
+class TensorFlowVariables(object):
+  """An object used to extract variables from a loss function.
+
+  This object also provides methods for getting and setting the weights of the
+  relevant variables.
+
+  Attributes:
+    sess (tf.Session): The tensorflow session used to run assignment.
+    loss: The loss function passed in by the user.
+    variables (List[tf.Variable]): Extracted variables from the loss.
+    assignment_placeholders (List[tf.placeholders]): The nodes that weights get
+      passed to.
+    assignment_nodes (List[tf.Tensor]): The nodes that assign the weights.
+  """
+  def __init__(self, loss, sess=None):
+    """Creates a TensorFlowVariables instance."""
+    import tensorflow as tf
+    self.sess = sess
+    self.loss = loss
+    variable_names = [op.node_def.name for op in loss.graph.get_operations() if op.node_def.op == "Variable"]
+    self.variables = [v for v in tf.trainable_variables() if v.op.node_def.name in variable_names]
+    self.assignment_placeholders = dict()
+    self.assignment_nodes = []
+
+    # Create new placeholders to put in custom weights.
+    for var in self.variables:
+      self.assignment_placeholders[var.op.node_def.name] = tf.placeholder(var.value().dtype, var.get_shape().as_list())
+      self.assignment_nodes.append(var.assign(self.assignment_placeholders[var.op.node_def.name]))
+
+  def set_session(self, sess):
+    """Modifies the current session used by the class."""
+    self.sess = sess
+
+  def get_flat_size(self):
+    return sum([np.prod(v.get_shape().as_list()) for v in self.variables])
+
+  def _check_sess(self):
+    """Checks if the session is set, and if not throw an error message."""
+    assert self.sess is not None, "The session is not set. Set the session either by passing it into the TensorFlowVariables constructor or by calling set_session(sess)."
+  
+  def get_flat(self):
+    """Gets the weights and returns them as a flat array."""
+    self._check_sess()
+    return np.concatenate([v.eval(session=self.sess).flatten() for v in self.variables])
+  
+  def set_flat(self, new_weights):
+    """Sets the weights to new_weights, converting from a flat array."""
+    self._check_sess()
+    shapes = [v.get_shape().as_list() for v in self.variables]
+    arrays = unflatten(new_weights, shapes)
+    placeholders = [self.assignment_placeholders[v.op.node_def.name] for v in self.variables]
+    self.sess.run(self.assignment_nodes, feed_dict=dict(zip(placeholders,arrays)))
+
+  def get_weights(self):
+    """Returns the weights of the variables of the loss function in a list."""
+    self._check_sess()
+    return {v.op.node_def.name: v.eval(session=self.sess) for v in self.variables}
+
+  def set_weights(self, new_weights):
+    """Sets the weights to new_weights."""
+    self._check_sess()
+    self.sess.run(self.assignment_nodes, feed_dict={self.assignment_placeholders[name]: value for (name, value) in new_weights.items()})

python/ray/experimental/utils.py

@@ -0,0 +1,74 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import io
+import os
+import tarfile
+import sys
+
+import ray
+
+def tarred_directory_as_bytes(source_dir):
+  """Tar a directory and return it as a byte string.
+
+  Args:
+    source_dir (str): The name of the directory to tar.
+
+  Returns:
+    A byte string representing the tarred file.
+  """
+  # Get a BytesIO object.
+  string_file = io.BytesIO()
+  # Create an in-memory tarfile of the source directory.
+  with tarfile.open(mode="w:gz", fileobj=string_file) as tar:
+    tar.add(source_dir, arcname=os.path.basename(source_dir))
+  string_file.seek(0)
+  return string_file.read()
+
+def tarred_bytes_to_directory(tarred_bytes, target_dir):
+  """Take a byte string and untar it.
+
+  Args:
+    tarred_bytes (str): A byte string representing the tarred file. This should
+      be the output of tarred_directory_as_bytes.
+    target_dir (str): The directory to create the untarred files in.
+  """
+  string_file = io.BytesIO(tarred_bytes)
+  with tarfile.open(fileobj=string_file) as tar:
+    tar.extractall(path=target_dir)
+
+def copy_directory(source_dir, target_dir=None):
+  """Copy a local directory to each machine in the Ray cluster.
+
+  Note that both source_dir and target_dir must have the same basename). For
+  example, source_dir can be /a/b/c and target_dir can be /d/e/c. In this case,
+  the directory /d/e will be added to the Python path of each worker.
+
+  Note that this method is not completely safe to use. For example, workers that
+  do not do the copying and only set their paths (only one worker per node does
+  the copying) may try to execute functions that use the files in the directory
+  being copied before the directory being copied has finished untarring.
+
+  Args:
+    source_dir (str): The directory to copy.
+    target_dir (str): The location to copy it to on the other machines. If this
+      is not provided, the source_dir will be used. If it is provided and is
+      different from source_dir, the source_dir also be copied to the target_dir
+      location on this machine.
+  """
+  target_dir = source_dir if target_dir is None else target_dir
+  source_dir = os.path.abspath(source_dir)
+  target_dir = os.path.abspath(target_dir)
+  source_basename = os.path.basename(source_dir)
+  target_basename = os.path.basename(target_dir)
+  if source_basename != target_basename:
+    raise Exception("The source_dir and target_dir must have the same base name, {} != {}".format(source_basename, target_basename))
+  tarred_bytes = tarred_directory_as_bytes(source_dir)
+  def f(worker_info):
+    if worker_info["counter"] == 0:
+      tarred_bytes_to_directory(tarred_bytes, os.path.dirname(target_dir))
+    sys.path.append(os.path.dirname(target_dir))
+  # Run this function on all workers to copy the directory to all nodes and to
+  # add the directory to the Python path of each worker.
+  ray.worker.global_worker.run_function_on_all_workers(f)

python/ray/pickling.py

@@ -0,0 +1,67 @@
+# Note that a little bit of code here is taken and slightly modified from the pickler because it was not possible to change its behavior otherwise.
+
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import sys
+from ctypes import c_void_p
+from cloudpickle import pickle, cloudpickle, CloudPickler, load, loads
+
+try:
+  from ctypes import pythonapi
+  pythonapi.PyCell_Set  # Make sure this exists
+except:
+  pythonapi = None
+
+def dump(obj, file, protocol=2):
+  return BetterPickler(file, protocol).dump(obj)
+
+def dumps(obj):
+  stringio = cloudpickle.StringIO()
+  dump(obj, stringio)
+  return stringio.getvalue()
+
+def _make_skel_func(code, closure, base_globals = None):
+  """ Creates a skeleton function object that contains just the provided
+      code and the correct number of cells in func_closure.  All other
+      func attributes (e.g. func_globals) are empty.
+  """
+  if base_globals is None: base_globals = {}
+  base_globals['__builtins__'] = __builtins__
+  return _make_skel_func.__class__(code, base_globals, None, None, tuple(closure))
+
+def _fill_function(func, globals, defaults, closure, dict):
+  """ Fills in the rest of function data into the skeleton function object
+      that were created via _make_skel_func(), including closures.
+  """
+  result = cloudpickle._fill_function(func, globals, defaults, dict)
+  if pythonapi is not None:
+    for i, v in enumerate(closure):
+      pythonapi.PyCell_Set(c_void_p(id(result.__closure__[i])), c_void_p(id(v)))
+  return result
+
+class BetterPickler(CloudPickler):
+  def save_function_tuple(self, func):
+    code, f_globals, defaults, closure, dct, base_globals = self.extract_func_data(func)
+
+    self.save(_fill_function)
+    self.write(pickle.MARK)
+
+    self.save(_make_skel_func if pythonapi else cloudpickle._make_skel_func)
+    self.save((code, map(lambda _: cloudpickle._make_cell(None), closure) if closure and pythonapi is not None else closure, base_globals))
+    self.write(pickle.REDUCE)
+    self.memoize(func)
+
+    self.save(f_globals)
+    self.save(defaults)
+    self.save(closure)
+    self.save(dct)
+    self.write(pickle.TUPLE)
+    self.write(pickle.REDUCE)
+  def save_cell(self, obj):
+    self.save(cloudpickle._make_cell)
+    self.save((obj.cell_contents,))
+    self.write(pickle.REDUCE)
+  dispatch = CloudPickler.dispatch.copy()
+  dispatch[(lambda _: lambda: _)(0).__closure__[0].__class__] = save_cell

python/ray/serialization.py

@@ -0,0 +1,143 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import numpy as np
+import numbuf
+
+import ray.pickling as pickling
+
+def check_serializable(cls):
+  """Throws an exception if Ray cannot serialize this class efficiently.
+
+  Args:
+    cls (type): The class to be serialized.
+
+  Raises:
+    Exception: An exception is raised if Ray cannot serialize this class
+      efficiently.
+  """
+  if is_named_tuple(cls):
+    # This case works.
+    return
+  if not hasattr(cls, "__new__"):
+    raise Exception("The class {} does not have a '__new__' attribute, and is probably an old-style class. We do not support this. Please either make it a new-style class by inheriting from 'object', or use 'ray.register_class(cls, pickle=True)'. However, note that pickle is inefficient.".format(cls))
+  try:
+    obj = cls.__new__(cls)
+  except:
+    raise Exception("The class {} has overridden '__new__', so Ray may not be able to serialize it efficiently. Try using 'ray.register_class(cls, pickle=True)'. However, note that pickle is inefficient.".format(cls))
+  if not hasattr(obj, "__dict__"):
+    raise Exception("Objects of the class {} do not have a `__dict__` attribute, so Ray cannot serialize it efficiently. Try using 'ray.register_class(cls, pickle=True)'. However, note that pickle is inefficient.".format(cls))
+  if hasattr(obj, "__slots__"):
+    raise Exception("The class {} uses '__slots__', so Ray may not be able to serialize it efficiently. Try using 'ray.register_class(cls, pickle=True)'. However, note that pickle is inefficient.".format(cls))
+
+# This field keeps track of a whitelisted set of classes that Ray will
+# serialize.
+whitelisted_classes = {}
+classes_to_pickle = set()
+custom_serializers = {}
+custom_deserializers = {}
+
+def class_identifier(typ):
+  """Return a string that identifies this type."""
+  return "{}.{}".format(typ.__module__, typ.__name__)
+
+def is_named_tuple(cls):
+  """Return True if cls is a namedtuple and False otherwise."""
+  b = cls.__bases__
+  if len(b) != 1 or b[0] != tuple:
+    return False
+  f = getattr(cls, "_fields", None)
+  if not isinstance(f, tuple):
+    return False
+  return all(type(n) == str for n in f)
+
+def add_class_to_whitelist(cls, pickle=False, custom_serializer=None, custom_deserializer=None):
+  """Add cls to the list of classes that we can serialize.
+
+  Args:
+    cls (type): The class that we can serialize.
+    pickle (bool): True if the serialization should be done with pickle. False
+      if it should be done efficiently with Ray.
+    custom_serializer: This argument is optional, but can be provided to
+      serialize objects of the class in a particular way.
+    custom_deserializer: This argument is optional, but can be provided to
+      deserialize objects of the class in a particular way.
+  """
+  class_id = class_identifier(cls)
+  whitelisted_classes[class_id] = cls
+  if pickle:
+    classes_to_pickle.add(class_id)
+  if custom_serializer is not None:
+    custom_serializers[class_id] = custom_serializer
+    custom_deserializers[class_id] = custom_deserializer
+
+# Here we define a custom serializer and deserializer for handling numpy
+# arrays that contain objects.
+def array_custom_serializer(obj):
+  return obj.tolist(), obj.dtype.str
+def array_custom_deserializer(serialized_obj):
+  return np.array(serialized_obj[0], dtype=np.dtype(serialized_obj[1]))
+add_class_to_whitelist(np.ndarray, pickle=False, custom_serializer=array_custom_serializer, custom_deserializer=array_custom_deserializer)
+
+def serialize(obj):
+  """This is the callback that will be used by numbuf.
+
+  If numbuf does not know how to serialize an object, it will call this method.
+
+  Args:
+    obj (object): A Python object.
+
+  Returns:
+    A dictionary that has the key "_pyttype_" to identify the class, and
+      contains all information needed to reconstruct the object.
+  """
+  class_id = class_identifier(type(obj))
+  if class_id not in whitelisted_classes:
+    raise Exception("Ray does not know how to serialize objects of type {}. To fix this, call 'ray.register_class' with this class.".format(type(obj)))
+  if class_id in classes_to_pickle:
+    serialized_obj = {"data": pickling.dumps(obj)}
+  elif class_id in custom_serializers.keys():
+    serialized_obj = {"data": custom_serializers[class_id](obj)}
+  else:
+    # Handle the namedtuple case.
+    if is_named_tuple(type(obj)):
+      serialized_obj = {}
+      serialized_obj["_ray_getnewargs_"] = obj.__getnewargs__()
+    elif hasattr(obj, "__dict__"):
+      serialized_obj = obj.__dict__
+    else:
+      raise Exception("We do not know how to serialize the object '{}'".format(obj))
+  result = dict(serialized_obj, **{"_pytype_": class_id})
+  return result
+
+def deserialize(serialized_obj):
+  """This is the callback that will be used by numbuf.
+
+  If numbuf encounters a dictionary that contains the key "_pytype_" during
+    deserialization, it will ask this callback to deserialize the object.
+
+  Args:
+    serialized_obj (object): A dictionary that contains the key "_pytype_".
+
+  Returns:
+    A Python object.
+  """
+  class_id = serialized_obj["_pytype_"]
+  cls = whitelisted_classes[class_id]
+  if class_id in classes_to_pickle:
+    obj = pickling.loads(serialized_obj["data"])
+  elif class_id in custom_deserializers.keys():
+    obj = custom_deserializers[class_id](serialized_obj["data"])
+  else:
+    # In this case, serialized_obj should just be the __dict__ field.
+    if "_ray_getnewargs_" in serialized_obj:
+      obj = cls.__new__(cls, *serialized_obj["_ray_getnewargs_"])
+    else:
+      obj = cls.__new__(cls)
+      serialized_obj.pop("_pytype_")
+      obj.__dict__.update(serialized_obj)
+  return obj
+
+# Register the callbacks with numbuf.
+numbuf.register_callbacks(serialize, deserialize)

python/ray/services.py

@@ -0,0 +1,529 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import psutil
+import os
+import random
+import redis
+import signal
+import socket
+import string
+import subprocess
+import sys
+import time
+from collections import namedtuple
+
+# Ray modules
+import photon
+import plasma
+import global_scheduler
+
+# all_processes is a list of the scheduler, object store, and worker processes
+# that have been started by this services module if Ray is being used in local
+# mode.
+all_processes = []
+
+# True if processes are run in the valgrind profiler.
+RUN_PHOTON_PROFILER = False
+RUN_PLASMA_MANAGER_PROFILER = False
+RUN_PLASMA_STORE_PROFILER = False
+
+# ObjectStoreAddress tuples contain all information necessary to connect to an
+# object store. The fields are:
+# - name: The socket name for the object store
+# - manager_name: The socket name for the object store manager
+# - manager_port: The Internet port that the object store manager listens on
+ObjectStoreAddress = namedtuple("ObjectStoreAddress", ["name",
+                                                       "manager_name",
+                                                       "manager_port"])
+
+def address(host, port):
+  return host + ":" + str(port)
+
+def get_port(address):
+  try:
+    port = int(address.split(":")[1])
+  except:
+    raise Exception("Unable to parse port from address {}".format(address))
+  return port
+
+def new_port():
+  return random.randint(10000, 65535)
+
+def random_name():
+  return str(random.randint(0, 99999999))
+
+def cleanup():
+  """When running in local mode, shutdown the Ray processes.
+
+  This method is used to shutdown processes that were started with
+  services.start_ray_local(). It kills all scheduler, object store, and worker
+  processes that were started by this services module. Driver processes are
+  started and disconnected by worker.py.
+  """
+  global all_processes
+  successfully_shut_down = True
+  # Terminate the processes in reverse order.
+  for p in all_processes[::-1]:
+    if p.poll() is not None: # process has already terminated
+      continue
+    if RUN_PHOTON_PROFILER or RUN_PLASMA_MANAGER_PROFILER or RUN_PLASMA_STORE_PROFILER:
+      os.kill(p.pid, signal.SIGINT) # Give process signal to write profiler data.
+      time.sleep(0.1) # Wait for profiling data to be written.
+    p.kill()
+    time.sleep(0.05) # is this necessary?
+    if p.poll() is not None:
+      continue
+    p.terminate()
+    time.sleep(0.05) # is this necessary?
+    if p.poll is not None:
+      continue
+    successfully_shut_down = False
+  if successfully_shut_down:
+    if len(all_processes) > 0:
+      print("Successfully shut down Ray.")
+  else:
+    print("Ray did not shut down properly.")
+  all_processes = []
+
+def all_processes_alive():
+  return all([p.poll() is None for p in all_processes])
+
+def get_node_ip_address(address="8.8.8.8:53"):
+  """Determine the IP address of the local node.
+
+  Args:
+    address (str): The IP address and port of any known live service on the
+      network you care about.
+
+  Returns:
+    The IP address of the current node.
+  """
+  host, port = address.split(":")
+  s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+  s.connect((host, int(port)))
+  return s.getsockname()[0]
+
+def wait_for_redis_to_start(redis_host, redis_port, num_retries=5):
+  """Wait for a Redis server to be available.
+
+  This is accomplished by creating a Redis client and sending a random command
+  to the server until the command gets through.
+
+  Args:
+    redis_host (str): The IP address of the redis server.
+    redis_port (int): The port of the redis server.
+    num_retries (int): The number of times to try connecting with redis. The
+      client will sleep for one second between attempts.
+
+  Raises:
+    Exception: An exception is raised if we could not connect with Redis.
+  """
+  redis_client = redis.StrictRedis(host=redis_host, port=redis_port)
+  # Wait for the Redis server to start.
+  counter = 0
+  while counter < num_retries:
+    try:
+      # Run some random command and see if it worked.
+      redis_client.client_list()
+    except redis.ConnectionError as e:
+      # Wait a little bit.
+      time.sleep(1)
+      print("Failed to connect to the redis server, retrying.")
+      counter += 1
+    else:
+      break
+  if counter == num_retries:
+    raise Exception("Unable to connect to Redis. If the Redis instance is on a different machine, check that your firewall is configured properly.")
+
+def start_redis(node_ip_address, num_retries=20, cleanup=True, redirect_output=False):
+  """Start a Redis server.
+
+  Args:
+    num_retries (int): The number of times to attempt to start Redis.
+    cleanup (bool): True if using Ray in local mode. If cleanup is true, then
+      this process will be killed by serices.cleanup() when the Python process
+      that imported services exits.
+    redirect_output (bool): True if stdout and stderr should be redirected to
+      /dev/null.
+
+  Returns:
+    The port used by Redis.
+
+  Raises:
+    Exception: An exception is raised if Redis could not be started.
+  """
+  redis_filepath = os.path.join(os.path.dirname(os.path.abspath(__file__)), "../core/src/common/thirdparty/redis/src/redis-server")
+  redis_module = os.path.join(os.path.dirname(os.path.abspath(__file__)), "../core/src/common/redis_module/libray_redis_module.so")
+  assert os.path.isfile(redis_filepath)
+  assert os.path.isfile(redis_module)
+  counter = 0
+  while counter < num_retries:
+    if counter > 0:
+      print("Redis failed to start, retrying now.")
+    port = new_port()
+    with open(os.devnull, "w") as FNULL:
+      stdout = FNULL if redirect_output else None
+      stderr = FNULL if redirect_output else None
+      p = subprocess.Popen([redis_filepath, "--port", str(port), "--loglevel", "warning", "--loadmodule", redis_module], stdout=stdout, stderr=stderr)
+    time.sleep(0.1)
+    # Check if Redis successfully started (or at least if it the executable did
+    # not exit within 0.1 seconds).
+    if p.poll() is None:
+      if cleanup:
+        all_processes.append(p)
+      break
+    counter += 1
+  if counter == num_retries:
+    raise Exception("Couldn't start Redis.")
+
+  # Create a Redis client just for configuring Redis.
+  redis_client = redis.StrictRedis(host="127.0.0.1", port=port)
+  # Wait for the Redis server to start.
+  wait_for_redis_to_start("127.0.0.1", port)
+  # Configure Redis to generate keyspace notifications. TODO(rkn): Change this
+  # to only generate notifications for the export keys.
+  redis_client.config_set("notify-keyspace-events", "Kl")
+  # Configure Redis to not run in protected mode so that processes on other
+  # hosts can connect to it. TODO(rkn): Do this in a more secure way.
+  redis_client.config_set("protected-mode", "no")
+  redis_address = address(node_ip_address, port)
+  return redis_address
+
+def start_global_scheduler(redis_address, cleanup=True, redirect_output=False):
+  """Start a global scheduler process.
+
+  Args:
+    redis_address (str): The address of the Redis instance.
+    cleanup (bool): True if using Ray in local mode. If cleanup is true, then
+      this process will be killed by serices.cleanup() when the Python process
+      that imported services exits.
+    redirect_output (bool): True if stdout and stderr should be redirected to
+      /dev/null.
+  """
+  p = global_scheduler.start_global_scheduler(redis_address, redirect_output=redirect_output)
+  if cleanup:
+    all_processes.append(p)
+
+def start_local_scheduler(redis_address, node_ip_address, plasma_store_name, plasma_manager_name, plasma_address=None, cleanup=True, redirect_output=False):
+  """Start a local scheduler process.
+
+  Args:
+    redis_address (str): The address of the Redis instance.
+    node_ip_address (str): The IP address of the node that this local scheduler
+      is running on.
+    plasma_store_name (str): The name of the plasma store socket to connect to.
+    plasma_manager_name (str): The name of the plasma manager socket to connect
+      to.
+    cleanup (bool): True if using Ray in local mode. If cleanup is true, then
+      this process will be killed by serices.cleanup() when the Python process
+      that imported services exits.
+    redirect_output (bool): True if stdout and stderr should be redirected to
+      /dev/null.
+
+  Return:
+    The name of the local scheduler socket.
+  """
+  local_scheduler_name, p = photon.start_local_scheduler(plasma_store_name, plasma_manager_name, node_ip_address=node_ip_address, redis_address=redis_address, plasma_address=plasma_address, use_profiler=RUN_PHOTON_PROFILER, redirect_output=redirect_output)
+  if cleanup:
+    all_processes.append(p)
+  return local_scheduler_name
+
+def start_objstore(node_ip_address, redis_address, cleanup=True, redirect_output=False, objstore_memory=None):
+  """This method starts an object store process.
+
+  Args:
+    node_ip_address (str): The IP address of the node running the object store.
+    redis_address (str): The address of the Redis instance to connect to.
+    cleanup (bool): True if using Ray in local mode. If cleanup is true, then
+      this process will be killed by serices.cleanup() when the Python process
+      that imported services exits.
+    redirect_output (bool): True if stdout and stderr should be redirected to
+      /dev/null.
+
+  Return:
+    A tuple of the Plasma store socket name, the Plasma manager socket name, and
+      the plasma manager port.
+  """
+  if objstore_memory is None:
+    # Compute a fraction of the system memory for the Plasma store to use.
+    system_memory = psutil.virtual_memory().total
+    if sys.platform == "linux" or sys.platform == "linux2":
+      # On linux we use /dev/shm, its size is half the size of the physical
+      # memory. To not overflow it, we set the plasma memory limit to 0.4 times
+      # the size of the physical memory.
+      objstore_memory = int(system_memory * 0.4)
+      # Compare the requested memory size to the memory available in /dev/shm.
+      shm_fd = os.open("/dev/shm", os.O_RDONLY)
+      try:
+        shm_fs_stats = os.fstatvfs(shm_fd)
+        # The value shm_fs_stats.f_bsize is the block size and the value
+        # shm_fs_stats.f_bavail is the number of available blocks.
+        shm_avail = shm_fs_stats.f_bsize * shm_fs_stats.f_bavail
+        if objstore_memory > shm_avail:
+          print("Warning: Reducing object store memory because /dev/shm has only {} bytes available. You may be able to free up space by deleting files in /dev/shm. If you are inside a Docker container, you may need to pass an argument with the flag '--shm-size' to 'docker run'.".format(shm_avail))
+          objstore_memory = int(shm_avail * 0.8)
+      finally:
+        os.close(shm_fd)
+    else:
+      objstore_memory = int(system_memory * 0.8)
+  # Start the Plasma store.
+  plasma_store_name, p1 = plasma.start_plasma_store(plasma_store_memory=objstore_memory, use_profiler=RUN_PLASMA_STORE_PROFILER, redirect_output=redirect_output)
+  # Start the plasma manager.
+  plasma_manager_name, p2, plasma_manager_port = plasma.start_plasma_manager(plasma_store_name, redis_address, node_ip_address=node_ip_address, run_profiler=RUN_PLASMA_MANAGER_PROFILER, redirect_output=redirect_output)
+  if cleanup:
+    all_processes.append(p1)
+    all_processes.append(p2)
+
+  return ObjectStoreAddress(plasma_store_name, plasma_manager_name,
+                            plasma_manager_port)
+
+def start_worker(node_ip_address, object_store_name, object_store_manager_name, local_scheduler_name, redis_address, worker_path, cleanup=True, redirect_output=False):
+  """This method starts a worker process.
+
+  Args:
+    node_ip_address (str): The IP address of the node that this worker is
+      running on.
+    object_store_name (str): The name of the object store.
+    object_store_manager_name (str): The name of the object store manager.
+    local_scheduler_name (str): The name of the local scheduler.
+    redis_address (int): The address that the Redis server is listening on.
+    worker_path (str): The path of the source code which the worker process will
+      run.
+    cleanup (bool): True if using Ray in local mode. If cleanup is true, then
+      this process will be killed by services.cleanup() when the Python process
+      that imported services exits. This is True by default.
+    redirect_output (bool): True if stdout and stderr should be redirected to
+      /dev/null.
+  """
+  command = ["python",
+             worker_path,
+             "--node-ip-address=" + node_ip_address,
+             "--object-store-name=" + object_store_name,
+             "--object-store-manager-name=" + object_store_manager_name,
+             "--local-scheduler-name=" + local_scheduler_name,
+             "--redis-address=" + str(redis_address)]
+  with open(os.devnull, "w") as FNULL:
+    stdout = FNULL if redirect_output else None
+    stderr = FNULL if redirect_output else None
+    p = subprocess.Popen(command, stdout=stdout, stderr=stderr)
+  if cleanup:
+    all_processes.append(p)
+
+def start_webui(redis_port, cleanup=True, redirect_output=False):
+  """This method starts the web interface.
+
+  Args:
+    redis_port (int): The redis server's port
+    cleanup (bool): True if using Ray in local mode. If cleanup is true, then
+      this process will be killed by services.cleanup() when the Python process
+      that imported services exits. This is True by default.
+    redirect_output (bool): True if stdout and stderr should be redirected to
+      /dev/null.
+  """
+  executable = "nodejs" if sys.platform == "linux" or sys.platform == "linux2" else "node"
+  command = [executable, os.path.join(os.path.abspath(os.path.dirname(__file__)), "../webui/index.js"), str(redis_port)]
+  with open("/tmp/webui_out.txt", "wb") as out:
+    with open(os.devnull, "w") as FNULL:
+      stdout = FNULL if redirect_output else out
+      stderr = FNULL if redirect_output else None
+      p = subprocess.Popen(command, stdout=stdout, stderr=stderr)
+  if cleanup:
+    all_processes.append(p)
+
+def start_ray_processes(address_info=None,
+                        node_ip_address="127.0.0.1",
+                        num_workers=0,
+                        num_local_schedulers=1,
+                        worker_path=None,
+                        cleanup=True,
+                        redirect_output=False,
+                        include_global_scheduler=False):
+  """Helper method to start Ray processes.
+
+  Args:
+    address_info (dict): A dictionary with address information for processes
+      that have already been started. If provided, address_info will be
+      modified to include processes that are newly started.
+    node_ip_address (str): The IP address of this node.
+    num_workers (int): The number of workers to start.
+    num_local_schedulers (int): The total number of local schedulers required.
+      This is also the total number of object stores required. This method will
+      start new instances of local schedulers and object stores until there are
+      num_local_schedulers existing instances of each, including ones already
+      registered with the given address_info.
+    worker_path (str): The path of the source code that will be run by the
+      worker.
+    cleanup (bool): If cleanup is true, then the processes started here will be
+      killed by services.cleanup() when the Python process that called this
+      method exits.
+    redirect_output (bool): True if stdout and stderr should be redirected to
+      /dev/null.
+    include_global_scheduler (bool): If include_global_scheduler is True, then
+      start a global scheduler process.
+
+  Returns:
+    A dictionary of the address information for the processes that were
+      started.
+  """
+  if address_info is None:
+    address_info = {}
+  address_info["node_ip_address"] = node_ip_address
+
+  if worker_path is None:
+    worker_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "workers/default_worker.py")
+
+  # Start Redis if there isn't already an instance running. TODO(rkn): We are
+  # suppressing the output of Redis because on Linux it prints a bunch of
+  # warning messages when it starts up. Instead of suppressing the output, we
+  # should address the warnings.
+  redis_address = address_info.get("redis_address")
+  if redis_address is None:
+    redis_address = start_redis(node_ip_address, cleanup=cleanup,
+                                redirect_output=redirect_output)
+    address_info["redis_address"] = redis_address
+    time.sleep(0.1)
+  redis_port = get_port(redis_address)
+
+  # Start the global scheduler, if necessary.
+  if include_global_scheduler:
+    start_global_scheduler(redis_address, cleanup=cleanup,
+                           redirect_output=redirect_output)
+
+  # Initialize with existing services.
+  if "object_store_addresses" not in address_info:
+    address_info["object_store_addresses"] = []
+  object_store_addresses = address_info["object_store_addresses"]
+  if "local_scheduler_socket_names" not in address_info:
+    address_info["local_scheduler_socket_names"] = []
+  local_scheduler_socket_names = address_info["local_scheduler_socket_names"]
+
+  # Start any object stores that do not yet exist.
+  for _ in range(num_local_schedulers - len(object_store_addresses)):
+    # Start Plasma.
+    object_store_address = start_objstore(node_ip_address, redis_address,
+                                          cleanup=cleanup,
+                                          redirect_output=redirect_output)
+    object_store_addresses.append(object_store_address)
+    time.sleep(0.1)
+
+  # Start any local schedulers that do not yet exist.
+  for i in range(len(local_scheduler_socket_names), num_local_schedulers):
+    # Connect the local scheduler to the object store at the same index.
+    object_store_address = object_store_addresses[i]
+    plasma_address = "{}:{}".format(node_ip_address,
+                                    object_store_address.manager_port)
+    # Start the local scheduler.
+    local_scheduler_name = start_local_scheduler(redis_address,
+                                                 node_ip_address,
+                                                 object_store_address.name,
+                                                 object_store_address.manager_name,
+                                                 plasma_address=plasma_address,
+                                                 cleanup=cleanup,
+                                                 redirect_output=redirect_output)
+    local_scheduler_socket_names.append(local_scheduler_name)
+    time.sleep(0.1)
+
+  # Make sure that we have exactly num_local_schedulers instances of object
+  # stores and local schedulers.
+  assert len(object_store_addresses) == num_local_schedulers
+  assert len(local_scheduler_socket_names) == num_local_schedulers
+
+  # Start the workers.
+  for i in range(num_workers):
+    object_store_address = object_store_addresses[i % num_local_schedulers]
+    local_scheduler_name = local_scheduler_socket_names[i % num_local_schedulers]
+    start_worker(node_ip_address,
+                 object_store_address.name,
+                 object_store_address.manager_name,
+                 local_scheduler_name,
+                 redis_address,
+                 worker_path,
+                 cleanup=cleanup,
+                 redirect_output=redirect_output)
+
+  # Return the addresses of the relevant processes.
+  return address_info
+
+def start_ray_node(node_ip_address,
+                   redis_address,
+                   num_workers=0,
+                   num_local_schedulers=1,
+                   worker_path=None,
+                   cleanup=True,
+                   redirect_output=False):
+  """Start the Ray processes for a single node.
+
+  This assumes that the Ray processes on some master node have already been
+  started.
+
+  Args:
+    node_ip_address (str): The IP address of this node.
+    redis_address (str): The address of the Redis server.
+    num_workers (int): The number of workers to start.
+    num_local_schedulers (int): The number of local schedulers to start. This is
+      also the number of plasma stores and plasma managers to start.
+    worker_path (str): The path of the source code that will be run by the
+      worker.
+    cleanup (bool): If cleanup is true, then the processes started here will be
+      killed by services.cleanup() when the Python process that called this
+      method exits.
+    redirect_output (bool): True if stdout and stderr should be redirected to
+      /dev/null.
+
+  Returns:
+    A dictionary of the address information for the processes that were
+      started.
+  """
+  address_info = {
+      "redis_address": redis_address,
+      }
+  return start_ray_processes(address_info=address_info,
+                             node_ip_address=node_ip_address,
+                             num_workers=num_workers,
+                             num_local_schedulers=num_local_schedulers,
+                             worker_path=worker_path,
+                             cleanup=cleanup,
+                             redirect_output=redirect_output)
+
+def start_ray_local(address_info=None,
+                    node_ip_address="127.0.0.1",
+                    num_workers=0,
+                    num_local_schedulers=1,
+                    worker_path=None,
+                    cleanup=True,
+                    redirect_output=False):
+  """Start Ray in local mode.
+
+  Args:
+    address_info (dict): A dictionary with address information for processes
+      that have already been started. If provided, address_info will be
+      modified to include processes that are newly started.
+    node_ip_address (str): The IP address of this node.
+    num_workers (int): The number of workers to start.
+    num_local_schedulers (int): The total number of local schedulers required.
+      This is also the total number of object stores required. This method will
+      start new instances of local schedulers and object stores until there are
+      at least num_local_schedulers existing instances of each, including ones
+      already registered with the given address_info.
+    worker_path (str): The path of the source code that will be run by the
+      worker.
+    cleanup (bool): If cleanup is true, then the processes started here will be
+      killed by services.cleanup() when the Python process that called this
+      method exits.
+    redirect_output (bool): True if stdout and stderr should be redirected to
+      /dev/null.
+
+  Returns:
+    A dictionary of the address information for the processes that were
+      started.
+  """
+  return start_ray_processes(address_info=address_info,
+                             node_ip_address=node_ip_address,
+                             num_workers=num_workers,
+                             num_local_schedulers=num_local_schedulers,
+                             worker_path=worker_path,
+                             cleanup=cleanup,
+                             redirect_output=redirect_output,
+                             include_global_scheduler=True)

python/ray/test/test_functions.py

@@ -0,0 +1,102 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import ray
+
+import numpy as np
+
+# Test simple functionality
+
+@ray.remote(num_return_vals=2)
+def handle_int(a, b):
+  return a + 1, b + 1
+
+# Test timing
+
+@ray.remote
+def empty_function():
+  pass
+
+@ray.remote
+def trivial_function():
+  return 1
+
+# Test keyword arguments
+
+@ray.remote
+def keyword_fct1(a, b="hello"):
+  return "{} {}".format(a, b)
+
+@ray.remote
+def keyword_fct2(a="hello", b="world"):
+  return "{} {}".format(a, b)
+
+@ray.remote
+def keyword_fct3(a, b, c="hello", d="world"):
+  return "{} {} {} {}".format(a, b, c, d)
+
+# Test variable numbers of arguments
+
+@ray.remote
+def varargs_fct1(*a):
+  return " ".join(map(str, a))
+
+@ray.remote
+def varargs_fct2(a, *b):
+  return " ".join(map(str, b))
+
+try:
+  @ray.remote
+  def kwargs_throw_exception(**c):
+    return ()
+  kwargs_exception_thrown = False
+except:
+  kwargs_exception_thrown = True
+
+try:
+  @ray.remote
+  def varargs_and_kwargs_throw_exception(a, b="hi", *c):
+    return "{} {} {}".format(a, b, c)
+  varargs_and_kwargs_exception_thrown = False
+except:
+  varargs_and_kwargs_exception_thrown = True
+
+# test throwing an exception
+
+@ray.remote
+def throw_exception_fct1():
+  raise Exception("Test function 1 intentionally failed.")
+
+@ray.remote
+def throw_exception_fct2():
+  raise Exception("Test function 2 intentionally failed.")
+
+@ray.remote(num_return_vals=3)
+def throw_exception_fct3(x):
+  raise Exception("Test function 3 intentionally failed.")
+
+# test Python mode
+
+@ray.remote
+def python_mode_f():
+  return np.array([0, 0])
+
+@ray.remote
+def python_mode_g(x):
+  x[0] = 1
+  return x
+
+# test no return values
+
+@ray.remote
+def no_op():
+  pass
+
+class TestClass(object):
+  def __init__(self):
+    self.a = 5
+
+@ray.remote
+def test_unknown_type():
+  return TestClass()

python/ray/workers/default_worker.py

@@ -0,0 +1,61 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import argparse
+import numpy as np
+import redis
+import traceback
+
+import ray
+
+parser = argparse.ArgumentParser(description="Parse addresses for the worker to connect to.")
+parser.add_argument("--node-ip-address", required=True, type=str, help="the ip address of the worker's node")
+parser.add_argument("--redis-address", required=True, type=str, help="the address to use for Redis")
+parser.add_argument("--object-store-name", required=True, type=str, help="the object store's name")
+parser.add_argument("--object-store-manager-name", required=True, type=str, help="the object store manager's name")
+parser.add_argument("--local-scheduler-name", required=True, type=str, help="the local scheduler's name")
+
+def random_string():
+  return np.random.bytes(20)
+
+if __name__ == "__main__":
+  args = parser.parse_args()
+  info = {"node_ip_address": args.node_ip_address,
+          "redis_address": args.redis_address,
+          "store_socket_name": args.object_store_name,
+          "manager_socket_name": args.object_store_manager_name,
+          "local_scheduler_socket_name": args.local_scheduler_name}
+  ray.worker.connect(info, ray.WORKER_MODE)
+
+  error_explanation = """
+This error is unexpected and should not have happened. Somehow a worker crashed
+in an unanticipated way causing the main_loop to throw an exception, which is
+being caught in "lib/python/ray/workers/default_worker.py".
+"""
+
+  while True:
+    try:
+      # This call to main_loop should never return if things are working. Most
+      # exceptions that are thrown (e.g., inside the execution of a task) should
+      # be caught and handled inside of the call to main_loop. If an exception
+      # is thrown here, then that means that there is some error that we didn't
+      # anticipate.
+      ray.worker.main_loop()
+    except Exception as e:
+      traceback_str = traceback.format_exc() + error_explanation
+      error_key = "WorkerError:{}".format(random_string())
+      redis_host, redis_port = args.redis_address.split(":")
+      # For this command to work, some other client (on the same machine as
+      # Redis) must have run "CONFIG SET protected-mode no".
+      redis_client = redis.StrictRedis(host=redis_host, port=int(redis_port))
+      redis_client.hmset(error_key, {"message": traceback_str,
+                                     "note": "This error is unexpected and should not have happened."})
+      redis_client.rpush("ErrorKeys", error_key)
+      # TODO(rkn): Note that if the worker was in the middle of executing a
+      # task, the any worker or driver that is blocking in a get call and
+      # waiting for the output of that task will hang. We need to address this.
+
+    # After putting the error message in Redis, this worker will attempt to
+    # reenter the main loop. TODO(rkn): We should probably reset it's state and
+    # call connect again.