diff --git a/CHANGELOG.md b/CHANGELOG.md index ab4c188e96..1158eb5304 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,7 @@ but cannot always guarantee backwards compatibility. Changes that may **break co - Fixed `TimeSeries.strip()` and `TimeSeries.longest_contiguous_slice()` returning unstripped NaNs on all-NaN series or series with all-partial-NaNs when `mode="any"`. [#3285](https://github.com/unit8co/darts/pull/3285) by [Shyam Sharma](https://github.com/shyamsharmas124-commits). - Fixed `ConformalNaiveModel` and `ConformalQRModel` accepting classification models (e.g. `SKLearnClassifierModel`), which resulted in meaningless intervals around the predicted class labels. A `ValueError` is now raised, since conformal models only support regression models. [#3282](https://github.com/unit8co/darts/pull/3282) by [Shiven Aggarwal](https://github.com/shivenaggarwal). - Fixed `ForecastingModel.historical_forecasts()` with `retrain=True` mutating caller-provided `Pipeline` instances under `data_transformers`. [#3312](https://github.com/unit8co/darts/pull/3312) by [Pallav Biyala](https://github.com/Pallav-Biyala). +- Fixed `darts.logging.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). [#3290](https://github.com/unit8co/darts/pull/3290) by [Manas Choksi](https://github.com/choksi2212). **Dependencies** diff --git a/darts/logging.py b/darts/logging.py index c278f2134d..f9f4a2aa9c 100644 --- a/darts/logging.py +++ b/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): diff --git a/darts/tests/test_logging.py b/darts/tests/test_logging.py index 500f87c65b..84fd61c638 100644 --- a/darts/tests/test_logging.py +++ b/darts/tests/test_logging.py @@ -1,5 +1,8 @@ +import errno import logging +import os import re +import threading import numpy as np import pandas as pd @@ -7,7 +10,12 @@ 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"