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

Inherits:
Object
  • Object
show all
Defined in:
ext/io/event/selector/uring.c

Instance Method Summary collapse

Constructor Details

#initialize(loop) ⇒ Object



325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
# File 'ext/io/event/selector/uring.c', line 325

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

#closeObject



404
405
406
407
408
409
410
411
# File 'ext/io/event/selector/uring.c', line 404

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)


413
414
415
416
417
418
# File 'ext/io/event/selector/uring.c', line 413

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_durationObject



395
396
397
398
399
400
401
402
# File 'ext/io/event/selector/uring.c', line 395

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



1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
# File 'ext/io/event/selector/uring.c', line 1462

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, _first, _second) ⇒ Object



1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
# File 'ext/io/event/selector/uring.c', line 1137

VALUE IO_Event_Selector_URing_io_pread(VALUE self, VALUE fiber, VALUE io, VALUE buffer, VALUE _from, VALUE _first, VALUE _second) {
	struct IO_Event_Selector_URing *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
	
#if RUBY_FIBER_SCHEDULER_VERSION >= 4
	struct io_read_locked_arguments arguments = {
		.selector = selector,
		.fiber = fiber,
		.io = io,
		.from = NUM2OFFT(_from),
		.offset = NUM2SIZET(_first),
		.length = NUM2SIZET(_second),
		.positional = true,
	};
	
	return rb_io_buffer_locked_for_writing(buffer, io_read_locked, (VALUE)&arguments);
#else
	void *base;
	size_t size;
	rb_io_buffer_get_bytes_for_writing(buffer, &base, &size);
	
	size_t length = NUM2SIZET(_first);
	size_t offset = NUM2SIZET(_second);
	size_t total = 0;
	
	// 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);
	}
	off_t from = NUM2OFFT(_from);
	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);
#endif
}

#io_pwrite(fiber, io, buffer, _from, _first, _second) ⇒ Object



1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
# File 'ext/io/event/selector/uring.c', line 1394

VALUE IO_Event_Selector_URing_io_pwrite(VALUE self, VALUE fiber, VALUE io, VALUE buffer, VALUE _from, VALUE _first, VALUE _second) {
	struct IO_Event_Selector_URing *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
	
#if RUBY_FIBER_SCHEDULER_VERSION >= 4
	struct io_write_locked_arguments arguments = {
		.selector = selector,
		.fiber = fiber,
		.io = io,
		.from = NUM2OFFT(_from),
		.offset = NUM2SIZET(_first),
		.length = NUM2SIZET(_second),
		.positional = true,
	};
	
	return rb_io_buffer_locked_for_reading(buffer, io_write_locked, (VALUE)&arguments);
#else
	const void *base;
	size_t size;
	rb_io_buffer_get_bytes_for_reading(buffer, &base, &size);
	
	size_t length = NUM2SIZET(_first);
	size_t offset = NUM2SIZET(_second);
	size_t total = 0;
	
	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);
	}
	off_t from = NUM2OFFT(_from);
	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);
#endif
}

#io_read(*args) ⇒ Object



1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
# File 'ext/io/event/selector/uring.c', line 1051

VALUE IO_Event_Selector_URing_io_read(VALUE self, VALUE fiber, VALUE io, VALUE buffer, VALUE _first, VALUE _second) {
	struct IO_Event_Selector_URing *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
	
#if RUBY_FIBER_SCHEDULER_VERSION >= 4
	struct io_read_locked_arguments arguments = {
		.selector = selector,
		.fiber = fiber,
		.io = io,
		.offset = NUM2SIZET(_first),
		.length = NUM2SIZET(_second),
		.positional = false,
	};
	
	return rb_io_buffer_locked_for_writing(buffer, io_read_locked, (VALUE)&arguments);
#else
	void *base;
	size_t size;
	rb_io_buffer_get_bytes_for_writing(buffer, &base, &size);
	
	size_t length = NUM2SIZET(_first);
	size_t offset = NUM2SIZET(_second);
	size_t total = 0;
	
	// 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);
	off_t from = io_seekable(descriptor);
	size_t maximum_size = size - offset;
	
	// Are we performing a non-blocking read?
	if (!length) {
		// If the (maximum) length is zero, that indicates we just want to read whatever is available without blocking.
		// If we schedule this read into the URing, it will block until data is available, rather than returning immediately.
		int state = IO_Event_Selector_nonblock_set(descriptor);
		
		int result = read(descriptor, (char*)base+offset, maximum_size);
		int error = errno;
		
		IO_Event_Selector_nonblock_restore(descriptor, state);
		return rb_fiber_scheduler_io_result(result, error);
	}
	
	while (maximum_size) {
		int result = io_read(selector, fiber, descriptor, (char*)base+offset, maximum_size, from);
		
		if (result > 0) {
			total += result;
			offset += 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);
#endif
}

#io_wait(fiber, io, events) ⇒ Object



875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
# File 'ext/io/event/selector/uring.c', line 875

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



1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
# File 'ext/io/event/selector/uring.c', line 1318

VALUE IO_Event_Selector_URing_io_write(VALUE self, VALUE fiber, VALUE io, VALUE buffer, VALUE _first, VALUE _second) {
	struct IO_Event_Selector_URing *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector);
	
#if RUBY_FIBER_SCHEDULER_VERSION >= 4
	struct io_write_locked_arguments arguments = {
		.selector = selector,
		.fiber = fiber,
		.io = io,
		.offset = NUM2SIZET(_first),
		.length = NUM2SIZET(_second),
		.positional = false,
	};
	
	return rb_io_buffer_locked_for_reading(buffer, io_write_locked, (VALUE)&arguments);
#else
	const void *base;
	size_t size;
	rb_io_buffer_get_bytes_for_reading(buffer, &base, &size);
	
	size_t length = NUM2SIZET(_first);
	size_t offset = NUM2SIZET(_second);
	size_t total = 0;
	
	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);
	off_t from = io_seekable(descriptor);
	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;
			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);
#endif
}

#loopObject



388
389
390
391
392
393
# File 'ext/io/event/selector/uring.c', line 388

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



744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
# File 'ext/io/event/selector/uring.c', line 744

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);
	
#ifndef IO_EVENT_SELECTOR_URING_USE_WAITID
	// `pidfd_open` 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);
	}
	
	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);
#endif
	
	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,
#ifdef IO_EVENT_SELECTOR_URING_USE_WAITID
		.siginfo = {0},
#else
		.descriptor = descriptor,
#endif
	};
	
	struct io_uring_sqe *sqe = io_get_sqe(selector);
	
#ifdef IO_EVENT_SELECTOR_URING_USE_WAITID
	id_t id;
	idtype_t idtype = process_waitid_type(pid, &id);
	if (DEBUG) fprintf(stderr, "IO_Event_Selector_URing_process_wait:io_uring_prep_waitid(fiber=%p, idtype=%d, id=%d, flags=%d)\n", (void*)fiber, idtype, (int)id, flags);
#ifdef HAVE_RB_PROCESS_STATUS_FOR
	// Reap the child directly; the completion contains enough information to construct the Ruby process status value:
	io_uring_prep_waitid(sqe, idtype, id, &process_wait_arguments.siginfo, WEXITED, 0);
#else
	// `WNOWAIT` leaves the child in a waitable state so we can reap it with `rb_process_status_wait` afterwards and build a correct `Process::Status`:
	io_uring_prep_waitid(sqe, idtype, id, &process_wait_arguments.siginfo, WEXITED | WNOWAIT, 0);
#endif
#else
	if (DEBUG) fprintf(stderr, "IO_Event_Selector_URing_process_wait:io_uring_prep_poll_add(%p)\n", (void*)fiber);
	io_uring_prep_poll_add(sqe, descriptor, POLLIN|POLLHUP|POLLERR);
#endif
	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



444
445
446
447
448
449
450
451
452
# File 'ext/io/event/selector/uring.c', line 444

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



454
455
456
457
458
459
460
# File 'ext/io/event/selector/uring.c', line 454

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)


462
463
464
465
466
467
# File 'ext/io/event/selector/uring.c', line 462

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



428
429
430
431
432
433
434
# File 'ext/io/event/selector/uring.c', line 428

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



1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
# File 'ext/io/event/selector/uring.c', line 1642

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);
}

#transferObject



420
421
422
423
424
425
426
# File 'ext/io/event/selector/uring.c', line 420

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);
}

#wakeupObject



1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
# File 'ext/io/event/selector/uring.c', line 1711

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->backend.blocked) {
		IO_Event_Interrupt_signal(&selector->interrupt);
		return Qtrue;
	}
	
	return Qfalse;
}

#yieldObject



436
437
438
439
440
441
442
# File 'ext/io/event/selector/uring.c', line 436

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);
}