From 0f2a2224c22c4a340ae1e3c6740491b70da3ef5e Mon Sep 17 00:00:00 2001
From: Gagandeep Singh <gdp.1807@gmail.com>
Date: Tue, 8 Feb 2022 15:29:46 +0530
Subject: [PATCH] PoolActor now uses num_cpus=0 to avoid any deadlock (#22048)

https://github.com/ray-project/ray/issues/21488#issuecomment-1027122177 :

> We discussed this issue in a bit more detail and came to the conclusion that we should set the CPU resource requirement for each actor in the actor pool to 0, to make the Ray Pool compatible/same behavior as the Python multiprocessing pool. Would that work for you @yogeveran ? (very similar to solution 4 mentioned above, but with 0.0 instead of 0.1, so it works in all cases).
---
 python/ray/tests/test_multiprocessing.py | 22 ++++++++++++++++++++++
 python/ray/util/multiprocessing/pool.py  |  2 +-
 2 files changed, 23 insertions(+), 1 deletion(-)

diff --git a/python/ray/tests/test_multiprocessing.py b/python/ray/tests/test_multiprocessing.py
index 8c72b07c0..415f87d45 100644
--- a/python/ray/tests/test_multiprocessing.py
+++ b/python/ray/tests/test_multiprocessing.py
@@ -6,6 +6,7 @@ import time
 import random
 from collections import defaultdict
 import queue
+import math
 
 import ray
 from ray._private.test_utils import SignalActor
@@ -46,6 +47,14 @@ def pool_4_processes_python_multiprocessing_lib():
     pool.join()
 
 
+@pytest.fixture
+def ray_start_1_cpu():
+    address_info = ray.init(num_cpus=1)
+    yield address_info
+    # The code after the yield will run as teardown code.
+    ray.shutdown()
+
+
 def test_ray_init(shutdown_only):
     def getpid(args):
         return os.getpid()
@@ -562,6 +571,19 @@ def test_maxtasksperchild(shutdown_only):
     pool.join()
 
 
+def test_deadlock_avoidance_in_recursive_tasks(ray_start_1_cpu):
+    def poolit_a(_):
+        with Pool(ray_address="auto") as pool:
+            return list(pool.map(math.sqrt, range(0, 2, 1)))
+
+    def poolit_b():
+        with Pool(ray_address="auto") as pool:
+            return list(pool.map(poolit_a, range(2, 4, 1)))
+
+    result = poolit_b()
+    assert result == [[0.0, 1.0], [0.0, 1.0]]
+
+
 if __name__ == "__main__":
     import pytest
 
diff --git a/python/ray/util/multiprocessing/pool.py b/python/ray/util/multiprocessing/pool.py
index 3a107022d..9305e8616 100644
--- a/python/ray/util/multiprocessing/pool.py
+++ b/python/ray/util/multiprocessing/pool.py
@@ -435,7 +435,7 @@ class UnorderedIMapIterator(IMapIterator):
         return self._ready_objects.popleft()
 
 
-@ray.remote(num_cpus=1)
+@ray.remote(num_cpus=0)
 class PoolActor:
     """Actor used to process tasks submitted to a Pool."""