fixed (bc8e0c4c) · Commits · jan.koester / libnetplus

src/event/epoll.cpp

+42 −76

Original line number	Diff line number	Diff line
		@@ -63,7 +63,7 @@ namespace netplus {
		std::atomic<bool> event::Restart(false);

		std::vector<socket*> SSOCKETS;
		std::map<int,con*> CONNECTIONS;
		std::map<int,std::shared_ptr<con>> CONNECTIONS;

		std::mutex POLL_HANDLER_MUTEX;

		@@ -168,16 +168,10 @@ namespace netplus {
		void ConnectEventHandler(int pos, const int tid, ULONG_PTR args) {
		NetException exception;

		std::unique_ptr<con> ccon;

		std::shared_ptr<con> ccon;
		evconnection->CreateConnection(ccon);

		if(!ccon){
		char errstr[255];
		strerror_r_netplus(errno,errstr,255);
		std::cerr << "ConnectEventHandler: Failed to accept connection or invalid socket: " << errstr << std::endl;
		return;
		}
		if (!ccon) return;

		if(_ServerSocket->_Type==sockettype::TCP){
		ccon->csock=std::make_unique<tcp>(-1);
		@@ -188,45 +182,32 @@ namespace netplus {
		ccon->csock=std::make_unique<ssl>( srv->_cert,-1);
		}

		if (!ccon->csock) {
		// The connection failed to establish or the socket is invalid.
		// Clean up and return, perhaps log an error.
		char errstr[255];
		strerror_r_netplus(errno,errstr,255);
		std::cerr << "ConnectEventHandler: Failed to accept connection or invalid socket: " << errstr << std::endl;
		return;
		}

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

		std::string ip;
		ccon->csock->getAddress(ip);
		std::cout << "Connected: " << ip << std::endl;

		// 3) Bridge to shared_ptr
		ccon->lasteventime = time(nullptr);

		struct epoll_event setevent { 0 };
		setevent.events = EPOLLIN \| EPOLLONESHOT \| EPOLLET;
		// 4) Register with Epoll
		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();

		int estate = epoll_ctl(_pollFD, EPOLL_CTL_ADD,ccon->csock->fd(), &setevent);

		if ( estate < 0 ) {
		if (epoll_ctl(_pollFD, EPOLL_CTL_ADD, ccon->csock->fd(), &setevent) < 0) {
		char errstr[255];
		strerror_r_netplus(errno, errstr, 255);
		exception[NetException::Error] << "ConnectEventHandler: can't add socket to epoll: " << errstr;
		exception[NetException::Error] << "ConnectEventHandler: epoll_ctl add failed: " << errstr;
		throw exception;
		}

		evconnection->ConnectEvent(*ccon, tid, args);

		{
		std::lock_guard<std::mutex> clock(ccon->event_mutex),global_lock(POLL_HANDLER_MUTEX);
		CONNECTIONS.emplace(ccon->csock->fd(),ccon.get());
		ccon.release();
		std::lock_guard<std::mutex> glk(POLL_HANDLER_MUTEX);
		CONNECTIONS.emplace(ccon->csock->fd(), std::move(ccon));
		}
		}

		};

		void ReadEventHandler(int pos, const int tid, ULONG_PTR args) {
		con* rcon = reinterpret_cast<con*>(_Events[pos].data.ptr);
		@@ -342,56 +323,41 @@ namespace netplus {
		}

		void CloseEventHandler(int pos, const int tid, ULONG_PTR args) {
		con* ccon = reinterpret_cast<con*>(_Events[pos].data.ptr);
		if (!ccon) return;

		// Lock connection first (prevents concurrent Read/Write handlers on same con)
		std::unique_lock<std::mutex> conn_lock(ccon->event_mutex);
		// Get the raw pointer from epoll data
		con* raw_ptr = reinterpret_cast<con*>(_Events[pos].data.ptr);
		if (!raw_ptr) return;

		// Cache fd early; after close it may be invalid / reused
		int fd = -1;
		if (ccon->csock)
		fd = ccon->csock->fd();
		std::shared_ptr<con> ccon;

		// 1) Remove from epoll (ignore ENOENT / EBADF)
		// 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) {
		if (epoll_ctl(_pollFD, EPOLL_CTL_DEL, fd, nullptr) < 0) {
		if (errno != ENOENT && errno != EBADF) {
		// Log, but continue cleanup anyway
		NetException except;
		char errstr[255];
		strerror_r_netplus(errno, errstr, 255);
		except[NetException::Error]
		<< "CloseEventHandler: failed to remove fd from epoll: " << errstr;
		// don't throw here; still must cleanup safely
		auto it = CONNECTIONS.find(fd);
		if (it != CONNECTIONS.end()) {
		ccon = it->second;
		CONNECTIONS.erase(it);
		}
		}
		}

		// 2) Inform application layer (do NOT hold global map lock here)
		try {
		evconnection->DisconnectEvent(*ccon, tid, args);
		} catch (...) {
		// don't let DisconnectEvent prevent cleanup
		}
		if (!ccon) return;

		// 3) Close socket
		if (ccon->csock) {
		try { ccon->csock->close(); } catch (...) {}
		}
		// 2) Synchronize
		std::unique_lock<std::mutex> conn_lock(ccon->event_mutex);

		// 4) Remove from global map (use cached fd)
		{
		std::lock_guard<std::mutex> global_lock(POLL_HANDLER_MUTEX);
		if (fd >= 0) CONNECTIONS.erase(fd);
		// 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 {
		evconnection->DisconnectEvent(*ccon, tid, args);
		ccon->csock->close();
		} catch (...) {}
		}

		// 5) Clear epoll event slot pointer
		_Events[pos].data.ptr = nullptr;

		// 6) Destroy connection object
		conn_lock.unlock();
		delete ccon;
		}

		private:

src/event/kqueue.cpp

+2 −2

Original line number	Diff line number	Diff line
		@@ -61,7 +61,7 @@ namespace netplus {
		std::mutex POLL_HANDLER_MUTEX;

		std::vector<socket*> SSOCKETS;
		std::map<int,std::unique_ptr<con>> CONNECTIONS;
		std::map<int,std::shared_ptr<con>> CONNECTIONS;

		class pollapi {
		public:
		@@ -173,7 +173,7 @@ namespace netplus {
		NetException exception;

		// 1) Create as unique_ptr to satisfy the API requirement
		std::unique_ptr<con> ucon;
		std::shared_ptr<con> ucon;
		evconnection->CreateConnection(ucon);

		if (!ucon) {

src/eventapi.h

+1 −1

Original line number	Diff line number	Diff line
		@@ -46,7 +46,7 @@ namespace netplus {
		virtual void ConnectEvent(con &curcon,const int tid, ULONG_PTR args)=0;
		virtual void DisconnectEvent(con &curcon,const int tid, ULONG_PTR args)=0;
		/memory allocation/
		virtual void CreateConnection(std::unique_ptr<con> &res)=0;
		virtual void CreateConnection(std::shared_ptr<con> &res)=0;
		protected:
		eventapi(){};
		};