def io_worker_receive_handler(self, io_worker):
    # Consume everything immediately
    data = io_worker.peek_receive_buf()
    io_worker.consume_receive_buf(len(data))
    if self._currently_blocked:
      # thread-safe queue
      self._receive_queue.put(data)
    else:
      self._actual_receive(data)
called from io_worker (recoco thread, after the Select loop pushes onto io_worker)
dataset/ETHPy150Open ucb-sts/sts/sts/util/deferred_io.py/DeferredIOWorker.io_worker_receive_handler