Commit 2841890e authored by jan.koester's avatar jan.koester
Browse files

perf

parent 09d8e029

src/posix/udp.cpp

+204 −60
Original line number Diff line number Diff line
@@ -428,6 +428,7 @@ ssize_t udp::sendBatch( 
    if (_Socket < 0) return -1;



    const size_t count = datagrams.size();

    auto& perf = quicPerf();



    // --- GSO path: coalesce uniform-sized datagrams into one sendmsg ---

    if (_gso_enabled && count > 1) {
@@ -449,7 +450,14 @@ ssize_t udp::sendBatch( 
            // array as a whole — it does not require the source bytes to

            // already be contiguous in user space. This removes one full

            // copy of every batch's total bytes on the GSO path.

            std::vector<struct iovec> iov(count);

            //

            // iovec array itself: thread_local, reused across calls instead

            // of a fresh std::vector<iovec>(count) every flush — mirrors

            // recvBatchAddr()'s thread_local iovec/mmsghdr arrays below.

            // resize() never shrinks capacity, so after the first call at

            // the connection's steady-state batch size this is alloc-free.

            static thread_local std::vector<struct iovec> iov;

            if (iov.size() < count) iov.resize(count);

            for (size_t i = 0; i < count; ++i) {

                iov[i].iov_base = const_cast<uint8_t*>(datagrams[i].first);

                iov[i].iov_len  = datagrams[i].second;
@@ -467,7 +475,7 @@ ssize_t udp::sendBatch( 
                msg.msg_namelen = dest_len;

            }

            msg.msg_iov        = iov.data();

            msg.msg_iovlen     = iov.size();

            msg.msg_iovlen     = count;

            msg.msg_control    = cmsg_buf.buf;

            msg.msg_controllen = sizeof(cmsg_buf.buf);
@@ -479,9 +487,18 @@ ssize_t udp::sendBatch( 
                static_cast<uint16_t>(seg_size);



            ssize_t ret = ::sendmsg(_Socket, &msg, MSG_DONTWAIT);

            if (ret > 0) return static_cast<ssize_t>(count);

            if (ret > 0) {

                perf.gso_sendmsg_calls.fetch_add(1, std::memory_order_relaxed);

                perf.gso_segments.fetch_add(count, std::memory_order_relaxed);

                perf.packets_sent.fetch_add(count, std::memory_order_relaxed);

                perf.bytes_sent.fetch_add(static_cast<uint64_t>(ret), std::memory_order_relaxed);

                return static_cast<ssize_t>(count);

            }

            if (errno == EAGAIN || errno == EWOULDBLOCK)

                perf.send_eagain.fetch_add(1, std::memory_order_relaxed);

            // GSO failed — fall through to sendmmsg

        }

        perf.gso_fallbacks.fetch_add(1, std::memory_order_relaxed);

    }



    // --- sendmmsg path ---
@@ -502,14 +519,24 @@ ssize_t udp::sendBatch( 
            r = ::send(_Socket, datagrams[i].first, datagrams[i].second, MSG_DONTWAIT);

        }

        if (r < 0) {

            if (errno == EAGAIN || errno == EWOULDBLOCK)

                perf.send_eagain.fetch_add(1, std::memory_order_relaxed);

            if (sent == 0) return -1;

            break;

        }

        perf.bytes_sent.fetch_add(static_cast<uint64_t>(r), std::memory_order_relaxed);

        ++sent;

    }

    perf.single_send_calls.fetch_add(static_cast<uint64_t>(sent), std::memory_order_relaxed);

    perf.packets_sent.fetch_add(static_cast<uint64_t>(sent), std::memory_order_relaxed);

#else

    std::vector<struct iovec> iovecs(count);

    std::vector<struct mmsghdr> msgs(count);

    // thread_local, reused across calls (same rationale as the GSO iovec

    // array above) — avoids two fresh heap allocations (iovec + mmsghdr

    // arrays) on every flushBatch() call.

    static thread_local std::vector<struct iovec> iovecs;

    static thread_local std::vector<struct mmsghdr> msgs;

    if (iovecs.size() < count) iovecs.resize(count);

    if (msgs.size() < count) msgs.resize(count);

    std::memset(msgs.data(), 0, sizeof(struct mmsghdr) * count);



    for (size_t i = 0; i < count; ++i) {
@@ -525,6 +552,15 @@ ssize_t udp::sendBatch( 


    int sent = ::sendmmsg(_Socket, msgs.data(),

                          static_cast<unsigned int>(count), MSG_DONTWAIT);

    perf.sendmmsg_calls.fetch_add(1, std::memory_order_relaxed);

    if (sent < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) {

        perf.send_eagain.fetch_add(1, std::memory_order_relaxed);

    } else if (sent > 0) {

        perf.packets_sent.fetch_add(static_cast<uint64_t>(sent), std::memory_order_relaxed);

        uint64_t bytes = 0;

        for (int i = 0; i < sent; ++i) bytes += datagrams[static_cast<size_t>(i)].second;

        perf.bytes_sent.fetch_add(bytes, std::memory_order_relaxed);

    }

#endif

    return sent;

}
@@ -538,6 +574,73 @@ size_t udp::recvBatch(std::vector<std::vector<uint8_t>>& out, int max_count) { 
    return recvBatchAddr(out, dummy_addrs, max_count);

}



#ifndef NETPLUS_NO_MMSG

namespace {

// Shared by recvBatchAddr() (owning-vector API, unchanged behavior) and

// recvBatchAddrViews() (zero-copy API) below: performs the actual

// recvmmsg() call into thread_local buffers and leaves the raw per-slot

// results (`msgs`/`flat_buf`/`peer_addrs`) populated for the caller to walk.

// Splitting this out means both APIs share one recvmmsg() call site instead

// of drifting apart, and neither has to re-declare the (sizeable) thread_local

// scratch state itself.

//

// Per-slot size is 65535, not the ~1472-byte PMTU ceiling a QUIC connection

// actually negotiates: this buffer is shared (thread_local) across every

// udp/quic instance serviced by this thread, and when GRO is enabled, a

// single recvmmsg slot can hold a UDP_GRO coalesced batch of multiple

// datagrams — up to the kernel's ~64KB GRO limit, not one MTU-sized packet.

// Sizing slots down to PMTU+margin would silently truncate/corrupt any

// GRO-coalesced receive on a thread that also happens to service a

// GRO-enabled socket. Safe to shrink only if GRO support were removed, or

// reworked to size per-socket instead of per-thread.

constexpr int kRecvMaxBatch = 64;

thread_local std::vector<uint8_t> g_recv_flat_buf(kRecvMaxBatch * 65535);

thread_local struct iovec g_recv_iovecs[kRecvMaxBatch];

thread_local struct mmsghdr g_recv_msgs[kRecvMaxBatch];

thread_local sockaddr_storage g_recv_peer_addrs[kRecvMaxBatch];

constexpr size_t kRecvCmsgBufSize = CMSG_SPACE(sizeof(uint16_t));

thread_local char g_recv_cmsg_bufs[kRecvMaxBatch * kRecvCmsgBufSize];



int recvBatchRaw(SOCKET sock, int batch, bool gro_enabled) {

    std::memset(g_recv_msgs, 0, sizeof(struct mmsghdr) * batch);

    std::memset(g_recv_peer_addrs, 0, sizeof(sockaddr_storage) * batch);



    for (int i = 0; i < batch; ++i) {

        g_recv_iovecs[i].iov_base = g_recv_flat_buf.data() + i * 65535;

        g_recv_iovecs[i].iov_len  = 65535;

        g_recv_msgs[i].msg_hdr.msg_iov     = &g_recv_iovecs[i];

        g_recv_msgs[i].msg_hdr.msg_iovlen  = 1;

        g_recv_msgs[i].msg_hdr.msg_name    = &g_recv_peer_addrs[i];

        g_recv_msgs[i].msg_hdr.msg_namelen = sizeof(sockaddr_storage);

        if (gro_enabled) {

            g_recv_msgs[i].msg_hdr.msg_control    = &g_recv_cmsg_bufs[i * kRecvCmsgBufSize];

            g_recv_msgs[i].msg_hdr.msg_controllen = kRecvCmsgBufSize;

        }

    }



    struct timespec timeout = {0, 0}; // non-blocking

    int received = ::recvmmsg(sock, g_recv_msgs, batch, MSG_DONTWAIT, &timeout);

    quicPerf().recvmmsg_calls.fetch_add(1, std::memory_order_relaxed);

    if (received < 0 && (errno == EAGAIN || errno == EWOULDBLOCK))

        quicPerf().recv_eagain.fetch_add(1, std::memory_order_relaxed);

    return received;

}



// Returns the UDP_GRO segment size for slot `i` (0 if none/not coalesced).

uint16_t recvBatchGroSegSize(int i, bool gro_enabled) {

    if (!gro_enabled) return 0;

    for (struct cmsghdr* cm = CMSG_FIRSTHDR(&g_recv_msgs[i].msg_hdr);

         cm != nullptr;

         cm = CMSG_NXTHDR(&g_recv_msgs[i].msg_hdr, cm)) {

        if (cm->cmsg_level == SOL_UDP && cm->cmsg_type == UDP_GRO) {

            return *reinterpret_cast<uint16_t*>(CMSG_DATA(cm));

        }

    }

    return 0;

}

} // namespace

#endif // NETPLUS_NO_MMSG



size_t udp::recvBatchAddr(std::vector<std::vector<uint8_t>>& out,

                           std::vector<sockaddr_storage>& addrs,

                           int max_count) {
@@ -562,65 +665,15 @@ size_t udp::recvBatchAddr(std::vector<std::vector<uint8_t>>& out, 
    }

    return out.size();

#else

    // Use thread_local static buffers to avoid 64×65KB heap allocation per call.

    // These are reused across calls on the same thread.

    //

    // Per-slot size is 65535, not the ~1472-byte PMTU ceiling a QUIC

    // connection actually negotiates: this buffer is shared (thread_local)

    // across every udp/quic instance serviced by this thread, and when

    // _gro_enabled is set, a single recvmmsg slot can hold a UDP_GRO

    // coalesced batch of multiple datagrams — up to the kernel's ~64KB GRO

    // limit, not one MTU-sized packet. Sizing slots down to PMTU+margin

    // would silently truncate/corrupt any GRO-coalesced receive on a thread

    // that also happens to service a GRO-enabled socket. Safe to shrink only

    // if GRO support were removed, or reworked to size per-socket instead of

    // per-thread.

    static constexpr int MAX_BATCH = 64;

    static thread_local std::vector<uint8_t> flat_buf(MAX_BATCH * 65535);

    static thread_local struct iovec iovecs[MAX_BATCH];

    static thread_local struct mmsghdr msgs[MAX_BATCH];

    static thread_local sockaddr_storage peer_addrs[MAX_BATCH];

    static constexpr size_t CMSG_BUF_SIZE = CMSG_SPACE(sizeof(uint16_t));

    static thread_local char cmsg_bufs[MAX_BATCH * CMSG_BUF_SIZE];



    std::memset(msgs, 0, sizeof(struct mmsghdr) * batch);

    std::memset(peer_addrs, 0, sizeof(sockaddr_storage) * batch);



    for (int i = 0; i < batch; ++i) {

        iovecs[i].iov_base = flat_buf.data() + i * 65535;

        iovecs[i].iov_len  = 65535;

        msgs[i].msg_hdr.msg_iov     = &iovecs[i];

        msgs[i].msg_hdr.msg_iovlen  = 1;

        msgs[i].msg_hdr.msg_name    = &peer_addrs[i];

        msgs[i].msg_hdr.msg_namelen = sizeof(sockaddr_storage);

        if (_gro_enabled) {

            msgs[i].msg_hdr.msg_control    = &cmsg_bufs[i * CMSG_BUF_SIZE];

            msgs[i].msg_hdr.msg_controllen = CMSG_BUF_SIZE;

        }

    }



    struct timespec timeout = {0, 0}; // non-blocking

    int received = ::recvmmsg(_Socket, msgs, batch, MSG_DONTWAIT, &timeout);

    int received = recvBatchRaw(_Socket, batch, _gro_enabled);

    if (received <= 0) return 0;



    for (int i = 0; i < received; ++i) {

        size_t total_len = msgs[i].msg_len;

        size_t total_len = g_recv_msgs[i].msg_len;

        if (total_len == 0) continue;



        uint8_t* base = flat_buf.data() + i * 65535;



        // Check for GRO coalescing

        uint16_t gro_seg = 0;

        if (_gro_enabled) {

            for (struct cmsghdr* cm = CMSG_FIRSTHDR(&msgs[i].msg_hdr);

                 cm != nullptr;

                 cm = CMSG_NXTHDR(&msgs[i].msg_hdr, cm)) {

                if (cm->cmsg_level == SOL_UDP && cm->cmsg_type == UDP_GRO) {

                    gro_seg = *reinterpret_cast<uint16_t*>(CMSG_DATA(cm));

                    break;

                }

            }

        }

        uint8_t* base = g_recv_flat_buf.data() + i * 65535;

        uint16_t gro_seg = recvBatchGroSegSize(i, _gro_enabled);



        if (gro_seg > 0 && total_len > gro_seg) {

            // Split coalesced GRO buffer into individual datagrams
@@ -629,13 +682,104 @@ size_t udp::recvBatchAddr(std::vector<std::vector<uint8_t>>& out, 
                size_t seg_len = std::min(static_cast<size_t>(gro_seg),

                                          total_len - off);

                out.emplace_back(base + off, base + off + seg_len);

                addrs.push_back(peer_addrs[i]);

                addrs.push_back(g_recv_peer_addrs[i]);

                off += seg_len;

                quicPerf().gro_packets_split.fetch_add(1, std::memory_order_relaxed);

            }

        } else {

            out.emplace_back(base, base + total_len);

            addrs.push_back(peer_addrs[i]);

            addrs.push_back(g_recv_peer_addrs[i]);

        }

    }



    if (!out.empty()) {

        auto& perf = quicPerf();

        perf.recv_batch_calls.fetch_add(1, std::memory_order_relaxed);

        perf.recv_batch_packets.fetch_add(out.size(), std::memory_order_relaxed);

        perf.packets_recv.fetch_add(out.size(), std::memory_order_relaxed);

        uint64_t bytes = 0;

        for (auto& dgram : out) bytes += dgram.size();

        perf.bytes_recv.fetch_add(bytes, std::memory_order_relaxed);

        uint64_t prev_max = perf.recv_batch_max.load(std::memory_order_relaxed);

        while (out.size() > prev_max &&

               !perf.recv_batch_max.compare_exchange_weak(prev_max, out.size(), std::memory_order_relaxed)) {}

    }



    return out.size();

#endif // NETPLUS_NO_MMSG

}



size_t udp::recvBatchViews(std::vector<DatagramView>& out, int max_count) {

    std::vector<sockaddr_storage> dummy_addrs;

    return recvBatchAddrViews(out, dummy_addrs, max_count);

}



size_t udp::recvBatchAddrViews(std::vector<DatagramView>& out,

                                std::vector<sockaddr_storage>& addrs,

                                int max_count) {

    out.clear();

    addrs.clear();

    if (_Socket < 0) return 0;



    const int batch = std::min(max_count, 64);



#ifdef NETPLUS_NO_MMSG

    // Fallback: recvfrom loop, straight into a thread_local per-slot buffer

    // pool so each DatagramView stays valid until the next call on this

    // thread — same lifetime contract as the recvmmsg path below.

    static thread_local std::vector<std::array<uint8_t, 65535>> slots;

    if (slots.size() < static_cast<size_t>(batch)) slots.resize(batch);

    for (int i = 0; i < batch; ++i) {

        sockaddr_storage peer{};

        socklen_t peer_len = sizeof(peer);

        ssize_t n = ::recvfrom(_Socket, slots[i].data(), slots[i].size(),

                               MSG_DONTWAIT,

                               reinterpret_cast<sockaddr*>(&peer), &peer_len);

        if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK))

            quicPerf().recv_eagain.fetch_add(1, std::memory_order_relaxed);

        if (n <= 0) break;

        out.push_back({slots[i].data(), static_cast<size_t>(n)});

        addrs.push_back(peer);

    }

    return out.size();

#else

    int received = recvBatchRaw(_Socket, batch, _gro_enabled);

    if (received <= 0) return 0;



    for (int i = 0; i < received; ++i) {

        size_t total_len = g_recv_msgs[i].msg_len;

        if (total_len == 0) continue;



        uint8_t* base = g_recv_flat_buf.data() + i * 65535;

        uint16_t gro_seg = recvBatchGroSegSize(i, _gro_enabled);



        if (gro_seg > 0 && total_len > gro_seg) {

            size_t off = 0;

            while (off < total_len) {

                size_t seg_len = std::min(static_cast<size_t>(gro_seg),

                                          total_len - off);

                out.push_back({base + off, seg_len});

                addrs.push_back(g_recv_peer_addrs[i]);

                off += seg_len;

                quicPerf().gro_packets_split.fetch_add(1, std::memory_order_relaxed);

            }

        } else {

            out.push_back({base, total_len});

            addrs.push_back(g_recv_peer_addrs[i]);

        }

    }



    if (!out.empty()) {

        auto& perf = quicPerf();

        perf.recv_batch_calls.fetch_add(1, std::memory_order_relaxed);

        perf.recv_batch_packets.fetch_add(out.size(), std::memory_order_relaxed);

        perf.packets_recv.fetch_add(out.size(), std::memory_order_relaxed);

        uint64_t bytes = 0;

        for (auto& dv : out) bytes += dv.len;

        perf.bytes_recv.fetch_add(bytes, std::memory_order_relaxed);

        uint64_t prev_max = perf.recv_batch_max.load(std::memory_order_relaxed);

        while (out.size() > prev_max &&

               !perf.recv_batch_max.compare_exchange_weak(prev_max, out.size(), std::memory_order_relaxed)) {}

    }



    return out.size();

src/quic.cpp

+289 −31

File changed.

Preview size limit exceeded, changes collapsed.

src/socket.h

+170 −0
Original line number Diff line number Diff line
@@ -164,6 +164,34 @@ namespace netplus { 
	};

#endif



	// Non-owning view of one received UDP datagram: a pointer into a

	// per-thread reusable receive buffer (see udp::recvBatchAddrViews()) plus

	// its length. Exists so the hot receive path can hand off datagrams

	// without heap-allocating one std::vector<uint8_t> per packet the way

	// recvBatch()/recvBatchAddr() do (kept as-is for API compatibility —

	// see their own comments). Valid only until the next recvBatch*() call

	// on the same thread; every call site that uses this type fully

	// processes the batch (or moves any bytes it needs to keep into

	// connection-owned storage) before the next receive call, exactly the

	// lifetime recvBatchAddr()'s own owning-vector callers already rely on.

	//

	// Deliberately mimics std::vector<uint8_t>'s read-only interface

	// (data()/size()/empty()/operator[]/begin()/end()) rather than exposing

	// `ptr`/`len` directly, so call sites written against the owning-vector

	// recvBatchAddr() (indexing, slicing into a sub-range via two iterators,

	// etc.) port over to the zero-copy path unchanged.

	struct DatagramView {

		const uint8_t* ptr = nullptr;

		size_t len = 0;



		const uint8_t* data() const { return ptr; }

		size_t size() const { return len; }

		bool empty() const { return len == 0; }

		const uint8_t& operator[](size_t i) const { return ptr[i]; }

		const uint8_t* begin() const { return ptr; }

		const uint8_t* end() const { return ptr + len; }

	};



	// ===================================================

	// ✅ SOCKET BASE CLASS (refactored)

	// ===================================================
@@ -400,6 +428,28 @@ namespace netplus { 
		                     std::vector<sockaddr_storage>& addrs,

		                     int max_count = 64);



		// Zero-copy siblings of recvBatch()/recvBatchAddr(): same recvmmsg+GRO

		// receive, but |out| is filled with DatagramView (pointer, length)

		// entries pointing directly into a per-thread reusable buffer instead

		// of one heap-allocated std::vector<uint8_t> per datagram. Eliminates

		// the dominant allocation source on the hot receive path (measured:

		// one heap alloc+copy per QUIC packet received, at up to several

		// hundred thousand packets/sec under sustained bulk transfer).

		//

		// Lifetime: every entry in |out| is valid only until the next

		// recvBatch*()/recvBatchAddr*() call on THIS thread (the underlying

		// storage is thread_local and reused/overwritten then) — callers

		// must fully consume |out| (decrypt/route/process, or copy out any

		// bytes that must outlive the batch) before calling any recv*()

		// variant again from the same thread. recvBatch()/recvBatchAddr()

		// above keep their original owning-vector contract unchanged for

		// existing/external callers; these views are for quic.cpp's internal

		// hot paths, which already respect exactly this lifetime.

		size_t recvBatchViews(std::vector<DatagramView>& out, int max_count = 64);

		size_t recvBatchAddrViews(std::vector<DatagramView>& out,

		                          std::vector<sockaddr_storage>& addrs,

		                          int max_count = 64);



		bool gsoEnabled() const { return _gso_enabled; }

		bool groEnabled() const { return _gro_enabled; }
@@ -528,6 +578,62 @@ namespace netplus { 
		friend class TestSSL;

	};



	// ============================================================================

	// Optional QUIC performance instrumentation (opt-in via QUIC_PERF=1).

	// Process-wide (not per-connection) counters updated with relaxed atomic

	// increments from the send/receive hot paths in quic.cpp/posix/udp.cpp —

	// cheap enough (a handful of ns) to leave permanently enabled rather than

	// branching on whether reporting is on; only the periodic formatting/

	// printing (see quicPerfReportingEnabled() and its reporter thread in

	// quic.cpp) is gated behind the env var. Diagnostic tool, not part of

	// protocol correctness: nothing in the send/receive/CC/retransmission

	// logic reads these values back.

	struct QuicPerfCounters {

		std::atomic<uint64_t> gso_sendmsg_calls{0};

		std::atomic<uint64_t> gso_segments{0};

		std::atomic<uint64_t> gso_fallbacks{0};      // GSO attempted but not uniform / not enabled

		std::atomic<uint64_t> sendmmsg_calls{0};

		std::atomic<uint64_t> single_send_calls{0};  // sendto()/send() fallback (no mmsg/GSO)

		std::atomic<uint64_t> recvmmsg_calls{0};

		std::atomic<uint64_t> gro_packets_split{0};

		std::atomic<uint64_t> send_eagain{0};

		std::atomic<uint64_t> recv_eagain{0};

		std::atomic<uint64_t> bytes_sent{0};

		std::atomic<uint64_t> bytes_recv{0};

		std::atomic<uint64_t> packets_sent{0};

		std::atomic<uint64_t> packets_recv{0};

		std::atomic<uint64_t> send_batch_flushes{0};   // flushBatch() calls that sent >=1 datagram

		std::atomic<uint64_t> send_batch_packets{0};   // sum of datagrams across those flushes

		std::atomic<uint64_t> send_batch_max{0};

		std::atomic<uint64_t> recv_batch_calls{0};     // recvBatch*() calls that returned >=1 datagram

		std::atomic<uint64_t> recv_batch_packets{0};

		std::atomic<uint64_t> recv_batch_max{0};

		std::atomic<uint64_t> packets_lost{0};

		std::atomic<uint64_t> retransmits{0};

		std::atomic<uint64_t> flow_control_stalls{0};  // DATA_BLOCKED/STREAM_DATA_BLOCKED sent

		std::atomic<uint64_t> congestion_stalls{0};    // cwndAllowsSend() block entries

		std::atomic<uint64_t> pumps_done{0};           // pumpIncomingLocked() invocations from the send loop

		std::atomic<uint64_t> pumps_skipped{0};        // adaptive-pump skips (had cwnd headroom)



		// Diagnostic snapshot of the most recently active connection's

		// congestion/RTT state — not authoritative across multiple

		// concurrent connections, but sufficient to see whether cwnd/RTT/

		// loss is the limiting factor for whichever connection is actually

		// under test (the common QUIC_PERF use case: one benchmark/transfer

		// at a time). Stored as integer microseconds/bytes rather than

		// std::atomic<double> so plain relaxed load/store suffices.

		std::atomic<uint64_t> last_cwnd{0};

		std::atomic<uint64_t> last_bytes_in_flight{0};

		std::atomic<uint64_t> last_ssthresh{0};

		std::atomic<uint64_t> last_rtt_us{0};

		std::atomic<uint64_t> last_min_rtt_us{0};

	};



	inline QuicPerfCounters& quicPerf() {

		static QuicPerfCounters counters;

		return counters;

	}



	// ===================================================

	// QUIC Protocol Implementation (RFC 9000)

	// UDP-based transport with TLS 1.3 encryption
@@ -711,6 +817,48 @@ namespace netplus { 
		// probed something larger.

		uint64_t getMaxUdpPayload() const { return _max_udp_payload; }



		// Opt-in resize of the OS-level SO_RCVBUF/SO_SNDBUF beyond the

		// constructors' conservative 4MB default (chosen so a server with

		// many concurrent connections doesn't multiply a large per-connection

		// buffer by connection count). Call after the socket exists — for a

		// client, that means after connect(); for a server/accepted

		// connection, any time. Values above net.core.rmem_max/wmem_max are

		// silently clamped by the kernel (Linux never fails SO_RCVBUF/

		// SO_SNDBUF for being "too large") rather than rejected — use

		// getSocketBufferSizes() afterward to see what was actually applied.

		// Never throws: a request the OS won't honor just leaves the

		// previous size in effect.

		void setSocketBufferSizes(size_t rcvbuf_bytes, size_t sndbuf_bytes);

		// Actual, OS-confirmed current sizes (post any clamping) — 0/0 if

		// the socket doesn't exist yet or the query itself failed.

		void getSocketBufferSizes(size_t& rcvbuf_bytes, size_t& sndbuf_bytes) const;



		// Flow control (RFC 9000 §4): raises this connection's advertised

		// receive-window credit. flushPendingFlowControlAndCheckLoss()

		// grants a fresh increment of this size via MAX_DATA each time the

		// peer has consumed about half of what's currently committed — this

		// isn't a one-time hard cap, but a bigger value means bigger (and

		// therefore less frequent) credit grants, which matters on

		// high-bandwidth-delay-product paths where the default (64MB)

		// isn't enough standing credit to keep a fast sender from stalling

		// on connection-level flow control between grants. Must be called

		// before any data has been received on this connection: client,

		// right after construction and before connect(); server, on the

		// LISTENER before it accepts anything (copied to every accepted

		// child the same way setPmtuCeiling()'s ceiling is — see accept()).

		void setConnectionRecvWindow(uint64_t bytes) {

			_max_data_local = bytes;

			_max_data_local_committed = bytes;

		}

		uint64_t getConnectionRecvWindow() const { return _max_data_local; }



		// Same idea, applied to every newly created stream (openStream()

		// and a peer-initiated stream's first frame) from this point

		// onward — does not retroactively change streams that already

		// exist. Default (16MB) matches the historical per-stream constant.

		void setDefaultStreamRecvWindow(uint64_t bytes) { _stream_recv_window_default = bytes; }

		uint64_t getDefaultStreamRecvWindow() const { return _stream_recv_window_default; }



		// Stream management

		uint64_t openStream(bool bidirectional = true);

		void closeStream(uint64_t stream_id);
@@ -1520,6 +1668,12 @@ namespace netplus { 
		uint64_t _max_data_local_committed = 67108864; // last advertised

		uint64_t _data_sent = 0;

		uint64_t _data_recv = 0;

		// Per-stream receive-window default applied to every newly created

		// Stream (openStream() and a peer-initiated stream's first frame —

		// see setDefaultStreamRecvWindow()'s comment). Mirrors Stream::

		// max_data_local's own historical default so leaving this untouched

		// changes nothing.

		uint64_t _stream_recv_window_default = 16777216;



		// Stream count limits (MAX_STREAMS)

		uint64_t _max_streams_bidi_local = 10000000;   // we allow peer to open
@@ -2068,6 +2222,22 @@ namespace netplus { 
		std::vector<uint8_t> _send_arena;

		size_t _send_arena_used = 0;

		std::vector<BatchEntry> _send_batch;

		// Adaptive pumpIncomingLocked() trigger in sendStreamData()'s main

		// loop: counts flushBatch() cycles since the last ACK/flow-control

		// pump. See its call site's comment for the full rationale — in

		// short, skip the pump while there's ample cwnd headroom (the

		// cwnd-block path already pumps when it's actually needed) but never

		// let more than PUMP_ADAPTIVE_MAX_BATCHES flushes go by without one,

		// so RTT samples/PTO timers/flow-control credit don't go stale on a

		// connection whose cwnd happens to be very large.

		size_t _flushes_since_pump = 0;

		static constexpr size_t PUMP_ADAPTIVE_MAX_BATCHES = 4;

		// Scratch pointer/length pairs built from _send_batch/_send_arena on

		// every flushBatch() call, reused instead of a fresh

		// std::vector<std::pair<...>> per call — resize() never shrinks

		// capacity, so after the first flush at the connection's steady-state

		// batch size this is alloc-free.

		std::vector<std::pair<const uint8_t*, size_t>> _send_batch_ptrs;



		// Each connection's own lock — see _quic_mutex's comment for why

		// this no longer delegates to the parent.

src/windows/udp.cpp

+43 −0

File changed.

Preview size limit exceeded, changes collapsed.

test/CMakeLists.txt

+7 −0
Original line number Diff line number Diff line
@@ -157,6 +157,13 @@ else() 
    target_link_libraries(benchmark_quic netplus-static)

endif()



add_executable(benchmark_quic_bulk benchmark_quic_bulk.cpp)

if(WIN32)

    target_link_libraries(benchmark_quic_bulk netplus-static ws2_32)

else()

    target_link_libraries(benchmark_quic_bulk netplus-static)

endif()



add_executable(quic_rfc9000_test quic_rfc9000_test.cpp)

if(WIN32)

    target_link_libraries(quic_rfc9000_test netplus-static ws2_32)