Class: Raptor::Reactor

Inherits:
Object
  • Object
show all
Defined in:
lib/raptor/reactor.rb,
sig/generated/raptor/reactor.rbs

Overview

High-performance I/O reactor for managing client connections and timeouts.

Reactor uses NIO selectors for efficient I/O multiplexing and implements client timeouts using a red-black tree for O(log n) timeout management. It coordinates between ractor pools for CPU-intensive HTTP parsing and thread pools for blocking operations, and provides backlog metrics that the server uses for backpressure control to prevent overload.

Examples:

reactor = Reactor.new(
  ractor_pool,
  thread_pool,
  connection_options: { first_data_timeout: 30, chunk_data_timeout: 10 },
  http1_options: { persistent_data_timeout: 65 }
)
reactor.run
reactor.add(id: client.object_id, socket: client)
# ... later
reactor.shutdown

Defined Under Namespace

Classes: TimeoutClient

Constant Summary collapse

CHUNK_SIZE =

Returns:

  • (Object) 
64 * 1024
TIMEOUT_RESPONSE =

Returns:

  • (::String) 
"HTTP/1.1 408 Request Timeout\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"

Instance Method Summary collapse

Constructor Details

#initialize(ractor_pool, thread_pool, connection_options:, http1_options:) ⇒ Reactor

Creates a new Reactor instance.

Parameters:

  • ractor_pool (RactorPool) —

    ractor pool for HTTP parsing

  • thread_pool (AtomicThreadPool) —

    thread pool for application processing

  • connection_options (Hash) —

    per-connection timeout configuration

  • http1_options (Hash) —

    HTTP/1.1-specific configuration

  • connection_options: (Hash[Symbol, untyped])
  • http1_options: (Hash[Symbol, untyped])

Options Hash (connection_options:):

  • :first_data_timeout (Integer) —

    timeout for initial data

  • :chunk_data_timeout (Integer) —

    timeout for subsequent chunks

Options Hash (http1_options:):

  • :persistent_data_timeout (Integer) —

    timeout for keep-alive idle connections



97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
 
# File 'lib/raptor/reactor.rb', line 97

def initialize(ractor_pool, thread_pool, connection_options:, http1_options:)
  @ractor_pool = ractor_pool
  @thread_pool = thread_pool
  @first_data_timeout = connection_options[:first_data_timeout]
  @chunk_data_timeout = connection_options[:chunk_data_timeout]
  @persistent_data_timeout = http1_options[:persistent_data_timeout]

  @selector = NIO::Selector.new
  @queue = Queue.new
  @timeouts = RedBlackTree.new

  @id_to_socket = {}
  @socket_to_state = {}
  @id_to_timeout = {}
  @id_to_writer = {}
  @id_to_flow_control = {}
end

Instance Method Details

#add(state) ⇒ void

This method returns an undefined value.

Adds a new client connection to the reactor.

Parameters:

  • state (Hash) —

    client connection state including socket and ID

Options Hash (state):

  • :socket (TCPSocket) —

    the client socket

  • :id (Integer) —

    unique identifier for the client



175
176
177
178
179
180
181
182
183
184
185
186
 
# File 'lib/raptor/reactor.rb', line 175

def add(state)
  socket = state[:socket]
  state.delete(:socket)
  writer = state.delete(:writer)
  flow_control = state.delete(:flow_control)
  @id_to_socket[state[:id]] = socket
  @socket_to_state[socket] = state
  @id_to_writer[state[:id]] = writer if writer
  @id_to_flow_control[state[:id]] = flow_control if flow_control

  read_and_queue_for_parse(socket, state)
end

#backlog ⇒ Integer

Returns the number of complete requests either being processed or awaiting processing.

Returns:

  • (Integer) —

    number of complete requests



346
347
348
 
# File 'lib/raptor/reactor.rb', line 346

def backlog
  @thread_pool.queue_size + @thread_pool.active_count
end

#cleanup(socket) ⇒ void

This method returns an undefined value.

Cleans up a client connection by removing it from tracking and closing the socket.

Parameters:

  • socket (TCPSocket) —

    the socket to clean up



426
427
428
429
430
431
432
 
# File 'lib/raptor/reactor.rb', line 426

def cleanup(socket)
  state = @socket_to_state.delete(socket)
  @id_to_socket.delete(state[:id])
  @id_to_writer.delete(state[:id])
  @id_to_flow_control.delete(state[:id])
  socket.close
end

#close_connection(id) ⇒ void

This method returns an undefined value.

Closes the socket for the given connection and drops all reactor state associated with it. Used to terminate HTTP/2 connections after sending a GOAWAY frame.

Parameters:

  • id (Integer) —

    unique client identifier



319
320
321
322
323
324
325
326
327
 
# File 'lib/raptor/reactor.rb', line 319

def close_connection(id)
  socket = @id_to_socket.delete(id)
  return unless socket

  @socket_to_state.delete(socket)
  @id_to_writer.delete(id)
  @id_to_flow_control.delete(id)
  socket.close rescue nil
end

#complete?(state) ⇒ Boolean

Checks if a request is complete i.e., processable.

Parameters:

  • state (Hash) —

    connection state

Returns:

  • (Boolean) —

    true if the request is complete



440
441
442
 
# File 'lib/raptor/reactor.rb', line 440

def complete?(state)
  state[:complete]
end

#first_data_received?(state) ⇒ Boolean

Checks if any data has been received for this connection.

Parameters:

  • state (Hash) —

    connection state

Returns:

  • (Boolean) —

    true if first data has been received



450
451
452
 
# File 'lib/raptor/reactor.rb', line 450

def first_data_received?(state)
  complete?(state) || (state.dig(:parse_data, :parse_count) || 0) >= 1
end

#flow_control_for(id) ⇒ Object?

Returns the flow controller associated with a given connection, if one was supplied when the connection was added. Used by HTTP/2 stream dispatchers to honour the peer's flow-control windows.

Parameters:

  • id (Integer) —

    unique client identifier

Returns:

  • (Object, nil) —

    the flow controller, if found



287
288
289
 
# File 'lib/raptor/reactor.rb', line 287

def flow_control_for(id)
  @id_to_flow_control[id]
end

#persist(socket, id, request_count, remote_addr:, url_scheme:) ⇒ void

This method returns an undefined value.

Re-registers a kept-alive connection for the next request cycle.

Called after successfully writing a response when keep-alive is active. Resets the connection state and re-queues the socket in the selector using the persistent data timeout.

Parameters:

  • socket (TCPSocket) —

    the kept-alive client socket

  • id (Integer) —

    the unique client identifier

  • request_count (Integer) —

    number of requests handled on this connection

  • remote_addr (String) —

    the client's remote IP address

  • url_scheme (String) —

    "http" or "https"

  • remote_addr: (String)
  • url_scheme: (String) 


237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
 
# File 'lib/raptor/reactor.rb', line 237

def persist(socket, id, request_count, remote_addr:, url_scheme:)
  state = {
    id: id,
    request_count: request_count,
    remote_addr: remote_addr,
    url_scheme: url_scheme,
    persisted: true
  }

  @id_to_socket[id] = socket
  @socket_to_state[socket] = state
  @queue << socket
  @selector.wakeup
rescue ClosedQueueError
  socket.close
end

#read_and_queue_for_parse(socket, state) ⇒ Hash?

Reads data from a socket and either queues it for parsing, or for selector registration.

Parameters:

  • socket (TCPSocket) —

    the socket to read from and queue

  • state (Hash) —

    current connection state

Returns:

  • (Hash, nil) —

    updated state, if successful



397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
 
# File 'lib/raptor/reactor.rb', line 397

def read_and_queue_for_parse(socket, state)
  data = begin
    socket.read_nonblock(CHUNK_SIZE)
  rescue IO::WaitReadable
    @queue << socket
    @selector.wakeup
    return
  rescue EOFError
    cleanup(socket)
    return
  end

  buffer = state[:buffer] ? state[:buffer].dup : String.new
  buffer << data

  while socket.respond_to?(:pending) && socket.pending > 0
    buffer << socket.read_nonblock(socket.pending)
  end

  state = state.frozen? ? state.merge(buffer: buffer) : state.merge!(buffer: buffer)
  @ractor_pool << Ractor.make_shareable(state)
end

#register(socket) ⇒ void

This method returns an undefined value.

Registers a socket with the NIO selector and sets up timeout tracking.

Parameters:

  • socket (TCPSocket) —

    the socket to register



358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
 
# File 'lib/raptor/reactor.rb', line 358

def register(socket)
  @selector.register(socket, :r).value = socket

  state = @socket_to_state[socket]
  client = TimeoutClient.new(state)
  timeout = if state[:persisted]
    @persistent_data_timeout
  elsif first_data_received?(state)
    @chunk_data_timeout
  else
    @first_data_timeout
  end
  client.timeout_at = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout
  @timeouts << client
  @id_to_timeout[state[:id]] = client
end

#remove(id) ⇒ TCPSocket?

Drops the reactor's references to a client whose parsed request has been handed off to the thread pool. The socket itself is kept open so the worker can write the response.

Parameters:

  • id (Integer) —

    unique client identifier

Returns:

  • (TCPSocket, nil) —

    the socket associated with id, if any



217
218
219
220
221
 
# File 'lib/raptor/reactor.rb', line 217

def remove(id)
  @id_to_socket.delete(id).tap do |socket|
    @socket_to_state.delete(socket)
  end
end

#run ⇒ Thread

Starts the reactor's main event loop in a new thread.

The event loop handles I/O events, processes timeouts, manages the registration queue, and controls server connection acceptance. It continues until the queue is closed and emptied.

Returns:

  • (Thread) —

    the thread running the reactor event loop



124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
 
# File 'lib/raptor/reactor.rb', line 124

def run
  Thread.new do
    Thread.current.name = "Reactor"

    until @queue.closed? && @queue.empty?
      begin
        timeout = @timeouts.min&.timeout(Process.clock_gettime(Process::CLOCK_MONOTONIC))
        @selector.select(timeout) do |monitor|
          wakeup!(monitor.value)
        end

        now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
        expired = []
        @timeouts.traverse do |to_client|
          break unless to_client.timeout(now) == 0

          expired << to_client
        end

        expired.each do |to_client|
          @timeouts.delete!(to_client)
          id = to_client.client_data[:id]
          @id_to_timeout.delete(id)
          socket = @id_to_socket[id]
          next unless socket

          @selector.deregister(socket)
          socket.write(TIMEOUT_RESPONSE) rescue nil
          cleanup(socket)
        end

        until @queue.empty?
          register(@queue.pop)
        end
      rescue => error
        Log.rescued_error(error)
      end
    end

    @selector.close
  end
end

#shutdown ⇒ void

This method returns an undefined value.

Closes the registration queue and wakes the selector so the event loop drains pending work and exits.



335
336
337
338
 
# File 'lib/raptor/reactor.rb', line 335

def shutdown
  @queue.close
  @selector.wakeup
end

#socket_for(id) ⇒ TCPSocket?

Returns the socket for a given client identifier without removing it.

Used by HTTP/2 connections where the socket remains registered across multiple stream requests.

Parameters:

  • id (Integer) —

    unique client identifier

Returns:

  • (TCPSocket, nil) —

    the socket, if found



263
264
265
 
# File 'lib/raptor/reactor.rb', line 263

def socket_for(id)
  @id_to_socket[id]
end

#update_http2_state(state) ⇒ void

This method returns an undefined value.

Updates connection state for an HTTP/2 connection after frame processing.

Re-registers the socket with the selector for further reads and stores the updated HPACK table and stream states.

Parameters:

  • state (Hash) —

    updated connection state from the ractor pool



300
301
302
303
304
305
306
307
308
309
 
# File 'lib/raptor/reactor.rb', line 300

def update_http2_state(state)
  socket = @id_to_socket[state[:id]]
  return unless socket

  @socket_to_state[socket] = state
  @queue << socket
  @selector.wakeup
rescue ClosedQueueError
  socket.close
end

#update_state(state) ⇒ void

This method returns an undefined value.

Updates the state of an existing client connection.

Called when an incomplete HTTP request needs to be re-registered with the reactor for further processing.

Parameters:

  • state (Hash) —

    updated client connection state

Options Hash (state):

  • :id (Integer) —

    client identifier



198
199
200
201
202
203
204
205
206
207
 
# File 'lib/raptor/reactor.rb', line 198

def update_state(state)
  socket = @id_to_socket[state[:id]]
  return unless socket

  @socket_to_state[socket] = state
  @queue << socket
  @selector.wakeup
rescue ClosedQueueError
  socket.close
end

#wakeup!(socket) ⇒ void

This method returns an undefined value.

Handles socket wakeup by deregistering and queuing for processing.

Parameters:

  • socket (TCPSocket) —

    the socket that became ready



381
382
383
384
385
386
387
 
# File 'lib/raptor/reactor.rb', line 381

def wakeup!(socket)
  @selector.deregister(socket)
  state = @socket_to_state[socket]
  to_client = @id_to_timeout.delete(state[:id])
  @timeouts.delete!(to_client)
  read_and_queue_for_parse(socket, state)
end

#writer_for(id) ⇒ Object?

Returns the writer object associated with a given connection, if one was supplied when the connection was added. Used by protocol handlers that need to coordinate concurrent socket writes.

Parameters:

  • id (Integer) —

    unique client identifier

Returns:

  • (Object, nil) —

    the writer, if found



275
276
277
 
# File 'lib/raptor/reactor.rb', line 275

def writer_for(id)
  @id_to_writer[id]
end