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(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