From f4c59a0d222c31c7a0b2323deaeb164e760bba37 Mon Sep 17 00:00:00 2001 From: Will Jordan Date: Thu, 24 Sep 2020 07:24:34 -0700 Subject: [PATCH] Better error handling during force shutdown (#2271) Only allow `ForceShutdown` to be raised in a thread during specific areas of the connection-processing cycle (marked by `with_force_shutdown` blocks), to ensure that the raised error is always rescued and handled cleanly. Fixes an issue where the `force_shutdown_after: 0` option throws uncaught exceptions from the threadpool on shutdown. --- History.md | 1 + lib/puma/server.rb | 20 ++++++++---- lib/puma/thread_pool.rb | 22 +++++++++++-- test/test_puma_server.rb | 67 ++++++++++++++++++++-------------------- test/test_thread_pool.rb | 12 ++++--- 5 files changed, 76 insertions(+), 46 deletions(-) diff --git a/History.md b/History.md index ee351a8b36..1fccfb70c7 100644 --- a/History.md +++ b/History.md @@ -5,6 +5,7 @@ * Bugfixes * Prevent connections from entering Reactor after shutdown begins (#2377) + * Better error handling during force shutdown (#2271) * Refactor * Change Events#ssl_error signature from (error, peeraddr, peercert) to (error, ssl_socket) (#2375) diff --git a/lib/puma/server.rb b/lib/puma/server.rb index 536e546502..804011e3ca 100644 --- a/lib/puma/server.rb +++ b/lib/puma/server.rb @@ -232,7 +232,9 @@ def run(background=true) if @queue_requests process_now = client.eagerly_finish else - client.finish(@first_data_timeout) + @thread_pool.with_force_shutdown do + client.finish(@first_data_timeout) + end process_now = true end rescue MiniSSL::SSLError => e @@ -244,7 +246,7 @@ def run(background=true) client.close @events.parse_error e, client - rescue ConnectionError, EOFError => e + rescue ConnectionError, EOFError, ThreadPool::ForceShutdown => e client.close @events.connection_error e, client @@ -422,7 +424,11 @@ def process_client(client, buffer) check_for_more_data = false end - unless client.reset(check_for_more_data) + next_request_ready = @thread_pool.with_force_shutdown do + client.reset(check_for_more_data) + end + + unless next_request_ready @shutdown_mutex.synchronize do return unless @queue_requests close_socket = false @@ -435,7 +441,7 @@ def process_client(client, buffer) end # The client disconnected while we were reading data - rescue ConnectionError + rescue ConnectionError, ThreadPool::ForceShutdown # Swallow them. The ensure tries to close +client+ down # SSL handshake error @@ -638,7 +644,9 @@ def handle_request(req, lines) begin begin - status, headers, res_body = @app.call(env) + status, headers, res_body = @thread_pool.with_force_shutdown do + @app.call(env) + end return :async if req.hijacked @@ -936,7 +944,7 @@ def graceful_shutdown if @thread_pool if timeout = @options[:force_shutdown_after] - @thread_pool.shutdown timeout.to_i + @thread_pool.shutdown timeout.to_f else @thread_pool.shutdown end diff --git a/lib/puma/thread_pool.rb b/lib/puma/thread_pool.rb index 0e44f7458d..a7966bcd78 100644 --- a/lib/puma/thread_pool.rb +++ b/lib/puma/thread_pool.rb @@ -62,6 +62,8 @@ def initialize(min, max, *extra, &block) end @clean_thread_locals = false + @force_shutdown = false + @shutdown_mutex = Mutex.new end attr_reader :spawned, :trim_requested, :waiting @@ -322,6 +324,19 @@ def auto_reap!(timeout=5) @reaper.start! end + # Allows ThreadPool::ForceShutdown to be raised within the + # provided block if the thread is forced to shutdown during execution. + def with_force_shutdown + t = Thread.current + @shutdown_mutex.synchronize do + raise ForceShutdown if @force_shutdown + t[:with_force_shutdown] = true + end + yield + ensure + t[:with_force_shutdown] = false + end + # Tell all threads in the pool to exit and wait for them to finish. # Wait +timeout+ seconds then raise +ForceShutdown+ in remaining threads. # Next, wait an extra +grace+ seconds then force-kill remaining threads. @@ -356,8 +371,11 @@ def shutdown(timeout=-1) join.call(timeout) # If threads are still running, raise ForceShutdown and wait to finish. - threads.each do |t| - t.raise ForceShutdown + @shutdown_mutex.synchronize do + @force_shutdown = true + threads.each do |t| + t.raise ForceShutdown if t[:with_force_shutdown] + end end join.call(SHUTDOWN_GRACE_TIME) diff --git a/test/test_puma_server.rb b/test/test_puma_server.rb index c1b2d25187..0953d6deb1 100644 --- a/test/test_puma_server.rb +++ b/test/test_puma_server.rb @@ -949,27 +949,37 @@ def assert_does_not_allow_http_injection(app, opts = {}) end # Perform a server shutdown while requests are pending (one in app-server response, one still sending client request). - def shutdown_requests(app_delay: 2, request_delay: 1, post: false, response:, **options) + def shutdown_requests(s1_complete: true, s1_response: nil, post: false, s2_response: nil, **options) @server = Puma::Server.new @app, @events, options - server_run app: ->(_) { - sleep app_delay + mutex = Mutex.new + app_finished = ConditionVariable.new + server_run app: ->(env) { + path = env['REQUEST_PATH'] + mutex.synchronize do + app_finished.signal + app_finished.wait(mutex) if path == '/s1' + end [204, {}, []] } - s1 = send_http "GET / HTTP/1.1\r\n\r\n" + s1 = nil s2 = send_http post ? - "POST / HTTP/1.1\r\nHost: test.com\r\nContent-Type: text/plain\r\nContent-Length: 5\r\n\r\nhi!" : - "GET / HTTP/1.1\r\n" - sleep 0.1 - + "POST /s2 HTTP/1.1\r\nHost: test.com\r\nContent-Type: text/plain\r\nContent-Length: 5\r\n\r\nhi!" : + "GET /s2 HTTP/1.1\r\n" + mutex.synchronize do + s1 = send_http "GET /s1 HTTP/1.1\r\n\r\n" + app_finished.wait(mutex) + app_finished.signal if s1_complete + end @server.stop - sleep request_delay + Thread.pass until @server.instance_variable_get(:@thread_pool).instance_variable_get(:@shutdown) - s2 << "\r\n" + assert_match(s1_response, s1.gets) if s1_response - assert_match(/204/, s1.gets) + # Send s2 after shutdown begins + s2 << "\r\n" unless IO.select([s2], nil, nil, 0.1) - assert IO.select([s2], nil, nil, app_delay), 'timeout waiting for response' + assert IO.select([s2], nil, nil, 10), 'timeout waiting for response' s2_result = begin s2.gets rescue Errno::ECONNABORTED, Errno::ECONNRESET @@ -977,38 +987,27 @@ def shutdown_requests(app_delay: 2, request_delay: 1, post: false, response:, ** post ? '408' : nil end - if response - assert_match response, s2_result + if s2_response + assert_match s2_response, s2_result else assert_nil s2_result end end - # Shutdown should allow pending requests to complete. + # Shutdown should allow pending requests and app-responses to complete. def test_shutdown_requests - shutdown_requests response: /204/ - shutdown_requests response: /204/, queue_requests: false - end - - # Requests stuck longer than `first_data_timeout` should have connection closed (408 w/pending POST body). - def test_shutdown_data_timeout - shutdown_requests request_delay: 3, first_data_timeout: 2, response: nil - shutdown_requests request_delay: 3, first_data_timeout: 2, response: nil, queue_requests: false - shutdown_requests request_delay: 3, first_data_timeout: 2, response: /408/, post: true + opts = {s1_response: /204/, s2_response: /204/} + shutdown_requests(**opts) + shutdown_requests(**opts, queue_requests: false) end # Requests still pending after `force_shutdown_after` should have connection closed (408 w/pending POST body). + # App-responses still pending should return 503 (uncaught Puma::ThreadPool::ForceShutdown exception). def test_force_shutdown - shutdown_requests request_delay: 4, response: nil, force_shutdown_after: 3 - shutdown_requests request_delay: 4, response: nil, force_shutdown_after: 3, queue_requests: false - shutdown_requests request_delay: 4, response: /408/, force_shutdown_after: 3, post: true - end - - # App-responses still pending during `force_shutdown_after` should return 503 - # (uncaught Puma::ThreadPool::ForceShutdown exception). - def test_force_shutdown_app - shutdown_requests app_delay: 3, response: /503/, force_shutdown_after: 3 - shutdown_requests app_delay: 3, response: /503/, force_shutdown_after: 3, queue_requests: false + opts = {s1_complete: false, s1_response: /503/, s2_response: nil, force_shutdown_after: 0} + shutdown_requests(**opts) + shutdown_requests(**opts, queue_requests: false) + shutdown_requests(**opts, post: true, s2_response: /408/) end def test_http11_connection_header_queue diff --git a/test/test_thread_pool.rb b/test/test_thread_pool.rb index 0e6f7645ab..b25f34ea67 100644 --- a/test/test_thread_pool.rb +++ b/test/test_thread_pool.rb @@ -222,8 +222,10 @@ def test_force_shutdown_immediately pool = mutex_pool(0, 1) do begin - pool.signal - sleep + pool.with_force_shutdown do + pool.signal + sleep + end rescue Puma::ThreadPool::ForceShutdown rescued = true end @@ -248,8 +250,10 @@ def test_shutdown_with_grace rescued = [] pool = mutex_pool(2, 2) do begin - pool.signal - sleep + pool.with_force_shutdown do + pool.signal + sleep + end rescue Puma::ThreadPool::ForceShutdown rescued << Thread.current sleep