InterfaceError: connection already closed (using django + celery + Scrapy)

Question

I am getting this when using a Scrapy parsing function (that can take till 10 minutes sometimes) inside a Celery task.

I use: - Django==1.6.5 - django-celery==3.1.16 - celery==3.1.16 - psycopg2==2.5.5 (I used also psycopg2==2.5.4)

[2015-07-19 11:27:49,488: CRITICAL/MainProcess] Task myapp.parse_items[63fc40eb-c0d6-46f4-a64e-acce8301d29a] INTERNAL ERROR: InterfaceError('connection already closed',)
Traceback (most recent call last):
  File "/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/celery/app/trace.py", line 284, in trace_task
    uuid, retval, SUCCESS, request=task_request,
  File "/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/celery/backends/base.py", line 248, in store_result
    request=request, **kwargs)
  File "/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/djcelery/backends/database.py", line 29, in _store_result
    traceback=traceback, children=self.current_task_children(request),
  File "/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/djcelery/managers.py", line 42, in _inner
    return fun(*args, **kwargs)
  File "/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/djcelery/managers.py", line 181, in store_result
    'meta': {'children': children}})
  File "/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/djcelery/managers.py", line 87, in update_or_create
    return get_queryset(self).update_or_create(**kwargs)
  File "/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/djcelery/managers.py", line 70, in update_or_create
    obj, created = self.get_or_create(**kwargs)
  File "/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/models/query.py", line 376, in get_or_create
    return self.get(**lookup), False
  File "/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/models/query.py", line 304, in get
    num = len(clone)
  File "/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/models/query.py", line 77, in __len__
    self._fetch_all()
  File "/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/models/query.py", line 857, in _fetch_all
    self._result_cache = list(self.iterator())
  File "/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/models/query.py", line 220, in iterator
    for row in compiler.results_iter():
  File "/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/models/sql/compiler.py", line 713, in results_iter
    for rows in self.execute_sql(MULTI):
  File "/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/models/sql/compiler.py", line 785, in execute_sql
    cursor = self.connection.cursor()
  File "/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/backends/__init__.py", line 160, in cursor
    cursor = self.make_debug_cursor(self._cursor())
  File "/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/backends/__init__.py", line 134, in _cursor
    return self.create_cursor()
  File "/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/utils.py", line 99, in __exit__
    six.reraise(dj_exc_type, dj_exc_value, traceback)
  File "/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/backends/__init__.py", line 134, in _cursor
    return self.create_cursor()
  File "/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/backends/postgresql_psycopg2/base.py", line 137, in create_cursor
    cursor = self.connection.cursor()
InterfaceError: connection already closed

Can you show the problematic code?

omri_saadon
– omri_saadon

2015-07-19 18:39:10 +00:00
Commented Jul 19, 2015 at 18:39 — omri_saadon
– omri_saadon, Commented Jul 19, 2015 at 18:39

mannysz · Accepted Answer · 2016-04-15 19:24:22Z

25

Unfortunately this is a problem with django + psycopg2 + celery combo. It's an old and unsolved problem.

Take a look on this thread to understand: https://github.com/celery/django-celery/issues/121

Basically, when celery starts a worker, it forks a database connection from django.db framework. If this connection drops for some reason, it doesn't create a new one. Celery has nothing to do with this problem once there is no way to detect when the database connection is dropped using django.db libraries. Django doesn't notifies when it happens, because it just start a connection and it receives a wsgi call (no connection pool). I had the same problem on a huge production environment with a lot of machine workers, and sometimes, these machines lost connectivity with postgres server.

I solved it putting each celery master process under a linux supervisord handler and a watcher and implemented a decorator that handles the psycopg2.InterfaceError, and when it happens this function dispatches a syscall to force supervisor restart gracefully with SIGINT the celery process.

Edit:

Found a better solution. I implemented a celery task baseclass like this:

from django.db import connection
import celery

class FaultTolerantTask(celery.Task):
    """ Implements after return hook to close the invalid connection.
    This way, django is forced to serve a new connection for the next
    task.
    """
    abstract = True

    def after_return(self, *args, **kwargs):
        connection.close()

@celery.task(base=FaultTolerantTask)
def my_task():
    # my database dependent code here

I believe it will fix your problem too.

edited Apr 15, 2016 at 19:24

answered Apr 12, 2016 at 17:52

mannysz

1,01910 silver badges19 bronze badges

Sign up to request clarification or add additional context in comments.

4 Comments

Mo J. Mughrabi Over a year ago

hi emanuelcds, can you share example? facing same issue and would help to see sample code. Thanks

mannysz Over a year ago

I got a better solution yesterday. I will implement and check if it works properly. Once it worked, I will edit this answer and let you know. But basicaly, you can use a base class on a celery task by using the 'base' argument o @app.task. I will implement something to restart the db connection if it fails. I will keep you posted here.

Noky Over a year ago

I'm seeing the exact same exception from a daemon python script which is using Django ORM to access a db (Django==1.9.5, psycopg2==2.6.1). If PostgreSQL gets restarted while the daemon is running (or a db connection becomes invalid for some other reason), the db connection never gets recreated. It is possible to catch every possible db exception and forcibly reset the connection, but this is a big hack: stackoverflow.com/questions/4447497. I'd love to see a general solution to address this.

mannysz Over a year ago

There is no general solution. In your case, just use a try except block catching a django.db.utils.Error, closing the django.db.connection like the way I did on my edited code, and retry the entire process again.

Community · Accepted Answer · 2017-05-23 12:00:25Z

14

Guys and emanuelcds,

I had the same problem, now I have updated my code and created a new loader for celery:

from djcelery.loaders import DjangoLoader
from django import db

class CustomDjangoLoader(DjangoLoader):
    def on_task_init(self, task_id, task):
        """Called before every task."""
        for conn in db.connections.all():
            conn.close_if_unusable_or_obsolete()
        super(CustomDjangoLoader, self).on_task_init(task_id, task)

This of course if you are using djcelery, it will also require something like this in the settings:

CELERY_LOADER = 'myproject.loaders.CustomDjangoLoader'
os.environ['CELERY_LOADER'] = CELERY_LOADER

I still have to test it, I will update.

edited May 23, 2017 at 12:00

CommunityBot

11 silver badge

answered Jun 17, 2016 at 22:19

PaoloC

3,6612 gold badges17 silver badges18 bronze badges

Comments

user14140381 · Accepted Answer · 2022-03-30 20:06:31Z

5

If you are running into this when running tests, then you can either change the test to TransactionTestCase class instead of TestCase or add the mark pytest.mark.django_db(transaction=True). This kept my db connection alive from creation of the pytest-celery fixtures to the database calls.

Github issue - https://github.com/Koed00/django-q/issues/167

For context, I am using pytest-celery with celery_app and celery_worker as fixtures in my tests. I am also trying to hit the test db in the tasks referenced in these tests.

If someone would explain switching to transaction=True keeps it open, that would be great!

answered Mar 30, 2022 at 20:06

user14140381

2711 gold badge4 silver badges13 bronze badges

1 Comment

Community Over a year ago

As it’s currently written, your answer is unclear. Please edit to add additional details that will help others understand how this addresses the question asked. You can find more information on how to write good answers in the help center.

trybek_4 · Accepted Answer · 2024-10-31 10:15:18Z

0

In my case it was caused by not closed transaction.atomic.

def foo():
    trans = transaction.atomic()
    trans.__enter__()
    ## some code
    if error:
        raise Exception
    trans.__exit__()

When an exception occured the transaction was kept open so the connection was. Even if I manually called connection.close() Connection already closed was occuring on the next call of the task.

I've fixed it with:

with transaction.atomic():
    ....

Check if you don't leave an open transactions ;)

answered Oct 31, 2024 at 10:15

trybek_4

971 silver badge3 bronze badges

Collectives™ on Stack Overflow

InterfaceError: connection already closed (using django + celery + Scrapy)

4 Answers 4

4 Comments

Comments

1 Comment

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

4 Answers 4

4 Comments

Comments

1 Comment

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related