Merge pull request #1579 from puma/schneems/waiting

[close #1577] Negative Backpressure Metric

Author: schneems

lib/puma/cluster.rb

@@ -292,7 +292,8 @@ def worker(index, master)
           begin
             b = server.backlog || 0
             r = server.running || 0
-            payload = %Q!#{base_payload}{ "backlog":#{b}, "running":#{r} }\n!
+            t = server.pool_capacity || 0
+            payload = %Q!#{base_payload}{ "backlog":#{b}, "running":#{r}, "pool_capacity":#{t} }\n!
             io << payload
           rescue IOError
             Thread.current.purge_interrupt_queue if Thread.current.respond_to? :purge_interrupt_queue

lib/puma/server.rb

@@ -168,6 +168,18 @@ def running
       @thread_pool and @thread_pool.spawned
     end
 
+
+    # This number represents the number of requests that
+    # the server is capable of taking right now.
+    #
+    # For example if the number is 5 then it means
+    # there are 5 threads sitting idle ready to take
+    # a request. If one request comes in, then the
+    # value would be 4 until it finishes processing.
+    def pool_capacity
+      @thread_pool and @thread_pool.pool_capacity
+    end
+
     # Lopez Mode == raw tcp apps
 
     def run_lopez_mode(background=true)

lib/puma/single.rb

@@ -14,7 +14,8 @@ class Single < Runner
     def stats
       b = @server.backlog || 0
       r = @server.running || 0
-      %Q!{ "backlog": #{b}, "running": #{r} }!
+      t = @server.pool_capacity || 0
+      %Q!{ "backlog": #{b}, "running": #{r}, "pool_capacity": #{t} }!
     end
 
     def restart

lib/puma/thread_pool.rb

@@ -58,7 +58,7 @@ def initialize(min, max, *extra, &block)
       @clean_thread_locals = false
     end
 
-    attr_reader :spawned, :trim_requested
+    attr_reader :spawned, :trim_requested, :waiting
     attr_accessor :clean_thread_locals
 
     def self.clean_thread_locals
@@ -73,6 +73,10 @@ def backlog
       @mutex.synchronize { @todo.size }
     end
 
+    def pool_capacity
+      waiting + (@max - spawned)
+    end
+
     # :nodoc:
     #
     # Must be called with @mutex held!

test/test_cli.rb

@@ -56,8 +56,8 @@ def test_control_for_tcp
     s = TCPSocket.new "127.0.0.1", 9877
     s << "GET /stats HTTP/1.0\r\n\r\n"
     body = s.read
-    assert_equal '{ "backlog": 0, "running": 0 }', body.split(/\r?\n/).last
-    assert_equal '{ "backlog": 0, "running": 0 }', Puma.stats
+    assert_equal '{ "backlog": 0, "running": 0, "pool_capacity": 16 }', body.split(/\r?\n/).last
+    assert_equal '{ "backlog": 0, "running": 0, "pool_capacity": 16 }', Puma.stats
 
     cli.launcher.stop
     t.join
@@ -95,7 +95,7 @@ def test_control_clustered
     s = UNIXSocket.new @tmp_path
     s << "GET /stats HTTP/1.0\r\n\r\n"
     body = s.read
-    assert_match(/\{ "workers": 2, "phase": 0, "booted_workers": 2, "old_workers": 0, "worker_status": \[\{ "pid": \d+, "index": 0, "phase": 0, "booted": true, "last_checkin": "[^"]+", "last_status": \{ "backlog":0, "running":2 \} \},\{ "pid": \d+, "index": 1, "phase": 0, "booted": true, "last_checkin": "[^"]+", "last_status": \{ "backlog":0, "running":2 \} \}\] \}/, body.split("\r\n").last)
+    assert_match(/\{ "workers": 2, "phase": 0, "booted_workers": 2, "old_workers": 0, "worker_status": \[\{ "pid": \d+, "index": 0, "phase": 0, "booted": true, "last_checkin": "[^"]+", "last_status": \{ "backlog":0, "running":2, "pool_capacity":2 \} \},\{ "pid": \d+, "index": 1, "phase": 0, "booted": true, "last_checkin": "[^"]+", "last_status": \{ "backlog":0, "running":2, "pool_capacity":2 \} \}\] \}/, body.split("\r\n").last)
 
     cli.launcher.stop
     t.join
@@ -118,7 +118,7 @@ def test_control
     s << "GET /stats HTTP/1.0\r\n\r\n"
     body = s.read
 
-    assert_equal '{ "backlog": 0, "running": 0 }', body.split("\r\n").last
+    assert_equal '{ "backlog": 0, "running": 0, "pool_capacity": 16 }', body.split("\r\n").last
 
     cli.launcher.stop
     t.join