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
91
92
 
# File 'lib/raptor/reuseport_bpf.rb', line 67

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

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

.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

Whether the preconditions for BPF-directed reuseport are met on this platform. Does not imply that the kernel will accept the program.

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



100
101
102
103
104
105
106
 
# File 'lib/raptor/reuseport_bpf.rb', line 100

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