Fix SuppressStdoutStderr permanently silencing the process when contexts overlap

CHANGELOG.md

@@ -13,6 +13,7 @@ but cannot always guarantee backwards compatibility. Changes that may **break co
 
 **Fixed**
 
+- Fixed `SuppressStdoutStderr` (used to suppress `Prophet` fit output) permanently redirecting the process-level stdout and stderr to `devnull` when contexts were used at the same time, and crashing in interpreters without usable standard streams (e.g. `pythonw` on Windows). The redirection is now shared behind a lock with a depth counter, and degrades to a no-op when the standard streams cannot be saved. [#3290](https://github.com/unit8co/darts/pull/3290) by [Manas Choksi](https://github.com/choksi2212).
 - Fixed compatibility with pandas 3.1: fitting and predicting models on series with a daily frequency failed with a `TypeError`, and `MIDAS` accepted a series whose frequency was already the target low frequency. [#3289](https://github.com/unit8co/darts/pull/3289) by [Nicolò Angileri](https://github.com/nicoloangileri).
 
 **Dependencies**

darts/logging.py

@@ -1,8 +1,10 @@
 import logging
 import os
 import sys
+import threading
 import time
 import warnings
+from contextlib import suppress
 from typing import NoReturn
 
 
@@ -116,28 +118,83 @@ class SuppressStdoutStderr:
     to stderr just before a script exits, and after the context manager has
     exited (at least, I think that is why it lets exceptions through).
 
+    The redirection targets the process-level file descriptors 1 and 2, so
+    instances used from different threads share a single redirection: the
+    first context to enter saves the real descriptors and the last one to
+    leave restores them, and a context entered while a redirection is
+    already active never mistakes the redirected descriptors for the real
+    ones. When the standard streams are not backed by usable file
+    descriptors (e.g. under pythonw on Windows), the context degrades to
+    doing nothing instead of raising.
+
     source:
     https://stackoverflow.com/questions/11130156/suppress-stdout-stderr-print-from-python-functions
     """
 
+    _lock = threading.Lock()
+    _depth = 0
+    _null_fds: tuple[int, int] | None = None
+    _saved_fds: tuple[int, int] | None = None
+
     def __init__(self):
-        # Open a pair of null files
-        self.null_fds = [os.open(os.devnull, os.O_RDWR) for x in range(2)]
-        # Save the actual stdout (1) and stderr (2) file descriptors.
-        self.save_fds = [os.dup(1), os.dup(2)]
+        # The descriptors are only touched on __enter__ / __exit__, so that
+        # creating the context manager neither fails nor leaks descriptors
+        # when the standard streams are not backed by real files.
+        self._active = False
 
     def __enter__(self):
-        # Assign the null pointers to stdout and stderr.
-        os.dup2(self.null_fds[0], 1)
-        os.dup2(self.null_fds[1], 2)
+        with SuppressStdoutStderr._lock:
+            if SuppressStdoutStderr._depth > 0:
+                # Someone else already redirected the descriptors; joining
+                # the active redirection must not overwrite the saved ones.
+                SuppressStdoutStderr._depth += 1
+                self._active = True
+                return self
+
+            null_a = null_b = saved_a = saved_b = -1
+            try:
+                null_a = os.open(os.devnull, os.O_RDWR)
+                null_b = os.open(os.devnull, os.O_RDWR)
+                saved_a = os.dup(1)
+                saved_b = os.dup(2)
+            except OSError:
+                # No usable standard streams (e.g. windowed GUI
+                # interpreters): nothing to save, nothing to restore.
+                # Close anything opened before the failure so it does not
+                # leak, and leave the redirection untouched.
+                for fd in (null_a, null_b, saved_a):
+                    if fd >= 0:
+                        with suppress(OSError):
+                            os.close(fd)
+                return self
+
+            os.dup2(null_a, 1)
+            os.dup2(null_b, 2)
+            SuppressStdoutStderr._null_fds = (null_a, null_b)
+            SuppressStdoutStderr._saved_fds = (saved_a, saved_b)
+            SuppressStdoutStderr._depth = 1
+            self._active = True
+        return self
 
     def __exit__(self, *_):
-        # Re-assign the real stdout/stderr back to (1) and (2)
-        os.dup2(self.save_fds[0], 1)
-        os.dup2(self.save_fds[1], 2)
-        # Close all file descriptors
-        for fd in self.null_fds + self.save_fds:
-            os.close(fd)
+        with SuppressStdoutStderr._lock:
+            if not self._active:
+                return
+            self._active = False
+            SuppressStdoutStderr._depth -= 1
+            if SuppressStdoutStderr._depth == 0:
+                saved_fds = SuppressStdoutStderr._saved_fds
+                null_fds = SuppressStdoutStderr._null_fds
+                if saved_fds is not None and null_fds is not None:
+                    # Re-assign the real stdout/stderr back to (1) and (2)
+                    os.dup2(saved_fds[0], 1)
+                    os.dup2(saved_fds[1], 2)
+                    # Close all file descriptors
+                    for fd in saved_fds + null_fds:
+                        with suppress(OSError):
+                            os.close(fd)
+                    SuppressStdoutStderr._saved_fds = None
+                    SuppressStdoutStderr._null_fds = None
 
 
 def execute_and_suppress_output(function, logger, suppression_threshold_level, *args):

darts/tests/test_logging.py

@@ -1,13 +1,21 @@
+import errno
 import logging
+import os
 import re
+import threading
 
 import numpy as np
 import pandas as pd
 import pytest
 from testfixtures import LogCapture
 
 from darts import TimeSeries
-from darts.logging import get_logger, raise_log, time_log
+from darts.logging import (
+    SuppressStdoutStderr,
+    get_logger,
+    raise_log,
+    time_log,
+)
 
 
 @pytest.fixture(scope="module", autouse=True)
@@ -87,3 +95,133 @@ def _my_timed_fn():
         )
         is not None
     )
+
+
+class _PipeOnFd1:
+    """
+    Point file descriptor 1 at the write end of a fresh pipe, so that
+    whatever a test writes to fd 1 can be read back afterwards. The saved
+    stream is restored on exit, and the pipe is only readable once fd 1
+    points elsewhere again.
+    """
+
+    def __enter__(self):
+        self.read_fd, write_fd = os.pipe()
+        self.saved_fd = os.dup(1)
+        os.dup2(write_fd, 1)
+        os.close(write_fd)
+        return self
+
+    def __exit__(self, *_):
+        os.dup2(self.saved_fd, 1)
+        os.close(self.saved_fd)
+
+    def drain(self):
+        data = b""
+        while True:
+            chunk = os.read(self.read_fd, 4096)
+            if not chunk:
+                break
+            data += chunk
+        os.close(self.read_fd)
+        return data
+
+
+def test_suppress_stdout_stderr_round_trip():
+    with _PipeOnFd1() as pipe:
+        with SuppressStdoutStderr():
+            os.write(1, b"suppressed")
+        os.write(1, b"restored")
+    assert pipe.drain() == b"restored"
+
+
+def test_suppress_stdout_stderr_threads_no_permanent_silencing():
+    """
+    Two threads use the context manager with overlapping lifetimes: the
+    first one to enter is still inside the redirection when the second one
+    enters, and the second one is the last to leave. After both are done,
+    fd 1 must still point at the real stream; with the pre-fix behaviour
+    the second context saved the devnull descriptors as if they were the
+    real ones and silenced the process for good (#3276).
+    """
+    a_entered = threading.Event()
+    b_entered = threading.Event()
+    a_exited = threading.Event()
+    b_exited = threading.Event()
+
+    def enter_and_exit(entered, may_exit, exited):
+        with SuppressStdoutStderr():
+            entered.set()
+            assert may_exit.wait(timeout=30)
+        exited.set()
+
+    with _PipeOnFd1() as pipe:
+        thread_a = threading.Thread(
+            target=enter_and_exit, args=(a_entered, b_entered, a_exited)
+        )
+        thread_b = threading.Thread(
+            target=enter_and_exit, args=(b_entered, a_exited, b_exited)
+        )
+
+        thread_a.start()
+        assert a_entered.wait(timeout=30)
+        thread_b.start()
+        assert b_entered.wait(timeout=30)
+        thread_a.join(timeout=30)
+        thread_b.join(timeout=30)
+
+        assert not thread_a.is_alive() and not thread_b.is_alive()
+        os.write(1, b"survivor")
+
+    assert pipe.drain() == b"survivor"
+
+
+def test_suppress_stdout_stderr_nested_only_outer_restores():
+    with _PipeOnFd1() as pipe:
+        with SuppressStdoutStderr():
+            os.write(1, b"outer")
+            with SuppressStdoutStderr():
+                os.write(1, b"inner")
+            # leaving the inner context must not restore the descriptors
+            # while the outer one is still active
+            os.write(1, b"between")
+        os.write(1, b"after")
+    assert pipe.drain() == b"after"
+
+
+def test_suppress_stdout_stderr_falls_back_without_real_streams(monkeypatch):
+    """
+    In interpreters without usable standard streams (pythonw on Windows,
+    GUI-embedded hosts) os.dup on fd 1/2 raises. The context must degrade
+    to a no-op instead of crashing, must not leak the descriptors opened
+    before the failure, and must not leave the shared state corrupted for
+    the next user.
+    """
+    opened_fds = []
+    real_open = os.open
+
+    def recording_open(path, flags, *args, **kwargs):
+        fd = real_open(path, flags, *args, **kwargs)
+        opened_fds.append(fd)
+        return fd
+
+    def bad_dup(fd):
+        raise OSError(errno.EBADF, "Bad file descriptor")
+
+    monkeypatch.setattr(os, "open", recording_open)
+    monkeypatch.setattr(os, "dup", bad_dup)
+
+    with SuppressStdoutStderr():
+        pass
+
+    monkeypatch.undo()
+
+    for fd in opened_fds:
+        with pytest.raises(OSError):
+            os.fstat(fd)
+
+    with _PipeOnFd1() as pipe:
+        with SuppressStdoutStderr():
+            os.write(1, b"suppressed")
+        os.write(1, b"restored")
+    assert pipe.drain() == b"restored"