threadpool.h

Go to the documentation of this file.

 
#pragma once
 
 
#include <promeki/config.h>
#if PROMEKI_ENABLE_CORE
#include <functional>
#include <memory>
#include <type_traits>
#include <promeki/basicthread.h>
#include <promeki/function.h>
#include <promeki/namespace.h>
#include <promeki/atomic.h>
#include <promeki/duration.h>
#include <promeki/string.h>
#include <promeki/error.h>
#include <promeki/hashmap.h>
#include <promeki/mutex.h>
#include <promeki/readwritelock.h>
#include <promeki/stringregistry.h>
#include <promeki/timestamp.h>
#include <promeki/uniqueptr.h>
#include <promeki/waitcondition.h>
#include <promeki/future.h>
#include <promeki/list.h>
 
PROMEKI_NAMESPACE_BEGIN
 
class ThreadPool {
        public:
                using WorkTag = StringRegistry<"ThreadPoolWorkTag">::Item;
 
                struct WorkStats {
                                WorkTag  tag;          
                                String   name;         
                                Duration totalWall;    
                                Duration totalCpu;     
                                Duration totalQueueWait; 
                                int64_t  count = 0;    
                };
 
                ThreadPool(int maxThreadCount = -1, bool lazy = true);
 
                ~ThreadPool();
 
                ThreadPool(const ThreadPool &) = delete;
                ThreadPool &operator=(const ThreadPool &) = delete;
                ThreadPool(ThreadPool &&) = delete;
                ThreadPool &operator=(ThreadPool &&) = delete;
 
                template <typename F> auto submit(F &&callable) -> Future<std::invoke_result_t<F>> {
                        return submit(WorkTag(), std::forward<F>(callable));
                }
 
                template <typename F> auto submit(WorkTag tag, F &&callable) -> Future<std::invoke_result_t<F>> {
                        using R = std::invoke_result_t<F>;
                        auto      task = std::make_shared<std::packaged_task<R()>>(std::forward<F>(callable));
                        Future<R> fut(task->get_future());
                        bool      runInline = false;
                        TaggedTask entry;
                        entry.tag = tag;
                        entry.enqueuedAt = TimeStamp::now();
                        entry.callable = [task]() { (*task)(); };
                        {
                                Mutex::Locker locker(_mutex);
                                if (_maxThreadCount == 0) {
                                        runInline = true;
                                } else {
                                        _tasks.pushToBack(std::move(entry));
                                        maybeSpawnOne();
                                        _cv.wakeOne();
                                }
                        }
                        if (runInline) {
                                runTaskWithStats(entry);
                        }
                        return fut;
                }
 
                void setNamePrefix(const String &prefix);
 
                String namePrefix() const;
 
                void setThreadCount(int count, bool lazy = true);
 
                int maxThreadCount() const;
 
                int threadCount() const;
 
                int activeThreadCount() const;
 
                void waitForDone();
 
                Error waitForDone(unsigned int timeoutMs);
 
                void clear();
 
                void setName(const String &name);
 
                String name() const;
 
                List<WorkStats> snapshotWorkStats() const;
 
                void resetWorkStats();
 
                static List<ThreadPool *> allPools();
 
        private:
                using Task = Function<void()>;
 
                struct TaggedTask {
                                Task      callable;
                                WorkTag   tag;
                                TimeStamp enqueuedAt;
                };
 
                struct WorkRecord {
                                Atomic<int64_t> totalWallNs{0};
                                Atomic<int64_t> totalCpuNs{0};
                                Atomic<int64_t> totalQueueWaitNs{0};
                                Atomic<int64_t> count{0};
                                String          name;
                                WorkTag         tag;
                };
 
                void workerFunc(int index);
                void spawnThreads(int count);
                void maybeSpawnOne();
 
                void runTaskWithStats(TaggedTask &t);
 
                WorkRecord *recordFor(WorkTag tag);
 
                static Mutex              &registryMutex();
                static List<ThreadPool *> &registry();
 
                mutable Mutex                                  _mutex;
                WaitCondition                                  _cv;
                WaitCondition                                  _doneCv;
                List<TaggedTask>                               _tasks;
                List<BasicThread>                              _threads;
                String                                         _namePrefix;
                String                                         _name;
                int                                            _maxThreadCount = 0;
                int                                            _threadCount = 0;
                int                                            _activeCount = 0;
                int                                            _waitingCount = 0;
                bool                                           _shutdown = false;
 
                mutable ReadWriteLock                          _statsLock;
                HashMap<uint64_t, UniquePtr<WorkRecord>> _stats;
};
 
PROMEKI_NAMESPACE_END
 
#endif // PROMEKI_ENABLE_CORE