Bug fix: now compare the actual objects instead of pointer values. Se…

…tup t1 and t2 depending on the thread id.
arangodb · neunhoef · Jan 8, 2019 · Jul 23, 2018 · Jul 24, 2018 · Jul 25, 2018
commit 43fb2d9d62a2f08244b52b25d93a5717f318b3d2
diff --git a/arangod/Scheduler/Scheduler.cpp b/arangod/Scheduler/Scheduler.cpp
@@ -110,11 +110,16 @@ void Scheduler::shutdown () {
 
 void Scheduler::runCron() {
 
+  std::unique_lock<std::mutex> guard(_priorityQueueMutex);
+
+  uint64_t tick = 0;
+
   while (!isStopping()) {
 
+    tick++;
+
     auto now = clock::now();
 
-    std::unique_lock<std::mutex> guard(_priorityQueueMutex);
 
     clock::duration sleepTime = std::chrono::milliseconds(50);
 
@@ -127,7 +132,7 @@ void Scheduler::runCron() {
       } else {
         auto then = (top->_due - now);
 
-        sleepTime = sleepTime > then ? then : sleepTime;
+        sleepTime = (sleepTime > then ? then : sleepTime);
         break ;
       }
     }
@@ -147,8 +152,8 @@ Scheduler::WorkHandle Scheduler::postDelay(clock::duration delay,
   }
 
   std::unique_lock<std::mutex> guard(_priorityQueueMutex);
-
   auto handle = std::make_shared<Scheduler::DelayedWorkItem>(callback, delay);
+
   _priorityQueue.push(handle);
 
   if (delay < std::chrono::milliseconds(50)) {

diff --git a/arangod/Scheduler/Scheduler.h b/arangod/Scheduler/Scheduler.h
@@ -92,10 +92,6 @@ class Scheduler {
       _handler(std::move(handler)), _due(clock::now() + delay), _cancelled(false) {}
 
     void cancel() { _cancelled = true; };
-    bool operator <(const DelayedWorkItem & rhs) const
-    {
-      return _due < rhs._due;
-    }
   };
 
   class WorkGuard {
@@ -154,9 +150,17 @@ class Scheduler {
   // For tasks above 50ms the Cron Thread is woken up to potentially update its sleep time, which
   // could now be shorter than before.
 
+  struct compare {
+    bool operator()(std::shared_ptr<DelayedWorkItem> const &left,
+      std::shared_ptr<DelayedWorkItem> const &right) {
+      // Reverse order, because std::priority_queue is a max heap.
+      return right->_due < left->_due;
+    }
+  };
+
   std::priority_queue<std::shared_ptr<DelayedWorkItem>,
                       std::vector<std::shared_ptr<DelayedWorkItem>>,
-                      std::greater<std::shared_ptr<DelayedWorkItem>>> _priorityQueue;
+                      compare> _priorityQueue;
   std::mutex _priorityQueueMutex;
   std::condition_variable _conditionCron;
 

diff --git a/arangod/Scheduler/SupervisedScheduler.cpp b/arangod/Scheduler/SupervisedScheduler.cpp
@@ -209,12 +209,6 @@ void SupervisedScheduler::shutdown () {
   Scheduler::shutdown();
 }
 
-SupervisedScheduler::WorkerState::WorkerState(SupervisedScheduler &scheduler) :
-  _queueRetryCount
8000
(100),
-  _sleepTimeout_ms(100),
-  _stop(false),
-  _thread(new SupervisedSchedulerWorkerThread(scheduler)) {}
-
 void SupervisedScheduler::runWorker()
 {
   uint64_t id;
@@ -228,6 +222,9 @@ void SupervisedScheduler::runWorker()
 
   auto &state = _workerStates[id];
 
+  state._sleepTimeout_ms = 20 * (id + 1);
+  state._queueRetryCount = (512 >> id) + 3;
+
   while (true) {
     std::unique_ptr<WorkItem> work = getWork(state);
     if (work == nullptr) {
@@ -380,6 +377,11 @@ SupervisedScheduler::WorkerState::WorkerState(SupervisedScheduler::WorkerState &
   _stop(that._stop.load()),
   _thread(std::move(that._thread)) {}
 
+SupervisedScheduler::WorkerState::WorkerState(SupervisedScheduler &scheduler) :
+  _queueRetryCount(100),
+  _sleepTimeout_ms(100),
+  _stop(false),
+  _thread(new SupervisedSchedulerWorkerThread(scheduler)) {}
 
 bool SupervisedScheduler::WorkerState::start() {
   return _thread->start();