Class: Plushie::TimerScheduler

Inherits:
Object
  • Object
show all
Defined in:
lib/plushie/timer_scheduler.rb,
sig/plushie/timer_scheduler.rbs

Overview

Single-thread timer scheduler for timer subscriptions.

Replaces the per-timer Thread.new pattern with one thread that manages all timers via IO.select with a deadline-based timeout. Adding or cancelling a timer wakes the scheduler through a pipe so it can recalculate the next deadline immediately.

Examples:

scheduler = TimerScheduler.new
scheduler.schedule(tag: "tick", interval_ms: 1000, event_queue: queue)
scheduler.cancel("tick")
scheduler.stop

Instance Method Summary collapse

Constructor Details

#initialize ⇒ TimerScheduler

Returns a new instance of TimerScheduler.



19
20
21
22
23
24
25
26
 
# File 'lib/plushie/timer_scheduler.rb', line 19

def initialize
  @pipe_r, @pipe_w = IO.pipe
  @timers = {}
  @mutex = Mutex.new
  @stop_mutex = Mutex.new
  @thread = Thread.new { run_loop }
  @thread.name = "plushie-timer-scheduler"
end

Instance Method Details

#cancel(tag)

This method returns an undefined value.

Parameters:

  • (Symbol) 


40
41
42
43
 
# File 'lib/plushie/timer_scheduler.rb', line 40

def cancel(tag)
  @mutex.synchronize { @timers.delete(tag) }
  wake
end

#drain_pipe

This method returns an undefined value.



68
69
70
71
72
 
# File 'lib/plushie/timer_scheduler.rb', line 68

def drain_pipe
  @pipe_r.read_nonblock(4096)
rescue IOError, IO::EAGAINWaitReadable
  nil
end

#run_loop

This method returns an undefined value.



74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/plushie/timer_scheduler.rb', line 74

def run_loop
  loop do
    now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
    next_deadline = nil

    timers_to_fire = []
    @mutex.synchronize do
      @timers.each do |tag, timer|
        if now >= timer[:deadline]
          timers_to_fire << [tag, timer[:queue]]
          timer[:deadline] = now + timer[:interval]
        end
        next_deadline = timer[:deadline] if next_deadline.nil? || timer[:deadline] < next_deadline
      end
    end

    timers_to_fire.each do |tag, queue|
      BoundedQueue.push(queue, [:timer_tick, tag])
    end

    drain_pipe

    timeout = next_deadline ? [next_deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC), 0.001].max : 3600
    IO.select([@pipe_r], [], [], timeout)
  end
rescue IOError
  # Pipe closed during shutdown
end

#schedule(tag:, interval_ms:, event_queue:)

This method returns an undefined value.

Parameters:

  • tag: (Symbol)
  • interval_ms: (Numeric)
  • event_queue: (Thread::Queue) 


28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/plushie/timer_scheduler.rb', line 28

def schedule(tag:, interval_ms:, event_queue:)
  interval = interval_ms / 1000.0
  @mutex.synchronize do
    @timers[tag] = {
      interval: interval,
      deadline: Process.clock_gettime(Process::CLOCK_MONOTONIC) + interval,
      queue: event_queue
    }
  end
  wake
end

#stop

This method returns an undefined value.



45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/plushie/timer_scheduler.rb', line 45

def stop
  thread = nil
  @stop_mutex.synchronize do
    thread = @thread
    @thread = nil
    @pipe_w&.close unless @pipe_w&.closed?
    @pipe_r&.close unless @pipe_r&.closed?
  end
  return unless thread
  return if thread == Thread.current

  thread.kill
  thread.join(0.5)
end

#wake

This method returns an undefined value.



62
63
64
65
66
 
# File 'lib/plushie/timer_scheduler.rb', line 62

def wake
  @pipe_w.write_nonblock(".")
rescue IOError, Errno::EPIPE
  nil
end