code: def test_parallel_actor_fill_plasma_retry(ray_start_cluster_head):
      @ray.remote
      class LargeMemoryActor:
          def some_expensive_task(self):
              return np.zeros(10**8 // 2, dtype=np.uint8)
  
      actors = [LargeMemoryActor.remote() for _ in range(5)]
      for _ in range(5):
          pending = [a.some_expensive_task.remote() for a in actors]
          while pending:
              [done], pending = ray.wait(pending, num_returns=1)
  
imports: import os
  import sys
  import signal
  import threading
  import json
  from pathlib import Path
  import ray
  import numpy
  import pytest
  import psutil
  import time
  from ray._private.test_utils import SignalActor, wait_for_pid_to_exit, wait_for_condition, run_string_as_driver_nonblocking
label_desc: Use image node_id 2 for calling a global function with example usage: test_parallel_actor_fill_plasma_retry(ray_start_cluster_head) and returns: np