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

# Source code for celery.utils.timer2

    """Scheduler for Python functions.

    .. note::
        This is used for the thread-based worker only,
        not for amqp/redis/sqs/qpid where :mod:`kombu.asynchronous.timer` is used.
    """
    import os
    import sys
    import threading
    from itertools import count
    from threading import TIMEOUT_MAX as THREAD_TIMEOUT_MAX
    from time import sleep
    from typing import Any, Callable, Iterator, Optional, Tuple

    from kombu.asynchronous.timer import Entry
    from kombu.asynchronous.timer import Timer as Schedule
    from kombu.asynchronous.timer import logger, to_timestamp

    TIMER_DEBUG = os.environ.get('TIMER_DEBUG')

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

    class Timer(threading.Thread):
        """Timer thread.

        Note:
            This is only used for transports not supporting AsyncIO.
        """

        Entry = Entry
        Schedule = Schedule

        running: bool = False
        on_tick: Optional[Callable[[float], None]] = None

        _timer_count: count = count(1)

        if TIMER_DEBUG:  # pragma: no cover
            def start(self, *args: Any, **kwargs: Any) -> None:
                import traceback
                print('- Timer starting')
                traceback.print_stack()
                super().start(*args, **kwargs)

        def __init__(self, schedule: Optional[Schedule] = None,
                     on_error: Optional[Callable[[Exception], None]] = None,
                     on_tick: Optional[Callable[[float], None]] = None,
                     on_start: Optional[Callable[['Timer'], None]] = None,
                     max_interval: Optional[float] = None, **kwargs: Any) -> None:
            self.schedule = schedule or self.Schedule(on_error=on_error,
                                                      max_interval=max_interval)
            self.on_start = on_start
            self.on_tick = on_tick or self.on_tick
            super().__init__()
            # `_is_stopped` is likely to be an attribute on `Thread` objects so we
            # double underscore these names to avoid shadowing anything and
            # potentially getting confused by the superclass turning these into
            # something other than an `Event` instance (e.g. a `bool`)
            self.__is_shutdown = threading.Event()
            self.__is_stopped = threading.Event()
            self.mutex = threading.Lock()
            self.not_empty = threading.Condition(self.mutex)
            self.daemon = True
            self.name = f'Timer-{next(self._timer_count)}'

        def _next_entry(self) -> Optional[float]:
            with self.not_empty:
                delay: Optional[float]
                entry: Optional[Entry]
                delay, entry = next(self.scheduler)
                if entry is None:
                    if delay is None:
                        self.not_empty.wait(1.0)
                    return delay
            return self.schedule.apply_entry(entry)
        __next__ = next = _next_entry  # for 2to3

        def run(self) -> None:
            try:
                self.running = True
                self.scheduler: Iterator[Tuple[Optional[float], Optional[Entry]]] = iter(self.schedule)

                while not self.__is_shutdown.is_set():
                    delay = self._next_entry()
                    if delay:
                        if self.on_tick:
                            self.on_tick(delay)
                        if sleep is None:  # pragma: no cover
                            break
                        sleep(delay)
                try:
                    self.__is_stopped.set()
                except TypeError:  # pragma: no cover
                    # we lost the race at interpreter shutdown,
                    # so gc collected built-in modules.
                    pass
            except Exception as exc:
                logger.error('Thread Timer crashed: %r', exc, exc_info=True)
                sys.stderr.flush()
                os._exit(1)

        def stop(self) -> None:
            self.__is_shutdown.set()
            if self.running:
                self.__is_stopped.wait()
                self.join(THREAD_TIMEOUT_MAX)
                self.running = False

        def ensure_started(self) -> None:
            if not self.running and not self.is_alive():
                if self.on_start:
                    self.on_start(self)
                self.start()

        def _do_enter(self, meth: str, *args: Any, **kwargs: Any) -> Entry:
            self.ensure_started()
            with self.mutex:
                entry = getattr(self.schedule, meth)(*args, **kwargs)
                self.not_empty.notify()
                return entry

        def enter(self, entry: Entry, eta: float, priority: Optional[int] = None) -> Entry:
            return self._do_enter('enter_at', entry, eta, priority=priority)

        def call_at(self, *args: Any, **kwargs: Any) -> Entry:
            return self._do_enter('call_at', *args, **kwargs)

        def enter_after(self, *args: Any, **kwargs: Any) -> Entry:
            return self._do_enter('enter_after', *args, **kwargs)

        def call_after(self, *args: Any, **kwargs: Any) -> Entry:
            return self._do_enter('call_after', *args, **kwargs)

        def call_repeatedly(self, *args: Any, **kwargs: Any) -> Entry:
            return self._do_enter('call_repeatedly', *args, **kwargs)

        def exit_after(self, secs: float, priority: int = 10) -> None:
            self.call_after(secs, sys.exit, priority)

        def cancel(self, tref: Entry) -> None:
            tref.cancel()

        def clear(self) -> None:
            self.schedule.clear()

        def empty(self) -> bool:
            return not len(self)

        def __len__(self) -> int:
            return len(self.schedule)

        def __bool__(self) -> bool:
            """``bool(timer)``."""
            return True
        __nonzero__ = __bool__

        @property
        def queue(self) -> list:
            return self.schedule.queue