Skip to content

Commit 5471bbe

Browse files
ioquatixioquatix
committed
Allow interruption of IO waits.
1 parent 0aa989c commit 5471bbe

File tree

11 files changed

+535
-14
lines changed

11 files changed

+535
-14
lines changed

‎ext/extconf.rb‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828

2929
have_func("rb_ext_ractor_safe")
3030
have_func("&rb_fiber_transfer")
31+
have_func("rb_io_interruptable_operation")
3132

3233
if have_library("uring") and have_header("liburing.h")
3334
# We might want to consider using this in the future:

‎ext/io/event/event.c‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,4 +29,10 @@ void Init_IO_Event(void)
2929
#ifdef IO_EVENT_SELECTOR_KQUEUE
3030
Init_IO_Event_Selector_KQueue(IO_Event_Selector);
3131
#endif
32+
33+
#ifdef HAVE_RB_IO_INTERRUPTABLE_OPERATION
34+
rb_define_const(IO_Event, "INTERRUPTABLE", Qtrue);
35+
#else
36+
rb_define_const(IO_Event, "INTERRUPTABLE", Qfalse);
37+
#endif
3238
}

‎ext/io/event/selector/epoll.c‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -420,7 +420,7 @@ VALUE IO_Event_Selector_EPoll_push(VALUE self, VALUE fiber)
420420

421421
IO_Event_Selector_ready_push(&selector->backend, fiber);
422422

423-
return Qnil;
423+
return fiber;
424424
}
425425

426426
VALUE IO_Event_Selector_EPoll_raise(int argc, VALUE *argv, VALUE self)
@@ -523,6 +523,8 @@ VALUE IO_Event_Selector_EPoll_process_wait(VALUE self, VALUE fiber, VALUE _pid,
523523
struct io_wait_arguments {
524524
struct IO_Event_Selector_EPoll *selector;
525525
struct IO_Event_Selector_EPoll_Waiting *waiting;
526+
527+
VALUE io;
526528
};
527529

528530
static
@@ -538,7 +540,7 @@ static
538540
VALUE io_wait_transfer(VALUE _arguments) {
539541
struct io_wait_arguments *arguments = (struct io_wait_arguments *)_arguments;
540542

541-
IO_Event_Selector_loop_yield(&arguments->selector->backend);
543+
IO_Event_Selector_loop_yield_io(&arguments->selector->backend, arguments->io);
542544

543545
if (arguments->waiting->ready) {
544546
return RB_INT2NUM(arguments->waiting->ready);
@@ -578,6 +580,7 @@ VALUE IO_Event_Selector_EPoll_io_wait(VALUE self, VALUE fiber, VALUE io, VALUE e
578580
struct io_wait_arguments io_wait_arguments = {
579581
.selector = selector,
580582
.waiting = &waiting,
583+
.io = io,
581584
};
582585

583586
return rb_ensure(io_wait_transfer, (VALUE)&io_wait_arguments, io_wait_ensure, (VALUE)&io_wait_arguments);

‎ext/io/event/selector/kqueue.c‎

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,8 @@ enum {
1919
DEBUG = 0,
2020
DEBUG_IO_READ = 0,
2121
DEBUG_IO_WRITE = 0,
22-
DEBUG_IO_WAIT = 0
22+
DEBUG_IO_WAIT = 0,
23+
DEBUG_IO_INTERRUPT = 1
2324
};
2425

2526
#ifndef EVFILT_USER
@@ -407,7 +408,7 @@ VALUE IO_Event_Selector_KQueue_push(VALUE self, VALUE fiber)
407408

408409
IO_Event_Selector_ready_push(&selector->backend, fiber);
409410

410-
return Qnil;
411+
return fiber;
411412
}
412413

413414
VALUE IO_Event_Selector_KQueue_raise(int argc, VALUE *argv, VALUE self)
@@ -516,6 +517,8 @@ VALUE IO_Event_Selector_KQueue_process_wait(VALUE self, VALUE fiber, VALUE _pid,
516517
struct io_wait_arguments {
517518
struct IO_Event_Selector_KQueue *selector;
518519
struct IO_Event_Selector_KQueue_Waiting *waiting;
520+
521+
VALUE io;
519522
};
520523

521524
static
@@ -531,7 +534,7 @@ static
531534
VALUE io_wait_transfer(VALUE _arguments) {
532535
struct io_wait_arguments *arguments = (struct io_wait_arguments *)_arguments;
533536

534-
IO_Event_Selector_loop_yield(&arguments->selector->backend);
537+
IO_Event_Selector_loop_yield_io(&arguments->selector->backend, arguments->io);
535538

536539
if (arguments->waiting->ready) {
537540
return RB_INT2NUM(arguments->waiting->ready);
@@ -564,6 +567,7 @@ VALUE IO_Event_Selector_KQueue_io_wait(VALUE self, VALUE fiber, VALUE io, VALUE
564567
struct io_wait_arguments io_wait_arguments = {
565568
.selector = selector,
566569
.waiting = &waiting,
570+
.io = io,
567571
};
568572

569573
if (DEBUG_IO_WAIT) fprintf(stderr, "IO_Event_Selector_KQueue_io_wait descriptor=%d\n", descriptor);

‎ext/io/event/selector/selector.c‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -116,6 +116,21 @@ VALUE IO_Event_Selector_loop_yield(struct IO_Event_Selector *backend)
116116
return IO_Event_Fiber_transfer(backend->loop, 0, NULL);
117117
}
118118

119+
#ifndef HAVE_RB_IO_INTERRUPTABLE_OPERATION
120+
static inline VALUE
121+
rb_io_interruptable_operation(VALUE self, VALUE(*function)(VALUE), VALUE argument) {
122+
return function(argument);
123+
}
124+
#endif
125+
126+
static VALUE IO_Event_Selector_loop_yield_io_interruptable(VALUE fiber) {
127+
return IO_Event_Fiber_transfer(fiber, 0, NULL);
128+
}
129+
130+
VALUE IO_Event_Selector_loop_yield_io(struct IO_Event_Selector *backend, VALUE io) {
131+
return rb_io_interruptable_operation(io, IO_Event_Selector_loop_yield_io_interruptable, backend->loop);
132+
}
133+
119134
struct wait_and_transfer_arguments {
120135
int argc;
121136
VALUE *argv;

‎ext/io/event/selector/selector.h‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -120,6 +120,9 @@ VALUE IO_Event_Selector_loop_resume(struct IO_Event_Selector *backend, VALUE fib
120120
// Strictly speaking, it's not a scheduling operation (does not schedule the current fiber).
121121
VALUE IO_Event_Selector_loop_yield(struct IO_Event_Selector *backend);
122122

123+
// Similar to `IO_Event_Selector_loop_yield` but allows the caller to specify an IO which may be interrupted.
124+
VALUE IO_Event_Selector_loop_yield_io(struct IO_Event_Selector *backend, VALUE io);
125+
123126
// Resume a specific fiber. This is a scheduling operation.
124127
// The first argument is the fiber, the rest are the arguments to the resume.
125128
//

‎ext/io/event/selector/uring.c‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -303,7 +303,7 @@ VALUE IO_Event_Selector_URing_push(VALUE self, VALUE fiber)
303303

304304
IO_Event_Selector_ready_push(&selector->backend, fiber);
305305

306-
return Qnil;
306+
return fiber;
307307
}
308308

309309
VALUE IO_Event_Selector_URing_raise(int argc, VALUE *argv, VALUE self)
@@ -522,6 +522,7 @@ int events_from_poll_flags(short flags) {
522522
struct io_wait_arguments {
523523
struct IO_Event_Selector_URing *selector;
524524
struct IO_Event_Selector_URing_Waiting *waiting;
525+
VALUE io;
525526
short flags;
526527
};
527528

@@ -548,7 +549,7 @@ VALUE io_wait_transfer(VALUE _arguments) {
548549
struct io_wait_arguments *arguments = (struct io_wait_arguments *)_arguments;
549550
struct IO_Event_Selector_URing *selector = arguments->selector;
550551

551-
IO_Event_Selector_loop_yield(&selector->backend);
552+
IO_Event_Selector_loop_yield_io(&selector->backend, arguments->io);
552553

553554
if (DEBUG) fprintf(stderr, "io_wait_transfer:waiting=%p, result=%d\n", (void*)arguments->waiting, arguments->waiting->result);
554555

@@ -588,7 +589,8 @@ VALUE IO_Event_Selector_URing_io_wait(VALUE self, VALUE fiber, VALUE io, VALUE e
588589
struct io_wait_arguments io_wait_arguments = {
589590
.selector = selector,
590591
.waiting = &waiting,
591-
.flags = flags
592+
.io = io,
593+
.flags = flags,
592594
};
593595

594596
return rb_ensure(io_wait_transfer, (VALUE)&io_wait_arguments, io_wait_ensure, (VALUE)&io_wait_arguments);

0 commit comments

Comments
 (0)