Why celery task fired from django pre_save signal never sees the object

Question

This is a followup queston from here.

Django 1.3.1, celery 2.2.7, python2.6.

I have the following in the fruits/models.py:

Consider the model:

class Fruit(models.Model):
    name = models.CharField(max_length=50)

    def __unicode__(self):
        return self.name

And the following in the fruits/tasks.py:

from django.dispatch import receiver
from django.db.models import signals

from celery.task import periodic_task, task
import fruits.models as m
import time

@task()
def check_fruit(id):
    time.sleep(2)
    try:
        fruit = m.Fruit.objects.get(pk=id)
        print "Fruit %s is found!" % fruit.name
    except m.Fruit.DoesNotExist:
        print "no such fruit"

@receiver(signals.pre_save, sender=m.Fruit, dispatch_uid="on_fruit_save")
def on_custom_feed_save(sender, instance, **kwargs):
    check_fruit.apply_async(args=[instance.id])

I launch celery daemon, then open django shell and type:

import fruits.tasks;
import fruits.models as m;
m.Fruit(name="plum").save()

Question: I would expect that the task would find the fruit, but it never does. Why?

(I'm launching task from pre-save signal on purpose to simulate a problem that happens on a large system).

Answer 1

Old question actually very simple, problem isn't race condition or bug etc.

The example creates new object from scratch and the object has not yet stored in database during pre_save signal. So 'instance.id' isn't set and it is None for newly created objects.

check_fruit.delay(None)

called by pre_save signal and it produces

fruit = m.Fruit.objects.get(pk=None)

There isn't any row with null primary key in DB.

Celery task is checking this query for newly created objects actually. the query throws exception always as expected.

Use post_save signal and check created parameter of the signal for new objects.

after using post_save, there is possibility for race condition related with transactions. Most of time you will not care, task retries works mostly. But if you care race condition, look at new django feature, on_commit hook

Answer 2

I ran into a very similar issue myself this week, except that I was using Celery without Django. I found that the sender parameter only worked when the instance of the task was passed to it, rather than simply a reference to the sender class itself.

I looked into the problem a bit, and found that the sender parameter is used by celery to compare id's of a specific task, and a signal's registered sender (which is set to filter for a specific sender.)

In the celery/utils/dispatch/signals.py module, the following performs this evaluation:

def _make_id(target):  # pragma: no cover
    if hasattr(target, 'im_func'):
        return (id(target.im_self), id(target.im_func))
    return id(target) 

First an id of some target object is fetched. When a sender is specified in the decorator, it is saved in tuple similar to this:

lookup_key = (_make_id(receiver), _make_id(sender))

Later, when a task fires, the _live_receivers method is called, which essentially performs an evaluation to see whether the target sender specified in the lookup_key matches the id of the current sender (the firing task):

def _live_receivers(self, senderkey):
        """Filter sequence of receivers to get resolved, live receivers.

        This checks for weak references and resolves them, then returning only
        live receivers.

        """
        none_senderkey = _make_id(None)
        receivers = []

        for (receiverkey, r_senderkey), receiver in self.receivers:
            if r_senderkey == none_senderkey or r_senderkey == senderkey:
                if isinstance(receiver, WEAKREF_TYPES):
                    # Dereference the weak reference.
                    receiver = receiver()
                    if receiver is not None:
                        receivers.append(receiver)
                else:
                    receivers.append(receiver)
        return receivers

Now the problem that I was experiencing was that even though I specified the task which I wanted to receive a signal on, when it was fired, the sender keys never matched.

The only way I could get this to work correctly was to register the signal from within the __init__ method of an abstract task class I had built for the specific task itself. By doing this, I was able to pass Celery the exact instance (self), that it had registered for the task.

I have no idea if this problem was with my logic or understanding of how signals work, or if it was a bug in Celery, but I do know that passing the task instance fixed the problem and everything worked fine thereafter.

Some things to note:

1. I wasn't using Django, and therefore was not using the Django-Celery extension
2. I am not confident that the problem was with Celery (it is more likely that I made a mistake in logic or misunderstood something, somewhere)
3. I don't know if this case even applies to you, because I am not sure how Django-Celery changes the way in which the Celery backend works.

Nonetheless, I hope that this is helpful.

Good Luck!

Answer 3

Old question and problem has probably been solved by now, but for future visitors...

Sounds like the celery worker has a long-running transaction open, and if it's never committed, it never sees the new object. Try adding this before you run a DB query in the task:

from django.db import transaction
transaction.commit()

Note that Django 1.6 is bringing some changes to transaction management, but it's not out yet, as of writing this.