test (ddc744f0) · Commits · jan.koester / libhttppp

src/httpd.cpp

+344 −6

Changes for src/httpd.cpp: 344 added lines, 6 removed lines.
Original line number	Diff line number	Diff line
		@@ -406,6 +406,8 @@ libhttppp::HttpEvent::HttpEvent(std::vector<netplus::socket*> serversocket, int
		// so this is the only opportunity to wire up Http3StreamEvent.
		for (auto* sock : serversocket) {
		if (auto* quicsock = dynamic_cast<netplus::quic*>(sock)) {
		// Request bodies are pulled, not pushed: see _h3PullStream.
		quicsock->setStreamPullDispatch(true);
		quicsock->setStreamCallback([this](netplus::socket *s,
		uint64_t stream_id,
		const std::vector<uint8_t> &data,
		@@ -2077,18 +2079,334 @@ done:



		namespace {

		// Body reader for a streamed H3 request (shouldStreamRequestBody). Pulls the stream's bytes
		// from QUIC itself as RequestEvent asks for them -- under pull dispatch QUIC only returns
		// stream credit for what's pulled, so a slow consumer throttles the client instead of filling
		// memory -- and strips the H3 framing: DATA payloads are the body; any other frame (trailing
		// HEADERS, reserved/unknown types) is skipped.
		class H3RequestBodyReader : public libhttppp::RequestBodyStream {
		public:
		H3RequestBodyReader(std::shared_ptr<netplus::quic> q, uint64_t sid, std::vector<uint8_t> leftover)
		: _q(std::move(q)), _sid(sid), _raw(std::move(leftover)) {}

		size_t read(char *buf, size_t len) override {
		if (_done \|\| len == 0) return 0;
		auto lastProgress = std::chrono::steady_clock::now();
		for (;;) {
		if (_aborted) fail("request body: aborted");
		const size_t avail = _raw.size() - _off;
		if (_payloadLeft > 0 && avail > 0) {
		size_t n = std::min<size_t>({len, static_cast<size_t>(_payloadLeft), avail});
		std::memcpy(buf, _raw.data() + _off, n);
		_off += n;
		_payloadLeft -= n;
		compact();
		return n;
		}
		if (_skipLeft > 0 && avail > 0) {
		size_t k = std::min<size_t>(static_cast<size_t>(_skipLeft), avail);
		_off += k;
		_skipLeft -= k;
		compact();
		continue;
		}
		if (_payloadLeft == 0 && _skipLeft == 0 && avail > 0) {
		size_t tb = 0, lb = 0;
		uint64_t type = h3DecodeVarInt(_raw.data() + _off, avail, tb);
		uint64_t flen = tb ? h3DecodeVarInt(_raw.data() + _off + tb, avail - tb, lb) : 0;
		if (tb && lb) {
		_off += tb + lb;
		(type == 0x00 ? _payloadLeft : _skipLeft) = flen;
		continue;
		}
		// incomplete frame header -- need more bytes
		}
		if (pull()) {
		lastProgress = std::chrono::steady_clock::now();
		continue;
		}
		if (_q->isStreamReset(_sid)) fail("request body: stream reset by peer");
		if (_q->isStreamComplete(_sid) && !_q->hasStreamData(_sid)) {
		if (_payloadLeft \|\| _skipLeft \|\| _off < _raw.size())
		fail("request body: stream ended inside an H3 frame");
		_done = true;
		return 0;
		}
		if (std::chrono::steady_clock::now() - lastProgress > std::chrono::seconds(60))
		fail("request body: timed out waiting for data");
		std::unique_lock<std::mutex> lk(_mtx);
		const uint64_t seen = _signals;
		// Bounded: a signal racing ahead of this wait is caught by the counter, anything
		// else (a lost signal, a stream that silently went away) by the next poll.
		_cv.wait_for(lk, std::chrono::milliseconds(200),
		[&] { return _signals != seen \|\| _aborted.load(); });
		}
		}

		bool finished() const override { return _done; }
		void abort() override {
		{ std::lock_guard<std::mutex> lk(_mtx); _aborted = true; }
		_cv.notify_all();
		}
		void notify() {
		{ std::lock_guard<std::mutex> lk(_mtx); ++_signals; }
		_cv.notify_all();
		}

		private:
		[[noreturn]] static void fail(const char *msg) {
		libhttppp::HTTPException e;
		e[libhttppp::HTTPException::Error] << msg;
		throw e;
		}

		bool pull() {
		uint8_t tmp[65536];
		size_t n = _q->recvStreamData(_sid, tmp, sizeof(tmp));
		if (n == 0) return false;
		_raw.insert(_raw.end(), tmp, tmp + n);
		_q->flushFlowControl(); // the credit for these bytes, now -- the client may be waiting on it
		return true;
		}

		void compact() {
		if (_off == _raw.size()) {
		_raw.clear();
		_off = 0;
		} else if (_off >= 65536 && _off * 2 >= _raw.size()) {
		_raw.erase(_raw.begin(), _raw.begin() + static_cast<std::ptrdiff_t>(_off));
		_off = 0;
		}
		}

		std::shared_ptr<netplus::quic> _q;
		uint64_t _sid;
		std::vector<uint8_t> _raw;
		size_t _off = 0;
		uint64_t _payloadLeft = 0;
		uint64_t _skipLeft = 0;
		bool _done = false;
		std::atomic<bool> _aborted{false};
		std::mutex _mtx;
		std::condition_variable _cv;
		uint64_t _signals = 0;
		};

		bool isBodyMethod(int requestType) {
		switch (requestType) {
		case POSTREQUEST: case PUTREQUEST: case PATCHREQUEST:
		case PROPFINDREQUEST: case PROPPATCHREQUEST: case MKCOLREQUEST: case COPYREQUEST:
		case MOVEREQUEST: case LOCKREQUEST: case UNLOCKREQUEST: case REPORTREQUEST:
		case SEARCHREQUEST:
		return true;
		default:
		return false;
		}
		}

		} // namespace

		bool libhttppp::HttpEvent::_startH3BodyStream(netplus::quic &q, uint64_t stream_id,
		std::unique_ptr<HttpRequest> &req,
		std::vector<uint8_t> leftover) {
		if (!_h2DispatchPool \|\| !isBodyMethod(req->getRequestType())) return false;
		if (auto *cl = req->getHeaderData("content-length")) {
		if (cl->getfirstValue() && cl->getfirstValue()->getvalue() == "0") return false;
		}
		if (!shouldOffloadH2Dispatch(*req, static_cast<uint32_t>(stream_id)) \|\|
		!shouldStreamRequestBody(*req))
		return false;
		// The worker outlives this callback; it has to pin the connection.
		std::shared_ptr<netplus::quic> pin = q.weak_from_this().lock();
		if (!pin) return false;

		std::shared_ptr<H3PullState> st;
		{
		std::lock_guard<std::mutex> lk(_h3BufferMutex);
		auto it = _h3PullStates.find({&q, stream_id});
		if (it == _h3PullStates.end()) return false; // can't happen: our caller created it
		st = it->second;
		}
		auto reader = std::make_shared<H3RequestBodyReader>(pin, stream_id, std::move(leftover));
		req->_bodyStream = reader;
		st->notifyReader = [reader] { reader->notify(); }; // caller holds st->mtx

		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.
		const bool unread = !reader->finished();
		reader->abort();
		r->_bodyStream.reset();
		try {
		_finishH3Request(pin.get(), stream_id, std::move(r), tid);
		} 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);
		};

		auto trBox = std::make_shared<std::unique_ptr<HttpRequest>>(std::move(req));
		bool accepted = _h2DispatchPool->submit([this, trBox, h3tid, finish]() {
		try {
		RequestEvent(**trBox, h3tid, 0);
		} catch (...) {
		// No :res-valid then -- _finishH3Request resets the stream.
		}
		finish(std::move(*trBox), h3tid);
		});
		if (!accepted) {
		// Pool full (h2DispatchQueueMax): 503 right here, like the H2 path.
		try {
		HttpResponse busy;
		busy.setState(HTTP503);
		busy.setContentType("text/plain");
		static const std::string kBusyMsg = "server overloaded, try again shortly";
		busy.send(**trBox, kBusyMsg, static_cast<int>(kBusyMsg.size()));
		} catch (const HTTPException &) {
		}
		finish(std::move(*trBox), h3tid);
		}
		return true;
		}

		void libhttppp::HttpEvent::_h3PullStream(netplus::quic &q, uint64_t stream_id) {
		const std::pair<netplus::quic*, uint64_t> key{&q, stream_id};
		std::shared_ptr<H3PullState> st;
		{
		std::lock_guard<std::mutex> lk(_h3BufferMutex);
		auto &slot = _h3PullStates[key];
		// A tombstone left by an earlier connection that happened to get the same address.
		if (slot && slot->conn.expired()) slot.reset();
		if (!slot) {
		slot = std::make_shared<H3PullState>();
		slot->conn = q.weak_from_this();
		// Drop tombstones old enough that no signal for them can still be in flight, and
		// anything whose connection is gone (e.g. died mid-upload), at most once a second.
		const auto now = std::chrono::steady_clock::now();
		if (now >= _h3PullPruneAt) {
		_h3PullPruneAt = now + std::chrono::seconds(1);
		for (auto it = _h3PullStates.begin(); it != _h3PullStates.end();) {
		const auto &e = it->second;
		bool stale = e != slot &&
		(e->conn.expired() \|\|
		(e->finished.load(std::memory_order_acquire) &&
		now - e->finishedAt > std::chrono::seconds(10)));
		it = stale ? _h3PullStates.erase(it) : std::next(it);
		}
		}
		}
		st = slot;
		}
		std::lock_guard<std::mutex> lk(st->mtx);
		if (st->finished.load(std::memory_order_acquire)) return;
		if (st->notifyReader) { // body belongs to an offloaded RequestEvent now
		st->notifyReader();
		return;
		}
		auto forget = [&] {
		st->data.clear();
		st->data.shrink_to_fit();
		st->finishedAt = std::chrono::steady_clock::now();
		st->finished.store(true, std::memory_order_release);
		};

		// Pull what has arrived. Buffered requests hold their body here exactly as before
		// (credit returned as pulled); streamed ones stop being pulled here once dispatched.
		uint8_t buf[65536];
		bool pulled = false;
		auto drain = [&] {
		while (size_t n = q.recvStreamData(stream_id, buf, sizeof(buf))) {
		st->data.insert(st->data.end(), buf, buf + n);
		pulled = true;
		}
		};
		drain();
		const bool complete = q.isStreamComplete(stream_id);
		if (complete) drain(); // bytes that landed between the drain and the check
		if (pulled) q.flushFlowControl();
		if (q.isStreamReset(stream_id)) {
		forget();
		q.resetStream(stream_id, 0x010c); // H3_REQUEST_CANCELLED -- also retires the stream
		return;
		}

		while (st->parseOffset < st->data.size()) {
		const uint8_t *p = st->data.data() + st->parseOffset;
		size_t remaining = st->data.size() - st->parseOffset;
		size_t tb = 0, lb = 0;
		uint64_t frame_type = h3DecodeVarInt(p, remaining, tb);
		if (tb == 0 \|\| tb >= remaining) break;
		uint64_t frame_len = h3DecodeVarInt(p + tb, remaining - tb, lb);
		if (lb == 0 \|\| tb + lb + frame_len > remaining) break;
		const uint8_t *frame_data = p + tb + lb;
		const size_t frame_end = st->parseOffset + tb + lb + frame_len;

		if (frame_type == 0x01 && !st->headersParsed) {
		st->headersParsed = true;
		auto decoded = qpack::Decoder::decode(frame_data, frame_len);
		auto req = std::make_unique<HttpRequest>();
		req->parseH3(decoded);
		std::string peerAddr;
		q.getAddress(peerAddr);
		req->setPeerAddress(peerAddr);
		std::vector<uint8_t> leftover(st->data.begin() + frame_end, st->data.end());
		if (_startH3BodyStream(q, stream_id, req, std::move(leftover))) {
		st->data.clear();
		st->data.shrink_to_fit();
		st->parseOffset = 0;
		return;
		}
		st->streaming = onH3StreamHeaders(&q, stream_id, decoded);
		} else if (frame_type == 0x00 && st->streaming) {
		onH3DataChunk(&q, stream_id, reinterpret_cast<const char*>(frame_data), frame_len,
		false);
		}
		st->parseOffset = frame_end;
		}
		if (st->streaming && st->parseOffset > 0) {
		// Already handed to onH3DataChunk -- don't keep it.
		st->data.erase(st->data.begin(), st->data.begin() + st->parseOffset);
		st->parseOffset = 0;
		}

		if (!complete) return;
		std::vector<uint8_t> completeData = std::move(st->data);
		forget();
		if (st->streaming) {
		onH3DataChunk(&q, stream_id, nullptr, 0, true);
		return;
		}
		if (!completeData.empty()) _dispatchH3Complete(q, stream_id, completeData);
		}

		void libhttppp::HttpEvent::Http3StreamEvent(netplus::socket *sock,
		uint64_t stream_id,
		const std::vector<uint8_t> &data,
		bool fin) {
		// Pull dispatch (enabled on every QUIC socket this class serves, see the ctor): request
		// streams only signal "bytes ready" -- _h3PullStream() pulls them itself.
		if (auto pq = dynamic_cast<netplus::quic>(sock);
		pq && pq->streamPullDispatch() && stream_id % 4 == 0 && data.empty()) {
		_h3PullStream(*pq, stream_id);
		return;
		}

		if (data.empty() && !fin) {
		return;
		}
		// Stream ids are per connection (every connection's first request is stream 0).
		const std::pair<netplus::socket*, uint64_t> key{sock, stream_id};

		// Accumulate stream data
		{
		std::lock_guard<std::mutex> lock(_h3BufferMutex);
		auto &state = _h3StreamStates[stream_id];
		auto &state = _h3StreamStates[key];
		if (!data.empty()) {
		state.data.insert(state.data.end(), data.begin(), data.end());
		}
		@@ -2100,7 +2418,7 @@ void libhttppp::HttpEvent::Http3StreamEvent(netplus::socket *sock,
		H3StreamState *sp = nullptr;
		{
		std::lock_guard<std::mutex> lock(_h3BufferMutex);
		auto it = _h3StreamStates.find(stream_id);
		auto it = _h3StreamStates.find(key);
		if (it != _h3StreamStates.end()) sp = &it->second;
		}
		if (!sp) return;
		@@ -2144,7 +2462,7 @@ void libhttppp::HttpEvent::Http3StreamEvent(netplus::socket *sock,
		// Final callback with fin=true
		onH3DataChunk(sock, stream_id, nullptr, 0, true);
		std::lock_guard<std::mutex> lock(_h3BufferMutex);
		_h3StreamStates.erase(stream_id);
		_h3StreamStates.erase(key);
		return;
		}

		@@ -2152,7 +2470,7 @@ void libhttppp::HttpEvent::Http3StreamEvent(netplus::socket *sock,
		std::vector<uint8_t> completeData;
		{
		std::lock_guard<std::mutex> lock(_h3BufferMutex);
		auto it = _h3StreamStates.find(stream_id);
		auto it = _h3StreamStates.find(key);
		if (it != _h3StreamStates.end()) {
		completeData = std::move(it->second.data);
		_h3StreamStates.erase(it);
		@@ -2167,6 +2485,14 @@ void libhttppp::HttpEvent::Http3StreamEvent(netplus::socket *sock,
		if (!q) {
		return;
		}
		_dispatchH3Complete(*q, stream_id, completeData);
		}

		// 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) {
		netplus::quic *q = &qref;

		// Two-pass approach: first collect HEADERS and DATA payloads,
		// then dispatch. This ensures POST body data is available
		@@ -2224,9 +2550,9 @@ void libhttppp::HttpEvent::Http3StreamEvent(netplus::socket *sock,
		// path's identical comment) -- its `slots` is always empty, so stash
		// the real peer address from the QUIC socket this stream actually
		// arrived on while we still have it.
		if (sock) {
		{
		std::string peerAddr;
		sock->getAddress(peerAddr);
		q->getAddress(peerAddr);
		tempreq.setPeerAddress(peerAddr);
		}

		@@ -2247,6 +2573,17 @@ void libhttppp::HttpEvent::Http3StreamEvent(netplus::socket *sock,

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

		// Turns a completed H3 request's :res-* pseudo-headers (what HttpResponse::send() left on the
		// per-stream request) into the HEADERS/DATA response on its stream -- or the streaming
		// continuation on _h3StreamPool, or RESET_STREAM if no response was produced. Shared by the
		// 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) {
		HttpRequest &tempreq = *tempreq_ptr;

		// Read structured response info from :res-* pseudo-headers
		auto *resValid = tempreq.getHeaderData(":res-valid");
		@@ -2812,6 +3149,7 @@ void libhttppp::HttpEvent::ResponseEvent(netplus::con &curcon,const int tid,ULON

		void libhttppp::HttpEvent::ConnectEvent(netplus::con &curcon,const int tid,ULONG_PTR args){
		if (auto quicsock = dynamic_cast<netplus::quic>(curcon.slots[0].csock.get())) {
		quicsock->setStreamPullDispatch(true);
		quicsock->setStreamCallback([this](netplus::socket *sock,
		uint64_t stream_id,
		const std::vector<uint8_t> &data,

src/httpd.h

+37 −2

Changes for src/httpd.h: 37 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -118,7 +118,9 @@ namespace libhttppp {
		// with the request's method/URL/headers already parsed (an H2 call sees a scratch
		// request object, not the one RequestEvent later gets). Only honored where the request
		// is also offloaded (shouldOffloadH1Dispatch/shouldOffloadH2Dispatch true, with a
		// pool), since reading the body blocks; H3 bodies are still buffered. Default false.
		// pool), since reading the body blocks. H3 requests use the H2 pool and
		// shouldOffloadH2Dispatch (memory bound: the QUIC stream receive window); for H1 and
		// H3 only body-bearing methods (POST/PUT/PATCH/WebDAV) qualify. Default false.
		virtual bool shouldStreamRequestBody(HttpRequest &req) const {
		return false;
		}
		@@ -185,7 +187,40 @@ namespace libhttppp {
		size_t parseOffset = 0;
		};
		std::mutex _h3BufferMutex;
		std::map<uint64_t, H3StreamState> _h3StreamStates;
		// Keyed by connection too: stream ids restart at 0 on every QUIC connection.
		std::map<std::pair<netplus::socket*, uint64_t>, H3StreamState> _h3StreamStates;

		// Request streams under QUIC pull dispatch (see _h3PullStream): what's been pulled
		// so far, until either the stream is complete (buffered request) or its body has been
		// handed to an offloaded RequestEvent (streamed request, notifyReader set).
		// Finished entries stay behind as tombstones for a while: a signal for the stream can
		// still be in flight on another pumping thread, and must not recreate it as new (see
		// _h3PullStream's pruning).
		struct H3PullState {
		std::mutex mtx; // serializes signals for one stream
		std::vector<uint8_t> data;
		size_t parseOffset = 0;
		bool headersParsed = false;
		bool streaming = false; // onH3StreamHeaders() took it
		std::atomic<bool> finished{false};
		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
		};
		std::map<std::pair<netplus::quic*, uint64_t>, std::shared_ptr<H3PullState>> _h3PullStates;
		std::chrono::steady_clock::time_point _h3PullPruneAt{};

		// Handles one "bytes ready" signal for a request stream (pull dispatch).
		void _h3PullStream(netplus::quic &q, uint64_t stream_id);
		// Whether this request's body goes to an offloaded RequestEvent incrementally
		// (shouldStreamRequestBody), and if so starts it. Takes the bytes after HEADERS.
		bool _startH3BodyStream(netplus::quic &q, uint64_t stream_id,
		std::unique_ptr<HttpRequest> &req,
		std::vector<uint8_t> leftover);
		void _dispatchH3Complete(netplus::quic &q, uint64_t stream_id,
		const std::vector<uint8_t> &completeData);
		void _finishH3Request(netplus::quic *q, uint64_t stream_id,
		std::unique_ptr<HttpRequest> tempreq, int h3tid);
		std::atomic<int> _h3NextTid{0};

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