Add driver ID to task spec and add driver ID to Python error handling. (#225)

* Add driver ID to task spec and add driver ID to Python error handling.

* Make constants global variables.

* Add test for error isolation.

.travis.yml

@@ -77,3 +77,4 @@ script:
   - python test/microbenchmarks.py
   - python test/stress_tests.py
   - python test/component_failures_test.py
+  - python test/multi_node_test.py

python/common/test/test.py

@@ -17,6 +17,9 @@ def random_object_id():
 def random_function_id():
   return photon.ObjectID(np.random.bytes(ID_SIZE))
 
+def random_driver_id():
+  return photon.ObjectID(np.random.bytes(ID_SIZE))
+
 def random_task_id():
   return photon.ObjectID(np.random.bytes(ID_SIZE))
 
@@ -125,6 +128,7 @@ class TestTask(unittest.TestCase):
 
   def test_create_and_serialize_task(self):
     # TODO(rkn): The function ID should be a FunctionID object, not an ObjectID.
+    driver_id = random_driver_id()
     parent_id = random_task_id()
     function_id = random_function_id()
     object_ids = [random_object_id() for _ in range(256)]
@@ -156,7 +160,7 @@ class TestTask(unittest.TestCase):
     ]
     for args in args_list:
       for num_return_vals in [0, 1, 2, 3, 5, 10, 100]:
-        task = photon.Task(function_id, args, num_return_vals, parent_id, 0)
+        task = photon.Task(driver_id, function_id, args, num_return_vals, parent_id, 0)
         self.check_task(task, function_id, num_return_vals, args)
         data = photon.task_to_string(task)
         task2 = photon.task_from_string(data)

python/global_scheduler/test/test.py

@@ -34,6 +34,9 @@ TASK_STATUS_DONE = 16
 DB_CLIENT_PREFIX = "CL:"
 TASK_PREFIX = "TT:"
 
+def random_driver_id():
+  return photon.ObjectID(np.random.bytes(ID_SIZE))
+
 def random_task_id():
   return photon.ObjectID(np.random.bytes(ID_SIZE))
 
@@ -150,7 +153,7 @@ class TestGlobalScheduler(unittest.TestCase):
     # Sleep before submitting task to photon.
     time.sleep(0.1)
     # Submit a task to Redis.
-    task = photon.Task(random_function_id(), [photon.ObjectID(object_dep)], num_return_vals[0], random_task_id(), 0)
+    task = photon.Task(random_driver_id(), random_function_id(), [photon.ObjectID(object_dep)], num_return_vals[0], random_task_id(), 0)
     self.photon_client.submit(task)
     time.sleep(0.1)
     # There should now be a task in Redis, and it should get assigned to the
@@ -194,7 +197,7 @@ class TestGlobalScheduler(unittest.TestCase):
       if timesync:
         # Give 10ms for object info handler to fire (long enough to yield CPU).
         time.sleep(0.010)
-      task = photon.Task(random_function_id(), [photon.ObjectID(object_dep)], num_return_vals[0], random_task_id(), 0)
+      task = photon.Task(random_driver_id(), random_function_id(), [photon.ObjectID(object_dep)], num_return_vals[0], random_task_id(), 0)
       self.photon_client.submit(task)
     # Check that there are the correct number of tasks in Redis and that they
     # all get assigned to the local scheduler.

python/photon/test/test.py

@@ -21,6 +21,9 @@ ID_SIZE = 20
 def random_object_id():
   return photon.ObjectID(np.random.bytes(ID_SIZE))
 
+def random_driver_id():
+  return photon.ObjectID(np.random.bytes(ID_SIZE))
+
 def random_task_id():
   return photon.ObjectID(np.random.bytes(ID_SIZE))
 
@@ -94,7 +97,7 @@ class TestPhotonClient(unittest.TestCase):
 
     for args in args_list:
       for num_return_vals in [0, 1, 2, 3, 5, 10, 100]:
-        task = photon.Task(function_id, args, num_return_vals, random_task_id(), 0)
+        task = photon.Task(random_driver_id(), function_id, args, num_return_vals, random_task_id(), 0)
         # Submit a task.
         self.photon_client.submit(task)
         # Get the task.
@@ -113,7 +116,7 @@ class TestPhotonClient(unittest.TestCase):
     # Submit all of the tasks.
     for args in args_list:
       for num_return_vals in [0, 1, 2, 3, 5, 10, 100]:
-        task = photon.Task(function_id, args, num_return_vals, random_task_id(), 0)
+        task = photon.Task(random_driver_id(), function_id, args, num_return_vals, random_task_id(), 0)
         self.photon_client.submit(task)
     # Get all of the tasks.
     for args in args_list:
@@ -123,7 +126,7 @@ class TestPhotonClient(unittest.TestCase):
   def test_scheduling_when_objects_ready(self):
     # Create a task and submit it.
     object_id = random_object_id()
-    task = photon.Task(random_function_id(), [object_id], 0, random_task_id(), 0)
+    task = photon.Task(random_driver_id(), random_function_id(), [object_id], 0, random_task_id(), 0)
     self.photon_client.submit(task)
     # Launch a thread to get the task.
     def get_task():
@@ -143,7 +146,7 @@ class TestPhotonClient(unittest.TestCase):
     # Create a task with two dependencies and submit it.
     object_id1 = random_object_id()
     object_id2 = random_object_id()
-    task = photon.Task(random_function_id(), [object_id1, object_id2], 0, random_task_id(), 0)
+    task = photon.Task(random_driver_id(), random_function_id(), [object_id1, object_id2], 0, random_task_id(), 0)
     self.photon_client.submit(task)
 
     # Launch a thread to get the task.

python/ray/worker.py

@@ -35,6 +35,10 @@ LOG_POINT = 0
 LOG_SPAN_START = 1
 LOG_SPAN_END = 2
 
+ERROR_KEY_PREFIX = b"Error:"
+DRIVER_ID_LENGTH = 20
+ERROR_ID_LENGTH = 20
+
 def random_string():
   return np.random.bytes(20)
 
@@ -479,7 +483,8 @@ class Worker(object):
           args_for_photon.append(put(arg))
 
       # Submit the task to Photon.
-      task = photon.Task(photon.ObjectID(function_id.id()),
+      task = photon.Task(self.task_driver_id,
+                         photon.ObjectID(function_id.id()),
                          args_for_photon,
                          self.num_return_vals[function_id.id()],
                          self.current_task_id,
@@ -520,11 +525,30 @@ class Worker(object):
       counter = self.redis_client.hincrby(self.node_ip_address, key, 1) - 1
       function({"counter": counter})
       # Run the function on all workers.
-      self.redis_client.hmset(key, {"function_id": function_to_run_id,
+      self.redis_client.hmset(key, {"driver_id": self.task_driver_id.id(),
+                                    "function_id": function_to_run_id,
                                     "function": pickling.dumps(function)})
       self.redis_client.rpush("Exports", key)
       self.driver_export_counter += 1
 
+  def push_error_to_driver(self, driver_id, error_type, message, data=None):
+    """Push an error message to the driver to be printed in the background.
+
+    Args:
+      driver_id: The ID of the driver to push the error message to.
+      error_type (str): The type of the error.
+      message (str): The message that will be printed in the background on the
+        driver.
+      data: This should be a dictionary mapping strings to strings. It will be
+        serialized with json and stored in Redis.
+    """
+    error_key = ERROR_KEY_PREFIX + driver_id + b":" + random_string()
+    data = {} if data is None else data
+    self.redis_client.hmset(error_key, {"type": error_type,
+                                        "message": message,
+                                        "data": data})
+    self.redis_client.rpush("ErrorKeys", error_key)
+
 global_worker = Worker()
 """Worker: The global Worker object for this worker process.
 
@@ -578,24 +602,29 @@ def print_failed_task(task_status):
       Error Message: \n{}
   """.format(task_status["function_name"], task_status["operationid"], task_status["error_message"]))
 
+def error_applies_to_driver(error_key, worker=global_worker):
+  """Return True if the error is for this driver and false otherwise."""
+  # TODO(rkn): Should probably check that this is only called on a driver.
+  # Check that the error key is formatted as in push_error_to_driver.
+  assert len(error_key) == len(ERROR_KEY_PREFIX) + DRIVER_ID_LENGTH + 1 + ERROR_ID_LENGTH, error_key
+  # If the driver ID in the error message is a sequence of all zeros, then the
+  # message is intended for all drivers.
+  generic_driver_id = DRIVER_ID_LENGTH * b"\x00"
+  driver_id = error_key[len(ERROR_KEY_PREFIX):(len(ERROR_KEY_PREFIX) + DRIVER_ID_LENGTH)]
+  return driver_id == worker.task_driver_id.id() or driver_id == generic_driver_id
+
 def error_info(worker=global_worker):
   """Return information about failed tasks."""
   check_connected(worker)
   check_main_thread()
-  result = {b"TaskError": [],
-            b"RemoteFunctionImportError": [],
-            b"EnvironmentVariableImportError": [],
-            b"EnvironmentVariableReinitializeError": [],
-            b"FunctionToRunError": [],
-            b"GenericWarning": [],
-            }
   error_keys = worker.redis_client.lrange("ErrorKeys", 0, -1)
+  errors = []
   for error_key in error_keys:
-    error_type = error_key.split(b":", 1)[0]
-    error_contents = worker.redis_client.hgetall(error_key)
-    result[error_type].append(error_contents)
+    if error_applies_to_driver(error_key, worker=worker):
+      error_contents = worker.redis_client.hgetall(error_key)
+      errors.append(error_contents)
 
-  return result
+  return errors
 
 def initialize_numbuf(worker=global_worker):
   """Initialize the serialization library.
@@ -866,25 +895,27 @@ If this driver is hanging, start a new one with
   # error_message_pubsub_client.psubscribe and before the call to
   # error_message_pubsub_client.listen will still be processed in the loop.
   worker.error_message_pubsub_client.psubscribe("__keyspace@0__:ErrorKeys")
-  num_errors_printed = 0
+  num_errors_received = 0
 
   # Get the exports that occurred before the call to psubscribe.
   with worker.lock:
     error_keys = worker.redis_client.lrange("ErrorKeys", 0, -1)
     for error_key in error_keys:
-      error_message = worker.redis_client.hget(error_key, "message").decode("ascii")
-      print(error_message)
-      print(helpful_message)
-      num_errors_printed += 1
+      if error_applies_to_driver(error_key, worker=worker):
+        error_message = worker.redis_client.hget(error_key, "message").decode("ascii")
+        print(error_message)
+        print(helpful_message)
+      num_errors_received += 1
 
   try:
     for msg in worker.error_message_pubsub_client.listen():
       with worker.lock:
-        for error_key in worker.redis_client.lrange("ErrorKeys", num_errors_printed, -1):
-          error_message = worker.redis_client.hget(error_key, "message").decode("ascii")
-          print(error_message)
-          print(helpful_message)
-          num_errors_printed += 1
+        for error_key in worker.redis_client.lrange("ErrorKeys", num_errors_received, -1):
+          if error_applies_to_driver(error_key, worker=worker):
+            error_message = worker.redis_client.hget(error_key, "message").decode("ascii")
+            print(error_message)
+            print(helpful_message)
+          num_errors_received += 1
   except redis.ConnectionError:
     # When Redis terminates the listen call will throw a ConnectionError, which
     # we catch here.
@@ -892,7 +923,7 @@ If this driver is hanging, start a new one with
 
 def fetch_and_register_remote_function(key, worker=global_worker):
   """Import a remote function."""
-  function_id_str, function_name, serialized_function, num_return_vals, module, function_export_counter = worker.redis_client.hmget(key, ["function_id", "name", "function", "num_return_vals", "module", "function_export_counter"])
+  driver_id, function_id_str, function_name, serialized_function, num_return_vals, module, function_export_counter = worker.redis_client.hmget(key, ["driver_id", "function_id", "name", "function", "num_return_vals", "module", "function_export_counter"])
   function_id = photon.ObjectID(function_id_str)
   function_name = function_name.decode("ascii")
   num_return_vals = int(num_return_vals)
@@ -915,11 +946,10 @@ def fetch_and_register_remote_function(key, worker=global_worker):
     # record the traceback and notify the scheduler of the failure.
     traceback_str = format_error_message(traceback.format_exc())
     # Log the error message.
-    error_key = "RemoteFunctionImportError:{}".format(function_id.id())
-    worker.redis_client.hmset(error_key, {"function_id": function_id.id(),
-                                          "function_name": function_name,
-                                          "message": traceback_str})
-    worker.redis_client.rpush("ErrorKeys", error_key)
+    worker.push_error_to_driver(driver_id, "register_remote_function",
+                                traceback_str,
+                                data={"function_id": function_id.id(),
+                                      "function_name": function_name})
   else:
     # TODO(rkn): Why is the below line necessary?
     function.__module__ = module
@@ -929,7 +959,7 @@ def fetch_and_register_remote_function(key, worker=global_worker):
 
 def fetch_and_register_environment_variable(key, worker=global_worker):
   """Import an environment variable."""
-  environment_variable_name, serialized_initializer, serialized_reinitializer = worker.redis_client.hmget(key, ["name", "initializer", "reinitializer"])
+  driver_id, environment_variable_name, serialized_initializer, serialized_reinitializer = worker.redis_client.hmget(key, ["driver_id", "name", "initializer", "reinitializer"])
   environment_variable_name = environment_variable_name.decode("ascii")
   try:
     initializer = pickling.loads(serialized_initializer)
@@ -940,14 +970,13 @@ def fetch_and_register_environment_variable(key, worker=global_worker):
     # record the traceback and notify the scheduler of the failure.
     traceback_str = format_error_message(traceback.format_exc())
     # Log the error message.
-    error_key = "EnvironmentVariableImportError:{}".format(random_string())
-    worker.redis_client.hmset(error_key, {"name": environment_variable_name,
-                                          "message": traceback_str})
-    worker.redis_client.rpush("ErrorKeys", error_key)
+    worker.push_error_to_driver(driver_id, "register_environment_variable",
+                                traceback_str,
+                                data={"name": environment_variable_name})
 
 def fetch_and_execute_function_to_run(key, worker=global_worker):
   """Run on arbitrary function on the worker."""
-  serialized_function, = worker.redis_client.hmget(key, ["function"])
+  driver_id, serialized_function = worker.redis_client.hmget(key, ["driver_id", "function"])
   # Get the number of workers on this node that have already started executing
   # this remote function, and increment that value. Subtract 1 so the counter
   # starts at 0.
@@ -963,10 +992,8 @@ def fetch_and_execute_function_to_run(key, worker=global_worker):
     traceback_str = traceback.format_exc()
     # Log the error message.
     name = function.__name__  if "function" in locals() and hasattr(function, "__name__") else ""
-    error_key = "FunctionToRunError:{}".format(random_string())
-    worker.redis_client.hmset(error_key, {"name": name,
-                                          "message": traceback_str})
-    worker.redis_client.rpush("ErrorKeys", error_key)
+    worker.push_error_to_driver(driver_id, "function_to_run", traceback_str,
+                                data={"name": name})
 
 def import_thread(worker):
   worker.import_pubsub_client = worker.redis_client.pubsub()
@@ -1062,7 +1089,8 @@ def connect(info, object_id_seed=None, mode=WORKER_MODE, worker=global_worker):
     worker.redis_client.hmset(b"Workers:" + worker.worker_id, {"node_ip_address": worker.node_ip_address})
   else:
     raise Exception("This code should be unreachable.")
-  # If this is a driver, set the current task ID and set the task index to 0.
+  # If this is a driver, set the current task ID, the task driver ID, and set
+  # the task index to 0.
   if mode in [SCRIPT_MODE, SILENT_MODE]:
     # If the user provided an object_id_seed, then set the current task ID
     # deterministically based on that seed (without altering the state of the
@@ -1075,6 +1103,11 @@ def connect(info, object_id_seed=None, mode=WORKER_MODE, worker=global_worker):
       # Try to use true randomness.
       np.random.seed(None)
     worker.current_task_id = photon.ObjectID(np.random.bytes(20))
+    # When tasks are executed on remote workers in the context of multiple
+    # drivers, the task driver ID is used to keep track of which driver is
+    # responsible for the task so that error messages will be propagated to the
+    # correct driver.
+    worker.task_driver_id = photon.ObjectID(worker.worker_id)
     # Reset the state of the numpy random number generator.
     np.random.set_state(numpy_state)
     # Set other fields needed for computing task IDs.
@@ -1328,7 +1361,7 @@ def wait(object_ids, num_returns=1, timeout=None, worker=global_worker):
     remaining_ids = [photon.ObjectID(object_id) for object_id in remaining_ids]
     return ready_ids, remaining_ids
 
-def wait_for_valid_import_counter(function_id, timeout=5, worker=global_worker):
+def wait_for_valid_import_counter(function_id, driver_id, timeout=5, worker=global_worker):
   """Wait until this worker has imported enough to execute the function.
 
   This method will simply loop until the import thread has imported enough of
@@ -1338,6 +1371,8 @@ def wait_for_valid_import_counter(function_id, timeout=5, worker=global_worker):
 
   Args:
     function_id (str): The ID of the function that we want to execute.
+    driver_id (str): The ID of the driver to push the error message to if this
+      times out.
   """
   start_time = time.time()
   # Only send the warning once.
@@ -1353,7 +1388,8 @@ def wait_for_valid_import_counter(function_id, timeout=5, worker=global_worker):
       else:
         warning_message = "This worker's import counter is too small."
       if not warning_sent:
-        push_warning_to_user(warning_message, worker=worker)
+        worker.push_error_to_driver(driver_id, "import_counter",
+                                    warning_message)
       warning_sent = True
     time.sleep(0.001)
 
@@ -1400,6 +1436,10 @@ def main_loop(worker=global_worker):
     were accessed by the task.
     """
     try:
+      # The ID of the driver that this task belongs to. This is needed so that
+      # if the task throws an exception, we propagate the error message to the
+      # correct driver.
+      worker.task_driver_id = task.driver_id()
       worker.current_task_id = task.task_id()
       worker.task_index = 0
       worker.put_index = 0
@@ -1440,11 +1480,10 @@ def main_loop(worker=global_worker):
       failure_objects = [failure_object for _ in range(len(return_object_ids))]
       store_outputs_in_objstore(return_object_ids, failure_objects, worker)
       # Log the error message.
-      error_key = "TaskError:{}".format(random_string())
-      worker.redis_client.hmset(error_key, {"function_id": function_id.id(),
-                                            "function_name": function_name,
-                                            "message": str(failure_object)})
-      worker.redis_client.rpush("ErrorKeys", error_key)
+      worker.push_error_to_driver(worker.task_driver_id.id(), "task",
+                                  str(failure_object),
+                                  data={"function_id": function_id.id(),
+                                        "function_name": function_name})
     try:
       # Reinitialize the values of environment variables that were used in the
       # task above so that changes made to their state do not affect other tasks.
@@ -1454,12 +1493,11 @@ def main_loop(worker=global_worker):
       # The attempt to reinitialize the environment variables threw an
       # exception. We record the traceback and notify the scheduler.
       traceback_str = format_error_message(traceback.format_exc())
-      error_key = "EnvironmentVariableReinitializeError:{}".format(random_string())
-      worker.redis_client.hmset(error_key, {"task_id": "NOTIMPLEMENTED",
-                                            "function_id": function_id.id(),
-                                            "function_name": function_name,
-                                            "message": traceback_str})
-      worker.redis_client.rpush("ErrorKeys", error_key)
+      worker.push_error_to_driver(worker.task_driver_id.id(),
+                                  "reinitialize_environment_variable",
+                                  traceback_str,
+                                  data={"function_id": function_id.id(),
+                                        "function_name": function_name})
 
   check_main_thread()
   while True:
@@ -1471,7 +1509,7 @@ def main_loop(worker=global_worker):
     # export counter for the task. If not, wait until we have imported enough.
     # We will push warnings to the user if we spend too long in this loop.
     with log_span("ray:wait_for_import_counter", worker=worker):
-      wait_for_valid_import_counter(function_id, worker=worker)
+      wait_for_valid_import_counter(function_id, task.driver_id().id(), worker=worker)
 
     # Execute the task.
     # TODO(rkn): Consider acquiring this lock with a timeout and pushing a
@@ -1490,11 +1528,6 @@ def main_loop(worker=global_worker):
     # Push all of the log events to the global state store.
     flush_log()
 
-def push_warning_to_user(message, worker=global_worker):
-  error_key = "GenericWarning:{}".format(random_string())
-  worker.redis_client.hmset(error_key, {"message": message})
-  worker.redis_client.rpush("ErrorKeys", error_key)
-
 def _submit_task(function_id, func_name, args, worker=global_worker):
   """This is a wrapper around worker.submit_task.
 
@@ -1538,7 +1571,8 @@ def _export_environment_variable(name, environment_variable, worker=global_worke
     raise Exception("_export_environment_variable can only be called on a driver.")
   environment_variable_id = name
   key = "EnvironmentVariables:{}".format(environment_variable_id)
-  worker.redis_client.hmset(key, {"name": name,
+  worker.redis_client.hmset(key, {"driver_id": worker.task_driver_id.id(),
+                                  "name": name,
                                   "initializer": pickling.dumps(environment_variable.initializer),
                                   "reinitializer": pickling.dumps(environment_variable.reinitializer)})
   worker.redis_client.rpush("Exports", key)
@@ -1551,7 +1585,8 @@ def export_remote_function(function_id, func_name, func, num_return_vals, worker
   key = "RemoteFunction:{}".format(function_id.id())
   worker.num_return_vals[function_id.id()] = num_return_vals
   pickled_func = pickling.dumps(func)
-  worker.redis_client.hmset(key, {"function_id": function_id.id(),
+  worker.redis_client.hmset(key, {"driver_id": worker.task_driver_id.id(),
+                                  "function_id": function_id.id(),
                                   "name": func_name,
                                   "module": func.__module__,
                                   "function": pickled_func,

python/ray/workers/default_worker.py

@@ -44,12 +44,16 @@ being caught in "lib/python/ray/workers/default_worker.py".
       ray.worker.main_loop()
     except Exception as e:
       traceback_str = traceback.format_exc() + error_explanation
-      error_key = "WorkerError:{}".format(random_string())
+      DRIVER_ID_LENGTH = 20
+      # We use a driver ID of all zeros to push an error message to all drivers.
+      driver_id = DRIVER_ID_LENGTH * b"\x00"
+      error_key = b"Error:" + driver_id + b":" + 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,
+      redis_client.hmset(error_key, {"type": "worker_crash",
+                                     "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

src/common/lib/python/common_extension.c

@@ -253,6 +253,7 @@ PyTypeObject PyObjectIDType = {
 /* Define the PyTask class. */
 
 static int PyTask_init(PyTask *self, PyObject *args, PyObject *kwds) {
+  unique_id driver_id;
   function_id function_id;
   /* Arguments of the task (can be PyObjectIDs or Python values). */
   PyObject *arguments;
@@ -264,9 +265,10 @@ static int PyTask_init(PyTask *self, PyObject *args, PyObject *kwds) {
   task_id parent_task_id;
   /* The number of tasks that the parent task has called prior to this one. */
   int parent_counter;
-  if (!PyArg_ParseTuple(args, "O&OiO&i", &PyObjectToUniqueID, &function_id,
-                        &arguments, &num_returns, &PyObjectToUniqueID,
-                        &parent_task_id, &parent_counter)) {
+  if (!PyArg_ParseTuple(args, "O&O&OiO&i", &PyObjectToUniqueID, &driver_id,
+                        &PyObjectToUniqueID, &function_id, &arguments,
+                        &num_returns, &PyObjectToUniqueID, &parent_task_id,
+                        &parent_counter)) {
     return -1;
   }
   Py_ssize_t size = PyList_Size(arguments);
@@ -285,9 +287,9 @@ static int PyTask_init(PyTask *self, PyObject *args, PyObject *kwds) {
   }
   /* Construct the task specification. */
   int val_repr_index = 0;
-  self->spec =
-      start_construct_task_spec(parent_task_id, parent_counter, function_id,
-                                size, num_returns, value_data_bytes);
+  self->spec = start_construct_task_spec(driver_id, parent_task_id,
+                                         parent_counter, function_id, size,
+                                         num_returns, value_data_bytes);
   /* Add the task arguments. */
   for (Py_ssize_t i = 0; i < size; ++i) {
     PyObject *arg = PyList_GetItem(arguments, i);
@@ -320,6 +322,11 @@ static PyObject *PyTask_function_id(PyObject *self) {
   return PyObjectID_make(function_id);
 }
 
+static PyObject *PyTask_driver_id(PyObject *self) {
+  unique_id driver_id = task_spec_driver_id(((PyTask *) self)->spec);
+  return PyObjectID_make(driver_id);
+}
+
 static PyObject *PyTask_task_id(PyObject *self) {
   task_id task_id = task_spec_id(((PyTask *) self)->spec);
   return PyObjectID_make(task_id);
@@ -362,6 +369,8 @@ static PyObject *PyTask_returns(PyObject *self) {
 static PyMethodDef PyTask_methods[] = {
     {"function_id", (PyCFunction) PyTask_function_id, METH_NOARGS,
      "Return the function ID for this task."},
+    {"driver_id", (PyCFunction) PyTask_driver_id, METH_NOARGS,
+     "Return the driver ID for this task."},
     {"task_id", (PyCFunction) PyTask_task_id, METH_NOARGS,
      "Return the task ID for this task."},
     {"arguments", (PyCFunction) PyTask_arguments, METH_NOARGS,

src/common/task.c

@@ -35,6 +35,8 @@ typedef struct {
 } task_arg;
 
 struct task_spec_impl {
+  /** ID of the driver that created this task. */
+  unique_id driver_id;
   /** Task ID of the task. */
   task_id task_id;
   /** Task ID of the parent task. */
@@ -136,7 +138,8 @@ object_id task_compute_put_id(task_id task_id, int64_t put_index) {
   return put_id;
 }
 
-task_spec *start_construct_task_spec(task_id parent_task_id,
+task_spec *start_construct_task_spec(unique_id driver_id,
+                                     task_id parent_task_id,
                                      int64_t parent_counter,
                                      function_id function_id,
                                      int64_t num_args,
@@ -145,6 +148,7 @@ task_spec *start_construct_task_spec(task_id parent_task_id,
   int64_t size = TASK_SPEC_SIZE(num_args, num_returns, args_value_size);
   task_spec *task = malloc(size);
   memset(task, 0, size);
+  task->driver_id = driver_id;
   task->task_id = NIL_TASK_ID;
   task->parent_task_id = parent_task_id;
   task->parent_counter = parent_counter;
@@ -171,7 +175,7 @@ void finish_construct_task_spec(task_spec *spec) {
 
 task_spec *alloc_nil_task_spec(task_id task_id) {
   task_spec *spec =
-      start_construct_task_spec(NIL_ID, 0, NIL_FUNCTION_ID, 0, 0, 0);
+      start_construct_task_spec(NIL_ID, NIL_ID, 0, NIL_FUNCTION_ID, 0, 0, 0);
   finish_construct_task_spec(spec);
   spec->task_id = task_id;
   return spec;
@@ -188,6 +192,12 @@ function_id task_function(task_spec *spec) {
   return spec->function_id;
 }
 
+unique_id task_spec_driver_id(task_spec *spec) {
+  /* Check that the task has been constructed. */
+  DCHECK(!task_ids_equal(spec->task_id, NIL_TASK_ID));
+  return spec->driver_id;
+}
+
 task_id task_spec_id(task_spec *spec) {
   /* Check that the task has been constructed. */
   DCHECK(!task_ids_equal(spec->task_id, NIL_TASK_ID));

src/common/task.h

@@ -78,6 +78,8 @@ bool function_id_is_nil(function_id id);
  * Begin constructing a task_spec. After this is called, the arguments must be
  * added to the task_spec and then finish_construct_task_spec must be called.
  *
+ * @param driver_id The ID of the driver whose job is responsible for the
+ *        creation of this task.
  * @param parent_task_id The task ID of the task that submitted this task.
  * @param parent_counter A counter indicating how many tasks were submitted by
  *        the parent task prior to this one.
@@ -88,7 +90,8 @@ bool function_id_is_nil(function_id id);
           ignoring object ID arguments.
  * @return The partially constructed task_spec.
  */
-task_spec *start_construct_task_spec(task_id parent_task_id,
+task_spec *start_construct_task_spec(unique_id driver_id,
+                                     task_id parent_task_id,
                                      int64_t parent_counter,
                                      function_id function_id,
                                      int64_t num_args,
@@ -121,6 +124,14 @@ int64_t task_spec_size(task_spec *spec);
  */
 function_id task_function(task_spec *spec);
 
+/**
+ * Return the driver ID of the task.
+ *
+ * @param spec The task_spec in question.
+ * @return The driver ID of the task.
+ */
+unique_id task_spec_driver_id(task_spec *spec);
+
 /**
  * Return the task ID of the task.
  *

src/common/test/task_tests.c

@@ -15,7 +15,7 @@ TEST task_test(void) {
   task_id parent_task_id = globally_unique_id();
   function_id func_id = globally_unique_id();
   task_spec *spec =
-      start_construct_task_spec(parent_task_id, 0, func_id, 4, 2, 10);
+      start_construct_task_spec(NIL_ID, parent_task_id, 0, func_id, 4, 2, 10);
   ASSERT(task_num_args(spec) == 4);
   ASSERT(task_num_returns(spec) == 2);
 
@@ -53,14 +53,14 @@ TEST deterministic_ids_test(void) {
 
   /* Construct a first task. */
   task_spec *spec1 =
-      start_construct_task_spec(parent_task_id, 0, func_id, 2, 3, 11);
+      start_construct_task_spec(NIL_ID, parent_task_id, 0, func_id, 2, 3, 11);
   task_args_add_ref(spec1, arg1);
   task_args_add_val(spec1, arg2, 11);
   finish_construct_task_spec(spec1);
 
   /* Construct a second identical task. */
   task_spec *spec2 =
-      start_construct_task_spec(parent_task_id, 0, func_id, 2, 3, 11);
+      start_construct_task_spec(NIL_ID, parent_task_id, 0, func_id, 2, 3, 11);
   task_args_add_ref(spec2, arg1);
   task_args_add_val(spec2, arg2, 11);
   finish_construct_task_spec(spec2);
@@ -78,21 +78,21 @@ TEST deterministic_ids_test(void) {
   /* Create more tasks that are only mildly different. */
 
   /* Construct a task with a different parent task ID. */
-  task_spec *spec3 =
-      start_construct_task_spec(globally_unique_id(), 0, func_id, 2, 3, 11);
+  task_spec *spec3 = start_construct_task_spec(NIL_ID, globally_unique_id(), 0,
+                                               func_id, 2, 3, 11);
   task_args_add_ref(spec3, arg1);
   task_args_add_val(spec3, arg2, 11);
   finish_construct_task_spec(spec3);
 
   /* Construct a task with a different parent counter. */
   task_spec *spec4 =
-      start_construct_task_spec(parent_task_id, 1, func_id, 2, 3, 11);
+      start_construct_task_spec(NIL_ID, parent_task_id, 1, func_id, 2, 3, 11);
   task_args_add_ref(spec4, arg1);
   task_args_add_val(spec4, arg2, 11);
   finish_construct_task_spec(spec4);
 
   /* Construct a task with a different function ID. */
-  task_spec *spec5 = start_construct_task_spec(parent_task_id, 0,
+  task_spec *spec5 = start_construct_task_spec(NIL_ID, parent_task_id, 0,
                                                globally_unique_id(), 2, 3, 11);
   task_args_add_ref(spec5, arg1);
   task_args_add_val(spec5, arg2, 11);
@@ -100,14 +100,14 @@ TEST deterministic_ids_test(void) {
 
   /* Construct a task with a different object ID argument. */
   task_spec *spec6 =
-      start_construct_task_spec(parent_task_id, 0, func_id, 2, 3, 11);
+      start_construct_task_spec(NIL_ID, parent_task_id, 0, func_id, 2, 3, 11);
   task_args_add_ref(spec6, globally_unique_id());
   task_args_add_val(spec6, arg2, 11);
   finish_construct_task_spec(spec6);
 
   /* Construct a task with a different value argument. */
   task_spec *spec7 =
-      start_construct_task_spec(parent_task_id, 0, func_id, 2, 3, 11);
+      start_construct_task_spec(NIL_ID, parent_task_id, 0, func_id, 2, 3, 11);
   task_args_add_ref(spec7, arg1);
   task_args_add_val(spec7, (uint8_t *) "hello_world", 11);
   finish_construct_task_spec(spec7);
@@ -149,7 +149,7 @@ TEST send_task(void) {
   task_id parent_task_id = globally_unique_id();
   function_id func_id = globally_unique_id();
   task_spec *spec =
-      start_construct_task_spec(parent_task_id, 0, func_id, 4, 2, 10);
+      start_construct_task_spec(NIL_ID, parent_task_id, 0, func_id, 4, 2, 10);
   task_args_add_ref(spec, globally_unique_id());
   task_args_add_val(spec, (uint8_t *) "Hello", 5);
   task_args_add_val(spec, (uint8_t *) "World", 5);

src/common/test/test_common.h

@@ -21,8 +21,9 @@ static inline task_spec *example_task_spec_with_args(int64_t num_args,
                                                      object_id arg_ids[]) {
   task_id parent_task_id = globally_unique_id();
   function_id func_id = globally_unique_id();
-  task_spec *task = start_construct_task_spec(
-      parent_task_id, 0, func_id, num_args, num_returns, arg_value_size);
+  task_spec *task =
+      start_construct_task_spec(NIL_ID, parent_task_id, 0, func_id, num_args,
+                                num_returns, arg_value_size);
   for (int64_t i = 0; i < num_args; ++i) {
     object_id arg_id;
     if (arg_ids == NULL) {

test/failure_test.py

@@ -12,11 +12,13 @@ if sys.version_info >= (3, 0):
 
 import ray.test.test_functions as test_functions
 
+def relevant_errors(error_type):
+  return [info for info in ray.error_info() if info[b"type"] == error_type]
+
 def wait_for_errors(error_type, num_errors, timeout=10):
   start_time = time.time()
   while time.time() - start_time < timeout:
-    error_info = ray.error_info()
-    if len(error_info[error_type]) >= num_errors:
+    if len(relevant_errors(error_type)) >= num_errors:
       return
     time.sleep(0.1)
   print("Timing out of wait.")
@@ -27,9 +29,9 @@ class FailureTest(unittest.TestCase):
     ray.init(num_workers=1, driver_mode=ray.SILENT_MODE)
 
     test_functions.test_unknown_type.remote()
-    wait_for_errors(b"TaskError", 1)
+    wait_for_errors(b"task", 1)
     error_info = ray.error_info()
-    self.assertEqual(len(error_info[b"TaskError"]), 1)
+    self.assertEqual(len(relevant_errors(b"task")), 1)
 
     ray.worker.cleanup()
 
@@ -61,10 +63,10 @@ class TaskStatusTest(unittest.TestCase):
 
     test_functions.throw_exception_fct1.remote()
     test_functions.throw_exception_fct1.remote()
-    wait_for_errors(b"TaskError", 2)
+    wait_for_errors(b"task", 2)
     result = ray.error_info()
-    self.assertEqual(len(result[b"TaskError"]), 2)
-    for task in result[b"TaskError"]:
+    self.assertEqual(len(relevant_errors(b"task")), 2)
+    for task in relevant_errors(b"task"):
       self.assertTrue(b"Test function 1 intentionally failed." in task.get(b"message"))
 
     x = test_functions.throw_exception_fct2.remote()
@@ -104,8 +106,8 @@ class TaskStatusTest(unittest.TestCase):
       def __call__(self):
         return
     f = ray.remote(Foo())
-    wait_for_errors(b"RemoteFunctionImportError", 2)
-    self.assertTrue(b"There is a problem here." in ray.error_info()[b"RemoteFunctionImportError"][0][b"message"])
+    wait_for_errors(b"register_remote_function", 2)
+    self.assertTrue(b"There is a problem here." in ray.error_info()[0][b"message"])
 
     # Check that if we try to call the function it throws an exception and does
     # not hang.
@@ -124,9 +126,9 @@ class TaskStatusTest(unittest.TestCase):
         raise Exception("The initializer failed.")
       return 0
     ray.env.foo = ray.EnvironmentVariable(initializer)
-    wait_for_errors(b"EnvironmentVariableImportError", 2)
+    wait_for_errors(b"register_environment_variable", 2)
     # Check that the error message is in the task info.
-    self.assertTrue(b"The initializer failed." in ray.error_info()[b"EnvironmentVariableImportError"][0][b"message"])
+    self.assertTrue(b"The initializer failed." in ray.error_info()[0][b"message"])
 
     ray.worker.cleanup()
 
@@ -142,9 +144,9 @@ class TaskStatusTest(unittest.TestCase):
     def use_foo():
       ray.env.foo
     use_foo.remote()
-    wait_for_errors(b"EnvironmentVariableReinitializeError", 1)
+    wait_for_errors(b"reinitialize_environment_variable", 1)
     # Check that the error message is in the task info.
-    self.assertTrue(b"The reinitializer failed." in ray.error_info()[b"EnvironmentVariableReinitializeError"][0][b"message"])
+    self.assertTrue(b"The reinitializer failed." in ray.error_info()[0][b"message"])
 
     ray.worker.cleanup()
 
@@ -155,11 +157,11 @@ class TaskStatusTest(unittest.TestCase):
       if ray.worker.global_worker.mode == ray.WORKER_MODE:
         raise Exception("Function to run failed.")
     ray.worker.global_worker.run_function_on_all_workers(f)
-    wait_for_errors(b"FunctionToRunError", 2)
+    wait_for_errors(b"function_to_run", 2)
     # Check that the error message is in the task info.
-    self.assertEqual(len(ray.error_info()[b"FunctionToRunError"]), 2)
-    self.assertTrue(b"Function to run failed." in ray.error_info()[b"FunctionToRunError"][0][b"message"])
-    self.assertTrue(b"Function to run failed." in ray.error_info()[b"FunctionToRunError"][1][b"message"])
+    self.assertEqual(len(ray.error_info()), 2)
+    self.assertTrue(b"Function to run failed." in ray.error_info()[0][b"message"])
+    self.assertTrue(b"Function to run failed." in ray.error_info()[1][b"message"])
 
     ray.worker.cleanup()
 

test/multi_node_test.py

@@ -0,0 +1,101 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import numpy as np
+import os
+import unittest
+import ray
+import subprocess
+import sys
+import tempfile
+import time
+
+start_ray_script = os.path.join(os.path.dirname(os.path.abspath(__file__)), "../scripts/start_ray.sh")
+stop_ray_script = os.path.join(os.path.dirname(os.path.abspath(__file__)), "../scripts/stop_ray.sh")
+
+class MultiNodeTest(unittest.TestCase):
+
+  def testErrorIsolation(self):
+    # Start the Ray processes on this machine.
+    out = subprocess.check_output([start_ray_script, "--head"]).decode("ascii")
+    # Get the redis address from the output.
+    redis_substring_prefix = "redis_address=\""
+    redis_address_location = out.find(redis_substring_prefix) + len(redis_substring_prefix)
+    redis_address = out[redis_address_location:]
+    redis_address = redis_address.split("\"")[0]
+    # Connect a driver to the Ray cluster.
+    ray.init(redis_address=redis_address, driver_mode=ray.SILENT_MODE)
+
+    # There shouldn't be any errors yet.
+    self.assertEqual(len(ray.error_info()), 0)
+
+    error_string1 = "error_string1"
+    error_string2 = "error_string2"
+
+    @ray.remote
+    def f():
+      raise Exception(error_string1)
+
+    # Run a remote function that throws an error.
+    with self.assertRaises(Exception):
+      ray.get(f.remote())
+
+    # Wait for the error to appear in Redis.
+    while len(ray.error_info()) != 1:
+      time.sleep(0.1)
+      print("Waiting for error to appear.")
+
+    # Make sure we got the error.
+    self.assertEqual(len(ray.error_info()), 1)
+    self.assertIn(error_string1, ray.error_info()[0][b"message"].decode("ascii"))
+
+    # Start another driver and make sure that it does not receive this error.
+    # Make the other driver throw an error, and make sure it receives that
+    # error.
+    driver_script = """
+import ray
+import time
+
+ray.init(redis_address="{}")
+
+time.sleep(1)
+assert len(ray.error_info()) == 0
+
+@ray.remote
+def f():
+  raise Exception("{}")
+
+try:
+  ray.get(f.remote())
+except Exception as e:
+  pass
+
+while len(ray.error_info()) != 1:
+  print(len(ray.error_info()))
+  time.sleep(0.1)
+assert len(ray.error_info()) == 1
+
+assert "{}" in ray.error_info()[0][b"message"].decode("ascii")
+
+print("success")
+""".format(redis_address, error_string2, error_string2)
+
+    # Save the driver script as a file so we can call it using subprocess.
+    with tempfile.NamedTemporaryFile() as f:
+      f.write(driver_script.encode("ascii"))
+      f.flush()
+      out = subprocess.check_output(["python", f.name]).decode("ascii")
+
+    # Make sure the other driver succeeded.
+    self.assertIn("success", out)
+
+    # Make sure that the other error message doesn't show up for this driver.
+    self.assertEqual(len(ray.error_info()), 1)
+    self.assertIn(error_string1, ray.error_info()[0][b"message"].decode("ascii"))
+
+    ray.worker.cleanup()
+    subprocess.Popen([stop_ray_script]).wait()
+
+if __name__ == "__main__":
+  unittest.main(verbosity=2)