code: def test_task_status(ray_start_regular):
      address = ray_start_regular["address"]
  
      @ray.remote
      def dep(sema, x=None):
          ray.get(sema.acquire.remote())
          return
  
      @ray.remote(num_gpus=1)
      def impossible():
          pass
  
      # Filter out actor handle refs.
      def filtered_summary():
          data = "\n".join(
              [
                  line
                  for line in memory_summary(
                      address, line_wrap=False
                  ).split("\n")
                  if "ACTOR_HANDLE" not in line
              ]
          )
          print(data)
          return data
  
      sema = Semaphore.remote(value=0)
      x = dep.remote(sema)
      y = dep.remote(sema, x=x)
      im = impossible.remote()  # noqa
      # x and its semaphore task are scheduled. im cannot
      # be scheduled, so it is pending forever.
      wait_for_condition(
          lambda: count(filtered_summary(), WAITING_FOR_EXECUTION) == 2
      )
      wait_for_condition(
          lambda: count(filtered_summary(), WAITING_FOR_DEPENDENCIES)
          == 1
      )
      wait_for_condition(
          lambda: count(filtered_summary(), SCHEDULED) == 1
      )
  
      z = dep.remote(sema, x=x)
      wait_for_condition(
          lambda: count(filtered_summary(), WAITING_FOR_DEPENDENCIES)
          == 2
      )
      wait_for_condition(
          lambda: count(filtered_summary(), WAITING_FOR_EXECUTION) == 2
      )
      wait_for_condition(
          lambda: count(filtered_summary(), FINISHED) == 0
      )
  
      sema.release.remote()
      time.sleep(2)
      wait_for_condition(
          lambda: count(filtered_summary(), FINISHED) == 1
      )
      wait_for_condition(
          lambda: count(filtered_summary(), WAITING_FOR_DEPENDENCIES)
          == 0
      )
      # y, z, and two semaphore tasks are scheduled.
      wait_for_condition(
          lambda: count(filtered_summary(), WAITING_FOR_EXECUTION) == 4
      )
  
      sema.release.remote()
      wait_for_condition(
          lambda: count(filtered_summary(), FINISHED) == 2
      )
      wait_for_condition(
          lambda: count(filtered_summary(), WAITING_FOR_DEPENDENCIES)
          == 0
      )
      wait_for_condition(
          lambda: count(filtered_summary(), WAITING_FOR_EXECUTION) == 2
      )
  
      sema.release.remote()
      ray.get(y)
      ray.get(z)
      wait_for_condition(
          lambda: count(filtered_summary(), FINISHED) == 3
      )
      wait_for_condition(
          lambda: count(filtered_summary(), WAITING_FOR_DEPENDENCIES)
          == 0
      )
      wait_for_condition(
          lambda: count(filtered_summary(), WAITING_FOR_EXECUTION) == 0
      )
      wait_for_condition(
          lambda: count(filtered_summary(), SCHEDULED) == 1
      )
  
file_path: ray/python/ray/tests/test_memstat.py
imports: import os
  import time
  import numpy
  import pytest
  import ray
  from ray._private.internal_api import memory_summary
  from ray._private.test_utils import Semaphore, wait_for_condition
  from ray.cluster_utils import Cluster, cluster_not_supported
label_desc: Use image node_id 12 for calling a global function with example usage: test_task_status(ray_start_regular) and returns: data