Python ThreadPoolExecutor 关闭行为因调用位置而异

Question

我已经运行了两种代码变体，对我来说，它们应该完全相同地运行 - 所以我很惊讶地看到每个代码的不同输出......

第一：

from concurrent.futures import ThreadPoolExecutor
from time import sleep

executor = ThreadPoolExecutor(max_workers=2)

def func(x):
    print(f"In func {x}")
    sleep(1)
    return True

foo = executor.map(func, range(0, 10))

for f in foo:
    print(f"blah {f}")
    if f:
        break

print("Shutting down")
executor.shutdown(wait=False)
print("Shut down")

这将输出以下内容 - 显示剩余的期货正在运行完成。虽然一开始这让我感到惊讶，但我相信它与文档一致（在没有将 cancel_futures 设置为 True 的情况下），根据 https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.Executor.shutdown“无论等待值如何，整个 Python 程序都不会退出直到所有未决期货执行完毕。”

In func 0
In func 1
In func 2
In func 3
blah True
Shutting down
Shut down
In func 4
In func 5
In func 6
In func 7
In func 8
In func 9

所以没关系。但奇怪的是——如果我重构为在一个函数中调用它，它的行为会有所不同。见小调整：

from concurrent.futures import ThreadPoolExecutor
from time import sleep

def run_test():
    executor = ThreadPoolExecutor(max_workers=2)

    def func(x):
        print(f"In func {x}")
        sleep(1)
        return True

    foo = executor.map(func, range(0, 10))

    for f in foo:
        print(f"blah {f}")
        if f:
            break

    print("Shutting down")
    executor.shutdown(wait=False)
    print("Shut down")

run_test()

这输出以下内容，表明未来是在这种情况下取​​消

In func 0
In func 1
In func 2
blah True
Shutting down
In func 3
Shut down

所以我猜当执行者在 run_test() 结束时超出范围时发生了什么事？但这似乎与文档相矛盾（没有提到这一点），并且执行者肯定在第一个脚本的末尾同样超出了范围？

见于 Python 3.8 和 3.9。

我预计两种情况下的输出相同，但它们不匹配

Answer 1

这也让我很吃惊。此代码还会重现您的行为

from concurrent.futures import ThreadPoolExecutor
from time import sleep


def run_test():
    executor = ThreadPoolExecutor(max_workers=2)

    def func(x):
        print(f"In func {x}")
        sleep(1)

    foo = executor.map(func, range(0, 10))

    # a
    x = next(foo)
    # b

    print("Shutting down")
    executor.shutdown(wait=False)
    print("Shut down")

run_test()

如果按原样运行它，它将运行 0 到 10 之间的前几个整数，然后退出。如果您注释掉 #a 和 #b 之间的行，那么它将运行全部 10。

据我所知，原因是，如果您完全遍历生成器对象 (foo)（或对其调用 next()），那么代码将在 CPython 并发的 this 迭代器函数中结束.futures._base 源代码。

当 run_test() 函数退出并且 foo 超出范围时，您最终会进入 this finally 块，这会取消所有未决的期货。

在您没有函数的示例中，我相信您的猜测是正确的，它与对象超出范围的顺序有关。您可以通过评论/取消评论下面# a 和# b 之间的行来看到这一点

from concurrent.futures import ThreadPoolExecutor
from time import sleep

executor = ThreadPoolExecutor(max_workers=2)

def func(x):
    print(f"In func {x}")
    sleep(1)
    return True

foo = executor.map(func, range(0, 10))

next(foo)

# a
# del foo
# b

print("Shutting down")
executor.shutdown(wait=False)
print("Shut down")