Issue #3595 - Add an actual scheduler; use joined threads for file system

src/Scheduler.cpp

Index: src/Scheduler.cpp
===================================================================
--- a/src/Scheduler.cpp
+++ b/src/Scheduler.cpp
@@ -27,7 +27,7 @@
   }
 }
 
-void Scheduler::StartImmediatelyInSingleUseThread(std::function<void()> task)
+void StartImmediatelyInSingleUseDetachedThread(std::function<void()> task)
 {
   if (!task)
   {
@@ -36,3 +36,82 @@
   auto th = std::thread(SingleUseThreadMain, task);
   th.detach();
 }
+
+OperationRunner::OperationRunner()
+  : isRunning(true)
+{
+  th = std::thread([this]() { ThreadMain(); });
+  /*
+   * Note that we don't need to wait for our thread to start running
+   *   before we consider the object fully constructed.
+   * Because of how the `wait()` statement in `ThreadMain` is written,
+   *   any operation arriving before the thread begins will be
+   *   processed immediately, without blocking on a notification.
+   */

[sergei, 2017/01/20 13:08:13]

That comment is good however I don't think we need it because ThreadMain will be
written that way or similar anyway. The worker thread can be paused for instance
between lines 68 and 69, and we add items into queue with notifying the
condition variable. Afterwards the worker thread continues the execution, of
course it should process already pending items as soon as possible. So it's not
related to construction or starting of a thread, it's just how it should be.

[Eric, 2017/03/30 17:16:58]

Behavior is always clearer *after* you read the comment.

It's also not a trivial comment. There could be other ways of implementing
`ThreadMain` and you have to avoid race conditions upon construction, where this
class gets a queued task but no notification, which would cause the object to
freeze up and never run anything.

[sergei, 2017/04/03 15:35:32]

I'm talking about implementation without race conditions and other bugs.

+}
+
+OperationRunner::~OperationRunner()
+{
+  // Shut down the operation queue
+  Run([this]() { isRunning = false; });
+  th.join();
+}
+
+void OperationRunner::Run(std::function<void()> f)
+{
+  UniqueLockType ul(m);
+  queue.push(f);
+  cv.notify_one();
+}
+
+void OperationRunner::ThreadMain()
+{
+  while (true)
+  {
+    std::function<void()> op;
+    {
+      UniqueLockType ul(m);
+      // Checking the flag here avoids needing a second lock object
+      if (!isRunning)

[sergei, 2017/01/20 13:08:13]

Why not to put isRunning into while condition? we set and read isRunning only
from worker thread, we don't need to have a lock to access it.

[Eric, 2017/03/30 17:16:58]

Because it's not correct. It opens up an opportunity for a race condition.

[sergei, 2017/04/03 15:35:31]

Could you please explain when that race condition occurs?

+      {
+        break;
+      }
+      cv.wait(ul, [this]() -> bool { return !queue.empty(); });
+      op = queue.front();
+      queue.pop();
+    }
+    // Assert `m` is unlocked

[sergei, 2017/01/20 13:08:13]

I would remove that comment.

[Eric, 2017/03/30 17:16:58]

Nope; it stays. Tracking when the mutex is locked and when it's unlocked is
critical to auditing this code. It wasn't there when I first drafted the code.

+    try
+    {
+      if (op)

[sergei, 2017/04/03 15:35:31]

Why not to prevent putting of empty functions in the queue?

+      {
+        op();
+      }
+    }
+    catch (std::exception& e)

[sergei, 2017/04/03 15:35:32]

it generates a warning about unused local variable.

+    {
+      // suppress
+    }
+  }
+}
+
+SingleUseWorker::SingleUseWorker()
+{}
+
+SingleUseWorker::~SingleUseWorker()
+{
+  if (th.joinable())

[sergei, 2017/01/20 13:08:13]

This implementation with if (th.joinable()) should be in Join method.

[Eric, 2017/03/30 17:16:58]

Yes, you need to be able to join the thread either explicitly or in the
destructor. I fixed this in a later revision by moving this code to Join() and
calling Join() here in the destructor.

+  {
+    th.join();
+  }
+}
+
+void SingleUseWorker::Run(std::function<void()> f)
+{
+  th = std::move(std::thread(f));

[sergei, 2017/01/20 13:08:13]

Do we really need to explicitly say std::move here?

[Eric, 2017/03/30 17:16:58]

It avoids a second copy, since there's already one in the call by value. 

I don't think there's a semantic rule in the C++ specification that makes the
`move` call redundant, but there might be.

[sergei, 2017/04/03 15:35:32]

Copy of what? JIC, std::thread cannot be copied.

+}
+
+void SingleUseWorker::Join()
+{
+  th.join();
+}