docs: def inference(cluster_info, feed_timeout=600, qname='input'):
    """Feeds Spark partitions into the shared multiprocessing.Queue and returns inference results.
  
    Args:
      :cluster_info: node reservation information for the cluster (e.g. host, executor_id, pid, ports, etc)
      :feed_timeout: number of seconds after which data feeding times out (600 sec default)
      :qname: *INTERNAL_USE*
  
    Returns:
      A dataRDD.mapPartitions() function
    """
    def _inference(iter):
      # get shared queue, reconnecting if necessary
      mgr = _get_manager(cluster_info, util.get_ip_address(), util.read_executor_id())
      try:
        queue_in = mgr.get_queue(qname)
        equeue = mgr.get_queue('error')
      except (AttributeError, KeyError):
        msg = "Queue '{}' not found on this node, check for exceptions on other nodes.".format(qname)
        raise Exception(msg)
  
      logging.info("Feeding partition {0} into {1} queue {2}".format(iter, qname, queue_in))
      count = 0
      for item in iter:
        count += 1
        queue_in.put(item, block=True)
  
      # signal "end of partition"
      queue_in.put(marker.EndPartition())
  
      # skip empty partitions
      if count == 0:
        return []
  
      # wait for consumers to finish processing all items in queue before "finishing" this iterator
      joinThr = Thread(target=queue_in.join)
      joinThr.start()
      timeout = feed_timeout
      while (joinThr.isAlive()):
        if (not equeue.empty()):
          e_str = equeue.get()
          equeue.task_done()
          raise Exception("exception in worker:\n" + e_str)
        time.sleep(1)
        timeout -= 1
        if timeout <= 0:
          raise Exception("Timeout while feeding partition")
  
      logging.info("Processed {0} items in partition".format(count))
  
      # read result queue
      results = []
      queue_out = mgr.get_queue('output')
      while count > 0:
        result = queue_out.get(block=True)
        results.append(result)
        count -= 1
        queue_out.task_done()
  
      logging.info("Finished processing partition")
      return results
  
    return _inference