async def _run_loop(self, executor):
        """Main loop."""
        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