Class: Hyperion::Worker

Inherits:
Object
  • Object
show all
Defined in:
lib/hyperion/worker.rb

Overview

Worker process. Receives a listening socket and runs a ‘Hyperion::Server` (fiber accept loop) until SIGTERM.

Two listener sources, picked by the master per-OS:

  • ‘:share` mode (macOS / BSD): the master forwards a pre-bound `TCPServer` / `OpenSSL::SSL::SSLServer` via the `listener:` kwarg. The worker uses it as-is — the fd was inherited across fork.

  • ‘:reuseport` mode (Linux): no listener is passed. The worker binds its own `Socket` with `SO_REUSEPORT` set so the kernel can hash incoming connections across the sibling sockets.

Instance Method Summary collapse

Constructor Details

#initialize(host:, port:, app:, read_timeout:, tls: nil, thread_count: Server::DEFAULT_THREAD_COUNT, config: nil, worker_index: 0, listener: nil) ⇒ Worker

Returns a new instance of Worker.



19
20
21
22
23
24
25
26
27
28
29
30
31
 
# File 'lib/hyperion/worker.rb', line 19

def initialize(host:, port:, app:, read_timeout:, tls: nil,
               thread_count: Server::DEFAULT_THREAD_COUNT,
               config: nil, worker_index: 0, listener: nil)
  @host         = host
  @port         = port
  @app          = app
  @read_timeout = read_timeout
  @tls          = tls
  @thread_count = thread_count
  @config       = config || Hyperion::Config.new
  @worker_index = worker_index
  @listener     = listener
end

Instance Method Details

#run ⇒ Object 



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/hyperion/worker.rb', line 33

def run
  scheme = @tls ? 'https' : 'http'
  Hyperion.logger.info do
    {
      message: 'worker listening',
      pid: Process.pid,
      worker_index: @worker_index,
      url: "#{scheme}://#{@host}:#{@port}"
    }
  end

  server = Server.new(host: @host, port: @port, app: @app,
                      read_timeout: @read_timeout, tls: @tls,
                      thread_count: @thread_count)
  tcp_server = @listener || build_reuseport_listener
  server.adopt_listener(tcp_server)

  Signal.trap('TERM') { server.stop }
  Signal.trap('INT')  { server.stop }

  # `on_worker_boot` runs in the child after fork, after the listener is
  # ready, and before we start accepting. App code reconnects DB/Redis
  # pools here so each worker has its own. Index identifies the slot
  # (0..workers-1) so apps can shard background work if they want.
  @config.on_worker_boot.each { |h| h.call(@worker_index) }

  begin
    server.start
  ensure
    # `on_worker_shutdown` fires when the accept loop exits — either
    # due to graceful SIGTERM or a hard error. Use it to flush metrics,
    # close DB connections cleanly, etc.
    @config.on_worker_shutdown.each { |h| h.call(@worker_index) }
  end
end