From e861344babe2bec1e10cbf0308df653dd021fc9a Mon Sep 17 00:00:00 2001 From: Chenpeng Su Date: Fri, 24 Jul 2026 15:28:29 +0800 Subject: [PATCH 1/3] Update test_client_session.py --- .../internal/pkg/client/test_client_session.py | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/tests/unit/sdk/internal/pkg/client/test_client_session.py b/tests/unit/sdk/internal/pkg/client/test_client_session.py index 8afc2ae7a..4342713bb 100644 --- a/tests/unit/sdk/internal/pkg/client/test_client_session.py +++ b/tests/unit/sdk/internal/pkg/client/test_client_session.py @@ -7,6 +7,8 @@ import pytest import responses +from requests import Request, Response +from requests.adapters import HTTPAdapter from swanlab.exceptions import ApiError from swanlab.sdk.internal.pkg.client.session import ( @@ -54,6 +56,19 @@ def test_session_send_success(session): assert resp.json() == {"ok": True} +def test_timeout_adapter_applies_default_when_requests_passes_none(mocker): + """requests 会显式传 timeout=None，此时 adapter 应替换为配置的默认超时。""" + adapter = TimeoutHTTPAdapter(timeout=30) + request = Request("GET", "http://api.test/data").prepare() + response = Response() + response.status_code = 200 + mock_send = mocker.patch.object(HTTPAdapter, "send", return_value=response) + + adapter.send(request, timeout=None) + + assert mock_send.call_args.kwargs["timeout"] == 30 + + @responses.activate() def test_session_send_api_error_custom_parsed(session): """测试 4xx/5xx 时，能否正确提取 JSON 中的错误信息并抛出 ApiError""" From 2fa14e1cf148b720ac6342c4498ec4645a61d58e Mon Sep 17 00:00:00 2001 From: Kang Li Date: Fri, 24 Jul 2026 15:50:58 +0800 Subject: [PATCH 2/3] Add debug logging throughout the terminal log pipeline --- swanlab/sdk/internal/core_python/core.py | 10 ++++++++++ swanlab/sdk/internal/core_python/store/__init__.py | 14 +++++++------- .../sdk/internal/core_python/transport/dispatch.py | 10 ++++++++++ .../sdk/internal/core_python/transport/sender.py | 3 +++ swanlab/sdk/internal/pkg/client/session.py | 4 ++-- .../internal/run/components/consumer/__init__.py | 3 +++ .../internal/run/components/terminal/__init__.py | 4 +++- 7 files changed, 38 insertions(+), 10 deletions(-) diff --git a/swanlab/sdk/internal/core_python/core.py b/swanlab/sdk/internal/core_python/core.py index 2cb8d8c9d..446251ce7 100644 --- a/swanlab/sdk/internal/core_python/core.py +++ b/swanlab/sdk/internal/core_python/core.py @@ -341,7 +341,17 @@ def _upsert_logs_when_offline(self, logs: List[LogRecord]) -> None: def _upsert_logs_when_online(self, logs: List[LogRecord]) -> None: records = [builder.build_log_record(self._counter, self._epoch, c) for c in logs] self._store_records(records) + if records: + console.debug( + f"Stored log records locally: count={len(records)}, nums={records[0].num}..{records[-1].num}", + write_to_tty=False, + ) self._transport_put(records) + if records: + console.debug( + f"Queued log records for upload: count={len(records)}, nums={records[0].num}..{records[-1].num}", + write_to_tty=False, + ) # ---- upsert_saves ---- diff --git a/swanlab/sdk/internal/core_python/store/__init__.py b/swanlab/sdk/internal/core_python/store/__init__.py index 4559e5580..6e4489fbd 100644 --- a/swanlab/sdk/internal/core_python/store/__init__.py +++ b/swanlab/sdk/internal/core_python/store/__init__.py @@ -89,13 +89,13 @@ def write(self, data: bytes) -> None: data_left -= LEVELDBLOG_DATA_LEN self._write_record(data[data_used:], LEVELDBLOG_LAST) # FIX: 【会导致磁盘高频 IO】每次 write 后统一 fsync，保证落盘 - try: - self._fp.flush() - # 指标量在较大的情况下高频 fsync 会影响上传效率 - # os.fsync(self._fp.fileno()) - except OSError: - pass - self._flush_offset = self._index + # try: + # self._fp.flush() + # # 指标量在较大的情况下高频 fsync 会影响上传效率 + # # os.fsync(self._fp.fileno()) + # except OSError: + # pass + # self._flush_offset = self._index def ensure_flushed(self) -> None: assert self._fp is not None, "writer is not open" diff --git a/swanlab/sdk/internal/core_python/transport/dispatch.py b/swanlab/sdk/internal/core_python/transport/dispatch.py index 8777fca02..953ef5377 100644 --- a/swanlab/sdk/internal/core_python/transport/dispatch.py +++ b/swanlab/sdk/internal/core_python/transport/dispatch.py @@ -68,9 +68,19 @@ def _upload_chunk(self, record_type: str, chunk: Sequence[Record]) -> bool: """单 chunk 上传。返回 True 表示成功，False 表示失败。""" if self._sender is None: raise RuntimeError("sender not set") + if record_type == "log": + console.debug( + f"Dispatching log records: count={len(chunk)}, nums={chunk[0].num}..{chunk[-1].num}", + write_to_tty=False, + ) with safe.block(message=f"record chunk upload failed, record_type={record_type!r}", write_to_tty=False): # ⚠️：在持续断网的状态下，上传失败可能会重复打印，破坏掉 transport 层的 UploadWarningThrottle self._sender.upload(record_type, chunk) + if record_type == "log": + console.debug( + f"Log sender completed: count={len(chunk)}, nums={chunk[0].num}..{chunk[-1].num}", + write_to_tty=False, + ) return True return False diff --git a/swanlab/sdk/internal/core_python/transport/sender.py b/swanlab/sdk/internal/core_python/transport/sender.py index 59574407c..c3e6ca752 100644 --- a/swanlab/sdk/internal/core_python/transport/sender.py +++ b/swanlab/sdk/internal/core_python/transport/sender.py @@ -244,6 +244,7 @@ def upload_media(self, records: Sequence[Record]) -> None: def upload_log(self, records: Sequence[Record]) -> None: with safe.block(message="Failed to upload terminal logs, skipping"): + console.debug(f"Preparing log HTTP request: records={len(records)}", write_to_tty=False) metrics: UploadLogBatch = [] for record in records: if not record.HasField("log"): @@ -258,7 +259,9 @@ def upload_log(self, records: Sequence[Record]) -> None: "create_time": create_time, } metrics.append(metric) + console.debug(f"Sending log HTTP request: metrics={len(metrics)}", write_to_tty=False) upload_log(self._project_id, self._experiment_id, metrics=metrics) + console.debug(f"Log HTTP request completed: metrics={len(metrics)}", write_to_tty=False) # ── 文件保存上传 ── diff --git a/swanlab/sdk/internal/pkg/client/session.py b/swanlab/sdk/internal/pkg/client/session.py index e297df5c9..52f3c813e 100644 --- a/swanlab/sdk/internal/pkg/client/session.py +++ b/swanlab/sdk/internal/pkg/client/session.py @@ -60,8 +60,8 @@ def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) def send(self, request, *args, **kwargs): - if self.timeout is not None: - kwargs.setdefault("timeout", self.timeout) + if self.timeout is not None and kwargs.get("timeout") is None: + kwargs["timeout"] = self.timeout # 2. 直接从上下文中读取 retries，无需触碰 request.headers retries = request_retries_ctx.get() diff --git a/swanlab/sdk/internal/run/components/consumer/__init__.py b/swanlab/sdk/internal/run/components/consumer/__init__.py index 93efff61d..ef2a6dad9 100644 --- a/swanlab/sdk/internal/run/components/consumer/__init__.py +++ b/swanlab/sdk/internal/run/components/consumer/__init__.py @@ -179,6 +179,7 @@ def _handle_event(self, event) -> None: self._save_batch.append(self._builder.build_config(event)) elif isinstance(event, LogEvent): self._log_batch.append(self._builder.build_log(event)) + console.debug(f"Buffered terminal log record: pending={len(self._log_batch)}", write_to_tty=False) elif isinstance(event, FileSaveEvent): self._save_batch.append(self._builder.build_save(event)) @@ -210,7 +211,9 @@ def _flush(self) -> None: on_error=lambda _: self._restore_batches(log_batch, column_batch, scalar_batch, media_batch, save_batch), ): if log_batch: + console.debug(f"Flushing log records to core: count={len(log_batch)}", write_to_tty=False) self._core.upsert_logs(log_batch) + console.debug(f"Core log call completed: count={len(log_batch)}", write_to_tty=False) self._callbacker.on_log_flush(log_batch) log_batch = [] diff --git a/swanlab/sdk/internal/run/components/terminal/__init__.py b/swanlab/sdk/internal/run/components/terminal/__init__.py index 6a0e7628b..aef83aad0 100644 --- a/swanlab/sdk/internal/run/components/terminal/__init__.py +++ b/swanlab/sdk/internal/run/components/terminal/__init__.py @@ -22,7 +22,7 @@ from swanlab.proto.swanlab.terminal.v1.log_pb2 import LogLevel from swanlab.sdk.internal.bus import EmitterProtocol, LogEvent -from swanlab.sdk.internal.pkg import safe +from swanlab.sdk.internal.pkg import console, safe from .capture import StreamCapture from .emulator import TerminalEmulator @@ -232,4 +232,6 @@ def _emit_line(self, line: str, log_level: LogLevel = LogLevel.LOG_LEVEL_INFO) - ts = Timestamp() ts.GetCurrentTime() + console.debug("Emitting terminal log event", write_to_tty=False) self._emitter.emit(LogEvent(line=line, level=log_level, timestamp=ts)) + console.debug("Terminal log event emitted", write_to_tty=False) From 950256a539cdbe4d7be5628ae34eee9d2983845b Mon Sep 17 00:00:00 2001 From: Kang Li Date: Fri, 24 Jul 2026 16:34:01 +0800 Subject: [PATCH 3/3] Comment out unnecessary flush/fsync calls in DataStoreWriter High-frequency flush/fsync on every write causes excessive disk I/O and can block the writer thread on network file systems like NAS. Flushing is now deferred to close(), with a small risk of losing buffered data on abnormal exit. A future Go rewrite will decouple local persistence from remote upload. --- swanlab/sdk/internal/core_python/store/__init__.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/swanlab/sdk/internal/core_python/store/__init__.py b/swanlab/sdk/internal/core_python/store/__init__.py index 6e4489fbd..182faf55e 100644 --- a/swanlab/sdk/internal/core_python/store/__init__.py +++ b/swanlab/sdk/internal/core_python/store/__init__.py @@ -88,11 +88,13 @@ def write(self, data: bytes) -> None: data_used += LEVELDBLOG_DATA_LEN data_left -= LEVELDBLOG_DATA_LEN self._write_record(data[data_used:], LEVELDBLOG_LAST) - # FIX: 【会导致磁盘高频 IO】每次 write 后统一 fsync，保证落盘 + # 不在每次 write 后调用 flush/fsync：高频指标会产生大量磁盘 IO；文件位于 NAS 等网络文件系统时， + # 网络重试还可能使写入线程长时间阻塞。正常关闭时由 close() 统一 flush；代价是进程异常退出时可能 + # 丢失尚在 Python 缓冲区中的尾部数据。若需要增强耐久性，应采用按批次或定时刷新，而非逐条刷新。 + # 后面用Go重写core时会考虑将本地持久化存储和远程上传分离在不同的channel中，避免阻塞写入线程，目前暂时注释掉flush/fsync # try: # self._fp.flush() - # # 指标量在较大的情况下高频 fsync 会影响上传效率 - # # os.fsync(self._fp.fileno()) + # os.fsync(self._fp.fileno()) # except OSError: # pass # self._flush_offset = self._index