code: def run_stateful(
      args: "argparse.Namespace",
      name: str,
      runtime_cls: Type[AsyncNewLoopRuntime],
      envs: Dict[str, str],
  ):
      """
      Method to be called in Docker containers when Stateful Executor is required. This will start
      2 processes in the Docker container.
      :param args: namespace args from the Pod
      :param name: name of the Pod to have proper logging
      :param runtime_cls: the runtime class to instantiate
      :param envs: a dictionary of environment variables to be set in the new Process
      """
      import signal
      from jina.jaml import JAML
  
      is_ready = multiprocessing.Event()
      is_shutdown = multiprocessing.Event()
      is_started = multiprocessing.Event()
      is_signal_handlers_installed = multiprocessing.Event()
      raft_worker = multiprocessing.Process(
          target=run_raft,
          kwargs={
              "args": args,
              "is_ready": is_ready,
          },
          name=name,
          daemon=True,
      )
      cargs = copy.deepcopy(args)
  
      from jina.constants import RAFT_TO_EXECUTOR_PORT
  
      if isinstance(cargs.port, int):
          cargs.port += RAFT_TO_EXECUTOR_PORT
      elif isinstance(cargs.port, list):
          cargs.port = [
              port + RAFT_TO_EXECUTOR_PORT for port in cargs.port
          ]
      worker = multiprocessing.Process(
          target=run,
          kwargs={
              "args": cargs,
              "name": name,
              "envs": envs,
              "is_started": is_started,
              "is_shutdown": is_shutdown,
              "is_ready": is_ready,
              "is_signal_handlers_installed": is_signal_handlers_installed,
              "runtime_cls": runtime_cls,
              "jaml_classes": JAML.registered_classes(),
          },
          name=name,
          daemon=False,
      )
  
      try:
          HANDLED_SIGNALS = (
              signal.SIGINT,  # Unix signal 2. Sent by Ctrl+C.
              signal.SIGTERM,  # Unix signal 15. Sent by `kill <pid>`.
              signal.SIGSEGV,
          )
  
          def signal_handler(*args, **kwargs):
              worker.terminate()
              raft_worker.terminate()
  
          for sig in HANDLED_SIGNALS:
              signal.signal(sig, signal_handler)
  
          worker.start()
          raft_worker.start()
  
          worker.join()
          raft_worker.join()
      except:
          worker.terminate()
          raft_worker.terminate()
  
file_path: jina/jina/serve/executors/run.py
imports: import argparse
  import multiprocessing
  import os
  import copy
  from typing import Dict, Optional, Type, Union, TYPE_CHECKING
  from jina.logging.logger import JinaLogger
  from jina.serve.runtimes.asyncio import AsyncNewLoopRuntime
  from jina.serve.helper import _get_workspace_from_name_and_shards
  from jina.constants import RAFT_TO_EXECUTOR_PORT
label_desc: Use image node_id 3 for calling a global function with example usage: run_stateful(args, name, runtime_cls, envs) without return types