test

#include <iostream>

#include <algorithm>

#include <chrono>

#include <ctime>

#include <memory>

#include <vector>

#include <atomic>

                                // True EOF — peer closed the connection

                                needClose = true;

                            } else if (rcv > 0) {

                                c->lasteventime = time(nullptr);

                                c->RecvData.append(recvBuf.data.buf, rcv);

                                evconnection->RequestEvent(*c, tid, args);

                                if (!c->slots[0].csock) {

                                    // flush_out below throws EAGAIN we must not

                                    // re-encrypt this chunk on the next cycle.

                                    if (consumed > 0) {

                                        c->lasteventime = time(nullptr);

                                        c->SendData.pos += consumed;

                                        if (c->SendData.pos >= c->SendData.size()) {

                                            c->SendData.clear();

            }

        }

        // ------------------------------------------------------------

        // Idle-connection reaper — see event::event()'s idleTimeoutSeconds doc comment

        // (eventapi.h) for why this exists at all.

        //

        // Called from the wait==0 branch of every worker thread's loop (EventWorker, below),

        // so with `threads` CPU-count worker threads sharing this listener's epoll fd, up to

        // `threads` poll instances -- one per thread -- can all decide to sweep around the same

        // moment. That's fine, not a race: _lastSweepAttempt is instance-local (each thread's

        // poll object is its own, never shared, so no synchronization is needed on it), so

        // worst case a few of them redundantly re-scan the same connection set once in a while;

        // CloseEventHandler's find-then-erase-under-lock (above) already makes a second close

        // attempt on an already-reaped fd a safe no-op. Deliberately NOT a global/shared gate

        // across listeners -- that would starve whichever listener's poll instance doesn't win

        // the race every interval.

        // ------------------------------------------------------------

        void reapIdleConnections(int idleTimeoutSeconds, int tid, ULONG_PTR args) {

            if (idleTimeoutSeconds <= 0) return;

            time_t now = time(nullptr);

            if (now - _lastSweepAttempt < kIdleSweepIntervalSeconds) return;

            _lastSweepAttempt = now;

            std::vector<int> idleFds;

            {

                std::shared_lock<std::shared_mutex> lk(POLL_HANDLER_MUTEX);

                for (auto& kv : CONNECTIONS) {

                    if (kv.second->ownerPollFd != _pollFD) continue; // not ours to close

                    if (now - kv.second->lasteventime < idleTimeoutSeconds) continue;

                    idleFds.push_back(kv.first);

                }

            }

            // Close outside POLL_HANDLER_MUTEX -- CloseEventHandler takes it again itself

            // (unique_lock), and it's not recursive.

            for (int fd : idleFds) CloseEventHandler(fd, tid, args);

        }

    private:

        int _pollFD;

        std::unique_ptr<epoll_event[]> _Events;

        socket* _ServerSocket;

        time_t _lastSweepAttempt = 0;

        static constexpr int kIdleSweepIntervalSeconds = 30;

    };

    // ------------------------------------------------------------

    class EventWorkerArgs {

    public:

        int timeout;

        int idleTimeout = 0;

        eventapi* event;

        std::map<int,socket*> ssocket;

    };

                                if (wait == 0) {

                                    // epoll_wait already blocked for the timeout

                                    // duration — no need for additional sleep.

                                    pollptr.reapIdleConnections(eargs->idleTimeout, tid, args);

                                    continue;

                                }

    // ------------------------------------------------------------

    // event class

    // ------------------------------------------------------------

    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;

        threads = sysconf(_SC_NPROCESSORS_ONLN);

    }

    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() {}

    void event::runEventloop(ULONG_PTR args) {

        }

        eargs.timeout = _Timeout;

        eargs.idleTimeout = _IdleTimeout;

        eargs.event = this;

        pthread_attr_t attr;

#include <iostream>

#include <algorithm>

#include <chrono>

#include <ctime>

#include <memory>

#include <vector>

#include <atomic>

                                // True EOF — peer closed the connection

                                needClose = true;

                            } else if (rcv > 0) {

                                c->lasteventime = time(nullptr);

                                c->RecvData.append(recvBuf.data.buf, rcv);

                                evconnection->RequestEvent(*c, tid, args);

                                if (!c->slots[0].csock) {

                                    // flush_out below throws EAGAIN we must not

                                    // re-encrypt this chunk on the next cycle.

                                    if (consumed > 0) {

                                        c->lasteventime = time(nullptr);

                                        c->SendData.pos += consumed;

                                        if (c->SendData.pos >= c->SendData.size()) {

                                            c->SendData.clear();

            }

        }

        // ------------------------------------------------------------

        // Idle-connection reaper — see event::event()'s idleTimeoutSeconds doc comment

        // (eventapi.h) for why this exists, and epoll.cpp's identical method for the full

        // reasoning on why per-instance (not shared/atomic) _lastSweepAttempt is safe here

        // despite multiple poll instances (one per worker thread) sharing this listener's kqfd.

        // ------------------------------------------------------------

        void reapIdleConnections(int idleTimeoutSeconds, int tid, ULONG_PTR args) {

            if (idleTimeoutSeconds <= 0) return;

            time_t now = time(nullptr);

            if (now - _lastSweepAttempt < kIdleSweepIntervalSeconds) return;

            _lastSweepAttempt = now;

            std::vector<int> idleFds;

            {

                std::shared_lock<std::shared_mutex> lk(POLL_HANDLER_MUTEX);

                for (auto& kv : CONNECTIONS) {

                    if (kv.second->ownerPollFd != _kqfd) continue; // not ours to close

                    if (now - kv.second->lasteventime < idleTimeoutSeconds) continue;

                    idleFds.push_back(kv.first);

                }

            }

            // Close outside POLL_HANDLER_MUTEX -- CloseEventHandler takes it again itself

            // (unique_lock), and it's not recursive.

            for (int fd : idleFds) CloseEventHandler(fd, tid, args);

        }

    private:

        int _kqfd;

        std::unique_ptr<struct kevent[]> _Events;

        socket* _ServerSocket;

        int _EventCount;

        KeventBatch _batch;

        time_t _lastSweepAttempt = 0;

        static constexpr int kIdleSweepIntervalSeconds = 30;

    };

    // ------------------------------------------------------------

    class EventWorkerArgs {

    public:

        int timeout;

        int idleTimeout = 0;

        eventapi* event;

        std::map<int,socket*> ssocket;

    };

                                    // kevent() already blocked for the full

                                    // timeout duration — no additional sleep

                                    // needed (matches epoll behavior).

                                    pollptr.reapIdleConnections(eargs->idleTimeout, tid, args);

                                    continue;

                                }

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

// event class implementation (outside namespace)

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

netplus::event::event(std::vector<netplus::socket*> serversockets, int timeout)

    : _ServerSockets(serversockets), _Timeout(timeout)

netplus::event::event(std::vector<netplus::socket*> serversockets, int timeout, int idleTimeoutSeconds)

    : _ServerSockets(serversockets), _Timeout(timeout), _IdleTimeout(idleTimeoutSeconds)

{

    if (_ServerSockets.empty()) {

        NetException e;

    threads = sysconf(_SC_NPROCESSORS_ONLN);

}

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

netplus::event::event(const netplus::event& src)

    : _ServerSockets(src._ServerSockets), _Timeout(src._Timeout), _IdleTimeout(src._IdleTimeout) {}

netplus::event::~event() {}

void netplus::event::runEventloop(ULONG_PTR args) {

    }

    eargs.timeout = _Timeout;

    eargs.idleTimeout = _IdleTimeout;

    eargs.event = this;

    pthread_attr_t attr;

        class event :public eventapi {

        public:

            event(std::vector<netplus::socket*> serversockets, int timeout=1000);

            // idleTimeoutSeconds: close an accepted connection once it's gone this long with

            // no genuine read/write activity (see con::lasteventime) -- e.g. a browser's

            // keep-alive connection outliving its tab, a mobile app backgrounded without a

            // clean close, or a NAT/middlebox silently dropping a connection's mapping without

            // ever sending a FIN/RST. Without this, such a connection's fd (and, on the epoll/

            // kqueue backends, its entry in the process-wide CONNECTIONS registry) is held for

            // the life of the process -- there was previously no time-based reclamation

            // anywhere in this stack, only reactive close-on-peer-error/EOF. 0 = disabled

            // (default): matches every accepted connection's previous behavior exactly, so

            // existing deployments are unaffected until they opt in. Implemented on the epoll

            // and kqueue backends only, same documented-no-op precedent as

            // requestWritablePoll()/detachConnection() for poll.cpp/iocp.cpp (connection.h).

            event(std::vector<netplus::socket*> serversockets, int timeout=1000,

                  int idleTimeoutSeconds=0);

            void runEventloop(ULONG_PTR args=0);

        private:

            std::vector<netplus::socket*> _ServerSockets;

            int           _Timeout;

            int           _IdleTimeout;

            int           _pollFD;

            friend class EventWorker;

        };

endif()

add_test(NAME tls_roundtrip_sha256_test COMMAND tls_roundtrip_sha256_test)

add_executable(tls_flush_out_blocking_test tls_flush_out_blocking_test.cpp)

if(WIN32)

    target_link_libraries(tls_flush_out_blocking_test netplus-static ws2_32)

else()

    target_link_libraries(tls_flush_out_blocking_test netplus-static)

endif()

add_test(NAME tls_flush_out_blocking_test COMMAND tls_flush_out_blocking_test)

add_executable(rwlock_writer_starvation_test rwlock_writer_starvation_test.cpp)

if(WIN32)

    target_link_libraries(rwlock_writer_starvation_test netplus-static ws2_32)

// tls_flush_out_blocking_test.cpp

//

// Regression test for tls::flush_out(): when the underlying transport socket

// is a BLOCKING socket and its sendRaw() reports NetException::Note (i.e.

// EAGAIN/EWOULDBLOCK), that can only mean a configured SO_SNDTIMEO deadline

// just expired -- a genuinely blocking send() with no timeout configured

// cannot return EAGAIN at all. Before the fix, flush_out() treated this

// exactly like the normal non-blocking "not writable yet, event loop will

// retry" case and silently returned as if the write had succeeded, leaving

// encrypted bytes stuck in send_queue/send_record forever while every caller

// up the stack (including tls::sendData()) believed the send had completed.

//

// The fix added netplus::socket::isBlocking() (backed by a tracked

// _blocking member kept in sync by setBlock()/setNonBlock()) so

// flush_out()'s catch(NetException::Note) branch can tell the two cases

// apart: when isBlocking() is true it now throws a hard NetException::Error

// instead of returning.

//

// This is a white-box unit test rather than a full TLS handshake driven to

// real kernel send-buffer exhaustion: reliably filling a real socket's send

// buffer to force a genuine SO_SNDTIMEO expiry is slow and environment

// -dependent (kernel buffer sizes vary), so it would make this test flaky.

// Instead we drive tls::flush_out() directly against a minimal fake

// transport socket (a netplus::tcp subclass with no real fd) whose

// sendRaw() always reports NetException::Note and whose isBlocking() is

// controlled directly -- this exercises the exact fixed branch

// deterministically. tls::queueRaw()/flush_out()/hasPendingWrite() are all

// public members of netplus::tls (see src/crypto/tls.h), so no friend/test

// -helper access is required.

//

// Two cases are covered, both sides of the branch:

//   1. isBlocking() == true  -> flush_out() must now throw NetException

//      with getErrorType() == NetException::Error (the fix).

//   2. isBlocking() == false -> flush_out() must still swallow the Note

//      and return normally, leaving the record queued for the event loop

//      to retry later (the original, still-correct non-blocking behavior).

#include <iostream>

#include <string>

#include <vector>

#include <cstdint>

#include "socket.h"

#include "exception.h"

using namespace netplus;

static int g_passed = 0, g_failed = 0;

static void check(bool ok, const std::string& name) {

    if (ok) { std::cout << "  PASS: " << name << std::endl; g_passed++; }

    else    { std::cout << "  FAIL: " << name << std::endl; g_failed++; }

}

namespace {

// Minimal transport double. Deliberately extends netplus::tcp (rather than

// netplus::socket directly) so all the other pure-virtual plumbing (bind,

// listen, connect, accept, getAddress, operator=(SOCKET), ...) is already

// implemented by the real tcp class and never gets exercised here. The

// default tcp() constructor sets _Socket = -1 and never opens a real fd,

// and ~tcp() only closes if _Socket >= 0, so this never touches the

// network at all -- sendRaw() is overridden to always simulate

// EAGAIN/EWOULDBLOCK before any real syscall would happen.

class FakeAlwaysWouldBlockSocket : public netplus::tcp {

public:

    using netplus::tcp::operator=;

    FakeAlwaysWouldBlockSocket() : netplus::tcp() {}

    size_t sendRaw(netplus::buffer& data, int flags = 0) override {

        (void)data; (void)flags;

        ++sendRawCalls;

        NetException n;

        n[NetException::Note] << "FakeAlwaysWouldBlockSocket::sendRaw: simulated EAGAIN/EWOULDBLOCK";

        throw n;

    }

    // Directly set the same _blocking flag that real setBlock()/

    // setNonBlock() overrides keep in sync (see src/posix/tcp.cpp) --

    // without needing a real fd to call fcntl() on. _blocking is

    // `protected` on netplus::socket, so it's reachable here through the

    // tcp -> socket inheritance chain.

    void setBlockingForTest(bool b) { _blocking = b; }

    int sendRawCalls = 0;

};

} // namespace

int main() {

    std::cout << "=== tls::flush_out() blocking-socket timeout regression test ===" << std::endl;

    const std::vector<uint8_t> payload = { 'h','e','l','l','o',' ','w','o','r','l','d' };

    // ------------------------------------------------------------------

    // Case 1 (the fix): BLOCKING socket + Note from sendRaw() must now

    // surface as a hard NetException::Error, not a silent success.

    // ------------------------------------------------------------------

    {

        std::cout << "\n--- Case 1: blocking socket, sendRaw() reports EAGAIN ---" << std::endl;

        netplus::tls t;

        FakeAlwaysWouldBlockSocket sock;

        sock.setBlockingForTest(true);

        check(sock.isBlocking(), "fake socket reports isBlocking() == true");

        t.setSocket(&sock);

        t.queueRaw(payload.data(), payload.size());

        check(t.hasPendingWrite(), "data is queued before flush_out() is called");

        bool threw = false;

        bool correctType = false;

        std::string what;

        try {

            t.flush_out();

        } catch (NetException& e) {

            threw = true;

            correctType = (e.getErrorType() == NetException::Error);

            what = e.what();

        } catch (...) {

            threw = true;

        }

        check(threw,

              "flush_out() throws when a BLOCKING socket reports EAGAIN "

              "(pre-fix: this was silently swallowed and flush_out() returned as if it succeeded)");

        check(correctType, "the thrown exception is NetException::Error, not Note/Warning/other");

        if (!what.empty())

            std::cout << "  (exception message: " << what << ")" << std::endl;

        check(sock.sendRawCalls >= 1, "sendRaw() was actually invoked -- the branch under test really ran");

        check(t.hasPendingWrite(),

              "queued bytes are NOT silently dropped -- send_queue/send_record still holds them after the throw");

    }

    // ------------------------------------------------------------------

    // Case 2 (unchanged behavior): NON-blocking socket + Note from

    // sendRaw() must still return normally (record stays queued for the

    // event loop to retry once the socket becomes writable), exactly as

    // before the fix.

    // ------------------------------------------------------------------

    {

        std::cout << "\n--- Case 2: non-blocking socket, sendRaw() reports EAGAIN ---" << std::endl;

        netplus::tls t;

        FakeAlwaysWouldBlockSocket sock;

        sock.setBlockingForTest(false);

        check(!sock.isBlocking(), "fake socket reports isBlocking() == false");

        t.setSocket(&sock);

        t.queueRaw(payload.data(), payload.size());

        check(t.hasPendingWrite(), "data is queued before flush_out() is called");

        bool threw = false;

        try {

            t.flush_out();

        } catch (NetException&) {

            threw = true;

        }

        check(!threw,

              "flush_out() does NOT throw for a non-blocking socket reporting EAGAIN "

              "(the event loop is expected to retry later)");

        check(sock.sendRawCalls >= 1, "sendRaw() was actually invoked -- the branch under test really ran");

        check(t.hasPendingWrite(), "record remains queued for the next flush_out() attempt, unchanged from before the fix");

    }

    std::cout << "\n==============================" << std::endl;

    std::cout << "Results: " << g_passed << " passed, " << g_failed << " failed" << std::endl;

    std::cout << "==============================" << std::endl;

    return (g_failed > 0) ? 1 : 0;

}