You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by be...@apache.org on 2025/07/23 09:47:03 UTC
(cassandra-accord) 01/01: Limit replay to those records not durably persisted to both command store and data store Also improve: - Limit truncation to TruncatedApplyWithOutcome until data is persisted durably to local store - start/stop progress log explicitly - limit progress log concurrency - clear heavy fields in some messages once processed Also fix: - relax paranoid CFK validation during restart
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-accord.git
commit 51f36f788b8095c0da9efaf1ea91bb9a7dd31181
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Thu Jul 17 14:40:10 2025 +0100
Limit replay to those records not durably persisted to both command store and data store
Also improve:
- Limit truncation to TruncatedApplyWithOutcome until data is persisted durably to local store
- start/stop progress log explicitly
- limit progress log concurrency
- clear heavy fields in some messages once processed
Also fix:
- relax paranoid CFK validation during restart
patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20780
---
.../src/main/java/accord/api/AsyncExecutor.java | 8 +-
.../src/main/java/accord/api/DataStore.java | 5 +-
accord-core/src/main/java/accord/api/Journal.java | 27 +--
.../src/main/java/accord/api/LocalListeners.java | 5 +
.../src/main/java/accord/api/ProgressLog.java | 4 +
.../main/java/accord/coordinate/Invalidate.java | 1 -
.../{AbstractLoader.java => AbstractReplayer.java} | 27 ++-
.../java/accord/impl/DefaultLocalListeners.java | 6 +
.../src/main/java/accord/impl/InMemoryAgent.java | 27 +++
.../java/accord/impl/InMemoryCommandStore.java | 210 +++++++++++-------
.../accord/impl/progresslog/CallbackInvoker.java | 7 +-
.../impl/progresslog/DefaultProgressLog.java | 241 +++++++++++++--------
.../impl/progresslog/DefaultProgressLogs.java | 42 +---
.../java/accord/impl/progresslog/HomeState.java | 6 +-
.../java/accord/impl/progresslog/WaitingState.java | 8 +-
.../src/main/java/accord/local/Cleanup.java | 4 +
.../src/main/java/accord/local/Command.java | 12 +
.../src/main/java/accord/local/CommandStore.java | 46 ++--
.../src/main/java/accord/local/Commands.java | 5 +-
.../src/main/java/accord/local/PreLoadContext.java | 1 -
.../main/java/accord/local/RedundantBefore.java | 30 ++-
.../main/java/accord/local/RedundantStatus.java | 32 ++-
.../src/main/java/accord/local/SafeCommand.java | 1 +
.../main/java/accord/local/SafeCommandStore.java | 5 +-
.../main/java/accord/local/cfk/CommandsForKey.java | 8 +-
.../src/main/java/accord/local/cfk/Serialize.java | 1 +
.../accord/local/durability/ShardDurability.java | 11 +-
.../src/main/java/accord/messages/Accept.java | 12 +-
.../src/main/java/accord/messages/Apply.java | 25 ++-
.../accord/messages/ApplyThenWaitUntilApplied.java | 18 +-
.../java/accord/messages/BeginInvalidation.java | 6 +-
.../java/accord/messages/ReadEphemeralTxnData.java | 2 +-
.../main/java/accord/messages/StableThenRead.java | 2 +-
.../main/java/accord/primitives/SaveStatus.java | 18 +-
.../src/main/java/accord/utils/ArrayBuffers.java | 9 +-
.../java/accord/utils/IntrusiveLinkedList.java | 3 +-
.../src/main/java/accord/utils/Invariants.java | 2 +-
.../java/accord/utils/ReducingIntervalMap.java | 10 +
.../main/java/accord/utils/async/AsyncResults.java | 32 +++
.../accord/utils/async/RunnableWithResult.java} | 46 ++--
.../src/test/java/accord/burn/BurnTestBase.java | 6 +-
.../coordinate/CoordinateTransactionTest.java | 8 -
.../test/java/accord/impl/RemoteListenersTest.java | 4 +-
.../src/test/java/accord/impl/basic/Cluster.java | 101 ++++-----
.../java/accord/impl/basic/InMemoryJournal.java | 5 +-
.../src/test/java/accord/impl/basic/NodeSink.java | 2 -
.../src/test/java/accord/impl/basic/Pending.java | 1 +
.../accord/impl/basic/TaskExecutorService.java | 4 +-
.../java/accord/impl/basic/TestProgressLogs.java | 69 ++++++
.../src/test/java/accord/impl/list/ListAgent.java | 34 ++-
.../accord/impl/list/ListFetchCoordinator.java | 2 +-
.../test/java/accord/impl/list/ListRequest.java | 15 +-
.../src/test/java/accord/impl/list/ListStore.java | 104 ++-------
.../test/java/accord/impl/list/Snapshotter.java | 135 ++++++++++++
.../src/test/java/accord/impl/mock/MockStore.java | 9 +-
.../java/accord/local/cfk/CommandsForKeyTest.java | 4 +-
.../main/java/accord/maelstrom/MaelstromStore.java | 13 +-
57 files changed, 937 insertions(+), 544 deletions(-)
diff --git a/accord-core/src/main/java/accord/api/AsyncExecutor.java b/accord-core/src/main/java/accord/api/AsyncExecutor.java
index 533eff7a..a4c8761d 100644
--- a/accord-core/src/main/java/accord/api/AsyncExecutor.java
+++ b/accord-core/src/main/java/accord/api/AsyncExecutor.java
@@ -20,20 +20,20 @@ package accord.api;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
-import java.util.concurrent.Executors;
import java.util.function.BiConsumer;
import accord.utils.async.AsyncChain;
+import accord.utils.async.RunnableWithResult;
public interface AsyncExecutor extends Executor
{
- default AsyncChain<?> build(Runnable task)
+ default AsyncChain<Void> build(Runnable task)
{
- return build(Executors.callable(task));
+ return build(new RunnableWithResult<>(task, null));
}
default <T> AsyncChain<T> build(Runnable task, T result)
{
- return build(Executors.callable(task, result));
+ return build(new RunnableWithResult<>(task, result));
}
<T> AsyncChain<T> build(Callable<T> task);
diff --git a/accord-core/src/main/java/accord/api/DataStore.java b/accord-core/src/main/java/accord/api/DataStore.java
index 55327cfd..750daa44 100644
--- a/accord-core/src/main/java/accord/api/DataStore.java
+++ b/accord-core/src/main/java/accord/api/DataStore.java
@@ -18,12 +18,13 @@
package accord.api;
+import accord.local.CommandStore;
import accord.local.Node;
+import accord.local.RedundantBefore;
import accord.local.SafeCommandStore;
import accord.primitives.Ranges;
import accord.primitives.SyncPoint;
import accord.primitives.Timestamp;
-import accord.primitives.TxnId;
import accord.utils.async.AsyncResult;
/**
@@ -124,5 +125,5 @@ public interface DataStore
* must be durable once the AsyncResult completes successfully. That is, a restart of the node must
* restore the DataStore to a state on or after the point at which snapshot was invoked.
*/
- AsyncResult<Void> snapshot(Ranges ranges, TxnId before);
+ void ensureDurable(CommandStore commandStore, Ranges ranges, RedundantBefore reportOnSuccess);
}
diff --git a/accord-core/src/main/java/accord/api/Journal.java b/accord-core/src/main/java/accord/api/Journal.java
index f491aafd..73dd5416 100644
--- a/accord-core/src/main/java/accord/api/Journal.java
+++ b/accord-core/src/main/java/accord/api/Journal.java
@@ -47,13 +47,20 @@ import org.agrona.collections.Int2ObjectHashMap;
*/
public interface Journal
{
+ enum Load
+ {
+ ALL,
+ PURGEABLE,
+ MINIMAL
+ }
+
void start(Node node);
Command loadCommand(int store, TxnId txnId, RedundantBefore redundantBefore, DurableBefore durableBefore);
default List<? extends Supplier<CommandChange.Builder>> debugCommand(int store, TxnId txnId) { throw new UnsupportedOperationException(); }
Command.Minimal loadMinimal(int store, TxnId txnId, Load load, RedundantBefore redundantBefore, DurableBefore durableBefore);
- // TODO (required): propagate exceptions (i.e. using OnDone instead of Runnable)
+ // TODO (required): propagate exceptions
void saveCommand(int store, CommandUpdate value, Runnable onFlush);
Iterator<? extends TopologyUpdate> replayTopologies();
@@ -163,25 +170,11 @@ public interface Journal
}
}
- enum Load
- {
- ALL,
- PURGEABLE,
- MINIMAL
- }
-
/**
* Helper for CommandStore to restore Command states.
*/
- interface Loader
- {
- AsyncChain<?> load(TxnId txnId);
- }
-
-
- interface OnDone
+ interface Replayer
{
- void success();
- void failure(Throwable t);
+ AsyncChain<?> replay(TxnId txnId);
}
}
diff --git a/accord-core/src/main/java/accord/api/LocalListeners.java b/accord-core/src/main/java/accord/api/LocalListeners.java
index fe2b9c52..173f365e 100644
--- a/accord-core/src/main/java/accord/api/LocalListeners.java
+++ b/accord-core/src/main/java/accord/api/LocalListeners.java
@@ -84,4 +84,9 @@ public interface LocalListeners
* Erase all listeners for transactions with a lower {@codfe TxnId} than {@code clearBefore}.
*/
void clearBefore(CommandStore safeStore, TxnId clearBefore);
+
+ /**
+ * Erase all listeners; used only for resetting state
+ */
+ void clear();
}
diff --git a/accord-core/src/main/java/accord/api/ProgressLog.java b/accord-core/src/main/java/accord/api/ProgressLog.java
index 5404da9d..978315ed 100644
--- a/accord-core/src/main/java/accord/api/ProgressLog.java
+++ b/accord-core/src/main/java/accord/api/ProgressLog.java
@@ -219,6 +219,8 @@ public interface ProgressLog
*/
void clear(TxnId txnId);
+ void start();
+ void stop();
void clear();
/**
@@ -240,6 +242,8 @@ public interface ProgressLog
@Override public void invalidIfUncommitted(TxnId txnId) {}
@Override public void clear(TxnId txnId) {}
@Override public void clearBefore(SafeCommandStore safeStore, TxnId clearWaitingBefore, TxnId clearAnyBefore) {}
+ @Override public void start() {}
+ @Override public void stop() {}
@Override public void clear() {}
}
}
diff --git a/accord-core/src/main/java/accord/coordinate/Invalidate.java b/accord-core/src/main/java/accord/coordinate/Invalidate.java
index 72886aa5..f57bebdb 100644
--- a/accord-core/src/main/java/accord/coordinate/Invalidate.java
+++ b/accord-core/src/main/java/accord/coordinate/Invalidate.java
@@ -20,7 +20,6 @@ package accord.coordinate;
import java.util.function.BiConsumer;
-import accord.api.TraceEventType;
import accord.coordinate.tracking.InvalidationTracker;
import accord.coordinate.tracking.InvalidationTracker.InvalidationShardTracker;
import accord.coordinate.tracking.RequestStatus;
diff --git a/accord-core/src/main/java/accord/impl/AbstractLoader.java b/accord-core/src/main/java/accord/impl/AbstractReplayer.java
similarity index 69%
rename from accord-core/src/main/java/accord/impl/AbstractLoader.java
rename to accord-core/src/main/java/accord/impl/AbstractReplayer.java
index a003bb2a..88281e8c 100644
--- a/accord-core/src/main/java/accord/impl/AbstractLoader.java
+++ b/accord-core/src/main/java/accord/impl/AbstractReplayer.java
@@ -23,21 +23,46 @@ import accord.local.Command;
import accord.local.CommandStore;
import accord.local.Commands;
import accord.local.PreLoadContext;
+import accord.local.RedundantBefore;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
+import accord.local.StoreParticipants;
import accord.primitives.Participants;
import accord.primitives.SaveStatus;
import accord.primitives.TxnId;
import accord.utils.Invariants;
+import static accord.local.RedundantStatus.Property.LOCALLY_DURABLE_TO_COMMAND_STORE;
+import static accord.local.RedundantStatus.Property.LOCALLY_DURABLE_TO_DATA_STORE;
import static accord.primitives.SaveStatus.Applying;
import static accord.primitives.SaveStatus.PreApplied;
import static accord.primitives.SaveStatus.TruncatedApplyWithOutcome;
import static accord.primitives.Status.Applied;
import static accord.primitives.Txn.Kind.Write;
-public abstract class AbstractLoader implements Journal.Loader
+public abstract class AbstractReplayer implements Journal.Replayer
{
+ final RedundantBefore redundantBefore;
+ final TxnId minReplay;
+
+ protected AbstractReplayer(RedundantBefore redundantBefore)
+ {
+ this.redundantBefore = redundantBefore;
+ this.minReplay = TxnId.noneIfNull(redundantBefore.foldl((b, v) -> TxnId.nonNullOrMin(v, TxnId.min(b.maxBound(LOCALLY_DURABLE_TO_DATA_STORE), b.maxBound(LOCALLY_DURABLE_TO_COMMAND_STORE))), null, ignore -> false));
+ }
+
+ protected boolean maybeShouldReplay(TxnId txnId)
+ {
+ return txnId.compareTo(minReplay) >= 0;
+ }
+
+ protected boolean shouldReplay(TxnId txnId, StoreParticipants participants)
+ {
+ Participants<?> search = participants.route();
+ if (search == null) search = participants.hasTouched();
+ return redundantBefore.foldlWithDefault(search, (b, v, id) -> v || b.maxBoundBoth(LOCALLY_DURABLE_TO_COMMAND_STORE, LOCALLY_DURABLE_TO_DATA_STORE).compareTo(id) <= 0, RedundantBefore.Bounds.NONE, false, txnId, i -> i);
+ }
+
protected void initialiseState(SafeCommandStore safeStore, TxnId txnId)
{
SafeCommand safeCommand = safeStore.unsafeGet(txnId);
diff --git a/accord-core/src/main/java/accord/impl/DefaultLocalListeners.java b/accord-core/src/main/java/accord/impl/DefaultLocalListeners.java
index a4c5d60f..cacc703c 100644
--- a/accord-core/src/main/java/accord/impl/DefaultLocalListeners.java
+++ b/accord-core/src/main/java/accord/impl/DefaultLocalListeners.java
@@ -516,4 +516,10 @@ public class DefaultLocalListeners implements LocalListeners
txnListeners = BTreeRemoval.remove(txnListeners, TxnListeners::compareListeners, entry);
}
}
+
+ public void clear()
+ {
+ txnListeners = BTree.empty();
+ complexListeners.clear();
+ }
}
diff --git a/accord-core/src/main/java/accord/impl/InMemoryAgent.java b/accord-core/src/main/java/accord/impl/InMemoryAgent.java
new file mode 100644
index 00000000..5571b236
--- /dev/null
+++ b/accord-core/src/main/java/accord/impl/InMemoryAgent.java
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package accord.impl;
+
+import accord.api.Agent;
+import accord.utils.async.AsyncResult;
+
+public interface InMemoryAgent extends Agent
+{
+ AsyncResult<Void> snapshot(InMemoryCommandStore commandStore);
+}
diff --git a/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java b/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
index 2a754a67..33ed1009 100644
--- a/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
+++ b/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
@@ -18,6 +18,8 @@
package accord.impl;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -32,12 +34,12 @@ import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
import javax.annotation.Nonnull;
-import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
@@ -56,15 +58,17 @@ import accord.local.CommandStore;
import accord.local.CommandStores.RangesForEpoch;
import accord.local.CommandSummaries;
import accord.local.Commands;
-import accord.local.LoadKeys;
import accord.local.NodeCommandStoreService;
import accord.local.PreLoadContext;
+import accord.local.PreLoadContext.Empty;
+import accord.local.RedundantBefore;
import accord.local.RejectBefore;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.local.StoreParticipants;
import accord.local.cfk.CommandsForKey;
import accord.local.cfk.SafeCommandsForKey;
+import accord.local.cfk.Serialize;
import accord.primitives.AbstractRanges;
import accord.primitives.AbstractUnseekableKeys;
import accord.primitives.PartialDeps;
@@ -84,13 +88,13 @@ import accord.primitives.Unseekables;
import accord.utils.Invariants;
import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncChains;
+import accord.utils.async.AsyncResult;
+import accord.utils.async.AsyncResults;
import accord.utils.async.Cancellable;
import org.agrona.collections.ObjectHashSet;
import static accord.local.Cleanup.Input.FULL;
-import static accord.local.LoadKeys.ASYNC;
import static accord.local.LoadKeys.NONE;
-import static accord.local.LoadKeysFor.READ_WRITE;
import static accord.local.LoadKeysFor.WRITE;
import static accord.local.RedundantStatus.Coverage.ALL;
import static accord.local.StoreParticipants.Filter.LOAD;
@@ -119,6 +123,62 @@ public abstract class InMemoryCommandStore extends CommandStore
private static final Logger logger = LoggerFactory.getLogger(InMemoryCommandStore.class);
private static final boolean CHECK_DEPENDENCY_INVARIANTS = false;
+ public static class Snapshot extends AsyncResults.SettableResult<Snapshot>
+ {
+ private static final AtomicLong nextId = new AtomicLong();
+
+ private final long id = nextId.incrementAndGet();
+ private final NavigableMap<RoutingKey, ByteBuffer> commandsForKey = new TreeMap<>();
+ private final TreeMap<TxnId, Ranges> rangeCommands = new TreeMap<>();
+ private int waitingForCfk;
+
+ private Snapshot(){}
+
+ public void restore(InMemoryCommandStore commandStore)
+ {
+ for (Map.Entry<RoutingKey, ByteBuffer> e : commandsForKey.entrySet())
+ commandStore.commandsForKey.computeIfAbsent(e.getKey(), GlobalCommandsForKey::new).value(Serialize.fromBytes(e.getKey(), e.getValue()));
+
+ for (Map.Entry<TxnId, Ranges> e : rangeCommands.entrySet())
+ {
+ RangeCommand rangeCommand = commandStore.rangeCommands.computeIfAbsent(e.getKey(), RangeCommand::new);
+ rangeCommand.ranges = e.getValue();
+ }
+ }
+
+ void save(CommandsForKey cfk)
+ {
+ commandsForKey.put(cfk.key(), Serialize.toBytesWithoutKey(cfk.maximalPrune()));
+ if (--waitingForCfk == 0)
+ trySuccess(this);
+ }
+
+ public static AsyncResult<Snapshot> snapshot(InMemoryCommandStore commandStore)
+ {
+ Snapshot snapshot = new Snapshot();
+ for (Map.Entry<TxnId, RangeCommand> e : commandStore.rangeCommands.entrySet())
+ snapshot.rangeCommands.put(e.getKey(), e.getValue().ranges);
+
+ for (Map.Entry<RoutableKey, GlobalCommandsForKey> e : commandStore.commandsForKey.entrySet())
+ {
+ GlobalCommandsForKey global = e.getValue();
+ CommandsForKey cfk = global.value();
+ if (cfk.isLoadingPruned())
+ {
+ ++snapshot.waitingForCfk;
+ global.pendingSnapshots.add(snapshot);
+ }
+ else
+ {
+ snapshot.commandsForKey.put(cfk.key(), Serialize.toBytesWithoutKey(cfk.maximalPrune()));
+ }
+ }
+ if (snapshot.waitingForCfk == 0)
+ snapshot.trySuccess(snapshot);
+ return snapshot;
+ }
+ }
+
final NavigableMap<TxnId, GlobalCommand> commands = new TreeMap<>();
final NavigableMap<Timestamp, GlobalCommand> commandsByExecuteAt = new TreeMap<>();
private final NavigableMap<RoutableKey, GlobalCommandsForKey> commandsForKey = new TreeMap<>();
@@ -263,19 +323,31 @@ public abstract class InMemoryCommandStore extends CommandStore
}
@Override
- protected void updatedRedundantBefore(SafeCommandStore safeStore, TxnId syncId, Ranges ranges)
+ protected void ensureDurable(Ranges ranges, RedundantBefore onCommandStoreDurable)
+ {
+ ((InMemoryAgent)agent).snapshot(this).begin((success, fail) -> execute((Empty)() -> "Report CommandStore Durable", safeStore -> {
+ safeStore.upsertRedundantBefore(onCommandStoreDurable);
+ }).begin(agent));
+ }
+
+ @Override
+ protected void updatedRedundantBefore(SafeCommandStore safeStore, RedundantBefore added)
{
InMemorySafeStore inMemorySafeStore = (InMemorySafeStore) safeStore;
- ranges.forEach(r -> {
- commandsForKey.subMap(r.start(), r.startInclusive(), r.end(), r.endInclusive()).forEach((forKey, forValue) -> {
- if (!forValue.isEmpty())
- {
- InMemorySafeCommandsForKey safeCfk = forValue.createSafeReference();
- inMemorySafeStore.commandsForKey.put(forKey, safeCfk);
- safeCfk.refresh(safeStore);
- }
- });
- });
+ for (int i = 0 ; i < added.size() ; ++i)
+ {
+ if (added.valueAt(i) != null)
+ {
+ commandsForKey.subMap(added.startAt(i), !added.inclusiveEnds(), added.startAt(i + 1), added.inclusiveEnds()).forEach((forKey, forValue) -> {
+ if (!forValue.isEmpty())
+ {
+ InMemorySafeCommandsForKey safeCfk = forValue.createSafeReference();
+ inMemorySafeStore.commandsForKey.put(forKey, safeCfk);
+ safeCfk.refresh(safeStore);
+ }
+ });
+ }
+ }
TxnId clearProgressLogBefore = unsafeGetRedundantBefore().minShardAndLocallyAppliedBefore();
List<TxnId> clearing = ((DefaultProgressLog) progressLog).activeBefore(clearProgressLogBefore);
for (TxnId txnId : clearing)
@@ -291,7 +363,7 @@ public abstract class InMemoryCommandStore extends CommandStore
((command.participants().stillExecutes() != null && command.participants().stillExecutes().isEmpty())
|| !Route.isFullRoute(command.route()))));
}
- super.updatedRedundantBefore(safeStore, syncId, ranges);
+ super.updatedRedundantBefore(safeStore, added);
}
@Override
@@ -314,7 +386,7 @@ public abstract class InMemoryCommandStore extends CommandStore
}
else
{
- GlobalCommand global = tx.getValue().command;
+ GlobalCommand global = commands.get(tx.getKey());
if (global != null)
maxRedundant = Timestamp.nonNullOrMax(maxRedundant, global.value().executeAt());
return true;
@@ -432,15 +504,8 @@ public abstract class InMemoryCommandStore extends CommandStore
static class RangeCommand
{
final TxnId txnId;
- @Nullable GlobalCommand command;
Ranges ranges;
- RangeCommand(GlobalCommand command)
- {
- this.txnId = command.txnId;
- this.command = command;
- }
-
RangeCommand(TxnId txnId)
{
this.txnId = txnId;
@@ -456,11 +521,6 @@ public abstract class InMemoryCommandStore extends CommandStore
if (ranges == null) ranges = add;
else ranges = ranges.with(add);
}
-
- void update(GlobalCommand set)
- {
- command = set;
- }
}
public static abstract class GlobalState<V>
@@ -513,12 +573,21 @@ public abstract class InMemoryCommandStore extends CommandStore
public static class GlobalCommandsForKey extends GlobalState<CommandsForKey>
{
private final RoutingKey key;
+ private final List<Snapshot> pendingSnapshots = new ArrayList<>();
public GlobalCommandsForKey(RoutableKey key)
{
this.key = (RoutingKey) key;
}
+ @Override
+ public GlobalState<CommandsForKey> value(CommandsForKey value)
+ {
+ if (!pendingSnapshots.isEmpty() && !value.isLoadingPruned())
+ pendingSnapshots.forEach(snapshot -> snapshot.save(value));
+ return super.value(value);
+ }
+
public InMemorySafeCommandsForKey createSafeReference()
{
return new InMemorySafeCommandsForKey(key, this);
@@ -676,7 +745,7 @@ public abstract class InMemoryCommandStore extends CommandStore
if (updated.saveStatus() == Erased || updated.saveStatus() == Vestigial)
return;
- commandStore().rangeCommands.computeIfAbsent(txnId, ignore -> new RangeCommand(commandStore().commands.get(txnId)))
+ commandStore().rangeCommands.computeIfAbsent(txnId, RangeCommand::new)
.update(((AbstractRanges)updated.participants().stillTouches()).toRanges());
}
@@ -740,7 +809,7 @@ public abstract class InMemoryCommandStore extends CommandStore
TreeMap<Timestamp, Summary> summaries = new TreeMap<>();
for (RangeCommand rangeCommand : commandStore().rangeCommands.values())
{
- GlobalCommand global = rangeCommand.command;
+ GlobalCommand global = commandStore().commands.get(rangeCommand.txnId);
Command command = global == null ? null : global.value();
Summary summary;
if (command == null)
@@ -1128,7 +1197,7 @@ public abstract class InMemoryCommandStore extends CommandStore
// redundantBefore, durableBefore, newBootstrapBeganAt, safeToRead, rangesForEpoch are
// not replayed here. It is assumed that persistence on the application side will ensure
// they are brought up to latest values _before_ replay.
- public void clearForTesting()
+ public void clear()
{
Invariants.require(current == null);
progressLog.clear();
@@ -1136,80 +1205,67 @@ public abstract class InMemoryCommandStore extends CommandStore
commandsByExecuteAt.clear();
commandsForKey.clear();
rangeCommands.clear();
+ listeners.clear();
unsafeSetRejectBefore(new RejectBefore());
}
- public Journal.Loader loader()
+ public Journal.Replayer replayer()
{
- return new CommandLoader(this);
+ return new CommandReplayer(this);
}
- private static class CommandLoader extends AbstractLoader
+ private static class CommandReplayer extends AbstractReplayer
{
private final InMemoryCommandStore commandStore;
- private CommandLoader(InMemoryCommandStore commandStore)
+ private CommandReplayer(InMemoryCommandStore commandStore)
{
+ super(commandStore.unsafeGetRedundantBefore());
this.commandStore = commandStore;
}
- private PreLoadContext context(Command command, LoadKeys loadKeys)
- {
- TxnId txnId = command.txnId();
- AbstractUnseekableKeys keys = null;
-
- if (CommandsForKey.manages(txnId))
- keys = (AbstractUnseekableKeys) command.participants().hasTouched();
- else if (!CommandsForKey.managesExecution(txnId) && command.hasBeen(Status.Stable) && !command.hasBeen(Status.Truncated))
- keys = command.asCommitted().waitingOn.keys;
-
- if (keys != null)
- return PreLoadContext.contextFor(txnId, keys, loadKeys, WRITE, "Replay");
-
- return PreLoadContext.contextFor(txnId, "Replay");
- }
-
- protected TxnId loadInternal(Command command, SafeCommandStore safeStore)
- {
- TxnId txnId = command.txnId();
- Cleanup cleanup = Cleanup.shouldCleanup(FULL, safeStore, command, command.participants());
- if (cleanup != Cleanup.NO)
- command = Commands.purge(safeStore, command, cleanup);
-
- safeStore.unsafeGetNoCleanup(txnId).update(safeStore, command);
- return txnId;
- }
-
- private AsyncChain<TxnId> load(Command command)
+ private AsyncChain<Void> apply(Command command)
{
return AsyncChains.success(commandStore.executeInContext(commandStore,
- context(command, ASYNC),
- (SafeCommandStore safeStore) -> loadInternal(command, safeStore)));
- }
-
- private AsyncChain<Void> apply(TxnId txnId)
- {
- return AsyncChains.success(commandStore.executeInContext(commandStore,
- PreLoadContext.contextFor(txnId, "Replay"),
+ PreLoadContext.contextFor(command.txnId(), "Replay"),
(SafeCommandStore safeStore) -> {
- initialiseState(safeStore, txnId);
+ initialiseState(safeStore, command.txnId());
return null;
}));
}
@Override
- public AsyncChain<Void> load(TxnId txnId)
+ public AsyncChain<Void> replay(TxnId txnId)
{
// TODO (required): consider this race condition some more:
// - can we avoid double-applying?
// - is this definitely safe?
+ Command command = null;
if (commandStore.hasCommand(txnId))
- return apply(txnId);
+ command = commandStore.commands.get(txnId).value();
- Command command = commandStore.journal.loadCommand(commandStore.id, txnId, commandStore.unsafeGetRedundantBefore(), commandStore.durableBefore());
if (command == null)
+ {
+ command = commandStore.journal.loadCommand(commandStore.id, txnId, commandStore.unsafeGetRedundantBefore(), commandStore.durableBefore());
+ if (command != null)
+ {
+ Cleanup cleanup = Cleanup.shouldCleanup(FULL, command, commandStore.unsafeGetRedundantBefore(), commandStore.durableBefore());
+ if (cleanup != Cleanup.NO)
+ command = Commands.purgeUnsafe(commandStore, command, cleanup);
+
+ // initialise basic state, but don't call safeStore.update so we don't initialise listeners etc
+ GlobalCommand global = commandStore.commands.computeIfAbsent(txnId, GlobalCommand::new);
+ global.value(command);
+ Timestamp executeAt = command.executeAtIfKnown();
+ if (executeAt != null)
+ commandStore.commandsByExecuteAt.put(executeAt, global);
+ }
+ }
+
+ if (command == null || !maybeShouldReplay(txnId) || !shouldReplay(txnId, command.participants()))
return AsyncChains.success(null);
- return load(command).flatMap(this::apply);
+
+ return apply(command);
}
}
diff --git a/accord-core/src/main/java/accord/impl/progresslog/CallbackInvoker.java b/accord-core/src/main/java/accord/impl/progresslog/CallbackInvoker.java
index 895a23d5..4667846b 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/CallbackInvoker.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/CallbackInvoker.java
@@ -44,7 +44,7 @@ class CallbackInvoker<P, V> implements BiConsumer<V, Throwable>, PreLoadContext
static <P, V> CallbackInvoker<P, V> invokeCallback(TxnStateKind kind, DefaultProgressLog owner, TxnId txnId, P param, Callback<P, V> callback)
{
CallbackInvoker<P, V> invoker = new CallbackInvoker<>(owner, kind, owner.nextInvokerId(), txnId, param, callback);
- owner.registerActive(kind, txnId, invoker);
+ owner.registerPending(kind, txnId, invoker);
return invoker;
}
@@ -77,14 +77,13 @@ class CallbackInvoker<P, V> implements BiConsumer<V, Throwable>, PreLoadContext
// we load safeCommand first so that if it clears the progress log we abandon the callback
SafeCommand safeCommand = safeStore.ifInitialised(txnId);
- if (!owner.deregisterActive(kind(), this))
+ if (!owner.complete(safeStore, kind(), id, this))
return;
if (safeCommand == null)
return;
- callback.callback(safeStore, safeCommand, owner, txnId, param, success, fail);
- owner.undebugActive(this);
+ callback.callback(safeStore, safeCommand, owner, txnId, param, success, fail);
}, owner.commandStore.agent());
}
diff --git a/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java b/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java
index 8c3346f2..48a9a3d6 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java
@@ -28,9 +28,6 @@ import java.util.function.Consumer;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import accord.api.ProgressLog;
import accord.api.RoutingKey;
import accord.local.Command;
@@ -46,6 +43,7 @@ import accord.primitives.ProgressToken;
import accord.primitives.Ranges;
import accord.primitives.Route;
import accord.primitives.TxnId;
+import accord.utils.ArrayBuffers;
import accord.utils.ArrayBuffers.BufferList;
import accord.utils.Invariants;
import accord.utils.LogGroupTimers;
@@ -76,10 +74,10 @@ import static accord.utils.btree.UpdateFunction.noOpReplace;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
// TODO (expected): for transactions that span multiple progress logs (notably: sync points) we need to coordinate *fetching* to avoid redundant work
+// TODO (expected): report transactions not making progress
+// TODO (desired): evict to disk
public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStore>
{
- private static final Logger logger = LoggerFactory.getLogger(DefaultProgressLog.class);
-
final Node node;
final CommandStore commandStore;
@@ -95,10 +93,10 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
* These callbacks are required to have hashCode() == txnId.hashCode() and equals(txnId) == true,
* so that we can manage overriding callbacks on the relevant TxnState.
*/
- private final ObjectHashSet<Object> activeWaiting = new ObjectHashSet<>();
- private final ObjectHashSet<Object> activeHome = new ObjectHashSet<>();
+ private final ObjectHashSet<Object> pendingWaiting = new ObjectHashSet<>();
+ private final ObjectHashSet<Object> pendingHome = new ObjectHashSet<>();
- private final Long2ObjectHashMap<Object> debugActive = Invariants.debug() ? new Long2ObjectHashMap<>() : null;
+ private final Long2ObjectHashMap<Object> active = new Long2ObjectHashMap<>();
private final Map<TxnId, StackTraceElement[]> debugDeleted = Invariants.debug() ? new Object2ObjectHashMap<>() : null;
private static final Object[] EMPTY_RUN_BUFFER = new Object[0];
@@ -106,15 +104,19 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
// The tasks whose timers have elapsed and are going to be run
// The queue is drained here first before processing tasks so that tasks can modify the queue
- private Object[] runBuffer;
- private int runBufferCount;
+ private Object[] runBuffer = EMPTY_RUN_BUFFER;
+ private int runBufferIndex, runBufferCount;
private RunInvoker[] awaitingEpochBuffer = EMPTY_AWAITING_EPOCH_BUFFER;
private int awaitingEpochBufferCount;
private boolean isAwaitingEpoch;
+ private boolean processing;
+
+ private volatile boolean stopped;
+ private int maxConcurrency = 128;
private long nextInvokerId;
- DefaultProgressLog(Node node, CommandStore commandStore)
+ protected DefaultProgressLog(Node node, CommandStore commandStore)
{
this.node = node;
this.commandStore = commandStore;
@@ -351,6 +353,22 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
}
}
+ @Override
+ public void start()
+ {
+ commandStore.maybeExecuteImmediately(() -> {
+ stopped = false;
+ accept(null);
+ });
+ }
+
+ @Override
+ public void stop()
+ {
+ stopped = true;
+ }
+
+ @Override
public void clear()
{
timers.clear();
@@ -358,13 +376,13 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
stateMap = BTree.empty();
progressTokenMap = BTree.empty();
- activeWaiting.clear();
- activeHome.clear();
+ pendingWaiting.clear();
+ pendingHome.clear();
if (debugDeleted != null)
debugDeleted.clear();
runBuffer = EMPTY_RUN_BUFFER;
- runBufferCount = 0;
+ runBufferIndex = runBufferCount = 0;
}
private void clear(TxnState state)
@@ -466,26 +484,22 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
@Override
public void accept(@Nullable SafeCommandStore safeStore)
{
+ if (stopped || processing)
+ return;
+
long nowMicros = node.elapsed(TimeUnit.MICROSECONDS);
+ processing = true;
try
{
- if (DefaultProgressLogs.pauseForTest(this))
- {
- logger.info("Skipping progress log because it is paused for test");
- return;
- }
-
- try (BufferList<RunInvoker> readyToRun = safeStore == null ? null : new BufferList<>())
+ processAwaitingEpoch();
+ try (BufferList<TxnState> preRunBuffer = new BufferList<>())
{
- processAwaitingEpoch(safeStore, readyToRun);
// drain to a buffer to avoid reentrancy in timers
- runBuffer = EMPTY_RUN_BUFFER;
- runBufferCount = 0;
- timers.advance(nowMicros, this, DefaultProgressLog::addToRunBuffer);
- processRunBuffer(safeStore, nowMicros, readyToRun);
- cachedAny().forceDiscard(runBuffer, runBufferCount);
- processReadyToRun(safeStore, readyToRun);
+ timers.advance(nowMicros, preRunBuffer, BufferList::add);
+ updateRunBuffer(nowMicros, preRunBuffer);
}
+ processRunBuffer(safeStore);
+
if (awaitingEpochBufferCount > 0)
rerunWithPendingEpoch();
}
@@ -493,13 +507,44 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
{
node.agent().onUncaughtException(t);
}
+ finally
+ {
+ processing = false;
+ }
}
- private void addToRunBuffer(TxnState add)
+ private void addToRunBuffer(RunInvoker readyToRun)
{
if (runBufferCount == runBuffer.length)
- runBuffer = cachedAny().resize(runBuffer, runBufferCount, Math.max(8, runBuffer.length * 2));
- runBuffer[runBufferCount++] = add;
+ {
+ int newCount = runBufferCount - runBufferIndex;
+ Object[] newBuffer = cachedAny().get(Math.max(8, newCount * 2));
+ replaceRunBuffer(newBuffer);
+ }
+ runBuffer[runBufferCount++] = readyToRun;
+ }
+
+ private void replaceRunBuffer(Object[] newBuffer)
+ {
+ Object[] prevBuffer = runBuffer;
+ int prevCount = runBufferCount;
+ int newCount = prevCount - runBufferIndex;
+ System.arraycopy(prevBuffer, runBufferIndex, newBuffer, 0, newCount);
+ runBuffer = newBuffer;
+ runBufferIndex = 0;
+ runBufferCount = newCount;
+ if (prevBuffer.length >= ArrayBuffers.MIN_BUFFER_SIZE)
+ cachedAny().forceDiscard(prevBuffer, prevCount);
+ }
+
+ private void maybeShrinkRunBuffer()
+ {
+ if (runBuffer.length <= (runBufferCount - runBufferIndex)/2)
+ {
+ int newCount = runBufferCount - runBufferIndex;
+ Object[] newBuffer = new Object[newCount + (newCount/2)];
+ replaceRunBuffer(newBuffer);
+ }
}
private void rerunWithPendingEpoch()
@@ -518,13 +563,12 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
}, node.agent()));
}
- private void processRunBuffer(@Nullable SafeCommandStore safeStore, long nowMicros, List<RunInvoker> readyToRun)
+ private void updateRunBuffer(long nowMicros, List<TxnState> preRunBuffer)
{
long hasEpoch = node.topology().epoch();
- for (int i = 0; i < runBufferCount; ++i)
+ for (TxnState run : preRunBuffer)
{
- TxnState run = (TxnState) runBuffer[i];
Invariants.require(!run.isScheduled());
TxnStateKind runKind = run.wasScheduledTimer();
validatePreRunState(run, runKind);
@@ -547,8 +591,8 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
long epoch = run.txnId.epoch();
if (epoch <= hasEpoch)
{
- addIfReadyElseSubmit(safeStore, invoker(run, runKind), readyToRun);
- if (invokeBoth) addIfReadyElseSubmit(safeStore, invoker(run, runKind.other()), readyToRun);
+ addToRunBuffer(invoker(run, runKind));
+ if (invokeBoth) addToRunBuffer(invoker(run, runKind.other()));
}
else
{
@@ -559,12 +603,9 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
if (invokeBoth) awaitingEpochBuffer[awaitingEpochBufferCount++] = invoker(run, runKind.other());
}
}
-
- Arrays.fill(runBuffer, 0, runBufferCount, null);
- runBufferCount = 0;
}
- private void processAwaitingEpoch(@Nullable SafeCommandStore safeStore, List<RunInvoker> readyToRun)
+ private void processAwaitingEpoch()
{
if (awaitingEpochBufferCount == 0)
return;
@@ -574,7 +615,7 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
for (int i = 0 ; i < awaitingEpochBufferCount ; ++i)
{
RunInvoker awaiting = awaitingEpochBuffer[i];
- if (awaiting.run.txnId.epoch() <= hasEpoch) addIfReadyElseSubmit(safeStore, awaiting, readyToRun);
+ if (awaiting.run.txnId.epoch() <= hasEpoch) addToRunBuffer(awaiting);
else awaitingEpochBuffer[retainCount++] = awaiting;
}
awaitingEpochBufferCount = retainCount;
@@ -582,24 +623,33 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
else if (retainCount < awaitingEpochBuffer.length / 2) awaitingEpochBuffer = Arrays.copyOf(awaitingEpochBuffer, retainCount);
}
- private void processReadyToRun(SafeCommandStore safeStore, List<RunInvoker> readyToRun)
+ private void processRunBuffer(SafeCommandStore safeStore)
{
- if (safeStore == null || readyToRun.isEmpty())
- return;
-
- int i = 0;
- try
- {
- while (i < readyToRun.size())
- readyToRun.get(i++).accept(safeStore);
- }
- finally
+ while (runBufferIndex < runBufferCount)
{
- while (i < readyToRun.size())
+ if (active.size() >= maxConcurrency)
+ {
+ maybeShrinkRunBuffer();
+ return;
+ }
+
+ RunInvoker run = (RunInvoker) runBuffer[runBufferIndex];
+ if (safeStore == null || !safeStore.canExecuteWith(run))
{
- commandStore.execute((PreLoadContext.Empty) () -> "Run ProgressLog", readyToRun.get(i++), node.agent());
+ maybeShrinkRunBuffer();
+ commandStore.execute(run, this);
+ return;
}
+
+ ++runBufferIndex;
+ try { run.accept(safeStore); }
+ catch (Throwable t) { node.agent().onUncaughtException(t); }
}
+
+ if (runBuffer.length > ArrayBuffers.MIN_BUFFER_SIZE)
+ cachedAny().forceDiscard(runBuffer, runBufferCount);
+ runBufferIndex = runBufferCount = 0;
+ runBuffer = EMPTY_RUN_BUFFER;
}
private void validatePreRunState(TxnState run, TxnStateKind kind)
@@ -608,27 +658,14 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
Invariants.require(progress != NoneExpected && progress != Querying);
}
- void addIfReadyElseSubmit(@Nullable SafeCommandStore safeStore, RunInvoker invoker, List<RunInvoker> ifReady)
- {
- if (safeStore != null && safeStore.canExecuteWith(invoker.run))
- {
- ifReady.add(invoker);
- }
- else
- {
- // TODO (required): if we fail execution, do we need to reschedule the state?
- commandStore.execute(invoker.run, invoker, commandStore.agent());
- }
- }
-
RunInvoker invoker(TxnState run, TxnStateKind runKind)
{
RunInvoker invoker = new RunInvoker(nextInvokerId(), run, runKind);
- registerActive(runKind, run.txnId, invoker);
+ registerPending(runKind, run.txnId, invoker);
return invoker;
}
- class RunInvoker implements Consumer<SafeCommandStore>
+ class RunInvoker implements Consumer<SafeCommandStore>, PreLoadContext
{
final long id;
final TxnState run;
@@ -644,7 +681,7 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
@Override
public void accept(SafeCommandStore safeStore)
{
- if (!deregisterActive(runKind, this))
+ if (!complete(safeStore, runKind, id, this))
return; // we've been cancelled
// we have to read safeCommand first as it may become truncated on load, which may clear the progress log and invalidate us
@@ -690,6 +727,18 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
{
return run.txnId.hashCode();
}
+
+ @Override
+ public TxnId primaryTxnId()
+ {
+ return run.txnId;
+ }
+
+ @Override
+ public String reason()
+ {
+ return "Invoke " + runKind + " Progress Log";
+ }
}
long nextInvokerId()
@@ -697,43 +746,40 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
return nextInvokerId++;
}
- ObjectHashSet<Object> active(TxnStateKind kind)
- {
- return kind == Waiting ? activeWaiting : activeHome;
- }
-
- void registerActive(TxnStateKind kind, TxnId txnId, Object object)
+ ObjectHashSet<Object> pending(TxnStateKind kind)
{
- ObjectHashSet<Object> active = active(kind);
- Invariants.require(!active.contains(txnId));
- active.add(object);
+ return kind == Waiting ? pendingWaiting : pendingHome;
}
- boolean hasActive(TxnStateKind kind, TxnId txnId)
+ void registerPending(TxnStateKind kind, TxnId txnId, Object object)
{
- return active(kind).contains(txnId);
+ ObjectHashSet<Object> pending = pending(kind);
+ Invariants.require(!pending.contains(txnId));
+ pending.add(object);
}
- void debugActive(Object debug, CallbackInvoker<?, ?> invoker)
+ boolean hasPending(TxnStateKind kind, TxnId txnId)
{
- if (debugActive != null)
- debugActive.put(invoker.id, debug);
+ return pending(kind).contains(txnId);
}
- void undebugActive(CallbackInvoker<?, ?> invoker)
+ void start(CallbackInvoker<?, ?> invoker, Object task)
{
- if (debugActive != null)
- debugActive.remove(invoker.id);
+ active.put(invoker.id, task);
}
- boolean deregisterActive(TxnStateKind kind, Object object)
+ boolean complete(SafeCommandStore safeStore, TxnStateKind kind, long id, Object active)
{
- return active(kind).remove(object);
+ this.active.remove(id);
+ boolean result = pending(kind).remove(active);
+ if (runBufferIndex < runBufferCount)
+ accept(safeStore);
+ return result;
}
- void clearActive(TxnStateKind kind, TxnId txnId)
+ void clearPending(TxnStateKind kind, TxnId txnId)
{
- active(kind).remove(txnId);
+ pending(kind).remove(txnId);
}
void unschedule(TxnState state)
@@ -767,6 +813,17 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
}
}
+ public void setMaxConcurrency(int maxConcurrency)
+ {
+ Invariants.requireArgument(maxConcurrency >= 1);
+ this.maxConcurrency = maxConcurrency;
+ }
+
+ public int maxConcurrency()
+ {
+ return maxConcurrency;
+ }
+
public ImmutableView immutableView()
{
return new ImmutableView(commandStore.id(), stateMap);
@@ -885,6 +942,4 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
return current.homeRetryCounter();
}
}
-
-
}
diff --git a/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLogs.java b/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLogs.java
index cafe2c1c..3abedd04 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLogs.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLogs.java
@@ -18,49 +18,13 @@
package accord.impl.progresslog;
-import java.util.List;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CopyOnWriteArrayList;
-
import accord.api.ProgressLog;
import accord.local.CommandStore;
import accord.local.Node;
-import accord.local.PreLoadContext;
-// TODO (desired, consider): consider propagating invalidations in the same way as we do applied
-// TODO (expected): report transactions not making progress
-// TODO (expected): evict to disk
public class DefaultProgressLogs implements ProgressLog.Factory
{
- static ConcurrentHashMap<DefaultProgressLog, Runnable> PAUSE_FOR_TEST;
- public static synchronized void unsafePauseForTesting(boolean pause)
- {
- if (pause)
- {
- PAUSE_FOR_TEST = new ConcurrentHashMap<>();
- }
- else
- {
- ConcurrentHashMap<DefaultProgressLog, Runnable> reschedule = PAUSE_FOR_TEST;
- PAUSE_FOR_TEST = null;
- if (reschedule != null)
- reschedule.values().forEach(Runnable::run);
- }
- }
-
- static boolean pauseForTest(DefaultProgressLog progressLog)
- {
- ConcurrentHashMap<DefaultProgressLog, Runnable> paused = PAUSE_FOR_TEST;
- if (paused == null)
- return false;
-
- if (!paused.containsKey(progressLog))
- paused.putIfAbsent(progressLog, () -> progressLog.commandStore.execute((PreLoadContext.Empty) () -> "Unpause", progressLog, progressLog.commandStore.agent()));
- return true;
- }
-
- final Node node;
- final List<DefaultProgressLog> instances = new CopyOnWriteArrayList<>();
+ protected final Node node;
public DefaultProgressLogs(Node node)
{
@@ -70,8 +34,6 @@ public class DefaultProgressLogs implements ProgressLog.Factory
@Override
public DefaultProgressLog create(CommandStore commandStore)
{
- DefaultProgressLog instance = new DefaultProgressLog(node, commandStore);
- instances.add(instance);
- return instance;
+ return new DefaultProgressLog(node, commandStore);
}
}
diff --git a/accord-core/src/main/java/accord/impl/progresslog/HomeState.java b/accord-core/src/main/java/accord/impl/progresslog/HomeState.java
index 322ed8f9..28190c25 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/HomeState.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/HomeState.java
@@ -129,7 +129,7 @@ abstract class HomeState extends WaitingState
if (newPhase.compareTo(phase()) > 0)
{
- instance.clearActive(Home, txnId);
+ instance.clearPending(Home, txnId);
clearHomeRetryCounter();
set(safeStore, instance, newPhase, newProgress);
}
@@ -154,7 +154,7 @@ abstract class HomeState extends WaitingState
if (tracing != null)
tracing.trace(safeStore.commandStore(), "Invoking MaybeRecover with progress token %s", maxProgressToken);
- instance.debugActive(MaybeRecover.maybeRecover(instance.node(), txnId, invalidIf(), command.route(), maxProgressToken, reportTo, invoker), invoker);
+ instance.start(invoker, MaybeRecover.maybeRecover(instance.node(), txnId, invalidIf(), command.route(), maxProgressToken, reportTo, invoker));
set(safeStore, instance, ReadyToExecute, Querying);
}
@@ -225,7 +225,7 @@ abstract class HomeState extends WaitingState
{
set(null, instance, Done, NoneExpected);
clearHomeRetryCounter();
- instance.clearActive(Home, txnId);
+ instance.clearPending(Home, txnId);
}
void setHomeDoneAndMaybeRemove(DefaultProgressLog instance)
diff --git a/accord-core/src/main/java/accord/impl/progresslog/WaitingState.java b/accord-core/src/main/java/accord/impl/progresslog/WaitingState.java
index e047caa1..e7eb0922 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/WaitingState.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/WaitingState.java
@@ -349,7 +349,7 @@ abstract class WaitingState extends BaseTxnState
void setWaitingDone(DefaultProgressLog owner)
{
set(null, owner, CanApply, NoneExpected);
- owner.clearActive(Waiting, txnId);
+ owner.clearPending(Waiting, txnId);
clearWaitingRetryCounter();
}
@@ -360,7 +360,7 @@ abstract class WaitingState extends BaseTxnState
{
clearAwaitState();
clearWaitingRetryCounter();
- owner.clearActive(Waiting, txnId);
+ owner.clearPending(Waiting, txnId);
set(safeStore, owner, blockedUntil, Queued);
}
}
@@ -374,7 +374,7 @@ abstract class WaitingState extends BaseTxnState
set(null, owner, isDone ? CanApply : currentlyBlockedUntil, NoneExpected);
if (isDone)
maybeRemove(owner);
- owner.clearActive(Waiting, txnId);
+ owner.clearPending(Waiting, txnId);
}
}
@@ -387,7 +387,7 @@ abstract class WaitingState extends BaseTxnState
{
BlockedUntil blockedUntil = blockedUntil();
Command command = safeCommand.current();
- Invariants.require(!owner.hasActive(Waiting, txnId));
+ Invariants.require(!owner.hasPending(Waiting, txnId));
Invariants.require(command.saveStatus().compareTo(blockedUntil.unblockedFrom) < 0,
"Command has met desired criteria (%s) but progress log entry has not been cancelled: %s", blockedUntil.unblockedFrom, command);
set(safeStore, owner, blockedUntil, Querying);
diff --git a/accord-core/src/main/java/accord/local/Cleanup.java b/accord-core/src/main/java/accord/local/Cleanup.java
index 569fa94d..0f379cd4 100644
--- a/accord-core/src/main/java/accord/local/Cleanup.java
+++ b/accord-core/src/main/java/accord/local/Cleanup.java
@@ -37,6 +37,7 @@ import static accord.local.Cleanup.Input.PARTIAL;
import static accord.local.RedundantStatus.Property.GC_BEFORE;
import static accord.local.RedundantStatus.Property.LOCALLY_APPLIED;
import static accord.local.RedundantStatus.Property.LOCALLY_DEFUNCT;
+import static accord.local.RedundantStatus.Property.LOCALLY_DURABLE_TO_DATA_STORE;
import static accord.local.RedundantStatus.Property.LOCALLY_REDUNDANT;
import static accord.local.RedundantStatus.Property.NOT_OWNED;
import static accord.local.RedundantStatus.Property.SHARD_APPLIED;
@@ -195,6 +196,9 @@ public enum Cleanup
Invariants.paranoid(redundant.all(SHARD_APPLIED));
+ if (!redundant.all(LOCALLY_DURABLE_TO_DATA_STORE) && participants.doesStillExecute())
+ return truncateWithOutcome(txnId, min);
+
if (saveStatus.compareTo(Vestigial) >= 0)
{
// we can't use durability from an Invalidated record to decide if we can erase/expunge,
diff --git a/accord-core/src/main/java/accord/local/Command.java b/accord-core/src/main/java/accord/local/Command.java
index 168c7b98..50c55110 100644
--- a/accord-core/src/main/java/accord/local/Command.java
+++ b/accord-core/src/main/java/accord/local/Command.java
@@ -313,6 +313,18 @@ public abstract class Command implements ICommand
return orElse;
}
+ public final Timestamp applyAtIfKnown()
+ {
+ return applyAtIfKnown(null);
+ }
+
+ public final Timestamp applyAtIfKnown(Timestamp orElse)
+ {
+ if (known().is(ApplyAtKnown))
+ return executeAt();
+ return orElse;
+ }
+
public final Timestamp executeAtOrTxnId()
{
Timestamp executeAt = executeAt();
diff --git a/accord-core/src/main/java/accord/local/CommandStore.java b/accord-core/src/main/java/accord/local/CommandStore.java
index 1b25a180..30d6018a 100644
--- a/accord-core/src/main/java/accord/local/CommandStore.java
+++ b/accord-core/src/main/java/accord/local/CommandStore.java
@@ -62,8 +62,10 @@ import org.agrona.collections.LongHashSet;
import static accord.api.ConfigurationService.EpochReady.DONE;
import static accord.api.ProtocolModifiers.Toggles.requiresUniqueHlcs;
-import static accord.local.RedundantStatus.SomeStatus.GC_BEFORE_AND_LOCALLY_APPLIED;
+import static accord.local.RedundantStatus.SomeStatus.GC_BEFORE_AND_LOCALLY_DURABLE;
import static accord.local.RedundantStatus.SomeStatus.LOCALLY_APPLIED_ONLY;
+import static accord.local.RedundantStatus.SomeStatus.LOCALLY_DURABLE_TO_COMMAND_STORE_ONLY;
+import static accord.local.RedundantStatus.SomeStatus.LOCALLY_DURABLE_TO_DATA_STORE_ONLY;
import static accord.local.RedundantStatus.SomeStatus.LOCALLY_WITNESSED_ONLY;
import static accord.local.RedundantStatus.SomeStatus.MAJORITY_APPLIED_ONLY;
import static accord.local.RedundantStatus.SomeStatus.PRE_BOOTSTRAP_ONLY;
@@ -202,7 +204,9 @@ public abstract class CommandStore implements SequentialAsyncExecutor
public void restore() {};
- public abstract Journal.Loader loader();
+ public abstract Journal.Replayer replayer();
+ // expected to invoke safeStore.upsertRedundantBefore at some future point, when the commandStore state is durably persisted
+ protected abstract void ensureDurable(Ranges ranges, RedundantBefore onCommandStoreDurable);
public Agent agent()
{
@@ -478,10 +482,12 @@ public abstract class CommandStore implements SequentialAsyncExecutor
{
// TODO (desired): narrow ranges to those that are owned
Invariants.requireArgument(txnId.is(ExclusiveSyncPoint));
- RedundantBefore newRedundantBefore = RedundantBefore.merge(redundantBefore, RedundantBefore.create(ranges, txnId, LOCALLY_APPLIED_ONLY));
- safeStore.upsertRedundantBefore(newRedundantBefore);
- unsafeSetRedundantBefore(newRedundantBefore);
- updatedRedundantBefore(safeStore, txnId, ranges);
+ RedundantBefore addNow = RedundantBefore.create(ranges, txnId, LOCALLY_APPLIED_ONLY);
+ safeStore.upsertRedundantBefore(addNow);
+ RedundantBefore addOnDataStoreDurable = RedundantBefore.create(ranges, txnId, LOCALLY_DURABLE_TO_DATA_STORE_ONLY);
+ RedundantBefore addOnCommandStoreDurable = RedundantBefore.create(ranges, txnId, LOCALLY_DURABLE_TO_COMMAND_STORE_ONLY);
+ dataStore.ensureDurable(this, ranges, addOnDataStoreDurable);
+ ensureDurable(ranges, addOnCommandStoreDurable);
}
/**
@@ -616,7 +622,6 @@ public abstract class CommandStore implements SequentialAsyncExecutor
updateMaxConflicts(ranges, globalSyncId);
RedundantBefore addRedundantBefore = RedundantBefore.create(ranges, Long.MIN_VALUE, Long.MAX_VALUE, globalSyncId, PRE_BOOTSTRAP_ONLY);
safeStore.upsertRedundantBefore(addRedundantBefore);
- updatedRedundantBefore(safeStore, globalSyncId, ranges);
}
// TODO (expected): we can immediately truncate dependencies locally once an exclusiveSyncPoint applies, we don't need to wait for the whole shard
@@ -628,9 +633,8 @@ public abstract class CommandStore implements SequentialAsyncExecutor
SomeStatus status = durability.compareTo(Durability.UniversalOrInvalidated) >= 0 ? SHARD_APPLIED_ONLY : MAJORITY_APPLIED_ONLY;
final Ranges slicedRanges = durableRanges.slice(safeStore.ranges().allUntil(globalSyncId.epoch()), Minimal);
TxnId locallyRedundantBefore = safeStore.redundantBefore().min(slicedRanges, Bounds::maxLocallyAppliedBefore);
- RedundantBefore addShardRedundant = RedundantBefore.create(slicedRanges, globalSyncId, status);
- safeStore.upsertRedundantBefore(addShardRedundant);
- updatedRedundantBefore(safeStore, globalSyncId, slicedRanges);
+ RedundantBefore addNow = RedundantBefore.create(slicedRanges, globalSyncId, status);
+ safeStore.upsertRedundantBefore(addNow);
if (status != SHARD_APPLIED_ONLY)
return;
@@ -649,23 +653,12 @@ public abstract class CommandStore implements SequentialAsyncExecutor
// TODO (desired): not all systems care about HLC_BOUND for GC, make configurable
if (globalSyncId.is(HLC_BOUND) || !requiresUniqueHlcs())
{
- safeStore = safeStore; // make unusable in lambda
- safeStore.dataStore().snapshot(slicedRanges, globalSyncId).begin((success, fail) -> {
- if (fail != null)
- {
- agent.onCaughtException(fail, "Unsuccessful dataStore snapshot; unable to update GC markers");
- return;
- }
-
- execute((PreLoadContext.Empty) () -> "Mark Shard Durable", safeStore0 -> {
- RedundantBefore addGc = RedundantBefore.create(slicedRanges, globalSyncId, GC_BEFORE_AND_LOCALLY_APPLIED);
- safeStore0.upsertRedundantBefore(addGc);
- }, agent());
- });
+ RedundantBefore addOnDataStoreDurable = RedundantBefore.create(slicedRanges, globalSyncId, GC_BEFORE_AND_LOCALLY_DURABLE);
+ dataStore.ensureDurable(this, slicedRanges, addOnDataStoreDurable);
}
}
- protected void updatedRedundantBefore(SafeCommandStore safeStore, TxnId syncId, Ranges ranges)
+ protected void updatedRedundantBefore(SafeCommandStore safeStore, RedundantBefore added)
{
TxnId clearWaitingBefore = redundantBefore.minShardAndLocallyAppliedBefore();
TxnId clearAllBefore = TxnId.min(clearWaitingBefore, durableBefore().min.majorityBefore);
@@ -675,9 +668,8 @@ public abstract class CommandStore implements SequentialAsyncExecutor
protected void markSynced(SafeCommandStore safeStore, TxnId syncId, Ranges ranges)
{
- RedundantBefore newRedundantBefore = RedundantBefore.merge(redundantBefore, RedundantBefore.create(ranges, syncId, LOCALLY_WITNESSED_ONLY));
- unsafeSetRedundantBefore(newRedundantBefore);
- updatedRedundantBefore(safeStore, syncId, ranges);
+ RedundantBefore addRedundantBefore = RedundantBefore.create(ranges, syncId, LOCALLY_WITNESSED_ONLY);
+ safeStore.upsertRedundantBefore(addRedundantBefore);
if (waitingOnSync.isEmpty())
return;
diff --git a/accord-core/src/main/java/accord/local/Commands.java b/accord-core/src/main/java/accord/local/Commands.java
index 94f33821..cd511b8e 100644
--- a/accord-core/src/main/java/accord/local/Commands.java
+++ b/accord-core/src/main/java/accord/local/Commands.java
@@ -1070,14 +1070,14 @@ public class Commands
if (command.durability().compareTo(durability) >= 0)
return command;
- Command updated = supplementParticipants(command, participants);
+ Command updated = supplementParticipants(safeStore, safeCommand, participants);
participants = updated.participants();
if (executeAt != null && command.status().hasBeen(Committed) && !command.executeAt().equals(executeAt))
safeStore.agent().onInconsistentTimestamp(command, command.asCommitted().executeAt(), executeAt);
if (command.durability().compareTo(durability) < 0)
{
- updated = updated.updateDurability(durability);
+ updated = safeCommand.update(safeStore, updated.updateDurability(durability));
TxnId txnId = command.txnId();
if (dependencyElision() == IF_DURABLE && CommandsForKey.manages(txnId))
{
@@ -1101,7 +1101,6 @@ public class Commands
}
}
- updated = safeCommand.update(safeStore, updated);
if (maybeCleanup(safeStore, safeCommand, updated, participants))
updated = safeCommand.current();
diff --git a/accord-core/src/main/java/accord/local/PreLoadContext.java b/accord-core/src/main/java/accord/local/PreLoadContext.java
index 5c4e6415..e4da78fc 100644
--- a/accord-core/src/main/java/accord/local/PreLoadContext.java
+++ b/accord-core/src/main/java/accord/local/PreLoadContext.java
@@ -34,7 +34,6 @@ import javax.annotation.Nullable;
import static accord.local.LoadKeys.NONE;
import static accord.local.LoadKeys.SYNC;
-import static accord.local.LoadKeysFor.READ_WRITE;
import static accord.local.LoadKeysFor.WRITE;
/**
diff --git a/accord-core/src/main/java/accord/local/RedundantBefore.java b/accord-core/src/main/java/accord/local/RedundantBefore.java
index 8e9a7c72..a5bd0b57 100644
--- a/accord-core/src/main/java/accord/local/RedundantBefore.java
+++ b/accord-core/src/main/java/accord/local/RedundantBefore.java
@@ -62,6 +62,8 @@ import static accord.local.RedundantStatus.PRE_BOOTSTRAP_OR_STALE_ONLY;
import static accord.local.RedundantStatus.Property.GC_BEFORE;
import static accord.local.RedundantStatus.Property.LOCALLY_APPLIED;
import static accord.local.RedundantStatus.Property.LOCALLY_DEFUNCT;
+import static accord.local.RedundantStatus.Property.LOCALLY_DURABLE_TO_COMMAND_STORE;
+import static accord.local.RedundantStatus.Property.LOCALLY_DURABLE_TO_DATA_STORE;
import static accord.local.RedundantStatus.Property.LOCALLY_REDUNDANT;
import static accord.local.RedundantStatus.Property.LOCALLY_SYNCED;
import static accord.local.RedundantStatus.Property.LOCALLY_WITNESSED;
@@ -133,6 +135,8 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
public static class Bounds extends QuickBounds
{
+ public static final Bounds NONE = new Bounds(null, Long.MIN_VALUE, Long.MAX_VALUE, TxnId.NO_TXNIDS, new short[0], null);
+
// TODO (desired): we don't need to maintain this now, can migrate to ReducingRangeMap.foldWithBounds
public final Range range;
@@ -402,7 +406,7 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
Invariants.require(a.mergeWithPreBootstrapOrStale);
Invariants.require(b != GC_BEFORE);
Invariants.require(b.mergeWithPreBootstrapOrStale);
- return maxBound(a, b);
+ return maxBoundBoth(a, b);
}
boolean noBoundMatches(TxnId txnId)
@@ -436,7 +440,7 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
return maxBound(bounds, statuses, property);
}
- public TxnId maxBound(Property a, Property b)
+ public TxnId maxBoundBoth(Property a, Property b)
{
return maxBound(bounds, statuses, mask(a, SOME) | mask(b, SOME));
}
@@ -736,7 +740,7 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
private boolean isRetired()
{
- return endEpoch <= maxBound(SHARD_APPLIED, LOCALLY_SYNCED).epoch();
+ return endEpoch <= maxBoundBoth(SHARD_APPLIED, LOCALLY_SYNCED).epoch();
}
private boolean isLocallyRetired()
@@ -890,7 +894,7 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
public static RedundantBefore create(AbstractRanges ranges, long startEpoch, long endEpoch, TxnId bound, SomeStatus status, @Nullable Timestamp staleUntilAtLeast)
{
if (ranges.isEmpty())
- return new RedundantBefore();
+ return EMPTY;
Bounds bounds = new Bounds(null, startEpoch, endEpoch, new TxnId[] { bound }, new short[] { (short) (status.encoded & ONLY_LE_MASK), status.encoded }, staleUntilAtLeast);
Builder builder = new Builder(ranges.get(0).endInclusive(), ranges.size() * 2);
@@ -1253,17 +1257,25 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
@Override
public String toString()
{
- return "gc:" + toString(GC_BEFORE) + "\nlocal:" + toString(LOCALLY_APPLIED) + "\nbootstrap:" + toString(PRE_BOOTSTRAP);
+ return "gc:" + toString(GC_BEFORE)
+ + "\nlocal:" + toString(LOCALLY_DURABLE_TO_DATA_STORE, LOCALLY_DURABLE_TO_COMMAND_STORE)
+ + "\nbootstrap:" + toString(PRE_BOOTSTRAP);
+ }
+
+ private String toString(Property p1)
+ {
+ return toString(p1, null);
}
- private String toString(Property property)
+ private String toString(Property p1, Property p2)
{
TreeMap<TxnId, List<Range>> map = new TreeMap<>();
- foldl((e, m, p, o) -> {
- m.computeIfAbsent(e.maxBound(p), ignore -> new ArrayList<>())
+ foldl((e, m, pp1, pp2) -> {
+ TxnId bound = pp2 == null ? e.maxBound(pp1) : e.maxBoundBoth(pp1, pp2);
+ m.computeIfAbsent(bound, ignore -> new ArrayList<>())
.add(e.range);
return m;
- }, map, property, null, i -> false);
+ }, map, p1, p2, i -> false);
return map.descendingMap().entrySet().stream()
.map(e -> (e.getKey().equals(TxnId.NONE) ? "none" : e.getKey().toString()) + ":" + Ranges.ofSorted(e.getValue().toArray(new Range[0])).mergeTouching())
diff --git a/accord-core/src/main/java/accord/local/RedundantStatus.java b/accord-core/src/main/java/accord/local/RedundantStatus.java
index 080eff86..ae977baa 100644
--- a/accord-core/src/main/java/accord/local/RedundantStatus.java
+++ b/accord-core/src/main/java/accord/local/RedundantStatus.java
@@ -28,6 +28,8 @@ import static accord.local.RedundantStatus.Coverage.ALL;
import static accord.local.RedundantStatus.Coverage.SOME;
import static accord.local.RedundantStatus.Property.GC_BEFORE;
import static accord.local.RedundantStatus.Property.LOCALLY_APPLIED;
+import static accord.local.RedundantStatus.Property.LOCALLY_DURABLE_TO_COMMAND_STORE;
+import static accord.local.RedundantStatus.Property.LOCALLY_DURABLE_TO_DATA_STORE;
import static accord.local.RedundantStatus.Property.LOCALLY_SYNCED;
import static accord.local.RedundantStatus.Property.LOCALLY_WITNESSED;
import static accord.local.RedundantStatus.Property.MAJORITY_APPLIED;
@@ -86,32 +88,46 @@ public class RedundantStatus
* See also {@link SafeCommandStore#safeToReadAt()}.
* TODO (expected): do we need to distinguish this case from DEFUNCT?
*/
- PRE_BOOTSTRAP_OR_STALE (true, true, LT, LOCALLY_DEFUNCT),
- PRE_BOOTSTRAP (true, true, LT, PRE_BOOTSTRAP_OR_STALE),
+ PRE_BOOTSTRAP_OR_STALE ( true, true, LT, LOCALLY_DEFUNCT),
+ PRE_BOOTSTRAP ( true, true, LT, PRE_BOOTSTRAP_OR_STALE),
LOCALLY_WITNESSED (false, true, LE),
// we've applied a sync point locally covering the transaction, but the transaction itself may not have applied
LOCALLY_SYNCED (false, true, LE, LOCALLY_REDUNDANT),
- LOCALLY_APPLIED (true, false, LE, LOCALLY_SYNCED),
+ LOCALLY_APPLIED ( true, false, LE, LOCALLY_SYNCED),
/**
* We have fully executed until across all a majority of replicas for the range in question,
* but not necessarily ourselves.
*/
- MAJORITY_APPLIED (false, true, LE),
+ MAJORITY_APPLIED (false, true, LE),
/**
* We have fully executed until across all healthy non-bootstrapping replicas for the range in question,
* but not necessarily ourselves.
*/
- SHARD_APPLIED (false, true, LE, MAJORITY_APPLIED),
+ SHARD_APPLIED (false, true, LE, MAJORITY_APPLIED),
TRUNCATE_BEFORE (false, true, LT, SHARD_APPLIED, LOCALLY_SYNCED),
GC_BEFORE (false, true, LT, TRUNCATE_BEFORE),
// not persisted
WAS_OWNED (false, false, LT, LOCALLY_DEFUNCT),
- NOT_OWNED (false, false, LT),
+ NOT_OWNED (false, false, LT),
+
+ // added later, so vaguely out of order (but no specific order to this enum)
+
+ /**
+ * We have applied the preceding transactions durably to the store, so that we can safely truncate the Write
+ * information as we will not need to replay it to the store
+ */
+ LOCALLY_DURABLE_TO_DATA_STORE (false, false, LE, LOCALLY_APPLIED),
+
+ /**
+ * We have applied the preceding transactions durably to all summary structures, so that on restart we do
+ * not need to replay the transaction to restore any internal state.
+ */
+ LOCALLY_DURABLE_TO_COMMAND_STORE (false, false, LE, LOCALLY_APPLIED),
;
@@ -158,7 +174,9 @@ public class RedundantStatus
public static final SomeStatus LOCALLY_APPLIED_ONLY = oneSlow(LOCALLY_APPLIED);
public static final SomeStatus MAJORITY_APPLIED_ONLY = oneSlow(MAJORITY_APPLIED);
public static final SomeStatus SHARD_APPLIED_ONLY = oneSlow(SHARD_APPLIED);
- public static final SomeStatus GC_BEFORE_AND_LOCALLY_APPLIED = multi(GC_BEFORE, LOCALLY_APPLIED);
+ public static final SomeStatus LOCALLY_DURABLE_TO_DATA_STORE_ONLY = oneSlow(LOCALLY_DURABLE_TO_DATA_STORE);
+ public static final SomeStatus LOCALLY_DURABLE_TO_COMMAND_STORE_ONLY = oneSlow(LOCALLY_DURABLE_TO_COMMAND_STORE);
+ public static final SomeStatus GC_BEFORE_AND_LOCALLY_DURABLE = multi(GC_BEFORE, LOCALLY_DURABLE_TO_DATA_STORE, LOCALLY_DURABLE_TO_COMMAND_STORE);
final short encoded;
public SomeStatus(short encoded)
diff --git a/accord-core/src/main/java/accord/local/SafeCommand.java b/accord-core/src/main/java/accord/local/SafeCommand.java
index 943c4483..d0cb5dae 100644
--- a/accord-core/src/main/java/accord/local/SafeCommand.java
+++ b/accord-core/src/main/java/accord/local/SafeCommand.java
@@ -70,6 +70,7 @@ public abstract class SafeCommand
if (prev == update && !force)
return update;
+ Invariants.require(prev == null || update == SaveStatus.max(update, prev, Command::saveStatus, Command::acceptedOrCommitted) || update.saveStatus() == SaveStatus.Erased || update.saveStatus() == SaveStatus.Vestigial);
set(update);
safeStore.progressLog().update(safeStore, txnId, prev, update, force);
safeStore.update(prev, update, force);
diff --git a/accord-core/src/main/java/accord/local/SafeCommandStore.java b/accord-core/src/main/java/accord/local/SafeCommandStore.java
index 814a4d94..da892a7a 100644
--- a/accord-core/src/main/java/accord/local/SafeCommandStore.java
+++ b/accord-core/src/main/java/accord/local/SafeCommandStore.java
@@ -33,12 +33,10 @@ import accord.local.cfk.CommandsForKey;
import accord.local.cfk.SafeCommandsForKey;
import accord.local.cfk.UpdateUnmanagedMode;
import accord.primitives.AbstractUnseekableKeys;
-import accord.primitives.FullRoute;
import accord.primitives.KeyDeps;
import accord.primitives.Participants;
import accord.primitives.RangeDeps;
import accord.primitives.Ranges;
-import accord.primitives.Routables;
import accord.primitives.RoutingKeys;
import accord.primitives.SaveStatus;
import accord.primitives.Status;
@@ -342,6 +340,7 @@ public abstract class SafeCommandStore implements RangesForEpochSupplier, Redund
protected void unsafeUpsertRedundantBefore(RedundantBefore addRedundantBefore)
{
commandStore().unsafeUpsertRedundantBefore(addRedundantBefore);
+ commandStore().updatedRedundantBefore(this, addRedundantBefore);
}
public void setBootstrapBeganAt(NavigableMap<TxnId, Ranges> newBootstrapBeganAt)
@@ -366,7 +365,7 @@ public abstract class SafeCommandStore implements RangesForEpochSupplier, Redund
TxnId txnId = next.txnId();
if (CommandsForKey.manages(txnId)) updateManagedCommandsForKey(this, prev, next);
- if (!CommandsForKey.managesExecution(txnId) && next.hasBeen(Status.Stable) && !next.hasBeen(Status.Truncated) && !prev.hasBeen(Status.Stable))
+ if (!CommandsForKey.managesExecution(txnId) && next.hasBeen(Status.Stable) && !next.hasBeen(Status.Truncated) && (force || !prev.hasBeen(Status.Stable)))
updateUnmanagedCommandsForKey(this, next, REGISTER);
// TODO (expected): register deps during Accept phase to more quickly sync epochs
// else if (txnId.is(Range) && next.known().deps.hasProposedOrDecidedDeps())
diff --git a/accord-core/src/main/java/accord/local/cfk/CommandsForKey.java b/accord-core/src/main/java/accord/local/cfk/CommandsForKey.java
index 58036157..710e086f 100644
--- a/accord-core/src/main/java/accord/local/cfk/CommandsForKey.java
+++ b/accord-core/src/main/java/accord/local/cfk/CommandsForKey.java
@@ -2315,7 +2315,13 @@ public class CommandsForKey extends CommandsForKeyUpdate
case APPLY:
{
int byExecuteAtIndex = SortedArrays.binarySearch(committedByExecuteAt, 0, committedByExecuteAt.length, unmanaged.waitingUntil, (f, i) -> f.compareTo(i.executeAt), FAST);
- Invariants.require(byExecuteAtIndex >= 0 && byExecuteAtIndex >= appliedBefore);
+ if (byExecuteAtIndex >= 0 && byExecuteAtIndex < appliedBefore)
+ {
+ Invariants.require(!reportLinearizabilityViolations);
+ int i = 0;
+ for (; i < appliedBefore && committedByExecuteAt[i].compareTo(APPLIED) >= 0 ; ++i){}
+ Invariants.require(i < appliedBefore || isLoadingPruned());
+ }
break;
}
}
diff --git a/accord-core/src/main/java/accord/local/cfk/Serialize.java b/accord-core/src/main/java/accord/local/cfk/Serialize.java
index 3c25a1bf..9162c71a 100644
--- a/accord-core/src/main/java/accord/local/cfk/Serialize.java
+++ b/accord-core/src/main/java/accord/local/cfk/Serialize.java
@@ -478,6 +478,7 @@ public class Serialize
long bits = txn.status().ordinal();
int bitIndex = 4;
+ // TODO (desired): these are all single bit flags, so we could do explicit shift and mask to produce this value
int statusHasExecuteAt = txn.hasExecuteAt() ? 1 : 0;
int statusHasDeps = txn.hasDeps() ? 1 : 0;
int statusHasBallot = txn.hasBallot() ? 1 : 0;
diff --git a/accord-core/src/main/java/accord/local/durability/ShardDurability.java b/accord-core/src/main/java/accord/local/durability/ShardDurability.java
index afd0f91d..01f90a53 100644
--- a/accord-core/src/main/java/accord/local/durability/ShardDurability.java
+++ b/accord-core/src/main/java/accord/local/durability/ShardDurability.java
@@ -138,7 +138,7 @@ public class ShardDurability
Invariants.require(shard == null || shard.range.equals(newShard.range));
shard = newShard;
ShardDistributor shardDistributor = node.commandStores().shardDistributor();
- maxSplits = Math.min(maxNumberOfShardSplits, shardDistributor.numberOfSplitsPossible(newShard.range));
+ maxSplits = Math.min(maxShardSplits, shardDistributor.numberOfSplitsPossible(newShard.range));
int targetSplits = Math.min(targetShardSplits, maxSplits);
if (currentSplits == 0 || currentSplits < targetSplits)
{
@@ -305,7 +305,7 @@ public class ShardDurability
sb.append(entry);
return sb;
}, new StringBuilder(), ignore -> false).toString();
- logger.info("{}", report);
+ logger.debug("{}", report);
}
else if (min.equals(cycleMin))
{
@@ -424,7 +424,7 @@ public class ShardDurability
* In each cycle, attempt to split the range into this many pieces; if we fail, we increase the number of pieces
*/
private int targetShardSplits = 64;
- private int maxNumberOfShardSplits = 1 << 10;
+ private int maxShardSplits = 1 << 10;
/*
* Target for how often the entire ring should be processed in microseconds. Every node will start at an offset in the current round that is based
@@ -463,7 +463,7 @@ public class ShardDurability
public void setMaxShardSplits(int maxNumberOfShardSplits)
{
- this.maxNumberOfShardSplits = maxNumberOfShardSplits;
+ this.maxShardSplits = maxNumberOfShardSplits;
}
public synchronized void setTargetShardSplits(int targetShardSplits)
@@ -482,8 +482,9 @@ public class ShardDurability
scheduled = node.scheduler().recurring(this::tick, shardCycleTimeMicros / targetShardSplits, MICROSECONDS);
}
- public synchronized void reconfigure(int targetShardSplits, long newShardCycleTime, TimeUnit units)
+ public synchronized void reconfigure(int targetShardSplits, int maxShardSplits, long newShardCycleTime, TimeUnit units)
{
+ this.maxShardSplits = maxShardSplits;
this.targetShardSplits = BitUtil.findNextPositivePowerOfTwo(targetShardSplits);
this.shardCycleTimeMicros = units.toMicros(newShardCycleTime);
Invariants.require(scheduled == null);
diff --git a/accord-core/src/main/java/accord/messages/Accept.java b/accord-core/src/main/java/accord/messages/Accept.java
index 47b47295..4a46e851 100644
--- a/accord-core/src/main/java/accord/messages/Accept.java
+++ b/accord-core/src/main/java/accord/messages/Accept.java
@@ -74,9 +74,11 @@ public class Accept extends TxnRequest.WithUnsynced<Accept.AcceptReply>
public final Kind kind;
public final Ballot ballot;
public final Timestamp executeAt;
- public final PartialDeps partialDeps;
+ private PartialDeps partialDeps;
public final boolean isPartialAccept;
+ public PartialDeps partialDeps() { return partialDeps; }
+
public Accept(Id to, Topologies topologies, Kind kind, Ballot ballot, TxnId txnId, FullRoute<?> route, Timestamp executeAt, Deps deps, boolean isPartialAccept)
{
super(to, topologies, txnId, route);
@@ -194,6 +196,14 @@ public class Accept extends TxnRequest.WithUnsynced<Accept.AcceptReply>
return node.mapReduceConsumeLocal(this, minEpoch, executeAt.epoch(), this);
}
+ @Override
+ protected void acceptInternal(AcceptReply reply, Throwable failure)
+ {
+ // finished processing, null out large objects
+ partialDeps = null;
+ super.acceptInternal(reply, failure);
+ }
+
@Override
public LoadKeys loadKeys()
{
diff --git a/accord-core/src/main/java/accord/messages/Apply.java b/accord-core/src/main/java/accord/messages/Apply.java
index 3cc7dd49..66745ca4 100644
--- a/accord-core/src/main/java/accord/messages/Apply.java
+++ b/accord-core/src/main/java/accord/messages/Apply.java
@@ -77,19 +77,24 @@ public class Apply extends TxnRequest<ApplyReply>
Apply create(Kind kind, Id to, Topologies participates, TxnId txnId, Ballot ballot, Route<?> scope, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute, ExecuteFlags flags);
}
+ public enum Kind { Minimal, Maximal }
+
public final Kind kind;
public final Ballot ballot;
public final Timestamp executeAt;
- public final PartialDeps deps; // not much benefit in nullability, as we try only to commit to readers, so most recipients will need to receive deps
- public final @Nullable PartialTxn txn;
+ private PartialDeps deps; // not much benefit in nullability, as we try only to commit to readers, so most recipients will need to receive deps
+ private @Nullable PartialTxn txn;
public final @Nullable FullRoute<?> fullRoute;
- public final @Nullable Writes writes;
- public final Result result;
+ private @Nullable Writes writes;
+ private Result result;
public final long minEpoch;
public final long maxEpoch;
public final ExecuteFlags flags;
- public enum Kind { Minimal, Maximal }
+ public PartialTxn txn() { return txn; }
+ public PartialDeps deps() { return deps; }
+ public Writes writes() { return writes; }
+ public Result result() { return result; }
protected Apply(Kind kind, Id to, Topologies participates, TxnId txnId, Ballot ballot, Route<?> sendTo, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute, ExecuteFlags flags)
{
@@ -151,6 +156,16 @@ public class Apply extends TxnRequest<ApplyReply>
return apply(safeStore, participants);
}
+ @Override
+ protected void acceptInternal(ApplyReply reply, Throwable failure)
+ {
+ txn = null;
+ deps = null;
+ writes = null;
+ result = null;
+ super.acceptInternal(reply, failure);
+ }
+
// TODO (desired): always applyDirect reads, whether or not the replica is ready, as reads are no-ops (note: affects linearizability warnings)
private AsyncChain<ApplyReply> applyDirect(CommandStore commandStore)
{
diff --git a/accord-core/src/main/java/accord/messages/ApplyThenWaitUntilApplied.java b/accord-core/src/main/java/accord/messages/ApplyThenWaitUntilApplied.java
index d2c49543..961f410b 100644
--- a/accord-core/src/main/java/accord/messages/ApplyThenWaitUntilApplied.java
+++ b/accord-core/src/main/java/accord/messages/ApplyThenWaitUntilApplied.java
@@ -63,10 +63,15 @@ public class ApplyThenWaitUntilApplied extends WaitUntilApplied
}
public final FullRoute<?> route;
- public final PartialTxn txn;
- public final PartialDeps deps;
- public final Writes writes;
- public final Result result;
+ private PartialTxn txn;
+ private PartialDeps deps;
+ private Writes writes;
+ private Result result;
+
+ public PartialTxn txn() { return txn; }
+ public PartialDeps deps() { return deps; }
+ public Writes writes() { return writes; }
+ public Result result() { return result; }
public ApplyThenWaitUntilApplied(Node.Id to, Topologies topologies, Timestamp executeAt, FullRoute<?> route, TxnId txnId, Txn txn, Deps deps, Participants<?> readScope, Writes writes, Result result)
{
@@ -135,6 +140,11 @@ public class ApplyThenWaitUntilApplied extends WaitUntilApplied
}
if (waiting && reply == null && failure == null)
node.reply(replyTo, replyContext, CommitOrReadNack.Waiting, null);
+
+ txn = null;
+ deps = null;
+ writes = null;
+ result = null;
}
@Override
diff --git a/accord-core/src/main/java/accord/messages/BeginInvalidation.java b/accord-core/src/main/java/accord/messages/BeginInvalidation.java
index 1f4e63f9..747773e1 100644
--- a/accord-core/src/main/java/accord/messages/BeginInvalidation.java
+++ b/accord-core/src/main/java/accord/messages/BeginInvalidation.java
@@ -238,17 +238,17 @@ public class BeginInvalidation extends AbstractRequest<BeginInvalidation.Invalid
public static InvalidateReply max(InvalidateReply[] invalidateReplies, Shard shard, SortedList<Id> nodeIds)
{
- return SaveStatus.max(nodeIds.select(invalidateReplies, shard.nodes), r -> r.maxStatus, r -> r.accepted, Objects::nonNull);
+ return SaveStatus.maxOfList(nodeIds.select(invalidateReplies, shard.nodes), r -> r.maxStatus, r -> r.accepted, Objects::nonNull);
}
public static InvalidateReply max(InvalidateReply[] invalidateReplies)
{
- return SaveStatus.max(Arrays.asList(invalidateReplies), r -> r.maxStatus, r -> r.accepted, Objects::nonNull);
+ return SaveStatus.maxOfList(Arrays.asList(invalidateReplies), r -> r.maxStatus, r -> r.accepted, Objects::nonNull);
}
public static InvalidateReply maxNotTruncated(InvalidateReply[] invalidateReplies)
{
- return SaveStatus.max(Arrays.asList(invalidateReplies), r -> r.maxKnowledgeStatus, r -> r.accepted, r -> r != null && !r.maxKnowledgeStatus.is(Status.Truncated));
+ return SaveStatus.maxOfList(Arrays.asList(invalidateReplies), r -> r.maxKnowledgeStatus, r -> r.accepted, r -> r != null && !r.maxKnowledgeStatus.is(Status.Truncated));
}
}
}
diff --git a/accord-core/src/main/java/accord/messages/ReadEphemeralTxnData.java b/accord-core/src/main/java/accord/messages/ReadEphemeralTxnData.java
index f14aa06d..54d92739 100644
--- a/accord-core/src/main/java/accord/messages/ReadEphemeralTxnData.java
+++ b/accord-core/src/main/java/accord/messages/ReadEphemeralTxnData.java
@@ -117,9 +117,9 @@ public class ReadEphemeralTxnData extends ReadData
@Override
public void accept(CommitOrReadNack reply, Throwable failure)
{
- super.accept(reply, failure);
partialDeps = null;
route = null;
+ super.accept(reply, failure);
}
@Override
diff --git a/accord-core/src/main/java/accord/messages/StableThenRead.java b/accord-core/src/main/java/accord/messages/StableThenRead.java
index acc0033b..c6fd56d9 100644
--- a/accord-core/src/main/java/accord/messages/StableThenRead.java
+++ b/accord-core/src/main/java/accord/messages/StableThenRead.java
@@ -99,8 +99,8 @@ public class StableThenRead extends ReadData
@Override
public void accept(CommitOrReadNack reply, Throwable failure)
{
- super.accept(reply, failure);
partialDeps = null;
+ super.accept(reply, failure);
}
@Override
diff --git a/accord-core/src/main/java/accord/primitives/SaveStatus.java b/accord-core/src/main/java/accord/primitives/SaveStatus.java
index 80c2080f..5c7c7090 100644
--- a/accord-core/src/main/java/accord/primitives/SaveStatus.java
+++ b/accord-core/src/main/java/accord/primitives/SaveStatus.java
@@ -425,7 +425,23 @@ public enum SaveStatus
return a.compareTo(b) >= 0 ? av : bv;
}
- public static <T> T max(List<T> list, Function<T, SaveStatus> getStatus, Function<T, Ballot> getAcceptedOrCommittedBallot, Predicate<T> filter)
+ public static <T> T max(T av, T bv, Function<T, SaveStatus> getStatus, Function<T, Ballot> getAcceptedOrCommittedBallot)
+ {
+ SaveStatus a = getStatus.apply(av), b = getStatus.apply(bv);
+
+ if (a == b)
+ return av;
+
+ if (a.phase != b.phase)
+ return a.phase.compareTo(b.phase) > 0 ? av : bv;
+
+ if (a.phase.tieBreakWithBallot)
+ return getAcceptedOrCommittedBallot.apply(av).compareTo(getAcceptedOrCommittedBallot.apply(bv)) >= 0 ? av : bv;
+
+ return a.compareTo(b) >= 0 ? av : bv;
+ }
+
+ public static <T> T maxOfList(List<T> list, Function<T, SaveStatus> getStatus, Function<T, Ballot> getAcceptedOrCommittedBallot, Predicate<T> filter)
{
T max = null;
SaveStatus maxStatus = null;
diff --git a/accord-core/src/main/java/accord/utils/ArrayBuffers.java b/accord-core/src/main/java/accord/utils/ArrayBuffers.java
index 3b5527ee..5e228e53 100644
--- a/accord-core/src/main/java/accord/utils/ArrayBuffers.java
+++ b/accord-core/src/main/java/accord/utils/ArrayBuffers.java
@@ -55,6 +55,7 @@ import static accord.utils.Invariants.illegalState;
public class ArrayBuffers
{
private static final boolean FULLY_UNCACHED = true;
+ public static final int MIN_BUFFER_SIZE = 8;
// TODO (low priority, efficiency): we should periodically clear the thread locals to ensure we aren't slowly accumulating unnecessarily large objects on every thread
private static final ThreadLocal<ShortBufferCache> SHORTS = ThreadLocal.withInitial(() -> new ShortBufferCache(4, 1 << 10));
@@ -510,8 +511,6 @@ public class ArrayBuffers
void clear(B array, int usedSize);
}
- static final int MIN_SIZE = 8;
-
final IntFunction<B> allocator;
final Clear<B> clear;
final B empty;
@@ -521,7 +520,7 @@ public class ArrayBuffers
AbstractBufferCache(IntFunction<B> allocator, Clear<B> clear, int maxCount, int maxSize)
{
this.allocator = allocator;
- this.maxSize = Math.max(MIN_SIZE, maxSize);
+ this.maxSize = Math.max(MIN_BUFFER_SIZE, maxSize);
this.cached = (B[])new Object[maxCount];
this.empty = allocator.apply(0);
this.clear = clear;
@@ -535,7 +534,7 @@ public class ArrayBuffers
if (minSize > maxSize)
return allocator.apply(minSize);
- minSize = Math.max(minSize, MIN_SIZE);
+ minSize = Math.max(minSize, MIN_BUFFER_SIZE);
for (int i = 0 ; i < cached.length ; ++i)
{
if (cached[i] != null && Array.getLength(cached[i]) >= minSize)
@@ -554,7 +553,7 @@ public class ArrayBuffers
if (bufferSize == 0 || bufferSize > maxSize)
return true;
- Invariants.require(bufferSize >= MIN_SIZE,
+ Invariants.require(bufferSize >= MIN_BUFFER_SIZE,
"Trying to return an array of size %d we cannot have allocated, suggesting a logic bug that could lead to overwriting an in-use array.",
bufferSize);
if (bufferSize == usedSize && !force)
diff --git a/accord-core/src/main/java/accord/utils/IntrusiveLinkedList.java b/accord-core/src/main/java/accord/utils/IntrusiveLinkedList.java
index f223fbea..209ea90a 100644
--- a/accord-core/src/main/java/accord/utils/IntrusiveLinkedList.java
+++ b/accord-core/src/main/java/accord/utils/IntrusiveLinkedList.java
@@ -42,8 +42,7 @@ public class IntrusiveLinkedList<O extends IntrusiveLinkedListNode> extends Intr
public void addFirst(O add)
{
- if (add.next != null)
- throw new IllegalStateException();
+ Invariants.requireArgument(add.next == null);
add(this, add, next);
}
diff --git a/accord-core/src/main/java/accord/utils/Invariants.java b/accord-core/src/main/java/accord/utils/Invariants.java
index bce558ea..0ac5aad3 100644
--- a/accord-core/src/main/java/accord/utils/Invariants.java
+++ b/accord-core/src/main/java/accord/utils/Invariants.java
@@ -99,7 +99,7 @@ public class Invariants
public static IllegalArgumentException illegalArgument(String fmt, Object ... args)
{
- throw new IllegalArgumentException(String.format(fmt, args));
+ throw illegalArgument(String.format(fmt, args));
}
private static IllegalArgumentException illegalArgument()
diff --git a/accord-core/src/main/java/accord/utils/ReducingIntervalMap.java b/accord-core/src/main/java/accord/utils/ReducingIntervalMap.java
index 37aaacf5..8882dabe 100644
--- a/accord-core/src/main/java/accord/utils/ReducingIntervalMap.java
+++ b/accord-core/src/main/java/accord/utils/ReducingIntervalMap.java
@@ -194,6 +194,11 @@ public class ReducingIntervalMap<K extends Comparable<? super K>, V>
return values.length;
}
+ public boolean isEmpty()
+ {
+ return values.length == 0;
+ }
+
RangeIterator rangeIterator()
{
return new RangeIterator();
@@ -608,4 +613,9 @@ public class ReducingIntervalMap<K extends Comparable<? super K>, V>
else
return rightIsInclusive;
}
+
+ public K[] unsafeStarts()
+ {
+ return starts;
+ }
}
diff --git a/accord-core/src/main/java/accord/utils/async/AsyncResults.java b/accord-core/src/main/java/accord/utils/async/AsyncResults.java
index 6c954376..9ed0061e 100644
--- a/accord-core/src/main/java/accord/utils/async/AsyncResults.java
+++ b/accord-core/src/main/java/accord/utils/async/AsyncResults.java
@@ -19,6 +19,7 @@
package accord.utils.async;
import java.util.concurrent.Callable;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.function.BiConsumer;
import java.util.function.Function;
@@ -266,6 +267,37 @@ public class AsyncResults
}
}
+ public static class CountingResult extends AbstractResult<Void>
+ {
+ private volatile int count;
+ private static final AtomicIntegerFieldUpdater<CountingResult> countUpdater = AtomicIntegerFieldUpdater.newUpdater(CountingResult.class, "count");
+
+ public CountingResult(int initialCount)
+ {
+ count = initialCount;
+ if (initialCount == 0)
+ trySuccess(null);
+ }
+
+ public void decrement()
+ {
+ if (0 == countUpdater.decrementAndGet(this))
+ trySuccess(null);
+ }
+
+ public void increment()
+ {
+ if (0 == countUpdater.getAndIncrement(this))
+ throw new IllegalStateException("Count was already zero");
+ }
+
+ @Override
+ public boolean tryFailure(Throwable throwable)
+ {
+ return super.tryFailure(throwable);
+ }
+ }
+
static class Immediate<V> implements AsyncResult<V>
{
private final V value;
diff --git a/accord-core/src/test/java/accord/impl/basic/Pending.java b/accord-core/src/main/java/accord/utils/async/RunnableWithResult.java
similarity index 50%
copy from accord-core/src/test/java/accord/impl/basic/Pending.java
copy to accord-core/src/main/java/accord/utils/async/RunnableWithResult.java
index fd84fa88..b47452a8 100644
--- a/accord-core/src/test/java/accord/impl/basic/Pending.java
+++ b/accord-core/src/main/java/accord/utils/async/RunnableWithResult.java
@@ -16,39 +16,31 @@
* limitations under the License.
*/
-package accord.impl.basic;
+package accord.utils.async;
-import accord.utils.Invariants;
+import java.util.concurrent.Callable;
-public interface Pending
+public class RunnableWithResult<V> implements Callable<V>
{
- Pending origin();
+ final Runnable run;
+ final V result;
- // a slightly hacky but easy way to track transitive recurring tasks
- class Global
+ public RunnableWithResult(Runnable run, V result)
{
- public static final Pending NONE = () -> null;
- private static Pending activeOrigin;
-
- public static void setActiveOrigin(Pending newActive)
- {
- Invariants.require(activeOrigin == null);
- activeOrigin = newActive.origin();
- }
-
- public static void clearActiveOrigin()
- {
- activeOrigin = null;
- }
+ this.run = run;
+ this.result = result;
+ }
- public static void setNoActiveOrigin()
- {
- activeOrigin = NONE;
- }
+ @Override
+ public V call() throws Exception
+ {
+ run.run();
+ return result;
+ }
- public static Pending activeOrigin()
- {
- return Invariants.nonNull(activeOrigin);
- }
+ @Override
+ public String toString()
+ {
+ return "[Run " + run.toString() + "; return " + result + ']';
}
}
diff --git a/accord-core/src/test/java/accord/burn/BurnTestBase.java b/accord-core/src/test/java/accord/burn/BurnTestBase.java
index 95b3018a..6a686fb8 100644
--- a/accord-core/src/test/java/accord/burn/BurnTestBase.java
+++ b/accord-core/src/test/java/accord/burn/BurnTestBase.java
@@ -57,6 +57,7 @@ import accord.api.Journal;
import accord.api.Key;
import accord.api.ProtocolModifiers.Toggles;
import accord.api.ProtocolModifiers.Toggles.FastExec;
+import accord.api.Scheduler;
import accord.burn.random.FrequentLargeRange;
import accord.impl.MessageListener;
import accord.impl.PrefixedIntHashKey;
@@ -99,6 +100,7 @@ import accord.utils.DefaultRandom;
import accord.utils.Gen;
import accord.utils.Gens;
import accord.utils.RandomSource;
+import accord.utils.TriFunction;
import accord.utils.UnhandledEnum;
import accord.utils.Utils;
import accord.api.AsyncExecutor;
@@ -412,8 +414,8 @@ public class BurnTestBase
.asLongSupplier(forked);
};
Supplier<TimeService> timeServiceSupplier = () -> TimeService.ofNonMonotonic(nowSupplier.get(), MILLISECONDS);
- BiFunction<BiConsumer<Timestamp, Ranges>, NodeSink.TimeoutSupplier, Agent> agentSupplier = (onStale, timeoutSupplier) -> new ListAgent(random.fork(), 1000L, failures::add, retryBootstrap, onStale, coordinationDelays, progressDelays, timeoutDelays, pendingQueue::nowInMillis, timeServiceSupplier.get(), timeoutSupplier);
- SimulatedDelayedExecutorService globalExecutor = new SimulatedDelayedExecutorService(queue, new ListAgent(random.fork(), 1000L, failures::add, retryBootstrap, (i1, i2) -> {
+ TriFunction<BiConsumer<Timestamp, Ranges>, Scheduler, NodeSink.TimeoutSupplier, Agent> agentSupplier = (onStale, scheduler, timeoutSupplier) -> new ListAgent(scheduler, random.fork(), 1000L, failures::add, retryBootstrap, onStale, coordinationDelays, progressDelays, timeoutDelays, pendingQueue::nowInMillis, timeServiceSupplier.get(), timeoutSupplier);
+ SimulatedDelayedExecutorService globalExecutor = new SimulatedDelayedExecutorService(queue, new ListAgent(null, random.fork(), 1000L, failures::add, retryBootstrap, (i1, i2) -> {
throw new IllegalAccessError("Global executor should enver get a stale event");
}, coordinationDelays, progressDelays, timeoutDelays, queue::nowInMillis, timeServiceSupplier.get(), null), null);
Verifier verifier = createVerifier(keyCount * prefixCount);
diff --git a/accord-core/src/test/java/accord/coordinate/CoordinateTransactionTest.java b/accord-core/src/test/java/accord/coordinate/CoordinateTransactionTest.java
index 7ffa4580..9658b4bb 100644
--- a/accord-core/src/test/java/accord/coordinate/CoordinateTransactionTest.java
+++ b/accord-core/src/test/java/accord/coordinate/CoordinateTransactionTest.java
@@ -21,14 +21,12 @@ package accord.coordinate;
import java.util.concurrent.ExecutionException;
import java.util.function.BiConsumer;
-import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import accord.api.LocalListeners;
import accord.api.Result;
import accord.impl.mock.MockCluster;
import accord.impl.mock.MockStore;
-import accord.impl.progresslog.DefaultProgressLogs;
import accord.local.Node;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
@@ -57,12 +55,6 @@ import static org.junit.jupiter.api.Assertions.fail;
public class CoordinateTransactionTest
{
- @AfterEach
- public void tearDown()
- {
- DefaultProgressLogs.unsafePauseForTesting(false);
- }
-
@Test
void simpleTest() throws Throwable
{
diff --git a/accord-core/src/test/java/accord/impl/RemoteListenersTest.java b/accord-core/src/test/java/accord/impl/RemoteListenersTest.java
index c8dafd88..4e0e9f80 100644
--- a/accord-core/src/test/java/accord/impl/RemoteListenersTest.java
+++ b/accord-core/src/test/java/accord/impl/RemoteListenersTest.java
@@ -55,6 +55,7 @@ import accord.local.RedundantBefore;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.primitives.RangeDeps;
+import accord.primitives.Ranges;
import accord.primitives.SaveStatus;
import accord.primitives.Status.Durability;
import accord.local.cfk.SafeCommandsForKey;
@@ -404,11 +405,12 @@ public class RemoteListenersTest
}
@Override
- public Journal.Loader loader()
+ public Journal.Replayer replayer()
{
throw new UnsupportedOperationException();
}
+ @Override protected void ensureDurable(Ranges ranges, RedundantBefore onCommandStoreDurable) {}
@Override public boolean inStore() { return false; }
@Override public AsyncChain<Void> build(PreLoadContext context, Consumer<? super SafeCommandStore> consumer) { return null; }
@Override public <T> AsyncChain<T> build(PreLoadContext context, Function<? super SafeCommandStore, T> apply) { return null; }
diff --git a/accord-core/src/test/java/accord/impl/basic/Cluster.java b/accord-core/src/test/java/accord/impl/basic/Cluster.java
index 3ba061e2..a0acf419 100644
--- a/accord-core/src/test/java/accord/impl/basic/Cluster.java
+++ b/accord-core/src/test/java/accord/impl/basic/Cluster.java
@@ -73,9 +73,9 @@ import accord.impl.PrefixedIntHashKey;
import accord.impl.SizeOfIntersectionSorter;
import accord.impl.TopologyFactory;
import accord.impl.basic.DelayedCommandStores.DelayedCommandStore;
+import accord.impl.basic.TestProgressLogs.TestProgressLog;
import accord.impl.list.ListAgent;
import accord.impl.list.ListStore;
-import accord.impl.progresslog.DefaultProgressLogs;
import accord.local.Cleanup;
import accord.local.Command;
import accord.local.CommandStore;
@@ -111,6 +111,7 @@ import accord.utils.LazyToString;
import accord.utils.RandomSource;
import accord.utils.ReflectionUtils;
import accord.utils.Timestamped;
+import accord.utils.TriFunction;
import accord.utils.UnhandledEnum;
import accord.utils.async.AsyncChains;
import accord.utils.async.AsyncResult;
@@ -240,6 +241,7 @@ public class Cluster
final MessageListener messageListener;
int clock;
BiFunction<Id, Id, Link> links;
+ long droppedAt;
public Cluster(RandomSource random, MessageListener messageListener, Supplier<PendingQueue> queueSupplier, Runnable checkFailures, Function<Id, Node> lookup, Function<Id, Journal> journalLookup, IntSupplier rf, Consumer<Packet> responseSink)
{
@@ -315,7 +317,7 @@ public class Cluster
{
checkFailures.run();
// All remaining tasks are recurring
- if (!hasNonRecurring())
+ if (!hasNonRecurring() && pending.nowInMillis() > droppedAt + TimeUnit.MINUTES.toMillis(1L))
return false;
Pending next = pending.poll();
@@ -379,6 +381,8 @@ public class Cluster
{
if (trace.isTraceEnabled())
trace.trace("{} DROP[{}] (from:{}, to:{}, {}:{}, body:{})", clock++, lookup.apply(to).epoch(), from, to, message instanceof Reply ? "replyTo" : "id", id, message);
+ if (!RecurringPendingRunnable.isRecurring(Pending.Global.activeOrigin()))
+ droppedAt = pending.nowInMillis();
}
public void onDone(Runnable run)
@@ -421,8 +425,8 @@ public class Cluster
.asLongSupplier(forked);
};
Supplier<TimeService> timeServiceSupplier = () -> TimeService.ofNonMonotonic(nowSupplier.get(), MILLISECONDS);
- BiFunction<BiConsumer<Timestamp, Ranges>, NodeSink.TimeoutSupplier, Agent> agentSupplier = (onStale, timeoutSupplier) -> new ListAgent(randomSupplier.get(), 1000L, failures::add, retryBootstrap, onStale, coordinationDelays, progressDelays, timeoutDelays, queue::nowInMillis, timeServiceSupplier.get(), timeoutSupplier);
- SimulatedDelayedExecutorService globalExecutor = new SimulatedDelayedExecutorService(queue, new ListAgent(randomSupplier.get(), 1000L, failures::add, retryBootstrap, (i1, i2) -> {
+ TriFunction<BiConsumer<Timestamp, Ranges>, Scheduler, NodeSink.TimeoutSupplier, Agent> agentSupplier = (onStale, scheduler, timeoutSupplier) -> new ListAgent(scheduler, randomSupplier.get(), 1000L, failures::add, retryBootstrap, onStale, coordinationDelays, progressDelays, timeoutDelays, queue::nowInMillis, timeServiceSupplier.get(), timeoutSupplier);
+ SimulatedDelayedExecutorService globalExecutor = new SimulatedDelayedExecutorService(queue, new ListAgent(null, randomSupplier.get(), 1000L, failures::add, retryBootstrap, (i1, i2) -> {
throw new IllegalAccessError("Global executor should never get a stale event");
}, () -> { throw new UnsupportedOperationException(); }, () -> { throw new UnsupportedOperationException(); }, timeoutDelays, queue::nowInMillis, timeServiceSupplier.get(), null), null);
TopologyFactory topologyFactory = new TopologyFactory(initialTopology.maxRf(), initialTopology.ranges().stream().toArray(Range[]::new))
@@ -616,7 +620,7 @@ public class Cluster
public static Map<MessageType, Stats> run(Id[] nodes, int[] prefixes, MessageListener messageListener, Supplier<PendingQueue> queueSupplier,
Function<Id, AsyncExecutor> nodeExecutorSupplier,
- BiFunction<BiConsumer<Timestamp, Ranges>, NodeSink.TimeoutSupplier, Agent> agentSupplier,
+ TriFunction<BiConsumer<Timestamp, Ranges>, Scheduler, NodeSink.TimeoutSupplier, Agent> agentSupplier,
Runnable checkFailures, Consumer<Packet> responseSink,
Supplier<RandomSource> randomSupplier,
Supplier<TimeService> timeServiceSupplier,
@@ -674,7 +678,7 @@ public class Cluster
@Override public TimeUnit units() { return MILLISECONDS; }
};
TopologyRandomizer configRandomizer = new TopologyRandomizer(randomSupplier, prefixes, topology, topologyUpdates, nodeMap::get, schemaApply);
- List<DurabilityService> durabilityService = new ArrayList<>();
+ List<DurabilityService> durabilityServices = new ArrayList<>();
List<Service> services = new ArrayList<>();
for (Id id : nodes)
{
@@ -683,7 +687,7 @@ public class Cluster
TimeService timeService = timeServiceSupplier.get();
BiConsumer<Timestamp, Ranges> onStale = (sinceAtLeast, ranges) -> configRandomizer.onStale(id, sinceAtLeast, ranges);
AsyncExecutor nodeExecutor = nodeExecutorSupplier.apply(id);
- Agent agent = agentSupplier.apply(onStale, timeouts);
+ Agent agent = agentSupplier.apply(onStale, scheduler, timeouts);
executorMap.put(id, nodeExecutor);
Journal journal = journalFactory.apply(id, random);
journalMap.put(id, journal);
@@ -693,16 +697,16 @@ public class Cluster
() -> new ListStore(scheduler, random, id), new ShardDistributor.EvenSplit<>(8, ignore -> new PrefixedIntHashKey.Splitter()),
agent,
randomSupplier.get(), scheduler, SizeOfIntersectionSorter.SUPPLIER, DefaultRemoteListeners::new, DefaultTimeouts::new,
- DefaultProgressLogs::new, DefaultLocalListeners.Factory::new, DelayedCommandStores.factory(sinks.pending, cacheLoading), new CoordinationAdapter.DefaultFactory(),
+ TestProgressLogs::new, DefaultLocalListeners.Factory::new, DelayedCommandStores.factory(sinks.pending, cacheLoading), new CoordinationAdapter.DefaultFactory(),
journal.durableBeforePersister(), journal);
journal.start(node);
DurabilityService durability = node.durability();
// TODO (desired): randomise
durability.shards().setShardCycleTime(30, SECONDS);
durability.global().setGlobalCycleTime(180, SECONDS);
- durabilityService.add(durability);
+ durabilityServices.add(durability);
nodeMap.put(id, node);
- durabilityService.add(new DurabilityService(node));
+ durabilityServices.add(new DurabilityService(node));
}
Runnable updateDurabilityRate;
@@ -714,7 +718,7 @@ public class Cluster
int c = targetSplits.getAsInt();
int s = shardCycleTimeSeconds.getAsInt() * topologyFactory.rf;
int g = globalCycleTimeSeconds.getAsInt();
- durabilityService.forEach(d -> {
+ durabilityServices.forEach(d -> {
d.shards().setTargetShardSplits(c);
d.shards().setShardCycleTime(s, SECONDS);
d.global().setGlobalCycleTime(g, SECONDS);
@@ -722,6 +726,15 @@ public class Cluster
};
}
updateDurabilityRate.run();
+
+ Runnable updateProgressLogConcurrency;
+ {
+ updateProgressLogConcurrency = () -> {
+ nodeMap.values().forEach(node -> node.commandStores().forEachCommandStore(cs -> ((TestProgressLog)cs.unsafeProgressLog()).setMaxConcurrency(random.nextInt(1, 16))));
+ };
+ }
+ updateProgressLogConcurrency.run();
+
schemaApply.onUpdate(topology);
Pending.Global.setNoActiveOrigin();
@@ -737,6 +750,8 @@ public class Cluster
sinks.links = sinks.linkConfig.overrideLinks.apply(nodesList);
if (random.decide(0.1f))
updateDurabilityRate.run();
+ if (random.decide(0.1f))
+ updateProgressLogConcurrency.run();
}, 5L, SECONDS);
Scheduled reconfigure = clusterScheduler.recurring(configRandomizer::maybeUpdateTopology, 1, SECONDS);
@@ -757,15 +772,16 @@ public class Cluster
// Clean data and restore from snapshot
ListStore listStore = (ListStore) nodeMap.get(id).commandStores().dataStore();
NavigableMap<RoutableKey, Timestamped<int[]>> prevData = listStore.copyOfCurrentData();
- listStore.clear();
- listStore.restoreFromSnapshot();
+ Int2ObjectHashMap<NavigableMap<TxnId, Command>> beforeStores = copyCommands(stores.all());
+ listStore.clear();
// We are simulating node restart, so its remote listeners will also be gone
((DefaultRemoteListeners) nodeMap.get(id).remoteListeners()).clear();
- Int2ObjectHashMap<NavigableMap<TxnId, Command>> beforeStores = copyCommands(stores.all());
-
for (CommandStore store : stores.all())
- ((InMemoryCommandStore) store).clearForTesting();
+ {
+ InMemoryCommandStore commandStore = (InMemoryCommandStore) store;
+ commandStore.clear();
+ }
// Replay journal
Journal journal = journalMap.get(id);
@@ -781,7 +797,9 @@ public class Cluster
if (lastUpdate != null)
((DelayedCommandStores) nodeMap.get(id).commandStores()).validateShardStateForTesting(lastUpdate);
- stores = nodeMap.get(id).commandStores();
+ listStore.restore();
+ for (CommandStore store : stores.all())
+ ((ListAgent) store.agent()).restore((InMemoryCommandStore) store);
journal.replay(stores);
// Re-enable safety checks
@@ -793,12 +811,12 @@ public class Cluster
trace.debug("Done with replay.");
}, () -> random.nextInt(10, 30), SECONDS);
- durabilityService.forEach(DurabilityService::start);
+ durabilityServices.forEach(DurabilityService::start);
services.forEach(Service::start);
Runnable stop = () -> {
reconfigure.cancel();
- durabilityService.forEach(DurabilityService::stop);
+ durabilityServices.forEach(DurabilityService::stop);
purge.cancel();
restart.cancel();
services.forEach(Service::close);
@@ -1035,51 +1053,6 @@ public class Cluster
void close();
}
- private static abstract class AbstractService implements Service, Runnable
- {
- protected final Node node;
- protected final RandomSource rs;
- private Scheduled scheduled;
-
- protected AbstractService(Node node, RandomSource rs)
- {
- this.node = node;
- this.rs = rs;
- }
-
- @Override
- public void start()
- {
- Invariants.require(scheduled == null, "Start already called...");
- this.scheduled = node.scheduler().recurring(this, 1, SECONDS);
- }
-
- protected abstract void doRun() throws Exception;
-
- @Override
- public final void run()
- {
- try
- {
- doRun();
- }
- catch (Throwable t)
- {
- node.agent().onUncaughtException(t);
- }
- }
-
- @Override
- public void close()
- {
- if (scheduled != null)
- {
- scheduled.cancel();
- scheduled = null;
- }
- }
- }
-
private static BiFunction<Id, Id, Link> partition(List<Id> nodes, RandomSource random, int rf, BiFunction<Id, Id, Link> up)
{
Collections.shuffle(nodes, random.asJdkRandom());
diff --git a/accord-core/src/test/java/accord/impl/basic/InMemoryJournal.java b/accord-core/src/test/java/accord/impl/basic/InMemoryJournal.java
index 8adf14ac..5920aad7 100644
--- a/accord-core/src/test/java/accord/impl/basic/InMemoryJournal.java
+++ b/accord-core/src/test/java/accord/impl/basic/InMemoryJournal.java
@@ -605,7 +605,7 @@ public class InMemoryJournal implements Journal
Map<TxnId, List<Diff>> diffs = new TreeMap<>();
InMemoryCommandStore commandStore = (InMemoryCommandStore) commandStores.forId(commandStoreId);
- Loader loader = commandStore.loader();
+ Replayer replayer = commandStore.replayer();
for (Map.Entry<TxnId, Diffs> e : diffEntry.getValue().entrySet())
diffs.put(e.getKey(), e.getValue().sorted(true));
@@ -614,9 +614,10 @@ public class InMemoryJournal implements Journal
{
if (e.getValue().isEmpty()) continue;
- AsyncResult<?> res = loader.load(e.getKey()).beginAsResult();
+ AsyncResult<?> res = replayer.replay(e.getKey()).beginAsResult();
AsyncChains.getUnchecked(res);
}
+
}
}
diff --git a/accord-core/src/test/java/accord/impl/basic/NodeSink.java b/accord-core/src/test/java/accord/impl/basic/NodeSink.java
index b9e2e64d..deb02505 100644
--- a/accord-core/src/test/java/accord/impl/basic/NodeSink.java
+++ b/accord-core/src/test/java/accord/impl/basic/NodeSink.java
@@ -126,8 +126,6 @@ public class NodeSink implements MessageSink
case DELIVER_WITH_FAILURE:
deliver(to, id, expiresAt, message, link);
case FAILURE:
- if (action == Action.FAILURE)
- parent.notifyDropped(self, to, id, message);
if (callback != null)
{
long failesAt = timeouts.failsAt();
diff --git a/accord-core/src/test/java/accord/impl/basic/Pending.java b/accord-core/src/test/java/accord/impl/basic/Pending.java
index fd84fa88..7214d5bc 100644
--- a/accord-core/src/test/java/accord/impl/basic/Pending.java
+++ b/accord-core/src/test/java/accord/impl/basic/Pending.java
@@ -34,6 +34,7 @@ public interface Pending
{
Invariants.require(activeOrigin == null);
activeOrigin = newActive.origin();
+ Invariants.require(activeOrigin != null);
}
public static void clearActiveOrigin()
diff --git a/accord-core/src/test/java/accord/impl/basic/TaskExecutorService.java b/accord-core/src/test/java/accord/impl/basic/TaskExecutorService.java
index fafef387..97d5255d 100644
--- a/accord-core/src/test/java/accord/impl/basic/TaskExecutorService.java
+++ b/accord-core/src/test/java/accord/impl/basic/TaskExecutorService.java
@@ -101,9 +101,9 @@ public abstract class TaskExecutorService extends AbstractExecutorService implem
}
@Override
- public Task<?> build(Runnable task)
+ public Task<Void> build(Runnable task)
{
- return (Task<?>) super.submit(task);
+ return (Task<Void>) super.submit(task);
}
@Override
diff --git a/accord-core/src/test/java/accord/impl/basic/TestProgressLogs.java b/accord-core/src/test/java/accord/impl/basic/TestProgressLogs.java
new file mode 100644
index 00000000..56e76f00
--- /dev/null
+++ b/accord-core/src/test/java/accord/impl/basic/TestProgressLogs.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package accord.impl.basic;
+
+import javax.annotation.Nullable;
+
+import accord.api.ProgressLog;
+import accord.impl.progresslog.DefaultProgressLog;
+import accord.local.CommandStore;
+import accord.local.Node;
+import accord.local.SafeCommandStore;
+
+public class TestProgressLogs implements ProgressLog.Factory
+{
+ static class TestProgressLog extends DefaultProgressLog
+ {
+ protected TestProgressLog(Node node, CommandStore commandStore)
+ {
+ super(node, commandStore);
+ }
+
+ @Override
+ public void accept(@Nullable SafeCommandStore safeStore)
+ {
+ Pending prev = Pending.Global.activeOrigin();
+ Pending.Global.setNoActiveOrigin();
+ try
+ {
+ super.accept(safeStore);
+ }
+ finally
+ {
+ if (prev != Pending.Global.NONE)
+ {
+ Pending.Global.clearActiveOrigin();
+ Pending.Global.setActiveOrigin(prev);
+ }
+ }
+ }
+ }
+
+ final Node node;
+ public TestProgressLogs(Node node)
+ {
+ this.node = node;
+ }
+
+ @Override
+ public DefaultProgressLog create(CommandStore commandStore)
+ {
+ return new TestProgressLog(node, commandStore);
+ }
+}
diff --git a/accord-core/src/test/java/accord/impl/list/ListAgent.java b/accord-core/src/test/java/accord/impl/list/ListAgent.java
index a67cffa1..24fde94f 100644
--- a/accord-core/src/test/java/accord/impl/list/ListAgent.java
+++ b/accord-core/src/test/java/accord/impl/list/ListAgent.java
@@ -30,13 +30,16 @@ import java.util.function.LongSupplier;
import javax.annotation.Nullable;
-import accord.api.Agent;
import accord.api.ProgressLog;
import accord.api.Result;
+import accord.api.Scheduler;
import accord.api.TraceEventType;
import accord.api.Tracing;
import accord.coordinate.CoordinationFailed;
import accord.coordinate.ExecuteSyncPoint;
+import accord.impl.InMemoryAgent;
+import accord.impl.InMemoryCommandStore;
+import accord.impl.InMemoryCommandStore.Snapshot;
import accord.impl.basic.NodeSink;
import accord.impl.basic.Packet;
import accord.impl.basic.SimulatedFault;
@@ -60,6 +63,7 @@ import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncChains;
import accord.utils.async.AsyncResult;
import accord.utils.async.AsyncResults;
+import org.agrona.collections.Int2ObjectHashMap;
import static accord.local.Node.Id.NONE;
import static com.google.common.base.Functions.identity;
@@ -67,8 +71,9 @@ import static java.util.concurrent.TimeUnit.MICROSECONDS;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
-public class ListAgent implements Agent
+public class ListAgent implements InMemoryAgent
{
+ final Scheduler scheduler;
final RandomSource rnd;
final long timeout;
final Consumer<Throwable> onFailure;
@@ -80,9 +85,11 @@ public class ListAgent implements Agent
final LongSupplier queueTimeMillis;
final TimeService time;
final NodeSink.TimeoutSupplier timeoutSupplier;
+ final Int2ObjectHashMap<Snapshotter<Snapshot>> snapshotters = new Int2ObjectHashMap<>();
- public ListAgent(RandomSource rnd, long timeout, Consumer<Throwable> onFailure, Consumer<Runnable> retryBootstrap, BiConsumer<Timestamp, Ranges> onStale, IntSupplier coordinationDelays, IntSupplier progressDelays, IntSupplier timeoutDelays, LongSupplier queueTimeMillis, TimeService time, NodeSink.TimeoutSupplier timeoutSupplier)
+ public ListAgent(Scheduler scheduler, RandomSource rnd, long timeout, Consumer<Throwable> onFailure, Consumer<Runnable> retryBootstrap, BiConsumer<Timestamp, Ranges> onStale, IntSupplier coordinationDelays, IntSupplier progressDelays, IntSupplier timeoutDelays, LongSupplier queueTimeMillis, TimeService time, NodeSink.TimeoutSupplier timeoutSupplier)
{
+ this.scheduler = scheduler;
this.rnd = rnd;
this.timeout = timeout;
this.onFailure = onFailure;
@@ -139,17 +146,15 @@ public class ListAgent implements Agent
onStale.accept(staleSince, ranges);
}
- private static final Set<Class<?>> expectedExceptions = new HashSet<>(Arrays.asList(SimulatedFault.class, ExecuteSyncPoint.SyncPointErased.class, CancellationException.class, TopologyManager.TopologyRetiredException.class));
+ private static final Set<Class<?>> expectedExceptions = new HashSet<>(Arrays.asList(SimulatedFault.class, ExecuteSyncPoint.SyncPointErased.class, CancellationException.class, TopologyManager.TopologyRetiredException.class, Snapshotter.SnapshotAborted.class));
@Override
public void onUncaughtException(Throwable t)
{
if (expectedExceptions.contains(t.getClass()))
return;
- // TODO (required): why are we now seeing SnapshotAborted? Nothing inherently wrong with it, but should find out what has changed.
if (!(t instanceof CoordinationFailed)
- && !(t.getCause() instanceof CancellationException)
- && !(t instanceof ListStore.SnapshotAborted))
+ && !(t.getCause() instanceof CancellationException))
onFailure.accept(t);
}
@@ -272,4 +277,19 @@ public class ListAgent implements Agent
{
return unit.convert(timeoutSupplier.expiresAt(), MICROSECONDS);
}
+
+ @Override
+ public AsyncResult<Void> snapshot(InMemoryCommandStore commandStore)
+ {
+ Snapshotter<Snapshot> snapshotter = snapshotters.computeIfAbsent(commandStore.id(), ignore -> new Snapshotter<>(scheduler, rnd));
+ return snapshotter.snapshot(false, (AsyncResult<Snapshot>) Snapshot.snapshot(commandStore));
+ }
+
+ public void restore(InMemoryCommandStore commandStore)
+ {
+ Snapshotter<Snapshot> snapshotter = snapshotters.get(commandStore.id());
+ if (snapshotter == null)
+ return;
+ snapshotter.restore(snapshot -> snapshot.restore(commandStore));
+ }
}
diff --git a/accord-core/src/test/java/accord/impl/list/ListFetchCoordinator.java b/accord-core/src/test/java/accord/impl/list/ListFetchCoordinator.java
index 65f9c10a..11ab47c1 100644
--- a/accord-core/src/test/java/accord/impl/list/ListFetchCoordinator.java
+++ b/accord-core/src/test/java/accord/impl/list/ListFetchCoordinator.java
@@ -66,7 +66,7 @@ public class ListFetchCoordinator extends AbstractFetchCoordinator
ListData listData = (ListData) data;
persisting.add(commandStore.build((PreLoadContext.Empty) () -> "List Fetch", safeStore -> {
listData.forEach((key, value) -> listStore.writeUnsafe(key, value));
- }).flatMap(ignore -> listStore.snapshot(true, received, syncPoint.syncId)).invoke((success, fail) -> {
+ }).flatMap(ignore -> listStore.snapshot(true)).invoke((success, fail) -> {
if (fail == null) success(from, received);
else fail(from, received, fail);
}).beginAsResult());
diff --git a/accord-core/src/test/java/accord/impl/list/ListRequest.java b/accord-core/src/test/java/accord/impl/list/ListRequest.java
index 61ed5a65..d1d7ec13 100644
--- a/accord-core/src/test/java/accord/impl/list/ListRequest.java
+++ b/accord-core/src/test/java/accord/impl/list/ListRequest.java
@@ -177,17 +177,17 @@ public class ListRequest implements Request
}
node.reply(client, replyContext, ListResult.heartBeat(client, ((Packet)replyContext).requestId, id), null);
- node.scheduler().once(() -> checkOnResult(homeKey, id, 0, null), 5L, TimeUnit.MINUTES);
+ node.scheduler().once(() -> checkOnResult(homeKey, id, 0, null), 1L, TimeUnit.MINUTES);
}
else if (fail instanceof SimulatedFault)
{
node.reply(client, replyContext, ListResult.heartBeat(client, ((Packet)replyContext).requestId, id), null);
- node.scheduler().once(() -> checkOnResult(null, id, 0, null), 5L, TimeUnit.MINUTES);
+ node.scheduler().once(() -> checkOnResult(null, id, 0, null), 1L, TimeUnit.MINUTES);
}
else if (fail instanceof CancellationException)
{
node.reply(client, replyContext, ListResult.heartBeat(client, ((Packet)replyContext).requestId, id), null);
- node.scheduler().once(() -> checkOnResult(null, id, 0, null), 5L, TimeUnit.MINUTES);
+ node.scheduler().once(() -> checkOnResult(null, id, 0, null), 1L, TimeUnit.MINUTES);
}
else
{
@@ -236,7 +236,7 @@ public class ListRequest implements Request
if (attempt < 1000 && (f instanceof Timeout || f instanceof SimulatedFault || f instanceof Exhausted))
{
if (attempt < 100) checkOnResult(finalHomeKey, txnId, attempt + 1, f);
- else node.scheduler().once(() -> checkOnResult(finalHomeKey, txnId, attempt + 1, null), 5L, TimeUnit.MINUTES);
+ else node.scheduler().once(() -> checkOnResult(finalHomeKey, txnId, attempt + 1, null), retryDelay(attempt), TimeUnit.MINUTES);
}
else
{
@@ -263,7 +263,7 @@ public class ListRequest implements Request
if (attempt < 100)
{
node.reply(client, replyContext, ListResult.heartBeat(client, ((Packet)replyContext).requestId, txnId), null);
- node.scheduler().once(() -> checkOnResult(finalHomeKey, txnId, attempt + 1, null), 5L, TimeUnit.MINUTES);
+ node.scheduler().once(() -> checkOnResult(finalHomeKey, txnId, attempt + 1, null), retryDelay(attempt), TimeUnit.MINUTES);
}
else
{
@@ -277,6 +277,11 @@ public class ListRequest implements Request
}
}
+ private static long retryDelay(int attempts)
+ {
+ return attempts <= 1 ? 1 : Math.min(5, attempts);
+ }
+
private final String description;
private final Function<Node, Txn> gen;
private final BiFunction<Node, Txn, TxnId> txnIdGen;
diff --git a/accord-core/src/test/java/accord/impl/list/ListStore.java b/accord-core/src/test/java/accord/impl/list/ListStore.java
index bcf1cc79..0fc14153 100644
--- a/accord-core/src/test/java/accord/impl/list/ListStore.java
+++ b/accord-core/src/test/java/accord/impl/list/ListStore.java
@@ -18,16 +18,12 @@
package accord.impl.list;
-import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Deque;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.TreeMap;
-import java.util.concurrent.TimeUnit;
-import java.util.function.Consumer;
import java.util.function.Predicate;
import java.util.stream.Collectors;
@@ -38,6 +34,8 @@ import accord.api.Scheduler;
import accord.local.CommandStore;
import accord.local.CommandStores;
import accord.local.Node;
+import accord.local.PreLoadContext;
+import accord.local.RedundantBefore;
import accord.local.SafeCommandStore;
import accord.primitives.Range;
import accord.primitives.Ranges;
@@ -58,7 +56,7 @@ import org.agrona.collections.LongArrayList;
import static accord.primitives.Routables.Slice.Minimal;
import static accord.utils.Invariants.illegalState;
-public class ListStore implements DataStore, ConfigurationService.Listener
+public class ListStore extends Snapshotter<ListStore.Snapshot> implements DataStore, ConfigurationService.Listener
{
private static class ChangeAt
{
@@ -128,14 +126,10 @@ public class ListStore implements DataStore, ConfigurationService.Listener
}
}
- public static class SnapshotAborted extends RuntimeException { SnapshotAborted() { super("Snapshot aborted due to earlier snapshot being restored"); } }
-
static final boolean VERIFY_ACCESS = Boolean.getBoolean(System.getProperty("accord.test.verify_liststore_access", "true"));
static final Timestamped<int[]> EMPTY = new Timestamped<>(Timestamp.NONE, new int[0], Arrays::toString);
final NavigableMap<RoutableKey, Timestamped<int[]>> data = new TreeMap<>();
- final RandomSource random;
- final Scheduler scheduler;
private final List<ChangeAt> addedAts = new ArrayList<>();
private final List<ChangeAt> removedAts = new ArrayList<>();
@@ -147,7 +141,7 @@ public class ListStore implements DataStore, ConfigurationService.Listener
// used to make sure removes are applied in epoch order and not in the order sync points complete in
private final LongArrayList pendingRemoves = new LongArrayList();
- private static final class Snapshot
+ public static final class Snapshot
{
private final NavigableMap<RoutableKey, Timestamped<int[]>> data;
private final List<ChangeAt> addedAts;
@@ -168,96 +162,37 @@ public class ListStore implements DataStore, ConfigurationService.Listener
}
}
- private static final class PendingSnapshot
- {
- // we don't restart bootstraps after clearing journal state and replaying, so must finish snapshots
- final boolean runBeforeRestart;
- final long delay;
- final Consumer<Boolean> onCompletion;
-
- private PendingSnapshot(boolean runBeforeRestart, long delay, Consumer<Boolean> onCompletion)
- {
- this.runBeforeRestart = runBeforeRestart;
- this.delay = delay;
- this.onCompletion = onCompletion;
- }
- }
-
- private Snapshot snapshot;
private Scheduler.Scheduled scheduled;
- private final Deque<PendingSnapshot> pendingSnapshots = new ArrayDeque<>();
- private long pendingDelay = 0;
+ /**
+ * Logical fsync-like operation: anything written to the store prior to the invocation of this method
+ * must be durable once the AsyncResult completes successfully. That is, a restart of the node must
+ * restore the DataStore to a state on or after the point at which snapshot was invoked.
+ */
@Override
- public AsyncResult<Void> snapshot(Ranges ranges, TxnId before)
+ public void ensureDurable(CommandStore commandStore, Ranges ranges, RedundantBefore onSuccess)
{
- return snapshot(false, ranges, before);
+ snapshot(false).begin((success, fail) -> {
+ if (fail == null) commandStore.execute((PreLoadContext.Empty)()->"Report DataStore Durable", safeStore -> safeStore.upsertRedundantBefore(onSuccess));
+ });
}
- public AsyncResult<Void> snapshot(boolean runBeforeRestart, Ranges ranges, TxnId before)
+ public AsyncResult<Void> snapshot(boolean runBeforeRestart)
{
Snapshot snapshot = new Snapshot(data, addedAts, removedAts, purgedAts, fetchCompletes, pendingRemoves);
- AsyncResult.Settable<Void> result = new AsyncResults.SettableResult<>();
- long delay = Math.max(1, random.nextBiasedLong(100, 1000, 5000) - pendingDelay);
- pendingDelay += delay;
-
- if (scheduled != null && runBeforeRestart && !pendingSnapshots.stream().anyMatch(p -> p.runBeforeRestart))
- {
- scheduled.cancel();
- scheduled = null;
- }
-
- pendingSnapshots.add(new PendingSnapshot(runBeforeRestart, delay, success -> {
- if (success)
- {
- this.snapshot = snapshot;
- result.setSuccess(null);
- }
- else
- {
- result.setFailure(new SnapshotAborted());
- }
- }));
-
- if (scheduled == null)
- scheduleRunSnapshot();
- return result;
+ return super.snapshot(runBeforeRestart, snapshot);
}
- private void scheduleRunSnapshot()
+ public void restore()
{
- Invariants.require(!pendingSnapshots.isEmpty());
- // schedule as recurring so that we don't run them
- Runnable run = () -> {
- scheduled = null;
- if (pendingSnapshots.isEmpty())
- return;
-
- PendingSnapshot pendingSnapshot = pendingSnapshots.pollFirst();
- pendingSnapshot.onCompletion.accept(true);
- pendingDelay -= pendingSnapshot.delay;
- if (!pendingSnapshots.isEmpty())
- scheduleRunSnapshot();
- };
-
- if (pendingSnapshots.stream().anyMatch(p -> p.runBeforeRestart)) scheduled = scheduler.once(run, pendingSnapshots.peekFirst().delay, TimeUnit.MILLISECONDS);
- else scheduled = scheduler.selfRecurring(run, pendingSnapshots.peekFirst().delay, TimeUnit.MILLISECONDS);
- }
-
- public void restoreFromSnapshot()
- {
- if (snapshot != null)
- {
+ super.restore(snapshot -> {
data.putAll(snapshot.data);
addedAts.addAll(snapshot.addedAts);
removedAts.addAll(snapshot.removedAts);
purgedAts.addAll(snapshot.purgedAts);
fetchCompletes.addAll(snapshot.fetchCompletes);
pendingRemoves.addAll(snapshot.pendingRemoves);
- }
-
- while (!pendingSnapshots.isEmpty())
- pendingSnapshots.pollFirst().onCompletion.accept(false);
+ });
}
public void clear()
@@ -275,8 +210,7 @@ public class ListStore implements DataStore, ConfigurationService.Listener
public ListStore(Scheduler scheduler, RandomSource random, Node.Id node)
{
- this.random = random;
- this.scheduler = scheduler;
+ super(scheduler, random);
this.node = node;
}
diff --git a/accord-core/src/test/java/accord/impl/list/Snapshotter.java b/accord-core/src/test/java/accord/impl/list/Snapshotter.java
new file mode 100644
index 00000000..0605ff93
--- /dev/null
+++ b/accord-core/src/test/java/accord/impl/list/Snapshotter.java
@@ -0,0 +1,135 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package accord.impl.list;
+
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+
+import accord.api.Scheduler;
+import accord.utils.Invariants;
+import accord.utils.RandomSource;
+import accord.utils.async.AsyncResult;
+import accord.utils.async.AsyncResults;
+
+// TODO (expected): move this to a test package, along with InMemoryX
+public class Snapshotter<S>
+{
+ public static class SnapshotAborted extends RuntimeException { SnapshotAborted() { super("Snapshot aborted due to earlier snapshot being restored"); } }
+
+ private static final class PendingSnapshot
+ {
+ // we don't restart bootstraps after clearing journal state and replaying, so must finish snapshots
+ final boolean runBeforeRestart;
+ final long delay;
+ final Consumer<Runnable> onReady;
+
+ private PendingSnapshot(boolean runBeforeRestart, long delay, Consumer<Runnable> onReady)
+ {
+ this.runBeforeRestart = runBeforeRestart;
+ this.delay = delay;
+ this.onReady = onReady;
+ }
+ }
+
+ final RandomSource random;
+ final Scheduler scheduler;
+ private S snapshot;
+ private Scheduler.Scheduled scheduled;
+ private final Deque<PendingSnapshot> pendingSnapshots = new ArrayDeque<>();
+ private long pendingDelay = 0;
+
+ protected Snapshotter(Scheduler scheduler, RandomSource random)
+ {
+ this.random = random;
+ this.scheduler = scheduler;
+ }
+
+ protected AsyncResult<Void> snapshot(boolean runBeforeRestart, S snapshot)
+ {
+ return snapshot(runBeforeRestart, AsyncResults.success(snapshot));
+ }
+
+ protected AsyncResult<Void> snapshot(boolean runBeforeRestart, AsyncResult<S> snapshot)
+ {
+ AsyncResult.Settable<Void> result = new AsyncResults.SettableResult<>();
+ long delay = Math.max(1, random.nextBiasedLong(100, 1000, 5000) - pendingDelay);
+ pendingDelay += delay;
+
+ if (scheduled != null && runBeforeRestart && !pendingSnapshots.stream().anyMatch(p -> p.runBeforeRestart))
+ {
+ scheduled.cancel();
+ scheduled = null;
+ }
+
+ pendingSnapshots.add(new PendingSnapshot(runBeforeRestart, delay, (continuation) -> {
+ if (continuation != null)
+ {
+ snapshot.begin((success, fail) -> {
+ if (fail != null) result.setFailure(fail);
+ else
+ {
+ this.snapshot = success;
+ result.setSuccess(null);
+ }
+ continuation.run();
+ });
+ }
+ else
+ {
+ result.setFailure(new SnapshotAborted());
+ }
+ }));
+
+ if (scheduled == null)
+ scheduleRunSnapshot();
+ return result;
+ }
+
+ private void scheduleRunSnapshot()
+ {
+ Invariants.require(!pendingSnapshots.isEmpty());
+ // schedule as recurring so that we don't run them
+ Runnable run = () -> {
+ scheduled = null;
+ if (pendingSnapshots.isEmpty())
+ return;
+
+ PendingSnapshot pendingSnapshot = pendingSnapshots.pollFirst();
+ pendingDelay -= pendingSnapshot.delay;
+ pendingSnapshot.onReady.accept(() -> {
+ if (!pendingSnapshots.isEmpty())
+ scheduleRunSnapshot();
+ });
+ };
+
+ if (pendingSnapshots.stream().anyMatch(p -> p.runBeforeRestart)) scheduled = scheduler.once(run, pendingSnapshots.peekFirst().delay, TimeUnit.MILLISECONDS);
+ else scheduled = scheduler.selfRecurring(run, pendingSnapshots.peekFirst().delay, TimeUnit.MILLISECONDS);
+ }
+
+ protected void restore(Consumer<S> restore)
+ {
+ if (snapshot != null)
+ restore.accept(snapshot);
+
+ while (!pendingSnapshots.isEmpty())
+ pendingSnapshots.pollFirst().onReady.accept(null);
+ }
+}
diff --git a/accord-core/src/test/java/accord/impl/mock/MockStore.java b/accord-core/src/test/java/accord/impl/mock/MockStore.java
index b7e5740a..fa3747ab 100644
--- a/accord-core/src/test/java/accord/impl/mock/MockStore.java
+++ b/accord-core/src/test/java/accord/impl/mock/MockStore.java
@@ -25,7 +25,10 @@ import accord.api.Read;
import accord.api.Result;
import accord.api.Update;
import accord.api.Write;
+import accord.local.CommandStore;
import accord.local.Node;
+import accord.local.PreLoadContext;
+import accord.local.RedundantBefore;
import accord.local.SafeCommandStore;
import accord.primitives.Participants;
import accord.primitives.Ranges;
@@ -33,11 +36,9 @@ import accord.primitives.Seekable;
import accord.primitives.Seekables;
import accord.primitives.SyncPoint;
import accord.primitives.Timestamp;
-import accord.primitives.TxnId;
import accord.primitives.Writes;
import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncChains;
-import accord.utils.async.AsyncResult;
import accord.utils.async.AsyncResults;
public class MockStore implements DataStore
@@ -159,8 +160,8 @@ public class MockStore implements DataStore
}
@Override
- public AsyncResult<Void> snapshot(Ranges ranges, TxnId before)
+ public void ensureDurable(CommandStore commandStore, Ranges ranges, RedundantBefore reportOnSuccess)
{
- return AsyncResults.success(null);
+ commandStore.execute((PreLoadContext.Empty)() -> "Report CommandStore Durable", safeStore -> safeStore.upsertRedundantBefore(reportOnSuccess));
}
}
diff --git a/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java b/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java
index 3639376a..17cd3c25 100644
--- a/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java
+++ b/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java
@@ -990,11 +990,13 @@ public class CommandsForKeyTest
return true;
}
- public Journal.Loader loader()
+ public Journal.Replayer replayer()
{
throw new UnsupportedOperationException();
}
+ @Override protected void ensureDurable(Ranges ranges, RedundantBefore onDataStoreDurable) {}
+
@Override
public AsyncChain<Void> build(PreLoadContext context, Consumer<? super SafeCommandStore> consumer)
{
diff --git a/accord-maelstrom/src/main/java/accord/maelstrom/MaelstromStore.java b/accord-maelstrom/src/main/java/accord/maelstrom/MaelstromStore.java
index 777700cc..e5ed170f 100644
--- a/accord-maelstrom/src/main/java/accord/maelstrom/MaelstromStore.java
+++ b/accord-maelstrom/src/main/java/accord/maelstrom/MaelstromStore.java
@@ -23,14 +23,14 @@ import java.util.concurrent.ConcurrentHashMap;
import accord.api.Key;
import accord.api.DataStore;
+import accord.local.CommandStore;
import accord.local.Node;
+import accord.local.PreLoadContext;
+import accord.local.RedundantBefore;
import accord.local.SafeCommandStore;
import accord.primitives.Ranges;
import accord.primitives.SyncPoint;
-import accord.primitives.TxnId;
import accord.utils.Timestamped;
-import accord.utils.async.AsyncResult;
-import accord.utils.async.AsyncResults;
import accord.utils.async.AsyncResults.SettableResult;
public class MaelstromStore implements DataStore
@@ -62,8 +62,7 @@ public class MaelstromStore implements DataStore
}
@Override
- public AsyncResult<Void> snapshot(Ranges ranges, TxnId before)
+ public void ensureDurable(CommandStore commandStore, Ranges ranges, RedundantBefore reportOnSuccess)
{
- return AsyncResults.success(null);
- }
-}
+ commandStore.execute((PreLoadContext.Empty)() -> "Report CommandStore Durable", safeStore -> safeStore.upsertRedundantBefore(reportOnSuccess));
+ }}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org