async def _run_loop(self, executor):
        """Main loop."""
        # SPEAD heap descriptor.
        # One channel per stream and one time index per heap.
        max_duration_sec = self._config['max_duration_sec']
        num_stations = self._config['heap']['num_stations']
        num_pols = self._config['heap']['num_pols']
        num_baselines = num_stations * (num_stations + 1) // 2
        reporting_interval_sec = self._config['reporting_interval_sec']
        start_chan = self._config['start_channel']
        dtype = [('TCI', 'i1'), ('FD', 'u1'), ('VIS', '<c8', num_pols)]
        descriptor = {
            'visibility_timestamp_count': {
                'id': 0x6000,
                'dtype': '<u4',
            },
            'visibility_timestamp_fraction': {
                'id': 0x6001,
                'dtype': '<u4',
            },
            'visibility_channel_id': {
                'id': 0x6002,
                'dtype': '<u4'
            },
            'visibility_channel_count': {
                'id': 0x6003,
                'dtype': '<u4'
            },
            'visibility_baseline_polarisation_id': {
                'id': 0x6004,
                'dtype': '<u4'
            },
            'visibility_baseline_count': {
                'id': 0x6005,
                'dtype': '<u4'
            },
            'phase_bin_id': {
                'id': 0x6006,
                'dtype': '<u2'
            },
            'phase_bin_count': {
                'id': 0x6007,
                'dtype': '<u2'
            },
            'scan_id': {
                'id': 0x6008,
                'dtype': '<u8'
            },
            'visibility_hardware_id': {
                'id': 0x6009,
                'dtype': '<u4'
            },
            'correlator_output_data': {
                'id': 0x600A,
                'dtype': dtype,
                'shape': (num_baselines,)
            }
        }

        # Create shadow heap buffers for each stream/channel.
        for i_buffer in range(2):
            self._buffer.append([])
            for _ in range(len(self._streams)):
                heap_data = {
                    'visibility_timestamp_count': 0,
                    'visibility_timestamp_fraction': 0,
                    'correlator_output_data': numpy.zeros(
                        (num_baselines,), dtype=dtype)
                }
                self._buffer[i_buffer].append(heap_data)

        # Add items to each stream's item group, and
        # send the start of stream message to each stream.
        tasks = []
        for i_stream, (stream, item_group) in enumerate(self._streams):
            for key, item in descriptor.items():
                item_shape = item['shape'] if 'shape' in item else tuple()
                item_group.add_item(
                    id=item['id'], name=key, description='',
                    shape=item_shape, dtype=item['dtype'])

            # Write the header information.
            item_group['visibility_channel_id'].value = i_stream + start_chan
            item_group['visibility_channel_count'].value = 1
            item_group['visibility_baseline_count'].value = num_baselines
            tasks.append(stream.async_send_heap(item_group.get_start()))
        await asyncio.gather(*tasks)

        # Calculate the total data size being sent.
        num_streams = self._config['num_streams']
        data_size = (num_baselines * (num_pols * 8 + 2)) / 1e6
        self._log.info('Heap size is %.3f MB', data_size)

        # Enter the main loop over time samples.
        done = False
        loop = asyncio.get_event_loop()
        heaps_sent = 0
        i_time = 0
        start_time = time.time()
        timer1 = time.time()
        while not done:
            # Schedule sends for the previous buffer, if available.
            tasks = []
            if i_time > 0:
                i_buffer = (i_time - 1) % 2
                for i_stream, (stream, item_group) in enumerate(self._streams):
                    for key, value in self._buffer[i_buffer][i_stream].items():
                        item_group[key].value = value  # Update values in heap.
                    # Always send the descriptors AND the static data!
                    tasks.append(stream.async_send_heap(
                        item_group.get_heap(descriptors='all', data='all')))
                    heaps_sent += 1

            # Fill a buffer by distributing it among worker threads.
            i_buffer = i_time % 2
            for i_stream in range(num_streams):
                tasks.append(loop.run_in_executor(
                    executor, self.fill_buffer,
                    self._buffer[i_buffer][i_stream], i_stream))

            # Ensure processing tasks and previous asynchronous sends are done.
            await asyncio.gather(*tasks)

            # Increment time index.
            i_time += 1

            # Periodically report the sender data rate.
            now = time.time()
            if (now - timer1) >= reporting_interval_sec:
                self._log.info('Sent %i heaps in %.3f sec (%.1f MB/s)',
                               heaps_sent, (now - timer1),
                               (data_size * heaps_sent) / (now - timer1))
                heaps_sent = 0
                timer1 = now

            # Check for completion.
            if now - start_time > max_duration_sec > 0:
                done = True

        # End the streams.
        tasks = []
        for stream, item_group in self._streams:
            tasks.append(stream.async_send_heap(item_group.get_end()))
        await asyncio.gather(*tasks)
Main loop.