We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Celery Worker crashing after first task with TypeError: 'NoneType' object is not callable
## Checklist

- [X] I have included the output of ``celery -A proj report`` in the issue.
      (if you are not able to do this, then at least specify the Celery
       version affected).
```
software -> celery:4.0.0 (latentcall) kombu:4.0.0 py:3.4.3
            billiard:3.5.0.2 py-amqp:2.1.1
platform -> system:Linux arch:64bit, ELF imp:CPython
loader   -> celery.loaders.default.Loader
settings -> transport:amqp results:disabled
```
- [X] I have verified that the issue exists against the `master` branch of Celery.
Yes I've tested and it behaves the same using master.

## Steps to reproduce
Not exactly sure, because other machines with the same specs and requirements are working.

## Expected behavior
Should consume tasks.

## Actual behavior
A task is accepted, then a traceback is logged, then the worker reconnects to the broker for some reason. This repeats forever:

```
[2016-11-23 23:09:00,468: INFO/MainProcess] Connected to amqp://user:**@10.136.131.6:5672//
[2016-11-23 23:09:00,484: INFO/MainProcess] mingle: searching for neighbors
[2016-11-23 23:09:01,921: INFO/MainProcess] mingle: sync with 1 nodes
[2016-11-23 23:09:01,922: INFO/MainProcess] mingle: sync complete
[2016-11-23 23:09:01,970: INFO/MainProcess] Received task: tasks.calculate_user_running_total[ddd103af-d527-4564-83f8-96b747767a0c]
[2016-11-23 23:09:01,972: CRITICAL/MainProcess] Unrecoverable error: TypeError("'NoneType' object is not callable",)
Traceback (most recent call last):
  File "./venv/lib/python3.4/site-packages/celery/worker/worker.py", line 203, in start
    self.blueprint.start(self)
  File "./venv/lib/python3.4/site-packages/celery/bootsteps.py", line 119, in start
    step.start(parent)
  File "./venv/lib/python3.4/site-packages/celery/bootsteps.py", line 370, in start
    return self.obj.start()
  File "./venv/lib/python3.4/site-packages/celery/worker/consumer/consumer.py", line 318, in start
    blueprint.start(self)
  File "./venv/lib/python3.4/site-packages/celery/bootsteps.py", line 119, in start
    step.start(parent)
  File "./venv/lib/python3.4/site-packages/celery/worker/consumer/consumer.py", line 584, in start
    c.loop(*c.loop_args())
  File "./venv/lib/python3.4/site-packages/celery/worker/loops.py", line 47, in asynloop
    consumer.consume()
  File "./venv/lib/python3.4/site-packages/kombu/messaging.py", line 470, in consume
    self._basic_consume(T, no_ack=no_ack, nowait=False)
  File "./venv/lib/python3.4/site-packages/kombu/messaging.py", line 591, in _basic_consume
    no_ack=no_ack, nowait=nowait)
  File "./venv/lib/python3.4/site-packages/kombu/entity.py", line 737, in consume
    arguments=self.consumer_arguments)
  File "./venv/lib/python3.4/site-packages/amqp/channel.py", line 1578, in basic_consume
    wait=None if nowait else spec.Basic.ConsumeOk,
  File "./venv/lib/python3.4/site-packages/amqp/abstract_channel.py", line 73, in send_method
    return self.wait(wait, returns_tuple=returns_tuple)
  File "./venv/lib/python3.4/site-packages/amqp/abstract_channel.py", line 93, in wait
    self.connection.drain_events(timeout=timeout)
  File "./venv/lib/python3.4/site-packages/amqp/connection.py", line 464, in drain_events
    return self.blocking_read(timeout)
  File "./venv/lib/python3.4/site-packages/amqp/connection.py", line 469, in blocking_read
    return self.on_inbound_frame(frame)
  File "./venv/lib/python3.4/site-packages/amqp/method_framing.py", line 88, in on_frame
    callback(channel, msg.frame_method, msg.frame_args, msg)
  File "./venv/lib/python3.4/site-packages/amqp/connection.py", line 473, in on_inbound_method
    method_sig, payload, content,
  File "./venv/lib/python3.4/site-packages/amqp/abstract_channel.py", line 142, in dispatch_method
    listener(*args)
  File "./venv/lib/python3.4/site-packages/amqp/channel.py", line 1613, in _on_basic_deliver
    fun(msg)
  File "./venv/lib/python3.4/site-packages/kombu/messaging.py", line 617, in _receive_callback
    return on_m(message) if on_m else self.receive(decoded, message)
  File "./venv/lib/python3.4/site-packages/celery/worker/consumer/consumer.py", line 558, in on_task_received
    callbacks,
  File "./venv/lib/python3.4/site-packages/celery/worker/strategy.py", line 145, in task_message_handler
    handle(req)
  File "./venv/lib/python3.4/site-packages/celery/worker/worker.py", line 221, in _process_task_sem
    return self._quick_acquire(self._process_task, req)
  File "./venv/lib/python3.4/site-packages/kombu/async/semaphore.py", line 62, in acquire
    callback(*partial_args, **partial_kwargs)
  File "./venv/lib/python3.4/site-packages/celery/worker/worker.py", line 226, in _process_task
    req.execute_using_pool(self.pool)
  File "./venv/lib/python3.4/site-packages/celery/worker/request.py", line 532, in execute_using_pool
    correlation_id=task_id,
  File "./venv/lib/python3.4/site-packages/celery/concurrency/base.py", line 155, in apply_async
    **options)
  File "./venv/lib/python3.4/site-packages/billiard/pool.py", line 1487, in apply_async
    self._quick_put((TASK, (result._job, None, func, args, kwds)))
TypeError: 'NoneType' object is not callable
```
The above lines are keep repeating every few seconds and no tasks are consumed from the queue.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `celery/worker/loops.py`
Content:
```
1 """The consumers highly-optimized inner loop."""
2 from __future__ import absolute_import, unicode_literals
3 import errno
4 import socket
5 from celery import bootsteps
6 from celery.exceptions import WorkerShutdown, WorkerTerminate, WorkerLostError
7 from celery.utils.log import get_logger
8 from . import state
9 
10 __all__ = ['asynloop', 'synloop']
11 
12 # pylint: disable=redefined-outer-name
13 # We cache globals and attribute lookups, so disable this warning.
14 
15 logger = get_logger(__name__)
16 
17 
18 def _quick_drain(connection, timeout=0.1):
19     try:
20         connection.drain_events(timeout=timeout)
21     except Exception as exc:  # pylint: disable=broad-except
22         exc_errno = getattr(exc, 'errno', None)
23         if exc_errno is not None and exc_errno != errno.EAGAIN:
24             raise
25 
26 
27 def _enable_amqheartbeats(timer, connection, rate=2.0):
28     if connection:
29         tick = connection.heartbeat_check
30         heartbeat = connection.get_heartbeat_interval()  # negotiated
31         if heartbeat and connection.supports_heartbeats:
32             timer.call_repeatedly(heartbeat / rate, tick, (rate,))
33 
34 
35 def asynloop(obj, connection, consumer, blueprint, hub, qos,
36              heartbeat, clock, hbrate=2.0):
37     """Non-blocking event loop."""
38     RUN = bootsteps.RUN
39     update_qos = qos.update
40     errors = connection.connection_errors
41 
42     on_task_received = obj.create_task_handler()
43 
44     _enable_amqheartbeats(hub.timer, connection, rate=hbrate)
45 
46     consumer.on_message = on_task_received
47     consumer.consume()
48     obj.on_ready()
49     obj.controller.register_with_event_loop(hub)
50     obj.register_with_event_loop(hub)
51 
52     # did_start_ok will verify that pool processes were able to start,
53     # but this will only work the first time we start, as
54     # maxtasksperchild will mess up metrics.
55     if not obj.restart_count and not obj.pool.did_start_ok():
56         raise WorkerLostError('Could not start worker processes')
57 
58     # consumer.consume() may have prefetched up to our
59     # limit - drain an event so we're in a clean state
60     # prior to starting our event loop.
61     if connection.transport.driver_type == 'amqp':
62         hub.call_soon(_quick_drain, connection)
63 
64     # FIXME: Use loop.run_forever
65     # Tried and works, but no time to test properly before release.
66     hub.propagate_errors = errors
67     loop = hub.create_loop()
68 
69     try:
70         while blueprint.state == RUN and obj.connection:
71             # shutdown if signal handlers told us to.
72             should_stop, should_terminate = (
73                 state.should_stop, state.should_terminate,
74             )
75             # False == EX_OK, so must use is not False
76             if should_stop is not None and should_stop is not False:
77                 raise WorkerShutdown(should_stop)
78             elif should_terminate is not None and should_stop is not False:
79                 raise WorkerTerminate(should_terminate)
80 
81             # We only update QoS when there's no more messages to read.
82             # This groups together qos calls, and makes sure that remote
83             # control commands will be prioritized over task messages.
84             if qos.prev != qos.value:
85                 update_qos()
86 
87             try:
88                 next(loop)
89             except StopIteration:
90                 loop = hub.create_loop()
91     finally:
92         try:
93             hub.reset()
94         except Exception as exc:  # pylint: disable=broad-except
95             logger.exception(
96                 'Error cleaning up after event loop: %r', exc)
97 
98 
99 def synloop(obj, connection, consumer, blueprint, hub, qos,
100             heartbeat, clock, hbrate=2.0, **kwargs):
101     """Fallback blocking event loop for transports that doesn't support AIO."""
102     RUN = bootsteps.RUN
103     on_task_received = obj.create_task_handler()
104     perform_pending_operations = obj.perform_pending_operations
105     if getattr(obj.pool, 'is_green', False):
106         _enable_amqheartbeats(obj.timer, connection, rate=hbrate)
107     consumer.on_message = on_task_received
108     consumer.consume()
109 
110     obj.on_ready()
111 
112     while blueprint.state == RUN and obj.connection:
113         state.maybe_shutdown()
114         if qos.prev != qos.value:
115             qos.update()
116         try:
117             perform_pending_operations()
118             connection.drain_events(timeout=2.0)
119         except socket.timeout:
120             pass
121         except socket.error:
122             if blueprint.state == RUN:
123                 raise
124
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/celery/worker/loops.py b/celery/worker/loops.py
--- a/celery/worker/loops.py
+++ b/celery/worker/loops.py
@@ -44,10 +44,10 @@
     _enable_amqheartbeats(hub.timer, connection, rate=hbrate)
 
     consumer.on_message = on_task_received
-    consumer.consume()
-    obj.on_ready()
     obj.controller.register_with_event_loop(hub)
     obj.register_with_event_loop(hub)
+    consumer.consume()
+    obj.on_ready()
 
     # did_start_ok will verify that pool processes were able to start,
     # but this will only work the first time we start, as

{"golden_diff": "diff --git a/celery/worker/loops.py b/celery/worker/loops.py\n--- a/celery/worker/loops.py\n+++ b/celery/worker/loops.py\n@@ -44,10 +44,10 @@\n     _enable_amqheartbeats(hub.timer, connection, rate=hbrate)\n \n     consumer.on_message = on_task_received\n-    consumer.consume()\n-    obj.on_ready()\n     obj.controller.register_with_event_loop(hub)\n     obj.register_with_event_loop(hub)\n+    consumer.consume()\n+    obj.on_ready()\n \n     # did_start_ok will verify that pool processes were able to start,\n     # but this will only work the first time we start, as\n", "issue": "Celery Worker crashing after first task with TypeError: 'NoneType' object is not callable\n## Checklist\r\n\r\n- [X] I have included the output of ``celery -A proj report`` in the issue.\r\n      (if you are not able to do this, then at least specify the Celery\r\n       version affected).\r\n```\r\nsoftware -> celery:4.0.0 (latentcall) kombu:4.0.0 py:3.4.3\r\n            billiard:3.5.0.2 py-amqp:2.1.1\r\nplatform -> system:Linux arch:64bit, ELF imp:CPython\r\nloader   -> celery.loaders.default.Loader\r\nsettings -> transport:amqp results:disabled\r\n```\r\n- [X] I have verified that the issue exists against the `master` branch of Celery.\r\nYes I've tested and it behaves the same using master.\r\n\r\n## Steps to reproduce\r\nNot exactly sure, because other machines with the same specs and requirements are working.\r\n\r\n## Expected behavior\r\nShould consume tasks.\r\n\r\n## Actual behavior\r\nA task is accepted, then a traceback is logged, then the worker reconnects to the broker for some reason. This repeats forever:\r\n\r\n```\r\n[2016-11-23 23:09:00,468: INFO/MainProcess] Connected to amqp://user:**@10.136.131.6:5672//\r\n[2016-11-23 23:09:00,484: INFO/MainProcess] mingle: searching for neighbors\r\n[2016-11-23 23:09:01,921: INFO/MainProcess] mingle: sync with 1 nodes\r\n[2016-11-23 23:09:01,922: INFO/MainProcess] mingle: sync complete\r\n[2016-11-23 23:09:01,970: INFO/MainProcess] Received task: tasks.calculate_user_running_total[ddd103af-d527-4564-83f8-96b747767a0c]\r\n[2016-11-23 23:09:01,972: CRITICAL/MainProcess] Unrecoverable error: TypeError(\"'NoneType' object is not callable\",)\r\nTraceback (most recent call last):\r\n  File \"./venv/lib/python3.4/site-packages/celery/worker/worker.py\", line 203, in start\r\n    self.blueprint.start(self)\r\n  File \"./venv/lib/python3.4/site-packages/celery/bootsteps.py\", line 119, in start\r\n    step.start(parent)\r\n  File \"./venv/lib/python3.4/site-packages/celery/bootsteps.py\", line 370, in start\r\n    return self.obj.start()\r\n  File \"./venv/lib/python3.4/site-packages/celery/worker/consumer/consumer.py\", line 318, in start\r\n    blueprint.start(self)\r\n  File \"./venv/lib/python3.4/site-packages/celery/bootsteps.py\", line 119, in start\r\n    step.start(parent)\r\n  File \"./venv/lib/python3.4/site-packages/celery/worker/consumer/consumer.py\", line 584, in start\r\n    c.loop(*c.loop_args())\r\n  File \"./venv/lib/python3.4/site-packages/celery/worker/loops.py\", line 47, in asynloop\r\n    consumer.consume()\r\n  File \"./venv/lib/python3.4/site-packages/kombu/messaging.py\", line 470, in consume\r\n    self._basic_consume(T, no_ack=no_ack, nowait=False)\r\n  File \"./venv/lib/python3.4/site-packages/kombu/messaging.py\", line 591, in _basic_consume\r\n    no_ack=no_ack, nowait=nowait)\r\n  File \"./venv/lib/python3.4/site-packages/kombu/entity.py\", line 737, in consume\r\n    arguments=self.consumer_arguments)\r\n  File \"./venv/lib/python3.4/site-packages/amqp/channel.py\", line 1578, in basic_consume\r\n    wait=None if nowait else spec.Basic.ConsumeOk,\r\n  File \"./venv/lib/python3.4/site-packages/amqp/abstract_channel.py\", line 73, in send_method\r\n    return self.wait(wait, returns_tuple=returns_tuple)\r\n  File \"./venv/lib/python3.4/site-packages/amqp/abstract_channel.py\", line 93, in wait\r\n    self.connection.drain_events(timeout=timeout)\r\n  File \"./venv/lib/python3.4/site-packages/amqp/connection.py\", line 464, in drain_events\r\n    return self.blocking_read(timeout)\r\n  File \"./venv/lib/python3.4/site-packages/amqp/connection.py\", line 469, in blocking_read\r\n    return self.on_inbound_frame(frame)\r\n  File \"./venv/lib/python3.4/site-packages/amqp/method_framing.py\", line 88, in on_frame\r\n    callback(channel, msg.frame_method, msg.frame_args, msg)\r\n  File \"./venv/lib/python3.4/site-packages/amqp/connection.py\", line 473, in on_inbound_method\r\n    method_sig, payload, content,\r\n  File \"./venv/lib/python3.4/site-packages/amqp/abstract_channel.py\", line 142, in dispatch_method\r\n    listener(*args)\r\n  File \"./venv/lib/python3.4/site-packages/amqp/channel.py\", line 1613, in _on_basic_deliver\r\n    fun(msg)\r\n  File \"./venv/lib/python3.4/site-packages/kombu/messaging.py\", line 617, in _receive_callback\r\n    return on_m(message) if on_m else self.receive(decoded, message)\r\n  File \"./venv/lib/python3.4/site-packages/celery/worker/consumer/consumer.py\", line 558, in on_task_received\r\n    callbacks,\r\n  File \"./venv/lib/python3.4/site-packages/celery/worker/strategy.py\", line 145, in task_message_handler\r\n    handle(req)\r\n  File \"./venv/lib/python3.4/site-packages/celery/worker/worker.py\", line 221, in _process_task_sem\r\n    return self._quick_acquire(self._process_task, req)\r\n  File \"./venv/lib/python3.4/site-packages/kombu/async/semaphore.py\", line 62, in acquire\r\n    callback(*partial_args, **partial_kwargs)\r\n  File \"./venv/lib/python3.4/site-packages/celery/worker/worker.py\", line 226, in _process_task\r\n    req.execute_using_pool(self.pool)\r\n  File \"./venv/lib/python3.4/site-packages/celery/worker/request.py\", line 532, in execute_using_pool\r\n    correlation_id=task_id,\r\n  File \"./venv/lib/python3.4/site-packages/celery/concurrency/base.py\", line 155, in apply_async\r\n    **options)\r\n  File \"./venv/lib/python3.4/site-packages/billiard/pool.py\", line 1487, in apply_async\r\n    self._quick_put((TASK, (result._job, None, func, args, kwds)))\r\nTypeError: 'NoneType' object is not callable\r\n```\r\nThe above lines are keep repeating every few seconds and no tasks are consumed from the queue.\n", "before_files": [{"content": "\"\"\"The consumers highly-optimized inner loop.\"\"\"\nfrom __future__ import absolute_import, unicode_literals\nimport errno\nimport socket\nfrom celery import bootsteps\nfrom celery.exceptions import WorkerShutdown, WorkerTerminate, WorkerLostError\nfrom celery.utils.log import get_logger\nfrom . import state\n\n__all__ = ['asynloop', 'synloop']\n\n# pylint: disable=redefined-outer-name\n# We cache globals and attribute lookups, so disable this warning.\n\nlogger = get_logger(__name__)\n\n\ndef _quick_drain(connection, timeout=0.1):\n    try:\n        connection.drain_events(timeout=timeout)\n    except Exception as exc:  # pylint: disable=broad-except\n        exc_errno = getattr(exc, 'errno', None)\n        if exc_errno is not None and exc_errno != errno.EAGAIN:\n            raise\n\n\ndef _enable_amqheartbeats(timer, connection, rate=2.0):\n    if connection:\n        tick = connection.heartbeat_check\n        heartbeat = connection.get_heartbeat_interval()  # negotiated\n        if heartbeat and connection.supports_heartbeats:\n            timer.call_repeatedly(heartbeat / rate, tick, (rate,))\n\n\ndef asynloop(obj, connection, consumer, blueprint, hub, qos,\n             heartbeat, clock, hbrate=2.0):\n    \"\"\"Non-blocking event loop.\"\"\"\n    RUN = bootsteps.RUN\n    update_qos = qos.update\n    errors = connection.connection_errors\n\n    on_task_received = obj.create_task_handler()\n\n    _enable_amqheartbeats(hub.timer, connection, rate=hbrate)\n\n    consumer.on_message = on_task_received\n    consumer.consume()\n    obj.on_ready()\n    obj.controller.register_with_event_loop(hub)\n    obj.register_with_event_loop(hub)\n\n    # did_start_ok will verify that pool processes were able to start,\n    # but this will only work the first time we start, as\n    # maxtasksperchild will mess up metrics.\n    if not obj.restart_count and not obj.pool.did_start_ok():\n        raise WorkerLostError('Could not start worker processes')\n\n    # consumer.consume() may have prefetched up to our\n    # limit - drain an event so we're in a clean state\n    # prior to starting our event loop.\n    if connection.transport.driver_type == 'amqp':\n        hub.call_soon(_quick_drain, connection)\n\n    # FIXME: Use loop.run_forever\n    # Tried and works, but no time to test properly before release.\n    hub.propagate_errors = errors\n    loop = hub.create_loop()\n\n    try:\n        while blueprint.state == RUN and obj.connection:\n            # shutdown if signal handlers told us to.\n            should_stop, should_terminate = (\n                state.should_stop, state.should_terminate,\n            )\n            # False == EX_OK, so must use is not False\n            if should_stop is not None and should_stop is not False:\n                raise WorkerShutdown(should_stop)\n            elif should_terminate is not None and should_stop is not False:\n                raise WorkerTerminate(should_terminate)\n\n            # We only update QoS when there's no more messages to read.\n            # This groups together qos calls, and makes sure that remote\n            # control commands will be prioritized over task messages.\n            if qos.prev != qos.value:\n                update_qos()\n\n            try:\n                next(loop)\n            except StopIteration:\n                loop = hub.create_loop()\n    finally:\n        try:\n            hub.reset()\n        except Exception as exc:  # pylint: disable=broad-except\n            logger.exception(\n                'Error cleaning up after event loop: %r', exc)\n\n\ndef synloop(obj, connection, consumer, blueprint, hub, qos,\n            heartbeat, clock, hbrate=2.0, **kwargs):\n    \"\"\"Fallback blocking event loop for transports that doesn't support AIO.\"\"\"\n    RUN = bootsteps.RUN\n    on_task_received = obj.create_task_handler()\n    perform_pending_operations = obj.perform_pending_operations\n    if getattr(obj.pool, 'is_green', False):\n        _enable_amqheartbeats(obj.timer, connection, rate=hbrate)\n    consumer.on_message = on_task_received\n    consumer.consume()\n\n    obj.on_ready()\n\n    while blueprint.state == RUN and obj.connection:\n        state.maybe_shutdown()\n        if qos.prev != qos.value:\n            qos.update()\n        try:\n            perform_pending_operations()\n            connection.drain_events(timeout=2.0)\n        except socket.timeout:\n            pass\n        except socket.error:\n            if blueprint.state == RUN:\n                raise\n", "path": "celery/worker/loops.py"}], "after_files": [{"content": "\"\"\"The consumers highly-optimized inner loop.\"\"\"\nfrom __future__ import absolute_import, unicode_literals\nimport errno\nimport socket\nfrom celery import bootsteps\nfrom celery.exceptions import WorkerShutdown, WorkerTerminate, WorkerLostError\nfrom celery.utils.log import get_logger\nfrom . import state\n\n__all__ = ['asynloop', 'synloop']\n\n# pylint: disable=redefined-outer-name\n# We cache globals and attribute lookups, so disable this warning.\n\nlogger = get_logger(__name__)\n\n\ndef _quick_drain(connection, timeout=0.1):\n    try:\n        connection.drain_events(timeout=timeout)\n    except Exception as exc:  # pylint: disable=broad-except\n        exc_errno = getattr(exc, 'errno', None)\n        if exc_errno is not None and exc_errno != errno.EAGAIN:\n            raise\n\n\ndef _enable_amqheartbeats(timer, connection, rate=2.0):\n    if connection:\n        tick = connection.heartbeat_check\n        heartbeat = connection.get_heartbeat_interval()  # negotiated\n        if heartbeat and connection.supports_heartbeats:\n            timer.call_repeatedly(heartbeat / rate, tick, (rate,))\n\n\ndef asynloop(obj, connection, consumer, blueprint, hub, qos,\n             heartbeat, clock, hbrate=2.0):\n    \"\"\"Non-blocking event loop.\"\"\"\n    RUN = bootsteps.RUN\n    update_qos = qos.update\n    errors = connection.connection_errors\n\n    on_task_received = obj.create_task_handler()\n\n    _enable_amqheartbeats(hub.timer, connection, rate=hbrate)\n\n    consumer.on_message = on_task_received\n    obj.controller.register_with_event_loop(hub)\n    obj.register_with_event_loop(hub)\n    consumer.consume()\n    obj.on_ready()\n\n    # did_start_ok will verify that pool processes were able to start,\n    # but this will only work the first time we start, as\n    # maxtasksperchild will mess up metrics.\n    if not obj.restart_count and not obj.pool.did_start_ok():\n        raise WorkerLostError('Could not start worker processes')\n\n    # consumer.consume() may have prefetched up to our\n    # limit - drain an event so we're in a clean state\n    # prior to starting our event loop.\n    if connection.transport.driver_type == 'amqp':\n        hub.call_soon(_quick_drain, connection)\n\n    # FIXME: Use loop.run_forever\n    # Tried and works, but no time to test properly before release.\n    hub.propagate_errors = errors\n    loop = hub.create_loop()\n\n    try:\n        while blueprint.state == RUN and obj.connection:\n            # shutdown if signal handlers told us to.\n            should_stop, should_terminate = (\n                state.should_stop, state.should_terminate,\n            )\n            # False == EX_OK, so must use is not False\n            if should_stop is not None and should_stop is not False:\n                raise WorkerShutdown(should_stop)\n            elif should_terminate is not None and should_stop is not False:\n                raise WorkerTerminate(should_terminate)\n\n            # We only update QoS when there's no more messages to read.\n            # This groups together qos calls, and makes sure that remote\n            # control commands will be prioritized over task messages.\n            if qos.prev != qos.value:\n                update_qos()\n\n            try:\n                next(loop)\n            except StopIteration:\n                loop = hub.create_loop()\n    finally:\n        try:\n            hub.reset()\n        except Exception as exc:  # pylint: disable=broad-except\n            logger.exception(\n                'Error cleaning up after event loop: %r', exc)\n\n\ndef synloop(obj, connection, consumer, blueprint, hub, qos,\n            heartbeat, clock, hbrate=2.0, **kwargs):\n    \"\"\"Fallback blocking event loop for transports that doesn't support AIO.\"\"\"\n    RUN = bootsteps.RUN\n    on_task_received = obj.create_task_handler()\n    perform_pending_operations = obj.perform_pending_operations\n    if getattr(obj.pool, 'is_green', False):\n        _enable_amqheartbeats(obj.timer, connection, rate=hbrate)\n    consumer.on_message = on_task_received\n    consumer.consume()\n\n    obj.on_ready()\n\n    while blueprint.state == RUN and obj.connection:\n        state.maybe_shutdown()\n        if qos.prev != qos.value:\n            qos.update()\n        try:\n            perform_pending_operations()\n            connection.drain_events(timeout=2.0)\n        except socket.timeout:\n            pass\n        except socket.error:\n            if blueprint.state == RUN:\n                raise\n", "path": "celery/worker/loops.py"}]}