Gino: Working with Multiple Event Loops

Created on 5 Sep 2019 · 7Comments · Source: python-gino/gino

GINO version: 0.7.0
Python version: 3.5.4

We have a service with two event loops on different threads (one for REST API, the other for background processing). Is there a way to use Gino with ORM classes that are accessed by both the event loops?

Thanks

question

Source

nilir

Most helpful comment

Tony's right. With vanilla GINO, none of Engine, Pool or underlying asyncpg.pool supports multiple event loops - they are coupled with a single loop instance so you have to create a separate engine for each event loop, and in order to share the same MetaData (gino.Gino) instance, you'll need to specify the bind explicitly for each DB query call.

However, it is also possible to make a custom event-loop-aware Pool that delegates and distributes DB query calls to underlying Pools based on the current event loop:

import asyncio
import random
import threading

import gino
from gino.dialects import base
from gino.dialects.asyncpg import Pool as AsyncpgPool

db = gino.Gino()


class Pool(base.Pool):
    def __init__(self, url, loop, **kwargs):
        self._url = url
        self._kwargs = kwargs
        self._pools = {}

    def __await__(self):
        async def _get_pool():
            await self._get_pool()
            return self

        return _get_pool().__await__()

    @property
    def raw_pool(self):
        return self._pools[asyncio.get_event_loop()]

    async def acquire(self, *, timeout=None):
        pool = await self._get_pool()
        return await pool.acquire(timeout=timeout)

    async def release(self, conn):
        pool = await self._get_pool()
        await pool.release(conn)

    async def close(self):
        pools = list(self._pools.values())
        self._pools.clear()
        for pool in pools:
            await pool.close()

    async def _get_pool(self):
        loop = asyncio.get_event_loop()
        rv = self._pools.get(loop)
        if rv is None:
            rv = self._pools[loop] = asyncio.Future()
            rv.set_result(await AsyncpgPool(self._url, loop, **self._kwargs))
        return await rv


class Sub(threading.Thread):
    def run(self):
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        loop.run_until_complete(run())


async def run_coro():
    count = 0
    while True:
        async with db.transaction():
            print(threading.current_thread(), count, '\tstart')
            start = await db.scalar('SELECT now()')
            await asyncio.sleep(random.random())
            print(threading.current_thread(), count, '\tend',
                  await db.scalar('SELECT now()') - start)
            count += 1


async def run():
    fs = [asyncio.ensure_future(run_coro()) for _ in range(3)]
    await asyncio.wait(fs)


async def main():
    await db.set_bind('postgresql://localhost/gino', pool_class=Pool)
    for _ in range(3):
        Sub(daemon=True).start()
    await run()


if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

fantix on 8 Sep 2019

👍2

All 7 comments

I think it's possible, but you may need to initialise two engines and pass them to each DB query call.

wwwjfy on 5 Sep 2019

However, it is also possible to make a custom event-loop-aware Pool that delegates and distributes DB query calls to underlying Pools based on the current event loop:

import asyncio
import random
import threading

import gino
from gino.dialects import base
from gino.dialects.asyncpg import Pool as AsyncpgPool

db = gino.Gino()


class Pool(base.Pool):
    def __init__(self, url, loop, **kwargs):
        self._url = url
        self._kwargs = kwargs
        self._pools = {}

    def __await__(self):
        async def _get_pool():
            await self._get_pool()
            return self

        return _get_pool().__await__()

    @property
    def raw_pool(self):
        return self._pools[asyncio.get_event_loop()]

    async def acquire(self, *, timeout=None):
        pool = await self._get_pool()
        return await pool.acquire(timeout=timeout)

    async def release(self, conn):
        pool = await self._get_pool()
        await pool.release(conn)

    async def close(self):
        pools = list(self._pools.values())
        self._pools.clear()
        for pool in pools:
            await pool.close()

    async def _get_pool(self):
        loop = asyncio.get_event_loop()
        rv = self._pools.get(loop)
        if rv is None:
            rv = self._pools[loop] = asyncio.Future()
            rv.set_result(await AsyncpgPool(self._url, loop, **self._kwargs))
        return await rv


class Sub(threading.Thread):
    def run(self):
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        loop.run_until_complete(run())


async def run_coro():
    count = 0
    while True:
        async with db.transaction():
            print(threading.current_thread(), count, '\tstart')
            start = await db.scalar('SELECT now()')
            await asyncio.sleep(random.random())
            print(threading.current_thread(), count, '\tend',
                  await db.scalar('SELECT now()') - start)
            count += 1


async def run():
    fs = [asyncio.ensure_future(run_coro()) for _ in range(3)]
    await asyncio.wait(fs)


async def main():
    await db.set_bind('postgresql://localhost/gino', pool_class=Pool)
    for _ in range(3):
        Sub(daemon=True).start()
    await run()


if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

fantix on 8 Sep 2019

👍2

Closing due to inactivity. Feel free to reopen.

fantix on 6 Feb 2020

@fantix Hi, I just stumbled upon this thread yesterday and I was having the same issue. Your solution with a custom pool works nicely, but it doesn't clean up properly after itself (the event loop closes before the pool for the current loop can release resources properly).

I'm currently trying to solve this issue. Would you be interested in a pull request if I happen to find a solution?

Traceback (most recent call last):
  File ".\asdf.py", line 97, in <module>
    asyncio.get_event_loop().run_until_complete(main())
  File "C:\Python38\lib\asyncio\base_events.py", line 612, in run_until_complete
    return future.result()
  File ".\asdf.py", line 90, in main
    pass
  File "E:\Workspace\liQuorice\.venv\lib\site-packages\gino\api.py", line 207, in __aexit__
    await self._args[0].pop_bind().close()
  File "E:\Workspace\liQuorice\.venv\lib\site-packages\gino\engine.py", line 719, in close
    await self._pool.close()
  File ".\asdf.py", line 43, in close
    await pool.result().close()
  File "E:\Workspace\liQuorice\.venv\lib\site-packages\gino\dialects\asyncpg.py", line 230, in close
    await self._pool.close()
  File "E:\Workspace\liQuorice\.venv\lib\site-packages\asyncpg\pool.py", line 685, in close
    await asyncio.gather(*release_coros)
  File "E:\Workspace\liQuorice\.venv\lib\site-packages\asyncpg\pool.py", line 229, in wait_until_released
    await self._in_use
RuntimeError: Task <Task pending name='Task-130' coro=<PoolConnectionHolder.wait_until_released() running at E:\Workspace\liQuorice\.venv\lib\site-packages\asyncpg\pool.py:229> cb=[gather.<locals>._done_callback() at C:\Python38\lib\asyncio\tasks.py:751]> got Future <Future pending> attached to 
a different loop

ghost-of-turek on 8 Feb 2020

I'm currently trying to solve this issue. Would you be interested in a pull request if I happen to find a solution?

Oh yes, absolutely. I appreciate that!

fantix on 8 Feb 2020

Ok the, I've spent some time on it and I just don't see a reasonable way for the Pool to close itself at the right time. So for now I'm stuck with this monster (I modified your example a little):

import asyncio
import random
import threading

import gino
from gino.dialects.base import Pool as GinoPool
from gino.dialects.asyncpg import Pool as AsyncpgPool


stop_ev = threading.Event()
db = gino.Gino()


class MultiLoopPool(GinoPool):
    def __init__(self, url, loop, **kwargs):
        self._url = url
        self._kwargs = kwargs
        self._pools = {}

    def __await__(self):
        async def _get_pool():
            await self._get_pool()
            return self

        return _get_pool().__await__()

    @property
    def raw_pool(self):
        return self._pools[asyncio.get_event_loop()]

    async def acquire(self, *, timeout=None):
        pool = await self._get_pool()
        return await pool.acquire(timeout=timeout)

    async def release(self, conn):
        pool = await self._get_pool()
        await pool.release(conn)

    async def close(self):
        pools = list(self._pools.values())
        self._pools.clear()
        for pool in pools:
            await pool.result().close()

    async def _get_pool(self):
        loop = asyncio.get_event_loop()
        rv = self._pools.get(loop)
        if rv is None:
            rv = self._pools[loop] = asyncio.Future()
            rv.set_result(await AsyncpgPool(self._url, loop, **self._kwargs))
        return await rv

    async def close_for_thread(self):
        await (await self._get_pool()).close()


class Sub(threading.Thread):
    def run(self):
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        loop.run_until_complete(run())
        loop.run_until_complete(db.bind._pool.close_for_thread())


async def run_coro():
    count = 0
    while not stop_ev.is_set():
        async with db.transaction():
            print(threading.current_thread(), count, '\tstart')
            start = await db.scalar('SELECT now()')
            await asyncio.sleep(random.random())
            print(threading.current_thread(), count, '\tend',
                  await db.scalar('SELECT now()') - start)
            count += 1


async def run():
    fs = [asyncio.ensure_future(run_coro()) for _ in range(3)]
    await asyncio.wait(fs)


async def main():
    async with db.with_bind(
        'postgresql://liquorice:liquorice@localhost:5432/liquorice',
        pool_class=MultiLoopPool,
    ):

        handles = [Sub(daemon=True) for _ in range(3)]

        for handle in handles:
            handle.start()
        try:
            while True:
                ...
        except KeyboardInterrupt:
            stop_ev.set()
        finally:
            for handle in handles:
                handle.join()


asyncio.run(main())

Sadly with this approach, every time you want to signal that you are done with the pool you need to do it manually via something like loop.run_until_complete(db.bind._pool.close_for_thread()).

I was thinking about exposing this as a method on GinoEngine, so at least it would be db.bind.close_for_thread() without accessing Gino's private API, but it's not a beautiful approach.

ghost-of-turek on 9 Feb 2020

Alright, I gave it some more thought, and I don't think I will create a PR for this. The code to handle this issue cannot be reasonably written and I think the solution I'm using is not really deserving of being a part of the library.

I refactored out some helper methods on my custom Gino object and it's pretty convenient now.

In case anyone ever needs it or deems is PR-worthy, here's what I've came up which works nicely and completely satisfies my needs. Do whatever you will with it.

import asyncio
import random
import threading

import gino
from gino import Gino as BaseGino
from gino.dialects.base import Pool as GinoPool
from gino.dialects.asyncpg import Pool as AsyncpgPool


class Gino(BaseGino):
    def set_bind(self, bind, loop=None, **kwargs):
        kwargs['pool_class'] = MultiLoopPool
        return super().set_bind(bind, loop=loop, **kwargs)

    def with_bind(self, bind, loop=None, **kwargs):
        kwargs['pool_class'] = MultiLoopPool
        return super().with_bind(bind, loop=loop, **kwargs)

    async def close_for_current_loop(self):
        await self.bind._pool.close_for_current_loop()


db = gino.Gino()


class MultiLoopPool(GinoPool):
    def __init__(self, url, loop, **kwargs):
        self._url = url
        self._kwargs = kwargs
        self._pools = {}

    def __await__(self):
        async def _get_pool():
            await self._get_pool()
            return self

        return _get_pool().__await__()

    @property
    def raw_pool(self):
        return self._pools[asyncio.get_event_loop()]

    async def acquire(self, *, timeout=None):
        pool = await self._get_pool()
        return await pool.acquire(timeout=timeout)

    async def release(self, conn):
        pool = await self._get_pool()
        await pool.release(conn)

    async def close(self):
        pools = list(self._pools.values())
        self._pools.clear()
        for pool in pools:
            await pool.close()

    async def _get_pool(self):
        loop = asyncio.get_event_loop()
        if loop not in self._pools:
            self._pools[loop] = await AsyncpgPool(
                self._url, loop, **self._kwargs,
            )
        return self._pools[loop]

    async def close_for_current_loop(self):
        await (await self._get_pool()).close()


class Sub(threading.Thread):
    def run(self):
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        loop.run_until_complete(run())

        # this is the key ingredient to correctly closing the pool
        loop.run_until_complete(await db.close_for_current_loop())


stop_ev = threading.Event()


async def run_coro():
    count = 0
    while not stop_ev.is_set():
        async with db.transaction():
            print(threading.current_thread(), count, '\tstart')
            start = await db.scalar('SELECT now()')
            await asyncio.sleep(random.random())
            print(threading.current_thread(), count, '\tend',
                  await db.scalar('SELECT now()') - start)
            count += 1


async def run():
    fs = [asyncio.ensure_future(run_coro()) for _ in range(3)]
    await asyncio.wait(fs)


async def main():
    async with db.with_bind(
        'postgresql://liquorice:liquorice@localhost:5432/liquorice',
    ):

        handles = [Sub(daemon=True) for _ in range(3)]

        for handle in handles:
            handle.start()
        try:
            while True:
                ...
        except KeyboardInterrupt:
            stop_ev.set()
        finally:
            for handle in handles:
                handle.join()


asyncio.run(main())

ghost-of-turek on 11 Feb 2020

👍1

Was this page helpful?

0 / 5 - 0 ratings