diff --git a/Doc/library/asyncio-eventloop.rst b/Doc/library/asyncio-eventloop.rst
index cae09bfae699abd..3210c97f44a3a4f 100644
--- a/Doc/library/asyncio-eventloop.rst
+++ b/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
 ^^^^^^^^^^^^^^^^^^^^
 
diff --git a/Doc/library/asyncio-llapi-index.rst b/Doc/library/asyncio-llapi-index.rst
index faf67209a2b0ef5..ea52bfaf42cd15f 100644
--- a/Doc/library/asyncio-llapi-index.rst
+++ b/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
diff --git a/Doc/library/asyncio-task.rst b/Doc/library/asyncio-task.rst
index 6df87d4d71a3657..a5aaa094a278a6e 100644
--- a/Doc/library/asyncio-task.rst
+++ b/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` ...
diff --git a/Doc/whatsnew/3.16.rst b/Doc/whatsnew/3.16.rst
index e342aa71c7924ea..38502ffdf7baf8a 100644
--- a/Doc/whatsnew/3.16.rst
+++ b/Doc/whatsnew/3.16.rst
@@ -144,6 +144,15 @@ asyncio
   socket file created for *path*.
   (Contributed by Sam Bull in :gh:`94984`.)
 
+* 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
 ---
diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py
index 9eb34c5f29fb551..ec29b13f4370cc0 100644
--- a/Lib/asyncio/base_events.py
+++ b/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.
         """
+        left = 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 left >= 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.
+            left = min(left, nready)
+            nnew = len(self._ready) - nready
+            self._run_handles(left, eager=False)
+            later = [self._ready.popleft() for i in range(nready - left)]
+            self._run_handles(nnew, eager=False)
+            self._ready.extendleft(reversed(later))
+            return
         ntodo = len(self._ready)
+        if left > 0:
+            left = min(left, ntodo)
+            self._run_handles(left, eager=False)
+            if self._stopping:
+                return
+            ntodo -= left
+            left = 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 left > 0:
+                self._eager_state = left
+            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)
+        remaining = settings.eager_max_callbacks
+        while True:
+            if left <= 0:
+                left = len(self._ready)
+            ntodo = min(left, remaining, constants.EAGER_BATCH_SIZE)
+            remaining -= ntodo
+            left -= self._run_handles(ntodo, eager=True)
+            if (not remaining or not self._ready
+                    or self._eager_state == constants.EAGER_BREAK
+                    or self._stopping or self.time() > deadline):
+                break
+        if left > 0 or self._stopping:
+            self._eager_state = left
+
+    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
diff --git a/Lib/asyncio/constants.py b/Lib/asyncio/constants.py
index b60c1e4236af1fa..db0706be725f8b3 100644
--- a/Lib/asyncio/constants.py
+++ b/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):
diff --git a/Lib/asyncio/events.py b/Lib/asyncio/events.py
index 6b2d34e733a6b1a..49bb8728156782f 100644
--- a/Lib/asyncio/events.py
+++ b/Lib/asyncio/events.py
@@ -7,6 +7,7 @@
 __all__ = (
     "AbstractEventLoop",
     "AbstractServer",
+    "EventLoopSettings",
     "Handle",
     "TimerHandle",
     "get_event_loop",
@@ -18,6 +19,7 @@
 )
 
 import contextvars
+import dataclasses
 import os
 import signal
 import socket
@@ -248,6 +250,31 @@ async def __aexit__(self, *exc):
         await self.wait_closed()
 
 
+@dataclasses.dataclass(frozen=True, kw_only=True)
+class EventLoopSettings:
+    eager_max_time: float
+    eager_max_callbacks: int
+
+    def __post_init__(self):
+        if (not isinstance(self.eager_max_time, (int, float))
+                or isinstance(self.eager_max_time, bool)):
+            raise TypeError(f"eager_max_time must be an int or float, "
+                            f"got {self.eager_max_time!r}")
+        if (not isinstance(self.eager_max_callbacks, int)
+                or isinstance(self.eager_max_callbacks, bool)):
+            raise TypeError(f"eager_max_callbacks must be an int, "
+                            f"got {self.eager_max_callbacks!r}")
+        if not 0 <= self.eager_max_time <= sys.float_info.max:
+            raise ValueError("eager_max_time must be a non-negative "
+                             "finite number")
+        if self.eager_max_callbacks < 0:
+            raise ValueError("eager_max_callbacks must be non-negative")
+
+
+EventLoopSettings.DEFAULT = EventLoopSettings(eager_max_time=1e-4,
+                                              eager_max_callbacks=100)
+
+
 class AbstractEventLoop:
     """Abstract event loop."""
 
@@ -665,6 +692,14 @@ def get_debug(self):
     def set_debug(self, enabled):
         raise NotImplementedError
 
+    # Settings management.
+
+    def get_settings(self):
+        raise NotImplementedError
+
+    def set_settings(self, settings):
+        raise NotImplementedError
+
 
 class _Local(threading.local):
     _loop = None
diff --git a/Lib/asyncio/tasks.py b/Lib/asyncio/tasks.py
index 6c8d7cbbda64407..4b8bc840e60fa9b 100644
--- a/Lib/asyncio/tasks.py
+++ b/Lib/asyncio/tasks.py
@@ -21,6 +21,7 @@
 from types import GenericAlias
 
 from . import base_tasks
+from . import constants
 from . import coroutines
 from . import events
 from . import exceptions
@@ -193,7 +194,7 @@ def cancel(self, msg=None):
         """Request that this task cancel itself.
 
         This arranges for a CancelledError to be thrown into the
-        wrapped coroutine on the next cycle through the event loop.
+        wrapped coroutine the next time the task runs.
         The coroutine then has a chance to clean up or even deny
         the request using try/except/finally.
 
@@ -718,6 +719,12 @@ def __sleep0():
 async def sleep(delay, result=None):
     """Coroutine that completes after a given time (in seconds)."""
     if delay <= 0:
+        loop = events._get_running_loop()
+        if loop is not None:
+            try:
+                loop._eager_state = constants.EAGER_BREAK
+            except AttributeError:
+                pass
         await __sleep0()
         return result
 
diff --git a/Lib/asyncio/timeouts.py b/Lib/asyncio/timeouts.py
index 65ddc285abd971d..e814d3ead4d5ca1 100644
--- a/Lib/asyncio/timeouts.py
+++ b/Lib/asyncio/timeouts.py
@@ -33,8 +33,8 @@ def __init__(self, when: float | None) -> None:
         """Schedule a timeout that will trigger at a given loop time.
 
         - 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 `loop.call_soon()`.
         """
         self._state = _State.CREATED
 
diff --git a/Lib/test/test_asyncio/test_base_events.py b/Lib/test/test_asyncio/test_base_events.py
index e11f77ef10c0c96..746ebf85822571f 100644
--- a/Lib/test/test_asyncio/test_base_events.py
+++ b/Lib/test/test_asyncio/test_base_events.py
@@ -425,9 +425,26 @@ def test_set_debug(self):
         self.loop.set_debug(False)
         self.assertFalse(self.loop.get_debug())
 
-    def test__run_once_schedule_handle(self):
+    def test_set_settings_closed_loop(self):
+        settings = asyncio.EventLoopSettings(eager_max_time=0,
+                                             eager_max_callbacks=0)
+        self.loop.close()
+        self.loop.set_settings(settings)
+        self.assertIs(self.loop.get_settings(), settings)
+
+    def test_set_settings_invalid_type(self):
+        settings = self.loop.get_settings()
+        for value in (None, {}, 'x'):
+            with self.subTest(value=value):
+                with self.assertRaises(TypeError):
+                    self.loop.set_settings(value)
+                self.assertIs(self.loop.get_settings(), settings)
+
+    def test__run_once_schedule_handle_non_eager(self):
         handle = None
         processed = False
+        self.loop._settings = asyncio.EventLoopSettings(
+            eager_max_time=0, eager_max_callbacks=0)
 
         def cb(loop):
             nonlocal processed, handle
@@ -444,6 +461,322 @@ def cb(loop):
         self.assertTrue(processed)
         self.assertEqual([handle], list(self.loop._ready))
 
+    def test__run_once_schedule_handle_eager(self):
+        handle = None
+        processed = False
+
+        def cb(loop):
+            nonlocal processed, handle
+            processed = True
+            handle = loop.call_soon(lambda: True)
+
+        h = asyncio.TimerHandle(time.monotonic() - 1, cb, (self.loop,),
+                                self.loop, None)
+
+        self.loop._process_events = mock.Mock()
+        self.loop._scheduled.append(h)
+        self.loop._run_once()
+
+        self.assertTrue(processed)
+        self.assertEqual([], list(self.loop._ready))
+
+    def test__run_once_eager_max_callbacks(self):
+        calls = 0
+
+        def cb(loop):
+            nonlocal calls
+            calls += 1
+            loop.call_soon(cb, loop)
+
+        self.loop._settings = asyncio.EventLoopSettings(
+            eager_max_time=60, eager_max_callbacks=5)
+        self.loop._process_events = mock.Mock()
+        self.loop.call_soon(cb, self.loop)
+        self.loop._run_once()
+
+        # One callback from the regular pass plus five eager ones.
+        self.assertEqual(calls, 6)
+        self.assertEqual(len(self.loop._ready), 1)
+
+    def test__run_once_eager_disabled(self):
+        def chain(loop, n):
+            calls.append(n)
+            if n:
+                loop.call_soon(chain, loop, n - 1)
+
+        for args in [(0, 100), (60, 0)]:
+            with self.subTest(args=args):
+                calls = []
+                self.loop._ready.clear()
+                self.loop.set_settings(asyncio.EventLoopSettings(
+                    eager_max_time=args[0], eager_max_callbacks=args[1]))
+                self.loop._process_events = mock.Mock()
+                self.loop.call_soon(chain, self.loop, 3)
+                self.loop._run_once()
+
+                self.assertEqual(calls, [3])
+                self.assertEqual(len(self.loop._ready), 1)
+
+    def test__run_once_eager_max_callbacks_fan_out(self):
+        calls = 0
+
+        def worker():
+            nonlocal calls
+            calls += 1
+
+        def fan_out(loop):
+            for _ in range(10):
+                loop.call_soon(worker)
+
+        self.loop._settings = asyncio.EventLoopSettings(
+            eager_max_time=60, eager_max_callbacks=4)
+        self.loop._process_events = mock.Mock()
+        self.loop.call_soon(fan_out, self.loop)
+        self.loop._run_once()
+
+        self.assertEqual(calls, 4)
+        self.assertEqual(len(self.loop._ready), 6)
+
+    def test__run_once_eager_max_time_fan_out(self):
+        calls = 0
+        now = 0.0
+
+        def fake_time():
+            return now
+
+        def worker():
+            nonlocal calls, now
+            calls += 1
+            now += 1.0
+
+        def fan_out(loop):
+            for _ in range(10):
+                loop.call_soon(worker)
+
+        self.loop.time = fake_time
+        self.loop._clock_resolution = 0.0
+        self.loop._settings = asyncio.EventLoopSettings(
+            eager_max_time=2.5, eager_max_callbacks=100)
+        self.loop._process_events = mock.Mock()
+        self.loop.call_soon(fan_out, self.loop)
+        with mock.patch.object(constants, 'EAGER_BATCH_SIZE', 2):
+            self.loop._run_once()
+
+        # The deadline is checked between batches of two callbacks.
+        self.assertEqual(calls, 4)
+        self.assertEqual(len(self.loop._ready), 6)
+
+    def test__run_once_eager_break_skips_eager_phase(self):
+        handle = None
+
+        def cb(loop):
+            nonlocal handle
+            handle = loop.call_soon(lambda: True)
+            loop._eager_state = constants.EAGER_BREAK
+
+        self.loop._process_events = mock.Mock()
+        self.loop.call_soon(cb, self.loop)
+        self.loop._run_once()
+
+        self.assertEqual([handle], list(self.loop._ready))
+
+    def test__run_once_stop_ends_eager_phase(self):
+        calls = []
+        handle = None
+
+        def first(loop):
+            loop.call_soon(stopper, loop)
+
+        def stopper(loop):
+            nonlocal handle
+            calls.append('stopper')
+            handle = loop.call_soon(lambda: calls.append('late'))
+            loop.stop()
+
+        self.loop._process_events = mock.Mock()
+        self.loop.call_soon(first, self.loop)
+        self.loop._run_once()
+
+        self.assertEqual(['stopper'], calls)
+        self.assertEqual([handle], list(self.loop._ready))
+
+    def test_run_forever_stop_during_eager_processing(self):
+        calls = []
+
+        def first(loop):
+            loop.call_soon(stopper, loop)
+            loop.call_soon(breaker, loop)
+            for i in range(150):
+                loop.call_soon(calls.append, i)
+
+        def stopper(loop):
+            calls.append('stopper')
+            loop.stop()
+
+        def breaker(loop):
+            calls.append('breaker')
+            loop._eager_state = constants.EAGER_BREAK
+
+        self.loop._process_events = mock.Mock()
+        self.loop.call_soon(first, self.loop)
+        self.loop.run_forever()
+
+        # Callbacks scheduled before stop() still run, also past the
+        # current eager batch.
+        self.assertEqual(['stopper', 'breaker', *range(150)], calls)
+        self.assertEqual([], list(self.loop._ready))
+        self.assertEqual(self.loop._selector.select.call_count, 2)
+
+    def test_run_forever_stop_keeps_later_callbacks(self):
+        calls = []
+
+        def first(loop):
+            loop.call_soon(stopper, loop)
+            loop.call_soon(calls.append, 'next')
+
+        def stopper(loop):
+            calls.append('stopper')
+            loop.stop()
+            loop.call_soon(calls.append, 'later')
+
+        self.loop._process_events = mock.Mock()
+        self.loop.call_soon(first, self.loop)
+        self.loop.run_forever()
+
+        # As without eager processing, a callback scheduled in the iteration
+        # in which stop() was called runs the next time the loop runs.
+        self.assertEqual(['stopper', 'next'], calls)
+        self.assertEqual(1, len(self.loop._ready))
+        self.loop.call_soon(self.loop.stop)
+        self.loop.run_forever()
+        self.assertEqual(['stopper', 'next', 'later'], calls)
+
+    def test_run_forever_stopped_before_start(self):
+        calls = 0
+
+        def cb(loop):
+            nonlocal calls
+            calls += 1
+            loop.call_soon(cb, loop)
+
+        self.loop._process_events = mock.Mock()
+        self.loop.call_soon(cb, self.loop)
+        self.loop.stop()
+        self.loop.run_forever()
+
+        # Only one iteration runs, without eager processing.
+        self.assertEqual(calls, 1)
+        self.assertEqual(self.loop._selector.select.call_count, 1)
+
+    def test__run_once_eager_break_stops_eager_phase(self):
+        calls = []
+        handle = None
+
+        def first(loop):
+            loop.call_soon(breaker, loop)
+
+        def breaker(loop):
+            nonlocal handle
+            calls.append('breaker')
+            handle = loop.call_soon(lambda: calls.append('late'))
+            loop._eager_state = constants.EAGER_BREAK
+
+        self.loop._process_events = mock.Mock()
+        self.loop.call_soon(first, self.loop)
+        self.loop._run_once()
+
+        self.assertEqual(['breaker'], calls)
+        self.assertEqual([handle], list(self.loop._ready))
+
+        self.loop._run_once()
+        self.assertEqual(['breaker', 'late'], calls)
+        self.assertEqual([], list(self.loop._ready))
+
+    def test__run_once_eager_break_cuts_batch(self):
+        calls = []
+
+        def first(loop):
+            loop.call_soon(breaker, loop)
+            loop.call_soon(calls.append, 'next')
+
+        def breaker(loop):
+            calls.append('breaker')
+            loop._eager_state = constants.EAGER_BREAK
+
+        self.loop._process_events = mock.Mock()
+        self.loop.call_soon(first, self.loop)
+        self.loop._run_once()
+
+        # The rest of the batch runs in the next iteration.
+        self.assertEqual(['breaker'], calls)
+        self.assertEqual(1, len(self.loop._ready))
+
+    def test__run_once_resets_eager_break(self):
+        calls = []
+
+        def chain(loop, n):
+            calls.append(n)
+            if n:
+                loop.call_soon(chain, loop, n - 1)
+
+        self.loop.set_settings(asyncio.EventLoopSettings(
+            eager_max_time=60, eager_max_callbacks=100))
+        self.loop._process_events = mock.Mock()
+        # Left over from a previous iteration.
+        self.loop._eager_state = constants.EAGER_BREAK
+        self.loop.call_soon(chain, self.loop, 3)
+        self.loop._run_once()
+
+        self.assertEqual(calls, [3, 2, 1, 0])
+
+    def test_sleep0_ends_eager_processing(self):
+        calls = []
+
+        async def main():
+            calls.append('before')
+            await asyncio.sleep(0)
+            calls.append('after')
+
+        self.loop._process_events = mock.Mock()
+        task = self.loop.create_task(main())
+        asyncio.events._set_running_loop(self.loop)
+        try:
+            self.loop._run_once()
+            # The task resumes only after the next I/O poll.
+            self.assertEqual(calls, ['before'])
+            self.loop._run_once()
+            self.assertEqual(calls, ['before', 'after'])
+        finally:
+            asyncio.events._set_running_loop(None)
+        self.assertTrue(task.done())
+
+    def test_sleep0_breaks_eager(self):
+        loop = mock.Mock(_eager_state=constants.EAGER_IDLE)
+        with mock.patch('asyncio.events._get_running_loop',
+                        return_value=loop):
+            coro = asyncio.sleep(0)
+            self.assertIsNone(coro.send(None))
+            self.assertEqual(loop._eager_state, constants.EAGER_BREAK)
+            with self.assertRaises(StopIteration):
+                coro.send(None)
+
+    def test_sleep0_loop_without_eager_state(self):
+        # A third-party loop may not accept the _eager_state attribute.
+        with mock.patch('asyncio.events._get_running_loop',
+                        return_value=object()):
+            coro = asyncio.sleep(0, result=42)
+            self.assertIsNone(coro.send(None))
+            with self.assertRaises(StopIteration) as cm:
+                coro.send(None)
+        self.assertEqual(cm.exception.value, 42)
+
+    def test_sleep0_without_running_loop(self):
+        coro = asyncio.sleep(0, result=42)
+        self.assertIsNone(coro.send(None))
+        with self.assertRaises(StopIteration) as cm:
+            coro.send(None)
+        self.assertEqual(cm.exception.value, 42)
+
     def test__run_once_cancelled_event_cleanup(self):
         self.loop._process_events = mock.Mock()
 
@@ -1274,6 +1607,52 @@ def setUp(self):
         self.loop = asyncio.SelectorEventLoop()
         self.set_event_loop(self.loop)
 
+    def test_run_until_complete_after_eager_limit(self):
+        # The eager limit ends eager processing before the done-callback
+        # that stops the loop runs; the callbacks queued before it still
+        # run before run_until_complete() returns.
+        self.loop.set_settings(asyncio.EventLoopSettings(
+            eager_max_time=60, eager_max_callbacks=5))
+        results = []
+
+        async def worker(ev, i):
+            await ev.wait()
+            return i
+
+        async def main():
+            ev = asyncio.Event()
+            for i in range(20):
+                task = asyncio.ensure_future(worker(ev, i))
+                task.add_done_callback(lambda t: results.append(t.result()))
+            self.loop.call_soon(ev.set)
+            await ev.wait()
+
+        self.loop.run_until_complete(main())
+        self.assertEqual(sorted(results), list(range(20)))
+
+    def test_run_until_complete_processes_ready_io(self):
+        rsock, wsock = socket.socketpair()
+        self.addCleanup(rsock.close)
+        self.addCleanup(wsock.close)
+        received = []
+        self.loop.add_reader(rsock, lambda: received.append(rsock.recv(1)))
+        self.addCleanup(self.loop.remove_reader, rsock)
+
+        async def main():
+            wsock.send(b'x')
+
+        for settings in (asyncio.EventLoopSettings(eager_max_time=0,
+                                                   eager_max_callbacks=0),
+                         self.loop.get_settings()):
+            with self.subTest(settings=settings):
+                received.clear()
+                self.loop.set_settings(settings)
+                # I/O made ready by the last step is processed before
+                # run_until_complete() returns.
+                self.loop.run_until_complete(main())
+                self.assertEqual(received, [b'x'])
+                test_utils.run_briefly(self.loop)
+
     @mock.patch('socket.getnameinfo')
     def test_getnameinfo(self, m_gai):
         m_gai.side_effect = lambda *args: 42
diff --git a/Lib/test/test_asyncio/test_events.py b/Lib/test/test_asyncio/test_events.py
index 58556cc87409f7f..82c8033043a3202 100644
--- a/Lib/test/test_asyncio/test_events.py
+++ b/Lib/test/test_asyncio/test_events.py
@@ -2771,6 +2771,43 @@ def noop(*args, **kwargs):
     pass
 
 
+class EventLoopSettingsTests(unittest.TestCase):
+
+    def make(self, max_time, max_callbacks):
+        return asyncio.EventLoopSettings(eager_max_time=max_time,
+                                         eager_max_callbacks=max_callbacks)
+
+    def test_default(self):
+        self.assertEqual(asyncio.EventLoopSettings.DEFAULT,
+                         self.make(1e-4, 100))
+        loop = asyncio.new_event_loop()
+        self.addCleanup(loop.close)
+        self.assertIs(loop.get_settings(), asyncio.EventLoopSettings.DEFAULT)
+
+    def test_valid(self):
+        for args in [(0, 0), (0, 10), (0.0, 0), (1e-6, 0), (1e-6, 1),
+                     (1, 100)]:
+            with self.subTest(args=args):
+                settings = self.make(*args)
+                self.assertEqual(settings.eager_max_time, args[0])
+                self.assertEqual(settings.eager_max_callbacks, args[1])
+
+    def test_invalid_type(self):
+        for args in [(None, 1), ('1', 1), (True, 1), (False, 0),
+                     (1e-6, 1.0), (1e-6, None), (1e-6, '1'), (1e-6, True),
+                     (0, False)]:
+            with self.subTest(args=args):
+                with self.assertRaises(TypeError):
+                    self.make(*args)
+
+    def test_invalid_value(self):
+        for args in [(-1e-6, 1), (float('inf'), 1), (float('nan'), 1),
+                     (10**400, 1), (1e-6, -1)]:
+            with self.subTest(args=args):
+                with self.assertRaises(ValueError):
+                    self.make(*args)
+
+
 class HandleTests(test_utils.TestCase):
 
     def setUp(self):
@@ -3173,6 +3210,10 @@ def test_not_implemented(self):
             NotImplementedError, loop.get_debug)
         self.assertRaises(
             NotImplementedError, loop.set_debug, f)
+        self.assertRaises(
+            NotImplementedError, loop.get_settings)
+        self.assertRaises(
+            NotImplementedError, loop.set_settings, f)
 
     def test_not_implemented_async(self):
 
diff --git a/Lib/test/test_asyncio/test_proactor_events.py b/Lib/test/test_asyncio/test_proactor_events.py
index d25c5fa47d1dc1d..b6adcec62fa743e 100644
--- a/Lib/test/test_asyncio/test_proactor_events.py
+++ b/Lib/test/test_asyncio/test_proactor_events.py
@@ -393,6 +393,8 @@ def test_write_eof_duplex_pipe(self):
         close_transport(tr)
 
     def test_pause_resume_reading(self):
+        self.loop.set_settings(asyncio.EventLoopSettings(
+            eager_max_time=0, eager_max_callbacks=0))
         tr = self.socket_transport()
         index = 0
         msgs = [b'data1', b'data2', b'data3', b'data4', b'data5', b'']
diff --git a/Lib/test/test_asyncio/test_windows_events.py b/Lib/test/test_asyncio/test_windows_events.py
index bb4ba74f19a17f0..2a36c36f17b7638 100644
--- a/Lib/test/test_asyncio/test_windows_events.py
+++ b/Lib/test/test_asyncio/test_windows_events.py
@@ -264,6 +264,10 @@ def test_address_argument_type_error(self):
         sock.close()
 
     def test_client_pipe_stat(self):
+        # The probes rely on the pipe server creating the next pipe
+        # instance between them, which eager processing would postpone.
+        self.loop.set_settings(asyncio.EventLoopSettings(
+            eager_max_time=0, eager_max_callbacks=0))
         res = self.loop.run_until_complete(self._test_client_pipe_stat())
         self.assertEqual(res, 'done')
 
diff --git a/Misc/NEWS.d/next/Library/2026-10-27-15-40-00.gh-issue-160760.plLyNA.rst b/Misc/NEWS.d/next/Library/2026-10-27-15-40-00.gh-issue-160760.plLyNA.rst
new file mode 100644
index 000000000000000..6089af11edbdcc6
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2026-10-27-15-40-00.gh-issue-160760.plLyNA.rst
@@ -0,0 +1,5 @@
+The :mod:`asyncio` event loop now runs callbacks scheduled by other
+callbacks eagerly, before polling for I/O again, within limits set by the
+new :class:`asyncio.EventLoopSettings` and :meth:`loop.set_settings()
+<asyncio.loop.set_settings>`. :func:`asyncio.sleep` with a delay of 0 or less
+ends eager processing in the current iteration of the loop.
diff --git a/Modules/_asynciomodule.c b/Modules/_asynciomodule.c
index 73404cb75b5d1ff..9d7423d06a76170 100644
--- a/Modules/_asynciomodule.c
+++ b/Modules/_asynciomodule.c
@@ -2551,7 +2551,7 @@ _asyncio.Task.cancel
 Request that this task cancel itself.
 
 This arranges for a CancelledError to be thrown into the
-wrapped coroutine on the next cycle through the event loop.
+wrapped coroutine the next time the task runs.
 The coroutine then has a chance to clean up or even deny
 the request using try/except/finally.
 
@@ -2572,7 +2572,7 @@ This also increases the task's count of cancellation requests.
 
 static PyObject *
 _asyncio_Task_cancel_impl(TaskObj *self, PyObject *msg)
-/*[clinic end generated code: output=c66b60d41c74f9f1 input=6125d45b9a6a5abd]*/
+/*[clinic end generated code: output=c66b60d41c74f9f1 input=aadb870f189f82b5]*/
 {
     self->task_log_tb = 0;
 
diff --git a/Modules/clinic/_asynciomodule.c.h b/Modules/clinic/_asynciomodule.c.h
index 7ffddc20ea3153b..5016a068cf1498b 100644
--- a/Modules/clinic/_asynciomodule.c.h
+++ b/Modules/clinic/_asynciomodule.c.h
@@ -960,7 +960,7 @@ PyDoc_STRVAR(_asyncio_Task_cancel__doc__,
 "Request that this task cancel itself.\n"
 "\n"
 "This arranges for a CancelledError to be thrown into the\n"
-"wrapped coroutine on the next cycle through the event loop.\n"
+"wrapped coroutine the next time the task runs.\n"
 "The coroutine then has a chance to clean up or even deny\n"
 "the request using try/except/finally.\n"
 "\n"
@@ -2104,4 +2104,4 @@ _asyncio_future_discard_from_awaited_by(PyObject *module, PyObject *const *args,
 
 #define _ASYNCIO_TASK__FUT_WAITER_GETSETDEF {"_fut_waiter", (getter)_asyncio_Task__fut_waiter_get, (setter)NULL, NULL},
 
-/*[clinic end generated code: output=419f6d2b73047616 input=a9049054013a1b77]*/
+/*[clinic end generated code: output=eccf9dfdf8c91532 input=a9049054013a1b77]*/