pythongh-108973: Fix asyncio test_subprocess_consistent_callbacks()

Subprocess events can be delivered in any order: tolerate that in the
test.

Revert commit 282edd7.
_child_watcher_callback() calls immediately _process_exited(): don't
add an additional delay with call_soon(). The reverted change didn't
make _process_exited() more determistic: it can still be called
before pipe_connection_lost() for example.

Co-authored-by: Davide Rizzo <[email protected]>

Author: vstinner

Doc/library/asyncio-llapi-index.rst

@@ -496,7 +496,9 @@ Protocol classes can implement the following **callback methods**:
 
     * - ``callback`` :meth:`process_exited()
         <SubprocessProtocol.process_exited>`
-      - Called when the child process has exited.
+      - Called when the child process has exited. This method is not always
+        the last one to be called: protocol methods are not called in a
+        determistic order.
 
 
 Event Loop Policies

Doc/library/asyncio-protocol.rst

@@ -688,6 +688,11 @@ Subprocess Protocol instances should be constructed by protocol
 factories passed to the :meth:`loop.subprocess_exec` and
 :meth:`loop.subprocess_shell` methods.
 
+The protocol methods are not called in a determistic order. The order depends
+on the event loop implementation and the operating system. The protocol should
+implement its own logic depending on the use case to decide when it's done with
+the subprocess.
+
 .. method:: SubprocessProtocol.pipe_data_received(fd, data)
 
    Called when the child process writes data into its stdout or stderr
@@ -1003,12 +1008,25 @@ The subprocess is created by the :meth:`loop.subprocess_exec` method::
         def __init__(self, exit_future):
             self.exit_future = exit_future
             self.output = bytearray()
+            self.pipe_closed = False
+            self.exited = False
+
+        def pipe_connection_lost(self, fd, exc):
+            self.pipe_closed = True
+            self.exit_maybe()
 
         def pipe_data_received(self, fd, data):
             self.output.extend(data)
 
         def process_exited(self):
-            self.exit_future.set_result(True)
+            self.exited = True
+            # process_exited() can be called before pipe_connection_lost():
+            # wait until both methods are called.
+            self.exit_maybe()
+
+        def exit_maybe(self):
+            if self.pipe_closed and self.exited:
+                self.exit_future.set_result(True)
 
     async def get_date():
         # Get a reference to the event loop as we plan to use

Lib/asyncio/unix_events.py

@@ -226,8 +226,7 @@ async def _make_subprocess_transport(self, protocol, args, shell,
         return transp
 
     def _child_watcher_callback(self, pid, returncode, transp):
-        # Skip one iteration for callbacks to be executed
-        self.call_soon_threadsafe(self.call_soon, transp._process_exited, returncode)
+        self.call_soon_threadsafe(transp._process_exited, returncode)
 
     async def create_unix_connection(
             self, protocol_factory, path=None, *,

Lib/test/test_asyncio/test_subprocess.py

@@ -753,21 +753,39 @@ async def main() -> None:
 
         self.loop.run_until_complete(main())
 
-    def test_subprocess_consistent_callbacks(self):
+    def test_subprocess_protocol_events(self):
+        # gh-108973: Test that all subprocess protocol methods are called.
+        # The protocol methods are not called in a determistic order.
+        # The order depends on the event loop and the operating system.
         events = []
+        expected = [
+            ('pipe_data_received', 1, b'stdout'),
+            ('pipe_data_received', 2, b'stderr'),
+            ('pipe_connection_lost', 1),
+            ('pipe_connection_lost', 2),
+            'process_exited',
+        ]
+
         class MyProtocol(asyncio.SubprocessProtocol):
             def __init__(self, exit_future: asyncio.Future) -> None:
                 self.exit_future = exit_future
 
             def pipe_data_received(self, fd, data) -> None:
                 events.append(('pipe_data_received', fd, data))
+                self.exit_maybe()
 
             def pipe_connection_lost(self, fd, exc) -> None:
-                events.append('pipe_connection_lost')
+                events.append(('pipe_connection_lost', fd))
+                self.exit_maybe()
 
             def process_exited(self) -> None:
                 events.append('process_exited')
-                self.exit_future.set_result(True)
+                self.exit_maybe()
+
+            def exit_maybe(self):
+                # Only exit when we got all expected events
+                if len(events) >= len(expected):
+                    self.exit_future.set_result(True)
 
         async def main() -> None:
             loop = asyncio.get_running_loop()
@@ -777,13 +795,9 @@ async def main() -> None:
                                                       sys.executable, '-c', code, stdin=None)
             await exit_future
             transport.close()
-            self.assertEqual(events, [
-                ('pipe_data_received', 1, b'stdout'),
-                ('pipe_data_received', 2, b'stderr'),
-                'pipe_connection_lost',
-                'pipe_connection_lost',
-                'process_exited',
-            ])
+
+            # Protocol methods can be called in any order
+            self.assertSetEqual(set(events), set(expected))
 
         self.loop.run_until_complete(main())