문제

파이썬 공식 문서에 따르면, Queue는 프로세스간 데이터를 공유할 수 있는 객체의 하나로 Process에게 전달하여 사용할 수 있습니다.

if __name__ == '__main__':
    queue = Queue()
    p1 = Process(work, (queue, ))
    p2 = Process(work, (queue, ))

    p1.start()
    p2.start()
    print(queue.get())
    print(queue.get())
    p1.join()
    p2.join()

위의 코드에서 각 프로세스가 queue에게 어떤 값을 집어넣었다면 메인 프로세스에서 이를 볼 수 있습니다.

하지만 이는 Pool에서는 그대로 수행할 수 없습니다.

if __name__ == '__main__':
    queue = Queue()
    
    with Pool(2) as p:
        p.apply(work, (queue, ))

다음을 실행하면 이런 오류가 발생합니다.

RuntimeError: Queue objects should only be shared between processes through inheritance

그렇다면 Process은 잘 받았던 Queue를 Pool은 왜 받을 수 없을까요?

Pool은 인자를 pickling 한다

def apply(self, func, args=(), kwds={}):
    '''
    Equivalent of `func(*args, **kwds)`.
    Pool must be running.
    '''
    return self.apply_async(func, args, kwds).get()


def apply_async(self, func, args=(), kwds={}, callback=None,
        error_callback=None):
    '''
    Asynchronous version of `apply()` method.
    '''
    self._check_running()
    result = ApplyResult(self, callback, error_callback)
    self._taskqueue.put(([(result._job, 0, func, args, kwds)], None))
    return result

다음 코드는 Pool.apply와 Pool.apply_async의 구현 코드입니다. 코드에서는 func와 args, kwds 모두 self._taskqueue에 집어넣는다는 것을 볼 수 있습니다.

class Pool(object):
    '''
    Class which supports an async version of applying functions to arguments.
    '''
    _wrap_exception = True

    @staticmethod
    def Process(ctx, *args, **kwds):
        return ctx.Process(*args, **kwds)

    def __init__(self, processes=None, initializer=None, initargs=(),
                 maxtasksperchild=None, context=None):
        # Attributes initialized early to make sure that they exist in
        # __del__() if __init__() raises an exception
        self._pool = []
        self._state = INIT

        self._ctx = context or get_context()
        self._setup_queues()
        self._taskqueue = queue.SimpleQueue()

그리고 self._taskqueue는 SimpleQueue 객체입니다.

그렇다면 SimpleQueue가 어떻게 저장하는지를 확인해봅시다.

def put(self, obj):
    # serialize the data before acquiring the lock
    obj = _ForkingPickler.dumps(obj)
    if self._wlock is None:
        # writes to a message oriented win32 pipe are atomic
        self._writer.send_bytes(obj)
    else:
        with self._wlock:
            self._writer.send_bytes(obj)

SimpleQueue.put의 구현 코드를 보면 _ForkingPickler.dumps를 통해 받은 obj 전체를 pickling 합니다.

여기서 obj에는 func 뿐만 아니라 args 모두 포함되어있어 pickling이 된다는 것을 확인했습니다.

즉, Pool.apply를 통해 Queue를 전달하면 Queue 객체가 pickling 되기에 이를 막기위해 오류가 발생했습니다.

그렇기 때문에 Manager 객체를 통해 pickling 되더라도 원본 객체에 접근할 수 있는 프록시 객체를 만들어서 전달하는 방법을 사용해야 합니다.