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

• UnexpectedFrame exception after "Connection to broker lost" #1779, "Received 0x01 while expecting 0xce" error when using python-amqp #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:

1. 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):

2. 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