class_docstr: PIDServer tracks PIDs reported by a set of pid_clients which connect to it.
  
  PIDServer.run() will return when all pid_clients have reported a graceful shutdown and have
  exited, or it will raise an exception if any pids disappear without reporting a graceful
  shutdown.
  
  PIDServer lets an sshd-based launch layer keep track of its worker processes, even when the
  worker processes aren't proper child processes.
class_docstr_tok: ["PIDServer","tracks","PIDs","reported","by","a","set","of","pid_clients","which","connect","to","it",".","PIDServer.run","(",")","will","return","when","all","pid_clients","have","reported","a","graceful","shutdown","and","have","exited",",","or","it","will","raise","an","exception","if","any","pids","disappear","without","reporting","a","graceful","shutdown",".","PIDServer","lets","an","sshd-based","launch","layer","keep","track","of","its","worker","processes",",","even","when","the","worker","processes","are","n't","proper","child","processes","."]
code: def __init__(
      self, addr: Union[str, int, Tuple[str, int]], num_clients: int
  ) -> None:
      self.addr = addr
      self.num_clients = num_clients
  
      self.started = False
      self.sel = None  # type: Optional[selectors.BaseSelector]
      self.listener = None  # type: Optional[socket.socket]
  
      self.pids = []  # type: List[int]
      self.graceful_shutdowns = []  # type: List[int]
      # maps a connection to its pid
      self.conns = {}  # type: Dict[socket.socket, int]
  
      self.done_accepting = False
  
imports: import logging
  import os
  import selectors
  import signal
  import socket
  import subprocess
  import time
  from typing import Any, Callable, Dict, List, Optional, Tuple, Union
  import determined
label_desc: Use image node_id 1 to create a new PIDServer object  with example: obj = PIDServer(addr, num_clients)