Skip to content

Add epoll operation queue #269

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
mrabine merged 1 commit into from
Sep 26, 2026
Merged

Add epoll operation queue #269

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
6 changes: 3 additions & 3 deletions core/include/join/async_datagram_socket.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,7 +170,7 @@ namespace join
if (this->_proactor->submit (read->op, flush, false) == -1)
{
// LCOV_EXCL_START
this->releaseOp (read);
this->releaseOperation (read);
return -1;
// LCOV_EXCL_STOP
}
Expand Down Expand Up @@ -218,7 +218,7 @@ namespace join
if (this->_proactor->submit (read->op, flush, false) == -1)
{
// LCOV_EXCL_START
this->releaseOp (read);
this->releaseOperation (read);
return -1;
// LCOV_EXCL_STOP
}
Expand Down Expand Up @@ -278,7 +278,7 @@ namespace join
if (this->_proactor->submit (write->op, flush, false) == -1)
{
// LCOV_EXCL_START
this->releaseOp (write);
this->releaseOperation (write);
return -1;
// LCOV_EXCL_STOP
}
Expand Down
22 changes: 11 additions & 11 deletions core/include/join/async_raw_socket.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -196,7 +196,7 @@ namespace join
if (this->_proactor->submit (read->op, flush, false) == -1)
{
// LCOV_EXCL_START
this->releaseOp (read);
this->releaseOperation (read);
return -1;
// LCOV_EXCL_STOP
}
Expand Down Expand Up @@ -240,7 +240,7 @@ namespace join
if (this->_proactor->submit (read->op, flush, false) == -1)
{
// LCOV_EXCL_START
this->releaseOp (read);
this->releaseOperation (read);
return -1;
// LCOV_EXCL_STOP
}
Expand Down Expand Up @@ -280,7 +280,7 @@ namespace join
if (this->_proactor->submit (read->op, flush, false) == -1)
{
// LCOV_EXCL_START
this->releaseOp (read);
this->releaseOperation (read);
return -1;
// LCOV_EXCL_STOP
}
Expand Down Expand Up @@ -331,7 +331,7 @@ namespace join
if (this->_proactor->submit (write->op, flush, false) == -1)
{
// LCOV_EXCL_START
this->releaseOp (write);
this->releaseOperation (write);
return -1;
// LCOV_EXCL_STOP
}
Expand Down Expand Up @@ -375,7 +375,7 @@ namespace join
if (this->_proactor->submit (write->op, flush, false) == -1)
{
// LCOV_EXCL_START
this->releaseOp (write);
this->releaseOperation (write);
return -1;
// LCOV_EXCL_STOP
}
Expand All @@ -396,7 +396,7 @@ namespace join

if ((op != nullptr) && (static_cast<IoOperation::Opcode> (op->code) == IoOperation::Opcode::Connect))
{
return this->cancelOp (*op);
return this->cancelOperation (*op);
}
}

Expand Down Expand Up @@ -461,7 +461,7 @@ namespace join
this->_socket._state = Socket::Connected;
}

this->releaseOp (connect);
this->releaseOperation (connect);

if (JOIN_LIKELY (handler))
{
Expand Down Expand Up @@ -540,7 +540,7 @@ namespace join
ReadHandler handler = std::move (read->readHandler);
ReadFromHandler fromHandler = std::move (read->readFromHandler);

this->releaseOp (read);
this->releaseOperation (read);

if (fromHandler)
{
Expand All @@ -563,7 +563,7 @@ namespace join

if (JOIN_LIKELY (!write->op.multishot))
{
this->releaseOp (write);
this->releaseOperation (write);
}

if (JOIN_LIKELY (handler))
Expand All @@ -578,7 +578,7 @@ namespace join
*/
AsyncRead* allocateRead () noexcept
{
return this->template allocateOp<AsyncRead> ();
return this->template allocateOperation<AsyncRead> ();
}

/**
Expand All @@ -587,7 +587,7 @@ namespace join
*/
AsyncWrite* allocateWrite () noexcept
{
return this->template allocateOp<AsyncWrite> ();
return this->template allocateOperation<AsyncWrite> ();
}
};
}
Expand Down
22 changes: 11 additions & 11 deletions core/include/join/async_socket.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -228,7 +228,7 @@ namespace join
return -1;
}

AsyncWait* wait = allocateOp<AsyncWait> ();
AsyncWait* wait = allocateOperation<AsyncWait> ();
if (JOIN_UNLIKELY (wait == nullptr))
{
lastError = make_error_code (Errc::OutOfMemory);
Expand All @@ -244,7 +244,7 @@ namespace join
if (_proactor->submit (wait->op, flush, false) == -1)
{
// LCOV_EXCL_START
releaseOp (wait);
releaseOperation (wait);
return -1;
// LCOV_EXCL_STOP
}
Expand Down Expand Up @@ -274,7 +274,7 @@ namespace join
return -1;
}

AsyncWait* wait = allocateOp<AsyncWait> ();
AsyncWait* wait = allocateOperation<AsyncWait> ();
if (JOIN_UNLIKELY (wait == nullptr))
{
lastError = make_error_code (Errc::OutOfMemory);
Expand All @@ -290,7 +290,7 @@ namespace join
if (_proactor->submit (wait->op, flush, false) == -1)
{
// LCOV_EXCL_START
releaseOp (wait);
releaseOperation (wait);
return -1;
// LCOV_EXCL_STOP
}
Expand All @@ -313,7 +313,7 @@ namespace join

IoOperation* op = _ops[index].load (std::memory_order_acquire);

return (op != nullptr) ? cancelOp (*op) : 0;
return (op != nullptr) ? cancelOperation (*op) : 0;
}

#ifdef JOIN_HAS_IO_URING
Expand Down Expand Up @@ -498,11 +498,11 @@ namespace join

if (wait->op.more)
{
cancelOp (wait->op);
cancelOperation (wait->op);
}
else
{
releaseOp (wait);
releaseOperation (wait);
}

if (handler)
Expand All @@ -516,7 +516,7 @@ namespace join
* @return allocated operation, or nullptr if the arena is exhausted.
*/
template <class Op>
Op* allocateOp () noexcept
Op* allocateOperation () noexcept
{
static_assert (sizeof (Op) <= _opSize, "operation larger than an arena slot");

Expand All @@ -537,7 +537,7 @@ namespace join
* @param operation operation to release.
*/
template <class Op>
void releaseOp (Op* operation) noexcept
void releaseOperation (Op* operation) noexcept
{
const uint32_t index = _arena.getIndex (operation);

Expand Down Expand Up @@ -586,7 +586,7 @@ namespace join
IoOperation* op = slot.load (std::memory_order_acquire);
if (op != nullptr)
{
cancelOp (*op);
cancelOperation (*op);
}
}
}
Expand Down Expand Up @@ -623,7 +623,7 @@ namespace join
* @param op operation to cancel.
* @return 0 on success, -1 on failure.
*/
int cancelOp (IoOperation& op) noexcept
int cancelOperation (IoOperation& op) noexcept
{
if (!inFlight (op))
{
Expand Down
2 changes: 1 addition & 1 deletion core/include/join/async_stream_socket.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,7 +136,7 @@ namespace join
if (this->_proactor->submit (connect->op, true, false) == -1)
{
// LCOV_EXCL_START
this->releaseOp (connect);
this->releaseOperation (connect);
this->_socket.close ();
return -1;
// LCOV_EXCL_STOP
Expand Down
7 changes: 6 additions & 1 deletion core/include/join/io_operation.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -389,8 +389,13 @@ namespace join
*/
int fd () const noexcept;

/// index of this operation in the proactor pending ops (io_uring only).
#ifdef JOIN_HAS_IO_URING
/// index of this operation in the proactor pending ops.
uint32_t index = 0;
#else
/// next operation queued on the same descriptor and direction.
IoOperation* next = nullptr;
#endif

/// operation code.
uint8_t code = 0;
Expand Down
49 changes: 40 additions & 9 deletions core/include/join/proactor.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -579,26 +579,57 @@ class join::BasicProactor : public join::EventHandler
void eventLoop (std::true_type, std::true_type) noexcept;
#else
/**
* @brief return true if operation requires EPOLLOUT.
* @param op operation.
* @return true for Connect, Write, WriteFixed, SendMsg, Send and Poll without POLLIN.
* @brief method called when events are reported on handle.
* @param fd file descriptor.
* @param revents events reported by the reactor.
*/
void onEvent (int fd, uint32_t revents) noexcept override;

/**
* @brief complete every operation of a queue detached from its descriptor, in submission order.
* @param tail last operation of the queue.
* @param result result reported to the operations.
* @param revents events reported by the reactor, handed to poll operations.
* @param cancelled true if the operations are cancelled.
*/
void drainQueue (IoOperation& tail, int result, uint32_t revents, bool cancelled) noexcept;

/**
* @brief notify every poll operation of a queue whose descriptor is ready and execute its first other operation.
* @param fd file descriptor.
* @param write true to serve the write queue, false to serve the read queue.
* @param revents events reported by the reactor.
*/
static bool isWriteOp (const IoOperation& op) noexcept;
void processQueue (int fd, bool write, uint32_t revents) noexcept;

/**
* @brief execute an operation whose descriptor is ready.
* @param op operation to execute.
* @param revents events reported by the reactor.
*/
void processOperation (IoOperation& op, uint32_t revents) noexcept;

/**
* @brief execute the syscall described by op.
* @param op operation to execute.
* @param revents events reported by the reactor.
* @return bytes transferred or ready events (>= 0), or -errno (< 0).
*/
static int executeOp (IoOperation& op, uint32_t revents) noexcept;
static int executeOperation (IoOperation& op, uint32_t revents) noexcept;

/**
* @brief method called when events are reported on handle.
* @param fd file descriptor.
* @param revents events reported by the reactor.
* @brief remove an operation from the queue of its descriptor and direction.
* @param op operation to remove.
* @return true if the operation was queued, false otherwise.
*/
void onEvent (int fd, uint32_t revents) noexcept override;
bool unqueueOperation (IoOperation& op) noexcept;

/**
* @brief return true if operation requires EPOLLOUT.
* @param op operation.
* @return true for Connect, Write, WriteFixed, SendMsg, Send and Poll without POLLIN.
*/
static bool isWriteOperation (const IoOperation& op) noexcept;
#endif

/// command queue size.
Expand Down
Loading
Loading