celery-celery-10218-10204

__all__ = ('asynloop', 'synloop')

# pylint: disable=redefined-outer-name
# We cache globals and attribute lookups, so disable this warning.

logger = get_logger(__name__)


def _quick_drain(connection, timeout=0.1):
    try:
        connection.drain_events(timeout=timeout)
    except Exception as exc:  # pylint: disable=broad-except
        exc_errno = getattr(exc, 'errno', None)
        if exc_errno is not None and exc_errno != errno.EAGAIN:
            raise


def _enable_amqheartbeats(timer, connection, rate=2.0):
    heartbeat_error = [None]

    if not connection:
        return heartbeat_error

    heartbeat = connection.get_heartbeat_interval()  # ne
…[3744 chars omitted]…
           qos.update()
        try:
            perform_pending_operations()
            connection.drain_events(timeout=2.0)
        except socket.timeout:
            pass
        except OSError:
            if blueprint.state == RUN:
                raise

    while blueprint.state == RUN and obj.connection:
        try:
            state.maybe_shutdown()
        finally:
            _loop_cycle()

    13	# pylint: disable=redefined-outer-name
    14	# We cache globals and attribute lookups, so disable this warning.
    15	
    16	logger = get_logger(__name__)
    17	
    18	
    19	def _quick_drain(connection, timeout=0.1):
    20	    try:
    21	        connection.drain_events(timeout=timeout)
    22	    except Exception as exc:  # pylint: disable=broad-except
    23	        exc_errno = getattr(exc, 'errno', None)
    24	        if exc_errno is not None and exc_errno != errno.EAGAIN:
    25	            raise
    26	
    27	
    28	def _enable_amqhear
…[8492 chars omitted]…
    x.hub.poller.poll.return_value = [(6, WRITE)]
391:        x.hub.remove_writer = Mock(name='hub.remove_writer()')
393:            asynloop(*x.args)
403:        x.hub.add_writer(6, gen)
404:        x.hub.remove = Mock(name='hub.remove()')
405:        x.hub.on_tick.add(x.close_then_error(Mock(name='tick'), 2))
406:        x.hub.poller.poll.return_value = [(6, WRITE)]
408:            asynloop(*x.args)

from celery.worker import state
from celery.worker.consumer import Consumer
from celery.worker.loops import _quick_drain, asynloop, synloop


class PromiseEqual:

    def __init__(self, fun, *args, **kwargs):
        self.fun = fun
        self.args = args
        self.kwargs = kwargs

    def __eq__(self, other):
        return (other.fun == self.fun and
                other.args == self.args and
                other.kwargs == self.kwargs)

    def __repr__(self):
        return '<promise: {0.fun!r} {0.args!r} {0
…[8629 chars omitted]…
[1]
        x.hub.reset = Mock(name='hub.reset()')
        x.close_then_error(x.hub.poller.poll)
        x.hub.fire_timers.return_value = 33.37
        poller = x.hub.poller
        poller.poll.return_value = []
        with pytest.raises(socket.error):
            asynloop(*x.args)
        x.hub.reset.assert_called_once()

    def test_hub_not_reset_on_graceful_shutdown(self):
        x = X(self.app)

    def test_on_task_message_missing_name(self):
        x, on_task, msg, strategy = self.task_context(self.add.s(2, 2))
        msg.headers.pop('task')
        on_task(msg)
        x.on_unknown_message.assert_called_with(msg.decode(), msg)

    def test_on_task_pool_raises(self):
        x, on_task, msg, strategy = self.task_context(self.add.s(2, 2))
        strategy.side_effect = ValueError()
        with pytest.raises(ValueEr
…[8629 chars omitted]…
untimeError from reset()
        with pytest.raises(socket.error):
            asynloop(*x.args)
        x.hub.reset.assert_called_once()


class test_synloop:

    def test_timeout_ignored(self):
        x = X(self.app)
        x.timeout_then_error(x.connection.drain_events)
        with pytest.raises(socket.error):
            synloop(*x.args)
        assert x.connection.drain_events.call_count == 2

celery/worker/consumer/consumer.py:569:            cset.add_queue(q)
celery/worker/consumer/consumer.py:573:    def cancel_task_queue(self, queue):
celery/worker/consumer/consumer.py:576:        self.ta
…[4329 chars omitted]…
/unit/worker/test_loops.py:67:        self.hub.timer._queue = [Mock()]
5.7.0
    def reset(self):
        # Unlike close(), keeps the waker: closing it could race with
        # another thread's write.
        self._close()
        self._create_poller()

    def close(self, *args):
        self._close()
        if self._waker is not None:
            self._waker.close()
            self._waker = None

    WRITE = WRITE

    #: Flag set on error, and the fd should be read from asap.
    ERR = ERR

    #: List of callbacks to be called when the loop is exiting,
    #: applied with the hub instance as sole argument.
    on_close = None

    def __init__(self, timer=None):
        self.timer = timer if timer is not None else Timer()

        self.readers = {}
        self.writers = {}
        self.on_tick = set()
        self.on_close = set()
        self._ready = set()
        self._ready_lock = threading.Lock()

        # Lets call_soon() from another thread interrupt the poller,
        # see _create_waker().
      
…[8629 chars omitted]…
   from .debug import repr_active
        return repr_active(self)

    def repr_events(self, events):
        from .debug import repr_events
        return repr_events(self, events or [])

    @cached_property
    def scheduler(self):
        return iter(self.timer)

    @property
    def loop(self):
        if self._loop is None:
            self._loop = self.create_loop()
        return self._loop

    def __enter__(self):
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None
    ) -> None:
        self.stop()

    def call_at(self, eta, fun, args=(), kwargs=None, priority=0):
        kwargs = {} if not kwargs else kwargs
        return self.enter_at(self.Entry(fun, args, kwargs), eta, priority)

    def call_after(self, secs, fun, args=(), kwargs=None, priority=0):
        kwargs = {} if not kwargs else kwargs
        return self.enter_after(secs, self.Entr
…[3303 chars omitted]…
ancel(self, tref):
        tref.cancel()

    def __len__(self):
        return len(self._queue)

    def __nonzero__(self):
        return True

    @property
    def queue(self, _pop=heapq.heappop):
        """Snapshot of underlying datastructure."""
        events = list(self._queue)
        return [_pop(v) for v in [events] * len(events)]

    @property
    def schedule(self):
        return self

celery
celery.egg-info
docker
docs
examples
extra
helm-chart
pyproject.toml
requirements
setup.cfg
setup.py
t
tox.ini

class test_synloop:

    def test_timeout_ignored(self):
        x = X(self.app)
        x.timeout_then_error(x.connection.drain_events)
        with pytest.raises(socket.error):
            synloop(*x.args)
        assert x.connection.drain_events.call_count == 2

    def test_updates_qos_when_changed(self):
        x = X(self.app)
        x.qos.prev = 2
        x.q
…[2150 chars omitted]…
   exc.errno = 313
        self.connection.drain_events.side_effect = exc
        with pytest.raises(KeyError):
            _quick_drain(self.connection, timeout=33.3)

    def test_drain_error_EAGAIN(self):
        exc = KeyError()
        exc.errno = errno.EAGAIN
        self.connection.drain_events.side_effect = exc
        _quick_drain(self.connection, timeout=33.3)
606 t/unit/worker/test_loops.py

t/unit/worker/test_loops.py:534:            synloop(*x.args)
t/unit/worker/test_loops.py:540:        assert synloop(*x.args) is None
t/unit/worker/test_loops.py:549:        synloop(*x.args)
t/unit/worker/test_loops.py:568:            synloop(*x.args)
t/unit/worker/test_loops.py:581:        synloop(*x.args)
t/unit/worker/test_worker.py:15:from kombu.asynchronous import get_event_loop

from celery.platforms import IS_WINDOWS
from celery.utils.log import worker_logger as logger

__all__ = ('Timer', 'Hub', 'Pool', 'Beat', 'StateDB', 'Consumer')

GREEN_POOLS = {'eventlet', 'gevent'}

ERR_B_GREEN = """\
-B option doesn't work with eventlet/gevent pools: \
use standalone beat instead.\
"""

W_POOL_SETTING = """
The worker_pool setting shouldn't be used to select the eventlet/gevent
pools, instead you *must use the -P* argument so that patches are applied

…[4667 chars omitted]…
g_operations was called even after SystemExit
    50	    obj.perform_pending_operations.assert_called_once()
    51	
    52	    # Assert that connection.drain_events was called
    53	    connection.drain_events.assert_called_with(timeout=2.0)
    54	
    55	    # Assert other important method calls
    56	    obj.on_ready.assert_called_once()
    57	    consumer.consume.assert_called_once()
---git---

        self._restart_state = restart_state(maxR=5, maxT=1)

        self._does_info = logger.isEnabledFor(logging.INFO)
        self._limit_order = 0
        self.on_task_request = on_task_request
        self.on_task_message = set()
        self.amqheartbeat_rate = self.app.conf.broker_heartbeat_checkrate
        self.disable_rate_limits = disable_rate_limi
…[2042 chars omitted]…
)
369:                        self.on_connection_error_before_connected(exc)
379:    def on_connection_error_before_connected(self, exc):
383:    def on_connection_error_after_connected(self, exc):
593:        message.reject_log_error(logger, self.connection_errors)
616:        message.reject_log_error(logger, self.connection_errors)
632:        message.reject_log_error(logger, self.connection_errors)

                pri = self._limit_order = (self._limit_order + 1) % 10
                hold = bucket.expected_time(tokens)
                self.timer.call_after(
                    hold, self._schedule_bucket_request, (bucket,),
                    priority=pri,
                )
                # no tokens, break
                break

    def _limit_task(self, request, bucket, tokens):
        bucket.add((request, tokens))
        return self._schedule_bucket_request(bucket)

    def _limit_post_eta(se
…[3319 chars omitted]…
cknowledged:
                    warn(TERMINATING_TASK_ON_RESTART_AFTER_A_CONNECTION_LOSS,
                         request)
                    request.cancel(self.pool)
        else:
            warnings.warn(CANCEL_TASKS_BY_DEFAULT, CPendingDeprecationWarning)

        if self.app.conf.worker_enable_prefetch_count_reduction:
            self.initial_prefetch_count = max(
453:    def on_close(self):

                not self.app.IS_WINDOWS)

    def stop(self, in_sighandler=False, exitcode=None):
        """Graceful shutdown of the worker server (Warm shutdown)."""
        if exitcode is not None:
            self.exitcode = exitcode
        if self.blueprint.state == RUN:
            self.signal_consumer_close()
            if not in_sighandler or self.pool.signal_safe:
                self._shutdown(warm=True)
        self._send_worker_shutdown()

    def terminate(self, in_sighandler=False):
        """Not so graceful shutdown of the worker server (Cold shutdown)."""

                    f"The prefetch count will be gradually restored 
…[2485 chars omitted]…
     return self.ensure_connected(
            self.app.connection_for_read(heartbeat=heartbeat))

    def connection_for_write(self, url=None, heartbeat=None):
        return self.ensure_connected(
            self.app.connection_for_write(url=url, heartbeat=heartbeat))

    def ensure_connected(self, conn):
        # Callback called for each retry while the connection
        # can't be established.

233-            hostname=w.hostname,
234-            task_events=w.task_events,
235-            init_callback=w.ready_callback,
236-            initial_prefetch_count=prefetch_count,
237-            pool=w.pool,
238-            timer=w.timer,
239-            app=w.app,
240-            controller=w,
241-            hub=w.hub,
242-            worker_options=w.options,
243-            disable_rate_limits=w.disable_rate_limits,
244-            prefetch_multiplier=w.prefetch_multiplier,
245-        )
246-        return c

celery/concurrency/asynpool.py:894:            hub.remove(fd)
celery/concurrency/asynpool.py:942:                hub.remove_writer(fd)
celery/concurrency/asynpool.py:1334:                        self.hub_remove(sock)
celery/concurrency/prefork.py:148:            # the pool drains. If not using an async transport, no hub exists
celery/concurrency/prefork.py:150:            hub = get_event_loop()

    def did_start_ok(self):
        return self._pool.did_start_ok()

    def register_with_event_loop(self, loop):
        try:
            reg = self._pool.register_with_event_loop
        except AttributeError:
            return
        return reg(loop)

    def on_stop(self):
        """Gracefully stop the pool."""
        if self._pool is not None and self._pool._state in (RUN, CLOSE):
            self._pool.close()

            # Keep firing timers (for heartbeats on async transports) while
            # 
…[769 chars omitted]…
r the pool join is complete and sleeping long enough to
                        # avoid excessive CPU usage.
                        time.sleep(0.5)

                timer_thread = threading.Thread(
                    target=fire_timers_loop,
                    daemon=True,
                    name="prefork-timer-shutdown",
                )
                timer_thread.start()

                try:

# pylint: disable=redefined-outer-name
# We cache globals and attribute lookups, so disable this warning.

logger = get_logger(__name__)


def _quick_drain(connection, timeout=0.1):
    try:
        connection.drain_events(timeout=timeout)
    except Exception as exc:  # pylint: disable=broad-except
        exc_errno = getattr(exc, 'errno', None)
        if exc_errno is not None and exc_errno != errno.EAGAIN:
            raise


def _cleanup_hub(hub):
    """Clean up the hub after a connection error.

    The hub may be left in an inconsistent state after the connection is
    lost, and timers registered for the old connection (such as the

…[5273 chars omitted]…
     try:
                state.maybe_shutdown()
            finally:
                _loop_cycle()
    except Exception as exc:
        # Only clean up on connection errors.  A normal exit (or a graceful
        # shutdown) must leave the timer alone so it can keep firing while
        # the pool drains.
        if isinstance(exc, tuple(connection_errors)):
            _cleanup_hub(hub)
        raise

drwxr-xr-x   2 root root 4096 Aug  3 00:00 mnt
drwxr-xr-x   2 root root 4096 Aug  3 00:00 opt
dr-xr-xr-x 320 root root    0 Nov  4 21:33 proc
drwxr-xr-x   1 root root 4096 Nov  4 21:35 repo
drwx------   1 root root 4096 Nov  4 21:34 root
drwxr-xr-x   1 root root 4096 Nov  4 21:34 run
lrwxrwxrwx   1 root root    8 Jul  4 09:0
…[292 chars omitted]…
v  4 21:36 .
drwxr-xr-x 1 root root 4096 Nov  4 21:33 ..
-rw-r--r-- 1 root root    0 Nov  4 21:34 install.log
total 24
drwx------ 1 root root 4096 Nov  4 21:34 .
drwxr-xr-x 1 root root 4096 Nov  4 21:33 ..
-rw-r--r-- 1 root root  607 Jul  4 09:05 .bashrc
drwxr-xr-x 3 root root 4096 Nov  4 21:34 .cache
-rw-r--r-- 1 root root  132 Jul  4 09:05 .profile
-rw-r--r-- 1 root root  169 Aug  5 01:04 .wget-hsts

             ^
    File "/repo/celery/app/trace.py", line 260, in handle_failure
      task.backend.mark_as_failure(
    File "/usr/local/lib/python3.12/unittest/mock.py", line 1139, in __call__
      return self._mock_call(*args, **kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    File "/usr/local/lib/pyth
…[39 chars omitted]…
 _mock_call
      return self._execute_mock_call(*args, **kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    File "/usr/local/lib/python3.12/unittest/mock.py", line 1198, in _execute_mock_call
      raise effect
  Exception
  
    warn(RuntimeWarning(

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
570 passed, 1 skipped, 24 warnings, 37 subtests passed in 24.04s

+    lost, and timers registered for the old connection (such as the
+    heartbeat timer) must not fire while the worker is reconnecting.
+
+    Cleanup errors are logged and suppressed so that the original
+    connection error is always propagated to the caller.
+    """
+    if hub is None:
+        return
+
+    try:
+        hub.reset()
+    except Exception as exc:  # pylint: disable=broad-except
+        logger.exception(
+            'Error cleaning up after event loop: %r', exc)
+
+    # Clear the
…[1748 chars omitted]…
+                state.maybe_shutdown()
+            finally:
+                _loop_cycle()
+    except Exception as exc:
+        # Only clean up on connection errors.  A normal exit (or a graceful
+        # shutdown) must leave the timer alone so it can keep firing while
+        # the pool drains.
+        if isinstance(exc, tuple(connection_errors)):
+            _cleanup_hub(hub)
+        raise

# pylint: disable=redefined-outer-name
# We cache globals and attribute lookups, so disable this warning.

logger = get_logger(__name__)


def _quick_drain(connection, timeout=0.1):
    try:
        connection.drain_events(timeout=timeout)
    except Exception as exc:  # pylint: disable=broad-except
        exc_errno = getattr(exc, 'errno', None)
        if exc_errno is not None and exc_errno != errno.EAGAIN:
            raise


def _reset_hub(hub):
    """Reset the hub, suppressing and logging any cleanup error."""
    if hub is None:
        return
    try:
        hub.reset()
    except Exception as exc:  # pylint: disable=broad-except
 
…[8628 chars omitted]…
+                state.maybe_shutdown()
+            finally:
+                _loop_cycle()
+    except Exception as exc:
+        # Only clean up on connection errors.  A normal exit (or a graceful
+        # shutdown) must leave the timer alone so it can keep firing while
+        # the pool drains.
+        if isinstance(exc, tuple(connection_errors)):
+            _cleanup_hub(hub)
+        raise

ERROR t/unit/backends/test_s3.py
ERROR t/unit/security/test_certificate.py
ERROR t/unit/security/test_key.py
ERROR t/unit/security/test_security.py
ERROR t/unit/security/test_serialization.py
ERROR t/unit/utils/test_annotations.py
!!!!!!!!!!!!!!!!!!! Interrupted: 14 errors during collection !!!!!!!!!!!!!!!!!!!
9 skipped, 14 errors in 8.61s

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
88 passed, 18 warnings, 37 subtests passed in 1.07s

# pylint: disable=redefined-outer-name
# We cache globals and attribute lookups, so disable this warning.

logger = get_logger(__name__)


def _quick_drain(connection, timeout=0.1):
    try:
        connection.drain_events(timeout=timeout)
    except Exception as exc:  # pylint: disable=broad-except
        exc_errno = getattr(exc, 'errno', None)
        if exc_errno is not None and exc_errno != errno.EAGAIN:
            raise


def _enable_amqheartbeats(timer, connection, rate=2.0):
    heartbeat_error = [None]

    if not connection:
        return heartbeat_error

    heartbeat = connection.get_heartbeat_interval()  # negotiated
  
…[7464 chars omitted]…
py:519:            synloop(*x.args)
t/unit/worker/test_loops.py:528:            synloop(*x.args)
t/unit/worker/test_loops.py:534:            synloop(*x.args)
t/unit/worker/test_loops.py:540:        assert synloop(*x.args) is None
t/unit/worker/test_loops.py:549:        synloop(*x.args)
t/unit/worker/test_loops.py:568:            synloop(*x.args)
t/unit/worker/test_loops.py:581:        synloop(*x.args)

from celery.worker import state
from celery.worker.consumer import Consumer
from celery.worker.loops import _quick_drain, asynloop, synloop


class PromiseEqual:

    def __init__(self, fun, *args, **kwargs):
        self.fun = fun
        self.args = args
        self.kwargs = kwargs

    def __eq__(self, other):
        return (other.fun == self.fun and
                other.args == self.args and
                other.kwargs == self.kwargs)

    def __repr__(self):
        return '<promise: {0.fun!r} {0.args!r} {0
…[8630 chars omitted]…
        x.hub.reset = Mock(name='hub.reset()')
t/unit/worker/test_loops.py:496:        x.hub.reset.assert_not_called()
t/unit/worker/test_loops.py:498:    def test_hub_reset_error_still_reraises_original(self):
t/unit/worker/test_loops.py:502:        x.hub.reset = Mock(name='hub.reset()', side_effect=RuntimeError('reset failed'))
t/unit/worker/test_loops.py:510:        x.hub.reset.assert_called_once()

    def test_drain_after_consume(self):
        x, _ = get_task_callback(self.app, transport_driver_type='amqp')
        assert _quick_drain in [p.fun for p in x.hub._ready]

    def test_pool_did_not_start_at_startup(self):
        x = X(self.app)
        x.obj.restart_count = 0
        x.obj.pool.did_start_ok.return_value = False
        with pytest.raises(WorkerLostError):
            asynloop(*x.args)

    def test_setup_heartbeat(self):
        x = X(self.app, heartbeat=10)
        x.hub.timer.call_repeatedly = Mock(name='x.hub.call_repeatedly()')
        x.blueprint.state = CLOSE
        asynloop(*x.args)
        x.consumer.consume.assert_called_with()
        x.o
…[8629 chars omitted]…
ion_errors = consumer.connection_errors
celery/worker/strategy.py:148:            eventer=eventer, task=task, connection_errors=connection_errors,
/usr/local/lib/python3.12/site-packages/kombu/asynchronous/hub.py:57:class Hub:
/usr/local/lib/python3.12/site-packages/kombu/asynchronous/hub.py:115:    def reset(self):
/usr/local/lib/python3.12/site-packages/kombu/pools.py:144:def reset(*args, **kwargs):

   402	        gen = Gen()
   403	        x.hub.add_writer(6, gen)
   404	        x.hub.remove = Mock(name='hub.remove()')
   405	        x.hub.on_tick.add(x.close_then_error(Mock(name='tick'), 2))
   406	        x.hub.poller.poll.return_value = [(6, WRITE)]
   407	        with pytest.raises(ValueError):
   408	            asynloop(*x.args)
   409	        assert gen.gi_frame is None
   410	        x.hub.remove.assert_called_with(6)
   41
…[8630 chars omitted]…
           warnings.warn(UserWarning(W_POOL_SETTING))
        threaded = not w.use_eventloop or IS_WINDOWS
        procs = w.min_concurrency
        w.process_task = w._process_task
        if not threaded:
            semaphore = w.semaphore = LaxBoundedSemaphore(procs)
            w._quick_acquire = w.semaphore.acquire
            w._quick_release = w.semaphore.release
            max_restarts = 100

from copy import copy
from queue import Empty
from time import sleep
from types import GeneratorType as generator

from vine import Thenable, promise

from kombu.log import get_logger
from kombu.utils.compat import fileno
from kombu.utils.eventio import ERR, READ, WRITE, poll
from kombu.utils.objects import c
…[8630 chars omitted]…
untimeError from reset()
        with pytest.raises(socket.error):
            asynloop(*x.args)
        x.hub.reset.assert_called_once()


class test_synloop:

    def test_timeout_ignored(self):
        x = X(self.app)
        x.timeout_then_error(x.connection.drain_events)
        with pytest.raises(socket.error):
            synloop(*x.args)
        assert x.connection.drain_events.call_count == 2

from weakref import proxy as weakrefproxy

from vine.utils import wraps

from kombu.log import get_logger

if sys.version_info >= (3, 9):
    from zoneinfo import ZoneInfo
else:
    from backports.zoneinfo import ZoneInfo

if TYPE_CHECKING:
    from types import TracebackType

__all__ = ('Entry', 'Timer', 'to_timestamp')

logger = get_logger(__name__)

DEFAULT_MAX_INTERVAL = 2
EPOCH = datetime.fromtimestamp(0, ZoneInfo("UTC"))
IS_PYPY = hasattr(sys, 'pypy_version_info')

scheduled = namedtuple('scheduled', ('eta', 'priority', 'entry'))


def to_timestamp(d, default_timezone=ZoneInfo("UTC"), time
…[8630 chars omitted]…
py:519:            synloop(*x.args)
t/unit/worker/test_loops.py:528:            synloop(*x.args)
t/unit/worker/test_loops.py:534:            synloop(*x.args)
t/unit/worker/test_loops.py:540:        assert synloop(*x.args) is None
t/unit/worker/test_loops.py:549:        synloop(*x.args)
t/unit/worker/test_loops.py:568:            synloop(*x.args)
t/unit/worker/test_loops.py:581:        synloop(*x.args)

477:            conn.transport.register_with_event_loop(conn.connection, self.hub)
548:        if self.hub:
549:            self.hub._ready.add(self._flush_events)
786:    def start(self, c):
                else self.qos.increment_eventually)(
            abs(index) * self.prefetch_multiplier)

    def _limit_move_to_
…[8629 chars omitted]…
                   except Empty:
                            pass
                if to_consolidate:
                    consolidate_callback(to_consolidate)
            else:
                # no sockets yet, startup is probably not done.
                sleep(min(poll_timeout, 0.1))
            yield

    def repr_active(self):
        from .debug import repr_active
        return repr_active(self)

        self.controller = controller
        self.init_callback = init_callback
        self.hostname = hostname or gethostname()
        self.pid = os.getpid()
        self.pool = pool
        self.timer = timer
        self.strategies = self.Strategies()
        self.conninfo = self.app.connection_for_read()
        self.connection_errors = self.conninfo.connection_errors
        self.channel_errors = self.conninfo.channel_erro
…[8629 chars omitted]…
s)
t/unit/worker/test_loops.py:534:            synloop(*x.args)
t/unit/worker/test_loops.py:540:        assert synloop(*x.args) is None
t/unit/worker/test_loops.py:549:        synloop(*x.args)
t/unit/worker/test_loops.py:568:            synloop(*x.args)
t/unit/worker/test_loops.py:581:        synloop(*x.args)
258:    def close(self, *args):
281:    def _discard(self, fd):
292:    def create_loop(self,

    def close(self, *args):
        [self._unregister(fd) for fd in self.readers]
        self.readers.clear()
        [self._unregister(f
…[1898 chars omitted]…
%s' % (self.repr_active(),))
            if readers or writers:
                to_consolidate = []
                try:
                    events = poll(poll_timeout)
                    #  print('[EVENTS]: %s' % (self.repr_events(events),))
                except ValueError:  # Issue celery/#882
                    return

271:class WorkerTerminate(SystemExit):
278:class WorkerShutdown(SystemExit):

…[212 chars omitted]…
2:            self.should_use_eventloop() if use_eventloop is None
celery/worker/worker.py:123:            else use_eventloop
celery/worker/worker.py:239:    def should_use_eventloop(self):
celery/worker/components.py:79:        # Since asynloop() no longer resets the hub on exit (to preserve timers
celery/worker/components.py:130:    def close(self, w):
celery/worker/state.py:225:    def close(self):

    def signal_consumer_close(self):
        try:
            self.consumer.close()
        except AttributeError:
            pass

    def should_use_eventloop(self):
        return (detect_environment() == 'default' and
                self._conninfo.transport.implements.asynchronous and
                not self.app.IS_WINDOWS)

    def stop(self, in_sighandler=False, exitcode=None):
        """Graceful shutdown of the worker server (Warm shutdown)."""
        if exitcode i
…[8629 chars omitted]…
elf):
celery/worker/pidbox.py:63:    def reset(self):
celery/worker/pidbox.py:96:    def reset(self):
celery/worker/state.py:77:def reset_state():
celery/worker/consumer/events.py:44:            buffer_group=['task'] if c.hub else None,
celery/worker/consumer/events.py:45:            on_send_buffered=c.on_send_event_buffered if c.hub else None,
celery/worker/consumer/gossip.py:57:            if c.hub:

        self.blueprint.shutdown(self)

    def stop(self):
        self.blueprint.stop(self)

    def on_ready(self):
        callback, self.init_callback = self.init_callback, None
        if callback:
            callback(self)

    def loop_args(self):
        return (self, self.connection, self.task_consumer,
                self.blueprint, self.hub, self.qos, self.amqheartbeat,
                self.app.clock, self.amqheartbeat_rate)

    def on_decode_error(se
…[874 chars omitted]…
=None, on_interval=None):
celery/backends/redis.py:146:    def drain_events(self, timeout=None):
celery/backends/rpc.py:66:    def drain_events(self, timeout=None):
celery/worker/loops.py:21:        connection.drain_events(timeout=timeout)
celery/worker/loops.py:139:            connection.drain_events(timeout=2.0)
celery/worker/pidbox.py:118:                        connection.drain_events(timeout=1.0)

129:    def _loop_cycle():

            prefetch_count = max(w.max_concurrency, 1) * w.prefetch_multiplier
        else:
            prefetch_count = w.concurrency * w.prefetch_multiplier
        c = w.consumer = self.instantiate(
            w.consumer_cls, w.process_task,
            hostname=w.hostname,
            task_events=w.task_events,
            init_callback=w.ready_callback,
            initial_prefetch_count=prefetch_count,
            pool=w.pool,
            timer=w.timer,
            app=w.app,
            controller=w,
            hub=w.hub,
            worker_options=w.options,
     
…[1191 chars omitted]…
 conn):
        # Callback called for each retry while the connection
        # can't be established.
        def _error_handler(exc, interval, next_step=CONNECTION_RETRY_STEP):
            if getattr(conn, 'alt', None) and interval == 0:
                next_step = CONNECTION_FAILOVER
            elif interval > 0:
                self.broker_connection_retry_attempt += 1
./pyproject.toml
./setup.cfg

rn logger.exception('election request missing field %s', exc)
celery/worker/consumer/gossip.py:128:                    logger.exception('Unknown election topic %r', topic)
celery/worker/consumer/gossip.py:153:                logger.exception(
celery/worker/consumer/mingle.py:10:debug, info, exception = logger.debug, logger.info, logger.exception
celery/worker/loops.py:110:            logger.exception(

            # no append here list should be fixed size=1
            heartbeat_error[0] = e

    timer.call_repeatedly(heartbeat / rate, tick, (rate,))
    return heartbeat_error


def _clean_up_after_connection_error(hub, clear_timer=False):
    if hub is None:
        return

    try:
        hub.reset()
    except Exception as exc:  # pylint: disable=broad-except
        logger.error(
            'Error cleaning up after event loop: %r', exc, exc_info=True)

    if clear_timer:
        try:
        
…[6766 chars omitted]…

+    try:
+        while blueprint.state == RUN and obj.connection:
+            try:
+                state.maybe_shutdown()
+            finally:
+                _loop_cycle()
+    except errors:
+        # A synchronous loop normally has no hub, but reset it if one is
+        # available to ensure old connection state is not retained.
+        _clean_up_after_connection_error(hub)
+        raise

    # Missing docstring in public method
    D102,
    # Missing docstring in public package
    D104,
    # Missing docstring in magic method
    D105,
    # Missing docstring in __init__
    D107,
    # First line should be in imperative mood; try rephrasing
    D401,
    # No blank lines allowed between a section header and its content
    D412,
    # ambiguous variable name '...'
    E741,
    # ambiguous class definition '...'
    E742,
per-file-ignores =
   t/*,setup.py,examples/*,docs/*,extra/*:
       # docstrings
       D,

[bdist_rpm]
requires = backports.zoneinfo>=0.2.1;python_version<'3.9'
           tzdata>=2022.7
           tzlocal
           billiard >=4.1.0,<5.0
           kombu >= 5.3.4,<6.0.0


[metadata]
license_files = LICENSE
setup.cfg:6:[flake8]
tox.ini:9:    flake8

+        return
+
+    try:
+        hub.reset()
+    except Exception as exc:  # pylint: disable=broad-except
+        logger.error(
+            'Error cleaning up after event loop: %r', exc, exc_info=True)
+
+    if clear_timer:
+        try:
+            hub.timer.clear()
+        except Exception as exc:  # pylint: disable=broad-except
+            logger.error(
+                'Error cleaning up after event loop: %r
…[2194 chars omitted]…
     state.maybe_shutdown()
+            finally:
+                _loop_cycle()
+    except errors:
+        # A synchronous loop normally has no hub, but reset it if one is
+        # available to ensure old connection state is not retained.
+        _clean_up_after_connection_error(hub)
+        raise
..............................................                           [100%]
46 passed in 1.55s