Merge lp:~albaguirre/mir/recycle-compositor-threads-take2 into lp:mir

[Alberto Aguirre]

Recycle compositor threads by using a thread pool (LP: #1362841)

Recycle compositor threads by using a thread pool

Adds a basic thread pool placed under server instead of common as I wanted to use mir::terminate_with_current_exception() and currently in use by server code only.

[PS Jenkins bot (ps-jenkins)]

FAILED: Continuous integration, rev:1895
http://jenkins.qa.ubuntu.com/job/mir-team-mir-development-branch-ci/2637/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-android-utopic-i386-build/1636
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-clang-utopic-amd64-build/1642
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-utopic-touch/1618/console
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-team-mir-development-branch-utopic-amd64-ci/1159
        deb: http://jenkins.qa.ubuntu.com/job/mir-team-mir-development-branch-utopic-amd64-ci/1159/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-builder-utopic-armhf/555
        deb: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-builder-utopic-armhf/555/artifact/work/output/*zip*/output.zip
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-runner-mako/2675/console
    SUCCESS: http://s-jenkins.ubuntu-ci:8080/job/touch-flash-device/12809

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/mir-team-mir-development-branch-ci/2637/rebuild

review: Needs Fixing (continuous-integration)

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:1896
http://jenkins.qa.ubuntu.com/job/mir-team-mir-development-branch-ci/2638/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-android-utopic-i386-build/1638
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-clang-utopic-amd64-build/1644
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-utopic-touch/1620
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-team-mir-development-branch-utopic-amd64-ci/1160
        deb: http://jenkins.qa.ubuntu.com/job/mir-team-mir-development-branch-utopic-amd64-ci/1160/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-builder-utopic-armhf/557
        deb: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-builder-utopic-armhf/557/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-runner-mako/2677
    SUCCESS: http://s-jenkins.ubuntu-ci:8080/job/touch-flash-device/12813

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/mir-team-mir-development-branch-ci/2638/rebuild

review: Approve (continuous-integration)

[Alexandros Frantzis (afrantzis)]

Looks good.

review: Approve

[PS Jenkins bot (ps-jenkins)]

FAILED: Autolanding.
More details in the following jenkins job:
http://jenkins.qa.ubuntu.com/job/mir-team-mir-development-branch-autolanding/1046/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-android-utopic-i386-build/1658
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-clang-utopic-amd64-build/1664
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-utopic-touch/1640
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-team-mir-development-branch-utopic-amd64-autolanding/430/console
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-builder-utopic-armhf/577
        deb: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-builder-utopic-armhf/577/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-runner-mako/2696
    SUCCESS: http://s-jenkins.ubuntu-ci:8080/job/touch-flash-device/12883

review: Needs Fixing (continuous-integration)

[Alberto Aguirre (albaguirre)]

^--DemoPrivateProtobuf.client_calls_server had an exception....seems unrelated. Re-approving...

[PS Jenkins bot (ps-jenkins)]

review: Approve (continuous-integration)

=== added directory 'include.private/server/mir/thread'
=== added file 'include.private/server/mir/thread/basic_thread_pool.h'
--- include.private/server/mir/thread/basic_thread_pool.h	1970-01-01 00:00:00 +0000
+++ include.private/server/mir/thread/basic_thread_pool.h	2014-09-06 14:29:21 +0000
@@ -0,0 +1,63 @@
+/*
+ * Copyright © 2014 Canonical Ltd.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser 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 Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * Authored by: Alberto Aguirre <[e-mail]>
+ */
+
+#ifndef MIR_THREAD_BASIC_THREAD_POOL_H_
+#define MIR_THREAD_BASIC_THREAD_POOL_H_
+
+#include <functional>
+#include <future>
+#include <vector>
+#include <memory>
+#include <mutex>
+
+namespace mir
+{
+namespace thread
+{
+
+class WorkerThread;
+class BasicThreadPool
+{
+public:
+    BasicThreadPool(int min_threads);
+    ~BasicThreadPool();
+
+    std::future<void> run(std::function<void()> const& task);
+
+    typedef void const* TaskId;
+    std::future<void> run(std::function<void()> const& task, TaskId id);
+
+    void shrink();
+
+private:
+    BasicThreadPool(BasicThreadPool const&) = delete;
+    BasicThreadPool& operator=(BasicThreadPool const&) = delete;
+
+    std::future<void> run(WorkerThread* t, std::function<void()> const& task, TaskId id);
+    WorkerThread *find_thread_by(TaskId id);
+    WorkerThread *find_idle_thread();
+
+    std::mutex mutex;
+    int const min_threads;
+    std::vector<std::unique_ptr<WorkerThread>> threads;
+};
+
+}
+}
+
+#endif

=== modified file 'src/server/CMakeLists.txt'
--- src/server/CMakeLists.txt	2014-09-05 03:37:19 +0000
+++ src/server/CMakeLists.txt	2014-09-06 14:29:21 +0000
@@ -13,6 +13,7 @@
 add_subdirectory(scene/)
 add_subdirectory(frontend/)
 add_subdirectory(shell/)
+add_subdirectory(thread/)
 
 set(PREFIX "${CMAKE_INSTALL_PREFIX}")
 set(EXEC_PREFIX "${CMAKE_INSTALL_PREFIX}")
@@ -48,6 +49,7 @@
   $<TARGET_OBJECTS:mirlogger>
   $<TARGET_OBJECTS:mirnestedgraphics>
   $<TARGET_OBJECTS:miroffscreengraphics>
+  $<TARGET_OBJECTS:mirthread>
 )
 
 set(MIR_SERVER_REFERENCES

=== modified file 'src/server/compositor/multi_threaded_compositor.cpp'
--- src/server/compositor/multi_threaded_compositor.cpp	2014-09-05 03:37:19 +0000
+++ src/server/compositor/multi_threaded_compositor.cpp	2014-09-06 14:29:21 +0000
@@ -195,7 +195,8 @@
       display_buffer_compositor_factory{db_compositor_factory},
       report{compositor_report},
       state{CompositorState::stopped},
-      compose_on_start{compose_on_start}
+      compose_on_start{compose_on_start},
+      thread_pool{1}
 {
     observer = std::make_shared<ms::LegacySceneChangeNotification>(
     [this]()
@@ -289,10 +290,12 @@
         auto thread_functor_raw = new mc::CompositingFunctor{display_buffer_compositor_factory, buffer, report};
         auto thread_functor = std::unique_ptr<mc::CompositingFunctor>(thread_functor_raw);
 
-        threads.push_back(std::thread{std::ref(*thread_functor)});
+        futures.push_back(thread_pool.run(std::ref(*thread_functor), &buffer));
         thread_functors.push_back(std::move(thread_functor));
     });
 
+    thread_pool.shrink();
+
     state = CompositorState::started;
 }
 
@@ -307,11 +310,11 @@
     for (auto& f : thread_functors)
         f->stop();
 
-    for (auto& t : threads)
-        t.join();
+    for (auto& f : futures)
+        f.wait();
 
     thread_functors.clear();
-    threads.clear();
+    futures.clear();
 
     report->stopped();
 

=== modified file 'src/server/compositor/multi_threaded_compositor.h'
--- src/server/compositor/multi_threaded_compositor.h	2014-09-05 03:37:19 +0000
+++ src/server/compositor/multi_threaded_compositor.h	2014-09-06 14:29:21 +0000
@@ -20,11 +20,12 @@
 #define MIR_COMPOSITOR_MULTI_THREADED_COMPOSITOR_H_
 
 #include "mir/compositor/compositor.h"
+#include "mir/thread/basic_thread_pool.h"
 
 #include <mutex>
 #include <memory>
 #include <vector>
-#include <thread>
+#include <future>
 
 namespace mir
 {
@@ -76,15 +77,16 @@
     std::shared_ptr<CompositorReport> const report;
 
     std::vector<std::unique_ptr<CompositingFunctor>> thread_functors;
-    std::vector<std::thread> threads;
+    std::vector<std::future<void>> futures;
 
     std::mutex state_guard;
     CompositorState state;
     bool compose_on_start;
 
     void schedule_compositing(int number_composites);
-    
+
     std::shared_ptr<mir::scene::Observer> observer;
+    mir::thread::BasicThreadPool thread_pool;
 };
 
 }

=== added directory 'src/server/thread'
=== added file 'src/server/thread/CMakeLists.txt'
--- src/server/thread/CMakeLists.txt	1970-01-01 00:00:00 +0000
+++ src/server/thread/CMakeLists.txt	2014-09-06 14:29:21 +0000
@@ -0,0 +1,12 @@
+set(
+  MIR_THREAD_SRCS
+
+  basic_thread_pool.cpp
+)
+
+ADD_LIBRARY(
+  mirthread OBJECT
+
+  ${MIR_THREAD_SRCS}
+)
+

=== added file 'src/server/thread/basic_thread_pool.cpp'
--- src/server/thread/basic_thread_pool.cpp	1970-01-01 00:00:00 +0000
+++ src/server/thread/basic_thread_pool.cpp	2014-09-06 14:29:21 +0000
@@ -0,0 +1,221 @@
+/*
+ * Copyright © 2014 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: Alberto Aguirre <[e-mail]>
+ */
+
+#include "mir/thread/basic_thread_pool.h"
+#include <mir/run_mir.h>
+
+#include <deque>
+#include <algorithm>
+#include <condition_variable>
+
+namespace mt = mir::thread;
+
+namespace
+{
+
+class Worker
+{
+public:
+    Worker() : exiting{false}
+    {
+    }
+
+    ~Worker()
+    {
+        exit();
+    }
+
+    void operator()() noexcept
+    try
+    {
+       std::unique_lock<std::mutex> lock{state_mutex};
+       while (!exiting)
+       {
+           task_available_cv.wait(lock, [&]{ return exiting || !tasks.empty(); });
+
+           if (!exiting)
+           {
+               auto& task = tasks.front();
+               lock.unlock();
+               task();
+               lock.lock();
+               tasks.pop_front();
+           }
+       }
+    }
+    catch(...)
+    {
+        mir::terminate_with_current_exception();
+    }
+
+    void queue_task(std::packaged_task<void()> task)
+    {
+        std::lock_guard<std::mutex> lock{state_mutex};
+        tasks.push_back(std::move(task));
+        task_available_cv.notify_one();
+    }
+
+    void exit()
+    {
+        std::lock_guard<std::mutex> lock{state_mutex};
+        exiting = true;
+        task_available_cv.notify_one();
+    }
+
+    bool is_idle() const
+    {
+        std::lock_guard<std::mutex> lock{state_mutex};
+        return tasks.empty();
+    }
+
+private:
+    std::deque<std::packaged_task<void()>> tasks;
+    bool exiting;
+    std::mutex mutable state_mutex;
+    std::condition_variable task_available_cv;
+};
+
+}
+
+namespace mir
+{
+namespace thread
+{
+class WorkerThread
+{
+public:
+    WorkerThread(mt::BasicThreadPool::TaskId an_id)
+        : thread{std::ref(worker)},
+          id_{an_id}
+    {}
+
+    ~WorkerThread()
+    {
+        worker.exit();
+        if (thread.joinable())
+            thread.join();
+    }
+
+    void queue_task(std::packaged_task<void()> task, mt::BasicThreadPool::TaskId the_id)
+    {
+        worker.queue_task(std::move(task));
+        id_ = the_id;
+    }
+
+    bool is_idle() const
+    {
+        return worker.is_idle();
+    }
+
+    mt::BasicThreadPool::TaskId current_id() const
+    {
+        return id_;
+    }
+
+private:
+    ::Worker worker;
+    std::thread thread;
+    mt::BasicThreadPool::TaskId id_;
+};
+}
+}
+
+mt::BasicThreadPool::BasicThreadPool(int min_threads)
+    : min_threads{min_threads}
+{
+}
+
+mt::BasicThreadPool::~BasicThreadPool() = default;
+
+std::future<void> mt::BasicThreadPool::run(std::function<void()> const& task)
+{
+    std::lock_guard<decltype(mutex)> lock{mutex};
+
+    TaskId const generic_id = nullptr;
+    WorkerThread* no_preferred_thread = nullptr;
+    return run(no_preferred_thread, task, generic_id);
+}
+
+std::future<void> mt::BasicThreadPool::run(std::function<void()> const& task, TaskId id)
+{
+    std::lock_guard<decltype(mutex)> lock{mutex};
+    return run(find_thread_by(id), task, id);
+}
+
+std::future<void> mt::BasicThreadPool::run(WorkerThread* worker_thread,
+                                           std::function<void()> const& task, TaskId id)
+{
+    // No pre-selected thread to execute task, find an idle thread
+    if (worker_thread == nullptr)
+        worker_thread = find_idle_thread();
+
+    if (worker_thread == nullptr)
+    {
+        // No idle threads available so create a new one
+        std::unique_ptr<WorkerThread> new_worker_thread{new WorkerThread{id}};
+        threads.push_back(std::move(new_worker_thread));
+        worker_thread = threads.back().get();
+    }
+
+    std::packaged_task<void()> a_task{task};
+    auto future = a_task.get_future();
+    worker_thread->queue_task(std::move(a_task), id);
+    return future;
+}
+
+
+void mt::BasicThreadPool::shrink()
+{
+    std::lock_guard<decltype(mutex)> lock{mutex};
+
+    int max_threads_to_remove = threads.size() - min_threads;
+    auto it = std::remove_if(threads.begin(), threads.end(),
+        [&max_threads_to_remove](std::unique_ptr<WorkerThread> const& worker_thread)
+        {
+            bool remove = worker_thread->is_idle() && max_threads_to_remove > 0;
+            if (remove)
+                max_threads_to_remove--;
+            return remove;
+        }
+    );
+    threads.erase(it, threads.end());
+}
+
+mt::WorkerThread* mt::BasicThreadPool::find_thread_by(TaskId id)
+{
+    auto it = std::find_if(threads.begin(), threads.end(),
+        [id](std::unique_ptr<WorkerThread> const& worker_thread)
+        {
+            return worker_thread->current_id() == id;
+        }
+    );
+
+    return it == threads.end() ? nullptr : it->get();
+}
+
+mt::WorkerThread *mt::BasicThreadPool::find_idle_thread()
+{
+    auto it = std::find_if(threads.begin(), threads.end(),
+        [](std::unique_ptr<WorkerThread> const& worker_thread)
+        {
+            return worker_thread->is_idle();
+        }
+    );
+
+    return it == threads.end() ? nullptr : it->get();
+}

=== modified file 'tests/unit-tests/CMakeLists.txt'
--- tests/unit-tests/CMakeLists.txt	2014-09-05 03:37:19 +0000
+++ tests/unit-tests/CMakeLists.txt	2014-09-06 14:29:21 +0000
@@ -36,6 +36,7 @@
 add_subdirectory(scene/)
 add_subdirectory(draw/)
 add_subdirectory(examples/)
+add_subdirectory(thread/)
 
 link_directories(${LIBRARY_OUTPUT_PATH})
 

=== added directory 'tests/unit-tests/thread'
=== added file 'tests/unit-tests/thread/CMakeLists.txt'
--- tests/unit-tests/thread/CMakeLists.txt	1970-01-01 00:00:00 +0000
+++ tests/unit-tests/thread/CMakeLists.txt	2014-09-06 14:29:21 +0000
@@ -0,0 +1,5 @@
+list(APPEND UNIT_TEST_SOURCES
+  ${CMAKE_CURRENT_SOURCE_DIR}/test_basic_thread_pool.cpp
+)
+
+set(UNIT_TEST_SOURCES ${UNIT_TEST_SOURCES} PARENT_SCOPE)

=== added file 'tests/unit-tests/thread/test_basic_thread_pool.cpp'
--- tests/unit-tests/thread/test_basic_thread_pool.cpp	1970-01-01 00:00:00 +0000
+++ tests/unit-tests/thread/test_basic_thread_pool.cpp	2014-09-06 14:29:21 +0000
@@ -0,0 +1,221 @@
+/*
+ * Copyright © 2014 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: Alberto Aguirre <[e-mail]>
+ */
+
+#include "mir/thread/basic_thread_pool.h"
+
+#include "mir/thread_name.h"
+#include "mir_test/current_thread_name.h"
+#include "mir_test/signal.h"
+
+#include <memory>
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+namespace mt = mir::test;
+namespace mth = mir::thread;
+
+namespace
+{
+class BasicThreadPool : public testing::Test
+{
+public:
+    BasicThreadPool()
+        : default_task_id{nullptr},
+          default_num_threads{3},
+          expected_name{"test_thread"}
+    {
+    }
+protected:
+    mth::BasicThreadPool::TaskId default_task_id;
+    int default_num_threads;
+    std::string expected_name;
+};
+
+class TestTask
+{
+public:
+    TestTask()
+        : TestTask("")
+    {
+    }
+
+    TestTask(std::string name)
+        : called{false},
+          block{false},
+          name{name}
+    {
+    }
+
+    ~TestTask()
+    {
+        unblock();
+    }
+
+    void operator()() noexcept
+    {
+        called = true;
+        if (!name.empty())
+            mir::set_thread_name(name);
+        else
+            name = mt::current_thread_name();
+
+        if (block)
+            signal.wait();
+    }
+
+    std::string thread_name() const
+    {
+        return name;
+    }
+
+    bool was_called() const
+    {
+        return called;
+    }
+
+    void block_on_execution()
+    {
+        block = true;
+    }
+
+    void unblock()
+    {
+        signal.raise();
+    }
+
+private:
+    bool called;
+    bool block;
+    std::string name;
+    mt::Signal signal;
+};
+}
+
+TEST_F(BasicThreadPool, executes_given_functor)
+{
+    using namespace testing;
+    mth::BasicThreadPool p{default_num_threads};
+
+    TestTask task;
+    auto future = p.run(std::ref(task));
+    future.wait();
+
+    EXPECT_TRUE(task.was_called());
+}
+
+TEST_F(BasicThreadPool, executes_on_preferred_thread)
+{
+    using namespace testing;
+    mth::BasicThreadPool p{default_num_threads};
+
+    TestTask task1{expected_name};
+    task1.block_on_execution();
+    auto future1 = p.run(std::ref(task1), default_task_id);
+
+    // This task should be queued to the thread associated with the given id
+    // even when its busy
+    TestTask task2;
+    auto future2 = p.run(std::ref(task2), default_task_id);
+
+    task1.unblock();
+    future1.wait();
+    future2.wait();
+
+    // Check if the second task executed on the same thread as the first task
+    EXPECT_TRUE(task1.was_called());
+    EXPECT_TRUE(task2.was_called());
+    EXPECT_THAT(task2.thread_name(), Eq(expected_name));
+}
+
+TEST_F(BasicThreadPool, recycles_threads)
+{
+    using namespace testing;
+    mth::BasicThreadPool p{2};
+
+    std::string const thread1_name = "thread1";
+    std::string const thread2_name = "thread2";
+
+    //Create a couple of blocking tasks, so that we force the pool
+    //to assign new threads to each task
+    TestTask task1{thread1_name};
+    task1.block_on_execution();
+    auto future1 = p.run(std::ref(task1));
+
+    TestTask task2{thread2_name};
+    task2.block_on_execution();
+    auto future2 = p.run(std::ref(task2));
+
+    task1.unblock();
+    task2.unblock();
+    future1.wait();
+    future2.wait();
+
+    // Now we should have some idle threads, the next task should
+    // run in any of the two previously created threads
+    TestTask task3;
+    auto future = p.run(std::ref(task3));
+    future.wait();
+
+    EXPECT_TRUE(task1.was_called());
+    EXPECT_TRUE(task2.was_called());
+    EXPECT_TRUE(task3.was_called());
+    EXPECT_THAT(task3.thread_name(), AnyOf(Eq(thread1_name), Eq(thread2_name)));
+}
+
+TEST_F(BasicThreadPool, creates_new_threads)
+{
+    using namespace testing;
+    mth::BasicThreadPool p{1};
+
+    TestTask task1{expected_name};
+    task1.block_on_execution();
+    auto future1 = p.run(std::ref(task1));
+
+    TestTask task2;
+    auto future2 = p.run(std::ref(task2));
+
+    task1.unblock();
+    future1.wait();
+    future2.wait();
+
+    EXPECT_TRUE(task1.was_called());
+    EXPECT_TRUE(task2.was_called());
+    EXPECT_THAT(task2.thread_name(), Ne(expected_name));
+}
+
+TEST_F(BasicThreadPool, can_shrink)
+{
+    using namespace testing;
+    mth::BasicThreadPool p{0};
+
+    TestTask task1{expected_name};
+    auto future = p.run(std::ref(task1));
+    future.wait();
+
+    // This should delete all threads since we specified a minimum of 0 threads for our pool
+    p.shrink();
+
+    TestTask task2;
+    future = p.run(std::ref(task2));
+    future.wait();
+
+    EXPECT_TRUE(task1.was_called());
+    EXPECT_TRUE(task2.was_called());
+    EXPECT_THAT(task2.thread_name(), Ne(expected_name));
+}