Class: IO::Event::Selector::URing

Inherits:

Object

• Object
• IO::Event::Selector::URing

Defined in:

ext/io/event/selector/uring.c

Instance Method Summary

• #close ⇒ Object
• #closed? ⇒ Boolean
• #idle_duration ⇒ Object
• #initialize(loop) ⇒ Object constructor
• #io_close(_descriptor) ⇒ Object
• #io_pread(fiber, io, buffer, _from, _length, _offset) ⇒ Object
• #io_pwrite(fiber, io, buffer, _from, _length, _offset) ⇒ Object
• #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/uring.c', line 265

VALUE IO_Event_Selector_URing_initialize(VALUE self, VALUE loop) {
	struct IO_Event_Selector_URing *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
	
	IO_Event_Selector_initialize(&selector->backend, self, loop);
	
	unsigned int flags = 0;
	// IORING_SETUP_SINGLE_ISSUER (kernel 6.0+): only the owner thread submits SQEs.
	// Safe here because wakeup() uses eventfd (no ring access from other threads).
#ifdef IORING_SETUP_SINGLE_ISSUER
	flags |= IORING_SETUP_SINGLE_ISSUER;
#endif
	// IORING_SETUP_DEFER_TASKRUN (kernel 6.1+, requires SINGLE_ISSUER): defer io_uring
	// task work to the application thread rather than a kernel thread, reducing
	// cross-CPU signaling overhead.
#ifdef IORING_SETUP_DEFER_TASKRUN
	flags |= IORING_SETUP_DEFER_TASKRUN;
#endif
	// IORING_SETUP_TASKRUN_FLAG (kernel 5.19+, always available alongside
	// DEFER_TASKRUN): the kernel surfaces IORING_SQ_TASKRUN in sq.flags whenever
	// task work is pending, so select() can skip the io_uring_get_events()
	// syscall when there is nothing deferred to flush.
#ifdef IORING_SETUP_TASKRUN_FLAG
	flags |= IORING_SETUP_TASKRUN_FLAG;
#endif
	// IORING_SETUP_SUBMIT_ALL (kernel 5.18+): keep processing the rest of the SQE
	// batch even when one fails, reducing the frequency of short submits.
#ifdef IORING_SETUP_SUBMIT_ALL
	flags |= IORING_SETUP_SUBMIT_ALL;
#endif
	
	int result = io_uring_queue_init(URING_ENTRIES, &selector->ring, flags);
	
#ifdef IORING_SETUP_SUBMIT_ALL
	if (result == -EINVAL) {
		// IORING_SETUP_SUBMIT_ALL was added in Linux 5.18; retry without it.
		if (DEBUG) fprintf(stderr, "IO_Event_Selector_URing_initialize: no IORING_SETUP_SUBMIT_ALL\n");
		flags &= ~IORING_SETUP_SUBMIT_ALL;
		result = io_uring_queue_init(URING_ENTRIES, &selector->ring, flags);
	}
#endif
	
	if (result < 0) {
		rb_syserr_fail(-result, "IO_Event_Selector_URing_initialize:io_uring_queue_init");
	}
	
	selector->owner = getpid();
	
	rb_update_max_fd(selector->ring.ring_fd);
	
	// Interrupt for cross-thread wakeup: another thread calls signal(); the owner
	// thread submits an async read before each blocking wait so the ring wakes up
	// without the waking thread ever touching the SQ.
	IO_Event_Interrupt_open(&selector->interrupt);
	if (selector->interrupt.descriptor < 0) {
		io_uring_queue_exit(&selector->ring);
		selector->ring.ring_fd = -1;
		rb_sys_fail("IO_Event_Selector_URing_initialize:IO_Event_Interrupt_open");
	}
	
	return self;
}

Instance Method Details

#close ⇒ Object

# File 'ext/io/event/selector/uring.c', line 344

VALUE IO_Event_Selector_URing_close(VALUE self) {
	struct IO_Event_Selector_URing *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
	
	close_internal(selector);
	
	return Qnil;
}

#closed? ⇒ Boolean

Returns:

• (Boolean)

# File 'ext/io/event/selector/uring.c', line 353

VALUE IO_Event_Selector_URing_closed_p(VALUE self) {
	struct IO_Event_Selector_URing *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
	
	return selector->ring.ring_fd < 0 || selector->owner != getpid() ? Qtrue : Qfalse;
}

#idle_duration ⇒ Object

# File 'ext/io/event/selector/uring.c', line 335

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

#io_close(_descriptor) ⇒ Object

# File 'ext/io/event/selector/uring.c', line 1084

VALUE IO_Event_Selector_URing_io_close(VALUE self, VALUE _descriptor) {
	struct IO_Event_Selector_URing *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
	
	// Ruby's fiber scheduler `io_close` hook is invoked with a raw integer file descriptor (Ruby 4.0+); it does not pass the `IO` object.
	int descriptor = RB_NUM2INT(_descriptor);
	
	if (ASYNC_CLOSE) {
		struct io_uring_sqe *sqe = io_get_sqe(selector);
		io_uring_prep_close(sqe, descriptor);
		io_uring_sqe_set_data(sqe, NULL);
		io_uring_submit_now(selector);
		
		// It would be nice to explore not flushing immediately, but instead deferring to the next select cycle.
		// The problem with this approach is that if the user expects the file descriptor to be closed immediately, (e.g. before fork), it may not be closed in time.
		// io_uring_submit_pending(selector);
	} else {
		close(descriptor);
	}
	
	// We don't wait for the result of close since it has no use in practice:
	return Qtrue;
}

#io_pread(fiber, io, buffer, _from, _length, _offset) ⇒ Object

# File 'ext/io/event/selector/uring.c', line 846

VALUE IO_Event_Selector_URing_io_pread(VALUE self, VALUE fiber, VALUE io, VALUE buffer, VALUE _from, VALUE _length, VALUE _offset) {
	struct IO_Event_Selector_URing *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
	
	void *base;
	size_t size;
	rb_io_buffer_get_bytes_for_writing(buffer, &base, &size);
	
	size_t length = NUM2SIZET(_length);
	size_t offset = NUM2SIZET(_offset);
	size_t total = 0;
	off_t from = NUM2OFFT(_from);
	
	// Ensure offset is within the bounds of the buffer to avoid size_t underflow and out-of-bounds pointer arithmetic on (char *)base + offset.
	if (offset > size) {
		return rb_fiber_scheduler_io_result(-1, EINVAL);
	} else if (offset == size) {
		return rb_fiber_scheduler_io_result(0, 0);
	}
	
	int descriptor = IO_Event_Selector_io_descriptor(io);
	
	size_t maximum_size = size - offset;
	while (maximum_size) {
		int result = io_read(selector, fiber, descriptor, (char*)base+offset, maximum_size, from);
		
		if (result > 0) {
			total += result;
			offset += result;
			from += result;
			if ((size_t)result >= length) break;
			length -= result;
		} else if (result == 0) {
			break;
		} else if (length > 0 && IO_Event_try_again(-result)) {
			IO_Event_Selector_URing_io_wait(self, fiber, io, RB_INT2NUM(IO_EVENT_READABLE));
		} else {
			return rb_fiber_scheduler_io_result(-1, -result);
		}
		
		maximum_size = size - offset;
	}
	
	return rb_fiber_scheduler_io_result(total, 0);
}

#io_pwrite(fiber, io, buffer, _from, _length, _offset) ⇒ Object

# File 'ext/io/event/selector/uring.c', line 1028

VALUE IO_Event_Selector_URing_io_pwrite(VALUE self, VALUE fiber, VALUE io, VALUE buffer, VALUE _from, VALUE _length, VALUE _offset) {
	struct IO_Event_Selector_URing *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
	
	const void *base;
	size_t size;
	rb_io_buffer_get_bytes_for_reading(buffer, &base, &size);
	
	size_t length = NUM2SIZET(_length);
	size_t offset = NUM2SIZET(_offset);
	size_t total = 0;
	off_t from = NUM2OFFT(_from);
	
	if (length > size) {
		rb_raise(rb_eRuntimeError, "Length exceeds size of buffer!");
	}

	// Ensure offset is within the bounds of the buffer to avoid size_t underflow and out-of-bounds pointer arithmetic on (char *)base + offset.
	if (offset > size) {
		return rb_fiber_scheduler_io_result(-1, EINVAL);
	} else if (offset == size) {
		return rb_fiber_scheduler_io_result(0, 0);
	}
	
	int descriptor = IO_Event_Selector_io_descriptor(io);
	
	size_t maximum_size = size - offset;
	while (maximum_size) {
		int result = io_write(selector, fiber, descriptor, (char*)base+offset, maximum_size, from);
		
		if (result > 0) {
			total += result;
			offset += result;
			from += result;
			if ((size_t)result >= length) break;
			length -= result;
		} else if (result == 0) {
			break;
		} else if (length > 0 && IO_Event_try_again(-result)) {
			IO_Event_Selector_URing_io_wait(self, fiber, io, RB_INT2NUM(IO_EVENT_WRITABLE));
		} else {
			return rb_fiber_scheduler_io_result(-1, -result);
		}
		
		maximum_size = size - offset;
	}
	
	return rb_fiber_scheduler_io_result(total, 0);
}

#io_read(*args) ⇒ Object

# File 'ext/io/event/selector/uring.c', line 833

static VALUE IO_Event_Selector_URing_io_read_compatible(int argc, VALUE *argv, VALUE self)
{
	rb_check_arity(argc, 4, 5);
	
	VALUE _offset = SIZET2NUM(0);
	
	if (argc == 5) {
		_offset = argv[4];
	}
	
	return IO_Event_Selector_URing_io_read(self, argv[0], argv[1], argv[2], argv[3], _offset);
}

#io_wait(fiber, io, events) ⇒ Object

# File 'ext/io/event/selector/uring.c', line 647

VALUE IO_Event_Selector_URing_io_wait(VALUE self, VALUE fiber, VALUE io, VALUE events) {
	struct IO_Event_Selector_URing *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
	
	int descriptor = IO_Event_Selector_io_descriptor(io);
	
	short flags = poll_flags_from_events(NUM2INT(events));
	
	if (DEBUG) fprintf(stderr, "IO_Event_Selector_URing_io_wait:io_uring_prep_poll_add(descriptor=%d, flags=%d, fiber=%p)\n", descriptor, flags, (void*)fiber);
	
	struct IO_Event_Selector_URing_Waiting waiting = {
		.fiber = fiber,
	};
	
	RB_OBJ_WRITTEN(self, Qundef, fiber);
	
	struct IO_Event_Selector_URing_Completion *completion = IO_Event_Selector_URing_Completion_acquire(selector, &waiting);
	
	struct io_uring_sqe *sqe = io_get_sqe(selector);
	io_uring_prep_poll_add(sqe, descriptor, flags);
	io_uring_sqe_set_data(sqe, completion);
	// If we are going to wait, we assume that we are waiting for a while:
	io_uring_submit_pending(selector);
	
	struct io_wait_arguments io_wait_arguments = {
		.selector = selector,
		.waiting = &waiting,
		.flags = flags
	};
	
	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/uring.c', line 1015

static VALUE IO_Event_Selector_URing_io_write_compatible(int argc, VALUE *argv, VALUE self)
{
	rb_check_arity(argc, 4, 5);
	
	VALUE _offset = SIZET2NUM(0);
	
	if (argc == 5) {
		_offset = argv[4];
	}
	
	return IO_Event_Selector_URing_io_write(self, argv[0], argv[1], argv[2], argv[3], _offset);
}

#loop ⇒ Object

# File 'ext/io/event/selector/uring.c', line 328

VALUE IO_Event_Selector_URing_loop(VALUE self) {
	struct IO_Event_Selector_URing *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
	
	return selector->backend.loop;
}

#process_wait(fiber, _pid, _flags) ⇒ Object

# File 'ext/io/event/selector/uring.c', line 536

VALUE IO_Event_Selector_URing_process_wait(VALUE self, VALUE fiber, VALUE _pid, VALUE _flags) {
	struct IO_Event_Selector_URing *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
	
	pid_t pid = NUM2PIDT(_pid);
	int flags = NUM2INT(_flags);
	
	int descriptor = pidfd_open(pid, 0);
	if (descriptor < 0) {
		rb_syserr_fail(errno, "IO_Event_Selector_URing_process_wait:pidfd_open");
	}
	rb_update_max_fd(descriptor);
	
	struct IO_Event_Selector_URing_Waiting waiting = {
		.fiber = fiber,
	};
	
	RB_OBJ_WRITTEN(self, Qundef, fiber);
	
	struct IO_Event_Selector_URing_Completion *completion = IO_Event_Selector_URing_Completion_acquire(selector, &waiting);
	
	struct process_wait_arguments process_wait_arguments = {
		.selector = selector,
		.waiting = &waiting,
		.pid = pid,
		.flags = flags,
		.descriptor = descriptor,
	};
	
	if (DEBUG) fprintf(stderr, "IO_Event_Selector_URing_process_wait:io_uring_prep_poll_add(%p)\n", (void*)fiber);
	struct io_uring_sqe *sqe = io_get_sqe(selector);
	io_uring_prep_poll_add(sqe, descriptor, POLLIN|POLLHUP|POLLERR);
	io_uring_sqe_set_data(sqe, completion);
	io_uring_submit_pending(selector);
	
	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/uring.c', line 384

VALUE IO_Event_Selector_URing_push(VALUE self, VALUE fiber)
{
	struct IO_Event_Selector_URing *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
	
	IO_Event_Selector_ready_push(&selector->backend, fiber);
	
	return Qnil;
}

#raise(*args) ⇒ Object

# File 'ext/io/event/selector/uring.c', line 394

VALUE IO_Event_Selector_URing_raise(int argc, VALUE *argv, VALUE self)
{
	struct IO_Event_Selector_URing *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
	
	return IO_Event_Selector_raise(&selector->backend, argc, argv);
}

#ready? ⇒ Boolean

Returns:

• (Boolean)

# File 'ext/io/event/selector/uring.c', line 402

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

#resume(*args) ⇒ Object

# File 'ext/io/event/selector/uring.c', line 368

VALUE IO_Event_Selector_URing_resume(int argc, VALUE *argv, VALUE self)
{
	struct IO_Event_Selector_URing *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
	
	return IO_Event_Selector_resume(&selector->backend, argc, argv);
}

#select(duration) ⇒ Object

# File 'ext/io/event/selector/uring.c', line 1273

VALUE IO_Event_Selector_URing_select(VALUE self, VALUE duration) {
	struct IO_Event_Selector_URing *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
	
	selector->idle_duration.tv_sec = 0;
	selector->idle_duration.tv_nsec = 0;
	
	// Flush any pending events:
	io_uring_submit_flush(selector);
	
#ifdef IORING_SETUP_DEFER_TASKRUN
	// With DEFER_TASKRUN the kernel holds completions as "deferred task work"
	// rather than placing them directly into the CQ.  We need to flush that work
	// into the CQ so the non-blocking select_process_completions below can see
	// it.  With TASKRUN_FLAG enabled the kernel sets IORING_SQ_TASKRUN in
	// sq.flags whenever task work is pending; a relaxed atomic load is enough
	// to check, and we only pay for an io_uring_enter syscall (via
	// io_uring_get_events) when there is actually deferred work to flush.
	if (selector->ring.flags & IORING_SETUP_DEFER_TASKRUN) {
#ifdef IORING_SETUP_TASKRUN_FLAG
		unsigned sq_flags = __atomic_load_n(selector->ring.sq.kflags, __ATOMIC_RELAXED);
		if (sq_flags & IORING_SQ_TASKRUN)
#endif
		{
			io_uring_get_events(&selector->ring);
		}
	}
#endif
	
	int ready = IO_Event_Selector_ready_flush(&selector->backend);
	
	int completed = select_process_completions(selector);
	
	// 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 && !completed && !selector->backend.ready) {
		// We might need to wait for events:
		struct select_arguments arguments = {
			.selector = selector,
			.result = 0,
			.timeout = NULL,
		};
		
		arguments.timeout = make_timeout(duration, &arguments.storage);
		
		if (!selector->backend.ready && select_blocking_allowed(arguments.timeout)) {
			struct timespec start_time;
			IO_Event_Time_current(&start_time);
			
			// This is a blocking operation, we wait for events:
			int 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);
			
			// After waiting/flushing the SQ, check if there are any completions:
			if (result > 0) {
				completed = select_process_completions(selector);
			}
		}
	}
	
	return RB_INT2NUM(completed);
}

#transfer ⇒ Object

# File 'ext/io/event/selector/uring.c', line 360

VALUE IO_Event_Selector_URing_transfer(VALUE self)
{
	struct IO_Event_Selector_URing *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
	
	return IO_Event_Selector_loop_yield(&selector->backend);
}

#wakeup ⇒ Object

# File 'ext/io/event/selector/uring.c', line 1342

VALUE IO_Event_Selector_URing_wakeup(VALUE self) {
	struct IO_Event_Selector_URing *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
	
	// Wake the selector by signalling the interrupt. This is safe from any thread
	// and never touches the ring's SQ, which is required for IORING_SETUP_SINGLE_ISSUER.
	if (selector->blocked) {
		IO_Event_Interrupt_signal(&selector->interrupt);
		return Qtrue;
	}
	
	return Qfalse;
}

#yield ⇒ Object

# File 'ext/io/event/selector/uring.c', line 376

VALUE IO_Event_Selector_URing_yield(VALUE self)
{
	struct IO_Event_Selector_URing *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
	
	return IO_Event_Selector_yield(&selector->backend);
}