@@ -44,6 +44,23 @@ def wrapped(self: "Cmd", *args: Any, **kwargs: Any) -> Any:
4444 return cast (_F , wrapped )
4545
4646
47+ def pipe_target (stream : Any ) -> Any :
48+ """Return the stream a pipe process can inherit, or ``None`` if its output must be captured.
49+
50+ A pipe process may be interactive, such as ``less`` or ``fzf``, so it needs the real
51+ terminal rather than a stream this process reads on its behalf. Look through a
52+ :class:`ToolbarStream` wrapper, but only hand back a stream owning a file descriptor.
53+ """
54+ if isinstance (stream , ToolbarStream ):
55+ stream = stream .original
56+ try :
57+ stream .fileno ()
58+ except (AttributeError , OSError ):
59+ # io.UnsupportedOperation, raised by streams like io.StringIO, subclasses OSError.
60+ return None
61+ return stream
62+
63+
4764class _ContextStdoutProxy (StdoutProxy ):
4865 """Keep stdout's flush worker in the toolbar's isolated application session."""
4966
@@ -57,19 +74,24 @@ def _start_write_thread(self) -> threading.Thread:
5774class ToolbarStream :
5875 """Keep a stable stream identity across suspensions and cmd2 redirections."""
5976
60- def __init__ (self , original : TextIO ) -> None :
77+ def __init__ (self , original : TextIO , lock : "threading.RLock" ) -> None :
6178 """Wrap a terminal stream while preserving its ordinary file attributes."""
6279 self .original = original
6380 self .proxy : StdoutProxy | None = None
81+ # Shared with the toolbar so a write from another thread cannot land on a proxy
82+ # that is being closed. Such a write is accepted by the dead proxy and discarded.
83+ self ._lock = lock
6484 self .buffer = _ToolbarBuffer (self )
6585
6686 def write (self , data : str ) -> int :
6787 """Write above the toolbar, or directly while the toolbar is suspended."""
68- return (self .proxy or self .original ).write (data )
88+ with self ._lock :
89+ return (self .proxy or self .original ).write (data )
6990
7091 def flush (self ) -> None :
7192 """Flush the currently active output stream."""
72- (self .proxy or self .original ).flush ()
93+ with self ._lock :
94+ (self .proxy or self .original ).flush ()
7395
7496 def __getattr__ (self , name : str ) -> Any :
7597 """Delegate file attributes to the original terminal stream."""
@@ -117,6 +139,8 @@ def __init__(self, cmd: "Cmd") -> None:
117139 self ._thread : threading .Thread | None = None
118140 self ._streams : list [ToolbarStream ] = []
119141 self ._proxy : StdoutProxy | None = None
142+ self ._lock = threading .RLock ()
143+ self ._pausing = False
120144
121145 session = cmd .main_session
122146 self .app = session .app
@@ -149,12 +173,19 @@ def interrupt(event: KeyPressEvent) -> None: # noqa: ARG001
149173 self ._keys .clear ()
150174 if sys .platform == "win32" :
151175 # os.kill(..., SIGINT) terminates the process on Windows instead
152- # of dispatching Python's signal handler.
176+ # of dispatching Python's signal handler. This reaches only this
177+ # process, so a console subprocess started by a command keeps
178+ # running until it is waited on.
153179 import _thread
154180
155181 _thread .interrupt_main ()
156182 else :
157- os .kill (os .getpid (), signal .SIGINT )
183+ # Raw mode clears ISIG, so no signal is generated for us. Signal the
184+ # foreground process group the way the terminal driver would, since a
185+ # command may be waiting on a subprocess that shares this group. Pipe
186+ # processes are excluded because cmd2 starts them in their own session
187+ # and forwards to them from sigint_handler().
188+ os .killpg (os .getpgrp (), signal .SIGINT )
158189
159190 @bindings .add (
160191 "c-z" ,
@@ -184,7 +215,7 @@ def start(self) -> None:
184215 for obj , name in ((self .cmd , "stdout" ), (sys , "stdout" ), (sys , "stderr" )):
185216 stream = getattr (obj , name )
186217 if stream .isatty ():
187- wrapper = ToolbarStream (stream )
218+ wrapper = ToolbarStream (stream , self . _lock )
188219 self ._streams .append (wrapper )
189220 setattr (obj , name , cast (TextIO , wrapper ))
190221 stack .callback (self ._restore_stream , obj , name , wrapper )
@@ -219,6 +250,7 @@ def run() -> None:
219250 self ._error = exc
220251 finally :
221252 self ._ready .set ()
253+ self ._app_exited ()
222254
223255 self ._thread = threading .Thread (target = context .run , args = (run ,), name = "cmd2-toolbar" , daemon = True )
224256 self ._thread .start ()
@@ -227,32 +259,86 @@ def run() -> None:
227259 raise self ._error
228260 # The worker already combines queued writes. A batching sleep would also
229261 # delay close(), which runs at each command finalization boundary.
230- self ._proxy = _ContextStdoutProxy (raw = True , sleep_between_writes = 0 )
231- for stream in self ._streams :
232- stream .proxy = self ._proxy
262+ proxy = _ContextStdoutProxy (raw = True , sleep_between_writes = 0 )
263+ with self ._lock :
264+ self ._proxy = proxy
265+ for stream in self ._streams :
266+ stream .proxy = proxy
267+
268+ def _app_exited (self ) -> None :
269+ """Give the terminal back to the streams when the display stops on its own.
270+
271+ ``_ready`` is set as soon as the first frame renders, so a failure after that is
272+ never seen by the command thread waiting in ``_resume()``. The display is gone at
273+ that point and its stdout proxy can no longer reach the terminal, so anything
274+ written through it would be discarded without a trace.
275+ """
276+ if self ._pausing :
277+ # A deliberate pause restores the streams itself, in the right order.
278+ return
279+
280+ with self ._lock :
281+ # Leave self._proxy set so that the next _pause() still drains and closes
282+ # it. With the display gone, its worker writes to the terminal directly.
283+ started = self ._proxy is not None
284+ for stream in self ._streams :
285+ stream .proxy = None
286+
287+ # A proxy exists only once _resume() has handed startup failures to the command
288+ # thread, so reporting here does not duplicate the exception it raises.
289+ if started and self ._error is not None :
290+ self .cmd .perror (f"Bottom toolbar stopped after an error: { self ._error !r} " )
291+
292+ def _exit (self ) -> None :
293+ """Stop the display unless it has already stopped on its own.
294+
295+ Application.exit() raises once the result is set, and this runs later than the
296+ check that scheduled it. Any exception here would reach the loop's default
297+ handler, which prints a traceback over the terminal.
298+
299+ Output queued before this does not need draining: Application.run_async() waits
300+ for cursor position reports and for run_in_terminal() calls still in flight
301+ before its loop closes.
302+ """
303+ if self .app .is_running and not self .app .is_done :
304+ self .app .exit ()
233305
234306 def _pause (self ) -> None :
307+ self ._pausing = True
235308 try :
236- if self ._proxy is not None :
237- self ._proxy .flush ()
238- self ._proxy .close ()
309+ try :
310+ # Hold off other threads while the proxy drains so their output is never
311+ # handed to a proxy whose worker has already stopped. Writes that arrive
312+ # after this go straight to the terminal, still in order.
313+ with self ._lock :
314+ try :
315+ if self ._proxy is not None :
316+ self ._proxy .flush ()
317+ self ._proxy .close ()
318+ finally :
319+ self ._proxy = None
320+ for stream in self ._streams :
321+ stream .proxy = None
322+ finally :
323+ # The lock is released before joining, since the toolbar thread may be
324+ # blocked writing through a stream that is waiting on it.
325+ if self .app .is_running and self .app .loop is not None :
326+ self .app .loop .call_soon_threadsafe (self ._exit )
327+ if self ._thread is not None :
328+ self ._thread .join ()
329+ self ._thread = None
330+ # Return the borrowed application to the main prompt, including on
331+ # proxy failures. The upstream toolbar owned a separate application.
332+ if self ._display_stack is not None :
333+ self ._display_stack .close ()
334+ self ._display_stack = None
335+ # Application.run() saves its unprocessed queue before the thread
336+ # exits. Those keys arrived after the ones handled by save_key().
337+ pending_keys = get_typeahead (self .app .input )
338+ store_typeahead (self .app .input , self ._keys + pending_keys )
339+ self ._keys .clear ()
239340 finally :
240- self ._proxy = None
241- for stream in self ._streams :
242- stream .proxy = None
243- if self .app .is_running and self .app .loop is not None :
244- self .app .loop .call_soon_threadsafe (self .app .exit )
245- if self ._thread is not None :
246- self ._thread .join ()
247- self ._thread = None
248- if self ._display_stack is not None :
249- self ._display_stack .close ()
250- self ._display_stack = None
251- # Application.run() saves its unprocessed queue before the thread
252- # exits. Those keys arrived after the ones handled by save_key().
253- pending_keys = get_typeahead (self .app .input )
254- store_typeahead (self .app .input , self ._keys + pending_keys )
255- self ._keys .clear ()
341+ self ._pausing = False
256342
257343 def stop (self ) -> None :
258344 """Flush output, stop rendering, and restore the terminal and its streams."""
0 commit comments