code: def get_or_create_manager(
      warn_on_creation: bool = True,
  ) -> "ActorHandle":
      """Get or create the storage manager."""
      # TODO(suquark): We should not get the actor everytime. We also need to
      # resume the actor if it failed. Using a global variable to cache the
      # actor seems not enough to resume the actor, because there is no
      # aliveness detection for an actor.
      try:
          return ray.get_actor(
              common.STORAGE_ACTOR_NAME,
              namespace=common.MANAGEMENT_ACTOR_NAMESPACE,
          )
      except ValueError:
          if warn_on_creation:
              logger.warning(
                  "Cannot access workflow serialization manager. It "
                  "could be because "
                  "the workflow manager exited unexpectedly. A new "
                  "workflow manager is being created. "
              )
          handle = Manager.options(
              name=common.STORAGE_ACTOR_NAME,
              namespace=common.MANAGEMENT_ACTOR_NAMESPACE,
              lifetime="detached",
          ).remote()
          ray.get(handle.ping.remote())
          return handle
  
imports: import contextlib
  from dataclasses import dataclass
  import logging
  import os
  import ray
  from ray import cloudpickle
  from ray.types import ObjectRef
  from ray.workflow import common, workflow_storage
  from typing import Any, Dict, Generator, List, Optional, Tuple, TYPE_CHECKING
  from collections import ChainMap
  import io
label_desc: Use image node_id 2 for calling a global function with example usage: get_or_create_manager(warn_on_creation) and returns: ray, handle