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. Yields between accepts when the thread pool queue exceeds the pool size so overloaded workers stay responsive to in-flight requests.

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(binder, reactor, thread_pool, http1, http2, connection_options:, listeners:, drain_accept_queue: false, worker_index: nil) ⇒ Server

Creates a new Server instance.

RBS:

  • (Binder binder, Reactor reactor, AtomicThreadPool thread_pool, Http1 http1, Http2 http2, connection_options: Hash[Symbol, untyped], listeners: Array[untyped], ?drain_accept_queue: bool, ?worker_index: Integer?) -> void

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) 


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

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.

RBS:

  • (TCPServer | UNIXServer | Binder::SslListener listener) -> bool

Parameters:

Returns:

  • (Boolean) —

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



162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
 
# File 'lib/raptor/server.rb', line 162

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.

RBS:

  • (Binder::SslListener listener, TCPSocket tcp_client, String remote_addr) -> void

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



203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
 
# File 'lib/raptor/server.rb', line 203

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.

RBS:

  • () -> void



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

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.

RBS:

  • (OpenSSL::SSL::SSLSocket ssl_socket) -> bool

Parameters:

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

    the SSL socket to hand-shake

Returns:

  • (Boolean) —

    true if the handshake completed



242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
 
# File 'lib/raptor/server.rb', line 242

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.

RBS:

  • () -> Thread

Returns:

  • (Thread) —

    the thread running the accept loop



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

def run
  spawn_load_reporter if @bpf_active

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

    backpressure_threshold = [(@thread_pool.size * BACKPRESSURE_THRESHOLD_MULTIPLIER).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

      if @thread_pool.queue_size > @thread_pool.size
        Thread.pass
        next
      end

      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.

RBS:

  • () -> void



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

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.

RBS:

  • () -> void



130
131
132
133
134
135
136
137
138
139
 
# File 'lib/raptor/server.rb', line 130

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.

RBS:

  • () -> void



106
107
108
109
 
# File 'lib/raptor/server.rb', line 106

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.

RBS:

  • (OpenSSL::SSL::SSLSocket ssl_socket, Float deadline, Symbol direction) -> bool

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



272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
 
# File 'lib/raptor/server.rb', line 272

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