terminal-web-server/src/api-driver/daemon.cpp

#include "daemon.h"
#include <utility>

#include "terminal_api_driver.h"

// минимальный порог для сна в цикле событий демона
static constexpr int64_t SLEEP_THRESHOLD = 10;

int64_t api_driver::TimeNow() {
    return std::chrono::time_point_cast<std::chrono::milliseconds>(std::chrono::system_clock::now()).time_since_epoch().count();
}


namespace api_driver {
    /**
     * Обертка для объектов, доступных для обновления
     * NOTE: перед вызовом функций, требующих `TSID`, необходимо захватить мютекс API.
     */
    class CpUpdatebleObject {
    public:
        int64_t lastUpdate = 0;
        int64_t updatePeriodMs = -1;

        /**
         * Функция для обновления (загрузки) объекта из CP API.
         */
        std::function<void ()> updateCallback;

        explicit CpUpdatebleObject(std::function<void ()> callback, int64_t period = -1): updatePeriodMs(period), updateCallback(std::move(callback)) {}

        bool checkNeedUpdate(int64_t now) const {
            if (updatePeriodMs < 0) return false;
            // тут нет смысла спать меньше чем на 20мс, поэтому можно разрешить чтение на некоторое время раньше
            return now - lastUpdate >= (updatePeriodMs - 20);
        }

        int64_t getNextUpdate(int64_t now) const {
            if (checkNeedUpdate(now)) {
                return 0;
            }
            auto next = now - lastUpdate;
            return next < 0 ? 0 : next;
        }

        ~CpUpdatebleObject() = default;
    };
}


void api_driver::TerminalApiDaemon::connectToApi() {
    {
        std::lock_guard _lock(this->stateMutex);
        this->state.fInitState = "Not connected to API";
    }
#ifdef API_OBJECT_NETWORK_SETTINGS_ENABLE
    {
        std::lock_guard _lock(this->settingsMutex);
        this->settingsNetwork.loadDefaults();
    }
#endif

    for (int connectAttempt = 0;; connectAttempt++) {
        BOOST_LOG_TRIVIAL(info) << "api_driver::TerminalApiDaemon::connectToApi(): Try to connect api (attempt " << connectAttempt << ")...";
        try {
            cp.connect();

            std::string tmp = cp.getDmaDebug("status_init");
            {
                std::lock_guard _lock(this->stateMutex);
                this->state.fInitState = tmp;
            }
            BOOST_LOG_TRIVIAL(info) << "api_driver::TerminalApiDaemon::connectToApi(): Success connect!";
            BOOST_LOG_TRIVIAL(info) << "api_driver::TerminalApiDaemon::connectToApi(): API status: " << tmp;

            obj::TerminalFirmwareVersion f;
            f.load(cp);

            {
                std::lock_guard _lock(this->firmwareMutex);
                this->firmware = f;
            }

            cp.lastCpError = OK;
            break;
        } catch (std::exception& e) {
            BOOST_LOG_TRIVIAL(error) << "api_driver::TerminalApiDaemon::connectToApi(): connect error " << e.what();
        }
        boost::this_thread::sleep_for(boost::chrono::duration(boost::chrono::milliseconds(1000)));
    }
}

void api_driver::TerminalApiDaemon::run() {
    // это демон, который в бесконечном цикле опрашивает API
    this->connectToApi();

    struct {
        CpUpdatebleObject uo;
        std::string updaterName;
    } updaters[] = {
#ifdef API_OBJECT_DEBUG_METRICS_ENABLE
        // обновление логов
        {.uo = CpUpdatebleObject([this]() {
            this->statsLogs.updateCallback(cp);
        }), .updaterName = "updateDebugMetrics"},
#endif
        // обновление статистики
        {.uo = CpUpdatebleObject([this]() {
            std::shared_lock _slock1(this->settingsMutex);
            obj::TerminalState tmp(state);
            _slock1.release();

            tmp.updateCallback(cp);
            std::lock_guard _slock2(this->settingsMutex);
            state = tmp;
        }, CACHE_STATISTICS_UPDATE_MS), .updaterName = "updateStatistics"},
        {.uo = CpUpdatebleObject([this]() {
            obj::TerminalDeviceState tmp(stateDev);
            tmp.updateCallback(cp);
            std::lock_guard _slock2(this->settingsMutex);
            stateDev = tmp;
        }, CACHE_STATISTICS_UPDATE_MS), .updaterName = "updateDeviceState"},
        // обновление кеша настроек
        {.uo = CpUpdatebleObject([this]() {
            obj::TerminalRxTxSettings rxtx;
            rxtx.updateCallback(cp);
            std::lock_guard _slock2(this->settingsMutex);
            settingsRxTx = rxtx;
        }, CACHE_SETTINGS_UPDATE_MS), .updaterName = "updateRxTxSettings"},
#ifdef API_OBJECT_NETWORK_SETTINGS_ENABLE
        {.uo = CpUpdatebleObject([this]() {
            std::shared_lock _slock1(this->settingsMutex);
            obj::TerminalNetworkSettings net(settingsNetwork);
            _slock1.release();

            net.updateCallback(cp);
            std::lock_guard _slock2(this->settingsMutex);
            settingsNetwork = net;
        }, CACHE_SETTINGS_UPDATE_MS), .updaterName = "updateNetworkSettings"},
#endif
#ifdef API_OBJECT_QOS_SETTINGS_ENABLE
        // обновление кеша QoS
        {.uo = CpUpdatebleObject([this]() {
            obj::TerminalQosSettings qos;
            qos.updateCallback(cp);
            std::lock_guard _slock(this->settingsMutex);
            settingsQos = qos;
        }, CACHE_SETTINGS_UPDATE_MS), .updaterName = "updateQosSettings"},
#endif
    };

    while (true) {
        if (this->cp.lastCpError == ERROR || this->cp.lastCpError == TIMEOUT) {
            BOOST_LOG_TRIVIAL(error) << "api_driver::TerminalApiDaemon::run(): close current daemon session caused error " << this->cp.lastCpError;
            cp.disconnect();
            this->connectToApi();
        }
#ifdef API_OBJECT_DEBUG_METRICS_ENABLE
        updaters[0].uo.updatePeriodMs = this->statsLogs.logEn ? this->statsLogs.logPeriodMs.load() : -1;
#endif
        int64_t sleepTime = 60000; // минута по-умолчанию
        auto now = TimeNow();
        for (auto& u: updaters) {
            if (u.uo.checkNeedUpdate(now)) {
                auto targetTime = u.uo.lastUpdate + u.uo.updatePeriodMs;
                if (targetTime + SLEEP_THRESHOLD <= now && targetTime - SLEEP_THRESHOLD >= now) {
                    u.uo.lastUpdate = targetTime;
                } else {
                    u.uo.lastUpdate = now;
                }

                try {
                    std::lock_guard _lock(this->cpApiMutex);
                    u.uo.updateCallback();
                    BOOST_LOG_TRIVIAL(debug) << "api_driver::TerminalApiDaemon::run()->" << u.updaterName << "(): success update!";
                } catch (std::exception& e) {
                    BOOST_LOG_TRIVIAL(error) << "api_driver::TerminalApiDaemon::run()->" << u.updaterName << "(): error " << e.what();
                }

                now = TimeNow();
            }
            if (u.uo.updatePeriodMs >= 0) {
                sleepTime = std::min(sleepTime, u.uo.getNextUpdate(now));
            }
        }

        if (sleepTime > 0) {
            boost::this_thread::sleep_for(boost::chrono::duration(boost::chrono::milliseconds(sleepTime)));
        }
    }
}

api_driver::TerminalApiDaemon::TerminalApiDaemon(): daemon([this]() { this->run(); }) {}

void api_driver::TerminalApiDaemon::getState(obj::TerminalState &dest) {
    std::shared_lock _lock(stateMutex);
    dest = state;
}

void api_driver::TerminalApiDaemon::getDeviceState(obj::TerminalDeviceState &dest) {
    std::shared_lock _lock(stateMutex);
    dest = stateDev;
}


api_driver::obj::TerminalFirmwareVersion api_driver::TerminalApiDaemon::getFirmware() {
    obj::TerminalFirmwareVersion res;
    {
        std::shared_lock _olock(firmwareMutex);
        res = firmware;
    }
    return res;
}

void api_driver::TerminalApiDaemon::resetPacketStatistics() {
    std::lock_guard lock(this->cpApiMutex);
    cp.getDmaDebug("reset_cnt_rx");
}

void api_driver::TerminalApiDaemon::resetDefaultSettings() {
    std::lock_guard lock(this->cpApiMutex);
    cp.setDmaDebug("begin_save_config", "");
    cp.setDmaDebug("default_params", "");
    cp.setDmaDebug("save_config", "");
}

api_driver::TerminalApiDaemon::~TerminalApiDaemon() {
    try {
        daemon.interrupt();
        daemon.try_join_for(boost::chrono::seconds(2));
    } catch (std::exception& e) {
        BOOST_LOG_TRIVIAL(error) << "api_driver::~TerminalApiDaemon(): " << e.what();
    }
}