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



182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
 
# File 'lib/raptor/server.rb', line 182

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



223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
 
# File 'lib/raptor/server.rb', line 223

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.



162
163
164
165
166
167
168
 
# File 'lib/raptor/server.rb', line 162

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



262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
 
# File 'lib/raptor/server.rb', line 262

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, stopping the accept loop and closing all listening sockets.



132
133
134
135
 
# File 'lib/raptor/server.rb', line 132

def shutdown
  stop_accepting
  @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.



145
146
147
148
149
150
151
152
153
154
 
# File 'lib/raptor/server.rb', line 145

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

#stop_accepting ⇒ void

This method returns an undefined value.

Stops the accept loop and, when drain_accept_queue is enabled, dispatches every connection already in the kernel accept queue. Leaves the listening sockets open so they can be handed to replacement workers.



121
122
123
124
 
# File 'lib/raptor/server.rb', line 121

def stop_accepting
  @running.make_false
  drain_accept_queue if @drain_accept_queue
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



292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
 
# File 'lib/raptor/server.rb', line 292

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