Class: Raptor::Server

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

Overview

Accepts client connections and dispatches them into the request pipeline. Skips acceptance when the reactor backlog is high so an overloaded process leaves connections for peers that can absorb them (via shared SO_REUSEPORT listeners).

Supports TCP, Unix, and SSL listeners. SSL handshakes are offloaded to the thread pool so a slow client can't pin the server thread. For HTTP/1.1 the first request is parsed inline and dispatched straight to the thread pool; HTTP/2 (negotiated via ALPN) is registered with the reactor for frame processing.

Examples:

binder = Binder.new(["tcp://0.0.0.0:3000"])
reactor = Reactor.new(ractor_pool, thread_pool, connection_options: {}, http1_options: {})
http1 = Http1.new(app, 3000)
http2 = Http2.new(app, 3000)
server = Server.new(binder, reactor, thread_pool, http1, http2, connection_options: { first_data_timeout: 30 }, listeners: binder.listeners)
server.run
# ... later
server.shutdown

Constant Summary collapse

HTTP_SCHEME =

Returns:

  • (::String) 
"http"
HTTPS_SCHEME =

Returns:

  • (::String) 
"https"
H2_PROTOCOL =

Returns:

  • (::String) 
"h2"
DEFAULT_REMOTE_ADDR =

Returns:

  • (::String) 
"127.0.0.1"
DEFAULT_SERVER_NAME =

Returns:

  • (::String) 
"localhost"
MIN_BACKPRESSURE_THRESHOLD =

Returns:

  • (::Integer) 
8

Instance Method Summary collapse

Constructor Details

#initialize(binder, reactor, thread_pool, http1, http2, connection_options:, listeners:, drain_accept_queue: false, worker_index: nil) ⇒ Server

Creates a new Server instance.

Parameters:

  • binder (Binder) —

    the binder managing listening sockets

  • reactor (Reactor) —

    the reactor for handling client connections

  • thread_pool (AtomicThreadPool) —

    thread pool for application processing

  • http1 (Http1) —

    the HTTP/1.1 handler

  • http2 (Http2) —

    the HTTP/2 handler (provides the initial SETTINGS frame)

  • connection_options (Hash) —

    per-connection timeout configuration, used to bound TLS handshakes

  • listeners (Array) —

    the per-worker listeners this server accepts on

  • drain_accept_queue (Boolean) (defaults to: false) —

    whether to drain the kernel accept queue on shutdown

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

    the slot index for BPF load reporting

  • connection_options: (Hash[Symbol, untyped])
  • listeners: (Array[untyped])
  • drain_accept_queue: (Boolean) (defaults to: false)
  • worker_index: (Integer, nil) (defaults to: nil) 


68
69
70
71
72
73
74
75
76
77
78
79
 
# File 'lib/raptor/server.rb', line 68

def initialize(binder, reactor, thread_pool, http1, http2, connection_options:, listeners:, drain_accept_queue: false, worker_index: nil)
  @binder = binder
  @listeners = listeners
  @reactor = reactor
  @thread_pool = thread_pool
  @http1 = http1
  @http2 = http2
  @first_data_timeout = connection_options[:first_data_timeout]
  @drain_accept_queue = drain_accept_queue
  @bpf_active = !!worker_index
  @running = AtomicBoolean.new(true)
end

Instance Method Details

#accept_connection(listener) ⇒ Boolean

Accepts a connection from the given listener and dispatches it.

For SSL listeners the TLS handshake is offloaded to the thread pool so a slow client cannot block the server thread. For SSL connections with h2 negotiated via ALPN, the server sends initial SETTINGS and adds the connection to the reactor as an HTTP/2 connection. All other connections follow the HTTP/1.1 path.

Parameters:

Returns:

  • (Boolean) —

    true if a connection was accepted, false if the listener had nothing to dispatch



174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
 
# File 'lib/raptor/server.rb', line 174

def accept_connection(listener)
  tcp_client = begin
    listener.is_a?(Binder::SslListener) ? listener.tcp_server.accept_nonblock : listener.accept_nonblock
  rescue IO::WaitReadable
    return false
  end

  if tcp_client.is_a?(TCPSocket)
    remote_addr = tcp_client.remote_address.ip_address
  else
    remote_addr = DEFAULT_REMOTE_ADDR
  end

  if listener.is_a?(Binder::SslListener)
    @thread_pool << proc do
      dispatch_ssl_connection(listener, tcp_client, remote_addr)
    end
    return true
  end

  @http1.eager_accept(
    tcp_client,
    tcp_client.object_id,
    @reactor,
    @thread_pool,
    remote_addr,
    HTTP_SCHEME
  )
  true
end

#dispatch_ssl_connection(listener, tcp_client, remote_addr) ⇒ void

This method returns an undefined value.

Performs the TLS handshake for an accepted SSL connection and dispatches it through the HTTP/2 or HTTP/1.1 path. The handshake is bounded by :first_data_timeout so a slow client cannot pin a worker thread.

Parameters:

  • listener (Binder::SslListener) —

    the SSL listener that accepted the connection

  • tcp_client (TCPSocket) —

    the accepted TCP socket

  • remote_addr (String) —

    the client's IP address



215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
 
# File 'lib/raptor/server.rb', line 215

def dispatch_ssl_connection(listener, tcp_client, remote_addr)
  ssl_socket = OpenSSL::SSL::SSLSocket.new(tcp_client, listener.ssl_context)
  ssl_socket.sync_close = true
  return unless perform_ssl_handshake(ssl_socket)

  if ssl_socket.alpn_protocol == H2_PROTOCOL
    ssl_socket.write(@http2.initial_settings_frame) rescue nil

    @reactor.add(
      id: ssl_socket.object_id,
      socket: ssl_socket,
      remote_addr: remote_addr,
      url_scheme: HTTPS_SCHEME,
      protocol: :http2,
      writer: @http2.create_writer,
      flow_control: Http2::FlowControl.new
    )

    return
  end

  @http1.eager_accept(
    ssl_socket,
    ssl_socket.object_id,
    @reactor,
    @thread_pool,
    remote_addr,
    HTTPS_SCHEME
  )
end

#drain_accept_queue ⇒ void

This method returns an undefined value.

Dispatches every connection already in the kernel accept queue for each listener until all are drained.



154
155
156
157
158
159
160
 
# File 'lib/raptor/server.rb', line 154

def drain_accept_queue
  loop do
    accepted = false
    @listeners.each { |listener| accepted = true if accept_connection(listener) }
    break unless accepted
  end
end

#perform_ssl_handshake(ssl_socket) ⇒ Boolean

Drives a non-blocking SSL handshake to completion, bounded by the configured first-data timeout. Returns true on success, false on timeout or SSL error.

Parameters:

  • ssl_socket (OpenSSL::SSL::SSLSocket) —

    the SSL socket to hand-shake

Returns:

  • (Boolean) —

    true if the handshake completed



254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
 
# File 'lib/raptor/server.rb', line 254

def perform_ssl_handshake(ssl_socket)
  deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + @first_data_timeout

  begin
    ssl_socket.accept_nonblock
    true
  rescue IO::WaitReadable
    return false unless wait_for_handshake(ssl_socket, deadline, :read)

    retry
  rescue IO::WaitWritable
    return false unless wait_for_handshake(ssl_socket, deadline, :write)

    retry
  rescue OpenSSL::SSL::SSLError => error
    Log.rescued_error(error)
    ssl_socket.close rescue nil
    false
  end
end

#run ⇒ Thread

Starts the server's main accept loop in a new thread.

The accept loop polls listening sockets for ready connections and accepts them when the reactor backlog is under the backpressure threshold. On Linux with BPF-directed dispatch active, a companion thread publishes this worker's backlog to the BPF map.

Returns:

  • (Thread) —

    the thread running the accept loop



91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
 
# File 'lib/raptor/server.rb', line 91

def run
  spawn_load_reporter if @bpf_active

  Thread.new do
    Thread.current.name = "Server"

    backpressure_threshold = [(@thread_pool.size * 1.2).ceil, MIN_BACKPRESSURE_THRESHOLD].max

    while @running.true?
      begin
        ready_servers, _, _ = IO.select(@listeners, nil, nil, 1)
      rescue IOError, Errno::EBADF
        break
      end

      next unless ready_servers
      next if @reactor.backlog >= backpressure_threshold

      ready_servers.each { |listener| accept_connection(listener) }
    end
  end
end

#shutdown ⇒ void

This method returns an undefined value.

Gracefully shuts down the server.

Stops accepting new connections and closes all listening sockets. When drain_accept_queue is enabled, dispatches every connection already in the kernel accept queue before closing the listeners.



123
124
125
126
127
 
# File 'lib/raptor/server.rb', line 123

def shutdown
  @running.make_false
  drain_accept_queue if @drain_accept_queue
  @listeners.each(&:close)
end

#spawn_load_reporter ⇒ void

This method returns an undefined value.

Starts a background thread that publishes this worker's reactor backlog to the BPF map for load-aware dispatch.



137
138
139
140
141
142
143
144
145
146
 
# File 'lib/raptor/server.rb', line 137

def spawn_load_reporter
  Thread.new do
    Thread.current.name = "Load Reporter"

    while @running.true?
      ReuseportBPF.update_load(@reactor.backlog)
      sleep 0.01
    end
  end
end

#wait_for_handshake(ssl_socket, deadline, direction) ⇒ Boolean

Waits up to deadline for the socket to become ready for the next step of the SSL handshake. Closes the socket and returns false on timeout.

Parameters:

  • ssl_socket (OpenSSL::SSL::SSLSocket) —

    the SSL socket

  • deadline (Float) —

    absolute monotonic deadline

  • direction (Symbol) —

    either :read or :write

Returns:

  • (Boolean) —

    true if the socket became ready before the deadline



284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
 
# File 'lib/raptor/server.rb', line 284

def wait_for_handshake(ssl_socket, deadline, direction)
  remaining = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC)
  ready = if remaining <= 0
    false
  elsif direction == :read
    ssl_socket.wait_readable(remaining)
  else
    ssl_socket.wait_writable(remaining)
  end
  return true if ready

  Log.warn "SSL handshake timed out"
  ssl_socket.close rescue nil
  false
end