Class: Mxrb::Runtime::Scheduler

Inherits:
Object
  • Object
show all
Defined in:
lib/mxrb/runtime/scheduler.rb

Overview

Small stdlib-only scheduled-event runner. Execution, time, and sleeping are injectable so applications can connect it to their Ruby microflow runtime and test it without waiting for wall-clock time.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(project, executor: nil, clock: -> { Time.now.utc }, sleeper: Kernel.method(:sleep), poll_interval: 1.0, skip_overlap: true, async: true, on_error: nil, logger: nil, coordinator: nil, lease_ttl: 300, owner: nil) ⇒ Scheduler

Returns a new instance of Scheduler.

Raises:

  • (ArgumentError) 


32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/mxrb/runtime/scheduler.rb', line 32

def initialize(project, executor: nil, clock: -> { Time.now.utc },
               sleeper: Kernel.method(:sleep), poll_interval: 1.0,
               skip_overlap: true, async: true, on_error: nil, logger: nil,
               coordinator: nil, lease_ttl: 300, owner: nil)
  raise ArgumentError, 'poll_interval must be positive' unless poll_interval.to_f.positive?
  raise ArgumentError, 'lease_ttl must be positive' unless lease_ttl.to_f.positive?

  @project = project
  @executor = executor || default_executor
  @clock = clock
  @sleeper = sleeper
  @poll_interval = poll_interval.to_f
  @skip_overlap = skip_overlap
  @async = async
  @on_error = on_error
  @logger = logger
  @coordinator = coordinator || MemorySharedStore.new
  @lease_ttl = lease_ttl.to_f
  @owner = owner || "#{Process.pid}-#{SecureRandom.uuid}"
  @time_zones = {}
  @jobs = load_jobs.freeze
  @errors = []
  @last_slots = {}
  @running_jobs = {}
  @workers = []
  @mutex = Mutex.new
  @stopping = false
  @thread = nil
end

Instance Attribute Details

#errors ⇒ Object (readonly)

Returns the value of attribute errors.



30
31
32
 
# File 'lib/mxrb/runtime/scheduler.rb', line 30

def errors
  @errors
end

#jobs ⇒ Object (readonly)

Returns the value of attribute jobs.



30
31
32
 
# File 'lib/mxrb/runtime/scheduler.rb', line 30

def jobs
  @jobs
end

Instance Method Details

#running? ⇒ Boolean

Returns:

  • (Boolean) 


62
 
# File 'lib/mxrb/runtime/scheduler.rb', line 62

def running? = @mutex.synchronize { @thread&.alive? == true }

#shutdown(wait: true) ⇒ Object Also known as: stop 



75
76
77
78
79
80
81
82
83
84
 
# File 'lib/mxrb/runtime/scheduler.rb', line 75

def shutdown(wait: true)
  thread = @mutex.synchronize do
    @stopping = true
    @thread
  end
  thread&.join if wait && thread != Thread.current
  wait_for_jobs if wait
  @mutex.synchronize { @thread = nil unless @thread&.alive? }
  self
end

#start ⇒ Object 



64
65
66
67
68
69
70
71
72
73
 
# File 'lib/mxrb/runtime/scheduler.rb', line 64

def start
  @mutex.synchronize do
    return self if @thread&.alive?

    @stopping = false
    @thread = Thread.new { run_loop }
    @thread.name = 'mxrb-scheduler' if @thread.respond_to?(:name=)
  end
  self
end

#tick(now = @clock.call, async: @async) ⇒ Object

Evaluates all jobs once and returns the jobs that were dispatched.



88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/mxrb/runtime/scheduler.rb', line 88

def tick(now = @clock.call, async: @async)
  cleanup_workers
  jobs.filter_map do |job|
    slot = due_slot(job, now)
    next unless slot
    next unless reserve(job, slot, now)

    dispatch(job, slot, async:)
    job
  end
end

#wait_for_jobs ⇒ Object 



100
101
102
103
104
105
106
107
108
109
 
# File 'lib/mxrb/runtime/scheduler.rb', line 100

def wait_for_jobs
  loop do
    workers = @mutex.synchronize { @workers.select(&:alive?) }
    break if workers.empty?

    workers.each { _1.join unless _1 == Thread.current }
  end
  cleanup_workers
  self
end