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

   1 /******************************************************************************
   2  * Icinga 2                                                                   *
   3  * Copyright (C) 2012-2018 Icinga Development Team (https://www.icinga.com/)  *
   4  *                                                                            *
   5  * This program is free software; you can redistribute it and/or              *
   6  * modify it under the terms of the GNU General Public License                *
   7  * as published by the Free Software Foundation; either version 2             *
   8  * of the License, or (at your option) any later version.                     *
   9  *                                                                            *
  10  * This program is distributed in the hope that it will be useful,            *
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of             *
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the              *
  13  * GNU General Public License for more details.                               *
  14  *                                                                            *
  15  * You should have received a copy of the GNU General Public License          *
  16  * along with this program; if not, write to the Free Software Foundation     *
  17  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA.             *
  18  ******************************************************************************/
  19
  20 #ifndef WORKQUEUE_H
  21 #define WORKQUEUE_H
  22
  23 #include "base/i2-base.hpp"
  24 #include "base/timer.hpp"
  25 #include "base/ringbuffer.hpp"
  26 #include <boost/thread/thread.hpp>
  27 #include <boost/thread/mutex.hpp>
  28 #include <boost/thread/condition_variable.hpp>
  29 #include <boost/exception_ptr.hpp>
  30 #include <queue>
  31 #include <deque>
  32 #include <atomic>
  33
  34 namespace icinga
  35 {
  36
  37 enum WorkQueuePriority
  38 {
  39         PriorityLow,
  40         PriorityNormal,
  41         PriorityHigh
  42 };
  43
  44 using TaskFunction = std::function<void ()>;
  45
  46 struct Task
  47 {
  48         Task() = default;
  49
  50         Task(TaskFunction function, WorkQueuePriority priority, int id)
  51                 : Function(std::move(function)), Priority(priority), ID(id)
  52         { }
  53
  54         TaskFunction Function;
  55         WorkQueuePriority Priority{PriorityNormal};
  56         int ID{-1};
  57 };
  58
  59 bool operator<(const Task& a, const Task& b);
  60
  61 /**
  62  * A workqueue.
  63  *
  64  * @ingroup base
  65  */
  66 class WorkQueue
  67 {
  68 public:
  69         typedef std::function<void (boost::exception_ptr)> ExceptionCallback;
  70
  71         WorkQueue(size_t maxItems = 0, int threadCount = 1);
  72         ~WorkQueue();
  73
  74         void SetName(const String& name);
  75         String GetName() const;
  76
  77         boost::mutex::scoped_lock AcquireLock();
  78         void EnqueueUnlocked(boost::mutex::scoped_lock& lock, TaskFunction&& function, WorkQueuePriority priority = PriorityNormal);
  79         void Enqueue(TaskFunction&& function, WorkQueuePriority priority = PriorityNormal,
  80                 bool allowInterleaved = false);
  81         void Join(bool stop = false);
  82
  83         template<typename VectorType, typename FuncType>
  84         void ParallelFor(const VectorType& items, const FuncType& func)
  85         {
  86                 using SizeType = decltype(items.size());
  87
  88                 SizeType totalCount = items.size();
  89
  90                 auto lock = AcquireLock();
  91
  92                 SizeType offset = 0;
  93
  94                 for (int i = 0; i < m_ThreadCount; i++) {
  95                         SizeType count = totalCount / static_cast<SizeType>(m_ThreadCount);
  96                         if (static_cast<SizeType>(i) < totalCount % static_cast<SizeType>(m_ThreadCount))
  97                                 count++;
  98
  99                         EnqueueUnlocked(lock, [&items, func, offset, count, this]() {
 100                                 for (SizeType j = offset; j < offset + count; j++) {
 101                                         RunTaskFunction([&func, &items, j]() {
 102                                                 func(items[j]);
 103                                         });
 104                                 }
 105                         });
 106
 107                         offset += count;
 108                 }
 109
 110                 ASSERT(offset == items.size());
 111         }
 112
 113         bool IsWorkerThread() const;
 114
 115         size_t GetLength() const;
 116         size_t GetTaskCount(RingBuffer::SizeType span);
 117
 118         void SetExceptionCallback(const ExceptionCallback& callback);
 119
 120         bool HasExceptions() const;
 121         std::vector<boost::exception_ptr> GetExceptions() const;
 122         void ReportExceptions(const String& facility) const;
 123
 124 protected:
 125         void IncreaseTaskCount();
 126
 127 private:
 128         int m_ID;
 129         String m_Name;
 130         static std::atomic<int> m_NextID;
 131         int m_ThreadCount;
 132         bool m_Spawned{false};
 133
 134         mutable boost::mutex m_Mutex;
 135         boost::condition_variable m_CVEmpty;
 136         boost::condition_variable m_CVFull;
 137         boost::condition_variable m_CVStarved;
 138         boost::thread_group m_Threads;
 139         size_t m_MaxItems;
 140         bool m_Stopped{false};
 141         int m_Processing{0};
 142         std::priority_queue<Task, std::deque<Task> > m_Tasks;
 143         int m_NextTaskID{0};
 144         ExceptionCallback m_ExceptionCallback;
 145         std::vector<boost::exception_ptr> m_Exceptions;
 146         Timer::Ptr m_StatusTimer;
 147         double m_StatusTimerTimeout;
 148
 149         RingBuffer m_TaskStats;
 150         size_t m_PendingTasks{0};
 151         double m_PendingTasksTimestamp{0};
 152
 153         void WorkerThreadProc();
 154         void StatusTimerHandler();
 155
 156         void RunTaskFunction(const TaskFunction& func);
 157 };
 158
 159 }
 160
 161 #endif /* WORKQUEUE_H */