From 0e0dc0cd38fd7855fa73e25f76f1889f85c8c493 Mon Sep 17 00:00:00 2001 From: Andrey Bychkov Date: Mon, 24 Aug 2026 17:20:49 +0300 Subject: [PATCH] test: make regression tests fail cleanly on pre-fix code - dispatcher tests: register the server before the client connects (Connect race), poll the Connected ack non-blocking with a deadline, keep the client socket non-blocking so a full buffer yields EAGAIN instead of a blocking send() past any deadline - piethernet test: use a TCP client against a never-reading listener (server-object writeDevice is a no-op) and 32 MB to exceed the ~10 MB loopback buffers With piethernet/pistreampacker reverted to master all four tests fail (8 ms .. 40 s); with the fixes ctest is 342/342. --- tests/cloud/dispatcher_test.cpp | 149 ++++++++++++++++++++------------ tests/io/testpiethernet.cpp | 65 ++++++++------ 2 files changed, 133 insertions(+), 81 deletions(-) diff --git a/tests/cloud/dispatcher_test.cpp b/tests/cloud/dispatcher_test.cpp index 9fdcde5a..fd4440b4 100644 --- a/tests/cloud/dispatcher_test.cpp +++ b/tests/cloud/dispatcher_test.cpp @@ -30,6 +30,7 @@ #include #include #include +#include #include static const ushort PACKET_SIGN = 0xAFBE; // PIStreamPacker default sign @@ -79,16 +80,22 @@ static void send_frame(int fd, uchar type, uchar role, const void * payload, int } } -// stream-safe non-blocking send of one FULL frame: on EAGAIN retry the -// SAME offset. A partial write followed by the next frame would corrupt -// the receiver's PIStreamPacker byte stream (half frame + next frame) -// and the resync would swallow the following frames. -static bool send_frame_nb(int fd, uchar type, uchar role, const void * payload, int plen) { +// non-blocking send of one FULL frame with a per-frame deadline. On +// EAGAIN retry the SAME offset; a partial write followed by the next +// frame would corrupt the receiver's PIStreamPacker byte stream, so the +// deadline only bounds how long one frame may wait for the buffer to +// drain. Returns false (the frame may be partially written) when the +// deadline expires: on the buggy code the reader never drains, so the +// flood exits via the caller's failure limit instead of hanging, and the +// already-jammed reader never parses the (possibly corrupted) tail. +static bool send_frame_nb(int fd, uchar type, uchar role, const void * payload, int plen, PISystemTime frame_timeout = 3_s) { uchar frame[65536]; frame_build(frame, type, role, payload, plen); int total = 6 + 3 + plen; int off = 0; + PITimeMeasurer tm; while (off < total) { + if (tm.elapsed() > frame_timeout) return false; ssize_t r = send(fd, frame + off, total - off, MSG_NOSIGNAL); if (r < 0) { if (errno == EAGAIN || errno == EWOULDBLOCK) { @@ -112,22 +119,48 @@ static bool wait_for(std::function cond, PISystemTime timeout) { return cond(); } -// waits for the 13-byte Connected ack frame on fd +// polls (non-blocking) for the 13-byte Connected ack frame on fd static bool wait_connected_ack(int fd) { uchar rbuf[64]; memset(rbuf, 0, sizeof(rbuf)); int rn = 0; - for (int i = 0; i < 50 && rn < (int)sizeof(rbuf); ++i) { + int fl = fcntl(fd, F_GETFL, 0); + fcntl(fd, F_SETFL, fl | O_NONBLOCK); + PITimeMeasurer tm; + while (tm.elapsed() < 5_s && rn < (int)sizeof(rbuf)) { ssize_t r = recv(fd, rbuf + rn, sizeof(rbuf) - rn, 0); - if (r <= 0) break; - rn += (int)r; - if (rn >= 13) // 2 sign + 4 size + 7 payload - return true; - piMinSleep(); + if (r > 0) { + rn += (int)r; + if (rn >= 13) // 2 sign + 4 size + 7 payload + break; + } else if (r < 0 && errno != EAGAIN && errno != EWOULDBLOCK) { + break; + } else + piMinSleep(); } + fcntl(fd, F_SETFL, fl); return rn >= 13; } +// connects the logical server A; the caller must wait for the dispatcher +// to register it (getServer() != nullptr) before connecting a client, +// otherwise the client's Connect may be processed first and rejected +static bool connect_server(uint16_t port, const uchar * suuid, int & A) { + A = open_conn(port); + if (A < 0) return false; + send_frame(A, 1 /*Connect*/, 1 /*Server*/, suuid, 32); + return true; +} + +// connects the logical client B (the server must already be registered) +// and waits for the Connected ack +static bool connect_client(uint16_t port, const uchar * suuid, int & B) { + B = open_conn(port); + if (B < 0) return false; + send_frame(B, 1 /*Connect*/, 2 /*Client*/, suuid, 32); + return wait_connected_ack(B); +} + // keeps the logical server alive past the 15 s ping-timeout removal struct PingKeeper { void start(int fd, const uchar * suuid) { @@ -143,38 +176,30 @@ struct PingKeeper { const uchar * suuid_ = nullptr; }; -// connects A (logical server) and B (logical client) to the dispatcher on -// port, both using the same server uuid, and waits for B's Connected ack -static bool connect_pair(uint16_t port, const uchar * suuid, int & A, int & B) { - A = open_conn(port); - if (A < 0) return false; - send_frame(A, 1 /*Connect*/, 1 /*Server*/, suuid, 32); - - B = open_conn(port); - if (B < 0) return false; - send_frame(B, 1 /*Connect*/, 2 /*Client*/, suuid, 32); - return wait_connected_ack(B); -} - - +// both tests run the server on the heap: with a buggy piethernet a failed +// assertion would unwind into the DispatcherServer destructor, which waits +// for the jammed read thread and hangs, so the server is deleted only on +// the success path TEST(PICloudDispatcher, ServerDisconnectWhileClientForwarding) { - constexpr uint16_t port = 10199; - DispatcherServer server(PINetworkAddress("127.0.0.1", port)); - server.start(); + constexpr uint16_t port = 10199; + DispatcherServer * server = new DispatcherServer(PINetworkAddress("127.0.0.1", port)); + server->start(); uchar suuid[32]; memset(suuid, 0xAA, 32); int A = -1, B = -1; - ASSERT_TRUE(connect_pair(port, suuid, A, B)); - - // the dispatcher must know about the server connection - ASSERT_TRUE(wait_for([&]() { return server.getServer(0) != nullptr; }, 5_s)); + ASSERT_TRUE(connect_server(port, suuid, A)); + // the dispatcher must know about the server before the client + // connects, otherwise the client's Connect may be rejected + ASSERT_TRUE(wait_for([&]() { return server->getServer(0) != nullptr; }, 5_s)); + ASSERT_TRUE(connect_client(port, suuid, B)); PingKeeper pinger; pinger.start(A, suuid); // flood B with data frames: B's read thread forwards them to A's - // socket, which nobody reads, so it jams inside the forward + // socket, which nobody reads, so it jams inside the forward; a few + // consecutive full-frame timeouts mean the reader is jammed { const int plen = 32 * 1024; uchar * data = new uchar[plen]; @@ -182,8 +207,8 @@ TEST(PICloudDispatcher, ServerDisconnectWhileClientForwarding) { int fl = fcntl(B, F_GETFL, 0); fcntl(B, F_SETFL, fl | O_NONBLOCK); int sent = 0, consec_failed = 0; - while (sent < 1024 && consec_failed < 2000) { - if (send_frame_nb(B, 3 /*Data*/, 2 /*Client*/, data, plen)) { + while (sent < 1024 && consec_failed < 4) { + if (send_frame_nb(B, 3 /*Data*/, 2 /*Client*/, data, plen, 5_s)) { ++sent; consec_failed = 0; } else @@ -199,56 +224,73 @@ TEST(PICloudDispatcher, ServerDisconnectWhileClientForwarding) { // the still-registered client under map_mutex (close()/stopAndWait() // on the jammed read thread) and froze forever uchar zero4[4] = {0, 0, 0, 0}; - send_frame(A, 2 /*Disconnect*/, 1 /*Server*/, zero4, 4); + send_frame_nb(A, 2 /*Disconnect*/, 1 /*Server*/, zero4, 4); // the dispatcher must stay responsive std::atomic_bool prober_done{false}; PIThread prober; prober.start([&] { - server.picoutStatus(); + server->picoutStatus(); prober_done.store(true); }); ASSERT_TRUE(wait_for([&]() { return prober_done.load(); }, 10_s)) << "dispatcher frozen: picoutStatus() did not return"; prober.stopAndWait(); - // both connections are removed - ASSERT_TRUE(wait_for([&]() { return server.getConnection(0) == nullptr && server.getConnection(1) == nullptr; }, 10_s)) + // both connections are removed from the bookkeeping + ASSERT_TRUE(wait_for([&]() { return server->getConnection(0) == nullptr && server->getConnection(1) == nullptr; }, 10_s)) << "connections were not removed after the server disconnect"; pinger.stop(); + + // the drainer must actually close the jammed client: + // ~DispatcherServer waits for all clients, so it must finish in finite + // time. On buggy code the wait is not interruptible, so on failure the + // deleting thread is detached (leaked) rather than cancelled: cancelling + // a thread inside PIP C++ code forces an unwind that aborts the process. + std::atomic_bool deleted{false}; + std::thread deleter([&] { + delete server; + deleted.store(true); + }); + bool destroyed = wait_for([&]() { return deleted.load(); }, 15_s); + if (destroyed) + deleter.join(); + else + deleter.detach(); + ASSERT_TRUE(destroyed) << "drainer stuck closing the jammed client: destructor did not finish"; } TEST(PICloudDispatcher, ClientRemovableWhileServerNotReading) { - constexpr uint16_t port = 10297; - DispatcherServer server(PINetworkAddress("127.0.0.1", port)); - server.start(); + constexpr uint16_t port = 10297; + DispatcherServer * server = new DispatcherServer(PINetworkAddress("127.0.0.1", port)); + server->start(); uchar suuid[32]; memset(suuid, 0xAA, 32); int A = -1, B = -1; - ASSERT_TRUE(connect_pair(port, suuid, A, B)); - ASSERT_TRUE(wait_for([&]() { return server.getServer(0) != nullptr; }, 5_s)); + ASSERT_TRUE(connect_server(port, suuid, A)); + ASSERT_TRUE(wait_for([&]() { return server->getServer(0) != nullptr; }, 5_s)); + ASSERT_TRUE(connect_client(port, suuid, B)); PingKeeper pinger; pinger.start(A, suuid); // same flood: B's read thread jams in A::writeDevice(), where A is - // alive and never stopped + // alive and never stopped. B stays non-blocking: a blocking send() + // into a full buffer would hang inside the call, past any deadline. { const int plen = 32 * 1024; uchar * data = new uchar[plen]; memset(data, 0x42, plen); - int fl = fcntl(B, F_GETFL, 0); - fcntl(B, F_SETFL, fl | O_NONBLOCK); + fcntl(B, F_SETFL, O_NONBLOCK); int sent = 0, consec_failed = 0; - while (sent < 256 && consec_failed < 2000) { - if (send_frame_nb(B, 3 /*Data*/, 2 /*Client*/, data, plen)) { + while (sent < 256 && consec_failed < 4) { + if (send_frame_nb(B, 3 /*Data*/, 2 /*Client*/, data, plen, 5_s)) { ++sent; consec_failed = 0; } else ++consec_failed; } - fcntl(B, F_SETFL, fl); delete[] data; } @@ -256,9 +298,10 @@ TEST(PICloudDispatcher, ClientRemovableWhileServerNotReading) { // the flood. Pre-fix B's read thread never returned from the write // to A, so the frame was never processed and B was never removed. uchar zero4[4] = {0, 0, 0, 0}; - send_frame(B, 2 /*Disconnect*/, 2 /*Client*/, zero4, 4); + ASSERT_TRUE(send_frame_nb(B, 2 /*Disconnect*/, 2 /*Client*/, zero4, 4, 10_s)) << "could not enqueue the disconnect frame"; - ASSERT_TRUE(wait_for([&]() { return server.getConnection(1) == nullptr && server.getConnection(0) != nullptr; }, 30_s)) + ASSERT_TRUE(wait_for([&]() { return server->getConnection(1) == nullptr && server->getConnection(0) != nullptr; }, 30_s)) << "client not removed while the server connection never read"; pinger.stop(); + delete server; } diff --git a/tests/io/testpiethernet.cpp b/tests/io/testpiethernet.cpp index 9fafa59b..5e793348 100644 --- a/tests/io/testpiethernet.cpp +++ b/tests/io/testpiethernet.cpp @@ -11,22 +11,6 @@ #include #include -static int connect_raw(uint16_t port) { - int fd = socket(AF_INET, SOCK_STREAM, 0); - if (fd < 0) return -1; - sockaddr_in sa; - memset(&sa, 0, sizeof(sa)); - sa.sin_family = AF_INET; - sa.sin_port = htons(port); - inet_pton(AF_INET, "127.0.0.1", &sa.sin_addr); - for (int i = 0; i < 100; ++i) { - if (connect(fd, (sockaddr *)&sa, sizeof(sa)) == 0) return fd; - usleep(100000); - } - ::close(fd); - return -1; -} - static bool wait_for(std::function cond, PISystemTime timeout) { PITimeMeasurer tm; while (tm.elapsed() < timeout) { @@ -41,30 +25,55 @@ static bool wait_for(std::function cond, PISystemTime timeout) { // busy writer must return. TEST(PIEthernet, WriteThreadStopsWhilePeerNotReading) { constexpr uint16_t port = 10299; - PIEthernet eth(PIEthernet::TCP_Server); - eth.setReadAddress(PINetworkAddress("127.0.0.1", port)); - ASSERT_TRUE(eth.listen(true)); - // a client that connects but never reads - int cs = connect_raw(port); - ASSERT_GE(cs, 0); - ASSERT_TRUE(wait_for([&]() { return eth.clientsCount() >= 1; }, 5_s)) << "server did not accept the connection"; + // a raw listener that accepts the connection but never reads it + int ls = socket(AF_INET, SOCK_STREAM, 0); + ASSERT_GE(ls, 0); + int on = 1; + setsockopt(ls, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)); + sockaddr_in sa; + memset(&sa, 0, sizeof(sa)); + sa.sin_family = AF_INET; + sa.sin_port = htons(port); + inet_pton(AF_INET, "127.0.0.1", &sa.sin_addr); + ASSERT_EQ(0, bind(ls, (sockaddr *)&sa, sizeof(sa))); + ASSERT_EQ(0, listen(ls, 1)); - eth.startThreadedWrite(); + // on the heap: if the stop below hangs, the destructor would wait for + // the stuck thread too, so on failure the device is deliberately + // leaked (the test process is about to exit anyway) + PIEthernet * eth = new PIEthernet(PIEthernet::TCP_Client); + // connect() only flags the device, the read thread performs the actual + // TCP connection; the read thread also provides the stopping state + // that the EAGAIN retry loop in writeDevice() observes + ASSERT_TRUE(eth->connect(PINetworkAddress("127.0.0.1", port))); + eth->startThreadedRead(); + eth->startThreadedWrite(); + + int acc = accept(ls, nullptr, nullptr); + ASSERT_GE(acc, 0); + ASSERT_TRUE(wait_for([&]() { return eth->isConnected(); }, 5_s)) << "client did not connect"; + + // well above the loopback kernel buffer capacity (tcp_wmem + tcp_rmem + // ~10 MB), so the send buffer is guaranteed to fill up and the write + // thread to reach the EAGAIN path in writeDevice() const size_t MB = 1024 * 1024; PIByteArray big(8 * MB, 0x42); - eth.writeThreaded(big); + for (int i = 0; i < 4; ++i) + eth->writeThreaded(big); - // let the write thread queue up and jam on the send + // let the write thread fill the buffers and jam on the send wait_for([] { return false; }, 2_s); std::atomic_bool done{false}; PIThread stopper; stopper.start([&] { - eth.stopAndWait(); + eth->stopAndWait(); done.store(true); }); ASSERT_TRUE(wait_for([&]() { return done.load(); }, 15_s)) << "stopAndWait() hung on a stalled write"; stopper.stopAndWait(); - ::close(cs); + delete eth; + ::close(acc); + ::close(ls); }