以下のコードはmax_workersを1にしてExecutorを作り、2つ仕事を与えて1つ目が終わる前に2つ目に中断を試みるものです。
import pytest
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, InterpreterPoolExecutor
@pytest.mark.parametrize(
"executor_cls",
(ProcessPoolExecutor, ThreadPoolExecutor, InterpreterPoolExecutor)
)
def test_cancel_before_start_executing(executor_cls):
import time
with executor_cls(max_workers=1) as executor:
executor.submit(time.sleep, 2) # 1つ目の仕事
time.sleep(.5)
fut = executor.submit(time.sleep, 1) # 2つ目の仕事
time.sleep(.5)
assert fut.cancel() # 中断
==================================================================== FAILURES ====================================================================
____________________________________________ test_cancel_before_start_executing[ProcessPoolExecutor] _____________________________________________
executor_cls = <class 'concurrent.futures.process.ProcessPoolExecutor'>
@pytest.mark.parametrize(
"executor_cls",
(ProcessPoolExecutor, ThreadPoolExecutor, InterpreterPoolExecutor)
)
def test_cancel_before_start_executing(executor_cls):
import time
with executor_cls(max_workers=1) as executor:
executor.submit(time.sleep, 2)
time.sleep(.5)
fut = executor.submit(time.sleep, 1)
time.sleep(.5)
> assert fut.cancel()
E assert False
E + where False = cancel()
E + where cancel = <Future at 0x7d59d31be5d0 state=running>.cancel
ProcessPoolExecutorだけが中断に失敗してしまいました。ただ以下のようにsubmitの直後に中断を試みた場合はどのExecutorでもうまく行きました。
fut = executor.submit(time.sleep, 1)
+ assert fut.cancel()
time.sleep(.5)
- assert fut.cancel()
環境
- CPython 3.14.5
- pytest 9.0.3