<p dir="auto">I'm receiving an AssertionError when running Celery. The issue is persistent and repeatable in that, if I run a job with ~700,000 subtasks, the issue occurs at some point well into the job. However, it does not occur if I run a job with only 180 subtasks.</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="[2014-06-10 10:27:51,355: ERROR/MainProcess] Unrecoverable error: AssertionError()
Traceback (most recent call last):
  File &quot;/Users/beccapetrin/Documents/Development/automated_reports/lib/python2.7/site-packages/celery/worker/__init__.py&quot;, line 206, in start
self.blueprint.start(self)
  File &quot;/Users/beccapetrin/Documents/Development/automated_reports/lib/python2.7/site-packages/celery/bootsteps.py&quot;, line 123, in start
step.start(parent)
  File &quot;/Users/beccapetrin/Documents/Development/automated_reports/lib/python2.7/site-packages/celery/bootsteps.py&quot;, line 373, in start
return self.obj.start()
  File &quot;/Users/beccapetrin/Documents/Development/automated_reports/lib/python2.7/site-packages/celery/worker/consumer.py&quot;, line 278, in start
blueprint.start(self)
  File &quot;/Users/beccapetrin/Documents/Development/automated_reports/lib/python2.7/site-packages/celery/bootsteps.py&quot;, line 123, in start
step.start(parent)
  File &quot;/Users/beccapetrin/Documents/Development/automated_reports/lib/python2.7/site-packages/celery/worker/consumer.py&quot;, line 821, in start
c.loop(*c.loop_args())
  File &quot;/Users/beccapetrin/Documents/Development/automated_reports/lib/python2.7/site-packages/celery/worker/loops.py&quot;, line 72, in asynloop
next(loop)
  File &quot;/Users/beccapetrin/Documents/Development/automated_reports/lib/python2.7/site-packages/kombu/async/hub.py&quot;, line 324, in create_loop
cb(*cbargs)
  File &quot;/Users/beccapetrin/Documents/Development/automated_reports/lib/python2.7/site-packages/celery/concurrency/asynpool.py&quot;, line 263, in on_result_readable
next(it)
  File &quot;/Users/beccapetrin/Documents/Development/automated_reports/lib/python2.7/site-packages/celery/concurrency/asynpool.py&quot;, line 200, in _recv_message
assert not isblocking(fd)
AssertionError"><pre class="notranslate"><code class="notranslate">[2014-06-10 10:27:51,355: ERROR/MainProcess] Unrecoverable error: AssertionError()
Traceback (most recent call last):
  File "/Users/beccapetrin/Documents/Development/automated_reports/lib/python2.7/site-packages/celery/worker/__init__.py", line 206, in start
self.blueprint.start(self)
  File "/Users/beccapetrin/Documents/Development/automated_reports/lib/python2.7/site-packages/celery/bootsteps.py", line 123, in start
step.start(parent)
  File "/Users/beccapetrin/Documents/Development/automated_reports/lib/python2.7/site-packages/celery/bootsteps.py", line 373, in start
return self.obj.start()
  File "/Users/beccapetrin/Documents/Development/automated_reports/lib/python2.7/site-packages/celery/worker/consumer.py", line 278, in start
blueprint.start(self)
  File "/Users/beccapetrin/Documents/Development/automated_reports/lib/python2.7/site-packages/celery/bootsteps.py", line 123, in start
step.start(parent)
  File "/Users/beccapetrin/Documents/Development/automated_reports/lib/python2.7/site-packages/celery/worker/consumer.py", line 821, in start
c.loop(*c.loop_args())
  File "/Users/beccapetrin/Documents/Development/automated_reports/lib/python2.7/site-packages/celery/worker/loops.py", line 72, in asynloop
next(loop)
  File "/Users/beccapetrin/Documents/Development/automated_reports/lib/python2.7/site-packages/kombu/async/hub.py", line 324, in create_loop
cb(*cbargs)
  File "/Users/beccapetrin/Documents/Development/automated_reports/lib/python2.7/site-packages/celery/concurrency/asynpool.py", line 263, in on_result_readable
next(it)
  File "/Users/beccapetrin/Documents/Development/automated_reports/lib/python2.7/site-packages/celery/concurrency/asynpool.py", line 200, in _recv_message
assert not isblocking(fd)
AssertionError
</code></pre></div>
<p dir="auto">Here are my config details:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="$ celery report

software -&gt; celery:3.1.12 (Cipater) kombu:3.0.19 py:2.7.5
            billiard:3.3.0.17 py-amqp:1.4.5
platform -&gt; system:Darwin arch:64bit imp:CPython
loader   -&gt; celery.loaders.default.Loader
settings -&gt; transport:amqp results:amqp://username:password@localhost:5672

CELERY_QUEUES:
    (&lt;unbound Queue automated_reports -&gt; &lt;unbound Exchange default(direct)&gt; -&gt; automated_reports&gt;,)
CELERY_DEFAULT_ROUTING_KEY: '********'
CELERY_IMPORTS:
    ('automated_reports.queue.tasks', 'automated_reports.queue.subtasks')
CELERY_RESULT_PERSISTENT: True
CELERY_ROUTES: {
    'automated_reports.queue.tasks.run_device_info_report': {   'queue': 'automated_reports'},
    'uploader.queue.subtasks.multithread_device_listing': {   'queue': 'automated_reports'},
    'uploader.queue.subtasks.multithread_individual_device': {   'queue': 'automated_reports'},
    'uploader.queue.tasks.multithread_device_listing': {   'queue': 'automated_reports'},
    'uploader.queue.tasks.multithread_individual_device': {   'queue': 'automated_reports'}}
CELERY_DEFAULT_QUEUE: 'automated_reports'
BROKER_URL: 'amqp://guest:********@localhost:5672//'
CELERY_RESULT_BACKEND: 'amqp://username:password@localhost:5672'"><pre class="notranslate"><code class="notranslate">$ celery report

software -&gt; celery:3.1.12 (Cipater) kombu:3.0.19 py:2.7.5
            billiard:3.3.0.17 py-amqp:1.4.5
platform -&gt; system:Darwin arch:64bit imp:CPython
loader   -&gt; celery.loaders.default.Loader
settings -&gt; transport:amqp results:amqp://username:password@localhost:5672

CELERY_QUEUES:
    (&lt;unbound Queue automated_reports -&gt; &lt;unbound Exchange default(direct)&gt; -&gt; automated_reports&gt;,)
CELERY_DEFAULT_ROUTING_KEY: '********'
CELERY_IMPORTS:
    ('automated_reports.queue.tasks', 'automated_reports.queue.subtasks')
CELERY_RESULT_PERSISTENT: True
CELERY_ROUTES: {
    'automated_reports.queue.tasks.run_device_info_report': {   'queue': 'automated_reports'},
    'uploader.queue.subtasks.multithread_device_listing': {   'queue': 'automated_reports'},
    'uploader.queue.subtasks.multithread_individual_device': {   'queue': 'automated_reports'},
    'uploader.queue.tasks.multithread_device_listing': {   'queue': 'automated_reports'},
    'uploader.queue.tasks.multithread_individual_device': {   'queue': 'automated_reports'}}
CELERY_DEFAULT_QUEUE: 'automated_reports'
BROKER_URL: 'amqp://guest:********@localhost:5672//'
CELERY_RESULT_BACKEND: 'amqp://username:password@localhost:5672'
</code></pre></div>
<p dir="auto">My worker startup command is:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="~/Documents/Development/automated_reports/bin/celeryd --loglevel=DEBUG --autoreload -A automated_reports.queue.tasks -Q automated_reports -B --schedule=~/Documents/Development/automated_reports/log/celerybeat --autoscale=10,3 -Ofair"><pre class="notranslate"><code class="notranslate">~/Documents/Development/automated_reports/bin/celeryd --loglevel=DEBUG --autoreload -A automated_reports.queue.tasks -Q automated_reports -B --schedule=~/Documents/Development/automated_reports/log/celerybeat --autoscale=10,3 -Ofair
</code></pre></div>
<p dir="auto">I run my application as a task, which then kicks off subtasks by using celery subtasks that are grouped together.</p>
<p dir="auto">OS version is Mac OS X, version 10.9.3.</p>
<p dir="auto">Issue was also posted on StackOverflow originally to post details and chat about before posting here: <a href="http://stackoverflow.com/questions/24152180/celery-assertionerror" rel="nofollow">http://stackoverflow.com/questions/24152180/celery-assertionerror</a> . . . I will add the outcome there too.</p>
<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>
(if you are not able to do this, then at least specify the Celery<br>
version affected).</li>
</ul>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="software -&gt; celery:4.0.2 (latentcall) kombu:4.0.2 py:3.5.2
            billiard:3.5.0.2 py-amqp:2.1.4
platform -&gt; system:Darwin arch:64bit imp:CPython
loader   -&gt; celery.loaders.app.AppLoader
settings -&gt; transport:amqp results:redis://localhost/

result_backend: 'redis://localhost/'"><pre class="notranslate"><code class="notranslate">software -&gt; celery:4.0.2 (latentcall) kombu:4.0.2 py:3.5.2
            billiard:3.5.0.2 py-amqp:2.1.4
platform -&gt; system:Darwin arch:64bit imp:CPython
loader   -&gt; celery.loaders.app.AppLoader
settings -&gt; transport:amqp results:redis://localhost/

result_backend: 'redis://localhost/'
</code></pre></div>
<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.</li>
</ul>
<h2 dir="auto">Steps to reproduce:</h2>
<p dir="auto">Create code for a random task and a <code class="notranslate">task_revoked</code> handler in <code class="notranslate">app.py</code>:</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="import time

from celery import Celery
from celery.signals import task_revoked

app = Celery('app', backend='redis://localhost')

@app.task
def test():
    # random task
    counter = 0
    while counter &lt; 20:
        print(str(counter))
        time.sleep(1)
        counter += 1

@task_revoked.connect
def on_task_revoked(*args, **kwargs):
    print(str(kwargs))
    print('task_revoked')"><pre class="notranslate"><span class="pl-k">import</span> <span class="pl-s1">time</span>

<span class="pl-k">from</span> <span class="pl-s1">celery</span> <span class="pl-k">import</span> <span class="pl-v">Celery</span>
<span class="pl-k">from</span> <span class="pl-s1">celery</span>.<span class="pl-s1">signals</span> <span class="pl-k">import</span> <span class="pl-s1">task_revoked</span>

<span class="pl-s1">app</span> <span class="pl-c1">=</span> <span class="pl-v">Celery</span>(<span class="pl-s">'app'</span>, <span class="pl-s1">backend</span><span class="pl-c1">=</span><span class="pl-s">'redis://localhost'</span>)

<span class="pl-en">@<span class="pl-s1">app</span>.<span class="pl-s1">task</span></span>
<span class="pl-k">def</span> <span class="pl-en">test</span>():
    <span class="pl-c"># random task</span>
    <span class="pl-s1">counter</span> <span class="pl-c1">=</span> <span class="pl-c1">0</span>
    <span class="pl-k">while</span> <span class="pl-s1">counter</span> <span class="pl-c1">&lt;</span> <span class="pl-c1">20</span>:
        <span class="pl-en">print</span>(<span class="pl-en">str</span>(<span class="pl-s1">counter</span>))
        <span class="pl-s1">time</span>.<span class="pl-en">sleep</span>(<span class="pl-c1">1</span>)
        <span class="pl-s1">counter</span> <span class="pl-c1">+=</span> <span class="pl-c1">1</span>

<span class="pl-en">@<span class="pl-s1">task_revoked</span>.<span class="pl-s1">connect</span></span>
<span class="pl-k">def</span> <span class="pl-en">on_task_revoked</span>(<span class="pl-c1">*</span><span class="pl-s1">args</span>, <span class="pl-c1">**</span><span class="pl-s1">kwargs</span>):
    <span class="pl-en">print</span>(<span class="pl-en">str</span>(<span class="pl-s1">kwargs</span>))
    <span class="pl-en">print</span>(<span class="pl-s">'task_revoked'</span>)</pre></div>
<p dir="auto">Start worker:<br>
<code class="notranslate">celery -A app worker --loglevel=info</code></p>
<p dir="auto">Then start the task and revoke it:</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="&gt;&gt;&gt; from celery.task.control import revoke
&gt;&gt;&gt; from app import test
&gt;&gt;&gt; id = test.delay().id
&gt;&gt;&gt; revoke(id, terminate=True)"><pre class="notranslate"><span class="pl-c1">&gt;&gt;</span><span class="pl-c1">&gt;</span> <span class="pl-k">from</span> <span class="pl-s1">celery</span>.<span class="pl-s1">task</span>.<span class="pl-s1">control</span> <span class="pl-k">import</span> <span class="pl-s1">revoke</span>
<span class="pl-c1">&gt;&gt;</span><span class="pl-c1">&gt;</span> <span class="pl-k">from</span> <span class="pl-s1">app</span> <span class="pl-k">import</span> <span class="pl-s1">test</span>
<span class="pl-c1">&gt;&gt;</span><span class="pl-c1">&gt;</span> <span class="pl-s1">id</span> <span class="pl-c1">=</span> <span class="pl-s1">test</span>.<span class="pl-en">delay</span>().<span class="pl-s1">id</span>
<span class="pl-c1">&gt;&gt;</span><span class="pl-c1">&gt;</span> <span class="pl-en">revoke</span>(<span class="pl-s1">id</span>, <span class="pl-s1">terminate</span><span class="pl-c1">=</span><span class="pl-c1">True</span>)</pre></div>
<h2 dir="auto">Expected behavior</h2>
<p dir="auto">I would expect <code class="notranslate">on_task_revoked</code> to be called once.</p>
<h2 dir="auto">Actual behavior</h2>
<p dir="auto">Looking at the output of the worker, it seems that <code class="notranslate">on_task_revoked</code> is actually called twice:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="...
[2017-02-01 17:08:53,361: WARNING/PoolWorker-1] 6
[2017-02-01 17:08:54,362: WARNING/PoolWorker-1] 7
[2017-02-01 17:08:55,363: WARNING/PoolWorker-1] 8
[2017-02-01 17:08:55,576: INFO/MainProcess] Terminating 9f6f5f7a-8e74-4c6a-be8c-ac773042ab0d (Signals.SIGTERM)
[2017-02-01 17:08:55,582: WARNING/MainProcess] {'request': &lt;Request: app.test[9f6f5f7a-8e74-4c6a-be8c-ac773042ab0d] () {}&gt;, 'signal': &lt;Signal: task_revoked providing_args={'request', 'expired', 'terminated', 'signum'}&gt;, 'sender': &lt;@task: app.test of app:0x10b3f5400&gt;, 'signum': &lt;Signals.SIGTERM: 15&gt;, 'expired': False, 'terminated': True}
[2017-02-01 17:08:55,582: WARNING/MainProcess] task_revoked
[2017-02-01 17:08:55,596: WARNING/MainProcess] {'request': &lt;Request: app.test[9f6f5f7a-8e74-4c6a-be8c-ac773042ab0d] () {}&gt;, 'signal': &lt;Signal: task_revoked providing_args={'request', 'expired', 'terminated', 'signum'}&gt;, 'sender': &lt;@task: app.test of app:0x10b3f5400&gt;, 'signum': '15', 'expired': False, 'terminated': True}
[2017-02-01 17:08:55,596: WARNING/MainProcess] task_revoked
[2017-02-01 17:08:55,596: ERROR/MainProcess] Task handler raised error: Terminated(15,)
Traceback (most recent call last):
  File &quot;/Users/jdepoix/Temp/celery_test/virtualenv/lib/python3.5/site-packages/billiard/pool.py&quot;, line 1726, in _set_terminated
    raise Terminated(-(signum or 0))
billiard.exceptions.Terminated: 15"><pre class="notranslate"><code class="notranslate">...
[2017-02-01 17:08:53,361: WARNING/PoolWorker-1] 6
[2017-02-01 17:08:54,362: WARNING/PoolWorker-1] 7
[2017-02-01 17:08:55,363: WARNING/PoolWorker-1] 8
[2017-02-01 17:08:55,576: INFO/MainProcess] Terminating 9f6f5f7a-8e74-4c6a-be8c-ac773042ab0d (Signals.SIGTERM)
[2017-02-01 17:08:55,582: WARNING/MainProcess] {'request': &lt;Request: app.test[9f6f5f7a-8e74-4c6a-be8c-ac773042ab0d] () {}&gt;, 'signal': &lt;Signal: task_revoked providing_args={'request', 'expired', 'terminated', 'signum'}&gt;, 'sender': &lt;@task: app.test of app:0x10b3f5400&gt;, 'signum': &lt;Signals.SIGTERM: 15&gt;, 'expired': False, 'terminated': True}
[2017-02-01 17:08:55,582: WARNING/MainProcess] task_revoked
[2017-02-01 17:08:55,596: WARNING/MainProcess] {'request': &lt;Request: app.test[9f6f5f7a-8e74-4c6a-be8c-ac773042ab0d] () {}&gt;, 'signal': &lt;Signal: task_revoked providing_args={'request', 'expired', 'terminated', 'signum'}&gt;, 'sender': &lt;@task: app.test of app:0x10b3f5400&gt;, 'signum': '15', 'expired': False, 'terminated': True}
[2017-02-01 17:08:55,596: WARNING/MainProcess] task_revoked
[2017-02-01 17:08:55,596: ERROR/MainProcess] Task handler raised error: Terminated(15,)
Traceback (most recent call last):
  File "/Users/jdepoix/Temp/celery_test/virtualenv/lib/python3.5/site-packages/billiard/pool.py", line 1726, in _set_terminated
    raise Terminated(-(signum or 0))
billiard.exceptions.Terminated: 15
</code></pre></div>