Renderive/YSGraphic_Core/base/Global.cpp

#include "asio.hpp"
#include "Global.h"
#include <QApplication>
#include <atomic>
#include <algorithm>
#include <condition_variable>
#include <mutex>
#include <thread>
#include "Plot_p.h"

namespace YSG {
    struct TimerThreadPrivate {
        asio::io_context mIoContext;
        std::unique_ptr<asio::executor_work_guard<asio::io_context::executor_type>> mWorkGuard;
        std::unique_ptr<asio::thread_pool> mCpuPool;
        std::thread::id mSchedulerThreadId;
        std::atomic_bool mIoActive{false};
        std::atomic_bool mCpuJoined{false};
    };

    TimerThread::TimerThread() : d(std::make_unique<TimerThreadPrivate>()) {
        connect(QApplication::instance(), &QApplication::aboutToQuit, [this]() {
            shutdown();
            delete this;
        });
    }
    TimerThread::~TimerThread() = default;

    void TimerThread::run() {
        d->mSchedulerThreadId = std::this_thread::get_id();
        d->mIoContext.restart();
        d->mWorkGuard = std::make_unique<asio::executor_work_guard<asio::io_context::executor_type>>(d->mIoContext.get_executor());
        unsigned cpuCount = std::max(1u, std::thread::hardware_concurrency() > 1 ? std::thread::hardware_concurrency() - 1 : 1u);
        d->mCpuPool = std::make_unique<asio::thread_pool>(cpuCount);
        d->mCpuJoined.store(false, std::memory_order_release);
        d->mIoActive.store(true, std::memory_order_release);
        for (auto &plot: mPlots) {
            plot->d->mTimerThread = this;
            plot->d->mRenderTimer = std::make_unique<asio::steady_timer>(d->mIoContext);
            if(plot->isVisible()) plot->d->mRenderEnabled = true;
            if(plot->d->mRenderEnabled) plot->scheduleRenderTimer();
            plot->requestRender();
        }
        d->mIoContext.run();
        d->mIoActive.store(false, std::memory_order_release);
        for (auto &plot: mPlots) {
            if(plot->d->mRenderTimer) {
                plot->d->mRenderTimer->cancel();
                plot->d->mRenderTimer.reset();
            }
        }
        joinCpuPool();
        d->mCpuPool.reset();
        d->mWorkGuard.reset();
    }
    void TimerThread::post(std::function<void()> task) {
        asio::post(d->mIoContext, std::move(task));
    }
    void TimerThread::postCpu(std::function<void()> task) {
        if(d->mCpuPool) {
            asio::post(*d->mCpuPool, std::move(task));
        } else {
            post(std::move(task));
        }
    }
    bool TimerThread::isSchedulerThread() const {
        return std::this_thread::get_id() == d->mSchedulerThreadId;
    }
    asio::io_context& TimerThread::ioContext() {
        return d->mIoContext;
    }
    void TimerThread::removePlot(Plot* plot) {
        postAndWait([this, plot]() {
            removePlotOnScheduler(plot);
        });
    }
    void TimerThread::shutdown() {
        if(!isRunning()) {
            shutdownOnScheduler();
            stopSchedulerOnScheduler();
            joinCpuPool();
            d->mCpuPool.reset();
            return;
        }
        if(isSchedulerThread()) {
            shutdownOnScheduler();
            stopSchedulerOnScheduler();
            return;
        }
        postAndWait([this]() {
            shutdownOnScheduler();
        });
        joinCpuPool();
        postAndWait([this]() {
            stopSchedulerOnScheduler();
        });
        wait();
    }
    void TimerThread::joinCpuPool() {
        if(d->mCpuPool && !d->mCpuJoined.exchange(true, std::memory_order_acq_rel)) d->mCpuPool->join();
    }
    void TimerThread::postAndWait(std::function<void()> task) {
        if(isSchedulerThread() || !isRunning() || !d->mIoActive.load(std::memory_order_acquire)) {
            task();
            return;
        }
        std::mutex mutex;
        std::condition_variable cv;
        bool done = false;
        post([&task, &mutex, &cv, &done]() {
            task();
            {
                std::lock_guard<std::mutex> lock(mutex);
                done = true;
            }
            cv.notify_one();
        });
        std::unique_lock<std::mutex> lock(mutex);
        cv.wait(lock, [&done]() {
            return done;
        });
    }
    void TimerThread::removePlotOnScheduler(Plot* plot) {
        mPlots.removeAll(plot);
        PlotPrivate* data = plot->d;
        data->mRenderEnabled = false;
        if(data->mRenderTimer) {
            data->mRenderTimer->cancel();
            data->mRenderTimer.reset();
        }
        data->mTimerThread = nullptr;
    }
    void TimerThread::shutdownOnScheduler() {
        for(Plot* plot : mPlots) {
            PlotPrivate* data = plot->d;
            data->mDestroying.store(true, std::memory_order_release);
            data->mRenderEnabled = false;
            if(data->mRenderTimer) {
                data->mRenderTimer->cancel();
                data->mRenderTimer.reset();
            }
            data->mTimerThread = nullptr;
        }
        mPlots.clear();
    }
    void TimerThread::stopSchedulerOnScheduler() {
        d->mWorkGuard.reset();
    }

    void Global::startAllTimeThread() {
        for (auto &thread: mTimerThreadMap) {
            if(!thread->isRunning()) thread->start();
        }
    }

    void Global::timerEvent(QTimerEvent* event) {
        Q_UNUSED(event)
    }
}