asyncio and blocking - an update

[Frank Millman]

"Chris Angelico"  wrote in message 
news:CAPTjJmrVCkKAEevc9TW8FYYTnZgRUMPHectz+bD=DQRphXYTpw at mail.gmail.com...
>
> Something worth checking would be real-world database performance metrics

[snip lots of valid questions]

My approach is guided by something I read a long time ago, and I don't know 
how true it is, but it feels plausible. This is a rough paraphrase.

Modern databases are highly optimised to execute a query and return the 
result as quickly as possible. A properly written database adaptor will work 
in conjunction with the database to optimise the retrieval of the result. 
Therefore the quickest way to get the result is to let the adaptor iterate 
over the cursor and let it figure out how best to achieve it.

Obviously you still have to tune your query to make make sure it is 
efficient, using indexes etc. But there is no point in trying to 
second-guess the database adaptor in figuring out the quickest way to get 
the result.

My theory rests on an assumption which may be faulty. I have assumed that, 
in order to execute a query using run_in_executor(), the way to get the 
result is to use cur.fetchall(). Maybe there are alternatives. However, 
based on that assumption, my theory contrasts the following two approaches -

1. In a separate thread, perform the following -

    cur.execute('SELECT ...')
    rows = cur.fetchall()
    return rows

    The awaiting function will perform the following -

    future = loop.run_in_executor('SELECT ...')
    await future
    rows = future.result()
    for row in rows:
        process row

    The SELECT will not block, because it is run in a separate thread. But 
it will return all the rows in a single list, and the calling function will 
block while it processes the rows, unless it takes the extra step of turning 
the list into an Asynchronous Iterator.

2. In a separate thread, perform the following -

    cur.execute('SELECT ...')
    for row in cur:
        build up block of 50 rows
        loop.call_soon_threadsafe(return_queue.put_nowait, block)

    The awaiting function will call the following -

        rows = AsyncCursor('SELECT ...')
        async for row in rows:
            process row

    AsyncCursor looks like this (abridged) -
        def __init__(self, sql, params):
            loop = asyncio.get_event_loop()
            self.return_queue = asyncio.Queue()
            request_queue.put((loop, sql, params, self.return_queue))
            self.rows = []
        async def __aiter__(self):
            return self
        async def __anext__(self):
            if self.rows:
                return self.rows.pop(0)
            self.rows = await self.return_queue.get()
            return self.rows.pop(0)

Hope this makes sense.

Frank