You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by ma...@apache.org on 2020/07/17 18:18:09 UTC
[lucene-solr] branch reference_impl updated (8a78a0c -> 2ca2e3f)
This is an automated email from the ASF dual-hosted git repository.
markrmiller pushed a change to branch reference_impl
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git.
discard 8a78a0c @222 - Work on fixing the wait for async request phaser. "I love the zoom..."
new 2ca2e3f @222 - Work on fixing the wait for async request phaser. "I love the zoom..."
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (8a78a0c)
\
N -- N -- N refs/heads/reference_impl (2ca2e3f)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../org/apache/solr/client/solrj/impl/Http2SolrClient.java | 5 +++++
.../org/apache/solr/common/util/SolrQueuedThreadPool.java | 12 ++++++++++++
2 files changed, 17 insertions(+)
[lucene-solr] 01/01: @222 - Work on fixing the wait for async
request phaser. "I love the zoom..."
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
markrmiller pushed a commit to branch reference_impl
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
commit 2ca2e3f1323eb12ea44277461b331da461b4555b
Author: markrmiller@gmail.com <ma...@gmail.com>
AuthorDate: Fri Jul 17 12:49:30 2020 -0500
@222 - Work on fixing the wait for async request phaser. "I love the zoom..."
---
.../solr/client/solrj/embedded/JettyConfig.java | 9 +-
.../client/solrj/embedded/JettySolrRunner.java | 2 +-
.../org/apache/solr/cloud/RecoveryStrategy.java | 8 +-
.../org/apache/solr/update/SolrCmdDistributor.java | 8 +-
.../processor/DistributedZkUpdateProcessor.java | 9 +-
.../solr/cloud/FullSolrCloudDistribCmdsTest.java | 2 +
.../solr/client/solrj/impl/Http2SolrClient.java | 132 ++-
.../solr/common/util/SolrQueuedThreadPool.java | 1000 +++++++++++++++++++-
8 files changed, 1090 insertions(+), 80 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/client/solrj/embedded/JettyConfig.java b/solr/core/src/java/org/apache/solr/client/solrj/embedded/JettyConfig.java
index 0abec45..a199203 100644
--- a/solr/core/src/java/org/apache/solr/client/solrj/embedded/JettyConfig.java
+++ b/solr/core/src/java/org/apache/solr/client/solrj/embedded/JettyConfig.java
@@ -16,6 +16,7 @@
*/
package org.apache.solr.client.solrj.embedded;
+import org.apache.solr.common.util.SolrQueuedThreadPool;
import org.eclipse.jetty.servlet.ServletHolder;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
@@ -49,11 +50,11 @@ public class JettyConfig {
public final boolean enableProxy;
- public final QueuedThreadPool qtp;
+ public final SolrQueuedThreadPool qtp;
private JettyConfig(boolean onlyHttp1, int port, int portRetryTime , String context, boolean stopAtShutdown,
Long waitForLoadingCoresToFinishMs, Map<ServletHolder, String> extraServlets,
- Map<Class<? extends Filter>, String> extraFilters, SSLConfig sslConfig, boolean enableV2, boolean enableProxy, QueuedThreadPool qtp) {
+ Map<Class<? extends Filter>, String> extraFilters, SSLConfig sslConfig, boolean enableV2, boolean enableProxy, SolrQueuedThreadPool qtp) {
this.onlyHttp1 = onlyHttp1;
this.port = port;
this.context = context;
@@ -102,7 +103,7 @@ public class JettyConfig {
SSLConfig sslConfig = null;
int portRetryTime = 60;
boolean enableProxy;
- QueuedThreadPool qtp;
+ SolrQueuedThreadPool qtp;
public Builder useOnlyHttp1(boolean useOnlyHttp1) {
this.onlyHttp1 = useOnlyHttp1;
@@ -170,7 +171,7 @@ public class JettyConfig {
return this;
}
- public Builder withExecutor(QueuedThreadPool qtp) {
+ public Builder withExecutor(SolrQueuedThreadPool qtp) {
this.qtp = qtp;
return this;
}
diff --git a/solr/core/src/java/org/apache/solr/client/solrj/embedded/JettySolrRunner.java b/solr/core/src/java/org/apache/solr/client/solrj/embedded/JettySolrRunner.java
index b5ea632..17cbdf6 100644
--- a/solr/core/src/java/org/apache/solr/client/solrj/embedded/JettySolrRunner.java
+++ b/solr/core/src/java/org/apache/solr/client/solrj/embedded/JettySolrRunner.java
@@ -301,7 +301,7 @@ public class JettySolrRunner implements Closeable {
private void init(int port) {
- QueuedThreadPool qtp;
+ SolrQueuedThreadPool qtp;
if (config.qtp != null) {
qtp = config.qtp;
} else {
diff --git a/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java b/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
index d50f25c..755af6f 100644
--- a/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
+++ b/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
@@ -137,7 +137,7 @@ public class RecoveryStrategy implements Runnable, Closeable {
private final CoreContainer cc;
protected RecoveryStrategy(CoreContainer cc, CoreDescriptor cd, RecoveryListener recoveryListener) {
- ObjectReleaseTracker.track(this);
+ // ObjectReleaseTracker.track(this);
this.cc = cc;
this.coreName = cd.getName();
this.recoveryListener = recoveryListener;
@@ -187,8 +187,8 @@ public class RecoveryStrategy implements Runnable, Closeable {
// (even though getRecoveryOnlyHttpClient() already has them set)
final UpdateShardHandlerConfig cfg = cc.getConfig().getUpdateShardHandlerConfig();
return (new HttpSolrClient.Builder(leaderUrl)
- .withConnectionTimeout(cfg.getDistributedConnectionTimeout())
- .withSocketTimeout(cfg.getDistributedSocketTimeout())
+ .withConnectionTimeout(3)
+ .withSocketTimeout(5)
.withHttpClient(cc.getUpdateShardHandler().getDefaultHttpClient())
.markInternalRequest()
).build();
@@ -230,7 +230,7 @@ public class RecoveryStrategy implements Runnable, Closeable {
}
log.warn("Stopping recovery for core=[{}] coreNodeName=[{}]", coreName, coreZkNodeName);
- ObjectReleaseTracker.release(this);
+ //ObjectReleaseTracker.release(this);
}
final private void recoveryFailed(final SolrCore core,
diff --git a/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java b/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java
index 690f71c..2d849f0 100644
--- a/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java
+++ b/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java
@@ -80,7 +80,7 @@ public class SolrCmdDistributor implements Closeable {
assert !finished : "lifecycle sanity check";
finished = true;
- blockAndDoRetries();
+ // blockAndDoRetries();
}
public void close() {
@@ -250,14 +250,20 @@ public class SolrCmdDistributor implements Closeable {
if (t instanceof SolrException) {
error.statusCode = ((SolrException) t).code();
}
+ boolean success = false;
if (checkRetry(error)) {
log.info("Retrying distrib update on error: {}", t.getMessage());
submit(req);
+ success = true;
} else {
allErrors.add(error);
latch.countDown();
}
+ if (!success) {
+ latch.countDown();
+ }
+
}});
} catch (Exception e) {
latch.countDown();
diff --git a/solr/core/src/java/org/apache/solr/update/processor/DistributedZkUpdateProcessor.java b/solr/core/src/java/org/apache/solr/update/processor/DistributedZkUpdateProcessor.java
index b8b05d0..e6c0da8 100644
--- a/solr/core/src/java/org/apache/solr/update/processor/DistributedZkUpdateProcessor.java
+++ b/solr/core/src/java/org/apache/solr/update/processor/DistributedZkUpdateProcessor.java
@@ -226,7 +226,6 @@ public class DistributedZkUpdateProcessor extends DistributedUpdateProcessor {
params.set(DISTRIB_FROM, ZkCoreNodeProps.getCoreUrl(
zkController.getBaseUrl(), req.getCore().getName()));
cmdDistrib.distribCommit(cmd, useNodes, params);
- cmdDistrib.blockAndDoRetries();
}
}
@@ -251,7 +250,7 @@ public class DistributedZkUpdateProcessor extends DistributedUpdateProcessor {
}
// if (useNodes != null && useNodes.size() > 0) {
- cmdDistrib.blockAndDoRetries();
+ // cmdDistrib.blockAndDoRetries();
// }
}
}
@@ -554,9 +553,9 @@ public class DistributedZkUpdateProcessor extends DistributedUpdateProcessor {
}
}
- if (someReplicas) {
- cmdDistrib.blockAndDoRetries();
- }
+// if (someReplicas) {
+// cmdDistrib.blockAndDoRetries();
+// }
}
// used for deleteByQuery to get the list of nodes this leader should forward to
diff --git a/solr/core/src/test/org/apache/solr/cloud/FullSolrCloudDistribCmdsTest.java b/solr/core/src/test/org/apache/solr/cloud/FullSolrCloudDistribCmdsTest.java
index 130317c..10d21a6 100644
--- a/solr/core/src/test/org/apache/solr/cloud/FullSolrCloudDistribCmdsTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/FullSolrCloudDistribCmdsTest.java
@@ -51,6 +51,7 @@ import org.apache.solr.common.params.SolrParams;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.BeforeClass;
+import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
@@ -60,6 +61,7 @@ import org.slf4j.LoggerFactory;
* Super basic testing, no shard restarting or anything.
*/
@Slow
+@Ignore // nocommit needs work
public class FullSolrCloudDistribCmdsTest extends SolrCloudTestCase {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private static final AtomicInteger NAME_COUNTER = new AtomicInteger(1);
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/Http2SolrClient.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/Http2SolrClient.java
index 602645a..b5ca7dc 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/Http2SolrClient.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/Http2SolrClient.java
@@ -20,6 +20,8 @@ import java.io.ByteArrayOutputStream;
import java.io.Closeable;
import java.io.IOException;
import java.io.InputStream;
+import java.io.PrintWriter;
+import java.io.StringWriter;
import java.lang.invoke.MethodHandles;
import java.net.ConnectException;
import java.net.MalformedURLException;
@@ -37,7 +39,6 @@ import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Phaser;
-import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -66,10 +67,8 @@ import org.apache.solr.common.params.SolrParams;
import org.apache.solr.common.params.UpdateParams;
import org.apache.solr.common.util.Base64;
import org.apache.solr.common.util.ContentStream;
-import org.apache.solr.common.util.ExecutorUtil;
import org.apache.solr.common.util.NamedList;
import org.apache.solr.common.util.ObjectReleaseTracker;
-import org.apache.solr.common.util.SolrNamedThreadFactory;
import org.apache.solr.common.util.SolrQueuedThreadPool;
import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.client.HttpClientTransport;
@@ -92,9 +91,9 @@ import org.eclipse.jetty.http.HttpHeader;
import org.eclipse.jetty.http.HttpMethod;
import org.eclipse.jetty.http2.client.HTTP2Client;
import org.eclipse.jetty.http2.client.http.HttpClientTransportOverHTTP2;
-import org.eclipse.jetty.util.BlockingArrayQueue;
import org.eclipse.jetty.util.Fields;
import org.eclipse.jetty.util.ssl.SslContextFactory;
+import org.eclipse.jetty.util.thread.Scheduler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -123,10 +122,12 @@ public class Http2SolrClient extends SolrClient {
private static final String DEFAULT_PATH = "/select";
private static final List<String> errPath = Arrays.asList("metadata", "error-class");
private final Map<String, String> headers;
+ private final SolrHttpClientScheduler scheduler;
private volatile HttpClient httpClient;
private volatile Set<String> queryParams = Collections.emptySet();
private int idleTimeout;
+ volatile String closed = null;
private volatile ResponseParser parser = new BinaryResponseParser();
private volatile RequestWriter requestWriter = new BinaryRequestWriter();
@@ -151,6 +152,8 @@ public class Http2SolrClient extends SolrClient {
this.serverBaseUrl = serverBaseUrl;
}
+ scheduler = new SolrHttpClientScheduler("JettyHttpClientScheduler", true, null, new ThreadGroup("JettyHttpClientScheduler"), 5);
+
this.headers = builder.headers;
if (builder.idleTimeout != null && builder.idleTimeout > 0) idleTimeout = builder.idleTimeout;
@@ -162,6 +165,7 @@ public class Http2SolrClient extends SolrClient {
} else {
httpClient = builder.http2SolrClient.httpClient;
}
+ httpClient.setScheduler(scheduler);
assert ObjectReleaseTracker.track(this);
}
@@ -244,8 +248,32 @@ public class Http2SolrClient extends SolrClient {
}
public void close() {
+ if (this.closed != null) {
+ throw new AlreadyClosedException("Already closed! " + this.closed);
+ }
+ StringWriter sw = new StringWriter();
+ PrintWriter pw = new PrintWriter(sw);
+ new AlreadyClosedException("Already closed at: ").printStackTrace(pw);
+ this.closed = sw.toString();
+ try {
+ httpClientExecutor.prepareToStop();
+ } catch (Exception e) {
+ ParWork.propegateInterrupt(e);
+ throw new RuntimeException(e);
+ }
+ try {
+ scheduler.stop();
+ } catch (Exception e) {
+ ParWork.propegateInterrupt(e);
+ throw new RuntimeException(e);
+ }
// we wait for async requests, so far devs don't want to give sugar for this
- asyncTracker.waitForComplete();
+ asyncTracker.waitForCompleteFinal();
+ try {
+ httpClientExecutor.waitForStopping();
+ } catch (InterruptedException e) {
+ ParWork.propegateInterrupt(e);
+ }
try (ParWork closer = new ParWork(this, true)) {
if (closeClient) {
@@ -263,7 +291,6 @@ public class Http2SolrClient extends SolrClient {
}
closer.addCollect("http2SolrClientClose");
}
-
assert ObjectReleaseTracker.release(this);
}
@@ -356,7 +383,7 @@ public class Http2SolrClient extends SolrClient {
decorateRequest(postRequest, updateRequest);
InputStreamResponseListener responseListener = new InputStreamResponseListener();
- postRequest.send(responseListener);
+ postRequest.onRequestQueued(asyncTracker.queuedListener).send(responseListener);
boolean isXml = ClientUtils.TEXT_XML.equals(requestWriter.getUpdateContentType());
OutStream outStream = new OutStream(collection, origParams, provider, responseListener,
@@ -398,36 +425,47 @@ public class Http2SolrClient extends SolrClient {
Request req = makeRequest(solrRequest, collection);
final ResponseParser parser = solrRequest.getResponseParser() == null
? this.parser: solrRequest.getResponseParser();
-
+ if (this.closed != null) {
+ throw new AlreadyClosedException();
+ }
if (onComplete != null) {
// This async call only suitable for indexing since the response size is limited by 5MB
- req.onRequestQueued(asyncTracker.queuedListener)
- .onComplete(asyncTracker.completeListener).send(new BufferingResponseListener(5 * 1024 * 1024) {
+ req.onRequestQueued(asyncTracker.queuedListener).send(new BufferingResponseListener(5 * 1024 * 1024) {
@Override
public void onComplete(Result result) {
- if (result.isFailed()) {
- onComplete.onFailure(result.getFailure());
- return;
- }
-
- NamedList<Object> rsp;
try {
- InputStream is = getContentAsInputStream();
- assert ObjectReleaseTracker.track(is);
- rsp = processErrorsAndResponse(result.getResponse(),
- parser, is, getMediaType(), getEncoding(), isV2ApiRequest(solrRequest));
- onComplete.onSuccess(rsp);
- } catch (Exception e) {
- onComplete.onFailure(e);
+ if (result.isFailed()) {
+ onComplete.onFailure(result.getFailure());
+ return;
+ }
+
+ NamedList<Object> rsp;
+ try {
+ InputStream is = getContentAsInputStream();
+ assert ObjectReleaseTracker.track(is);
+ rsp = processErrorsAndResponse(req, result.getResponse(),
+ parser, is, getMediaType(), getEncoding(), isV2ApiRequest(solrRequest));
+ onComplete.onSuccess(rsp);
+ } catch (Exception e) {
+ onComplete.onFailure(e);
+ }
+ } finally {
+ asyncTracker.completeListener.onComplete(result);
}
}
});
return null;
} else {
try {
- InputStreamResponseListener listener = new InputStreamResponseListener();
- req.send(listener);
+ InputStreamResponseListener listener = new InputStreamResponseListener() {
+ @Override
+ public void onComplete(Result result) {
+ super.onComplete(result);
+ asyncTracker.completeListener.onComplete(result);
+ }
+ };
+ req.onRequestQueued(asyncTracker.queuedListener).send(listener);
Response response = listener.get(idleTimeout, TimeUnit.MILLISECONDS);
InputStream is = listener.getInputStream();
assert ObjectReleaseTracker.track(is);
@@ -439,7 +477,7 @@ public class Http2SolrClient extends SolrClient {
mimeType = contentType.getMimeType();
encoding = contentType.getCharset() != null? contentType.getCharset().name() : null;
}
- return processErrorsAndResponse(response, parser, is, mimeType, encoding, isV2ApiRequest(solrRequest));
+ return processErrorsAndResponse(req, response, parser, is, mimeType, encoding, isV2ApiRequest(solrRequest));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException(e);
@@ -680,7 +718,7 @@ public class Http2SolrClient extends SolrClient {
return processor == null || processor instanceof InputStreamResponseParser;
}
- private NamedList<Object> processErrorsAndResponse(Response response,
+ private NamedList<Object> processErrorsAndResponse(Request req, Response response,
final ResponseParser processor,
InputStream is,
String mimeType,
@@ -811,32 +849,31 @@ public class Http2SolrClient extends SolrClient {
return serverBaseUrl;
}
- private static class AsyncTracker {
+ private class AsyncTracker {
+ private final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ // nocommit - look at outstanding max again
private static final int MAX_OUTSTANDING_REQUESTS = 1000;
// wait for async requests
- private final Phaser phaser;
+ private final Phaser phaser = new Phaser(1) {
+ @Override
+ protected boolean onAdvance(int phase, int parties) {
+ return false;
+ }
+ };
// maximum outstanding requests left
- private final Semaphore available;
private final Request.QueuedListener queuedListener;
private final Response.CompleteListener completeListener;
AsyncTracker() {
- // TODO: what about shared instances?
- phaser = new Phaser(1);
- available = new Semaphore(MAX_OUTSTANDING_REQUESTS, false);
queuedListener = request -> {
phaser.register();
- try {
- available.acquire();
- } catch (InterruptedException ignored) {
- ParWork.propegateInterrupt(ignored);
- throw new AlreadyClosedException("Interrupted");
- }
+ if (log.isDebugEnabled()) log.debug("Request queued registered: {} arrived: {}", phaser.getRegisteredParties(), phaser.getArrivedParties());
};
completeListener = result -> {
+ if (log.isDebugEnabled()) log.debug("Request complete registered: {} arrived: {}", phaser.getRegisteredParties(), phaser.getArrivedParties());
phaser.arriveAndDeregister();
- available.release();
};
}
@@ -846,8 +883,19 @@ public class Http2SolrClient extends SolrClient {
}
public void waitForComplete() {
- phaser.arriveAndAwaitAdvance();
- phaser.arriveAndDeregister();
+ if (Http2SolrClient.this.closed != null) {
+ throw new IllegalStateException("Already closed! " + Http2SolrClient.this.closed );
+ }
+ if (log.isDebugEnabled()) log.debug("Before wait for outstanding requests registered: {} arrived: {}", phaser.getRegisteredParties(), phaser.getArrivedParties());
+ int arrival = phaser.arriveAndAwaitAdvance();
+ if (log.isDebugEnabled()) log.debug("After wait for outstanding requests registered: {} arrived: {} ourArrival#: {}", phaser.getRegisteredParties(), phaser.getArrivedParties(), arrival);
+ }
+
+ public void waitForCompleteFinal() {
+ if (log.isDebugEnabled()) log.debug("Before wait for complete final registered: {} arrived: {}", phaser.getRegisteredParties(), phaser.getArrivedParties());
+ int arrival = phaser.awaitAdvance(phaser.arriveAndDeregister());
+
+ if (log.isDebugEnabled()) log.debug("After wait for complete final registered: {} arrived: {}", phaser.getRegisteredParties(), phaser.getArrivedParties());
}
}
diff --git a/solr/solrj/src/java/org/apache/solr/common/util/SolrQueuedThreadPool.java b/solr/solrj/src/java/org/apache/solr/common/util/SolrQueuedThreadPool.java
index de92108..7e7d6c6 100644
--- a/solr/solrj/src/java/org/apache/solr/common/util/SolrQueuedThreadPool.java
+++ b/solr/solrj/src/java/org/apache/solr/common/util/SolrQueuedThreadPool.java
@@ -17,32 +17,799 @@
package org.apache.solr.common.util;
import java.io.Closeable;
+import java.io.IOException;
import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.solr.common.ParWork;
import org.apache.solr.common.util.ObjectReleaseTracker;
import org.apache.solr.common.util.SolrNamedThreadFactory;
+import org.eclipse.jetty.util.AtomicBiInteger;
+import org.eclipse.jetty.util.BlockingArrayQueue;
+import org.eclipse.jetty.util.StringUtil;
+import org.eclipse.jetty.util.annotation.ManagedAttribute;
+import org.eclipse.jetty.util.annotation.ManagedOperation;
+import org.eclipse.jetty.util.annotation.Name;
+import org.eclipse.jetty.util.component.ContainerLifeCycle;
+import org.eclipse.jetty.util.component.Dumpable;
+import org.eclipse.jetty.util.component.DumpableCollection;
+import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
+import org.eclipse.jetty.util.thread.ReservedThreadExecutor;
+import org.eclipse.jetty.util.thread.ThreadPool;
+import org.eclipse.jetty.util.thread.ThreadPoolBudget;
+import org.eclipse.jetty.util.thread.TryExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class SolrQueuedThreadPool extends QueuedThreadPool implements Closeable {
- private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private final String name;
- private volatile Error error;
- private final Object notify = new Object();
-
+public class SolrQueuedThreadPool extends ContainerLifeCycle implements ThreadFactory, ThreadPool.SizedThreadPool, Dumpable, TryExecutor, Closeable {
+ private static final org.eclipse.jetty.util.log.Logger LOG = Log.getLogger(QueuedThreadPool.class);
+ private static Runnable NOOP = () ->
+ {
+ };
+ /**
+ * Encodes thread counts:
+ * <dl>
+ * <dt>Hi</dt><dd>Total thread count or Integer.MIN_VALUE if the pool is stopping</dd>
+ * <dt>Lo</dt><dd>Net idle threads == idle threads - job queue size. Essentially if positive,
+ * this represents the effective number of idle threads, and if negative it represents the
+ * demand for more threads</dd>
+ * </dl>
+ */
+ private final AtomicBiInteger _counts = new AtomicBiInteger(Integer.MIN_VALUE, 0);
+ private final AtomicLong _lastShrink = new AtomicLong();
+ private final Set<Thread> _threads = ConcurrentHashMap.newKeySet();
+ private final Object _joinLock = new Object();
+ private final BlockingQueue<Runnable> _jobs;
+ private final ThreadGroup _threadGroup;
+ private final ThreadFactory _threadFactory;
+ private String _name = "qtp" + hashCode();
+ private int _idleTimeout;
+ private int _maxThreads;
+ private int _minThreads;
+ private int _reservedThreads = -1;
+ private TryExecutor _tryExecutor = TryExecutor.NO_TRY;
+ private int _priority = Thread.NORM_PRIORITY;
+ private boolean _daemon = false;
+ private boolean _detailedDump = false;
+ private int _lowThreadsThreshold = 1;
+ private ThreadPoolBudget _budget;
public SolrQueuedThreadPool(String name) {
- super(10000, 15,
- 15000, -1,
- null, null,
- new SolrNamedThreadFactory(name));
+ this(10000, 15,
+ 15000, -1,
+ null, null,
+ new SolrNamedThreadFactory(name));
this.name = name;
}
+ public SolrQueuedThreadPool(@Name("maxThreads") int maxThreads)
+ {
+ this(maxThreads, Math.min(8, maxThreads));
+ }
+
+ public SolrQueuedThreadPool(@Name("maxThreads") int maxThreads, @Name("minThreads") int minThreads)
+ {
+ this(maxThreads, minThreads, 60000);
+ }
+
+ public SolrQueuedThreadPool(@Name("maxThreads") int maxThreads, @Name("minThreads") int minThreads, @Name("queue") BlockingQueue<Runnable> queue)
+ {
+ this(maxThreads, minThreads, 60000, -1, queue, null);
+ }
+
+ public SolrQueuedThreadPool(@Name("maxThreads") int maxThreads, @Name("minThreads") int minThreads, @Name("idleTimeout") int idleTimeout)
+ {
+ this(maxThreads, minThreads, idleTimeout, null);
+ }
+
+ public SolrQueuedThreadPool(@Name("maxThreads") int maxThreads, @Name("minThreads") int minThreads, @Name("idleTimeout") int idleTimeout, @Name("queue") BlockingQueue<Runnable> queue)
+ {
+ this(maxThreads, minThreads, idleTimeout, queue, null);
+ }
+
+ public SolrQueuedThreadPool(@Name("maxThreads") int maxThreads, @Name("minThreads") int minThreads, @Name("idleTimeout") int idleTimeout, @Name("queue") BlockingQueue<Runnable> queue, @Name("threadGroup") ThreadGroup threadGroup)
+ {
+ this(maxThreads, minThreads, idleTimeout, -1, queue, threadGroup);
+ }
+
+ public SolrQueuedThreadPool(@Name("maxThreads") int maxThreads, @Name("minThreads") int minThreads,
+ @Name("idleTimeout") int idleTimeout, @Name("reservedThreads") int reservedThreads,
+ @Name("queue") BlockingQueue<Runnable> queue, @Name("threadGroup") ThreadGroup threadGroup)
+ {
+ this(maxThreads, minThreads, idleTimeout, reservedThreads, queue, threadGroup, null);
+ }
+
+ public SolrQueuedThreadPool(@Name("maxThreads") int maxThreads, @Name("minThreads") int minThreads,
+ @Name("idleTimeout") int idleTimeout, @Name("reservedThreads") int reservedThreads,
+ @Name("queue") BlockingQueue<Runnable> queue, @Name("threadGroup") ThreadGroup threadGroup,
+ @Name("threadFactory") ThreadFactory threadFactory)
+ {
+ if (maxThreads < minThreads)
+ throw new IllegalArgumentException("max threads (" + maxThreads + ") less than min threads (" + minThreads + ")");
+ setMinThreads(minThreads);
+ setMaxThreads(maxThreads);
+ setIdleTimeout(idleTimeout);
+ setStopTimeout(5000);
+ setReservedThreads(reservedThreads);
+ if (queue == null)
+ {
+ int capacity = Math.max(_minThreads, 8) * 1024;
+ queue = new BlockingArrayQueue<>(capacity, capacity);
+ }
+ _jobs = queue;
+ _threadGroup = threadGroup;
+ setThreadPoolBudget(new ThreadPoolBudget(this));
+ _threadFactory = threadFactory == null ? this : threadFactory;
+ }
+
+ @Override
+ public ThreadPoolBudget getThreadPoolBudget()
+ {
+ return _budget;
+ }
+
+ public void setThreadPoolBudget(ThreadPoolBudget budget)
+ {
+ if (budget != null && budget.getSizedThreadPool() != this)
+ throw new IllegalArgumentException();
+ _budget = budget;
+ }
+
+ @Override
+ protected void doStart() throws Exception
+ {
+ if (_reservedThreads == 0)
+ {
+ _tryExecutor = NO_TRY;
+ }
+ else
+ {
+ ReservedThreadExecutor reserved = new ReservedThreadExecutor(this, _reservedThreads);
+ reserved.setIdleTimeout(_idleTimeout, TimeUnit.MILLISECONDS);
+ _tryExecutor = reserved;
+ }
+ addBean(_tryExecutor);
+
+ _lastShrink.set(System.nanoTime());
+
+ super.doStart();
+ // The threads count set to MIN_VALUE is used to signal to Runners that the pool is stopped.
+ _counts.set(0, 0); // threads, idle
+ ensureThreads();
+ }
+
+
+ public void prepareToStop() throws Exception {
+ super.doStop();
+ }
+
+
+ @Override
+ protected void doStop() throws Exception
+ {
+ if (LOG.isDebugEnabled())
+ LOG.debug("Stopping {}", this);
+
+ super.doStop();
+
+ removeBean(_tryExecutor);
+ _tryExecutor = TryExecutor.NO_TRY;
+
+ // Signal the Runner threads that we are stopping
+ int threads = _counts.getAndSetHi(Integer.MIN_VALUE);
+
+ // If stop timeout try to gracefully stop
+ long timeout = getStopTimeout();
+ BlockingQueue<Runnable> jobs = getQueue();
+ if (timeout > 0)
+ {
+ // Fill the job queue with noop jobs to wakeup idle threads.
+ for (int i = 0; i < threads; ++i)
+ {
+ jobs.offer(NOOP);
+ }
+
+ // try to let jobs complete naturally for half our stop time
+ joinThreads(System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(timeout) / 2);
+
+ // If we still have threads running, get a bit more aggressive
+
+ // interrupt remaining threads
+ for (Thread thread : _threads)
+ {
+ if (LOG.isDebugEnabled())
+ LOG.debug("Interrupting {}", thread);
+ thread.interrupt();
+ }
+
+ // wait again for the other half of our stop time
+ joinThreads(System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(timeout) / 2);
+
+ Thread.yield();
+ if (LOG.isDebugEnabled())
+ {
+ for (Thread unstopped : _threads)
+ {
+ StringBuilder dmp = new StringBuilder();
+ for (StackTraceElement element : unstopped.getStackTrace())
+ {
+ dmp.append(System.lineSeparator()).append("\tat ").append(element);
+ }
+ LOG.warn("Couldn't stop {}{}", unstopped, dmp.toString());
+ }
+ }
+ else
+ {
+ for (Thread unstopped : _threads)
+ {
+ LOG.warn("{} Couldn't stop {}", this, unstopped);
+ }
+ }
+ }
+
+ // Close any un-executed jobs
+ while (!_jobs.isEmpty())
+ {
+ Runnable job = _jobs.poll();
+ if (job instanceof Closeable)
+ {
+ try
+ {
+ ((Closeable)job).close();
+ }
+ catch (Throwable t)
+ {
+ LOG.warn(t);
+ }
+ }
+ else if (job != NOOP)
+ LOG.warn("Stopped without executing or closing {}", job);
+ }
+
+ if (_budget != null)
+ _budget.reset();
+
+ synchronized (_joinLock)
+ {
+ _joinLock.notifyAll();
+ }
+ }
+
+ private void joinThreads(long stopByNanos) throws InterruptedException
+ {
+ for (Thread thread : _threads)
+ {
+ long canWait = TimeUnit.NANOSECONDS.toMillis(stopByNanos - System.nanoTime());
+ if (LOG.isDebugEnabled())
+ LOG.debug("Waiting for {} for {}", thread, canWait);
+ if (canWait > 0)
+ thread.join(canWait);
+ }
+ }
+
+ /**
+ * Thread Pool should use Daemon Threading.
+ *
+ * @param daemon true to enable delegation
+ * @see Thread#setDaemon(boolean)
+ */
+ public void setDaemon(boolean daemon)
+ {
+ _daemon = daemon;
+ }
+
+ /**
+ * Set the maximum thread idle time.
+ * Threads that are idle for longer than this period may be
+ * stopped.
+ *
+ * @param idleTimeout Max idle time in ms.
+ * @see #getIdleTimeout
+ */
+ public void setIdleTimeout(int idleTimeout)
+ {
+ _idleTimeout = idleTimeout;
+ ReservedThreadExecutor reserved = getBean(ReservedThreadExecutor.class);
+ if (reserved != null)
+ reserved.setIdleTimeout(idleTimeout, TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * Set the maximum number of threads.
+ *
+ * @param maxThreads maximum number of threads.
+ * @see #getMaxThreads
+ */
+ @Override
+ public void setMaxThreads(int maxThreads)
+ {
+ if (_budget != null)
+ _budget.check(maxThreads);
+ _maxThreads = maxThreads;
+ if (_minThreads > _maxThreads)
+ _minThreads = _maxThreads;
+ }
+
+ /**
+ * Set the minimum number of threads.
+ *
+ * @param minThreads minimum number of threads
+ * @see #getMinThreads
+ */
+ @Override
+ public void setMinThreads(int minThreads)
+ {
+ _minThreads = minThreads;
+
+ if (_minThreads > _maxThreads)
+ _maxThreads = _minThreads;
+
+ if (isStarted())
+ ensureThreads();
+ }
+
+ /**
+ * Set the number of reserved threads.
+ *
+ * @param reservedThreads number of reserved threads or -1 for heuristically determined
+ * @see #getReservedThreads
+ */
+ public void setReservedThreads(int reservedThreads)
+ {
+ if (isRunning())
+ throw new IllegalStateException(getState());
+ _reservedThreads = reservedThreads;
+ }
+
+ /**
+ * @param name Name of this thread pool to use when naming threads.
+ */
+ public void setName(String name)
+ {
+ if (isRunning())
+ throw new IllegalStateException("started");
+ _name = name;
+ }
+
+ /**
+ * Set the priority of the pool threads.
+ *
+ * @param priority the new thread priority.
+ */
+ public void setThreadsPriority(int priority)
+ {
+ _priority = priority;
+ }
+
+ /**
+ * Get the maximum thread idle time.
+ *
+ * @return Max idle time in ms.
+ * @see #setIdleTimeout
+ */
+ @ManagedAttribute("maximum time a thread may be idle in ms")
+ public int getIdleTimeout()
+ {
+ return _idleTimeout;
+ }
+
+ /**
+ * Get the maximum number of threads.
+ *
+ * @return maximum number of threads.
+ * @see #setMaxThreads
+ */
+ @Override
+ @ManagedAttribute("maximum number of threads in the pool")
+ public int getMaxThreads()
+ {
+ return _maxThreads;
+ }
+
+ /**
+ * Get the minimum number of threads.
+ *
+ * @return minimum number of threads.
+ * @see #setMinThreads
+ */
+ @Override
+ @ManagedAttribute("minimum number of threads in the pool")
+ public int getMinThreads()
+ {
+ return _minThreads;
+ }
+
+ /**
+ * Get the number of reserved threads.
+ *
+ * @return number of reserved threads or or -1 for heuristically determined
+ * @see #setReservedThreads
+ */
+ @ManagedAttribute("the number of reserved threads in the pool")
+ public int getReservedThreads()
+ {
+ if (isStarted())
+ {
+ ReservedThreadExecutor reservedThreadExecutor = getBean(ReservedThreadExecutor.class);
+ if (reservedThreadExecutor != null)
+ return reservedThreadExecutor.getCapacity();
+ }
+ return _reservedThreads;
+ }
+
+ /**
+ * @return The name of the this thread pool
+ */
+ @ManagedAttribute("name of the thread pool")
+ public String getName()
+ {
+ return _name;
+ }
+
+ /**
+ * Get the priority of the pool threads.
+ *
+ * @return the priority of the pool threads.
+ */
+ @ManagedAttribute("priority of threads in the pool")
+ public int getThreadsPriority()
+ {
+ return _priority;
+ }
+
+ /**
+ * Get the size of the job queue.
+ *
+ * @return Number of jobs queued waiting for a thread
+ */
+ @ManagedAttribute("size of the job queue")
+ public int getQueueSize()
+ {
+ // The idle counter encodes demand, which is the effective queue size
+ int idle = _counts.getLo();
+ return Math.max(0, -idle);
+ }
+
+ /**
+ * @return whether this thread pool is using daemon threads
+ * @see Thread#setDaemon(boolean)
+ */
+ @ManagedAttribute("thread pool uses daemon threads")
+ public boolean isDaemon()
+ {
+ return _daemon;
+ }
+
+ @ManagedAttribute("reports additional details in the dump")
+ public boolean isDetailedDump()
+ {
+ return _detailedDump;
+ }
+
+ public void setDetailedDump(boolean detailedDump)
+ {
+ _detailedDump = detailedDump;
+ }
+
+ @ManagedAttribute("threshold at which the pool is low on threads")
+ public int getLowThreadsThreshold()
+ {
+ return _lowThreadsThreshold;
+ }
+
+ public void setLowThreadsThreshold(int lowThreadsThreshold)
+ {
+ _lowThreadsThreshold = lowThreadsThreshold;
+ }
+
+ @Override
+ public void execute(Runnable job)
+ {
+ // Determine if we need to start a thread, use and idle thread or just queue this job
+ int startThread;
+ while (true)
+ {
+ // Get the atomic counts
+ long counts = _counts.get();
+
+ // Get the number of threads started (might not yet be running)
+ int threads = AtomicBiInteger.getHi(counts);
+ if (threads == Integer.MIN_VALUE)
+ throw new RejectedExecutionException(job.toString());
+
+ // Get the number of truly idle threads. This count is reduced by the
+ // job queue size so that any threads that are idle but are about to take
+ // a job from the queue are not counted.
+ int idle = AtomicBiInteger.getLo(counts);
+
+ // Start a thread if we have insufficient idle threads to meet demand
+ // and we are not at max threads.
+ startThread = (idle <= 0 && threads < _maxThreads) ? 1 : 0;
+
+ // The job will be run by an idle thread when available
+ if (!_counts.compareAndSet(counts, threads + startThread, idle + startThread - 1))
+ continue;
+
+ break;
+ }
+
+ if (!_jobs.offer(job))
+ {
+ // reverse our changes to _counts.
+ if (addCounts(-startThread, 1 - startThread))
+ LOG.warn("{} rejected {}", this, job);
+ throw new RejectedExecutionException(job.toString());
+ }
+
+ if (LOG.isDebugEnabled())
+ LOG.debug("queue {} startThread={}", job, startThread);
+
+ // Start a thread if one was needed
+ while (startThread-- > 0)
+ startThread();
+ }
+
+ @Override
+ public boolean tryExecute(Runnable task)
+ {
+ TryExecutor tryExecutor = _tryExecutor;
+ return tryExecutor != null && tryExecutor.tryExecute(task);
+ }
+
+ @Override
+ public void join() throws InterruptedException
+ {
+// synchronized (_joinLock)
+// {
+// while (isRunning())
+// {
+// _joinLock.wait();
+// }
+// }
+//
+// while (isStopping())
+// {
+// Thread.sleep(1);
+// }
+ }
+
+ /**
+ * @return the total number of threads currently in the pool
+ */
+ @Override
+ @ManagedAttribute("number of threads in the pool")
+ public int getThreads()
+ {
+ int threads = _counts.getHi();
+ return Math.max(0, threads);
+ }
+
+ /**
+ * @return the number of idle threads in the pool
+ */
+ @Override
+ @ManagedAttribute("number of idle threads in the pool")
+ public int getIdleThreads()
+ {
+ int idle = _counts.getLo();
+ return Math.max(0, idle);
+ }
+
+ /**
+ * @return the number of busy threads in the pool
+ */
+ @ManagedAttribute("number of busy threads in the pool")
+ public int getBusyThreads()
+ {
+ int reserved = _tryExecutor instanceof ReservedThreadExecutor ? ((ReservedThreadExecutor)_tryExecutor).getAvailable() : 0;
+ return getThreads() - getIdleThreads() - reserved;
+ }
+
+ /**
+ * <p>Returns whether this thread pool is low on threads.</p>
+ * <p>The current formula is:</p>
+ * <pre>
+ * maxThreads - threads + idleThreads - queueSize <= lowThreadsThreshold
+ * </pre>
+ *
+ * @return whether the pool is low on threads
+ * @see #getLowThreadsThreshold()
+ */
+ @Override
+ @ManagedAttribute(value = "thread pool is low on threads", readonly = true)
+ public boolean isLowOnThreads()
+ {
+ return getMaxThreads() - getThreads() + getIdleThreads() - getQueueSize() <= getLowThreadsThreshold();
+ }
+
+ private void ensureThreads()
+ {
+ while (true)
+ {
+ long counts = _counts.get();
+ int threads = AtomicBiInteger.getHi(counts);
+ if (threads == Integer.MIN_VALUE)
+ break;
+
+ // If we have less than min threads
+ // OR insufficient idle threads to meet demand
+ int idle = AtomicBiInteger.getLo(counts);
+ if (threads < _minThreads || (idle < 0 && threads < _maxThreads))
+ {
+ // Then try to start a thread.
+ if (_counts.compareAndSet(counts, threads + 1, idle + 1))
+ startThread();
+ // Otherwise continue to check state again.
+ continue;
+ }
+ break;
+ }
+ }
+
+ protected void startThread()
+ {
+ boolean started = false;
+ try
+ {
+ Thread thread = _threadFactory.newThread(_runnable);
+ if (LOG.isDebugEnabled())
+ LOG.debug("Starting {}", thread);
+ _threads.add(thread);
+ _lastShrink.set(System.nanoTime());
+ thread.start();
+ started = true;
+ }
+ finally
+ {
+ if (!started)
+ addCounts(-1, -1); // threads, idle
+ }
+ }
+
+ private boolean addCounts(int deltaThreads, int deltaIdle)
+ {
+ while (true)
+ {
+ long encoded = _counts.get();
+ int threads = AtomicBiInteger.getHi(encoded);
+ int idle = AtomicBiInteger.getLo(encoded);
+ if (threads == Integer.MIN_VALUE) // This is a marker that the pool is stopped.
+ return false;
+ long update = AtomicBiInteger.encode(threads + deltaThreads, idle + deltaIdle);
+ if (_counts.compareAndSet(encoded, update))
+ return true;
+ }
+ }
+
+ @Override
+ public Thread newThread(Runnable runnable)
+ {
+ Thread thread = new Thread(_threadGroup, runnable);
+ thread.setDaemon(isDaemon());
+ thread.setPriority(getThreadsPriority());
+ thread.setName(_name + "-" + thread.getId());
+ return thread;
+ }
+
+ protected void removeThread(Thread thread)
+ {
+ _threads.remove(thread);
+ }
+
+ @Override
+ public void dump(Appendable out, String indent) throws IOException
+ {
+ List<Object> threads = new ArrayList<>(getMaxThreads());
+ for (final Thread thread : _threads)
+ {
+ final StackTraceElement[] trace = thread.getStackTrace();
+ String knownMethod = "";
+ for (StackTraceElement t : trace)
+ {
+ if ("idleJobPoll".equals(t.getMethodName()) && t.getClassName().equals(SolrQueuedThreadPool.Runner.class.getName()))
+ {
+ knownMethod = "IDLE ";
+ break;
+ }
+
+ if ("reservedWait".equals(t.getMethodName()) && t.getClassName().endsWith("ReservedThread"))
+ {
+ knownMethod = "RESERVED ";
+ break;
+ }
+
+ if ("select".equals(t.getMethodName()) && t.getClassName().endsWith("SelectorProducer"))
+ {
+ knownMethod = "SELECTING ";
+ break;
+ }
+
+ if ("accept".equals(t.getMethodName()) && t.getClassName().contains("ServerConnector"))
+ {
+ knownMethod = "ACCEPTING ";
+ break;
+ }
+ }
+ final String known = knownMethod;
+
+ if (isDetailedDump())
+ {
+ threads.add(new Dumpable()
+ {
+ @Override
+ public void dump(Appendable out, String indent) throws IOException
+ {
+ if (StringUtil.isBlank(known))
+ Dumpable.dumpObjects(out, indent, String.format("%s %s %s %d", thread.getId(), thread.getName(), thread.getState(), thread.getPriority()), (Object[])trace);
+ else
+ Dumpable.dumpObjects(out, indent, String.format("%s %s %s %s %d", thread.getId(), thread.getName(), known, thread.getState(), thread.getPriority()));
+ }
+
+ @Override
+ public String dump()
+ {
+ return null;
+ }
+ });
+ }
+ else
+ {
+ int p = thread.getPriority();
+ threads.add(thread.getId() + " " + thread.getName() + " " + known + thread.getState() + " @ " + (trace.length > 0 ? trace[0] : "???") + (p == Thread.NORM_PRIORITY ? "" : (" prio=" + p)));
+ }
+ }
+
+ if (isDetailedDump())
+ {
+ List<Runnable> jobs = new ArrayList<>(getQueue());
+ dumpObjects(out, indent, new DumpableCollection("threads", threads), new DumpableCollection("jobs", jobs));
+ }
+ else
+ {
+ dumpObjects(out, indent, new DumpableCollection("threads", threads));
+ }
+ }
+
+ @Override
+ public String toString()
+ {
+ long count = _counts.get();
+ int threads = Math.max(0, AtomicBiInteger.getHi(count));
+ 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,q=%d}[%s]",
+ getClass().getSimpleName(),
+ _name,
+ hashCode(),
+ getState(),
+ getMinThreads(),
+ threads,
+ getMaxThreads(),
+ idle,
+ getReservedThreads(),
+ queue,
+ _tryExecutor);
+ }
+
+ private final Runnable _runnable = new SolrQueuedThreadPool.Runner();
+
+ /**
+ * <p>Runs the given job in the {@link Thread#currentThread() current thread}.</p>
+ * <p>Subclasses may override to perform pre/post actions before/after the job is run.</p>
+ *
+ * @param job the job to run
+ */
protected void runJob(Runnable job) {
try {
job.run();
@@ -54,10 +821,201 @@ public class SolrQueuedThreadPool extends QueuedThreadPool implements Closeable
notify.notifyAll();
}
}
+ /**
+ * @return the job queue
+ */
+ protected BlockingQueue<Runnable> getQueue()
+ {
+ return _jobs;
+ }
+
+ /**
+ * @param queue the job queue
+ * @deprecated pass the queue to the constructor instead
+ */
+ @Deprecated
+ public void setQueue(BlockingQueue<Runnable> queue)
+ {
+ throw new UnsupportedOperationException("Use constructor injection");
+ }
+
+ /**
+ * @param id the thread ID to interrupt.
+ * @return true if the thread was found and interrupted.
+ */
+ @ManagedOperation("interrupts a pool thread")
+ public boolean interruptThread(@Name("id") long id)
+ {
+ for (Thread thread : _threads)
+ {
+ if (thread.getId() == id)
+ {
+ thread.interrupt();
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * @param id the thread ID to interrupt.
+ * @return the stack frames dump
+ */
+ @ManagedOperation("dumps a pool thread stack")
+ public String dumpThread(@Name("id") long id)
+ {
+ for (Thread thread : _threads)
+ {
+ if (thread.getId() == id)
+ {
+ StringBuilder buf = new StringBuilder();
+ buf.append(thread.getId()).append(" ").append(thread.getName()).append(" ");
+ buf.append(thread.getState()).append(":").append(System.lineSeparator());
+ for (StackTraceElement element : thread.getStackTrace())
+ {
+ buf.append(" at ").append(element.toString()).append(System.lineSeparator());
+ }
+ return buf.toString();
+ }
+ }
+ return null;
+ }
+
+ private class Runner implements Runnable
+ {
+ private Runnable idleJobPoll(long idleTimeout) throws InterruptedException
+ {
+ if (idleTimeout <= 0)
+ return _jobs.take();
+ return _jobs.poll(idleTimeout, TimeUnit.MILLISECONDS);
+ }
+
+ @Override
+ public void run()
+ {
+ if (LOG.isDebugEnabled())
+ LOG.debug("Runner started for {}", SolrQueuedThreadPool.this);
+
+ boolean idle = true;
+ try
+ {
+ Runnable job = null;
+ while (true)
+ {
+ // If we had a job,
+ if (job != null)
+ {
+ // signal that we are idle again
+ if (!addCounts(0, 1))
+ break;
+ idle = true;
+ }
+ // else check we are still running
+ else if (_counts.getHi() == Integer.MIN_VALUE)
+ {
+ break;
+ }
+
+ try
+ {
+ // Look for an immediately available job
+ job = _jobs.poll();
+ if (job == null)
+ {
+ // No job immediately available maybe we should shrink?
+ long idleTimeout = getIdleTimeout();
+ if (idleTimeout > 0 && getThreads() > _minThreads)
+ {
+ long last = _lastShrink.get();
+ long now = System.nanoTime();
+ if ((now - last) > TimeUnit.MILLISECONDS.toNanos(idleTimeout) && _lastShrink.compareAndSet(last, now))
+ {
+ if (LOG.isDebugEnabled())
+ LOG.debug("shrinking {}", SolrQueuedThreadPool.this);
+ break;
+ }
+ }
+
+ // Wait for a job, only after we have checked if we should shrink
+ job = idleJobPoll(idleTimeout);
+
+ // If still no job?
+ if (job == null)
+ // continue to try again
+ continue;
+ }
+
+ idle = false;
+
+ // run job
+ if (LOG.isDebugEnabled())
+ LOG.debug("run {} in {}", job, SolrQueuedThreadPool.this);
+ runJob(job);
+ if (LOG.isDebugEnabled())
+ LOG.debug("ran {} in {}", job, SolrQueuedThreadPool.this);
+ }
+ catch (InterruptedException e)
+ {
+ if (LOG.isDebugEnabled())
+ LOG.debug("interrupted {} in {}", job, SolrQueuedThreadPool.this);
+ LOG.ignore(e);
+ }
+ catch (Throwable e)
+ {
+ LOG.warn(e);
+ }
+ finally
+ {
+ // Clear any interrupted status
+ Thread.interrupted();
+ }
+ }
+ }
+ finally
+ {
+ Thread thread = Thread.currentThread();
+ removeThread(thread);
+
+ // Decrement the total thread count and the idle count if we had no job
+ addCounts(-1, idle ? -1 : 0);
+ if (LOG.isDebugEnabled())
+ LOG.debug("{} exited for {}", thread, SolrQueuedThreadPool.this);
+
+ // There is a chance that we shrunk just as a job was queued for us, so
+ // check again if we have sufficient threads to meet demand
+ ensureThreads();
+ }
+ }
+ }
+
+
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+ private String name;
+ private volatile Error error;
+ private final Object notify = new Object();
+
+
+
+ public void waitForStopping() throws InterruptedException {
+ int threads = _counts.getAndSetHi(Integer.MIN_VALUE);
+ BlockingQueue<Runnable> jobs = getQueue();
+ // Fill the job queue with noop jobs to wakeup idle threads.
+ for (int i = 0; i < threads; ++i)
+ {
+ jobs.offer(NOOP);
+ }
+
+ // try to let jobs complete naturally for half our stop time
+ joinThreads( TimeUnit.MILLISECONDS.toNanos(10000));
+
+ }
+
public void close() {
try {
- doStop();
+ if (!isStopping() || !isStopped()) {
+ stop();
+ }
while (isStopping()) {
Thread.sleep(1);
}
@@ -68,17 +1026,13 @@ public class SolrQueuedThreadPool extends QueuedThreadPool implements Closeable
assert ObjectReleaseTracker.release(this);
}
- @Override
- public void doStop() throws Exception {
- super.doStop();
- }
-
- public void stdStop() throws Exception {
- super.doStop();
- }
+// @Override
+// public void doStop() throws Exception {
+// super.doStop();
+// }
+//
+// public void stdStop() throws Exception {
+// super.doStop();
+// }
- @Override
- public void join() throws InterruptedException {
-
- }
}
\ No newline at end of file