(unet): receiver refactoring: remove "thread strategy",

extensions/UNetUDP/UDPPacket.cc

@@ -23,6 +23,7 @@
 // myvar = LE_TO_H(myvar)
 // -------------------------------------------------------------------------
 #if __BYTE_ORDER == __LITTLE_ENDIAN
+static bool HostIsBigEndian = false;
 #define LE_TO_H(x) {}
 #elif INTPTR_MAX == INT64_MAX
 #define LE_TO_H(x) x = le64toh(x)
@@ -33,6 +34,7 @@
 #endif
 
 #if __BYTE_ORDER == __BIG_ENDIAN
+static bool HostIsBigEndian = true;
 #define BE_TO_H(x) {}
 #elif INTPTR_MAX == INT64_MAX
 #define BE_TO_H(x) x = be64toh(x)
@@ -253,7 +255,7 @@ namespace uniset
 	// -----------------------------------------------------------------------------
 	size_t UDPMessage::transport_msg( UDPPacket& p ) const noexcept
 	{
-		memset(&p, 0, sizeof(UDPPacket));
+		p = UDPPacket{}; // reset data
 
 		size_t i = 0;
 		memcpy(&(p.data[i]), this, sizeof(UDPHeader));
@@ -311,7 +313,8 @@ namespace uniset
 	// -----------------------------------------------------------------------------
 	size_t UDPMessage::getMessage( UDPMessage& m, UDPPacket& p ) noexcept
 	{
-		memset(&m, 0, sizeof(m));
+		// reset data
+		m = UDPMessage{};
 
 		size_t i = 0;
 		memcpy(&m, &(p.data[i]), sizeof(UDPHeader));
@@ -320,7 +323,7 @@ namespace uniset
 		// byte order from packet
 		u_char be_order = m._be_order;
 
-		if( be_order )
+		if( be_order && !HostIsBigEndian )
 		{
 			BE_TO_H(m.magic);
 			BE_TO_H(m.num);
@@ -329,7 +332,7 @@ namespace uniset
 			BE_TO_H(m.dcount);
 			BE_TO_H(m.acount);
 		}
-		else
+		else if( !be_order && HostIsBigEndian )
 		{
 			LE_TO_H(m.magic);
 			LE_TO_H(m.num);
@@ -385,6 +388,8 @@ namespace uniset
 
 		// CONVERT DATA TO HOST BYTE ORDER
 		// -------------------------------
+		if( (be_order && !HostIsBigEndian) || (!be_order && HostIsBigEndian) )
+		{
 			for( size_t n = 0; n < m.acount; n++ )
 			{
 				if( be_order )
@@ -410,6 +415,7 @@ namespace uniset
 					LE_TO_H(m.d_id[n]);
 				}
 			}
+		}
 
 		return i + sz;
 	}

extensions/UNetUDP/UNetExchange.cc

@@ -84,9 +84,9 @@ UNetExchange::UNetExchange(uniset::ObjectId objId, uniset::ObjectId shmId, const
 	int lostTimeout = conf->getArgPInt("--" + prefix + "-lost-timeout", it.getProp("lostTimeout"), 2 * updatepause);
 	steptime = conf->getArgPInt("--" + prefix + "-steptime", it.getProp("steptime"), 1000);
 	int maxDiff = conf->getArgPInt("--" + prefix + "-maxdifferense", it.getProp("maxDifferense"), 100);
-	int maxProcessingCount = conf->getArgPInt("--" + prefix + "-maxprocessingcount", it.getProp("maxProcessingCount"), 100);
 	int checkConnectionPause = conf->getArgPInt("--" + prefix + "-checkconnection-pause", it.getProp("checkConnectionPause"), 10000);
 	int initpause = conf->getArgPInt("--" + prefix + "-initpause", it.getProp("initpause"), 5000);
+	int recvBufferSize = conf->getArgPInt("--" + prefix + "-recv-buffer-size", it.getProp("recvBufferSize"), 3000);
 
 	std::string updateStrategy = conf->getArg2Param("--" + prefix + "-update-strategy", it.getProp("updateStrategy"), "evloop");
 
@@ -333,16 +333,6 @@ UNetExchange::UNetExchange(uniset::ObjectId objId, uniset::ObjectId shmId, const
 			}
 		}
 
-		UNetReceiver::UpdateStrategy r_upStrategy = UNetReceiver::strToUpdateStrategy( n_it.getProp2("unet_update_strategy", updateStrategy) );
-
-		if( r_upStrategy == UNetReceiver::useUpdateUnknown )
-		{
-			ostringstream err;
-			err << myname << ": Unknown update strategy!!! '" << n_it.getProp2("unet_update_strategy", updateStrategy) << "'" << endl;
-			unetcrit << myname << "(init): " << err.str() << endl;
-			throw SystemError(err.str());
-		}
-
 		unetinfo << myname << "(init): (node='" << n << "') add  basic receiver "
 				 << h << ":" << p << endl;
 		auto r = make_shared<UNetReceiver>(h, p, shm, false, prefix);
@@ -361,11 +351,10 @@ UNetExchange::UNetExchange(uniset::ObjectId objId, uniset::ObjectId shmId, const
 		r->setCheckConnectionPause(checkConnectionPause);
 		r->setInitPause(initpause);
 		r->setMaxDifferens(maxDiff);
-		r->setMaxProcessingCount(maxProcessingCount);
 		r->setRespondID(resp_id, resp_invert);
 		r->setLostPacketsID(lp_id);
 		r->connectEvent( sigc::mem_fun(this, &UNetExchange::receiverEvent) );
-		r->setUpdateStrategy(r_upStrategy);
+		r->setBufferSize(recvBufferSize);
 
 		shared_ptr<UNetReceiver> r2(nullptr);
 
@@ -391,11 +380,10 @@ UNetExchange::UNetExchange(uniset::ObjectId objId, uniset::ObjectId shmId, const
 				r2->setCheckConnectionPause(checkConnectionPause);
 				r2->setInitPause(initpause);
 				r2->setMaxDifferens(maxDiff);
-				r2->setMaxProcessingCount(maxProcessingCount);
 				r2->setRespondID(resp2_id, resp_invert);
 				r2->setLostPacketsID(lp2_id);
 				r2->connectEvent( sigc::mem_fun(this, &UNetExchange::receiverEvent) );
-				r2->setUpdateStrategy(r_upStrategy);
+				r2->setBufferSize(recvBufferSize);
 			}
 		}
 		catch(...)
@@ -547,7 +535,7 @@ void UNetExchange::step() noexcept
 		}
 	}
 
-	for( auto && it : recvlist )
+	for( auto&& it : recvlist )
 		it.step(shm, myname, unetlog);
 }
 
@@ -755,10 +743,10 @@ void UNetExchange::askSensors( UniversalIO::UIOCommand cmd )
 void UNetExchange::sensorInfo( const uniset::SensorMessage* sm )
 {
 	if( sender )
-		sender->updateSensor( sm->id , sm->value );
+		sender->updateSensor( sm->id, sm->value );
 
 	if( sender2 )
-		sender2->updateSensor( sm->id , sm->value );
+		sender2->updateSensor( sm->id, sm->value );
 }
 // ------------------------------------------------------------------------------------------
 bool UNetExchange::activateObject()
@@ -839,7 +827,7 @@ void UNetExchange::initIterators() noexcept
 	if( sender2 )
 		sender2->initIterators();
 
-	for( auto && it : recvlist )
+	for( auto&& it : recvlist )
 		it.initIterators(shm);
 }
 // -----------------------------------------------------------------------------
@@ -855,13 +843,8 @@ void UNetExchange::help_print( int argc, const char* argv[] ) noexcept
 	cout << "--prefix-steptime msec           - Пауза между обновлением информации о связи с узлами." << endl;
 	cout << "--prefix-checkconnection-pause msec  - Пауза между попытками открыть соединение (если это не удалось до этого). По умолчанию: 10000 (10 сек)" << endl;
 	cout << "--prefix-maxdifferense num       - Маскимальная разница в номерах пакетов для фиксации события 'потеря пакетов' " << endl;
-	cout << "--prefix-maxprocessingcount num  - Максимальное количество пакетов обрабатываемых за один раз (если их слишком много)" << endl;
 	cout << "--prefix-nosender [0,1]          - Отключить посылку." << endl;
-	cout << "--prefix-update-strategy [thread,evloop] - Стратегия обновления данных в SM. " << endl;
-	cout << "                                         'thread' - у каждого UNetReceiver отдельный поток" << endl;
-	cout << "                                         'evloop' - используется общий (с приёмом сообщений) event loop" << endl;
-	cout << "                                 По умолчанию: evloop" << endl;
-
+	cout << "--prefix-recv-buffer-size sz     - Размер циклического буфера для приёма сообщений. По умолчанию: 3000" << endl;
 	cout << "--prefix-sm-ready-timeout msec   - Время ожидание я готовности SM к работе. По умолчанию 120000" << endl;
 	cout << "--prefix-filter-field name       - Название фильтрующего поля при формировании списка датчиков посылаемых данным узлом" << endl;
 	cout << "--prefix-filter-value name       - Значение фильтрующего поля при формировании списка датчиков посылаемых данным узлом" << endl;
@@ -913,7 +896,7 @@ std::shared_ptr<UNetExchange> UNetExchange::init_unetexchange(int argc, const ch
 // -----------------------------------------------------------------------------
 void UNetExchange::receiverEvent( const shared_ptr<UNetReceiver>& r, UNetReceiver::Event ev ) noexcept
 {
-	for( auto && it : recvlist )
+	for( auto&& it : recvlist )
 	{
 		if( it.r1 == r )
 		{

extensions/UNetUDP/UNetExchange.h

@@ -58,7 +58,6 @@ namespace uniset
 	    пар [id,value]. Другие узлы принимают их. Помимо этого данный процесс запускает
 		"получателей" по одному на каждый (другой) узел и ловит пакеты от них, сохраняя данные в SM.
 		При этом "получатели" работают на одном(!) потоке с использованием событий libev (см. UNetReceiver).
-		или каждый на своём потоке. Это определяется параметром \b unet_update_strategy.
 
 	    \par
 	        При своём старте процесс считывает из секции \<nodes> список узлов которые необходимо "слушать",
@@ -78,15 +77,13 @@ namespace uniset
 	            ...
 	        </iocards>
 	        </item>
-			<item ip="192.168.56.10" name="Node1" textname="Node1" unet_port="3001" unet_update_strategy="evloop"/>
+			<item ip="192.168.56.10" name="Node1" textname="Node1" unet_port="3001"/>
 	        <item ip="192.168.56.11" name="Node2" textname="Node2" unet_port="3002"/>
 	    </nodes>
 	    \endcode
 
-		* \b unet_update_strategy - задаёт стратегию обновления данных в SM.
-		   Поддерживается два варианта:
-		- 'thread' - отдельный поток обновления
-		- 'evloop' - использование общего с приёмом event loop (libev)
+		Буфер для приёма сообщений можно настроить параметром \b recvBufferSize="1000" в конфигурационной секции
+		или аргументом командной строки \b --prefix-recv-buffer-size sz
 
 		\note Имеется возможность задавать отдельную настроечную секцию для "списка узлов" при помощи параметра
 		 --prefix-nodes-confnode name. По умолчанию настройка ведётся по секции <nodes>

extensions/UNetUDP/UNetSender.cc

@@ -156,7 +156,7 @@ namespace uniset
 	// -----------------------------------------------------------------------------
 	void UNetSender::updateFromSM()
 	{
-		for( auto && it : items )
+		for( auto&& it : items )
 		{
 			UItem& i = it.second;
 
@@ -243,7 +243,7 @@ namespace uniset
 				if( !shm->isLocalwork() )
 					updateFromSM();
 
-				for( auto && it : mypacks )
+				for( auto&& it : mypacks )
 				{
 					if( it.first > 1 && (ncycle % it.first) != 0 )
 						continue;
@@ -504,6 +504,7 @@ namespace uniset
 			if( p.pack_ind >= maxAData )
 			{
 				anum++;
+
 				if( anum >= pk.size() )
 					pk.resize(anum + 1);
 
@@ -551,7 +552,7 @@ namespace uniset
 	// -----------------------------------------------------------------------------
 	void UNetSender::initIterators()
 	{
-		for( auto && it : items )
+		for( auto&& it : items )
 			shm->initIterator(it.second.ioit);
 	}
 	// -----------------------------------------------------------------------------

extensions/UNetUDP/tests/Makefile.am

 if HAVE_TESTS
 
 noinst_PROGRAMS = tests-with-sm urecv-perf-test
+#noinst_PROGRAMS = urecv-perf-test
 
 tests_with_sm_SOURCES   = tests_with_sm.cc test_unetudp.cc
 tests_with_sm_LDADD     = $(top_builddir)/lib/libUniSet2.la $(top_builddir)/extensions/lib/libUniSet2Extensions.la \

extensions/UNetUDP/tests/test_unetudp.cc

@@ -96,45 +96,6 @@ void send( UniSetUDP::UDPMessage& pack, int tout = 2000 )
 	size_t ret = udp_s->sendTo(&s_buf.data, s_buf.len, s_addr);
 	REQUIRE( ret == s_buf.len );
 }
-// -----------------------------------------------------------------------------
-TEST_CASE("[UNetUDP]: queue sort", "[unetudp][packetqueue]")
-{
-	UNetReceiver::PacketQueue q;
-
-	UniSetUDP::UDPMessage m1;
-	m1.num = 10;
-	UniSetUDP::UDPMessage m2;
-	m2.num = 11;
-	UniSetUDP::UDPMessage m3;
-	m3.num = 13;
-
-	UniSetUDP::UDPMessage m4;
-	m4.num = 100;
-
-	// специально складываем в обратном порядке
-	// чтобы проверить "сортировку"
-	q.push(m1);
-	q.push(m3);
-	q.push(m2);
-	q.push(m4);
-
-	UniSetUDP::UDPMessage t = q.top();
-	REQUIRE( t.num == 10 );
-	q.pop();
-
-	t = q.top();
-	REQUIRE( t.num == 11 );
-	q.pop();
-
-	t = q.top();
-	REQUIRE( t.num == 13 );
-	q.pop();
-
-	t = q.top();
-	REQUIRE( t.num == 100 );
-	q.pop();
-}
-// -----------------------------------------------------------------------------
 TEST_CASE("[UNetUDP]: UDPMessage", "[unetudp][udpmessage]")
 {
 	SECTION("UDPMessage::isFull()")

extensions/UNetUDP/tests/tests_with_sm_thread.sh

-#!/bin/sh
-
-# '--' - нужен для отделения аргументов catch, от наших..
-cd ../../../Utilities/Admin/
-./uniset2-start.sh -f ./create_links.sh
-./uniset2-start.sh -f ./create
-
-./uniset2-start.sh -f ./exist | grep -q UNISET_PLC/Controllers || exit 1
-cd -
-
-./uniset2-start.sh -f ./tests-with-sm $* -- --confile unetudp-test-configure.xml --e-startup-pause 10 \
---unet-name UNetExchange --unet-filter-field unet --unet-filter-value 1 --unet-maxdifferense 5 \
---unet-recv-timeout 1000 --unet-sendpause 500  --unet-update-strategy thread
-#--unet-log-add-levels any

extensions/UNetUDP/tests/unetudp-tests.at

-AT_SETUP([UNetUDP tests (with SM)(thread)])
-AT_CHECK([$abs_top_builddir/testsuite/at-test-launch.sh $abs_top_builddir/extensions/UNetUDP/tests tests_with_sm_thread.sh],[0],[ignore],[ignore])
-AT_CLEANUP
-
 AT_SETUP([UNetUDP tests (with SM)(evloop)])
 AT_CHECK([$abs_top_builddir/testsuite/at-test-launch.sh $abs_top_builddir/extensions/UNetUDP/tests tests_with_sm_evloop.sh],[0],[ignore],[ignore])
 AT_CLEANUP

extensions/UNetUDP/tests/urecv_perf_test.cc

@@ -42,7 +42,9 @@ static void run_senders( size_t max, const std::string& s_host, size_t count = 5
 	{
 		try
 		{
+			cout << "create sender: " << s_host << ":" << begPort + i << endl;
 			auto s = make_shared<UDPSocketU>(s_host, begPort + i);
+			s->setBroadcast(true);
 			vsend.emplace_back(s);
 		}
 		catch( Poco::Net::NetException& e )
@@ -103,7 +105,7 @@ static void run_senders( size_t max, const std::string& s_host, size_t count = 5
 		if( packetnum == 0 )
 			packetnum = 1;
 
-		for( auto && udp : vsend )
+		for( auto&& udp : vsend )
 		{
 			try
 			{
@@ -139,6 +141,7 @@ static void run_test( size_t max, const std::string& host )
 	// make receivers..
 	for( size_t i = 0; i < max; i++ )
 	{
+		cout << "create receiver: " << host << ":" << begPort + i << endl;
 		auto r = make_shared<UNetReceiver>(host, begPort + i, smiInstance());
 		r->setLockUpdate(true);
 		vrecv.emplace_back(r);
@@ -147,7 +150,7 @@ static void run_test( size_t max, const std::string& host )
 	size_t count = 0;
 
 	// Run receivers..
-	for( auto && r : vrecv )
+	for( auto&& r : vrecv )
 	{
 		if( r )
 		{
@@ -156,12 +159,12 @@ static void run_test( size_t max, const std::string& host )
 		}
 	}
 
-	cerr << "RUn " << count << " receivers..." << endl;
+	cerr << "RUN " << count << " receivers..." << endl;
 
 	// wait..
 	pause();
 
-	for( auto && r : vrecv )
+	for( auto&& r : vrecv )
 	{
 		if(r)
 			r->stop();

extensions/UNetUDP/unet-udp-tester.cc

@@ -29,6 +29,7 @@ static struct option longopts[] =
 	{ "prof", required_argument, 0, 'y' },
 	{ "a-data", required_argument, 0, 'a' },
 	{ "d-data", required_argument, 0, 'i' },
+	{ "pack-num", required_argument, 0, 'u' },
 	{ NULL, 0, 0, 0 }
 };
 // --------------------------------------------------------------------------
@@ -79,10 +80,11 @@ int main(int argc, char* argv[])
 	unsigned int nprof = 0;
 	std::string d_data = "";
 	std::string a_data = "";
+	size_t packetnum = 1;
 
 	while(1)
 	{
-		opt = getopt_long(argc, argv, "hs:c:r:p:n:t:x:blvdz:y:a:i:", longopts, &optindex);
+		opt = getopt_long(argc, argv, "hs:c:r:p:n:t:x:blvdz:y:a:i:u:", longopts, &optindex);
 
 		if( opt == -1 )
 			break;
@@ -106,6 +108,7 @@ int main(int argc, char* argv[])
 				cout << "[-y|--prof] num          - Print receive statistics every NUM packets (for -r only)" << endl;
 				cout << "[-a|--a-data] id1=val1,id2=val2,... - Analog data. Send: id1=id1,id2=id2,.. for analog sensors" << endl;
 				cout << "[-i|--d-data] id1=val1,id2=val2,... - Digital data. Send: id1=id1,id2=id2,.. for digital sensors" << endl;
+				cout << "[-u|--pack-num] num      - first packet numbrt (default: 1)" << endl;
 				cout << endl;
 				return 0;
 
@@ -171,6 +174,10 @@ int main(int argc, char* argv[])
 				ncycles = atoi(optarg);
 				break;
 
+			case 'u':
+				packetnum = atoi(optarg);
+				break;
+
 			case '?':
 			default:
 				cerr << "? argumnet" << endl;
@@ -345,7 +352,6 @@ int main(int argc, char* argv[])
 
 				Poco::Net::SocketAddress sa(s_host, port);
 				udp->connect(sa);
-				size_t packetnum = 0;
 
 				UniSetUDP::UDPPacket s_buf;
 
@@ -358,7 +364,7 @@ int main(int argc, char* argv[])
 				{
 					mypack.num = packetnum++;
 
-					// при переходе черех максимум (UniSetUDP::MaxPacketNum)
+					// при переходе через максимум (UniSetUDP::MaxPacketNum)
 					// пакет опять должен иметь номер "1"
 					if( packetnum == 0 )
 						packetnum = 1;