concurrent_batch_iter.py 文件源码

def next(self):
    self.scan(do_print=(self.counter % self.log_queue == 0))
    if self.counter % self.log_queue == 0:
      self.counter = 0
    batch = self.q.get()
    self.q.task_done()
    self.counter += 1
    while batch is None:
      self.info("Got an empty batch. Ending iteration.")
      self.relaunch = False
      try:
        batch = self.q.get(False)
        self.q.task_done()
        qempty = False
      except Queue.Empty:
        qempty = True
        pass

      if qempty:
        self.info("Queue empty. Scanning for alive thread.")
        # Scan for alive thread.
        found_alive = False
        for ff in self.fetchers:
          if ff.is_alive():
            found_alive = True
            break

        self.info("No alive thread found. Joining.")
        # If no alive thread, join all.
        if not found_alive:
          for ff in self.fetchers:
            ff.join()
          raise StopIteration
      else:
        self.info("Got another batch from the queue.")
    return batch