Class: Goru::Scheduler

Inherits:
Object
  • Object
show all
Includes:
Is::Global, MonitorMixin
Defined in:
lib/goru/scheduler.rb

Overview

public

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(count: self.class.default_scheduler_count) ⇒ Scheduler

Returns a new instance of Scheduler.



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/goru/scheduler.rb', line 37

def initialize(count: self.class.default_scheduler_count)
  super()

  @stopping = false
  @routines = Thread::Queue.new
  @condition = new_cond

  @reactors = count.times.map {
    Reactor.new(queue: @routines, scheduler: self)
  }

  @threads = @reactors.map { |reactor|
    Thread.new {
      Thread.handle_interrupt(Interrupt => :never) do
        reactor.run
      end
    }
  }
end

Class Method Details

.default_scheduler_count ⇒ Object

public


32
33
34
 
# File 'lib/goru/scheduler.rb', line 32

def default_scheduler_count
  Etc.nprocessors
end

.go ⇒ Object

Prevent issues when including ‘Goru` at the toplevel.

public


26
27
28
 
# File 'lib/goru/scheduler.rb', line 26

def go(...)
  global.go(...)
end

Instance Method Details

#go(state = nil, io: nil, channel: nil, intent: nil, &block) ⇒ Object

public

Raises:

  • (ArgumentError) 


59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
 
# File 'lib/goru/scheduler.rb', line 59

def go(state = nil, io: nil, channel: nil, intent: nil, &block)
  raise ArgumentError, "cannot set both `io` and `channel`" if io && channel

  routine = if io
    Routines::IO.new(state, io: io, intent: intent, &block)
  elsif channel
    case intent
    when :r
      Routines::Channels::Readable.new(state, channel: channel, &block)
    when :w
      Routines::Channels::Writable.new(state, channel: channel, &block)
    end
  else
    Routine.new(state, &block)
  end

  @routines << routine
  @reactors.each(&:signal)

  routine
end

#signal(reactor) ⇒ Object

public


105
106
107
108
109
110
111
112
113
 
# File 'lib/goru/scheduler.rb', line 105

def signal(reactor)
  synchronize do
    if @reactors.all?(&:finished?)
      @stopping = true
    end

    @condition.signal
  end
end

#stop ⇒ Object

public


96
97
98
99
100
101
 
# File 'lib/goru/scheduler.rb', line 96

def stop
  @stopping = true
  @routines.close
  @reactors.each(&:stop)
  @threads.each(&:join)
end

#wait ⇒ Object

public


83
84
85
86
87
88
89
90
91
92
 
# File 'lib/goru/scheduler.rb', line 83

def wait
  synchronize do
    @condition.wait_until do
      @stopping
    end
  end
rescue Interrupt
ensure
  stop
end