Just call Executor.shutdown:
shutdown(wait=True)Signal the executor that it should free any resources that it is using when the currently pending futures are done executing. Calls to
Executor.submit()andExecutor.map()made after shutdown will raiseRuntimeError.If wait is
Truethen this method will not return until all the pending futures are done executing and the resources associated with the executor have been freed.
However if you keep track of your futures in a list then you can avoid shutting the executor down for future use using the futures.wait() function:
concurrent.futures.wait(fs, timeout=None, return_when=ALL_COMPLETED)Wait for the
Futureinstances (possibly created by differentExecutorinstances) given byfsto complete. Returns a named 2-tuple of sets. The first set, named done, contains the futures that completed (finished or were cancelled) before the wait completed. The second set, named not_done, contains uncompleted futures.
note that if you don't provide a timeout it waits until all futures have completed.
You can also use futures.as_completed() instead, however you'd have to iterate over it.
Just call Executor.shutdown:
shutdown(wait=True)Signal the executor that it should free any resources that it is using when the currently pending futures are done executing. Calls to
Executor.submit()andExecutor.map()made after shutdown will raiseRuntimeError.If wait is
Truethen this method will not return until all the pending futures are done executing and the resources associated with the executor have been freed.
However if you keep track of your futures in a list then you can avoid shutting the executor down for future use using the futures.wait() function:
concurrent.futures.wait(fs, timeout=None, return_when=ALL_COMPLETED)Wait for the
Futureinstances (possibly created by differentExecutorinstances) given byfsto complete. Returns a named 2-tuple of sets. The first set, named done, contains the futures that completed (finished or were cancelled) before the wait completed. The second set, named not_done, contains uncompleted futures.
note that if you don't provide a timeout it waits until all futures have completed.
You can also use futures.as_completed() instead, however you'd have to iterate over it.
As stated before, one can use Executor.shutdown(wait=True), but also pay attention to the following note in the documentation:
You can avoid having to call this method explicitly if you use the
withstatement, which will shutdown theExecutor(waiting as ifExecutor.shutdown()were called withwaitset toTrue):import shutil with ThreadPoolExecutor(max_workers=4) as e: e.submit(shutil.copy, 'src1.txt', 'dest1.txt') e.submit(shutil.copy, 'src2.txt', 'dest2.txt') e.submit(shutil.copy, 'src3.txt', 'dest3.txt') e.submit(shutil.copy, 'src4.txt', 'dest4.txt')
The call to ThreadPoolExecutor.map does not block until all of its tasks are complete. Use wait to do this.
Copyfrom concurrent.futures import wait, ALL_COMPLETED
...
futures = [pool.submit(fn, args) for args in arg_list]
wait(futures, timeout=whatever, return_when=ALL_COMPLETED) # ALL_COMPLETED is actually the default
do_other_stuff()
You could also call list(results) on the generator returned by pool.map to force the evaluation (which is what you're doing in your original example). If you're not actually using the values returned from the tasks, though, wait is the way to go.
It's true that Executor.map() will not wait for all futures to finish. Because it returns a lazy iterator like @MisterMiyagi said.
But we can accomplish this by using with:
Copyimport time
from concurrent.futures import ThreadPoolExecutor
def hello(i):
time.sleep(i)
print(i)
with ThreadPoolExecutor(max_workers=2) as executor:
executor.map(hello, [1, 2, 3])
print("finish")
# output
# 1
# 2
# 3
# finish
As you can see, finish is printed after 1,2,3. It works because Executor has a __exit__() method, code is
Copydef __exit__(self, exc_type, exc_val, exc_tb):
self.shutdown(wait=True)
return False
the shutdown method of ThreadPoolExecutor is
Copydef shutdown(self, wait=True, *, cancel_futures=False):
with self._shutdown_lock:
self._shutdown = True
if cancel_futures:
# Drain all work items from the queue, and then cancel their
# associated futures.
while True:
try:
work_item = self._work_queue.get_nowait()
except queue.Empty:
break
if work_item is not None:
work_item.future.cancel()
# Send a wake-up to prevent threads calling
# _work_queue.get(block=True) from permanently blocking.
self._work_queue.put(None)
if wait:
for t in self._threads:
t.join()
shutdown.__doc__ = _base.Executor.shutdown.__doc__
So by using with, we can get the ability to wait until all futures finish.