class_docstr: Executes a blocking function asynchronously on a queue of items.
docstr: Buffers elements in a queue and runs `fn` asynchronously..
  
  NOTE: Once closed, `AsyncExecutor` will block until current `fn` finishes
    but is not guaranteed to dequeue all elements currently stored in
    the data queue. This is intentional so as to prevent a blocking `fn` call
    from preventing `AsyncExecutor` from closing.
  
  Args:
    fn: A callable to be executed upon dequeuing an element from data
      queue.
    queue_size: The maximum size of the synchronized buffer queue.
    interruptible_interval_secs: Timeout interval in seconds for blocking
      queue operations after which the background threads check for errors and
      if background threads should stop.
code: def __init__(
      self,
      fn: Callable[[E], None],
      queue_size: int = 1,
      interruptible_interval_secs: float = 1.0,
  ):
      """Buffers elements in a queue and runs `fn` asynchronously..
  
      NOTE: Once closed, `AsyncExecutor` will block until current `fn` finishes
        but is not guaranteed to dequeue all elements currently stored in
        the data queue. This is intentional so as to prevent a blocking `fn` call
        from preventing `AsyncExecutor` from closing.
  
      Args:
        fn: A callable to be executed upon dequeuing an element from data
          queue.
        queue_size: The maximum size of the synchronized buffer queue.
        interruptible_interval_secs: Timeout interval in seconds for blocking
          queue operations after which the background threads check for errors and
          if background threads should stop.
      """
      self._data = queue.Queue(maxsize=queue_size)
      self._should_stop = threading.Event()
      self._errors = queue.Queue()
      self._interruptible_interval_secs = interruptible_interval_secs
  
      def _dequeue() -> None:
          """Dequeue data from a queue and invoke blocking call."""
          while not self._should_stop.is_set():
              try:
                  element = self._data.get(
                      timeout=self._interruptible_interval_secs
                  )
                  # Execute fn upon dequeuing an element from the data queue.
                  fn(element)
              except queue.Empty:
                  # If queue is Empty for longer than the specified time interval,
                  # check again if should_stop has been requested and retry.
                  continue
              except Exception as e:
                  logging.error(
                      "AsyncExecuter thread terminated with error."
                  )
                  logging.exception(e)
                  self._errors.put(e)
                  self._should_stop.set()
                  raise  # Never caught by anything, just terminates the thread.
  
      self._thread = threading.Thread(target=_dequeue, daemon=True)
      self._thread.start()
  
file_path: acme/acme/utils/async_utils.py
imports: import queue
  import threading
  from typing import Callable, TypeVar, Generic
  from absl import logging
label_desc: Use image node_id 1 to create a new AsyncExecutor object from inherited base classes: Generic with example: obj = AsyncExecutor(fn, queue_size, interruptible_interval_secs)