You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by yo...@apache.org on 2019/11/21 23:59:44 UTC
[lucene-solr] branch jira/SOLR-13101 updated: SOLR-13101:These are
pretty much non functional changes,
related to SHARED replica concurrent updates (#1028)
This is an automated email from the ASF dual-hosted git repository.
yonik pushed a commit to branch jira/SOLR-13101
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
The following commit(s) were added to refs/heads/jira/SOLR-13101 by this push:
new 9201c58 SOLR-13101:These are pretty much non functional changes, related to SHARED replica concurrent updates (#1028)
9201c58 is described below
commit 9201c586ba05537a049e6406070361cdda561a4c
Author: Bilal Waheed <23...@users.noreply.github.com>
AuthorDate: Thu Nov 21 15:59:37 2019 -0800
SOLR-13101:These are pretty much non functional changes, related to SHARED replica concurrent updates (#1028)
https://github.com/apache/lucene-solr/pull/983/commits/581f468f9914ce2488201efbed42fd43dc4b481
- Pull read lock is only required before the local indexing starts. If indexing has to pull it can acquire and release write lock without anything to do with read lock first. Removing the unneeded lock upgrade/downgrade logic to simplify things.
- Added one pager explanation in the start of SharedCoreConcurrencyController around overall concurrency design.
- Added mores comments around BlobCoreMetadata#generation number and its usage.
---
.../solr/handler/admin/RequestApplyUpdatesOp.java | 11 +-
.../solr/store/blob/client/BlobCoreMetadata.java | 24 +++-
.../apache/solr/store/blob/process/CorePusher.java | 10 ++
.../shared/SharedCoreConcurrencyController.java | 83 +++++++++++-
.../processor/DistributedZkUpdateProcessor.java | 146 ++++++++++-----------
5 files changed, 190 insertions(+), 84 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/handler/admin/RequestApplyUpdatesOp.java b/solr/core/src/java/org/apache/solr/handler/admin/RequestApplyUpdatesOp.java
index 3658ed0..5007ade 100644
--- a/solr/core/src/java/org/apache/solr/handler/admin/RequestApplyUpdatesOp.java
+++ b/solr/core/src/java/org/apache/solr/handler/admin/RequestApplyUpdatesOp.java
@@ -100,11 +100,18 @@ class RequestApplyUpdatesOp implements CoreAdminHandler.CoreAdminOp {
// As of now we are only taking care of basic happy path. We still need to evaluate what will happen
// if a split is abandoned because of failure(e.g. long GC pause) and is re-tried?
// How to make sure our re-attempt wins even when the ghost of previous attempt resumes and intervenes?
+ //
+ // TODO: There is an assumption here that for SHARED replicas this code path is only called for shard splits.
+ // If that is true, can we establish/assert that explicitly?
+ // If that is not true, then we need to understand those other use cases and update following logic accordingly.
if (!SharedShardMetadataController.METADATA_NODE_DEFAULT_VALUE.equals(shardVersionMetadata.getMetadataSuffix())) {
- throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "New sub shard has zk information that is not default");
+ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,
+ String.format("New sub shard has zk information that is not default, collection=%s, shard=%s, core=%s",
+ collectionName, shardName, coreName));
}
- // sync local cache with zk's default information i.e. equivalent of no-op pull
+ // sync local cache with zk's default information i.e. equivalent of no-op pull
+ // this syncing is necessary for the zk conditional update to succeed at the end of core push
SharedCoreConcurrencyController concurrencyController = cc.getSharedStoreManager().getSharedCoreConcurrencyController();
String sharedBlobName = Assign.buildSharedShardName(collectionName, shardName);
concurrencyController.updateCoreVersionMetadata(collectionName, shardName, coreName,
diff --git a/solr/core/src/java/org/apache/solr/store/blob/client/BlobCoreMetadata.java b/solr/core/src/java/org/apache/solr/store/blob/client/BlobCoreMetadata.java
index dc36dc9..3a9749f 100644
--- a/solr/core/src/java/org/apache/solr/store/blob/client/BlobCoreMetadata.java
+++ b/solr/core/src/java/org/apache/solr/store/blob/client/BlobCoreMetadata.java
@@ -6,6 +6,11 @@ import java.util.Objects;
import java.util.Set;
import java.util.UUID;
+import org.apache.solr.store.blob.metadata.PushPullData;
+import org.apache.solr.store.blob.metadata.ServerSideMetadata;
+import org.apache.solr.store.blob.metadata.SharedStoreResolutionUtil;
+import org.apache.solr.store.blob.process.CorePusher;
+
/**
* Object defining metadata stored in blob store for a Shared Collection shard and its builders.
* This metadata includes all actual segment files as well as the segments_N file of the commit point.
@@ -20,11 +25,20 @@ public class BlobCoreMetadata {
*/
private final String sharedBlobName;
- /**
- * Generation number of index represented by this metadata.
- * This generation number is only meant to identify a scenario where local index generation number is higher than
- * what we have in blob. In that scenario we would switch index to a new directory when pulling contents from blob.
- * Because in the presence of higher generation number locally, blob contents cannot establish their legitimacy.
+ /**
+ * Generation number of index represented by this metadata. Generation numbers across replicas don't carry any meaning
+ * since each replica can be doing its own indexing. We are only using this generation for optimizations in the context of
+ * single replica. It should not be used beyond following stated purposes. If its mere existence is popping it into
+ * consideration for solving unintended use cases, then we can remove it and implement the suggested alternatives
+ * for following usages.
+ * 1. identify a scenario where local index generation number is higher than what we have in shared store. In that scenario
+ * we would switch index to a new directory when pulling contents from shared store. Because in the presence of higher
+ * generation number locally, shared store contents cannot establish their legitimacy. Storing this number is just
+ * an optimization. We can always infer that number from segment_N file and get rid of this usage.
+ * {@link SharedStoreResolutionUtil#resolveMetadata(ServerSideMetadata, BlobCoreMetadata)}
+ * 2. {@link CorePusher#pushCoreToBlob(PushPullData)} piggy back's on cached BlobCoreMetadata's generation number instead
+ * of carrying a separate lastGenerationPushed property in the cache. This is also an optimization, we can always
+ * add lastGenerationPushed property to cache and get rid of this usage.
*/
private final long generation;
diff --git a/solr/core/src/java/org/apache/solr/store/blob/process/CorePusher.java b/solr/core/src/java/org/apache/solr/store/blob/process/CorePusher.java
index 97f51aa..2712b8d 100644
--- a/solr/core/src/java/org/apache/solr/store/blob/process/CorePusher.java
+++ b/solr/core/src/java/org/apache/solr/store/blob/process/CorePusher.java
@@ -98,6 +98,16 @@ public class CorePusher {
// This can happen if another indexing batch comes in and acquires the push lock first and ends up pushing segments
// produced by this indexing batch.
// This optimization saves us from creating somewhat expensive ServerSideMetadata.
+ // Note1! At this point we might not be the leader and shared store might have already received pushes from
+ // new leader. It is still ok to declare success since our indexing batch was correctly pushed earlier
+ // by another thread before the new leader could have pushed its batches.
+ // Note2! It is important to note that we are piggybacking on cached BlobCoreMetadata's generation number here.
+ // A more accurate representation of this optimization would be to add a "lastGenerationPushed" property
+ // to per core cache and update it to whatever generation gets pushed successfully and to -1 on each
+ // successful pull. But that is not necessary since local core will be on blobCoreMetadata's generation
+ // number after pull and on push blobCoreMetadata get generation number from local core.
+ // The reason it is important to call it out here is that BlobCoreMetadata' generation also gets
+ // persisted to shared store which is not the requirement for this optimization.
log.info(String.format("Nothing to push, pushLockTime=%s pushPullData=%s", lockAcquisitionTime, pushPullData.toString()));
return;
}
diff --git a/solr/core/src/java/org/apache/solr/store/shared/SharedCoreConcurrencyController.java b/solr/core/src/java/org/apache/solr/store/shared/SharedCoreConcurrencyController.java
index 9a9d52e..2adde86 100644
--- a/solr/core/src/java/org/apache/solr/store/shared/SharedCoreConcurrencyController.java
+++ b/solr/core/src/java/org/apache/solr/store/shared/SharedCoreConcurrencyController.java
@@ -28,6 +28,7 @@ import org.apache.solr.common.SolrException;
import org.apache.solr.common.StringUtils;
import org.apache.solr.common.cloud.DocCollection;
import org.apache.solr.core.CoreContainer;
+import org.apache.solr.servlet.HttpSolrCall;
import org.apache.solr.store.blob.client.BlobCoreMetadata;
import org.apache.solr.store.blob.metadata.PushPullData;
import org.apache.solr.store.blob.process.CorePullTask;
@@ -42,6 +43,80 @@ import org.slf4j.LoggerFactory;
/**
* This class helps coordinate synchronization of concurrent indexing, pushes and pulls
* happening on a core of a shared collection {@link DocCollection#getSharedIndex()}
+ * The objective of synchronization is not only to be functionally correct but also efficient i.e.
+ * we don't unnecessarily fail things that are happening concurrently on a single node.
+ * Following paragraphs explain those needs and how are those addressed.
+ *
+ * General requirement: The source of truth for shared cores is shared store. We pull from shared store if a core is
+ * locally absent or is stale. We push to shared store at the end of indexing. Queries can only trigger pulls. Whereas,
+ * indexing does a pull (only if the core is stale) then local indexing and finally a push to shared store. It is
+ * important that local indexing always happen on source of truth that was in the shared store when the batch arrived,
+ * so that at the push time new source of truth is whatever was there before plus new local indexing on top of that. In
+ * scenarios below we will talk about concurrent local indexing. Solr already takes care of that. Our job is to make
+ * sure that shared cores play nicely and efficiently with that i.e. we do not end up completely serializing each
+ * indexing batch.
+ *
+ *
+ * Scenario#1: Since leadership can change anytime, we need to protect the source of truth from being incorrectly
+ * overwritten by an indexing batch on ghost leader (was a real leader when received the batch but at the time of push
+ * world has changed).
+ * Solution: We use optimistic concurrency by maintaining a unique identifier (metadataSuffix) in zookeeper that is
+ * changed at each push. At push time we make sure that metadataSuffix is still at the value on which we started the
+ * local indexing i.e. no one else became the leader while we were indexing locally and did a push to shared store.
+ * If metadataSuffix in zookeeper happens to have changed we fail the indexing batch.
+ * {@link SharedShardVersionMetadata} is representative of that unique identifier in zookeeper.
+ *
+ * Scenario#2: A pull is going on and another pull comes in, they will step over each other. Even if we pull into a
+ * new index directory on each pull we will have to ensure that pull corresponding to later contents of shared store wins.
+ * Solution: We serialize pulls.
+ * {@link #getCorePullLock(String, String, String)}'s write lock is used for that purpose.
+ *
+ * Scenario#3: The core is stale and an indexing batch comes in, refreshes it and proceeds to local indexing. After the
+ * local indexing is finished and before the push to shared store, a query comes in and triggers the pull. Because of
+ * local indexing, local contents are different from shared store, pull will go ahead and overwrite them with what is on
+ * shared store(it being the source of truth). Now the indexing batch will go ahead and finish its push and declare
+ * success. The only problem is it did not push anything since whatever it had indexed locally was undone by the query pull.
+ * Note! In this scenario query pull can also be replaced by a pull of another concurrent indexing request.
+ * Solution: We need a cache telling us the last shared store state that was pulled in successfully i.e. a value of
+ * metadataSuffix that we also for conditional zk update at push time(as explained earlier scenario#1). This cache not
+ * only provides functional correctness but also prevents from unnecessary comparison of local and shared store contents.
+ * {@link #coresVersionMetadata} is that per core cache.
+ *
+ * Scenario#4: Local indexing is finished and server goes into a long GC pause and loses leadership. The new leader makes
+ * progress on indexing. The old leader comes out of GC pause. Before the push of stalled batch resumes, a query comes
+ * in and triggers a pull. That pull will overwrite local index with whatever the new leader has pushed to shared store.
+ * We will lose the local indexing of stalled batch. Because of successful pull, our cache now matches with what's in
+ * shared store. The push will be successful but end up pushing nothing.
+ * Note! In this scenario query pull can also be replaced by a pull of another concurrent indexing request(assuming old
+ * leader becomes leader again).
+ * Solution: We need to protect indexing(including pushes) from pulls.
+ * {@link #getCorePullLock(String, String, String)}'s read lock is used for that protection.
+ *
+ * Scenario#5: Two concurrent indexing threads finish up their local indexing and are ready to push. One pushes and advances
+ * metadaSuffix in zookeeper. The second one being on previous cached version would fail on conditional zk update. This
+ * will functionally be correct but inefficient since second's batch progress will go to waste.
+ * Solution: We serialize pushes to shared store. So that concurrent pushes on the same node do not fail.
+ * {@link #getCorePushLock(String, String, String)} is used for that serialization.
+ * Corollary: When an indexing thread goes into push phase it will not only push its own batch rather it will end up pushing
+ * all other batches that have been committed up till that point. Later on, the threads responsible for other batches will
+ * just do a touch and go(assuming no other batch came afterwards). In other words, push phase of indexing can be described
+ * as: "push if there is something to be pushed (my batch could be part of it or someone else has already pushed that for me)"
+ *
+ *
+ * Miscellaneous Notes:
+ * 1. Since we are doing synchronous pushes of files from index directory to shared store we only support hard commits
+ * for shared collections. We ensure that each batch has a hard commit {@link HttpSolrCall#addCommitIfAbsent()}.
+ * 2. In steady state a leader does not need to consult zookeeper all the time since it is the only one updating the
+ * metadataSuffix. For that we rely on {@link SharedCoreVersionMetadata#softGuaranteeOfEquality}. Leaders set it to
+ * "true" when they successfully pull from or push to shared store. If we happen to be incorrect, conditional update
+ * of zookeeper will fail at push time.
+ * 3. We also cache {@link SharedCoreVersionMetadata#blobCoreMetadata} so that in steady state leaders don't need to read
+ * it from shared store at push time.
+ * 4. {@link #coresVersionMetadata} cache:
+ * a. It is a {@link ConcurrentHashMap} and consists of immutable values, therefore, updating it is an atomic operation.
+ * b. We initialize it only once with values never to be found in zookeeper{@link #initializeCoreVersionMetadata(String, String, String)}.
+ * c. We can only update {@link #coresVersionMetadata} under either a pull write lock or push lock along with pull read lock.
+ * {@link #updateCoreVersionMetadata(String, String, String, SharedCoreVersionMetadata, SharedCoreVersionMetadata)}
*/
public class SharedCoreConcurrencyController {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
@@ -155,6 +230,11 @@ public class SharedCoreConcurrencyController {
}
private void updateCoreVersionMetadata(String collectionName, String shardName, String coreName, SharedCoreVersionMetadata currentMetadata, SharedCoreVersionMetadata updatedMetadata) {
+ // either have a pull write lock or push lock along with pull read lock
+ // TODO: Core split does not acquire these locks because as of current understanding there is no concurrency involved.
+ // But at least for semantic correctness we should acquire locks in that path too and enable this assert.
+ // assert (currentMetadata.getCorePullLock().getWriteHoldCount() > 0) ||
+ // (currentMetadata.getCorePushLock().getHoldCount() > 0 && currentMetadata.getCorePullLock().getReadHoldCount() > 0);
log.info(String.format("updateCoreVersionMetadata: collection=%s shard=%s core=%s current={%s} updated={%s}",
collectionName, shardName, coreName, currentMetadata.toString(), updatedMetadata.toString()));
coresVersionMetadata.put(coreName, updatedMetadata);
@@ -172,6 +252,7 @@ public class SharedCoreConcurrencyController {
private SharedCoreVersionMetadata initializeCoreVersionMetadata(String collectionName, String shardName, String coreName) {
// computeIfAbsent to ensure we only do single initialization
return coresVersionMetadata.computeIfAbsent(coreName, k -> {
+ // TODO: This metadata should not be created here. It should only be created on shard creation or zk recovery time.
ensureShardVersionMetadataNodeExists(collectionName, shardName);
// a value not to be found as a zk node version
int version = -1;
@@ -241,7 +322,7 @@ public class SharedCoreConcurrencyController {
* {@link CorePusher#pushCoreToBlob(PushPullData)}
* and pull
* {@link BlobStoreUtils#syncLocalCoreWithSharedStore(String, String, String, CoreContainer, SharedShardVersionMetadata, boolean)}
- * {@link CorePullTask#pullCoreFromBlob()}
+ * {@link CorePullTask#pullCoreFromBlob(boolean)} ()}
* since followers cannot index. In presence of this guarantee we can skip consulting zookeeper before processing an indexing batch.
*/
private final boolean softGuaranteeOfEquality;
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 5caf0f8..b43aa6c 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
@@ -1125,7 +1125,7 @@ public class DistributedZkUpdateProcessor extends DistributedUpdateProcessor {
}
// this lock acquire/release logic is built on the assumption that one particular instance of this processor
- // will solely be consumed by a single thread.
+ // will solely be consumed by a single thread. And all the documents of indexing batch will be processed by this one instance.
// Following pull logic should only run once before the first document of indexing batch(add/delete) is processed by this processor
if (corePullLock != null) {
// we already have a lock i.e. we have already read from the shared store (if needed)
@@ -1136,90 +1136,84 @@ public class DistributedZkUpdateProcessor extends DistributedUpdateProcessor {
SharedCoreConcurrencyController concurrencyController = coreContainer.getSharedStoreManager().getSharedCoreConcurrencyController();
concurrencyController.recordState(collectionName, shardName, coreName, SharedCoreStage.IndexingBatchReceived);
corePullLock = concurrencyController.getCorePullLock(collectionName, shardName, coreName);
- // acquire lock for the whole duration of update
- // it will be release in close method
- corePullLock.readLock().lock();
// from this point on wards we should always exit this method with read lock (no matter failure or what)
+ try {
+ SharedCoreVersionMetadata coreVersionMetadata = concurrencyController.getCoreVersionMetadata(collectionName, shardName, coreName);
+ /**
+ * we only need to sync if there is no soft guarantee of being in sync.
+ * if there is one we will rely on that, and if we turned out to be wrong indexing will fail at push time
+ * and will remove this guarantee in {@link CorePusher#pushCoreToBlob(PushPullData)}
+ */
+ if (!coreVersionMetadata.isSoftGuaranteeOfEquality()) {
+ SharedShardMetadataController metadataController = coreContainer.getSharedStoreManager().getSharedShardMetadataController();
+ SharedShardVersionMetadata shardVersionMetadata = metadataController.readMetadataValue(collectionName, shardName);
+ if (!concurrencyController.areVersionsEqual(coreVersionMetadata, shardVersionMetadata)) {
+ acquireWriteLockAndPull(collectionName, shardName, coreName, coreContainer);
+ }
+ }
+ } finally {
+ // acquire lock for the whole duration of update
+ // we should always leave with read lock acquired(failure or success), since it is the job of close method to release it
+ corePullLock.readLock().lock();
+ concurrencyController.recordState(collectionName, shardName, coreName, SharedCoreStage.LocalIndexingStarted);
+ }
+ }
- SharedCoreVersionMetadata coreVersionMetadata = concurrencyController.getCoreVersionMetadata(collectionName, shardName, coreName);
- /**
- * we only need to sync if there is no soft guarantee of being in sync.
- * if there is one we will rely on that, and if we turned out to be wrong indexing will fail at push time
- * and will remove this guarantee in {@link CorePusher#pushCoreToBlob(PushPullData)}
- */
- if (!coreVersionMetadata.isSoftGuaranteeOfEquality()) {
- SharedShardMetadataController metadataController = coreContainer.getSharedStoreManager().getSharedShardMetadataController();
- SharedShardVersionMetadata shardVersionMetadata = metadataController.readMetadataValue(collectionName, shardName);
- if (!concurrencyController.areVersionsEqual(coreVersionMetadata, shardVersionMetadata)) {
- // we need to pull before indexing therefore we need to upgrade to write lock
- // we have to release read lock before we can acquire write lock
- corePullLock.readLock().unlock();
- boolean reacquireReadLock = true;
- try {
- // There is a likelihood that many indexing requests came at once and realized we are out of sync.
- // They all would try to acquire write lock. One of them makes progress to pull from shared store.
- // After that regular indexing will see soft guarantee of equality and moves straight to indexing
- // under read lock. Now it is possible that new indexing keeps coming in and read lock is never free.
- // In that case the poor guys that came in earlier and wanted to pull will still be struggling(starving) to
- // acquire write lock. Since we know that write lock is only needed by one to do the work, we will
- // try time boxed acquisition and in case of failed acquisition we will see if some one else has already completed the pull.
- // We will make few attempts before we bail out. Ideally bail out scenario should never happen.
- // If it does then either we are too slow in pulling and can tune following parameters or something else is wrong.
- int attempt = 1;
- while (true) {
- try {
- // try acquiring write lock
- if (corePullLock.writeLock().tryLock(SharedCoreConcurrencyController.SECONDS_TO_WAIT_INDEXING_PULL_WRITE_LOCK, TimeUnit.SECONDS)) {
- try {
- // in between upgrading locks things might have updated, should reestablish if pull is still needed
- coreVersionMetadata = concurrencyController.getCoreVersionMetadata(collectionName, shardName, coreName);
- if (!coreVersionMetadata.isSoftGuaranteeOfEquality()) {
- shardVersionMetadata = metadataController.readMetadataValue(collectionName, shardName);
- if (!concurrencyController.areVersionsEqual(coreVersionMetadata, shardVersionMetadata)) {
- BlobStoreUtils.syncLocalCoreWithSharedStore(collectionName, coreName, shardName, coreContainer, shardVersionMetadata, /* isLeaderSyncing */true);
- }
- }
- // reacquire read lock for the remainder of indexing before releasing write lock that was acquired for pull part
- corePullLock.readLock().lock();
- reacquireReadLock = false;
- } finally {
- corePullLock.writeLock().unlock();
- }
- // write lock acquisition was successful and we are in sync with shared store
- break;
- } else {
- // we could not acquire write lock but see if some other thread has already done the pulling
- coreVersionMetadata = concurrencyController.getCoreVersionMetadata(collectionName, shardName, coreName);
- if (coreVersionMetadata.isSoftGuaranteeOfEquality()) {
- log.info(String.format("Indexing thread waited to acquire to write lock and could not. " +
- "But someone else has done the pulling so we are good. attempt=%s collection=%s shard=%s core=%s",
- attempt, collectionName, shardName, coreName));
- break;
- }
- // no one else has pulled yet either, lets make another attempt ourselves
- attempt++;
- if (attempt > SharedCoreConcurrencyController.MAX_ATTEMPTS_INDEXING_PULL_WRITE_LOCK) {
- throw new SolrException(ErrorCode.SERVER_ERROR, String.format("Indexing thread failed to acquire write lock for pull in %s seconds. " +
- "And no one else either has done the pull during that time. collection=%s shard=%s core=%s",
- Integer.toString(SharedCoreConcurrencyController.SECONDS_TO_WAIT_INDEXING_PULL_WRITE_LOCK * SharedCoreConcurrencyController.MAX_ATTEMPTS_INDEXING_PULL_WRITE_LOCK),
- collectionName, shardName, coreName));
- }
+ private void acquireWriteLockAndPull(String collectionName, String shardName, String coreName, CoreContainer coreContainer) {
+ // There is a likelihood that many indexing requests came at once and realized we are out of sync.
+ // They all would try to acquire write lock. One of them makes progress to pull from shared store.
+ // After that regular indexing will see soft guarantee of equality and moves straight to indexing
+ // under read lock. Now it is possible that new indexing keeps coming in and read lock is never free.
+ // In that case the poor guys that came in earlier and wanted to pull will still be struggling(starving) to
+ // acquire write lock. Since we know that write lock is only needed by one to do the work, we will
+ // try time boxed acquisition and in case of failed acquisition we will see if some one else has already completed the pull.
+ // We will make few attempts before we bail out. Ideally bail out scenario should never happen.
+ // If it does then either we are too slow in pulling and can tune following parameters or something else is wrong.
+ int attempt = 1;
+ while (true) {
+ SharedCoreConcurrencyController concurrencyController = coreContainer.getSharedStoreManager().getSharedCoreConcurrencyController();
+ try {
+ // try acquiring write lock
+ if (corePullLock.writeLock().tryLock(SharedCoreConcurrencyController.SECONDS_TO_WAIT_INDEXING_PULL_WRITE_LOCK, TimeUnit.SECONDS)) {
+ try {
+ // while acquiring write lock things might have updated, should reestablish if pull is still needed
+ SharedCoreVersionMetadata coreVersionMetadata = concurrencyController.getCoreVersionMetadata(collectionName, shardName, coreName);
+ if (!coreVersionMetadata.isSoftGuaranteeOfEquality()) {
+ SharedShardMetadataController metadataController = coreContainer.getSharedStoreManager().getSharedShardMetadataController();
+ SharedShardVersionMetadata shardVersionMetadata = metadataController.readMetadataValue(collectionName, shardName);
+ if (!concurrencyController.areVersionsEqual(coreVersionMetadata, shardVersionMetadata)) {
+ BlobStoreUtils.syncLocalCoreWithSharedStore(collectionName, coreName, shardName, coreContainer, shardVersionMetadata, /* isLeaderSyncing */true);
}
- } catch (InterruptedException ie) {
- Thread.currentThread().interrupt();
- throw new SolrException(ErrorCode.SERVER_ERROR, String.format("Indexing thread interrupted while trying to acquire pull write lock." +
- " collection=%s shard=%s core=%s", collectionName, shardName, coreName), ie);
}
+ } finally {
+ corePullLock.writeLock().unlock();
}
- } finally {
- // we should always leave with read lock acquired(failure or success), since it is the job of close method to release it
- if (reacquireReadLock) {
- corePullLock.readLock().lock();
+ // write lock acquisition was successful and we are in sync with shared store
+ break;
+ } else {
+ // we could not acquire write lock but see if some other thread has already done the pulling
+ SharedCoreVersionMetadata coreVersionMetadata = concurrencyController.getCoreVersionMetadata(collectionName, shardName, coreName);
+ if (coreVersionMetadata.isSoftGuaranteeOfEquality()) {
+ log.info(String.format("Indexing thread waited to acquire to write lock and could not. " +
+ "But someone else has done the pulling so we are good. attempt=%s collection=%s shard=%s core=%s",
+ attempt, collectionName, shardName, coreName));
+ break;
+ }
+ // no one else has pulled yet either, lets make another attempt ourselves
+ attempt++;
+ if (attempt > SharedCoreConcurrencyController.MAX_ATTEMPTS_INDEXING_PULL_WRITE_LOCK) {
+ throw new SolrException(ErrorCode.SERVER_ERROR, String.format("Indexing thread failed to acquire write lock for pull in %s seconds. " +
+ "And no one else either has done the pull during that time. collection=%s shard=%s core=%s",
+ Integer.toString(SharedCoreConcurrencyController.SECONDS_TO_WAIT_INDEXING_PULL_WRITE_LOCK * SharedCoreConcurrencyController.MAX_ATTEMPTS_INDEXING_PULL_WRITE_LOCK),
+ collectionName, shardName, coreName));
}
}
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ throw new SolrException(ErrorCode.SERVER_ERROR, String.format("Indexing thread interrupted while trying to acquire pull write lock." +
+ " collection=%s shard=%s core=%s", collectionName, shardName, coreName), ie);
}
}
- concurrencyController.recordState(collectionName, shardName, coreName, SharedCoreStage.LocalIndexingStarted);
}
// TODO: optionally fail if n replicas are not reached...