class Fiber::Scheduler

VALUE
rb_fiber_scheduler_address_resolve(VALUE scheduler, VALUE hostname)
{
    VALUE arguments[] = {
        hostname
    };

    return rb_check_funcall(scheduler, id_address_resolve, 1, arguments);
}

Вызывается любым методом, выполняющим DNS-поиск без обратного разрешения. Наиболее примечательный из таких методов — Addrinfo.getaddrinfo, но есть и многие другие.

Ожидается, что метод вернёт массив строк, соответствующих IP-адресам, в которые разрешается hostname, или nil, если разрешить его не удалось.

Достаточно полный список всех возможных мест вызова:

• Addrinfo.getaddrinfo

• Addrinfo.tcp

• Addrinfo.udp

• Addrinfo.ip

• Addrinfo.new

• Addrinfo.marshal_load

• SOCKSSocket.new

• TCPServer.new

• TCPSocket.new

• IPSocket.getaddress

• TCPSocket.gethostbyname

• UDPSocket#connect

• UDPSocket#bind

• UDPSocket#send

• Socket.getaddrinfo

• Socket.gethostbyname

• Socket.pack_sockaddr_in

• Socket.sockaddr_in

• Socket.unpack_sockaddr_in

VALUE
rb_fiber_scheduler_block(VALUE scheduler, VALUE blocker, VALUE timeout)
{
    return rb_funcall(scheduler, id_block, 2, blocker, timeout);
}

Вызывается такими методами, как Thread.join, а также Mutex, чтобы указать, что текущее Fiber заблокировано до дальнейшего уведомления (например, unblock) или до истечения timeout.

blocker — это то, чего мы ожидаем; значение носит исключительно информационный характер (для отладки и ведения журнала). Его значение не гарантируется.

Ожидается, что метод вернёт логическое значение, указывающее, успешно ли выполнена операция блокировки.

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)
{
    struct rb_blocking_operation_wait_arguments arguments = {
        .function = function,
        .data = data,
        .unblock_function = unblock_function,
        .data2 = data2,
        .flags = flags,
        .state = state
    };

    VALUE proc = rb_proc_new(rb_fiber_scheduler_blocking_operation_wait_proc, (VALUE)&arguments);

    return rb_check_funcall(scheduler, id_blocking_operation_wait, 1, &proc);
}

Вызывается основными методами Ruby для выполнения блокирующей операции неблокирующим способом.

Минимальная рекомендуемая реализация:

def blocking_operation_wait(work)
  Thread.new(&work).join
end

VALUE
rb_fiber_scheduler_close(VALUE scheduler)
{
    RUBY_ASSERT(ruby_thread_has_gvl_p());

    VALUE result;

    // The reason for calling `scheduler_close` before calling `close` is for
    // legacy schedulers which implement `close` and expect the user to call
    // it. Subsequently, that method would call `Fiber.set_scheduler(nil)`
    // which should call `scheduler_close`. If it were to call `close`, it
    // would create an infinite loop.

    result = rb_check_funcall(scheduler, id_scheduler_close, 0, NULL);
    if (!UNDEF_P(result)) return result;

    result = rb_check_funcall(scheduler, id_close, 0, NULL);
    if (!UNDEF_P(result)) return result;

    return Qnil;
}

Вызывается при завершении текущего потока. Ожидается, что планировщик реализует этот метод, чтобы все ожидающие волокна могли завершить выполнение.

Рекомендуется реализовать основной цикл обработки событий в методе close.

Реализация метода Fiber.schedule. Ожидается, что этот метод немедленно выполнит переданный блок кода в отдельном неблокирующем волокне и вернёт это Fiber.

Минимальная рекомендуемая реализация:

def fiber(&block)
  fiber = Fiber.new(blocking: false, &block)
  fiber.resume
  fiber
end

VALUE
rb_fiber_scheduler_io_pread(VALUE scheduler, VALUE io, rb_off_t from, VALUE buffer, size_t length, size_t offset)
{
    VALUE arguments[] = {
        io, buffer, OFFT2NUM(from), SIZET2NUM(length), SIZET2NUM(offset)
    };

    return rb_check_funcall(scheduler, id_io_pread, 5, arguments);
}

Вызывается методами IO#pread или IO::Buffer#pread для чтения length байт из io со смещения from в указанный buffer (см. IO::Buffer) по заданному offset.

Семантически этот метод совпадает с io_read, но позволяет задать смещение для чтения и часто лучше подходит для асинхронного IO из одного и того же файла.

Метод следует считать экспериментальным.

VALUE
rb_fiber_scheduler_io_pwrite(VALUE scheduler, VALUE io, rb_off_t from, VALUE buffer, size_t length, size_t offset)
{
    VALUE arguments[] = {
        io, buffer, OFFT2NUM(from), SIZET2NUM(length), SIZET2NUM(offset)
    };

    return rb_check_funcall(scheduler, id_io_pwrite, 5, arguments);
}

Вызывается методами IO#pwrite или IO::Buffer#pwrite для записи length байт в io со смещения from в указанный buffer (см. IO::Buffer) по заданному offset.

Семантически этот метод совпадает с io_write, но позволяет задать смещение для записи и часто лучше подходит для асинхронного IO в один и тот же файл.

Метод следует считать экспериментальным.

VALUE
rb_fiber_scheduler_io_read(VALUE scheduler, VALUE io, VALUE buffer, size_t length, size_t offset)
{
    VALUE arguments[] = {
        io, buffer, SIZET2NUM(length), SIZET2NUM(offset)
    };

    return rb_check_funcall(scheduler, id_io_read, 4, arguments);
}

Вызывается методами IO#read или IO#Buffer.read для чтения length байт из io в указанный buffer (см. IO::Buffer) по заданному offset.

Аргумент length означает «минимальное количество байт для чтения». Если размер буфера IO составляет 8 КиБ, а length равен 1024 (1 КиБ), может быть прочитано до 8 КиБ, но не менее 1 КиБ. Как правило, единственный случай, когда будет прочитано меньше данных, чем length, — это ошибка при чтении данных.

Допустимо указать length равным 0; это означает, что следует хотя бы один раз попытаться прочитать данные и вернуть все доступные данные.

Рекомендуемая реализация должна пытаться читать из io в неблокирующем режиме и вызывать io_wait, если io не готов (при этом управление будет передано другим волокнам).

Описание интерфейса для возврата данных см. в разделе IO::Buffer.

Ожидается, что метод вернёт количество прочитанных байт или, в случае ошибки, -errno (отрицательное значение, соответствующее коду системной ошибки).

Метод следует считать экспериментальным.

VALUE rb_fiber_scheduler_io_select(VALUE scheduler, VALUE readables, VALUE writables, VALUE exceptables, VALUE timeout)
{
    VALUE arguments[] = {
        readables, writables, exceptables, timeout
    };

    return rb_fiber_scheduler_io_selectv(scheduler, 4, arguments);
}

Вызывается методом IO.select, чтобы проверить, готовы ли указанные дескрипторы к заданным событиям в течение указанного timeout.

Ожидается, что метод вернёт кортеж из трёх Array объектов IO, готовых к работе.

VALUE
rb_fiber_scheduler_io_wait(VALUE scheduler, VALUE io, VALUE events, VALUE timeout)
{
    return rb_funcall(scheduler, id_io_wait, 3, io, events, timeout);
}

Вызывается методами IO#wait, IO#wait_readable, IO#wait_writable, чтобы проверить, готов ли указанный дескриптор к заданным событиям в течение указанного timeout.

events — это битовая маска из IO::READABLE, IO::WRITABLE и IO::PRIORITY.

Рекомендуемая реализация должна зарегистрировать, какое Fiber ожидает какие ресурсы, и немедленно вызвать Fiber.yield, чтобы передать управление другим волокнам. Затем в методе close планировщик может распределить все ресурсы ввода-вывода между ожидающими их волокнами.

Ожидается, что метод вернёт подмножество событий, готовых немедленно.

VALUE
rb_fiber_scheduler_io_write(VALUE scheduler, VALUE io, VALUE buffer, size_t length, size_t offset)
{
    VALUE arguments[] = {
        io, buffer, SIZET2NUM(length), SIZET2NUM(offset)
    };

    return rb_check_funcall(scheduler, id_io_write, 4, arguments);
}

Вызывается методами IO#write или IO::Buffer#write для записи length байт в io из указанного buffer (см. IO::Buffer) по заданному offset.

Аргумент length означает «минимальное количество байт для записи». Если размер буфера IO составляет 8 КиБ, а заданное значение length равно 1024 (1 КиБ), будет записано не более 8 КиБ, но не менее 1 КиБ. Как правило, единственный случай, когда будет записано меньше данных, чем length, — это ошибка при записи данных.

Допустимо указать length равным 0; это означает, что следует хотя бы один раз попытаться записать как можно больше данных.

Рекомендуемая реализация должна пытаться записывать в io в неблокирующем режиме и вызывать io_wait, если io не готов (при этом управление будет передано другим волокнам).

Описание интерфейса для эффективного получения данных из буфера см. в разделе IO::Buffer.

Ожидается, что метод вернёт количество записанных байт или, в случае ошибки, -errno (отрицательное значение, соответствующее коду системной ошибки).

Метод следует считать экспериментальным.

VALUE
rb_fiber_scheduler_kernel_sleep(VALUE scheduler, VALUE timeout)
{
    return rb_funcall(scheduler, id_kernel_sleep, 1, timeout);
}

Вызывается методами Kernel#sleep и Mutex#sleep. Ожидается, что метод обеспечит неблокирующую реализацию ожидания. Реализация может зарегистрировать текущее волокно в списке «какое волокно и до какого момента ожидает», вызвать Fiber.yield, чтобы передать управление, а затем в методе close возобновить волокна, время ожидания которых истекло.

VALUE
rb_fiber_scheduler_process_wait(VALUE scheduler, rb_pid_t pid, int flags)
{
    VALUE arguments[] = {
        PIDT2NUM(pid), RB_INT2NUM(flags)
    };

    return rb_check_funcall(scheduler, id_process_wait, 2, arguments);
}

Вызывается методом Process::Status.wait для ожидания указанного процесса. Описание аргументов см. в описании этого метода.

Рекомендуемая минимальная реализация:

Thread.new do
  Process::Status.wait(pid, flags)
end.value

Этот перехватчик необязателен: если в текущем планировщике он отсутствует, Process::Status.wait будет работать как блокирующий метод.

Ожидается, что метод вернёт экземпляр Process::Status.

VALUE
rb_fiber_scheduler_timeout_after(VALUE scheduler, VALUE timeout, VALUE exception, VALUE message)
{
    VALUE arguments[] = {
        timeout, exception, message
    };

    return rb_check_funcall(scheduler, id_timeout_after, 3, arguments);
}

Вызывается методом Timeout.timeout для выполнения переданного block в течение заданного duration. Его также может напрямую вызвать планировщик или пользовательский код.

По возможности пытается ограничить время выполнения заданного block указанным duration. Если неблокирующая операция приводит к тому, что время выполнения block превышает заданное duration, эта неблокирующая операция должна быть прервана путём возбуждения указанного exception_class, созданного с переданными exception_arguments.

Общие тайм-ауты выполнения часто считают рискованными. Эта реализация прерывает только неблокирующие операции. Это сделано намеренно: неблокирующие операции могут завершаться сбоем по разным непредсказуемым причинам, поэтому приложения уже должны быть устойчивы к таким ситуациям и, следовательно, к тайм-аутам.

Однако вследствие такого подхода, если block не вызывает неблокирующих операций, прервать его будет невозможно. Если требуется обеспечить предсказуемые точки срабатывания тайм-аутов, добавьте +sleep(0)+.

Если блок выполнится успешно, будет возвращён его результат.

Исключение обычно возбуждается с помощью Fiber#raise.

VALUE
rb_fiber_scheduler_unblock(VALUE scheduler, VALUE blocker, VALUE fiber)
{
    RUBY_ASSERT(rb_obj_is_fiber(fiber));

    VALUE result;
    enum ruby_tag_type state;

    // `rb_fiber_scheduler_unblock` can be called from points where `errno` is expected to be preserved. Therefore, we should save and restore it. For example `io_binwrite` calls `rb_fiber_scheduler_unblock` and if `errno` is reset to 0 by user code, it will break the error handling in `io_write`.
    //
    // If we explicitly preserve `errno` in `io_binwrite` and other similar functions (e.g. by returning it), this code is no longer needed. I hope in the future we will be able to remove it.
    int saved_errno = errno;

    // We must prevent interrupts while invoking the unblock method, because otherwise fibers can be left permanently blocked if an interrupt occurs during the execution of user code.
    rb_execution_context_t *ec = GET_EC();
    int saved_interrupt_mask = ec->interrupt_mask;
    ec->interrupt_mask |= PENDING_INTERRUPT_MASK;

    EC_PUSH_TAG(ec);
    if ((state = EC_EXEC_TAG()) == TAG_NONE) {
        result = rb_funcall(scheduler, id_unblock, 2, blocker, fiber);
    }
    EC_POP_TAG();

    ec->interrupt_mask = saved_interrupt_mask;

    if (state) {
        EC_JUMP_TAG(ec, state);
    }

    RUBY_VM_CHECK_INTS(ec);

    errno = saved_errno;

    return result;
}

Вызывается для пробуждения Fiber, ранее заблокированного с помощью block (например, Mutex#lock вызывает block, а Mutex#unlock вызывает unblock). Планировщик должен использовать параметр fiber, чтобы определить, какое волокно разблокируется.

blocker — это то, чего ожидали, но значение носит исключительно информационный характер (для отладки и ведения журнала), и не гарантируется, что оно совпадает со значением blocker для block.