Celery redis backend not always returning result
<p>I'm running a celery worker such that:</p>

<pre><code> -------------- celery@ v3.1.23 (Cipater)
---- **** ----- 
--- * ***  * -- Linux-4.4.0-31-generic-x86_64-with-debian-stretch-sid
-- * - **** --- 
- ** ---------- [config]
- ** ---------- .&gt; app:         __main__:0x7fe76cd42400
- ** ---------- .&gt; transport:   amqp://
- ** ---------- .&gt; results:     redis://
- *** --- * --- .&gt; concurrency: 4 (prefork)
-- ******* ---- 
--- ***** ----- [queues]
 -------------- .&gt; celery           exchange=celery(direct) key=celery
[tasks]
  . tasks.mytask
</code></pre>

<p><strong>tasks.py</strong>:</p>

<pre><code>@celery_app.task(bind=True, ignore_result=False)
def mytask(task):
    r = redis.StrictRedis()
    r.rpush('/task_finished', task.request.id)
    return {'result': 42}
</code></pre>

<p>When I try to run the following code, and run 2 task one after the other it works when getting the first result but fails to return the second one.</p>

<pre><code>import celery.result
import redis

r = redis.StrictRedis()
celery_app = Celery(name="my_long_task", backend="redis://")

while True:
    _, resp = r.blpop('/task_finished')
    task_id = resp.decode('utf-8')
    task = celery.result.AsyncResult(task_id, app=celery_app)
    print(task)
    print(task.result)
</code></pre>

<p>Will return:</p>

<p><em>First loop</em>:  </p>

<pre><code>[1] 990e2d04-5664-4d7c-8a5c-e9cb4ef45e24  
[2] {'result': 42}
</code></pre>

<p><em>Second loop</em> (fails to return the result):  </p>

<pre><code>[3] 8463cc46-0884-4bf7-b838-f0614f74b271  
[4] {}
</code></pre>

<p>However if I instantiate <code>celery_app = Celery(name="my_long_task", backend="redis://")</code> in the <code>while</code> loop it will work each time.<br>
What is wrong with not reinstantiating <code>celery_app</code> ? What am I missing ?</p>

<p><strong>Edit:</strong></p>

<p>Waiting a bit for the result (in case of latency) won't work too</p>

<pre><code>while True:
    _, resp = r.blpop('/task_finished')
    task_id = resp.decode('utf-8')
    for i in range(0, 20):
        # Won't work because I need to re instantiate celery_app
        task = celery.result.AsyncResult(task_id, app=celery_app)
        print(task.result)
        time.sleep(1)
</code></pre>

<p><strong>You have a race condition.</strong> This is what happens:</p>

<ol>
<li><p>The loop arrives at <code>_, resp = r.blpop('/task_finished')</code> and blocks there.</p></li>
<li><p>The task executes <code>r.rpush('/task_finished', task.request.id)</code></p></li>
<li><p>The loop unblocks, executes <code>task = celery.result.AsyncResult(task_id, app=celery_app)</code> and gets an empty result because the task has not yet recorded its result to the database.</p></li>
</ol>

<p>There may be a way do the <code>r.rpush</code> <em>after</em> celery has committed the results to the backend. Perhaps creating a custom class derived from <code>Task</code> would do it. But that's not something I've tried.</p>

<p>However, you could certainly modify your code to store the results <em>together</em> with the task id. Something like: </p>

<pre><code>r.rpush('/task_finished', json.dumps({ "task_id": task.request.id, "result": 42 }))
</code></pre>

<p>I've used a JSON serialization for the sake of illustration. You can use whatever scheme you want. On reading:</p>

<pre><code>_, resp = r.blpop('/task_finished')
resp = json.loads(resp)
</code></pre>

<p>With this, you might want to change <code>ignore_result=False</code> to <code>ignore_result=True</code>.</p>

python rabbitmq celery