class_docstr: Keep the ownership and manage the workflow output.
docstr: Submit workflow. A submitted workflow can be executed later.
  
  Args:
      workflow_id: ID of the workflow.
      state: The initial state of the workflow.
      ignore_existing: Ignore existing executed workflows.
docstr_tok: ["Submit","workflow",".","A","submitted","workflow","can","be","executed","later",".","Args",":","workflow_id",":","ID","of","the","workflow",".","state",":","The","initial","state","of","the","workflow",".","ignore_existing",":","Ignore","existing","executed","workflows","."]
code: def submit_workflow(
      self,
      workflow_id: str,
      state: WorkflowExecutionState,
      ignore_existing: bool = False,
  ):
      """Submit workflow. A submitted workflow can be executed later.
  
      Args:
          workflow_id: ID of the workflow.
          state: The initial state of the workflow.
          ignore_existing: Ignore existing executed workflows.
      """
      if workflow_id in self._workflow_executors:
          raise RuntimeError(
              f"Workflow[id={workflow_id}] is being executed."
          )
      if (
          workflow_id in self._executed_workflows
          and not ignore_existing
      ):
          raise RuntimeError(
              f"Workflow[id={workflow_id}] has been executed."
          )
  
      if state.output_task_id is None:
          raise ValueError(
              "No root DAG specified that generates output for the workflow."
          )
  
      wf_store = workflow_storage.WorkflowStorage(workflow_id)
      if (
          self._max_running_workflows != -1
          and len(self._running_workflows)
          >= self._max_running_workflows
      ):
          try:
              self._workflow_queue.put_nowait(workflow_id)
              self._queued_workflows[workflow_id] = asyncio.Future()
              wf_store.update_workflow_status(WorkflowStatus.PENDING)
          except queue.Full:
              # override with our error message
              raise queue.Full("Workflow queue has been full") from None
      else:
          self._running_workflows.add(workflow_id)
          wf_store.update_workflow_status(WorkflowStatus.RUNNING)
      # initialize executor
      self._workflow_executors[workflow_id] = WorkflowExecutor(state)
  
imports: import asyncio
  import logging
  import queue
  from typing import Dict, List, Set, Optional, TYPE_CHECKING
  import ray
  from ray.workflow import common
  from ray.workflow.common import WorkflowStatus, TaskID
  from ray.workflow import workflow_state_from_storage
  from ray.workflow import workflow_context
  from ray.workflow import workflow_storage
  from ray.workflow.exceptions import WorkflowCancellationError, WorkflowNotFoundError, WorkflowNotResumableError, WorkflowStillActiveError
  from ray.workflow.workflow_executor import WorkflowExecutor
  from ray.workflow.workflow_state import WorkflowExecutionState
  from ray.workflow.workflow_context import WorkflowTaskContext
label_desc: Use image node_id 4 for calling the WorkflowManagementActor obj's underlying member method code with example usage: obj.submit_workflow(workflow_id, state, ignore_existing) without return types