gh-160760: asyncio: run newly scheduled callbacks eagerly before polling for I/O

Doc/library/asyncio-eventloop.rst

@@ -130,7 +130,12 @@ Running and stopping the loop
    the loop will run the current batch of callbacks and then exit.
    Note that new callbacks scheduled by callbacks will not run in this
    case; instead, they will run the next time :meth:`run_forever` or
-   :meth:`run_until_complete` is called.
+   :meth:`run_until_complete` is called.  If
+   :ref:`eager callback processing <asyncio-eager-callbacks>` ran in
+   the current or the previous iteration of the loop, the loop runs one
+   more iteration after :meth:`stop` is called, which polls for I/O and
+   runs all ready callbacks, including those scheduled after
+   :meth:`stop`.
 
 .. method:: loop.stop()
 
@@ -213,7 +218,10 @@ Scheduling callbacks
 .. method:: loop.call_soon(callback, *args, context=None)
 
    Schedule the *callback* :term:`callback` to be called with
-   *args* arguments at the next iteration of the event loop.
+   *args* arguments after the callbacks that are already scheduled.
+   If it is scheduled by another callback, it may run in the same
+   iteration of the event loop, see
+   :ref:`eager callback processing <asyncio-eager-callbacks>`.
 
    Return an instance of :class:`asyncio.Handle`,
    which can be used later to cancel the callback.
@@ -1578,6 +1586,94 @@ Enabling debug mode
    The :ref:`debug mode of asyncio <asyncio-debug-mode>`.
 
 
+.. _asyncio-eager-callbacks:
+
+Eager callback processing
+^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Each iteration of the event loop polls for I/O events and then runs the
+callbacks that are ready.  Callbacks scheduled by these callbacks, for
+example tasks woken up by a completed future, do not have to wait for
+the next I/O poll: the loop keeps running them, in the order they were
+scheduled, for a short time.  This saves one I/O poll per hop and speeds
+up code that runs many short callbacks.
+
+Eager processing in the current iteration ends when:
+
+* there are no more ready callbacks;
+* *eager_max_callbacks* additional callbacks have run;
+* more than *eager_max_time* seconds have passed; the time is checked
+  between batches of callbacks, and a batch ends at the latest where the
+  callbacks scheduled while it runs begin;
+* :func:`asyncio.sleep` is called with a delay of 0 or less; the rest of
+  the current batch then runs after the next I/O poll; or
+* :meth:`loop.stop` is called; the loop then runs one more iteration,
+  which polls for I/O and runs the remaining ready callbacks without
+  eager processing, before :meth:`loop.run_forever` returns.
+
+I/O events and timers are not processed while eager processing runs,
+so larger limits can delay them.
+
+.. note::
+
+   Eager processing can end before all ready callbacks have run: when
+   ``asyncio.sleep(0)`` is called or a limit is reached, the callbacks
+   still queued are postponed to the next iteration of the loop, after
+   the I/O poll.  How far a chain of callbacks gets before pending I/O
+   events and timers are handled therefore depends on its position in
+   the queue and on timing.  Application code should not rely on this
+   order, but some unit tests do.  Such tests can disable eager
+   processing::
+
+      loop.set_settings(asyncio.EventLoopSettings(eager_max_time=0,
+                                                  eager_max_callbacks=0))
+
+.. class:: EventLoopSettings(*, eager_max_time, eager_max_callbacks)
+
+   Immutable settings of an event loop.
+
+   *eager_max_time* is the time in seconds after which eager processing
+   in one iteration of the loop stops.  It is checked between batches of
+   callbacks, so eager processing can take longer if the callbacks are
+   slow.  It must be a non-negative finite :class:`int` or :class:`float`;
+   ``0`` disables eager processing.
+
+   *eager_max_callbacks* is the maximum number of callbacks run eagerly
+   in one iteration of the loop.  It must be a non-negative integer;
+   ``0`` disables eager processing.
+
+   :exc:`TypeError` is raised if a value has a wrong type and
+   :exc:`ValueError` if it is out of range.
+
+   .. attribute:: DEFAULT
+
+      The default settings of a new event loop:
+      ``EventLoopSettings(eager_max_time=0.0001, eager_max_callbacks=100)``.
+
+   .. versionadded:: next
+
+.. method:: loop.get_settings()
+
+   Return the :class:`EventLoopSettings` of the event loop.
+
+   The default value is :attr:`EventLoopSettings.DEFAULT`.
+
+   .. versionadded:: next
+
+.. method:: loop.set_settings(settings)
+
+   Set the :class:`EventLoopSettings` of the event loop.
+
+   The new settings are used starting from the next iteration of the loop.
+
+   To change a single setting, use :func:`copy.replace`::
+
+      loop.set_settings(copy.replace(asyncio.EventLoopSettings.DEFAULT,
+                                     eager_max_callbacks=1000))
+
+   .. versionadded:: next
+
+
 Running subprocesses
 ^^^^^^^^^^^^^^^^^^^^
 

Doc/library/asyncio-llapi-index.rst

@@ -78,6 +78,19 @@ See also the main documentation section about the
       - Get the current debug mode.
 
 
+.. rubric:: Loop settings
+.. list-table::
+    :widths: 50 50
+    :class: full-width-table
+
+    * - :meth:`loop.set_settings`
+      - Set the :ref:`eager callback processing <asyncio-eager-callbacks>`
+        settings.
+
+    * - :meth:`loop.get_settings`
+      - Get the current :class:`EventLoopSettings`.
+
+
 .. rubric:: Scheduling Callbacks
 .. list-table::
     :widths: 50 50

Doc/library/asyncio-task.rst

@@ -511,6 +511,9 @@ Sleeping
    Setting the delay to 0 provides an optimized path to allow other
    tasks to run. This can be used by long-running functions to avoid
    blocking the event loop for the full duration of the function call.
+   It also ends :ref:`eager callback processing <asyncio-eager-callbacks>`
+   in the current iteration of the loop, so the task resumes only after
+   the loop has polled for I/O.
 
    .. _asyncio_example_sleep:
 
@@ -538,6 +541,10 @@ Sleeping
    .. versionchanged:: 3.13
       Raises :exc:`ValueError` if *delay* is :data:`~math.nan`.
 
+   .. versionchanged:: next
+      A delay of 0 or less ends eager callback processing in the current
+      iteration of the event loop.
+
 
 Running tasks concurrently
 ==========================
@@ -798,8 +805,8 @@ Timeouts
        as measured by the event loop's clock:
 
        - If ``when`` is ``None``, the timeout will never trigger.
-       - If ``when < loop.time()``, the timeout will trigger on the next
-         iteration of the event loop.
+       - If ``when < loop.time()``, the timeout will trigger as soon as
+         possible, like a callback scheduled with :meth:`loop.call_soon`.
 
         .. method:: when() -> float | None
 
@@ -1436,7 +1443,7 @@ Task object
       otherwise, return ``True``.
 
       The method arranges for a :exc:`CancelledError` exception to be thrown
-      into the wrapped coroutine on the next cycle of the event loop.
+      into the wrapped coroutine the next time the task runs.
 
       The coroutine then has a chance to clean up or even deny the
       request by suppressing the exception with a :keyword:`try` ...

Doc/whatsnew/3.16.rst

@@ -148,6 +148,15 @@ asyncio
   :class:`asyncio.Runner` class constructor.
   (Contributed by Andrew Svetlov in :gh:`128289`.)
 
+* The event loop now runs callbacks scheduled by other callbacks eagerly,
+  within a small time and count budget, before polling for I/O again.
+  This speeds up code that resolves futures or wakes tasks in chains.
+  The limits can be changed with the new :meth:`loop.set_settings()
+  <asyncio.loop.set_settings>` method and :class:`asyncio.EventLoopSettings`
+  class.  ``asyncio.sleep(0)`` ends eager processing in the current
+  iteration.  See :ref:`asyncio-eager-callbacks`.
+  (Contributed by Andrew Svetlov in :gh:`160760`.)
+
 
 bdb
 ---
@@ -1243,6 +1252,13 @@ Porting to Python 3.16
 This section lists previously described changes and other bugfixes
 that may require changes to your code.
 
+* :mod:`asyncio` event loops now run callbacks scheduled by other callbacks
+  eagerly, before polling for I/O again.  Code or tests that rely on the
+  exact order of callbacks relative to I/O events may need updating; eager
+  processing can be disabled with ``loop.set_settings(asyncio.EventLoopSettings(
+  eager_max_time=0, eager_max_callbacks=0))``.
+  (Contributed by Andrew Svetlov in :gh:`160760`.)
+
 * In :mod:`tkinter`, the *name* parameter of the
   :meth:`~tkinter.Misc.wait_variable`, :meth:`~tkinter.Misc.setvar` and
   :meth:`~tkinter.Misc.getvar` methods and the *value* parameter of

Lib/asyncio/base_events.py

@@ -434,6 +434,8 @@ def __init__(self):
         self._thread_id = None
         self._clock_resolution = time.get_clock_info('monotonic').resolution
         self._exception_handler = None
+        self._settings = events.EventLoopSettings.DEFAULT
+        self._eager_state = constants.EAGER_IDLE
         self.set_debug(coroutines._is_debug_mode())
         # The preserved state of async generator hooks.
         self._old_agen_hooks = None
@@ -667,6 +669,7 @@ def _run_forever_cleanup(self):
         loop cleanup logic.
         """
         self._stopping = False
+        self._eager_state = constants.EAGER_IDLE
         self._thread_id = None
         events._set_running_loop(None)
         self._set_coroutine_origin_tracking(False)
@@ -682,6 +685,8 @@ def run_forever(self):
             while True:
                 self._run_once()
                 if self._stopping:
+                    if self._eager_state >= 0:
+                        self._run_once()
                     break
         finally:
             self._run_forever_cleanup()
@@ -2009,10 +2014,13 @@ def _timer_handle_cancelled(self, handle):
     def _run_once(self):
         """Run one full iteration of the event loop.
 
-        This calls all currently ready callbacks, polls for I/O,
-        schedules the resulting callbacks, and finally schedules
-        'call_later' callbacks.
+        This polls for I/O, schedules the resulting callbacks and the
+        'call_later' callbacks that are due, and calls all ready
+        callbacks.  Then it calls the callbacks scheduled by them
+        eagerly, within the limits of the loop settings.
         """
+        nleft = self._eager_state
+        nready = len(self._ready)
 
         sched_count = len(self._scheduled)
         if (sched_count > _MIN_SCHEDULED_TIMER_HANDLES and
@@ -2068,11 +2076,59 @@ def _run_once(self):
 
         # This is the only place where callbacks are actually *called*.
         # All other places just add them to ready.
-        # Note: We run all currently scheduled callbacks, but not any
-        # callbacks scheduled by callbacks run this time around --
-        # they will be run the next time (after another I/O poll).
+        # Note: We run all currently scheduled callbacks.  Callbacks
+        # scheduled by them are run eagerly in batches, within the limits
+        # of the loop settings; the rest will be run the next time (after
+        # another I/O poll).
         # Use an idiom that is thread-safe without using locks.
+        self._eager_state = constants.EAGER_IDLE
+        settings = self._settings
+        if nleft >= 0 and self._stopping:
+            # The final iteration after stop() during eager processing: run
+            # what the loop would run without eager processing, and keep the
+            # callbacks scheduled for later iterations queued.
+            nleft = min(nleft, nready)
+            nnew = len(self._ready) - nready
+            self._run_handles(nleft, eager=False)
+            later = [self._ready.popleft() for i in range(nready - nleft)]
+            self._run_handles(nnew, eager=False)
+            self._ready.extendleft(reversed(later))
+            return
         ntodo = len(self._ready)
+        if nleft > 0:
+            nleft = min(nleft, ntodo)
+            self._run_handles(nleft, eager=False)
+            if self._stopping:
+                return
+            ntodo -= nleft
+            nleft = len(self._ready) - ntodo
+        self._run_handles(ntodo, eager=False)
+        if (not self._ready or self._eager_state == constants.EAGER_BREAK
+                or self._stopping or settings.eager_max_time <= 0
+                or not settings.eager_max_callbacks):
+            if nleft > 0:
+                self._eager_state = nleft
+            return
+
+        # Eager processing: run the callbacks scheduled by the callbacks
+        # above, in batches, within the limits of the loop settings.
+        deadline = (self.time() + settings.eager_max_time
+                    + self._clock_resolution)
+        nremaining = settings.eager_max_callbacks
+        while True:
+            if nleft <= 0:
+                nleft = len(self._ready)
+            ntodo = min(nleft, nremaining, constants.EAGER_BATCH_SIZE)
+            nremaining -= ntodo
+            nleft -= self._run_handles(ntodo, eager=True)
+            if (not nremaining or not self._ready
+                    or self._eager_state == constants.EAGER_BREAK
+                    or self._stopping or self.time() > deadline):
+                break
+        if nleft > 0 or self._stopping:
+            self._eager_state = nleft
+
+    def _run_handles(self, ntodo, *, eager):
         for i in range(ntodo):
             handle = self._ready.popleft()
             if handle._cancelled:
@@ -2090,7 +2146,11 @@ def _run_once(self):
                     self._current_handle = None
             else:
                 handle._run()
+            if eager and self._eager_state == constants.EAGER_BREAK:
+                ntodo = i + 1
+                break
         handle = None  # Needed to break cycles when an exception occurs.
+        return ntodo
 
     def _set_coroutine_origin_tracking(self, enabled):
         if bool(enabled) == bool(self._coroutine_origin_tracking_enabled):
@@ -2115,3 +2175,11 @@ def set_debug(self, enabled):
 
         if self.is_running():
             self.call_soon_threadsafe(self._set_coroutine_origin_tracking, enabled)
+
+    def get_settings(self):
+        return self._settings
+
+    def set_settings(self, settings):
+        if not isinstance(settings, events.EventLoopSettings):
+            raise TypeError('settings must be EventLoopSettings instance')
+        self._settings = settings

Lib/asyncio/constants.py

@@ -33,6 +33,17 @@
 # Default timeout for joining the threads in the threadpool
 THREAD_JOIN_TIMEOUT = 300
 
+# Maximum number of callbacks run during eager processing between checks
+# of the eager_max_time deadline.  A batch also ends where the callbacks
+# scheduled during it begin, so for chains of callbacks the deadline is
+# checked after every callback.
+EAGER_BATCH_SIZE = 100
+
+# Values of BaseEventLoop._eager_state other than a number of callbacks.
+EAGER_IDLE = -1
+# Set by asyncio.sleep(0) to end eager processing.
+EAGER_BREAK = -2
+
 # The enum should be here to break circular dependencies between
 # base_events and sslproto
 class _SendfileMode(enum.Enum):