docstr: The core logic of executing a workflow.
  
  This class is responsible for:
  
  - Dependency resolving.
  - Task scheduling.
  - Reference counting.
  - Garbage collection.
  - Continuation handling and scheduling.
  - Error handling.
  - Responding callbacks.
  
  It borrows some design of event loop in asyncio,
  e.g., 'run_until_complete'.
  
  Args:
      state: The initial state of the workflow.
docstr_tok: ["The","core","logic","of","executing","a","workflow",".","This","class","is","responsible","for",":","-","Dependency","resolving",".","-","Task","scheduling",".","-","Reference","counting",".","-","Garbage","collection",".","-","Continuation","handling","and","scheduling",".","-","Error","handling",".","-","Responding","callbacks",".","It","borrows","some","design","of","event","loop","in","asyncio",",","e.g.",",","'run_until_complete","'",".","Args",":","state",":","The","initial","state","of","the","workflow","."]
code: def __init__(
      self,
      state: WorkflowExecutionState,
  ):
      """The core logic of executing a workflow.
  
      This class is responsible for:
  
      - Dependency resolving.
      - Task scheduling.
      - Reference counting.
      - Garbage collection.
      - Continuation handling and scheduling.
      - Error handling.
      - Responding callbacks.
  
      It borrows some design of event loop in asyncio,
      e.g., 'run_until_complete'.
  
      Args:
          state: The initial state of the workflow.
      """
      self._state = state
      self._completion_queue = asyncio.Queue()
      self._task_done_callbacks: Dict[
          TaskID, List[asyncio.Future]
      ] = defaultdict(list)
  
imports: from typing import Dict, List, Iterator, Optional, Tuple, TYPE_CHECKING
  import asyncio
  import logging
  import time
  from collections import defaultdict
  import ray
  from ray.exceptions import RayTaskError, RayError
  from ray.workflow.common import WorkflowRef, WorkflowExecutionMetadata, WorkflowStatus, TaskID
  from ray.workflow.exceptions import WorkflowCancellationError, WorkflowExecutionError
  from ray.workflow.task_executor import get_task_executor, _BakedWorkflowInputs
  from ray.workflow.workflow_state import WorkflowExecutionState, TaskExecutionMetadata, Task
label_desc: Use image node_id 1 to create a new WorkflowExecutor object  with example: obj = WorkflowExecutor(state)