code: def submit_jobs(args, udf_command):
      """Submit distributed jobs (server and client processes) via ssh"""
      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)
      ip_config = 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 = DEFAULT_PORT
                  hosts.append((ip, port))
              else:
                  raise RuntimeError("Format error of ip_config.")
              server_count_per_machine = args.num_proc_per_machine
  
      # launch server tasks
      server_env_vars = construct_dgl_server_env_vars(
          ip_config=args.ip_config,
          num_proc_per_machine=args.num_proc_per_machine,
          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} RANK={i} MASTER_ADDR={hosts[0][0]} MASTER_PORT={args.master_port}"
          cmd = wrap_cmd_with_local_envvars(
              udf_command, server_env_vars_cur
          )
          print(cmd)
          thread_list.append(
              execute_remote(
                  cmd, ip, args.ssh_port, username=args.ssh_username
              )
          )
  
      # 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)
  
      for thread in thread_list:
          thread.join()
      # The training processes complete. We should tell the cleanup process to exit.
      conn2.send("exit")
      process.join()
  
imports: import argparse
  import json
  import logging
  import multiprocessing
  import os
  import re
  import signal
  import stat
  import subprocess
  import sys
  import time
  from functools import partial
  from threading import Thread
  from typing import Optional
label_desc: Use image node_id 12 for calling a global function with example usage: submit_jobs(args, udf_command) without return types