@@ -180,6 +180,8 @@ def __init__(
180180 self .terminate_timeout = terminate_timeout
181181 self .collectors : list [asyncio .Task [None ] | asyncio .Task [int ]] = []
182182 self .main_task : asyncio .Task [bool ] | None = None
183+ self .is_terminating = False
184+ self .do_terminate = False
183185
184186 async def _shutdown_process (self , proc : Process ) -> None :
185187 """Gracefully shutdown a child process."""
@@ -203,6 +205,8 @@ async def _main_task(self, proc: Process, wait_task: asyncio.Task[int]) -> bool:
203205 Return ``True`` if the process had a non-zero exit code during shutdown.
204206 """
205207 try :
208+ if self .do_terminate :
209+ self ._handle_sigint ()
206210 await proc .wait ()
207211 except asyncio .CancelledError :
208212 # SIGINT causes the task to be cancelled. We first need to uncancel it
@@ -260,7 +264,8 @@ async def _read_stream(
260264 # should be data and assume that the chunk is complete and EOF has been reached.
261265 if chunk_part :
262266 continue
263- is_eof = True
267+ # We mark the following line as 'no cover' since it should never happen.
268+ is_eof = True # pragma: no cover
264269
265270 # Re-combine the line (including separator, if EOF wasn't found).
266271 chunk = b"" .join (chunk_parts )
@@ -278,6 +283,7 @@ async def _stream_subprocess(
278283 args : Sequence [str ],
279284 * ,
280285 env : Mapping [str , str ] | None ,
286+ extra_tasks : Sequence [asyncio .Task [None ]],
281287 ) -> tuple [bool , int ]:
282288 """
283289 Start the process, all tasks, and wait for them to finish.
@@ -317,14 +323,33 @@ async def _stream_subprocess(
317323 self .main_task = self .loop .create_task (self ._main_task (proc , wait_task ))
318324
319325 # We wait for all tasks to finish and extract the return code.
320- await asyncio .wait ([* self .collectors , self .main_task ])
326+ await asyncio .wait ([* self .collectors , self .main_task , * extra_tasks ])
321327 return self .main_task .result (), wait_task .result ()
322328
329+ def _handle_sigint (self ) -> None :
330+ """
331+ SIGINT handler for event loop.
332+ """
333+ # In case this is called before the main task is created,
334+ # set a flag and that's it.
335+ if not self .main_task :
336+ self .do_terminate = True
337+ return
338+
339+ # Don't handle it twice.
340+ if self .is_terminating :
341+ return
342+ self .is_terminating = True
343+
344+ # Cancel the main task. This triggers graceful shutdown.
345+ self .main_task .cancel ()
346+
323347 def run (
324348 self ,
325349 args : Sequence [str ],
326350 * ,
327351 env : Mapping [str , str ] | None = None ,
352+ extra_tasks : Sequence [asyncio .Task [None ]] | None = None ,
328353 ) -> tuple [int , bytes , bytes ]:
329354 """
330355 Run the command with the given environment.
@@ -333,22 +358,10 @@ def run(
333358 Note that this function is **NOT** thread-safe, since we have to
334359 add a SIGINT handler to the event loop and later remove it.
335360 """
336- is_terminating = [False ]
337-
338- def handle_sigint () -> None :
339- # Don't handle it twice.
340- if is_terminating [0 ]:
341- return
342- is_terminating [0 ] = True
343-
344- # Cancel the main task. This triggers graceful shutdown.
345- if self .main_task :
346- self .main_task .cancel ()
347-
348- self .loop .add_signal_handler (signal .SIGINT , handle_sigint )
361+ self .loop .add_signal_handler (signal .SIGINT , self ._handle_sigint )
349362 try :
350363 is_canceled , return_code = self .loop .run_until_complete (
351- self ._stream_subprocess (args , env = env )
364+ self ._stream_subprocess (args , env = env , extra_tasks = extra_tasks or [] )
352365 )
353366 finally :
354367 self .loop .remove_signal_handler (signal .SIGINT )
0 commit comments