fix: dual zmq pub for sdr and bridge, sdr pythonpath - #17
Conversation
📝 WalkthroughWalkthroughPR 将 ZMQ PUB 从单一 ChangesZMQ 多端点发布
运行环境与视频提示
Estimated code review effort: 3 (Moderate) | ~20 minutes Sequence Diagram(s)sequenceDiagram
participant RadarApp
participant ZmqPubRuntime
participant zmq_start_pub
participant PUBSockets
RadarApp->>ZmqPubRuntime: 启动并传递两个绑定地址
ZmqPubRuntime->>PUBSockets: 创建 :5557 和 :5558 的 PUB socket
ZmqPubRuntime->>zmq_start_pub: 传递 socket 向量
zmq_start_pub->>PUBSockets: 向每个 socket 发布同步消息
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (2)
src/services/script_runner.rs (1)
244-244: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win保留继承的
PYTHONPATH。
.env("PYTHONPATH", ".:/usr/lib/python3/dist-packages")会替换父进程提供的PYTHONPATH。如果部署环境通过现有值提供其他 SDR 模块路径，SDR 启动时可能无法导入thread_init.py依赖。请在现有值基础上追加仓库根目录和/usr/lib/python3/dist-packages。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/services/script_runner.rs` at line 244, 更新 script runner 中设置 PYTHONPATH 的逻辑，保留父进程现有的 PYTHONPATH，并在其基础上追加仓库根目录和 /usr/lib/python3/dist-packages；当父进程未设置该变量时仍使用这两个路径，确保现有 SDR 模块路径不被覆盖。tests/runtime/zmq.rs (1)
276-276: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick win补充双端点 fan-out 回归测试。
Line 276 和 Line 328 都只传入
vec![pub_sock]。这些测试只能验证参数类型变化，不能验证第二个 PUB socket 是否收到同一条消息。请为IDX_GAME_STATE、IDX_RADAR_MARK_PROCESS和IDX_RADAR_AUTONOMOUS_DECISION_SYNC使用两个独立的 PUB/SUB pair，并断言两个 subscriber 都收到正确的cmd_id和 payload。Also applies to: 328-328
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/runtime/zmq.rs` at line 276, 扩展 tests/runtime/zmq.rs 中由 zmq_start_pub 启动的相关测试，为 IDX_GAME_STATE、IDX_RADAR_MARK_PROCESS 和 IDX_RADAR_AUTONOMOUS_DECISION_SYNC 分别建立两个独立的 PUB/SUB pair，并将两个 PUB socket 传入 fan-out 流程；随后分别从两个 subscriber 接收消息，断言其 cmd_id 和 payload 均正确且一致，覆盖第二个端点的实际转发行为。
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@docs/data-flow.md`:
- Line 27: 统一 docs/data-flow.md 中所有 PUB 端口描述：将 Line 53 和 Line 194 的单一
tcp://*:5557 更新为 :5557/:5558 双端点，并明确 5557 与 5558 的消费者职责；同时核对相关端口说明保持与 Lines
27、201、249 一致，准确反映当前 ZMQ/ROS2 Radar 设计。
In `@src/zmq/zmq.rs`:
- Around line 119-121: Update each publishing loop around zmq_send so individual
socket send errors are handled instead of discarded with .ok(). On Err, log the
socket’s endpoint identity and apply an explicit policy—retry, remove the failed
socket, or terminate the publishing thread—consistently across all three publish
branches; keep Value::to_string() unchanged.
- Around line 119-121: Move the blocking zmq_send calls out of the SharedData
lock scope in the payload construction flow. Read the required SharedData fields
and build the message while holding the lock, then release the lock before
iterating over pub_sockets and sending to each endpoint; preserve the existing
per-socket send behavior and error handling.
---
Nitpick comments:
In `@src/services/script_runner.rs`:
- Line 244: 更新 script runner 中设置 PYTHONPATH 的逻辑，保留父进程现有的
PYTHONPATH，并在其基础上追加仓库根目录和
/usr/lib/python3/dist-packages；当父进程未设置该变量时仍使用这两个路径，确保现有 SDR 模块路径不被覆盖。
In `@tests/runtime/zmq.rs`:
- Line 276: 扩展 tests/runtime/zmq.rs 中由 zmq_start_pub 启动的相关测试，为
IDX_GAME_STATE、IDX_RADAR_MARK_PROCESS 和 IDX_RADAR_AUTONOMOUS_DECISION_SYNC
分别建立两个独立的 PUB/SUB pair，并将两个 PUB socket 传入 fan-out 流程；随后分别从两个 subscriber 接收消息，断言其
cmd_id 和 payload 均正确且一致，覆盖第二个端点的实际转发行为。
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 49cda069-14e1-434c-af74-156f29f6615b
📒 Files selected for processing (7)
AGENTS.mddocs/data-flow.mdsrc/app/mod.rssrc/runtime/mod.rssrc/services/script_runner.rssrc/zmq/zmq.rstests/runtime/zmq.rs
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@tests/runtime/zmq.rs`:
- Around line 443-451: Update test_zmq_pub_fans_out_to_all_sockets to replace
both fixed thread::sleep waits with deterministic subscription-readiness
synchronization, using XPUB subscription events to confirm both SUB sockets are
registered before sending IDX_GAME_STATE. Preserve the existing single-send
behavior and subsequent receives from sub_a and sub_b.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: fc385792-f004-4957-bed3-3fa6a8758d46
📒 Files selected for processing (4)
docs/data-flow.mdsrc/services/script_runner.rssrc/zmq/zmq.rstests/runtime/zmq.rs
🚧 Files skipped from review as they are similar to previous changes (3)
- src/services/script_runner.rs
- src/zmq/zmq.rs
- docs/data-flow.md
| thread::sleep(Duration::from_millis(50)); | ||
|
|
||
| tx.send(IDX_GAME_STATE).unwrap(); | ||
| thread::sleep(Duration::from_millis(100)); | ||
|
|
||
| let mut buf_a = zmq2::Message::new(); | ||
| let mut buf_b = zmq2::Message::new(); | ||
| sub_a.recv(&mut buf_a, 0).unwrap(); | ||
| sub_b.recv(&mut buf_b, 0).unwrap(); |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== files =="
git ls-files | rg '(^|/)zmq\.rs$|runtime' | head -200
echo "== target file outline/size =="
wc -l tests/runtime/zmq.rs
ast-grep outline tests/runtime/zmq.rs --view compact || true
echo "== relevant section =="
sed -n '380,470p' tests/runtime/zmq.rs
echo "== search zmq pub/sub start functions =="
rg -n "zmq_start_pub|zmq_start_sub|XPUB|setsockopt|subscribe|send\\(" tests/runtime/zmq.rs src runtime 2>/dev/null || trueRepository: Alliance-Algorithm/radar-egui
Length of output: 12127
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== zmq implementation =="
sed -n '1,220p' src/zmq/zmq.rs
echo "== helper functions =="
sed -n '1,80p' tests/runtime/zmq.rs
sed -n '260,310p' tests/runtime/zmq.rs
sed -n '320,375p' tests/runtime/zmq.rs
echo "== deterministic recv semantics probe =="
python3 - <<'PY'
from pathlib import Path
p = Path('src/zmq/zmq.rs')
text = p.read_text()
checks = {
'zmq_start_pub sends once after notification': 'payloads.push(payload);\n while let Some(msg) = pub_rx.try_recv() {' in text or 'pub_rx.try_recv()' in text,
'zmq_send sends with nonblocking flag': 'pub_socket.send(msg, 0)' in text,
'sub_socket set empty subscribe': 'sub_socket.set_subscribe(b"")?' in text,
'fanout test publishes after 50ms sleep': 'thread::sleep(Duration::from_millis(50));' in Path('tests/runtime/zmq.rs').read_text(),
'fanout test publishes once': 'tx.send(IDX_GAME_STATE).unwrap();' in Path('tests/runtime/zmq.rs').read_text().split('fn test_zmq_pub_fans_out_to_all_sockets')[1].split('stop.store')[0],
}
for k,v in checks.items():
print(f'{k}: {v}')
# Check ZeroMQ PUB/SUB semantics from installed docs/pkg? Read-only best-effort local package metadata only when present.
PYRepository: Alliance-Algorithm/radar-egui
Length of output: 15958
🌐 Web query:
ZeroMQ PUB SUB subscription propagation PUB drop messages if no subscribers documentation
💡 Result:
In ZeroMQ, the PUB socket pattern is designed for high-performance, transient event distribution rather than reliable messaging [1][2]. Regarding your specific questions about message drops and subscription propagation: Message Drops with No Subscribers If a PUB socket has no connected subscribers, it will silently drop all messages sent to it [3][4][5]. The socket does not buffer these messages internally for future subscribers [5]. Once a connection is established, the PUB socket maintains an outgoing message queue for that subscriber, but messages are only queued if the subscriber is connected (or has been connected) and falls behind [1][2][5]. Subscription Propagation and Missed Messages Even when subscribers are present, you may experience message loss during the connection phase [6][3]. This occurs because: 1. Asynchronous Connection: The connection process is asynchronous [6][7]. When a subscriber connects to a publisher, there is a delay while the subscription information (the topic filter) is exchanged [7]. If the publisher sends messages immediately after the connection is initiated but before the subscription handshake completes, those messages may be lost [6][7][3]. 2. Publisher-Side Filtering: In ZeroMQ (v3.x and later), filtering is performed on the publisher side [8][9]. For the publisher to filter correctly, it must have received the subscription command from the subscriber [8][7]. 3. Connection Order: A common issue arises when a PUB socket connects to a SUB socket (rather than the SUB connecting to the PUB) [6][7]. Because the SUB must send its subscription to the PUB, and the connection is asynchronous, the PUB may begin sending messages before it knows the SUB's subscription requirements [6][7]. Mitigation Strategies * Synchronization: To avoid missing the first messages, applications often implement a "side-channel" or use socket monitoring (e.g., ZMQ_EVENT_HANDSHAKE_SUCCEEDED) to ensure the subscriber is fully connected and the subscription has been registered before the publisher begins sending critical data [7][5]. However, even with these events, perfect synchronization can be difficult [7]. * Idempotency and Re-publishing: Because PUB/SUB is inherently unreliable, reliable messaging should be handled at the application level [3][10]. This often involves assigning IDs to messages and having subscribers request missed data or having publishers periodically re-publish state [8][4]. * Architectural Choice: If guaranteed delivery is required, PUB/SUB is generally not the appropriate pattern; consider other patterns (like PUSH/PULL) or implementing a custom reliability layer [3][10]. Top results: [1], [6], [7], [3], [5]
Citations:
- 1: https://rfc.zeromq.org/spec/29/
- 2: http://zeromq-rfc.wikidot.com/spec:29
- 3: https://zguide.zeromq.org/docs/chapter5/
- 4: https://lists.zeromq.org/pipermail/zeromq-dev/2010-September/005828.html
- 5: https://stackoverflow.com/questions/58443889/how-to-make-zmq-pub-client-socket-buffer-messages-while-sub-server-socket-is-dow
- 6: First subscriptions seem to be dropped when a PUB connects to a SUB zeromq/libzmq#2267
- 7: https://www.mail-archive.com/zeromq-dev@lists.zeromq.org/msg31517.html
- 8: https://stackoverflow.com/questions/28609867/zeromq-pub-xpub-xsub-sub-filtering
- 9: http://learning-0mq-with-pyzmq.readthedocs.io/en/latest/pyzmq/patterns/pubsub.html
- 10: https://stackoverflow.com/questions/18008482/how-to-avoid-dropping-messages-zeromq-pub-sub
使用订阅就绪同步替代固定等待。
test_zmq_pub_fans_out_to_all_sockets 第 443 行和第 446 行的固定等待不能保证两个 SUB 的订阅事件已到达 PUB。零MQ 的订阅传播是异步且可能丢失首条消息，这里发出后不再重发的 IDX_GAME_STATE 可能被丢弃，导致测试间歇失败。发送 IDX_GAME_STATE 前，请通过 XPUB 订阅事件等确定性机制等待订阅就绪。
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/runtime/zmq.rs` around lines 443 - 451, Update
test_zmq_pub_fans_out_to_all_sockets to replace both fixed thread::sleep waits
with deterministic subscription-readiness synchronization, using XPUB
subscription events to confirm both SUB sockets are registered before sending
IDX_GAME_STATE. Preserve the existing single-send behavior and subsequent
receives from sub_a and sub_b.
1 participant
Summary
tcp://*:5557(SDR zmq_sub, for GameState and 0x020E encryption rank -> noise grade) andtcp://*:5558(radar_bridge TransmitGameState), so the updated SDR bridge receives match state without changing the SDR repostart_sdrPYTHONPATH now includes/usr/lib/python3/dist-packagesfor system GNU Radio, matching the SDR repostart-sdr.shZmqPubRuntime::startaccepts multiple bind addresses;zmq_start_pubfans out to all socketsVerification
Summary by CodeRabbit
新功能
:5557和:5558端口发布数据，提升连接兼容性。改进
Bug 修复
测试