CODEWITHSUNDEEP
pythonpostgresqlconnection-poolingcelery
oneself
oneself

Reputation: 40321

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?

Upvotes: 63

Views: 36565

Answers (6)

ThatAintWorking
ThatAintWorking

Reputation: 1370

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()

Upvotes: 45

Robin Loxley
Robin Loxley

Reputation: 824

Contribute back my findings by implementing and monitoring.

Welcome feedback.

Reference: use pooling http://www.prschmid.com/2013/04/using-sqlalchemy-with-celery-tasks.html

Each worker process (prefork mode specified by -c k) will establish one new connection to DB without pooling or reusing. So if using pooling, the pool is seen only at each worker process level. So pool size > 1 is not useful, but reusing connection is still fine for saving connection from open & close.

If using one connection per worker process, 1 DB connection is established per worker process (prefork mode celery -A app worker -c k) at initialization phase. It saves connection from open & close repeatedly.

No matter how many worker thread (eventlet), each worker thread (celery -A app worker -P eventlet) only establish one connection to DB without pooling or reusing. So for eventlet, all worker threads (eventlets) on one celery process (celery -A app worker ...) have 1 db connection at each moment.

According to celery docs

but you need to ensure your tasks do not perform blocking calls, as this will halt all other operations in the worker until the blocking call returns.

It is probably due to the way of MYSQL DB connection is blocking calls.

Upvotes: 0

kev
kev

Reputation: 9795

Perhaps you can use pgbouncer. For celery nothing should change and the connection pooling is done outside of the processes. I have the same issue.

('perhaps' because I am not sure if there could be any side effects)

Upvotes: 0

tigeronk2
tigeronk2

Reputation: 2650

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.

Upvotes: 3

p7k
p7k

Reputation: 161

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.

Upvotes: 1

Loren Abrams
Loren Abrams

Reputation: 1002

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.

Upvotes: 2

Related Questions