ThreadPool.h

/*
** libutil++
** $Id: ThreadPool.h 1653 2016-02-28 19:54:59Z sella $
** Copyright (c) 2011-2016 Digital Genesis, LLC. All Rights Reserved.
** Released under the LGPL Version 2.1 License.
** http://www.digitalgenesis.com
*/

#ifndef __libutilxx__sella__util__ThreadPool_H__
#define __libutilxx__sella__util__ThreadPool_H__

#include "../../common.h"
#include "ThreadPoolException.h"
#include "Task.h"
#include "TaskQueue.h"
#include "Log.h"

#include <sched.h>

#include <set>
#include <thread>
#include <chrono>
#include <condition_variable>

#if defined __GNUC__ && (__GNUC__ > 4 || (__GNUC__ == 4 && __GNUC_MINOR__ >= 5))
        #include <atomic> // GCC 4.6 and up
#else
        #include <stdatomic.h> // GCC 4.4 and 4.5 only
#endif

namespace sella {
        namespace util {
                class ThreadPool;
        }
}

class sella::util::ThreadPool : protected TaskQueue { /* NOTE: ThreadPool is thread safe. */
        public:
                typedef std::shared_ptr<ThreadPool> shared;

        public:
                static const size_t DefaultMaximumWorkers = 0; /* 0 to auto assign available core count */
                static const size_t DefaultMinimumWorkers = 2;
                static const size_t DefaultSpareWorkers = 2; /* Number of non-active workers. */

        public:
                ThreadPool(size_t maximumWorkers = DefaultMaximumWorkers) throw (ThreadPoolException); /* Simple non-scaling pool. */
                ThreadPool(size_t maximumWorkers, size_t minimumWorkers, size_t spareWorkers = DefaultSpareWorkers) throw (ThreadPoolException); /* Auto-scaling pool. */
                ThreadPool(Log::shared log, size_t maximumWorkers = DefaultMaximumWorkers) throw (ThreadPoolException);
                ThreadPool(Log::shared log, size_t maximumWorkers, size_t minimumWorkers, size_t spareWorkers = DefaultSpareWorkers) throw (ThreadPoolException);
                ThreadPool(const ThreadPool &other) = delete;
                virtual ~ThreadPool();

                void schedule(Task::shared &task, int priority = 0);
                void schedule(Task::shared &task, const std::chrono::system_clock::duration &delay, int priority = 0);
                void schedule(Task::shared &task, const std::chrono::system_clock::time_point &time, int priority = 0);
                void schedule(Task::shared &task, time_t sec, suseconds_t usec, int priority = 0);
                void schedule(Task::shared &task, timeval &tv, int priority = 0);
                bool reschedule(Task::shared &task, const std::chrono::system_clock::duration &delay, int priority = 0, bool strict = false);
                bool reschedule(Task::shared &task, const std::chrono::system_clock::time_point &time, int priority = 0, bool strict = false);
                bool reschedule(Task::shared &task, time_t sec, suseconds_t usec = 0, int priority = 0, bool strict = false);
                bool reschedule(Task::shared &task, timeval &tv, int priority = 0, bool strict = false);
                bool unschedule(Task::shared &task, bool firstonly = true);

                void pause(bool on = true);
                void resume(void) { pause(false); }

                size_t getCPUCount(void);
                int getCPUSetCount(void);
                bool addCPUAffinity(int cpu);
                bool addCPUAffinity(std::set<int> cpus);
                bool setCPUAffinity(cpu_set_t &cpuset);
                bool clearCPUAffinity(void);
                bool clearCPUAffinity(int cpu);
                bool clearCPUAffinity(std::set<int> cpus);

                size_t size(void) const;
                size_t workers(void) const;
                size_t active(void) const;
                bool empty(void) const;
                void clear(void);
                
                Task::shared next(void);
                std::chrono::microseconds duration(void) const;
                std::chrono::microseconds max(void) const;
                bool ready(void) const;

                bool wait(void); /* Wait until completely empty */
                bool wait(const std::chrono::system_clock::duration &delay);
                bool wait(const std::chrono::system_clock::time_point &time);

                using TaskQueue::capacity;
                void reserve(size_t capacity);
                void print(void) const;
                using TaskQueue::setLog;

                static ThreadPool::shared shared_ptr(size_t maximumWorkers = DefaultMaximumWorkers) throw (ThreadPoolException);
                static ThreadPool::shared shared_ptr(size_t maximumWorkers, size_t minimumWorkers, size_t spareWorkers = DefaultSpareWorkers) throw (ThreadPoolException);
                static ThreadPool::shared shared_ptr(Log::shared log, size_t maximumWorkers = DefaultMaximumWorkers) throw (ThreadPoolException);
                static ThreadPool::shared shared_ptr(Log::shared log, size_t maximumWorkers, size_t minimumWorkers, size_t spareWorkers = DefaultSpareWorkers) throw (ThreadPoolException);

        protected:
                std::thread managerTID;
                std::set<pthread_t> tids;
                std::atomic<size_t> threads;
                std::atomic<size_t> creating;
                std::atomic<size_t> actives;
                std::atomic<size_t> overage;
                bool shutdown;
                bool paused;
                bool allcpus;
                cpu_set_t cpuset;

                mutable std::mutex heapMutex;
                mutable std::mutex tidsMutex;
                mutable std::mutex managerMutex;
                mutable std::mutex workerMutex;
                mutable std::mutex overageMutex;
                mutable std::mutex emptyMutex;
                std::condition_variable managerCV;
                std::condition_variable workerCV;
                std::condition_variable emptyCV;

                size_t maximum;
                size_t minimum;
                size_t spares;

                void manager(void);
                void worker(void);
                void normalize(void);
                void createWorker(void);
                bool setThreadsAffinity(void);
                bool setThreadAffinity(pthread_t tid);
};

#endif

/*
** vim: noet ts=3 sw=3
*/