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 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 diff --git a/TestConcoreZmq.cpp b/TestConcoreZmq.cpp new file mode 100644 index 0000000..094c82d --- /dev/null +++ b/TestConcoreZmq.cpp @@ -0,0 +1,123 @@ +/** + * 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; } + +// 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"); + 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)); +} + +// 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() { + 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); + + 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_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/concore_base.hpp b/concore_base.hpp index 524d5ba..01c70a6 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. */ @@ -424,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); 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",