<h2 dir="auto">Checklist</h2>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I have included the output of <code class="notranslate">celery -A proj report</code> in the issue.<br>
software -&gt; celery:4.2.1 (windowlicker) kombu:4.2.1 py:3.6.6<br>
billiard:3.5.0.4 py-amqp:2.3.2<br>
platform -&gt; system:Linux arch:64bit, ELF imp:CPython<br>
loader   -&gt; celery.loaders.app.AppLoader<br>
settings -&gt; transport:pyamqp results:redis://localhost/</li>
</ul>
<p dir="auto">broker_url: 'amqp://guest:********<a class="user-mention notranslate" data-hovercard-type="user" data-hovercard-url="/users/localhost/hovercard" data-octo-click="hovercard-link-click" data-octo-dimensions="link_type:self" href="https://github.com/localhost">@localhost</a>:5672//'<br>
result_backend: 'redis://localhost/'</p>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> I have verified that the issue exists against the <code class="notranslate">master</code> branch of Celery.<br>
<em>I've installed Celery using pip.</em></li>
</ul>
<h2 dir="auto">Environment</h2>
<p dir="auto">Miniconda - Python 3.6.6 (see requirements.txt for details below)<br>
Debian 9 running in Virtualbox. VM Config: 4 cores and 8GB RAM<br>
Redis 4.0.11<br>
RabbitMQ 3.6.6 on Erlang 19.2.1</p>
<h2 dir="auto">Steps to reproduce</h2>
<ol start="0" dir="auto">
<li>
<p dir="auto">Install RabbitMQ and Redis, use default config.</p>
</li>
<li>
<p dir="auto">Create conda environment using <a href="https://github.com/celery/celery/files/2360605/requirements.txt">requirements.txt</a></p>
</li>
<li>
<p dir="auto">Save the script below as tasks.py</p>
</li>
<li>
<p dir="auto">Start a worker: <code class="notranslate">celery -A tasks worker -l info</code></p>
</li>
<li>
<p dir="auto">Start python and call the <code class="notranslate">start()</code> function</p>
</li>
</ol>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="&gt;&gt; from tasks import *
&gt;&gt; start(20,100)
"><pre class="notranslate"><code class="notranslate">&gt;&gt; from tasks import *
&gt;&gt; start(20,100)

</code></pre></div>
<p dir="auto">tasks.py:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="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))"><pre class="notranslate"><code class="notranslate">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))
</code></pre></div>
<h2 dir="auto">Expected behavior</h2>
<p dir="auto">The script should run without error even if the parallel_tasks value is relatively high.</p>
<h2 dir="auto">Actual behavior</h2>
<p dir="auto">Execution stops somewhere between the 5th and 12th iteration in the <code class="notranslate">start()</code> function. Note that the higher the <code class="notranslate">parallel_tasks</code> value the sooner process stuck. The example script runs just fine when  the <code class="notranslate">parallel_tasks</code> value is relatively low. (e.g. 4).<br>
If is set a timeout <code class="notranslate">results = job.apply_async().join(timeout=120)</code> it throws the following exception:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="Traceback (most recent call last):
  File &quot;/home/bakcsa/tools/miniconda3/envs/cel/lib/python3.6/site-packages/celery/backends/async.py&quot;, line 255, in _wait_for_pending
    on_interval=on_interval):
  File &quot;/home/bakcsa/tools/miniconda3/envs/cel/lib/python3.6/site-packages/celery/backends/async.py&quot;, 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 &quot;&lt;stdin&gt;&quot;, line 1, in &lt;module&gt;
  File &quot;/home/bakcsa/work/awst/projects/qa4sm/celery-test/celery_test/tasks2.py&quot;, line 31, in start
    compute(parallel_tasks)
  File &quot;/home/bakcsa/work/awst/projects/qa4sm/celery-test/celery_test/tasks2.py&quot;, line 25, in compute
    print(result.get(timeout=20))
  File &quot;/home/bakcsa/tools/miniconda3/envs/cel/lib/python3.6/site-packages/celery/result.py&quot;, line 224, in get
    on_message=on_message,
  File &quot;/home/bakcsa/tools/miniconda3/envs/cel/lib/python3.6/site-packages/celery/backends/async.py&quot;, line 188, in wait_for_pending
    for _ in self._wait_for_pending(result, **kwargs):
  File &quot;/home/bakcsa/tools/miniconda3/envs/cel/lib/python3.6/site-packages/celery/backends/async.py&quot;, line 259, in _wait_for_pending
    raise TimeoutError('The operation timed out.')
celery.exceptions.TimeoutError: The operation timed out."><pre class="notranslate"><code class="notranslate">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 "&lt;stdin&gt;", line 1, in &lt;module&gt;
  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.
</code></pre></div>
<h2 dir="auto">Checklist</h2>
<p dir="auto">software -&gt; celery:4.2.0 (windowlicker) kombu:4.2.1 py:3.6.3<br>
billiard:3.5.0.4 py-amqp:2.3.2<br>
platform -&gt; system:Linux arch:64bit imp:CPython<br>
loader   -&gt; celery.loaders.app.AppLoader<br>
settings -&gt; transport:amqp results:redis://127.0.0.1:6379/</p>
<p dir="auto">BROKER_URL: 'amqp://guest:********@127.0.0.1:5672//'<br>
CELERY_ACCEPT_CONTENT: ['json']<br>
CELERY_RESULT_BACKEND: 'redis://127.0.0.1:6379/'<br>
CELERY_RESULT_SERIALIZER: 'json'<br>
CELERY_TASK_PROTOCOL: 2<br>
CELERY_TASK_SERIALIZER: 'json'<br>
os: &lt;module 'os' from '/home/ehengao/miniconda3/envs/nia-dcs/lib/python3.6/os.py'&gt;<br>
CELERY_INCLUDE: ['nia_dcs.ccwp.tasks']</p>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I have verified that the issue exists against the <code class="notranslate">master</code> branch of Celery.<br>
yes</li>
</ul>
<h2 dir="auto">Steps to reproduce</h2>
<p dir="auto">Run following test</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="def test_add_with_group():
    &quot;&quot;&quot;test add task from the tutorial as a group
    &quot;&quot;&quot;
    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)"><pre class="notranslate"><code class="notranslate">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)
</code></pre></div>
<h2 dir="auto">Expected behavior</h2>
<p dir="auto">Task would pass every time. (randomly not always happen)</p>
<h2 dir="auto">Actual behavior</h2>
<p dir="auto">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.<strong>del</strong> 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.</p>