introduce ThreadTaskRunner and use it in heapprofd's UnwindingWorker
As discussed last week, proposing a wrapper object that owns a UnixTaskRunner & its task thread.
For the review, please focus on the following:
* that the writeup of the thread-checking considerations in unix_task_runner.h is accurate
* that the thread_task_runner implementation itself is sane
* that its use in heapprofd is reasonable
I have no strong opinions on the following, and simply chose one option for the initial review:
* Naming of the new utility - ThreadUnixTaskRunner would more accurately reflect that this isn't
cross-platform?
* Whether it belongs in base::, or should be in profiling::memory:: for now.
We can slot this into consumer_api.cc (which does a very similar handshake) in a followup cl,
so I thought base:: might be a reasonable place.
Note: with this change, heapprofd works in debug builds e2e afaict.
Change-Id: Ia12c89c6963bf0a659bca7994f4fe87aa3bfde58
diff --git a/src/base/BUILD.gn b/src/base/BUILD.gn
index 25f3691..447a9b1 100644
--- a/src/base/BUILD.gn
+++ b/src/base/BUILD.gn
@@ -41,6 +41,7 @@
"event.cc",
"pipe.cc",
"temp_file.cc",
+ "thread_task_runner.cc",
"unix_task_runner.cc",
]
}
@@ -157,6 +158,7 @@
"task_runner_unittest.cc",
"temp_file_unittest.cc",
"thread_checker_unittest.cc",
+ "thread_task_runner_unittest.cc",
"utils_unittest.cc",
]
}
diff --git a/src/base/thread_task_runner.cc b/src/base/thread_task_runner.cc
new file mode 100644
index 0000000..a545e6b
--- /dev/null
+++ b/src/base/thread_task_runner.cc
@@ -0,0 +1,81 @@
+/*
+ * Copyright (C) 2019 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "perfetto/base/thread_task_runner.h"
+
+#include <condition_variable>
+#include <functional>
+#include <mutex>
+#include <thread>
+
+#include "perfetto/base/logging.h"
+#include "perfetto/base/unix_task_runner.h"
+
+namespace perfetto {
+namespace base {
+
+ThreadTaskRunner::ThreadTaskRunner(ThreadTaskRunner&& other) noexcept
+ : thread_(std::move(other.thread_)), task_runner_(other.task_runner_) {
+ other.task_runner_ = nullptr;
+}
+
+ThreadTaskRunner& ThreadTaskRunner::operator=(ThreadTaskRunner&& other) {
+ this->~ThreadTaskRunner();
+ new (this) ThreadTaskRunner(std::move(other));
+ return *this;
+}
+
+ThreadTaskRunner::~ThreadTaskRunner() {
+ if (task_runner_) {
+ PERFETTO_CHECK(!task_runner_->QuitCalled());
+ task_runner_->Quit();
+
+ PERFETTO_DCHECK(thread_.joinable());
+ }
+ if (thread_.joinable())
+ thread_.join();
+}
+
+ThreadTaskRunner::ThreadTaskRunner() {
+ std::mutex init_lock;
+ std::condition_variable init_cv;
+
+ std::function<void(UnixTaskRunner*)> initializer =
+ [this, &init_lock, &init_cv](UnixTaskRunner* task_runner) {
+ std::lock_guard<std::mutex> lock(init_lock);
+ task_runner_ = task_runner;
+ // Notify while still holding the lock, as init_cv ceases to exist as
+ // soon as the main thread observes a non-null task_runner_, and it can
+ // wake up spuriously (i.e. before the notify if we had unlocked before
+ // notifying).
+ init_cv.notify_one();
+ };
+ thread_ = std::thread(&ThreadTaskRunner::RunTaskThread, this,
+ std::move(initializer));
+
+ std::unique_lock<std::mutex> lock(init_lock);
+ init_cv.wait(lock, [this] { return !!task_runner_; });
+}
+
+void ThreadTaskRunner::RunTaskThread(
+ std::function<void(UnixTaskRunner*)> initializer) {
+ UnixTaskRunner task_runner;
+ task_runner.PostTask(std::bind(std::move(initializer), &task_runner));
+ task_runner.Run();
+}
+
+} // namespace base
+} // namespace perfetto
diff --git a/src/base/thread_task_runner_unittest.cc b/src/base/thread_task_runner_unittest.cc
new file mode 100644
index 0000000..823967b
--- /dev/null
+++ b/src/base/thread_task_runner_unittest.cc
@@ -0,0 +1,144 @@
+/*
+ * Copyright (C) 2019 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "perfetto/base/thread_task_runner.h"
+
+#include <thread>
+
+#include "gtest/gtest.h"
+#include "perfetto/base/thread_checker.h"
+
+namespace perfetto {
+namespace base {
+namespace {
+
+class ThreadTaskRunnerTest : public ::testing::Test {
+ protected:
+ std::atomic<bool> atomic_flag_{false};
+};
+
+TEST_F(ThreadTaskRunnerTest, ConstructedRunning) {
+ ThreadTaskRunner task_runner = ThreadTaskRunner::CreateAndStart();
+ task_runner.get()->PostTask([this] { atomic_flag_ = true; });
+ // main thread not blocked, wait on the task explicitly
+ while (!atomic_flag_) {
+ std::this_thread::yield();
+ }
+}
+
+TEST_F(ThreadTaskRunnerTest, RunsTasksOnOneDedicatedThread) {
+ ThreadTaskRunner task_runner = ThreadTaskRunner::CreateAndStart();
+ EXPECT_FALSE(task_runner.get()->RunsTasksOnCurrentThread());
+
+ ThreadChecker thread_checker;
+ task_runner.get()->PostTask([&thread_checker] {
+ // make thread_checker track the task thread
+ thread_checker.DetachFromThread();
+ EXPECT_TRUE(thread_checker.CalledOnValidThread());
+ });
+ task_runner.get()->PostTask([this, &thread_checker] {
+ // called on the same thread
+ EXPECT_TRUE(thread_checker.CalledOnValidThread());
+ atomic_flag_ = true;
+ });
+
+ while (!atomic_flag_) {
+ std::this_thread::yield();
+ }
+}
+
+TEST_F(ThreadTaskRunnerTest, MovableOwnership) {
+ ThreadTaskRunner task_runner = ThreadTaskRunner::CreateAndStart();
+ UnixTaskRunner* runner_ptr = task_runner.get();
+ EXPECT_NE(runner_ptr, nullptr);
+
+ ThreadChecker thread_checker;
+ task_runner.get()->PostTask([&thread_checker] {
+ // make thread_checker track the task thread
+ thread_checker.DetachFromThread();
+ EXPECT_TRUE(thread_checker.CalledOnValidThread());
+ });
+
+ // move ownership and destroy old instance
+ ThreadTaskRunner task_runner2 = std::move(task_runner);
+ EXPECT_EQ(task_runner.get(), nullptr);
+ task_runner.~ThreadTaskRunner();
+
+ // runner pointer is stable, and remains usable
+ EXPECT_EQ(task_runner2.get(), runner_ptr);
+ task_runner2.get()->PostTask([this, &thread_checker] {
+ // task thread remains the same
+ EXPECT_TRUE(thread_checker.CalledOnValidThread());
+ atomic_flag_ = true;
+ });
+
+ while (!atomic_flag_) {
+ std::this_thread::yield();
+ }
+}
+
+// Test helper callable that remembers a copy of a given ThreadChecker, and
+// checks that this class' destructor runs on the remembered thread. Note that
+// it is copyable so that it can be passed as a task (i.e. std::function) to a
+// task runner. Also note that all instances of this class will thread-check,
+// including those that have been moved-from.
+class DestructorThreadChecker {
+ public:
+ DestructorThreadChecker(ThreadChecker checker) : checker_(checker) {}
+ DestructorThreadChecker(const DestructorThreadChecker&) = default;
+ DestructorThreadChecker& operator=(const DestructorThreadChecker&) = default;
+ DestructorThreadChecker(DestructorThreadChecker&&) = default;
+ DestructorThreadChecker& operator=(DestructorThreadChecker&&) = default;
+
+ ~DestructorThreadChecker() { EXPECT_TRUE(checker_.CalledOnValidThread()); }
+
+ void operator()() { GTEST_FAIL() << "shouldn't be called"; }
+
+ ThreadChecker checker_;
+};
+
+// Checks that the still-pending tasks (and therefore the UnixTaskRunner itself)
+// are destructed on the task thread, and not the thread that destroys the
+// ThreadTaskRunner.
+TEST_F(ThreadTaskRunnerTest, EnqueuedTasksDestructedOnTaskThread) {
+ ThreadChecker thread_checker;
+ ThreadTaskRunner task_runner = ThreadTaskRunner::CreateAndStart();
+
+ task_runner.get()->PostTask([this, &thread_checker, &task_runner] {
+ // make thread_checker track the task thread
+ thread_checker.DetachFromThread();
+ EXPECT_TRUE(thread_checker.CalledOnValidThread());
+ // Post a follow-up delayed task and unblock the main thread, which will
+ // destroy the ThreadTaskRunner while this task is still pending.
+ //
+ // Note: DestructorThreadChecker will thread-check (at least) twice:
+ // * for the temporary that was moved-from to construct the task
+ // std::function. Will pass as we're posting from a task thread.
+ // * for the still-pending task once the ThreadTaskRunner destruction causes
+ // the destruction of UnixTaskRunner.
+ task_runner.get()->PostDelayedTask(DestructorThreadChecker(thread_checker),
+ 100 * 1000 /*ms*/);
+ atomic_flag_ = true;
+ });
+
+ while (!atomic_flag_) {
+ std::this_thread::yield();
+ }
+}
+
+} // namespace
+} // namespace base
+} // namespace perfetto
diff --git a/src/base/unix_task_runner.cc b/src/base/unix_task_runner.cc
index 937a4b1..03a1935 100644
--- a/src/base/unix_task_runner.cc
+++ b/src/base/unix_task_runner.cc
@@ -72,6 +72,11 @@
WakeUp();
}
+bool UnixTaskRunner::QuitCalled() {
+ std::lock_guard<std::mutex> lock(lock_);
+ return quit_;
+}
+
bool UnixTaskRunner::IsIdleForTesting() {
std::lock_guard<std::mutex> lock(lock_);
return immediate_tasks_.empty();
