Question about asyncio and blocking operations

[Ian Kelly]

On Wed, Jan 27, 2016 at 7:40 AM, Frank Millman <frank at chagford.com> wrote:
> "Ian Kelly"  wrote in message
> news:CALwzidk-RBkB-vi6CgcEeoFHQrsoTFvqX9MqzDD=rnY5bOCRUg at mail.gmail.com...
>
>> You probably want an asynchronous iterator here. If the cursor doesn't
>> provide that, then you can wrap it in one. In fact, this is basically
>> one of the examples in the PEP:
>> https://www.python.org/dev/peps/pep-0492/#example-1
>>
>
> Thanks, Ian. I had a look, and it does seem to fit the bill, but I could not
> get it to work, and I am running out of time.
>
> Specifically, I tried to get it working with the sqlite3 cursor. I am no
> expert, but after some googling I tried this -
>
> import sqlite3
> conn = sqlite3.connect('/sqlite_db')
> cur = conn.cursor()
>
> async def __aiter__(self):
>    return self
>
> async def __anext__(self):
>    loop = asyncio.get_event_loop()
>    return await loop.run_in_executor(None, self.__next__)
>
> import types
> cur.__aiter__ = types.MethodType( __aiter__, cur )
> cur.__anext__ = types.MethodType( __anext__, cur )
>
> It failed with this exception -
>
> AttributeError: 'sqlite3.Cursor' object has no attribute '__aiter__'
>
> I think this is what happens if a class uses 'slots' to define its
> attributes - it will not permit the creation of a new one.

This is why I suggested wrapping the cursor instead. Something like this:

class CursorWrapper:

    def __init__(self, cursor):
        self._cursor = cursor

    async def __aiter__(self):
        return self

    async def __anext__(self):
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(None, next, self._cursor)

> Anyway, moving on, I decided to change tack. Up to now I have been trying to
> isolate the function where I actually communicate with the database, and
> wrap that in a Future with 'run_in_executor'.
>
> In practice, the vast majority of my interactions with the database consist
> of very small CRUD commands, and will have minimal impact on response times
> even if they block. So I decided to focus on a couple of functions which are
> larger, and try to wrap the entire function in a Future with
> 'run_in_executor'.
>
> It seems to be working, but it looks a bit odd, so I will show what I am
> doing and ask for feedback.
>
> Assume a slow function -
>
> async def slow_function(arg1, arg2):
>    [do stuff]
>
> It now looks like this -
>
> async def slow_function(arg1, arg2):
>    loop = asyncio.get_event_loop()
>    await loop.run_in_executor(None, slow_function_1, arg1, arg2)
>
> def slow_function_1(self, arg1, arg2):
>    loop = asyncio.new_event_loop()
>    asyncio.set_event_loop(loop)
>    loop.run_until_complete(slow_function_2(arg1, arg2))
>
> async slow_function_2(arg1, arg2):
>    [do stuff]
>
> Does this look right?

I'm not sure I understand what you're trying to accomplish by running
a second event loop inside the executor thread. It will only be useful
for scheduling asynchronous operations, and if they're asynchronous
then why not schedule them on the original event loop?