merge causes anext(): asynchronous generator is already running Exception about aiostream HOT 2 CLOSED

Hi @sebastianjanisch,

My guess is that an async generator somehow gets shared between task1 and task2, most probably because both tasks use the same async with self.

So if both tasks try to enter or exit at the same time, the second task won't be able to access the async generator as it's already running in the first one.

I guess you could easily check that by protecting self with an async lock and see whether the problem remains. However, the sharing of the async generator is very suspicious so there's probably another problem somewhere else.

I'm not familiar with reflex but I don't think the problem comes from aiostream. My guess is that you would observe the same issue using asyncio.gather(self.task1(), self.task2()).

Hope this helps, and feel free to re-open the issue if you have more elements pointing towards aiostream.

from aiostream.

Hi @vxgmichel thanks for responding so quickly. You are right, I slapped a asyncio lock around the async with self block (i.e. async with _lock, self: which did the trick. I'll relay to the guys from Reflex to see if they want to investigate on their side.

from aiostream.