Introduce clean transition on primary promotion #24925

jasontedor · 2017-05-28T16:46:40Z

This commit introduces a clean transition from the old primary term to the new primary term when a replica is promoted primary. To accomplish this, we delay all operations before incrementing the primary term. The delay is guaranteed to be in place before we increment the term, and then all operations that are delayed are executed after the delay is removed which asynchronously happens on another thread. This thread does not progress until in-flight operations that were executing are completed, and after these operations drain, the delayed operations re-acquire permits and are executed.

Relates #10708

This commit introduces a clean transition from the old primary term to the new primary term when a replica is promoted primary. To accomplish this, we delay all operations before incrementing the primary term. The delay is guaranteed to be in place before we increment the term, and then all operations that are delayed are executed after the delay is removed which asynchronously happens on another thread. This thread does not progress until in-flight operations that were executing are completed, and after these operations drain, the delayed operations re-acquire permits and are executed.

bleskes

Thanks @jasontedor . Basics flow looks good. I left some suggestions.

bleskes · 2017-05-29T11:38:39Z

core/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java

    private final ShardId shardId;
    private final Logger logger;
    private final ThreadPool threadPool;

    private static final int TOTAL_PERMITS = Integer.MAX_VALUE;
-    // fair semaphore to ensure that blockOperations() does not starve under thread contention
-    final Semaphore semaphore = new Semaphore(TOTAL_PERMITS, true);
+    final Semaphore semaphore = new Semaphore(Integer.MAX_VALUE, true); // fair to ensure a blocking thread is not starved


why switch from the TOTAL_PERMITS field?

I can't even up come with a plausible theory how that happened, that was completely unintentional.

Oh, I know what happened! I was playing around with the idea of using a shared/exclusive lock so removed the semaphore and then added it back manually (so with this change) after I decided against it (because it doesn't give an easy way to see how many threads hold the shared lock which we use in testing).

bleskes · 2017-05-29T11:39:51Z

core/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java

     * @throws IndexShardClosedException if operation permit has been closed
     */
-    public <E extends Exception> void blockOperations(long timeout, TimeUnit timeUnit, CheckedRunnable<E> onBlocked) throws
-        InterruptedException, TimeoutException, E {
+    <E extends Exception> void syncBlockOperations(


nit - I like the blockOperations naming... syncX is just one letter away from asyncX..

bleskes · 2017-05-29T11:42:10Z

core/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java

@@ -92,6 +154,7 @@ public void close() {
            synchronized (this) {
                queuedActions = delayedOperations;
                delayedOperations = null;
+                delayed = false;


assert again that it was true?

bleskes · 2017-05-29T11:42:50Z

core/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java

+        }
+    }
+
+    private <E extends Exception> void doBlockOperations(


can we doc / make it clearer in the name that this methods releases the delayed operations?

better yet, how about the following? This way it's clear how a delayOperations call is matched by the undelay logic.

diff --git a/core/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java b/core/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java index 7fc56a1..f1c8b0d 100644 --- a/core/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java +++ b/core/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java @@ -27,6 +27,7 @@ import org.elasticsearch.action.support.ThreadedActionListener; import org.elasticsearch.common.CheckedRunnable; import org.elasticsearch.common.Nullable; import org.elasticsearch.common.lease.Releasable; +import org.elasticsearch.common.util.concurrent.AbstractRunnable; import org.elasticsearch.common.util.concurrent.ThreadContext.StoredContext; import org.elasticsearch.threadpool.ThreadPool; @@ -95,7 +96,11 @@ final class IndexShardOperationPermits implements Closeable { throw new IndexShardClosedException(shardId); } delayOperations(); - doBlockOperations(timeout, timeUnit, onBlocked); + try { + doBlockOperations(timeout, timeUnit, onBlocked); + } finally { + releasedDelayedOperations(); + } } /** @@ -110,12 +115,21 @@ final class IndexShardOperationPermits implements Closeable { */ <E extends Exception> void asyncBlockOperations(final long timeout, final TimeUnit timeUnit, final CheckedRunnable<E> onBlocked) { delayOperations(); - threadPool.executor(ThreadPool.Names.GENERIC).execute(() -> { - try { - doBlockOperations(timeout, timeUnit, onBlocked); - } catch (final Exception e) { + threadPool.executor(ThreadPool.Names.GENERIC).execute(new AbstractRunnable() { + @Override + public void onFailure(Exception e) { throw new RuntimeException(e); } + + @Override + protected void doRun() throws Exception { + doBlockOperations(timeout, timeUnit, onBlocked); + } + + @Override + public void onAfter() { + releasedDelayedOperations(); + } }); } @@ -150,25 +164,30 @@ final class IndexShardOperationPermits implements Closeable { throw new TimeoutException("timed out during blockOperations"); } } finally { - final List<ActionListener<Releasable>> queuedActions; - synchronized (this) { - queuedActions = delayedOperations; - delayedOperations = null; - delayed = false; - } - if (queuedActions != null) { - // Try acquiring permits on fresh thread (for two reasons): - // - blockOperations can be called on recovery thread which can be expected to be interrupted when recovery is cancelled. - // Interruptions are bad here as permit acquisition will throw an InterruptedException which will be swallowed by - // ThreadedActionListener if the queue of the thread pool on which it submits is full. - // - if permit is acquired and queue of the thread pool which the ThreadedActionListener uses is full, the onFailure - // handler is executed on the calling thread. This should not be the recovery thread as it would delay the recovery. - threadPool.executor(ThreadPool.Names.GENERIC).execute(() -> { - for (ActionListener<Releasable> queuedAction : queuedActions) { - acquire(queuedAction, null, false); - } - }); - } + releasedDelayedOperations(); + } + } + + private void releasedDelayedOperations() { + final List<ActionListener<Releasable>> queuedActions; + synchronized (this) { + assert delayed; + queuedActions = delayedOperations; + delayedOperations = null; + delayed = false; + } + if (queuedActions != null) { + // Try acquiring permits on fresh thread (for two reasons): + // - blockOperations can be called on recovery thread which can be expected to be interrupted when recovery is cancelled. + // Interruptions are bad here as permit acquisition will throw an InterruptedException which will be swallowed by + // ThreadedActionListener if the queue of the thread pool on which it submits is full. + // - if permit is acquired and queue of the thread pool which the ThreadedActionListener uses is full, the onFailure + // handler is executed on the calling thread. This should not be the recovery thread as it would delay the recovery. + threadPool.executor(ThreadPool.Names.GENERIC).execute(() -> { + for (ActionListener<Releasable> queuedAction : queuedActions) { + acquire(queuedAction, null, false); + } + }); } }

bleskes · 2017-05-29T11:52:15Z

core/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java

+            try {
+                doBlockOperations(timeout, timeUnit, onBlocked);
+            } catch (final Exception e) {
+                throw new RuntimeException(e);


I wonder if we should fail the shard if something goes wrong here... maybe we should make the CheckedRunnable it's own AbstractRunnable so we can call onFailure and let the caller decide what to do?

bleskes · 2017-05-29T11:54:03Z

core/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java

+            if (delayed) {
+                throw new IllegalStateException("operations are already delayed");
+            } else {
+                delayed = true;


this can pre-create the delayedOperations list. Stronger yet, maybe we should use the existence of the list as a marker that ops should be delayed?

I strongly prefer the explicit boolean flag rather than implicit dependency on the list being null versus not null.

Fair enough. We can also maybe just have a something we can drain and let it be a final field.

bleskes · 2017-05-29T12:00:16Z

core/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java

        try {
            synchronized (this) {
                releasable = tryAcquire();
                if (releasable == null) {
-                    // blockOperations is executing, this operation will be retried by blockOperations once it finishes
+                    assert delayed;


can we used delay to control the flow here? I think it be easier to read that to make tryAcquire check on delay. WDYT?

diff --git a/core/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java b/core/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java index 7fc56a1..ebab08d 100644 --- a/core/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java +++ b/core/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java @@ -190,10 +190,7 @@ final class IndexShardOperationPermits implements Closeable { final Releasable releasable; try { synchronized (this) { - releasable = tryAcquire(); - if (releasable == null) { - assert delayed; - // operations are delayed, this operation will be retried by doBlockOperations once the delay is remoked + if (delayed) { if (delayedOperations == null) { delayedOperations = new ArrayList<>(); } @@ -205,6 +202,13 @@ final class IndexShardOperationPermits implements Closeable { } else { delayedOperations.add(new ContextPreservingActionListener<>(contextSupplier, onAcquired)); } + releasable = null; + } else { + releasable = tryAcquire(); + assert releasable != null; + } + if (releasable == null) { + assert delayed; return; } } @@ -212,7 +216,10 @@ final class IndexShardOperationPermits implements Closeable { onAcquired.onFailure(e); return; } - onAcquired.onResponse(releasable); + if (releasable != null) { + // if it's null operations are delayed, this operation will be retried by doBlockOperations once the delay is remoked + onAcquired.onResponse(releasable); + } } @Nullable private Releasable tryAcquire() throws InterruptedException {

I think we can take it a little more, see my push.

jasontedor · 2017-05-29T18:33:11Z

Back to you @bleskes.

* master: Add a second refresh to concurrent relocation test Add a dummy_index to upgrade tests to ensure we recover fine with replicas (elastic#24937) Rework bwc snapshot projects to build up to two bwc versions (elastic#24870) Move the IndexDeletionPolicy to be engine internal (elastic#24930) [Tests] Harden InternalExtendedStatsTests (elastic#24934) TCorrecting api name (elastic#24924) Add search method to high level REST client (elastic#24796) Add fromXContent method to ClearScrollResponse (elastic#24909) ClearScrollRequest to implement ToXContentObject (elastic#24907) SearchScrollRequest to implement ToXContentObject (elastic#24906) Fix bug in weight computation for query cache

jasontedor · 2017-05-30T00:11:21Z

@bleskes I think that a4d2bf3 is necessary. What do you think?

* master: Fix typo in comment in ReplicationOperation.java Prevent Index & Delete request primaryTerm getter/setter, setShardId setter Drop name from TokenizerFactory (elastic#24869) Correctly set doc_count when MovAvg "predicts" values on existing buckets (elastic#24892) Handle primary failure handling replica response Add missing word to terms-query.asciidoc (elastic#24960) Correct some spelling in match-phrase-prefix docs (elastic#24956) testConcurrentWriteViewsAndSnapshot shouldn't flush concurrently [TEST] Fix FieldSortIT failures Add doc_count to ParsedMatrixStats (elastic#24952) Add document count to Matrix Stats aggregation response (elastic#24776) Fix script field sort returning Double.MAX_VALUE for all documents (elastic#24942)

bleskes

LGTM. Left some minor suggestions.

bleskes · 2017-05-30T10:18:22Z

core/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java

-                }
+            doBlockOperations(timeout, timeUnit, onBlocked);
+        } finally {
+            releasedDelayedOperations();


nit - an extra d? releasedDelayed..

bleskes · 2017-05-30T10:23:31Z

core/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java

+            final long timeout,
+            final TimeUnit timeUnit,
+            final CheckedRunnable<E> onBlocked) throws InterruptedException, TimeoutException, E {
+        if (Assertions.ENABLED) {
            synchronized (this) {


can you add a comment this is for visibility?

bleskes · 2017-05-30T10:25:46Z

core/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java

                    } else {
                        delayedOperations.add(new ContextPreservingActionListener<>(contextSupplier, onAcquired));
                    }
                    return;
+                } else {
+                    releasable = tryAcquire();


should we just make this be called acquire and make it never return null (but blow up if it can't acquire (as it is not non-delayed)?

Okay, I will do that in a follow-up immediately after this PR.

I opened the follow-up: #24971

jasontedor added :Sequence IDs >enhancement v6.0.0 labels May 28, 2017

jasontedor requested a review from bleskes May 28, 2017 16:46

jasontedor force-pushed the primary-promotion-transition branch from 4f7ba68 to 8d0b64a Compare May 28, 2017 16:47

bleskes mentioned this pull request May 28, 2017

Add Sequence Numbers to write operations #10708

Closed

64 tasks

jasontedor force-pushed the primary-promotion-transition branch from 8d0b64a to 3b4a5c0 Compare May 28, 2017 17:55

bleskes suggested changes May 29, 2017

View reviewed changes

jasontedor added 5 commits May 29, 2017 12:50

Separate block/release

069745e

Fix permits

46978ee

Failure listener

b420228

Simpler

ebb9f3d

Delayed

e9224dd

jasontedor added 4 commits May 29, 2017 14:38

Fix comment formatting

73c48fe

Comments

f04ff7e

Latch to ensure ordering

a4d2bf3

jasontedor added 3 commits May 29, 2017 20:37

Formatting so I can stop fighting IntelliJ

27b4f85

Rename

419528a

bleskes approved these changes May 30, 2017

View reviewed changes

Last comments

632af5a

jasontedor merged commit ddbc468 into elastic:master May 30, 2017

jasontedor mentioned this pull request May 30, 2017

Fill gaps on primary promotion #24945

Merged

jasontedor deleted the primary-promotion-transition branch May 30, 2017 15:47

jasontedor mentioned this pull request May 30, 2017

Clarify acquiring index shard permit #24971

Merged

jasontedor mentioned this pull request Jun 2, 2017

Handle already closed while filling gaps #25021

Merged

clintongormley added v6.0.0-alpha2 and removed v6.0.0 labels Jun 6, 2017

clintongormley added :Distributed/Engine Anything around managing Lucene and the Translog in an open shard. and removed :Sequence IDs labels Feb 14, 2018

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Introduce clean transition on primary promotion #24925

Introduce clean transition on primary promotion #24925

jasontedor commented May 28, 2017

bleskes left a comment

bleskes May 29, 2017

jasontedor May 29, 2017

jasontedor May 29, 2017 •

edited

bleskes May 29, 2017

jasontedor May 29, 2017

bleskes May 29, 2017

jasontedor May 29, 2017

bleskes May 29, 2017

bleskes May 29, 2017

jasontedor May 29, 2017

bleskes May 29, 2017

jasontedor May 29, 2017

bleskes May 29, 2017

jasontedor May 29, 2017

bleskes May 29, 2017

jasontedor May 29, 2017

bleskes May 29, 2017

jasontedor May 29, 2017

jasontedor commented May 29, 2017

jasontedor commented May 30, 2017

bleskes left a comment

bleskes May 30, 2017

jasontedor May 30, 2017

bleskes May 30, 2017

jasontedor May 30, 2017

bleskes May 30, 2017

jasontedor May 30, 2017

jasontedor May 30, 2017

Introduce clean transition on primary promotion #24925

Introduce clean transition on primary promotion #24925

Conversation

jasontedor commented May 28, 2017

bleskes left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jasontedor May 29, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jasontedor commented May 29, 2017

jasontedor commented May 30, 2017

bleskes left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jasontedor May 29, 2017 •

edited