docstr: Returns results from worker pool
  
  :param timeout: If None, will block forever, otherwise will raise :class:`.TimeoutWaitingForResultError`
      exception if no data received within the timeout (in seconds)
  :return: arguments passed to ``publish_func(...)`` by a worker. If no more results are anticipated,
      :class:`.EmptyResultError` is raised.
docstr_tok: ["Returns","results","from","worker","pool",":","param","timeout",":","If","None",",","will","block","forever",",","otherwise","will","raise",":","class",":","`",".TimeoutWaitingForResultError","`","exception","if","no","data","received","within","the","timeout","(","in","seconds",")",":","return",":","arguments","passed","to","``","publish_func","(","...",")","``","by","a","worker",".","If","no","more","results","are","anticipated",",",":","class",":","`",".EmptyResultError","`","is","raised","."]
code: def get_results(self):
      """Returns results from worker pool
  
      :param timeout: If None, will block forever, otherwise will raise :class:`.TimeoutWaitingForResultError`
          exception if no data received within the timeout (in seconds)
      :return: arguments passed to ``publish_func(...)`` by a worker. If no more results are anticipated,
          :class:`.EmptyResultError` is raised.
      """
  
      while True:
          # If there is no more work to do, raise an EmptyResultError
          logger.debug(
              "ventilated_items=%d ventilated_items_processed=%d ventilator.completed=%s",
              self._ventilated_items,
              self._ventilated_items_processed,
              str(self._ventilator.completed())
              if self._ventilator
              else "N/A",
          )
          if self._ventilated_items == self._ventilated_items_processed:
              # We also need to check if we are using a ventilator and if it is completed
              if not self._ventilator or self._ventilator.completed():
                  logger.debug(
                      "ventilator reported it has completed. Reporting end of results"
                  )
                  raise EmptyResultError()
  
          logger.debug("get_results polling on the next result")
          socks = self._results_receiver_poller.poll(
              _VERIFY_END_OF_VENTILATION_PERIOD * 1e3
          )
          if not socks:
              continue
          # Result message is a tuple containing data payload and possible exception (or None).
          (
              fast_serialized,
              pickle_serialized,
          ) = self._results_receiver.recv_multipart(
              copy=self._zmq_copy_buffers
          )
          pickle_serialized = pickle.loads(pickle_serialized)
  
          if pickle_serialized:
              logger.debug(
                  "get_results a pickled message %s",
                  type(pickle_serialized),
              )
              if isinstance(
                  pickle_serialized, VentilatedItemProcessedMessage
              ):
                  self._ventilated_items_processed += 1
                  if self._ventilator:
                      self._ventilator.processed_item()
              elif isinstance(pickle_serialized, Exception):
                  self.stop()
                  self.join()
                  raise pickle_serialized
          else:
              logger.debug("get_results received new results")
              if self._zmq_copy_buffers:
                  deserialized_result = self._serializer.deserialize(
                      fast_serialized
                  )
              else:
                  deserialized_result = self._serializer.deserialize(
                      fast_serialized.buffer
                  )
              return deserialized_result
  
imports: import logging
  import pickle
  import sys
  import os
  from time import sleep, time
  from traceback import format_exc
  from threading import Thread
  from psutil import pid_exists
  import zmq
  from zmq import ZMQBaseError
  from petastorm.reader_impl.pickle_serializer import PickleSerializer
  from petastorm.workers_pool import EmptyResultError, VentilatedItemProcessedMessage
  from petastorm.workers_pool.exec_in_new_process import exec_in_new_process
label_desc: Use image node_id 6 for calling the ProcessPool obj's underlying member method code with example usage: obj.get_results() and returns: deserialized_result