Class: Event::Selector::URing
- Inherits:
-
Object
- Object
- Event::Selector::URing
- Defined in:
- ext/event/selector/uring.c
Instance Method Summary collapse
- #close ⇒ Object
- #initialize(loop) ⇒ Object constructor
- #io_close(io) ⇒ Object
- #io_read(fiber, io, buffer, _length) ⇒ Object
- #io_wait(fiber, io, events) ⇒ Object
- #io_write(fiber, io, buffer, _length) ⇒ Object
- #process_wait(fiber, pid, flags) ⇒ Object
- #push(fiber) ⇒ Object
- #raise(*args) ⇒ Object
- #ready? ⇒ Boolean
- #select(duration) ⇒ Object
- #transfer(*args) ⇒ Object
- #yield ⇒ Object
Constructor Details
#initialize(loop) ⇒ Object
93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 |
# File 'ext/event/selector/uring.c', line 93
VALUE Event_Selector_URing_initialize(VALUE self, VALUE loop) {
struct Event_Selector_URing *data = NULL;
TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
Event_Selector_initialize(&data->backend, loop);
int result = io_uring_queue_init(URING_ENTRIES, &data->ring, 0);
if (result < 0) {
rb_syserr_fail(-result, "io_uring_queue_init");
}
rb_update_max_fd(data->ring.ring_fd);
return self;
}
|
Instance Method Details
#close ⇒ Object
109 110 111 112 113 114 115 116 |
# File 'ext/event/selector/uring.c', line 109
VALUE Event_Selector_URing_close(VALUE self) {
struct Event_Selector_URing *data = NULL;
TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
close_internal(data);
return Qnil;
}
|
#io_close(io) ⇒ Object
459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 |
# File 'ext/event/selector/uring.c', line 459
VALUE Event_Selector_URing_io_close(VALUE self, VALUE io) {
struct Event_Selector_URing *data = NULL;
TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
int descriptor = Event_Selector_io_descriptor(io);
if (ASYNC_CLOSE) {
struct io_uring_sqe *sqe = io_get_sqe(data);
io_uring_prep_close(sqe, descriptor);
io_uring_sqe_set_data(sqe, NULL);
io_uring_submit_now(data);
} else {
close(descriptor);
}
// We don't wait for the result of close since it has no use in pratice:
return Qtrue;
}
|
#io_read(fiber, io, buffer, _length) ⇒ Object
372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 |
# File 'ext/event/selector/uring.c', line 372
VALUE Event_Selector_URing_io_read(VALUE self, VALUE fiber, VALUE io, VALUE buffer, VALUE _length) {
struct Event_Selector_URing *data = NULL;
TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
int descriptor = Event_Selector_io_descriptor(io);
void *base;
size_t size;
rb_io_buffer_get_mutable(buffer, &base, &size);
size_t offset = 0;
size_t length = NUM2SIZET(_length);
while (length > 0) {
size_t maximum_size = size - offset;
int result = io_read(data, fiber, descriptor, (char*)base+offset, maximum_size);
if (result == 0) {
break;
} else if (result > 0) {
offset += result;
if ((size_t)result >= length) break;
length -= result;
} else if (-result == EAGAIN || -result == EWOULDBLOCK) {
Event_Selector_URing_io_wait(self, fiber, io, RB_INT2NUM(EVENT_READABLE));
} else {
rb_syserr_fail(-result, strerror(-result));
}
}
return SIZET2NUM(offset);
}
|
#io_wait(fiber, io, events) ⇒ Object
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 |
# File 'ext/event/selector/uring.c', line 329
VALUE Event_Selector_URing_io_wait(VALUE self, VALUE fiber, VALUE io, VALUE events) {
struct Event_Selector_URing *data = NULL;
TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
int descriptor = Event_Selector_io_descriptor(io);
struct io_uring_sqe *sqe = io_get_sqe(data);
short flags = poll_flags_from_events(NUM2INT(events));
if (DEBUG) fprintf(stderr, "Event_Selector_URing_io_wait:io_uring_prep_poll_add(descriptor=%d, flags=%d, fiber=%p)\n", descriptor, flags, (void*)fiber);
io_uring_prep_poll_add(sqe, descriptor, flags);
io_uring_sqe_set_data(sqe, (void*)fiber);
// If we are going to wait, we assume that we are waiting for a while:
io_uring_submit_pending(data);
struct io_wait_arguments io_wait_arguments = {
.data = data,
.fiber = fiber,
.flags = flags
};
return rb_rescue(io_wait_transfer, (VALUE)&io_wait_arguments, io_wait_rescue, (VALUE)&io_wait_arguments);
}
|
#io_write(fiber, io, buffer, _length) ⇒ Object
421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 |
# File 'ext/event/selector/uring.c', line 421
VALUE Event_Selector_URing_io_write(VALUE self, VALUE fiber, VALUE io, VALUE buffer, VALUE _length) {
struct Event_Selector_URing *data = NULL;
TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
int descriptor = Event_Selector_io_descriptor(io);
const void *base;
size_t size;
rb_io_buffer_get_immutable(buffer, &base, &size);
size_t offset = 0;
size_t length = NUM2SIZET(_length);
if (length > size) {
rb_raise(rb_eRuntimeError, "Length exceeds size of buffer!");
}
while (length > 0) {
int result = io_write(data, fiber, descriptor, (char*)base+offset, length);
if (result >= 0) {
offset += result;
if ((size_t)result >= length) break;
length -= result;
} else if (-result == EAGAIN || -result == EWOULDBLOCK) {
Event_Selector_URing_io_wait(self, fiber, io, RB_INT2NUM(EVENT_WRITABLE));
} else {
rb_syserr_fail(-result, strerror(-result));
}
}
return SIZET2NUM(offset);
}
|
#process_wait(fiber, pid, flags) ⇒ Object
245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 |
# File 'ext/event/selector/uring.c', line 245
VALUE Event_Selector_URing_process_wait(VALUE self, VALUE fiber, VALUE pid, VALUE flags) {
struct Event_Selector_URing *data = NULL;
TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
struct process_wait_arguments process_wait_arguments = {
.data = data,
.pid = NUM2PIDT(pid),
.flags = NUM2INT(flags),
};
process_wait_arguments.descriptor = pidfd_open(process_wait_arguments.pid, 0);
rb_update_max_fd(process_wait_arguments.descriptor);
struct io_uring_sqe *sqe = io_get_sqe(data);
if (DEBUG) fprintf(stderr, "Event_Selector_URing_process_wait:io_uring_prep_poll_add(%p)\n", (void*)fiber);
io_uring_prep_poll_add(sqe, process_wait_arguments.descriptor, POLLIN|POLLHUP|POLLERR);
io_uring_sqe_set_data(sqe, (void*)fiber);
io_uring_submit_pending(data);
return rb_ensure(process_wait_transfer, (VALUE)&process_wait_arguments, process_wait_ensure, (VALUE)&process_wait_arguments);
}
|
#push(fiber) ⇒ Object
138 139 140 141 142 143 144 145 146 |
# File 'ext/event/selector/uring.c', line 138
VALUE Event_Selector_URing_push(VALUE self, VALUE fiber)
{
struct Event_Selector_URing *data = NULL;
TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
Event_Selector_queue_push(&data->backend, fiber);
return Qnil;
}
|
#raise(*args) ⇒ Object
148 149 150 151 152 153 154 |
# File 'ext/event/selector/uring.c', line 148
VALUE Event_Selector_URing_raise(int argc, VALUE *argv, VALUE self)
{
struct Event_Selector_URing *data = NULL;
TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
return Event_Selector_wait_and_raise(&data->backend, argc, argv);
}
|
#ready? ⇒ Boolean
156 157 158 159 160 161 |
# File 'ext/event/selector/uring.c', line 156
VALUE Event_Selector_URing_ready_p(VALUE self) {
struct Event_Selector_URing *data = NULL;
TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
return data->backend.ready ? Qtrue : Qfalse;
}
|
#select(duration) ⇒ Object
579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 |
# File 'ext/event/selector/uring.c', line 579
VALUE Event_Selector_URing_select(VALUE self, VALUE duration) {
struct Event_Selector_URing *data = NULL;
TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
int ready = Event_Selector_queue_flush(&data->backend);
int result = select_process_completions(&data->ring);
// If the ready list was empty and we didn't process any completions:
if (!ready && result == 0) {
// We might need to wait for events:
struct select_arguments arguments = {
.data = data,
.timeout = NULL,
};
arguments.timeout = make_timeout(duration, &arguments.storage);
if (!data->backend.ready && !timeout_nonblocking(arguments.timeout)) {
// This is a blocking operation, we wait for events:
result = select_internal_without_gvl(&arguments);
} else {
// The timeout specified required "nonblocking" behaviour so we just flush the SQ if required:
io_uring_submit_flush(data);
}
// After waiting/flushing the SQ, check if there are any completions:
result = select_process_completions(&data->ring);
}
return RB_INT2NUM(result);
}
|
#transfer(*args) ⇒ Object
118 119 120 121 122 123 124 125 126 |
# File 'ext/event/selector/uring.c', line 118
VALUE Event_Selector_URing_transfer(int argc, VALUE *argv, VALUE self)
{
struct Event_Selector_URing *data = NULL;
TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
Event_Selector_wait_and_transfer(&data->backend, argc, argv);
return Qnil;
}
|
#yield ⇒ Object
128 129 130 131 132 133 134 135 136 |
# File 'ext/event/selector/uring.c', line 128
VALUE Event_Selector_URing_yield(VALUE self)
{
struct Event_Selector_URing *data = NULL;
TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
Event_Selector_yield(&data->backend);
return Qnil;
}
|