如何解决创建过多进程间管道的问题？

Question

我对 Python 还很陌生，所以请不要评判我 :) 我是一名网络工程师，正在开发 Python 程序，该程序连接到 600 多个设备以收集某些数据。它正在使用多处理。我决定基本上一次运行，因为我使用的服务器可以很好地处理这种方法，它极大地加快了整个过程。我刚刚注意到我的程序使用了大量的文件描述符（超过 200K）。我能够追踪到正在创建的进程间（我假设）管道。看起来每个连续生成的进程不仅向其父进程（即主程序）打开管道，还向在他之前生成的所有“兄弟”打开管道。老实说，我不知道这是正常行为还是我的代码有问题。我确实需要进程将数据发送回它们的父级，但我绝对不需要它们相互通信。当我生成子进程时，我能做些什么来防止创建完整的管道网格？

这说明了问题：

root@vf1netcat2:~# lsof -p 15531 | grep 管道
python3 15531 netcat 4r FIFO 0,12 0t0 334887 管道
python3 15531 netcat 5w FIFO 0,12 0t0 334887 管道
python3 15531 netcat 6r FIFO 0,12 0t0 334888 管道
python3 15531 netcat 7w FIFO 0,12 0t0 334888 管道
python3 15531 netcat 9w FIFO 0,12 0t0 334889 管道

root@vf1netcat2:~# lsof -p 15532 | grep 管道
python3 15532 netcat 4r FIFO 0,12 0t0 334887 管道
python3 15532 netcat 5w FIFO 0,12 0t0 334887 管道
python3 15532 netcat 6r FIFO 0,12 0t0 334888 管道
python3 15532 netcat 7w FIFO 0,12 0t0 334888 管道
python3 15532 netcat 8r FIFO 0,12 0t0 334889 管道
python3 15532 netcat 10w FIFO 0,12 0t0 334890 管道

root@vf1netcat2:~# lsof -p 15533 | grep 管道
python3 15533 netcat 4r FIFO 0,12 0t0 334887 管道
python3 15533 netcat 5w FIFO 0,12 0t0 334887 管道
python3 15533 netcat 6r FIFO 0,12 0t0 334888 管道
python3 15533 netcat 7w FIFO 0,12 0t0 334888 管道
python3 15533 netcat 8r FIFO 0,12 0t0 334889 管道
python3 15533 netcat 9r FIFO 0,12 0t0 334890 管道
python3 15533 netcat 11w FIFO 0,12 0t0 334891 管道

...

root@vf1netcat2:~# lsof -p 15735 | grep 管道 | wc -l
209
root@vf1netcat2:~# lsof -p 16035 | grep 管道 | wc -l
509
root@vf1netcat2:~# lsof -p 16100 | grep 管道 | wc -l


root@vf1netcat2:~# lsof | wc -l
3838
root@vf1netcat2:~# lsof | wc -l
184686
root@vf1netcat2:~# lsof | wc -l
237127
root@vf1netcat2:~# lsof | wc -l
228187

这就是我生成子进程的方式：

def execute_data_processing_function(data_list: List[Any], data_processing_function: Callable[..., List[Any]], *args: Any, **kwargs: Any) -> List[Any]:
    """ Execute generic data processing function in single or multiprocess manner and return merged list of results """

    if SINGLE_PROCESS_MODE:
        results = [data_processing_function(_, *args, **kwargs) for _ in data_list]

    else:
        with concurrent.futures.ProcessPoolExecutor(max_workers=len(data_list)) as executor:
            process_pool = [executor.submit(data_processing_function, _, *args, **kwargs) for _ in data_list]

        results = [_.result() for _ in process_pool if not _.exception() and _.result()]

    return [_ for __ in results for _ in __]

Answer 1

您可能想查看set_start_method()。使用'forkserver' 方法将消除额外的管道文件描述符。