optimize

namespace netplus {

namespace {

// Connection IDs are peer-chosen and RFC 9000 allows 0-20 bytes, so they

// can't be packed into a fixed-width integer key without either truncating

// (risking two different CIDs colliding on the same key) or zero-padding

// (risking a short CID colliding with a longer one that happens to start

// with the same bytes followed by zeros). std::string has neither problem

// (it carries its own length) and, for CIDs up to the small-string-optimized

// capacity (>=15 bytes on libstdc++/libc++) — which covers every CID this

// codebase generates (always 8 bytes, see generateConnectionId) and most

// real-world peer-chosen ones — needs no heap allocation, unlike the

// std::vector<uint8_t> key this replaces.

inline std::string cidKey(const uint8_t* data, size_t len) {

    return std::string(reinterpret_cast<const char*>(data), len);

}

inline std::string cidKey(const std::vector<uint8_t>& v) {

    return cidKey(v.data(), v.size());

}

} // namespace

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

// Constructors & Destructor

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

    std::lock_guard<std::recursive_mutex> lock(quic_mtx());

    if (_parent != nullptr) {

        _parent->_child_connections.erase(_original_dcid);

        _parent->_child_connections.erase(_local_cid);

        _parent->_child_connections.erase(cidKey(_original_dcid));

        _parent->_child_connections.erase(cidKey(_local_cid));

        // Child connections share the parent's socket fd — do NOT close it.

        // Invalidate _Socket so that ~udp() won't call ::close() on the

        // parent's listening socket (which would cause a double-close /

}

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

// AES-NI Batched Encryption: encrypt multiple QUIC packets in a pipeline

// Amortises AES-GCM setup cost across N packets

// Batched processing of 1-RTT packets

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

void quic::batchEncryptPackets(

    const uint8_t* key, const uint8_t* iv, aes* cached,

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

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

    const std::vector<uint64_t>& pns,

    std::vector<std::vector<uint8_t>>& out_packets)

void quic::processApplicationPacketsBatch(

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

{

    const size_t n = headers.size();

    out_packets.resize(n);

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

        const uint8_t* hdr = headers[i].first;

        size_t hdr_len     = headers[i].second;

        const uint8_t* pt  = plains[i].first;

        size_t pt_len      = plains[i].second;

        uint64_t pn        = pns[i];

        // Build nonce: IV XOR packet_number

        uint8_t nonce[12];

        std::memcpy(nonce, iv, 12);

        for (int j = 0; j < 8; ++j)

            nonce[11 - j] ^= static_cast<uint8_t>((pn >> (8 * j)) & 0xFF);

    _current_enc_level = EncryptionLevel::Application;

        auto& pkt = out_packets[i];

        pkt.resize(hdr_len + pt_len + 16);

        std::memcpy(pkt.data(), hdr, hdr_len);

    const uint8_t* key = _is_server ? _app_key_client : _app_key_server;

    const uint8_t* iv  = _is_server ? _app_iv_client  : _app_iv_server;

    const uint8_t* hp  = _is_server ? _app_hp_client  : _app_hp_server;

        uint8_t* ct_dst  = pkt.data() + hdr_len;

        uint8_t* tag_dst = ct_dst + pt_len;

    std::vector<uint8_t> header, payload;

    for (auto& [data, len] : packets) {

        uint64_t pn;

        if (cached) {

            cached->aes_gcm_encrypt(nonce, hdr, hdr_len, pt, pt_len, ct_dst, tag_dst);

        } else {

            // Fallback: create a temporary AES instance

            const bool use_aes256 = (_selected_cipher == 0x1302);

            if (use_aes256) {

                aes256 tmp(std::vector<uint8_t>(key, key + 32));

                tmp.aes_gcm_encrypt(nonce, hdr, hdr_len, pt, pt_len, ct_dst, tag_dst);

            } else {

                aes128 tmp(std::vector<uint8_t>(key, key + 16));

                tmp.aes_gcm_encrypt(nonce, hdr, hdr_len, pt, pt_len, ct_dst, tag_dst);

        if (!unprotectPacket(data, len, header, payload, pn, key, iv, hp, false)) {

            QUIC_DBG("processApplicationPacketsBatch: decrypt FAILED");

            continue;

        }

        QUIC_DBG("processApplicationPacketsBatch: decrypt OK pn=%lu payload=%zu",

                 (unsigned long)pn, payload.size());

        if (pn > _app_pn_recv) {

            _app_pn_recv = pn;

        }

        trackRecvPN(pn, _app_pn_recv_ranges);

        // Defer ACK across the whole batch — sent once below (or piggybacked

        // sooner if frame processing triggers an outgoing data packet), so N

        // packets arriving in one recv cycle cost one ACK instead of N.

        _app_ack_pending = true;

        size_t offset = 0;

        while (offset < payload.size()) {

            processFrame(payload.data(), payload.size(), offset);

        }

    }

bool quic::batchDecryptPacket(

    const uint8_t* key, const uint8_t* iv, const uint8_t* hp_key,

    aes* cached_aes, aes* cached_hp,

    const uint8_t* data, size_t len, bool force_aes128,

    std::vector<uint8_t>& header, std::vector<uint8_t>& payload,

    uint64_t& packet_number)

{

    // Delegates to unprotectPacket but uses provided cached instances

    return unprotectPacket(data, len, header, payload, packet_number,

                           key, iv, hp_key, force_aes128);

    if (_app_ack_pending) {

        sendAppAck();

        _app_ack_pending = false;

    }

}

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

        std::remove_if(_child_owned_connections.begin(), _child_owned_connections.end(),

            [this](const std::unique_ptr<quic>& p) {

                if (p && p->_conn_state.load() == ConnectionState::Closed) {

                    _child_connections.erase(p->_original_dcid);

                    _child_connections.erase(p->_local_cid);

                    _child_connections.erase(cidKey(p->_original_dcid));

                    _child_connections.erase(cidKey(p->_local_cid));

                    return true;

                }

                return false;

    std::vector<std::vector<uint8_t>> datagrams;

    std::vector<sockaddr_storage> addrs;

    // Packets for existing connections, grouped by connection, deferred

    // until the end of each recvBatchAddr cycle so a burst of several

    // short-header packets for the same connection is decrypted and

    // processed as one batch (see processApplicationPacketsBatch) instead

    // of one ACK-flush-dispatch round trip per packet. Datagram storage

    // (`datagrams`) stays valid until the next recvBatchAddr call, so this

    // must be flushed before that call and before any early return.

    std::map<quic*, std::vector<std::pair<const uint8_t*, size_t>>> pending_app_pkts;

    auto flush_pending_app = [&]() {

        for (auto& [existing, pkts] : pending_app_pkts) {

            existing->processApplicationPacketsBatch(pkts);

            if (existing->_handshake_complete &&

                existing->_conn_state.load() == ConnectionState::Connected) {

                for (auto& [sid, limit] : existing->_pending_max_stream_data) {

                    existing->sendMaxStreamData(sid, limit);

                }

                existing->_pending_max_stream_data.clear();

                if (existing->_pending_max_data) {

                    existing->sendMaxData(existing->_pending_max_data_value);

                    existing->_pending_max_data = false;

                }

            }

            if (!existing->_pending_dispatches.empty() && existing->_stream_callback) {

                std::vector<PendingDispatch> dispatches = std::move(existing->_pending_dispatches);

                existing->_pending_dispatches.clear();

                lock.unlock();

                for (auto& pd : dispatches) {

                    existing->_stream_callback(existing, pd.stream_id, pd.data, pd.fin);

                }

                lock.lock();

            }

        }

        pending_app_pkts.clear();

    };

    while (total_processed < MAX_DRAIN) {

    int batch_to_read = std::min(64, MAX_DRAIN - total_processed);

            continue;

        }

        std::vector<uint8_t> dcid(dgram.begin() + 1, dgram.begin() + 1 + cid_len);

        // Look up the connection

        auto it = _child_connections.find(dcid);

        // Look up the connection directly from the datagram bytes — no

        // intermediate std::vector allocation just to build a lookup key.

        auto it = _child_connections.find(cidKey(&dgram[1], cid_len));

        if (it != _child_connections.end() && it->second != nullptr) {

            quic* existing = it->second;

                existing->_conn_state.load() == ConnectionState::Draining) {

                _child_connections.erase(it);

            } else {

                // Parent lock covers child state — no separate child lock needed

                existing->processIncomingPacket(dgram.data(), dgram.size());

                // Flush child's pending flow control frames

                if (existing->_handshake_complete &&

                    existing->_conn_state.load() == ConnectionState::Connected) {

                    for (auto& [sid, limit] : existing->_pending_max_stream_data) {

                        existing->sendMaxStreamData(sid, limit);

                    }

                    existing->_pending_max_stream_data.clear();

                    if (existing->_pending_max_data) {

                        existing->sendMaxData(existing->_pending_max_data_value);

                        existing->_pending_max_data = false;

                    }

                }

                // Dispatch stream callbacks outside the lock

                if (!existing->_pending_dispatches.empty() && existing->_stream_callback) {

                    std::vector<PendingDispatch> dispatches = std::move(existing->_pending_dispatches);

                    existing->_pending_dispatches.clear();

                    lock.unlock();

                    for (auto& pd : dispatches) {

                        existing->_stream_callback(existing, pd.stream_id, pd.data, pd.fin);

                    }

                    lock.lock();

                }

                // Parent lock covers child state — defer to end-of-cycle batch

                // (flush_pending_app) instead of processing immediately.

                pending_app_pkts[existing].emplace_back(dgram.data(), dgram.size());

            }

        }

    // Check if this is for an existing connection (by original DCID)

    // The DCID in subsequent packets should match our local_cid of an existing connection

    auto it = _child_connections.find(dcid);

    auto it = _child_connections.find(cidKey(dcid));

    if (it != _child_connections.end() && it->second != nullptr) {

        quic* existing = it->second;

             existing->_conn_state.load() == ConnectionState::Draining)) {

            // Remove from routing maps

            _child_connections.erase(existing->_local_cid);

            _child_connections.erase(existing->_original_dcid);

            _child_connections.erase(cidKey(existing->_local_cid));

            _child_connections.erase(cidKey(existing->_original_dcid));

            // Remove from ownership vector (destroys the old connection)

            _child_owned_connections.erase(

    new_quic->_cert_map = _cert_map;        // Copy certificate configuration

    new_quic->_parent = this;               // Link to parent for routing

    new_quic->_stream_callback = _stream_callback;  // Copy stream callback

    new_quic->_incremental_stream_dispatch = _incremental_stream_dispatch;

    // Generate our own connection ID for the client to use

    new_quic->generateConnectionId(new_quic->_local_cid, 8);

    _child_owned_connections.push_back(std::move(new_quic));

    // Register this connection with both the original DCID and our local CID

    _child_connections[dcid] = child_ptr;

    _child_connections[child_ptr->_local_cid] = child_ptr;

    _child_connections[cidKey(dcid)] = child_ptr;

    _child_connections[cidKey(child_ptr->_local_cid)] = child_ptr;

    // Store peer address

    // The child uses the parent socket with sendto() for sending

    child_ptr->_Socket = _Socket;  // Share parent socket

    // GSO/GRO were already probed on this fd when the listener was set up

    // (probeGSOGRO() in the constructor); child connections share that same

    // fd but are freshly constructed quic objects, so without copying these

    // flags every child would silently fall back to unbatched sendto/recvfrom.

    child_ptr->_gso_enabled      = _gso_enabled;

    child_ptr->_gro_enabled      = _gro_enabled;

    child_ptr->_gso_segment_size = _gso_segment_size;

    // Set non-blocking if requested (parent should already be non-blocking)

    if (nonblock) {

        child_ptr->setNonBlock();

    // Don't transfer ownership to csock (would cause double-free).

    // QUIC connections share the parent socket and are routed via accept().

    csock = nullptr;

    flush_pending_app();  // don't drop already-scanned packets from this cycle

    return;  // New connection created — exit drain loop

    } // end msg_i loop

    flush_pending_app();

    } // end recvmmsg batch loop

    // Reached MAX_DRAIN limit — will be called again by epoll

    // Remove from parent's child map (same lock — quic_mtx() is parent's mutex)

    if (_parent != nullptr) {

        _parent->_child_connections.erase(_original_dcid);

        _parent->_child_connections.erase(_local_cid);

        _parent->_child_connections.erase(cidKey(_original_dcid));

        _parent->_child_connections.erase(cidKey(_local_cid));

    }

}

    // Delegate address resolution, socket creation and UDP connect to the base class

    udp::connect(addr, port, nonblock);

    // NOTE: enlarging SO_RCVBUF/SO_SNDBUF here (matching the server ctors'

    // 4MB) was tried and reverted — bisection showed it reproducibly breaks

    // quic_roundtrip_sha256_test (hangs/aborts on the very first round,

    // where the pre-change baseline passes cleanly). SO_RCVBUF alone is

    // enough to trigger it, independent of SO_SNDBUF, GSO/GRO, or any of

    // the other changes in this diff (each ruled out individually via

    // bisection). The setsockopt calls themselves report success with the

    // expected doubled kernel buffer size, so this isn't a syscall failure

    // — something about the larger client receive queue interacts badly

    // with a downstream path (client-side batch pump is the prime

    // remaining suspect, but not confirmed). Left as probeGSOGRO() only,

    // which bisected clean on its own.

    probeGSOGRO();

    // Generate random DCID for the server (we don't know their CID yet)

    generateConnectionId(_remote_cid, 8);

    _original_dcid = _remote_cid;  // Save for key derivation

    if (!datagrams.empty()) {

        std::lock_guard<std::recursive_mutex> lock(quic_mtx());

        // Route short-header (1-RTT) datagrams — the entire steady-state

        // traffic once the handshake is done — through the batch path so

        // N packets arriving in this one recvBatch() cycle cost one ACK

        // instead of N (same fix as the server's accept() drain loop).

        // Long-header packets (Initial/Handshake) are rare here (a client

        // only sees them during/around the handshake) and keep using the

        // per-packet path, which also handles coalesced long-header packets.

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

        for (auto& dgram : datagrams) {

            if (!dgram.empty() && (dgram[0] & 0x80) == 0) {

                app_pkts.emplace_back(dgram.data(), dgram.size());

            } else {

                processIncomingPacket(dgram.data(), dgram.size());

            }

        }

        if (!app_pkts.empty()) {

            processApplicationPacketsBatch(app_pkts);

        }

        // Flush pending flow control frames so the peer doesn't stall

        if (_handshake_complete && _conn_state.load() == ConnectionState::Connected) {

    // so that long-running callbacks (e.g. import) don't block PING ACKs

    // and don't deadlock when the callback calls sendStreamData/flush_out.

    // ALL stream types are deferred via _pending_dispatches.

    if (_is_server && _stream_callback) {

    if (_is_server && _stream_callback && _incremental_stream_dispatch) {

        // Opt-in path (setIncrementalStreamDispatch): dispatch each newly

        // contiguous chunk as it arrives instead of waiting for the whole

        // message, so a server can start responding while the peer is

        // still sending. fin is true only on the chunk that completes the

        // stream; everything before that is fin=false, mirroring how the

        // client side already reads via recv_contiguous (see recvData()).

        if (stream.recv_contiguous > stream.recv_dispatched_offset) {

            std::vector<uint8_t> chunk(

                stream.recv_buffer.begin() + stream.recv_dispatched_offset,

                stream.recv_buffer.begin() + stream.recv_contiguous);

            stream.recv_dispatched_offset = stream.recv_contiguous;

            bool chunk_fin = stream.recv_complete;

            if (chunk_fin) {

                stream.recv_dispatched = true;

                stream.recv_buffer.clear();

                stream.recv_buffer.shrink_to_fit();

                stream.recv_ranges.clear();

            }

            _pending_dispatches.push_back({stream_id, std::move(chunk), chunk_fin});

        }

    } else if (_is_server && _stream_callback) {

        bool is_bidi_request = (stream_id % 4 == 0);

        if (is_bidi_request && stream.recv_complete && !stream.recv_dispatched) {

            stream.recv_dispatched = true;

    // Reset PTO back-off counter once ACKs are flowing

    _pto_count = 0;

    // Time-based loss threshold (RFC 9002 §6.1.2)

    // Time-based loss threshold (RFC 9002 §6.1.2). Floor raised from 1ms to

    // 5ms: on loopback/same-host connections srtt/rttvar are tiny, but two

    // independently scheduled userspace threads (sender + receiver event

    // loop) routinely see >1ms scheduling jitter under load that is not

    // packet loss. At a 1ms floor that jitter was misclassified as loss,

    // halving cwnd (onPacketLostCC) for packets that were merely late —

    // which forces more send rounds, which gives the false-positive more

    // chances to fire again, compounding into a throughput collapse that

    // gets worse the larger the transfer.

    double loss_delay = TIME_THRESHOLD * std::max(_srtt, _rttvar);

    if (loss_delay < 0.001) loss_delay = 0.001; // min 1ms

    if (loss_delay < 0.005) loss_delay = 0.005; // min 5ms

    auto loss_time = std::chrono::duration<double>(loss_delay);

    std::vector<SentPacket> lost;

    size_t sent_total = 0;

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

    static constexpr size_t BATCH_FLUSH = 32; // flush every N packets via sendmmsg/GSO

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

    // never fires first — flushing at half that size doubled sendmmsg syscalls

    // for no benefit (cwnd-constrained sends already flush earlier via the

    // congestion-wait path below, so this only changes the already-healthy,

    // not-blocked-on-cwnd case).

    static constexpr size_t BATCH_FLUSH = BATCH_MAX;

    while (sent_total < send_len) {

        size_t chunk = std::min(send_len - sent_total, max_chunk);

            } else {

                static constexpr size_t CID_LEN = 8;

                if (len < 1 + CID_LEN) continue;

                std::vector<uint8_t> dcid(data + 1, data + 1 + CID_LEN);

                auto it = parent->_child_connections.find(dcid);

                auto it = parent->_child_connections.find(cidKey(data + 1, CID_LEN));

                if (it != parent->_child_connections.end() && it->second) {

                    it->second->processIncomingPacket(data, len);

                    if (it->second->_handshake_complete &&

            if (len >= 7) {

                size_t dcid_len = static_cast<size_t>(data[5]);

                if (6 + dcid_len <= len) {

                    std::vector<uint8_t> dcid(data + 6, data + 6 + dcid_len);

                    auto it = parent->_child_connections.find(dcid);

                    auto it = parent->_child_connections.find(cidKey(data + 6, dcid_len));

                    if (it != parent->_child_connections.end() && it->second)

                        it->second->processIncomingPacket(data, len);

                }

    auto t0 = hrc::now();

    int send_idx = 0;

    size_t send_off = 0;  // bytes of the current iteration's payload already sent

    while (total_recv < target_bytes) {

        // Send next chunk if we haven't sent everything yet

        // Send next chunk if we haven't sent everything yet. sendStreamData()

        // may return less than requested when congestion control blocks —

        // only advance to the next iteration once this one's payload (and,

        // on the last iteration, the FIN) has actually gone out; otherwise a

        // partial send silently drops the tail of the transfer (including

        // FIN on the last iteration), and the receive loop below waits

        // forever for bytes that were never sent.

        if (send_idx < iterations) {

            bool fin = (send_idx == iterations - 1);

            size_t sent = client.sendStreamData(stream_id, payload.data(), payload.size(), fin);

            size_t sent = client.sendStreamData(stream_id, payload.data() + send_off,

                                                 payload.size() - send_off, fin);

            total_sent += sent;

            send_off += sent;

            if (send_off >= payload.size()) {

                send_off = 0;

                send_idx++;

            }

        }

        // Pump and receive — drain all available data

        client.pumpNetwork(MSG_DONTWAIT);