perf

    if (_Socket < 0) return;

    // Probe UDP_SEGMENT (GSO) — a setsockopt that succeeds means the

    // kernel supports generic segmentation offload for UDP.

    uint16_t seg = segment_size;

#ifdef __linux__

    // Probe UDP_SEGMENT (GSO) -- a setsockopt that succeeds means the

    // kernel supports generic segmentation offload for UDP. The option is

    // an int: udp_lib_setsockopt() rejects anything shorter with EINVAL, so

    // a uint16_t here silently disables GSO on every kernel. Set back to 0

    // right away: a nonzero value is sticky and would apply to every later

    // send on this socket; sendBatch() asks per message via cmsg instead.

    int seg = segment_size;

    if (::setsockopt(_Socket, SOL_UDP, UDP_SEGMENT, &seg, sizeof(seg)) == 0) {

        seg = 0;

        ::setsockopt(_Socket, SOL_UDP, UDP_SEGMENT, &seg, sizeof(seg));

        _gso_enabled = true;

    }

    // Probe UDP_GRO

    // Probe UDP_GRO. Linux only: the recvmmsg path is the one that reads

    // the segment size back and splits, the fallback loop does not.

    int gro_val = 1;

    if (::setsockopt(_Socket, SOL_UDP, UDP_GRO, &gro_val, sizeof(gro_val)) == 0) {

        _gro_enabled = true;

    }

#endif

}

// The real wire length of a peer address, by family — NOT

// UDP Batched Send: sendmmsg + GSO when available

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

// UDP_MAX_SEGMENTS is 64 on older kernels (128 since 6.x), and one GSO send is

// still a single IP datagram to the stack, capped at 65535 minus the headers.

static constexpr size_t kGsoMaxSegments = 64;

static constexpr size_t kGsoMaxPayload = 65000;

ssize_t udp::sendBatch(

    const std::vector<std::pair<const uint8_t*, size_t>>& datagrams,

    sockaddr_storage* dest, socklen_t dest_len)

    const size_t count = datagrams.size();

    auto& perf = quicPerf();

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

    if (_gso_enabled && count > 1) {

        bool uniform = true;

        size_t seg_size = datagrams[0].second;

        // Last datagram may be shorter (tail segment) — GSO allows that.

        for (size_t i = 1; i < count - 1; ++i) {

            if (datagrams[i].second != seg_size) { uniform = false; break; }

        }

        if (uniform) {

            // F21: one iovec per datagram, pointing straight at each

            // caller-owned buffer — no intermediate "coalesced" heap

            // allocation + memcpy of the whole batch. sendmsg()'s own

            // copy_from_iter() into the outgoing skb already walks a

            // multi-entry iovec exactly like a single contiguous buffer

            // (that's what iovec/scatter-gather I/O is for), and UDP_SEGMENT

            // segments the *logical* byte stream described by the iovec

            // 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.

            //

            // 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;

            }

            // cmsg carrying UDP_SEGMENT size

            union {

                char buf[CMSG_SPACE(sizeof(uint16_t))];

                struct cmsghdr align;

            } cmsg_buf;

            struct msghdr msg{};

            if (dest) {

                msg.msg_name    = dest;

                msg.msg_namelen = dest_len;

            }

            msg.msg_iov        = iov.data();

            msg.msg_iovlen     = count;

            msg.msg_control    = cmsg_buf.buf;

            msg.msg_controllen = sizeof(cmsg_buf.buf);

            struct cmsghdr* cm = CMSG_FIRSTHDR(&msg);

            cm->cmsg_level = SOL_UDP;

            cm->cmsg_type  = UDP_SEGMENT;

            cm->cmsg_len   = CMSG_LEN(sizeof(uint16_t));

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

                static_cast<uint16_t>(seg_size);

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

            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 ---

#ifdef NETPLUS_NO_MMSG

    // Fallback: sendto loop for platforms without a trusted sendmmsg.

    }

    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);

    return sent;

#else

    // 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.

    // thread_local, reused across calls -- avoids two fresh heap allocations

    // (iovec + mmsghdr arrays) on every flushBatch() call.

    //

    // One iovec per datagram, pointing straight at the caller-owned buffer.

    // With GSO, a run of equal-sized datagrams becomes one message with a

    // UDP_SEGMENT cmsg of its own instead of a whole batch either being

    // uniform or not: an ACK or a shorter tail in the middle of a batch only

    // ends the current run, and everything still goes out in one syscall.

    union CmsgBuf {

        char buf[CMSG_SPACE(sizeof(uint16_t))];

        struct cmsghdr align;

    };

    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);

    static thread_local std::vector<CmsgBuf> cmsgs;

    static thread_local std::vector<uint16_t> msg_counts;

    if (iovecs.size() < count) {

        iovecs.resize(count);

        msgs.resize(count);

        cmsgs.resize(count);

        msg_counts.resize(count);

    }

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

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

        iovecs[i].iov_len  = datagrams[i].second;

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

        msgs[i].msg_hdr.msg_iovlen = 1;

    }

    size_t nmsgs = 0;

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

        const size_t seg_size = datagrams[i].second;

        size_t run = 1;

        if (_gso_enabled && seg_size > 0 && seg_size <= kGsoMaxPayload) {

            const size_t max_run = std::min(kGsoMaxSegments, kGsoMaxPayload / seg_size);

            while (i + run < count && run < max_run && datagrams[i + run].second == seg_size)

                ++run;

            // Only the last segment may be shorter, and it has to be nonempty.

            if (i + run < count && run < max_run && run > 0 &&

                datagrams[i + run].second > 0 && datagrams[i + run].second < seg_size)

                ++run;

        }

        struct mmsghdr& m = msgs[nmsgs];

        std::memset(&m, 0, sizeof(m));

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

        m.msg_hdr.msg_iovlen = run;

        if (dest) {

            msgs[i].msg_hdr.msg_name    = dest;

            msgs[i].msg_hdr.msg_namelen = dest_len;

            m.msg_hdr.msg_name    = dest;

            m.msg_hdr.msg_namelen = dest_len;

        }

        if (run > 1) {

            m.msg_hdr.msg_control    = cmsgs[nmsgs].buf;

            m.msg_hdr.msg_controllen = sizeof(cmsgs[nmsgs].buf);

            struct cmsghdr* cm = CMSG_FIRSTHDR(&m.msg_hdr);

            cm->cmsg_level = SOL_UDP;

            cm->cmsg_type  = UDP_SEGMENT;

            cm->cmsg_len   = CMSG_LEN(sizeof(uint16_t));

            uint16_t seg = static_cast<uint16_t>(seg_size);

            std::memcpy(CMSG_DATA(cm), &seg, sizeof(seg));

        }

        msg_counts[nmsgs] = static_cast<uint16_t>(run);

        ++nmsgs;

        i += run;

    }

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

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

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

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

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

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

    if (sent_msgs <= 0) {

        if (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);

        } else if (nmsgs < count && (errno == EIO || errno == EINVAL)) {

            // The route can't do it: EIO without checksum offload, EINVAL when

            // a segment doesn't fit the MTU. The probe can't see either.

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

            _gso_enabled = false;

            return sendBatch(datagrams, dest, dest_len);

        }

        return sent_msgs;

    }

    size_t sent = 0;

    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);

    for (int m = 0; m < sent_msgs; ++m) {

        const size_t run = msg_counts[static_cast<size_t>(m)];

        bytes += msgs[static_cast<size_t>(m)].msg_len;

        sent += run;

        if (run > 1) {

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

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

        }

    }

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

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

    return static_cast<ssize_t>(sent);

#endif

    return sent;

}

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

        return;

    }

    // RFC 9002 7.8: no growth while something else limits the sender (flow

    // control, the application, the CPU). Otherwise slow start keeps doubling

    // a window nothing has tested, and the first time the sender can fill it

    // that is one burst far beyond what the path takes.

    if (_bytes_in_flight * 2 < _cwnd) return;

    // If this packet was sent during recovery, don't grow cwnd

    if (_in_recovery && pn <= _recovery_start_pn) {

        return;

    // Flow control: after consuming data, check if we should send

    // MAX_STREAM_DATA to open the receive window for the peer.

    // Trigger when >50% of the advertised window has been consumed.

    uint64_t consumed_from_committed = stream.recv_offset;

    if (consumed_from_committed > stream.max_data_local_committed / 2) {

    // Trigger when less than half a window of credit is left. Both sides of

    // this are absolute offsets, so comparing against committed / 2 instead

    // is only right for the first window: after that committed = offset +

    // window, the test becomes offset > window, and it fires on every packet.

    if (stream.recv_offset + stream.max_data_local / 2 > stream.max_data_local_committed) {

        uint64_t new_limit = stream.recv_offset + stream.max_data_local;

        if (new_limit > stream.max_data_local_committed) {

            stream.max_data_local_committed = new_limit;

    }

    // Connection-level flow control: update MAX_DATA after consumption

    if (_data_recv > _max_data_local_committed / 2) {

    if (_data_recv + _max_data_local / 2 > _max_data_local_committed) {

        uint64_t new_limit = _data_recv + _max_data_local;

        if (new_limit > _max_data_local_committed) {

            _max_data_local_committed = new_limit;

    // Update connection-level receive counter (only new unique bytes)

    _data_recv += new_unique_bytes;

    // Flow control: defer MAX_STREAM_DATA when >50% of window consumed.

    // Flow control: defer MAX_STREAM_DATA once less than half a window of

    // credit is left (see recvStreamData() on why not committed / 2).

    // recv_base_offset + recv_buffer.size() is the highest absolute offset

    // the buffer currently extends to — equivalent to the old plain

    // recv_buffer.size() when recv_base_offset is 0 (always true outside

    // instead -- granting it on arrival is exactly what lets a fast peer outrun a slow reader.

    const bool pull_stream = _is_server && _stream_pull_dispatch && (stream_id % 4 == 0);

    uint64_t consumed = stream.recv_base_offset + stream.recv_buffer.size();

    if (!pull_stream && consumed > stream.max_data_local_committed / 2) {

    if (!pull_stream && consumed + stream.max_data_local / 2 > stream.max_data_local_committed) {

        uint64_t new_limit = consumed + stream.max_data_local;

        stream.max_data_local_committed = new_limit;

        _pending_max_stream_data[stream_id] = new_limit;

    }

    // Flow control: defer MAX_DATA when >50% of connection window consumed

    if (_data_recv > _max_data_local_committed / 2) {

    // Flow control: same for MAX_DATA and the connection window

    if (_data_recv + _max_data_local / 2 > _max_data_local_committed) {

        uint64_t new_limit = _data_recv + _max_data_local;

        _max_data_local_committed = new_limit;

        _pending_max_data_value = new_limit;

    stream.stream_id = stream_id;

    // Check flow control

    size_t send_len = len;

    if (_data_sent + send_len > _max_data_remote) {

        send_len = (_max_data_remote > _data_sent) ?

    auto clampToCredit = [&]() {

        size_t n = len;

        if (_data_sent + n > _max_data_remote) {

            n = (_max_data_remote > _data_sent) ?

                (_max_data_remote - _data_sent) : 0;

        }

    if (stream.send_offset + send_len > stream.max_data_remote) {

        send_len = (stream.max_data_remote > stream.send_offset) ?

        if (stream.send_offset + n > stream.max_data_remote) {

            n = (stream.max_data_remote > stream.send_offset) ?

                (stream.max_data_remote - stream.send_offset) : 0;

        }

        return n;

    };

    size_t send_len = clampToCredit();

    // Out of credit: the MAX_DATA/MAX_STREAM_DATA that ends this may already

    // be in the socket, and a caller retrying on 0 has nothing else to read

    // it. Without this, the only pumps were the ones on the cwnd path, so a

    // sender that runs out of credit before cwnd stalled forever.

    if (send_len == 0 && len > 0) {

        pumpIncomingLocked();

        send_len = clampToCredit();

    }

    // Still none: MAX_DATA/MAX_STREAM_DATA aren't retransmitted, so if ours

    // got lost only this gets the peer to resend it (0x14/0x15 in

    // processFrame()). Paced, a caller may be retrying this in a tight loop.

    if (send_len == 0 && len > 0) {

        auto now = std::chrono::steady_clock::now();

        if (now - _last_blocked_sent >= std::chrono::duration<double>(std::max(_rto, 0.01))) {

            _last_blocked_sent = now;

            if (_data_sent >= _max_data_remote) sendDataBlocked(_max_data_remote);

            if (stream.send_offset >= stream.max_data_remote)

                sendStreamDataBlocked(stream_id, stream.max_data_remote);

        }

    }

    // A STREAM FIN fixes the final stream size permanently.  Completion of a

    // flow-control-clamped prefix is therefore not completion of the caller's

        pt_buf_base = pt_stack;

    }

    // Adaptive pump: while there's ample cwnd headroom, several GSO/sendmmsg

    // batches can go out before ACK/flow-control processing needs to run --

    // the cwnd-blocked wait path below already forces a pump exactly when

    // headroom actually runs out, so skipping it here only removes redundant

    // recvmmsg syscalls + routing work in the common, not-blocked case, not

    // correctness. Still bounded by PUMP_ADAPTIVE_MAX_BATCHES so a connection

    // with a very large cwnd (BBR-lite ramped up, or a high-BDP path) doesn't

    // go arbitrarily long without processing ACKs/RTT samples/flow-control

    // updates. Returns false once the connection is going away.

    auto pumpAfterFlush = [&]() {

        ++_flushes_since_pump;

        bool cwnd_getting_tight = _handshake_complete &&

            _cwnd > 0 && (_bytes_in_flight * 2 >= _cwnd);

        if (!cwnd_getting_tight && _flushes_since_pump < PUMP_ADAPTIVE_MAX_BATCHES) {

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

            return true;

        }

        // quic_mtx() is already held throughout this function (F31).

        pumpIncomingLocked();

        _flushes_since_pump = 0;

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

        return _conn_state.load() != ConnectionState::Closed &&

               _conn_state.load() != ConnectionState::Draining;

    };

    size_t sent_total = 0;

    size_t batch_count = 0;         // packets accumulated since last flush

    // Match BATCH_MAX so _send_batch's own auto-flush (batchPacket(), quic.cpp:300)

        // Congestion control: use estimated actual packet size, not max_datagram

        size_t est_pkt_size = header_overhead + frame_hdr_max + chunk;

        // Once stalled, wait for room for a real burst rather than one

        // packet. Resuming on the first ACK sends one packet per ACK, each

        // getting an ACK of its own: a stable state with no batching left

        // on either side, GSO and GRO doing nothing and a syscall per packet.

        size_t resume_bytes = std::max<size_t>(est_pkt_size,

            std::min<size_t>(est_pkt_size * CWND_RESUME_PACKETS, _cwnd / 4));

        if (_handshake_complete && !cwndAllowsSend(est_pkt_size)) {

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

            // Flush any batched packets first so the peer can ACK them

                pumpIncomingLocked();

                _flushes_since_pump = 0;

                ++spin_count;

                if (cwndAllowsSend(est_pkt_size)) { unblocked = true; break; }

                if (cwndAllowsSend(resume_bytes)) { unblocked = true; break; }

            }

            quicPerf().cc_spin_iterations.fetch_add(static_cast<uint64_t>(spin_count), std::memory_order_relaxed);

            // If still blocked, wait with a real timeout instead of

                    pumpIncomingLocked();

                    _flushes_since_pump = 0;

                    ++wait_count;

                    if (cwndAllowsSend(est_pkt_size)) { unblocked = true; break; }

                    if (cwndAllowsSend(resume_bytes)) { unblocked = true; break; }

                }

                quicPerf().cc_wait_iterations.fetch_add(static_cast<uint64_t>(wait_count), std::memory_order_relaxed);

            }

            if (!unblocked) break;

            if (!unblocked && !cwndAllowsSend(est_pkt_size)) break;

        }

        // === Build short header directly into pkt_buf ===

        if (batch_count >= BATCH_FLUSH) {

            flushBatch();

            batch_count = 0;

            ++_flushes_since_pump;

            // Adaptive pump: while there's ample cwnd headroom, several

            // GSO/sendmmsg batches can go out before ACK/flow-control

            // processing needs to run — the cwnd-blocked wait path above

            // already forces a pump exactly when headroom actually runs

            // out, so skipping it here only removes redundant recvmmsg

            // syscalls + routing work in the common, not-blocked case, not

            // correctness. Still bounded by PUMP_ADAPTIVE_MAX_BATCHES so a

            // connection with a very large cwnd (BBR-lite ramped up, or a

            // high-BDP path) doesn't go arbitrarily long without processing

            // ACKs/RTT samples/flow-control updates.

            bool cwnd_getting_tight = _handshake_complete &&

                _cwnd > 0 && (_bytes_in_flight * 2 >= _cwnd);

            if (cwnd_getting_tight || _flushes_since_pump >= PUMP_ADAPTIVE_MAX_BATCHES) {

                // quic_mtx() is already held throughout this function (F31).

                pumpIncomingLocked();

                _flushes_since_pump = 0;

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

                if (_conn_state.load() == ConnectionState::Closed ||

                    _conn_state.load() == ConnectionState::Draining) {

                    break;

                }

            } else {

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

            }

            if (!pumpAfterFlush()) break;

        }

    }

    // Flush remaining batched packets

    // Flush remaining batched packets. Counts towards the pump as well: an

    // application writing less than BATCH_FLUSH packets per call never gets

    // to the flush in the loop above, and without a pump here it reads

    // nothing at all until cwnd or flow control stalls it. By then ACKs and

    // MAX_DATA have piled up in the socket buffer and start getting dropped.

    if (batch_count > 0) {

        flushBatch();

        pumpAfterFlush();

    }

    // Only mark FIN if we actually sent all requested data (FIN is on the last chunk).

		void sendDataBlocked(uint64_t limit);

		/// Send a STREAM_DATA_BLOCKED frame

		void sendStreamDataBlocked(uint64_t stream_id, uint64_t limit);

		/// Paces the two above when a caller keeps retrying out of credit.

		std::chrono::steady_clock::time_point _last_blocked_sent{};

		/// Send an ACK frame for 1-RTT packets

		void sendAppAck();

		// this file was added for.

		static constexpr double HYSTART_MIN_RTT_THRESH = 0.004;  // 4ms

		static constexpr double HYSTART_MAX_RTT_THRESH = 0.016;  // 16ms

		static constexpr int HYSTART_N_RTT_SAMPLE = 8;

		// RFC 9406 says 8, which assumes an ACK every other packet. A sample

		// here is one ACK, and with GSO/GRO that covers a whole receive

		// batch: a round rarely has 8 of them, and a round with too few is

		// never evaluated, so slow start would simply never exit.

		static constexpr int HYSTART_N_RTT_SAMPLE = 3;

		static constexpr int HYSTART_CSS_ROUNDS = 5;

		static constexpr uint64_t HYSTART_CSS_GROWTH_DIVISOR = 4;

		// connection whose cwnd happens to be very large.

		size_t _flushes_since_pump = 0;

		static constexpr size_t PUMP_ADAPTIVE_MAX_BATCHES = 4;

		// How much cwnd room, in packets, sendStreamData() waits for after

		// stalling. Capped at cwnd / 4 there so small windows keep moving.

		static constexpr size_t CWND_RESUME_PACKETS = 16;

		// 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