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:10 UTC
[lucene-solr] 01/01: @222 - Work on fixing the wait for async
request phaser. "I love the zoom..."
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