1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
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
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
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
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
198
199
200
201
202
203
204
205
206
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
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
| | # -*- encoding: binary -*-
# Copyright (C) 2013, Eric Wong <normalperson@yhbt.net> and all contributors
# License: GPLv3 or later (https://www.gnu.org/licenses/gpl-3.0.txt)
class Yahns::Server # :nodoc:
QUEUE_SIGS = [ :WINCH, :QUIT, :INT, :TERM, :USR1, :USR2, :HUP, :TTIN, :TTOU ]
attr_accessor :daemon_pipe
attr_accessor :logger
attr_writer :worker_processes
include Yahns::SocketHelper
def initialize(config)
@reexec_pid = 0
@daemon_pipe = nil # writable IO or true
@config = config
@workers = {} # pid -> workers
@sig_queue = [] # nil in forked workers
@logger = Logger.new($stderr)
@sev = Yahns::Sigevent.new
@listeners = []
@pid = nil
@worker_processes = nil
@user = nil
@queues = []
@thr = []
end
def sqwakeup(sig)
@sig_queue << sig
@sev.sev_signal
end
def start
@config.commit!(self)
inherit_listeners!
# we try inheriting listeners first, so we bind them later.
# we don't write the pid file until we've bound listeners in case
# yahns was started twice by mistake. Even though our #pid= method
# checks for stale/existing pid files, race conditions are still
# possible (and difficult/non-portable to avoid) and can be likely
# to clobber the pid if the second start was in quick succession
# after the first, so we rely on the listener binding to fail in
# that case. Some tests (in and outside of this source tree) and
# monitoring tools may also rely on pid files existing before we
# attempt to connect to the listener(s)
# setup signal handlers before writing pid file in case people get
# trigger happy and send signals as soon as the pid file exists.
QUEUE_SIGS.each { |sig| trap(sig) { sqwakeup(sig) } }
self.pid = @config.value(:pid) # write pid file
bind_new_listeners!
if @worker_processes
require 'yahns/server_mp'
extend Yahns::ServerMP
mp_init
end
self
end
# replaces current listener set with +listeners+. This will
# close the socket if it will not exist in the new listener set
def listeners=(listeners)
cur_names, dead_names = [], []
listener_names.each do |name|
if ?/ == name[0]
# mark unlinked sockets as dead so we can rebind them
(File.socket?(name) ? cur_names : dead_names) << name
else
cur_names << name
end
end
set_names = listener_names(listeners)
dead_names.concat(cur_names - set_names).uniq!
@listeners.delete_if do |io|
if dead_names.include?(sock_name(io))
(io.close rescue nil).nil? # true
else
set_server_sockopt(io, sock_opts(io))
false
end
end
(set_names - cur_names).each { |addr| listen(addr) }
end
# sets the path for the PID file of the master process
def pid=(path)
if path
if x = valid_pid?(path)
return path if @pid && path == @pid && x == $$
if x == @reexec_pid && @pid =~ /\.oldbin\z/
@logger.warn("will not set pid=#{path} while reexec-ed "\
"child is running PID:#{x}")
return
end
raise ArgumentError, "Already running on PID:#{x} " \
"(or pid=#{path} is stale)"
end
end
unlink_pid_safe(@pid) if @pid
if path
fp = begin
tmp = "#{File.dirname(path)}/#{rand}.#$$"
File.open(tmp, File::RDWR|File::CREAT|File::EXCL, 0644)
rescue Errno::EEXIST
retry
end
fp.syswrite("#$$\n")
File.rename(fp.path, path)
fp.close
end
@pid = path
end
# add a given address to the +listeners+ set, idempotently
# Allows workers to add a private, per-process listener via the
# after_fork hook. Very useful for debugging and testing.
# +:tries+ may be specified as an option for the number of times
# to retry, and +:delay+ may be specified as the time in seconds
# to delay between retries.
# A negative value for +:tries+ indicates the listen will be
# retried indefinitely, this is useful when workers belonging to
# different masters are spawned during a transparent upgrade.
def listen(address)
address = @config.expand_addr(address)
return if String === address && listener_names.include?(address)
begin
io = bind_listen(address, sock_opts(address))
unless Kgio::TCPServer === io || Kgio::UNIXServer === io
io = server_cast(io)
end
@logger.info "listening on addr=#{sock_name(io)} fd=#{io.fileno}"
@listeners << io
io
rescue Errno::EADDRINUSE => err
@logger.error "adding listener failed addr=#{address} (in use)"
rescue => err
@logger.fatal "error adding listener addr=#{address}"
raise err
end
end
def daemon_ready
@daemon_pipe or return
@daemon_pipe.syswrite("#$$")
@daemon_pipe.close
@daemon_pipe = true # for SIGWINCH
end
# reexecutes the Yahns::START with a new binary
def reexec
if @reexec_pid > 0
begin
Process.kill(0, @reexec_pid)
@logger.error "reexec-ed child already running PID:#@reexec_pid"
return
rescue Errno::ESRCH
@reexec_pid = 0
end
end
if @pid
old_pid = "#@pid.oldbin"
begin
self.pid = old_pid # clear the path for a new pid file
rescue ArgumentError
@logger.error "old PID:#{valid_pid?(old_pid)} running with " \
"existing pid=#{old_pid}, refusing rexec"
return
rescue => e
@logger.error "error writing pid=#{old_pid} #{e.class} #{e.message}"
return
end
end
opts = {}
@listeners.each { |sock| opts[sock.fileno] = sock }
env = { "YAHNS_FD" => opts.keys.map(&:to_s).join(',') }
opts[:chdir] = @config.value(:working_directory) || Yahns::START[:cwd]
cmd = [ Yahns::START[0] ].concat(Yahns::START[:argv])
@logger.info "spawning #{cmd.inspect} (in #{opts[:chdir]})"
cmd << opts
@reexec_pid = Process.spawn(env, *cmd)
proc_name 'master (old)'
end
# unlinks a PID file at given +path+ if it contains the current PID
# still potentially racy without locking the directory (which is
# non-portable and may interact badly with other programs), but the
# window for hitting the race condition is small
def unlink_pid_safe(path)
(File.read(path).to_i == $$ and File.unlink(path)) rescue nil
end
# returns a PID if a given path contains a non-stale PID file,
# nil otherwise.
def valid_pid?(path)
wpid = File.read(path).to_i
wpid <= 0 and return
Process.kill(0, wpid)
wpid
rescue Errno::EPERM
@logger.info "pid=#{path} possibly stale, got EPERM signalling PID:#{wpid}"
nil
rescue Errno::ESRCH, Errno::ENOENT
# don't unlink stale pid files, racy without non-portable locking...
end
def load_config!
@logger.info "reloading config_file=#{@config.config_file}"
@config.config_reload!
@config.commit!(self)
kill_each_worker(:QUIT)
Yahns::Log.reopen_all
@logger.info "done reloading config_file=#{@config.config_file}"
rescue StandardError, LoadError, SyntaxError => e
Yahns::Log.exception(@logger,
"error reloading config_file=#{@config.config_file}", e)
end
# returns an array of string names for the given listener array
def listener_names(listeners = @listeners)
listeners.map { |io| sock_name(io) }
end
def sock_opts(io)
@config.config_listeners[sock_name(io)]
end
def inherit_listeners!
# inherit sockets from parents, they need to be plain Socket objects
# before they become Kgio::UNIXServer or Kgio::TCPServer
inherited = ENV['YAHNS_FD'].to_s.split(/,/).map do |fd|
io = Socket.for_fd(fd.to_i)
set_server_sockopt(io, sock_opts(io))
@logger.info "inherited addr=#{sock_name(io)} fd=#{fd}"
server_cast(io)
end
@listeners.replace(inherited)
end
# call only after calling inherit_listeners!
# This binds any listeners we did NOT inherit from the parent
def bind_new_listeners!
self.listeners = @config.config_listeners.keys
raise ArgumentError, "no listeners" if @listeners.empty?
@listeners.each { |l| l.extend(Yahns::Acceptor) }
end
def proc_name(tag)
s = Yahns::START
$0 = ([ File.basename(s[0]), tag ]).concat(s[:argv]).join(' ')
end
# spins up processing threads of the server
def fdmap_init
thresh = @config.value(:client_expire_threshold)
# keeps track of all connections, like ObjectSpace, but only for IOs
fdmap = Yahns::Fdmap.new(@logger, thresh)
# initialize queues (epoll/kqueue) and associated worker threads
queues = {}
@config.qeggs.each do |name, qe|
queue = qe.vivify(fdmap)
qe.worker_threads.times do
@thr << queue.worker_thread(@logger, qe.max_events)
end
@queues << queue
queues[qe] = queue
end
# spin up applications (which are preload: false)
@config.app_ctx.each { |ctx| ctx.after_fork_init }
# spin up acceptor threads, clients flow into worker queues after this
@listeners.each do |l|
ctx = sock_opts(l)[:yahns_app_ctx]
qegg = ctx.qegg || @config.qeggs[:default]
# acceptors feed the the queues
@thr << l.spawn_acceptor(@logger, ctx, queues[qegg])
end
fdmap
end
def usr1_reopen(prefix)
@logger.info "#{prefix}reopening logs..."
Yahns::Log.reopen_all
@logger.info "#{prefix}done reopening logs"
end
def quit_enter(alive)
self.listeners = [] # close acceptors, we close epolls in quit_done
exit(0) unless alive # drop connections immediately if signaled twice
@config.config_listeners.each_value do |opts|
ctx = opts[:yahns_app_ctx] or next
ctx.persistent_connections = false # Yahns::HttpContext
end
false
end
# drops all the the IO objects we have threads waiting on before exiting
def quit_finish
@queues.each(&:close)
self.listeners = [] # just in case, this is used in ensure
@thr.each(&:join)
end
def sp_sig_handle(alive)
@sev.kgio_wait_readable(alive ? nil : 0.01)
@sev.yahns_step
case sig = @sig_queue.shift
when :QUIT, :TERM, :INT
return quit_enter(alive)
when :USR1
usr1_reopen('')
when :USR2
reexec
when :HUP
reexec
return false
when :TTIN, :TTOU, :WINCH
@logger.info("SIG#{sig} ignored in single-process mode")
end
alive
end
# single-threaded only, this is overriden if @worker_processes is non-nil
def join
daemon_ready
fdmap = fdmap_init
alive = true
begin
alive = sp_sig_handle(alive)
rescue => e
Yahns::Log.exception(@logger, "main loop", e)
end while alive || fdmap.size > 0
unlink_pid_safe(@pid) if @pid
ensure
quit_finish
end
end
|