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/17 06:17:24 UTC
(cassandra-accord) branch trunk updated: PreLoadContext descriptions Also: - Introduce LoadKeysFor so we can avoid loading range transactions except where necessary
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
The following commit(s) were added to refs/heads/trunk by this push:
new a1c1ed91 PreLoadContext descriptions Also: - Introduce LoadKeysFor so we can avoid loading range transactions except where necessary
a1c1ed91 is described below
commit a1c1ed91cfeb904be9b490d498c1c52aba2253c6
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Wed Jun 25 14:45:11 2025 +0100
PreLoadContext descriptions
Also:
- Introduce LoadKeysFor so we can avoid loading range transactions except where necessary
patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20758
---
.../java/accord/api/CoordinatorEventListener.java | 83 +++++++++++
.../accord/coordinate/CoordinateTransaction.java | 20 ++-
.../main/java/accord/coordinate/ExecuteTxn.java | 6 +
.../main/java/accord/coordinate/FetchRoute.java | 3 +-
.../src/main/java/accord/coordinate/Infer.java | 18 ++-
.../main/java/accord/coordinate/Invalidate.java | 2 +-
.../main/java/accord/coordinate/KeyBarriers.java | 4 +-
.../src/main/java/accord/coordinate/Stabilise.java | 3 +-
.../src/main/java/accord/impl/AbstractLoader.java | 3 +-
.../java/accord/impl/AbstractSafeCommandStore.java | 20 ++-
.../java/accord/impl/DefaultLocalListeners.java | 19 ++-
.../java/accord/impl/InMemoryCommandStore.java | 26 ++--
.../accord/impl/progresslog/CallbackInvoker.java | 20 ++-
.../impl/progresslog/DefaultProgressLog.java | 14 +-
.../impl/progresslog/DefaultProgressLogs.java | 5 +-
.../java/accord/impl/progresslog/TxnState.java | 16 ++-
.../src/main/java/accord/local/Bootstrap.java | 11 +-
.../src/main/java/accord/local/CommandStore.java | 13 +-
.../src/main/java/accord/local/CommandStores.java | 24 +---
.../main/java/accord/local/CommandSummaries.java | 15 +-
.../src/main/java/accord/local/Commands.java | 17 ++-
.../local/{KeyHistory.java => LoadKeys.java} | 21 +--
.../src/main/java/accord/local/LoadKeysFor.java | 49 +++++++
accord-core/src/main/java/accord/local/Node.java | 6 +-
.../src/main/java/accord/local/PreLoadContext.java | 154 +++++++++------------
.../main/java/accord/local/SafeCommandStore.java | 17 +--
.../src/main/java/accord/local/cfk/NotifySink.java | 12 +-
.../main/java/accord/local/cfk/PostProcess.java | 5 +-
.../src/main/java/accord/local/cfk/Updating.java | 5 +-
.../main/java/accord/messages/AbstractRequest.java | 7 +
.../src/main/java/accord/messages/Accept.java | 19 ++-
.../src/main/java/accord/messages/Apply.java | 6 +-
.../src/main/java/accord/messages/Await.java | 12 ++
.../java/accord/messages/BeginInvalidation.java | 6 +
.../main/java/accord/messages/BeginRecovery.java | 16 ++-
.../src/main/java/accord/messages/Commit.java | 12 +-
.../java/accord/messages/GetDurableBefore.java | 6 +
.../java/accord/messages/GetEphemeralReadDeps.java | 13 +-
.../main/java/accord/messages/GetLatestDeps.java | 13 +-
.../main/java/accord/messages/GetMaxConflict.java | 12 +-
.../main/java/accord/messages/InformDurable.java | 8 +-
.../src/main/java/accord/messages/PreAccept.java | 13 +-
.../src/main/java/accord/messages/Propagate.java | 6 +
.../src/main/java/accord/messages/ReadData.java | 6 +
.../java/accord/messages/ReadEphemeralTxnData.java | 8 +-
.../src/main/java/accord/messages/ReadTxnData.java | 6 +
.../java/accord/messages/SetGloballyDurable.java | 6 +
.../main/java/accord/messages/StableThenRead.java | 6 +-
.../src/main/java/accord/primitives/TxnId.java | 10 +-
accord-core/src/main/java/accord/utils/Reduce.java | 2 +-
.../main/java/accord/utils/async/AsyncChains.java | 18 +--
.../java/accord/coordinate/TopologyChangeTest.java | 6 +-
.../accord/impl/list/ListFetchCoordinator.java | 2 +-
.../java/accord/local/ImmutableCommandTest.java | 6 +-
.../test/java/accord/messages/PreAcceptTest.java | 7 +-
.../test/java/accord/messages/ReadDataTest.java | 10 +-
56 files changed, 580 insertions(+), 273 deletions(-)
diff --git a/accord-core/src/main/java/accord/api/CoordinatorEventListener.java b/accord-core/src/main/java/accord/api/CoordinatorEventListener.java
new file mode 100644
index 00000000..e44a7073
--- /dev/null
+++ b/accord-core/src/main/java/accord/api/CoordinatorEventListener.java
@@ -0,0 +1,83 @@
+/*
+ * 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.api;
+
+import javax.annotation.Nullable;
+
+import accord.coordinate.ExecutePath;
+import accord.local.Node;
+import accord.primitives.Ballot;
+import accord.primitives.Deps;
+import accord.primitives.Status.Durability;
+import accord.primitives.TxnId;
+
+// TODO (required): revisit the call-sites, and boundary with C*
+public interface CoordinatorEventListener
+{
+ default void onPreAccepted(TxnId txnId, Deps deps, boolean isStable)
+ {
+ }
+
+ default void onAccepted(TxnId txnId, Ballot ballot, Deps deps, boolean isStable)
+ {
+ }
+
+ default void onStabilised(TxnId txnId, Ballot ballot, Deps deps)
+ {
+ }
+
+ default void onExecuting(TxnId txnId, @Nullable Ballot ballot, @Nullable ExecutePath path)
+ {
+ }
+
+ default void onExecuted(TxnId txnId, Ballot ballot)
+ {
+ }
+
+ default void onDurable(Durability durability, @Nullable Ballot ballot, TxnId txnId)
+ {
+ }
+
+ default void onRecoveryStarted(TxnId txnId, Ballot ballot)
+ {
+ }
+
+ /**
+ * For use by implementations to decide what to do about successfully recovered transactions.
+ * Specifically intended to define if and how they should inform clients of the result.
+ * e.g. in Maelstrom we send the full result directly, in other impls we may simply acknowledge success via the coordinator
+ *
+ * Note: may be invoked multiple times in different places
+ */
+ default void onRecoveryStopped(Node node, TxnId txnId, Ballot ballot, Result success, Throwable fail)
+ {
+ }
+
+ default void onInvalidated(TxnId txnId)
+ {
+ }
+
+ default void onRejected(TxnId txnId)
+ {
+ }
+
+ CoordinatorEventListener NOOP = new CoordinatorEventListener()
+ {
+ };
+}
diff --git a/accord-core/src/main/java/accord/coordinate/CoordinateTransaction.java b/accord-core/src/main/java/accord/coordinate/CoordinateTransaction.java
index 2f57baa2..c8963918 100644
--- a/accord-core/src/main/java/accord/coordinate/CoordinateTransaction.java
+++ b/accord-core/src/main/java/accord/coordinate/CoordinateTransaction.java
@@ -32,7 +32,8 @@ import accord.coordinate.ExecuteFlag.ExecuteFlags;
import accord.local.Command;
import accord.local.Commands;
import accord.local.DepsCalculator;
-import accord.local.KeyHistory;
+import accord.local.LoadKeys;
+import accord.local.LoadKeysFor;
import accord.local.PreLoadContext;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
@@ -244,7 +245,6 @@ public class CoordinateTransaction extends CoordinatePreAccept<Result>
setFailure(new Preempted(txnId, route.homeKey()));
}
}
-
}
@Override
@@ -346,9 +346,21 @@ public class CoordinateTransaction extends CoordinatePreAccept<Result>
}
@Override
- public KeyHistory keyHistory()
+ public String reason()
+ {
+ return "Local PreAccept";
+ }
+
+ @Override
+ public LoadKeys loadKeys()
+ {
+ return LoadKeys.SYNC;
+ }
+
+ @Override
+ public LoadKeysFor loadKeysFor()
{
- return KeyHistory.SYNC;
+ return LoadKeysFor.READ_WRITE;
}
@Override
diff --git a/accord-core/src/main/java/accord/coordinate/ExecuteTxn.java b/accord-core/src/main/java/accord/coordinate/ExecuteTxn.java
index e8bb33a4..9c352de4 100644
--- a/accord-core/src/main/java/accord/coordinate/ExecuteTxn.java
+++ b/accord-core/src/main/java/accord/coordinate/ExecuteTxn.java
@@ -350,5 +350,11 @@ public class ExecuteTxn extends ReadCoordinator<ReadReply>
@Override
public MessageType type() { throw new UnsupportedOperationException(); }
+
+ @Override
+ public String reason()
+ {
+ return "Local Execute";
+ }
}
}
diff --git a/accord-core/src/main/java/accord/coordinate/FetchRoute.java b/accord-core/src/main/java/accord/coordinate/FetchRoute.java
index e7bd8daa..7a4f149b 100644
--- a/accord-core/src/main/java/accord/coordinate/FetchRoute.java
+++ b/accord-core/src/main/java/accord/coordinate/FetchRoute.java
@@ -25,6 +25,7 @@ import javax.annotation.Nullable;
import accord.api.Tracing;
import accord.local.Commands;
import accord.local.Node;
+import accord.local.PreLoadContext;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.primitives.Known;
@@ -100,7 +101,7 @@ public class FetchRoute extends CheckShards<Participants<?>>
else
{
StoreSelector selector = reportTo.refine(txnId, null, query);
- node.mapReduceConsumeLocal(txnId, selector, new MapReduceConsume<>()
+ node.mapReduceConsumeLocal(PreLoadContext.contextFor(txnId, "Report Route"), selector, new MapReduceConsume<>()
{
@Override
public void accept(Object result, Throwable failure)
diff --git a/accord-core/src/main/java/accord/coordinate/Infer.java b/accord-core/src/main/java/accord/coordinate/Infer.java
index 71ee91fb..97e09874 100644
--- a/accord-core/src/main/java/accord/coordinate/Infer.java
+++ b/accord-core/src/main/java/accord/coordinate/Infer.java
@@ -28,6 +28,7 @@ import accord.local.CommandStores.StoreFinder;
import accord.local.CommandStores.StoreSelector;
import accord.local.Commands;
import accord.local.Node;
+import accord.local.PreLoadContext;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.primitives.Status;
@@ -104,7 +105,7 @@ public class Infer
}
// TODO (required): audit all use cases
- private static abstract class CleanupAndCallback<T> implements MapReduceConsume<SafeCommandStore, Void>
+ private static abstract class CleanupAndCallback<T> implements MapReduceConsume<SafeCommandStore, Void>, PreLoadContext
{
final Node node;
final TxnId txnId;
@@ -129,7 +130,7 @@ public class Infer
void start()
{
- node.mapReduceConsumeLocal(txnId, reportTo.refine(txnId, null, participants), this);
+ node.mapReduceConsumeLocal(this, reportTo.refine(txnId, null, participants), this);
}
@Override
@@ -153,6 +154,13 @@ public class Infer
{
callback.accept(param, failure);
}
+
+ @Nullable
+ @Override
+ public TxnId primaryTxnId()
+ {
+ return txnId;
+ }
}
static class InvalidateAndCallback<T> extends CleanupAndCallback<T>
@@ -183,5 +191,11 @@ public class Infer
Commands.commitInvalidate(safeStore, safeCommand, participants);
return null;
}
+
+ @Override
+ public String reason()
+ {
+ return "Invalidate " + txnId + " with " + participants;
+ }
}
}
diff --git a/accord-core/src/main/java/accord/coordinate/Invalidate.java b/accord-core/src/main/java/accord/coordinate/Invalidate.java
index b2f0137e..72886aa5 100644
--- a/accord-core/src/main/java/accord/coordinate/Invalidate.java
+++ b/accord-core/src/main/java/accord/coordinate/Invalidate.java
@@ -303,7 +303,7 @@ public class Invalidate implements Callback<InvalidateReply>
{
// TODO (desired): merge with FetchData.InvalidateOnDone
// TODO (desired): when sending to network, register a callback for when local application of commitInvalidate message ahs been performed, so no need to special-case
- node.forEachLocal(txnId, reportTo.refine(txnId, null, commitTo), safeStore -> {
+ node.forEachLocal(PreLoadContext.contextFor(txnId, "Local Commit Invalidate"), reportTo.refine(txnId, null, commitTo), safeStore -> {
// TODO (expected): consid
StoreParticipants participants = StoreParticipants.notAccept(safeStore, commitTo, txnId);
Commands.commitInvalidate(safeStore, safeStore.get(txnId, participants), commitTo);
diff --git a/accord-core/src/main/java/accord/coordinate/KeyBarriers.java b/accord-core/src/main/java/accord/coordinate/KeyBarriers.java
index 0c8f6065..54ba8cda 100644
--- a/accord-core/src/main/java/accord/coordinate/KeyBarriers.java
+++ b/accord-core/src/main/java/accord/coordinate/KeyBarriers.java
@@ -24,6 +24,7 @@ import org.slf4j.LoggerFactory;
import accord.api.RoutingKey;
import accord.local.CommandSummaries;
import accord.local.Node;
+import accord.local.PreLoadContext;
import accord.local.SafeCommandStore;
import accord.local.SequentialAsyncExecutor;
import accord.local.durability.DurabilityService.SyncLocal;
@@ -49,7 +50,6 @@ import static accord.local.CommandSummaries.SummaryStatus.COMMITTED;
import static accord.local.CommandSummaries.SummaryStatus.INVALIDATED;
import static accord.local.CommandSummaries.ComputeIsDep.IGNORE;
import static accord.local.CommandSummaries.TestStartedAt.STARTED_AFTER;
-import static accord.local.PreLoadContext.contextFor;
import static accord.local.durability.DurabilityService.SyncLocal.NoLocal;
import static accord.local.durability.DurabilityService.SyncLocal.Self;
import static accord.local.durability.DurabilityService.SyncRemote.NoRemote;
@@ -86,7 +86,7 @@ public class KeyBarriers
public static AsyncChain<Found> find(Node node, Timestamp min, RoutingKey key, SyncLocal syncLocal, SyncRemote syncRemote)
{
- return node.commandStores().mapReduce(contextFor(key), RoutingKeys.of(key), min.epoch(), Long.MAX_VALUE, new Find(min, key, syncLocal, syncRemote));
+ return node.commandStores().mapReduce((PreLoadContext.Empty) () -> "Key Barrier", RoutingKeys.of(key), min.epoch(), Long.MAX_VALUE, new Find(min, key, syncLocal, syncRemote));
}
/*
diff --git a/accord-core/src/main/java/accord/coordinate/Stabilise.java b/accord-core/src/main/java/accord/coordinate/Stabilise.java
index 77136cbd..ce72c59b 100644
--- a/accord-core/src/main/java/accord/coordinate/Stabilise.java
+++ b/accord-core/src/main/java/accord/coordinate/Stabilise.java
@@ -42,7 +42,6 @@ import accord.utils.SortedArrays.SortedArrayList;
import accord.utils.SortedListMap;
import accord.utils.UnhandledEnum;
-import static accord.coordinate.ExecutePath.RECOVER;
import static accord.coordinate.ExecutePath.SLOW;
import static accord.coordinate.tracking.RequestStatus.Failed;
import static accord.messages.Commit.Kind.CommitWithTxn;
@@ -161,7 +160,7 @@ public abstract class Stabilise<R> implements Callback<ReadReply>
protected void onStabilised()
{
- adapter().execute(node, executor, allTopologies, route, ballot, ballot == Ballot.ZERO ? SLOW : RECOVER, CoordinationFlags.none(), txnId, txn, executeAt, stabiliseDeps, stabiliseDeps, callback);
+ adapter().execute(node, executor, allTopologies, route, ballot, SLOW, CoordinationFlags.none(), txnId, txn, executeAt, stabiliseDeps, stabiliseDeps, callback);
}
protected abstract CoordinationAdapter<R> adapter();
diff --git a/accord-core/src/main/java/accord/impl/AbstractLoader.java b/accord-core/src/main/java/accord/impl/AbstractLoader.java
index 864ccd90..a003bb2a 100644
--- a/accord-core/src/main/java/accord/impl/AbstractLoader.java
+++ b/accord-core/src/main/java/accord/impl/AbstractLoader.java
@@ -22,6 +22,7 @@ import accord.api.Journal;
import accord.local.Command;
import accord.local.CommandStore;
import accord.local.Commands;
+import accord.local.PreLoadContext;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.primitives.Participants;
@@ -55,7 +56,7 @@ public abstract class AbstractLoader implements Journal.Loader
{
command.writes()
.apply(safeStore, executes, command.partialTxn())
- .invoke(() -> unsafeStore.build(txnId, ss -> {
+ .invoke(() -> unsafeStore.build(PreLoadContext.contextFor(txnId, "Replay"), ss -> {
Commands.postApply(ss, txnId, -1, true);
}))
.begin(safeStore.agent());
diff --git a/accord-core/src/main/java/accord/impl/AbstractSafeCommandStore.java b/accord-core/src/main/java/accord/impl/AbstractSafeCommandStore.java
index 0e9a1ebb..84fa5b67 100644
--- a/accord-core/src/main/java/accord/impl/AbstractSafeCommandStore.java
+++ b/accord-core/src/main/java/accord/impl/AbstractSafeCommandStore.java
@@ -24,7 +24,7 @@ import java.util.NavigableMap;
import accord.api.RoutingKey;
import accord.local.CommandStore;
-import accord.local.KeyHistory;
+import accord.local.LoadKeys;
import accord.local.PreLoadContext;
import accord.local.RedundantBefore;
import accord.local.SafeCommand;
@@ -81,8 +81,16 @@ extends SafeCommandStore
if (with.keys().domain() == Routable.Domain.Range)
return with.isSubsetOf(this.context) ? with : null;
- if (!context().keyHistory().satisfiesIfPresent(with.keyHistory()))
- return null;
+ LoadKeys require = with.loadKeys();
+ if (require != LoadKeys.NONE)
+ {
+ PreLoadContext context = context();
+ if (!context.loadKeys().satisfiesIfPresent(require))
+ return null;
+
+ if (with.loadKeysFor().compareTo(context.loadKeysFor()) > 0)
+ return null;
+ }
try (Caches caches = tryGetCaches())
{
@@ -101,8 +109,8 @@ extends SafeCommandStore
add(safeCommand, caches);
}
- KeyHistory keyHistory = with.keyHistory();
- if (keyHistory == KeyHistory.NONE)
+ LoadKeys loadKeys = with.loadKeys();
+ if (loadKeys == LoadKeys.NONE)
return with;
List<RoutingKey> unavailable = null;
@@ -136,7 +144,7 @@ extends SafeCommandStore
if (unavailable.size() == keys.size())
return null;
- return PreLoadContext.contextFor(with.primaryTxnId(), with.additionalTxnId(), keys.without(RoutingKeys.ofSortedUnique(unavailable)), keyHistory);
+ return PreLoadContext.contextFor(with.primaryTxnId(), with.additionalTxnId(), keys.without(RoutingKeys.ofSortedUnique(unavailable)), loadKeys, context.loadKeysFor(), context.reason());
}
}
diff --git a/accord-core/src/main/java/accord/impl/DefaultLocalListeners.java b/accord-core/src/main/java/accord/impl/DefaultLocalListeners.java
index 97fca67a..a4c5d60f 100644
--- a/accord-core/src/main/java/accord/impl/DefaultLocalListeners.java
+++ b/accord-core/src/main/java/accord/impl/DefaultLocalListeners.java
@@ -23,6 +23,8 @@ import java.util.EnumMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.BiFunction;
+import javax.annotation.Nullable;
+
import accord.api.LocalListeners;
import accord.api.RemoteListeners;
import accord.local.Command;
@@ -95,7 +97,7 @@ public class DefaultLocalListeners implements LocalListeners
//noinspection SillyAssignment,ConstantConditions
safeStore = safeStore; // prevent use in lambda
TxnId updatedId = safeCommand.txnId();
- PreLoadContext context = PreLoadContext.contextFor(listenerId, updatedId);
+ PreLoadContext context = PreLoadContext.contextFor(listenerId, updatedId, "Notify");
safeStore.commandStore().execute(context, safeStore0 -> notify(safeStore0, listenerId, updatedId), safeStore.agent());
}
}
@@ -115,7 +117,7 @@ public class DefaultLocalListeners implements LocalListeners
/*
* A list that allows duplicates and sorts and removes duplicates on notify and when the list would have to resize
*/
- static class TxnListeners extends TxnId
+ static class TxnListeners extends TxnId implements PreLoadContext
{
final SaveStatus await;
TxnId[] listeners = NO_TXNIDS;
@@ -203,6 +205,19 @@ public class DefaultLocalListeners implements LocalListeners
listeners[count++] = listener;
}
+
+ @Nullable
+ @Override
+ public TxnId primaryTxnId()
+ {
+ return this;
+ }
+
+ @Override
+ public String reason()
+ {
+ return "Notify Listeners";
+ }
}
class RegisteredComplexListener implements Registered, BiFunction<TxnId, RegisteredComplexListeners, RegisteredComplexListeners>
diff --git a/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java b/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
index 857c98f5..2a754a67 100644
--- a/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
+++ b/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
@@ -56,7 +56,7 @@ import accord.local.CommandStore;
import accord.local.CommandStores.RangesForEpoch;
import accord.local.CommandSummaries;
import accord.local.Commands;
-import accord.local.KeyHistory;
+import accord.local.LoadKeys;
import accord.local.NodeCommandStoreService;
import accord.local.PreLoadContext;
import accord.local.RejectBefore;
@@ -88,8 +88,10 @@ import accord.utils.async.Cancellable;
import org.agrona.collections.ObjectHashSet;
import static accord.local.Cleanup.Input.FULL;
-import static accord.local.KeyHistory.ASYNC;
-import static accord.local.KeyHistory.NONE;
+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;
import static accord.primitives.Known.KnownRoute.MaybeRoute;
@@ -344,17 +346,16 @@ public abstract class InMemoryCommandStore extends CommandStore
{
case Key:
RoutableKey key = (RoutableKey) unseekable;
- switch (context.keyHistory())
+ switch (context.loadKeys())
{
case NONE:
continue;
case INCR:
case SYNC:
case ASYNC:
- case RECOVER:
commandsForKey.put(key, commandsForKey((RoutingKey) key).createSafeReference());
break;
- default: throw new UnsupportedOperationException("Unknown key history: " + context.keyHistory());
+ default: throw new UnsupportedOperationException("Unknown key history: " + context.loadKeys());
}
break;
case Range:
@@ -634,7 +635,7 @@ public abstract class InMemoryCommandStore extends CommandStore
@Override
protected InMemorySafeCommandsForKey ifLoadedInternal(RoutingKey key)
{
- if (context.keyHistory() != NONE && context.keys().domain() == Range && context.keys().contains(key))
+ if (context.loadKeys() != NONE && context.keys().domain() == Range && context.keys().contains(key))
{
GlobalCommandsForKey globalCfk = commandStore().commandsForKey.get(key);
if (globalCfk == null)
@@ -734,7 +735,8 @@ public abstract class InMemoryCommandStore extends CommandStore
if (commandsForRanges != null)
return commandsForRanges;
- Summary.Loader loader = Summary.Loader.loader(redundantBefore(), context.primaryTxnId(), context.keyHistory(), context.keys());
+ Invariants.require(context.loadKeysFor() != WRITE);
+ Summary.Loader loader = Summary.Loader.loader(redundantBefore(), context.primaryTxnId(), context.loadKeysFor(), context.keys());
TreeMap<Timestamp, Summary> summaries = new TreeMap<>();
for (RangeCommand rangeCommand : commandStore().rangeCommands.values())
{
@@ -1151,7 +1153,7 @@ public abstract class InMemoryCommandStore extends CommandStore
this.commandStore = commandStore;
}
- private PreLoadContext context(Command command, KeyHistory keyHistory)
+ private PreLoadContext context(Command command, LoadKeys loadKeys)
{
TxnId txnId = command.txnId();
AbstractUnseekableKeys keys = null;
@@ -1162,9 +1164,9 @@ public abstract class InMemoryCommandStore extends CommandStore
keys = command.asCommitted().waitingOn.keys;
if (keys != null)
- return PreLoadContext.contextFor(txnId, keys, keyHistory);
+ return PreLoadContext.contextFor(txnId, keys, loadKeys, WRITE, "Replay");
- return txnId;
+ return PreLoadContext.contextFor(txnId, "Replay");
}
protected TxnId loadInternal(Command command, SafeCommandStore safeStore)
@@ -1188,7 +1190,7 @@ public abstract class InMemoryCommandStore extends CommandStore
private AsyncChain<Void> apply(TxnId txnId)
{
return AsyncChains.success(commandStore.executeInContext(commandStore,
- txnId,
+ PreLoadContext.contextFor(txnId, "Replay"),
(SafeCommandStore safeStore) -> {
initialiseState(safeStore, txnId);
return null;
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 c1c155e0..895a23d5 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/CallbackInvoker.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/CallbackInvoker.java
@@ -20,13 +20,16 @@ package accord.impl.progresslog;
import java.util.function.BiConsumer;
+import javax.annotation.Nullable;
+
+import accord.local.PreLoadContext;
import accord.local.SafeCommand;
import accord.primitives.TxnId;
import static accord.impl.progresslog.TxnStateKind.Home;
import static accord.impl.progresslog.TxnStateKind.Waiting;
-class CallbackInvoker<P, V> implements BiConsumer<V, Throwable>
+class CallbackInvoker<P, V> implements BiConsumer<V, Throwable>, PreLoadContext
{
static <P, V> CallbackInvoker<P, V> invokeWaitingCallback(DefaultProgressLog instance, TxnId txnId, P param, Callback<P, V> callback)
{
@@ -70,7 +73,7 @@ class CallbackInvoker<P, V> implements BiConsumer<V, Throwable>
@Override
public void accept(V success, Throwable fail)
{
- owner.commandStore.execute(txnId, safeStore -> {
+ owner.commandStore.execute(this, safeStore -> {
// we load safeCommand first so that if it clears the progress log we abandon the callback
SafeCommand safeCommand = safeStore.ifInitialised(txnId);
@@ -106,5 +109,18 @@ class CallbackInvoker<P, V> implements BiConsumer<V, Throwable>
{
return txnId + (isHome ? ":Home:" : ":Waiting:") + owner.commandStore;
}
+
+ @Nullable
+ @Override
+ public TxnId primaryTxnId()
+ {
+ return txnId;
+ }
+
+ @Override
+ public String reason()
+ {
+ return "Callback " + this;
+ }
}
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 f349e8e0..8c3346f2 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java
@@ -36,6 +36,7 @@ import accord.api.RoutingKey;
import accord.local.Command;
import accord.local.CommandStore;
import accord.local.Node;
+import accord.local.PreLoadContext;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.primitives.SaveStatus;
@@ -66,7 +67,6 @@ import static accord.impl.progresslog.Progress.Queued;
import static accord.impl.progresslog.TxnStateKind.Home;
import static accord.impl.progresslog.TxnStateKind.Waiting;
import static accord.local.Command.NotDefined.uninitialised;
-import static accord.local.PreLoadContext.empty;
import static accord.primitives.Routables.Slice.Minimal;
import static accord.primitives.Status.PreApplied;
import static accord.primitives.Status.PreCommitted;
@@ -333,7 +333,7 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
{
// the command might be invalidated, which should be established on load, so simply load the command
TxnId txnId = state.txnId;
- safeStore.commandStore().execute(txnId, safeStore0 -> {
+ safeStore.commandStore().execute(PreLoadContext.contextFor(txnId, "Clear Progress"), safeStore0 -> {
safeStore0.unsafeGet(txnId);
}, node.agent());
}
@@ -512,7 +512,7 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
minEpoch = Math.min(awaitingEpochBuffer[i].run.txnId.epoch(), minEpoch);
Invariants.requireArgument(minEpoch != Long.MAX_VALUE);
isAwaitingEpoch = true;
- node.withEpochAtLeast(minEpoch, commandStore, (success, fail) -> commandStore.execute(empty(), ss -> {
+ node.withEpochAtLeast(minEpoch, commandStore, (success, fail) -> commandStore.execute((PreLoadContext.Empty) () -> "Run ProgressLog", ss -> {
isAwaitingEpoch = false;
accept(ss);
}, node.agent()));
@@ -597,7 +597,7 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
{
while (i < readyToRun.size())
{
- commandStore.execute(empty(), readyToRun.get(i++), node.agent());
+ commandStore.execute((PreLoadContext.Empty) () -> "Run ProgressLog", readyToRun.get(i++), node.agent());
}
}
}
@@ -610,14 +610,14 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
void addIfReadyElseSubmit(@Nullable SafeCommandStore safeStore, RunInvoker invoker, List<RunInvoker> ifReady)
{
- if (safeStore != null && safeStore.canExecuteWith(invoker.run.txnId))
+ 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.txnId, invoker, commandStore.agent());
+ commandStore.execute(invoker.run, invoker, commandStore.agent());
}
}
@@ -763,7 +763,7 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
{
long now = node.elapsed(MICROSECONDS);
if (timers.shouldWake(now))
- commandStore.execute(empty(), this, node.agent());
+ commandStore.execute((PreLoadContext.Empty) () -> "Run ProgressLog", this, node.agent());
}
}
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 5453f19f..cafe2c1c 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLogs.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLogs.java
@@ -25,8 +25,7 @@ import java.util.concurrent.CopyOnWriteArrayList;
import accord.api.ProgressLog;
import accord.local.CommandStore;
import accord.local.Node;
-
-import static accord.local.PreLoadContext.empty;
+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
@@ -56,7 +55,7 @@ public class DefaultProgressLogs implements ProgressLog.Factory
return false;
if (!paused.containsKey(progressLog))
- paused.putIfAbsent(progressLog, () -> progressLog.commandStore.execute(empty(), progressLog, progressLog.commandStore.agent()));
+ paused.putIfAbsent(progressLog, () -> progressLog.commandStore.execute((PreLoadContext.Empty) () -> "Unpause", progressLog, progressLog.commandStore.agent()));
return true;
}
diff --git a/accord-core/src/main/java/accord/impl/progresslog/TxnState.java b/accord-core/src/main/java/accord/impl/progresslog/TxnState.java
index 2ed92399..044a9327 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/TxnState.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/TxnState.java
@@ -23,6 +23,7 @@ import javax.annotation.Nullable;
import com.google.common.primitives.Ints;
import accord.api.ProgressLog.BlockedUntil;
+import accord.local.PreLoadContext;
import accord.local.SafeCommandStore;
import accord.primitives.TxnId;
import accord.utils.Invariants;
@@ -30,7 +31,7 @@ import accord.utils.UnhandledEnum;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
-final class TxnState extends HomeState
+final class TxnState extends HomeState implements PreLoadContext
{
TxnState(TxnId txnId)
{
@@ -163,4 +164,17 @@ final class TxnState extends HomeState
{
return runKind == TxnStateKind.Home ? isHomeDone() : isWaitingDone();
}
+
+ @Nullable
+ @Override
+ public TxnId primaryTxnId()
+ {
+ return txnId;
+ }
+
+ @Override
+ public String reason()
+ {
+ return "Progress";
+ }
}
diff --git a/accord-core/src/main/java/accord/local/Bootstrap.java b/accord-core/src/main/java/accord/local/Bootstrap.java
index 8743d5ad..bf335167 100644
--- a/accord-core/src/main/java/accord/local/Bootstrap.java
+++ b/accord-core/src/main/java/accord/local/Bootstrap.java
@@ -40,7 +40,6 @@ import accord.utils.ReducingRangeMap;
import accord.utils.async.AsyncResult;
import accord.utils.async.AsyncResults;
-import static accord.local.PreLoadContext.empty;
import static accord.primitives.Routables.Slice.Minimal;
import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
import static accord.utils.Invariants.illegalState;
@@ -125,7 +124,7 @@ class Bootstrap
if (!node.topology().hasEpoch(globalSyncId.epoch()))
{
// Ignore timeouts fetching the epoch, always keep trying to bootstrap
- node.withEpochAtLeast(globalSyncId.epoch(), null, (ignored, failure) -> store.execute(empty(), Attempt.this::start, (ignored1, failure2) -> {
+ node.withEpochAtLeast(globalSyncId.epoch(), null, (ignored, failure) -> store.execute((PreLoadContext.Empty) () -> "Start Bootstrap", Attempt.this::start, (ignored1, failure2) -> {
if (failure2 != null)
node.agent().acceptAndWrap(null, failure2);
}));
@@ -138,12 +137,12 @@ class Bootstrap
safeStore = safeStore;
// we submit a separate execution so that we know markBootstrapping is durable before we initiate the fetch
safeStore.commandStore()
- .build(empty(), safeStore0 -> {
+ .build((PreLoadContext.Empty) () -> "Start Bootstrap RX", safeStore0 -> {
store.markBootstrapping(safeStore0, globalSyncId, commitRanges);
return CoordinateSyncPoint.exclusive(node, globalSyncId, commitRanges);
})
.flatMap(i -> i)
- .flatMap(syncPoint -> node.withEpochAtLeast(epoch, null, () -> store.build(empty(), safeStore1 -> {
+ .flatMap(syncPoint -> node.withEpochAtLeast(epoch, null, () -> store.build((PreLoadContext.Empty) () -> "Start Bootstrap Fetch", safeStore1 -> {
if (valid.isEmpty()) // we've lost ownership of the range
return AsyncResults.success(Ranges.EMPTY);
return fetch = safeStore1.dataStore().fetch(node, safeStore1, valid, syncPoint, this);
@@ -274,7 +273,7 @@ class Bootstrap
store.agent().onFailedBootstrap(attempt, "PartialFetch", newFailures, () -> {
node.scheduler().selfRecurring(() -> {
- store.execute(empty(), safeStore -> restart(safeStore, newFailures.slice(allValid), attempt + 1), store.agent());
+ store.execute((PreLoadContext.Empty) () -> "Restart Bootstrap", safeStore -> restart(safeStore, newFailures.slice(allValid), attempt + 1), store.agent());
}, 0L, TimeUnit.NANOSECONDS);
}, failure);
Invariants.require(!newFailures.intersects(fetchedAndSafeToRead));
@@ -350,7 +349,7 @@ class Bootstrap
{
store.agent().onFailedBootstrap(attempt, "Fetch", retry, () -> {
node.scheduler().selfRecurring(() -> {
- store.execute(empty(), safeStore -> restart(safeStore, retry, attempt + 1), node.agent());
+ store.execute((PreLoadContext.Empty) () -> "Restart Bootstrap", safeStore -> restart(safeStore, retry, attempt + 1), node.agent());
}, 0L, TimeUnit.NANOSECONDS);
}, fetchOutcome);
}
diff --git a/accord-core/src/main/java/accord/local/CommandStore.java b/accord-core/src/main/java/accord/local/CommandStore.java
index e189753c..1b25a180 100644
--- a/accord-core/src/main/java/accord/local/CommandStore.java
+++ b/accord-core/src/main/java/accord/local/CommandStore.java
@@ -62,7 +62,6 @@ import org.agrona.collections.LongHashSet;
import static accord.api.ConfigurationService.EpochReady.DONE;
import static accord.api.ProtocolModifiers.Toggles.requiresUniqueHlcs;
-import static accord.local.PreLoadContext.empty;
import static accord.local.RedundantStatus.SomeStatus.GC_BEFORE_AND_LOCALLY_APPLIED;
import static accord.local.RedundantStatus.SomeStatus.LOCALLY_APPLIED_ONLY;
import static accord.local.RedundantStatus.SomeStatus.LOCALLY_WITNESSED_ONLY;
@@ -529,7 +528,7 @@ public abstract class CommandStore implements SequentialAsyncExecutor
final Supplier<EpochReady> bootstrapper(Node node, Ranges newRanges, long epoch)
{
return () -> {
- AsyncResult<EpochReady> metadata = submit(empty(), safeStore -> {
+ AsyncResult<EpochReady> metadata = submit((PreLoadContext.Empty) () -> "New Epoch", safeStore -> {
Bootstrap bootstrap = new Bootstrap(node, this, epoch, newRanges);
bootstraps.add(bootstrap);
bootstrap.start(safeStore);
@@ -591,7 +590,7 @@ public abstract class CommandStore implements SequentialAsyncExecutor
Supplier<EpochReady> unbootstrap(long epoch, Ranges removedRanges)
{
return () -> {
- AsyncResult<Void> done = submit(empty(), safeStore -> {
+ AsyncResult<Void> done = submit((PreLoadContext.Empty) () -> "Unbootstrap", safeStore -> {
for (Bootstrap prev : bootstraps)
{
Ranges abort = prev.allValid.slice(removedRanges, Minimal);
@@ -658,7 +657,7 @@ public abstract class CommandStore implements SequentialAsyncExecutor
return;
}
- execute(PreLoadContext.empty(), safeStore0 -> {
+ execute((PreLoadContext.Empty) () -> "Mark Shard Durable", safeStore0 -> {
RedundantBefore addGc = RedundantBefore.create(slicedRanges, globalSyncId, GC_BEFORE_AND_LOCALLY_APPLIED);
safeStore0.upsertRedundantBefore(addGc);
}, agent());
@@ -744,7 +743,7 @@ public abstract class CommandStore implements SequentialAsyncExecutor
Supplier<EpochReady> initialise(long epoch, Ranges ranges)
{
return () -> {
- AsyncResult<Void> done = execute(empty(), (safeStore) -> {
+ AsyncResult<Void> done = execute((PreLoadContext.Empty) () -> "Initialise New Epoch", (safeStore) -> {
// Merge in a base for any ranges that needs to be covered
Ranges newBootstrapRanges = ranges;
for (Ranges existing : bootstrapBeganAt.values())
@@ -802,7 +801,7 @@ public abstract class CommandStore implements SequentialAsyncExecutor
{
if (safeToRead.values().stream().anyMatch(r -> r.intersects(ranges)))
{
- execute(empty(), safeStore -> {
+ execute((PreLoadContext.Empty) () -> "Mark Unsafe To Read", safeStore -> {
safeStore.setSafeToRead(purgeHistory(safeToRead, ranges));
}, agent);
}
@@ -815,7 +814,7 @@ public abstract class CommandStore implements SequentialAsyncExecutor
final synchronized void markSafeToRead(Timestamp forBootstrapAt, Timestamp at, Ranges ranges)
{
- execute(empty(), safeStore -> {
+ execute((PreLoadContext.Empty) () -> "Mark Safe To Read", safeStore -> {
// TODO (required): handle weird edge cases like newer at having a lower HLC than prior existing at, but higher epoch
Ranges validatedSafeToRead = redundantBefore.validateSafeToRead(forBootstrapAt, ranges);
safeStore.setSafeToRead(purgeAndInsert(safeToRead, at, validatedSafeToRead));
diff --git a/accord-core/src/main/java/accord/local/CommandStores.java b/accord-core/src/main/java/accord/local/CommandStores.java
index 779c1091..80f697e1 100644
--- a/accord-core/src/main/java/accord/local/CommandStores.java
+++ b/accord-core/src/main/java/accord/local/CommandStores.java
@@ -81,7 +81,6 @@ import org.agrona.collections.Int2IntHashMap;
import org.agrona.collections.Int2ObjectHashMap;
import static accord.api.ConfigurationService.EpochReady.done;
-import static accord.local.PreLoadContext.empty;
import static accord.primitives.Routables.Slice.Minimal;
import static accord.utils.Invariants.illegalState;
import static java.util.stream.Collectors.toList;
@@ -813,17 +812,13 @@ public abstract class CommandStores implements AsyncExecutorFactory
return initial;
}
- public AsyncChain<Void> forEach(Consumer<SafeCommandStore> forEach)
+ public AsyncChain<Void> forEach(PreLoadContext context, Consumer<SafeCommandStore> forEach)
{
List<AsyncChain<Void>> list = new ArrayList<>();
Snapshot snapshot = current;
for (ShardHolder shard : snapshot.shards)
- list.add(shard.store.build(empty(), forEach));
-
- if (list.isEmpty())
- return AsyncResults.success(null);
-
- return AsyncChains.reduce(list, Reduce.toNull());
+ list.add(shard.store.build(context, forEach));
+ return AsyncChains.reduce(list, Reduce.toNull(), null);
}
public void forEachCommandStore(Consumer<CommandStore> forEach)
@@ -933,18 +928,18 @@ public abstract class CommandStores implements AsyncExecutorFactory
return mapReduce(CommandStore::build, context, mapReduce, mapReduce, unseekables, minEpoch, maxEpoch);
}
- public <O> Cancellable mapReduceConsume(Unseekables<?> unseekables, long minEpoch, long maxEpoch, Function<? super CommandStore, AsyncChain<O>> map, Reduce<? super O, ? extends O> reduce, BiConsumer<? super O, Throwable> consume)
+ public <O> Cancellable mapReduceConsume(Unseekables<?> unseekables, long minEpoch, long maxEpoch, Function<? super CommandStore, AsyncChain<O>> map, Reduce<O, O> reduce, BiConsumer<? super O, Throwable> consume)
{
AsyncChain<O> reduced = mapReduce(unseekables, minEpoch, maxEpoch, map, reduce);
return reduced.begin(consume);
}
- public <O> AsyncChain<O> mapReduce(Unseekables<?> unseekables, long minEpoch, long maxEpoch, Function<? super CommandStore, AsyncChain<O>> map, Reduce<? super O, ? extends O> reducer)
+ public <O> AsyncChain<O> mapReduce(Unseekables<?> unseekables, long minEpoch, long maxEpoch, Function<? super CommandStore, AsyncChain<O>> map, Reduce<O, O> reducer)
{
return mapReduce((commandStore, i, f) -> f.apply(commandStore), null, map, reducer, unseekables, minEpoch, maxEpoch);
}
- public <O, P1, P2> AsyncChain<O> mapReduce(TriFunction<CommandStore, P1, P2, AsyncChain<O>> applyMap, P1 p1, P2 p2, Reduce<? super O, ? extends O> reducer, Unseekables<?> unseekables, long minEpoch, long maxEpoch)
+ public <O, P1, P2> AsyncChain<O> mapReduce(TriFunction<CommandStore, P1, P2, AsyncChain<O>> applyMap, P1 p1, P2 p2, Reduce<O, O> reducer, Unseekables<?> unseekables, long minEpoch, long maxEpoch)
{
return mapReduce(StoreFinder.selector(unseekables, minEpoch, maxEpoch), applyMap, p1, p2, reducer);
}
@@ -954,7 +949,7 @@ public abstract class CommandStores implements AsyncExecutorFactory
return mapReduce(selector, CommandStore::build, context, mapReduce, mapReduce);
}
- public <O, P1, P2> AsyncChain<O> mapReduce(StoreSelector selector, TriFunction<CommandStore, P1, P2, AsyncChain<O>> applyMap, P1 p1, P2 p2, Reduce<? super O, ? extends O> reducer)
+ public <O, P1, P2> AsyncChain<O> mapReduce(StoreSelector selector, TriFunction<CommandStore, P1, P2, AsyncChain<O>> applyMap, P1 p1, P2 p2, Reduce<O, O> reducer)
{
Snapshot snapshot = current;
Iterator<CommandStore> stores = selector.select(snapshot);
@@ -992,11 +987,6 @@ public abstract class CommandStores implements AsyncExecutorFactory
}
- public <O> AsyncChain<List<O>> map(Function<? super SafeCommandStore, O> mapper)
- {
- return map(PreLoadContext.empty(), mapper);
- }
-
public <O> AsyncChain<List<O>> map(PreLoadContext context, Function<? super SafeCommandStore, O> mapper)
{
ShardHolder[] shards = current.shards;
diff --git a/accord-core/src/main/java/accord/local/CommandSummaries.java b/accord-core/src/main/java/accord/local/CommandSummaries.java
index f9724155..2295ae49 100644
--- a/accord-core/src/main/java/accord/local/CommandSummaries.java
+++ b/accord-core/src/main/java/accord/local/CommandSummaries.java
@@ -39,6 +39,7 @@ import accord.utils.Invariants;
import accord.utils.UnhandledEnum;
import static accord.local.CommandSummaries.SummaryStatus.ACCEPTED;
+import static accord.local.LoadKeysFor.RECOVERY;
import static accord.primitives.Routables.Slice.Minimal;
import static accord.primitives.Status.Durability.NotDurable;
import static accord.primitives.Txn.Kind.AnyGloballyVisible;
@@ -120,20 +121,20 @@ public interface CommandSummaries
// TODO (expected): provide executeAt to PreLoadContext so we can more aggressively filter what we load, esp. by Kind
public static Loader loader(RedundantBefore redundantBefore, PreLoadContext context)
{
- return loader(redundantBefore, context.primaryTxnId(), context.keyHistory(), context.keys());
+ return loader(redundantBefore, context.primaryTxnId(), context.loadKeysFor(), context.keys());
}
- public static Loader loader(RedundantBefore redundantBefore, @Nullable TxnId primaryTxnId, KeyHistory keyHistory, Unseekables<?> keysOrRanges)
+ public static Loader loader(RedundantBefore redundantBefore, @Nullable TxnId primaryTxnId, LoadKeysFor loadKeysFor, Unseekables<?> keysOrRanges)
{
- return loader(redundantBefore, primaryTxnId, keyHistory, keysOrRanges, Loader::new);
+ return loader(redundantBefore, primaryTxnId, loadKeysFor, keysOrRanges, Loader::new);
}
- public static <L extends Loader> L loader(RedundantBefore redundantBefore, @Nullable TxnId primaryTxnId, KeyHistory keyHistory, Unseekables<?> keysOrRanges, Factory<L> factory)
+ public static <L extends Loader> L loader(RedundantBefore redundantBefore, @Nullable TxnId primaryTxnId, LoadKeysFor loadKeysFor, Unseekables<?> keysOrRanges, Factory<L> factory)
{
TxnId minTxnId = redundantBefore.min(keysOrRanges, Bounds::gcBefore);
- Timestamp maxTxnId = primaryTxnId == null || keyHistory == KeyHistory.RECOVER || !primaryTxnId.is(ExclusiveSyncPoint) ? Timestamp.MAX : primaryTxnId;
- TxnId findAsDep = primaryTxnId != null && keyHistory == KeyHistory.RECOVER ? primaryTxnId : null;
- Kinds kinds = primaryTxnId == null ? AnyGloballyVisible : primaryTxnId.witnesses().or(keyHistory == KeyHistory.RECOVER ? primaryTxnId.witnessedBy() : Nothing);
+ Timestamp maxTxnId = primaryTxnId == null || loadKeysFor == RECOVERY || !primaryTxnId.is(ExclusiveSyncPoint) ? Timestamp.MAX : primaryTxnId;
+ TxnId findAsDep = primaryTxnId != null && loadKeysFor == RECOVERY ? primaryTxnId : null;
+ Kinds kinds = primaryTxnId == null ? AnyGloballyVisible : primaryTxnId.witnesses().or(loadKeysFor == RECOVERY ? primaryTxnId.witnessedBy() : Nothing);
return factory.create(primaryTxnId, keysOrRanges, redundantBefore, kinds, minTxnId, maxTxnId, findAsDep);
}
diff --git a/accord-core/src/main/java/accord/local/Commands.java b/accord-core/src/main/java/accord/local/Commands.java
index 72bcd3d5..94f33821 100644
--- a/accord-core/src/main/java/accord/local/Commands.java
+++ b/accord-core/src/main/java/accord/local/Commands.java
@@ -77,8 +77,9 @@ import static accord.local.Commands.Validated.UPDATE_TXN_IGNORE_DEPS;
import static accord.local.Commands.Validated.UPDATE_TXN_KEEP_DEPS;
import static accord.local.Commands.Validated.UPDATE_TXN_AND_DEPS;
import static accord.local.Commands.Validated.UPDATE_TXN_MERGE_DEPS;
-import static accord.local.KeyHistory.INCR;
-import static accord.local.KeyHistory.SYNC;
+import static accord.local.LoadKeys.INCR;
+import static accord.local.LoadKeys.SYNC;
+import static accord.local.LoadKeysFor.WRITE;
import static accord.local.PreLoadContext.contextFor;
import static accord.local.RedundantStatus.Property.LOCALLY_APPLIED;
import static accord.local.RedundantStatus.Property.LOCALLY_DEFUNCT;
@@ -655,7 +656,7 @@ public class Commands
Participants<?> executes = command.participants().stillExecutes(); // including any keys we aren't writing
return command.writes().apply(safeStore, executes, command.partialTxn())
// TODO (expected): once we guarantee execution order KeyHistory can be ASYNC
- .flatMap(unused -> unsafeStore.build(contextFor(txnId, executes, SYNC), ss -> {
+ .flatMap(unused -> unsafeStore.build(contextFor(txnId, executes, SYNC, WRITE, "Post Apply"), ss -> {
postApply(ss, txnId, t0, false);
return null;
}));
@@ -1080,7 +1081,7 @@ public class Commands
TxnId txnId = command.txnId();
if (dependencyElision() == IF_DURABLE && CommandsForKey.manages(txnId))
{
- PreLoadContext context = PreLoadContext.contextFor(updated.participants().touches(), INCR);
+ PreLoadContext context = PreLoadContext.contextFor(updated.participants().touches(), INCR, WRITE, "Set Durable");
PreLoadContext execute = safeStore.canExecute(context);
if (execute != null)
{
@@ -1089,7 +1090,7 @@ public class Commands
if (execute != context)
{
if (execute != null)
- context = contextFor(context.keys().without(execute.keys()), INCR);
+ context = contextFor(context.keys().without(execute.keys()), INCR, WRITE, "Set Durable");
Invariants.require(!context.keys().isEmpty());
safeStore = safeStore; // prevent accidental usage inside lambda
@@ -1299,6 +1300,12 @@ public class Commands
return waitingId;
}
+ @Override
+ public String reason()
+ {
+ return "NotifyWaitingOn{waiting=" + waitingId + ",on=" + loadDepId + '}';
+ }
+
@Override
public TxnId additionalTxnId()
{
diff --git a/accord-core/src/main/java/accord/local/KeyHistory.java b/accord-core/src/main/java/accord/local/LoadKeys.java
similarity index 78%
rename from accord-core/src/main/java/accord/local/KeyHistory.java
rename to accord-core/src/main/java/accord/local/LoadKeys.java
index 90ffd934..be1263aa 100644
--- a/accord-core/src/main/java/accord/local/KeyHistory.java
+++ b/accord-core/src/main/java/accord/local/LoadKeys.java
@@ -21,10 +21,10 @@ package accord.local;
import accord.utils.UnhandledEnum;
/**
- * For operations that need information on historical operations, this indicates the
- * amount of data needed.
+ * For operations that need information associated with keys, this indicates the
+ * amount of data needed, and whether it is needed immediately or asynchronously.
*/
-public enum KeyHistory
+public enum LoadKeys
{
NONE,
@@ -44,32 +44,25 @@ public enum KeyHistory
/**
* Load all keys into memory before processing the command.
*/
- SYNC,
+ SYNC;
- /**
- * Load recovery information for all keys into memory before processing the command.
- */
- RECOVER;
-
- public boolean satisfiesIfPresent(KeyHistory that)
+ public boolean satisfiesIfPresent(LoadKeys that)
{
return satisfies(that, ASYNC);
}
- public boolean satisfies(KeyHistory that)
+ public boolean satisfies(LoadKeys that)
{
return satisfies(that, SYNC);
}
- private boolean satisfies(KeyHistory that, KeyHistory ifSyncRequireAtLeast)
+ private boolean satisfies(LoadKeys that, LoadKeys ifSyncRequireAtLeast)
{
switch (that)
{
default: throw new UnhandledEnum(that);
case NONE:
return true;
- case RECOVER:
- return this == that;
case ASYNC:
case INCR:
case SYNC:
diff --git a/accord-core/src/main/java/accord/local/LoadKeysFor.java b/accord-core/src/main/java/accord/local/LoadKeysFor.java
new file mode 100644
index 00000000..1e4ba7f0
--- /dev/null
+++ b/accord-core/src/main/java/accord/local/LoadKeysFor.java
@@ -0,0 +1,49 @@
+/*
+ * 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.local;
+
+/**
+ * For operations that need information associated with keys or ranges,
+ * this indicates whether the data will be queried, updated, or both.
+ */
+public enum LoadKeysFor
+{
+ /**
+ * WRITE covers only updating the relevant key state for the primaryTxnId,
+ * that is for key transactions this means updating key summaries, and for
+ * range transactions this means updating any range summaries.
+ * Importantly, this does not mean range transactions must be able to
+ * synchronously (or otherwise) write to all intersecting key summaries.
+ */
+ WRITE,
+
+ /**
+ * READ covers all intersecting summaries of relevant key or range transactions that might be
+ * witnessed by the primaryTxnId, for any commands that should be witnessed by primaryTxnId.
+ *
+ * This means range transactions MUST be able to consult all intersecting key summaries.
+ */
+ READ_WRITE,
+
+ /**
+ * RECOVERY is READ_WRITE + summary information of keys/transactions that should have witnessed
+ * the primaryTxnId.
+ */
+ RECOVERY
+}
diff --git a/accord-core/src/main/java/accord/local/Node.java b/accord-core/src/main/java/accord/local/Node.java
index 792e4623..400a0c0c 100644
--- a/accord-core/src/main/java/accord/local/Node.java
+++ b/accord-core/src/main/java/accord/local/Node.java
@@ -574,7 +574,7 @@ public class Node implements ConfigurationService.Listener, NodeCommandStoreServ
return commandStores.mapReduceConsume(request, request.scope(), minEpoch, maxEpoch, mapReduceConsume);
}
- public <T> Cancellable mapReduceConsumeLocal(Unseekables<?> keys, long minEpoch, long maxEpoch, Function<? super CommandStore, AsyncChain<T>> map, Reduce<? super T, ? extends T> reduce, BiConsumer<? super T, Throwable> consume)
+ public <T> Cancellable mapReduceConsumeLocal(Unseekables<?> keys, long minEpoch, long maxEpoch, Function<? super CommandStore, AsyncChain<T>> map, Reduce<T, T> reduce, BiConsumer<? super T, Throwable> consume)
{
return commandStores.mapReduceConsume(keys, minEpoch, maxEpoch, map, reduce, consume);
}
@@ -925,8 +925,8 @@ public class Node implements ConfigurationService.Listener, NodeCommandStoreServ
public void updateMinHlc(long minHlc)
{
- commandStores().forEach(safeStore -> {
- safeStore.commandStore().updateMinHlc(minHlc);
+ commandStores().forEach((commandStore, ranges) -> {
+ commandStore.updateMinHlc(minHlc);
});
}
diff --git a/accord-core/src/main/java/accord/local/PreLoadContext.java b/accord-core/src/main/java/accord/local/PreLoadContext.java
index acfe2305..5c4e6415 100644
--- a/accord-core/src/main/java/accord/local/PreLoadContext.java
+++ b/accord-core/src/main/java/accord/local/PreLoadContext.java
@@ -32,8 +32,10 @@ import java.util.List;
import java.util.function.Consumer;
import javax.annotation.Nullable;
-import static accord.local.KeyHistory.NONE;
-import static accord.local.KeyHistory.SYNC;
+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;
/**
* Lists txnids and keys of commands and commands for key that will be needed for an operation. Used
@@ -42,6 +44,7 @@ import static accord.local.KeyHistory.SYNC;
public interface PreLoadContext
{
@Nullable TxnId primaryTxnId();
+ String reason();
/**
* @return ids of the {@link Command} objects that need to be loaded into memory before this operation is run
@@ -90,7 +93,8 @@ public interface PreLoadContext
*/
default Unseekables<?> keys() { return RoutingKeys.EMPTY; }
- default KeyHistory keyHistory() { return NONE; }
+ default LoadKeys loadKeys() { return NONE; }
+ default LoadKeysFor loadKeysFor() { return WRITE; }
default boolean isEmpty()
{
@@ -98,7 +102,7 @@ public interface PreLoadContext
Invariants.require(additionalTxnId() == null);
return isEmpty;
}
-
+
/**
* Is the provided PreLoadContext guaranteed to have a superset of our requested information?
* Note that for this calculation we are asking if all the information we want is known to be available,
@@ -110,12 +114,14 @@ public interface PreLoadContext
Unseekables<?> keys = keys();
if (!keys.isEmpty())
{
- KeyHistory requiredHistory = keyHistory();
+ LoadKeys requiredHistory = loadKeys();
if (requiredHistory != NONE)
{
if (requiredHistory.compareTo(SYNC) < 0)
requiredHistory = SYNC;
- if (requiredHistory.compareTo(superset.keyHistory()) < 0)
+ if (requiredHistory.compareTo(superset.loadKeys()) < 0)
+ return false;
+ if (loadKeysFor().compareTo(superset.loadKeysFor()) > 0)
return false;
}
@@ -139,10 +145,26 @@ public interface PreLoadContext
}
}
- static PreLoadContext contextFor(@Nullable TxnId primary, @Nullable TxnId additional, Unseekables<?> keys, KeyHistory keyHistory)
+ default String describe()
+ {
+ List<TxnId> txnIds = txnIds();
+ Unseekables<?> keys = keys();
+ return reason() + (txnIds.isEmpty() ? "" : " for " + txnIds) + (keys.isEmpty() ? "" : (txnIds.isEmpty() ? " for " : " and ") + keys());
+ }
+
+ static PreLoadContext contextFor(@Nullable TxnId primary, @Nullable TxnId additional, Unseekables<?> keys, LoadKeys loadKeys, LoadKeysFor loadKeysFor, String reason)
{
Invariants.require(primary == null ? additional == null : !primary.equals(additional));
- return new Standard(primary, additional, keys, keyHistory);
+ return new PreLoadContext()
+ {
+ @Override public @Nullable TxnId primaryTxnId() { return primary; }
+ @Override public @Nullable TxnId additionalTxnId() { return additional; }
+ @Override public Unseekables<?> keys() { return keys; }
+ @Override public LoadKeys loadKeys() { return loadKeys; }
+ @Override public LoadKeysFor loadKeysFor() { return loadKeysFor; }
+ @Override public String reason() { return reason; }
+ @Override public String toString() { return describe(); }
+ };
}
default boolean contains(TxnId txnId)
@@ -151,100 +173,60 @@ public interface PreLoadContext
return primaryTxnId != null && (txnId.equals(primaryTxnId) || txnId.equals(additionalTxnId()));
}
-
- class Standard implements PreLoadContext
+ static PreLoadContext contextFor(TxnId primary, TxnId additional, String reason)
{
- private final @Nullable TxnId primary;
- private final @Nullable TxnId additional;
- private final Unseekables<?> keys;
- private final KeyHistory keyHistory;
-
- public Standard(@Nullable TxnId primary, @Nullable TxnId additional, Unseekables<?> keys, KeyHistory keyHistory)
- {
- Invariants.require(primary != null || additional == null);
- this.primary = primary;
- this.additional = additional;
- this.keys = keys;
- this.keyHistory = keyHistory;
- }
-
- public String toString()
- {
- return "PreLoadContext{" +
- "primary=" + primary +
- ", additional=" + additional +
- ", keys=" + keys +
- ", keyHistory=" + keyHistory +
- '}';
- }
-
- @Override
- @Nullable
- public TxnId primaryTxnId()
- {
- return primary;
- }
-
- @Override
- public TxnId additionalTxnId()
- {
- return additional;
- }
-
- @Override
- public Unseekables<?> keys() { return keys; }
-
- @Override
- public KeyHistory keyHistory()
+ return new PreLoadContext()
{
- return keyHistory;
- }
- }
-
- static PreLoadContext contextFor(TxnId primary, TxnId additional, Unseekables<?> keys)
- {
- return contextFor(primary, additional, keys, NONE);
- }
-
- static PreLoadContext contextFor(TxnId primary, TxnId additional)
- {
- return contextFor(primary, additional, RoutingKeys.EMPTY);
- }
-
- static PreLoadContext contextFor(TxnId txnId, Unseekables<?> keysOrRanges, KeyHistory keyHistory)
- {
- return contextFor(txnId, null, keysOrRanges, keyHistory);
- }
-
- static PreLoadContext contextFor(TxnId txnId, Unseekables<?> keysOrRanges)
- {
- return contextFor(txnId, keysOrRanges, NONE);
+ @Override public @Nullable TxnId primaryTxnId() { return primary; }
+ @Override public @Nullable TxnId additionalTxnId() { return additional; }
+ @Override public String reason() { return reason; }
+ @Override public String toString() { return describe(); }
+ };
}
- static PreLoadContext contextFor(RoutingKey key, KeyHistory keyHistory)
+ static PreLoadContext contextFor(TxnId primary, String reason)
{
- return contextFor(null, null, RoutingKeys.of(key), keyHistory);
+ return new PreLoadContext()
+ {
+ @Override public @Nullable TxnId primaryTxnId() { return primary; }
+ @Override public String reason() { return reason; }
+ @Override public String toString() { return describe(); }
+ };
}
- static PreLoadContext contextFor(RoutingKey key)
+ static PreLoadContext contextFor(TxnId txnId, Unseekables<?> keys, LoadKeys loadKeys, LoadKeysFor loadKeysFor, String reason)
{
- return contextFor(key, NONE);
+ return new PreLoadContext()
+ {
+ @Override public @Nullable TxnId primaryTxnId() { return txnId; }
+ @Override public Unseekables<?> keys() { return keys; }
+ @Override public LoadKeys loadKeys() { return loadKeys; }
+ @Override public LoadKeysFor loadKeysFor() { return loadKeysFor; }
+ @Override public String reason() { return reason; }
+ @Override public String toString() { return describe(); }
+ };
}
- static PreLoadContext contextFor(Unseekables<?> keys)
+ static PreLoadContext contextFor(RoutingKey key, LoadKeys loadKeys, LoadKeysFor loadKeysFor, String describe)
{
- return contextFor(null, null, keys);
+ return contextFor(RoutingKeys.of(key), loadKeys, loadKeysFor, describe);
}
- static PreLoadContext contextFor(Unseekables<?> keys, KeyHistory keyHistory)
+ static PreLoadContext contextFor(Unseekables<?> keys, LoadKeys loadKeys, LoadKeysFor loadKeysFor, String reason)
{
- return contextFor(null, null, keys, keyHistory);
+ return new PreLoadContext()
+ {
+ @Override public @Nullable TxnId primaryTxnId() { return null; }
+ @Override public Unseekables<?> keys() { return keys; }
+ @Override public LoadKeys loadKeys() { return loadKeys; }
+ @Override public LoadKeysFor loadKeysFor() { return loadKeysFor; }
+ @Override public String reason() { return reason; }
+ @Override public String toString() { return describe(); }
+ };
}
- static PreLoadContext empty()
+ interface Empty extends PreLoadContext
{
- return EMPTY_PRELOADCONTEXT;
+ @Override default @Nullable TxnId primaryTxnId() { return null; }
}
-
- PreLoadContext EMPTY_PRELOADCONTEXT = contextFor(null, null, RoutingKeys.EMPTY);
}
diff --git a/accord-core/src/main/java/accord/local/SafeCommandStore.java b/accord-core/src/main/java/accord/local/SafeCommandStore.java
index 2526899b..814a4d94 100644
--- a/accord-core/src/main/java/accord/local/SafeCommandStore.java
+++ b/accord-core/src/main/java/accord/local/SafeCommandStore.java
@@ -51,8 +51,9 @@ import accord.utils.SimpleBitSet;
import accord.utils.SortedList;
import accord.utils.async.AsyncChain;
-import static accord.local.KeyHistory.INCR;
-import static accord.local.KeyHistory.NONE;
+import static accord.local.LoadKeys.INCR;
+import static accord.local.LoadKeys.NONE;
+import static accord.local.LoadKeysFor.WRITE;
import static accord.local.RedundantStatus.SomeStatus.LOCALLY_WITNESSED_ONLY;
import static accord.local.RedundantStatus.Property.LOCALLY_REDUNDANT;
import static accord.local.RedundantStatus.Property.SHARD_APPLIED;
@@ -230,7 +231,7 @@ public abstract class SafeCommandStore implements RangesForEpochSupplier, Redund
if (safeCfk != null)
return maybeCleanup(safeCfk);
- if (context().keyHistory() != NONE && context().keys().contains(key)) throw illegalState("%s was specified in %s but was not returned by getInternal(key)", key, context().keys());
+ if (context().loadKeys() != NONE && context().keys().contains(key)) throw illegalState("%s was specified in %s but was not returned by getInternal(key)", key, context().keys());
else throw illegalArgument("%s was not specified in %s", key, context());
}
@@ -382,7 +383,7 @@ public abstract class SafeCommandStore implements RangesForEpochSupplier, Redund
return;
// TODO (expected): we don't want to insert any dependencies for those we only touch; we just need to record them as decided/applied for execution
- PreLoadContext context = PreLoadContext.contextFor(next.txnId(), update, INCR);
+ PreLoadContext context = PreLoadContext.contextFor(next.txnId(), update, INCR, WRITE, "Update CommandsForKey");
PreLoadContext execute = safeStore.canExecute(context);
if (execute != null)
{
@@ -391,7 +392,7 @@ public abstract class SafeCommandStore implements RangesForEpochSupplier, Redund
if (execute != context)
{
if (execute != null)
- context = PreLoadContext.contextFor(next.txnId(), update.without(execute.keys()), INCR);
+ context = PreLoadContext.contextFor(next.txnId(), update.without(execute.keys()), INCR, WRITE, "Update CommandsForKey");
Invariants.require(!context.keys().isEmpty());
safeStore = safeStore; // prevent accidental usage inside lambda
@@ -459,7 +460,7 @@ public abstract class SafeCommandStore implements RangesForEpochSupplier, Redund
}
// TODO (required): use StoreParticipants.executes()
// TODO (required): consider how execution works for transactions that await future deps and where the command store inherits additional keys in execution epoch
- PreLoadContext context = PreLoadContext.contextFor(txnId, keys, INCR);
+ PreLoadContext context = PreLoadContext.contextFor(txnId, keys, INCR, WRITE, "Update Unmanaged CommandsForKey");
PreLoadContext execute = safeStore.canExecute(context);
// TODO (expected): execute immediately for any keys we already have loaded, and save only those we haven't for later
if (execute != null)
@@ -474,13 +475,13 @@ public abstract class SafeCommandStore implements RangesForEpochSupplier, Redund
else
{
if (execute != null)
- context = PreLoadContext.contextFor(txnId, keys.without(execute.keys()), INCR);
+ context = PreLoadContext.contextFor(txnId, keys.without(execute.keys()), INCR, WRITE, "Update Unmanaged CommandsForKey");
safeStore = safeStore;
CommandStore unsafeStore = safeStore.commandStore();
AsyncChain<Void> submit = unsafeStore.build(context, safeStore0 -> { updateUnmanagedCommandsForKey(safeStore0, safeStore0.context().keys() , txnId, mode); });
if (next.txnId().is(Range))
- submit = submit.flatMap(success -> unsafeStore.build(PreLoadContext.empty(), safeStore0 -> { registerTransitive(safeStore0, txnId, next); }));
+ submit = submit.flatMap(success -> unsafeStore.build((PreLoadContext.Empty) () -> "Register Transitive Dependencies", safeStore0 -> { registerTransitive(safeStore0, txnId, next); }));
submit.begin(safeStore.commandStore().agent);
}
}
diff --git a/accord-core/src/main/java/accord/local/cfk/NotifySink.java b/accord-core/src/main/java/accord/local/cfk/NotifySink.java
index 8a4df5a6..cfd0f9e0 100644
--- a/accord-core/src/main/java/accord/local/cfk/NotifySink.java
+++ b/accord-core/src/main/java/accord/local/cfk/NotifySink.java
@@ -34,7 +34,8 @@ import accord.primitives.TxnId;
import accord.utils.Invariants;
import static accord.local.Command.NotDefined.uninitialised;
-import static accord.local.KeyHistory.SYNC;
+import static accord.local.LoadKeys.SYNC;
+import static accord.local.LoadKeysFor.WRITE;
import static accord.primitives.Status.Truncated;
interface NotifySink
@@ -58,7 +59,7 @@ interface NotifySink
}
else
{
- safeStore.commandStore().execute(txnId, safeStore0 -> {
+ safeStore.commandStore().execute(PreLoadContext.contextFor(txnId, "Notify"), safeStore0 -> {
notWaiting(safeStore0, safeStore0.unsafeGet(txnId), key, uniqueHlc);
}, safeStore.agent());
}
@@ -74,12 +75,13 @@ interface NotifySink
{
TxnId txnId = notify.plainTxnId();
- if (safeStore.canExecuteWith(txnId) && safeStore.tryRecurse())
+ PreLoadContext context = PreLoadContext.contextFor(txnId, "Key Waiting On");
+ if (safeStore.canExecuteWith(context) && safeStore.tryRecurse())
{
try { doNotifyWaitingOn(safeStore, txnId, key, waitingOnStatus, blockedUntil, notifyCfk); }
finally { safeStore.unrecurse(); }
}
- else safeStore.commandStore().execute(txnId, safeStore0 -> {
+ else safeStore.commandStore().execute(context, safeStore0 -> {
doNotifyWaitingOn(safeStore0, txnId, key, waitingOnStatus, blockedUntil, notifyCfk);
}, safeStore.agent());
}
@@ -131,7 +133,7 @@ interface NotifySink
RoutingKeys keys = RoutingKeys.of(key);
//noinspection ConstantConditions,SillyAssignment
safeStore = safeStore; // prevent use in lambda
- safeStore.commandStore().execute(PreLoadContext.contextFor(txnId, keys, SYNC), safeStore0 -> {
+ safeStore.commandStore().execute(PreLoadContext.contextFor(txnId, keys, SYNC, WRITE, "Notify"), safeStore0 -> {
doNotifyAlreadyReady(safeStore0, txnId, key);
}, safeStore.agent());
}
diff --git a/accord-core/src/main/java/accord/local/cfk/PostProcess.java b/accord-core/src/main/java/accord/local/cfk/PostProcess.java
index ec3046b6..5c08f8b4 100644
--- a/accord-core/src/main/java/accord/local/cfk/PostProcess.java
+++ b/accord-core/src/main/java/accord/local/cfk/PostProcess.java
@@ -44,7 +44,8 @@ import accord.utils.SortedArrays;
import accord.utils.btree.BTree;
import static accord.local.CommandSummaries.SummaryStatus.APPLIED;
-import static accord.local.KeyHistory.SYNC;
+import static accord.local.LoadKeys.SYNC;
+import static accord.local.LoadKeysFor.WRITE;
import static accord.local.PreLoadContext.contextFor;
import static accord.local.cfk.CommandsForKey.InternalStatus.INVALIDATED;
import static accord.local.cfk.CommandsForKey.InternalStatus.STABLE;
@@ -102,7 +103,7 @@ abstract class PostProcess
try { load(safeStore, safeCommand, safeCfk, notifySink); }
finally { safeStore.unrecurse(); }
}
- else safeStore.commandStore().execute(contextFor(txnId, RoutingKeys.of(key), SYNC), safeStore0 -> {
+ else safeStore.commandStore().execute(contextFor(txnId, RoutingKeys.of(key), SYNC, WRITE, "Load Pruned CommandsForKey"), safeStore0 -> {
load(safeStore0, safeStore0.unsafeGet(txnId), safeStore0.get(key), notifySink);
}, safeStore.agent());
}
diff --git a/accord-core/src/main/java/accord/local/cfk/Updating.java b/accord-core/src/main/java/accord/local/cfk/Updating.java
index fd3b0a43..4bddc48e 100644
--- a/accord-core/src/main/java/accord/local/cfk/Updating.java
+++ b/accord-core/src/main/java/accord/local/cfk/Updating.java
@@ -55,7 +55,8 @@ import accord.utils.SortedList.MergeCursor;
import static accord.api.ProtocolModifiers.Toggles.isTransitiveDependencyVisible;
import static accord.local.CommandSummaries.SummaryStatus.APPLIED;
-import static accord.local.KeyHistory.SYNC;
+import static accord.local.LoadKeys.SYNC;
+import static accord.local.LoadKeysFor.WRITE;
import static accord.local.cfk.CommandsForKey.InternalStatus.COMMITTED;
import static accord.local.cfk.CommandsForKey.InternalStatus.INVALIDATED;
import static accord.local.cfk.CommandsForKey.InternalStatus.PRUNED;
@@ -845,7 +846,7 @@ class Updating
static void updateUnmanagedAsync(CommandStore commandStore, TxnId txnId, RoutingKey key, NotifySink notifySink)
{
- PreLoadContext context = PreLoadContext.contextFor(txnId, RoutingKeys.of(key), SYNC);
+ PreLoadContext context = PreLoadContext.contextFor(txnId, RoutingKeys.of(key), SYNC, WRITE, "Update unmanaged CommandsForKey");
commandStore.execute(context, safeStore -> {
SafeCommandsForKey safeCommandsForKey = safeStore.get(key);
CommandsForKey cur = safeCommandsForKey.current();
diff --git a/accord-core/src/main/java/accord/messages/AbstractRequest.java b/accord-core/src/main/java/accord/messages/AbstractRequest.java
index 7b7eddfb..afc61fcc 100644
--- a/accord-core/src/main/java/accord/messages/AbstractRequest.java
+++ b/accord-core/src/main/java/accord/messages/AbstractRequest.java
@@ -190,4 +190,11 @@ public abstract class AbstractRequest<R extends Reply> implements PreLoadContext
if (cancel != null)
cancel.cancel();
}
+
+
+ @Override
+ public String reason()
+ {
+ return getClass().getSimpleName() + (txnId == null ? "" : "{" + txnId + '}');
+ }
}
diff --git a/accord-core/src/main/java/accord/messages/Accept.java b/accord-core/src/main/java/accord/messages/Accept.java
index 43c92999..47b47295 100644
--- a/accord-core/src/main/java/accord/messages/Accept.java
+++ b/accord-core/src/main/java/accord/messages/Accept.java
@@ -26,7 +26,8 @@ import accord.local.Command;
import accord.local.Commands;
import accord.local.Commands.AcceptOutcome;
import accord.local.DepsCalculator;
-import accord.local.KeyHistory;
+import accord.local.LoadKeys;
+import accord.local.LoadKeysFor;
import accord.local.Node.Id;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
@@ -50,7 +51,7 @@ import static accord.api.ProtocolModifiers.Toggles.filterDuplicateDependenciesFr
import static accord.local.Commands.AcceptOutcome.Redundant;
import static accord.local.Commands.AcceptOutcome.RejectedBallot;
import static accord.local.Commands.AcceptOutcome.Success;
-import static accord.local.KeyHistory.SYNC;
+import static accord.local.LoadKeys.SYNC;
import static accord.messages.MessageType.StandardMessage.ACCEPT_REQ;
import static accord.messages.MessageType.StandardMessage.ACCEPT_RSP;
import static accord.messages.MessageType.StandardMessage.NOT_ACCEPT_REQ;
@@ -194,11 +195,17 @@ public class Accept extends TxnRequest.WithUnsynced<Accept.AcceptReply>
}
@Override
- public KeyHistory keyHistory()
+ public LoadKeys loadKeys()
{
return SYNC;
}
+ @Override
+ public LoadKeysFor loadKeysFor()
+ {
+ return LoadKeysFor.READ_WRITE;
+ }
+
@Override
public MessageType type()
{
@@ -411,6 +418,12 @@ public class Accept extends TxnRequest.WithUnsynced<Accept.AcceptReply>
return "NotAccept{kind: " + status + ", ballot:" + ballot + ", txnId:" + txnId + ", key:" + participants + '}';
}
+ @Override
+ public String reason()
+ {
+ return status + "{" + txnId + '}';
+ }
+
@Override
public long waitForEpoch()
{
diff --git a/accord-core/src/main/java/accord/messages/Apply.java b/accord-core/src/main/java/accord/messages/Apply.java
index de299297..3cc7dd49 100644
--- a/accord-core/src/main/java/accord/messages/Apply.java
+++ b/accord-core/src/main/java/accord/messages/Apply.java
@@ -26,7 +26,7 @@ import accord.coordinate.ExecuteFlag.ExecuteFlags;
import accord.local.CommandStore;
import accord.local.CommandStores;
import accord.local.Commands;
-import accord.local.KeyHistory;
+import accord.local.LoadKeys;
import accord.local.Node;
import accord.local.Node.Id;
import accord.local.SafeCommand;
@@ -228,10 +228,10 @@ public class Apply extends TxnRequest<ApplyReply>
}
@Override
- public KeyHistory keyHistory()
+ public LoadKeys loadKeys()
{
// TODO (expected): need to guarantee execution order then can make this ASYNC
- return KeyHistory.SYNC;
+ return LoadKeys.SYNC;
}
@Override
diff --git a/accord-core/src/main/java/accord/messages/Await.java b/accord-core/src/main/java/accord/messages/Await.java
index 799f3104..cdc3dea2 100644
--- a/accord-core/src/main/java/accord/messages/Await.java
+++ b/accord-core/src/main/java/accord/messages/Await.java
@@ -255,6 +255,12 @@ public class Await implements Request, MapReduceConsume<SafeCommandStore, Void>,
return txnId;
}
+ @Override
+ public String reason()
+ {
+ return "AwaitComplete{" + blockedUntil + ',' + txnId + '}';
+ }
+
@Override
public MessageType type()
{
@@ -354,6 +360,12 @@ public class Await implements Request, MapReduceConsume<SafeCommandStore, Void>,
return txnId;
}
+ @Override
+ public String reason()
+ {
+ return "AwaitComplete{" + newStatus + ',' + txnId + '}';
+ }
+
@Override
public void accept(SafeCommandStore safeStore)
{
diff --git a/accord-core/src/main/java/accord/messages/BeginInvalidation.java b/accord-core/src/main/java/accord/messages/BeginInvalidation.java
index b2284b7d..1f4e63f9 100644
--- a/accord-core/src/main/java/accord/messages/BeginInvalidation.java
+++ b/accord-core/src/main/java/accord/messages/BeginInvalidation.java
@@ -123,6 +123,12 @@ public class BeginInvalidation extends AbstractRequest<BeginInvalidation.Invalid
return BEGIN_INVALIDATE_REQ;
}
+ @Override
+ public String reason()
+ {
+ return "Invalidate{" + txnId + '}';
+ }
+
@Override
public String toString()
{
diff --git a/accord-core/src/main/java/accord/messages/BeginRecovery.java b/accord-core/src/main/java/accord/messages/BeginRecovery.java
index e858a659..15e3d883 100644
--- a/accord-core/src/main/java/accord/messages/BeginRecovery.java
+++ b/accord-core/src/main/java/accord/messages/BeginRecovery.java
@@ -222,9 +222,15 @@ public class BeginRecovery extends TxnRequest.WithUnsynced<BeginRecovery.Recover
}
@Override
- public KeyHistory keyHistory()
+ public LoadKeys loadKeys()
{
- return KeyHistory.RECOVER;
+ return LoadKeys.SYNC;
+ }
+
+ @Override
+ public LoadKeysFor loadKeysFor()
+ {
+ return LoadKeysFor.RECOVERY;
}
@Override
@@ -233,6 +239,12 @@ public class BeginRecovery extends TxnRequest.WithUnsynced<BeginRecovery.Recover
return BEGIN_RECOVER_REQ;
}
+ @Override
+ public String reason()
+ {
+ return "Recover{" + txnId + '}';
+ }
+
@Override
public String toString()
{
diff --git a/accord-core/src/main/java/accord/messages/Commit.java b/accord-core/src/main/java/accord/messages/Commit.java
index ded359c0..89d8ea5d 100644
--- a/accord-core/src/main/java/accord/messages/Commit.java
+++ b/accord-core/src/main/java/accord/messages/Commit.java
@@ -23,7 +23,7 @@ import javax.annotation.Nullable;
import accord.coordinate.ExecuteFlag.CoordinationFlags;
import accord.coordinate.ExecuteFlag.ExecuteFlags;
import accord.local.Commands;
-import accord.local.KeyHistory;
+import accord.local.LoadKeys;
import accord.local.Node;
import accord.local.Node.Id;
import accord.local.PreLoadContext;
@@ -269,10 +269,10 @@ public class Commit extends TxnRequest.WithUnsynced<CommitOrReadNack>
}
@Override
- public KeyHistory keyHistory()
+ public LoadKeys loadKeys()
{
// TODO (expected): need to guarantee execution order then can make this ASYNC
- return KeyHistory.SYNC;
+ return LoadKeys.SYNC;
}
@Override
@@ -391,6 +391,12 @@ public class Commit extends TxnRequest.WithUnsynced<CommitOrReadNack>
return txnId;
}
+ @Override
+ public String reason()
+ {
+ return "Commit Invalidate";
+ }
+
@Override
public long waitForEpoch()
{
diff --git a/accord-core/src/main/java/accord/messages/GetDurableBefore.java b/accord-core/src/main/java/accord/messages/GetDurableBefore.java
index 4a71cf88..d5812f08 100644
--- a/accord-core/src/main/java/accord/messages/GetDurableBefore.java
+++ b/accord-core/src/main/java/accord/messages/GetDurableBefore.java
@@ -59,6 +59,12 @@ public class GetDurableBefore implements Request, PreLoadContext
return null;
}
+ @Override
+ public String reason()
+ {
+ return "GetDurableBefore";
+ }
+
public static class DurableBeforeReply implements Reply
{
public final DurableBefore durableBeforeMap;
diff --git a/accord-core/src/main/java/accord/messages/GetEphemeralReadDeps.java b/accord-core/src/main/java/accord/messages/GetEphemeralReadDeps.java
index a0829543..f2ab3659 100644
--- a/accord-core/src/main/java/accord/messages/GetEphemeralReadDeps.java
+++ b/accord-core/src/main/java/accord/messages/GetEphemeralReadDeps.java
@@ -23,7 +23,8 @@ import javax.annotation.Nullable;
import accord.coordinate.ExecuteFlag.ExecuteFlags;
import accord.local.DepsCalculator;
-import accord.local.KeyHistory;
+import accord.local.LoadKeys;
+import accord.local.LoadKeysFor;
import accord.local.Node.Id;
import accord.local.SafeCommandStore;
import accord.local.StoreParticipants;
@@ -111,9 +112,15 @@ public class GetEphemeralReadDeps extends TxnRequest.WithUnsynced<GetEphemeralRe
}
@Override
- public KeyHistory keyHistory()
+ public LoadKeys loadKeys()
{
- return KeyHistory.SYNC;
+ return LoadKeys.SYNC;
+ }
+
+ @Override
+ public LoadKeysFor loadKeysFor()
+ {
+ return LoadKeysFor.READ_WRITE;
}
public static class GetEphemeralReadDepsOk implements Reply
diff --git a/accord-core/src/main/java/accord/messages/GetLatestDeps.java b/accord-core/src/main/java/accord/messages/GetLatestDeps.java
index 5548ed37..082d2ec7 100644
--- a/accord-core/src/main/java/accord/messages/GetLatestDeps.java
+++ b/accord-core/src/main/java/accord/messages/GetLatestDeps.java
@@ -23,7 +23,8 @@ import javax.annotation.Nullable;
import accord.local.Command;
import accord.local.DepsCalculator;
-import accord.local.KeyHistory;
+import accord.local.LoadKeys;
+import accord.local.LoadKeysFor;
import accord.local.Node.Id;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
@@ -124,9 +125,15 @@ public class GetLatestDeps extends TxnRequest.WithUnsynced<GetLatestDeps.GetLate
}
@Override
- public KeyHistory keyHistory()
+ public LoadKeys loadKeys()
{
- return KeyHistory.SYNC;
+ return LoadKeys.SYNC;
+ }
+
+ @Override
+ public LoadKeysFor loadKeysFor()
+ {
+ return LoadKeysFor.READ_WRITE;
}
public interface GetLatestDepsReply extends Reply
diff --git a/accord-core/src/main/java/accord/messages/GetMaxConflict.java b/accord-core/src/main/java/accord/messages/GetMaxConflict.java
index 7064be09..ea1ed311 100644
--- a/accord-core/src/main/java/accord/messages/GetMaxConflict.java
+++ b/accord-core/src/main/java/accord/messages/GetMaxConflict.java
@@ -20,7 +20,7 @@ package accord.messages;
import javax.annotation.Nonnull;
-import accord.local.KeyHistory;
+import accord.local.LoadKeys;
import accord.local.Node;
import accord.local.SafeCommandStore;
import accord.primitives.FullRoute;
@@ -98,6 +98,12 @@ public class GetMaxConflict extends TxnRequest.WithUnsynced<GetMaxConflict.GetMa
return null;
}
+ @Override
+ public String reason()
+ {
+ return toString();
+ }
+
@Override
public Unseekables<?> keys()
{
@@ -105,9 +111,9 @@ public class GetMaxConflict extends TxnRequest.WithUnsynced<GetMaxConflict.GetMa
}
@Override
- public KeyHistory keyHistory()
+ public LoadKeys loadKeys()
{
- return KeyHistory.NONE;
+ return LoadKeys.NONE;
}
public static class GetMaxConflictOk implements Reply
diff --git a/accord-core/src/main/java/accord/messages/InformDurable.java b/accord-core/src/main/java/accord/messages/InformDurable.java
index 7f145ede..d73ffde9 100644
--- a/accord-core/src/main/java/accord/messages/InformDurable.java
+++ b/accord-core/src/main/java/accord/messages/InformDurable.java
@@ -20,7 +20,7 @@ package accord.messages;
import javax.annotation.Nullable;
import accord.local.Commands;
-import accord.local.KeyHistory;
+import accord.local.LoadKeys;
import accord.local.Node;
import accord.local.Node.Id;
import accord.local.PreLoadContext;
@@ -110,7 +110,7 @@ public class InformDurable extends TxnRequest<Reply> implements PreLoadContext
public Cancellable submit()
{
// TODO (expected): do not load from disk to perform this update, just write a delta to any journal
- return node.mapReduceConsumeLocal(txnId, scope, minEpoch, maxEpoch, this);
+ return node.mapReduceConsumeLocal(this, scope, minEpoch, maxEpoch, this);
}
@Override
@@ -126,9 +126,9 @@ public class InformDurable extends TxnRequest<Reply> implements PreLoadContext
}
@Override
- public KeyHistory keyHistory()
+ public LoadKeys loadKeys()
{
- return dependencyElision() == IF_DURABLE ? KeyHistory.ASYNC : KeyHistory.NONE;
+ return dependencyElision() == IF_DURABLE ? LoadKeys.ASYNC : LoadKeys.NONE;
}
@Override
diff --git a/accord-core/src/main/java/accord/messages/PreAccept.java b/accord-core/src/main/java/accord/messages/PreAccept.java
index dabff631..0711fda1 100644
--- a/accord-core/src/main/java/accord/messages/PreAccept.java
+++ b/accord-core/src/main/java/accord/messages/PreAccept.java
@@ -28,7 +28,8 @@ import accord.coordinate.ExecuteFlag.ExecuteFlags;
import accord.local.Command;
import accord.local.Commands;
import accord.local.DepsCalculator;
-import accord.local.KeyHistory;
+import accord.local.LoadKeys;
+import accord.local.LoadKeysFor;
import accord.local.Node.Id;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
@@ -95,9 +96,15 @@ public class PreAccept extends WithUnsynced<PreAccept.PreAcceptReply>
}
@Override
- public KeyHistory keyHistory()
+ public LoadKeys loadKeys()
{
- return KeyHistory.SYNC;
+ return LoadKeys.SYNC;
+ }
+
+ @Override
+ public LoadKeysFor loadKeysFor()
+ {
+ return LoadKeysFor.READ_WRITE;
}
@Override
diff --git a/accord-core/src/main/java/accord/messages/Propagate.java b/accord-core/src/main/java/accord/messages/Propagate.java
index a616a021..f030f7bb 100644
--- a/accord-core/src/main/java/accord/messages/Propagate.java
+++ b/accord-core/src/main/java/accord/messages/Propagate.java
@@ -183,6 +183,12 @@ public class Propagate implements PreLoadContext, MapReduceConsume<SafeCommandSt
return txnId;
}
+ @Override
+ public String reason()
+ {
+ return "Propagate";
+ }
+
@Override
public Void apply(SafeCommandStore safeStore)
{
diff --git a/accord-core/src/main/java/accord/messages/ReadData.java b/accord-core/src/main/java/accord/messages/ReadData.java
index 0d4f14d8..6b77a1c6 100644
--- a/accord-core/src/main/java/accord/messages/ReadData.java
+++ b/accord-core/src/main/java/accord/messages/ReadData.java
@@ -862,4 +862,10 @@ public abstract class ReadData implements PreLoadContext, Request, MapReduceCons
return true;
}
}
+
+ @Override
+ public String reason()
+ {
+ return getClass().getSimpleName();
+ }
}
diff --git a/accord-core/src/main/java/accord/messages/ReadEphemeralTxnData.java b/accord-core/src/main/java/accord/messages/ReadEphemeralTxnData.java
index dd07adf9..f14aa06d 100644
--- a/accord-core/src/main/java/accord/messages/ReadEphemeralTxnData.java
+++ b/accord-core/src/main/java/accord/messages/ReadEphemeralTxnData.java
@@ -177,7 +177,7 @@ public class ReadEphemeralTxnData extends ReadData
while (iter.hasNext())
{
node.commandStores().forId(iter.nextValue())
- .execute(PreLoadContext.empty(), safeStore -> {
+ .execute((PreLoadContext.Empty) () -> "Timeout Ephemeral Read", safeStore -> {
eraseEphemeralRead(safeStore, txnId);
}, node.agent());
}
@@ -196,4 +196,10 @@ public class ReadEphemeralTxnData extends ReadData
{
return READ_EPHEMERAL_REQ;
}
+
+ @Override
+ public String reason()
+ {
+ return "Ephemeral Read";
+ }
}
diff --git a/accord-core/src/main/java/accord/messages/ReadTxnData.java b/accord-core/src/main/java/accord/messages/ReadTxnData.java
index 2c0c0543..995febe1 100644
--- a/accord-core/src/main/java/accord/messages/ReadTxnData.java
+++ b/accord-core/src/main/java/accord/messages/ReadTxnData.java
@@ -81,4 +81,10 @@ public class ReadTxnData extends ReadData
{
return READ_REQ;
}
+
+ @Override
+ public String reason()
+ {
+ return "Read";
+ }
}
diff --git a/accord-core/src/main/java/accord/messages/SetGloballyDurable.java b/accord-core/src/main/java/accord/messages/SetGloballyDurable.java
index 473ea98d..cff218fb 100644
--- a/accord-core/src/main/java/accord/messages/SetGloballyDurable.java
+++ b/accord-core/src/main/java/accord/messages/SetGloballyDurable.java
@@ -63,4 +63,10 @@ public class SetGloballyDurable implements Request, PreLoadContext
{
return null;
}
+
+ @Override
+ public String reason()
+ {
+ return "SetGloballyDurable";
+ }
}
diff --git a/accord-core/src/main/java/accord/messages/StableThenRead.java b/accord-core/src/main/java/accord/messages/StableThenRead.java
index 9eccece8..acc0033b 100644
--- a/accord-core/src/main/java/accord/messages/StableThenRead.java
+++ b/accord-core/src/main/java/accord/messages/StableThenRead.java
@@ -21,7 +21,7 @@ package accord.messages;
import javax.annotation.Nullable;
import accord.local.Commands;
-import accord.local.KeyHistory;
+import accord.local.LoadKeys;
import accord.local.Node;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
@@ -121,9 +121,9 @@ public class StableThenRead extends ReadData
}
@Override
- public KeyHistory keyHistory()
+ public LoadKeys loadKeys()
{
- return KeyHistory.ASYNC;
+ return LoadKeys.ASYNC;
}
@Override
diff --git a/accord-core/src/main/java/accord/primitives/TxnId.java b/accord-core/src/main/java/accord/primitives/TxnId.java
index 7328be1d..48d477ae 100644
--- a/accord-core/src/main/java/accord/primitives/TxnId.java
+++ b/accord-core/src/main/java/accord/primitives/TxnId.java
@@ -24,7 +24,6 @@ import java.util.regex.Pattern;
import javax.annotation.Nullable;
import accord.local.Node.Id;
-import accord.local.PreLoadContext;
import accord.primitives.Routable.Domain;
import accord.primitives.Txn.Kind;
import accord.primitives.Txn.Kind.Kinds;
@@ -39,7 +38,7 @@ import static accord.primitives.TxnId.Cardinality.Any;
import static accord.primitives.TxnId.MediumPath.NoMediumPath;
import static accord.utils.Invariants.illegalArgument;
-public class TxnId extends Timestamp implements PreLoadContext
+public class TxnId extends Timestamp
{
public enum FastPath
{
@@ -416,13 +415,6 @@ public class TxnId extends Timestamp implements PreLoadContext
return merge(this, that, TxnId::fromBits);
}
- @Nullable
- @Override
- public TxnId primaryTxnId()
- {
- return this;
- }
-
@Override
public String toString()
{
diff --git a/accord-core/src/main/java/accord/utils/Reduce.java b/accord-core/src/main/java/accord/utils/Reduce.java
index 2150054a..be6dbf22 100644
--- a/accord-core/src/main/java/accord/utils/Reduce.java
+++ b/accord-core/src/main/java/accord/utils/Reduce.java
@@ -21,5 +21,5 @@ package accord.utils;
public interface Reduce<I, O>
{
static <I, O> Reduce<I, O> toNull() { return (a, b) -> null; }
- O reduce(I o1, I o2);
+ O reduce(O o1, I o2);
}
diff --git a/accord-core/src/main/java/accord/utils/async/AsyncChains.java b/accord-core/src/main/java/accord/utils/async/AsyncChains.java
index 135cc65c..5fd001ce 100644
--- a/accord-core/src/main/java/accord/utils/async/AsyncChains.java
+++ b/accord-core/src/main/java/accord/utils/async/AsyncChains.java
@@ -458,15 +458,15 @@ public abstract class AsyncChains<V> implements AsyncChain<V>
@VisibleForTesting
static class ReducingAsyncChain<V> extends AbstractReducingAsyncChain<V, V>
{
- private final Reduce<? super V, ? extends V> reducer;
+ private final Reduce<V, V> reducer;
- private ReducingAsyncChain(List<? extends AsyncChain<? extends V>> list, Reduce<? super V, ? extends V> reducer)
+ private ReducingAsyncChain(List<? extends AsyncChain<? extends V>> list, Reduce<V, V> reducer)
{
super(list);
this.reducer = reducer;
}
- private ReducingAsyncChain(AsyncChain<V> accum, AsyncChain<V> add, Reduce<? super V, ? extends V> reducer)
+ private ReducingAsyncChain(AsyncChain<V> accum, AsyncChain<V> add, Reduce<V, V> reducer)
{
super(accum, add);
this.reducer = reducer;
@@ -703,34 +703,34 @@ public abstract class AsyncChains<V> implements AsyncChain<V>
};
}
- public static <V> AsyncChain<V> reduce(List<? extends AsyncChain<? extends V>> chains, Reduce<? super V, ? extends V> reducer)
+ public static <V> AsyncChain<V> reduce(List<? extends AsyncChain<? extends V>> chains, Reduce<V, V> reducer)
{
if (chains.size() == 1)
return (AsyncChain<V>) chains.get(0);
return new ReducingAsyncChain<>(chains, reducer);
}
- public static <I, O> AsyncChain<O> reduce(List<? extends AsyncChain<? extends I>> chains, BiFunction<? super I, ? super O, ? extends O> reducer, O identity)
+ public static <I, O> AsyncChain<O> reduce(List<? extends AsyncChain<? extends I>> chains, Reduce<I, O> reducer, O identity)
{
switch (chains.size())
{
case 0: return AsyncChains.success(identity);
- case 1: return chains.get(0).map(a -> reducer.apply(a, identity));
+ case 1: return chains.get(0).map(a -> reducer.reduce(identity, a));
}
- return new AbstractReducingAsyncChain<I, O>(chains)
+ return new AbstractReducingAsyncChain<>(chains)
{
@Override
O process(I[] inputs)
{
O result = identity;
for (I input : inputs)
- result = reducer.apply(input, identity);
+ result = reducer.reduce(result, input);
return result;
}
};
}
- public static <V> AsyncChain<V> reduce(AsyncChain<V> accum, AsyncChain<V> add, Reduce<? super V, ? extends V> reducer)
+ public static <V> AsyncChain<V> reduce(AsyncChain<V> accum, AsyncChain<V> add, Reduce<V, V> reducer)
{
if (ReducingAsyncChain.match(accum, reducer))
{
diff --git a/accord-core/src/test/java/accord/coordinate/TopologyChangeTest.java b/accord-core/src/test/java/accord/coordinate/TopologyChangeTest.java
index 37cfc743..8adb26a3 100644
--- a/accord-core/src/test/java/accord/coordinate/TopologyChangeTest.java
+++ b/accord-core/src/test/java/accord/coordinate/TopologyChangeTest.java
@@ -72,7 +72,7 @@ public class TopologyChangeTest
TxnId txnId1 = node1.nextTxnId(Write, Key);
Txn txn1 = writeTxn(keys);
getUninterruptibly(node1.coordinate(txnId1, txn1));
- getUninterruptibly(node1.commandStores().forEach(txnId1, keys.toParticipants(), 1, 1, safeStore -> {
+ getUninterruptibly(node1.commandStores().forEach(PreLoadContext.contextFor(txnId1, "Test"), keys.toParticipants(), 1, 1, safeStore -> {
StoreParticipants participants = StoreParticipants.read(safeStore, keys.toParticipants(), txnId1);
Command command = safeStore.get(txnId1, participants).current();
Assertions.assertTrue(command.partialDeps().isEmpty());
@@ -99,7 +99,7 @@ public class TopologyChangeTest
cluster.nodes(4, 5).forEach(node -> {
try
{
- getUninterruptibly(node.commandStores().forEach(contextFor(txnId1, txnId2), keys.toParticipants(), 2, 2, safeStore -> {
+ getUninterruptibly(node.commandStores().forEach(contextFor(txnId1, txnId2, "Test"), keys.toParticipants(), 2, 2, safeStore -> {
StoreParticipants participants = StoreParticipants.update(safeStore, keys.toParticipants(), txnId2.epoch(), txnId2, txnId2.epoch());
Command command = safeStore.get(txnId2, participants).current();
Assertions.assertTrue(command.partialDeps().contains(txnId1));
@@ -130,7 +130,7 @@ public class TopologyChangeTest
if (node.epoch() < epoch)
throw new AssertionError(String.format("node[%s] epoch %s is less than check epoch %s", node.id(), node.epoch(), epoch));
- node.forEachLocal(PreLoadContext.contextFor(keys.toParticipants()), participants, 1, node.epoch(), safeStore -> {
+ node.forEachLocal((PreLoadContext.Empty) () -> "Test", participants, 1, node.epoch(), safeStore -> {
boolean rejected = safeStore.commandStore().isRejectedIfNotPreAccepted(TxnId.minForEpoch(epoch), participants);
if (rejected != rejectionExpected)
{
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 43eda4c0..65f9c10a 100644
--- a/accord-core/src/test/java/accord/impl/list/ListFetchCoordinator.java
+++ b/accord-core/src/test/java/accord/impl/list/ListFetchCoordinator.java
@@ -64,7 +64,7 @@ public class ListFetchCoordinator extends AbstractFetchCoordinator
return;
ListData listData = (ListData) data;
- persisting.add(commandStore.build(PreLoadContext.empty(), safeStore -> {
+ 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) -> {
if (fail == null) success(from, received);
diff --git a/accord-core/src/test/java/accord/local/ImmutableCommandTest.java b/accord-core/src/test/java/accord/local/ImmutableCommandTest.java
index 34cd2463..a1ddd093 100644
--- a/accord-core/src/test/java/accord/local/ImmutableCommandTest.java
+++ b/accord-core/src/test/java/accord/local/ImmutableCommandTest.java
@@ -136,7 +136,7 @@ public class ImmutableCommandTest
}
commands.execute(() -> {
- SafeCommandStore safeStore = commands.beginOperation(PreLoadContext.contextFor(txnId, keys.toParticipants()));
+ SafeCommandStore safeStore = commands.beginOperation(PreLoadContext.contextFor(txnId, "Test"));
try
{
StoreParticipants participants = StoreParticipants.update(safeStore, ROUTE, txnId.epoch(), txnId, txnId.epoch());
@@ -169,7 +169,7 @@ public class ImmutableCommandTest
Assertions.assertEquals(Status.NotDefined, command.status());
Assertions.assertNull(command.executeAt());
}
- PreLoadContext context = PreLoadContext.contextFor(txnId, keys.toParticipants());
+ PreLoadContext context = PreLoadContext.contextFor(txnId, "Test");
setTopologyEpoch(support.local, 2);
((TestableConfigurationService)node.configService()).reportTopology(TopologyUtils.withEpoch(support.local.get(), 2));
@@ -178,7 +178,7 @@ public class ImmutableCommandTest
StoreParticipants participants = StoreParticipants.update(safeStore, ROUTE, txnId.epoch(), txnId, 2);
Commands.preaccept(safeStore, safeStore.get(txnId, participants), participants, txnId, txn.slice(FULL_RANGES, true), null, false);
}));
- commands.build(PreLoadContext.contextFor(txnId, txn.keys().toParticipants()), safeStore -> {
+ commands.build(PreLoadContext.contextFor(txnId, "Test"), safeStore -> {
Command command = safeStore.get(txnId).current();
Assertions.assertEquals(Status.PreAccepted, command.status());
Assertions.assertEquals(expectedTimestamp, command.executeAt());
diff --git a/accord-core/src/test/java/accord/messages/PreAcceptTest.java b/accord-core/src/test/java/accord/messages/PreAcceptTest.java
index 1c23179a..ca5d6244 100644
--- a/accord-core/src/test/java/accord/messages/PreAcceptTest.java
+++ b/accord-core/src/test/java/accord/messages/PreAcceptTest.java
@@ -61,7 +61,8 @@ import static accord.Utils.writeTxn;
import static accord.impl.InMemoryCommandStore.inMemory;
import static accord.impl.IntKey.routing;
import static accord.impl.mock.MockCluster.configService;
-import static accord.local.KeyHistory.SYNC;
+import static accord.local.LoadKeys.SYNC;
+import static accord.local.LoadKeysFor.WRITE;
import static accord.primitives.Routable.Domain.Key;
import static accord.primitives.Txn.Kind.Write;
import static accord.topology.Topologies.SelectNodeOwnership.SHARE;
@@ -105,7 +106,7 @@ public class PreAcceptTest
clock.increment(10);
preAccept.process(node, ID2, REPLY_CONTEXT);
- commandStore.build(PreLoadContext.contextFor(txnId, txn.keys().toParticipants(), SYNC), safeStore -> {
+ commandStore.build(PreLoadContext.contextFor(txnId, txn.keys().toParticipants(), SYNC, WRITE, "Test"), safeStore -> {
CommandsForKey cfk = safeStore.get(key.toUnseekable()).current();
TxnId commandId = cfk.get(0).plainTxnId();
Command command = safeStore.ifInitialised(commandId).current();
@@ -274,7 +275,7 @@ public class PreAcceptTest
clock.increment(10);
preAccept.process(node, ID2, REPLY_CONTEXT);
- commandStore.build(PreLoadContext.contextFor(txnId, txn.keys().toParticipants(), SYNC), safeStore -> {
+ commandStore.build(PreLoadContext.contextFor(txnId, txn.keys().toParticipants(), SYNC, WRITE, "Test"), safeStore -> {
CommandsForKey cfk = safeStore.get(key.toUnseekable()).current();
TxnId commandId = cfk.get(0).plainTxnId();
Command command = safeStore.ifInitialised(commandId).current();
diff --git a/accord-core/src/test/java/accord/messages/ReadDataTest.java b/accord-core/src/test/java/accord/messages/ReadDataTest.java
index 6c083ea9..a0ec3fa3 100644
--- a/accord-core/src/test/java/accord/messages/ReadDataTest.java
+++ b/accord-core/src/test/java/accord/messages/ReadDataTest.java
@@ -153,7 +153,7 @@ class ReadDataTest
{
// status=Commit, will listen waiting for ReadyToExecute; obsolete marked by status listener
test(state -> {
- state.forEach(store -> check(store.build(PreLoadContext.contextFor(state.txnId, state.route), safe -> {
+ state.forEach(store -> check(store.build(PreLoadContext.contextFor(state.txnId, "Test"), safe -> {
CheckedCommands.preaccept(safe, state.txnId, state.partialTxn, state.route);
CheckedCommands.accept(safe, state.txnId, Ballot.ZERO, state.partialRoute, state.executeAt, state.deps);
@@ -192,7 +192,7 @@ class ReadDataTest
state.readyToExecute(store);
store = stores.get(1);
- check(store.build(PreLoadContext.contextFor(state.txnId, state.route), safeStore -> {
+ check(store.build(PreLoadContext.contextFor(state.txnId, "Test"), safeStore -> {
StoreParticipants participants = StoreParticipants.notAccept(safeStore, state.route, state.txnId);
SafeCommand safeCommand = safeStore.get(state.txnId, participants);
Command prev = safeCommand.current();
@@ -210,7 +210,7 @@ class ReadDataTest
{
test(state -> {
List<CommandStore> stores = stores(state);
- stores.forEach(store -> check(store.build(PreLoadContext.contextFor(state.txnId, state.route), safeStore -> {
+ stores.forEach(store -> check(store.build(PreLoadContext.contextFor(state.txnId, "Test"), safeStore -> {
StoreParticipants participants = StoreParticipants.notAccept(safeStore, state.route, state.txnId);
SafeCommand command = safeStore.get(state.txnId, participants);
command.commitInvalidated(safeStore);
@@ -279,7 +279,7 @@ class ReadDataTest
void readyToExecute(CommandStore store)
{
- check(store.build(PreLoadContext.contextFor(txnId, route), safe -> {
+ check(store.build(PreLoadContext.contextFor(txnId, "Test"), safe -> {
CheckedCommands.preaccept(safe, txnId, partialTxn, route);
CheckedCommands.accept(safe, txnId, Ballot.ZERO, partialRoute, executeAt, deps);
CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, partialTxn, executeAt, deps);
@@ -303,7 +303,7 @@ class ReadDataTest
Mockito.when(write.apply(any(), any(), any(), any(), any())).thenReturn(writeResult);
Writes writes = new Writes(txnId, executeAt, keys, write);
- forEach(store -> check(store.build(PreLoadContext.contextFor(txnId, route), safe -> {
+ forEach(store -> check(store.build(PreLoadContext.contextFor(txnId, "Test"), safe -> {
CheckedCommands.apply(safe, txnId, route, executeAt, deps, partialTxn, writes, Mockito.mock(Result.class));
})));
return writeResult;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org