diff --git a/xinference/deploy/test/test_log_rotation.py b/xinference/deploy/test/test_log_rotation.py index 87bda22c16..b68f658f50 100644 --- a/xinference/deploy/test/test_log_rotation.py +++ b/xinference/deploy/test/test_log_rotation.py @@ -15,12 +15,15 @@ import logging import os +import shutil import sys +import threading import time from unittest import mock import pytest +from .. import utils as deploy_utils from ..utils import SafeTimedAndSizeRotatingFileHandler, get_config_dict _skip_on_windows = pytest.mark.skipif( @@ -450,6 +453,227 @@ def test_rolling_rename_on_size_trigger(self, tmp_path): assert "test.log.2" in files +class _FakeMsvcrt: + LK_LOCK = 1 + LK_UNLCK = 2 + + def __init__(self): + self.actions = [] + + def locking(self, fd, mode, count): + self.actions.append((fd, mode, count)) + + +class _BlockingFakeMsvcrt(_FakeMsvcrt): + def __init__(self): + super().__init__() + self.lock = threading.Lock() + self.acquire_attempted = threading.Event() + + def locking(self, fd, mode, count): + super().locking(fd, mode, count) + if mode == self.LK_LOCK: + self.acquire_attempted.set() + self.lock.acquire() + else: + self.lock.release() + + +class TestWindowsRotationFallback: + def test_lock_file_is_closed_when_locking_fails(self, tmp_path): + from ..utils import SafeRotatingFileHandler + + class _FailingMsvcrt(_FakeMsvcrt): + def locking(self, fd, mode, count): + raise OSError("lock failed") + + log_path = str(tmp_path / "test.log") + h = SafeRotatingFileHandler(log_path, maxBytes=100, backupCount=1) + opened = [] + real_open = open + + def tracking_open(*args, **kwargs): + lock_fd = real_open(*args, **kwargs) + opened.append(lock_fd) + return lock_fd + + with ( + mock.patch.object(deploy_utils, "msvcrt", _FailingMsvcrt()), + mock.patch("builtins.open", side_effect=tracking_open), + ): + with pytest.raises(OSError, match="lock failed"): + h._acquire_rotation_lock() + h.close() + + assert len(opened) == 1 + assert opened[0].closed + + def test_msvcrt_lock_is_acquired_and_released(self, tmp_path): + from ..utils import SafeRotatingFileHandler + + fake_msvcrt = _FakeMsvcrt() + log_path = str(tmp_path / "test.log") + with mock.patch.object(deploy_utils, "msvcrt", fake_msvcrt): + h = SafeRotatingFileHandler(log_path, maxBytes=100, backupCount=1) + fake_msvcrt.actions.clear() + lock_fd = h._acquire_rotation_lock() + h._release_rotation_lock(lock_fd) + h.close() + + assert [action[1] for action in fake_msvcrt.actions] == [ + fake_msvcrt.LK_LOCK, + fake_msvcrt.LK_UNLCK, + ] + assert all(action[2] == 1 for action in fake_msvcrt.actions) + + def test_copy_truncate_fallback_preserves_live_file(self, tmp_path): + from ..utils import SafeRotatingFileHandler + + fake_msvcrt = _FakeMsvcrt() + log_path = str(tmp_path / "test.log") + archive_path = str(tmp_path / "test.log.1") + with mock.patch.object(deploy_utils, "msvcrt", fake_msvcrt): + h = SafeRotatingFileHandler(log_path, maxBytes=1, backupCount=1) + h.stream.write("before rotation\n") + h.stream.flush() + with mock.patch( + "logging.handlers.BaseRotatingHandler.rotate", + side_effect=PermissionError(13, "file is in use"), + ): + h.doRollover() + h.close() + + with open(archive_path, encoding="utf8") as archive: + assert archive.read() == "before rotation\n" + with open(log_path, encoding="utf8") as live: + assert live.read() == "" + + def test_copy_truncate_blocks_concurrent_writes(self, tmp_path): + from ..utils import SafeRotatingFileHandler + + fake_msvcrt = _BlockingFakeMsvcrt() + log_path = str(tmp_path / "test.log") + archive_path = tmp_path / "test.log.1" + copy_finished = threading.Event() + allow_truncate = threading.Event() + errors = [] + real_copy2 = shutil.copy2 + + def pausing_copy(source, dest): + result = real_copy2(source, dest) + copy_finished.set() + if not allow_truncate.wait(timeout=5): + raise TimeoutError("test did not release the truncate step") + return result + + def capture_errors(target): + try: + target() + except BaseException as exc: + errors.append(exc) + + with mock.patch.object(deploy_utils, "msvcrt", fake_msvcrt): + h1 = SafeRotatingFileHandler(log_path, maxBytes=1000, backupCount=1) + h2 = SafeRotatingFileHandler(log_path, maxBytes=1000, backupCount=1) + h1.setFormatter(logging.Formatter("%(message)s")) + h2.setFormatter(logging.Formatter("%(message)s")) + h1.emit(_make_record("before rotation")) + + with ( + mock.patch( + "logging.handlers.BaseRotatingHandler.rotate", + side_effect=PermissionError(13, "file is in use"), + ), + mock.patch("shutil.copy2", side_effect=pausing_copy), + ): + rotate_thread = threading.Thread( + target=capture_errors, args=(h1.doRollover,) + ) + rotate_thread.start() + assert copy_finished.wait(timeout=5) + + fake_msvcrt.acquire_attempted.clear() + write_thread = threading.Thread( + target=capture_errors, + args=(lambda: h2.emit(_make_record("during rotation")),), + ) + write_thread.start() + assert fake_msvcrt.acquire_attempted.wait(timeout=5) + assert write_thread.is_alive() + + allow_truncate.set() + rotate_thread.join(timeout=5) + write_thread.join(timeout=5) + + h1.close() + h2.close() + + assert not rotate_thread.is_alive() + assert not write_thread.is_alive() + assert errors == [] + assert archive_path.read_text(encoding="utf8") == "before rotation\n" + assert (tmp_path / "test.log").read_text(encoding="utf8") == ( + "during rotation\n" + ) + + def test_size_rotation_generation_prevents_duplicate_archive(self, tmp_path): + from ..utils import SafeRotatingFileHandler + + fake_msvcrt = _FakeMsvcrt() + log_path = str(tmp_path / "test.log") + with mock.patch.object(deploy_utils, "msvcrt", fake_msvcrt): + h1 = SafeRotatingFileHandler(log_path, maxBytes=1, backupCount=2) + h2 = SafeRotatingFileHandler(log_path, maxBytes=1, backupCount=2) + h1.stream.write("one archive only\n") + h1.stream.flush() + with mock.patch( + "logging.handlers.BaseRotatingHandler.rotate", + side_effect=PermissionError(13, "file is in use"), + ): + h1.doRollover() + h2.doRollover() + h1.close() + h2.close() + + assert (tmp_path / "test.log.1").read_text(encoding="utf8") == ( + "one archive only\n" + ) + assert not (tmp_path / "test.log.2").exists() + + def test_timed_rotation_marker_prevents_archive_clobber(self, tmp_path): + fake_msvcrt = _FakeMsvcrt() + log_path = str(tmp_path / "test.log") + with mock.patch.object(deploy_utils, "msvcrt", fake_msvcrt): + h1 = SafeTimedAndSizeRotatingFileHandler( + log_path, when="midnight", backupCount=2, maxBytes=1000 + ) + h2 = SafeTimedAndSizeRotatingFileHandler( + log_path, when="midnight", backupCount=2, maxBytes=1000 + ) + scheduled = int(time.time()) - 1 + h1.rolloverAt = scheduled + h2.rolloverAt = scheduled + h1.stream.write("keep this archive\n") + h1.stream.flush() + with mock.patch( + "logging.handlers.BaseRotatingHandler.rotate", + side_effect=PermissionError(13, "file is in use"), + ): + h1.doRollover() + h2.doRollover() + h1.close() + h2.close() + + archives = [ + path + for path in tmp_path.iterdir() + if path.name.startswith("test.log.") and not path.name.endswith(".lock") + ] + assert len(archives) == 1 + assert archives[0].read_text(encoding="utf8") == "keep this archive\n" + assert h2.rolloverAt > time.time() + + def _mp_worker_size(proc_id, log_path, max_bytes, n_records): """Module-level worker for multiprocessing (size mode).""" from ..utils import SafeRotatingFileHandler diff --git a/xinference/deploy/utils.py b/xinference/deploy/utils.py index f1036b3892..aa6ca8098c 100644 --- a/xinference/deploy/utils.py +++ b/xinference/deploy/utils.py @@ -27,9 +27,12 @@ if sys.platform == "win32": fcntl = None + import msvcrt else: import fcntl + msvcrt = None + import xoscar as xo from ..constants import ( @@ -45,7 +48,136 @@ logger = logging.getLogger(__name__) -class SafeRotatingFileHandler(logging.handlers.RotatingFileHandler): +class _SafeFileRotationMixin: + """Cross-process coordination shared by size and timed handlers.""" + + def _init_rotation_coordination(self, filename): + directory = os.path.dirname(filename) + basename = os.path.basename(filename) + self._lock_path = os.path.join(directory, basename + ".rotate.lock") + with open(self._lock_path, "a+b") as lock_fd: + if os.fstat(lock_fd.fileno()).st_size == 0: + lock_fd.write(b"\0") + lock_fd.flush() + + lock_fd = self._acquire_rotation_lock() + try: + generation, _ = self._read_rotation_state(lock_fd) + self._rotation_generation = generation + finally: + self._release_rotation_lock(lock_fd) + + def _acquire_rotation_lock(self): + active_lock = getattr(self, "_rotation_lock_fd", None) + if active_lock is not None: + self._rotation_lock_depth += 1 + return active_lock + + lock_fd = open(self._lock_path, "r+b") + try: + if msvcrt is not None: + lock_fd.seek(0) + msvcrt.locking(lock_fd.fileno(), msvcrt.LK_LOCK, 1) + else: + fcntl.flock(lock_fd, fcntl.LOCK_EX) + except BaseException: + lock_fd.close() + raise + self._rotation_lock_fd = lock_fd + self._rotation_lock_depth = 1 + return lock_fd + + def _release_rotation_lock(self, lock_fd): + if lock_fd is None: + return + if lock_fd is getattr(self, "_rotation_lock_fd", None): + self._rotation_lock_depth -= 1 + if self._rotation_lock_depth > 0: + return + self._rotation_lock_fd = None + self._rotation_lock_depth = 0 + try: + if msvcrt is not None: + lock_fd.seek(0) + msvcrt.locking(lock_fd.fileno(), msvcrt.LK_UNLCK, 1) + else: + fcntl.flock(lock_fd, fcntl.LOCK_UN) + finally: + lock_fd.close() + + def emit(self, record): + if msvcrt is None: + return super().emit(record) + + # Windows cannot rename a log file held open by sibling processes, so + # rotate() may need to copy and truncate it in place. Coordinate the + # entire rollover check and write to prevent a sibling from writing in + # the otherwise lossy interval between that copy and truncate. + try: + lock_fd = self._acquire_rotation_lock() + try: + return super().emit(record) + finally: + self._release_rotation_lock(lock_fd) + except Exception: + self.handleError(record) + + def _read_rotation_state(self, lock_fd): + try: + lock_fd.seek(1) + payload = lock_fd.read().decode("ascii") + state = json.loads(payload) if payload else {} + generation = state.get("generation", 0) + timed_rollover = state.get("timed_rollover", 0) + if not isinstance(generation, int) or generation < 0: + generation = 0 + if not isinstance(timed_rollover, (int, float)) or timed_rollover < 0: + timed_rollover = 0 + return generation, timed_rollover + except (OSError, UnicodeError, ValueError, TypeError): + return 0, 0 + + def _write_rotation_state(self, lock_fd, generation, timed_rollover): + payload = json.dumps( + { + "generation": generation, + "timed_rollover": timed_rollover, + }, + separators=(",", ":"), + ).encode("ascii") + lock_fd.seek(1) + lock_fd.truncate() + lock_fd.write(payload) + lock_fd.flush() + os.fsync(lock_fd.fileno()) + + def _record_rotation(self, lock_fd, state, timed_rollover=None): + generation, last_timed_rollover = state + generation += 1 + if timed_rollover is not None: + last_timed_rollover = max(last_timed_rollover, timed_rollover) + self._write_rotation_state(lock_fd, generation, last_timed_rollover) + self._rotation_generation = generation + + def rotate(self, source, dest): + try: + super().rotate(source, dest) + except PermissionError: + if msvcrt is None or self.rotator is not None: + raise + # Windows refuses to rename a log held by sibling processes. Copy + # the archive first, then truncate the live file in place so every + # process keeps writing through its existing handle. + import shutil + + shutil.copy2(source, dest) + with open(source, "r+b") as source_file: + source_file.truncate(0) + + +class SafeRotatingFileHandler( + _SafeFileRotationMixin, logging.handlers.RotatingFileHandler +): """RotatingFileHandler that auto-creates parent directories. Python's standard RotatingFileHandler raises FileNotFoundError if @@ -56,46 +188,24 @@ class SafeRotatingFileHandler(logging.handlers.RotatingFileHandler): calls ``dictConfig``. This subclass ensures the directory is (re-)created before the file is opened. - Multi-process safety (2026-06-25): - 1. fcntl file lock serializes ``doRollover`` across processes. + Multi-process safety: + 1. A platform-native file lock serializes ``doRollover`` across processes. + On Windows it also serializes normal writes with copy/truncate rotation. 2. ``shouldRollover`` checks inode at entry; if another process renamed the file, reopen the stream. 3. Size check uses ``os.fstat().st_size`` instead of ``stream.tell()`` so it reflects the true file size (including writes from other processes). + 4. On Windows, an in-place copy/truncate fallback preserves open sibling + handles, while a shared generation prevents duplicate rotations. - Assumption: log directory is on a local filesystem (fcntl.flock - is unreliable on NFS). - - Windows caveat: ``fcntl`` is unavailable on Windows, so the - rotation lock is a no-op there. Single-process use is unaffected; - multi-process use on Windows may regress to lost archives or - cleanup of files still held by another process. + Assumption: the log directory is on a local filesystem. """ def __init__(self, filename, *args, **kwargs): os.makedirs(os.path.dirname(filename), exist_ok=True) super().__init__(filename, *args, **kwargs) - self._lock_path = os.path.join( - os.path.dirname(filename), - os.path.basename(filename) + ".rotate.lock", - ) - # Pre-create the lock file so it exists even if doRollover is never - # called (e.g. short-lived processes). The file is empty and harmless. - open(self._lock_path, "a").close() - - def _acquire_rotation_lock(self): - if fcntl is None: - return None - lock_fd = open(self._lock_path, "w") - fcntl.flock(lock_fd, fcntl.LOCK_EX) - return lock_fd - - def _release_rotation_lock(self, lock_fd): - if lock_fd is None: - return - fcntl.flock(lock_fd, fcntl.LOCK_UN) - lock_fd.close() + self._init_rotation_coordination(filename) def _check_inode_and_reopen(self): """Reopen stream if another process renamed the base file. @@ -140,6 +250,11 @@ def shouldRollover(self, record): def doRollover(self): lock_fd = self._acquire_rotation_lock() try: + state = self._read_rotation_state(lock_fd) + if state[0] != self._rotation_generation: + self._rotation_generation = state[0] + self._check_inode_and_reopen() + return # If another process rotated the file while we were waiting # for the lock, the stream has been reopened to a fresh file # and there is nothing to do. @@ -148,6 +263,7 @@ def doRollover(self): # Trust shouldRollover's judgment: if we got here, either time # or size trigger fired and no other process has rotated since. self._do_rolling_rollover() + self._record_rotation(lock_fd, state) finally: self._release_rotation_lock(lock_fd) @@ -171,12 +287,65 @@ def _do_rolling_rollover(self): self.stream = self._open() -class SafeTimedRotatingFileHandler(logging.handlers.TimedRotatingFileHandler): - """TimedRotatingFileHandler that auto-creates parent directories.""" +class SafeTimedRotatingFileHandler( + _SafeFileRotationMixin, logging.handlers.TimedRotatingFileHandler +): + """Timed handler with cross-platform, cross-process rotation safety.""" def __init__(self, filename, *args, **kwargs): os.makedirs(os.path.dirname(filename), exist_ok=True) super().__init__(filename, *args, **kwargs) + self._init_rotation_coordination(filename) + + def _check_inode_and_reopen(self): + if self.stream is None: + return False + try: + current_inode = os.stat(self.baseFilename).st_ino + stream_inode = os.fstat(self.stream.fileno()).st_ino + if current_inode != stream_inode: + self.stream.close() + self.stream = self._open() + if time.time() >= self.rolloverAt: + self.rolloverAt = self.computeRollover(time.time()) + return True + except OSError: + try: + self.stream.close() + except Exception: + pass + self.stream = None + return False + + def _foreign_rotation_handled(self, state, current_time): + self._rotation_generation = state[0] + self._check_inode_and_reopen() + if current_time < self.rolloverAt: + return True + if state[1] >= self.rolloverAt: + self.rolloverAt = self.computeRollover(current_time) + return True + return False + + def shouldRollover(self, record): + self._check_inode_and_reopen() + return super().shouldRollover(record) + + def doRollover(self): + lock_fd = self._acquire_rotation_lock() + try: + state = self._read_rotation_state(lock_fd) + current_time = time.time() + if state[0] != self._rotation_generation: + if self._foreign_rotation_handled(state, current_time): + return + elif self._check_inode_and_reopen(): + return + scheduled_rollover = self.rolloverAt + logging.handlers.TimedRotatingFileHandler.doRollover(self) + self._record_rotation(lock_fd, state, scheduled_rollover) + finally: + self._release_rotation_lock(lock_fd) class SafeTimedAndSizeRotatingFileHandler(SafeTimedRotatingFileHandler): @@ -196,21 +365,19 @@ class SafeTimedAndSizeRotatingFileHandler(SafeTimedRotatingFileHandler): ``%Y-%m-%d``). The handler is only used with midnight rotation in ``get_config_dict()``, so this assumption holds. - Multi-process safety (2026-06-25): - 1. fcntl file lock serializes ``doRollover`` across processes. + Multi-process safety: + 1. A platform-native file lock serializes ``doRollover`` across processes. + On Windows it also serializes normal writes with copy/truncate rotation. 2. ``shouldRollover`` checks inode at entry; if another process renamed the file, reopen the stream. 3. Size check uses ``os.fstat().st_size`` instead of ``stream.tell()`` so it reflects the true file size (including writes from other processes). - Assumption: log directory is on a local filesystem (fcntl.flock - is unreliable on NFS). + 4. On Windows, copy/truncate keeps sibling handles valid and shared state + distinguishes time rotations from size rotations. - Windows caveat: ``fcntl`` is unavailable on Windows, so the - rotation lock is a no-op there. Single-process use is unaffected; - multi-process use on Windows may regress to lost archives or - cleanup of files still held by another process. + Assumption: the log directory is on a local filesystem. """ _rotated_re = re.compile(r"^\d{4}-\d{2}-\d{2}(\.\d+)?$") @@ -242,62 +409,10 @@ def __init__( atTime=atTime, errors=errors, ) - self._lock_path = os.path.join( - os.path.dirname(filename), - os.path.basename(filename) + ".rotate.lock", - ) - # Pre-create the lock file so it exists even if doRollover is never - # called (e.g. short-lived processes). The file is empty and harmless. - open(self._lock_path, "a").close() - - def _acquire_rotation_lock(self): - if fcntl is None: - return None - lock_fd = open(self._lock_path, "w") - fcntl.flock(lock_fd, fcntl.LOCK_EX) - return lock_fd - - def _release_rotation_lock(self, lock_fd): - if lock_fd is None: - return - fcntl.flock(lock_fd, fcntl.LOCK_UN) - lock_fd.close() - - def _check_inode_and_reopen(self): - """Reopen stream if another process renamed the base file. - - Returns True if the stream was reopened (i.e. another process - rotated the file), False otherwise. - """ - if self.stream is None: - return False - try: - current_inode = os.stat(self.baseFilename).st_ino - stream_inode = os.fstat(self.stream.fileno()).st_ino - if current_inode != stream_inode: - self.stream.close() - self.stream = self._open() - # Another process performed the time-based rollover for us. - # Advance rolloverAt so the next shouldRollover/doRollover - # does not see a stale past-due time and perform a redundant, - # destructive rotation (clobbers the archive just written by - # the other process). Only advance when we are actually past - # the scheduled time — a size-only foreign rotation leaves - # the pending time rollover intact. - if time.time() >= self.rolloverAt: - self.rolloverAt = self.computeRollover(time.time()) - return True - except OSError: - try: - self.stream.close() - except Exception: - pass - self.stream = None - return False def shouldRollover(self, record): self._check_inode_and_reopen() - if super().shouldRollover(record): + if logging.handlers.TimedRotatingFileHandler.shouldRollover(self, record): return True if self.maxBytes > 0: if self.stream is None: @@ -314,15 +429,23 @@ def shouldRollover(self, record): def doRollover(self): lock_fd = self._acquire_rotation_lock() try: + state = self._read_rotation_state(lock_fd) + current_time = time.time() + if state[0] != self._rotation_generation: + if self._foreign_rotation_handled(state, current_time): + return # If another process rotated the file while we were waiting # for the lock, the stream has been reopened to a fresh file # and there is nothing to do. - if self._check_inode_and_reopen(): + elif self._check_inode_and_reopen(): return - if time.time() >= self.rolloverAt: - super().doRollover() + if current_time >= self.rolloverAt: + scheduled_rollover = self.rolloverAt + logging.handlers.TimedRotatingFileHandler.doRollover(self) + self._record_rotation(lock_fd, state, scheduled_rollover) else: self._do_size_rollover() + self._record_rotation(lock_fd, state) finally: self._release_rotation_lock(lock_fd)