Module: Lich::Common::Buffer

Defined in:
documented/common/buffer.rb

Overview

Thread-safe circular buffer for game server output and script input streams.

Buffer maintains a fixed-size ring of lines, each tagged with a stream identifier that indicates the source (game server downstream output, upstream commands, script output, etc.). Each thread maintains its own read cursor and stream filter, allowing independent consumption of the shared buffer while preserving thread safety via mutex.

Threads are registered automatically on their first call to .gets, .gets?, .rewind, or .clear; stale entries are cleaned up automatically via throttled sweeps to prevent unbounded memory growth in long-running sessions.

Constant Summary collapse

DOWNSTREAM_STRIPPED = 
1
DOWNSTREAM_RAW = 
2
DOWNSTREAM_MOD = 
4
UPSTREAM = 
8
UPSTREAM_MOD = 
16
SCRIPT_OUTPUT = 
32
@@index = 
Hash.new
@@streams = 
Hash.new
@@mutex = 
Mutex.new
@@offset = 
0
@@buffer = 
Array.new
@@max_size = 
3000
@@cleanup_throttle =

@@index / @@streams are keyed by Thread#object_id and were never pruned, so every thread that ever read the buffer leaked an entry that outlived it. maybe_cleanup sweeps dead-thread entries from the registration path, at most once every 60s via this throttle.

Throttle.new(60.0)

Class Method Summary collapse

Class Method Details

.cleanup ⇒ Object

rubocop:enable Lint/HashCompareByIdentity Removes @@index / @@streams 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 same hashes.



260
261
262
263
264
265
266
267
 
# File 'documented/common/buffer.rb', line 260

def self.cleanup
  @@mutex.synchronize {
    live_ids = Thread.list.map(&:object_id)
    @@index.delete_if { |k, _v| !live_ids.include?(k) }
    @@streams.delete_if { |k, _v| !live_ids.include?(k) }
  }
  return self
end

.clear ⇒ Array<Object>

Reads all remaining lines from the buffer for the calling thread, returning them as an array and advancing the cursor to the end.

Filters the buffer by the calling thread's current stream mask, returning only lines that match. If the thread's cursor has fallen behind the buffer (e.g., due to buffer rotation), it is advanced to the oldest line. Registers the thread on first call with a default stream filter of DOWNSTREAM_STRIPPED.

Examples:

lines = Buffer.clear
# => [line1, line2, line3]

Returns:

  • (Array<Object>) —

    an array of line objects matching the calling thread's stream filter, or an empty array if no matching lines remain



152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
 
# File 'documented/common/buffer.rb', line 152

def self.clear
  thread_id = Thread.current.object_id
  if @@index[thread_id].nil?
    @@mutex.synchronize {
      @@index[thread_id] = (@@offset + @@buffer.length)
      @@streams[thread_id] ||= DOWNSTREAM_STRIPPED
    }
    maybe_cleanup
  end
  lines = Array.new
  loop {
    if (@@index[thread_id] - @@offset) >= @@buffer.length
      return lines
    end

    line = nil
    @@mutex.synchronize {
      if @@index[thread_id] < @@offset
        @@index[thread_id] = @@offset
      end
      line = @@buffer[@@index[thread_id] - @@offset]
    }
    @@index[thread_id] += 1
    lines.push(line) if ((line.stream & @@streams[thread_id]) != 0)
  }
  return lines
end

.gets ⇒ Object

Reads the next line from the buffer for the calling thread, blocking until one is available.

Filters the buffer by the calling thread's current stream mask, skipping lines that do not match. If the thread's cursor has fallen behind the buffer (e.g., due to buffer rotation), it is advanced to the oldest line. Registers the thread on first call with a default stream filter of DOWNSTREAM_STRIPPED.

Examples:

line = Buffer.gets
# => line.stream => 1, line => "You scan the area...\n"

Returns:

  • (Object) —

    a line object with .stream and text attributes



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
 
# File 'documented/common/buffer.rb', line 51

def self.gets
  thread_id = Thread.current.object_id
  if @@index[thread_id].nil?
    @@mutex.synchronize {
      @@index[thread_id] = (@@offset + @@buffer.length)
      @@streams[thread_id] ||= DOWNSTREAM_STRIPPED
    }
    maybe_cleanup
  end
  line = nil
  loop {
    if (@@index[thread_id] - @@offset) >= @@buffer.length
      sleep 0.05 while ((@@index[thread_id] - @@offset) >= @@buffer.length)
    end
    @@mutex.synchronize {
      if @@index[thread_id] < @@offset
        @@index[thread_id] = @@offset
      end
      line = @@buffer[@@index[thread_id] - @@offset]
    }
    @@index[thread_id] += 1
    break if ((line.stream & @@streams[thread_id]) != 0)
  }
  return line
end

.gets? ⇒ Object?

Reads the next line from the buffer for the calling thread, returning nil if no matching line is available instead of blocking.

Filters the buffer by the calling thread's current stream mask, skipping lines that do not match. If the thread's cursor has fallen behind the buffer (e.g., due to buffer rotation), it is advanced to the oldest line. Registers the thread on first call with a default stream filter of DOWNSTREAM_STRIPPED.

Examples:

line = Buffer.gets?
# => line.stream => 1, line => "You scan the area...\n" OR nil

Returns:

  • (Object, nil) —

    a line object with .stream and text attributes, or nil if no matching line is available



90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
 
# File 'documented/common/buffer.rb', line 90

def self.gets?
  thread_id = Thread.current.object_id
  if @@index[thread_id].nil?
    @@mutex.synchronize {
      @@index[thread_id] = (@@offset + @@buffer.length)
      @@streams[thread_id] ||= DOWNSTREAM_STRIPPED
    }
    maybe_cleanup
  end
  line = nil
  loop {
    if (@@index[thread_id] - @@offset) >= @@buffer.length
      return nil
    end

    @@mutex.synchronize {
      if @@index[thread_id] < @@offset
        @@index[thread_id] = @@offset
      end
      line = @@buffer[@@index[thread_id] - @@offset]
    }
    @@index[thread_id] += 1
    break if ((line.stream & @@streams[thread_id]) != 0)
  }
  return line
end

.rewind ⇒ self

Resets the calling thread's read cursor to the oldest line in the buffer.

Registers the thread with a default stream filter of DOWNSTREAM_STRIPPED if it has not yet accessed the buffer. Acquires the mutex to prevent race conditions during thread registration and cleanup.

Examples:

Buffer.rewind
first_line = Buffer.gets

Returns:

  • (self) —

    the Buffer module



127
128
129
130
131
132
133
134
135
136
137
 
# File 'documented/common/buffer.rb', line 127

def self.rewind
  thread_id = Thread.current.object_id
  # Hold the mutex: for a thread whose first Buffer call is rewind this
  # adds new keys, which must not race a concurrent cleanup delete_if
  # (Ruby raises on a key added during iteration).
  @@mutex.synchronize {
    @@index[thread_id] = @@offset
    @@streams[thread_id] ||= DOWNSTREAM_STRIPPED
  }
  return self
end

.streams ⇒ Integer?

rubocop:disable Lint/HashCompareByIdentity Returns the stream filter bitmask for the calling thread.

The result is a bitmask composed of stream constants (DOWNSTREAM_STRIPPED, DOWNSTREAM_RAW, etc.) that determines which lines the thread will receive from .gets, .gets?, and .clear. Returns nil if the thread has never accessed the buffer.

Examples:

Buffer.streams #=> 1  (DOWNSTREAM_STRIPPED)
Buffer.streams = Buffer::UPSTREAM
Buffer.streams #=> 8

Returns:

  • (Integer, nil) —

    the stream filter bitmask, or nil if the thread is not registered



226
227
228
 
# File 'documented/common/buffer.rb', line 226

def self.streams
  @@mutex.synchronize { @@streams[Thread.current.object_id] }
end

.streams=(val) ⇒ void

This method returns an undefined value.

Sets the stream filter bitmask for the calling thread.

The bitmask must be a non-zero Integer with at least one of the six stream bits set (bits 0-5 corresponding to the six stream constants). Acquires the mutex to prevent race conditions during thread registration and cleanup. Raises no exception on invalid input; instead logs an error message via respond and returns nil.

Examples:

Buffer.streams = Buffer::DOWNSTREAM_STRIPPED | Buffer::UPSTREAM
Buffer.streams #=> 9

Parameters:

  • val (Integer) —

    a non-zero bitmask composed of stream constants (DOWNSTREAM_STRIPPED, DOWNSTREAM_RAW, DOWNSTREAM_MOD, UPSTREAM, UPSTREAM_MOD, SCRIPT_OUTPUT)



244
245
246
247
248
249
250
251
252
253
 
# File 'documented/common/buffer.rb', line 244

def self.streams=(val)
  if (!val.is_a?(Integer)) or ((val & 63) == 0)
    respond "--- Lich: error: invalid streams value\n\t#{$!.caller[0..2].join("\n\t")}"
  else
    # Hold the mutex: setting streams for a thread that has not registered
    # yet adds a new key, which must not race the cleanup delete_if sweep
    # (Ruby raises on a key added during iteration).
    @@mutex.synchronize { @@streams[Thread.current.object_id] = val }
  end
end

.update(line, stream = nil) ⇒ self

Appends a line to the buffer, optionally overriding its stream identifier.

The line is duplicated and frozen before insertion to prevent external mutation. If the buffer exceeds the maximum size (3000 lines), the oldest line is discarded and the offset is incremented. Acquires the mutex for thread safety.

Examples:

line = Object.new
line.stream = Buffer::DOWNSTREAM_STRIPPED
Buffer.update(line)
# or
Buffer.update(line, Buffer::UPSTREAM)

Parameters:

  • line (Object) —

    a line object with text and optional .stream attribute

  • stream (Integer, nil) (defaults to: nil) —

    optional stream identifier (a bitmask constant like DOWNSTREAM_STRIPPED) to set on the frozen copy; if nil, the line's existing stream is preserved

Returns:

  • (self) —

    the Buffer module



197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
 
# File 'documented/common/buffer.rb', line 197

def self.update(line, stream = nil)
  @@mutex.synchronize {
    frozen_line = line.dup
    unless stream.nil?
      frozen_line.stream = stream
    end
    frozen_line.freeze
    @@buffer.push(frozen_line)
    while (@@buffer.length > @@max_size)
      @@buffer.shift
      @@offset += 1
    end
  }
  return self
end