Can I somehow share an asynchronous queue with a subprocess?

Question

I would like to use a queue for passing data from a parent to a child process which is launched via multiprocessing.Process. However, since the parent process uses Python's new asyncio library, the queue methods need to be non-blocking. As far as I understand, asyncio.Queue is made for inter-task communication and cannot be used for inter-process communication. Also, I know that multiprocessing.Queue has the put_nowait() and get_nowait() methods but I actually need coroutines that would still block the current task (but not the whole process). Is there some way to create coroutines that wrap put_nowait()/get_nowait()? On another note, are the threads that multiprocessing.Queue uses internally compatible after all with an event loop running in the same process?

If not, what other options do I have? I know I could implement such a queue myself by making use of asynchronous sockets but I hoped I could avoid that…

EDIT: I also considered using pipes instead of sockets but it seems asyncio is not compatible with multiprocessing.Pipe(). More precisely, Pipe() returns a tuple of Connection objects which are not file-like objects. However, asyncio.BaseEventLoop's methods add_reader()/add_writer() methods and connect_read_pipe()/connect_write_pipe() all expect file-like objects, so it is impossible to asynchronously read from/write to such a Connection. In contrast, the usual file-like objects that the subprocess package uses as pipes pose no problem at all and can easily be used in combination with asyncio.

UPDATE: I decided to explore the pipe approach a bit further: I converted the Connection objects returned by multiprocessing.Pipe() into file-like objects by retrieving the file descriptor via fileno() and passing it to os.fdopen(). Finally, I passed the resulting file-like object to the event loop's connect_read_pipe()/connect_write_pipe(). (There is some mailing list discussion on a related issue if someone is interested in the exact code.) However, read()ing the stream gave me an OSError: [Errno 9] Bad file descriptor and I didn't manage to fix this. Also considering the missing support for Windows, I will not pursue this any further.

How is the child process being launched?

dano
– dano

2014-07-10 22:23:57 +00:00
Commented Jul 10, 2014 at 22:23 — dano
– dano, Commented Jul 10, 2014 at 22:23
The child process is created via multiprocessing.Process.

balu
– balu

2014-07-11 06:04:19 +00:00
Commented Jul 11, 2014 at 6:04 — balu
– balu, Commented Jul 11, 2014 at 6:04

dano · Accepted Answer · 2014-07-11 19:30:25Z

Here is an implementation of a multiprocessing.Queue object that can be used with asyncio. It provides the entire multiprocessing.Queue interface, with the addition of coro_get and coro_put methods, which are asyncio.coroutines that can be used to asynchronously get/put from/into the queue. The implementation details are essentially the same as the second example of my other answer: ThreadPoolExecutor is used to make the get/put asynchronous, and a multiprocessing.managers.SyncManager.Queue is used to share the queue between processes. The only additional trick is implementing __getstate__ to keep the object picklable despite using a non-picklable ThreadPoolExecutor as an instance variable.

from multiprocessing import Manager, cpu_count
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor

def AsyncProcessQueue(maxsize=0):
    m = Manager()
    q = m.Queue(maxsize=maxsize)
    return _ProcQueue(q)   

class _ProcQueue(object):
    def __init__(self, q):
        self._queue = q
        self._real_executor = None
        self._cancelled_join = False

    @property
    def _executor(self):
        if not self._real_executor:
            self._real_executor = ThreadPoolExecutor(max_workers=cpu_count())
        return self._real_executor

    def __getstate__(self):
        self_dict = self.__dict__
        self_dict['_real_executor'] = None
        return self_dict

    def __getattr__(self, name):
        if name in ['qsize', 'empty', 'full', 'put', 'put_nowait',
                    'get', 'get_nowait', 'close']:
            return getattr(self._queue, name)
        else:
            raise AttributeError("'%s' object has no attribute '%s'" % 
                                    (self.__class__.__name__, name))

    @asyncio.coroutine
    def coro_put(self, item):
        loop = asyncio.get_event_loop()
        return (yield from loop.run_in_executor(self._executor, self.put, item))

    @asyncio.coroutine    
    def coro_get(self):
        loop = asyncio.get_event_loop()
        return (yield from loop.run_in_executor(self._executor, self.get))

    def cancel_join_thread(self):
        self._cancelled_join = True
        self._queue.cancel_join_thread()

    def join_thread(self):
        self._queue.join_thread()
        if self._real_executor and not self._cancelled_join:
            self._real_executor.shutdown()

@asyncio.coroutine
def _do_coro_proc_work(q, stuff, stuff2):
    ok = stuff + stuff2
    print("Passing %s to parent" % ok)
    yield from q.coro_put(ok)  # Non-blocking
    item = q.get() # Can be used with the normal blocking API, too
    print("got %s back from parent" % item)

def do_coro_proc_work(q, stuff, stuff2):
    loop = asyncio.get_event_loop()
    loop.run_until_complete(_do_coro_proc_work(q, stuff, stuff2))

@asyncio.coroutine
def do_work(q):
    loop.run_in_executor(ProcessPoolExecutor(max_workers=1),
                         do_coro_proc_work, q, 1, 2)
    item = yield from q.coro_get()
    print("Got %s from worker" % item)
    item = item + 25
    q.put(item)

if __name__  == "__main__":
    q = AsyncProcessQueue()
    loop = asyncio.get_event_loop()
    loop.run_until_complete(do_work(q))

Output:

Passing 3 to parent
Got 3 from worker
got 28 back from parent

As you can see, you can use the AsyncProcessQueue both synchronously and asynchronously, from either the parent or child process. It doesn't require any global state, and by encapsulating most of the complexity in a class, is more elegant to use than my original answer.

You'll probably be able to get better performance using sockets directly, but getting that working in a cross-platform way seems to be pretty tricky. This also has the advantage of being usable across multiple workers, won't require you to pickle/unpickle yourself, etc.

For what it's worth, I ended up taking this idea and building a complete library, called aioprocessing, that provides similar functionality for all the multiprocessing classes.

dano · Accepted Answer · 2014-07-11 14:15:46Z

7

The multiprocessing library isn't particularly well-suited for use with asyncio, unfortunately. Depending on how you were planning to use the multiprocessing/multprocessing.Queue, however, you may be able to replace it completely with a concurrent.futures.ProcessPoolExecutor:

import asyncio
from concurrent.futures import ProcessPoolExecutor


def do_proc_work(stuff, stuff2):  # This runs in a separate process
    return stuff + stuff2

@asyncio.coroutine
def do_work():
    out = yield from loop.run_in_executor(ProcessPoolExecutor(max_workers=1),
                                          do_proc_work, 1, 2)
    print(out)

if __name__  == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(do_work())

Output:

If you absolutely need a multiprocessing.Queue, It seems like it will behave ok when combined with ProcessPoolExecutor:

import asyncio
import time
import multiprocessing
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor


def do_proc_work(q, stuff, stuff2):
    ok = stuff + stuff2
    time.sleep(5) # Artificial delay to show that it's running asynchronously
    print("putting output in queue")
    q.put(ok)

@asyncio.coroutine
def async_get(q):
    """ Calls q.get() in a separate Thread. 

    q.get is an I/O call, so it should release the GIL.
    Ideally there would be a real non-blocking I/O-based 
    Queue.get call that could be used as a coroutine instead 
    of this, but I don't think one exists.

    """
    return (yield from loop.run_in_executor(ThreadPoolExecutor(max_workers=1), 
                                           q.get))

@asyncio.coroutine
def do_work(q):
    loop.run_in_executor(ProcessPoolExecutor(max_workers=1),
                         do_proc_work, q, 1, 2)
    coro = async_get(q) # You could do yield from here; I'm not just to show that it's asynchronous
    print("Getting queue result asynchronously")
    print((yield from coro))

if __name__  == "__main__":
    m = multiprocessing.Manager()
    q = m.Queue() # The queue must be inherited by our worker, it can't be explicitly passed in
    loop = asyncio.get_event_loop()
    loop.run_until_complete(do_work(q))

Output:

Getting queue result asynchronously
putting output in queue
3

edited Jul 11, 2014 at 14:15

answered Jul 10, 2014 at 23:12

dano

95.5k21 gold badges234 silver badges231 bronze badges

6 Comments

balu Over a year ago

I really need a queue since both processes run indefinitely and I will need to pass data from one to the other all the time. More specifically, the child process forwards queries to an SQLite database that arrive through the queue. The reasoning behind this was, ironically, to be able to run those queries asynchronously by putting them in a queue and executing them in a different process (as SQLite calls are blocking). Anyway, your 2nd suggestion looks interesting. Though, I feel it's more efficient and, considering the global state, more elegant to just use a socket.

balu Over a year ago

By the way: Could you explain why it's not possible to explicitly pass the queue to the worker?

dano Over a year ago

@balu Attempting to pass a multprocessing.Queue directly raises RuntimeError: Queue objects should only be shared between processes through inheritance. Letting the Queue be inherited in the example code works on Linux, but I found it actually hangs on Windows. Using a multprocessing.manager.Queue (which can be passed explicitly between processes) seems to work on all platforms, though. I've updated my answer to reflect that.

balu Over a year ago

Thank you for your answer. I'm a bit confused, now, because the code example in the Python docs explicitly passes a multiprocessing.Queue to multiprocessing.Process.

dano Over a year ago

@balu Yes, I'm not exactly what about the implementation makes it legal, but you're allowed to pass a Queue to the constructor of a Process or a Pool (using the initializer/initargs keyword arguments). However, if you try to pass a Queue to a pool.apply call, it will raise the RuntimeError. It appears that if you pass the queue along prior to the child process actually starting, it's allowed. In the case of pool.apply and passing it to a ProcessPoolExecutor, the child process is already started, so the Queue can't be passed.

|

Greemo · Accepted Answer · 2020-10-20 14:46:41Z

1

aiopipe (https://pypi.org/project/aiopipe/) looks like it hits the nail on the head here.

At least it helped me..

answered Oct 20, 2020 at 14:46

Greemo

4794 silver badges3 bronze badges

1 Comment

user48956 Over a year ago

That's nice, but it makes me said that it demands using its own system for starting subprocesses.

Collectives™ on Stack Overflow

Can I somehow share an asynchronous queue with a subprocess?

3 Answers 3

1 Comment

6 Comments

1 Comment

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

3 Answers 3

1 Comment

6 Comments

1 Comment

Your Answer

Sign up or log in

Post as a guest

Linked

Related