Class: IO::Event::Selector::URing
- Inherits:
-
Object
- Object
- IO::Event::Selector::URing
- Defined in:
- ext/io/event/selector/uring.c
Instance Method Summary collapse
- #close ⇒ Object
- #closed? ⇒ Boolean
- #idle_duration ⇒ Object
- #initialize(loop) ⇒ Object constructor
- #io_close(_descriptor) ⇒ Object
- #io_pread(fiber, io, buffer, _from, _first, _second) ⇒ Object
- #io_pwrite(fiber, io, buffer, _from, _first, _second) ⇒ Object
- #io_read(*args) ⇒ Object
- #io_wait(fiber, io, events) ⇒ Object
- #io_write(*args) ⇒ Object
- #loop ⇒ Object
- #process_wait(fiber, _pid, _flags) ⇒ Object
- #push(fiber) ⇒ Object
- #raise(*args) ⇒ Object
- #ready? ⇒ Boolean
- #resume(*args) ⇒ Object
- #select(duration) ⇒ Object
- #transfer ⇒ Object
- #wakeup ⇒ Object
- #yield ⇒ Object
Constructor Details
#initialize(loop) ⇒ Object
325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 |
# File 'ext/io/event/selector/uring.c', line 325
VALUE IO_Event_Selector_URing_initialize(VALUE self, VALUE loop) {
struct IO_Event_Selector_URing *selector = NULL;
TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
IO_Event_Selector_initialize(&selector->backend, self, loop);
unsigned int flags = 0;
// IORING_SETUP_SINGLE_ISSUER (kernel 6.0+): only the owner thread submits SQEs.
// Safe here because wakeup() uses eventfd (no ring access from other threads).
#ifdef IORING_SETUP_SINGLE_ISSUER
flags |= IORING_SETUP_SINGLE_ISSUER;
#endif
// IORING_SETUP_DEFER_TASKRUN (kernel 6.1+, requires SINGLE_ISSUER): defer io_uring
// task work to the application thread rather than a kernel thread, reducing
// cross-CPU signaling overhead.
#ifdef IORING_SETUP_DEFER_TASKRUN
flags |= IORING_SETUP_DEFER_TASKRUN;
#endif
// IORING_SETUP_TASKRUN_FLAG (kernel 5.19+, always available alongside
// DEFER_TASKRUN): the kernel surfaces IORING_SQ_TASKRUN in sq.flags whenever
// task work is pending, so select() can skip the io_uring_get_events()
// syscall when there is nothing deferred to flush.
#ifdef IORING_SETUP_TASKRUN_FLAG
flags |= IORING_SETUP_TASKRUN_FLAG;
#endif
// IORING_SETUP_SUBMIT_ALL (kernel 5.18+): keep processing the rest of the SQE
// batch even when one fails, reducing the frequency of short submits.
#ifdef IORING_SETUP_SUBMIT_ALL
flags |= IORING_SETUP_SUBMIT_ALL;
#endif
int result = io_uring_queue_init(URING_ENTRIES, &selector->ring, flags);
#ifdef IORING_SETUP_SUBMIT_ALL
if (result == -EINVAL) {
// IORING_SETUP_SUBMIT_ALL was added in Linux 5.18; retry without it.
if (DEBUG) fprintf(stderr, "IO_Event_Selector_URing_initialize: no IORING_SETUP_SUBMIT_ALL\n");
flags &= ~IORING_SETUP_SUBMIT_ALL;
result = io_uring_queue_init(URING_ENTRIES, &selector->ring, flags);
}
#endif
if (result < 0) {
rb_syserr_fail(-result, "IO_Event_Selector_URing_initialize:io_uring_queue_init");
}
selector->owner = getpid();
rb_update_max_fd(selector->ring.ring_fd);
// Interrupt for cross-thread wakeup: another thread calls signal(); the owner
// thread submits an async read before each blocking wait so the ring wakes up
// without the waking thread ever touching the SQ.
IO_Event_Interrupt_open(&selector->interrupt);
if (selector->interrupt.descriptor < 0) {
io_uring_queue_exit(&selector->ring);
selector->ring.ring_fd = -1;
rb_sys_fail("IO_Event_Selector_URing_initialize:IO_Event_Interrupt_open");
}
return self;
}
|
Instance Method Details
#close ⇒ Object
404 405 406 407 408 409 410 411 |
# File 'ext/io/event/selector/uring.c', line 404
VALUE IO_Event_Selector_URing_close(VALUE self) {
struct IO_Event_Selector_URing *selector = NULL;
TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
close_internal(selector);
return Qnil;
}
|
#closed? ⇒ Boolean
413 414 415 416 417 418 |
# File 'ext/io/event/selector/uring.c', line 413
VALUE IO_Event_Selector_URing_closed_p(VALUE self) {
struct IO_Event_Selector_URing *selector = NULL;
TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
return selector->ring.ring_fd < 0 || selector->owner != getpid() ? Qtrue : Qfalse;
}
|
#idle_duration ⇒ Object
395 396 397 398 399 400 401 402 |
# File 'ext/io/event/selector/uring.c', line 395
VALUE IO_Event_Selector_URing_idle_duration(VALUE self) {
struct IO_Event_Selector_URing *selector = NULL;
TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
double duration = selector->idle_duration.tv_sec + (selector->idle_duration.tv_nsec / 1000000000.0);
return DBL2NUM(duration);
}
|
#io_close(_descriptor) ⇒ Object
1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 |
# File 'ext/io/event/selector/uring.c', line 1462
VALUE IO_Event_Selector_URing_io_close(VALUE self, VALUE _descriptor) {
struct IO_Event_Selector_URing *selector = NULL;
TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
// Ruby's fiber scheduler `io_close` hook is invoked with a raw integer file descriptor (Ruby 4.0+); it does not pass the `IO` object.
int descriptor = RB_NUM2INT(_descriptor);
if (ASYNC_CLOSE) {
struct io_uring_sqe *sqe = io_get_sqe(selector);
io_uring_prep_close(sqe, descriptor);
io_uring_sqe_set_data(sqe, NULL);
io_uring_submit_now(selector);
// It would be nice to explore not flushing immediately, but instead deferring to the next select cycle.
// The problem with this approach is that if the user expects the file descriptor to be closed immediately, (e.g. before fork), it may not be closed in time.
// io_uring_submit_pending(selector);
} else {
close(descriptor);
}
// We don't wait for the result of close since it has no use in practice:
return Qtrue;
}
|
#io_pread(fiber, io, buffer, _from, _first, _second) ⇒ Object
1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 |
# File 'ext/io/event/selector/uring.c', line 1137
VALUE IO_Event_Selector_URing_io_pread(VALUE self, VALUE fiber, VALUE io, VALUE buffer, VALUE _from, VALUE _first, VALUE _second) {
struct IO_Event_Selector_URing *selector = NULL;
TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
#if RUBY_FIBER_SCHEDULER_VERSION >= 4
struct io_read_locked_arguments arguments = {
.selector = selector,
.fiber = fiber,
.io = io,
.from = NUM2OFFT(_from),
.offset = NUM2SIZET(_first),
.length = NUM2SIZET(_second),
.positional = true,
};
return rb_io_buffer_locked_for_writing(buffer, io_read_locked, (VALUE)&arguments);
#else
void *base;
size_t size;
rb_io_buffer_get_bytes_for_writing(buffer, &base, &size);
size_t length = NUM2SIZET(_first);
size_t offset = NUM2SIZET(_second);
size_t total = 0;
// Ensure offset is within the bounds of the buffer to avoid size_t underflow and out-of-bounds pointer arithmetic on (char *)base + offset.
if (offset > size) {
return rb_fiber_scheduler_io_result(-1, EINVAL);
} else if (offset == size) {
return rb_fiber_scheduler_io_result(0, 0);
}
off_t from = NUM2OFFT(_from);
int descriptor = IO_Event_Selector_io_descriptor(io);
size_t maximum_size = size - offset;
while (maximum_size) {
int result = io_read(selector, fiber, descriptor, (char*)base+offset, maximum_size, from);
if (result > 0) {
total += result;
offset += result;
from += result;
if ((size_t)result >= length) break;
length -= result;
} else if (result == 0) {
break;
} else if (length > 0 && IO_Event_try_again(-result)) {
IO_Event_Selector_URing_io_wait(self, fiber, io, RB_INT2NUM(IO_EVENT_READABLE));
} else {
return rb_fiber_scheduler_io_result(-1, -result);
}
maximum_size = size - offset;
}
return rb_fiber_scheduler_io_result(total, 0);
#endif
}
|
#io_pwrite(fiber, io, buffer, _from, _first, _second) ⇒ Object
1394 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448 1449 1450 1451 1452 1453 1454 |
# File 'ext/io/event/selector/uring.c', line 1394
VALUE IO_Event_Selector_URing_io_pwrite(VALUE self, VALUE fiber, VALUE io, VALUE buffer, VALUE _from, VALUE _first, VALUE _second) {
struct IO_Event_Selector_URing *selector = NULL;
TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
#if RUBY_FIBER_SCHEDULER_VERSION >= 4
struct io_write_locked_arguments arguments = {
.selector = selector,
.fiber = fiber,
.io = io,
.from = NUM2OFFT(_from),
.offset = NUM2SIZET(_first),
.length = NUM2SIZET(_second),
.positional = true,
};
return rb_io_buffer_locked_for_reading(buffer, io_write_locked, (VALUE)&arguments);
#else
const void *base;
size_t size;
rb_io_buffer_get_bytes_for_reading(buffer, &base, &size);
size_t length = NUM2SIZET(_first);
size_t offset = NUM2SIZET(_second);
size_t total = 0;
if (length > size) {
rb_raise(rb_eRuntimeError, "Length exceeds size of buffer!");
}
// Ensure offset is within the bounds of the buffer to avoid size_t underflow and out-of-bounds pointer arithmetic on (char *)base + offset.
if (offset > size) {
return rb_fiber_scheduler_io_result(-1, EINVAL);
} else if (offset == size) {
return rb_fiber_scheduler_io_result(0, 0);
}
off_t from = NUM2OFFT(_from);
int descriptor = IO_Event_Selector_io_descriptor(io);
size_t maximum_size = size - offset;
while (maximum_size) {
int result = io_write(selector, fiber, descriptor, (char*)base+offset, maximum_size, from);
if (result > 0) {
total += result;
offset += result;
from += result;
if ((size_t)result >= length) break;
length -= result;
} else if (result == 0) {
break;
} else if (length > 0 && IO_Event_try_again(-result)) {
IO_Event_Selector_URing_io_wait(self, fiber, io, RB_INT2NUM(IO_EVENT_WRITABLE));
} else {
return rb_fiber_scheduler_io_result(-1, -result);
}
maximum_size = size - offset;
}
return rb_fiber_scheduler_io_result(total, 0);
#endif
}
|
#io_read(*args) ⇒ Object
1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 |
# File 'ext/io/event/selector/uring.c', line 1051
VALUE IO_Event_Selector_URing_io_read(VALUE self, VALUE fiber, VALUE io, VALUE buffer, VALUE _first, VALUE _second) {
struct IO_Event_Selector_URing *selector = NULL;
TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
#if RUBY_FIBER_SCHEDULER_VERSION >= 4
struct io_read_locked_arguments arguments = {
.selector = selector,
.fiber = fiber,
.io = io,
.offset = NUM2SIZET(_first),
.length = NUM2SIZET(_second),
.positional = false,
};
return rb_io_buffer_locked_for_writing(buffer, io_read_locked, (VALUE)&arguments);
#else
void *base;
size_t size;
rb_io_buffer_get_bytes_for_writing(buffer, &base, &size);
size_t length = NUM2SIZET(_first);
size_t offset = NUM2SIZET(_second);
size_t total = 0;
// Ensure offset is within the bounds of the buffer to avoid size_t underflow and out-of-bounds pointer arithmetic on (char *)base + offset.
if (offset > size) {
return rb_fiber_scheduler_io_result(-1, EINVAL);
} else if (offset == size) {
return rb_fiber_scheduler_io_result(0, 0);
}
int descriptor = IO_Event_Selector_io_descriptor(io);
off_t from = io_seekable(descriptor);
size_t maximum_size = size - offset;
// Are we performing a non-blocking read?
if (!length) {
// If the (maximum) length is zero, that indicates we just want to read whatever is available without blocking.
// If we schedule this read into the URing, it will block until data is available, rather than returning immediately.
int state = IO_Event_Selector_nonblock_set(descriptor);
int result = read(descriptor, (char*)base+offset, maximum_size);
int error = errno;
IO_Event_Selector_nonblock_restore(descriptor, state);
return rb_fiber_scheduler_io_result(result, error);
}
while (maximum_size) {
int result = io_read(selector, fiber, descriptor, (char*)base+offset, maximum_size, from);
if (result > 0) {
total += result;
offset += result;
if ((size_t)result >= length) break;
length -= result;
} else if (result == 0) {
break;
} else if (length > 0 && IO_Event_try_again(-result)) {
IO_Event_Selector_URing_io_wait(self, fiber, io, RB_INT2NUM(IO_EVENT_READABLE));
} else {
return rb_fiber_scheduler_io_result(-1, -result);
}
maximum_size = size - offset;
}
return rb_fiber_scheduler_io_result(total, 0);
#endif
}
|
#io_wait(fiber, io, events) ⇒ Object
875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 |
# File 'ext/io/event/selector/uring.c', line 875
VALUE IO_Event_Selector_URing_io_wait(VALUE self, VALUE fiber, VALUE io, VALUE events) {
struct IO_Event_Selector_URing *selector = NULL;
TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
int descriptor = IO_Event_Selector_io_descriptor(io);
short flags = poll_flags_from_events(NUM2INT(events));
if (DEBUG) fprintf(stderr, "IO_Event_Selector_URing_io_wait:io_uring_prep_poll_add(descriptor=%d, flags=%d, fiber=%p)\n", descriptor, flags, (void*)fiber);
struct IO_Event_Selector_URing_Waiting waiting = {
.fiber = fiber,
};
RB_OBJ_WRITTEN(self, Qundef, fiber);
struct IO_Event_Selector_URing_Completion *completion = IO_Event_Selector_URing_Completion_acquire(selector, &waiting);
struct io_uring_sqe *sqe = io_get_sqe(selector);
io_uring_prep_poll_add(sqe, descriptor, flags);
io_uring_sqe_set_data(sqe, completion);
// If we are going to wait, we assume that we are waiting for a while:
io_uring_submit_pending(selector);
struct io_wait_arguments io_wait_arguments = {
.selector = selector,
.waiting = &waiting,
.flags = flags
};
return rb_ensure(io_wait_transfer, (VALUE)&io_wait_arguments, io_wait_ensure, (VALUE)&io_wait_arguments);
}
|
#io_write(*args) ⇒ Object
1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348 1349 1350 1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377 |
# File 'ext/io/event/selector/uring.c', line 1318
VALUE IO_Event_Selector_URing_io_write(VALUE self, VALUE fiber, VALUE io, VALUE buffer, VALUE _first, VALUE _second) {
struct IO_Event_Selector_URing *selector = NULL;
TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
#if RUBY_FIBER_SCHEDULER_VERSION >= 4
struct io_write_locked_arguments arguments = {
.selector = selector,
.fiber = fiber,
.io = io,
.offset = NUM2SIZET(_first),
.length = NUM2SIZET(_second),
.positional = false,
};
return rb_io_buffer_locked_for_reading(buffer, io_write_locked, (VALUE)&arguments);
#else
const void *base;
size_t size;
rb_io_buffer_get_bytes_for_reading(buffer, &base, &size);
size_t length = NUM2SIZET(_first);
size_t offset = NUM2SIZET(_second);
size_t total = 0;
if (length > size) {
rb_raise(rb_eRuntimeError, "Length exceeds size of buffer!");
}
// Ensure offset is within the bounds of the buffer to avoid size_t underflow and out-of-bounds pointer arithmetic on (char *)base + offset.
if (offset > size) {
return rb_fiber_scheduler_io_result(-1, EINVAL);
} else if (offset == size) {
return rb_fiber_scheduler_io_result(0, 0);
}
int descriptor = IO_Event_Selector_io_descriptor(io);
off_t from = io_seekable(descriptor);
size_t maximum_size = size - offset;
while (maximum_size) {
int result = io_write(selector, fiber, descriptor, (char*)base+offset, maximum_size, from);
if (result > 0) {
total += result;
offset += result;
if ((size_t)result >= length) break;
length -= result;
} else if (result == 0) {
break;
} else if (length > 0 && IO_Event_try_again(-result)) {
IO_Event_Selector_URing_io_wait(self, fiber, io, RB_INT2NUM(IO_EVENT_WRITABLE));
} else {
return rb_fiber_scheduler_io_result(-1, -result);
}
maximum_size = size - offset;
}
return rb_fiber_scheduler_io_result(total, 0);
#endif
}
|
#loop ⇒ Object
388 389 390 391 392 393 |
# File 'ext/io/event/selector/uring.c', line 388
VALUE IO_Event_Selector_URing_loop(VALUE self) {
struct IO_Event_Selector_URing *selector = NULL;
TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
return selector->backend.loop;
}
|
#process_wait(fiber, _pid, _flags) ⇒ Object
744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 |
# File 'ext/io/event/selector/uring.c', line 744
VALUE IO_Event_Selector_URing_process_wait(VALUE self, VALUE fiber, VALUE _pid, VALUE _flags) {
struct IO_Event_Selector_URing *selector = NULL;
TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
pid_t pid = NUM2PIDT(_pid);
int flags = NUM2INT(_flags);
#ifndef IO_EVENT_SELECTOR_URING_USE_WAITID
// `pidfd_open` can only refer to a specific process, so waiting for any child or a process group (pid <= 0) is delegated to the threaded fallback:
if (pid <= 0) {
return IO_Event_Selector_process_wait(pid, flags);
}
int descriptor = pidfd_open(pid, 0);
if (descriptor < 0) {
rb_syserr_fail(errno, "IO_Event_Selector_URing_process_wait:pidfd_open");
}
rb_update_max_fd(descriptor);
#endif
struct IO_Event_Selector_URing_Waiting waiting = {
.fiber = fiber,
};
RB_OBJ_WRITTEN(self, Qundef, fiber);
struct IO_Event_Selector_URing_Completion *completion = IO_Event_Selector_URing_Completion_acquire(selector, &waiting);
struct process_wait_arguments process_wait_arguments = {
.selector = selector,
.waiting = &waiting,
.pid = pid,
.flags = flags,
#ifdef IO_EVENT_SELECTOR_URING_USE_WAITID
.siginfo = {0},
#else
.descriptor = descriptor,
#endif
};
struct io_uring_sqe *sqe = io_get_sqe(selector);
#ifdef IO_EVENT_SELECTOR_URING_USE_WAITID
id_t id;
idtype_t idtype = process_waitid_type(pid, &id);
if (DEBUG) fprintf(stderr, "IO_Event_Selector_URing_process_wait:io_uring_prep_waitid(fiber=%p, idtype=%d, id=%d, flags=%d)\n", (void*)fiber, idtype, (int)id, flags);
#ifdef HAVE_RB_PROCESS_STATUS_FOR
// Reap the child directly; the completion contains enough information to construct the Ruby process status value:
io_uring_prep_waitid(sqe, idtype, id, &process_wait_arguments.siginfo, WEXITED, 0);
#else
// `WNOWAIT` leaves the child in a waitable state so we can reap it with `rb_process_status_wait` afterwards and build a correct `Process::Status`:
io_uring_prep_waitid(sqe, idtype, id, &process_wait_arguments.siginfo, WEXITED | WNOWAIT, 0);
#endif
#else
if (DEBUG) fprintf(stderr, "IO_Event_Selector_URing_process_wait:io_uring_prep_poll_add(%p)\n", (void*)fiber);
io_uring_prep_poll_add(sqe, descriptor, POLLIN|POLLHUP|POLLERR);
#endif
io_uring_sqe_set_data(sqe, completion);
io_uring_submit_pending(selector);
return rb_ensure(process_wait_transfer, (VALUE)&process_wait_arguments, process_wait_ensure, (VALUE)&process_wait_arguments);
}
|
#push(fiber) ⇒ Object
444 445 446 447 448 449 450 451 452 |
# File 'ext/io/event/selector/uring.c', line 444
VALUE IO_Event_Selector_URing_push(VALUE self, VALUE fiber)
{
struct IO_Event_Selector_URing *selector = NULL;
TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
IO_Event_Selector_ready_push(&selector->backend, fiber);
return Qnil;
}
|
#raise(*args) ⇒ Object
454 455 456 457 458 459 460 |
# File 'ext/io/event/selector/uring.c', line 454
VALUE IO_Event_Selector_URing_raise(int argc, VALUE *argv, VALUE self)
{
struct IO_Event_Selector_URing *selector = NULL;
TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
return IO_Event_Selector_raise(&selector->backend, argc, argv);
}
|
#ready? ⇒ Boolean
462 463 464 465 466 467 |
# File 'ext/io/event/selector/uring.c', line 462
VALUE IO_Event_Selector_URing_ready_p(VALUE self) {
struct IO_Event_Selector_URing *selector = NULL;
TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
return selector->backend.ready ? Qtrue : Qfalse;
}
|
#resume(*args) ⇒ Object
428 429 430 431 432 433 434 |
# File 'ext/io/event/selector/uring.c', line 428
VALUE IO_Event_Selector_URing_resume(int argc, VALUE *argv, VALUE self)
{
struct IO_Event_Selector_URing *selector = NULL;
TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
return IO_Event_Selector_resume(&selector->backend, argc, argv);
}
|
#select(duration) ⇒ Object
1642 1643 1644 1645 1646 1647 1648 1649 1650 1651 1652 1653 1654 1655 1656 1657 1658 1659 1660 1661 1662 1663 1664 1665 1666 1667 1668 1669 1670 1671 1672 1673 1674 1675 1676 1677 1678 1679 1680 1681 1682 1683 1684 1685 1686 1687 1688 1689 1690 1691 1692 1693 1694 1695 1696 1697 1698 1699 1700 1701 1702 1703 1704 1705 1706 1707 1708 1709 |
# File 'ext/io/event/selector/uring.c', line 1642
VALUE IO_Event_Selector_URing_select(VALUE self, VALUE duration) {
struct IO_Event_Selector_URing *selector = NULL;
TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
selector->idle_duration.tv_sec = 0;
selector->idle_duration.tv_nsec = 0;
// Flush any pending events:
io_uring_submit_flush(selector);
#ifdef IORING_SETUP_DEFER_TASKRUN
// With DEFER_TASKRUN the kernel holds completions as "deferred task work"
// rather than placing them directly into the CQ. We need to flush that work
// into the CQ so the non-blocking select_process_completions below can see
// it. With TASKRUN_FLAG enabled the kernel sets IORING_SQ_TASKRUN in
// sq.flags whenever task work is pending; a relaxed atomic load is enough
// to check, and we only pay for an io_uring_enter syscall (via
// io_uring_get_events) when there is actually deferred work to flush.
if (selector->ring.flags & IORING_SETUP_DEFER_TASKRUN) {
#ifdef IORING_SETUP_TASKRUN_FLAG
unsigned sq_flags = __atomic_load_n(selector->ring.sq.kflags, __ATOMIC_RELAXED);
if (sq_flags & IORING_SQ_TASKRUN)
#endif
{
io_uring_get_events(&selector->ring);
}
}
#endif
int ready = IO_Event_Selector_ready_flush(&selector->backend);
int completed = select_process_completions(selector);
// If we:
// 1. Didn't process any ready fibers, and
// 2. Didn't process any events from non-blocking select (above), and
// 3. There are no items in the ready list,
// then we can perform a blocking select.
if (!ready && !completed && !selector->backend.ready) {
// We might need to wait for events:
struct select_arguments arguments = {
.selector = selector,
.result = 0,
.timeout = NULL,
};
arguments.timeout = make_timeout(duration, &arguments.storage);
if (!selector->backend.ready && select_blocking_allowed(arguments.timeout)) {
struct timespec start_time;
IO_Event_Time_current(&start_time);
// This is a blocking operation, we wait for events:
int result = select_internal_without_gvl(&arguments);
struct timespec end_time;
IO_Event_Time_current(&end_time);
IO_Event_Time_elapsed(&start_time, &end_time, &selector->idle_duration);
// After waiting/flushing the SQ, check if there are any completions:
if (result > 0) {
completed = select_process_completions(selector);
}
}
}
return RB_INT2NUM(completed);
}
|
#transfer ⇒ Object
420 421 422 423 424 425 426 |
# File 'ext/io/event/selector/uring.c', line 420
VALUE IO_Event_Selector_URing_transfer(VALUE self)
{
struct IO_Event_Selector_URing *selector = NULL;
TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
return IO_Event_Selector_loop_yield(&selector->backend);
}
|
#wakeup ⇒ Object
1711 1712 1713 1714 1715 1716 1717 1718 1719 1720 1721 1722 1723 |
# File 'ext/io/event/selector/uring.c', line 1711
VALUE IO_Event_Selector_URing_wakeup(VALUE self) {
struct IO_Event_Selector_URing *selector = NULL;
TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
// Wake the selector by signalling the interrupt. This is safe from any thread
// and never touches the ring's SQ, which is required for IORING_SETUP_SINGLE_ISSUER.
if (selector->backend.blocked) {
IO_Event_Interrupt_signal(&selector->interrupt);
return Qtrue;
}
return Qfalse;
}
|
#yield ⇒ Object
436 437 438 439 440 441 442 |
# File 'ext/io/event/selector/uring.c', line 436
VALUE IO_Event_Selector_URing_yield(VALUE self)
{
struct IO_Event_Selector_URing *selector = NULL;
TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
return IO_Event_Selector_yield(&selector->backend);
}
|