Wait for loop to exit in EventLoop destructor. (#198)

Author: gschmottlach-xse

trantor/net/EventLoop.cc

@@ -25,7 +25,9 @@
 #include <thread>
 #include <assert.h>
 #ifdef _WIN32
+#include <windows.h>
 #include <io.h>
+#include <synchapi.h>
 using ssize_t = long long;
 #else
 #include <poll.h>
@@ -101,7 +103,7 @@ EventLoop::EventLoop()
 void EventLoop::resetTimerQueue()
 {
     assertInLoopThread();
-    assert(!looping_);
+    assert(!looping_.load(std::memory_order_acquire));
     timerQueue_->reset();
 }
 #endif
@@ -111,8 +113,29 @@ void EventLoop::resetAfterFork()
 }
 EventLoop::~EventLoop()
 {
+#ifdef _WIN32
+    DWORD delay = 1; /* 1 msec */
+#else
+    struct timespec delay = {0, 1000000}; /* 1 msec */
+#endif
+
     quit();
-    assert(!looping_);
+
+    // Spin waiting for the loop to exit because
+    // this may take some time to complete. We
+    // assume the loop thread will *always* exit.
+    // If this cannot be guaranteed then one option
+    // might be to abort waiting and
+    // assert(!looping_) after some delay;
+    while (looping_.load(std::memory_order_acquire))
+    {
+#ifdef _WIN32
+        Sleep(delay);
+#else
+        nanosleep(&delay, nullptr);
+#endif
+    }
+
     t_loopInThisThread = nullptr;
 #ifdef __linux__
     close(wakeupFd_);
@@ -140,7 +163,7 @@ void EventLoop::removeChannel(Channel *channel)
 }
 void EventLoop::quit()
 {
-    quit_ = true;
+    quit_.store(true, std::memory_order_release);
 
     Func f;
     while (funcsOnQuit_.dequeue(f))
@@ -160,10 +183,10 @@ void EventLoop::loop()
 {
     assert(!looping_);
     assertInLoopThread();
-    looping_ = true;
-    quit_ = false;
+    looping_.store(true, std::memory_order_release);
+    quit_.store(false, std::memory_order_release);
 
-    while (!quit_)
+    while (!quit_.load(std::memory_order_acquire))
     {
         activeChannels_.clear();
 #ifdef __linux__
@@ -187,7 +210,7 @@ void EventLoop::loop()
         // std::cout << "looping" << endl;
         doRunInLoopFuncs();
     }
-    looping_ = false;
+    looping_.store(false, std::memory_order_release);
 }
 void EventLoop::abortNotInLoopThread()
 {
@@ -198,15 +221,15 @@ void EventLoop::abortNotInLoopThread()
 void EventLoop::queueInLoop(const Func &cb)
 {
     funcs_.enqueue(cb);
-    if (!isInLoopThread() || !looping_)
+    if (!isInLoopThread() || !looping_.load(std::memory_order_acquire))
     {
         wakeup();
     }
 }
 void EventLoop::queueInLoop(Func &&cb)
 {
     funcs_.enqueue(std::move(cb));
-    if (!isInLoopThread() || !looping_)
+    if (!isInLoopThread() || !looping_.load(std::memory_order_acquire))
     {
         wakeup();
     }

trantor/net/EventLoop.h

@@ -28,6 +28,7 @@
 #include <functional>
 #include <chrono>
 #include <limits>
+#include <atomic>
 
 namespace trantor
 {
@@ -270,7 +271,8 @@ class TRANTOR_EXPORT EventLoop : NonCopyable
      */
     bool isRunning()
     {
-        return looping_ && (!quit_);
+        return looping_.load(std::memory_order_acquire) &&
+               (!quit_.load(std::memory_order_acquire));
     }
 
     /**
@@ -297,9 +299,9 @@ class TRANTOR_EXPORT EventLoop : NonCopyable
     void abortNotInLoopThread();
     void wakeup();
     void wakeupRead();
-    bool looping_;
+    std::atomic<bool> looping_;
     std::thread::id threadId_;
-    bool quit_;
+    std::atomic<bool> quit_;
     std::unique_ptr<Poller> poller_;
 
     ChannelList activeChannels_;