AsyncPipeline.hxx

Go to the documentation of this file.

//
// Created by Zhiping Jiang on 22-7-21.
//
 
#ifndef PICOSCENES_PLATFORM_ASYNCPIPELINE_HXX
#define PICOSCENES_PLATFORM_ASYNCPIPELINE_HXX
 
#include "SystemTools.hxx"
#include "TaggedThreadPool.hxx"
 
using namespace std::chrono_literals;
 
template<typename ObjectType>
class AsyncPipeline {
public:
    AsyncPipeline &startService() {
        pipelineState = true;
        pipelinePaused = false;
        std::this_thread::sleep_for(1ms);
        ThreadPoolSingleton::getInstance().AddJob([&] {
            serviceLoop();
        });
        return *this;
    }
 
    void stopService() {
        if (!pipelineState)
            return;
 
        pipelineState = false;
        pipelinePaused = false;
        bufferMutex.lock();
        handleMutex.lock();
        handlers.clear();
        buffer.clear();
        objectTrackingIdQueue.clear();
        bufferMutex.unlock();
        handleMutex.unlock();
        std::this_thread::sleep_for(200ms);
    }
 
    template<typename T>
    uint64_t send(T&& object) {
        std::lock_guard lock(bufferMutex);
        buffer.emplace_back(std::forward<T>(object));
        auto trackingId = currentTrackingId;
        objectTrackingIdQueue.emplace_back(currentTrackingId++);
        bufferCV.notify_all();
        return trackingId;
    }
 
    template<typename T>
    uint64_t sendSync(T&& object) {
        auto trackingId = send(std::forward<T>(object));
        std::unique_lock syncLock(sendSyncMutex);
        while (pipelineState) {
            syncSendCV.wait_for(syncLock, 1ms, [&]() {
                return std::find(objectTrackingIdQueue.begin(), objectTrackingIdQueue.end(), trackingId) == objectTrackingIdQueue.end();
            });
 
            if (std::find(objectTrackingIdQueue.begin(), objectTrackingIdQueue.end(), trackingId) == objectTrackingIdQueue.end())
                break;
        }
        return trackingId;
    }
 
    int pause() {
        pipelinePaused = true;
        pauseCV.notify_all();
        return 0;
    }
 
    int resume() {
        pipelinePaused = false;
        pauseCV.notify_all();
        return 0;
    }
 
    AsyncPipeline<ObjectType> &registerAsyncHandler(std::string handlerName, std::function<void(const ObjectType &object)> handler) {
        auto guardHandle = std::lock_guard(handleMutex);
        handlers[handlerName] = handler;
 
        return *this;
    }
 
    std::optional<ObjectType> selectDataByCondition(const std::function<bool(const ObjectType &)> &condition, const std::chrono::milliseconds &maxWait_ms, std::optional<std::string> conditionName) {
        thread_local static uint64_t conditionCount = 0;
        std::optional<ObjectType> returnValue;
        std::unique_lock syncLock(handleSyncMutex);
        if (!conditionName)
            conditionName = "Conditional" + std::to_string(conditionCount);
 
        registerAsyncHandler(*conditionName, [=, this, &returnValue](const ObjectType &itemToBeProcessed) -> bool {
            if (condition && condition(itemToBeProcessed)) {
                returnValue = itemToBeProcessed;
                handleSyncCV.notify_all();
            }
            return false;
        });
 
        handleSyncCV.wait_for(syncLock, maxWait_ms, [&]() -> bool {
            return returnValue.has_value();
        });
 
        unregisterHandler(*conditionName);
        conditionCount++;
 
        return returnValue;
    }
 
    AsyncPipeline<ObjectType> &unregisterHandler(const std::string &handlerName) {
        auto guardHandle = std::lock_guard(handleMutex);
        if (handlers.contains(handlerName)) {
            handlers.erase(handlerName);
        }
 
        return *this;
    }
 
    bool hasHandler(const std::string &handlerName) {
        return handlers.contains(handlerName);
    }
 
    [[nodiscard]] bool isPipelineRunning() const {
        return pipelineState;
    }
 
    [[nodiscard]] bool isPipelinePaused() const {
        return pipelinePaused;
    }
 
private:
    bool pipelineState = false;
 
    bool pipelinePaused = false;
 
    std::mutex bufferMutex, handleMutex, waitMutex, sendSyncMutex, handleSyncMutex, pauseMutex;
    std::condition_variable bufferCV, syncSendCV, handleSyncCV, pauseCV;
 
    std::deque<ObjectType> buffer;
 
    std::deque<uint64_t> objectTrackingIdQueue;
    uint64_t currentTrackingId{0UL};
 
    std::map<std::string, std::function<void(const ObjectType &object)>> handlers;
 
    void serviceLoop() {
        auto lock = std::unique_lock(waitMutex);
        auto pauseLock = std::unique_lock(pauseMutex);
        while (pipelineState) {
            bufferCV.wait_for(lock, 50ms, [&] {
                return pipelineState && !buffer.empty();
            });
 
            pauseCV.wait(pauseLock, [&] {
                return !pipelinePaused;
            });
 
            if (pipelineState && !buffer.empty() && !pipelinePaused) {
                const auto &front = buffer.front();
                auto guardHandle = std::lock_guard(handleMutex);
                std::for_each(handlers.cbegin(), handlers.cend(), [&](const auto &handle) {
                    handle.second(front);
                });
 
                auto guard = std::lock_guard(bufferMutex);
                buffer.pop_front();
                objectTrackingIdQueue.pop_front();
            }
        }
    }
};
 
#endif //PICOSCENES_PLATFORM_ASYNCPIPELINE_HXX