perf (b569f535) · Commits · jan.koester / libnetplus

src/crypto/aes.cpp

+86 −49

Changes for src/crypto/aes.cpp: 86 added lines, 49 removed lines.
Original line number	Diff line number	Diff line
		@@ -231,40 +231,89 @@ static void gcm_mul_clmul(std::array<uint8_t,16>& X, const uint8_t H[16]) {
		_mm_storeu_si128(reinterpret_cast<__m128i*>(X.data()), reflect128(result));
		}

		// ---- Inline __m128i GHASH multiply for interleaved use ----
		// Operates directly on __m128i in GCM byte order (no array conversion).
		// ---- GHASH for the interleaved loops ----
		// Works on byte-swapped blocks rather than bit-reflected ones (Gueron & Kounavis,
		// "Intel Carry-Less Multiplication Instruction and its Usage for Computing the
		// GCM Mode", algorithm 5): a shuffle per block instead of the reflect128()
		// sequence, and the shift-left-by-one then makes up for the bit order. Blocks
		// go in four at a time against H^4..H^1, so those four share one reduction.
		struct GhashKey {
		__m128i h[4]; // H^1..H^4, byte-swapped
		};

		AESNI_FUNC
		static inline __m128i gcm_mul_m128i(__m128i X, __m128i H_reflected) {
		__m128i x = reflect128(X);
		static inline __m128i ghash_bswap(__m128i v) {
		return _mm_shuffle_epi8(v, _mm_setr_epi8(15,14,13,12,11,10,9,8,7,6,5,4,3,2,1,0));
		}

		__m128i ll = _mm_clmulepi64_si128(x, H_reflected, 0x00);
		__m128i hh = _mm_clmulepi64_si128(x, H_reflected, 0x11);
		__m128i mid = _mm_xor_si128(_mm_clmulepi64_si128(x, H_reflected, 0x01),
		_mm_clmulepi64_si128(x, H_reflected, 0x10));
		__m128i lo128 = _mm_xor_si128(ll, _mm_slli_si128(mid, 8));
		__m128i hi128 = _mm_xor_si128(hh, _mm_srli_si128(mid, 8));
		AESNI_FUNC
		static inline void ghash_clmul(__m128i a, __m128i b, __m128i& lo, __m128i& mid, __m128i& hi) {
		lo = _mm_xor_si128(lo, _mm_clmulepi64_si128(a, b, 0x00));
		hi = _mm_xor_si128(hi, _mm_clmulepi64_si128(a, b, 0x11));
		mid = _mm_xor_si128(mid, _mm_xor_si128(_mm_clmulepi64_si128(a, b, 0x01),
		_mm_clmulepi64_si128(a, b, 0x10)));
		}

		__m128i s1 = shl128(hi128, 1);
		__m128i s2 = shl128(hi128, 2);
		__m128i s7 = shl128(hi128, 7);
		__m128i fold = _mm_xor_si128(hi128, _mm_xor_si128(s1, _mm_xor_si128(s2, s7)));
		// Reduce the 256-bit product [hi:lo] (with the middle term still separate)
		// mod x^128 + x^7 + x^2 + x + 1.
		AESNI_FUNC
		static inline __m128i ghash_reduce(__m128i lo, __m128i mid, __m128i hi) {
		lo = _mm_xor_si128(lo, _mm_slli_si128(mid, 8));
		hi = _mm_xor_si128(hi, _mm_srli_si128(mid, 8));

		// Shift [hi:lo] left by one
		__m128i c_lo = _mm_srli_epi32(lo, 31);
		__m128i c_hi = _mm_srli_epi32(hi, 31);
		lo = _mm_slli_epi32(lo, 1);
		hi = _mm_slli_epi32(hi, 1);
		hi = _mm_or_si128(hi, _mm_srli_si128(c_lo, 12));
		hi = _mm_or_si128(hi, _mm_slli_si128(c_hi, 4));
		lo = _mm_or_si128(lo, _mm_slli_si128(c_lo, 4));

		// First phase
		__m128i t = _mm_xor_si128(_mm_xor_si128(_mm_slli_epi32(lo, 31), _mm_slli_epi32(lo, 30)),
		_mm_slli_epi32(lo, 25));
		__m128i carry = _mm_srli_si128(t, 4);
		lo = _mm_xor_si128(lo, _mm_slli_si128(t, 12));

		// Second phase
		t = _mm_xor_si128(_mm_xor_si128(_mm_srli_epi32(lo, 1), _mm_srli_epi32(lo, 2)),
		_mm_srli_epi32(lo, 7));
		lo = _mm_xor_si128(lo, _mm_xor_si128(t, carry));
		return _mm_xor_si128(hi, lo);
		}

		alignas(16) uint64_t hipair[2];
		_mm_store_si128(reinterpret_cast<__m128i*>(hipair), hi128);
		uint64_t hi_hi = hipair[1];
		uint64_t ov = (hi_hi >> 63) ^ (hi_hi >> 62) ^ (hi_hi >> 57);
		uint64_t ov2 = ov ^ (ov << 1) ^ (ov << 2) ^ (ov << 7);
		AESNI_FUNC
		static inline __m128i ghash_mul(__m128i a, __m128i b) {
		__m128i lo = _mm_setzero_si128(), mid = lo, hi = lo;
		ghash_clmul(a, b, lo, mid, hi);
		return ghash_reduce(lo, mid, hi);
		}

		__m128i result = _mm_xor_si128(lo128, fold);
		result = _mm_xor_si128(result, _mm_set_epi64x(0, static_cast<long long>(ov2)));
		AESNI_FUNC
		static inline GhashKey ghash_key(const uint8_t H[16]) {
		GhashKey k;
		k.h[0] = ghash_bswap(_mm_loadu_si128(reinterpret_cast<const __m128i*>(H)));
		for (int i = 1; i < 4; ++i) k.h[i] = ghash_mul(k.h[i - 1], k.h[0]);
		return k;
		}

		return reflect128(result);
		// ---- GHASH update: Y = (Y ^ block) * H. Y is byte-swapped, block isn't ----
		AESNI_FUNC
		static inline __m128i ghash_update(__m128i Y, __m128i block, const GhashKey& key) {
		return ghash_mul(_mm_xor_si128(Y, ghash_bswap(block)), key.h[0]);
		}

		// ---- GHASH update: Y = gcm_mul(Y ^ block, H) ----
		// Same result as four ghash_update() calls in a row.
		AESNI_FUNC
		static inline __m128i ghash_update(__m128i Y, __m128i block, __m128i H_reflected) {
		return gcm_mul_m128i(_mm_xor_si128(Y, block), H_reflected);
		static inline __m128i ghash_update4(__m128i Y, __m128i b0, __m128i b1, __m128i b2, __m128i b3,
		const GhashKey& key) {
		__m128i lo = _mm_setzero_si128(), mid = lo, hi = lo;
		ghash_clmul(_mm_xor_si128(Y, ghash_bswap(b0)), key.h[3], lo, mid, hi);
		ghash_clmul(ghash_bswap(b1), key.h[2], lo, mid, hi);
		ghash_clmul(ghash_bswap(b2), key.h[1], lo, mid, hi);
		ghash_clmul(ghash_bswap(b3), key.h[0], lo, mid, hi);
		return ghash_reduce(lo, mid, hi);
		}

		// ---- Interleaved AES-CTR + GHASH for AES-128 GCM ----
		@@ -280,7 +329,7 @@ static void aesni_gcm_ctr_ghash_128(
		const uint8_t H[16],
		uint8_t tag16[16]
		) {
		__m128i H_ref = reflect128(_mm_loadu_si128(reinterpret_cast<const __m128i*>(H)));
		GhashKey H_ref = ghash_key(H);
		__m128i Y = _mm_setzero_si128(); // GHASH accumulator

		// Process AAD first
		@@ -351,10 +400,7 @@ static void aesni_gcm_ctr_ghash_128(
		_mm_storeu_si128(reinterpret_cast<__m128i*>(out + off + 48), ct3);

		// GHASH the 4 ciphertext blocks we just produced
		Y = ghash_update(Y, ct0, H_ref);
		Y = ghash_update(Y, ct1, H_ref);
		Y = ghash_update(Y, ct2, H_ref);
		Y = ghash_update(Y, ct3, H_ref);
		Y = ghash_update4(Y, ct0, ct1, ct2, ct3, H_ref);

		off += 64;
		}
		@@ -415,7 +461,7 @@ static void aesni_gcm_ctr_ghash_128(
		j0_block = _mm_aesenc_si128(j0_block, _mm_loadu_si128(reinterpret_cast<const __m128i>(round_keys + i 16)));
		j0_block = _mm_aesenclast_si128(j0_block, _mm_loadu_si128(reinterpret_cast<const __m128i*>(round_keys + 160)));

		__m128i tag = _mm_xor_si128(j0_block, Y);
		__m128i tag = _mm_xor_si128(j0_block, ghash_bswap(Y));
		_mm_storeu_si128(reinterpret_cast<__m128i*>(tag16), tag);
		}

		@@ -437,7 +483,7 @@ static void aesni_gcm_ctr_ghash_decrypt_128(
		const uint8_t H[16],
		uint8_t tag16[16]
		) {
		__m128i H_ref = reflect128(_mm_loadu_si128(reinterpret_cast<const __m128i*>(H)));
		GhashKey H_ref = ghash_key(H);
		__m128i Y = _mm_setzero_si128(); // GHASH accumulator

		// Process AAD first
		@@ -469,10 +515,7 @@ static void aesni_gcm_ctr_ghash_decrypt_128(
		__m128i ct2 = _mm_loadu_si128(reinterpret_cast<const __m128i*>(in + off + 32));
		__m128i ct3 = _mm_loadu_si128(reinterpret_cast<const __m128i*>(in + off + 48));

		Y = ghash_update(Y, ct0, H_ref);
		Y = ghash_update(Y, ct1, H_ref);
		Y = ghash_update(Y, ct2, H_ref);
		Y = ghash_update(Y, ct3, H_ref);
		Y = ghash_update4(Y, ct0, ct1, ct2, ct3, H_ref);

		__m128i c0 = make_ctr_block(J0, ++ctr);
		__m128i c1 = make_ctr_block(J0, ++ctr);
		@@ -563,7 +606,7 @@ static void aesni_gcm_ctr_ghash_decrypt_128(
		j0_block = _mm_aesenc_si128(j0_block, _mm_loadu_si128(reinterpret_cast<const __m128i>(round_keys + i 16)));
		j0_block = _mm_aesenclast_si128(j0_block, _mm_loadu_si128(reinterpret_cast<const __m128i*>(round_keys + 160)));

		__m128i tag = _mm_xor_si128(j0_block, Y);
		__m128i tag = _mm_xor_si128(j0_block, ghash_bswap(Y));
		_mm_storeu_si128(reinterpret_cast<__m128i*>(tag16), tag);
		}

		@@ -577,7 +620,7 @@ static void aesni_gcm_ctr_ghash_256(
		const uint8_t H[16],
		uint8_t tag16[16]
		) {
		__m128i H_ref = reflect128(_mm_loadu_si128(reinterpret_cast<const __m128i*>(H)));
		GhashKey H_ref = ghash_key(H);
		__m128i Y = _mm_setzero_si128();

		// Process AAD
		@@ -641,10 +684,7 @@ static void aesni_gcm_ctr_ghash_256(
		_mm_storeu_si128(reinterpret_cast<__m128i*>(out + off + 32), ct2);
		_mm_storeu_si128(reinterpret_cast<__m128i*>(out + off + 48), ct3);

		Y = ghash_update(Y, ct0, H_ref);
		Y = ghash_update(Y, ct1, H_ref);
		Y = ghash_update(Y, ct2, H_ref);
		Y = ghash_update(Y, ct3, H_ref);
		Y = ghash_update4(Y, ct0, ct1, ct2, ct3, H_ref);

		off += 64;
		}
		@@ -703,7 +743,7 @@ static void aesni_gcm_ctr_ghash_256(
		j0_block = _mm_aesenc_si128(j0_block, _mm_loadu_si128(reinterpret_cast<const __m128i>(round_keys + i 16)));
		j0_block = _mm_aesenclast_si128(j0_block, _mm_loadu_si128(reinterpret_cast<const __m128i*>(round_keys + 224)));

		__m128i tag = _mm_xor_si128(j0_block, Y);
		__m128i tag = _mm_xor_si128(j0_block, ghash_bswap(Y));
		_mm_storeu_si128(reinterpret_cast<__m128i*>(tag16), tag);
		}

		@@ -718,7 +758,7 @@ static void aesni_gcm_ctr_ghash_decrypt_256(
		const uint8_t H[16],
		uint8_t tag16[16]
		) {
		__m128i H_ref = reflect128(_mm_loadu_si128(reinterpret_cast<const __m128i*>(H)));
		GhashKey H_ref = ghash_key(H);
		__m128i Y = _mm_setzero_si128();

		{
		@@ -745,10 +785,7 @@ static void aesni_gcm_ctr_ghash_decrypt_256(
		__m128i ct2 = _mm_loadu_si128(reinterpret_cast<const __m128i*>(in + off + 32));
		__m128i ct3 = _mm_loadu_si128(reinterpret_cast<const __m128i*>(in + off + 48));

		Y = ghash_update(Y, ct0, H_ref);
		Y = ghash_update(Y, ct1, H_ref);
		Y = ghash_update(Y, ct2, H_ref);
		Y = ghash_update(Y, ct3, H_ref);
		Y = ghash_update4(Y, ct0, ct1, ct2, ct3, H_ref);

		__m128i c0 = make_ctr_block(J0, ++ctr);
		__m128i c1 = make_ctr_block(J0, ++ctr);
		@@ -836,7 +873,7 @@ static void aesni_gcm_ctr_ghash_decrypt_256(
		j0_block = _mm_aesenc_si128(j0_block, _mm_loadu_si128(reinterpret_cast<const __m128i>(round_keys + i 16)));
		j0_block = _mm_aesenclast_si128(j0_block, _mm_loadu_si128(reinterpret_cast<const __m128i*>(round_keys + 224)));

		__m128i tag = _mm_xor_si128(j0_block, Y);
		__m128i tag = _mm_xor_si128(j0_block, ghash_bswap(Y));
		_mm_storeu_si128(reinterpret_cast<__m128i*>(tag16), tag);
		}

src/quic.cpp

+76 −45

Changes for src/quic.cpp: 76 added lines, 45 removed lines.
Original line number	Diff line number	Diff line
		@@ -949,18 +949,18 @@ void quic::onPacketAckedCC(size_t packet_size, uint64_t pn) {
		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;
		}
		_in_recovery = false;

		// 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 (!_cwnd_limited) return;

		// Slow start: cwnd grows by bytes acked (or, under HyStart++
		// Conservative Slow Start, by a fraction of that — see
		// hystartOnRttSample()).
		@@ -2287,16 +2287,7 @@ void quic::handshake_after_accept() {

		// Try to receive more data if handshake not complete
		if (!_handshake_complete && _Socket >= 0) {
		buffer recv_buf(65535);
		try {
		size_t recv_len = udp::recvData(recv_buf, MSG_DONTWAIT);
		if (recv_len > 0) {
		processIncomingPacket(reinterpret_cast<const uint8_t*>(recv_buf.data.buf),
		recv_len);
		}
		} catch (NetException&) {
		// recvData throws on EAGAIN/EWOULDBLOCK or errors
		}
		processReceived();
		{

		// If no data on child socket and parent exists, check if parent has data for us
		@@ -4586,6 +4577,19 @@ std::vector<uint8_t> quic::buildCryptoFrame(const std::vector<uint8_t>& data, ui
		// Blocking Handshake Completion
		// ============================================================================

		// Reads and processes whatever is queued on the socket. For every path that
		// isn't a batch path already: a socket with UDP_GRO hands back a whole GSO burst
		// as one buffer, a plain recv() has no way to learn the segment size, and
		// taking it for a single datagram fails the AEAD and drops all of it.
		size_t quic::processReceived() {
		std::vector<DatagramView> datagrams;
		size_t count = recvBatchViews(datagrams, 64);
		for (auto& dgram : datagrams) {
		processIncomingPacket(dgram.data(), dgram.size());
		}
		return count;
		}

		void quic::completeHandshake() {
		// Receive and process packets until handshake is complete or timeout
		const auto start = std::chrono::steady_clock::now();
		@@ -4614,14 +4618,10 @@ void quic::completeHandshake() {
		if (!sw.waitRead(*this, 5)) continue; // no data within 5ms, retry

		try {
		buffer recv_buf_wrapper(65535);
		// recvBatchViews(), not recvData(): only the former splits what
		// UDP_GRO coalesced, see processReceived().
		recv_attempts++;
		size_t recv_len = udp::recvData(recv_buf_wrapper, MSG_DONTWAIT);
		if (recv_len > 0) {
		recv_got++;
		const uint8_t* pkt = reinterpret_cast<const uint8_t*>(recv_buf_wrapper.data.buf);
		processIncomingPacket(pkt, recv_len);
		}
		recv_got += static_cast<int>(processReceived());
		} catch (NetException& e) {
		if (e.getErrorType() != NetException::Note) {
		throw;
		@@ -4677,20 +4677,20 @@ size_t quic::recvData(buffer& data, int flags) {
		check_stream(it->second);
		} else {
		// No buffered data, try to receive from network
		int recv_flags = (flags & MSG_DONTWAIT) ? MSG_DONTWAIT : 0;
		buffer recv_buf_wrapper(65535);
		size_t recv_len;
		try {
		recv_len = udp::recvData(recv_buf_wrapper, recv_flags);
		} catch (NetException&) {
		throw;
		if (!(flags & MSG_DONTWAIT)) {
		socketwait sw;
		while (!sw.waitRead(*this, 1000)) {
		if (_conn_state.load() == ConnectionState::Closed) break;
		}
		}
		size_t recv_count = processReceived();

		if (recv_len == 0) {
		data.size = 0;
		result = 0;
		if (recv_count == 0) {
		// What udp::recvData() does on an empty socket
		NetException e;
		e[NetException::Note] << "quic::recvData would block";
		throw e;
		} else {
		processIncomingPacket(reinterpret_cast<const uint8_t*>(recv_buf_wrapper.data.buf), recv_len);

		// After processing, check for buffered stream data
		bool found = false;
		@@ -5962,12 +5962,10 @@ void quic::processStreamFrame(const uint8_t* data, size_t len, size_t& offset, u
		// 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) {
		// recv_dispatched_offset always equals recv_base_offset here
		// (the buffer is trimmed to it immediately below, every time),
		// so the chunk always starts at the beginning of recv_buffer.
		uint64_t rel_dispatched = stream.recv_dispatched_offset - stream.recv_base_offset;
		uint64_t rel_contiguous = stream.recv_contiguous - stream.recv_base_offset;
		std::vector<uint8_t> chunk(
		stream.recv_buffer.begin(),
		stream.recv_buffer.begin() + rel_dispatched,
		stream.recv_buffer.begin() + rel_contiguous);
		stream.recv_dispatched_offset = stream.recv_contiguous;
		bool chunk_fin = stream.recv_complete;
		@@ -5984,10 +5982,21 @@ void quic::processStreamFrame(const uint8_t* data, size_t len, size_t& offset, u
		// streams (e.g. large uploads), which could hold the
		// entire transfer in memory even though the application had
		// already consumed most of it via the callback.
		//
		// Only once the prefix makes up half the buffer: behind a
		// hole the buffer holds up to a whole window, and moving all
		// of that for every packet that fills one slows the receiver
		// down enough to cause the next holes. A connection stayed
		// stuck at a tenth of its throughput that way.
		static constexpr uint64_t MIN_TRIM_BYTES = 1024 * 1024;
		if (rel_contiguous >= MIN_TRIM_BYTES &&
		rel_contiguous >= stream.recv_buffer.size() / 2) {
		stream.recv_buffer.erase(stream.recv_buffer.begin(),
		stream.recv_buffer.begin() + rel_contiguous);
		}
		stream.recv_base_offset = stream.recv_dispatched_offset;
		}
		}
		if (chunk_fin) stream.recv_base_offset = stream.recv_dispatched_offset;
		_pending_dispatches.push_back({stream_id, std::move(chunk), chunk_fin});
		}
		} else if (_is_server && _stream_callback) {
		@@ -6133,6 +6142,11 @@ void quic::processAckFrame(const uint8_t* data, size_t len, size_t& offset, bool
		// while some older ACK exists must not erase the backoff.
		bool newly_acked_ack_eliciting = false;

		// Taken once, before this ACK shrinks the count: mid-loop the answer would
		// depend on how many packets the ACK covers, and with a window of two
		// packets it would always be no, leaving cwnd at the minimum for good.
		_cwnd_limited = _bytes_in_flight * 2 >= _cwnd;

		// Remove acknowledged packets from _sent_packets and update RTT
		for (size_t ri = 0; ri < acked_count; ++ri) {
		auto& ar = acked_ranges[ri];
		@@ -6326,7 +6340,13 @@ void quic::checkLossAndRetransmit() {
		// _rttvar, which is a slow-moving variance estimate (0.25 weight per
		// update) and reacts far more sluggishly to a genuine RTT jump (path
		// change, mobile handover) than the actual latest sample does.
		double loss_delay = TIME_THRESHOLD * std::max(_srtt, _latest_rtt);
		// Without a sample yet both are 0, and the floor below alone would call
		// anything older than 5ms lost, on any path slower than that. That false
		// alarm on a connection's first packets is what cost it slow start: cwnd
		// halved from its initial value and grew by one packet per RTT from then
		// on. RFC 9002 6.2.2 says to assume kInitialRtt until measured.
		double loss_delay = TIME_THRESHOLD *
		(_rtt_initialized ? std::max(_srtt, _latest_rtt) : INITIAL_RTT);
		if (loss_delay < 0.005) loss_delay = 0.005; // min 5ms
		auto loss_time = std::chrono::duration<double>(loss_delay);

		@@ -8121,7 +8141,7 @@ size_t quic::sendStreamData(uint64_t stream_id, const uint8_t* data, size_t len,
		// 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();
		drainIncomingLocked();
		send_len = clampToCredit();
		}

		@@ -8214,7 +8234,7 @@ size_t quic::sendStreamData(uint64_t stream_id, const uint8_t* data, size_t len,
		return true;
		}
		// quic_mtx() is already held throughout this function (F31).
		pumpIncomingLocked();
		drainIncomingLocked();
		_flushes_since_pump = 0;
		quicPerf().pumps_done.fetch_add(1, std::memory_order_relaxed);
		return _conn_state.load() != ConnectionState::Closed &&
		@@ -8267,7 +8287,7 @@ size_t quic::sendStreamData(uint64_t stream_id, const uint8_t* data, size_t len,
		// quic_mtx() is already held throughout this function (see
		// `lock` above) — pumpIncomingLocked() skips the redundant
		// recursive re-lock pumpIncoming() would otherwise do here.
		pumpIncomingLocked();
		drainIncomingLocked();
		_flushes_since_pump = 0;
		++spin_count;
		if (cwndAllowsSend(resume_bytes)) { unblocked = true; break; }
		@@ -8325,7 +8345,7 @@ size_t quic::sendStreamData(uint64_t stream_id, const uint8_t* data, size_t len,
		lock.lock();
		// Re-locked just above — see the spin-loop's identical
		// comment on pumpIncomingLocked() (F31).
		pumpIncomingLocked();
		drainIncomingLocked();
		_flushes_since_pump = 0;
		++wait_count;
		if (cwndAllowsSend(resume_bytes)) { unblocked = true; break; }
		@@ -8754,7 +8774,17 @@ void quic::pumpIncoming() {
		pumpIncomingLocked();
		}

		void quic::pumpIncomingLocked() {
		// Reads until the socket is empty instead of one batch's worth. For a sender
		// that needs something specific out of the queue: a MAX_STREAM_DATA arriving
		// behind a window's worth of ACKs sits there for as many pumps as it takes
		// to get to it, 64 datagrams at a time. Bounded so a peer flooding the socket
		// can't keep this thread in here forever.
		void quic::drainIncomingLocked() {
		static constexpr int MAX_DRAIN_ROUNDS = 32;
		for (int i = 0; i < MAX_DRAIN_ROUNDS && pumpIncomingLocked() >= 64; ++i) {}
		}

		size_t quic::pumpIncomingLocked() {
		quic* parent = _parent ? _parent : this;

		// Use udp::recvBatchViews (sendmmsg/recvmmsg + GRO) to read datagrams.
		@@ -8936,6 +8966,7 @@ void quic::pumpIncomingLocked() {
		// retransmit — both operate on `this`'s own already-locked state (see
		// pumpIncomingLocked()'s precondition).
		flushPendingFlowControlAndCheckLoss();
		return datagrams.size();
		}

		void quic::flushPendingFlowControlAndCheckLoss() {

src/socket.h

+8 −4

Changes for src/socket.h: 8 added lines, 4 removed lines.
Original line number	Diff line number	Diff line
		@@ -766,9 +766,9 @@ namespace netplus {
		// stream from offset 0, as before); incremental dispatch
		// advances it to recv_dispatched_offset each time it trims the
		// already-dispatched prefix out of recv_buffer (see
		// processStreamFrame), so the buffer only ever holds
		// undispatched bytes instead of growing to the stream's full
		// cumulative size.
		// processStreamFrame), so the buffer doesn't grow to the stream's
		// full cumulative size. Trimming is amortized: up to half the
		// buffer may still be already-dispatched bytes.
		uint64_t recv_base_offset = 0;
		uint64_t recv_frame_count = 0; // also used to detect a peer-initiated stream's first frame (processStreamFrame)
		// Track received byte ranges for gap detection
		@@ -1115,7 +1115,8 @@ namespace netplus {
		// spin loop). Every other connection's own mutex (`t->quic_mtx()`
		// for `t != this` in the grouped-datagram loop) is still acquired
		// normally — those are genuinely different locks, not redundant.
		void pumpIncomingLocked();
		size_t pumpIncomingLocked();
		void drainIncomingLocked();
		// Shared tail of pumpIncoming()/pumpIncomingLocked(): flush pending
		// flow-control frames and check for retransmits, both operating on
		// `this`'s own already-locked state.
		@@ -1502,6 +1503,7 @@ namespace netplus {
		// dispatch only helps once the object itself is the derived type.
		virtual std::shared_ptr<quic> createChild() const;
		void completeHandshake();
		size_t processReceived();

		// Packet number handling
		uint64_t decodePacketNumber(uint32_t truncated_pn, size_t pn_len, uint64_t largest_pn);
		@@ -2158,6 +2160,8 @@ namespace netplus {
		// Loss detection
		static constexpr uint64_t PACKET_THRESHOLD = 3; // RFC 9002 kPacketThreshold
		static constexpr double TIME_THRESHOLD = 9.0/8; // RFC 9002 kTimeThreshold
		static constexpr double INITIAL_RTT = 0.333; // RFC 9002 kInitialRtt
		bool _cwnd_limited = true; // see processAckFrame()

		// Shared by both buildAckFrame() (encoder) and processAckFrame()
		// (decoder) so they can never disagree about how many ACK ranges