Class: Raptor::Reactor

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

Overview

Multiplexes client connections and closes them when they overrun their per-phase timeouts (first data, subsequent chunks, and keep-alive idle). Feeds ready bytes to the parser pipeline and hands completed requests back to the caller-provided handlers.

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(http1_ractor_pool, http2_ractor_pool, thread_pool, connection_options:, http1_options:) ⇒ Reactor

Creates a new Reactor instance.

RBS:

  • (untyped http1_ractor_pool, untyped http2_ractor_pool, untyped thread_pool, connection_options: Hash[Symbol, untyped], http1_options: Hash[Symbol, untyped]) -> void

Parameters:

  • http1_ractor_pool (RactorPool) —

    ractor pool for HTTP/1.x parsing

  • http2_ractor_pool (RactorPool) —

    ractor pool for HTTP/2 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



84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/raptor/reactor.rb', line 84

def initialize(http1_ractor_pool, http2_ractor_pool, thread_pool, connection_options:, http1_options:)
  @http1_ractor_pool = http1_ractor_pool
  @http2_ractor_pool = http2_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.

RBS:

  • (Hash[Symbol, untyped] state) -> void

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



160
161
162
163
164
165
166
167
168
169
170
171
 
# File 'lib/raptor/reactor.rb', line 160

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

#attach_http2(id:, socket:, state:, writer:, flow_control:) ⇒ void

This method returns an undefined value.

Stores an HTTP/2 connection's socket, state, writer, and flow controller in the reactor's per-connection maps.

RBS:

  • (id: Integer, socket: TCPSocket, state: Hash[Symbol, untyped], writer: untyped, flow_control: untyped) -> void

Parameters:

  • id (Integer) —

    unique client identifier

  • socket (TCPSocket) —

    the connection socket

  • state (Hash) —

    initial connection state

  • writer (Http2::Writer) —

    per-connection frame writer

  • flow_control (Http2::FlowControl) —

    per-connection outbound flow controller

  • id: (Integer)
  • socket: (TCPSocket)
  • state: (Hash[Symbol, untyped])
  • writer: (Object)
  • flow_control: (Object) 


277
278
279
280
281
282
 
# File 'lib/raptor/reactor.rb', line 277

def attach_http2(id:, socket:, state:, writer:, flow_control:)
  @id_to_socket[id] = socket
  @socket_to_state[socket] = state
  @id_to_writer[id] = writer
  @id_to_flow_control[id] = flow_control
end

#backlog ⇒ Integer

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

RBS:

  • () -> Integer

Returns:

  • (Integer) —

    number of complete requests



352
353
354
 
# File 'lib/raptor/reactor.rb', line 352

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.

RBS:

  • (TCPSocket socket) -> void

Parameters:

  • socket (TCPSocket) —

    the socket to clean up



433
434
435
436
437
438
439
 
# File 'lib/raptor/reactor.rb', line 433

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.

RBS:

  • (Integer id) -> void

Parameters:

  • id (Integer) —

    unique client identifier



325
326
327
328
329
330
331
332
333
 
# File 'lib/raptor/reactor.rb', line 325

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

Returns true when the request has been fully parsed.

RBS:

  • (Hash[Symbol, untyped] state) -> bool

Parameters:

  • state (Hash) —

    connection state

Returns:

  • (Boolean) —

    true if the request is complete



447
448
449
 
# File 'lib/raptor/reactor.rb', line 447

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

#first_data_received?(state) ⇒ Boolean

Checks if any data has been received for this connection.

RBS:

  • (Hash[Symbol, untyped] state) -> bool

Parameters:

  • state (Hash) —

    connection state

Returns:

  • (Boolean) —

    true if first data has been received



457
458
459
 
# File 'lib/raptor/reactor.rb', line 457

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.

RBS:

  • (Integer id) -> untyped?

Parameters:

  • id (Integer) —

    unique client identifier

Returns:

  • (Object, nil) —

    the flow controller, if found



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

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 under the persistent-data timeout.

RBS:

  • (TCPSocket socket, Integer id, Integer request_count, remote_addr: String, url_scheme: String) -> void

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) 


217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
 
# File 'lib/raptor/reactor.rb', line 217

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 returns it to the selector.

RBS:

  • (TCPSocket socket, Hash[Symbol, untyped] state) -> Hash[Symbol, untyped]?

Parameters:

  • socket (TCPSocket) —

    the socket to read from and queue

  • state (Hash) —

    current connection state

Returns:

  • (Hash, nil) —

    updated state, if successful



403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
 
# File 'lib/raptor/reactor.rb', line 403

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.positive?
    buffer << socket.read_nonblock(socket.pending)
  end

  state = state.frozen? ? state.merge(buffer: buffer) : state.merge!(buffer: buffer)
  pool = state[:protocol] == :http2 ? @http2_ractor_pool : @http1_ractor_pool
  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.

RBS:

  • (TCPSocket socket) -> void

Parameters:

  • socket (TCPSocket) —

    the socket to register



364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
 
# File 'lib/raptor/reactor.rb', line 364

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.

RBS:

  • (Integer id) -> TCPSocket?

Parameters:

  • id (Integer) —

    unique client identifier

Returns:

  • (TCPSocket, nil) —

    the socket associated with id, if any



200
201
202
203
204
 
# File 'lib/raptor/reactor.rb', line 200

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. Runs until the registration queue is closed and drained.

RBS:

  • () -> Thread

Returns:

  • (Thread) —

    the thread running the reactor event loop



109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
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
 
# File 'lib/raptor/reactor.rb', line 109

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).zero?

          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.

RBS:

  • () -> void



341
342
343
344
 
# File 'lib/raptor/reactor.rb', line 341

def shutdown
  @queue.close
  @selector.wakeup
end

#socket_for(id) ⇒ TCPSocket?

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

RBS:

  • (Integer id) -> TCPSocket?

Parameters:

  • id (Integer) —

    unique client identifier

Returns:

  • (TCPSocket, nil) —

    the socket, if found



240
241
242
 
# File 'lib/raptor/reactor.rb', line 240

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 and re-registers the socket for further reads.

RBS:

  • (Hash[Symbol, untyped] state) -> void

Parameters:

  • state (Hash) —

    updated connection state from the ractor pool



307
308
309
310
311
312
313
314
315
316
 
# File 'lib/raptor/reactor.rb', line 307

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 and re-registers it for further I/O.

RBS:

  • (Hash[Symbol, untyped] state) -> void

Parameters:

  • state (Hash) —

    updated client connection state

Options Hash (state):

  • :id (Integer) —

    client identifier



181
182
183
184
185
186
187
188
189
190
 
# File 'lib/raptor/reactor.rb', line 181

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.

RBS:

  • (TCPSocket socket) -> void

Parameters:

  • socket (TCPSocket) —

    the socket that became ready



387
388
389
390
391
392
393
 
# File 'lib/raptor/reactor.rb', line 387

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

#watch(id) ⇒ void

This method returns an undefined value.

Registers an attached socket for future reactor-driven reads.

RBS:

  • (Integer id) -> void

Parameters:

  • id (Integer) —

    unique client identifier



290
291
292
293
294
295
296
297
298
 
# File 'lib/raptor/reactor.rb', line 290

def watch(id)
  socket = @id_to_socket[id]
  return unless socket

  @queue << socket
  @selector.wakeup
rescue ClosedQueueError
  socket.close
end

#writer_for(id) ⇒ Object?

Returns the writer object associated with a given connection, if one was supplied when the connection was added.

RBS:

  • (Integer id) -> untyped?

Parameters:

  • id (Integer) —

    unique client identifier

Returns:

  • (Object, nil) —

    the writer, if found



251
252
253
 
# File 'lib/raptor/reactor.rb', line 251

def writer_for(id)
  @id_to_writer[id]
end