Issue 1:
Checklist

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

software -> celery:4.2.1 (windowlicker) kombu:4.2.1 py:3.6.6

billiard:3.5.0.4 py-amqp:2.3.2

platform -> system:Linux arch:64bit, ELF imp:CPython

loader -> celery.loaders.app.AppLoader

settings -> transport:pyamqp results:redis://localhost/

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

result_backend: 'redis://localhost/'

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

I've installed Celery using pip.

Environment

Miniconda - Python 3.6.6 (see requirements.txt for details below)

Debian 9 running in Virtualbox. VM Config: 4 cores and 8GB RAM

Redis 4.0.11

RabbitMQ 3.6.6 on Erlang 19.2.1

Steps to reproduce

Install RabbitMQ and Redis, use default config.

Create conda environment using requirements.txt

Save the script below as tasks.py

Start a worker: celery -A tasks worker -l info

Start python and call the start() function

>> from tasks import *
>> start(20,100)

tasks.py:

from celery import Celery
from celery import group
import time

app = Celery()
app.conf.update(
broker_url='pyamqp://guest@localhost//',
result_backend='redis://localhost',
)

@app.task
def my_calc(data):
for i in range(100):
data[0]=data[0]/1.04856
data[1]=data[1]/1.02496
return data

def compute(parallel_tasks):
tasks=[]
for i in range(parallel_tasks):
tasks.append([i+1.3,i+2.65])

job = group([my_calc.s(task) for task in tasks])
results = job.apply_async().join()
#results = job.apply_async().join(timeout=120)

def start(parallel_tasks,iterations):
for i in range(iterations):
exec_time = int(round(time.time() * 1000))
compute(parallel_tasks)
exec_time = int(round(time.time() * 1000)) - exec_time
print('Iteration {} Exec time: {}ms'.format(i,exec_time))

Expected behavior

The script should run without error even if the parallel_tasks value is relatively high.

Actual behavior

Execution stops somewhere between the 5th and 12th iteration in the start() function. Note that the higher the parallel_tasks value the sooner process stuck. The example script runs just fine when the parallel_tasks value is relatively low. (e.g. 4).

If is set a timeout results = job.apply_async().join(timeout=120) it throws the following exception:

Traceback (most recent call last):
File "/home/bakcsa/tools/miniconda3/envs/cel/lib/python3.6/site-packages/celery/backends/async.py", line 255, in _wait_for_pending
on_interval=on_interval):
File "/home/bakcsa/tools/miniconda3/envs/cel/lib/python3.6/site-packages/celery/backends/async.py", line 54, in drain_events_until
raise socket.timeout()
socket.timeout

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/home/bakcsa/work/awst/projects/qa4sm/celery-test/celery_test/tasks2.py", line 31, in start
compute(parallel_tasks)
File "/home/bakcsa/work/awst/projects/qa4sm/celery-test/celery_test/tasks2.py", line 25, in compute
print(result.get(timeout=20))
File "/home/bakcsa/tools/miniconda3/envs/cel/lib/python3.6/site-packages/celery/result.py", line 224, in get
on_message=on_message,
File "/home/bakcsa/tools/miniconda3/envs/cel/lib/python3.6/site-packages/celery/backends/async.py", line 188, in wait_for_pending
for _ in self._wait_for_pending(result, **kwargs):
File "/home/bakcsa/tools/miniconda3/envs/cel/lib/python3.6/site-packages/celery/backends/async.py", line 259, in _wait_for_pending
raise TimeoutError('The operation timed out.')
celery.exceptions.TimeoutError: The operation timed out.

Issue 2:
Checklist

software -> celery:4.2.0 (windowlicker) kombu:4.2.1 py:3.6.3

billiard:3.5.0.4 py-amqp:2.3.2

platform -> system:Linux arch:64bit imp:CPython

loader -> celery.loaders.app.AppLoader

settings -> transport:amqp results:redis://127.0.0.1:6379/

BROKER_URL: 'amqp://guest:********@127.0.0.1:5672//'

CELERY_ACCEPT_CONTENT: ['json']

CELERY_RESULT_BACKEND: 'redis://127.0.0.1:6379/'

CELERY_RESULT_SERIALIZER: 'json'

CELERY_TASK_PROTOCOL: 2

CELERY_TASK_SERIALIZER: 'json'

os: <module 'os' from '/home/ehengao/miniconda3/envs/nia-dcs/lib/python3.6/os.py'>

CELERY_INCLUDE: ['nia_dcs.ccwp.tasks']

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

yes

Steps to reproduce

Run following test

def test_add_with_group():
"""test add task from the tutorial as a group
"""
for i in range(100):
print(f'Runnint test {i}')
add1 = add.s(1, 2)
add2 = add.s(3, 4)
g = group(add1, add2)()
print(f'GroupTasks:{g} sent!')
r = g.get()
print(r)

Expected behavior

Task would pass every time. (randomly not always happen)

Actual behavior

Task some time blocked and causing pytest in continues integration never returns. After doing some digging into the celery code. noticed that things randomly happen when the AsyncResult.del triggered somehow to delete RedisBacken._pending_result before the result is adding to the ResultConsumer.buckets causing the GroupResult still waiting for one of the task to finish.