Class: Lich::Common::SynchronizedSocket

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

Overview

Thread-safe socket wrapper with write-side resilience.

Wraps a delegate socket (typically TCPSocket) with a mutex to serialize writes and an internal liveness flag that transitions irreversibly to dead on any fatal write error.

When a fatal write error occurs the wrapper:

  1. Sets @alive to false
  2. Closes the delegate socket (unblocking any blocked readers)
  3. Logs the error via Lich.log

A WriteQueueOverflow additionally dumps every still-pending write to Lich.log, so a stalled frontend can be diagnosed from the queue contents rather than from the summary line alone.

When the pending-write budget is exhausted the queue is first compacted by discarding the oldest prompt-delimited groups; the session only ends if compaction cannot free space. Compaction drops display data the parser has already consumed, so a slow frontend costs a gap in its scrollback rather than the whole session.

Subsequent writes short-circuit without touching the delegate. Reads still delegate via method_missing so lifecycle threads can detect the closed socket through normal IOError propagation.

Frontend writes are queued through one socket-local writer thread so a slow frontend cannot block the game parser. Main-stream output is deferred while a frontend stream is open and released after the matching popStream or prompt has entered the queue.

Defined Under Namespace

Classes: WriteQueueOverflow

Constant Summary collapse

FATAL_WRITE_ERRORS =

Errors that indicate a permanently broken write path.

[
  Errno::ECONNRESET,
  Errno::EPIPE,
  Errno::ECONNABORTED,
  Errno::ENOTCONN,
  IOError,
].freeze
WRITER_INIT_MUTEX =

Mutex guarding lazy initialization of the writer thread and associated state.

Synchronizes access to @write_queue, @stream_mutex, @stream_stack, @deferred_main_stream, and @writer_thread to prevent race conditions when the first write arrives and those structures are created.

Mutex.new
DEFAULT_WRITE_QUEUE_CAPACITY = 
4_096
OVERFLOW_KEEP_PROMPT_GROUPS =

Number of trailing prompt-delimited groups retained when the write queue is compacted. A <prompt> clears the stream stack, so a group boundary is a point where stream nesting is balanced and older output can be discarded without orphaning a pushStream in the frontend.

10
DROP_PREAMBLE_SIZE =

Items injected ahead of the retained groups on compaction: the replayed resync prompt and the client-visible notice. Kept in sync with drop_preamble, which never returns more than this many items.

2
PROMPT_TAG =

Pattern matching the opening tag of a prompt element in game server XML output.

A prompt clears the stream stack, making it a safe boundary point for discarding older display data during write-queue compaction.

Examples:

Matching a prompt tag

text = "<prompt time='1234567890'>Some&gt; </prompt>"
SynchronizedSocket::PROMPT_TAG.match?(text) #=> true

See Also:

  • #note_stream_xml!
/<prompt\b/i
OVERFLOW_DUMP_MAX_BYTES_PER_ARG =

Per-argument byte cap applied when dumping pending writes after an overflow. Longer payloads are truncated with their full byte size noted.

1_024
OVERFLOW_DUMP_EDGE_ENTRIES =

Leading and trailing entries emitted per section by the overflow dump. Bounds a single log record at roughly 400 entries per section instead of the full queue capacity.

200
ROLES =

Valid roles for a SynchronizedSocket.

primary

A fatal write error ends the session immediately. Used for the main frontend connection.

detachable

A fatal write error closes the connection but does not end the session. Used for secondary connections that can be reconnected without restarting the game.

See Also:

%i[primary detachable].freeze
ATTACHMENT_STREAM_SENTINEL =

Sentinel marking the boundary between a detachable socket's pre-attachment and post-attachment stream stack entries.

When a detachable connection is attached, this sentinel is pushed onto the stream stack so that any popStream before the first pushStream after attachment cannot accidentally pop the initial sentinel, leaving the stack corrupted.

Object.new.freeze

Instance Method Summary collapse

Constructor Details

#initialize(delegate, role: :primary, write_queue_capacity: DEFAULT_WRITE_QUEUE_CAPACITY) ⇒ SynchronizedSocket

Returns a new instance of SynchronizedSocket.

Parameters:

  • delegate (#puts, #write, #gets, #close) —

    the underlying socket

  • role (Symbol) (defaults to: :primary) —

    whether failure should end the session or only the detachable connection

  • write_queue_capacity (Integer) (defaults to: DEFAULT_WRITE_QUEUE_CAPACITY) —

    maximum pending writes

Raises:

  • (ArgumentError) 


111
112
113
114
115
116
117
118
119
120
121
122
123
 
# File 'documented/common/class_exts/synchronizedsocket.rb', line 111

def initialize(delegate, role: :primary, write_queue_capacity: DEFAULT_WRITE_QUEUE_CAPACITY)
  raise ArgumentError, "unknown socket role: #{role.inspect}" unless ROLES.include?(role)
  unless write_queue_capacity.is_a?(Integer) && write_queue_capacity.positive?
    raise ArgumentError, 'write_queue_capacity must be a positive Integer'
  end

  @delegate = delegate
  @mutex = Mutex.new
  @state_mutex = Mutex.new
  @alive = true
  @role = role
  @write_queue_capacity = write_queue_capacity
end

Dynamic Method Handling

This class handles dynamic methods through the method_missing method

#method_missing(method, *args, &block) ⇒ Object

Delegates all non-write methods to the underlying socket.

Read-side errors propagate normally so that connection lifecycle threads can detect disconnects.



194
195
196
 
# File 'documented/common/class_exts/synchronizedsocket.rb', line 194

def method_missing(method, *args, &block)
  @delegate.__send__(method, *args, &block)
end

Instance Method Details

#alive? ⇒ Boolean

Whether the socket is usable for I/O.

Returns false once a fatal write error has occurred or the delegate has been closed by any means. This transition is one-way -- a dead socket cannot be revived.

Returns:

  • (Boolean) 


132
133
134
 
# File 'documented/common/class_exts/synchronizedsocket.rb', line 132

def alive?
  @alive && !@delegate.closed?
end

#close(*args, &block) ⇒ Object

Marks the socket dead, closes the delegate, and stops the writer.



182
183
184
185
186
187
188
 
# File 'documented/common/class_exts/synchronizedsocket.rb', line 182

def close(*args, &block)
  @alive = false
  @delegate.close(*args, &block) unless @delegate.closed?
  @write_queue.push([:stop, [], nil], true) if @write_queue.is_a?(Queue)
rescue ThreadError
  nil
end

#puts(*args, &block) ⇒ nil

Queue a newline-terminated write for the writer thread.

Returns:

  • (nil) —

    the write happens asynchronously



139
140
141
 
# File 'documented/common/class_exts/synchronizedsocket.rb', line 139

def puts(*args, &block)
  enqueue_write(:puts, args, block)
end

#puts_if(*args) ⇒ Object

Compatibility alias for scripts using the historical name. Blocks were never part of the asynchronous contract and are intentionally ignored.



170
171
172
 
# File 'documented/common/class_exts/synchronizedsocket.rb', line 170

def puts_if(*args)
  puts_main_stream(*args)
end

#puts_main_stream(*args) ⇒ Boolean

Queue output for the next main-stream opportunity.

Returns:

  • (Boolean) —

    true if accepted, false if the socket is dead



146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
 
# File 'documented/common/class_exts/synchronizedsocket.rb', line 146

def puts_main_stream(*args)
  return false unless @alive

  ensure_writer!
  queued_args = copy_args(args)

  @stream_mutex.synchronize do
    return false unless alive?

    if stream_open?
      ensure_pending_capacity!
      @deferred_main_stream << queued_args
    else
      queue_puts_locked(queued_args)
    end
  end
  true
rescue StandardError => e
  handle_enqueue_error('puts_main_stream', e)
  false
end

#respond_to_missing?(method, include_private = false) ⇒ Boolean

Returns whether the delegate responds to method.

Returns:

  • (Boolean) —

    whether the delegate responds to method



199
200
201
 
# File 'documented/common/class_exts/synchronizedsocket.rb', line 199

def respond_to_missing?(method, include_private = false)
  @delegate.respond_to?(method, include_private) || super
end

#write(*args, &block) ⇒ nil

Queue a raw write for the writer thread.

Returns:

  • (nil) —

    the write happens asynchronously



177
178
179
 
# File 'documented/common/class_exts/synchronizedsocket.rb', line 177

def write(*args, &block)
  enqueue_write(:write, args, block)
end