(UNet): небольшая оптимизация: сделал mutex более гранулярным

extensions/SharedMemoryPlus/start_fg.sh

@@ -47,7 +47,7 @@ ${START} -f ./uniset2-smemory-plus --smemory-id SharedMemory  --confile test.xml
      --ulog-add-levels system \
      --add-unet \
 	--unet-name UNetExchange --unet-run-logserver \
-    --unet-filter-field rs --unet-filter-value 2 --unet-maxdifferense 40 --unet-sendpause 1000     
+    --unet-filter-field rs --unet-filter-value 2 --unet-maxdifferense 40 --unet-sendpause 1000 \
      $*
 #	 --add-rtu \
 #	 --rs-dev /dev/cbsideA1 \

extensions/UNetUDP/UDPPacket.h

@@ -92,6 +92,12 @@ namespace UniSetUDP
 	{
 		UDPMessage();
 
+		UDPMessage(UDPMessage&& m) = default;
+		UDPMessage& operator=(UDPMessage&&) = default;
+
+		UDPMessage( const UDPMessage& m ) = default;
+		UDPMessage& operator=(const UDPMessage&) = default;
+
 		explicit UDPMessage( UDPPacket& p );
 		size_t transport_msg( UDPPacket& p );
 		static size_t getMessage( UDPMessage& m, UDPPacket& p );

extensions/UNetUDP/UNetSender.cc

@@ -40,7 +40,6 @@ UNetSender::UNetSender(const std::string& _host, const int _port, const std::sha
 	packsendpause(5),
 	activated(false),
 	items(100),
-	maxItem(0),
 	packetnum(1),
 	lastcrc(0),
 	maxAData(maxACount),
@@ -74,10 +73,13 @@ UNetSender::UNetSender(const std::string& _host, const int _port, const std::sha
 	mypacks[0].resize(1);
 	packs_anum[0] = 0;
 	packs_dnum[0] = 0;
-	UniSetUDP::UDPMessage& mypack(mypacks[0][0]);
+	auto& mypack(mypacks[0][0]);
 	// выставляем поля, которые не меняются
-	mypack.nodeID = uniset_conf()->getLocalNode();
-	mypack.procID = shm->ID();
+	{
+		uniset_rwmutex_wrlock l(mypack.mut);
+		mypack.msg.nodeID = uniset_conf()->getLocalNode();
+		mypack.msg.procID = shm->ID();
+	}
 
 	// -------------------------------
 	if( shm->isLocalwork() )
@@ -164,13 +166,14 @@ void UNetSender::updateSensor( UniSetTypes::ObjectId id, long value )
 // -----------------------------------------------------------------------------
 void UNetSender::updateItem( UItem& it, long value )
 {
-	UniSetTypes::uniset_rwmutex_wrlock l(pack_mutex);
 	auto& pk = mypacks[it.pack_sendfactor];
-	UniSetUDP::UDPMessage& mypack(pk[it.pack_num]);
+
+	auto& mypack(pk[it.pack_num]);
+	UniSetTypes::uniset_rwmutex_wrlock l(mypack.mut);
 	if( it.iotype == UniversalIO::DI || it.iotype == UniversalIO::DO )
-		mypack.setDData(it.pack_ind, value);
+		mypack.msg.setDData(it.pack_ind, value);
 	else if( it.iotype == UniversalIO::AI || it.iotype == UniversalIO::AO )
-		mypack.setAData(it.pack_ind, value);
+		mypack.msg.setAData(it.pack_ind, value);
 }
 // -----------------------------------------------------------------------------
 void UNetSender::setCheckConnectionPause( int msec )
@@ -262,17 +265,17 @@ void UNetSender::send()
 // -----------------------------------------------------------------------------
 // #define UNETUDP_DISABLE_OPTIMIZATION_N1
 
-void UNetSender::real_send( UniSetUDP::UDPMessage& mypack )
+void UNetSender::real_send( PackMessage& mypack )
 {
-	UniSetTypes::uniset_rwmutex_rlock l(pack_mutex);
+	UniSetTypes::uniset_rwmutex_rlock l(mypack.mut);
 #ifdef UNETUDP_DISABLE_OPTIMIZATION_N1
-	mypack.num = packetnum++;
+	mypack.msg.num = packetnum++;
 #else
-	uint16_t crc = mypack.getDataCRC();
+	uint16_t crc = mypack.msg.getDataCRC();
 
 	if( crc != lastcrc )
 	{
-		mypack.num = packetnum++;
+		mypack.msg.num = packetnum++;
 		lastcrc = crc;
 	}
 
@@ -286,7 +289,7 @@ void UNetSender::real_send( UniSetUDP::UDPMessage& mypack )
 	if( !udp || !udp->poll(writeTimeout * 1000, Poco::Net::Socket::SELECT_WRITE) )
 		return;
 
-	mypack.transport_msg(s_msg);
+	mypack.msg.transport_msg(s_msg);
 
 	try
 	{
@@ -380,7 +383,7 @@ bool UNetSender::initItem( UniXML::iterator& it )
 
 	int priority = it.getPIntProp(prefix + "_sendfactor", 0);
 
-	auto pk = mypacks[priority];
+	auto& pk = mypacks[priority];
 
 	UItem p;
 	p.iotype = UniSetTypes::getIOType(it.getProp("iotype"));
@@ -401,9 +404,11 @@ bool UNetSender::initItem( UniXML::iterator& it )
 		if( pk.size() <= dnum )
 			pk.resize(dnum + 1);
 
-		UniSetUDP::UDPMessage& mypack(pk[dnum]);
+		auto& mypack(pk[dnum]);
+
+		uniset_rwmutex_wrlock l(mypack.mut);
 
-		p.pack_ind = mypack.addDData(sid, 0);
+		p.pack_ind = mypack.msg.addDData(sid, 0);
 
 		if( p.pack_ind >= maxDData )
 		{
@@ -412,10 +417,11 @@ bool UNetSender::initItem( UniXML::iterator& it )
 			if( dnum >= pk.size() )
 				pk.resize(dnum + 1);
 
-			UniSetUDP::UDPMessage& mypack( pk[dnum] );
-			p.pack_ind = mypack.addDData(sid, 0);
-			mypack.nodeID = uniset_conf()->getLocalNode();
-			mypack.procID = shm->ID();
+			auto& mypack2( pk[dnum] );
+			uniset_rwmutex_wrlock l2(mypack2.mut);
+			p.pack_ind = mypack2.msg.addDData(sid, 0);
+			mypack2.msg.nodeID = uniset_conf()->getLocalNode();
+			mypack2.msg.procID = shm->ID();
 		}
 
 		p.pack_num = dnum;
@@ -438,9 +444,9 @@ bool UNetSender::initItem( UniXML::iterator& it )
 		if( pk.size() <= anum )
 			pk.resize(anum + 1);
 
-		UniSetUDP::UDPMessage& mypack(pk[anum]);
-
-		p.pack_ind = mypack.addAData(sid, 0);
+		auto& mypack(pk[anum]);
+		uniset_rwmutex_wrlock l(mypack.mut);
+		p.pack_ind = mypack.msg.addAData(sid, 0);
 
 		if( p.pack_ind >= maxAData )
 		{
@@ -449,10 +455,11 @@ bool UNetSender::initItem( UniXML::iterator& it )
 			if( anum >= pk.size() )
 				pk.resize(anum + 1);
 
-			UniSetUDP::UDPMessage& mypack(pk[anum]);
-			p.pack_ind = mypack.addAData(sid, 0);
-			mypack.nodeID = uniset_conf()->getLocalNode();
-			mypack.procID = shm->ID();
+			auto& mypack2(pk[anum]);
+			uniset_rwmutex_wrlock l2(mypack2.mut);
+			p.pack_ind = mypack2.msg.addAData(sid, 0);
+			mypack2.msg.nodeID = uniset_conf()->getLocalNode();
+			mypack2.msg.procID = shm->ID();
 		}
 
 		p.pack_num = anum;
@@ -469,12 +476,17 @@ bool UNetSender::initItem( UniXML::iterator& it )
 		}
 	}
 
-	mypacks[priority] = pk;
-
-	items[p.id] = p;
-	maxItem++;
-
 	unetinfo << myname << "(initItem): add " << p << endl;
+	auto i = items.find(p.id);
+	if( i != items.end() )
+	{
+		unetcrit << myname
+				 << "(readItem): Sensor (" << p.id << ")" << sname << " ALREADY ADDED!!  ABORT!" << endl;
+		raise(SIGTERM);
+		return false;
+	}
+
+	items.emplace(p.id, std::move(p));
 	return true;
 }
 
@@ -511,7 +523,7 @@ const std::string UNetSender::getShortInfo() const
 	s << setw(15) << std::right << getAddress() << ":" << std::left << setw(6) << getPort()
 	  << " lastpacknum=" << packetnum
 	  << " lastcrc=" << setw(6) << lastcrc
-	  << " items=" << maxItem << " maxAData=" << getADataSize() << " maxDData=" << getDDataSize()
+	  << " items=" << items.size() << " maxAData=" << getADataSize() << " maxDData=" << getDDataSize()
 	  << endl
 	  << "\t   packs([sendfactor]=num): "
 	  << endl;
@@ -522,8 +534,9 @@ const std::string UNetSender::getShortInfo() const
 		size_t n=0;
 		for( const auto& p: i->second )
 		{
-			s << "        \t\t[" << (n++) << "]=" << p.sizeOf() << " bytes"
-			  << " ( numA=" << setw(5) << p.asize() << " numD=" << setw(5) << p.dsize() << ")"
+			//uniset_rwmutex_rlock l(p->mut);
+			s << "        \t\t[" << (n++) << "]=" << p.msg.sizeOf() << " bytes"
+			  << " ( numA=" << setw(5) << p.msg.asize() << " numD=" << setw(5) << p.msg.dsize() << ")"
 			  << endl;
 		}
 	}

extensions/UNetUDP/UNetSender.h

@@ -41,12 +41,13 @@
  *    Причем так как дискретные и аналоговые датчики обрабатываются отдельно (но пересылаются в одном пакете),
  *    то датчики, которые первые переполнятся приводят к тому, что создаётся новый пакет и они добавляются в него,
  *    в свою очередь остальные продолжают "добивать" предыдущий пакет.
- *    В свою очередь в initItem() каждому UItem в dlist кроме pack_ind присваивается еще и номер пакета pack_num, который гарантировано соответствует
+ *    В initItem() каждому UItem в dlist кроме pack_ind присваивается еще и номер пакета pack_num, который гарантировано соответствует
  *    существующему пакету, поэтому в дальнейшем при использовании pack_num в качестве ключа в mypacks мы не проверяем пакет на существование.
  *
- *    ОПТИМИЗАЦИЯ N1: Для оптимизации обработки посылаемых пакетов (на стороне UNetReceiver) сделана следующая логика:
+ *    ОПТИМИЗАЦИЯ N1: Для оптимизации обработки посылаемых пакетов (на стороне UNetSender) сделана следующая логика:
  *                  Номер очередного посылаемого пакета меняется (увеличивается) только, если изменились данные с момента
-                    последней посылки. Для этого по данным каждый раз производится расчёт UNetUDP::makeCRC() и сравнивается с последним..
+					последней посылки. Для этого по данным каждый раз производится расчёт UNetUDP::makeCRC() и сравнивается с последним.
+					На стороне UNetReceiver пакаеты с повторными номерами (т.е. уже обработанные) - откидываются.
  *
  *
  * Создание соединения
@@ -87,7 +88,6 @@ class UNetSender
 			size_t pack_num;
 			size_t pack_ind;
 			sendfactor_t pack_sendfactor = { 0 };
-
 			friend std::ostream& operator<<( std::ostream& os, UItem& p );
 		};
 
@@ -99,7 +99,19 @@ class UNetSender
 		void stop();
 
 		void send();
-		void real_send(UniSetUDP::UDPMessage& mypack);
+
+		struct PackMessage
+		{
+			PackMessage( UniSetUDP::UDPMessage&& m ):msg(std::move(m)){}
+			PackMessage( const UniSetUDP::UDPMessage& m ) = delete;
+
+			PackMessage(){}
+
+			UniSetUDP::UDPMessage msg;
+			UniSetTypes::uniset_rwmutex mut;
+		};
+
+		void real_send( PackMessage& mypack );
 
 		/*! (принудительно) обновить все данные (из SM) */
 		void updateFromSM();
@@ -184,15 +196,14 @@ class UNetSender
 		std::atomic_bool activated = { false };
 		PassiveTimer ptCheckConnection;
 
-		UniSetTypes::uniset_rwmutex pack_mutex;
-
-		typedef std::unordered_map<sendfactor_t, std::vector<UniSetUDP::UDPMessage>> Packs;
+		typedef std::unordered_map<sendfactor_t, std::vector<PackMessage>> Packs;
 
+		// mypacks заполняется в начале и дальше с ним происходит только чтение
+		// поэтому mutex-ом его не защищаем
 		Packs mypacks;
 		std::unordered_map<sendfactor_t, size_t> packs_anum;
 		std::unordered_map<sendfactor_t, size_t> packs_dnum;
 		UItemMap items;
-		size_t maxItem = { 0 };
 		size_t packetnum = { 1 }; /*!< номер очередного посылаемого пакета */
 		uint16_t lastcrc = { 0 };
 		UniSetUDP::UDPPacket s_msg;

include/ThreadCreator.h

@@ -109,6 +109,8 @@ class ThreadCreator:
 		void stop();
 		void start();
 
+		void sleep( long milliseconds );
+
 		inline bool isRunning()
 		{
 			return thr.isRunning();
@@ -200,6 +202,12 @@ void ThreadCreator<ThreadMaster>::start()
 }
 //----------------------------------------------------------------------------------------
 template <class ThreadMaster>
+void ThreadCreator<ThreadMaster>::sleep( long milliseconds )
+{
+	thr.sleep(milliseconds);
+}
+//----------------------------------------------------------------------------------------
+template <class ThreadMaster>
 ThreadCreator<ThreadMaster>::ThreadCreator():
 	m(0),
 	act(0),