From 0ae732ff8c64ca9beb307ebbf5205df1034035ff Mon Sep 17 00:00:00 2001 From: Segfault <5221072+Segfaultd@users.noreply.github.com> Date: Thu, 23 Jul 2026 12:45:46 +0200 Subject: [PATCH 1/6] feat: prototype recvmmsg/sendmmsg batched datagram I/O Add batched UDP send/receive behind default-off Linux build flags (MAFIANET_USE_RECVMMSG / MAFIANET_USE_SENDMMSG), plus a portable, unit-tested core. Portable core (MmsgBatch.h/.cpp), compiled and tested on every platform: - DriveBatchedSend: sendmmsg partial-send resume state machine. - SockaddrToSystemAddress: byte-order-correct sockaddr -> SystemAddress. - DispatchRecvBatch: fan a received batch to the event handler, freeing zero-length reads and the unused tail. Linux syscall wiring (guarded, inert by default): - RNS2_Berkley::RecvFromBatchedLoop drains the socket with one recvmmsg(MSG_WAITFORONE) per burst instead of one recvfrom per packet. - RNS2_Linux::SendBatch coalesces datagrams via sendmmsg; a loop-local RNS2SendBatch in ReliabilityLayer's resend loop defers only the transmit (encryption/simulation/metrics still run per datagram) and flushes at the loop's single exit. Tests: 11 hermetic unit tests for the core (Tests/Unit/MmsgBatchTests.cpp). The Linux path is compile-verified (gcc 13, both flags on); runtime validation on Linux via the integration suite is still pending. --- Source/CMakeLists.txt | 10 + Source/include/mafianet/MmsgBatch.h | 156 +++++++++++++ Source/include/mafianet/ReliabilityLayer.h | 8 +- Source/include/mafianet/socket2.h | 11 + Source/src/MmsgBatch.cpp | 67 ++++++ Source/src/RakNetSocket2.cpp | 65 +++++- Source/src/RakNetSocket2_Berkley.cpp | 64 ++++++ Source/src/ReliabilityLayer.cpp | 40 +++- Tests/Unit/MmsgBatchTests.cpp | 244 +++++++++++++++++++++ 9 files changed, 657 insertions(+), 8 deletions(-) create mode 100644 Source/include/mafianet/MmsgBatch.h create mode 100644 Source/src/MmsgBatch.cpp create mode 100644 Tests/Unit/MmsgBatchTests.cpp diff --git a/Source/CMakeLists.txt b/Source/CMakeLists.txt index 0913d5544..c3443b348 100644 --- a/Source/CMakeLists.txt +++ b/Source/CMakeLists.txt @@ -3,6 +3,12 @@ # Copyright (c) 2024, MafiaHub # Licensed under MIT-style license +# Batched datagram I/O (Linux). OFF by default; the portable helpers in +# MmsgBatch.h are always compiled and unit-tested, but the recvmmsg/sendmmsg +# syscall paths and their wiring only activate on Linux when these are ON. +option(MAFIANET_USE_RECVMMSG "Batch UDP receives with recvmmsg (Linux only)" OFF) +option(MAFIANET_USE_SENDMMSG "Batch UDP sends with sendmmsg (Linux only)" OFF) + # Source files (explicit list, no GLOB) set(MAFIANET_SOURCES src/_FindFirst.cpp @@ -47,6 +53,7 @@ set(MAFIANET_SOURCES src/LocklessTypes.cpp src/LogCommandParser.cpp src/MessageFilter.cpp + src/MmsgBatch.cpp src/NatPunchthroughClient.cpp src/NatPunchthroughServer.cpp src/NatTypeDetectionClient.cpp @@ -200,6 +207,7 @@ set(MAFIANET_HEADERS include/mafianet/memoryoverride.h include/mafianet/MessageFilter.h include/mafianet/MessageIdentifiers.h + include/mafianet/MmsgBatch.h include/mafianet/MTUSize.h include/mafianet/NativeFeatureIncludes.h include/mafianet/NativeFeatureIncludesOverrides.h @@ -337,6 +345,8 @@ function(mafianet_configure_target target_name) PRIVATE $<$:_DEBUG> $<$:NDEBUG> + $<$:MAFIANET_USE_RECVMMSG> + $<$:MAFIANET_USE_SENDMMSG> ) set_target_properties(${target_name} PROPERTIES diff --git a/Source/include/mafianet/MmsgBatch.h b/Source/include/mafianet/MmsgBatch.h new file mode 100644 index 000000000..6387ec19a --- /dev/null +++ b/Source/include/mafianet/MmsgBatch.h @@ -0,0 +1,156 @@ +/* + * Copyright (c) 2026, MafiaHub + * + * This source code is licensed under the MIT-style license found in the + * license.txt file in the root directory of this source tree. + */ + +/// \file MmsgBatch.h +/// \brief Portable helpers for batched datagram I/O (recvmmsg / sendmmsg). +/// +/// The Linux-only syscall glue lives in RakNetSocket2_Berkley.cpp behind the +/// MAFIANET_USE_RECVMMSG / MAFIANET_USE_SENDMMSG build flags. The logic that is +/// actually bug-prone -- the sendmmsg partial-send resume loop, the byte-order +/// handling when turning a raw sockaddr into a SystemAddress, and fanning a +/// received batch out to the event handler -- is factored out here so it is +/// portable and unit-testable on platforms that lack the mmsg syscalls (macOS, +/// Windows, the BSDs). + +#ifndef __MAFIANET_MMSG_BATCH_H +#define __MAFIANET_MMSG_BATCH_H + +#include "mafianet/socket2.h" + +#include // memcpy + +namespace MafiaNet +{ + +/// Maximum datagrams coalesced into a single recvmmsg/sendmmsg system call. +static const unsigned MMSG_BATCH_MAX = 64; + +/// Convert a raw sockaddr (IPv4 or IPv6) into a SystemAddress, reproducing the +/// byte-order handling of the scalar recvfrom path in RNS2_Berkley. \a out +/// receives the address; its port is stored in network order in the address +/// union and mirrored (host order) into debugPort. +void SockaddrToSystemAddress(const sockaddr_storage &from, SystemAddress *out); + +/// Drive a sendmmsg-style batched send to completion, handling partial sends. +/// +/// \a transmit is invoked as transmit(offset, count) to send the messages in +/// [offset, offset+count). It must return: +/// - a positive count of messages accepted this call (advance and continue), +/// - 0 to signal no further progress is possible right now (stop), or +/// - a negative errno-style value meaning the call failed with nothing sent. +/// +/// This mirrors sendmmsg(2), which returns the number of messages sent and only +/// reports an error (-1) when *no* datagram could be sent. Returns the total +/// number of messages sent (0..total). If the very first call fails before any +/// message is sent, the negative error code is propagated unchanged so the +/// caller can inspect errno exactly as it would for a scalar sendto. +template +int DriveBatchedSend(unsigned total, TransmitFn transmit) +{ + unsigned sent = 0; + while (sent < total) + { + int r = transmit(sent, total - sent); + if (r > 0) + { + sent += (unsigned) r; // accepted r messages; retry the remainder + continue; + } + if (r == 0) + break; // no progress possible right now; don't spin + // r < 0: this call failed with nothing sent. Propagate the error only + // if we have not managed to send anything yet (mirrors sendmmsg, which + // reports -1 solely when no datagram at all could be sent); otherwise + // report the progress already made. + if (sent == 0) + return r; + break; + } + return (int) sent; +} + +/// Fan a received batch out to the event handler. +/// +/// \a slots[0..allocated) are pre-allocated recv structs whose .data buffers +/// already hold the received bytes (the mmsghdr iovecs pointed at them). For +/// each i in [0,received) with lens[i] > 0 the struct is stamped with the byte +/// count, source address (from addrs[i]), timestamp \a now and \a socket, then +/// handed to handler->OnRNS2Recv. Every other struct -- short reads +/// (lens[i] <= 0) and the unused tail [received,allocated) -- is returned via +/// handler->DeallocRNS2RecvStruct so no buffer leaks. +void DispatchRecvBatch(RNS2EventHandler *handler, + RNS2RecvStruct **slots, unsigned allocated, + const int *lens, const sockaddr_storage *addrs, + unsigned received, RakNetSocket2 *socket, + MafiaNet::TimeUS now); + +#if defined(MAFIANET_USE_SENDMMSG) +/// Accumulates already-prepared datagrams (post-encryption, post packet-loss +/// simulation) destined for a single peer and flushes them via +/// RakNetSocket2::SendBatch -- one sendmmsg per burst on Linux. +/// +/// Bytes are copied in because the caller's serialization buffer is reused for +/// the next datagram. It is meant to be loop-local: construct it before a send +/// loop, Flush() at the loop's single exit (the destructor also flushes as a +/// backstop), so a datagram is never stranded across ticks. +class RNS2SendBatch +{ +public: + RNS2SendBatch(RakNetSocket2 *socket, const SystemAddress &dest) + : socket(socket), dest(dest), count(0) + { + // Heap-backed so a full batch (MMSG_BATCH_MAX * MTU) never sits on the + // network thread's stack. + buffers = new char[(size_t) MMSG_BATCH_MAX * MAXIMUM_MTU_SIZE]; + } + ~RNS2SendBatch() + { + Flush(); + delete[] buffers; + } + + void Add(const char *data, int length) + { + if (count == MMSG_BATCH_MAX) + Flush(); + if (length > MAXIMUM_MTU_SIZE) + length = MAXIMUM_MTU_SIZE; + memcpy(Slot(count), data, (size_t) length); + lengths[count] = length; + ++count; + } + + void Flush() + { + if (count == 0) + return; + RNS2_SendParameters sends[MMSG_BATCH_MAX]; + for (unsigned i = 0; i < count; ++i) + { + sends[i].data = Slot(i); + sends[i].length = lengths[i]; + sends[i].systemAddress = dest; + sends[i].ttl = 0; + } + socket->SendBatch(sends, count, __FILE__, __LINE__); + count = 0; + } + +private: + char *Slot(unsigned i) { return buffers + (size_t) i * MAXIMUM_MTU_SIZE; } + + RakNetSocket2 *socket; + SystemAddress dest; + unsigned count; + char *buffers; + int lengths[MMSG_BATCH_MAX]; +}; +#endif // MAFIANET_USE_SENDMMSG + +} // namespace MafiaNet + +#endif // __MAFIANET_MMSG_BATCH_H diff --git a/Source/include/mafianet/ReliabilityLayer.h b/Source/include/mafianet/ReliabilityLayer.h index 67979d049..e623270c2 100644 --- a/Source/include/mafianet/ReliabilityLayer.h +++ b/Source/include/mafianet/ReliabilityLayer.h @@ -63,6 +63,7 @@ namespace MafiaNet { /// Forward declarations class PluginInterface2; class RakNetRandom; +class RNS2SendBatch; // MmsgBatch.h; only defined when MAFIANET_USE_SENDMMSG is set typedef uint64_t reliabilityHeapWeightType; // #med - consider a more suitable name for the class / maybe even make an internal class to SplitPacketChannel? @@ -271,7 +272,12 @@ class ReliabilityLayer// /// \param[in] s The socket used for sending data /// \param[in] systemAddress The address and port to send to /// \param[in] bitStream The data to send. - void SendBitStream( RakNetSocket2 *s, SystemAddress &systemAddress, MafiaNet::BitStream *bitStream, RakNetRandom *rnr, CCTimeType currentTime); + /// \param[in] sendBatch If non-null, the fully-prepared datagram is appended + /// to this batch (coalesced into a sendmmsg) instead of being sent + /// immediately. All other processing (encryption, loss simulation, + /// metrics) still happens per datagram. Only used when + /// MAFIANET_USE_SENDMMSG is enabled. + void SendBitStream( RakNetSocket2 *s, SystemAddress &systemAddress, MafiaNet::BitStream *bitStream, RakNetRandom *rnr, CCTimeType currentTime, RNS2SendBatch *sendBatch = nullptr); ///Parse an internalPacket and create a bitstream to represent this data /// \return Returns number of bits used diff --git a/Source/include/mafianet/socket2.h b/Source/include/mafianet/socket2.h index 592afef50..71ade21bf 100644 --- a/Source/include/mafianet/socket2.h +++ b/Source/include/mafianet/socket2.h @@ -113,6 +113,10 @@ class RakNetSocket2 // In order for the handler to trigger, some platforms must call PollRecvFrom, some platforms this create an internal thread. void SetRecvEventHandler(RNS2EventHandler *_eventHandler); virtual RNS2SendResult Send( RNS2_SendParameters *sendParameters, const char *file, unsigned int line )=0; + // Batched send. The base implementation simply loops Send() so every socket + // type has a working default; RNS2_Linux overrides it with sendmmsg when + // MAFIANET_USE_SENDMMSG is enabled. Returns the total datagrams accepted. + virtual RNS2SendResult SendBatch( RNS2_SendParameters *sends, unsigned count, const char *file, unsigned int line ); RNS2Type GetSocketType(void) const; void SetSocketType(RNS2Type t); bool IsBerkleySocket(void) const; @@ -196,6 +200,10 @@ class RNS2_Berkley : public IRNS2_Berkley RNS2_BerkleyBindParameters binding; unsigned RecvFromLoopInt(void); + // Batched drain of the recv socket via recvmmsg (Linux + MAFIANET_USE_RECVMMSG). + // Declared unconditionally; only ever called from RecvFromLoopInt under the + // same guard, and only defined on that platform. + void RecvFromBatchedLoop(void); MafiaNet::LocklessUint32_t isRecvFromLoopThreadActive; volatile bool endThreads; // Constructor not called! @@ -256,6 +264,9 @@ class RNS2_Linux : public RNS2_Berkley, public RNS2_Windows_Linux_360 public: RNS2BindResult Bind( RNS2_BerkleyBindParameters *bindParameters, const char *file, unsigned int line ); RNS2SendResult Send( RNS2_SendParameters *sendParameters, const char *file, unsigned int line ); +#if defined(MAFIANET_USE_SENDMMSG) + RNS2SendResult SendBatch( RNS2_SendParameters *sends, unsigned count, const char *file, unsigned int line ); +#endif // ----------- STATICS ------------ static void GetMyIP( SystemAddress addresses[MAXIMUM_NUMBER_OF_INTERNAL_IDS] ); diff --git a/Source/src/MmsgBatch.cpp b/Source/src/MmsgBatch.cpp new file mode 100644 index 000000000..1be2cf1a5 --- /dev/null +++ b/Source/src/MmsgBatch.cpp @@ -0,0 +1,67 @@ +/* + * Copyright (c) 2026, MafiaHub + * + * This source code is licensed under the MIT-style license found in the + * license.txt file in the root directory of this source tree. + */ + +#include "mafianet/MmsgBatch.h" + +namespace MafiaNet +{ + +void SockaddrToSystemAddress(const sockaddr_storage &from, SystemAddress *out) +{ + // Mirrors the byte-order handling of RNS2_Berkley::RecvFromBlockingIPV4And6: + // the sockaddr keeps the port in network order (copied verbatim into the + // address union), and debugPort caches the host-order value. + if (from.ss_family == AF_INET) + { + memcpy(&out->address.addr4, + reinterpret_cast(&from), sizeof(sockaddr_in)); + out->debugPort = ntohs(out->address.addr4.sin_port); + } +#if RAKNET_SUPPORT_IPV6 == 1 + else + { + memcpy(&out->address.addr6, + reinterpret_cast(&from), sizeof(sockaddr_in6)); + out->debugPort = ntohs(out->address.addr6.sin6_port); + } +#endif +} + +void DispatchRecvBatch(RNS2EventHandler *handler, + RNS2RecvStruct **slots, unsigned allocated, + const int *lens, const sockaddr_storage *addrs, + unsigned received, RakNetSocket2 *socket, + MafiaNet::TimeUS now) +{ + if (received > allocated) + received = allocated; + + for (unsigned i = 0; i < received; ++i) + { + RNS2RecvStruct *s = slots[i]; + if (lens[i] > 0) + { + s->bytesRead = lens[i]; + s->timeRead = now; + s->socket = socket; + SockaddrToSystemAddress(addrs[i], &s->systemAddress); + handler->OnRNS2Recv(s); + } + else + { + // Zero-length read: same as the scalar bytesRead<=0 branch -- free + // the struct rather than surfacing an empty packet. + handler->DeallocRNS2RecvStruct(s, __FILE__, __LINE__); + } + } + + // Hand back the unused tail so no preallocated buffer leaks. + for (unsigned i = received; i < allocated; ++i) + handler->DeallocRNS2RecvStruct(slots[i], __FILE__, __LINE__); +} + +} // namespace MafiaNet diff --git a/Source/src/RakNetSocket2.cpp b/Source/src/RakNetSocket2.cpp index 6575a8d69..8cf9782f9 100644 --- a/Source/src/RakNetSocket2.cpp +++ b/Source/src/RakNetSocket2.cpp @@ -14,6 +14,7 @@ */ #include "mafianet/socket2.h" +#include "mafianet/MmsgBatch.h" #include "mafianet/memoryoverride.h" #include "mafianet/assert.h" #include "mafianet/sleep.h" @@ -54,6 +55,19 @@ void RakNetSocket2Allocator::DeallocRNS2(RakNetSocket2 *s) { MafiaNet::OP_DELETE RakNetSocket2::RakNetSocket2() {eventHandler=0;} RakNetSocket2::~RakNetSocket2() {} void RakNetSocket2::SetRecvEventHandler(RNS2EventHandler *_eventHandler) {eventHandler=_eventHandler;} +RNS2SendResult RakNetSocket2::SendBatch( RNS2_SendParameters *sends, unsigned count, const char *file, unsigned int line ) +{ + // Portable default: send each datagram individually. Platforms with a real + // batched syscall (Linux/sendmmsg) override this. + RNS2SendResult total=0; + for (unsigned i=0; i0) + total += r; + } + return total; +} RNS2Type RakNetSocket2::GetSocketType(void) const {return socketType;} void RakNetSocket2::SetSocketType(RNS2Type t) {socketType=t;} bool RakNetSocket2::IsBerkleySocket(void) const { @@ -154,7 +168,13 @@ RAK_THREAD_DECLARATION(RNS2_Berkley::RecvFromLoop) unsigned RNS2_Berkley::RecvFromLoopInt(void) { isRecvFromLoopThreadActive.Increment(); - + +#if defined(MAFIANET_USE_RECVMMSG) && defined(__linux__) + // Drain the socket in batches with a single recvmmsg per burst instead of + // one recvfrom per datagram. Falls through to the scalar loop below on any + // other platform / when the flag is off. + RecvFromBatchedLoop(); +#else while ( endThreads == false ) { RNS2RecvStruct *recvFromStruct; @@ -176,6 +196,7 @@ unsigned RNS2_Berkley::RecvFromLoopInt(void) } } } +#endif // MAFIANET_USE_RECVMMSG && __linux__ isRecvFromLoopThreadActive.Decrement(); return 0; @@ -263,5 +284,47 @@ SocketLayerOverride* RNS2_Windows::GetSocketLayerOverride(void) {return slo;} #else RNS2BindResult RNS2_Linux::Bind( RNS2_BerkleyBindParameters *bindParameters, const char *file, unsigned int line ) {return BindShared(bindParameters, file, line);} RNS2SendResult RNS2_Linux::Send( RNS2_SendParameters *sendParameters, const char *file, unsigned int line ) {return Send_Windows_Linux_360NoVDP(rns2Socket,sendParameters, file, line);} +#if defined(MAFIANET_USE_SENDMMSG) +RNS2SendResult RNS2_Linux::SendBatch( RNS2_SendParameters *sends, unsigned count, const char *file, unsigned int line ) +{ + (void) file; + (void) line; + + // DriveBatchedSend handles the partial-send resume; each transmit() call + // coalesces up to MMSG_BATCH_MAX datagrams into one sendmmsg. sendmmsg + // returns the number of messages sent (>=1) or -1 (nothing sent), which + // DriveBatchedSend interprets exactly as documented. + const RNS2SendResult sent = DriveBatchedSend(count, + [&](unsigned offset, unsigned remaining) -> int + { + unsigned chunk = remaining < MMSG_BATCH_MAX ? remaining : MMSG_BATCH_MAX; + struct mmsghdr msgs[MMSG_BATCH_MAX]; + struct iovec iovecs[MMSG_BATCH_MAX]; + for (unsigned i=0; idata; + iovecs[i].iov_len = (size_t) p->length; + memset(&msgs[i], 0, sizeof(msgs[i])); + msgs[i].msg_hdr.msg_iov = &iovecs[i]; + msgs[i].msg_hdr.msg_iovlen = 1; + if (p->systemAddress.address.addr4.sin_family==AF_INET) + { + msgs[i].msg_hdr.msg_name = (void*) &p->systemAddress.address.addr4; + msgs[i].msg_hdr.msg_namelen = sizeof(sockaddr_in); + } + else + { +#if RAKNET_SUPPORT_IPV6==1 + msgs[i].msg_hdr.msg_name = (void*) &p->systemAddress.address.addr6; + msgs[i].msg_hdr.msg_namelen = sizeof(sockaddr_in6); +#endif + } + } + return sendmmsg(rns2Socket, msgs, chunk, 0); + }); + return sent; +} +#endif void RNS2_Linux::GetMyIP( SystemAddress addresses[MAXIMUM_NUMBER_OF_INTERNAL_IDS] ) {return GetMyIP_Windows_Linux(addresses);} #endif // Linux diff --git a/Source/src/RakNetSocket2_Berkley.cpp b/Source/src/RakNetSocket2_Berkley.cpp index eeddc302b..a4e4e4c4c 100644 --- a/Source/src/RakNetSocket2_Berkley.cpp +++ b/Source/src/RakNetSocket2_Berkley.cpp @@ -528,6 +528,70 @@ void RNS2_Berkley::RecvFromBlocking(RNS2RecvStruct *recvFromStruct) #endif } +#if defined(MAFIANET_USE_RECVMMSG) && defined(__linux__) +void RNS2_Berkley::RecvFromBatchedLoop(void) +{ + RNS2RecvStruct *slots[MMSG_BATCH_MAX]; + struct mmsghdr msgs[MMSG_BATCH_MAX]; + struct iovec iovecs[MMSG_BATCH_MAX]; + sockaddr_storage addrs[MMSG_BATCH_MAX]; + int lens[MMSG_BATCH_MAX]; + + while ( endThreads == false ) + { + // Preallocate a batch of recv structs and point each iovec at its data + // buffer, so recvmmsg writes straight into the structs we will hand to + // the event handler -- no extra copy. + unsigned allocated=0; + for (; allocatedAllocRNS2RecvStruct(_FILE_AND_LINE_); + if (s==nullptr) + break; + s->socket=this; + slots[allocated]=s; + iovecs[allocated].iov_base=s->data; + iovecs[allocated].iov_len=sizeof(s->data); + memset(&msgs[allocated], 0, sizeof(msgs[allocated])); + msgs[allocated].msg_hdr.msg_iov=&iovecs[allocated]; + msgs[allocated].msg_hdr.msg_iovlen=1; + msgs[allocated].msg_hdr.msg_name=&addrs[allocated]; + msgs[allocated].msg_hdr.msg_namelen=sizeof(addrs[allocated]); + } + if (allocated==0) + { + RakSleep(0); + continue; + } + + // MSG_WAITFORONE: block until at least one datagram is ready, then return + // everything already queued. Without it recvmmsg would wait to fill the + // whole array (or hit a timeout), adding latency at low packet rates. + int n = recvmmsg(rns2Socket, msgs, allocated, MSG_WAITFORONE, nullptr); + MafiaNet::TimeUS now = MafiaNet::GetTimeUS(); + + if (n<0) + { + // Interrupted / error (includes the shutdown poke unblocking us). + // Release the whole batch and re-evaluate endThreads. + for (unsigned i=0; iDeallocRNS2RecvStruct(slots[i], _FILE_AND_LINE_); + RakSleep(0); + continue; + } + + for (unsigned i=0; i0) @@ -2275,7 +2281,11 @@ void ReliabilityLayer::UpdateInternal( RakNetSocket2 *s, SystemAddress &systemAd congestionManager.OnSendBytes(time,UDP_HEADER_SIZE+DatagramHeaderFormat::GetDataHeaderByteLength()); +#if defined(MAFIANET_USE_SENDMMSG) + SendBitStream( s, systemAddress, &updateBitStream, rnr, time, &sendBatch ); +#else SendBitStream( s, systemAddress, &updateBitStream, rnr, time ); +#endif bandwidthExceededStatistic=outgoingPacketBuffer.Size()>0; // bandwidthExceededStatistic=sendPacketSet[0].IsEmpty()==false || @@ -2291,6 +2301,11 @@ void ReliabilityLayer::UpdateInternal( RakNetSocket2 *s, SystemAddress &systemAd timeOfLastContinualSend=0; } +#if defined(MAFIANET_USE_SENDMMSG) + // Single flush point for the whole update's datagrams. + sendBatch.Flush(); +#endif + ClearPacketsAndDatagrams(); // Any data waiting to send after attempting to send, then bandwidth is exceeded @@ -2309,10 +2324,11 @@ void ReliabilityLayer::UpdateInternal( RakNetSocket2 *s, SystemAddress &systemAd //------------------------------------------------------------------------------------------------------- // Writes a bitstream to the socket //------------------------------------------------------------------------------------------------------- -void ReliabilityLayer::SendBitStream( RakNetSocket2 *s, SystemAddress &systemAddress, MafiaNet::BitStream *bitStream, RakNetRandom *rnr, CCTimeType currentTime) +void ReliabilityLayer::SendBitStream( RakNetSocket2 *s, SystemAddress &systemAddress, MafiaNet::BitStream *bitStream, RakNetRandom *rnr, CCTimeType currentTime, RNS2SendBatch *sendBatch) { (void) systemAddress; (void) rnr; + (void) sendBatch; unsigned int length; @@ -2395,11 +2411,23 @@ void ReliabilityLayer::SendBitStream( RakNetSocket2 *s, SystemAddress &systemAdd #else // SocketLayer::SendTo( s, ( char* ) bitStream->GetData(), length, systemAddress, __FILE__, __LINE__ ); - RNS2_SendParameters bsp; - bsp.data = (char*) bitStream->GetData(); - bsp.length = length; - bsp.systemAddress = systemAddress; - s->Send(&bsp, _FILE_AND_LINE_); +#if defined(MAFIANET_USE_SENDMMSG) + if (sendBatch) + { + // Defer only the transmit: the datagram above is fully prepared + // (encrypted, metered). The batch flushes via sendmmsg at the caller's + // loop boundary. + sendBatch->Add((const char*) bitStream->GetData(), (int) length); + } + else +#endif + { + RNS2_SendParameters bsp; + bsp.data = (char*) bitStream->GetData(); + bsp.length = length; + bsp.systemAddress = systemAddress; + s->Send(&bsp, _FILE_AND_LINE_); + } #endif } diff --git a/Tests/Unit/MmsgBatchTests.cpp b/Tests/Unit/MmsgBatchTests.cpp new file mode 100644 index 000000000..904f6be21 --- /dev/null +++ b/Tests/Unit/MmsgBatchTests.cpp @@ -0,0 +1,244 @@ +/* + * Copyright (c) 2026, MafiaHub + * + * This source code is licensed under the MIT-style license found in the + * license.txt file in the root directory of this source tree. + * + * Hermetic unit tests for the portable batched-datagram helpers in + * MmsgBatch.h. These exercise the logic that is genuinely bug-prone in a + * recvmmsg/sendmmsg integration -- the sendmmsg partial-send resume loop, the + * byte-order handling when decoding a raw sockaddr, and fanning a received + * batch out to the event handler -- without touching any actual syscall, so + * they run everywhere including platforms that lack recvmmsg/sendmmsg. + */ + +#include + +#include "mafianet/MmsgBatch.h" +#include "mafianet/socket2.h" +#include "mafianet/types.h" + +#include +#include +#include + +using namespace MafiaNet; + +namespace +{ + sockaddr_storage MakeV4(const char *ip, unsigned short port) + { + sockaddr_storage ss; + memset(&ss, 0, sizeof(ss)); + sockaddr_in *in = reinterpret_cast(&ss); + in->sin_family = AF_INET; + in->sin_port = htons(port); + inet_pton(AF_INET, ip, &in->sin_addr); + return ss; + } + + // Records everything DispatchRecvBatch does so tests can assert on it. No + // allocation happens here -- the tests own the RNS2RecvStruct storage. + struct RecordingHandler : RNS2EventHandler + { + struct Received + { + int bytesRead; + unsigned short port; + RakNetSocket2 *socket; + MafiaNet::TimeUS timeRead; + }; + + std::vector received; + std::vector deallocated; + + void OnRNS2Recv(RNS2RecvStruct *s) override + { + received.push_back({s->bytesRead, s->systemAddress.GetPort(), + s->socket, s->timeRead}); + } + void DeallocRNS2RecvStruct(RNS2RecvStruct *s, const char *, unsigned int) override + { + deallocated.push_back(s); + } + RNS2RecvStruct *AllocRNS2RecvStruct(const char *, unsigned int) override + { + return nullptr; // unused by DispatchRecvBatch + } + }; + + // A sentinel non-null socket pointer; DispatchRecvBatch only stores it. + RakNetSocket2 *const kSentinelSocket = reinterpret_cast(0xF00D); +} + +// --------------------------------------------------------------------------- +// DriveBatchedSend -- the sendmmsg partial-send resume state machine. +// --------------------------------------------------------------------------- + +TEST(DriveBatchedSend, SendsWholeBatchInOneCall) +{ + int calls = 0; + int sent = DriveBatchedSend(5, [&](unsigned offset, unsigned count) { + ++calls; + EXPECT_EQ(offset, 0u); + EXPECT_EQ(count, 5u); + return (int) count; + }); + EXPECT_EQ(sent, 5); + EXPECT_EQ(calls, 1); +} + +TEST(DriveBatchedSend, ResumesFromOffsetAcrossPartialSends) +{ + // sendmmsg returns short counts; the loop must retry the remainder from the + // advancing offset until the whole batch is sent. + std::vector offsets; + int sent = DriveBatchedSend(7, [&](unsigned offset, unsigned count) { + offsets.push_back(offset); + return (int) std::min(count, 3u); // accept at most 3 per call + }); + EXPECT_EQ(sent, 7); + EXPECT_EQ(offsets, (std::vector{0u, 3u, 6u})); +} + +TEST(DriveBatchedSend, ReturnsErrorWhenFirstCallSendsNothing) +{ + // sendmmsg only reports -1 when NO datagram could be sent; propagate it so + // the caller sees errno exactly as for a scalar sendto. + int sent = DriveBatchedSend(4, [&](unsigned, unsigned) { return -1; }); + EXPECT_EQ(sent, -1); +} + +TEST(DriveBatchedSend, ReturnsProgressWhenErrorFollowsPartialSend) +{ + // First call sends 4, second fails: we already made progress, so report the + // 4 that went out rather than the error. + int call = 0; + int sent = DriveBatchedSend(9, [&](unsigned, unsigned) { + return (call++ == 0) ? 4 : -1; + }); + EXPECT_EQ(sent, 4); +} + +TEST(DriveBatchedSend, StopsOnNoProgressWithoutLooping) +{ + // A zero return means "no progress possible right now" -- stop instead of + // spinning forever. + int calls = 0; + int sent = DriveBatchedSend(6, [&](unsigned, unsigned) { + ++calls; + return 0; + }); + EXPECT_EQ(sent, 0); + EXPECT_EQ(calls, 1); +} + +TEST(DriveBatchedSend, EmptyBatchNeverCallsTransmit) +{ + int calls = 0; + int sent = DriveBatchedSend(0, [&](unsigned, unsigned) { + ++calls; + return 0; + }); + EXPECT_EQ(sent, 0); + EXPECT_EQ(calls, 0); +} + +// --------------------------------------------------------------------------- +// SockaddrToSystemAddress -- byte-order handling. +// --------------------------------------------------------------------------- + +TEST(SockaddrToSystemAddress, DecodesIPv4PortToHostOrder) +{ + sockaddr_storage ss = MakeV4("192.0.2.1", 4660); + SystemAddress out; + SockaddrToSystemAddress(ss, &out); + EXPECT_EQ(out.GetPort(), 4660); + EXPECT_EQ(out.debugPort, 4660); + EXPECT_EQ(out.GetIPVersion(), 4); +} + +TEST(SockaddrToSystemAddress, PreservesIPv4Address) +{ + sockaddr_storage ss = MakeV4("203.0.113.7", 1234); + const sockaddr_in *in = reinterpret_cast(&ss); + SystemAddress out; + SockaddrToSystemAddress(ss, &out); + EXPECT_EQ(out.address.addr4.sin_addr.s_addr, in->sin_addr.s_addr); +} + +// --------------------------------------------------------------------------- +// DispatchRecvBatch -- fanning a received batch out to the handler. +// --------------------------------------------------------------------------- + +TEST(DispatchRecvBatch, DispatchesEachReceivedDatagramAndFreesTheTail) +{ + const unsigned allocated = 4; + std::vector storage(allocated); + RNS2RecvStruct *slots[allocated]; + for (unsigned i = 0; i < allocated; ++i) + slots[i] = &storage[i]; + + int lens[allocated] = {11, 22, 0, 0}; + sockaddr_storage addrs[allocated] = { + MakeV4("10.0.0.1", 1111), MakeV4("10.0.0.2", 2222), {}, {}}; + + RecordingHandler handler; + DispatchRecvBatch(&handler, slots, allocated, lens, addrs, + /*received=*/2, kSentinelSocket, /*now=*/12345); + + ASSERT_EQ(handler.received.size(), 2u); + EXPECT_EQ(handler.received[0].bytesRead, 11); + EXPECT_EQ(handler.received[0].port, 1111); + EXPECT_EQ(handler.received[0].socket, kSentinelSocket); + EXPECT_EQ(handler.received[0].timeRead, 12345u); + EXPECT_EQ(handler.received[1].bytesRead, 22); + EXPECT_EQ(handler.received[1].port, 2222); + + // The two unused tail slots must be handed back, not leaked. + EXPECT_EQ(handler.deallocated.size(), 2u); +} + +TEST(DispatchRecvBatch, FreesZeroLengthDatagramsInsteadOfDispatching) +{ + // A zero-length read mirrors the scalar path's bytesRead<=0 branch: free it, + // don't surface it as a packet. + const unsigned allocated = 3; + std::vector storage(allocated); + RNS2RecvStruct *slots[allocated]; + for (unsigned i = 0; i < allocated; ++i) + slots[i] = &storage[i]; + + int lens[allocated] = {0, 22, 0}; + sockaddr_storage addrs[allocated] = { + MakeV4("10.0.0.1", 1111), MakeV4("10.0.0.2", 2222), {}}; + + RecordingHandler handler; + DispatchRecvBatch(&handler, slots, allocated, lens, addrs, + /*received=*/2, kSentinelSocket, /*now=*/1); + + ASSERT_EQ(handler.received.size(), 1u); + EXPECT_EQ(handler.received[0].bytesRead, 22); + EXPECT_EQ(handler.received[0].port, 2222); + // Freed: the zero-length slot 0 and the unused tail slot 2. + EXPECT_EQ(handler.deallocated.size(), 2u); +} + +TEST(DispatchRecvBatch, FreesEverythingWhenNothingReceived) +{ + const unsigned allocated = 3; + std::vector storage(allocated); + RNS2RecvStruct *slots[allocated]; + for (unsigned i = 0; i < allocated; ++i) + slots[i] = &storage[i]; + + int lens[allocated] = {0, 0, 0}; + sockaddr_storage addrs[allocated] = {{}, {}, {}}; + + RecordingHandler handler; + DispatchRecvBatch(&handler, slots, allocated, lens, addrs, + /*received=*/0, kSentinelSocket, /*now=*/1); + + EXPECT_TRUE(handler.received.empty()); + EXPECT_EQ(handler.deallocated.size(), 3u); +} From 6afccc83c0f759a3fde1ff9eb160de800c481cb6 Mon Sep 17 00:00:00 2001 From: Segfault <5221072+Segfaultd@users.noreply.github.com> Date: Thu, 23 Jul 2026 13:10:00 +0200 Subject: [PATCH 2/6] fix(mmsg): portable test include, no truncation/heap churn, PUBLIC flags Address CI + review feedback on the batched I/O prototype: - Windows build: MmsgBatchTests.cpp used ; guard the include ( on Windows) so the hermetic tests build cross-platform. - RNS2SendBatch::Add rejected oversized datagrams instead of silently clamping to the MTU, which diverged from the scalar send path and could ship truncated payloads in release builds. - RNS2SendBatch no longer heap-allocates ~95 KB per UpdateInternal call; it reuses a thread_local buffer (single update thread per RakPeer, flushed synchronously) so the send path stays allocation-free. - MAFIANET_USE_RECVMMSG/SENDMMSG are now PUBLIC compile definitions so consumers see the same conditional declarations the library was built with (no header/ABI mismatch). --- Source/CMakeLists.txt | 5 ++++ Source/include/mafianet/MmsgBatch.h | 37 ++++++++++++++++------------- Tests/Unit/MmsgBatchTests.cpp | 6 ++++- 3 files changed, 31 insertions(+), 17 deletions(-) diff --git a/Source/CMakeLists.txt b/Source/CMakeLists.txt index c3443b348..3f6fbe647 100644 --- a/Source/CMakeLists.txt +++ b/Source/CMakeLists.txt @@ -345,6 +345,11 @@ function(mafianet_configure_target target_name) PRIVATE $<$:_DEBUG> $<$:NDEBUG> + # PUBLIC so consumers including MmsgBatch.h / socket2.h see the same + # conditional declarations (RNS2SendBatch, the SendBatch override) the + # library was compiled with -- avoids a header/ABI mismatch when a + # downstream build turns batching on. + PUBLIC $<$:MAFIANET_USE_RECVMMSG> $<$:MAFIANET_USE_SENDMMSG> ) diff --git a/Source/include/mafianet/MmsgBatch.h b/Source/include/mafianet/MmsgBatch.h index 6387ec19a..534a240d4 100644 --- a/Source/include/mafianet/MmsgBatch.h +++ b/Source/include/mafianet/MmsgBatch.h @@ -20,6 +20,7 @@ #define __MAFIANET_MMSG_BATCH_H #include "mafianet/socket2.h" +#include "mafianet/assert.h" #include // memcpy @@ -101,24 +102,21 @@ class RNS2SendBatch { public: RNS2SendBatch(RakNetSocket2 *socket, const SystemAddress &dest) - : socket(socket), dest(dest), count(0) - { - // Heap-backed so a full batch (MMSG_BATCH_MAX * MTU) never sits on the - // network thread's stack. - buffers = new char[(size_t) MMSG_BATCH_MAX * MAXIMUM_MTU_SIZE]; - } - ~RNS2SendBatch() - { - Flush(); - delete[] buffers; - } - + : socket(socket), dest(dest), count(0) {} + ~RNS2SendBatch() { Flush(); } + + /// Append one already-prepared datagram. Oversized datagrams are rejected + /// rather than truncated: the scalar send path never truncates, and silently + /// clamping would ship corrupted bytes with no error signal. Callers already + /// keep datagrams within the MTU (RakAssert in ReliabilityLayer), so this is + /// a release-build backstop, not the normal path. void Add(const char *data, int length) { + RakAssert(length >= 0 && length <= MAXIMUM_MTU_SIZE); + if (length < 0 || length > MAXIMUM_MTU_SIZE) + return; // drop, don't corrupt; the reliability layer will resend if (count == MMSG_BATCH_MAX) Flush(); - if (length > MAXIMUM_MTU_SIZE) - length = MAXIMUM_MTU_SIZE; memcpy(Slot(count), data, (size_t) length); lengths[count] = length; ++count; @@ -141,12 +139,19 @@ class RNS2SendBatch } private: - char *Slot(unsigned i) { return buffers + (size_t) i * MAXIMUM_MTU_SIZE; } + // One buffer block per update thread, reused across every connection and + // tick -- the batch is filled and flushed synchronously within a single + // UpdateInternal call, so there is no reentrancy. This keeps the send path + // allocation-free (the old per-tick new[] churned ~95 KB each call). + static char *Slot(unsigned i) + { + static thread_local char buffers[MMSG_BATCH_MAX][MAXIMUM_MTU_SIZE]; + return buffers[i]; + } RakNetSocket2 *socket; SystemAddress dest; unsigned count; - char *buffers; int lengths[MMSG_BATCH_MAX]; }; #endif // MAFIANET_USE_SENDMMSG diff --git a/Tests/Unit/MmsgBatchTests.cpp b/Tests/Unit/MmsgBatchTests.cpp index 904f6be21..c5c335778 100644 --- a/Tests/Unit/MmsgBatchTests.cpp +++ b/Tests/Unit/MmsgBatchTests.cpp @@ -18,7 +18,11 @@ #include "mafianet/socket2.h" #include "mafianet/types.h" -#include +#ifdef _WIN32 +#include // inet_pton, htons (winsock2 pulled in via socket2.h) +#else +#include // inet_pton, htons +#endif #include #include From 89fed16082ece73e665debe8abf885cce0ae5401 Mon Sep 17 00:00:00 2001 From: Segfault <5221072+Segfaultd@users.noreply.github.com> Date: Sun, 26 Jul 2026 18:44:34 +0200 Subject: [PATCH 3/6] fix(mmsg): unify SendBatch contract, guard recv address decode, cover in CI Pre-landing review follow-ups on the batched datagram I/O prototype: - SendBatch returned a byte total from the portable base implementation but a datagram count from the sendmmsg override. Both now return a datagram count (and the error only when nothing was sent) by routing the scalar loop through the same DriveBatchedSend state machine. - SockaddrToSystemAddress silently left *out untouched for an address family the build cannot decode, so a recycled recv struct would attribute the datagram to its previous sender. It now returns bool, sets UNASSIGNED_SYSTEM_ADDRESS, and DispatchRecvBatch frees instead of dispatching. - RNS2SendBatch is non-copyable and asserts against a second live batch on the same thread; its shared buffer block moved from TLS to a lazily allocated per-thread block, so threads that never batch pay nothing. - The batched recv loop kept reallocating the whole slot batch on every recvmmsg failure. Slots are now held across failures with progressive back-off, so a persistent async error (ECONNREFUSED on a closed peer port) cannot spin the polling thread. - RNS2SendBatch is compiled unconditionally (it only needs the portable SendBatch), with unit tests for payload copying, batch-boundary flush, destructor backstop, and the oversized-datagram drop. - New linux-mmsg CI job builds and tests with both flags ON; the syscall paths and their reliability-layer wiring were previously never compiled anywhere. Verified: 98/98 (unit + integration) on Linux with MAFIANET_USE_RECVMMSG and MAFIANET_USE_SENDMMSG ON; 69/69 unit on macOS with both OFF. --- .github/workflows/build.yml | 41 ++++ Source/CMakeLists.txt | 7 +- Source/include/mafianet/MmsgBatch.h | 82 ++++++-- Source/include/mafianet/ReliabilityLayer.h | 2 +- Source/include/mafianet/socket2.h | 6 +- Source/src/MmsgBatch.cpp | 21 +- Source/src/RakNetSocket2.cpp | 22 +- Source/src/RakNetSocket2_Berkley.cpp | 54 +++-- Tests/Unit/MmsgBatchTests.cpp | 222 +++++++++++++++++++++ 9 files changed, 402 insertions(+), 55 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 2592f246d..9d60a894b 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -34,6 +34,47 @@ jobs: path: test-results/junit.xml if-no-files-found: ignore + # Linux with batched datagram I/O turned on. The recvmmsg/sendmmsg paths and + # their reliability-layer wiring are compiled out of every other job, so + # without this they would never even be built. Runs both suites so the + # batched send path is exercised over real loopback traffic. + linux-mmsg: + runs-on: ubuntu-latest + steps: + - name: Checkout repository + uses: actions/checkout@v4 + with: + persist-credentials: false + + - name: Install OpenSSL + run: sudo apt-get update && sudo apt-get install -y libssl-dev + + - name: Configure CMake + run: | + cmake -B build \ + -DMAFIANET_BUILD_TESTS=ON \ + -DMAFIANET_USE_RECVMMSG=ON \ + -DMAFIANET_USE_SENDMMSG=ON + + - name: Build + run: cmake --build build --config Release --parallel 4 + + - name: Run tests + run: | + ctest --test-dir build \ + --output-on-failure \ + --repeat until-pass:3 \ + --timeout 600 \ + --output-junit junit.xml + + - name: Upload test results + if: always() + uses: actions/upload-artifact@v4 + with: + name: test-results-linux-mmsg + path: build/junit.xml + if-no-files-found: ignore + # macOS: Build and run both suites (unit + integration). macos: runs-on: macos-latest diff --git a/Source/CMakeLists.txt b/Source/CMakeLists.txt index 3f6fbe647..799ff14af 100644 --- a/Source/CMakeLists.txt +++ b/Source/CMakeLists.txt @@ -3,9 +3,10 @@ # Copyright (c) 2024, MafiaHub # Licensed under MIT-style license -# Batched datagram I/O (Linux). OFF by default; the portable helpers in -# MmsgBatch.h are always compiled and unit-tested, but the recvmmsg/sendmmsg -# syscall paths and their wiring only activate on Linux when these are ON. +# Batched datagram I/O (Linux). OFF by default; everything in MmsgBatch.h -- +# the helpers and the RNS2SendBatch accumulator alike -- is always compiled and +# unit-tested, but the recvmmsg/sendmmsg syscall paths and their wiring into the +# reliability layer only activate on Linux when these are ON. option(MAFIANET_USE_RECVMMSG "Batch UDP receives with recvmmsg (Linux only)" OFF) option(MAFIANET_USE_SENDMMSG "Batch UDP sends with sendmmsg (Linux only)" OFF) diff --git a/Source/include/mafianet/MmsgBatch.h b/Source/include/mafianet/MmsgBatch.h index 534a240d4..fcfcb54e3 100644 --- a/Source/include/mafianet/MmsgBatch.h +++ b/Source/include/mafianet/MmsgBatch.h @@ -9,18 +9,18 @@ /// \brief Portable helpers for batched datagram I/O (recvmmsg / sendmmsg). /// /// The Linux-only syscall glue lives in RakNetSocket2_Berkley.cpp behind the -/// MAFIANET_USE_RECVMMSG / MAFIANET_USE_SENDMMSG build flags. The logic that is -/// actually bug-prone -- the sendmmsg partial-send resume loop, the byte-order -/// handling when turning a raw sockaddr into a SystemAddress, and fanning a -/// received batch out to the event handler -- is factored out here so it is -/// portable and unit-testable on platforms that lack the mmsg syscalls (macOS, -/// Windows, the BSDs). +/// MAFIANET_USE_RECVMMSG / MAFIANET_USE_SENDMMSG build flags. Everything in this +/// header is portable and compiled unconditionally -- including RNS2SendBatch, +/// which only needs RakNetSocket2::SendBatch (whose base implementation works on +/// every platform) -- so the logic that is actually bug-prone is unit-testable +/// even where the mmsg syscalls do not exist (macOS, Windows, the BSDs). #ifndef __MAFIANET_MMSG_BATCH_H #define __MAFIANET_MMSG_BATCH_H #include "mafianet/socket2.h" #include "mafianet/assert.h" +#include "mafianet/memoryoverride.h" #include // memcpy @@ -30,11 +30,22 @@ namespace MafiaNet /// Maximum datagrams coalesced into a single recvmmsg/sendmmsg system call. static const unsigned MMSG_BATCH_MAX = 64; +/// Upper bound (ms) of the progressive back-off the batched recv loop applies +/// after consecutive recvmmsg failures, so a persistent asynchronous socket +/// error cannot spin the polling thread. +static const unsigned MMSG_ERROR_BACKOFF_MAX_MS = 10; + /// Convert a raw sockaddr (IPv4 or IPv6) into a SystemAddress, reproducing the /// byte-order handling of the scalar recvfrom path in RNS2_Berkley. \a out /// receives the address; its port is stored in network order in the address /// union and mirrored (host order) into debugPort. -void SockaddrToSystemAddress(const sockaddr_storage &from, SystemAddress *out); +/// +/// Returns false for an address family this build cannot represent (anything +/// other than AF_INET when RAKNET_SUPPORT_IPV6 is off, or AF_UNSPEC), in which +/// case \a out is set to UNASSIGNED_SYSTEM_ADDRESS rather than left holding +/// whatever the recycled recv struct contained. Callers must not dispatch a +/// datagram whose address failed to decode. +bool SockaddrToSystemAddress(const sockaddr_storage &from, SystemAddress *out); /// Drive a sendmmsg-style batched send to completion, handling partial sends. /// @@ -81,7 +92,8 @@ int DriveBatchedSend(unsigned total, TransmitFn transmit) /// each i in [0,received) with lens[i] > 0 the struct is stamped with the byte /// count, source address (from addrs[i]), timestamp \a now and \a socket, then /// handed to handler->OnRNS2Recv. Every other struct -- short reads -/// (lens[i] <= 0) and the unused tail [received,allocated) -- is returned via +/// (lens[i] <= 0), datagrams whose source address failed to decode, and the +/// unused tail [received,allocated) -- is returned via /// handler->DeallocRNS2RecvStruct so no buffer leaks. void DispatchRecvBatch(RNS2EventHandler *handler, RNS2RecvStruct **slots, unsigned allocated, @@ -89,10 +101,10 @@ void DispatchRecvBatch(RNS2EventHandler *handler, unsigned received, RakNetSocket2 *socket, MafiaNet::TimeUS now); -#if defined(MAFIANET_USE_SENDMMSG) /// Accumulates already-prepared datagrams (post-encryption, post packet-loss /// simulation) destined for a single peer and flushes them via -/// RakNetSocket2::SendBatch -- one sendmmsg per burst on Linux. +/// RakNetSocket2::SendBatch -- one sendmmsg per burst on Linux, a plain Send() +/// loop everywhere else. /// /// Bytes are copied in because the caller's serialization buffer is reused for /// the next datagram. It is meant to be loop-local: construct it before a send @@ -102,8 +114,25 @@ class RNS2SendBatch { public: RNS2SendBatch(RakNetSocket2 *socket, const SystemAddress &dest) - : socket(socket), dest(dest), count(0) {} - ~RNS2SendBatch() { Flush(); } + : socket(socket), dest(dest), count(0) + { + // The datagram buffers are a single block shared by every batch on this + // thread (see Slot), so two live batches would silently overwrite each + // other's payloads. The class is loop-local by construction; this catches + // a future caller that nests one inside another. + RakAssert(ThreadBatchLive() == false && "RNS2SendBatch is not reentrant"); + ThreadBatchLive() = true; + } + ~RNS2SendBatch() + { + Flush(); + ThreadBatchLive() = false; + } + + // Non-copyable, non-movable: a copy would alias the same shared buffers and + // flush the same datagrams a second time from its destructor. + RNS2SendBatch(const RNS2SendBatch &) = delete; + RNS2SendBatch &operator=(const RNS2SendBatch &) = delete; /// Append one already-prepared datagram. Oversized datagrams are rejected /// rather than truncated: the scalar send path never truncates, and silently @@ -139,14 +168,30 @@ class RNS2SendBatch } private: - // One buffer block per update thread, reused across every connection and - // tick -- the batch is filled and flushed synchronously within a single - // UpdateInternal call, so there is no reentrancy. This keeps the send path - // allocation-free (the old per-tick new[] churned ~95 KB each call). + // One buffer block per thread that actually batches, reused across every + // connection and tick -- the batch is filled and flushed synchronously + // within a single UpdateInternal call, so there is no reentrancy. This keeps + // the send path allocation-free (the old per-tick new[] churned ~95 KB each + // call) while keeping the block off the stack and out of TLS for the many + // threads that never send a batch: it is allocated on this thread's first + // Add() and released when the thread exits. static char *Slot(unsigned i) { - static thread_local char buffers[MMSG_BATCH_MAX][MAXIMUM_MTU_SIZE]; - return buffers[i]; + struct Block + { + char *bytes; + Block() : bytes(MafiaNet::OP_NEW_ARRAY(MMSG_BATCH_MAX * MAXIMUM_MTU_SIZE, _FILE_AND_LINE_)) {} + ~Block() { MafiaNet::OP_DELETE_ARRAY(bytes, _FILE_AND_LINE_); } + }; + static thread_local Block block; + return block.bytes + (size_t) i * MAXIMUM_MTU_SIZE; + } + + // Debug-only guard against a second live batch on the same thread. + static bool &ThreadBatchLive() + { + static thread_local bool live = false; + return live; } RakNetSocket2 *socket; @@ -154,7 +199,6 @@ class RNS2SendBatch unsigned count; int lengths[MMSG_BATCH_MAX]; }; -#endif // MAFIANET_USE_SENDMMSG } // namespace MafiaNet diff --git a/Source/include/mafianet/ReliabilityLayer.h b/Source/include/mafianet/ReliabilityLayer.h index e623270c2..0ec4bd1f3 100644 --- a/Source/include/mafianet/ReliabilityLayer.h +++ b/Source/include/mafianet/ReliabilityLayer.h @@ -63,7 +63,7 @@ namespace MafiaNet { /// Forward declarations class PluginInterface2; class RakNetRandom; -class RNS2SendBatch; // MmsgBatch.h; only defined when MAFIANET_USE_SENDMMSG is set +class RNS2SendBatch; // MmsgBatch.h typedef uint64_t reliabilityHeapWeightType; // #med - consider a more suitable name for the class / maybe even make an internal class to SplitPacketChannel? diff --git a/Source/include/mafianet/socket2.h b/Source/include/mafianet/socket2.h index 71ade21bf..0e7a2b884 100644 --- a/Source/include/mafianet/socket2.h +++ b/Source/include/mafianet/socket2.h @@ -115,7 +115,11 @@ class RakNetSocket2 virtual RNS2SendResult Send( RNS2_SendParameters *sendParameters, const char *file, unsigned int line )=0; // Batched send. The base implementation simply loops Send() so every socket // type has a working default; RNS2_Linux overrides it with sendmmsg when - // MAFIANET_USE_SENDMMSG is enabled. Returns the total datagrams accepted. + // MAFIANET_USE_SENDMMSG is enabled. + // Returns the number of datagrams accepted (0..count) -- a count, NOT the + // byte total Send() returns -- or a negative error code if the very first + // datagram failed, mirroring sendmmsg(2). Both implementations must agree on + // this; see MmsgBatchTests. virtual RNS2SendResult SendBatch( RNS2_SendParameters *sends, unsigned count, const char *file, unsigned int line ); RNS2Type GetSocketType(void) const; void SetSocketType(RNS2Type t); diff --git a/Source/src/MmsgBatch.cpp b/Source/src/MmsgBatch.cpp index 1be2cf1a5..5e6be15d3 100644 --- a/Source/src/MmsgBatch.cpp +++ b/Source/src/MmsgBatch.cpp @@ -10,7 +10,7 @@ namespace MafiaNet { -void SockaddrToSystemAddress(const sockaddr_storage &from, SystemAddress *out) +bool SockaddrToSystemAddress(const sockaddr_storage &from, SystemAddress *out) { // Mirrors the byte-order handling of RNS2_Berkley::RecvFromBlockingIPV4And6: // the sockaddr keeps the port in network order (copied verbatim into the @@ -20,15 +20,24 @@ void SockaddrToSystemAddress(const sockaddr_storage &from, SystemAddress *out) memcpy(&out->address.addr4, reinterpret_cast(&from), sizeof(sockaddr_in)); out->debugPort = ntohs(out->address.addr4.sin_port); + return true; } #if RAKNET_SUPPORT_IPV6 == 1 - else + if (from.ss_family == AF_INET6) { memcpy(&out->address.addr6, reinterpret_cast(&from), sizeof(sockaddr_in6)); out->debugPort = ntohs(out->address.addr6.sin6_port); + return true; } #endif + + // A family this build cannot represent (an IPv6 source on an IPv4-only + // build, or AF_UNSPEC from a failed read). Overwrite rather than leave the + // recycled struct's previous sender in place -- that would attribute the + // datagram to the wrong peer. + *out = UNASSIGNED_SYSTEM_ADDRESS; + return false; } void DispatchRecvBatch(RNS2EventHandler *handler, @@ -43,18 +52,18 @@ void DispatchRecvBatch(RNS2EventHandler *handler, for (unsigned i = 0; i < received; ++i) { RNS2RecvStruct *s = slots[i]; - if (lens[i] > 0) + if (lens[i] > 0 && SockaddrToSystemAddress(addrs[i], &s->systemAddress)) { s->bytesRead = lens[i]; s->timeRead = now; s->socket = socket; - SockaddrToSystemAddress(addrs[i], &s->systemAddress); handler->OnRNS2Recv(s); } else { - // Zero-length read: same as the scalar bytesRead<=0 branch -- free - // the struct rather than surfacing an empty packet. + // Zero-length read (same as the scalar bytesRead<=0 branch) or an + // undecodable source address -- free the struct rather than + // surfacing a packet we cannot attribute to a peer. handler->DeallocRNS2RecvStruct(s, __FILE__, __LINE__); } } diff --git a/Source/src/RakNetSocket2.cpp b/Source/src/RakNetSocket2.cpp index 8cf9782f9..c0582d3f7 100644 --- a/Source/src/RakNetSocket2.cpp +++ b/Source/src/RakNetSocket2.cpp @@ -58,15 +58,19 @@ void RakNetSocket2::SetRecvEventHandler(RNS2EventHandler *_eventHandler) {eventH RNS2SendResult RakNetSocket2::SendBatch( RNS2_SendParameters *sends, unsigned count, const char *file, unsigned int line ) { // Portable default: send each datagram individually. Platforms with a real - // batched syscall (Linux/sendmmsg) override this. - RNS2SendResult total=0; - for (unsigned i=0; i0) - total += r; - } - return total; + // batched syscall (Linux/sendmmsg) override this. Driving it through + // DriveBatchedSend -- one datagram per "call" -- gives the same return + // contract as the sendmmsg override: a datagram count, and the error code + // only when nothing at all went out. Send() reports bytes, so a successful + // send is normalized to 1 (a zero-byte return still counts as one datagram + // accepted; it must not be mistaken for DriveBatchedSend's "no progress"). + return DriveBatchedSend(count, + [&](unsigned offset, unsigned remaining) -> int + { + (void) remaining; + RNS2SendResult r = Send(&sends[offset], file, line); + return r>=0 ? 1 : (int) r; + }); } RNS2Type RakNetSocket2::GetSocketType(void) const {return socketType;} void RakNetSocket2::SetSocketType(RNS2Type t) {socketType=t;} diff --git a/Source/src/RakNetSocket2_Berkley.cpp b/Source/src/RakNetSocket2_Berkley.cpp index a4e4e4c4c..a0b2dec2a 100644 --- a/Source/src/RakNetSocket2_Berkley.cpp +++ b/Source/src/RakNetSocket2_Berkley.cpp @@ -537,12 +537,19 @@ void RNS2_Berkley::RecvFromBatchedLoop(void) sockaddr_storage addrs[MMSG_BATCH_MAX]; int lens[MMSG_BATCH_MAX]; + // Recv structs are held across a failed recvmmsg rather than reallocated + // each pass: Linux reports asynchronous errors on a connectionless UDP + // socket (ECONNREFUSED when a peer's port is closed), and rebuilding the + // whole batch on every one of those would churn MMSG_BATCH_MAX alloc/free + // round trips per pass instead of the scalar path's one. + unsigned allocated=0; + unsigned consecutiveErrors=0; + while ( endThreads == false ) { - // Preallocate a batch of recv structs and point each iovec at its data - // buffer, so recvmmsg writes straight into the structs we will hand to + // Top the batch back up to full, then point each iovec at the struct's + // data buffer so recvmmsg writes straight into the structs we hand to // the event handler -- no extra copy. - unsigned allocated=0; for (; allocatedAllocRNS2RecvStruct(_FILE_AND_LINE_); @@ -550,20 +557,27 @@ void RNS2_Berkley::RecvFromBatchedLoop(void) break; s->socket=this; slots[allocated]=s; - iovecs[allocated].iov_base=s->data; - iovecs[allocated].iov_len=sizeof(s->data); - memset(&msgs[allocated], 0, sizeof(msgs[allocated])); - msgs[allocated].msg_hdr.msg_iov=&iovecs[allocated]; - msgs[allocated].msg_hdr.msg_iovlen=1; - msgs[allocated].msg_hdr.msg_name=&addrs[allocated]; - msgs[allocated].msg_hdr.msg_namelen=sizeof(addrs[allocated]); } if (allocated==0) { - RakSleep(0); + // Out of recv structs entirely; back off rather than spin. + RakSleep(1); continue; } + // Rebuilt every pass, not just on allocation: recvmmsg writes back + // msg_namelen and msg_flags, so a reused header must be reset. + for (unsigned i=0; idata; + iovecs[i].iov_len=sizeof(slots[i]->data); + memset(&msgs[i], 0, sizeof(msgs[i])); + msgs[i].msg_hdr.msg_iov=&iovecs[i]; + msgs[i].msg_hdr.msg_iovlen=1; + msgs[i].msg_hdr.msg_name=&addrs[i]; + msgs[i].msg_hdr.msg_namelen=sizeof(addrs[i]); + } + // MSG_WAITFORONE: block until at least one datagram is ready, then return // everything already queued. Without it recvmmsg would wait to fill the // whole array (or hit a timeout), adding latency at low packet rates. @@ -573,22 +587,30 @@ void RNS2_Berkley::RecvFromBatchedLoop(void) if (n<0) { // Interrupted / error (includes the shutdown poke unblocking us). - // Release the whole batch and re-evaluate endThreads. - for (unsigned i=0; iDeallocRNS2RecvStruct(slots[i], _FILE_AND_LINE_); - RakSleep(0); + // Keep the batch and re-evaluate endThreads. Back off progressively + // so a persistent error cannot spin this thread at 100%; the loop + // condition is still checked first, so shutdown stays prompt. + if (consecutiveErrorsDeallocRNS2RecvStruct(slots[i], _FILE_AND_LINE_); } #endif // MAFIANET_USE_RECVMMSG && __linux__ diff --git a/Tests/Unit/MmsgBatchTests.cpp b/Tests/Unit/MmsgBatchTests.cpp index c5c335778..d6a752134 100644 --- a/Tests/Unit/MmsgBatchTests.cpp +++ b/Tests/Unit/MmsgBatchTests.cpp @@ -24,6 +24,7 @@ #include // inet_pton, htons #endif #include +#include #include using namespace MafiaNet; @@ -73,6 +74,47 @@ namespace // A sentinel non-null socket pointer; DispatchRecvBatch only stores it. RakNetSocket2 *const kSentinelSocket = reinterpret_cast(0xF00D); + + sockaddr_storage MakeFamily(unsigned short family) + { + sockaddr_storage ss; + memset(&ss, 0, sizeof(ss)); + ss.ss_family = family; + return ss; + } + + // Records every datagram handed to Send(), so tests can assert on what the + // batch flushed and when. SendBatch is deliberately NOT overridden: these + // tests exercise the portable base implementation. + struct RecordingSocket : RakNetSocket2 + { + struct Sent + { + std::string payload; + unsigned short port; + }; + + std::vector sent; + unsigned sendCalls = 0; + int failFrom = -1; // fail every Send() from this index on + + RNS2SendResult Send(RNS2_SendParameters *p, const char *, unsigned int) override + { + const unsigned index = sendCalls++; + if (failFrom >= 0 && index >= (unsigned) failFrom) + return -1; + sent.push_back({std::string(p->data, (size_t) p->length), + p->systemAddress.GetPort()}); + return p->length; // Send() reports bytes, not a datagram count + } + }; + + SystemAddress MakeDest(unsigned short port) + { + SystemAddress a; + a.FromStringExplicitPort("127.0.0.1", port); + return a; + } } // --------------------------------------------------------------------------- @@ -171,6 +213,20 @@ TEST(SockaddrToSystemAddress, PreservesIPv4Address) EXPECT_EQ(out.address.addr4.sin_addr.s_addr, in->sin_addr.s_addr); } +TEST(SockaddrToSystemAddress, RejectsUndecodableFamilyWithoutKeepingStaleAddress) +{ + // recv structs are recycled, so `out` arrives holding the previous sender. + // An address family this build cannot decode must not leave that in place -- + // the datagram would be attributed to the wrong peer. + SystemAddress out; + ASSERT_TRUE(SockaddrToSystemAddress(MakeV4("198.51.100.9", 5555), &out)); + ASSERT_EQ(out.GetPort(), 5555); + + sockaddr_storage unknown = MakeFamily(AF_UNSPEC); + EXPECT_FALSE(SockaddrToSystemAddress(unknown, &out)); + EXPECT_EQ(out, UNASSIGNED_SYSTEM_ADDRESS); +} + // --------------------------------------------------------------------------- // DispatchRecvBatch -- fanning a received batch out to the handler. // --------------------------------------------------------------------------- @@ -246,3 +302,169 @@ TEST(DispatchRecvBatch, FreesEverythingWhenNothingReceived) EXPECT_TRUE(handler.received.empty()); EXPECT_EQ(handler.deallocated.size(), 3u); } + +TEST(DispatchRecvBatch, FreesDatagramsWhoseSourceAddressCannotBeDecoded) +{ + const unsigned allocated = 2; + std::vector storage(allocated); + RNS2RecvStruct *slots[allocated]; + for (unsigned i = 0; i < allocated; ++i) + slots[i] = &storage[i]; + + int lens[allocated] = {11, 22}; + sockaddr_storage addrs[allocated] = {MakeFamily(AF_UNSPEC), + MakeV4("10.0.0.2", 2222)}; + + RecordingHandler handler; + DispatchRecvBatch(&handler, slots, allocated, lens, addrs, + /*received=*/2, kSentinelSocket, /*now=*/1); + + // Slot 0 has bytes but no usable source address: free it rather than + // surfacing a packet attributed to the recycled struct's previous sender. + ASSERT_EQ(handler.received.size(), 1u); + EXPECT_EQ(handler.received[0].bytesRead, 22); + EXPECT_EQ(handler.received[0].port, 2222); + ASSERT_EQ(handler.deallocated.size(), 1u); + EXPECT_EQ(handler.deallocated[0], slots[0]); +} + +// --------------------------------------------------------------------------- +// RakNetSocket2::SendBatch -- the portable default must match the sendmmsg +// override's contract: a datagram COUNT, and an error only if nothing went out. +// --------------------------------------------------------------------------- + +TEST(SocketSendBatch, ReturnsDatagramCountNotByteTotal) +{ + RecordingSocket socket; + RNS2_SendParameters sends[3]; + char payload[64] = {}; + for (unsigned i = 0; i < 3; ++i) + { + sends[i].data = payload; + sends[i].length = 40; // 3 * 40 bytes, but only 3 datagrams + sends[i].systemAddress = MakeDest(1000); + } + + EXPECT_EQ(socket.SendBatch(sends, 3, _FILE_AND_LINE_), 3); + EXPECT_EQ(socket.sent.size(), 3u); +} + +TEST(SocketSendBatch, PropagatesErrorOnlyWhenNothingWasSent) +{ + RNS2_SendParameters sends[4]; + char payload[8] = {}; + for (unsigned i = 0; i < 4; ++i) + { + sends[i].data = payload; + sends[i].length = 8; + sends[i].systemAddress = MakeDest(1000); + } + + RecordingSocket allFail; + allFail.failFrom = 0; + EXPECT_EQ(allFail.SendBatch(sends, 4, _FILE_AND_LINE_), -1); + + // Two out, then a failure: report the progress made, like sendmmsg does. + RecordingSocket partial; + partial.failFrom = 2; + EXPECT_EQ(partial.SendBatch(sends, 4, _FILE_AND_LINE_), 2); + EXPECT_EQ(partial.sent.size(), 2u); +} + +// --------------------------------------------------------------------------- +// RNS2SendBatch -- accumulating datagrams and flushing them through SendBatch. +// --------------------------------------------------------------------------- + +TEST(RNS2SendBatch, CopiesPayloadsSoCallerCanReuseItsBuffer) +{ + // The reliability layer reuses one serialization buffer per datagram, so the + // batch must copy rather than alias. + RecordingSocket socket; + char scratch[8]; + { + RNS2SendBatch batch(&socket, MakeDest(4242)); + memcpy(scratch, "first", 5); + batch.Add(scratch, 5); + memcpy(scratch, "secnd", 5); + batch.Add(scratch, 5); + EXPECT_TRUE(socket.sent.empty()) << "nothing should go out before Flush"; + } + + ASSERT_EQ(socket.sent.size(), 2u); + EXPECT_EQ(socket.sent[0].payload, "first"); + EXPECT_EQ(socket.sent[1].payload, "secnd"); + EXPECT_EQ(socket.sent[0].port, 4242); + EXPECT_EQ(socket.sent[1].port, 4242); +} + +TEST(RNS2SendBatch, DestructorFlushesAsBackstop) +{ + RecordingSocket socket; + { + RNS2SendBatch batch(&socket, MakeDest(1234)); + batch.Add("x", 1); + } + ASSERT_EQ(socket.sent.size(), 1u); + EXPECT_EQ(socket.sent[0].payload, "x"); +} + +TEST(RNS2SendBatch, ExplicitFlushIsNotRepeatedByTheDestructor) +{ + RecordingSocket socket; + { + RNS2SendBatch batch(&socket, MakeDest(1234)); + batch.Add("x", 1); + batch.Flush(); + EXPECT_EQ(socket.sent.size(), 1u); + } + EXPECT_EQ(socket.sent.size(), 1u) << "destructor must not re-send a flushed batch"; +} + +TEST(RNS2SendBatch, FlushesAtTheBatchBoundaryInsteadOfOverrunning) +{ + RecordingSocket socket; + { + RNS2SendBatch batch(&socket, MakeDest(7000)); + for (unsigned i = 0; i < MMSG_BATCH_MAX + 3; ++i) + { + const char byte = (char) ('a' + (i % 26)); + batch.Add(&byte, 1); + } + // The first MMSG_BATCH_MAX are flushed when slot MMSG_BATCH_MAX is added. + EXPECT_EQ(socket.sent.size(), (size_t) MMSG_BATCH_MAX); + } + + ASSERT_EQ(socket.sent.size(), (size_t) MMSG_BATCH_MAX + 3); + for (unsigned i = 0; i < MMSG_BATCH_MAX + 3; ++i) + EXPECT_EQ(socket.sent[i].payload, std::string(1, (char) ('a' + (i % 26)))) + << "datagram " << i << " out of order or corrupted across the boundary"; +} + +TEST(RNS2SendBatch, DropsOversizedDatagramsInsteadOfTruncating) +{ + // Silently clamping would ship corrupted bytes with no error signal; the + // reliability layer resends what never went out. +#if defined(_DEBUG) + GTEST_SKIP() << "the oversized case trips RakAssert by design in debug builds"; +#else + RecordingSocket socket; + std::vector oversized(MAXIMUM_MTU_SIZE + 1, 'z'); + { + RNS2SendBatch batch(&socket, MakeDest(1234)); + batch.Add(oversized.data(), MAXIMUM_MTU_SIZE + 1); + batch.Add("ok", 2); + } + ASSERT_EQ(socket.sent.size(), 1u); + EXPECT_EQ(socket.sent[0].payload, "ok"); +#endif +} + +TEST(RNS2SendBatch, EmptyBatchSendsNothing) +{ + RecordingSocket socket; + { + RNS2SendBatch batch(&socket, MakeDest(1234)); + (void) batch; + } + EXPECT_EQ(socket.sendCalls, 0u); +} From 5994f84e69aa4cb91c9d87afb2b7a8656b866d34 Mon Sep 17 00:00:00 2001 From: Segfault <5221072+Segfaultd@users.noreply.github.com> Date: Sun, 26 Jul 2026 19:05:02 +0200 Subject: [PATCH 4/6] fix(mmsg): retain recv slots, honour TTL, report dropped sends Pre-landing review follow-ups on the batched datagram I/O work. - RecvFromBatchedLoop refilled all MMSG_BATCH_MAX slots every pass and handed the whole array to DispatchRecvBatch, so a pass that received a single datagram still cost 64 allocs plus 63 frees -- each taking the event handler's pool mutex. Only the slots that actually received a datagram now change hands; the untouched tail carries over, putting the steady state at one alloc/free round trip per datagram, the same as the scalar path. - RNS2_Linux::SendBatch silently ignored RNS2_SendParameters::ttl, which the scalar Send() honours by bracketing the sendto with setsockopt(IP_TTL). sendmmsg has no per-message TTL, so a batch carrying one is now deferred to the portable base loop. Nothing sets ttl today (RNS2SendBatch always flushes with zero), so this closes a trap rather than a live bug. - RNS2SendBatch::Flush discarded SendBatch's result: a failed or partial sendmmsg dropped datagrams with no trace, where the scalar path at least reports its sendto failures. Both that shortfall and a dropped oversized datagram now emit RAKNET_DEBUG_PRINTF. Also drops the dead lens[] writes past the received count, corrects the comment claiming shutdown arrives on the recvmmsg error branch (the poke is a real datagram, so it surfaces as a normal pass), and switches to _FILE_AND_LINE_ to match the rest of the codebase. Tests: cover the DispatchRecvBatch clamp, the IPv6 decode branch of SockaddrToSystemAddress (and its IPv4-only-build rejection), TTL pass-through in the base SendBatch, and that RNS2SendBatch flushes with a zero TTL so the sendmmsg path is never deferred. Verified with the full suite on Linux with both flags on (102/102). --- Source/include/mafianet/MmsgBatch.h | 16 ++++- Source/include/mafianet/socket2.h | 2 + Source/src/MmsgBatch.cpp | 4 +- Source/src/RakNetSocket2.cpp | 12 ++++ Source/src/RakNetSocket2_Berkley.cpp | 51 +++++++++----- Tests/Unit/MmsgBatchTests.cpp | 100 ++++++++++++++++++++++++++- 6 files changed, 162 insertions(+), 23 deletions(-) diff --git a/Source/include/mafianet/MmsgBatch.h b/Source/include/mafianet/MmsgBatch.h index fcfcb54e3..c8ab1a159 100644 --- a/Source/include/mafianet/MmsgBatch.h +++ b/Source/include/mafianet/MmsgBatch.h @@ -23,6 +23,7 @@ #include "mafianet/memoryoverride.h" #include // memcpy +#include // RAKNET_DEBUG_PRINTF defaults to printf namespace MafiaNet { @@ -143,7 +144,13 @@ class RNS2SendBatch { RakAssert(length >= 0 && length <= MAXIMUM_MTU_SIZE); if (length < 0 || length > MAXIMUM_MTU_SIZE) - return; // drop, don't corrupt; the reliability layer will resend + { + // Drop rather than corrupt. Reliable traffic is resent by the + // reliability layer; an unreliable datagram is simply lost, which is + // why the diagnostic below is the only trace it leaves. + RAKNET_DEBUG_PRINTF("RNS2SendBatch dropped an oversized datagram (%d bytes).\n", length); + return; + } if (count == MMSG_BATCH_MAX) Flush(); memcpy(Slot(count), data, (size_t) length); @@ -163,7 +170,12 @@ class RNS2SendBatch sends[i].systemAddress = dest; sends[i].ttl = 0; } - socket->SendBatch(sends, count, __FILE__, __LINE__); + // SendBatch returns a datagram count, or a negative error when nothing at + // all went out. Either shortfall means datagrams were dropped; the scalar + // path reports its sendto failures the same way, so don't lose the signal. + const RNS2SendResult sent = socket->SendBatch(sends, count, _FILE_AND_LINE_); + if (sent < 0 || (unsigned) sent < count) + RAKNET_DEBUG_PRINTF("SendBatch sent %d of %u datagrams.\n", (int) sent, count); count = 0; } diff --git a/Source/include/mafianet/socket2.h b/Source/include/mafianet/socket2.h index 0e7a2b884..7fa7c158a 100644 --- a/Source/include/mafianet/socket2.h +++ b/Source/include/mafianet/socket2.h @@ -120,6 +120,8 @@ class RakNetSocket2 // byte total Send() returns -- or a negative error code if the very first // datagram failed, mirroring sendmmsg(2). Both implementations must agree on // this; see MmsgBatchTests. + // RNS2_SendParameters::ttl is honoured either way: sendmmsg has no per-message + // TTL, so the override defers a batch carrying one to this base loop. virtual RNS2SendResult SendBatch( RNS2_SendParameters *sends, unsigned count, const char *file, unsigned int line ); RNS2Type GetSocketType(void) const; void SetSocketType(RNS2Type t); diff --git a/Source/src/MmsgBatch.cpp b/Source/src/MmsgBatch.cpp index 5e6be15d3..386372303 100644 --- a/Source/src/MmsgBatch.cpp +++ b/Source/src/MmsgBatch.cpp @@ -64,13 +64,13 @@ void DispatchRecvBatch(RNS2EventHandler *handler, // Zero-length read (same as the scalar bytesRead<=0 branch) or an // undecodable source address -- free the struct rather than // surfacing a packet we cannot attribute to a peer. - handler->DeallocRNS2RecvStruct(s, __FILE__, __LINE__); + handler->DeallocRNS2RecvStruct(s, _FILE_AND_LINE_); } } // Hand back the unused tail so no preallocated buffer leaks. for (unsigned i = received; i < allocated; ++i) - handler->DeallocRNS2RecvStruct(slots[i], __FILE__, __LINE__); + handler->DeallocRNS2RecvStruct(slots[i], _FILE_AND_LINE_); } } // namespace MafiaNet diff --git a/Source/src/RakNetSocket2.cpp b/Source/src/RakNetSocket2.cpp index c0582d3f7..cd9df4048 100644 --- a/Source/src/RakNetSocket2.cpp +++ b/Source/src/RakNetSocket2.cpp @@ -291,6 +291,18 @@ RNS2SendResult RNS2_Linux::Send( RNS2_SendParameters *sendParameters, const char #if defined(MAFIANET_USE_SENDMMSG) RNS2SendResult RNS2_Linux::SendBatch( RNS2_SendParameters *sends, unsigned count, const char *file, unsigned int line ) { + // sendmmsg has no per-message TTL, whereas the scalar Send() honours + // RNS2_SendParameters::ttl by bracketing the sendto with setsockopt(IP_TTL). + // Batching must not silently drop that: hand any batch carrying a TTL back to + // the portable base implementation, which is the same Send() loop. The + // reliability layer never sets ttl, so this is not the hot path -- only NAT + // punchthrough style callers reach it. + for (unsigned i=0; i0) + return RakNetSocket2::SendBatch(sends, count, file, line); + } + (void) file; (void) line; diff --git a/Source/src/RakNetSocket2_Berkley.cpp b/Source/src/RakNetSocket2_Berkley.cpp index a0b2dec2a..32c5f4f64 100644 --- a/Source/src/RakNetSocket2_Berkley.cpp +++ b/Source/src/RakNetSocket2_Berkley.cpp @@ -537,11 +537,17 @@ void RNS2_Berkley::RecvFromBatchedLoop(void) sockaddr_storage addrs[MMSG_BATCH_MAX]; int lens[MMSG_BATCH_MAX]; - // Recv structs are held across a failed recvmmsg rather than reallocated - // each pass: Linux reports asynchronous errors on a connectionless UDP - // socket (ECONNREFUSED when a peer's port is closed), and rebuilding the - // whole batch on every one of those would churn MMSG_BATCH_MAX alloc/free - // round trips per pass instead of the scalar path's one. + // Recv structs live across passes rather than being rebuilt each time. Only + // the slots that actually received a datagram change hands; the untouched + // tail is carried over, so the steady state costs one alloc/free round trip + // per datagram -- the same as the scalar path -- instead of MMSG_BATCH_MAX of + // them per pass. Each of those round trips takes the event handler's pool + // mutex, and at typical packet rates recvmmsg returns far fewer than + // MMSG_BATCH_MAX datagrams, so rebuilding the batch would dominate the + // syscall saving the batching exists for. The same carry-over covers failed + // passes: Linux reports asynchronous errors on a connectionless UDP socket + // (ECONNREFUSED when a peer's port is closed) and those must not churn the + // pool either. unsigned allocated=0; unsigned consecutiveErrors=0; @@ -586,10 +592,13 @@ void RNS2_Berkley::RecvFromBatchedLoop(void) if (n<0) { - // Interrupted / error (includes the shutdown poke unblocking us). - // Keep the batch and re-evaluate endThreads. Back off progressively - // so a persistent error cannot spin this thread at 100%; the loop - // condition is still checked first, so shutdown stays prompt. + // Interrupted or an asynchronous socket error. (Shutdown does not + // come through here: BlockOnStopRecvPollingThread pokes the socket + // with a real datagram, so it surfaces as a normal n>=1 pass and the + // loop condition below catches endThreads.) Keep the batch and back + // off progressively so a persistent error cannot spin this thread at + // 100%; the loop condition is still checked first, so shutdown stays + // prompt. if (consecutiveErrors0) + memmove(slots, slots+received, allocated*sizeof(slots[0])); } // Release whatever the last pass was still holding. diff --git a/Tests/Unit/MmsgBatchTests.cpp b/Tests/Unit/MmsgBatchTests.cpp index d6a752134..fcfefe550 100644 --- a/Tests/Unit/MmsgBatchTests.cpp +++ b/Tests/Unit/MmsgBatchTests.cpp @@ -83,6 +83,19 @@ namespace return ss; } +#if RAKNET_SUPPORT_IPV6 == 1 + sockaddr_storage MakeV6(const char *ip, unsigned short port) + { + sockaddr_storage ss; + memset(&ss, 0, sizeof(ss)); + sockaddr_in6 *in6 = reinterpret_cast(&ss); + in6->sin6_family = AF_INET6; + in6->sin6_port = htons(port); + inet_pton(AF_INET6, ip, &in6->sin6_addr); + return ss; + } +#endif + // Records every datagram handed to Send(), so tests can assert on what the // batch flushed and when. SendBatch is deliberately NOT overridden: these // tests exercise the portable base implementation. @@ -92,6 +105,7 @@ namespace { std::string payload; unsigned short port; + int ttl; }; std::vector sent; @@ -104,7 +118,7 @@ namespace if (failFrom >= 0 && index >= (unsigned) failFrom) return -1; sent.push_back({std::string(p->data, (size_t) p->length), - p->systemAddress.GetPort()}); + p->systemAddress.GetPort(), p->ttl}); return p->length; // Send() reports bytes, not a datagram count } }; @@ -213,6 +227,31 @@ TEST(SockaddrToSystemAddress, PreservesIPv4Address) EXPECT_EQ(out.address.addr4.sin_addr.s_addr, in->sin_addr.s_addr); } +#if RAKNET_SUPPORT_IPV6 == 1 +TEST(SockaddrToSystemAddress, DecodesIPv6PortToHostOrder) +{ + sockaddr_storage ss = MakeV6("2001:db8::1", 4660); + const sockaddr_in6 *in6 = reinterpret_cast(&ss); + SystemAddress out; + ASSERT_TRUE(SockaddrToSystemAddress(ss, &out)); + EXPECT_EQ(out.GetPort(), 4660); + EXPECT_EQ(out.debugPort, 4660); + EXPECT_EQ(out.GetIPVersion(), 6); + EXPECT_EQ(memcmp(&out.address.addr6.sin6_addr, &in6->sin6_addr, + sizeof(in6->sin6_addr)), + 0); +} +#else +TEST(SockaddrToSystemAddress, RejectsIPv6SourceOnIPv4OnlyBuild) +{ + // An IPv6 sender reaching an IPv4-only build has no representable address; + // the caller must be told so rather than handed a half-decoded one. + SystemAddress out; + EXPECT_FALSE(SockaddrToSystemAddress(MakeFamily(AF_INET6), &out)); + EXPECT_EQ(out, UNASSIGNED_SYSTEM_ADDRESS); +} +#endif + TEST(SockaddrToSystemAddress, RejectsUndecodableFamilyWithoutKeepingStaleAddress) { // recv structs are recycled, so `out` arrives holding the previous sender. @@ -328,6 +367,30 @@ TEST(DispatchRecvBatch, FreesDatagramsWhoseSourceAddressCannotBeDecoded) EXPECT_EQ(handler.deallocated[0], slots[0]); } +TEST(DispatchRecvBatch, ClampsAReceivedCountLargerThanTheAllocation) +{ + // Defensive clamp: a received count above the number of slots would otherwise + // walk off the end of the arrays. Nothing beyond the allocation is touched. + const unsigned allocated = 2; + std::vector storage(allocated); + RNS2RecvStruct *slots[allocated]; + for (unsigned i = 0; i < allocated; ++i) + slots[i] = &storage[i]; + + int lens[allocated] = {11, 22}; + sockaddr_storage addrs[allocated] = {MakeV4("10.0.0.1", 1111), + MakeV4("10.0.0.2", 2222)}; + + RecordingHandler handler; + DispatchRecvBatch(&handler, slots, allocated, lens, addrs, + /*received=*/9, kSentinelSocket, /*now=*/1); + + ASSERT_EQ(handler.received.size(), 2u); + EXPECT_EQ(handler.received[0].port, 1111); + EXPECT_EQ(handler.received[1].port, 2222); + EXPECT_TRUE(handler.deallocated.empty()); +} + // --------------------------------------------------------------------------- // RakNetSocket2::SendBatch -- the portable default must match the sendmmsg // override's contract: a datagram COUNT, and an error only if nothing went out. @@ -371,10 +434,45 @@ TEST(SocketSendBatch, PropagatesErrorOnlyWhenNothingWasSent) EXPECT_EQ(partial.sent.size(), 2u); } +TEST(SocketSendBatch, ForwardsPerDatagramTtl) +{ + // sendmmsg has no per-message TTL, so RNS2_Linux::SendBatch defers a batch + // carrying one to this base loop. That only preserves the TTL if the loop + // hands the whole parameter through to Send() untouched. + RecordingSocket socket; + RNS2_SendParameters sends[2]; + char payload[4] = {}; + for (unsigned i = 0; i < 2; ++i) + { + sends[i].data = payload; + sends[i].length = 4; + sends[i].systemAddress = MakeDest(1000); + sends[i].ttl = (int) (i + 1); + } + + EXPECT_EQ(socket.SendBatch(sends, 2, _FILE_AND_LINE_), 2); + ASSERT_EQ(socket.sent.size(), 2u); + EXPECT_EQ(socket.sent[0].ttl, 1); + EXPECT_EQ(socket.sent[1].ttl, 2); +} + // --------------------------------------------------------------------------- // RNS2SendBatch -- accumulating datagrams and flushing them through SendBatch. // --------------------------------------------------------------------------- +TEST(RNS2SendBatch, FlushesWithoutATtlSoTheSendmmsgPathIsNeverDeferred) +{ + // The reliability layer never asks for a TTL; keeping it zero is what lets + // RNS2_Linux::SendBatch take the real sendmmsg path on every flush. + RecordingSocket socket; + { + RNS2SendBatch batch(&socket, MakeDest(1234)); + batch.Add("x", 1); + } + ASSERT_EQ(socket.sent.size(), 1u); + EXPECT_EQ(socket.sent[0].ttl, 0); +} + TEST(RNS2SendBatch, CopiesPayloadsSoCallerCanReuseItsBuffer) { // The reliability layer reuses one serialization buffer per datagram, so the From 392ea0adddf23d5435c8c5368c8d96cd75c90f05 Mon Sep 17 00:00:00 2001 From: Segfault <5221072+Segfaultd@users.noreply.github.com> Date: Sun, 26 Jul 2026 19:20:38 +0200 Subject: [PATCH 5/6] fix(mmsg): drop only the failed datagram, not the rest of the batch DriveBatchedSend treated any negative transmit() return as "stop", so a single undeliverable datagram silently discarded every datagram queued behind it -- the scalar Send() loop it replaced would have delivered those. Split the contract: 0 now means a transient socket-wide condition (stop, don't spin), negative means the message at that offset is itself undeliverable (drop it and continue). RNS2_Linux::SendBatch classifies errno accordingly, since sendmmsg funnels both through the same -1. Also: - Assert in RNS2SendBatch::Flush that SendBatch returned a datagram count rather than a byte total. That is the one contract the sendmmsg override cannot be unit-tested against, so check it at runtime. - Build the linux-mmsg CI job in Debug. It is the only job that compiles these paths, and RakAssert -- which guards the return contract, the RNS2SendBatch reentrancy invariant and the IPv4-only address branch -- is compiled out unless _DEBUG. - Make the IPv4-only branch of the sendmmsg address setup explicit instead of silently leaving msg_name null. - Cover the previously untested failure paths: a partially failing flush, a fully failing flush not resending its batch, per-message skip, and first-error propagation. - Ignore build-*/ and docs/superpowers/. --- .github/workflows/build.yml | 8 +- .gitignore | 6 +- Source/include/mafianet/MmsgBatch.h | 60 ++++++++++----- Source/include/mafianet/socket2.h | 8 +- Source/src/RakNetSocket2.cpp | 31 +++++++- Tests/Unit/MmsgBatchTests.cpp | 111 ++++++++++++++++++++++++++-- 6 files changed, 192 insertions(+), 32 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 9d60a894b..207b0d635 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -50,14 +50,20 @@ jobs: run: sudo apt-get update && sudo apt-get install -y libssl-dev - name: Configure CMake + # Debug on purpose: this is the only job that compiles the mmsg paths, + # and RakAssert (which is compiled out unless _DEBUG) is what guards the + # invariants they rely on -- the SendBatch count-not-bytes return + # contract, the RNS2SendBatch reentrancy guard, the IPv4-only address + # branch. In Release those checks would not run anywhere. run: | cmake -B build \ + -DCMAKE_BUILD_TYPE=Debug \ -DMAFIANET_BUILD_TESTS=ON \ -DMAFIANET_USE_RECVMMSG=ON \ -DMAFIANET_USE_SENDMMSG=ON - name: Build - run: cmake --build build --config Release --parallel 4 + run: cmake --build build --parallel 4 - name: Run tests run: | diff --git a/.gitignore b/.gitignore index fccf37cec..32682e097 100644 --- a/.gitignore +++ b/.gitignore @@ -14,6 +14,7 @@ # Build directories build/ Build/ +build-*/ out/ # Compiled libraries @@ -46,4 +47,7 @@ Makefile # Package files *.tar.gz -*.zip \ No newline at end of file +*.zip + +# Agent scratch space (plans, specs) -- kept on disk, never committed +docs/superpowers/ \ No newline at end of file diff --git a/Source/include/mafianet/MmsgBatch.h b/Source/include/mafianet/MmsgBatch.h index c8ab1a159..fb02a3fbc 100644 --- a/Source/include/mafianet/MmsgBatch.h +++ b/Source/include/mafianet/MmsgBatch.h @@ -53,36 +53,52 @@ bool SockaddrToSystemAddress(const sockaddr_storage &from, SystemAddress *out); /// \a transmit is invoked as transmit(offset, count) to send the messages in /// [offset, offset+count). It must return: /// - a positive count of messages accepted this call (advance and continue), -/// - 0 to signal no further progress is possible right now (stop), or -/// - a negative errno-style value meaning the call failed with nothing sent. +/// - 0 when no further progress is possible right now -- a transient, +/// whole-socket condition such as EAGAIN/ENOBUFS (stop, don't spin), or +/// - a negative errno-style value meaning the message at \a offset itself +/// cannot be sent (a permanent per-message error such as EMSGSIZE or +/// EDESTADDRREQ): it is dropped and the rest of the batch continues. /// -/// This mirrors sendmmsg(2), which returns the number of messages sent and only -/// reports an error (-1) when *no* datagram could be sent. Returns the total -/// number of messages sent (0..total). If the very first call fails before any -/// message is sent, the negative error code is propagated unchanged so the -/// caller can inspect errno exactly as it would for a scalar sendto. +/// The 0-vs-negative split matters: sendmmsg reports both through the same -1, +/// so the caller's transmit() must classify errno. Treating a permanent +/// per-message error as "stop" would silently discard every datagram *after* +/// the bad one, which the scalar Send() loop would have delivered. +/// +/// Returns the total number of messages actually sent (0..total), or -- when +/// nothing at all went out and at least one message failed permanently -- the +/// first negative error code seen. Note that the drop-and-continue behaviour +/// means further syscalls may run after that first failure, so errno is not +/// preserved; transmit() should encode the error in its return value if the +/// caller needs it. template int DriveBatchedSend(unsigned total, TransmitFn transmit) { - unsigned sent = 0; - while (sent < total) + unsigned sent = 0; // messages actually accepted by the socket + unsigned offset = 0; // next message to attempt; may run ahead of `sent` + int firstError = 0; // first permanent per-message error seen, if any + while (offset < total) { - int r = transmit(sent, total - sent); + int r = transmit(offset, total - offset); if (r > 0) { - sent += (unsigned) r; // accepted r messages; retry the remainder + // Accepted r messages; retry the remainder from the new offset. + sent += (unsigned) r; + offset += (unsigned) r; continue; } if (r == 0) break; // no progress possible right now; don't spin - // r < 0: this call failed with nothing sent. Propagate the error only - // if we have not managed to send anything yet (mirrors sendmmsg, which - // reports -1 solely when no datagram at all could be sent); otherwise - // report the progress already made. - if (sent == 0) - return r; - break; + // r < 0: the message at `offset` is undeliverable. Drop just that one + // and keep going, so a single bad datagram cannot take the rest of the + // batch down with it. + if (firstError == 0) + firstError = r; + ++offset; } + // Mirror sendmmsg, which reports an error only when no datagram at all went + // out; otherwise report the progress made. + if (sent == 0 && firstError != 0) + return firstError; return (int) sent; } @@ -174,8 +190,16 @@ class RNS2SendBatch // all went out. Either shortfall means datagrams were dropped; the scalar // path reports its sendto failures the same way, so don't lose the signal. const RNS2SendResult sent = socket->SendBatch(sends, count, _FILE_AND_LINE_); + // A return above `count` means an override handed back a byte total + // instead of a datagram count -- the one contract the RNS2_Linux sendmmsg + // override cannot be unit-tested against, so check it at runtime here. + RakAssert(sent <= (RNS2SendResult) count && "SendBatch must return a datagram count, not a byte total"); if (sent < 0 || (unsigned) sent < count) RAKNET_DEBUG_PRINTF("SendBatch sent %d of %u datagrams.\n", (int) sent, count); + // Cleared unconditionally: a datagram the socket refused is dropped here, + // not carried into the next flush. Reliable traffic is resent by the + // reliability layer; retrying in place would reorder it against the + // datagrams queued after it. count = 0; } diff --git a/Source/include/mafianet/socket2.h b/Source/include/mafianet/socket2.h index 7fa7c158a..b09ace9b5 100644 --- a/Source/include/mafianet/socket2.h +++ b/Source/include/mafianet/socket2.h @@ -117,9 +117,11 @@ class RakNetSocket2 // type has a working default; RNS2_Linux overrides it with sendmmsg when // MAFIANET_USE_SENDMMSG is enabled. // Returns the number of datagrams accepted (0..count) -- a count, NOT the - // byte total Send() returns -- or a negative error code if the very first - // datagram failed, mirroring sendmmsg(2). Both implementations must agree on - // this; see MmsgBatchTests. + // byte total Send() returns -- or a negative error code when no datagram at + // all went out, mirroring sendmmsg(2). A datagram that fails on its own + // (bad destination, oversized) is dropped and the rest of the batch is still + // sent, so the count may be short without an error being reported. Both + // implementations must agree on this; see MmsgBatchTests. // RNS2_SendParameters::ttl is honoured either way: sendmmsg has no per-message // TTL, so the override defers a batch carrying one to this base loop. virtual RNS2SendResult SendBatch( RNS2_SendParameters *sends, unsigned count, const char *file, unsigned int line ); diff --git a/Source/src/RakNetSocket2.cpp b/Source/src/RakNetSocket2.cpp index cd9df4048..bdf9b2999 100644 --- a/Source/src/RakNetSocket2.cpp +++ b/Source/src/RakNetSocket2.cpp @@ -64,6 +64,9 @@ RNS2SendResult RakNetSocket2::SendBatch( RNS2_SendParameters *sends, unsigned co // only when nothing at all went out. Send() reports bytes, so a successful // send is normalized to 1 (a zero-byte return still counts as one datagram // accepted; it must not be mistaken for DriveBatchedSend's "no progress"). + // A failed Send() is reported as a per-message error, so DriveBatchedSend + // drops that datagram and continues -- matching the plain Send() loop this + // replaced, which ignored an individual sendto failure and kept going. return DriveBatchedSend(count, [&](unsigned offset, unsigned remaining) -> int { @@ -308,8 +311,10 @@ RNS2SendResult RNS2_Linux::SendBatch( RNS2_SendParameters *sends, unsigned count // DriveBatchedSend handles the partial-send resume; each transmit() call // coalesces up to MMSG_BATCH_MAX datagrams into one sendmmsg. sendmmsg - // returns the number of messages sent (>=1) or -1 (nothing sent), which - // DriveBatchedSend interprets exactly as documented. + // returns the number of messages sent (>=1) or -1 (nothing sent). It funnels + // both transient socket-wide conditions and permanent per-message failures + // through that same -1, so classify errno before handing it back: the two + // mean opposite things to DriveBatchedSend (stop vs. drop one and continue). const RNS2SendResult sent = DriveBatchedSend(count, [&](unsigned offset, unsigned remaining) -> int { @@ -334,10 +339,30 @@ RNS2SendResult RNS2_Linux::SendBatch( RNS2_SendParameters *sends, unsigned count #if RAKNET_SUPPORT_IPV6==1 msgs[i].msg_hdr.msg_name = (void*) &p->systemAddress.address.addr6; msgs[i].msg_hdr.msg_namelen = sizeof(sockaddr_in6); +#else + // Nothing to point msg_name at: this build has no sockaddr_in6. + // Leaving it null makes sendmmsg fail EDESTADDRREQ for this one + // datagram, which DriveBatchedSend drops while still delivering + // the rest of the batch. + RakAssert(false && "non-IPv4 destination on an IPv4-only build"); #endif } } - return sendmmsg(rns2Socket, msgs, chunk, 0); + const int r = sendmmsg(rns2Socket, msgs, chunk, 0); + if (r>=0) + return r; + const int err = errno; + // Transient and socket-wide: the send buffer is full or the call was + // interrupted. Nothing in this batch can go out right now, and the + // messages themselves are fine -- stop rather than burn a syscall per + // remaining datagram. Reliable traffic is resent by the reliability + // layer on a later tick. + if (err==EAGAIN || err==EWOULDBLOCK || err==ENOBUFS || err==EINTR) + return 0; + // Anything else (EMSGSIZE, EINVAL, EDESTADDRREQ, an ICMP error posted + // against a prior send) is a property of the message at `offset`. + // Report it as a per-message failure so the remainder still ships. + return -err; }); return sent; } diff --git a/Tests/Unit/MmsgBatchTests.cpp b/Tests/Unit/MmsgBatchTests.cpp index fcfefe550..3c6bb62c8 100644 --- a/Tests/Unit/MmsgBatchTests.cpp +++ b/Tests/Unit/MmsgBatchTests.cpp @@ -23,6 +23,7 @@ #else #include // inet_pton, htons #endif +#include #include #include #include @@ -111,12 +112,15 @@ namespace std::vector sent; unsigned sendCalls = 0; int failFrom = -1; // fail every Send() from this index on + int failOnly = -1; // fail only the Send() at this index RNS2SendResult Send(RNS2_SendParameters *p, const char *, unsigned int) override { const unsigned index = sendCalls++; if (failFrom >= 0 && index >= (unsigned) failFrom) return -1; + if (failOnly >= 0 && index == (unsigned) failOnly) + return -1; sent.push_back({std::string(p->data, (size_t) p->length), p->systemAddress.GetPort(), p->ttl}); return p->length; // Send() reports bytes, not a datagram count @@ -161,18 +165,43 @@ TEST(DriveBatchedSend, ResumesFromOffsetAcrossPartialSends) EXPECT_EQ(offsets, (std::vector{0u, 3u, 6u})); } -TEST(DriveBatchedSend, ReturnsErrorWhenFirstCallSendsNothing) +TEST(DriveBatchedSend, ReturnsErrorWhenNothingCouldBeSent) { - // sendmmsg only reports -1 when NO datagram could be sent; propagate it so - // the caller sees errno exactly as for a scalar sendto. + // Every message fails permanently, so nothing goes out: report the first + // error, mirroring sendmmsg's -1-only-when-nothing-was-sent contract. int sent = DriveBatchedSend(4, [&](unsigned, unsigned) { return -1; }); EXPECT_EQ(sent, -1); } -TEST(DriveBatchedSend, ReturnsProgressWhenErrorFollowsPartialSend) +TEST(DriveBatchedSend, ReturnsTheFirstErrorNotTheLast) +{ + int call = 0; + int sent = DriveBatchedSend(3, [&](unsigned, unsigned) { + return (call++ == 0) ? -EMSGSIZE : -EINVAL; + }); + EXPECT_EQ(sent, -EMSGSIZE); +} + +TEST(DriveBatchedSend, SkipsAPermanentlyFailedMessageAndSendsTheRest) +{ + // A negative return means the message at `offset` itself is undeliverable. + // Stopping there would silently discard every datagram after it -- the + // scalar Send() loop would have delivered those. + std::vector offsets; + int sent = DriveBatchedSend(5, [&](unsigned offset, unsigned count) { + offsets.push_back(offset); + if (offset == 2) + return -EMSGSIZE; // the third datagram is the bad one + return (int) std::min(count, 1u); + }); + EXPECT_EQ(sent, 4) << "only the bad datagram should be lost"; + EXPECT_EQ(offsets, (std::vector{0u, 1u, 2u, 3u, 4u})); +} + +TEST(DriveBatchedSend, ReturnsProgressWhenEveryRemainingMessageFails) { - // First call sends 4, second fails: we already made progress, so report the - // 4 that went out rather than the error. + // First call sends 4, the rest are all undeliverable: we already made + // progress, so report the 4 that went out rather than the error. int call = 0; int sent = DriveBatchedSend(9, [&](unsigned, unsigned) { return (call++ == 0) ? 4 : -1; @@ -180,6 +209,19 @@ TEST(DriveBatchedSend, ReturnsProgressWhenErrorFollowsPartialSend) EXPECT_EQ(sent, 4); } +TEST(DriveBatchedSend, StopsOnATransientFailureInsteadOfSkippingMessages) +{ + // A transient socket-wide condition (EAGAIN) is reported as 0, not as a + // negative: the messages are fine, so they must not be dropped one by one. + int calls = 0; + int sent = DriveBatchedSend(5, [&](unsigned, unsigned) { + ++calls; + return calls == 1 ? 2 : 0; + }); + EXPECT_EQ(sent, 2); + EXPECT_EQ(calls, 2) << "must not burn a call per remaining datagram"; +} + TEST(DriveBatchedSend, StopsOnNoProgressWithoutLooping) { // A zero return means "no progress possible right now" -- stop instead of @@ -434,6 +476,28 @@ TEST(SocketSendBatch, PropagatesErrorOnlyWhenNothingWasSent) EXPECT_EQ(partial.sent.size(), 2u); } +TEST(SocketSendBatch, DropsOnlyTheDatagramThatFailed) +{ + // One bad destination in the middle must not take the rest of the batch with + // it -- the plain Send() loop this replaced kept going after a failed sendto. + RecordingSocket socket; + socket.failOnly = 1; + RNS2_SendParameters sends[4]; + char payload[4] = {}; + for (unsigned i = 0; i < 4; ++i) + { + sends[i].data = payload; + sends[i].length = 4; + sends[i].systemAddress = MakeDest((unsigned short) (1000 + i)); + } + + EXPECT_EQ(socket.SendBatch(sends, 4, _FILE_AND_LINE_), 3); + ASSERT_EQ(socket.sent.size(), 3u); + EXPECT_EQ(socket.sent[0].port, 1000); + EXPECT_EQ(socket.sent[1].port, 1002) << "the datagram after the bad one must still go out"; + EXPECT_EQ(socket.sent[2].port, 1003); +} + TEST(SocketSendBatch, ForwardsPerDatagramTtl) { // sendmmsg has no per-message TTL, so RNS2_Linux::SendBatch defers a batch @@ -557,6 +621,41 @@ TEST(RNS2SendBatch, DropsOversizedDatagramsInsteadOfTruncating) #endif } +TEST(RNS2SendBatch, FailedFlushDropsTheBatchInsteadOfRetryingIt) +{ + // Flush clears the batch unconditionally. Carrying refused datagrams into + // the next flush would resend them behind whatever was queued meanwhile, + // reordering the stream; the reliability layer already handles the resend. + RecordingSocket socket; + { + RNS2SendBatch batch(&socket, MakeDest(1234)); + socket.failFrom = 0; + batch.Add("dropped", 7); + batch.Flush(); + EXPECT_TRUE(socket.sent.empty()); + + socket.failFrom = -1; + batch.Add("fresh", 5); + } + ASSERT_EQ(socket.sent.size(), 1u) << "the refused datagram must not be resent"; + EXPECT_EQ(socket.sent[0].payload, "fresh"); +} + +TEST(RNS2SendBatch, PartiallyFailedFlushStillDeliversTheRest) +{ + RecordingSocket socket; + socket.failOnly = 1; + { + RNS2SendBatch batch(&socket, MakeDest(1234)); + batch.Add("aa", 2); + batch.Add("bb", 2); + batch.Add("cc", 2); + } + ASSERT_EQ(socket.sent.size(), 2u); + EXPECT_EQ(socket.sent[0].payload, "aa"); + EXPECT_EQ(socket.sent[1].payload, "cc"); +} + TEST(RNS2SendBatch, EmptyBatchSendsNothing) { RecordingSocket socket; From 28925c0a0d7a8509bb4fe17c801001db94566bc9 Mon Sep 17 00:00:00 2001 From: Segfault <5221072+Segfaultd@users.noreply.github.com> Date: Mon, 27 Jul 2026 11:22:45 +0200 Subject: [PATCH 6/6] fix(mmsg): silence release-build drop spam, guard hot spins, test errno split Follow-up to the review of the batched datagram I/O branch. - RAKNET_DEBUG_PRINTF is a plain printf in every configuration, so the two dropped-datagram diagnostics in RNS2SendBatch wrote to stdout from the network thread in Release. A routine ENOBUFS burst on a loaded server turned them into unbounded, stdout-lock-taking spam inside the very loop batching exists to speed up. Both now go through MMSG_BATCH_DEBUG_PRINTF, which is debug-only -- matching the scalar send path, which ignores an individual sendto failure silently. - RecvFromBatchedLoop treated only n<0 as a failed recvmmsg pass. A pass returning 0 reset the backoff, dispatched nothing and slept not at all, i.e. spun the polling thread at 100%. Not expected under MSG_WAITFORONE, but the n<0 path was hardened against exactly this, so n<=0 takes it now. - Extracted the sendmmsg errno classification into ClassifySendmmsgErrno and unit-tested it. The transient (stop) vs. permanent (drop one, ship the rest) split is where getting it backwards loses traffic silently, and it was the one part of the glue no test covered -- the linux-mmsg CI job only ever exercises the happy path. - RNS2_Linux::SendBatch was guarded on MAFIANET_USE_SENDMMSG alone, but RNS2_Linux is the non-Windows socket class, so macOS and the BSDs compiled it too and failed on the missing mmsghdr/sendmmsg. Now requires __linux__ as well, so the flag is a no-op off Linux (portable base SendBatch) instead of a build break. Found by configuring the flags on macOS; no CI job covers that combination. Both suites pass in the default and the -DMAFIANET_USE_RECVMMSG=ON -DMAFIANET_USE_SENDMMSG=ON configurations (113/113 each). --- Source/include/mafianet/MmsgBatch.h | 52 +++++++++++++++++++-- Source/include/mafianet/socket2.h | 6 ++- Source/src/RakNetSocket2.cpp | 22 +++------ Source/src/RakNetSocket2_Berkley.cpp | 6 ++- Tests/Unit/MmsgBatchTests.cpp | 69 ++++++++++++++++++++++++++++ 5 files changed, 133 insertions(+), 22 deletions(-) diff --git a/Source/include/mafianet/MmsgBatch.h b/Source/include/mafianet/MmsgBatch.h index fb02a3fbc..964123c59 100644 --- a/Source/include/mafianet/MmsgBatch.h +++ b/Source/include/mafianet/MmsgBatch.h @@ -24,6 +24,20 @@ #include // memcpy #include // RAKNET_DEBUG_PRINTF defaults to printf +#include // error numbers classified by ClassifySendmmsgErrno + +/// Diagnostics for dropped datagrams, debug builds only. RAKNET_DEBUG_PRINTF +/// resolves to a plain printf in *every* configuration, and both call sites +/// below sit on the network thread's send path -- a routine ENOBUFS burst on a +/// loaded server would turn them into unbounded, stdout-lock-taking spam inside +/// the very loop batching exists to speed up. Staying silent in a shipping +/// build matches the scalar send path, which ignores an individual sendto +/// failure without a word. +#if defined(_DEBUG) +#define MMSG_BATCH_DEBUG_PRINTF(...) RAKNET_DEBUG_PRINTF(__VA_ARGS__) +#else +#define MMSG_BATCH_DEBUG_PRINTF(...) ((void) 0) +#endif namespace MafiaNet { @@ -102,6 +116,35 @@ int DriveBatchedSend(unsigned total, TransmitFn transmit) return (int) sent; } +/// Map the errno of a failed sendmmsg(2) onto DriveBatchedSend's transmit() +/// return contract. sendmmsg funnels two opposite conditions through the same +/// -1, so this split is the whole correctness of the batched send path: +/// - transient and socket-wide (the send buffer is full, or the call was +/// interrupted) -> 0, "no progress possible right now". The messages +/// themselves are fine; stop rather than burn a syscall per remaining +/// datagram. Reliable traffic is resent by the reliability layer on a later +/// tick. +/// - anything else (EMSGSIZE, EINVAL, EDESTADDRREQ, an ICMP error posted +/// against a prior send) -> -err, a property of the message at the current +/// offset. DriveBatchedSend drops that one datagram and still ships the +/// rest; treating it as "stop" would silently discard every datagram after +/// the bad one, which the scalar Send() loop would have delivered. +/// +/// Free function rather than inline in the override so the mapping is unit +/// testable without a real socket -- it is the one part of the sendmmsg glue +/// where getting it backwards loses traffic silently. +inline int ClassifySendmmsgErrno(int err) +{ + // A caller that hands us a non-error (errno unset, or a nonsensical + // negative) gets "stop": it cannot be attributed to a single message, and + // stopping can never spin. + if (err <= 0) + return 0; + if (err == EAGAIN || err == EWOULDBLOCK || err == ENOBUFS || err == EINTR) + return 0; + return -err; +} + /// Fan a received batch out to the event handler. /// /// \a slots[0..allocated) are pre-allocated recv structs whose .data buffers @@ -162,9 +205,10 @@ class RNS2SendBatch if (length < 0 || length > MAXIMUM_MTU_SIZE) { // Drop rather than corrupt. Reliable traffic is resent by the - // reliability layer; an unreliable datagram is simply lost, which is - // why the diagnostic below is the only trace it leaves. - RAKNET_DEBUG_PRINTF("RNS2SendBatch dropped an oversized datagram (%d bytes).\n", length); + // reliability layer; an unreliable datagram is simply lost. The + // RakAssert above and the diagnostic below are both debug-only, so in + // a shipping build this drop is silent by design. + MMSG_BATCH_DEBUG_PRINTF("RNS2SendBatch dropped an oversized datagram (%d bytes).\n", length); return; } if (count == MMSG_BATCH_MAX) @@ -195,7 +239,7 @@ class RNS2SendBatch // override cannot be unit-tested against, so check it at runtime here. RakAssert(sent <= (RNS2SendResult) count && "SendBatch must return a datagram count, not a byte total"); if (sent < 0 || (unsigned) sent < count) - RAKNET_DEBUG_PRINTF("SendBatch sent %d of %u datagrams.\n", (int) sent, count); + MMSG_BATCH_DEBUG_PRINTF("SendBatch sent %d of %u datagrams.\n", (int) sent, count); // Cleared unconditionally: a datagram the socket refused is dropped here, // not carried into the next flush. Reliable traffic is resent by the // reliability layer; retrying in place would reorder it against the diff --git a/Source/include/mafianet/socket2.h b/Source/include/mafianet/socket2.h index b09ace9b5..4ac9256e6 100644 --- a/Source/include/mafianet/socket2.h +++ b/Source/include/mafianet/socket2.h @@ -272,7 +272,11 @@ class RNS2_Linux : public RNS2_Berkley, public RNS2_Windows_Linux_360 public: RNS2BindResult Bind( RNS2_BerkleyBindParameters *bindParameters, const char *file, unsigned int line ); RNS2SendResult Send( RNS2_SendParameters *sendParameters, const char *file, unsigned int line ); -#if defined(MAFIANET_USE_SENDMMSG) + // __linux__ as well as the flag: RNS2_Linux is the non-Windows socket class, + // so macOS and the BSDs compile it too, and sendmmsg/mmsghdr do not exist + // there. Those platforms keep the portable base SendBatch (a Send() loop), + // which is why turning the flag on off-Linux is a no-op rather than an error. +#if defined(MAFIANET_USE_SENDMMSG) && defined(__linux__) RNS2SendResult SendBatch( RNS2_SendParameters *sends, unsigned count, const char *file, unsigned int line ); #endif diff --git a/Source/src/RakNetSocket2.cpp b/Source/src/RakNetSocket2.cpp index bdf9b2999..d03b5e4b8 100644 --- a/Source/src/RakNetSocket2.cpp +++ b/Source/src/RakNetSocket2.cpp @@ -291,7 +291,9 @@ SocketLayerOverride* RNS2_Windows::GetSocketLayerOverride(void) {return slo;} #else RNS2BindResult RNS2_Linux::Bind( RNS2_BerkleyBindParameters *bindParameters, const char *file, unsigned int line ) {return BindShared(bindParameters, file, line);} RNS2SendResult RNS2_Linux::Send( RNS2_SendParameters *sendParameters, const char *file, unsigned int line ) {return Send_Windows_Linux_360NoVDP(rns2Socket,sendParameters, file, line);} -#if defined(MAFIANET_USE_SENDMMSG) +// See the declaration in socket2.h for why __linux__ is required here and not +// just the build flag. +#if defined(MAFIANET_USE_SENDMMSG) && defined(__linux__) RNS2SendResult RNS2_Linux::SendBatch( RNS2_SendParameters *sends, unsigned count, const char *file, unsigned int line ) { // sendmmsg has no per-message TTL, whereas the scalar Send() honours @@ -313,8 +315,9 @@ RNS2SendResult RNS2_Linux::SendBatch( RNS2_SendParameters *sends, unsigned count // coalesces up to MMSG_BATCH_MAX datagrams into one sendmmsg. sendmmsg // returns the number of messages sent (>=1) or -1 (nothing sent). It funnels // both transient socket-wide conditions and permanent per-message failures - // through that same -1, so classify errno before handing it back: the two - // mean opposite things to DriveBatchedSend (stop vs. drop one and continue). + // through that same -1, so ClassifySendmmsgErrno (unit-tested in + // MmsgBatchTests) splits them before handing the result back: the two mean + // opposite things to DriveBatchedSend (stop vs. drop one and continue). const RNS2SendResult sent = DriveBatchedSend(count, [&](unsigned offset, unsigned remaining) -> int { @@ -351,18 +354,7 @@ RNS2SendResult RNS2_Linux::SendBatch( RNS2_SendParameters *sends, unsigned count const int r = sendmmsg(rns2Socket, msgs, chunk, 0); if (r>=0) return r; - const int err = errno; - // Transient and socket-wide: the send buffer is full or the call was - // interrupted. Nothing in this batch can go out right now, and the - // messages themselves are fine -- stop rather than burn a syscall per - // remaining datagram. Reliable traffic is resent by the reliability - // layer on a later tick. - if (err==EAGAIN || err==EWOULDBLOCK || err==ENOBUFS || err==EINTR) - return 0; - // Anything else (EMSGSIZE, EINVAL, EDESTADDRREQ, an ICMP error posted - // against a prior send) is a property of the message at `offset`. - // Report it as a per-message failure so the remainder still ships. - return -err; + return ClassifySendmmsgErrno(errno); }); return sent; } diff --git a/Source/src/RakNetSocket2_Berkley.cpp b/Source/src/RakNetSocket2_Berkley.cpp index 32c5f4f64..7a7270ec7 100644 --- a/Source/src/RakNetSocket2_Berkley.cpp +++ b/Source/src/RakNetSocket2_Berkley.cpp @@ -590,9 +590,11 @@ void RNS2_Berkley::RecvFromBatchedLoop(void) int n = recvmmsg(rns2Socket, msgs, allocated, MSG_WAITFORONE, nullptr); MafiaNet::TimeUS now = MafiaNet::GetTimeUS(); - if (n<0) + if (n<=0) { - // Interrupted or an asynchronous socket error. (Shutdown does not + // Interrupted, an asynchronous socket error, or -- not expected under + // MSG_WAITFORONE, but not worth a hot spin if a kernel ever does it -- + // a pass that returned no datagrams at all. (Shutdown does not // come through here: BlockOnStopRecvPollingThread pokes the socket // with a real datagram, so it surfaces as a normal n>=1 pass and the // loop condition below catches endThreads.) Keep the batch and back diff --git a/Tests/Unit/MmsgBatchTests.cpp b/Tests/Unit/MmsgBatchTests.cpp index 3c6bb62c8..cd89cb025 100644 --- a/Tests/Unit/MmsgBatchTests.cpp +++ b/Tests/Unit/MmsgBatchTests.cpp @@ -246,6 +246,75 @@ TEST(DriveBatchedSend, EmptyBatchNeverCallsTransmit) EXPECT_EQ(calls, 0); } +// --------------------------------------------------------------------------- +// ClassifySendmmsgErrno -- the transient-vs-permanent split the RNS2_Linux +// sendmmsg override feeds into DriveBatchedSend. Getting this backwards loses +// traffic silently (a permanent error read as "stop" discards the whole batch +// tail; a transient one read as "drop" throws away a healthy datagram), and the +// syscall itself cannot be provoked from a unit test -- hence testing the +// mapping directly. +// --------------------------------------------------------------------------- + +TEST(ClassifySendmmsgErrno, TransientSocketWideErrorsStopTheBatch) +{ + // 0 == "no progress possible right now", so DriveBatchedSend stops without + // dropping anything. EAGAIN and EWOULDBLOCK are the same value on Linux; + // both are listed because that is not guaranteed everywhere. + EXPECT_EQ(ClassifySendmmsgErrno(EAGAIN), 0); + EXPECT_EQ(ClassifySendmmsgErrno(EWOULDBLOCK), 0); + EXPECT_EQ(ClassifySendmmsgErrno(ENOBUFS), 0); + EXPECT_EQ(ClassifySendmmsgErrno(EINTR), 0); +} + +TEST(ClassifySendmmsgErrno, PermanentPerMessageErrorsDropOnlyThatDatagram) +{ + // A negative return names the message at the current offset as undeliverable, + // so DriveBatchedSend drops it and still ships the rest of the batch. + EXPECT_EQ(ClassifySendmmsgErrno(EMSGSIZE), -EMSGSIZE); + EXPECT_EQ(ClassifySendmmsgErrno(EINVAL), -EINVAL); + EXPECT_EQ(ClassifySendmmsgErrno(EDESTADDRREQ), -EDESTADDRREQ); + EXPECT_EQ(ClassifySendmmsgErrno(ECONNREFUSED), -ECONNREFUSED); +} + +TEST(ClassifySendmmsgErrno, UnsetErrnoStopsRatherThanSpinning) +{ + // errno left at 0 cannot be attributed to a single message, and -0 == 0 would + // read as "stop" anyway. Pinned so it can never become a value that makes + // DriveBatchedSend loop. + EXPECT_EQ(ClassifySendmmsgErrno(0), 0); + EXPECT_EQ(ClassifySendmmsgErrno(-1), 0); +} + +TEST(ClassifySendmmsgErrno, DrivesTheBatchTailPastAPermanentFailure) +{ + // The end-to-end reason the split exists: a permanent error on the first + // message must not cost the remaining four. + std::vector offsets; + int sent = DriveBatchedSend(5, [&](unsigned offset, unsigned remaining) { + offsets.push_back(offset); + if (offset == 0) + return ClassifySendmmsgErrno(EMSGSIZE); + return (int) remaining; + }); + EXPECT_EQ(sent, 4); + EXPECT_EQ(offsets, (std::vector{0, 1})); +} + +TEST(ClassifySendmmsgErrno, StopsTheBatchTailOnATransientFailure) +{ + // The mirror image: a full send buffer must not be mistaken for a bad + // datagram, which would drop one healthy message per remaining slot. + std::vector offsets; + int sent = DriveBatchedSend(5, [&](unsigned offset, unsigned) { + offsets.push_back(offset); + if (offset == 0) + return 2; // two accepted, then the buffer fills up + return ClassifySendmmsgErrno(ENOBUFS); + }); + EXPECT_EQ(sent, 2); + EXPECT_EQ(offsets, (std::vector{0, 2})); +} + // --------------------------------------------------------------------------- // SockaddrToSystemAddress -- byte-order handling. // ---------------------------------------------------------------------------