Class: PumaPlus::ControlChannel

Inherits:
Object
  • Object
show all
Defined in:
lib/puma_plus/control_channel.rb

Overview

The control connection to the Go server, shared by every kind of worker supervisor.

Extracted because Shepherd (which forks processes) and RactorWorker (which spawns Ractors) had byte-identical copies of connect, command loop, heartbeat and RSS reading. The only genuinely different parts are what SET_SLOTS means and what goes in the heartbeat -- which is exactly what the handler protocol below covers. Duplicating the rest meant a protocol change had to be made in two files, and nothing would have failed if it were made in one.

A supervisor is the handler. It must respond to:

running?       -- keep looping?
set_slots(n)   -- SET_SLOTS arrived; n is the desired capacity
quiesce        -- stop replacing capacity that dies
shutdown(ms)   -- stop, with this much grace
heartbeat_kv   -- extra [key, value] pairs for WORKER_STATUS

and may respond to:

on_idle        -- called each pass, whether or not a frame arrived, for
                work that cannot wait on the socket (the shepherd reaps
                dead children here)

Constant Summary collapse

SELECT_TIMEOUT =

Long enough to be cheap, short enough that signal flags and child deaths are noticed promptly rather than only when Go happens to send something.

0.25

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(socket_path:, logger: $stderr) ⇒ ControlChannel

Returns a new instance of ControlChannel.



37
38
39
40
41
42
43
44
45
46
 
# File 'lib/puma_plus/control_channel.rb', line 37

def initialize(socket_path:, logger: $stderr)
  @socket_path = socket_path
  @logger = logger
  # Guards the socket. The heartbeat thread writes WORKER_STATUS while the
  # command loop writes PONG, and without this the two can interleave
  # mid-frame and hand Go a corrupt stream -- a race that existed in both
  # copies of this code and had simply never been hit, since PING is only
  # sent when a worker looks suspect.
  @write_mu = Mutex.new
end

Class Method Details

.rss_kb ⇒ Object

Advisory only. Go reads /proc itself for the memory guard rather than trusting a number reported by a process it may need to kill.



75
76
77
78
79
 
# File 'lib/puma_plus/control_channel.rb', line 75

def self.rss_kb
  File.read("/proc/self/statm").split[1].to_i * (Etc.sysconf(Etc::SC_PAGESIZE) / 1024)
rescue StandardError
  0
end

Instance Method Details

#close ⇒ Object 



67
68
69
70
71
 
# File 'lib/puma_plus/control_channel.rb', line 67

def close
  @conn&.close
rescue IOError
  nil
end

#connect! ⇒ Object 



48
49
50
51
52
53
54
55
56
57
 
# File 'lib/puma_plus/control_channel.rb', line 48

def connect!
  @conn = UNIXSocket.new(@socket_path)
  @conn.sync = true
  write(Wire::HELLO, Wire.encode_kv([
                                      ["version", Wire::VERSION],
                                      ["role", "control"],
                                      ["pid", Process.pid]
                                    ]))
  self
end

#run(handler) ⇒ Object

Serve commands until the connection closes or the handler stops running.



60
61
62
63
64
65
 
# File 'lib/puma_plus/control_channel.rb', line 60

def run(handler)
  heartbeat = Thread.new { heartbeat_loop(handler) }
  command_loop(handler)
ensure
  heartbeat&.kill
end