code: def _distribute_save_datasets(self, scenes_iter, client, batch_size=1, **kwargs):
          
          def load_data(q):
              idx = 0
              while True:
                  future_list = q.get()
                  if future_list is None:
                      break
  
                  # save_datasets shouldn't be returning anything
                  for future in future_list:
                      future.result()
                      log.info("Finished saving %d scenes", idx)
                      idx += 1
                  q.task_done()
  
          input_q = Queue(batch_size if batch_size is not None else 1)
          load_thread = Thread(target=load_data, args=(input_q,))
          load_thread.start()
  
          for scene in scenes_iter:
              delayed = scene.save_datasets(compute=False, **kwargs)
              if isinstance(delayed, (list, tuple)) and len(delayed) == 2:
                  # TODO Make this work for (source, target) datasets
                  # given a target, source combination
                  raise NotImplementedError("Distributed save_datasets does not support writers "
                                            "that return (source, target) combinations at this time. Use "
                                            "the non-distributed save_datasets instead.")
              future = client.compute(delayed)
              input_q.put(future)
          input_q.put(None)
  
          log.debug("Waiting for child thread to get saved results...")
          load_thread.join()
          log.debug("Child thread died successfully")
docs: Distribute save_datasets across a cluster.