Module: Raptor::ReuseportBPF

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

Overview

Routes incoming connections across worker processes to the worker with the lowest reactor backlog.

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. 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__)

Class Method Summary collapse

Class Method Details

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

Returns tcp listening sockets bound and registered for worker_index. Skips non-tcp bind URIs.

Parameters:

  • bind_uris (Array<String>) —

    the configured bind URIs

  • 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



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

def self.create_worker_listeners(bind_uris, worker_index, socket_backlog)
  bind_uris.filter_map do |bind_uri|
    uri = URI(bind_uri)
    next unless uri.scheme == "tcp"

    host = uri.host
    host = host[1..-2] if host&.start_with?("[")

    addrinfo = Addrinfo.tcp(host, uri.port)
    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.bind(addrinfo)
    socket.listen(socket_backlog)

    LibBPFRuby.sockmap_update(@socks_fd, [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



99
100
101
102
 
# File 'lib/raptor/reuseport_bpf.rb', line 99

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

.mark_unavailable(worker_index) ⇒ void

This method returns an undefined value.

Marks a worker's slot as unavailable by parking its load at the sentinel maximum so the BPF program's least-loaded pick skips it.

Parameters:

  • worker_index (Integer) —

    slot index for the worker to mark



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

def self.mark_unavailable(worker_index)
  return unless @loads_fd

  LibBPFRuby.map_update(@loads_fd, [worker_index + 1].pack("L"), [0xffffffff].pack("L"))
end

.setup(worker_count) ⇒ Boolean

Prepares BPF-directed routing for worker_count workers.

Returns true when BPF-directed dispatch is active. Returns false when the platform is unsupported. Raises when the kernel refuses the program.

Parameters:

  • worker_count (Integer) —

    number of worker processes that will bind sockets

Returns:

  • (Boolean) 


46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/raptor/reuseport_bpf.rb', line 46

def self.setup(worker_count)
  return false unless supported?

  @object = LibBPFRuby::Object.new(BPF_OBJECT_PATH)
  @program_fd = @object.program_fd("select_least_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) 


32
33
34
 
# File 'lib/raptor/reuseport_bpf.rb', line 32

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

.update_load(backlog) ⇒ void

This method returns an undefined value.

Publishes this worker's current backlog to the BPF map.

Parameters:

  • backlog (Integer) —

    the worker's current reactor backlog



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

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