Issue 1:
Checklist

I have verified that the issue exists against the master branch of Celery.

This has already been asked to the discussion group first.

I have read the relevant section in the

contribution guide

on reporting bugs.

I have checked the issues list

for similar or identical bug reports.

I have checked the pull requests list

for existing proposed fixes.

I have checked the commit log

to find out if the bug was already fixed in the master branch.

I have included all related issues and possible duplicate issues

in this issue (If there are none, check this box anyway).

Related Issues

celery/py-amqp#330

Possible Duplicates

#1779, #2066

Minimally Reproducible Test Case

See celery/py-amqp#330 (comment)

Expected Behavior

Celery should be able to be called from multiple threads - e.g. from gunicorn container

Actual Behavior

Due race conditions inside backends, Celery serving data is stalled or it returns "0x01 while expecting 0xce" errors, or others (depends on backend) - see celery/py-amqp#330, #1779, #2066

Details

The issue is caused by sharing resources between threads. This issue can be broken into 2 separate sub-issues:

Underlying Connection is shared between threads:

When rpc backend is used, kombu/py-amqp backend is used. This backend supports only one connection per thread. Unfortunately since using single backend object in all threads, all threads shares single connections causing "0x01 while expecting 0xce" errors or other creashes:

celery/celery/app/base.py

Lines 1166 to 1167
in
05da357

@cached_property

def backend(self):

Single oid is shared between threads:

This unique identifier is used to define reply-to queue of rpc backend. Again, since the oid is global it is caused that all threads are having single result queue causing not receiving responses:

celery/celery/app/base.py

Lines 1152 to 1153
in
05da357

@cached_property

def oid(self):

Possible solution

Store these global variables to thread local storage:

https://docs.python.org/3/library/threading.html#thread-local-data

Issue 2:
Checklist

I have included the output of celery -A proj report in the issue.

(if you are not able to do this, then at least specify the Celery

version affected).

I have verified that the issue exists against the master branch of Celery.

software -> celery:4.1.0 (latentcall) kombu:4.1.0 py:3.5.4
billiard:3.5.0.3 py-amqp:2.2.2
platform -> system:Linux arch:64bit imp:CPython
loader -> celery.loaders.app.AppLoader
settings -> transport:amqp results:rpc:///

result_backend: 'rpc:///'
broker_url: 'amqp://guest:********@localhost:5672//'

Steps to reproduce

Using code from the user manual, but two threads running concurrently:

from celery import Celery
import time
from threading import Thread

app = Celery('test', broker='amqp://guest@localhost//', backend='rpc://')

@app.task(bind=True)
def hello(self, a, b):
time.sleep(1)
self.update_state(state="PROGRESS", meta={'progress': 50})
time.sleep(1)
self.update_state(state="PROGRESS", meta={'progress': 90})
time.sleep(1)
return 'hello world: %i' % (a+b)

if __name__ == '__main__':
def run ():
handle = hello.delay (1, 2)
print (handle.get ())
t1 = Thread (target=run)
t2 = Thread (target=run)
t1.start ()
t2.start ()
t1.join ()
t2.join ()

Expected behavior

hello world: 3
hello world: 3

Actual behavior

Different exceptions, depending on timing. For instance:

Exception in thread Thread-1:
Traceback (most recent call last):
File "/…/sandbox/lib/python3.5/site-packages/celery/backends/async.py", line 154, in add_pending_result
self._maybe_resolve_from_buffer(result)
File "/…/sandbox/lib/python3.5/site-packages/celery/backends/async.py", line 160, in _maybe_resolve_from_buffer
result._maybe_set_cache(self._pending_messages.take(result.id))
File "/…/sandbox/lib/python3.5/site-packages/celery/utils/collections.py", line 871, in take
raise self.Empty()
queue.Empty

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/…/sandbox/lib/python3.5/site-packages/amqp/connection.py", line 456, in channel
return self.channels[channel_id]
KeyError: None

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/usr/lib64/python3.5/threading.py", line 914, in _bootstrap_inner
self.run()
File "/usr/lib64/python3.5/threading.py", line 862, in run
self._target(*self._args, **self._kwargs)
File "test.py", line 19, in run
print (handle.get ())
File "/…/sandbox/lib/python3.5/site-packages/celery/result.py", line 193, in get
self.backend.add_pending_result(self)
File "/…/sandbox/lib/python3.5/site-packages/celery/backends/async.py", line 156, in add_pending_result
self._add_pending_result(result.id, result, weak=weak)
File "/…/sandbox/lib/python3.5/site-packages/celery/backends/async.py", line 166, in _add_pending_result
self.result_consumer.consume_from(task_id)
File "/…/sandbox/lib/python3.5/site-packages/celery/backends/rpc.py", line 81, in consume_from
return self.start(task_id)
File "/…/sandbox/lib/python3.5/site-packages/celery/backends/rpc.py", line 56, in start
self._connection.default_channel, [initial_queue],
File "/…/sandbox/lib/python3.5/site-packages/kombu/connection.py", line 821, in default_channel
self._default_channel = self.channel()
File "/…/sandbox/lib/python3.5/site-packages/kombu/connection.py", line 266, in channel
chan = self.transport.create_channel(self.connection)
File "/…/sandbox/lib/python3.5/site-packages/kombu/transport/pyamqp.py", line 100, in create_channel
return connection.channel()
File "/…/sandbox/lib/python3.5/site-packages/amqp/connection.py", line 459, in channel
channel.open()
File "/…/sandbox/lib/python3.5/site-packages/amqp/channel.py", line 432, in open
spec.Channel.Open, 's', ('',), wait=spec.Channel.OpenOk,
File "/…/sandbox/lib/python3.5/site-packages/amqp/abstract_channel.py", line 59, in send_method
return self.wait(wait, returns_tuple=returns_tuple)
File "/…/sandbox/lib/python3.5/site-packages/amqp/abstract_channel.py", line 79, in wait
self.connection.drain_events(timeout=timeout)
File "/…/sandbox/lib/python3.5/site-packages/amqp/connection.py", line 468, in drain_events
while not self.blocking_read(timeout):
File "/…/sandbox/lib/python3.5/site-packages/amqp/connection.py", line 473, in blocking_read
frame = self.transport.read_frame()
File "/…/sandbox/lib/python3.5/site-packages/amqp/transport.py", line 252, in read_frame
payload = read(size)
File "/…/sandbox/lib/python3.5/site-packages/amqp/transport.py", line 417, in _read
s = recv(n - len(rbuf))
socket.timeout: timed out

or

Exception in thread Thread-1:
Traceback (most recent call last):
File "/…/sandbox/lib/python3.5/site-packages/celery/backends/async.py", line 154, in add_pending_result
self._maybe_resolve_from_buffer(result)
File "/…/sandbox/lib/python3.5/site-packages/celery/backends/async.py", line 160, in _maybe_resolve_from_buffer
result._maybe_set_cache(self._pending_messages.take(result.id))
File "/…/sandbox/lib/python3.5/site-packages/celery/utils/collections.py", line 871, in take
raise self.Empty()
queue.Empty

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/usr/lib64/python3.5/threading.py", line 914, in _bootstrap_inner
self.run()
File "/usr/lib64/python3.5/threading.py", line 862, in run
self._target(*self._args, **self._kwargs)
File "test.py", line 19, in run
print (handle.get ())
File "/…/sandbox/lib/python3.5/site-packages/celery/result.py", line 193, in get
self.backend.add_pending_result(self)
File "/…/sandbox/lib/python3.5/site-packages/celery/backends/async.py", line 156, in add_pending_result
self._add_pending_result(result.id, result, weak=weak)
File "/…/sandbox/lib/python3.5/site-packages/celery/backends/async.py", line 166, in _add_pending_result
self.result_consumer.consume_from(task_id)
File "/…/sandbox/lib/python3.5/site-packages/celery/backends/rpc.py", line 81, in consume_from
return self.start(task_id)
File "/…/sandbox/lib/python3.5/site-packages/celery/backends/rpc.py", line 59, in start
self._consumer.consume()
File "/…/sandbox/lib/python3.5/site-packages/kombu/messaging.py", line 477, in consume
self._basic_consume(T, no_ack=no_ack, nowait=False)
File "/…/sandbox/lib/python3.5/site-packages/kombu/messaging.py", line 598, in _basic_consume
no_ack=no_ack, nowait=nowait)
File "/…/sandbox/lib/python3.5/site-packages/kombu/entity.py", line 737, in consume
arguments=self.consumer_arguments)
File "/…/sandbox/lib/python3.5/site-packages/amqp/channel.py", line 1564, in basic_consume
wait=None if nowait else spec.Basic.ConsumeOk,
File "/…/sandbox/lib/python3.5/site-packages/amqp/abstract_channel.py", line 59, in send_method
return self.wait(wait, returns_tuple=returns_tuple)
File "/…/sandbox/lib/python3.5/site-packages/amqp/abstract_channel.py", line 79, in wait
self.connection.drain_events(timeout=timeout)
File "/…/sandbox/lib/python3.5/site-packages/amqp/connection.py", line 471, in drain_events
while not self.blocking_read(timeout):
File "/…/sandbox/lib/python3.5/site-packages/amqp/connection.py", line 476, in blocking_read
frame = self.transport.read_frame()
File "/…/sandbox/lib/python3.5/site-packages/amqp/transport.py", line 254, in read_frame
'Received {0:#04x} while expecting 0xce'.format(ch))
amqp.exceptions.UnexpectedFrame: Received 0x3c while expecting 0xce