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/08/11 12:46:44 UTC
[lucene-solr] branch reference_impl_dev updated (1d7c7b9 -> c371780)
This is an automated email from the ASF dual-hosted git repository.
markrmiller pushed a change to branch reference_impl_dev
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git.
from 1d7c7b9 @505 Keep playing with executor shutdown.
new 6baefc4 @506 More exec shutdown work.
new c371780 @507 Don't dump IW logging by default.
The 2 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:
.../src/java/org/apache/solr/core/SolrCore.java | 7 +-
solr/core/src/test-files/log4j2.xml | 1 +
.../solr/cloud/FullSolrCloudDistribCmdsTest.java | 3 +-
.../solr/client/solrj/cloud/DistributedLock.java | 444 +++++++++++----------
.../src/java/org/apache/solr/common/ParWork.java | 15 +-
.../org/apache/solr/common/ParWorkExecutor.java | 1 -
6 files changed, 233 insertions(+), 238 deletions(-)
[lucene-solr] 01/02: @506 More exec shutdown work.
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_dev
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
commit 6baefc493debf369e632bde0880c60db25bf5616
Author: markrmiller@gmail.com <ma...@gmail.com>
AuthorDate: Tue Aug 11 07:45:04 2020 -0500
@506 More exec shutdown work.
---
.../src/java/org/apache/solr/core/SolrCore.java | 7 +-
.../solr/cloud/FullSolrCloudDistribCmdsTest.java | 3 +-
.../solr/client/solrj/cloud/DistributedLock.java | 444 +++++++++++----------
.../src/java/org/apache/solr/common/ParWork.java | 15 +-
.../org/apache/solr/common/ParWorkExecutor.java | 1 -
5 files changed, 232 insertions(+), 238 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/core/SolrCore.java b/solr/core/src/java/org/apache/solr/core/SolrCore.java
index bf9c2a2..b97c63f 100644
--- a/solr/core/src/java/org/apache/solr/core/SolrCore.java
+++ b/solr/core/src/java/org/apache/solr/core/SolrCore.java
@@ -1039,10 +1039,10 @@ public final class SolrCore implements SolrInfoBean, Closeable {
// cause the executor to stall so firstSearcher events won't fire
// until after inform() has been called for all components.
// searchExecutor must be single-threaded for this to work
- searcherExecutor.doSubmit(() -> {
+ searcherExecutor.submit(() -> {
boolean success = latch.await(10000, TimeUnit.MILLISECONDS);
return null;
- }, true);
+ });
this.updateHandler = initUpdateHandler(updateHandler);
@@ -1879,7 +1879,8 @@ public final class SolrCore implements SolrInfoBean, Closeable {
private final LinkedList<RefCounted<SolrIndexSearcher>> _searchers = new LinkedList<>();
private final LinkedList<RefCounted<SolrIndexSearcher>> _realtimeSearchers = new LinkedList<>();
- final ParWorkExecService searcherExecutor = (ParWorkExecService) ParWork.getExecutorService(1);
+ final ExecutorService searcherExecutor = ExecutorUtil.newMDCAwareSingleThreadExecutor(
+ new SolrNamedThreadFactory("searcherExecutor"));
private AtomicInteger onDeckSearchers = new AtomicInteger(); // number of searchers preparing
// Lock ordering: one can acquire the openSearcherLock and then the searcherLock, but not vice-versa.
private final Object searcherLock = new Object(); // the sync object for the searcher
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 5c929f4..8638ef1 100644
--- a/solr/core/src/test/org/apache/solr/cloud/FullSolrCloudDistribCmdsTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/FullSolrCloudDistribCmdsTest.java
@@ -63,7 +63,6 @@ import org.slf4j.LoggerFactory;
* Super basic testing, no shard restarting or anything.
*/
@Slow
-
public class FullSolrCloudDistribCmdsTest extends SolrCloudTestCase {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private static final AtomicInteger NAME_COUNTER = new AtomicInteger(1);
@@ -401,7 +400,6 @@ public class FullSolrCloudDistribCmdsTest extends SolrCloudTestCase {
checkShardConsistency(params("q","*:*", "rows", ""+(1 + numDocs),"_trace","addAll"));
}
- // @Ignore // nocommit debug
public void testIndexingBatchPerRequestWithHttpSolrClient() throws Exception {
final CloudHttp2SolrClient cloudClient = cluster.getSolrClient();
final String collectionName = createAndSetNewDefaultCollection();
@@ -458,6 +456,7 @@ public class FullSolrCloudDistribCmdsTest extends SolrCloudTestCase {
cloudClient.commit(collectionName);
assertEquals(expectedDocCount.get(), cloudClient.query(params("q","*:*")).getResults().getNumFound());
+ cluster.waitForActiveCollection(collectionName, 2, 4);
checkShardConsistency(params("q","*:*", "rows", ""+totalDocsExpected, "_trace","batches_done"));
}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/DistributedLock.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/DistributedLock.java
index 474d994..1207fe9 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/DistributedLock.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/DistributedLock.java
@@ -45,254 +45,260 @@ import java.util.TreeSet;
*/
public class DistributedLock extends ProtocolSupport {
- private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+ private static final Logger log = LoggerFactory
+ .getLogger(MethodHandles.lookup().lookupClass());
- private final String dir;
- private String id;
- private ZNodeName idName;
- private String ownerId;
- private String lastChildId;
- private byte[] data = {0x12, 0x34};
- private LockListener callback;
- private LockZooKeeperOperation zop;
+ private final String dir;
+ private String id;
+ private ZNodeName idName;
+ private String ownerId;
+ private String lastChildId;
+ private byte[] data = {0x12, 0x34};
+ private LockListener callback;
+ private LockZooKeeperOperation zop;
- /**
- * zookeeper contructor for writelock.
- *
- * @param zookeeper zookeeper client instance
- * @param dir the parent path you want to use for locking
- * @param acl the acls that you want to use for all the paths, if null world read/write is used.
- */
- public DistributedLock(SolrZkClient zookeeper, String dir, List<ACL> acl) {
- super(zookeeper);
- this.dir = dir;
- if (acl != null) {
- setAcl(acl);
- }
- this.zop = new LockZooKeeperOperation();
+ /**
+ * zookeeper contructor for writelock.
+ *
+ * @param zookeeper zookeeper client instance
+ * @param dir the parent path you want to use for locking
+ * @param acl the acls that you want to use for all the paths, if null world read/write is used.
+ */
+ public DistributedLock(SolrZkClient zookeeper, String dir, List<ACL> acl) {
+ super(zookeeper);
+ this.dir = dir;
+ if (acl != null) {
+ setAcl(acl);
}
+ this.zop = new LockZooKeeperOperation();
+ }
- /**
- * zookeeper contructor for writelock with callback.
- *
- * @param zookeeper the zookeeper client instance
- * @param dir the parent path you want to use for locking
- * @param acl the acls that you want to use for all the paths
- * @param callback the call back instance
- */
- public DistributedLock(
- SolrZkClient zookeeper,
- String dir,
- List<ACL> acl,
- LockListener callback) {
- this(zookeeper, dir, acl);
- this.callback = callback;
- }
+ /**
+ * zookeeper contructor for writelock with callback.
+ *
+ * @param zookeeper the zookeeper client instance
+ * @param dir the parent path you want to use for locking
+ * @param acl the acls that you want to use for all the paths
+ * @param callback the call back instance
+ */
+ public DistributedLock(SolrZkClient zookeeper, String dir, List<ACL> acl,
+ LockListener callback) {
+ this(zookeeper, dir, acl);
+ this.callback = callback;
+ }
- /**
- * return the current locklistener.
- *
- * @return the locklistener
- */
- public synchronized LockListener getLockListener() {
- return this.callback;
- }
+ /**
+ * return the current locklistener.
+ *
+ * @return the locklistener
+ */
+ public synchronized LockListener getLockListener() {
+ return this.callback;
+ }
- /**
- * register a different call back listener.
- *
- * @param callback the call back instance
- */
- public synchronized void setLockListener(LockListener callback) {
- this.callback = callback;
- }
+ /**
+ * register a different call back listener.
+ *
+ * @param callback the call back instance
+ */
+ public synchronized void setLockListener(LockListener callback) {
+ this.callback = callback;
+ }
- /**
- * Removes the lock or associated znode if
- * you no longer require the lock. this also
- * removes your request in the queue for locking
- * in case you do not already hold the lock.
- *
- * @throws RuntimeException throws a runtime exception
- * if it cannot connect to zookeeper.
- */
- public synchronized void unlock() throws RuntimeException {
+ /**
+ * Removes the lock or associated znode if
+ * you no longer require the lock. this also
+ * removes your request in the queue for locking
+ * in case you do not already hold the lock.
+ *
+ * @throws RuntimeException throws a runtime exception
+ * if it cannot connect to zookeeper.
+ */
+ public synchronized void unlock() throws RuntimeException {
- if (!isClosed() && id != null) {
- // we don't need to retry this operation in the case of failure
- // as ZK will remove ephemeral files and we don't wanna hang
- // this process when closing if we cannot reconnect to ZK
- try {
+ if (!isClosed() && id != null) {
+ // we don't need to retry this operation in the case of failure
+ // as ZK will remove ephemeral files and we don't wanna hang
+ // this process when closing if we cannot reconnect to ZK
+ try {
- ZooKeeperOperation zopdel = () -> {
- zookeeper.getSolrZooKeeper().delete(id, -1);
- return Boolean.TRUE;
- };
- zopdel.execute();
- } catch (InterruptedException e) {
- log.warn("Unexpected exception", e);
- // set that we have been interrupted.
- Thread.currentThread().interrupt();
- } catch (KeeperException.NoNodeException e) {
- // do nothing
- } catch (KeeperException e) {
- log.warn("Unexpected exception", e);
- throw new RuntimeException(e.getMessage(), e);
- } finally {
- LockListener lockListener = getLockListener();
- if (lockListener != null) {
- lockListener.lockReleased();
- }
- id = null;
- }
+ ZooKeeperOperation zopdel = () -> {
+ zookeeper.getSolrZooKeeper().delete(id, -1);
+ return Boolean.TRUE;
+ };
+ zopdel.execute();
+ } catch (InterruptedException e) {
+ log.warn("Unexpected exception", e);
+ // set that we have been interrupted.
+ Thread.currentThread().interrupt();
+ } catch (KeeperException.NoNodeException e) {
+ // do nothing
+ } catch (KeeperException e) {
+ log.warn("Unexpected exception", e);
+ throw new RuntimeException(e.getMessage(), e);
+ } finally {
+ LockListener lockListener = getLockListener();
+ if (lockListener != null) {
+ lockListener.lockReleased();
}
+ id = null;
+ }
}
+ }
- /**
- * the watcher called on
- * getting watch while watching
- * my predecessor.
- */
- private class LockWatcher implements Watcher {
+ /**
+ * the watcher called on
+ * getting watch while watching
+ * my predecessor.
+ */
+ private class LockWatcher implements Watcher {
- public void process(WatchedEvent event) {
- // lets either become the leader or watch the new/updated node
- log.debug("Watcher fired: {}", event);
- try {
- lock();
- } catch (Exception e) {
+ public void process(WatchedEvent event) {
+ if (Event.EventType.None.equals(event.getType())) {
+ return;
+ }
+ // lets either become the leader or watch the new/updated node
+ if (log.isDebugEnabled()) log.debug("Watcher fired: {}", event);
+ try {
+ lock();
+ } catch (Exception e) {
+ log.warn("Failed to acquire lock", e);
+ }
+ }
+ }
- log.warn("Failed to acquire lock", e);
- }
- }
+ /**
+ * a zookeeper operation that is mainly responsible
+ * for all the magic required for locking.
+ */
+ private class LockZooKeeperOperation implements ZooKeeperOperation {
+ /**
+ * find if we have been created earler if not create our node.
+ *
+ * @param prefix the prefix node
+ * @param zookeeper teh zookeeper client
+ * @param dir the dir parent
+ * @throws KeeperException on zk exception
+ * @throws InterruptedException on interrupt
+ */
+ private void findPrefixInChildren(String prefix, SolrZkClient zookeeper,
+ String dir) throws KeeperException, InterruptedException {
+ List<String> names = zookeeper.getSolrZooKeeper().getChildren(dir, false);
+ for (String name : names) {
+ if (name.startsWith(prefix)) {
+ id = name;
+ log.debug("Found id created last time: {}", id);
+ break;
+ }
+ }
+ if (id == null) {
+ id = zookeeper.getSolrZooKeeper().create(dir + "/" + prefix, data,
+ zookeeper.getZkACLProvider().getACLsToAdd(dir + "/" + prefix),
+ EPHEMERAL_SEQUENTIAL);
+ log.debug("Created id: {}", id);
+ }
}
/**
- * a zookeeper operation that is mainly responsible
- * for all the magic required for locking.
+ * the command that is run and retried for actually
+ * obtaining the lock.
+ *
+ * @return if the command was successful or not
*/
- private class LockZooKeeperOperation implements ZooKeeperOperation {
-
- /**
- * find if we have been created earler if not create our node.
- *
- * @param prefix the prefix node
- * @param zookeeper teh zookeeper client
- * @param dir the dir parent
- * @throws KeeperException on zk exception
- * @throws InterruptedException on interrupt
- */
- private void findPrefixInChildren(String prefix, SolrZkClient zookeeper, String dir)
- throws KeeperException, InterruptedException {
- List<String> names = zookeeper.getSolrZooKeeper().getChildren(dir, false);
- for (String name : names) {
- if (name.startsWith(prefix)) {
- id = name;
- log.debug("Found id created last time: {}", id);
- break;
- }
- }
- if (id == null) {
- id = zookeeper.getSolrZooKeeper().create(dir + "/" + prefix, data, zookeeper.getZkACLProvider().getACLsToAdd(dir + "/" + prefix), EPHEMERAL_SEQUENTIAL);
- log.debug("Created id: {}", id);
- }
+ public boolean execute() throws KeeperException, InterruptedException {
+ do {
+ if (id == null) {
+ long sessionId = zookeeper.getSolrZooKeeper().getSessionId();
+ String prefix = "x-" + sessionId + "-";
+ // lets try look up the current ID if we failed
+ // in the middle of creating the znode
+ findPrefixInChildren(prefix, zookeeper, dir);
+ idName = new ZNodeName(id);
}
-
- /**
- * the command that is run and retried for actually
- * obtaining the lock.
- *
- * @return if the command was successful or not
- */
- public boolean execute() throws KeeperException, InterruptedException {
- do {
- if (id == null) {
- long sessionId = zookeeper.getSolrZooKeeper().getSessionId();
- String prefix = "x-" + sessionId + "-";
- // lets try look up the current ID if we failed
- // in the middle of creating the znode
- findPrefixInChildren(prefix, zookeeper, dir);
- idName = new ZNodeName(id);
- }
- List<String> names = zookeeper.getSolrZooKeeper().getChildren(dir, false);
- if (names.isEmpty()) {
- log.warn("No children in: {} when we've just created one! Lets recreate it...", dir);
- // lets force the recreation of the id
- id = null;
- } else {
- // lets sort them explicitly (though they do seem to come back in order ususally :)
- SortedSet<ZNodeName> sortedNames = new TreeSet<>();
- for (String name : names) {
- sortedNames.add(new ZNodeName(dir + "/" + name));
- }
- ownerId = sortedNames.first().getName();
- SortedSet<ZNodeName> lessThanMe = sortedNames.headSet(idName);
- if (!lessThanMe.isEmpty()) {
- ZNodeName lastChildName = lessThanMe.last();
- lastChildId = lastChildName.getName();
- log.debug("Watching less than me node: {}", lastChildId);
- Stat stat = zookeeper.getSolrZooKeeper().exists(lastChildId, new LockWatcher());
- if (stat != null) {
- return Boolean.FALSE;
- } else {
- log.warn("Could not find the stats for less than me: {}", lastChildName.getName());
- }
- } else {
- if (isOwner()) {
- LockListener lockListener = getLockListener();
- if (lockListener != null) {
- lockListener.lockAcquired();
- }
- return Boolean.TRUE;
- }
- }
- }
+ List<String> names = zookeeper.getSolrZooKeeper()
+ .getChildren(dir, false);
+ if (names.isEmpty()) {
+ log.warn(
+ "No children in: {} when we've just created one! Lets recreate it...",
+ dir);
+ // lets force the recreation of the id
+ id = null;
+ } else {
+ // lets sort them explicitly (though they do seem to come back in order ususally :)
+ SortedSet<ZNodeName> sortedNames = new TreeSet<>();
+ for (String name : names) {
+ sortedNames.add(new ZNodeName(dir + "/" + name));
+ }
+ ownerId = sortedNames.first().getName();
+ SortedSet<ZNodeName> lessThanMe = sortedNames.headSet(idName);
+ if (!lessThanMe.isEmpty()) {
+ ZNodeName lastChildName = lessThanMe.last();
+ lastChildId = lastChildName.getName();
+ log.debug("Watching less than me node: {}", lastChildId);
+ Stat stat = zookeeper.getSolrZooKeeper()
+ .exists(lastChildId, new LockWatcher());
+ if (stat != null) {
+ return Boolean.FALSE;
+ } else {
+ log.warn("Could not find the stats for less than me: {}",
+ lastChildName.getName());
+ }
+ } else {
+ if (isOwner()) {
+ LockListener lockListener = getLockListener();
+ if (lockListener != null) {
+ lockListener.lockAcquired();
+ }
+ return Boolean.TRUE;
}
- while (id == null);
- return Boolean.FALSE;
+ }
}
-
+ } while (id == null);
+ return Boolean.FALSE;
}
- /**
- * Attempts to acquire the exclusive write lock returning whether or not it was
- * acquired. Note that the exclusive lock may be acquired some time later after
- * this method has been invoked due to the current lock owner going away.
- */
- public synchronized boolean lock() throws KeeperException, InterruptedException {
- if (isClosed()) {
- return false;
- }
+ }
- return (Boolean) retryOperation(zop);
+ /**
+ * Attempts to acquire the exclusive write lock returning whether or not it was
+ * acquired. Note that the exclusive lock may be acquired some time later after
+ * this method has been invoked due to the current lock owner going away.
+ */
+ public synchronized boolean lock()
+ throws KeeperException, InterruptedException {
+ if (isClosed() || !zookeeper.isConnected()) {
+ return false;
}
- /**
- * return the parent dir for lock.
- *
- * @return the parent dir used for locks.
- */
- public String getDir() {
- return dir;
- }
+ return (Boolean) retryOperation(zop);
+ }
- /**
- * Returns true if this node is the owner of the
- * lock (or the leader).
- */
- public boolean isOwner() {
- return id != null && id.equals(ownerId);
- }
+ /**
+ * return the parent dir for lock.
+ *
+ * @return the parent dir used for locks.
+ */
+ public String getDir() {
+ return dir;
+ }
- /**
- * return the id for this lock.
- *
- * @return the id for this lock
- */
- public String getId() {
- return this.id;
- }
+ /**
+ * Returns true if this node is the owner of the
+ * lock (or the leader).
+ */
+ public boolean isOwner() {
+ return id != null && id.equals(ownerId);
+ }
+
+ /**
+ * return the id for this lock.
+ *
+ * @return the id for this lock
+ */
+ public String getId() {
+ return this.id;
+ }
}
diff --git a/solr/solrj/src/java/org/apache/solr/common/ParWork.java b/solr/solrj/src/java/org/apache/solr/common/ParWork.java
index 6cdb655..d28605f 100644
--- a/solr/solrj/src/java/org/apache/solr/common/ParWork.java
+++ b/solr/solrj/src/java/org/apache/solr/common/ParWork.java
@@ -64,7 +64,7 @@ public class ParWork implements Closeable {
protected final static ThreadLocal<ExecutorService> THREAD_LOCAL_EXECUTOR = new ThreadLocal<>();
private final boolean requireAnotherThread;
- private volatile Set<Object> collectSet = null;
+ private volatile Set<Object> collectSet = ConcurrentHashMap.newKeySet(32);
private static volatile ThreadPoolExecutor EXEC;
@@ -249,10 +249,6 @@ public class ParWork implements Closeable {
if (object == null) {
return;
}
- if (collectSet == null) {
- collectSet = ConcurrentHashMap.newKeySet(32);
- }
-
collectSet.add(object);
}
@@ -261,10 +257,6 @@ public class ParWork implements Closeable {
* used to identify it.
*/
public void collect(Callable<?> callable) {
-
- if (collectSet == null) {
- collectSet = ConcurrentHashMap.newKeySet(32);
- }
collectSet.add(callable);
}
@@ -276,14 +268,11 @@ public class ParWork implements Closeable {
if (runnable == null) {
return;
}
- if (collectSet == null) {
- collectSet = ConcurrentHashMap.newKeySet(32);
- }
collectSet.add(runnable);
}
public void addCollect(String label) {
- if (collectSet == null) {
+ if (collectSet.isEmpty()) {
log.info("No work collected to submit");
return;
}
diff --git a/solr/solrj/src/java/org/apache/solr/common/ParWorkExecutor.java b/solr/solrj/src/java/org/apache/solr/common/ParWorkExecutor.java
index 95373a4..311b1b7c 100644
--- a/solr/solrj/src/java/org/apache/solr/common/ParWorkExecutor.java
+++ b/solr/solrj/src/java/org/apache/solr/common/ParWorkExecutor.java
@@ -68,7 +68,6 @@ public class ParWorkExecutor extends ExecutorUtil.MDCAwareThreadPoolExecutor {
public void shutdown() {
allowCoreThreadTimeOut(true);
- setKeepAliveTime(1, TimeUnit.NANOSECONDS);
super.shutdown();
}
}
[lucene-solr] 02/02: @507 Don't dump IW logging by default.
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_dev
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
commit c3717806d1c00cc3292def1e961e84c004db571d
Author: markrmiller@gmail.com <ma...@gmail.com>
AuthorDate: Tue Aug 11 07:45:25 2020 -0500
@507 Don't dump IW logging by default.
---
solr/core/src/test-files/log4j2.xml | 1 +
1 file changed, 1 insertion(+)
diff --git a/solr/core/src/test-files/log4j2.xml b/solr/core/src/test-files/log4j2.xml
index 46ad20c..1928dbd 100644
--- a/solr/core/src/test-files/log4j2.xml
+++ b/solr/core/src/test-files/log4j2.xml
@@ -28,6 +28,7 @@
</Console>
</Appenders>
<Loggers>
+ <AsyncLogger name="org.apache.solr.update.LoggingInfoStream" level="off"/>
<AsyncLogger name="org.apache.zookeeper" level="WARN"/>
<AsyncLogger name="org.apache.hadoop" level="WARN"/>
<AsyncLogger name="org.apache.directory" level="WARN"/>