Class: IO::Event::Selector::KQueue

Inherits:

Object

• Object
• IO::Event::Selector::KQueue

Defined in:

ext/io/event/selector/kqueue.c

Instance Method Summary

• #close ⇒ Object
• #closed? ⇒ Boolean
• #idle_duration ⇒ Object
• #initialize(loop) ⇒ Object constructor
• #io_read(*args) ⇒ Object
• #io_wait(fiber, io, events) ⇒ Object
• #io_write(*args) ⇒ Object
• #loop ⇒ Object
• #process_wait(fiber, _pid, _flags) ⇒ Object
• #push(fiber) ⇒ Object
• #raise(*args) ⇒ Object
• #ready? ⇒ Boolean
• #resume(*args) ⇒ Object
• #select(duration) ⇒ Object
• #transfer ⇒ Object
• #wakeup ⇒ Object
• #yield ⇒ Object

Constructor Details

#initialize(loop) ⇒ Object

# File 'ext/io/event/selector/kqueue.c', line 323

VALUE IO_Event_Selector_KQueue_initialize(VALUE self, VALUE loop) {
	struct IO_Event_Selector_KQueue *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_KQueue, &IO_Event_Selector_KQueue_Type, selector);
	
	IO_Event_Selector_initialize(&selector->backend, self, loop);
	int result = kqueue();
	
	if (result == -1) {
		rb_sys_fail("IO_Event_Selector_KQueue_initialize:kqueue");
	} else {
		// Make sure the descriptor is closed on exec.
		ioctl(result, FIOCLEX);
		
		selector->descriptor = result;
		selector->owner = getpid();
		
		rb_update_max_fd(selector->descriptor);
	}
	
#ifdef IO_EVENT_SELECTOR_KQUEUE_USE_INTERRUPT
	IO_Event_Interrupt_open(&selector->interrupt);
	IO_Event_Interrupt_add(&selector->interrupt, selector);
#endif
	
	return self;
}

Instance Method Details

#close ⇒ Object

# File 'ext/io/event/selector/kqueue.c', line 366

VALUE IO_Event_Selector_KQueue_close(VALUE self) {
	struct IO_Event_Selector_KQueue *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_KQueue, &IO_Event_Selector_KQueue_Type, selector);
	
	close_internal(selector);
	
	return Qnil;
}

#closed? ⇒ Boolean

Returns:

• (Boolean)

# File 'ext/io/event/selector/kqueue.c', line 375

VALUE IO_Event_Selector_KQueue_closed_p(VALUE self) {
	struct IO_Event_Selector_KQueue *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_KQueue, &IO_Event_Selector_KQueue_Type, selector);
	
	return selector->descriptor < 0 || selector->owner != getpid() ? Qtrue : Qfalse;
}

#idle_duration ⇒ Object

# File 'ext/io/event/selector/kqueue.c', line 357

VALUE IO_Event_Selector_KQueue_idle_duration(VALUE self) {
	struct IO_Event_Selector_KQueue *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_KQueue, &IO_Event_Selector_KQueue_Type, selector);
	
	double duration = selector->idle_duration.tv_sec + (selector->idle_duration.tv_nsec / 1000000000.0);
	
	return DBL2NUM(duration);
}

#io_read(*args) ⇒ Object

# File 'ext/io/event/selector/kqueue.c', line 654

VALUE IO_Event_Selector_KQueue_io_read(VALUE self, VALUE fiber, VALUE io, VALUE buffer, VALUE _first, VALUE _second) {
	struct IO_Event_Selector_KQueue *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_KQueue, &IO_Event_Selector_KQueue_Type, selector);
	
	void *base;
	size_t size;
	rb_io_buffer_get_bytes_for_writing(buffer, &base, &size);
	
#if RUBY_FIBER_SCHEDULER_VERSION >= 4
	size_t offset = NUM2SIZET(_first);
	size_t length = NUM2SIZET(_second);
	
	if (!IO_Event_Selector_valid_buffer_range(size, offset, length)) {
		return rb_fiber_scheduler_io_result(-1, EINVAL);
	} else if (length == 0) {
		return rb_fiber_scheduler_io_result(0, 0);
	}
	
	base = (char*)base + offset;
	size = length;
#else
	size_t length = NUM2SIZET(_first);
	size_t offset = NUM2SIZET(_second);
	
	if (offset > size) {
		return rb_fiber_scheduler_io_result(-1, EINVAL);
	} else if (offset == size) {
		return rb_fiber_scheduler_io_result(0, 0);
	}
	
	base = (char*)base + offset;
	size -= offset;
#endif
	
	int descriptor = IO_Event_Selector_io_descriptor(io);
	
	struct io_read_arguments io_read_arguments = {
#if RUBY_FIBER_SCHEDULER_VERSION < 4
		.self = self,
		.fiber = fiber,
		.io = io,
#endif
		
		.flags = IO_Event_Selector_nonblock_set(descriptor),
		.descriptor = descriptor,
		.base = base,
		.size = size,
#if RUBY_FIBER_SCHEDULER_VERSION < 4
		.length = length,
#endif
	};
	
	RB_OBJ_WRITTEN(self, Qundef, fiber);
	
	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/io/event/selector/kqueue.c', line 553

VALUE IO_Event_Selector_KQueue_io_wait(VALUE self, VALUE fiber, VALUE io, VALUE events) {
	struct IO_Event_Selector_KQueue *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_KQueue, &IO_Event_Selector_KQueue_Type, selector);
	
	int descriptor = IO_Event_Selector_io_descriptor(io);
	
	struct IO_Event_Selector_KQueue_Waiting waiting = {
		.list = {.type = &IO_Event_Selector_KQueue_io_wait_list_type},
		.fiber = fiber,
		.events = RB_NUM2INT(events),
	};
	
	RB_OBJ_WRITTEN(self, Qundef, fiber);
	
	int result = IO_Event_Selector_KQueue_Waiting_register(selector, descriptor, &waiting);
	if (result == -1) {
		rb_sys_fail("IO_Event_Selector_KQueue_io_wait:IO_Event_Selector_KQueue_Waiting_register");
	}
	
	struct io_wait_arguments io_wait_arguments = {
		.selector = selector,
		.waiting = &waiting,
	};
	
	if (DEBUG_IO_WAIT) fprintf(stderr, "IO_Event_Selector_KQueue_io_wait descriptor=%d\n", descriptor);
	
	return rb_ensure(io_wait_transfer, (VALUE)&io_wait_arguments, io_wait_ensure, (VALUE)&io_wait_arguments);
}

#io_write(*args) ⇒ Object

# File 'ext/io/event/selector/kqueue.c', line 796

VALUE IO_Event_Selector_KQueue_io_write(VALUE self, VALUE fiber, VALUE io, VALUE buffer, VALUE _first, VALUE _second) {
	struct IO_Event_Selector_KQueue *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_KQueue, &IO_Event_Selector_KQueue_Type, selector);
	
	const void *base;
	size_t size;
	rb_io_buffer_get_bytes_for_reading(buffer, &base, &size);
	
#if RUBY_FIBER_SCHEDULER_VERSION >= 4
	size_t offset = NUM2SIZET(_first);
	size_t length = NUM2SIZET(_second);
	
	if (!IO_Event_Selector_valid_buffer_range(size, offset, length)) {
		return rb_fiber_scheduler_io_result(-1, EINVAL);
	} else if (length == 0) {
		return rb_fiber_scheduler_io_result(0, 0);
	}
	
	base = (const char*)base + offset;
	size = length;
#else
	size_t length = NUM2SIZET(_first);
	size_t offset = NUM2SIZET(_second);
	
	if (length > size) {
		rb_raise(rb_eRuntimeError, "Length exceeds size of buffer!");
	}
	
	if (offset > size) {
		return rb_fiber_scheduler_io_result(-1, EINVAL);
	} else if (offset == size) {
		return rb_fiber_scheduler_io_result(0, 0);
	}
	
	base = (const char*)base + offset;
	size -= offset;
#endif
	
	int descriptor = IO_Event_Selector_io_descriptor(io);
	
	struct io_write_arguments io_write_arguments = {
#if RUBY_FIBER_SCHEDULER_VERSION < 4
		.self = self,
		.fiber = fiber,
		.io = io,
#endif
		
		.flags = IO_Event_Selector_nonblock_set(descriptor),
		.descriptor = descriptor,
		.base = base,
		.size = size,
#if RUBY_FIBER_SCHEDULER_VERSION < 4
		.length = length,
#endif
	};
	
	RB_OBJ_WRITTEN(self, Qundef, fiber);
	
	return rb_ensure(io_write_loop, (VALUE)&io_write_arguments, io_write_ensure, (VALUE)&io_write_arguments);
}

#loop ⇒ Object

# File 'ext/io/event/selector/kqueue.c', line 350

VALUE IO_Event_Selector_KQueue_loop(VALUE self) {
	struct IO_Event_Selector_KQueue *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_KQueue, &IO_Event_Selector_KQueue_Type, selector);
	
	return selector->backend.loop;
}

#process_wait(fiber, _pid, _flags) ⇒ Object

# File 'ext/io/event/selector/kqueue.c', line 482

VALUE IO_Event_Selector_KQueue_process_wait(VALUE self, VALUE fiber, VALUE _pid, VALUE _flags) {
	struct IO_Event_Selector_KQueue *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_KQueue, &IO_Event_Selector_KQueue_Type, selector);
	
	pid_t pid = NUM2PIDT(_pid);
	int flags = NUM2INT(_flags);
	
	// `EVFILT_PROC` can only refer to a specific process, so waiting for any child or a process group (pid <= 0) is delegated to the threaded fallback:
	if (pid <= 0) {
		return IO_Event_Selector_process_wait(pid, flags);
	}
	
	struct IO_Event_Selector_KQueue_Waiting waiting = {
		.list = {.type = &IO_Event_Selector_KQueue_process_wait_list_type},
		.fiber = fiber,
		.events = IO_EVENT_EXIT,
	};
	
	RB_OBJ_WRITTEN(self, Qundef, fiber);
	
	struct process_wait_arguments process_wait_arguments = {
		.selector = selector,
		.waiting = &waiting,
		.pid = pid,
		.flags = flags,
	};
	
	int result = IO_Event_Selector_KQueue_Waiting_register(selector, pid, &waiting);
	if (result == -1) {
		// OpenBSD/NetBSD return ESRCH when attempting to register an EVFILT_PROC event for a zombie process.
		if (errno == ESRCH) {
			process_prewait(pid);
			
			return IO_Event_Selector_process_status_reap(pid, flags);
		}
		
		rb_sys_fail("IO_Event_Selector_KQueue_process_wait:IO_Event_Selector_KQueue_Waiting_register");
	}
	
	return rb_ensure(process_wait_transfer, (VALUE)&process_wait_arguments, process_wait_ensure, (VALUE)&process_wait_arguments);
}

#push(fiber) ⇒ Object

# File 'ext/io/event/selector/kqueue.c', line 406

VALUE IO_Event_Selector_KQueue_push(VALUE self, VALUE fiber)
{
	struct IO_Event_Selector_KQueue *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_KQueue, &IO_Event_Selector_KQueue_Type, selector);
	
	IO_Event_Selector_ready_push(&selector->backend, fiber);
	
	return Qnil;
}

#raise(*args) ⇒ Object

# File 'ext/io/event/selector/kqueue.c', line 416

VALUE IO_Event_Selector_KQueue_raise(int argc, VALUE *argv, VALUE self)
{
	struct IO_Event_Selector_KQueue *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_KQueue, &IO_Event_Selector_KQueue_Type, selector);
	
	return IO_Event_Selector_raise(&selector->backend, argc, argv);
}

#ready? ⇒ Boolean

Returns:

• (Boolean)

# File 'ext/io/event/selector/kqueue.c', line 424

VALUE IO_Event_Selector_KQueue_ready_p(VALUE self) {
	struct IO_Event_Selector_KQueue *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_KQueue, &IO_Event_Selector_KQueue_Type, selector);
	
	return selector->backend.ready ? Qtrue : Qfalse;
}

#resume(*args) ⇒ Object

# File 'ext/io/event/selector/kqueue.c', line 390

VALUE IO_Event_Selector_KQueue_resume(int argc, VALUE *argv, VALUE self)
{
	struct IO_Event_Selector_KQueue *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_KQueue, &IO_Event_Selector_KQueue_Type, selector);
	
	return IO_Event_Selector_resume(&selector->backend, argc, argv);
}

#select(duration) ⇒ Object

# File 'ext/io/event/selector/kqueue.c', line 1045

VALUE IO_Event_Selector_KQueue_select(VALUE self, VALUE duration) {
	struct IO_Event_Selector_KQueue *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_KQueue, &IO_Event_Selector_KQueue_Type, selector);
	
	selector->idle_duration.tv_sec = 0;
	selector->idle_duration.tv_nsec = 0;
	
	int ready = IO_Event_Selector_ready_flush(&selector->backend);
	
	struct select_arguments arguments = {
		.selector = selector,
		.count = KQUEUE_MAX_EVENTS,
		.result = 0,
		.storage = {
			.tv_sec = 0,
			.tv_nsec = 0
		},
		.saved = {},
	};
	
	arguments.timeout = &arguments.storage;
	
	// We break this implementation into two parts.
	// (1) result = kevent(..., timeout = 0)
	// (2) without gvl: kevent(..., timeout = 0) if result == 0 and timeout != 0
	// This allows us to avoid releasing and reacquiring the GVL.
	// Non-comprehensive testing shows this gives a 1.5x speedup.
	
	// First do the syscall with no timeout to get any immediately available events:
	if (DEBUG) fprintf(stderr, "\r\nselect_internal_with_gvl timeout=" IO_EVENT_TIME_PRINTF_TIMESPEC "\r\n", IO_EVENT_TIME_PRINTF_TIMESPEC_ARGUMENTS(arguments.storage));
	int result = select_internal_with_gvl(&arguments);
	if (DEBUG) fprintf(stderr, "\r\nselect_internal_with_gvl done\r\n");
	
	// If we:
	// 1. Didn't process any ready fibers, and
	// 2. Didn't process any events from non-blocking select (above), and
	// 3. There are no items in the ready list,
	// then we can perform a blocking select.
	if (!ready && !result && !selector->backend.ready) {
		arguments.timeout = make_timeout(duration, &arguments.storage);
		
		if (select_blocking_allowed(arguments.timeout)) {
			struct timespec start_time;
			IO_Event_Time_current(&start_time);
			
			if (DEBUG) fprintf(stderr, "IO_Event_Selector_KQueue_select timeout=" IO_EVENT_TIME_PRINTF_TIMESPEC "\n", IO_EVENT_TIME_PRINTF_TIMESPEC_ARGUMENTS(arguments.storage));
			result = select_internal_without_gvl(&arguments);
			
			struct timespec end_time;
			IO_Event_Time_current(&end_time);
			IO_Event_Time_elapsed(&start_time, &end_time, &selector->idle_duration);
		}
	}
	
	if (result) {
		return rb_ensure(select_handle_events, (VALUE)&arguments, select_handle_events_ensure, (VALUE)&arguments);
	} else {
		return RB_INT2NUM(0);
	}
}

#transfer ⇒ Object

# File 'ext/io/event/selector/kqueue.c', line 382

VALUE IO_Event_Selector_KQueue_transfer(VALUE self)
{
	struct IO_Event_Selector_KQueue *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_KQueue, &IO_Event_Selector_KQueue_Type, selector);
	
	return IO_Event_Selector_loop_yield(&selector->backend);
}

#wakeup ⇒ Object

# File 'ext/io/event/selector/kqueue.c', line 1106

VALUE IO_Event_Selector_KQueue_wakeup(VALUE self) {
	struct IO_Event_Selector_KQueue *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_KQueue, &IO_Event_Selector_KQueue_Type, selector);
	
	if (selector->backend.blocked) {
#ifdef IO_EVENT_SELECTOR_KQUEUE_USE_INTERRUPT
		IO_Event_Interrupt_signal(&selector->interrupt);
#else
		struct kevent trigger = {0};
		
		trigger.filter = EVFILT_USER;
		trigger.flags = EV_ADD | EV_CLEAR;
		
		int result = kevent(selector->descriptor, &trigger, 1, NULL, 0, NULL);
		
		if (result == -1) {
			rb_sys_fail("IO_Event_Selector_KQueue_wakeup:kevent");
		}
		
		// FreeBSD apparently only works if the NOTE_TRIGGER is done as a separate call.
		trigger.flags = 0;
		trigger.fflags = NOTE_TRIGGER;
		
		result = kevent(selector->descriptor, &trigger, 1, NULL, 0, NULL);
		
		if (result == -1) {
			rb_sys_fail("IO_Event_Selector_KQueue_wakeup:kevent");
		}
#endif
		
		return Qtrue;
	}
	
	return Qfalse;
}

#yield ⇒ Object

# File 'ext/io/event/selector/kqueue.c', line 398

VALUE IO_Event_Selector_KQueue_yield(VALUE self)
{
	struct IO_Event_Selector_KQueue *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_KQueue, &IO_Event_Selector_KQueue_Type, selector);
	
	return IO_Event_Selector_yield(&selector->backend);
}