opendev/gerrit
Code Issues Proposed changes

Merge "RepoSequence: Release counter lock while blocking for retry"

Browse Source
This commit is contained in:
Patrick Hiesel
2019-06-13 08:17:42 +00:00
committed by Gerrit Code Review
parent 36a7cb6f92 577e8fc66c
commit ae29e38a3a
2 changed files with 164 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

140
java/com/google/gerrit/server/notedb/RepoSequence.java
View File

@@ -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<RefUpdate> retryerBuilder() {
return RetryerBuilder.<RefUpdate>newBuilder()
.retryIfResult(ru -> ru != null && RefUpdate.Result.LOCK_FAILURE.equals(ru.getResult()))
static RetryerBuilder<ImmutableList<Integer>> retryerBuilder() {
return RetryerBuilder.<ImmutableList<Integer>>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<RefUpdate> RETRYER = retryerBuilder().build();
private static final Retryer<ImmutableList<Integer>> 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<RefUpdate> retryer;
private final Retryer<ImmutableList<Integer>> retryer;
// Protects all non-final fields.
private final Lock counterLock;
@@ -147,7 +151,7 @@ public class RepoSequence {
Seed seed,
int batchSize,
Runnable afterReadRef,
Retryer<RefUpdate> retryer) {
Retryer<ImmutableList<Integer>> 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<RefUpdate> retryer,
Retryer<ImmutableList<Integer>> 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.
*
* <p>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.
*
* <p>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<Integer> next(int count) {
if (count == 0) {
return ImmutableList.of();
}
checkArgument(count > 0, "count is negative: %s", count);
counterLock.lock();
try {
List<Integer> 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<Integer> 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<RefUpdate> {
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.
*
* <p><strong>Note:</strong> 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<IntBlob> 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);
}
}

92
javatests/com/google/gerrit/server/notedb/RepoSequenceTest.java
View File

@@ -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<RefUpdate> RETRYER =
private static final Retryer<ImmutableList<Integer>> 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.<RefUpdate>newBuilder()
RetryerBuilder.<ImmutableList<Integer>>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<RefUpdate> retryer) {
Retryer<ImmutableList<Integer>> retryer) {
return new RepoSequence(
repoManager,
GitReferenceUpdated.DISABLED,