Thread-safe memory object streams

[davidbrochart]

Memory object streams are not thread-safe. Since they don't (de)serialize, they seem to be a good fit to pass data between different threads.
In a scenario where there is an event loop running in the main thread, and a separate event loop running in another thread, how can memory object streams be used to pass data between both threads?
This hangs:

from functools import partial
from anyio import create_memory_object_stream, create_task_group, run, sleep, to_thread

async def process_items(send_stream):
    async with send_stream:
        for num in range(10):
            await sleep(1)
            await send_stream.send(f"number {num}")

async def main():
    send_stream, receive_stream = create_memory_object_stream()
    async with create_task_group() as tg:
        # tg.start_soon(process_items, send_stream)  # works fine
        tg.start_soon(to_thread.run_sync, partial(run, partial(process_items, send_stream)))
        async with receive_stream:
            async for item in receive_stream:
                print("received", item)

run(main)

[agronholm]

In order to send messages between event loop threads, you would have to create a memory object stream and start a blocking portal in one event loop, and then use to_thread.run_sync() in the other one to use that portal to call send() on that memory object stream.

[davidbrochart]

In particular, to_thread.run_sync() expects a regular function, so I cannot call the memory object stream's send(), unless I wrap it in a regular function. In the end, that's what I have and it still hangs:

from anyio import start_blocking_portal, create_memory_object_stream, create_task_group, run, sleep, to_thread, from_thread

def send_wrapper(send_stream, num):
    from_thread.run(send_stream.send, f"number {num}")

async def process_items(send_stream):
    async with send_stream:
        for num in range(10):
            await sleep(1)
            await to_thread.run_sync(send_wrapper, send_stream, num)

async def main(portal):
    send_stream, receive_stream = create_memory_object_stream()
    portal.start_task_soon(process_items, send_stream)
    async with create_task_group() as tg:
        async with receive_stream:
            async for item in receive_stream:
                print("received", item)

with start_blocking_portal() as portal:
    run(main, portal)

[davidbrochart]

This doesn't use a blocking portal and seems to work:

from anyio import create_memory_object_stream, create_task_group, run, sleep, to_thread, from_thread

def process_items(send_stream, tg):
    async def main():
        async with send_stream:
            for num in range(10):
                await sleep(1)
                item = f"number {num}"
                from_thread.run_sync(tg.start_soon, send_stream.send, item)
                print("sent", item)
    run(main)

async def main():
    send_stream, receive_stream = create_memory_object_stream()
    async with create_task_group() as tg:
        tg.start_soon(to_thread.run_sync, process_items, send_stream, tg)
        async with receive_stream:
            async for item in receive_stream:
                print("received", item)

run(main)

[davidbrochart]

Ah but if the main thread blocks then the other thread blocks too... I'm kind of lost now.

[davidbrochart]

I think the solution would be something like https://github.com/aio-libs/janus.

[davidbrochart]

I found a solution, but using sockets and thread-safe queues (to do something similar to janus), instead of AnyIO's streams:

import queue
import time

import zmq.asyncio
import zmq
from anyio import create_task_group, run, sleep, to_thread


def process_items(q, socket_send):
    async def main():
        for num in range(10):
            await sleep(1)
            item = f"number {num}"
            q.put(item)
            await socket_send.send(b"")  # notify receiver
            print("sent", item)
    run(main)

async def main():
    q = queue.Queue()
    context = zmq.asyncio.Context()
    socket_recv = context.socket(zmq.PAIR)
    socket_recv.bind("inproc://event")
    socket_send = context.socket(zmq.PAIR)
    socket_send.connect("inproc://event")
    async with create_task_group() as tg:
        tg.start_soon(to_thread.run_sync, process_items, q, socket_send)
        while True:
            await socket_recv.recv()
            item = q.get()  # we know there is an item
            print("received", item)
            time.sleep(2)  # simulate blocking code

run(main)