From 4f0f6a18a8e9680824e5e77d384b10d1085f829e Mon Sep 17 00:00:00 2001 From: Sahil Lenka <76817449+Sahil-u07@users.noreply.github.com> Date: Tue, 3 Nov 2026 03:23:45 +0530 Subject: [PATCH 1/5] fix(cpp): include zmq.hpp outside the concore_base namespace concore_base.hpp included inside namespace concore_base, so every standard header it pulls in got wrapped in that namespace too. Building any C++ node with -DCONCORE_USE_ZMQ failed with ~700 errors, for both concore.hpp and concoredocker.hpp. Moved the include above the namespace, still behind CONCORE_USE_ZMQ. Added TestConcoreZmq.cpp, which sends a request and a reply between two Concore objects over a local socket and checks values and simtime. Build it with -DTEST_DOCKER_HEADER to run it against concoredocker.hpp. --- TestConcoreZmq.cpp | 57 ++++++++++++++++++++++++++++++++++++++++++++++ concore_base.hpp | 6 +++-- 2 files changed, 61 insertions(+), 2 deletions(-) create mode 100644 TestConcoreZmq.cpp diff --git a/TestConcoreZmq.cpp b/TestConcoreZmq.cpp new file mode 100644 index 0000000..071b8ec --- /dev/null +++ b/TestConcoreZmq.cpp @@ -0,0 +1,57 @@ +/** + * TestConcoreZmq.cpp + * + * Tests for the C++ ZMQ transport (CONCORE_USE_ZMQ). + * + * Compile: g++ -std=c++11 -DCONCORE_USE_ZMQ -I. -o TestConcoreZmq TestConcoreZmq.cpp -lzmq + * Add -DTEST_DOCKER_HEADER to run the same tests against concoredocker.hpp. + */ + +#include +#include +#include +#include + +#ifdef TEST_DOCKER_HEADER +#include "concoredocker.hpp" +#else +#include "concore.hpp" +#endif + +static int passed = 0; +static int failed = 0; + +static void check(const std::string& testName, bool condition) { + if (condition) { + std::cout << "PASS: " << testName << std::endl; + ++passed; + } else { + std::cout << "FAIL: " << testName << std::endl; + ++failed; + } +} + +static bool approx(double a, double b) { return std::fabs(a - b) < 1e-9; } + +static void test_round_trip() { + Concore server, client; + server.init_zmq_port("p", "bind", "tcp://127.0.0.1:5591", "REP"); + client.init_zmq_port("p", "connect", "tcp://127.0.0.1:5591", "REQ"); + + client.simtime = 3; + client.write("p", "u", std::vector{1.5, 2.5}, 1); + std::vector u = server.read("p", "u", "[0.0]"); + check("request values arrive", u.size() == 2 && approx(u[0], 1.5) && approx(u[1], 2.5)); + check("request timestamp moves server simtime", approx(server.simtime, 4)); + + server.write("p", "ym", std::vector{9}, 0); + std::vector ym = client.read("p", "ym", "[0.0]"); + check("reply values arrive", ym.size() == 1 && approx(ym[0], 9)); + check("client simtime after reply", approx(client.simtime, 4)); +} + +int main() { + test_round_trip(); + std::cout << "\n" << passed << " passed, " << failed << " failed" << std::endl; + return failed ? 1 : 0; +} diff --git a/concore_base.hpp b/concore_base.hpp index 524d5ba..f8cf486 100644 --- a/concore_base.hpp +++ b/concore_base.hpp @@ -17,6 +17,10 @@ #include #include +#ifdef CONCORE_USE_ZMQ +#include +#endif + namespace concore_base { // =================================================================== @@ -408,8 +412,6 @@ inline std::string tryparam( // ZeroMQ Transport (opt-in: compile with -DCONCORE_USE_ZMQ) // =================================================================== #ifdef CONCORE_USE_ZMQ -#include - /** * ZMQ socket wrapper with bind/connect, timeouts, and retry. */ From 6d13bc7875a04ea31d0855fb26f2b69df5484851 Mon Sep 17 00:00:00 2001 From: Sahil Lenka <76817449+Sahil-u07@users.noreply.github.com> Date: Tue, 3 Nov 2026 03:24:13 +0530 Subject: [PATCH 2/5] fix(cpp): let ZMQ REQ ports recover after a missed reply Same problem Java (#577) and Python (#599) had: once a REQ socket misses a reply, every later send fails with "Operation cannot be accomplished in current state", so the node can never write on that port again. Set ZMQ_REQ_RELAXED and ZMQ_REQ_CORRELATE on REQ sockets, like the other two runtimes. Also moved the existing options to set(zmq::sockopt::...) since setsockopt() is deprecated in cppzmq and warns with -Wall. The new test is the C++ version of the Python one: the first reply never comes, the second request still goes out and the late reply is dropped. --- TestConcoreZmq.cpp | 38 ++++++++++++++++++++++++++++++++++++++ concore_base.hpp | 12 +++++++++--- 2 files changed, 47 insertions(+), 3 deletions(-) diff --git a/TestConcoreZmq.cpp b/TestConcoreZmq.cpp index 071b8ec..dec400f 100644 --- a/TestConcoreZmq.cpp +++ b/TestConcoreZmq.cpp @@ -50,8 +50,46 @@ static void test_round_trip() { check("client simtime after reply", approx(client.simtime, 4)); } +// Same scenario as the Python test for issue #598: the reply to the first +// request never comes, the REQ port must still be able to send again. +static void test_req_recovers_after_missing_reply() { + zmq::context_t ctx(1); + zmq::socket_t peer(ctx, ZMQ_REP); + peer.set(zmq::sockopt::linger, 0); + peer.set(zmq::sockopt::rcvtimeo, 2000); + peer.bind("tcp://127.0.0.1:5592"); + + concore_base::ZeroMQPort req("connect", "tcp://127.0.0.1:5592", ZMQ_REQ); + req.socket.set(zmq::sockopt::rcvtimeo, 50); + + auto peer_recv = [&]() { + zmq::message_t m; + auto res = peer.recv(m, zmq::recv_flags::none); + return res ? m.to_string() : std::string(""); + }; + auto peer_send = [&](const std::string& s) { + try { + peer.send(zmq::buffer(s), zmq::send_flags::none); + } catch (const zmq::error_t&) {} // peer is out of step when the REQ side is stuck + }; + + req.send_with_retry({0, 1}); + check("first request arrives", peer_recv() == "[0, 1]"); + check("missing reply times out", req.recv_with_retry().empty()); + + req.send_with_retry({0, 2}); + peer_send("[0, 99]"); // late reply to the first request + check("second request arrives", peer_recv() == "[0, 2]"); + peer_send("[0, 7]"); + + req.socket.set(zmq::sockopt::rcvtimeo, 2000); + std::vector reply = req.recv_with_retry(); + check("late reply is dropped", reply.size() == 2 && approx(reply[1], 7)); +} + int main() { test_round_trip(); + test_req_recovers_after_missing_reply(); std::cout << "\n" << passed << " passed, " << failed << " failed" << std::endl; return failed ? 1 : 0; } diff --git a/concore_base.hpp b/concore_base.hpp index f8cf486..01c70a6 100644 --- a/concore_base.hpp +++ b/concore_base.hpp @@ -426,9 +426,15 @@ class ZeroMQPort { : context(1), socket(context, socket_type), port_type(port_type_), address(address_) { - socket.setsockopt(ZMQ_RCVTIMEO, 2000); - socket.setsockopt(ZMQ_SNDTIMEO, 2000); - socket.setsockopt(ZMQ_LINGER, 0); + socket.set(zmq::sockopt::rcvtimeo, 2000); + socket.set(zmq::sockopt::sndtimeo, 2000); + socket.set(zmq::sockopt::linger, 0); + + // Let REQ send again after a missed reply, and drop stale replies + if (socket_type == ZMQ_REQ) { + socket.set(zmq::sockopt::req_relaxed, 1); + socket.set(zmq::sockopt::req_correlate, 1); + } if (port_type == "bind") socket.bind(address); From 468b2255e27ea8316bc8ddcd2e27a7ffba2a6201 Mon Sep 17 00:00:00 2001 From: Sahil Lenka <76817449+Sahil-u07@users.noreply.github.com> Date: Tue, 3 Nov 2026 03:26:07 +0530 Subject: [PATCH 3/5] test(cpp): run the write_zmq conformance cases on the C++ runtime The matrix marked both C++ write_zmq rows as passing from reading the code, since there was no ZMQ test. Added both fixture cases to TestConcoreZmq.cpp: a list write sends [simtime+delta, ...] and leaves simtime alone, and a string write is sent as is. Updated the two notes to point to the test. --- TestConcoreZmq.cpp | 70 +++++++++++++------ .../cross_runtime_matrix.phase2.json | 4 +- 2 files changed, 51 insertions(+), 23 deletions(-) diff --git a/TestConcoreZmq.cpp b/TestConcoreZmq.cpp index dec400f..094c82d 100644 --- a/TestConcoreZmq.cpp +++ b/TestConcoreZmq.cpp @@ -33,6 +33,28 @@ static void check(const std::string& testName, bool condition) { static bool approx(double a, double b) { return std::fabs(a - b) < 1e-9; } +// Plain REP socket on the other end, so tests can see the exact bytes sent. +struct RepPeer { + zmq::context_t ctx{1}; + zmq::socket_t sock{ctx, ZMQ_REP}; + + explicit RepPeer(const std::string& address) { + sock.set(zmq::sockopt::linger, 0); + sock.set(zmq::sockopt::rcvtimeo, 2000); + sock.bind(address); + } + std::string recv() { + zmq::message_t m; + auto res = sock.recv(m, zmq::recv_flags::none); + return res ? m.to_string() : std::string(""); + } + void send(const std::string& s) { + try { + sock.send(zmq::buffer(s), zmq::send_flags::none); + } catch (const zmq::error_t&) {} // out of step when the REQ side is stuck + } +}; + static void test_round_trip() { Concore server, client; server.init_zmq_port("p", "bind", "tcp://127.0.0.1:5591", "REP"); @@ -50,37 +72,42 @@ static void test_round_trip() { check("client simtime after reply", approx(client.simtime, 4)); } +// The two write_zmq cases from tests/protocol_fixtures/python_phase1_cases.json +static void test_write_zmq_fixture_cases() { + RepPeer peer("tcp://127.0.0.1:5593"); + Concore c; + c.init_zmq_port("p", "connect", "tcp://127.0.0.1:5593", "REQ"); + c.simtime = 10; + c.write("p", "data", std::vector{1.5, 2.5}, 2); + check("list payload gets simtime+delta in front", peer.recv() == "[12, 1.5, 2.5]"); + check("list write leaves simtime alone", approx(c.simtime, 10)); + +#ifndef TEST_DOCKER_HEADER // concoredocker.hpp has no string write + peer.send("[0]"); + c.read("p", "ack", "[0]"); + c.simtime = 10; + c.delay = 0; + c.write("p", "status", std::string("ok"), 3); + check("string payload is sent as is", peer.recv() == "ok"); + check("string write leaves simtime alone", approx(c.simtime, 10)); +#endif +} + // Same scenario as the Python test for issue #598: the reply to the first // request never comes, the REQ port must still be able to send again. static void test_req_recovers_after_missing_reply() { - zmq::context_t ctx(1); - zmq::socket_t peer(ctx, ZMQ_REP); - peer.set(zmq::sockopt::linger, 0); - peer.set(zmq::sockopt::rcvtimeo, 2000); - peer.bind("tcp://127.0.0.1:5592"); - + RepPeer peer("tcp://127.0.0.1:5592"); concore_base::ZeroMQPort req("connect", "tcp://127.0.0.1:5592", ZMQ_REQ); req.socket.set(zmq::sockopt::rcvtimeo, 50); - auto peer_recv = [&]() { - zmq::message_t m; - auto res = peer.recv(m, zmq::recv_flags::none); - return res ? m.to_string() : std::string(""); - }; - auto peer_send = [&](const std::string& s) { - try { - peer.send(zmq::buffer(s), zmq::send_flags::none); - } catch (const zmq::error_t&) {} // peer is out of step when the REQ side is stuck - }; - req.send_with_retry({0, 1}); - check("first request arrives", peer_recv() == "[0, 1]"); + check("first request arrives", peer.recv() == "[0, 1]"); check("missing reply times out", req.recv_with_retry().empty()); req.send_with_retry({0, 2}); - peer_send("[0, 99]"); // late reply to the first request - check("second request arrives", peer_recv() == "[0, 2]"); - peer_send("[0, 7]"); + peer.send("[0, 99]"); // late reply to the first request + check("second request arrives", peer.recv() == "[0, 2]"); + peer.send("[0, 7]"); req.socket.set(zmq::sockopt::rcvtimeo, 2000); std::vector reply = req.recv_with_retry(); @@ -89,6 +116,7 @@ static void test_req_recovers_after_missing_reply() { int main() { test_round_trip(); + test_write_zmq_fixture_cases(); test_req_recovers_after_missing_reply(); std::cout << "\n" << passed << " passed, " << failed << " failed" << std::endl; return failed ? 1 : 0; diff --git a/tests/protocol_fixtures/cross_runtime_matrix.phase2.json b/tests/protocol_fixtures/cross_runtime_matrix.phase2.json index 6aedeb3..6e9e27c 100644 --- a/tests/protocol_fixtures/cross_runtime_matrix.phase2.json +++ b/tests/protocol_fixtures/cross_runtime_matrix.phase2.json @@ -178,7 +178,7 @@ "cpp": { "status": "observed_pass", "classification": "required", - "note": "write_ZMQ(vector) in concore.hpp inserts simtime+delta at the front of the outgoing payload and sends it, without writing back to the Concore instance's simtime member (the code even carries an explicit comment, 'simtime must not be mutated here (issue #385)'). Matches this fixture's semantics. Gated behind the CONCORE_USE_ZMQ build macro; TestConcoreHpp.cpp has no ZMQ coverage, so this was verified by reading concore.hpp directly rather than by an existing test." + "note": "write_ZMQ(vector) in concore.hpp inserts simtime+delta at the front of the outgoing payload and sends it, without writing back to the Concore instance's simtime member (the code even carries an explicit comment, 'simtime must not be mutated here (issue #385)'). Matches this fixture's semantics. Gated behind the CONCORE_USE_ZMQ build macro; TestConcoreZmq.cpp runs this case in CI against concore.hpp and concoredocker.hpp." }, "java": { "status": "observed_pass", @@ -214,7 +214,7 @@ "cpp": { "status": "observed_pass", "classification": "required", - "note": "write_ZMQ(string) in concore.hpp sends the string as-is with no timestamp prepended and no simtime mutation, matching this fixture. Same CONCORE_USE_ZMQ caveat as the list-payload case; verified by reading concore.hpp, no existing ZMQ test exercises it." + "note": "write_ZMQ(string) in concore.hpp sends the string as-is with no timestamp prepended and no simtime mutation, matching this fixture. Same CONCORE_USE_ZMQ caveat as the list-payload case; TestConcoreZmq.cpp runs this case in CI (concoredocker.hpp has no string write)." }, "java": { "status": "observed_pass", From cc7cb9080c2314a898bc524135b2bea4abd714ae Mon Sep 17 00:00:00 2001 From: Sahil Lenka <76817449+Sahil-u07@users.noreply.github.com> Date: Tue, 3 Nov 2026 03:26:07 +0530 Subject: [PATCH 4/5] ci: build and run the C++ ZMQ tests on Linux, Windows and macOS Nothing built C++ with -DCONCORE_USE_ZMQ before, which is how the namespace include went unnoticed. Installs cppzmq on each runner and runs TestConcoreZmq.cpp. The Linux job also runs it against concoredocker.hpp. --- .github/workflows/ci.yml | 21 ++++++++++++++++++--- 1 file changed, 18 insertions(+), 3 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bb47fa0..69c6d6e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -189,8 +189,8 @@ jobs: steps: - uses: actions/checkout@v4 - - name: Install g++ - run: sudo apt-get update && sudo apt-get install -y g++ + - name: Install g++ and ZeroMQ + run: sudo apt-get update && sudo apt-get install -y g++ libzmq3-dev cppzmq-dev - name: Parse concore.hpp standalone # -fsyntax-only parses concore.hpp without producing an object @@ -209,6 +209,13 @@ jobs: g++ -std=c++11 -Wall -Wextra -I. -o TestConcoreHpp TestConcoreHpp.cpp ./TestConcoreHpp + - name: Build and run C++ ZMQ tests + run: | + g++ -std=c++11 -Wall -Wextra -DCONCORE_USE_ZMQ -I. -o TestConcoreZmq TestConcoreZmq.cpp -lzmq + ./TestConcoreZmq + g++ -std=c++11 -Wall -Wextra -DCONCORE_USE_ZMQ -DTEST_DOCKER_HEADER -I. -o TestConcoreZmqDocker TestConcoreZmq.cpp -lzmq + ./TestConcoreZmqDocker + cpp-test-windows: runs-on: windows-latest defaults: @@ -220,7 +227,7 @@ jobs: - uses: msys2/setup-msys2@v2 with: msystem: UCRT64 - install: mingw-w64-ucrt-x86_64-gcc + install: mingw-w64-ucrt-x86_64-gcc mingw-w64-ucrt-x86_64-cppzmq - name: Build and run C++ tests run: | @@ -228,15 +235,23 @@ jobs: ./TestLiteralEvalCpp g++ -std=c++11 -I. -o TestConcoreHpp TestConcoreHpp.cpp ./TestConcoreHpp + g++ -std=c++11 -DCONCORE_USE_ZMQ -I. -o TestConcoreZmq TestConcoreZmq.cpp -lzmq + ./TestConcoreZmq cpp-test-macos: runs-on: macos-latest steps: - uses: actions/checkout@v4 + - name: Install ZeroMQ + run: brew install cppzmq + - name: Build and run C++ tests run: | g++ -std=c++11 -I. -o TestLiteralEvalCpp TestLiteralEvalCpp.cpp ./TestLiteralEvalCpp g++ -std=c++11 -I. -o TestConcoreHpp TestConcoreHpp.cpp ./TestConcoreHpp + BREW=$(brew --prefix) + g++ -std=c++11 -DCONCORE_USE_ZMQ -I. -I$BREW/include -L$BREW/lib -o TestConcoreZmq TestConcoreZmq.cpp -lzmq + ./TestConcoreZmq From b7f36eb579a8f7ffcd9e5932e5c0094c29aeca89 Mon Sep 17 00:00:00 2001 From: Sahil Lenka <76817449+Sahil-u07@users.noreply.github.com> Date: Tue, 3 Nov 2026 03:40:11 +0530 Subject: [PATCH 5/5] docs: note the cppzmq version the C++ ZMQ transport needs zmq::sockopt, used for the socket options now, came in cppzmq 4.7. --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 1d59a86..e54f4f2 100644 --- a/README.md +++ b/README.md @@ -129,7 +129,7 @@ For a detailed and more scientific documentation, please read our extensive [ope `concore.hpp` supports ZMQ-based communication as an opt-in transport alongside the default file-based I/O. -To enable it, compile with `-DCONCORE_USE_ZMQ` and link against cppzmq: +To enable it, compile with `-DCONCORE_USE_ZMQ` and link against cppzmq (4.7 or newer): ```bash g++ -DCONCORE_USE_ZMQ my_node.cpp -lzmq -o my_node