class_docstr: The Runner manages two threads (per instance) and all of the
  workers (per class, i.e., singletons).
  
  The two threads run interface() and jobserver(), below.
  
  interface() is responsible grouping jobs for dispatch, tracking job
  progress, integrating measurements returned from workers.
  
  jobserver() is a lightweight thread that serves jobs and receives their
  requests, acting as a switchboard between workers, interface(), and
  whatever event_listener is present (via post_event()).
  
  workers are stored in AnalysisRunner.workers[], and are separate processes.
  They are expected to exit if their stdin() closes, e.g., if the parent
  process goes away.
  
  interface() and jobserver() communicate via Queues and using condition
  variables to get each other's attention.  zmqrequest is used to communicate
  between jobserver() and workers[].
class_docstr_tok: ["The","Runner","manages","two","threads","(","per","instance",")","and","all","of","the","workers","(","per","class",",","i.e.",",","singletons",")",".","The","two","threads","run","interface","(",")","and","jobserver","(",")",",","below",".","interface","(",")","is","responsible","grouping","jobs","for","dispatch",",","tracking","job","progress",",","integrating","measurements","returned","from","workers",".","jobserver","(",")","is","a","lightweight","thread","that","serves","jobs","and","receives","their","requests",",","acting","as","a","switchboard","between","workers",",","interface","(",")",",","and","whatever","event_listener","is","present","(","via","post_event","(",")",")",".","workers","are","stored","in","AnalysisRunner.workers","[","]",",","and","are","separate","processes",".","They","are","expected","to","exit","if","their","stdin","(",")","closes",",","e.g.",",","if","the","parent","process","goes","away",".","interface","(",")","and","jobserver","(",")","communicate","via","Queues","and","using","condition","variables","to","get","each","other","'s","attention",".","zmqrequest","is","used","to","communicate","between","jobserver","(",")","and","workers","[","]","."]
code: def jobserver(self, start_signal):
      # this server subthread should be very lightweight, as it has to handle
      # all the requests from workers, of which there might be several.
  
      # start the zmqrequest Boundary
      request_queue = queue.Queue()
      self.boundary.register_analysis(self.analysis_id, request_queue)
      #
      # The boundary is announcing our analysis at this point. Workers
      # will get announcements if they connect.
      #
      start_signal.release()
  
      # XXX - is this just to keep from posting another AnalysisPaused event?
      # If so, probably better to simplify the code and keep sending them
      # (should be only one per second).
      i_was_paused_before = False
  
      # start serving work until the analysis is done (or changed)
      while not self.cancelled:
          with self.jobserver_work_cv:
              if self.paused and not i_was_paused_before:
                  self.post_event(Paused())
                  i_was_paused_before = True
              if self.paused or request_queue.empty():
                  self.jobserver_work_cv.wait(
                      1
                  )  # we timeout in order to keep announcing ourselves.
                  continue  # back to while... check that we're still running
  
          if i_was_paused_before:
              self.post_event(Resumed())
              i_was_paused_before = False
  
          try:
              req = request_queue.get(timeout=0.25)
          except queue.Empty:
              continue
  
          if isinstance(req, anarequest.PipelinePreferences):
              LOGGER.debug("Received pipeline preferences request")
              req.reply(
                  Reply(
                      pipeline_blob=numpy.array(
                          self.pipeline_as_string()
                      ),
                      preferences=preferences_as_dict(),
                  )
              )
              LOGGER.debug("Replied to pipeline preferences request")
          elif isinstance(req, anarequest.InitialMeasurements):
              LOGGER.debug("Received initial measurements request")
              req.reply(Reply(buf=self.initial_measurements_buf))
              LOGGER.debug("Replied to initial measurements request")
          elif isinstance(req, anarequest.Work):
              if not self.work_queue.empty():
                  LOGGER.debug("Received work request")
                  (
                      job,
                      worker_runs_post_group,
                      wants_dictionary,
                  ) = self.work_queue.get()
                  req.reply(
                      anareply.Work(
                          image_set_numbers=job,
                          worker_runs_post_group=worker_runs_post_group,
                          wants_dictionary=wants_dictionary,
                      )
                  )
                  self.queue_dispatched_job(job)
                  LOGGER.debug(
                      "Dispatched job: image sets=%s"
                      % ",".join([str(i) for i in job])
                  )
              else:
                  # there may be no work available, currently, but there
                  # may be some later.
                  req.reply(anareply.NoWork())
          elif isinstance(req, anareply.ImageSetSuccess):
              # interface() is responsible for replying, to allow it to
              # request the shared_state dictionary if needed.
              LOGGER.debug("Received ImageSetSuccess")
              self.queue_imageset_finished(req)
              LOGGER.debug("Enqueued ImageSetSuccess")
          elif isinstance(req, anarequest.SharedDictionary):
              LOGGER.debug("Received shared dictionary request")
              req.reply(
                  anareply.SharedDictionary(
                      dictionaries=self.shared_dicts
                  )
              )
              LOGGER.debug("Sent shared dictionary reply")
          elif isinstance(req, anarequest.MeasurementsReport):
              LOGGER.debug("Received measurements report")
              self.queue_received_measurements(
                  req.image_set_numbers, req.buf
              )
              req.reply(anareply.Ack())
              LOGGER.debug("Acknowledged measurements report")
          elif isinstance(req, anarequest.AnalysisCancel):
              # Signal the interface that we are cancelling
              LOGGER.debug("Received analysis worker cancel request")
              with self.interface_work_cv:
                  self.cancelled = True
                  self.interface_work_cv.notify()
              req.reply(anareply.Ack())
          elif isinstance(
              req,
              (
                  anarequest.Interaction,
                  anarequest.Display,
                  anarequest.DisplayPostGroup,
                  anarequest.ExceptionReport,
                  anarequest.DebugWaiting,
                  anarequest.DebugComplete,
                  anarequest.OmeroLogin,
              ),
          ):
              LOGGER.debug("Enqueueing interactive request")
              # bump upward
              self.post_event(req)
              LOGGER.debug("Interactive request enqueued")
          else:
              msg = "Unknown request from worker: %s of type %s" % (
                  req,
                  type(req),
              )
              LOGGER.error(msg)
              raise ValueError(msg)
  
      # stop the ZMQ-boundary thread - will also deal with any requests waiting on replies
      if self.boundary is not None:
          self.boundary.cancel(self.analysis_id)
  
imports: import collections
  import io
  import logging
  import os
  import queue
  import subprocess
  import sys
  import tempfile
  import threading
  import re
  import numpy
  import psutil
  from .event import Finished
  from .event import Paused
  from .event import Progress
  from .event import Resumed
  from .event import Started
  from .None import reply
  from .None import request
  from ..image import ImageSetList
  from ..measurement import Measurements
  from ..utilities.measurement import load_measurements_from_buffer
  from ..pipeline import dump
  from ..preferences import get_plugin_directory
  from ..preferences import get_conserve_memory
  from ..preferences import get_temporary_directory
  from ..preferences import get_always_continue
  from ..preferences import preferences_as_dict
  from ..utilities.analysis import close_all_on_exec, start_daemon_thread
  from ..utilities.analysis import find_analysis_worker_source
  from ..utilities.analysis import find_python
  from ..utilities.analysis import find_worker_env
  from ..utilities.zmq import start_boundary, Boundary
  from ..utilities.zmq.communicable.reply import Reply
  from ..workspace import Workspace
label_desc: Use image node_id 12 for calling the Runner obj's underlying member method code with example usage: obj.jobserver(start_signal) without return types