SHS/pip
4
3
Fork 0
Code Issues 14 Pull Requests 4 Actions Releases Wiki Activity

test: make regression tests fail cleanly on pre-fix code

Browse Source 
- 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.
This commit is contained in:
andrey
2026-08-24 17:20:49 +03:00
parent 3abf0a0808
commit 0e0dc0cd38
2 changed files with 133 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/cloud/dispatcher_test.cpp
+96 -53
View File
@@ -30,6 +30,7 @@
#include <functional>
#include <netinet/in.h>
#include <sys/socket.h>
#include <thread>
#include <unistd.h>
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<bool()> 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;
}
tests/io/testpiethernet.cpp
+37 -28
View File
@@ -11,22 +11,6 @@
#include <sys/socket.h>
#include <unistd.h>
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<bool()> cond, PISystemTime timeout) {
PITimeMeasurer tm;
while (tm.elapsed() < timeout) {
@@ -41,30 +25,55 @@ static bool wait_for(std::function<bool()> 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);
}