(UNET): версия UReceiver без использования std::mutex

extensions/UNetUDP/UNetReceiver.cc

@@ -76,6 +76,8 @@ UNetReceiver::UNetReceiver( const std::string& s_host, const ost::tpport_t _port
 
 	if( !createConnection(nocheckConnection /* <-- это флаг throwEx */) )
 		evCheckConnection.set<UNetReceiver, &UNetReceiver::checkConnectionEvent>(this);
+
+	evForceUpdate.set<UNetReceiver, &UNetReceiver::forceUpdateEvent>(this);
 }
 // -----------------------------------------------------------------------------
 UNetReceiver::~UNetReceiver()
@@ -148,7 +150,6 @@ void UNetReceiver::setLostPacketsID( UniSetTypes::ObjectId id )
 // -----------------------------------------------------------------------------
 void UNetReceiver::setLockUpdate( bool st )
 {
-	uniset_rwmutex_wrlock l(lockMutex);
 	lockUpdate = st;
 
 	if( !st )
@@ -226,6 +227,8 @@ void UNetReceiver::start()
 // -----------------------------------------------------------------------------
 void UNetReceiver::evprepare( const ev::loop_ref& eloop )
 {
+	evForceUpdate.set(eloop);
+
 	if( !udp )
 	{
 		evCheckConnection.set(eloop);
@@ -245,12 +248,8 @@ void UNetReceiver::evfinish( const ev::loop_ref& eloop )
 {
 	activated = false;
 
-	{
-		std::lock_guard<std::mutex> l(checkConnMutex);
-
-		if( evCheckConnection.is_active() )
-			evCheckConnection.stop();
-	}
+	if( evCheckConnection.is_active() )
+		evCheckConnection.stop();
 
 	if( evReceive.is_active() )
 		evReceive.stop();
@@ -258,13 +257,26 @@ void UNetReceiver::evfinish( const ev::loop_ref& eloop )
 	if( evUpdate.is_active() )
 		evUpdate.stop();
 
+	if( evForceUpdate.is_active() )
+		evForceUpdate.stop();
+
 	//udp->disconnect();
 	udp = nullptr;
 }
 // -----------------------------------------------------------------------------
 void UNetReceiver::forceUpdate()
 {
-	uniset_rwmutex_wrlock l(packMutex);
+	evForceUpdate.start(0, 0.001);
+}
+// -----------------------------------------------------------------------------
+void UNetReceiver::forceUpdateEvent( ev::timer& watcher, int revents )
+{
+	if( EV_ERROR & revents )
+	{
+		unetcrit << myname << "(forceUpdateEvent): EVENT ERROR.." << endl;
+		return;
+	}
+
 	pnum = 0; // сбрасываем запомненый номер последнего обработанного пакета
 	// и тем самым заставляем обновить данные в SM (см. update)
 }
@@ -281,9 +293,6 @@ void UNetReceiver::update()
 	while( k > 0 )
 	{
 		{
-			// lock qpack
-			uniset_rwmutex_wrlock l(packMutex);
-
 			if( qpack.empty() )
 				return;
 
@@ -374,12 +383,8 @@ void UNetReceiver::update()
 				}
 
 				// обновление данных в SM (блокировано)
-				{
-					uniset_rwmutex_rlock l(lockMutex);
-
-					if( lockUpdate )
-						continue;
-				}
+				if( lockUpdate )
+					continue;
 
 				shm->localSetValue(ii.ioit, id, val, shm->ID());
 			}
@@ -412,12 +417,8 @@ void UNetReceiver::update()
 				}
 
 				// обновление данных в SM (блокировано)
-				{
-					uniset_rwmutex_rlock l(lockMutex);
-
-					if( lockUpdate )
-						continue;
-				}
+				if( lockUpdate )
+					continue;
 
 				shm->localSetValue(ii.ioit, d.id, d.val, shm->ID());
 			}
@@ -554,8 +555,6 @@ void UNetReceiver::checkConnectionEvent( ev::periodic& tm, int revents )
 
 	unetinfo << myname << "(checkConnectionEvent): check connection event.." << endl;
 
-	std::lock_guard<std::mutex> l(checkConnMutex);
-
 	if( !createConnection(false) )
 		tm.again();
 }
@@ -640,9 +639,6 @@ bool UNetReceiver::receive()
 #endif
 
 	{
-		// lock qpack
-		uniset_rwmutex_wrlock l(packMutex);
-
 		if( !waitClean )
 		{
 			qpack.push(pack);

extensions/UNetUDP/UNetReceiver.h

@@ -174,6 +174,7 @@ class UNetReceiver:
 		void readEvent( ev::io& watcher );
 		void updateEvent( ev::periodic& watcher, int revents );
 		void checkConnectionEvent( ev::periodic& watcher, int revents );
+		void forceUpdateEvent( ev::timer& watcher, int revents );
 		virtual void evprepare( const ev::loop_ref& eloop ) override;
 		virtual void evfinish(const ev::loop_ref& eloop ) override;
 		virtual std::string wname() override
@@ -211,15 +212,14 @@ class UNetReceiver:
 		ev::io evReceive;
 		ev::periodic evUpdate;
 		ev::periodic evCheckConnection;
+		ev::timer evForceUpdate;
 
 		// делаем loop общим.. одним на всех!
 		static CommonEventLoop loop;
 
 		double updateTime = { 0.01 };
 		double checkConnectionTime = { 10.0 }; // sec
-		std::mutex checkConnMutex;
 
-		UniSetTypes::uniset_rwmutex pollMutex;
 		PassiveTimer ptRecvTimeout;
 		PassiveTimer ptPrepare;
 		timeout_t recvTimeout = { 5000 }; // msec
@@ -239,7 +239,6 @@ class UNetReceiver:
 		PacketQueue qpack;    /*!< очередь принятых пакетов (отсортированных по возрастанию номера пакета) */
 		UniSetUDP::UDPMessage pack;        /*!< просто буфер для получения очередного сообщения */
 		UniSetUDP::UDPPacket r_buf;
-		UniSetTypes::uniset_rwmutex packMutex; /*!< mutex для работы с очередью */
 		size_t pnum = { 0 };    /*!< текущий номер обработанного сообщения, для проверки непрерывности последовательности пакетов */
 
 		/*! максимальная разница межд номерами пакетов, при которой считается, что счётчик пакетов
@@ -253,8 +252,7 @@ class UNetReceiver:
 
 		size_t maxProcessingCount = { 100 }; /*!< максимальное число обрабатываемых за один раз сообщений */
 
-		bool lockUpdate = { false }; /*!< флаг блокировки сохранения принятых данных в SM */
-		UniSetTypes::uniset_rwmutex lockMutex;
+		std::atomic_bool lockUpdate = { false }; /*!< флаг блокировки сохранения принятых данных в SM */
 
 		EventSlot slEvent;
 		Trigger trTimeout;

extensions/UNetUDP/tests/Makefile.am

 if HAVE_TESTS
 
-noinst_PROGRAMS = tests-with-sm
+noinst_PROGRAMS = tests-with-sm 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 \
@@ -11,6 +11,16 @@ tests_with_sm_CPPFLAGS  = -I$(top_builddir)/include -I$(top_builddir)/extensions
 	-I$(top_builddir)/extensions/UNetUDP \
 	-I$(top_builddir)/extensions/SharedMemory $(SIGC_CFLAGS)  $(COMCPP_CFLAGS)
 
+urecv_perf_test_SOURCES   = urecv_perf_test.cc
+urecv_perf_test_LDADD     = $(top_builddir)/lib/libUniSet2.la $(top_builddir)/extensions/lib/libUniSet2Extensions.la \
+	$(top_builddir)/extensions/UNetUDP/libUniSet2UNetUDP.la \
+	$(top_builddir)/extensions/SharedMemory/libUniSet2SharedMemory.la \
+	$(SIGC_LIBS) $(COMCPP_LIBS)
+urecv_perf_test_CPPFLAGS  = -I$(top_builddir)/include -I$(top_builddir)/extensions/include \
+	-I$(top_builddir)/extensions/UNetUDP \
+	-I$(top_builddir)/extensions/SharedMemory $(SIGC_CFLAGS)  $(COMCPP_CFLAGS)
+
+
 include $(top_builddir)/testsuite/testsuite-common.mk
 
 check-local: atconfig package.m4 $(TESTSUITE) unetudp-tests.at

extensions/UNetUDP/tests/unet_perf_test_senders.sh

+#!/bin/sh
+
+
+for p in `seq 50001 50010`; do
+	uniset2-unet-udp-tester -s 127.255.255.255:$p -x 50 &
+done

extensions/UNetUDP/tests/unet_perf_test_senders_stop.sh

+#!/bin/sh
+
+killall uniset2-unet-udp-tester

extensions/UNetUDP/tests/urecv_perf_test.cc

+#include <memory>
+#include <chrono>
+#include <string>
+#include "Debug.h"
+#include "UNetReceiver.h"
+#include "SMInterface.h"
+#include "Extensions.h"
+// --------------------------------------------------------------------------
+using namespace std;
+using namespace UniSetTypes;
+using namespace UniSetExtensions;
+// --------------------------------------------------------------------------
+static shared_ptr<SMInterface> smi;
+static shared_ptr<UInterface> ui;
+static ObjectId myID = 6000;
+static ObjectId shmID = 5003;
+static int begPort = 50000;
+// --------------------------------------------------------------------------
+shared_ptr<SMInterface> smiInstance()
+{
+	if( smi == nullptr )
+	{
+		if( ui == nullptr )
+			ui = make_shared<UInterface>();
+
+		smi = make_shared<SMInterface>(shmID, ui, myID );
+	}
+
+	return smi;
+}
+// --------------------------------------------------------------------------
+static void run_senders( size_t max, const std::string& s_host, size_t count=50, timeout_t usecpause=50 )
+{
+	ost::IPV4Host host(s_host.c_str());
+
+	std::vector< std::shared_ptr<ost::UDPSocket> > vsend;
+	vsend.reserve(max);
+
+	cout << "Run " << max << " senders (" << s_host << ")" << endl;
+
+	// make sendesrs..
+	for( size_t i=0; i<max; i++ )
+	{
+		try
+		{
+			auto s = make_shared<ost::UDPSocket>(host,begPort+i);
+			vsend.emplace_back(s);
+		}
+		catch( ost::SockException& e )
+		{
+			cerr << "(run_senders): " << e.getString() << " (" << host << ")" << endl;
+			throw;
+		}
+		catch( std::exception& ex)
+		{
+			cerr << "(run_senders): " << ex.what() << endl;
+			throw;
+		}
+	}
+
+	UniSetUDP::UDPMessage mypack;
+	mypack.nodeID = 100;
+	mypack.procID = 100;
+
+	for( size_t i = 0; i < count; i++ )
+	{
+		UniSetUDP::UDPAData d(i, i);
+		mypack.addAData(d);
+	}
+
+	for( size_t i = 0; i < count; i++ )
+		mypack.addDData(i, i);
+
+	for( size_t i=0; i<max; i++ )
+	{
+		try
+		{
+			if( vsend[i] )
+				vsend[i]->setPeer(host, begPort+i);
+		}
+		catch( ost::SockException& e )
+		{
+			cerr << "(run_senders): " << e.getString() << " (" << host << ")" << endl;
+			throw;
+		}
+		catch( std::exception& ex)
+		{
+			cerr << "(run_senders): " << ex.what() << endl;
+			throw;
+		}
+	}
+
+	size_t packetnum = 0;
+	UniSetUDP::UDPPacket s_buf;
+
+	size_t nc = 1;
+
+	while( nc )
+	{
+		mypack.num = packetnum++;
+
+		// при переходе черех максимум (UniSetUDP::MaxPacketNum)
+		// пакет опять должен иметь номер "1"
+		if( packetnum == 0 )
+			packetnum = 1;
+
+		for( auto&& udp: vsend )
+		{
+			try
+			{
+				if( udp->isPending(ost::Socket::pendingOutput, 100) )
+				{
+					mypack.transport_msg(s_buf);
+					size_t ret = udp->send((char*)&s_buf.data, s_buf.len);
+
+					if( ret < s_buf.len )
+						cerr << "(send): FAILED ret=" << ret << " < sizeof=" << s_buf.len << endl;
+				}
+			}
+			catch( ost::SockException& e )
+			{
+				cerr << "(send): " << e.getString() << " (" << host << ")" << endl;
+			}
+			catch( ... )
+			{
+				cerr << "(send): catch ..." << endl;
+			}
+
+		}
+		usleep(usecpause);
+	}
+}
+// --------------------------------------------------------------------------
+static void run_test( size_t max, const std::string& host )
+{
+	std::vector< std::shared_ptr<UNetReceiver> > vrecv;
+	vrecv.reserve(max);
+
+	// make receivers..
+	for( size_t i=0; i<max; i++ )
+	{
+		auto r = make_shared<UNetReceiver>(host,begPort+i,smiInstance());
+		vrecv.emplace_back(r);
+	}
+
+	size_t count = 0;
+	// Run receivers..
+	for( auto&& r: vrecv )
+	{
+		if( r )
+		{
+			count++;
+			r->start();
+		}
+	}
+
+	cerr << "RUn " << count << " receivers..." << endl;
+
+	// wait..
+	pause();
+
+	for( auto&& r: vrecv )
+	{
+		if(r)
+			r->stop();
+	}
+}
+// --------------------------------------------------------------------------
+int main(int argc, char* argv[] )
+{
+	std::string host = "127.255.255.255";
+	try
+	{
+		auto conf = uniset_init(argc, argv);
+
+		if( argc > 1 && !strcmp(argv[1],"s") )
+			run_senders(10,host);
+		else
+			run_test(10,host);
+
+		return 0;
+	}
+	catch( const SystemError& err )
+	{
+		cerr << "(urecv-perf-test): " << err << endl;
+	}
+	catch( const Exception& ex )
+	{
+		cerr << "(urecv-perf-test): " << ex << endl;
+	}
+	catch( const std::exception& e )
+	{
+		cerr << "(tests_with_sm): " << e.what() << endl;
+	}
+	catch(...)
+	{
+		cerr << "(urecv-perf-test): catch(...)" << endl;
+	}
+
+	return 1;
+}

extensions/UNetUDP/tests/urecv_perf_test_start.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 ./urecv-perf-test $* -- --confile unetudp-test-configure.xml

extensions/UNetUDP/tests/urecv_perf_test_start_gdb.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 -g ./urecv-perf-test $* -- --confile unetudp-test-configure.xml

uniset2.files

@@ -223,6 +223,7 @@ extensions/UNetUDP/tests/test_unetudp.cc
 extensions/UNetUDP/tests/tests_individual_process.cc
 extensions/UNetUDP/tests/tests_with_sm.cc
 extensions/UNetUDP/tests/unetudp-test-configure.xml
+extensions/UNetUDP/tests/urecv_perf_test.cc
 extensions/UNetUDP/libUniSet2UNetUDP.pc.in
 extensions/UNetUDP/Makefile.am
 extensions/UNetUDP/test.xml