async_logger.h

Go to the documentation of this file.

 
#ifndef CPPTRAIL_ASYNC_LOGGER_H
#define CPPTRAIL_ASYNC_LOGGER_H
 
#include <atomic>
#include <condition_variable>
#include <mutex>
#include <queue>
#include <stdexcept>
#include <thread>
#include <vector>
 
#include "cpptrail/base_logger.h"
 
namespace CppTrail {
    class LoggerOverflowError : public std::runtime_error {
    public:
        LoggerOverflowError()
            : std::runtime_error("LoggerOverflowError") {
        }
    };
 
    template<typename char_t>
    class BasicAsyncLoggerImpl : public BasicLoggerImpl<char_t> {
    public:
        using char_type = typename BasicLoggerImpl<char_t>::char_type;
 
        using string_type = typename BasicLoggerImpl<char_t>::string_type;
 
        using message_type = typename BasicLoggerImpl<char_t>::message_type;
 
    protected:
        BasicAsyncLoggerImpl() : BasicAsyncLoggerImpl(
            std::numeric_limits<std::size_t>::max(),
            true, 1
        ) {
        }
 
        BasicAsyncLoggerImpl(
            std::size_t nMaxEntryCount,
            bool bThrowOnOverflow,
            std::size_t nWorkerCount
        );
 
    protected:
        ~BasicAsyncLoggerImpl() override = default;
 
    protected:
        virtual void work(message_type oMessage) = 0;
 
    public:
        void log(message_type oMessage) final;
 
        void start() final;
 
        void stop() final;
 
        void signalStop() final;
 
        void join() final;
 
        Status status() final;
 
    protected:
        virtual Status serviceStatus() = 0;
 
        virtual void serviceStart() = 0;
 
        virtual void serviceStop() = 0;
 
    private:
        std::mutex m_oQueueMutex;
        std::deque<message_type> m_qEntryQueue;
        std::size_t m_nMaxEntryCount;
        bool m_bThrowOnOverflow;
        std::mutex m_oWorkerMutex;
        std::condition_variable m_oWorkerCv;
        std::vector<std::thread> m_vWorkers;
        std::size_t m_nWorkerCount;
        std::atomic_bool m_bWorkerEngineOn{false};
        std::mutex m_oKillerMutex;
        std::vector<std::thread> m_oKillerThread;
        std::condition_variable m_oKillerCv;
    private:
        void work();
 
        void startWorkers();
 
        void stopWorkers();
 
        void waitForKiller();
    };
 
    using AsyncLoggerImpl = BasicAsyncLoggerImpl<char>;
 
    using wAsyncLoggerImpl = BasicAsyncLoggerImpl<wchar_t>;
 
#if __cplusplus >= 202002L
    using u8AsyncLoggerImpl = BasicAsyncLoggerImpl<char8_t>;
#endif
 
    using u16AsyncLoggerImpl = BasicAsyncLoggerImpl<char16_t>;
 
    using u32AsyncLoggerImpl = BasicAsyncLoggerImpl<char32_t>;
 
    template<typename char_t>
    BasicAsyncLoggerImpl<char_t>::BasicAsyncLoggerImpl(
        const std::size_t nMaxEntryCount,
        const bool bThrowOnOverflow,
        const std::size_t nWorkerCount
    ) : m_nMaxEntryCount(nMaxEntryCount),
        m_bThrowOnOverflow(bThrowOnOverflow),
        m_nWorkerCount(nWorkerCount) {
        // Reserve workers
        this->m_vWorkers.reserve(nWorkerCount);
        this->m_oKillerThread.reserve(1);
    }
 
    template<typename char_t>
    void BasicAsyncLoggerImpl<char_t>::log(
        message_type oMessage
    ) {
        // Check if we accept log requests
        if (!m_bWorkerEngineOn.load())return;
 
        // Queue lock
        std::unique_lock<std::mutex> oQueueLock(this->m_oQueueMutex);
 
        // Overflow check
        if (this->m_qEntryQueue.size() + 1 > this->m_nMaxEntryCount) {
            // Throw when overflow
            if (this->m_bThrowOnOverflow)
                throw LoggerOverflowError{};
            // Ignore when overflow
            return;
        }
 
        // Push back message
        this->m_qEntryQueue.push_back(std::move(oMessage));
 
        // Wake up call
        this->m_oWorkerCv.notify_one();
    }
 
 
    template<typename char_t>
    void BasicAsyncLoggerImpl<char_t>::startWorkers() {
        if (!this->m_bWorkerEngineOn.load()) {
            // Create workers
            this->m_vWorkers.clear();
            for (std::size_t i = 0; i < this->m_nWorkerCount; i++) {
                this->m_vWorkers.emplace_back([this] {
                    this->work();
                });
            }
            // Update status
            this->m_bWorkerEngineOn.store(true);
        }
    }
 
    template<typename char_t>
    void BasicAsyncLoggerImpl<char_t>::stopWorkers() {
        if (this->m_bWorkerEngineOn.load()) {
            // Ask for worker termination
            this->m_bWorkerEngineOn.store(false);
            this->m_oWorkerCv.notify_all();
            // Kill workers
            for (auto &oWorker: this->m_vWorkers) {
                if (oWorker.joinable())
                    oWorker.join();
            }
            this->m_vWorkers.clear();
        }
    }
 
    template<typename char_t>
    void BasicAsyncLoggerImpl<char_t>::waitForKiller() {
        // Ensure killer thread is off
        if (!this->m_oKillerThread.empty()) {
            auto &oKiller = this->m_oKillerThread.front();
            if (oKiller.joinable())
                oKiller.join();
            this->m_oKillerThread.clear();
        }
    }
 
    template<typename char_t>
    void BasicAsyncLoggerImpl<char_t>::start() {
        // Ensure killer thread is OFF
        this->m_oKillerMutex.lock();
        this->waitForKiller();
 
        // Worker lock
        std::unique_lock<std::mutex> oWorkerLock(this->m_oWorkerMutex);
 
        // Release the killer lock
        // We no longer need it
        this->m_oKillerMutex.unlock();
 
        // Ensure killer thread is OFF
        this->waitForKiller();
 
        // Get service status
        const auto nStatus = this->serviceStatus();
 
        // Start service
        if (nStatus != Status::TRASCENDENT && nStatus != Status::RUNNING) {
            // External service if OFF
            this->serviceStart();
        }
 
        // Start the working engine
        this->startWorkers();
    }
 
    template<typename char_t>
    void BasicAsyncLoggerImpl<char_t>::stop() {
        // Ensure killer thread is OFF
        this->m_oKillerMutex.lock();
        this->waitForKiller();
 
        // Worker lock
        std::unique_lock<std::mutex> oWorkerLock(this->m_oWorkerMutex);
 
        // Release the killer lock
        // We no longer need it
        this->m_oKillerMutex.unlock();
 
        // Stop the working engine
        this->stopWorkers();
 
        // Get service status
        const auto nStatus = this->serviceStatus();
 
        // Stop service
        if (nStatus == Status::TRASCENDENT || nStatus == Status::RUNNING) {
            // External service if ON
            this->serviceStop();
        }
    }
 
    template<typename char_t>
    void BasicAsyncLoggerImpl<char_t>::signalStop() {
        // Killer lock
        std::unique_lock<std::mutex> oKillerLock(this->m_oKillerMutex);
 
        // Ensure killer thread is OFF
        this->waitForKiller();
 
        // Worker lock
        std::unique_lock<std::mutex> oWorkerLock(this->m_oWorkerMutex);
 
        // Killer thread
        this->m_oKillerThread.emplace_back(
            [this] {
                // Worker lock
                std::unique_lock<std::mutex> oKillerWorkerLock(this->m_oWorkerMutex);
 
                // Stop the working engine
                this->stopWorkers();
 
                // Get service status
                const auto nStatus = this->serviceStatus();
 
                // Stop service
                if (nStatus == Status::TRASCENDENT || nStatus == Status::RUNNING) {
                    // External service if ON
                    this->serviceStop();
                }
            }
        );
    }
 
    template<typename char_t>
    void BasicAsyncLoggerImpl<char_t>::join() {
        // Killer lock
        std::unique_lock<std::mutex> oKillerLock(this->m_oKillerMutex);
        // Ensure killer thread is OFF
        this->waitForKiller();
    }
 
    template<typename char_t>
    Status BasicAsyncLoggerImpl<char_t>::status() {
        // Get service status
        auto nStatus = this->serviceStatus();
 
        // Calculate logger status
        switch (nStatus) {
            case Status::TRASCENDENT:
            case Status::RUNNING:
                return m_bWorkerEngineOn.load() ? nStatus : Status::BROKEN;
            case Status::STOPPED:
                return !m_bWorkerEngineOn.load() ? nStatus : Status::BROKEN;
            case Status::BROKEN:
            default:
                return Status::BROKEN;
        }
    }
 
    template<typename char_t>
    void BasicAsyncLoggerImpl<char_t>::work() {
        bool bContinue;
        do {
            // Queue lock
            std::unique_lock<std::mutex> oQueueLock(this->m_oQueueMutex);
 
            // Continue while there is something to do
            // Continue to end if stop signal
            this->m_oWorkerCv.wait(oQueueLock, [this] {
                // Checkif there are entries queued
                if (!this->m_qEntryQueue.empty())return true;
                // Continue to stop
                return !this->m_bWorkerEngineOn.load();
            });
 
            // Check entry
            if (!this->m_qEntryQueue.empty()) {
                // Get entry
                message_type oEntry = std::move(this->m_qEntryQueue.front());
                this->m_qEntryQueue.pop_front();
 
                // Do work (heavy lifting)
                this->m_oQueueMutex.unlock();
                this->work(std::move(oEntry));
                this->m_oQueueMutex.lock();
            }
 
            // Determine wherther continue
            const bool bEmpty = this->m_qEntryQueue.empty();
            const bool bEngine = this->m_bWorkerEngineOn.load();
            bContinue = bEngine || !bEmpty;
        } while (bContinue);
    }
}
 
#endif