code: async def _run_loop(self, executor):
          
          loop = asyncio.get_event_loop()
  
          # Get first heap in each stream (should be empty).
          self._log.info("Waiting for %d streams to start...", self._num_streams)
          for stream in self._streams:
              await stream.get(loop=loop)
  
          i_block = 0
          receive_buffer = [None, None]
          while True:
              # Process the previous buffer, if available.
              processing_tasks = None
              if i_block > 0:
                  i_buffer_proc = (i_block - 1) % 2
                  processing_tasks = loop.run_in_executor(
                      executor, self.process_buffer, i_block - 1,
                      receive_buffer[i_buffer_proc])
  
              # Set up asynchronous receive on all streams.
              i_buffer_recv = i_block % 2
              receive_tasks = []
              for stream in self._streams:
                  for _ in range(self._num_buffer_times):
                      # NOTE: loop.create_task() is needed to schedule these
                      # in the right order!
                      receive_tasks.append(
                          loop.create_task(stream.get(loop=loop)))
              receive_buffer[i_buffer_recv] = asyncio.gather(*receive_tasks)
  
              # Ensure asynchronous receives and previous processing tasks are
              # done.
              self._log.info("Receiving block %i", i_block)
              try:
                  await receive_buffer[i_buffer_recv]
                  self._log.info("Received block %i", i_block)
              except spead2.Stopped:
                  self._log.info("Stream Stopped")
                  break
              if processing_tasks:
                  await processing_tasks
  
              i_block += 1
docs: Main loop.