Commit a8c6f301 authored by Pavel Vainerman's avatar Pavel Vainerman Committed by Pavel Vainerman

[uwebsocket]: refactoring: rename parameters, update help, added runlock

parent 1ce93d28
...@@ -49,6 +49,7 @@ ...@@ -49,6 +49,7 @@
<BackendOpenTSDB name="BackendOpenTSDB" host="localhost" filter_field="tsdb" filter_value="1" tags="host=localhost uniset=1" prefix="uniset"/> <BackendOpenTSDB name="BackendOpenTSDB" host="localhost" filter_field="tsdb" filter_value="1" tags="host=localhost uniset=1" prefix="uniset"/>
<UWebSocketGate name="UWebSocketGate1"/> <UWebSocketGate name="UWebSocketGate1"/>
<UWebSocketGate name="UWebSocketGate2" port="8082"/>
<settings> <settings>
...@@ -5523,6 +5524,7 @@ ...@@ -5523,6 +5524,7 @@
<item id="6089" name="MQTTPublisher1"/> <item id="6089" name="MQTTPublisher1"/>
<item id="6090" name="BackendOpenTSDB"/> <item id="6090" name="BackendOpenTSDB"/>
<item id="6091" name="UWebSocketGate1"/> <item id="6091" name="UWebSocketGate1"/>
<item id="6092" name="UWebSocketGate2"/>
<item id="6101" name="MBTCP1"/> <item id="6101" name="MBTCP1"/>
<item id="6102" name="MBTCP2"/> <item id="6102" name="MBTCP2"/>
<item id="6103" name="MBTCP3"/> <item id="6103" name="MBTCP3"/>
......
...@@ -25,6 +25,7 @@ ...@@ -25,6 +25,7 @@
#include "unisetstd.h" #include "unisetstd.h"
#include <Poco/Net/NetException.h> #include <Poco/Net/NetException.h>
#include <Poco/Net/WebSocket.h> #include <Poco/Net/WebSocket.h>
#include <Poco/Net/ServerSocket.h>
#include "ujson.h" #include "ujson.h"
#include "UWebSocketGate.h" #include "UWebSocketGate.h"
#include "Configuration.h" #include "Configuration.h"
...@@ -35,6 +36,7 @@ ...@@ -35,6 +36,7 @@
#include "ORepHelpers.h" #include "ORepHelpers.h"
#include "UWebSocketGateSugar.h" #include "UWebSocketGateSugar.h"
#include "SMonitor.h" #include "SMonitor.h"
#include "UTCPSocket.h"
// -------------------------------------------------------------------------- // --------------------------------------------------------------------------
using namespace uniset; using namespace uniset;
using namespace std; using namespace std;
...@@ -58,6 +60,9 @@ UWebSocketGate::UWebSocketGate( uniset::ObjectId id ...@@ -58,6 +60,9 @@ UWebSocketGate::UWebSocketGate( uniset::ObjectId id
conf->initLogStream(mylog, s.str()); conf->initLogStream(mylog, s.str());
} }
if( mylog->verbose() == 0 )
mylog->verbose(1);
UniXML::iterator it(cnode); UniXML::iterator it(cnode);
int maxCacheSize = conf->getArgPInt("--" + prefix + "max-ui-cache-size", it.getProp("msgUIChacheSize"), 5000); int maxCacheSize = conf->getArgPInt("--" + prefix + "max-ui-cache-size", it.getProp("msgUIChacheSize"), 5000);
...@@ -65,7 +70,7 @@ UWebSocketGate::UWebSocketGate( uniset::ObjectId id ...@@ -65,7 +70,7 @@ UWebSocketGate::UWebSocketGate( uniset::ObjectId id
shm = make_shared<SMInterface>(shmID, ui, getId(), ic); shm = make_shared<SMInterface>(shmID, ui, getId(), ic);
maxwsocks = conf->getArgPInt("--" + prefix + "ws-max", it.getProp("wsMax"), maxwsocks); maxwsocks = conf->getArgPInt("--" + prefix + "max-conn", it.getProp("wsMaxConnection"), maxwsocks);
wscmd = make_shared<ev::async>(); wscmd = make_shared<ev::async>();
wsactivate.set<UWebSocketGate, &UWebSocketGate::onActivate>(this); wsactivate.set<UWebSocketGate, &UWebSocketGate::onActivate>(this);
...@@ -92,21 +97,62 @@ UWebSocketGate::UWebSocketGate( uniset::ObjectId id ...@@ -92,21 +97,62 @@ UWebSocketGate::UWebSocketGate( uniset::ObjectId id
wsMaxSend = conf->getArgPInt("--" + prefix + "max-send", it.getProp("wsMaxSend"), wsMaxSend); wsMaxSend = conf->getArgPInt("--" + prefix + "max-send", it.getProp("wsMaxSend"), wsMaxSend);
wsMaxCmd = conf->getArgPInt("--" + prefix + "max-cmd", it.getProp("wsMaxCmd"), wsMaxCmd); wsMaxCmd = conf->getArgPInt("--" + prefix + "max-cmd", it.getProp("wsMaxCmd"), wsMaxCmd);
myinfoV(1) << myname << "maxSend=" << wsMaxSend << " maxCmd=" << wsMaxCmd << endl; myinfoV(1) << myname << "(init): maxSend=" << wsMaxSend
<< " maxCmd=" << wsMaxCmd
<< " maxConnections=" << maxwsocks
<< endl;
httpHost = conf->getArgParam("--" + prefix + "httpserver-host", "localhost"); httpHost = conf->getArgParam("--" + prefix + "host", it.getProp2("host", "localhost"));
httpPort = conf->getArgPInt("--" + prefix + "httpserver-port", 8081); httpPort = conf->getArgPInt("--" + prefix + "port", it.getPIntProp("port", 8081));
httpCORS_allow = conf->getArgParam("--" + prefix + "httpserver-cors-allow", "*"); httpCORS_allow = conf->getArgParam("--" + prefix + "cors-allow", "*");
myinfoV(1) << myname << "(init): http server parameters " << httpHost << ":" << httpPort << endl; myinfoV(1) << myname << "(init): http server parameters " << httpHost << ":" << httpPort << endl;
Poco::Net::SocketAddress sa(httpHost, httpPort); Poco::Net::SocketAddress sa(httpHost, httpPort);
ostringstream lockfile;
lockfile << conf->getLockDir() << "ws" << id;
myinfoV(1) << myname << "(init): lockfile " << lockfile.str() << endl;
runlock = unisetstd::make_unique<uniset::RunLock>(lockfile.str());
if( runlock->isLocked() )
{
ostringstream err;
err << myname << "(init): lock failed! UWebSocketGate id=" << id
<< " Already running!" << endl
<< " ..or delete the lockfile " << lockfile.str() << " to run..";
throw uniset::SystemError(err.str());
}
if( !runlock->lock() )
{
ostringstream err;
err << myname << "(init): failed create lockfile " << lockfile.str();
throw uniset::SystemError(err.str());
}
// check create socket
try
{
myinfoV(4) << myname << "(init): check socket create " << httpHost << ":" << httpPort << endl;
Poco::Net::ServerSocket ss;
ss.bind(sa);
ss.close();
}
catch( std::exception& ex )
{
ostringstream err;
err << myname << "(init): create socket " << httpHost << ":" << httpPort << " error or already use..";
throw uniset::SystemError(err.str());
}
try try
{ {
Poco::Net::HTTPServerParams* httpParams = new Poco::Net::HTTPServerParams; Poco::Net::HTTPServerParams* httpParams = new Poco::Net::HTTPServerParams;
int maxQ = conf->getArgPInt("--" + prefix + "httpserver-max-queued", it.getProp("httpMaxQueued"), 100); int maxQ = conf->getArgPInt("--" + prefix + "max-queued", it.getProp("maxQueued"), 100);
int maxT = conf->getArgPInt("--" + prefix + "httpserver-max-threads", it.getProp("httpMaxThreads"), 3); int maxT = conf->getArgPInt("--" + prefix + "max-threads", it.getProp("maxThreads"), 3);
httpParams->setMaxQueued(maxQ); httpParams->setMaxQueued(maxQ);
httpParams->setMaxThreads(maxT); httpParams->setMaxThreads(maxT);
...@@ -129,6 +175,9 @@ UWebSocketGate::~UWebSocketGate() ...@@ -129,6 +175,9 @@ UWebSocketGate::~UWebSocketGate()
if( httpserv ) if( httpserv )
httpserv->stop(); httpserv->stop();
if( runlock )
runlock->unlock();
} }
//-------------------------------------------------------------------------------------------- //--------------------------------------------------------------------------------------------
void UWebSocketGate::onTerminate( ev::sig& evsig, int revents ) void UWebSocketGate::onTerminate( ev::sig& evsig, int revents )
...@@ -282,23 +331,29 @@ void UWebSocketGate::help_print() ...@@ -282,23 +331,29 @@ void UWebSocketGate::help_print()
{ {
cout << "Default: prefix='ws'" << endl; cout << "Default: prefix='ws'" << endl;
cout << "--prefix-name name - Имя. Для поиска настроечной секции в configure.xml" << endl; cout << "--prefix-name name - Имя. Для поиска настроечной секции в configure.xml" << endl;
cout << "--uniset-object-size-message-queue num - Размер uniset-очереди сообщений" << endl; cout << "--uniset-object-size-message-queue num - Размер uniset-очереди сообщений. По умолчанию: 10000" << endl;
cout << "--prefix-msg-check-time msec - Период опроса uniset-очереди сообщений, для обработки новых сообщений. По умолчанию: 10 мсек" << endl; cout << "--prefix-msg-check-time msec - Период опроса uniset-очереди сообщений, для обработки новых сообщений. По умолчанию: 10 мсек" << endl;
cout << "--prefix-max-messages-processing num - Количество uniset-сообщений обрабатывамых за один раз. По умолчанию 50. По умолчанию: 100" << endl; cout << "--prefix-max-messages-processing num - Количество uniset-сообщений обрабатывамых за один раз. По умолчанию: 100" << endl;
cout << "websockets: " << endl; cout << "websockets: " << endl;
cout << "--prefix-max num - Максимальное количество websocket-ов" << endl; cout << "--prefix-max-conn num - Максимальное количество одновременных подключений (клиентов). По усмолчанию: 50" << endl;
cout << "--prefix-heartbeat-time msec - Период сердцебиения в соединении. По умолчанию: 3000 мсек" << endl; cout << "--prefix-heartbeat-time msec - Период сердцебиения в соединении. По умолчанию: 3000 мсек" << endl;
cout << "--prefix-send-time msec - Период посылки сообщений. По умолчанию: 500 мсек" << endl; cout << "--prefix-send-time msec - Период посылки сообщений. По умолчанию: 500 мсек" << endl;
cout << "--prefix-max-send num - Максимальное число сообщений посылаемых за один раз. По умолчанию: 5000" << endl; cout << "--prefix-max-send num - Максимальное число сообщений посылаемых за один раз. По умолчанию: 5000" << endl;
cout << "--prefix-max-cmd num - Максимальное число команд обрабатываемых за один раз. По умолчанию: 200" << endl; cout << "--prefix-max-cmd num - Максимальное число команд обрабатываемых за один раз. По умолчанию: 200" << endl;
cout << "http: " << endl; cout << "http: " << endl;
cout << "--prefix-httpserver-host ip - IP на котором слушает http сервер. По умолчанию: localhost" << endl; cout << "--prefix-host ip - IP на котором слушает http сервер. По умолчанию: localhost" << endl;
cout << "--prefix-httpserver-port num - Порт на котором принимать запросы. По умолчанию: 8080" << endl; cout << "--prefix-port num - Порт на котором принимать запросы. По умолчанию: 8081" << endl;
cout << "--prefix-httpserver-max-queued num - Размер очереди запросов к http серверу. По умолчанию: 100" << endl; cout << "--prefix-max-queued num - Размер очереди запросов к http серверу. По умолчанию: 100" << endl;
cout << "--prefix-httpserver-max-threads num - Разрешённое количество потоков для http-сервера. По умолчанию: 3" << endl; cout << "--prefix-max-threads num - Разрешённое количество потоков для http-сервера. По умолчанию: 3" << endl;
cout << "--prefix-httpserver-cors-allow addr - (CORS): Access-Control-Allow-Origin. Default: *" << endl; cout << "--prefix-cors-allow addr - (CORS): Access-Control-Allow-Origin. Default: *" << endl;
cout << "logs: " << endl;
cout << "--prefix-log-add-levels [crit,warn,info..] - Уровень логов" << endl;
cout << "--prefix-log-verbosity N - Уровень подробностей [1...5]" << endl;
cout << " Пример параметров для запуска с подробными логами: " << endl;
cout << " --ws-log-add-levels any --ws-log-verbosity 5" << endl;
} }
// ----------------------------------------------------------------------------- // -----------------------------------------------------------------------------
void UWebSocketGate::run( bool async ) void UWebSocketGate::run( bool async )
......
...@@ -41,6 +41,7 @@ ...@@ -41,6 +41,7 @@
#include "UHttpRequestHandler.h" #include "UHttpRequestHandler.h"
#include "UHttpServer.h" #include "UHttpServer.h"
#include "UTCPCore.h" #include "UTCPCore.h"
#include "RunLock.h"
// ------------------------------------------------------------------------- // -------------------------------------------------------------------------
namespace uniset namespace uniset
{ {
...@@ -72,7 +73,7 @@ namespace uniset ...@@ -72,7 +73,7 @@ namespace uniset
<UWebSocketGate name="UWebSocketGate" .../> <UWebSocketGate name="UWebSocketGate" .../>
\endcode \endcode
Количество создаваемых websocket-ов можно ограничить при помощи параметра maxWebsockets (--prefix-ws-max). Количество создаваемых websocket-ов можно ограничить при помощи параметра maxWebsockets (--prefix-max-conn).
\section sec_UWebSocketGate_DETAIL UWebSocketGate: Технические детали \section sec_UWebSocketGate_DETAIL UWebSocketGate: Технические детали
Вся релизация построена на "однопоточном" eventloop. Если датчики долго не меняются, то периодически посылается "ping" сообщение. Вся релизация построена на "однопоточном" eventloop. Если датчики долго не меняются, то периодически посылается "ping" сообщение.
...@@ -257,6 +258,7 @@ namespace uniset ...@@ -257,6 +258,7 @@ namespace uniset
std::shared_ptr<DebugStream> mylog; std::shared_ptr<DebugStream> mylog;
std::shared_ptr<SMInterface> shm; std::shared_ptr<SMInterface> shm;
std::unique_ptr<uniset::RunLock> runlock;
#ifndef DISABLE_REST_API #ifndef DISABLE_REST_API
std::shared_ptr<Poco::Net::HTTPServer> httpserv; std::shared_ptr<Poco::Net::HTTPServer> httpserv;
......
/*
* Copyright (c) 2021 Pavel Vainerman.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Lesser General Public License as
* published by the Free Software Foundation, version 2.1.
*
* This program is distributed in the hope that it will be useful, but
* WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Lesser Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
// --------------------------------------------------------------------------
#ifndef RunLock_H_
#define RunLock_H_
// ---------------------------------------------------------------------------
#include <string>
// ---------------------------------------------------------------------------
namespace uniset {
// ---------------------------------------------------------------------------
/*!
* RunLock неявно использует pid процесса который вызывает lock();
*/
class RunLock
{
public:
RunLock( const std::string& lockfile );
~RunLock();
bool isLocked() const;
bool lock() const;
bool unlock() const;
bool isLockOwner() const;
protected:
const std::string lockfile;
};
// ----------------------------------------------------------------------------
} // end of namespace uniset
// ----------------------------------------------------------------------------
#endif
...@@ -6,7 +6,8 @@ noinst_LTLIBRARIES = libProcesses.la ...@@ -6,7 +6,8 @@ noinst_LTLIBRARIES = libProcesses.la
libProcesses_la_CXXFLAGS = -I$(top_builddir)/include $(SIGC_CFLAGS) $(EV_CFLAGS) libProcesses_la_CXXFLAGS = -I$(top_builddir)/include $(SIGC_CFLAGS) $(EV_CFLAGS)
libProcesses_la_LIBADD = $(SIGC_LIBS) $(EV_LIBS) libProcesses_la_LIBADD = $(SIGC_LIBS) $(EV_LIBS)
libProcesses_la_SOURCES = IOController_iSK.cc IOController.cc IONotifyController.cc \ libProcesses_la_SOURCES = IOController_iSK.cc IOController.cc IONotifyController.cc \
IOConfig_XML.cc EventLoopServer.cc CommonEventLoop.cc ProxyManager.cc PassiveObject.cc IOConfig_XML.cc EventLoopServer.cc CommonEventLoop.cc ProxyManager.cc PassiveObject.cc \
RunLock.cc
# NCRestorer.cc NCRestorer_XML.cc # NCRestorer.cc NCRestorer_XML.cc
......
/*
* Copyright (c) 2021 Pavel Vainerman.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Lesser General Public License as
* published by the Free Software Foundation, version 2.1.
*
* This program is distributed in the hope that it will be useful, but
* WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Lesser Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
// --------------------------------------------------------------------------
//#include <sstream>
#include <fstream>
#include <Poco/File.h>
#include <Poco/Process.h>
#include "RunLock.h"
#include "Exceptions.h"
#include "UniSetTypes.h"
// --------------------------------------------------------------------------
using namespace std;
using namespace uniset;
// --------------------------------------------------------------------------
RunLock::RunLock( const std::string& _lockfile ):
lockfile(_lockfile)
{
}
RunLock::~RunLock()
{
if( isLocked() )
{
try
{
unlock();
}
catch(...){}
}
}
// --------------------------------------------------------------------------
bool RunLock::isLockOwner() const
{
if( !uniset::file_exist(lockfile) )
return false;
ifstream pidfile(lockfile.c_str());
if( !pidfile )
throw ORepFailed("(RunLock): can't read lockfile " + lockfile);
int pid;
pidfile >> pid;
return pid == Poco::Process::id();
}
// --------------------------------------------------------------------------
bool RunLock::isLocked() const
{
if( !uniset::file_exist(lockfile) )
return false;
ifstream pidfile(lockfile.c_str());
if( !pidfile )
throw ORepFailed("(RunLock): can't read lockfile " + lockfile);
int pid;
pidfile >> pid;
return Poco::Process::isRunning(pid);
}
// --------------------------------------------------------------------------
bool RunLock::lock() const
{
if( isLocked() )
return true;
ofstream pidfile(lockfile.c_str(), ios::out | ios::trunc);
if( !pidfile )
return false;
// throw ORepFailed("(RunLock): can't create lockfile " + lockfile);
pidfile << Poco::Process::id() << endl;
pidfile.close();
return true;
}
// --------------------------------------------------------------------------
bool RunLock::unlock() const
{
if( !isLocked() )
return true;
if( !isLockOwner() )
return false;
Poco::File f(lockfile);
try
{
f.remove(false);
return true;
}
catch(...) {}
return false;
}
// --------------------------------------------------------------------------
...@@ -40,6 +40,7 @@ tests_with_conf_SOURCES = tests_with_conf.cc \ ...@@ -40,6 +40,7 @@ tests_with_conf_SOURCES = tests_with_conf.cc \
test_conftest.cc \ test_conftest.cc \
test_ui.cc \ test_ui.cc \
test_iorfile.cc \ test_iorfile.cc \
test_runlock.cc \
test_messagetype.cc \ test_messagetype.cc \
test_utypes.cc \ test_utypes.cc \
test_mqueue.cc \ test_mqueue.cc \
......
#include <catch.hpp>
#include "Configuration.h"
#include "UniSetTypes.h"
#include "RunLock.h"
using namespace std;
using namespace uniset;
TEST_CASE("RunLock", "[runlock][basic]" )
{
REQUIRE( uniset_conf() != nullptr );
const std::string lname = uniset_conf()->getLockDir() + "testLock.lock";
RunLock rlock(lname);
REQUIRE_FALSE( rlock.isLocked() );
REQUIRE( rlock.lock() );
REQUIRE( rlock.isLocked() );
REQUIRE( rlock.isLockOwner() );
// double check
REQUIRE( rlock.isLocked() );
REQUIRE( rlock.isLockOwner() );
REQUIRE( rlock.unlock() );
REQUIRE_FALSE( rlock.isLocked() );
REQUIRE_FALSE( rlock.isLockOwner() );
}
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