new locking (c5c22754) · Commits · jan.koester / libnetplus

src/eventapi.h

+2 −2

Original line number	Diff line number	Diff line
		@@ -25,6 +25,8 @@
		* SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
		*******************************************************************************/

		#include <mutex>

		#include "socket.h"
		#include "connection.h"

		@@ -83,12 +85,10 @@ namespace netplus {
		void unlockCon(int pos);
		bool trylockCon(int pos);
		private:

		void _setpollEvents(con *curcon,int events);
		int _pollFD;
		struct poll_event *_Events;
		socket *_ServerSocket;
		std::atomic<bool> _WaitLock;
		};

		class event : public poll {

src/linux/event.cpp

+80 −52

Original line number	Diff line number	Diff line
		@@ -57,7 +57,6 @@ namespace netplus {

		poll::poll(socket* serversocket) {
		_ServerSocket = serversocket;
		_WaitLock=false;
		};

		poll::~poll() {
		@@ -101,6 +100,7 @@ namespace netplus {
		};

		int poll::pollState(int pos){

		con pcon = (con)_Events[pos].data.ptr;
		NetException exception;

		@@ -118,7 +118,6 @@ namespace netplus {
		}

		unsigned int poll::waitEventHandler() {
		bool expected = false;

		int ret = epoll_wait(_pollFD, (struct epoll_event*)_Events, _ServerSocket->getMaxconnections(), -1);

		@@ -135,6 +134,11 @@ namespace netplus {
		con *ccon;
		try {
		ccon = new con(this);
		bool expected = false;
		if(!ccon->conlock.compare_exchange_strong(expected,true)){
		exception[NetException::Note] << "ConnectEventHandler: connection can'T lock this shoulden't happend";
		throw exception;
		}
		ccon->csock = _ServerSocket->accept();
		ccon->csock->setnonblocking();
		std::string ip;
		@@ -283,15 +287,28 @@ namespace netplus {
		bool event::_Run = true;
		bool event::_Restart = false;

		struct WorkerArgs {
		WorkerArgs(){
		};

		WorkerArgs(const WorkerArgs &args){
		eventptr=args.eventptr;
		jobs=args.jobs;
		};

		std::mutex sync;
		eventapi* eventptr;
		int jobs;
		};

		class EventWorker{
		public:

		EventWorker(void* args) {
		eventapi* eventptr = ((eventapi*)args);
		eventapi* eventptr = ((WorkerArgs*)args)->eventptr;
		while (event::_Run) {
		try {
		int wfd = eventptr->waitEventHandler();
		for (int i = 0; i < wfd; ++i) {
		((WorkerArgs*)args)->sync.lock();
		for (int i = 0; i < ((WorkerArgs*)args)->jobs; ++i) {
		try {
		int state = eventptr->pollState(i);
		if(eventptr->trylockCon(i)){
		@@ -303,8 +320,6 @@ namespace netplus {
		case poll::EventHandlerStatus::EVOUT:
		eventptr->WriteEventHandler(i);
		break;
		case poll::EventHandlerStatus::EVWAIT:
		continue;
		default:
		NetException excep;
		excep[NetException::Note] << "Eventworker: nothing todo close connection";
		@@ -320,6 +335,8 @@ namespace netplus {
		case poll::EventHandlerStatus::EVCON:
		eventptr->ConnectEventHandler(i);
		break;
		case poll::EventHandlerStatus::EVWAIT:
		continue;
		default:
		break;
		}
		@@ -334,16 +351,7 @@ namespace netplus {

		}
		}
		}
		catch (NetException& e) {
		switch (e.getErrorType()) {
		case NetException::Critical:
		throw e;
		default:
		std::cerr << e.what() << std::endl;
		break;
		}
		}
		((WorkerArgs*)args)->sync.lock();
		}
		}
		};
		@@ -384,12 +392,18 @@ namespace netplus {
		signal(SIGPIPE, SIG_IGN);
		initEventHandler();
		MAINWORKERLOOP:

		std::vector<WorkerArgs> targs;
		std::vector<std::thread> thpool;

		for (unsigned long i = 0; i < thrs; i++) {
		try {
		thpool.push_back(std::thread([this](){
		new EventWorker((void*)this);
		WorkerArgs args;
		args.eventptr=this;
		args.jobs=0;
		args.sync.lock();
		targs.push_back(args);
		thpool.push_back(std::thread([i,targs](){
		new EventWorker((void*)&targs[i]);
		}));
		}
		catch (NetException& e) {
		@@ -397,6 +411,20 @@ namespace netplus {
		}
		}


		int expected=0;
		while (event::_Run) {
		try {
		int wait=waitEventHandler();
		for(auto arg : targs){
		arg.jobs=wait;
		arg.sync.unlock();
		}
		}catch(NetException &e){
		std::cerr << e.what() << std::endl;
		}
		}

		for(std::vector<std::thread>::iterator thd = thpool.begin(); thd!=thpool.end(); ++thd){
		thd->join();
		}