Commented on source code and added TODOs.

arangodb · neunhoef · Jan 8, 2019 · Jul 23, 2018 · Jul 24, 2018 · Jul 25, 2018
commit 5e6a3f26513dcde1a6b5a572c01ee46b38d7adf6
diff --git a/arangod/Pregel/Conductor.cpp b/arangod/Pregel/Conductor.cpp
@@ -469,7 +469,7 @@ void Conductor::startRecovery() {
       cancelNoLock();
       LOG_TOPIC(ERR, Logger::PREGEL) << "Compensation failed";
     }
-  });
+  }).detach();
 }
 
 // resolves into an ordered list of shards for each collection on each server

diff --git a/arangod/Pregel/Worker.cpp b/arangod/Pregel/Worker.cpp
@@ -599,7 +599,7 @@ void Worker<V, E, M>::_continueAsync() {
       _workerAggregators->resetValues();
       _startProcessing();
     }
-  });
+  }).detach();
 }
 
 template <typename V, typename E, typename M>

diff --git a/arangod/Scheduler/Scheduler.h b/arangod/Scheduler/Scheduler.h
@@ -67,12 +67,17 @@ class Scheduler {
     uint64_t _queued;
   };
 
+  // Enqueues a task at highest priority
   virtual void post(std::function<void()> const& callback) = 0;
+  // Enqueues a task at given priority
   virtual bool queue(RequestPriority prio, std::function<void()> const&) = 0;
 
+  // A spinning threads looks at the queues in a cyclic way, i.e.
+  // 0 1 2 0 1 2 0 1 2 0 1 2 ..
 
   typedef std::chrono::steady_clock clock;
 
+private:
   struct DelayedWorkItem {
     std::function<void(bool cancelled)> _handler;
     const clock::time_point _due;
@@ -103,6 +108,7 @@ class Scheduler {
     void detach() { _item.reset(); }
   };
 
+public:
   class WorkHandle {
     std::shared_ptr<WorkGuard> _guard;
   public:
@@ -113,8 +119,14 @@ class Scheduler {
     void detach() { _guard->detach(); }
   };
 
+  // postDelay returns a WorkHandler. You can cancel the job by calling cancel. The job is also
+  // cancelled if all WorkHandles are destructed. To disable this behavior call detach.
   WorkHandle postDelay(clock::duration delay, std::function<void(bool cancelled)> const& callback);
 
+  // TODO: This method is unintuitive to use. If you call it and you are not interested in canceling
+  // you have to call detach(). Otherwise the WorkHandle goes out of scope and the task is canceled
+  // immediately.
+
   virtual void addQueueStatistics(velocypack::Builder&) const = 0;
   virtual QueueStatistics queueStatistics() const = 0;
   virtual std::string infoStatus() const = 0;
@@ -134,6 +146,14 @@ class Sche
10000
duler {
  private:
   friend class SchedulerCronThread;
 
+  // The priority queue is managed by a Cron Thread. It wakes up on a regular basis (10ms currently)
+  // and looks at queue.top(). It the _due time is smaller than now() and the task is not canceled
+  // it is posted on the scheduler. The next sleep time is computed depending on queue top.
+  //
+  // Note that tasks that have a delay of less than 1ms are posted directly.
+  // For tasks above 50ms the Cron Thread is woken up to potentially update its sleep time, which
+  // could now be shorter than before.
+
   std::priority_queue<std::shared_ptr<DelayedWorkItem>,
                       std::vector<std::shared_ptr<DelayedWorkItem>>,
                       std::greater<std::shared_ptr<DelayedWorkItem>>> _priorityQueue;
@@ -142,7 +162,6 @@ class Scheduler {
 
   void runCron();
 
-
   std::unique_ptr<SchedulerCronThread> _cronThread;
 };
 }

diff --git a/arangod/Scheduler/SupervisedScheduler.cpp b/arangod/Scheduler/SupervisedScheduler.cpp
@@ -101,24 +101,6 @@ SupervisedScheduler::SupervisedScheduler(uint64_t minThreads,
   _queue[1].reserve(fifo1Size);
   _queue[2].reserve(fifo2Size);
   _workerStates.reserve(maxThreads);
-
-#ifdef _WIN32
-// Windows does not support POSIX signal handling
-#else
-  struct sigaction action;
-  memset(&action, 0, sizeof(action));
-  sigfillset(&action.sa_mask);
-
-  // ignore broken pipes
-  action.sa_handler = SIG_IGN;
-
-  int res = sigaction(SIGPIPE, &action, nullptr);
-
-  if (res < 0) {
-    LOG_TOPIC(ERR, arangodb::Logger::FIXME)
-        << "cannot initialize signal handlers for pipe";
-  }
-#endif
 }
 
 SupervisedScheduler::~SupervisedScheduler() {}
@@ -170,6 +152,8 @@ bool SupervisedScheduler::queue(RequestPriority prio, std::function<void()> cons
 
 
 std::string  SupervisedScheduler::infoStatus() const {
+  // TODO: compare with old output format
+  // Does some code rely on that string or is it for humans?
   uint64_t numWorker = _numWorker.load(std::memory_order_relaxed);
   uint64_t queueLength = _jobsSubmitted.load(std::memory_order_relaxed)
     - _jobsDone.load(std::memory_order_relaxed);
@@ -193,6 +177,7 @@ void SupervisedScheduler::addQueueStatistics(velocypack::Builder& b) const {
   uint64_t queueLength = _jobsSubmitted.load(std::memory_order_relaxed)
     - _jobsDone.load(std::memory_order_relaxed);
 
+  // TODO: previous scheduler filled out a lot more fields, relevant?
   b.add("scheduler-threads",
         VPackValue(static_cast<int32_t>(numWorker)));
   b.add("queued", VPackValue(static_cast<int32_t>(queueLength)));
@@ -280,12 +265,20 @@ void SupervisedScheduler::runSupervisor()
 
     if (jobsDone == lastJobsDone && (jobsDone < jobsSubmitted)) {
       jobsStallingTick++;
-    } else {
-      jobsStallingTick = jobsStallingTick == 0 ? 0 : jobsStallingTick - 1;
+    } else if(jobsStallingTick != 0) {
+      jobsStallingTick--;
     }
 
     queueLength = jobsSubmitted - jobsDone;
 
+    // TODO: Currently the conditions to spawn a new thread are not very sophisticated.
+    // A thread is stoped when the queue is empty. In real life this should rarely happen
+    // thus a thread that was started once, never terminates.
+    // A new thread is started when:
+    //  1. jobsStallingTick > 5, which means that 5 times in a row, the supervisor observed that no
+    //      job finished.
+    //  2. The queue length is bigger than 50 and 1.5 times of what it was last time.
+    //      (this is a spike detector)
     bool doStartOneThread =
       (jobsStallingTick > 5) || ((queueLength >= 50) && (1.5 * lastQueueLength < queueLength));
 

diff --git a/arangod/Scheduler/SupervisedScheduler.h b/arangod/Scheduler/SupervisedScheduler.h
@@ -80,18 +80,34 @@ class SupervisedScheduler : public Scheduler {
 
   std::atomic<uint64_t> _jobsSubmitted, _jobsDone;
 
+  // During a queue operation there a two reasons to manually wake up a worker
+  //  1. the queue length is bigger than _wakeupQueueLength and the last submit time
+  //      is bigger than _wakeupTime_ns.
+  //  2. the last submit time is bigger than _definitiveWakeupTime_ns.
+  //
+  // The last submit time is a thread local variable that stores the time of the last
+  // queue operation.
   std::atomic<uint64_t> _wakeupQueueLength;                        // q1
   std::atomic<uint64_t> _wakeupTime_ns, _definitiveWakeupTime_ns;  // t3, t4
 
 
+  // each worker thread has a state block which contains configuration values.
+  // _queueRetryCount is the number of spins this particular thread should perform
+  // before going to sleep.
+  // _sleepTimeout_ms is the amount of ms the thread should sleep before waking
+  // up again. Note that each worker wakes up constantly, even if there is no work.
+  //
+  // All those values are maintained by the supervisor thread.
+  // Currently they are set once and for all the same, however a future
+  // implementation my alter those values for each thread individually.
   struct WorkerState {
     uint64_t _queueRetryCount;     // t1
     uint64_t _sleepTimeout_ms;    // t2
     std::atomic<bool> _stop;
     std::unique_ptr<SupervisedSchedulerWorkerThread> _thread;
     char padding[40];
 
-    // initialise with harmless defaults: spin once, sleep forever
+    // initialize with harmless defaults: spin once, sleep forever
     WorkerState(SupervisedScheduler &scheduler);
     WorkerState(WorkerState &&that);