ifesdjeen commented on code in PR #168:
URL: https://github.com/apache/cassandra-accord/pull/168#discussion_r1953975419
##########
accord-core/src/main/java/accord/coordinate/ExecuteTxn.java:
##########
@@ -192,11 +198,25 @@ protected Action process(Id from, ReadReply reply)
CommitOrReadNack nack = (CommitOrReadNack) reply;
switch (nack)
{
- default: throw new IllegalStateException();
+ default: throw new UnhandledEnum(nack);
+ case Waiting:
+ if (from.id != node.id().id)
+ throw new UnhandledEnum(nack);
Review Comment:
nit: should we throw unhandled enum here, or maybe something more
informative, like node id mismatch?
##########
accord-core/src/main/java/accord/coordinate/CollectLatestDeps.java:
##########
@@ -95,10 +109,15 @@ public void onFailure(Id from, Throwable failure)
if (isDone)
return;
+ if (failure != null)
+ this.failure = FailureAccumulator.append(this.failure, failure);
+
if (tracker.recordFailure(from) == Failed)
{
isDone = true;
- callback.accept(null, new Timeout(txnId, route.homeKey()));
+ if (this.failure == null)
Review Comment:
looks like above we can report Failure as a replay, so it not necessarily is
a timeout, or?
##########
accord-core/src/main/java/accord/impl/CommandChange.java:
##########
@@ -264,84 +271,53 @@ public Cleanup shouldCleanup(Input input, Agent agent,
RedundantBefore redundant
return cleanup;
}
- public Builder maybeCleanup(Cleanup cleanup)
+ public boolean maybeCleanup(Input input, Agent agent, RedundantBefore
redundantBefore, DurableBefore durableBefore)
{
- if (saveStatus == null)
- return this;
-
- switch (cleanup)
- {
- case EXPUNGE:
- case ERASE:
- return null;
-
- case VESTIGIAL:
- case INVALIDATE:
- return saveStatusOnly();
-
- case TRUNCATE_WITH_OUTCOME:
- case TRUNCATE:
- return expungePartial(cleanup, cleanup.appliesIfNot,
cleanup == TRUNCATE_WITH_OUTCOME);
-
- case NO:
- return this;
- default:
- throw new UnhandledEnum(cleanup);
- }
+ Cleanup cleanup = shouldCleanup(input, agent, redundantBefore,
durableBefore);
+ return maybeCleanup(cleanup);
}
- public Builder expungePartial(Cleanup cleanup, SaveStatus saveStatus,
boolean includeOutcome)
+ public boolean maybeCleanup(Cleanup cleanup)
{
- Invariants.require(txnId != null);
- Builder builder = new Builder(txnId, ALL);
+ if (saveStatus == null)
+ return false;
- builder.count++;
- builder.nextCalled = true;
+ cleanup = cleanup.filter(saveStatus);
+ if (cleanup == NO)
+ return false;
- Invariants.require(saveStatus != null);
- builder.flags = setChanged(SAVE_STATUS, builder.flags);
- builder.saveStatus = saveStatus;
- builder.flags = setChanged(CLEANUP, builder.flags);
- builder.cleanup = cleanup;
- if (executeAt != null)
- {
- builder.flags = setChanged(EXECUTE_AT, builder.flags);
- builder.executeAt = executeAt;
- }
- if (durability != null)
- {
- builder.flags = setChanged(DURABILITY, builder.flags);
- builder.durability = durability;
- }
- if (participants != null)
- {
- builder.flags = setChanged(PARTICIPANTS, builder.flags);
- builder.participants = participants;
- }
- if (includeOutcome && builder.writes != null)
- {
- builder.flags = setChanged(WRITES, builder.flags);
- builder.writes = writes;
- }
-
- return builder;
+ truncate(cleanup.appliesIfNot);
+ return true;
}
- public Builder saveStatusOnly()
+ protected void truncate(SaveStatus newSaveStatus)
{
- Invariants.require(txnId != null);
- Builder builder = new Builder(txnId, ALL);
-
- builder.count++;
- builder.nextCalled = true;
-
- if (saveStatus != null)
+ int mask = saveStatusMasks[newSaveStatus.ordinal()];
+ // low flag bits represent fields already nulled out, so no need
to visit them again
+ int iterable = toIterableSetFields(mask) & ~flags;
+ for (Field next = nextSetField(iterable); next != null; iterable =
unsetIterable(next, iterable), next = nextSetField(iterable))
Review Comment:
nice, i really like this new implementation of `truncate`
##########
accord-core/src/main/java/accord/coordinate/ProposeOnly.java:
##########
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package accord.coordinate;
+
+import java.util.function.BiConsumer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import accord.local.Node;
+import accord.messages.Accept;
+import accord.primitives.Ballot;
+import accord.primitives.Deps;
+import accord.primitives.FullRoute;
+import accord.primitives.Route;
+import accord.primitives.Timestamp;
+import accord.primitives.Txn;
+import accord.primitives.TxnId;
+import accord.topology.Topologies;
+
+public class ProposeOnly extends Propose<Deps>
+{
+ @SuppressWarnings("unused")
+ private static final Logger logger =
LoggerFactory.getLogger(ProposeOnly.class);
+
+ ProposeOnly(Node node, Topologies topologies, Route<?> sendTo,
FullRoute<?> route, Accept.Kind kind, Ballot ballot, TxnId txnId, Txn txn,
Timestamp executeAt, Deps deps, BiConsumer<? super Deps, Throwable> callback)
+ {
+ super(node, topologies, kind, ballot, txnId, txn, sendTo, route,
executeAt, deps, callback);
+ }
+
+ @Override
+ void onAccepted()
+ {
+ Deps deps = mergeDeps();
+ callback.accept(deps, null);
+ }
+
+ @Override
+ CoordinationAdapter<Deps> adapter()
+ {
+ throw new UnsupportedOperationException();
Review Comment:
I might be missing context, but could you elaborate why we do not need an
adapter for this one?
##########
accord-core/src/main/java/accord/coordinate/Propose.java:
##########
@@ -102,32 +108,37 @@ public void onSuccess(Id from, AcceptReply reply)
if (isDone)
return;
- switch (reply.outcome())
+ switch (reply.outcome)
{
default: throw new AssertionError("Unhandled AcceptOutcome: " +
reply.outcome());
- case Truncated:
- isDone = true;
- callback.accept(null, new Truncated(txnId, route.homeKey()));
- break;
-
case RejectedBallot:
isDone = true;
callback.accept(null, new Preempted(txnId, route.homeKey()));
break;
+ case Truncated:
case Redundant:
- if (reply.supersededBy != null || ballot.equals(Ballot.ZERO))
+ if (sendTo == route || !isSufficientPartialReply(reply, from))
{
- isDone = true;
- callback.accept(null, new Preempted(txnId,
route.homeKey()));
+ Throwable failNow = reply.outcome ==
AcceptOutcome.Truncated
Review Comment:
could we unwrap this ternary?
```
if (reply.outcome == AcceptOutcome.Truncated)
failNow = new Truncated(txnId, route.homeKey());
else if (reply.supersededBy != null ||
ballot.equals(Ballot.ZERO))
failNow = new Preempted(txnId, route.homeKey());
else if (reply.committedExecuteAt != null)
failNow = new Redundant(txnId, route.homeKey(),
reply.committedExecuteAt);
else
failNow = null;
```
(please feel free to ignore if i am the only person who struggles parsing
long ternaries)
##########
accord-core/src/main/java/accord/coordinate/Propose.java:
##########
@@ -140,17 +151,27 @@ else if (reply.committedExecuteAt != null)
}
}
+ private boolean isSufficientPartialReply(AcceptReply reply, Id from)
+ {
+ return reply.successful != null &&
reply.successful.containsAll(sendTo.slice(acceptTracker.topologies().computeRangesForNode(from),
Minimal));
+ }
+
@Override
public void onFailure(Id from, Throwable failure)
{
+ // TODO (desired): this is a common pattern, find a way to more fully
share it
if (isDone)
return;
- // TODO (required): we aren't tracking the specific failure here to
report
+ if (failure != null)
+ this.failure = FailureAccumulator.append(this.failure, failure);
+
if (acceptTracker.recordFailure(from) == Failed)
{
isDone = true;
- callback.accept(null, new Timeout(txnId, route.homeKey()));
+ if (this.failure == null)
+ this.failure = new Timeout(txnId, route.homeKey());
Review Comment:
same question about timeout here: since we are calling `onFailure` with
`null` failure from success on nack, is it right to replace failure with
timeout here?
##########
accord-core/src/main/java/accord/coordinate/CoordinationAdapter.java:
##########
@@ -133,16 +136,33 @@ public static CoordinationAdapter<Result>
recoverExclusiveSyncPoint()
return RecoverExclusiveSyncPointAdapter.INSTANCE;
}
- public static abstract class AbstractTxnAdapter implements
CoordinationAdapter<Result>
+ public static class TxnAdapter implements CoordinationAdapter<Result>
{
+ static final TxnAdapter STANDARD = new TxnAdapter(Minimal);
+ static final TxnAdapter RECOVERY = new TxnAdapter(Maximal);
+
+ final Apply.Kind applyKind;
+ public TxnAdapter(Apply.Kind applyKind)
+ {
+ this.applyKind = applyKind;
+ }
+
@Override
public void propose(Node node, @Nullable Topologies
preacceptOrRecovery, FullRoute<?> route, Accept.Kind kind, Ballot ballot, TxnId
txnId, Txn txn, Timestamp executeAt, Deps deps, BiConsumer<? super Result,
Throwable> callback)
{
Topologies accept =
node.topology().reselect(preacceptOrRecovery,
QuorumEpochIntersections.preacceptOrRecover,
- route, txnId,
executeAt, QuorumEpochIntersections.accept);
+ route, txnId,
executeAt, SHARE, QuorumEpochIntersections.accept);
new ProposeTxn(node, accept, route, kind, ballot, txnId, txn,
executeAt, deps, callback).start();
}
+ @Override
+ public void proposeOnly(Node node, Route<?> require, Route<?>
sendTo, SelectNodeOwnership selectNodeOwnership, FullRoute<?> route,
Accept.Kind kind, Ballot ballot, TxnId txnId, Txn txn, Timestamp executeAt,
Deps deps, BiConsumer<? super Deps, Throwable> callback)
+ {
+ Topologies accept = node.topology().reselect(null,
QuorumEpochIntersections.preacceptOrRecover,
+ sendTo, txnId,
executeAt, selectNodeOwnership, QuorumEpochIntersections.accept);
+ new ProposeOnly(node, accept, sendTo, route, kind, ballot,
txnId, txn, executeAt, deps, callback).start();
Review Comment:
nit: looks like this `start` might be quite easy to miss. If we always call
it (which we seem to do), should we maybe move the start call to constructor,
or maybe make a static method?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
