code: def launch(
      program: lp_program.Program,
      test_case: absltest.TestCase,
      local_resources: Optional[Mapping[str, Any]] = None,
  ):
      """Launches a program using multiple processes as a test."""
      # Set up the launch context (launch type & launch config) for all nodes
      local_resources = local_resources or {}
      for label, nodes in program.groups.items():
          launch_config = local_resources.get(label, None)
          for node in nodes:
              node._initialize_context(
                  context.LaunchType.TEST_MULTI_PROCESSING,
                  launch_config=launch_config,
              )
  
      # Notify the input handles
      for label, nodes in program.groups.items():
          for node in nodes:
              for handle in node.input_handles:
                  handle.connect(node, label)
  
      # Bind addresses
      for node in program.get_all_nodes():
          node.bind_addresses()
  
      label_to_commands = {}
      process_handles = {}
      for label, nodes in program.groups.items():
          # to_executables() is a static method, so we can call it from any of the
          # nodes in this group.
          # Somehow pytype thinks to_executables() gets wrong arg count.
          # pytype: disable=wrong-arg-count
          commands = nodes[0].to_executables(
              nodes, label, nodes[0].launch_context
          )
          # pytype: enable=wrong-arg-count
          if commands:
              label_to_commands[label] = commands
              # Not to create to actual processes, in case of failures in this loop.
              process_handles[label] = []
  
      if lp_flags.LP_WORKER_MANAGER_V2.value:
          manager = worker_manager_v2.WorkerManager(
              # Subprocess sends SIGINT back via lp.stop(), which stops the program.
              kill_workers_upon_sigint=True,
              # Kill all processes immediately.
              termination_notice_secs=0,
          )
      else:
          manager = worker_manager.WorkerManager(kill_main_thread=False)
      for label, commands in label_to_commands.items():
          for command in commands:
              env = {}
              env.update(os.environ)
              env.update(command.env_overrides)
              manager.process_worker(
                  label, command.command_as_list, env=env
              )
  
      def _cleanup(manager=manager):
          if lp_flags.LP_WORKER_MANAGER_V2.value:
              manager = typing.cast(
                  worker_manager_v2.WorkerManager, manager
              )
  
              manager._set_stop_event_and_terminate_process_workers(
                  sig=signal.SIGKILL
              )
  
          else:
              typing.cast(
                  worker_manager.WorkerManager, manager
              ).cleanup_after_test(test_case)
  
      test_case.addCleanup(_cleanup)
      return manager
  
file_path: launchpad/launchpad/launch/test_multi_processing/launch.py
imports: import os
  import signal
  import typing
  from typing import Any, Mapping, Optional
  from absl.testing import absltest
  from launchpad import context
  from launchpad import flags
  from launchpad import program
  from launchpad.launch import worker_manager
  from launchpad.launch import worker_manager_v2
label_desc: Use image node_id 1 for calling a global function with example usage: launch(program, test_case, local_resources) and returns: manager