granicus.if.org Git - icinga2/blob - lib/base/workqueue.hpp

   1 /* Icinga 2 | (c) 2012 Icinga GmbH | GPLv2+ */
   2
   3 #ifndef WORKQUEUE_H
   4 #define WORKQUEUE_H
   5
   6 #include "base/i2-base.hpp"
   7 #include "base/timer.hpp"
   8 #include "base/ringbuffer.hpp"
   9 #include <boost/thread/thread.hpp>
  10 #include <boost/thread/mutex.hpp>
  11 #include <boost/thread/condition_variable.hpp>
  12 #include <boost/exception_ptr.hpp>
  13 #include <queue>
  14 #include <deque>
  15 #include <atomic>
  16
  17 namespace icinga
  18 {
  19
  20 enum WorkQueuePriority
  21 {
  22         PriorityLow = 0,
  23         PriorityNormal = 1,
  24         PriorityHigh = 2,
  25         PriorityImmediate = 4
  26 };
  27
  28 using TaskFunction = std::function<void ()>;
  29
  30 struct Task
  31 {
  32         Task() = default;
  33
  34         Task(TaskFunction function, WorkQueuePriority priority, int id)
  35                 : Function(std::move(function)), Priority(priority), ID(id)
  36         { }
  37
  38         TaskFunction Function;
  39         WorkQueuePriority Priority{PriorityNormal};
  40         int ID{-1};
  41 };
  42
  43 bool operator<(const Task& a, const Task& b);
  44
  45 /**
  46  * A workqueue.
  47  *
  48  * @ingroup base
  49  */
  50 class WorkQueue
  51 {
  52 public:
  53         typedef std::function<void (boost::exception_ptr)> ExceptionCallback;
  54
  55         WorkQueue(size_t maxItems = 0, int threadCount = 1);
  56         ~WorkQueue();
  57
  58         void SetName(const String& name);
  59         String GetName() const;
  60
  61         boost::mutex::scoped_lock AcquireLock();
  62         void EnqueueUnlocked(boost::mutex::scoped_lock& lock, TaskFunction&& function, WorkQueuePriority priority = PriorityNormal);
  63         void Enqueue(TaskFunction&& function, WorkQueuePriority priority = PriorityNormal,
  64                 bool allowInterleaved = false);
  65         void Join(bool stop = false);
  66
  67         template<typename VectorType, typename FuncType>
  68         void ParallelFor(const VectorType& items, const FuncType& func)
  69         {
  70                 using SizeType = decltype(items.size());
  71
  72                 SizeType totalCount = items.size();
  73
  74                 auto lock = AcquireLock();
  75
  76                 SizeType offset = 0;
  77
  78                 for (int i = 0; i < m_ThreadCount; i++) {
  79                         SizeType count = totalCount / static_cast<SizeType>(m_ThreadCount);
  80                         if (static_cast<SizeType>(i) < totalCount % static_cast<SizeType>(m_ThreadCount))
  81                                 count++;
  82
  83                         EnqueueUnlocked(lock, [&items, func, offset, count, this]() {
  84                                 for (SizeType j = offset; j < offset + count; j++) {
  85                                         RunTaskFunction([&func, &items, j]() {
  86                                                 func(items[j]);
  87                                         });
  88                                 }
  89                         });
  90
  91                         offset += count;
  92                 }
  93
  94                 ASSERT(offset == items.size());
  95         }
  96
  97         bool IsWorkerThread() const;
  98
  99         size_t GetLength() const;
 100         size_t GetTaskCount(RingBuffer::SizeType span);
 101
 102         void SetExceptionCallback(const ExceptionCallback& callback);
 103
 104         bool HasExceptions() const;
 105         std::vector<boost::exception_ptr> GetExceptions() const;
 106         void ReportExceptions(const String& facility) const;
 107
 108 protected:
 109         void IncreaseTaskCount();
 110
 111 private:
 112         int m_ID;
 113         String m_Name;
 114         static std::atomic<int> m_NextID;
 115         int m_ThreadCount;
 116         bool m_Spawned{false};
 117
 118         mutable boost::mutex m_Mutex;
 119         boost::condition_variable m_CVEmpty;
 120         boost::condition_variable m_CVFull;
 121         boost::condition_variable m_CVStarved;
 122         boost::thread_group m_Threads;
 123         size_t m_MaxItems;
 124         bool m_Stopped{false};
 125         int m_Processing{0};
 126         std::priority_queue<Task, std::deque<Task> > m_Tasks;
 127         int m_NextTaskID{0};
 128         ExceptionCallback m_ExceptionCallback;
 129         std::vector<boost::exception_ptr> m_Exceptions;
 130         Timer::Ptr m_StatusTimer;
 131         double m_StatusTimerTimeout;
 132
 133         RingBuffer m_TaskStats;
 134         size_t m_PendingTasks{0};
 135         double m_PendingTasksTimestamp{0};
 136
 137         void WorkerThreadProc();
 138         void StatusTimerHandler();
 139
 140         void RunTaskFunction(const TaskFunction& func);
 141 };
 142
 143 }
 144
 145 #endif /* WORKQUEUE_H */