opendev/gerrit
Code Issues Proposed changes

BatchUpdate: Support executing multiple updates at once

Browse Source 
Execute all repo updates, then all DB updates, etc. Allow callers to
provide a listener instance with methods that get called between each
phase. This object doesn't receive any Context objects, because each
of those is tied to a single Op instance. Callers can keep track of
their own state as necessary.

Change-Id: Ia69a3e40cae2fd7bd51307e5e161755c19873ca0
This commit is contained in:
Dave Borowitz 2015-12-27 12:31:57 -08:00
parent f8e35efeed
commit b009b9d9e2
1 changed files with 119 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

159
gerrit-server/src/main/java/com/google/gerrit/server/git/BatchUpdate.java
View File

@ -20,6 +20,7 @@ import static com.google.common.base.Preconditions.checkState;
import com.google.common.base.Throwables;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ListMultimap;
import com.google.common.util.concurrent.CheckedFuture;
import com.google.gerrit.extensions.restapi.RestApiException;
@ -239,6 +240,112 @@ public class BatchUpdate implements AutoCloseable {
}
}
/**
* Interface for listening during batch update execution.
* <p>
* When used during execution of multiple batch updates, the {@code after*}
* methods are called after that phase has been completed for <em>all</em> updates.
*/
public static class Listener {
private static final Listener NONE = new Listener();
/**
* Called after updating all repositories and flushing objects but before
* updating any refs.
*/
public void afterUpdateRepos() throws Exception {
}
/** Called after updating all refs. */
public void afterRefUpdates() throws Exception {
}
/** Called after updating all changes. */
public void afterUpdateChanges() throws Exception {
}
}
private static Order getOrder(Collection<BatchUpdate> updates) {
Order o = null;
for (BatchUpdate u : updates) {
if (o == null) {
o = u.order;
} else if (u.order != o) {
throw new IllegalArgumentException("cannot mix execution orders");
}
}
return o;
}
static void execute(Collection<BatchUpdate> updates, Listener listener)
throws UpdateException, RestApiException {
if (updates.isEmpty()) {
return;
}
try {
Order order = getOrder(updates);
switch (order) {
case REPO_BEFORE_DB:
for (BatchUpdate u : updates) {
u.executeUpdateRepo();
}
listener.afterUpdateRepos();
for (BatchUpdate u : updates) {
u.executeRefUpdates();
}
listener.afterRefUpdates();
for (BatchUpdate u : updates) {
u.executeChangeOps();
}
listener.afterUpdateChanges();
break;
case DB_BEFORE_REPO:
for (BatchUpdate u : updates) {
u.executeChangeOps();
}
listener.afterUpdateChanges();
for (BatchUpdate u : updates) {
u.executeUpdateRepo();
}
listener.afterUpdateRepos();
for (BatchUpdate u : updates) {
u.executeRefUpdates();
}
listener.afterRefUpdates();
break;
default:
throw new IllegalStateException("invalid execution order: " + order);
}
List<CheckedFuture<?, IOException>> indexFutures = new ArrayList<>();
for (BatchUpdate u : updates) {
indexFutures.addAll(u.indexFutures);
}
ChangeIndexer.allAsList(indexFutures).get();
for (BatchUpdate u : updates) {
if (u.batchRefUpdate != null) {
// Fire ref update events only after all mutations are finished, since
// callers may assume a patch set ref being created means the change
// was created, or a branch advancing meaning some changes were
// closed.
u.gitRefUpdated.fire(u.project, u.batchRefUpdate);
}
}
for (BatchUpdate u : updates) {
u.executePostOps();
}
} catch (UpdateException | RestApiException e) {
// Propagate REST API exceptions thrown by operations; they commonly throw
// exceptions like ResourceConflictException to indicate an atomic update
// failure.
throw e;
} catch (Exception e) {
Throwables.propagateIfPossible(e);
throw new UpdateException(e);
}
}
private final ReviewDb db;
private final GitRepositoryManager repoManager;
@ -357,59 +464,35 @@ public class BatchUpdate implements AutoCloseable {
}
public void execute() throws UpdateException, RestApiException {
try {
switch (order) {
case REPO_BEFORE_DB:
executeRefUpdates();
executeChangeOps();
break;
case DB_BEFORE_REPO:
executeChangeOps();
executeRefUpdates();
break;
default:
throw new IllegalStateException("invalid execution order: " + order);
}
reindexChanges();
if (batchRefUpdate != null) {
// Fire ref update events only after all mutations are finished, since
// callers may assume a patch set ref being created means the change was
// created, or a branch advancing meaning some changes were closed.
gitRefUpdated.fire(project, batchRefUpdate);
}
executePostOps();
} catch (UpdateException | RestApiException e) {
// Propagate REST API exceptions thrown by operations; they commonly throw
// exceptions like ResourceConflictException to indicate an atomic update
// failure.
throw e;
} catch (Exception e) {
Throwables.propagateIfPossible(e);
throw new UpdateException(e);
}
execute(Listener.NONE);
}
private void executeRefUpdates()
throws IOException, UpdateException, RestApiException {
public void execute(Listener listener)
throws UpdateException, RestApiException {
execute(ImmutableList.of(this), listener);
}
private void executeUpdateRepo() throws UpdateException, RestApiException {
try {
RepoContext ctx = new RepoContext();
for (Op op : ops.values()) {
op.updateRepo(ctx);
}
if (inserter != null) {
inserter.flush();
}
} catch (Exception e) {
Throwables.propagateIfPossible(e, RestApiException.class);
throw new UpdateException(e);
}
}
private void executeRefUpdates() throws IOException, UpdateException {
if (commands.isEmpty()) {
return;
}
// May not be opened if the caller added ref updates but no new objects.
initRepository();
inserter.flush();
batchRefUpdate = repo.getRefDatabase().newBatchUpdate();
commands.addTo(batchRefUpdate);
batchRefUpdate.execute(revWalk, NullProgressMonitor.INSTANCE);
@ -465,10 +548,6 @@ public class BatchUpdate implements AutoCloseable {
changeControlFactory.controlFor(c, user));
}
private void reindexChanges() throws IOException {
ChangeIndexer.allAsList(indexFutures).checkedGet();
}
private void executePostOps() throws Exception {
Context ctx = new Context();
for (Op op : ops.values()) {