docstr: An actor that periodically issues state API
  
  Args:
      - apis: List of StateAPICallSpec
      - call_interval_s: State apis in the `apis` will be issued
          every `call_interval_s` seconds.
      - print_interval_s: How frequent state api stats will be dumped.
      - wait_after_stop: When true, call to `ray.get(actor.stop.remote())`
          will wait for all pending state APIs to return.
          Setting it to `False` might miss some long-running state apis calls.
      - print_result: True if result of each API call is printed. Default False.
code: def __init__(
      self,
      apis: List[StateAPICallSpec],
      call_interval_s: float = 5.0,
      print_interval_s: float = 20.0,
      wait_after_stop: bool = True,
      print_result: bool = False,
  ) -> None:
      """An actor that periodically issues state API
  
      Args:
          - apis: List of StateAPICallSpec
          - call_interval_s: State apis in the `apis` will be issued
              every `call_interval_s` seconds.
          - print_interval_s: How frequent state api stats will be dumped.
          - wait_after_stop: When true, call to `ray.get(actor.stop.remote())`
              will wait for all pending state APIs to return.
              Setting it to `False` might miss some long-running state apis calls.
          - print_result: True if result of each API call is printed. Default False.
      """
      # Configs
      self._apis = apis
      self._call_interval_s = call_interval_s
      self._print_interval_s = print_interval_s
      self._wait_after_cancel = wait_after_stop
      self._logger = logging.getLogger(self.__class__.__name__)
      self._print_result = print_result
  
      # States
      self._tasks = None
      self._fut_queue = None
      self._executor = None
      self._loop = None
      self._stopping = False
      self._stopped = False
      self._stats = StateAPIStats()
  
imports: import asyncio
  import sys
  from copy import deepcopy
  from collections import defaultdict
  import concurrent.futures
  from dataclasses import dataclass, field
  import logging
  import numpy
  import pprint
  import time
  import traceback
  from typing import Callable, Dict, List, Optional, Tuple, Union
  from ray.util.state import list_tasks
  import ray
  from ray.actor import ActorHandle
  from ray.util.state import list_workers
  from ray._private.gcs_utils import GcsAioClient, GcsChannel
  from ray.util.state.state_manager import StateDataSourceClient
  from ray.dashboard.state_aggregator import StateAPIManager
label_desc: Use image node_id 1 to create a new StateAPIGeneratorActor object  with example: obj = StateAPIGeneratorActor(apis, call_interval_s, print_interval_s, wait_after_stop, print_result)