futures.append(executor.submit(self._exec_all_actions, batch, True))
// wait until all batches have been processed
_ = [future.result() for future in futures]
else:
self._run_seq(batch_generator)
return self
After Change
self.prefetch_queue = q.Queue(maxsize=prefetch)
service_executor = cf.ThreadPoolExecutor(max_workers=2)
service_executor.submit(self._put_batches_into_queue, batch_generator)
future = service_executor.submit(self._run_batches_from_queue)
// wait until all batches have been processed
_ = future.result()
else:
self.prefetch_queue = None
