diff --git a/src/SVClient/README.md b/src/SVClient/README.md new file mode 100644 index 00000000..0e626f51 --- /dev/null +++ b/src/SVClient/README.md @@ -0,0 +1,103 @@ +# SVClient + +Клиентская библиотека отправки сигналов SVisual: накопление значений по циклам, формирование бинарных пакетов и отправка на сервер по TCP. + +Собирается как shared library (`libSVClient.so` / `SVClient.dll`), публичный API — `SVClient.h`. + +## Сборка + +```bash +cd src/build +cmake .. +cmake --build . --target SVClient +``` + +Зависимости: `SVBase`, `SVMisc` (`TCPClient`). + +## Lifecycle + +``` +Idle ──svConnect()──► SessionActive ──svDisconnect()──► Idle + │ + ├── sender thread running (always after first svConnect) + ├── _isConnect true/false (TCP state) + └── automatic reconnect in sender thread +``` + +| Метод | Поведение | +|-------|-----------| +| `svSetParam(cycleRecMs, packetSz)` | Только до первого `svAdd*Value` и до `svConnect`; иначе `false` | +| `svConnect(module, ip, port)` | Стартует sender thread один раз; возвращает результат **первого** TCP connect | +| `svConnect()` (повторный) | При активной сессии — `false`/`true` по текущему `_isConnect`, второй поток не создаётся | +| `svAdd*Value(name, …)` | Запись значения в текущий цикл; имя должно быть уникальным по типу | +| `svDisconnect()` | Останавливает поток, закрывает сокет; **обязателен** перед выгрузкой DLL / exit | + +`false` от `svConnect()` **не означает**, что сессия не запущена: sender thread уже работает, reconnect идёт в фоне. + +Порядок вызовов для типичного приложения: + +```cpp +SV::svSetParam(100, 10); // до connect и add +SV::svConnect("myModule", "127.0.0.1", 2144); // false ok — reconnect в фоне +while (running) { + SV::svAddIntValue("temp", value); + // ... +} +SV::svDisconnect(); // обязательно перед unload +``` + +Пример: [`../SVClientExample/src/main.cpp`](../SVClientExample/src/main.cpp). + +## Публичный API + +| Функция | Назначение | +|---------|------------| +| `svSetParam(cycleRecMs, packetSz)` | Период записи (мс) и размер пакета (число циклов) | +| `svConnect(moduleName, ipAddr, port)` | Подключение и старт sender thread | +| `svDisconnect()` | Остановка потока и отключение | +| `svAddBoolValue(name, value, onlyFront)` | Bool-сигнал; `onlyFront` — импульс только в активном цикле | +| `svAddIntValue(name, value)` | Int-сигнал | +| `svAddFloatValue(name, value)` | Float-сигнал | + +### Ограничения имён и параметров + +- `moduleName`, `name` — non-null, длина `< 24` (`SV_NAMESZ`), без подстрок `"=begin="` и `"=end="`. +- `ipAddr` — non-null; на Linux только dotted-decimal IP (`inet_aton` в `SVMisc`). +- `cycleRecMs >= 1`, `1 <= packetSz <= SV_PACKETSZ_MAX`. + +## Wire-format + +Пакет совместим с парсером [`SVServer`](../SVServer/src/buffer_data.cpp): + +``` +=begin= | int32 dataSz | module[24] | (name[24] | ValueType | Value[packetSz])... | =end= +``` + +Fixed-size поля (`module`, `name`) zero-padded до 24 байт. + +## Внутренняя структура + +``` +SVClient.h — публичный C API +src/client.cpp — svConnect/svDisconnect, addValue, prepareData, sendCycle +src/dllmain.cpp — Windows DLL entry (пустой; cleanup через svDisconnect) +``` + +### Синхронизация + +| Mutex / флаг | Область | +|--------------|---------| +| `_sessionMtx` | `svConnect` / `svDisconnect`, `_thr` | +| `_tcpMtx` | `TCPClient::connect` / `disconnect` / `sendData` | +| `_mtx` | `_values`, `_curCycle`, `cng` | +| `_isConnect` (`atomic_bool`) | Snapshot TCP-состояния; записи под `_tcpMtx`, чтения lock-free | + +Порядок захвата: `_sessionMtx` → `_mtx`. `_tcpMtx` и `_mtx` **не держать одновременно**. + +## Потребители + +| Модуль | Использование | +|--------|---------------| +| SVClientExample | Пример отправки синусоиды | +| SVMonitor | Клиентские модули через TCP | +| SVArduino | Порт клиента для Arduino | diff --git a/src/SVClient/SVClient.h b/src/SVClient/SVClient.h index 6a90a036..0d3bfc1c 100644 --- a/src/SVClient/SVClient.h +++ b/src/SVClient/SVClient.h @@ -43,26 +43,34 @@ namespace SV{ - // connect of server - // moduleName - module (leng max 24) - // ipAddrServ - ip + // Connect to server and start sender thread. + // moduleName - module name (non-null, length < 24, must not contain "=begin=" or "=end=") + // ipAddrServ - server IP address (non-null; on Linux dotted-decimal only) + // portServ - server port + // Returns true if the initial TCP connect succeeded, false otherwise. + // false does not mean the session was not started: the sender thread is already + // running and automatic reconnect continues in the background. + // Safe to call again while session is active: returns current connection state, + // does not start a second sender thread. SV_API bool svConnect(const char *moduleName, const char *ipAddrServ, int portServ); - // disconnect of server + // Stop sender thread and disconnect. Must be called before DLL unload / process exit. SV_API void svDisconnect(); - // add Value for rec - // name - Value name (leng max 24) + // Add bool signal value for current cycle. + // name - signal name (non-null, length < 24, must not contain "=begin=" or "=end=") + // onlyFront - if true, bool pulse is cleared on inactive cycles SV_API bool svAddBoolValue(const char *name, bool Value, bool onlyFront = false); - // add Value for rec - // name - Value name (leng max 24) + // Add int signal value for current cycle. + // name - signal name (non-null, length < 24, must not contain "=begin=" or "=end=") SV_API bool svAddIntValue(const char *name, int Value); - // add Value for rec - // name - Value name (leng max 24) + // Add float signal value for current cycle. + // name - signal name (non-null, length < 24, must not contain "=begin=" or "=end=") SV_API bool svAddFloatValue(const char *name, float Value); - // frecuence rec + // Set sampling period (cycleRecMs) and packet size (packetSz). + // Must be called before first svAdd*Value and before svConnect; returns false otherwise. SV_API bool svSetParam(int cycleRecMs, int packetSz); -} \ No newline at end of file +} diff --git a/src/SVClient/src/client.cpp b/src/SVClient/src/client.cpp index 8c9c31e4..51f622ac 100644 --- a/src/SVClient/src/client.cpp +++ b/src/SVClient/src/client.cpp @@ -33,6 +33,7 @@ #include #include #include +#include #include #include @@ -46,18 +47,20 @@ struct Config { }; struct ValueRec { - bool isActive; - bool isOnlyFront; - char name[SV_NAMESZ]; - SV_Base::ValueType type; - SV_Base::Value* vals; + bool isActive{}; + bool isOnlyFront{}; + char name[SV_NAMESZ]{}; + SV_Base::ValueType type{}; + std::vector vals; }; -std::atomic_bool _isConnect = false, - _thrStop = false; +std::atomic_bool _isConnect{false}; +std::atomic_bool _thrStop{false}; std::thread _thr; std::mutex _mtx; +std::mutex _sessionMtx; +std::mutex _tcpMtx; std::map _values; @@ -77,36 +80,58 @@ namespace SV { void sendCycle(); bool svConnect(const char *moduleName, const char *ipAddr, int port) { - - if (_isConnect) return true; + + if (!moduleName || !ipAddr) + return false; if ((strlen(moduleName) == 0) || (strlen(moduleName) >= SV_NAMESZ) || strstr(moduleName, messEnd) || strstr(moduleName, messBegin)) { return false; } + std::lock_guard sessionLck(_sessionMtx); + + if (_thr.joinable()) + return _isConnect.load(std::memory_order_acquire); + + _thrStop.store(false); + + { + std::lock_guard dataLck(_mtx); + _curCycle = 0; + } + _module = moduleName; _addrServ = ipAddr; _portServ = port; - _isConnect = SV_Misc::TCPClient::connect(ipAddr, port); - - if (_isConnect) { - _thr = std::thread(sendCycle); + bool connected = false; + { + std::lock_guard tcpLck(_tcpMtx); + connected = SV_Misc::TCPClient::connect(ipAddr, port); + _isConnect.store(connected, std::memory_order_release); } - return _isConnect; + _thr = std::thread(sendCycle); + + return connected; } void svDisconnect() { - - _thrStop = true; - if (_thr.joinable()) _thr.join(); - if (_isConnect){ + std::lock_guard sessionLck(_sessionMtx); + + _thrStop.store(true); + if (_thr.joinable()) + _thr.join(); + + { + std::lock_guard tcpLck(_tcpMtx); SV_Misc::TCPClient::disconnect(); - _isConnect = false; + _isConnect.store(false, std::memory_order_release); } + + _thrStop.store(false); } bool addValue(const char* name, SV_Base::ValueType type, SV_Base::Value val, bool onlyPosFront); @@ -133,38 +158,45 @@ namespace SV { } bool svSetParam(int cycleRecMs, int packetSz) { - - std::lock_guard lck(_mtx); - cng = Config(cycleRecMs, packetSz); + std::lock_guard sessionLck(_sessionMtx); + const bool sessionActive = _thr.joinable(); + + std::lock_guard dataLck(_mtx); + if (!_values.empty() || sessionActive) + return false; + if (cycleRecMs < 1 || packetSz < 1 || packetSz > SV_PACKETSZ_MAX) + return false; + cng = Config(cycleRecMs, packetSz); return true; } - + bool addValue(const char* name, SV_Base::ValueType type, SV_Base::Value val, bool onlyPosFront) { - - if (_values.find(name) == _values.end()) { - if ((strlen(name) == 0) || (strlen(name) >= SV_NAMESZ) || - strstr(name, "=end=") || strstr(name, "=begin=")){ - return false; - } - ValueRec vr; - vr.vals = new SV_Base::Value[SV_PACKETSZ]; - memset(vr.vals, 0, sizeof(SV_Base::Value) * SV_PACKETSZ); - - strcpy(vr.name, name); + + if (!name) + return false; + if (strlen(name) == 0 || strlen(name) >= SV_NAMESZ || + strstr(name, messEnd) || strstr(name, messBegin)) { + return false; + } + + std::lock_guard lck(_mtx); + + auto [it, inserted] = _values.try_emplace(name); + if (inserted) { + ValueRec& vr = it->second; + strncpy(vr.name, name, SV_NAMESZ - 1); vr.type = type; vr.isOnlyFront = onlyPosFront; vr.isActive = false; - { std::lock_guard lck(_mtx); - _values.insert({ name, vr }); - } - } - { std::lock_guard lck(_mtx); - ValueRec& vr = _values[name]; - vr.vals[_curCycle] = val; - vr.isActive = true; + vr.vals.resize(cng.packetSz); + } else if (it->second.type != type) { + return false; } + + it->second.vals[_curCycle] = val; + it->second.isActive = true; return true; } @@ -172,28 +204,36 @@ namespace SV { if (_values.empty()) return ""; - size_t SINT = sizeof(int32_t), - /* val name type vals */ - valSz = SV_NAMESZ + SINT + SINT * SV_PACKETSZ, - /* mod name vals */ - dataSz = SV_NAMESZ + valSz * _values.size(), - - startSz = strlen(messBegin), endSz = strlen(messEnd), offs = 0, - /* dataSz */ - messSz = startSz + SINT + dataSz + endSz; + const int packetSz = cng.packetSz; + const size_t szInt = sizeof(int32_t); + const size_t valSz = SV_NAMESZ + sizeof(SV_Base::ValueType) + + sizeof(SV_Base::Value) * static_cast(packetSz); + const size_t dataSz = SV_NAMESZ + valSz * _values.size(); + const size_t startSz = strlen(messBegin); + const size_t endSz = strlen(messEnd); + const size_t messSz = startSz + szInt + dataSz + endSz; std::string data(messSz, '\0'); - char* dptr = (char*)data.c_str(); + char* dptr = data.data(); + size_t offs = 0; memcpy(dptr, messBegin, startSz); offs += startSz; - memcpy(dptr + offs, &dataSz, SINT); offs += SINT; - memcpy(dptr + offs, _module.data(), SV_NAMESZ); offs += SV_NAMESZ; + const int32_t dataSzField = static_cast(dataSz); + memcpy(dptr + offs, &dataSzField, szInt); offs += szInt; + + char modBuf[SV_NAMESZ] = {}; + strncpy(modBuf, _module.c_str(), SV_NAMESZ - 1); + memcpy(dptr + offs, modBuf, SV_NAMESZ); offs += SV_NAMESZ; + + const size_t valsBytes = sizeof(SV_Base::Value) * static_cast(packetSz); for (const auto& v : _values) { memcpy(dptr + offs, v.second.name, SV_NAMESZ); - memcpy(dptr + offs + SV_NAMESZ, &v.second.type, SINT); - memcpy(dptr + offs + SV_NAMESZ + SINT, v.second.vals, SV_PACKETSZ * SINT); - offs += valSz; + offs += SV_NAMESZ; + memcpy(dptr + offs, &v.second.type, sizeof(SV_Base::ValueType)); + offs += sizeof(SV_Base::ValueType); + memcpy(dptr + offs, v.second.vals.data(), valsBytes); + offs += valsBytes; } memcpy(dptr + offs, messEnd, endSz); @@ -201,11 +241,6 @@ namespace SV { return data; } - bool sendData(const std::string& data){ - std::string out; - return SV_Misc::TCPClient::sendData(data, out, false, true); - } - void sendCycle() { uint64_t cTm = SV_Misc::currDateTimeSinceEpochMs(), @@ -213,50 +248,64 @@ namespace SV { int cDelay = 0; - while (!_thrStop) { + while (!_thrStop.load(std::memory_order_acquire)) { - if (!_isConnect){ - _isConnect = SV_Misc::TCPClient::connect(_addrServ, _portServ); + if (!_isConnect.load(std::memory_order_acquire)) { + std::lock_guard tcpLck(_tcpMtx); + const bool ok = SV_Misc::TCPClient::connect(_addrServ, _portServ); + _isConnect.store(ok, std::memory_order_release); } cTm = SV_Misc::currDateTimeSinceEpochMs(); int tmDiff = int(cTm - prevTm) - cDelay; prevTm = cTm; - + std::string data; - { std::lock_guard lck(_mtx); - - int prevCyc = _curCycle - 1; - if (prevCyc < 0) - prevCyc = SV_PACKETSZ - 1; - - for (auto it = _values.begin(); it != _values.end(); ++it) { - if (!it->second.isActive) { - it->second.vals[_curCycle] = it->second.vals[prevCyc]; - - if ((it->second.type == SV_Base::ValueType::BOOL) && it->second.isOnlyFront) - it->second.vals[_curCycle].vBool = false; - } - it->second.isActive = false; - } + int cycleRecMs = 0; + const bool isConnected = _isConnect.load(std::memory_order_acquire); + + { + std::lock_guard lck(_mtx); + + cycleRecMs = cng.cycleRecMs; + const int packetSz = cng.packetSz; + + int prevCyc = _curCycle - 1; + if (prevCyc < 0) + prevCyc = packetSz - 1; - if (_curCycle < SV_PACKETSZ - 1) { - ++_curCycle; - }else { - _curCycle = 0; - if (_isConnect && !_values.empty()){ - data = prepareData(); - } + for (auto it = _values.begin(); it != _values.end(); ++it) { + if (!it->second.isActive) { + it->second.vals[_curCycle] = it->second.vals[prevCyc]; + + if ((it->second.type == SV_Base::ValueType::BOOL) && it->second.isOnlyFront) + it->second.vals[_curCycle].vBool = false; + } + it->second.isActive = false; + } + + if (_curCycle < packetSz - 1) { + ++_curCycle; + } else { + _curCycle = 0; + if (isConnected && !_values.empty()) { + data = prepareData(); } + } } - if (_isConnect && !data.empty()){ - _isConnect = sendData(data); + + if (isConnected && !data.empty()) { + std::lock_guard tcpLck(_tcpMtx); + std::string out; + const bool ok = SV_Misc::TCPClient::sendData(data, out, false, true); + if (!ok) + _isConnect.store(false, std::memory_order_release); } - - cDelay = (SV_CYCLEREC_MS - tmDiff) > 0 ? (SV_CYCLEREC_MS - tmDiff) : 0; - if (cDelay > 0) { - SV_Misc::sleepMs(cDelay); + + cDelay = (cycleRecMs - tmDiff) > 0 ? (cycleRecMs - tmDiff) : 0; + if (cDelay > 0) { + SV_Misc::sleepMs(cDelay); } } } -} \ No newline at end of file +} diff --git a/src/SVDocker/src/main.cpp b/src/SVDocker/src/main.cpp index 3a24dc99..628985f7 100644 --- a/src/SVDocker/src/main.cpp +++ b/src/SVDocker/src/main.cpp @@ -122,11 +122,6 @@ int main(int argc, char* argv[]){ QCoreApplication a(argc, argv); - SV_Misc::TCPServer::setErrorCBack(statusMess); - SV_Misc::TCPServer::setDataCBack(SV_Srv::receiveData); - - SV_Srv::setStatusCBack(statusMess); - config cng; std::string iniPath = argc > 1 ? argv[1] : a.applicationDirPath().toStdString() + "/svdocker.ini"; @@ -141,8 +136,14 @@ int main(int argc, char* argv[]){ scng.outArchiveName = cng.outArchiveName; scng.outArchivePath = cng.outArchivePath; - if (SV_Srv::startServer(scng) && SV_Misc::TCPServer::start("0.0.0.0", cng.tcp_port)){ - statusMess("TCP server run port: " + std::to_string(cng.tcp_port)); + if (SV_Srv::startServer(scng, statusMess)) { + const auto tcpErr = SV_Misc::TCPServer::start("0.0.0.0", cng.tcp_port, SV_Srv::receiveData, statusMess); + if (tcpErr.empty()) + statusMess("TCP server run port: " + std::to_string(cng.tcp_port)); + else { + statusMess("TCP server not run port: " + std::to_string(cng.tcp_port) + ": " + tcpErr); + return -1; + } } else{ statusMess("TCP server not run port: " + std::to_string(cng.tcp_port)); diff --git a/src/SVMisc/README.md b/src/SVMisc/README.md new file mode 100644 index 00000000..ea7999b3 --- /dev/null +++ b/src/SVMisc/README.md @@ -0,0 +1,149 @@ +# SVMisc + +Статическая библиотека общих утилит SVisual: TCP client/server (libuv), асинхронный logger, функции времени/файловой системы, PLC-style таймеры и фронты. + +Собирается как `libSVMisc.a` / static lib на Windows. + +## Сборка + +```bash +cd src/build +cmake .. +cmake --build . --target SVMisc +``` + +| Платформа | Зависимости | +|-----------|-------------| +| Linux | `libuv`, `pthread` | +| Windows | `libuv`, `ws2_32`, `Iphlpapi`, `psapi`, `Userenv` | + +## Состав модуля + +| Компонент | Header | Назначение | +|-----------|--------|------------| +| `TCPServer` | `SVMisc/tcp_server.h` | TCP-сервер на libuv (worker thread) | +| `TCPClient` | `SVMisc/tcp_client.h` | Блокирующий TCP-клиент (sockets) | +| `Logger` | `SVMisc/logger.h` | Асинхронная запись логов в файл | +| helpers | `SVMisc/misc.h` | Дата/время, `createSubDirectory`, `sleepMs` | +| `TimerDelay` | `SVMisc/timer_delay.h` | ON/OFF delay, импульсы сек/мин/час | +| `Front` | `SVMisc/front.h` | Положительный/отрицательный фронт bool | + +Использование в проекте: + +- `TCPServer` — `SVDocker`, `SVWriterExample`, `SVMonitor` → callback `SV_Srv::receiveData` +- `TCPClient` — `SVClient` (отправка пакетов, `onlySend=true`) +- `Logger` — `SVMonitor` +- `misc` — `SVServer` (archive, timestamps), Logger, UI + +## TCPServer + +### API + +```cpp +const auto err = SV_Misc::TCPServer::start("0.0.0.0", 2144, + SV_Srv::receiveData, + [](const std::string& err) { /* ... */ }); +if (err.empty()) { + // listen готов (async mode, timeout 5 s) +} +``` + +| Метод | Поведение | +|-------|-----------| +| `start(...)` | Пустая строка — успех; иначе текст ошибки (init/listen/timeout/already running) | +| `stop()` | Идempotent shutdown; безопасен из `dataCBack` (см. ниже) | + +Повторный `start()` при already running — `"already running"` без блокировки. + +```cpp +SV_Misc::TCPServer::stop(); +``` + +### Lifecycle (async mode) + +``` +Idle ──start()──► Running ──stop()──► Draining ──► Idle + │ │ │ + └── start() ──► "already running" └── start() снова OK +``` + +- Собственный `uv_loop_t` (не `uv_default_loop()`). +- `stop()`: async → закрытие server/clients/async → `uv_run` возвращается → `uv_loop_close` → `u_loop=nullptr`. +- При startup abort/timeout worker проверяет `stopping` и выходит без зависания `join()`. +- `start()` перед spawn: `error.clear()`, join только завершённого failed worker (`!running`). + +### stop() из dataCBack + +`dataCBack` вызывается в libuv worker thread. `stop()` из callback: + +1. Выставляет `stopping`, шлёт `uv_async_send` (если loop уже init). +2. **Не делает self-join** — вместо этого `detach()` worker thread. +3. Worker завершает `uv_run` и выходит; повторный `start()` join'ит detached thread не требует (не joinable). + +Рекомендация: callbacks и error handler задавать **до** `start()`. + +### Sync mode (`intoOtherThread=false`) + +`startImpl` блокирует caller thread в `uv_run`. Возвращает `error.empty()` после штатной остановки (не проверяет `running` — он уже сброшен в `drainAndCloseLoop`). + +В текущих call sites используется только async mode (default). + +## TCPClient + +Один глобальный socket на процесс (namespace `SV_Misc::TCPClient`). + +```cpp +if (SV_Misc::TCPClient::connect("127.0.0.1", 2144)) { + std::string out; + SV_Misc::TCPClient::sendData(packet, out, false, true); // onlySend +} +SV_Misc::TCPClient::disconnect(); +``` + +| Особенность | Детали | +|-------------|--------| +| Win32 | `WSAStartup` один раз на сессию, `freeaddrinfo` на всех путях | +| Linux | cleanup fd на fail-path connect; `errno` в recv loop | +| `onlySend=true` | Ответ не читается (режим SVClient) | + +Не thread-safe: один поток отправки на процесс. + +## Logger + +Header-only класс, writer thread + ring buffer (100 сообщений). + +```cpp +SV_Misc::Logger lg("/var/log/sv/monitor.log"); +lg.writeLine("message"); +lg.setPathFile("/other/path.log"); +``` + +| Поведение | Детали | +|-----------|--------| +| Overflow | Новое сообщение отбрасывается | +| Shutdown | Drain очереди в деструкторе | +| Path | `createSubDirectory` для parent path файла | + +## misc helpers + +```cpp +SV_Misc::currDateTime(); // UI: %d-%m-%y %H:%M:%S +SV_Misc::currDateTimeMs(); // лог: %Y-%m-%d %H:%M:%S:ms +SV_Misc::currDateTimeSinceEpochMs(); +SV_Misc::createSubDirectory("path/to/file.log"); // создаёт parent dirs +SV_Misc::sleepMs(100); +``` + +- Функции времени — thread-safe (`localtime_r` / `localtime_s`). +- `createSubDirectory("")` и `"file.log"` → `true`; при `EEXIST` проверяется, что путь — directory (не regular file). + +## TimerDelay / Front + +Single-threaded PLC-style утилиты для циклов опроса (archive, trigger/script dialogs). Вызывать из одного потока после `update()` раз в цикл. + +## Ограничения + +- `TCPServer` — singleton (`static srv`); один экземпляр на процесс. +- `TCPClient` — не для параллельных connect/send из разных потоков. +- IPv4 only (`uv_ip4_addr` / `AF_INET`). +- Ответ сервера клиенту через `outMess` в `dataCBack` поддерживается на уровне libuv, но `SV_Srv::receiveData` ответ не использует. diff --git a/src/SVMisc/logger.h b/src/SVMisc/logger.h index a25cd7a2..6282061d 100644 --- a/src/SVMisc/logger.h +++ b/src/SVMisc/logger.h @@ -32,102 +32,117 @@ #include #include #include -#include namespace SV_Misc { - class Logger { + class Logger { public: - - Logger(const std::string& pathFile = ""): - pathFile_(pathFile){ + explicit Logger(const std::string& pathFile = "") : + pathFile_(pathFile) { createSubDirectory(pathFile); - deqMess_.resize(MAX_CNT_MESS); - thrWriteMess_ = std::thread(&Logger::writeCycle, this); } - + ~Logger() { { - std::lock_guard lck (mtxRd_); + std::lock_guard lck(mtx_); fStop_ = true; cval_.notify_one(); - } - if (thrWriteMess_.joinable()){ - thrWriteMess_.join(); } + if (thrWriteMess_.joinable()) + thrWriteMess_.join(); } void setPathFile(const std::string& pathFile) { createSubDirectory(pathFile); - - std::lock_guard lck(mtxRd_); - + std::lock_guard lck(mtx_); pathFile_ = pathFile; } - - void writeLine(const std::string &mess) { - std::lock_guard lck(mtxWr_); + void writeLine(const std::string& mess) { + std::lock_guard lck(mtx_); + + const int nextWrite = (writeMessCnt_ + 1) % MAX_CNT_MESS; + if (nextWrite == readMessCnt_) + return; deqMess_[writeMessCnt_] = Message{ currDateTimeMs(), mess }; - ++writeMessCnt_; - if (writeMessCnt_ >= MAX_CNT_MESS){ - writeMessCnt_ = 0; - } + writeMessCnt_ = nextWrite; cval_.notify_one(); } private: + static const int MAX_CNT_MESS = 100; - const int MAX_CNT_MESS = 100; + struct Message { + std::string cTime; + std::string mess; + }; std::string pathFile_; + int readMessCnt_ = 0; + int writeMessCnt_ = 0; - int readMessCnt_ = 0, writeMessCnt_ = 0; - - struct Message { - std::string cTime; - std::string mess; - }; std::vector deqMess_; - std::mutex mtxWr_, mtxRd_; + std::mutex mtx_; std::thread thrWriteMess_; std::condition_variable cval_; - std::atomic_bool fStop_{}; - - void writeCycle() { + bool fStop_ = false; - while (!fStop_) { + std::vector drainLocked() { + std::vector mess; + while (readMessCnt_ != writeMessCnt_) { + mess.push_back(deqMess_[readMessCnt_]); + readMessCnt_ = (readMessCnt_ + 1) % MAX_CNT_MESS; + } + return mess; + } + + static void writeBatch(const std::string& path, const std::vector& mess) { + if (mess.empty() || path.empty()) + return; - std::unique_lock lck(mtxRd_); - cval_.wait(lck); + std::ofstream slg(path.c_str(), std::ios::app); + if (!slg.good()) + return; - std::ofstream slg(pathFile_.c_str(), std::ios::app); - std::vector mess; - while (readMess(mess)) { - for (const auto& m : mess){ - slg << "[" << m.cTime << "] " << m.mess << std::endl; + for (const auto& m : mess) + slg << "[" << m.cTime << "] " << m.mess << std::endl; + } + + void writeCycle() { + while (true) { + std::vector batch; + std::string path; + bool stop = false; + + { + std::unique_lock lck(mtx_); + cval_.wait(lck, [&] { + return fStop_ || readMessCnt_ != writeMessCnt_; + }); + + batch = drainLocked(); + path = pathFile_; + stop = fStop_; + } + + writeBatch(path, batch); + + if (stop) { + std::vector tail; + std::string tailPath; + { + std::lock_guard lck(mtx_); + tail = drainLocked(); + tailPath = pathFile_; } + writeBatch(tailPath, tail); + break; } - slg.close(); - } - } - bool readMess(std::vector& mess){ - std::lock_guard lck(mtxWr_); - - mess.clear(); - while (readMessCnt_ != writeMessCnt_){ - mess.push_back(deqMess_[readMessCnt_]); - - ++readMessCnt_; - if (readMessCnt_ >= MAX_CNT_MESS){ - readMessCnt_ = 0; - } } - return !mess.empty(); } }; -} \ No newline at end of file +} diff --git a/src/SVMisc/misc.h b/src/SVMisc/misc.h index 672d0743..f498fe55 100644 --- a/src/SVMisc/misc.h +++ b/src/SVMisc/misc.h @@ -25,7 +25,6 @@ #pragma once #include -#include #include namespace SV_Misc { @@ -36,23 +35,18 @@ namespace SV_Misc { // тек дата-время %Y%m%d_%H%M std::string currDateTimeEx(); - // тек дата-время %Y-%m-%d %H:%M:%S + // тек дата-время %d-%m-%y %H:%M:%S std::string currDateTime(); // тек дата-время %Y-%m-%d %H:%M:%S:%MS std::string currDateTimeMs(); - // тек дата-время %Y-%m-%d %H:%M:%S - std::string currDateTimeSQL(); - uint64_t currDateTimeSinceEpochMs(); int hourOffsFromUTC(); - std::vector split(std::string str, const char *sep); - // автосоздание субдиректорий bool createSubDirectory(std::string strDirs); void sleepMs(uint64_t ms); -} \ No newline at end of file +} diff --git a/src/SVMisc/spin_lock.h b/src/SVMisc/spin_lock.h deleted file mode 100644 index 7bc690de..00000000 --- a/src/SVMisc/spin_lock.h +++ /dev/null @@ -1,65 +0,0 @@ -#include -#include -#include - - -namespace SV_Misc { - -class SpinLock{ -public: - SpinLock() = default; - - SpinLock(const SpinLock&) = delete; - SpinLock& operator=(const SpinLock&) = delete; - - enum Action{ READ, WRITE }; - - void lock(Action act){ - bool isBlock = false; - while (!m_readLock.compare_exchange_weak(isBlock, true)){ - isBlock = false; - std::this_thread::yield(); - } - - isBlock = (act == WRITE) && (m_readCount.load() > 0); - while (!m_writeLock.compare_exchange_weak(isBlock, act == WRITE ? true : false)){ - isBlock = (act == WRITE) && (m_readCount.load() > 0); - } - - if (act == READ){ - ++m_readCount; - } - - m_readLock.store(false); - } - - void unlock(Action act){ - if (act == READ){ - --m_readCount; - }else{ - m_writeLock.store(false); - } - } - -private: - std::atomic m_readCount{0}; - std::atomic m_readLock{false}; - std::atomic m_writeLock{false}; -}; - -class Locker { -public: - Locker(SpinLock& lock, SpinLock::Action act) : - m_lock(lock), - m_act(act) { - m_lock.lock(act); - } - ~Locker() { - m_lock.unlock(m_act); - } - -private: - SpinLock& m_lock; - SpinLock::Action m_act; -}; -} \ No newline at end of file diff --git a/src/SVMisc/src/aux_func.cpp b/src/SVMisc/src/aux_func.cpp index c8abd725..f8682272 100644 --- a/src/SVMisc/src/aux_func.cpp +++ b/src/SVMisc/src/aux_func.cpp @@ -23,133 +23,178 @@ // THE SOFTWARE. // #include "SVMisc/misc.h" -#include #include #include #include -#include #ifdef WIN32 #include #else -#include #include +#include #endif -using namespace std; - namespace SV_Misc { +namespace { - // тек дата %Y%m%d - string currDateS() { +bool localTm(time_t t, tm& out) { +#ifdef _WIN32 + return localtime_s(&out, &t) == 0; +#else + return localtime_r(&t, &out) != nullptr; +#endif +} - time_t ct = time(nullptr); - tm* lct = localtime(&ct); +bool gmTm(time_t t, tm& out) { +#ifdef _WIN32 + return gmtime_s(&out, &t) == 0; +#else + return gmtime_r(&t, &out) != nullptr; +#endif +} - char curDate[32]; - strftime(curDate, 32, "%Y%m%d", lct); +bool isDirectory(const std::string& path) { +#if defined(_WIN32) + const DWORD attr = GetFileAttributesA(path.c_str()); + if (attr == INVALID_FILE_ATTRIBUTES) + return false; + return (attr & FILE_ATTRIBUTE_DIRECTORY) != 0; +#else + struct stat st{}; + if (stat(path.c_str(), &st) != 0) + return false; + return S_ISDIR(st.st_mode); +#endif +} - return curDate; - } +bool mkdirSegment(const std::string& path) { + if (path.empty()) + return true; +#if defined(_WIN32) + if (CreateDirectoryA(path.c_str(), NULL)) + return true; + if (GetLastError() == ERROR_ALREADY_EXISTS) + return isDirectory(path); + return false; +#else + if (mkdir(path.c_str(), 0755) == 0) + return true; + if (errno == EEXIST) + return isDirectory(path); + return false; +#endif +} - // тек дата-время %Y%m%d_%H%M - string currDateTimeEx() { +std::string parentDirectoryPath(const std::string& path) { + const auto pos = path.find_last_of("/\\"); + if (pos == std::string::npos) + return {}; + return path.substr(0, pos); +} +} // namespace + + std::string currDateS() { time_t ct = time(nullptr); - tm* lct = localtime(&ct); + tm lct{}; + if (!localTm(ct, lct)) + return {}; char curDate[32]; - strftime(curDate, 32, "%Y%m%d_%H%M", lct); - + strftime(curDate, sizeof(curDate), "%Y%m%d", &lct); return curDate; } - // тек дата-время %Y-%m-%d %H:%M:%S - string currDateTime() { - + std::string currDateTimeEx() { time_t ct = time(nullptr); - tm* lct = localtime(&ct); + tm lct{}; + if (!localTm(ct, lct)) + return {}; char curDate[32]; - strftime(curDate, 32, "%d-%m-%y %H:%M:%S", lct); - + strftime(curDate, sizeof(curDate), "%Y%m%d_%H%M", &lct); return curDate; } - // тек дата-время %Y-%m-%d %H:%M:%S:%MS - string currDateTimeMs() { - + std::string currDateTime() { time_t ct = time(nullptr); - tm* lct = localtime(&ct); - - uint64_t ms = std::chrono::time_point_cast(std::chrono::system_clock::now()).time_since_epoch().count(); - uint64_t mspr = ms / 1000; - ms -= mspr * 1000; + tm lct{}; + if (!localTm(ct, lct)) + return {}; char curDate[32]; - strftime(curDate, 32, "%Y-%m-%d %H:%M:%S:", lct); - - return string(curDate) + to_string(ms); + strftime(curDate, sizeof(curDate), "%d-%m-%y %H:%M:%S", &lct); + return curDate; } - // тек дата-время %Y-%m-%d %H:%M:%S - std::string currDateTimeSQL() { - + std::string currDateTimeMs() { time_t ct = time(nullptr); - tm* lct = localtime(&ct); + tm lct{}; + if (!localTm(ct, lct)) + return {}; - char curDate[32]; - strftime(curDate, 32, "%Y-%m-%d %H:%M:%S", (const tm *)&lct); + const uint64_t ms = std::chrono::time_point_cast( + std::chrono::system_clock::now()).time_since_epoch().count() % 1000; - return curDate; + char curDate[32]; + strftime(curDate, sizeof(curDate), "%Y-%m-%d %H:%M:%S:", &lct); + return std::string(curDate) + std::to_string(ms); } - uint64_t currDateTimeSinceEpochMs(){ - - auto now = std::chrono::system_clock::now(); - auto now_ms = std::chrono::time_point_cast(now); - - return now_ms.time_since_epoch().count(); + uint64_t currDateTimeSinceEpochMs() { + const auto now = std::chrono::system_clock::now(); + return std::chrono::time_point_cast(now) + .time_since_epoch().count(); } int hourOffsFromUTC() { + const time_t rawtime = time(nullptr); + tm gmt{}; + tm local{}; + if (!gmTm(rawtime, gmt) || !localTm(rawtime, local)) + return 0; + + const time_t gmtSec = mktime(&gmt); + return int((rawtime - gmtSec + (local.tm_isdst ? 3600 : 0)) / 3600); + } - time_t rawtime = time(NULL); - struct tm *ptm = gmtime(&rawtime); - time_t gmt = mktime(ptm); - ptm = localtime(&rawtime); + bool createSubDirectory(std::string path) { + if (path.empty()) + return true; - return int((rawtime - gmt + (ptm->tm_isdst ? 3600 : 0)) / 3600); - } - - bool isFileExist(const std::string &name) { - ifstream f(name.c_str()); - return f.good(); - } + const std::string dirPath = parentDirectoryPath(path); + if (dirPath.empty()) + return true; - // автосоздание субдиректорий - bool createSubDirectory(string strDirs) { - if (isFileExist(strDirs)) return true; + std::string accumulated; + accumulated.reserve(dirPath.size()); + + for (size_t i = 0; i < dirPath.size(); ++i) { + const char ch = dirPath[i]; + if (ch != '\\' && ch != '/') { + accumulated += ch; + continue; + } - int sz = int(strDirs.size()), ret = 0; - string strTmp = ""; - for (int i = 0; i < sz; ++i) { - char ch = strDirs[i]; - if (ch != '\\' && ch != '/') strTmp += ch; - else { #if defined(_WIN32) - strTmp += "\\"; - ret = CreateDirectoryA(strTmp.c_str(), NULL); + accumulated += '\\'; #else - strTmp += "/"; - ret = mkdir(strTmp.c_str(), 0733); + accumulated += '/'; #endif - } + if (accumulated.size() <= 1) + continue; + + if (!mkdirSegment(accumulated)) + return false; } - return ret == 0; + + if (!accumulated.empty() && !mkdirSegment(accumulated)) + return false; + + return true; } - void sleepMs(uint64_t ms){ + void sleepMs(uint64_t ms) { std::this_thread::sleep_for(std::chrono::milliseconds(ms)); } -} \ No newline at end of file +} diff --git a/src/SVMisc/src/tcp_client.cpp b/src/SVMisc/src/tcp_client.cpp index 2f23a30b..518b1db1 100644 --- a/src/SVMisc/src/tcp_client.cpp +++ b/src/SVMisc/src/tcp_client.cpp @@ -28,86 +28,117 @@ #include -namespace SV_Misc{ - namespace TCPClient { - #ifdef WIN32 - #include #include +#else +#include +#include +#include +#include +#include +#include +#include +#endif + +namespace SV_Misc { +namespace TCPClient { + +#ifdef WIN32 + +namespace { SOCKET _socket = INVALID_SOCKET; - -bool connect(const std::string& addr, int port, bool noBlock) { +bool _wsaStarted = false; +bool ensureWsa() { + if (_wsaStarted) + return true; WSADATA wsaData; - _socket = INVALID_SOCKET; - struct addrinfo *adInfo = NULL, hints; - - // Initialize Winsock - if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0) { + if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0) return false; + _wsaStarted = true; + return true; +} + +void releaseWsa() { + if (!_wsaStarted) + return; + WSACleanup(); + _wsaStarted = false; +} + +void closeSocket() { + if (_socket != INVALID_SOCKET) { + shutdown(_socket, SD_BOTH); + closesocket(_socket); + _socket = INVALID_SOCKET; } +} + +} // namespace + +bool connect(const std::string& addr, int port, bool noBlock) { + closeSocket(); - ZeroMemory(&hints, sizeof(hints)); + if (!ensureWsa()) + return false; + + addrinfo hints{}; hints.ai_family = AF_INET; hints.ai_socktype = SOCK_STREAM; hints.ai_protocol = IPPROTO_TCP; - // Resolve the server address and port char cport[11]; - (sprintf)(cport, "%d", port); - if (getaddrinfo(addr.c_str(), cport, &hints, &adInfo) != 0) { - WSACleanup(); + (sprintf)(cport, "%d", port); + + addrinfo* adInfo = nullptr; + if (getaddrinfo(addr.c_str(), cport, &hints, &adInfo) != 0) return false; - } _socket = socket(AF_INET, SOCK_STREAM, 0); if (_socket == INVALID_SOCKET) { - WSACleanup(); + freeaddrinfo(adInfo); return false; } - // connects to server. - if (connect(_socket, adInfo->ai_addr, (int)adInfo->ai_addrlen) == SOCKET_ERROR) { - closesocket(_socket); + if (::connect(_socket, adInfo->ai_addr, (int)adInfo->ai_addrlen) == SOCKET_ERROR) { + freeaddrinfo(adInfo); + closeSocket(); return false; } + freeaddrinfo(adInfo); + if (noBlock) { - BOOL l = TRUE; - int er = ioctlsocket(_socket, FIONBIO, (unsigned long *)&l); + u_long mode = 1; + ioctlsocket(_socket, FIONBIO, &mode); } return true; } bool disconnect() { - - if (_socket != INVALID_SOCKET){ - shutdown(_socket, SD_BOTH); - closesocket(_socket); - WSACleanup(); - } + closeSocket(); + releaseWsa(); return true; } -int sendAll(const std::string& mess, int flags){ - - int total = 0, - len = int(mess.size()); - while (total < len){ - int n = send(_socket, mess.c_str() + total, len - total, flags); - if (n == SOCKET_ERROR){ - total = SOCKET_ERROR; - break; - } +int sendAll(const std::string& mess, int flags) { + int total = 0; + const int len = int(mess.size()); + while (total < len) { + const int n = send(_socket, mess.c_str() + total, len - total, flags); + if (n == SOCKET_ERROR) + return SOCKET_ERROR; total += n; } return total; } -bool sendData(const std::string& in, std::string &out, bool disconn, bool onlySend) { +bool sendData(const std::string& in, std::string& out, bool disconn, bool onlySend) { + if (_socket == INVALID_SOCKET) + return false; if (sendAll(in, 0) == SOCKET_ERROR) { disconnect(); @@ -115,59 +146,69 @@ bool sendData(const std::string& in, std::string &out, bool disconn, bool onlySe } if (!onlySend) { - - // Receive until the peer closes the connection const int recvbuflen = 2048; static char recvbuf[recvbuflen]; out.clear(); while (true) { - int rlen = recv(_socket, recvbuf, recvbuflen, 0); + const int rlen = recv(_socket, recvbuf, recvbuflen, 0); if (rlen > 0) { - size_t csz = out.size(); + const size_t csz = out.size(); out.resize(csz + size_t(rlen)); memcpy((char*)out.data() + csz, recvbuf, rlen); - if (recvbuf[rlen - 1] == '\0') break; - } - else if (WSAGetLastError() == WSAEWOULDBLOCK) { - SV_Misc::sleepMs(0); + if (recvbuf[rlen - 1] == '\0') + break; + } else if (rlen == 0) { + break; + } else if (rlen == SOCKET_ERROR && WSAGetLastError() == WSAEWOULDBLOCK) { + SV_Misc::sleepMs(1); + } else { + break; } - else break; } } - if (disconn) disconnect(); + if (disconn) + disconnect(); return true; } - + #else -#include -#include -#include -#include -#include -#include +namespace { int _socket = -1; -bool connect(const std::string& addr, int port, bool noBlock) { +void closeSocket() { + if (_socket >= 0) { + ::close(_socket); + _socket = -1; + } +} - sockaddr_in saddr; +} // namespace - _socket = socket(AF_INET, SOCK_STREAM, 0); - if(_socket < 0) +bool connect(const std::string& addr, int port, bool noBlock) { + closeSocket(); + + _socket = ::socket(AF_INET, SOCK_STREAM, 0); + if (_socket < 0) return false; + sockaddr_in saddr{}; saddr.sin_family = AF_INET; saddr.sin_port = htons(port); - if (inet_aton(addr.c_str(), &saddr.sin_addr) == 0) + if (inet_aton(addr.c_str(), &saddr.sin_addr) == 0) { + closeSocket(); return false; + } - if(connect(_socket, (struct sockaddr*)&saddr, sizeof(saddr)) < 0) + if (::connect(_socket, reinterpret_cast(&saddr), sizeof(saddr)) < 0) { + closeSocket(); return false; + } if (noBlock) fcntl(_socket, F_SETFL, O_NONBLOCK); @@ -176,69 +217,60 @@ bool connect(const std::string& addr, int port, bool noBlock) { } bool disconnect() { - - if (_socket > 0) close(_socket); - _socket = -1; + closeSocket(); return true; } -int sendAll(const std::string& mess, int flags){ - - int total = 0, - len = int(mess.size()); +int sendAll(const std::string& mess, int flags) { + int total = 0; + const int len = int(mess.size()); while (total < len) { - int n = send(_socket, mess.c_str() + total, len - total, flags); - if (n == -1) { - total = -1; - break; - } + const int n = send(_socket, mess.c_str() + total, len - total, flags); + if (n == -1) + return -1; total += n; } return total; } -bool sendData(const std::string& in, std::string &out, bool disconn, bool onlySend) { - - if (_socket < 0){ +bool sendData(const std::string& in, std::string& out, bool disconn, bool onlySend) { + if (_socket < 0) return false; - } - // Send an initial buffer - int ret = sendAll(in, 0); + const int ret = sendAll(in, 0); if (ret == -1) { disconnect(); return false; } if (!onlySend) { - - // Receive until the peer closes the connection const int recvbuflen = 2048; static char recvbuf[recvbuflen]; out.clear(); while (true) { - int rlen = recv(_socket, recvbuf, recvbuflen, 0); + const int rlen = recv(_socket, recvbuf, recvbuflen, 0); if (rlen > 0) { - - size_t csz = out.size(); + const size_t csz = out.size(); out.resize(csz + rlen); - memcpy((char*) out.data() + csz, recvbuf, rlen); - - if (recvbuf[rlen - 1] == '\0') break; - } - else if (rlen == -1) { - SV_Misc::sleepMs(0); - } - else break; + memcpy((char*)out.data() + csz, recvbuf, rlen); + + if (recvbuf[rlen - 1] == '\0') + break; + } else if (rlen == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) { + SV_Misc::sleepMs(1); + } else { + break; + } } } - if (disconn) disconnect(); + if (disconn) + disconnect(); return true; } #endif - + +} } -} \ No newline at end of file diff --git a/src/SVMisc/src/tcp_server.cpp b/src/SVMisc/src/tcp_server.cpp index 8ba0f346..d0fe6bdf 100644 --- a/src/SVMisc/src/tcp_server.cpp +++ b/src/SVMisc/src/tcp_server.cpp @@ -22,43 +22,63 @@ // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN // THE SOFTWARE. // -#include -#include -#include +#include +#include #include #include +#include +#include +#include +#include #include "libuv/uv.h" #include "SVMisc/tcp_server.h" #ifdef __linux__ - #define ULONG size_t + #define ULONG size_t #endif namespace SV_Misc { namespace TCPServer { - + + struct Client_m; + + struct WriteReq { + uv_write_t req{}; + std::string payload; + }; + struct Server_m { - uv_tcp_t u_server; - uv_loop_t* u_loop; + uv_loop_t u_loop_storage{}; + uv_loop_t* u_loop = nullptr; + uv_tcp_t u_server{}; + uv_async_t stop_async{}; + + std::atomic running{false}; + std::atomic stopping{false}; + std::atomic stop_async_inited{false}; + std::atomic server_inited{false}; + DataCBack dataCBack; ErrorCBack errCBack; std::string error; std::thread thr; - - Server_m(): - dataCBack(nullptr), - errCBack(nullptr), - u_loop(nullptr) {}; - }srv; + + std::mutex lifecycle_mtx; + std::mutex clients_mtx; + std::unordered_set clients; + + std::mutex start_mtx; + std::condition_variable start_cval; + bool listen_ready = false; + } srv; struct Client_m { - uv_tcp_t u_client; - uv_write_t u_writeReq; - uv_buf_t u_buf; + uv_tcp_t u_client{}; + uv_buf_t u_buf{}; std::string inMess; std::string outMess; - + Client_m() { u_buf.base = nullptr; u_buf.len = 0; @@ -70,145 +90,380 @@ namespace SV_Misc { } }; -#define CHECK(func, mess){ \ - int fsts = func; \ - if (fsts != 0){ \ - srv.error = std::string(mess) + " code " + std::to_string(fsts); \ - if (srv.errCBack) \ - srv.errCBack(srv.error); \ - return; \ - } \ - } - void on_close(uv_handle_t* u_client) { - Client_m* client = static_cast(u_client->data); + void reportError(const std::string& msg) { + { + std::lock_guard lck(srv.lifecycle_mtx); + srv.error = msg; + } + if (srv.errCBack) + srv.errCBack(msg); + } + + void signalListenReady() { + { + std::lock_guard lck(srv.start_mtx); + srv.listen_ready = true; + } + srv.start_cval.notify_one(); + } + + void on_client_close(uv_handle_t* handle); + + void closeAllClientsSnapshot() { + std::vector snapshot; + { + std::lock_guard lck(srv.clients_mtx); + snapshot.assign(srv.clients.begin(), srv.clients.end()); + } + for (Client_m* client : snapshot) { + if (!uv_is_closing(reinterpret_cast(&client->u_client))) + uv_close(reinterpret_cast(&client->u_client), on_client_close); + } + } + + void drainAndCloseLoop() { + uv_loop_t* loop = nullptr; + { + std::lock_guard lck(srv.lifecycle_mtx); + if (!srv.u_loop) + return; + loop = srv.u_loop; + } + + while (uv_loop_alive(loop)) + uv_run(loop, UV_RUN_ONCE); + + { + std::lock_guard lck(srv.lifecycle_mtx); + if (srv.u_loop != loop) + return; + + uv_loop_close(srv.u_loop); + srv.u_loop = nullptr; + srv.server_inited.store(false, std::memory_order_release); + srv.stop_async_inited.store(false, std::memory_order_release); + srv.running.store(false, std::memory_order_release); + srv.stopping.store(false, std::memory_order_release); + } + } + + void on_client_close(uv_handle_t* handle) { + Client_m* client = static_cast(handle->data); + { + std::lock_guard lck(srv.clients_mtx); + srv.clients.erase(client); + } delete client; } - void alloc_cb(uv_handle_t* u_client, size_t suggested_size, uv_buf_t* buf) { + void on_write(uv_write_t* req, int /*status*/) { + delete static_cast(req->data); + } + void alloc_cb(uv_handle_t* u_client, size_t suggested_size, uv_buf_t* buf) { Client_m* client = static_cast(u_client->data); - if (client->u_buf.len < (ULONG)suggested_size) { - client->u_buf.base = (char*)realloc(client->u_buf.base, suggested_size); - client->u_buf.len = (ULONG)suggested_size; + if (client->u_buf.len < suggested_size) { + char* base = static_cast(realloc(client->u_buf.base, suggested_size)); + if (!base) { + buf->base = nullptr; + buf->len = 0; + if (!uv_is_closing(u_client)) + uv_close(u_client, on_client_close); + return; + } + client->u_buf.base = base; + client->u_buf.len = suggested_size; } *buf = client->u_buf; } - void on_read(uv_stream_t* u_client, ssize_t nread, const uv_buf_t* buf) { - + void on_read(uv_stream_t* u_client, ssize_t nread, const uv_buf_t* /*buf*/) { if (nread > 0) { - Client_m* client = static_cast(u_client->data); - - // копим данные - auto csz = client->inMess.size(); - client->inMess.resize(csz + nread); - memcpy((char*)client->inMess.data() + csz, buf->base, nread); - // передача пользователю + const auto csz = client->inMess.size(); + client->inMess.resize(csz + static_cast(nread)); + memcpy(client->inMess.data() + csz, client->u_buf.base, static_cast(nread)); + if (srv.dataCBack) srv.dataCBack(client->inMess, client->outMess); - // ответ if (!client->outMess.empty()) { + auto* wr = new WriteReq(); + wr->payload = std::move(client->outMess); + wr->req.data = wr; + uv_buf_t resbuf; - resbuf.base = (char*)client->outMess.c_str(); - resbuf.len = (ULONG)client->outMess.size() + 1; + resbuf.base = wr->payload.data(); + resbuf.len = static_cast(wr->payload.size()); - uv_write(&client->u_writeReq, u_client, &resbuf, 1, nullptr); + if (uv_write(&wr->req, u_client, &resbuf, 1, on_write) != 0) { + delete wr; + } } - } - else if (nread < 0) { - if (!uv_is_closing((uv_handle_t*)u_client)) - uv_close((uv_handle_t*)u_client, on_close); + } else if (nread < 0) { + if (!uv_is_closing(reinterpret_cast(u_client))) + uv_close(reinterpret_cast(u_client), on_client_close); } } void on_connect(uv_stream_t* u_server, int sts) { - - CHECK(sts, "on_connect::sts error"); - + if (sts != 0) { + reportError("on_connect::sts error code " + std::to_string(sts)); + return; + } + Client_m* client = new Client_m(); - - CHECK(uv_tcp_init(u_server->loop, &client->u_client), - "on_connect::uv_tcp_init error"); + + if (uv_tcp_init(u_server->loop, &client->u_client) != 0) { + reportError("on_connect::uv_tcp_init error"); + delete client; + return; + } client->u_client.data = client; - CHECK(uv_accept(u_server, (uv_stream_t*)&client->u_client), - "on_connect::uv_accept error"); + if (uv_accept(u_server, reinterpret_cast(&client->u_client)) != 0) { + reportError("on_connect::uv_accept error"); + uv_close(reinterpret_cast(&client->u_client), on_client_close); + return; + } - CHECK(uv_read_start((uv_stream_t *)&client->u_client, alloc_cb, on_read), - "on_connect::uv_read_start error"); + if (uv_read_start(reinterpret_cast(&client->u_client), alloc_cb, on_read) != 0) { + reportError("on_connect::uv_read_start error"); + uv_close(reinterpret_cast(&client->u_client), on_client_close); + return; + } + + { + std::lock_guard lck(srv.clients_mtx); + srv.clients.insert(client); + } } - - bool start(const std::string& addr, uint16_t port, bool intoOtherThread) { - - if (srv.u_loop) return true; - std::condition_variable cval; - auto startImpl = [&cval](std::string addr, uint16_t port){ - srv.u_loop = uv_default_loop(); + void on_server_close(uv_handle_t* /*handle*/); - CHECK(uv_tcp_init(srv.u_loop, &srv.u_server), - "initConnection::uv_tcp_init error"); + void closeStartupHandles() { + if (srv.server_inited.load(std::memory_order_acquire) && + !uv_is_closing(reinterpret_cast(&srv.u_server))) { + uv_close(reinterpret_cast(&srv.u_server), on_server_close); + } + if (srv.stop_async_inited.load(std::memory_order_acquire) && + !uv_is_closing(reinterpret_cast(&srv.stop_async))) { + uv_close(reinterpret_cast(&srv.stop_async), nullptr); + } + } - CHECK(uv_tcp_keepalive(&srv.u_server, true, 60), - "initConnection::uv_tcp_keepalive error"); + void finishStartup(bool signalReady) { + drainAndCloseLoop(); + if (signalReady) + signalListenReady(); + } - struct sockaddr_in address; - CHECK(uv_ip4_addr(addr.c_str(), port, &address), - "initConnection::uv_ip4_addr error"); + void failStartup(const std::string& msg) { + reportError(msg); + closeStartupHandles(); + finishStartup(true); + } - CHECK(uv_tcp_bind(&srv.u_server, (const struct sockaddr*)&address, 0), - "initConnection::uv_tcp_bind error"); + bool abortStartupIfRequested() { + if (!srv.stopping.load()) + return false; - CHECK(uv_listen((uv_stream_t*)&srv.u_server, 1000, on_connect), - "initConnection::uv_listen error"); + closeStartupHandles(); + finishStartup(true); + return true; + } - cval.notify_one(); + void on_server_close(uv_handle_t* /*handle*/) {} - CHECK(uv_run(srv.u_loop, UV_RUN_DEFAULT), - "initConnection::uv_run error"); - }; - if (intoOtherThread){ - srv.thr = std::thread(startImpl, addr, port); - std::mutex mtx; - std::unique_lock lck(mtx); - cval.wait_for(lck, std::chrono::milliseconds(100)); - }else{ - startImpl(addr, port); - } - return srv.error.empty(); + void on_stop_async(uv_async_t* /*handle*/) { + closeAllClientsSnapshot(); + closeStartupHandles(); } - void stop() { - if (srv.u_loop){ - uv_async_t async; - uv_async_init(srv.u_loop, &async, [](uv_async_t* handle){ - uv_stop(srv.u_loop); - uv_loop_close(srv.u_loop); - }); - uv_async_send(&async); + void startImpl(std::string addr, uint16_t port) { + { + std::lock_guard lck(srv.lifecycle_mtx); + srv.u_loop = &srv.u_loop_storage; + } - if (srv.thr.joinable()) srv.thr.join(); - } - } + if (uv_loop_init(srv.u_loop) != 0) { + { + std::lock_guard lck(srv.lifecycle_mtx); + srv.u_loop = nullptr; + } + reportError("initConnection::uv_loop_init error"); + signalListenReady(); + return; + } - bool setDataCBack(DataCBack dcb) { - srv.dataCBack = dcb; - return true; + if (uv_async_init(srv.u_loop, &srv.stop_async, on_stop_async) != 0) { + failStartup("initConnection::uv_async_init error"); + return; + } + srv.stop_async_inited.store(true, std::memory_order_release); + + if (uv_tcp_init(srv.u_loop, &srv.u_server) != 0) { + failStartup("initConnection::uv_tcp_init error"); + return; + } + srv.server_inited.store(true, std::memory_order_release); + + if (uv_tcp_keepalive(&srv.u_server, 1, 60) != 0) { + failStartup("initConnection::uv_tcp_keepalive error"); + return; + } + + sockaddr_in address{}; + if (uv_ip4_addr(addr.c_str(), port, &address) != 0) { + failStartup("initConnection::uv_ip4_addr error"); + return; + } + + if (uv_tcp_bind(&srv.u_server, reinterpret_cast(&address), 0) != 0) { + failStartup("initConnection::uv_tcp_bind error"); + return; + } + + if (abortStartupIfRequested()) + return; + + if (uv_listen(reinterpret_cast(&srv.u_server), 1000, on_connect) != 0) { + failStartup("initConnection::uv_listen error"); + return; + } + + if (abortStartupIfRequested()) + return; + + srv.running.store(true, std::memory_order_release); + signalListenReady(); + + if (abortStartupIfRequested()) + return; + + uv_run(srv.u_loop, UV_RUN_DEFAULT); + + drainAndCloseLoop(); } - bool setErrorCBack(ErrorCBack ecb) { - srv.errCBack = ecb; - return true; + // join from external thread; detach when stop/abort is invoked on worker thread + void joinWorkerThreadIfExternal() { + if (!srv.thr.joinable()) + return; + if (std::this_thread::get_id() == srv.thr.get_id()) { + srv.thr.detach(); + return; + } + srv.thr.join(); + } + + void abortWorker(bool sendAsyncIfPossible) { + bool sendAsync = false; + bool needJoin = false; + + { + std::lock_guard lck(srv.lifecycle_mtx); + srv.stopping.store(true, std::memory_order_release); + sendAsync = sendAsyncIfPossible && + srv.stop_async_inited.load(std::memory_order_acquire) && + srv.u_loop != nullptr; + needJoin = srv.thr.joinable(); + } + + if (sendAsync) + uv_async_send(&srv.stop_async); + + if (needJoin) + joinWorkerThreadIfExternal(); } - std::string errorStr() { + std::string start(const std::string& addr, uint16_t port, + DataCBack dataCb, ErrorCBack errorCb, bool intoOtherThread) { + { + std::lock_guard lck(srv.lifecycle_mtx); + srv.error.clear(); + if (srv.running.load(std::memory_order_acquire) && srv.u_loop != nullptr) + return "already running"; + if (dataCb) + srv.dataCBack = std::move(dataCb); + if (errorCb) + srv.errCBack = std::move(errorCb); + } + + if (srv.thr.joinable() && !srv.running.load(std::memory_order_acquire)) + srv.thr.join(); + + { + std::lock_guard lck(srv.lifecycle_mtx); + if (srv.running.load(std::memory_order_acquire) && srv.u_loop != nullptr) + return "already running"; + srv.stopping.store(false, std::memory_order_release); + } + + { + std::lock_guard lck(srv.start_mtx); + srv.listen_ready = false; + } + + if (intoOtherThread) { + srv.thr = std::thread(startImpl, addr, port); + std::unique_lock lck(srv.start_mtx); + const bool ready = srv.start_cval.wait_for(lck, std::chrono::seconds(5), [] { + return srv.listen_ready; + }); + lck.unlock(); + + if (!ready) { + reportError("initConnection::listen timeout"); + abortWorker(true); + std::lock_guard lifeLck(srv.lifecycle_mtx); + return srv.error; + } + + std::lock_guard lifeLck(srv.lifecycle_mtx); + if (!srv.error.empty()) + return srv.error; + if (!srv.running.load(std::memory_order_acquire)) + return "failed to start"; + return {}; + } + + startImpl(addr, port); + std::lock_guard lck(srv.lifecycle_mtx); return srv.error; } + + void stop() { + bool sendAsync = false; + bool needJoin = false; + + { + std::lock_guard lck(srv.lifecycle_mtx); + if (srv.u_loop == nullptr && !srv.thr.joinable()) + return; + + if (srv.stopping.load()) { + needJoin = srv.thr.joinable(); + } else if (srv.u_loop != nullptr && srv.running.load(std::memory_order_acquire)) { + srv.stopping.store(true, std::memory_order_release); + sendAsync = srv.stop_async_inited.load(std::memory_order_acquire); + needJoin = srv.thr.joinable(); + } else if (srv.thr.joinable()) { + srv.stopping.store(true, std::memory_order_release); + sendAsync = srv.stop_async_inited.load(std::memory_order_acquire) && srv.u_loop != nullptr; + needJoin = true; + } + } + + if (sendAsync) + uv_async_send(&srv.stop_async); + + if (needJoin) + joinWorkerThreadIfExternal(); + } } } diff --git a/src/SVMisc/tcp_server.h b/src/SVMisc/tcp_server.h index 319b6726..f8268df9 100644 --- a/src/SVMisc/tcp_server.h +++ b/src/SVMisc/tcp_server.h @@ -34,14 +34,12 @@ namespace SV_Misc { using ErrorCBack = std::function; - bool start(const std::string& addr, uint16_t port, bool intoOtherThread = true); + std::string start(const std::string& addr, uint16_t port, + DataCBack dataCb = {}, + ErrorCBack errorCb = {}, + bool intoOtherThread = true); + // Safe from dataCBack/worker thread: initiates shutdown and detaches worker (no self-join). void stop(); - - bool setDataCBack(DataCBack); - - bool setErrorCBack(ErrorCBack); - - std::string errorStr(); } } diff --git a/src/SVMonitor/src/main_win.cpp b/src/SVMonitor/src/main_win.cpp index ca14e80b..0b470757 100644 --- a/src/SVMonitor/src/main_win.cpp +++ b/src/SVMonitor/src/main_win.cpp @@ -104,7 +104,12 @@ MainWin::MainWin(QWidget *parent) srvCng.outDataBaseName = cng.outDataBaseName.toStdString(); } + auto statusCb = [this](const std::string& mess){ + statusMess(QString::fromLocal8Bit(mess.c_str())); + }; + if (cng.com_ena){ + const bool srvStarted = SV_Srv::startServer(srvCng, statusCb); for (auto& port : cng.com_ports){ if (port.first.isEmpty() || port.second.isEmpty()) continue; @@ -113,21 +118,23 @@ MainWin::MainWin(QWidget *parent) auto comReader = new SerialPortReader(ccng, this); if (comReader->start()) { statusMess(QString(tr("Прослушивание %1 порта запущено")).arg(port.first)); - SV_Srv::startServer(srvCng); - comReader->setDataCBack(SV_Srv::receiveData); + if (srvStarted) { + comReader->setDataCBack(SV_Srv::receiveData); + } } else { statusMess(QString(tr("%1 порт недоступен")).arg(port.first)); } comReaders_.push_back(comReader); } }else{ - if (SV_Misc::TCPServer::start(cng.tcp_addr.toStdString(), cng.tcp_port)){ + const auto tcpErr = SV_Misc::TCPServer::start(cng.tcp_addr.toStdString(), cng.tcp_port, SV_Srv::receiveData); + if (tcpErr.empty()) { statusMess(QString(tr("TCP cервер запущен: адрес %1 порт %2").arg(cng.tcp_addr).arg(cng.tcp_port))); - SV_Srv::startServer(srvCng); - SV_Misc::TCPServer::setDataCBack(SV_Srv::receiveData); + SV_Srv::startServer(srvCng, statusCb); } else - statusMess(QString(tr("Не удалось запустить TCP сервер: адрес %1 порт %2").arg(cng.tcp_addr).arg(cng.tcp_port))); + statusMess(QString(tr("Не удалось запустить TCP сервер: адрес %1 порт %2 (%3)") + .arg(cng.tcp_addr).arg(cng.tcp_port).arg(QString::fromStdString(tcpErr)))); } // web @@ -819,9 +826,6 @@ void MainWin::load(){ SV_Exp::setGetCopyModuleRefCBack(exportDialog_, getCopyModuleRefSrv); SV_Exp::setGetSignalDataCBack(exportDialog_, getSignalDataSrv); - SV_Srv::setStatusCBack([this](const std::string& mess){ - statusMess(QString::fromLocal8Bit(mess.c_str())); - }); SV_Srv::setOnUpdateSignalsCBack([this](){ QMetaObject::invokeMethod(this, "updateSignals", Qt::AutoConnection); }); @@ -1270,19 +1274,6 @@ MainWin::Config MainWin::getConfig(){ } void MainWin::updateConfig(const MainWin::Config& newCng){ - - SV_Srv::Config srvCng; { - srvCng.cycleRecMs = cng.cycleRecMs; - srvCng.packetSz = cng.packetSz; - srvCng.outArchiveEna = newCng.outArchiveEna; - srvCng.outArchiveHourCnt = newCng.outArchiveHourCnt; - srvCng.outArchiveName = newCng.outArchiveName.toStdString(); - srvCng.outArchivePath = newCng.outArchivePath.toStdString(); - srvCng.outDataBaseEna = newCng.outDataBaseEna; - srvCng.outDataBaseName = newCng.outDataBaseName.toStdString(); - srvCng.outDataBaseAddr = newCng.outDataBaseAddr.toStdString(); - } - SV_Srv::setConfig(srvCng); cng = newCng; } diff --git a/src/SVServer/README.md b/src/SVServer/README.md new file mode 100644 index 00000000..ed4561e8 --- /dev/null +++ b/src/SVServer/README.md @@ -0,0 +1,110 @@ +# SVServer + +Библиотека приёма и обработки потока сигналов SVisual: ring-buffer входящих данных, worker-поток обновления сигналов, запись архива на диск и (опционально) в ClickHouse. + +Собирается как shared library (`libSVServer.so` / `SVServer.dll`), публичный API — `sv_server.h`. + +## Сборка + +```bash +cd src/build +cmake .. -DUSE_ClickHouseDB=OFF # или ON при установленном clickhouse-cpp +cmake --build . --target SVServer +``` + +| Опция CMake | Описание | +|-------------|----------| +| `USE_ClickHouseDB` | Включить `ClickHouseDB` и запись в ClickHouse | + +Зависимости: `SVBase`, `SVMisc`, `zlib` / `z`. + +## Lifecycle + +Сервер стартует **один раз за жизнь процесса**. Config и status-callback задаются только при старте; hot-reload настроек архива/ClickHouse не поддерживается. + +``` +Idle ──startServer()──► Running ──stopServer()──► Stopped + │ │ │ + │ receiveData OK receiveData no-op + │ │ │ + └── startServer() ──► false (повторный старт) ◄────┘ +``` + +| Метод | Поведение | +|-------|-----------| +| `startServer(config, statusCb)` | Первый вызов — `true`; все последующие (в т.ч. после `stopServer`) — `false` | +| `stopServer()` | Идемпотентен; `receiveData` сразу становится no-op | +| `receiveData(inout, out)` | Принимает TCP/COM-пакеты только пока сервер running | + +`startServer` / `stopServer` сериализуются внутри (`m_mtxLifecycle`); безопасно вызывать из разных потоков. + +При неуспешном старте после CAS (`m_wasStarted`) повторный `startServer` также вернёт `false` — rollback не выполняется. + +### Пример + +```cpp +SV_Srv::Config cfg; +cfg.cycleRecMs = 100; +cfg.packetSz = 10; +cfg.outArchiveEna = true; +cfg.outArchivePath = "/data/archive"; + +if (!SV_Srv::startServer(cfg, [](const std::string& msg) { + std::cout << msg << '\n'; + })) { + // уже стартовал ранее +} + +// передать пакет из TCP/COM callback: +SV_Srv::receiveData(buffer, dummyOut); + +SV_Srv::stopServer(); +``` + +В **SVMonitor** при нескольких COM/TCP-портах: один `startServer` до цикла чтения, `receiveData` — из каждого reader'а. + +## Публичный API + +| Функция | Назначение | +|---------|------------| +| `setOnUpdateSignalsCBack` | Сигналы обновлены | +| `setOnAddSignalsCBack` | Добавлены новые сигналы | +| `setOnModuleConnectCBack` / `setOnModuleDisconnectCBack` | Модуль подключён / отключён | +| `getCopyModuleRef` / `getCopySignalRef` | Снимок реестра | +| `getSignalData(sign)` | Данные сигнала (`sign` = `name + module`) | +| `lockReadSData` / `unlockReadSData` | Shared-lock на данные сигналов | +| `lockWriteSData` / `unlockWriteSData` | Exclusive-lock | +| `signalBufferEna(sign)` | Включить буфер графика для сигнала | + +Удалены (v2): `setStatusCBack()`, `setConfig()` — callback статуса передаётся в `startServer`, конфиг архива/CH применяется только при следующем запуске приложения. + +## Внутренняя структура + +``` +sv_server.h — публичный API +src/server.cpp — lifecycle, receiveData, signalBufferEna, реестр модулей/сигналов +src/buffer_data.* — ring buffer входящих пакетов +src/thread_update.* — worker: разбор буфера, addSignal, callbacks, archive tick +src/archive.* — запись бинарного архива на диск (+ ClickHouse при USE_ClickHouseDB) +src/clickhouse_db.* — асинхронная запись в ClickHouse (только USE_ClickHouseDB) +``` + +### Синхронизация + +| Mutex | Область | +|-------|---------| +| `m_mtxLifecycle` | `startServer` / `stopServer` | +| `m_mtxCommon` | Реестр `m_moduleData` / `m_signalData` | +| `m_mtxRW` (`shared_mutex`) | Данные сигналов, `isBuffEnable`, буферы графика | + +`signalBufferEna`: поиск сигнала под `m_mtxCommon`, чтение/запись `isBuffEnable` и публикация буфера — под `m_mtxRW`. Владение памятью буфера — `m_signalBuffStorage` в `server.cpp`. + +`ClickHouseDB`: `m_mtx` — `m_signals` / `m_signalBlock`; `m_workersMtx` — пул worker-потоков; `m_stopping` — запрет новых задач в деструкторе. `joinWorkers()` переносит потоки из-под lock и join'ит локально. + +## Потребители + +| Модуль | Использование | +|--------|---------------| +| SVMonitor | Основной сервер записи | +| SVDocker | Docker-обёртка | +| SVWriterExample | Пример writer | diff --git a/src/SVServer/src/archive.cpp b/src/SVServer/src/archive.cpp index 17a1dc0a..9db04d52 100644 --- a/src/SVServer/src/archive.cpp +++ b/src/SVServer/src/archive.cpp @@ -4,30 +4,14 @@ // // This code is licensed under the MIT License. // -// Permission is hereby granted, free of charge, to any person obtaining a copy -// of this software and associated documentation files(the "Software"), to deal -// in the Software without restriction, including without limitation the rights -// to use, copy, modify, merge, publish, distribute, sublicense, and / or sell -// copies of the Software, and to permit persons to whom the Software is -// furnished to do so, subject to the following conditions : -// -// The above copyright notice and this permission notice shall be included in -// all copies or substantial portions of the Software. -// -// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE -// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN -// THE SOFTWARE. -// #include "archive.h" #include "clickhouse_db.h" #include "SVMisc/misc.h" #include "SVBase/sv_limits.h" +#include "SVServer/sv_server.h" #include "zlib/zlib.h" +#include #include #include #include @@ -38,12 +22,36 @@ using namespace std; void statusMessage(const std::string& mess); +namespace { + void writeFixedField(char* dst, size_t fieldSz, const std::string& s) { + memset(dst, 0, fieldSz); + if (!s.empty()) { + memcpy(dst, s.c_str(), std::min(s.size(), fieldSz - 1)); + } + } +} + +Archive::~Archive() { + joinSaveThread(); +#ifdef USE_ClickHouseDB + delete m_chdb; + m_chdb = nullptr; +#endif +} + +void Archive::joinSaveThread() { + if (m_saveThread && m_saveThread->joinable()) { + m_saveThread->join(); + } +} + void Archive::init(const SV_Srv::Config& cng_) { cng = cng_; + cng.outArchiveHourCnt = std::max(1, cng.outArchiveHourCnt); m_copyStartTime = SV_Misc::currDateTimeEx(); m_copyDateMem = SV_Misc::currDateS(); - m_copySz = std::max(10, 600000 / SV_CYCLESAVE_MS); // 10мин + m_copySz = std::max(size_t(10), size_t(600000 / SV_CYCLESAVE_MS)); #ifdef USE_ClickHouseDB if(cng.outDataBaseEna && !cng.outDataBaseName.empty() && !cng.outDataBaseAddr.empty()){ @@ -52,19 +60,6 @@ void Archive::init(const SV_Srv::Config& cng_) { #endif } -void Archive::setConfig(const SV_Srv::Config& cng_){ - - cng.outArchivePath = cng_.outArchivePath; - cng.outArchiveName = cng_.outArchiveName; - cng.outArchiveHourCnt = cng_.outArchiveHourCnt; - -#ifdef USE_ClickHouseDB - if (!m_chdb && cng_.outDataBaseEna && !cng_.outDataBaseName.empty() && !cng_.outDataBaseAddr.empty()){ - m_chdb = new ClickHouseDB(cng); - } -#endif -} - void Archive::addSignal(const std::string& sname, const std::string& module, SV_Base::ValueType stype) { for (int aIndex = 0; aIndex < 2; ++aIndex){ @@ -77,11 +72,13 @@ void Archive::addSignal(const std::string& sname, const std::string& module, SV_ auto& vpos = m_valPos[aIndex]; vpos[sign] = 0; - SV_Base::Value* buff = new SV_Base::Value[SV_PACKETSZ * m_copySz]; - memset(buff, 0, SV_PACKETSZ * m_copySz * sizeof(SV_Base::Value)); + auto storage = std::make_unique(SV_PACKETSZ * m_copySz); + memset(storage.get(), 0, SV_PACKETSZ * m_copySz * sizeof(SV_Base::Value)); + SV_Base::Value* buff = storage.get(); for (size_t i = 0; i < m_copySz; ++i){ archiveData[sign][i].vals = &buff[i * SV_PACKETSZ]; } + m_archiveValueStorage[aIndex][sign] = std::move(storage); #ifdef USE_ClickHouseDB if (m_chdb){ m_chdb->addSignal(sname, module, stype); @@ -100,15 +97,13 @@ void Archive::addValue(const string& sign, const SV_Base::RecData& rd) { ++valPos[sign]; - if (valPos[sign] == m_copySz) { + if (valPos[sign] == static_cast(m_copySz)) { copyToDisk(false); } } void Archive::copyToDisk(bool isStop){ - if (m_saveThread && m_saveThread->joinable()){ - m_saveThread->join(); - } + joinSaveThread(); if (!isStop){ m_saveThread = std::make_shared([this, aIndex = m_archiveIndex]{ copyToDiskImpl(false, aIndex); @@ -129,7 +124,7 @@ void Archive::copyToDiskImpl(bool isStop, int archiveIndex){ } if (cng.outArchiveEna){ - size_t SMAXCNT = 100; // макс кол-во сигналов в посылке + size_t SMAXCNT = 100; if (SV_PACKETSZ > 100000){ SMAXCNT = 1; }else if (SV_PACKETSZ > 10000){ @@ -140,7 +135,6 @@ void Archive::copyToDiskImpl(bool isStop, int archiveIndex){ tmSz = sizeof(uint64_t), vlSz = sizeof(SV_Base::Value) * SV_PACKETSZ; - // name module group comment type vCnt const size_t headSz = SV_NAMESZ + SV_NAMESZ + SV_NAMESZ + SV_COMMENTSZ + intSz + intSz; vector inArr((tmSz + vlSz) * m_copySz * SMAXCNT + headSz * SMAXCNT); @@ -161,17 +155,21 @@ void Archive::copyToDiskImpl(bool isStop, int archiveIndex){ for (const auto& ad : archiveData) { const auto sign = SV_Srv::getSignalData(ad.first); + if (!sign) { + ++ix; + continue; + } char* pIn = inArr.data(); int vCnt = valPos[ad.first]; if (vCnt > 0) { - memcpy(pIn + csize, sign->name.c_str(), SV_NAMESZ); csize += SV_NAMESZ; - memcpy(pIn + csize, sign->module.c_str(), SV_NAMESZ); csize += SV_NAMESZ; - memcpy(pIn + csize, sign->group.c_str(), SV_NAMESZ); csize += SV_NAMESZ; - memcpy(pIn + csize, sign->comment.c_str(), SV_COMMENTSZ); csize += SV_COMMENTSZ; - memcpy(pIn + csize, &sign->type, intSz); csize += intSz; - memcpy(pIn + csize, &vCnt, intSz); csize += intSz; + writeFixedField(pIn + csize, SV_NAMESZ, sign->name); csize += SV_NAMESZ; + writeFixedField(pIn + csize, SV_NAMESZ, sign->module); csize += SV_NAMESZ; + writeFixedField(pIn + csize, SV_NAMESZ, sign->group); csize += SV_NAMESZ; + writeFixedField(pIn + csize, SV_COMMENTSZ, sign->comment); csize += SV_COMMENTSZ; + memcpy(pIn + csize, &sign->type, intSz); csize += intSz; + memcpy(pIn + csize, &vCnt, intSz); csize += intSz; for (int j = 0; j < vCnt; ++j) { memcpy(pIn + csize, &ad.second[j].beginTime, tmSz); csize += tmSz; @@ -207,8 +205,6 @@ void Archive::copyToDiskImpl(bool isStop, int archiveIndex){ for(auto& v : valPos){ v.second = 0; } - - return; } bool Archive::compressData(size_t inSz, const vector& inArr, size_t& outsz, vector& outArr) { @@ -224,7 +220,7 @@ bool Archive::compressData(size_t inSz, const vector& inArr, size_t& outsz return res == Z_OK; } - catch (exception e) { + catch (const exception& e) { statusMessage("Archive::compressData exception " + string(e.what())); return false; } @@ -261,12 +257,19 @@ string Archive::getOutPath(bool isStop) { bool Archive::isCopyTimeHour() { time_t t = time(nullptr); - tm* lct = localtime(&t); + tm lctBuf{}; +#ifdef _WIN32 + localtime_s(&lctBuf, &t); + tm* lct = &lctBuf; +#else + tm* lct = localtime_r(&t, &lctBuf); + if (!lct) return false; +#endif if (m_front.PosFront(lct->tm_min == 0, 0)) ++m_crtFileHour; bool req = false; - bool isCheck = cng.outArchiveHourCnt % 2 == 0; // кратно 2м часам? + bool isCheck = cng.outArchiveHourCnt % 2 == 0; bool isHourCheck = lct->tm_hour % 2 == 0; bool isNDay = m_front.PosFront(lct->tm_hour == 0, 1); if (((int(m_crtFileHour) >= cng.outArchiveHourCnt) || isNDay) && (!isCheck || isHourCheck)) { @@ -276,4 +279,3 @@ bool Archive::isCopyTimeHour() { return req; } - diff --git a/src/SVServer/src/archive.h b/src/SVServer/src/archive.h index f4f3dfa0..562d8616 100644 --- a/src/SVServer/src/archive.h +++ b/src/SVServer/src/archive.h @@ -4,40 +4,24 @@ // // This code is licensed under the MIT License. // -// Permission is hereby granted, free of charge, to any person obtaining a copy -// of this software and associated documentation files(the "Software"), to deal -// in the Software without restriction, including without limitation the rights -// to use, copy, modify, merge, publish, distribute, sublicense, and / or sell -// copies of the Software, and to permit persons to whom the Software is -// furnished to do so, subject to the following conditions : -// -// The above copyright notice and this permission notice shall be included in -// all copies or substantial portions of the Software. -// -// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE -// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN -// THE SOFTWARE. -// #pragma once #include "SVMisc/front.h" #include "SVServer/sv_server.h" #include +#include #include +#include class ClickHouseDB; -// копирование архива на диск class Archive { public: Archive() = default; + ~Archive(); void init(const SV_Srv::Config&); @@ -46,9 +30,8 @@ class Archive void addSignal(const std::string& sname, const std::string& module, SV_Base::ValueType stype); void addValue(const std::string& sign, const SV_Base::RecData& rd); - void setConfig(const SV_Srv::Config&); - private: + void joinSaveThread(); void copyToDiskImpl(bool isStop, int archiveIndex); bool isCopyTimeHour(); std::string getOutPath(bool isStop); @@ -69,6 +52,7 @@ class Archive SV_Misc::Front m_front; std::map> m_archiveData[2]; + std::map> m_archiveValueStorage[2]; ClickHouseDB* m_chdb{}; std::shared_ptr m_saveThread; diff --git a/src/SVServer/src/buffer_data.cpp b/src/SVServer/src/buffer_data.cpp index 6b893a1d..574ac50b 100644 --- a/src/SVServer/src/buffer_data.cpp +++ b/src/SVServer/src/buffer_data.cpp @@ -4,24 +4,6 @@ // // This code is licensed under the MIT License. // -// Permission is hereby granted, free of charge, to any person obtaining a copy -// of this software and associated documentation files(the "Software"), to deal -// in the Software without restriction, including without limitation the rights -// to use, copy, modify, merge, publish, distribute, sublicense, and / or sell -// copies of the Software, and to permit persons to whom the Software is -// furnished to do so, subject to the following conditions : -// -// The above copyright notice and this permission notice shall be included in -// all copies or substantial portions of the Software. -// -// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE -// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN -// THE SOFTWARE. -// #include "buffer_data.h" @@ -30,28 +12,52 @@ using namespace std; +namespace { + size_t calcBuffSz(int packetSz) { + size_t buffSz = SV_VALUE_MAX_CNT * 10; + if (packetSz > 100000) { + buffSz /= 100; + } else if (packetSz > 10000) { + buffSz /= 10; + } + return buffSz; + } + + size_t boundedCStrLen(const char* s, size_t maxLen) { + const void* end = std::memchr(s, '\0', maxLen); + return end ? static_cast(end) - s : maxLen; + } +} + void BufferData::init(const SV_Srv::Config& cng_) { cng = cng_; - if (SV_PACKETSZ > 100000){ - m_buffSz /= 100; - }else if (SV_PACKETSZ > 10000){ - m_buffSz /= 10; - } + m_buffSz = calcBuffSz(cng.packetSz); + m_buffReadPos = 0; + m_buffWritePos = 0; + m_buffWritePosForReader = 0; + m_timeOffsetMs.clear(); + + m_valueStorage = std::make_unique(SV_PACKETSZ * m_buffSz); + memset(m_valueStorage.get(), 0, SV_PACKETSZ * m_buffSz * sizeof(SV_Base::Value)); + m_buffer.resize(m_buffSz); - SV_Base::Value* buff = new SV_Base::Value[SV_PACKETSZ * m_buffSz]; - memset(buff, 0, SV_PACKETSZ * m_buffSz * sizeof(SV_Base::Value)); for (size_t i = 0; i < m_buffSz; ++i){ - m_buffer[i].data.vals = &buff[i * SV_PACKETSZ]; + m_buffer[i].data.vals = &m_valueStorage[i * SV_PACKETSZ]; } } void BufferData::updateDataSignals(std::string_view indata, uint64_t bTm){ - size_t dsz = indata.size(), - valSz = SV_NAMESZ + sizeof(SV_Base::ValueType) + sizeof(SV_Base::Value) * SV_PACKETSZ, - cPos = SV_NAMESZ; - size_t valCnt = std::max(size_t(0), std::min((dsz - cPos) / valSz, m_buffSz / 10)); // 10 сек - запас + const size_t dsz = indata.size(); + if (dsz < SV_NAMESZ) return; + + const size_t valSz = SV_NAMESZ + sizeof(SV_Base::ValueType) + sizeof(SV_Base::Value) * SV_PACKETSZ; + const size_t maxBySize = (dsz - SV_NAMESZ) / valSz; + const size_t valCnt = std::min(maxBySize, m_buffSz / 10); + + if (valCnt == 0) return; + size_t wPos, wPosMem; { std::lock_guard lck(m_mtxWrite); @@ -61,7 +67,8 @@ void BufferData::updateDataSignals(std::string_view indata, uint64_t bTm){ m_buffWritePos -= m_buffSz; } if (cng.offsetMs > 0){ - const std::string module = indata.data(); + const size_t modLen = boundedCStrLen(indata.data(), SV_NAMESZ); + const std::string module(indata.data(), modLen); if (m_timeOffsetMs.count(module)){ m_timeOffsetMs[module] += cng.offsetMs; }else{ @@ -70,11 +77,17 @@ void BufferData::updateDataSignals(std::string_view indata, uint64_t bTm){ bTm += m_timeOffsetMs[module]; } } - size_t vlsz = sizeof(SV_Base::Value) * SV_PACKETSZ, - cvalCnt = 0; - while (cPos < dsz && cvalCnt < valCnt){ - m_buffer[wPos].module = indata.data(); - m_buffer[wPos].name = indata.data() + cPos; + + const size_t vlsz = sizeof(SV_Base::Value) * SV_PACKETSZ; + size_t cPos = SV_NAMESZ; + size_t cvalCnt = 0; + + const size_t modLen = boundedCStrLen(indata.data(), SV_NAMESZ); + const std::string moduleStr(indata.data(), modLen); + + while (cPos + valSz <= dsz && cvalCnt < valCnt){ + m_buffer[wPos].module = moduleStr; + m_buffer[wPos].name = std::string(indata.data() + cPos, SV_NAMESZ); memcpy(&m_buffer[wPos].type, indata.data() + cPos + SV_NAMESZ, sizeof(int32_t)); memcpy(m_buffer[wPos].data.vals, indata.data() + cPos + SV_NAMESZ + sizeof(SV_Base::ValueType), vlsz); m_buffer[wPos].data.beginTime = bTm; @@ -85,10 +98,15 @@ void BufferData::updateDataSignals(std::string_view indata, uint64_t bTm){ cPos += valSz; ++cvalCnt; } + + if (cvalCnt == 0) { + return; + } + while (true){ std::lock_guard lck(m_mtxRead); - if (wPosMem == m_buffWritePosForReader){ // для защиты от гонки, если из др потока сюда дойдут раньше - m_buffWritePosForReader += valCnt; + if (wPosMem == m_buffWritePosForReader){ + m_buffWritePosForReader += cvalCnt; if (m_buffWritePosForReader >= m_buffSz){ m_buffWritePosForReader -= m_buffSz; } diff --git a/src/SVServer/src/buffer_data.h b/src/SVServer/src/buffer_data.h index 417a12d0..389cd9be 100644 --- a/src/SVServer/src/buffer_data.h +++ b/src/SVServer/src/buffer_data.h @@ -4,29 +4,13 @@ // // This code is licensed under the MIT License. // -// Permission is hereby granted, free of charge, to any person obtaining a copy -// of this software and associated documentation files(the "Software"), to deal -// in the Software without restriction, including without limitation the rights -// to use, copy, modify, merge, publish, distribute, sublicense, and / or sell -// copies of the Software, and to permit persons to whom the Software is -// furnished to do so, subject to the following conditions : -// -// The above copyright notice and this permission notice shall be included in -// all copies or substantial portions of the Software. -// -// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE -// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN -// THE SOFTWARE. -// #pragma once #include "SVServer/sv_server.h" #include "SVBase/sv_limits.h" +#include +#include #include class BufferData @@ -52,7 +36,7 @@ class BufferData SV_Srv::Config cng; - size_t m_buffSz = SV_VALUE_MAX_CNT * 10; // 10 сек - запас + size_t m_buffSz = 0; std::vector m_buffer; size_t m_buffReadPos = 0; @@ -61,5 +45,7 @@ class BufferData std::map m_timeOffsetMs; + std::unique_ptr m_valueStorage; + std::mutex m_mtxRead, m_mtxWrite; }; diff --git a/src/SVServer/src/clickhouse_db.cpp b/src/SVServer/src/clickhouse_db.cpp index 28aa4d63..815a61a3 100644 --- a/src/SVServer/src/clickhouse_db.cpp +++ b/src/SVServer/src/clickhouse_db.cpp @@ -6,7 +6,9 @@ #include "SVBase/sv_limits.h" #include -#include + +#include +#include namespace ch = clickhouse; @@ -15,11 +17,11 @@ void statusMessage(const std::string&); ClickHouseDB::ClickHouseDB(const SV_Srv::Config& _cng): cng(_cng) { - std::thread([this]() { + emplaceWorker([this]() { try { - std::lock_guard lk(m_mtx); if (auto clt = newClient(); clt) { clt->Select("SELECT id, sname, module FROM tblSignal;", [this](const ch::Block& block) { + std::lock_guard lk(m_mtx); for (size_t i = 0; i < block.GetRowCount(); ++i) { int sId = block[0]->As()->At(i); std::string sname = std::string(block[1]->As()->At(i)); @@ -30,10 +32,45 @@ ClickHouseDB::ClickHouseDB(const SV_Srv::Config& _cng): }); } } - catch (std::exception& e) { + catch (const std::exception& e) { statusMessage("ClickHouseDB read signals error: " + std::string(e.what())); } - }).detach(); + }); +} + +ClickHouseDB::~ClickHouseDB() { + m_stopping.store(true, std::memory_order_release); + joinWorkers(); +} + +void ClickHouseDB::reapWorkersLocked() { + m_workers.erase( + std::remove_if(m_workers.begin(), m_workers.end(), + [](std::future& f) { + return f.wait_for(std::chrono::seconds(0)) == std::future_status::ready; + }), + m_workers.end()); +} + +void ClickHouseDB::emplaceWorker(std::function fn) { + if (m_stopping.load(std::memory_order_acquire)) return; + std::lock_guard lk(m_workersMtx); + if (m_stopping.load(std::memory_order_acquire)) return; + reapWorkersLocked(); + m_workers.push_back(std::async(std::launch::async, std::move(fn))); +} + +void ClickHouseDB::joinWorkers() { + std::vector> workers; + { + std::lock_guard lk(m_workersMtx); + workers = std::move(m_workers); + } + for (auto& f : workers) { + if (f.valid()) { + f.wait(); + } + } } bool ClickHouseDB::isConnect()const @@ -43,7 +80,7 @@ bool ClickHouseDB::isConnect()const clt->Ping(); return true; } - }catch(std::exception& e){ + }catch(const std::exception& e){ statusMessage("ClickHouseDB connect error: " + std::string(e.what())); } return false; @@ -51,41 +88,43 @@ bool ClickHouseDB::isConnect()const void ClickHouseDB::addSignal(const std::string& sname, const std::string& module, SV_Base::ValueType stype) { - if (!m_signals.count(sname + module)){ - m_signals.insert({sname + module, m_signals.size()}); + const std::string key = sname + module; + bool needWorker = false; + { std::lock_guard lk(m_mtx); + if (m_signals.count(key)) return; + m_signals.insert({key, static_cast(m_signals.size())}); - bool isNew = false; if (!m_signalBlock){ m_signalBlock = newSignalBlock(); - isNew = true; + needWorker = true; } auto cId = column(m_signalBlock, "id")->AsStrict(); auto cSName = column(m_signalBlock, "sname")->AsStrict(); auto cModule = column(m_signalBlock, "module")->AsStrict(); auto cType = column(m_signalBlock, "stype")->AsStrict(); - cId->Append(m_signals[sname + module]); + cId->Append(m_signals[key]); cSName->Append(sname); cModule->Append(module); cType->Append(int(stype)); + } - if (isNew){ - std::thread([this](){ - SV_Misc::sleepMs(1000); - try{ - std::lock_guard lk(m_mtx); - if (auto clt = newClient(); clt){ - m_signalBlock->RefreshRowCount(); - clt->Insert("tblSignal", *m_signalBlock); - } - m_signalBlock.reset(); - }catch(std::exception& e){ - statusMessage("ClickHouseDB::addSignal save error: " + std::string(e.what())); + if (needWorker){ + emplaceWorker([this](){ + SV_Misc::sleepMs(1000); + try{ + std::lock_guard lk(m_mtx); + if (auto clt = newClient(); clt && m_signalBlock){ + m_signalBlock->RefreshRowCount(); + clt->Insert("tblSignal", *m_signalBlock); } - }).detach(); - } + m_signalBlock.reset(); + }catch(const std::exception& e){ + statusMessage("ClickHouseDB::addSignal save error: " + std::string(e.what())); + } + }); } } @@ -99,12 +138,21 @@ void ClickHouseDB::saveSData(bool onClose, const std::map& val for (const auto& sd : sdata){ - if (!m_signals.count(sd.first) || valPos.at(sd.first) == 0) continue; + int sid = 0; + int vcnt = 0; + { + std::lock_guard lk(m_mtx); + auto it = m_signals.find(sd.first); + if (it == m_signals.end()) continue; + auto vpIt = valPos.find(sd.first); + if (vpIt == valPos.end() || vpIt->second == 0) continue; + sid = it->second; + vcnt = vpIt->second; + } auto sign = SV_Srv::getSignalData(sd.first); - - int sid = m_signals[sd.first]; - int vcnt = valPos.at(sd.first); + if (!sign) continue; + if (sign->type == SV_Base::ValueType::FLOAT){ for(int i = 0; i < vcnt; ++i){ const auto& rd = sd.second[i]; @@ -132,14 +180,14 @@ void ClickHouseDB::saveSData(bool onClose, const std::map& val dblock->RefreshRowCount(); clt->Insert("tblSData", *dblock); } - }catch(std::exception& e){ + }catch(const std::exception& e){ statusMessage("ClickHouseDB::addSData save error: " + std::string(e.what())); } }; if (!onClose){ - std::thread([this, sendToDb, dblock = std::move(dataBlock)](){ + emplaceWorker([this, sendToDb, dblock = std::move(dataBlock)](){ sendToDb(dblock); - }).detach(); + }); }else{ sendToDb(dataBlock); } @@ -160,10 +208,11 @@ std::unique_ptr ClickHouseDB::newClient()const { ch::ClientOptions opts;{ opts.SetDefaultDatabase(cng.outDataBaseName); - auto sp = cng.outDataBaseAddr.find(':'); - opts.SetHost(sp > 0 ? cng.outDataBaseAddr.substr(0, sp) : cng.outDataBaseAddr); - if (sp > 0){ - opts.SetPort(stoi(cng.outDataBaseAddr.substr(sp + 1))); + const auto sp = cng.outDataBaseAddr.find(':'); + opts.SetHost(sp == std::string::npos ? cng.outDataBaseAddr + : cng.outDataBaseAddr.substr(0, sp)); + if (sp != std::string::npos && sp + 1 < cng.outDataBaseAddr.size()){ + opts.SetPort(std::stoi(cng.outDataBaseAddr.substr(sp + 1))); } } return std::make_unique(opts); diff --git a/src/SVServer/src/clickhouse_db.h b/src/SVServer/src/clickhouse_db.h index f03b8bfb..71e944d4 100644 --- a/src/SVServer/src/clickhouse_db.h +++ b/src/SVServer/src/clickhouse_db.h @@ -4,28 +4,13 @@ // // This code is licensed under the MIT License. // -// Permission is hereby granted, free of charge, to any person obtaining a copy -// of this software and associated documentation files(the "Software"), to deal -// in the Software without restriction, including without limitation the rights -// to use, copy, modify, merge, publish, distribute, sublicense, and / or sell -// copies of the Software, and to permit persons to whom the Software is -// furnished to do so, subject to the following conditions : -// -// The above copyright notice and this permission notice shall be included in -// all copies or substantial portions of the Software. -// -// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE -// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN -// THE SOFTWARE. -// #pragma once #include "SVServer/sv_server.h" +#include +#include +#include #include #include #include @@ -44,6 +29,7 @@ class ClickHouseDB{ public: ClickHouseDB(const SV_Srv::Config&); + ~ClickHouseDB(); bool isConnect()const; @@ -52,6 +38,9 @@ class ClickHouseDB{ void saveSData(bool onClose, const std::map& valPos, const std::map>&); private: + void joinWorkers(); + void emplaceWorker(std::function fn); + void reapWorkersLocked(); std::unique_ptr newClient()const; std::unique_ptr newSignalBlock()const; std::unique_ptr newSDataBlock()const; @@ -61,8 +50,11 @@ class ClickHouseDB{ private: SV_Srv::Config cng; - std::map m_signals; // key - sname, value - id + std::map m_signals; std::unique_ptr m_signalBlock; std::mutex m_mtx; + std::mutex m_workersMtx; + std::atomic_bool m_stopping{false}; + std::vector> m_workers; }; diff --git a/src/SVServer/src/server.cpp b/src/SVServer/src/server.cpp index c9052481..f29ee749 100644 --- a/src/SVServer/src/server.cpp +++ b/src/SVServer/src/server.cpp @@ -4,31 +4,15 @@ // // This code is licensed under the MIT License. // -// Permission is hereby granted, free of charge, to any person obtaining a copy -// of this software and associated documentation files(the "Software"), to deal -// in the Software without restriction, including without limitation the rights -// to use, copy, modify, merge, publish, distribute, sublicense, and / or sell -// copies of the Software, and to permit persons to whom the Software is -// furnished to do so, subject to the following conditions : -// -// The above copyright notice and this permission notice shall be included in -// all copies or substantial portions of the Software. -// -// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE -// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN -// THE SOFTWARE. -// #include "sv_server.h" #include "buffer_data.h" #include "thread_update.h" #include "SVMisc/misc.h" +#include #include +#include #include using namespace std; @@ -52,55 +36,68 @@ namespace SV_Srv { std::map m_moduleData; std::map m_signalData; + std::map> m_signalBuffStorage; + std::atomic_bool m_wasStarted{false}; + std::atomic_bool m_isRunning{false}; + + std::mutex m_mtxLifecycle; std::mutex m_mtxCommon; std::shared_mutex m_mtxRW; - - void setStatusCBack(statusCBack cback){ - pfStatusCBack = cback; - } - - bool startServer(const Config& _cng){ - std::lock_guard lck(m_mtxCommon); + bool startServer(const Config& _cng, statusCBack stsCBack){ + std::lock_guard lifecycle(m_mtxLifecycle); - if (m_pThrUpdSignal) return true; + bool expected = false; + if (!m_wasStarted.compare_exchange_strong(expected, true)) + return false; + pfStatusCBack = std::move(stsCBack); cng = _cng; m_buffData.init(cng); - m_pThrUpdSignal = new ThreadUpdate(cng, m_buffData); + auto thr = std::make_unique(cng, m_buffData); + { + std::lock_guard lck(m_mtxCommon); + m_pThrUpdSignal = thr.release(); + } + m_isRunning.store(true, std::memory_order_release); return true; } void stopServer(){ - if (m_pThrUpdSignal){ - delete m_pThrUpdSignal; + std::lock_guard lifecycle(m_mtxLifecycle); + + m_isRunning.store(false, std::memory_order_release); + ThreadUpdate* thr = nullptr; + { + std::lock_guard lck(m_mtxCommon); + std::swap(thr, m_pThrUpdSignal); } - } - - void setConfig(const Config& cng){ - if (m_pThrUpdSignal){ - m_pThrUpdSignal->setArchiveConfig(cng); - } + delete thr; } void receiveData(std::string& inout, std::string& out){ + (void)out; + if (!m_isRunning.load(std::memory_order_acquire)) return; + vector> bePos; const std::string_view beginMess = "=begin="; const std::string_view endMess = "=end="; const size_t mlen = 4, beginLen = beginMess.size(), endLen = endMess.size(); size_t stPos = inout.find(beginMess), endPos = 0; - while (stPos != std::string::npos && stPos + beginLen + mlen < inout.size()){ + while (stPos != std::string::npos && stPos + beginLen + mlen <= inout.size()){ int32_t allSz; std::memcpy(&allSz, inout.c_str() + stPos + beginLen, sizeof(int32_t)); - if (allSz > 0){ - endPos = stPos + beginLen + mlen + allSz; + const size_t payloadStart = stPos + beginLen + mlen; + const size_t maxPayload = inout.size() - payloadStart; + if (allSz > 0 && static_cast(allSz) <= maxPayload){ + endPos = payloadStart + allSz; if (endPos + endLen <= inout.size() && endMess == std::string_view(inout.data() + endPos, endLen)){ - bePos.push_back({stPos + beginLen + mlen, endPos}); + bePos.push_back({payloadStart, endPos}); stPos = inout.find(beginMess, endPos + endLen); continue; } @@ -209,21 +206,34 @@ namespace SV_Srv { } bool signalBufferEna(const std::string& sign){ - std::lock_guard lck(m_mtxCommon); + SV_Base::SignalData* sd = nullptr; + { + std::lock_guard lck(m_mtxCommon); + auto it = m_signalData.find(sign); + if (it == m_signalData.end()) return false; + sd = it->second; + } - if (m_signalData.find(sign) == m_signalData.end()) return false; + { + std::lock_guard lck(m_mtxRW); + if (sd->isBuffEnable) return true; + } - if (!m_signalData[sign]->isBuffEnable){ - int buffSz = std::max(10, (2 * 3600000) / SV_CYCLESAVE_MS); // размер буфера 2 часа - m_signalData[sign]->buffData.resize(buffSz); + const int buffSz = std::max(10, static_cast((2 * 3600000) / SV_CYCLESAVE_MS)); + std::vector preparedBuffData(buffSz); + auto preparedValueBlock = std::make_unique(SV_PACKETSZ * buffSz); + for (int i = 0; i < buffSz; ++i){ + preparedBuffData[i].vals = &preparedValueBlock[i * SV_PACKETSZ]; + } - SV_Base::Value* buff = new SV_Base::Value[SV_PACKETSZ * buffSz]; - for (int i = 0; i < buffSz; ++i){ - m_signalData[sign]->buffData[i].vals = &buff[i * SV_PACKETSZ]; - } - {std::lock_guard lckw(m_mtxRW); - m_signalData[sign]->isBuffEnable = true; - } + { + std::lock_guard lckw(m_mtxRW); + if (sd->isBuffEnable) return true; + sd->buffData = std::move(preparedBuffData); + m_signalBuffStorage[sign] = std::move(preparedValueBlock); + sd->buffBeginPos = 0; + sd->buffValuePos = 0; + sd->isBuffEnable = true; } return true; } diff --git a/src/SVServer/src/thread_update.cpp b/src/SVServer/src/thread_update.cpp index 4109d22f..0c1f3c79 100644 --- a/src/SVServer/src/thread_update.cpp +++ b/src/SVServer/src/thread_update.cpp @@ -4,24 +4,6 @@ // // This code is licensed under the MIT License. // -// Permission is hereby granted, free of charge, to any person obtaining a copy -// of this software and associated documentation files(the "Software"), to deal -// in the Software without restriction, including without limitation the rights -// to use, copy, modify, merge, publish, distribute, sublicense, and / or sell -// copies of the Software, and to permit persons to whom the Software is -// furnished to do so, subject to the following conditions : -// -// The above copyright notice and this permission notice shall be included in -// all copies or substantial portions of the Software. -// -// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE -// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN -// THE SOFTWARE. -// #include "SVServer/sv_server.h" #include "SVMisc/timer_delay.h" @@ -60,14 +42,6 @@ ThreadUpdate::~ThreadUpdate(){ if (m_thr.joinable()) m_thr.join(); } -void ThreadUpdate::setArchiveConfig(const SV_Srv::Config& cng_){ - - cng.outArchiveEna = cng_.outArchiveEna; - cng.outDataBaseEna = cng_.outDataBaseEna; - - m_archive.setConfig(cng_); -} - void ThreadUpdate::addSignal(const BufferData::InputData& bp){ SignalData* sd = new SignalData(); @@ -106,16 +80,14 @@ void ThreadUpdate::addSignal(const BufferData::InputData& bp){ void ThreadUpdate::updateSignals(std::map& sref, std::map& mref, std::map& signActive, std::map& moduleActive){ - std::lock_guard lock(SV_Srv::m_mtxRW); - bool isNewSign = false, isBuffActive = false; - size_t buffSz = 2 * 3600000 / SV_CYCLESAVE_MS; // 2 часа + size_t buffSz = 2 * 3600000 / SV_CYCLESAVE_MS; if (buffSz == 0){ buffSz = 1; } const size_t packSz = SV_PACKETSZ * sizeof(Value); - + std::vector bufData; while (m_buffData.getDataByReadPos(bufData)){ for(const auto& bufPos : bufData){ @@ -132,25 +104,29 @@ void ThreadUpdate::updateSignals(std::map& sr auto sdata = sref[sign]; - signActive[sign] = true; - moduleActive[bufPos.module] = true; + { + std::lock_guard lock(SV_Srv::m_mtxRW); - sdata->lastData.beginTime = bufPos.data.beginTime; - memcpy(sdata->lastData.vals, bufPos.data.vals, packSz); - if (sdata->isBuffEnable) { - size_t vp = sdata->buffValuePos; - sdata->buffData[vp].beginTime = bufPos.data.beginTime; - memcpy(sdata->buffData[vp].vals, bufPos.data.vals, packSz); + signActive[sign] = true; + moduleActive[bufPos.module] = true; - updateSignalsBuff(sdata, sdata->buffBeginPos, vp); + sdata->lastData.beginTime = bufPos.data.beginTime; + memcpy(sdata->lastData.vals, bufPos.data.vals, packSz); + if (sdata->isBuffEnable) { + size_t vp = sdata->buffValuePos; + sdata->buffData[vp].beginTime = bufPos.data.beginTime; + memcpy(sdata->buffData[vp].vals, bufPos.data.vals, packSz); - ++vp; - if (vp == buffSz) vp = 0; - sdata->buffValuePos = vp; + updateSignalsBuff(sdata, sdata->buffBeginPos, vp); - if (vp == sdata->buffBeginPos) { - ++sdata->buffBeginPos; - if (sdata->buffBeginPos >= buffSz) sdata->buffBeginPos = 0; + ++vp; + if (vp == buffSz) vp = 0; + sdata->buffValuePos = vp; + + if (vp == sdata->buffBeginPos) { + ++sdata->buffBeginPos; + if (sdata->buffBeginPos >= buffSz) sdata->buffBeginPos = 0; + } } } if (cng.outArchiveEna || cng.outDataBaseEna){ @@ -231,32 +207,42 @@ void ThreadUpdate::updateCycle(){ updateSignals(sref, mref, signActive, moduleActive); - // архив if (tmDelay.hourOnc() && (cng.outArchiveEna || cng.outDataBaseEna)){ m_archive.copyToDisk(false); } - // проверка связи if (tmDelay.onDelayMS(true, checkConnectTout, 0)){ tmDelay.onDelayMS(false, 0, 0); - std::lock_guard lock(SV_Srv::m_mtxRW); + std::vector connectModules; + std::vector disconnectModules; + + { + std::lock_guard lock(SV_Srv::m_mtxRW); - for (auto& s : signActive){ - sref[s.first]->isActive = s.second; - s.second = false; + for (auto& s : signActive){ + sref[s.first]->isActive = s.second; + s.second = false; + } + for (auto& m : moduleActive){ + if (m.first == "Virtual") continue; + + if (!mref[m.first]->isActive && m.second){ + connectModules.push_back(m.first); + } + else if (mref[m.first]->isActive && !m.second){ + disconnectModules.push_back(m.first); + } + mref[m.first]->isActive = m.second; + m.second = false; + } } - for (auto& m : moduleActive){ - if (m.first == "Virtual") continue; - if (!mref[m.first]->isActive && m.second){ - moduleConnect(m.first); - } - else if (mref[m.first]->isActive && !m.second){ - moduleDisconnect(m.first); - } - mref[m.first]->isActive = m.second; - m.second = false; + for (const auto& mod : connectModules) { + moduleConnect(mod); + } + for (const auto& mod : disconnectModules) { + moduleDisconnect(mod); } } @@ -270,4 +256,3 @@ void ThreadUpdate::updateCycle(){ m_archive.copyToDisk(true); } } - diff --git a/src/SVServer/src/thread_update.h b/src/SVServer/src/thread_update.h index ac8dd36f..f86a5209 100644 --- a/src/SVServer/src/thread_update.h +++ b/src/SVServer/src/thread_update.h @@ -4,24 +4,6 @@ // // This code is licensed under the MIT License. // -// Permission is hereby granted, free of charge, to any person obtaining a copy -// of this software and associated documentation files(the "Software"), to deal -// in the Software without restriction, including without limitation the rights -// to use, copy, modify, merge, publish, distribute, sublicense, and / or sell -// copies of the Software, and to permit persons to whom the Software is -// furnished to do so, subject to the following conditions : -// -// The above copyright notice and this permission notice shall be included in -// all copies or substantial portions of the Software. -// -// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE -// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN -// THE SOFTWARE. -// #pragma once #include "buffer_data.h" @@ -38,8 +20,6 @@ class ThreadUpdate{ ~ThreadUpdate(); - void setArchiveConfig(const SV_Srv::Config&); - private: SV_Srv::Config cng; diff --git a/src/SVServer/sv_server.h b/src/SVServer/sv_server.h index d012c198..ba54d734 100644 --- a/src/SVServer/sv_server.h +++ b/src/SVServer/sv_server.h @@ -68,20 +68,16 @@ namespace SV_Srv { offsetMs(0) {} }; - /// задать статус callback - /// \param pfStatusMess using statusCBack = std::function; - SV_API void setStatusCBack(statusCBack stsCBack); - /// старт сервера - SV_API bool startServer(const Config&); + /// старт сервера (один раз за жизнь процесса; повторный вызов — false) + /// \param stsCBack опциональный callback статусных сообщений + /// startServer/stopServer сериализуются внутри; безопасно вызывать из разных потоков. + SV_API bool startServer(const Config&, statusCBack stsCBack = nullptr); - /// стоп сервера + /// стоп сервера (идемпотентен). Сериализуется с startServer. SV_API void stopServer(); - /// задать конфиг - SV_API void setConfig(const Config&); - // получение данных сервером SV_API void receiveData(std::string& inout, std::string& out); diff --git a/src/SVWriterExample/src/main.cpp b/src/SVWriterExample/src/main.cpp index b302f90b..6a0c9a88 100644 --- a/src/SVWriterExample/src/main.cpp +++ b/src/SVWriterExample/src/main.cpp @@ -36,21 +36,23 @@ void statusMess(const string& mess){ int main(int argc, char* argv[]){ - SV_Misc::TCPServer::setErrorCBack(statusMess); - SV_Srv::setStatusCBack(statusMess); - - SV_Misc::TCPServer::setDataCBack(SV_Srv::receiveData); - SV_Srv::Config scng; scng.outArchiveEna = true; string addr = argc > 1 ? argv[1] : "127.0.0.1"; int port = argc > 2 ? atoi(argv[2]) : 2144; - if (SV_Srv::startServer(scng) && SV_Misc::TCPServer::start(addr, port)){ - statusMess("Run " + addr + " " + to_string(port)); - while (true) - SV_Misc::sleepMs(1000); + if (SV_Srv::startServer(scng, statusMess)) { + const auto tcpErr = SV_Misc::TCPServer::start(addr, port, SV_Srv::receiveData, statusMess); + if (tcpErr.empty()) { + statusMess("Run " + addr + " " + to_string(port)); + while (true) + SV_Misc::sleepMs(1000); + } + else { + statusMess("No run " + addr + " " + to_string(port) + ": " + tcpErr); + cin.get(); + } } else {