Can connection pools be used for lots of long lived connections?

[Directory]

This is more of a question then an issue. For my case i have a websocket server using the python websockets package. Since its async i decided to use this library and my implementation of it works. At least on the surface level. i did some stress tests and immediately found out that concurrency is an issue. it raises asyncpg.exceptions._base.InterfaceError: cannot perform operation: another operation is in progress. not the libs fault is obvious whats going on. i am creating a connection object at the global scope of the script and using that same object throughout the entire program for everything including for all different connections. These connections are long lived and ideally i would like to be able to support hundreds to thousands of websocket connections (users). A naive approach is opening a new asyncpg connection for every websocket connection but i doubt its a smart idea to open thousands of database connections if working with thousands of websocket connections.

In the documentation i found connection pooling however a concern of mine is in its example its used for short lived connections in the context of an http request. not a long lived socket. my idea was to have a pool connection object at the global scope and only acquire the pool for every database operation that happens during the life of that websocket connection. which under peak loads i would say is about once per second (for each connection). My concern with this however is the performance. Does it take time to acquire the pool or is it instant? and what happens under high loads with lots of concurrent operations where the pool is trying to be acquired but some other operation still hasn't finished its with block? can multiple concurrent pool acquisitions happen? and about how much?

I'm going to attempt this implementation and test and respond with my findings. But if someone else can give an insight if this is a good idea and if it will scale well it would be greatly appreciated.

[elprans]

only acquire the pool for every database operation that happens during the life of that websocket connection.

This is appropriate.

which under peak loads i would say is about once per second

1 query per second is not that intensive and should be plenty of time to acquire/query/release multiple times over. I did a small benchmark on my laptop for a Stack Overflow answer a while ago, where a thousand acquire/query/release iterations completed in under 0.3s.

Does it take time to acquire the pool or is it instant?

If there's no contention, acquiring a connection is very quick.

and what happens under high loads with lots of concurrent operations where the pool is trying to be acquired but some other operation still hasn't finished its with block?

If the pool is contended, tasks will wait in the fifo queue until a connection becomes available. There's no hard limit on the number of waiters.

can multiple concurrent pool acquisitions happen? and about how much?

I'm not sure I understand the question completely. The number of concurrent connections in the pool are specified when you call create_pool(max_size=<max-connections>), which is 10 by default, but you should adjust it to balance the database server resource utilization with connection acquisition latency.

[Directory]

ok so i made 1 single pool object globally and acquired it for every fetch and execute and it made 0 difference no matter what i still get asyncpg.exceptions._base.InterfaceError: cannot perform operation: another operation is in progress at any point when 2 operations are ran right next to eachother. im also timing the operation of just a simple INSERT or fetch (including pool acquisition) and it takes somewhere up to 0.001 which is kinda slow. about half of that time is just spent on the pool acquisition and codec type setting and the rest on the actual db operation. i dont know what im supposed to do about concurrency at this point. i thought that using a global pool object and acquriing it was good enough but the exception keeps happening when its concurrent.

my last idea is to give each websocket user connection its own pool object rather then sharing 1 for each ws connection. however i am worried about this since that means up to thousands of pool objects to accommodate every socket. does a pool open a database connection? would this cause issues? not only that but its still a bandaid fix since 1 socket could potentially also have to handle multiple pool acquisitions in parallel and once again i would run into the same exception

[elprans]

and it made 0 difference

Well, that's unexpected. This should not happen as long as you're not sharing connection objects across concurrent tasks. Is the source available somewhere? Are you able to provide a minimal repro?

[Directory]

as long as you're not sharing connection objects across concurrent tasks

hmmm that might be it. im still unfamiliar with async in python so im not entirly sure what is considered a "concurrent task" however here is an as stripped down as possible version of my websocket server. take note of the pool and database functions being in the global scope separate from the connection_handler.

import websockets
import asyncio
import asyncpg
import json

global pool

async def init():
    global pool
    pool = await asyncpg.create_pool(host='localhost', port=5432, user='user', password='***', database='testdb')
    # when server starts it does some initial db stuff
    await junk_db_operation()

    await websockets.serve(connection_handler, '0.0.0.0', 5000)

async def junk_db_operation():
    async with pool.acquire() as conn:
        # other type codecs may also be in set here
        await conn.set_type_codec('json', encoder=json.dumps, decoder=json.loads, schema='pg_catalog')
        await conn.execute('INSERT INTO test VALUES ($1, $2)', 'lorem', 'ipsum')

async def connection_handler(connection: websockets.server.WebSocketServerProtocol, path):
    # random db operation on websocket establishment (to test in the benchmark client)
    await junk_db_operation()


    async for frame in connection:  # waits for data to come in
        print(frame)
        # just an example. per each frame a db operation may occur
        await junk_db_operation()

loop = asyncio.get_event_loop()
loop.run_until_complete(init())
loop.run_forever()

and the client i use to benchmark. the gather does the connections in parrallel so the exception raises on line 24 and tracebacks to 20 (on the server) where all 10 sockets are doing that same insert.

import asyncio
import websockets

conns = []

async def make_conn():
    ws = await websockets.connect('ws://localhost:5000')
    conns.append(ws)

async def bench():
    await asyncio.gather(*[make_conn() for _ in range(10)])
    await asyncio.sleep(30)

asyncio.get_event_loop().run_until_complete(bench())

as you can see, pool object in global scope. and wherever needed in the program a db operation is ran where a pool acquisition happens. what am i doing wrong here? the conn object in the pool acquisition with doesnt leave its test function just like in my real code.

[elprans]

Well, the above code looks correct to me and I cannot reproduce the cannot perform operation with it. Also, you don't need to set_type_codec after each acquire, you can do it in the init callback passed to create_pool instead, which would only be called when an actual new connection is opened (as opposed to reusing an existing connection).

[Directory]

oh wow. that was actually it. haha welp. yea i was running a set_caster function inside each acquire to add the codecs to the conn object. i wasnt aware of the init callback but im now doing that there and no more exception issues. that was the issue. it cant both set codecs and execute queries at the same time. however it doesnt raise on the set_type_codec line interestingly enough.