[core] Pin arguments during task execution (#13737)

* tmp

* Pin task args

* unit tests

* update

* test

* Fix

python/ray/tests/test_object_manager.py

@@ -296,9 +296,6 @@ def test_pull_request_retry(shutdown_only):
     ray.get(driver.remote())
 
 
-@pytest.mark.skip(
-    reason="This hangs due to a deadlock between a worker getting its "
-    "arguments and the node pulling arguments for the next task queued.")
 @pytest.mark.timeout(30)
 def test_pull_bundles_admission_control(shutdown_only):
     cluster = Cluster()
@@ -333,9 +330,6 @@ def test_pull_bundles_admission_control(shutdown_only):
     ray.get(tasks)
 
 
-@pytest.mark.skip(
-    reason="This hangs due to a deadlock between a worker getting its "
-    "arguments and the node pulling arguments for the next task queued.")
 @pytest.mark.timeout(30)
 def test_pull_bundles_admission_control_dynamic(shutdown_only):
     # This test is the same as test_pull_bundles_admission_control, except that
@@ -358,11 +352,13 @@ def test_pull_bundles_admission_control_dynamic(shutdown_only):
     cluster.wait_for_nodes()
 
     @ray.remote
-    def foo(*args):
+    def foo(i, *args):
+        print("foo", i)
         return
 
     @ray.remote
-    def allocate(*args):
+    def allocate(i):
+        print("allocate", i)
         return np.zeros(object_size, dtype=np.uint8)
 
     args = []
@@ -373,8 +369,8 @@ def test_pull_bundles_admission_control_dynamic(shutdown_only):
         ]
         args.append(task_args)
 
-    tasks = [foo.remote(*task_args) for task_args in args]
-    allocated = [allocate.remote() for _ in range(num_objects)]
+    tasks = [foo.remote(i, *task_args) for i, task_args in enumerate(args)]
+    allocated = [allocate.remote(i) for i in range(num_objects)]
     ray.get(tasks)
     del allocated
 

python/ray/tests/test_object_spilling.py

@@ -618,9 +618,6 @@ def test_release_during_plasma_fetch(object_spilling_config, shutdown_only):
     do_test_release_resource(object_spilling_config, expect_released=True)
 
 
-@pytest.mark.skip(
-    reason="This hangs due to a deadlock between a worker getting its "
-    "arguments and the node pulling arguments for the next task queued.")
 @pytest.mark.skipif(
     platform.system() == "Windows", reason="Failing on Windows.")
 @pytest.mark.timeout(30)