test (09d8e029) · Commits · jan.koester / libnetplus

src/event/iocp.cpp

+77 −4

Original line number	Diff line number	Diff line
		@@ -37,6 +37,8 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
		#include <signal.h>

		#include <atomic>
		#include <cstdlib>
		#include <ctime>
		#include <thread>
		#include <chrono>
		#include <vector>
		@@ -71,6 +73,14 @@ namespace netplus {
		std::vector<std::thread> workers;
		std::vector<std::thread> acceptThreads;
		int threads = 1;

		// Idle-connection reaper throttle (see reapIdleConnections() below). Unlike epoll/
		// kqueue's per-poll-instance _lastSweepAttempt (safe as a plain member there because
		// each poll object is thread-exclusive), EventState is shared by every worker thread
		// for this event instance -- they all call GetQueuedCompletionStatus on the same
		// st->iocp -- so this needs to be atomic, with a compare-exchange ensuring only one of
		// the racing threads actually performs a given interval's scan.
		std::atomic<time_t> lastSweepAttempt{0};
		};

		static std::mutex g_stateMutex;
		@@ -121,6 +131,61 @@ namespace netplus {
		return true;
		}

		// Test-only escape hatch, mirrors epoll.cpp/kqueue.cpp's identical hook (same env var) --
		// a real deployment never sets this, so this returns kIdleSweepIntervalSeconds (30)
		// unconditionally. Exists purely so an idle-reaper regression test isn't forced to block
		// for a real 30s per run just to observe one sweep.
		static int sweepIntervalSeconds() {
		static const int v = []() -> int {
		constexpr int kIdleSweepIntervalSeconds = 30;
		if (const char* env = std::getenv("NETPLUS_IDLE_SWEEP_INTERVAL_SECONDS")) {
		int parsed = std::atoi(env);
		if (parsed > 0) return parsed;
		}
		return kIdleSweepIntervalSeconds;
		}();
		return v;
		}

		// Idle-connection reaper -- see event::event()'s idleTimeoutSeconds doc comment
		// (eventapi.h) for why this exists at all, and epoll.cpp's identical-in-spirit method for
		// the epoll/kqueue side of this same feature.
		//
		// Called from the GetQueuedCompletionStatus timeout branch of every worker thread's loop
		// (EventWorker::operator(), below). Unlike epoll/kqueue -- where each worker thread owns
		// its own poll object and CONNECTIONS is filtered by ownerPollFd -- every worker thread
		// here shares the same EventState (and its one sockToCon map) for this event instance, so
		// the throttle must be a genuine cross-thread gate (EventState::lastSweepAttempt, atomic
		// CAS) rather than a plain per-instance member: without it, all `st->threads` workers
		// would redundantly re-scan the same connection set on every ~1s GetQueuedCompletionStatus
		// timeout instead of once per interval.
		//
		// try_cleanup_con() already declines to close a connection with a write in flight
		// (WritePending) -- it just marks it Closing and lets the write-completion handler finish
		// the teardown once that write lands -- so no separate pending-op check is needed here.
		static void reapIdleConnections(event* ev, EventState* st, int idleTimeoutSeconds, int tid) {
		if (idleTimeoutSeconds <= 0) return;

		time_t now = time(nullptr);
		time_t last = st->lastSweepAttempt.load(std::memory_order_relaxed);
		if (now - last < sweepIntervalSeconds()) return;
		if (!st->lastSweepAttempt.compare_exchange_strong(last, now, std::memory_order_relaxed)) return;

		std::vector<SOCKET> idleSocks;
		{
		std::shared_lock<std::shared_mutex> lk(st->conMutex);
		for (auto& kv : st->sockToCon) {
		if (now - kv.second->lasteventime < idleTimeoutSeconds) continue;
		idleSocks.push_back(kv.first);
		}
		}
		for (SOCKET s : idleSocks) {
		auto c = find_con(st, s);
		if (!c) continue; // already torn down by something else in the meantime
		try_cleanup_con(ev, st, c.get(), s, tid);
		}
		}

		// -------------------------------------------------------------------------
		// Ensure persistent recv buffer exists for a connection
		// -------------------------------------------------------------------------
		@@ -360,7 +425,12 @@ namespace netplus {

		BOOL ok = GetQueuedCompletionStatus(st->iocp, &bytes, &key, &pov, 1000);
		if (!event::Running) break;
		if (!ok && !pov) continue; // timeout or spurious wake
		if (!ok && !pov) {
		// ev->_IdleTimeout: EventWorker is a declared friend of event
		// (eventapi.h), same as the epoll/kqueue backends.
		reapIdleConnections(ev, st, ev->_IdleTimeout, tid);
		continue; // timeout or spurious wake
		}
		if (!pov && key == 0) continue;

		socket* sockObj = reinterpret_cast<socket*>(key);
		@@ -500,6 +570,7 @@ namespace netplus {

		if (wouldBlock \|\| decLen == 0) break;

		owner->lasteventime = time(nullptr);
		owner->RecvData.append(decrypted.data.buf, decLen);
		ev->RequestEvent(*owner, tid, (ULONG_PTR)decLen);

		@@ -611,6 +682,7 @@ namespace netplus {
		{
		std::lock_guard<std::recursive_mutex> lock(owner->event_mutex);

		owner->lasteventime = time(nullptr);
		owner->slots[0].WritePending.store(false);
		owner->slots[0].csock->setPendingWrite(false);

		@@ -686,8 +758,8 @@ namespace netplus {
		// -------------------------------------------------------------------------
		// event implementation
		// -------------------------------------------------------------------------
		event::event(std::vector<socket*> serversockets, int timeout)
		: _ServerSockets(serversockets), _Timeout(timeout)
		event::event(std::vector<socket*> serversockets, int timeout, int idleTimeoutSeconds)
		: _ServerSockets(serversockets), _Timeout(timeout), _IdleTimeout(idleTimeoutSeconds)
		{
		if (_ServerSockets.empty()) {
		NetException e;
		@@ -713,7 +785,8 @@ namespace netplus {
		threads = (int)std::max<unsigned>(1, std::thread::hardware_concurrency());
		}

		event::event(const event& src) : _ServerSockets(src._ServerSockets), _Timeout(src._Timeout) {}
		event::event(const event& src)
		: _ServerSockets(src._ServerSockets), _Timeout(src._Timeout), _IdleTimeout(src._IdleTimeout) {}

		event::~event() {
		Running = false;

src/event/poll.cpp

+8 −3

Original line number	Diff line number	Diff line
		@@ -757,8 +757,12 @@ namespace netplus {
		// ------------------------------------------------------------
		// event class
		// ------------------------------------------------------------
		event::event(std::vector<socket*> serversockets, int timeout)
		: _ServerSockets(serversockets), _Timeout(timeout)
		// idleTimeoutSeconds is accepted (for signature parity with the epoll/kqueue/iocp
		// backends -- see eventapi.h's doc comment) but not yet acted on: this select-based
		// fallback backend has no idle-connection reaper, same documented-no-op precedent as
		// requestWritablePoll()/detachConnection() elsewhere in connection.h for this backend.
		event::event(std::vector<socket*> serversockets, int timeout, int idleTimeoutSeconds)
		: _ServerSockets(serversockets), _Timeout(timeout), _IdleTimeout(idleTimeoutSeconds)
		{
		if (_ServerSockets.empty()) {
		NetException e;
		@@ -793,7 +797,8 @@ namespace netplus {
		#endif
		}

		event::event(const event& src) : _ServerSockets(src._ServerSockets), _Timeout(src._Timeout) {}
		event::event(const event& src)
		: _ServerSockets(src._ServerSockets), _Timeout(src._Timeout), _IdleTimeout(src._IdleTimeout) {}

		event::~event() {
		#ifdef _WIN32