From 2bc002cb4f6ce16c38a96d7427902e66868f315c Mon Sep 17 00:00:00 2001 From: jschen069 <3563624058@qq.com> Date: Thu, 23 Jul 2026 12:51:27 +0800 Subject: [PATCH 1/6] fix: add cross-platform file locking and background progress thread - file.py: Add OS-level file locking (msvcrt on Windows, fcntl on Unix) to prevent race conditions in concurrent JSON status file writes - base.py: Wrap progress monitoring in daemon thread with join to properly handle background task progress tracking --- ais_bench/benchmark/runners/base.py | 19 +++- ais_bench/benchmark/utils/file/file.py | 122 ++++++++++++++++++------- 2 files changed, 106 insertions(+), 35 deletions(-) diff --git a/ais_bench/benchmark/runners/base.py b/ais_bench/benchmark/runners/base.py index 5494815b..20dd6a39 100644 --- a/ais_bench/benchmark/runners/base.py +++ b/ais_bench/benchmark/runners/base.py @@ -2,6 +2,7 @@ import time import psutil import shutil +import threading from tqdm import tqdm from abc import abstractmethod from typing import Any, Dict, List, Tuple @@ -56,6 +57,9 @@ def __init__(self, self.refresh_interval = refresh_interval self.run_in_background = self.is_running_in_background() if not self.is_debug else True self.last_table = None + + self._progress_thread = None + self.logger.info(f"Launch TasksMonitor, " f"PID: {os.getpid()}, " f"Refresh interval: {self.refresh_interval}, " @@ -90,7 +94,14 @@ def launch_state_board(self): print(self.last_table) else: self.logger.debug("Running task progress monitor in background mode") - self._update_tasks_progress() + self._progress_thread = threading.Thread( + target=self._update_tasks_progress, + name="ProgressMonitor", + daemon=True + ) + self._progress_thread.start() + # Wait for the progress thread to finish (all tasks done) + self._progress_thread.join() def _is_all_task_done(self): unfinished_tasks = [] @@ -168,14 +179,18 @@ def _get_task_states(self): def _update_tasks_progress(self): pbar = tqdm(total=len(self.tasks_state_map), desc="Monitoring tasks progress") + headers = ["Task Name", "Process", "Progress", "Time Cost", "Status", "Log Path", "Extend Parameters"] while True: self._refresh_task_state() - _ = self._get_task_states() + data = self._get_task_states() cur_count = 0 for _, state in self.tasks_state_map.items(): if state.get("status") == "finish" or state.get("status") == "error": cur_count += 1 + # Format progress table + full_table = tabulate(data, headers=headers, tablefmt="grid") + if cur_count > pbar.n: pbar.update(cur_count - pbar.n) # break when all the task finished diff --git a/ais_bench/benchmark/utils/file/file.py b/ais_bench/benchmark/utils/file/file.py index f15a26c4..56654641 100644 --- a/ais_bench/benchmark/utils/file/file.py +++ b/ais_bench/benchmark/utils/file/file.py @@ -1,3 +1,4 @@ +import sys from typing import List, Tuple, Union import os import json @@ -20,9 +21,46 @@ logger = AISLogger() + +# --------------------------------------------------------------------------- +# Cross-platform file locking +# --------------------------------------------------------------------------- + +if sys.platform == 'win32': + import msvcrt + + def _lock_file(f): + """Blocking exclusive lock on an open file (Windows).""" + f.seek(0) + msvcrt.locking(f.fileno(), msvcrt.LK_LOCK, 1) + + def _unlock_file(f): + """Release lock on an open file (Windows).""" + try: + f.seek(0) + msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, 1) + except IOError: + pass +else: + import fcntl + + def _lock_file(f): + """Blocking exclusive lock on an open file (Unix).""" + fcntl.lockf(f.fileno(), fcntl.LOCK_EX) + + def _unlock_file(f): + """Release lock on an open file (Unix).""" + try: + fcntl.lockf(f.fileno(), fcntl.LOCK_UN) + except IOError: + pass + def write_status(file_path, status): """Write status to a JSON file, appending to existing content. + Uses file locking to safely handle concurrent writes from multiple + processes to the same file (read-modify-write is atomic). + Args: file_path: Path to the status file status: Status data to append @@ -30,32 +68,38 @@ def write_status(file_path, status): Returns: bool: True if successful, False otherwise """ - # read existing content - existing_data = [] - if os.path.exists(file_path): + # Ensure the file exists before opening in r+ mode + if not os.path.exists(file_path): try: - with open(file_path, 'r', encoding='utf-8') as f: - existing_data = json.load(f) - except json.JSONDecodeError as e: - logger.warning( - f"Failed to parse JSON from status file '{file_path}': {e}. " - f"Starting with empty status list." - ) - existing_data = [] - except IOError as e: - logger.warning( - f"Failed to read status file '{file_path}': {e}. " - f"Starting with empty status list." - ) - existing_data = [] - - # add new status - existing_data.append(status) + with open(file_path, 'w', encoding='utf-8') as f: + json.dump([], f) + except IOError: + pass - # write to file try: - with open(file_path, 'w', encoding='utf-8') as f: - json.dump(existing_data, f) + with open(file_path, 'r+', encoding='utf-8') as f: + _lock_file(f) + try: + # read existing content + content = f.read() + if content.strip(): + try: + existing_data = json.loads(content) + except json.JSONDecodeError: + existing_data = [] + else: + existing_data = [] + + # add new status + existing_data.append(status) + + # write back + f.seek(0) + f.truncate() + json.dump(existing_data, f) + f.flush() + finally: + _unlock_file(f) return True except IOError as e: logger.warning(f"Failed to write status to '{file_path}': {e}") @@ -82,17 +126,29 @@ def read_and_clear_statuses(tmp_file_dir, tmp_file_name_list): logger.debug(f"Reading and clearing {len(abs_path_list)} status files from '{tmp_file_dir}'") for tmp_file in abs_path_list: - try: - # read existing content - with open(tmp_file, 'r', encoding='utf-8') as f: - data = json.load(f) - - status_count = len(data) - all_status.extend(data) + if not os.path.exists(tmp_file): + continue - # clear file content - with open(tmp_file, 'w', encoding='utf-8') as f: - json.dump([], f) + try: + with open(tmp_file, 'r+', encoding='utf-8') as f: + _lock_file(f) + try: + content = f.read() + if content.strip(): + data = json.loads(content) + else: + data = [] + + status_count = len(data) + all_status.extend(data) + + # clear file content + f.seek(0) + f.truncate() + json.dump([], f) + f.flush() + finally: + _unlock_file(f) logger.debug(f"Read {status_count} statuses from '{tmp_file}' and cleared file") From c41b9d6d0dbe7742714544853bbfe084f5f6972e Mon Sep 17 00:00:00 2001 From: jschen69 <3563624058@qq.com> Date: Fri, 24 Jul 2026 17:39:38 +0800 Subject: [PATCH 2/6] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E7=94=A8=E4=BE=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ais_bench/benchmark/utils/file/file.py | 6 +++++- tests/UT/utils/file/test_file.py | 14 ++++++++++---- 2 files changed, 15 insertions(+), 5 deletions(-) diff --git a/ais_bench/benchmark/utils/file/file.py b/ais_bench/benchmark/utils/file/file.py index 56654641..4452316a 100644 --- a/ais_bench/benchmark/utils/file/file.py +++ b/ais_bench/benchmark/utils/file/file.py @@ -85,7 +85,11 @@ def write_status(file_path, status): if content.strip(): try: existing_data = json.loads(content) - except json.JSONDecodeError: + except json.JSONDecodeError as e: + logger.warning( + f"Failed to parse JSON from status file '{file_path}': {e}. " + f"Starting with empty status list." + ) existing_data = [] else: existing_data = [] diff --git a/tests/UT/utils/file/test_file.py b/tests/UT/utils/file/test_file.py index 3b33dce3..b5d5d967 100644 --- a/tests/UT/utils/file/test_file.py +++ b/tests/UT/utils/file/test_file.py @@ -120,11 +120,17 @@ def test_ioerror_on_clear_logs_warning(self): with open(p, "w", encoding="utf-8") as f: json.dump([{"k": 1}], f) - # Mock open to raise IOError when writing to the file + # Mock the file object's flush to raise IOError during the clear phase, + # since the new code uses a single r+ handle with truncate+flush. original_open = open - def mock_open(file_path, mode='r', *args, **kwargs): - if mode == 'w' and file_path == p: - raise IOError("Permission denied") + def mock_open(file_path, mode='r+', *args, **kwargs): + if file_path == p: + real_file = original_open(file_path, mode, *args, **kwargs) + original_flush = real_file.flush + def failing_flush(*a, **kw): + raise IOError("Permission denied") + real_file.flush = failing_flush + return real_file return original_open(file_path, mode, *args, **kwargs) with patch('builtins.open', side_effect=mock_open): From 48feb847d0bedb8f9dedb907794b99268ef664e9 Mon Sep 17 00:00:00 2001 From: jschen069 <3563624058@qq.com> Date: Fri, 24 Jul 2026 18:06:14 +0800 Subject: [PATCH 3/6] refactor: remove win32 cross-platform file locking, keep fcntl only for Linux --- ais_bench/benchmark/utils/file/file.py | 38 +++++++------------------- 1 file changed, 10 insertions(+), 28 deletions(-) diff --git a/ais_bench/benchmark/utils/file/file.py b/ais_bench/benchmark/utils/file/file.py index 4452316a..e5ce9af6 100644 --- a/ais_bench/benchmark/utils/file/file.py +++ b/ais_bench/benchmark/utils/file/file.py @@ -22,38 +22,20 @@ logger = AISLogger() -# --------------------------------------------------------------------------- -# Cross-platform file locking -# --------------------------------------------------------------------------- +import fcntl -if sys.platform == 'win32': - import msvcrt - def _lock_file(f): - """Blocking exclusive lock on an open file (Windows).""" - f.seek(0) - msvcrt.locking(f.fileno(), msvcrt.LK_LOCK, 1) +def _lock_file(f): + """Blocking exclusive lock on an open file.""" + fcntl.lockf(f.fileno(), fcntl.LOCK_EX) - def _unlock_file(f): - """Release lock on an open file (Windows).""" - try: - f.seek(0) - msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, 1) - except IOError: - pass -else: - import fcntl - - def _lock_file(f): - """Blocking exclusive lock on an open file (Unix).""" - fcntl.lockf(f.fileno(), fcntl.LOCK_EX) - def _unlock_file(f): - """Release lock on an open file (Unix).""" - try: - fcntl.lockf(f.fileno(), fcntl.LOCK_UN) - except IOError: - pass +def _unlock_file(f): + """Release lock on an open file.""" + try: + fcntl.lockf(f.fileno(), fcntl.LOCK_UN) + except IOError: + pass def write_status(file_path, status): """Write status to a JSON file, appending to existing content. From 1258b155dfda73b26b61114c91da4b8824bd6a45 Mon Sep 17 00:00:00 2001 From: jschen069 <3563624058@qq.com> Date: Sat, 25 Jul 2026 09:15:35 +0800 Subject: [PATCH 4/6] revert: remove background progress thread changes from commit 2bc002c --- ais_bench/benchmark/runners/base.py | 19 ++----------------- 1 file changed, 2 insertions(+), 17 deletions(-) diff --git a/ais_bench/benchmark/runners/base.py b/ais_bench/benchmark/runners/base.py index 20dd6a39..5494815b 100644 --- a/ais_bench/benchmark/runners/base.py +++ b/ais_bench/benchmark/runners/base.py @@ -2,7 +2,6 @@ import time import psutil import shutil -import threading from tqdm import tqdm from abc import abstractmethod from typing import Any, Dict, List, Tuple @@ -57,9 +56,6 @@ def __init__(self, self.refresh_interval = refresh_interval self.run_in_background = self.is_running_in_background() if not self.is_debug else True self.last_table = None - - self._progress_thread = None - self.logger.info(f"Launch TasksMonitor, " f"PID: {os.getpid()}, " f"Refresh interval: {self.refresh_interval}, " @@ -94,14 +90,7 @@ def launch_state_board(self): print(self.last_table) else: self.logger.debug("Running task progress monitor in background mode") - self._progress_thread = threading.Thread( - target=self._update_tasks_progress, - name="ProgressMonitor", - daemon=True - ) - self._progress_thread.start() - # Wait for the progress thread to finish (all tasks done) - self._progress_thread.join() + self._update_tasks_progress() def _is_all_task_done(self): unfinished_tasks = [] @@ -179,18 +168,14 @@ def _get_task_states(self): def _update_tasks_progress(self): pbar = tqdm(total=len(self.tasks_state_map), desc="Monitoring tasks progress") - headers = ["Task Name", "Process", "Progress", "Time Cost", "Status", "Log Path", "Extend Parameters"] while True: self._refresh_task_state() - data = self._get_task_states() + _ = self._get_task_states() cur_count = 0 for _, state in self.tasks_state_map.items(): if state.get("status") == "finish" or state.get("status") == "error": cur_count += 1 - # Format progress table - full_table = tabulate(data, headers=headers, tablefmt="grid") - if cur_count > pbar.n: pbar.update(cur_count - pbar.n) # break when all the task finished From 3bbb813a839e215005d6e579f3eea62f072017e2 Mon Sep 17 00:00:00 2001 From: jschen069 <3563624058@qq.com> Date: Sat, 25 Jul 2026 15:10:23 +0800 Subject: [PATCH 5/6] =?UTF-8?q?=E5=88=A0=E9=99=A4=E6=97=A0=E7=94=A8?= =?UTF-8?q?=E5=AF=BC=E5=85=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ais_bench/benchmark/utils/file/file.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/ais_bench/benchmark/utils/file/file.py b/ais_bench/benchmark/utils/file/file.py index e5ce9af6..77a8ef13 100644 --- a/ais_bench/benchmark/utils/file/file.py +++ b/ais_bench/benchmark/utils/file/file.py @@ -1,4 +1,3 @@ -import sys from typing import List, Tuple, Union import os import json @@ -34,8 +33,8 @@ def _unlock_file(f): """Release lock on an open file.""" try: fcntl.lockf(f.fileno(), fcntl.LOCK_UN) - except IOError: - pass + except IOError as e: + logger.warning(f"Failed to release file lock: {e}") def write_status(file_path, status): """Write status to a JSON file, appending to existing content. From cf3cd251ccc01e776cc324adcf99b7e91dfb5f8a Mon Sep 17 00:00:00 2001 From: jschen069 <3563624058@qq.com> Date: Sat, 25 Jul 2026 15:32:58 +0800 Subject: [PATCH 6/6] =?UTF-8?q?=E6=89=93=E5=8D=B0=E6=97=A5=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ais_bench/benchmark/utils/file/file.py | 1 + 1 file changed, 1 insertion(+) diff --git a/ais_bench/benchmark/utils/file/file.py b/ais_bench/benchmark/utils/file/file.py index 77a8ef13..02062551 100644 --- a/ais_bench/benchmark/utils/file/file.py +++ b/ais_bench/benchmark/utils/file/file.py @@ -35,6 +35,7 @@ def _unlock_file(f): fcntl.lockf(f.fileno(), fcntl.LOCK_UN) except IOError as e: logger.warning(f"Failed to release file lock: {e}") + def write_status(file_path, status): """Write status to a JSON file, appending to existing content.