Commit 39a2cd8e authored by jan.koester's avatar jan.koester
Browse files

crypto merge

parent 079dfee3

src/quic.cpp

+53 −14
Original line number Diff line number Diff line
@@ -4162,6 +4162,21 @@ void quic::processCryptoFrame(const uint8_t* data, size_t len, size_t& offset) { 
        return;

    }



    // Fast path: the overwhelmingly common case — this fragment arrives

    // exactly at the expected next offset, with no other out-of-order

    // fragments currently buffered. Append directly instead of paying for

    // a CryptoRange allocation, an O(n) merge-scan that rebuilds the

    // entire ranges list even when there's nothing to merge with, and a

    // full re-sort — all of which the general path below does

    // unconditionally, every single CRYPTO frame.

    if (crypto_ranges->empty() && crypto_offset == *crypto_next_offset) {

        crypto_buffer->insert(crypto_buffer->end(), &data[offset], &data[offset + crypto_len]);

        *crypto_next_offset += crypto_len;

        offset += crypto_len;

        processTLSMessages();

        return;

    }



    // Create a range for this data

    CryptoRange range;

    range.offset = crypto_offset;
@@ -4209,8 +4224,8 @@ void quic::processCryptoFrame(const uint8_t* data, size_t len, size_t& offset) { 
            new_ranges.push_back(existing);

        }

    }

    new_ranges.push_back(range);

    *crypto_ranges = new_ranges;

    new_ranges.push_back(std::move(range));

    *crypto_ranges = std::move(new_ranges);

    

    // Check if we can assemble contiguous data starting from next_offset

    // First, sort ranges by offset to handle out-of-order arrivals
@@ -4311,7 +4326,7 @@ void quic::processStreamFrame(const uint8_t* data, size_t len, size_t& offset) { 
    // and received).  Retransmitted packets for erased streams would otherwise

    // auto-create a new Stream object, inflating _data_recv and potentially

    // re-dispatching stale data to the application.

    if (_closed_streams.count(stream_id)) {

    if (isStreamClosed(stream_id)) {

        offset += stream_len;

        return;

    }
@@ -4422,18 +4437,42 @@ void quic::processStreamFrame(const uint8_t* data, size_t len, size_t& offset) { 
        // dispatched (or belong to a duplicate/retransmitted frame) and

        // must not be written — only the portion of this frame at or after

        // recv_base_offset, if any, still needs storing.

        uint64_t buf_end = (end_offset > stream.recv_base_offset)

                                ? (end_offset - stream.recv_base_offset) : 0;

        if (buf_end > stream.recv_buffer.size()) {

            stream.recv_buffer.resize(buf_end, 0);

        }



        uint64_t write_start = std::max(stream_offset, stream.recv_base_offset);

        if (write_start < end_offset) {

            size_t copy_len = static_cast<size_t>(end_offset - write_start);

            size_t src_off = static_cast<size_t>(offset + (write_start - stream_offset));

            std::memcpy(&stream.recv_buffer[write_start - stream.recv_base_offset],

                         &data[src_off], copy_len);

            uint64_t rel_write_start = write_start - stream.recv_base_offset;

            uint64_t cur_size = stream.recv_buffer.size();



            if (rel_write_start > cur_size) {

                // Genuine gap (bytes we don't have yet and this frame

                // doesn't cover) — these must be zero-filled since their

                // content is still unknown.

                stream.recv_buffer.resize(rel_write_start, 0);

                cur_size = rel_write_start;

            }



            if (rel_write_start < cur_size) {

                // Overlaps already-sized buffer (retransmit/duplicate) —

                // copy the overlapping prefix in place.

                size_t in_place_len = std::min<size_t>(copy_len, static_cast<size_t>(cur_size - rel_write_start));

                std::memcpy(&stream.recv_buffer[rel_write_start], &data[src_off], in_place_len);

                src_off += in_place_len;

                rel_write_start += in_place_len;

                copy_len -= in_place_len;

            }



            if (copy_len > 0) {

                // Remainder extends past the current buffer end — append

                // it directly (the vector grows by exactly this many

                // elements, each constructed once from source data)

                // instead of zero-filling via resize() and then

                // overwriting via memcpy, which wrote every new byte

                // twice — the common case for in-order delivery, where

                // this is the only branch that ever does anything.

                stream.recv_buffer.insert(stream.recv_buffer.end(),

                                           &data[src_off], &data[src_off + copy_len]);

            }

        }



        // Track received range — merge with adjacent/overlapping ranges.
@@ -6333,7 +6372,7 @@ size_t quic::sendStreamData(uint64_t stream_id, const uint8_t* data, size_t len, 
        if (stream.recv_fin) {

            bool peer_bidi = _is_server ? (stream_id % 4 == 0) : (stream_id % 4 == 1);

            bool peer_uni  = _is_server ? (stream_id % 4 == 2) : (stream_id % 4 == 3);

            _closed_streams.insert(stream_id);

            markStreamClosed(stream_id);

            _streams.erase(stream_id);

            if (peer_bidi) {

                _max_streams_bidi_local++;
@@ -6861,7 +6900,7 @@ void quic::retireStreamIfDone(uint64_t stream_id) { 
    if (stream.recv_fin && stream.send_fin) {

        bool peer_bidi = _is_server ? (stream_id % 4 == 0) : (stream_id % 4 == 1);

        bool peer_uni  = _is_server ? (stream_id % 4 == 2) : (stream_id % 4 == 3);

        _closed_streams.insert(stream_id);

        markStreamClosed(stream_id);

        _streams.erase(it);

        if (peer_bidi) {

            _max_streams_bidi_local++;

src/socket.h

+42 −1
Original line number Diff line number Diff line
@@ -1171,7 +1171,48 @@ namespace netplus { 


		// Streams

		std::map<uint64_t, Stream> _streams;

		std::set<uint64_t> _closed_streams;  // IDs of fully-closed streams (prevent resurrection)



		// Bounded tracking for "has this stream_id been fully closed"

		// (prevents resurrection via a stale/retransmitted frame after

		// both sides have FIN'd — see retireStreamIfDone()/closeStream()).

		// A plain std::set<uint64_t> of every closed stream_id ever would

		// grow without bound over a connection's lifetime (permanent

		// memory plus ever-costlier lookups on every incoming STREAM

		// frame). QUIC stream IDs are allocated gaplessly within each of

		// the 4 (initiator, direction) classes (stream_id % 4), so

		// closures are almost always contiguous within a class too — track

		// a contiguous-closed-count per class plus a small set of any

		// out-of-order closures still waiting to be absorbed into that

		// count, instead of remembering every ID forever.

		struct ClosedStreamTracker {

			uint64_t contiguous = 0; // stream indices [0, contiguous) in this class are closed

			std::set<uint64_t> gaps; // out-of-order closed indices >= contiguous

		};

		std::array<ClosedStreamTracker, 4> _closed_stream_trackers;



		void markStreamClosed(uint64_t stream_id) {

			auto& t = _closed_stream_trackers[stream_id % 4];

			uint64_t idx = stream_id / 4;

			if (idx < t.contiguous) return; // already recorded

			if (idx == t.contiguous) {

				++t.contiguous;

				while (true) {

					auto it = t.gaps.find(t.contiguous);

					if (it == t.gaps.end()) break;

					t.gaps.erase(it);

					++t.contiguous;

				}

			} else {

				t.gaps.insert(idx);

			}

		}



		bool isStreamClosed(uint64_t stream_id) const {

			const auto& t = _closed_stream_trackers[stream_id % 4];

			uint64_t idx = stream_id / 4;

			return idx < t.contiguous || t.gaps.count(idx) > 0;

		}



		uint64_t _next_stream_id_bidi = 0;

		uint64_t _next_stream_id_uni = 2;