title: Celery worker connection with Rabbitmq met broken pipe error In gevent or eventlet Mode
body: <p>I run to issue that Celery worker connection with Rabbitmq met broken pipe error IN Gevent Mode. While no problem when Celery worker work in Process pool mode (without gevent without monkey patch). </p>
  
  <p>After that, Celery workers will not get task messages from Rabbitmq anymore until they are restarted.</p>
  
  <p>That issue always happen when the speed of Celery workers consuming task messages slower than Django applications producing messages, and about 3 thounds of messages piled in Rabbitmq.</p>
  
  <p>Gevent version 1.1.0</p>
  
  <p>Celery version 3.1.22</p>
  
  <p>======  Celery log ======</p>
  
  <pre><code>[2016-08-08 13:52:06,913: CRITICAL/MainProcess] Couldn't ack 293, reason:error(32, 'Broken pipe')
  Traceback (most recent call last):
    File "/usr/local/lib/python2.7/site-packages/kombu/message.py", line 93, in ack_log_error
      self.ack()
    File "/usr/local/lib/python2.7/site-packages/kombu/message.py", line 88, in ack
      self.channel.basic_ack(self.delivery_tag)
    File "/usr/local/lib/python2.7/site-packages/amqp/channel.py", line 1584, in basic_ack
      self._send_method((60, 80), args)
    File "/usr/local/lib/python2.7/site-packages/amqp/abstract_channel.py", line 56, in _send_method
      self.channel_id, method_sig, args, content,
    File "/usr/local/lib/python2.7/site-packages/amqp/method_framing.py", line 221, in write_method
      write_frame(1, channel, payload)
    File "/usr/local/lib/python2.7/site-packages/amqp/transport.py", line 182, in write_frame
      frame_type, channel, size, payload, 0xce,
    File "/usr/local/lib/python2.7/site-packages/gevent/_socket2.py", line 412, in sendall
      timeleft = self.__send_chunk(chunk, flags, timeleft, end)
    File "/usr/local/lib/python2.7/site-packages/gevent/_socket2.py", line 351, in __send_chunk
      data_sent += self.send(chunk, flags)
    File "/usr/local/lib/python2.7/site-packages/gevent/_socket2.py", line 320, in send
      return sock.send(data, flags)
  error: [Errno 32] Broken pipe
  </code></pre>
  
  <p>======= Rabbitmq log ==================</p>
  
  <pre><code>=ERROR REPORT==== 8-Aug-2016::14:28:33 ===
  closing AMQP connection &lt;0.15928.4&gt; (10.26.39.183:60732 -&gt; 10.26.39.183:5672):
  {writer,send_failed,{error,enotconn}}
  
  =ERROR REPORT==== 8-Aug-2016::14:29:03 ===
  closing AMQP connection &lt;0.15981.4&gt; (10.26.39.183:60736 -&gt; 10.26.39.183:5672):
  {writer,send_failed,{error,enotconn}}
  
  =ERROR REPORT==== 8-Aug-2016::14:29:03 ===
  closing AMQP connection &lt;0.15955.4&gt; (10.26.39.183:60734 -&gt; 10.26.39.183:5672):
  {writer,send_failed,{error,enotconn}}
  </code></pre>
  
  <p>The similar issue appears when Celery worker use eventlet.</p>
  
  <pre><code>[2016-08-09 17:41:37,952: CRITICAL/MainProcess] Couldn't ack 583, reason:error(32, 'Broken pipe')
  Traceback (most recent call last):
    File "/usr/local/lib/python2.7/site-packages/kombu/message.py", line 93, in ack_log_error
      self.ack()
    File "/usr/local/lib/python2.7/site-packages/kombu/message.py", line 88, in ack
      self.channel.basic_ack(self.delivery_tag)
    File "/usr/local/lib/python2.7/site-packages/amqp/channel.py", line 1584, in basic_ack
      self._send_method((60, 80), args)
    File "/usr/local/lib/python2.7/site-packages/amqp/abstract_channel.py", line 56, in _send_method
      self.channel_id, method_sig, args, content,
    File "/usr/local/lib/python2.7/site-packages/amqp/method_framing.py", line 221, in write_method
      write_frame(1, channel, payload)
    File "/usr/local/lib/python2.7/site-packages/amqp/transport.py", line 182, in write_frame
      frame_type, channel, size, payload, 0xce,
    File "/usr/local/lib/python2.7/site-packages/eventlet/greenio/base.py", line 385, in sendall
      tail = self.send(data, flags)
    File "/usr/local/lib/python2.7/site-packages/eventlet/greenio/base.py", line 379, in send
      return self._send_loop(self.fd.send, data, flags)
    File "/usr/local/lib/python2.7/site-packages/eventlet/greenio/base.py", line 366, in _send_loop
      return send_method(data, *args)
  error: [Errno 32] Broken pipe
  </code></pre>
  
  <hr>
  
  <p>Add setup and load test info </p>
  
  <p>We use supervisor to launch celery with the following options</p>
  
  <pre><code>celery worker -A celerytasks.celery_worker_init -Q default -P gevent -c 1000 --loglevel=info
  </code></pre>
  
  <p>And Celery use Rabbitmq as broker.</p>
  
  <p>And we have 4 Celery worker processes by specifying "numprocs=4" in supervisor configurations.</p>
  
  <p>We use jmeter to emulate web access load, Django applications will produces tasks for Celery workers to consume. Those tasks basically need to access Mysql DB to get/update some data.</p>
  
  <p>From rabbitmq web admin page, tasks-producing speed is like 50 per seconds while consuming speed is like 20 per seconds. After about 1 miniutes load testing, log file shows many connections between Rabbitmq and Celery met Broken-Pipe error</p>