CODEWITHSUNDEEP
pythonmysqldjangocelerydjango-celery
user2302807
user2302807

Reputation: 335

python django race condition with celery

Working on a python django project, here is what I want:

  1. User access Page1 with object argument, function longFunction() of the object is triggered and passed to celery so the page can be returned immediately

  2. If user tries to access Page2 with same object argument, I want the page to hang until object function longFunction() triggered by Page1 is terminated.

So I tried by locking mysql db row with objects.select_for_update() but it doesn't work.

Here is a simplified version of my code:

def Page1(request, arg_id):
    obj = Vm.objects.select_for_update().get(id=arg_id)
    obj.longFunction.delay()
    return render_to_response(...)

def Page2(request, arg_id):
    vm = Vm.objects.select_for_update().get(id=arg_id)
    return render_to_response(...)

I want that Page2 hangs at the line vm = Vm.objects.select_for_update().get(id=arg_id) until longFunction() is completed. I'm new to celery and it looks like the mysql connection initiated on Page1 is lost when the Page1 returns, even if longFunction() is not finished.

Is there another way I can achieve that?

Thanks

Upvotes: 2

Views: 1685

Answers (2)

joshua
joshua

Reputation: 2519

The database lock from select_for_update is released when the transaction closes (in page 1). This lock doesn't get carried to the celery task. You can lock in the celery task but that won't solve your problem because page 2 might get loaded before the celery task obtains the lock.

Mikel's answer will work. You could also put a lock in the cache as described in the celery cookbook.

Upvotes: 0

Mikel Emaldi Manrique
Mikel Emaldi Manrique

Reputation: 633

Maybe this can be helpul for you: 

from celery.result import AsyncResult
from yourapp.celery import app

def Page1(request, arg_id):
    obj = Vm.objects.select_for_update().get(id=arg_id)
    celery_task_id = obj.longFunction.delay()
    return render_to_response(...)

def Page2(request, arg_id, celery_task_id):
    task = AsyncResult(app=app, id=celery_task_id)
    state = task.state
    while state != "SUCCESFUL":
        # wait or do whatever you want
    vm = Vm.objects.select_for_update().get(id=arg_id)
    return render_to_response(...)

More info at http://docs.celeryproject.org/en/latest/reference/celery.states.html

Upvotes: 1

Related Questions