Class: FiberAudit::Runtime::Heartbeat

Inherits:
Object
  • Object
show all
Defined in:
lib/fiber_audit/runtime/heartbeat.rb

Overview

Scheduler-owned progress fiber observed by the watchdog thread.

Defined Under Namespace

Classes: Snapshot

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(interval_ns:, clock: Clock.new, owner_thread: Thread.current, on_tick: ->(_heartbeat) {}, on_error: ->(_heartbeat, _error) {}) ⇒ Heartbeat

Returns a new instance of Heartbeat.

Raises:



13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
 
# File 'lib/fiber_audit/runtime/heartbeat.rb', line 13

def initialize(
  interval_ns:,
  clock: Clock.new,
  owner_thread: Thread.current,
  on_tick: ->(_heartbeat) {},
  on_error: ->(_heartbeat, _error) {}
)
  raise RuntimeContractError, 'clock must be a FiberAudit::Runtime::Clock' unless clock.is_a?(Clock)
  unless interval_ns.is_a?(Integer) && interval_ns.positive?
    raise RuntimeContractError, 'interval_ns must be a positive Integer'
  end
  raise RuntimeContractError, 'owner_thread must be a Thread' unless owner_thread.is_a?(Thread)
  raise RuntimeContractError, 'on_tick must respond to call' unless on_tick.respond_to?(:call)
  raise RuntimeContractError, 'on_error must respond to call' unless on_error.respond_to?(:call)

  @clock = clock
  @interval_ns = interval_ns
  @owner_thread = owner_thread
  @on_tick = on_tick
  @on_error = on_error
  @mutex = Mutex.new
  @sequence = 0
  @last_progress_ns = nil
  @fiber_id = nil
  @started = false
  @start_requested = false
  @stop_requested = false
end

Instance Attribute Details

#owner_thread ⇒ Object (readonly)

Returns the value of attribute owner_thread.



11
12
13
 
# File 'lib/fiber_audit/runtime/heartbeat.rb', line 11

def owner_thread
  @owner_thread
end

Instance Method Details

#request_stop ⇒ Object 



70
71
72
73
 
# File 'lib/fiber_audit/runtime/heartbeat.rb', line 70

def request_stop
  @mutex.synchronize { @stop_requested = true }
  self
end

#snapshot ⇒ Object 



75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/fiber_audit/runtime/heartbeat.rb', line 75

def snapshot
  @mutex.synchronize do
    Snapshot.new(
      sequence: @sequence,
      last_progress_ns: @last_progress_ns,
      thread_id: owner_thread.object_id,
      fiber_id: @fiber_id,
      started: @started,
      stop_requested: @stop_requested
    )
  end
end

#start(schedule: Fiber.method(:schedule), sleeper: Kernel.method(:sleep)) ⇒ Object 



42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/fiber_audit/runtime/heartbeat.rb', line 42

def start(schedule: Fiber.method(:schedule), sleeper: Kernel.method(:sleep))
  raise RuntimeContractError, 'schedule must respond to call' unless schedule.respond_to?(:call)
  raise RuntimeContractError, 'sleeper must respond to call' unless sleeper.respond_to?(:call)

  @mutex.synchronize do
    return self if @start_requested

    @start_requested = true
  end
  schedule.call { run(sleeper) }
  self
rescue StandardError
  @mutex.synchronize { @start_requested = false }
  raise
end

#started? ⇒ Boolean

Returns:

  • (Boolean) 


88
89
90
 
# File 'lib/fiber_audit/runtime/heartbeat.rb', line 88

def started?
  @mutex.synchronize { @started }
end

#stop_requested? ⇒ Boolean

Returns:

  • (Boolean) 


92
93
94
 
# File 'lib/fiber_audit/runtime/heartbeat.rb', line 92

def stop_requested?
  @mutex.synchronize { @stop_requested }
end

#tick ⇒ Object 



58
59
60
61
62
63
64
65
66
67
68
 
# File 'lib/fiber_audit/runtime/heartbeat.rb', line 58

def tick
  now_ns = @clock.monotonic_ns
  @mutex.synchronize do
    @sequence += 1
    @last_progress_ns = now_ns
    @fiber_id ||= Fiber.current.object_id
    @started = true
  end
  @on_tick.call(self)
  self
end