From 422612a29e8514c3bb9fbf09970db1407f49ec4a Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Wed, 22 Jul 2026 02:36:08 +0300 Subject: [PATCH 1/3] fix: advance server write offset on completion --- include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp | 1 - 1 file changed, 1 deletion(-) diff --git a/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp b/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp index 2e4f978..341d5d1 100644 --- a/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp +++ b/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp @@ -425,7 +425,6 @@ namespace SimpleNamedPipe { size_t bytes_to_copy = (std::min)(buffer_size, remaining); auto& buffer = m_write_buffers[index]; buffer.assign(cmd.message.begin() + msg_offset, cmd.message.begin() + msg_offset + bytes_to_copy); - cmd.offset += bytes_to_copy; OVERLAPPED* ov = &m_write_overlapped[index]; memset(ov, 0, sizeof(OVERLAPPED)); From 59515c49af9fbf168f45882345332cfa5c07e5dc Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Wed, 22 Jul 2026 03:56:37 +0300 Subject: [PATCH 2/3] test: cover large named pipe messages --- include/SimpleNamedPipe/NamedPipeServer.hpp | 1 + .../NamedPipeServer/NamedPipeServer.ipp | 111 +++++++++++++----- tests/client_server_test.cpp | 67 +++++++++++ 3 files changed, 150 insertions(+), 29 deletions(-) diff --git a/include/SimpleNamedPipe/NamedPipeServer.hpp b/include/SimpleNamedPipe/NamedPipeServer.hpp index 0108683..cf47e62 100644 --- a/include/SimpleNamedPipe/NamedPipeServer.hpp +++ b/include/SimpleNamedPipe/NamedPipeServer.hpp @@ -162,6 +162,7 @@ namespace SimpleNamedPipe { void process_write_commands(); void process_write_commands(size_t index); void post_next_write(size_t index); + bool post_next_read(size_t index, HANDLE completion_port, OVERLAPPED* ov); void handle_write_completion(size_t index, size_t bytes_transferred); void handle_close(size_t index, HANDLE completion_port); void cleanup_pending_operations(const std::error_code& reason); diff --git a/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp b/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp index 341d5d1..399686f 100644 --- a/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp +++ b/include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp @@ -301,13 +301,30 @@ namespace SimpleNamedPipe { } if (err == ERROR_BROKEN_PIPE || - err == ERROR_NO_DATA) { + err == ERROR_NO_DATA || + err == ERROR_PIPE_NOT_CONNECTED) { notify_disconnected(index, std::error_code(err, std::system_category())); DisconnectNamedPipe(m_pipes[index]); reconnect_client(index, completion_port, &m_read_overlapped[index]); continue; } + if (err == ERROR_MORE_DATA && + index < MAX_CLIENTS && + ov == &m_read_overlapped[index]) { + if (bytes_transferred > 0) { + m_message_buffers[index].append( + m_read_buffers[index].data(), + bytes_transferred); + } + if (!m_is_connected[index].load(std::memory_order_acquire)) { + continue; + } + + post_next_read(index, completion_port, &m_read_overlapped[index]); + continue; + } + notify_error(std::error_code(static_cast(err), std::system_category())); continue; } @@ -329,39 +346,73 @@ namespace SimpleNamedPipe { } else // Handle read completion if (ov == &m_read_overlapped[index] && bytes_transferred > 0) { - err = GetLastError(); m_message_buffers[index].append(m_read_buffers[index].data(), bytes_transferred); - if (err != ERROR_MORE_DATA) { - notify_message(index); - } + notify_message(index); } // Skip reading if still not connected if (!m_is_connected[index].load(std::memory_order_acquire)) continue; - DWORD dummy = 0; - OVERLAPPED* new_ov = &m_read_overlapped[index]; - BOOL result = ReadFile(m_pipes[index], m_read_buffers[index].data(), static_cast(m_read_buffers[index].size()), &dummy, new_ov); + post_next_read(index, completion_port, &m_read_overlapped[index]); + } - err = GetLastError(); - if (!result && err != ERROR_IO_PENDING) { - if (err == ERROR_BROKEN_PIPE || - err == ERROR_NO_DATA) { - notify_disconnected(index, std::error_code(static_cast(err), std::system_category())); - DisconnectNamedPipe(m_pipes[index]); - reconnect_client(index, completion_port, new_ov); - continue; - } else - if (err == ERROR_OPERATION_ABORTED) { - continue; - } else { - notify_error(std::error_code(static_cast(err), std::system_category())); - continue; + notify_stop(config); + } + + SIMPLE_NAMED_PIPE_INLINE bool NamedPipeServer::post_next_read( + size_t index, + HANDLE completion_port, + OVERLAPPED* ov) { + if (index >= MAX_CLIENTS) { + notify_error(make_error_code(NamedPipeErrc::ClientIndexOutOfRange)); + return false; + } + if (!m_is_connected[index].load(std::memory_order_acquire)) { + return true; + } + + for (;;) { + memset(ov, 0, sizeof(OVERLAPPED)); + + DWORD bytes_read = 0; + BOOL result = ReadFile( + m_pipes[index], + m_read_buffers[index].data(), + static_cast(m_read_buffers[index].size()), + &bytes_read, + ov); + + DWORD err = GetLastError(); + if (result || err == ERROR_IO_PENDING) { + return true; + } + + if (err == ERROR_MORE_DATA) { + if (bytes_read > 0) { + m_message_buffers[index].append( + m_read_buffers[index].data(), + bytes_read); } + continue; } - } - notify_stop(config); + if (err == ERROR_BROKEN_PIPE || + err == ERROR_NO_DATA || + err == ERROR_PIPE_NOT_CONNECTED) { + notify_disconnected( + index, + std::error_code(static_cast(err), std::system_category())); + DisconnectNamedPipe(m_pipes[index]); + reconnect_client(index, completion_port, ov); + return false; + } + if (err == ERROR_OPERATION_ABORTED) { + return false; + } + + notify_error(std::error_code(static_cast(err), std::system_category())); + return false; + } } // Process all accumulated write commands @@ -417,20 +468,21 @@ namespace SimpleNamedPipe { return; } - size_t buffer_size = m_write_buffers[index].size(); size_t msg_offset = cmd.offset; size_t remaining = (msg_offset < cmd.message.size()) ? (cmd.message.size() - msg_offset) : 0; - size_t bytes_to_copy = (std::min)(buffer_size, remaining); - auto& buffer = m_write_buffers[index]; - buffer.assign(cmd.message.begin() + msg_offset, cmd.message.begin() + msg_offset + bytes_to_copy); OVERLAPPED* ov = &m_write_overlapped[index]; memset(ov, 0, sizeof(OVERLAPPED)); DWORD bytes_written = 0; - BOOL success = WriteFile(m_pipes[index], buffer.data(), static_cast(bytes_to_copy), &bytes_written, ov); + BOOL success = WriteFile( + m_pipes[index], + cmd.message.data() + msg_offset, + static_cast(remaining), + &bytes_written, + ov); DWORD err = GetLastError(); if (!success && err != ERROR_IO_PENDING) { if (cmd.on_done) cmd.on_done(std::error_code(static_cast(err), std::system_category())); @@ -459,6 +511,7 @@ namespace SimpleNamedPipe { } if (err == ERROR_BROKEN_PIPE || err == ERROR_NO_DATA || + err == ERROR_PIPE_NOT_CONNECTED || err == ERROR_OPERATION_ABORTED) { DisconnectNamedPipe(m_pipes[index]); continue; diff --git a/tests/client_server_test.cpp b/tests/client_server_test.cpp index a92725f..cdbde85 100644 --- a/tests/client_server_test.cpp +++ b/tests/client_server_test.cpp @@ -266,6 +266,71 @@ void basic_io_edges(TestRunner& tr) { tr.expect(harness.error_count() == 0, with_error("server reported error: ", harness.error())); } +void large_messages_cross_transport_buffers(TestRunner& tr) { + ServerHarness harness("large_buffers", 64, 25); + harness.start(); + if (!tr.expect(harness.wait_started(), "server did not start")) return; + + auto client = make_client(harness.pipe_name, 37, 3000); + std::error_code ec; + if (!tr.expect(client->connect(&ec), with_error("client connect failed: ", ec))) return; + tr.expect(harness.wait_connected_at_least(1), "server did not observe client connect"); + + std::string payload; + payload.reserve(8192); + for (int i = 0; i < 8192; ++i) { + payload.push_back(static_cast('A' + (i % 26))); + } + + tr.expect(client->write(payload, &ec), with_error("large message write failed: ", ec)); + std::string response; + tr.expect(client->read(response, 3000, &ec), with_error("large message read failed: ", ec)); + const std::string expected = "Echo: " + payload; + if (!tr.expect(response == expected, "large message echo bytes did not match")) { + std::cerr << "expected bytes=" << expected.size() + << " actual bytes=" << response.size() << "\n"; + } + tr.expect(harness.wait_messages_at_least(1), "server did not receive large message"); + + client->close(); + tr.expect(harness.wait_disconnected_at_least(1), "server did not observe client close"); + harness.stop(); + tr.expect(harness.error_count() == 0, with_error("server reported error: ", harness.error())); +} + +void adjacent_messages_are_not_merged(TestRunner& tr) { + ServerHarness harness("adjacent_messages", 1024, 25); + harness.start(); + if (!tr.expect(harness.wait_started(), "server did not start")) return; + + auto client = make_client(harness.pipe_name, 1024, 3000); + std::error_code ec; + if (!tr.expect(client->connect(&ec), with_error("client connect failed: ", ec))) return; + tr.expect(harness.wait_connected_at_least(1), "server did not observe client connect"); + + const std::string first = "first-json-rpc-request"; + const std::string second = "second-json-rpc-request"; + tr.expect(client->write(first, &ec), with_error("first adjacent write failed: ", ec)); + tr.expect(client->write(second, &ec), with_error("second adjacent write failed: ", ec)); + + std::string first_response; + std::string second_response; + tr.expect(client->read(first_response, 3000, &ec), + with_error("first adjacent read failed: ", ec)); + tr.expect(client->read(second_response, 3000, &ec), + with_error("second adjacent read failed: ", ec)); + tr.expect(first_response == "Echo: " + first, + "first adjacent response should echo only first message"); + tr.expect(second_response == "Echo: " + second, + "second adjacent response should echo only second message"); + tr.expect(harness.wait_messages_at_least(2), "server did not receive two adjacent messages"); + + client->close(); + tr.expect(harness.wait_disconnected_at_least(1), "server did not observe client close"); + harness.stop(); + tr.expect(harness.error_count() == 0, with_error("server reported error: ", harness.error())); +} + void repeated_connect_and_open_contract(TestRunner& tr) { ServerHarness harness("repeat", 1024, 25); harness.start(); @@ -446,6 +511,8 @@ int main() { TestRunner runner; runner.run("basic_io_edges", basic_io_edges); + runner.run("large_messages_cross_transport_buffers", large_messages_cross_transport_buffers); + runner.run("adjacent_messages_are_not_merged", adjacent_messages_are_not_merged); runner.run("repeated_connect_and_open_contract", repeated_connect_and_open_contract); runner.run("server_disconnect_notifies_client", server_disconnect_notifies_client); runner.run("churn_clients", churn_clients); From e3810accb9bf5910ec881383876ab836d47c3829 Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Wed, 22 Jul 2026 13:00:33 +0300 Subject: [PATCH 3/3] test: define missing pipe error in ODR stub --- tests/stubs/windows.h | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/stubs/windows.h b/tests/stubs/windows.h index f9b485a..a4f2358 100644 --- a/tests/stubs/windows.h +++ b/tests/stubs/windows.h @@ -63,6 +63,7 @@ using LPOVERLAPPED = OVERLAPPED*; #define ERROR_OPERATION_ABORTED 995 #define ERROR_MORE_DATA 234 #define ERROR_BROKEN_PIPE 109 +#define ERROR_PIPE_NOT_CONNECTED 233 #define ERROR_SUCCESS 0 #define ERROR_NO_DATA 232 #define INFINITE 0xFFFFFFFF