gh-124309: Modernize the staggered_race implementation to support eager task factories

Lib/asyncio/base_events.py

@@ -1144,7 +1144,7 @@ async def create_connection(
                     (functools.partial(self._connect_sock,
                                        exceptions, addrinfo, laddr_infos)
                      for addrinfo in infos),
-                    happy_eyeballs_delay, loop=self)
+                    happy_eyeballs_delay)
 
             if sock is None:
                 exceptions = [exc for sub in exceptions for exc in sub]

Lib/asyncio/staggered.py

@@ -12,7 +12,8 @@
 class _Done(Exception):
     pass
 
-async def staggered_race(coro_fns, delay, *, loop=None):
+
+async def staggered_race(coro_fns, delay):
     """Run coroutines with staggered start times and take the first to finish.
 
     This method takes an iterable of coroutine functions. The first one is
@@ -44,8 +45,6 @@ async def staggered_race(coro_fns, delay, *, loop=None):
         delay: amount of time, in seconds, between starting coroutines. If
             ``None``, the coroutines will run sequentially.
 
-        loop: the event loop to use.
-
     Returns:
         tuple *(winner_result, winner_index, exceptions)* where
 
@@ -79,7 +78,7 @@ async def run_one_coro(this_index, coro_fn, this_failed):
         else:
             # Store winner's results
             nonlocal winner_index, winner_result
-            # There could be more than one winner
+            assert winner_index is None
             winner_index = this_index
             winner_result = result
             raise _Done

Lib/test/test_asyncio/test_staggered.py

@@ -82,19 +82,44 @@ async def test_none_successful(self):
         async def coro(index):
             raise ValueError(index)
 
+        for delay in [None, 0, 0.1, 1]:
+            with self.subTest(delay=delay):
+                winner, index, excs = await staggered_race(
+                    [
+                        lambda: coro(0),
+                        lambda: coro(1),
+                    ],
+                    delay=delay,
+                )
+
+                self.assertIs(winner, None)
+                self.assertIs(index, None)
+                self.assertEqual(len(excs), 2)
+                self.assertIsInstance(excs[0], ValueError)
+                self.assertIsInstance(excs[1], ValueError)
+
+    async def test_long_delay_early_failure(self):
+        async def coro(index):
+            await asyncio.sleep(0)  # Dummy coroutine for the 1 case
+            if index == 0:
+                await asyncio.sleep(0.1)  # Dummy coroutine
+                raise ValueError(index)
+
+            return f'Res: {index}'
+
         winner, index, excs = await staggered_race(
             [
                 lambda: coro(0),
                 lambda: coro(1),
             ],
-            delay=None,
+            delay=10,
         )
 
-        self.assertIs(winner, None)
-        self.assertIs(index, None)
+        self.assertEqual(winner, 'Res: 1')
+        self.assertEqual(index, 1)
         self.assertEqual(len(excs), 2)
         self.assertIsInstance(excs[0], ValueError)
-        self.assertIsInstance(excs[1], ValueError)
+        self.assertIsNone(excs[1], None)
 
 
 if __name__ == "__main__":