Python concurrent.futures饥饿

Question

我使用Python concurrent.futures，执行父多线程，每个父线程执行子线程。当ThreadPoolExecutor少于所需的父线程数时，我会遇到饥饿和程序卡住。

实现以下目标的最佳方法是什么：

1. 使用const ThreadPoolExecutor

1. Do not get starvation

请查看以下示例代码：

import time
import sys
import concurrent.futures


MAX_THREAD_EXECUTORS = 5
threadPool = concurrent.futures.ThreadPoolExecutor(MAX_THREAD_EXECUTORS)
threads = []
command_threads = []


def main():
    start_tests()
    join_threads()


def start_tests():
    for i in range(1,14):
       threads.append(threadPool.submit(start_test_flow, i))


def start_test_flow(test):
    print(f"Start test flow for: {test}")
    execute_commands()
    join_command_threads()
    

def execute_commands():
    for i in range(1,5):
        command_threads.append(threadPool.submit(start_command, i))


def start_command(command):
    print(f"Start command for: {command}")
    time.sleep(120)


def join_threads():
    for thread in threads:
        result = thread.result()
        print(f"test result={result}")


def join_command_threads():
    for thread in command_threads:
        result = thread.result()
        print(f"command result={result}")

if __name__ == '__main__':
    main()
    sys.exit(0)

致以最好的问候，摩西

Answer 1

您实际需要的最小线程数量是不确定的，并且取决于时间，尽管有一个数字(13 + 1，即每个父线程一个线程，至少一个线程运行子线程)可以保证您永远不会停止。最可能发生的情况是，您正在快速创建5个父线程，然后等待创建更多的父线程和子线程，因为您只有5个工作线程。但是直到你能够创建4个子线程(在execute_commands中)并运行它们直到完成，父线程才会完成，因此你被卡住了。

现在，例如，在函数start_tests中插入对time.sleep(1)的调用，如下所示：

def start_tests():
    for i in range(1,14):
       threads.append(threadPool.submit(start_test_flow, i))
       time.sleep(1)

这将允许创建4个子线程，并且会有一些进展。但根据时间的不同，你可能最终会停滞不前。为了保证您永远不会停顿，您必须休眠足够长的时间，以便在尝试启动下一个父线程之前让所有4个子线程完成。

底线是你没有足够的工作线程(13 + 1)来保证你不会停顿。