| [fix] Make the controller survive bad ZMQ messages and repair the pickle fallback (#143) ## Motivation A long-running RL job hit an occasional failure that took the whole run down: Exception in thread TransferQueueControllerProcessRequestThread: Traceback (most recent call last): ... File "transfer_queue/controller.py", line 1771, in _process_request request_msg = ZMQMessage.deserialize(serialized_msg) File "transfer_queue/utils/zmq_utils.py", line 182, in deserialize result = decode(frames) File "transfer_queue/utils/serial_utils.py", line 262, in decode result = self.decoder.decode(bufs[0]) msgspec.DecodeError: Input data was truncated _process_request had no exception handling, so this one message killed TransferQueueControllerProcessRequestThread permanently. The controller stayed alive as an actor but stopped answering every request from that point on, and the job hung. While investigating I found more defects on the same path. ## What was wrong **1. One bad message killed the request loop.** _process_request ran recv_multipart → deserialize → dispatch → send_multipart with no guard anywhere. Note that _wait_connection in the same file already logs and continues; the request loop did not. **2. Dropped requests hang the caller forever.** Clients issue requests through with_controller_socket with no RCVTIMEO and then block on await socket.recv_multipart(). Any controller path that fails to reply — an undecodable request, an unhandled request type, a handler that raises — leaves that caller waiting for the rest of the run. **3. decode() could never detect the pickle fallback marker.** It located the marker with frames[0] == _PICKLE_FALLBACK_SENTINEL. Every receiver calls recv_multipart(copy=False) and therefore holds zmq.Frame objects, and zmq.Frame defines no __eq__ against bytes, so the comparison was *always* False and the marker frame was handed to msgpack. **4. encode() almost never reached that fallback anyway.** It caught only TypeError/ValueError, but msgspec reports unrepresentable values with OverflowError (an ArithmeticError), RecursionError (a RuntimeError) and its own MsgspecError — none of which derive from ValueError. Those escaped to the caller instead of degrading to pickle, even though pickle handles all of them (arbitrary-precision ints, self-referential containers). **5. Unhandled request types replied with a stale response.** Every branch in the dispatch chain is an if/elif with no else, so an unknown request_type fell through to send_multipart([identity, *response_msg.serialize()]) still holding the *previous* iteration's response_msg, sending an unrelated requester someone else's answer. ## What this changes - _process_request is split into a thin supervisor plus _process_request_loop; anything that escapes the per-request handling restarts the loop, so the thread survives for the controller's lifetime. The ROUTER socket stays bound across a restart. - The dispatch chain moves into _handle_request(), which returns the response or None. Every request now gets **exactly one reply**: - the handler's normal response; - a new generic ZMQRequestType.REQUEST_ERROR (built by _make_error_response()) when the handler raises — the reply body carries {request_type} failed: {error}, so the caller raises immediately instead of hanging; - REQUEST_ERROR ("no handler for request_type ...") when no branch matches, replacing the stale-response replay; - REQUEST_ERROR for undecodable requests too — the ROUTER identity frame is prepended by the transport and survives payload corruption, so the reply still reaches the requester. No client change is needed: every client call site already treats an unexpected response type as RuntimeError(body["message"]). - ZMQMessage.deserialize self-checks the frame layout and raises ZMQMessageDecodeError carrying frame count and per-frame sizes. It lives in deserialize, so controller, storage, client and manager all get the diagnostics. - decode() compares buffer *contents* via _is_pickle_fallback(). Size is tested first, so the msgpack path stays copy-free. - encode() catches the errors msgspec actually raises, via a named _ENCODE_FALLBACK_ERRORS tuple documenting why none of them are ValueError. Falling back now logs at warning level — it is a performance degradation path and should be visible at default log levels. - mypy now runs over the whole package in pre-commit (pass_filenames: false); with follow_imports = "skip", passing only changed files produced false errors. ## Scope This does **not** fix the root cause of the truncation. The evidence says the ZMQ multipart frame boundaries shift, so frame 0 stops being the msgpack header — decoding an empty buffer produces exactly Input data was truncated, and empty frames are routine here since empty tensors serialize to zero-length frames. What this PR does is stop that from taking the job down, and emit the frame layout needed to confirm it. The frame sizes make the diagnosis immediate: healthy message: num_frames=3, frame_sizes=[184, 512, 512] empty leading frame: leading frame is empty; num_frames=4, frame_sizes=[0, 184, 512, 512] boundary on a tensor frame: trailing characters (byte 1); num_frames=2, frame_sizes=[512, 512] truncated header: Input data was truncated; num_frames=3, frame_sizes=[12, 512, 512] A healthy message is one small header frame followed by large buffers. A leading 0, a missing small header, or an implausibly small header all point at shifted boundaries. One caveat to note for rollout: the pickle fallback only works end to end when *both* peers run this version — an old receiver (recv_multipart(copy=False) + == marker check) cannot recognise fallback frames from a new sender, and new senders produce them in more situations than before. Upgrade controller, storage units and clients together. ## Tests TestPickleFallback in tests/test_serial_utils_on_cpu.py (13 cases): oversized ints and self-referential containers degrade instead of raising, round-trips through zmq.Frame, marker detection across bytes/bytearray/memoryview/zmq.Frame, four near-miss negatives, and an end-to-end round trip over a real ROUTER with recv_multipart(copy=False) — the transport that hid the bug. TestTransferQueueControllerBadRequests in tests/test_controller.py (3 cases): - an empty leading frame (shifted multipart boundary) gets a REQUEST_ERROR reply and the controller keeps serving; - an unhandled PUT_DATA gets a REQUEST_ERROR reply instead of replaying the previous iteration's stale response; - a GET_META body missing its keys makes the handler raise KeyError, and the requester gets a REQUEST_ERROR ("GET_META failed: KeyError ...") while the loop keeps answering subsequent requests. All three were confirmed to **fail** against the pre-fix code and pass after. Results: 115 passed across the two serialization suites, 25 passed across tests/test_controller.py (including the 3 new cases), and 76 passed across tests/e2e/. pre-commit run --all-files is green (ruff, ruff-format, mypy). ## Unrelated CI fix bundled here The MooncakeStore e2e job started failing on libcudart.so.12: cannot open shared object file — upstream mooncake-transfer-engine-non-cuda 0.3.12 ships a mooncake_master binary linked against the CUDA 12 runtime (verified by diffing the DT_NEEDED entries of the 0.3.11.post1 and 0.3.12.post1 wheels; 0.3.11.post1 has no CUDA dependency). The two workflows that install it now pin mooncake-transfer-engine-non-cuda<0.3.12 until upstream fixes the wheel. --------- Signed-off-by: 0oshowero0 <o0shower0o@outlook.com> | 1 天前 |
| [feat] support GDR in mooncake backend (#131) Add GDR support for mooncake backend Signed-off-by: xupinjie <xupinjie321@outlook.com> | 20 天前 |
| [feat] Support cross-job actor discovery via explicit namespace (#115) When multiple Ray Jobs share the same Ray cluster, Named Actors are isolated by namespace. Without an explicit namespace, a TQ Controller created by one job is invisible to workers in another job. This commit adds namespace="transfer_queue" to both: - ray.get_actor() in _init_from_existing() - TransferQueueController.options() in init() This ensures that the TQ Controller is always registered and discovered in the fixed "transfer_queue" namespace, enabling cross-job TQ sharing (e.g., a teacher server job creates TQ, and a trainer job connects to it). This change is backward-compatible: single-job usage is unaffected since the namespace is consistent between creation and discovery. Signed-off-by: huniu20 <huniumail@gmail.com> | 1 个月前 |
| [feat] support GDR in mooncake backend (#131) Add GDR support for mooncake backend Signed-off-by: xupinjie <xupinjie321@outlook.com> | 20 天前 |
| [fix] Make the controller survive bad ZMQ messages and repair the pickle fallback (#143) ## Motivation A long-running RL job hit an occasional failure that took the whole run down: Exception in thread TransferQueueControllerProcessRequestThread: Traceback (most recent call last): ... File "transfer_queue/controller.py", line 1771, in _process_request request_msg = ZMQMessage.deserialize(serialized_msg) File "transfer_queue/utils/zmq_utils.py", line 182, in deserialize result = decode(frames) File "transfer_queue/utils/serial_utils.py", line 262, in decode result = self.decoder.decode(bufs[0]) msgspec.DecodeError: Input data was truncated _process_request had no exception handling, so this one message killed TransferQueueControllerProcessRequestThread permanently. The controller stayed alive as an actor but stopped answering every request from that point on, and the job hung. While investigating I found more defects on the same path. ## What was wrong **1. One bad message killed the request loop.** _process_request ran recv_multipart → deserialize → dispatch → send_multipart with no guard anywhere. Note that _wait_connection in the same file already logs and continues; the request loop did not. **2. Dropped requests hang the caller forever.** Clients issue requests through with_controller_socket with no RCVTIMEO and then block on await socket.recv_multipart(). Any controller path that fails to reply — an undecodable request, an unhandled request type, a handler that raises — leaves that caller waiting for the rest of the run. **3. decode() could never detect the pickle fallback marker.** It located the marker with frames[0] == _PICKLE_FALLBACK_SENTINEL. Every receiver calls recv_multipart(copy=False) and therefore holds zmq.Frame objects, and zmq.Frame defines no __eq__ against bytes, so the comparison was *always* False and the marker frame was handed to msgpack. **4. encode() almost never reached that fallback anyway.** It caught only TypeError/ValueError, but msgspec reports unrepresentable values with OverflowError (an ArithmeticError), RecursionError (a RuntimeError) and its own MsgspecError — none of which derive from ValueError. Those escaped to the caller instead of degrading to pickle, even though pickle handles all of them (arbitrary-precision ints, self-referential containers). **5. Unhandled request types replied with a stale response.** Every branch in the dispatch chain is an if/elif with no else, so an unknown request_type fell through to send_multipart([identity, *response_msg.serialize()]) still holding the *previous* iteration's response_msg, sending an unrelated requester someone else's answer. ## What this changes - _process_request is split into a thin supervisor plus _process_request_loop; anything that escapes the per-request handling restarts the loop, so the thread survives for the controller's lifetime. The ROUTER socket stays bound across a restart. - The dispatch chain moves into _handle_request(), which returns the response or None. Every request now gets **exactly one reply**: - the handler's normal response; - a new generic ZMQRequestType.REQUEST_ERROR (built by _make_error_response()) when the handler raises — the reply body carries {request_type} failed: {error}, so the caller raises immediately instead of hanging; - REQUEST_ERROR ("no handler for request_type ...") when no branch matches, replacing the stale-response replay; - REQUEST_ERROR for undecodable requests too — the ROUTER identity frame is prepended by the transport and survives payload corruption, so the reply still reaches the requester. No client change is needed: every client call site already treats an unexpected response type as RuntimeError(body["message"]). - ZMQMessage.deserialize self-checks the frame layout and raises ZMQMessageDecodeError carrying frame count and per-frame sizes. It lives in deserialize, so controller, storage, client and manager all get the diagnostics. - decode() compares buffer *contents* via _is_pickle_fallback(). Size is tested first, so the msgpack path stays copy-free. - encode() catches the errors msgspec actually raises, via a named _ENCODE_FALLBACK_ERRORS tuple documenting why none of them are ValueError. Falling back now logs at warning level — it is a performance degradation path and should be visible at default log levels. - mypy now runs over the whole package in pre-commit (pass_filenames: false); with follow_imports = "skip", passing only changed files produced false errors. ## Scope This does **not** fix the root cause of the truncation. The evidence says the ZMQ multipart frame boundaries shift, so frame 0 stops being the msgpack header — decoding an empty buffer produces exactly Input data was truncated, and empty frames are routine here since empty tensors serialize to zero-length frames. What this PR does is stop that from taking the job down, and emit the frame layout needed to confirm it. The frame sizes make the diagnosis immediate: healthy message: num_frames=3, frame_sizes=[184, 512, 512] empty leading frame: leading frame is empty; num_frames=4, frame_sizes=[0, 184, 512, 512] boundary on a tensor frame: trailing characters (byte 1); num_frames=2, frame_sizes=[512, 512] truncated header: Input data was truncated; num_frames=3, frame_sizes=[12, 512, 512] A healthy message is one small header frame followed by large buffers. A leading 0, a missing small header, or an implausibly small header all point at shifted boundaries. One caveat to note for rollout: the pickle fallback only works end to end when *both* peers run this version — an old receiver (recv_multipart(copy=False) + == marker check) cannot recognise fallback frames from a new sender, and new senders produce them in more situations than before. Upgrade controller, storage units and clients together. ## Tests TestPickleFallback in tests/test_serial_utils_on_cpu.py (13 cases): oversized ints and self-referential containers degrade instead of raising, round-trips through zmq.Frame, marker detection across bytes/bytearray/memoryview/zmq.Frame, four near-miss negatives, and an end-to-end round trip over a real ROUTER with recv_multipart(copy=False) — the transport that hid the bug. TestTransferQueueControllerBadRequests in tests/test_controller.py (3 cases): - an empty leading frame (shifted multipart boundary) gets a REQUEST_ERROR reply and the controller keeps serving; - an unhandled PUT_DATA gets a REQUEST_ERROR reply instead of replaying the previous iteration's stale response; - a GET_META body missing its keys makes the handler raise KeyError, and the requester gets a REQUEST_ERROR ("GET_META failed: KeyError ...") while the loop keeps answering subsequent requests. All three were confirmed to **fail** against the pre-fix code and pass after. Results: 115 passed across the two serialization suites, 25 passed across tests/test_controller.py (including the 3 new cases), and 76 passed across tests/e2e/. pre-commit run --all-files is green (ruff, ruff-format, mypy). ## Unrelated CI fix bundled here The MooncakeStore e2e job started failing on libcudart.so.12: cannot open shared object file — upstream mooncake-transfer-engine-non-cuda 0.3.12 ships a mooncake_master binary linked against the CUDA 12 runtime (verified by diffing the DT_NEEDED entries of the 0.3.11.post1 and 0.3.12.post1 wheels; 0.3.11.post1 has no CUDA dependency). The two workflows that install it now pin mooncake-transfer-engine-non-cuda<0.3.12 until upstream fixes the wheel. --------- Signed-off-by: 0oshowero0 <o0shower0o@outlook.com> | 1 天前 |
| [fix] Make the controller survive bad ZMQ messages and repair the pickle fallback (#143) ## Motivation A long-running RL job hit an occasional failure that took the whole run down: Exception in thread TransferQueueControllerProcessRequestThread: Traceback (most recent call last): ... File "transfer_queue/controller.py", line 1771, in _process_request request_msg = ZMQMessage.deserialize(serialized_msg) File "transfer_queue/utils/zmq_utils.py", line 182, in deserialize result = decode(frames) File "transfer_queue/utils/serial_utils.py", line 262, in decode result = self.decoder.decode(bufs[0]) msgspec.DecodeError: Input data was truncated _process_request had no exception handling, so this one message killed TransferQueueControllerProcessRequestThread permanently. The controller stayed alive as an actor but stopped answering every request from that point on, and the job hung. While investigating I found more defects on the same path. ## What was wrong **1. One bad message killed the request loop.** _process_request ran recv_multipart → deserialize → dispatch → send_multipart with no guard anywhere. Note that _wait_connection in the same file already logs and continues; the request loop did not. **2. Dropped requests hang the caller forever.** Clients issue requests through with_controller_socket with no RCVTIMEO and then block on await socket.recv_multipart(). Any controller path that fails to reply — an undecodable request, an unhandled request type, a handler that raises — leaves that caller waiting for the rest of the run. **3. decode() could never detect the pickle fallback marker.** It located the marker with frames[0] == _PICKLE_FALLBACK_SENTINEL. Every receiver calls recv_multipart(copy=False) and therefore holds zmq.Frame objects, and zmq.Frame defines no __eq__ against bytes, so the comparison was *always* False and the marker frame was handed to msgpack. **4. encode() almost never reached that fallback anyway.** It caught only TypeError/ValueError, but msgspec reports unrepresentable values with OverflowError (an ArithmeticError), RecursionError (a RuntimeError) and its own MsgspecError — none of which derive from ValueError. Those escaped to the caller instead of degrading to pickle, even though pickle handles all of them (arbitrary-precision ints, self-referential containers). **5. Unhandled request types replied with a stale response.** Every branch in the dispatch chain is an if/elif with no else, so an unknown request_type fell through to send_multipart([identity, *response_msg.serialize()]) still holding the *previous* iteration's response_msg, sending an unrelated requester someone else's answer. ## What this changes - _process_request is split into a thin supervisor plus _process_request_loop; anything that escapes the per-request handling restarts the loop, so the thread survives for the controller's lifetime. The ROUTER socket stays bound across a restart. - The dispatch chain moves into _handle_request(), which returns the response or None. Every request now gets **exactly one reply**: - the handler's normal response; - a new generic ZMQRequestType.REQUEST_ERROR (built by _make_error_response()) when the handler raises — the reply body carries {request_type} failed: {error}, so the caller raises immediately instead of hanging; - REQUEST_ERROR ("no handler for request_type ...") when no branch matches, replacing the stale-response replay; - REQUEST_ERROR for undecodable requests too — the ROUTER identity frame is prepended by the transport and survives payload corruption, so the reply still reaches the requester. No client change is needed: every client call site already treats an unexpected response type as RuntimeError(body["message"]). - ZMQMessage.deserialize self-checks the frame layout and raises ZMQMessageDecodeError carrying frame count and per-frame sizes. It lives in deserialize, so controller, storage, client and manager all get the diagnostics. - decode() compares buffer *contents* via _is_pickle_fallback(). Size is tested first, so the msgpack path stays copy-free. - encode() catches the errors msgspec actually raises, via a named _ENCODE_FALLBACK_ERRORS tuple documenting why none of them are ValueError. Falling back now logs at warning level — it is a performance degradation path and should be visible at default log levels. - mypy now runs over the whole package in pre-commit (pass_filenames: false); with follow_imports = "skip", passing only changed files produced false errors. ## Scope This does **not** fix the root cause of the truncation. The evidence says the ZMQ multipart frame boundaries shift, so frame 0 stops being the msgpack header — decoding an empty buffer produces exactly Input data was truncated, and empty frames are routine here since empty tensors serialize to zero-length frames. What this PR does is stop that from taking the job down, and emit the frame layout needed to confirm it. The frame sizes make the diagnosis immediate: healthy message: num_frames=3, frame_sizes=[184, 512, 512] empty leading frame: leading frame is empty; num_frames=4, frame_sizes=[0, 184, 512, 512] boundary on a tensor frame: trailing characters (byte 1); num_frames=2, frame_sizes=[512, 512] truncated header: Input data was truncated; num_frames=3, frame_sizes=[12, 512, 512] A healthy message is one small header frame followed by large buffers. A leading 0, a missing small header, or an implausibly small header all point at shifted boundaries. One caveat to note for rollout: the pickle fallback only works end to end when *both* peers run this version — an old receiver (recv_multipart(copy=False) + == marker check) cannot recognise fallback frames from a new sender, and new senders produce them in more situations than before. Upgrade controller, storage units and clients together. ## Tests TestPickleFallback in tests/test_serial_utils_on_cpu.py (13 cases): oversized ints and self-referential containers degrade instead of raising, round-trips through zmq.Frame, marker detection across bytes/bytearray/memoryview/zmq.Frame, four near-miss negatives, and an end-to-end round trip over a real ROUTER with recv_multipart(copy=False) — the transport that hid the bug. TestTransferQueueControllerBadRequests in tests/test_controller.py (3 cases): - an empty leading frame (shifted multipart boundary) gets a REQUEST_ERROR reply and the controller keeps serving; - an unhandled PUT_DATA gets a REQUEST_ERROR reply instead of replaying the previous iteration's stale response; - a GET_META body missing its keys makes the handler raise KeyError, and the requester gets a REQUEST_ERROR ("GET_META failed: KeyError ...") while the loop keeps answering subsequent requests. All three were confirmed to **fail** against the pre-fix code and pass after. Results: 115 passed across the two serialization suites, 25 passed across tests/test_controller.py (including the 3 new cases), and 76 passed across tests/e2e/. pre-commit run --all-files is green (ruff, ruff-format, mypy). ## Unrelated CI fix bundled here The MooncakeStore e2e job started failing on libcudart.so.12: cannot open shared object file — upstream mooncake-transfer-engine-non-cuda 0.3.12 ships a mooncake_master binary linked against the CUDA 12 runtime (verified by diffing the DT_NEEDED entries of the 0.3.11.post1 and 0.3.12.post1 wheels; 0.3.11.post1 has no CUDA dependency). The two workflows that install it now pin mooncake-transfer-engine-non-cuda<0.3.12 until upstream fixes the wheel. --------- Signed-off-by: 0oshowero0 <o0shower0o@outlook.com> | 1 天前 |
| [feat] Support cross-job actor discovery via explicit namespace (#115) When multiple Ray Jobs share the same Ray cluster, Named Actors are isolated by namespace. Without an explicit namespace, a TQ Controller created by one job is invisible to workers in another job. This commit adds namespace="transfer_queue" to both: - ray.get_actor() in _init_from_existing() - TransferQueueController.options() in init() This ensures that the TQ Controller is always registered and discovered in the fixed "transfer_queue" namespace, enabling cross-job TQ sharing (e.g., a teacher server job creates TQ, and a trainer job connects to it). This change is backward-compatible: single-job usage is unaffected since the namespace is consistent between creation and discovery. Signed-off-by: huniu20 <huniumail@gmail.com> | 1 个月前 |
| [Perf] Refactor performance test for different kv store backends (#52) # Description 1. support different kv backends 2. support intra-node and inter-node client placement for yr 3. output to csv 4. remove the non-tensor part when create_complex_test_case 5. remove ray bandwidth test 6. add readme for perf test 7. test 3 times to mitigate variance (warmup) 8. use kv client to simplify usage # Usage bash usage: perftest.py [-h] --backend_config BACKEND_CONFIG [--backend BACKEND] [--device {cpu,npu,gpu}] [--global_batch_size GLOBAL_BATCH_SIZE] [--field_num FIELD_NUM] [--seq_len SEQ_LEN] [--num_test_iterations NUM_TEST_ITERATIONS] --head_node_ip HEAD_NODE_IP [--worker_node_ip WORKER_NODE_IP] [--output_csv OUTPUT_CSV] [--use_complex_case] TransferQueue Throughput Test options: -h, --help show this help message and exit --backend_config BACKEND_CONFIG Path to backend config YAML file --backend BACKEND Override storage_backend in config (e.g. SimpleStorage, Yuanrong, MooncakeStore) --device {cpu,npu,gpu} Device to use (default: cpu) --global_batch_size GLOBAL_BATCH_SIZE Global batch size (default: 1024) --field_num FIELD_NUM Number of fields (default: 10) --seq_len SEQ_LEN Sequence length (default: 8192) --num_test_iterations NUM_TEST_ITERATIONS Number of test iterations (default: 4) --head_node_ip HEAD_NODE_IP Head node IP address --worker_node_ip WORKER_NODE_IP Worker node IP address (required for Yuanrong) --output_csv OUTPUT_CSV Path to output CSV file (optional) --use_complex_case Use complex test case with nested tensors and nontensor fields (default: False, simple case) closes #51 --------- Signed-off-by: tianyi-ge <tianyig@outlook.com> Signed-off-by: 0oshowero0 <o0shower0o@outlook.com> Co-authored-by: 0oshowero0 <o0shower0o@outlook.com> | 4 个月前 |
| [fix] Make the controller survive bad ZMQ messages and repair the pickle fallback (#143) ## Motivation A long-running RL job hit an occasional failure that took the whole run down: Exception in thread TransferQueueControllerProcessRequestThread: Traceback (most recent call last): ... File "transfer_queue/controller.py", line 1771, in _process_request request_msg = ZMQMessage.deserialize(serialized_msg) File "transfer_queue/utils/zmq_utils.py", line 182, in deserialize result = decode(frames) File "transfer_queue/utils/serial_utils.py", line 262, in decode result = self.decoder.decode(bufs[0]) msgspec.DecodeError: Input data was truncated _process_request had no exception handling, so this one message killed TransferQueueControllerProcessRequestThread permanently. The controller stayed alive as an actor but stopped answering every request from that point on, and the job hung. While investigating I found more defects on the same path. ## What was wrong **1. One bad message killed the request loop.** _process_request ran recv_multipart → deserialize → dispatch → send_multipart with no guard anywhere. Note that _wait_connection in the same file already logs and continues; the request loop did not. **2. Dropped requests hang the caller forever.** Clients issue requests through with_controller_socket with no RCVTIMEO and then block on await socket.recv_multipart(). Any controller path that fails to reply — an undecodable request, an unhandled request type, a handler that raises — leaves that caller waiting for the rest of the run. **3. decode() could never detect the pickle fallback marker.** It located the marker with frames[0] == _PICKLE_FALLBACK_SENTINEL. Every receiver calls recv_multipart(copy=False) and therefore holds zmq.Frame objects, and zmq.Frame defines no __eq__ against bytes, so the comparison was *always* False and the marker frame was handed to msgpack. **4. encode() almost never reached that fallback anyway.** It caught only TypeError/ValueError, but msgspec reports unrepresentable values with OverflowError (an ArithmeticError), RecursionError (a RuntimeError) and its own MsgspecError — none of which derive from ValueError. Those escaped to the caller instead of degrading to pickle, even though pickle handles all of them (arbitrary-precision ints, self-referential containers). **5. Unhandled request types replied with a stale response.** Every branch in the dispatch chain is an if/elif with no else, so an unknown request_type fell through to send_multipart([identity, *response_msg.serialize()]) still holding the *previous* iteration's response_msg, sending an unrelated requester someone else's answer. ## What this changes - _process_request is split into a thin supervisor plus _process_request_loop; anything that escapes the per-request handling restarts the loop, so the thread survives for the controller's lifetime. The ROUTER socket stays bound across a restart. - The dispatch chain moves into _handle_request(), which returns the response or None. Every request now gets **exactly one reply**: - the handler's normal response; - a new generic ZMQRequestType.REQUEST_ERROR (built by _make_error_response()) when the handler raises — the reply body carries {request_type} failed: {error}, so the caller raises immediately instead of hanging; - REQUEST_ERROR ("no handler for request_type ...") when no branch matches, replacing the stale-response replay; - REQUEST_ERROR for undecodable requests too — the ROUTER identity frame is prepended by the transport and survives payload corruption, so the reply still reaches the requester. No client change is needed: every client call site already treats an unexpected response type as RuntimeError(body["message"]). - ZMQMessage.deserialize self-checks the frame layout and raises ZMQMessageDecodeError carrying frame count and per-frame sizes. It lives in deserialize, so controller, storage, client and manager all get the diagnostics. - decode() compares buffer *contents* via _is_pickle_fallback(). Size is tested first, so the msgpack path stays copy-free. - encode() catches the errors msgspec actually raises, via a named _ENCODE_FALLBACK_ERRORS tuple documenting why none of them are ValueError. Falling back now logs at warning level — it is a performance degradation path and should be visible at default log levels. - mypy now runs over the whole package in pre-commit (pass_filenames: false); with follow_imports = "skip", passing only changed files produced false errors. ## Scope This does **not** fix the root cause of the truncation. The evidence says the ZMQ multipart frame boundaries shift, so frame 0 stops being the msgpack header — decoding an empty buffer produces exactly Input data was truncated, and empty frames are routine here since empty tensors serialize to zero-length frames. What this PR does is stop that from taking the job down, and emit the frame layout needed to confirm it. The frame sizes make the diagnosis immediate: healthy message: num_frames=3, frame_sizes=[184, 512, 512] empty leading frame: leading frame is empty; num_frames=4, frame_sizes=[0, 184, 512, 512] boundary on a tensor frame: trailing characters (byte 1); num_frames=2, frame_sizes=[512, 512] truncated header: Input data was truncated; num_frames=3, frame_sizes=[12, 512, 512] A healthy message is one small header frame followed by large buffers. A leading 0, a missing small header, or an implausibly small header all point at shifted boundaries. One caveat to note for rollout: the pickle fallback only works end to end when *both* peers run this version — an old receiver (recv_multipart(copy=False) + == marker check) cannot recognise fallback frames from a new sender, and new senders produce them in more situations than before. Upgrade controller, storage units and clients together. ## Tests TestPickleFallback in tests/test_serial_utils_on_cpu.py (13 cases): oversized ints and self-referential containers degrade instead of raising, round-trips through zmq.Frame, marker detection across bytes/bytearray/memoryview/zmq.Frame, four near-miss negatives, and an end-to-end round trip over a real ROUTER with recv_multipart(copy=False) — the transport that hid the bug. TestTransferQueueControllerBadRequests in tests/test_controller.py (3 cases): - an empty leading frame (shifted multipart boundary) gets a REQUEST_ERROR reply and the controller keeps serving; - an unhandled PUT_DATA gets a REQUEST_ERROR reply instead of replaying the previous iteration's stale response; - a GET_META body missing its keys makes the handler raise KeyError, and the requester gets a REQUEST_ERROR ("GET_META failed: KeyError ...") while the loop keeps answering subsequent requests. All three were confirmed to **fail** against the pre-fix code and pass after. Results: 115 passed across the two serialization suites, 25 passed across tests/test_controller.py (including the 3 new cases), and 76 passed across tests/e2e/. pre-commit run --all-files is green (ruff, ruff-format, mypy). ## Unrelated CI fix bundled here The MooncakeStore e2e job started failing on libcudart.so.12: cannot open shared object file — upstream mooncake-transfer-engine-non-cuda 0.3.12 ships a mooncake_master binary linked against the CUDA 12 runtime (verified by diffing the DT_NEEDED entries of the 0.3.11.post1 and 0.3.12.post1 wheels; 0.3.11.post1 has no CUDA dependency). The two workflows that install it now pin mooncake-transfer-engine-non-cuda<0.3.12 until upstream fixes the wheel. --------- Signed-off-by: 0oshowero0 <o0shower0o@outlook.com> | 1 天前 |
| initialize TransferQueue repository Co-authored-by: 0oshowero0<o0shower0o@outlook.com> # message auto-generated for no-merge-commit merge: !1 merge main into main [chore] initialize TransferQueue repository Created-by: hanzhenyu8 Commit-by: 0oshowero0 Merged-by: ascend-robot Description: ## Background This PR marks the **initial commit** of the TransferQueue repository. We would like to express our sincere gratitude to the community for their invaluable contributions during the **incubation** phase of TransferQueue. - [@NINGBENZHE](https://github.com/NINGBENZHE): [[Feat]: add check_data_production_status and check_consumption_status and support Polling get metadata](https://github.com/TransferQueue/TransferQueue/pull/157). - [@zhaohaidao](https://github.com/zhaohaidao): [[Feat] Support Mooncake Store backend](https://github.com/TransferQueue/TransferQueue/pull/162). ### Historical Context To preserve the project's heritage, the early development history remains accessible at: https://github.com/TransferQueue/TransferQueue. Moving forward, we will maintain a mirror repository under the [Ascend organization](https://github.com/Ascend) on GitHub. You are welcome to submit contributions or propose new ideas on either platform. **<span style="color:#e60000;">We look forward to continuing this journey with all of you!</span>** See merge request: Ascend/TransferQueue!1 | 6 个月前 |
| [chore] Update README (#128) As title Signed-off-by: 0oshowero0 <o0shower0o@outlook.com> | 1 个月前 |
| [chore] Bump version from 0.1.9.dev0 to 0.1.9 & update dependency (#139) As title Signed-off-by: 0oshowero0 <o0shower0o@outlook.com> | 20 天前 |
| [chore] Relax numpy version constraints (#113) As title Signed-off-by: 0oshowero0 <o0shower0o@outlook.com> | 1 个月前 |
