title: Undesired delay in the celery process
body: <p>I am encountering an undesired delay in the celery process that I cannot explain. My intent is to manage live processing of incoming data (at a rate of 10 to 60 data per seconds). Processing of one piece of data is divided into two fully sequential tasks but parallelization is used to start processing the next piece of data (with task 1) while processing the current one (with task 2) is not finished yet. Getting the shortest delay in the process is of at-most importance since it is a live application. </p>
  
  <p>Once in a while, I encounter a freeze in the process. To see where this problem came from I started monitoring the occupation of my workers. It appeared that it happened during the communication between workers. I designed the lightest and simplest example to illustrate it here. </p>
  
  <p>Here is my code, as you can see I have two tasks doing nothing but waiting 10ms each. I call them by using celery chains once every 20ms. I track each workers occupation by using prerun and postrun along with logging. In most of the case all is happening sequentially as time spent by both the workers doesn't exceed the send rate. </p>
  
  <pre><code>from __future__ import absolute_import
  
  import time
  from celery import chain
  from celery.signals import task_prerun, task_postrun
  from celery import Celery
  from kombu import Queue, Exchange
  
  
  N_ITS = 100000  # Total number of chains sent
  LOG_FILE = 'log_file.txt'  # Path to the log file
  
  
  def write_to_log_file(text):
      with open(LOG_FILE, 'a') as f:
          f.write(text)
  
  
  # Create celery app
  app = Celery('live')
  app.config_from_object('celeryconfig')
  
  default_exchange = Exchange('default', type='direct')
  app.conf.task_queues = tuple(Queue(route['queue'], default_exchange, routing_key=route['queue'])
                                      for route in app.conf.task_routes.values() + [{'queue': 'default'}])
  app.conf.update(result_expires=3600)
  
  
  # Define functions that record timings
  @task_prerun.connect()
  def task_prerun(signal=None, sender=None, task_id=None, task=None, **kwargs):
      text = 'task_prerun; {0}; {1:.16g}\n'.format(task.name, time.time())
      write_to_log_file(text)
  
  
  @task_postrun.connect()
  def task_postrun(signal=None, sender=None, task_id=None, task=None, **kwargs):
      text = 'task_postrun; {0}; {1:.16g}\n'.format(task.name, time.time())
      write_to_log_file(text)
  
  
  # Define tasks
  @app.task
  def task_1(i):
      print 'Executing task_1: {}'.format(i)
      time.sleep(0.01)
  
  
  @app.task
  def task_2(i):
      print 'Executing task_2: {}'.format(i)
      time.sleep(0.01)
  
  
  # Send chained tasks
  def main():
      celery_chains = []
      for i in range(N_ITS):
          print '[{}] - Dispatching tasks'.format(i)
          celery_chains.append(chain(task_1.si(i) | task_2.si(i))())
          time.sleep(0.02)
  
      # wait for all tasks to complete
      [c.get() for c in celery_chains]
  
  
  if __name__ == '__main__':
      main()
  </code></pre>
  
  <p>I also give the configuration of celery if needed:</p>
  
  <pre><code>from __future__ import absolute_import
  
  import os
  
  name = 'live'
  
  broker_url = 'pyamqp://{}'.format(os.environ.get('RMQ_HOST', 'localhost'))
  print 'broker_url:', broker_url
  include = ['live']
  
  DEFAULT_QUEUE = 'celery'
  
  # A named queue that's not already defined in task_queues will be created automatically.
  task_create_missing_queues = True
  
  broker_pool_limit = 10000
  
  task_routes = {
      'live.task_1': {'queue': 'worker_1'},
      'live.task_2': {'queue': 'worker_2'}
  }
  
  # We always set the routing key to be the queue name so we do it here automatically.
  for v in task_routes.values():
      v.update({'routing_key': v['queue']})
  
  task_serializer = 'pickle'
  result_serializer = 'pickle'
  accept_content = ['json', 'pickle']
  timezone = 'Europe/Paris'
  enable_utc = True
  </code></pre>
  
  <p>For the broker, I use the docker image rabbitmq:3.6-alpine with basic configurations appart that I enabled rabbitmq_management.</p>
  
  <p>This resuts in the following worker occupation chronogram: (the color indicates the index of the data being processed, so you can link tasks belonging to the same chain)</p>
  
  <p><img src="https://i.stack.imgur.com/F5e7Z.png" alt="enter image description here"></p>
  
  <p>As you can see, usually everything goes well and task 2 is called right after task 1 is finished. However, sometimes (indicated by the arrows on the figure) task 2 doesn't start immediately even though worker 2 isn't occupied. It imputes a delay of 27ms, which is more than twice the duration of a single task. This happened approximately every 2 seconds during this execution. </p>
  
  <p>I made some additionnal investigation using firehose to study the message exchange in rabbitmq and it resulted that the messages are effectively sent on time. To my understanding, the worker waits to go fetch the message and process the task, but I cannot understand why. </p>
  
  <p>I tried setting the broker pool limit to a high number but the issue remains. </p>