Merge lp:~albaguirre/unity-system-compositor/add-worker-thread into lp:unity-system-compositor

[Alberto Aguirre]

Add worker thread to handle incoming DBus requests.

The worker thread can optionally coalesce multiple queued tasks (LP: #1369790)
DBusScreen uses worker thread to serialize requests that were previously dispatched without any particular order (LP: #1368786)

Add worker thread to handle incoming DBus requests.

The worker thread can optionally coalesce multiple queued tasks (LP: #1369790)
DBusScreen uses worker thread to serialize requests that were previously dispatched without any particular order (LP: #1368786)

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:174
http://jenkins.qa.ubuntu.com/job/unity-system-compositor-ci/152/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-system-compositor-utopic-amd64-ci/48
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-system-compositor-utopic-armhf-ci/48

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/unity-system-compositor-ci/152/rebuild

review: Approve (continuous-integration)

[lp:~albaguirre/unity-system-compositor/add-worker-thread]

175. By Alberto Aguirre on 2014-09-16

Small cleanup

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:175
http://jenkins.qa.ubuntu.com/job/unity-system-compositor-ci/153/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-system-compositor-utopic-amd64-ci/49
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-system-compositor-utopic-armhf-ci/49

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/unity-system-compositor-ci/153/rebuild

review: Approve (continuous-integration)

[Andreas Pokorny (andreas-pokorny)]

lgtm

review: Approve

=== modified file 'src/CMakeLists.txt'
--- src/CMakeLists.txt	2014-08-15 02:16:33 +0000
+++ src/CMakeLists.txt	2014-09-16 15:04:17 +0000
@@ -26,6 +26,7 @@
   session_switcher.cpp
   surface_coordinator.cpp
   system_compositor.cpp
+  worker_thread.cpp
 )
 
 qt5_generate_dbus_interface(dbus_screen.h com.canonical.Unity.Screen.xml)
 
=== modified file 'src/dbus_screen.cpp'
--- src/dbus_screen.cpp	2014-08-14 16:06:33 +0000
+++ src/dbus_screen.cpp	2014-09-16 15:04:17 +0000
@@ -18,6 +18,7 @@
 #include "dbus_screen_adaptor.h"
 #include "dbus_screen_observer.h"
 #include "power_state_change_reason.h"
+#include "worker_thread.h"
 
 #include <atomic>
 #include <memory>
@@ -29,6 +30,8 @@
 #include <QDBusServiceWatcher>
 #include <QDebug>
 
+namespace
+{
 bool is_valid_reason(int raw_reason)
 {
     auto reason = static_cast<PowerStateChangeReason>(raw_reason);
@@ -43,12 +46,19 @@
     return false;
 }
 
+enum DBusHandlerTaskId
+{
+    set_power_mode
+};
+}
+
 
 DBusScreen::DBusScreen(DBusScreenObserver& observer, QObject *parent)
     : QObject(parent),
       dbus_adaptor{new DBusScreenAdaptor(this)},
       service_watcher{new QDBusServiceWatcher()},
-      observer{&observer}
+      observer{&observer},
+      worker_thread{new usc::WorkerThread("USC/DBusHandler")}
 {
     QDBusConnection bus = QDBusConnection::systemBus();
     bus.registerObject("/com/canonical/Unity/Screen", this);
@@ -90,9 +100,9 @@
     }
 
     //This call may block - avoid blocking this dbus handling thread
-    std::thread{[this, newPowerMode, reason]{
+    worker_thread->queue_task([this, newPowerMode, reason]{
         observer->set_screen_power_mode(newPowerMode, static_cast<PowerStateChangeReason>(reason));
-    }}.detach();
+    }, DBusHandlerTaskId::set_power_mode);
 
     return true;
 }
@@ -119,43 +129,62 @@
     static std::atomic<uint32_t> request_id{0};
 
     int id = request_id.fetch_add(1);
-
-    //This call may block - avoid blocking this dbus handling thread
-    std::thread{[this]{observer->keep_display_on(true);}}.detach();
-
     auto const& caller = message().service();
-    auto& caller_requests = display_requests[caller.toStdString()];
-
-    if (caller_requests.size() == 0)
-        service_watcher->addWatchedService(caller);
-
-    caller_requests.insert(id);
-    std::cerr << "keepDisplayOn request id:" << id;
-    std::cerr << " requested by \"" << caller.toStdString() << "\"" << std::endl;
+
+    worker_thread->queue_task([this, id, caller]{
+        std::lock_guard<decltype(guard)> lock{guard};
+
+        //Check that the owner of the request is still valid
+        auto system_bus_if = QDBusConnection::systemBus().interface();
+        QDBusReply<QString> reply = system_bus_if->serviceOwner(caller);
+        if (!reply.isValid())
+            return;
+
+        auto& caller_requests = display_requests[caller.toStdString()];
+
+        if (caller_requests.size() == 0)
+            service_watcher->addWatchedService(caller);
+
+        caller_requests.insert(id);
+
+        //This call may block so it needs to be executed outside the thread
+        //that received the dbus call
+        observer->keep_display_on(true);
+
+        std::cout << "keepDisplayOn request id:" << id;
+        std::cout << " requested by \"" << caller.toStdString() << "\"" << std::endl;
+    });
     return id;
 }
 
 void DBusScreen::removeDisplayOnRequest(int cookie)
 {
+    std::lock_guard<decltype(guard)> lock{guard};
     auto const& requestor = message().service();
 
     auto it = display_requests.find(requestor.toStdString());
     if (it == display_requests.end())
         return;
 
-    std::cerr << "removeDisplayOnRequest id:" << cookie;
-    std::cerr << " requested by \"" << requestor.toStdString() << "\"" << std::endl;
+    std::cout << "removeDisplayOnRequest id:" << cookie;
+    std::cout << " requested by \"" << requestor.toStdString() << "\"" << std::endl;
 
     auto caller_requests = it->second;
     caller_requests.erase(cookie);
     if (caller_requests.size() == 0)
-        remove_display_on_requestor(requestor);
+        remove_requestor(requestor, lock);
 }
 
 void DBusScreen::remove_display_on_requestor(QString const& requestor)
 {
-    std::cerr << "remove_display_on_requestor \"" << requestor.toStdString() << "\"";
-    std::cerr << std::endl;
+    std::lock_guard<decltype(guard)> lock{guard};
+    remove_requestor(requestor, lock);
+}
+
+void DBusScreen::remove_requestor(QString const& requestor, std::lock_guard<std::mutex> const&)
+{
+    std::cout << "remove_display_on_requestor \"" << requestor.toStdString() << "\"";
+    std::cout << std::endl;
 
     display_requests.erase(requestor.toStdString());
     service_watcher->removeWatchedService(requestor);
 
=== modified file 'src/dbus_screen.h'
--- src/dbus_screen.h	2014-08-14 16:06:33 +0000
+++ src/dbus_screen.h	2014-09-16 15:04:17 +0000
@@ -20,6 +20,7 @@
 #include <mir_toolkit/common.h>
 
 #include <memory>
+#include <mutex>
 #include <unordered_map>
 #include <unordered_set>
 
@@ -27,6 +28,8 @@
 #include <QtCore>
 #include <QDBusContext>
 
+namespace usc {class WorkerThread;}
+
 class DBusScreenAdaptor;
 class DBusScreenObserver;
 class QDBusInterface;
@@ -59,10 +62,14 @@
     void remove_display_on_requestor(QString const& requestor);
 
 private:
+    void remove_requestor(QString const& requestor, std::lock_guard<std::mutex> const& lock);
+
+    std::mutex guard;
     std::unique_ptr<DBusScreenAdaptor> dbus_adaptor;
     std::unique_ptr<QDBusServiceWatcher> service_watcher;
     std::unordered_map<std::string, std::unordered_set<int>> display_requests;
     DBusScreenObserver* const observer;
+    std::unique_ptr<usc::WorkerThread> worker_thread;
 };
 
 #endif /* DBUS_SCREEN_H_ */
 
=== added file 'src/worker_thread.cpp'
--- src/worker_thread.cpp	1970-01-01 00:00:00 +0000
+++ src/worker_thread.cpp	2014-09-16 15:04:17 +0000
@@ -0,0 +1,141 @@
+/*
+ * 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 "worker_thread.h"
+#include <mir/run_mir.h>
+
+#include <algorithm>
+#include <condition_variable>
+#include <deque>
+#include <map>
+
+#ifndef _GNU_SOURCE
+#define _GNU_SOURCE
+#endif
+#include <pthread.h>
+
+namespace usc
+{
+
+class Worker
+{
+public:
+    Worker(std::string name)
+        : name{name.substr(0, 15)}, exiting{false}
+    {
+    }
+
+    ~Worker()
+    {
+        exit();
+    }
+
+    void operator()() noexcept
+    try
+    {
+       pthread_setname_np(pthread_self(), name.c_str());
+
+       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::function<void()> task)
+    {
+        std::lock_guard<std::mutex> lock{state_mutex};
+        tasks.push_back(std::move(task));
+        task_available_cv.notify_one();
+    }
+
+    void queue_task(std::function<void()> task, int id)
+    {
+        std::lock_guard<std::mutex> lock{state_mutex};
+        coalesced_tasks[id] = task;
+        tasks.push_back([this, id]{ run_coalesced_task(id);});
+        task_available_cv.notify_one();
+    }
+
+    void run_coalesced_task(int id)
+    {
+        std::unique_lock<std::mutex> lock{state_mutex};
+        auto it = coalesced_tasks.find(id);
+        if (it != coalesced_tasks.end())
+        {
+            auto task = it->second;
+            coalesced_tasks.erase(it);
+            lock.unlock();
+            task();
+        }
+    }
+
+    void exit()
+    {
+        std::lock_guard<std::mutex> lock{state_mutex};
+        exiting = true;
+        task_available_cv.notify_one();
+    }
+
+private:
+    std::mutex mutable state_mutex;
+    std::string name;
+    bool exiting;
+    std::map<int, std::function<void()>> coalesced_tasks;
+    std::deque<std::function<void()>> tasks;
+    std::condition_variable task_available_cv;
+};
+
+}
+
+usc::WorkerThread::WorkerThread(std::string name)
+    : worker{new Worker(name)},
+      thread{std::ref(*worker)}
+{
+}
+
+usc::WorkerThread::~WorkerThread()
+{
+    worker->exit();
+    if (thread.joinable())
+        thread.join();
+}
+
+void usc::WorkerThread::queue_task(std::function<void()> task)
+{
+    worker->queue_task(std::move(task));
+}
+
+void usc::WorkerThread::queue_task(std::function<void()> task, int id)
+{
+    worker->queue_task(std::move(task), id);
+}
 
=== added file 'src/worker_thread.h'
--- src/worker_thread.h	1970-01-01 00:00:00 +0000
+++ src/worker_thread.h	2014-09-16 15:04:17 +0000
@@ -0,0 +1,55 @@
+/*
+ * 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]>
+ */
+
+#ifndef USC_WORKER_THREAD_H_
+#define USC_WORKER_THREAD_H_
+
+#include <memory>
+#include <string>
+#include <thread>
+
+namespace usc
+{
+class Worker;
+class WorkerThread
+{
+public:
+    WorkerThread(std::string name);
+    ~WorkerThread();
+
+    /**
+     * Queues a task to be executed on the worker thread
+     */
+    void queue_task(std::function<void()> task);
+
+    /**
+     * Queues and coalesces tasks that share the same id
+     */
+    void queue_task(std::function<void()> task, int id);
+
+private:
+    WorkerThread(WorkerThread const&) = delete;
+    WorkerThread& operator=(WorkerThread const&) = delete;
+
+    std::unique_ptr<Worker> worker;
+    std::thread thread;
+};
+
+}
+
+#endif