Работа на корректным завершением для ModbusTCPMaster и MBSlave

TODO

@@ -67,6 +67,10 @@ Version 2.5
 UNET: сделать чтобы "канал"(UReceiver, USender) не вылетал если не удалось создать сокет, а собственно,
 делал периодические попытки пока не получиться...(т.к. сеть может подняться позже внешними системами мониторинга)
 
+libev
+=======
+- переписать UNetUDP
+- UniSetActivator (обработка сигналов и возможно вынести сюда DefaultEventLoop)
 
 SM: подумать насчёт асинхронности публикации событий и посылки уведомления (setValue/push) через очередь..
 

Utilities/MBTester/mbtcpserver.cc

@@ -49,7 +49,7 @@ int main( int argc, char** argv )
 	{
 		while(1)
 		{
-			opt = getopt_long(argc, argv, "hva:p:i:bc:", longopts, &optindex);
+			opt = getopt_long(argc, argv, "hva:p:i:c:", longopts, &optindex);
 
 			if( opt == -1 )
 				break;

Utilities/MBTester/mbtcptester.cc

@@ -121,8 +121,13 @@ int main( int argc, char** argv )
 
 	try
 	{
-		while( (opt = getopt_long(argc, argv, "hvna:w:z:r:x:c:b:d:s:t:p:i:ol:d:e:u:", longopts, &optindex)) != -1 )
+		while( 1 )
 		{
+			opt = getopt_long(argc, argv, "hvna:w:z:r:x:c:b:d:s:t:p:i:ol:d:e:u:", longopts, &optindex);
+
+			if( opt == -1 )
+				break;
+
 			switch (opt)
 			{
 				case 'h':

extensions/ModbusMaster/tests/MBTCPTestServer.cc

@@ -27,8 +27,6 @@ MBTCPTestServer::MBTCPTestServer( const std::unordered_set<ModbusAddr>& _vaddr,
 	lastWriteOutputSingleRegister(0),
 	lastForceCoilsQ(0, 0),
 	lastWriteOutputQ(0, 0),
-	thr(0),
-	isrunning(false),
 	disabled(false)
 {
 	ost::InetAddress ia(inetaddr.c_str());
@@ -87,15 +85,8 @@ MBTCPTestServer::MBTCPTestServer( const std::unordered_set<ModbusAddr>& _vaddr,
 // -------------------------------------------------------------------------
 MBTCPTestServer::~MBTCPTestServer()
 {
-	if( thr )
-	{
-		thr->stop();
-
-		if( thr->isRunning() )
-			thr->join();
-	}
-
-	delete sslot;
+	if( sslot )
+		sslot->terminate();
 }
 // -------------------------------------------------------------------------
 void MBTCPTestServer::setLog( std::shared_ptr<DebugStream> dlog )
@@ -104,21 +95,16 @@ void MBTCPTestServer::setLog( std::shared_ptr<DebugStream> dlog )
 		sslot->setLog(dlog);
 }
 // -------------------------------------------------------------------------
-void MBTCPTestServer::runThread()
-{
-	thr = new ThreadCreator<MBTCPTestServer>(this, &MBTCPTestServer::execute);
-	thr->start();
-}
-// -------------------------------------------------------------------------
 void MBTCPTestServer::execute()
 {
-	isrunning = true;
-	sslot->run( vaddr );
-	isrunning = false;
+	if( sslot )
+		sslot->run( vaddr, true );
 }
 // -------------------------------------------------------------------------
 void MBTCPTestServer::sigterm( int signo )
 {
+	if( sslot )
+		sslot->terminate();
 }
 // -------------------------------------------------------------------------
 ModbusRTU::mbErrCode MBTCPTestServer::readCoilStatus( ReadCoilMessage& query,

extensions/ModbusMaster/tests/MBTCPTestServer.h

@@ -5,7 +5,6 @@
 #include <atomic>
 #include <ostream>
 #include <unordered_set>
-#include "ThreadCreator.h"
 #include "modbus/ModbusTCPServerSlot.h"
 // -------------------------------------------------------------------------
 /*! Реализация MBTCPTestServer для тестирования */
@@ -31,14 +30,12 @@ class MBTCPTestServer
 				sslot->setIgnoreAddrMode(state);
 		}
 
-		void runThread(); /*!< запуск с отдельным потоком */
-
 		void execute();    /*!< основной цикл работы */
 		void setLog( std::shared_ptr<DebugStream> dlog );
 
 		inline bool isRunning()
 		{
-			return isrunning;
+			return ( sslot && sslot->isActive() );
 		}
 
 		inline void disableExchange( bool set = true )
@@ -157,8 +154,6 @@ class MBTCPTestServer
 #endif
 
 	private:
-		ThreadCreator<MBTCPTestServer>* thr;
-		std::atomic_bool isrunning;
 		bool disabled;
 		std::string myname;
 };

extensions/ModbusMaster/tests/test_mbtcpmaster.cc

@@ -72,7 +72,7 @@ static void InitTest()
 
 		//mbs->setVerbose(true);
 		CHECK( mbs != nullptr );
-		mbs->runThread();
+		mbs->execute();
 
 		for( int i = 0; !mbs->isRunning() && i < 10; i++ )
 			msleep(200);

extensions/ModbusMaster/tests/test_mbtcpmultimaster.cc

@@ -73,7 +73,7 @@ static void InitTest()
 
 		CHECK( mbs1 != nullptr );
 		mbs1->setReply(0);
-		mbs1->runThread();
+		mbs1->execute();
 
 		for( int i = 0; !mbs1->isRunning() && i < 10; i++ )
 			msleep(200);
@@ -105,7 +105,7 @@ static void InitTest()
 
 		CHECK( mbs2 != nullptr );
 		mbs2->setReply(0);
-		mbs2->runThread();
+		mbs2->execute();
 
 		for( int i = 0; !mbs2->isRunning() && i < 10; i++ )
 			msleep(200);

extensions/ModbusSlave/MBSlave.cc

@@ -147,9 +147,9 @@ MBSlave::MBSlave(UniSetTypes::ObjectId objId, UniSetTypes::ObjectId shmId, const
 
 	timeout_t aftersend_pause = conf->getArgInt("--" + prefix + "-aftersend-pause", it.getProp("afterSendPause"));
 
-	string stype = conf->getArgParam("--" + prefix + "-type", it.getProp("type"));
+	mbtype = conf->getArgParam("--" + prefix + "-type", it.getProp("type"));
 
-	if( stype == "RTU" )
+	if( mbtype == "RTU" )
 	{
 		// ---------- init RS ----------
 		string dev    = conf->getArgParam("--" + prefix + "-dev", it.getProp("device"));
@@ -183,7 +183,7 @@ MBSlave::MBSlave(UniSetTypes::ObjectId objId, UniSetTypes::ObjectId shmId, const
 		rs->setLog(l);
 		conf->initLogStream(l, prefix + "-exchangelog");
 	}
-	else if( stype == "TCP" )
+	else if( mbtype == "TCP" )
 	{
 		string iaddr = conf->getArgParam("--" + prefix + "-inet-addr", it.getProp("iaddr"));
 
@@ -201,8 +201,8 @@ MBSlave::MBSlave(UniSetTypes::ObjectId objId, UniSetTypes::ObjectId shmId, const
 		tcpserver->setReplyTimeout(reply_tout);
 
 		mbslot = std::static_pointer_cast<ModbusServerSlot>(tcpserver);
-		thr = make_shared< ThreadCreator<MBSlave> >(this, &MBSlave::execute_tcp);
-		thr->setFinalAction(this, &MBSlave::finalThread);
+		//thr = make_shared< ThreadCreator<MBSlave> >(this, &MBSlave::execute_tcp);
+		//thr->setFinalAction(this, &MBSlave::finalThread);
 		mbinfo << myname << "(init): init TCP connection ok. " << " inet=" << iaddr << " port=" << port << endl;
 
 		ostringstream n;
@@ -490,31 +490,14 @@ MBSlave::~MBSlave()
 {
 	cancelled = true;
 
-	if( tcpserver )
-	{
-		if( !tcpCancelled )
-		{
-			sigterm(SIGTERM);
-			timeout_t waitPause = updateStatTime / 10;
-
-			// специально делаем больше шагов(15).. чтобы с запасом..
-			for( int i = 0; i < 15 && !tcpCancelled; i++ )
-				msleep(waitPause);
-		}
-
-		if( !tcpCancelled )
-		{
-			dcrit << myname << "(~): TCP NOT CANCELED" << endl;
-		}
-	}
+	if( tcpserver && tcpserver->isActive() )
+		tcpserver->terminate();
 
 	if( thr && thr->isRunning() )
 	{
 		thr->stop();
 		//        thr->join();
 	}
-
-
 }
 // -----------------------------------------------------------------------------
 void MBSlave::finalThread()
@@ -633,11 +616,10 @@ void MBSlave::execute_tcp()
 
 	tcpCancelled = false;
 
-	tcpserver->run( vaddr );
+	tcpserver->run( vaddr, true );
 
-	tcpCancelled = true;
-
-	mbinfo << myname << "(execute_tcp): tcpserver stopped.." << endl;
+	//	tcpCancelled = true;
+	//	mbinfo << myname << "(execute_tcp): tcpserver stopped.." << endl;
 }
 // -------------------------------------------------------------------------
 void MBSlave::updateStatistics()
@@ -726,6 +708,7 @@ void MBSlave::updateTCPStatistics()
 
 		// суммарное количество по всем
 		askCount = 0;
+
 		for( const auto& s : sess )
 		{
 			if( !activated || cancelled )
@@ -796,7 +779,7 @@ void MBSlave::updateTCPStatistics()
 // -------------------------------------------------------------------------
 void MBSlave::updateThresholds()
 {
-	for( auto&& i: thrlist )
+	for( auto && i : thrlist )
 	{
 		try
 		{
@@ -857,7 +840,11 @@ void MBSlave::sysCommand( const UniSetTypes::SystemMessage* sm )
 			{
 				UniSetTypes::uniset_rwmutex_rlock l(mutex_start);
 				askSensors(UniversalIO::UIONotify);
-				thr->start();
+
+				if( mbtype == "RTU" && thr )
+					thr->start();
+				else if( mbtype == "TCP")
+					execute_tcp();
 			}
 
 			break;
@@ -1011,12 +998,13 @@ bool MBSlave::deactivateObject()
 {
 	mbinfo << myname << "(deactivateObject): ..." << endl;
 
+	if( cancelled )
+		return UniSetObject::deactivateObject();
+
 	activated = false;
 	cancelled = true;
 
-	if( tcpserver && tcpserver->isAcive() )
-		tcpserver->sigterm(SIGTERM);
-	else
+	if( mbtype == "RTU" )
 	{
 		try
 		{
@@ -1029,21 +1017,30 @@ bool MBSlave::deactivateObject()
 		}
 	}
 
+
 	return UniSetObject::deactivateObject();
 }
 // ------------------------------------------------------------------------------------------
 void MBSlave::sigterm( int signo )
 {
 	mbinfo << myname << ": ********* SIGTERM(" << signo << ") ********" << endl;
+
+	if( cancelled )
+	{
+		UniSetObject::sigterm(signo);
+		return;
+	}
+
 	activated = false;
 	cancelled = true;
 
 	if( tcpserver )
 	{
 		cancelled = true;
+		cerr << "********* MBSlave::sigterm" << endl;
 
-		if( tcpserver && tcpserver->isAcive() )
-			tcpserver->sigterm(signo);
+		if( tcpserver->isActive() )
+			tcpserver->terminate();
 	}
 	else
 	{
@@ -2464,6 +2461,7 @@ ModbusRTU::mbErrCode MBSlave::diagnostics( ModbusRTU::DiagnosticMessage& query,
 		ModbusRTU::DiagnosticRetMessage& reply )
 {
 	auto mbserver = dynamic_pointer_cast<ModbusServer>(mbslot);
+
 	if( !mbserver )
 		return ModbusRTU::erHardwareError;
 
@@ -2497,8 +2495,8 @@ ModbusRTU::mbErrCode MBSlave::diagnostics( ModbusRTU::DiagnosticMessage& query,
 	if( query.subf == ModbusRTU::dgClearCounters )
 	{
 		mbserver->resetAskCounter();
-		mbserver->resetErrCount(erOperationFailed,0);
-		mbserver->resetErrCount(ModbusRTU::erBadCheckSum,0);
+		mbserver->resetErrCount(erOperationFailed, 0);
+		mbserver->resetErrCount(ModbusRTU::erBadCheckSum, 0);
 		// другие счётчики пока не сбрасываем..
 		reply = query;
 		return ModbusRTU::erNoError;

extensions/ModbusSlave/MBSlave.h

@@ -572,6 +572,7 @@ class MBSlave:
 		std::string logserv_host = {""};
 		int logserv_port = {0};
 		VMonitor vmon;
+		std::string mbtype = { "" };
 
 		// ----------------------------------------------------------------------------
 		// TCPServer section..

extensions/SharedMemory/SharedMemory.cc

@@ -276,8 +276,10 @@ void SharedMemory::sysCommand( const SystemMessage* sm )
 // ------------------------------------------------------------------------------------------
 bool SharedMemory::deactivateObject()
 {
-	if( logserv )
-		logserv = nullptr;
+	cerr << myname << "************* deactivateObject()..." << endl;
+	workready = false;
+//	if( logserv && logserv->isRunning() )
+//		logserv->terminate();
 
 	return IONotifyController::deactivateObject();
 }
@@ -317,7 +319,7 @@ bool SharedMemory::activateObject()
 		activated = true;
 	}
 
-	cerr << "************************** activate: " << pt.getCurrent() << " msec " << endl;
+	cout << myname << ": ********** activate: " << pt.getCurrent() << " msec " << endl;
 	return res;
 }
 // ------------------------------------------------------------------------------------------
@@ -328,10 +330,14 @@ CORBA::Boolean SharedMemory::exist()
 // ------------------------------------------------------------------------------------------
 void SharedMemory::sigterm( int signo )
 {
+	cerr << myname << "************* SIGTERM...." << endl;
+	workready = false;
 	if( signo == SIGTERM && wdt )
 		wdt->stop();
 
-	//    raise(SIGKILL);
+	if( logserv && logserv->isRunning() )
+		logserv->terminate();
+
 	IONotifyController::sigterm(signo);
 }
 // ------------------------------------------------------------------------------------------

include/DefaultEventLoop.h

@@ -13,8 +13,8 @@
 class EventWatcher
 {
 	public:
-		EventWatcher(){}
-		virtual ~EventWatcher(){}
+		EventWatcher() {}
+		~EventWatcher() {}
 };
 // -------------------------------------------------------------------------
 /*!

include/LogSession.h

@@ -34,7 +34,7 @@ class LogSession
 	public:
 
 		LogSession(int sock, std::shared_ptr<DebugStream>& log, timeout_t cmdTimeout = 2000 );
-		virtual ~LogSession();
+		~LogSession();
 
 		typedef sigc::slot<void, LogSession*> FinalSlot;
 		void connectFinalSession( FinalSlot sl );

include/UTCPCore.h

@@ -24,7 +24,8 @@ namespace UTCPCore
 		{
 			pos = 0;
 			len = nbytes;
-			if( len <=0 ) // ??!!
+
+			if( len <= 0 ) // ??!!
 				return;
 
 			data = new unsigned char[nbytes];
@@ -36,7 +37,7 @@ namespace UTCPCore
 			pos = 0;
 			len = s.length();
 
-			if( len <=0 ) // ??!!
+			if( len <= 0 ) // ??!!
 				return;
 
 			data = new unsigned char[len];

include/modbus/ModbusRTUSlave.h

@@ -19,6 +19,7 @@
     \todo Разобратся с тем как отвечать на неправильные запросы!
         Формат ответных сообщений!!! Коды ошибок!!!
     \todo Доработать terminate, чтобы можно было прервать ожидание
+	\todo Перейти на libev..
 */
 class ModbusRTUSlave:
 	public ModbusServer
@@ -38,7 +39,7 @@ class ModbusRTUSlave:
 		}
 
 		virtual void terminate() override;
-		virtual bool isAcive() override;
+		virtual bool isActive() override;
 
 	protected:
 

include/modbus/ModbusServer.h

@@ -58,11 +58,11 @@ class ModbusServer
 		// ---------------------------------------------------------------------------------------
 		// сигнал вызова receive, ДО обработки realReceive()
 		// \return ModbusRTU::errNoError, тогда обработка продолжиться.
-		typedef sigc::signal<ModbusRTU::mbErrCode,const std::unordered_set<ModbusRTU::ModbusAddr>, timeout_t> PreReceiveSignal;
+		typedef sigc::signal<ModbusRTU::mbErrCode, const std::unordered_set<ModbusRTU::ModbusAddr>, timeout_t> PreReceiveSignal;
 		PreReceiveSignal signal_pre_receive();
 
 		// сигнал после обработки realReceive()
-		typedef sigc::signal<void,ModbusRTU::mbErrCode> PostReceiveSignal;
+		typedef sigc::signal<void, ModbusRTU::mbErrCode> PostReceiveSignal;
 		PostReceiveSignal signal_post_receive();
 		// ---------------------------------------------------------------------------------------
 
@@ -142,16 +142,19 @@ class ModbusServer
 		virtual void cleanupChannel() {}
 		virtual void terminate() {}
 
-		virtual bool isAcive() = 0;
+		virtual bool isActive() = 0;
 
 		// ------------ Статистика ---------------
 		typedef std::unordered_map<ModbusRTU::mbErrCode, size_t> ExchangeErrorMap;
 
 		ExchangeErrorMap getErrorMap();
 		size_t getErrCount( ModbusRTU::mbErrCode e );
-		size_t resetErrCount( ModbusRTU::mbErrCode e, size_t set=0 );
+		size_t resetErrCount( ModbusRTU::mbErrCode e, size_t set = 0 );
 
-		inline size_t getAskCount() { return askCount; }
+		inline size_t getAskCount()
+		{
+			return askCount;
+		}
 		void resetAskCounter();
 
 	protected:

include/modbus/ModbusTCPServer.h

@@ -111,6 +111,8 @@ class ModbusTCPServer:
 		virtual ModbusRTU::mbErrCode realReceive( const std::unordered_set<ModbusRTU::ModbusAddr>& vaddr, timeout_t msecTimeout ) override;
 
 		virtual void mainLoop();
+		void finish();
+
 		virtual void ioAccept(ev::io& watcher, int revents);
 		void onTimer( ev::timer& t, int revents );
 

include/modbus/ModbusTCPSession.h

@@ -55,7 +55,7 @@ class ModbusTCPSession:
 		// запуск обработки входящих запросов
 		void run();
 
-		virtual bool isAcive() override;
+		virtual bool isActive() override;
 
 	protected:
 

src/Communications/Modbus/ModbusRTUSlave.cc

@@ -141,7 +141,7 @@ void ModbusRTUSlave::terminate()
 	catch(...) {}
 }
 // -------------------------------------------------------------------------
-bool ModbusRTUSlave::isAcive()
+bool ModbusRTUSlave::isActive()
 {
 	return false;
 }

src/Communications/Modbus/ModbusServer.cc

@@ -1530,22 +1530,23 @@ mbErrCode ModbusServer::receive(const std::unordered_set<ModbusAddr>& vaddr, tim
 
 	if( !m_pre_signal.empty() )
 	{
-		ret = m_pre_signal.emit(vaddr,msecTimeout);
+		ret = m_pre_signal.emit(vaddr, msecTimeout);
+
 		if( ret != erNoError && ret != erSessionClosed )
 		{
-			errmap[ret] +=1;
+			errmap[ret] += 1;
 			return ret;
 		}
 	}
 
-	ret = realReceive(vaddr,msecTimeout);
+	ret = realReceive(vaddr, msecTimeout);
 
 	// собираем статистику..
 	if( ret != erTimeOut && ret != erSessionClosed )
 		askCount++;
 
 	if( ret != erNoError && ret != erSessionClosed )
-		errmap[ret] +=1;
+		errmap[ret] += 1;
 
 	if( ret != erSessionClosed )
 		m_post_signal.emit(ret);
@@ -1668,6 +1669,7 @@ ModbusServer::ExchangeErrorMap ModbusServer::getErrorMap()
 size_t ModbusServer::getErrCount( mbErrCode e )
 {
 	auto i = errmap.find(e);
+
 	if( i == errmap.end() )
 		return 0;
 
@@ -1677,6 +1679,7 @@ size_t ModbusServer::getErrCount( mbErrCode e )
 size_t ModbusServer::resetErrCount( mbErrCode e, size_t set )
 {
 	auto i = errmap.find(e);
+
 	if( i == errmap.end() )
 		return 0;
 

src/Communications/Modbus/ModbusTCPServer.cc

@@ -45,8 +45,8 @@ ModbusTCPServer::ModbusTCPServer( ost::InetAddress& ia, int _port ):
 // -------------------------------------------------------------------------
 ModbusTCPServer::~ModbusTCPServer()
 {
-	if( evloop )
-		evloop->terminate(this);
+	if( cancelled )
+		finish();
 }
 // -------------------------------------------------------------------------
 void ModbusTCPServer::setMaxSessions( unsigned int num )
@@ -120,7 +120,7 @@ void ModbusTCPServer::mainLoop()
 
 	{
 		evloop = DefaultEventLoop::inst();
-		evloop->run(this,false);
+		evloop->run(this, false);
 	}
 
 	if( dlog->is_info() )
@@ -131,6 +131,11 @@ void ModbusTCPServer::mainLoop()
 // -------------------------------------------------------------------------
 void ModbusTCPServer::terminate()
 {
+	finish();
+}
+// -------------------------------------------------------------------------
+void ModbusTCPServer::finish()
+{
 	if( cancelled )
 		return;
 
@@ -187,7 +192,7 @@ void ModbusTCPServer::getSessions( Sessions& lst )
 	}
 }
 // -------------------------------------------------------------------------
-bool ModbusTCPServer::isAcive()
+bool ModbusTCPServer::isActive()
 {
 	return !cancelled;
 }
@@ -328,6 +333,6 @@ mbErrCode ModbusTCPServer::preReceiveEvent(const std::unordered_set<ModbusAddr>
 	if( m_pre_signal.empty() )
 		return ModbusRTU::erNoError;
 
-	return m_pre_signal.emit(vaddr,tout);
+	return m_pre_signal.emit(vaddr, tout);
 }
 // -------------------------------------------------------------------------

src/Communications/Modbus/ModbusTCPServerSlot.cc

@@ -171,6 +171,6 @@ void ModbusTCPServerSlot::sigterm( int signo )
 	{
 		terminate();
 	}
-	catch(...) {}
+	catch( std::exception& ex ) {}
 }
 // -------------------------------------------------------------------------

src/Communications/Modbus/ModbusTCPSession.cc

@@ -98,7 +98,7 @@ void ModbusTCPSession::run()
 	ioTimeout.start(sessTimeout);
 }
 // -------------------------------------------------------------------------
-bool ModbusTCPSession::isAcive()
+bool ModbusTCPSession::isActive()
 {
 	return io.is_active();
 }

src/Log/LogServer.cc

@@ -27,7 +27,6 @@ using namespace UniSetTypes;
 // -------------------------------------------------------------------------
 LogServer::~LogServer()
 {
-	cerr << myname << "  ~LogServer() " << endl;
 	if( running )
 		terminate();
 }
@@ -84,18 +83,19 @@ void LogServer::terminate()
 	running = false;
 
 	if( evloop )
-	{
-		cerr << myname << ": terminate evloop.." << endl;
 		evloop->terminate(this);
-	}
 }
 // -------------------------------------------------------------------------
 void LogServer::run( const std::string& _addr, ost::tpport_t _port, bool thread )
 {
 	addr = _addr;
 	port = _port;
+
 	if( !running )
 	{
+		if( !thread )
+			running = true;
+
 		mainLoop( thread );
 		running = true;
 	}
@@ -107,6 +107,7 @@ void LogServer::mainLoop( bool thread )
 	{
 		if( elog->is_crit() )
 			elog->crit() << myname << "(LogServer::mainLoopt): ALREADY RUNNING.." << endl;
+
 		return;
 	}
 
@@ -131,6 +132,7 @@ void LogServer::mainLoop( bool thread )
 
 		if( mylog.is_crit() )
 			mylog.crit() << myname << "(LogServer): " << err.str() << endl;
+
 		throw SystemError( err.str() );
 	}
 
@@ -140,7 +142,7 @@ void LogServer::mainLoop( bool thread )
 	io.set<LogServer, &LogServer::ioAccept>(this);
 	io.start(sock->getSocket(), ev::READ);
 
-	// скобки специально чтобы пораньшк выйти из "зоны" видимости
+	// скобки специально чтобы пораньше освободить evloop (выйти из "зоны" видимости)
 	{
 		evloop = DefaultEventLoop::inst();
 		evloop->run( this, thread );
@@ -168,6 +170,7 @@ void LogServer::ioAccept( ev::io& watcher, int revents )
 
 	{
 		uniset_rwmutex_wrlock l(mutSList);
+
 		if( slist.size() >= sessMaxCount )
 		{
 			if( mylog.is_crit() )

src/Log/LogSession.cc

@@ -90,7 +90,7 @@ LogSession::LogSession( int sfd, std::shared_ptr<DebugStream>& _log, timeout_t _
 
 	io.set<LogSession, &LogSession::callback>(this);
 	cmdTimer.set<LogSession, &LogSession::onCmdTimeout>(this);
-	asyncEvent.set<LogSession,&LogSession::event>(this);
+	asyncEvent.set<LogSession, &LogSession::event>(this);
 }
 // -------------------------------------------------------------------------
 void LogSession::logOnEvent( const std::string& s )
@@ -100,6 +100,7 @@ void LogSession::logOnEvent( const std::string& s )
 
 	std::unique_lock<std::mutex> lk(logbuf_mutex);
 	logbuf.emplace(new UTCPCore::Buffer(s));
+
 	if( asyncEvent.is_active() )
 		asyncEvent.send();
 }
@@ -112,7 +113,7 @@ void LogSession::run()
 		mylog.info() << peername << "(run): run session.." << endl;
 
 	io.start(sock->getSocket(), ev::READ);
-	cmdTimer.start( cmdTimeout/1000. );
+	cmdTimer.start( cmdTimeout / 1000. );
 	// asyncEvent.start(); // слать логи начинаем только после обработки команд.. если есть..
 }
 // -------------------------------------------------------------------------
@@ -122,18 +123,22 @@ void LogSession::terminate()
 		mylog.info() << peername << "(terminate)..." << endl;
 
 	cancelled = true;
-	std::unique_lock<std::mutex> lk(logbuf_mutex);
+
+	{
+		std::unique_lock<std::mutex> lk2(io_mutex);
+		io.stop();
+		cmdTimer.stop();
+		asyncEvent.stop();
+		conn.disconnect();
+	}
+
 	{
+		std::unique_lock<std::mutex> lk(logbuf_mutex);
+
 		while( !logbuf.empty() )
 			logbuf.pop();
 	}
 
-	std::unique_lock<std::mutex> lk2(io_mutex);
-	io.stop();
-	cmdTimer.stop();
-	asyncEvent.stop();
-	conn.disconnect();
-
 	sock.reset(); // close..
 	final();
 }
@@ -196,6 +201,7 @@ void LogSession::writeEvent( ev::io& watcher )
 		return;
 
 	auto buffer = logbuf.front();
+
 	if( !buffer )
 		return;
 
@@ -226,6 +232,7 @@ void LogSession::writeEvent( ev::io& watcher )
 	}
 
 	std::unique_lock<std::mutex> lk1(io_mutex);
+
 	if( logbuf.empty() )
 		io.set(ev::READ);
 	else
@@ -268,6 +275,7 @@ void LogSession::readEvent( ev::io& watcher )
 	{
 		if( mylog.is_warn() )
 			mylog.warn() << peername << "(readEvent): BAD MESSAGE..." << endl;
+
 		return;
 	}
 
@@ -356,38 +364,38 @@ void LogSession::cmdProcessing( const string& cmdLogName, const LogServerTypes::
 		{
 			case LogServerTypes::cmdSetLevel:
 				l.log->level( (Debug::type)msg.data );
-			break;
+				break;
 
 			case LogServerTypes::cmdAddLevel:
 				l.log->addLevel( (Debug::type)msg.data );
-			break;
+				break;
 
 			case LogServerTypes::cmdDelLevel:
 				l.log->delLevel( (Debug::type)msg.data );
-			break;
+				break;
 
 			case LogServerTypes::cmdRotate:
 				l.log->onLogFile(true);
-			break;
+				break;
 
 			case LogServerTypes::cmdList: // обработали выше (в начале)
-			break;
+				break;
 
 			case LogServerTypes::cmdOffLogFile:
 				l.log->offLogFile();
-			break;
+				break;
 
 			case LogServerTypes::cmdOnLogFile:
 				l.log->onLogFile();
-			break;
+				break;
 
 			case LogServerTypes::cmdFilterMode:
 				l.log->signal_stream_event().connect( sigc::mem_fun(this, &LogSession::logOnEvent) );
-			break;
+				break;
 
 			default:
 				mylog.warn() << peername << "(run): Unknown command '" << msg.cmd << "'" << endl;
-			break;
+				break;
 		}
 	} // end if for
 

src/ObjectRepository/UniSetActivator.cc

@@ -37,7 +37,6 @@
 #include "Debug.h"
 #include "Configuration.h"
 #include "Mutex.h"
-
 // ------------------------------------------------------------------------------------------
 using namespace UniSetTypes;
 using namespace std;

src/Processes/DefaultEventLoop.cc

@@ -31,8 +31,9 @@ void DefaultEventLoop::run( EventWatcher* s, bool thread )
 
 	{
 		std::unique_lock<std::mutex> lk(m_run_mutex);
+
 		if( !thr )
-			thr = make_shared<std::thread>( [=] { defaultLoop(); } );
+			thr = make_shared<std::thread>( [ = ] { defaultLoop(); } );
 	}
 
 	{
@@ -43,6 +44,7 @@ void DefaultEventLoop::run( EventWatcher* s, bool thread )
 	if( !thread )
 	{
 		std::unique_lock<std::mutex> lk(m_mutex);
+
 		while( !m_notify )
 			m_event.wait(lk);
 
@@ -60,7 +62,8 @@ void DefaultEventLoop::terminate( EventWatcher* s )
 {
 	cerr << "(DefaultEventLoop::defaultLoop): terminate.." << endl;
 	std::unique_lock<std::mutex> lk(m_slist_mutex);
-	for( auto i=slist.begin(); i!=slist.end(); i++ )
+
+	for( auto i = slist.begin(); i != slist.end(); i++ )
 	{
 		if( (*i) == s )
 		{
@@ -77,6 +80,7 @@ void DefaultEventLoop::finish()
 {
 	cerr << "(DefaultEventLoop::fini): TERMINATE EVENT LOOP.." << endl;
 	cancelled = true;
+
 	if( !evloop )
 		return;
 
@@ -84,7 +88,7 @@ void DefaultEventLoop::finish()
 	evloop->break_loop(ev::ALL);
 
 	std::unique_lock<std::mutex> lk(m_mutex);
-	m_event.wait_for(lk, std::chrono::seconds(3), [=]()
+	m_event.wait_for(lk, std::chrono::seconds(1), [ = ]()
 	{
 		return (m_notify == true);
 	} );

tests/test_logserver.cc

@@ -230,11 +230,11 @@ TEST_CASE("MaxSessions", "[LogServer]" )
 
 	{
 		uniset_mutex_lock l(r2_mutex);
-/*
-		// Ищем часть сообщения об ошибке: '(LOG SERVER): Exceeded the limit on the number of sessions = 1'
-		size_t pos = msg2.str().find("Exceeded the limit");
-		REQUIRE( pos != std::string::npos );
-*/
+		/*
+				// Ищем часть сообщения об ошибке: '(LOG SERVER): Exceeded the limit on the number of sessions = 1'
+				size_t pos = msg2.str().find("Exceeded the limit");
+				REQUIRE( pos != std::string::npos );
+		*/
 		// ничего не получили..
 		REQUIRE( msg2.str() == "" );
 	}