class_docstr: This object serves as the interface between a ZMQ socket passing
  Requests and Replies, and a thread or threads serving those requests.
  Received requests are received on a ZMQ socket and placed on upward_queue,
  and notify_all() is called on updward_cv.  Replies (via the Request.reply()
  method) are dispatched to their requesters via a downward queue.
  
  The Boundary wakes up the socket thread via the notify socket. This lets
  the socket thread poll for changes on the notify and request sockets, but
  allows it to receive Python objects via the downward queue.
class_docstr_tok: ["This","object","serves","as","the","interface","between","a","ZMQ","socket","passing","Requests","and","Replies",",","and","a","thread","or","threads","serving","those","requests",".","Received","requests","are","received","on","a","ZMQ","socket","and","placed","on","upward_queue",",","and","notify_all","(",")","is","called","on","updward_cv",".","Replies","(","via","the","Request.reply","(",")","method",")","are","dispatched","to","their","requesters","via","a","downward","queue",".","The","Boundary","wakes","up","the","socket","thread","via","the","notify","socket",".","This","lets","the","socket","thread","poll","for","changes","on","the","notify","and","request","sockets",",","but","allows","it","to","receive","Python","objects","via","the","downward","queue","."]
code: def spin(self):
      try:
          # socket for handling downward notifications
          selfnotify_socket = self.zmq_context.socket(zmq.SUB)
          selfnotify_socket.connect(
              cellprofiler_core.utilities.zmq.NOTIFY_SOCKET_ADDR
          )
          selfnotify_socket.setsockopt(zmq.SUBSCRIBE, b"")
  
          # socket where we receive Requests
          request_socket = self.zmq_context.socket(zmq.ROUTER)
          request_socket.setsockopt(zmq.LINGER, 0)
          request_port = request_socket.bind_to_random_port(
              self.zmq_address
          )
          self.request_address = self.zmq_address + (
              ":%d" % request_port
          )
  
          poller = zmq.Poller()
          poller.register(selfnotify_socket, zmq.POLLIN)
          poller.register(request_socket, zmq.POLLIN)
  
          received_stop = False
          while not received_stop:
              # Tell the workers to stay alive
              self.keepalive_socket.send(NOTIFY_RUN)
              poll_result = poller.poll(1000)
              #
              # Under all circumstances, read everything from the queue
              #
              try:
                  while True:
                      (
                          notification,
                          arg,
                      ) = self.downward_queue.get_nowait()
                      if notification == self.NOTIFY_REPLY_READY:
                          req, rep = arg
                          self.handle_reply(req, rep)
                      elif notification == self.NOTIFY_CANCEL_ANALYSIS:
                          analysis_id, response_queue = arg
                          self.handle_cancel(
                              analysis_id, response_queue
                          )
                      elif notification == self.NOTIFY_STOP:
                          received_stop = True
              except queue.Empty:
                  pass
              #
              # Then process the poll result
              #
              for s, state in poll_result:
                  if s == selfnotify_socket:
                      # This isn't really used for message transmission.
                      # Instead, it pokes this spin thread to get it to check
                      # the non-ZMQ message queue.
                      msg = selfnotify_socket.recv()
                      # Let's watch out for stop signals anyway
                      if msg == NOTIFY_STOP:
                          LOGGER.warning(
                              "Captured a stop message over zmq"
                          )
                          received_stop = True
                      continue
                  req = Communicable.recv(s, routed=True)
                  req.set_boundary(self)
                  if not isinstance(req, AnalysisRequest):
                      for request_class in self.request_dictionary:
                          if isinstance(req, request_class):
                              q = self.request_dictionary[request_class]
                              q.put([self, self.NOTIFY_REQUEST, req])
                              break
                      else:
                          LOGGER.warning(
                              "Received a request that wasn't an AnalysisRequest: %s"
                              % str(type(req))
                          )
                          req.reply(BoundaryExited())
                      continue
                  if s != request_socket:
                      # Request is on the external socket
                      LOGGER.warning(
                          "Received a request on the external socket"
                      )
                      req.reply(BoundaryExited())
                      continue
                  #
                  # Filter out requests for cancelled analyses.
                  #
                  with self.analysis_context_lock:
                      if not self.analysis_context.enqueue(req):
                          continue
          # Give the workers an explicit stop command
          self.keepalive_socket.send(NOTIFY_STOP)
          self.keepalive_socket.close()
          #
          # We assume here that workers trying to communicate with us will
          # be shut down without needing replies to pending requests.
          # There's not much we can do in terms of handling that in a more
          # orderly fashion since workers might be formulating requests as or
          # after we have shut down. But calling cancel on all the analysis
          # contexts will raise exceptions in threads waiting for a rep/rep.
          #
          # You could call analysis_context.handle_cancel() here, what if it
          # blocks?
          with self.analysis_context_lock:
              self.analysis_context.cancel()
              for (
                  request_class_queue
              ) in self.request_dictionary.values():
                  #
                  # Tell each response class to stop. Wait for a reply
                  # which may be a thread instance. If so, join to the
                  # thread so there will be an orderly shutdown.
                  #
                  response_queue = queue.Queue()
                  request_class_queue.put(
                      [self, self.NOTIFY_STOP, response_queue]
                  )
                  thread = response_queue.get()
                  if isinstance(thread, threading.Thread):
                      thread.join()
  
          request_socket.close()
          selfnotify_socket.close()
          LOGGER.debug("Shutting down the boundary thread")
      except:
          #
          # Pretty bad - a logic error or something extremely unexpected
          #              We're close to hosed here, better die an ugly death.
          #
          LOGGER.critical(
              "Unhandled exception in boundary thread.", exc_info=True
          )
          import os
  
          os._exit(-1)
  
imports: import logging
  import queue
  import threading
  import zmq
  import cellprofiler_core.utilities.zmq
  from cellprofiler_core.utilities.zmq._analysis_context import AnalysisContext
  from .communicable import Communicable
  from .communicable.reply.upstream_exit import BoundaryExited
  from .communicable.request import AnalysisRequest, Request
  from ...constants.worker import NOTIFY_STOP, NOTIFY_RUN
label_desc: Use image node_id 8 for calling the Boundary obj's underlying member method code with example usage: obj.spin() without return types