多处理，ATEXIT和全局数据出错

Question

预先抱歉，这会很长...

可能相关：

Python多处理atexit错误“ atexit._run_exitfuncs中的错误”

绝对相关：

带有全局数据的python并行映射（multiprocessing.Pool.map）

键盘中断与python的多处理池

这是我一起破解的一个“简单”脚本，用以说明我的问题...

import time
import multiprocessing as multi
import atexit

cleanup_stuff=multi.Manager().list([])

##################################################
# Some code to allow keyboard interrupts  
##################################################
was_interrupted=multi.Manager().list([])
class _interrupt(object):
    """
    Toy class to allow retrieval of the interrupt that triggered it's execution
    """
    def __init__(self,interrupt):
        self.interrupt=interrupt

def interrupt():
    was_interrupted.append(1)

def interruptable(func):
    """
    decorator to allow functions to be "interruptable" by
    a keyboard interrupt when in python's multiprocessing.Pool.map
    **Note**, this won't actually cause the Map to be interrupted,
    It will merely cause the following functions to be not executed.
    """
    def newfunc(*args,**kwargs):
        try:
            if(not was_interrupted):
                return func(*args,**kwargs)
            else:
                return False
        except KeyboardInterrupt as e:
            interrupt()
            return _interrupt(e)  #If we really want to know about the interrupt...
    return newfunc

@atexit.register
def cleanup():
    for i in cleanup_stuff:
        print(i)
    return

@interruptable
def func(i):
    print(i)
    cleanup_stuff.append(i)
    time.sleep(float(i)/10.)
    return i

#Must wrap func here, otherwise it won't be found in __main__'s dict
#Maybe because it was created dynamically using the decorator?
def wrapper(*args):
    return func(*args)


if __name__ == "__main__":

    #This is an attempt to use signals -- I also attempted something similar where
    #The signals were only caught in the child processes...Or only on the main process...
    #
    #import signal
    #def onSigInt(*args): interrupt()
    #signal.signal(signal.SIGINT,onSigInt)

    #Try 2 with signals (only catch signal on main process)
    #import signal
    #def onSigInt(*args): interrupt()
    #signal.signal(signal.SIGINT,onSigInt)
    #def startup(): signal.signal(signal.SIGINT,signal.SIG_IGN)
    #p=multi.Pool(processes=4,initializer=startup)

    #Try 3 with signals (only catch signal on child processes)
    #import signal
    #def onSigInt(*args): interrupt()
    #signal.signal(signal.SIGINT,signal.SIG_IGN)
    #def startup(): signal.signal(signal.SIGINT,onSigInt)
    #p=multi.Pool(processes=4,initializer=startup)


    p=multi.Pool(4)
    try:
        out=p.map(wrapper,range(30))
        #out=p.map_async(wrapper,range(30)).get()  #This doesn't work either...

        #The following lines don't work either
        #Effectively trying to roll my own p.map() with p.apply_async 
        # results=[p.apply_async(wrapper,args=(i,)) for i in range(30)]
        # out = [ r.get() for r in results() ]
    except KeyboardInterrupt:
        print ("Hello!")
        out=None
    finally:
        p.terminate()
        p.join()

    print (out)

如果没有引发KeyboardInterrupt，这将很好地工作。但是，如果我提出一个，则会发生以下异常：

10
7
9
12
^CHello!
None
Error in atexit._run_exitfuncs:
Traceback (most recent call last):
  File "/usr/lib/python2.6/atexit.py", line 24, in _run_exitfuncs
    func(*targs, **kargs)
  File "test.py", line 58, in cleanup
    for i in cleanup_stuff:
  File "<string>", line 2, in __getitem__
  File "/usr/lib/python2.6/multiprocessing/managers.py", line 722, in _callmethod
    self._connect()
  File "/usr/lib/python2.6/multiprocessing/managers.py", line 709, in _connect
    conn = self._Client(self._token.address, authkey=self._authkey)
  File "/usr/lib/python2.6/multiprocessing/connection.py", line 143, in Client
   c = SocketClient(address)
  File "/usr/lib/python2.6/multiprocessing/connection.py", line 263, in SocketClient
   s.connect(address)
  File "<string>", line 1, in connect
error: [Errno 2] No such file or directory
Error in sys.exitfunc:
Traceback (most recent call last):
  File "/usr/lib/python2.6/atexit.py", line 24, in _run_exitfuncs
    func(*targs, **kargs)
  File "test.py", line 58, in cleanup
    for i in cleanup_stuff:
  File "<string>", line 2, in __getitem__
  File "/usr/lib/python2.6/multiprocessing/managers.py", line 722, in _callmethod
    self._connect()
  File "/usr/lib/python2.6/multiprocessing/managers.py", line 709, in _connect
    conn = self._Client(self._token.address, authkey=self._authkey)
  File "/usr/lib/python2.6/multiprocessing/connection.py", line 143, in Client
    c = SocketClient(address)
  File "/usr/lib/python2.6/multiprocessing/connection.py", line 263, in SocketClient
    s.connect(address)
  File "<string>", line 1, in connect
socket.error: [Errno 2] No such file or directory

有趣的是，该代码确实退出了Pool.map函数，而没有调用任何其他函数……问题似乎是在某些时候无法正确处理KeyboardInterrupt，但这在哪里有点令人困惑，并且为什么不以可中断方式处理它。谢谢。

请注意，如果我使用同样的问题 out=p.map_async(wrapper,range(30)).get()

编辑1

稍微靠近一点……如果我将out=p.map(...)in放在一个try,except,finally子句中，它将摆脱第一个异常……但是其他异常仍在atexit中引发。上面的代码和追溯已更新。

编辑2

其他无效的内容已作为注释添加到上面的代码中。（相同的错误）。此尝试的灵感来自：

http://jessenoller.com/2009/01/08/multiprocessingpool-and-keyboardinterrupt/

编辑3

使用添加到上述代码中的信号的另一次尝试失败。

编辑4

我已经弄清楚了如何重组我的代码，以便不再需要上面的代码。在（不太可能）的事件中，有人以与我相同的用例偶然发现此线程，我将描述我的解决方案...

用例

我有一个使用该tempfile模块生成临时文件的功能。我希望在程序退出时清除这些临时文件。我最初的尝试是将每个临时文件名打包到一个列表中，然后使用通过注册的函数删除列表中的所有元素atexit.register。问题在于更新的列表没有跨多个进程进行更新。这是我multiprocessing.Manager用来管理列表数据的主意。不幸的是，KeyboardInterrupt无论我怎么努力，这都失败了，因为进程之间的通信套接字由于某种原因而中断了。解决这个问题很简单。在使用多重处理之前，请设置临时文件目录...类似tempfile.tempdir=tempfile.mkdtemp()然后注册删除临时目录的功能。每个进程都写入相同的临时目录，因此可以正常工作。 当然，此解决方案仅在共享数据是需要在程序寿命结束时删除的文件列表的情况下才有效。