dcapwell commented on code in PR #219:
URL: https://github.com/apache/cassandra-accord/pull/219#discussion_r2226162970
##########
accord-core/src/main/java/accord/api/Agent.java:
##########
@@ -40,22 +40,11 @@
/**
* Facility for augmenting node behaviour at specific points
- *
- * TODO (expected): rationalise LocalConfig and Agent
*/
public interface Agent extends UncaughtExceptionListener
{
default @Nullable Tracing trace(TxnId txnId, TraceEventType eventType) {
return null; }
- /**
- * For use by implementations to decide what to do about successfully
recovered transactions.
- * Specifically intended to define if and how they should inform clients
of the result.
- * e.g. in Maelstrom we send the full result directly, in other impls we
may simply acknowledge success via the coordinator
- *
- * Note: may be invoked multiple times in different places
- */
- void onRecover(Node node, Result success, Throwable fail);
Review Comment:
for me: this was replaced by
`accord.api.CoordinatorEventListener#onRecoveryStopped`?
##########
accord-core/src/main/java/accord/coordinate/Propose.java:
##########
@@ -197,9 +197,10 @@ void onAccepted()
// In this case either id' needs to wait (which requires potentially
more states like the alternative medium path)
// Or we must pick it up as an Unstable dependency here.
Deps newDeps = mergeNewDeps();
- Deps stableDeps = mergeDeps(newDeps);
- if (kind == Kind.MEDIUM) adapter().execute(node, executor,
acceptTracker.topologies(), route, ballot, MEDIUM, CoordinationFlags.none(),
txnId, txn, executeAt, stableDeps, newDeps, callback);
- else adapter().stabilise(node, executor, acceptTracker.topologies(),
route, ballot, txnId, txn, executeAt, stableDeps, callback);
+ Deps deps = mergeDeps(newDeps);
+ node.agent().coordinatorEvents().onAccepted(txnId, ballot, deps, kind
== Kind.MEDIUM);
Review Comment:
similar comment to other method, maybe pass in `kind` rather than `boolean
isStable`?
##########
accord-core/src/main/java/accord/coordinate/CoordinateTransaction.java:
##########
@@ -128,8 +129,8 @@ void onPreAccepted(Topologies topologies, Timestamp
executeAt, SortedListMap<Nod
// note: we merge all Deps regardless of witnessedAt. While we
only need fast path votes,
// 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, deps,
true);
Review Comment:
wonder if enum would help here? Had to go read the method signature to see
that this is `isStable`
##########
accord-core/src/main/java/accord/coordinate/AbstractCoordinatePreAccept.java:
##########
@@ -114,16 +114,6 @@ void setFailure(Throwable failure)
// we may already be complete, as we may receive a failure from a
later phase; but it's fine to redundantly mark done
isDone = true;
callback.accept(null, failure);
- if (failure instanceof CoordinationFailed)
Review Comment:
why was this removed? Is the assumption that the caller to `coordinator`
will see this failure so can handle that there?
##########
accord-core/src/main/java/accord/coordinate/CoordinateSyncPoint.java:
##########
@@ -168,9 +167,8 @@ void onPreAccepted(Topologies topologies, Timestamp
executeAt, SortedListMap<Nod
if (txnId.is(ExclusiveSyncPoint) && txnId.epoch() ==
executeAt.epoch())
withFlags = txnId.addFlag(HLC_BOUND);
Deps deps = Deps.merge(oks.valuesAsNullableList(),
oks.domainSize(), List::get, ok -> ok.deps);
- if (tracker.hasFastPathAccepted())
- adapter.execute(node, executor, topologies, route,
Ballot.ZERO, FAST, CoordinationFlags.none(), txnId, txn, withFlags, deps, deps,
callback);
Review Comment:
why was fast path removed?
##########
accord-core/src/main/java/accord/coordinate/CoordinateTransaction.java:
##########
@@ -138,21 +139,21 @@ else if (tracker.hasMediumPathAccepted() &&
txnId.hasMediumPath())
Deps deps = mergeFastOrMediumDeps(oks);
if (deps != null)
{
- proposeAdapter().propose(node, executor, topologies, route,
Accept.Kind.MEDIUM, Ballot.ZERO, txnId, txn, txnId, deps, callback);
- node.agent().eventListener().onMediumPathTaken(txnId, deps);
+ node.agent().coordinatorEvents().onPreAccepted(txnId, deps,
false);
+ proposeAdapter().propose(node, executor, topologies, route,
MEDIUM, Ballot.ZERO, txnId, txn, txnId, deps, callback);
return;
}
}
else if (executeAt.is(REJECTED))
{
proposeAndCommitInvalidate(node, executor, Ballot.ZERO, txnId,
route.homeKey(), route, executeAt, callback);
- node.agent().eventListener().onRejected(txnId);
+ node.agent().coordinatorEvents().onRejected(txnId);
return;
}
Deps deps = Deps.merge(oks.valuesAsNullableList(), oks.domainSize(),
List::get, ok -> ok.deps);
- proposeAdapter().propose(node, executor, topologies, route,
Accept.Kind.SLOW, Ballot.ZERO, txnId, txn, executeAt, deps, callback);
- node.agent().eventListener().onSlowPathTaken(txnId, deps);
+ node.agent().coordinatorEvents().onPreAccepted(txnId, deps, false);
Review Comment:
should we the enum showing fast/med/slow so we can pass this detail into
`onPreAccepted`? We had 3 methods before and they get replaced with this
single one, which looks like we loose that level of detail?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscr...@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscr...@cassandra.apache.org
For additional commands, e-mail: pr-h...@cassandra.apache.org
