Eclipse
/
jetty.project
mirror of https://github.com/jetty/jetty.project.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fixes #6652 - Improve ReservedThreadExecutor dump. (#6653) 

Fixes #6652 - Improve ReservedThreadExecutor dump.

Filtering out non-reserved threads in dump() and doStop().

Signed-off-by: Simone Bordet <simone.bordet@gmail.com>
Co-authored-by: Greg Wilkins <gregw@webtide.com>
(cherry picked from commit b2a023675c)
This commit is contained in:
Simone Bordet 2021-08-24 23:12:08 +02:00
parent 18b66d1d47
commit e8beb58baf
3 changed files with 37 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
jetty-util/src/main/java/org/eclipse/jetty/util/thread/ReservedThreadExecutor.java
View File

@ -14,6 +14,7 @@
package org.eclipse.jetty.util.thread;
import java.io.IOException;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
@ -21,6 +22,7 @@ import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import org.eclipse.jetty.util.AtomicBiInteger;
import org.eclipse.jetty.util.ProcessorUtils;
@ -38,14 +40,14 @@ import static org.eclipse.jetty.util.AtomicBiInteger.getLo;
/**
* An Executor using pre-allocated/reserved Threads from a wrapped Executor.
* <p>Calls to {@link #execute(Runnable)} on a {@link ReservedThreadExecutor} will either succeed
* with a Thread immediately being assigned the Runnable task, or fail if no Thread is
* available.
* <p>Threads are reserved lazily, with a new reserved threads being allocated from the
* <p>A TryExecutor using pre-allocated/reserved threads from an external Executor.</p>
* <p>Calls to {@link #tryExecute(Runnable)} on ReservedThreadExecutor will either
* succeed with a reserved thread immediately being assigned the task, or fail if
* no reserved thread is available.</p>
* <p>Threads are reserved lazily, with new reserved threads being allocated from the external
* {@link Executor} passed to the constructor. Whenever 1 or more reserved threads have been
* idle for more than {@link #getIdleTimeoutMs()} then one reserved thread will return to
* the executor.
* the external Executor.</p>
*/
@ManagedObject("A pool for reserved threads")
public class ReservedThreadExecutor extends AbstractLifeCycle implements TryExecutor, Dumpable
@ -62,7 +64,7 @@ public class ReservedThreadExecutor extends AbstractLifeCycle implements TryExec
@Override
public String toString()
{
return "STOP!";
return "STOP";
}
};
@ -72,7 +74,6 @@ public class ReservedThreadExecutor extends AbstractLifeCycle implements TryExec
private final SynchronousQueue<Runnable> _queue = new SynchronousQueue<>(false);
private final AtomicBiInteger _count = new AtomicBiInteger(); // hi=pending; lo=size;
private final AtomicLong _lastEmptyTime = new AtomicLong(System.nanoTime());
private ThreadPoolBudget.Lease _lease;
private long _idleTimeNanos = DEFAULT_IDLE_TIMEOUT;
@ -175,20 +176,25 @@ public class ReservedThreadExecutor extends AbstractLifeCycle implements TryExec
super.doStop();
// Offer STOP task to all waiting reserved threads.
for (int i = _count.getAndSetLo(-1); i-- > 0;)
// Mark this instance as stopped.
int size = _count.getAndSetLo(-1);
// Offer the STOP task to all waiting reserved threads.
for (int i = 0; i < size; ++i)
{
// yield to wait for any reserved threads that have incremented the size but not yet polled
// Yield to wait for any reserved threads that
// have incremented the size but not yet polled.
Thread.yield();
_queue.offer(STOP);
}
// Interrupt any reserved thread missed the offer so it doesn't wait too long.
for (ReservedThread reserved : _threads)
{
Thread thread = reserved._thread;
if (thread != null)
thread.interrupt();
}
// Interrupt any reserved thread missed the offer,
// so they do not wait for the whole idle timeout.
_threads.stream()
.filter(ReservedThread::isReserved)
.map(t -> t._thread)
.filter(Objects::nonNull)
.forEach(Thread::interrupt);
_threads.clear();
_count.getAndSetHi(0);
}
@ -264,13 +270,16 @@ public class ReservedThreadExecutor extends AbstractLifeCycle implements TryExec
public void dump(Appendable out, String indent) throws IOException
{
Dumpable.dumpObjects(out, indent, this,
new DumpableCollection("reserved", _threads));
new DumpableCollection("threads",
_threads.stream()
.filter(ReservedThread::isReserved)
.collect(Collectors.toList())));
}
@Override
public String toString()
{
return String.format("%s@%x{s=%d/%d,p=%d}",
return String.format("%s@%x{reserved=%d/%d,pending=%d}",
getClass().getSimpleName(),
hashCode(),
_count.getLo(),
@ -293,6 +302,11 @@ public class ReservedThreadExecutor extends AbstractLifeCycle implements TryExec
private volatile State _state = State.PENDING;
private volatile Thread _thread;
private boolean isReserved()
{
return _state == State.RESERVED;
}
private Runnable reservedWait()
{
if (LOG.isDebugEnabled())
@ -321,7 +335,6 @@ public class ReservedThreadExecutor extends AbstractLifeCycle implements TryExec
}
_state = size >= 0 ? State.IDLE : State.STOPPED;
return STOP;
}
catch (InterruptedException e)
{

4
jetty-util/src/test/java/org/eclipse/jetty/util/thread/QueuedThreadPoolTest.java
View File

@ -831,7 +831,7 @@ public class QueuedThreadPoolTest extends AbstractThreadPoolTest
dump = pool.dump();
assertThat(count(dump, " - STARTED"), is(2));
assertThat(dump, containsString(",3<=3<=4,i=2,r=2,q=0"));
assertThat(dump, containsString("s=0/2"));
assertThat(dump, containsString("reserved=0/2"));
assertThat(dump, containsString("[ReservedThreadExecutor@"));
assertThat(count(dump, " IDLE"), is(2));
assertThat(count(dump, " WAITING"), is(1));
@ -846,7 +846,7 @@ public class QueuedThreadPoolTest extends AbstractThreadPoolTest
dump = pool.dump();
assertThat(count(dump, " - STARTED"), is(2));
assertThat(dump, containsString(",3<=3<=4,i=1,r=2,q=0"));
assertThat(dump, containsString("s=1/2"));
assertThat(dump, containsString("reserved=1/2"));
assertThat(dump, containsString("[ReservedThreadExecutor@"));
assertThat(count(dump, " IDLE"), is(1));
assertThat(count(dump, " WAITING"), is(1));

1
jetty-util/src/test/java/org/eclipse/jetty/util/thread/ReservedThreadExecutorTest.java
View File

@ -359,6 +359,5 @@ public class ReservedThreadExecutorTest
assertThat(usedReserved.get(), greaterThan(0));
assertThat(usedReserved.get() + usedPool.get(), is(LOOPS));
// System.err.printf("reserved=%d pool=%d total=%d%n", usedReserved.get(), usedPool.get(), LOOPS);
}
}