Celery Worker Database Connection Pooling

I am using Celery standalone (not within Django). I am planning to have one worker task type running on multiple physical machines. The task does the following

1. Accept an XML document.
2. Transform it.
3. Make multiple database reads and writes.

I'm using PostgreSQL, but this would apply equally to other store types that use connections. In the past, I've used a database connection pool to avoid creating a new database connection on every request or avoid keeping the connection open too long. However, since each Celery worker runs in a separate process, I'm not sure how they would actually be able to share the pool. Am I missing something? I know that Celery allows you to persist a result returned from a Celery worker, but that is not what I'm trying to do here. Each task can do several different updates or inserts depending on the data processed.

What is the right way to access a database from within a Celery worker?

Is it possible to share a pool across multiple workers/tasks or is there some other way to do this?

Solution 1:

I like tigeronk2's idea of one connection per worker. As he says, Celery maintains its own pool of workers so there really isn't a need for a separate database connection pool. The Celery Signal docs explain how to do custom initialization when a worker is created so I added the following code to my tasks.py and it seems to work exactly like you would expect. I was even able to close the connections when the workers are shutdown:

from celery.signals import worker_process_init, worker_process_shutdown

db_conn = None

@worker_process_init.connect
def init_worker(**kwargs):
    global db_conn
    print('Initializing database connection for worker.')
    db_conn = db.connect(DB_CONNECT_STRING)


@worker_process_shutdown.connect
def shutdown_worker(**kwargs):
    global db_conn
    if db_conn:
        print('Closing database connectionn for worker.')
        db_conn.close()

Solution 2:

Have one DB connection per worker process. Since celery itself maintains a pool of worker processes, your db connections will always be equal to the number of celery workers. Flip side, sort of, it will tie up db connection pooling to celery worker process management. But that should be fine given that GIL allows only one thread at a time in a process.

Solution 3:

You can override the default behavior to have threaded workers instead of a worker per process in your celery config:

CELERYD_POOL = "celery.concurrency.threads.TaskPool"

Then you can store the shared pool instance on your task instance and reference it from each threaded task invocation.

Solution 4:

Perhaps, celery.concurrency.gevent could provide the pool sharing and not aggravate the GIL. However, it's support is still "experimental".

And a psycopg2.pool.SimpleConnectionPool to share amongst greenlets (coroutines) which will all run in a single process/thread.

Tiny bit of other stack discussion on the topic.