with
инструкция убирает за ThreadPoolExecutor
: вызывает на выходе pool.shutdown(wait=True)
. Она не может вернуться пока задачи в пуле выполняются.
Чтобы это поправить, нужно заменить return
на yield
and использовать print(next(calc()))
.
Если последующие результаты не нужны, то необходимо предусмотреть возможность отмены уже запущенных задач.
Очевидный код с concurrent.futures.ThreadPoolExecutor
не заработал (см. код в конце вопроса). Вот похожий пример c multiprocessing
ThreadPool
, который работает:
#!/urs/bin/env python3
import contextlib
import logging
import time
from multiprocessing.pool import ThreadPool as Pool
@contextlib.contextmanager
def logged(message):
logging.info('before {}'.format(message))
try:
yield
finally:
logging.info('after {}'.format(message))
def foo(a):
with logged('sleep'):
time.sleep(a)
return a
def calc():
with Pool(2) as pool:
for result in pool.imap_unordered(foo, [10, 1]):
with logged('result'):
yield result
logging.basicConfig(format="%(relativeCreated)5d %(threadName)s %(message)s",
level=logging.DEBUG)
with logged('calc'):
print(next(calc()), flush=True)
Результат
4 MainThread before calc
8 Thread-2 before sleep
8 Thread-1 before sleep
1010 Thread-1 after sleep
1010 MainThread before result
1010 MainThread after result
1
1109 MainThread after calc
Единичка показана после одной секунды. Последующие задачи отменены.
Нерабочий код с concurrent.futures
пулом:
#!/urs/bin/env python3
import contextlib
import logging
import time
from concurrent.futures import ThreadPoolExecutor as Pool, as_completed
@contextlib.contextmanager
def logged(message):
logging.info('before {}'.format(message))
try:
yield
finally:
logging.info('after {}'.format(message))
def foo(a):
with logged('sleep'):
time.sleep(a)
return a
def calc():
with Pool(2) as pool:
for future in as_completed(pool.submit(foo, i) for i in [10, 1]):
with logged('result'):
yield future.result()
logging.basicConfig(format="%(relativeCreated)5d %(threadName)s %(message)s",
level=logging.DEBUG)
with logged('calc'):
print(next(calc()), flush=True)
Результат
43 MainThread before calc
43 Thread-1 before sleep
44 Thread-2 before sleep
1045 Thread-2 after sleep
1045 MainThread before result
1046 MainThread after result
10053 Thread-1 after sleep
1
10054 MainThread after calc
Единичка показана только после 10 секунд.
Вариант с multiprocessing
ThreadPool
работает в этом случае даже если использовать return
. with
инструкция вызывает на выходе pool.terminate()
в данном случае:
#!/urs/bin/env python3
import contextlib
import logging
import time
from multiprocessing.pool import ThreadPool as Pool
@contextlib.contextmanager
def logged(message):
logging.info('before {}'.format(message))
try:
yield
finally:
logging.info('after {}'.format(message))
def foo(a):
with logged('sleep'):
time.sleep(a)
return a
def calc():
with Pool(2) as pool:
for result in pool.imap_unordered(foo, [10, 1]):
with logged('result'):
return result
logging.basicConfig(format="%(relativeCreated)5d %(threadName)s %(message)s",
level=logging.DEBUG)
with logged('calc'):
print(calc(), flush=True)
Результат
10 MainThread before calc
13 Thread-2 before sleep
13 Thread-1 before sleep
1014 Thread-1 after sleep
1014 MainThread before result
1015 MainThread after result
1
1114 MainThread after calc
Единичка показана после одной секунды.