test

// change against it — a blindly added hash-sharding layer would be

// nothing more than an unsubstantiated guess here.

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

// Tier 1: opt-in parallel receive-side decrypt fan-out (see

// quic::processApplicationPacketsBatchParallel()). Separate pool from

// quicDispatchPool() above on purpose: that one is deliberately

// oversubscribed (2x cores) because its work is I/O-wait-bound (waiting on

// a callback that itself waits on network round trips); this one is

// CPU-bound AES-NI/header-protection decrypt, where oversubscribing would

// just cause cache-line/AES-NI-unit contention between worker threads for

// no benefit — sized near physical core count instead.

//

// Off by default: batch-size measurement on this sandbox's loopback bulk

// benchmark showed recvmmsg batches are essentially always exactly 1

// datagram (a single CPU-bound consumer thread never lets the kernel queue

// build up any deeper — see project memory), meaning there is nothing to

// fan out across on THIS workload regardless of how much decrypt costs.

// This is built for deployments where real network jitter/bursty delivery

// produces genuinely multi-packet batches, which this sandbox cannot

// reproduce to validate a throughput gain against — hence opt-in

// (QUIC_PARALLEL_DECRYPT=1) rather than a new default.

ThreadPool& quicDecryptPool() {

    static ThreadPool pool((std::max<unsigned>)(2, std::thread::hardware_concurrency()));

    return pool;

}

// Cached once, same reasoning as quicTraceEnabled()/ensureQuicPerfReporterStarted().

inline bool quicParallelDecryptEnabled() {

    static const bool enabled = (getenv("QUIC_PARALLEL_DECRYPT") != nullptr);

    return enabled;

}

// Below this many packets, thread dispatch/join overhead would dominate

// whatever decrypt time is saved -- not measured against a real batch on

// this sandbox (see above), chosen conservatively pending real-world data.

constexpr size_t kParallelDecryptMinBatch = 8;

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

// QUIC_PERF: optional, opt-in (env var QUIC_PERF=1) throughput/diagnostics

// reporter. Prints one line/sec to stderr summarizing the QuicPerfCounters

{

    _current_enc_level = EncryptionLevel::Application;

    if (quicParallelDecryptEnabled() && packets.size() >= kParallelDecryptMinBatch) {

        processApplicationPacketsBatchParallel(packets);

        return;

    }

    std::vector<uint8_t>& header = _recv_header_scratch;

    std::vector<uint8_t>& payload = _recv_payload_scratch;

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

    }

}

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

// Tier 1: parallel receive-side decrypt fan-out (opt-in, see

// quicParallelDecryptEnabled()). Same externally-visible batch semantics as

// processApplicationPacketsBatch() above -- one ACK per batch, packets

// applied to connection state in original arrival order -- the only

// difference is that the header-protection-removal + AEAD-decrypt step of

// each packet (unprotectPacket()) runs on quicDecryptPool()'s workers

// instead of inline, one at a time, on this thread.

//

// Why this is safe to parallelize (unlike the send path's F30, which was

// analyzed and rejected — see sendStreamData()'s comment): unprotectPacket()

// only reads state that's either immutable for the connection's life

// (_local_cid, _selected_cipher, the AEAD/header-protection key material in

// _app_keys.recv — aes instances are documented safe for concurrent calls

// across threads as long as output buffers don't overlap, see aes.h) or a

// single connection-wide counter, _app_pn_recv, used to reconstruct each

// packet's full packet number from its truncated wire form. That counter is

// NOT written anywhere during the parallel phase below (the real update —

// `if (pn > _app_pn_recv) _app_pn_recv = pn;` — is deferred to the serial

// phase, exactly where the original loop did it) — every worker thread only

// ever *reads* the same frozen value, which is not a data race (only a

// write conflicting with a read/write is). It's also safe to freeze rather

// than update live: PN reconstruction is self-correcting (a stale candidate

// just fails the AEAD tag instead of corrupting state), and a batch this

// size is nowhere near PN-space wraparound distance. Each worker gets its

// own local header/payload buffers instead of the connection-level

// _recv_header_scratch/_recv_payload_scratch (which stay reserved for the

// serial path above) — unprotectPacket() already takes them by reference,

// so nothing about that function itself changes.

//

// Lock discipline is unchanged from the caller's perspective: whoever

// called processApplicationPacketsBatch() already holds quic_mtx() for the

// whole call (see e.g. pumpIncomingLocked()'s comment), and this function

// keeps holding it for the whole parallel-then-serial duration too (it

// blocks on latch.wait()) — no other thread can touch this connection's

// state while decrypt fan-out is in flight, exactly as before. The

// datagram pointers in `packets` point into the calling thread's

// thread_local recvmmsg buffer (see DatagramView's lifetime contract) —

// safe for worker threads to read from concurrently as long as the owning

// thread doesn't call recvBatchViews() again before they're done, which is

// guaranteed here since it's blocked on latch.wait() the whole time.

void quic::processApplicationPacketsBatchParallel(

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

{

    struct DecryptedPacket {

        bool ok = false;

        uint64_t pn = 0;

        std::vector<uint8_t> payload;

    };

    const size_t n = packets.size();

    std::vector<DecryptedPacket> results(n);

    CountdownLatch latch(n);

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

        quicDecryptPool().submit([this, i, &packets, &results, &latch]() {

            std::vector<uint8_t> header;  // per-task, not the shared connection-level scratch

            uint64_t pn = 0;

            const uint8_t* data = packets[i].first;

            size_t len = packets[i].second;

            results[i].ok = unprotectPacket(data, len, header, results[i].payload, pn,

                                             EncryptionLevel::Application, _app_keys.recv);

            results[i].pn = pn;

            latch.count_down();

        });

    }

    latch.wait();

    // Serial phase: apply results in original arrival order, identical to

    // processApplicationPacketsBatch()'s own loop body, so batch semantics

    // (largest_pn tracking, trackRecvPN, frame processing, ACK coalescing)

    // are unchanged from the non-parallel path.

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

        if (!results[i].ok) {

            QUIC_DBG("processApplicationPacketsBatchParallel: decrypt FAILED");

            continue;

        }

        uint64_t pn = results[i].pn;

        std::vector<uint8_t>& payload = results[i].payload;

        QUIC_DBG("processApplicationPacketsBatchParallel: 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);

        size_t offset = 0;

        while (offset < payload.size()) {

            bool ack_eliciting = false;

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

            if (ack_eliciting) {

                if (!_app_ack_pending) _app_ack_elicit_since = std::chrono::steady_clock::now();

                _app_ack_pending = true;

            }

        }

    }

    if (_app_ack_pending) {

        sendAppAck();

        _app_ack_pending = false;

    }

}

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

// Congestion Control: NewReno (RFC 9002)

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

		void processApplicationPacketsBatch(

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

		// Tier 1 (opt-in, see quicParallelDecryptEnabled()): same batch

		// semantics as processApplicationPacketsBatch() (one ACK per batch,

		// packets applied to connection state in original arrival order),

		// but fans the header-protection-removal + AEAD-decrypt step of

		// each packet out across quicDecryptPool() first, then applies all

		// results serially. Called from processApplicationPacketsBatch()

		// itself when the opt-in is enabled and the batch is large enough

		// to be worth it — not part of the public API.

		void processApplicationPacketsBatchParallel(

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

		// Retry/NEW_TOKEN address-validation tokens (RFC 9000 §8.1). Both

		// token kinds share one AEAD-sealed wire format and this one

		// generate/validate pair — see generateToken()'s definition for the

    std::condition_variable _done_cv;

};

// One-shot fan-out/join barrier: N producers each call count_down() once

// when their submitted task finishes; one consumer calls wait() to block

// until all N have. ThreadPool::submit() deliberately has no per-task

// completion signal (see its own comment above) -- this is exactly the

// "caller's own serialization on top" that comment calls for, kept generic

// here (not tied to any one call site) so any future CPU-bound fan-out/join

// use in this codebase can reuse it instead of hand-rolling the same

// mutex+condvar+counter pattern again.

class CountdownLatch {

public:

    explicit CountdownLatch(size_t count) : _count(count) {}

    CountdownLatch(const CountdownLatch&) = delete;

    CountdownLatch& operator=(const CountdownLatch&) = delete;

    void count_down() {

        std::lock_guard<std::mutex> lk(_mtx);

        if (_count > 0 && --_count == 0) {

            _cv.notify_all();

        }

    }

    void wait() {

        std::unique_lock<std::mutex> lk(_mtx);

        _cv.wait(lk, [this] { return _count == 0; });

    }

private:

    std::mutex _mtx;

    std::condition_variable _cv;

    size_t _count;

};

} // namespace netplus