celery-celery-10218

docker
docs
examples
extra
helm-chart
pyproject.toml
requirements
setup.cfg
setup.py
t
tox.ini
/repo/celery/app/defaults.py
/repo/celery/concurrency/asynpool.py
/repo/celery/concurrency/base.py
/repo/celery/concurrency/eventlet.py
/repo/celery/concurrency/gevent.py
/repo/celery/concurrency/prefork.py
/repo/celery/contrib/abortable.py
/repo/celery/events/snapshot.py
/repo/celery/platforms.py
/repo/celery/utils/timer2.py

# 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
    if n
…[3727 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()

        # To avoid infinite loop where one of the callables adds items
        # to self._ready (via call_soon or otherwise).
        # we create new list with current self._ready
        todos = self._pop_ready()
        for item in todos:
            item()

        # Clear global event loop variable if this hub is the current loop
        if _current_loop is self:
            set_event_loop(None)

    def _create_poller(self):
        self._poller = poll()
        self._register_fd = self._poller.register
        self._unregister_fd = self._poller.unregister

    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

286:            asynloop(*x.args)
299:            asynloop(*x.args)
311:            asynloop(*x.args)
323:            asynloop(*x.args)
334:            asynloop(*x.args)
349:            asynloop(*x.args)
360:            asynloop(*x.args)
377:            asynloop(*x.args)
393:            asynloop(*x.args)
408:            asynloop(*x.args)
420:            asynloop(*x.args)
429:        asynloop(*x.args)
443:            asynloop(*x.args)
452:        asynloop(*x.args)
456:    def test_hub_reset_on_connection_error(self):
460:        x.hub.reset = Mock(name='hub.reset()')
466:            asynloop(*x.args)
467:        x.hub.reset.assert_called_once()

        x.hub.timer.call_repeatedly.assert_not_called()

    def test_hub_reset_on_connection_error(self):
        x = X(self.app)
        x.hub.readers = {6: Mock()}
        x.hub.timer._queue = [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)

        try:
            try:
                hub.reset()
            except Exception as exc:  # pylint:
…[469 chars omitted]…
clock, hbrate=2.0, **kwargs):
    """Fallback blocking event loop for transports that doesn't support AIO."""
    RUN = bootsteps.RUN
    on_task_received = obj.create_task_handler()
    perform_pending_operations = obj.perform_pending_operations
    heartbeat_error = [None]
    if getattr(obj.pool, 'is_green', False):
        heartbeat_error = _enable_amqheartbeats(obj.timer, connection, rate=hbrate)

                         hub,
                         Mock(name='qos'),
                         heartbeat,
                         Mock(name='clock')]
        self.connection.supports_heartbeats = True
        self.connection.get_heartbeat_interval.side_effect = (
            lambda: self.heartbeat
        )
        self.consumer.callbacks = []
        self.obj.strategies = {}
        self.connection.connection_errors = (socket.error,)
        if transport_driver_type:
            self.connection.
…[2064 chars omitted]…
ueprint.state = CLOSE

    def closer(self, mock=None, mod=0):
        mock = Mock() if mock is None else mock

        def closing(*args, **kwargs):
            if not mod or mock.call_count >= mod:
                self.close()
        mock.side_effect = closing
        return mock


def get_task_callback(*args, **kwargs):
    x = X(*args, **kwargs)
    x.blueprint.state = CLOSE
    asynloop(*x.args)

  File "/usr/local/lib/python3.12/site-packages/_pytest/python.py", line 167 in pytest_pyfunc_call
  File "/usr/local/lib/python3.12/site-packages/pluggy/_callers.py", line 121 in _multicall
  File "/usr/local/lib/python3.12/site-packages/pluggy/_manager.py", line 120 in _hookexec
  File "/usr/local/lib/python
…[1998 chars omitted]…
  File "/usr/local/lib/python3.12/site-packages/pluggy/_manager.py", line 120 in _hookexec
  File "/usr/local/lib/python3.12/site-packages/pluggy/_hooks.py", line 512 in __call__
  File "/usr/local/lib/python3.12/site-packages/_pytest/config/__init__.py", line 229 in _main
  File "/usr/local/lib/python3.12/site-packages/_pytest/config/__init__.py", line 201 in main
  File "<stdin>", line 4 in <module>

        while 1:
            todo = self._pop_ready()

            for item in todo:
                if item:
                    item()

            poll_timeout = fire_timers(propagate=propagate) if scheduled else 1

            for tick_callback in copy(self.on_tick):
                tick_callback()

            #  print('
…[2241 chars omitted]…
                   else:
                        try:
                            cb(*cbargs)
                        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

+        # connection and cause repeated loop restarts before reconnecting.
+        # Cleanup is attempted even if resetting the hub fails, and any
+        # cleanup error must not 
…[621 chars omitted]…
ar()`` is
+                # unavailable or ineffective for the timer implementation.
+                queue = getattr(timer, '_queue', None)
+                if queue:
+                    queue.clear()
+            except Exception as exc:  # pylint: disable=broad-except
+                logger.exception(
+                    'Error clearing timer queue after event loop: %r', exc)
         raise