Class: Event::Selector::URing

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

Instance Method Summary collapse

Constructor Details

#initialize(loop) ⇒ Object 



93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
 
# File 'ext/event/selector/uring.c', line 93

VALUE Event_Selector_URing_initialize(VALUE self, VALUE loop) {
	struct Event_Selector_URing *data = NULL;
	TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
	
	Event_Selector_initialize(&data->backend, loop);
	int result = io_uring_queue_init(URING_ENTRIES, &data->ring, 0);
	
	if (result < 0) {
		rb_syserr_fail(-result, "io_uring_queue_init");
	}
	
	rb_update_max_fd(data->ring.ring_fd);
	
	return self;
}

Instance Method Details

#close ⇒ Object 



109
110
111
112
113
114
115
116
 
# File 'ext/event/selector/uring.c', line 109

VALUE Event_Selector_URing_close(VALUE self) {
	struct Event_Selector_URing *data = NULL;
	TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
	
	close_internal(data);
	
	return Qnil;
}

#io_close(io) ⇒ Object 



463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
 
# File 'ext/event/selector/uring.c', line 463

VALUE Event_Selector_URing_io_close(VALUE self, VALUE io) {
	struct Event_Selector_URing *data = NULL;
	TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
	
	int descriptor = Event_Selector_io_descriptor(io);

	if (ASYNC_CLOSE) {
		struct io_uring_sqe *sqe = io_get_sqe(data);
		
		io_uring_prep_close(sqe, descriptor);
		io_uring_sqe_set_data(sqe, NULL);
		io_uring_submit_now(data);
	} else {
		close(descriptor);
	}

	// We don't wait for the result of close since it has no use in pratice:
	return Qtrue;
}

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



376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
 
# File 'ext/event/selector/uring.c', line 376

VALUE Event_Selector_URing_io_read(VALUE self, VALUE fiber, VALUE io, VALUE buffer, VALUE _length) {
	struct Event_Selector_URing *data = NULL;
	TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
	
	int descriptor = Event_Selector_io_descriptor(io);
	
	void *base;
	size_t size;
	rb_io_buffer_get_mutable(buffer, &base, &size);
	
	size_t offset = 0;
	size_t length = NUM2SIZET(_length);
	
	while (length > 0) {
		size_t maximum_size = size - offset;
		int result = io_read(data, fiber, descriptor, (char*)base+offset, maximum_size);
		
		if (result == 0) {
			break;
		} else if (result > 0) {
			offset += result;
			if ((size_t)result >= length) break;
			length -= result;
		} else if (-result == EAGAIN || -result == EWOULDBLOCK) {
			Event_Selector_URing_io_wait(self, fiber, io, RB_INT2NUM(EVENT_READABLE));
		} else {
			rb_syserr_fail(-result, strerror(-result));
		}
	}
	
	return SIZET2NUM(offset);
}

#io_wait(fiber, io, events) ⇒ Object 



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
 
# File 'ext/event/selector/uring.c', line 333

VALUE Event_Selector_URing_io_wait(VALUE self, VALUE fiber, VALUE io, VALUE events) {
	struct Event_Selector_URing *data = NULL;
	TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
	
	int descriptor = Event_Selector_io_descriptor(io);
	struct io_uring_sqe *sqe = io_get_sqe(data);
	
	short flags = poll_flags_from_events(NUM2INT(events));
	
	if (DEBUG) fprintf(stderr, "Event_Selector_URing_io_wait:io_uring_prep_poll_add(descriptor=%d, flags=%d, fiber=%p)\n", descriptor, flags, (void*)fiber);
	
	io_uring_prep_poll_add(sqe, descriptor, flags);
	io_uring_sqe_set_data(sqe, (void*)fiber);
	
	// If we are going to wait, we assume that we are waiting for a while:
	io_uring_submit_pending(data);
	
	struct io_wait_arguments io_wait_arguments = {
		.data = data,
		.fiber = fiber,
		.flags = flags
	};
	
	return rb_rescue(io_wait_transfer, (VALUE)&io_wait_arguments, io_wait_rescue, (VALUE)&io_wait_arguments);
}

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



425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
 
# File 'ext/event/selector/uring.c', line 425

VALUE Event_Selector_URing_io_write(VALUE self, VALUE fiber, VALUE io, VALUE buffer, VALUE _length) {
	struct Event_Selector_URing *data = NULL;
	TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
	
	int descriptor = Event_Selector_io_descriptor(io);
	
	const void *base;
	size_t size;
	rb_io_buffer_get_immutable(buffer, &base, &size);
	
	size_t offset = 0;
	size_t length = NUM2SIZET(_length);
	
	if (length > size) {
		rb_raise(rb_eRuntimeError, "Length exceeds size of buffer!");
	}
	
	while (length > 0) {
		int result = io_write(data, fiber, descriptor, (char*)base+offset, length);
		
		if (result >= 0) {
			offset += result;
			if ((size_t)result >= length) break;
			length -= result;
		} else if (-result == EAGAIN || -result == EWOULDBLOCK) {
			Event_Selector_URing_io_wait(self, fiber, io, RB_INT2NUM(EVENT_WRITABLE));
		} else {
			rb_syserr_fail(-result, strerror(-result));
		}
	}
	
	return SIZET2NUM(offset);
}

#process_wait(fiber, pid, flags) ⇒ Object 



249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
 
# File 'ext/event/selector/uring.c', line 249

VALUE Event_Selector_URing_process_wait(VALUE self, VALUE fiber, VALUE pid, VALUE flags) {
	struct Event_Selector_URing *data = NULL;
	TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
	
	struct process_wait_arguments process_wait_arguments = {
		.data = data,
		.pid = NUM2PIDT(pid),
		.flags = NUM2INT(flags),
	};
	
	process_wait_arguments.descriptor = pidfd_open(process_wait_arguments.pid, 0);
	rb_update_max_fd(process_wait_arguments.descriptor);
	
	struct io_uring_sqe *sqe = io_get_sqe(data);

	if (DEBUG) fprintf(stderr, "Event_Selector_URing_process_wait:io_uring_prep_poll_add(%p)\n", (void*)fiber);
	io_uring_prep_poll_add(sqe, process_wait_arguments.descriptor, POLLIN|POLLHUP|POLLERR);
	io_uring_sqe_set_data(sqe, (void*)fiber);
	io_uring_submit_pending(data);

	return rb_ensure(process_wait_transfer, (VALUE)&process_wait_arguments, process_wait_ensure, (VALUE)&process_wait_arguments);
}

#push(fiber) ⇒ Object 



142
143
144
145
146
147
148
149
150
 
# File 'ext/event/selector/uring.c', line 142

VALUE Event_Selector_URing_push(VALUE self, VALUE fiber)
{
	struct Event_Selector_URing *data = NULL;
	TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
	
	Event_Selector_queue_push(&data->backend, fiber);
	
	return Qnil;
}

#raise(*args) ⇒ Object 



152
153
154
155
156
157
158
 
# File 'ext/event/selector/uring.c', line 152

VALUE Event_Selector_URing_raise(int argc, VALUE *argv, VALUE self)
{
	struct Event_Selector_URing *data = NULL;
	TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
	
	return Event_Selector_raise(&data->backend, argc, argv);
}

#ready? ⇒ Boolean

Returns:

  • (Boolean) 


160
161
162
163
164
165
 
# File 'ext/event/selector/uring.c', line 160

VALUE Event_Selector_URing_ready_p(VALUE self) {
	struct Event_Selector_URing *data = NULL;
	TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
	
	return data->backend.ready ? Qtrue : Qfalse;
}

#resume(*args) ⇒ Object 



126
127
128
129
130
131
132
 
# File 'ext/event/selector/uring.c', line 126

VALUE Event_Selector_URing_resume(int argc, VALUE *argv, VALUE self)
{
	struct Event_Selector_URing *data = NULL;
	TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
	
	return Event_Selector_resume(&data->backend, argc, argv);
}

#select(duration) ⇒ Object 



583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
 
# File 'ext/event/selector/uring.c', line 583

VALUE Event_Selector_URing_select(VALUE self, VALUE duration) {
	struct Event_Selector_URing *data = NULL;
	TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
	
	int ready = Event_Selector_queue_flush(&data->backend);
	
	int result = select_process_completions(&data->ring);

	// If the ready list was empty and we didn't process any completions:
	if (!ready && result == 0) {
		// We might need to wait for events:
		struct select_arguments arguments = {
			.data = data,
			.timeout = NULL,
		};
		
		arguments.timeout = make_timeout(duration, &arguments.storage);
		
		if (!data->backend.ready && !timeout_nonblocking(arguments.timeout)) {
			// This is a blocking operation, we wait for events:
			result = select_internal_without_gvl(&arguments);
		} else {
			// The timeout specified required "nonblocking" behaviour so we just flush the SQ if required:
			io_uring_submit_flush(data);
		}

		// After waiting/flushing the SQ, check if there are any completions:
		result = select_process_completions(&data->ring);
	}
	
	return RB_INT2NUM(result);
}

#transfer ⇒ Object 



118
119
120
121
122
123
124
 
# File 'ext/event/selector/uring.c', line 118

VALUE Event_Selector_URing_transfer(VALUE self)
{
	struct Event_Selector_URing *data = NULL;
	TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
	
	return Event_Selector_fiber_transfer(data->backend.loop, 0, NULL);
}

#yield ⇒ Object 



134
135
136
137
138
139
140
 
# File 'ext/event/selector/uring.c', line 134

VALUE Event_Selector_URing_yield(VALUE self)
{
	struct Event_Selector_URing *data = NULL;
	TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
	
	return Event_Selector_yield(&data->backend);
}