Блог

Я пытаюсь создать файлоподобный объект, который должен быть назначен 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: %rn' % (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% случаев, но имеет три проблемы с граничным случаем. Я должен запустить тестовый скрипт в быстром цикле while, чтобы воспроизвести их.

1. В 3% случаев выходные данные родительского процесса отражаются не полностью. Я предполагаю, что это связано с тем, что данные используются до того, как поток очистки очереди сможет их догнать. Я не нашел способа дождаться потока без взаимоблокировки.
2. . в 5% случаев происходит обратная трассировка от мультипроцесса.Реализация очереди
3. .01% времени PID-идентификаторы изменяются, и поэтому сортировка по PID приводит к неправильному порядку.

В самом худшем случае (вероятность: один к 70 миллионам) результат будет выглядеть следующим образом:

 START

DONE

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

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
  

Как мне избавиться от этих крайних случаев?

Решение состояло из двух частей. Я успешно запустил тестовую программу 200 тысяч раз без каких-либо изменений в выходных данных.

Проще всего было использовать multiprocessing.current_process()._identity для сортировки сообщений. Это не часть опубликованного API, но это уникальный, детерминированный идентификатор каждого процесса. Это устранило проблему с переносом PIDS и неправильным порядком вывода.

Другой частью решения было использование многопроцессорности.Manager().Queue() вместо многопроцессорной обработки.Очередь. Это устраняет проблему № 2, описанную выше, поскольку менеджер находится в отдельном процессе, и таким образом позволяет избежать некоторых неприятных особых случаев при использовании очереди из процесса-владельца. Исправлено # 3, поскольку очередь полностью исчерпана, и поток фидера естественным образом умирает до того, как python начнет завершать работу и закроет stdin.

Я столкнулся с гораздо меньшим количеством multiprocessing ошибок с Python 2.7, чем с Python 2.6. Сказав это, решение, которое я использовал, чтобы избежать « Exception in thread QueueFeederThread » проблемы, заключается в том, чтобы sleep мгновенно, возможно, в течение 0,01 с, в каждом процессе, в котором используется Queue . Это правда, что использование sleep нежелательно или даже ненадежно, но было замечено, что указанная продолжительность на практике работает достаточно хорошо для меня. Вы также можете попробовать 0,1 секунды.