Merge lp:~michihenning/unity-scopes-api/fix-threadpool-race into lp:unity-scopes-api

[Michi Henning]

Fixed race condition that could cause a segfault if a thread pool was created and immediately destroyed again.
This gets rid of the occasional failures we've been seeing in the Runtime and RuntimeImpl tests.

Fixed race condition that could cause a segfault if a thread pool was created and immediately destroyed again.
This gets rid of the occasional failures we've been seeing in the Runtime and RuntimeImpl tests.

[Michi Henning (michihenning)]

Could someone review this please?

[lp:~michihenning/unity-scopes-api/fix-threadpool-race]

45. By Michi Henning on 2013-11-11

Merged trunk.

46. By Michi Henning on 2013-11-12

Made changes to work with latest trunk.

Added interlock to ThreadPool to wait until all the worker threads have been created, to prevent a worker thread
from calling into an already-destroyed ThreadSafeQueue.

[Paweł Stołowski (stolowski)]

Looks good and I don't see any random test failures anymore. +1

review: Approve

=== modified file 'include/scopes/internal/ThreadPool.h'
--- include/scopes/internal/ThreadPool.h	2013-11-01 12:44:48 +0000
+++ include/scopes/internal/ThreadPool.h	2013-11-12 00:50:34 +0000
@@ -24,7 +24,6 @@
 #include <unity/UnityExceptions.h>
 
 #include <future>
-#include <thread>
 
 namespace unity
 {
@@ -58,8 +57,8 @@
     std::unique_ptr<TaskQueue> queue_;
     std::vector<std::thread> threads_;
     std::mutex mutex_;
-    std::condition_variable cond_;
-    bool done_;
+    std::promise<void> threads_ready_;
+    int num_threads_;
 };
 
 template<typename F>

=== modified file 'include/scopes/internal/ThreadSafeQueue.h'
--- include/scopes/internal/ThreadSafeQueue.h	2013-09-23 06:13:06 +0000
+++ include/scopes/internal/ThreadSafeQueue.h	2013-11-12 00:50:34 +0000
@@ -21,11 +21,12 @@
 
 #include <unity/util/NonCopyable.h>
 
-#include <cassert>
+#include <atomic>
 #include <condition_variable>
 #include <mutex>
 #include <queue>
 #include <stdexcept>
+#include <thread>
 
 namespace unity
 {
@@ -51,6 +52,7 @@
     ThreadSafeQueue();
     ~ThreadSafeQueue() noexcept;
 
+    void destroy() noexcept;
     void push(T const& item);
     void push(T&& item);
     T wait_and_pop();
@@ -63,20 +65,38 @@
     mutable std::mutex mutex_;
     std::condition_variable cond_;
     bool done_;
+    std::atomic<int> num_waiters_;
 };
 
 template<typename T>
 ThreadSafeQueue<T>::ThreadSafeQueue() :
-    done_(false)
+    done_(false),
+    num_waiters_(0)
 {
 }
 
 template<typename T>
 ThreadSafeQueue<T>::~ThreadSafeQueue() noexcept
 {
+    destroy();
+
+    // Don't destroy the object while there are still threads in wait_and_pop(), otherwise
+    // a thread that wakes up in wait_and_pop() will try to re-lock the already-destroyed
+    // mutex.
+    while (num_waiters_.load() > 0)
+        std::this_thread::yield();  // LCOV_EXCL_LINE (impossible to reliably hit with a test)
+}
+
+template<typename T>
+void ThreadSafeQueue<T>::destroy() noexcept
+{
     std::lock_guard<std::mutex> lock(mutex_);
+    if (done_)
+    {
+        return;
+    }
     done_ = true;
-    cond_.notify_all();
+    cond_.notify_all(); // Wake up anyone asleep in wait_and_pop()
 }
 
 template<typename T>
@@ -99,13 +119,17 @@
 T ThreadSafeQueue<T>::wait_and_pop()
 {
     std::unique_lock<std::mutex> lock(mutex_);
+    ++num_waiters_;
     cond_.wait(lock, [this] { return done_ || queue_.size() != 0; });
     if (done_)
     {
+        lock.unlock();
+        --num_waiters_;
         throw std::runtime_error("ThreadSafeQueue: queue destroyed while thread was blocked in wait_and_pop()");
     }
     T item = std::move(queue_.front());
     queue_.pop();
+    --num_waiters_;
     return item;
 }
 

=== modified file 'src/internal/ThreadPool.cpp'
--- src/internal/ThreadPool.cpp	2013-11-01 12:44:48 +0000
+++ src/internal/ThreadPool.cpp	2013-11-12 00:50:34 +0000
@@ -20,6 +20,8 @@
 
 #include <unity/UnityExceptions.h>
 
+#include <cassert>
+
 using namespace std;
 
 namespace unity
@@ -35,6 +37,7 @@
 {
 
 ThreadPool::ThreadPool(int size)
+    : num_threads_(size)
 {
     if (size < 1)
     {
@@ -45,22 +48,29 @@
 
     try
     {
+        {
+            lock_guard<mutex> lock(mutex_);
+            threads_ready_ = std::promise<void>();
+        }
         for (int i = 0; i < size; ++i)
         {
             threads_.push_back(std::thread(&ThreadPool::run, this));
         }
+        auto future = threads_ready_.get_future();
+        future.wait();
+        future.get();
     }
     catch (std::exception const&)   // LCOV_EXCL_LINE
     {
-        throw unity::ResourceException("ThreadPool(): exception during pool creation");  // LCOV_EXCL_LINE
+        throw ResourceException("ThreadPool(): exception during pool creation");  // LCOV_EXCL_LINE
     }
 }
 
 ThreadPool::~ThreadPool() noexcept
 {
+    queue_->destroy();
     try
     {
-        queue_.reset(nullptr);
         for (size_t i = 0; i < threads_.size(); ++i)
         {
             threads_[i].join();
@@ -74,14 +84,21 @@
 
 void ThreadPool::run()
 {
+    TaskQueue::value_type task;
     for (;;)
     {
-        TaskQueue::value_type task;
         try
         {
+            {
+                lock_guard<mutex> lock(mutex_);
+                if (--num_threads_ == 0)
+                {
+                    threads_ready_.set_value();
+                }
+            }
             task = queue_->wait_and_pop();
         }
-        catch (...)
+        catch (runtime_error const&)
         {
             return; // wait_and_pop() throws if the queue is destroyed while threads are blocked on it.
         }

=== modified file 'test/gtest/unity/api/scopes/internal/CMakeLists.txt'
--- test/gtest/unity/api/scopes/internal/CMakeLists.txt	2013-10-28 07:06:08 +0000
+++ test/gtest/unity/api/scopes/internal/CMakeLists.txt	2013-11-12 00:50:34 +0000
@@ -8,6 +8,7 @@
 add_subdirectory(RuntimeConfig)
 add_subdirectory(RuntimeImpl)
 add_subdirectory(ScopeLoader)
+add_subdirectory(ThreadPool)
 add_subdirectory(ThreadSafeQueue)
 add_subdirectory(UniqueID)
 add_subdirectory(zmq_middleware)

=== added directory 'test/gtest/unity/api/scopes/internal/ThreadPool'
=== added file 'test/gtest/unity/api/scopes/internal/ThreadPool/CMakeLists.txt'
--- test/gtest/unity/api/scopes/internal/ThreadPool/CMakeLists.txt	1970-01-01 00:00:00 +0000
+++ test/gtest/unity/api/scopes/internal/ThreadPool/CMakeLists.txt	2013-11-12 00:50:34 +0000
@@ -0,0 +1,4 @@
+add_executable(ThreadPool_test ThreadPool_test.cpp)
+target_link_libraries(ThreadPool_test ${TESTLIBS})
+
+add_test(ThreadPool ThreadPool_test)

=== added file 'test/gtest/unity/api/scopes/internal/ThreadPool/ThreadPool_test.cpp'
--- test/gtest/unity/api/scopes/internal/ThreadPool/ThreadPool_test.cpp	1970-01-01 00:00:00 +0000
+++ test/gtest/unity/api/scopes/internal/ThreadPool/ThreadPool_test.cpp	2013-11-12 00:50:34 +0000
@@ -0,0 +1,106 @@
+/*
+ * Copyright (C) 2013 Canonical Ltd
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 3 as
+ * published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * Authored by: Michi Henning <[e-mail]>
+ */
+
+#include <scopes/internal/ThreadPool.h>
+
+#include <gtest/gtest.h>
+
+using namespace std;
+using namespace unity::api::scopes::internal;
+
+TEST(ThreadPool, basic)
+{
+    // Creation and destruction in quick succession
+    {
+        ThreadPool p(1);
+    }
+    {
+        ThreadPool p(5);
+    }
+    {
+        ThreadPool p(20);
+    }
+}
+
+atomic_int count;
+
+void f()
+{
+    ++count;
+    this_thread::sleep_for(chrono::milliseconds(200));
+}
+
+void g()
+{
+    ++count;
+}
+
+TEST(ThreadPool, submit)
+{
+    {
+        ThreadPool p(1);
+        p.submit(f);
+        p.submit(f);
+    }
+
+    {
+        ThreadPool p(5);
+        p.submit(f);
+        p.submit(f);
+    }
+
+    {
+        count = 0;
+        ThreadPool p(20);
+        p.submit(f);
+        p.submit(f);
+        this_thread::sleep_for(chrono::milliseconds(300));
+        EXPECT_EQ(2, count);
+    }
+
+    {
+        count = 0;
+        ThreadPool p(20);
+        p.submit(g);
+        p.submit(g);
+        p.submit(g);
+        this_thread::sleep_for(chrono::milliseconds(300));
+        EXPECT_EQ(3, count);
+    }
+}
+
+TEST(ThreadPool, exception)
+{
+    try
+    {
+        ThreadPool p(0);
+    }
+    catch (unity::InvalidArgumentException const& e)
+    {
+        EXPECT_EQ("unity::InvalidArgumentException: ThreadPool(): invalid pool size: 0", e.to_string());
+    }
+
+    try
+    {
+        ThreadPool p(-1);
+    }
+    catch (unity::InvalidArgumentException const& e)
+    {
+        EXPECT_EQ("unity::InvalidArgumentException: ThreadPool(): invalid pool size: -1", e.to_string());
+    }
+}

=== modified file 'test/gtest/unity/api/scopes/internal/ThreadSafeQueue/ThreadSafeQueue_test.cpp'
--- test/gtest/unity/api/scopes/internal/ThreadSafeQueue/ThreadSafeQueue_test.cpp	2013-11-01 12:44:48 +0000
+++ test/gtest/unity/api/scopes/internal/ThreadSafeQueue/ThreadSafeQueue_test.cpp	2013-11-12 00:50:34 +0000
@@ -77,13 +77,49 @@
     unique_ptr<ThreadSafeQueue<string>> q(new ThreadSafeQueue<string>);
     q->push("fred");
     auto f = waiter_ready.get_future();
-    auto t = thread(&waiter_thread, q.get());
+    auto t = thread(waiter_thread, q.get());
     f.wait();
     this_thread::sleep_for(chrono::milliseconds(50));   // Make sure child thread has time to call wait_and_pop()
     q.reset();
     t.join();
 }
 
+atomic_int count;
+
+void int_reader_thread(ThreadSafeQueue<int>* q)
+{
+    try
+    {
+        q->wait_and_pop();
+        FAIL();
+    }
+    catch (std::runtime_error const&)
+    {
+        ++count;
+    }
+}
+
+TEST(ThreadSafeQueue, wait_for_threads)
+{
+    ThreadSafeQueue<int> q;
+    count = 0;
+    vector<thread> threads;
+    for (auto i = 0; i < 20; ++i)
+    {
+        threads.push_back(thread(int_reader_thread, &q));
+    }
+    this_thread::sleep_for(chrono::milliseconds(300));
+
+    // Destroy the queue while multiple threads are sleeping in wait_and_pop().
+    q.destroy();
+
+    for (auto& t : threads)
+    {
+        t.join();
+    }
+    EXPECT_EQ(20, count);
+}
+
 class MoveOnly
 {
 public: