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","."]
docstr: Handle an event on a connection socket.
  
  You can simulate an EVENT_READ on in-memory data by setting mask==0 and data!=None.
docstr_tok: ["Handle","an","event","on","a","connection","socket",".","You","can","simulate","an","EVENT_READ","on","in-memory","data","by","setting","mask==0","and","data","!","=None","."]
code: def handle_conn(
      self, conn: socket.socket, mask: int, data: Optional[bytes] = None
  ) -> None:
      """
      Handle an event on a connection socket.
  
      You can simulate an EVENT_READ on in-memory data by setting mask==0 and data!=None.
      """
      assert self.sel
      pid = self.conns[conn]
      if mask & selectors.EVENT_READ:
          data = conn.recv(4096)
      # Messages are all one-byte codes for easy parsing.
      # The protocol is "any number of keepalive "k"s followed by a quit "q"", so we can
      # safely ignore everything except the final byte of the message.
      if data:
          if data[-1:] == b"k":
              # keepalive message; leave the connection alone.
              return
          elif data[-1:] == b"q":
              # Graceful shutdown code.
              self.graceful_shutdowns.append(pid)
          else:
              raise ValueError("invalid message from pid_client:", data)
  
      # Error, EOF, or anything else.
  
      if self.listener is not None:
          raise det.errors.WorkerError(
              "worker died before all workers connected"
          )
  
      self.sel.unregister(conn)
      conn.close()
      del self.conns[conn]
  
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 7 for calling the PIDServer obj's underlying member method code with example usage: obj.handle_conn(conn, mask, data) without return types