1 // Copyright 2019 The Chromium Authors
2 // Use of this source code is governed by a BSD-style license that can be
3 // found in the LICENSE file.
4
5 #include "base/tracing/perfetto_task_runner.h"
6
7 #include <memory>
8 #include <utility>
9 #include <vector>
10
11 #include "base/files/scoped_file.h"
12 #include "base/memory/raw_ptr.h"
13 #include "base/memory/weak_ptr.h"
14 #include "base/task/sequenced_task_runner.h"
15 #include "base/task/thread_pool.h"
16 #include "base/test/bind.h"
17 #include "base/test/task_environment.h"
18 #include "base/threading/simple_thread.h"
19 #include "build/build_config.h"
20 #include "testing/gtest/include/gtest/gtest.h"
21
22 #if BUILDFLAG(IS_POSIX) && !BUILDFLAG(IS_NACL)
23 #include <sys/socket.h>
24 #include <sys/types.h>
25 #include "base/posix/eintr_wrapper.h"
26 #endif
27
28 namespace base {
29 namespace tracing {
30 namespace {
31
32 class TaskDestination {
33 public:
TaskDestination(size_t number_of_sequences,size_t expected_tasks,base::OnceClosure on_complete)34 TaskDestination(size_t number_of_sequences,
35 size_t expected_tasks,
36 base::OnceClosure on_complete)
37 : expected_tasks_(expected_tasks),
38 on_complete_(std::move(on_complete)),
39 last_task_id_(number_of_sequences) {}
40
tasks_run() const41 size_t tasks_run() const { return tasks_run_; }
42
TestTask(int n,size_t sequence_number=0)43 void TestTask(int n, size_t sequence_number = 0) {
44 EXPECT_LT(sequence_number, last_task_id_.size());
45 EXPECT_GT(expected_tasks_, tasks_run_);
46 EXPECT_GE(n, last_task_id_[sequence_number]);
47 last_task_id_[sequence_number] = n;
48
49 if (++tasks_run_ == expected_tasks_) {
50 std::move(on_complete_).Run();
51 }
52 }
53
GetWeakPtr()54 base::WeakPtr<TaskDestination> GetWeakPtr() {
55 return weak_ptr_factory_.GetWeakPtr();
56 }
57
58 private:
59 const size_t expected_tasks_;
60 base::OnceClosure on_complete_;
61 std::vector<int> last_task_id_;
62 size_t tasks_run_ = 0;
63
64 base::WeakPtrFactory<TaskDestination> weak_ptr_factory_{this};
65 };
66
67 class PosterThread : public base::SimpleThread {
68 public:
PosterThread(PerfettoTaskRunner * task_runner,base::WeakPtr<TaskDestination> weak_ptr,int n,size_t sequence_number)69 PosterThread(PerfettoTaskRunner* task_runner,
70 base::WeakPtr<TaskDestination> weak_ptr,
71 int n,
72 size_t sequence_number)
73 : SimpleThread("TaskPostThread"),
74 task_runner_(task_runner),
75 weak_ptr_(weak_ptr),
76 n_(n),
77 sequence_number_(sequence_number) {}
~PosterThread()78 ~PosterThread() override {}
79
80 // base::SimpleThread overrides.
BeforeStart()81 void BeforeStart() override {}
BeforeJoin()82 void BeforeJoin() override {}
83
Run()84 void Run() override {
85 for (int i = 0; i < n_; ++i) {
86 auto weak_ptr = weak_ptr_;
87 auto sequence_number = sequence_number_;
88 task_runner_->PostTask([weak_ptr, i, sequence_number]() {
89 weak_ptr->TestTask(i, sequence_number);
90 });
91 }
92 }
93
94 private:
95 raw_ptr<PerfettoTaskRunner> task_runner_;
96 base::WeakPtr<TaskDestination> weak_ptr_;
97 const int n_;
98 const size_t sequence_number_;
99 };
100
101 class PerfettoTaskRunnerTest : public testing::Test {
102 public:
SetUp()103 void SetUp() override {
104 sequenced_task_runner_ = CreateNewTaskrunner();
105 task_runner_ = std::make_unique<PerfettoTaskRunner>(sequenced_task_runner_);
106 }
107
CreateNewTaskrunner()108 scoped_refptr<base::SequencedTaskRunner> CreateNewTaskrunner() {
109 return base::ThreadPool::CreateSingleThreadTaskRunner(
110 {base::MayBlock()}, base::SingleThreadTaskRunnerThreadMode::DEDICATED);
111 }
SetTaskExpectations(base::OnceClosure on_complete,size_t expected_tasks,size_t number_of_sequences=1)112 void SetTaskExpectations(base::OnceClosure on_complete,
113 size_t expected_tasks,
114 size_t number_of_sequences = 1) {
115 task_destination_ = std::make_unique<TaskDestination>(
116 number_of_sequences, expected_tasks, std::move(on_complete));
117 }
118
TearDown()119 void TearDown() override {
120 sequenced_task_runner_->DeleteSoon(FROM_HERE, std::move(task_runner_));
121 }
122
task_runner()123 PerfettoTaskRunner* task_runner() { return task_runner_.get(); }
destination()124 TaskDestination* destination() { return task_destination_.get(); }
task_environment()125 base::test::TaskEnvironment& task_environment() { return task_environment_; }
126
127 private:
128 base::test::TaskEnvironment task_environment_;
129 scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner_;
130 std::unique_ptr<PerfettoTaskRunner> task_runner_;
131 std::unique_ptr<TaskDestination> task_destination_;
132 };
133
TEST_F(PerfettoTaskRunnerTest,SequentialTasks)134 TEST_F(PerfettoTaskRunnerTest, SequentialTasks) {
135 base::RunLoop wait_for_tasks;
136 SetTaskExpectations(wait_for_tasks.QuitClosure(), 3);
137
138 auto weak_ptr = destination()->GetWeakPtr();
139 for (int i = 1; i <= 3; ++i) {
140 task_runner()->PostTask([=]() mutable {
141 auto* dest = weak_ptr.get();
142 // The weak pointer must be reset before TestTask() is called, otherwise
143 // there will be a race where the factory could be destructed on main
144 // thread while still bound to the task runner sequence.
145 weak_ptr.reset();
146 dest->TestTask(i);
147 });
148 }
149
150 wait_for_tasks.Run();
151 }
152
153 #if BUILDFLAG(IS_POSIX) && !BUILDFLAG(IS_NACL)
154 // Tests file descriptor reuse that causes crashes.
TEST_F(PerfettoTaskRunnerTest,FileDescriptorReuse)155 TEST_F(PerfettoTaskRunnerTest, FileDescriptorReuse) {
156 int sockets[2];
157 // Use sockets because we need a FD that supports epoll().
158 ASSERT_EQ(0, socketpair(AF_UNIX, SOCK_STREAM, 0, sockets));
159 base::ScopedFD fd(sockets[0]), write_fd(sockets[1]);
160 ASSERT_TRUE(fd.is_valid());
161
162 constexpr int data_value = 0x12ab34cd;
163 bool run_callback_1 = false, run_callback_2 = false;
164 int data = data_value;
165 constexpr ssize_t data_size = static_cast<ssize_t>(sizeof(data));
166
167 // Trigger the file descriptor watcher callback.
168 ASSERT_EQ(data_size, HANDLE_EINTR(write(write_fd.get(), &data, data_size)));
169 data = 0;
170
171 base::RunLoop run_loop;
172
173 task_runner()->GetOrCreateTaskRunner()->PostTask(
174 FROM_HERE, base::BindLambdaForTesting([&]() {
175 // The 1st add operation posts a task.
176 task_runner()->AddFileDescriptorWatch(fd.get(), [&]() {
177 run_callback_1 = true;
178 ASSERT_EQ(data_size, HANDLE_EINTR(read(fd.get(), &data, data_size)));
179 run_loop.Quit();
180 });
181 // Remove so the 2nd add operation can succeed.
182 task_runner()->RemoveFileDescriptorWatch(fd.get());
183
184 // Simulate FD reuse. The 2nd add operation also posts a task.
185 task_runner()->AddFileDescriptorWatch(fd.get(), [&]() {
186 run_callback_2 = true;
187 ASSERT_EQ(data_size, HANDLE_EINTR(read(fd.get(), &data, data_size)));
188 run_loop.Quit();
189 });
190 }));
191
192 // Make all posted tasks run.
193 run_loop.Run();
194
195 ASSERT_FALSE(run_callback_1);
196 ASSERT_TRUE(run_callback_2);
197 ASSERT_EQ(data, data_value);
198
199 task_runner()->GetOrCreateTaskRunner()->PostTask(
200 FROM_HERE, base::BindLambdaForTesting([&]() {
201 // Cleanup the FD watcher.
202 task_runner()->RemoveFileDescriptorWatch(fd.get());
203 }));
204 task_environment().RunUntilIdle();
205 }
206 #endif
207 } // namespace
208 } // namespace tracing
209 } // namespace base
210