Commit ef6c70aa authored by Pavel Vainerman's avatar Pavel Vainerman

(MBTCPMaster): Реализация с отдельным потоком обмена.

parent f47ea61b
...@@ -23,10 +23,10 @@ mbregFromID(false), ...@@ -23,10 +23,10 @@ mbregFromID(false),
activated(false), activated(false),
noQueryOptimization(false), noQueryOptimization(false),
force_disconnect(true), force_disconnect(true),
allNotRespond(false),
prefix(prefix), prefix(prefix),
no_extimer(false), no_extimer(false),
poll_count(0) poll_count(0),
pollThread(0)
{ {
// cout << "$ $" << endl; // cout << "$ $" << endl;
...@@ -66,10 +66,10 @@ poll_count(0) ...@@ -66,10 +66,10 @@ poll_count(0)
throw UniSetTypes::SystemError(myname+"(MBMaster): Unknown inet port...(Use: " + tmp +")" ); throw UniSetTypes::SystemError(myname+"(MBMaster): Unknown inet port...(Use: " + tmp +")" );
recv_timeout = conf->getArgPInt("--" + prefix + "-recv-timeout",it.getProp("recv_timeout"), 50); recv_timeout = conf->getArgPInt("--" + prefix + "-recv-timeout",it.getProp("recv_timeout"), 500);
int alltout = conf->getArgPInt("--" + prefix + "-all-timeout",it.getProp("all_timeout"), 2000); int tout = conf->getArgPInt("--" + prefix + "-timeout",it.getProp("timeout"), 5000);
ptAllNotRespond.setTiming(alltout); ptTimeout.setTiming(tout);
noQueryOptimization = conf->getArgInt("--" + prefix + "-no-query-optimization",it.getProp("no_query_optimization")); noQueryOptimization = conf->getArgInt("--" + prefix + "-no-query-optimization",it.getProp("no_query_optimization"));
...@@ -84,6 +84,7 @@ poll_count(0) ...@@ -84,6 +84,7 @@ poll_count(0)
force_out = conf->getArgInt("--" + prefix + "-force-out",it.getProp("force_out")); force_out = conf->getArgInt("--" + prefix + "-force-out",it.getProp("force_out"));
force_disconnect = conf->getArgInt("--" + prefix + "-persistent-connection",it.getProp("persistent_connection")) ? false : true; force_disconnect = conf->getArgInt("--" + prefix + "-persistent-connection",it.getProp("persistent_connection")) ? false : true;
dlog[Debug::INFO] << myname << "(init): persisten-connection=" << (!force_disconnect) << endl;
if( shm->isLocalwork() ) if( shm->isLocalwork() )
{ {
...@@ -138,6 +139,8 @@ poll_count(0) ...@@ -138,6 +139,8 @@ poll_count(0)
// abort(); // abort();
poll_count = -1; poll_count = -1;
pollThread = new ThreadCreator<MBTCPMaster>(this, &MBTCPMaster::poll_thread);
} }
// ----------------------------------------------------------------------------- // -----------------------------------------------------------------------------
MBTCPMaster::~MBTCPMaster() MBTCPMaster::~MBTCPMaster()
...@@ -151,6 +154,7 @@ MBTCPMaster::~MBTCPMaster() ...@@ -151,6 +154,7 @@ MBTCPMaster::~MBTCPMaster()
delete it1->second; delete it1->second;
} }
delete pollThread;
delete mb; delete mb;
delete shm; delete shm;
} }
...@@ -193,6 +197,8 @@ void MBTCPMaster::initMB( bool reopen ) ...@@ -193,6 +197,8 @@ void MBTCPMaster::initMB( bool reopen )
delete mb; delete mb;
mb = 0; mb = 0;
} }
} }
// ----------------------------------------------------------------------------- // -----------------------------------------------------------------------------
void MBTCPMaster::waitSMReady() void MBTCPMaster::waitSMReady()
...@@ -226,10 +232,7 @@ void MBTCPMaster::timerInfo( TimerMessage *tm ) ...@@ -226,10 +232,7 @@ void MBTCPMaster::timerInfo( TimerMessage *tm )
// ----------------------------------------------------------------------------- // -----------------------------------------------------------------------------
void MBTCPMaster::step() void MBTCPMaster::step()
{ {
{ updateRespondSensors();
uniset_mutex_lock l(pollMutex,2000);
poll();
}
if( !activated ) if( !activated )
return; return;
...@@ -250,33 +253,121 @@ void MBTCPMaster::step() ...@@ -250,33 +253,121 @@ void MBTCPMaster::step()
} }
// ----------------------------------------------------------------------------- // -----------------------------------------------------------------------------
void MBTCPMaster::poll() void MBTCPMaster::updateRespondSensors()
{
bool tcpIsTimeout = false;
{
uniset_mutex_lock l(tcpMutex);
tcpIsTimeout = pollActivated && ptTimeout.checkTime();
}
if( dlog.debugging(Debug::LEVEL4) )
dlog[Debug::LEVEL4] << myname << ": tcpTimeout=" << tcpIsTimeout << endl;
for( MBTCPMaster::RTUDeviceMap::iterator it1=rmap.begin(); it1!=rmap.end(); ++it1 )
{
RTUDevice* d(it1->second);
if( tcpIsTimeout )
d->resp_real = false;
if( dlog.debugging(Debug::LEVEL4) )
{
dlog[Debug::LEVEL4] << myname << ": check respond addr=" << ModbusRTU::addr2str(d->mbaddr)
<< " respond_id=" << d->resp_id
<< " real=" << d->resp_real
<< " state=" << d->resp_state
<< endl;
}
if( d->checkRespond() && d->resp_id != DefaultObjectId )
{
try
{
bool set = d->resp_invert ? !d->resp_state : d->resp_state;
shm->localSaveState(d->resp_dit,d->resp_id,set,getId());
}
catch( Exception& ex )
{
dlog[Debug::CRIT] << myname << "(step): (respond) " << ex << std::endl;
}
}
}
}
// -----------------------------------------------------------------------------
void MBTCPMaster::poll_thread()
{ {
if( trAllNotRespond.hi(allNotRespond) ) cerr << "*********** polling starting.." << endl;
ptAllNotRespond.reset();
{
uniset_mutex_lock l(pollMutex,300);
ptTimeout.reset();
}
if( allNotRespond && mb && ptAllNotRespond.checkTime() ) while( checkProcActive() )
{ {
ptAllNotRespond.reset(); try
// initMB(true); {
poll();
} }
catch(...){}
if( !checkProcActive() )
break;
msleep(polltime);
}
cerr << "*********** polling finished.." << endl;
}
// -----------------------------------------------------------------------------
bool MBTCPMaster::checkProcActive()
{
uniset_mutex_lock l(actMutex, 300);
return activated;
}
// -----------------------------------------------------------------------------
void MBTCPMaster::setProcActive( bool st )
{
uniset_mutex_lock l(actMutex, 400);
activated = st;
}
// -----------------------------------------------------------------------------
void MBTCPMaster::poll()
{
if( !mb ) if( !mb )
{ {
{
uniset_mutex_lock l(pollMutex, 300);
pollActivated = false;
initMB(false); initMB(false);
if( !mb ) if( !mb )
{ {
for( MBTCPMaster::RTUDeviceMap::iterator it=rmap.begin(); it!=rmap.end(); ++it ) for( MBTCPMaster::RTUDeviceMap::iterator it=rmap.begin(); it!=rmap.end(); ++it )
it->second->resp_real = false; it->second->resp_real = false;
} }
}
if( !checkProcActive() )
return;
updateSM(); updateSM();
allInitOK = false; allInitOK = false;
return; return;
} }
{
uniset_mutex_lock l(pollMutex);
pollActivated = true;
ptTimeout.reset();
}
if( !allInitOK ) if( !allInitOK )
firstInitRegisters(); firstInitRegisters();
if( !checkProcActive() )
return;
for( MBTCPMaster::RTUDeviceMap::iterator it1=rmap.begin(); it1!=rmap.end(); ++it1 ) for( MBTCPMaster::RTUDeviceMap::iterator it1=rmap.begin(); it1!=rmap.end(); ++it1 )
{ {
RTUDevice* d(it1->second); RTUDevice* d(it1->second);
...@@ -288,6 +379,9 @@ void MBTCPMaster::poll() ...@@ -288,6 +379,9 @@ void MBTCPMaster::poll()
d->resp_real = false; d->resp_real = false;
for( MBTCPMaster::RegMap::iterator it=d->regmap.begin(); it!=d->regmap.end(); ++it ) for( MBTCPMaster::RegMap::iterator it=d->regmap.begin(); it!=d->regmap.end(); ++it )
{ {
if( !checkProcActive() )
return;
try try
{ {
if( d->dtype==MBTCPMaster::dtRTU || d->dtype==MBTCPMaster::dtMTR ) if( d->dtype==MBTCPMaster::dtRTU || d->dtype==MBTCPMaster::dtMTR )
...@@ -320,6 +414,9 @@ void MBTCPMaster::poll() ...@@ -320,6 +414,9 @@ void MBTCPMaster::poll()
if( it==d->regmap.end() ) if( it==d->regmap.end() )
break; break;
if( !checkProcActive() )
return;
} }
if( stat_time > 0 ) if( stat_time > 0 )
...@@ -336,6 +433,14 @@ void MBTCPMaster::poll() ...@@ -336,6 +433,14 @@ void MBTCPMaster::poll()
// mb->disconnect(); // mb->disconnect();
} }
{
uniset_mutex_lock l(pollMutex);
pollActivated = false;
}
if( !checkProcActive() )
return;
// update SharedMemory... // update SharedMemory...
updateSM(); updateSM();
...@@ -345,6 +450,9 @@ void MBTCPMaster::poll() ...@@ -345,6 +450,9 @@ void MBTCPMaster::poll()
RTUDevice* d(it1->second); RTUDevice* d(it1->second);
for( MBTCPMaster::RegMap::iterator it=d->regmap.begin(); it!=d->regmap.end(); ++it ) for( MBTCPMaster::RegMap::iterator it=d->regmap.begin(); it!=d->regmap.end(); ++it )
{ {
if( !checkProcActive() )
return;
RegInfo* r(it->second); RegInfo* r(it->second);
for( PList::iterator i=r->slst.begin(); i!=r->slst.end(); ++i ) for( PList::iterator i=r->slst.begin(); i!=r->slst.end(); ++i )
IOBase::processingThreshold( &(*i),shm,force); IOBase::processingThreshold( &(*i),shm,force);
...@@ -777,6 +885,13 @@ bool MBTCPMaster::initSMValue( ModbusRTU::ModbusData* data, int count, RSPropert ...@@ -777,6 +885,13 @@ bool MBTCPMaster::initSMValue( ModbusRTU::ModbusData* data, int count, RSPropert
bool MBTCPMaster::RTUDevice::checkRespond() bool MBTCPMaster::RTUDevice::checkRespond()
{ {
bool prev = resp_state; bool prev = resp_state;
if( resp_ptTimeout.getInterval() <= 0 )
{
resp_state = resp_real;
return (prev != resp_state);
}
if( resp_trTimeout.hi(resp_real) ) if( resp_trTimeout.hi(resp_real) )
{ {
if( resp_real ) if( resp_real )
...@@ -803,38 +918,13 @@ bool MBTCPMaster::RTUDevice::checkRespond() ...@@ -803,38 +918,13 @@ bool MBTCPMaster::RTUDevice::checkRespond()
// ----------------------------------------------------------------------------- // -----------------------------------------------------------------------------
void MBTCPMaster::updateSM() void MBTCPMaster::updateSM()
{ {
allNotRespond = true;
for( MBTCPMaster::RTUDeviceMap::iterator it1=rmap.begin(); it1!=rmap.end(); ++it1 ) for( MBTCPMaster::RTUDeviceMap::iterator it1=rmap.begin(); it1!=rmap.end(); ++it1 )
{ {
RTUDevice* d(it1->second); RTUDevice* d(it1->second);
if( dlog.debugging(Debug::LEVEL4) ) // обновление датчиков связи происходит в другом потоке
{ // чтобы не зависеть от TCP таймаутов
dlog[Debug::LEVEL4] << "check respond addr=" << ModbusRTU::addr2str(d->mbaddr) // см. updateRespondSensors()
<< " respond_id=" << d->resp_id
<< " real=" << d->resp_real
<< " state=" << d->resp_state
<< endl;
}
if( d->resp_real )
allNotRespond = false;
// update respond sensors...
if( d->checkRespond() && d->resp_id != DefaultObjectId )
{
try
{
bool set = d->resp_invert ? !d->resp_state : d->resp_state;
shm->localSaveState(d->resp_dit,d->resp_id,set,getId());
}
catch(Exception& ex)
{
dlog[Debug::CRIT] << myname
<< "(step): (respond) " << ex << std::endl;
}
}
// cerr << "*********** allNotRespond=" << allNotRespond << endl;
// update values... // update values...
for( MBTCPMaster::RegMap::iterator it=d->regmap.begin(); it!=d->regmap.end(); ++it ) for( MBTCPMaster::RegMap::iterator it=d->regmap.begin(); it!=d->regmap.end(); ++it )
...@@ -967,15 +1057,8 @@ void MBTCPMaster::sysCommand( UniSetTypes::SystemMessage *sm ) ...@@ -967,15 +1057,8 @@ void MBTCPMaster::sysCommand( UniSetTypes::SystemMessage *sm )
initOutput(); initOutput();
} }
// начальная инициализация
if( !force )
{
uniset_mutex_lock l(pollMutex,2000);
force = true;
poll();
force = false;
}
askTimer(tmExchange,polltime); askTimer(tmExchange,polltime);
pollThread->start();
break; break;
} }
...@@ -1123,13 +1206,13 @@ bool MBTCPMaster::activateObject() ...@@ -1123,13 +1206,13 @@ bool MBTCPMaster::activateObject()
// пока не пройдёт инициализация датчиков // пока не пройдёт инициализация датчиков
// см. sysCommand() // см. sysCommand()
{ {
activated = false; setProcActive(false);
UniSetTypes::uniset_mutex_lock l(mutex_start, 5000); UniSetTypes::uniset_mutex_lock l(mutex_start, 5000);
UniSetObject_LT::activateObject(); UniSetObject_LT::activateObject();
if( !shm->isLocalwork() ) if( !shm->isLocalwork() )
rtuQueryOptimization(rmap); rtuQueryOptimization(rmap);
initIterators(); initIterators();
activated = true; setProcActive(true);
} }
return true; return true;
...@@ -1138,7 +1221,7 @@ bool MBTCPMaster::activateObject() ...@@ -1138,7 +1221,7 @@ bool MBTCPMaster::activateObject()
void MBTCPMaster::sigterm( int signo ) void MBTCPMaster::sigterm( int signo )
{ {
cerr << myname << ": ********* SIGTERM(" << signo <<") ********" << endl; cerr << myname << ": ********* SIGTERM(" << signo <<") ********" << endl;
activated = false; setProcActive(false);
/*! \todo Доделать выставление безопасного состояния на выходы. /*! \todo Доделать выставление безопасного состояния на выходы.
И нужно ли это. Ведь может не хватить времени на "обмен" И нужно ли это. Ведь может не хватить времени на "обмен"
...@@ -1182,7 +1265,7 @@ void MBTCPMaster::readConfiguration() ...@@ -1182,7 +1265,7 @@ void MBTCPMaster::readConfiguration()
UniXML_iterator it(root); UniXML_iterator it(root);
if( !it.goChildren() ) if( !it.goChildren() )
{ {
std::cerr << myname << "(readConfiguration): раздел <sensors> не содержит секций ?!!\n"; dlog[Debug::CRIT] << myname << "(readConfiguration): раздел <sensors> не содержит секций ?!!\n";
return; return;
} }
...@@ -1736,16 +1819,29 @@ void MBTCPMaster::initIterators() ...@@ -1736,16 +1819,29 @@ void MBTCPMaster::initIterators()
void MBTCPMaster::help_print( int argc, const char* const* argv ) void MBTCPMaster::help_print( int argc, const char* const* argv )
{ {
cout << "Default: prefix='mbtcp'" << endl; cout << "Default: prefix='mbtcp'" << endl;
cout << "--prefix-name name - ObjectId (имя) процесса. По умолчанию: MBTCPMaster1" << endl;
cout << "--prefix-confnode name - Настроечная секция в конф. файле <name>. " << endl;
cout << "--prefix-polltime msec - Пауза между опросаом карт. По умолчанию 200 мсек." << endl; cout << "--prefix-polltime msec - Пауза между опросаом карт. По умолчанию 200 мсек." << endl;
cout << "--prefix-heartbeat-id - Данный процесс связан с указанным аналоговым heartbeat-дачиком." << endl; cout << "--prefix-heartbeat-id name - Данный процесс связан с указанным аналоговым heartbeat-дачиком." << endl;
cout << "--prefix-heartbeat-max - Максимальное значение heartbeat-счётчика для данного процесса. По умолчанию 10." << endl; cout << "--prefix-heartbeat-max val - Максимальное значение heartbeat-счётчика для данного процесса. По умолчанию 10." << endl;
cout << "--prefix-ready-timeout - Время ожидания готовности SM к работе, мсек. (-1 - ждать 'вечно')" << endl; cout << "--prefix-ready-timeout msec - Время ожидания готовности SM к работе, мсек. (-1 - ждать 'вечно')" << endl;
cout << "--prefix-force - Сохранять значения в SM, независимо от, того менялось ли значение" << endl; cout << "--prefix-force 0,1 - Сохранять значения в SM, независимо от, того менялось ли значение" << endl;
cout << "--prefix-initPause - Задержка перед инициализацией (время на активизация процесса)" << endl; cout << "--prefix-force-out 0,1 - Считывать значения 'выходов' кажый раз SM (а не по изменению)" << endl;
cout << "--prefix-initPause msec - Задержка перед инициализацией (время на активизация процесса)" << endl;
cout << "--prefix-no-query-optimization 0,1 - Не оптимизировать запросы (не объединять соседние регистры в один запрос)" << endl;
cout << "--prefix-reg-from-id 0,1 - Использовать в качестве регистра sensor ID" << endl;
cout << "--prefix-filter-field name - Считывать список опрашиваемых датчиков, только у которых есть поле field" << endl;
cout << "--prefix-filter-value val - Считывать список опрашиваемых датчиков, только у которых field=value" << endl;
cout << "--prefix-statistic-sec sec - Выводить статистику опроса каждые sec секунд" << endl;
// ---------- init MBTCP ----------
// cout << "--prefix-sm-ready-timeout - время на ожидание старта SM" << endl; // cout << "--prefix-sm-ready-timeout - время на ожидание старта SM" << endl;
cout << " Настройки протокола TCP: " << endl; cout << " Настройки протокола TCP: " << endl;
cout << "--prefix-recv-timeout - Таймаут на ожидание ответа." << endl; cout << "--prefix-gateway hostname,IP - IP опрашиваемого узла" << endl;
cout << "--prefix-persistent-connection - Не закрывать соединение на каждом цикле опроса" << endl; cout << "--prefix-gateway-port num - port на опрашиваемом узле" << endl;
cout << "--prefix-recv-timeout msec - Таймаут на приём одного сообщения." << endl;
cout << "--prefix-timeout msec - Таймаут для определения отсутсвия соединения.в" << endl;
cout << "--prefix-persistent-connection 0,1 - Не закрывать соединение на каждом цикле опроса" << endl;
} }
// ----------------------------------------------------------------------------- // -----------------------------------------------------------------------------
MBTCPMaster* MBTCPMaster::init_mbmaster( int argc, const char* const* argv, UniSetTypes::ObjectId icID, SharedMemory* ic, MBTCPMaster* MBTCPMaster::init_mbmaster( int argc, const char* const* argv, UniSetTypes::ObjectId icID, SharedMemory* ic,
...@@ -1858,12 +1954,12 @@ bool MBTCPMaster::initDeviceInfo( RTUDeviceMap& m, ModbusRTU::ModbusAddr a, UniX ...@@ -1858,12 +1954,12 @@ bool MBTCPMaster::initDeviceInfo( RTUDeviceMap& m, ModbusRTU::ModbusAddr a, UniX
} }
dlog[Debug::INFO] << myname << "(initDeviceInfo): add addr=" << ModbusRTU::addr2str(a) << endl; dlog[Debug::INFO] << myname << "(initDeviceInfo): add addr=" << ModbusRTU::addr2str(a) << endl;
int tout = it.getPIntProp("timeout", UniSetTimer::WaitUpTime); int tout = it.getPIntProp("timeout",5000);
d->second->resp_ptTimeout.setTiming(tout); d->second->resp_ptTimeout.setTiming(tout);
d->second->resp_invert = it.getIntProp("invert"); d->second->resp_invert = it.getIntProp("invert");
// d->second->no_clean_input = it.getIntProp("no_clean_input"); // d->second->no_clean_input = it.getIntProp("no_clean_input");
// dlog[Debug::INFO] << myname << "(initDeviceInfo): add " << d->second << endl; dlog[Debug::INFO] << myname << "(initDeviceInfo): add " << d->second << endl;
return true; return true;
} }
......
...@@ -14,6 +14,7 @@ ...@@ -14,6 +14,7 @@
#include "Calibration.h" #include "Calibration.h"
#include "SMInterface.h" #include "SMInterface.h"
#include "SharedMemory.h" #include "SharedMemory.h"
#include "ThreadCreator.h"
#include "IOBase.h" #include "IOBase.h"
#include "VTypes.h" #include "VTypes.h"
#include "MTR.h" #include "MTR.h"
...@@ -68,10 +69,11 @@ ...@@ -68,10 +69,11 @@
Порт задаётся в конфигурационном файле параметром \b gateway_port или Порт задаётся в конфигурационном файле параметром \b gateway_port или
параметром командной строки \b --xxx-gateway-port. По умолчанию используется порт \b 502. параметром командной строки \b --xxx-gateway-port. По умолчанию используется порт \b 502.
\b --xxx-recv-timeout или \b recv_timeout msec - таймаут на приём сообщений. По умолчанию 2000 мсек. \b --xxx-recv-timeout или \b recv_timeout msec - таймаут на приём одного сообщения. По умолчанию 100 мсек.
\b --xxx-all-timeout или \b all_timeout msec - таймаут на определение отсутсвия связи \b --xxx-timeout или \b timeout msec - таймаут на определение отсутсвия связи
(после этого идёт попытка реинициализировать соединение) (после этого идёт попытка реинициализировать соединение)
По умолчанию 5000 мсек.
\b --xxx-no-query-optimization или \b no_query_optimization - [1|0] отключить оптимизацию запросов \b --xxx-no-query-optimization или \b no_query_optimization - [1|0] отключить оптимизацию запросов
...@@ -158,14 +160,18 @@ ...@@ -158,14 +160,18 @@
При этом будет записывыться значение "default". При этом будет записывыться значение "default".
\warning Регистр должен быть уникальным. И может повторятся только если указан параметр \a nbit или \a nbyte. \warning Регистр должен быть уникальным. И может повторятся только если указан параметр \a nbit или \a nbyte.
*/ */
// ----------------------------------------------------------------------------- // -----------------------------------------------------------------------------
/*! /*!
Реализация Modbus TCP Master для обмена с многими ModbusRTU устройствами \par Реализация Modbus TCP Master для обмена с многими ModbusRTU устройствами
через один modbus tcp шлюз. через один modbus tcp шлюз.
\par Чтобы не зависеть от таймаутов TCP соединений, которые могут неопределённо зависать
на создании соединения с недоступным хостом. Обмен вынесен в отдельный поток.
При этом в этом же потоке обновляются данные в SM. В свою очередь информация о датчиках
связи обновляется в основном потоке (чтобы не зависеть от TCP).
*/ */
class MBTCPMaster: class MBTCPMaster:
public UniSetObject_LT public UniSetObject_LT
...@@ -353,6 +359,7 @@ class MBTCPMaster: ...@@ -353,6 +359,7 @@ class MBTCPMaster:
SMInterface* shm; SMInterface* shm;
void step(); void step();
void poll_thread();
void poll(); void poll();
bool pollRTU( RTUDevice* dev, RegMap::iterator& it ); bool pollRTU( RTUDevice* dev, RegMap::iterator& it );
...@@ -360,6 +367,7 @@ class MBTCPMaster: ...@@ -360,6 +367,7 @@ class MBTCPMaster:
void updateRTU(RegMap::iterator& it); void updateRTU(RegMap::iterator& it);
void updateMTR(RegMap::iterator& it); void updateMTR(RegMap::iterator& it);
void updateRSProperty( RSProperty* p, bool write_only=false ); void updateRSProperty( RSProperty* p, bool write_only=false );
void updateRespondSensors();
virtual void processingMessage( UniSetTypes::VoidMessage *msg ); virtual void processingMessage( UniSetTypes::VoidMessage *msg );
void sysCommand( UniSetTypes::SystemMessage *msg ); void sysCommand( UniSetTypes::SystemMessage *msg );
...@@ -398,6 +406,8 @@ class MBTCPMaster: ...@@ -398,6 +406,8 @@ class MBTCPMaster:
void readConfiguration(); void readConfiguration();
bool check_item( UniXML_iterator& it ); bool check_item( UniXML_iterator& it );
bool checkProcActive();
void setProcActive( bool st );
private: private:
MBTCPMaster(); MBTCPMaster();
...@@ -415,17 +425,13 @@ class MBTCPMaster: ...@@ -415,17 +425,13 @@ class MBTCPMaster:
IOController::AIOStateList::iterator aitHeartBeat; IOController::AIOStateList::iterator aitHeartBeat;
UniSetTypes::ObjectId test_id; UniSetTypes::ObjectId test_id;
UniSetTypes::uniset_mutex pollMutex; UniSetTypes::uniset_mutex actMutex;
bool activated; bool activated;
int activateTimeout; int activateTimeout;
bool noQueryOptimization; bool noQueryOptimization;
bool force_disconnect; bool force_disconnect;
bool allNotRespond;
Trigger trAllNotRespond;
PassiveTimer ptAllNotRespond;
std::string prefix; std::string prefix;
bool no_extimer; bool no_extimer;
...@@ -433,6 +439,16 @@ class MBTCPMaster: ...@@ -433,6 +439,16 @@ class MBTCPMaster:
timeout_t stat_time; /*!< время сбора статистики обмена */ timeout_t stat_time; /*!< время сбора статистики обмена */
int poll_count; int poll_count;
PassiveTimer ptStatistic; /*!< таймер для сбора статистики обмена */ PassiveTimer ptStatistic; /*!< таймер для сбора статистики обмена */
// т.к. TCP может "зависнуть" на подключении к недоступному узлу
// делаем опрос в отдельном потоке
ThreadCreator<MBTCPMaster>* pollThread; /*!< поток опроса */
bool pollActivated;
UniSetTypes::uniset_mutex pollMutex;
// определение timeout для соединения
PassiveTimer ptTimeout;
UniSetTypes::uniset_mutex tcpMutex;
}; };
// ----------------------------------------------------------------------------- // -----------------------------------------------------------------------------
#endif // _MBTCPMaster_H_ #endif // _MBTCPMaster_H_
......
/* $Id: SMInterface.h,v 1.1 2008/12/14 21:57:50 vpashka Exp $ */
//--------------------------------------------------------------------------
#ifndef SMInterface_H_
#define SMInterface_H_
//--------------------------------------------------------------------------
#include <string>
#include "UniSetTypes.h"
#include "IONotifyController.h"
#include "UniversalInterface.h"
class SMInterface
{
public:
SMInterface( UniSetTypes::ObjectId _shmID, UniversalInterface* ui,
UniSetTypes::ObjectId myid, IONotifyController* ic=0 );
~SMInterface();
void setState ( UniSetTypes::ObjectId, bool state );
void setValue ( UniSetTypes::ObjectId, long value );
bool saveState ( IOController_i::SensorInfo& si, bool state, UniversalIO::IOTypes type, UniSetTypes::ObjectId supplier );
bool saveValue ( IOController_i::SensorInfo& si, long value, UniversalIO::IOTypes type, UniSetTypes::ObjectId supplier );
bool saveLocalState ( UniSetTypes::ObjectId id, bool state, UniversalIO::IOTypes type=UniversalIO::DigitalInput );
bool saveLocalValue ( UniSetTypes::ObjectId id, long value, UniversalIO::IOTypes type=UniversalIO::AnalogInput );
void setUndefinedState( IOController_i::SensorInfo& si, bool undefined, UniSetTypes::ObjectId supplier );
long getValue ( UniSetTypes::ObjectId id );
bool getState ( UniSetTypes::ObjectId id );
void askSensor( UniSetTypes::ObjectId id, UniversalIO::UIOCommand cmd,
UniSetTypes::ObjectId backid = UniSetTypes::DefaultObjectId );
inline bool alarm( UniSetTypes::AlarmMessage& msg, UniSetTypes::ObjectId messenger )
{
return ui->alarm(msg,messenger);
}
IOController_i::DSensorInfoSeq* getDigitalSensorsMap();
IOController_i::ASensorInfoSeq* getAnalogSensorsMap();
IONotifyController_i::ThresholdsListSeq* getThresholdsList();
void localSaveValue( IOController::AIOStateList::iterator& it,
UniSetTypes::ObjectId sid,
CORBA::Long newvalue, UniSetTypes::ObjectId sup_id );
void localSaveState( IOController::DIOStateList::iterator& it,
UniSetTypes::ObjectId sid,
CORBA::Boolean newstate, UniSetTypes::ObjectId sup_id );
void localSetState( IOController::DIOStateList::iterator& it,
UniSetTypes::ObjectId sid,
CORBA::Boolean newstate, UniSetTypes::ObjectId sup_id );
void localSetValue( IOController::AIOStateList::iterator& it,
UniSetTypes::ObjectId sid,
CORBA::Long value, UniSetTypes::ObjectId sup_id );
bool localGetState( IOController::DIOStateList::iterator& it,
UniSetTypes::ObjectId sid );
long localGetValue( IOController::AIOStateList::iterator& it,
UniSetTypes::ObjectId sid );
/*! функция выставления признака неопределённого состояния для аналоговых датчиков
// для дискретных датчиков необходимости для подобной функции нет.
// см. логику выставления в функции localSaveState
*/
void localSetUndefinedState( IOController::AIOStateList::iterator& it,
bool undefined, UniSetTypes::ObjectId sid );
// специальные функции
IOController::DIOStateList::iterator dioEnd();
IOController::AIOStateList::iterator aioEnd();
void initAIterator( IOController::AIOStateList::iterator& it );
void initDIterator( IOController::DIOStateList::iterator& it );
bool exist();
bool waitSMready( int msec, int pause=5000 );
bool waitSMworking( UniSetTypes::ObjectId, int msec, int pause=3000 );
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;
UniversalInterface* ui;
CORBA::Object_var oref;
UniSetTypes::ObjectId shmID;
UniSetTypes::ObjectId myid;
};
//--------------------------------------------------------------------------
#endif
...@@ -10,6 +10,7 @@ using namespace UniSetTypes; ...@@ -10,6 +10,7 @@ using namespace UniSetTypes;
#define BEG_FUNC(name) \ #define BEG_FUNC(name) \
try \ try \
{ \ { \
uniset_mutex_lock l(shmMutex,500); \
IONotifyController_i_var shm;\ IONotifyController_i_var shm;\
for( unsigned int i=0; i<conf->getRepeatCount(); i++)\ for( unsigned int i=0; i<conf->getRepeatCount(); i++)\
{\ {\
...@@ -32,6 +33,7 @@ using namespace UniSetTypes; ...@@ -32,6 +33,7 @@ using namespace UniSetTypes;
#define BEG_FUNC1(name) \ #define BEG_FUNC1(name) \
try \ try \
{ \ { \
uniset_mutex_lock l(shmMutex,500); \
if( true ) \ if( true ) \
{ \ { \
try \ try \
......
...@@ -45,6 +45,8 @@ mbErrCode ModbusTCPMaster::sendData( unsigned char* buf, int len ) ...@@ -45,6 +45,8 @@ mbErrCode ModbusTCPMaster::sendData( unsigned char* buf, int len )
mbErrCode ModbusTCPMaster::query( ModbusAddr addr, ModbusMessage& msg, mbErrCode ModbusTCPMaster::query( ModbusAddr addr, ModbusMessage& msg,
ModbusMessage& reply, timeout_t timeout ) ModbusMessage& reply, timeout_t timeout )
{ {
try
{
if( iaddr.empty() ) if( iaddr.empty() )
{ {
dlog[Debug::WARN] << "(query): unknown ip address for server..." << endl; dlog[Debug::WARN] << "(query): unknown ip address for server..." << endl;
...@@ -69,10 +71,10 @@ mbErrCode ModbusTCPMaster::query( ModbusAddr addr, ModbusMessage& msg, ...@@ -69,10 +71,10 @@ mbErrCode ModbusTCPMaster::query( ModbusAddr addr, ModbusMessage& msg,
tcp->setTimeout(timeout); tcp->setTimeout(timeout);
ost::Thread::setException(ost::Thread::throwException); // ost::Thread::setException(ost::Thread::throwException);
// ost::tpport_t port;
// cerr << "****** peer: " << tcp->getPeer(&port) << " err: " << tcp->getErrorNumber() << endl;
try
{
if( nTransaction >= numeric_limits<ModbusRTU::ModbusData>::max() ) if( nTransaction >= numeric_limits<ModbusRTU::ModbusData>::max() )
nTransaction = 0; nTransaction = 0;
...@@ -136,6 +138,7 @@ mbErrCode ModbusTCPMaster::query( ModbusAddr addr, ModbusMessage& msg, ...@@ -136,6 +138,7 @@ mbErrCode ModbusTCPMaster::query( ModbusAddr addr, ModbusMessage& msg,
while( !qrecv.empty() ) while( !qrecv.empty() )
qrecv.pop(); qrecv.pop();
tcp->sync();
if( tcp->isPending(ost::Socket::pendingInput,timeout) ) if( tcp->isPending(ost::Socket::pendingInput,timeout) )
{ {
/* /*
...@@ -158,9 +161,14 @@ mbErrCode ModbusTCPMaster::query( ModbusAddr addr, ModbusMessage& msg, ...@@ -158,9 +161,14 @@ mbErrCode ModbusTCPMaster::query( ModbusAddr addr, ModbusMessage& msg,
if( ret < (int)sizeof(rmh) ) if( ret < (int)sizeof(rmh) )
{ {
ost::tpport_t port;
if( dlog.debugging(Debug::INFO) ) if( dlog.debugging(Debug::INFO) )
dlog[Debug::INFO] << "(ModbusTCPMaster::query): ret=" << (int)ret dlog[Debug::INFO] << "(ModbusTCPMaster::query): ret=" << (int)ret
<< " < rmh=" << (int)sizeof(rmh) << endl; << " < rmh=" << (int)sizeof(rmh)
<< " err: " << tcp->getErrorNumber()
<< " perr: " << tcp->getPeer(&port)
<< endl;
disconnect(); disconnect();
return erTimeOut; // return erHardwareError; return erTimeOut; // return erHardwareError;
} }
...@@ -211,17 +219,18 @@ mbErrCode ModbusTCPMaster::query( ModbusAddr addr, ModbusMessage& msg, ...@@ -211,17 +219,18 @@ mbErrCode ModbusTCPMaster::query( ModbusAddr addr, ModbusMessage& msg,
{ {
dlog[Debug::WARN] << "(query): " << ex << endl; dlog[Debug::WARN] << "(query): " << ex << endl;
} }
catch(SystemError& err) catch( SystemError& err )
{ {
dlog[Debug::WARN] << "(query): " << err << endl; dlog[Debug::WARN] << "(query): " << err << endl;
} }
catch(Exception& ex) catch( Exception& ex )
{ {
dlog[Debug::WARN] << "(query): " << ex << endl; dlog[Debug::WARN] << "(query): " << ex << endl;
} }
catch( ost::SockException& e ) catch( ost::SockException& e )
{ {
dlog[Debug::WARN] << e.getString() << ": " << e.getSystemErrorString() << endl; dlog[Debug::WARN] << "(query): tcp error: " << e.getString() << endl;
return erTimeOut;
} }
catch(...) catch(...)
{ {
...@@ -252,19 +261,42 @@ void ModbusTCPMaster::reconnect() ...@@ -252,19 +261,42 @@ void ModbusTCPMaster::reconnect()
// cerr << "tcp diconnect..." << endl; // cerr << "tcp diconnect..." << endl;
tcp->disconnect(); tcp->disconnect();
delete tcp; delete tcp;
tcp = 0;
} }
ost::Thread::setException(ost::Thread::throwException); ost::Thread::setException(ost::Thread::throwException);
// cerr << "create new tcp..." << endl; try
tcp = new ost::TCPStream(iaddr.c_str()); {
tcp->setTimeout(500); // TCPStream (const char *name, Family family=IPV4, unsigned mss=536, bool throwflag=false, timeout_t timer=0)
tcp = new ost::TCPStream(iaddr.c_str(),ost::Socket::IPV4,536,true,500);
tcp->setTimeout(replyTimeOut_ms);
}
catch(ost::Socket *socket)
{
ost::tpport_t port;
int err = socket->getErrorNumber();
ost::InetAddress saddr = (ost::InetAddress)socket->getPeer(&port);
dlog[Debug::CRIT] << "tcp error " << saddr.getHostname() << ":" << port << " = " << err << endl;
}
catch( ost::SockException& e)
{
dlog[Debug::CRIT] << "tcp error: " << e.getString() << endl;
}
catch(...)
{
dlog[Debug::CRIT] << "create TCPStream[" << iaddr << "] error..." << endl;
}
} }
// ------------------------------------------------------------------------- // -------------------------------------------------------------------------
void ModbusTCPMaster::connect( ost::InetAddress addr, int port ) void ModbusTCPMaster::connect( ost::InetAddress addr, int port )
{ {
if( tcp ) if( tcp )
{
disconnect(); disconnect();
delete tcp;
tcp = 0;
}
// if( !tcp ) // if( !tcp )
// { // {
...@@ -275,9 +307,26 @@ void ModbusTCPMaster::connect( ost::InetAddress addr, int port ) ...@@ -275,9 +307,26 @@ void ModbusTCPMaster::connect( ost::InetAddress addr, int port )
dlog[Debug::INFO] << "(ModbusTCPMaster): connect to " << s.str() << endl; dlog[Debug::INFO] << "(ModbusTCPMaster): connect to " << s.str() << endl;
iaddr = s.str(); iaddr = s.str();
try
{
tcp = new ost::TCPStream(iaddr.c_str()); tcp = new ost::TCPStream(iaddr.c_str());
tcp->setTimeout(500); tcp->setTimeout(replyTimeOut_ms);
}
catch(ost::Socket *socket)
{
ost::tpport_t port;
int err = socket->getErrorNumber();
ost::InetAddress saddr = (ost::InetAddress)socket->getPeer(&port);
dlog[Debug::CRIT] << ": tcp error " << saddr.getHostname() << ":" << port << " = " << err << endl;
}
catch( ost::SockException& e)
{
dlog[Debug::CRIT] << "tcp error: " << e.getString() << endl;
}
catch(...)
{
dlog[Debug::CRIT] << "create TCPStream[" << iaddr << "] error..." << endl;
}
// } // }
} }
// ------------------------------------------------------------------------- // -------------------------------------------------------------------------
......
...@@ -10,7 +10,7 @@ PassiveTimer pt(1000); ...@@ -10,7 +10,7 @@ PassiveTimer pt(1000);
int main() int main()
{ {
PassiveTimer pt1(5000); PassiveTimer pt1(5000);
cout << " pt1.getInterval()=" << pt1.getInterval() << endl; cout << " pt1.getInterval()=" << pt1.getInterval() << " TEST: " << ((pt1.getInterval()==5000) ? "OK" : "FAILED") << endl;
PassiveTimer pt2; PassiveTimer pt2;
cout << " pt2.getInterval()=" << pt2.getInterval() << endl; cout << " pt2.getInterval()=" << pt2.getInterval() << endl;
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment