test

                    ccon->slots[0].csock = std::make_unique<ssl>(*srv->_tls.cert_map, -1);

                } else if (_ServerSocket->_Type == sockettype::QUIC) {

                    // QUIC: accept() drains up to 512 queued datagrams in

                    // one call (see quic::accept). The listener fd is level

                    // triggered without EPOLLONESHOT (see runEventloop), so

                    // no rearm is needed here — and no rearm-with-ONESHOT

                    // should happen here either, since that would silently

                    // put the fd back into single-thread-at-a-time mode

                    // after the first connection event.

                    // one call (see quic::accept). The listener fd is

                    // registered EPOLLONESHOT (see runEventloop) so we're

                    // the only thread servicing it right now — rearm on

                    // scope exit (even if accept() throws) so the next

                    // datagram wakes a thread again.

                    int qfd = _ServerSocket->fd();

                    struct RearmOnExit {

                        poll* p; int fd;

                        ~RearmOnExit() {

                            try { p->setpollEventsFd(fd, EPOLLIN | EPOLLONESHOT); }

                            catch (...) {}

                        }

                    } rearm{this, qfd};

                    std::unique_ptr<socket> quic_csock;

                    _ServerSocket->accept(quic_csock, true);

                    return;

            epoll_event setevent{};

            setevent.events = EPOLLIN;

            // QUIC multiplexes every client (and all of their ongoing

            // traffic) over this one UDP socket for the connection's whole

            // lifetime — unlike TCP/SSL, where only accept() shares the

            // listener and established connections get their own

            // EPOLLONESHOT'd fd. This used to also be EPOLLONESHOT'd for

            // QUIC, because quic_mtx() was one lock shared by every

            // connection on the listener (see socket.h's history) — level

            // triggered + shared epoll fd meant every datagram woke all

            // `threads` workers, and only one could ever do anything

            // useful while the rest contended on that single mutex for

            // nothing.

            // traffic) over this one UDP socket for the connection's

            // whole lifetime — unlike TCP/SSL, where only accept() shares

            // the listener and established connections get their own

            // EPOLLONESHOT'd fd. Without ONESHOT here, every incoming

            // datagram wakes all `threads` workers (level-triggered,

            // shared epoll fd); ONESHOT + explicit rearm in

            // ConnectEventHandler limits servicing to one thread at a

            // time, same as TCP/SSL already do per-connection.

            //

            // Each connection now has its own quic_mtx(), and

            // _child_connections lookups/mutations go through their own

            // short-lived _registry_mutex (see quic::accept()), so

            // different connections CAN be processed by different threads

            // in parallel instead of serializing on one lock. But plain

            // EPOLLIN without ONESHOT still wakes every worker per event

            // (thundering herd): measured under concurrent load (24

            // simultaneous new connections, see quic_concurrent_test) this

            // made things drastically worse, not better — `threads`

            // workers (one per core) all piling into accept()/recvmmsg for

            // the same handful of datagrams starved the CPU time actual

            // packet processing needed, cascading into handshake timeouts.

            // EPOLLEXCLUSIVE (Linux >= 4.5) is the fix purpose-built for

            // this: only one waiter is woken per readiness event instead

            // of all of them, but — unlike EPOLLONESHOT — no explicit

            // rearm is needed and a different thread is free to pick up

            // the next event, so cross-connection parallelism is kept

            // without the herd. (Not recommended to combine with

            // EPOLLONESHOT, which is why this is the only flag added.)

            // Scoped to QUIC only — TCP/SSL listeners weren't part of what

            // this was measured against, so left as before.

            // This was changed to EPOLLEXCLUSIVE for a while, on the

            // theory that per-connection locking (each quic object now

            // has its own quic_mtx(), see socket.h) meant concurrent

            // accept() calls could process different connections in

            // parallel instead of serializing on one shared lock. Measured

            // and reverted: even for a *single* connection, multiple

            // threads independently draining the same UDP socket

            // (recvmmsg on the same fd from separate threads is safe, but

            // not free) split what would otherwise be one large,

            // uncontended batch into several smaller ones handled by

            // different threads, which then contend on that one

            // connection's mutex instead of never touching it concurrently

            // at all — a 10x throughput regression on the basic

            // single-connection benchmark, confirmed by forcing

            // single-threaded servicing (`event::threads = 1`) and seeing

            // full throughput return. The per-connection locking itself

            // stays (it's correct, and free when only one thread is ever

            // in accept() at a time) as groundwork for a real fix later —

            // e.g. sharding connections to dedicated threads instead of

            // letting any thread grab any datagram — but that needs its

            // own design and measurement, not a blanket EPOLLEXCLUSIVE.

            if (ss->_Type == sockettype::QUIC) {

                setevent.events |= EPOLLEXCLUSIVE;

                setevent.events |= EPOLLONESHOT;

            }

            setevent.data.fd = ss->fd();

#include "eventapi.h"

#include "exception.h"

#include "random.h"

#include "threadpool.h"

#include "crypto/aes.h"

#include "crypto/sha.h"

#include "crypto/curve25519.h"

inline std::string cidKey(const std::vector<uint8_t>& v) {

    return cidKey(v.data(), v.size());

}

// Shared pool for running stream-callback dispatches off the epoll worker

// threads (see quic::scheduleDispatches). Process-wide and lazily

// constructed on first use (C++11 guarantees thread-safe init of function-

// local statics); sized generously since this work is dominated by waiting

// on network round trips, not CPU, so oversubscribing relative to core

// count is fine — the cost of a mostly-idle extra thread is far smaller

// than the cost of a slow callback blocking one of the few epoll workers.

ThreadPool& quicDispatchPool() {

    static ThreadPool pool((std::max<unsigned>)(4, std::thread::hardware_concurrency() * 2));

    return pool;

}

} // namespace

// ============================================================================

    }

}

// ============================================================================

// Off-thread stream-callback dispatch (see declaration comment in socket.h)

// ============================================================================

void quic::scheduleDispatches(std::shared_ptr<quic> self,

                               std::vector<PendingDispatch> dispatches) {

    if (dispatches.empty() || !self->_stream_callback) return;

    if (!self->_dispatch_promoted.load()) {

        // Fast path: run inline on the calling thread, exactly like a

        // direct call — this is the common case (small/fast callbacks)

        // and must not pay any queuing or thread-hop cost. Timed so a

        // callback that turns out to be slow promotes this connection to

        // the async path for its *next* dispatch — this one has already

        // paid the cost of blocking inline, nothing to be done for it

        // retroactively.

        auto t0 = std::chrono::steady_clock::now();

        for (auto& pd : dispatches) {

            self->_stream_callback(self.get(), pd.stream_id, pd.data, pd.fin);

        }

        if (std::chrono::steady_clock::now() - t0 > kSlowDispatchThreshold) {

            self->_dispatch_promoted.store(true);

        }

        return;

    }

    bool need_to_start;

    {

        std::lock_guard<std::mutex> lk(self->_dispatch_queue_mutex);

        self->_dispatch_queue.push_back(std::move(dispatches));

        // exchange(true) returns the previous value: if it was already

        // true, a runner is already draining _dispatch_queue and will pick

        // up this batch itself — starting a second one would let two

        // threads race on the same connection's callbacks.

        need_to_start = !self->_dispatch_running.exchange(true);

    }

    if (need_to_start) {

        quicDispatchPool().submit([self]() {

            self->runDispatchQueue(self);

        });

    }

}

void quic::runDispatchQueue(std::shared_ptr<quic> self) {

    for (;;) {

        std::vector<PendingDispatch> batch;

        {

            std::lock_guard<std::mutex> lk(_dispatch_queue_mutex);

            if (_dispatch_queue.empty()) {

                _dispatch_running.store(false);

                return;

            }

            batch = std::move(_dispatch_queue.front());

            _dispatch_queue.pop_front();

        }

        for (auto& pd : batch) {

            _stream_callback(this, pd.stream_id, pd.data, pd.fin);

        }

    }

    // `self` (kept alive across the whole loop, including while the queue

    // was briefly empty between batches being enqueued and us checking)

    // goes out of scope here, dropping this function's reference.

}

// ============================================================================

// Socket Interface: accept (Server-side connection acceptance)

// ============================================================================

                }

            }

            if (!existing->_pending_dispatches.empty() && existing->_stream_callback) {

            if (!existing->_pending_dispatches.empty()) {

                std::vector<PendingDispatch> dispatches = std::move(existing->_pending_dispatches);

                existing->_pending_dispatches.clear();

                child_lock.unlock();

                for (auto& pd : dispatches) {

                    existing->_stream_callback(existing.get(), pd.stream_id, pd.data, pd.fin);

                }

                quic::scheduleDispatches(existing, std::move(dispatches));

                child_lock.lock();

            }

        }

                }

            }

            // Dispatch stream callbacks outside the connection's own lock

            if (!existing->_pending_dispatches.empty() && existing->_stream_callback) {

            // Dispatch stream callbacks off-thread (see scheduleDispatches)

            if (!existing->_pending_dispatches.empty()) {

                std::vector<PendingDispatch> dispatches = std::move(existing->_pending_dispatches);

                existing->_pending_dispatches.clear();

                child_lock.unlock();

                for (auto& pd : dispatches) {

                    existing->_stream_callback(existing.get(), pd.stream_id, pd.data, pd.fin);

                }

                quic::scheduleDispatches(existing, std::move(dispatches));

                child_lock.lock();

            }

            continue;

            }

        }

        if (!winner->_pending_dispatches.empty() && winner->_stream_callback) {

        if (!winner->_pending_dispatches.empty()) {

            std::vector<PendingDispatch> dispatches = std::move(winner->_pending_dispatches);

            winner->_pending_dispatches.clear();

            winner_lock.unlock();

            for (auto& pd : dispatches) {

                winner->_stream_callback(winner.get(), pd.stream_id, pd.data, pd.fin);

            }

            quic::scheduleDispatches(winner, std::move(dispatches));

            winner_lock.lock();

        }

        continue;

        }

    }

    // Dispatch any pending child callbacks outside its own lock

    {

        std::vector<PendingDispatch> child_dispatches;

        child_dispatches = std::move(child_ptr->_pending_dispatches);

    // Dispatch any pending child callbacks off-thread (see scheduleDispatches)

    if (!child_ptr->_pending_dispatches.empty()) {

        std::vector<PendingDispatch> child_dispatches = std::move(child_ptr->_pending_dispatches);

        child_ptr->_pending_dispatches.clear();

        if (!child_dispatches.empty() && child_ptr->_stream_callback) {

        child_lock.unlock();

            for (auto& pd : child_dispatches) {

                child_ptr->_stream_callback(child_ptr.get(), pd.stream_id, pd.data, pd.fin);

            }

        quic::scheduleDispatches(child_ptr, std::move(child_dispatches));

        child_lock.lock();

    }

    }

    // Child connection is owned jointly by its two _child_connections

    // registry entries (shared_ptr) — don't also transfer ownership to

		};

		std::vector<PendingDispatch> _pending_dispatches;

		// Dispatches _stream_callback batches — inline on the calling

		// thread by default (same cost as calling the callback directly:

		// this is the common case, e.g. an echo server's fast reply, and

		// must not pay any queuing/thread-hop overhead for it), but

		// permanently promotes a connection to running its callbacks on a

		// shared thread pool instead, once one of its dispatches is

		// observed taking longer than kSlowDispatchThreshold. A callback

		// that takes a while (e.g. sending a large reply, which needs

		// several congestion-window-limited round trips) would otherwise

		// tie up whichever epoll worker thread is servicing accept() for

		// its whole duration, and enough concurrent slow callbacks can

		// exhaust the pool and starve unrelated connections'

		// handshakes/ACKs — but that's only worth paying the async-path

		// cost for once a connection has actually shown that behavior.

		// Never auto-demotes back to inline: a connection that needed it

		// once (e.g. mid-transfer) is likely to need it again for the

		// rest of that transfer, and flapping between paths would add

		// its own overhead.

		//

		// Per-connection FIFO ordering on the async path (a stream's

		// chunks must still arrive in order, with fin=true only on the

		// last one) is preserved by only ever having one runner active

		// per connection at a time: scheduleDispatches() enqueues a batch

		// and, if no runner is currently active for this connection,

		// submits one; that runner drains _dispatch_queue until empty

		// before clearing _dispatch_running, so a batch enqueued while a

		// runner is already active is picked up by that same runner

		// rather than racing a second one on another thread.

		//

		// Takes the shared_ptr as a parameter (rather than using

		// enable_shared_from_this) because callers already hold one — this

		// is only ever reached for child connections, always constructed

		// via make_shared (see quic::accept()); the listener and client

		// connections never call it.

		static constexpr std::chrono::microseconds kSlowDispatchThreshold{2000};

		static void scheduleDispatches(std::shared_ptr<quic> self,

		                                std::vector<PendingDispatch> dispatches);

		void runDispatchQueue(std::shared_ptr<quic> self);

		std::atomic<bool> _dispatch_promoted{false};

		std::mutex _dispatch_queue_mutex;

		std::deque<std::vector<PendingDispatch>> _dispatch_queue;

		std::atomic<bool> _dispatch_running{false};

		// Handshake timeout tracking

		// Detects when client sends incomplete ClientHello with large gaps (likely packet loss)

		std::chrono::steady_clock::time_point _handshake_start_time;

/*******************************************************************************

* Copyright (c) 2026, Jan Koester jan.koester@gmx.net

* All rights reserved.

*

* Redistribution and use in source and binary forms, with or without

* modification, are permitted provided that the following conditions are met:

* Redistributions of source code must retain the above copyright

*      notice, this list of conditions and the following disclaimer.

* Redistributions in binary form must reproduce the above copyright

*      notice, this list of conditions and the following disclaimer in the

*      documentation and/or other materials provided with the distribution.

* Neither the name of the <organization> nor the

*      names of its contributors may be used to endorse or promote products

*      derived from this software without specific prior written permission.

*

* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND

* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

* WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE

* DISCLAIMED. IN NO EVENT SHALL <COPYRIGHT HOLDER> BE LIABLE FOR ANY

* DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES

* (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;

* LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND

* ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT

* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS

* SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

*******************************************************************************/

#pragma once

#include <vector>

#include <deque>

#include <thread>

#include <functional>

#include <mutex>

#include <condition_variable>

namespace netplus {

// Minimal fixed-size worker pool for offloading tasks that don't belong on

// whatever thread submits them (e.g. a caller-supplied callback that may

// block for a while, run from code that also needs its thread free for

// other work). Not a general scheduler: no priorities, no per-task futures,

// no work-stealing — just a shared FIFO queue drained by a fixed set of

// worker threads. Tasks are run in submission order across the pool, but

// nothing here orders tasks *relative to each other* beyond that — callers

// needing per-key ordering (e.g. "this connection's callbacks must run in

// sequence") need their own serialization on top, since two tasks for the

// same key could otherwise be picked up by two different idle workers at

// once.

class ThreadPool {

public:

    explicit ThreadPool(size_t num_threads) {

        _workers.reserve(num_threads);

        for (size_t i = 0; i < num_threads; ++i) {

            _workers.emplace_back([this] { workerLoop(); });

        }

    }

    ~ThreadPool() {

        {

            std::lock_guard<std::mutex> lk(_mtx);

            _stopping = true;

        }

        _cv.notify_all();

        for (auto& t : _workers) {

            if (t.joinable()) t.join();

        }

    }

    ThreadPool(const ThreadPool&) = delete;

    ThreadPool& operator=(const ThreadPool&) = delete;

    // Queues a task for execution on some worker thread. Safe to call from

    // any thread, including from within a task running on this same pool.

    void submit(std::function<void()> task) {

        {

            std::lock_guard<std::mutex> lk(_mtx);

            _tasks.push_back(std::move(task));

        }

        _cv.notify_one();

    }

private:

    void workerLoop() {

        for (;;) {

            std::function<void()> task;

            {

                std::unique_lock<std::mutex> lk(_mtx);

                _cv.wait(lk, [this] { return _stopping || !_tasks.empty(); });

                if (_tasks.empty()) {

                    // Only reachable via _stopping — drain fully before exiting

                    // (queued work still runs; shutdown just stops accepting

                    // the wait once there's nothing left).

                    return;

                }

                task = std::move(_tasks.front());

                _tasks.pop_front();

            }

            // A task throwing must not crash the whole pool (and, since

            // nothing else on this thread's call stack will catch it,

            // the whole process) — matches the epoll worker loop's own

            // per-event try/catch (src/event/epoll.cpp), which is what

            // used to protect a callback like this before it ran here.

            try {

                task();

            } catch (...) {

            }

        }

    }

    std::vector<std::thread> _workers;

    std::deque<std::function<void()>> _tasks;

    std::mutex _mtx;

    std::condition_variable _cv;

    bool _stopping = false;

};

} // namespace netplus