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

Issue #1851 - Improve insufficient thread warnings/errors. 

Added ExecutorSizedThreadPool, a wrapper around JDK's ThreadPoolExecutor
that implements SizedThreadPool (and therefore returns a ThreadPoolBudget).

Deprecated ExecutorThreadPool, an older version of ExecutorSizedThreadPool.
This commit is contained in:
Simone Bordet 2017-10-06 16:32:53 +02:00
parent da72bff86a
commit c93360ad9c
3 changed files with 135 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

109
jetty-util/src/main/java/org/eclipse/jetty/util/thread/ExecutorSizedThreadPool.java Normal file
View File

@ -0,0 +1,109 @@
//
// ========================================================================
// Copyright (c) 1995-2017 Mort Bay Consulting Pty. Ltd.
// ------------------------------------------------------------------------
// All rights reserved. This program and the accompanying materials
// are made available under the terms of the Eclipse Public License v1.0
// and Apache License v2.0 which accompanies this distribution.
//
// The Eclipse Public License is available at
// http://www.eclipse.org/legal/epl-v10.html
//
// The Apache License v2.0 is available at
// http://www.opensource.org/licenses/apache2.0.php
//
// You may elect to redistribute this code under either of these licenses.
// ========================================================================
//
package org.eclipse.jetty.util.thread;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.eclipse.jetty.util.component.AbstractLifeCycle;
/**
* A {@link org.eclipse.jetty.util.thread.ThreadPool.SizedThreadPool} wrapper around {@link ThreadPoolExecutor}.
*/
public class ExecutorSizedThreadPool extends AbstractLifeCycle implements ThreadPool.SizedThreadPool
{
private final ThreadPoolExecutor executor;
public ExecutorSizedThreadPool()
{
this(new ThreadPoolExecutor(8, 200, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>()));
}
public ExecutorSizedThreadPool(ThreadPoolExecutor executor)
{
this.executor = executor;
}
@Override
public int getMinThreads()
{
return executor.getCorePoolSize();
}
@Override
public int getMaxThreads()
{
return executor.getMaximumPoolSize();
}
@Override
public void setMinThreads(int threads)
{
executor.setCorePoolSize(threads);
}
@Override
public void setMaxThreads(int threads)
{
executor.setMaximumPoolSize(threads);
}
@Override
public int getThreads()
{
return executor.getPoolSize();
}
@Override
public int getIdleThreads()
{
return getThreads() - executor.getActiveCount();
}
@Override
public void execute(Runnable command)
{
executor.execute(command);
}
@Override
public boolean isLowOnThreads()
{
return getThreads() == getMaxThreads() && executor.getQueue().size() >= getIdleThreads();
}
@Override
protected void doStop() throws Exception
{
executor.shutdownNow();
}
@Override
public void join() throws InterruptedException
{
executor.awaitTermination(getStopTimeout(), TimeUnit.MILLISECONDS);
}
@Override
public ThreadPoolBudget getThreadPoolBudget()
{
return new ThreadPoolBudget(this);
}
}

57
jetty-util/src/main/java/org/eclipse/jetty/util/thread/ExecutorThreadPool.java
View File

@ -32,24 +32,24 @@ import org.eclipse.jetty.util.component.LifeCycle;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
/* ------------------------------------------------------------ */
/**
* Jetty ThreadPool using java 5 ThreadPoolExecutor
* This class wraps a {@link ExecutorService} as a {@link ThreadPool} and
* {@link LifeCycle} interfaces so that it may be used by the Jetty <code>org.eclipse.jetty.server.Server</code>
* {@link LifeCycle} interfaces so that it may be used by the Jetty {@code org.eclipse.jetty.server.Server}
*
* @deprecated use {@link ExecutorSizedThreadPool} instead
*/
@Deprecated
public class ExecutorThreadPool extends AbstractLifeCycle implements ThreadPool, LifeCycle
{
private static final Logger LOG = Log.getLogger(ExecutorThreadPool.class);
private final ExecutorService _executor;
/* ------------------------------------------------------------ */
public ExecutorThreadPool(ExecutorService executor)
{
_executor = executor;
}
/* ------------------------------------------------------------ */
/**
* Wraps an {@link ThreadPoolExecutor}.
* Max pool size is 256, pool thread timeout after 60 seconds and
@ -59,74 +59,70 @@ public class ExecutorThreadPool extends AbstractLifeCycle implements ThreadPool,
{
// Using an unbounded queue makes the maxThreads parameter useless
// Refer to ThreadPoolExecutor javadocs for details
this(new ThreadPoolExecutor(256, 256, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>()));
this(new ThreadPoolExecutor(256, 256, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>()));
}
/* ------------------------------------------------------------ */
/**
* Wraps an {@link ThreadPoolExecutor}.
* Max pool size is 256, pool thread timeout after 60 seconds, and core pool size is 32 when queueSize &gt;= 0.
*
* @param queueSize can be -1 for using an unbounded {@link LinkedBlockingQueue}, 0 for using a
* {@link SynchronousQueue}, greater than 0 for using a {@link ArrayBlockingQueue} of the given size.
* {@link SynchronousQueue}, greater than 0 for using a {@link ArrayBlockingQueue} of the given size.
*/
public ExecutorThreadPool(int queueSize)
{
this(queueSize < 0 ? new ThreadPoolExecutor(256, 256, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>()) :
queueSize == 0 ? new ThreadPoolExecutor(32, 256, 60, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()) :
new ThreadPoolExecutor(32, 256, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(queueSize)));
this(queueSize < 0 ? new ThreadPoolExecutor(256, 256, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>()) :
queueSize == 0 ? new ThreadPoolExecutor(32, 256, 60, TimeUnit.SECONDS, new SynchronousQueue<>()) :
new ThreadPoolExecutor(32, 256, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(queueSize)));
}
/* ------------------------------------------------------------ */
/**
* Wraps an {@link ThreadPoolExecutor} using
* an unbounded {@link LinkedBlockingQueue} is used for the jobs queue;
* @param corePoolSize must be equal to maximumPoolSize
*
* @param corePoolSize must be equal to maximumPoolSize
* @param maximumPoolSize the maximum number of threads to allow in the pool
* @param keepAliveTime the max time a thread can remain idle, in milliseconds
* @param keepAliveTime the max time a thread can remain idle, in milliseconds
*/
public ExecutorThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime)
{
this(corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit.MILLISECONDS);
}
/* ------------------------------------------------------------ */
/**
* Wraps an {@link ThreadPoolExecutor} using
* an unbounded {@link LinkedBlockingQueue} is used for the jobs queue.
* @param corePoolSize must be equal to maximumPoolSize
*
* @param corePoolSize must be equal to maximumPoolSize
* @param maximumPoolSize the maximum number of threads to allow in the pool
* @param keepAliveTime the max time a thread can remain idle
* @param unit the unit for the keepAliveTime
* @param keepAliveTime the max time a thread can remain idle
* @param unit the unit for the keepAliveTime
*/
public ExecutorThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit)
{
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, new LinkedBlockingQueue<Runnable>());
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, new LinkedBlockingQueue<>());
}
/* ------------------------------------------------------------ */
/**
* Wraps an {@link ThreadPoolExecutor}
* @param corePoolSize the number of threads to keep in the pool, even if they are idle
*
* @param corePoolSize the number of threads to keep in the pool, even if they are idle
* @param maximumPoolSize the maximum number of threads to allow in the pool
* @param keepAliveTime the max time a thread can remain idle
* @param unit the unit for the keepAliveTime
* @param workQueue the queue to use for holding tasks before they are executed
* @param keepAliveTime the max time a thread can remain idle
* @param unit the unit for the keepAliveTime
* @param workQueue the queue to use for holding tasks before they are executed
*/
public ExecutorThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue)
{
this(new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue));
}
/* ------------------------------------------------------------ */
@Override
public void execute(Runnable job)
{
_executor.execute(job);
}
/* ------------------------------------------------------------ */
public boolean dispatch(Runnable job)
{
try
@ -134,14 +130,13 @@ public class ExecutorThreadPool extends AbstractLifeCycle implements ThreadPool,
_executor.execute(job);
return true;
}
catch(RejectedExecutionException e)
catch (RejectedExecutionException e)
{
LOG.warn(e);
return false;
}
}
/* ------------------------------------------------------------ */
public int getIdleThreads()
{
if (_executor instanceof ThreadPoolExecutor)
@ -152,7 +147,6 @@ public class ExecutorThreadPool extends AbstractLifeCycle implements ThreadPool,
return -1;
}
/* ------------------------------------------------------------ */
public int getThreads()
{
if (_executor instanceof ThreadPoolExecutor)
@ -163,7 +157,6 @@ public class ExecutorThreadPool extends AbstractLifeCycle implements ThreadPool,
return -1;
}
/* ------------------------------------------------------------ */
public boolean isLowOnThreads()
{
if (_executor instanceof ThreadPoolExecutor)
@ -176,13 +169,11 @@ public class ExecutorThreadPool extends AbstractLifeCycle implements ThreadPool,
return false;
}
/* ------------------------------------------------------------ */
public void join() throws InterruptedException
{
_executor.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
}
/* ------------------------------------------------------------ */
@Override
protected void doStop() throws Exception
{

4
jetty-websocket/websocket-common/src/test/java/org/eclipse/jetty/websocket/common/io/LocalWebSocketConnection.java
View File

@ -24,7 +24,7 @@ import java.util.concurrent.Executor;
import org.eclipse.jetty.io.ByteBufferPool;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.util.thread.ExecutorThreadPool;
import org.eclipse.jetty.util.thread.ExecutorSizedThreadPool;
import org.eclipse.jetty.websocket.api.BatchMode;
import org.eclipse.jetty.websocket.api.StatusCode;
import org.eclipse.jetty.websocket.api.SuspendToken;
@ -57,7 +57,7 @@ public class LocalWebSocketConnection implements LogicalConnection, IncomingFram
{
this.id = id;
this.bufferPool = bufferPool;
this.executor = new ExecutorThreadPool();
this.executor = new ExecutorSizedThreadPool();
this.ioState.addListener(this);
}