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



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
 
# 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])
  @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 has nothing to race.
    access_timing: !!settings[:log_requests] && (@mode == :threaded || @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]
  @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
  @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
  @worker_threads = []
  @worker_threads_lock = Mutex.new
  @supervisor = nil
  @quarantine_monitor = 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:



235
236
237
 
# File 'lib/kino/server.rb', line 235

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:



264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
 
# File 'lib/kino/server.rb', line 264

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



244
245
246
247
248
249
250
251
252
253
254
255
256
257
 
# File 'lib/kino/server.rb', line 244

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) 


182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
 
# File 'lib/kino/server.rb', line 182

def shutdown(timeout: nil)
  return unless @started

  @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
  @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:



134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
 
# File 'lib/kino/server.rb', line 134

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,
      tokio_threads: @tokio_threads,
      tls_cert: @tls&.fetch(:cert), tls_key: @tls&.fetch(:key),
      lanes: @lanes, log_requests: @log_requests,
      mode: @mode.to_s, workers: @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
    @supervisor = RactorSupervisor.new(@id, @app, workers: @workers, threads: @threads,
      batch: @batch, hooks: @worker_hooks, on_worker_exit: @on_worker_exit).start
  else
    @worker_threads = (@workers * @threads).times.map { spawn_worker_thread }
  end
  start_quarantine_monitor if @quarantine_timeout_ms
  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



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

def stats
  base = {
    mode: @mode, lanes: @lanes, workers: @workers, threads: @threads,
    batch: @batch, respawns: 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
  rows = Native.worker_stats(@id)
  base[:worker_status] = rows.map do |index, served, in_flight, busy_ms, quarantined|
    {index:, served:, in_flight:, busy_ms:, quarantined:}
  end
  base[:quarantined] = rows.count { |_index, _served, _in_flight, _busy_ms, quarantined| quarantined }
  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).



225
226
227
 
# File 'lib/kino/server.rb', line 225

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