Skip to content

Rollback io_uring completion setup failures #221

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

Open
samuel-williams-shopify wants to merge 4 commits into
base: main
Choose a base branch
from
Open

Rollback io_uring completion setup failures #221

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
d57664e
Add SQE acquisition failure regression test
samuel-williams-shopify Aug 27, 2026
dd9b627
Rollback completion state when SQE setup fails
samuel-williams-shopify Aug 27, 2026
34958a5
Fix test formatting
samuel-williams-shopify Aug 27, 2026
d01bba0
Remove runtime fault injection hooks
samuel-williams-shopify Aug 27, 2026
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
59 changes: 50 additions & 9 deletions ext/io/event/selector/uring.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -213,11 +213,18 @@ struct IO_Event_Selector_URing_Completion * IO_Event_Selector_URing_Completion_a

waiting->completion = completion;
completion->waiting = waiting;
completion->operation_pending = true;
// The operation is not pending until an SQE has been prepared successfully.

return completion;
}

inline static
void IO_Event_Selector_URing_Completion_submit(struct IO_Event_Selector_URing_Completion *completion)
{
assert(!completion->operation_pending);
completion->operation_pending = true;
}

inline static
void IO_Event_Selector_URing_Completion_cancel(struct IO_Event_Selector_URing_Completion *completion)
{
Expand All @@ -239,6 +246,16 @@ void IO_Event_Selector_URing_Completion_recycle(struct IO_Event_Selector_URing *
IO_Event_List_prepend(&selector->free_list, &completion->list);
}

inline static
void IO_Event_Selector_URing_Completion_abandon(struct IO_Event_Selector_URing *selector, struct IO_Event_Selector_URing_Completion *completion)
{
assert(!completion->operation_pending);
assert(!completion->cancellation_pending);

IO_Event_Selector_URing_Completion_cancel(completion);
IO_Event_Selector_URing_Completion_recycle(selector, completion);
}

inline static
void IO_Event_Selector_URing_Completion_complete(struct IO_Event_Selector_URing *selector, struct IO_Event_Selector_URing_Completion *completion)
{
Expand Down Expand Up @@ -558,14 +575,32 @@ struct io_uring_sqe * io_get_sqe(struct IO_Event_Selector_URing *selector) {
return sqe;
}

static VALUE IO_Event_Selector_URing_get_sqe_protected(VALUE _selector) {
struct IO_Event_Selector_URing *selector = (struct IO_Event_Selector_URing *)(uintptr_t)_selector;

return (VALUE)(uintptr_t)io_get_sqe(selector);
}

static struct io_uring_sqe * IO_Event_Selector_URing_Completion_get_sqe(struct IO_Event_Selector_URing *selector, struct IO_Event_Selector_URing_Completion *completion) {
int state = 0;
VALUE result = rb_protect(IO_Event_Selector_URing_get_sqe_protected, (VALUE)selector, &state);

if (state) {
IO_Event_Selector_URing_Completion_abandon(selector, completion);
rb_jump_tag(state);
}

return (struct io_uring_sqe *)(uintptr_t)result;
}

static
void IO_Event_Selector_URing_Completion_cancel_async(struct IO_Event_Selector_URing *selector, struct IO_Event_Selector_URing_Completion *completion)
{
if (completion->cancellation_pending) return;

completion->cancellation_pending = true;

struct io_uring_sqe *sqe = io_get_sqe(selector);
completion->cancellation_pending = true;

io_uring_prep_cancel(sqe, completion, 0);
io_uring_sqe_set_data64(sqe, IO_Event_Selector_URing_Completion_cancellation_data(completion));
io_uring_submit_pending(selector);
Expand Down Expand Up @@ -781,7 +816,7 @@ VALUE IO_Event_Selector_URing_process_wait(VALUE self, VALUE fiber, VALUE _pid,
#endif
};

struct io_uring_sqe *sqe = io_get_sqe(selector);
struct io_uring_sqe *sqe = IO_Event_Selector_URing_Completion_get_sqe(selector, completion);

#ifdef IO_EVENT_SELECTOR_URING_USE_WAITID
id_t id;
Expand All @@ -799,6 +834,7 @@ VALUE IO_Event_Selector_URing_process_wait(VALUE self, VALUE fiber, VALUE _pid,
io_uring_prep_poll_add(sqe, descriptor, POLLIN|POLLHUP|POLLERR);
#endif
io_uring_sqe_set_data(sqe, completion);
IO_Event_Selector_URing_Completion_submit(completion);
io_uring_submit_pending(selector);

return rb_ensure(process_wait_transfer, (VALUE)&process_wait_arguments, process_wait_ensure, (VALUE)&process_wait_arguments);
Expand Down Expand Up @@ -890,9 +926,10 @@ VALUE IO_Event_Selector_URing_io_wait(VALUE self, VALUE fiber, VALUE io, VALUE e

struct IO_Event_Selector_URing_Completion *completion = IO_Event_Selector_URing_Completion_acquire(selector, &waiting);

struct io_uring_sqe *sqe = io_get_sqe(selector);
struct io_uring_sqe *sqe = IO_Event_Selector_URing_Completion_get_sqe(selector, completion);
io_uring_prep_poll_add(sqe, descriptor, flags);
io_uring_sqe_set_data(sqe, completion);
IO_Event_Selector_URing_Completion_submit(completion);
// If we are going to wait, we assume that we are waiting for a while:
io_uring_submit_pending(selector);

Expand Down Expand Up @@ -944,9 +981,11 @@ io_read_submit(VALUE _arguments)

if (DEBUG) fprintf(stderr, "io_read_submit:io_uring_prep_read(waiting=%p, completion=%p, descriptor=%d, buffer=%p, length=%ld)\n", (void*)arguments->waiting, (void*)arguments->waiting->completion, arguments->descriptor, arguments->buffer, arguments->length);

struct io_uring_sqe *sqe = io_get_sqe(selector);
struct IO_Event_Selector_URing_Completion *completion = arguments->waiting->completion;
struct io_uring_sqe *sqe = IO_Event_Selector_URing_Completion_get_sqe(selector, completion);
io_uring_prep_read(sqe, arguments->descriptor, arguments->buffer, arguments->length, arguments->offset);
io_uring_sqe_set_data(sqe, arguments->waiting->completion);
io_uring_sqe_set_data(sqe, completion);
IO_Event_Selector_URing_Completion_submit(completion);
io_uring_submit_pending(selector);

IO_Event_Selector_loop_yield(&selector->backend);
Expand Down Expand Up @@ -1211,9 +1250,11 @@ io_write_submit(VALUE _argument)

if (DEBUG) fprintf(stderr, "io_write_submit:io_uring_prep_write(waiting=%p, completion=%p, descriptor=%d, buffer=%p, length=%ld)\n", (void*)arguments->waiting, (void*)arguments->waiting->completion, arguments->descriptor, arguments->buffer, arguments->length);

struct io_uring_sqe *sqe = io_get_sqe(selector);
struct IO_Event_Selector_URing_Completion *completion = arguments->waiting->completion;
struct io_uring_sqe *sqe = IO_Event_Selector_URing_Completion_get_sqe(selector, completion);
io_uring_prep_write(sqe, arguments->descriptor, arguments->buffer, arguments->length, arguments->offset);
io_uring_sqe_set_data(sqe, arguments->waiting->completion);
io_uring_sqe_set_data(sqe, completion);
IO_Event_Selector_URing_Completion_submit(completion);
io_uring_submit_pending(selector);

IO_Event_Selector_loop_yield(&selector->backend);
Expand Down
Loading