celery-celery-10086

./.github/ISSUE_TEMPLATE.md
./.github/opencollective.yml
./.github/PULL_REQUEST_TEMPLATE.md
./.gitignore
./.pre-commit-config.yaml
./.readthedocs.yaml
./bandit.json
./celery/beat.py
./celery/bootsteps.py
./celery/canvas.py
./celery/exceptions.py
./celery/local.py
./celery/platforms.py
./celery/result.py
./celery/schedules.py
./celery/signals.py
./celery/states.py
./celery/_state.py
./celery/__init__.py
./celery/__main__.py
./celery.egg-info/dependency_links.txt
./celery.egg-info/entry_points.txt
./celery.egg-info/PKG-INFO
./celery.egg-info/requires.txt
./celery.egg-info/SOURCES.txt
./celery.egg-info/top_level.txt
./Changelog.rst
./CONTRIBUTING.rst
./CONTRIBUTORS.txt
./do
…[8629 chars omitted]…
 allows users to remap the SIGTERM signal to SIGQUIT, to initiate a soft or a cold shutdown using :sig:`TERM`
./docs/history/changelog-5.5.rst:1069:The soft shutdown is a new mechanism in Celery that sits between the warm shutdown and the cold shutdown.
./docs/history/changelog-5.5.rst:1071:After the soft shutdown ends, the worker will initiate a graceful cold shutdown, stopping all tasks and exiting.

…[1672 chars omitted]…
        pool = asynpool.AsynPool(processes=1, threads=False)
t/unit/concurrency/test_prefork.py:499:        x = asynpool.ResultHandler(
t/unit/concurrency/test_prefork.py:528:        assert pool._pool._state == asynpool.RUN
t/unit/worker/test_worker.py:751:        from celery.concurrency.asynpool import Worker
t/unit/worker/test_worker.py:890:        # worker.pool._pool is the asynpool.AsynPool class.

    def test_iterate_file_descriptors_safely_source_data_dict(self):
        # Given: a list of integers that could be file descriptors
        fd_iter = {1: 1, 2: 2, 3: 3, 4: 4, 5: 5}

        # Given: a mock hub method that does nothing to call
        def _fake_hub(*args, **kwargs):
            raise OSError

        # When Calling the helper to iterate_file_descriptors_safely
        iterate_file_descriptors_safely(
            fd_iter, fd_iter, _fake_hub,
     
…[5545 chars omitted]…
ader, 3, x.on_state_change,
        )


class test_TaskPool:

    def test_start(self):
        pool = TaskPool(10)
        pool.start()
        assert pool._pool.started
        assert pool._pool._state == asynpool.RUN

        _pool = pool._pool
celery/concurrency/asynpool.py:730:        hub_add, hub_remove = hub.add, hub.remove
celery/concurrency/asynpool.py:801:        self.hub_remove = hub_remove

from celery.app.defaults import DEFAULTS
from celery.concurrency.asynpool import iterate_file_descriptors_safely
from celery.utils.collections import AttributeDict
from celery.utils.functional import noop
from celery.utils.objects import Bunch

try:
    from celery.concurrency import asynpool
    from celery.concurrency import prefork as mp
except ImportError:

    class _mp:
        RUN = 0x1

        class TaskPool:
            _pool = Mock()

            def __init__(self, *args, **kwargs):
                pass

            def start(self):
                pass

            def stop(self):
                pass

            def apply_async(self, *args, **kwargs):
 
…[1083 chars omitted]…
ker.com')
            _signals.ignore.assert_any_call(*WORKER_SIGIGNORE)
            _signals.reset.assert_any_call(*WORKER_SIGRESET)
./celery/concurrency/asynpool.py:497:        self._untrack_child_process(proc, hub)
./celery/concurrency/asynpool.py:515:    def _untrack_child_process(self, proc, hub):
./celery/concurrency/asynpool.py:703:            self._untrack_child_process(proc, hub)
ecd736e base

                os.kill(proc.pid, 9)

        def on_process_up(proc):
            """Called when a process has started."""
            # If we got the same fd as a previous process then we'll also
            # receive jobs in the old buffer, so we need to reset the
            # job._write_to and job._sc
…[3667 chars omitted]…
.py:475:        pool.register_with_event_loop(hub)
t/unit/concurrency/test_prefork.py:511:        x.register_with_event_loop(hub)
t/unit/concurrency/test_prefork.py:554:    def test_register_with_event_loop(self):
t/unit/concurrency/test_prefork.py:558:        pool.register_with_event_loop(loop)
t/unit/concurrency/test_prefork.py:559:        pool._pool.register_with_event_loop.assert_called_with(loop)

+        if fd is not None:
+            proc._sentinel_poll = None
             hub.remove(fd)
             os.close(fd)

+        if fd is not None:
+            proc._sentinel_poll = None
             hub.remove(fd)
             os.close(fd)

./.github/copilot-instructions.md
./.github/dependabot.yml
./.github/opencollective.yml
./.gitignore
./.pre-commit-config.yaml
./.readthedocs.yaml
./CONTRIBUTING.rst
./CONTRIBUTORS.txt
./Changelog.rst
./LICENSE
./MANIFEST.in
./Makefile
./README.rst
./SECURITY.md
./TODO
./bandit.json
./celery.egg-info/PKG-INFO
./celery.egg-info/SOURCES.txt
./celery.egg-info/dependency_links.txt
./celery.egg-info/entry_points.txt
./celery.egg-info/requires.txt
./celery.egg-info/top_level.txt
./celery/__init__.py
./celery/__main__.py
./celery/_state.py
./celery/beat.py
./celery/bootsteps.py
./celery/canvas.py
./celery/exceptions.py
./celery/local.py
./celery/platforms.py
./celery/result
…[8629 chars omitted]…
search.py:107:        res = x.get(sentinel.task_id)
./t/unit/backends/test_elasticsearch.py:117:        res = x.get(sentinel.task_id)
./t/unit/backends/test_elasticsearch.py:124:        x._server.delete.return_value = sentinel.result
./t/unit/backends/test_elasticsearch.py:126:        assert x.delete(sentinel.task_id) is None
./t/unit/backends/test_elasticsearch.py:128:            id=sentinel.task_id,

        # Set of active co-routines currently writing jobs.
        self._active_writers = set()

        # Set of fds that are busy (executing task)
        self._busy_workers = set()
        self._mark_worker_as_available = self._busy_workers.discard

        # Holds jobs waiting to be written to child processes.
        self.outbound_buffer = deque()

        self.write_stats = Counter()

        super().__init__(processes, *args, **kwargs)

        for proc in self._pool:
            # create
…[2560 chars omitted]…
f test_cleanup_after_job_finish(self):
t/unit/concurrency/test_eventlet.py:164:        TaskPool._cleanup_after_job_finish(None, testMap, '1')
t/unit/concurrency/test_gevent.py:112:    def test_cleanup_after_job_finish(self):
t/unit/concurrency/test_gevent.py:114:        TaskPool._cleanup_after_job_finish(None, testMap, '1')
sed: can't read t/unit/concurrency/test_asynpool.py: No such file or directory

1192:    def _stop_task_handler(task_handler):

        waiting_to_start = self._waiting_to_start

        def verify_process_alive(proc):
            proc = proc()  # is a weakref
            if (proc is not None and proc._is_alive() and
                    proc in waiting_to_start):
                assert proc.outqR_fd in fileno_to_outq
                assert fileno_to_outq[proc.outqR_fd] 
…[7220 chars omitted]…
                    raise

    def create_result_handler(self):
        return super().create_result_handler(
            fileno_to_outq=self._fileno_to_outq,
            on_process_alive=self.on_process_alive,
        )
t/unit/worker/test_worker.py:890:        # worker.pool._pool is the asynpool.AsynPool class.
t/unit/worker/test_worker.py:900:        worker.pool._pool._track_child_process(proc, hub)

+        fd = getattr(proc, '_sentinel_poll', None)
+        if fd is not None:
+            proc._sentinel_poll = None
             hub.remove(fd)
             os.close(fd)

def long_something():
    time.sleep(1)


def raise_something(i):
    try:
        raise KeyError('FOO EXCEPTION')
    except KeyError:
        return ExceptionInfo()


class test_TaskPool:

    def setup_method(self):
        from celery.concurrency.prefork import TaskPool
        self.TaskPool = TaskPool

    def test_attrs(self):
        p = self.TaskPool(2)
        assert p.limit == 2
        assert p._pool is None

    def x_apply(self):
        p = self.TaskPool(2)
        p.start()
        scratchpad = {}
        proc_counter = itertools.count()

        def mycallback(ret_value):
            process = next(proc_counter)
            scratchpad[process] = {}
            scratchpad[process]['ret_value'] = ret_value

…[2473 chars omitted]…
_or_terminate(self):
        worker = self.app.WorkController(concurrency=1, loglevel=0)
        worker.stop()
        assert worker.blueprint.state != CLOSE
        worker.terminate()
        assert worker.blueprint.state != CLOSE
ERROR: usage: __main__.py [options] [file_or_dir] [file_or_dir] [...]
__main__.py: error: unrecognized arguments: --no-cov
  inifile: /repo/pyproject.toml
  rootdir: /repo

+            proc._sentinel_poll = None
             hub.remove(fd)
             os.close(fd)
 
 M celery/concurrency/asynpool.py

-rw-r--r-- 1 root root  44764 Sep 30 17:21 CONTRIBUTING.rst
-rw-r--r-- 1 root root   8550 Sep 30 17:21 CONTRIBUTORS.txt
-rw-r--r-- 1 root root 128886 Sep 30 17:21 Changelog.rst
-rw-r--r-- 1 root root   2717 Sep 30 17:21 LICENSE
-rw-r--r-- 1 root root    709 Sep 30 17:21 MANIFEST.in
-rw-r--r-- 1 root root   5866 Sep 30 17:21 Makefile

…[2821 chars omitted]…
elery/apps/multi.py
./celery/apps/worker.py
./celery/apps/__init__.py
./celery/backends/arangodb.py
./celery/backends/asynchronous.py
./celery/backends/azureblockblob.py
./celery/backends/base.py
./celery/backends/cache.py
./celery/backends/cassandra.py
./celery/backends/consul.py
./celery/backends/cosmosdbsql.py
./celery/backends/couchbase.py
./celery/backends/couchdb.py
./celery/backends/dynamodb.py

celery/backends/mongodb.py:266:    def cleanup(self):
celery/backends/rpc.py:34:def _on_after_fork_cleanup_backend(backend):
celery/concurrency/asynpool.
…[54 chars omitted]…
elery/concurrency/eventlet.py:180:    def _cleanup_after_job_finish(greenlet, pool_map, pid):
celery/concurrency/gevent.py:165:    def _cleanup_after_job_finish(greenlet, pool_map, pid):
celery/events/snapshot.py:54:    def on_cleanup(self):
celery/events/snapshot.py:57:    def cleanup(self):
celery/loaders/base.py:71:    def on_process_cleanup(self):
celery/utils/threads.py:292:    def cleanup(self):

def long_something():
    time.sleep(1)


def raise_something(i):
    try:
        raise KeyError('FOO EXCEPTION')
    except KeyError:
        return ExceptionInfo()


class test_TaskPool:

    def setup_method(self):
        from celery.concurrency.prefork import TaskPool
        self.TaskPool = TaskPool

    def test_attrs(self):
        p = self.TaskPool(2)
        assert p.limit == 2
        assert p._pool is None

    def x_apply(self):
        p = self.TaskPool(2)
        p.start()
        scratchpad = {}
        proc_counter = itertools.count()

        def mycallback(ret_value):
            process = next(proc_counter)
            scratchpad[process] = {}
            scratchpad[process]['ret_value'] = ret_value

…[3529 chars omitted]…
worker.blueprint.state = RUN
            worker.stop(in_sighandler=True)
            assert worker.blueprint.state != CLOSE
            worker.terminate(in_sighandler=True)
            assert worker.blueprint.state != CLOSE
        finally:
            worker.pool.signal_safe = sigsafe

    def test_on_timer_error(self):
        worker = self.app.WorkController(concurrency=1, loglevel=0)

        try:

+        if fd is not None:
+            proc._sentinel_poll = None
             hub.remove(fd)
             os.close(fd)

./.github/copilot-instructions.md
./.github/dependabot.yml
./.github/opencollective.yml
./.gitignore
./.pre-commit-config.yaml
./.readthedocs.yaml
./CONTRIBUTING.rst
./CONTRIBUTORS.txt
./Changelog.rst
./LICENSE
./MANIFEST.in
./Makefile
./README.rst
./SECURITY.md
./TODO
./bandit.json
./celery.egg-info/PKG-INFO
./celery.egg-info/SOURCES.txt
./celery.egg-info/dependency_links.txt
./celery.egg-info/entry_points.txt
./celery.egg-info/requires.txt
./celery.egg-info/top_level.txt
./celery/__init__.py
./celery/__main__.py
./celery/_state.py
./celery/beat.py
./celery/bootsteps.py
./celery/canvas.py
./celery/exceptions.py
./celery/local.py
./celery/platforms.py
./celery/result
…[8630 chars omitted]…
151:        def test_cold_shutdown(self, celery_setup: CeleryTestSetup):
./t/smoke/tests/test_worker.py:212:            worker.assert_log_exists("Waiting gracefully for cold shutdown to complete...")
./t/smoke/tests/test_worker.py:312:                worker.assert_log_exists("Waiting gracefully for cold shutdown to complete...")
./t/unit/worker/test_worker.py:897:        proc = Mock(_sentinel_poll=42)

        # outqueue fileno -> process mapping
        self._fileno_to_outq = {}
        # synqueue fileno -> process mapping
        self._fileno_to_synq = {}

        # We keep track of processes that haven't yet
        # sent a WORKER_UP message.  If a process fails to send
        # this message within _proc_alive_timeout we terminate it
        # and hope the next proces
…[3567 chars omitted]…
all_repeatedly(interval, handler)
t/unit/concurrency/test_prefork.py:137:        self.maintain_pool = Mock()
t/unit/concurrency/test_prefork.py:503:            on_process_alive=Mock(),
t/unit/concurrency/test_prefork.py:137:        self.maintain_pool = Mock()
t/unit/concurrency/test_prefork.py:503:            on_process_alive=Mock(),
t/unit/backends/test_asynchronous.py
t/unit/concurrency/test_pool.py

def long_something():
    time.sleep(1)


def raise_something(i):
    try:
        raise KeyError('FOO EXCEPTION')
    except KeyError:
        return ExceptionInfo()


class test_TaskPool:

    def setup_method(self):
        from celery.concurrency.prefork import TaskPool
        self.TaskPool = TaskPool

    def test_attrs(self):
        p = self.TaskPool(2)
        assert p.limit == 2
        assert p._pool is None

    def x_apply(self):
        p = self.TaskPool(2)
        p.start()
        scratchpad = {}
        proc_counter = itertools.count()

        def mycallback(ret_value):
            process = next(proc_counter)
            scratchpad[process] = {}
            scratchpad[process]['ret_value'] = ret_value

…[793 chars omitted]…
llback=mycallback)

        assert res3.get() == 900
        time.sleep(0.5)
        assert scratchpad.get(3)['ret_value'] == 900
        p.stop()
./celery/concurrency/asynpool.py:497:        self._untrack_child_process(proc, hub)
./celery/concurrency/asynpool.py:515:    def _untrack_child_process(self, proc, hub):
./celery/concurrency/asynpool.py:703:            self._untrack_child_process(proc, hub)

                if job._write_to and job._write_to.inqW_fd == infd:
                    job._write_to = proc
                if job._scheduled_for and job._scheduled_for.inqW_fd == infd:
                    job._scheduled_for = proc
            fileno_to_outq[proc.outqR_fd] = proc

            # maint
…[8629 chars omitted]…
sk[fd] = fd_to_mask.get(fd, 0) | POLLIN
        if writers:
            for fd in map(_ensure_integral_fd, writers):
                fd_to_mask[fd] = fd_to_mask.get(fd, 0) | POLLOUT
        if err:
            for fd in map(_ensure_integral_fd, err):
                fd_to_mask[fd] = fd_to_mask.get(fd, 0) | POLLERR

        for fd, event_mask in fd_to_mask.items():
            register(fd, event_mask)

            poll.side_effect.errno = 34134
            with pytest.raises(socket.error):
                asynpool._select({3}, poll=poll)

    def test_select_unpatched(self):
        with tempfile.TemporaryFile('w') as f:
            _, writeable, _ = asynpool._select(writers={f, }, err={f, })
            assert f.fileno() in writeable

        with tempfile.TemporaryFile('r') as f:
            readable,
…[8041 chars omitted]…
orker.py:897:        proc = Mock(_sentinel_poll=42)
celery/concurrency/asynpool.py:503:            fd = proc._sentinel_poll
celery/concurrency/asynpool.py:509:            fd = proc._sentinel_poll = os.dup(proc._popen.sentinel)
celery/concurrency/asynpool.py:516:        if proc._sentinel_poll is not None:
celery/concurrency/asynpool.py:517:            fd, proc._sentinel_poll = proc._sentinel_poll, None

        hub.add_reader = throw_file_not_found_error
        hub.remove = throw_file_not_found_error

        # When: Calling again to register with event loop ...
        worker.pool.register_with_event_loop(hub)
        worker.pool._pool.register_with_event_loop(hub)
        # Then: test did not raise OSError
        # Note: worker.pool is prefork.TaskPool whereas
        # worker.pool._pool is the asynpool.AsynPool class.

        # When: Calling th
…[621 chars omitted]…
 test_dont_stop_or_terminate(self):
        worker = self.app.WorkController(concurrency=1, loglevel=0)
        worker.stop()
        assert worker.blueprint.state != CLOSE
        worker.terminate()
        assert worker.blueprint.state != CLOSE
celery/concurrency/asynpool.py:687:            if getattr(proc, 'dead', None):
celery/concurrency/asynpool.py:687:            if getattr(proc, 'dead', None):

+        if fd is not None:
+            proc._sentinel_poll = None
             hub.remove(fd)
             os.close(fd)

+            proc._sentinel_poll = None
             hub.remove(fd)
             os.close(fd)
 
 M celery/concurrency/asynpool.py

        try:
            fd = proc._sentinel_poll
        except AttributeError:
            # we need to duplicate the fd here to carefully
            # control when the fd is removed from the process table,
            # as once the original fd is closed we cannot unregister
            # the fd from epoll(7) anymore, causing a 100% CPU poll loop.
            fd = proc._sentinel_poll = os.dup(proc
…[516 chars omitted]…
vent loop."""
        self._result_handler.register_with_event_loop(hub)
        self.handle_result_event = self._result_handler.handle_event
        self._create_timelimit_handlers(hub)
        self._create_process_handlers(hub)
        self._create_write_handlers(hub)

        # Add handler for when a process exits (calls maintain_pool)
        [self._track_child_process(w, hub) for w in self._pool]

    def register_with_event_loop(self, hub):
        """Register the async pool with the current event loop."""