Блог

У меня есть код, подобный следующему,

 class _Process(multiprocessing.Process):

    STOP = multiprocessing.Manager().Event()

    def __init__(self, queue, process_fn):

        self._q = queue
        self._p = process_fn
        super().__init__()

    def run(self):

        while True:
            dat = self._q.get()
            if not dat is _Process.STOP:
                self._p(dat, self._q)
                self._q.task_done()
            else:
                self._q.task_done()
                break
  

но я не могу STOP успешно сравнить. Это неудивительно, когда я использую is , поскольку, я полагаю, is сравнивает идентификаторы объектов и из документов «… Это адрес объекта в памяти «.Итак, поскольку я использую несколько процессов, адрес памяти будет другим. (Я не могу сравнить это ни с == тем, ни с другим, и я не уверен, почему это так).

Это происходит с любым объектом, с которым я создаю Manager() , но если я использую «истинный» синглтон ( True или False или None ), он работает. Хотя это не подходящее решение, поскольку любое из этих значений может быть допустимым в очереди.

Итак, как я могу создать переменную, подобную синглетонам, которую можно сравнивать между процессами?

(Примечание: Я тоже пытался использовать выделенный класс, но получаю ошибки о том, что его невозможно обработать.)

Обновление: ответ, похоже, заключается в использовании класса, но я получал проблемы с маринованием, поскольку я пытался только с внутренним классом. Перемещение ее в область модуля исправило ошибку, и она работает нормально. — Спасибо @Schnouki!

Вот пример (и бессмысленный) использования кода, который показывает ошибку …

 def f(data, queue):
    print(data)

q = multiprocessing.JoinableQueue()

for i in range(4):
    p = _Process(q, f)
    p.daemon = True
    p.start()
    q.put(i)

q.join()

for i in range(4):
    q.put(_Process.STOP)

q.join()
  

Это странный способ использования Event объекта… Если вы не можете использовать None или логическое значение, я предлагаю вам использовать выделенный класс и протестировать тип того, что вы получаете из очереди:

 class StopProcessing(object):
    pass

#...

q.put(StopProcessing())

#...

while True:
    dat = self._q.get()
    if type(dat) is StopProcessing:
        # ...
  

Или, конечно, вы могли бы просто продолжать использовать multiprocessing.Event и проверять его тип. Однако это, вероятно, будет вводить в заблуждение кого-то другого, читающего ваш код; использование выделенного типа кажется мне намного более чистым и питоническим.

РЕДАКТИРОВАТЬ: Хорошо, так что, по-видимому, это не работает, потому что новый класс не может быть выбран. Итак, вот еще одна идея: что, если вы напрямую поместите тип в свою очередь, вот так:

 class StopProcessing(object):
    pass
#...
q.put(StopProcessing)
#...
while True:
    dat = self._q.get()
    if dat is StopProcessing:
        #...
  

Согласно документу pickle, «классы, определенные на верхнем уровне модуля», могут быть изменены.