This document describes the current stable version of Celery (5.2). For development docs, go here.
Source code for celery.worker.consumer.mingle
"""Worker <-> Worker Sync at startup (Bootstep)."""
from celery import bootsteps
from celery.utils.log import get_logger
from .events import Events
__all__ = ('Mingle',)
logger = get_logger(__name__)
debug, info, exception = logger.debug, logger.info, logger.exception
[docs]class Mingle(bootsteps.StartStopStep):
"""Bootstep syncing state with neighbor workers.
At startup, or upon consumer restart, this will:
- Sync logical clocks.
- Sync revoked tasks.
"""
label = 'Mingle'
requires = (Events,)
compatible_transports = {'amqp', 'redis'}
def __init__(self, c, without_mingle=False, **kwargs):
self.enabled = not without_mingle and self.compatible_transport(c.app)
super().__init__(
c, without_mingle=without_mingle, **kwargs)
[docs] def compatible_transport(self, app):
with app.connection_for_read() as conn:
return conn.transport.driver_type in self.compatible_transports
[docs] def sync(self, c):
info('mingle: searching for neighbors')
replies = self.send_hello(c)
if replies:
info('mingle: sync with %s nodes',
len([reply for reply, value in replies.items() if value]))
[self.on_node_reply(c, nodename, reply)
for nodename, reply in replies.items() if reply]
info('mingle: sync complete')
else:
info('mingle: all alone')
[docs] def send_hello(self, c):
inspect = c.app.control.inspect(timeout=1.0, connection=c.connection)
our_revoked = c.controller.state.revoked
replies = inspect.hello(c.hostname, our_revoked._data) or {}
replies.pop(c.hostname, None) # delete my own response
return replies
[docs] def on_node_reply(self, c, nodename, reply):
debug('mingle: processing reply from %s', nodename)
try:
self.sync_with_node(c, **reply)
except MemoryError:
raise
except Exception as exc: # pylint: disable=broad-except
exception('mingle: sync with %s failed: %r', nodename, exc)
[docs] def sync_with_node(self, c, clock=None, revoked=None, **kwargs):
self.on_clock_event(c, clock)
self.on_revoked_received(c, revoked)
[docs] def on_clock_event(self, c, clock):
c.app.clock.adjust(clock) if clock else c.app.clock.forward()
[docs] def on_revoked_received(self, c, revoked):
if revoked:
c.controller.state.revoked.update(revoked)
