Class: IO::Event::Selector::EPoll

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

Instance Method Summary collapse

Constructor Details

#initialize(loop) ⇒ Object 



340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
 
# File 'ext/io/event/selector/epoll.c', line 340

VALUE IO_Event_Selector_EPoll_initialize(VALUE self, VALUE loop) {
	struct IO_Event_Selector_EPoll *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_EPoll, &IO_Event_Selector_EPoll_Type, selector);
	
	IO_Event_Selector_initialize(&selector->backend, self, loop);
	int result = epoll_create1(EPOLL_CLOEXEC);
	
	if (result == -1) {
		rb_sys_fail("IO_Event_Selector_EPoll_initialize:epoll_create");
	} else {
		selector->descriptor = result;
		selector->owner = getpid();
		
		rb_update_max_fd(selector->descriptor);
	}
	
	IO_Event_Interrupt_open(&selector->interrupt);
	IO_Event_Interrupt_add(&selector->interrupt, selector);
	
	return self;
}

Instance Method Details

#close ⇒ Object 



378
379
380
381
382
383
384
385
 
# File 'ext/io/event/selector/epoll.c', line 378

VALUE IO_Event_Selector_EPoll_close(VALUE self) {
	struct IO_Event_Selector_EPoll *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_EPoll, &IO_Event_Selector_EPoll_Type, selector);
	
	close_internal(selector);
	
	return Qnil;
}

#closed? ⇒ Boolean

Returns:

  • (Boolean) 


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

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

#idle_duration ⇒ Object 



369
370
371
372
373
374
375
376
 
# File 'ext/io/event/selector/epoll.c', line 369

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

#io_read(*args) ⇒ Object 



658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
 
# File 'ext/io/event/selector/epoll.c', line 658

VALUE IO_Event_Selector_EPoll_io_read(VALUE self, VALUE fiber, VALUE io, VALUE buffer, VALUE _first, VALUE _second) {
	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 



559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
 
# File 'ext/io/event/selector/epoll.c', line 559

VALUE IO_Event_Selector_EPoll_io_wait(VALUE self, VALUE fiber, VALUE io, VALUE events) {
	struct IO_Event_Selector_EPoll *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_EPoll, &IO_Event_Selector_EPoll_Type, selector);
	
	int descriptor = IO_Event_Selector_io_descriptor(io); 
	
	struct IO_Event_Selector_EPoll_Waiting waiting = {
		.list = {.type = &IO_Event_Selector_EPoll_io_wait_list_type},
		.fiber = fiber,
		.events = RB_NUM2INT(events),
	};
	
	RB_OBJ_WRITTEN(self, Qundef, fiber);
	
	int result = IO_Event_Selector_EPoll_Waiting_register(selector, io, descriptor, &waiting);
	
	if (result == -1) {
		if (errno == EPERM) {
			IO_Event_Selector_ready_push(&selector->backend, fiber);
			IO_Event_Selector_yield(&selector->backend);
			return events;
		}
		
		rb_sys_fail("IO_Event_Selector_EPoll_io_wait:IO_Event_Selector_EPoll_Waiting_register");
	}
	
	struct io_wait_arguments io_wait_arguments = {
		.selector = selector,
		.waiting = &waiting,
	};
	
	return rb_ensure(io_wait_transfer, (VALUE)&io_wait_arguments, io_wait_ensure, (VALUE)&io_wait_arguments);
}

#io_write(*args) ⇒ Object 



790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
 
# File 'ext/io/event/selector/epoll.c', line 790

VALUE IO_Event_Selector_EPoll_io_write(VALUE self, VALUE fiber, VALUE io, VALUE buffer, VALUE _first, VALUE _second) {
	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 



362
363
364
365
366
367
 
# File 'ext/io/event/selector/epoll.c', line 362

VALUE IO_Event_Selector_EPoll_loop(VALUE self) {
	struct IO_Event_Selector_EPoll *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_EPoll, &IO_Event_Selector_EPoll_Type, selector);
	
	return selector->backend.loop;
}

#process_wait(fiber, _pid, _flags) ⇒ Object 



477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
 
# File 'ext/io/event/selector/epoll.c', line 477

VALUE IO_Event_Selector_EPoll_process_wait(VALUE self, VALUE fiber, VALUE _pid, VALUE _flags) {
	struct IO_Event_Selector_EPoll *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_EPoll, &IO_Event_Selector_EPoll_Type, selector);
	
	pid_t pid = NUM2PIDT(_pid);
	int flags = NUM2INT(_flags);
	
	// `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 == -1) {
		rb_sys_fail("IO_Event_Selector_EPoll_process_wait:pidfd_open");
	}
	
	rb_update_max_fd(descriptor);
	
	// `pidfd_open` (above) may be edge triggered, so we need to check if the process is already exited, and if so, return immediately, otherwise we will block indefinitely.
	VALUE status = IO_Event_Selector_process_status_reap(pid, flags);
	if (status != Qnil) {
		close(descriptor);
		return status;
	}
	
	struct IO_Event_Selector_EPoll_Waiting waiting = {
		.list = {.type = &IO_Event_Selector_EPoll_process_wait_list_type},
		.fiber = fiber,
		.events = IO_EVENT_READABLE,
	};
	
	RB_OBJ_WRITTEN(self, Qundef, fiber);
	
	int result = IO_Event_Selector_EPoll_Waiting_register(selector, _pid, descriptor, &waiting);
	
	if (result == -1) {
		close(descriptor);
		rb_sys_fail("IO_Event_Selector_EPoll_process_wait:IO_Event_Selector_EPoll_Waiting_register");
	}
	
	struct process_wait_arguments process_wait_arguments = {
		.selector = selector,
		.pid = pid,
		.flags = flags,
		.descriptor = descriptor,
		.waiting = &waiting,
	};
	
	return rb_ensure(process_wait_transfer, (VALUE)&process_wait_arguments, process_wait_ensure, (VALUE)&process_wait_arguments);
}

#push(fiber) ⇒ Object 



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

VALUE IO_Event_Selector_EPoll_push(VALUE self, VALUE fiber)
{
	struct IO_Event_Selector_EPoll *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_EPoll, &IO_Event_Selector_EPoll_Type, selector);
	
	IO_Event_Selector_ready_push(&selector->backend, fiber);
	
	return Qnil;
}

#raise(*args) ⇒ Object 



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

VALUE IO_Event_Selector_EPoll_raise(int argc, VALUE *argv, VALUE self)
{
	struct IO_Event_Selector_EPoll *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_EPoll, &IO_Event_Selector_EPoll_Type, selector);
	
	return IO_Event_Selector_raise(&selector->backend, argc, argv);
}

#ready? ⇒ Boolean

Returns:

  • (Boolean) 


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

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

#resume(*args) ⇒ Object 



402
403
404
405
406
407
408
 
# File 'ext/io/event/selector/epoll.c', line 402

VALUE IO_Event_Selector_EPoll_resume(int argc, VALUE *argv, VALUE self)
{
	struct IO_Event_Selector_EPoll *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_EPoll, &IO_Event_Selector_EPoll_Type, selector);
	
	return IO_Event_Selector_resume(&selector->backend, argc, argv);
}

#select(duration) ⇒ Object 



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
1121
1122
1123
1124
1125
 
# File 'ext/io/event/selector/epoll.c', line 1074

VALUE IO_Event_Selector_EPoll_select(VALUE self, VALUE duration) {
	struct IO_Event_Selector_EPoll *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_EPoll, &IO_Event_Selector_EPoll_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 = EPOLL_MAX_EVENTS,
		.result = 0,
		.storage = {
			.tv_sec = 0,
			.tv_nsec = 0
		},
		.saved = {},
	};

	arguments.timeout = &arguments.storage;

	// Process any currently pending events:
	int result = select_internal_with_gvl(&arguments);
	
	// 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);
			
			// Wait for events to occur:
			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 



394
395
396
397
398
399
400
 
# File 'ext/io/event/selector/epoll.c', line 394

VALUE IO_Event_Selector_EPoll_transfer(VALUE self)
{
	struct IO_Event_Selector_EPoll *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_EPoll, &IO_Event_Selector_EPoll_Type, selector);
	
	return IO_Event_Selector_loop_yield(&selector->backend);
}

#wakeup ⇒ Object 



1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
 
# File 'ext/io/event/selector/epoll.c', line 1127

VALUE IO_Event_Selector_EPoll_wakeup(VALUE self) {
	struct IO_Event_Selector_EPoll *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_EPoll, &IO_Event_Selector_EPoll_Type, selector);
	
	// If we are blocking, we can schedule a nop event to wake up the selector:
	if (selector->backend.blocked) {
		IO_Event_Interrupt_signal(&selector->interrupt);
		
		return Qtrue;
	}
	
	return Qfalse;
}

#yield ⇒ Object 



410
411
412
413
414
415
416
 
# File 'ext/io/event/selector/epoll.c', line 410

VALUE IO_Event_Selector_EPoll_yield(VALUE self)
{
	struct IO_Event_Selector_EPoll *selector = NULL;
	TypedData_Get_Struct(self, struct IO_Event_Selector_EPoll, &IO_Event_Selector_EPoll_Type, selector);
	
	return IO_Event_Selector_yield(&selector->backend);
}