diff --git a/java/com/google/gerrit/server/notedb/RepoSequence.java b/java/com/google/gerrit/server/notedb/RepoSequence.java index 1c33a68450..4d96565eb1 100644 --- a/java/com/google/gerrit/server/notedb/RepoSequence.java +++ b/java/com/google/gerrit/server/notedb/RepoSequence.java @@ -29,8 +29,10 @@ import com.github.rholder.retry.WaitStrategies; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Throwables; import com.google.common.collect.ImmutableList; +import com.google.common.collect.Iterables; import com.google.common.util.concurrent.Runnables; import com.google.gerrit.exceptions.StorageException; +import com.google.gerrit.git.LockFailureException; import com.google.gerrit.git.RefUpdateUtil; import com.google.gerrit.reviewdb.client.Project; import com.google.gerrit.reviewdb.client.RefNames; @@ -40,7 +42,6 @@ import java.io.IOException; import java.util.ArrayList; import java.util.List; import java.util.Optional; -import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Lock; @@ -69,9 +70,12 @@ public class RepoSequence { } @VisibleForTesting - static RetryerBuilder retryerBuilder() { - return RetryerBuilder.newBuilder() - .retryIfResult(ru -> ru != null && RefUpdate.Result.LOCK_FAILURE.equals(ru.getResult())) + static RetryerBuilder> retryerBuilder() { + return RetryerBuilder.>newBuilder() + .retryIfException( + t -> + t instanceof StorageException + && ((StorageException) t).getCause() instanceof LockFailureException) .withWaitStrategy( WaitStrategies.join( WaitStrategies.exponentialWait(5, TimeUnit.SECONDS), @@ -79,7 +83,7 @@ public class RepoSequence { .withStopStrategy(StopStrategies.stopAfterDelay(30, TimeUnit.SECONDS)); } - private static final Retryer RETRYER = retryerBuilder().build(); + private static final Retryer> RETRYER = retryerBuilder().build(); private final GitRepositoryManager repoManager; private final GitReferenceUpdated gitRefUpdated; @@ -89,7 +93,7 @@ public class RepoSequence { private final int floor; private final int batchSize; private final Runnable afterReadRef; - private final Retryer retryer; + private final Retryer> retryer; // Protects all non-final fields. private final Lock counterLock; @@ -147,7 +151,7 @@ public class RepoSequence { Seed seed, int batchSize, Runnable afterReadRef, - Retryer retryer) { + Retryer> retryer) { this(repoManager, gitRefUpdated, projectName, name, seed, batchSize, afterReadRef, retryer, 0); } @@ -159,7 +163,7 @@ public class RepoSequence { Seed seed, int batchSize, Runnable afterReadRef, - Retryer retryer, + Retryer> retryer, int floor) { this.repoManager = requireNonNull(repoManager, "repoManager"); this.gitRefUpdated = requireNonNull(gitRefUpdated, "gitRefUpdated"); @@ -184,78 +188,85 @@ public class RepoSequence { counterLock = new ReentrantLock(true); } + /** + * Retrieves the next available sequence number. + * + *

This method is thread-safe. + * + * @return the next available sequence number + */ public int next() { - counterLock.lock(); - try { - if (counter >= limit) { - acquire(batchSize); - } - return counter++; - } finally { - counterLock.unlock(); - } + return Iterables.getOnlyElement(next(1)); } + /** + * Retrieves the next N available sequence number. + * + *

This method is thread-safe. + * + * @param count the number of sequence numbers which should be returned + * @return the next N available sequence numbers + */ public ImmutableList next(int count) { if (count == 0) { return ImmutableList.of(); } checkArgument(count > 0, "count is negative: %s", count); - counterLock.lock(); - try { - List ids = new ArrayList<>(count); - while (counter < limit) { - ids.add(counter++); - if (ids.size() == count) { - return ImmutableList.copyOf(ids); - } - } - acquire(Math.max(count - ids.size(), batchSize)); - while (ids.size() < count) { - ids.add(counter++); - } - return ImmutableList.copyOf(ids); - } finally { - counterLock.unlock(); - } - } - private void acquire(int count) { - try (Repository repo = repoManager.openRepository(projectName); - RevWalk rw = new RevWalk(repo)) { - TryAcquire attempt = new TryAcquire(repo, rw, count); - RefUpdateUtil.checkResult(retryer.call(attempt)); - counter = attempt.next; - limit = counter + count; - acquireCount++; + try { + return retryer.call( + () -> { + counterLock.lock(); + try { + if (count == 1) { + if (counter >= limit) { + acquire(batchSize); + } + return ImmutableList.of(counter++); + } + + List ids = new ArrayList<>(count); + while (counter < limit) { + ids.add(counter++); + if (ids.size() == count) { + return ImmutableList.copyOf(ids); + } + } + acquire(Math.max(count - ids.size(), batchSize)); + while (ids.size() < count) { + ids.add(counter++); + } + return ImmutableList.copyOf(ids); + } finally { + counterLock.unlock(); + } + }); } catch (ExecutionException | RetryException e) { if (e.getCause() != null) { Throwables.throwIfInstanceOf(e.getCause(), StorageException.class); } throw new StorageException(e); - } catch (IOException e) { - throw new StorageException(e); } } - private class TryAcquire implements Callable { - private final Repository repo; - private final RevWalk rw; - private final int count; - - private int next; - - private TryAcquire(Repository repo, RevWalk rw, int count) { - this.repo = repo; - this.rw = rw; - this.count = count; - } - - @Override - public RefUpdate call() throws Exception { + /** + * Updates the next available sequence number in NoteDb in order to have a batch of sequence + * numbers available that can be handed out. {@link #counter} stores the next sequence number that + * can be handed out. When {@link #limit} is reached a new batch of sequence numbers needs to be + * retrieved by calling this method. + * + *

Note: Callers are required to acquire the {@link #counterLock} before + * calling this method. + * + * @param count the number of sequence numbers which should be retrieved + */ + private void acquire(int count) { + try (Repository repo = repoManager.openRepository(projectName); + RevWalk rw = new RevWalk(repo)) { Optional blob = IntBlob.parse(repo, refName, rw); afterReadRef.run(); ObjectId oldId; + int next; if (!blob.isPresent()) { oldId = ObjectId.zeroId(); next = seed.get(); @@ -264,7 +275,14 @@ public class RepoSequence { next = blob.get().value(); } next = Math.max(floor, next); - return IntBlob.tryStore(repo, rw, projectName, refName, oldId, next + count, gitRefUpdated); + RefUpdate refUpdate = + IntBlob.tryStore(repo, rw, projectName, refName, oldId, next + count, gitRefUpdated); + RefUpdateUtil.checkResult(refUpdate); + counter = next; + limit = counter + count; + acquireCount++; + } catch (IOException e) { + throw new StorageException(e); } } diff --git a/javatests/com/google/gerrit/server/notedb/RepoSequenceTest.java b/javatests/com/google/gerrit/server/notedb/RepoSequenceTest.java index f088a79164..6baa3e4544 100644 --- a/javatests/com/google/gerrit/server/notedb/RepoSequenceTest.java +++ b/javatests/com/google/gerrit/server/notedb/RepoSequenceTest.java @@ -20,9 +20,12 @@ import static com.google.gerrit.testing.GerritJUnit.assertThrows; import static java.nio.charset.StandardCharsets.UTF_8; import static org.eclipse.jgit.lib.Constants.OBJ_BLOB; +import com.github.rholder.retry.BlockStrategy; import com.github.rholder.retry.Retryer; import com.github.rholder.retry.RetryerBuilder; import com.github.rholder.retry.StopStrategies; +import com.google.common.collect.ImmutableList; +import com.google.common.truth.Expect; import com.google.common.util.concurrent.Runnables; import com.google.gerrit.exceptions.StorageException; import com.google.gerrit.reviewdb.client.Project; @@ -30,7 +33,9 @@ import com.google.gerrit.reviewdb.client.RefNames; import com.google.gerrit.server.extensions.events.GitReferenceUpdated; import com.google.gerrit.testing.InMemoryRepositoryManager; import java.io.IOException; -import java.util.concurrent.ExecutionException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import org.eclipse.jgit.errors.IncorrectObjectTypeException; @@ -41,11 +46,14 @@ import org.eclipse.jgit.lib.RefUpdate; import org.eclipse.jgit.lib.Repository; import org.eclipse.jgit.revwalk.RevWalk; import org.junit.Before; +import org.junit.Rule; import org.junit.Test; public class RepoSequenceTest { + @Rule public final Expect expect = Expect.create(); + // Don't sleep in tests. - private static final Retryer RETRYER = + private static final Retryer> RETRYER = RepoSequence.retryerBuilder().withBlockStrategy(t -> {}).build(); private InMemoryRepositoryManager repoManager; @@ -160,7 +168,7 @@ public class RepoSequenceTest { RepoSequence s = newSequence("id", 1, 10, bgUpdate, RETRYER); assertThat(doneBgUpdate.get()).isFalse(); assertThat(s.next()).isEqualTo(1234); - // Single acquire call that results in 2 ref reads. + // Two acquire calls, but only one successful. assertThat(s.acquireCount).isEqualTo(1); assertThat(doneBgUpdate.get()).isTrue(); } @@ -182,8 +190,7 @@ public class RepoSequenceTest { tr.branch(RefNames.REFS_SEQUENCES + "id").commit().create(); StorageException e = assertThrows(StorageException.class, () -> newSequence("id", 1, 3).next()); - assertThat(e.getCause()).isInstanceOf(ExecutionException.class); - assertThat(e.getCause().getCause()).isInstanceOf(IncorrectObjectTypeException.class); + assertThat(e.getCause()).isInstanceOf(IncorrectObjectTypeException.class); } } @@ -196,7 +203,7 @@ public class RepoSequenceTest { 1, 10, () -> writeBlob("id", Integer.toString(bgCounter.getAndAdd(1000))), - RetryerBuilder.newBuilder() + RetryerBuilder.>newBuilder() .withStopStrategy(StopStrategies.stopAfterAttempt(3)) .build()); StorageException thrown = assertThrows(StorageException.class, () -> s.next()); @@ -205,6 +212,77 @@ public class RepoSequenceTest { .contains("Failed to update refs/sequences/id: LOCK_FAILURE"); } + @Test + public void idCanBeRetrievedFromOtherThreadWhileWaitingToRetry() throws Exception { + // Seed existing ref value. + writeBlob("id", "1"); + + // Let the first update of the sequence fail with LOCK_FAILURE, so that the update is retried. + CountDownLatch lockFailure = new CountDownLatch(1); + CountDownLatch parallelSuccessfulSequenceGeneration = new CountDownLatch(1); + AtomicBoolean doneBgUpdate = new AtomicBoolean(false); + Runnable bgUpdate = + () -> { + if (!doneBgUpdate.getAndSet(true)) { + writeBlob("id", "1234"); + } + }; + + BlockStrategy blockStrategy = + t -> { + // Keep blocking until we verified that another thread can retrieve a sequence number + // while we are blocking here. + lockFailure.countDown(); + parallelSuccessfulSequenceGeneration.await(); + }; + + // Use batch size = 1 to make each call go to NoteDb. + RepoSequence s = + newSequence( + "id", + 1, + 1, + bgUpdate, + RepoSequence.retryerBuilder().withBlockStrategy(blockStrategy).build()); + + assertThat(doneBgUpdate.get()).isFalse(); + + // Start a thread to get a sequence number. This thread needs to update the sequence in NoteDb, + // but due to the background update (see bgUpdate) the first attempt to update NoteDb fails + // with LOCK_FAILURE. RepoSequence uses a retryer to retry the NoteDb update on LOCK_FAILURE, + // but our block strategy ensures that this retry only happens after isBlocking was set to + // false. + Future future = + Executors.newFixedThreadPool(1) + .submit( + () -> { + // The background update sets the next available sequence number to 1234. Then the + // test thread retrieves one sequence number, so that the next available sequence + // number for this thread is 1235. + expect.that(s.next()).isEqualTo(1235); + }); + + // Wait until the LOCK_FAILURE has happened and the block strategy was entered. + lockFailure.await(); + + // Verify that the background update was done now. + assertThat(doneBgUpdate.get()).isTrue(); + + // Verify that we can retrieve a sequence number while the other thread is blocked. If the + // s.next() call hangs it means that the RepoSequence.counterLock was not released before the + // background thread started to block for retry. In this case the test would time out. + assertThat(s.next()).isEqualTo(1234); + + // Stop blocking the retry of the background thread (and verify that it was still blocked). + parallelSuccessfulSequenceGeneration.countDown(); + + // Wait until the background thread is done. + future.get(); + + // Two successful acquire calls (because batch size == 1). + assertThat(s.acquireCount).isEqualTo(2); + } + @Test public void nextWithCountOneCaller() throws Exception { RepoSequence s = newSequence("id", 1, 3); @@ -260,7 +338,7 @@ public class RepoSequenceTest { final int start, int batchSize, Runnable afterReadRef, - Retryer retryer) { + Retryer> retryer) { return new RepoSequence( repoManager, GitReferenceUpdated.DISABLED,