test (d975de7f) · Commits · jan.koester / libnetplus

src/connection.cpp

+1 −0

Original line number	Diff line number	Diff line
		@@ -52,6 +52,7 @@ netplus::condata<char>::condata() : vector(){
		}

		netplus::condata<char>::condata(const condata<char> &src) : vector(src) {
		pos=src.pos;;
		}

		netplus::condata<char>::condata(const std::vector<char> &src){

src/event/epoll.cpp

+181 −146

Original line number	Diff line number	Diff line
		@@ -131,16 +131,17 @@ namespace netplus {
		return _Events[pos].events;
		}

		int getEventFd(int pos) {
		return _Events[pos].data.fd;
		}

		void setpollEvents(int pos,int events){
		NetException except;
		int fd = _Events[pos].data.fd;
		struct epoll_event setevent = { 0 };
		setevent.events = events;
		setevent.data.ptr = _Events[pos].data.ptr;

		con curcon = reinterpret_cast<con>(_Events[pos].data.ptr);

		if (epoll_ctl(_pollFD, EPOLL_CTL_MOD,curcon->csock->fd()
		,&setevent) < 0) {
		setevent.data.fd = fd;
		if (epoll_ctl(_pollFD, EPOLL_CTL_MOD,fd,&setevent) < 0) {
		NetException except;
		except[NetException::Error] << "_setEpollEvents: can change socket!";
		throw except;
		}
		@@ -182,7 +183,13 @@ namespace netplus {
		ccon->csock=std::make_unique<ssl>( srv->_cert,-1);
		}

		try {
		_ServerSocket->accept(ccon->csock);
		} catch (NetException& e) {
		if (e.getErrorType() == NetException::Note) return;
		throw e;
		}

		ccon->csock->setFlag(O_NONBLOCK, 1);

		// 3) Bridge to shared_ptr
		@@ -192,9 +199,12 @@ namespace netplus {
		struct epoll_event setevent = { 0 };
		// We use EPOLLET (Edge Triggered) and EPOLLONESHOT for thread safety
		setevent.events = EPOLLIN \| EPOLLOUT \| EPOLLRDHUP \| EPOLLONESHOT \| EPOLLET;
		setevent.data.ptr = ccon.get();
		setevent.data.fd = ccon->csock->fd();

		if (epoll_ctl(_pollFD, EPOLL_CTL_ADD, ccon->csock->fd(), &setevent) < 0) {
		if (errno == EAGAIN \|\| errno == EWOULDBLOCK) {
		return;
		}
		char errstr[255];
		strerror_r_netplus(errno, errstr, 255);
		exception[NetException::Error] << "ConnectEventHandler: epoll_ctl add failed: " << errstr;
		@@ -210,154 +220,168 @@ namespace netplus {
		}

		void ReadEventHandler(int pos, const int tid, ULONG_PTR args) {
		con* rcon = reinterpret_cast<con*>(_Events[pos].data.ptr);
		if (!rcon \|\| !rcon->csock) return;
		int fd = _Events[pos].data.fd;
		std::shared_ptr<con> rcon;

		{
		std::lock_guard<std::mutex> lk(POLL_HANDLER_MUTEX);
		auto it = CONNECTIONS.find(fd);
		if (it != CONNECTIONS.end()) {
		rcon = it->second;
		}
		}

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

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

		try {
		for (;;) {
		buffer buf(BLOCKSIZE);

		size_t rcv = 0;
		try {
		rcv = rcon->csock->recvData(buf, 0);
		} catch (NetException& e) {
		if (e.getErrorType() == NetException::Note) {
		// would-block: stop draining
		break;
		}
		throw; // real error
		// EAGAIN: Keine weiteren Daten im Kernel-Buffer
		setpollEvents(pos, EPOLLIN \| EPOLLONESHOT \| EPOLLET \| EPOLLRDHUP);
		return;
		}

		if (rcv == 0) {
		// peer closed
		NetException e;
		e[NetException::Close] << "ReadEventHandler: peer closed connection (EOF)";
		throw e;
		throw; // Kritischer Fehler
		}

		if (rcv > 0) {
		rcon->RecvData.append(buf.data.buf, rcv);
		rcon->lasteventime = time(nullptr);

		// Let protocol layer consume/produce data NOW
		evconnection->RequestEvent(*rcon, tid, args);

		// If we now have pending output -> switch to write
		if (rcon->SendOff < rcon->SendData.size()) {
		setpollEvents(pos, EPOLLOUT \| EPOLLONESHOT);
		return;
		break;
		}

		// else continue draining reads until would-block
		else {
		if (!rcon->csock->getHandshakeDone()) {
		continue;
		} else {
		// Handshake ist fertig und wir lesen 0? -> Verbindung zu.
		throw NetException() << "Client closed connection";
		}

		// Draining done. Rearm interest.
		if (rcon->SendOff < rcon->SendData.size())
		setpollEvents(pos, EPOLLOUT \| EPOLLONESHOT);
		else
		setpollEvents(pos, EPOLLIN \| EPOLLONESHOT);
		return;

		} catch (NetException& e) {
		if (e.getErrorType() == NetException::Note) {
		// Usually won't happen anymore because we catch Note above,
		// but keep it defensive.
		if (rcon->SendOff < rcon->SendData.size())
		setpollEvents(pos, EPOLLOUT \| EPOLLONESHOT);
		else
		setpollEvents(pos, EPOLLIN \| EPOLLONESHOT);
		return;
		}
		throw; // Close/Error -> your outer loop should cleanup
		}
		// Entscheidung nach der Schleife: Müssen wir jetzt senden?
		if (!rcon->SendData.empty()) {
		setpollEvents(pos, EPOLLOUT \| EPOLLONESHOT \| EPOLLET \| EPOLLRDHUP);
		} else {
		setpollEvents(pos, EPOLLIN \| EPOLLONESHOT \| EPOLLET \| EPOLLRDHUP);
		}
		} catch (NetException& e) {
		CloseEventHandler(pos, tid, args);
		}
		}

		void WriteEventHandler(int pos, const int tid, ULONG_PTR args) {
		con* wcon = reinterpret_cast<con*>(_Events[pos].data.ptr);
		// 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;
		}
		}

		// 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 {
		// 1) Flush existing queued plaintext as far as possible
		// 4) Sende-Schleife: Versuche SendData mit Hilfe der buffer struct zu leeren
		while (!wcon->SendData.empty()) {
		const size_t toSend = std::min<size_t>(BLOCKSIZE, wcon->SendData.size());
		buffer out(wcon->SendData.data(), toSend);
		// Bestimme Blockgröße (entweder restlicher Buffer oder BLOCKSIZE)
		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);

		const size_t consumed = wcon->csock->sendData(out, 0); // plaintext consumed
		size_t consumed = 0;
		try {
		// SSL oder TCP sendData
		consumed = wcon->csock->sendData(out, 0);
		} 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);
		return;
		}
		throw; // Kritischer Fehler (z.B. Connection Reset)
		}

		if (consumed == 0) {
		// would-block OR ssl is still flushing internal TLS record
		setpollEvents(pos, EPOLLOUT \| EPOLLONESHOT);
		// Socket-Buffer des Kernels voll
		setpollEvents(pos, EPOLLOUT \| EPOLLONESHOT \| EPOLLET \| EPOLLRDHUP);
		return;
		}

		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);
		}

		// 2) Queue is empty -> ask protocol layer for more (ONCE)
		// 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()) {
		evconnection->ResponseEvent(*wcon, tid, args);
		}

		// 3) If ResponseEvent produced more output, stay in write mode
		// 6) Finaler Zustandswechsel ("New Way")
		if (!wcon->SendData.empty()) {
		setpollEvents(pos, EPOLLOUT \| EPOLLONESHOT);
		return;
		// 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);
		}

		// 4) Otherwise go back to read
		setpollEvents(pos, EPOLLIN \| EPOLLONESHOT);
		return;

		} catch (NetException& e) {
		if (e.getErrorType() == NetException::Note) {
		setpollEvents(pos, EPOLLOUT \| EPOLLONESHOT);
		return;
		}
		throw;
		// Bei Fehlern (außer "Note") Verbindung schließen
		CloseEventHandler(pos, tid, args);
		}
		}

		void CloseEventHandler(int pos, const int tid, ULONG_PTR args) {
		// Get the raw pointer from epoll data
		con* raw_ptr = reinterpret_cast<con*>(_Events[pos].data.ptr);
		if (!raw_ptr) return;

		int fd = _Events[pos].data.fd;
		std::shared_ptr<con> ccon;

		// 1) Extract from map
		{
		std::lock_guard<std::mutex> global_lock(POLL_HANDLER_MUTEX);
		int fd = (raw_ptr->csock) ? raw_ptr->csock->fd() : -1;
		if (fd >= 0) {
		std::lock_guard<std::mutex> lock(POLL_HANDLER_MUTEX);
		auto it = CONNECTIONS.find(fd);
		if (it != CONNECTIONS.end()) {
		ccon = it->second;
		// Move the pointer out of the map into our local variable
		ccon = std::move(it->second);
		// The map entry now contains an empty shared_ptr, we must erase it
		CONNECTIONS.erase(it);
		}
		}
		}
		} // Lock released here

		if (!ccon) return;
		if (ccon) {
		// Now 'ccon' is the sole owner if it was the only one in the map
		std::lock_guard<std::mutex> conn_lock(ccon->event_mutex);

		// 2) Synchronize
		std::unique_lock<std::mutex> conn_lock(ccon->event_mutex);
		// 1. Remove from Epoll first to stop getting events
		epoll_ctl(_pollFD, EPOLL_CTL_DEL, fd, nullptr);

		// 3) Epoll Cleanup
		// On Linux, always remove from epoll before closing the FD
		if (ccon->csock) {
		epoll_ctl(_pollFD, EPOLL_CTL_DEL, ccon->csock->fd(), nullptr);
		try {
		// 2. Notify protocol layer
		evconnection->DisconnectEvent(*ccon, tid, args);

		// 3. Physical close
		if(ccon->csock)
		ccon->csock->close();
		} catch (...) {}
		}

		_Events[pos].data.ptr = nullptr;
		}

		private:
		@@ -392,60 +416,71 @@ namespace netplus {
		class EventWorker {
		public:
		EventWorker(int tid, ULONG_PTR args, EventWorkerArgs* eargs) {

		poll pollptr(eargs->ssocket, eargs->event, eargs->pollfd, eargs->timeout);

		int wait = 0;

		// In EventWorker::EventWorker:
		while (event::Running) {
		try {
		wait = pollptr.waitEventHandler(eargs->timeout);
		for (int i = 0; i < wait; ++i) {
		try {
		int events = pollptr.pollState(i);

		// Check for listener socket (data.ptr is nullptr)
		if (events == pollapi::EventHandlerStatus::EVCON) {
		// 1. Prüfen, ob es sich um den Server-Socket (Listening) handelt
		int state = pollptr.pollState(i);
		if (state == pollapi::EventHandlerStatus::EVCON) {
		pollptr.ConnectEventHandler(i, tid, args);
		continue;
		}

		events = pollptr.getEventFlags(i);
		// 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;

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

		// Check for error/hangs (EPOLLHUP/EPOLLERR)
		if (events & (EPOLLHUP \| EPOLLERR)) {
		// Treat as close event if any of these flags are set
		// 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
		if (events & (EPOLLHUP \| EPOLLERR \| EPOLLRDHUP)) {
		pollptr.CloseEventHandler(i, tid, args);
		continue;
		}

		// Prioritize read events (new data from client)
		// Daten empfangen (Read)
		if (events & EPOLLIN) {
		pollptr.ReadEventHandler(i, tid, args);
		continue;
		}

		// Handle write events (ready to send data)
		if (events & EPOLLOUT) {
		// Daten senden (Write)
		else if (events & EPOLLOUT) {
		pollptr.WriteEventHandler(i, tid, args);
		continue;
		}

		// FALLBACK/DEFAULT HANDLER
		NetException e;
		e[NetException::Error] << "EventWorker: Unhandled epoll event type!";
		throw e;

		} catch (NetException& e) {
		std::cerr << 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);
		}

		if (e.getErrorType() == NetException::Critical)
		throw e;
		}
		}
		} catch (NetException& e) {
		std::cerr << e.what() << std::endl;
		std::cerr << "Poll Error: " << e.what() << std::endl;
		}
		}
		}
		@@ -545,7 +580,7 @@ namespace netplus {

		struct epoll_event setevent = {0};

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

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

src/event/iocp.cpp

+106 −158

File changed.

Preview size limit exceeded, changes collapsed.

src/socket.h

+16 −2

Original line number	Diff line number	Diff line
		@@ -169,6 +169,8 @@ namespace netplus {

		virtual bool hasPendingWrite() const { return false; }

		virtual bool getHandshakeDone() { return true; }

		virtual void connect(std::unique_ptr<socket>& csock) = 0;

		virtual void getAddress(std::string& addr) = 0;
		@@ -295,9 +297,14 @@ namespace netplus {
		#endif
		bool loadServerPrivateKeyDer(const std::string& keyDerPath);

		bool handleHandshakeAsynch();

		bool hasPendingWrite() const override {
		return !_send_record.empty(); // your TLS ciphertext buffer
		}

		bool getHandshakeDone() override { return _handshakeDone; }

		netplus::x509cert _cert;
		private:
		// --- Cryptographic Components ---
		@@ -325,12 +332,14 @@ namespace netplus {

		std::vector<uint8_t> _recv_record;
		std::vector<uint8_t> _rx_netbuf;
		std::vector<uint8_t> _rx_handshake_buffer;
		#ifdef Windows
		AcceptContext _acpt;
		std::array<char, 2 * (sizeof(SOCKADDR_STORAGE) + 16)> _acceptBuf{};

		std::vector<uint8_t> _hs_tx;
		size_t _hs_tx_off = 0;
		#endif

		enum class HsState {
		WAIT_CH,
		@@ -345,7 +354,7 @@ namespace netplus {

		HsState _hs_state = HsState::WAIT_CH;
		std::vector<uint8_t> _masterSecret;
		#endif

		size_t _recv_off = 0;

		std::vector<uint8_t> _handshake_transcript;
		@@ -404,6 +413,11 @@ namespace netplus {
		const std::vector<uint8_t>& content
		);

		bool _popHandshakeMsg(
		std::vector<uint8_t>& out,
		uint8_t& type
		);

		friend class poll;
		friend class event;
		friend class EventWorker;

src/ssl.cpp

+361 −238

File changed.

Preview size limit exceeded, changes collapsed.