Simplify and rework CubeCleanerController.trySaveSnapshotAndCleanupChunks
@@ -3,8 +3,6 @@ package io.datakernel.cube.service;
import io.datakernel.aggregation.AggregationChunk;
import io.datakernel.aggregation.LocalFsChunkStorage;
import io.datakernel.async.AsyncCallable;
-import io.datakernel.async.Callback;
-import io.datakernel.async.SettableStage;
import io.datakernel.async.Stage;
import io.datakernel.cube.ot.CubeDiff;
import io.datakernel.eventloop.Eventloop;
@@ -21,7 +19,6 @@ import java.util.stream.Stream;
import static io.datakernel.async.AsyncCallable.sharedCall;
import static io.datakernel.async.StageConsumer.transform;
-import static io.datakernel.async.Stages.collectToList;
import static io.datakernel.util.CollectionUtils.*;
import static io.datakernel.util.LogUtils.Level.TRACE;
import static io.datakernel.util.LogUtils.thisMethod;
@@ -138,41 +135,30 @@ public final class CubeCleanerController implements EventloopJmxMBeanEx {
Stage<Void> trySaveSnapshotAndCleanupChunks(Integer checkpointNode) {
return algorithms.checkout(checkpointNode)
- .thenCompose(checkpointDiffs -> {
- long cleanupTimestamp = eventloop.currentTimeMillis();
-
- return remote.saveSnapshot(checkpointNode, checkpointDiffs)
- .thenCompose($ -> {
- SettableStage<Optional<Integer>> result = SettableStage.create();
- findSnapshot(singleton(checkpointNode), extraSnapshotsCount, result);
- return result;
- })
+ .thenCompose(checkpointDiffs ->
+ remote.saveSnapshot(checkpointNode, checkpointDiffs)
+ .thenCompose($ -> findSnapshot(singleton(checkpointNode), extraSnapshotsCount))
.thenCompose(lastSnapshot -> lastSnapshot.isPresent() ?
- collectRequiredChunks(checkpointNode)
+ collectRequiredChunks(lastSnapshot.get())
.thenApply(extractedChunks -> union(chunks(checkpointDiffs), extractedChunks))
.thenCompose(chunks ->
- remote.getHeads()
- .thenCompose(heads -> collectToList(heads.stream().map(remote::loadCommit)))
- .thenApply(headCommits -> Math.min(cleanupTimestamp,
- headCommits.stream()
- .mapToLong(OTCommit::getTimestamp)
- .min()
- .getAsLong()) - chunksCleanupDelay.toMillis())
- .thenCompose(timestamp -> cleanup(lastSnapshot.get(), chunks, timestamp))
+ remote.loadCommit(lastSnapshot.get())
+ .thenApply(commit -> commit.getTimestamp() - chunksCleanupDelay.toMillis())
+ .thenCompose(timestamp ->
+ cleanup(lastSnapshot.get(), chunks, timestamp))
) :
- notEnoughSnapshots());
- })
+ notEnoughSnapshots()))
.whenComplete(toLogger(logger, thisMethod(), checkpointNode));
}
- void findSnapshot(Set<Integer> heads, int skipSnapshots, Callback<Optional<Integer>> cb) {
- algorithms.findParent(heads, DiffsReducer.toVoid(), OTCommit::isSnapshot, null)
- .whenResult(findResult -> {
- if (!findResult.isFound()) cb.set(Optional.empty());
- else if (skipSnapshots <= 0) cb.set(Optional.of(findResult.getCommit()));
- else findSnapshot(findResult.getCommitParents(), skipSnapshots - 1, cb);
- })
- .whenException(cb::setException);
+ Stage<Optional<Integer>> findSnapshot(Set<Integer> heads, int skipSnapshots) {
+ return algorithms.findParent(heads, DiffsReducer.toVoid(), OTCommit::isSnapshot, null)
+ .post()
+ .thenCompose(findResult -> {
+ if (!findResult.isFound()) return Stage.of(Optional.empty());
+ else if (skipSnapshots <= 0) return Stage.of(Optional.of(findResult.getCommit()));
+ else return findSnapshot(findResult.getCommitParents(), skipSnapshots - 1);
+ });
}
private Stage<Void> notEnoughSnapshots() {
