Allow QueuedThreadPool to implement ExecutorService, add a join method overload with timeout
- Dominant language
- Java
- Stars
- 4.1k
- Forks
- 2k
- Avg merge
- 3d 56m
- Merged PRs (30d)
- 48
Description
**Jetty version(s)**
12.0.15
**Enhancement Description**
**JDK's [ThreadPoolExecutor is severely flawed](https://stackoverflow.com/questions/27987724/threadpoolexecutor-similar-to-executors-cachedthreadpool-but-with-max-threads-an#comment128085631_27987867) by design.** It doesn't allow using a queue only for overflowing tasks, and it consistently creates new threads instead of reusing them from the pool (when their count is less than core, or when the queue has rejected them - so an [Apache Tomcat's TaskQueue workaround](https://github.com/apache/tomcat/blob/fc1448c6acda5dcb5b3b5a043af53c3f84fcfad0/java/org/apache/tomcat/util/threads/TaskQueue.java#L69) has the same bug).
We consider the advantages of **migrating its usages to a superior `QueuedThreadPool` implementation**. It implements JDK's `Executor`, too. However, it doesn't implement JDK's `ExecutorService` interface. A useful `java.util.concurrent.AbstractExecutorService` abstract class exists, however, Java doesn't allow multiple inheritance.
* 📔 Note: compared to JDK's TPE:
* It has better heuristics of creating threads when low on threads, and can reserve them.
* It has a slightly different API, but supports statistics for JMX and is Dumpable, so key data shouldn't be missing.
* A key drawback is that it prioritizes queuing tasks over creating new threads once minThreads is exceeded, using "burst" threads more reluctantly, so minThreads must reflect normal operating load. An option to reconfigure this could possibly become important.
* It's missing a `rejectedExecutionHandler`.
My idea is to define an inner static class wrapper that extends the `AbstractExecutorService`. However, even then, there is one more blocker: if you want to implement the `awaitTermination(long timeout, TimeUnit unit)` method, you cannot use the `qtp.join()`, because it lacks the `timeout` parameter. Internally, it calls `l.await()`, which provides a different overload for `boolean await(long time, TimeUnit unit)`, but we can't use this overload ourselves due to a private `this._joinLock`.
* One solution would be if you moved the `join()`'s dependent logic to a new protected method, where the `l -> l.await()` would be a lambda.
* Other solution would be adding a compatible `join` overload directly.
* I'd like to also hear your opinions on supporting the `ExecutorService` interface directly.
* Similarly, I'd appreciate if you reviewed my wrapper solution and possibly documented an official workaround, supported with no risk of undocumented breaking changes.
The [QueuedThreadPool configuration docs](https://jetty.org/docs/jetty/12/programming-guide/arch/threads.html#thread-pool-configuration) also doesn't really document `setStopTimeout`, as values `>0` behave different from `0`. The difference between usages in `shutdown` and `shutdownNow` may require confirming these definitions. **A graceful shutdown may need to be supported.**
* (📔 Note: an equivalent translation of metrics between `ThreadPoolExecutor` and `QueuedThreadPool` would also be a nice addition to the docs - e.g. the `getActiveCount()` becomes `getBusyThreads()` (or `getThreads() - getIdleThreads()`, which isn't even an atomic operation). The `getQueue().size()` has protected access -> there is `getQueueSize()`.)
* Also see the next comment for a separate issue of a limited queue not being supported.
Here is an illustration of the solution:
```java
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.*;
public class MyQTPExecutorService extends QueuedThreadPool implements ExecutorService {
private final StartedExecutorServiceQTPWrapper wrapper;
public MyQTPExecutorService(){
wrapper = new StartedExecutorServiceQTPWrapper(this);
}
@Override
public void shutdown() {
wrapper.shutdown();
}
@Override
public List shutdownNow() {
return wrapper.shutdownNow();
}
@Override
public boolean isShutdown() {
return wrapper.isShutdown();
}
@Override
public boolean isTerminated() {
return wrapper.isTerminated();
}
@Override
public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
return wrapper.awaitTermination(timeout, unit);
}
@Override
public Future submit(Callable task) {
return wrapper.submit(task);
}
@Override
public Future submit(Runnable task, T result) {
return wrapper.submit(task, result);
}
@Override
public Future submit(Runnable task) {
return wrapper.submit(task);
}
@Override
public List> invokeAll(Collection> tasks) throws InterruptedException {
return wrapper.invokeAll(tasks);
}
@Override
public List> invokeAll(Collection> tasks, long timeout, TimeUnit unit) throws InterruptedException {
return wrapper.invokeAll(tasks, timeout, unit);
}
@Override
public T invokeAny(Collection> tasks) throws InterruptedException, ExecutionException {
return wrapper.invokeAny(tasks);
}
@Override
public T invokeAny(Collection> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
return wrapper.invokeAny(tasks, timeout, unit);
}
public static class StartedExecutorServiceQTPWrapper extends AbstractExecutorService {
private final MyQTPExecutorService qtp;
public StartedExecutorServiceQTPWrapper(MyQTPExecutorService qtp) {
this.qtp = qtp;
qtp.setStopTimeout(0);
try {
qtp.start();
} catch (Exception e) {
throw new RuntimeException("Failed to start wrapped QueuedThreadPool", e);
}
}
// Available option to override `newTaskFor` methods
@Override
public void shutdown() {
// TODO: implement non-blocking shutdown that doesn't interrupt tasks
// Equivalent graceful shutdown as long as `setStopTimeout(0)` is configured
try {
qtp.stop();
} catch (Exception e) {
throw new RuntimeException("Exception during wrapped QueuedThreadPool's shutdown", e);
}
}
@Override
public List shutdownNow() {
// Already configured: `qtp.setStopTimeout(0)`
shutdown();
ArrayList waitingTasks = new ArrayList<>(qtp.getQueue().size());
qtp.getQueue().drainTo(waitingTasks);
return waitingTasks;
}
@Override
public boolean isShutdown() {
return qtp.isStopped() || qtp.isStopping();
}
@Override
public boolean isTerminated() {
return qtp.isStopped();
}
@Override
public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
// TODO: extend QueuedThreadPool's join method to support timeout
qtp.join();
return isTerminated();
}
@Override
public void execute(Runnable command) {
qtp.execute(command);
}
}
}
```
Contributor guide
Assessment
This issue has not been assessed yet.