Loading .gitignore +18 −1 Original line number Diff line number Diff line Loading @@ -40,7 +40,24 @@ debian/netplus-certgen.substvars debian/debhelper-build-stamp debian/files debian/debhelper-build-stamp debian/files debian/tmp debian/.debhelper debian/libnetplus-dev .claude/* .claude/worktrees/* debian/libnetplus-dev.substvars debian/libnetplus1.substvars debian/libnetplus1/DEBIAN/control debian/libnetplus1/DEBIAN/md5sums debian/libnetplus1/DEBIAN/shlibs debian/libnetplus1/DEBIAN/triggers debian/libnetplus1/usr/lib/x86_64-linux-gnu/libnetplus.so.1 debian/libnetplus1/usr/lib/x86_64-linux-gnu/libnetplus.so.1.0.1 debian/libnetplus1/usr/share/doc/libnetplus1/changelog.gz debian/libnetplus1/usr/share/doc/libnetplus1/copyright debian/netplus-certgen/usr/share/doc/netplus-certgen/copyright debian/netplus-certgen/usr/share/doc/netplus-certgen/changelog.gz debian/netplus-certgen/usr/bin/netplus-certgen debian/netplus-certgen/DEBIAN/md5sums debian/netplus-certgen/DEBIAN/control src/quic.cpp +54 −3 Original line number Diff line number Diff line Loading @@ -2947,6 +2947,7 @@ void quic::accept(std::unique_ptr<socket>& csock, bool nonblock) { child_ptr->_parent = this; // Link to parent for routing child_ptr->_stream_callback = _stream_callback; // Copy stream callback child_ptr->_incremental_stream_dispatch = _incremental_stream_dispatch; child_ptr->_stream_pull_dispatch = _stream_pull_dispatch; if (_start_dispatch_promoted) child_ptr->_dispatch_promoted.store(true); child_ptr->_pmtud_ceiling = _pmtud_ceiling; // Copy PMTU discovery ceiling // Copy configured flow-control window defaults (see Loading Loading @@ -4863,9 +4864,23 @@ size_t quic::recvStreamData(uint64_t stream_id, uint8_t* data, size_t max_len) { size_t available = static_cast<size_t>(stream.recv_contiguous - stream.recv_offset); size_t to_copy = std::min(max_len, available); std::memcpy(data, stream.recv_buffer.data() + stream.recv_offset, to_copy); // recv_buffer[0] is stream offset recv_base_offset (0 unless something trimmed it). std::memcpy(data, stream.recv_buffer.data() + (stream.recv_offset - stream.recv_base_offset), to_copy); stream.recv_offset += to_copy; // Pull dispatch: drop the consumed prefix, so the buffer holds at most one receive // window instead of the whole stream. Only once it's at least half the buffer (and not // tiny), keeping the front-erase memmove amortized O(1) per byte. if (_is_server && _stream_pull_dispatch && (stream_id % 4 == 0)) { uint64_t consumed_rel = stream.recv_offset - stream.recv_base_offset; if (consumed_rel >= 65536 && consumed_rel * 2 >= stream.recv_buffer.size()) { stream.recv_buffer.erase(stream.recv_buffer.begin(), stream.recv_buffer.begin() + consumed_rel); stream.recv_base_offset = stream.recv_offset; } } // 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. Loading Loading @@ -4915,6 +4930,21 @@ bool quic::isStreamComplete(uint64_t stream_id) const { return it->second.recv_complete; } void quic::flushFlowControl() { std::lock_guard<std::recursive_mutex> lock(quic_mtx()); if (!_handshake_complete || _conn_state.load() != ConnectionState::Connected) return; if (_pending_max_stream_data.empty() && !_pending_max_data) return; for (auto& [sid, limit] : _pending_max_stream_data) { sendMaxStreamData(sid, limit); } _pending_max_stream_data.clear(); if (_pending_max_data) { sendMaxData(_pending_max_data_value); _pending_max_data = false; } flushBatch(); } bool quic::isStreamReset(uint64_t stream_id) const { std::lock_guard<std::recursive_mutex> lock(quic_mtx()); Loading Loading @@ -5050,6 +5080,13 @@ void quic::processFrame(const uint8_t* data, size_t len, size_t& offset, bool& a it->second.reset = true; it->second.recv_fin = true; it->second.recv_fin_offset = final_size; // Pull dispatch: a reader may be waiting for bytes that will now never // come -- give it its final signal so it finds isStreamReset() at once. if (_is_server && _stream_pull_dispatch && (sid % 4 == 0) && _stream_callback && !it->second.recv_dispatched) { it->second.recv_dispatched = true; _pending_dispatches.push_back({sid, {}, true}); } } (void)app_error; // Deliberately not retired here: the application may still Loading Loading @@ -5781,8 +5818,11 @@ void quic::processStreamFrame(const uint8_t* data, size_t len, size_t& offset, u // recv_buffer.size() when recv_base_offset is 0 (always true outside // incremental dispatch's trimming), and still correct once trimming // has shifted the buffer forward. // Pull dispatch (setStreamPullDispatch): credit follows recvStreamData()'s consumption // 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 (consumed > stream.max_data_local_committed / 2) { if (!pull_stream && consumed > stream.max_data_local_committed / 2) { uint64_t new_limit = consumed + stream.max_data_local; stream.max_data_local_committed = new_limit; _pending_max_stream_data[stream_id] = new_limit; Loading Loading @@ -5812,7 +5852,18 @@ void quic::processStreamFrame(const uint8_t* data, size_t len, size_t& offset, u // 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 && _incremental_stream_dispatch) { if (pull_stream && _stream_callback) { // Pull dispatch: announce, don't hand over -- the data stays in recv_buffer for // recvStreamData(). One signal per batch of newly contiguous bytes, plus exactly one // fin=true signal once the stream is complete. if (stream.recv_contiguous > stream.recv_notified_offset || (stream.recv_complete && !stream.recv_dispatched)) { stream.recv_notified_offset = stream.recv_contiguous; bool complete = stream.recv_complete; if (complete) stream.recv_dispatched = true; _pending_dispatches.push_back({stream_id, {}, complete}); } } else 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 Loading src/socket.h +28 −1 Original line number Diff line number Diff line Loading @@ -651,7 +651,11 @@ namespace netplus { // QUIC Protocol Implementation (RFC 9000) // UDP-based transport with TLS 1.3 encryption // =================================================== class quic : public udp { // enable_shared_from_this: lets code handed a raw quic* (e.g. a stream callback) pin an // accepted connection -- always make_shared (see accept()) -- while it works on one of its // streams off-thread. weak_from_this() is empty for a listener/client not owned by a // shared_ptr. class quic : public udp, public std::enable_shared_from_this<quic> { public: // QUIC version constants static constexpr uint32_t QUIC_VERSION_1 = 0x00000001; Loading Loading @@ -735,6 +739,7 @@ namespace netplus { bool recv_fin = false; bool recv_dispatched = false; uint64_t recv_dispatched_offset = 0; // bytes already handed to the stream callback (incremental dispatch only) uint64_t recv_notified_offset = 0; // contiguous bytes already announced to the callback (pull dispatch only) bool recv_complete = false; // all data up to FIN received bool send_fin = false; bool reset = false; Loading Loading @@ -1004,6 +1009,27 @@ namespace netplus { // across calls (or opt out and keep today's behavior). void setIncrementalStreamDispatch(bool enable) { _incremental_stream_dispatch = enable; } // Opt-in, default off, server side (inherited by accepted children like the flag // above): peer-initiated bidirectional streams are no longer handed to the stream // callback as data. The callback is only a "bytes are ready" signal -- empty data, // fin=false while the stream is open, one fin=true call once it's complete -- and the // application pulls the bytes with recvStreamData(). Stream-level flow-control credit // is then granted as recvStreamData() consumes, not as data arrives, and consumed // bytes are trimmed from the receive buffer, so memory per stream stays within its // receive window however slowly the application reads (e.g. relaying an upload to a // slower upstream). Signals can arrive from several pumping threads and out of order // relative to each other; recvStreamData() is the ordered source of truth. Takes // precedence over setIncrementalStreamDispatch() for those streams; unidirectional // streams keep their normal dispatch. void setStreamPullDispatch(bool enable) { _stream_pull_dispatch = enable; } bool streamPullDispatch() const { return _stream_pull_dispatch; } // Sends any MAX_DATA / MAX_STREAM_DATA credit recvStreamData() has queued right away, // instead of with the next packet the connection happens to send -- for a reader // outside the pump loop (pull dispatch), whose peer may be blocked on exactly that // credit and have nothing else to send that would trigger it. void flushFlowControl(); // Opt-in, default off: seed every accepted child connection's // _dispatch_promoted as already true, instead of waiting for its // first dispatch to be observed slower than kSlowDispatchThreshold. Loading Loading @@ -1938,6 +1964,7 @@ namespace netplus { using QuicStreamCallback = std::function<void(quic*, uint64_t, const std::vector<uint8_t>&, bool)>; QuicStreamCallback _stream_callback; bool _incremental_stream_dispatch = false; // see setIncrementalStreamDispatch() bool _stream_pull_dispatch = false; // see setStreamPullDispatch() bool _start_dispatch_promoted = false; // see setStartDispatchPromoted() // Deferred stream dispatches — filled inside _quic_mutex, Loading Loading
.gitignore +18 −1 Original line number Diff line number Diff line Loading @@ -40,7 +40,24 @@ debian/netplus-certgen.substvars debian/debhelper-build-stamp debian/files debian/debhelper-build-stamp debian/files debian/tmp debian/.debhelper debian/libnetplus-dev .claude/* .claude/worktrees/* debian/libnetplus-dev.substvars debian/libnetplus1.substvars debian/libnetplus1/DEBIAN/control debian/libnetplus1/DEBIAN/md5sums debian/libnetplus1/DEBIAN/shlibs debian/libnetplus1/DEBIAN/triggers debian/libnetplus1/usr/lib/x86_64-linux-gnu/libnetplus.so.1 debian/libnetplus1/usr/lib/x86_64-linux-gnu/libnetplus.so.1.0.1 debian/libnetplus1/usr/share/doc/libnetplus1/changelog.gz debian/libnetplus1/usr/share/doc/libnetplus1/copyright debian/netplus-certgen/usr/share/doc/netplus-certgen/copyright debian/netplus-certgen/usr/share/doc/netplus-certgen/changelog.gz debian/netplus-certgen/usr/bin/netplus-certgen debian/netplus-certgen/DEBIAN/md5sums debian/netplus-certgen/DEBIAN/control
src/quic.cpp +54 −3 Original line number Diff line number Diff line Loading @@ -2947,6 +2947,7 @@ void quic::accept(std::unique_ptr<socket>& csock, bool nonblock) { child_ptr->_parent = this; // Link to parent for routing child_ptr->_stream_callback = _stream_callback; // Copy stream callback child_ptr->_incremental_stream_dispatch = _incremental_stream_dispatch; child_ptr->_stream_pull_dispatch = _stream_pull_dispatch; if (_start_dispatch_promoted) child_ptr->_dispatch_promoted.store(true); child_ptr->_pmtud_ceiling = _pmtud_ceiling; // Copy PMTU discovery ceiling // Copy configured flow-control window defaults (see Loading Loading @@ -4863,9 +4864,23 @@ size_t quic::recvStreamData(uint64_t stream_id, uint8_t* data, size_t max_len) { size_t available = static_cast<size_t>(stream.recv_contiguous - stream.recv_offset); size_t to_copy = std::min(max_len, available); std::memcpy(data, stream.recv_buffer.data() + stream.recv_offset, to_copy); // recv_buffer[0] is stream offset recv_base_offset (0 unless something trimmed it). std::memcpy(data, stream.recv_buffer.data() + (stream.recv_offset - stream.recv_base_offset), to_copy); stream.recv_offset += to_copy; // Pull dispatch: drop the consumed prefix, so the buffer holds at most one receive // window instead of the whole stream. Only once it's at least half the buffer (and not // tiny), keeping the front-erase memmove amortized O(1) per byte. if (_is_server && _stream_pull_dispatch && (stream_id % 4 == 0)) { uint64_t consumed_rel = stream.recv_offset - stream.recv_base_offset; if (consumed_rel >= 65536 && consumed_rel * 2 >= stream.recv_buffer.size()) { stream.recv_buffer.erase(stream.recv_buffer.begin(), stream.recv_buffer.begin() + consumed_rel); stream.recv_base_offset = stream.recv_offset; } } // 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. Loading Loading @@ -4915,6 +4930,21 @@ bool quic::isStreamComplete(uint64_t stream_id) const { return it->second.recv_complete; } void quic::flushFlowControl() { std::lock_guard<std::recursive_mutex> lock(quic_mtx()); if (!_handshake_complete || _conn_state.load() != ConnectionState::Connected) return; if (_pending_max_stream_data.empty() && !_pending_max_data) return; for (auto& [sid, limit] : _pending_max_stream_data) { sendMaxStreamData(sid, limit); } _pending_max_stream_data.clear(); if (_pending_max_data) { sendMaxData(_pending_max_data_value); _pending_max_data = false; } flushBatch(); } bool quic::isStreamReset(uint64_t stream_id) const { std::lock_guard<std::recursive_mutex> lock(quic_mtx()); Loading Loading @@ -5050,6 +5080,13 @@ void quic::processFrame(const uint8_t* data, size_t len, size_t& offset, bool& a it->second.reset = true; it->second.recv_fin = true; it->second.recv_fin_offset = final_size; // Pull dispatch: a reader may be waiting for bytes that will now never // come -- give it its final signal so it finds isStreamReset() at once. if (_is_server && _stream_pull_dispatch && (sid % 4 == 0) && _stream_callback && !it->second.recv_dispatched) { it->second.recv_dispatched = true; _pending_dispatches.push_back({sid, {}, true}); } } (void)app_error; // Deliberately not retired here: the application may still Loading Loading @@ -5781,8 +5818,11 @@ void quic::processStreamFrame(const uint8_t* data, size_t len, size_t& offset, u // recv_buffer.size() when recv_base_offset is 0 (always true outside // incremental dispatch's trimming), and still correct once trimming // has shifted the buffer forward. // Pull dispatch (setStreamPullDispatch): credit follows recvStreamData()'s consumption // 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 (consumed > stream.max_data_local_committed / 2) { if (!pull_stream && consumed > stream.max_data_local_committed / 2) { uint64_t new_limit = consumed + stream.max_data_local; stream.max_data_local_committed = new_limit; _pending_max_stream_data[stream_id] = new_limit; Loading Loading @@ -5812,7 +5852,18 @@ void quic::processStreamFrame(const uint8_t* data, size_t len, size_t& offset, u // 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 && _incremental_stream_dispatch) { if (pull_stream && _stream_callback) { // Pull dispatch: announce, don't hand over -- the data stays in recv_buffer for // recvStreamData(). One signal per batch of newly contiguous bytes, plus exactly one // fin=true signal once the stream is complete. if (stream.recv_contiguous > stream.recv_notified_offset || (stream.recv_complete && !stream.recv_dispatched)) { stream.recv_notified_offset = stream.recv_contiguous; bool complete = stream.recv_complete; if (complete) stream.recv_dispatched = true; _pending_dispatches.push_back({stream_id, {}, complete}); } } else 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 Loading
src/socket.h +28 −1 Original line number Diff line number Diff line Loading @@ -651,7 +651,11 @@ namespace netplus { // QUIC Protocol Implementation (RFC 9000) // UDP-based transport with TLS 1.3 encryption // =================================================== class quic : public udp { // enable_shared_from_this: lets code handed a raw quic* (e.g. a stream callback) pin an // accepted connection -- always make_shared (see accept()) -- while it works on one of its // streams off-thread. weak_from_this() is empty for a listener/client not owned by a // shared_ptr. class quic : public udp, public std::enable_shared_from_this<quic> { public: // QUIC version constants static constexpr uint32_t QUIC_VERSION_1 = 0x00000001; Loading Loading @@ -735,6 +739,7 @@ namespace netplus { bool recv_fin = false; bool recv_dispatched = false; uint64_t recv_dispatched_offset = 0; // bytes already handed to the stream callback (incremental dispatch only) uint64_t recv_notified_offset = 0; // contiguous bytes already announced to the callback (pull dispatch only) bool recv_complete = false; // all data up to FIN received bool send_fin = false; bool reset = false; Loading Loading @@ -1004,6 +1009,27 @@ namespace netplus { // across calls (or opt out and keep today's behavior). void setIncrementalStreamDispatch(bool enable) { _incremental_stream_dispatch = enable; } // Opt-in, default off, server side (inherited by accepted children like the flag // above): peer-initiated bidirectional streams are no longer handed to the stream // callback as data. The callback is only a "bytes are ready" signal -- empty data, // fin=false while the stream is open, one fin=true call once it's complete -- and the // application pulls the bytes with recvStreamData(). Stream-level flow-control credit // is then granted as recvStreamData() consumes, not as data arrives, and consumed // bytes are trimmed from the receive buffer, so memory per stream stays within its // receive window however slowly the application reads (e.g. relaying an upload to a // slower upstream). Signals can arrive from several pumping threads and out of order // relative to each other; recvStreamData() is the ordered source of truth. Takes // precedence over setIncrementalStreamDispatch() for those streams; unidirectional // streams keep their normal dispatch. void setStreamPullDispatch(bool enable) { _stream_pull_dispatch = enable; } bool streamPullDispatch() const { return _stream_pull_dispatch; } // Sends any MAX_DATA / MAX_STREAM_DATA credit recvStreamData() has queued right away, // instead of with the next packet the connection happens to send -- for a reader // outside the pump loop (pull dispatch), whose peer may be blocked on exactly that // credit and have nothing else to send that would trigger it. void flushFlowControl(); // Opt-in, default off: seed every accepted child connection's // _dispatch_promoted as already true, instead of waiting for its // first dispatch to be observed slower than kSlowDispatchThreshold. Loading Loading @@ -1938,6 +1964,7 @@ namespace netplus { using QuicStreamCallback = std::function<void(quic*, uint64_t, const std::vector<uint8_t>&, bool)>; QuicStreamCallback _stream_callback; bool _incremental_stream_dispatch = false; // see setIncrementalStreamDispatch() bool _stream_pull_dispatch = false; // see setStreamPullDispatch() bool _start_dispatch_promoted = false; // see setStartDispatchPromoted() // Deferred stream dispatches — filled inside _quic_mutex, Loading