Return a single batch from queue or ``None`` signaling epoch end.

        :raise ChildProcessError: if the enqueueing thread ended unexpectedly
345369
def _dequeue_batch(self) -> Optional[Batch]:
        
        if self._enqueueing_thread is None:
            raise ValueError('StreamWrapper `{}` with buffer of size `{}` was used outside with-resource environment.'
                             .format(self._name, self._buffer_size))
        if not self._enqueueing_thread.is_alive() and self._queue.empty():
            self._start_thread()
        while True:
            try:
                batch = self._queue.get(timeout=2)
                self._queue.task_done()
                break
            except Empty:
                if not self._enqueueing_thread.is_alive():
                    try:
                        # the enqueueing thread may just finished properly so lets check the queue eagerly
                        batch = self._queue.get_nowait()
                        self._queue.task_done()
                        break
                    except Empty:
                        # so we failed to retrieve a batch and the enqueueing thread is dead
                        # there is no hope, something must went wrong
                        raise ChildProcessError('Enqueueing thread ended unexpectedly.')
        return batch
664416
def get_next_batch(self):
        
        if self.collection_scanner.is_enabled:
            batch = self.collection_scanner.get_new_batch()
            for item in batch:
                base_item = BaseRecord(item)
                self.increase_read()
                self.last_position['last_key'] = item['_key']
                yield base_item
            self.logger.debug('Done reading batch')
        else:
            self.logger.debug('No more batches')
            self.finished = True
725065
def process_single_message_from_queue(self):
        
        try:
            message = self.message_queue.get_nowait()
            task_id, data = message
            task = self.task_id_to_task[task_id]
            task.on_message(data)
            return True
        except queue.Empty:
            return False
180962
def get_batch_children(self):
        
        if not self.batch:
            raise SbgError("This task is not a batch task.")
        return self.query(parent=self.id, api=self._api)
570454
def on_batch_end(self, train, **kwargs):
        ""
        if train: self.stats.append(self.hooks.stored)
965013
def queue_get_stoppable(self, q):
        
        while not self.stopped():
            try:
                return q.get(timeout=5)
            except queue.Empty:
                pass
152495
def get(self, block=True, timeout=None):
        
        value = self._queue.get(block, timeout)
        if self._queue.empty():
            self.clear()
        return value
1351016
def enqueue(self, future):
    

    future.enqueue()
    with self._lock:
      if self._shutdown:
        raise RuntimeError('ThreadPool has been shut down and can no '
          'longer accept futures.')
      self._queue.append(future)
      if len(self._running) == len(self._workers):
        self._new_worker()
      self._lock.notify_all()
920906
def process_queue(queue=None, **kwargs):
    
    while True:
        item = queue.get()
        if item is None:
            queue.task_done()
            logger.info(f"{queue}: exiting process queue.")
            break
        filename = os.path.basename(item)
        try:
            queue.next_task(item, **kwargs)
        except Exception as e:
            queue.task_done()
            logger.warn(f"{queue}: item={filename}. {e}\n")
            logger.exception(e)
            sys.stdout.write(
                style.ERROR(
                    f"{queue}. item={filename}. {e}. Exception has been logged.\n"
                )
            )
            sys.stdout.flush()
            break
        else:
            logger.info(f"{queue}: Successfully processed {filename}.\n")
        queue.task_done()
699778
def wait(self, timeout=-1):
        
        if self._process is None:
            raise RuntimeError('no child process')
        if timeout == -1:
            timeout = self._timeout
        if not self._child_exited.wait(timeout):
            raise Timeout('timeout waiting for child to exit')
        return self.returncode
465714
def next(self) -> mx.io.DataBatch:
        
        if self.iter_next():
            return self.next_batch
        raise StopIteration
580401
def PopItem(self):
    
    try:
      # If no timeout is specified the queue will block if empty otherwise
      # a Queue.Empty exception is raised.
      return self._queue.get(timeout=self._timeout)

    except KeyboardInterrupt:
      raise errors.QueueClose

    # If close() is called on the multiprocessing.Queue while it is blocking
    # on get() it will raise IOError.
    except IOError:
      raise errors.QueueClose

    except Queue.Empty:
      raise errors.QueueEmpty
1127846
def bft(self):
        
        queue = deque([self])
        while queue:
            node = queue.pop()
            yield node
            if hasattr(node, "childs"):
                queue.extendleft(node.childs)
223333
async def get(self, *, encoding=None, decoder=None):
        
        assert decoder is None or callable(decoder), decoder
        if self._queue.exhausted:
            raise ChannelClosedError()
        msg = await self._queue.get()
        if msg is EndOfStream:
            # TODO: maybe we need an explicit marker for "end of stream"
            #       currently, returning None may overlap with
            #       possible return value from `decoder`
            #       so the user would have to check `ch.is_active`
            #       to determine if its EoS or payload
            return
        if self._is_pattern:
            dest_channel, msg = msg
        if encoding is not None:
            msg = msg.decode(encoding)
        if decoder is not None:
            msg = decoder(msg)
        if self._is_pattern:
            return dest_channel, msg
        return msg
46748
async def run(self):
        
        # Implement pull max batch from queue instead of one message at a time
        while self.eh_partition_pump.pump_status != "Errored" and not self.eh_partition_pump.is_closing():
            if self.eh_partition_pump.partition_receive_handler:
                try:
                    msgs = await self.eh_partition_pump.partition_receive_handler.receive(
                        max_batch_size=self.max_batch_size,
                        timeout=self.recieve_timeout)
                except Exception as e:  # pylint: disable=broad-except
                    _logger.info("Error raised while attempting to receive messages: %r", e)
                    await self.process_error_async(e)
                else:
                    if not msgs:
                        _logger.info("No events received, queue size %r, release %r",
                                     self.eh_partition_pump.partition_receive_handler.queue_size,
                                     self.eh_partition_pump.host.eph_options.release_pump_on_timeout)
                        if self.eh_partition_pump.host.eph_options.release_pump_on_timeout:
                            await self.process_error_async(TimeoutError("No events received"))
                    else:
                        await self.process_events_async(msgs)
469056
def finish(self):
    
    if self.proc is None:
      return None
    self.proc.stdin.close()
    for thread in (self._out_thread, self._err_thread):
      thread.join()
    (out, err) = [
        b"".join(chunks) for chunks in (self._out_chunks, self._err_chunks)
    ]
    self.proc.stdout.close()
    self.proc.stderr.close()
    if self.proc.returncode:
      err = "\n".join([" ".join(self.cmd), err.decode("utf8")])
      raise IOError(err)
    del self.proc
    self.proc = None
    return out
574637
def on_batch_end(self, last_target, train, **kwargs):
        ""
        if train: return
        bs = last_target.size(0)
        for name in self.names:
            self.metrics[name] += bs * self.learn.loss_func.metrics[name].detach().cpu()
        self.nums += bs