test (4efef906) · Commits · jan.koester / libnetplus

src/connection.h

+2 −1

Original line number	Diff line number	Diff line
		@@ -176,9 +176,10 @@ namespace netplus {
		int DebugId=0;

		std::mutex event_mutex;

		time_t lasteventime;
		protected:
		con();
		time_t lasteventime;
		friend class poll;
		};
		};

src/event/epoll.cpp

+174 −147

Original line number	Diff line number	Diff line
		@@ -148,7 +148,7 @@ namespace netplus {
		}

		int pollState(int pos){
		if(!_Events[pos].data.ptr)
		if(_Events[pos].data.fd == _ServerSocket->fd())
		return EventHandlerStatus::EVCON;
		return _Events[pos].events;
		}
		@@ -200,7 +200,7 @@ namespace netplus {
		}

		struct epoll_event setevent = {0};
		setevent.events =EPOLLIN \| EPOLLRDHUP \| EPOLLONESHOT \| EPOLLET;
		setevent.events =EPOLLIN \| EPOLLRDHUP \| EPOLLONESHOT;
		setevent.data.fd = fd;

		epoll_ctl(_pollFD, EPOLL_CTL_ADD, fd, &setevent);
		@@ -215,135 +215,168 @@ namespace netplus {
		{
		std::lock_guard<std::mutex> lk(POLL_HANDLER_MUTEX);
		auto it = CONNECTIONS.find(fd);
		if (it != CONNECTIONS.end()) {
		rcon = it->second;
		}
		if (it != CONNECTIONS.end()) rcon = it->second;
		}

		if(!rcon \|\| !rcon->csock)
		return;
		if (!rcon \|\| !rcon->csock) return;

		std::lock_guard<std::mutex> lk(rcon->event_mutex);
		if (!rcon->csock) return;

		try {
		// ------------------------------------------------------------
		// 1) HANDSHAKE PHASE (async)
		// ------------------------------------------------------------
		if (!rcon->csock->getHandshakeDone()) {
		// Run as far as possible (may throw Note)
		rcon->csock->handshake_after_accept();

		// Handshake may have produced outgoing bytes -> flush now
		rcon->csock->flush_out();
		}

		// ------------------------------------------------------------
		// 2) DATA PHASE (after handshake)
		// ------------------------------------------------------------
		if (rcon->csock->getHandshakeDone()) {
		bool dataReceived = false;

		for (;;) {
		buffer buf(BLOCKSIZE);
		size_t rcv = 0;

		try {
		rcv = rcon->csock->recvData(buf, 0);
		} catch (NetException& e) {
		if (e.getErrorType() == NetException::Note) {
		// Falls während des Handshakes Sendedaten (z.B. ServerHello)
		// hängen geblieben sind, müssen wir auf EPOLLOUT warten
		int event = EPOLLIN;
		if (rcon->csock->_Type == sockettype::SSL && !rcon->SendData.empty())
		event \|= EPOLLOUT;

		setpollEvents(pos, event \| EPOLLONESHOT \| EPOLLET \| EPOLLRDHUP);
		return;
		}
		if (e.getErrorType() == NetException::Note) break; // kernel empty
		throw;
		}

		if (rcv > 0) {
		rcon->RecvData.append(buf.data.buf, rcv);
		rcon->lasteventime = time(nullptr);
		evconnection->RequestEvent(*rcon, tid, args);
		// Nach dem RequestEvent brechen wir ab, damit die
		// Antwort (SendData) im nächsten Schritt verarbeitet wird.
		break;
		} else {
		// rcv == 0: SSL Handshake Paket verarbeitet oder Connection Closed
		if (rcon->csock->_Type == sockettype::SSL && !rcon->csock->getHandshakeDone()) {
		dataReceived = true;
		continue;
		}
		throw NetException() << "Client closed connection";
		}

		// rcv == 0 => peer closed
		break;
		}

		// Nach der Schleife (wenn break aufgerufen wurde oder Handshake fertig ist)
		if (!rcon->SendData.empty()) {
		setpollEvents(pos, EPOLLOUT \| EPOLLONESHOT \| EPOLLET \| EPOLLRDHUP);
		} else {
		setpollEvents(pos, EPOLLIN \| EPOLLONESHOT \| EPOLLET \| EPOLLRDHUP);
		if (dataReceived && !rcon->RecvData.empty()) {
		evconnection->RequestEvent(*rcon, tid, args);
		}
		}

		// ------------------------------------------------------------
		// 3) Rearm ONESHOT
		// ------------------------------------------------------------
		int nextevent = EPOLLIN \| EPOLLRDHUP \| EPOLLONESHOT;

		// If app has data to send OR socket has pending write -> need OUT
		if (!rcon->SendData.empty() \|\| rcon->csock->hasPendingWrite())
		nextevent \|= EPOLLOUT;

		setpollEvents(pos, nextevent);

		} catch (NetException& e) {
		if (e.getErrorType() == NetException::Note) {
		// Need more IO. Always arm both IN and OUT.
		setpollEvents(pos, EPOLLIN \| EPOLLOUT \| EPOLLRDHUP \| EPOLLONESHOT);
		return;
		}
		// fatal -> close
		CloseEventHandler(pos, tid, args);
		}
		}


		void WriteEventHandler(int pos, const int tid, ULONG_PTR args) {
		// 1) FD aus epoll-Event holen
		int fd = _Events[pos].data.fd;
		std::shared_ptr<con> wcon;

		// 2) shared_ptr sicher aus der globalen Map holen
		{
		std::lock_guard<std::mutex> lk(POLL_HANDLER_MUTEX);
		auto it = CONNECTIONS.find(fd);
		if (it != CONNECTIONS.end()) {
		wcon = it->second;
		}
		if (it != CONNECTIONS.end()) wcon = it->second;
		}

		// Falls Verbindung bereits weg ist, abbrechen
		if (!wcon \|\| !wcon->csock) return;

		// 3) Verbindungs-Mutex sperren (Thread-Sicherheit für SSL und Buffer)
		std::lock_guard<std::mutex> lk(wcon->event_mutex);
		if (!wcon->csock) return;

		try {
		// 4) Sende-Schleife: Versuche SendData mit Hilfe der buffer struct zu leeren
		// ------------------------------------------------------------
		// 0) flush socket pending bytes first (handles partial send)
		// ------------------------------------------------------------
		wcon->csock->flush_out();

		// ------------------------------------------------------------
		// 1) HANDSHAKE (async)
		// ------------------------------------------------------------
		if (!wcon->csock->getHandshakeDone()) {
		wcon->csock->handshake_after_accept();
		wcon->csock->flush_out(); // handshake may queue more
		}

		// ------------------------------------------------------------
		// 2) APPLICATION SEND (only after handshake)
		// ------------------------------------------------------------
		if (wcon->csock->getHandshakeDone()) {

		while (!wcon->SendData.empty()) {
		// Bestimme Blockgröße (entweder restlicher Buffer oder BLOCKSIZE)
		size_t sendlen = (wcon->SendData.size() > BLOCKSIZE) ? BLOCKSIZE : wcon->SendData.size();
		size_t sendlen = (wcon->SendData.size() > BLOCKSIZE)
		? BLOCKSIZE
		: wcon->SendData.size();

		// Nutze deine buffer struct: Sie kapselt den Pointer und die Länge
		// Annahme: wcon->SendData ist ein std::vector oder eine ähnliche Struktur
		buffer out(wcon->SendData.data(), sendlen);

		size_t consumed = 0;

		try {
		// SSL oder TCP sendData
		consumed = wcon->csock->sendData(out, 0);
		// sendData may queue TLS records internally -> flush them now
		wcon->csock->flush_out();
		} catch (NetException& e) {
		if (e.getErrorType() == NetException::Note) {
		// SSL_WANT_WRITE oder EAGAIN: Wir müssen auf das nächste EPOLLOUT warten
		setpollEvents(pos, EPOLLOUT \| EPOLLONESHOT \| EPOLLET \| EPOLLRDHUP);
		// need OUT again
		int ev = EPOLLIN \| EPOLLOUT \| EPOLLRDHUP \| EPOLLONESHOT;
		setpollEvents(pos, ev);
		return;
		}
		throw; // Kritischer Fehler (z.B. Connection Reset)
		throw;
		}

		if (consumed == 0) {
		// Socket-Buffer des Kernels voll
		setpollEvents(pos, EPOLLOUT \| EPOLLONESHOT \| EPOLLET \| EPOLLRDHUP);
		return;
		}
		if (consumed == 0) break;

		wcon->SendData.erase(
		wcon->SendData.begin(),
		wcon->SendData.begin() + consumed
		);

		// Gesendete Daten aus dem Send-Vektor löschen
		wcon->SendData.erase(wcon->SendData.begin(), wcon->SendData.begin() + consumed);
		wcon->lasteventime = time(nullptr);
		}

		// 5) Wenn alles gesendet wurde: Frage die Protokollschicht nach mehr Daten
		// (Wichtig für Chunked Transfer oder große HTTP-Antworten)
		if (wcon->SendData.empty()) {
		// ------------------------------------------------------------
		// 3) Response event + Rearm
		// ------------------------------------------------------------
		if (wcon->SendData.empty() && !wcon->csock->hasPendingWrite()) {
		evconnection->ResponseEvent(*wcon, tid, args);
		setpollEvents(pos, EPOLLIN \| EPOLLRDHUP \| EPOLLONESHOT);
		} else {
		// still pending send
		setpollEvents(pos, EPOLLIN \| EPOLLOUT \| EPOLLRDHUP \| EPOLLONESHOT);
		}

		// 6) Finaler Zustandswechsel ("New Way")
		if (!wcon->SendData.empty()) {
		// ResponseEvent hat neue Daten hinzugefügt -> Weiterhin im Schreib-Modus bleiben
		setpollEvents(pos, EPOLLOUT \| EPOLLONESHOT \| EPOLLET \| EPOLLRDHUP);
		} else {
		// Alles fertig gesendet -> Zurück in den Lese-Modus (Warten auf nächsten Request)
		setpollEvents(pos, EPOLLIN \| EPOLLONESHOT \| EPOLLET \| EPOLLRDHUP);
		// Handshake not done yet -> keep both
		setpollEvents(pos, EPOLLIN \| EPOLLOUT \| EPOLLRDHUP \| EPOLLONESHOT);
		}

		} catch (NetException& e) {
		// Bei Fehlern (außer "Note") Verbindung schließen
		if (e.getErrorType() == NetException::Note) {
		setpollEvents(pos, EPOLLIN \| EPOLLOUT \| EPOLLRDHUP \| EPOLLONESHOT);
		return;
		}
		CloseEventHandler(pos, tid, args);
		}
		}
		@@ -417,55 +450,48 @@ namespace netplus {
		while (event::Running) {
		try {
		wait = pollptr.waitEventHandler(eargs->timeout);

		for (int i = 0; i < wait; ++i) {
		try {
		// 1. Prüfen, ob es sich um den Server-Socket (Listening) handelt
		// Server socket accept?
		int state = pollptr.pollState(i);
		if (state == pollapi::EventHandlerStatus::EVCON) {
		pollptr.ConnectEventHandler(i, tid, args);
		continue;
		}

		// 2. Verbindung über die Map absichern
		// Wir holen uns den FD aus dem epoll-Event
		int fd = pollptr.getEventFd(i);
		std::shared_ptr<netplus::con> curcon;

		// keep connection alive during handler
		std::shared_ptr<netplus::con> curcon;
		{
		std::lock_guard<std::mutex> lock(POLL_HANDLER_MUTEX);
		auto it = CONNECTIONS.find(fd);
		if (it != CONNECTIONS.end()) {
		curcon = it->second; // Erhöht den Ref-Count, con lebt sicher weiter
		}
		if (it != CONNECTIONS.end()) curcon = it->second;
		}
		if (!curcon) continue;

		// Wenn die Verbindung bereits gelöscht wurde, ignorieren wir das Event
		if (!curcon) {
		continue;
		}

		// 3. Flags abrufen und verarbeiten
		int events = pollptr.getEventFlags(i);

		// Fehler oder Hangup -> Verbindung schließen
		// Hard errors -> close
		if (events & (EPOLLHUP \| EPOLLERR \| EPOLLRDHUP)) {
		pollptr.CloseEventHandler(i, tid, args);
		continue;
		}

		// Daten empfangen (Read)
		if (events & EPOLLIN) {
		pollptr.ReadEventHandler(i, tid, args);
		}
		// Daten senden (Write)
		else if (events & EPOLLOUT) {

		// IMPORTANT: handle both IN and OUT if both are set
		if (events & EPOLLOUT) {
		pollptr.WriteEventHandler(i, tid, args);
		}

		} catch (NetException& e) {
		std::cerr << "EventWorker Thread " << tid << ": " << e.what() << std::endl;
		std::cerr << "EventWorker Thread " << tid
		<< ": " << e.what() << std::endl;

		// Bei Fehlern (außer reinen Notizen) Verbindung trennen
		if (e.getErrorType() != NetException::Note) {
		pollptr.CloseEventHandler(i, tid, args);
		}
		@@ -474,6 +500,7 @@ namespace netplus {
		throw e;
		}
		}

		} catch (NetException& e) {
		std::cerr << "Poll Error: " << e.what() << std::endl;
		}
		@@ -576,7 +603,7 @@ namespace netplus {

		struct epoll_event setevent = {0};

		setevent.events = EPOLLIN \| EPOLLEXCLUSIVE \| EPOLLET;
		setevent.events = EPOLLIN \| EPOLLEXCLUSIVE;
		setevent.data.ptr = nullptr;

		if (epoll_ctl(_pollFD, EPOLL_CTL_ADD,_ServerSocket->fd(),&setevent) < 0) {

src/exception.cpp

+1 −1

Original line number	Diff line number	Diff line
		@@ -36,7 +36,7 @@ namespace netplus {
		};

		netplus::NetException::NetException() {
		curCType=Note;
		curCType=Error;
		};

		netplus::NetException::NetException(const NetException &exp){

src/netdebug.h

0 → 100644

+57 −0

Original line number	Diff line number	Diff line

		#pragma once
		#include <cstdio>
		#include <cerrno>
		#include <cstring>
		#include <chrono>
		#include <thread>

		#ifndef NETPLUS_DEBUG
		#define NETPLUS_DEBUG 0
		#endif

		#if NETPLUS_DEBUG

		// optional: include time stamps
		#ifndef NETPLUS_DEBUG_TS
		#define NETPLUS_DEBUG_TS 1
		#endif

		static inline unsigned long long _ndbg_now_us() {
		using namespace std::chrono;
		return (unsigned long long)duration_cast<microseconds>(
		steady_clock::now().time_since_epoch()
		).count();
		}

		static inline void _ndbg_print_prefix(int tid, int fd, const char* tag) {
		#if NETPLUS_DEBUG_TS
		std::fprintf(stderr, "[%llu][T%02d][fd=%d][%s] ",
		_ndbg_now_us(), tid, fd, tag);
		#else
		std::fprintf(stderr, "[T%02d][fd=%d][%s] ", tid, fd, tag);
		#endif
		}

		#define NDBG(tid, fd, tag, fmt, ...) \
		do { \
		_ndbg_print_prefix((tid), (fd), (tag)); \
		std::fprintf(stderr, fmt "\n", ##__VA_ARGS__); \
		std::fflush(stderr); \
		} while(0)

		#define NDBG_ERR(tid, fd, tag, fmt, ...) \
		do { \
		int _e = errno; \
		_ndbg_print_prefix((tid), (fd), (tag)); \
		std::fprintf(stderr, fmt " (errno=%d:%s)\n", ##__VA_ARGS__, _e, std::strerror(_e)); \
		std::fflush(stderr); \
		} while(0)

		#else

		// compiled out
		#define NDBG(tid, fd, tag, fmt, ...) do {} while(0)
		#define NDBG_ERR(tid, fd, tag, fmt, ...) do {} while(0)

		#endif

src/posix/tcp.cpp

+31 −44

Original line number	Diff line number	Diff line
		@@ -183,64 +183,51 @@ void netplus::tcp::bind(){
		}

		size_t netplus::tcp::sendData(buffer &data,int flags){
		NetException exception;

		int rval=0;
		int s=0;

		if(data.ptr)
		rval=::send(_Socket,data.data.ptr,data.size,flags);
		s=::send(_Socket,data.data.ptr,data.size,flags);
		else
		rval=::send(_Socket,data.data.buf,data.size,flags);

		if (rval < 0) {
		int etype = NetException::Error;

		if (errno == EAGAIN \|\| errno == EWOULDBLOCK)
		etype = NetException::Note;
		else if (errno == EPIPE \|\| errno == ECONNRESET)
		etype = NetException::Error;

		NetException ex;
		char errstr[512];
		strerror_r_netplus(errno, errstr, 512);
		ex[etype] << "Socket senddata failed on Socket: " << _Socket << " ErrorMsg: " << errstr;
		throw ex;
		}

		s=::send(_Socket,data.data.buf,data.size,flags);

		return (size_t)rval;
		if (s < 0) {
		if (errno == EAGAIN \|\| errno == EWOULDBLOCK) {
		NetException exception;
		exception[NetException::Note] << "Resource temporarily unavailable";
		throw exception;
		}

		size_t netplus::tcp::recvData(buffer &data,int flags){
		NetException exception;

		if(data.ptr){
		exception[NetException::Error] << "Socket recvdata error buffer with const ptr not supported!";
		exception[NetException::Error] << "send failed: " << strerror(errno);
		throw exception;
		}

		int recvsize=::recv(_Socket,
		data.data.buf,
		data.size,
		flags
		);

		if(recvsize<0){
		int etype=NetException::Error;
		return (size_t)s;
		}

		if(errno==EAGAIN \|\| errno==EWOULDBLOCK)
		etype=NetException::Note;
		size_t netplus::tcp::recvData(buffer& data, int flags){
		ssize_t r = ::recv(_Socket, data.data.buf, data.size, flags);

		if (r > 0) {
		data.size = (size_t)r;
		return (size_t)r;
		}

		char errstr[512];
		strerror_r_netplus(errno,errstr,512);
		if (r == 0) {
		return 0;
		}

		exception[etype] << "Socket recvdata failed on Socket: " << _Socket
		<< " ErrorMsg: " << errstr;
		if (r < 0) {
		if (errno == EAGAIN \|\| errno == EWOULDBLOCK) {
		NetException exception;
		exception[NetException::Note] << "Resource temporarily unavailable";
		throw exception;
		}
		return (size_t)recvsize;
		NetException exception;
		exception[NetException::Error] << "recv failed: " << strerror(errno);
		throw exception;
		}
		}


		void netplus::tcp::connect(std::unique_ptr<socket> &srvsock){
		NetException exception;