docs: def hybrid_threaded_worker(selector, workers):
      """Runs a set of workers, each in a separate thread.
  
      :param selector:
          A function that takes a hints-tuple and returns a key
          indexing a worker in the `workers` dictionary.
      :param workers:
          A dictionary of workers.
  
      :returns:
          A connection for the scheduler.
      :rtype: Connection
  
      The hybrid worker dispatches jobs to the different workers
      based on the information contained in the hints. If no hints
      were given, the job is run in the main thread.
  
      Dispatching is done in the main thread. Retrieving results is
      done in a separate thread for each worker. In this design it is
      assumed that dispatching a job takes little time, while waiting for
      one to return a result may take a long time.
      """
      result_queue = Queue()
  
      job_sink = {k: w.sink() for k, w in workers.items()}
  
      @push
      def dispatch_job():
          default_sink = result_queue.sink()
  
          while True:
              msg = yield
  
              if msg is EndOfQueue:
                  for k in workers.keys():
                      try:
                          job_sink[k].send(EndOfQueue)
                      except StopIteration:
                          pass
                  return
  
              if msg is FlushQueue:
                  for k in workers.keys():
                      try:
                          job_sink[k].send(FlushQueue)
                      except StopIteration:
                          pass
                  return
  
              worker = selector(msg.node)
              if worker:
                  job_sink[worker].send(msg)
              else:
                  default_sink.send(run_job(*msg))
  
      for key, worker in workers.items():
          t = threading.Thread(
              target=patch,
              args=(worker.source, result_queue.sink))
          t.daemon = True
          t.start()
  
      return Connection(result_queue.source, dispatch_job)