strand.h

Go to the documentation of this file.

 
#pragma once
 
 
#include <promeki/config.h>
#if PROMEKI_ENABLE_CORE
#include <functional>
#include <type_traits>
#include <memory>
#include <promeki/function.h>
#include <promeki/deque.h>
#include <promeki/namespace.h>
#include <promeki/mutex.h>
#include <promeki/waitcondition.h>
#include <promeki/future.h>
#include <promeki/promise.h>
#include <promeki/threadpool.h>
 
PROMEKI_NAMESPACE_BEGIN
 
class Strand {
        public:
                using TaskFunc = Function<void()>;
 
                explicit Strand(ThreadPool &pool) : _pool(pool) {}
 
                ~Strand() { waitForIdle(); }
 
                Strand(const Strand &) = delete;
                Strand &operator=(const Strand &) = delete;
                Strand(Strand &&) = delete;
                Strand &operator=(Strand &&) = delete;
 
                template <typename F>
                auto submit(F &&callable, TaskFunc onCancel = {}) -> Future<std::invoke_result_t<F>> {
                        return submitImpl(std::forward<F>(callable), std::move(onCancel), false);
                }
 
                template <typename F>
                auto submitUrgent(F &&callable, TaskFunc onCancel = {}) -> Future<std::invoke_result_t<F>> {
                        return submitImpl(std::forward<F>(callable), std::move(onCancel), true);
                }
 
                size_t cancelPending() {
                        EntryQueue toCancel;
                        {
                                Mutex::Locker lock(_mutex);
                                toCancel = std::move(_queue);
                                _queue.clear();
                        }
                        size_t count = toCancel.size();
                        for (auto &entry : toCancel) {
                                entry.cancel();
                        }
                        return count;
                }
 
                void waitForIdle() {
                        Mutex::Locker lock(_mutex);
                        _idleCv.wait(_mutex, [this] { return !_running; });
                }
 
                bool isBusy() const {
                        Mutex::Locker lock(_mutex);
                        return _running;
                }
 
                size_t pendingCount() const {
                        Mutex::Locker lock(_mutex);
                        return _queue.size();
                }
 
                void setWorkTag(ThreadPool::WorkTag tag) {
                        Mutex::Locker lock(_mutex);
                        _workTag = tag;
                }
 
                ThreadPool::WorkTag workTag() const {
                        Mutex::Locker lock(_mutex);
                        return _workTag;
                }
 
        private:
                struct Entry {
                                TaskFunc run;
                                TaskFunc cancel;
                };
 
                template <typename F>
                auto submitImpl(F &&callable, TaskFunc onCancel, bool urgent) -> Future<std::invoke_result_t<F>> {
                        using R = std::invoke_result_t<F>;
                        auto      promise = std::make_shared<Promise<R>>();
                        Future<R> future = promise->future();
 
                        Entry entry;
                        entry.run = [promise, f = std::forward<F>(callable)]() mutable {
                                if constexpr (std::is_void_v<R>) {
                                        f();
                                        promise->setValue();
                                } else {
                                        promise->setValue(f());
                                }
                        };
                        entry.cancel = [promise, onCancel = std::move(onCancel)]() mutable {
                                if (onCancel) onCancel();
                                promise->setError(Error::Cancelled);
                        };
 
                        ThreadPool::WorkTag spawnTag;
                        bool                needSpawn = false;
                        {
                                Mutex::Locker lock(_mutex);
                                if (urgent) {
                                        _queue.pushToFront(std::move(entry));
                                } else {
                                        _queue.pushToBack(std::move(entry));
                                }
                                if (!_running) {
                                        _running = true;
                                        needSpawn = true;
                                }
                                spawnTag = _workTag;
                        }
                        if (needSpawn) {
                                _pool.submit(spawnTag, [this] { runNext(); });
                        }
                        return future;
                }
 
                void runNext() {
                        Entry entry;
                        bool  haveTask = false;
                        {
                                Mutex::Locker lock(_mutex);
                                if (!_queue.isEmpty()) {
                                        entry = _queue.popFromFront();
                                        haveTask = true;
                                }
                        }
 
                        // Possible if cancelPending() emptied the queue
                        // between spawn and execution.
                        if (haveTask) entry.run();
 
                        // Either re-submit (more work pending) or mark idle.
                        bool reSpawn = false;
                        {
                                Mutex::Locker lock(_mutex);
                                if (_queue.isEmpty()) {
                                        _running = false;
                                        _idleCv.wakeAll();
                                } else {
                                        reSpawn = true;
                                }
                        }
 
                        if (reSpawn) {
                                ThreadPool::WorkTag tag;
                                {
                                        Mutex::Locker lock(_mutex);
                                        tag = _workTag;
                                }
                                _pool.submit(tag, [this] { runNext(); });
                        }
                }
 
                using EntryQueue = Deque<Entry>;
 
                ThreadPool         &_pool;
                mutable Mutex       _mutex;
                WaitCondition       _idleCv;
                EntryQueue          _queue;
                ThreadPool::WorkTag _workTag;
                bool                _running = false;
};
 
PROMEKI_NAMESPACE_END
 
#endif // PROMEKI_ENABLE_CORE