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

src/httpd.cpp

+83 −25

Changes for src/httpd.cpp: 83 added lines, 25 removed lines.
Original line number	Diff line number	Diff line
		@@ -290,7 +290,10 @@ static std::vector<uint8_t> h3BuildResponse(uint16_t status_code,
		return out;
		}

		libhttppp::HttpEvent::HttpEvent(std::vector<netplus::socket*> serversocket, int timeout) : netplus::event(serversocket, timeout) {
		libhttppp::HttpEvent::HttpEvent(std::vector<netplus::socket*> serversocket, int timeout, size_t h2OffloadThreads)
		: netplus::event(serversocket, timeout) {
		if (h2OffloadThreads > 0)
		_h2DispatchPool = std::make_unique<netplus::ThreadPool>(h2OffloadThreads);
		// Set the HTTP/3 stream callback on QUIC server sockets so that
		// child connections created by accept() inherit it. The event loop
		// never creates a con / calls ConnectEvent for QUIC children,
		@@ -406,8 +409,57 @@ void libhttppp::HttpEvent::_dispatchH2Stream(HttpRequest &cureq,
		tempreq.setHeaderData("content-length")->push_back(std::to_string(reqBody.size()));
		}

		if (_h2DispatchPool && shouldOffloadH2Dispatch(tempreq, sid)) {
		int connFd = cureq.slots[0].csock ? cureq.slots[0].csock->fd() : -1;
		std::shared_ptr<netplus::con> connOwner =
		connFd >= 0 ? netplus::lookupConnection(connFd) : nullptr;
		if (connOwner) {
		// std::function (what ThreadPool::submit takes) requires its
		// target to be copy-constructible even though it's only ever
		// moved/invoked here — a bare move-captured unique_ptr fails
		// that static check. Box it in a shared_ptr (copy-constructible
		// regardless of the pointee) purely to satisfy that; ownership
		// stays effectively unique since nothing else ever sees this
		// particular shared_ptr.
		auto trBox = std::make_shared<std::unique_ptr<HttpRequest>>(std::move(tempreq_ptr));
		_h2DispatchPool->submit(
		[this, connOwner, trBox, sid, tid, args]() mutable {
		// The ONLY blocking call in this task — touches
		// *trBox (private to this task) only, never
		// cureq/h2state().
		RequestEvent(**trBox, tid, args);

		HttpRequest &cureq2 = static_cast<HttpRequest&>(*connOwner);
		std::lock_guard<std::recursive_mutex> lk(cureq2.event_mutex);
		if (!cureq2.slots[0].csock)
		return; // connection torn down while we were blocked

		std::string out2;
		_finishH2Dispatch(cureq2, out2, sid, std::move(*trBox), tid, args);
		if (!out2.empty())
		cureq2.SendData.append(out2.data(), out2.size());
		_flushSendDataNow(cureq2);
		if (cureq2.SendData.pos < cureq2.SendData.size())
		netplus::requestWritablePoll(cureq2);
		});
		return;
		}
		// Connection lookup failed (shouldn't happen while we're still
		// holding event_mutex for this call) — fall through to the
		// synchronous path rather than silently dropping the request.
		}

		// Dispatch to user's virtual RequestEvent
		RequestEvent(tempreq, tid, args);
		_finishH2Dispatch(cureq, out, sid, std::move(tempreq_ptr), tid, args);
		}

		void libhttppp::HttpEvent::_finishH2Dispatch(HttpRequest &cureq,
		std::string &out,
		uint32_t sid,
		std::unique_ptr<HttpRequest> tempreq_ptr,
		const int tid, ULONG_PTR args) {
		HttpRequest &tempreq = *tempreq_ptr;

		// Read structured response info from :res-* pseudo-headers
		auto *resValid = tempreq.getHeaderData(":res-valid");
		@@ -629,6 +681,35 @@ void libhttppp::HttpEvent::_dispatchH2Stream(HttpRequest &cureq,
		}
		}

		bool libhttppp::HttpEvent::_flushSendDataNow(HttpRequest &cureq) {
		if (!(cureq.SendData.pos < cureq.SendData.size() && cureq.slots[0].csock))
		return cureq.SendData.pos >= cureq.SendData.size();

		try {
		while (cureq.SendData.pos < cureq.SendData.size()) {
		size_t remaining = cureq.SendData.size() - cureq.SendData.pos;
		size_t sendlen = std::min(remaining, static_cast<size_t>(16384));
		netplus::buffer buf(cureq.SendData.data() + cureq.SendData.pos, sendlen);
		size_t sent = cureq.slots[0].csock->sendData(buf, 0);
		if (sent > 0) {
		cureq.SendData.pos += sent;
		if (cureq.slots[0].csock->hasPendingWrite())
		cureq.slots[0].csock->flush_out();
		} else {
		break;
		}
		}
		} catch (...) {
		// Socket busy or error — remaining data will be flushed
		// by IoEventHandler's send loop after we return.
		}
		if (cureq.SendData.pos >= cureq.SendData.size()) {
		cureq.SendData.clear();
		cureq.SendData.pos = 0;
		}
		return cureq.SendData.pos >= cureq.SendData.size();
		}

		void libhttppp::HttpEvent::_resumeH2Streams(HttpRequest &cureq,
		std::string &out,
		const int /tid/, ULONG_PTR /args/) {
		@@ -1145,30 +1226,7 @@ done:
		// all remaining dispatches. Without this, a slow RequestEvent
		// (e.g. on-demand preview rendering) serialises all streams and the
		// client may time out before later streams are served.
		if (cureq.SendData.pos < cureq.SendData.size() && cureq.slots[0].csock) {
		try {
		while (cureq.SendData.pos < cureq.SendData.size()) {
		size_t remaining = cureq.SendData.size() - cureq.SendData.pos;
		size_t sendlen = std::min(remaining, static_cast<size_t>(16384));
		netplus::buffer buf(cureq.SendData.data() + cureq.SendData.pos, sendlen);
		size_t sent = cureq.slots[0].csock->sendData(buf, 0);
		if (sent > 0) {
		cureq.SendData.pos += sent;
		if (cureq.slots[0].csock->hasPendingWrite())
		cureq.slots[0].csock->flush_out();
		} else {
		break;
		}
		}
		} catch (...) {
		// Socket busy or error — remaining data will be flushed
		// by IoEventHandler's send loop after we return.
		}
		if (cureq.SendData.pos >= cureq.SendData.size()) {
		cureq.SendData.clear();
		cureq.SendData.pos = 0;
		}
		}
		_flushSendDataNow(cureq);

		if (off > 0) {
		cureq.RecvData.erase(cureq.RecvData.begin(), cureq.RecvData.begin() + off);

src/httpd.h

+39 −1

Changes for src/httpd.h: 39 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -27,6 +27,7 @@

		#include <netplus/socket.h>
		#include <netplus/eventapi.h>
		#include <netplus/threadpool.h>
		#include <atomic>
		#include <cstdint>
		#include <map>
		@@ -49,7 +50,25 @@ namespace cmdplus {
		namespace libhttppp {
		class HttpEvent : public netplus::event {
		public:
		HttpEvent(std::vector<netplus::socket*> serversocket,int timeout = 1000);
		// h2OffloadThreads: when non-zero, spins up a background thread pool
		// (see shouldOffloadH2Dispatch) that lets HTTP/2 streams whose
		// RequestEvent does slow, blocking work (e.g. a backend network
		// round-trip) run without blocking every other stream multiplexed
		// on the same connection. Zero (the default) preserves the original
		// fully-synchronous H2 dispatch behavior for every existing caller.
		HttpEvent(std::vector<netplus::socket*> serversocket,int timeout = 1000,
		size_t h2OffloadThreads = 0);

		// Return true to have this stream's RequestEvent run on the H2
		// offload thread pool instead of inline in the frame-processing
		// loop. Called right after the per-stream request's headers have
		// been parsed (so :path/:method are already available), before
		// RequestEvent runs. Defaults to false — every route stays on the
		// original synchronous path unless a subclass opts a specific
		// route in. Has no effect if h2OffloadThreads is 0.
		virtual bool shouldOffloadH2Dispatch(HttpRequest &tempreq, uint32_t streamId) const {
		return false;
		}

		virtual void RequestEvent(HttpRequest &curreq,const int tid,ULONG_PTR args);
		virtual void ResponseEvent(HttpRequest &curreq,const int tid,ULONG_PTR args);
		@@ -115,11 +134,30 @@ namespace libhttppp {
		std::mutex _h3BufferMutex;
		std::map<uint64_t, H3StreamState> _h3StreamStates;
		std::atomic<int> _h3NextTid{0};

		// Non-null only when h2OffloadThreads > 0 was passed to the
		// constructor. See shouldOffloadH2Dispatch / _dispatchH2Stream.
		std::unique_ptr<netplus::ThreadPool> _h2DispatchPool;

		void _dispatchH2Stream(HttpRequest &cureq, std::string &out,
		uint32_t sid,
		const std::vector<hpack::HeaderField> &decoded,
		const std::string &reqBody,
		const int tid, ULONG_PTR args);
		// The part of stream dispatch that must run on the connection's
		// owning thread: extracts the plugin's :res-* response headers off
		// an already-completed tempreq, HPACK-encodes them, and frames the
		// response (or sets up activeStreams for a streaming response).
		// Shared by both the synchronous path and the offload-completion
		// path so they behave identically once RequestEvent has returned.
		void _finishH2Dispatch(HttpRequest &cureq, std::string &out,
		uint32_t sid,
		std::unique_ptr<HttpRequest> tempreq,
		const int tid, ULONG_PTR args);
		// Factored out of the synchronous dispatch loop's immediate-flush
		// step so the offload-completion path (which has no "loop" to fall
		// through to) can reuse the exact same send behavior.
		bool _flushSendDataNow(HttpRequest &cureq);
		void _resumeH2Streams(HttpRequest &cureq, std::string &out,
		const int tid, ULONG_PTR args);
		void _reapStalledH2Streams(HttpRequest &cureq, std::string &out);