code: def test_deadlock_two_sets_of_actor_with_nested_task(
      ray_with_memory_monitor,
  ):
      barrier = BarrierActor.options(
          max_restarts=0, max_task_retries=0, max_concurrency=2
      ).remote(2)
      leaker1 = ActorWithNestedTask.options(
          max_restarts=0, max_task_retries=0
      ).remote(barrier)
      leaker2 = ActorWithNestedTask.options(
          max_restarts=0, max_task_retries=0
      ).remote(barrier)
      parent_bytes = get_additional_bytes_to_reach_memory_usage_pct(
          (memory_usage_threshold - 0.05) / 2
      )
      nested_bytes = (
          get_additional_bytes_to_reach_memory_usage_pct(
              memory_usage_threshold + 0.1
          )
          - 2 * parent_bytes
      )
      ref1 = leaker1.perform_allocations.remote(
          parent_bytes, nested_bytes
      )
      ref2 = leaker2.perform_allocations.remote(
          parent_bytes, nested_bytes
      )
  
      with pytest.raises(ray.exceptions.RayTaskError) as _:
          ray.get(ref1)
      with pytest.raises(ray.exceptions.RayTaskError) as _:
          ray.get(ref2)
  
imports: import sys
  import threading
  import pytest
  import ray
  from ray.tests.test_memory_pressure import allocate_memory, Leaker, get_additional_bytes_to_reach_memory_usage_pct, memory_usage_threshold, memory_monitor_refresh_ms
label_desc: Use image node_id 9 for calling a global function with example usage: test_deadlock_two_sets_of_actor_with_nested_task(ray_with_memory_monitor) without return types