<p dir="auto">Healthcheck fails after a few tries about in 5 minutes, and when I ping all the workers using</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="celery insect ping -A project"><pre class="notranslate"><code class="notranslate">celery insect ping -A project
</code></pre></div>
<p dir="auto">I don't find the worker, I find the other ones by the way.<br>
I have three dockerized workers, using Amazon ECS two of which have light work to do and are not dependent on a VPN so I'm using fargate and the one that gives me problems is on EC2 using host networking mode in a server with a VPN. The other two workers don't have the same start up problem.<br>
This worker has a beat that makes the worker do some work to update some sensors statuses every minute. It uses group calls to do parallel work.</p>
<h2 dir="auto">REPORT with part of django settings</h2>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="software -&gt; celery:4.2.1 (windowlicker) kombu:4.2.1 py:3.6.8
            billiard:3.5.0.4 redis:2.10.6
platform -&gt; system:Linux arch:64bit imp:CPython
loader   -&gt; celery.loaders.app.AppLoader
settings -&gt; transport:redis results:disabled

CELERY_ACCEPT_CONTENT: ['application/json']
CELERY_BEAT_SCHEDULE: {
    'curator_indexes_migration': {   'schedule': &lt;crontab: 0 4 * * * (m/h/d/dM/MY)&gt;,
                                     'task': 'sensor.tasks.statuses.curator_indexes_migration'},
    'fetch_all_configurations_backup': {   'schedule': &lt;crontab: 0 9 * * * (m/h/d/dM/MY)&gt;,
                                           'task': 'sensor.tasks.statuses.fetch_all_configurations_backup'},
    'update_images': {   'schedule': &lt;crontab: 0 * * * * (m/h/d/dM/MY)&gt;,
                         'task': 'sensor.tasks.statuses.update_images'},
    'update_statuses': {   'schedule': &lt;crontab: * * * * * (m/h/d/dM/MY)&gt;,
                           'task': 'sensor.tasks.statuses.update_statuses'}}
CELERY_BROKER_URL: 'redis://safety-staging.h1q2ld.0001.euw1.cache.amazonaws.com:6379/1'
CELERY_RESULT_BACKEND: None
CELERY_RESULT_SERIALIZER: 'json'
CELERY_TASK_ROUTES: {
    'notification.tasks.*': {'queue': 'notification'},
    'sensor.tasks.*': {'queue': 'sensor'}}
CELERY_TASK_SERIALIZER: 'json'
CELERY_TIMEZONE: 'Europe/Rome'"><pre class="notranslate"><code class="notranslate">software -&gt; celery:4.2.1 (windowlicker) kombu:4.2.1 py:3.6.8
            billiard:3.5.0.4 redis:2.10.6
platform -&gt; system:Linux arch:64bit imp:CPython
loader   -&gt; celery.loaders.app.AppLoader
settings -&gt; transport:redis results:disabled

CELERY_ACCEPT_CONTENT: ['application/json']
CELERY_BEAT_SCHEDULE: {
    'curator_indexes_migration': {   'schedule': &lt;crontab: 0 4 * * * (m/h/d/dM/MY)&gt;,
                                     'task': 'sensor.tasks.statuses.curator_indexes_migration'},
    'fetch_all_configurations_backup': {   'schedule': &lt;crontab: 0 9 * * * (m/h/d/dM/MY)&gt;,
                                           'task': 'sensor.tasks.statuses.fetch_all_configurations_backup'},
    'update_images': {   'schedule': &lt;crontab: 0 * * * * (m/h/d/dM/MY)&gt;,
                         'task': 'sensor.tasks.statuses.update_images'},
    'update_statuses': {   'schedule': &lt;crontab: * * * * * (m/h/d/dM/MY)&gt;,
                           'task': 'sensor.tasks.statuses.update_statuses'}}
CELERY_BROKER_URL: 'redis://safety-staging.h1q2ld.0001.euw1.cache.amazonaws.com:6379/1'
CELERY_RESULT_BACKEND: None
CELERY_RESULT_SERIALIZER: 'json'
CELERY_TASK_ROUTES: {
    'notification.tasks.*': {'queue': 'notification'},
    'sensor.tasks.*': {'queue': 'sensor'}}
CELERY_TASK_SERIALIZER: 'json'
CELERY_TIMEZONE: 'Europe/Rome'
</code></pre></div>
<h2 dir="auto">part of requirements.txt concerning celery and redis</h2>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="aioredis==1.2.0
celery==4.2.1
Django==1.11
hiredis==0.2.0
kombu==4.2.1
redis==2.10.6"><pre class="notranslate"><code class="notranslate">aioredis==1.2.0
celery==4.2.1
Django==1.11
hiredis==0.2.0
kombu==4.2.1
redis==2.10.6
</code></pre></div>
<h2 dir="auto">start script</h2>
<div class="highlight highlight-source-shell notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="#!/usr/bin/env bash
CELERY_DOMAIN=$(cat /dev/urandom | tr -dc 'a-zA-Z0-9' | fold -w 32 | head -n 1)
echo ${CELERY_DOMAIN} &gt; /home/user/celery_domain.txt
printf &quot;Celery domain is %s\n&quot; &quot;$CELERY_DOMAIN&quot;
celery worker -A centroservizi -l INFO --autoscale=4,2 -Q sensor -n sensor@${CELERY_DOMAIN}"><pre class="notranslate"><span class="pl-c"><span class="pl-c">#!</span>/usr/bin/env bash</span>
CELERY_DOMAIN=<span class="pl-s"><span class="pl-pds">$(</span>cat /dev/urandom <span class="pl-k">|</span> tr -dc <span class="pl-s"><span class="pl-pds">'</span>a-zA-Z0-9<span class="pl-pds">'</span></span> <span class="pl-k">|</span> fold -w 32 <span class="pl-k">|</span> head -n 1<span class="pl-pds">)</span></span>
<span class="pl-c1">echo</span> <span class="pl-smi">${CELERY_DOMAIN}</span> <span class="pl-k">&gt;</span> /home/user/celery_domain.txt
<span class="pl-c1">printf</span> <span class="pl-s"><span class="pl-pds">"</span>Celery domain is %s\n<span class="pl-pds">"</span></span> <span class="pl-s"><span class="pl-pds">"</span><span class="pl-smi">$CELERY_DOMAIN</span><span class="pl-pds">"</span></span>
celery worker -A centroservizi -l INFO --autoscale=4,2 -Q sensor -n sensor@<span class="pl-smi">${CELERY_DOMAIN}</span></pre></div>
<h2 dir="auto">health check script</h2>
<div class="highlight highlight-source-shell notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="#!/usr/bin/env bash
CELERY_DOMAIN=$(cat /home/user/celery_domain.txt)
printf &quot;Celery domain is %s\n&quot; &quot;$CELERY_DOMAIN&quot;
celery inspect ping -A centroservizi -d sensor@${CELERY_DOMAIN}"><pre class="notranslate"><span class="pl-c"><span class="pl-c">#!</span>/usr/bin/env bash</span>
CELERY_DOMAIN=<span class="pl-s"><span class="pl-pds">$(</span>cat /home/user/celery_domain.txt<span class="pl-pds">)</span></span>
<span class="pl-c1">printf</span> <span class="pl-s"><span class="pl-pds">"</span>Celery domain is %s\n<span class="pl-pds">"</span></span> <span class="pl-s"><span class="pl-pds">"</span><span class="pl-smi">$CELERY_DOMAIN</span><span class="pl-pds">"</span></span>
celery inspect ping -A centroservizi -d sensor@<span class="pl-smi">${CELERY_DOMAIN}</span></pre></div>
<h2 dir="auto">Start log</h2>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="Celery domain is G9ehGHbxIlsYFPb1JeW6Pdlar0gMIUue
[2019-01-08 07:14:57,184: INFO/MainProcess] Connected to redis://&lt;elastic cache&gt;.0001.euw1.cache.amazonaws.com:6379/1
[2019-01-08 07:14:57,200: INFO/MainProcess] mingle: searching for neighbors
[2019-01-08 07:14:58,306: INFO/MainProcess] mingle: sync with 2 nodes
[2019-01-08 07:14:58,307: INFO/MainProcess] mingle: sync complete
[2019-01-08 07:14:58,332: INFO/MainProcess] sensor@G9ehGHbxIlsYFPb1JeW6Pdlar0gMIUue ready.
[2019-01-08 07:14:59,026: INFO/MainProcess] Received task: sensor.tasks.statuses.update_sensor_status[2fa6e725-0a24-42c9-9171-572d7db3d31e]
[2019-01-08 07:14:59,033: INFO/MainProcess] Received task: sensor.tasks.statuses.update_sensor_status[296aaeb7-926a-4d1b-b62d-cec1783dbc70]
[2019-01-08 07:14:59,036: INFO/MainProcess] Received task: sensor.tasks.statuses.update_sensor_status[43cd23b1-5304-41df-ac82-1168cf53da60]
[2019-01-08 07:14:59,036: INFO/MainProcess] Scaling up 1 processes.
[2019-01-08 07:14:59,204: WARNING/MainProcess] consumer: Connection to broker lost. Trying to re-establish the connection...
Traceback (most recent call last):
File &quot;/usr/local/lib/python3.6/site-packages/redis/connection.py&quot;, line 590, in send_packed_command
self._sock.sendall(item)
BrokenPipeError: [Errno 32] Broken pipe
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File &quot;/usr/local/lib/python3.6/site-packages/celery/worker/consumer/consumer.py&quot;, line 317, in start
blueprint.start(self)
File &quot;/usr/local/lib/python3.6/site-packages/celery/bootsteps.py&quot;, line 119, in start
step.start(parent)
File &quot;/usr/local/lib/python3.6/site-packages/celery/worker/consumer/consumer.py&quot;, line 593, in start
c.loop(*c.loop_args())
File &quot;/usr/local/lib/python3.6/site-packages/celery/worker/loops.py&quot;, line 91, in asynloop
next(loop)
File &quot;/usr/local/lib/python3.6/site-packages/kombu/asynchronous/hub.py&quot;, line 276, in create_loop
tick_callback()
File &quot;/usr/local/lib/python3.6/site-packages/kombu/transport/redis.py&quot;, line 1033, in on_poll_start
cycle_poll_start()
File &quot;/usr/local/lib/python3.6/site-packages/kombu/transport/redis.py&quot;, line 315, in on_poll_start
self._register_BRPOP(channel)
File &quot;/usr/local/lib/python3.6/site-packages/kombu/transport/redis.py&quot;, line 301, in _register_BRPOP
channel._brpop_start()
File &quot;/usr/local/lib/python3.6/site-packages/kombu/transport/redis.py&quot;, line 707, in _brpop_start
self.client.connection.send_command('BRPOP', *keys)
File &quot;/usr/local/lib/python3.6/site-packages/redis/connection.py&quot;, line 610, in send_command
self.send_packed_command(self.pack_command(*args))
File &quot;/usr/local/lib/python3.6/site-packages/redis/connection.py&quot;, line 603, in send_packed_command
(errno, errmsg))
redis.exceptions.ConnectionError: Error 32 while writing to socket. Broken pipe."><pre class="notranslate"><code class="notranslate">Celery domain is G9ehGHbxIlsYFPb1JeW6Pdlar0gMIUue
[2019-01-08 07:14:57,184: INFO/MainProcess] Connected to redis://&lt;elastic cache&gt;.0001.euw1.cache.amazonaws.com:6379/1
[2019-01-08 07:14:57,200: INFO/MainProcess] mingle: searching for neighbors
[2019-01-08 07:14:58,306: INFO/MainProcess] mingle: sync with 2 nodes
[2019-01-08 07:14:58,307: INFO/MainProcess] mingle: sync complete
[2019-01-08 07:14:58,332: INFO/MainProcess] sensor@G9ehGHbxIlsYFPb1JeW6Pdlar0gMIUue ready.
[2019-01-08 07:14:59,026: INFO/MainProcess] Received task: sensor.tasks.statuses.update_sensor_status[2fa6e725-0a24-42c9-9171-572d7db3d31e]
[2019-01-08 07:14:59,033: INFO/MainProcess] Received task: sensor.tasks.statuses.update_sensor_status[296aaeb7-926a-4d1b-b62d-cec1783dbc70]
[2019-01-08 07:14:59,036: INFO/MainProcess] Received task: sensor.tasks.statuses.update_sensor_status[43cd23b1-5304-41df-ac82-1168cf53da60]
[2019-01-08 07:14:59,036: INFO/MainProcess] Scaling up 1 processes.
[2019-01-08 07:14:59,204: WARNING/MainProcess] consumer: Connection to broker lost. Trying to re-establish the connection...
Traceback (most recent call last):
File "/usr/local/lib/python3.6/site-packages/redis/connection.py", line 590, in send_packed_command
self._sock.sendall(item)
BrokenPipeError: [Errno 32] Broken pipe
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/usr/local/lib/python3.6/site-packages/celery/worker/consumer/consumer.py", line 317, in start
blueprint.start(self)
File "/usr/local/lib/python3.6/site-packages/celery/bootsteps.py", line 119, in start
step.start(parent)
File "/usr/local/lib/python3.6/site-packages/celery/worker/consumer/consumer.py", line 593, in start
c.loop(*c.loop_args())
File "/usr/local/lib/python3.6/site-packages/celery/worker/loops.py", line 91, in asynloop
next(loop)
File "/usr/local/lib/python3.6/site-packages/kombu/asynchronous/hub.py", line 276, in create_loop
tick_callback()
File "/usr/local/lib/python3.6/site-packages/kombu/transport/redis.py", line 1033, in on_poll_start
cycle_poll_start()
File "/usr/local/lib/python3.6/site-packages/kombu/transport/redis.py", line 315, in on_poll_start
self._register_BRPOP(channel)
File "/usr/local/lib/python3.6/site-packages/kombu/transport/redis.py", line 301, in _register_BRPOP
channel._brpop_start()
File "/usr/local/lib/python3.6/site-packages/kombu/transport/redis.py", line 707, in _brpop_start
self.client.connection.send_command('BRPOP', *keys)
File "/usr/local/lib/python3.6/site-packages/redis/connection.py", line 610, in send_command
self.send_packed_command(self.pack_command(*args))
File "/usr/local/lib/python3.6/site-packages/redis/connection.py", line 603, in send_packed_command
(errno, errmsg))
redis.exceptions.ConnectionError: Error 32 while writing to socket. Broken pipe.
</code></pre></div>
<p dir="auto">If I can give you other info on the issue just ask</p>
<h3 dir="auto">Versions</h3>
<p dir="auto">Kombu: 4.0.2<br>
Celery: 4.0.2<br>
redis-py: 2.10.5<br>
Redis: 3.2.6<br>
Python: 2.7.12</p>
<h3 dir="auto">Steps to reproduce</h3>
<p dir="auto">Start a <code class="notranslate">redis</code> server with <code class="notranslate">timeout 1</code>, I use the following config:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="daemonize yes
pidfile ./redis.pid
port 0
unixsocket /tmp/celery.redis.test.sock
unixsocketperm 755
timeout 1
loglevel notice
logfile ./redis.log
databases 1"><pre class="notranslate"><code class="notranslate">daemonize yes
pidfile ./redis.pid
port 0
unixsocket /tmp/celery.redis.test.sock
unixsocketperm 755
timeout 1
loglevel notice
logfile ./redis.log
databases 1
</code></pre></div>
<p dir="auto">(I shove the unix socket into <code class="notranslate">/tmp/</code> because there's a limit to how long a unix socket path can be.)</p>
<p dir="auto">I put the Celery configuration in <code class="notranslate">celeryconfig.py</code>:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="CELERY_BROKER_URL = 'redis+socket:///tmp/celery.redis.test.sock'
CELERY_RESULT_BACKEND = CELERY_BROKER_URL"><pre class="notranslate"><code class="notranslate">CELERY_BROKER_URL = 'redis+socket:///tmp/celery.redis.test.sock'
CELERY_RESULT_BACKEND = CELERY_BROKER_URL
</code></pre></div>
<p dir="auto">The Celery app is in <code class="notranslate">tasks.py</code>:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="import sys
import time

from celery import Celery, Task

app = Celery('tasks')
app.config_from_object('celeryconfig', namespace='CELERY')

@app.task
def add(x, y):
    return x + y"><pre class="notranslate"><code class="notranslate">import sys
import time

from celery import Celery, Task

app = Celery('tasks')
app.config_from_object('celeryconfig', namespace='CELERY')

@app.task
def add(x, y):
    return x + y
</code></pre></div>
<p dir="auto">I have <code class="notranslate">test.py</code>:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="from celery.bin.multi import MultiTool

from tasks import app, add

workers = [
    &quot;A&quot;,
    &quot;B&quot;
]

while True:
    for worker in workers:
        retcode = MultiTool().execute_from_commandline([&quot;multi&quot;, &quot;start&quot;, &quot;-A&quot;,
                                                        &quot;tasks&quot;, worker])
        print &quot;STARTED {0} WITH {1}&quot;.format(worker, retcode)

    print &quot;PING AFTER START&quot;, app.control.inspect().ping()
    print add.delay(1, 2).get()
    print &quot;PING AFTER TASK&quot;, app.control.inspect().ping()
    for worker in workers:

        retcode = MultiTool().execute_from_commandline([&quot;multi&quot;, &quot;stopwait&quot;, &quot;-A&quot;,
                                                        &quot;tasks&quot;, worker])
        print &quot;STOPPED {0} WITH {1}&quot;.format(worker, retcode)
    print &quot;PING AFTER STOP&quot;, app.control.inspect().ping()"><pre class="notranslate"><code class="notranslate">from celery.bin.multi import MultiTool

from tasks import app, add

workers = [
    "A",
    "B"
]

while True:
    for worker in workers:
        retcode = MultiTool().execute_from_commandline(["multi", "start", "-A",
                                                        "tasks", worker])
        print "STARTED {0} WITH {1}".format(worker, retcode)

    print "PING AFTER START", app.control.inspect().ping()
    print add.delay(1, 2).get()
    print "PING AFTER TASK", app.control.inspect().ping()
    for worker in workers:

        retcode = MultiTool().execute_from_commandline(["multi", "stopwait", "-A",
                                                        "tasks", worker])
        print "STOPPED {0} WITH {1}".format(worker, retcode)
    print "PING AFTER STOP", app.control.inspect().ping()
</code></pre></div>
<p dir="auto">Run <code class="notranslate">python test.py</code>.</p>
<h3 dir="auto">Expected Results</h3>
<p dir="auto">Should run forever without a crash.</p>
<h3 dir="auto">Actual Results</h3>
<p dir="auto">Crashes at the end of the first iteration:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="PING AFTER STOP
Traceback (most recent call last):
  File &quot;test.py&quot;, line 24, in &lt;module&gt;
    print &quot;PING AFTER STOP&quot;, app.control.inspect().ping()
  File &quot;/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/celery/app/control.py&quot;, line 113, in ping
    return self._request('ping')
  File &quot;/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/celery/app/control.py&quot;, line 81, in _request
    timeout=self.timeout, reply=True,
  File &quot;/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/celery/app/control.py&quot;, line 436, in broadcast
    limit, callback, channel=channel,
  File &quot;/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/kombu/pidbox.py&quot;, line 321, in _broadcast
    channel=chan)
  File &quot;/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/kombu/pidbox.py&quot;, line 360, in _collect
    self.connection.drain_events(timeout=timeout)
  File &quot;/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/kombu/connection.py&quot;, line 301, in drain_events
    return self.transport.drain_events(self.connection, **kwargs)
  File &quot;/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/kombu/transport/virtual/base.py&quot;, line 961, in drain_events
    get(self._deliver, timeout=timeout)
  File &quot;/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/kombu/transport/redis.py&quot;, line 352, in get
    self._register_BRPOP(channel)
  File &quot;/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/kombu/transport/redis.py&quot;, line 301, in _register_BRPOP
    channel._brpop_start()
  File &quot;/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/kombu/transport/redis.py&quot;, line 707, in _brpop_start
    self.client.connection.send_command('BRPOP', *keys)
  File &quot;/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/redis/connection.py&quot;, line 563, in send_command
    self.send_packed_command(self.pack_command(*args))
  File &quot;/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/redis/connection.py&quot;, line 556, in send_packed_command
    (errno, errmsg))
redis.exceptions.ConnectionError: Error 32 while writing to socket. Broken pipe."><pre class="notranslate"><code class="notranslate">PING AFTER STOP
Traceback (most recent call last):
  File "test.py", line 24, in &lt;module&gt;
    print "PING AFTER STOP", app.control.inspect().ping()
  File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/celery/app/control.py", line 113, in ping
    return self._request('ping')
  File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/celery/app/control.py", line 81, in _request
    timeout=self.timeout, reply=True,
  File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/celery/app/control.py", line 436, in broadcast
    limit, callback, channel=channel,
  File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/kombu/pidbox.py", line 321, in _broadcast
    channel=chan)
  File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/kombu/pidbox.py", line 360, in _collect
    self.connection.drain_events(timeout=timeout)
  File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/kombu/connection.py", line 301, in drain_events
    return self.transport.drain_events(self.connection, **kwargs)
  File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/kombu/transport/virtual/base.py", line 961, in drain_events
    get(self._deliver, timeout=timeout)
  File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/kombu/transport/redis.py", line 352, in get
    self._register_BRPOP(channel)
  File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/kombu/transport/redis.py", line 301, in _register_BRPOP
    channel._brpop_start()
  File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/kombu/transport/redis.py", line 707, in _brpop_start
    self.client.connection.send_command('BRPOP', *keys)
  File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/redis/connection.py", line 563, in send_command
    self.send_packed_command(self.pack_command(*args))
  File "/home/ldd/src/celery_issues/celery_issue_2/.venv/local/lib/python2.7/site-packages/redis/connection.py", line 556, in send_packed_command
    (errno, errmsg))
redis.exceptions.ConnectionError: Error 32 while writing to socket. Broken pipe.
</code></pre></div>
<h3 dir="auto">Observations</h3>
<p dir="auto">The code above in <code class="notranslate">test.py</code> imitates a sequence of operations that happen while testing an actual Django application of mine. The test suite starts and stops workers and executes some tasks on them.</p>
<p dir="auto">Though the problem surfaced after I upgraded to Celery 4.x and thus started uising Kombu 4.x, a summary inspection of Kombu's source code in the 3.x series suggests the problem is present there too. It is unclear to me why I did not run into it in when running Kombu 3.x.</p>
<p dir="auto">In my actual Redis setup I do not use a <code class="notranslate">timeout 1</code> setting. Using <code class="notranslate">timeout 1</code> is an easy way to cause Redis to close a connection. Other situations may be because the <code class="notranslate">tcp-keepalive</code> timeout deemed a connection "dead". Or because a client violated an output buffer limit, or for some other reason. What is clear is that Redis clients should be resilient in the face of connections that got closed by the server.</p>
<p dir="auto">Indeed, the code of <code class="notranslate">redis-py</code> will generally retry sending commands that fail. This can be seen in <a href="https://github.com/andymccurdy/redis-py/blob/2.10.5/redis/client.py#L566"><code class="notranslate">execute_command</code></a>:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="    def execute_command(self, *args, **options):
        &quot;Execute a command and return a parsed response&quot;
        pool = self.connection_pool
        command_name = args[0]
        connection = pool.get_connection(command_name, **options)
        try:
            connection.send_command(*args)
            return self.parse_response(connection, command_name, **options)
        except (ConnectionError, TimeoutError) as e:
            connection.disconnect()
            if not connection.retry_on_timeout and isinstance(e, TimeoutError):
                raise
            connection.send_command(*args)
            return self.parse_response(connection, command_name, **options)
        finally:
            pool.release(connection)"><pre class="notranslate"><code class="notranslate">    def execute_command(self, *args, **options):
        "Execute a command and return a parsed response"
        pool = self.connection_pool
        command_name = args[0]
        connection = pool.get_connection(command_name, **options)
        try:
            connection.send_command(*args)
            return self.parse_response(connection, command_name, **options)
        except (ConnectionError, TimeoutError) as e:
            connection.disconnect()
            if not connection.retry_on_timeout and isinstance(e, TimeoutError):
                raise
            connection.send_command(*args)
            return self.parse_response(connection, command_name, **options)
        finally:
            pool.release(connection)
</code></pre></div>
<p dir="auto">Kombu's Redis <code class="notranslate">Channel</code> will generally benefit from <code class="notranslate">redis-py</code>'s automatic retrying because most of the time the methods it calls on its client ultimately run <code class="notranslate">execute_command</code>. However, <a href="https://github.com/celery/kombu/blob/v4.0.2/kombu/transport/redis.py#L700"><code class="notranslate">_brpop_start</code></a> calls <code class="notranslate">self.client.connection.send_command('BRPOP', *keys)</code>. If this call fails, the failure is automatically sent  up the stack without a retry.</p>
<p dir="auto">I've been able to work around the issue by changing the code of <code class="notranslate">_brpop_start</code> to:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="    def _brpop_start(self, timeout=1):
        queues = self._queue_cycle.consume(len(self.active_queues))
        if not queues:
            return
        keys = [self._q_for_pri(queue, pri) for pri in self.priority_steps
                for queue in queues] + [timeout or 0]
        self._in_poll = self.client.connection
        from redis.exceptions import ConnectionError
        try:
            self.client.connection.send_command('BRPOP', *keys)
        except ConnectionError:
            self.client.connection.send_command('BRPOP', *keys)"><pre class="notranslate"><code class="notranslate">    def _brpop_start(self, timeout=1):
        queues = self._queue_cycle.consume(len(self.active_queues))
        if not queues:
            return
        keys = [self._q_for_pri(queue, pri) for pri in self.priority_steps
                for queue in queues] + [timeout or 0]
        self._in_poll = self.client.connection
        from redis.exceptions import ConnectionError
        try:
            self.client.connection.send_command('BRPOP', *keys)
        except ConnectionError:
            self.client.connection.send_command('BRPOP', *keys)
</code></pre></div>