Class: Yahns::Server
- Inherits:
-
Object
- Object
- Yahns::Server
- Includes:
- SocketHelper
- Defined in:
- lib/yahns/server.rb
Overview
:nodoc:
Constant Summary collapse
- QUEUE_SIGS =
[ :WINCH, :QUIT, :INT, :TERM, :USR1, :USR2, :HUP, :TTIN, :TTOU, :CHLD ]
Instance Attribute Summary collapse
-
#atfork_child ⇒ Object
writeonly
Sets the attribute atfork_child.
-
#atfork_parent ⇒ Object
writeonly
Sets the attribute atfork_parent.
-
#atfork_prepare ⇒ Object
writeonly
Sets the attribute atfork_prepare.
-
#before_exec ⇒ Object
writeonly
Sets the attribute before_exec.
-
#daemon_pipe ⇒ Object
Returns the value of attribute daemon_pipe.
-
#logger ⇒ Object
Returns the value of attribute logger.
-
#shutdown_timeout ⇒ Object
writeonly
Sets the attribute shutdown_timeout.
-
#user ⇒ Object
writeonly
Sets the attribute user.
-
#worker_processes ⇒ Object
writeonly
Sets the attribute worker_processes.
Instance Method Summary collapse
-
#bind_new_listeners! ⇒ Object
call only after calling inherit_listeners! This binds any listeners we did NOT inherit from the parent.
- #clobber_pid(path) ⇒ Object
- #daemon_ready ⇒ Object
- #drop_acceptors ⇒ Object
- #dropping(fdmap) ⇒ Object
-
#fdmap_init ⇒ Object
spins up processing threads of the server.
- #inherit_listeners! ⇒ Object
-
#initialize(config) ⇒ Server
constructor
A new instance of Server.
-
#join ⇒ Object
single-threaded only, this is overriden if @worker_processes is non-nil.
-
#listen(address) ⇒ Object
add a given address to the
listeners
set, idempotently Allows workers to add a private, per-process listener via the after_fork hook. -
#listener_names(listeners = @listeners) ⇒ Object
returns an array of string names for the given listener array.
-
#listeners=(listeners) ⇒ Object
replaces current listener set with
listeners
. - #load_config! ⇒ Object
-
#pid=(path) ⇒ Object
sets the path for the PID file of the master process.
- #proc_name(tag) ⇒ Object
- #qegg_vivify(qegg, fdmap) ⇒ Object
- #quit_enter(alive) ⇒ Object
-
#quit_finish ⇒ Object
drops all the the IO objects we have threads waiting on before exiting This just injects the QueueQuitter object which acts like a monkey wrench thrown into a perfectly good engine :).
- #reap_reexec ⇒ Object
-
#reexec ⇒ Object
reexecutes the Yahns::START with a new binary.
- #sock_opts(io) ⇒ Object
- #sp_sig_handle(alive) ⇒ Object
- #sqwakeup(sig) ⇒ Object
- #start ⇒ Object
- #switch_user(user, group = nil) ⇒ Object
-
#unlink_pid_safe(path) ⇒ Object
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. - #usr1_reopen(prefix) ⇒ Object
-
#valid_pid?(path) ⇒ Boolean
returns a PID if a given path contains a non-stale PID file, nil otherwise.
Methods included from SocketHelper
#bind_listen, #log_buffer_sizes, #new_tcp_server, #server_cast, #set_server_sockopt, #so_reuseport, #sock_name, #tcp_name
Constructor Details
#initialize(config) ⇒ Server
Returns a new instance of Server.
22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 |
# File 'lib/yahns/server.rb', line 22 def initialize(config) @shutdown_expire = nil @shutdown_timeout = nil @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 @before_exec = nil @atfork_prepare = @atfork_parent = @atfork_child = nil @user = nil @queues = [] @wthr = [] end |
Instance Attribute Details
#atfork_child=(value) ⇒ Object (writeonly)
Sets the attribute atfork_child
19 20 21 |
# File 'lib/yahns/server.rb', line 19 def atfork_child=(value) @atfork_child = value end |
#atfork_parent=(value) ⇒ Object (writeonly)
Sets the attribute atfork_parent
18 19 20 |
# File 'lib/yahns/server.rb', line 18 def atfork_parent=(value) @atfork_parent = value end |
#atfork_prepare=(value) ⇒ Object (writeonly)
Sets the attribute atfork_prepare
17 18 19 |
# File 'lib/yahns/server.rb', line 17 def atfork_prepare=(value) @atfork_prepare = value end |
#before_exec=(value) ⇒ Object (writeonly)
Sets the attribute before_exec
14 15 16 |
# File 'lib/yahns/server.rb', line 14 def before_exec=(value) @before_exec = value end |
#daemon_pipe ⇒ Object
Returns the value of attribute daemon_pipe.
11 12 13 |
# File 'lib/yahns/server.rb', line 11 def daemon_pipe @daemon_pipe end |
#logger ⇒ Object
Returns the value of attribute logger.
12 13 14 |
# File 'lib/yahns/server.rb', line 12 def logger @logger end |
#shutdown_timeout=(value) ⇒ Object (writeonly)
Sets the attribute shutdown_timeout
16 17 18 |
# File 'lib/yahns/server.rb', line 16 def shutdown_timeout=(value) @shutdown_timeout = value end |
#user=(value) ⇒ Object (writeonly)
Sets the attribute user
13 14 15 |
# File 'lib/yahns/server.rb', line 13 def user=(value) @user = value end |
#worker_processes=(value) ⇒ Object (writeonly)
Sets the attribute worker_processes
15 16 17 |
# File 'lib/yahns/server.rb', line 15 def worker_processes=(value) @worker_processes = value end |
Instance Method Details
#bind_new_listeners! ⇒ Object
call only after calling inherit_listeners! This binds any listeners we did NOT inherit from the parent
341 342 343 344 |
# File 'lib/yahns/server.rb', line 341 def bind_new_listeners! self.listeners = @config.config_listeners.keys raise ArgumentError, "no listeners" if @listeners.empty? end |
#clobber_pid(path) ⇒ Object
120 121 122 123 124 125 126 127 128 129 130 131 132 133 |
# File 'lib/yahns/server.rb', line 120 def clobber_pid(path) 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 end |
#daemon_ready ⇒ Object
203 204 205 206 207 208 209 210 211 212 |
# File 'lib/yahns/server.rb', line 203 def daemon_ready @daemon_pipe.respond_to?(:syswrite) or return begin @daemon_pipe.syswrite("#$$") rescue => e @logger.warn("grandparent died too soon?: #{e.} (#{e.class})") end @daemon_pipe.close @daemon_pipe = true # for SIGWINCH end |
#drop_acceptors ⇒ Object
82 83 84 |
# File 'lib/yahns/server.rb', line 82 def drop_acceptors @listeners.delete_if(&:ac_quit) end |
#dropping(fdmap) ⇒ Object
499 500 501 502 503 504 505 506 507 508 |
# File 'lib/yahns/server.rb', line 499 def dropping(fdmap) if drop_acceptors[0] || fdmap.size > 0 timeout = @shutdown_expire < Yahns.now ? -1 : @shutdown_timeout n = fdmap.desperate_expire(timeout) $0 = "yahns quitting, #{n} FD(s) remain" true else false end end |
#fdmap_init ⇒ Object
spins up processing threads of the server
361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 |
# File 'lib/yahns/server.rb', line 361 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) # once initialize queues (epoll/kqueue) and associated worker threads queues = {} # spin up applications (which are preload: false) @config.app_ctx.each(&:after_fork_init) @shutdown_timeout ||= @config.app_ctx.map(&:client_timeout).max # spin up acceptor threads, clients flow into worker queues after this @listeners.each do |l| opts = sock_opts(l) ctx = opts[:yahns_app_ctx] ctx_list = opts[:yahns_app_ctx_list] ||= [] qegg = ctx.qegg || @config.qeggs[:default] ctx.queue = queues[qegg] ||= qegg_vivify(qegg, fdmap) ctx = ctx.dup ctx.__send__(:include, l.expire_mod) if ssl_ctx = opts[:ssl_ctx] ctx.__send__(:include, Yahns::OpenSSLClient) env = ctx.app_defaults = ctx.app_defaults.dup env['HTTPS'] = 'on' # undocumented, but Rack::Request uses this env['rack.url_scheme'] = 'https' # avoid "session id context uninitialized" errors when a client # attempts to reuse a cached SSL session. Server admins may # configure their own cache and session_id_context if desired. # 32 bytes is SSL_MAX_SSL_SESSION_ID_LENGTH and has been since # the SSLeay days ssl_ctx.session_id_context ||= OpenSSL::Random.random_bytes(32) # call OpenSSL::SSL::SSLContext#setup explicitly here to detect # errors and avoid race conditions. We avoid calling this in the # parent process (if we have multiple workers) in case the # setup code starts TCP connections to memcached or similar # for session caching. ssl_ctx.setup end ctx_list << ctx # acceptors feed the the queues l.spawn_acceptor(opts[:threads] || 1, @logger, ctx) end fdmap end |
#inherit_listeners! ⇒ Object
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 |
# File 'lib/yahns/server.rb', line 303 def inherit_listeners! # inherit sockets from parents, they need to be plain Socket objects # before they become Yahns::UNIXServer or Yahns::TCPServer # # Note: we intentionally use a yahns-specific environment variable # here because existing servers may use non-blocking listen sockets. # yahns uses _blocking_ listen sockets exclusively. We cannot # change an existing socket to blocking mode if two servers are # running (one expecting blocking, one expecting non-blocking) # because that can completely break the non-blocking one. # Unfortunately, there is no one-off MSG_DONTWAIT-like flag for # accept4(2). inherited = ENV['YAHNS_FD'].to_s.split(',') # emulate sd_listen_fds() for systemd sd_pid, sd_fds = ENV.values_at('LISTEN_PID', 'LISTEN_FDS') if sd_pid.to_i == $$ # 3 = SD_LISTEN_FDS_START inherited.concat((3...(3 + sd_fds.to_i)).to_a) end # to ease debugging, we will not unset LISTEN_PID and LISTEN_FDS inherited.map! do |fd| io = Socket.for_fd(fd.to_i) opts = sock_opts(io) io = server_cast(io, opts) set_server_sockopt(io, opts) name = sock_name(io) @logger.info "inherited addr=#{name} fd=#{io.fileno}" @config.register_inherited(name) io end @listeners.replace(inherited) end |
#join ⇒ Object
single-threaded only, this is overriden if @worker_processes is non-nil
511 512 513 514 515 516 517 518 519 520 521 522 523 |
# File 'lib/yahns/server.rb', line 511 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 || dropping(fdmap) unlink_pid_safe(@pid) if @pid ensure quit_finish end |
#listen(address) ⇒ Object
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.
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 |
# File 'lib/yahns/server.rb', line 174 def listen(address) address = @config.(address) return if String === address && listener_names.include?(address) delay = 0.5 tries = 5 begin opts = sock_opts(address) io = bind_listen(address, opts) io = server_cast(io, opts) unless io.class.name.start_with?('Yahns::') @logger.info "listening on addr=#{sock_name(io)} fd=#{io.fileno}" @listeners << io io rescue Errno::EADDRINUSE => err if tries == 0 @logger.error "adding listener failed addr=#{address} (in use)" raise err end tries -= 1 @logger.warn "retrying in #{delay} seconds " \ "(#{tries < 0 ? 'infinite' : tries} tries left)" sleep(delay) retry rescue => err @logger.fatal "error adding listener addr=#{address}" raise err end end |
#listener_names(listeners = @listeners) ⇒ Object
returns an array of string names for the given listener array
295 296 297 |
# File 'lib/yahns/server.rb', line 295 def listener_names(listeners = @listeners) listeners.map { |io| sock_name(io) } end |
#listeners=(listeners) ⇒ Object
replaces current listener set with listeners
. This will close the socket if it will not exist in the new listener set
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 |
# File 'lib/yahns/server.rb', line 88 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)) if io.ac_quit true else << io false end else set_server_sockopt(io, sock_opts(io)) false end end .delete_if(&:ac_quit) while [0] (set_names - cur_names).each { |addr| listen(addr) } end |
#load_config! ⇒ Object
282 283 284 285 286 287 288 289 290 291 292 |
# File 'lib/yahns/server.rb', line 282 def load_config! @logger.info "reloading config_file=#{@config.config_file}" @config.config_reload! @config.commit!(self) soft_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 |
#pid=(path) ⇒ Object
sets the path for the PID file of the master process
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 |
# File 'lib/yahns/server.rb', line 136 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 # rename the old pid if possible if @pid && path begin File.rename(@pid, path) rescue Errno::ENOENT, Errno::EXDEV # a user may have accidentally removed the original, # obviously cross-FS renames don't work, either. clobber_pid(path) end else clobber_pid(path) end @pid = path end |
#proc_name(tag) ⇒ Object
346 347 348 349 |
# File 'lib/yahns/server.rb', line 346 def proc_name(tag) s = Yahns::START $0 = ([ File.basename(s[0]), tag ]).concat(s[:argv]).join(' ') end |
#qegg_vivify(qegg, fdmap) ⇒ Object
351 352 353 354 355 356 357 358 |
# File 'lib/yahns/server.rb', line 351 def qegg_vivify(qegg, fdmap) queue = qegg.vivify(fdmap) qegg.worker_threads.times do @wthr << queue.worker_thread(@logger, qegg.max_events) end @queues << queue queue end |
#quit_enter(alive) ⇒ Object
417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 |
# File 'lib/yahns/server.rb', line 417 def quit_enter(alive) if alive @logger.info("gracefully exiting shutdown_timeout=#@shutdown_timeout") @shutdown_expire ||= Yahns.now + @shutdown_timeout + 1 else # drop connections immediately if signaled twice @logger.info("graceful exit aborted, exiting immediately") # we will still call any app-defined at_exit hooks here # use SIGKILL if you don't want that. exit end drop_acceptors # stop acceptors, we close epolls in quit_done @config.config_listeners.each_value do |opts| list= opts[:yahns_app_ctx_list] or next # Yahns::HttpContext#persistent_connections= list.each { |ctx| ctx.persistent_connections = false } end false end |
#quit_finish ⇒ Object
drops all the the IO objects we have threads waiting on before exiting This just injects the QueueQuitter object which acts like a monkey wrench thrown into a perfectly good engine :)
440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 |
# File 'lib/yahns/server.rb', line 440 def quit_finish quitter = Yahns::QueueQuitter.new # throw the monkey wrench into the worker threads @queues.each { |q| q.queue_add(quitter, Yahns::Queue::QEV_QUIT) } # watch the monkey wrench destroy all the threads! # Ugh, this may fail if we have dedicated threads trickling # response bodies out (e.g. "tail -F") Oh well, have a timeout begin @wthr.delete_if { |t| t.join(0.01) } end while @wthr[0] && Yahns.now <= @shutdown_expire # cleanup, our job is done @queues.each(&:close).clear # we must not let quitter get GC-ed if we have any worker threads leftover @quitter = quitter quitter.close rescue => e Yahns::Log.exception(@logger, "quit finish", e) ensure if (@wthr.size + @listeners.size) > 0 @logger.warn("still active wthr=#{@wthr.size} "\ "listeners=#{@listeners.size}") end end |
#reap_reexec ⇒ Object
469 470 471 472 473 474 475 476 |
# File 'lib/yahns/server.rb', line 469 def reap_reexec @reexec_pid > 0 or return wpid, status = Process.waitpid2(@reexec_pid, Process::WNOHANG) wpid or return @logger.error "reaped #{status.inspect} exec()-ed" @reexec_pid = 0 self.pid = @pid.chomp('.oldbin') if @pid end |
#reexec ⇒ Object
reexecutes the Yahns::START with a new binary
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 |
# File 'lib/yahns/server.rb', line 215 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.}" return end end opt = {} @listeners.each { |sock| opt[sock.fileno] = sock } env = ENV.to_hash env['YAHNS_FD'] = opt.keys.join(',') opt[:close_others] = true cmd = [ Yahns::START[0] ].concat(Yahns::START[:argv]) dir = @config.value(:working_directory) || Yahns::START[:cwd] @logger.info "spawning #{cmd.inspect} (in #{dir})" @reexec_pid = if @before_exec fork do Dir.chdir(dir) @before_exec.call(cmd) exec(env, *cmd, opt) end else opt[:chdir] = dir spawn(env, *cmd, opt) end end |
#sock_opts(io) ⇒ Object
299 300 301 |
# File 'lib/yahns/server.rb', line 299 def sock_opts(io) @config.config_listeners[sock_name(io)] || {} end |
#sp_sig_handle(alive) ⇒ Object
478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 |
# File 'lib/yahns/server.rb', line 478 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 :CHLD reap_reexec when :USR1 usr1_reopen(nil) when :USR2 reexec when :HUP reexec return quit_enter(alive) when :TTIN, :TTOU, :WINCH @logger.info("SIG#{sig} ignored in single-process mode") end alive end |
#sqwakeup(sig) ⇒ Object
42 43 44 45 |
# File 'lib/yahns/server.rb', line 42 def sqwakeup(sig) @sig_queue << sig @sev.sev_signal end |
#start ⇒ Object
47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 |
# File 'lib/yahns/server.rb', line 47 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. # 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) } } bind_new_listeners! self.pid = @config.value(:pid) # write pid file if @worker_processes require_relative 'server_mp' extend Yahns::ServerMP else switch_user(*@user) if @user end self end |
#switch_user(user, group = nil) ⇒ Object
68 69 70 71 72 73 74 75 76 77 78 79 80 |
# File 'lib/yahns/server.rb', line 68 def switch_user(user, group = nil) # we do not protect the caller, checking Process.euid == 0 is # insufficient because modern systems have fine-grained # capabilities. Let the caller handle any and all errors. uid = Etc.getpwnam(user).uid gid = Etc.getgrnam(group).gid if group Yahns::Log.chown_all(uid, gid) if gid && Process.egid != gid Process.initgroups(user, gid) Process::GID.change_privilege(gid) end Process.euid != uid and Process::UID.change_privilege(uid) end |
#unlink_pid_safe(path) ⇒ Object
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
264 265 266 |
# File 'lib/yahns/server.rb', line 264 def unlink_pid_safe(path) (File.read(path).to_i == $$ and File.unlink(path)) rescue nil end |
#usr1_reopen(prefix) ⇒ Object
411 412 413 414 415 |
# File 'lib/yahns/server.rb', line 411 def usr1_reopen(prefix) @logger.info "#{prefix}reopening logs..." Yahns::Log.reopen_all @logger.info "#{prefix}done reopening logs" end |
#valid_pid?(path) ⇒ Boolean
returns a PID if a given path contains a non-stale PID file, nil otherwise.
270 271 272 273 274 275 276 277 278 279 280 |
# File 'lib/yahns/server.rb', line 270 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 |