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

Raises:

  • (ArgumentError) 


61
62
63
64
65
66
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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
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
 
# File 'lib/kino/server.rb', line 61

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])
  # The pool ceiling; equal to the floor for a fixed pool.
  @max_workers = settings[:max_workers].nil? ? @workers : Integer(settings[:max_workers])
  if @max_workers < @workers
    raise ArgumentError, "max_workers (#{@max_workers}) must be at least workers (#{@workers})"
  end
  @scale_down_after = settings[:scale_down_after].nil? ? 30.0 : Float(settings[:scale_down_after])
  raise ArgumentError, "scale_down_after must be positive" unless @scale_down_after.positive?
  if !settings[:scale_down_after].nil? && !elastic?
    Log.warn("scale_down_after has no effect unless max_workers is above workers")
  end
  @on_error = validate_hook(settings[:on_error], :on_error)
  @after_worker_boot = validate_hook(settings[:after_worker_boot], :after_worker_boot)
  @after_request_complete = validate_hook(settings[:after_request_complete], :after_request_complete)
  @after_boot = validate_hook(settings[:after_boot], :after_boot)
  @on_worker_exit = validate_hook(settings[:on_worker_exit], :on_worker_exit)
  @mode = resolve_mode(settings[:mode])
  @worker_hooks = WorkerHooks.new(
    on_error: @on_error,
    after_worker_boot: @after_worker_boot,
    after_request_complete: @after_request_complete,
    # The access log's GC and allocation figures come from the VM's
    # process-wide counters, so they are measured only where one
    # request at a time can own them: the GVL serializes :threaded
    # mode, and a single ractor (a pool that can never grow past one)
    # has nothing to race.
    access_timing: !!settings[:log_requests] && (@mode == :threaded || @max_workers == 1)
  )
  # 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]
  @io_shards = !!settings[:io_shards]
  @io_threads = Integer(settings[:io_threads]) unless settings[:io_threads].nil?
  if @io_threads && @io_threads < 1
    raise ArgumentError, "io_threads must be >= 1"
  end
  Log.warn("io_threads has no effect unless io_shards is true") if @io_threads && !@io_shards
  @tokio_threads = settings[:tokio_threads]
  @tls = validate_tls(settings[:tls])
  if @tls && unix?
    raise ArgumentError, "TLS is not supported on a unix socket bind; terminate TLS at the proxy in front"
  end
  @http2 = settings.fetch(:http2, true) ? true : false
  @pidfile = settings[:pidfile]
  @control_bind = settings[:control_bind]&.to_s
  @control_token = settings[:control_token]&.to_s
  # An empty token (e.g. control_token ENV["KINO_CONTROL_TOKEN"] with the
  # var unset) must not half-disable auth: treat it as auth off, not as
  # "require a zero-length Bearer token".
  @control_token = nil if @control_token && @control_token.empty?
  @quarantine_timeout_ms = settings[:quarantine_timeout] ? (Float(settings[:quarantine_timeout]) * 1000).round : nil
  @quarantine_max =
    if settings[:quarantine_max]
      Integer(settings[:quarantine_max])
    elsif @mode == :ractor
      @workers
    else
      @workers * @threads
    end
  @supervisor = nil
  @threaded_pool = nil
  @quarantine_monitor = nil
  @pool_scaler = nil
  @started = false
end

Instance Attribute Details

#bind ⇒ String (readonly)

Returns the bind address.

Returns:

  • (String) —

    the bind address



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

def bind
  @bind
end

#control_port ⇒ Integer? (readonly)

Returns the control plane's TCP port (nil until #start, when the control plane is off, or for a unix-socket bind).

Returns:

  • (Integer, nil) —

    the control plane's TCP port (nil until #start, when the control plane is off, or for a unix-socket bind)



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

def control_port
  @control_port
end

#mode ⇒ Symbol (readonly)

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

Returns:

  • (Symbol) —

    the resolved dispatch mode, :ractor or :threaded



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

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: build the server and #run it. The kino CLI funnels into this too (CLI#serve).

Parameters:

  • app (#call) —

    a Rack 3 application

  • opts (Hash) —

    see #initialize

Returns:



260
261
262
 
# File 'lib/kino/server.rb', line 260

def self.run(app, **opts)
  new(app, **opts).run
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:



289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
 
# File 'lib/kino/server.rb', line 289

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 { Log.info(CLI.stats_line(server.stats)) }
  end
  signaled = false
  %w[INT TERM].each do |signal|
    trap(signal) do
      Process.exit!(1) if signaled
      signaled = true
      Log.warn("draining (signal again to force exit)")
      # Trap context forbids mutexes; do the real work on a thread.
      Thread.new { server.shutdown }
    end
  end
end

Instance Method Details

#control_url ⇒ String?

Where the control plane listens, once started, or nil when it is off: http://host:port, or its unix:// socket path.

Returns:

  • (String, nil) 


46
47
48
49
50
51
 
# File 'lib/kino/server.rb', line 46

def control_url
  return nil unless @control_bind
  return @control_bind if @control_bind.start_with?("unix://")

  "http://#{@control_bind.rpartition(":").first}:#{@control_port}"
end

#run ⇒ self

Serve until shut down: start, print the banner, trap INT/TERM for graceful shutdown (second signal force-exits), block until done. The Rack handler calls this on a server it built itself.

Returns:

  • (self) —

    after shutdown



269
270
271
272
273
274
275
276
277
278
279
280
281
282
 
# File 'lib/kino/server.rb', line 269

def run
  # Startup output must land immediately even when stdout is a pipe or
  # file (process supervisors, `kino > server.log`, `rails server`
  # under Docker); block buffering would hold the banner back until
  # exit.
  $stdout.sync = true
  CLI.opening_credits
  start
  CLI.action!(self)
  CLI.fin_at_exit
  self.class.trap_signals(self)
  wait
  self
end

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


207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
 
# File 'lib/kino/server.rb', line 207

def shutdown(timeout: nil)
  return unless @started

  @pool_scaler&.stop
  @quarantine_monitor&.stop
  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 control thread reports "draining" for the whole drain and stops
  # only now, once there is nothing left to report.
  Native.control_stop(@id)
  # 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
  @started = false
  remove_pidfile if @pidfile
  nil
end

#start ⇒ self

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

Returns:

  • (self)

Raises:



152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
 
# File 'lib/kino/server.rb', line 152

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, @control_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,
      io_shards: @io_shards,
      io_threads: @io_threads,
      tokio_threads: @tokio_threads,
      tls_cert: @tls&.fetch(:cert), tls_key: @tls&.fetch(:key),
      http2: @http2,
      lanes: @lanes, log_requests: @log_requests,
      mode: @mode.to_s, workers: @workers, max_workers: @max_workers,
      threads: @threads, batch: @batch,
      control_bind: @control_bind, control_token: @control_token
    )
    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
    warn_scheduler_cap
    @supervisor = RactorSupervisor.new(@id, @app, workers: @workers, threads: @threads,
      batch: @batch, hooks: @worker_hooks, on_worker_exit: @on_worker_exit).start
  else
    @threaded_pool = ThreadedPool.new(@id, @app, threads: @threads, batch: @batch,
      hooks: @worker_hooks, on_worker_exit: @on_worker_exit).start(@workers)
  end
  start_quarantine_monitor if @quarantine_timeout_ms
  start_pool_scaler if elastic?
  Native.control_ready(@id)
  HookFire.fire(@after_boot, "after_boot")
  @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, worker_status, quarantined, queue_time (and lane_depths in lanes mode) once started



314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
 
# File 'lib/kino/server.rb', line 314

def stats
  base = {
    mode: @mode, lanes: @lanes, workers: @workers, max_workers: @max_workers,
    threads: @threads, batch: @batch, respawns: 0,
    active_workers: @workers, scale_ups: 0, scale_downs: 0
  }
  return base unless @started

  queued, in_flight, served, rejected, timeouts, respawns, lane_depths = Native.server_stats(@id)
  base.merge!(queued:, in_flight:, served:, rejected:, timeouts:, respawns:)
  base[:lane_depths] = lane_depths if lane_depths
  active_workers, _max_workers, scale_ups, scale_downs = Native.pool_stats(@id)
  base.merge!(active_workers:, scale_ups:, scale_downs:)
  rows = Native.worker_stats(@id)
  base[:worker_status] = rows.map do |index, served, in_flight, busy_ms, quarantined, retired|
    {index:, served:, in_flight:, busy_ms:, quarantined:, retired:}
  end
  base[:quarantined] = rows.count { |row| row[4] }
  count, sum_seconds = Native.queue_time(@id)
  base[:queue_time] = {count:, sum_seconds:}
  base
end

#tls? ⇒ Boolean

Returns whether TLS termination is configured.

Returns:

  • (Boolean) —

    whether TLS termination is configured



26
27
28
 
# File 'lib/kino/server.rb', line 26

def tls?
  !@tls.nil?
end

#unix? ⇒ Boolean

Returns whether the bind is a unix domain socket ("unix:///path/to.sock").

Returns:

  • (Boolean) —

    whether the bind is a unix domain socket ("unix:///path/to.sock")



32
33
34
 
# File 'lib/kino/server.rb', line 32

def unix?
  @bind.start_with?("unix://")
end

#url ⇒ String

Where the server listens, once started: http://host:port (https under TLS), or the unix:// socket path.

Returns:

  • (String) 


39
40
41
 
# File 'lib/kino/server.rb', line 39

def url
  unix? ? @bind : "http#{"s" if tls?}://#{@bind}:#{@port}"
end

#wait ⇒ void

This method returns an undefined value.

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



250
251
252
 
# File 'lib/kino/server.rb', line 250

def wait
  pool.join
end