Script 'mail_helper' called by obssrc
Hello community,
here is the log from the commit of package python-futurist for openSUSE:Factory
checked in at 2026-08-22 21:34:59
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Comparing /work/SRC/openSUSE:Factory/python-futurist (Old)
and /work/SRC/openSUSE:Factory/.python-futurist.new.1258 (New)
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Package is "python-futurist"
Sat Aug 22 21:34:59 2026 rev:20 rq:1372930 version:3.5.0
Changes:
--------
--- /work/SRC/openSUSE:Factory/python-futurist/python-futurist.changes
2026-07-07 21:03:40.819255261 +0200
+++
/work/SRC/openSUSE:Factory/.python-futurist.new.1258/python-futurist.changes
2026-08-22 21:37:04.494217515 +0200
@@ -1,0 +2,11 @@
+Fri Aug 21 20:10:04 UTC 2026 - Dirk Müller <[email protected]>
+
+- update to 3.5.0:
+ * typing: Tighten signature for ProcessPoolExecutor.__init__
+ * typing: Expose function signature
+ * Don't alias typing library
+ * Use "futures" consistently in log messages
+ * Add DelayedExecutorMixin variants
+ * ruff: Configure hacking as external linter
+
+-------------------------------------------------------------------
Old:
----
futurist-3.4.0.tar.gz
New:
----
futurist-3.5.0.tar.gz
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Other differences:
------------------
++++++ python-futurist.spec ++++++
--- /var/tmp/diff_new_pack.In0Et3/_old 2026-08-22 21:37:05.148240893 +0200
+++ /var/tmp/diff_new_pack.In0Et3/_new 2026-08-22 21:37:05.150240965 +0200
@@ -17,7 +17,7 @@
Name: python-futurist
-Version: 3.4.0
+Version: 3.5.0
Release: 0
Summary: Useful additions to futures, from the future.
License: Apache-2.0
++++++ futurist-3.4.0.tar.gz -> futurist-3.5.0.tar.gz ++++++
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/futurist-3.4.0/.zuul.yaml
new/futurist-3.5.0/.zuul.yaml
--- old/futurist-3.4.0/.zuul.yaml 2026-06-26 15:32:16.000000000 +0200
+++ new/futurist-3.5.0/.zuul.yaml 2026-08-20 17:19:07.487014000 +0200
@@ -1,8 +1,11 @@
+---
+
- project:
templates:
- check-requirements
- lib-forward-testing-python3
- openstack-python3-jobs
+ - openstack-python3-next-jobs
- periodic-stable-jobs
- publish-openstack-docs-pti
- release-notes-jobs-python3
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/futurist-3.4.0/AUTHORS new/futurist-3.5.0/AUTHORS
--- old/futurist-3.4.0/AUTHORS 2026-06-26 15:33:06.000000000 +0200
+++ new/futurist-3.5.0/AUTHORS 2026-08-20 17:19:40.562633500 +0200
@@ -3,6 +3,7 @@
Akihiro Motoki <[email protected]>
Andreas Jaeger <[email protected]>
Anh Tran <[email protected]>
+Balazs Gibizer <[email protected]>
ChangBo Guo(gcb) <[email protected]>
Chenjun Shen <[email protected]>
Corey Bryant <[email protected]>
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/futurist-3.4.0/ChangeLog new/futurist-3.5.0/ChangeLog
--- old/futurist-3.4.0/ChangeLog 2026-06-26 15:33:06.000000000 +0200
+++ new/futurist-3.5.0/ChangeLog 2026-08-20 17:19:40.548633300 +0200
@@ -1,12 +1,23 @@
CHANGES
=======
+3.5.0
+-----
+
+* typing: Tighten signature for ProcessPoolExecutor.\_\_init\_\_
+* typing: Expose function signature
+* Don't alias typing library
+* Use "futures" consistently in log messages
+* Add DelayedExecutorMixin variants
+* ruff: Configure hacking as external linter
+
3.4.0
-----
* Add release note for mp\_context kwarg
* tests: Fix ProcessPoolExecutor race on Python 3.14+
* Drop Python 3.10 support
+* zuul: Use openstack-python3-next-jobs template
* pre-commit: Bump versions
* typing: Fixes for mypy 2.0.0
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/futurist-3.4.0/PKG-INFO new/futurist-3.5.0/PKG-INFO
--- old/futurist-3.4.0/PKG-INFO 2026-06-26 15:33:06.589815900 +0200
+++ new/futurist-3.5.0/PKG-INFO 2026-08-20 17:19:40.599634400 +0200
@@ -1,6 +1,6 @@
Metadata-Version: 2.4
Name: futurist
-Version: 3.4.0
+Version: 3.5.0
Summary: Useful additions to futures, from the future.
Author-email: OpenStack <[email protected]>
License: Apache-2.0
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/futurist-3.4.0/doc/source/reference/index.rst
new/futurist-3.5.0/doc/source/reference/index.rst
--- old/futurist-3.4.0/doc/source/reference/index.rst 2026-06-26
15:32:16.000000000 +0200
+++ new/futurist-3.5.0/doc/source/reference/index.rst 2026-08-20
17:19:07.489014100 +0200
@@ -26,6 +26,19 @@
:members:
:special-members: __init__
+~~~~~~~~~~~~~~~~~
+Delayed Executors
+~~~~~~~~~~~~~~~~~
+
+.. autoclass:: futurist.DelayedExecutorMixinBase
+ :members:
+
+.. autoclass:: futurist.DelayedExecutorMixin
+ :members:
+
+.. autoclass:: futurist.GreenDelayedExecutorMixin
+ :members:
+
-------
Futures
-------
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/futurist-3.4.0/doc/source/user/examples.rst
new/futurist-3.5.0/doc/source/user/examples.rst
--- old/futurist-3.4.0/doc/source/user/examples.rst 2026-06-26
15:32:16.000000000 +0200
+++ new/futurist-3.5.0/doc/source/user/examples.rst 2026-08-20
17:19:07.489014100 +0200
@@ -96,6 +96,49 @@
hello
+------------------------------------------------------
+Creating and using a delayed (time-deferred) executor
+------------------------------------------------------
+
+``DelayedExecutorMixin`` adds ``submit_after(delay, fn, …)`` to any futurist
+executor. The scheduler keeps a single background thread and a min-heap of
+deadlines, so arbitrarily many delayed tasks share just one extra thread.
+
+::
+
+ import time
+
+ import futurist
+
+
+ class DelayedThreadExecutor(
+ futurist.DelayedExecutorMixin,
+ futurist.ThreadPoolExecutor,
+ ):
+ pass
+
+
+ def greet(name):
+ return f"Hello, {name}!"
+
+
+ e = DelayedThreadExecutor(max_workers=4)
+
+ # Schedule greet() to run 0.5 seconds from now.
+ fut = e.submit_after(0.5, greet, "world")
+
+ # The future can be cancelled at any point before the deadline.
+ # fut.cancel()
+
+ print(fut.result()) # blocks until the task completes
+ e.shutdown()
+
+**Expected output:**
+
+::
+
+ Hello, world!
+
-------------------------------------------
Creating and using a process-based executor
-------------------------------------------
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/futurist-3.4.0/doc/source/user/features.rst
new/futurist-3.5.0/doc/source/user/features.rst
--- old/futurist-3.4.0/doc/source/user/features.rst 2026-06-26
15:32:16.000000000 +0200
+++ new/futurist-3.5.0/doc/source/user/features.rst 2026-08-20
17:19:07.489014100 +0200
@@ -19,6 +19,18 @@
execution statistics. It returns instances
of :py:class:`.futurist.Future` objects.
+Delayed Execution
+-----------------
+
+* A :py:class:`.futurist.DelayedExecutorMixin` that can be combined with any
+ :class:`~concurrent.futures.Executor` subclass to add a
+ ``submit_after(delay, fn, *args, **kwargs)`` method. A single background
+ scheduler thread maintains a min-heap of pending tasks ordered by deadline;
+ tasks can be cancelled while waiting via the returned
+ :class:`~concurrent.futures.Future`. Use
+ :py:class:`.futurist.GreenDelayedExecutorMixin` when working with eventlet-
+ based executors.
+
Periodics
---------
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/futurist-3.4.0/futurist/__init__.py
new/futurist-3.5.0/futurist/__init__.py
--- old/futurist-3.4.0/futurist/__init__.py 2026-06-26 15:32:16.000000000
+0200
+++ new/futurist-3.5.0/futurist/__init__.py 2026-08-20 17:19:07.489014100
+0200
@@ -20,7 +20,10 @@
from futurist._futures import CancelledError
from futurist._futures import TimeoutError
+from futurist._futures import DelayedExecutorMixin
+from futurist._futures import DelayedExecutorMixinBase
from futurist._futures import DynamicThreadPoolExecutor
+from futurist._futures import GreenDelayedExecutorMixin
from futurist._futures import GreenThreadPoolExecutor
from futurist._futures import ProcessPoolExecutor
from futurist._futures import SynchronousExecutor
@@ -41,6 +44,9 @@
'SynchronousExecutor',
'ThreadPoolExecutor',
'DynamicThreadPoolExecutor',
+ 'DelayedExecutorMixin',
+ 'GreenDelayedExecutorMixin',
'RejectedSubmission',
'ExecutorStatistics',
+ 'DelayedExecutorMixinBase',
]
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/futurist-3.4.0/futurist/_futures.py
new/futurist-3.5.0/futurist/_futures.py
--- old/futurist-3.4.0/futurist/_futures.py 2026-06-26 15:32:16.000000000
+0200
+++ new/futurist-3.5.0/futurist/_futures.py 2026-08-20 17:19:07.490014000
+0200
@@ -14,13 +14,16 @@
from __future__ import annotations
+import abc
from collections.abc import Callable
import functools
+import heapq
import logging
+import multiprocessing.context
import queue
import threading
import time
-from typing import Any, ParamSpec, Self, TypeVar
+from typing import Any, Generic, ParamSpec, Self, TypeVar
from concurrent import futures as _futures
from concurrent.futures import process as _process
@@ -33,6 +36,7 @@
_P = ParamSpec('_P')
_R = TypeVar('_R')
+T = TypeVar('T')
LOG = logging.getLogger(__name__)
@@ -463,7 +467,7 @@
def __init__(
self,
max_workers: int | None = None,
- mp_context: Any = None,
+ mp_context: multiprocessing.context.BaseContext | None = None,
) -> None:
if max_workers is None:
max_workers = _utils.get_optimal_process_count()
@@ -813,3 +817,402 @@
'cancelled': self._cancelled,
}
)
+
+
+class DelayedExecutorMixinBase(_futures.Executor, abc.ABC, Generic[T]):
+ """Mixin that adds submit_after(delay, fn, ...) to any Executor subclass.
+
+ Maintains a single background scheduler thread that keeps a min-heap of
+ pending tasks ordered by deadline. When a task's deadline arrives the
+ scheduler submits it to the underlying executor. Only one extra thread is
+ required regardless of how many delayed tasks are pending, making this
+ suitable for use with native threads where thread creation is expensive.
+
+ Concrete subclasses must implement the factory / threading primitives:
+ ``_get_condition_object``, ``_get_future_object``, ``_start_thread``,
+ ``_join_thread``, and ``_is_scheduler_alive``.
+
+ .. note::
+
+ When combining this mixin with
+ :class:`~futurist.ProcessPoolExecutor`, you **must** use the
+ ``spawn`` multiprocessing start method (e.g.
+ ``mp_context=multiprocessing.get_context('spawn')``). The
+ scheduler runs a background thread in the parent process; using
+ the ``fork`` or ``forkserver`` start methods can leave
+ threading primitives (conditions, locks) in a corrupted state in
+ child worker processes, leading to hangs or deadlocks.
+ """
+
+ class Task(Generic[_R]):
+ """A single delayed task entry in the scheduler priority queue.
+
+ Instances are ordered by ``deadline`` so that the min-heap in
+ :class:`DelayedExecutorMixinBase` always pops the task whose
+ deadline is closest.
+ """
+
+ def __init__(
+ self,
+ fn: Callable[..., _R],
+ args: tuple[Any, ...],
+ kwargs: dict[str, Any],
+ future: Future[_R],
+ delay: float,
+ ) -> None:
+ self.fn = fn
+ self.args = args
+ self.kwargs = kwargs
+ self.future = future
+ self.deadline = time.monotonic() + delay
+
+ @property
+ def _remaining_delay(self) -> float:
+ return self.deadline - time.monotonic()
+
+ def __repr__(self) -> str:
+ return (
+ f"Task={self.fn}, "
+ f"remaining_delay={self._remaining_delay}, "
+ f"future={self.future}"
+ )
+
+ def __lt__(self, other: object) -> bool:
+ if not isinstance(other, DelayedExecutorMixinBase.Task):
+ return NotImplemented
+ return self.deadline < other.deadline
+
+ class _SentinelTask(Task[Any]):
+ """Sentinel pushed to the queue to signal the scheduler to exit.
+
+ Inherits :class:`Task` so the scheduler queue can be typed as
+ ``list[Task]`` and ordering falls out naturally from deadline
+ comparison. Setting ``deadline`` to ``float('inf')`` guarantees
+ the sentinel always sorts after every real task, so all pending
+ work is drained before the scheduler thread notices the shutdown
+ signal.
+ """
+
+ def __init__(self) -> None:
+ self.deadline = float("inf")
+
+ def __repr__(self) -> str:
+ return "SentinelTask"
+
+ def __init__(self, *args: Any, **kwargs: Any) -> None:
+ super().__init__(*args, **kwargs)
+
+ self._queue_changed = self._get_condition_object()
+ # A min-heap of Task instances (including the _SentinelTask
+ # sentinel, which inherits Task and always sorts last).
+ self._queue: list[DelayedExecutorMixinBase.Task[Any]] = []
+
+ self._sentinel = self._SentinelTask()
+
+ self._shutdown_requested = False
+ self._cancel_futures_on_shutdown = False
+
+ self._scheduler = self._start_thread(self._schedule)
+
+ def _task_wrapper(self, task: Task[Any]) -> None:
+ try:
+ task.future.set_result(task.fn(*task.args, **task.kwargs))
+ except BaseException as e:
+ task.future.set_exception(e)
+
+ def _schedule(self) -> None:
+ """Run the scheduler, submitting delayed tasks as they become due."""
+ while True:
+ task_to_submit: DelayedExecutorMixinBase.Task[Any] | None = None
+ with self._queue_changed:
+ LOG.debug("Waiting for task")
+ self._queue_changed.wait_for(lambda: bool(self._queue))
+ task: DelayedExecutorMixinBase.Task[Any] = heapq.heappop(
+ self._queue
+ )
+ LOG.debug("%s received", task)
+
+ if task is self._sentinel:
+ # We are being shutdown. As the sentinel always sorts
+ # after every real task we know all pending tasks have
+ # been submitted to the inner executor. Shut down the
+ # inner executor now so resources are freed whether or
+ # not the caller passes wait=True.
+ LOG.debug("Sentinel task received, exiting scheduler")
+ super().shutdown(
+ wait=False,
+ cancel_futures=self._cancel_futures_on_shutdown,
+ )
+ break
+
+ # task is now the one with the closest deadline
+ if task.future.cancelled():
+ # task is cancelled, just start over
+ LOG.debug("%s was cancelled while in the queue", task)
+ continue
+
+ # wait for the deadline or an outside trigger
+ LOG.debug("Waiting for the deadline of %s", task)
+ changed = self._queue_changed.wait(task._remaining_delay)
+ LOG.debug(
+ "Awaken while waiting on %s, queue changed=%s",
+ task,
+ changed,
+ )
+
+ if changed:
+ # Something changed before the deadline.
+ if task.future.cancelled():
+ # nothing to do with this task just start over
+ LOG.debug("%s is cancelled, skipping", task)
+ continue
+ else:
+ # Reinsert the task to the queue and start over; this
+ # allows handling a newer task with a closer deadline.
+ heapq.heappush(self._queue, task)
+ LOG.debug("Queue changed, re-scheduling %s", task)
+ continue
+ else:
+ # We hit the deadline of the task so if it is not
+ # cancelled then we can execute it.
+ run_it = task.future.set_running_or_notify_cancel()
+ if not run_it:
+ # Task is cancelled just start over
+ LOG.debug("%s is cancelled, skipping", task)
+ continue
+ # Mark for submission outside the lock to avoid holding
+ # the condition while dispatching to the inner executor.
+ task_to_submit = task
+
+ # Submit outside the lock to avoid holding the condition
+ # longer than needed and to let concurrent submit_after
+ # calls proceed without waiting for this dispatch.
+ if task_to_submit is not None:
+ try:
+ self.submit(self._task_wrapper, task_to_submit)
+ LOG.debug("Task submitted %s", task_to_submit)
+ except BaseException as e:
+ LOG.debug("Failed to submit %s: %s", task_to_submit, e)
+ task_to_submit.future.set_exception(e)
+
+ LOG.debug("Scheduler thread finished")
+
+ def submit_after(
+ self,
+ delay: float,
+ fn: Callable[_P, _R],
+ *args: _P.args,
+ **kwargs: _P.kwargs,
+ ) -> _futures.Future[_R]:
+ """Schedule *fn* to run after *delay* seconds.
+
+ :param delay: Number of seconds to wait before executing *fn*.
+ Must be a non-negative value.
+ :param fn: Callable to execute after the delay.
+ :param args: Positional arguments forwarded to *fn*.
+ :param kwargs: Keyword arguments forwarded to *fn*.
+ :returns: A :class:`~concurrent.futures.Future` that will be
+ resolved with the return value of *fn*, or with an exception
+ if *fn* raises. The future can be cancelled while the task
+ is still waiting for its deadline.
+ :raises RuntimeError: If the executor has already been shut down.
+ :raises ValueError: If *delay* is negative.
+ """
+ if delay < 0:
+ raise ValueError(
+ f"delay must be a non-negative number, got {delay!r}"
+ )
+ with self._queue_changed:
+ if self._shutdown_requested:
+ raise RuntimeError(
+ "Cannot schedule new futures after being shutdown"
+ )
+
+ task = self.Task(
+ fn, args, kwargs, self._get_future_object(), delay
+ )
+ heapq.heappush(self._queue, task)
+ self._queue_changed.notify_all()
+ LOG.debug("%s is queued", task)
+ return task.future
+
+ def shutdown(
+ self, wait: bool = True, *, cancel_futures: bool = False
+ ) -> None:
+ """Shutdown the executor.
+
+ :param wait: If ``True`` (the default) block until the scheduler
+ thread and all submitted work have completed.
+ :param cancel_futures: If ``True``, cancel any delayed tasks that
+ are still waiting for their deadline before shutting down.
+ """
+ LOG.debug("Shutdown requested")
+ with self._queue_changed:
+ self._cancel_futures_on_shutdown = cancel_futures
+ if cancel_futures:
+ for item in self._queue:
+ if item is not self._sentinel:
+ item.future.cancel()
+ if not self._shutdown_requested:
+ # Ensure that our thread wakes at least one more time to allow
+ # it to exit by queuing up a sentinel task after the shutdown
+ # condition is set. This task won't be executed.
+ heapq.heappush(self._queue, self._sentinel)
+ # We only want to queue 1 sentinel even if multiple shutdown
+ # calls happen.
+ self._shutdown_requested = True
+ self._queue_changed.notify_all()
+ LOG.debug("Shutdown signalled to the scheduler thread")
+
+ # If wait is set we need to wait for our sentinel to be processed and
+ # therefore our thread to exit. The scheduler thread shuts down the
+ # inner executor when it processes the sentinel; calling
+ # super().shutdown(wait=True) here then waits for any still-running
+ # inner-executor workers to finish.
+ if wait:
+ LOG.debug("Waiting for the scheduler thread to finish")
+ self._join_thread(self._scheduler)
+ LOG.debug("Waiting for the internal executor to finish")
+ super().shutdown(wait=wait, cancel_futures=cancel_futures)
+ # With wait=False we return immediately. The scheduler thread is
+ # still running, will submit any remaining delayed tasks to the
+ # inner executor, and will then shut down the inner executor itself.
+
+ @property
+ def alive(self) -> bool:
+ """``True`` while the scheduler or inner executor is still active.
+ Returns ``False`` only once both have fully stopped. The
+ ``or super().alive`` term is defensive: in normal operation the
+ scheduler calls ``super().shutdown()`` before exiting so
+ ``super().alive`` is already ``False`` when the scheduler dies, but
+ if the scheduler exits abnormally without that call, the inner
+ executor may still be running and this ensures we report it.
+
+ Note: ``alive`` is not defined on the stdlib
+ :class:`~concurrent.futures.Executor` base class, but every futurist
+ executor exposes it. This mixin is designed to be combined with a
+ futurist executor, so ``super().alive`` will always resolve correctly
+ at runtime even though the static type checker cannot verify it.
+ """
+ return self._is_scheduler_alive() or super().alive # type:
ignore[misc]
+
+ @abc.abstractmethod
+ def _get_condition_object(self) -> threading.Condition:
+ raise NotImplementedError
+
+ @abc.abstractmethod
+ def _get_future_object(self) -> Future[Any]:
+ raise NotImplementedError
+
+ @abc.abstractmethod
+ def _start_thread(self, fn: Callable[[], None]) -> T:
+ raise NotImplementedError
+
+ @abc.abstractmethod
+ def _join_thread(self, t: T) -> None:
+ raise NotImplementedError
+
+ @abc.abstractmethod
+ def _is_scheduler_alive(self) -> bool:
+ raise NotImplementedError
+
+
+class GreenDelayedExecutorMixin(DelayedExecutorMixinBase[Any]):
+ """Variant of :class:`DelayedExecutorMixinBase` that uses eventlet
+ primitives.
+
+ This allows the mixin to work correctly in environments where eventlet is
+ present but the standard library has *not* been monkey-patched: all
+ threading primitives (conditions, threads, futures) are substituted with
+ their eventlet equivalents so that the scheduler integrates properly with
+ the eventlet event loop.
+
+ Example usage::
+
+ class DelayedGreenExecutor(
+ GreenDelayedExecutorMixin, GreenThreadPoolExecutor
+ ):
+ pass
+
+
+ executor = DelayedGreenExecutor(max_workers=4)
+ fut = executor.submit_after(0.5, my_func, arg1, kwarg=val)
+ result = fut.result()
+ executor.shutdown()
+ """
+
+ @staticmethod
+ def _get_condition_object() -> threading.Condition:
+ if _green.threading is None:
+ raise RuntimeError(
+ "GreenDelayedExecutorMixin requires eventlet to be installed"
+ )
+ return _green.threading.condition_object()
+
+ @staticmethod
+ def _get_future_object() -> GreenFuture:
+ return GreenFuture()
+
+ @staticmethod
+ def _start_thread(fn: Callable[[], None]) -> Any:
+ if _green.spawn is None:
+ raise RuntimeError(
+ "GreenDelayedExecutorMixin requires eventlet to be installed"
+ )
+ return _green.spawn(fn)
+
+ @staticmethod
+ def _join_thread(t: Any) -> None:
+ t.wait()
+
+ def _is_scheduler_alive(self) -> bool:
+ # GreenThread exposes a ``dead`` attribute rather than ``is_alive()``.
+ return not self._scheduler.dead
+
+
+class DelayedExecutorMixin(DelayedExecutorMixinBase[threading.Thread]):
+ """Variant of :class:`DelayedExecutorMixinBase` that uses native threads.
+
+ Use this class together with any standard-library
+ :class:`~concurrent.futures.Executor` subclass (e.g.
+ :class:`~futurist.ThreadPoolExecutor`) to add
+ :meth:`~DelayedExecutorMixinBase.submit_after` support backed by a single
+ daemon scheduler thread.
+
+ Example usage::
+
+ class DelayedThreadExecutor(DelayedExecutorMixin, ThreadPoolExecutor):
+ pass
+
+
+ executor = DelayedThreadExecutor(max_workers=4)
+ fut = executor.submit_after(0.5, my_func, arg1, kwarg=val)
+ result = fut.result()
+ executor.shutdown()
+ """
+
+ @staticmethod
+ def _get_condition_object() -> threading.Condition:
+ return threading.Condition()
+
+ @staticmethod
+ def _get_future_object() -> Future[Any]:
+ return Future()
+
+ @staticmethod
+ def _start_thread(fn: Callable[[], None]) -> threading.Thread:
+ t = threading.Thread(
+ target=fn,
+ name="futurist-delayed-scheduler",
+ daemon=True,
+ )
+ t.start()
+ return t
+
+ @staticmethod
+ def _join_thread(t: threading.Thread) -> None:
+ t.join()
+
+ def _is_scheduler_alive(self) -> bool:
+ t: threading.Thread = self._scheduler
+ return t.is_alive()
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/futurist-3.4.0/futurist/_green.py
new/futurist-3.5.0/futurist/_green.py
--- old/futurist-3.4.0/futurist/_green.py 2026-06-26 15:32:16.000000000
+0200
+++ new/futurist-3.5.0/futurist/_green.py 2026-08-20 17:19:07.490014000
+0200
@@ -28,8 +28,10 @@
from eventlet import queue as greenqueue
from eventlet.green import threading as greenthreading
+ from eventlet import greenthread
except ImportError:
- greenpatcher, greenpool, greenqueue, greenthreading = (
+ greenpatcher, greenpool, greenqueue, greenthreading, greenthread = (
+ None,
None,
None,
None,
@@ -42,6 +44,7 @@
Pool = greenpool.GreenPool
Queue = greenqueue.Queue
is_monkey_patched = greenpatcher.is_monkey_patched
+ spawn = greenthread.spawn
class GreenThreading:
@staticmethod
@@ -71,6 +74,8 @@
def is_monkey_patched(mod: str) -> bool:
return False
+ spawn = None
+
class GreenWorker:
def __init__(
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/futurist-3.4.0/futurist/tests/test_executors.py
new/futurist-3.5.0/futurist/tests/test_executors.py
--- old/futurist-3.4.0/futurist/tests/test_executors.py 2026-06-26
15:32:16.000000000 +0200
+++ new/futurist-3.5.0/futurist/tests/test_executors.py 2026-08-20
17:19:07.491014200 +0200
@@ -34,6 +34,14 @@
return 1
+def returns_args(foo, bar):
+ return foo + bar
+
+
+def returns_clock():
+ return time.monotonic()
+
+
def blows_up():
raise RuntimeError("no worky")
@@ -410,3 +418,365 @@
created_worker = mock_create_thread.return_value
created_worker.start.assert_called_with()
self.assertEqual(13, created_worker.start.call_count)
+
+
+class _GreenDelayedExecutor(
+ futurist.GreenDelayedExecutorMixin, futurist.GreenThreadPoolExecutor
+):
+ pass
+
+
+class _ThreadDelayedExecutor(
+ futurist.DelayedExecutorMixin, futurist.ThreadPoolExecutor
+):
+ pass
+
+
+class TestDelayedExecutorMixin(testscenarios.TestWithScenarios, base.TestCase):
+ executor_cls: type
+ scenarios = [
+ ('green', {'executor_cls': _GreenDelayedExecutor}),
+ ('thread', {'executor_cls': _ThreadDelayedExecutor}),
+ ]
+
+ def setUp(self):
+ super().setUp()
+ self.executor = self.executor_cls(max_workers=1)
+ self.addCleanup(self.executor.shutdown, wait=True)
+
+ def test_tasks_ordered_by_earliest_deadline(self):
+ def task(delay):
+ return self.executor.Task(lambda: None, (), {}, None, delay)
+
+ t10 = task(10)
+ t9 = task(9)
+ t11 = task(11)
+ t8 = task(8)
+ t1 = task(1)
+
+ self.assertEqual(
+ [t1, t8, t9, t10, t11], sorted([t10, t9, t11, t8, t1])
+ )
+
+ def test_execute_one(self):
+ task = self.executor.submit_after(0.1, returns_args, 13, bar=42)
+
+ self.assertEqual(13 + 42, task.result())
+
+ def test_execute_two_sequential(self):
+ task1 = self.executor.submit_after(0.1, returns_clock)
+ task2 = self.executor.submit_after(0.1, returns_clock)
+
+ t1_at = task1.result()
+ t2_at = task2.result()
+ self.assertLess(t1_at, t2_at)
+
+ def test_submit_second_while_waiting_on_first_sequential(self):
+ task1 = self.executor.submit_after(0.5, returns_clock)
+ time.sleep(0.2)
+ task2 = self.executor.submit_after(0.5, returns_clock)
+
+ t1_at = task1.result()
+ t2_at = task2.result()
+ self.assertLess(t1_at, t2_at)
+
+ def test_submit_second_preempts_first(self):
+ task1 = self.executor.submit_after(0.3, returns_clock)
+ task2 = self.executor.submit_after(0.1, returns_clock)
+
+ t1_at = task1.result()
+ t2_at = task2.result()
+ self.assertLess(t2_at, t1_at)
+
+ def test_submit_preempts_first_while_waiting(self):
+ task1 = self.executor.submit_after(0.5, returns_clock)
+ time.sleep(0.2)
+ task2 = self.executor.submit_after(0.1, returns_clock)
+
+ t1_at = task1.result()
+ t2_at = task2.result()
+ self.assertLess(t2_at, t1_at)
+
+ def test_zero_delay_executes(self):
+ """A zero-delay task executes without error."""
+ task = self.executor.submit_after(0, returns_one)
+ self.assertEqual(1, task.result(timeout=5))
+
+ def test_multiple_tasks_execute_in_delay_order(self):
+ """Three tasks with distinct delays finish in deadline order."""
+ # Submit in non-deadline order to prove the scheduler sorts by
+ # deadline, not by submission order: task_long is submitted first
+ # but has the furthest deadline, so task_medium (submitted second)
+ # must execute before it.
+ task_long = self.executor.submit_after(0.3, returns_clock)
+ task_medium = self.executor.submit_after(0.2, returns_clock)
+ task_short = self.executor.submit_after(0.1, returns_clock)
+
+ t_short = task_short.result()
+ t_medium = task_medium.result()
+ t_long = task_long.result()
+
+ self.assertLess(t_short, t_medium)
+ self.assertLess(t_medium, t_long)
+
+ def test_delayed_even_with_idle_workers(self):
+ """Non-zero delay is respected even when the executor has free workers.
+
+ With max_workers=4, all worker slots are available immediately.
+ The task must still wait for its full delay before executing.
+ """
+ executor = self.executor_cls(max_workers=4)
+ self.addCleanup(executor.shutdown, wait=True)
+
+ delay = 0.2
+ start = time.monotonic()
+ task = executor.submit_after(delay, returns_clock)
+ ran_at = task.result(timeout=10)
+ elapsed = ran_at - start
+
+ self.assertGreaterEqual(
+ elapsed,
+ delay,
+ "task ran before its scheduled delay elapsed",
+ )
+
+ def test_submit_after_raises_after_shutdown(self):
+ """submit_after raises RuntimeError once the executor is shut down."""
+ # Shut down the executor created in setUp and verify a subsequent
+ # submit_after call raises RuntimeError. The addCleanup will call
+ # shutdown again, which is a safe no-op.
+ self.executor.shutdown(wait=True)
+ self.assertRaises(
+ RuntimeError, self.executor.submit_after, 0, returns_one
+ )
+
+ def test_cancel_before_deadline_skips_execution(self):
+ """Cancelling a future before its deadline prevents the callable
+ from running.
+ """
+ call_count = []
+
+ def fn():
+ call_count.append(1)
+ return 1
+
+ task = self.executor.submit_after(0.5, fn)
+ cancelled = task.cancel()
+
+ self.assertTrue(cancelled)
+ # Wait well past the deadline so the scheduler processes the
+ # cancellation even on a loaded CI system (0.5s deadline + 0.5s
+ # margin = 1.0s total).
+ time.sleep(1.0)
+ self.assertTrue(task.cancelled())
+ self.assertEqual([], call_count)
+
+ def test_exception_propagation(self):
+ """An exception raised by the callable is reflected in
+ future.exception().
+ """
+ task = self.executor.submit_after(0.1, blows_up)
+ self.assertRaises(RuntimeError, task.result)
+ self.assertIsInstance(task.exception(), RuntimeError)
+
+ def test_concurrent_submissions_all_complete(self):
+ """Many tasks submitted in rapid succession all complete."""
+ num_tasks = 20
+ futures_list = [
+ self.executor.submit_after(0.1, returns_one)
+ for _ in range(num_tasks)
+ ]
+ results = [f.result(timeout=30) for f in futures_list]
+ self.assertEqual([1] * num_tasks, results)
+
+ def test_executor_forwards_args_and_kwargs(self):
+ """The callable is invoked with the exact positional and keyword
+ args given to submit_after.
+ """
+ received = []
+
+ def capture(pos, *, kw):
+ received.append((pos, kw))
+ return pos + kw
+
+ task = self.executor.submit_after(0.1, capture, 7, kw=3)
+ result = task.result()
+
+ self.assertEqual(10, result)
+ self.assertEqual([(7, 3)], received)
+
+ def test_negative_delay_raises_value_error(self):
+ """submit_after rejects negative delay values immediately."""
+ self.assertRaises(
+ ValueError, self.executor.submit_after, -1, returns_one
+ )
+
+ def test_double_shutdown_is_safe(self):
+ """Calling shutdown() twice must not raise or deadlock."""
+ self.executor.shutdown(wait=True)
+ # Second call — must be a safe no-op.
+ self.executor.shutdown(wait=True)
+
+ def test_alive_false_after_no_wait_shutdown(self):
+ self.executor.shutdown(wait=False)
+ self.executor.shutdown(wait=True) # drain
+ self.assertFalse(self.executor.alive)
+
+ def test_scheduler_thread_is_daemon(self):
+ if not isinstance(self.executor._scheduler, threading.Thread):
+ self.skipTest(
+ "daemon attribute only applies to native threading.Thread; "
+ "eventlet GreenThreads exit with the hub unconditionally"
+ )
+ self.assertTrue(
+ self.executor._scheduler.daemon,
+ "scheduler thread must be a daemon so the process can exit "
+ "cleanly if shutdown() is never called",
+ )
+
+ def test_shutdown_cancel_futures_cancels_pending_tasks(self):
+ """shutdown(cancel_futures=True) cancels tasks still in the queue."""
+ # Use a very long delay so both tasks are guaranteed to remain in the
+ # scheduler queue (i.e. well before their deadline) when shutdown is
+ # called.
+ task1 = self.executor.submit_after(100.0, returns_one)
+ task2 = self.executor.submit_after(100.0, returns_one)
+
+ self.executor.shutdown(cancel_futures=True, wait=True)
+
+ self.assertTrue(task1.cancelled(), "task1 should have been cancelled")
+ self.assertTrue(task2.cancelled(), "task2 should have been cancelled")
+
+ def test_alive_true_before_shutdown(self):
+ """The executor is alive as soon as it is created."""
+ self.assertTrue(self.executor.alive)
+
+ def test_alive_false_immediately_after_wait_true_shutdown(self):
+ self.assertTrue(self.executor.alive)
+ self.executor.shutdown(wait=True)
+ self.assertFalse(self.executor.alive)
+
+ def test_submit_after_raises_after_wait_false_shutdown(self):
+ self.executor.shutdown(wait=False)
+ self.assertRaises(
+ RuntimeError, self.executor.submit_after, 0, returns_one
+ )
+
+ def test_task_cancelled_while_in_queue_is_skipped(self):
+ call_count = []
+
+ def fn():
+ call_count.append(1)
+
+ # quick (0.1 s) has a shorter deadline than slow (100 s).
+ # The scheduler pops quick first, leaving slow in the queue.
+ quick = self.executor.submit_after(0.1, returns_one)
+ slow = self.executor.submit_after(100.0, fn)
+
+ # Cancel slow while the scheduler is busy waiting on quick's deadline.
+ # future.cancel() does NOT notify _queue_changed, so the scheduler
+ # is not woken; slow simply sits in the queue as CANCELLED.
+ self.assertTrue(slow.cancel())
+
+ # Wait for quick to complete; the scheduler then loops and pops slow.
+ self.assertEqual(1, quick.result(timeout=5))
+
+ # Shut down and wait for the scheduler to drain fully rather than
+ # sleeping for a fixed time, which would be racy on a loaded system.
+ self.executor.shutdown(wait=True)
+
+ self.assertTrue(slow.cancelled())
+ self.assertEqual([], call_count, "slow callable must not have run")
+
+ def test_task_cancelled_during_scheduler_wakeup_is_skipped(self):
+ call_count = []
+
+ def fn():
+ call_count.append(1)
+ return 1
+
+ # task_a has a long deadline — the scheduler pops it and waits.
+ task_a = self.executor.submit_after(5.0, fn)
+ # Ensure the scheduler has had time to pop task_a and begin waiting.
+ time.sleep(0.05)
+
+ # Cancel task_a (does NOT wake the scheduler on its own).
+ self.assertTrue(task_a.cancel())
+
+ # Submitting task_b calls notify_all, waking the scheduler.
+ # It finds changed=True and task_a.future.cancelled() → True,
+ # so it takes the L963-966 branch: continue without executing task_a.
+ task_b = self.executor.submit_after(0.1, returns_one)
+
+ self.assertEqual(1, task_b.result(timeout=5))
+ self.assertTrue(task_a.cancelled())
+ self.assertEqual([], call_count, "task_a callable must not have run")
+
+ def test_running_task_not_cancelled_by_cancel_futures(self):
+ executor = self.executor_cls(max_workers=2)
+ self.addCleanup(executor.shutdown, wait=True)
+
+ # Use futures compatible with the executor's threading model as
+ # synchronisation points. In the green variant these are
+ # GreenFuture objects whose .result() yields to the eventlet hub,
+ # so the test and the worker interleave correctly without
+ # time.sleep() blocking the entire event loop.
+ running_signal = executor._get_future_object()
+ hold_signal = executor._get_future_object()
+
+ # Release hold_signal in cleanup so the worker is never left dangling
+ # if the test fails before the explicit set_result() call below.
+ def _release_hold():
+ try:
+ hold_signal.set_result(None)
+ except Exception:
+ pass # already set by the test body
+
+ self.addCleanup(_release_hold)
+
+ def controlled_task():
+ running_signal.set_result(True) # signal: task is now running
+ hold_signal.result(timeout=10) # park until the test proceeds
+
+ slow_future = executor.submit_after(0.0, controlled_task)
+ # Block until controlled_task has actually started executing. In the
+ # green variant GreenFuture.result() yields to the hub so the green
+ # worker can run and signal without blocking the event loop.
+ running_signal.result(timeout=10)
+
+ # At this point slow_future is in RUNNING state: the scheduler called
+ # set_running_or_notify_cancel() on it before dispatching to the inner
+ # executor, and controlled_task() is still parked at hold_signal.
+ # It is no longer in the delay queue, so cancel_futures=True cannot
+ # affect it.
+ self.assertTrue(
+ slow_future.running(),
+ "slow_future must be in RUNNING state before shutdown is called",
+ )
+
+ # Release the parked task so that shutdown(wait=True) can complete.
+ hold_signal.set_result(None)
+ executor.shutdown(cancel_futures=True, wait=True)
+
+ self.assertFalse(slow_future.cancelled())
+ self.assertTrue(slow_future.done())
+
+ def test_inner_submit_failure_propagated_to_future(self):
+ with mock.patch.object(
+ self.executor,
+ 'submit',
+ side_effect=RuntimeError("injected dispatch failure"),
+ ):
+ future = self.executor.submit_after(0.0, returns_one)
+ exc = future.exception(timeout=5)
+
+ self.assertIsInstance(exc, RuntimeError)
+ self.assertIn("injected dispatch failure", str(exc))
+
+ def test_sentinel_task_has_no_future_attribute(self):
+ self.assertFalse(
+ hasattr(self.executor._sentinel, 'future'),
+ "_SentinelTask must not define a .future attribute — "
+ "the cancel-loop guard `if item is not self._sentinel` "
+ "exists precisely because it is absent.",
+ )
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/futurist-3.4.0/futurist/tests/test_waiters.py
new/futurist-3.5.0/futurist/tests/test_waiters.py
--- old/futurist-3.4.0/futurist/tests/test_waiters.py 2026-06-26
15:32:16.000000000 +0200
+++ new/futurist-3.5.0/futurist/tests/test_waiters.py 2026-08-20
17:19:07.492014200 +0200
@@ -10,6 +10,7 @@
# License for the specific language governing permissions and limitations
# under the License.
+import multiprocessing
import time
import eventlet
@@ -71,7 +72,14 @@
'process',
{
'executor_cls': futurist.ProcessPoolExecutor,
- 'executor_kwargs': {},
+ 'executor_kwargs': {
+ # Use 'spawn' rather than the platform default 'fork'.
+ # This test module imports eventlet at module level, which
+ # creates background threads; forking a multithreaded
+ # process can deadlock the child due to inherited locks.
+ # 'spawn' starts a clean interpreter and avoids this.
+ 'mp_context': multiprocessing.get_context('spawn'),
+ },
'use_eventlet_sleep': False,
},
),
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/futurist-3.4.0/futurist.egg-info/PKG-INFO
new/futurist-3.5.0/futurist.egg-info/PKG-INFO
--- old/futurist-3.4.0/futurist.egg-info/PKG-INFO 2026-06-26
15:33:06.000000000 +0200
+++ new/futurist-3.5.0/futurist.egg-info/PKG-INFO 2026-08-20
17:19:40.564633600 +0200
@@ -1,6 +1,6 @@
Metadata-Version: 2.4
Name: futurist
-Version: 3.4.0
+Version: 3.5.0
Summary: Useful additions to futures, from the future.
Author-email: OpenStack <[email protected]>
License: Apache-2.0
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/futurist-3.4.0/futurist.egg-info/SOURCES.txt
new/futurist-3.5.0/futurist.egg-info/SOURCES.txt
--- old/futurist-3.4.0/futurist.egg-info/SOURCES.txt 2026-06-26
15:33:06.000000000 +0200
+++ new/futurist-3.5.0/futurist.egg-info/SOURCES.txt 2026-08-20
17:19:40.586634200 +0200
@@ -48,6 +48,7 @@
futurist/tests/test_waiters.py
releasenotes/notes/add-mp_context-arg-db2ca2d0b3f24b93.yaml
releasenotes/notes/add-reno-996dd44974d53238.yaml
+releasenotes/notes/delayed-executor-mixin-a4f3c8d1b9e20571.yaml
releasenotes/notes/deprecate-eventlet-d96deb8b97930fef.yaml
releasenotes/notes/drop-python-3.10-1ffc51bb720262e3.yaml
releasenotes/notes/drop-python27-support-5757997ea990b7ca.yaml
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/futurist-3.4.0/futurist.egg-info/pbr.json
new/futurist-3.5.0/futurist.egg-info/pbr.json
--- old/futurist-3.4.0/futurist.egg-info/pbr.json 2026-06-26
15:33:06.000000000 +0200
+++ new/futurist-3.5.0/futurist.egg-info/pbr.json 2026-08-20
17:19:40.570633600 +0200
@@ -1 +1 @@
-{"git_version": "85f1968", "is_release": true}
\ No newline at end of file
+{"git_version": "b429036", "is_release": true}
\ No newline at end of file
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/futurist-3.4.0/pyproject.toml
new/futurist-3.5.0/pyproject.toml
--- old/futurist-3.4.0/pyproject.toml 2026-06-26 15:32:16.000000000 +0200
+++ new/futurist-3.5.0/pyproject.toml 2026-08-20 17:19:07.492014200 +0200
@@ -45,6 +45,7 @@
[tool.ruff.lint]
select = ["E4", "E5", "E7", "E9", "F", "G", "LOG", "S", "UP"]
+external = ["H"]
ignore = [
"S101", # asserts are only used for type narrowing
]
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore'
old/futurist-3.4.0/releasenotes/notes/delayed-executor-mixin-a4f3c8d1b9e20571.yaml
new/futurist-3.5.0/releasenotes/notes/delayed-executor-mixin-a4f3c8d1b9e20571.yaml
---
old/futurist-3.4.0/releasenotes/notes/delayed-executor-mixin-a4f3c8d1b9e20571.yaml
1970-01-01 01:00:00.000000000 +0100
+++
new/futurist-3.5.0/releasenotes/notes/delayed-executor-mixin-a4f3c8d1b9e20571.yaml
2026-08-20 17:19:07.493014000 +0200
@@ -0,0 +1,32 @@
+---
+features:
+ - |
+ Adds ``DelayedExecutorMixin`` and ``GreenDelayedExecutorMixin`` mixin
+ classes that extend any :class:`~concurrent.futures.Executor` subclass
+ with a ``submit_after(delay, fn, *args, **kwargs)`` method.
+
+ A single background scheduler thread maintains a min-heap of pending
+ tasks ordered by their deadline. When a task's deadline arrives the
+ scheduler submits it to the underlying executor. Tasks can be
+ cancelled while waiting for their deadline via the returned
+ :class:`~concurrent.futures.Future`.
+
+ Use ``DelayedExecutorMixin`` with native-thread executors (e.g.
+ :class:`~futurist.ThreadPoolExecutor`) and ``GreenDelayedExecutorMixin``
+ with eventlet-based executors (e.g.
+ :class:`~futurist.GreenThreadPoolExecutor`). Each ``submit_after``
+ call accepts its own *delay* value, so different tasks submitted to the
+ same executor can have different delays.
+
+ Example::
+
+ class DelayedThreadExecutor(
+ futurist.DelayedExecutorMixin,
+ futurist.ThreadPoolExecutor,
+ ):
+ pass
+
+ executor = DelayedThreadExecutor(max_workers=4)
+ fut = executor.submit_after(0.5, my_func, arg1, kwarg=val)
+ result = fut.result()
+ executor.shutdown()
