UDP: Версия с буфером

extensions/UDPExchange/UDPExchange.cc

@@ -228,7 +228,7 @@ void UDPExchange::send()
 	UniSetUDP::UDPHeader h;
 	h.nodeID = conf->getLocalNode();
 	h.procID = getId();
-	h.dcount = mypack.size();
+	h.dcount = mypack.msg.header.dcount;
 	h.num = packetnum++;
 
 	mypack.msg.header = h;
@@ -239,12 +239,16 @@ void UDPExchange::send()
 	memcpy( &(udpbuf[ind]),&(mypack.data),mypack.size());
 	ind += mypack.size();
 */
+	cout << "************* send header: " << mypack.msg.header << endl;
+	int sz = mypack.size() * sizeof(UniSetUDP::UDPHeader);
 	if( udp->isPending(ost::Socket::pendingOutput) )
 	{
-		ssize_t ret = udp->send( (char*)&(mypack.msg),sizeof(mypack.msg));
-		if( ret<sizeof(mypack.msg) )
+//		ssize_t ret = udp->send( (char*)&(mypack.msg),sizeof(mypack.msg));
+//		if( ret<sizeof(mypack.msg) )
+		ssize_t ret = udp->send( (char*)&(mypack.msg),sz);
+		if( ret < sz )
 		{
-			cerr << myname << "(send data header): ret=" << ret << " sizeof=" << sizeof(mypack.msg) << endl;
+			cerr << myname << "(send data header): ret=" << ret << " sizeof=" << sz << endl;
 			return;
 		}
 

extensions/UDPExchange/UDPPacket.cc

@@ -31,8 +31,9 @@ bool UDPMessage::addData( const UniSetUDP::UDPData& dat )
 	if( count >= MaxDataCount )
 		return false;
 
-	msg.dat[sizeof(UniSetUDP::UDPHeader)+count] = dat;
+	msg.dat[count] = dat;
 	count++;
+	msg.header.dcount = count;
 	return true;
 }
 // -----------------------------------------------------------------------------

extensions/UDPExchange/UDPPacket.h

@@ -11,10 +11,11 @@ namespace UniSetUDP
 {
 	struct UDPHeader
 	{
-		long num;
+		UDPHeader():num(0),nodeID(0),procID(0),dcount(0){}
+		unsigned long num;
 		long nodeID;
 		long procID;
-		long dcount;
+		size_t dcount;
 
 		friend std::ostream& operator<<( std::ostream& os, UDPHeader& p );
 	}__attribute__((packed));

extensions/UDPExchange/UDPReceiver.cc

@@ -7,6 +7,15 @@ using namespace std;
 using namespace UniSetTypes;
 using namespace UniSetExtensions;
 // -----------------------------------------------------------------------------
+bool UDPReceiver::PacketCompare::operator()(const UniSetUDP::UDPMessage& lhs,
+											const UniSetUDP::UDPMessage& rhs) const
+{
+//	if( lhs.msg.header.num == rhs.msg.header.num )
+//		return (lhs.msg < rhs.msg);
+
+	return lhs.msg.header.num > rhs.msg.header.num;
+}
+// ------------------------------------------------------------------------------------------
 UDPReceiver::UDPReceiver( UniSetTypes::ObjectId objId, UniSetTypes::ObjectId shmId, SharedMemory* ic ):
 UniSetObject_LT(objId),
 shm(0),
@@ -62,6 +71,8 @@ activated(false)
 	recvTimeout = conf->getArgPInt("--udp-recv-timeout",it.getProp("recvTimeout"), 5000);
 	polltime = conf->getArgPInt("--udp-polltime",it.getProp("polltime"), 100);
 
+	ptUpdate.setTiming(100);
+
 	// -------------------------------
 	// ********** HEARTBEAT *************
 	string heart = conf->getArgParam("--udp-heartbeat-id",it.getProp("heartbeat_id"));
@@ -131,24 +142,23 @@ void UDPReceiver::waitSMReady()
 	}
 }
 // -----------------------------------------------------------------------------
-/*
 void UDPReceiver::timerInfo( TimerMessage *tm )
 {
 	if( tm->id == tmExchange )
 		step();
 }
-*/
 // -----------------------------------------------------------------------------
 void UDPReceiver::step()
 {
-//	{
-//		uniset_mutex_lock l(pollMutex,2000);
-//		poll();
-//	}
-
 	if( !activated )
 		return;
 
+//	if(  ptUpdate.checkTime() )
+//	{
+		  update_data();
+		  ptUpdate.reset();
+//	}
+
 	if( sidHeartBeat!=DefaultObjectId && ptHeartBeat.checkTime() )
 	{
 		try
@@ -165,26 +175,56 @@ void UDPReceiver::step()
 }
 
 // -----------------------------------------------------------------------------
-void UDPReceiver::poll()
+void UDPReceiver::update_data()
 {
-	try
+	if( !activated )
+		return;
+
+	UniSetUDP::UDPMessage p;
+	bool buf_ok = false;
 	{
-//		udp->connect(host,port);
-//		udp->UDPSocket::setPeer(host,port);
+		uniset_mutex_lock l(packMutex);
+		if( qpack.size() <= max_buf_size )
+			return;
+
+		buf_ok = true;
 	}
-	catch( UniSetTypes::Exception& ex)
+
+	while( buf_ok )
 	{
-		cerr << myname << "(step): " << ex << std::endl;
-//		reise(SIGTERM);
-		return;
+		{
+			uniset_mutex_lock l(packMutex);
+			p = qpack.top();
+			qpack.pop();
+		}
+
+		if( labs(p.msg.header.num - pnum) > 1 )
+		{
+			cerr << "************ FAILED! ORDER PACKETS! recv.num=" << pack.msg.header.num
+				  << " num=" << pnum << endl;
+		}
+
+		pnum = p.msg.header.num;
+
+		{
+			uniset_mutex_lock l(packMutex);
+			buf_ok =  ( qpack.size() > max_buf_size );
 		}
 
+		cerr << myname << "(step): recv DATA OK. header: " << p.msg.header << endl;
+	}
+}
+
+// -----------------------------------------------------------------------------
+
+void UDPReceiver::poll()
+{
+	cerr << "******************* pool start" << endl;
 	while( activated )
 	{
 		try
 		{
 			recv();
-//			send();
 		}
 		catch( ost::SockException& e )
 		{
@@ -192,11 +232,11 @@ void UDPReceiver::poll()
 		}
 		catch( UniSetTypes::Exception& ex)
 		{
-			cerr << myname << "(step): " << ex << std::endl;
+			cerr << myname << "(poll): " << ex << std::endl;
 		}
 		catch(...)
 		{
-			cerr << myname << "(step): catch ..." << std::endl;
+			cerr << myname << "(poll): catch ..." << std::endl;
 		}
 
 		msleep(polltime);
@@ -207,75 +247,42 @@ void UDPReceiver::poll()
 // -----------------------------------------------------------------------------
 void UDPReceiver::recv()
 {
-	cout << myname << ": recv....(timeout=" << recvTimeout << ")" << endl;
-//	UniSetUDP::UDPHeader h;
-	// receive
-	UniSetUDP::UDPMessage pack;
 	if( udp->isInputReady(recvTimeout) )
 	{
 		ssize_t ret = udp->UDPReceive::receive(&(pack.msg),sizeof(pack.msg));
-		if( ret<(ssize_t)sizeof(pack.msg) )
+		if( ret < sizeof(UniSetUDP::UDPHeader) )
 		{
-			cerr << myname << "(receive): FAILED ret=" << ret << " sizeof=" << sizeof(pack.msg) << endl;
+			cerr << myname << "(receive): FAILED header ret=" << ret << " sizeof=" << sizeof(UniSetUDP::UDPHeader) << endl;
 			return;
 		}
 
-		cerr << myname << "(receive): OK. ret=" << ret << " sizeof=" << sizeof(pack.msg)
-			  << " header: " << pack.msg.header << endl;
-	}
-/*
-	cout << myname << ": recv....(timeout=" << recvTimeout << ")" << endl;
-	UniSetUDP::UDPHeader h;
-	// receive
-	if( udp->isInputReady(recvTimeout) )
+		ssize_t sz = pack.msg.header.dcount * sizeof(UniSetUDP::UDPData) + sizeof(UniSetUDP::UDPHeader);
+		if( ret < sz )
 		{
-		ssize_t ret = udp->UDPReceive::receive(&h,sizeof(h));
-		if( ret<(ssize_t)sizeof(h) )
-		{
-			cerr << myname << "(receive): ret=" << ret << " sizeof=" << sizeof(h) << endl;
+			cerr << myname << "(receive): FAILED data ret=" << ret << " sizeof=" << sz << endl;
 			return;
 		}
 
-		cout << myname << "(receive): header: " << h << endl;
-		if( h.dcount <=0 )
+
+//		cerr << myname << "(receive): recv DATA OK. ret=" << ret << " sizeof=" << sz
+//			  << " header: " << pack.msg.header << endl;
+/*
+		if( labs(pack.msg.header.num - pnum) > 1 )
 		{
-			cout << " data=0" << endl;
-			return;
+			cerr << "************ FAILED! ORDER PACKETS! recv.num=" << pack.msg.header.num
+					<< " num=" << pnum << endl;
 		}
 
-		UniSetUDP::UDPData d;
-		// ignore echo...
+		pnum = pack.msg.header.num;
 */
-#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
-#if 0
-		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;
+			uniset_mutex_lock l(packMutex);
+//			qpack[pack.msg.header.num] = pack;
+			qpack.push(pack);
 		}
 
-			cout << myname << "(receive data " << i << "): " << d << endl;
-		}
+		return;
 	}
-//	else
-//	{
-//		cout << "no InputReady.." << endl;
-//	}
-#endif
 }
 // -----------------------------------------------------------------------------
 void UDPReceiver::processingMessage(UniSetTypes::VoidMessage *msg)
@@ -298,6 +305,13 @@ void UDPReceiver::processingMessage(UniSetTypes::VoidMessage *msg)
 			}
 			break;
 
+			case Message::Timer:
+			{
+				TimerMessage tm(msg);
+				timerInfo(&tm);
+			}
+			break;
+
 			default:
 				break;
 		}
@@ -318,7 +332,7 @@ void UDPReceiver::processingMessage(UniSetTypes::VoidMessage *msg)
 	}
 }
 // -----------------------------------------------------------------------------
-void UDPReceiver::sysCommand(UniSetTypes::SystemMessage *sm)
+void UDPReceiver::sysCommand( UniSetTypes::SystemMessage *sm )
 {
 	switch( sm->command )
 	{
@@ -346,6 +360,7 @@ void UDPReceiver::sysCommand(UniSetTypes::SystemMessage *sm)
 				askSensors(UniversalIO::UIONotify);
 			}
 			thr->start();
+			askTimer(tmExchange,1000);
 		}
 
 		case SystemMessage::FoldUp:

extensions/UDPExchange/UDPReceiver.h

@@ -3,7 +3,8 @@
 // -----------------------------------------------------------------------------
 #include <ostream>
 #include <string>
-#include <vector>
+#include <map>
+#include <queue>
 #include <cc++/socket.h>
 #include "UniSetObject_LT.h"
 #include "Trigger.h"
@@ -26,7 +27,6 @@ class UDPReceiver:
 
 		/*! глобальная функция для вывода help-а */
 		static void help_print( int argc, char* argv[] );
-
 	protected:
 
 		xmlNode* cnode;
@@ -37,11 +37,13 @@ class UDPReceiver:
 
 		void poll();
 		void recv();
-		void step();
+		virtual void step();
+		void update_data();
 
 		virtual void processingMessage( UniSetTypes::VoidMessage *msg );
 		void sysCommand( UniSetTypes::SystemMessage *msg );
 		void sensorInfo( UniSetTypes::SensorMessage*sm );
+		void timerInfo( UniSetTypes::TimerMessage *tm );
 		void askSensors( UniversalIO::UIOCommand cmd );
 		void waitSMReady();
 
@@ -52,6 +54,11 @@ class UDPReceiver:
 
 		void initIterators();
 
+		enum Timer
+		{
+			tmExchange
+		};
+
 	private:
 		UDPReceiver();
 		bool initPause;
@@ -76,7 +83,27 @@ class UDPReceiver:
 		bool activated;
 		int activateTimeout;
 
+		long pnum;
+
 		ThreadCreator<UDPReceiver>* thr;
+
+//		typedef std::map<unsigned long,UniSetUDP::UDPMessage> QueuePacket;
+//		QueuePacket qpack;
+		UniSetUDP::UDPMessage pack;
+		UniSetTypes::uniset_mutex packMutex;
+
+		// функция определения приоритетного сообщения для обработки
+		struct PacketCompare:
+		public std::binary_function<UniSetUDP::UDPMessage, UniSetUDP::UDPMessage, bool>
+		{
+			bool operator()(const UniSetUDP::UDPMessage& lhs,
+							const UniSetUDP::UDPMessage& rhs) const;
+		};
+		typedef std::priority_queue<UniSetUDP::UDPMessage,std::vector<UniSetUDP::UDPMessage>,PacketCompare> PacketQueue;
+		PacketQueue qpack;
+		static const int max_buf_size = 20;
+
+		PassiveTimer ptUpdate;
 };
 // -----------------------------------------------------------------------------
 #endif // UDPReceiver_H_