code: def _distribute_frame_compute(self, writers, frame_keys, frames_to_write, client, batch_size=1):
          
          def load_data(frame_gen, q):
              for frame_arrays in frame_gen:
                  future_list = client.compute(frame_arrays)
                  for frame_key, arr_future in zip(frame_keys, future_list):
                      q.put({frame_key: arr_future})
              q.put(None)
  
          input_q = Queue(batch_size if batch_size is not None else 1)
          load_thread = Thread(target=load_data, args=(frames_to_write, input_q,))
          remote_q = client.gather(input_q)
          load_thread.start()
  
          while True:
              future_dict = remote_q.get()
              if future_dict is None:
                  break
  
              # write the current frame
              # this should only be one element in the dictionary, but this is
              # also the easiest way to get access to the data
              for frame_key, result in future_dict.items():
                  # frame_key = rev_future_dict[future]
                  w = writers[frame_key]
                  w.append_data(result)
              input_q.task_done()
  
          log.debug("Waiting for child thread...")
          load_thread.join(10)
          if load_thread.is_alive():
              import warnings
              warnings.warn("Background thread still alive after failing to die gracefully")
          else:
              log.debug("Child thread died successfully")
docs: Use ``dask.distributed`` to compute multiple frames at a time.