Register Common.error with local scheduler extension module. (#1316)

* Register Common.error with local scheduler extension module.

* Add test.

python/ray/local_scheduler/__init__.py

@@ -4,9 +4,9 @@ from __future__ import print_function
 
 from ray.core.src.local_scheduler.liblocal_scheduler_library import (
     Task, LocalSchedulerClient, ObjectID, check_simple_value, task_from_string,
-    task_to_string, _config)
+    task_to_string, _config, common_error)
 from .local_scheduler_services import start_local_scheduler
 
 __all__ = ["Task", "LocalSchedulerClient", "ObjectID", "check_simple_value",
            "task_from_string", "task_to_string", "start_local_scheduler",
-           "_config"]
+           "_config", "common_error"]

src/local_scheduler/local_scheduler_extension.cc

@@ -287,10 +287,10 @@ MOD_INIT(liblocal_scheduler_library) {
 
   g_task_builder = make_task_builder();
 
-  char local_scheduler_error[] = "local_scheduler.error";
-  LocalSchedulerError = PyErr_NewException(local_scheduler_error, NULL, NULL);
-  Py_INCREF(LocalSchedulerError);
-  PyModule_AddObject(m, "local_scheduler_error", LocalSchedulerError);
+  char common_error[] = "common.error";
+  CommonError = PyErr_NewException(common_error, NULL, NULL);
+  Py_INCREF(CommonError);
+  PyModule_AddObject(m, "common_error", CommonError);
 
   Py_INCREF(&PyRayConfigType);
   PyModule_AddObject(m, "RayConfig", (PyObject *) &PyRayConfigType);

test/runtest.py

@@ -284,6 +284,24 @@ class SerializationTest(unittest.TestCase):
 
         ray.worker.cleanup()
 
+    def testPuttingObjectThatClosesOverObjectID(self):
+        # This test is here to prevent a regression of
+        # https://github.com/ray-project/ray/issues/1317.
+        ray.init(num_workers=0)
+
+        class Foo(object):
+            def __init__(self):
+                self.val = ray.put(0)
+
+            def method(self):
+                f
+
+        f = Foo()
+        with self.assertRaises(ray.local_scheduler.common_error):
+            ray.put(f)
+
+        ray.worker.cleanup()
+
 
 class WorkerTest(unittest.TestCase):
     def testPythonWorkers(self):