Python:使用 ProcessPoolExecutor 执行

Question

from concurrent.futures import ProcessPoolExecutor
import os
import time

def parInnerLoop(item):
    a = 2+item
    print(f'A. {a} Processing {os.getpid()} done on {item}\n')
    exec(open('mainWork.py').read())
    print(f'D. {a} Processing {os.getpid()} done on {item}\n')

def main():
    with ProcessPoolExecutor(max_workers=4) as executor:
        for itemNo in range(10):
            executor.submit(parInnerLoop, itemNo)
    print('done')

if __name__ == '__main__':
    main()

主工作文件

print(f'B. {a} Processing {os.getpid()} done on {item}\n')
a = 12
print(f'C. {a} Processing {os.getpid()} done on {item}\n')

错误:

Error in atexit._run_exitfuncs:
Traceback (most recent call last):
  File "D:\Program Files\Python\Python37\Lib\concurrent\futures\process.py", line 101, in _python_exit
    thread_wakeup.wakeup()
  File "D:\Program Files\Python\Python37\Lib\concurrent\futures\process.py", line 89, in wakeup
    self._writer.send_bytes(b"")
  File "D:\Program Files\Python\Python37\Lib\multiprocessing\connection.py", line 183, in send_bytes
    self._check_closed()
  File "D:\Program Files\Python\Python37\Lib\multiprocessing\connection.py", line 136, in _check_closed
    raise OSError("handle is closed")
OSError: handle is closed

问题:

错误，它出现在程序的末尾。

我想执行mainWork.py的所有代码作为 parInnerLoop 的一部分，使得 a 的值发生变化应该反射(reflect)在parInnerLoop以及在 mainWork.py反之亦然。我得到如下输出。 a的值不改变打印开始的位置 D. ... ，我希望它是 12。

项目输出 0

A. 2 Processing 19784 done on 0
B. 2 Processing 19784 done on 0
C. 12 Processing 19784 done on 0
D. 2 Processing 19784 done on 0

到底是怎么回事？请任何帮助。

我最终想要实现的是:在不同的设置下并行运行相同的代码，互不干扰

额外的:

我测试了 ProcessPoolExecutor 正常工作，如图所示 here .

我测试了 exec(open('run.py').read())使用以下代码按预期工作:

主文件

def myFun():
    a = 1
    b = 2
    print(a)
    exec(open('run.py').read())
    #execfile("run.py")
    print(a)
    print(b)
    print(c)

def main():
    myFun()

if __name__ == '__main__':
    main()

运行文件

a = a+5
b = 10
c = 15

输出

1
6
10
15

Answer 1

不是完整的答案，但似乎与

Thank you for the report! After a quick investigation, it seems like the issue is because of recent changes in the concurrent.futures module. In short, they maintain a dictionary of weak references to some service threads (named "QueueManagerThread") and try to shut down whatever threads are in this dictionary when exiting a Python process. The dictionary is usually empty but if a reference to a thread exists this may cause an attempt of closing the thread which is not alive. In this case the original exception is raised.

https://youtrack.jetbrains.com/issue/PY-34432