Lock free task runner [WIP] WORK IN PROGRESS A lock-free TaskRunner implementation. The core logic is there, missing bits: - Dealing with Windows - File descriptor watches - Flagging
diff --git a/Android.bp b/Android.bp index 2f40772..9cf7b5f 100644 --- a/Android.bp +++ b/Android.bp
@@ -12283,6 +12283,7 @@ "src/base/getopt_compat.cc", "src/base/intrusive_list.cc", "src/base/intrusive_tree.cc", + "src/base/lock_free_task_runner.cc", "src/base/logging.cc", "src/base/metatrace.cc", "src/base/paged_memory.cc",
diff --git a/BUILD b/BUILD index 8410372..398f88d 100644 --- a/BUILD +++ b/BUILD
@@ -824,6 +824,7 @@ name = "include_perfetto_ext_base_base", srcs = [ "include/perfetto/ext/base/android_utils.h", + "include/perfetto/ext/base/atomic_shared_ptr.h", "include/perfetto/ext/base/base64.h", "include/perfetto/ext/base/bits.h", "include/perfetto/ext/base/circular_queue.h", @@ -840,6 +841,7 @@ "include/perfetto/ext/base/getopt.h", "include/perfetto/ext/base/getopt_compat.h", "include/perfetto/ext/base/hash.h", + "include/perfetto/ext/base/lock_free_task_runner.h", "include/perfetto/ext/base/metatrace.h", "include/perfetto/ext/base/metatrace_events.h", "include/perfetto/ext/base/murmur_hash.h", @@ -1297,6 +1299,7 @@ "src/base/intrusive_list.h", "src/base/intrusive_tree.cc", "src/base/intrusive_tree.h", + "src/base/lock_free_task_runner.cc", "src/base/log_ring_buffer.h", "src/base/logging.cc", "src/base/metatrace.cc",
diff --git a/gn/BUILD.gn b/gn/BUILD.gn index 37d2e3e..c2aebf3 100644 --- a/gn/BUILD.gn +++ b/gn/BUILD.gn
@@ -107,6 +107,7 @@ "PERFETTO_THREAD_SAFETY_ANNOTATIONS=$perfetto_thread_safety_annotations", "PERFETTO_ENABLE_WINSCOPE=$enable_perfetto_winscope", "PERFETTO_ENABLE_RT_MUTEX=$enable_perfetto_rt_mutex", + "PERFETTO_ENABLE_LOCKFREE_TASKRUNNER=$enable_perfetto_lockfree_taskrunner", ] rel_out_path = rebase_path(gen_header_path, "$root_build_dir")
diff --git a/gn/perfetto.gni b/gn/perfetto.gni index c52143a..06aefb9 100644 --- a/gn/perfetto.gni +++ b/gn/perfetto.gni
@@ -277,6 +277,9 @@ (!is_wasm && ((build_with_chromium || perfetto_build_standalone) && (current_cpu == "x64" || current_cpu == "arm64"))) || is_perfetto_build_generator + + # TODO DNS. chill out here. True just for CI testing. + enable_perfetto_lockfree_taskrunner = true } declare_args() {
diff --git a/include/perfetto/base/build_configs/android_tree/perfetto_build_flags.h b/include/perfetto/base/build_configs/android_tree/perfetto_build_flags.h index 72bedff..a85b6b4 100644 --- a/include/perfetto/base/build_configs/android_tree/perfetto_build_flags.h +++ b/include/perfetto/base/build_configs/android_tree/perfetto_build_flags.h
@@ -53,6 +53,7 @@ #define PERFETTO_BUILDFLAG_DEFINE_PERFETTO_THREAD_SAFETY_ANNOTATIONS() (0) #define PERFETTO_BUILDFLAG_DEFINE_PERFETTO_ENABLE_WINSCOPE() (1) #define PERFETTO_BUILDFLAG_DEFINE_PERFETTO_ENABLE_RT_MUTEX() (1) +#define PERFETTO_BUILDFLAG_DEFINE_PERFETTO_ENABLE_LOCKFREE_TASKRUNNER() (1) struct PerfettoBuildFlag { const char* name; @@ -89,6 +90,7 @@ {"PERFETTO_THREAD_SAFETY_ANNOTATIONS", PERFETTO_BUILDFLAG_DEFINE_PERFETTO_THREAD_SAFETY_ANNOTATIONS()}, {"PERFETTO_ENABLE_WINSCOPE", PERFETTO_BUILDFLAG_DEFINE_PERFETTO_ENABLE_WINSCOPE()}, {"PERFETTO_ENABLE_RT_MUTEX", PERFETTO_BUILDFLAG_DEFINE_PERFETTO_ENABLE_RT_MUTEX()}, + {"PERFETTO_ENABLE_LOCKFREE_TASKRUNNER", PERFETTO_BUILDFLAG_DEFINE_PERFETTO_ENABLE_LOCKFREE_TASKRUNNER()}, }; static const int kPerfettoBuildFlagsCount = sizeof(kPerfettoBuildFlags) / sizeof(struct PerfettoBuildFlag);
diff --git a/include/perfetto/base/build_configs/bazel/perfetto_build_flags.h b/include/perfetto/base/build_configs/bazel/perfetto_build_flags.h index 1c8e829..8bbcf0e 100644 --- a/include/perfetto/base/build_configs/bazel/perfetto_build_flags.h +++ b/include/perfetto/base/build_configs/bazel/perfetto_build_flags.h
@@ -53,6 +53,7 @@ #define PERFETTO_BUILDFLAG_DEFINE_PERFETTO_THREAD_SAFETY_ANNOTATIONS() (0) #define PERFETTO_BUILDFLAG_DEFINE_PERFETTO_ENABLE_WINSCOPE() (1) #define PERFETTO_BUILDFLAG_DEFINE_PERFETTO_ENABLE_RT_MUTEX() (1) +#define PERFETTO_BUILDFLAG_DEFINE_PERFETTO_ENABLE_LOCKFREE_TASKRUNNER() (1) struct PerfettoBuildFlag { const char* name; @@ -89,6 +90,7 @@ {"PERFETTO_THREAD_SAFETY_ANNOTATIONS", PERFETTO_BUILDFLAG_DEFINE_PERFETTO_THREAD_SAFETY_ANNOTATIONS()}, {"PERFETTO_ENABLE_WINSCOPE", PERFETTO_BUILDFLAG_DEFINE_PERFETTO_ENABLE_WINSCOPE()}, {"PERFETTO_ENABLE_RT_MUTEX", PERFETTO_BUILDFLAG_DEFINE_PERFETTO_ENABLE_RT_MUTEX()}, + {"PERFETTO_ENABLE_LOCKFREE_TASKRUNNER", PERFETTO_BUILDFLAG_DEFINE_PERFETTO_ENABLE_LOCKFREE_TASKRUNNER()}, }; static const int kPerfettoBuildFlagsCount = sizeof(kPerfettoBuildFlags) / sizeof(struct PerfettoBuildFlag);
diff --git a/include/perfetto/base/flat_set.h b/include/perfetto/base/flat_set.h index 5be590e..898eb0d 100644 --- a/include/perfetto/base/flat_set.h +++ b/include/perfetto/base/flat_set.h
@@ -92,6 +92,8 @@ size_t size() const { return entries_.size(); } const_iterator begin() const { return entries_.begin(); } const_iterator end() const { return entries_.end(); } + const T& back() const { return entries_.back(); } + void pop_back() { entries_.pop_back(); } private: std::vector<T> entries_;
diff --git a/include/perfetto/ext/base/BUILD.gn b/include/perfetto/ext/base/BUILD.gn index 481ecbf..1eb0a05 100644 --- a/include/perfetto/ext/base/BUILD.gn +++ b/include/perfetto/ext/base/BUILD.gn
@@ -17,6 +17,7 @@ source_set("base") { sources = [ "android_utils.h", + "atomic_shared_ptr.h", "base64.h", "bits.h", "circular_queue.h", @@ -33,6 +34,7 @@ "getopt.h", "getopt_compat.h", "hash.h", + "lock_free_task_runner.h", "metatrace.h", "metatrace_events.h", "murmur_hash.h",
diff --git a/include/perfetto/ext/base/atomic_shared_ptr.h b/include/perfetto/ext/base/atomic_shared_ptr.h new file mode 100644 index 0000000..6ca7ebf --- /dev/null +++ b/include/perfetto/ext/base/atomic_shared_ptr.h
@@ -0,0 +1,85 @@ +/* + * Copyright (C) 2023 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. + */ + +#ifndef INCLUDE_PERFETTO_EXT_BASE_ATOMIC_SHARED_PTR_H_ +#define INCLUDE_PERFETTO_EXT_BASE_ATOMIC_SHARED_PTR_H_ + +#include <atomic> +#include <memory> + +namespace perfetto { +namespace base { + +// Wrapper providing a uniform API for atomic shared_ptr operations across +// pre-C++20 (free functions) and C++20+ (std::atomic<std::shared_ptr<T>>). +template <typename T> +class AtomicSharedPtr { + public: + AtomicSharedPtr() noexcept = default; + explicit AtomicSharedPtr(std::shared_ptr<T> p) noexcept { + store(std::move(p)); + } + + // Non-copyable (like many atomic wrappers). Movable for convenience. + AtomicSharedPtr(const AtomicSharedPtr&) = delete; + AtomicSharedPtr& operator=(const AtomicSharedPtr&) = delete; + AtomicSharedPtr(AtomicSharedPtr&&) = delete; + AtomicSharedPtr& operator=(AtomicSharedPtr&&) = delete; + + std::shared_ptr<T> load( + std::memory_order order = std::memory_order_seq_cst) const noexcept { +#if defined(__cpp_lib_atomic_shared_ptr) + return ptr_.load(order); +#else + return std::atomic_load_explicit(&ptr_, order); +#endif + } + + void store(std::shared_ptr<T> desired, + std::memory_order order = std::memory_order_seq_cst) noexcept { +#if defined(__cpp_lib_atomic_shared_ptr) + ptr_.store(std::move(desired), order); +#else + std::atomic_store_explicit(&ptr_, std::move(desired), order); +#endif + } + + bool compare_exchange_strong( + std::shared_ptr<T>& expected, + std::shared_ptr<T> desired, + std::memory_order success = std::memory_order_seq_cst, + std::memory_order failure = std::memory_order_seq_cst) noexcept { +#if defined(__cpp_lib_atomic_shared_ptr) + return ptr_.compare_exchange_strong(expected, std::move(desired), success, + failure); +#else + return std::atomic_compare_exchange_strong_explicit( + &ptr_, &expected, std::move(desired), success, failure); +#endif + } + + private: +#if defined(__cpp_lib_atomic_shared_ptr) // C++20 and later + std::atomic<std::shared_ptr<T>> ptr_{}; +#else // pre-C++20: use free-function atomics with a plain shared_ptr + mutable std::shared_ptr<T> ptr_{}; +#endif +}; + +} // namespace base +} // namespace perfetto + +#endif // INCLUDE_PERFETTO_EXT_BASE_ATOMIC_SHARED_PTR_H_
diff --git a/include/perfetto/ext/base/flags.h b/include/perfetto/ext/base/flags.h index b57af9a..ca71497 100644 --- a/include/perfetto/ext/base/flags.h +++ b/include/perfetto/ext/base/flags.h
@@ -38,7 +38,11 @@ X(ftrace_clear_offline_cpus_only, NonAndroidPlatformDefault_TRUE) \ X(use_rt_mutex, PERFETTO_BUILDFLAG(PERFETTO_ENABLE_RT_MUTEX) \ ? NonAndroidPlatformDefault_TRUE \ - : NonAndroidPlatformDefault_FALSE) + : NonAndroidPlatformDefault_FALSE) \ + X(use_lockfree_taskrunner, \ + PERFETTO_BUILDFLAG(PERFETTO_ENABLE_LOCKFREE_TASKRUNNER) \ + ? NonAndroidPlatformDefault_TRUE \ + : NonAndroidPlatformDefault_FALSE) //////////////////////////////////////////////////////////////////////////////// // //
diff --git a/include/perfetto/ext/base/lock_free_task_runner.h b/include/perfetto/ext/base/lock_free_task_runner.h new file mode 100644 index 0000000..0a793c9 --- /dev/null +++ b/include/perfetto/ext/base/lock_free_task_runner.h
@@ -0,0 +1,254 @@ +/* + * Copyright (C) 2025 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. + */ + +#ifndef INCLUDE_PERFETTO_EXT_BASE_LOCK_FREE_TASK_RUNNER_H_ +#define INCLUDE_PERFETTO_EXT_BASE_LOCK_FREE_TASK_RUNNER_H_ + +#include "perfetto/base/flat_set.h" +#include "perfetto/base/task_runner.h" +#include "perfetto/base/thread_annotations.h" +#include "perfetto/base/time.h" +#include "perfetto/ext/base/atomic_shared_ptr.h" +#include "perfetto/ext/base/event_fd.h" +#include "perfetto/ext/base/flags.h" +#include "perfetto/ext/base/scoped_file.h" +#include "perfetto/ext/base/unix_task_runner.h" + +#if !PERFETTO_BUILDFLAG(PERFETTO_OS_WIN) +#include <poll.h> +#endif + +#include <array> +#include <atomic> +#include <map> +#include <thread> + +namespace perfetto { +namespace base { + +// This class implements a lock-less multi-producer single-consumer task runner. +// This is achieved by using a linked list of "slabs". Each slab is a fixed-size +// array of tasks. +// +// The overall architecture is as follows: +// - There is one "main" thread, which is the only thread that is allowed to +// invoke Run(). This is the consumer thread. +// - There can be multiple "writer" threads, which are the threads that call +// PostTask(). These are the producer threads. +// +// The slabs are organized as a singly-linked list, linked from the tail: +// tail -> [Slab N] -> [Slab N-1] -> ... -> [Slab 0] -> null +// The tail points to the latest Slab. In nominal cases (i.e. in absence of +// PostTask() bursts, assuming Run catches up) there is only one (or at most +// two) Slabs in the list. +// +// Writer threads atomically try to reserve a slot in the current `tail` slab. +// If the slab is full, they allocate a new slab and atomically swap the +// `tail` pointer to point to the new slab, linking the old tail as `prev`. +// +// The key design element is that writer threads only ever access the `tail` +// slab and never look at the `->prev` pointer / never iterate the list. +// Only the main Run() thread iterates the list. This makes the design simpler +// to reason about. +// +// The main thread, instead, is the only one that is allowed to follow the +// `->prev` pointers to drain the tasks. +// +// Slab lifecycle: +// - A new slab is created by a writer thread when the current slab is full. +// - The main thread drains tasks from slabs (from 0 to N). When a slab becomes +// empty, it's destroyed using a shared_ptr, which guarrantees that the slab +// is not destroyed while another writer thread is trying to append tasks. +// - As a further optimization, empty slabs are kept around in a free-list of +// size 1. This is makes it so that in absence of bursts this class doesn't +// perform any allocation. +// +// tail_ (atomic_shared_ptr) +// | +// ▼ +// +-----------------+ +-----------------+ +-----------------+ +// | Slab N | | Slab N-1 | | Slab 0 | +// | tasks: [....] | | tasks: [....] | | tasks: [....] | +// | next_task_slot | | next_task_slot | | next_task_slot | +// | prev (sptr) ----+----->| prev (sptr) ----+----->| prev = nullptr | +// +-----------------+ +-----------------+ +-----------------+ +// +class PERFETTO_EXPORT_COMPONENT LockFreeTaskRunner : public TaskRunner { + public: + LockFreeTaskRunner(); + ~LockFreeTaskRunner() override; + + void Run(); + void Quit(); + + // Checks whether there are any pending immediate tasks to run. Note that + // delayed tasks don't count even if they are due to run. + bool IsIdleForTesting(); + + // TaskRunner implementation: + void PostTask(std::function<void()>) override; + void PostDelayedTask(std::function<void()>, uint32_t delay_ms) override; + void AddFileDescriptorWatch(PlatformHandle, std::function<void()>) override; + void RemoveFileDescriptorWatch(PlatformHandle) override; + bool RunsTasksOnCurrentThread() const override; + + // Pretends (for the purposes of running delayed tasks) that time advanced by + // `ms`. + void AdvanceTimeForTesting(uint32_t ms); + + static constexpr size_t kSlabSize = 512; // Exposed for testing. + + // Stats for testing. + size_t slabs_allocated() const { + return slabs_allocated_.load(std::memory_order_relaxed); + } + size_t slabs_freed() const { + return slabs_freed_.load(std::memory_order_relaxed); + } + + private: + // A slab is a fixed-size array of tasks. The lifecycle of a task slot + // within a slab goes through three phases: + // + // 1. Reservation: A writer thread atomically increments `next_task_slot` to + // reserve a slot in the `tasks` array. This reservation establishes the + // implicit order in which the consumer will attempt to read tasks (but + // only if they are published in the bitmap, see below). + // + // 2. Publishing: After writing the task into its reserved slot, the writer + // thread atomically sets the corresponding bit in the `tasks_written` + // bitmask. This acts as a memory barrier and makes the task visible to + // the consumer (main) thread. + // + // 3. Consumption: The main thread acquire-reads the `tasks_written` bitmask. + // For each bit that is set, it processes the task and then sets the + // corresponding bit in its private `tasks_read` bitmask to prevent + // reading the same task again. + struct Slab { + Slab(); + ~Slab(); + + // `tasks` and `next_task_slot` are accessed by writer threads only. + // The main thread can access `tasks[i]` but only after ensuring that the + // corresponding bit in `tasks_written` is set. + std::array<std::function<void()>, kSlabSize> tasks{}; + std::atomic<size_t> next_task_slot{0}; + + // A bitmask indicating which tasks in the `tasks` array have been written + // and are ready to be read by the main thread. + // This is atomically updated by writer threads and read by the main thread. + using BitWord = size_t; + static constexpr size_t kBitsPerWord = sizeof(BitWord) * 8; + static constexpr size_t kNumWords = kSlabSize / kBitsPerWord; + std::array<std::atomic<BitWord>, kNumWords> tasks_written{}; + + // A bitmask indicating which tasks have been read by the main thread. + // This is accessed only by the main thread, so no atomicity is required. + std::array<BitWord, kNumWords> tasks_read{}; + + // The link to the previous slab. + // This is written by writer threads when they create a new slab and link it + // to the previous tail. But they do so when nobody else can see the Slab, + // so there is no need for an AtomicSharedPtr. After the initial creation, + // this is accessed only by the main thread when: + // 1. draining tasks (to walk back to the oldest slab) + // 2. deleting slabs, setting it to nullptr, when they are fully consumed. + std::shared_ptr<Slab> prev; + }; + + struct DelayedTask { + TimeMillis time; + uint64_t seq; + std::function<void()> task; + + // Note that the < operator keeps the DelayedTasks sorted in reverse order + // (the latest one is first, the earliest one is last). This is so we can + // have a FIFO queue using a vector by just doing an O(1) pop_back(). + bool operator<(const DelayedTask& other) const { + if (time != other.time) + return time > other.time; + return seq > other.seq; + } + bool operator==(const DelayedTask& other) const { + return time == other.time && seq == other.seq; + } + }; + + std::function<void()> PopNextImmediateTask(); + std::function<void()> PopTaskRecursive(const std::shared_ptr<Slab>&, + Slab* next_slab); + void EnqueueExpiredDelayedTasks(); + int GetDelayMsToNextTask() const; + void WakeUp() { wakeup_event_.Notify(); } + std::shared_ptr<Slab> AllocNewSlab(); + void PostFileDescriptorWatches(uint64_t windows_wait_result); + void RunFileDescriptorWatch(PlatformHandle); + void UpdateWatchTasks(); + + // This is semantically a unique_ptr, but is accessed from different threads. + std::atomic<Slab*> free_slab_{}; + + EventFd wakeup_event_; + std::atomic<bool> quit_{false}; + std::thread::id run_task_thread_id_; + + // Delayed tasks, accessed only by the main thread. Items are stored in + // reverse temporal order, see comment in the operator<. + FlatSet<DelayedTask> delayed_tasks_; + uint64_t next_delayed_task_seq_ = 0; + std::atomic<uint32_t> advanced_time_for_testing_{}; + + // The array of fds/handles passed to poll(2) / WaitForMultipleObjects(). +#if PERFETTO_BUILDFLAG(PERFETTO_OS_WIN) + std::vector<PlatformHandle> poll_fds_; +#else + std::vector<struct pollfd> poll_fds_; +#endif + + struct WatchTask { + std::function<void()> callback; +#if PERFETTO_BUILDFLAG(PERFETTO_OS_WIN) + // On UNIX systems we make the FD number negative in |poll_fds_| to avoid + // polling it again until the queued task runs. On Windows we can't do that. + // Instead we keep track of its state here. + bool pending = false; +#else + size_t poll_fd_index; // Index into |poll_fds_|. +#endif + }; + + // Accessed only from the main thread. + std::map<PlatformHandle, WatchTask> watch_tasks_; + bool watch_tasks_changed_ = false; + + // Stats for testing. + std::atomic<size_t> slabs_allocated_{}; + std::atomic<size_t> slabs_freed_{}; + + // Keep last, so deletion of slabs happens before invalidating the remaining + // state. + AtomicSharedPtr<Slab> tail_; +}; + +using MaybeLockFreeTaskRunner = + std::conditional_t<base::flags::use_lockfree_taskrunner, + LockFreeTaskRunner, + UnixTaskRunner>; + +} // namespace base +} // namespace perfetto + +#endif // INCLUDE_PERFETTO_EXT_BASE_LOCK_FREE_TASK_RUNNER_H_
diff --git a/include/perfetto/ext/base/thread_task_runner.h b/include/perfetto/ext/base/thread_task_runner.h index 950de9b..95dad40 100644 --- a/include/perfetto/ext/base/thread_task_runner.h +++ b/include/perfetto/ext/base/thread_task_runner.h
@@ -20,17 +20,19 @@ #include <functional> #include <thread> -#include "perfetto/ext/base/unix_task_runner.h" +#include "perfetto/ext/base/lock_free_task_runner.h" namespace perfetto { namespace base { -// A UnixTaskRunner backed by a dedicated task thread. Shuts down the runner and -// joins the thread upon destruction. Can be moved to transfer ownership. +// A MaybeLockFreeTaskRunner backed by a dedicated task thread. Shuts down the +// runner and joins the thread upon destruction. Can be moved to transfer +// ownership. // // Guarantees that: -// * the UnixTaskRunner will be constructed and destructed on the task thread. -// * the task thread will live for the lifetime of the UnixTaskRunner. +// * the MaybeLockFreeTaskRunner will be constructed and destructed on the task +// thread. +// * the task thread will live for the lifetime of the MaybeLockFreeTaskRunner. // class PERFETTO_EXPORT_COMPONENT ThreadTaskRunner : public TaskRunner { public: @@ -55,14 +57,14 @@ PlatformThreadId GetThreadIdForTesting(); - // Returns a pointer to the UnixTaskRunner, which is valid for the lifetime of - // this ThreadTaskRunner object (unless this object is moved-from, in which - // case the pointer remains valid for the lifetime of the new owning + // Returns a pointer to the MaybeLockFreeTaskRunner, which is valid for the + // lifetime of this ThreadTaskRunner object (unless this object is moved-from, + // in which case the pointer remains valid for the lifetime of the new owning // ThreadTaskRunner). // // Warning: do not call Quit() on the returned runner pointer, the termination // should be handled exclusively by this class' destructor. - UnixTaskRunner* get() const { return task_runner_; } + MaybeLockFreeTaskRunner* get() const { return task_runner_; } // TaskRunner implementation. // These methods just proxy to the underlying task_runner_. @@ -74,11 +76,11 @@ private: explicit ThreadTaskRunner(const std::string& name); - void RunTaskThread(std::function<void(UnixTaskRunner*)> initializer); + void RunTaskThread(std::function<void(MaybeLockFreeTaskRunner*)> initializer); std::thread thread_; std::string name_; - UnixTaskRunner* task_runner_ = nullptr; + MaybeLockFreeTaskRunner* task_runner_ = nullptr; }; } // namespace base
diff --git a/src/base/BUILD.gn b/src/base/BUILD.gn index 83877fe..d253d6b 100644 --- a/src/base/BUILD.gn +++ b/src/base/BUILD.gn
@@ -72,6 +72,7 @@ if (!is_nacl) { sources += [ + "lock_free_task_runner.cc", "thread_task_runner.cc", "unix_task_runner.cc", ]
diff --git a/src/base/lock_free_task_runner.cc b/src/base/lock_free_task_runner.cc new file mode 100644 index 0000000..5c7f05d --- /dev/null +++ b/src/base/lock_free_task_runner.cc
@@ -0,0 +1,509 @@ + +/* + * Copyright (C) 2025 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 of 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/ext/base/lock_free_task_runner.h" + +#include "perfetto/base/build_config.h" + +#if !PERFETTO_BUILDFLAG(PERFETTO_OS_WIN) +#include <poll.h> +#endif + +#include <thread> + +#include "perfetto/base/logging.h" +#include "perfetto/base/time.h" +#include "perfetto/ext/base/bits.h" +#include "perfetto/ext/base/platform.h" +#include "perfetto/ext/base/utils.h" +#include "perfetto/ext/base/watchdog.h" + +namespace perfetto { +namespace base { + +// --- LockFreeTaskRunner::Slab --- +LockFreeTaskRunner::Slab::Slab() = default; +LockFreeTaskRunner::Slab::~Slab() = default; + +// --- LockFreeTaskRunner --- +LockFreeTaskRunner::LockFreeTaskRunner() + : run_task_thread_id_(std::this_thread::get_id()) { + static_assert((kSlabSize & (kSlabSize - 1)) == 0, "kSlabSize must be a pow2"); + static_assert(kSlabSize >= Slab::kBitsPerWord); + + AddFileDescriptorWatch(wakeup_event_.fd(), [] { + // Not reached -- see PostFileDescriptorWatches(). + PERFETTO_DFATAL("Should be unreachable."); + }); +} + +LockFreeTaskRunner::~LockFreeTaskRunner() { + PERFETTO_DCHECK(RunsTasksOnCurrentThread()); + delete free_slab_.exchange(nullptr); // no check needed, delete(null) is ok. +} + +void LockFreeTaskRunner::PostTask(std::function<void()> closure) { + // We use nullity of std::function in PopTaskRecursive() to determine exit + // criteria. Posting a null task would break that logic. Also a null task + // would cause an abort when trying to run it later on. + PERFETTO_CHECK(PERFETTO_LIKELY(closure)); + using BitWord = Slab::BitWord; + for (;;) { + std::shared_ptr<Slab> slab = tail_.load(std::memory_order_acquire); + if (PERFETTO_UNLIKELY(!slab)) { + // This happens on the very first call, and on each call after the reader + // has consumed a full slab (once every kSlabSize tasks). + tail_.compare_exchange_strong(slab, AllocNewSlab()); + // If the cmpxcgh fails, another thread allocated a new slab and won the + // race. It doesn't matter really as long as we have a slab. + // In either case retry, as now we should have a slab. + continue; + } + + // We have 3 cases here: + // 1. slot < kSlabSize: the nominal case. Append the task and return. + // 2. slot == kSlabSize: the common overflow case: The slab was full and we + // tried to allocate the N+1 th element. We have to allocate a new Slab. + // 3. slot > kSlabSize: like 2, but but two (or more) threads raced on it. + // One thread will win the race and alloc a new slab, the other one will + // repeat. + size_t slot = slab->next_task_slot.fetch_add(1, std::memory_order_relaxed); + + if (slot >= kSlabSize) { // Cases 2,3 + std::shared_ptr<Slab> new_slab = AllocNewSlab(); + + new_slab->prev = slab; + new_slab->next_task_slot.store(1, std::memory_order_relaxed); + slot = 0; + if (PERFETTO_UNLIKELY(!tail_.compare_exchange_strong(slab, new_slab))) { + // If the cmpxcgh fails, another thread tried to allocate a new tail + // slab and won the race. Do another round, we'll observe the new slab. + + // The reset() below is not really needed as the dtor (triggered by + // `new_slab` going out of scope without any further refcount) resets + // the `prev` shared_ptr. This is here just for future-proofness, in + // case somebody in future changes the freelist logic and forgets to + // invoke the dtor. Doing so would leak slabs, which could go unnoticed. + new_slab->prev.reset(); + continue; + } + slab = new_slab; + } + + // Nominal case: publish the task and return. + PERFETTO_DCHECK(!slab->tasks[slot]); + slab->tasks[slot] = std::move(closure); + size_t s_word = slot / Slab::kBitsPerWord; + size_t s_bit = slot % Slab::kBitsPerWord; + BitWord s_mask = BitWord(1) << s_bit; + PERFETTO_DCHECK( + (slab->tasks_written[s_word].load(std::memory_order_relaxed) & + s_mask) == 0); + slab->tasks_written[s_word].fetch_or(s_mask, std::memory_order_release); + + if (!RunsTasksOnCurrentThread()) { + // We don't need any clever logic to avoid spurious wake ups from other + // threads. Most PostTask()s in our codebase are done by the main thread. + // In the rare cases of a PostTask() coming from another thread, the odds + // of the main thread being woken up at the same time are tiny. + WakeUp(); + } + return; + } +} + +void LockFreeTaskRunner::Run() { + run_task_thread_id_ = std::this_thread::get_id(); + quit_.store(false, std::memory_order_relaxed); + + while (!quit_.load(std::memory_order_relaxed)) { + // Step 1: if any delayed task is expired, post it now and turn it into an + // immediate task. + if (!delayed_tasks_.empty()) { + EnqueueExpiredDelayedTasks(); + } + + // Step 2: extract an immediate task, if any. + int poll_timeout_ms = -1; + std::function<void()> imm_task = PopNextImmediateTask(); + poll_timeout_ms = imm_task ? 0 : GetDelayMsToNextTask(); + + // Step 3: run the poll(). We need it for two different reasons: + // 1. Blocks until the next event on the horizon, which: + // - If we pulled an immediate task, poll in non-blocking mode (0 delay) + // because we want to run the task immediately. + // - If there is a delayed task, compute the time remaining for it. + // - Otherwise polls indefinitely, waiting for a PostTask() or a Quit() + // call from another thread. + // 2. Regardless of timing, we need to read the FD watches. + // We want to do this even if we know already that we have an immediate + // task (when poll_timeout_ms = 0) to ensure fairness. + // TODO(primiano): we could optimize this and avoid a syscall for each + // task when we have bursts of tasks. But today RunUntilIdle() relies on + // FD watches to be polled before we run the current task (which might be + // last). + + // Recompute the list of FDs to watch. + UpdateWatchTasks(); + + uint64_t windows_wait_res = 0; +#if PERFETTO_BUILDFLAG(PERFETTO_OS_WIN) + // Unlike poll(2), WaitForMultipleObjects() returns only *one* handle in the + // set, even when >1 is signalled. In order to avoid starvation, + // PostFileDescriptorWatches() will WaitForSingleObject() each other handle + // to ensure fairness. |windows_wait_res| is passed just to avoid an extra + // WaitForSingleObject() for the one handle that WaitForMultipleObject() + // returned. + DWORD timeout = + poll_timeout_ms >= 0 ? static_cast<DWORD>(poll_timeout_ms) : INFINITE; + windows_wait_res = + WaitForMultipleObjects(static_cast<DWORD>(poll_fds_.size()), + &poll_fds_[0], /*bWaitAll=*/false, timeout); +#else + platform::BeforeMaybeBlockingSyscall(); + int ret = PERFETTO_EINTR(poll( + &poll_fds_[0], static_cast<nfds_t>(poll_fds_.size()), poll_timeout_ms)); + platform::AfterMaybeBlockingSyscall(); + PERFETTO_CHECK(ret >= 0); +#endif + PostFileDescriptorWatches(windows_wait_res); + + if (imm_task) { + errno = 0; + RunTaskWithWatchdogGuard(std::move(imm_task)); + } + } +} + +std::function<void()> LockFreeTaskRunner::PopNextImmediateTask() { + std::shared_ptr<Slab> tail = tail_.load(std::memory_order_acquire); + if (!tail) { + // The tail can be null: + // 1. When invoking Run() with no task present (e.g. after ctor). + // 2. When PopTaskRecursive() below deletes the only slab present. + return nullptr; + } + return PopTaskRecursive(tail, nullptr); +} + +std::function<void()> LockFreeTaskRunner::PopTaskRecursive( + const std::shared_ptr<Slab>& slab, + Slab* next_slab) { + PERFETTO_DCHECK(RunsTasksOnCurrentThread()); + const std::shared_ptr<Slab>& prev = slab->prev; + if (PERFETTO_UNLIKELY(prev)) { + // In practice it's extemely unlikely that a slab has >1 predecessors. + // In nominal conditions it is going to have 0 predecessors most of the + // times and 1 predecessors 1 every kSlabSize times. + auto task = PopTaskRecursive(prev, slab.get()); + if (task) + return task; + } + + size_t words_fully_consumed = 0; + for (size_t w = 0; w < Slab::kNumWords; ++w) { + using BitWord = Slab::BitWord; + BitWord wr_word = slab->tasks_written[w].load(std::memory_order_acquire); + BitWord rd_word = slab->tasks_read[w]; + words_fully_consumed += base::AllBitsSet(rd_word) ? 1 : 0; + BitWord unread_word = wr_word & ~rd_word; + + if (unread_word == 0) + continue; + + // Find the first unread task in the word. + uint32_t bit = base::CountTrailZeros(unread_word); + BitWord bit_mask = BitWord(1) << bit; + size_t slot = w * Slab::kBitsPerWord + bit; + std::function<void()> task = std::move(slab->tasks[slot]); + slab->tasks[slot] = nullptr; + slab->tasks_read[w] |= bit_mask; + return task; + } // for(word in tasks_written) + + // There are no unconsumed tasks in this Slab. Reached this point, this + // invocation will return null. However, before doing so, if the slab is fully + // written (are no slots left) and fully consumed delete it. We delete only + // slabs that have no predecessor, from the oldest to the newest, to keep the + // logic simpler as slabs are fully consumed in that order. The only thing + // that could keep a slab alive is a thread getting descheduled between the + // acquisition of the slot and the publishing of the written bit. This is very + // unlikely as a thread should be descheduled for the time it takes to fill up + // and consume another slab. Even if it happens, it will just delay a bit the + // deletion of the chain. + + bool slab_fully_consumed = words_fully_consumed == Slab::kNumWords; + + if (slab_fully_consumed && !prev) { + // NOTE: only the main thread follows the `prev` linked list, writers never + // look at `prev`. The only contention entrypoint is the `tail_` pointer, + // which can be modified both by us and by writers. + PERFETTO_DCHECK(slab->prev.get() == nullptr); + if (next_slab) { + next_slab->prev.reset(); + // The current `slab` might get deleted at this point, as the shared_ptr + // of next_slab.prev might be the only one refcounting it. + } else { + // If we get here, `slab` is the only Slab: it has no prev, and it is the + // one `tail_` is pointing to. We need to update the `tail_` pointer but, + // by doing so, we might race with a writer thread allocating a new slab + // (and pointing back to us). + + std::shared_ptr<Slab> expected = slab; + tail_.compare_exchange_strong(expected, nullptr); + + // If the compxcgh fails, another thread managed to add a new slab, which + // points back to us, essentially invalidating our attempt to remove the + // current slab. This is not a big deal really. We will try again removing + // this slab on the next invocation. + } + } + + return nullptr; +} + +void LockFreeTaskRunner::Quit() { + quit_.store(true, std::memory_order_relaxed); + WakeUp(); +} + +bool LockFreeTaskRunner::IsIdleForTesting() { + PERFETTO_DCHECK(RunsTasksOnCurrentThread()); + for (std::shared_ptr<Slab> slab = tail_.load(std::memory_order_acquire); slab; + slab = slab->prev) { + for (size_t i = 0; i < Slab::kNumWords; ++i) { + if (slab->tasks_written[i] & ~slab->tasks_read[i]) { + return false; + } + } + } + return true; +} + +void LockFreeTaskRunner::EnqueueExpiredDelayedTasks() { + PERFETTO_DCHECK(RunsTasksOnCurrentThread()); + TimeMillis now = + GetWallTimeMs() + + TimeMillis(advanced_time_for_testing_.load(std::memory_order_relaxed)); + while (!delayed_tasks_.empty() && delayed_tasks_.back().time <= now) { + auto task = std::move(const_cast<DelayedTask&>(delayed_tasks_.back()).task); + delayed_tasks_.pop_back(); + PostTask(std::move(task)); + } +} + +int LockFreeTaskRunner::GetDelayMsToNextTask() const { + PERFETTO_DCHECK(RunsTasksOnCurrentThread()); + if (delayed_tasks_.empty()) { + return -1; + } + TimeMillis now = + GetWallTimeMs() + + TimeMillis(advanced_time_for_testing_.load(std::memory_order_relaxed)); + TimeMillis deadline = delayed_tasks_.back().time; + if (deadline <= now) { + return 0; + } + return static_cast<int>((deadline - now).count()); +} + +std::shared_ptr<LockFreeTaskRunner::Slab> LockFreeTaskRunner::AllocNewSlab() { + Slab* free_slab = free_slab_.exchange(nullptr); + + auto deleter = [this](Slab* s) { + // Reset the slab. + s->~Slab(); + new (s) Slab(); + + Slab* null_slab = nullptr; + if (!free_slab_.compare_exchange_strong(null_slab, s)) { + slabs_freed_.fetch_add(1, std::memory_order_relaxed); + delete s; + } + }; + + if (free_slab) { + return std::shared_ptr<Slab>(free_slab, deleter); + } + slabs_allocated_.fetch_add(1, std::memory_order_relaxed); + return std::shared_ptr<Slab>(new Slab(), deleter); +} + +void LockFreeTaskRunner::PostDelayedTask(std::function<void()> task, + uint32_t delay_ms) { + if (!RunsTasksOnCurrentThread()) { + PostTask(std::bind(&LockFreeTaskRunner::PostDelayedTask, this, + std::move(task), delay_ms)); + return; + } + PERFETTO_DCHECK(RunsTasksOnCurrentThread()); + TimeMillis runtime = + GetWallTimeMs() + TimeMillis(delay_ms) + + TimeMillis(advanced_time_for_testing_.load(std::memory_order_relaxed)); + delayed_tasks_.insert( + DelayedTask{runtime, next_delayed_task_seq_++, std::move(task)}); +} + +void LockFreeTaskRunner::AdvanceTimeForTesting(uint32_t ms) { + PERFETTO_DCHECK(RunsTasksOnCurrentThread()); + advanced_time_for_testing_.fetch_add(ms); + WakeUp(); +} + +void LockFreeTaskRunner::PostFileDescriptorWatches( + uint64_t windows_wait_result) { + PERFETTO_DCHECK(RunsTasksOnCurrentThread()); + for (size_t i = 0; i < poll_fds_.size(); i++) { +#if PERFETTO_BUILDFLAG(PERFETTO_OS_WIN) + const PlatformHandle handle = poll_fds_[i]; + // |windows_wait_result| is the result of WaitForMultipleObjects() call. If + // one of the objects was signalled, it will have a value between + // [0, poll_fds_.size()]. + if (i != windows_wait_result && + WaitForSingleObject(handle, 0) != WAIT_OBJECT_0) { + continue; + } +#else + base::ignore_result(windows_wait_result); + const PlatformHandle handle = poll_fds_[i].fd; + if (!(poll_fds_[i].revents & (POLLIN | POLLHUP))) + continue; + poll_fds_[i].revents = 0; +#endif + + // The wake-up event is handled inline to avoid an infinite recursion of + // posted tasks. + if (handle == wakeup_event_.fd()) { + wakeup_event_.Clear(); + continue; + } + + // Binding to |this| is safe since we are the only object executing the + // task. + PostTask( + std::bind(&LockFreeTaskRunner::RunFileDescriptorWatch, this, handle)); + + // Flag the task as pending. +#if PERFETTO_BUILDFLAG(PERFETTO_OS_WIN) + // On Windows this is done by marking the WatchTask entry as pending. This + // is more expensive than Linux as requires rebuilding the |poll_fds_| + // vector on each call. There doesn't seem to be a good alternative though. + auto it = watch_tasks_.find(handle); + PERFETTO_CHECK(it != watch_tasks_.end()); + PERFETTO_DCHECK(!it->second.pending); + it->second.pending = true; +#else + // On UNIX systems instead, we just make the fd negative while its task is + // pending. This makes poll(2) ignore the fd. + PERFETTO_DCHECK(poll_fds_[i].fd >= 0); + poll_fds_[i].fd = -poll_fds_[i].fd; +#endif + } +} + +void LockFreeTaskRunner::RunFileDescriptorWatch(PlatformHandle fd) { + PERFETTO_DCHECK(RunsTasksOnCurrentThread()); + + std::function<void()> task; + auto it = watch_tasks_.find(fd); + if (it == watch_tasks_.end()) + return; + WatchTask& watch_task = it->second; + + // Make poll(2) pay attention to the fd again. Since another thread may have + // updated this watch we need to refresh the set first. + // TODO check if this still holds. + UpdateWatchTasks(); + +#if PERFETTO_BUILDFLAG(PERFETTO_OS_WIN) + // On Windows we manually track the presence of outstanding tasks for the + // watch. The UpdateWatchTasksLocked() in the Run() loop will re-add the + // task to the |poll_fds_| vector. + PERFETTO_DCHECK(watch_task.pending); + watch_task.pending = false; +#else + size_t fd_index = watch_task.poll_fd_index; + PERFETTO_DCHECK(fd_index < poll_fds_.size()); + PERFETTO_DCHECK(::abs(poll_fds_[fd_index].fd) == fd); + poll_fds_[fd_index].fd = fd; +#endif + task = watch_task.callback; + errno = 0; + RunTaskWithWatchdogGuard(task); +} + +void LockFreeTaskRunner::UpdateWatchTasks() { + PERFETTO_DCHECK(RunsTasksOnCurrentThread()); +#if !PERFETTO_BUILDFLAG(PERFETTO_OS_WIN) + if (!watch_tasks_changed_) + return; + watch_tasks_changed_ = false; +#endif + poll_fds_.clear(); + for (auto& it : watch_tasks_) { + PlatformHandle handle = it.first; + WatchTask& watch_task = it.second; +#if PERFETTO_BUILDFLAG(PERFETTO_OS_WIN) + if (!watch_task.pending) + poll_fds_.push_back(handle); +#else + watch_task.poll_fd_index = poll_fds_.size(); + poll_fds_.push_back({handle, POLLIN | POLLHUP, 0}); +#endif + } +} + +void LockFreeTaskRunner::AddFileDescriptorWatch(PlatformHandle fd, + std::function<void()> task) { + PERFETTO_DCHECK(PlatformHandleChecker::IsValid(fd)); + + if (!RunsTasksOnCurrentThread()) { + PostTask(std::bind(&LockFreeTaskRunner::AddFileDescriptorWatch, this, fd, + std::move(task))); + return; + } + + PERFETTO_DCHECK(RunsTasksOnCurrentThread()); + PERFETTO_DCHECK(!watch_tasks_.count(fd)); + WatchTask& watch_task = watch_tasks_[fd]; + watch_task.callback = std::move(task); +#if PERFETTO_BUILDFLAG(PERFETTO_OS_WIN) + watch_task.pending = false; +#else + watch_task.poll_fd_index = SIZE_MAX; +#endif + watch_tasks_changed_ = true; +} + +void LockFreeTaskRunner::RemoveFileDescriptorWatch(PlatformHandle fd) { + if (!RunsTasksOnCurrentThread()) { + PostTask( + std::bind(&LockFreeTaskRunner::RemoveFileDescriptorWatch, this, fd)); + return; + } + + PERFETTO_DCHECK(RunsTasksOnCurrentThread()); + PERFETTO_DCHECK(watch_tasks_.count(fd)); + watch_tasks_.erase(fd); + watch_tasks_changed_ = true; +} + +bool LockFreeTaskRunner::RunsTasksOnCurrentThread() const { + return run_task_thread_id_ == std::this_thread::get_id(); +} + +} // namespace base +} // namespace perfetto \ No newline at end of file
diff --git a/src/base/task_runner_unittest.cc b/src/base/task_runner_unittest.cc index 060cf43..06dc794 100644 --- a/src/base/task_runner_unittest.cc +++ b/src/base/task_runner_unittest.cc
@@ -16,27 +16,95 @@ #include "perfetto/base/build_config.h" -#include "perfetto/ext/base/unix_task_runner.h" +#include "perfetto/ext/base/lock_free_task_runner.h" +#include <random> #include <thread> #include "perfetto/ext/base/event_fd.h" #include "perfetto/ext/base/file_utils.h" #include "perfetto/ext/base/pipe.h" #include "perfetto/ext/base/scoped_file.h" +#include "perfetto/ext/base/unix_task_runner.h" #include "perfetto/ext/base/utils.h" +#include "perfetto/ext/base/waitable_event.h" #include "test/gtest_and_gmock.h" namespace perfetto { namespace base { namespace { +template <typename TaskRunnerType> class TaskRunnerTest : public ::testing::Test { public: - UnixTaskRunner task_runner; + TaskRunnerType task_runner; }; -TEST_F(TaskRunnerTest, PostImmediateTask) { +struct TaskRunnerTestNames { + template <typename T> + static std::string GetName(int) { + if (std::is_same<T, UnixTaskRunner>::value) + return "UnixTaskRunner"; + if (std::is_same<T, LockFreeTaskRunner>::value) + return "LockFreeTaskRunner"; + return testing::internal::GetTypeName<T>(); + } +}; + +using TaskRunnerTypes = ::testing::Types<UnixTaskRunner, LockFreeTaskRunner>; +TYPED_TEST_SUITE(TaskRunnerTest, TaskRunnerTypes, TaskRunnerTestNames); + +TYPED_TEST(TaskRunnerTest, QuitImmediately) { + this->task_runner.PostTask([&] { this->task_runner.Quit(); }); + this->task_runner.Run(); +} + +TYPED_TEST(TaskRunnerTest, OneTaskFromAnotherThread) { + WaitableEvent task_runner_started; + std::thread t1([&] { + task_runner_started.Wait(); + this->task_runner.PostTask([&] { this->task_runner.Quit(); }); + }); + this->task_runner.PostTask([&] { task_runner_started.Notify(); }); + this->task_runner.Run(); + t1.join(); +} + +TYPED_TEST(TaskRunnerTest, PostTaskSimple) { + std::string str; + this->task_runner.PostTask([&str] { str.append("a"); }); + this->task_runner.PostTask([&str] { str.append("b"); }); + this->task_runner.PostTask([&str] { str.append("c"); }); + this->task_runner.PostTask([&str, tr = &this->task_runner] { + tr->PostTask([&str] { str.append("d"); }); + tr->PostTask([&str] { str.append("e"); }); + tr->PostTask([&str] { str.append("f"); }); + tr->PostTask([tr] { tr->Quit(); }); + }); + this->task_runner.Run(); + EXPECT_EQ(str, "abcdef"); +} + +TYPED_TEST(TaskRunnerTest, ManyTasksPostedBeforeRun) { + constexpr size_t kNumTasks = 10000; + std::function<void()> post_another; + size_t last_task_id = 0; + auto task = [&](size_t n) { + ASSERT_EQ(last_task_id, n - 1); + last_task_id = n; + if (n == kNumTasks) + this->task_runner.Quit(); + }; + + for (size_t i = 1; i <= kNumTasks; i++) { + this->task_runner.PostTask(std::bind(task, i)); + } + + this->task_runner.Run(); + EXPECT_EQ(last_task_id, kNumTasks); +} + +TYPED_TEST(TaskRunnerTest, PostImmediateTask) { auto& task_runner = this->task_runner; int counter = 0; task_runner.PostTask([&counter] { counter = (counter << 4) | 1; }); @@ -48,19 +116,36 @@ EXPECT_EQ(0x1234, counter); } -TEST_F(TaskRunnerTest, PostDelayedTask) { - auto& task_runner = this->task_runner; - int counter = 0; - task_runner.PostDelayedTask([&counter] { counter = (counter << 4) | 1; }, 5); - task_runner.PostDelayedTask([&counter] { counter = (counter << 4) | 2; }, 10); - task_runner.PostDelayedTask([&counter] { counter = (counter << 4) | 3; }, 15); - task_runner.PostDelayedTask([&counter] { counter = (counter << 4) | 4; }, 15); - task_runner.PostDelayedTask([&task_runner] { task_runner.Quit(); }, 20); - task_runner.Run(); - EXPECT_EQ(0x1234, counter); +TYPED_TEST(TaskRunnerTest, PostDelayedTask) { + std::vector<int> executed_tasks; + this->task_runner.PostDelayedTask( + [&] { + executed_tasks.push_back(5); + this->task_runner.Quit(); + }, + 100); + this->task_runner.PostDelayedTask([&] { executed_tasks.push_back(2); }, 20); + this->task_runner.PostDelayedTask([&] { executed_tasks.push_back(3); }, 20); + this->task_runner.PostDelayedTask([&] { executed_tasks.push_back(4); }, 80); + this->task_runner.PostDelayedTask([&] { executed_tasks.push_back(1); }, 10); + this->task_runner.PostTask([&] { + this->task_runner.AdvanceTimeForTesting(10); // Executes task 1. + }); + this->task_runner.PostTask([&] { + this->task_runner.AdvanceTimeForTesting(10); // Executes tasks 2 and 3. + }); + this->task_runner.PostTask([&] { + this->task_runner.AdvanceTimeForTesting(60); // Executes task 4. + }); + this->task_runner.PostTask([&] { + this->task_runner.AdvanceTimeForTesting(20); // Executes task 5. + }); + this->task_runner.Run(); + + EXPECT_THAT(executed_tasks, ::testing::ElementsAre(1, 2, 3, 4, 5)); } -TEST_F(TaskRunnerTest, PostImmediateTaskFromTask) { +TYPED_TEST(TaskRunnerTest, PostImmediateTaskFromTask) { auto& task_runner = this->task_runner; task_runner.PostTask([&task_runner] { task_runner.PostTask([&task_runner] { task_runner.Quit(); }); @@ -68,7 +153,7 @@ task_runner.Run(); } -TEST_F(TaskRunnerTest, PostDelayedTaskFromTask) { +TYPED_TEST(TaskRunnerTest, PostDelayedTaskFromTask) { auto& task_runner = this->task_runner; task_runner.PostTask([&task_runner] { task_runner.PostDelayedTask([&task_runner] { task_runner.Quit(); }, 10); @@ -76,7 +161,7 @@ task_runner.Run(); } -TEST_F(TaskRunnerTest, PostImmediateTaskFromOtherThread) { +TYPED_TEST(TaskRunnerTest, PostImmediateTaskFromOtherThread) { auto& task_runner = this->task_runner; ThreadChecker thread_checker; int counter = 0; @@ -95,7 +180,7 @@ EXPECT_EQ(0x1234, counter); } -TEST_F(TaskRunnerTest, PostDelayedTaskFromOtherThread) { +TYPED_TEST(TaskRunnerTest, PostDelayedTaskFromOtherThread) { auto& task_runner = this->task_runner; std::thread thread([&task_runner] { task_runner.PostDelayedTask([&task_runner] { task_runner.Quit(); }, 10); @@ -104,7 +189,7 @@ thread.join(); } -TEST_F(TaskRunnerTest, AddFileDescriptorWatch) { +TYPED_TEST(TaskRunnerTest, AddFileDescriptorWatch) { auto& task_runner = this->task_runner; EventFd evt; task_runner.AddFileDescriptorWatch(evt.fd(), @@ -113,7 +198,7 @@ task_runner.Run(); } -TEST_F(TaskRunnerTest, RemoveFileDescriptorWatch) { +TYPED_TEST(TaskRunnerTest, RemoveFileDescriptorWatch) { auto& task_runner = this->task_runner; EventFd evt; evt.Notify(); @@ -128,7 +213,7 @@ EXPECT_FALSE(watch_ran); } -TEST_F(TaskRunnerTest, RemoveFileDescriptorWatchFromTask) { +TYPED_TEST(TaskRunnerTest, RemoveFileDescriptorWatchFromTask) { auto& task_runner = this->task_runner; EventFd evt; evt.Notify(); @@ -145,7 +230,7 @@ EXPECT_FALSE(watch_ran); } -TEST_F(TaskRunnerTest, AddFileDescriptorWatchFromAnotherWatch) { +TYPED_TEST(TaskRunnerTest, AddFileDescriptorWatchFromAnotherWatch) { auto& task_runner = this->task_runner; EventFd evt; EventFd evt2; @@ -159,7 +244,7 @@ task_runner.Run(); } -TEST_F(TaskRunnerTest, RemoveFileDescriptorWatchFromAnotherWatch) { +TYPED_TEST(TaskRunnerTest, RemoveFileDescriptorWatchFromAnotherWatch) { auto& task_runner = this->task_runner; EventFd evt; EventFd evt2; @@ -179,7 +264,7 @@ EXPECT_FALSE(watch_ran); } -TEST_F(TaskRunnerTest, ReplaceFileDescriptorWatchFromAnotherWatch) { +TYPED_TEST(TaskRunnerTest, ReplaceFileDescriptorWatchFromAnotherWatch) { auto& task_runner = this->task_runner; EventFd evt; EventFd evt2; @@ -200,7 +285,7 @@ EXPECT_FALSE(watch_ran); } -TEST_F(TaskRunnerTest, AddFileDescriptorWatchFromAnotherThread) { +TYPED_TEST(TaskRunnerTest, AddFileDescriptorWatchFromAnotherThread) { auto& task_runner = this->task_runner; EventFd evt; evt.Notify(); @@ -213,7 +298,7 @@ thread.join(); } -TEST_F(TaskRunnerTest, FileDescriptorWatchWithMultipleEvents) { +TYPED_TEST(TaskRunnerTest, FileDescriptorWatchWithMultipleEvents) { auto& task_runner = this->task_runner; EventFd evt; evt.Notify(); @@ -232,7 +317,7 @@ task_runner.Run(); } -TEST_F(TaskRunnerTest, PostManyDelayedTasks) { +TYPED_TEST(TaskRunnerTest, PostManyDelayedTasks) { // Check that PostTask doesn't start failing if there are too many scheduled // wake-ups. auto& task_runner = this->task_runner; @@ -242,7 +327,66 @@ task_runner.Run(); } -TEST_F(TaskRunnerTest, RunAgain) { +TYPED_TEST(TaskRunnerTest, PostTaskReentrant) { + constexpr int kMaxTasks = 10000; + int next_task_id = 0; + int tasks_executed = 0; + std::vector<int> execution_log; + std::map<int, int> parent_of; // task_id -> parent_id + + std::function<void(int)> post_recursive; + post_recursive = [&](int parent_id) { + int task_id = next_task_id++; + if (task_id >= kMaxTasks) { + return; + } + parent_of[task_id] = parent_id; + this->task_runner.PostTask([&, task_id] { + { + execution_log.push_back(task_id); + } + + // Spawn sub-tasks. + std::minstd_rand0 rnd(static_cast<uint32_t>(task_id)); + int num_children = std::uniform_int_distribution<int>(0, 5)(rnd); + for (int i = 0; i < num_children; ++i) { + post_recursive(task_id); + } + + if (tasks_executed++ == kMaxTasks - 1) { + this->task_runner.Quit(); + } + }); + }; + + // Start with a few top-level tasks. + post_recursive(-1); + post_recursive(-1); + post_recursive(-1); + + this->task_runner.Run(); + + EXPECT_EQ(tasks_executed, kMaxTasks); + EXPECT_EQ(execution_log.size(), static_cast<size_t>(kMaxTasks)); + + // Verify that parents are always executed before children. + std::map<int, size_t> execution_pos; + for (size_t i = 0; i < execution_log.size(); ++i) { + execution_pos[execution_log[i]] = i; + } + + for (const auto& it : parent_of) { + int task_id = it.first; + int parent_id = it.second; + if (parent_id == -1) + continue; + ASSERT_TRUE(execution_pos.count(task_id)); + ASSERT_TRUE(execution_pos.count(parent_id)); + EXPECT_LT(execution_pos[parent_id], execution_pos[task_id]); + } +} + +TYPED_TEST(TaskRunnerTest, RunAgain) { auto& task_runner = this->task_runner; int counter = 0; task_runner.PostTask([&task_runner, &counter] { @@ -258,31 +402,33 @@ EXPECT_EQ(2, counter); } -void RepeatingTask(UnixTaskRunner* task_runner) { - task_runner->PostTask(std::bind(&RepeatingTask, task_runner)); +template <typename T> +void RepeatingTask(T* task_runner) { + task_runner->PostTask(std::bind(&RepeatingTask<T>, task_runner)); } -TEST_F(TaskRunnerTest, FileDescriptorWatchesNotStarved) { +TYPED_TEST(TaskRunnerTest, FileDescriptorWatchesNotStarved) { auto& task_runner = this->task_runner; EventFd evt; evt.Notify(); - task_runner.PostTask(std::bind(&RepeatingTask, &task_runner)); + task_runner.PostTask(std::bind(&RepeatingTask<TypeParam>, &task_runner)); task_runner.AddFileDescriptorWatch(evt.fd(), [&task_runner] { task_runner.Quit(); }); task_runner.Run(); } -void CountdownTask(UnixTaskRunner* task_runner, int* counter) { +template <typename T> +void CountdownTask(T* task_runner, int* counter) { if (!--(*counter)) { task_runner->Quit(); return; } - task_runner->PostDelayedTask(std::bind(&CountdownTask, task_runner, counter), - 1); + task_runner->PostDelayedTask( + std::bind(&CountdownTask<T>, task_runner, counter), 1); } -TEST_F(TaskRunnerTest, NoDuplicateFileDescriptorWatchCallbacks) { +TYPED_TEST(TaskRunnerTest, NoDuplicateFileDescriptorWatchCallbacks) { auto& task_runner = this->task_runner; EventFd evt; evt.Notify(); @@ -294,11 +440,12 @@ evt.Clear(); watch_called = true; }); - task_runner.PostTask(std::bind(&CountdownTask, &task_runner, &counter)); + task_runner.PostTask( + std::bind(&CountdownTask<TypeParam>, &task_runner, &counter)); task_runner.Run(); } -TEST_F(TaskRunnerTest, ReplaceFileDescriptorWatchFromOtherThread) { +TYPED_TEST(TaskRunnerTest, ReplaceFileDescriptorWatchFromOtherThread) { auto& task_runner = this->task_runner; EventFd evt; evt.Notify(); @@ -318,10 +465,14 @@ thread.join(); } -TEST_F(TaskRunnerTest, IsIdleForTesting) { +TYPED_TEST(TaskRunnerTest, IsIdleForTesting) { auto& task_runner = this->task_runner; + // This first task fails because by the time we get to Run(), there is another + // one (below) queued up already. task_runner.PostTask( [&task_runner] { EXPECT_FALSE(task_runner.IsIdleForTesting()); }); + + // This one succeeds because it's the last one and there is no further task. task_runner.PostTask([&task_runner] { EXPECT_TRUE(task_runner.IsIdleForTesting()); task_runner.Quit(); @@ -329,12 +480,44 @@ task_runner.Run(); } -TEST_F(TaskRunnerTest, RunsTasksOnCurrentThread) { +// Covers a corner cases that TestTaskRunner::RunUntilIdle relies on: +// IsIdleForTesting() is supposed to check for all type of upcoming tasks, +// including FD watches. This is to check that the TaskRunner implementation +// doesn't have off-by one behaviours where the FD watch is only observed on +// the next task. +// It's debatable on whether we need to preserve this behaviour in production +// code, if we assume FDs are unpredictable events and we shouldn't expect +// timing correlations with current tasks. +TYPED_TEST(TaskRunnerTest, IsIdleForTesting_WithFd) { + auto& task_runner = this->task_runner; + EventFd efd; + bool efd_observed = false; + + // This will fail the IsIdleForTesting() check because by the time we get + // to run, the eventfd is notified. + task_runner.PostTask( + [&task_runner] { EXPECT_FALSE(task_runner.IsIdleForTesting()); }); + + task_runner.AddFileDescriptorWatch(efd.fd(), [&] { + efd.Clear(); + efd_observed = true; + task_runner.PostTask([&task_runner] { + EXPECT_TRUE(task_runner.IsIdleForTesting()); + task_runner.Quit(); + }); + }); + efd.Notify(); + + task_runner.Run(); + EXPECT_TRUE(efd_observed); +} + +TYPED_TEST(TaskRunnerTest, RunsTasksOnCurrentThread) { auto& main_tr = this->task_runner; EXPECT_TRUE(main_tr.RunsTasksOnCurrentThread()); std::thread thread([&main_tr] { - typename std::remove_reference<decltype(main_tr)>::type second_tr; + TypeParam second_tr; second_tr.PostTask([&main_tr, &second_tr] { EXPECT_FALSE(main_tr.RunsTasksOnCurrentThread()); EXPECT_TRUE(second_tr.RunsTasksOnCurrentThread()); @@ -347,7 +530,7 @@ thread.join(); } -TEST_F(TaskRunnerTest, FileDescriptorWatchFairness) { +TYPED_TEST(TaskRunnerTest, FileDescriptorWatchFairness) { auto& task_runner = this->task_runner; EventFd evt[5]; std::map<PlatformHandle, int /*num_tasks*/> num_tasks; @@ -376,7 +559,7 @@ #if !PERFETTO_BUILDFLAG(PERFETTO_OS_WIN) // This tests UNIX-specific behavior on pipe closure. -TEST_F(TaskRunnerTest, FileDescriptorClosedEvent) { +TYPED_TEST(TaskRunnerTest, FileDescriptorClosedEvent) { auto& task_runner = this->task_runner; Pipe pipe = Pipe::Create(); pipe.wr.reset(); @@ -387,6 +570,87 @@ #endif +TYPED_TEST(TaskRunnerTest, MultiThreadedStress) { + constexpr size_t kNumThreads = 4; + constexpr size_t kNumTasksPerThread = 1000; + constexpr size_t kTotalTasks = kNumThreads * kNumTasksPerThread; + + struct ThreadState { + size_t next_gen = 0; + size_t next_expected = 0; + std::minstd_rand0 rnd{}; + std::thread thread; + }; + + std::array<ThreadState, kNumThreads> threads{}; + std::atomic<size_t> tasks_posted{0}; + std::atomic<size_t> tasks_executed{0}; + + std::function<void(size_t, size_t)> worker_fn; + worker_fn = [&](size_t thread_idx, size_t expected_id) { + ThreadState& thd = threads[thread_idx]; + + ASSERT_EQ(expected_id, thd.next_expected); + ++thd.next_expected; + tasks_executed.fetch_add(1, std::memory_order_relaxed); + + int num_subtasks = std::uniform_int_distribution<int>(1, 32)(thd.rnd); + for (int i = 0; i < num_subtasks; ++i) { + auto total_num = tasks_posted.fetch_add(1, std::memory_order_relaxed); + if (total_num >= kTotalTasks) { + if (total_num == kTotalTasks) { + this->task_runner.PostTask([&] { this->task_runner.Quit(); }); + } + return; + } + size_t next_id = ++thd.next_gen; + this->task_runner.PostTask(std::bind(worker_fn, thread_idx, next_id)); + } + std::this_thread::yield(); + }; + + tasks_posted.store(kNumThreads); // Account for the initial calls. + for (size_t i = 0; i < kNumThreads; ++i) { + threads[i].thread = std::thread(std::bind(worker_fn, i, 0)); + } + + this->task_runner.Run(); + + for (auto& thread_state : threads) { + thread_state.thread.join(); + } + EXPECT_EQ(tasks_executed.load(), kTotalTasks); +} + +// [LockFreeTaskRunner-only] Covers the slab allocator logic, ensuring that +// slabs are recycled properly and are not leaked. It run tasks in bursts +// (one tasks spwaning up to kBurstMax subtasks), catches up, then repeats. +TEST(TaskRunnerTest, NoSlabLeaks) { + constexpr int kMaxTasks = 10000; + constexpr int kBurstMax = LockFreeTaskRunner::kSlabSize - 2; + size_t tasks_posted = 0; + std::function<void()> task_fn; + std::minstd_rand0 rnd; + LockFreeTaskRunner task_runner; + + task_fn = [&] { + int burst_count = std::uniform_int_distribution<int>(1, kBurstMax)(rnd); + for (int i = 0; i < burst_count; i++, tasks_posted++) { + task_runner.PostTask([] {}); + } + if (tasks_posted < kMaxTasks) { + task_runner.PostTask(task_fn); + } else { + task_runner.PostTask([&] { task_runner.Quit(); }); + } + }; + + task_fn(); + task_runner.Run(); + + EXPECT_LE(task_runner.slabs_allocated(), 2u); +} + } // namespace } // namespace base } // namespace perfetto
diff --git a/src/base/test/test_task_runner.h b/src/base/test/test_task_runner.h index 142970d..e6ba703 100644 --- a/src/base/test/test_task_runner.h +++ b/src/base/test/test_task_runner.h
@@ -24,8 +24,8 @@ #include "perfetto/base/build_config.h" #include "perfetto/base/compiler.h" +#include "perfetto/ext/base/lock_free_task_runner.h" #include "perfetto/ext/base/thread_checker.h" -#include "perfetto/ext/base/unix_task_runner.h" namespace perfetto { namespace base { @@ -63,7 +63,7 @@ std::string pending_checkpoint_; std::map<std::string, bool> checkpoints_; - base::UnixTaskRunner task_runner_; + base::MaybeLockFreeTaskRunner task_runner_; ThreadChecker thread_checker_; };
diff --git a/src/base/thread_task_runner.cc b/src/base/thread_task_runner.cc index 2669477..f58d008 100644 --- a/src/base/thread_task_runner.cc +++ b/src/base/thread_task_runner.cc
@@ -24,8 +24,8 @@ #include <thread> #include "perfetto/base/logging.h" +#include "perfetto/ext/base/lock_free_task_runner.h" #include "perfetto/ext/base/thread_utils.h" -#include "perfetto/ext/base/unix_task_runner.h" #if PERFETTO_BUILDFLAG(PERFETTO_OS_LINUX_BUT_NOT_QNX) || \ PERFETTO_BUILDFLAG(PERFETTO_OS_ANDROID) @@ -48,7 +48,6 @@ ThreadTaskRunner::~ThreadTaskRunner() { if (task_runner_) { - PERFETTO_CHECK(!task_runner_->QuitCalled()); task_runner_->Quit(); PERFETTO_DCHECK(thread_.joinable()); @@ -61,8 +60,8 @@ std::mutex init_lock; std::condition_variable init_cv; - std::function<void(UnixTaskRunner*)> initializer = - [this, &init_lock, &init_cv](UnixTaskRunner* task_runner) { + std::function<void(MaybeLockFreeTaskRunner*)> initializer = + [this, &init_lock, &init_cv](MaybeLockFreeTaskRunner* 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 @@ -80,12 +79,12 @@ } void ThreadTaskRunner::RunTaskThread( - std::function<void(UnixTaskRunner*)> initializer) { + std::function<void(MaybeLockFreeTaskRunner*)> initializer) { if (!name_.empty()) { base::MaybeSetThreadName(name_); } - UnixTaskRunner task_runner; + MaybeLockFreeTaskRunner task_runner; task_runner.PostTask(std::bind(std::move(initializer), &task_runner)); task_runner.Run(); }
diff --git a/src/base/thread_task_runner_unittest.cc b/src/base/thread_task_runner_unittest.cc index ae7dacd..b137672 100644 --- a/src/base/thread_task_runner_unittest.cc +++ b/src/base/thread_task_runner_unittest.cc
@@ -18,6 +18,7 @@ #include <thread> +#include "perfetto/ext/base/lock_free_task_runner.h" #include "perfetto/ext/base/no_destructor.h" #include "perfetto/ext/base/thread_checker.h" #include "test/gtest_and_gmock.h" @@ -66,7 +67,7 @@ NoDestructor<ThreadTaskRunner> ttr{ThreadTaskRunner::CreateAndStart()}; ThreadTaskRunner& task_runner = ttr.ref(); - UnixTaskRunner* runner_ptr = task_runner.get(); + MaybeLockFreeTaskRunner* runner_ptr = task_runner.get(); EXPECT_NE(runner_ptr, nullptr); ThreadChecker thread_checker; @@ -114,9 +115,9 @@ 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. +// Checks that the still-pending tasks (and therefore the +// MaybeLockFreeTaskRunner 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(); @@ -132,7 +133,7 @@ // * 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. + // the destruction of MaybeLockFreeTaskRunner. task_runner.get()->PostDelayedTask(DestructorThreadChecker(thread_checker), 100 * 1000 /*ms*/); atomic_flag_ = true;
diff --git a/src/perfetto_cmd/perfetto_cmd.h b/src/perfetto_cmd/perfetto_cmd.h index e821fdc..38c3b5e 100644 --- a/src/perfetto_cmd/perfetto_cmd.h +++ b/src/perfetto_cmd/perfetto_cmd.h
@@ -27,10 +27,10 @@ #include "perfetto/base/build_config.h" #include "perfetto/ext/base/event_fd.h" +#include "perfetto/ext/base/lock_free_task_runner.h" #include "perfetto/ext/base/pipe.h" #include "perfetto/ext/base/scoped_file.h" #include "perfetto/ext/base/thread_task_runner.h" -#include "perfetto/ext/base/unix_task_runner.h" #include "perfetto/ext/base/uuid.h" #include "perfetto/ext/base/weak_ptr.h" #include "perfetto/ext/tracing/core/basic_types.h" @@ -147,7 +147,7 @@ void LogTriggerEvents(PerfettoTriggerAtom atom, const std::vector<std::string>& trigger_names); - base::UnixTaskRunner task_runner_; + base::MaybeLockFreeTaskRunner task_runner_; std::unique_ptr<perfetto::TracingService::ConsumerEndpoint> consumer_endpoint_;
diff --git a/src/perfetto_cmd/trigger_perfetto.cc b/src/perfetto_cmd/trigger_perfetto.cc index c1711d4..36c303c 100644 --- a/src/perfetto_cmd/trigger_perfetto.cc +++ b/src/perfetto_cmd/trigger_perfetto.cc
@@ -19,7 +19,7 @@ #include "perfetto/base/logging.h" #include "perfetto/ext/base/getopt.h" -#include "perfetto/ext/base/unix_task_runner.h" +#include "perfetto/ext/base/lock_free_task_runner.h" #include "perfetto/ext/traced/traced.h" #include "src/android_stats/statsd_logging_helper.h" #include "src/perfetto_cmd/trigger_producer.h" @@ -87,7 +87,7 @@ } bool finished_with_success = false; - base::UnixTaskRunner task_runner; + base::MaybeLockFreeTaskRunner task_runner; TriggerProducer producer( &task_runner, [&task_runner, &finished_with_success](bool success) {
diff --git a/src/profiling/memory/client_api_factory_standalone.cc b/src/profiling/memory/client_api_factory_standalone.cc index e9d5ae6..70e7ca3 100644 --- a/src/profiling/memory/client_api_factory_standalone.cc +++ b/src/profiling/memory/client_api_factory_standalone.cc
@@ -17,9 +17,9 @@ #include "src/profiling/memory/client_api_factory.h" #include "perfetto/base/logging.h" +#include "perfetto/ext/base/lock_free_task_runner.h" #include "perfetto/ext/base/scoped_file.h" #include "perfetto/ext/base/unix_socket.h" -#include "perfetto/ext/base/unix_task_runner.h" #include "perfetto/ext/base/utils.h" #include "perfetto/ext/base/watchdog.h" #include "perfetto/heap_profile.h" @@ -144,7 +144,7 @@ srv_sock.SetBlocking(false); - base::UnixTaskRunner task_runner; + base::MaybeLockFreeTaskRunner task_runner; base::Watchdog::GetInstance()->Start(); // crash on exceedingly long tasks HeapprofdProducer producer(HeapprofdMode::kChild, &task_runner, /* exit_when_done= */ false);
diff --git a/src/profiling/perf/traced_perf.cc b/src/profiling/perf/traced_perf.cc index 6c530cd..a00d048 100644 --- a/src/profiling/perf/traced_perf.cc +++ b/src/profiling/perf/traced_perf.cc
@@ -21,7 +21,7 @@ #include "perfetto/ext/base/file_utils.h" #include "perfetto/ext/base/getopt.h" -#include "perfetto/ext/base/unix_task_runner.h" +#include "perfetto/ext/base/lock_free_task_runner.h" #include "perfetto/ext/base/version.h" #include "perfetto/tracing/default_socket.h" #include "src/profiling/perf/perf_producer.h" @@ -81,7 +81,7 @@ base::Daemonize([] { return 0; }); } - base::UnixTaskRunner task_runner; + base::MaybeLockFreeTaskRunner task_runner; // TODO(rsavitski): support standalone --root or similar on android. #if PERFETTO_BUILDFLAG(PERFETTO_ANDROID_BUILD)
diff --git a/src/trace_processor/rpc/httpd.cc b/src/trace_processor/rpc/httpd.cc index 441b7f8..6ee79cf 100644 --- a/src/trace_processor/rpc/httpd.cc +++ b/src/trace_processor/rpc/httpd.cc
@@ -26,9 +26,9 @@ #include "perfetto/base/logging.h" #include "perfetto/base/status.h" #include "perfetto/ext/base/http/http_server.h" +#include "perfetto/ext/base/lock_free_task_runner.h" #include "perfetto/ext/base/string_utils.h" #include "perfetto/ext/base/string_view.h" -#include "perfetto/ext/base/unix_task_runner.h" #include "perfetto/protozero/scattered_heap_buffer.h" #include "perfetto/trace_processor/trace_processor.h" #include "src/trace_processor/rpc/httpd.h" @@ -66,7 +66,7 @@ static void ServeHelpPage(const base::HttpRequest&); Rpc global_trace_processor_rpc_; - base::UnixTaskRunner task_runner_; + base::MaybeLockFreeTaskRunner task_runner_; base::HttpServer http_srv_; };
diff --git a/src/traced/probes/probes.cc b/src/traced/probes/probes.cc index 5704a3c..f2ee006 100644 --- a/src/traced/probes/probes.cc +++ b/src/traced/probes/probes.cc
@@ -22,7 +22,7 @@ #include "perfetto/base/logging.h" #include "perfetto/ext/base/file_utils.h" #include "perfetto/ext/base/getopt.h" -#include "perfetto/ext/base/unix_task_runner.h" +#include "perfetto/ext/base/lock_free_task_runner.h" #include "perfetto/ext/base/utils.h" #include "perfetto/ext/base/version.h" #include "perfetto/ext/traced/traced.h" @@ -114,7 +114,7 @@ PERFETTO_DCHECK(res == 0); } - base::UnixTaskRunner task_runner; + base::MaybeLockFreeTaskRunner task_runner; ProbesProducer producer; // If the TRACED_PROBES_NOTIFY_FD env var is set, write 1 and close the FD, // when all data sources have been registered. This is used for //src/tracebox
diff --git a/src/traced/service/service.cc b/src/traced/service/service.cc index f006dfe..3b804ca 100644 --- a/src/traced/service/service.cc +++ b/src/traced/service/service.cc
@@ -21,9 +21,9 @@ #include "perfetto/ext/base/android_utils.h" #include "perfetto/ext/base/file_utils.h" #include "perfetto/ext/base/getopt.h" +#include "perfetto/ext/base/lock_free_task_runner.h" #include "perfetto/ext/base/string_utils.h" #include "perfetto/ext/base/unix_socket.h" -#include "perfetto/ext/base/unix_task_runner.h" #include "perfetto/ext/base/utils.h" #include "perfetto/ext/base/version.h" #include "perfetto/ext/base/watchdog.h" @@ -130,7 +130,7 @@ base::Daemonize([] { return 0; }); } - base::UnixTaskRunner task_runner; + base::MaybeLockFreeTaskRunner task_runner; std::unique_ptr<ServiceIPCHost> svc; TracingService::InitOpts init_opts = {}; #if PERFETTO_BUILDFLAG(PERFETTO_ZLIB)
diff --git a/src/traced_relay/relay_service_main.cc b/src/traced_relay/relay_service_main.cc index 9d17269..8f49d7e 100644 --- a/src/traced_relay/relay_service_main.cc +++ b/src/traced_relay/relay_service_main.cc
@@ -16,8 +16,8 @@ #include "perfetto/ext/base/file_utils.h" #include "perfetto/ext/base/getopt.h" +#include "perfetto/ext/base/lock_free_task_runner.h" #include "perfetto/ext/base/string_utils.h" -#include "perfetto/ext/base/unix_task_runner.h" #include "perfetto/ext/base/version.h" #include "perfetto/ext/base/watchdog.h" #include "perfetto/ext/traced/traced.h" @@ -115,7 +115,7 @@ base::Daemonize([] { return 0; }); } - base::UnixTaskRunner task_runner; + base::MaybeLockFreeTaskRunner task_runner; auto svc = std::make_unique<RelayService>(&task_runner); // traced_relay binds to the producer socket of the `traced` service. When
diff --git a/src/websocket_bridge/websocket_bridge.cc b/src/websocket_bridge/websocket_bridge.cc index 240e17c..0c58e90 100644 --- a/src/websocket_bridge/websocket_bridge.cc +++ b/src/websocket_bridge/websocket_bridge.cc
@@ -24,9 +24,9 @@ #include <vector> #include "perfetto/ext/base/http/http_server.h" +#include "perfetto/ext/base/lock_free_task_runner.h" #include "perfetto/ext/base/string_utils.h" #include "perfetto/ext/base/unix_socket.h" -#include "perfetto/ext/base/unix_task_runner.h" #include "perfetto/tracing/default_socket.h" namespace perfetto { @@ -60,7 +60,7 @@ private: base::HttpServerConnection* GetWebsocket(base::UnixSocket*); - base::UnixTaskRunner task_runner_; + base::MaybeLockFreeTaskRunner task_runner_; std::vector<Endpoint> endpoints_; std::map<base::HttpServerConnection*, std::unique_ptr<base::UnixSocket>> conns_;