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, and SSL listeners and dispatches them into the request pipeline. Backs off from accepting when the reactor backlog is high so an overloaded process leaves connections for peers to absorb.

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) 


52
53
54
55
56
57
58
59
60
61
62
63
 
# File 'lib/raptor/server.rb', line 52

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 into the HTTP/1.1 or SSL/HTTP-2 pipeline.

Parameters:

Returns:

  • (Boolean) —

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



156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
 
# File 'lib/raptor/server.rb', line 156

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, 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's IP address



197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
 
# File 'lib/raptor/server.rb', line 197

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.



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

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



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 236

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.

Returns:

  • (Thread) —

    the thread running the accept loop



70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/raptor/server.rb', line 70

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.



111
112
113
114
 
# File 'lib/raptor/server.rb', line 111

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.



124
125
126
127
128
129
130
131
132
133
 
# File 'lib/raptor/server.rb', line 124

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.



100
101
102
103
 
# File 'lib/raptor/server.rb', line 100

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



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

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