code: def main(
      master_url: str,
      cert_name: str,
      cert_file: str,
      metadata: Dict[str, str],
      token: str,
      emit_stdout_logs: bool,
      cmd: List[str],
      log_wait_time: int,
  ) -> int:
      logq: queue.Queue = queue.Queue()
      doneq: queue.Queue = queue.Queue()
  
      waiter_started = False
      stdout_started = False
      stderr_started = False
      shipper_started = False
  
      # Normally we like structured concurrency; i.e. a function that owns a thread must not exit
      # until that thead has been properly cleaned up.  However, it is important that the log shipper
      # is not allowed to keep a task container alive too long after the child process has exited.  We
      # want to guarantee that we exit about DET_LOG_WAIT_TIME seconds after the child process exits.
      #
      # However, interruping a synchronous HTTP call form urllib is nearly impossible; even if you
      # were to select() until the underlying file descriptor had something to read before calling
      # Request.read(), there are many buffered readers in there and most likely multiple os.read()
      # calls would occur and you'd be blocking anyway.
      #
      # So as an easy workaround, we set daemon=True and just exit the process if it's not done on
      # time.
      shipper = Shipper(
          master_url,
          token,
          cert_name,
          cert_file,
          logq,
          doneq,
          daemon=True,
      )
      shipper_timed_out = False
  
      shipper.assert_master_is_reachable()
  
      # Start the process or ship a special log message to the master why we couldn't.
      try:
          # Don't rely on Popen's standard line buffering; we want to do our own line buffering.
          p = subprocess.Popen(
              cmd,
              stdin=subprocess.DEVNULL,
              stdout=subprocess.PIPE,
              stderr=subprocess.PIPE,
              bufsize=0,
          )
      except FileNotFoundError:
          shipper.ship_special(
              f"FileNotFoundError executing {cmd}",
              metadata,
              emit_stdout_logs,
          )
          # 127 is the standard bash exit code for file-not-found.
          return 127
      except PermissionError:
          # Unable to read or to execute the command.
          shipper.ship_special(
              f"PermissionError executing {cmd}",
              metadata,
              emit_stdout_logs,
          )
          # 126 is the standard bash exit code for permission failure.
          return 126
      except Exception:
          msg = (
              f"unexpected failure executing {cmd}:\n"
              + traceback.format_exc()
          )
          shipper.ship_special(msg, metadata, emit_stdout_logs)
          # 80 is the exit code we use to signal "ship_logs.py failed"
          return 80
  
      # Just for mypy.
      assert isinstance(p.stdout, io.RawIOBase) and isinstance(
          p.stderr, io.RawIOBase
      )
  
      try:
          waiter = Waiter(p, doneq)
          waiter.start()
          waiter_started = True
  
          # Set up signal forwarding.
          def signal_passthru(signum: Any, frame: Any):
              p.send_signal(signum)
  
          for sig in [
              signal.SIGINT,
              signal.SIGTERM,
              signal.SIGHUP,
              signal.SIGUSR1,
              signal.SIGUSR2,
              signal.SIGWINCH,
          ]:
              signal.signal(sig, signal_passthru)
  
          stdout = Collector(
              p.stdout,
              "stdout",
              emit_stdout_logs,
              metadata,
              logq,
              doneq,
          )
          stderr = Collector(
              p.stderr,
              "stderr",
              emit_stdout_logs,
              metadata,
              logq,
              doneq,
          )
  
          stdout.start()
          stdout_started = True
  
          stderr.start()
          stderr_started = True
  
          shipper.start()
          shipper_started = True
  
          exit_code = None
          deadline: Optional[float] = None
          # Expect 4 messages on the doneq, one for each thread we started.
          for _ in range(4):
              # Wait for an event, possibly with a deadline (if the child process already exited).
              try:
                  timeout = (
                      None
                      if deadline is None
                      else deadline - time.time()
                  )
                  if timeout is not None and timeout <= 0:
                      raise queue.Empty()
                  donemsg = doneq.get(timeout=timeout)
                  assert isinstance(donemsg, DoneMsg)
              except queue.Empty:
                  # Deadline is done, just abandon the shipper.
                  shipper_timed_out = True
                  logging.error(
                      f"waited {log_wait_time} seconds for shipper to finish after child exit; "
                      "giving up now"
                  )
                  break
  
              if donemsg.who == "shipper":
                  # There's no point in collecting logs after the shipper is gone.
                  stdout.shipper_died = True
                  stderr.shipper_died = True
  
              if donemsg.error is not None:
                  # Something in our shipping machinery broke.
                  raise RuntimeError(
                      f"failure in log shipper; {donemsg.who} thread died"
                  ) from donemsg.error
  
              if donemsg.who == "waiter":
                  # After the log shipper exits, the shipping code is on a deadline.
                  #
                  # Note: we could almost just return exit_code here and let the cleanup and shipping
                  # timeout happen in the finally block, but the finally block can't easily detect
                  # exceptions that arrive on the doneq, so we stay in this for...range(4) loop until
                  # we're confident all threads shut down without error.
                  deadline = time.time() + log_wait_time
                  exit_code = donemsg.exit_code
  
          # Mypy doesn't know that we're guaranteed to have an exit_code by now.
          #
          # It's guaranteed because the only scenario where we exit the loop without waiting on all
          # the threads is if the Waiter exited but the Shipper didn't.  And the only case where the
          # Waiter doesn't set the exit_code is in the case that p.wait() threw an error, in which
          # case we would have raised an exception due to a non-None donemsg.error.
          assert exit_code is not None
  
          # Convert signal exits to standard bash-style exits.
          if exit_code < 0:
              return 128 - exit_code
          return exit_code
  
      finally:
          # If our logging infrastructure ever crashes, just give up on the child process.
          p.kill()
          if waiter_started:
              waiter.join()
          # After p is dead, the Collectors should run out of input and exit quickly.
          if stdout_started:
              stdout.join()
          if stderr_started:
              stderr.join()
          # After everyone else is dead, the shipper could still be in a retry loop for a long time,
          # so we wait up to DET_LOG_WAIT_TIME seconds for it to exit, then we give up on it.
          if shipper_started and not shipper_timed_out:
              shipper.join(timeout=log_wait_time)
              if shipper.is_alive():
                  # The timeout was reached.
                  logging.error(
                      f"waited {log_wait_time} seconds for shipper to finish after crash; "
                      "giving up now"
                  )
  
imports: import datetime
  import io
  import json
  import logging
  import os
  import queue
  import re
  import signal
  import socket
  import ssl
  import subprocess
  import sys
  import threading
  import time
  import traceback
  import urllib.error
  import urllib.request
  from typing import Any, Dict, Iterator, List, NamedTuple, Optional, cast
label_desc: Use image node_id 3 for calling a global function with example usage: main(master_url, cert_name, cert_file, metadata, token, emit_stdout_logs, cmd, log_wait_time) and returns: exit_code, int, int, int, unknown