Работа на обменом через UDP

conf/test.xml

@@ -103,8 +103,9 @@
 	textname 	- текстовое имя датчика
 -->
 	<nodes port="2809">
-		<item name="LocalhostNode"	alias="" textname="Локальный узел"	ip="127.0.0.1" infserver="InfoServer" dbserver=""/>
-		<item name="Node2"			alias="" textname="Локальный узел"	ip="127.0.0.1" infserver="InfoServer" dbserver=""/>
+		<item name="LocalhostNode"	alias="" textname="Локальный узел"	ip="127.0.0.1" udp_port="3000" infserver="InfoServer" dbserver=""/>
+		<item name="Node1"			alias="" textname="Node1" ip="192.168.56.10" udp_port="3001" />
+		<item name="Node2"			alias="" textname="Node2" ip="192.168.56.11" udp_port="3002" />
 	</nodes>
 
 	<!-- ************************ Датчики ********************** -->

configure.ac

@@ -3,7 +3,7 @@
 # See doc: http://www.gnu.org/software/hello/manual/autoconf/Generic-Programs.html
 
 # AC_PREREQ(2.59)
-AC_INIT([uniset], [0.98.1], pv@etersoft.ru)
+AC_INIT([uniset], [0.99.1], pv@etersoft.ru)
 AM_INIT_AUTOMAKE(AC_PACKAGE_NAME,AC_PACKAGE_VERSION)
 
 # AC_CONFIG_MACRO_DIR([m4])

extensions/UDPExchange/Makefile.am

@@ -6,7 +6,7 @@ libUniSetUDP_la_LIBADD		= $(top_builddir)/lib/libUniSet.la \
 								$(top_builddir)/extensions/lib/libUniSetExtensions.la \
 								$(SIGC_LIBS) $(COMCPP_LIBS)
 libUniSetUDP_la_CXXFLAGS	= -I$(top_builddir)/extensions/include -I$(top_builddir)/extensions/SharedMemory $(SIGC_CFLAGS) $(COMCPP_CFLAGS)
-libUniSetUDP_la_SOURCES 	= UDPPacket.cc UDPExchange.cc UDPSender.cc UDPReceiver.cc
+libUniSetUDP_la_SOURCES 	= UDPPacket.cc UDPExchange.cc UDPSender.cc UDPNReceiver.cc UDPReceiver.cc
 
 #UDPSender.cc
 

extensions/UDPExchange/UDPExchange.cc

-// $Id: UDPExchange.cc,v 1.1 2009/02/10 20:38:27 vpashka Exp $
-// -----------------------------------------------------------------------------
 #include <sstream>
 #include "Exceptions.h"
 #include "Extensions.h"
@@ -18,8 +16,6 @@ activated(false),
 dlist(100),
 maxItem(0)
 {
-	cout << "$Id: UDPExchange.cc,v 1.1 2009/02/10 20:38:27 vpashka Exp $" << endl;
-
 	if( objId == DefaultObjectId )
 		throw UniSetTypes::SystemError("(UDPExchange): objId=-1?!! Use --udp-name" );
 
@@ -40,23 +36,24 @@ maxItem(0)
 
 	// ---------- init RS ----------
 //	UniXML_iterator it(cnode);
-	string s_host	= conf->getArgParam("--udp-host",it.getProp("host"));
+	s_host	= conf->getArgParam("--udp-host",it.getProp("host"));
 	if( s_host.empty() )
 		throw UniSetTypes::SystemError(myname+"(UDPExchange): Unknown host. Use --udp-host" );
 
-	port = conf->getArgInt("--udp-port",it.getProp("port"));
-	if( port <= 0 )
-		throw UniSetTypes::SystemError(myname+"(UDPExchange): Unknown port address. Use --udp-port" );
+	host = s_host.c_str();
+
+	buildReceiverList();
+
+//	port = conf->getArgInt("--udp-port",it.getProp("port"));
+	if( port <= 0 || port == DefaultObjectId )
+		throw UniSetTypes::SystemError(myname+"(UDPExchange): Unknown port address" );
 
 	if( dlog.debugging(Debug::INFO) )
 		dlog[Debug::INFO] << "(UDPExchange): UDP set to " << s_host << ":" << port << endl;
 	
-	host = s_host.c_str();
 	try
 	{
-		udp = new ost::UDPDuplex(host,port);
-//		udp->UDPTransmit::setBroadcast(false);
-//		udp->UDPTransmit::setRouting(false);
+		udp = new ost::UDPBroadcast(host,port);	  
 	}
 	catch( ost::SockException& e )
 	{
@@ -68,9 +65,7 @@ maxItem(0)
 	thr = new ThreadCreator<UDPExchange>(this, &UDPExchange::poll);
 
 	recvTimeout = conf->getArgPInt("--udp-recv-timeout",it.getProp("recvTimeout"), 5000);
-
 	sendTimeout = conf->getArgPInt("--udp-send-timeout",it.getProp("sendTimeout"), 5000);
-
 	polltime = conf->getArgPInt("--udp-polltime",it.getProp("polltime"), 100);
 
 	// -------------------------------
@@ -128,6 +123,9 @@ maxItem(0)
 // -----------------------------------------------------------------------------
 UDPExchange::~UDPExchange()
 {
+	for( ReceiverList::iterator it=rlist.begin(); it!=rlist.end(); it++ )
+		delete (*it);
+
 	delete udp;
 	delete shm;
 	delete thr;
@@ -151,21 +149,8 @@ void UDPExchange::waitSMReady()
 	}
 }
 // -----------------------------------------------------------------------------
-/*
-void UDPExchange::timerInfo( TimerMessage *tm )
-{
-	if( tm->id == tmExchange )
-		step();
-}
-*/
-// -----------------------------------------------------------------------------
 void UDPExchange::step()
 {
-//	{
-//		uniset_mutex_lock l(pollMutex,2000);
-//		poll();
-//	}
-
 	if( !activated )
 		return;
 
@@ -190,24 +175,31 @@ void UDPExchange::poll()
 	dlist.resize(maxItem);
 	dlog[Debug::INFO] << myname << "(init): dlist size = " << dlist.size() << endl;
 
+	for( ReceiverList::iterator it=rlist.begin(); it!=rlist.end(); it++ )
+	{
+		(*it)->setReceiveTimeout(recvTimeout);		
+		if( dlog.debugging(Debug::INFO) )
+			dlog[Debug::INFO] << myname << "(poll): start exchange for " << (*it)->getName() << endl;
+		(*it)->start();
+	}
+
+	ost::IPV4Broadcast h = s_host.c_str();
 	try
 	{			
-		udp->connect(host,port);
+		udp->setPeer(h,port);
 	}
-	catch( UniSetTypes::Exception& ex)
+	catch( ost::SockException& e )
 	{
-		cerr << myname << "(step): " << ex << std::endl;
-//		reise(SIGTERM);
-		return;
+		ostringstream s;
+		s << e.getString() << ": " << e.getSystemErrorString();
+		dlog[Debug::CRIT] << myname << "(poll): " << s.str() << endl;
+		throw SystemError(s.str());
 	}
 
-
-
 	while( activated )
 	{
 		try
 		{
-			recv();
 			send();
 		}
 		catch( ost::SockException& e )
@@ -229,82 +221,40 @@ void UDPExchange::poll()
 	cerr << "************* execute FINISH **********" << endl;
 }
 // -----------------------------------------------------------------------------
-void UDPExchange::recv()
-{	
-	cout << myname << ": recv...." << endl;
-	UniSetUDP::UDPHeader h;
-	// receive
-	if( udp->isInputReady(recvTimeout) )
-	{
-		ssize_t ret = udp->UDPReceive::receive(&h,sizeof(h));
-		if( ret<(ssize_t)sizeof(h) )
-		{
-			cerr << myname << "(receive): ret=" << ret << " sizeof=" << sizeof(h) << endl;
-			return;
-		}
-		
-		cout << myname << "(receive): header: " << h << endl;
-
-		UniSetUDP::UDPData d;
-		// ignore echo...
-#if 0
-		if( h.nodeID == conf->getLocalNode() && h.procID == getId() )
-		{
-			for( int i=0; i<h.dcount;i++ )
-			{
-				ssize_t ret = udp->UDPReceive::receive(&d,sizeof(d));
-				if( ret < (ssize_t)sizeof(d) )
-					return;
-			}
-			return;
-		}
-#endif 		
-		for( int i=0; i<h.dcount;i++ )
-		{
-			ssize_t ret = udp->UDPReceive::receive(&d,sizeof(d));
-			if( ret<(ssize_t)sizeof(d) )
-			{
-				cerr << myname << "(receive data " << i << "): ret=" << ret << " sizeof=" << sizeof(d) << endl;
-				break;
-			}
-			
-			cout << myname << "(receive data " << i << "): " << d << endl;
-		}
-	}
-}
-// -----------------------------------------------------------------------------
 void UDPExchange::send()
 {
 	cout << myname << ": send..." << endl;
-
+/*
 	UniSetUDP::UDPHeader h;
 	h.nodeID = conf->getLocalNode();
 	h.procID = getId();
 	h.dcount = mypack.size();
-	// receive
-	if( udp->isOutputReady(sendTimeout) )
+	if( udp->isPending(ost::Socket::pendingOutput) )
 	{
-		ssize_t ret = udp->transmit((char*)(&h),sizeof(h));
+		ssize_t ret = udp->send((char*)(&h),sizeof(h));
 		if( ret<(ssize_t)sizeof(h) )
 		{
 			cerr << myname << "(send data header): ret=" << ret << " sizeof=" << sizeof(h) << endl;
 			return;
 		}
-
+*/
 #warning use mutex for list!!!
 		UniSetUDP::UDPMessage::UDPDataList::iterator it = mypack.dlist.begin();
+		
 		for( ; it!=mypack.dlist.end(); ++it )
 		{
+//			while( !udp->isPending(ost::Socket::pendingOutput) )
+//				msleep(30);
+
         	cout << myname << "(send): " << (*it) << endl;
-			ssize_t ret = udp->transmit((char*)(&(*it)),sizeof(*it));
-			if( ret<(ssize_t)sizeof(*it) )
+			ssize_t ret = udp->send((char*)(&(*it)),sizeof(UniSetUDP::UDPData));
+			if( ret<(ssize_t)sizeof(UniSetUDP::UDPData) )
 			{
-				cerr << myname << "(send data): ret=" << ret << " sizeof=" << sizeof(*it) << endl;
+				cerr << myname << "(send data): ret=" << ret << " sizeof=" << sizeof(UniSetUDP::UDPData) << endl;
 				break;
 			}
 		}
-	}
-	
+//	}
 }
 // -----------------------------------------------------------------------------
 void UDPExchange::processingMessage(UniSetTypes::VoidMessage *msg)
@@ -488,6 +438,9 @@ void UDPExchange::sigterm( int signo )
 	cerr << myname << ": ********* SIGTERM(" << signo <<") ********" << endl;
 	activated = false;
 	udp->disconnect();
+	for( ReceiverList::iterator it=rlist.begin(); it!=rlist.end(); it++ )
+		(*it)->stop();
+
 	UniSetObject_LT::sigterm(signo);
 }
 // ------------------------------------------------------------------------------------------
@@ -607,11 +560,9 @@ void UDPExchange::help_print( int argc, char* argv[] )
 	cout << "--udp-initPause		- Задержка перед инициализацией (время на активизация процесса)" << endl;
 	cout << "--udp-notRespondSensor - датчик связи для данного процесса " << endl;
 	cout << "--udp-sm-ready-timeout - время на ожидание старта SM" << endl;
-	cout << " Настройки протокола RS: " << endl;
-	cout << "--udp-dev devname  - файл устройства" << endl;
-	cout << "--udp-speed        - Скорость обмена (9600,19920,38400,57600,115200)." << endl;
-	cout << "--udp-my-addr      - адрес текущего узла" << endl;
-	cout << "--udp-recv-timeout - Таймаут на ожидание ответа." << endl;
+	cout << " Настройки протокола UDP: " << endl;
+	cout << "--udp-host [ip|hostname]  - Адрес сервера" << endl;
+	cout << "--udp-send-timeout - Таймаут на посылку ответа." << endl;
 }
 // -----------------------------------------------------------------------------
 UDPExchange* UDPExchange::init_udpexchange( int argc, char* argv[], UniSetTypes::ObjectId icID, SharedMemory* ic )
@@ -641,3 +592,46 @@ std::ostream& operator<<( std::ostream& os, UDPExchange::UItem& p )
 	return os 	<< " sid=" << p.si.id;
 }
 // -----------------------------------------------------------------------------
+void UDPExchange::buildReceiverList()
+{
+	xmlNode* n = conf->getXMLNodesSection();
+	if( !n )
+	{
+		dlog[Debug::WARN] << myname << "(buildReceiverList): <nodes> not found! ignore..." << endl;
+		return;
+	}
+	
+	UniXML_iterator it(n);
+	if( !it.goChildren() )
+	{
+		dlog[Debug::WARN] << myname << "(buildReceiverList): <nodes> is empty?! ignore..." << endl;
+		return;
+	}
+	
+	for( ; it.getCurrent(); it.goNext() )
+	{
+		ObjectId n_id = conf->getNodeID( it.getProp("name") );
+		if( n_id == conf->getLocalNode() )
+		{
+			port = it.getIntProp("udp_port");
+			if( port<=0 )
+				port = n_id;
+			dlog[Debug::INFO] << myname << "(buildReceiverList): init myport port=" << port << endl;
+			continue;
+		}
+
+		int p = it.getIntProp("udp_port");
+		if( p <=0 )
+			p = n_id;
+
+		if( p == DefaultObjectId )
+		{
+			dlog[Debug::WARN] << myname << "(buildReceiverList): node=" << it.getProp("name") << " unknown port. ignore..." << endl;
+			continue;
+		}
+
+		UDPNReceiver* r = new UDPNReceiver(p,host,shm->getSMID(),shm->SM());
+		rlist.push_back(r);
+	}
+}
+// ------------------------------------------------------------------------------------------

extensions/UDPExchange/UDPExchange.h

-// $Id: UDPExchange.h,v 1.1 2009/02/10 20:38:27 vpashka Exp $
-// -----------------------------------------------------------------------------
 #ifndef UDPExchange_H_
 #define UDPExchange_H_
 // -----------------------------------------------------------------------------
@@ -14,6 +12,7 @@
 #include "SharedMemory.h"
 #include "ThreadCreator.h"
 #include "UDPPacket.h"
+#include "UDPNReceiver.h"
 // -----------------------------------------------------------------------------
 class UDPExchange:
 	public UniSetObject_LT
@@ -76,6 +75,7 @@ class UDPExchange:
 
 		void readConfiguration();
 		bool check_item( UniXML_iterator& it );
+		void buildReceiverList();
 
 	private:
 		UDPExchange();
@@ -90,9 +90,10 @@ class UDPExchange:
 
 		int polltime;	/*!< переодичность обновления данных, [мсек] */
 
-		ost::UDPDuplex* udp;
+		ost::UDPBroadcast* udp;
 		ost::IPV4Host host;
 		ost::tpport_t port;
+		std::string s_host;
 
 		UniSetTypes::uniset_mutex pollMutex;
 		Trigger trTimeout;
@@ -108,6 +109,9 @@ class UDPExchange:
 		int maxItem;
 		
 
+		typedef std::list<UDPNReceiver*> ReceiverList;
+		ReceiverList rlist;
+		
 		ThreadCreator<UDPExchange>* thr;
 };
 // -----------------------------------------------------------------------------

extensions/UDPExchange/UDPReceiver.cc

-// $Id: UDPReceiver.cc,v 1.1 2009/02/10 20:38:27 vpashka Exp $
-// -----------------------------------------------------------------------------
 #include <sstream>
 #include "Exceptions.h"
 #include "Extensions.h"
@@ -62,9 +60,6 @@ activated(false)
 	thr = new ThreadCreator<UDPReceiver>(this, &UDPReceiver::poll);
 
 	recvTimeout = conf->getArgPInt("--udp-recv-timeout",it.getProp("recvTimeout"), 5000);
-
-	sendTimeout = conf->getArgPInt("--udp-send-timeout",it.getProp("sendTimeout"), 5000);
-
 	polltime = conf->getArgPInt("--udp-polltime",it.getProp("polltime"), 100);
 
 	// -------------------------------
@@ -464,8 +459,3 @@ UDPReceiver* UDPReceiver::init_udpreceiver( int argc, char* argv[], UniSetTypes:
 	return new UDPReceiver(ID,icID,ic);
 }
 // -----------------------------------------------------------------------------
-std::ostream& operator<<( std::ostream& os, UDPReceiver::UItem& p )
-{
-	return os 	<< " sid=" << p.si.id;
-}
-// -----------------------------------------------------------------------------

extensions/UDPExchange/UDPReceiver.h

-// $Id: UDPExchange.h,v 1.1 2009/02/10 20:38:27 vpashka Exp $
-// -----------------------------------------------------------------------------
-#ifndef UDPExchange_H_
+#ifndef UDPReceiver_H_
 #define UDPReceiver_H_
 // -----------------------------------------------------------------------------
 #include <ostream>
@@ -29,22 +27,6 @@ class UDPReceiver:
 		/*! глобальная функция для вывода help-а */
 		static void help_print( int argc, char* argv[] );
 
-		struct UItem
-		{
-			UItem():
-				val(0)
-			{}
-
-			IOController_i::SensorInfo si;
-			IOController::AIOStateList::iterator ait;
-			IOController::DIOStateList::iterator dit;
-			UniSetTypes::uniset_spin_mutex val_lock;
-			UniSetUDP::UDPMessage::UDPDataList::iterator pack_it;
-			long val;
-
-			friend std::ostream& operator<<( std::ostream& os, UItem& p );
-		};
-
 	protected:
 
 		xmlNode* cnode;
@@ -90,7 +72,6 @@ class UDPReceiver:
 		UniSetTypes::uniset_mutex pollMutex;
 		Trigger trTimeout;
 		int recvTimeout;
-		int sendTimeout;
 		
 		bool activated;
 		int activateTimeout;

extensions/UDPExchange/start_fg.sh

 #!/bin/sh
 
-uniset-start.sh -f ./uniset-udpexchange --udp-name UDPExchange --udp-host 192.168.56.255 --udp-port 2050 --udp-broadcast 1 \
+uniset-start.sh -f ./uniset-udpexchange --udp-name UDPExchange --udp-host 192.168.56.255 \
+	--udp-broadcast 1 --udp-polltime 1000 \
 	--confile test.xml \
-	--udp-filter-field udp --udp-filter-value 1 \
 	--dlog-add-levels info,crit,warn
+#	--udp-filter-field udp --udp-filter-value 1 \
 

extensions/UDPExchange/start_fg_receiver.sh

 #!/bin/sh
 
 uniset-start.sh -f ./uniset-udpreceiver --udp-name UDPExchange \
-	--udp-host 192.168.56.255 --udp-port 2050 \
+	--udp-host 192.168.56.255 --udp-port 3000 \
 	--confile test.xml \
 	--udp-filter-field udp --udp-filter-value 1 \
 	--dlog-add-levels info,crit,warn

extensions/UDPExchange/start_fg_sender.sh

@@ -2,7 +2,7 @@
 
 uniset-start.sh -f ./uniset-udpsender --udp-name UDPExchange \
 	--udp-host 192.168.56.255 --udp-port 2050 --udp-broadcast 1\
-	--udp-sendtime 100 \
+	--udp-sendtime 2000 \
 	--confile test.xml \
 	--dlog-add-levels info,crit,warn
 #	--udp-filter-field udp --udp-filter-value 1 \

extensions/UDPExchange/udpexchange.cc

@@ -11,46 +11,6 @@ using namespace UniSetExtensions;
 // -----------------------------------------------------------------------------
 int main( int argc, char** argv )
 {
-	ost::IPV4Host host = "192.168.56.10";
-	ost::tpport_t port = 2050;
-	try
-	{
-		ost::UDPSocket* udp = new ost::UDPSocket();
-		udp->setPeer(host,port);
-//		udp->UDPTransmit::setBroadcast(false);
-//		udp->UDPTransmit::setRouting(false);
-		UniSetUDP::UDPHeader h;
-		h.nodeID = 20;
-		h.procID = 20;
-		h.dcount = 0;
-		
-//		udp->connect(host,port);
-		
-		
-//		ost::IPV4Host h1 = udp->getPeer();
-//		cout << "peer: " << h1 << endl;
-		
-		//if( udp->isOutputReady(5000) )
-		while(1)
-		{
-		  ssize_t ret = udp->send((char*)(&h),sizeof(h));
-		  if( ret<(ssize_t)sizeof(h) )
-		  {
-			  cerr << "(send data header): ret=" << ret << " sizeof=" << sizeof(h) << endl;
-//			  return 0;
-		  }
-		  else
-			  cout << "SEND OK..." << endl;
-		  
-		  msleep(1000);
-		}
-	}
-	catch( ost::SockException& e )
-	{
-		cerr  << e.getString() << ": " << e.getSystemErrorString() << endl;
-		return 0;
-	}  
-  
 	try
 	{
 		if( argc>1 && (!strcmp(argv[1],"--help") || !strcmp(argv[1],"-h")) )
@@ -105,9 +65,9 @@ int main( int argc, char** argv )
 		act.broadcast( sm.transport_msg() );
 
 		unideb(Debug::ANY) << "\n\n\n";
-		unideb[Debug::ANY] << "(main): -------------- RTU Exchange START -------------------------\n\n";
+		unideb[Debug::ANY] << "(main): -------------- UDP Exchange START -------------------------\n\n";
 		dlog(Debug::ANY) << "\n\n\n";
-		dlog[Debug::ANY] << "(main): -------------- RTU Exchange START -------------------------\n\n";
+		dlog[Debug::ANY] << "(main): -------------- UDP Exchange START -------------------------\n\n";
 
 		act.run(false);
 //		msleep(500);

extensions/include/SMInterface.h

@@ -80,6 +80,8 @@ class SMInterface
 
 		inline bool isLocalwork(){ return (ic==NULL); }
 		inline UniSetTypes::ObjectId ID(){ return myid; }
+		inline IONotifyController* SM(){ return ic; }
+		inline UniSetTypes::ObjectId getSMID(){ return shmID; }
 		
 	protected:
 		IONotifyController* ic;

include/Configuration.h

@@ -107,6 +107,7 @@ namespace UniSetTypes
 		xmlNode* getXMLObjectsSection();
 		xmlNode* getXMLControllersSection();
 		xmlNode* getXMLServicesSection();
+		xmlNode* getXMLNodesSection();
 		xmlNode* getXMLObjectNode( UniSetTypes::ObjectId );
 	
 		// net
@@ -219,6 +220,7 @@ namespace UniSetTypes
 		static xmlNode* xmlObjectsSec;
 		static xmlNode* xmlControllersSec;
 		static xmlNode* xmlServicesSec;
+		static xmlNode* xmlNodesSec;
 
 		ObjectId localTimerService;
 		ObjectId localDBServer;

src/Various/Configuration.cc

@@ -974,6 +974,7 @@ xmlNode* Configuration::xmlSensorsSec = 0;
 xmlNode* Configuration::xmlObjectsSec = 0;
 xmlNode* Configuration::xmlControllersSec = 0;
 xmlNode* Configuration::xmlServicesSec = 0;
+xmlNode* Configuration::xmlNodesSec = 0;
 // -------------------------------------------------------------------------
 xmlNode* Configuration::getXMLSensorsSection()
 {
@@ -1012,6 +1013,15 @@ xmlNode* Configuration::getXMLServicesSection()
 	return xmlServicesSec;
 }
 // -------------------------------------------------------------------------
+xmlNode* Configuration::getXMLNodesSection()
+{
+	if( xmlNodesSec )
+		return xmlNodesSec;
+	
+	xmlNodesSec = unixml.findNode(unixml.getFirstNode(),"nodes");
+	return xmlNodesSec;
+}
+// -------------------------------------------------------------------------
 xmlNode* Configuration::getXMLObjectNode( UniSetTypes::ObjectId id )
 {
 	const ObjectInfo* i = oind->getObjectInfo(id);