AsyncConnection.notifies() gernerator blocks AsyncConnection.execute()

[saschathiergart]

I have a use case where a single AsyncConnection needs to listen on different channels. Which to listen on becomes known in runtime, e.g. connected clients determine what is listened and unlistened on. Unfortunately, when using the AsynConnection.notifies() generator, further AsyncConnection.execute() calls will hang.

async def test_notifies():
    con = await psycopg.AsyncConnection.connect(PSYCOPG_DSN, autocommit=True)
    await con.set_autocommit(True)

    async def notifier():
        i = 0
        while True: 
            await con.execute(f'notify foo, \'{i}\'')
            print(f"notified {i}")
            await asyncio.sleep(1)
            i += 1
            if i==10:
                break
        await con.close()
        
    async def receiver():
        print("receiver: LISTENING ")
        cur = con.cursor()
        await cur.execute("listen foo")
        gen = con.notifies()
        async for n in gen:
            print("receiver: notified ")
            print(n)

    workers = [notifier(), receiver()]
    await asyncio.gather(*workers)
    
asyncio.run(test_notifies())

Output of the program:

receiver: LISTENING 
notified 0

After that the program hangs.

[dvarrazzo]

Yes, you are not supposed to run queries while you are listening.

If you need to issue a new LISTEN on that connection, you can stop the notifies generator (using generators.close(), see docs), execute the statement, and call notifies() again. You won't lose messages doing so: they will be buffered in the libpq connection anyway.

[saschathiergart]

Thank you @dvarrazzo for the suggestion. How do you close a running generator outside of the generator, i.e. outside of async for n in gen:?

Example:

async def test_notifies():
	con = await psycopg.AsyncConnection.connect(PSYCOPG_DSN, autocommit=True)
	await con.set_autocommit(True)
	ctx = {'gen': None}

	async def notifier(ctx):
		await asyncio.sleep(1)
		i = 0
		while True: 
			gen = ctx.get("gen")
			if gen:
				await gen.aclose()
			cur = con.cursor()
			await cur.execute(f'notify foo, \'{i}\'')
			ctx['gen']=con.notifies()
			await asyncio.sleep(1)
			i += 1
			if i==10:
				break
		await con.close()
		

	async def receiver(ctx):
		print("receiver: LISTENING ")
		cur = con.cursor()
		await cur.execute("listen foo")
		gen = con.notifies()
		ctx['gen']=gen
		async for n in gen:
			print("receiver: notified ")
			print(n)

	workers = [notifier(ctx), receiver(ctx)]
	await asyncio.gather(*workers)
	
asyncio.run(test_notifies())

will result in a RunTimeError: RuntimeError: aclose(): asynchronous generator is already running.

I am wondering if using a generator for notifies is actually the right concept. Maybe using an event emitter like approach would make it more useful.

[dvarrazzo]

How do you close a running generator outside of the generator, i.e. outside of async for n in gen:?

Maybe you could use a different task to call gen.close().

I am wondering if using a generator for notifies is actually the right concept. Maybe using an event emitter like approach would make it more useful.

If you wanted to do that you can use add_notify_handler() and then use select() to wait for the connection to have activity, and run a select 1 to gather the notifies received.

[saschathiergart]

Maybe you could use a different task to call gen.close().

As far as i understand, i am doing that in my example. This makes the AsyncConnection.notifies only usable for special use cases.

If you wanted to do that you can use add_notify_handler() and then use select() to wait for the connection to have activity, and run a select 1 to gather the notifies received.

Sure, add_notify_handler() could be used but executing select 1 to receive notifications is an overhead that can become big, if the application needs to get that notifications quickly. Not really an ideal solution imo.

[dvarrazzo]

As far as i understand, i am doing that in my example.

Ah, sorry, I missed the aclose(). Anyway, that's strange.

There is a test in the test suite using that featue: do you want to check what it is doing?

I will try to put together a small demo program doing what you need to do (closing and reopening a notifies generator). But I don't have time right now. leaving the ticket open as a reminder.

[saschathiergart]

Yes i checked the test and my example is inspired by that. There you are closing the generator within the generator loop and im my example i am trying closing it from outside.

[saschathiergart]

Thinking about it, wouldn't it make sense to declare notifies as asyncio.Queue ? and put notification in the queue as they appear, thus isolating the notifications from the connection? Maybe the generator approach was taken over from psycopg2 but imo a queue fits better to the requirement of an inherently non-blocking asynchronous database connector.

[dvarrazzo]

The problem you are having here is that you want to be able to send statements on the same connection you are listening from. In that moment you have to stop listening, send the statement, and start listening again.

I suspect we can do this operation with the primitives that psycopg exposes. I can try and put together a recipe for it. If it turns out that that it can't be done, we can think about other approach. If the pattern is clean enough we might make it into ab exposed functionality.

Being the producer a queue is just an interface, which can be created using whatever component available; I don't think it changes radically the problem of stopping and restarting the listening. If you think otherwise, I invite you to look at how generators.notifies() is implemented: you will see that it basically just two calls to libpq functions, organised in a way to be called as soon as the connection is ready for reading. You can try to create your own notifies producer out of them, and if you come out with something that is easier to compose with other code I'm happy to consider its inclusion in thel library 🙂

[jamesfmilne]

The notifies() generator has an outer loop inside which it fetches a batch of notifications, then yields them.

The generator only yields Notify objects, but if it yielded a tuple (Notify, bool), where the second item is a flag indicating that it's the last Notify in the batch, then you could close the generator after the last notification in a batch?

Alternatively, for my purposes, it might be enough to add a timeout parameter to notifies(). I want to interleave waiting on notifications with work doing periodic background tasks which involve queries on the same connection. A timeout could cause the generator to exit.

[dvarrazzo]

I made some tests about this problem, some time ago, but I can't find my finding back. I seem to remember I could cancel the async version using this pattern and I can't remember if I could cancel successfully the sync one.

Adding a timeout parameter might be a good idea too.

[AlanCoding]

I hit a version of this issue because we had existing non-async code using an old recipe for psycopg2. In that case, using select.select you wait until notification(s) come, and then loop over the current queue. Once you've exhausted the notifications, you can do other stuff before waiting for more messages.

The headache with this library is that I don't want that outer while True: here:

psycopg/psycopg/psycopg/connection.py

Lines 918 to 927 in 2ca04b7

	while True:
	with self.lock:
	try:
	ns = self.wait(notifies(self.pgconn))
	except e._NO_TRACEBACK as ex:
	raise ex.with_traceback(None)
	enc = pgconn_encoding(self.pgconn)
	for pgn in ns:
	n = Notify(pgn.relname.decode(enc), pgn.extra.decode(enc), pgn.be_pid)
	yield n

I have existing machinery (that still works) to wait until a notification is received. You can follow my specific problem at ansible/awx#14222, where I duplicated the method so I could remove that outer loop. So far it works like a charm.

It's very useful to block until "notification is received or 5 seconds passes". This way, the process dispatching those notifications can also be used to do periodic work. I can see how python threading Event objects could be used to solve the problem, as an alternative. An asyncio version could do something like asyncio.wait_for(gen.__anext__(), 5), but that feels wrong, you'd probably prefer to add another task. In either case there's an added challenge of closing it out for the exit condition.

I'm agreeing with other comments that it would be nice for .notifies() to either take a timeout (presumably yielding a non-Notify object on timeouts), or directly expose the inner loop to only get queued notifications.

[dvarrazzo]

Hello, Sorry for leaving this topic long unanswered. I worked on a few ways to interrupt the generator, but the sync and asymc approaches were different and now I can't remember what I was doing anymore... I'm totally ok to add a timeout if this solves all the use cases we would like to cover. The participants in this thread have concrete use cases, so I'm happy to hear from you if this would cover your problems or if a different interface would be preferable. MRs are welcome too! :)

[AlanCoding]

I have not tested it (and much could be wrong), but I envision something like this https://github.com/psycopg/psycopg/compare/master...AlanCoding:yield_timeouts?expand=1, and this would be applied AsyncConnection.notifies method as well.