Dynamically grow worker pool to partially solve hanging workloads (#286)

* First pass at a policy to solve deadlock

* Address Robert's comments

* stress test

* unit test

* Fix test cases

* Fix test for python3

* add more logging

* White space.

test/runtest.py

@@ -367,7 +367,7 @@ class APITest(unittest.TestCase):
     ray.worker.cleanup()
 
   def testWait(self):
-    ray.init(num_workers=1)
+    ray.init(num_workers=1, num_cpus=1)
 
     @ray.remote
     def f(delay):
@@ -1115,6 +1115,55 @@ class ResourcesTest(unittest.TestCase):
 
     ray.worker.cleanup()
 
+class WorkerPoolTests(unittest.TestCase):
+
+  def tearDown(self):
+    ray.worker.cleanup()
+
+  def testNoWorkers(self):
+    ray.init(num_workers=0)
+
+    @ray.remote
+    def f():
+      return 1
+
+    # Make sure we can call a remote function. This will require starting a new
+    # worker.
+    ray.get(f.remote())
+
+    ray.get([f.remote() for _ in range(100)])
+
+  def testBlockingTasks(self):
+    ray.init(num_workers=1)
+
+    @ray.remote
+    def f(i, j):
+      return (i, j)
+
+    @ray.remote
+    def g(i):
+      # Each instance of g submits and blocks on the result of another remote
+      # task.
+      object_ids = [f.remote(i, j) for j in range(10)]
+      return ray.get(object_ids)
+
+    ray.get([g.remote(i) for i in range(100)])
+
+    @ray.remote
+    def _sleep(i):
+      time.sleep(1)
+      return (i)
+
+    @ray.remote
+    def sleep():
+      # Each instance of sleep submits and blocks on the result of another
+      # remote task, which takes one second to execute.
+      ray.get([_sleep.remote(i) for i in range(10)])
+
+    ray.get(sleep.remote())
+
+    ray.worker.cleanup()
+
 class SchedulingAlgorithm(unittest.TestCase):
 
   def attempt_to_load_balance(self, remote_function, args, total_tasks,

test/stress_tests.py

@@ -300,5 +300,27 @@ class ReconstructionTestsMultinode(ReconstructionTests):
   # one worker each.
   num_local_schedulers = 4
 
+# NOTE(swang): This test tries to launch 1000 workers and breaks.
+#class WorkerPoolTests(unittest.TestCase):
+#
+#  def tearDown(self):
+#    ray.worker.cleanup()
+#
+#  def testBlockingTasks(self):
+#    @ray.remote
+#    def f(i, j):
+#      return (i, j)
+#
+#    @ray.remote
+#    def g(i):
+#      # Each instance of g submits and blocks on the result of another remote
+#      # task.
+#      object_ids = [f.remote(i, j) for j in range(10)]
+#      return ray.get(object_ids)
+#
+#    ray.init(num_workers=1)
+#    ray.get([g.remote(i) for i in range(1000)])
+#    ray.worker.cleanup()
+
 if __name__ == "__main__":
   unittest.main(verbosity=2)