test (400fe580) · Commits · jan.koester / libhttppp

src/http.cpp

+10 −0

Changes for src/http.cpp: 10 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -2992,6 +2992,7 @@ const std::vector<char> libhttppp::HttpClient::_h2Request(
		// vector overload sends nothing for empty data), so the last DATA frame has to
		// carry it -- and a chunk is only known to be last once the next read returns 0.
		std::vector<uint8_t> cur(64 * 1024), next(64 * 1024);
		try {
		size_t curLen = (bodySource)(reinterpret_cast<char >(cur.data()), cur.size());
		q->sendStreamData(stream_id, stream_payload, curLen == 0);
		while (curLen > 0) {
		@@ -3002,6 +3003,11 @@ const std::vector<char> libhttppp::HttpClient::_h2Request(
		cur.swap(next);
		curLen = nextLen;
		}
		} catch (netplus::NetException &) {
		// STOP_SENDING: the server answered without wanting the rest of the body
		// (RFC 9114 §4.1.2) -- its response is what matters now, go read it.
		if (!q->isStreamSendStopped(stream_id)) throw;
		}
		}

		std::vector<uint8_t> raw;
		@@ -3177,6 +3183,10 @@ const std::vector<char> libhttppp::HttpClient::_h2Request(
		}
		}

		// Retire the stream: without this every request left its Stream -- and the response
		// bytes in its receive buffer -- in the connection's stream table for as long as the
		// connection lived (libnetplus retires a stream only on closeStream()/resetStream()).
		q->closeStream(stream_id);
		return ret;
		}

src/httpd.cpp

+52 −11

Changes for src/httpd.cpp: 52 added lines, 11 removed lines.
Original line number	Diff line number	Diff line
		@@ -2236,20 +2236,26 @@ bool libhttppp::HttpEvent::_startH3BodyStream(netplus::quic &q, uint64_t stream_
		const int h3tid = _h3NextTid.fetch_add(1, std::memory_order_relaxed) % std::max(threads, 1);
		auto finish = [this, pin, stream_id, reader, st](std::unique_ptr<HttpRequest> r, int tid) {
		// Whatever of the body RequestEvent didn't read (it answered first) is unwanted:
		// STOP_SENDING(H3_NO_ERROR) after the response, RFC 9114 §4.1.2.
		// STOP_SENDING(H3_NO_ERROR) after the response, RFC 9114 §4.1.2. The request side is
		// then done once the peer's RESET_STREAM (or the rest of the body) arrives -- see the
		// finished-state branch of _h3PullStream.
		const bool unread = !reader->finished();
		std::cerr << "[DBG-H3] finish sid=" << stream_id << " unread=" << unread << " resvalid=" << (r->getHeaderData(":res-valid") != nullptr) << std::endl;
		reader->abort();
		r->_bodyStream.reset();
		st->finishedAt = std::chrono::steady_clock::now();
		st->finished.store(true, std::memory_order_release);
		try {
		_finishH3Request(pin.get(), stream_id, std::move(r), tid);
		_finishH3Request(pin.get(), stream_id, std::move(r), tid,
		[this, pin, stream_id, st] {
		_h3StreamSideDone(pin, stream_id, st, false);
		});
		} catch (netplus::NetException &) {
		pin->resetStream(stream_id, 0x0102); // H3_INTERNAL_ERROR
		}
		if (unread && !pin->isStreamReset(stream_id))
		pin->stopSending(stream_id, 0x0100); // H3_NO_ERROR
		st->finishedAt = std::chrono::steady_clock::now();
		st->finished.store(true, std::memory_order_release);
		if (!unread \|\| pin->isStreamReset(stream_id) \|\| pin->isStreamComplete(stream_id))
		_h3StreamSideDone(pin, stream_id, st, true);
		};

		auto trBox = std::make_shared<std::unique_ptr<HttpRequest>>(std::move(req));
		@@ -2305,7 +2311,13 @@ void libhttppp::HttpEvent::_h3PullStream(netplus::quic &q, uint64_t stream_id) {
		st = slot;
		}
		std::lock_guard<std::mutex> lk(st->mtx);
		if (st->finished.load(std::memory_order_acquire)) return;
		if (st->finished.load(std::memory_order_acquire)) {
		// Request side ends after we're done with the stream when the peer still had body
		// in flight past an early answer: its RESET_STREAM (or the end of that body).
		if (q.isStreamReset(stream_id) \|\| q.isStreamComplete(stream_id))
		_h3StreamSideDone(q, stream_id, *st, true);
		return;
		}
		if (st->notifyReader) { // body belongs to an offloaded RequestEvent now
		st->notifyReader();
		return;
		@@ -2333,6 +2345,7 @@ void libhttppp::HttpEvent::_h3PullStream(netplus::quic &q, uint64_t stream_id) {
		if (pulled) q.flushFlowControl();
		if (q.isStreamReset(stream_id)) {
		forget();
		st->closed = true;
		q.resetStream(stream_id, 0x010c); // H3_REQUEST_CANCELLED -- also retires the stream
		return;
		}
		@@ -2379,11 +2392,30 @@ void libhttppp::HttpEvent::_h3PullStream(netplus::quic &q, uint64_t stream_id) {
		if (!complete) return;
		std::vector<uint8_t> completeData = std::move(st->data);
		forget();
		_h3StreamSideDone(q, stream_id, *st, true);
		if (st->streaming) {
		// The app answers on its own (sendH3StreamResponse etc.), no completion hook there.
		onH3DataChunk(&q, stream_id, nullptr, 0, true);
		return;
		}
		if (!completeData.empty()) _dispatchH3Complete(q, stream_id, completeData);
		if (completeData.empty()) return;
		std::shared_ptr<netplus::quic> pin = q.weak_from_this().lock();
		std::function<void()> onDone;
		if (pin) onDone = [this, pin, stream_id, st] { _h3StreamSideDone(pin, stream_id, st, false); };
		_dispatchH3Complete(q, stream_id, completeData, std::move(onDone));
		}

		void libhttppp::HttpEvent::_h3StreamSideDone(netplus::quic &q, uint64_t stream_id,
		H3PullState &st, bool requestSide) {
		bool close;
		{
		std::lock_guard<std::mutex> lk(st.doneMtx);
		(requestSide ? st.requestDone : st.responseDone) = true;
		close = st.requestDone && st.responseDone && !st.closed;
		if (close) st.closed = true;
		}
		// Both directions are over: FIN and STOP_SENDING are no-ops now, this just retires it.
		if (close) q.closeStream(stream_id);
		}

		void libhttppp::HttpEvent::Http3StreamEvent(netplus::socket *sock,
		@@ -2492,7 +2524,8 @@ void libhttppp::HttpEvent::Http3StreamEvent(netplus::socket *sock,
		// A complete buffered H3 request stream (HEADERS + all DATA): parse, run RequestEvent inline,
		// send the response.
		void libhttppp::HttpEvent::_dispatchH3Complete(netplus::quic &qref, uint64_t stream_id,
		const std::vector<uint8_t> &completeData) {
		const std::vector<uint8_t> &completeData,
		std::function<void()> onResponseDone) {
		netplus::quic *q = &qref;

		// Two-pass approach: first collect HEADERS and DATA payloads,
		@@ -2574,7 +2607,7 @@ void libhttppp::HttpEvent::_dispatchH3Complete(netplus::quic &qref, uint64_t str

		// Call user's RequestEvent
		RequestEvent(tempreq, h3tid, 0);
		_finishH3Request(q, stream_id, std::move(tempreq_ptr), h3tid);
		_finishH3Request(q, stream_id, std::move(tempreq_ptr), h3tid, std::move(onResponseDone));
		}

		// Turns a completed H3 request's :res-* pseudo-headers (what HttpResponse::send() left on the
		@@ -2583,7 +2616,11 @@ void libhttppp::HttpEvent::_dispatchH3Complete(netplus::quic &qref, uint64_t str
		// inline buffered path and the offloaded streamed-body path.
		void libhttppp::HttpEvent::_finishH3Request(netplus::quic *q, uint64_t stream_id,
		std::unique_ptr<HttpRequest> tempreq_ptr,
		int h3tid) {
		int h3tid,
		std::function<void()> onResponseDone) {
		auto responseDone = [&onResponseDone] {
		if (onResponseDone) onResponseDone();
		};
		HttpRequest &tempreq = *tempreq_ptr;

		// Read structured response info from :res-* pseudo-headers
		@@ -2675,6 +2712,7 @@ void libhttppp::HttpEvent::_finishH3Request(netplus::quic *q, uint64_t stream_id
		tempreq.deldata(":res-status");
		tempreq.deldata(":res-content-type");
		tempreq.deldata(":res-content-length");
		responseDone();
		return;
		}

		@@ -2695,7 +2733,7 @@ void libhttppp::HttpEvent::_finishH3Request(netplus::quic *q, uint64_t stream_id
		// trBox workaround for a move-only unique_ptr capture.
		auto trBox = std::make_shared<std::unique_ptr<HttpRequest>>(std::move(tempreq_ptr));
		_h3StreamPool->submit([this, q, stream_id, h3tid,
		trBox,
		trBox, onResponseDone,
		content_length, total_sent]() mutable {
		HttpRequest *tr = trBox->get();
		size_t sent = total_sent;
		@@ -2784,6 +2822,7 @@ void libhttppp::HttpEvent::_finishH3Request(netplus::quic *q, uint64_t stream_id
		// handling above.
		q->resetStream(stream_id, 0x0102); // H3_INTERNAL_ERROR
		}
		if (onResponseDone) onResponseDone();
		});

		return;
		@@ -2814,12 +2853,14 @@ void libhttppp::HttpEvent::_finishH3Request(netplus::quic *q, uint64_t stream_id
		tempreq.deldata(":res-status");
		tempreq.deldata(":res-content-type");
		tempreq.deldata(":res-content-length");
		responseDone();
		} else {
		// The application never produced a response for this H3 request
		// (mirrors the H2 case in _finishH2Dispatch). Without this, the
		// stream is left open with nothing ever sent, and the client
		// hangs forever waiting for a reply that will never arrive.
		q->resetStream(stream_id, 0x0102); // H3_INTERNAL_ERROR
		responseDone();
		}
		}

src/httpd.h

+14 −2

Changes for src/httpd.h: 14 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -206,7 +206,15 @@ namespace libhttppp {
		std::chrono::steady_clock::time_point finishedAt{}; // valid once finished
		std::weak_ptr<netplus::quic> conn; // expired = connection gone
		std::function<void()> notifyReader; // streamed body: wake its reader
		// closeStream() once both sides are through (see _h3StreamSideDone) -- libnetplus
		// only retires a stream there, so skipping it leaks the stream and its buffer.
		std::mutex doneMtx;
		bool requestDone = false;
		bool responseDone = false;
		bool closed = false;
		};
		void _h3StreamSideDone(netplus::quic &q, uint64_t stream_id, H3PullState &st,
		bool requestSide);
		std::map<std::pair<netplus::quic*, uint64_t>, std::shared_ptr<H3PullState>> _h3PullStates;
		std::chrono::steady_clock::time_point _h3PullPruneAt{};

		@@ -217,10 +225,14 @@ namespace libhttppp {
		bool _startH3BodyStream(netplus::quic &q, uint64_t stream_id,
		std::unique_ptr<HttpRequest> &req,
		std::vector<uint8_t> leftover);
		// onResponseDone: called once the response has been fully sent (or the stream reset),
		// possibly much later from _h3StreamPool for a streamed response.
		void _dispatchH3Complete(netplus::quic &q, uint64_t stream_id,
		const std::vector<uint8_t> &completeData);
		const std::vector<uint8_t> &completeData,
		std::function<void()> onResponseDone = nullptr);
		void _finishH3Request(netplus::quic *q, uint64_t stream_id,
		std::unique_ptr<HttpRequest> tempreq, int h3tid);
		std::unique_ptr<HttpRequest> tempreq, int h3tid,
		std::function<void()> onResponseDone = nullptr);
		std::atomic<int> _h3NextTid{0};

		// Non-null only when h2OffloadThreads > 0 was passed to the