code: def init(
      *,
      max_running_workflows: Optional[int] = None,
      max_pending_workflows: Optional[int] = None,
  ) -> None:
      """Initialize workflow.
  
      If Ray is not initialized, we will initialize Ray and
      use ``/tmp/ray/workflow_data`` as the default storage.
  
      Args:
          max_running_workflows: The maximum number of concurrently running workflows.
              Use -1 as infinity. 'None' means preserving previous setting or initialize
              the setting with infinity.
          max_pending_workflows: The maximum number of queued workflows.
              Use -1 as infinity. 'None' means preserving previous setting or initialize
              the setting with infinity.
      """
      usage_lib.record_library_usage("workflow")
  
      if max_running_workflows is not None:
          if not isinstance(max_running_workflows, int):
              raise TypeError(
                  "'max_running_workflows' must be None or an integer."
              )
          if max_running_workflows < -1 or max_running_workflows == 0:
              raise ValueError(
                  "'max_running_workflows' must be a positive integer "
                  "or use -1 as infinity."
              )
      if max_pending_workflows is not None:
          if not isinstance(max_pending_workflows, int):
              raise TypeError(
                  "'max_pending_workflows' must be None or an integer."
              )
          if max_pending_workflows < -1:
              raise ValueError(
                  "'max_pending_workflows' must be a non-negative integer "
                  "or use -1 as infinity."
              )
  
      if not ray.is_initialized():
          # We should use get_temp_dir_path, but for ray client, we don't
          # have this one. We need a flag to tell whether it's a client
          # or a driver to use the right dir.
          # For now, just use $TMP/ray/workflow_data
          workflow_dir = (
              Path(tempfile.gettempdir()) / "ray" / "workflow_data"
          )
          ray.init(storage=workflow_dir.as_uri())
      workflow_access.init_management_actor(
          max_running_workflows, max_pending_workflows
      )
      serialization.init_manager()
  
imports: import functools
  import logging
  import tempfile
  from typing import Dict, Set, List, Tuple, Union, Optional, Any
  import time
  import uuid
  from pathlib import Path
  import ray
  from ray.dag import DAGNode
  from ray.dag.input_node import DAGInputData
  from ray.remote_function import RemoteFunction
  from ray.workflow.common import WorkflowStatus, Event, asyncio_run, validate_user_metadata
  from ray.workflow import serialization, workflow_access, workflow_context
  from ray.workflow.event_listener import EventListener, EventListenerType, TimerListener
  from ray.workflow.workflow_storage import WorkflowStorage
  from ray.workflow.workflow_state_from_dag import workflow_state_from_dag
  from ray.util.annotations import PublicAPI
  from ray._private.usage import usage_lib
label_desc: Use image node_id 1 for calling a global function with example usage: init() without return types