Class: SRT::Poll

Inherits:

Object

• Object
• SRT::Poll

Defined in:

ext/rbsrt/rbsrt.c

Instance Method Summary

• #add(*args) ⇒ Object
• #initialize ⇒ Object constructor
• #remove(socket_to_remove) ⇒ Object
• #update(*args) ⇒ Object
• #wait(*args) ⇒ Object

Constructor Details

#initialize ⇒ Object

# File 'ext/rbsrt/rbsrt.c', line 1956

VALUE rbsrt_poll_initialize(VALUE self)
{
    RBSRT_DEBUG_PRINT("initialize poll");

    RBSRT_POLL_UNWRAP(self, poll);

    poll->epollid = srt_epoll_create();

    VALUE objects_by_socket = rb_hash_new();

    rb_hash_clear(objects_by_socket);

    rb_ivar_set(self, rb_intern("@sockets"), objects_by_socket);
    
    return self;
}

Instance Method Details

#add(*args) ⇒ Object

# File 'ext/rbsrt/rbsrt.c', line 1993

VALUE rbsrt_poll_add_socket(int argc, VALUE* argv, VALUE self)
{
    RBSRT_DEBUG_PRINT("poll add socket");

    RBSRT_POLL_UNWRAP(self, poll);

    VALUE arg1;
    VALUE splat;

    rb_scan_args(argc, argv, "1*", &arg1, &splat);

    RBSRT_SOCKET_BASE_UNWRAP(arg1, socket);

    int events = rbsrt_epoll_event_with_splat(splat, rb_array_len(splat));

    if (srt_epoll_add_usock(poll->epollid, socket->socket, &events) == SRT_ERROR)
    {
        rbsrt_raise_last_srt_error();
    }

    VALUE sockets = rb_ivar_get(self, rb_intern("@sockets"));

    rb_hash_aset(sockets, INT2FIX(socket->socket), arg1);

    return self;
}

#remove(socket_to_remove) ⇒ Object

# File 'ext/rbsrt/rbsrt.c', line 1973

VALUE rbsrt_poll_remove_socket(VALUE self, VALUE socket_to_remove)
{
    RBSRT_DEBUG_PRINT("poll remove socket");

    RBSRT_POLL_UNWRAP(self, poll);

    RBSRT_SOCKET_BASE_UNWRAP(socket_to_remove, socket);

    VALUE sockets = rb_ivar_get(self, rb_intern("@sockets"));

    VALUE removed_socket = rb_hash_delete(sockets, INT2FIX(socket->socket));

    if (srt_epoll_remove_usock(poll->epollid, socket->socket) == SRT_ERROR)
    {
        rbsrt_raise_last_srt_error();
    }

    return removed_socket;
}

#update(*args) ⇒ Object

# File 'ext/rbsrt/rbsrt.c', line 2020

VALUE rbsrt_poll_update_socket(int argc, VALUE* argv, VALUE self)
{
    RBSRT_DEBUG_PRINT("poll update socket");

    RBSRT_POLL_UNWRAP(self, poll);

    VALUE arg1;
    VALUE splat;

    rb_scan_args(argc, argv, "1*", &arg1, &splat);

    int events = rbsrt_epoll_event_with_splat(splat, rb_array_len(splat));

    RBSRT_SOCKET_BASE_UNWRAP(arg1, socket);

    if (srt_epoll_update_usock(poll->epollid, socket->socket, &events) == SRT_ERROR)
    {
        rbsrt_raise_last_srt_error();
    }

    return Qnil;
}

#wait(*args) ⇒ Object

# File 'ext/rbsrt/rbsrt.c', line 2061

VALUE rbsrt_poll_wait(int argc, VALUE* argv, VALUE self)
{
    RBSRT_DEBUG_PRINT("poll wait");

    RBSRT_POLL_UNWRAP(self, poll);

    VALUE timeout;
    VALUE block;

    rb_scan_args(argc, argv, "01&", &timeout, &block);

    int epoll_timeout;

    if (!NIL_P(timeout))
    {
        Check_Type(timeout, T_FIXNUM);

        epoll_timeout = FIX2INT(timeout);
    }

    else
    {
        epoll_timeout = -1;
    }
    
    VALUE sockets = rb_ivar_get(self, rb_intern("@sockets"));
    int num_sockets = FIX2INT(rb_hash_size(sockets));

    SRT_EPOLL_EVENT events[num_sockets < 8 ? 8 : num_sockets + 8];

    VALUE readables = rb_ary_new();
    VALUE writables = rb_ary_new();
    VALUE errors = rb_ary_new();

    rbsrt_poll_wait_arg_t arg = {
        .epollid = poll->epollid,
        .timeout = epoll_timeout,
        .num_sockets = num_sockets,
        .num_events = 0,
        .events = events
    };

    rb_thread_call_without_gvl(rbsrt_poll_wait_without_gvl, &arg, RUBY_UBF_IO, NULL);

    RBSRT_DEBUG_PRINT("poll did wait");

    for (int i = 0; i < arg.num_events; i++)
    {
        SRT_EPOLL_EVENT *event = &events[i];

        VALUE sock = rb_hash_aref(sockets, INT2FIX(event->fd));

        if (!RTEST(sock))
        {
            RBSRT_DEBUG_PRINT("poll matched socket not in socket list");

            // TODO: remove unknown socket

            continue;
        }

        if (event->events & SRT_EPOLL_IN)
        {
            RBSRT_DEBUG_PRINT("poll add readables");

            rb_ary_push(readables, sock);
        }

        if (event->events & SRT_EPOLL_OUT)
        {
            RBSRT_DEBUG_PRINT("poll add writables");

            rb_ary_push(writables, sock);
        }

       if (event->events & SRT_EPOLL_ERR)
       {
           RBSRT_DEBUG_PRINT("poll add error");

           rb_ary_push(errors, sock);
       }
    }

    if (rb_block_given_p())
    {
        rb_yield_values(3, readables, writables, errors);

        return Qtrue;
    }

    return rb_ary_new_from_args(3, readables, writables, errors);
}