1 // Copyright 2024 The Pigweed Authors
2 //
3 // Licensed under the Apache License, Version 2.0 (the "License"); you may not
4 // use this file except in compliance with the License. You may obtain a copy of
5 // the License at
6 //
7 // https://www.apache.org/licenses/LICENSE-2.0
8 //
9 // Unless required by applicable law or agreed to in writing, software
10 // distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11 // WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12 // License for the specific language governing permissions and limitations under
13 // the License.
14
15 #include "pw_channel/epoll_channel.h"
16
17 #include <fcntl.h>
18 #include <sys/stat.h>
19 #include <sys/types.h>
20 #include <unistd.h>
21
22 #include "pw_assert/check.h"
23 #include "pw_async2/dispatcher.h"
24 #include "pw_bytes/array.h"
25 #include "pw_bytes/suffix.h"
26 #include "pw_channel/channel.h"
27 #include "pw_multibuf/simple_allocator_for_test.h"
28 #include "pw_status/status.h"
29 #include "pw_thread/sleep.h"
30 #include "pw_thread/thread.h"
31 #include "pw_thread_stl/options.h"
32 #include "pw_unit_test/framework.h"
33
34 namespace {
35
36 using namespace std::chrono_literals;
37
38 using ::pw::async2::Context;
39 using ::pw::async2::Dispatcher;
40 using ::pw::async2::Pending;
41 using ::pw::async2::Poll;
42 using ::pw::async2::Ready;
43 using ::pw::async2::Task;
44 using ::pw::channel::ByteReader;
45 using ::pw::channel::ByteWriter;
46 using ::pw::channel::EpollChannel;
47 using ::pw::multibuf::MultiBuf;
48 using ::pw::multibuf::test::SimpleAllocatorForTest;
49
50 template <typename ChannelKind>
51 class ReaderTask : public Task {
52 public:
ReaderTask(ChannelKind & channel,int num_reads)53 ReaderTask(ChannelKind& channel, int num_reads)
54 : channel_(channel), num_reads_(num_reads) {}
55
56 int poll_count = 0;
57 int read_count = 0;
58 int bytes_read = 0;
59 pw::Status read_status = pw::Status::Unknown();
60
61 private:
DoPend(Context & cx)62 Poll<> DoPend(Context& cx) final {
63 ++poll_count;
64 while (read_count < num_reads_) {
65 auto result = channel_.PendRead(cx);
66 if (result.IsPending()) {
67 return Pending();
68 }
69 read_status = result->status();
70 if (!result->ok()) {
71 // We hit an error-- call it quits.
72 return Ready();
73 }
74 ++read_count;
75 bytes_read += (**result).size();
76
77 (**result).Release();
78 }
79
80 return Ready();
81 }
82
83 ChannelKind& channel_;
84 int num_reads_;
85 };
86
87 template <typename ChannelKind>
88 class CloseTask : public Task {
89 public:
CloseTask(ChannelKind & channel)90 CloseTask(ChannelKind& channel) : channel_(channel) {}
91
92 pw::Status close_status = pw::Status::Unknown();
93
94 private:
DoPend(Context & cx)95 Poll<> DoPend(Context& cx) final {
96 auto result = channel_.PendClose(cx);
97 if (result.IsPending()) {
98 return Pending();
99 }
100
101 close_status = *result;
102 return Ready();
103 }
104
105 ChannelKind& channel_;
106 };
107
108 class EpollChannelTest : public ::testing::Test {
109 protected:
EpollChannelTest()110 EpollChannelTest() {
111 int pipefd[2];
112 PW_CHECK_INT_NE(pipe(pipefd), -1);
113 read_fd_ = pipefd[0];
114 write_fd_ = pipefd[1];
115 }
116
~EpollChannelTest()117 ~EpollChannelTest() override {
118 close(read_fd_);
119 close(write_fd_);
120 }
121
122 int read_fd_;
123 int write_fd_;
124 };
125
TEST_F(EpollChannelTest,Read_ValidData_Succeeds)126 TEST_F(EpollChannelTest, Read_ValidData_Succeeds) {
127 SimpleAllocatorForTest alloc;
128 Dispatcher dispatcher;
129
130 EpollChannel channel(read_fd_, dispatcher, alloc);
131 ASSERT_TRUE(channel.is_read_open());
132 ASSERT_TRUE(channel.is_write_open());
133
134 ReaderTask<ByteReader> read_task(channel.channel(), 1);
135 dispatcher.Post(read_task);
136
137 EXPECT_EQ(dispatcher.RunUntilStalled(), Pending());
138 EXPECT_EQ(read_task.poll_count, 1);
139 EXPECT_EQ(read_task.read_count, 0);
140 EXPECT_EQ(read_task.bytes_read, 0);
141
142 pw::Thread work_thread(pw::thread::stl::Options(), [this] {
143 pw::this_thread::sleep_for(500ms);
144 const char* data = "hello world";
145 PW_CHECK_INT_EQ(write(write_fd_, data, 11), 11);
146 });
147 work_thread.join();
148
149 dispatcher.RunToCompletion();
150 EXPECT_EQ(read_task.read_status, pw::OkStatus());
151 EXPECT_EQ(read_task.poll_count, 2);
152 EXPECT_EQ(read_task.read_count, 1);
153 EXPECT_EQ(read_task.bytes_read, 11);
154
155 CloseTask close_task(channel);
156 dispatcher.Post(close_task);
157 EXPECT_EQ(dispatcher.RunUntilStalled(), Ready());
158 EXPECT_EQ(close_task.close_status, pw::OkStatus());
159 }
160
TEST_F(EpollChannelTest,Read_Closed_ReturnsFailedPrecondition)161 TEST_F(EpollChannelTest, Read_Closed_ReturnsFailedPrecondition) {
162 SimpleAllocatorForTest alloc;
163 Dispatcher dispatcher;
164
165 EpollChannel channel(read_fd_, dispatcher, alloc);
166 ASSERT_TRUE(channel.is_read_open());
167 ASSERT_TRUE(channel.is_write_open());
168
169 CloseTask close_task(channel);
170 dispatcher.Post(close_task);
171 EXPECT_EQ(dispatcher.RunUntilStalled(), Ready());
172 EXPECT_EQ(close_task.close_status, pw::OkStatus());
173
174 ReaderTask<ByteReader> read_task(channel.channel(), 1);
175 dispatcher.Post(read_task);
176
177 EXPECT_EQ(dispatcher.RunUntilStalled(), Ready());
178 EXPECT_EQ(read_task.read_status, pw::Status::FailedPrecondition());
179 }
180
181 template <typename ChannelKind>
182 class WriterTask : public Task {
183 public:
WriterTask(ChannelKind & channel,int num_writes,pw::ConstByteSpan data_to_write)184 WriterTask(ChannelKind& channel,
185 int num_writes,
186 pw::ConstByteSpan data_to_write)
187 : max_writes(num_writes),
188 channel_(channel),
189 data_to_write_(data_to_write) {}
190
191 int poll_count = 0;
192 int write_pending_count = 0;
193 int write_count = 0;
194 int max_writes = 0;
195 pw::Status last_write_status = pw::Status::Unknown();
196
197 private:
DoPend(Context & cx)198 Poll<> DoPend(Context& cx) final {
199 ++poll_count;
200
201 while (write_count < max_writes) {
202 auto result = channel_.PendReadyToWrite(cx);
203 if (result.IsPending()) {
204 ++write_pending_count;
205 return Pending();
206 }
207 last_write_status = *result;
208 if (!result->ok()) {
209 // We hit an error-- call it quits.
210 return Ready();
211 }
212 ++write_count;
213
214 Poll<std::optional<MultiBuf>> multibuf_result =
215 channel_.PendAllocateWriteBuffer(cx, data_to_write_.size());
216 PW_CHECK(multibuf_result.IsReady());
217 PW_CHECK(multibuf_result->has_value());
218 MultiBuf& multibuf = **multibuf_result;
219 std::copy(data_to_write_.begin(), data_to_write_.end(), multibuf.begin());
220
221 last_write_status = channel_.StageWrite(std::move(multibuf));
222
223 Poll<pw::Status> write_status = channel_.PendWrite(cx);
224 if (write_status.IsPending()) {
225 return Pending();
226 }
227
228 PW_CHECK_OK(*write_status);
229 }
230
231 return Ready();
232 }
233
234 ChannelKind& channel_;
235 pw::ConstByteSpan data_to_write_;
236 };
237
TEST_F(EpollChannelTest,Write_ValidData_Succeeds)238 TEST_F(EpollChannelTest, Write_ValidData_Succeeds) {
239 SimpleAllocatorForTest alloc;
240 Dispatcher dispatcher;
241
242 EpollChannel channel(write_fd_, dispatcher, alloc);
243 ASSERT_TRUE(channel.is_read_open());
244 ASSERT_TRUE(channel.is_write_open());
245
246 constexpr auto kData = pw::bytes::Initialized<32>(0x3f);
247 WriterTask<ByteWriter> write_task(channel.channel(), 1, kData);
248 dispatcher.Post(write_task);
249
250 dispatcher.RunToCompletion();
251 EXPECT_EQ(write_task.last_write_status, pw::OkStatus());
252
253 std::array<std::byte, 64> buffer;
254 EXPECT_EQ(read(read_fd_, buffer.data(), buffer.size()),
255 static_cast<int>(kData.size()));
256 EXPECT_EQ(std::memcmp(buffer.data(), kData.data(), kData.size()), 0);
257
258 CloseTask close_task(channel);
259 dispatcher.Post(close_task);
260 EXPECT_EQ(dispatcher.RunUntilStalled(), Ready());
261 EXPECT_EQ(close_task.close_status, pw::OkStatus());
262 }
263
TEST_F(EpollChannelTest,Write_EmptyData_Succeeds)264 TEST_F(EpollChannelTest, Write_EmptyData_Succeeds) {
265 SimpleAllocatorForTest alloc;
266 Dispatcher dispatcher;
267
268 EpollChannel channel(write_fd_, dispatcher, alloc);
269 ASSERT_TRUE(channel.is_read_open());
270 ASSERT_TRUE(channel.is_write_open());
271
272 WriterTask<ByteWriter> write_task(channel.channel(), 1, {});
273 dispatcher.Post(write_task);
274
275 dispatcher.RunToCompletion();
276 EXPECT_EQ(write_task.last_write_status, pw::OkStatus());
277
278 CloseTask close_task(channel);
279 dispatcher.Post(close_task);
280 EXPECT_EQ(dispatcher.RunUntilStalled(), Ready());
281 EXPECT_EQ(close_task.close_status, pw::OkStatus());
282 }
283
TEST_F(EpollChannelTest,Write_Closed_ReturnsFailedPrecondition)284 TEST_F(EpollChannelTest, Write_Closed_ReturnsFailedPrecondition) {
285 SimpleAllocatorForTest alloc;
286 Dispatcher dispatcher;
287
288 EpollChannel channel(write_fd_, dispatcher, alloc);
289 ASSERT_TRUE(channel.is_read_open());
290 ASSERT_TRUE(channel.is_write_open());
291
292 CloseTask close_task(channel);
293 dispatcher.Post(close_task);
294 EXPECT_EQ(dispatcher.RunUntilStalled(), Ready());
295 EXPECT_EQ(close_task.close_status, pw::OkStatus());
296
297 WriterTask<ByteWriter> write_task(channel.channel(), 1, {});
298 dispatcher.Post(write_task);
299
300 dispatcher.RunToCompletion();
301 EXPECT_EQ(write_task.last_write_status, pw::Status::FailedPrecondition());
302 }
303
TEST_F(EpollChannelTest,Destructor_ClosesFileDescriptor)304 TEST_F(EpollChannelTest, Destructor_ClosesFileDescriptor) {
305 SimpleAllocatorForTest alloc;
306 Dispatcher dispatcher;
307
308 {
309 EpollChannel channel(write_fd_, dispatcher, alloc);
310 ASSERT_TRUE(channel.is_read_open());
311 ASSERT_TRUE(channel.is_write_open());
312 }
313
314 const char kArbitraryByte = 'b';
315 EXPECT_EQ(write(write_fd_, &kArbitraryByte, 1), -1);
316 EXPECT_EQ(errno, EBADF);
317 }
318
TEST_F(EpollChannelTest,PendReadyToWrite_BlocksWhenUnavailable)319 TEST_F(EpollChannelTest, PendReadyToWrite_BlocksWhenUnavailable) {
320 SimpleAllocatorForTest alloc;
321 Dispatcher dispatcher;
322 EpollChannel channel(write_fd_, dispatcher, alloc);
323 ASSERT_TRUE(channel.is_read_open());
324 ASSERT_TRUE(channel.is_write_open());
325
326 constexpr auto kData =
327 pw::bytes::Initialized<decltype(alloc)::data_size_bytes()>('c');
328 WriterTask<ByteWriter> write_task(
329 channel.channel(),
330 100, // Max writes set to some high number so the task fills the pipe.
331 pw::ConstByteSpan(kData));
332 dispatcher.Post(write_task);
333
334 // Try to write a bunch of data, eventually filling the pipe and blocking the
335 // task.
336 EXPECT_EQ(dispatcher.RunUntilStalled(), Pending());
337 EXPECT_EQ(write_task.poll_count, 1);
338 EXPECT_EQ(write_task.write_pending_count, 1);
339 EXPECT_EQ(write_task.last_write_status, pw::Status::Unavailable());
340
341 const int writes_to_drain = write_task.write_count;
342
343 // End the task on the next successful write.
344 write_task.max_writes = write_task.write_count + 1;
345
346 // Drain the pipe to make it writable again after a delay.
347 auto delayed_read = [this, writes_to_drain] {
348 pw::this_thread::sleep_for(500ms);
349 for (int i = 0; i < writes_to_drain; ++i) {
350 std::array<std::byte, decltype(alloc)::data_size_bytes()> buffer;
351 PW_CHECK_INT_GT(read(read_fd_, buffer.data(), buffer.size()), 0);
352 }
353 };
354 pw::Thread work_thread(pw::thread::stl::Options(),
355 [&delayed_read] { delayed_read(); });
356
357 dispatcher.RunToCompletion();
358 work_thread.join();
359
360 EXPECT_EQ(write_task.poll_count, 2);
361 EXPECT_EQ(write_task.last_write_status, pw::OkStatus());
362 }
363
364 } // namespace
365