h1dispatch added (833e0326) · Commits · jan.koester / libhttppp

src/httpd.cpp

+75 −25

Original line number	Diff line number	Diff line
		@@ -374,10 +374,13 @@ static std::vector<uint8_t> h3BuildResponse(uint16_t status_code,
		}

		libhttppp::HttpEvent::HttpEvent(std::vector<netplus::socket*> serversocket, int timeout,
		size_t h2OffloadThreads, int idleTimeoutSeconds)
		size_t h2OffloadThreads, int idleTimeoutSeconds,
		size_t h1OffloadThreads)
		: netplus::event(serversocket, timeout, idleTimeoutSeconds) {
		if (h2OffloadThreads > 0)
		_h2DispatchPool = std::make_unique<netplus::ThreadPool>(h2OffloadThreads);
		if (h1OffloadThreads > 0)
		_h1DispatchPool = std::make_unique<netplus::ThreadPool>(h1OffloadThreads);
		// Bounded like quicDispatchPool() in libnetplus (same rationale: this
		// work is dominated by waiting on upstream/network, not CPU, so a
		// generous multiple of core count is cheap) -- caps how many OS
		@@ -572,6 +575,67 @@ bool libhttppp::HttpEvent::_dispatchH2Stream(HttpRequest &cureq,
		return false;
		}

		void libhttppp::HttpEvent::_dispatchH1Request(HttpRequest &cureq, size_t consumeBodyBytes,
		const int tid, ULONG_PTR args) {
		// Bounds the blocking response-flush step below (see this function's doc comment in
		// httpd.h for why the socket has to be in blocking mode at all here) -- generous headroom
		// above any real client's RTT, just enough to stop a peer that's stopped reading entirely
		// from pinning a pool thread forever, since nothing else (no EPOLLOUT re-arm, no idle
		// reaper) can reach a detached connection to notice that.
		constexpr int kH1OffloadSocketTimeoutMs = 30000;

		if (_h1DispatchPool && shouldOffloadH1Dispatch(cureq)) {
		int fd = (!cureq.slots.empty() && cureq.slots[0].csock) ? cureq.slots[0].csock->fd() : -1;
		std::shared_ptr<netplus::con> connOwner = fd >= 0 ? netplus::lookupConnection(fd) : nullptr;
		std::unique_ptr<netplus::socket> sock = connOwner ? netplus::detachConnection(cureq) : nullptr;
		if (connOwner && sock) {
		cureq.slots[0].csock = std::move(sock);
		cureq.slots[0].csock->setBlock();
		cureq.slots[0].csock->setTimeout(kH1OffloadSocketTimeoutMs);

		_h1DispatchPool->submit(
		[this, connOwner, consumeBodyBytes, tid, args]() mutable {
		HttpRequest &cureq2 = static_cast<HttpRequest&>(*connOwner);
		try {
		// The blocking work this whole detach/reattach dance exists to keep
		// off the shared epoll/kqueue workers.
		RequestEvent(cureq2, tid, args);
		if (consumeBodyBytes > 0) {
		cureq2.RecvData.erase(cureq2.RecvData.begin(),
		cureq2.RecvData.begin() + consumeBodyBytes);
		}
		cureq2._RequestType = PARSEREQUEST;
		if (!cureq2.flushSendData()) {
		// Peer stopped reading within our bounded blocking window --
		// nothing left to retry against (no future EPOLLOUT will ever
		// come for a detached fd), so give up on the connection rather
		// than reattach it half-sent.
		cureq2.slots[0].csock->close();
		return;
		}
		} catch (const std::exception &) {
		// Either RequestEvent or the flush hit a hard error (peer reset,
		// socket-level failure) -- nothing to reattach.
		cureq2.slots[0].csock->close();
		return;
		}
		cureq2.slots[0].csock->setNonBlock();
		netplus::reattachConnection(connOwner);
		});
		return;
		}
		// Connection lookup/detach 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, mirrors _dispatchH2Stream's identical fallback.
		}

		RequestEvent(cureq, tid, args);
		if (consumeBodyBytes > 0) {
		cureq.RecvData.erase(cureq.RecvData.begin(), cureq.RecvData.begin() + consumeBodyBytes);
		}
		cureq._RequestType = PARSEREQUEST;
		}

		void libhttppp::HttpEvent::_finishH2Dispatch(HttpRequest &cureq,
		std::string &out,
		uint32_t sid,
		@@ -1967,20 +2031,16 @@ REQUESTHANDLING:
		goto REQUESTHANDLING;
		break;
		case GETREQUEST:
		RequestEvent(cureq,tid,args);
		cureq._RequestType=PARSEREQUEST;
		_dispatchH1Request(cureq, 0, tid, args);
		break;
		case DELETEREQUEST:
		RequestEvent(cureq,tid,args);
		cureq._RequestType=PARSEREQUEST;
		_dispatchH1Request(cureq, 0, tid, args);
		break;
		case OPTIONSREQUEST:
		RequestEvent(cureq,tid,args);
		cureq._RequestType=PARSEREQUEST;
		_dispatchH1Request(cureq, 0, tid, args);
		break;
		case HEADREQUEST:
		RequestEvent(cureq,tid,args);
		cureq._RequestType=PARSEREQUEST;
		_dispatchH1Request(cureq, 0, tid, args);
		break;
		case PUTREQUEST:
		case PATCHREQUEST: {
		@@ -1994,19 +2054,14 @@ REQUESTHANDLING:
		throw re;
		}
		size_t clen = cureq.getContentLength();
		RequestEvent(cureq,tid,args);
		cureq.RecvData.erase(cureq.RecvData.begin(),cureq.RecvData.begin()+clen);
		cureq._RequestType=PARSEREQUEST;
		_dispatchH1Request(cureq, clen, tid, args);
		break;
		}
		size_t clen = cureq.getContentLength();
		if(clen == 0){
		RequestEvent(cureq,tid,args);
		cureq._RequestType=PARSEREQUEST;
		_dispatchH1Request(cureq, 0, tid, args);
		} else if(cureq.RecvData.size()>=clen){
		RequestEvent(cureq,tid,args);
		cureq.RecvData.erase(cureq.RecvData.begin(),cureq.RecvData.begin()+clen);
		cureq._RequestType=PARSEREQUEST;
		_dispatchH1Request(cureq, clen, tid, args);
		}
		break;
		}
		@@ -2021,19 +2076,14 @@ REQUESTHANDLING:
		throw re;
		}
		size_t clen = cureq.getContentLength();
		RequestEvent(cureq,tid,args);
		cureq.RecvData.erase(cureq.RecvData.begin(),cureq.RecvData.begin()+clen);
		cureq._RequestType=PARSEREQUEST;
		_dispatchH1Request(cureq, clen, tid, args);
		break;
		}
		size_t clen = cureq.getContentLength();
		if(clen == 0){
		RequestEvent(cureq,tid,args);
		cureq._RequestType=PARSEREQUEST;
		_dispatchH1Request(cureq, 0, tid, args);
		} else if(cureq.RecvData.size()>=clen){
		RequestEvent(cureq,tid,args);
		cureq.RecvData.erase(cureq.RecvData.begin(),cureq.RecvData.begin()+clen);
		cureq._RequestType=PARSEREQUEST;
		_dispatchH1Request(cureq, clen, tid, args);
		}
		break;
		}

src/httpd.h

+49 −1

Original line number	Diff line number	Diff line
		@@ -61,8 +61,16 @@ namespace libhttppp {
		// with no genuine read/write activity. 0 (the default) preserves the original
		// behavior: a connection stays open until the peer closes it or a transport error
		// occurs, however long that takes.
		// h1OffloadThreads: HTTP/1.x analogue of h2OffloadThreads (see shouldOffloadH1Dispatch)
		// -- zero (the default) preserves the original fully-synchronous H1 dispatch behavior
		// for every existing caller. Unlike H2, an H1 connection has no per-stream separation
		// (cureq is the connection, one request in flight at a time), so offloading it
		// safely needs a detach/reattach round trip through netplus::detachConnection()/
		// reattachConnection() rather than H2's "hand off a throwaway tempreq" trick -- see
		// _dispatchH1Request's doc comment for the full mechanics.
		HttpEvent(std::vector<netplus::socket*> serversocket,int timeout = 1000,
		size_t h2OffloadThreads = 0, int idleTimeoutSeconds = 0);
		size_t h2OffloadThreads = 0, int idleTimeoutSeconds = 0,
		size_t h1OffloadThreads = 0);

		// Return true to have this stream's RequestEvent run on the H2
		// offload thread pool instead of inline in the frame-processing
		@@ -75,6 +83,16 @@ namespace libhttppp {
		return false;
		}

		// Return true to have this HTTP/1.x request's RequestEvent run on the H1 offload
		// thread pool instead of inline on the epoll/kqueue worker that read it. Called right
		// after the request is fully parsed (headers and, for a body-bearing method, the
		// complete body already buffered), before RequestEvent runs. Defaults to false --
		// every route stays on the original synchronous path unless a subclass opts in. Has
		// no effect if h1OffloadThreads is 0.
		virtual bool shouldOffloadH1Dispatch(HttpRequest &cureq) 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);
		virtual void ConnectEvent(HttpRequest &curreq,const int tid,ULONG_PTR args);
		@@ -144,6 +162,10 @@ namespace libhttppp {
		// constructor. See shouldOffloadH2Dispatch / _dispatchH2Stream.
		std::unique_ptr<netplus::ThreadPool> _h2DispatchPool;

		// Non-null only when h1OffloadThreads > 0 was passed to the constructor. See
		// shouldOffloadH1Dispatch / _dispatchH1Request.
		std::unique_ptr<netplus::ThreadPool> _h1DispatchPool;

		// Always constructed (unlike _h2DispatchPool, which is opt-in):
		// every streaming H3 response needs somewhere to run its
		// continuation loop (see Http3StreamEvent). Bounded so a burst of
		@@ -161,6 +183,32 @@ namespace libhttppp {
		const std::vector<hpack::HeaderField> &decoded,
		const std::string &reqBody,
		const int tid, ULONG_PTR args);
		// Runs RequestEvent(cureq,...) for a fully-parsed HTTP/1.x request, offloading to
		// _h1DispatchPool when shouldOffloadH1Dispatch() opts in. consumeBodyBytes is how many
		// already-fully-buffered request body bytes to erase from RecvData once RequestEvent
		// has returned (0 for GET/DELETE/OPTIONS/HEAD, which never have one) -- mirrors
		// exactly what each REQUESTHANDLING case in RequestEvent(netplus::con&,...) used to do
		// inline before this existed.
		//
		// Unlike _dispatchH2Stream (which hands a throwaway per-stream tempreq to the pool,
		// since H2 multiplexes many streams per connection), H1's cureq is the connection --
		// there is no separate object to hand off while leaving the epoll worker free to keep
		// servicing the same fd. Offloading therefore round-trips through
		// netplus::detachConnection()/reattachConnection(): detach before submitting (so the
		// fd leaves the epoll/kqueue interest set and no second dispatch can ever race the
		// in-flight one), run RequestEvent + the body-erase + the response flush on the pool
		// thread with the socket in blocking mode (HttpResponse::send() only ever appends to
		// SendData -- something has to actually write it, and off the event loop nothing will
		// do that later the way EPOLLOUT normally would), then reattach so keep-alive/
		// pipelining resumes normally. A peer that stops reading mid-flush past the bounded
		// timeout, or any exception, closes the connection instead of reattaching it
		// half-sent.
		//
		// Always leaves cureq fully handled by the time this returns: either it ran
		// (synchronously) right here, or it's been handed to the pool and the caller must not
		// touch cureq again.
		void _dispatchH1Request(HttpRequest &cureq, size_t consumeBodyBytes,
		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