[FIXED] A blocked Python async function invocation also block another async function

Issue

I use FastAPI to develope data layer APIs accessing SQL Server. No mater using pytds or pyodbc, if there is a database transaction caused any request hangs, all the other requests would be blocked. (even without database operation)

Reproduce:

1. Intentaionally do a serializable SQL Server session, begin a transaction and do not rollback or commit

    INSERT INTO [dbo].[KVStore] VALUES ('1', '1', 0)
    begin tran
    SET TRANSACTION ISOLATION LEVEL Serializable
    SELECT * FROM [dbo].[KVStore]

2. Send a request to the API with async handler function like this:

    def kv_delete_by_key_2_sql():
            conn = pytds.connect(dsn='192.168.0.1', database=cfg.kvStore_db, user=cfg.kvStore_uid,
                                 password=cfg.kvStore_upwd, port=1435, autocommit=True)
            engine = conn.cursor()
            try:
                sql = "delete KVStore; commit"

                with concurrent.futures.ThreadPoolExecutor() as executor:
                    future = executor.submit(engine.execute, sql)
                    rs = future.result()
                    j = {
                        'success': True,
                        'rowcount': rs.rowcount
                    }
                    return jsonable_encoder(j)
            except Exception as exn:
                j = {
                    'success': False,
                    'reason': exn_handle(exn)
                }
                return jsonable_encoder(j)

    @app.post("/kvStore/delete")
    async def kv_delete(request: Request, type_: Optional[str] = Query(None, max_length=50)):
        request_data = await request.json()

        return kv_delete_by_key_2_sql()

3. And send a request to the API of the same app with async handler function like this:

    async def hangit0(request: Request, t: int = Query(0)):
        print(t, datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3])
        await asyncio.sleep(t)
        print(t, datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3])
        j = {
            'success': True
        }
        return jsonable_encoder(j)
    
    @app.get("/kvStore/hangit/")
    async def hangit(request: Request, t: int = Query(0)):
        return await hangit0(request, t)

I expected step.2 would hang and step.3 should directly return after 2 seconds. However step.3 never return if the transaction doesn't commit or rollback...

How do I make these handler functions work concurrently?

Solution

The reason is that rs = future.result() is actually a blocking call - see python docs. Unfortunately, executor.submit() doesn't return an awaitable object (concurrent.futures.Future is different from asyncio.Future.

You can use asyncio.wrap_future which takes concurrent.futures.Future and returns asyncio.Future (see python docs). The new Future object is awaitable thus you can convert your blocking function into an async function.

An Example:

import asyncio
import concurrent.futures

async def my_async():
    with concurrent.futures.ThreadPoolExecutor() as executor:
        future = executor.submit(lambda x: x + 1, 1)
        return await asyncio.wrap_future(future)

print(asyncio.run(my_async()))

In your code, simply change the rs = future.result() to rs = await asyncio.wrap_future(future) and make the whole function async. That should do the magic, good luck! :)

Answered By - sukovanej