Class: IO::Event::Selector::Select

Inherits:
Object
  • Object
show all
Defined in:
lib/io/event/selector/select.rb

Overview

A pure-Ruby implementation of the event selector.

Defined Under Namespace

Classes: Optional, Waiter

Constant Summary collapse

EAGAIN = 
-Errno::EAGAIN::Errno
EWOULDBLOCK = 
-Errno::EWOULDBLOCK::Errno

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(loop) ⇒ Select

Initialize the selector with the given event loop fiber.



14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
 
# File 'lib/io/event/selector/select.rb', line 14

def initialize(loop)
	@loop = loop
	@owner = Process.pid
	
	@waiting = Hash.new.compare_by_identity
	
	# Flag indicating whether the selector is currently blocked in a system call.
	# Set to true when blocked in ::IO.select, false otherwise.
	# Used by wakeup() to determine if an interrupt signal is needed.
	@blocked = false
	
	@ready = Queue.new
	@interrupt = Interrupt.attach(self)
	
	@idle_duration = 0.0
end

Instance Attribute Details

#idle_duration ⇒ Object (readonly)

Returns the value of attribute idle_duration.



35
36
37
 
# File 'lib/io/event/selector/select.rb', line 35

def idle_duration
  @idle_duration
end

#loop ⇒ Object (readonly)

Returns the value of attribute loop.



32
33
34
 
# File 'lib/io/event/selector/select.rb', line 32

def loop
  @loop
end

#The event loop fiber.(eventloopfiber.) ⇒ Object (readonly) 



32
 
# File 'lib/io/event/selector/select.rb', line 32

attr :loop

Instance Method Details

#close ⇒ Object

Close the selector and release any resources.



49
50
51
52
53
54
 
# File 'lib/io/event/selector/select.rb', line 49

def close
	@interrupt.close
	
	@loop = nil
	@waiting = nil
end

#closed? ⇒ Boolean

Returns:

  • (Boolean) 


57
58
59
 
# File 'lib/io/event/selector/select.rb', line 57

def closed?
	@loop.nil? || @owner != Process.pid
end

#io_read(fiber, io, buffer, length, offset = 0) ⇒ Object

Read from the given IO to the buffer.



205
206
207
208
209
210
211
212
213
214
215
 
# File 'lib/io/event/selector/select.rb', line 205

def io_read(fiber, io, buffer, offset, length)
	if offset > buffer.size || length > buffer.size - offset
		return -Errno::EINVAL::Errno
	elsif length == 0
		return 0
	end
	
	Selector.nonblock(io) do
		return Fiber.blocking{buffer.read(io, offset, length)}
	end
end

#io_select(readable, writable, priority, timeout) ⇒ Object

Wait for multiple IO objects to become readable or writable.



186
187
188
189
190
 
# File 'lib/io/event/selector/select.rb', line 186

def io_select(readable, writable, priority, timeout)
	Thread.new do
		IO.select(readable, writable, priority, timeout)
	end.value
end

#io_wait(fiber, io, events) ⇒ Object

Wait for the given IO to become readable or writable.



173
174
175
176
177
178
179
 
# File 'lib/io/event/selector/select.rb', line 173

def io_wait(fiber, io, events)
	waiter = @waiting[io] = Waiter.new(fiber, events, @waiting[io])
	
	@loop.transfer || false
ensure
	waiter&.invalidate
end

#io_write(fiber, io, buffer, length, offset = 0) ⇒ Object

Write to the given IO from the buffer.



221
222
223
224
225
226
227
228
229
230
231
 
# File 'lib/io/event/selector/select.rb', line 221

def io_write(fiber, io, buffer, offset, length)
	if offset > buffer.size || length > buffer.size - offset
		return -Errno::EINVAL::Errno
	elsif length == 0
		return 0
	end
	
	Selector.nonblock(io) do
		return Fiber.blocking{buffer.write(io, offset, length)}
	end
end

#process_wait(fiber, pid, flags) ⇒ Object

Wait for a process to change state.



314
315
316
 
# File 'lib/io/event/selector/select.rb', line 314

def process_wait(fiber, pid, flags)
	Selector.process_wait(pid, flags)
end

#push(fiber) ⇒ Object

Append the given fiber into the ready list.



105
106
107
 
# File 'lib/io/event/selector/select.rb', line 105

def push(fiber)
	@ready.push(fiber)
end

#raise(fiber, *arguments, **options) ⇒ Object

Transfer to the given fiber and raise an exception. Put the current fiber into the ready list.



110
111
112
113
114
115
116
117
 
# File 'lib/io/event/selector/select.rb', line 110

def raise(fiber, *arguments, **options)
	optional = Optional.new(Fiber.current)
	@ready.push(optional)
	
	fiber.raise(*arguments, **options)
ensure
	optional.nullify
end

#ready? ⇒ Boolean

Returns:

  • (Boolean) 


120
121
122
 
# File 'lib/io/event/selector/select.rb', line 120

def ready?
	!@ready.empty?
end

#resume(fiber, *arguments) ⇒ Object

Transfer from the current fiber to the specified fiber. Put the current fiber into the ready list.



85
86
87
88
89
90
91
92
 
# File 'lib/io/event/selector/select.rb', line 85

def resume(fiber, *arguments)
	optional = Optional.new(Fiber.current)
	@ready.push(optional)
	
	fiber.transfer(*arguments)
ensure
	optional.nullify
end

#select(duration = nil) ⇒ Object

Wait for IO events or a timeout.



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
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
 
# File 'lib/io/event/selector/select.rb', line 335

def select(duration = nil)
	if pop_ready
		# If we have popped items from the ready list, they may influence the duration calculation, so we don't delay the event loop:
		duration = 0
	end
	
	readable = Array.new
	writable = Array.new
	priority = Array.new
	
	@waiting.delete_if do |io, waiter|
		if io.closed?
			# When an IO is closed, we silently drop it. Ruby 4's `rb_thread_io_close_interrupt` will take care of interrupting any fibers waiting on the closed IO, so we don't need to do anything here.
			true
		else
			waiter.each do |fiber, events|
				if (events & IO::READABLE) > 0
					readable << io
				end
				
				if (events & IO::WRITABLE) > 0
					writable << io
				end
				
				if (events & IO::PRIORITY) > 0
					priority << io
				end
			end
			
			false
		end
	end
	
	duration = 0 unless @ready.empty?
	error = nil
	
	if duration&.>(0)
		start_time = Process.clock_gettime(Process::CLOCK_MONOTONIC)
	else
		@idle_duration = 0.0
	end
	
	# We need to handle interrupts on blocking IO. Every other implementation uses EINTR, but that doesn't work with `::IO.select` as it will retry the call on EINTR.
	Thread.handle_interrupt(::Exception => :on_blocking) do
		@blocked = true
		readable, writable, priority = ::IO.select(readable, writable, priority, duration)
	rescue ::Exception => error
		# Requeue below...
	ensure
		@blocked = false
		if start_time
			end_time = Process.clock_gettime(Process::CLOCK_MONOTONIC)
			@idle_duration = end_time - start_time
		end
	end
	
	if error
		if error.is_a?(IOError) || error.is_a?(Errno::EBADF)
			# This can happen if an IO is closed while we're blocked in ::IO.select. Ruby 4's `rb_thread_io_close_interrupt` will take care of interrupting any fibers waiting on the closed IO, so we don't need to do anything here, except try again:
			return 0
		end
		
		# For all other errors (e.g. thread interrupts), re-queue on the scheduler thread:
		Thread.current.raise(error)
		return 0
	end
	
	ready = Hash.new(0).compare_by_identity
	
	readable&.each do |io|
		# Skip any IO that was closed/reused after IO.select returned - its fd number
		# may now belong to a different file, so resuming the waiter would be wrong:
		ready[io] |= IO::READABLE unless io.closed?
	end
	
	writable&.each do |io|
		ready[io] |= IO::WRITABLE unless io.closed?
	end
	
	priority&.each do |io|
		ready[io] |= IO::PRIORITY unless io.closed?
	end
	
	ready.each do |io, events|
		@waiting.delete(io).dispatch(events) do |waiter|
			# Re-schedule the waiting IO:
			waiter.tail = @waiting[io]
			@waiting[io] = waiter
		end
	end
	
	return ready.size
end

#This is the amount of time the event loop was idle during the last select call.=(istheamountoftimetheeventloopwasidleduringthelastselectcall. = (value)) ⇒ Object 



35
 
# File 'lib/io/event/selector/select.rb', line 35

attr :idle_duration

#transfer ⇒ Object

Transfer from the current fiber to the event loop.



80
81
82
 
# File 'lib/io/event/selector/select.rb', line 80

def transfer
	@loop.transfer
end

#wakeup ⇒ Object

Wake up the event loop if it is currently sleeping.



38
39
40
41
42
43
44
45
46
 
# File 'lib/io/event/selector/select.rb', line 38

def wakeup
	if @blocked
		@interrupt.signal
		
		return true
	end
	
	return false
end

#yield ⇒ Object

Yield from the current fiber back to the event loop. Put the current fiber into the ready list.



95
96
97
98
99
100
101
102
 
# File 'lib/io/event/selector/select.rb', line 95

def yield
	optional = Optional.new(Fiber.current)
	@ready.push(optional)
	
	@loop.transfer
ensure
	optional.nullify
end