Class: Event::Backend::KQueue

Inherits:

Object

• Object
• Event::Backend::KQueue

Defined in:

ext/event/backend/kqueue.c

Instance Method Summary

• #close ⇒ Object
• #initialize(loop) ⇒ Object constructor
• #io_read(fiber, io, buffer, _offset, _length) ⇒ Object
• #io_wait(fiber, io, events) ⇒ Object
• #io_write(fiber, io, buffer, _offset, _length) ⇒ Object
• #process_wait(fiber, pid, flags) ⇒ Object
• #select(duration) ⇒ Object

Constructor Details

#initialize(loop) ⇒ Object

# File 'ext/event/backend/kqueue.c', line 88

VALUE Event_Backend_KQueue_initialize(VALUE self, VALUE loop) {
  struct Event_Backend_KQueue *data = NULL;
  TypedData_Get_Struct(self, struct Event_Backend_KQueue, &Event_Backend_KQueue_Type, data);
  
  data->loop = loop;
  int result = kqueue();
  
  if (result == -1) {
    rb_sys_fail("kqueue");
  } else {
    ioctl(result, FIOCLEX);
    data->descriptor = result;
    
    rb_update_max_fd(data->descriptor);
  }
  
  return self;
}

Instance Method Details

#close ⇒ Object

# File 'ext/event/backend/kqueue.c', line 107

VALUE Event_Backend_KQueue_close(VALUE self) {
  struct Event_Backend_KQueue *data = NULL;
  TypedData_Get_Struct(self, struct Event_Backend_KQueue, &Event_Backend_KQueue_Type, data);
  
  close_internal(data);
  
  return Qnil;
}

#io_read(fiber, io, buffer, _offset, _length) ⇒ Object

# File 'ext/event/backend/kqueue.c', line 355

VALUE Event_Backend_KQueue_io_read(VALUE self, VALUE fiber, VALUE io, VALUE buffer, VALUE _offset, VALUE _length) {
  struct Event_Backend_KQueue *data = NULL;
  TypedData_Get_Struct(self, struct Event_Backend_KQueue, &Event_Backend_KQueue_Type, data);
  
  int descriptor = RB_NUM2INT(rb_funcall(io, id_fileno, 0));
  
  size_t offset = NUM2SIZET(_offset);
  size_t length = NUM2SIZET(_length);
  
  struct io_read_arguments io_read_arguments = {
    .self = self,
    .fiber = fiber,
    .io = io,
    
    .flags = Event_Backend_nonblock_set(descriptor),
    .descriptor = descriptor,
    .buffer = buffer,
    .offset = offset,
    .length = length,
  };
  
  return rb_ensure(io_read_loop, (VALUE)&io_read_arguments, io_read_ensure, (VALUE)&io_read_arguments);
}

#io_wait(fiber, io, events) ⇒ Object

# File 'ext/event/backend/kqueue.c', line 289

VALUE Event_Backend_KQueue_io_wait(VALUE self, VALUE fiber, VALUE io, VALUE events) {
  struct Event_Backend_KQueue *data = NULL;
  TypedData_Get_Struct(self, struct Event_Backend_KQueue, &Event_Backend_KQueue_Type, data);
  
  int descriptor = RB_NUM2INT(rb_funcall(io, id_fileno, 0));
  
  struct io_wait_arguments io_wait_arguments = {
    .events = io_add_filters(data->descriptor, descriptor, RB_NUM2INT(events), fiber),
    .data = data,
    .descriptor = descriptor,
  };
  
  return rb_rescue(io_wait_transfer, (VALUE)&io_wait_arguments, io_wait_rescue, (VALUE)&io_wait_arguments);
}

#io_write(fiber, io, buffer, _offset, _length) ⇒ Object

# File 'ext/event/backend/kqueue.c', line 428

VALUE Event_Backend_KQueue_io_write(VALUE self, VALUE fiber, VALUE io, VALUE buffer, VALUE _offset, VALUE _length) {
  struct Event_Backend_KQueue *data = NULL;
  TypedData_Get_Struct(self, struct Event_Backend_KQueue, &Event_Backend_KQueue_Type, data);
  
  int descriptor = NUM2INT(rb_funcall(io, id_fileno, 0));
  
  size_t offset = NUM2SIZET(_offset);
  size_t length = NUM2SIZET(_length);
  
  struct io_write_arguments io_write_arguments = {
    .self = self,
    .fiber = fiber,
    .io = io,
    
    .flags = Event_Backend_nonblock_set(descriptor),
    .descriptor = descriptor,
    .buffer = buffer,
    .offset = offset,
    .length = length,
  };
  
  return rb_ensure(io_write_loop, (VALUE)&io_write_arguments, io_write_ensure, (VALUE)&io_write_arguments);
}

#process_wait(fiber, pid, flags) ⇒ Object

# File 'ext/event/backend/kqueue.c', line 177

VALUE Event_Backend_KQueue_process_wait(VALUE self, VALUE fiber, VALUE pid, VALUE flags) {
  struct Event_Backend_KQueue *data = NULL;
  TypedData_Get_Struct(self, struct Event_Backend_KQueue, &Event_Backend_KQueue_Type, data);
  
  struct process_wait_arguments process_wait_arguments = {
    .data = data,
    .pid = NUM2PIDT(pid),
    .flags = RB_NUM2INT(flags),
  };
  
  int waiting = process_add_filters(data->descriptor, process_wait_arguments.pid, fiber);
  
  if (waiting) {
    return rb_rescue(process_wait_transfer, (VALUE)&process_wait_arguments, process_wait_rescue, (VALUE)&process_wait_arguments);
  } else {
    return Event_Backend_process_status_wait(process_wait_arguments.pid);
  }
}

#select(duration) ⇒ Object

# File 'ext/event/backend/kqueue.c', line 520

VALUE Event_Backend_KQueue_select(VALUE self, VALUE duration) {
  struct Event_Backend_KQueue *data = NULL;
  TypedData_Get_Struct(self, struct Event_Backend_KQueue, &Event_Backend_KQueue_Type, data);
  
  struct select_arguments arguments = {
    .data = data,
    .count = KQUEUE_MAX_EVENTS,
    .storage = {
      .tv_sec = 0,
      .tv_nsec = 0
    }
  };
  
  // We break this implementation into two parts.
  // (1) count = kevent(..., timeout = 0)
  // (2) without gvl: kevent(..., timeout = 0) if count == 0 and timeout != 0
  // This allows us to avoid releasing and reacquiring the GVL.
  // Non-comprehensive testing shows this gives a 1.5x speedup.
  arguments.timeout = &arguments.storage;
  
  // First do the syscall with no timeout to get any immediately available events:
  select_internal_with_gvl(&arguments);
  
  // If there were no pending events, if we have a timeout, wait for more events:
  if (arguments.count == 0) {
    arguments.timeout = make_timeout(duration, &arguments.storage);
    
    if (!timeout_nonblocking(arguments.timeout)) {
      arguments.count = KQUEUE_MAX_EVENTS;
      
      select_internal_without_gvl(&arguments);
    }
  }
  
  for (int i = 0; i < arguments.count; i += 1) {
    VALUE fiber = (VALUE)arguments.events[i].udata;
    VALUE result = INT2NUM(arguments.events[i].filter);
    
    Event_Backend_transfer_result(fiber, result);
  }
  
  return INT2NUM(arguments.count);
}