code: def _launch_task_servers(
      all_host_names,
      local_host_names,
      driver_addresses,
      num_hosts,
      tmout,
      key,
      ssh_port=None,
  ):
      """
      executes the task server and service client task for registration on the
      hosts.
      :param all_host_names: list of addresses. for example,
          ['worker-0','worker-1']
          ['10.11.11.11', '10.11.11.12']
      :type all_host_names: list(string)
      :param local_host_names: names that are resolved to one of the addresses
      of local hosts interfaces. For example,
          set(['localhost', '127.0.0.1'])
      :type local_host_names: set
      :param driver_addresses: map of interfaces and their address and port for
      the service. For example:
          {
              'lo': [('127.0.0.1', 34588)],
              'docker0': [('172.122.10.1', 34588)],
              'eth0': [('11.111.33.73', 34588)]
          }
      :type driver_addresses: map
      :param num_hosts:
      :type num_hosts: int
      :param tmout:
      :type tmout: horovod.spark.util.timeout.Timeout
      :return:
      :rtype:
      """
  
      def _exec_command(command):
          host_output = six.StringIO()
          try:
              exit_code = safe_shell_exec.execute(
                  command, stdout=host_output, stderr=host_output
              )
              if exit_code != 0:
                  print(
                      "Launching bluefog task function was not "
                      "successful:\n{host_output}".format(
                          host_output=host_output.getvalue()
                      )
                  )
                  os._exit(exit_code)
          finally:
              host_output.close()
          return exit_code
  
      if ssh_port:
          ssh_port_arg = "-p {ssh_port}".format(ssh_port=ssh_port)
      else:
          ssh_port_arg = ""
  
      args_list = []
      for index, host_name in enumerate(all_host_names):
          if host_name in local_host_names:
              command = (
                  "{python} -m bluefog.run.horovodrun.task_fn {index} "
                  "{driver_addresses} {num_hosts} {timeout} {key}".format(
                      python=sys.executable,
                      index=codec.dumps_base64(index),
                      driver_addresses=codec.dumps_base64(
                          driver_addresses
                      ),
                      num_hosts=codec.dumps_base64(num_hosts),
                      timeout=codec.dumps_base64(tmout),
                      key=codec.dumps_base64(key),
                  )
              )
          else:
              command = (
                  "ssh -o StrictHostKeyChecking=no {host} {ssh_port_arg} "
                  "'{python} -m bluefog.run.horovodrun.task_fn {index} "
                  "{driver_addresses} {num_hosts} {timeout} {key}'".format(
                      host=host_name,
                      ssh_port_arg=ssh_port_arg,
                      python=sys.executable,
                      index=codec.dumps_base64(index),
                      driver_addresses=codec.dumps_base64(
                          driver_addresses
                      ),
                      num_hosts=codec.dumps_base64(num_hosts),
                      timeout=codec.dumps_base64(tmout),
                      key=codec.dumps_base64(key),
                  )
              )
          args_list.append([command])
      # Each thread will use ssh command to launch the server on one task. If an
      # error occurs in one thread, entire process will be terminated. Otherwise,
      # threads will keep running and ssh session -- and the the task server --
      # will be bound to the thread. In case, the horovodrun process dies, all
      # the ssh sessions and all the task servers will die as well.
      threads.execute_function_multithreaded(
          _exec_command, args_list, block_until_all_done=False
      )
  
file_path: bluefog/bluefog/run/horovod_driver.py
imports: import os
  import sys
  import six
  from bluefog.run.horovodrun.common.util import codec, safe_shell_exec, timeout, secret
  from bluefog.run.horovodrun.driver import driver_service
  from bluefog.run.horovodrun.task import task_service
  from bluefog.run.horovodrun.util import threads
label_desc: Use image node_id 1 for calling a global function with example usage: _launch_task_servers(all_host_names, local_host_names, driver_addresses, num_hosts, tmout, key, ssh_port) and returns: exit_code