Class: Pitchfork::HttpServer
- Inherits:
-
Object
- Object
- Pitchfork::HttpServer
- Includes:
- HttpResponse, SocketHelper
- Defined in:
- lib/pitchfork/http_server.rb
Overview
This is the process manager of Pitchfork. This manages worker processes which in turn handle the I/O and application process. Listener sockets are started in the master process and shared with forked worker children.
Defined Under Namespace
Classes: TimeoutHandler
Constant Summary collapse
- LISTENERS =
all bound listener sockets note: this is public used by raindrops, but not recommended for use in new projects
[]
- NOOP =
'.'.freeze
- START_CTX =
:startdoc: This Hash is considered a stable interface and changing its contents will allow you to switch between different installations of Pitchfork or even different installations of the same applications without downtime. Keys of this constant Hash are described as follows:
-
0 - the path to the pitchfork executable
-
:argv - a deep copy of the ARGV array the executable originally saw
-
:cwd - the working directory of the application, this is where
you originally started Pitchfork. TODO: Can we get rid of this?
-
{ :argv => ARGV.map(&:dup), 0 => $0.dup, }
Constants included from HttpResponse
Pitchfork::HttpResponse::ILLEGAL_HEADER_VALUE, Pitchfork::HttpResponse::STATUS_CODES
Constants included from SocketHelper
Instance Attribute Summary collapse
-
#after_mold_fork ⇒ Object
:stopdoc:.
-
#after_monitor_ready ⇒ Object
writeonly
Sets the attribute after_monitor_ready.
-
#after_request_complete ⇒ Object
writeonly
Sets the attribute after_request_complete.
-
#after_worker_exit ⇒ Object
writeonly
Sets the attribute after_worker_exit.
-
#after_worker_fork ⇒ Object
:stopdoc:.
-
#after_worker_hard_timeout ⇒ Object
writeonly
Sets the attribute after_worker_hard_timeout.
-
#after_worker_ready ⇒ Object
writeonly
Sets the attribute after_worker_ready.
-
#after_worker_timeout ⇒ Object
writeonly
Sets the attribute after_worker_timeout.
-
#app ⇒ Object
:stopdoc:.
-
#before_fork ⇒ Object
:stopdoc:.
-
#before_worker_exit ⇒ Object
writeonly
Sets the attribute before_worker_exit.
-
#children ⇒ Object
:stopdoc:.
-
#cleanup_timeout ⇒ Object
:stopdoc:.
-
#config ⇒ Object
:stopdoc:.
-
#default_middleware ⇒ Object
:stopdoc:.
-
#early_hints ⇒ Object
:stopdoc:.
-
#listener_opts ⇒ Object
:stopdoc:.
-
#logger ⇒ Object
Returns the value of attribute logger.
-
#orig_app ⇒ Object
:stopdoc:.
-
#ready_pipe ⇒ Object
:stopdoc:.
-
#refork_condition ⇒ Object
writeonly
Sets the attribute refork_condition.
-
#soft_timeout ⇒ Object
:stopdoc:.
-
#spawn_timeout ⇒ Object
:stopdoc:.
-
#timeout ⇒ Object
:stopdoc:.
-
#timeout_signal ⇒ Object
:stopdoc:.
-
#worker_processes ⇒ Object
:stopdoc:.
Instance Method Summary collapse
- #check_client_connection ⇒ Object
- #check_client_connection=(bool) ⇒ Object
- #client_body_buffer_size ⇒ Object
- #client_body_buffer_size=(bytes) ⇒ Object
-
#initialize(app, options = {}) ⇒ HttpServer
constructor
Creates a working server on host:port (strange things happen if port isn’t a Number).
-
#join ⇒ Object
monitors children and receives signals forever (or until a termination signal is sent).
-
#listen(address, opt = {}.merge(listener_opts[address] || {})) ⇒ Object
add a given address to the
listeners
set, idempotently Allows workers to add a private, per-process listener via the after_worker_fork hook. -
#listeners=(listeners) ⇒ Object
replaces current listener set with
listeners
. - #monitor_loop(sleep = true) ⇒ Object
- #rewindable_input ⇒ Object
- #rewindable_input=(bool) ⇒ Object
-
#start(sync = true) ⇒ Object
Runs the thing.
-
#stop(graceful = true) ⇒ Object
Terminates all workers, but does not exit master process.
- #worker_exit(worker) ⇒ Object
Methods included from HttpResponse
#append_header, #err_response, #http_response_write, #httpdate
Methods included from SocketHelper
#bind_listen, #log_buffer_sizes, #new_tcp_server, #server_cast, #set_server_sockopt, #set_tcp_sockopt, sock_name, tcp_name
Constructor Details
#initialize(app, options = {}) ⇒ HttpServer
Creates a working server on host:port (strange things happen if port isn’t a Number). Use HttpServer::run to start the server and HttpServer.run.join to join the thread that’s processing incoming requests on the socket.
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 |
# File 'lib/pitchfork/http_server.rb', line 126 def initialize(app, = {}) @exit_status = 0 @app = app @respawn = false @last_check = Pitchfork.time_now @promotion_lock = Flock.new("pitchfork-promotion") Info.keep_io(@promotion_lock) = .dup @ready_pipe = .delete(:ready_pipe) @init_listeners = [:listeners].dup || [] [:use_defaults] = true self.config = Pitchfork::Configurator.new() self.listener_opts = {} proc_name role: 'monitor', status: START_CTX[:argv].join(' ') # We use @control_socket differently in the master and worker processes: # # * The master process never closes or reinitializes this once # initialized. Signal handlers in the master process will write to # it to wake up the master from IO.select in exactly the same manner # djb describes in https://cr.yp.to/docs/selfpipe.html # # * The workers immediately close the pipe they inherit. See the # Pitchfork::Worker class for the pipe workers use. @control_socket = [] @children = Children.new @sig_queue = [] # signal queue used for self-piping @pid = nil # we try inheriting listeners first, so we bind them later. # we don't write the pid file until we've bound listeners in case # pitchfork 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) config.commit!(self, :skip => [:listeners, :pid]) @orig_app = app # list of signals we care about and trap in master. @queue_sigs = [ :QUIT, :INT, :TERM, :USR2, :TTIN, :TTOU ] Info.workers_count = worker_processes SharedMemory.preallocate_drops(worker_processes) end |
Instance Attribute Details
#after_mold_fork ⇒ Object
:stopdoc:
77 78 79 |
# File 'lib/pitchfork/http_server.rb', line 77 def after_mold_fork @after_mold_fork end |
#after_monitor_ready=(value) ⇒ Object (writeonly)
Sets the attribute after_monitor_ready
82 83 84 |
# File 'lib/pitchfork/http_server.rb', line 82 def after_monitor_ready=(value) @after_monitor_ready = value end |
#after_request_complete=(value) ⇒ Object (writeonly)
Sets the attribute after_request_complete
82 83 84 |
# File 'lib/pitchfork/http_server.rb', line 82 def after_request_complete=(value) @after_request_complete = value end |
#after_worker_exit=(value) ⇒ Object (writeonly)
Sets the attribute after_worker_exit
82 83 84 |
# File 'lib/pitchfork/http_server.rb', line 82 def after_worker_exit=(value) @after_worker_exit = value end |
#after_worker_fork ⇒ Object
:stopdoc:
77 78 79 |
# File 'lib/pitchfork/http_server.rb', line 77 def after_worker_fork @after_worker_fork end |
#after_worker_hard_timeout=(value) ⇒ Object (writeonly)
Sets the attribute after_worker_hard_timeout
82 83 84 |
# File 'lib/pitchfork/http_server.rb', line 82 def after_worker_hard_timeout=(value) @after_worker_hard_timeout = value end |
#after_worker_ready=(value) ⇒ Object (writeonly)
Sets the attribute after_worker_ready
82 83 84 |
# File 'lib/pitchfork/http_server.rb', line 82 def after_worker_ready=(value) @after_worker_ready = value end |
#after_worker_timeout=(value) ⇒ Object (writeonly)
Sets the attribute after_worker_timeout
82 83 84 |
# File 'lib/pitchfork/http_server.rb', line 82 def after_worker_timeout=(value) @after_worker_timeout = value end |
#app ⇒ Object
:stopdoc:
77 78 79 |
# File 'lib/pitchfork/http_server.rb', line 77 def app @app end |
#before_fork ⇒ Object
:stopdoc:
77 78 79 |
# File 'lib/pitchfork/http_server.rb', line 77 def before_fork @before_fork end |
#before_worker_exit=(value) ⇒ Object (writeonly)
Sets the attribute before_worker_exit
82 83 84 |
# File 'lib/pitchfork/http_server.rb', line 82 def before_worker_exit=(value) @before_worker_exit = value end |
#children ⇒ Object
:stopdoc:
77 78 79 |
# File 'lib/pitchfork/http_server.rb', line 77 def children @children end |
#cleanup_timeout ⇒ Object
:stopdoc:
77 78 79 |
# File 'lib/pitchfork/http_server.rb', line 77 def cleanup_timeout @cleanup_timeout end |
#config ⇒ Object
:stopdoc:
77 78 79 |
# File 'lib/pitchfork/http_server.rb', line 77 def config @config end |
#default_middleware ⇒ Object
:stopdoc:
77 78 79 |
# File 'lib/pitchfork/http_server.rb', line 77 def default_middleware @default_middleware end |
#early_hints ⇒ Object
:stopdoc:
77 78 79 |
# File 'lib/pitchfork/http_server.rb', line 77 def early_hints @early_hints end |
#listener_opts ⇒ Object
:stopdoc:
77 78 79 |
# File 'lib/pitchfork/http_server.rb', line 77 def listener_opts @listener_opts end |
#logger ⇒ Object
Returns the value of attribute logger.
85 86 87 |
# File 'lib/pitchfork/http_server.rb', line 85 def logger @logger end |
#orig_app ⇒ Object
:stopdoc:
77 78 79 |
# File 'lib/pitchfork/http_server.rb', line 77 def orig_app @orig_app end |
#ready_pipe ⇒ Object
:stopdoc:
77 78 79 |
# File 'lib/pitchfork/http_server.rb', line 77 def ready_pipe @ready_pipe end |
#refork_condition=(value) ⇒ Object (writeonly)
Sets the attribute refork_condition
82 83 84 |
# File 'lib/pitchfork/http_server.rb', line 82 def refork_condition=(value) @refork_condition = value end |
#soft_timeout ⇒ Object
:stopdoc:
77 78 79 |
# File 'lib/pitchfork/http_server.rb', line 77 def soft_timeout @soft_timeout end |
#spawn_timeout ⇒ Object
:stopdoc:
77 78 79 |
# File 'lib/pitchfork/http_server.rb', line 77 def spawn_timeout @spawn_timeout end |
#timeout ⇒ Object
:stopdoc:
77 78 79 |
# File 'lib/pitchfork/http_server.rb', line 77 def timeout @timeout end |
#timeout_signal ⇒ Object
:stopdoc:
77 78 79 |
# File 'lib/pitchfork/http_server.rb', line 77 def timeout_signal @timeout_signal end |
#worker_processes ⇒ Object
:stopdoc:
77 78 79 |
# File 'lib/pitchfork/http_server.rb', line 77 def worker_processes @worker_processes end |
Instance Method Details
#check_client_connection ⇒ Object
447 448 449 |
# File 'lib/pitchfork/http_server.rb', line 447 def check_client_connection Pitchfork::HttpParser.check_client_connection end |
#check_client_connection=(bool) ⇒ Object
451 452 453 |
# File 'lib/pitchfork/http_server.rb', line 451 def check_client_connection=(bool) Pitchfork::HttpParser.check_client_connection = bool end |
#client_body_buffer_size ⇒ Object
439 440 441 |
# File 'lib/pitchfork/http_server.rb', line 439 def client_body_buffer_size Pitchfork::TeeInput.client_body_buffer_size end |
#client_body_buffer_size=(bytes) ⇒ Object
443 444 445 |
# File 'lib/pitchfork/http_server.rb', line 443 def client_body_buffer_size=(bytes) Pitchfork::TeeInput.client_body_buffer_size = bytes end |
#join ⇒ Object
monitors children and receives signals forever (or until a termination signal is sent). This handles signals one-at-a-time time and we’ll happily drop signals in case somebody is signalling us too often.
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 |
# File 'lib/pitchfork/http_server.rb', line 294 def join @respawn = true proc_name role: 'monitor', status: START_CTX[:argv].join(' ') logger.info "master process ready" # test_exec.rb relies on this message if @ready_pipe begin @ready_pipe.syswrite($$.to_s) rescue => e logger.warn("grandparent died too soon?: #{e.} (#{e.class})") end @ready_pipe = @ready_pipe.close rescue nil end while true begin if monitor_loop == StopIteration break end rescue => e Pitchfork.log_error(@logger, "master loop error", e) end end stop # gracefully shutdown all workers on our way out logger.info "master complete status=#{@exit_status}" @exit_status end |
#listen(address, opt = {}.merge(listener_opts[address] || {})) ⇒ Object
add a given address to the listeners
set, idempotently Allows workers to add a private, per-process listener via the after_worker_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.
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 |
# File 'lib/pitchfork/http_server.rb', line 260 def listen(address, opt = {}.merge(listener_opts[address] || {})) address = config.(address) return if String === address && listener_names.include?(address) delay = opt[:delay] || 0.5 tries = opt[:tries] || 5 begin io = bind_listen(address, opt) unless TCPServer === io || UNIXServer === io io.autoclose = false io = server_cast(io) end logger.info "listening on addr=#{sock_name(io)} fd=#{io.fileno}" Info.keep_io(io) LISTENERS << io io rescue Errno::EADDRINUSE => err logger.error "adding listener failed addr=#{address} (in use)" raise err if tries == 0 tries -= 1 logger.error "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 |
#listeners=(listeners) ⇒ Object
replaces current listener set with listeners
. This will close the socket if it will not exist in the new listener set
222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 |
# File 'lib/pitchfork/http_server.rb', line 222 def listeners=(listeners) cur_names, dead_names = [], [] listener_names.each do |name| if name.start_with?('/') # 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, listener_opts[sock_name(io)]) false end end (set_names - cur_names).each { |addr| listen(addr) } end |
#monitor_loop(sleep = true) ⇒ Object
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 347 348 349 350 351 352 353 354 355 356 357 358 359 360 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 |
# File 'lib/pitchfork/http_server.rb', line 322 def monitor_loop(sleep = true) reap_all_workers if REFORKING_AVAILABLE && @respawn && @children.molds.empty? logger.info("No mold alive, shutting down") @exit_status = 1 @sig_queue << :TERM @respawn = false end case = @sig_queue.shift when nil # avoid murdering workers after our master process (or the # machine) comes out of suspend/hibernation if (@last_check + @timeout) >= (@last_check = Pitchfork.time_now) sleep_time = murder_lazy_workers else sleep_time = @timeout/2.0 + 1 @logger.debug("waiting #{sleep_time}s after suspend/hibernation") end if @respawn maintain_worker_count restart_outdated_workers if REFORKING_AVAILABLE end master_sleep(sleep_time) if sleep when :QUIT, :TERM # graceful shutdown SharedMemory.shutting_down! logger.info "#{} received, starting graceful shutdown" return StopIteration when :INT # immediate shutdown SharedMemory.shutting_down! logger.info "#{} received, starting immediate shutdown" stop(false) return StopIteration when :USR2 # trigger a promotion if @respawn trigger_refork else logger.error "Can't trigger a refork as the server is shutting down" end when :TTIN @respawn = true self.worker_processes += 1 when :TTOU self.worker_processes -= 1 if self.worker_processes > 0 when Message::WorkerSpawned worker = @children.update() # TODO: should we send a message to the worker to acknowledge? logger.info "worker=#{worker.nr} pid=#{worker.pid} gen=#{worker.generation} registered" when Message::MoldSpawned new_mold = @children.update() logger.info("mold pid=#{new_mold.pid} gen=#{new_mold.generation} spawned") when Message::MoldReady old_molds = @children.molds new_mold = @children.update() logger.info("mold pid=#{new_mold.pid} gen=#{new_mold.generation} ready") old_molds.each do |old_mold| logger.info("Terminating old mold pid=#{old_mold.pid} gen=#{old_mold.generation}") old_mold.soft_kill(:TERM) end else logger.error("Unexpected message in sig_queue #{.inspect}") logger.error(@sig_queue.inspect) end end |
#rewindable_input ⇒ Object
430 431 432 |
# File 'lib/pitchfork/http_server.rb', line 430 def rewindable_input Pitchfork::HttpParser.input_class.method_defined?(:rewind) end |
#rewindable_input=(bool) ⇒ Object
434 435 436 437 |
# File 'lib/pitchfork/http_server.rb', line 434 def rewindable_input=(bool) Pitchfork::HttpParser.input_class = bool ? Pitchfork::TeeInput : Pitchfork::StreamInput end |
#start(sync = true) ⇒ Object
Runs the thing. Returns self so you can run join on it
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 |
# File 'lib/pitchfork/http_server.rb', line 178 def start(sync = true) Pitchfork.enable_child_subreaper # noop if not supported # This socketpair is used to wake us up from select(2) in #join when signals # are trapped. See trap_deferred. # It's also used by newly spawned children to send their soft_signal pipe # to the master when they are spawned. @control_socket.replace(Pitchfork.socketpair) Info.keep_ios(@control_socket) @master_pid = $$ # setup signal handlers before writing pid file in case people get # trigger happy and send signals as soon as the pid file exists. # Note that signals don't actually get handled until the #join method @queue_sigs.each { |sig| trap(sig) { @sig_queue << sig; awaken_master } } trap(:CHLD) { awaken_master } if REFORKING_AVAILABLE spawn_initial_mold wait_for_pending_workers unless @children.mold raise BootFailure, "The initial mold failed to boot" end else build_app! bind_listeners! after_mold_fork.call(self, Worker.new(nil, pid: $$).promoted!(@spawn_timeout)) end if sync spawn_missing_workers # We could just return here as we'd register them later in #join. # However a good part of the test suite assumes #start only return # once all initial workers are spawned. wait_for_pending_workers end @after_monitor_ready&.call(self) self end |
#stop(graceful = true) ⇒ Object
Terminates all workers, but does not exit master process
390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 |
# File 'lib/pitchfork/http_server.rb', line 390 def stop(graceful = true) proc_name role: 'monitor', status: 'shutting down' @respawn = false SharedMemory.shutting_down! wait_for_pending_workers self.listeners = [] limit = Pitchfork.time_now + timeout until @children.empty? || Pitchfork.time_now > limit if graceful @children.soft_kill_all(:TERM) else @children.hard_kill_all(:INT) end if monitor_loop(false) == StopIteration return StopIteration end end @children.each do |child| if child.pid @children.hard_kill(@timeout_signal.call(child.pid), child) end end @promotion_lock.unlink end |
#worker_exit(worker) ⇒ Object
416 417 418 419 420 421 422 423 424 425 426 427 428 |
# File 'lib/pitchfork/http_server.rb', line 416 def worker_exit(worker) logger.info "worker=#{worker.nr} pid=#{worker.pid} gen=#{worker.generation} exiting" proc_name status: "exiting" if @before_worker_exit begin @before_worker_exit.call(self, worker) rescue => error Pitchfork.log_error(logger, "before_worker_exit error", error) end end Process.exit end |