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/09/18 17:17:14 UTC
(cassandra-accord) branch trunk-tmp updated (730a6ff6 -> 8c079883)
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a change to branch trunk-tmp
in repository https://gitbox.apache.org/repos/asf/cassandra-accord.git
omit 730a6ff6 Improve tracing: - Support pattern trace table - Add basic tracing to all coordinations
new 8c079883 Improve tracing: - Support pattern trace table - Add basic tracing to all coordinations
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (730a6ff6)
\
N -- N -- N refs/heads/trunk-tmp (8c079883)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
(cassandra-accord) 01/01: Improve tracing: - Support pattern trace table - Add basic tracing to all coordinations
Posted by be...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a commit to branch trunk-tmp
in repository https://gitbox.apache.org/repos/asf/cassandra-accord.git
commit 8c0798839775415f035fc72dc2255072bf98407c
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Mon Sep 15 10:03:57 2025 +0100
Improve tracing:
- Support pattern trace table
- Add basic tracing to all coordinations
patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20911
---
accord-core/src/main/java/accord/api/Agent.java | 4 +-
.../java/accord/api/CoordinatorEventListener.java | 4 +
.../src/main/java/accord/api/TraceEventType.java | 23 ----
accord-core/src/main/java/accord/api/Tracing.java | 83 ++++++++------
.../coordinate/AbstractCoordinatePreAccept.java | 15 +--
.../accord/coordinate/AbstractCoordination.java | 31 +++++-
.../coordinate/AbstractSimpleCoordination.java | 9 +-
.../java/accord/coordinate/AsynchronousAwait.java | 18 +---
.../main/java/accord/coordinate/CheckShards.java | 24 +----
.../java/accord/coordinate/CollectLatestDeps.java | 14 +--
.../accord/coordinate/CoordinateEphemeralRead.java | 8 +-
.../accord/coordinate/CoordinateMaxConflict.java | 2 +-
.../accord/coordinate/CoordinatePreAccept.java | 8 +-
.../accord/coordinate/CoordinateSyncPoint.java | 6 +-
.../accord/coordinate/CoordinateTransaction.java | 21 ++--
.../main/java/accord/coordinate/Coordination.java | 21 ++--
.../accord/coordinate/ExecuteEphemeralRead.java | 6 +-
.../main/java/accord/coordinate/ExecuteFlag.java | 10 +-
.../java/accord/coordinate/ExecuteSyncPoint.java | 12 +--
.../main/java/accord/coordinate/ExecuteTxn.java | 5 +-
.../java/accord/coordinate/FetchCoordinator.java | 12 +--
.../src/main/java/accord/coordinate/FetchData.java | 18 ++--
.../main/java/accord/coordinate/FetchRoute.java | 19 ++--
.../src/main/java/accord/coordinate/Infer.java | 4 +-
.../main/java/accord/coordinate/Invalidate.java | 30 ++----
.../main/java/accord/coordinate/MaybeRecover.java | 24 ++---
.../src/main/java/accord/coordinate/Persist.java | 21 ++--
.../java/accord/coordinate/PrepareRecovery.java | 47 ++++----
.../src/main/java/accord/coordinate/Propose.java | 43 +++-----
.../java/accord/coordinate/ReadCoordinator.java | 42 +++++++-
.../src/main/java/accord/coordinate/Recover.java | 119 +++++++++------------
.../src/main/java/accord/coordinate/Stabilise.java | 27 ++---
.../java/accord/coordinate/SynchronousAwait.java | 15 +--
.../accord/coordinate/SynchronousRecoverAwait.java | 5 +-
.../java/accord/impl/progresslog/HomeState.java | 12 +--
.../java/accord/impl/progresslog/WaitingState.java | 36 ++++---
.../src/main/java/accord/local/Command.java | 6 +-
accord-core/src/main/java/accord/local/Node.java | 13 +--
.../src/main/java/accord/local/SafeCommand.java | 12 +++
.../main/java/accord/local/StoreParticipants.java | 5 +
.../src/main/java/accord/messages/Propagate.java | 5 +-
.../src/main/java/accord/primitives/Routable.java | 2 +-
.../src/main/java/accord/primitives/Txn.java | 13 ++-
.../src/main/java/accord/primitives/TxnId.java | 4 +-
.../src/main/java/accord/utils/TinyEnumSet.java | 95 ++++++++++++++--
.../src/test/java/accord/impl/list/ListAgent.java | 8 +-
46 files changed, 488 insertions(+), 473 deletions(-)
diff --git a/accord-core/src/main/java/accord/api/Agent.java b/accord-core/src/main/java/accord/api/Agent.java
index 9219e4a4..3fbb4066 100644
--- a/accord-core/src/main/java/accord/api/Agent.java
+++ b/accord-core/src/main/java/accord/api/Agent.java
@@ -23,10 +23,12 @@ import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
import accord.api.ProgressLog.BlockedUntil;
+import accord.coordinate.Coordination.CoordinationKind;
import accord.local.Node;
import accord.local.SafeCommandStore;
import accord.local.TimeService;
import accord.messages.ReplyContext;
+import accord.primitives.Participants;
import accord.primitives.Routable.Domain;
import accord.primitives.Status.Phase;
import accord.primitives.Txn;
@@ -40,7 +42,7 @@ import accord.utils.async.AsyncChain;
*/
public interface Agent extends UncaughtExceptionListener
{
- default @Nullable Tracing trace(TxnId txnId, TraceEventType eventType) { return null; }
+ default @Nullable Tracing trace(@Nullable TxnId txnId, @Nullable Participants<?> participants, CoordinationKind eventType) { return null; }
OwnershipEventListener ownershipEvents();
diff --git a/accord-core/src/main/java/accord/api/CoordinatorEventListener.java b/accord-core/src/main/java/accord/api/CoordinatorEventListener.java
index 7f771934..cd2254ce 100644
--- a/accord-core/src/main/java/accord/api/CoordinatorEventListener.java
+++ b/accord-core/src/main/java/accord/api/CoordinatorEventListener.java
@@ -20,15 +20,19 @@ package accord.api;
import javax.annotation.Nullable;
+import accord.coordinate.Coordination;
import accord.coordinate.ExecutePath;
import accord.local.Node;
import accord.primitives.Ballot;
import accord.primitives.Deps;
+import accord.primitives.Participants;
import accord.primitives.Status.Durability;
import accord.primitives.TxnId;
public interface CoordinatorEventListener
{
+ default void onFailed(Throwable failure, TxnId txnId, Participants<?> participants, Coordination coordination) {}
+
default void onPreAccepted(TxnId txnId) {}
default void onAccepted(TxnId txnId, Ballot ballot) {}
default void onExecuting(TxnId txnId, @Nullable Ballot ballot, Deps deps, @Nullable ExecutePath path) {}
diff --git a/accord-core/src/main/java/accord/api/TraceEventType.java b/accord-core/src/main/java/accord/api/TraceEventType.java
deleted file mode 100644
index 22afbfac..00000000
--- a/accord-core/src/main/java/accord/api/TraceEventType.java
+++ /dev/null
@@ -1,23 +0,0 @@
-/*
- * Licensed to the Apache Software ation (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;
-
-public enum TraceEventType
-{
- FETCH, HOME_PROGRESS, WAIT_PROGRESS, RECOVER
-}
diff --git a/accord-core/src/main/java/accord/api/Tracing.java b/accord-core/src/main/java/accord/api/Tracing.java
index 214632c0..7d5e7661 100644
--- a/accord-core/src/main/java/accord/api/Tracing.java
+++ b/accord-core/src/main/java/accord/api/Tracing.java
@@ -17,10 +17,6 @@
*/
package accord.api;
-import java.util.function.BiConsumer;
-
-import javax.annotation.Nullable;
-
import accord.local.CommandStore;
public interface Tracing
@@ -34,35 +30,61 @@ public interface Tracing
static String safeFormat(String fmt, Object ... args)
{
- try
- {
- return String.format(fmt, args);
- }
- catch (Throwable t)
+ if (args.length == 0)
+ return fmt;
+
+ StringBuilder out = new StringBuilder();
+ int prev = 0;
+ for (int argIndex = 0 ; argIndex < args.length ; ++argIndex)
{
+ Object arg = args[argIndex];
+ int next = fmt.indexOf('%', prev);
+ if (next < 0)
+ break;
+
+ out.append(fmt, prev, next);
+ if (++next == fmt.length())
+ throw new IllegalArgumentException("Invalid substitution declaration: % not followed by d, s or %");
+
+ char ch = fmt.charAt(next);
+ prev = next + 1;
+
+ if (ch == '%')
+ {
+ out.append('%');
+ prev = next + 1;
+ continue;
+ }
+
+ if (ch != 's' && ch != 'd')
+ throw new IllegalArgumentException("Invalid substitution declaration: % not followed by d, s or %");
+
+ if (arg == null)
+ {
+ out.append("null");
+ continue;
+ }
+
try
{
- String thrown = format(t);
- StringBuilder argsStr = new StringBuilder();
- if (args == null) argsStr.append("null");
- else
- {
- argsStr.append('[');
- for (int i = 0 ; i < args.length ; i++)
- {
- if (i > 0) argsStr.append(',');
- try { argsStr.append(args[i]); }
- catch (Throwable t2) { argsStr.append("<Could not invoke toString(): ").append(format(t2)).append('>'); }
- }
- argsStr.append(']');
- }
- return "<Could not invoke String.format(\"" + fmt + "\", " + argsStr + "): " + thrown + '>';
+ if (arg instanceof Throwable)
+ arg = format((Throwable) arg);
+ out.append(arg);
}
- catch (Throwable t2)
+ catch (Throwable t)
{
- return "<Could not format string or failure info>";
+ try
+ {
+ out.append("<Could not invoke toString(): ").append(format(t)).append('>');
+ }
+ catch (Throwable t2)
+ {
+ out.append("<Could not invoke toString() on argument ").append(argIndex).append('>');
+ }
}
}
+ out.append(fmt, prev, fmt.length());
+ return out.toString();
}
static String format(Throwable failure)
@@ -71,13 +93,4 @@ public interface Tracing
return failure.getClass().getSimpleName() + ':' + failure.getLocalizedMessage()
+ (ste.length > 0 ? " (@" + ste[0].getClassName() + '.' + ste[0].getMethodName() + ':' + ste[0].getLineNumber() + ')' : "");
}
-
- static <V> BiConsumer<V, Throwable> wrap(BiConsumer<V, Throwable> wrap, String context, @Nullable Tracing tracing)
- {
- if (tracing == null) return wrap;
- return (success, fail) -> {
- if (fail != null) tracing.trace(null, "Failure when %s: %s", context, format(fail));
- wrap.accept(success, fail);
- };
- }
}
diff --git a/accord-core/src/main/java/accord/coordinate/AbstractCoordinatePreAccept.java b/accord-core/src/main/java/accord/coordinate/AbstractCoordinatePreAccept.java
index 54b76afd..dcb60592 100644
--- a/accord-core/src/main/java/accord/coordinate/AbstractCoordinatePreAccept.java
+++ b/accord-core/src/main/java/accord/coordinate/AbstractCoordinatePreAccept.java
@@ -27,7 +27,6 @@ import accord.local.SequentialAsyncExecutor;
import accord.messages.Callback;
import accord.primitives.FullRoute;
import accord.primitives.TxnId;
-import accord.primitives.Unseekables;
import accord.topology.Topologies;
import accord.utils.SortedList;
@@ -38,9 +37,8 @@ import static accord.topology.Topologies.SelectNodeOwnership.SHARE;
* Abstract parent class for implementing preaccept-like operations where we may need to fetch additional replies
* from future epochs.
*/
-abstract class AbstractCoordinatePreAccept<Result, Reply extends accord.messages.Reply, Ok> extends AbstractCoordination<Result, Reply, Ok> implements Callback<Reply>
+abstract class AbstractCoordinatePreAccept<Result, Reply extends accord.messages.Reply, Ok> extends AbstractCoordination<FullRoute<?>, Result, Reply, Ok> implements Callback<Reply>
{
- final FullRoute<?> route;
final Topologies topologies;
AbstractCoordinatePreAccept(Node node, SequentialAsyncExecutor executor, FullRoute<?> route, @Nonnull TxnId txnId, BiConsumer<? super Result, Throwable> callback)
@@ -50,8 +48,7 @@ abstract class AbstractCoordinatePreAccept<Result, Reply extends accord.messages
AbstractCoordinatePreAccept(Node node, SequentialAsyncExecutor executor, FullRoute<?> route, @Nonnull TxnId txnId, Topologies topologies, BiConsumer<? super Result, Throwable> callback)
{
- super(node, executor, txnId, topologies.nodes(), callback);
- this.route = route;
+ super(node, executor, txnId, route, topologies.nodes(), callback);
this.topologies = topologies;
}
@@ -68,7 +65,7 @@ abstract class AbstractCoordinatePreAccept<Result, Reply extends accord.messages
final void onPreAcceptedInNewEpoch(Topologies topologies, long latestEpoch)
{
- TopologyMismatch mismatch = TopologyMismatch.checkForMismatch(node.topology().globalForEpoch(latestEpoch), txnId, route.homeKey(), route);
+ TopologyMismatch mismatch = TopologyMismatch.checkForMismatch(node.topology().globalForEpoch(latestEpoch), txnId, scope.homeKey(), scope);
if (mismatch == null) onPreAccepted(topologies);
else onNewEpochTopologyMismatch(mismatch);
}
@@ -79,12 +76,6 @@ abstract class AbstractCoordinatePreAccept<Result, Reply extends accord.messages
return CoordinationKind.PreAccept;
}
- @Override
- public Unseekables<?> scope()
- {
- return route;
- }
-
@Override
public SortedList<Id> nodes()
{
diff --git a/accord-core/src/main/java/accord/coordinate/AbstractCoordination.java b/accord-core/src/main/java/accord/coordinate/AbstractCoordination.java
index 262d1752..1d5a813d 100644
--- a/accord-core/src/main/java/accord/coordinate/AbstractCoordination.java
+++ b/accord-core/src/main/java/accord/coordinate/AbstractCoordination.java
@@ -27,11 +27,13 @@ import java.util.function.Predicate;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
+import accord.api.Tracing;
import accord.coordinate.tracking.AbstractTracker;
import accord.local.Node;
import accord.local.SequentialAsyncExecutor;
import accord.messages.Callback;
import accord.messages.Request;
+import accord.primitives.Participants;
import accord.primitives.Route;
import accord.primitives.TxnId;
import accord.utils.DebugMap;
@@ -46,22 +48,25 @@ import accord.utils.async.AsyncChain;
import accord.utils.async.Cancellable;
// TODO (expected): move message sending here, so we can invalidate any pending callbacks when we advance the state machine
-public abstract class AbstractCoordination<Result, Reply extends accord.messages.Reply, Ok> extends AbstractSimpleCoordination implements Callback<Reply>
+public abstract class AbstractCoordination<P extends Participants<?>, Result, Reply extends accord.messages.Reply, Ok> extends AbstractSimpleCoordination<P> implements Callback<Reply>
{
- private BiConsumer<? super Result, Throwable> callback;
private final SortedArrayList<Node.Id> nodes;
private final SimpleBitSet expectingReply;
- private Object[] replyState;
+ protected final @Nullable Tracing tracing;
private final @Nullable DebugMap debug;
+
+ private BiConsumer<? super Result, Throwable> callback;
+ private Object[] replyState;
private int replyCount;
- protected AbstractCoordination(Node node, SequentialAsyncExecutor executor, TxnId txnId, SortedArrayList<Node.Id> nodes, BiConsumer<? super Result, Throwable> callback)
+ protected AbstractCoordination(Node node, SequentialAsyncExecutor executor, TxnId txnId, P participants, SortedArrayList<Node.Id> nodes, BiConsumer<? super Result, Throwable> callback)
{
- super(node, executor, txnId);
+ super(node, executor, txnId, participants);
this.nodes = nodes;
this.callback = Invariants.nonNull(callback);
this.replyState = new Object[nodes.size()];
this.expectingReply = SimpleBitSet.allocate(nodes.size());
+ this.tracing = node.agent().trace(txnId, participants, kind());
this.debug = Invariants.debug() ? new DebugMap(nodes) : null;
}
@@ -195,7 +200,11 @@ public abstract class AbstractCoordination<Result, Reply extends accord.messages
public final void onSlowResponse(Node.Id from)
{
if (!isDoneWithReplies())
+ {
+ if (tracing != null)
+ tracing.trace(null, "marking %s slow", from);
onSlowResponseInternal(from);
+ }
}
@Override
@@ -205,6 +214,12 @@ public abstract class AbstractCoordination<Result, Reply extends accord.messages
if (fromIndex < 0)
return;
+ if (tracing != null)
+ {
+ if (failure == null) tracing.trace(null, "timeout %s", from);
+ else tracing.trace(null, "received failure reply from %s: %s", from, failure);
+ }
+
onFailureInternal(from, fromIndex, failure);
Invariants.require(!expectingReply.isEmpty() || isFinishing() || isDone(), "%s", this);
}
@@ -222,6 +237,9 @@ public abstract class AbstractCoordination<Result, Reply extends accord.messages
if (debug != null)
debug.debug(from, reply);
+ if (tracing != null)
+ tracing.trace(null, "from %s: %s", from, reply);
+
if (isFinal)
{
boolean expecting = expectingReply.unset(fromIndex);
@@ -246,12 +264,15 @@ public abstract class AbstractCoordination<Result, Reply extends accord.messages
BiConsumer<?, Throwable> callback = tryTakeCallback();
if (callback != null) callback.accept(null, failure);
else node.agent().onUncaughtException(failure);
+ if (tracing != null)
+ tracing.trace(null, "callback failure processing from %s: %s", from, failure);
return true;
}
void finishAndInvokeCallback(Result success, Throwable failure)
{
finishAndTakeCallback().accept(success, failure);
+ if (failure != null) node.agent().coordinatorEvents().onFailed(failure, txnId, scope, this);
}
BiConsumer<? super Result, Throwable> finishAndTakeCallback()
diff --git a/accord-core/src/main/java/accord/coordinate/AbstractSimpleCoordination.java b/accord-core/src/main/java/accord/coordinate/AbstractSimpleCoordination.java
index c1e4de7e..798e4950 100644
--- a/accord-core/src/main/java/accord/coordinate/AbstractSimpleCoordination.java
+++ b/accord-core/src/main/java/accord/coordinate/AbstractSimpleCoordination.java
@@ -22,26 +22,29 @@ import javax.annotation.Nullable;
import accord.local.Node;
import accord.local.SequentialAsyncExecutor;
+import accord.primitives.Participants;
import accord.primitives.TxnId;
import accord.utils.Invariants;
// TODO (desired): move failure and callback handling here,
// so we can standardise and cleanup as we advance the state machines
-public abstract class AbstractSimpleCoordination implements Coordination
+public abstract class AbstractSimpleCoordination<P extends Participants<?>> implements Coordination
{
final long coordinationId;
protected final Node node;
protected final SequentialAsyncExecutor executor;
protected final TxnId txnId;
+ protected final P scope;
private Throwable failure;
private boolean isDoneWithReplies, isFinishing, isDone;
- protected AbstractSimpleCoordination(Node node, SequentialAsyncExecutor executor, TxnId txnId)
+ protected AbstractSimpleCoordination(Node node, SequentialAsyncExecutor executor, TxnId txnId, P scope)
{
this.coordinationId = node.nextCoordinationId();
this.node = node;
this.executor = executor;
this.txnId = Invariants.nonNull(txnId);
+ this.scope = scope;
}
@Override
@@ -56,6 +59,8 @@ public abstract class AbstractSimpleCoordination implements Coordination
return txnId;
}
+ public final P scope() { return scope; }
+
@Override
public final SequentialAsyncExecutor executor()
{
diff --git a/accord-core/src/main/java/accord/coordinate/AsynchronousAwait.java b/accord-core/src/main/java/accord/coordinate/AsynchronousAwait.java
index 4b6003aa..3a53d9e4 100644
--- a/accord-core/src/main/java/accord/coordinate/AsynchronousAwait.java
+++ b/accord-core/src/main/java/accord/coordinate/AsynchronousAwait.java
@@ -51,7 +51,7 @@ import static accord.coordinate.tracking.RequestStatus.Success;
*
* Asynchronous awaits will not time out if the wait is longer than message/request timeouts.
*/
-public class AsynchronousAwait extends AbstractCoordination<AsynchronousAwait.SynchronousResult, AwaitOk, Void> implements Callback<AwaitOk>
+public class AsynchronousAwait extends AbstractCoordination<Participants<?>, AsynchronousAwait.SynchronousResult, AwaitOk, Void> implements Callback<AwaitOk>
{
// TODO (desired, efficiency): this should collect the executeAt of any commit, and terminate as soon as one is found
// that is earlier than TxnId for the Txn we are recovering; if all commits we wait for
@@ -69,7 +69,6 @@ public class AsynchronousAwait extends AbstractCoordination<AsynchronousAwait.Sy
}
}
- final Participants<?> contact;
final AwaitTracker tracker;
final Await.Until until;
final int asynchronousCallbackId;
@@ -77,8 +76,7 @@ public class AsynchronousAwait extends AbstractCoordination<AsynchronousAwait.Sy
public AsynchronousAwait(Node node, SequentialAsyncExecutor executor, Participants<?> contact, TxnId txnId, AwaitTracker tracker, Await.Until until, boolean notifyProgressLog, int asynchronousCallbackId, BiConsumer<SynchronousResult, Throwable> synchronousCallback)
{
- super(node, executor, txnId, tracker.nodes(), synchronousCallback);
- this.contact = contact;
+ super(node, executor, txnId, contact, tracker.nodes(), synchronousCallback);
this.tracker = tracker;
this.until = until;
this.asynchronousCallbackId = asynchronousCallbackId;
@@ -107,7 +105,7 @@ public class AsynchronousAwait extends AbstractCoordination<AsynchronousAwait.Sy
void start()
{
super.start();
- contact(to -> new Await(to, tracker.topologies(), txnId, contact, until, asynchronousCallbackId, notifyProgressLog));
+ contact(to -> new Await(to, tracker.topologies(), txnId, scope, until, asynchronousCallbackId, notifyProgressLog));
}
@Override
@@ -136,8 +134,8 @@ public class AsynchronousAwait extends AbstractCoordination<AsynchronousAwait.Sy
private void onSuccess()
{
- Unseekables<?> ready = tracker.ready(contact);
- Unseekables<?> notReady = tracker.notReady(contact);
+ Unseekables<?> ready = tracker.ready(scope);
+ Unseekables<?> notReady = tracker.notReady(scope);
if (notReady.isEmpty())
notReady = null;
@@ -150,12 +148,6 @@ public class AsynchronousAwait extends AbstractCoordination<AsynchronousAwait.Sy
return CoordinationKind.AsyncAwait;
}
- @Override
- public Unseekables<?> scope()
- {
- return contact;
- }
-
@Override
public AbstractTracker<?> tracker()
{
diff --git a/accord-core/src/main/java/accord/coordinate/CheckShards.java b/accord-core/src/main/java/accord/coordinate/CheckShards.java
index 53b65998..1439e1eb 100644
--- a/accord-core/src/main/java/accord/coordinate/CheckShards.java
+++ b/accord-core/src/main/java/accord/coordinate/CheckShards.java
@@ -21,7 +21,6 @@ package accord.coordinate;
import java.util.function.BiConsumer;
import javax.annotation.Nullable;
-import accord.api.Tracing;
import accord.local.Node;
import accord.local.Node.Id;
import accord.local.SequentialAsyncExecutor;
@@ -52,7 +51,6 @@ public abstract class CheckShards<R, U extends Participants<?>> extends ReadCoor
final IncludeInfo includeInfo;
final @Nullable Ballot bumpBallot;
final Infer.InvalidIf previouslyKnownToBeInvalidIf;
- final @Nullable Tracing tracing;
protected CheckStatusOk merged;
protected boolean truncated;
@@ -63,25 +61,14 @@ public abstract class CheckShards<R, U extends Participants<?>> extends ReadCoor
this(node, executor, txnId, query, txnId.epoch(), includeInfo, bumpBallot, previouslyKnownToBeInvalidIf, callback);
}
- protected CheckShards(Node node, SequentialAsyncExecutor executor, TxnId txnId, U query, IncludeInfo includeInfo, @Nullable Ballot bumpBallot, Infer.InvalidIf previouslyKnownToBeInvalidIf, BiConsumer<? super R, Throwable> callback, @Nullable Tracing tracing)
- {
- this(node, executor, txnId, query, txnId.epoch(), includeInfo, bumpBallot, previouslyKnownToBeInvalidIf, callback, tracing);
- }
-
protected CheckShards(Node node, SequentialAsyncExecutor executor, TxnId txnId, U query, long srcEpoch, IncludeInfo includeInfo, @Nullable Ballot bumpBallot, Infer.InvalidIf previouslyKnownToBeInvalidIf, BiConsumer<? super R, Throwable> callback)
{
- this(node, executor, txnId, query, srcEpoch, includeInfo, bumpBallot, previouslyKnownToBeInvalidIf, callback, null);
- }
-
- protected CheckShards(Node node, SequentialAsyncExecutor executor, TxnId txnId, U query, long srcEpoch, IncludeInfo includeInfo, @Nullable Ballot bumpBallot, Infer.InvalidIf previouslyKnownToBeInvalidIf, BiConsumer<? super R, Throwable> callback, @Nullable Tracing tracing)
- {
- super(node, executor, topologyFor(node, txnId, query, srcEpoch), txnId, callback);
+ super(node, executor, topologyFor(node, txnId, query, srcEpoch), txnId, query, callback);
this.sourceEpoch = srcEpoch;
this.query = query;
this.includeInfo = includeInfo;
this.bumpBallot = bumpBallot;
this.previouslyKnownToBeInvalidIf = previouslyKnownToBeInvalidIf;
- this.tracing = tracing;
Invariants.require(txnId.isVisible());
}
@@ -95,8 +82,6 @@ public abstract class CheckShards<R, U extends Participants<?>> extends ReadCoor
protected void contact(Id id)
{
Participants<?> unseekables = query.slice(topologies().computeRangesForNode(id));
- if (tracing != null)
- tracing.trace(null, "%s contacting %s for %s", getClass().getSimpleName(), id, unseekables);
node.send(id, new CheckStatus(txnId, unseekables, sourceEpoch, includeInfo, bumpBallot), executor, this);
}
@@ -106,8 +91,6 @@ public abstract class CheckShards<R, U extends Participants<?>> extends ReadCoor
protected Action checkSufficient(Id from, CheckStatusOk ok)
{
Action action = isSufficient(from, ok) ? Action.Approve : Action.ApproveIfQuorum;
- if (tracing != null)
- tracing.trace(null, "%s %s reply %s from %s", getClass().getSimpleName(), action, ok, from);
return action;
}
@@ -123,9 +106,6 @@ public abstract class CheckShards<R, U extends Participants<?>> extends ReadCoor
}
else
{
- if (tracing != null)
- tracing.trace(null, "%s received failure reply %s from %s", getClass().getSimpleName(), reply, from);
-
switch ((CheckStatus.CheckStatusNack)reply)
{
default: throw new AssertionError(String.format("Unexpected status: %s", reply));
@@ -144,7 +124,7 @@ public abstract class CheckShards<R, U extends Participants<?>> extends ReadCoor
}
@Override
- public Unseekables<?> scope()
+ public Participants<?> scope()
{
return query;
}
diff --git a/accord-core/src/main/java/accord/coordinate/CollectLatestDeps.java b/accord-core/src/main/java/accord/coordinate/CollectLatestDeps.java
index 5df56b77..c967c162 100644
--- a/accord-core/src/main/java/accord/coordinate/CollectLatestDeps.java
+++ b/accord-core/src/main/java/accord/coordinate/CollectLatestDeps.java
@@ -47,9 +47,8 @@ import static accord.coordinate.tracking.RequestStatus.Failed;
import static accord.coordinate.tracking.RequestStatus.Success;
import static accord.primitives.Routables.Slice.Minimal;
-public class CollectLatestDeps extends AbstractCoordination<List<LatestDeps>, GetLatestDepsReply, GetLatestDepsOk>
+public class CollectLatestDeps extends AbstractCoordination<Route<?>, List<LatestDeps>, GetLatestDepsReply, GetLatestDepsOk>
{
- final Route<?> route;
final Timestamp executeAt;
final @Nullable Ballot ballot;
@@ -57,8 +56,7 @@ public class CollectLatestDeps extends AbstractCoordination<List<LatestDeps>, Ge
CollectLatestDeps(Node node, Topologies topologies, TxnId txnId, Route<?> route, @Nullable Ballot ballot, Timestamp executeAt, BiConsumer<List<LatestDeps>, Throwable> callback)
{
- super(node, node.someSequentialExecutor(), txnId, topologies.nodes(), callback);
- this.route = route;
+ super(node, node.someSequentialExecutor(), txnId, route, topologies.nodes(), callback);
this.executeAt = executeAt;
this.ballot = ballot;
this.tracker = new QuorumTracker(topologies);
@@ -83,7 +81,7 @@ public class CollectLatestDeps extends AbstractCoordination<List<LatestDeps>, Ge
else
{
super.start();
- contact(to -> new GetLatestDeps(to, tracker.topologies(), route, txnId, ballot, executeAt));
+ contact(to -> new GetLatestDeps(to, tracker.topologies(), scope, txnId, ballot, executeAt));
}
}
@@ -126,12 +124,6 @@ public class CollectLatestDeps extends AbstractCoordination<List<LatestDeps>, Ge
return CoordinationKind.CollectLatestDeps;
}
- @Override
- public Unseekables<?> scope()
- {
- return route;
- }
-
@Override
public SortedList<Id> nodes()
{
diff --git a/accord-core/src/main/java/accord/coordinate/CoordinateEphemeralRead.java b/accord-core/src/main/java/accord/coordinate/CoordinateEphemeralRead.java
index ee181ca5..4306de45 100644
--- a/accord-core/src/main/java/accord/coordinate/CoordinateEphemeralRead.java
+++ b/accord-core/src/main/java/accord/coordinate/CoordinateEphemeralRead.java
@@ -118,7 +118,7 @@ public class CoordinateEphemeralRead extends AbstractCoordinatePreAccept<Result,
void start()
{
super.start();
- contact(to -> new GetEphemeralReadDeps(to, topologies, route, txnId, executeAtEpoch));
+ contact(to -> new GetEphemeralReadDeps(to, topologies, scope, txnId, executeAtEpoch));
}
@Override
@@ -160,7 +160,7 @@ public class CoordinateEphemeralRead extends AbstractCoordinatePreAccept<Result,
private void retry()
{
- awaitEpochExactToFinish(retryInEpoch, () -> coordinate(node, route, txnId.withEpoch(retryInEpoch), txn, finishAndTakeCallback()));
+ awaitEpochExactToFinish(retryInEpoch, () -> coordinate(node, scope, txnId.withEpoch(retryInEpoch), txn, finishAndTakeCallback()));
}
@Override
@@ -174,12 +174,12 @@ public class CoordinateEphemeralRead extends AbstractCoordinatePreAccept<Result,
{
SortedListMap<Node.Id, GetEphemeralReadDepsOk> oks = finishOks();
Deps deps = Deps.merge(oks, oks.domainSize(), SortedListMap::getValue, ok -> ok.deps);
- topologies = node.topology().reselect(topologies, QuorumEpochIntersections.preaccept.include, route, executeAtEpoch, executeAtEpoch, SHARE, Owned);
+ topologies = node.topology().reselect(topologies, QuorumEpochIntersections.preaccept.include, scope, executeAtEpoch, executeAtEpoch, SHARE, Owned);
CoordinationFlags flags = oks.foldlNonNull((d, k, v, out) -> {
ExecuteFlags.collect(out, k, v.flags, d, v.deps);
return out;
}, deps, empty(oks.domain()));
- new ExecuteEphemeralRead(node, executor, topologies, route, txnId.withEpoch(executeAtEpoch), txn, deps, flags, finishAndTakeCallback()).start();
+ new ExecuteEphemeralRead(node, executor, topologies, scope, txnId.withEpoch(executeAtEpoch), txn, deps, flags, finishAndTakeCallback()).start();
if (!Invariants.debug()) oks.clear();
}
diff --git a/accord-core/src/main/java/accord/coordinate/CoordinateMaxConflict.java b/accord-core/src/main/java/accord/coordinate/CoordinateMaxConflict.java
index b66531cc..8ae42af8 100644
--- a/accord-core/src/main/java/accord/coordinate/CoordinateMaxConflict.java
+++ b/accord-core/src/main/java/accord/coordinate/CoordinateMaxConflict.java
@@ -105,7 +105,7 @@ public class CoordinateMaxConflict extends AbstractCoordinatePreAccept<Timestamp
void start()
{
super.start();
- contact(to -> new GetMaxConflict(to, topologies, route, executionEpoch));
+ contact(to -> new GetMaxConflict(to, topologies, scope, executionEpoch));
}
@Override
diff --git a/accord-core/src/main/java/accord/coordinate/CoordinatePreAccept.java b/accord-core/src/main/java/accord/coordinate/CoordinatePreAccept.java
index 2ae88576..fab939fb 100644
--- a/accord-core/src/main/java/accord/coordinate/CoordinatePreAccept.java
+++ b/accord-core/src/main/java/accord/coordinate/CoordinatePreAccept.java
@@ -78,12 +78,12 @@ abstract class CoordinatePreAccept<T> extends AbstractCoordinatePreAccept<T, Pre
void contact(@Nullable Deps deps, boolean hasCoordinatorVote)
{
- contact(to -> new PreAccept(to, topologies, txnId, txn, deps, hasCoordinatorVote, route));
+ contact(to -> new PreAccept(to, topologies, txnId, txn, deps, hasCoordinatorVote, scope));
}
void contactNotSelf(@Nullable Deps deps, boolean hasCoordinatorVote)
{
- contact(to -> new PreAccept(to, topologies, txnId, txn, deps, hasCoordinatorVote, route), id -> !id.equals(node.id()));
+ contact(to -> new PreAccept(to, topologies, txnId, txn, deps, hasCoordinatorVote, scope), id -> !id.equals(node.id()));
}
@Override
@@ -115,7 +115,7 @@ abstract class CoordinatePreAccept<T> extends AbstractCoordinatePreAccept<T, Pre
if (!reply.isOk())
{
// we've been preempted by a recovery coordinator; defer to it, and wait to hear any result
- finishWithFailureOverride(Preempted.preempted(node.agent(), txnId, route.homeKey()));
+ finishWithFailureOverride(Preempted.preempted(node.agent(), txnId, scope.homeKey()));
}
else
{
@@ -143,7 +143,7 @@ abstract class CoordinatePreAccept<T> extends AbstractCoordinatePreAccept<T, Pre
* participating keys/ranges, so we propose that the transaction is invalidated in its coordination epoch
*/
BiConsumer<? super T, Throwable> callback = finishAndTakeCallback();
- proposeInvalidate(node, executor, node.uniqueTimestamp(Ballot::fromValues), txnId, route.homeKey(), (outcome, failure) -> {
+ proposeInvalidate(node, executor, node.uniqueTimestamp(Ballot::fromValues), txnId, scope.homeKey(), (outcome, failure) -> {
if (failure != null)
mismatch.addSuppressed(failure);
callback.accept(null, mismatch);
diff --git a/accord-core/src/main/java/accord/coordinate/CoordinateSyncPoint.java b/accord-core/src/main/java/accord/coordinate/CoordinateSyncPoint.java
index 851e2613..5fe8159f 100644
--- a/accord-core/src/main/java/accord/coordinate/CoordinateSyncPoint.java
+++ b/accord-core/src/main/java/accord/coordinate/CoordinateSyncPoint.java
@@ -170,7 +170,7 @@ public class CoordinateSyncPoint<R> extends CoordinatePreAccept<R>
if (executeAt.is(REJECTED))
{
node.agent().coordinatorEvents().onRejected(txnId);
- proposeAndCommitInvalidate(node, executor, Ballot.ZERO, txnId, route.homeKey(), route, executeAt, finishAndTakeCallback());
+ proposeAndCommitInvalidate(node, executor, Ballot.ZERO, txnId, scope.homeKey(), scope, executeAt, finishAndTakeCallback());
}
else
{
@@ -181,9 +181,9 @@ public class CoordinateSyncPoint<R> extends CoordinatePreAccept<R>
Deps deps = Deps.merge(oks.valuesAsNullableList(), oks.domainSize(), List::get, ok -> ok.deps);
node.agent().coordinatorEvents().onPreAccepted(txnId);
if (tracker.hasMediumPathAccepted() && txnId.hasMediumPath())
- adapter.propose(node, executor, topologies, route, Accept.Kind.MEDIUM, Ballot.ZERO, txnId, txn, withFlags, deps, finishAndTakeCallback());
+ adapter.propose(node, executor, topologies, scope, Accept.Kind.MEDIUM, Ballot.ZERO, txnId, txn, withFlags, deps, finishAndTakeCallback());
else
- adapter.propose(node, executor, topologies, route, Accept.Kind.SLOW, Ballot.ZERO, txnId, txn, executeAt, deps, finishAndTakeCallback());
+ adapter.propose(node, executor, topologies, scope, Accept.Kind.SLOW, Ballot.ZERO, txnId, txn, executeAt, deps, finishAndTakeCallback());
}
}
diff --git a/accord-core/src/main/java/accord/coordinate/CoordinateTransaction.java b/accord-core/src/main/java/accord/coordinate/CoordinateTransaction.java
index 42aa3683..1065ccd0 100644
--- a/accord-core/src/main/java/accord/coordinate/CoordinateTransaction.java
+++ b/accord-core/src/main/java/accord/coordinate/CoordinateTransaction.java
@@ -43,7 +43,6 @@ import accord.messages.PreAccept.PreAcceptNack;
import accord.messages.PreAccept.PreAcceptReply;
import accord.primitives.Route;
import accord.primitives.Status;
-import accord.primitives.Unseekables;
import accord.topology.Topologies;
import accord.local.Node;
import accord.messages.PreAccept.PreAcceptOk;
@@ -147,7 +146,7 @@ public class CoordinateTransaction extends CoordinatePreAccept<Result>
// we must include Deps from fast path votes from earlier epochs that may have witnessed later transactions
// TODO (desired): we might mask some bugs by merging more responses than we strictly need, so optimise this to optionally merge minimal deps
node.agent().coordinatorEvents().onPreAccepted(txnId);
- executeAdapter().execute(node, executor, topologies, route, Ballot.ZERO, FAST, flags, txnId, txn, txnId, deps, deps, finishAndTakeCallback());
+ executeAdapter().execute(node, executor, topologies, scope, Ballot.ZERO, FAST, flags, txnId, txn, txnId, deps, deps, finishAndTakeCallback());
return;
}
}
@@ -157,20 +156,20 @@ public class CoordinateTransaction extends CoordinatePreAccept<Result>
if (deps != null)
{
node.agent().coordinatorEvents().onPreAccepted(txnId);
- proposeAdapter().propose(node, executor, topologies, route, MEDIUM, Ballot.ZERO, txnId, txn, txnId, deps, finishAndTakeCallback());
+ proposeAdapter().propose(node, executor, topologies, scope, MEDIUM, Ballot.ZERO, txnId, txn, txnId, deps, finishAndTakeCallback());
return;
}
}
else if (executeAt.is(REJECTED))
{
- proposeAndCommitInvalidate(node, executor, Ballot.ZERO, txnId, route.homeKey(), route, executeAt, finishAndTakeCallback());
+ proposeAndCommitInvalidate(node, executor, Ballot.ZERO, txnId, scope.homeKey(), scope, executeAt, finishAndTakeCallback());
node.agent().coordinatorEvents().onRejected(txnId);
return;
}
Deps deps = Deps.merge(oks.valuesAsNullableList(), oks.domainSize(), List::get, ok -> ok.deps);
node.agent().coordinatorEvents().onPreAccepted(txnId);
- proposeAdapter().propose(node, executor, topologies, route, SLOW, Ballot.ZERO, txnId, txn, executeAt, deps, finishAndTakeCallback());
+ proposeAdapter().propose(node, executor, topologies, scope, SLOW, Ballot.ZERO, txnId, txn, executeAt, deps, finishAndTakeCallback());
}
private Deps mergeFastOrMediumDeps(SortedListMap<?, PreAcceptOk> oks)
@@ -211,7 +210,7 @@ public class CoordinateTransaction extends CoordinatePreAccept<Result>
protected LocalExecute()
{
- super(route);
+ super(CoordinateTransaction.this.scope);
}
void start()
@@ -263,7 +262,7 @@ public class CoordinateTransaction extends CoordinatePreAccept<Result>
}
else
{
- finishWithFailureOverride(Preempted.preempted(node.agent(), txnId, route.homeKey()));
+ finishWithFailureOverride(Preempted.preempted(node.agent(), txnId, scope.homeKey()));
}
}
});
@@ -273,7 +272,7 @@ public class CoordinateTransaction extends CoordinatePreAccept<Result>
public PreAcceptReply applyInternal(SafeCommandStore safeStore)
{
long minEpoch = topologies.oldestEpoch();
- StoreParticipants participants = StoreParticipants.update(safeStore, route, minEpoch, txnId, txnId.epoch());
+ StoreParticipants participants = StoreParticipants.update(safeStore, scope, minEpoch, txnId, txnId.epoch());
SafeCommand safeCommand = safeStore.get(txnId, participants);
ExecuteFlags flags;
@@ -384,11 +383,5 @@ public class CoordinateTransaction extends CoordinatePreAccept<Result>
{
return LoadKeysFor.READ_WRITE;
}
-
- @Override
- public Unseekables<?> keys()
- {
- return route;
- }
}
}
diff --git a/accord-core/src/main/java/accord/coordinate/Coordination.java b/accord-core/src/main/java/accord/coordinate/Coordination.java
index 55764c60..51c413fb 100644
--- a/accord-core/src/main/java/accord/coordinate/Coordination.java
+++ b/accord-core/src/main/java/accord/coordinate/Coordination.java
@@ -24,34 +24,41 @@ import accord.coordinate.tracking.AbstractTracker;
import accord.local.Node.Id;
import accord.local.SequentialAsyncExecutor;
import accord.primitives.Ballot;
+import accord.primitives.Participants;
import accord.primitives.TxnId;
-import accord.primitives.Unseekables;
import accord.utils.SortedList;
import accord.utils.SortedListMap;
import accord.utils.TinyEnumSet;
public interface Coordination
{
+ // approximately ordered, that is we should ordinarily move forwards in the state machine
enum CoordinationKind
{
- PreAccept, Propose, ProposeInvalidate, Stabilise, Execute, ExecuteSyncPoint, Persist,
- MaybeRecover, PrepareRecovery, BeginRecovery, RecoverAwait, BeginInvalidate,
- AsyncAwait, SyncAwait,
+ HomeProgress, WaitProgress, // not strictly coordinations, but for tracing simplicity we treat them as such here (maybe make this a more general enum)
Fetch, FetchRoute,
- CollectLatestDeps, Bootstrap,
- Other;
+ BeginInvalidate, MaybeRecover, PrepareRecovery, BeginRecovery, RecoverAwait, CollectLatestDeps,
+ PreAccept, Propose, ProposeInvalidate, Stabilise, Execute, ExecuteSyncPoint, Persist,
+ AsyncAwait, SyncAwait, Bootstrap, Other;
+ private static final CoordinationKind[] LOOKUP = values();
public static final TinyEnumSet<CoordinationKind> COORDINATES_STATE_MACHINE = TinyEnumSet.of(
PreAccept, Propose, ProposeInvalidate, Stabilise, Execute, ExecuteSyncPoint, Persist,
MaybeRecover, PrepareRecovery, BeginRecovery, RecoverAwait, BeginInvalidate
);
+ public static final TinyEnumSet<CoordinationKind> ALL = TinyEnumSet.of(LOOKUP);
+
+ public static CoordinationKind forOrdinal(int ordinal)
+ {
+ return LOOKUP[ordinal];
+ }
}
long coordinationId();
TxnId txnId();
CoordinationKind kind();
- Unseekables<?> scope();
+ Participants<?> scope();
default @Nullable Ballot ballot() { return null; }
diff --git a/accord-core/src/main/java/accord/coordinate/ExecuteEphemeralRead.java b/accord-core/src/main/java/accord/coordinate/ExecuteEphemeralRead.java
index 3e3bb369..1729b346 100644
--- a/accord-core/src/main/java/accord/coordinate/ExecuteEphemeralRead.java
+++ b/accord-core/src/main/java/accord/coordinate/ExecuteEphemeralRead.java
@@ -44,10 +44,10 @@ import accord.messages.SafeCallback;
import accord.messages.StableThenRead;
import accord.primitives.Deps;
import accord.primitives.FullRoute;
+import accord.primitives.Participants;
import accord.primitives.Ranges;
import accord.primitives.Txn;
import accord.primitives.TxnId;
-import accord.primitives.Unseekables;
import accord.topology.Topologies;
import accord.utils.Invariants;
import accord.utils.UnhandledEnum;
@@ -77,7 +77,7 @@ public class ExecuteEphemeralRead extends ReadCoordinator<Result, ReadReply>
{
// we need to send Stable to the origin epoch as well as the execution epoch
// TODO (desired): permit slicing Topologies by key (though unnecessary if we eliminate the concept of non-participating home keys)
- super(node, executor, topologies, txnId, callback);
+ super(node, executor, topologies, txnId, route, callback);
Invariants.requireArgument(txnId.kind() == EphemeralRead);
Invariants.require(topologies.currentEpoch() == txnId.epoch());
this.txn = txn;
@@ -252,7 +252,7 @@ public class ExecuteEphemeralRead extends ReadCoordinator<Result, ReadReply>
}
@Override
- public Unseekables<?> scope()
+ public Participants<?> scope()
{
return route;
}
diff --git a/accord-core/src/main/java/accord/coordinate/ExecuteFlag.java b/accord-core/src/main/java/accord/coordinate/ExecuteFlag.java
index e81ab6ba..10bc9749 100644
--- a/accord-core/src/main/java/accord/coordinate/ExecuteFlag.java
+++ b/accord-core/src/main/java/accord/coordinate/ExecuteFlag.java
@@ -28,9 +28,14 @@ public enum ExecuteFlag
{
READY_TO_EXECUTE, HAS_UNIQUE_HLC;
+ private static final ExecuteFlag[] LOOKUP = values();
+ public static ExecuteFlag forOrdinal(int ordinal)
+ {
+ return LOOKUP[ordinal];
+ }
+
public static final class ExecuteFlags extends TinyEnumSet<ExecuteFlag>
{
- private static final ExecuteFlag[] UNIVERSE = ExecuteFlag.values();
private static final ExecuteFlags[] LOOKUP = new ExecuteFlags[1 << ExecuteFlag.values().length];
static
{
@@ -46,14 +51,13 @@ public enum ExecuteFlag
public ExecuteFlags without(ExecuteFlag a) { return LOOKUP[bitset & ~encode(a)]; }
public ExecuteFlags or(ExecuteFlags that) { return LOOKUP[this.bitset | that.bitset]; }
public ExecuteFlags and(ExecuteFlags that) { return LOOKUP[this.bitset & that.bitset]; }
- public boolean isEmpty() { return bitset == 0; }
public int bits() { return bitset; }
private ExecuteFlags(int bits) { super(bits); }
@Override
public String toString()
{
- return toString(UNIVERSE);
+ return toString(ExecuteFlag::forOrdinal);
}
public static void collect(CoordinationFlags into, Node.Id id, ExecuteFlags add, Object expectIfReadyToExecute, Object actualReadyToExecute)
diff --git a/accord-core/src/main/java/accord/coordinate/ExecuteSyncPoint.java b/accord-core/src/main/java/accord/coordinate/ExecuteSyncPoint.java
index 23d31b13..c4962126 100644
--- a/accord-core/src/main/java/accord/coordinate/ExecuteSyncPoint.java
+++ b/accord-core/src/main/java/accord/coordinate/ExecuteSyncPoint.java
@@ -40,10 +40,10 @@ import accord.messages.ReadData;
import accord.messages.ReadData.CommitOrReadNack;
import accord.messages.ReadData.ReadReply;
import accord.messages.SetShardDurable;
+import accord.primitives.FullRoute;
import accord.primitives.Range;
import accord.primitives.SyncPoint;
import accord.primitives.Txn;
-import accord.primitives.Unseekables;
import accord.topology.Topologies;
import accord.utils.SortedArrays.SortedArrayList;
import accord.utils.UnhandledEnum;
@@ -56,7 +56,7 @@ import static accord.primitives.Status.Durability.HasOutcome.Quorum;
import static accord.primitives.Status.Durability.HasOutcome.Universal;
import static accord.topology.Topologies.SelectNodeOwnership.SHARE;
-public class ExecuteSyncPoint extends AbstractCoordination<DurabilityResult, ReadReply, Void> implements Callback<ReadReply>
+public class ExecuteSyncPoint extends AbstractCoordination<FullRoute<?>, DurabilityResult, ReadReply, Void> implements Callback<ReadReply>
{
public static class SyncPointErased extends Throwable implements WrappableException<SyncPointErased>
{
@@ -116,7 +116,7 @@ public class ExecuteSyncPoint extends AbstractCoordination<DurabilityResult, Rea
ExecuteSyncPoint(Node node, SyncPoint<Range> syncPoint, Topologies topologies, Set<Node.Id> excludeSuccess, SequentialAsyncExecutor executor, int attempt, DurabilityResult partialResult, DurabilityResults callback)
{
- super(node, executor, syncPoint.syncId, topologies.nodes(), callback);
+ super(node, executor, syncPoint.syncId, syncPoint.route, topologies.nodes(), callback);
this.syncPoint = syncPoint;
this.partialResult = partialResult;
this.excludeSuccess = excludeSuccess;
@@ -317,12 +317,6 @@ public class ExecuteSyncPoint extends AbstractCoordination<DurabilityResult, Rea
return CoordinationKind.ExecuteSyncPoint;
}
- @Override
- public Unseekables<?> scope()
- {
- return syncPoint.route;
- }
-
@Override
public AbstractTracker<?> tracker()
{
diff --git a/accord-core/src/main/java/accord/coordinate/ExecuteTxn.java b/accord-core/src/main/java/accord/coordinate/ExecuteTxn.java
index 43a8679a..270139b2 100644
--- a/accord-core/src/main/java/accord/coordinate/ExecuteTxn.java
+++ b/accord-core/src/main/java/accord/coordinate/ExecuteTxn.java
@@ -60,7 +60,6 @@ import accord.primitives.Timestamp;
import accord.primitives.TimestampWithUniqueHlc;
import accord.primitives.Txn;
import accord.primitives.TxnId;
-import accord.primitives.Unseekables;
import accord.primitives.Writes;
import accord.topology.Topologies;
import accord.utils.Invariants;
@@ -163,7 +162,7 @@ public class ExecuteTxn extends ReadCoordinator<Result, ReadReply>
ExecuteTxn(Node node, SequentialAsyncExecutor executor, Topologies topologies, FullRoute<?> route, Ballot ballot, ExecutePath path, CoordinationFlags flags, TxnId txnId, Txn txn, Timestamp executeAt, Deps stableDeps, Deps sendDeps, BiConsumer<? super Result, Throwable> callback)
{
- super(node, executor, topologies.forEpoch(executeAt.epoch()), txnId, callback);
+ super(node, executor, topologies.forEpoch(executeAt.epoch()), txnId, route, callback);
this.path = ballot == Ballot.ZERO ? path : RECOVER;
this.txn = txn;
this.route = route;
@@ -565,7 +564,7 @@ public class ExecuteTxn extends ReadCoordinator<Result, ReadReply>
}
@Override
- public Unseekables<?> scope()
+ public Participants<?> scope()
{
return route;
}
diff --git a/accord-core/src/main/java/accord/coordinate/FetchCoordinator.java b/accord-core/src/main/java/accord/coordinate/FetchCoordinator.java
index 934ef38a..2bc3c217 100644
--- a/accord-core/src/main/java/accord/coordinate/FetchCoordinator.java
+++ b/accord-core/src/main/java/accord/coordinate/FetchCoordinator.java
@@ -25,9 +25,9 @@ import accord.api.DataStore.StartingRangeFetch;
import accord.api.DataStore.FetchRanges;
import accord.local.Node;
import accord.local.SequentialAsyncExecutor;
+import accord.primitives.FullRoute;
import accord.primitives.Ranges;
import accord.primitives.SyncPoint;
-import accord.primitives.Unseekables;
import accord.topology.Topology;
import accord.utils.Invariants;
import accord.utils.SortedList;
@@ -43,7 +43,7 @@ import static accord.utils.Invariants.illegalState;
* This coordinates both the fetching of data, and the point-in-time that data will be safe to read,
* i.e. some point-in-time known to occur after any entry in the data that was fetched.
*/
-public abstract class FetchCoordinator extends AbstractSimpleCoordination
+public abstract class FetchCoordinator extends AbstractSimpleCoordination<FullRoute<?>>
{
/**
* For each node, maintain what ranges have been requested, successfully retrieved or not, and whether we are
@@ -145,7 +145,7 @@ public abstract class FetchCoordinator extends AbstractSimpleCoordination
protected FetchCoordinator(Node node, SequentialAsyncExecutor executor, Ranges ranges, SyncPoint syncPoint, FetchRanges fetchRanges)
{
- super(node, executor, syncPoint.syncId);
+ super(node, executor, syncPoint.syncId, syncPoint.route);
this.ranges = ranges;
this.syncPoint = syncPoint;
this.fetchRanges = fetchRanges;
@@ -336,12 +336,6 @@ public abstract class FetchCoordinator extends AbstractSimpleCoordination
trySendMore();
}
- @Override
- public Unseekables<?> scope()
- {
- return syncPoint.route;
- }
-
@Override
public CoordinationKind kind()
{
diff --git a/accord-core/src/main/java/accord/coordinate/FetchData.java b/accord-core/src/main/java/accord/coordinate/FetchData.java
index 12c06a6e..98beaeec 100644
--- a/accord-core/src/main/java/accord/coordinate/FetchData.java
+++ b/accord-core/src/main/java/accord/coordinate/FetchData.java
@@ -40,7 +40,6 @@ import accord.utils.Invariants;
import javax.annotation.Nonnull;
-import static accord.api.TraceEventType.FETCH;
import static accord.coordinate.Infer.InvalidIf.NotKnownToBeInvalid;
/**
@@ -99,7 +98,6 @@ public class FetchData extends CheckShards<FetchData.FetchResult, Route<?>>
@Override
public void accept(FetchResult success, Throwable fail)
{
- if (fail != null && tracing != null) tracing.trace(null, "Failed fetch: %s", Tracing.format(fail));
callback.accept(success, fail);
}
}
@@ -117,7 +115,7 @@ public class FetchData extends CheckShards<FetchData.FetchResult, Route<?>>
*/
public static Object fetchSpecific(Known fetch, Node node, TxnId txnId, InvalidIf invalidIf, @Nullable Timestamp executeAt, Route<?> query, Route<?> maxRoute, LatentStoreSelector reportTo, BiConsumer<? super FetchResult, Throwable> callback)
{
- return fetchSpecific(node, query, maxRoute, new FetchRequest(node.someSequentialExecutor(), fetch, txnId, invalidIf, executeAt, maxRoute, reportTo, callback, node.agent().trace(txnId, FETCH)));
+ return fetchSpecific(node, query, maxRoute, new FetchRequest(node.someSequentialExecutor(), fetch, txnId, invalidIf, executeAt, maxRoute, reportTo, callback, null));
}
public static Object fetchSpecific(Node node, Route<?> query, Route<?> maxRoute, FetchRequest request)
@@ -143,15 +141,15 @@ public class FetchData extends CheckShards<FetchData.FetchResult, Route<?>>
// (i.e. if preaccept/accept contact a later epoch than execution is decided for)
final LatentStoreSelector reportTo;
- private FetchData(Node node, Known target, TxnId txnId, InvalidIf invalidIf, Route<?> route, Route<?> maxRoute, long sourceEpoch, LatentStoreSelector reportTo, BiConsumer<? super FetchResult, Throwable> callback, @Nullable Tracing tracing)
+ private FetchData(Node node, Known target, TxnId txnId, InvalidIf invalidIf, Route<?> route, Route<?> maxRoute, long sourceEpoch, LatentStoreSelector reportTo, BiConsumer<? super FetchResult, Throwable> callback)
{
- this(node, target, txnId, invalidIf, route, route.withHomeKey(), maxRoute, sourceEpoch, reportTo, callback, tracing);
+ this(node, target, txnId, invalidIf, route, route.withHomeKey(), maxRoute, sourceEpoch, reportTo, callback);
}
- private FetchData(Node node, Known target, TxnId txnId, InvalidIf invalidIf, Route<?> route, Route<?> routeWithHomeKey, Route<?> maxRoute, long sourceEpoch, LatentStoreSelector reportTo, BiConsumer<? super FetchResult, Throwable> callback, @Nullable Tracing tracing)
+ private FetchData(Node node, Known target, TxnId txnId, InvalidIf invalidIf, Route<?> route, Route<?> routeWithHomeKey, Route<?> maxRoute, long sourceEpoch, LatentStoreSelector reportTo, BiConsumer<? super FetchResult, Throwable> callback)
{
// TODO (desired, efficiency): restore behaviour of only collecting info if e.g. Committed or Executed
- super(node, node.someSequentialExecutor(), txnId, routeWithHomeKey, sourceEpoch, CheckStatus.IncludeInfo.All, null, invalidIf, callback, tracing);
+ super(node, node.someSequentialExecutor(), txnId, routeWithHomeKey, sourceEpoch, CheckStatus.IncludeInfo.All, null, invalidIf, callback);
this.reportTo = reportTo;
this.maxRoute = maxRoute;
Invariants.requireArgument(routeWithHomeKey.contains(route.homeKey()), "route %s does not contain %s", routeWithHomeKey, route.homeKey());
@@ -161,14 +159,14 @@ public class FetchData extends CheckShards<FetchData.FetchResult, Route<?>>
private static FetchData fetchData(Node node, Route<?> route, Route<?> maxRoute, FetchRequest req)
{
Invariants.require(!req.contactable.isEmpty());
- FetchData fetch = new FetchData(node, req.fetch, req.txnId, req.invalidIf, route, maxRoute, req.srcEpoch, req.reportTo, req.callback, req.tracing);
+ FetchData fetch = new FetchData(node, req.fetch, req.txnId, req.invalidIf, route, maxRoute, req.srcEpoch, req.reportTo, req.callback);
fetch.start();
return fetch;
}
private static FetchData fetchData(Node node, Known fetch, TxnId txnId, InvalidIf invalidIf, Route<?> route, Route<?> maxRoute, long sourceEpoch, StoreSelector reportTo, BiConsumer<? super FetchResult, Throwable> callback)
{
- FetchData fetchData = new FetchData(node, fetch, txnId, invalidIf, route, maxRoute, sourceEpoch, reportTo, callback, node.agent().trace(txnId, FETCH));
+ FetchData fetchData = new FetchData(node, fetch, txnId, invalidIf, route, maxRoute, sourceEpoch, reportTo, callback);
fetchData.start();
return fetchData;
}
@@ -203,8 +201,6 @@ public class FetchData extends CheckShards<FetchData.FetchResult, Route<?>>
Invariants.require((success == null) != (failure == null));
if (failure != null)
{
- if (tracing != null)
- tracing.trace(null, "%s completed with failure %s", getClass().getSimpleName(), Tracing.format(failure));
invokeCallback(null, failure);
}
else
diff --git a/accord-core/src/main/java/accord/coordinate/FetchRoute.java b/accord-core/src/main/java/accord/coordinate/FetchRoute.java
index f8e00e7a..ab16b01d 100644
--- a/accord-core/src/main/java/accord/coordinate/FetchRoute.java
+++ b/accord-core/src/main/java/accord/coordinate/FetchRoute.java
@@ -37,7 +37,6 @@ import accord.primitives.Participants;
import accord.primitives.Route;
import accord.primitives.TxnId;
-import static accord.api.TraceEventType.FETCH;
import static accord.coordinate.Infer.InvalidIf.NotKnownToBeInvalid;
import static accord.coordinate.Infer.InvalidateAndCallback.locallyInvalidateAndCallback;
import static accord.local.CommandStores.*;
@@ -51,29 +50,35 @@ public class FetchRoute extends CheckShards<Route<?>, Participants<?>>
{
final LatentStoreSelector reportTo;
- FetchRoute(Node node, TxnId txnId, Infer.InvalidIf invalidIf, Participants<?> contactable, LatentStoreSelector reportTo, BiConsumer<Route<?>, Throwable> callback, @Nullable Tracing tracing)
+ FetchRoute(Node node, TxnId txnId, Infer.InvalidIf invalidIf, Participants<?> contactable, LatentStoreSelector reportTo, BiConsumer<Route<?>, Throwable> callback)
{
- super(node, node.someSequentialExecutor(), txnId, contactable, txnId.epoch(), IncludeInfo.Route, null, invalidIf, callback, tracing);
+ super(node, node.someSequentialExecutor(), txnId, contactable, txnId.epoch(), IncludeInfo.Route, null, invalidIf, callback);
this.reportTo = reportTo;
}
- public static Object fetchRoute(Node node, TxnId txnId, Infer.InvalidIf invalidIf, Participants<?> unseekables, LatentStoreSelector reportTo, BiConsumer<Route<?>, Throwable> callback, Tracing tracing)
+ public static Object fetchRoute(Node node, TxnId txnId, Infer.InvalidIf invalidIf, Participants<?> participants, LatentStoreSelector reportTo, BiConsumer<Route<?>, Throwable> callback)
+ {
+ Tracing tracing = node.agent().trace(txnId, participants, CoordinationKind.FetchRoute);
+ return fetchRoute(node, txnId, invalidIf, participants, reportTo, callback, tracing);
+ }
+
+ private static Object fetchRoute(Node node, TxnId txnId, Infer.InvalidIf invalidIf, Participants<?> participants, LatentStoreSelector reportTo, BiConsumer<Route<?>, Throwable> callback, @Nullable Tracing tracing)
{
if (!node.topology().hasEpoch(txnId.epoch()))
{
if (tracing != null)
tracing.trace(null, "Waiting for epoch %d", txnId.epoch());
- return node.withEpochAtLeast(txnId.epoch(), null, callback, () -> fetchRoute(node, txnId, invalidIf, unseekables, reportTo, callback, tracing));
+ return node.withEpochAtLeast(txnId.epoch(), null, callback, () -> fetchRoute(node, txnId, invalidIf, participants, reportTo, callback, tracing));
}
- FetchRoute fetchRoute = new FetchRoute(node, txnId, invalidIf, unseekables, reportTo, callback, tracing);
+ FetchRoute fetchRoute = new FetchRoute(node, txnId, invalidIf, participants, reportTo, callback);
fetchRoute.start();
return fetchRoute;
}
public static Object fetchRoute(Node node, TxnId txnId, Participants<?> contactable, LatentStoreSelector reportTo, BiConsumer<Route<?>, Throwable> callback)
{
- return fetchRoute(node, txnId, NotKnownToBeInvalid, contactable, reportTo, callback, node.agent().trace(txnId, FETCH));
+ return fetchRoute(node, txnId, NotKnownToBeInvalid, contactable, reportTo, callback, null);
}
@Override
diff --git a/accord-core/src/main/java/accord/coordinate/Infer.java b/accord-core/src/main/java/accord/coordinate/Infer.java
index a50ba81d..08c473f9 100644
--- a/accord-core/src/main/java/accord/coordinate/Infer.java
+++ b/accord-core/src/main/java/accord/coordinate/Infer.java
@@ -113,9 +113,9 @@ public class Infer
final StoreSelector reportTo;
final T param;
final BiConsumer<T, Throwable> callback;
- final Tracing tracing;
+ final @Nullable Tracing tracing;
- private CleanupAndCallback(Node node, TxnId txnId, StoreSelector reportTo, Participants<?> participants, T param, BiConsumer<T, Throwable> callback, Tracing tracing)
+ private CleanupAndCallback(Node node, TxnId txnId, StoreSelector reportTo, Participants<?> participants, T param, BiConsumer<T, Throwable> callback, @Nullable Tracing tracing)
{
super(participants);
this.node = node;
diff --git a/accord-core/src/main/java/accord/coordinate/Invalidate.java b/accord-core/src/main/java/accord/coordinate/Invalidate.java
index bfd471f4..6e4f134e 100644
--- a/accord-core/src/main/java/accord/coordinate/Invalidate.java
+++ b/accord-core/src/main/java/accord/coordinate/Invalidate.java
@@ -42,7 +42,6 @@ import accord.utils.UnhandledEnum;
import javax.annotation.Nullable;
-import static accord.api.TraceEventType.RECOVER;
import static accord.coordinate.Infer.InvalidIf.NotKnownToBeInvalid;
import static accord.coordinate.Propose.NotAccept.proposeInvalidate;
import static accord.primitives.Status.AcceptedMedium;
@@ -52,10 +51,9 @@ import static accord.primitives.ProgressToken.TRUNCATED_DURABLE_OR_INVALIDATED;
import static accord.topology.Topologies.SelectNodeOwnership.SHARE;
import static accord.utils.Invariants.illegalState;
-public class Invalidate extends AbstractCoordination<Outcome, InvalidateReply, InvalidateReply>
+public class Invalidate extends AbstractCoordination<Participants<?>, Outcome, InvalidateReply, InvalidateReply>
{
private final Ballot ballot;
- private final Participants<?> invalidateWith;
private final boolean transitivelyInvokedByPriorInvalidation;
private final InvalidationTracker tracker;
@@ -67,10 +65,9 @@ public class Invalidate extends AbstractCoordination<Outcome, InvalidateReply, I
}
private Invalidate(Node node, SequentialAsyncExecutor executor, Topologies topologies, Ballot ballot, TxnId txnId, Participants<?> invalidateWith, boolean transitivelyInvokedByPriorInvalidation, LatentStoreSelector reportTo, BiConsumer<? super Outcome, Throwable> callback)
{
- super(node, executor, txnId, topologies.nodes(), callback);
+ super(node, executor, txnId, invalidateWith, topologies.nodes(), callback);
Invariants.require(topologies.size() == 1);
this.ballot = ballot;
- this.invalidateWith = invalidateWith;
this.transitivelyInvokedByPriorInvalidation = transitivelyInvokedByPriorInvalidation;
this.reportTo = reportTo;
this.tracker = new InvalidationTracker(topologies, txnId);
@@ -105,7 +102,7 @@ public class Invalidate extends AbstractCoordination<Outcome, InvalidateReply, I
else
{
super.start();
- contact(to -> new BeginInvalidation(to, tracker.topologies(), txnId, invalidateWith, ballot));
+ contact(to -> new BeginInvalidation(to, tracker.topologies(), txnId, scope, ballot));
}
}
@@ -114,7 +111,7 @@ public class Invalidate extends AbstractCoordination<Outcome, InvalidateReply, I
{
recordOk(fromIndex, reply);
Participants<?> truncated = reply.truncated;
- Participants<?> notTruncated = truncated == null ? invalidateWith : invalidateWith.without(truncated);
+ Participants<?> notTruncated = truncated == null ? scope : scope.without(truncated);
Participants<?> promised = reply.isPromiseRejected() ? null : notTruncated;
handle(tracker.recordSuccess(from, promised, notTruncated, truncated, reply.hasDecision(), reply.acceptedFastPath));
}
@@ -183,7 +180,7 @@ public class Invalidate extends AbstractCoordination<Outcome, InvalidateReply, I
case Stable:
case Committed:
case PreCommitted:
- Invariants.require(maxNotTruncated.maxKnowledgeStatus.status == PreAccepted || !invalidateWith.contains(someRoute.homeKey()) || fullRoute != null);
+ Invariants.require(maxNotTruncated.maxKnowledgeStatus.status == PreAccepted || !scope.contains(someRoute.homeKey()) || fullRoute != null);
case AcceptedMedium:
case AcceptedSlow:
@@ -205,11 +202,10 @@ public class Invalidate extends AbstractCoordination<Outcome, InvalidateReply, I
if (!witnessedByInvalidation.hasBeen(AcceptedMedium))
{
Invariants.require(tracker.all(InvalidationShardTracker::isPromised));
- if (!invalidateWith.containsAll(fullRoute))
+ if (!scope.containsAll(fullRoute))
witnessedByInvalidation = null;
}
- PrepareRecovery.recover(node, executor, txnId, NotKnownToBeInvalid, fullRoute, witnessedByInvalidation, reportTo, finishAndTakeCallback(),
- node.agent().trace(txnId, RECOVER));
+ PrepareRecovery.recover(node, executor, txnId, NotKnownToBeInvalid, fullRoute, witnessedByInvalidation, reportTo, finishAndTakeCallback());
return;
case Invalidated:
@@ -244,7 +240,7 @@ public class Invalidate extends AbstractCoordination<Outcome, InvalidateReply, I
// Probably simplest to do so, but perhaps better for user if we don't.
Ranges ranges = Ranges.of(tracker.promisedShard().range);
// we look up by TxnId at the target node, so it's fine to pick a RoutingKey even if it's a range transaction
- RoutingKey someKey = invalidateWith.slice(ranges).get(0).someIntersectingRoutingKey(ranges);
+ RoutingKey someKey = scope.slice(ranges).get(0).someIntersectingRoutingKey(ranges);
BiConsumer<? super Outcome, Throwable> callback = finishAndTakeCallback();
proposeInvalidate(node, executor, ballot, txnId, someKey, (success, fail) -> {
/*
@@ -263,12 +259,12 @@ public class Invalidate extends AbstractCoordination<Outcome, InvalidateReply, I
try
{
@Nullable Route<?> route = InvalidateReply.mergeRoutes(oks.valuesAsNullableList());
- if (route == null && Route.isRoute(invalidateWith)) route = Route.castToRoute(invalidateWith);
+ if (route == null && Route.isRoute(scope)) route = Route.castToRoute(scope);
if (route != null) route = route.withHomeKey();
// TODO (desired, efficiency): commitInvalidate (and others) should skip the network for local applications,
// so we do not need to explicitly do so here before notifying the waiter
- Participants<?> commitTo = Participants.merge(route, (Participants) invalidateWith);
+ Participants<?> commitTo = Participants.merge(route, (Participants) scope);
Commit.Invalidate.commitInvalidate(node, txnId, commitTo, txnId);
commitInvalidateLocal(commitTo, reportTo.refine(txnId, null, commitTo), callback);
}
@@ -326,12 +322,6 @@ public class Invalidate extends AbstractCoordination<Outcome, InvalidateReply, I
return CoordinationKind.BeginInvalidate;
}
- @Override
- public Unseekables<?> scope()
- {
- return invalidateWith;
- }
-
@Override
public Ballot ballot()
{
diff --git a/accord-core/src/main/java/accord/coordinate/MaybeRecover.java b/accord-core/src/main/java/accord/coordinate/MaybeRecover.java
index 4e72a036..171f92da 100644
--- a/accord-core/src/main/java/accord/coordinate/MaybeRecover.java
+++ b/accord-core/src/main/java/accord/coordinate/MaybeRecover.java
@@ -20,7 +20,6 @@ package accord.coordinate;
import java.util.function.BiConsumer;
-import accord.api.Tracing;
import accord.local.CommandStores.LatentStoreSelector;
import accord.local.SequentialAsyncExecutor;
import accord.messages.InformDurable;
@@ -32,7 +31,6 @@ import accord.messages.CheckStatus.CheckStatusOk;
import accord.messages.CheckStatus.IncludeInfo;
import accord.utils.UnhandledEnum;
-import static accord.api.TraceEventType.RECOVER;
import static accord.coordinate.Infer.InvalidateAndCallback.locallyInvalidateAndCallback;
import static accord.messages.Commit.Invalidate.commitInvalidate;
import static accord.primitives.Known.Outcome.Unknown;
@@ -50,7 +48,7 @@ public class MaybeRecover extends CheckShards<Outcome, Route<?>>
MaybeRecover(Node node, SequentialAsyncExecutor executor, TxnId txnId, Infer.InvalidIf invalidIf, Route<?> someRoute, ProgressToken prevProgress, LatentStoreSelector reportTo, BiConsumer<? super Outcome, Throwable> callback)
{
// we only want to enquire with the home shard, but we prefer maximal route information for running Invalidation against, if necessary
- super(node, executor, txnId, someRoute.withHomeKey(), IncludeInfo.Route, null, invalidIf, callback, node.agent().trace(txnId, RECOVER));
+ super(node, executor, txnId, someRoute.withHomeKey(), IncludeInfo.Route, null, invalidIf, callback);
this.prevProgress = prevProgress;
this.reportTo = reportTo;
}
@@ -82,8 +80,6 @@ public class MaybeRecover extends CheckShards<Outcome, Route<?>>
// this can be helpful in mitigating flakiness and helping forward progress for large transactions spanning many shards
if (fail != null)
{
- if (tracing != null)
- tracing.trace(null, "MaybeRecover failed: " + Tracing.format(fail));
invokeCallback(null, fail);
}
else
@@ -91,7 +87,7 @@ public class MaybeRecover extends CheckShards<Outcome, Route<?>>
Invariants.require(merged != null);
CheckStatusOk full = merged.finish(this.query, this.query, this.query, success.withQuorum, previouslyKnownToBeInvalidIf);
if (tracing != null)
- tracing.trace(null, "MaybeRecover merged: " + full);
+ tracing.trace(null, "merged: " + full);
Known known = full.maxKnown();
Route<?> someRoute = full.route;
@@ -113,7 +109,7 @@ public class MaybeRecover extends CheckShards<Outcome, Route<?>>
if (success.withQuorum == HasQuorum && known.canProposeInvalidation() && !Route.isFullRoute(full.route))
{
if (tracing != null)
- tracing.trace(null, "MaybeRecover found quorum permitting invalidation: " + known);
+ tracing.trace(null, "found quorum permitting invalidation: " + known);
// for correctness reasons, we have not necessarily preempted the initial pre-accept round and
// may have raced with it, so we must attempt to recover anything we see pre-accepted.
@@ -121,7 +117,7 @@ public class MaybeRecover extends CheckShards<Outcome, Route<?>>
break;
}
if (tracing != null)
- tracing.trace(null, "MaybeRecover found quorum of Unknown that did not permit invalidation; falling through.");
+ tracing.trace(null, "found quorum of Unknown that did not permit invalidation; falling through.");
// fall through otherwise to recovery
}
case WasApply:
@@ -130,7 +126,7 @@ public class MaybeRecover extends CheckShards<Outcome, Route<?>>
{
ProgressToken progressToken = full.toProgressToken();
if (tracing != null)
- tracing.trace(null, "MaybeRecover found %s which need not be recovered; reporting %s", known.outcome(), progressToken);
+ tracing.trace(null, "found %s which need not be recovered; reporting %s", known.outcome(), progressToken);
invokeCallback(progressToken, null);
break;
}
@@ -142,7 +138,7 @@ public class MaybeRecover extends CheckShards<Outcome, Route<?>>
{
ProgressToken progressToken = full.toProgressToken();
if (tracing != null)
- tracing.trace(null, "MaybeRecover found %s; reporting progress token %s", hasMadeProgress(full) ? "progress" : "no route", progressToken);
+ tracing.trace(null, "found %s; reporting progress token %s", hasMadeProgress(full) ? "progress" : "no route", progressToken);
if (full.durability.isDurable())
InformDurable.informDefault(node, topologies, txnId, query, bumpBallot, full.executeAtIfKnown(), full.durability);
invokeCallback(full.toProgressToken(), null);
@@ -151,8 +147,8 @@ public class MaybeRecover extends CheckShards<Outcome, Route<?>>
{
Invariants.expect(!full.durability.isDurableOrInvalidated());
if (tracing != null)
- tracing.trace(null, "MaybeRecover invoking RecoverWithRoute");
- node.recover(txnId, full.invalidIf, Route.castToFullRoute(someRoute), reportTo, tracing).begin(takeCallback());
+ tracing.trace(null, "invoking RecoverWithRoute");
+ node.recover(txnId, full.invalidIf, Route.castToFullRoute(someRoute), reportTo).begin(takeCallback());
}
break;
}
@@ -169,14 +165,14 @@ public class MaybeRecover extends CheckShards<Outcome, Route<?>>
}
ProgressToken progressToken = full.toProgressToken();
if (tracing != null)
- tracing.trace(null, "MaybeRecover found %s which cannot be recovered; reporting %s", known.outcome(), progressToken);
+ tracing.trace(null, "found %s which cannot be recovered; reporting %s", known.outcome(), progressToken);
invokeCallback(progressToken, null);
break;
}
case Abort:
{
if (tracing != null)
- tracing.trace(null, "MaybeRecover found Abort; invalidating locally", known.outcome());
+ tracing.trace(null, "found Abort; invalidating locally");
commitInvalidate(node, txnId, Route.merge(full.route, (Route) query), txnId.epoch());
locallyInvalidateAndCallback(node, txnId, txnId.epoch(), txnId.epoch(), someRoute, full.toProgressToken(), takeCallback(), null);
diff --git a/accord-core/src/main/java/accord/coordinate/Persist.java b/accord-core/src/main/java/accord/coordinate/Persist.java
index abd43238..80edbc9d 100644
--- a/accord-core/src/main/java/accord/coordinate/Persist.java
+++ b/accord-core/src/main/java/accord/coordinate/Persist.java
@@ -36,7 +36,6 @@ import accord.primitives.Route;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.TxnId;
-import accord.primitives.Unseekables;
import accord.primitives.Writes;
import accord.topology.Topologies;
import accord.utils.Invariants;
@@ -47,7 +46,7 @@ import static accord.messages.Apply.Kind.Maximal;
import static accord.primitives.Status.Durability.AllQuorums;
import static accord.topology.Topologies.SelectNodeOwnership.SHARE;
-public abstract class Persist extends AbstractCoordination<Void, ApplyReply, Void>
+public abstract class Persist extends AbstractCoordination<FullRoute<?>, Void, ApplyReply, Void>
{
protected final Ballot ballot;
protected final Route<?> sendTo;
@@ -56,7 +55,6 @@ public abstract class Persist extends AbstractCoordination<Void, ApplyReply, Voi
protected final Deps stableDeps;
protected final Writes writes;
protected final Result result;
- protected final FullRoute<?> route;
protected final CoordinationFlags flags;
// TODO (expected): track separate ALL and Quorum, so we can report Universal durability to permit faster GC
protected final QuorumTracker tracker;
@@ -66,7 +64,7 @@ public abstract class Persist extends AbstractCoordination<Void, ApplyReply, Voi
protected Persist(Node node, SequentialAsyncExecutor executor, Topologies all, TxnId txnId, Ballot ballot, Route<?> sendTo, Txn txn, Timestamp executeAt, Deps stableDeps, Writes writes, Result result, FullRoute<?> route, CoordinationFlags flags, boolean informDurableOnDone, Apply.Factory factory, Apply.Kind applyKind)
{
- super(node, executor, txnId, all.nodes(), node.agent());
+ super(node, executor, txnId, route, all.nodes(), node.agent());
this.ballot = ballot;
this.sendTo = sendTo;
this.txn = txn;
@@ -74,7 +72,6 @@ public abstract class Persist extends AbstractCoordination<Void, ApplyReply, Voi
this.stableDeps = stableDeps;
this.writes = writes;
this.result = result;
- this.route = route;
this.flags = flags;
this.tracker = new QuorumTracker(all);
this.factory = factory;
@@ -98,7 +95,7 @@ public abstract class Persist extends AbstractCoordination<Void, ApplyReply, Voi
// but we make this explicit for the caller with informDurableOnDone
finishWithSuccess(null);
if (informDurableOnDone)
- InformDurable.informDefault(node, tracker.topologies(), txnId, route, ballot, executeAt, AllQuorums);
+ InformDurable.informDefault(node, tracker.topologies(), txnId, scope, ballot, executeAt, AllQuorums);
}
break;
case RaceWithRecovery:
@@ -110,12 +107,12 @@ public abstract class Persist extends AbstractCoordination<Void, ApplyReply, Voi
break;
case Insufficient:
Invariants.expect(applyKind != Maximal, "Received Insufficient reply from %s, but already sent Maximal", from);
- node.send(from, factory.create(Maximal, from, tracker.topologies(), txnId, ballot, sendTo, txn, executeAt, stableDeps, writes, result, route, flags.get(from)));
+ node.send(from, factory.create(Maximal, from, tracker.topologies(), txnId, ballot, sendTo, txn, executeAt, stableDeps, writes, result, scope, flags.get(from)));
break;
case InsufficientEpochs:
Invariants.requireArgument(txnId.isSyncPoint());
- node.send(from, factory.create(Maximal, from, node.topology().preciseEpochs(route, reply.minEpoch(), tracker.topologies().currentEpoch(), SHARE) , txnId, ballot, sendTo, txn, executeAt, stableDeps, writes, result, route, flags.get(from)));
+ node.send(from, factory.create(Maximal, from, node.topology().preciseEpochs(scope, reply.minEpoch(), tracker.topologies().currentEpoch(), SHARE) , txnId, ballot, sendTo, txn, executeAt, stableDeps, writes, result, scope, flags.get(from)));
break;
}
}
@@ -143,7 +140,7 @@ public abstract class Persist extends AbstractCoordination<Void, ApplyReply, Voi
else
{
super.start();
- contact(to -> factory.create(applyKind, to, tracker.topologies(), txnId, ballot, sendTo, txn, executeAt, stableDeps, writes, result, route, flags.get(to)));
+ contact(to -> factory.create(applyKind, to, tracker.topologies(), txnId, ballot, sendTo, txn, executeAt, stableDeps, writes, result, scope, flags.get(to)));
}
}
@@ -153,12 +150,6 @@ public abstract class Persist extends AbstractCoordination<Void, ApplyReply, Voi
return CoordinationKind.Persist;
}
- @Override
- public Unseekables<?> scope()
- {
- return route;
- }
-
@Override
public AbstractTracker<?> tracker()
{
diff --git a/accord-core/src/main/java/accord/coordinate/PrepareRecovery.java b/accord-core/src/main/java/accord/coordinate/PrepareRecovery.java
index 95051b07..c5588e26 100644
--- a/accord-core/src/main/java/accord/coordinate/PrepareRecovery.java
+++ b/accord-core/src/main/java/accord/coordinate/PrepareRecovery.java
@@ -21,7 +21,6 @@ package accord.coordinate;
import java.util.function.BiConsumer;
import javax.annotation.Nullable;
-import accord.api.Tracing;
import accord.local.CommandStores.LatentStoreSelector;
import accord.coordinate.ExecuteFlag.CoordinationFlags;
import accord.local.Node;
@@ -67,9 +66,9 @@ public class PrepareRecovery extends CheckShards<Outcome, FullRoute<?>>
final Status witnessedByInvalidation;
final LatentStoreSelector reportTo;
- private PrepareRecovery(Node node, SequentialAsyncExecutor executor, Topologies topologies, TxnId txnId, Infer.InvalidIf invalidIf, FullRoute<?> route, Status witnessedByInvalidation, LatentStoreSelector reportTo, BiConsumer<? super Outcome, Throwable> callback, @Nullable Tracing tracing)
+ private PrepareRecovery(Node node, SequentialAsyncExecutor executor, Topologies topologies, TxnId txnId, Infer.InvalidIf invalidIf, FullRoute<?> route, Status witnessedByInvalidation, LatentStoreSelector reportTo, BiConsumer<? super Outcome, Throwable> callback)
{
- super(node, executor, txnId, route, IncludeInfo.All, node.uniqueTimestamp(Ballot::fromValues), invalidIf, callback, tracing);
+ super(node, executor, txnId, route, IncludeInfo.All, node.uniqueTimestamp(Ballot::fromValues), invalidIf, callback);
this.reportTo = reportTo;
// if witnessedByInvalidation == AcceptedInvalidate then we cannot assume its definition was known, and our comparison with the status is invalid
Invariants.require(witnessedByInvalidation != Status.AcceptedInvalidate);
@@ -79,14 +78,14 @@ public class PrepareRecovery extends CheckShards<Outcome, FullRoute<?>>
assert topologies.oldestEpoch() == topologies.currentEpoch() && topologies.currentEpoch() == txnId.epoch();
}
- public static PrepareRecovery recover(Node node, SequentialAsyncExecutor executor, TxnId txnId, Infer.InvalidIf invalidIf, FullRoute<?> route, @Nullable Status witnessedByInvalidation, LatentStoreSelector reportTo, BiConsumer<? super Outcome, Throwable> callback, @Nullable Tracing tracing)
+ public static PrepareRecovery recover(Node node, SequentialAsyncExecutor executor, TxnId txnId, Infer.InvalidIf invalidIf, FullRoute<?> route, @Nullable Status witnessedByInvalidation, LatentStoreSelector reportTo, BiConsumer<? super Outcome, Throwable> callback)
{
- return recover(node, executor, node.topology().forEpoch(route, txnId.epoch(), SHARE), txnId, invalidIf, route, witnessedByInvalidation, reportTo, callback, tracing);
+ return recover(node, executor, node.topology().forEpoch(route, txnId.epoch(), SHARE), txnId, invalidIf, route, witnessedByInvalidation, reportTo, callback);
}
- private static PrepareRecovery recover(Node node, SequentialAsyncExecutor executor, Topologies topologies, TxnId txnId, Infer.InvalidIf invalidIf, FullRoute<?> route, @Nullable Status witnessedByInvalidation, LatentStoreSelector reportTo, BiConsumer<? super Outcome, Throwable> callback, @Nullable Tracing tracing)
+ private static PrepareRecovery recover(Node node, SequentialAsyncExecutor executor, Topologies topologies, TxnId txnId, Infer.InvalidIf invalidIf, FullRoute<?> route, @Nullable Status witnessedByInvalidation, LatentStoreSelector reportTo, BiConsumer<? super Outcome, Throwable> callback)
{
- PrepareRecovery recover = new PrepareRecovery(node, executor, topologies, txnId, invalidIf, route, witnessedByInvalidation, reportTo, callback, tracing);
+ PrepareRecovery recover = new PrepareRecovery(node, executor, topologies, txnId, invalidIf, route, witnessedByInvalidation, reportTo, callback);
recover.start();
return recover;
}
@@ -130,8 +129,6 @@ public class PrepareRecovery extends CheckShards<Outcome, FullRoute<?>>
{
if (failure != null)
{
- if (tracing != null)
- tracing.trace(null, "RecoverWithRoute failed: " + Tracing.format(failure));
invokeCallback(null, failure);
return;
}
@@ -139,7 +136,7 @@ public class PrepareRecovery extends CheckShards<Outcome, FullRoute<?>>
CheckStatusOkFull full = ((CheckStatusOkFull) this.merged).finish(query, query, query, success.withQuorum, previouslyKnownToBeInvalidIf);
Known known = full.knownFor(txnId, query, query);
if (tracing != null)
- tracing.trace(null, "RecoverWithRoute merged: " + full);
+ tracing.trace(null, "merged: " + full);
// TODO (required): audit this logic, and centralise with e.g. FetchData inferences
// TODO (expected): skip straight to ExecuteTxn if we have a Stable reply from each shard
@@ -151,15 +148,15 @@ public class PrepareRecovery extends CheckShards<Outcome, FullRoute<?>>
if (known.definition().isKnown())
{
if (tracing != null)
- tracing.trace(null, "RecoverWithRoute found definition; invoking Recover.");
+ tracing.trace(null, "found definition; invoking Recover.");
Txn txn = full.partialTxn.reconstitute(query);
- Recover.recover(node, txnId, txn, query, full.durability.isFastPathDurablyDecided(), reportTo, takeCallback(), tracing);
+ Recover.recover(node, txnId, txn, query, full.durability.isFastPathDurablyDecided(), reportTo, takeCallback());
}
else if (!known.definition().isOrWasKnown())
{
if (tracing != null)
- tracing.trace(null, "RecoverWithRoute found no current or erased transaction; invoking Invalidate.");
+ tracing.trace(null, "found no current or erased transaction; invoking Invalidate.");
if (witnessedByInvalidation != null && witnessedByInvalidation.compareTo(Status.PreAccepted) > 0)
throw illegalState("We previously invalidated, finding a status that should be recoverable");
@@ -170,7 +167,7 @@ public class PrepareRecovery extends CheckShards<Outcome, FullRoute<?>>
{
ProgressToken progressToken = full.toProgressToken();
if (tracing != null)
- tracing.trace(null, "RecoverWithRoute found insufficient information to Recover or Invalidate; calling back with %s.", progressToken);
+ tracing.trace(null, "found insufficient information to Recover or Invalidate; calling back with %s.", progressToken);
invokeCallback(progressToken, null);
}
break;
@@ -183,7 +180,7 @@ public class PrepareRecovery extends CheckShards<Outcome, FullRoute<?>>
if (!known.isTruncated() && !known.isInvalidated())
{
if (tracing != null)
- tracing.trace(null, "RecoverWithRoute found Apply/WasApply, but no definition, truncation or invalidation; must have raced with Apply, reporting no progress in expectation next attempt is successful.");
+ tracing.trace(null, "found Apply/WasApply, but no definition, truncation or invalidation; must have raced with Apply, reporting no progress in expectation next attempt is successful.");
// we must have raced with a successful apply, so should simply abort
invokeCallback(ProgressToken.NONE, null);
@@ -202,7 +199,7 @@ public class PrepareRecovery extends CheckShards<Outcome, FullRoute<?>>
if (known.isInvalidated())
{
if (tracing != null)
- tracing.trace(null, "RecoverWithRoute found partially truncated Invalidate; committing to shards " + trySendTo);
+ tracing.trace(null, "found partially truncated Invalidate; committing to shards " + trySendTo);
Commit.Invalidate.commitInvalidate(node, txnId, trySendTo, txnId);
}
else
@@ -214,7 +211,7 @@ public class PrepareRecovery extends CheckShards<Outcome, FullRoute<?>>
if (!known.is(DepsKnown))
{
if (tracing != null)
- tracing.trace(null, "RecoverWithRoute found partially truncated Apply with incomplete Deps; advancing state machine for shards " + trySendTo);
+ tracing.trace(null, "found partially truncated Apply with incomplete Deps; advancing state machine for shards " + trySendTo);
Invariants.require(txnId.isSystemTxn() || full.partialTxn.covers(trySendTo));
Participants<?> haveStable = full.map.knownFor(Known.DepsOnly, trySendTo);
@@ -229,7 +226,7 @@ public class PrepareRecovery extends CheckShards<Outcome, FullRoute<?>>
else
{
if (tracing != null)
- tracing.trace(null, "RecoverWithRoute found partially truncated Apply; persisting to shards " + trySendTo);
+ tracing.trace(null, "found partially truncated Apply; persisting to shards " + trySendTo);
Invariants.require(full.stableDeps.covers(trySendTo));
Invariants.require(txnId.isSystemTxn() || full.partialTxn.covers(trySendTo));
@@ -242,7 +239,7 @@ public class PrepareRecovery extends CheckShards<Outcome, FullRoute<?>>
else
{
if (tracing != null)
- tracing.trace(null, "RecoverWithRoute found Apply truncated at all shards; advancing Durability to at least Majority");
+ tracing.trace(null, "found Apply truncated at all shards; advancing Durability to at least Majority");
propagate = full.merge(AllQuorums);
}
@@ -264,7 +261,7 @@ public class PrepareRecovery extends CheckShards<Outcome, FullRoute<?>>
if (known.is(DepsKnown))
{
if (tracing != null)
- tracing.trace(null, "RecoverWithRoute found Apply with DepsKnown; persisting to all shards.");
+ tracing.trace(null, "found Apply with DepsKnown; persisting to all shards.");
deps = full.stableDeps.reconstitute(query);
missingDeps = query.slice(0, 0);
}
@@ -273,7 +270,7 @@ public class PrepareRecovery extends CheckShards<Outcome, FullRoute<?>>
Participants<?> hasDeps = full.map.knownFor(Known.DepsOnly, query);
missingDeps = query.without(hasDeps);
if (tracing != null)
- tracing.trace(null, "RecoverWithRoute found Apply, with deps missing for %s; advancing state machine for these shards.", missingDeps);
+ tracing.trace(null, "found Apply, with deps missing for %s; advancing state machine for these shards.", missingDeps);
if (full.stableDeps == null)
{
@@ -298,16 +295,16 @@ public class PrepareRecovery extends CheckShards<Outcome, FullRoute<?>>
else
{
if (tracing != null)
- tracing.trace(null, "RecoverWithRoute found %s; invoking Recover", known);
+ tracing.trace(null, "found %s; invoking Recover", known);
- Recover.recover(node, txnId, txn, query, full.durability.isFastPathDurablyDecided(), takeCallback(), tracing);
+ Recover.recover(node, txnId, txn, query, full.durability.isFastPathDurablyDecided(), takeCallback());
}
break;
}
case Abort:
{
if (tracing != null)
- tracing.trace(null, "RecoverWithRoute Found Abort; propagating locally.");
+ tracing.trace(null, "Found Abort; propagating locally.");
if (witnessedByInvalidation != null && witnessedByInvalidation.hasBeen(Status.PreCommitted))
throw illegalState("We previously invalidated, finding a status that should be recoverable");
@@ -318,7 +315,7 @@ public class PrepareRecovery extends CheckShards<Outcome, FullRoute<?>>
case Erased:
{
if (tracing != null)
- tracing.trace(null, "RecoverWithRoute found Erased; propagating locally.");
+ tracing.trace(null, "found Erased; propagating locally.");
// we should only be able to hit the Erased case if every participating shard has advanced past this TxnId, so we don't need to recover it
Propagate.propagate(node, txnId, previouslyKnownToBeInvalidIf, sourceEpoch, success.withQuorum, query, query, reportTo, null, full, (s, f) -> invokeCallback(f == null ? TRUNCATED_DURABLE_OR_INVALIDATED : null, f), tracing);
diff --git a/accord-core/src/main/java/accord/coordinate/Propose.java b/accord-core/src/main/java/accord/coordinate/Propose.java
index 59ed6942..1e0aec04 100644
--- a/accord-core/src/main/java/accord/coordinate/Propose.java
+++ b/accord-core/src/main/java/accord/coordinate/Propose.java
@@ -42,7 +42,6 @@ import accord.primitives.Status;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.TxnId;
-import accord.primitives.Unseekables;
import accord.topology.Topologies;
import accord.utils.Invariants;
import accord.utils.SortedArrays;
@@ -58,12 +57,11 @@ import static accord.primitives.Status.AcceptedInvalidate;
import static accord.primitives.TxnId.MediumPath.TrackStable;
import static accord.topology.Topologies.SelectNodeOwnership.SHARE;
-abstract class Propose<R> extends AbstractCoordination<R, AcceptReply, AcceptReply>
+abstract class Propose<R> extends AbstractCoordination<FullRoute<?>, R, AcceptReply, AcceptReply>
{
final Accept.Kind kind;
final Ballot ballot;
final Txn txn;
- final FullRoute<?> route;
final Route<?> require;
final Deps deps;
@@ -72,12 +70,11 @@ abstract class Propose<R> extends AbstractCoordination<R, AcceptReply, AcceptRep
Propose(Node node, SequentialAsyncExecutor executor, Topologies topologies, Accept.Kind kind, Ballot ballot, TxnId txnId, Txn txn, Route<?> require, FullRoute<?> route, Timestamp executeAt, Deps deps, BiConsumer<? super R, Throwable> callback)
{
- super(node, executor, txnId, topologies.nodes(), callback);
+ super(node, executor, txnId, route, topologies.nodes(), callback);
this.kind = kind;
this.ballot = ballot;
this.txn = txn;
this.require = require;
- this.route = route;
this.deps = deps;
this.executeAt = executeAt;
this.tracker = new QuorumTracker(topologies);
@@ -97,7 +94,7 @@ abstract class Propose<R> extends AbstractCoordination<R, AcceptReply, AcceptRep
else
{
super.start();
- contact(to -> new Accept(to, tracker.topologies(), kind, ballot, txnId, route, executeAt, deps, require != route));
+ contact(to -> new Accept(to, tracker.topologies(), kind, ballot, txnId, scope, executeAt, deps, require != scope));
}
}
@@ -108,23 +105,23 @@ abstract class Propose<R> extends AbstractCoordination<R, AcceptReply, AcceptRep
{
default: throw new AssertionError("Unhandled AcceptOutcome: " + reply.outcome());
case RejectedBallot:
- finishWithFailureOverride(Preempted.preempted(node.agent(), txnId, route.homeKey()));
+ finishWithFailureOverride(Preempted.preempted(node.agent(), txnId, scope.homeKey()));
break;
case Truncated:
case Redundant:
- if (require == route || !isSufficientPartialReply(reply, from))
+ if (require == scope || !isSufficientPartialReply(reply, from))
{
Throwable failNow = null;
if (reply.outcome == AcceptOutcome.Truncated)
- failNow = new Truncated(txnId, route.homeKey());
+ failNow = new Truncated(txnId, scope.homeKey());
else if (reply.supersededBy != null || ballot.equals(Ballot.ZERO))
- failNow = Preempted.preempted(node.agent(), txnId, route.homeKey());
+ failNow = Preempted.preempted(node.agent(), txnId, scope.homeKey());
if (failNow != null)
finishWithFailureOverride(failNow);
else
- onFailureInternal(from, fromIndex, reply.committedExecuteAt == null ? null : new Redundant(txnId, route.homeKey(), reply.committedExecuteAt));
+ onFailureInternal(from, fromIndex, reply.committedExecuteAt == null ? null : new Redundant(txnId, scope.homeKey(), reply.committedExecuteAt));
break;
}
@@ -158,8 +155,8 @@ abstract class Propose<R> extends AbstractCoordination<R, AcceptReply, AcceptRep
Deps newDeps = mergeNewDeps();
Deps deps = mergeDeps(newDeps);
node.agent().coordinatorEvents().onAccepted(txnId, ballot);
- if (kind == Accept.Kind.MEDIUM) adapter().execute(node, executor, tracker.topologies(), route, ballot, ExecutePath.MEDIUM, CoordinationFlags.none(), txnId, txn, executeAt, deps, newDeps, finishAndTakeCallback());
- else adapter().stabilise(node, executor, tracker.topologies(), route, ballot, txnId, txn, executeAt, deps, finishAndTakeCallback());
+ if (kind == Accept.Kind.MEDIUM) adapter().execute(node, executor, tracker.topologies(), scope, ballot, ExecutePath.MEDIUM, CoordinationFlags.none(), txnId, txn, executeAt, deps, newDeps, finishAndTakeCallback());
+ else adapter().stabilise(node, executor, tracker.topologies(), scope, ballot, txnId, txn, executeAt, deps, finishAndTakeCallback());
}
Deps mergeDeps()
@@ -200,12 +197,6 @@ abstract class Propose<R> extends AbstractCoordination<R, AcceptReply, AcceptRep
return CoordinationKind.Propose;
}
- @Override
- public Unseekables<?> scope()
- {
- return route;
- }
-
@Override
public Ballot ballot()
{
@@ -225,30 +216,28 @@ abstract class Propose<R> extends AbstractCoordination<R, AcceptReply, AcceptRep
}
// A special version for proposing the invalidation of a transaction; only needs to succeed on one shard
- static class NotAccept extends AbstractCoordination<Void, AcceptReply, Void> implements Callback<AcceptReply>
+ static class NotAccept extends AbstractCoordination<Participants<?>, Void, AcceptReply, Void> implements Callback<AcceptReply>
{
final Status status;
final Ballot ballot;
final TxnId txnId;
- final Participants<?> someParticipants;
private final SimpleTracker<?> tracker;
NotAccept(Node node, SequentialAsyncExecutor executor, Status status, Topologies topologies, Ballot ballot, TxnId txnId, Participants<?> someParticipants, BiConsumer<Void, Throwable> callback)
{
- super(node, executor, txnId, topologies.nodes(), callback);
+ super(node, executor, txnId, someParticipants, topologies.nodes(), callback);
this.status = status;
this.tracker = new QuorumTracker(topologies);
this.ballot = ballot;
this.txnId = txnId;
- this.someParticipants = someParticipants;
}
@Override
void start()
{
super.start();
- contact(to -> new Accept.NotAccept(status, ballot, txnId, someParticipants));
+ contact(to -> new Accept.NotAccept(status, ballot, txnId, scope));
}
public static void proposeInvalidate(Node node, SequentialAsyncExecutor executor, Ballot ballot, TxnId txnId, RoutingKey invalidateWithParticipant, BiConsumer<Void, Throwable> callback)
@@ -316,12 +305,6 @@ abstract class Propose<R> extends AbstractCoordination<R, AcceptReply, AcceptRep
return CoordinationKind.ProposeInvalidate;
}
- @Override
- public Unseekables<?> scope()
- {
- return someParticipants;
- }
-
@Override
public Ballot ballot()
{
diff --git a/accord-core/src/main/java/accord/coordinate/ReadCoordinator.java b/accord-core/src/main/java/accord/coordinate/ReadCoordinator.java
index 1cb7303f..0a5337d6 100644
--- a/accord-core/src/main/java/accord/coordinate/ReadCoordinator.java
+++ b/accord-core/src/main/java/accord/coordinate/ReadCoordinator.java
@@ -23,6 +23,9 @@ import java.util.Arrays;
import java.util.List;
import java.util.function.BiConsumer;
+import javax.annotation.Nullable;
+
+import accord.api.Tracing;
import accord.coordinate.tracking.AbstractTracker;
import accord.coordinate.tracking.ReadTracker;
import accord.coordinate.tracking.RequestStatus;
@@ -30,6 +33,7 @@ import accord.local.Node;
import accord.local.Node.Id;
import accord.local.SequentialAsyncExecutor;
import accord.messages.Callback;
+import accord.primitives.Participants;
import accord.primitives.Route;
import accord.primitives.WithQuorum;
import accord.primitives.Ranges;
@@ -109,20 +113,23 @@ public abstract class ReadCoordinator<Result, Reply extends accord.messages.Repl
protected final Node node;
protected final SequentialAsyncExecutor executor;
protected final TxnId txnId;
+ protected final @Nullable Tracing tracing;
+ private final DebugMap debug;
+
private BiConsumer<? super Result, Throwable> callback;
private boolean isDone;
private Throwable failure;
- final DebugMap debug;
- protected ReadCoordinator(Node node, SequentialAsyncExecutor executor, Topologies topologies, TxnId txnId, BiConsumer<? super Result, Throwable> callback)
+ protected ReadCoordinator(Node node, SequentialAsyncExecutor executor, Topologies topologies, TxnId txnId, Participants<?> participants, BiConsumer<? super Result, Throwable> callback)
{
super(topologies);
this.coordinationId = node.nextCoordinationId();
this.node = node;
this.executor = executor;
this.txnId = txnId;
- this.callback = callback;
+ this.tracing = node.agent().trace(txnId, participants, kind());
this.debug = debug() ? new DebugMap(topologies.nodes()) : null;
+ this.callback = callback;
}
protected abstract Action process(Id from, Reply reply);
@@ -142,6 +149,8 @@ public abstract class ReadCoordinator<Result, Reply extends accord.messages.Repl
debug.debug(from, reply);
Action action = process(from, reply);
+ if (tracing != null)
+ tracing.trace(null, "received reply from %s: (%s) %s", from, action, reply);
switch (action)
{
default: throw new UnhandledEnum(action);
@@ -178,7 +187,11 @@ public abstract class ReadCoordinator<Result, Reply extends accord.messages.Repl
public void onSlowResponse(Id from)
{
if (!isDone)
+ {
+ if (tracing != null)
+ tracing.trace(null, "marking %s slow", from);
handle(recordSlowResponse(from));
+ }
}
@Override
@@ -190,6 +203,12 @@ public abstract class ReadCoordinator<Result, Reply extends accord.messages.Repl
if (debug != null)
debug.debug(from, failure);
+ if (tracing != null)
+ {
+ if (failure == null) tracing.trace(null, "timeout %s", from);
+ else tracing.trace(null, "received failure response from %s: %s", from, failure);
+ }
+
this.failure = FailureAccumulator.append(this.failure, failure);
handle(recordFailure(from));
}
@@ -199,6 +218,9 @@ public abstract class ReadCoordinator<Result, Reply extends accord.messages.Repl
{
if (isDone) return false;
+ if (tracing != null)
+ tracing.trace(null, "callback failed processing reply from %s: %s", from, failure);
+
finishWithFailureOverride(failure);
return true;
}
@@ -230,8 +252,11 @@ public abstract class ReadCoordinator<Result, Reply extends accord.messages.Repl
{
Invariants.require(!isDone);
Invariants.require(failure != null);
- setDone();
onDone(null, failure);
+ node.agent().coordinatorEvents().onFailed(failure, txnId, scope(), this);
+ if (tracing != null)
+ tracing.trace(null, "finishing with failure: %s", failure);
+ setDone();
}
protected void finishOnExhaustion()
@@ -267,9 +292,12 @@ public abstract class ReadCoordinator<Result, Reply extends accord.messages.Repl
break;
case Success:
Invariants.require(!isDone);
- onDone(waitingOnData == 0 ? Success.Success : Success.Quorum, null);
+ Success success = waitingOnData == 0 ? Success.Success : Success.Quorum;
+ onDone(success, null);
// isDone = true needs to be last or exceptions thrown by onDone are ignored and this never finishes
setDone();
+ if (tracing != null)
+ tracing.trace(null, "finishing with %s", success);
break;
case Failed:
finishOnExhaustion();
@@ -278,6 +306,8 @@ public abstract class ReadCoordinator<Result, Reply extends accord.messages.Repl
protected void start(List<Id> to)
{
+ if (tracing != null)
+ tracing.trace(null, "contacting %s", to);
to.forEach(this::contact);
}
@@ -305,6 +335,8 @@ public abstract class ReadCoordinator<Result, Reply extends accord.messages.Repl
// queueing callbacks externally, so two may not be in-flight at once
List<Id> contact = new ArrayList<>(1);
RequestStatus status = trySendMore(List::add, contact);
+ if (tracing != null)
+ tracing.trace(null, "contacting %s", contact);
contact.forEach(this::contact);
return status;
}
diff --git a/accord-core/src/main/java/accord/coordinate/Recover.java b/accord-core/src/main/java/accord/coordinate/Recover.java
index b3275148..bf10f913 100644
--- a/accord-core/src/main/java/accord/coordinate/Recover.java
+++ b/accord-core/src/main/java/accord/coordinate/Recover.java
@@ -28,8 +28,6 @@ import javax.annotation.Nullable;
import accord.api.Result;
import accord.api.RoutingKey;
-import accord.api.TraceEventType;
-import accord.api.Tracing;
import accord.coordinate.ExecuteFlag.CoordinationFlags;
import accord.coordinate.tracking.AbstractTracker;
import accord.coordinate.tracking.RecoveryTracker;
@@ -57,7 +55,6 @@ import accord.primitives.ProgressToken;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.TxnId;
-import accord.primitives.Unseekables;
import accord.topology.Shard;
import accord.topology.Topologies;
import accord.topology.Topology;
@@ -96,7 +93,7 @@ import static accord.utils.SortedArrays.Search.FLOOR;
// TODO (expected): lifetime of Recovery currently overlaps with follow-up work (and we callback to the Recovery).
// this is suboptimal - should setDone() and takeCallback() before passing onto next step
-public class Recover extends AbstractCoordination<Outcome, RecoverReply, RecoverOk> implements Callback<RecoverReply>
+public class Recover extends AbstractCoordination<FullRoute<?>, Outcome, RecoverReply, RecoverOk> implements Callback<RecoverReply>
{
public enum InferredFastPath
{
@@ -111,61 +108,57 @@ public class Recover extends AbstractCoordination<Outcome, RecoverReply, Recover
private final CoordinationAdapter<Result> adapter;
private final Ballot ballot;
private final Txn txn;
- private final FullRoute<?> route;
private final @Nullable Timestamp committedExecuteAt;
private final boolean isFastPathDecided;
private final LatentStoreSelector reportTo;
- private final @Nullable Tracing tracing;
private final RecoveryTracker tracker;
private Recover(Node node, SequentialAsyncExecutor executor, Topologies topologies, Ballot ballot, TxnId txnId, Txn txn, FullRoute<?> route,
@Nullable Timestamp committedExecuteAt, boolean isFastPathDecided, LatentStoreSelector reportTo,
- BiConsumer<? super Outcome, Throwable> callback, @Nullable Tracing tracing)
+ BiConsumer<? super Outcome, Throwable> callback)
{
- super(node, executor, txnId, topologies.nodes(), callback);
+ super(node, executor, txnId, route, topologies.nodes(), callback);
this.isFastPathDecided = isFastPathDecided;
Invariants.require(txnId.isVisible());
this.adapter = node.coordinationAdapter(txnId, Recovery);
this.ballot = ballot;
this.txn = txn;
- this.route = route;
this.committedExecuteAt = committedExecuteAt;
this.reportTo = reportTo;
- this.tracing = tracing;
this.tracker = new RecoveryTracker(topologies);
}
- public static Recover recover(Node node, TxnId txnId, Txn txn, FullRoute<?> route, boolean isFastPathDecided, BiConsumer<? super Outcome, Throwable> callback, @Nullable Tracing tracing)
+ public static Recover recover(Node node, TxnId txnId, Txn txn, FullRoute<?> route, boolean isFastPathDecided, BiConsumer<? super Outcome, Throwable> callback)
{
- return recover(node, txnId, txn, route, isFastPathDecided, LatentStoreSelector.standard(), callback, tracing);
+ return recover(node, txnId, txn, route, isFastPathDecided, LatentStoreSelector.standard(), callback);
}
- public static Recover recover(Node node, TxnId txnId, Txn txn, FullRoute<?> route, boolean isFastPathDecided, LatentStoreSelector reportTo, BiConsumer<? super Outcome, Throwable> callback, @Nullable Tracing tracing)
+ public static Recover recover(Node node, TxnId txnId, Txn txn, FullRoute<?> route, boolean isFastPathDecided, LatentStoreSelector reportTo, BiConsumer<? super Outcome, Throwable> callback)
{
Ballot ballot = node.uniqueTimestamp(Ballot::fromValues);
- return recover(node, ballot, txnId, txn, route, isFastPathDecided, reportTo, callback, tracing);
+ return recover(node, ballot, txnId, txn, route, isFastPathDecided, reportTo, callback);
}
- private static Recover recover(Node node, Ballot ballot, TxnId txnId, Txn txn, FullRoute<?> route, boolean isFastPathDecided, LatentStoreSelector reportTo, BiConsumer<? super Outcome, Throwable> callback, @Nullable Tracing tracing)
+ private static Recover recover(Node node, Ballot ballot, TxnId txnId, Txn txn, FullRoute<?> route, boolean isFastPathDecided, LatentStoreSelector reportTo, BiConsumer<? super Outcome, Throwable> callback)
{
- return recover(node, ballot, txnId, txn, route, null, isFastPathDecided, reportTo, callback, tracing);
+ return recover(node, ballot, txnId, txn, route, null, isFastPathDecided, reportTo, callback);
}
- public static Recover recover(Node node, Ballot ballot, TxnId txnId, Txn txn, FullRoute<?> route, @Nullable Timestamp committedExecuteAt, boolean isFastPathDecided, BiConsumer<? super Outcome, Throwable> callback, @Nullable Tracing tracing)
+ public static Recover recover(Node node, Ballot ballot, TxnId txnId, Txn txn, FullRoute<?> route, @Nullable Timestamp committedExecuteAt, boolean isFastPathDecided, BiConsumer<? super Outcome, Throwable> callback)
{
- return recover(node, ballot, txnId, txn, route, committedExecuteAt, isFastPathDecided, null, callback, tracing);
+ return recover(node, ballot, txnId, txn, route, committedExecuteAt, isFastPathDecided, null, callback);
}
- public static Recover recover(Node node, Ballot ballot, TxnId txnId, Txn txn, FullRoute<?> route, @Nullable Timestamp committedExecuteAt, boolean isFastPathDecided, LatentStoreSelector reportTo, BiConsumer<? super Outcome, Throwable> callback, @Nullable Tracing tracing)
+ public static Recover recover(Node node, Ballot ballot, TxnId txnId, Txn txn, FullRoute<?> route, @Nullable Timestamp committedExecuteAt, boolean isFastPathDecided, LatentStoreSelector reportTo, BiConsumer<? super Outcome, Throwable> callback)
{
Topologies topologies = node.topology().select(route, txnId, committedExecuteAt == null ? txnId : committedExecuteAt, SHARE, QuorumEpochIntersections.recover);
- return recover(node, topologies, ballot, txnId, txn, route, committedExecuteAt, isFastPathDecided, reportTo, callback, tracing);
+ return recover(node, topologies, ballot, txnId, txn, route, committedExecuteAt, isFastPathDecided, reportTo, callback);
}
- private static Recover recover(Node node, Topologies topologies, Ballot ballot, TxnId txnId, Txn txn, FullRoute<?> route, Timestamp committedExecuteAt, boolean isFastPathDecided, LatentStoreSelector reportTo, BiConsumer<? super Outcome, Throwable> callback, @Nullable Tracing tracing)
+ private static Recover recover(Node node, Topologies topologies, Ballot ballot, TxnId txnId, Txn txn, FullRoute<?> route, Timestamp committedExecuteAt, boolean isFastPathDecided, LatentStoreSelector reportTo, BiConsumer<? super Outcome, Throwable> callback)
{
- Recover recover = new Recover(node, node.someSequentialExecutor(), topologies, ballot, txnId, txn, route, committedExecuteAt, isFastPathDecided, reportTo, callback, tracing);
+ Recover recover = new Recover(node, node.someSequentialExecutor(), topologies, ballot, txnId, txn, route, committedExecuteAt, isFastPathDecided, reportTo, callback);
recover.start();
return recover;
}
@@ -175,15 +168,12 @@ public class Recover extends AbstractCoordination<Outcome, RecoverReply, Recover
{
super.start();
node.agent().coordinatorEvents().onRecoveryStarted(txnId, ballot);
- contact(to -> new BeginRecovery(to, tracker.topologies(), txnId, committedExecuteAt, isFastPathDecided, txn, route, ballot));
+ contact(to -> new BeginRecovery(to, tracker.topologies(), txnId, committedExecuteAt, isFastPathDecided, txn, scope, ballot));
}
@Override
public void onSuccessInternal(Id from, int fromIndex, RecoverReply reply)
{
- if (tracing != null)
- tracing.trace(null, "Recover received from %s: %s", from, reply);
-
boolean acceptsFastPath;
switch (reply.kind())
{
@@ -191,7 +181,7 @@ public class Recover extends AbstractCoordination<Outcome, RecoverReply, Recover
case Reject:
case Truncated:
// TODO (required): handle partial truncations (both within a shard e.g. pre-bootstrap, and for some shards)
- finishWithFailureOverride(Preempted.preempted(node.agent(), txnId, route.homeKey()));
+ finishWithFailureOverride(Preempted.preempted(node.agent(), txnId, scope.homeKey()));
return;
case Ok:
@@ -242,7 +232,7 @@ public class Recover extends AbstractCoordination<Outcome, RecoverReply, Recover
{
Timestamp committedExecuteAt = ((Redundant) failure).committedExecuteAt;
if (tracing != null)
- tracing.trace(null, "Recover found Redundant; retrying with known committedExecuteAt " + committedExecuteAt);
+ tracing.trace(null, "found Redundant; retrying with known committedExecuteAt " + committedExecuteAt);
retry(committedExecuteAt, callback);
}
else
@@ -283,7 +273,7 @@ public class Recover extends AbstractCoordination<Outcome, RecoverReply, Recover
case Invalidated:
{
if (tracing != null)
- tracing.trace(null, "Recover found Invalidated: committing to all shards.");
+ tracing.trace(null, "found Invalidated: committing to all shards.");
commitInvalidate(invalidateUntil(oks), finishAndTakeCallback());
return;
@@ -292,7 +282,7 @@ public class Recover extends AbstractCoordination<Outcome, RecoverReply, Recover
case AcceptedInvalidate:
{
if (tracing != null)
- tracing.trace(null, "Recover found AcceptedInvalidate: continuing Invalidate.");
+ tracing.trace(null, "found AcceptedInvalidate: continuing Invalidate.");
invalidate(oks);
return;
@@ -304,7 +294,7 @@ public class Recover extends AbstractCoordination<Outcome, RecoverReply, Recover
}
LatestDeps.Merge merge = mergeDeps(okList);
- Participants<?> await = merge.notAccepted(route);
+ Participants<?> await = merge.notAccepted(scope);
awaitPartialEarlier(okList, await, () -> {
BiConsumer<Result, Throwable> callback = finishAndTakeResultCallback();
switch (status)
@@ -314,10 +304,10 @@ public class Recover extends AbstractCoordination<Outcome, RecoverReply, Recover
case PreApplied:
{
if (tracing != null)
- tracing.trace(null, "Recover found Applied; persisting.");
+ tracing.trace(null, "found Applied; persisting.");
withStableDeps(merge, executeAt, (i, t) -> node.agent().acceptAndWrap(i, t), stableDeps -> {
- adapter.persist(node, executor, tracker.topologies(), route, ballot, CoordinationFlags.none(), txnId, txn, executeAt, stableDeps, acceptOrCommitNotTruncated.writes, acceptOrCommitNotTruncated.result, (i, t) -> node.agent().acceptAndWrap(i, t));
+ adapter.persist(node, executor, tracker.topologies(), scope, ballot, CoordinationFlags.none(), txnId, txn, executeAt, stableDeps, acceptOrCommitNotTruncated.writes, acceptOrCommitNotTruncated.result, (i, t) -> node.agent().acceptAndWrap(i, t));
});
callback.accept(acceptOrCommitNotTruncated.result, null);
return;
@@ -326,10 +316,10 @@ public class Recover extends AbstractCoordination<Outcome, RecoverReply, Recover
case Stable:
{
if (tracing != null)
- tracing.trace(null, "Recover found Stable; executing.");
+ tracing.trace(null, "found Stable; executing.");
withStableDeps(merge, executeAt, callback, stableDeps -> {
- adapter.execute(node, executor, tracker.topologies(), route, ballot, RECOVER, CoordinationFlags.none(), txnId, txn, executeAt, stableDeps, stableDeps, callback);
+ adapter.execute(node, executor, tracker.topologies(), scope, ballot, RECOVER, CoordinationFlags.none(), txnId, txn, executeAt, stableDeps, stableDeps, callback);
});
return;
}
@@ -338,10 +328,10 @@ public class Recover extends AbstractCoordination<Outcome, RecoverReply, Recover
case Committed:
{
if (tracing != null)
- tracing.trace(null, "Recover found Committed; stabilising.");
+ tracing.trace(null, "found Committed; stabilising.");
withCommittedDeps(merge, executeAt, callback, committedDeps -> {
- adapter.stabilise(node, executor, tracker.topologies(), route, ballot, txnId, txn, executeAt, committedDeps, callback);
+ adapter.stabilise(node, executor, tracker.topologies(), scope, ballot, txnId, txn, executeAt, committedDeps, callback);
});
return;
}
@@ -350,7 +340,7 @@ public class Recover extends AbstractCoordination<Outcome, RecoverReply, Recover
case AcceptedMedium:
{
if (tracing != null)
- tracing.trace(null, "Recover found Accepted; re-proposing.");
+ tracing.trace(null, "found Accepted; re-proposing.");
// TODO (desired): if we have a quorum of Accept with matching ballot or proposal we can go straight to Commit
// TODO (desired): if we didn't find Accepted in *every* shard, consider invalidating for consistency of behaviour
@@ -379,7 +369,7 @@ public class Recover extends AbstractCoordination<Outcome, RecoverReply, Recover
if (allShardsTruncated)
{
if (tracing != null)
- tracing.trace(null, "Recover found all shards truncated; terminating.");
+ tracing.trace(null, "found all shards truncated; terminating.");
// TODO (required, correctness): this is not a safe inference in the case of an ErasedOrInvalidOrVestigial response.
// We need to tighten up the inference and spread of truncation/invalid outcomes.
// In this case, at minimum this can lead to liveness violations as the home shard stops coordinating
@@ -399,7 +389,7 @@ public class Recover extends AbstractCoordination<Outcome, RecoverReply, Recover
boolean coordinatorInRecoveryQuorum = oks.get(txnId.node) != null;
Participants<?> extraCoordVotes = extraCoordinatorVotes(txnId, coordinatorInRecoveryQuorum, okList);
Participants<?> extraRejects = Deps.merge(okList, okList.size(), List::get, ok -> ok.laterCoordRejects)
- .intersecting(route, id -> !oks.containsKey(id.node));
+ .intersecting(scope, id -> !oks.containsKey(id.node));
InferredFastPath fastPath;
if (txnId.hasPrivilegedCoordinator() && coordinatorInRecoveryQuorum) fastPath = Reject;
else if (txnId.isSyncPoint()) fastPath = Reject;
@@ -413,7 +403,7 @@ public class Recover extends AbstractCoordination<Outcome, RecoverReply, Recover
case Reject:
{
if (tracing != null)
- tracing.trace(null, "Recover found fast path rejection; invoking Invalidate.");
+ tracing.trace(null, "found fast path rejection; invoking Invalidate.");
invalidate(oks);
return;
@@ -448,21 +438,21 @@ public class Recover extends AbstractCoordination<Outcome, RecoverReply, Recover
case Accept:
{
if (tracing != null)
- tracing.trace(null, "Recover found accepted fast path; proposing.");
+ tracing.trace(null, "found accepted fast path; proposing.");
propose(SLOW, txnId, okList);
break;
}
case Unknown:
{
if (tracing != null)
- tracing.trace(null, "Recover found unknown fast path decision; retrying.");
+ tracing.trace(null, "found unknown fast path decision; retrying.");
retry(committedExecuteAt, finishAndUnwrapCallback());
break;
}
case Reject:
{
if (tracing != null)
- tracing.trace(null, "Recover found fast path rejection; invoking Invalidate.");
+ tracing.trace(null, "found fast path rejection; invoking Invalidate.");
invalidate(oks);
break;
@@ -473,7 +463,7 @@ public class Recover extends AbstractCoordination<Outcome, RecoverReply, Recover
else
{
if (tracing != null)
- tracing.trace(null, "Recover found unknown fast path decision, but no preceding or superseding transactions awaiting decisions; proposing.");
+ tracing.trace(null, "found unknown fast path decision, but no preceding or superseding transactions awaiting decisions; proposing.");
propose(SLOW, txnId, okList);
}
}
@@ -543,19 +533,19 @@ public class Recover extends AbstractCoordination<Outcome, RecoverReply, Recover
private void withCommittedDeps(LatestDeps.Merge merge, Timestamp executeAt, BiConsumer<?, Throwable> failureCallback, Consumer<Deps> withDeps)
{
- LatestDeps.withCommitted(adapter, node, executor, merge, route, ballot, txnId, executeAt, txn, failureCallback, withDeps);
+ LatestDeps.withCommitted(adapter, node, executor, merge, scope, ballot, txnId, executeAt, txn, failureCallback, withDeps);
}
private void withStableDeps(LatestDeps.Merge merge, Timestamp executeAt, BiConsumer<?, Throwable> failureCallback, Consumer<Deps> withDeps)
{
- LatestDeps.withStable(adapter, node, executor, merge, Deps.NONE, route, null, null, route, ballot, txnId, executeAt, txn, failureCallback, withDeps);
+ LatestDeps.withStable(adapter, node, executor, merge, Deps.NONE, scope, null, null, scope, ballot, txnId, executeAt, txn, failureCallback, withDeps);
}
private void invalidate(SortedListMap<Id, RecoverOk> recoverOks)
{
Timestamp invalidateUntil = invalidateUntil(recoverOks);
BiConsumer<? super Outcome, Throwable> callback = finishAndTakeCallback();
- proposeInvalidate(node, executor, ballot, txnId, route.homeKey(), (success, fail) -> {
+ proposeInvalidate(node, executor, ballot, txnId, scope.homeKey(), (success, fail) -> {
if (fail != null) callback.accept(null, fail);
else commitInvalidate(invalidateUntil, callback);
});
@@ -572,9 +562,9 @@ public class Recover extends AbstractCoordination<Outcome, RecoverReply, Recover
private void commitInvalidate(Timestamp invalidateUntil, BiConsumer<? super Outcome, Throwable> callback)
{
- locallyInvalidateAndCallback(node, txnId, reportTo.refine(txnId, null, route), route, ProgressToken.INVALIDATED, callback, null);
+ locallyInvalidateAndCallback(node, txnId, reportTo.refine(txnId, null, scope), scope, ProgressToken.INVALIDATED, callback, tracing);
node.withEpochAtLeast(invalidateUntil.epoch(), executor, node.agent(), () -> {
- Commit.Invalidate.commitInvalidate(node, txnId, route, invalidateUntil);
+ Commit.Invalidate.commitInvalidate(node, txnId, scope, invalidateUntil);
});
}
@@ -594,7 +584,7 @@ public class Recover extends AbstractCoordination<Outcome, RecoverReply, Recover
return;
}
node.withEpochAtLeast(executeAt.epoch(), executor, callback, () -> {
- adapter.propose(node, executor, null, route, kind, ballot, txnId, txn, executeAt, proposeDeps, callback);
+ adapter.propose(node, executor, null, scope, kind, ballot, txnId, txn, executeAt, proposeDeps, callback);
});
}
@@ -602,17 +592,16 @@ public class Recover extends AbstractCoordination<Outcome, RecoverReply, Recover
{
Topologies topologies = tracker.topologies();
if (executeAt != null && executeAt.epoch() != (this.committedExecuteAt == null ? txnId : this.committedExecuteAt).epoch())
- topologies = node.topology().select(route, txnId, executeAt, SHARE, QuorumEpochIntersections.recover);
+ topologies = node.topology().select(scope, txnId, executeAt, SHARE, QuorumEpochIntersections.recover);
Ballot ballot = node.uniqueTimestamp(Ballot::fromValues);
- Tracing tracing = node.agent().trace(txnId, TraceEventType.RECOVER);
- Recover.recover(node, topologies, ballot, txnId, txn, route, executeAt, isFastPathDecided, reportTo, callback, tracing);
+ Recover.recover(node, topologies, ballot, txnId, txn, scope, executeAt, isFastPathDecided, reportTo, callback);
}
AsyncChain<InferredFastPath> awaitEarlier(Node node, Deps waitOn, Await.Until awaitUntil)
{
if (tracing != null)
- tracing.trace(null, "Recover awaiting earlier decisions: " + waitOn.txnIds());
+ tracing.trace(null, "awaiting earlier decisions: " + waitOn.txnIds());
long requireEpoch = waitOn.maxTxnId(txnId).epoch();
return node.withEpochAtLeast(requireEpoch, executor, () -> {
@@ -638,22 +627,22 @@ public class Recover extends AbstractCoordination<Outcome, RecoverReply, Recover
AsyncChain<InferredFastPath> awaitLater(Node node, Deps waitOn, Await.Until awaitUntil, @Nullable Participants<?> selfCoordVotes)
{
if (tracing != null)
- tracing.trace(null, "Recover awaiting later decisions or recoveries: " + waitOn.txnIds());
+ tracing.trace(null, "awaiting later decisions or recoveries: " + waitOn.txnIds());
if (waitOn.isEmpty())
return AsyncChains.success(InferredFastPath.Accept);
Participants<?> reliesOnAwaitIdCoordVote;
Topology topology = tracker.topologies().current();
- switch (route.domain())
+ switch (scope.domain())
{
- default: throw new UnhandledEnum(route.domain());
+ default: throw new UnhandledEnum(scope.domain());
case Key:
try (BufferList<RoutingKey> tmp = new BufferList<>())
{
- for (int j = 0 ; j < route.size() ; ++j)
+ for (int j = 0; j < scope.size() ; ++j)
{
- RoutingKey key = (RoutingKey)route.get(j);
+ RoutingKey key = (RoutingKey) scope.get(j);
RecoveryTracker.RecoveryShardTracker shardTracker = tracker.get(0, topology.indexForKey(key));
if (shardTracker.fastPathReliesOnUnwitnessedCoordinatorVote(txnId, selfCoordVotes))
tmp.add(key);
@@ -664,9 +653,9 @@ public class Recover extends AbstractCoordination<Outcome, RecoverReply, Recover
case Range:
try (BufferList<Range> tmp = new BufferList<>())
{
- for (int j = 0 ; j < route.size() ; ++j)
+ for (int j = 0; j < scope.size() ; ++j)
{
- Range range = (Range)route.get(j);
+ Range range = (Range) scope.get(j);
for (int k = topology.indexForRange(range, CEIL), maxk = topology.indexForRange(range, FLOOR); k <= maxk ; k++)
{
RecoveryTracker.RecoveryShardTracker shardTracker = tracker.get(0, k);
@@ -719,12 +708,6 @@ public class Recover extends AbstractCoordination<Outcome, RecoverReply, Recover
return CoordinationKind.BeginRecovery;
}
- @Override
- public Unseekables<?> scope()
- {
- return route;
- }
-
@Override
public Ballot ballot()
{
diff --git a/accord-core/src/main/java/accord/coordinate/Stabilise.java b/accord-core/src/main/java/accord/coordinate/Stabilise.java
index c7906874..cb3c52ab 100644
--- a/accord-core/src/main/java/accord/coordinate/Stabilise.java
+++ b/accord-core/src/main/java/accord/coordinate/Stabilise.java
@@ -37,7 +37,6 @@ import accord.primitives.Route;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.TxnId;
-import accord.primitives.Unseekables;
import accord.topology.Topologies;
import accord.utils.SortedArrays.SortedArrayList;
import accord.utils.UnhandledEnum;
@@ -48,10 +47,9 @@ import static accord.messages.Commit.Kind.CommitSlowPath;
import static accord.messages.Commit.Kind.CommitWithTxn;
import static accord.topology.Topologies.SelectNodeOwnership.SHARE;
-public abstract class Stabilise<R> extends AbstractCoordination<R, ReadReply, Void> implements Callback<ReadReply>
+public abstract class Stabilise<R> extends AbstractCoordination<FullRoute<?>, R, ReadReply, Void> implements Callback<ReadReply>
{
final Txn txn;
- final FullRoute<?> route;
final Route<?> sendTo;
final Ballot ballot;
final Timestamp executeAt;
@@ -62,10 +60,9 @@ public abstract class Stabilise<R> extends AbstractCoordination<R, ReadReply, Vo
public Stabilise(Node node, SequentialAsyncExecutor executor, Topologies coordinates, Topologies allTopologies, Route<?> sendTo, FullRoute<?> route, TxnId txnId, Ballot ballot, Txn txn, Timestamp executeAt, Deps stabiliseDeps, BiConsumer<? super R, Throwable> callback)
{
- super(node, executor, txnId, coordinates.nodes(), callback);
+ super(node, executor, txnId, route, coordinates.nodes(), callback);
this.txn = txn;
this.sendTo = sendTo;
- this.route = route;
this.ballot = ballot;
this.executeAt = executeAt;
this.stabiliseDeps = stabiliseDeps;
@@ -88,7 +85,7 @@ public abstract class Stabilise<R> extends AbstractCoordination<R, ReadReply, Vo
else
{
super.start();
- contact(to -> new Commit(CommitSlowPath, to, allTopologies, txnId, txn, route, ballot, executeAt, stabiliseDeps));
+ contact(to -> new Commit(CommitSlowPath, to, allTopologies, txnId, txn, scope, ballot, executeAt, stabiliseDeps));
}
}
@@ -107,18 +104,18 @@ public abstract class Stabilise<R> extends AbstractCoordination<R, ReadReply, Vo
{
default: throw new UnhandledEnum(nack.kind);
case Redundant:
- finishWithFailureOverride(new Redundant(txnId, route.homeKey(), executeAt));
+ finishWithFailureOverride(new Redundant(txnId, scope.homeKey(), executeAt));
break;
case Rejected:
- recordFailure(from, Preempted.preempted(node.agent(), txnId, route.homeKey()));
+ recordFailure(from, Preempted.preempted(node.agent(), txnId, scope.homeKey()));
break;
case Insufficient:
node.send(from, new Commit(CommitWithTxn, from, allTopologies,
- txnId, txn, route, ballot, executeAt, stabiliseDeps));
+ txnId, txn, scope, ballot, executeAt, stabiliseDeps));
break;
case InsufficientEpochs:
- node.send(from, new Commit(CommitWithTxn, from, node.topology().preciseEpochs(route, Math.min(allTopologies.oldestEpoch(), nack.minEpoch()), allTopologies.currentEpoch(), SHARE),
- txnId, txn, route, ballot, executeAt, stabiliseDeps));
+ node.send(from, new Commit(CommitWithTxn, from, node.topology().preciseEpochs(scope, Math.min(allTopologies.oldestEpoch(), nack.minEpoch()), allTopologies.currentEpoch(), SHARE),
+ txnId, txn, scope, ballot, executeAt, stabiliseDeps));
break;
}
}
@@ -139,7 +136,7 @@ public abstract class Stabilise<R> extends AbstractCoordination<R, ReadReply, Vo
protected void onStabilised()
{
- adapter().execute(node, executor, allTopologies, route, ballot, SLOW, CoordinationFlags.none(), txnId, txn, executeAt, stabiliseDeps, stabiliseDeps, finishAndTakeCallback());
+ adapter().execute(node, executor, allTopologies, scope, ballot, SLOW, CoordinationFlags.none(), txnId, txn, executeAt, stabiliseDeps, stabiliseDeps, finishAndTakeCallback());
}
@Override
@@ -148,12 +145,6 @@ public abstract class Stabilise<R> extends AbstractCoordination<R, ReadReply, Vo
return CoordinationKind.Stabilise;
}
- @Override
- public Unseekables<?> scope()
- {
- return route;
- }
-
@Override
public Ballot ballot()
{
diff --git a/accord-core/src/main/java/accord/coordinate/SynchronousAwait.java b/accord-core/src/main/java/accord/coordinate/SynchronousAwait.java
index d425b3b3..56a350b5 100644
--- a/accord-core/src/main/java/accord/coordinate/SynchronousAwait.java
+++ b/accord-core/src/main/java/accord/coordinate/SynchronousAwait.java
@@ -31,7 +31,6 @@ import accord.local.SequentialAsyncExecutor;
import accord.messages.Await;
import accord.primitives.Participants;
import accord.primitives.TxnId;
-import accord.primitives.Unseekables;
import accord.topology.Topologies;
import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncChains;
@@ -42,17 +41,15 @@ import accord.utils.async.Cancellable;
* This may or may not be a condition we expect to reach promptly, but we will wait only until the timeout passes
* at which point we will report failure.
*/
-public class SynchronousAwait extends AbstractCoordination<Boolean, Await.AwaitOk, Void>
+public class SynchronousAwait extends AbstractCoordination<Participants<?>, Boolean, Await.AwaitOk, Void>
{
- final Participants<?> participants;
final SimpleTracker<?> tracker;
final Await.Until until;
final boolean notifyProgressLog;
public SynchronousAwait(Node node, SequentialAsyncExecutor executor, TxnId txnId, Participants<?> participants, SimpleTracker<?> tracker, Await.Until until, boolean notifyProgressLog, BiConsumer<? super Boolean, Throwable> callback)
{
- super(node, executor, txnId, tracker.nodes(), callback);
- this.participants = participants;
+ super(node, executor, txnId, participants, tracker.nodes(), callback);
this.until = until;
this.notifyProgressLog = notifyProgressLog;
this.tracker = tracker;
@@ -62,7 +59,7 @@ public class SynchronousAwait extends AbstractCoordination<Boolean, Await.AwaitO
void start()
{
super.start();
- contact(to -> new Await(to, tracker.topologies(), txnId, participants, until, notifyProgressLog));
+ contact(to -> new Await(to, tracker.topologies(), txnId, scope, until, notifyProgressLog));
}
public static AsyncChain<Boolean> awaitQuorum(Node node, SequentialAsyncExecutor executor, Topologies topologies, TxnId txnId, Participants<?> participants, Await.Until until, boolean notifyProgressLog)
@@ -108,12 +105,6 @@ public class SynchronousAwait extends AbstractCoordination<Boolean, Await.AwaitO
return CoordinationKind.SyncAwait;
}
- @Override
- public Unseekables<?> scope()
- {
- return participants;
- }
-
@Override
public AbstractTracker<?> tracker()
{
diff --git a/accord-core/src/main/java/accord/coordinate/SynchronousRecoverAwait.java b/accord-core/src/main/java/accord/coordinate/SynchronousRecoverAwait.java
index 1a0b05a6..8319f463 100644
--- a/accord-core/src/main/java/accord/coordinate/SynchronousRecoverAwait.java
+++ b/accord-core/src/main/java/accord/coordinate/SynchronousRecoverAwait.java
@@ -31,7 +31,6 @@ import accord.messages.RecoverAwait;
import accord.messages.RecoverAwait.RecoverAwaitOk;
import accord.primitives.Participants;
import accord.primitives.TxnId;
-import accord.primitives.Unseekables;
import accord.topology.Topologies;
import accord.utils.Invariants;
import accord.utils.UnhandledEnum;
@@ -59,7 +58,7 @@ public class SynchronousRecoverAwait extends ReadCoordinator<InferredFastPath, R
private Participants<?> waitingOn;
public SynchronousRecoverAwait(Node node, SequentialAsyncExecutor executor, Topologies topologies, TxnId txnId, Participants<?> participants, Await.Until until, boolean notifyProgressLog, TxnId recoverId, BiConsumer<? super InferredFastPath, Throwable> callback)
{
- super(node, executor, topologies, txnId, callback);
+ super(node, executor, topologies, txnId, participants, callback);
this.participants = participants;
this.until = until;
this.notifyProgressLog = notifyProgressLog;
@@ -134,7 +133,7 @@ public class SynchronousRecoverAwait extends ReadCoordinator<InferredFastPath, R
}
@Override
- public Unseekables<?> scope()
+ public Participants<?> scope()
{
return participants;
}
diff --git a/accord-core/src/main/java/accord/impl/progresslog/HomeState.java b/accord-core/src/main/java/accord/impl/progresslog/HomeState.java
index eacba2e0..5aedbc1b 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/HomeState.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/HomeState.java
@@ -37,7 +37,7 @@ import accord.primitives.Status.Durability.HasOutcome;
import accord.primitives.TxnId;
import accord.utils.Invariants;
-import static accord.api.TraceEventType.HOME_PROGRESS;
+import static accord.coordinate.Coordination.CoordinationKind.HomeProgress;
import static accord.impl.progresslog.CallbackInvoker.invokeHomeCallback;
import static accord.impl.progresslog.CoordinatePhase.Done;
import static accord.impl.progresslog.Progress.NoneExpected;
@@ -140,9 +140,9 @@ abstract class HomeState extends WaitingState
final void runHome(DefaultProgressLog instance, SafeCommandStore safeStore, SafeCommand safeCommand)
{
incrementHomeRunCounter();
- Tracing tracing = instance.node.agent().trace(txnId, HOME_PROGRESS);
Invariants.require(!isHomeDoneOrUninitialised());
Command command = safeCommand.current();
+ Tracing tracing = instance.node.agent().trace(txnId, command.participants().max(), HomeProgress);
// note: we may truncate locally based on shard-specific criteria, but this doesn't mean we're globally persisted
if (command.saveStatus() == SaveStatus.Erased // TODO (expected): improve progressLog.clear() so we can expect these to be cleared from the progress log
@@ -180,16 +180,12 @@ abstract class HomeState extends WaitingState
static void recoverCallback(SafeCommandStore safeStore, SafeCommand safeCommand, DefaultProgressLog instance, TxnId txnId, @Nullable ProgressToken prevProgressToken, Outcome success, Throwable fail)
{
- Tracing tracing = instance.node.agent().trace(safeCommand.txnId(), HOME_PROGRESS);
HomeState state = instance.get(txnId);
if (state == null)
- {
- if (tracing != null)
- tracing.trace(safeStore.commandStore(), "No HomeState to process recovery callback");
return;
- }
Command command = safeCommand.current();
+ Tracing tracing = instance.node.agent().trace(safeCommand.txnId(), command.participants().max(), HomeProgress);
CoordinatePhase status = state.phase();
if (status.isAtMostReadyToExecute() && state.homeProgress() == Querying)
{
@@ -197,7 +193,7 @@ abstract class HomeState extends WaitingState
{
if (tracing != null)
{
- tracing.trace(safeStore.commandStore(), "Failed to recover: " + Tracing.format(fail));
+ tracing.trace(safeStore.commandStore(), "Failed to recover: %s", fail);
tracing.trace(safeStore.commandStore(), "Waiting to retry (%d) with progress token %s", state.homeRunCounter(), prevProgressToken);
}
diff --git a/accord-core/src/main/java/accord/impl/progresslog/WaitingState.java b/accord-core/src/main/java/accord/impl/progresslog/WaitingState.java
index cd994cac..85a6b26c 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/WaitingState.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/WaitingState.java
@@ -52,7 +52,7 @@ import static accord.api.ProgressLog.BlockedUntil.CanApply;
import static accord.api.ProgressLog.BlockedUntil.HasDecidedExecuteAt;
import static accord.api.ProgressLog.BlockedUntil.Query.HOME;
import static accord.api.ProgressLog.BlockedUntil.Query.SHARD;
-import static accord.api.TraceEventType.WAIT_PROGRESS;
+import static accord.coordinate.Coordination.CoordinationKind.WaitProgress;
import static accord.impl.progresslog.CallbackInvoker.invokeWaitingCallback;
import static accord.impl.progresslog.PackedKeyTracker.bitSet;
import static accord.impl.progresslog.PackedKeyTracker.clearRoundState;
@@ -71,6 +71,7 @@ import static accord.impl.progresslog.TxnStateKind.Waiting;
import static accord.impl.progresslog.WaitingState.CallbackKind.AwaitHome;
import static accord.impl.progresslog.WaitingState.CallbackKind.AwaitSlice;
import static accord.impl.progresslog.WaitingState.CallbackKind.Fetch;
+import static accord.local.SafeCommand.maxParticipants;
import static accord.primitives.Known.KnownExecuteAt.ExecuteAtErased;
import static accord.topology.Topologies.SelectNodeOwnership.SHARE;
@@ -302,7 +303,7 @@ abstract class WaitingState extends BaseTxnState
if (offset >= 3)
{
offset = 3;
- lowEpoch = safeStore.ranges().latestEarlierEpochThatFullyCovers(lowEpoch, command.maxContactable());
+ lowEpoch = safeStore.ranges().latestEarlierEpochThatFullyCovers(lowEpoch, command.maxParticipants());
}
encodedState = encodedState & ~(0x3L << AWAIT_EPOCH_SHIFT);
encodedState |= ((long)offset) << AWAIT_EPOCH_SHIFT;
@@ -345,7 +346,7 @@ abstract class WaitingState extends BaseTxnState
if (offset >= 3)
{
offset = 3;
- highEpoch = safeStore.ranges().earliestLaterEpochThatFullyCovers(highEpoch, command.maxContactable());
+ highEpoch = safeStore.ranges().earliestLaterEpochThatFullyCovers(highEpoch, command.maxParticipants());
}
encodedState = encodedState & ~(0xCL << AWAIT_EPOCH_SHIFT);
encodedState |= ((long)offset) << (AWAIT_EPOCH_SHIFT + 2);
@@ -444,7 +445,7 @@ abstract class WaitingState extends BaseTxnState
final void runWaiting(DefaultProgressLog owner, SafeCommandStore safeStore, SafeCommand safeCommand)
{
- runInternal(safeStore, safeCommand, owner, owner.node.agent().trace(txnId, WAIT_PROGRESS));
+ runInternal(safeStore, safeCommand, owner, owner.node.agent().trace(txnId, maxParticipants(safeCommand), WaitProgress));
}
private void runInternal(SafeCommandStore safeStore, SafeCommand safeCommand, DefaultProgressLog owner, @Nullable Tracing tracing)
@@ -463,7 +464,7 @@ abstract class WaitingState extends BaseTxnState
TxnId txnId = safeCommand.txnId();
// first make sure we have enough information to obtain the command locally
Timestamp executeAt = command.executeAtIfKnown();
- Participants<?> maxContactable = Invariants.nonNull(command.maxContactable());
+ Participants<?> maxContactable = Invariants.nonNull(command.maxParticipants());
if (!Route.isRoute(maxContactable))
{
@@ -486,7 +487,7 @@ abstract class WaitingState extends BaseTxnState
tracing.trace(owner.commandStore, "Blocked until %s. Waiting for home key %s to satisfy %s.", blockedUntil, route.homeKey(), next);
clearAwaitState();
set(safeStore, owner, next, Querying);
- awaitHomeKey(owner, next, txnId, executeAt, route);
+ awaitHomeKey(owner, next, txnId, executeAt, route, tracing);
return;
}
@@ -555,7 +556,7 @@ abstract class WaitingState extends BaseTxnState
if (tracing != null)
tracing.trace(owner.commandStore, "Blocked until %s. Waiting for %s to satisfy %s; round %d of %d.", blockedUntil, awaitRoute, querying, roundIndex, (awaitRoute.size() + (roundSize - 1))/roundSize);
setAwaitStarted();
- awaitSlice(owner, querying, txnId, executeAt, awaitRoute, (roundIndex << 1) | 1);
+ awaitSlice(owner, querying, txnId, executeAt, awaitRoute, (roundIndex << 1) | 1, tracing);
}
// note that ready and notReady may include keys not requested by this progressLog
@@ -569,7 +570,7 @@ abstract class WaitingState extends BaseTxnState
Command command = safeCommand.current();
Route<?> route = command.route();
- Tracing tracing = owner.node.agent().trace(txnId, WAIT_PROGRESS);
+ Tracing tracing = owner.node.agent().trace(txnId, command.participants().max(), WaitProgress);
if (fail == null)
{
@@ -782,7 +783,9 @@ abstract class WaitingState extends BaseTxnState
if ((callbackId & 1) != 1)
return;
- Tracing tracing = owner.node.agent().trace(txnId, WAIT_PROGRESS);
+ SafeCommand safeCommand = safeStore.unsafeGet(txnId);
+ Tracing tracing = owner.node.agent().trace(txnId, maxParticipants(safeCommand), WaitProgress);
+
BlockedUntil querying = querying();
if (callbackId == AWAITING_HOME_KEY_CALLBACKID)
{
@@ -806,7 +809,6 @@ abstract class WaitingState extends BaseTxnState
return;
}
- SafeCommand safeCommand = safeStore.unsafeGet(txnId);
if (safeCommand != null)
{
if (tracing != null)
@@ -831,8 +833,8 @@ abstract class WaitingState extends BaseTxnState
}
callbackId >>= 1;
- SafeCommand safeCommand = Invariants.nonNull(safeStore.unsafeGet(txnId));
- Route<?> route = Route.castToRoute(safeCommand.current().maxContactable());
+ Invariants.nonNull(safeCommand);
+ Route<?> route = Route.castToRoute(safeCommand.current().maxParticipants());
long lowEpoch = readLowEpoch(safeStore, txnId, route);
long highEpoch = readHighEpoch(safeStore, txnId, route);
Route<?> slicedRoute = slicedRoute(safeStore, txnId, route, lowEpoch, highEpoch);
@@ -902,20 +904,20 @@ abstract class WaitingState extends BaseTxnState
owner.start(invoker, FetchData.fetchSpecific(querying.unblockedFrom.known, owner.node(), txnId, invalidIf, executeAt, fetchRoute, maxRoute, new IncludingSpecificStoreSelector(owner.commandStore.id()), invoker));
}
- void awaitHomeKey(DefaultProgressLog owner, BlockedUntil blockedUntil, TxnId txnId, Timestamp executeAt, Route<?> route)
+ void awaitHomeKey(DefaultProgressLog owner, BlockedUntil blockedUntil, TxnId txnId, Timestamp executeAt, Route<?> route, @Nullable Tracing tracing)
{
// TODO (expected): special-case when this shard is home key to avoid remote messages
- await(owner, blockedUntil, txnId, executeAt, route.homeKeyOnlyRoute(), AWAITING_HOME_KEY_CALLBACKID, WaitingState::synchronousAwaitHomeCallback);
+ await(owner, blockedUntil, txnId, executeAt, route.homeKeyOnlyRoute(), AWAITING_HOME_KEY_CALLBACKID, WaitingState::synchronousAwaitHomeCallback, tracing);
}
- void awaitSlice(DefaultProgressLog owner, BlockedUntil blockedUntil, TxnId txnId, Timestamp executeAt, Route<?> route, int callbackId)
+ void awaitSlice(DefaultProgressLog owner, BlockedUntil blockedUntil, TxnId txnId, Timestamp executeAt, Route<?> route, int callbackId, @Nullable Tracing tracing)
{
Invariants.require(blockedUntil.waitsOn == SHARD);
// TODO (expected): special-case when this shard is home key to avoid remote messages
- await(owner, blockedUntil, txnId, executeAt, route, callbackId, WaitingState::synchronousAwaitSliceCallback);
+ await(owner, blockedUntil, txnId, executeAt, route, callbackId, WaitingState::synchronousAwaitSliceCallback, tracing);
}
- void await(DefaultProgressLog owner, BlockedUntil blockedUntil, TxnId txnId, Timestamp executeAt, Route<?> route, int callbackId, Callback<BlockedUntil, AsynchronousAwait.SynchronousResult> callback)
+ void await(DefaultProgressLog owner, BlockedUntil blockedUntil, TxnId txnId, Timestamp executeAt, Route<?> route, int callbackId, Callback<BlockedUntil, AsynchronousAwait.SynchronousResult> callback, @Nullable Tracing tracing)
{
long epoch = blockedUntil.fetchEpoch(txnId, executeAt);
Await.Until awaitUntil = blockedUntil.toAwait();
diff --git a/accord-core/src/main/java/accord/local/Command.java b/accord-core/src/main/java/accord/local/Command.java
index b9d5b812..a9a66196 100644
--- a/accord-core/src/main/java/accord/local/Command.java
+++ b/accord-core/src/main/java/accord/local/Command.java
@@ -268,11 +268,9 @@ public abstract class Command implements ICommand
@Nullable
public final Route<?> route() { return participants().route(); }
- public Participants<?> maxContactable()
+ public Participants<?> maxParticipants()
{
- Route<?> route = route();
- if (route != null) return route.withHomeKey();
- else return participants().hasTouched();
+ return participants.max();
}
/**
diff --git a/accord-core/src/main/java/accord/local/Node.java b/accord-core/src/main/java/accord/local/Node.java
index 3cfc3844..e91d0f4c 100644
--- a/accord-core/src/main/java/accord/local/Node.java
+++ b/accord-core/src/main/java/accord/local/Node.java
@@ -51,7 +51,6 @@ import accord.api.RoutingKey;
import accord.api.Scheduler;
import accord.api.Timeouts;
import accord.api.TopologySorter;
-import accord.api.Tracing;
import accord.coordinate.CoordinateEphemeralRead;
import accord.coordinate.CoordinateTransaction;
import accord.coordinate.Coordination;
@@ -763,17 +762,11 @@ public class Node implements ConfigurationService.Listener, NodeCommandStoreServ
private AsyncChain<Result> initiateCoordination(TxnId txnId, Txn txn)
{
+ FullRoute<?> route = computeRoute(txnId, txn.keys());
if (txnId.kind() == Txn.Kind.EphemeralRead)
- {
- // TODO (expected): once non-participating home keys are removed, this can be simplified to share computeRoute
- FullRoute<?> route = txn.keys().toRoute(txn.keys().get(0).someIntersectingRoutingKey(null));
return CoordinateEphemeralRead.coordinate(this, route, txnId, txn);
- }
else
- {
- FullRoute<?> route = computeRoute(txnId, txn.keys());
return CoordinateTransaction.coordinate(this, route, txnId, txn);
- }
}
public FullRoute<?> computeRoute(TxnId txnId, Routables<?> keysOrRanges)
@@ -798,7 +791,7 @@ public class Node implements ConfigurationService.Listener, NodeCommandStoreServ
return keysOrRanges.get(random.nextInt(keysOrRanges.size())).someIntersectingRoutingKey(null);
}
- public AsyncChain<? extends Outcome> recover(TxnId txnId, InvalidIf invalidIf, FullRoute<?> route, LatentStoreSelector reportTo, @Nullable Tracing tracing)
+ public AsyncChain<? extends Outcome> recover(TxnId txnId, InvalidIf invalidIf, FullRoute<?> route, LatentStoreSelector reportTo)
{
SequentialAsyncExecutor executor = someSequentialExecutor();
return withEpochExact(txnId.epoch(), executor, () -> new AsyncChains.Head<>()
@@ -806,7 +799,7 @@ public class Node implements ConfigurationService.Listener, NodeCommandStoreServ
@Override
protected Cancellable start(BiConsumer<? super Outcome, Throwable> callback)
{
- PrepareRecovery.recover(Node.this, executor, txnId, invalidIf, route, null, reportTo, callback, tracing);
+ PrepareRecovery.recover(Node.this, executor, txnId, invalidIf, route, null, reportTo, callback);
return null;
}
});
diff --git a/accord-core/src/main/java/accord/local/SafeCommand.java b/accord-core/src/main/java/accord/local/SafeCommand.java
index 1d67a95c..794496ff 100644
--- a/accord-core/src/main/java/accord/local/SafeCommand.java
+++ b/accord-core/src/main/java/accord/local/SafeCommand.java
@@ -19,12 +19,14 @@
package accord.local;
import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import accord.api.Result;
import accord.local.Command.Truncated;
import accord.primitives.Ballot;
import accord.primitives.PartialDeps;
import accord.primitives.PartialTxn;
+import accord.primitives.Participants;
import accord.primitives.SaveStatus;
import accord.primitives.Status;
import accord.primitives.Timestamp;
@@ -198,4 +200,14 @@ public abstract class SafeCommand
return current;
return incidentalUpdate(Command.NotDefined.notDefined(current, current.promised()));
}
+
+ public static @Nullable Participants<?> maxParticipants(@Nullable SafeCommand safeCommand)
+ {
+ if (safeCommand == null)
+ return null;
+ Command command = safeCommand.current();
+ if (command == null)
+ return null;
+ return command.maxParticipants();
+ }
}
diff --git a/accord-core/src/main/java/accord/local/StoreParticipants.java b/accord-core/src/main/java/accord/local/StoreParticipants.java
index 14ae4c19..ec7c7d86 100644
--- a/accord-core/src/main/java/accord/local/StoreParticipants.java
+++ b/accord-core/src/main/java/accord/local/StoreParticipants.java
@@ -251,6 +251,11 @@ public class StoreParticipants
return owns;
}
+ public Participants<?> max()
+ {
+ return route != null ? route.withHomeKey() : hasTouched();
+ }
+
public boolean touchesOnlyOwned()
{
return true;
diff --git a/accord-core/src/main/java/accord/messages/Propagate.java b/accord-core/src/main/java/accord/messages/Propagate.java
index 4a9bd59a..beb1dad0 100644
--- a/accord-core/src/main/java/accord/messages/Propagate.java
+++ b/accord-core/src/main/java/accord/messages/Propagate.java
@@ -122,7 +122,8 @@ public class Propagate extends MapReduceConsumeCommandStores<Route<?>, Void>
@Nullable Timestamp committedExecuteAt,
@Nullable Writes writes,
@Nullable Result result,
- BiConsumer<? super FetchResult, Throwable> callback, @Nullable Tracing tracing)
+ BiConsumer<? super FetchResult, Throwable> callback,
+ @Nullable Tracing tracing)
{
super(route);
this.node = node;
@@ -537,7 +538,7 @@ public class Propagate extends MapReduceConsumeCommandStores<Route<?>, Void>
if (null != callback)
callback.accept(failure != null ? null : finaliseFetchResult(), failure);
if (failure != null && tracing != null)
- tracing.trace(null, "Failed propagate: %s", Tracing.format(failure));
+ tracing.trace(null, "Failed propagate: %s", failure);
}
private static void confirm(Commands.CommitOutcome outcome)
diff --git a/accord-core/src/main/java/accord/primitives/Routable.java b/accord-core/src/main/java/accord/primitives/Routable.java
index 45e40445..94b797d5 100644
--- a/accord-core/src/main/java/accord/primitives/Routable.java
+++ b/accord-core/src/main/java/accord/primitives/Routable.java
@@ -42,7 +42,7 @@ public interface Routable
return this == Range;
}
- public static Routable.Domain ofOrdinal(int ordinal)
+ public static Routable.Domain forOrdinal(int ordinal)
{
return VALUES[ordinal];
}
diff --git a/accord-core/src/main/java/accord/primitives/Txn.java b/accord-core/src/main/java/accord/primitives/Txn.java
index bba808d4..5b0449f0 100644
--- a/accord-core/src/main/java/accord/primitives/Txn.java
+++ b/accord-core/src/main/java/accord/primitives/Txn.java
@@ -190,7 +190,6 @@ public interface Txn
this.witnessedBy = witnessedBy;
}
-
public boolean isWrite()
{
return this == Write;
@@ -224,7 +223,7 @@ public interface Txn
return awaitsOnlyDeps;
}
- public static Kind ofOrdinal(int ordinal)
+ public static Kind forOrdinal(int ordinal)
{
return VALUES[ordinal];
}
@@ -288,6 +287,16 @@ public interface Txn
{
return shortName;
}
+
+ public static Txn.Kind forShortName(char ch)
+ {
+ for (Txn.Kind kind : VALUES)
+ {
+ if (kind.shortName == ch)
+ return kind;
+ }
+ return null;
+ }
}
class InMemory implements Txn
diff --git a/accord-core/src/main/java/accord/primitives/TxnId.java b/accord-core/src/main/java/accord/primitives/TxnId.java
index 63835e29..a7d041bf 100644
--- a/accord-core/src/main/java/accord/primitives/TxnId.java
+++ b/accord-core/src/main/java/accord/primitives/TxnId.java
@@ -446,12 +446,12 @@ public class TxnId extends Timestamp
private static Kind kind(int flags)
{
- return Kind.ofOrdinal(kindOrdinal(flags));
+ return Kind.forOrdinal(kindOrdinal(flags));
}
private static Domain domain(int flags)
{
- return Domain.ofOrdinal(domainOrdinal(flags));
+ return Domain.forOrdinal(domainOrdinal(flags));
}
public static int kindOrdinal(int flags)
diff --git a/accord-core/src/main/java/accord/utils/TinyEnumSet.java b/accord-core/src/main/java/accord/utils/TinyEnumSet.java
index 9f7ea7c9..ad540222 100644
--- a/accord-core/src/main/java/accord/utils/TinyEnumSet.java
+++ b/accord-core/src/main/java/accord/utils/TinyEnumSet.java
@@ -18,6 +18,8 @@
package accord.utils;
+import java.util.Iterator;
+import java.util.function.Function;
import java.util.function.IntFunction;
public class TinyEnumSet<E extends Enum<E>>
@@ -50,6 +52,11 @@ public class TinyEnumSet<E extends Enum<E>>
return new TinyEnumSet<>(-1 >>> (32 - count));
}
+ public static <E extends Enum<E>> TinyEnumSet<E> of()
+ {
+ return new TinyEnumSet<>(0);
+ }
+
public static <E extends Enum<E>> TinyEnumSet<E> of(Enum<E> value)
{
return new TinyEnumSet<>(encode(value));
@@ -86,6 +93,16 @@ public class TinyEnumSet<E extends Enum<E>>
return 0 != (bitset & (1 << ordinal));
}
+ public boolean isEmpty()
+ {
+ return bitset == 0;
+ }
+
+ public int size()
+ {
+ return Integer.bitCount(bitset);
+ }
+
public boolean contains(E value)
{
return contains(bitset, value.ordinal());
@@ -101,6 +118,16 @@ public class TinyEnumSet<E extends Enum<E>>
return or(this, or, TinyEnumSet::new);
}
+ public TinyEnumSet<E> and(TinyEnumSet<E> and)
+ {
+ return and(this, and, TinyEnumSet::new);
+ }
+
+ public TinyEnumSet<E> not(TinyEnumSet<E> not)
+ {
+ return not(this, not, TinyEnumSet::new);
+ }
+
public TinyEnumSet<E> with(E flag)
{
return new TinyEnumSet<>(bitset | encode(flag));
@@ -112,6 +139,18 @@ public class TinyEnumSet<E extends Enum<E>>
return newBitset == a.bitset ? a : newBitset == b.bitset ? b : constructor.apply(newBitset);
}
+ public static <S extends TinyEnumSet<?>> S and(S a, S b, IntFunction<S> constructor)
+ {
+ int newBitset = a.bitset & b.bitset;
+ return newBitset == a.bitset ? a : newBitset == b.bitset ? b : constructor.apply(newBitset);
+ }
+
+ public static <S extends TinyEnumSet<?>> S not(S a, S b, IntFunction<S> constructor)
+ {
+ int newBitset = a.bitset & ~b.bitset;
+ return newBitset == a.bitset ? a : newBitset == b.bitset ? b : constructor.apply(newBitset);
+ }
+
public boolean test(E kind)
{
return testOrdinal(kind.ordinal());
@@ -132,25 +171,61 @@ public class TinyEnumSet<E extends Enum<E>>
return bitset;
}
- protected String toString(Object[] universe)
+ protected String toString(IntFunction<E> universe)
{
- if (bitset == 0)
+ return toString(bitset, universe);
+ }
+
+ public static <E extends Enum<E>> String toString(int bits, IntFunction<E> universe)
+ {
+ return toString(bits, universe, Enum::toString);
+ }
+
+ public static <E extends Enum<E>> String toString(int bits, IntFunction<E> universe, Function<E, String> print)
+ {
+ if (bits == 0)
return "{}";
- StringBuilder sb = new StringBuilder();
- sb.append('{');
- int bits = this.bitset;
+ StringBuilder out = new StringBuilder();
+ out.append('{');
+ append(bits, universe, print, out);
+ out.append('}');
+ return out.toString();
+ }
+
+ public static <E extends Enum<E>> void append(int bits, IntFunction<E> universe, Function<E, String> print, StringBuilder out)
+ {
+ boolean comma = false;
while (bits != 0)
{
- if (sb.length() > 1)
- sb.append(',');
+ if (comma)
+ out.append(',');
int i = Integer.numberOfTrailingZeros(bits);
- sb.append(universe[i]);
+ out.append(print.apply(universe.apply(i)));
bits ^= 1 << i;
+ comma = true;
}
- sb.append('}');
- return sb.toString();
}
+ public Iterable<E> iterable(IntFunction<E> lookup)
+ {
+ return () -> new Iterator<>()
+ {
+ int remaining = bitset;
+ @Override
+ public boolean hasNext()
+ {
+ return remaining != 0;
+ }
+
+ @Override
+ public E next()
+ {
+ E next = lookup.apply(Integer.numberOfTrailingZeros(remaining));
+ remaining ^= Integer.lowestOneBit(remaining);
+ return next;
+ }
+ };
+ }
}
diff --git a/accord-core/src/test/java/accord/impl/list/ListAgent.java b/accord-core/src/test/java/accord/impl/list/ListAgent.java
index 6c280661..d4cb96ff 100644
--- a/accord-core/src/test/java/accord/impl/list/ListAgent.java
+++ b/accord-core/src/test/java/accord/impl/list/ListAgent.java
@@ -36,8 +36,8 @@ import accord.api.OwnershipEventListener;
import accord.api.ProgressLog.BlockedUntil;
import accord.api.Result;
import accord.api.Scheduler;
-import accord.api.TraceEventType;
import accord.api.Tracing;
+import accord.coordinate.Coordination;
import accord.coordinate.CoordinationFailed;
import accord.coordinate.ExecuteSyncPoint;
import accord.impl.InMemoryAgent;
@@ -56,6 +56,7 @@ import accord.local.TimeService;
import accord.messages.ReplyContext;
import accord.primitives.Ballot;
import accord.primitives.Keys;
+import accord.primitives.Participants;
import accord.primitives.Ranges;
import accord.primitives.Routable.Domain;
import accord.primitives.Status;
@@ -126,10 +127,11 @@ public class ListAgent implements InMemoryAgent, CoordinatorEventListener, Owner
@Nullable
@Override
- public Tracing trace(TxnId txnId, TraceEventType eventType)
+ public Tracing trace(TxnId txnId, Participants<?> participants, Coordination.CoordinationKind eventType)
{
if (rnd.nextFloat() < 0.01f)
- return (i1, i2) -> {};
+ return (store, message) -> {};
+
return null;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org