Class: Raptor::Server

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

Overview

Accepts client connections on TCP, Unix, SSL, and cleartext HTTP/2 listeners and dispatches them into the request pipeline.

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"
BACKPRESSURE_THRESHOLD_MULTIPLIER =

Returns:

  • (::Float) 
1.2
MIN_BACKPRESSURE_THRESHOLD =

Returns:

  • (::Integer) 
8

Instance Method Summary collapse

Constructor Details

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

Creates a new Server.

Parameters:

  • 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

  • connection_options (Hash) —

    per-connection timeout configuration, used to bound TLS handshakes

  • listeners (Array<TCPServer, UNIXServer, Binder::SslListener, Binder::H2cListener>) —

    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) —

    this worker's BPF slot index, or nil when BPF dispatch is inactive

  • connection_options: (Hash[Symbol, untyped])
  • listeners: (Array[TCPServer | UNIXServer | Binder::SslListener | Binder::H2cListener])
  • drain_accept_queue: (Boolean) (defaults to: false)
  • worker_index: (Integer, nil) (defaults to: nil)

Options Hash (connection_options:):

  • :first_data_timeout (Integer) —

    timeout for completing the TLS handshake



56
57
58
59
60
61
62
63
64
65
66
67
68
 
# File 'lib/raptor/server.rb', line 56

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

Instance Method Details

#accept_connection(listener) ⇒ Boolean

Accepts a connection from the given listener, immediately publishes the pending work to the BPF map, and dispatches the socket into the SSL, cleartext HTTP/2, or HTTP/1.1 pipeline.

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
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
 
# 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

  remote_addr = begin
    tcp_client.is_a?(TCPSocket) ? tcp_client.remote_address.ip_address : DEFAULT_REMOTE_ADDR
  rescue SystemCallError
    tcp_client.close
    return true
  end

  ReuseportBPF.update_accepted_load(@reactor.backlog + 1) if @worker_index && @running.true?

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

  if listener.is_a?(Binder::H2cListener)
    @http2.eager_accept(
      tcp_client,
      tcp_client.object_id,
      @reactor,
      @thread_pool,
      remote_addr,
      HTTP_SCHEME
    )
    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, bounded by :first_data_timeout, and dispatches it through the HTTP/2 or HTTP/1.1 path.

Parameters:

  • listener (Binder::SslListener) —

    the SSL listener that accepted the connection

  • tcp_client (TCPSocket) —

    the accepted TCP socket

  • remote_addr (String) —

    the client IP address



230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
 
# File 'lib/raptor/server.rb', line 230

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
    @http2.eager_accept(
      ssl_socket,
      ssl_socket.object_id,
      @reactor,
      @thread_pool,
      remote_addr,
      HTTPS_SCHEME
    )
    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.



158
159
160
161
162
163
164
 
# File 'lib/raptor/server.rb', line 158

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



265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
 
# File 'lib/raptor/server.rb', line 265

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
  rescue SystemCallError
    ssl_socket.close rescue nil
    false
  end
end

#run ⇒ Thread

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

Returns:

  • (Thread) —

    the thread running the accept loop



75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
 
# File 'lib/raptor/server.rb', line 75

def run
  @load_reporter = spawn_load_reporter if @worker_index

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

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

      next unless ready_servers

      backpressure_threshold = [(@thread_pool.size * BACKPRESSURE_THRESHOLD_MULTIPLIER).ceil, MIN_BACKPRESSURE_THRESHOLD].max
      if @reactor.backlog >= backpressure_threshold || @thread_pool.queue_size > @thread_pool.size
        CPU.yield
        next
      end

      break if @running.false?

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

    if @worker_index
      # Stop load updates before parking the slot so the sentinel survives.
      @load_reporter.join
      ReuseportBPF.mark_unavailable(@worker_index)
    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.



128
129
130
131
 
# File 'lib/raptor/server.rb', line 128

def shutdown
  stop_accepting
  @listeners.each(&:close)
end

#spawn_load_reporter ⇒ Thread

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

Returns:

  • (Thread) —

    the load reporter



141
142
143
144
145
146
147
148
149
150
 
# File 'lib/raptor/server.rb', line 141

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

    while @running.true?
      ReuseportBPF.update_load(@reactor.backlog)
      sleep 0.001
    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.



116
117
118
119
120
 
# File 'lib/raptor/server.rb', line 116

def stop_accepting
  @running.make_false
  @thread&.wakeup rescue nil
  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



298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
 
# File 'lib/raptor/server.rb', line 298

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