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

src/event/epoll.cpp

+94 −177

Original line number	Diff line number	Diff line
		@@ -86,8 +86,7 @@ namespace netplus {
		/EventHandler/
		virtual int waitEventHandler(int timeout)=0;
		virtual void ConnectEventHandler(int pos,const int tid,ULONG_PTR args)=0;
		virtual void ReadEventHandler(int pos,const int tid,ULONG_PTR args)=0;
		virtual void WriteEventHandler(int pos,const int tid,ULONG_PTR args)=0;
		virtual void IoEventHandler(int pos, int events, const int tid, ULONG_PTR args)=0;
		virtual void CloseEventHandler(int pos,const int tid,ULONG_PTR args)=0;
		protected:
		eventapi *evconnection;
		@@ -135,14 +134,14 @@ namespace netplus {
		return _Events[pos].data.fd;
		}

		void setpollEvents(int pos,int events){
		int fd = _Events[pos].data.fd;
		struct epoll_event setevent = { 0 };
		setevent.events = events;
		setevent.data.fd = fd;
		if (epoll_ctl(_pollFD, EPOLL_CTL_MOD,fd,&setevent) < 0) {
		void setpollEventsFd(int fd, int events) {
		epoll_event ev {};
		ev.events = events;
		ev.data.fd = fd;

		if (epoll_ctl(_pollFD, EPOLL_CTL_MOD, fd, &ev) < 0) {
		NetException except;
		except[NetException::Error] << "_setEpollEvents: can change socket!";
		except[NetException::Error] << "setpollEventsFd: epoll_ctl MOD failed: " << strerror(errno);
		throw except;
		}
		}
		@@ -168,7 +167,7 @@ namespace netplus {

		void ConnectEventHandler(int pos, const int tid, ULONG_PTR args) {
		NetException exception;

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

		@@ -206,175 +205,100 @@ namespace netplus {
		epoll_ctl(_pollFD, EPOLL_CTL_ADD, fd, &setevent);

		evconnection->ConnectEvent(*ccon, tid, args);
		}catch(NetException &e){
		if(e.getErrorType()==NetException::Note)
		return;
		throw;
		}
		}

		void ReadEventHandler(int pos, const int tid, ULONG_PTR args) {
		void IoEventHandler(int pos, int events, const int tid, ULONG_PTR args) {
		int fd = _Events[pos].data.fd;
		std::shared_ptr<con> rcon;
		std::shared_ptr<con> c;

		{
		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()) c = it->second;
		}
		if (!c \|\| !c->csock) return;

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

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

		try {
		// ------------------------------------------------------------
		// 1) HANDSHAKE PHASE (async)
		// ------------------------------------------------------------
		if (!rcon->csock->getHandshakeDone()) {
		// Run as far as possible (may throw Note)
		rcon->csock->handshake_after_accept();
		// 0) always flush socket pending bytes first
		c->csock->flush_out();

		// Handshake may have produced outgoing bytes -> flush now
		rcon->csock->flush_out();
		// 1) handshake progression (may need IN or OUT)
		if (!c->csock->getHandshakeDone()) {
		c->csock->handshake_after_accept();
		c->csock->flush_out();
		}

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

		// 2) read application data if handshake done
		if (c->csock->getHandshakeDone() && (events & EPOLLIN)) {
		bool got = false;
		for (;;) {
		buffer buf(BLOCKSIZE);
		size_t rcv = 0;

		try {
		rcv = rcon->csock->recvData(buf, 0);
		rcv = c->csock->recvData(buf, 0);
		} catch (NetException& e) {
		if (e.getErrorType() == NetException::Note) break; // kernel empty
		if (e.getErrorType() == NetException::Note) break;
		throw;
		}

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

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

		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);
		if (rcv == 0) break;
		c->RecvData.append(buf.data.buf, rcv);
		got = true;
		}
		if (got && !c->RecvData.empty()) {
		evconnection->RequestEvent(*c, tid, args);
		}


		void WriteEventHandler(int pos, const int tid, ULONG_PTR args) {
		int fd = _Events[pos].data.fd;
		std::shared_ptr<con> wcon;

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

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

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

		try {
		// ------------------------------------------------------------
		// 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()) {
		size_t sendlen = (wcon->SendData.size() > BLOCKSIZE)
		? BLOCKSIZE
		: wcon->SendData.size();

		buffer out(wcon->SendData.data(), sendlen);
		// 3) send app data if handshake done
		if (c->csock->getHandshakeDone() && (events & EPOLLOUT)) {
		while (!c->SendData.empty()) {
		size_t sendlen = std::min((size_t)BLOCKSIZE, c->SendData.size());
		buffer out(c->SendData.data(), sendlen);
		size_t consumed = 0;

		try {
		consumed = wcon->csock->sendData(out, 0);
		// sendData may queue TLS records internally -> flush them now
		wcon->csock->flush_out();
		consumed = c->csock->sendData(out, 0);
		c->csock->flush_out();
		} catch (NetException& e) {
		if (e.getErrorType() == NetException::Note) {
		// need OUT again
		int ev = EPOLLIN \| EPOLLOUT \| EPOLLRDHUP \| EPOLLONESHOT;
		setpollEvents(pos, ev);
		return;
		}
		if (e.getErrorType() == NetException::Note) break;
		throw;
		}

		if (consumed == 0) break;

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

		wcon->lasteventime = time(nullptr);
		c->SendData.erase(c->SendData.begin(), c->SendData.begin() + consumed);
		}

		// ------------------------------------------------------------
		// 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);
		if (c->SendData.empty() && !c->csock->hasPendingWrite()) {
		evconnection->ResponseEvent(*c, tid, args);
		}
		}

		// 4) rearm ONESHOT ONLY ONCE
		int next = EPOLLRDHUP \| EPOLLONESHOT;

		if (!c->csock->getHandshakeDone()) {
		// during handshake: always both!
		next \|= EPOLLIN \| EPOLLOUT;
		} else {
		// Handshake not done yet -> keep both
		setpollEvents(pos, EPOLLIN \| EPOLLOUT \| EPOLLRDHUP \| EPOLLONESHOT);
		next \|= EPOLLIN;
		if (!c->SendData.empty() \|\| c->csock->hasPendingWrite())
		next \|= EPOLLOUT;
		}

		setpollEventsFd(fd, next);

		} catch (NetException& e) {
		if (e.getErrorType() == NetException::Note) {
		setpollEvents(pos, EPOLLIN \| EPOLLOUT \| EPOLLRDHUP \| EPOLLONESHOT);
		// Need more IO, arm both
		setpollEventsFd(fd, EPOLLIN \| EPOLLOUT \| EPOLLRDHUP \| EPOLLONESHOT);
		return;
		}
		CloseEventHandler(pos, tid, args);
		@@ -479,14 +403,7 @@ namespace netplus {
		continue;
		}

		if (events & EPOLLIN) {
		pollptr.ReadEventHandler(i, tid, args);
		}

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

		} catch (NetException& e) {
		std::cerr << "EventWorker Thread " << tid
		@@ -603,8 +520,8 @@ namespace netplus {

		struct epoll_event setevent = {0};

		setevent.events = EPOLLIN \| EPOLLEXCLUSIVE;
		setevent.data.ptr = nullptr;
		setevent.events = EPOLLIN;
		setevent.data.fd = _ServerSocket->fd();

		if (epoll_ctl(_pollFD, EPOLL_CTL_ADD,_ServerSocket->fd(),&setevent) < 0) {
		exception[NetException::Critical] << "initEventHandler: can't create epoll";