Class: Lich::Common::SharedBuffer

Inherits:
Object
  • Object
show all
Defined in:
documented/common/sharedbuffer.rb

Overview

Thread-safe circular buffer for sharing lines between reader threads.

Each calling thread maintains its own position in the buffer, allowing independent consumers to read at different speeds. The buffer automatically discards old entries when it exceeds #max_size, and periodically removes index entries for dead threads to prevent memory leaks.

See Also:

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(args = {}) ⇒ void

Creates a new shared buffer.

Parameters:

  • args (Hash) (defaults to: {}) —

    optional configuration

Options Hash (args):

  • :max_size (Integer) — default: 500 —

    maximum number of lines to retain



27
28
29
30
31
32
33
34
35
36
37
 
# File 'documented/common/sharedbuffer.rb', line 27

def initialize(args = {})
  @buffer = Array.new
  @buffer_offset = 0
  @buffer_index = Hash.new
  @buffer_mutex = Mutex.new
  @max_size = args[:max_size] || 500
  # Sweeps dead-thread entries from @buffer_index (keyed by
  # Thread#object_id, previously never pruned) at most once every 60s.
  @cleanup_throttle = Throttle.new(60.0)
  # return self # rubocop does not like this - Lint/ReturnInVoidContext
end

Instance Attribute Details

#max_size ⇒ Object

Returns the value of attribute max_size.



20
21
22
 
# File 'documented/common/sharedbuffer.rb', line 20

def max_size
  @max_size
end

Instance Method Details

#cleanup_threads ⇒ Object

Removes @buffer_index entries whose thread is no longer alive. Snapshots the live thread ids once rather than recomputing them per entry, and holds the mutex so it cannot race with a concurrent reader mutating the hash.



181
182
183
184
185
186
187
 
# File 'documented/common/sharedbuffer.rb', line 181

def cleanup_threads
  @buffer_mutex.synchronize {
    live_ids = Thread.list.map(&:object_id)
    @buffer_index.delete_if { |k, _v| !live_ids.include?(k) }
  }
  return self
end

#clear ⇒ Array<String>

Note:

If the thread's position has fallen behind (line was deleted due to buffer overflow), silently jumps forward to the oldest available line.

Returns all available lines for the calling thread and advances to the end.

Equivalent to calling #gets? repeatedly until nil. On first call, registers the thread and initializes its position to the end of the buffer.

Examples:

# In a reader thread
lines = buffer.clear  #=> ["You say, \"hello\"", "Person says, \"hi\""]

Returns:

  • (Array<String>) —

    all buffered lines not yet read by this thread, or an empty array if none are available



115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
 
# File 'documented/common/sharedbuffer.rb', line 115

def clear
  thread_id = Thread.current.object_id
  if @buffer_index[thread_id].nil?
    @buffer_mutex.synchronize { @buffer_index[thread_id] = (@buffer_offset + @buffer.length) }
    maybe_cleanup_threads
    return Array.new
  end
  if (@buffer_index[thread_id] - @buffer_offset) >= @buffer.length
    return Array.new
  end

  lines = Array.new
  @buffer_mutex.synchronize {
    if @buffer_index[thread_id] < @buffer_offset
      @buffer_index[thread_id] = @buffer_offset
    end
    lines = @buffer[(@buffer_index[thread_id] - @buffer_offset)..-1]
    @buffer_index[thread_id] = (@buffer_offset + @buffer.length)
  }
  return lines
end

#gets ⇒ String?

Note:

If the thread's position has fallen behind (line was deleted due to buffer overflow), silently jumps forward to the oldest available line.

Waits for and returns the next line from the buffer for the calling thread.

Blocks with sleep 0.05s until a line is available. On first call, registers the thread and initializes its position to the end of the buffer.

Examples:

# In a reader thread
line = buffer.gets  #=> "You say, \"hello\""

Returns:

  • (String, nil) —

    the next buffered line, or nil if the buffer is empty



50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
 
# File 'documented/common/sharedbuffer.rb', line 50

def gets
  thread_id = Thread.current.object_id
  if @buffer_index[thread_id].nil?
    @buffer_mutex.synchronize { @buffer_index[thread_id] = (@buffer_offset + @buffer.length) }
    maybe_cleanup_threads
  end
  if (@buffer_index[thread_id] - @buffer_offset) >= @buffer.length
    sleep 0.05 while ((@buffer_index[thread_id] - @buffer_offset) >= @buffer.length)
  end
  line = nil
  @buffer_mutex.synchronize {
    if @buffer_index[thread_id] < @buffer_offset
      @buffer_index[thread_id] = @buffer_offset
    end
    line = @buffer[@buffer_index[thread_id] - @buffer_offset]
  }
  @buffer_index[thread_id] += 1
  return line
end

#gets? ⇒ String?

Note:

If the thread's position has fallen behind (line was deleted due to buffer overflow), silently jumps forward to the oldest available line.

Returns the next line from the buffer for the calling thread, or nil if no line is ready.

Non-blocking variant of #gets: returns immediately. On first call, registers the thread and initializes its position to the end of the buffer.

Examples:

# In a reader thread
line = buffer.gets?  #=> "You say, \"hello\"" or nil

Returns:

  • (String, nil) —

    the next buffered line, or nil if none is available



82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'documented/common/sharedbuffer.rb', line 82

def gets?
  thread_id = Thread.current.object_id
  if @buffer_index[thread_id].nil?
    @buffer_mutex.synchronize { @buffer_index[thread_id] = (@buffer_offset + @buffer.length) }
    maybe_cleanup_threads
  end
  if (@buffer_index[thread_id] - @buffer_offset) >= @buffer.length
    return nil
  end

  line = nil
  @buffer_mutex.synchronize {
    if @buffer_index[thread_id] < @buffer_offset
      @buffer_index[thread_id] = @buffer_offset
    end
    line = @buffer[@buffer_index[thread_id] - @buffer_offset]
  }
  @buffer_index[thread_id] += 1
  return line
end

#rewind ⇒ SharedBuffer

rubocop:disable Lint/HashCompareByIdentity Resets the calling thread's position to the beginning of the buffer.

The next call to #gets, #gets?, or #clear will return the oldest available line. On first call, registers the thread.

Examples:

# In a reader thread, replay the buffer from the start
buffer.rewind.gets  #=> oldest line

Returns:



147
148
149
150
151
152
 
# File 'documented/common/sharedbuffer.rb', line 147

def rewind
  # Hold the mutex: a first-call rewind adds a new key, which must not
  # race a concurrent cleanup_threads delete_if.
  @buffer_mutex.synchronize { @buffer_index[Thread.current.object_id] = @buffer_offset }
  return self
end

#update(line) ⇒ SharedBuffer

rubocop:enable Lint/HashCompareByIdentity Appends a line to the buffer and removes old lines if the buffer exceeds max_size.

The line is frozen to prevent accidental mutation by readers. Old lines are removed from the front (FIFO) as needed.

Examples:

buffer.update("You say, \"hello\"")

Parameters:

  • line (String) —

    the line to append; will be duplicated and frozen

Returns:



164
165
166
167
168
169
170
171
172
173
174
175
 
# File 'documented/common/sharedbuffer.rb', line 164

def update(line)
  @buffer_mutex.synchronize {
    fline = line.dup
    fline.freeze
    @buffer.push(fline)
    while (@buffer.length > @max_size)
      @buffer.shift
      @buffer_offset += 1
    end
  }
  return self
end