[Переход на libPoco]: Modbus{Master,Slave} добился прохождения тестов

Utilities/MBTester/MBSlave.h

@@ -35,7 +35,6 @@ class MBSlave
 
 		void execute();    /*!< основной цикл работы */
 
-
 		void setLog( std::shared_ptr<DebugStream> dlog );
 
 	protected:

Utilities/MBTester/MBTCPServer.cc

@@ -68,6 +68,11 @@ void MBTCPServer::setLog(std::shared_ptr<DebugStream>& dlog )
 		sslot->setLog(dlog);
 }
 // -------------------------------------------------------------------------
+void MBTCPServer::setMaxSessions( size_t max )
+{
+	sslot->setMaxSessions(max);
+}
+// -------------------------------------------------------------------------
 void MBTCPServer::execute()
 {
 	sslot->run( vaddr, false );

Utilities/MBTester/MBTCPServer.h

@@ -31,6 +31,8 @@ class MBTCPServer
 		void execute();    /*!< основной цикл работы */
 		void setLog( std::shared_ptr<DebugStream>& dlog );
 
+		void setMaxSessions( size_t max );
+
 	protected:
 		// действия при завершении работы
 		void sigterm( int signo );

Utilities/MBTester/mbtcpserver.cc

@@ -16,6 +16,7 @@ static struct option longopts[] =
 	{ "port", required_argument, 0, 'p' },
 	{ "const-reply", required_argument, 0, 'c' },
 	{ "after-send-pause", required_argument, 0, 's' },
+	{ "max-sessions", required_argument, 0, 'm' },
 	{ NULL, 0, 0, 0 }
 };
 // --------------------------------------------------------------------------
@@ -30,6 +31,7 @@ static void print_help()
 	printf("[-p|--port] port               - Server port. Default: 502.\n");
 	printf("[-c|--const-reply] val         - Reply 'val' for all queries\n");
 	printf("[-s|--after-send-pause] msec   - Pause after send request. Default: 0\n");
+	printf("[-m|--max-sessions] num        - Set the maximum number of sessions. Default: 10\n");
 }
 // --------------------------------------------------------------------------
 int main( int argc, char** argv )
@@ -44,12 +46,13 @@ int main( int argc, char** argv )
 	auto dlog = make_shared<DebugStream>();
 	int replyVal = -1;
 	timeout_t afterpause = 0;
+	size_t maxSessions = 10;
 
 	try
 	{
 		while(1)
 		{
-			opt = getopt_long(argc, argv, "hva:p:i:c:s:", longopts, &optindex);
+			opt = getopt_long(argc, argv, "hva:p:i:c:s:m:", longopts, &optindex);
 
 			if( opt == -1 )
 				break;
@@ -84,6 +87,10 @@ int main( int argc, char** argv )
 					afterpause = uni_atoi(optarg);
 					break;
 
+				case 'm':
+					maxSessions = uni_atoi(optarg);
+					break;
+
 				case '?':
 				default:
 					printf("? argumnet\n");
@@ -110,6 +117,7 @@ int main( int argc, char** argv )
 		mbs.setLog(dlog);
 		mbs.setVerbose(verb);
 		mbs.setAfterSendPause(afterpause);
+		mbs.setMaxSessions(maxSessions);
 
 		if( replyVal != -1 )
 			mbs.setReply(replyVal);

conf/test.xml

@@ -92,7 +92,7 @@
 			<DeviceList>
 			</DeviceList>
 			<GateList>
-				<item ip="192.168.3.12" port="2048" timeout="3000" recv_timeout="1000"/>
+				<item ip="localhost" port="2048" timeout="3000" recv_timeout="1000"/>
 				<item ip="192.168.3.12" port="2049" timeout="3000" recv_timeout="1000"/>
 			</GateList>
 		</MBPerfTestMaster>

extensions/ModbusMaster/MBTCPMaster.cc

@@ -220,7 +220,20 @@ void MBTCPMaster::sigterm( int signo )
 		std::clog << (p ? p.__cxa_exception_type()->name() : "null") << std::endl;
 	}
 }
+// -----------------------------------------------------------------------------
+bool MBTCPMaster::deactivateObject()
+{
+	setProcActive(false);
+	if( pollThread )
+	{
+		pollThread->stop();
 
+		if( pollThread->isRunning() )
+			pollThread->join();
+	}
+
+	return MBExchange::deactivateObject();
+}
 // -----------------------------------------------------------------------------
 void MBTCPMaster::help_print( int argc, const char* const* argv )
 {

extensions/ModbusMaster/MBTCPMaster.h

@@ -237,6 +237,7 @@ class MBTCPMaster:
 		virtual void sysCommand( const UniSetTypes::SystemMessage* sm ) override;
 		virtual std::shared_ptr<ModbusClient> initMB( bool reopen = false ) override;
 		virtual void sigterm( int signo ) override;
+		virtual bool deactivateObject() override;
 
 		std::string iaddr;
 		int port;

extensions/ModbusMaster/MBTCPMultiMaster.cc

@@ -335,6 +335,9 @@ bool MBTCPMultiMaster::MBSlaveInfo::init( std::shared_ptr<DebugStream>& mblog )
 
 			initOK = true;
 		}
+
+		mbinfo << myname << "(init): connect " << mbtcp->isConnection() << endl;
+
 		return mbtcp->isConnection();
 	}
 	catch( ModbusRTU::mbException& ex )
@@ -507,6 +510,28 @@ void MBTCPMultiMaster::sigterm( int signo )
 	//		std::clog << (p ? p.__cxa_exception_type()->name() : "null") << std::endl;
 	//	}
 }
+// -----------------------------------------------------------------------------
+bool MBTCPMultiMaster::deactivateObject()
+{
+	setProcActive(false);
+	if( pollThread )
+	{
+		pollThread->stop();
+
+		if( pollThread->isRunning() )
+			pollThread->join();
+	}
+
+	if( checkThread )
+	{
+		checkThread->stop();
+
+		if( checkThread->isRunning() )
+			checkThread->join();
+	}
+
+	return MBExchange::deactivateObject();
+}
 
 // -----------------------------------------------------------------------------
 void MBTCPMultiMaster::help_print( int argc, const char* const* argv )

extensions/ModbusMaster/MBTCPMultiMaster.h

@@ -262,6 +262,7 @@ class MBTCPMultiMaster:
 		virtual void initIterators() override;
 		virtual std::shared_ptr<ModbusClient> initMB( bool reopen = false ) override;
 		virtual void sigterm( int signo ) override;
+		virtual bool deactivateObject() override;
 
 		void poll_thread();
 		void check_thread();

extensions/ModbusMaster/mb-perf-test.cc

@@ -58,6 +58,8 @@ int main( int argc, const char** argv )
 
 		int count = conf->getArgPInt("--count", 50);
 
+		cerr << "RUN " << count << " MBTCPMultiMaster.." << endl;
+
 		for( int i = 1; i <= count; i++ )
 		{
 			ostringstream prefix;

include/modbus/ModbusTCPSession.h

@@ -9,8 +9,8 @@
 #include "ModbusServerSlot.h"
 #include "ModbusServer.h"
 #include "PassiveTimer.h"
-#include "USocket.h"
 #include "UTCPCore.h"
+#include "UTCPStream.h"
 // -------------------------------------------------------------------------
 /*!
  * \brief The ModbusTCPSession class
@@ -33,7 +33,7 @@ class ModbusTCPSession:
 {
 	public:
 
-		ModbusTCPSession( int sock, const std::unordered_set<ModbusRTU::ModbusAddr>& vmbaddr, timeout_t timeout );
+		ModbusTCPSession( const Poco::Net::StreamSocket& s, const std::unordered_set<ModbusRTU::ModbusAddr>& vmbaddr, timeout_t timeout );
 		virtual ~ModbusTCPSession();
 
 		void cleanInputStream();
@@ -129,7 +129,7 @@ class ModbusTCPSession:
 		ev::io  io;
 		ev::timer ioTimeout;
 
-		std::shared_ptr<USocket> sock;
+		std::shared_ptr<UTCPStream> sock;
 		std::queue<UTCPCore::Buffer*> qsend;
 		double sessTimeout = { 10.0 };
 

src/Communications/Modbus/ModbusTCPMaster.cc

@@ -35,11 +35,8 @@ ModbusTCPMaster::ModbusTCPMaster():
 	force_disconnect(true)
 {
 	setCRCNoCheckit(true);
-	/*
-	    dlog->addLevel(Debug::INFO);
-	    dlog->addLevel(Debug::WARN);
-	    dlog->addLevel(Debug::CRIT);
-	*/
+
+//	 dlog->level(Debug::ANY);
 }
 
 // -------------------------------------------------------------------------
@@ -64,10 +61,12 @@ void ModbusTCPMaster::setChannelTimeout( timeout_t msec )
 	Poco::Timespan old = tcp->getReceiveTimeout();;
 	//timeout_t old = tcp->getReceiveTimeout();
 
+	Poco::Timespan tmsec(msec*1000);
+
 	if( old == msec )
 		return;
 
-	tcp->setReceiveTimeout(msec*1000);
+	tcp->setReceiveTimeout(tmsec);
 
 	int oldKeepAlive = keepAliveTimeout;
 	keepAliveTimeout = (msec > 1000 ? msec / 1000 : 1);
@@ -122,7 +121,7 @@ mbErrCode ModbusTCPMaster::query( ModbusAddr addr, ModbusMessage& msg,
 		for( size_t i = 0; i < 2; i++ )
 		{
 			//if( tcp->isPending(ost::Socket::pendingOutput, timeout) )
-			if( tcp->poll(timeout*1000,Poco::Net::Socket::SELECT_READ ) )
+			if( tcp->poll(timeout*1000,Poco::Net::Socket::SELECT_WRITE) )
 			{
 				mbErrCode res = send(msg);
 
@@ -266,7 +265,7 @@ mbErrCode ModbusTCPMaster::query( ModbusAddr addr, ModbusMessage& msg,
 				// при штатном обмене..лучше дождаться конца "посылки"..
 				// поэтому применяем disconnect(), а не forceDisconnect()
 				// (с учётом выставленной опции setLinger(true))
-				tcp->shutdown();
+				tcp->close();
 			}
 
 			return res;
@@ -360,7 +359,8 @@ bool ModbusTCPMaster::checkConnection( const std::string& ip, int port, int time
 		t.create(ip, port, timeout_msec);
 		t.setKeepAliveParams( (timeout_msec > 1000 ? timeout_msec / 1000 : 1), 1, 1);
 		t.setNoDelay(true);
-		t.shutdown();
+		//t.shutdown();
+		t.close();
 		return true;
 	}
 	catch(...)
@@ -419,7 +419,8 @@ void ModbusTCPMaster::connect( const Poco::Net::SocketAddress& addr, int _port )
 {
 	if( tcp )
 	{
-		disconnect();
+		//disconnect();
+		forceDisconnect();
 		tcp.reset();
 	}
 
@@ -432,31 +433,43 @@ void ModbusTCPMaster::connect( const Poco::Net::SocketAddress& addr, int _port )
 	try
 	{
 		tcp = make_shared<UTCPStream>();
-		tcp->connect(addr,500);
+		tcp->create(iaddr,port,500);
+		//tcp->connect(addr,500);
 		tcp->setReceiveTimeout(replyTimeOut_ms*1000);
 		tcp->setKeepAlive(true); // tcp->setKeepAliveParams((replyTimeOut_ms > 1000 ? replyTimeOut_ms / 1000 : 1));
 		tcp->setNoDelay(true);
 	}
+	catch( Poco::Net::NetException& ex)
+	{
+		if( dlog->debugging(Debug::CRIT) )
+		{
+			ostringstream s;
+			s << "(ModbusTCPMaster): create connection " << iaddr << ":" << port << " error: " << ex.displayText();
+			dlog->crit() << iaddr << std::endl;
+		}
+
+		tcp = nullptr;
+	}
 	catch( const std::exception& e )
 	{
 		if( dlog->debugging(Debug::CRIT) )
 		{
 			ostringstream s;
-			s << "(ModbusTCPMaster): connection " << s.str() << " error: " << e.what();
+			s << "(ModbusTCPMaster): connection " << iaddr << ":" << port << " error: " << e.what();
 			dlog->crit() << iaddr << std::endl;
 		}
+		tcp = nullptr;
 	}
 	catch( ... )
 	{
 		if( dlog->debugging(Debug::CRIT) )
 		{
 			ostringstream s;
-			s << "(ModbusTCPMaster): connection " << s.str() << " error: catch ...";
+			s << "(ModbusTCPMaster): connection " << iaddr << ":" << port << " error: catch ...";
 			dlog->crit() << s.str() << std::endl;
 		}
+		tcp = nullptr;
 	}
-
-	//    }
 }
 // -------------------------------------------------------------------------
 void ModbusTCPMaster::disconnect()
@@ -467,7 +480,7 @@ void ModbusTCPMaster::disconnect()
 	if( !tcp )
 		return;
 
-	tcp->shutdown();
+	tcp->close();
 	tcp.reset();
 }
 // -------------------------------------------------------------------------
@@ -480,11 +493,20 @@ void ModbusTCPMaster::forceDisconnect()
 		return;
 
 	tcp->forceDisconnect();
-	tcp.reset();
+	tcp = nullptr;
 }
 // -------------------------------------------------------------------------
 bool ModbusTCPMaster::isConnection() const
 {
 	return tcp && tcp->isConnected();
+#if 0
+	if( !tcp )
+		return false;
+
+	if( tcp->poll({0,5},Poco::Net::Socket::SELECT_READ) )
+		return (tcp->available() > 0);
+
+	return false;
+#endif
 }
 // -------------------------------------------------------------------------

src/Communications/Modbus/ModbusTCPServer.cc

@@ -245,7 +245,9 @@ void ModbusTCPServer::ioAccept(ev::io& watcher, int revents)
 
 	try
 	{
-		auto s = make_shared<ModbusTCPSession>(watcher.fd, *vmbaddr, sessTimeout);
+		Poco::Net::StreamSocket ss = sock->acceptConnection();
+
+		auto s = make_shared<ModbusTCPSession>(ss, *vmbaddr, sessTimeout);
 		s->connectReadCoil( sigc::mem_fun(this, &ModbusTCPServer::readCoilStatus) );
 		s->connectReadInputStatus( sigc::mem_fun(this, &ModbusTCPServer::readInputStatus) );
 		s->connectReadOutput( sigc::mem_fun(this, &ModbusTCPServer::readOutputRegisters) );

src/Communications/Modbus/ModbusTCPSession.cc

@@ -42,7 +42,7 @@ ModbusTCPSession::~ModbusTCPSession()
 		ioTimeout.stop();
 }
 // -------------------------------------------------------------------------
-ModbusTCPSession::ModbusTCPSession( int sfd, const std::unordered_set<ModbusAddr>& a, timeout_t timeout ):
+ModbusTCPSession::ModbusTCPSession(const Poco::Net::StreamSocket& s, const std::unordered_set<ModbusAddr>& a, timeout_t timeout ):
 	vaddr(a),
 	timeout(timeout),
 	peername(""),
@@ -51,7 +51,7 @@ ModbusTCPSession::ModbusTCPSession( int sfd, const std::unordered_set<ModbusAddr
 {
 	try
 	{
-		sock = make_shared<USocket>(sfd);
+		sock = make_shared<UTCPStream>(s);
 
 		// если стремиться к "оптимизации по скорости"
 		// то getpeername "медленная" операция и может стоит от неё отказаться.

src/Communications/TCP/TCPCheck.cc

@@ -76,7 +76,7 @@ void TCPCheck::check_thread()
 		t.create(ip, port, tout_msec);
 		t.setKeepAliveParams( (tout_msec > 1000 ? tout_msec / 1000 : 1) );
 		setResult(true);
-		t.shutdown();
+		t.close();
 	}
 	catch( ... ) {}
 }

src/Communications/TCP/UTCPSocket.cc

@@ -42,7 +42,7 @@ UTCPSocket::UTCPSocket( int sock ):
 }
 // -------------------------------------------------------------------------
 UTCPSocket::UTCPSocket( const string& host, int port ):
-	Poco::Net::ServerSocket(Poco::Net::SocketAddress(host,port))
+	Poco::Net::ServerSocket(Poco::Net::SocketAddress(host,port),true)
 {
 	init();
 }

src/Communications/TCP/UTCPStream.cc

@@ -19,6 +19,7 @@
 #include <fcntl.h>
 #include <errno.h>
 #include <cstring>
+#include <Poco/Net/NetException.h>
 #include "UTCPStream.h"
 #include "PassiveTimer.h"
 #include "UniSetTypes.h"
@@ -56,12 +57,20 @@ bool UTCPStream::isSetLinger() const
 // -------------------------------------------------------------------------
 void UTCPStream::forceDisconnect()
 {
+	try
+	{
 		bool on;
 		int sec;
 		Poco::Net::StreamSocket::getLinger(on,sec);
 		setLinger(false,0);
-	shutdown();
+		close();
+		//shutdown();
 		Poco::Net::StreamSocket::setLinger(on,sec);
+	}
+	catch( Poco::Net::NetException& )
+	{
+
+	}
 }
 // -------------------------------------------------------------------------
 bool UTCPStream::setNoDelay(bool enable)
@@ -103,6 +112,13 @@ void UTCPStream::create(const std::string& hname, int port, timeout_t tout_msec
 bool UTCPStream::isConnected()
 {
 	//return ( Poco::Net::StreamSocket::sockfd() > 0 );
+	try
+	{
 		return ( Poco::Net::StreamSocket::peerAddress().addr() != 0 );
+	}
+	catch( Poco::Net::NetException& ex )
+	{
+	}
+	return false;
 }
 // -------------------------------------------------------------------------