Re: LISTEN, select/poll question/clarification

On Tue, Dec 5, 2017 at 6:31 PM, Israel Brewster <israel@ravnalaska.net> wrote:
> I am using the PostgreSQL PubSub feature with psycopg2 and gevent in the
> following manner (dbconn is, of course, a psycopg2 connection object on
> which LISTEN has been called):
>
> while True:
>     if gevent.select.select([dbconn], [], []) != ([], [], []):
>         dbconn.poll()
>         gevent.spawn(process_result,dbconn.notifies)

>
>
> Which works fine. Now my understanding of how this all works is that when
> dbconn.poll() is called, it should pull in *all* pending NOTIFYs and append
> them to the dbconn.notifies object. So lets say I have NOTIFYs coming in at
> a rate of 1 per second. I would think that if I added a "gevent.sleep(5)" to
> the end of the above while loop, then each time through I should have 5
> notifies in the dbconn.notifies list, since it has been 5 seconds since I
> last checked. However, that doesn't appear to be the case - rather, no mater
> how long a sleep I put in, I still only get *one* item in the notifies list,
> making me think that I am missing data.

Uhm, from the top of my very rusty familiarity with gevent, I think
that gevent.select() is woken up as soon as the notify is received,
and that exactly because you have gevent.sleep(5) that is saying "I
don't have anything else to do for 5 seconds, so take a look if there
is any other fd that needs attention".

    while True:
        if gevent.select.select([dbconn], [], []) != ([], [], []):
            dbconn.poll()
            gevent.spawn(process_result,dbconn.notifies)
>
> Can someone explain why this is? Why am I not getting 5 at a time under that
> scenario?
>
> In case someone was wondering, the reasoning behind adding the sleep() lies
> in that process_result function. Due to the looping it contains, it can
> process, say, 5 data points in one call much more efficiently than it can
> process those same 5 data points in 5 calls of 1 data point each. So rather
> than run it every time a data point comes in, I would like to let the data
> "pile up" as it were for a short period of time before processing. I was
> thinking I could easily accomplish this by simply waiting some appropriate
> period of time between poll() ing the database for new NOTIFYs, but
> apparently that's not working. The select() is there because the data isn't
> coming in at a regular speed like in my example, but rather it could come in
> much faster, or not at all for a while, depending on the time of day.

At the most simple (and still very IIRC), you can just ignore select
and poll every 5 seconds, and you should receive the batches.

    while True:
        gevent.sleep(5)
        dbconn.poll()
        if dbconn.notifies:
            gevent.spawn(process_result,dbconn.notifies)

A bit more complex, you can do something at application level, such as
pushing your notifies in a queue as fast as they are received by
select(), and on the other side of the queue having a consumer stuck
on get(): as soon as it receives an object, the consumer can sleep for
5 seconds, then get_nowait() until the queue is empty to gather
everything received while it was napping, and call process_result()
with the whole set.


-- Daniele