This document describes the current stable version of Celery (5.6). For development docs, go here.

celery.worker.state

Internal worker state (global).

This includes the currently active and reserved tasks, statistics, and revoked tasks.

class celery.worker.state.Persistent(state, filename, clock=None)

Stores worker state between restarts.

This is the persistent data stored by the worker when celery worker --statedb is enabled.

Currently only stores revoked task id’s.

close()

compress(data, /, level=-1, wbits=15)

Returns a bytes object containing compressed data.

data

Binary data to be compressed.

level

Compression level, in 0-9 or -1.

wbits

The window buffer size and container format.

property db

decompress(data, /, wbits=15, bufsize=16384)

Returns a bytes object containing the uncompressed data.

data

Compressed data.

wbits

The window buffer size and container format.

bufsize

The initial output buffer size.

merge()

open()

protocol = 4

save()

storage = <module 'shelve' from '/home/docs/.asdf/installs/python/3.11.15/lib/python3.11/shelve.py'>

sync()

celery.worker.state.SOFTWARE_INFO = {'sw_ident': 'py-celery', 'sw_sys': 'Linux', 'sw_ver': '5.6.2'}

Worker software/platform information.

celery.worker.state.active_requests = set()

set of currently active Request’s.

celery.worker.state.maybe_shutdown()

Shutdown if flags have been set.

celery.worker.state.reserved_requests = set()

set of all reserved Request’s.

celery.worker.state.revoked = <LimitedSet(0): maxlen=50000, expires=10800.0, minlen=0>

the list of currently revoked tasks. Persistent if statedb set.

celery.worker.state.scheduled_requests = set()

set of Request’s scheduled for an ETA/countdown and not yet handed over to the pool.

A request is discarded from here by task_reserved() once its ETA/countdown has elapsed. Note that for a rate-limited task the ETA firing only moves the request into its token bucket (Consumer._limit_post_eta); it stays in this set until a token frees up and Consumer._limit_move_to_pool reserves it, so such a request keeps reporting scheduled after its ETA has passed even though inspect scheduled no longer lists it.

celery.worker.state.task_accepted(request, _all_total_count=None, add_request=<method-wrapper '__setitem__' of dict object>, add_active_request=<bound method WeakSet.add of set()>, add_to_total_count=<bound method Counter.update of Counter()>)

Update global state when a task has been accepted.

celery.worker.state.task_ready(request, successful=False, remove_request=<built-in method pop of dict object>, discard_active_request=<bound method WeakSet.discard of set()>, discard_reserved_request=<bound method WeakSet.discard of set()>, discard_scheduled_request=<bound method WeakSet.discard of set()>)

Update global state when a task is ready.

celery.worker.state.task_reserved(request, add_request=<method-wrapper '__setitem__' of dict object>, add_reserved_request=<bound method WeakSet.add of set()>, discard_scheduled_request=<bound method WeakSet.discard of set()>)

Update global state when a task has been reserved.

celery.worker.state.task_scheduled(request, add_request=<method-wrapper '__setitem__' of dict object>, add_scheduled_request=<bound method WeakSet.add of set()>, all_reserved_requests=set(), all_active_requests=set())

Update global state when a task has been scheduled for an ETA/countdown.

Unlike task_reserved(), this doesn’t add the request to reserved_requests: the request isn’t waiting for a worker pool slot yet, it’s only registered so that it can be found (e.g. by the query_task remote control command) before its ETA/countdown elapses.

This is a no-op for a request that already moved on to being reserved or active: with a threaded timer (celery.utils.timer2.Timer, used by the non-eventloop pools) an ETA that’s already in the past fires on the timer thread right away, so apply_eta_task() -> task_reserved() can run before the strategy gets here. Adding the request back to scheduled_requests then would misreport its state and let Consumer.on_close() drop a still-running task from requests.

celery.worker.state.total_count = {}

count of tasks accepted by the worker, sorted by type.