爲什麼僅在訪問多處理特定場景中的共享列表時發生「Broken pipe」錯誤？

Question

在我開始提問之前，讓我提一下我已經知道下面的多處理代碼被破壞了。其中有TOCTOU錯誤。以下代碼旨在爲我提供教學目的，以便我可以更多地瞭解代碼是如何被破壞的。所以我的問題是關於破碎代碼的一個特定方面。首先，讓我展示我的代碼。

現在，您可以完全忽略worker_b，因爲我們現在不在任何地方使用它。我們稍後再回來。

import Queue 
import multiprocessing 
import time 

lock = multiprocessing.Lock() 

def pprint(s): 
    lock.acquire() 
    print(s) 
    lock.release() 

def worker_a(i, stack): 
    if stack: 
     data = stack.pop() 
     pprint('worker %d got %d' % (i, data)) 
     time.sleep(2) 
     pprint('worker %d exiting ...' % i) 
    else: 
     pprint('worker %d has nothing to do!' % i) 

def worker_b(i, stack): 
    if stack: 
     data = stack.pop() 
     pprint('worker %d got %d (stack length: %d)' % (i, data, len(stack))) 
     time.sleep(2) 
     pprint('worker %d exiting ... (stack length: %d)' % (i, len(stack))) 
    else: 
     pprint('worker %d has nothing to do!' % i) 

manager = multiprocessing.Manager() 
stack = manager.list() 

def master(): 
    for i in range(5): 
     stack.append(i) 
     pprint('master put %d' % i) 

    i = 0 
    while stack: 
     t = multiprocessing.Process(target=worker_a, args=(i, stack)) 
     t.start() 
     time.sleep(1) 
     i += 1 

    pprint('master returning ...') 

master() 

pprint('master returned!') 

以上破碎的代碼似乎工作正常。

$ python mplifo.py 
master put 0 
master put 1 
master put 2 
master put 3 
master put 4 
worker 0 got 4 
worker 1 got 3 
worker 0 exiting ... 
worker 2 got 2 
worker 1 exiting ... 
worker 3 got 1 
worker 2 exiting ... 
worker 4 got 0 
worker 3 exiting ... 
master returning ... 
master returned! 
worker 4 exiting ... 

然而，如果我請worker_b代替worker_a，即改變

 t = multiprocessing.Process(target=worker_a, args=(i, stack)) 

到

 t = multiprocessing.Process(target=worker_b, args=(i, stack)) 

出現以下錯誤。

$ python mplifo.py 
master put 0 
master put 1 
master put 2 
master put 3 
master put 4 
worker 0 got 4 (stack length: 4) 
worker 1 got 3 (stack length: 3) 
worker 0 exiting ... (stack length: 3) 
worker 2 got 2 (stack length: 2) 
worker 1 exiting ... (stack length: 2) 
worker 3 got 1 (stack length: 1) 
worker 2 exiting ... (stack length: 1) 
worker 4 got 0 (stack length: 0) 
worker 3 exiting ... (stack length: 0) 
master returning ... 
master returned! 
Process Process-6: 
Traceback (most recent call last): 
    File "/usr/local/Cellar/python/2.7.13/Frameworks/Python.framework/Versions/2.7/lib/python2.7/multiprocessing/process.py", line 258, in _bootstrap 
    self.run() 
    File "/usr/local/Cellar/python/2.7.13/Frameworks/Python.framework/Versions/2.7/lib/python2.7/multiprocessing/process.py", line 114, in run 
    self._target(*self._args, **self._kwargs) 
    File "mplifo.py", line 27, in worker_b 
    pprint('worker %d exiting ... (stack length: %d)' % (i, len(stack))) 
    File "<string>", line 2, in __len__ 
    File "/usr/local/Cellar/python/2.7.13/Frameworks/Python.framework/Versions/2.7/lib/python2.7/multiprocessing/managers.py", line 758, in _callmethod 
    conn.send((self._id, methodname, args, kwds)) 
IOError: [Errno 32] Broken pipe 

• 爲什麼只出現在worker_b情況下，這個錯誤？
• 爲什麼這個錯誤只發生在第二個pprint()呼叫worker_b而不是第一個pprint()呼叫？

Answer 1

追溯的這一部分給你一個提示：

File "mplifo.py", line 27, in worker_b 
    pprint('worker %d exiting ... (stack length: %d)' % (i, len(stack))) 
    File "<string>", line 2, in __len__ 
    File "/usr/local/Cellar/python/2.7.13/Frameworks/Python.framework/Versions/2.7/lib/python2.7/multiprocessing/managers.py", line 758, in _callmethod 
    conn.send((self._id, methodname, args, kwds)) 

在工作進程，stack不是一個Python列表。它是由multiprocessing.Manager創建的代理，它包含駐留在主進程中的列表。當最後一個worker_b退出時，它會評估代理必須從主進程請求的len(stack)。但到那時，主人已經退出了 - 通往它的通訊管道被打破了。

這不會在worker_a中發生，因爲它在退出前不會嘗試評估len(stack)。