docs: 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)