SCALRCORE-8536 (FAM-328) Enable heartbeats on gevent

celery/worker/consumer/consumer.py

@@ -158,6 +158,9 @@ class Blueprint(bootsteps.Blueprint):
         """Consumer blueprint."""
 
         name = 'Consumer'
+        # [FAM-328] We remove Events, Gossip, Heartbeat and Mingle from default
+        # steps because they are not used by us but interfere with amqp
+        # heartbeats we need.
         default_steps = [
             'celery.worker.consumer.connection:Connection',
             'celery.worker.consumer.mingle:Mingle',
@@ -206,8 +209,10 @@ def __init__(self, on_task_request,
         self.task_buckets = defaultdict(lambda: None)
         self.reset_rate_limits()
 
+        self.gevent_env = _detect_environment() == 'gevent'
+
         self.hub = hub
-        if self.hub or getattr(self.pool, 'is_green', False):
+        if self.hub or getattr(self.pool, 'is_green', False) or self.gevent_env:
             self.amqheartbeat = amqheartbeat
             if self.amqheartbeat is None:
                 self.amqheartbeat = self.app.conf.broker_heartbeat
@@ -217,7 +222,7 @@ def __init__(self, on_task_request,
         if not hasattr(self, 'loop'):
             self.loop = loops.asynloop if hub else loops.synloop
 
-        if _detect_environment() == 'gevent':
+        if self.gevent_env:
             # there's a gevent bug that causes timeouts to not be reset,
             # so if the connection timeout is exceeded once, it can NEVER
             # connect again.
@@ -428,6 +433,13 @@ def connect(self):
         conn = self.connection_for_read(heartbeat=self.amqheartbeat)
         if self.hub:
             conn.transport.register_with_event_loop(conn.connection, self.hub)
+
+        # Install timer to send heartbeats on gevent
+        if self.gevent_env and self.amqheartbeat:
+            hbtick = conn.heartbeat_check
+            logger.debug('Registring heartbeat timer for connection: {}'.format(id(conn)))
+            self.timer.call_repeatedly(int(self.amqheartbeat / self.amqheartbeat_rate), hbtick, (self.amqheartbeat_rate,))
+
         return conn
 
     def connection_for_read(self, heartbeat=None):

celery/worker/consumer/tasks.py

@@ -4,7 +4,7 @@
 from celery import bootsteps
 from celery.utils.log import get_logger
 
-from .mingle import Mingle
+from .connection import Connection
 
 __all__ = ('Tasks',)
 
@@ -15,7 +15,7 @@
 class Tasks(bootsteps.StartStopStep):
     """Bootstep starting the task message consumer."""
 
-    requires = (Mingle,)
+    requires = (Connection,)
 
     def __init__(self, c, **kwargs):
         c.task_consumer = c.qos = None