Skip to content

Commit

Permalink
pw_channel: Remove manual registration from epoll channel
Browse files Browse the repository at this point in the history 
This hides the Register method from the epoll channel's public API,
instead automatically calling in the constructor, and relying on the
base Channel's read and write open flags to detect errors.

Change-Id: I16a7b416750956cb8e86b3069b9135587cf065ad
Reviewed-on: https://pigweed-review.googlesource.com/c/pigweed/pigweed/+/213653
Pigweed-Auto-Submit: Alexei Frolov <[email protected]>
Lint: Lint 🤖 <[email protected]>
Presubmit-Verified: CQ Bot Account <[email protected]>
Commit-Queue: Auto-Submit <[email protected]>
Reviewed-by: Wyatt Hepler <[email protected]>
  • Loading branch information
@frolv
frolv authored and CQ Bot Account committed Jun 3, 2024
1 parent 778137c commit 60ae08a
Show file tree
Hide file tree
Showing 3 changed files with 39 additions and 69 deletions.
31 changes: 15 additions & 16 deletions pw_channel/epoll_channel.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,29 +22,28 @@

namespace pw::channel {

Status EpollChannel::Register() {
if (is_open()) {
return Status::FailedPrecondition();
}

void EpollChannel::Register() {
if (fcntl(channel_fd_, F_SETFL, O_NONBLOCK) != 0) {
PW_LOG_ERROR("Failed to make channel file descriptor nonblocking: %s",
std::strerror(errno));
return Status::Internal();
set_closed();
return;
}

PW_TRY(dispatcher_->NativeRegisterFileDescriptor(
channel_fd_, async2::Dispatcher::FileDescriptorType::kReadWrite));
if (!dispatcher_
->NativeRegisterFileDescriptor(
channel_fd_, async2::Dispatcher::FileDescriptorType::kReadWrite)
.ok()) {
set_closed();
return;
}

open_ = true;
ready_to_write_ = true;

return OkStatus();
}

async2::Poll<Result<multibuf::MultiBuf>> EpollChannel::DoPendRead(
async2::Context& cx) {
if (!is_open()) {
if (!is_read_open()) {
return Status::FailedPrecondition();
}

Expand Down Expand Up @@ -88,7 +87,7 @@ async2::Poll<Result<multibuf::MultiBuf>> EpollChannel::DoPendRead(
}

async2::Poll<Status> EpollChannel::DoPendReadyToWrite(async2::Context& cx) {
if (!is_open()) {
if (!is_write_open()) {
return Status::FailedPrecondition();
}

Expand All @@ -106,7 +105,7 @@ async2::Poll<Status> EpollChannel::DoPendReadyToWrite(async2::Context& cx) {
}

Result<channel::WriteToken> EpollChannel::DoWrite(multibuf::MultiBuf&& data) {
if (!is_open()) {
if (!is_write_open()) {
return Status::FailedPrecondition();
}

Expand All @@ -131,9 +130,9 @@ Result<channel::WriteToken> EpollChannel::DoWrite(multibuf::MultiBuf&& data) {
}

void EpollChannel::Cleanup() {
if (is_open()) {
if (is_read_or_write_open()) {
dispatcher_->NativeUnregisterFileDescriptor(channel_fd_).IgnoreError();
open_ = false;
set_closed();
}
close(channel_fd_);
}
Expand Down
47 changes: 14 additions & 33 deletions pw_channel/epoll_channel_test.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -139,7 +139,8 @@ TEST_F(EpollChannelTest, Read_ValidData_Succeeds) {
Dispatcher dispatcher;

EpollChannel channel(read_fd_, dispatcher, alloc);
ASSERT_EQ(channel.Register(), pw::OkStatus());
ASSERT_TRUE(channel.is_read_open());
ASSERT_TRUE(channel.is_write_open());

ReaderTask<ByteReader> read_task(channel, 1);
dispatcher.Post(read_task);
Expand Down Expand Up @@ -170,25 +171,13 @@ TEST_F(EpollChannelTest, Read_ValidData_Succeeds) {
EXPECT_EQ(close_task.close_status, pw::OkStatus());
}

TEST_F(EpollChannelTest, Read_Unregistered_ReturnsFailedPrecondition) {
SimpleAllocatorForTest alloc;
Dispatcher dispatcher;

EpollChannel channel(read_fd_, dispatcher, alloc);

ReaderTask<ByteReader> read_task(channel, 1);
dispatcher.Post(read_task);

EXPECT_EQ(dispatcher.RunUntilStalled(), Ready());
EXPECT_EQ(read_task.read_status, pw::Status::FailedPrecondition());
}

TEST_F(EpollChannelTest, Read_Closed_ReturnsFailedPrecondition) {
SimpleAllocatorForTest alloc;
Dispatcher dispatcher;

EpollChannel channel(read_fd_, dispatcher, alloc);
ASSERT_EQ(channel.Register(), pw::OkStatus());
ASSERT_TRUE(channel.is_read_open());
ASSERT_TRUE(channel.is_write_open());

CloseTask close_task(channel);
dispatcher.Post(close_task);
Expand Down Expand Up @@ -265,7 +254,8 @@ TEST_F(EpollChannelTest, Write_ValidData_Succeeds) {
Dispatcher dispatcher;

EpollChannel channel(write_fd_, dispatcher, alloc);
ASSERT_EQ(channel.Register(), pw::OkStatus());
ASSERT_TRUE(channel.is_read_open());
ASSERT_TRUE(channel.is_write_open());

constexpr auto kData = pw::bytes::Initialized<32>(0x3f);
WriterTask<ByteWriter> write_task(channel, 1, kData);
Expand All @@ -290,7 +280,8 @@ TEST_F(EpollChannelTest, Write_EmptyData_Succeeds) {
Dispatcher dispatcher;

EpollChannel channel(write_fd_, dispatcher, alloc);
ASSERT_EQ(channel.Register(), pw::OkStatus());
ASSERT_TRUE(channel.is_read_open());
ASSERT_TRUE(channel.is_write_open());

WriterTask<ByteWriter> write_task(channel, 1, {});
dispatcher.Post(write_task);
Expand All @@ -304,25 +295,13 @@ TEST_F(EpollChannelTest, Write_EmptyData_Succeeds) {
EXPECT_EQ(close_task.close_status, pw::OkStatus());
}

TEST_F(EpollChannelTest, Write_Unregistered_ReturnsFailedPrecondition) {
SimpleAllocatorForTest alloc;
Dispatcher dispatcher;

EpollChannel channel(write_fd_, dispatcher, alloc);

WriterTask<ByteWriter> write_task(channel, 1, {});
dispatcher.Post(write_task);

dispatcher.RunToCompletion();
EXPECT_EQ(write_task.last_write_status, pw::Status::FailedPrecondition());
}

TEST_F(EpollChannelTest, Write_Closed_ReturnsFailedPrecondition) {
SimpleAllocatorForTest alloc;
Dispatcher dispatcher;

EpollChannel channel(write_fd_, dispatcher, alloc);
ASSERT_EQ(channel.Register(), pw::OkStatus());
ASSERT_TRUE(channel.is_read_open());
ASSERT_TRUE(channel.is_write_open());

CloseTask close_task(channel);
dispatcher.Post(close_task);
Expand All @@ -342,7 +321,8 @@ TEST_F(EpollChannelTest, Destructor_ClosesFileDescriptor) {

{
EpollChannel channel(write_fd_, dispatcher, alloc);
ASSERT_EQ(channel.Register(), pw::OkStatus());
ASSERT_TRUE(channel.is_read_open());
ASSERT_TRUE(channel.is_write_open());
}

const char kArbitraryByte = 'b';
Expand All @@ -354,7 +334,8 @@ TEST_F(EpollChannelTest, PendReadyToWrite_BlocksWhenUnavailable) {
SimpleAllocatorForTest alloc;
Dispatcher dispatcher;
EpollChannel channel(write_fd_, dispatcher, alloc);
ASSERT_EQ(channel.Register(), pw::OkStatus());
ASSERT_TRUE(channel.is_read_open());
ASSERT_TRUE(channel.is_write_open());

constexpr auto kData =
pw::bytes::Initialized<decltype(alloc)::data_size_bytes()>('c');
Expand Down
30 changes: 10 additions & 20 deletions pw_channel/public/pw_channel/epoll_channel.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -42,12 +42,13 @@ class EpollChannel : public ByteReaderWriter {
async2::Dispatcher& dispatcher,
multibuf::MultiBufAllocator& allocator)
: channel_fd_(channel_fd),
open_(false),
ready_to_write_(false),
write_token_(0),
allocation_future_(std::nullopt),
dispatcher_(&dispatcher),
allocator_(&allocator) {}
allocator_(&allocator) {
Register();
}

~EpollChannel() override { Cleanup(); }

Expand All @@ -57,27 +58,12 @@ class EpollChannel : public ByteReaderWriter {
EpollChannel(EpollChannel&&) = default;
EpollChannel& operator=(EpollChannel&&) = default;

/// Registers the channel's file descriptor on its EpollDisptacher.
/// Must be called before any other channel operations.
///
/// @return @rst
///
/// .. pw-status-codes::
///
/// OK: The channel has been registered and can be used.
///
/// INTERNAL: One of the underlying Linux syscalls failed. A log message
/// with the value of ``errno`` is sent.
///
/// @endrst
Status Register();

constexpr bool is_open() const { return open_; }

private:
static constexpr size_t kMinimumReadSize = 64;
static constexpr size_t kDesiredReadSize = 1024;

void Register();

async2::Poll<Result<multibuf::MultiBuf>> DoPendRead(
async2::Context& cx) override;

Expand All @@ -99,10 +85,14 @@ class EpollChannel : public ByteReaderWriter {
return async2::Ready(OkStatus());
}

void set_closed() {
set_read_closed();
set_write_closed();
}

void Cleanup();

int channel_fd_;
bool open_;
bool ready_to_write_;
uint32_t write_token_;
std::optional<multibuf::MultiBufAllocationFuture> allocation_future_;
Expand Down

0 comments on commit 60ae08a

Please sign in to comment.