QueuedThreadPool will not reserve a thread with jobs waiting

jetty-core/jetty-http2/jetty-http2-tests/src/test/java/org/eclipse/jetty/http2/tests/BlockedWritesWithSmallThreadPoolTest.java

@@ -45,6 +45,7 @@
 import org.eclipse.jetty.util.Promise;
 import org.eclipse.jetty.util.component.LifeCycle;
 import org.eclipse.jetty.util.thread.QueuedThreadPool;
+import org.hamcrest.Matchers;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 
@@ -259,7 +260,7 @@ public void onDataAvailable(Stream stream)
 
         // Handler.handle() should have returned, make sure we block that thread.
         long delaySeconds = 10;
-        await().atMost(5, SECONDS).until(() -> serverThreads.getIdleThreads() == 1);
+        await().atMost(5, SECONDS).until(serverThreads::getIdleThreads, Matchers.equalTo(1));
         CountDownLatch serverBlockLatch = new CountDownLatch(1);
         serverThreads.execute(() ->
         {
@@ -276,7 +277,7 @@ public void onDataAvailable(Stream stream)
         if (serverThreads.getCurrentReservedThreads() != 1)
         {
             assertFalse(serverThreads.tryExecute(() -> {}));
-            await().atMost(5, SECONDS).until(() -> serverThreads.getCurrentReservedThreads() == 1);
+            await().atMost(5, SECONDS).until(serverThreads::getCurrentReservedThreads, Matchers.equalTo(1));
         }
         // Use the reserved thread for a blocking operation, simulating another blocking write.
         CountDownLatch reservedBlockLatch = new CountDownLatch(1);

jetty-core/jetty-http2/jetty-http2-tests/src/test/java/org/eclipse/jetty/http2/tests/IdleTimeoutTest.java

@@ -666,6 +666,8 @@ public void onHeaders(Stream stream, HeadersFrame frame)
     @Test
     public void testServerIdleTimeoutIsEnforcedForQueuedRequest() throws Exception
     {
+        // This test is fragile with regards to reserved thread strategies.
+
         long idleTimeout = 750;
         // Use a small thread pool to cause request queueing.
         QueuedThreadPool serverExecutor = new QueuedThreadPool(5);
@@ -755,9 +757,9 @@ public void onHeaders(Stream stream, HeadersFrame frame)
 
         // Wait for the server to finish serving requests.
         await().atMost(5, TimeUnit.SECONDS).until(handled::get, is(0));
-        assertThat(requests.get(), is(count - 1));
+        assertThat(requests.get(), is(count));
 
-        // The first 2 requests are handled normally and responded with 200, the last 2 are
+        // The first 3 requests are handled normally and responded with 200, the last one is
         // not handled due to timeout while queued, but they are responded anyway with a 500.
         await().atMost(5, TimeUnit.SECONDS).until(responses::get, is(count + 1));
     }

jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/ServerTest.java

@@ -153,7 +153,7 @@ public void testDump() throws Exception
         String dump = _server.dump();
         assertThat(dump, containsString("oejs.Server@"));
         assertThat(dump, containsString("oejut.QueuedThreadPool"));
-        assertThat(dump, containsString("+= oejut.ReservedThreadExecutor@"));
+        assertThat(dump, containsString("+= oejut.QueuedThreadPool$ReservedThreadExecutor@"));
         assertThat(dump, containsString("oeji.ArrayByteBufferPool@"));
         assertThat(dump, containsString("+- System Properties size="));
         assertThat(dump, containsString("+> java.home: "));

jetty-core/jetty-util/src/main/java/org/eclipse/jetty/util/thread/QueuedThreadPool.java

@@ -229,7 +229,7 @@ protected void doStart() throws Exception
             }
             else
             {
-                ReservedThreadExecutor rte = new ReservedThreadExecutor(this, _reservedThreads);
+                ReservedThreadExecutor rte = newReservedThreadExecutor();
                 rte.setIdleTimeout(_idleTimeout, TimeUnit.MILLISECONDS);
                 _tryExecutor = rte;
             }
@@ -244,6 +244,11 @@ protected void doStart() throws Exception
         ensureThreads();
     }
 
+    protected ReservedThreadExecutor newReservedThreadExecutor()
+    {
+        return new ReservedThreadExecutor();
+    }
+
     @Override
     protected void doStop() throws Exception
     {
@@ -265,8 +270,10 @@ protected void doStop() throws Exception
         {
             // Fill the job queue with noop jobs to wakeup idle threads.
             for (int i = 0; i < threads; ++i)
+            {
                 if (!jobs.offer(NOOP))
                     break;
+            }
 
             // try to let jobs complete naturally for half our stop time
             joinThreads(NanoTime.now() + TimeUnit.MILLISECONDS.toNanos(timeout) / 2);
@@ -679,7 +686,8 @@ public int getReadyThreads()
     @ManagedAttribute("number of threads leased for use by jetty components")
     public int getLeasedThreads()
     {
-        return getMaxLeasedThreads() - getMaxReservedThreads();
+        ThreadPoolBudget budget = _budget;
+        return budget == null ? 0 : budget.getLeasedThreads();
     }
 
     /**
@@ -694,8 +702,7 @@ public int getLeasedThreads()
     @ManagedAttribute("maximum number of threads leased to internal components")
     public int getMaxLeasedThreads()
     {
-        ThreadPoolBudget budget = _budget;
-        return budget == null ? 0 : budget.getLeasedThreads();
+        return getLeasedThreads();
     }
 
     /**
@@ -723,7 +730,7 @@ public int getIdleThreads()
     @ManagedAttribute("number of threads executing internal and unleased jobs")
     public int getBusyThreads()
     {
-        return getThreads() - getReadyThreads();
+        return Math.max(0, getThreads() - getIdleThreads());
     }
 
     /**
@@ -735,7 +742,7 @@ public int getBusyThreads()
     @ManagedAttribute("number of threads executing unleased jobs")
     public int getUtilizedThreads()
     {
-        return getThreads() - getLeasedThreads() - getReadyThreads();
+        return Math.max(0, getThreads() - getIdleThreads() - getLeasedThreads() + getReservedThreads() - getCurrentReservedThreads());
     }
 
     /**
@@ -745,8 +752,7 @@ public int getUtilizedThreads()
      * @deprecated The term available can be applied to either idle or reserved threads, but not both.
      * Use {@link #getCurrentReservedThreads()} or {@link #getIdleThreads()} instead.
      */
-    @ManagedAttribute("maximum number of threads available to run unleased jobs (deprecated")
-    @Deprecated(forRemoval = true, since = "12.1.0")
+    @ManagedAttribute("maximum number of threads available to run unleased jobs")
     public int getMaxAvailableThreads()
     {
         return getMaxThreads() - getLeasedThreads();
@@ -764,7 +770,10 @@ public int getMaxAvailableThreads()
     @ManagedAttribute("utilization rate of threads executing unleased jobs")
     public double getUtilizationRate()
     {
-        return (double)getUtilizedThreads() / (getMaxThreads() - getLeasedThreads());
+        int maxAvailableThreads = getMaxAvailableThreads();
+        if (maxAvailableThreads <= 0)
+            return 0.0D;
+        return (double)getUtilizedThreads() / maxAvailableThreads;
     }
 
     /**
@@ -1135,7 +1144,7 @@ public String toString()
         int idle = Math.max(0, AtomicBiInteger.getLo(count));
         int queue = getQueueSize();
 
-        return String.format("%s[%s]@%x{%s,%d<=%d<=%d,i=%d,r=%d,t=%dms,q=%d}[%s]",
+        return String.format("%s[%s]@%x{%s,%d<=%d<=%d,i=%d,r=%d/%d,t=%dms,q=%d}[%s]",
             TypeUtil.toShortName(getClass()),
             _name,
             hashCode(),
@@ -1144,6 +1153,7 @@ public String toString()
             threads,
             getMaxThreads(),
             idle,
+            getCurrentReservedThreads(),
             getReservedThreads(),
             NanoTime.millisUntil(_evictThreshold.get()),
             queue,
@@ -1239,4 +1249,43 @@ private void doRunJob(Runnable job)
             }
         }
     }
+
+    /**
+     * A ReservedThreadExecutor that will not reserve a thread if the QTP has jobs waiting
+     */
+    protected class ReservedThreadExecutor extends org.eclipse.jetty.util.thread.ReservedThreadExecutor
+    {
+        public ReservedThreadExecutor()
+        {
+            super(QueuedThreadPool.this, QueuedThreadPool.this._reservedThreads);
+        }
+
+        public ReservedThreadExecutor(Executor executor, int capacity, int minSize, int maxPending)
+        {
+            super(executor, capacity, minSize, maxPending);
+        }
+
+        @Override
+        protected void startReservedThread()
+        {
+            // If we have more jobs queued than the number of pending reserved threads,
+            // then we have some real jobs in the queue.
+            // So we will not start a reserved thread as it may stop a real job from being executed.
+            if (isTaskWaiting())
+                return;
+            super.startReservedThread();
+        }
+
+        @Override
+        protected boolean isTaskWaiting()
+        {
+            long counts = _counts.get();
+            int threads = Math.max(0, AtomicBiInteger.getHi(counts));
+            if (threads < getMaxThreads())
+                return false;
+            int queueSize = Math.max(0, -AtomicBiInteger.getLo(counts));
+            int pending = getPending();
+            return pending >= 0 && queueSize > pending;
+        }
+    }
 }

jetty-core/jetty-util/src/main/java/org/eclipse/jetty/util/thread/ReservedThreadExecutor.java

@@ -149,11 +149,10 @@ public int getAvailable()
         return _threads.size();
     }
 
-    @ManagedAttribute(value = "pending reserved threads (deprecated)", readonly = true)
-    @Deprecated
+    @ManagedAttribute(value = "pending reserved threads", readonly = true)
     public int getPending()
     {
-        return 0;
+        return _pending.getOpaque();
     }
 
     @ManagedAttribute(value = "idle timeout in ms", readonly = true)
@@ -225,9 +224,9 @@ public boolean tryExecute(Runnable task)
         return false;
     }
 
-    private void startReservedThread()
+    protected void startReservedThread()
     {
-        if (_maxPending > 0 && _pending.incrementAndGet() >= _maxPending)
+        if (_pending.incrementAndGet() > _maxPending && _maxPending > 0)
         {
             _pending.decrementAndGet();
             return;
@@ -273,7 +272,8 @@ public void run()
             _thread = Thread.currentThread();
             try
             {
-                _pending.decrementAndGet();
+                if (_pending.decrementAndGet() < 0)
+                    System.err.println("XXXXXXXXXXXXXXXXX " + this);
 
                 while (true)
                 {
@@ -321,7 +321,10 @@ public void run()
                         // clear interrupted status between reserved thread iterations
                         if (Thread.interrupted() && LOG.isDebugEnabled())
                             LOG.debug("{} was interrupted", _thread);
+
                     }
+                    if (isTaskWaiting())
+                        return;
                 }
             }
             catch (Throwable t)
@@ -386,4 +389,14 @@ public String toString()
                 _thread);
         }
     }
+
+    /**
+     * Test if there is a better task for a thread to do other than to become a reserved thread?
+     *
+     * @return true if a thread could be used to do something better than be reserved.
+     */
+    protected boolean isTaskWaiting()
+    {
+        return false;
+    }
 }

jetty-core/jetty-util/src/main/java/org/eclipse/jetty/util/thread/ThreadIdPool.java

@@ -247,9 +247,10 @@ public void dump(Appendable out, String indent) throws IOException
     @Override
     public String toString()
     {
-        return String.format("%s@%x{capacity=%d}",
+        return String.format("%s@%x{size=%d/%d}",
             TypeUtil.toShortName(getClass()),
             hashCode(),
+            size(),
             capacity());
     }
 }