[fix] Retry a lost storage-unit request instead of failing the job - #171
Conversation
CLA Signature Guide@adoda , thanks for your pull request. The following commit(s) are not associated with a signed Contributor License Agreement (CLA).
To sign CLA, click here. To check if your email is configured correctly, refer to the FAQs. Once you've signed the CLA or updating your email, please comment |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: bfa75f2ee9
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| self._put_to_single_storage_unit, | ||
| group.global_indexes, | ||
| storage_data, | ||
| target_storage_unit=su_id, | ||
| data_parser=data_parser, |
There was a problem hiding this comment.
When a PUT is committed by the storage unit but its reply is lost, this retry path sends data_parser again with a new request ID, and _handle_put executes the callable a second time. The supported public APIs currently accept arbitrary parsers without requiring idempotence, so parsers that consume references, perform external writes, or generate nondeterministic values can duplicate side effects or silently overwrite the first result. Deduplicate PUTs at the unit or avoid replaying parser-backed PUTs unless this constraint is enforced at the public API.
AGENTS.md reference: AGENTS.md:L17-L18
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Confirmed and fixed in 613bba8. kv_put / kv_batch_put / put document a contract for data_parser covering keys, element count and ordering, but never side effects, so treating replay-safety as an unwritten requirement was wrong.
Rather than dedupe at the unit, which needs per-request state there, a parser-backed put is now sent once: put_data passes max_attempts=1 when data_parser is not None, so it fails exactly as it did before this change. The retry stays in force for the ordinary put, which is a plain overwrite of the same global indexes.
Tests: test_parser_backed_put_is_not_replayed asserts one attempt, test_put_without_a_parser_is_still_retried asserts the retry is not lost for the ordinary path.
| endpoint = self._describe_storage_unit(target_storage_unit) | ||
| for attempt in range(1, TQ_SIMPLE_STORAGE_MAX_ATTEMPTS + 1): | ||
| try: | ||
| return await make_request() |
There was a problem hiding this comment.
If TQ_SIMPLE_STORAGE_MAX_ATTEMPTS is configured as 0 or a negative value, this loop never runs and _request_with_retry returns None. For PUTs, asyncio.gather therefore reports success and put_data notifies the controller even though no payload was sent, leaving published metadata that points to missing data. Validate the new configuration as at least one attempt before serving requests.
AGENTS.md reference: AGENTS.md:L17-L18
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Confirmed and fixed in 177667f. Reproduced first: with the count patched to 0, _request_with_retry returned None without raising, and for a put that reads as success, after which put_data would notify the controller with metadata for data that was never sent.
The attempt count is now floored at one where it is consumed, attempts_allowed = max(1, ...), so every caller gets one attempt regardless of how the value was configured. I put the floor at the use site rather than at import so it also covers the internal max_attempts override, and so it is testable without reloading the module.
Test: test_a_nonpositive_attempt_count_still_issues_one_request.
A storage unit stops answering and every client routed to it fails when its own recv timeout expires. Raising that timeout (400s to 1800s in our deployment) changed nothing. Evidence from one such failure: the unit's node was alive and serving other traffic throughout, a TCP connect to its put_get_socket succeeded, and the unit's own counters showed it fully healthy (1025 GET_DATA served, 9.6ms p99, 1.33GB RSS). It had served exactly one GET_DATA fewer than its cohort. So the unit never saw the request that timed out; it was lost between the two ends, not queued behind slow work. Ascend#168 protected the worker thread from dying, which is a different cause of the same symptom; here the thread was intact. That loss is invisible by construction. ZMQ connect is asynchronous and SNDHWM is 0, so a DEALER accepts send() into an unbounded local queue for a peer it has not reached yet. The message sits there and the caller only learns anything when its own RCVTIMEO expires, which is why no timeout value can distinguish a lost request from a slow one. Lowering SNDHWM would not help either, it only trades silent queuing for silent dropping. Retry the request on a new socket and TCP connection, which is the part that matters, up to TQ_SIMPLE_STORAGE_MAX_ATTEMPTS (default 3). Only a missing answer is retried: zmq.error.Again now raises StorageUnitTimeout, while an error the unit actually reported still surfaces on the first attempt. Replaying an attempt is safe: put is keyed by global index and overwrites, get is read-only. Make the residual failure self-diagnosing, so a next occurrence does not need another round of manual probing. After the last attempt, ask the unit for its own counters over a fresh socket with a short timeout. That probe is served by the same worker thread as put and get, so an answer proves the unit is serving and the request was lost in flight, while silence means the unit itself stopped. The failure log now carries that verdict plus tcp reachability, the unit's op counts and RSS, and the shape of the request that failed. Log volume is unchanged in the steady state. A recovered request logs one line and skips the diagnosis entirely; storage units log a request only above TQ_STORAGE_SLOW_REQUEST_SECONDS (5s) or TQ_STORAGE_LARGE_PAYLOAD_MB (256MB), both far above the single-digit-millisecond norm, so tripping either one is itself the finding. The put_data failure log no longer dumps every routed unit id, which on a large job was thousands of them per line, matching what get_data already does. Tests cover recovery on retry, the bounded attempt count, that reported errors are not retried, and each diagnosis verdict. Signed-off-by: jathonzhang <jathonzhang@tencent.com>
Codex review on Ascend#171: with TQ_SIMPLE_STORAGE_MAX_ATTEMPTS set to 0 or a negative value the retry loop never ran, so _request_with_retry returned None without raising. For a put that reads as success, and put_data then notified the controller, publishing metadata that points at data which was never sent. Floor the attempt count where it is consumed, so every caller gets one attempt regardless of how the value was configured. Signed-off-by: jathonzhang <jathonzhang@tencent.com>
|
Codex review on Ascend#171: a put whose reply was lost has already been committed by the unit, so a retry re-runs data_parser there a second time. Replaying the write itself is harmless, it overwrites the same global indexes, but the parser is not: kv_put, kv_batch_put and put accept an arbitrary callable and constrain only its keys, element count and ordering, never its side effects. A parser that consumes references or writes externally would see those effects duplicated. Send a parser-backed put once, so it fails exactly as it did before this series, and keep the retry for the ordinary put, which is a plain overwrite. Signed-off-by: jathonzhang <jathonzhang@tencent.com>
Codex review on Ascend#171: with TQ_SIMPLE_STORAGE_MAX_ATTEMPTS set to 0 or a negative value the retry loop never ran, so _request_with_retry returned None without raising. For a put that reads as success, and put_data then notified the controller, publishing metadata that points at data which was never sent. Floor the attempt count where it is consumed, so every caller gets one attempt regardless of how the value was configured. Signed-off-by: jathonzhang <jathonzhang@tencent.com> Co-authored-by: Cursor <cursoragent@cursor.com>
Codex review on Ascend#171: a put whose reply was lost has already been committed by the unit, so a retry re-runs data_parser there a second time. Replaying the write itself is harmless, it overwrites the same global indexes, but the parser is not: kv_put, kv_batch_put and put accept an arbitrary callable and constrain only its keys, element count and ordering, never its side effects. A parser that consumes references or writes externally would see those effects duplicated. Send a parser-backed put once, so it fails exactly as it did before this series, and keep the retry for the ordinary put, which is a plain overwrite. Signed-off-by: jathonzhang <jathonzhang@tencent.com> Co-authored-by: Cursor <cursoragent@cursor.com>
bfa75f2 to
613bba8
Compare
CLA Signature Guide@adoda , thanks for your pull request. The following commit(s) are not associated with a signed Contributor License Agreement (CLA).
To sign CLA, click here. To check if your email is configured correctly, refer to the FAQs. Once you've signed the CLA or updating your email, please comment |
|
/check-cla |
CLA Signature Guide@adoda , thanks for your pull request. The following commit(s) are not associated with a signed Contributor License Agreement (CLA).
To sign CLA, click here. To check if your email is configured correctly, refer to the FAQs. Once you've signed the CLA or updating your email, please comment |
|
/check-cla |
CLA Signature Passadoda, thanks for your pull request. All authors of the commits have signed the CLA. 👍 |
|
/check-cla |
613bba8 to
28d7a13
Compare
CLA Signature Passadoda, thanks for your pull request. All authors of the commits have signed the CLA. 👍 |
1 similar comment
CLA Signature Passadoda, thanks for your pull request. All authors of the commits have signed the CLA. 👍 |
| {f: self._select_by_positions(data[f], group.batch_positions) for f in data.keys()}, | ||
| target_storage_unit=su_id, | ||
| data_parser=data_parser, | ||
| # Replaying a put is safe because it overwrites the same global indexes, but a retry also |
There was a problem hiding this comment.
We can remove some comments
There was a problem hiding this comment.
Done in 44b7573: shortened to one line that states the constraint the max_attempts=1 branch is enforcing.
| return env_value_lower in true_values | ||
|
|
||
|
|
||
| def estimate_payload_bytes(field_data: Any) -> int: |
There was a problem hiding this comment.
This estimation might be highly inaccurate. Maybe a better solution is let the underlying _put_to_single_storage_unit to return the payload size and throw an error that will be captured by the retry layer?
For instance, to get the accurate size, we can
frames = request_msg.serialize()
serialized_bytes = sum(memoryview(frame).nbytes for frame in frames)
await socket.send_multipart(frames, copy=False)And raise when meeting errors to let retry layer know:
except zmq.error.Again as exc:
serialized_bytes = serialized_nbytes(frames)
raise StorageUnitTimeout(
f"put timeout: unit={target_storage_unit}, "
f"samples={len(global_indexes)}, "
f"serialized_bytes={serialized_bytes}"
) from excThere was a problem hiding this comment.
Agreed and fixed in 2bcbcf8. estimate_payload_bytes only walked tensor nbytes, so NonTensorStack / object fields, msgpack/pickle overhead, and multipart framing were all invisible, and a data_parser that expands references was measured before expansion.
_put_to_single_storage_unit now measures the frames after serialize() with frame_nbytes, puts serialized_mb on StorageUnitTimeout, and the retry layer logs the exception text so the wire size shows up on both intermediate retries and the final failure. The call-site request_context no longer estimates payload size.
There was a problem hiding this comment.
We can also refactor the get side accordingly, and totally remove this function
There was a problem hiding this comment.
Done in the same commit. Get now measures serialized response frames the same way put measures request frames, so estimate_payload_bytes has no callers left and is removed.
| poller.unregister(worker_socket) | ||
| worker_socket.close(linger=0) | ||
|
|
||
| def _log_if_heavy(self, operation: str, started: float, field_data: Any, num_samples: int, fields: Any) -> None: |
There was a problem hiding this comment.
Now the log only happens at simple storage side. Maybe a better choice is log for put in manager side, and log for get in simple storage side?
There was a problem hiding this comment.
Done in 91a9e46. Put heavy logging moved to the manager, using the same serialized frame size and the end-to-end RTT around send/recv. Get heavy logging stays on the unit, where the response payload is built. The unit-side PUT _log_if_heavy call is gone.
Signed-off-by: jathonzhang <jathonzhang@tencent.com>
estimate_payload_bytes only walked tensor nbytes, so NonTensorStack fields, msgpack/pickle overhead, and multipart framing were all invisible. Measure the frames after serialize in _put_to_single_storage_unit and put that size on StorageUnitTimeout; the retry layer now logs the exception text so the true wire size shows up on both intermediate retries and the final failure. Signed-off-by: jathonzhang <jathonzhang@tencent.com>
The manager sees the wire size and end-to-end RTT of a put; the unit only sees local processing after deserialize. Move put heavy logging to the manager using the serialized frame size already measured for timeouts, and keep get heavy logging on the unit where the response payload is built. Signed-off-by: jathonzhang <jathonzhang@tencent.com>
CLA Signature Passadoda, thanks for your pull request. All authors of the commits have signed the CLA. 👍 |
| field_schema, | ||
| ) | ||
|
|
||
| def _log_if_heavy_put( |
There was a problem hiding this comment.
We better extract this as a util function log_heavy_operation, passing id, criteria, warning message into it. Therefore we can reuse it for SimpleStorage side
There was a problem hiding this comment.
Done in f3d31af. Extracted log_heavy_operation(component_id, operation, elapsed, payload_bytes, detail) into transfer_queue/utils/common.py next to the shared thresholds, and both the manager put path and the unit get path call it.
Both ends already serialize the payload they care about, so report that size instead of estimating tensor nbytes. Fold the duplicated threshold check into log_heavy_operation and drop estimate_payload_bytes. Signed-off-by: jathonzhang <jathonzhang@tencent.com>
CLA Signature Passadoda, thanks for your pull request. All authors of the commits have signed the CLA. 👍 |
The timeout exception is what put_data and get_data callers see, so it names the unit, its endpoint and the timeout; the retry logs now add only the attempt and the request shape. Both operations report the same fields. Signed-off-by: jathonzhang <jathonzhang@tencent.com>
CLA Signature Passadoda, thanks for your pull request. All authors of the commits have signed the CLA. 👍 |
|
Pushed 7f28a13. The retry logs repeated what the exception already said: the unit id appeared three times in one line, the endpoint, Verified: |
Keep storage RPC pooling from Ascend#167 while preserving fresh-connection retry: a failed lease is discarded, and the diagnostic probe uses its own short-timeout pool. Compact the retry tests without dropping behavior coverage. Signed-off-by: jathonzhang <jathonzhang@tencent.com>
CLA Signature Passadoda, thanks for your pull request. All authors of the commits have signed the CLA. 👍 |
Builds on the retry in Ascend#171, which established that a request is lost between the two ends rather than queued behind slow work. Two things that retry does not cover. Raise the client-facing ROUTER's accept queue from ZMQ's default of 100, which was just the unset default. A full accept queue is emptied without sending an RST (tcp_abort_on_overflow=0), leaving the client established and waiting for a reply nobody will send, while the server never learns the request existed. The observed hung unit's listen socket had already charged 23 drops, and machine-wide ListenOverflows equalled ListenDrops, which is the signature of exactly that. It is a second silent-loss path alongside the unbounded DEALER queue. Tunable via TQ_STORAGE_ZMQ_BACKLOG; set it to 100 to restore the old value for an A/B run. Count requests as the worker decodes them, and sample the accept queue when asked. The per-op counters advance inside monitor.measure(), so they only move once a request completes and a request that arrived but never finished reads exactly like one that never arrived; the diagnostic probe cannot tell those apart either. An arrival count next to the completion histograms does, and a shortfall against the caller's send count localizes the loss to one side. The queue-depth probe is opt-in via TQ_ACCEPT_PROBE_INTERVAL and off by default: it shells out to ss on a timer, and depth has to be sampled sub-second because the queue drains in milliseconds, which is why inspecting a unit after it hung always read zero. Tests cover the probe's peak and delta arithmetic and the levels it logs at. Signed-off-by: OutstanderWang <wangweiyanster@gmail.com>
## Summary A storage-unit request can be lost between the manager and the unit without either end raising an error. #171 recovers such a request by retrying it; this PR closes one of the paths it can be lost on, and adds the instrumentation needed to tell the loss apart from a slow unit. It also corrects two defects in the diagnosis that #171 introduced — both are on `main` today. ## What's here ### 1. Raise the client-facing ROUTER's accept queue ZMQ defaults `BACKLOG` to 100. That was simply the unset default: a full accept queue is drained **without an RST** (`tcp_abort_on_overflow=0`), so the client stays in `ESTABLISHED` waiting for a reply nobody will ever send, and the server never learns the request existed. Neither end raises. Default is now 4096, tunable via `TQ_STORAGE_ZMQ_BACKLOG` (set it to 100 to restore ZMQ's default for an A/B run). This only raises a queue ceiling — it cannot reject a connection that previously succeeded. ### 2. Make the loss measurable rather than inferred - **Arrival counters** (`requests_arrived`, `arrivals_by_op` in `get_metrics`). The per-op counters advance inside `monitor.measure()`, so they only move once a request *completes* — a request that arrived and then stalled is indistinguishable from one that never arrived. Counting at decode time separates the two, and a shortfall against the caller's send count localizes the loss to one side of the connection. - **Accept-queue probe** (`transfer_queue/utils/accept_probe.py`). Samples `Recv-Q` / `sk_drops` / `ListenOverflows` from the kernel's own view. **Opt-in and off by default** (`TQ_ACCEPT_PROBE_INTERVAL=0`): it shells out to `ss` on a timer, and depth must be sampled sub-second because the queue drains in milliseconds — which is why inspecting a unit *after* it hung always read zero. ### 3. Report an accept-queue drop once per drop `sk_drops` is a cumulative kernel counter, but the initial check compared it against the probe's *first* sample. Once a socket had ever dropped a connection the condition stayed true for the life of the process — at a 0.1 s interval that is ten errors per second per storage unit. The flood also destroyed the signal: repeating the same cumulative total says nothing about *when* drops happened. Now compares against the previous sample and reports both deltas. ### 4. Base the timeout diagnosis on evidence (fixes two defects on `main`) `_diagnose_storage_unit` reported `verdict=request_lost_in_flight` **unconditionally** whenever the post-failure probe answered. But a unit that resumed inside the 10 s diagnostic window answers that probe too, having merely finished the original request late — so the verdict asserted *where* the request was lost on evidence that only shows the unit is serving *now*. It also printed `ops=` from `op_stats`, which `_handle_get_metrics` populates only when Prometheus is enabled, so the line read `ops={}` as if the unit had served nothing. The verdict now follows the unit's own arrival counter for the failed operation: | Probe answered, and… | Verdict | |---|---| | no arrival recorded for that op | `request_lost_in_flight` | | an arrival recorded for that op | `arrived_but_unfinished` | | counters unavailable (e.g. older unit) | `unit_serving_again` | `op_stats` is now reported as `unavailable(prometheus_disabled)` rather than as an empty dict. **The counters could not actually be correlated as intended.** The worker keyed `arrivals_by_op` by `str(operation)`, which for a `(str, Enum)` member renders as `"ZMQRequestType.GET_DATA"`, while `op_stats` is keyed `"GET_DATA"` — the two dicts shared no keys at all. Fixed by keying on `operation.name`. Note `.value` would *also* miss: the enum values are short wire tokens (`GET_DATA = "GET"`). ### 5. Stop the accept probe on shutdown `AcceptQueueProbe.start()` was called but `stop()` had no caller. The finalizer tore down only the ZMQ resources, so a unit finalized without its process exiting left the daemon thread spawning `ss` on a timer indefinitely, and the window summary was never logged. Now stopped in `_shutdown_resources`, before the ZMQ teardown. ## Behavior and compatibility - **No public API change.** `transfer_queue/utils/zmq_utils.py` is untouched. - **One default changes:** ROUTER `BACKLOG` 100 → 4096. - **The probe is off by default** and imported lazily, so a run that does not enable it never loads the module or spawns the thread. - **Steady-state log volume is unchanged**; the near-full-queue warning fires once per probe, not per sample. ## New configuration | Variable | Default | Purpose | |---|---|---| | `TQ_STORAGE_ZMQ_BACKLOG` | `4096` | Accept-queue depth for the client-facing ROUTER | | `TQ_ACCEPT_PROBE_INTERVAL` | `0` (off) | Accept-queue sampling period, seconds; sub-second when enabled | ## Note on the rebase This branch previously carried #171's commits as its base and included a per-call socket-timeout change. Since then #171 merged and #167 replaced per-call sockets with `ZMQSocketPool`, which resolves timeouts from the pool instead of the decorator. That change therefore had nowhere left to live and has been dropped — `main` already routes the two timeouts through separate `storage_rpc_pool` / `storage_probe_pool` instances, which satisfies the same intent. Its test file went with it; pool timeout behaviour is covered by `tests/test_zmq_socket_pool.py::test_timed_out_socket_is_not_reused`. The branch is now four commits on top of `ba8880c`, all mine. ## Testing - `tests/test_accept_probe.py` — utilization and delta arithmetic, peak tracking across a drained burst, drop reported once per drop, each new drop still reported, warn-once behaviour, zero-backlog not dividing by zero, and the probe being stopped on shutdown. - `tests/test_storage_request_retry.py` — the diagnosis table above, plus the empty-`op_stats` case and the arrival-key spelling. - Full suite: **644 passed, 10 skipped**, against a **623**-passing baseline measured on `ba8880c` itself — the +21 are the new tests, no regressions. - `check_license.py`, `check_docstrings.py`, `ruff check` and `ruff format --check` all clean. `tests/test_yuanrong_storage_client_e2e.py` is excluded from those runs: it errors identically on the unmodified base (optional Yuanrong backend not installed locally). --------- Signed-off-by: OutstanderWang <wangweiyanster@gmail.com>
3 participants
Problem
A storage unit stops answering and every client routed to it fails when its own recv timeout
expires. Raising the timeout (400s to 1800s in our deployment) changed nothing.
Evidence from one such failure: the unit's node was alive and serving other traffic throughout, a
TCP connect to its
put_get_socketsucceeded, and the unit's own counters showed it fully healthy(1025 GET_DATA served, 9.6ms p99, 1.33GB RSS). It had served exactly one GET_DATA fewer than its
cohort. So the unit never saw the request that timed out; it was lost between the two ends, not
queued behind slow work. #168 protected the worker thread from dying, which is a different cause of
the same symptom; here the thread was intact.
That loss is invisible by construction. ZMQ connect is asynchronous and SNDHWM is 0, so a DEALER
accepts
send()into an unbounded local queue for a peer it has not reached yet. The caller onlylearns anything when its own RCVTIMEO expires, which is why no timeout value can distinguish a lost
request from a slow one. Lowering SNDHWM would not help either, it only trades silent queuing for
silent dropping.
Fix
TQ_SIMPLE_STORAGE_MAX_ATTEMPTS(default 3, floored at 1).A fresh socket and TCP connection is the part that matters. Only a missing answer is retried:
zmq.error.Againnow raisesStorageUnitTimeout, while an error the unit reported surfaces onthe first attempt. Replaying is safe where it is used: get is read-only and put overwrites the
same global indexes. A put carrying a
data_parseris the exception and is sent once, because aretry would re-run the parser on the unit and the public API does not constrain its side
effects.
over a fresh socket with a short timeout. That probe is served by the same worker thread as put
and get, so an answer proves the request was lost in flight and silence means the unit itself
stopped. The failure log carries that verdict plus tcp reachability, the unit's op counts and RSS,
and the wire size of the request that failed (measured from the serialized ZMQ frames on put).
diagnosis entirely. Heavy requests are logged only above
TQ_STORAGE_SLOW_REQUEST_SECONDS(5s)or
TQ_STORAGE_LARGE_PAYLOAD_MB(256MB), via sharedlog_heavy_operation: puts on the manager(serialized request frames + end-to-end RTT), gets on the unit (serialized response frames +
local handling time). No tensor-
nbytesestimate remains.Tests
tests/test_storage_request_retry.py: 14 passed. Covers recovery on retry, the bounded attemptcount, that a nonpositive count still issues one request, that errors reported by the unit are not
retried, that a parser-backed put is not replayed while the ordinary put still is, that retry logs
carry wire-size detail from the failed attempt, shared heavy-request logging thresholds, and each
diagnosis verdict.
Full non-e2e run: 623 passed, 10 skipped, plus
python -m compileall -q transfer_queue tutorial tests.test_yuanrong_storage_client_e2e.pyis excluded locally, it needs the optionalopenyuanrong-datasystemdependency.