Module: Raptor::ReuseportBPF

Defined in:
lib/raptor/reuseport_bpf.rb,
sig/generated/raptor/reuseport_bpf.rbs

Overview

Routes incoming connections across worker processes using two-choice load sampling, atomically reserving the selected slot before routing.

Auto-enabled on Linux when the libbpf-ruby gem is installed and the BPF object file has been compiled. Falls back silently to standard SO_REUSEPORT when either is missing, or when there are more workers or listeners than the BPF maps have slots for. Raises when the kernel refuses the BPF program. Wraps tcp:// bindings only; non-tcp bindings are left to the binder's shared listeners.

Constant Summary collapse

LIBBPF_LOADED =

Returns:

  • (Object) 
!!defined?(LibBPFRuby)
BPF_OBJECT_PATH =

Returns:

  • (Object) 
File.expand_path("../../ext/raptor_bpf/reuseport_select.bpf.o", __dir__)
TCP_INFO_UNACKED_OFFSET =

Returns:

  • (::Integer) 
24
UNAVAILABLE_LOAD =

Returns:

  • (::Integer) 
0xffffffff
MAX_WORKERS =

Returns:

  • (::Integer) 
64
MAX_LISTENERS =

Returns:

  • (::Integer) 
16

Class Method Summary collapse

Class Method Details

.create_worker_listeners(worker_index, socket_backlog) ⇒ Array<TCPServer>

Creates this worker's BPF-routed listeners at the shared listeners' bound addresses, including resolved localhost addresses and ports assigned by the kernel for port 0.

Parameters:

  • worker_index (Integer) —

    slot index for this worker in the socks map

  • socket_backlog (Integer) —

    kernel listen() queue depth

Returns:

  • (Array<TCPServer>) —

    the tcp listening sockets created for this worker



85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
 
# File 'lib/raptor/reuseport_bpf.rb', line 85

def self.create_worker_listeners(worker_index, socket_backlog)
  @worker_listeners = @shared_listeners.each_with_index.map do |shared_listener, listener_index|
    addrinfo = shared_listener.local_address
    socket = Socket.new(addrinfo.afamily, Socket::SOCK_STREAM, 0)
    socket.setsockopt(Socket::SOL_SOCKET, Socket::SO_REUSEADDR, true)
    socket.setsockopt(Socket::SOL_SOCKET, Socket::SO_REUSEPORT, true)
    socket.setsockopt(Socket::IPPROTO_TCP, Socket::TCP_NODELAY, 1)
    socket.setsockopt(Socket::IPPROTO_TCP, Socket::TCP_DEFER_ACCEPT, 1)
    socket.bind(addrinfo)
    socket.listen(socket_backlog)

    LibBPFRuby.sockmap_update(@socks_fd, [listener_index * MAX_WORKERS + worker_index].pack("L"), socket)
    LibBPFRuby.attach_reuseport(socket, @program_fd)

    tcp_server = TCPServer.for_fd(socket.fileno)
    socket.autoclose = false
    tcp_server
  end
end

.enable_load_reporting(worker_index) ⇒ void

This method returns an undefined value.

Prepares this worker's load-reporting slot so subsequent update_load calls can publish its backlog.

Parameters:

  • worker_index (Integer) —

    slot index for this worker



112
113
114
115
116
 
# File 'lib/raptor/reuseport_bpf.rb', line 112

def self.enable_load_reporting(worker_index)
  @load_key = [worker_index + 1].pack("L")
  @load_value = String.new("\x00\x00\x00\x00", encoding: Encoding::ASCII_8BIT)
  @accepted_load_value = String.new("\x00\x00\x00\x00", encoding: Encoding::ASCII_8BIT)
end

.mark_unavailable(worker_index) ⇒ void

This method returns an undefined value.

Parks a worker's load at the maximum so BPF favors other workers. Stop publishing its load first to avoid overwriting the sentinel.

Parameters:

  • worker_index (Integer) —

    slot index for the worker to mark



157
158
159
 
# File 'lib/raptor/reuseport_bpf.rb', line 157

def self.mark_unavailable(worker_index)
  LibBPFRuby.map_update(@loads_fd, [worker_index + 1].pack("L"), [UNAVAILABLE_LOAD].pack("L"))
end

.setup(worker_count, shared_listeners) ⇒ Boolean

Prepares BPF-directed routing for worker_count workers across the addresses shared_listeners are bound to.

Returns true when BPF-directed dispatch is active. Returns false when the platform is unsupported or the counts exceed the map slots. Raises when the kernel refuses the program.

Parameters:

  • worker_count (Integer) —

    number of worker processes that will bind sockets

  • shared_listeners (Array<TCPServer>) —

    the binder's tcp listeners

Returns:

  • (Boolean) 


62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/raptor/reuseport_bpf.rb', line 62

def self.setup(worker_count, shared_listeners)
  return false unless supported? && worker_count <= MAX_WORKERS && shared_listeners.length <= MAX_LISTENERS

  @shared_listeners = shared_listeners

  @object = LibBPFRuby::Object.new(BPF_OBJECT_PATH)
  @program_fd = @object.program_fd("select_less_loaded")
  @socks_fd = @object.map_fd("socks")
  @loads_fd = @object.map_fd("loads")

  LibBPFRuby.map_update(@loads_fd, [0].pack("L"), [worker_count].pack("L"))
  true
end

.supported? ⇒ Boolean

Returns true when the preconditions for BPF-directed reuseport are met on this platform.

Returns:

  • (Boolean) 


46
47
48
 
# File 'lib/raptor/reuseport_bpf.rb', line 46

def self.supported?
  LIBBPF_LOADED && File.exist?(BPF_OBJECT_PATH)
end

.update_accepted_load(backlog) ⇒ void

This method returns an undefined value.

Publishes an accept-side load estimate. This uses a separate value buffer because the server and load reporter call into the map from different threads.

Parameters:

  • backlog (Integer) —

    the estimated backlog after accepting a socket



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

def self.update_accepted_load(backlog)
  load = backlog + accept_queue_length
  @accepted_load_value.setbyte(0, load & 0xff)
  @accepted_load_value.setbyte(1, (load >> 8) & 0xff)
  @accepted_load_value.setbyte(2, (load >> 16) & 0xff)
  @accepted_load_value.setbyte(3, (load >> 24) & 0xff)
  LibBPFRuby.map_update(@loads_fd, @load_key, @accepted_load_value)
end

.update_load(backlog) ⇒ void

This method returns an undefined value.

Publishes this worker's load to the BPF map.

Parameters:

  • backlog (Integer) —

    the worker's current reactor backlog



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

def self.update_load(backlog)
  load = backlog + accept_queue_length
  @load_value.setbyte(0, load & 0xff)
  @load_value.setbyte(1, (load >> 8) & 0xff)
  @load_value.setbyte(2, (load >> 16) & 0xff)
  @load_value.setbyte(3, (load >> 24) & 0xff)
  LibBPFRuby.map_update(@loads_fd, @load_key, @load_value)
end