(SM): добавил ведение статистики по "недоступным заказчикам"

conf/libuniset2.spec

@@ -504,7 +504,9 @@ mv -f %buildroot%python_sitelibdir_noarch/* %buildroot%python_sitelibdir/%oname
 %exclude %_pkgconfigdir/libUniSet2.pc
         
 # history of current unpublished changes
-# ..
+# - remove 'fastSaveValue'
+# - add suppor HTTP REST API
+# - (SM): add new statistics for consumers
 
 %changelog
 * Mon Oct 24 2016 Pavel Vainerman <pv@altlinux.ru> 2.5-alt20

conf/test.xml

@@ -18,6 +18,7 @@
 		<AutoStartUpTime name="1"/>
 		<DumpStateTime name="10"/>
 		<SleepTickMS name="500"/>
+		<ConsumerMaxAttempts name="3"/>
 		<UniSetDebug levels="" name="ulog"/>
 		<ConfDir name="./"/>
 		<DataDir name="./"/>

extensions/SharedMemory/SharedMemory.h

@@ -312,7 +312,7 @@
 									   limit - количество в ответе.
 
 	/consumers                       - получить список заказчиков по каждому датчику
-
+	/lost                            - получить список заказчиков с которыми терялась связь (и они удалялись из списка)
 */
 class SharedMemory:
 	public IONotifyController

extensions/SharedMemory/start_fg_testlost.sh

@@ -5,7 +5,7 @@ export LD_LIBRARY_PATH="../../lib/.libs;../lib/.libs"
 ulimit -Sc 10000000000
 
 ./uniset2-start.sh -f ./uniset2-smemory --smemory-id SharedMemory  \
---confile test-lost.xml --datfile test-lost.xml --ulog-add-levels system,level1 \
+--confile test-lost.xml --datfile test-lost.xml --ulog-add-levels crit,warn,system,level1 \
 --sm-log-add-levels any $* --sm-run-logserver --activate-timeout 320000
 
 #--pulsar-id DO_C --pulsar-iotype DO --pulsar-msec 100

extensions/tests/SMemoryTest/LostTestProc.cc

@@ -137,7 +137,8 @@ void LostTestProc::timerInfo( const TimerMessage* tm )
 
 				if(ui->getValue(s.first) != (s.second+1) )
 				{
-					cerr << myname << "(check): SAVE TO SM ERROR!! smValue=" << smValue  << endl;
+					cerr << myname << "(check): SAVE TO SM ERROR!! sid=" << s.first
+						 << " value=" << smValue << " != " << (s.second+1) << endl;
 					UniSetTypes::SimpleInfo_var i = getInfo();
 					cerr << i->info << endl;
 					std::abort();

extensions/tests/SMemoryTest/start_fg_losttest.sh

@@ -2,7 +2,7 @@
 
 START=uniset2-start.sh
 
-LOGS="crit,warn,info"
+LOGLEVEL="crit,warn,info"
 
 ${START} -f ./sm-lostmessage-test --confile ./test-lost.xml \
 --numproc 3 \

include/IOController.h

@@ -309,6 +309,8 @@ class IOController:
 			long d_off_value = { 0 }; /*!< блокирующее значение */
 			std::shared_ptr<USensorInfo> d_usi; // shared_ptr на датчик от которого зависит этот.
 
+			size_t nchanges = { 0 }; // количество изменений датчика
+
 			// функция обработки информации об изменении состояния датчика, от которого зависит данный
 			void checkDepend( std::shared_ptr<USensorInfo>& d_usi, IOController* );
 

include/IONotifyController.h

@@ -160,6 +160,12 @@ class IONotifyController:
 
 		// --------------------------------------------
 
+		// http API
+		virtual nlohmann::json httpHelp( const Poco::URI::QueryParameters& p ) override;
+		nlohmann::json httpRequest( const string& req, const Poco::URI::QueryParameters& p );
+
+
+		// --------------------------------------------
 		/*! Информация о заказчике */
 		struct ConsumerInfoExt:
 			public UniSetTypes::ConsumerInfo
@@ -172,6 +178,7 @@ class IONotifyController:
 			UniSetObject_i_var ref;
 			size_t attempt;
 			size_t lostEvents = { 0 }; // количество потерянных сообщений (не смогли послать)
+			size_t smCount = { 0 }; // количество посланных SensorMessage
 
 			ConsumerInfoExt( const ConsumerInfoExt& ) = default;
 			ConsumerInfoExt& operator=( const ConsumerInfoExt& ) = default;
@@ -270,11 +277,6 @@ class IONotifyController:
 		/*! словарь: аналоговый датчик --> список порогов по нему */
 		typedef std::unordered_map<UniSetTypes::ObjectId, ThresholdsListInfo> AskThresholdMap;
 
-
-		// http API
-		virtual nlohmann::json httpHelp( const Poco::URI::QueryParameters& p ) override;
-		nlohmann::json httpRequest( const string& req, const Poco::URI::QueryParameters& p );
-
 	protected:
 		IONotifyController();
 		virtual bool activateObject() override;
@@ -320,7 +322,8 @@ class IONotifyController:
 		};
 
 		// http api
-		virtual nlohmann::json request_consumers( const std::string& req, const Poco::URI::QueryParameters& p );
+		nlohmann::json request_consumers( const std::string& req, const Poco::URI::QueryParameters& p );
+		nlohmann::json request_lost( const string& req, const Poco::URI::QueryParameters& p );
 
 	private:
 		friend class NCRestorer;
@@ -346,10 +349,27 @@ class IONotifyController:
 		/*! замок для блокирования совместного доступа к cписку потребителей пороговых датчиков */
 		UniSetTypes::uniset_rwmutex trshMutex;
 
-		int maxAttemtps; /*! timeout for consumer */
-
 		sigc::connection conInit;
 		sigc::connection conUndef;
+
+		int maxAttemtps; /*! timeout for consumer */
+
+		std::mutex lostConsumersMutex;
+
+		struct LostConsumerInfo
+		{
+			size_t count = { 0 }; // количество "пропаданий"
+			bool lost = { false }; // флаг означающий что "заказчик пропал"
+			// lost нужен чтобы в count не увеличивать, на send() по каждому датчику, если заказчик заказывал больше одного датчика)
+			// флаг сбрасывается при перезаказе датчика..
+		};
+
+		/*! map для хранения информации о заказчиках с которыми была потеряна связь
+		 * и которые были удалены из списка заказчиков
+		 * size_t - количество раз
+		 * ObjectId - id заказчика
+		 */
+		std::unordered_map<UniSetTypes::ObjectId, LostConsumerInfo> lostConsumers;
 };
 // --------------------------------------------------------------------------
 #endif

src/Interfaces/UInterface.cc

@@ -937,6 +937,11 @@ throw(UniSetTypes::ResolveNameError, UniSetTypes::TimeOut )
 		// uwarn << "UI(resolve): CORBA::SystemException" << endl;
 		throw UniSetTypes::TimeOut();
 	}
+	catch( std::exception& ex )
+	{
+		ucrit << "UI(resolve): myID=" << myid <<  ": resolve id=" << rid << "@" << node
+			  << " catch " << ex.what() << endl;
+	}
 
 	throw UniSetTypes::ResolveNameError();
 }

src/Processes/IOController.cc

@@ -291,7 +291,10 @@ long IOController::localSetValue( std::shared_ptr<USensorInfo>& usi,
 		bool blocked = ( usi->blocked || usi->undefined );
 		changed = ( usi->real_value != value );
 
-		// если поменялось состояние блокировки
+		// Смотрим поменялось ли состояние блокировки.
+		// т.е. смотрим записано ли у нас уже value = d_off_value и флаг блокировки
+		// т.к. если blocked=true то должно быть usi->value = usi->d_off_value
+		// если флаг снимется, то значит должны "восстанавливать" значение из real_value
 		blockChanged = ( blocked != (usi->value == usi->d_off_value ) );
 
 		if( changed || blockChanged )
@@ -309,6 +312,8 @@ long IOController::localSetValue( std::shared_ptr<USensorInfo>& usi,
 			usi->value = (blocked ? usi->d_off_value : value);
 			retValue = usi->value;
 
+			usi->nchanges++; // статистика
+
 			// запоминаем время изменения
 			try
 			{
@@ -950,6 +955,7 @@ void IOController::getSensorInfo( nlohmann::json& jdata, std::shared_ptr<USensor
 	jsens["type"] = UniSetTypes::iotype2str(s->type);
 	jsens["default_val"] = s->default_val;
 	jsens["dbignore"] = s->dbignore;
+	jsens["nchanges"] = s->nchanges;
 	jsens["calibration"] = {
 		{ "cmin",s->ci.minCal},
 		{ "cmax",s->ci.maxCal},
@@ -958,7 +964,6 @@ void IOController::getSensorInfo( nlohmann::json& jdata, std::shared_ptr<USensor
 		{ "precision",s->ci.precision}
 	};
 
-
 	//	::CORBA::Boolean undefined;
 	//	::CORBA::Boolean blocked;
 	//	::CORBA::Long priority;

src/Processes/IONotifyController.cc

@@ -40,7 +40,7 @@ using namespace std;
 IONotifyController::IONotifyController():
 	askIOMutex("askIOMutex"),
 	trshMutex("trshMutex"),
-	maxAttemtps(uniset_conf()->getPIntField("ConsumerMaxAttempts", 5))
+	maxAttemtps(uniset_conf()->getPIntField("ConsumerMaxAttempts", 15))
 {
 
 }
@@ -50,7 +50,7 @@ IONotifyController::IONotifyController(const string& name, const string& section
 	restorer(d),
 	askIOMutex(name + "askIOMutex"),
 	trshMutex(name + "trshMutex"),
-	maxAttemtps(uniset_conf()->getPIntField("ConsumerMaxAttempts", 5))
+	maxAttemtps(uniset_conf()->getPIntField("ConsumerMaxAttempts", 15))
 {
 	conUndef = signal_change_undefined_state().connect(sigc::mem_fun(*this, &IONotifyController::onChangeUndefinedState));
 	conInit = signal_init().connect(sigc::mem_fun(*this, &IONotifyController::initItem));
@@ -61,7 +61,7 @@ IONotifyController::IONotifyController( ObjectId id, std::shared_ptr<NCRestorer>
 	restorer(d),
 	askIOMutex(string(uniset_conf()->oind->getMapName(id)) + "_askIOMutex"),
 	trshMutex(string(uniset_conf()->oind->getMapName(id)) + "_trshMutex"),
-	maxAttemtps(uniset_conf()->getPIntField("ConsumerMaxAttempts", 5))
+	maxAttemtps(uniset_conf()->getPIntField("ConsumerMaxAttempts", 15))
 {
 	conUndef = signal_change_undefined_state().connect(sigc::mem_fun(*this, &IONotifyController::onChangeUndefinedState));
 	conInit = signal_init().connect(sigc::mem_fun(*this, &IONotifyController::initItem));
@@ -81,12 +81,30 @@ SimpleInfo* IONotifyController::getInfo( ::CORBA::Long userparam )
 
 	inf << i->info << endl;
 
+	auto oind = uniset_conf()->oind;
+
+	{
+		std::lock_guard<std::mutex> lock(lostConsumersMutex);
+		if( lostConsumers.size() > 0 )
+		{
+			inf << "-------------------------- lost consumers list [maxAttemtps=" << maxAttemtps << "] ------------------" << endl;
+			for( const auto& l: lostConsumers )
+			{
+				inf << "        " << "(" << setw(6) << l.first << ")"
+					<< setw(35) << std::left << ORepHelpers::getShortName(oind->getMapName(l.first))
+					<< " lostCount=" << l.second.count
+					<< endl;
+			}
+		}
+		inf << "----------------------------------------------------------------------------------" << endl;
+	}
+
 	if( userparam == 1 || userparam == 2 )
 	{
 		inf << "------------------------------- consumers list ------------------------------" << endl;
-		{
-			auto oind = uniset_conf()->oind;
+		inf << "[userparam=" << userparam << "]" << endl;
 
+		{
 			uniset_rwmutex_rlock lock(askIOMutex);
 
 			for( auto && a : askIOList )
@@ -99,22 +117,23 @@ SimpleInfo* IONotifyController::getInfo( ::CORBA::Long userparam )
 				if( i.clst.empty() )
 					continue;
 
-				// если надо выводить только тех, у кого есть "потери"(lostEvent>0)
-				// то надо сперва смотреть список, а потом выводить
+				// Т.к. сперва выводится имя датчика, а только потом его заказчики
+				// то если надо выводить только тех, у кого есть "потери"(lostEvent>0)
+				// предварительно смотрим список есть ли там хоть один с "потерями", а потом уже выводим
 				if( userparam == 2 )
 				{
-					bool ok = false;
+					bool lost = false;
 
 					for( const auto& c : i.clst )
 					{
 						if( c.lostEvents > 0 )
 						{
-							ok = true;
+							lost = true;
 							break;
 						}
 					}
 
-					if( !ok )
+					if( !lost )
 						continue;
 
 					// выводим тех у кого lostEvent>0
@@ -126,7 +145,11 @@ SimpleInfo* IONotifyController::getInfo( ::CORBA::Long userparam )
 						{
 							inf << "        " << "(" << setw(6) << c.id << ")"
 								<< setw(35) << ORepHelpers::getShortName(oind->getMapName(c.id))
-								<< " [lostEvents=" << c.lostEvents << " attempt=" << c.attempt << "]"
+								<< " ["
+								<< " lostEvents=" << c.lostEvents
+								<< " attempt=" << c.attempt
+								<< " smCount=" << c.smCount
+								<< "]"
 								<< endl;
 						}
 					}
@@ -139,7 +162,11 @@ SimpleInfo* IONotifyController::getInfo( ::CORBA::Long userparam )
 					{
 						inf << "        " << "(" << setw(6) << c.id << ")"
 							<< setw(35) << ORepHelpers::getShortName(oind->getMapName(c.id))
-							<< " [lostEvents=" << c.lostEvents << " attempt=" << c.attempt << "]"
+							<< " ["
+							<< " lostEvents=" << c.lostEvents
+							<< " attempt=" << c.attempt
+							<< " smCount=" << c.smCount
+							<< "]"
 							<< endl;
 					}
 				}
@@ -168,11 +195,23 @@ bool IONotifyController::addConsumer( ConsumerListInfo& lst, const ConsumerInfo&
 {
 	uniset_rwmutex_wrlock l(lst.mut);
 
-	for( const auto& it :  lst.clst )
+	for( auto&& it :  lst.clst )
 	{
 		if( it.id == ci.id && it.node == ci.node )
+		{
+			// при перезаказе датчиков количество неудачных попыток послать сообщение
+			// считаем что "заказчик" опять на связи
+			it.attempt = maxAttemtps;
+
+			// выставляем флаг, что заказчик опять "на связи"
+			std::lock_guard<std::mutex> lock(lostConsumersMutex);
+			auto c = lostConsumers.find(ci.id);
+			if( c!= lostConsumers.end() )
+				c->second.lost = false;
+
 			return false;
 		}
+	}
 
 	ConsumerInfoExt cinf(ci, 0, maxAttemtps);
 
@@ -185,6 +224,13 @@ bool IONotifyController::addConsumer( ConsumerListInfo& lst, const ConsumerInfo&
 	catch(...) {}
 
 	lst.clst.emplace_front( std::move(cinf) );
+
+	// выставляем флаг, что клиент опять "на связи"
+	std::lock_guard<std::mutex> lock(lostConsumersMutex);
+	auto c = lostConsumers.find(ci.id);
+	if( c!= lostConsumers.end() )
+		c->second.lost = false;
+
 	return true;
 }
 // ------------------------------------------------------------------------------------------
@@ -367,6 +413,7 @@ void IONotifyController::ask( AskMap& askLst, const UniSetTypes::ObjectId sid,
 long IONotifyController::localSetValue( std::shared_ptr<IOController::USensorInfo>& usi,
 										CORBA::Long value, UniSetTypes::ObjectId sup_id )
 {
+	// оптимизация:
 	// if( !usi ) - не проверяем, т.к. считаем что это внутренние функции и несуществующий указатель передать не могут
 
 	CORBA::Long prevValue = value;
@@ -375,17 +422,17 @@ long IONotifyController::localSetValue( std::shared_ptr<IOController::USensorInf
 		prevValue = usi->value;
 	}
 
-	CORBA::Long realValue = IOController::localSetValue(usi, value, sup_id);
+	CORBA::Long curValue = IOController::localSetValue(usi, value, sup_id);
 
 	// Рассылаем уведомления только в случае изменения значения
 	// --------
-	if( prevValue == realValue )
-		return realValue;
+	if( prevValue == curValue )
+		return curValue;
 
 	SensorMessage sm(1); // <-- вызываем dummy конструктор т.к. потом все поля всё-равно сами инициализируем
 	sm.id           = usi->si.id;
 	sm.node         = usi->si.node; // uniset_conf()->getLocalNode();
-	sm.value        = realValue;
+	sm.value        = curValue;
 	sm.priority     = (Message::Priority)usi->priority;
 	sm.supplier     = sup_id; // owner_id
 	sm.sensor_type  = usi->type;
@@ -427,7 +474,7 @@ long IONotifyController::localSetValue( std::shared_ptr<IOController::USensorInf
 	}
 	catch(...) {}
 
-	return realValue;
+	return curValue;
 }
 // -----------------------------------------------------------------------------------------
 /*!
@@ -455,23 +502,25 @@ void IONotifyController::send( ConsumerListInfo& lst, const UniSetTypes::SensorM
 
 				tmsg.consumer = li->id;
 				li->ref->push( tmsg );
+				li->smCount++;
 				li->attempt = maxAttemtps; // reinit attempts
 				break;
 			}
 			catch( const CORBA::SystemException& ex )
 			{
-				uwarn << myname << "(IONotifyController::send): "
+				uwarn << myname << "(IONotifyController::send): attempt=" << li->attempt << " "
 					  << uniset_conf()->oind->getNameById(li->id) << "@" << li->node << " (CORBA::SystemException): "
 					  << ex.NP_minorString() << endl;
 			}
 			catch( const std::exception& ex )
 			{
-				uwarn << myname << "(IONotifyController::send): " << ex.what()
+				uwarn << myname << "(IONotifyController::send): attempt=" << li->attempt << " "
+					  << ex.what()
 					  << " for " << uniset_conf()->oind->getNameById(li->id) << "@" << li->node << endl;
 			}
 			catch(...)
 			{
-				ucrit << myname << "(IONotifyController::send): "
+				ucrit << myname << "(IONotifyController::send): attempt=" << li->attempt << " "
 					  << uniset_conf()->oind->getNameById(li->id) << "@" << li->node
 					  << " catch..." << endl;
 			}
@@ -480,15 +529,40 @@ void IONotifyController::send( ConsumerListInfo& lst, const UniSetTypes::SensorM
 			if( i > 0 )
 				li->lostEvents++;
 
+			try
+			{
 				if( maxAttemtps > 0 && --(li->attempt) <= 0 )
 				{
+					uwarn << myname << "(IONotifyController::send): ERASE FROM CONSUMERS:  "
+						  << uniset_conf()->oind->getNameById(li->id) << "@" << li->node << endl;
+
+					{
+						std::lock_guard<std::mutex> lock(lostConsumersMutex);
+						auto& c = lostConsumers[li->id];
+						// если уже выставлен флаг что "заказчик" пропал, то не надо увеличивать "счётчик"
+						// видимо мы уже зафиксировали его пропажу на другом датчике...
+						if( !c.lost )
+						{
+							c.count += 1;
+							c.lost = true;
+						}
+					}
+
 					li = lst.clst.erase(li);
 					--li;
 					break;
 				}
 
+
 				li->ref = UniSetObject_i::_nil();
 			}
+			catch( const std::exception& ex )
+			{
+				uwarn << myname << "(IONotifyController::send): UniSetObject_i::_nil() "
+					  << ex.what()
+					  << " for " << uniset_conf()->oind->getNameById(li->id) << "@" << li->node << endl;
+			}
+		}
 	}
 }
 // --------------------------------------------------------------------------------------------------------------
@@ -1089,6 +1163,7 @@ nlohmann::json IONotifyController::httpHelp(const Poco::URI::QueryParameters& p)
 {
 	nlohmann::json jdata = IOController::httpHelp(p);
 	jdata[myname]["help"]["consumers"]["desc"] = "get consumers list";
+	jdata[myname]["help"]["lost"]["desc"] = "get lost consumers list";
 	return std::move(jdata);
 }
 // -----------------------------------------------------------------------------
@@ -1097,6 +1172,9 @@ nlohmann::json IONotifyController::httpRequest( const string& req, const Poco::U
 	if( req == "consumers" )
 		return request_consumers(req,p);
 
+	if( req == "lost" )
+		return request_lost(req,p);
+
 	return IOController::httpRequest(req,p);
 }
 // -----------------------------------------------------------------------------
@@ -1136,9 +1214,35 @@ nlohmann::json IONotifyController::request_consumers(const string& req, const Po
 			jconsinfo["name"] = ORepHelpers::getShortName(oind->getMapName(c.id));
 			jconsinfo["lostEvents"] = c.lostEvents;
 			jconsinfo["attempt"] = c.attempt;
+			jconsinfo["smCount"] = c.smCount;
 		}
 	}
 
 	return std::move(json);
 }
 // -----------------------------------------------------------------------------
+nlohmann::json IONotifyController::request_lost( const string& req, const Poco::URI::QueryParameters& p )
+{
+	//! \todo Не реализовано
+	nlohmann::json json;
+
+	auto& jdata = json[myname]["lost consumers"];
+
+	auto oind = uniset_conf()->oind;
+
+	std::lock_guard<std::mutex> lock(lostConsumersMutex);
+
+	for( const auto& c: lostConsumers )
+	{
+		string cid( std::to_string(c.first) );
+		auto& jcons = jdata[cid];
+
+		jcons["id"] = c.first;
+		jcons["name"] = ORepHelpers::getShortName(oind->getMapName(c.first));
+		jcons["lostCount"] = c.second.count;
+		jcons["lost"] = c.second.lost;
+	}
+
+	return std::move(json);
+}
+// -----------------------------------------------------------------------------