Class: Raptor::Reactor

Inherits:
Object
  • Object
show all
Defined in:
lib/raptor/reactor.rb

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 thread pools for blocking operations and ractor pools for CPU-intensive HTTP parsing, and provides backlog metrics that the server uses for backpressure control to prevent overload.

Examples:

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

Defined Under Namespace

Classes: TimeoutClient

Constant Summary collapse

CHUNK_SIZE = 
64 * 1024
TIMEOUT_RESPONSE = 
"HTTP/1.1 408 Request Timeout\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"

Instance Method Summary collapse

Constructor Details

#initialize(thread_pool, ractor_pool, client_options:) ⇒ void

Creates a new Reactor instance.

Parameters:

  • thread_pool (AtomicThreadPool) —

    thread pool for application processing

  • ractor_pool (RactorPool) —

    ractor pool for HTTP parsing

  • client_options (Hash) —

    timeout configuration options

Options Hash (client_options:):

  • :first_data_timeout (Integer) —

    timeout for initial data

  • :chunk_data_timeout (Integer) —

    timeout for subsequent chunks

  • :persistent_data_timeout (Integer) —

    timeout for keep-alive connections



91
92
93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/raptor/reactor.rb', line 91

def initialize(thread_pool, ractor_pool, client_options:)
  @thread_pool = thread_pool
  @ractor_pool = ractor_pool
  @client_options = client_options

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

  @id_to_socket = {}
  @socket_to_state = {}
  @id_to_timeout = {}
  @id_to_mutex = {}
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



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

def add(state)
  socket = state[:socket]
  state.delete(:socket)
  @id_to_socket[state[:id]] = socket
  @socket_to_state[socket] = state

  if state[:protocol] == :http2
    @id_to_mutex[state[:id]] = Mutex.new
  end

  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



304
305
306
 
# File 'lib/raptor/reactor.rb', line 304

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

#mutex_for(id) ⇒ Mutex?

Returns the mutex for a given HTTP/2 connection.

Parameters:

  • id (Integer) —

    unique client identifier

Returns:

  • (Mutex, nil) —

    the mutex, if found



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

def mutex_for(id)
  @id_to_mutex[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”



225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
 
# File 'lib/raptor/reactor.rb', line 225

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

#remove(id) ⇒ TCPSocket?

Removes a client connection from the reactor.

Called when an HTTP request is complete and ready for application processing. Triggers server accept re-enabling if system capacity allows.

Parameters:

  • id (Integer) —

    unique client identifier

Returns:

  • (TCPSocket, nil) —

    the removed socket, if found



205
206
207
208
209
 
# File 'lib/raptor/reactor.rb', line 205

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



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
151
152
 
# File 'lib/raptor/reactor.rb', line 115

def run
  Thread.new do
    Thread.current.name = self.class.name

    until @queue.closed? && @queue.empty?
      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
    end

    @selector.close
  end
end

#shutdown ⇒ void

This method returns an undefined value.

Initiates reactor shutdown.

Closes the registration queue and wakes up the selector to begin graceful shutdown process.



293
294
295
296
 
# File 'lib/raptor/reactor.rb', line 293

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



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

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



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

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



185
186
187
188
189
190
191
192
193
194
 
# File 'lib/raptor/reactor.rb', line 185

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