7

確定的な出力を提供するために、テスト中に sys.stdout/sys.stderr に割り当てられることを意図したオブジェクトのようなファイルを作成しようとしています。高速であることを意図したものではなく、信頼性が高いだけです。これまでのところ、ほとんど機能していますが、最後のいくつかのエッジケースエラーを取り除くには、助けが必要です.

これが私の現在の実装です。

try:
    from cStringIO import StringIO
except ImportError:
    from StringIO import StringIO

from os import getpid
class MultiProcessFile(object):
    """
    helper for testing multiprocessing

    multiprocessing poses a problem for doctests, since the strategy
    of replacing sys.stdout/stderr with file-like objects then
    inspecting the results won't work: the child processes will
    write to the objects, but the data will not be reflected
    in the parent doctest-ing process.

    The solution is to create file-like objects which will interact with
    multiprocessing in a more desirable way.

    All processes can write to this object, but only the creator can read.
    This allows the testing system to see a unified picture of I/O.
    """
    def __init__(self):
        # per advice at:
        #    http://docs.python.org/library/multiprocessing.html#all-platforms
        from multiprocessing import Queue
        self.__master = getpid()
        self.__queue = Queue()
        self.__buffer = StringIO()
        self.softspace = 0

    def buffer(self):
        if getpid() != self.__master:
            return

        from Queue import Empty
        from collections import defaultdict
        cache = defaultdict(str)
        while True:
            try:
                pid, data = self.__queue.get_nowait()
            except Empty:
                break
            cache[pid] += data
        for pid in sorted(cache):
            self.__buffer.write( '%s wrote: %r\n' % (pid, cache[pid]) )
    def write(self, data):
        self.__queue.put((getpid(), data))
    def __iter__(self):
        "getattr doesn't work for iter()"
        self.buffer()
        return self.__buffer
    def getvalue(self):
        self.buffer()
        return self.__buffer.getvalue()
    def flush(self):
        "meaningless"
        pass

...そして簡単なテストスクリプト:

#!/usr/bin/python2.6

from multiprocessing import Process
from mpfile import MultiProcessFile

def printer(msg):
    print msg

processes = []
for i in range(20):
    processes.append( Process(target=printer, args=(i,), name='printer') )

print 'START'
import sys
buffer = MultiProcessFile()
sys.stdout = buffer

for p in processes:
    p.start()
for p in processes:
    p.join()

for i in range(20):
    print i,
print

sys.stdout = sys.__stdout__
sys.stderr = sys.__stderr__
print 
print 'DONE'
print
buffer.buffer()
print buffer.getvalue()

これは 95% の確率で完全に機能しますが、3 つの特殊な問題があります。これらを再現するには、テスト スクリプトを高速な while ループで実行する必要があります。

  1. 3% の確率で、親プロセスの出力が完全に反映されていません。これは、Queue-flushing スレッドが追いつく前にデータが消費されているためだと思います。デッドロックせずにスレッドを待つ方法はありません。
  2. .5% の確率で、multiprocess.Queue 実装からのトレースバックがあります。
  3. 0.01% の確率で、PID がラップアラウンドするため、PID でソートすると間違った順序になります。

最悪の場合 (確率: 7000 万分の 1)、出力は次のようになります。

START

DONE

302 wrote: '19\n'
32731 wrote: '0 1 2 3 4 5 6 7 8 '
32732 wrote: '0\n'
32734 wrote: '1\n'
32735 wrote: '2\n'
32736 wrote: '3\n'
32737 wrote: '4\n'
32738 wrote: '5\n'
32743 wrote: '6\n'
32744 wrote: '7\n'
32745 wrote: '8\n'
32749 wrote: '9\n'
32751 wrote: '10\n'
32752 wrote: '11\n'
32753 wrote: '12\n'
32754 wrote: '13\n'
32756 wrote: '14\n'
32757 wrote: '15\n'
32759 wrote: '16\n'
32760 wrote: '17\n'
32761 wrote: '18\n'

Exception in thread QueueFeederThread (most likely raised during interpreter shutdown):
Traceback (most recent call last):
  File "/usr/lib/python2.6/threading.py", line 532, in __bootstrap_inner
  File "/usr/lib/python2.6/threading.py", line 484, in run
      File "/usr/lib/python2.6/multiprocessing/queues.py", line 233, in _feed
<type 'exceptions.TypeError'>: 'NoneType' object is not callable

python2.7 では、例外が少し異なります。

Exception in thread QueueFeederThread (most likely raised during interpreter shutdown):
Traceback (most recent call last):
  File "/usr/lib/python2.7/threading.py", line 552, in __bootstrap_inner
  File "/usr/lib/python2.7/threading.py", line 505, in run
  File "/usr/lib/python2.7/multiprocessing/queues.py", line 268, in _feed
<type 'exceptions.IOError'>: [Errno 32] Broken pipe

これらのエッジケースを取り除くにはどうすればよいですか?

4

2 に答える 2

9

解決策は 2 つの部分に分かれていました。テスト プログラムを 20 万回実行しても、出力に変化はありませんでした。

簡単な部分は、 multiprocessing.current_process()._identity を使用してメッセージをソートすることでした。これは公開された API の一部ではありませんが、各プロセスの一意の決定論的な識別子です。これにより、PID がラップアラウンドし、出力の順序が正しくないという問題が修正されました。

ソリューションの他の部分は、multiprocessing.Queue ではなく multiprocessing.Manager().Queue() を使用することでした。これにより、上記の問題 #2 が修正されます。これは、マネージャーが別のプロセスに存在し、所有プロセスから Queue を使用する際の悪い特殊なケースのいくつかを回避するためです。＃3は、キューが完全に使い果たされ、Pythonがシャットダウンを開始してstdinを閉じる前にフィーダースレッドが自然に死ぬため、修正されました。

于 2011-04-29T15:29:37.877 に答える
0

multiprocessingPython 2.6 よりも Python 2.7 の方がはるかに少ないバグに遭遇しました。Exception in thread QueueFeederThreadそうは言っても、" " 問題を回避するために私が使用した解決策はsleep、 が使用される各プロセスで、おそらく 0.01 秒間、一時的に実行することQueueです。使用が望ましくない、または信頼性が低いことは事実ですsleepが、指定された期間は実際には十分に機能することが観察されました。0.1s を試すこともできます。

于 2012-07-17T14:52:14.130 に答える