Skip to content

Epoll #116

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sgerbino merged 1 commit into from
Feb 10, 2026
Merged

Epoll #116

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 18 additions & 28 deletions src/corosio/src/detail/epoll/acceptors.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -195,52 +195,42 @@ accept(
svc_.work_started();
op.impl_ptr = shared_from_this();

bool perform_now = false;
{
std::lock_guard lock(desc_state_.mutex);
if (desc_state_.read_ready)
{
desc_state_.read_ready = false;
perform_now = true;
}
else
{
desc_state_.read_op = &op;
}
}

if (perform_now)
std::lock_guard lock(desc_state_.mutex);
if (desc_state_.read_ready)
{
desc_state_.read_ready = false;
op.perform_io();
if (op.errn == EAGAIN || op.errn == EWOULDBLOCK)
{
op.errn = 0;
std::lock_guard lock(desc_state_.mutex);
desc_state_.read_op = &op;
if (op.cancelled.load(std::memory_order_acquire))
{
svc_.post(&op);
svc_.work_finished();
}
else
{
desc_state_.read_op = &op;
}
}
else
{
svc_.post(&op);
svc_.work_finished();
}
return std::noop_coroutine();
}

if (op.cancelled.load(std::memory_order_acquire))
else
{
epoll_op* claimed = nullptr;
if (op.cancelled.load(std::memory_order_acquire))
{
std::lock_guard lock(desc_state_.mutex);
if (desc_state_.read_op == &op)
claimed = std::exchange(desc_state_.read_op, nullptr);
svc_.post(&op);
svc_.work_finished();
}
if (claimed)
else
{
svc_.post(claimed);
svc_.work_finished();
desc_state_.read_op = &op;
}
}
// completion is always posted to scheduler queue, never inline.
return std::noop_coroutine();
}

Expand Down
133 changes: 66 additions & 67 deletions src/corosio/src/detail/epoll/scheduler.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,88 +174,87 @@ operator()()
err = EIO;
}

epoll_op* rd = nullptr;
epoll_op* wr = nullptr;
epoll_op* cn = nullptr;
{
std::lock_guard lock(mutex);
if (ev & EPOLLIN)
{
rd = std::exchange(read_op, nullptr);
if (!rd)
if (read_op)
{
auto* rd = read_op;
if (err)
rd->complete(err, 0);
else
rd->perform_io();

if (rd->errn == EAGAIN || rd->errn == EWOULDBLOCK)
{
rd->errn = 0;
}
else
{
read_op = nullptr;
local_ops.push(rd);
}
}
else
{
read_ready = true;
}
}
if (ev & EPOLLOUT)
{
cn = std::exchange(connect_op, nullptr);
wr = std::exchange(write_op, nullptr);
if (!cn && !wr)
bool had_write_op = (connect_op || write_op);
if (connect_op)
{
auto* cn = connect_op;
if (err)
cn->complete(err, 0);
else
cn->perform_io();
connect_op = nullptr;
local_ops.push(cn);
}
if (write_op)
{
auto* wr = write_op;
if (err)
wr->complete(err, 0);
else
wr->perform_io();

if (wr->errn == EAGAIN || wr->errn == EWOULDBLOCK)
{
wr->errn = 0;
}
else
{
write_op = nullptr;
local_ops.push(wr);
}
}
if (!had_write_op)
write_ready = true;
}
if (err && !(ev & (EPOLLIN | EPOLLOUT)))
{
rd = std::exchange(read_op, nullptr);
wr = std::exchange(write_op, nullptr);
cn = std::exchange(connect_op, nullptr);
}
}

// Non-null after I/O means EAGAIN; re-register under lock below
if (rd)
{
if (err)
rd->complete(err, 0);
else
rd->perform_io();

if (rd->errn == EAGAIN || rd->errn == EWOULDBLOCK)
{
rd->errn = 0;
}
else
{
local_ops.push(rd);
rd = nullptr;
}
}

if (cn)
{
if (err)
cn->complete(err, 0);
else
cn->perform_io();
local_ops.push(cn);
cn = nullptr;
}

if (wr)
{
if (err)
wr->complete(err, 0);
else
wr->perform_io();

if (wr->errn == EAGAIN || wr->errn == EWOULDBLOCK)
{
wr->errn = 0;
}
else
{
local_ops.push(wr);
wr = nullptr;
if (read_op)
{
read_op->complete(err, 0);
local_ops.push(std::exchange(read_op, nullptr));
}
if (write_op)
{
write_op->complete(err, 0);
local_ops.push(std::exchange(write_op, nullptr));
}
if (connect_op)
{
connect_op->complete(err, 0);
local_ops.push(std::exchange(connect_op, nullptr));
}
}
}

if (rd || wr)
{
std::lock_guard lock(mutex);
if (rd)
read_op = rd;
if (wr)
write_op = wr;
}

// Execute first handler inline — the scheduler's work_cleanup
// accounts for this as the "consumed" work item
scheduler_op* first = local_ops.pop();
Expand Down
Loading
Loading