code: def submit_jobs(args, udf_command, dry_run=False):
      """Submit distributed jobs (server and client processes) via ssh"""
      if dry_run:
          print(
              "Currently it's in dry run mode which means no jobs will be launched."
          )
      servers_cmd = []
      clients_cmd = []
      hosts = []
      thread_list = []
      server_count_per_machine = 0
  
      # Get the IP addresses of the cluster.
      ip_config = os.path.join(args.workspace, args.ip_config)
      with open(ip_config) as f:
          for line in f:
              result = line.strip().split()
              if len(result) == 2:
                  ip = result[0]
                  port = int(result[1])
                  hosts.append((ip, port))
              elif len(result) == 1:
                  ip = result[0]
                  port = get_available_port(ip)
                  hosts.append((ip, port))
              else:
                  raise RuntimeError("Format error of ip_config.")
              server_count_per_machine = args.num_servers
      # Get partition info of the graph data
      part_config = os.path.join(args.workspace, args.part_config)
      with open(part_config) as conf_f:
          part_metadata = json.load(conf_f)
      assert "num_parts" in part_metadata, "num_parts does not exist."
      # The number of partitions must match the number of machines in the cluster.
      assert part_metadata["num_parts"] == len(
          hosts
      ), "The number of graph partitions has to match the number of machines in the cluster."
  
      state_q = queue.Queue()
      tot_num_clients = (
          args.num_trainers * (1 + args.num_samplers) * len(hosts)
      )
      # launch server tasks
      server_env_vars = construct_dgl_server_env_vars(
          num_samplers=args.num_samplers,
          num_server_threads=args.num_server_threads,
          tot_num_clients=tot_num_clients,
          part_config=args.part_config,
          ip_config=args.ip_config,
          num_servers=args.num_servers,
          graph_format=args.graph_format,
          pythonpath=os.environ.get("PYTHONPATH", ""),
      )
      for i in range(len(hosts) * server_count_per_machine):
          ip, _ = hosts[int(i / server_count_per_machine)]
          server_env_vars_cur = f"{server_env_vars} DGL_SERVER_ID={i}"
          cmd = wrap_cmd_with_local_envvars(
              udf_command, server_env_vars_cur
          )
          cmd = (
              wrap_cmd_with_extra_envvars(cmd, args.extra_envs)
              if len(args.extra_envs) > 0
              else cmd
          )
          cmd = "cd " + str(args.workspace) + "; " + cmd
          servers_cmd.append(cmd)
          if not dry_run:
              thread_list.append(
                  execute_remote(
                      cmd,
                      state_q,
                      ip,
                      args.ssh_port,
                      username=args.ssh_username,
                  )
              )
  
      # launch client tasks
      client_env_vars = construct_dgl_client_env_vars(
          num_samplers=args.num_samplers,
          tot_num_clients=tot_num_clients,
          part_config=args.part_config,
          ip_config=args.ip_config,
          num_servers=args.num_servers,
          graph_format=args.graph_format,
          num_omp_threads=os.environ.get(
              "OMP_NUM_THREADS", str(args.num_omp_threads)
          ),
          group_id=0,
          pythonpath=os.environ.get("PYTHONPATH", ""),
      )
  
      master_addr = hosts[0][0]
      master_port = get_available_port(master_addr)
      for node_id, host in enumerate(hosts):
          ip, _ = host
          # Transform udf_command to follow torch's dist launcher format: `PYTHON_BIN -m torch.distributed.run ... UDF`
          torch_dist_udf_command = wrap_udf_in_torch_dist_launcher(
              udf_command=udf_command,
              num_trainers=args.num_trainers,
              num_nodes=len(hosts),
              node_rank=node_id,
              master_addr=master_addr,
              master_port=master_port,
          )
          cmd = wrap_cmd_with_local_envvars(
              torch_dist_udf_command, client_env_vars
          )
          cmd = (
              wrap_cmd_with_extra_envvars(cmd, args.extra_envs)
              if len(args.extra_envs) > 0
              else cmd
          )
          cmd = "cd " + str(args.workspace) + "; " + cmd
          clients_cmd.append(cmd)
          if not dry_run:
              thread_list.append(
                  execute_remote(
                      cmd,
                      state_q,
                      ip,
                      args.ssh_port,
                      username=args.ssh_username,
                  )
              )
  
      # return commands of clients/servers directly if in dry run mode
      if dry_run:
          return clients_cmd, servers_cmd
  
      # Start a cleanup process dedicated for cleaning up remote training jobs.
      conn1, conn2 = multiprocessing.Pipe()
      func = partial(
          get_all_remote_pids, hosts, args.ssh_port, udf_command
      )
      process = multiprocessing.Process(
          target=cleanup_proc, args=(func, conn1)
      )
      process.start()
  
      def signal_handler(signal, frame):
          logging.info("Stop launcher")
          # We need to tell the cleanup process to kill remote training jobs.
          conn2.send("cleanup")
          sys.exit(0)
  
      signal.signal(signal.SIGINT, signal_handler)
  
      err = 0
      for thread in thread_list:
          thread.join()
          err_code = state_q.get()
          if err_code != 0:
              # Record err_code
              # We record one of the error if there are multiple
              err = err_code
  
      # The training processes complete. We should tell the cleanup process to exit.
      conn2.send("exit")
      process.join()
      if err != 0:
          print("Task failed")
          sys.exit(-1)
  
imports: import argparse
  import json
  import logging
  import multiprocessing
  import os
  import queue
  import re
  import signal
  import subprocess
  import sys
  import time
  from functools import partial
  from threading import Thread
  from typing import Optional
label_desc: Use image node_id 14 for calling a global function with example usage: submit_jobs(args, udf_command, dry_run) and returns: clients_cmd,  servers_cmd