Class: Kino::Server

Inherits:
Object
  • Object
show all
Defined in:
lib/kino/server.rb

Overview

Public server API. All network I/O lives in Rust (tokio + hyper); this class only manages lifecycle and the Ruby worker pool.

Topology is Puma-style two-level: workers ractors × threads threads per ractor in :ractor mode; the same total capacity flattened onto plain Threads in :threaded mode (which runs ANY Rack app, Rails included).

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(app, config_file: nil, **options) ⇒ Server

Settings precedence: explicit kwargs > config_file DSL > defaults.

Examples:

Kino::Server.new(app, config_file: "kino.rb", port: 3000)

Parameters:

  • app (#call) —

    a Rack 3 application

  • config_file (String, nil) (defaults to: nil) —

    path to a kino.rb config file

  • options (Hash) —

    any Configuration setting, e.g. port:, workers:, threads:, mode:, request_timeout:, tls: cert/key Hash



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
 
# File 'lib/kino/server.rb', line 34

def initialize(app, config_file: nil, **options)
  config = Configuration.new
  config.load_file(config_file) if config_file
  config.merge!(options)
  settings = config.to_h

  @app = app
  @bind = settings[:bind]
  @requested_port = settings[:port]
  @workers = Integer(settings[:workers])
  @on_error = validate_on_error(settings[:on_error])
  @mode = resolve_mode(settings[:mode])
  # Default threads per mode: 1 in :ractor (threads inside a ractor
  # share its lock; a measured +17% on fast handlers; raise `workers`
  # for I/O concurrency instead), 3 in :threaded (threads ARE the
  # concurrency there).
  @threads = Integer(settings[:threads] || ((@mode == :ractor) ? 1 : 3))
  @queue_depth = Integer(settings[:queue_depth])
  @queue_timeout_ms = (Float(settings[:queue_timeout]) * 1000).round
  @request_timeout_ms = settings[:request_timeout] ? (Float(settings[:request_timeout]) * 1000).round : 0
  @max_connections = settings[:max_connections] ? Integer(settings[:max_connections]) : default_max_connections
  @max_body_size = Integer(settings[:max_body_size] || 0)
  @batch = [Integer(settings[:batch]), 1].max
  @lanes = !!settings[:lanes]
  @log_requests = !!settings[:log_requests]
  @shutdown_timeout = settings[:shutdown_timeout]
  @tokio_threads = settings[:tokio_threads]
  @tls = validate_tls(settings[:tls])
  @pidfile = settings[:pidfile]
  @worker_threads = []
  @supervisor = nil
  @started = false
end

Instance Attribute Details

#bind ⇒ String (readonly)

Returns the bind address.

Returns:

  • (String) —

    the bind address



19
20
21
 
# File 'lib/kino/server.rb', line 19

def bind
  @bind
end

#mode ⇒ Symbol (readonly)

Returns the resolved dispatch mode, :ractor or :threaded.

Returns:

  • (Symbol) —

    the resolved dispatch mode, :ractor or :threaded



16
17
18
 
# File 'lib/kino/server.rb', line 16

def mode
  @mode
end

#port ⇒ Integer? (readonly)

Returns the bound port (nil until #start; the actual port when configured with port 0).

Returns:

  • (Integer, nil) —

    the bound port (nil until #start; the actual port when configured with port 0)



13
14
15
 
# File 'lib/kino/server.rb', line 13

def port
  @port
end

Class Method Details

.run(app, **opts) ⇒ Kino::Server

Production entry point: start, print the banner, trap INT/TERM for graceful shutdown (second signal force-exits), block until done. The kino CLI funnels into this too (CLI#serve).

Parameters:

  • app (#call) —

    a Rack 3 application

  • opts (Hash) —

    see #initialize

Returns:



169
170
171
172
173
174
175
176
177
178
 
# File 'lib/kino/server.rb', line 169

def self.run(app, **opts)
  server = new(app, **opts)
  CLI.opening_credits
  server.start
  CLI.action!(server)
  CLI.fin_at_exit
  trap_signals(server)
  server.wait
  server
end

.trap_signals(server) ⇒ void

This method returns an undefined value.

Signal handling shared by Server.run and the kino CLI: INT/TERM drain gracefully (a second signal force-exits), USR1 prints a stats line.

Parameters:



185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
 
# File 'lib/kino/server.rb', line 185

def self.trap_signals(server)
  # kill -USR1 <pid> prints a one-line stats snapshot (find the pid in
  # the pidfile when configured).
  trap("USR1") do
    Thread.new { $stdout.puts Kino::CLI.stats_line(server.stats) }
  end
  signaled = false
  %w[INT TERM].each do |signal|
    trap(signal) do
      Process.exit!(1) if signaled
      signaled = true
      $stderr.write("Kino: draining (signal again to force exit)\n")
      # Trap context forbids mutexes; do the real work on a thread.
      Thread.new { server.shutdown }
    end
  end
end

Instance Method Details

#shutdown(timeout: nil) ⇒ nil

Graceful shutdown: stop accepting, drain in-flight work up to the deadline, then escalate: abort remaining clients (500), interrupt blocked workers, kill stragglers; and tear down the runtime. Always returns by ~deadline + a small epsilon; idempotent.

Parameters:

  • timeout (Numeric, nil) (defaults to: nil) —

    drain deadline in seconds (default: the configured shutdown_timeout)

Returns:

  • (nil) 


119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
 
# File 'lib/kino/server.rb', line 119

def shutdown(timeout: nil)
  return unless @started

  deadline = monotonic_now + (timeout || @shutdown_timeout)
  Native.stop_accepting(@id)

  # Drain: wait for queued + in-flight to reach zero, bounded by deadline.
  until monotonic_now >= deadline
    queued, in_flight = Native.queue_stats(@id)
    break if queued.zero? && in_flight.zero?

    sleep 0.01
  end

  # Idle workers see the closed queue and exit their loops.
  Native.close_queue(@id)
  join_workers(deadline)

  unless workers_done?
    # Past the deadline with stuck handlers: free the clients first,
    # then try to unblock and reap the workers.
    Native.abort_all_inflight(@id)
    Native.interrupt_all_workers(@id)
    join_workers(monotonic_now + 0.2)
    kill_stragglers
  end

  Native.shutdown_runtime(@id, 1_000)
  # The runtime is gone, so hyper has dropped every pinned buffer;
  # the keeper (and the strings it marked) may now be collected.
  @pin_keeper = nil
  @worker_threads.clear
  @started = false
  remove_pidfile if @pidfile
  nil
end

#start ⇒ self

Bind, boot the native front-end, and spawn the worker pool.

Returns:

  • (self)

Raises:



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
98
99
100
101
102
103
104
105
106
107
108
109
 
# File 'lib/kino/server.rb', line 73

def start
  raise Error, "server already started" if @started

  # Claim the pidfile before binding: refusing to start (another
  # instance is alive) must not leave a booted native runtime behind.
  write_pidfile if @pidfile
  booted = false
  begin
    @id, @port = Native.server_start(
      bind: @bind, port: @requested_port,
      queue_depth: @queue_depth, queue_timeout_ms: @queue_timeout_ms,
      request_timeout_ms: @request_timeout_ms,
      max_connections: @max_connections,
      max_body_size: @max_body_size,
      tokio_threads: @tokio_threads,
      tls_cert: @tls&.fetch(:cert), tls_key: @tls&.fetch(:key),
      lanes: @lanes, log_requests: @log_requests
    )
    booted = true
  ensure
    remove_pidfile if @pidfile && !booted
  end
  # GC anchor for zero-copy response buffers: held for the server's
  # lifetime so in-flight buffers survive even a worker ractor crash.
  @pin_keeper = Native.pin_keeper(@id)
  if @mode == :ractor
    @supervisor = RactorSupervisor.new(@id, @app, workers: @workers, threads: @threads,
      batch: @batch, on_error: @on_error).start
  else
    @worker_threads = (@workers * @threads).times.map do
      worker_id = Native.register_worker(@id)
      Thread.new { Worker.run(@id, worker_id, @app, @batch, @on_error) }
    end
  end
  @started = true
  self
end

#stats ⇒ Hash{Symbol => Object}

Live snapshot. Counters come from the native layer (one relaxed atomic per request); config echo makes the line self-describing.

Returns:

  • (Hash{Symbol => Object}) —

    mode, lanes, workers, threads, batch, respawns; plus queued, in_flight, served, rejected, timeouts (and lane_depths in lanes mode) once started



209
210
211
212
213
214
215
216
217
218
219
220
 
# File 'lib/kino/server.rb', line 209

def stats
  base = {
    mode: @mode, lanes: @lanes, workers: @workers, threads: @threads,
    batch: @batch, respawns: @supervisor ? @supervisor.respawns : 0
  }
  return base unless @started

  queued, in_flight, served, rejected, timeouts, lane_depths = Native.server_stats(@id)
  base.merge!(queued:, in_flight:, served:, rejected:, timeouts:)
  base[:lane_depths] = lane_depths if lane_depths
  base
end

#tls? ⇒ Boolean

Returns whether TLS termination is configured.

Returns:

  • (Boolean) —

    whether TLS termination is configured



22
23
24
 
# File 'lib/kino/server.rb', line 22

def tls?
  !@tls.nil?
end

#wait ⇒ void

This method returns an undefined value.

Block until every worker has exited (i.e. until shutdown).



158
159
160
 
# File 'lib/kino/server.rb', line 158

def wait
  @supervisor ? @supervisor.join : @worker_threads.each(&:join)
end