12#include "eval_intern.h"
20#include "internal/thread.h"
23#include "ruby_atomic.h"
26static ID id_scheduler_close;
31static ID id_timeout_after;
32static ID id_kernel_sleep;
33static ID id_process_wait;
35static ID id_io_read, id_io_pread;
36static ID id_io_write, id_io_pwrite;
38static ID id_io_select;
41static ID id_address_resolve;
43static ID id_blocking_operation_wait;
44static ID id_fiber_interrupt;
46static ID id_fiber_schedule;
49static VALUE rb_cFiberSchedulerBlockingOperation;
58 RB_FIBER_SCHEDULER_BLOCKING_OPERATION_STATUS_QUEUED,
59 RB_FIBER_SCHEDULER_BLOCKING_OPERATION_STATUS_EXECUTING,
60 RB_FIBER_SCHEDULER_BLOCKING_OPERATION_STATUS_COMPLETED,
61 RB_FIBER_SCHEDULER_BLOCKING_OPERATION_STATUS_CANCELLED
62} rb_fiber_blocking_operation_status_t;
65 void *(*function)(
void *);
79blocking_operation_mark(
void *ptr)
85blocking_operation_free(
void *ptr)
88 ruby_xfree(blocking_operation);
92blocking_operation_memsize(
const void *ptr)
98 "Fiber::Scheduler::BlockingOperation",
100 blocking_operation_mark,
101 blocking_operation_free,
102 blocking_operation_memsize,
104 0, 0, RUBY_TYPED_FREE_IMMEDIATELY | RUBY_TYPED_WB_PROTECTED
111blocking_operation_alloc(
VALUE klass)
116 blocking_operation->function = NULL;
117 blocking_operation->data = NULL;
118 blocking_operation->unblock_function = NULL;
119 blocking_operation->data2 = NULL;
120 blocking_operation->flags = 0;
121 blocking_operation->state = NULL;
122 blocking_operation->status = RB_FIBER_SCHEDULER_BLOCKING_OPERATION_STATUS_QUEUED;
131get_blocking_operation(
VALUE obj)
135 return blocking_operation;
147blocking_operation_call(
VALUE self)
151 if (blocking_operation->status != RB_FIBER_SCHEDULER_BLOCKING_OPERATION_STATUS_QUEUED) {
152 rb_raise(
rb_eRuntimeError,
"Blocking operation has already been executed!");
155 if (blocking_operation->function == NULL) {
156 rb_raise(
rb_eRuntimeError,
"Blocking operation has no function to execute!");
159 if (blocking_operation->state == NULL) {
164 blocking_operation->status = RB_FIBER_SCHEDULER_BLOCKING_OPERATION_STATUS_EXECUTING;
167 blocking_operation->state->result =
rb_nogvl(blocking_operation->function, blocking_operation->data,
168 blocking_operation->unblock_function, blocking_operation->data2,
169 blocking_operation->flags);
170 blocking_operation->state->saved_errno = rb_errno();
173 blocking_operation->status = RB_FIBER_SCHEDULER_BLOCKING_OPERATION_STATUS_COMPLETED;
191 return get_blocking_operation(self);
206 if (blocking_operation == NULL) {
210 if (blocking_operation->function == NULL || blocking_operation->state == NULL) {
215 rb_thread_resolve_unblock_function(&blocking_operation->unblock_function, &blocking_operation->data2, GET_THREAD());
218 rb_atomic_t expected = RB_FIBER_SCHEDULER_BLOCKING_OPERATION_STATUS_QUEUED;
219 if (
RUBY_ATOMIC_CAS(blocking_operation->status, expected, RB_FIBER_SCHEDULER_BLOCKING_OPERATION_STATUS_EXECUTING) != expected) {
225 blocking_operation->state->result = blocking_operation->function(blocking_operation->data);
226 blocking_operation->state->saved_errno =
errno;
229 expected = RB_FIBER_SCHEDULER_BLOCKING_OPERATION_STATUS_EXECUTING;
230 if (
RUBY_ATOMIC_CAS(blocking_operation->status, expected, RB_FIBER_SCHEDULER_BLOCKING_OPERATION_STATUS_COMPLETED) == expected) {
235 blocking_operation->state->saved_errno = EINTR;
247rb_fiber_scheduler_blocking_operation_new(
void *(*function)(
void *),
void *data,
251 VALUE self = blocking_operation_alloc(rb_cFiberSchedulerBlockingOperation);
254 blocking_operation->function = function;
255 blocking_operation->data = data;
256 blocking_operation->unblock_function = unblock_function;
257 blocking_operation->data2 = data2;
258 blocking_operation->flags = flags;
259 blocking_operation->state = state;
317Init_Fiber_Scheduler(
void)
340 id_blocking_operation_wait =
rb_intern_const(
"blocking_operation_wait");
347 rb_cFiberSchedulerBlockingOperation =
rb_class_new(rb_cObject);
349 rb_define_method(rb_cFiberSchedulerBlockingOperation,
"call", blocking_operation_call, 0);
352 rb_gc_register_mark_object(rb_cFiberSchedulerBlockingOperation);
366 rb_define_method(rb_cFiberScheduler,
"timeout_after", rb_fiber_scheduler_timeout_after, 3);
382 return thread->scheduler;
386verify_interface(
VALUE scheduler)
389 rb_raise(rb_eArgError,
"Scheduler must implement #block");
393 rb_raise(rb_eArgError,
"Scheduler must implement #unblock");
397 rb_raise(rb_eArgError,
"Scheduler must implement #kernel_sleep");
401 rb_raise(rb_eArgError,
"Scheduler must implement #io_wait");
405 rb_warn(
"Scheduler should implement #fiber_interrupt");
410fiber_scheduler_close(
VALUE scheduler)
416fiber_scheduler_close_ensure(
VALUE _thread)
419 thread->scheduler =
Qnil;
432 if (scheduler !=
Qnil) {
433 verify_interface(scheduler);
440 if (thread->scheduler !=
Qnil) {
442 rb_ensure(fiber_scheduler_close, thread->scheduler, fiber_scheduler_close_ensure, (
VALUE)thread);
445 thread->scheduler = scheduler;
447 return thread->scheduler;
451rb_fiber_scheduler_current_for_threadptr(
rb_thread_t *thread)
455 if (thread->blocking == 0) {
456 return thread->scheduler;
466 return rb_fiber_scheduler_current_for_threadptr(GET_THREAD());
471 return rb_fiber_scheduler_current_for_threadptr(rb_thread_ptr(thread));
498 if (!UNDEF_P(result))
return result;
501 if (!UNDEF_P(result))
return result;
510 return rb_float_new((
double)timeout->tv_sec + (0.000001 * timeout->tv_usec));
530 return rb_funcall(scheduler, id_kernel_sleep, 1, timeout);
536 return rb_funcallv(scheduler, id_kernel_sleep, argc, argv);
571 VALUE arguments[] = {
572 timeout, exception, message
579rb_fiber_scheduler_timeout_afterv(
VALUE scheduler,
int argc,
VALUE * argv)
606 VALUE arguments[] = {
630 return rb_funcall(scheduler, id_block, 2, blocker, timeout);
652 enum ruby_tag_type state;
657 int saved_errno =
errno;
661 int saved_interrupt_mask = ec->interrupt_mask;
662 ec->interrupt_mask |= PENDING_INTERRUPT_MASK;
665 if ((state = EC_EXEC_TAG()) == TAG_NONE) {
666 result =
rb_funcall(scheduler, id_unblock, 2, blocker, fiber);
670 ec->interrupt_mask = saved_interrupt_mask;
673 EC_JUMP_TAG(ec, state);
676 RUBY_VM_CHECK_INTS(ec);
703fiber_scheduler_io_wait(
VALUE _argument) {
706 return rb_funcallv(arguments[0], id_io_wait, 3, arguments + 1);
712 VALUE arguments[] = {
713 scheduler, io, events, timeout
717 return rb_thread_io_blocking_operation(io, fiber_scheduler_io_wait, (
VALUE)&arguments);
719 return fiber_scheduler_io_wait((
VALUE)&arguments);
747 VALUE arguments[] = {
748 readables, writables, exceptables, timeout
791fiber_scheduler_io_read(
VALUE _argument) {
794 return rb_funcallv(arguments[0], id_io_read, 4, arguments + 1);
804 VALUE arguments[] = {
809 return rb_thread_io_blocking_operation(io, fiber_scheduler_io_read, (
VALUE)&arguments);
811 return fiber_scheduler_io_read((
VALUE)&arguments);
830fiber_scheduler_io_pread(
VALUE _argument) {
833 return rb_funcallv(arguments[0], id_io_pread, 5, arguments + 1);
843 VALUE arguments[] = {
848 return rb_thread_io_blocking_operation(io, fiber_scheduler_io_pread, (
VALUE)&arguments);
850 return fiber_scheduler_io_pread((
VALUE)&arguments);
883fiber_scheduler_io_write(
VALUE _argument) {
886 return rb_funcallv(arguments[0], id_io_write, 4, arguments + 1);
896 VALUE arguments[] = {
901 return rb_thread_io_blocking_operation(io, fiber_scheduler_io_write, (
VALUE)&arguments);
903 return fiber_scheduler_io_write((
VALUE)&arguments);
923fiber_scheduler_io_pwrite(
VALUE _argument) {
926 return rb_funcallv(arguments[0], id_io_pwrite, 5, arguments + 1);
936 VALUE arguments[] = {
941 return rb_thread_io_blocking_operation(io, fiber_scheduler_io_pwrite, (
VALUE)&arguments);
943 return fiber_scheduler_io_pwrite((
VALUE)&arguments);
950 VALUE buffer = rb_io_buffer_new(base, size, RB_IO_BUFFER_LOCKED);
954 rb_io_buffer_free_locked(buffer);
962 VALUE buffer = rb_io_buffer_new((
void*)base, size, RB_IO_BUFFER_LOCKED|RB_IO_BUFFER_READONLY);
966 rb_io_buffer_free_locked(buffer);
974 VALUE buffer = rb_io_buffer_new(base, size, RB_IO_BUFFER_LOCKED);
978 rb_io_buffer_free_locked(buffer);
986 VALUE buffer = rb_io_buffer_new((
void*)base, size, RB_IO_BUFFER_LOCKED|RB_IO_BUFFER_READONLY);
990 rb_io_buffer_free_locked(buffer);
998 VALUE arguments[] = {io};
1038 VALUE arguments[] = {
1064 if (!
rb_respond_to(scheduler, id_blocking_operation_wait)) {
1069 VALUE blocking_operation = rb_fiber_scheduler_blocking_operation_new(function, data, unblock_function, data2, flags, state);
1071 VALUE result =
rb_funcall(scheduler, id_blocking_operation_wait, 1, blocking_operation);
1078 operation->function = NULL;
1079 operation->state = NULL;
1080 operation->data = NULL;
1081 operation->data2 = NULL;
1082 operation->unblock_function = NULL;
1085 if (current_status == RB_FIBER_SCHEDULER_BLOCKING_OPERATION_STATUS_QUEUED) {
1094 VALUE arguments[] = {
1099 enum ruby_tag_type state;
1103 int saved_interrupt_mask = ec->interrupt_mask;
1104 ec->interrupt_mask |= PENDING_INTERRUPT_MASK;
1107 if ((state = EC_EXEC_TAG()) == TAG_NONE) {
1112 ec->interrupt_mask = saved_interrupt_mask;
1115 EC_JUMP_TAG(ec, state);
1118 RUBY_VM_CHECK_INTS(ec);
1156 if (blocking_operation == NULL) {
1162 switch (current_state) {
1163 case RB_FIBER_SCHEDULER_BLOCKING_OPERATION_STATUS_QUEUED:
1165 if (
RUBY_ATOMIC_CAS(blocking_operation->status, current_state, RB_FIBER_SCHEDULER_BLOCKING_OPERATION_STATUS_CANCELLED) == current_state) {
1171 case RB_FIBER_SCHEDULER_BLOCKING_OPERATION_STATUS_EXECUTING:
1173 if (
RUBY_ATOMIC_CAS(blocking_operation->status, current_state, RB_FIBER_SCHEDULER_BLOCKING_OPERATION_STATUS_CANCELLED) != current_state) {
1179 if (unblock_function) {
1181 blocking_operation->unblock_function(blocking_operation->data2);
1186 case RB_FIBER_SCHEDULER_BLOCKING_OPERATION_STATUS_COMPLETED:
1187 case RB_FIBER_SCHEDULER_BLOCKING_OPERATION_STATUS_CANCELLED:
#define RUBY_ASSERT(...)
Asserts that the given expression is truthy if and only if RUBY_DEBUG is truthy.
#define RUBY_ATOMIC_CAS(var, oldval, newval)
Atomic compare-and-swap.
std::atomic< unsigned > rb_atomic_t
Type that is eligible for atomic operations.
#define RUBY_ATOMIC_LOAD(var)
Atomic load.
#define rb_define_method(klass, mid, func, arity)
Defines klass#mid.
VALUE rb_class_new(VALUE super)
Creates a new, anonymous class.
VALUE rb_define_class_under(VALUE outer, const char *name, VALUE super)
Defines a class under the namespace of outer.
#define Qundef
Old name of RUBY_Qundef.
#define SIZET2NUM
Old name of RB_SIZE2NUM.
#define Qnil
Old name of RUBY_Qnil.
VALUE rb_eRuntimeError
RuntimeError exception.
void rb_warn(const char *fmt,...)
Identical to rb_warning(), except it reports unless $VERBOSE is nil.
VALUE rb_funcall(VALUE recv, ID mid, int n,...)
Calls a method.
VALUE rb_funcall_passing_block_kw(VALUE recv, ID mid, int argc, const VALUE *argv, int kw_splat)
Identical to rb_funcallv_passing_block(), except you can specify how to handle the last element of th...
void rb_unblock_function_t(void *)
This is the type of UBFs.
int rb_respond_to(VALUE obj, ID mid)
Queries if the object responds to the method.
VALUE rb_check_funcall(VALUE recv, ID mid, int argc, const VALUE *argv)
Identical to rb_funcallv(), except it returns RUBY_Qundef instead of raising rb_eNoMethodError.
void rb_define_alloc_func(VALUE klass, rb_alloc_func_t func)
Sets the allocator function of a class.
static ID rb_intern_const(const char *str)
This is a "tiny optimisation" over rb_intern().
VALUE rb_io_timeout(VALUE io)
Get the timeout associated with the specified io object.
@ RUBY_IO_READABLE
IO::READABLE
@ RUBY_IO_WRITABLE
IO::WRITABLE
void * rb_nogvl(void *(*func)(void *), void *data1, rb_unblock_function_t *ubf, void *data2, int flags)
Identical to rb_thread_call_without_gvl(), except it additionally takes "flags" that change the behav...
#define RB_UINT2NUM
Just another name of rb_uint2num_inline.
#define RB_INT2NUM
Just another name of rb_int2num_inline.
VALUE rb_ensure(type *q, VALUE w, type *e, VALUE r)
An equivalent of ensure clause.
#define OFFT2NUM
Converts a C's off_t into an instance of rb_cInteger.
#define PIDT2NUM
Converts a C's pid_t into an instance of rb_cInteger.
#define TypedData_Get_Struct(obj, type, data_type, sval)
Obtains a C struct from inside of a wrapper Ruby object.
#define TypedData_Make_Struct(klass, type, data_type, sval)
Identical to TypedData_Wrap_Struct, except it allocates a new data region internally instead of takin...
#define errno
Ractor-aware version of errno.
VALUE rb_fiber_scheduler_blocking_operation_wait(VALUE scheduler, void *(*function)(void *), void *data, rb_unblock_function_t *unblock_function, void *data2, int flags, struct rb_fiber_scheduler_blocking_operation_state *state)
Defer the execution of the passed function to the scheduler.
VALUE rb_fiber_scheduler_current(void)
Identical to rb_fiber_scheduler_get(), except it also returns RUBY_Qnil in case of a blocking fiber.
VALUE rb_fiber_scheduler_io_pread_memory(VALUE scheduler, VALUE io, rb_off_t from, void *base, size_t size, size_t length)
Non-blocking pread from the passed IO using a native buffer.
VALUE rb_fiber_scheduler_make_timeout(struct timeval *timeout)
Converts the passed timeout to an expression that rb_fiber_scheduler_block() etc.
VALUE rb_fiber_scheduler_io_wait_readable(VALUE scheduler, VALUE io)
Non-blocking wait until the passed IO is ready for reading.
VALUE rb_fiber_scheduler_io_read_memory(VALUE scheduler, VALUE io, void *base, size_t size, size_t length)
Non-blocking read from the passed IO using a native buffer.
VALUE rb_fiber_scheduler_io_pwrite(VALUE scheduler, VALUE io, rb_off_t from, VALUE buffer, size_t length, size_t offset)
Non-blocking write to the passed IO at the specified offset.
VALUE rb_fiber_scheduler_kernel_sleepv(VALUE scheduler, int argc, VALUE *argv)
Identical to rb_fiber_scheduler_kernel_sleep(), except it can pass multiple arguments.
VALUE rb_fiber_scheduler_fiber_interrupt(VALUE scheduler, VALUE fiber, VALUE exception)
Interrupt a fiber by raising an exception.
VALUE rb_fiber_scheduler_io_wait(VALUE scheduler, VALUE io, VALUE events, VALUE timeout)
Non-blocking version of rb_io_wait().
VALUE rb_fiber_scheduler_io_select(VALUE scheduler, VALUE readables, VALUE writables, VALUE exceptables, VALUE timeout)
Non-blocking version of IO.select.
VALUE rb_fiber_scheduler_io_read(VALUE scheduler, VALUE io, VALUE buffer, size_t length, size_t offset)
Non-blocking read from the passed IO.
int rb_fiber_scheduler_blocking_operation_cancel(rb_fiber_scheduler_blocking_operation_t *blocking_operation)
Cancel a blocking operation.
VALUE rb_fiber_scheduler_io_selectv(VALUE scheduler, int argc, VALUE *argv)
Non-blocking version of IO.select, argv variant.
VALUE rb_fiber_scheduler_process_wait(VALUE scheduler, rb_pid_t pid, int flags)
Non-blocking waitpid.
VALUE rb_fiber_scheduler_block(VALUE scheduler, VALUE blocker, VALUE timeout)
Non-blocking wait for the passed "blocker", which is for instance Thread.join or Mutex....
int rb_fiber_scheduler_blocking_operation_execute(rb_fiber_scheduler_blocking_operation_t *blocking_operation)
Execute blocking operation from handle (GVL not required).
VALUE rb_fiber_scheduler_io_pread(VALUE scheduler, VALUE io, rb_off_t from, VALUE buffer, size_t length, size_t offset)
Non-blocking read from the passed IO at the specified offset.
VALUE rb_fiber_scheduler_io_pwrite_memory(VALUE scheduler, VALUE io, rb_off_t from, const void *base, size_t size, size_t length)
Non-blocking pwrite to the passed IO using a native buffer.
VALUE rb_fiber_scheduler_io_write(VALUE scheduler, VALUE io, VALUE buffer, size_t length, size_t offset)
Non-blocking write to the passed IO.
VALUE rb_fiber_scheduler_close(VALUE scheduler)
Closes the passed scheduler object.
rb_fiber_scheduler_blocking_operation_t * rb_fiber_scheduler_blocking_operation_extract(VALUE self)
Extract the blocking operation handle from a BlockingOperationRuby object.
VALUE rb_fiber_scheduler_current_for_thread(VALUE thread)
Identical to rb_fiber_scheduler_current(), except it queries for that of the passed thread instead of...
VALUE rb_fiber_scheduler_kernel_sleep(VALUE scheduler, VALUE duration)
Non-blocking sleep.
VALUE rb_fiber_scheduler_address_resolve(VALUE scheduler, VALUE hostname)
Non-blocking DNS lookup.
VALUE rb_fiber_scheduler_set(VALUE scheduler)
Destructively assigns the passed scheduler to that of the current thread that is calling this functio...
VALUE rb_fiber_scheduler_io_write_memory(VALUE scheduler, VALUE io, const void *base, size_t size, size_t length)
Non-blocking write to the passed IO using a native buffer.
VALUE rb_fiber_scheduler_io_wait_writable(VALUE scheduler, VALUE io)
Non-blocking wait until the passed IO is ready for writing.
VALUE rb_fiber_scheduler_io_close(VALUE scheduler, VALUE io)
Non-blocking close the given IO.
VALUE rb_fiber_scheduler_get(void)
Queries the current scheduler of the current thread that is calling this function.
VALUE rb_fiber_scheduler_unblock(VALUE scheduler, VALUE blocker, VALUE fiber)
Wakes up a fiber previously blocked using rb_fiber_scheduler_block().
VALUE rb_fiber_scheduler_fiber(VALUE scheduler, int argc, VALUE *argv, int kw_splat)
Create and schedule a non-blocking fiber.
@ RUBY_Qundef
Represents so-called undef.
This is the struct that holds necessary info for a struct.
uintptr_t ID
Type that represents a Ruby identifier such as a variable name.
uintptr_t VALUE
Type that represents a Ruby object.