класс 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_close(VALUE scheduler)
{
    VM_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::Buffer#pread. См. этот метод для описания аргументов.

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::Buffer#pwrite. См. этот метод для описания аргументов.

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 для чтения length байтов из io в указанный buffer (см. IO::Buffer).

Аргумент 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.

Ожидается, что будет возвращен 3-кортеж из Array готовых I/O.

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, планировщик может отправить все ресурсы I/O волокнам, ожидающим их.

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

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 для записи length байтов в io из указанного buffer (см. IO::Buffer).

Аргумент 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)
{
    VM_ASSERT(rb_obj_is_fiber(fiber));

    return rb_funcall(scheduler, id_unblock, 2, blocker, fiber);
}

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

blocker - это то, что ожидалось, но это только информационная информация (для отладки и ведения журнала), и не гарантируется, что она будет таким же значением, как и blocker для метода block.