Непредсказуемое поведение при запуске очередей Python - PullRequest
Новая
       25

Непредсказуемое поведение при запуске очередей Python

0 голосов
/ 30 сентября 2019

Я пытаюсь сгенерировать различные комбинации объектов очереди на основе ввода, который содержится в переменной mode. Когда mode установлен на 'distributed', поведение программы непредсказуемо. Иногда он запускается, иногда выдает мне странные ошибки (TypeError('cannot unpack non-iterable NoneType object')), но в основном он просто зависает при создании экземпляров data_queue1, data_queue2 и output_queue. (См. Код ниже.)

Два объекта data_queue генерируются классом ModeQueue, который использует Manager для хранения очереди (ей). output_queue создается с помощью класса QueueManager, который наследуется от BaseManager. Эта очередь должна быть доступна с удаленных компьютеров. 

import multiprocessing as mp
from multiprocessing.managers import BaseManager

from lib_a import ModeQueue


class QueueManager(BaseManager):
    pass


def serve_queue(queue):
    """Start a queue server"""
    QueueManager.register('get_queue', callable=lambda: queue)
    m = QueueManager(address=('localhost', 50000), authkey=b'key')
    s = m.get_server()
    s.serve_forever()


def distributed_queue():
    """Connect to distributed queue and return object"""
    QueueManager.register('get_queue')
    m = QueueManager(address=('localhost', 50000), authkey=b'key')
    m.connect()
    return m.get_queue()


def main():
    mode = 'distributed'
    if mode == 'distributed':
        q = mp.Queue(maxsize=10)
        q_server = mp.Process(target=serve_queue, args=(q,))
        q_server.start()

        data_queue1 = ModeQueue(mode, 20)
        data_queue2 = ModeQueue(mode, 10)
        output_queue = distributed_queue()
    else:
        pass
    # more code...


if __name__ == '__main__':
    main()

# lib_a
class ModeQueue:
    def __new__(cls, mode, maxsize):
        from multiprocessing import Manager
        q = Manager().Queue(maxsize=maxsize)
        setattr(q, 'mode', mode)
        setattr(q, 'maxsize', maxsize)
        return q

Я действительно не понимаю, откуда происходит непредсказуемое поведение. Я попытался найти ошибку с обширным журналом отладки, но каждый раз, когда программа зависает, никакой вывод не записывается в мой журнал. Можете ли вы обнаружить ошибку или объяснить, почему программа не работает должным образом?

...