Class: FiberAudit::Runtime::ActiveOperations

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

Overview

Bounded process-local registry shared by runtime observers.

Defined Under Namespace

Classes: Entry, Handle, Snapshot

Constant Summary collapse

MAX_ENTRIES = 
10_000
MAX_SNAPSHOT = 
100

Instance Method Summary collapse

Constructor Details

#initialize(pid_source: Process.method(:pid), capacity: MAX_ENTRIES, snapshot_limit: MAX_SNAPSHOT) ⇒ ActiveOperations

Returns a new instance of ActiveOperations.



41
42
43
44
45
46
47
48
49
 
# File 'lib/fiber_audit/runtime/active_operations.rb', line 41

def initialize(pid_source: Process.method(:pid), capacity: MAX_ENTRIES, snapshot_limit: MAX_SNAPSHOT)
  validate_source!(pid_source)
  validate_limit!(capacity, 'capacity', MAX_ENTRIES)
  validate_limit!(snapshot_limit, 'snapshot_limit', MAX_SNAPSHOT)
  @pid_source = pid_source
  @capacity = capacity
  @snapshot_limit = snapshot_limit
  reset_for_process!(current_pid)
end

Instance Method Details

#finish(handle) ⇒ Object 



87
88
89
90
91
92
 
# File 'lib/fiber_audit/runtime/active_operations.rb', line 87

def finish(handle)
  ensure_current_process!
  return unless handle.is_a?(Handle) && handle.pid == @owner_pid

  @mutex.synchronize { @entries.delete(handle) }
end

#register(operation:, monotonic_ns:, location: nil, execution_context: :unknown, thread: Thread.current, fiber: Fiber.current, scheduler_snapshot: nil) ⇒ Object 



51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
 
# File 'lib/fiber_audit/runtime/active_operations.rb', line 51

def register(
  operation:,
  monotonic_ns:,
  location: nil,
  execution_context: :unknown,
  thread: Thread.current,
  fiber: Fiber.current,
  scheduler_snapshot: nil
)
  ensure_current_process!
  values = normalize_entry(
    operation: operation,
    location: location,
    execution_context: execution_context,
    monotonic_ns: monotonic_ns,
    thread: thread,
    fiber: fiber,
    scheduler_snapshot: scheduler_snapshot
  )

  @mutex.synchronize do
    return if @entries.size >= @capacity

    @sequence += 1
    entry = Entry.new(sequence: @sequence, **values)
    handle = Handle.new(
      pid: @owner_pid,
      thread_id: entry.thread_id,
      fiber_id: entry.fiber_id,
      sequence: entry.sequence
    )
    @entries[handle] = entry
    handle
  end
end

#size ⇒ Object 



107
108
109
110
 
# File 'lib/fiber_audit/runtime/active_operations.rb', line 107

def size
  ensure_current_process!
  @mutex.synchronize { @entries.size }
end

#snapshot(thread_id: nil) ⇒ Object 



94
 
# File 'lib/fiber_audit/runtime/active_operations.rb', line 94

def snapshot(thread_id: nil) = snapshot_with_metadata(thread_id: thread_id).entries

#snapshot_with_metadata(thread_id: nil) ⇒ Object 



96
97
98
99
100
101
102
103
104
105
 
# File 'lib/fiber_audit/runtime/active_operations.rb', line 96

def snapshot_with_metadata(thread_id: nil)
  ensure_current_process!
  normalized_thread_id = Validation.integer(thread_id, 'thread_id', allow_nil: true)
  @mutex.synchronize do
    entries = @entries.values
    entries = entries.select { |entry| entry.thread_id == normalized_thread_id } if normalized_thread_id
    ordered = entries.sort_by(&:sequence)
    Snapshot.new(entries: ordered.first(@snapshot_limit), total_count: ordered.size)
  end
end