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 2024/11/11 11:49:04 UTC
(cassandra-accord) 01/01: Facilitate queueing, execution and performance improvements
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-accord.git
commit d13b88fd48aebf56838fb7f820faa406aeaa481f
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Thu Oct 17 17:26:14 2024 +0100
Facilitate queueing, execution and performance improvements
patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20069
---
accord-core/src/main/java/accord/api/Agent.java | 8 +-
.../main/java/accord/api/ConfigurationService.java | 8 +-
.../api/{RequestTimeouts.java => Timeouts.java} | 17 +-
.../coordinate/AbstractCoordinatePreAccept.java | 2 +-
.../src/main/java/accord/coordinate/Barrier.java | 2 +-
.../accord/coordinate/CoordinatePreAccept.java | 1 -
.../accord/coordinate/ExecuteEphemeralRead.java | 2 +-
.../main/java/accord/coordinate/ExecuteTxn.java | 19 ++-
.../main/java/accord/coordinate/MaybeRecover.java | 3 +-
.../src/main/java/accord/coordinate/Recover.java | 2 +-
.../src/main/java/accord/coordinate/Stabilise.java | 2 +-
.../coordinate/tracking/FastPathTracker.java | 2 +-
.../accord/impl/AbstractConfigurationService.java | 8 +-
.../java/accord/impl/AbstractSafeCommandStore.java | 69 ++++----
...tRequestTimeouts.java => AbstractTimeouts.java} | 25 ++-
.../java/accord/impl/DefaultLocalListeners.java | 8 +-
.../java/accord/impl/DefaultRemoteListeners.java | 2 +-
...ltRequestTimeouts.java => DefaultTimeouts.java} | 8 +-
.../java/accord/impl/DurabilityScheduling.java | 38 ++++-
.../java/accord/impl/InMemoryCommandStore.java | 48 ++++--
.../main/java/accord/impl/RequestCallbacks.java | 3 +-
.../main/java/accord/impl/TimestampsForKeys.java | 2 +-
.../accord/impl/progresslog/CallbackInvoker.java | 11 +-
.../impl/progresslog/DefaultProgressLog.java | 41 +++--
.../java/accord/impl/progresslog/HomeState.java | 8 +-
.../src/main/java/accord/local/Cleanup.java | 30 ++--
.../src/main/java/accord/local/Command.java | 15 ++
.../src/main/java/accord/local/CommandStore.java | 51 ++++--
.../src/main/java/accord/local/CommandStores.java | 28 ++--
.../src/main/java/accord/local/Commands.java | 48 ++++--
.../src/main/java/accord/local/KeyHistory.java | 32 +++-
accord-core/src/main/java/accord/local/Node.java | 56 ++++---
.../java/accord/local/NodeCommandStoreService.java | 2 +
.../src/main/java/accord/local/PreLoadContext.java | 143 ++++++++--------
.../main/java/accord/local/RedundantBefore.java | 90 +++++++---
.../main/java/accord/local/SafeCommandStore.java | 182 +++++++++++++++------
.../main/java/accord/local/ShardDistributor.java | 9 +
.../main/java/accord/local/cfk/CommandsForKey.java | 16 +-
.../src/main/java/accord/local/cfk/NotifySink.java | 6 +-
.../main/java/accord/local/cfk/PostProcess.java | 8 +-
.../src/main/java/accord/local/cfk/Updating.java | 4 +-
.../java/accord/messages/AbstractEpochRequest.java | 67 --------
.../main/java/accord/messages/AbstractRequest.java | 180 ++++++++++++++++++++
.../src/main/java/accord/messages/Accept.java | 19 +--
.../src/main/java/accord/messages/Apply.java | 13 +-
.../src/main/java/accord/messages/Await.java | 19 ++-
.../java/accord/messages/BeginInvalidation.java | 7 +-
.../main/java/accord/messages/BeginRecovery.java | 13 +-
.../main/java/accord/messages/CalculateDeps.java | 23 +--
.../src/main/java/accord/messages/CheckStatus.java | 10 +-
.../src/main/java/accord/messages/Commit.java | 76 +++++----
.../java/accord/messages/GetEphemeralReadDeps.java | 13 +-
.../main/java/accord/messages/GetMaxConflict.java | 11 +-
.../main/java/accord/messages/InformDurable.java | 11 +-
.../src/main/java/accord/messages/PreAccept.java | 13 +-
.../src/main/java/accord/messages/Propagate.java | 14 --
.../src/main/java/accord/messages/ReadData.java | 138 ++++++++--------
.../java/accord/messages/ReadEphemeralTxnData.java | 53 +++++-
.../src/main/java/accord/messages/ReadTxnData.java | 1 -
.../src/main/java/accord/messages/Request.java | 1 -
.../main/java/accord/messages/SetShardDurable.java | 8 +-
.../src/main/java/accord/messages/TxnRequest.java | 25 +--
.../src/main/java/accord/primitives/Keys.java | 14 +-
.../main/java/accord/primitives/Participants.java | 1 +
.../src/main/java/accord/primitives/Route.java | 1 +
.../main/java/accord/primitives/RoutingKeys.java | 7 +
.../src/main/java/accord/utils/ArrayBuffers.java | 12 ++
.../java/accord/utils/IntrusiveLinkedList.java | 13 +-
.../java/accord/utils/IntrusivePriorityHeap.java | 16 +-
.../src/main/java/accord/utils/Invariants.java | 31 ++++
.../src/main/java/accord/utils/LogGroupTimers.java | 1 +
.../utils/MergeFewDisjointSortedListsCursor.java | 135 ++++++++-------
.../Request.java => utils/QuadConsumer.java} | 12 +-
.../Request.java => utils/QuintConsumer.java} | 12 +-
.../main/java/accord/utils/RelationMultiMap.java | 25 +--
.../src/main/java/accord/utils/SimpleBitSet.java | 14 +-
.../src/main/java/accord/utils/SortedArrays.java | 10 ++
.../UncheckedInterruptedException.java} | 31 ++--
.../main/java/accord/utils/async/AsyncChain.java | 2 +
accord-core/src/test/java/accord/Utils.java | 4 +-
.../src/test/java/accord/burn/BurnTest.java | 2 +-
.../java/accord/coordinate/TopologyChangeTest.java | 12 +-
.../test/java/accord/impl/RemoteListenersTest.java | 9 +-
.../src/test/java/accord/impl/TestAgent.java | 10 +-
.../src/test/java/accord/impl/basic/Cluster.java | 4 +-
.../accord/impl/basic/DelayedCommandStores.java | 5 +-
.../src/test/java/accord/impl/basic/Journal.java | 2 +-
.../src/test/java/accord/impl/list/ListAgent.java | 14 +-
.../test/java/accord/impl/mock/MockCluster.java | 4 +-
.../java/accord/local/ImmutableCommandTest.java | 4 +-
.../java/accord/local/cfk/CommandsForKeyTest.java | 33 ++--
.../test/java/accord/messages/PreAcceptTest.java | 2 +-
.../test/java/accord/utils/LogGroupTimersTest.java | 4 +-
.../src/main/java/accord/maelstrom/Cluster.java | 4 +-
.../main/java/accord/maelstrom/MaelstromAgent.java | 4 +-
.../src/main/java/accord/maelstrom/Main.java | 4 +-
96 files changed, 1388 insertions(+), 844 deletions(-)
diff --git a/accord-core/src/main/java/accord/api/Agent.java b/accord-core/src/main/java/accord/api/Agent.java
index 3e352431..3d44358c 100644
--- a/accord-core/src/main/java/accord/api/Agent.java
+++ b/accord-core/src/main/java/accord/api/Agent.java
@@ -32,6 +32,8 @@ import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.topology.Topologies;
+import static accord.utils.Invariants.illegalState;
+
/**
* Facility for augmenting node behaviour at specific points
*
@@ -122,8 +124,6 @@ public interface Agent extends UncaughtExceptionListener
*/
default Topologies selectPreferred(Node.Id from, Topologies to) { return to; }
- long replyTimeout(ReplyContext replyContext, TimeUnit units);
-
/**
* This method permits implementations to configure the time at which a local home shard will attempt
* to coordinate a transaction to completion.
@@ -148,4 +148,8 @@ public interface Agent extends UncaughtExceptionListener
* and re-query the local state.
*/
long retryAwaitTimeout(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, BlockedUntil retrying, TimeUnit units);
+
+ long expiresAt(ReplyContext replyContext, TimeUnit unit);
+
+ default void onViolation(String message) { throw illegalState(message); }
}
diff --git a/accord-core/src/main/java/accord/api/ConfigurationService.java b/accord-core/src/main/java/accord/api/ConfigurationService.java
index b7e7ee83..cfe52546 100644
--- a/accord-core/src/main/java/accord/api/ConfigurationService.java
+++ b/accord-core/src/main/java/accord/api/ConfigurationService.java
@@ -74,9 +74,9 @@ public interface ConfigurationService
public final AsyncResult<Void> metadata;
/**
- * The node has retrieved enough remote information to become a coordinator for the new epoch.
+ * The node has retrieved enough remote information to become a fast path coordinator for the new epoch.
*/
- public final AsyncResult<Void> coordination;
+ public final AsyncResult<Void> fastPath;
/**
* The node has successfully replicated the underlying DataStore information for the new epoch, but may need
@@ -90,11 +90,11 @@ public interface ConfigurationService
*/
public final AsyncResult<Void> reads;
- public EpochReady(long epoch, AsyncResult<Void> metadata, AsyncResult<Void> coordination, AsyncResult<Void> data, AsyncResult<Void> reads)
+ public EpochReady(long epoch, AsyncResult<Void> metadata, AsyncResult<Void> fastPath, AsyncResult<Void> data, AsyncResult<Void> reads)
{
this.epoch = epoch;
this.metadata = metadata;
- this.coordination = coordination;
+ this.fastPath = fastPath;
this.data = data;
this.reads = reads;
}
diff --git a/accord-core/src/main/java/accord/api/RequestTimeouts.java b/accord-core/src/main/java/accord/api/Timeouts.java
similarity index 77%
copy from accord-core/src/main/java/accord/api/RequestTimeouts.java
copy to accord-core/src/main/java/accord/api/Timeouts.java
index 83c3d2bc..8a96840a 100644
--- a/accord-core/src/main/java/accord/api/RequestTimeouts.java
+++ b/accord-core/src/main/java/accord/api/Timeouts.java
@@ -20,19 +20,22 @@ package accord.api;
import java.util.concurrent.TimeUnit;
-public interface RequestTimeouts
-{
- interface RegisteredTimeout
- {
- void cancel();
- }
+import accord.utils.async.Cancellable;
+public interface Timeouts
+{
interface Timeout
{
void timeout();
int stripe();
}
- RegisteredTimeout register(Timeout timeout, long delay, TimeUnit units);
+ interface RegisteredTimeout extends Cancellable
+ {
+ void cancel();
+ }
+
+ RegisteredTimeout registerWithDelay(Timeout timeout, long delay, TimeUnit units);
+ RegisteredTimeout registerAt(Timeout timeout, long deadline, TimeUnit units);
void maybeNotify();
}
diff --git a/accord-core/src/main/java/accord/coordinate/AbstractCoordinatePreAccept.java b/accord-core/src/main/java/accord/coordinate/AbstractCoordinatePreAccept.java
index 1dd04ad6..e6f45e3a 100644
--- a/accord-core/src/main/java/accord/coordinate/AbstractCoordinatePreAccept.java
+++ b/accord-core/src/main/java/accord/coordinate/AbstractCoordinatePreAccept.java
@@ -208,7 +208,7 @@ abstract class AbstractCoordinatePreAccept<T, R> extends SettableResult<T> imple
{
long latestEpoch = executeAtEpoch();
if (latestEpoch > topologies.currentEpoch()) node.withEpoch(latestEpoch, node.agent(), () -> onPreAcceptedInNewEpoch(topologies, latestEpoch));
- else onPreAcceptedInNewEpoch(topologies, latestEpoch);
+ else onPreAccepted(topologies);
return;
}
diff --git a/accord-core/src/main/java/accord/coordinate/Barrier.java b/accord-core/src/main/java/accord/coordinate/Barrier.java
index 54de2fa2..ab794c7b 100644
--- a/accord-core/src/main/java/accord/coordinate/Barrier.java
+++ b/accord-core/src/main/java/accord/coordinate/Barrier.java
@@ -233,7 +233,7 @@ public class Barrier extends AsyncResults.AbstractResult<TxnId>
ExistingTransactionCheck check = new ExistingTransactionCheck();
RoutingKey k = route.get(0).asRoutingKey();
node.commandStores().mapReduceConsume(
- contextFor(k, KeyHistory.COMMANDS),
+ contextFor(k, KeyHistory.SYNC),
k.toUnseekable(),
minEpoch,
Long.MAX_VALUE,
diff --git a/accord-core/src/main/java/accord/coordinate/CoordinatePreAccept.java b/accord-core/src/main/java/accord/coordinate/CoordinatePreAccept.java
index 68a5c7e8..baeca339 100644
--- a/accord-core/src/main/java/accord/coordinate/CoordinatePreAccept.java
+++ b/accord-core/src/main/java/accord/coordinate/CoordinatePreAccept.java
@@ -52,7 +52,6 @@ import static accord.utils.Functions.foldl;
*/
abstract class CoordinatePreAccept<T> extends AbstractCoordinatePreAccept<T, PreAcceptReply>
{
- // TODO (required): do not use FastPathTracker for coordinating exclusive sync points - these always take the slow path
final PreAcceptTracker<?> tracker;
// TODO (expected): this can be cleared after preaccept
// TODO (expected): back by SortedListMap; must handle additional preaccepts (but this is no longer ordinarily enabled)
diff --git a/accord-core/src/main/java/accord/coordinate/ExecuteEphemeralRead.java b/accord-core/src/main/java/accord/coordinate/ExecuteEphemeralRead.java
index e0f11de2..d518595d 100644
--- a/accord-core/src/main/java/accord/coordinate/ExecuteEphemeralRead.java
+++ b/accord-core/src/main/java/accord/coordinate/ExecuteEphemeralRead.java
@@ -98,7 +98,7 @@ public class ExecuteEphemeralRead extends ReadCoordinator<ReadReply>
if (ok.futureEpoch > allTopologies.currentEpoch())
{
// TODO (expected): only submit new requests for the keys that execute in a later epoch
- new ExecuteEphemeralRead(node, node.topology().preciseEpochs(route, ok.futureEpoch, ok.futureEpoch), route, txnId, txn, ok.futureEpoch, deps, callback).start();
+ new ExecuteEphemeralRead(node, node.topology().preciseEpochs(route, ok.futureEpoch, ok.futureEpoch), route, txnId.withEpoch(ok.futureEpoch), txn, ok.futureEpoch, deps, callback).start();
return Aborted;
}
diff --git a/accord-core/src/main/java/accord/coordinate/ExecuteTxn.java b/accord-core/src/main/java/accord/coordinate/ExecuteTxn.java
index 089b6816..ca41b9a7 100644
--- a/accord-core/src/main/java/accord/coordinate/ExecuteTxn.java
+++ b/accord-core/src/main/java/accord/coordinate/ExecuteTxn.java
@@ -54,6 +54,8 @@ public class ExecuteTxn extends ReadCoordinator<ReadReply>
{
@SuppressWarnings("unused")
private static final Logger logger = LoggerFactory.getLogger(ExecuteTxn.class);
+ private static boolean SEND_MINIMUM_STABLE_MESSAGES = true;
+ public static void setSendMinimumStableMessages(boolean sendMin) {SEND_MINIMUM_STABLE_MESSAGES = sendMin; }
final ExecutePath path;
final Txn txn;
@@ -85,22 +87,25 @@ public class ExecuteTxn extends ReadCoordinator<ReadReply>
{
IntHashSet readSet = new IntHashSet();
to.forEach(i -> readSet.add(i.id));
- Commit.Kind kind;
+ Commit.stableAndRead(node, allTopologies, commitKind(), txnId, txn, route, readScope, executeAt, stableDeps, readSet, this, SEND_MINIMUM_STABLE_MESSAGES);
+ }
+
+ private Commit.Kind commitKind()
+ {
switch (path)
{
default: throw new AssertionError("Unhandled path: " + path);
- case FAST: kind = StableFastPath; break;
- case SLOW: kind = StableSlowPath; break;
- case RECOVER: kind = StableWithTxnAndDeps; break;
+ case FAST: return StableFastPath;
+ case SLOW: return StableSlowPath;
+ case RECOVER: return StableWithTxnAndDeps;
}
- // we want to send to all topologies, but we only want to track responses from the readScope
- Commit.stableAndRead(node, allTopologies, kind, txnId, txn, route, readScope, executeAt, stableDeps, readSet, this);
}
@Override
public void contact(Id to)
{
- node.send(to, new ReadTxnData(to, topologies(), txnId, readScope, executeAt.epoch()), this);
+ if (SEND_MINIMUM_STABLE_MESSAGES) Commit.stableAndRead(to, node, allTopologies, commitKind(), txnId, txn, route, readScope, executeAt, stableDeps, this, SEND_MINIMUM_STABLE_MESSAGES);
+ else node.send(to, new ReadTxnData(to, topologies(), txnId, readScope, executeAt.epoch()), this);
}
@Override
diff --git a/accord-core/src/main/java/accord/coordinate/MaybeRecover.java b/accord-core/src/main/java/accord/coordinate/MaybeRecover.java
index b568a65a..34de22f1 100644
--- a/accord-core/src/main/java/accord/coordinate/MaybeRecover.java
+++ b/accord-core/src/main/java/accord/coordinate/MaybeRecover.java
@@ -47,10 +47,11 @@ public class MaybeRecover extends CheckShards<Route<?>>
this.callback = callback;
}
- public static void maybeRecover(Node node, TxnId txnId, Route<?> someRoute, ProgressToken prevProgress, BiConsumer<Outcome, Throwable> callback)
+ public static Object maybeRecover(Node node, TxnId txnId, Route<?> someRoute, ProgressToken prevProgress, BiConsumer<Outcome, Throwable> callback)
{
MaybeRecover maybeRecover = new MaybeRecover(node, txnId, someRoute, prevProgress, callback);
maybeRecover.start();
+ return maybeRecover;
}
@Override
diff --git a/accord-core/src/main/java/accord/coordinate/Recover.java b/accord-core/src/main/java/accord/coordinate/Recover.java
index 09ec89e5..dcbf9417 100644
--- a/accord-core/src/main/java/accord/coordinate/Recover.java
+++ b/accord-core/src/main/java/accord/coordinate/Recover.java
@@ -140,7 +140,7 @@ public class Recover implements Callback<RecoverReply>, BiConsumer<Result, Throw
node.agent().metricsEventsListener().onPreempted(txnId);
else if (failure instanceof Timeout)
node.agent().metricsEventsListener().onTimeout(txnId);
- else if (failure instanceof Invalidated)
+ else if (failure instanceof Invalidated) // TODO (expected): should we tick this counter? we haven't invalidated anything
node.agent().metricsEventsListener().onInvalidated(txnId);
}
diff --git a/accord-core/src/main/java/accord/coordinate/Stabilise.java b/accord-core/src/main/java/accord/coordinate/Stabilise.java
index 05993692..1300da86 100644
--- a/accord-core/src/main/java/accord/coordinate/Stabilise.java
+++ b/accord-core/src/main/java/accord/coordinate/Stabilise.java
@@ -83,7 +83,7 @@ public abstract class Stabilise<R> implements Callback<ReadReply>
contact = contact.with(allTopologies.nodes().without(stableTracker.nodes()).without(allTopologies::isFaulty));
if (contact == null) callback.accept(null, new Exhausted(txnId, route.homeKey(), null));
- else Commit.commitMinimal(contact, node, stableTracker.topologies(), allTopologies, ballot, txnId, txn, route, executeAt, stabiliseDeps, this);
+ else Commit.commitMinimalNoRead(contact, node, stableTracker.topologies(), allTopologies, ballot, txnId, txn, route, executeAt, stabiliseDeps, this);
}
@Override
diff --git a/accord-core/src/main/java/accord/coordinate/tracking/FastPathTracker.java b/accord-core/src/main/java/accord/coordinate/tracking/FastPathTracker.java
index d3c503b8..d42e198a 100644
--- a/accord-core/src/main/java/accord/coordinate/tracking/FastPathTracker.java
+++ b/accord-core/src/main/java/accord/coordinate/tracking/FastPathTracker.java
@@ -104,7 +104,7 @@ public class FastPathTracker extends PreAcceptTracker<FastPathTracker.FastPathSh
final boolean isNewFastPathSuccess()
{
- return fastPathAccepts == shard.fastPathQuorumSize && fastPathDelayed < shard.fastPathRejectSize;
+ return fastPathAccepts == shard.fastPathQuorumSize && (fastPathDelayed < shard.fastPathRejectSize || isNewSlowPathSuccess());
}
@VisibleForTesting
diff --git a/accord-core/src/main/java/accord/impl/AbstractConfigurationService.java b/accord-core/src/main/java/accord/impl/AbstractConfigurationService.java
index bb70b14a..7edc3114 100644
--- a/accord-core/src/main/java/accord/impl/AbstractConfigurationService.java
+++ b/accord-core/src/main/java/accord/impl/AbstractConfigurationService.java
@@ -150,6 +150,7 @@ public abstract class AbstractConfigurationService<EpochState extends AbstractCo
public void receive(Topology topology)
{
long epoch = topology.epoch();
+ logger.debug("Receiving epoch {}", epoch);
Invariants.checkState(lastReceived == epoch - 1 || epoch == 0 || lastReceived == 0,
"Epoch %d != %d + 1", epoch, lastReceived);
lastReceived = epoch;
@@ -171,6 +172,7 @@ public abstract class AbstractConfigurationService<EpochState extends AbstractCo
public void acknowledge(EpochReady ready)
{
long epoch = ready.epoch;
+ logger.debug("Acknowledging epoch {}", epoch);
Invariants.checkState(lastAcknowledged == epoch - 1 || epoch == 0 || lastAcknowledged == 0,
"Epoch %d != %d + 1", epoch, lastAcknowledged);
lastAcknowledged = epoch;
@@ -256,7 +258,7 @@ public abstract class AbstractConfigurationService<EpochState extends AbstractCo
epochs.acknowledge(ready);
}
});
- ready.coordination.addCallback(() -> {
+ ready.fastPath.addCallback(() -> {
synchronized (AbstractConfigurationService.this)
{
localSyncComplete(epochs.getOrCreate(ready.epoch).topology, startSync);
@@ -275,6 +277,7 @@ public abstract class AbstractConfigurationService<EpochState extends AbstractCo
if (lastReceived > 0 && topology.epoch() > lastReceived + 1)
{
+ logger.debug("Epoch {} received; waiting to receive {} before reporting", topology.epoch(), lastReceived + 1);
fetchTopologyForEpoch(lastReceived + 1);
epochs.receiveFuture(lastReceived + 1).addCallback(() -> reportTopology(topology, startSync, isLoad));
return;
@@ -283,15 +286,16 @@ public abstract class AbstractConfigurationService<EpochState extends AbstractCo
long lastAcked = epochs.lastAcknowledged;
if (lastAcked == 0 && lastReceived > 0)
{
+ logger.debug("Epoch {} received; waiting for {} to ack before reporting", topology.epoch(), epochs.minEpoch());
epochs.acknowledgeFuture(epochs.minEpoch()).addCallback(() -> reportTopology(topology, startSync, isLoad));
return;
}
if (lastAcked > 0 && topology.epoch() > lastAcked + 1)
{
+ logger.debug("Epoch {} received; waiting for {} to ack before reporting", topology.epoch(), lastAcked + 1);
epochs.acknowledgeFuture(lastAcked + 1).addCallback(() -> reportTopology(topology, startSync, isLoad));
return;
}
- logger.trace("Epoch {} received by {}", topology.epoch(), localId);
epochs.receive(topology);
topologyUpdatePreListenerNotify(topology);
diff --git a/accord-core/src/main/java/accord/impl/AbstractSafeCommandStore.java b/accord-core/src/main/java/accord/impl/AbstractSafeCommandStore.java
index fc68cbfd..3523656b 100644
--- a/accord-core/src/main/java/accord/impl/AbstractSafeCommandStore.java
+++ b/accord-core/src/main/java/accord/impl/AbstractSafeCommandStore.java
@@ -44,7 +44,7 @@ public abstract class AbstractSafeCommandStore<CommandType extends SafeCommand,
this.context = context;
}
- private static <K, V> V getIfLoaded(K key, Function<K, V> get, Consumer<V> add, Function<K, V> getIfLoaded)
+ private static <K, V> V getIfLoadedUnsafe(K key, Function<K, V> get, Consumer<V> add, Function<K, V> getIfLoaded)
{
V value = get.apply(key);
if (value != null)
@@ -57,22 +57,22 @@ public abstract class AbstractSafeCommandStore<CommandType extends SafeCommand,
return value;
}
- protected abstract CommandType getCommandInternal(TxnId txnId);
- protected abstract void addCommandInternal(CommandType command);
- protected abstract CommandType getIfLoaded(TxnId txnId);
+ protected abstract CommandType getCommandUnsafe(TxnId txnId);
+ protected abstract void addCommandUnsafe(CommandType command);
+ protected abstract CommandType getIfLoadedUnsafe(TxnId txnId);
- protected abstract TimestampsForKeyType getTimestampsForKeyInternal(RoutingKey key);
- protected abstract void addTimestampsForKeyInternal(TimestampsForKeyType cfk);
- protected abstract TimestampsForKeyType getTimestampsForKeyIfLoaded(RoutingKey key);
+ protected abstract TimestampsForKeyType getTimestampsForKeyUnsafe(RoutingKey key);
+ protected abstract void addTimestampsForKeyUnsafe(TimestampsForKeyType cfk);
+ protected abstract TimestampsForKeyType getTimestampsForKeyIfUnsafe(RoutingKey key);
- protected abstract CommandsForKeyType getCommandsForKeyInternal(RoutingKey key);
- protected abstract void addCommandsForKeyInternal(CommandsForKeyType cfk);
- protected abstract CommandsForKeyType getCommandsForKeyIfLoaded(RoutingKey key);
+ protected abstract CommandsForKeyType getCommandsForKeyUnsafe(RoutingKey key);
+ protected abstract void addCommandsForKeyUnsafe(CommandsForKeyType cfk);
+ protected abstract CommandsForKeyType getCommandsForKeyIfUnsafe(RoutingKey key);
@Override
- protected CommandType getInternalIfLoadedAndInitialised(TxnId txnId)
+ protected CommandType ifLoadedAndInitialisedAndNotErasedInternal(TxnId txnId)
{
- CommandType command = getIfLoaded(txnId, this::getCommandInternal, this::addCommandInternal, this::getIfLoaded);
+ CommandType command = getIfLoadedUnsafe(txnId, this::getCommandUnsafe, this::addCommandUnsafe, this::getIfLoadedUnsafe);
if (command == null || command.isUnset())
return null;
return command;
@@ -81,43 +81,29 @@ public abstract class AbstractSafeCommandStore<CommandType extends SafeCommand,
@Override
public CommandType getInternal(TxnId txnId)
{
- CommandType command = getCommandInternal(txnId);
- if (command == null)
- throw illegalState(format("%s was not specified in PreLoadContext", txnId));
- if (command.isUnset())
- command.uninitialised();
- return command;
+ return getCommandUnsafe(txnId);
}
- private CommandsForKeyType getCommandsIfLoaded(RoutingKey key)
+ private CommandsForKeyType getCommandsIfLoadedUnsafe(RoutingKey key)
{
- return getIfLoaded(key, this::getCommandsForKeyInternal, this::addCommandsForKeyInternal, this::getCommandsForKeyIfLoaded);
+ return getIfLoadedUnsafe(key, this::getCommandsForKeyUnsafe, this::addCommandsForKeyUnsafe, this::getCommandsForKeyIfUnsafe);
}
- protected CommandsForKeyType getInternalIfLoadedAndInitialised(RoutingKey key)
+ protected CommandsForKeyType ifLoadedInternal(RoutingKey key)
{
- CommandsForKeyType cfk = getCommandsIfLoaded(key);
- if (cfk == null)
- return null;
- if (cfk.isUnset())
- cfk.initialize();
- return cfk;
+ return getCommandsIfLoadedUnsafe(key);
}
@VisibleForTesting
protected CommandsForKeyType getInternal(RoutingKey key)
{
- CommandsForKeyType cfk = getCommandsIfLoaded(key);
- Invariants.checkState(cfk != null, "%s was not specified in PreLoadContext", key);
- if (cfk.isUnset())
- cfk.initialize();
- return cfk;
+ return getCommandsIfLoadedUnsafe(key);
}
@VisibleForImplementation
public CommandsForKeyType maybeCommandsForKey(RoutingKey key)
{
- CommandsForKeyType cfk = getCommandsIfLoaded(key);
+ CommandsForKeyType cfk = getCommandsIfLoadedUnsafe(key);
if (cfk == null || cfk.isUnset())
return null;
return cfk;
@@ -125,7 +111,7 @@ public abstract class AbstractSafeCommandStore<CommandType extends SafeCommand,
public TimestampsForKeyType timestampsIfLoadedAndInitialised(RoutingKey key)
{
- TimestampsForKeyType cfk = getIfLoaded(key, this::getTimestampsForKeyInternal, this::addTimestampsForKeyInternal, this::getTimestampsForKeyIfLoaded);
+ TimestampsForKeyType cfk = getIfLoadedUnsafe(key, this::getTimestampsForKeyUnsafe, this::addTimestampsForKeyUnsafe, this::getTimestampsForKeyIfUnsafe);
if (cfk == null)
return null;
if (cfk.isUnset())
@@ -137,7 +123,7 @@ public abstract class AbstractSafeCommandStore<CommandType extends SafeCommand,
public TimestampsForKeyType timestampsForKey(RoutingKey key)
{
- TimestampsForKeyType tfk = getIfLoaded(key, this::getTimestampsForKeyInternal, this::addTimestampsForKeyInternal, this::getTimestampsForKeyIfLoaded);
+ TimestampsForKeyType tfk = getIfLoadedUnsafe(key, this::getTimestampsForKeyUnsafe, this::addTimestampsForKeyUnsafe, this::getTimestampsForKeyIfUnsafe);
Invariants.checkState(tfk != null, "%s was not specified in PreLoadContext", key);
if (tfk.isUnset())
tfk.initialize();
@@ -148,17 +134,22 @@ public abstract class AbstractSafeCommandStore<CommandType extends SafeCommand,
@VisibleForImplementation
public TimestampsForKeyType maybeTimestampsForKey(RoutingKey key)
{
- TimestampsForKeyType tfk = getIfLoaded(key, this::getTimestampsForKeyInternal, this::addTimestampsForKeyInternal, this::getTimestampsForKeyIfLoaded);
+ TimestampsForKeyType tfk = getIfLoadedUnsafe(key, this::getTimestampsForKeyUnsafe, this::addTimestampsForKeyUnsafe, this::getTimestampsForKeyIfUnsafe);
if (tfk == null || tfk.isUnset())
return null;
return tfk;
}
@Override
- public boolean canExecuteWith(PreLoadContext context)
+ public PreLoadContext context()
+ {
+ return this.context;
+ }
+
+ @Override
+ public PreLoadContext canExecute(PreLoadContext context)
{
- // TODO (required): check if data is in cache, and if so simply add it to our context
- return context.isSubsetOf(this.context);
+ return context.isSubsetOf(this.context) ? context : null;
}
public void postExecute()
diff --git a/accord-core/src/main/java/accord/impl/AbstractRequestTimeouts.java b/accord-core/src/main/java/accord/impl/AbstractTimeouts.java
similarity index 87%
rename from accord-core/src/main/java/accord/impl/AbstractRequestTimeouts.java
rename to accord-core/src/main/java/accord/impl/AbstractTimeouts.java
index dd3c55e9..df29d1f8 100644
--- a/accord-core/src/main/java/accord/impl/AbstractRequestTimeouts.java
+++ b/accord-core/src/main/java/accord/impl/AbstractTimeouts.java
@@ -23,14 +23,14 @@ import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;
import java.util.function.IntFunction;
-import accord.api.RequestTimeouts;
+import accord.api.Timeouts;
import accord.local.TimeService;
import accord.utils.ArrayBuffers.BufferList;
import accord.utils.LogGroupTimers;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
-public class AbstractRequestTimeouts<S extends AbstractRequestTimeouts.Stripe> implements RequestTimeouts
+public class AbstractTimeouts<S extends AbstractTimeouts.Stripe> implements Timeouts
{
protected interface Expiring
{
@@ -193,12 +193,12 @@ public class AbstractRequestTimeouts<S extends AbstractRequestTimeouts.Stripe> i
final TimeService time;
final S[] stripes;
- protected AbstractRequestTimeouts(TimeService time, IntFunction<S[]> arrayAllocator, Function<TimeService, S> stripeFactory)
+ protected AbstractTimeouts(TimeService time, IntFunction<S[]> arrayAllocator, Function<TimeService, S> stripeFactory)
{
this(time, 8, arrayAllocator, stripeFactory);
}
- public AbstractRequestTimeouts(TimeService time, int stripeCount, IntFunction<S[]> allocator, Function<TimeService, S> stripeFactory)
+ public AbstractTimeouts(TimeService time, int stripeCount, IntFunction<S[]> allocator, Function<TimeService, S> stripeFactory)
{
if (stripeCount > 1024)
throw new IllegalArgumentException("Far too many stripes requested");
@@ -209,14 +209,27 @@ public class AbstractRequestTimeouts<S extends AbstractRequestTimeouts.Stripe> i
}
@Override
- public RegisteredTimeout register(Timeout timeout, long delay, TimeUnit units)
+ public RegisteredTimeout registerWithDelay(Timeout timeout, long delay, TimeUnit units)
{
long now = time.elapsed(MICROSECONDS);
long deadline = now + Math.max(1, units.toMicros(delay));
+ return registerAt(timeout, now, deadline);
+ }
+
+ @Override
+ public RegisteredTimeout registerAt(Timeout timeout, long deadline, TimeUnit units)
+ {
+ long now = time.elapsed(MICROSECONDS);
+ deadline = units.toMicros(deadline);
+ return registerAt(timeout, now, deadline);
+ }
+
+ private RegisteredTimeout registerAt(Timeout timeout, long nowMicros, long deadlineMicros)
+ {
int i = timeout.stripe() & (stripes.length - 1);
while (true)
{
- RegisteredTimeout result = stripes[i].tryRegister(timeout, now, deadline);
+ RegisteredTimeout result = stripes[i].tryRegister(timeout, nowMicros, deadlineMicros);
if (result != null)
return result;
i = (i + 1) & (stripes.length - 1);
diff --git a/accord-core/src/main/java/accord/impl/DefaultLocalListeners.java b/accord-core/src/main/java/accord/impl/DefaultLocalListeners.java
index bfcdb8f5..daa5593c 100644
--- a/accord-core/src/main/java/accord/impl/DefaultLocalListeners.java
+++ b/accord-core/src/main/java/accord/impl/DefaultLocalListeners.java
@@ -69,6 +69,12 @@ public class DefaultLocalListeners implements LocalListeners
{
void notify(SafeCommandStore safeStore, SafeCommand safeCommand, TxnId listener);
boolean notify(SafeCommandStore safeStore, SafeCommand safeCommand, ComplexListener listener);
+
+ class NoOpNotifySink implements NotifySink
+ {
+ @Override public void notify(SafeCommandStore safeStore, SafeCommand safeCommand, TxnId listener) {}
+ @Override public boolean notify(SafeCommandStore safeStore, SafeCommand safeCommand, LocalListeners.ComplexListener listener) { return false; }
+ }
}
public static class DefaultNotifySink implements NotifySink
@@ -78,7 +84,7 @@ public class DefaultLocalListeners implements LocalListeners
@Override
public void notify(SafeCommandStore safeStore, SafeCommand safeCommand, TxnId listenerId)
{
- SafeCommand listener = safeStore.ifLoadedAndInitialised(listenerId);
+ SafeCommand listener = safeStore.ifLoadedAndInitialisedAndNotErased(listenerId);
if (listener != null) Commands.listenerUpdate(safeStore, listener, safeCommand);
else
{
diff --git a/accord-core/src/main/java/accord/impl/DefaultRemoteListeners.java b/accord-core/src/main/java/accord/impl/DefaultRemoteListeners.java
index e545ea3b..e8a9dc03 100644
--- a/accord-core/src/main/java/accord/impl/DefaultRemoteListeners.java
+++ b/accord-core/src/main/java/accord/impl/DefaultRemoteListeners.java
@@ -35,7 +35,7 @@ import accord.primitives.Route;
import accord.primitives.TxnId;
import accord.utils.Invariants;
-// TODO (required): evict to disk
+// TODO (expected): evict to disk
public class DefaultRemoteListeners implements RemoteListeners
{
public interface NotifySink
diff --git a/accord-core/src/main/java/accord/impl/DefaultRequestTimeouts.java b/accord-core/src/main/java/accord/impl/DefaultTimeouts.java
similarity index 79%
rename from accord-core/src/main/java/accord/impl/DefaultRequestTimeouts.java
rename to accord-core/src/main/java/accord/impl/DefaultTimeouts.java
index 10ff3bb4..13120e3c 100644
--- a/accord-core/src/main/java/accord/impl/DefaultRequestTimeouts.java
+++ b/accord-core/src/main/java/accord/impl/DefaultTimeouts.java
@@ -18,12 +18,12 @@
package accord.impl;
-import accord.local.Node;
+import accord.local.TimeService;
-public class DefaultRequestTimeouts extends AbstractRequestTimeouts<AbstractRequestTimeouts.Stripe>
+public class DefaultTimeouts extends AbstractTimeouts<AbstractTimeouts.Stripe>
{
- public DefaultRequestTimeouts(Node node)
+ public DefaultTimeouts(TimeService time)
{
- super(node, Stripe[]::new, Stripe::new);
+ super(time, Stripe[]::new, Stripe::new);
}
}
diff --git a/accord-core/src/main/java/accord/impl/DurabilityScheduling.java b/accord-core/src/main/java/accord/impl/DurabilityScheduling.java
index 10934617..41cd96f2 100644
--- a/accord-core/src/main/java/accord/impl/DurabilityScheduling.java
+++ b/accord-core/src/main/java/accord/impl/DurabilityScheduling.java
@@ -149,11 +149,12 @@ public class DurabilityScheduling implements ConfigurationService.Listener
int nodeOffset;
int index;
- int numberOfSplits;
+ int numberOfSplits, maxNumberOfSplits;
Scheduler.Scheduled scheduled;
long rangeStartedAtMicros, cycleStartedAtMicros;
long retryDelayMicros = defaultRetryDelayMicros;
boolean defunct;
+ SyncPoint<Range> lastApplied, lastAgreed;
private ShardScheduler()
{
@@ -193,9 +194,14 @@ public class DurabilityScheduling implements ConfigurationService.Listener
if (nowMicros > scheduleAt + (shardCycleTimeMicros / numberOfSplits))
scheduleAt += shardCycleTimeMicros;
- if (numberOfSplits < targetShardSplits)
- numberOfSplits = targetShardSplits;
+ maxNumberOfSplits = Math.max(1, Math.min(DurabilityScheduling.this.maxNumberOfSplits, node.commandStores().shardDistributor().numberOfSplitsPossible(shard.range)));
+ int target = Math.min(targetShardSplits, maxNumberOfSplits);
+ if (numberOfSplits < target)
+ numberOfSplits = target;
+ if (numberOfSplits > maxNumberOfSplits)
+ numberOfSplits = maxNumberOfSplits;
+ index = 0;
cycleStartedAtMicros = scheduleAt;
scheduleAt(nowMicros, scheduleAt);
}
@@ -245,10 +251,23 @@ public class DurabilityScheduling implements ConfigurationService.Listener
Invariants.checkState(index < numberOfSplits);
int i = index;
Range selectRange = null;
- while (selectRange == null)
- selectRange = distributor.splitRange(shard.range, index, ++i, numberOfSplits);
- range = selectRange;
+ while (selectRange == null && ++i <= numberOfSplits)
+ selectRange = distributor.splitRange(shard.range, index, i, numberOfSplits);
+ if (selectRange == null)
+ {
+ if (index == 0)
+ {
+ logger.warn("Range {} appears to be impossible to split. Using full range.", shard.range);
+ selectRange = shard.range;
+ }
+ else
+ {
+ restart();
+ return;
+ }
+ }
nextIndex = i;
+ range = selectRange;
}
Runnable schedule = () -> {
@@ -298,7 +317,8 @@ public class DurabilityScheduling implements ConfigurationService.Listener
else
{ // TODO (required): decouple CoordinateShardDurable concurrency from CoordinateSyncPoint (i.e., permit at least one CoordinateSyncPoint to queue up while we're coordinating durability)
coordinateShardDurableAfterExclusiveSyncPoint(node, success, nextIndex);
- logger.trace("{}: Successfully coordinated ExclusiveSyncPoint for local shard durability of {}", syncId, ranges);
+ logger.debug("{}: Successfully coordinated ExclusiveSyncPoint for local shard durability of {}", syncId, ranges);
+ lastAgreed = success;
}
});
}), txnIdLagMicros, MICROSECONDS);
@@ -326,7 +346,6 @@ public class DurabilityScheduling implements ConfigurationService.Listener
index = nextIndex;
if (index >= numberOfSplits)
{
- index = 0;
long nowMicros = node.elapsed(MICROSECONDS);
long timeTakenSeconds = MICROSECONDS.toSeconds(nowMicros - cycleStartedAtMicros);
long targetTimeSeconds = MICROSECONDS.toSeconds(shardCycleTimeMicros);
@@ -346,11 +365,12 @@ public class DurabilityScheduling implements ConfigurationService.Listener
}
else if (logger.isTraceEnabled())
{
- logger.trace("Successfully coordinated shard durability for range {} in {}s", shard.range, MICROSECONDS.toSeconds(nowMicros - rangeStartedAtMicros));
+ logger.debug("Successfully coordinated shard durability for range {} in {}s", shard.range, MICROSECONDS.toSeconds(nowMicros - rangeStartedAtMicros));
}
schedule();
}
}
+ lastApplied = success;
}
catch (Throwable t)
{
diff --git a/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java b/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
index 7e2d507f..69e547c5 100644
--- a/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
+++ b/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
@@ -92,7 +92,7 @@ import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncChains;
import accord.utils.async.Cancellable;
-import static accord.local.KeyHistory.COMMANDS;
+import static accord.local.KeyHistory.ASYNC;
import static accord.local.SafeCommandStore.TestDep.ANY_DEPS;
import static accord.local.SafeCommandStore.TestDep.WITH;
import static accord.local.SafeCommandStore.TestStartedAt.STARTED_BEFORE;
@@ -464,8 +464,10 @@ public abstract class InMemoryCommandStore extends CommandStore
{
case NONE:
continue;
- case COMMANDS:
- case RECOVERY:
+ case INCR:
+ case SYNC:
+ case ASYNC:
+ case RECOVER:
commandsForKey.put(key, commandsForKey((RoutingKey) key).createSafeReference());
break;
case TIMESTAMPS:
@@ -683,34 +685,48 @@ public abstract class InMemoryCommandStore extends CommandStore
this.commandsForKey = commandsForKey;
this.timestampsForKey = timestampsForKey;
this.ranges = ranges;
+ for (InMemorySafeCommand cmd : commands.values())
+ {
+ if (cmd.isUnset()) cmd.uninitialised();
+ }
+ for (InMemorySafeTimestampsForKey tfk : timestampsForKey.values())
+ {
+ if (tfk.isUnset()) tfk.initialize();
+ }
+ for (InMemorySafeCommandsForKey cfk : commandsForKey.values())
+ {
+ if (cfk.isUnset()) cfk.initialize();
+ }
}
@Override
- protected InMemorySafeCommand getCommandInternal(TxnId txnId)
+ protected InMemorySafeCommand getCommandUnsafe(TxnId txnId)
{
return commands.get(txnId);
}
@Override
- protected void addCommandInternal(InMemorySafeCommand command)
+ protected void addCommandUnsafe(InMemorySafeCommand command)
{
+ if (command.isUnset()) command.uninitialised();
commands.put(command.txnId(), command);
}
@Override
- protected InMemorySafeTimestampsForKey getTimestampsForKeyInternal(RoutingKey key)
+ protected InMemorySafeTimestampsForKey getTimestampsForKeyUnsafe(RoutingKey key)
{
return timestampsForKey.get(key);
}
@Override
- protected void addTimestampsForKeyInternal(InMemorySafeTimestampsForKey tfk)
+ protected void addTimestampsForKeyUnsafe(InMemorySafeTimestampsForKey tfk)
{
+ if (tfk.isUnset()) tfk.initialize();
timestampsForKey.put(tfk.key(), tfk);
}
@Override
- protected InMemorySafeTimestampsForKey getTimestampsForKeyIfLoaded(RoutingKey key)
+ protected InMemorySafeTimestampsForKey getTimestampsForKeyIfUnsafe(RoutingKey key)
{
if (!commandStore.canExposeUnloaded())
return null;
@@ -719,7 +735,7 @@ public abstract class InMemoryCommandStore extends CommandStore
}
@Override
- protected InMemorySafeCommand getIfLoaded(TxnId txnId)
+ protected InMemorySafeCommand getIfLoadedUnsafe(TxnId txnId)
{
if (!commandStore.canExposeUnloaded())
return null;
@@ -728,19 +744,20 @@ public abstract class InMemoryCommandStore extends CommandStore
}
@Override
- protected InMemorySafeCommandsForKey getCommandsForKeyInternal(RoutingKey key)
+ protected InMemorySafeCommandsForKey getCommandsForKeyUnsafe(RoutingKey key)
{
return commandsForKey.get(key);
}
@Override
- protected void addCommandsForKeyInternal(InMemorySafeCommandsForKey cfk)
+ protected void addCommandsForKeyUnsafe(InMemorySafeCommandsForKey cfk)
{
+ if (cfk.isUnset()) cfk.initialize();
commandsForKey.put(cfk.key(), cfk);
}
@Override
- protected InMemorySafeCommandsForKey getCommandsForKeyIfLoaded(RoutingKey key)
+ protected InMemorySafeCommandsForKey getCommandsForKeyIfUnsafe(RoutingKey key)
{
if (!commandStore.canExposeUnloaded())
return null;
@@ -996,7 +1013,6 @@ public abstract class InMemoryCommandStore extends CommandStore
return accumulate;
}
- @Override
public void postExecute()
{
commands.values().forEach(c -> {
@@ -1197,10 +1213,10 @@ public abstract class InMemoryCommandStore extends CommandStore
}
@Override
- public InMemorySafeCommand getInternalIfLoadedAndInitialised(TxnId txnId)
+ public InMemorySafeCommand ifLoadedAndInitialisedAndNotErasedInternal(TxnId txnId)
{
assertThread();
- return super.getInternalIfLoadedAndInitialised(txnId);
+ return super.ifLoadedAndInitialisedAndNotErasedInternal(txnId);
}
@Override
@@ -1372,7 +1388,7 @@ public abstract class InMemoryCommandStore extends CommandStore
TxnId txnId = command.txnId();
executeInContext(InMemoryCommandStore.this,
- context(command, COMMANDS),
+ context(command, ASYNC),
safeStore -> {
Command local = command;
if (local.status() != Truncated && local.status() != Invalidated)
diff --git a/accord-core/src/main/java/accord/impl/RequestCallbacks.java b/accord-core/src/main/java/accord/impl/RequestCallbacks.java
index 37e0f5fc..7f35d637 100644
--- a/accord-core/src/main/java/accord/impl/RequestCallbacks.java
+++ b/accord-core/src/main/java/accord/impl/RequestCallbacks.java
@@ -33,7 +33,8 @@ import org.agrona.collections.Long2ObjectHashMap;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
-public class RequestCallbacks extends AbstractRequestTimeouts<RequestCallbacks.CallbackStripe>
+// TODO (expected): tie shards to an AccordExecutor
+public class RequestCallbacks extends AbstractTimeouts<RequestCallbacks.CallbackStripe>
{
private static final Logger logger = LoggerFactory.getLogger(RequestCallbacks.class);
diff --git a/accord-core/src/main/java/accord/impl/TimestampsForKeys.java b/accord-core/src/main/java/accord/impl/TimestampsForKeys.java
index 987400b0..d54080a6 100644
--- a/accord-core/src/main/java/accord/impl/TimestampsForKeys.java
+++ b/accord-core/src/main/java/accord/impl/TimestampsForKeys.java
@@ -81,7 +81,7 @@ public class TimestampsForKeys
}
@VisibleForImplementation
- public static <D> TimestampsForKey updateLastExecutionTimestamps(AbstractSafeCommandStore<?,?,?> safeStore, RoutingKey key, TxnId txnId, Timestamp executeAt, boolean isForWriteTxn)
+ public static <D> TimestampsForKey updateLastExecutionTimestamps(SafeCommandStore safeStore, RoutingKey key, TxnId txnId, Timestamp executeAt, boolean isForWriteTxn)
{
return updateLastExecutionTimestamps(safeStore, safeStore.timestampsForKey(key), txnId, executeAt, isForWriteTxn);
}
diff --git a/accord-core/src/main/java/accord/impl/progresslog/CallbackInvoker.java b/accord-core/src/main/java/accord/impl/progresslog/CallbackInvoker.java
index 0170cc11..a5f18aa4 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/CallbackInvoker.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/CallbackInvoker.java
@@ -22,7 +22,6 @@ import java.util.function.BiConsumer;
import accord.local.SafeCommand;
import accord.primitives.TxnId;
-import accord.utils.Invariants;
import static accord.impl.progresslog.TxnStateKind.Home;
import static accord.impl.progresslog.TxnStateKind.Waiting;
@@ -30,17 +29,17 @@ import static accord.local.PreLoadContext.contextFor;
class CallbackInvoker<P, V> implements BiConsumer<V, Throwable>
{
- static <P, V> BiConsumer<V, Throwable> invokeWaitingCallback(DefaultProgressLog instance, TxnId txnId, P param, Callback<P, V> callback)
+ static <P, V> CallbackInvoker<P, V> invokeWaitingCallback(DefaultProgressLog instance, TxnId txnId, P param, Callback<P, V> callback)
{
return invokeCallback(Waiting, instance, txnId, param, callback);
}
- static <P, V> BiConsumer<V, Throwable> invokeHomeCallback(DefaultProgressLog owner, TxnId txnId, P param, Callback<P, V> callback)
+ static <P, V> CallbackInvoker<P, V> invokeHomeCallback(DefaultProgressLog owner, TxnId txnId, P param, Callback<P, V> callback)
{
return invokeCallback(Home, owner, txnId, param, callback);
}
- static <P, V> BiConsumer<V, Throwable> invokeCallback(TxnStateKind kind, DefaultProgressLog owner, TxnId txnId, P param, Callback<P, V> callback)
+ static <P, V> CallbackInvoker<P, V> invokeCallback(TxnStateKind kind, DefaultProgressLog owner, TxnId txnId, P param, Callback<P, V> callback)
{
CallbackInvoker<P, V> invoker = new CallbackInvoker<>(owner, kind, owner.nextInvokerId(), txnId, param, callback);
owner.registerActive(kind, txnId, invoker);
@@ -79,8 +78,10 @@ class CallbackInvoker<P, V> implements BiConsumer<V, Throwable>
if (!owner.deregisterActive(kind(), this))
return;
- Invariants.checkState(safeCommand != null);
+ if (safeCommand == null)
+ return;
callback.callback(safeStore, safeCommand, owner, txnId, param, success, fail);
+ owner.undebugActive(this);
}).begin(owner.commandStore.agent());
}
diff --git a/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java b/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java
index 06b65370..67df93d3 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java
@@ -47,6 +47,7 @@ import accord.utils.Invariants;
import accord.utils.LogGroupTimers;
import accord.utils.btree.BTree;
import accord.utils.btree.BTreeRemoval;
+import org.agrona.collections.Long2ObjectHashMap;
import org.agrona.collections.Object2ObjectHashMap;
import org.agrona.collections.ObjectHashSet;
@@ -89,7 +90,8 @@ public class DefaultProgressLog implements ProgressLog, Runnable
private final ObjectHashSet<Object> activeWaiting = new ObjectHashSet<>();
private final ObjectHashSet<Object> activeHome = new ObjectHashSet<>();
- private final Map<TxnId, StackTraceElement[]> deleted = Invariants.debug() ? new Object2ObjectHashMap<>() : null;
+ private final Long2ObjectHashMap<Object> debugActive = Invariants.debug() ? new Long2ObjectHashMap<>() : null;
+ private final Map<TxnId, StackTraceElement[]> debugDeleted = Invariants.debug() ? new Object2ObjectHashMap<>() : null;
private static final Object[] EMPTY_RUN_BUFFER = new Object[0];
@@ -133,7 +135,7 @@ public class DefaultProgressLog implements ProgressLog, Runnable
TxnState result = BTree.<TxnId, TxnState>find(stateMap, (id, state) -> id.compareTo(state.txnId), txnId);
if (result == null)
{
- Invariants.checkState(deleted == null || !deleted.containsKey(txnId));
+ Invariants.checkState(debugDeleted == null || !debugDeleted.containsKey(txnId));
node.agent().metricsEventsListener().onProgressLogSizeChange(txnId, 1);
stateMap = BTree.update(stateMap, BTree.singleton(result = new TxnState(txnId)), TxnState::compareTo);
}
@@ -142,7 +144,7 @@ public class DefaultProgressLog implements ProgressLog, Runnable
private TxnState insert(TxnId txnId)
{
- Invariants.checkState(deleted == null || !deleted.containsKey(txnId));
+ Invariants.checkState(debugDeleted == null || !debugDeleted.containsKey(txnId));
node.agent().metricsEventsListener().onProgressLogSizeChange(txnId, 1);
TxnState result = new TxnState(txnId);
stateMap = BTree.update(stateMap, BTree.singleton(result), TxnState::compareTo);
@@ -192,7 +194,7 @@ public class DefaultProgressLog implements ProgressLog, Runnable
if (after.durability().isDurableOrInvalidated())
{
- state.setHomeDoneAnyMaybeRemove(this);
+ state.setHomeDoneAndMaybeRemove(this);
state = maybeFetch(safeStore, txnId, after, state);
}
else
@@ -208,7 +210,7 @@ public class DefaultProgressLog implements ProgressLog, Runnable
{
state = get(txnId);
if (state != null)
- state.setHomeDoneAnyMaybeRemove(this);
+ state.setHomeDoneAndMaybeRemove(this);
state = maybeFetch(safeStore, txnId, after, state);
}
@@ -279,8 +281,8 @@ public class DefaultProgressLog implements ProgressLog, Runnable
activeWaiting.clear();
activeHome.clear();
- if (deleted != null)
- deleted.clear();
+ if (debugDeleted != null)
+ debugDeleted.clear();
runBuffer = EMPTY_RUN_BUFFER;
runBufferCount = 0;
@@ -300,8 +302,8 @@ public class DefaultProgressLog implements ProgressLog, Runnable
if (stateMap != newStateMap)
node.agent().metricsEventsListener().onProgressLogSizeChange(txnId, -1);
stateMap = newStateMap;
- if (deleted != null)
- deleted.put(txnId, Thread.currentThread().getStackTrace());
+ if (debugDeleted != null)
+ debugDeleted.put(txnId, Thread.currentThread().getStackTrace());
}
@Override
@@ -447,13 +449,14 @@ public class DefaultProgressLog implements ProgressLog, Runnable
@Override
public void accept(SafeCommandStore safeStore)
{
- // we have to read safeCommand first as it may become truncated on load, which may clear the progress log and invalidate us
- SafeCommand safeCommand = safeStore.ifInitialised(run.txnId);
- Invariants.checkState(safeCommand != null);
-
if (!deregisterActive(runKind, this))
return; // we've been cancelled
+ // we have to read safeCommand first as it may become truncated on load, which may clear the progress log and invalidate us
+ SafeCommand safeCommand = safeStore.ifInitialised(run.txnId);
+ if (safeCommand == null)
+ return;
+
Invariants.checkState(get(run.txnId) == run, "Transaction state for %s does not match expected one %s", run.txnId, run);
Invariants.checkState(run.scheduledTimer() != runKind, "We are actively executing %s, but we are also scheduled to run this same TxnState later. This should not happen.", runKind);
Invariants.checkState(run.pendingTimer() != runKind, "We are actively executing %s, but we also have a pending scheduled task to run this same TxnState later. This should not happen.", runKind);
@@ -512,6 +515,18 @@ public class DefaultProgressLog implements ProgressLog, Runnable
return active(kind).contains(txnId);
}
+ void debugActive(Object debug, CallbackInvoker<?, ?> invoker)
+ {
+ if (debugActive != null)
+ debugActive.put(invoker.id, debug);
+ }
+
+ void undebugActive(CallbackInvoker<?, ?> invoker)
+ {
+ if (debugActive != null)
+ debugActive.remove(invoker.id);
+ }
+
boolean deregisterActive(TxnStateKind kind, Object object)
{
return active(kind).remove(object);
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 c7186450..d8378f53 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/HomeState.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/HomeState.java
@@ -145,7 +145,9 @@ abstract class HomeState extends WaitingState
Invariants.checkState(!command.durability().isDurableOrInvalidated(), "Command is durable or invalidated, but we have not cleared the ProgressLog");
ProgressToken maxProgressToken = instance.savedProgressToken(txnId).merge(command);
- MaybeRecover.maybeRecover(instance.node(), txnId, command.route(), maxProgressToken, invokeHomeCallback(instance, txnId, maxProgressToken, HomeState::recoverCallback));
+
+ CallbackInvoker<ProgressToken, Outcome> invoker = invokeHomeCallback(instance, txnId, maxProgressToken, HomeState::recoverCallback);
+ instance.debugActive(MaybeRecover.maybeRecover(instance.node(), txnId, command.route(), maxProgressToken, invoker), invoker);
set(safeStore, instance, ReadyToExecute, Querying);
}
@@ -178,7 +180,7 @@ abstract class HomeState extends WaitingState
if (token.durability.isDurableOrInvalidated())
{
- state.setHomeDoneAnyMaybeRemove(instance);
+ state.setHomeDoneAndMaybeRemove(instance);
}
else
{
@@ -197,7 +199,7 @@ abstract class HomeState extends WaitingState
instance.clearActive(Home, txnId);
}
- void setHomeDoneAnyMaybeRemove(DefaultProgressLog instance)
+ void setHomeDoneAndMaybeRemove(DefaultProgressLog instance)
{
setHomeDone(instance);
maybeRemove(instance);
diff --git a/accord-core/src/main/java/accord/local/Cleanup.java b/accord-core/src/main/java/accord/local/Cleanup.java
index 0c0cecf7..195f9fd6 100644
--- a/accord-core/src/main/java/accord/local/Cleanup.java
+++ b/accord-core/src/main/java/accord/local/Cleanup.java
@@ -21,6 +21,7 @@ package accord.local;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
+import accord.api.Agent;
import accord.api.VisibleForImplementation;
import accord.primitives.FullRoute;
import accord.primitives.Route;
@@ -85,29 +86,29 @@ public enum Cleanup
public static Cleanup shouldCleanup(SafeCommandStore safeStore, Command command, @Nonnull StoreParticipants participants)
{
- return shouldCleanup(command.txnId(), command.saveStatus(), command.durability(), participants,
+ return shouldCleanup(safeStore.agent(), command.txnId(), command.saveStatus(), command.durability(), participants,
safeStore.redundantBefore(), safeStore.durableBefore());
}
- public static Cleanup shouldCleanup(Command command, RedundantBefore redundantBefore, DurableBefore durableBefore)
+ public static Cleanup shouldCleanup(Agent agent, Command command, RedundantBefore redundantBefore, DurableBefore durableBefore)
{
- return shouldCleanup(command.txnId(), command.saveStatus(), command.durability(), command.participants(),
+ return shouldCleanup(agent, command.txnId(), command.saveStatus(), command.durability(), command.participants(),
redundantBefore, durableBefore);
}
- public static Cleanup shouldCleanup(TxnId txnId, SaveStatus status, Durability durability, StoreParticipants participants, RedundantBefore redundantBefore, DurableBefore durableBefore)
+ public static Cleanup shouldCleanup(Agent agent, TxnId txnId, SaveStatus status, Durability durability, StoreParticipants participants, RedundantBefore redundantBefore, DurableBefore durableBefore)
{
- return shouldCleanupInternal(txnId, status, durability, participants, redundantBefore, durableBefore).filter(status);
+ return shouldCleanupInternal(agent, txnId, status, durability, participants, redundantBefore, durableBefore).filter(status);
}
// TODO (required): simulate compaction of log records in burn test
@VisibleForImplementation
- public static Cleanup shouldCleanupPartial(TxnId txnId, SaveStatus status, Durability durability, StoreParticipants participants, RedundantBefore redundantBefore, DurableBefore durableBefore)
+ public static Cleanup shouldCleanupPartial(Agent agent, TxnId txnId, SaveStatus status, Durability durability, StoreParticipants participants, RedundantBefore redundantBefore, DurableBefore durableBefore)
{
- return shouldCleanupPartialInternal(txnId, status, durability, participants, redundantBefore, durableBefore).filter(status);
+ return shouldCleanupPartialInternal(agent, txnId, status, durability, participants, redundantBefore, durableBefore).filter(status);
}
- private static Cleanup shouldCleanupInternal(TxnId txnId, SaveStatus saveStatus, Durability durability, StoreParticipants participants, RedundantBefore redundantBefore, DurableBefore durableBefore)
+ private static Cleanup shouldCleanupInternal(Agent agent, TxnId txnId, SaveStatus saveStatus, Durability durability, StoreParticipants participants, RedundantBefore redundantBefore, DurableBefore durableBefore)
{
if (txnId.kind() == EphemeralRead)
return NO; // TODO (required): clean-up based on timeout
@@ -142,10 +143,10 @@ public enum Cleanup
return Cleanup.NO;
}
- return cleanupWithFullRoute(false, participants, txnId, saveStatus, durability, redundantBefore, durableBefore);
+ return cleanupWithFullRoute(agent, false, participants, txnId, saveStatus, durability, redundantBefore, durableBefore);
}
- private static Cleanup shouldCleanupPartialInternal(TxnId txnId, SaveStatus status, @Nullable Durability durability, @Nullable StoreParticipants participants, RedundantBefore redundantBefore, DurableBefore durableBefore)
+ private static Cleanup shouldCleanupPartialInternal(Agent agent, TxnId txnId, SaveStatus status, @Nullable Durability durability, @Nullable StoreParticipants participants, RedundantBefore redundantBefore, DurableBefore durableBefore)
{
if (txnId.kind() == EphemeralRead)
return NO; // TODO (required): clean-up based on timeout
@@ -177,10 +178,10 @@ public enum Cleanup
}
}
- return cleanupWithFullRoute(true, participants, txnId, status, durability, redundantBefore, durableBefore);
+ return cleanupWithFullRoute(agent, true, participants, txnId, status, durability, redundantBefore, durableBefore);
}
- private static Cleanup cleanupWithFullRoute(boolean isPartial, StoreParticipants participants, TxnId txnId, SaveStatus saveStatus, Durability durability, RedundantBefore redundantBefore, DurableBefore durableBefore)
+ private static Cleanup cleanupWithFullRoute(Agent agent, boolean isPartial, StoreParticipants participants, TxnId txnId, SaveStatus saveStatus, Durability durability, RedundantBefore redundantBefore, DurableBefore durableBefore)
{
// We first check if the command is redundant locally, i.e. whether it has been applied to all non-faulty replicas of the local shard
// If not, we don't want to truncate its state else we may make catching up for these other replicas much harder
@@ -200,7 +201,10 @@ public enum Cleanup
return Cleanup.NO;
case SHARD_REDUNDANT:
if (!isPartial && saveStatus.hasBeen(PreCommitted) && !saveStatus.hasBeen(Applied) && redundantBefore.preBootstrapOrStale(txnId, participants.owns) != FULLY)
- illegalState("Loading redundant command that has been PreCommitted but not Applied.");
+ {
+ agent.onViolation(String.format("Loading SHARD_REDUNDANT command %s with status %s (that should have been Applied). Expected to be witnessed and executed by %s.", txnId, saveStatus, redundantBefore.max(participants.route, e -> e.shardAppliedOrInvalidatedBefore)));
+ return TRUNCATE;
+ }
case WAS_OWNED:
}
diff --git a/accord-core/src/main/java/accord/local/Command.java b/accord-core/src/main/java/accord/local/Command.java
index 379e2999..cf64e6cd 100644
--- a/accord-core/src/main/java/accord/local/Command.java
+++ b/accord-core/src/main/java/accord/local/Command.java
@@ -1260,6 +1260,21 @@ public abstract class Command implements CommonAttributes
return waitingOn.get(txnIdCount() + keyIndex);
}
+ public RoutingKeys waitingOnKeys()
+ {
+ int offset = txnIdCount();
+ int count = waitingOn.getSetBitCount(offset, waitingOn.size());
+ if (count == 0)
+ return RoutingKeys.EMPTY;
+ if (count == keys.size())
+ return keys;
+ RoutingKey[] selected = new RoutingKey[count];
+ int c = 0;
+ for (int i = waitingOn.nextSetBit(offset, -1) ; i >= 0 ; i = waitingOn.nextSetBit(i + 1, -1))
+ selected[c++] = keys.get(i - offset);
+ return RoutingKeys.ofSortedUnique(selected);
+ }
+
public boolean isWaitingOnCommand()
{
return waitingOn.firstSetBit() < txnIdCount();
diff --git a/accord-core/src/main/java/accord/local/CommandStore.java b/accord-core/src/main/java/accord/local/CommandStore.java
index 76e83b04..d6b42421 100644
--- a/accord-core/src/main/java/accord/local/CommandStore.java
+++ b/accord-core/src/main/java/accord/local/CommandStore.java
@@ -103,13 +103,13 @@ public abstract class CommandStore implements AgentExecutor
// TODO (desired): can better encapsulate by accepting only the newRangesForEpoch and deriving the add/remove ranges
public void add(long epoch, RangesForEpoch newRangesForEpoch, Ranges addRanges)
{
- RedundantBefore addRedundantBefore = RedundantBefore.create(addRanges, epoch, Long.MAX_VALUE, TxnId.NONE, TxnId.NONE, TxnId.NONE, TxnId.minForEpoch(epoch));
+ RedundantBefore addRedundantBefore = RedundantBefore.create(addRanges, epoch, Long.MAX_VALUE, TxnId.NONE, TxnId.NONE, TxnId.NONE, TxnId.NONE, TxnId.minForEpoch(epoch));
update(newRangesForEpoch, addRedundantBefore);
}
public void remove(long epoch, RangesForEpoch newRangesForEpoch, Ranges removeRanges)
{
- RedundantBefore addRedundantBefore = RedundantBefore.create(removeRanges, Long.MIN_VALUE, epoch, TxnId.NONE, TxnId.NONE, TxnId.NONE, TxnId.NONE);
+ RedundantBefore addRedundantBefore = RedundantBefore.create(removeRanges, Long.MIN_VALUE, epoch, TxnId.NONE, TxnId.NONE, TxnId.NONE, TxnId.NONE, TxnId.NONE);
update(newRangesForEpoch, addRedundantBefore);
}
@@ -225,12 +225,25 @@ public abstract class CommandStore implements AgentExecutor
safeStore.setRangesForEpoch(update.newRangesForEpoch);
}
+ @VisibleForTesting
+ public void unsafeUpdateRangesForEpoch()
+ {
+ EpochUpdate update = epochUpdateHolder.getAndSet(null);
+ if (update == null)
+ return;
+
+ if (update.addRedundantBefore.size() > 0)
+ unsafeUpsertRedundantBefore(update.addRedundantBefore);
+ if (update.newRangesForEpoch != null)
+ unsafeSetRangesForEpoch(update.newRangesForEpoch);
+ }
+
public RangesForEpoch unsafeRangesForEpoch()
{
return rangesForEpoch;
}
- protected void unsafeSetRangesForEpoch(RangesForEpoch newRangesForEpoch)
+ public void unsafeSetRangesForEpoch(RangesForEpoch newRangesForEpoch)
{
rangesForEpoch = nonNull(newRangesForEpoch);
}
@@ -342,7 +355,7 @@ public abstract class CommandStore implements AgentExecutor
{
// TODO (desired): narrow ranges to those that are owned
Invariants.checkArgument(txnId.is(ExclusiveSyncPoint));
- RedundantBefore newRedundantBefore = RedundantBefore.merge(redundantBefore, RedundantBefore.create(ranges, txnId, TxnId.NONE, TxnId.NONE, TxnId.NONE));
+ RedundantBefore newRedundantBefore = RedundantBefore.merge(redundantBefore, RedundantBefore.create(ranges, txnId, txnId, TxnId.NONE, TxnId.NONE, TxnId.NONE));
unsafeSetRedundantBefore(newRedundantBefore);
updatedRedundantBefore(safeStore, txnId, ranges);
}
@@ -466,7 +479,7 @@ public abstract class CommandStore implements AgentExecutor
}).beginAsResult();
return new EpochReady(epoch, metadata.<Void>map(ignore -> null).beginAsResult(),
- metadata.flatMap(e -> e.coordination).beginAsResult(),
+ metadata.flatMap(e -> e.fastPath).beginAsResult(),
metadata.flatMap(e -> e.data).beginAsResult(),
metadata.flatMap(e -> e.reads).beginAsResult());
};
@@ -478,12 +491,15 @@ public abstract class CommandStore implements AgentExecutor
* So, the outer future's success is sufficient for the topology to be acknowledged, and the inner future for the
* bootstrap to be complete.
*/
- protected Supplier<EpochReady> sync(Node node, Ranges ranges, long epoch, boolean isLoad)
+ protected Supplier<EpochReady> sync(Node node, Ranges ranges, long epoch)
{
return () -> {
+ if (redundantBefore.min(ranges, RedundantBefore.Entry::locallyWitnessedBefore).epoch() >= epoch)
+ return new EpochReady(epoch, DONE, DONE, DONE, DONE);
+
AsyncResults.SettableResult<Void> whenDone = new AsyncResults.SettableResult<>();
waitingOnSync.put(epoch, new WaitingOnSync(whenDone, ranges));
- return new EpochReady(epoch, DONE, whenDone, whenDone, whenDone);
+ return new EpochReady(epoch, DONE, whenDone, DONE, DONE);
};
}
@@ -512,7 +528,7 @@ public abstract class CommandStore implements AgentExecutor
final void markBootstrapping(SafeCommandStore safeStore, TxnId globalSyncId, Ranges ranges)
{
safeStore.setBootstrapBeganAt(bootstrap(globalSyncId, ranges, bootstrapBeganAt));
- RedundantBefore addRedundantBefore = RedundantBefore.create(ranges, Long.MIN_VALUE, Long.MAX_VALUE, TxnId.NONE, TxnId.NONE, TxnId.NONE, globalSyncId);
+ RedundantBefore addRedundantBefore = RedundantBefore.create(ranges, Long.MIN_VALUE, Long.MAX_VALUE, TxnId.NONE, TxnId.NONE, TxnId.NONE, TxnId.NONE, globalSyncId);
safeStore.upsertRedundantBefore(addRedundantBefore);
updatedRedundantBefore(safeStore, globalSyncId, ranges);
}
@@ -522,7 +538,7 @@ public abstract class CommandStore implements AgentExecutor
{
final Ranges slicedRanges = durableRanges.slice(safeStore.ranges().allUntil(globalSyncId.epoch()), Minimal);
TxnId locallyRedundantBefore = safeStore.redundantBefore().min(slicedRanges, e -> e.locallyAppliedOrInvalidatedBefore);
- RedundantBefore addShardRedundant = RedundantBefore.create(slicedRanges, Long.MIN_VALUE, Long.MAX_VALUE, TxnId.NONE, globalSyncId, TxnId.NONE, TxnId.NONE);
+ RedundantBefore addShardRedundant = RedundantBefore.create(slicedRanges, Long.MIN_VALUE, Long.MAX_VALUE, globalSyncId, TxnId.NONE, globalSyncId, TxnId.NONE, TxnId.NONE);
safeStore.upsertRedundantBefore(addShardRedundant);
updatedRedundantBefore(safeStore, globalSyncId, slicedRanges);
safeStore = safeStore; // make unusable in lambda
@@ -546,7 +562,7 @@ public abstract class CommandStore implements AgentExecutor
}
execute(PreLoadContext.empty(), safeStore0 -> {
- RedundantBefore addGc = RedundantBefore.create(slicedRanges, Long.MIN_VALUE, Long.MAX_VALUE, globalSyncId, globalSyncId, globalSyncId, TxnId.NONE);
+ RedundantBefore addGc = RedundantBefore.create(slicedRanges, Long.MIN_VALUE, Long.MAX_VALUE, globalSyncId, globalSyncId, globalSyncId, globalSyncId, TxnId.NONE);
safeStore0.upsertRedundantBefore(addGc);
}).begin(agent());
});
@@ -556,8 +572,12 @@ public abstract class CommandStore implements AgentExecutor
{
}
- protected void markSynced(TxnId syncId, Ranges ranges)
+ protected void markSynced(SafeCommandStore safeStore, TxnId syncId, Ranges ranges)
{
+ RedundantBefore newRedundantBefore = RedundantBefore.merge(redundantBefore, RedundantBefore.create(ranges, syncId, TxnId.NONE, TxnId.NONE, TxnId.NONE, TxnId.NONE));
+ unsafeSetRedundantBefore(newRedundantBefore);
+ updatedRedundantBefore(safeStore, syncId, ranges);
+
if (waitingOnSync.isEmpty())
return;
@@ -605,7 +625,7 @@ public abstract class CommandStore implements AgentExecutor
}
agent.onStale(staleSince, ranges);
- RedundantBefore addRedundantBefore = RedundantBefore.create(ranges, TxnId.NONE, TxnId.NONE, TxnId.NONE, TxnId.NONE, staleUntilAtLeast);
+ RedundantBefore addRedundantBefore = RedundantBefore.create(ranges, TxnId.NONE, TxnId.NONE, TxnId.NONE, TxnId.NONE, TxnId.NONE, staleUntilAtLeast);
safeStore.upsertRedundantBefore(addRedundantBefore);
// find which ranges need to bootstrap, subtracting those already in progress that cover the id
@@ -626,6 +646,7 @@ public abstract class CommandStore implements AgentExecutor
// TODO (review): Convoluted check to not overwrite existing bootstraps with TxnId.NONE
// If loading from disk didn't finish before this then we might initialize the range at TxnId.NONE?
// Does CommandStores.topology ensure that doesn't happen? Is it fine if it does because it will get superseded?
+
Ranges newBootstrapRanges = ranges;
for (Ranges existing : bootstrapBeganAt.values())
newBootstrapRanges = newBootstrapRanges.without(existing);
@@ -722,4 +743,10 @@ public abstract class CommandStore implements AgentExecutor
return in;
return new SimpleImmutableEntry<>(in.getKey(), without);
}
+
+ @Override
+ public int hashCode()
+ {
+ return id;
+ }
}
diff --git a/accord-core/src/main/java/accord/local/CommandStores.java b/accord-core/src/main/java/accord/local/CommandStores.java
index 3804254c..10fbb098 100644
--- a/accord-core/src/main/java/accord/local/CommandStores.java
+++ b/accord-core/src/main/java/accord/local/CommandStores.java
@@ -60,6 +60,7 @@ import accord.utils.MapReduceConsume;
import accord.utils.RandomSource;
import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncChains;
+import accord.utils.async.Cancellable;
import org.agrona.collections.Hashing;
import org.agrona.collections.Int2ObjectHashMap;
@@ -402,7 +403,7 @@ public abstract class CommandStores
return newLocalTopology.epoch() != 1;
}
- private synchronized TopologyUpdate updateTopology(Node node, Snapshot prev, Topology newTopology, boolean isLoad, boolean startSync)
+ private synchronized TopologyUpdate updateTopology(Node node, Snapshot prev, Topology newTopology, boolean startSync)
{
checkArgument(!newTopology.isSubset(), "Use full topology for CommandStores.updateTopology");
@@ -451,14 +452,13 @@ public abstract class CommandStores
// ranges can be empty when ranges are lost or consolidated across epochs.
if (epoch > 1 && startSync && requiresSync(ranges, prev.global, newTopology))
{
- bootstrapUpdates.add(shard.store.sync(node, ranges, epoch, isLoad));
+ bootstrapUpdates.add(shard.store.sync(node, ranges, epoch));
}
result.add(shard);
}
if (!added.isEmpty())
{
- // TODO (required): shards must rebalance
for (Ranges addRanges : shardDistributor.split(added))
{
EpochUpdateHolder updateHolder = new EpochUpdateHolder();
@@ -480,7 +480,7 @@ public abstract class CommandStores
List<EpochReady> list = bootstrapUpdates.stream().map(Supplier::get).collect(toList());
return new EpochReady(epoch,
AsyncChains.reduce(list.stream().map(b -> b.metadata).collect(toList()), (a, b) -> null).beginAsResult(),
- AsyncChains.reduce(list.stream().map(b -> b.coordination).collect(toList()), (a, b) -> null).beginAsResult(),
+ AsyncChains.reduce(list.stream().map(b -> b.fastPath).collect(toList()), (a, b) -> null).beginAsResult(),
AsyncChains.reduce(list.stream().map(b -> b.data).collect(toList()), (a, b) -> null).beginAsResult(),
AsyncChains.reduce(list.stream().map(b -> b.reads).collect(toList()), (a, b) -> null).beginAsResult()
);
@@ -591,9 +591,9 @@ public abstract class CommandStores
/**
* See {@link #mapReduceConsume(PreLoadContext, Routables, long, long, MapReduceConsume)}
*/
- public <O> void mapReduceConsume(PreLoadContext context, RoutingKey key, long minEpoch, long maxEpoch, MapReduceConsume<? super SafeCommandStore, O> mapReduceConsume)
+ public <O> Cancellable mapReduceConsume(PreLoadContext context, RoutingKey key, long minEpoch, long maxEpoch, MapReduceConsume<? super SafeCommandStore, O> mapReduceConsume)
{
- mapReduceConsume(context, RoutingKeys.of(key), minEpoch, maxEpoch, mapReduceConsume);
+ return mapReduceConsume(context, RoutingKeys.of(key), minEpoch, maxEpoch, mapReduceConsume);
}
/**
@@ -605,16 +605,16 @@ public abstract class CommandStores
*
* Implementations are expected to invoke {@link #mapReduceConsume(PreLoadContext, Routables, long, long, MapReduceConsume)}
*/
- protected <O> void mapReduceConsume(PreLoadContext context, Routables<?> keys, long minEpoch, long maxEpoch, MapReduceConsume<? super SafeCommandStore, O> mapReduceConsume)
+ protected <O> Cancellable mapReduceConsume(PreLoadContext context, Routables<?> keys, long minEpoch, long maxEpoch, MapReduceConsume<? super SafeCommandStore, O> mapReduceConsume)
{
AsyncChain<O> reduced = mapReduce(context, keys, minEpoch, maxEpoch, mapReduceConsume);
- reduced.begin(mapReduceConsume);
+ return reduced.begin(mapReduceConsume);
}
- public <O> void mapReduceConsume(PreLoadContext context, IntStream commandStoreIds, MapReduceConsume<? super SafeCommandStore, O> mapReduceConsume)
+ public <O> Cancellable mapReduceConsume(PreLoadContext context, IntStream commandStoreIds, MapReduceConsume<? super SafeCommandStore, O> mapReduceConsume)
{
AsyncChain<O> reduced = mapReduce(context, commandStoreIds, mapReduceConsume);
- reduced.begin(mapReduceConsume);
+ return reduced.begin(mapReduceConsume);
}
public <O> AsyncChain<O> mapReduce(PreLoadContext context, Routables<?> keys, long minEpoch, long maxEpoch, MapReduce<? super SafeCommandStore, O> mapReduce)
@@ -667,10 +667,10 @@ public abstract class CommandStores
return chain == null ? AsyncChains.success(null) : chain;
}
- public <O> void mapReduceConsume(PreLoadContext context, MapReduceConsume<? super SafeCommandStore, O> mapReduceConsume)
+ public <O> Cancellable mapReduceConsume(PreLoadContext context, MapReduceConsume<? super SafeCommandStore, O> mapReduceConsume)
{
AsyncChain<O> reduced = mapReduce(context, mapReduceConsume);
- reduced.begin(mapReduceConsume);
+ return reduced.begin(mapReduceConsume);
}
protected <O> AsyncChain<O> mapReduce(PreLoadContext context, MapReduce<? super SafeCommandStore, O> mapReduce)
@@ -687,9 +687,9 @@ public abstract class CommandStores
return chain == null ? AsyncChains.success(null) : chain;
}
- public synchronized Supplier<EpochReady> updateTopology(Node node, Topology newTopology, boolean isLoad, boolean startSync)
+ public synchronized Supplier<EpochReady> updateTopology(Node node, Topology newTopology, boolean startSync)
{
- TopologyUpdate update = updateTopology(node, current, newTopology, isLoad, startSync);
+ TopologyUpdate update = updateTopology(node, current, newTopology, startSync);
current = update.snapshot;
return update.bootstrap;
}
diff --git a/accord-core/src/main/java/accord/local/Commands.java b/accord-core/src/main/java/accord/local/Commands.java
index ab0b4235..3974d86d 100644
--- a/accord-core/src/main/java/accord/local/Commands.java
+++ b/accord-core/src/main/java/accord/local/Commands.java
@@ -18,8 +18,6 @@
package accord.local;
-import java.util.Collection;
-import java.util.Collections;
import java.util.function.BiPredicate;
import java.util.function.Consumer;
import javax.annotation.Nonnull;
@@ -81,6 +79,7 @@ import static accord.primitives.Status.PreCommitted;
import static accord.primitives.Status.Stable;
import static accord.primitives.Status.Truncated;
import static accord.primitives.Route.isFullRoute;
+import static accord.primitives.Txn.Kind.EphemeralRead;
import static accord.primitives.Txn.Kind.Read;
import static accord.utils.Invariants.illegalState;
@@ -338,18 +337,13 @@ public class Commands
safeStore.notifyListeners(safeCommand, command);
}
- public static void ephemeralRead(SafeCommandStore safeStore, SafeCommand safeCommand, Route<?> route, TxnId txnId, PartialTxn partialTxn, PartialDeps partialDeps, long executeAtEpoch)
+ public static void ephemeralRead(SafeCommandStore safeStore, SafeCommand safeCommand, Route<?> route, TxnId txnId, PartialTxn partialTxn, PartialDeps partialDeps)
{
// TODO (expected): introduce in-memory only commands
Command command = safeCommand.current();
if (command.hasBeen(Stable))
return;
- // TODO (required): by creating synthetic TxnId in future epochs we may not be evictable
- // but for ephemeral reads we want parallel eviction - or preferably no durability - anyway
- // REVISIT
- txnId = txnId.withEpoch(executeAtEpoch);
-
// BREAKING CHANGE NOTE: if in future we support a CommandStore adopting additional ranges (rather than only shedding them)
// then we need to revisit how we execute transactions that awaitsOnlyDeps, as they may need additional
// information to execute in the eventual execution epoch (that they didn't know they needed when they were made stable)
@@ -361,6 +355,19 @@ public class Commands
maybeExecute(safeStore, safeCommand, false, true);
}
+ public static void eraseEphemeralRead(SafeCommandStore safeStore, TxnId txnId)
+ {
+ SafeCommand safeCommand = safeStore.unsafeGetNoCleanup(txnId);
+ if (safeCommand == null)
+ return;
+
+ Command command = safeCommand.current();
+ if (command.hasBeen(Truncated))
+ return;
+
+ safeCommand.set(erased(command));
+ }
+
public static void markBootstrapComplete(SafeCommandStore safeStore, TxnId localSyncId, Seekables<?, ?> keys)
{
SafeCommand safeCommand = safeStore.get(localSyncId);
@@ -641,7 +648,7 @@ public class Commands
redundantBefore.removeRedundantDependencies(participants, update);
update.forEachWaitingOnId(safeStore, update, waiting, executeAt, (store, upd, w, exec, i) -> {
- SafeCommand dep = store.ifLoadedAndInitialised(upd.txnId(i));
+ SafeCommand dep = store.ifLoadedAndInitialisedAndNotErased(upd.txnId(i));
if (dep == null || !dep.current().hasBeen(PreCommitted))
return;
updateWaitingOn(store, w, exec, upd, dep);
@@ -737,17 +744,24 @@ public class Commands
public static void removeWaitingOnKeyAndMaybeExecute(SafeCommandStore safeStore, SafeCommand safeCommand, RoutingKey key)
{
- if (safeCommand.current().hasBeen(Applied))
+ Command current = safeCommand.current();
+ if (current.saveStatus().compareTo(SaveStatus.Applied) >= 0)
return;
- Command.Committed command = safeCommand.current().asCommitted();
+ if (current.saveStatus().compareTo(SaveStatus.Committed) < 0)
+ { // ephemeral reads can be erased without warning
+ Invariants.checkState(current.txnId().is(EphemeralRead));
+ return;
+ }
- WaitingOn currentWaitingOn = command.waitingOn;
+ Command.Committed committed = safeCommand.current().asCommitted();
+
+ WaitingOn currentWaitingOn = committed.waitingOn;
int keyIndex = currentWaitingOn.keys.indexOf(key);
if (keyIndex < 0 || !currentWaitingOn.isWaitingOnKey(keyIndex))
return;
- WaitingOn.Update waitingOn = new WaitingOn.Update(command);
+ WaitingOn.Update waitingOn = new WaitingOn.Update(committed);
waitingOn.removeWaitingOnKey(keyIndex);
safeCommand.updateWaitingOn(waitingOn);
if (!waitingOn.isWaiting())
@@ -815,7 +829,7 @@ public class Commands
safeCommand.update(safeStore, result);
safeStore.progressLog().clear(safeCommand.txnId());
if (notifyListeners)
- safeStore.notifyListeners(safeCommand, result);
+ safeStore.notifyListeners(safeCommand, command);
return result;
}
@@ -984,7 +998,7 @@ public class Commands
}
}
- depSafe = safeStore.ifLoadedAndInitialised(directlyBlockedOn);
+ depSafe = safeStore.ifLoadedAndInitialisedAndNotErased(directlyBlockedOn);
if (depSafe == null)
{
loadDepId = directlyBlockedOn;
@@ -1094,9 +1108,9 @@ public class Commands
}
@Override
- public Collection<TxnId> additionalTxnIds()
+ public TxnId additionalTxnId()
{
- return loadDepId == null ? Collections.emptyList() : Collections.singletonList(loadDepId);
+ return loadDepId;
}
}
diff --git a/accord-core/src/main/java/accord/local/KeyHistory.java b/accord-core/src/main/java/accord/local/KeyHistory.java
index 40dfe4ef..a3b46dc5 100644
--- a/accord-core/src/main/java/accord/local/KeyHistory.java
+++ b/accord-core/src/main/java/accord/local/KeyHistory.java
@@ -24,13 +24,31 @@ package accord.local;
*/
public enum KeyHistory
{
+ // TODO (required): deprecate
TIMESTAMPS,
- COMMANDS,
- RECOVERY,
- NONE;
- public boolean isNone()
- {
- return this == NONE;
- }
+ /**
+ * This command will likely need to use the provided keys, but can be processed before they are loaded;
+ * any work touching keys not loaded will be submitted as a follow-up INCR or SYNC command.
+ */
+ ASYNC,
+
+ /**
+ * Load and process the requested keys incrementally; the operation will be invoked multiples times
+ * as keys are loaded, until all the keys have been processed. The keys to be processed must be loaded
+ * into memory
+ */
+ INCR,
+
+ /**
+ * Load all keys into memory before processing the command.
+ */
+ SYNC,
+
+ /**
+ * Load recovery information for all keys into memory before processing the command.
+ */
+ RECOVER,
+
+ NONE
}
diff --git a/accord-core/src/main/java/accord/local/Node.java b/accord-core/src/main/java/accord/local/Node.java
index 15347737..a213c112 100644
--- a/accord-core/src/main/java/accord/local/Node.java
+++ b/accord-core/src/main/java/accord/local/Node.java
@@ -48,8 +48,8 @@ import accord.api.LocalConfig;
import accord.api.LocalListeners;
import accord.api.MessageSink;
import accord.api.ProgressLog;
+import accord.api.Timeouts;
import accord.api.RemoteListeners;
-import accord.api.RequestTimeouts;
import accord.api.Result;
import accord.api.RoutingKey;
import accord.api.Scheduler;
@@ -95,6 +95,7 @@ import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncExecutor;
import accord.utils.async.AsyncResult;
import accord.utils.async.AsyncResults;
+import accord.utils.async.Cancellable;
import net.nicoulaj.compilecommand.annotations.Inline;
import static accord.utils.Invariants.illegalState;
@@ -159,7 +160,7 @@ public class Node implements ConfigurationService.Listener, NodeCommandStoreServ
private final ConfigurationService configService;
private final TopologyManager topology;
private final RemoteListeners listeners;
- private final RequestTimeouts timeouts;
+ private final Timeouts timeouts;
private final CommandStores commandStores;
private final CoordinationAdapter.Factory coordinationAdapters;
@@ -183,7 +184,7 @@ public class Node implements ConfigurationService.Listener, NodeCommandStoreServ
public Node(Id id, MessageSink messageSink,
ConfigurationService configService, TimeService time,
Supplier<DataStore> dataSupplier, ShardDistributor shardDistributor, Agent agent, RandomSource random, Scheduler scheduler, TopologySorter.Supplier topologySorter,
- Function<Node, RemoteListeners> remoteListenersFactory, Function<Node, RequestTimeouts> requestTimeoutsFactory, Function<Node, ProgressLog.Factory> progressLogFactory,
+ Function<Node, RemoteListeners> remoteListenersFactory, Function<Node, Timeouts> requestTimeoutsFactory, Function<Node, ProgressLog.Factory> progressLogFactory,
Function<Node, LocalListeners.Factory> localListenersFactory, CommandStores.Factory factory, CoordinationAdapter.Factory coordinationAdapters,
Persister<DurableBefore, DurableBefore> durableBeforePersister,
LocalConfig localConfig)
@@ -240,9 +241,9 @@ public class Node implements ConfigurationService.Listener, NodeCommandStoreServ
@VisibleForTesting
public AsyncResult<Void> unsafeStart()
{
- EpochReady ready = onTopologyUpdateInternal(configService.currentTopology(), false, false);
+ EpochReady ready = onTopologyUpdateInternal(configService.currentTopology(), false);
durabilityScheduling.updateTopology();
- ready.coordination.addCallback(() -> this.topology.onEpochSyncComplete(id, topology.epoch()));
+ ready.fastPath.addCallback(() -> this.topology.onEpochSyncComplete(id, topology.epoch()));
configService.acknowledgeEpoch(ready, false);
return ready.metadata;
}
@@ -312,24 +313,24 @@ public class Node implements ConfigurationService.Listener, NodeCommandStoreServ
return topology().epoch();
}
- private synchronized EpochReady onTopologyUpdateInternal(Topology topology, boolean isLoad, boolean startSync)
+ private synchronized EpochReady onTopologyUpdateInternal(Topology topology, boolean startSync)
{
- Supplier<EpochReady> bootstrap = commandStores.updateTopology(this, topology, isLoad, startSync);
- Supplier<EpochReady> ordering = () -> {
+ Supplier<EpochReady> bootstrap = commandStores.updateTopology(this, topology, startSync);
+ Supplier<EpochReady> orderFastPathReporting = () -> {
if (this.topology.isEmpty()) return bootstrap.get();
- return order(this.topology.epochReady(topology.epoch() - 1), bootstrap.get());
+ return orderFastPathReporting(this.topology.epochReady(topology.epoch() - 1), bootstrap.get());
};
- return this.topology.onTopologyUpdate(topology, ordering);
+ return this.topology.onTopologyUpdate(topology, orderFastPathReporting);
}
- private static EpochReady order(EpochReady previous, EpochReady next)
+ private static EpochReady orderFastPathReporting(EpochReady previous, EpochReady next)
{
if (previous.epoch + 1 != next.epoch)
throw new IllegalArgumentException("Attempted to order epochs but they are not next to each other... previous=" + previous.epoch + ", next=" + next.epoch);
- if (previous.coordination.isDone()) return next;
+ if (previous.fastPath.isDone()) return next;
return new EpochReady(next.epoch,
next.metadata,
- previous.coordination.flatMap(ignore -> next.coordination).beginAsResult(),
+ previous.fastPath.flatMap(ignore -> next.fastPath).beginAsResult(),
next.data,
next.reads);
}
@@ -339,10 +340,10 @@ public class Node implements ConfigurationService.Listener, NodeCommandStoreServ
{
if (topology.epoch() <= this.topology.epoch())
return AsyncResults.success(null);
- EpochReady ready = onTopologyUpdateInternal(topology, isLoad, startSync);
- ready.coordination.addCallback(() -> this.topology.onEpochSyncComplete(id, topology.epoch()));
+ EpochReady ready = onTopologyUpdateInternal(topology, startSync);
+ ready.fastPath.addCallback(() -> this.topology.onEpochSyncComplete(id, topology.epoch()));
configService.acknowledgeEpoch(ready, startSync);
- return ready.coordination;
+ return ready.fastPath;
}
@Override
@@ -512,29 +513,29 @@ public class Node implements ConfigurationService.Listener, NodeCommandStoreServ
return commandStores.ifLocal(context, key, epoch, epoch, ifLocal);
}
- public <T> void mapReduceConsumeLocal(TxnRequest<?> request, long minEpoch, long maxEpoch, MapReduceConsume<SafeCommandStore, T> mapReduceConsume)
+ public <T> Cancellable mapReduceConsumeLocal(TxnRequest<?> request, long minEpoch, long maxEpoch, MapReduceConsume<SafeCommandStore, T> mapReduceConsume)
{
- commandStores.mapReduceConsume(request, request.scope(), minEpoch, maxEpoch, mapReduceConsume);
+ return commandStores.mapReduceConsume(request, request.scope(), minEpoch, maxEpoch, mapReduceConsume);
}
- public <T> void mapReduceConsumeLocal(PreLoadContext context, RoutingKey key, long atEpoch, MapReduceConsume<SafeCommandStore, T> mapReduceConsume)
+ public <T> Cancellable mapReduceConsumeLocal(PreLoadContext context, RoutingKey key, long atEpoch, MapReduceConsume<SafeCommandStore, T> mapReduceConsume)
{
- mapReduceConsumeLocal(context, key, atEpoch, atEpoch, mapReduceConsume);
+ return mapReduceConsumeLocal(context, key, atEpoch, atEpoch, mapReduceConsume);
}
- public <T> void mapReduceConsumeLocal(PreLoadContext context, RoutingKey key, long minEpoch, long maxEpoch, MapReduceConsume<SafeCommandStore, T> mapReduceConsume)
+ public <T> Cancellable mapReduceConsumeLocal(PreLoadContext context, RoutingKey key, long minEpoch, long maxEpoch, MapReduceConsume<SafeCommandStore, T> mapReduceConsume)
{
- commandStores.mapReduceConsume(context, key, minEpoch, maxEpoch, mapReduceConsume);
+ return commandStores.mapReduceConsume(context, key, minEpoch, maxEpoch, mapReduceConsume);
}
- public <T> void mapReduceConsumeLocal(PreLoadContext context, Routables<?> keys, long minEpoch, long maxEpoch, MapReduceConsume<SafeCommandStore, T> mapReduceConsume)
+ public <T> Cancellable mapReduceConsumeLocal(PreLoadContext context, Routables<?> keys, long minEpoch, long maxEpoch, MapReduceConsume<SafeCommandStore, T> mapReduceConsume)
{
- commandStores.mapReduceConsume(context, keys, minEpoch, maxEpoch, mapReduceConsume);
+ return commandStores.mapReduceConsume(context, keys, minEpoch, maxEpoch, mapReduceConsume);
}
- public <T> void mapReduceConsumeAllLocal(PreLoadContext context, MapReduceConsume<SafeCommandStore, T> mapReduceConsume)
+ public <T> Cancellable mapReduceConsumeAllLocal(PreLoadContext context, MapReduceConsume<SafeCommandStore, T> mapReduceConsume)
{
- commandStores.mapReduceConsume(context, mapReduceConsume);
+ return commandStores.mapReduceConsume(context, mapReduceConsume);
}
// send to every node besides ourselves
@@ -814,7 +815,8 @@ public class Node implements ConfigurationService.Listener, NodeCommandStoreServ
return listeners;
}
- public RequestTimeouts requestTimeouts()
+ @Override
+ public Timeouts timeouts()
{
return timeouts;
}
diff --git a/accord-core/src/main/java/accord/local/NodeCommandStoreService.java b/accord-core/src/main/java/accord/local/NodeCommandStoreService.java
index bbf6d80d..4ee7686e 100644
--- a/accord-core/src/main/java/accord/local/NodeCommandStoreService.java
+++ b/accord-core/src/main/java/accord/local/NodeCommandStoreService.java
@@ -18,12 +18,14 @@
package accord.local;
+import accord.api.Timeouts;
import accord.primitives.Timestamp;
public interface NodeCommandStoreService extends TimeService
{
long epoch();
Node.Id id();
+ Timeouts timeouts();
DurableBefore durableBefore();
Timestamp uniqueNow();
Timestamp uniqueNow(Timestamp atLeast);
diff --git a/accord-core/src/main/java/accord/local/PreLoadContext.java b/accord-core/src/main/java/accord/local/PreLoadContext.java
index dfbfc66a..97697d06 100644
--- a/accord-core/src/main/java/accord/local/PreLoadContext.java
+++ b/accord-core/src/main/java/accord/local/PreLoadContext.java
@@ -22,22 +22,21 @@ import accord.api.RoutingKey;
import accord.local.cfk.CommandsForKey;
import accord.primitives.RoutingKeys;
import accord.primitives.TxnId;
-import com.google.common.collect.Iterators;
import accord.primitives.Unseekables;
import accord.utils.Invariants;
import net.nicoulaj.compilecommand.annotations.Inline;
-import com.google.common.collect.Sets;
-
-import java.util.AbstractCollection;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.Set;
+import java.util.AbstractList;
+import java.util.List;
import java.util.function.Consumer;
import javax.annotation.Nullable;
+import static accord.local.KeyHistory.ASYNC;
+import static accord.local.KeyHistory.INCR;
+import static accord.local.KeyHistory.NONE;
+import static accord.local.KeyHistory.SYNC;
+
/**
* Lists txnids and keys of commands and commands for key that will be needed for an operation. Used
* to ensure the necessary state is in memory for an operation before it executes.
@@ -50,32 +49,33 @@ public interface PreLoadContext
/**
* @return ids of the {@link Command} objects that need to be loaded into memory before this operation is run
*
+ * This should ONLY be non-null if primaryTxnId() is non-null
+ *
* TODO (expected): this is used for Apply, NotifyWaitingOn and listenerContexts; others only use a single txnId
* firstly, it would be nice to simply have that txnId as a single value.
* In the case of Apply, we can likely avoid loading all dependent transactions, if we can track which ranges
* out of memory have un-applied transactions (and try not to evict those that are not applied).
* Either way, the information we need in memory is super minimal for secondary transactions.
*/
- default Collection<TxnId> additionalTxnIds() { return Collections.emptyList(); }
+ default @Nullable TxnId additionalTxnId() { return null; }
- default Collection<TxnId> txnIds()
+ default List<TxnId> txnIds()
{
TxnId primaryTxnId = primaryTxnId();
- Collection<TxnId> additional = additionalTxnIds();
- if (primaryTxnId == null) return additional;
- if (additional.isEmpty()) return Collections.singleton(primaryTxnId);
- return new AbstractCollection<>()
+ TxnId additionalTxnId = additionalTxnId();
+ Invariants.checkState(primaryTxnId != null || additionalTxnId == null);
+ return new AbstractList<>()
{
@Override
- public Iterator<TxnId> iterator()
+ public TxnId get(int index)
{
- return Iterators.concat(Iterators.singletonIterator(primaryTxnId), additional.iterator());
+ return index == 0 ? primaryTxnId : additionalTxnId;
}
@Override
public int size()
{
- return 1 + additional.size();
+ return primaryTxnId == null ? 0 : additionalTxnId == null ? 1 : 2;
}
};
}
@@ -86,7 +86,9 @@ public interface PreLoadContext
TxnId primaryTxnId = primaryTxnId();
if (primaryTxnId != null)
consumer.accept(primaryTxnId);
- additionalTxnIds().forEach(consumer);
+ TxnId additionalTxnId = additionalTxnId();
+ if (additionalTxnId != null)
+ consumer.accept(additionalTxnId);
}
/**
@@ -99,57 +101,71 @@ public interface PreLoadContext
// TODO (required): specify epochs for which we should load, so we can narrow to owned keys
default Unseekables<?> keys() { return RoutingKeys.EMPTY; }
- default KeyHistory keyHistory() { return KeyHistory.NONE; }
+ default KeyHistory keyHistory() { return NONE; }
+ default boolean isEmpty()
+ {
+ boolean isEmpty = primaryTxnId() == null && keys().isEmpty();
+ Invariants.checkState(additionalTxnId() == null);
+ return isEmpty;
+ }
+
+ /**
+ * Is the provided PreLoadContext guaranteed to have a superset of our requested information?
+ * Note that for this calculation we are asking if all the information we want is known to be available,
+ * not whether a subset has been requested - that is, a superset with INCR or ASYNC key information
+ * cannot be relied upon for serving INCR or ASYNC subsets in this calculation.
+ */
default boolean isSubsetOf(PreLoadContext superset)
{
- KeyHistory requiredHistory = keyHistory();
- if (requiredHistory != KeyHistory.NONE && requiredHistory != superset.keyHistory())
- return false;
+ Unseekables<?> keys = keys();
+ if (!keys.isEmpty())
+ {
+ KeyHistory requiredHistory = keyHistory();
+ if (requiredHistory != NONE)
+ {
+ if (requiredHistory == INCR || requiredHistory == ASYNC)
+ requiredHistory = SYNC;
+ if (requiredHistory != superset.keyHistory())
+ return false;
+ }
- if (superset.keys().domain() != keys().domain() || !superset.keys().containsAll(keys()))
- return false;
+ Unseekables<?> supersetKeys = superset.keys();
+ if (supersetKeys.domain() != keys.domain() || !supersetKeys.containsAll(keys()))
+ return false;
+ }
TxnId primaryId = primaryTxnId();
- Collection<TxnId> additionalIds = additionalTxnIds();
- if (additionalIds.isEmpty())
+ TxnId additionalId = additionalTxnId();
+ if (additionalId == null)
{
- if (primaryId == null || primaryId.equals(superset.primaryTxnId()))
- return true;
-
- return superset.additionalTxnIds().contains(primaryTxnId());
+ return primaryId == null || primaryId.equals(superset.primaryTxnId()) || primaryId.equals(superset.additionalTxnId());
}
else
{
+ Invariants.checkState(primaryId != null);
TxnId supersetPrimaryId = superset.primaryTxnId();
- Set<TxnId> supersetAdditionalIds = superset.additionalTxnIds() instanceof Set ? (Set<TxnId>) superset.additionalTxnIds() : Sets.newHashSet(superset.additionalTxnIds());
- if (primaryId != null && !primaryId.equals(supersetPrimaryId) && !supersetAdditionalIds.contains(primaryId))
- return false;
-
- for (TxnId txnId : additionalIds)
- {
- if (!txnId.equals(supersetPrimaryId) && !supersetAdditionalIds.contains(txnId))
- return false;
- }
- return true;
+ TxnId supersetAdditionalId = superset.additionalTxnId();
+ return (primaryId.equals(supersetPrimaryId) || primaryId.equals(supersetAdditionalId)) && (additionalId.equals(supersetAdditionalId) || additionalId.equals(supersetPrimaryId));
}
}
- static PreLoadContext contextFor(TxnId primary, Collection<TxnId> additional, Unseekables<?> keys, KeyHistory keyHistory)
+ static PreLoadContext contextFor(@Nullable TxnId primary, @Nullable TxnId additional, Unseekables<?> keys, KeyHistory keyHistory)
{
- Invariants.checkState(!additional.contains(primary));
+ Invariants.checkState(primary == null ? additional == null : !primary.equals(additional));
return new Standard(primary, additional, keys, keyHistory);
}
class Standard implements PreLoadContext
{
- private final TxnId primary;
- private final Collection<TxnId> additional;
+ private final @Nullable TxnId primary;
+ private final @Nullable TxnId additional;
private final Unseekables<?> keys;
private final KeyHistory keyHistory;
- public Standard(TxnId primary, Collection<TxnId> additional, Unseekables<?> keys, KeyHistory keyHistory)
+ public Standard(@Nullable TxnId primary, @Nullable TxnId additional, Unseekables<?> keys, KeyHistory keyHistory)
{
+ Invariants.checkState(primary != null || additional == null);
this.primary = primary;
this.additional = additional;
this.keys = keys;
@@ -174,7 +190,7 @@ public interface PreLoadContext
}
@Override
- public Collection<TxnId> additionalTxnIds()
+ public TxnId additionalTxnId()
{
return additional;
}
@@ -189,29 +205,24 @@ public interface PreLoadContext
}
}
- static PreLoadContext contextFor(TxnId primary, Collection<TxnId> additional, Unseekables<?> keys)
+ static PreLoadContext contextFor(TxnId primary, TxnId additional, Unseekables<?> keys)
{
- return contextFor(primary, additional, keys, KeyHistory.NONE);
+ return contextFor(primary, additional, keys, NONE);
}
static PreLoadContext contextFor(TxnId primary, TxnId additional)
{
- return contextFor(primary, Collections.singletonList(additional), RoutingKeys.EMPTY);
- }
-
- static PreLoadContext contextFor(TxnId primary, TxnId additional, Unseekables<?> keys)
- {
- return contextFor(primary, Collections.singletonList(additional), keys);
+ return contextFor(primary, additional, RoutingKeys.EMPTY);
}
static PreLoadContext contextFor(TxnId txnId, Unseekables<?> keysOrRanges, KeyHistory keyHistory)
{
- return contextFor(txnId, Collections.emptyList(), keysOrRanges, keyHistory);
+ return contextFor(txnId, null, keysOrRanges, keyHistory);
}
static PreLoadContext contextFor(TxnId txnId, Unseekables<?> keysOrRanges)
{
- return contextFor(txnId, keysOrRanges, KeyHistory.NONE);
+ return contextFor(txnId, keysOrRanges, NONE);
}
static PreLoadContext contextFor(TxnId txnId)
@@ -219,34 +230,24 @@ public interface PreLoadContext
return contextFor(txnId, RoutingKeys.EMPTY);
}
- static PreLoadContext contextFor(TxnId primary, Collection<TxnId> additional)
- {
- return contextFor(primary, additional, RoutingKeys.EMPTY);
- }
-
static PreLoadContext contextFor(RoutingKey key, KeyHistory keyHistory)
{
- return contextFor(null, Collections.emptyList(), RoutingKeys.of(key), keyHistory);
+ return contextFor(null, null, RoutingKeys.of(key), keyHistory);
}
static PreLoadContext contextFor(RoutingKey key)
{
- return contextFor(key, KeyHistory.NONE);
- }
-
- static PreLoadContext contextFor(Collection<TxnId> ids, Unseekables<?> keys)
- {
- return contextFor(null, ids, keys);
+ return contextFor(key, NONE);
}
static PreLoadContext contextFor(Unseekables<?> keys)
{
- return contextFor(null, Collections.emptyList(), keys);
+ return contextFor(null, null, keys);
}
static PreLoadContext contextFor(Unseekables<?> keys, KeyHistory keyHistory)
{
- return contextFor(null, Collections.emptyList(), keys, keyHistory);
+ return contextFor(null, null, keys, keyHistory);
}
static PreLoadContext empty()
@@ -254,5 +255,5 @@ public interface PreLoadContext
return EMPTY_PRELOADCONTEXT;
}
- PreLoadContext EMPTY_PRELOADCONTEXT = contextFor(null, Collections.emptyList(), RoutingKeys.EMPTY);
+ PreLoadContext EMPTY_PRELOADCONTEXT = contextFor(null, null, RoutingKeys.EMPTY);
}
diff --git a/accord-core/src/main/java/accord/local/RedundantBefore.java b/accord-core/src/main/java/accord/local/RedundantBefore.java
index bcd6bcc1..4aef63a2 100644
--- a/accord-core/src/main/java/accord/local/RedundantBefore.java
+++ b/accord-core/src/main/java/accord/local/RedundantBefore.java
@@ -76,6 +76,11 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Entry>
// start inclusive, end exclusive
public final long startOwnershipEpoch, endOwnershipEpoch;
+ /**
+ * Represents the maximum TxnId we know to have a record of transactions before, or else they will be invalidated.
+ */
+ public final @Nonnull TxnId locallyWitnessedOrInvalidatedBefore;
+
/**
* Represents the maximum TxnId we know to have fully executed until locally for the range in question.
* Unless we are stale or pre-bootstrap, in which case no such guarantees can be made.
@@ -90,14 +95,20 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Entry>
* and for which we guarantee that any distributed decision that might need to be consulted is also recorded
* locally (i.e. it is known to be Stable locally, or else it did not execute)
*
- * We maintain locallyDecidedAndExecutedBefore that were reached prior to a new bootstrap exceeding them,
- * as these were reached correctly and can be used for GC.
+ * We maintain locallyDecidedAndAppliedOrInvalidatedBefore that were reached prior to a new bootstrap exceeding them,
+ * as these were reached correctly and can be used for pruning CommandsForKey.
*
* However, once a bootstrap has begun we cannot safely advance until shardAppliedOrInvalidatedBefore goes
* ahead of the bootstrappedAt, because we cannot guarantee to have any intervening decision recorded locally.
*/
public final @Nonnull TxnId locallyDecidedAndAppliedOrInvalidatedBefore;
+ /**
+ * Represents the maximum TxnId we know to have fully executed until across all healthy non-bootstrapping replicas
+ * for the range in question.
+ */
+ public final @Nonnull TxnId shardOnlyAppliedOrInvalidatedBefore;
+
/**
* Represents the maximum TxnId we know to have fully executed until across all healthy non-bootstrapping replicas
* for the range in question, including ourselves.
@@ -110,12 +121,6 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Entry>
*/
public final @Nonnull TxnId gcBefore;
- /**
- * Represents the maximum TxnId we know to have fully executed until across all healthy non-bootstrapping replicas
- * for the range in question.
- */
- public final @Nonnull TxnId shardOnlyAppliedOrInvalidatedBefore;
-
/**
* bootstrappedAt defines the txnId bounds we expect to maintain data for locally.
*
@@ -137,18 +142,20 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Entry>
*/
public final @Nullable Timestamp staleUntilAtLeast;
- public Entry(Range range, long startOwnershipEpoch, long endOwnershipEpoch, @Nonnull TxnId locallyAppliedOrInvalidatedBefore, @Nonnull TxnId locallyDecidedAndAppliedOrInvalidatedBefore, @Nonnull TxnId shardAppliedOrInvalidatedBefore, @Nonnull TxnId shardOnlyAppliedOrInvalidatedBefore, @Nonnull TxnId gcBefore, @Nonnull TxnId bootstrappedAt, @Nullable Timestamp staleUntilAtLeast)
+ public Entry(Range range, long startOwnershipEpoch, long endOwnershipEpoch, @Nonnull TxnId locallyWitnessedOrInvalidatedBefore, @Nonnull TxnId locallyAppliedOrInvalidatedBefore, @Nonnull TxnId locallyDecidedAndAppliedOrInvalidatedBefore, @Nonnull TxnId shardOnlyAppliedOrInvalidatedBefore, @Nonnull TxnId shardAppliedOrInvalidatedBefore, @Nonnull TxnId gcBefore, @Nonnull TxnId bootstrappedAt, @Nullable Timestamp staleUntilAtLeast)
{
this.range = range;
this.startOwnershipEpoch = startOwnershipEpoch;
this.endOwnershipEpoch = endOwnershipEpoch;
+ this.locallyWitnessedOrInvalidatedBefore = locallyWitnessedOrInvalidatedBefore;
this.locallyAppliedOrInvalidatedBefore = locallyAppliedOrInvalidatedBefore;
this.locallyDecidedAndAppliedOrInvalidatedBefore = locallyDecidedAndAppliedOrInvalidatedBefore;
+ this.shardOnlyAppliedOrInvalidatedBefore = shardOnlyAppliedOrInvalidatedBefore;
this.shardAppliedOrInvalidatedBefore = shardAppliedOrInvalidatedBefore;
this.gcBefore = gcBefore;
- this.shardOnlyAppliedOrInvalidatedBefore = shardOnlyAppliedOrInvalidatedBefore;
this.bootstrappedAt = bootstrappedAt;
this.staleUntilAtLeast = staleUntilAtLeast;
+ Invariants.checkArgument(locallyWitnessedOrInvalidatedBefore.equals(TxnId.NONE) || locallyWitnessedOrInvalidatedBefore.domain().isRange());
Invariants.checkArgument(locallyAppliedOrInvalidatedBefore.equals(TxnId.NONE) || locallyAppliedOrInvalidatedBefore.domain().isRange());
Invariants.checkArgument(locallyDecidedAndAppliedOrInvalidatedBefore.equals(TxnId.NONE) || locallyDecidedAndAppliedOrInvalidatedBefore.domain().isRange());
Invariants.checkArgument(shardAppliedOrInvalidatedBefore.equals(TxnId.NONE) || shardAppliedOrInvalidatedBefore.domain().isRange());
@@ -174,6 +181,7 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Entry>
long startEpoch = Long.max(cur.startOwnershipEpoch, add.startOwnershipEpoch);
long endEpoch = Long.min(cur.endOwnershipEpoch, add.endOwnershipEpoch);
+ int cw = cur.locallyWitnessedOrInvalidatedBefore.compareTo(add.locallyWitnessedOrInvalidatedBefore);
int cl = cur.locallyAppliedOrInvalidatedBefore.compareTo(add.locallyAppliedOrInvalidatedBefore);
int cd = cur.locallyDecidedAndAppliedOrInvalidatedBefore.compareTo(add.locallyDecidedAndAppliedOrInvalidatedBefore);
int cs = cur.shardOnlyAppliedOrInvalidatedBefore.compareTo(add.shardOnlyAppliedOrInvalidatedBefore);
@@ -181,11 +189,12 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Entry>
int cb = cur.bootstrappedAt.compareTo(add.bootstrappedAt);
int csu = compareStaleUntilAtLeast(cur.staleUntilAtLeast, add.staleUntilAtLeast);
- if (range.equals(cur.range) && startEpoch == cur.startOwnershipEpoch && endEpoch == cur.endOwnershipEpoch && cl >= 0 && cd >= 0 && cs >= 0 && cg >= 0 && cb >= 0 && csu >= 0)
+ if (range.equals(cur.range) && startEpoch == cur.startOwnershipEpoch && endEpoch == cur.endOwnershipEpoch && cw >= 0 && cl >= 0 && cd >= 0 && cs >= 0 && cg >= 0 && cb >= 0 && csu >= 0)
return cur;
- if (range.equals(add.range) && startEpoch == add.startOwnershipEpoch && endEpoch == add.endOwnershipEpoch && cl <= 0 && cd >= 0 && cs <= 0 && cg <= 0 && cb <= 0 && csu <= 0)
+ if (range.equals(add.range) && startEpoch == add.startOwnershipEpoch && endEpoch == add.endOwnershipEpoch && cw <= 0 && cl <= 0 && cd >= 0 && cs <= 0 && cg <= 0 && cb <= 0 && csu <= 0)
return add;
+ TxnId locallyWitnessedOrInvalidatedBefore = cw >= 0 ? cur.locallyWitnessedOrInvalidatedBefore : add.locallyWitnessedOrInvalidatedBefore;
TxnId locallyAppliedOrInvalidatedBefore = cl >= 0 ? cur.locallyAppliedOrInvalidatedBefore : add.locallyAppliedOrInvalidatedBefore;
TxnId locallyDecidedAndAppliedOrInvalidatedBefore = cd >= 0 ? cur.locallyDecidedAndAppliedOrInvalidatedBefore : add.locallyDecidedAndAppliedOrInvalidatedBefore;
TxnId shardOnlyAppliedOrInvalidatedBefore = cs >= 0 ? cur.shardOnlyAppliedOrInvalidatedBefore : add.shardOnlyAppliedOrInvalidatedBefore;
@@ -208,7 +217,7 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Entry>
if (staleUntilAtLeast != null && bootstrappedAt.compareTo(staleUntilAtLeast) >= 0)
staleUntilAtLeast = null;
- return new Entry(range, startEpoch, endEpoch, locallyAppliedOrInvalidatedBefore, locallyDecidedAndAppliedOrInvalidatedBefore, shardAppliedOrInvalidatedBefore, shardOnlyAppliedOrInvalidatedBefore, gcBefore, bootstrappedAt, staleUntilAtLeast);
+ return new Entry(range, startEpoch, endEpoch, locallyWitnessedOrInvalidatedBefore, locallyAppliedOrInvalidatedBefore, locallyDecidedAndAppliedOrInvalidatedBefore, shardOnlyAppliedOrInvalidatedBefore, shardAppliedOrInvalidatedBefore, gcBefore, bootstrappedAt, staleUntilAtLeast);
}
public Entry withGcBeforeBeforeAtLeast(TxnId newGcBefore)
@@ -219,13 +228,13 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Entry>
TxnId locallyAppliedOrInvalidatedBefore = TxnId.nonNullOrMax(this.locallyAppliedOrInvalidatedBefore, newGcBefore);
TxnId shardAppliedOrInvalidatedBefore = TxnId.nonNullOrMax(this.shardAppliedOrInvalidatedBefore, newGcBefore);
TxnId shardOnlyAppliedOrInvalidatedBefore = TxnId.nonNullOrMax(this.shardOnlyAppliedOrInvalidatedBefore, newGcBefore);
- return new Entry(range, startOwnershipEpoch, endOwnershipEpoch, locallyAppliedOrInvalidatedBefore, locallyDecidedAndAppliedOrInvalidatedBefore, shardAppliedOrInvalidatedBefore, shardOnlyAppliedOrInvalidatedBefore, newGcBefore, bootstrappedAt, staleUntilAtLeast);
+ return new Entry(range, startOwnershipEpoch, endOwnershipEpoch, locallyWitnessedOrInvalidatedBefore, locallyAppliedOrInvalidatedBefore, locallyDecidedAndAppliedOrInvalidatedBefore, shardOnlyAppliedOrInvalidatedBefore, shardAppliedOrInvalidatedBefore, newGcBefore, bootstrappedAt, staleUntilAtLeast);
}
@VisibleForImplementation
public Entry withEpochs(long start, long end)
{
- return new Entry(range, start, end, locallyAppliedOrInvalidatedBefore, locallyDecidedAndAppliedOrInvalidatedBefore, shardAppliedOrInvalidatedBefore, shardOnlyAppliedOrInvalidatedBefore, gcBefore, bootstrappedAt, staleUntilAtLeast);
+ return new Entry(range, start, end, locallyWitnessedOrInvalidatedBefore, locallyAppliedOrInvalidatedBefore, locallyDecidedAndAppliedOrInvalidatedBefore, shardOnlyAppliedOrInvalidatedBefore, shardAppliedOrInvalidatedBefore, gcBefore, bootstrappedAt, staleUntilAtLeast);
}
static @Nonnull RedundantStatus getAndMerge(Entry entry, @Nonnull RedundantStatus prev, TxnId txnId, Object ignore)
@@ -391,6 +400,11 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Entry>
return gcBefore;
}
+ public final TxnId locallyWitnessedBefore()
+ {
+ return locallyWitnessedOrInvalidatedBefore;
+ }
+
public final TxnId locallyRedundantBefore()
{
return locallyAppliedOrInvalidatedBefore;
@@ -425,12 +439,12 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Entry>
Entry withEpochs(int startEpoch, int endEpoch)
{
- return new Entry(range, startEpoch, endEpoch, locallyAppliedOrInvalidatedBefore, locallyDecidedAndAppliedOrInvalidatedBefore, shardAppliedOrInvalidatedBefore, shardOnlyAppliedOrInvalidatedBefore, gcBefore, bootstrappedAt, staleUntilAtLeast);
+ return new Entry(range, startEpoch, endEpoch, locallyWitnessedOrInvalidatedBefore, locallyAppliedOrInvalidatedBefore, locallyDecidedAndAppliedOrInvalidatedBefore, shardOnlyAppliedOrInvalidatedBefore, shardAppliedOrInvalidatedBefore, gcBefore, bootstrappedAt, staleUntilAtLeast);
}
public Entry withRange(Range range)
{
- return new Entry(range, startOwnershipEpoch, endOwnershipEpoch, locallyAppliedOrInvalidatedBefore, locallyDecidedAndAppliedOrInvalidatedBefore, shardAppliedOrInvalidatedBefore, shardOnlyAppliedOrInvalidatedBefore, gcBefore, bootstrappedAt, staleUntilAtLeast);
+ return new Entry(range, startOwnershipEpoch, endOwnershipEpoch, locallyWitnessedOrInvalidatedBefore, locallyAppliedOrInvalidatedBefore, locallyDecidedAndAppliedOrInvalidatedBefore, shardOnlyAppliedOrInvalidatedBefore, shardAppliedOrInvalidatedBefore, gcBefore, bootstrappedAt, staleUntilAtLeast);
}
public boolean equals(Object that)
@@ -469,16 +483,30 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Entry>
public static RedundantBefore EMPTY = new RedundantBefore();
private final Ranges staleRanges;
+ private final Timestamp maxBootstrap, maxGcBefore;
private RedundantBefore()
{
staleRanges = Ranges.EMPTY;
+ maxBootstrap = Timestamp.NONE;
+ maxGcBefore = Timestamp.NONE;
}
RedundantBefore(boolean inclusiveEnds, RoutingKey[] starts, Entry[] values)
{
super(inclusiveEnds, starts, values);
staleRanges = extractStaleRanges(values);
+ Timestamp maxBootstrap = Timestamp.NONE, maxGcBefore = Timestamp.NONE;
+ for (Entry entry : values)
+ {
+ if (entry == null) continue;
+ if (entry.bootstrappedAt.compareTo(maxBootstrap) > 0)
+ maxBootstrap = entry.bootstrappedAt;
+ if (entry.gcBefore.compareTo(maxGcBefore) > 0)
+ maxGcBefore = entry.gcBefore;
+ }
+ this.maxBootstrap = maxBootstrap;
+ this.maxGcBefore = maxGcBefore;
checkParanoid(starts, values);
}
@@ -504,29 +532,29 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Entry>
return Ranges.ofSortedAndDeoverlapped(staleRanges).mergeTouching();
}
- public static RedundantBefore create(AbstractRanges ranges, @Nonnull TxnId locallyAppliedOrInvalidatedBefore, @Nonnull TxnId shardAppliedOrInvalidatedBefore, @Nonnull TxnId gcBefore, @Nonnull TxnId bootstrappedAt)
+ public static RedundantBefore create(AbstractRanges ranges, @Nonnull TxnId locallyWitnessedOrInvalidatedBefore, @Nonnull TxnId locallyAppliedOrInvalidatedBefore, @Nonnull TxnId shardAppliedOrInvalidatedBefore, @Nonnull TxnId gcBefore, @Nonnull TxnId bootstrappedAt)
{
- return create(ranges, locallyAppliedOrInvalidatedBefore, shardAppliedOrInvalidatedBefore, gcBefore, bootstrappedAt, null);
+ return create(ranges, locallyWitnessedOrInvalidatedBefore, locallyAppliedOrInvalidatedBefore, shardAppliedOrInvalidatedBefore, gcBefore, bootstrappedAt, null);
}
- public static RedundantBefore create(AbstractRanges ranges, @Nonnull TxnId locallyAppliedOrInvalidatedBefore, @Nonnull TxnId shardAppliedOrInvalidatedBefore, @Nonnull TxnId gcBefore, @Nonnull TxnId bootstrappedAt, @Nullable Timestamp staleUntilAtLeast)
+ public static RedundantBefore create(AbstractRanges ranges, @Nonnull TxnId locallyWitnessedOrInvalidatedBefore, @Nonnull TxnId locallyAppliedOrInvalidatedBefore, @Nonnull TxnId shardAppliedOrInvalidatedBefore, @Nonnull TxnId gcBefore, @Nonnull TxnId bootstrappedAt, @Nullable Timestamp staleUntilAtLeast)
{
- return create(ranges, Long.MIN_VALUE, Long.MAX_VALUE, locallyAppliedOrInvalidatedBefore, shardAppliedOrInvalidatedBefore, gcBefore, bootstrappedAt, staleUntilAtLeast);
+ return create(ranges, Long.MIN_VALUE, Long.MAX_VALUE, locallyWitnessedOrInvalidatedBefore, locallyAppliedOrInvalidatedBefore, shardAppliedOrInvalidatedBefore, gcBefore, bootstrappedAt, staleUntilAtLeast);
}
- public static RedundantBefore create(AbstractRanges ranges, long startEpoch, long endEpoch, @Nonnull TxnId locallyAppliedOrInvalidatedBefore, @Nonnull TxnId shardOnlyAppliedOrInvalidatedBefore, @Nonnull TxnId gcBefore, @Nonnull TxnId bootstrappedAt)
+ public static RedundantBefore create(AbstractRanges ranges, long startEpoch, long endEpoch, @Nonnull TxnId locallyWitnessedOrInvalidatedBefore, @Nonnull TxnId locallyAppliedOrInvalidatedBefore, @Nonnull TxnId shardOnlyAppliedOrInvalidatedBefore, @Nonnull TxnId gcBefore, @Nonnull TxnId bootstrappedAt)
{
- return create(ranges, startEpoch, endEpoch, locallyAppliedOrInvalidatedBefore, shardOnlyAppliedOrInvalidatedBefore, gcBefore, bootstrappedAt, null);
+ return create(ranges, startEpoch, endEpoch, locallyWitnessedOrInvalidatedBefore, locallyAppliedOrInvalidatedBefore, shardOnlyAppliedOrInvalidatedBefore, gcBefore, bootstrappedAt, null);
}
- public static RedundantBefore create(AbstractRanges ranges, long startEpoch, long endEpoch, @Nonnull TxnId locallyAppliedOrInvalidatedBefore, @Nonnull TxnId shardOnlyAppliedOrInvalidatedBefore, @Nonnull TxnId gcBefore, @Nonnull TxnId bootstrappedAt, @Nullable Timestamp staleUntilAtLeast)
+ public static RedundantBefore create(AbstractRanges ranges, long startEpoch, long endEpoch, @Nonnull TxnId locallyWitnessedOrInvalidatedBefore, @Nonnull TxnId locallyAppliedOrInvalidatedBefore, @Nonnull TxnId shardOnlyAppliedOrInvalidatedBefore, @Nonnull TxnId gcBefore, @Nonnull TxnId bootstrappedAt, @Nullable Timestamp staleUntilAtLeast)
{
if (ranges.isEmpty())
return new RedundantBefore();
TxnId locallyDecidedAndAppliedOrInvalidatedBefore = locallyAppliedOrInvalidatedBefore;
TxnId shardAppliedOrInvalidatedBefore = TxnId.min(locallyAppliedOrInvalidatedBefore, shardOnlyAppliedOrInvalidatedBefore);
- Entry entry = new Entry(null, startEpoch, endEpoch, locallyAppliedOrInvalidatedBefore, locallyDecidedAndAppliedOrInvalidatedBefore, shardAppliedOrInvalidatedBefore, shardOnlyAppliedOrInvalidatedBefore, gcBefore, bootstrappedAt, staleUntilAtLeast);
+ Entry entry = new Entry(null, startEpoch, endEpoch, locallyWitnessedOrInvalidatedBefore, locallyAppliedOrInvalidatedBefore, locallyDecidedAndAppliedOrInvalidatedBefore, shardOnlyAppliedOrInvalidatedBefore, shardAppliedOrInvalidatedBefore, gcBefore, bootstrappedAt, staleUntilAtLeast);
Builder builder = new Builder(ranges.get(0).endInclusive(), ranges.size() * 2);
for (int i = 0 ; i < ranges.size() ; ++i)
{
@@ -610,6 +638,8 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Entry>
public Ranges removeShardRedundant(TxnId txnId, @Nonnull Timestamp executeAt, Ranges ranges)
{
Invariants.checkArgument(executeAt != null, "executeAt must not be null");
+ if (txnId.compareTo(maxGcBefore) >= 0)
+ return ranges;
return foldl(ranges, Entry::removeShardRedundant, ranges, txnId, executeAt, r -> false);
}
@@ -618,6 +648,8 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Entry>
*/
public Ranges removePreBootstrap(TxnId txnId, Ranges ranges)
{
+ if (maxBootstrap.compareTo(txnId) <= 0)
+ return ranges;
return foldl(ranges, Entry::removePreBootstrap, ranges, txnId, null, r -> false);
}
@@ -627,6 +659,8 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Entry>
public Ranges expectToExecute(TxnId txnId, @Nonnull Timestamp executeAt, Ranges ranges)
{
Invariants.checkArgument(executeAt != null, "executeAt must not be null");
+ if (maxBootstrap.compareTo(txnId) <= 0 && (staleRanges == null || !staleRanges.intersects(ranges)))
+ return ranges;
return foldl(ranges, Entry::expectToExecute, ranges, txnId, executeAt, r -> false);
}
@@ -635,6 +669,8 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Entry>
*/
public Ranges everExpectToExecute(TxnId txnId, Ranges ranges)
{
+ if (maxBootstrap.compareTo(txnId) <= 0 && (staleRanges == null || !staleRanges.intersects(ranges)))
+ return ranges;
return foldl(ranges, Entry::expectToExecute, ranges, txnId, null, r -> false);
}
@@ -660,7 +696,7 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Entry>
if (v.range.start().equals(start) && v.range.end().equals(end))
return v;
- return new Entry(v.range.newRange(start, end), v.startOwnershipEpoch, v.endOwnershipEpoch, v.locallyAppliedOrInvalidatedBefore, v.locallyDecidedAndAppliedOrInvalidatedBefore, v.shardAppliedOrInvalidatedBefore, v.shardOnlyAppliedOrInvalidatedBefore, v.gcBefore, v.bootstrappedAt, v.staleUntilAtLeast);
+ return new Entry(v.range.newRange(start, end), v.startOwnershipEpoch, v.endOwnershipEpoch, v.locallyWitnessedOrInvalidatedBefore, v.locallyAppliedOrInvalidatedBefore, v.locallyDecidedAndAppliedOrInvalidatedBefore, v.shardOnlyAppliedOrInvalidatedBefore, v.shardAppliedOrInvalidatedBefore, v.gcBefore, v.bootstrappedAt, v.staleUntilAtLeast);
}
@Override
@@ -679,7 +715,7 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Entry>
return new Entry(a.range.newRange(
a.range.start().compareTo(b.range.start()) <= 0 ? a.range.start() : b.range.start(),
a.range.end().compareTo(b.range.end()) >= 0 ? a.range.end() : b.range.end()
- ), a.startOwnershipEpoch, a.endOwnershipEpoch, a.locallyAppliedOrInvalidatedBefore, a.locallyDecidedAndAppliedOrInvalidatedBefore, a.shardAppliedOrInvalidatedBefore, a.shardOnlyAppliedOrInvalidatedBefore, a.gcBefore, a.bootstrappedAt, a.staleUntilAtLeast);
+ ), a.startOwnershipEpoch, a.endOwnershipEpoch, a.locallyWitnessedOrInvalidatedBefore, a.locallyAppliedOrInvalidatedBefore, a.locallyDecidedAndAppliedOrInvalidatedBefore, a.shardOnlyAppliedOrInvalidatedBefore, a.shardAppliedOrInvalidatedBefore, a.gcBefore, a.bootstrappedAt, a.staleUntilAtLeast);
}
@Override
diff --git a/accord-core/src/main/java/accord/local/SafeCommandStore.java b/accord-core/src/main/java/accord/local/SafeCommandStore.java
index 5de54fed..72fb5bed 100644
--- a/accord-core/src/main/java/accord/local/SafeCommandStore.java
+++ b/accord-core/src/main/java/accord/local/SafeCommandStore.java
@@ -27,6 +27,7 @@ import accord.api.DataStore;
import accord.api.LocalListeners;
import accord.api.ProgressLog;
import accord.api.RoutingKey;
+import accord.impl.SafeTimestampsForKey;
import accord.local.cfk.CommandsForKey;
import accord.local.cfk.SafeCommandsForKey;
import accord.local.cfk.UpdateUnmanagedMode;
@@ -43,13 +44,16 @@ import accord.primitives.TxnId;
import accord.primitives.Unseekable;
import accord.primitives.Unseekables;
import accord.utils.Invariants;
+import accord.utils.async.AsyncChain;
-import static accord.local.KeyHistory.COMMANDS;
+import static accord.local.KeyHistory.INCR;
import static accord.local.RedundantBefore.PreBootstrapOrStale.FULLY;
import static accord.local.cfk.UpdateUnmanagedMode.REGISTER;
import static accord.primitives.Routable.Domain.Range;
import static accord.primitives.Routables.Slice.Minimal;
import static accord.primitives.SaveStatus.Applied;
+import static accord.utils.Invariants.illegalArgument;
+import static accord.utils.Invariants.illegalState;
/**
* A CommandStore with exclusive access; a reference to this should not be retained outside of the scope of the method
@@ -86,7 +90,7 @@ public abstract class SafeCommandStore
Command command = safeCommand.current();
if (command.saveStatus().isUninitialised())
return null;
- return maybeCleanup(safeCommand);
+ return safeCommand;
}
// decidedExecuteAt == null if not yet PreCommitted
@@ -111,9 +115,36 @@ public abstract class SafeCommandStore
*/
public SafeCommand get(TxnId txnId, StoreParticipants participants)
{
- SafeCommand safeCommand = get(txnId);
- Commands.maybeCleanup(this, safeCommand, safeCommand.current(), participants);
- return safeCommand;
+ SafeCommand safeCommand = getInternal(txnId);
+ if (safeCommand == null)
+ throw notFound(txnId);
+
+ return maybeCleanup(safeCommand, safeCommand.current(), participants);
+ }
+
+ protected SafeCommand get(TxnId txnId)
+ {
+ SafeCommand safeCommand = getInternal(txnId);
+ if (safeCommand == null)
+ throw notFound(txnId);
+
+ return maybeCleanup(safeCommand, safeCommand.current(), StoreParticipants.empty(txnId));
+ }
+
+ public SafeCommand unsafeGet(TxnId txnId)
+ {
+ return get(txnId);
+ }
+
+ public SafeCommand unsafeGetNoCleanup(TxnId txnId)
+ {
+ return getInternal(txnId);
+ }
+
+ private RuntimeException notFound(TxnId txnId)
+ {
+ if (context().txnIds().contains(txnId)) throw illegalState("%s was specified in %s but was not returned by getInternal(key)", txnId, context().txnIds());
+ else throw illegalArgument("%s was not specified in %s", txnId, context().txnIds());
}
protected SafeCommand maybeCleanup(SafeCommand safeCommand)
@@ -133,25 +164,19 @@ public abstract class SafeCommandStore
*
* This permits efficient operation when a transaction involved in processing another transaction happens to be in memory.
*/
- public SafeCommand ifLoadedAndInitialised(TxnId txnId)
+ public SafeCommand ifLoadedAndInitialisedAndNotErased(TxnId txnId)
{
- SafeCommand safeCommand = getInternalIfLoadedAndInitialised(txnId);
- if (safeCommand == null)
+ SafeCommand safeCommand = getInternal(txnId);
+ if (safeCommand != null)
+ return safeCommand.current().saveStatus().isUninitialised() ? null : safeCommand;
+
+ safeCommand = ifLoadedAndInitialisedAndNotErasedInternal(txnId);
+ if (safeCommand == null || safeCommand.current().saveStatus().isUninitialised())
return null;
- return maybeCleanup(safeCommand, safeCommand.current(), StoreParticipants.empty(txnId));
- }
- protected SafeCommand get(TxnId txnId)
- {
- SafeCommand safeCommand = getInternal(txnId);
return maybeCleanup(safeCommand, safeCommand.current(), StoreParticipants.empty(txnId));
}
- public SafeCommand unsafeGet(TxnId txnId)
- {
- return get(txnId);
- }
-
protected SafeCommandsForKey maybeCleanup(SafeCommandsForKey safeCfk)
{
RedundantBefore.Entry entry = redundantBefore().get(safeCfk.key().toUnseekable());
@@ -168,7 +193,11 @@ public abstract class SafeCommandStore
*/
public final SafeCommandsForKey ifLoadedAndInitialised(RoutingKey key)
{
- SafeCommandsForKey safeCfk = getInternalIfLoadedAndInitialised(key);
+ SafeCommandsForKey safeCfk = getInternal(key);
+ if (safeCfk != null)
+ return safeCfk;
+
+ safeCfk = ifLoadedInternal(key);
if (safeCfk == null)
return null;
return maybeCleanup(safeCfk);
@@ -177,7 +206,11 @@ public abstract class SafeCommandStore
public SafeCommandsForKey get(RoutingKey key)
{
SafeCommandsForKey safeCfk = getInternal(key);
- return maybeCleanup(safeCfk);
+ if (safeCfk != null)
+ return maybeCleanup(safeCfk);
+
+ if (context().keys().contains(key)) throw illegalState("%s was specified in %s but was not returned by getInternal(key)", key, context().keys());
+ else throw illegalArgument("%s was not specified in %s", key, context());
}
public long preAcceptTimeout()
@@ -185,11 +218,30 @@ public abstract class SafeCommandStore
return agent().preAcceptTimeout();
}
+ /** Get anything already referenced (should include anything in PreLoadContext). If returned, should be initialised. */
protected abstract SafeCommand getInternal(TxnId txnId);
- protected abstract SafeCommand getInternalIfLoadedAndInitialised(TxnId txnId);
+ /** Get if available and initialised */
+ protected abstract SafeCommand ifLoadedAndInitialisedAndNotErasedInternal(TxnId txnId);
+ /** Get anything already referenced (should include anything in PreLoadContext) */
protected abstract SafeCommandsForKey getInternal(RoutingKey key);
- protected abstract SafeCommandsForKey getInternalIfLoadedAndInitialised(RoutingKey key);
- public abstract boolean canExecuteWith(PreLoadContext context);
+ /** Get if available */
+ protected abstract SafeCommandsForKey ifLoadedInternal(RoutingKey key);
+
+ // TODO (required): to be deprecated
+ public abstract SafeTimestampsForKey timestampsForKey(RoutingKey key);
+
+ public final boolean canExecuteWith(PreLoadContext context) { return canExecute(context) == context; }
+
+ /**
+ * Attempt to ready the provided PreLoadContext; if this can only be achieved partially, a new PreLoadContext
+ * will be returned containing the readily available data. If nothing is available, null will be returned.
+ */
+ public abstract @Nullable PreLoadContext canExecute(PreLoadContext context);
+
+ /**
+ * The current PreLoadContext, excluding any upgrade.
+ */
+ public abstract PreLoadContext context();
protected void update(Command prev, Command updated)
{
@@ -281,25 +333,35 @@ public abstract class SafeCommandStore
private static void updateManagedCommandsForKey(SafeCommandStore safeStore, Command prev, Command next)
{
- TxnId txnId = next.txnId();
StoreParticipants participants = next.participants().supplement(prev.participants());
Participants<?> update = next.hasBeen(Status.Committed) ? participants.hasTouched : participants.touches;
+ if (update.isEmpty())
+ return;
// TODO (expected): we don't want to insert any dependencies for those we only touch; we just need to record them as decided/applied for execution
- PreLoadContext context = PreLoadContext.contextFor(txnId, update, COMMANDS);
- // TODO (expected): execute immediately for any keys we already have loaded, and save only those we haven't for later
- if (safeStore.canExecuteWith(context))
+ PreLoadContext context = PreLoadContext.contextFor(update, INCR);
+ PreLoadContext execute = safeStore.canExecute(context);
+ if (execute != null)
{
- for (RoutingKey key : (AbstractUnseekableKeys)update)
- {
- safeStore.get(key).update(safeStore, next, update != participants.touches && !participants.touches(key));
- }
+ updateManagedCommandsForKey(safeStore, execute.keys(), participants, next);
}
- else
+ if (execute != context)
{
+ if (execute != null)
+ context = PreLoadContext.contextFor(update.without(execute.keys()), INCR);
+
+ Invariants.checkState(!context.keys().isEmpty());
safeStore = safeStore; // prevent accidental usage inside lambda
- safeStore.commandStore().execute(context, safeStore0 -> updateManagedCommandsForKey(safeStore0, prev, next))
- .begin(safeStore.commandStore().agent);
+ safeStore.commandStore().execute(context, safeStore0 -> updateManagedCommandsForKey(safeStore0, safeStore0.context().keys(), participants, next))
+ .begin(safeStore.commandStore().agent);
+ }
+ }
+
+ private static void updateManagedCommandsForKey(SafeCommandStore safeStore, Unseekables<?> update, StoreParticipants participants, Command next)
+ {
+ for (RoutingKey key : (AbstractUnseekableKeys)update)
+ {
+ safeStore.get(key).update(safeStore, next, update != participants.touches && !participants.touches(key));
}
}
@@ -308,35 +370,51 @@ public abstract class SafeCommandStore
TxnId txnId = next.txnId();
// TODO (required): use StoreParticipants.executes()
Command.WaitingOn waitingOn = next.asCommitted().waitingOn();
- RoutingKeys keys = waitingOn.keys;
+ RoutingKeys keys = waitingOn.waitingOnKeys();
// TODO (required): consider how execution works for transactions that await future deps and where the command store inherits additional keys in execution epoch
- PreLoadContext context = PreLoadContext.contextFor(txnId, keys, COMMANDS);
+ PreLoadContext context = PreLoadContext.contextFor(txnId, keys, INCR);
+ PreLoadContext execute = safeStore.canExecute(context);
// TODO (expected): execute immediately for any keys we already have loaded, and save only those we haven't for later
- if (safeStore.canExecuteWith(context))
+ if (execute != null)
+ {
+ updateUnmanagedCommandsForKey(safeStore, execute.keys(), txnId, mode);
+ }
+ if (execute == context)
{
- int index = 0;
- for (RoutingKey key : keys)
- {
- if (!waitingOn.isWaitingOnKey(index++)) continue;
- safeStore.get(key).registerUnmanaged(safeStore, safeStore.get(txnId), mode);
- }
-
if (next.txnId().is(Range))
- {
- CommandStore commandStore = safeStore.commandStore();
- Ranges ranges = next.participants().touches.toRanges();
- commandStore.registerTransitive(safeStore, next.partialDeps().rangeDeps);
- commandStore.markSynced(txnId, ranges);
- }
+ registerTransitive(safeStore, txnId, next);
}
else
{
+ if (execute != null)
+ context = PreLoadContext.contextFor(txnId, keys.without(execute.keys()), INCR);
+
safeStore = safeStore;
- safeStore.commandStore().execute(context, safeStore0 -> updateUnmanagedCommandsForKey(safeStore0, next, mode))
- .begin(safeStore.commandStore().agent);
+ CommandStore unsafeStore = safeStore.commandStore();
+ AsyncChain<Void> submit = unsafeStore.execute(context, safeStore0 -> updateUnmanagedCommandsForKey(safeStore0, safeStore0.context().keys(), txnId, mode));
+ if (next.txnId().is(Range))
+ submit = submit.flatMap(success -> unsafeStore.execute(PreLoadContext.empty(), safeStore0 -> registerTransitive(safeStore0, txnId, next)));
+ submit.begin(safeStore.commandStore().agent);
}
}
+ private static void updateUnmanagedCommandsForKey(SafeCommandStore safeStore, Unseekables<?> update, TxnId txnId, UpdateUnmanagedMode mode)
+ {
+ SafeCommand safeCommand = safeStore.get(txnId);
+ for (RoutingKey key : (AbstractUnseekableKeys)update)
+ {
+ safeStore.get(key).registerUnmanaged(safeStore, safeCommand, mode);
+ }
+ }
+
+ private static void registerTransitive(SafeCommandStore safeStore, TxnId txnId, Command next)
+ {
+ CommandStore commandStore = safeStore.commandStore();
+ Ranges ranges = next.participants().touches.toRanges();
+ commandStore.registerTransitive(safeStore, next.partialDeps().rangeDeps);
+ commandStore.markSynced(safeStore, txnId, ranges);
+ }
+
/**
* Visits keys first and then ranges, both in ascending order.
* Within each key or range visits all visible txnids needed for the given scope in ascending order of queried timestamp.
diff --git a/accord-core/src/main/java/accord/local/ShardDistributor.java b/accord-core/src/main/java/accord/local/ShardDistributor.java
index 9a69549f..7245af54 100644
--- a/accord-core/src/main/java/accord/local/ShardDistributor.java
+++ b/accord-core/src/main/java/accord/local/ShardDistributor.java
@@ -43,6 +43,8 @@ public interface ShardDistributor
*/
@Nullable Range splitRange(Range range, int from, int to, int totalSplits);
+ int numberOfSplitsPossible(Range range);
+
class EvenSplit<T> implements ShardDistributor
{
public interface Splitter<T>
@@ -71,6 +73,13 @@ public interface ShardDistributor
this.splitter = splitter;
}
+ @Override
+ public int numberOfSplitsPossible(Range range)
+ {
+ Splitter<T> splitter = this.splitter.apply(Ranges.of(range));
+ return splitter.min(splitter.sizeOf(range), Integer.MAX_VALUE);
+ }
+
@Override
public Range splitRange(Range range, int from, int to, int numSplits)
{
diff --git a/accord-core/src/main/java/accord/local/cfk/CommandsForKey.java b/accord-core/src/main/java/accord/local/cfk/CommandsForKey.java
index 4ee98a64..428d2ca0 100644
--- a/accord-core/src/main/java/accord/local/cfk/CommandsForKey.java
+++ b/accord-core/src/main/java/accord/local/cfk/CommandsForKey.java
@@ -33,7 +33,6 @@ import accord.api.RoutingKey;
import accord.api.VisibleForImplementation;
import accord.impl.CommandsSummary;
import accord.local.Command;
-import accord.local.CommandStore;
import accord.local.RedundantBefore;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
@@ -208,7 +207,7 @@ public class CommandsForKey extends CommandsForKeyUpdate implements CommandsSumm
private static boolean reportLinearizabilityViolations = true;
private static final boolean ELIDE_TRANSITIVE_DEPENDENCIES = true;
- public static final RedundantBefore.Entry NO_BOUNDS_INFO = new RedundantBefore.Entry(null, Long.MIN_VALUE, Long.MAX_VALUE, TxnId.NONE, TxnId.NONE, TxnId.NONE, TxnId.NONE, TxnId.NONE, TxnId.NONE, null);
+ public static final RedundantBefore.Entry NO_BOUNDS_INFO = new RedundantBefore.Entry(null, Long.MIN_VALUE, Long.MAX_VALUE, TxnId.NONE, TxnId.NONE, TxnId.NONE, TxnId.NONE, TxnId.NONE, TxnId.NONE, TxnId.NONE, null);
public static final TxnInfo NO_INFO = TxnInfo.create(TxnId.NONE, HISTORICAL, false, TxnId.NONE, Ballot.ZERO);
public static final TxnInfo[] NO_INFOS = new TxnInfo[0];
public static final Unmanaged[] NO_PENDING_UNMANAGED = new Unmanaged[0];
@@ -1665,9 +1664,6 @@ public class CommandsForKey extends CommandsForKeyUpdate implements CommandsSumm
return new CommandsForKeyUpdateWithPostProcess(newCfk, newPostProcess);
}
- if (CommandStore.current().toString().equals("DelayedCommandStore{id=33,node=5}") && key.toString().equals("393#12078"))
- System.out.println();
-
TxnInfo[] newById = pruneById(byId, boundsInfo, newBoundsInfo);
int newPrunedBeforeById = prunedBeforeId(newById, prunedBefore(), redundantBefore(newBoundsInfo));
Object[] newLoadingPruned = Pruning.removeRedundantLoadingPruned(loadingPruned, redundantBefore(newBoundsInfo));
@@ -1702,6 +1698,11 @@ public class CommandsForKey extends CommandsForKeyUpdate implements CommandsSumm
return Pruning.maybePrune(this, pruneInterval, minHlcDelta);
}
+ public CommandsForKey maximalPrune()
+ {
+ return Pruning.maybePrune(this, 0, 0);
+ }
+
int insertPos(Timestamp timestamp)
{
return insertPos(byId, timestamp);
@@ -1918,6 +1919,11 @@ public class CommandsForKey extends CommandsForKeyUpdate implements CommandsSumm
}
}
+ public boolean isEmpty()
+ {
+ return byId.length == 0 && unmanageds.length == 0;
+ }
+
public static boolean reportLinearizabilityViolations()
{
return reportLinearizabilityViolations;
diff --git a/accord-core/src/main/java/accord/local/cfk/NotifySink.java b/accord-core/src/main/java/accord/local/cfk/NotifySink.java
index 5a6c7f5b..b6a415df 100644
--- a/accord-core/src/main/java/accord/local/cfk/NotifySink.java
+++ b/accord-core/src/main/java/accord/local/cfk/NotifySink.java
@@ -34,7 +34,7 @@ import accord.primitives.RoutingKeys;
import accord.primitives.TxnId;
import accord.utils.Invariants;
-import static accord.local.KeyHistory.COMMANDS;
+import static accord.local.KeyHistory.SYNC;
interface NotifySink
{
@@ -49,7 +49,7 @@ interface NotifySink
@Override
public void notWaiting(SafeCommandStore safeStore, TxnId txnId, RoutingKey key)
{
- SafeCommand safeCommand = safeStore.ifLoadedAndInitialised(txnId);
+ SafeCommand safeCommand = safeStore.ifLoadedAndInitialisedAndNotErased(txnId);
if (safeCommand != null) notWaiting(safeStore, safeCommand, key);
else
{
@@ -122,7 +122,7 @@ interface NotifySink
RoutingKeys keys = RoutingKeys.of(key);
//noinspection ConstantConditions,SillyAssignment
safeStore = safeStore; // prevent use in lambda
- safeStore.commandStore().execute(PreLoadContext.contextFor(txnId, keys, COMMANDS), safeStore0 -> {
+ safeStore.commandStore().execute(PreLoadContext.contextFor(txnId, keys, SYNC), safeStore0 -> {
doNotifyAlreadyReady(safeStore0, txnId, key);
}).begin(safeStore.agent());
}
diff --git a/accord-core/src/main/java/accord/local/cfk/PostProcess.java b/accord-core/src/main/java/accord/local/cfk/PostProcess.java
index 8b21352b..1c187a00 100644
--- a/accord-core/src/main/java/accord/local/cfk/PostProcess.java
+++ b/accord-core/src/main/java/accord/local/cfk/PostProcess.java
@@ -40,7 +40,7 @@ import accord.utils.Invariants;
import accord.utils.SortedArrays;
import accord.utils.btree.BTree;
-import static accord.local.KeyHistory.COMMANDS;
+import static accord.local.KeyHistory.SYNC;
import static accord.local.cfk.CommandsForKey.InternalStatus.APPLIED;
import static accord.local.cfk.CommandsForKey.InternalStatus.INVALID_OR_TRUNCATED_OR_PRUNED;
import static accord.local.cfk.CommandsForKey.InternalStatus.STABLE;
@@ -91,10 +91,10 @@ abstract class PostProcess
for (TxnId txnId : load)
{
safeStore = safeStore; // make it unsafe for use in lambda
- SafeCommand safeCommand = safeStore.ifLoadedAndInitialised(txnId);
+ SafeCommand safeCommand = safeStore.ifLoadedAndInitialisedAndNotErased(txnId);
if (safeCommand != null) load(safeStore, safeCommand, safeCfk, notifySink);
else
- safeStore.commandStore().execute(PreLoadContext.contextFor(txnId, RoutingKeys.of(key), COMMANDS), safeStore0 -> {
+ safeStore.commandStore().execute(PreLoadContext.contextFor(txnId, RoutingKeys.of(key), SYNC), safeStore0 -> {
load(safeStore0, safeStore0.unsafeGet(txnId), safeStore0.get(key), notifySink);
}).begin(safeStore.agent());
}
@@ -196,7 +196,7 @@ abstract class PostProcess
List<PostProcess> nestedNotify = new ArrayList<>();
for (TxnId txnId : notify)
{
- SafeCommand safeCommand = safeStore.ifLoadedAndInitialised(txnId);
+ SafeCommand safeCommand = safeStore.ifLoadedAndInitialisedAndNotErased(txnId);
if (safeCommand != null)
{
CommandsForKeyUpdate update = updateUnmanaged(cfk, safeCommand, UPDATE, addUnmanageds);
diff --git a/accord-core/src/main/java/accord/local/cfk/Updating.java b/accord-core/src/main/java/accord/local/cfk/Updating.java
index 97adbb49..079f8b45 100644
--- a/accord-core/src/main/java/accord/local/cfk/Updating.java
+++ b/accord-core/src/main/java/accord/local/cfk/Updating.java
@@ -46,7 +46,7 @@ import accord.utils.RelationMultiMap;
import accord.utils.SortedArrays;
import accord.utils.SortedCursor;
-import static accord.local.KeyHistory.COMMANDS;
+import static accord.local.KeyHistory.SYNC;
import static accord.local.cfk.CommandsForKey.InternalStatus.APPLIED;
import static accord.local.cfk.CommandsForKey.InternalStatus.COMMITTED;
import static accord.local.cfk.CommandsForKey.InternalStatus.INVALID_OR_TRUNCATED_OR_PRUNED;
@@ -702,7 +702,7 @@ class Updating
static void updateUnmanagedAsync(CommandStore commandStore, TxnId txnId, RoutingKey key, NotifySink notifySink)
{
- PreLoadContext context = PreLoadContext.contextFor(txnId, RoutingKeys.of(key), COMMANDS);
+ PreLoadContext context = PreLoadContext.contextFor(txnId, RoutingKeys.of(key), SYNC);
commandStore.execute(context, safeStore -> {
SafeCommandsForKey safeCommandsForKey = safeStore.get(key);
CommandsForKey cur = safeCommandsForKey.current();
diff --git a/accord-core/src/main/java/accord/messages/AbstractEpochRequest.java b/accord-core/src/main/java/accord/messages/AbstractEpochRequest.java
deleted file mode 100644
index 482fcf33..00000000
--- a/accord-core/src/main/java/accord/messages/AbstractEpochRequest.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * 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.messages;
-
-import accord.local.Node;
-import accord.local.PreLoadContext;
-import accord.local.SafeCommandStore;
-import accord.primitives.TxnId;
-import accord.utils.MapReduceConsume;
-
-public abstract class AbstractEpochRequest<R extends Reply> implements PreLoadContext, Request, MapReduceConsume<SafeCommandStore, R>
-{
- public final TxnId txnId;
- protected transient Node node;
- protected transient Node.Id replyTo;
- protected transient ReplyContext replyContext;
-
- protected AbstractEpochRequest(TxnId txnId)
- {
- this.txnId = txnId;
- }
-
- @Override
- public void process(Node on, Node.Id replyTo, ReplyContext replyContext)
- {
- this.node = on;
- this.replyTo = replyTo;
- this.replyContext = replyContext;
- process();
- }
-
- protected abstract void process();
-
- @Override
- public R reduce(R o1, R o2)
- {
- throw new IllegalStateException();
- }
-
- @Override
- public void accept(R reply, Throwable failure)
- {
- node.reply(replyTo, replyContext, reply, failure);
- }
-
- @Override
- public TxnId primaryTxnId()
- {
- return txnId;
- }
-}
diff --git a/accord-core/src/main/java/accord/messages/AbstractRequest.java b/accord-core/src/main/java/accord/messages/AbstractRequest.java
new file mode 100644
index 00000000..0306d551
--- /dev/null
+++ b/accord-core/src/main/java/accord/messages/AbstractRequest.java
@@ -0,0 +1,180 @@
+/*
+ * 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.messages;
+
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import javax.annotation.Nullable;
+
+import accord.api.Timeouts;
+import accord.api.Timeouts.RegisteredTimeout;
+import accord.local.Node;
+import accord.local.PreLoadContext;
+import accord.local.SafeCommandStore;
+import accord.primitives.TxnId;
+import accord.utils.MapReduceConsume;
+import accord.utils.async.Cancellable;
+
+import static java.util.concurrent.TimeUnit.MICROSECONDS;
+
+public abstract class AbstractRequest<R extends Reply> implements PreLoadContext, Request, MapReduceConsume<SafeCommandStore, R>, Timeouts.Timeout
+{
+ static class Cancellation implements Cancellable
+ {
+ final RegisteredTimeout timeout;
+ final Cancellable cancel;
+
+ Cancellation(RegisteredTimeout timeout, Cancellable cancel)
+ {
+ this.timeout = timeout;
+ this.cancel = cancel;
+ }
+
+ @Override
+ public void cancel()
+ {
+ if (timeout != null) timeout.cancel();
+ if (cancel != null) cancel.cancel();
+ }
+ }
+
+ private static final Cancellation DONE = new Cancellation(null, null);
+ private static final Cancellation EMPTY = new Cancellation(null, null);
+
+ public final TxnId txnId;
+ protected transient Node node;
+ protected transient Node.Id replyTo;
+ protected transient ReplyContext replyContext;
+
+ private transient volatile Cancellation cancellation;
+ private static final AtomicReferenceFieldUpdater<AbstractRequest, Cancellation> cancellationUpdater = AtomicReferenceFieldUpdater.newUpdater(AbstractRequest.class, Cancellation.class, "cancellation");
+
+ protected AbstractRequest(TxnId txnId)
+ {
+ this.txnId = txnId;
+ }
+
+ @Override
+ public final void process(Node on, Node.Id replyTo, ReplyContext replyContext)
+ {
+ this.node = on;
+ this.replyTo = replyTo;
+ this.replyContext = replyContext;
+ Cancellable cancel = submit();
+ if (cancel != null)
+ {
+ long expiresAt = node.agent().expiresAt(replyContext, MICROSECONDS);
+ if (expiresAt > 0)
+ {
+ RegisteredTimeout timeout = node.timeouts().registerWithDelay(this, expiresAt, MICROSECONDS);
+ Cancellation cancellation = new Cancellation(timeout, cancel);
+ if (!cancellationUpdater.compareAndSet(this, null, cancellation))
+ cancellation.cancel();
+ }
+ }
+ }
+
+ protected abstract Cancellable submit();
+
+ @Override
+ public final void accept(R reply, Throwable failure)
+ {
+ cleanup(processedInternal());
+ acceptInternal(reply, failure);
+ }
+
+ protected @Nullable Cancellable processedInternal()
+ {
+ return clearInternal().timeout;
+ }
+
+ protected void acceptInternal(R reply, Throwable failure)
+ {
+ node.reply(replyTo, replyContext, reply, failure);
+ }
+
+ @Override
+ public final void timeout()
+ {
+ cleanup(timeoutInternal());
+ }
+
+ protected @Nullable Cancellable timeoutInternal()
+ {
+ return clearInternal().cancel;
+ }
+
+ protected boolean cancel()
+ {
+ Cancellation clear = clearInternal();
+ if (clear == DONE)
+ return false;
+
+ cleanup(clear);
+ return true;
+ }
+
+ /**
+ * invoked on any termination, to ensure state is cleared
+ * @return
+ */
+ protected void clear()
+ {
+ cleanup(clearInternal());
+ }
+
+ /**
+ * invoked on any termination, to ensure state is cleared
+ * @return
+ */
+ protected Cancellation clearInternal()
+ {
+ while (true)
+ {
+ // can loop at most once
+ Cancellation cur = cancellation;
+ if (cur == DONE || cancellationUpdater.compareAndSet(this, cur, DONE))
+ return cur != null ? cur : EMPTY;
+ }
+ }
+
+ @Override
+ public R reduce(R o1, R o2)
+ {
+ throw new IllegalStateException();
+ }
+
+ @Override
+ public TxnId primaryTxnId()
+ {
+ return txnId;
+ }
+
+ @Override
+ public int stripe()
+ {
+ return txnId.hashCode();
+ }
+
+ private static void cleanup(@Nullable Cancellable cancel)
+ {
+ if (cancel != null)
+ cancel.cancel();
+ }
+}
diff --git a/accord-core/src/main/java/accord/messages/Accept.java b/accord-core/src/main/java/accord/messages/Accept.java
index 3731cea2..1ca608ae 100644
--- a/accord-core/src/main/java/accord/messages/Accept.java
+++ b/accord-core/src/main/java/accord/messages/Accept.java
@@ -42,6 +42,7 @@ import accord.primitives.TxnId;
import accord.primitives.Unseekables;
import accord.topology.Topologies;
import accord.utils.Invariants;
+import accord.utils.async.Cancellable;
import static accord.local.Commands.AcceptOutcome.Redundant;
import static accord.local.Commands.AcceptOutcome.RejectedBallot;
@@ -121,15 +122,9 @@ public class Accept extends TxnRequest.WithUnsynced<Accept.AcceptReply>
}
@Override
- public void accept(AcceptReply reply, Throwable failure)
+ public Cancellable submit()
{
- node.reply(replyTo, replyContext, reply, failure);
- }
-
- @Override
- public void process()
- {
- node.mapReduceConsumeLocal(this, minEpoch, executeAt.epoch(), this);
+ return node.mapReduceConsumeLocal(this, minEpoch, executeAt.epoch(), this);
}
@Override
@@ -147,7 +142,7 @@ public class Accept extends TxnRequest.WithUnsynced<Accept.AcceptReply>
@Override
public KeyHistory keyHistory()
{
- return KeyHistory.COMMANDS;
+ return KeyHistory.SYNC;
}
@Override
@@ -248,7 +243,7 @@ public class Accept extends TxnRequest.WithUnsynced<Accept.AcceptReply>
}
}
- public static class Invalidate extends AbstractEpochRequest<AcceptReply>
+ public static class Invalidate extends AbstractRequest<AcceptReply>
{
public final Ballot ballot;
// should not be a non-participating home key
@@ -262,9 +257,9 @@ public class Accept extends TxnRequest.WithUnsynced<Accept.AcceptReply>
}
@Override
- public void process()
+ public Cancellable submit()
{
- node.mapReduceConsumeLocal(this, someKey, txnId.epoch(), this);
+ return node.mapReduceConsumeLocal(this, someKey, txnId.epoch(), this);
}
@Override
diff --git a/accord-core/src/main/java/accord/messages/Apply.java b/accord-core/src/main/java/accord/messages/Apply.java
index 5d50815e..862e2658 100644
--- a/accord-core/src/main/java/accord/messages/Apply.java
+++ b/accord-core/src/main/java/accord/messages/Apply.java
@@ -42,6 +42,7 @@ import accord.primitives.Unseekables;
import accord.primitives.Writes;
import accord.topology.Topologies;
import accord.utils.Invariants;
+import accord.utils.async.Cancellable;
public class Apply extends TxnRequest<ApplyReply>
{
@@ -144,9 +145,9 @@ public class Apply extends TxnRequest<ApplyReply>
}
@Override
- public void process()
+ public Cancellable submit()
{
- node.mapReduceConsumeLocal(this, minEpoch(txnId), executeAt.epoch(), this);
+ return node.mapReduceConsumeLocal(this, minEpoch(txnId), executeAt.epoch(), this);
}
private long minEpoch(TxnId txnId)
@@ -195,12 +196,6 @@ public class Apply extends TxnRequest<ApplyReply>
return a.compareTo(b) >= 0 ? a : b;
}
- @Override
- public void accept(ApplyReply reply, Throwable failure)
- {
- node.reply(replyTo, replyContext, reply, failure);
- }
-
@Override
public TxnId primaryTxnId()
{
@@ -216,7 +211,7 @@ public class Apply extends TxnRequest<ApplyReply>
@Override
public KeyHistory keyHistory()
{
- return KeyHistory.COMMANDS;
+ return KeyHistory.ASYNC;
}
@Override
diff --git a/accord-core/src/main/java/accord/messages/Await.java b/accord-core/src/main/java/accord/messages/Await.java
index dfb7ee7e..53a89a0e 100644
--- a/accord-core/src/main/java/accord/messages/Await.java
+++ b/accord-core/src/main/java/accord/messages/Await.java
@@ -29,8 +29,8 @@ import javax.annotation.Nullable;
import accord.api.LocalListeners;
import accord.api.ProgressLog.BlockedUntil;
import accord.api.RemoteListeners;
-import accord.api.RequestTimeouts;
-import accord.api.RequestTimeouts.RegisteredTimeout;
+import accord.api.Timeouts;
+import accord.api.Timeouts.RegisteredTimeout;
import accord.local.Command;
import accord.local.Commands;
import accord.local.Node;
@@ -49,7 +49,7 @@ import accord.utils.Invariants;
import accord.utils.MapReduceConsume;
import static accord.messages.TxnRequest.computeScope;
-import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.MICROSECONDS;
/**
* Contact a replica to perform a synchronous or asynchronous wait on some condition for a transaction + some keys.
@@ -59,7 +59,7 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS;
*
* TODO (desired): return an OK message indicating we're waiting synchronously
*/
-public class Await implements Request, MapReduceConsume<SafeCommandStore, Void>, PreLoadContext, LocalListeners.ComplexListener, RequestTimeouts.Timeout
+public class Await implements Request, MapReduceConsume<SafeCommandStore, Void>, PreLoadContext, LocalListeners.ComplexListener, Timeouts.Timeout
{
public static class SerializerSupport
{
@@ -194,10 +194,13 @@ public class Await implements Request, MapReduceConsume<SafeCommandStore, Void>,
int waitingOn = synchronouslyWaitingOnUpdater.decrementAndGet(this);
if (waitingOn >= 0)
{
- long replyTimeout = node.agent().replyTimeout(replyContext, MILLISECONDS);
- timeout = node.requestTimeouts().register(this, replyTimeout, MILLISECONDS);
- if (-1 == synchronouslyWaitingOn)
- timeout.cancel(); // we could leave a dangling timeout in this rare race condition
+ long expiresAtMicros = node.agent().expiresAt(replyContext, MICROSECONDS);
+ if (expiresAtMicros > 0)
+ {
+ timeout = node.timeouts().registerAt(this, expiresAtMicros, MICROSECONDS);
+ if (-1 == synchronouslyWaitingOn)
+ timeout.cancel(); // we could leave a dangling timeout in this rare race condition
+ }
}
else
{
diff --git a/accord-core/src/main/java/accord/messages/BeginInvalidation.java b/accord-core/src/main/java/accord/messages/BeginInvalidation.java
index a1491d2d..8d826135 100644
--- a/accord-core/src/main/java/accord/messages/BeginInvalidation.java
+++ b/accord-core/src/main/java/accord/messages/BeginInvalidation.java
@@ -25,6 +25,7 @@ import accord.primitives.*;
import accord.topology.Shard;
import accord.topology.Topologies;
import accord.utils.SortedList;
+import accord.utils.async.Cancellable;
import javax.annotation.Nullable;
import java.util.Arrays;
@@ -34,7 +35,7 @@ import static accord.primitives.Route.castToFullRoute;
import static accord.primitives.Route.isFullRoute;
import static accord.utils.Functions.mapReduceNonNull;
-public class BeginInvalidation extends AbstractEpochRequest<BeginInvalidation.InvalidateReply> implements Request, PreLoadContext
+public class BeginInvalidation extends AbstractRequest<BeginInvalidation.InvalidateReply> implements Request, PreLoadContext
{
public final Ballot ballot;
public final Participants<?> participants;
@@ -54,9 +55,9 @@ public class BeginInvalidation extends AbstractEpochRequest<BeginInvalidation.In
}
@Override
- public void process()
+ public Cancellable submit()
{
- node.mapReduceConsumeLocal(this, participants, txnId.epoch(), txnId.epoch(), this);
+ return node.mapReduceConsumeLocal(this, participants, txnId.epoch(), txnId.epoch(), this);
}
@Override
diff --git a/accord-core/src/main/java/accord/messages/BeginRecovery.java b/accord-core/src/main/java/accord/messages/BeginRecovery.java
index e91a4b05..8d4b483d 100644
--- a/accord-core/src/main/java/accord/messages/BeginRecovery.java
+++ b/accord-core/src/main/java/accord/messages/BeginRecovery.java
@@ -39,6 +39,7 @@ import accord.primitives.Unseekables;
import accord.primitives.Writes;
import accord.topology.Topologies;
import accord.utils.Invariants;
+import accord.utils.async.Cancellable;
import static accord.local.SafeCommandStore.TestDep.WITH;
import static accord.local.SafeCommandStore.TestDep.WITHOUT;
@@ -90,9 +91,9 @@ public class BeginRecovery extends TxnRequest.WithUnsynced<BeginRecovery.Recover
}
@Override
- protected void process()
+ protected Cancellable submit()
{
- node.mapReduceConsumeLocal(this, minEpoch, executeAtOrTxnIdEpoch, this);
+ return node.mapReduceConsumeLocal(this, minEpoch, executeAtOrTxnIdEpoch, this);
}
@Override
@@ -193,12 +194,6 @@ public class BeginRecovery extends TxnRequest.WithUnsynced<BeginRecovery.Recover
);
}
- @Override
- public void accept(RecoverReply reply, Throwable failure)
- {
- node.reply(replyTo, replyContext, reply, failure);
- }
-
@Override
public TxnId primaryTxnId()
{
@@ -214,7 +209,7 @@ public class BeginRecovery extends TxnRequest.WithUnsynced<BeginRecovery.Recover
@Override
public KeyHistory keyHistory()
{
- return KeyHistory.RECOVERY;
+ return KeyHistory.RECOVER;
}
@Override
diff --git a/accord-core/src/main/java/accord/messages/CalculateDeps.java b/accord-core/src/main/java/accord/messages/CalculateDeps.java
index 1bcb89d7..593f8eb9 100644
--- a/accord-core/src/main/java/accord/messages/CalculateDeps.java
+++ b/accord-core/src/main/java/accord/messages/CalculateDeps.java
@@ -32,11 +32,12 @@ import accord.primitives.TxnId;
import accord.primitives.Unseekables;
import accord.topology.Topologies;
import accord.utils.Invariants;
+import accord.utils.async.Cancellable;
import static accord.messages.PreAccept.calculateDeps;
import static accord.primitives.EpochSupplier.constant;
-public class CalculateDeps extends TxnRequest.WithUnsynced<Deps>
+public class CalculateDeps extends TxnRequest.WithUnsynced<CalculateDeps.CalculateDepsOk>
{
public static final class SerializationSupport
{
@@ -61,28 +62,22 @@ public class CalculateDeps extends TxnRequest.WithUnsynced<Deps>
}
@Override
- public void process()
+ public Cancellable submit()
{
- node.mapReduceConsumeLocal(this, minEpoch, executeAt.epoch(), this);
+ return node.mapReduceConsumeLocal(this, minEpoch, executeAt.epoch(), this);
}
@Override
- public Deps apply(SafeCommandStore safeStore)
+ public CalculateDepsOk apply(SafeCommandStore safeStore)
{
StoreParticipants participants = StoreParticipants.read(safeStore, scope, txnId, minEpoch, executeAt.epoch());
- return calculateDeps(safeStore, txnId, participants, constant(minEpoch), executeAt);
+ return new CalculateDepsOk(calculateDeps(safeStore, txnId, participants, constant(minEpoch), executeAt));
}
@Override
- public Deps reduce(Deps deps1, Deps deps2)
+ public CalculateDepsOk reduce(CalculateDepsOk ok1, CalculateDepsOk ok2)
{
- return deps1.with(deps2);
- }
-
- @Override
- public void accept(Deps result, Throwable failure)
- {
- node.reply(replyTo, replyContext, result != null ? new CalculateDepsOk(result) : null, failure);
+ return new CalculateDepsOk(ok1.deps.with(ok2.deps));
}
@Override
@@ -116,7 +111,7 @@ public class CalculateDeps extends TxnRequest.WithUnsynced<Deps>
@Override
public KeyHistory keyHistory()
{
- return KeyHistory.COMMANDS;
+ return KeyHistory.SYNC;
}
public static class CalculateDepsOk implements Reply
diff --git a/accord-core/src/main/java/accord/messages/CheckStatus.java b/accord-core/src/main/java/accord/messages/CheckStatus.java
index 27ab0ea2..24a32e0d 100644
--- a/accord-core/src/main/java/accord/messages/CheckStatus.java
+++ b/accord-core/src/main/java/accord/messages/CheckStatus.java
@@ -62,6 +62,8 @@ import static accord.primitives.Status.Durability.ShardUniversal;
import static accord.primitives.Status.Durability.Universal;
import accord.primitives.Known;
+import accord.utils.async.Cancellable;
+
import static accord.primitives.Status.NotDefined;
import static accord.primitives.Status.Stable;
import static accord.primitives.Status.Truncated;
@@ -70,7 +72,7 @@ import static accord.primitives.WithQuorum.HasQuorum;
import static accord.primitives.Route.castToRoute;
import static accord.primitives.Route.isRoute;
-public class CheckStatus extends AbstractEpochRequest<CheckStatus.CheckStatusReply>
+public class CheckStatus extends AbstractRequest<CheckStatus.CheckStatusReply>
implements Request, PreLoadContext, MapReduceConsume<SafeCommandStore, CheckStatus.CheckStatusReply>
{
public static class SerializationSupport
@@ -130,10 +132,10 @@ public class CheckStatus extends AbstractEpochRequest<CheckStatus.CheckStatusRep
}
@Override
- public void process()
+ public Cancellable submit()
{
// TODO (expected): only contact sourceEpoch
- node.mapReduceConsumeLocal(this, query, txnId.epoch(), sourceEpoch, this);
+ return node.mapReduceConsumeLocal(this, query, txnId.epoch(), sourceEpoch, this);
}
@Override
@@ -199,7 +201,7 @@ public class CheckStatus extends AbstractEpochRequest<CheckStatus.CheckStatusRep
}
@Override
- public void accept(CheckStatusReply ok, Throwable failure)
+ protected void acceptInternal(CheckStatusReply ok, Throwable failure)
{
if (failure != null) node.reply(replyTo, replyContext, ok, failure);
else if (ok == null) node.reply(replyTo, replyContext, CheckStatusNack.NotOwned, null);
diff --git a/accord-core/src/main/java/accord/messages/Commit.java b/accord-core/src/main/java/accord/messages/Commit.java
index f2b29a65..25ff08dd 100644
--- a/accord-core/src/main/java/accord/messages/Commit.java
+++ b/accord-core/src/main/java/accord/messages/Commit.java
@@ -48,6 +48,7 @@ import accord.topology.Topology;
import accord.utils.Invariants;
import accord.utils.SortedArrays.SortedArrayList;
import accord.utils.TriFunction;
+import accord.utils.async.Cancellable;
import org.agrona.collections.IntHashSet;
import static accord.primitives.SaveStatus.Committed;
@@ -157,54 +158,71 @@ public class Commit extends TxnRequest.WithUnsynced<CommitOrReadNack>
this.readData = readData;
}
- public static void commitMinimal(SortedArrayList<Id> contact, Node node, Topologies stabilise, Topologies all, Ballot ballot, TxnId txnId, Txn txn, FullRoute<?> route, Timestamp executeAt, Deps unstableDeps, Callback<ReadReply> callback)
+ public static void commitMinimalNoRead(SortedArrayList<Id> contact, Node node, Topologies stabilise, Topologies all, Ballot ballot, TxnId txnId, Txn txn, FullRoute<?> route, Timestamp executeAt, Deps unstableDeps, Callback<ReadReply> callback)
{
Invariants.checkArgument(stabilise.size() == 1, "Invalid coordinate epochs: %s", stabilise);
// we want to send to everyone, and we want to include all of the relevant data, but we stabilise on the coordination epoch replica responses
Topology coordinates = stabilise.forEpoch(txnId.epoch());
- send(contact, null, (i1, i2) -> true, null, node, coordinates, coordinates, all, Kind.CommitSlowPath, ballot,
- txnId, txn, route, executeAt, unstableDeps, callback);
+ sendTo(contact, null, (i1, i2) -> true, null, node, coordinates, coordinates, all, Kind.CommitSlowPath, ballot,
+ txnId, txn, route, executeAt, unstableDeps, callback, false);
}
// TODO (desired, efficiency): do not commit if we're already ready to execute (requires extra info in Accept responses)
- public static void stableAndRead(Node node, Topologies executeEpochOnly, Kind kind, TxnId txnId, Txn txn, FullRoute<?> route, Participants<?> readScope, Timestamp executeAt, Deps stableDeps, IntHashSet readSet, Callback<ReadReply> callback)
+ public static void stableAndRead(Node node, Topologies all, Kind kind, TxnId txnId, Txn txn, FullRoute<?> route, Participants<?> readScope, Timestamp executeAt, Deps stableDeps, IntHashSet readSet, Callback<ReadReply> callback, boolean onlyContactOldAndReadSet)
{
- Topologies all = executeEpochOnly;
- if (txnId.epoch() != executeAt.epoch())
- all = node.topology().preciseEpochs(route, txnId.epoch(), executeAt.epoch());
-
- Topology executes = executeEpochOnly.forEpoch(executeAt.epoch());
+ Invariants.checkState(all.oldestEpoch() == txnId.epoch());
+ Invariants.checkState(all.currentEpoch() == executeAt.epoch());
+ Topology executes = all.forEpoch(executeAt.epoch());
Topology coordinates = all.forEpoch(txnId.epoch());
SortedArrayList<Id> contact = all.nodes().without(all::isFaulty);
- send(contact, readSet, (set, id) -> set.contains(id.id), readScope, node, coordinates, executes, all, kind, Ballot.ZERO,
- txnId, txn, route, executeAt, stableDeps, callback);
+ sendTo(contact, readSet, (set, id) -> set.contains(id.id), readScope, node, coordinates, executes, all, kind, Ballot.ZERO,
+ txnId, txn, route, executeAt, stableDeps, callback, onlyContactOldAndReadSet);
}
- private static <P> void send(SortedArrayList<Id> contact, P param, BiPredicate<P, Id> shouldRegisterCallback, @Nullable Participants<?> readScopeIfCallback,
- Node node, Topology coordinates, Topology primary, Topologies all, Kind kind, Ballot ballot,
- TxnId txnId, @Nullable Txn txn, FullRoute<?> route, Timestamp executeAt, @Nullable Deps deps,
- Callback<ReadReply> callback)
+ public static void stableAndRead(Id to, Node node, Topologies all, Kind kind, TxnId txnId, Txn txn, FullRoute<?> route, Participants<?> readScope, Timestamp executeAt, Deps stableDeps, Callback<ReadReply> callback, boolean onlyContactOldAndReadSet)
+ {
+ Invariants.checkState(all.oldestEpoch() == txnId.epoch());
+ Invariants.checkState(all.currentEpoch() == executeAt.epoch());
+ Topology executes = all.forEpoch(executeAt.epoch());
+ Topology coordinates = all.forEpoch(txnId.epoch());
+
+ sendTo(to, true, readScope, node, coordinates, executes, all, kind, Ballot.ZERO, txnId, txn, route, executeAt, stableDeps, callback, false);
+ }
+
+ private static <P> void sendTo(SortedArrayList<Id> contact, P param, BiPredicate<P, Id> shouldRegisterCallback, @Nullable Participants<?> readScopeIfCallback,
+ Node node, Topology coordinates, Topology primary, Topologies all, Kind kind, Ballot ballot,
+ TxnId txnId, @Nullable Txn txn, FullRoute<?> route, Timestamp executeAt, @Nullable Deps deps,
+ Callback<ReadReply> callback, boolean onlyContactOldAndCallbacks)
{
for (Node.Id to : contact)
{
- if (all.size() == 1 || primary.contains(to))
+ boolean registerCallback = shouldRegisterCallback.test(param, to);
+ sendTo(to, registerCallback, readScopeIfCallback, node, coordinates, primary, all, kind, ballot, txnId, txn, route, executeAt, deps, callback, onlyContactOldAndCallbacks);
+ }
+ }
+
+ private static <P> void sendTo(Id to, boolean registerCallback, Participants<?> readScope,
+ Node node, Topology coordinates, Topology primary, Topologies all, Kind kind, Ballot ballot,
+ TxnId txnId, @Nullable Txn txn, FullRoute<?> route, Timestamp executeAt, @Nullable Deps deps,
+ Callback<ReadReply> callback, boolean onlyContactOldAndCallbacks)
+ {
+ if ((all.size() == 1 || primary.contains(to)))
+ {
+ if (registerCallback || onlyContactOldAndCallbacks)
{
- boolean registerCallback = shouldRegisterCallback.test(param, to);
// if we register a callback, supply the provided readScope (which may be null)
- Participants<?> readScope = registerCallback ? readScopeIfCallback : null;
Commit send = new Commit(kind, to, coordinates, all, txnId, txn, route, ballot, executeAt, deps, readScope);
if (registerCallback) node.send(to, send, callback);
else node.send(to, send);
}
- else
- {
- boolean registerCallback = shouldRegisterCallback.test(param, to);
- Commit send = new Commit(kind, to, coordinates, all, txnId, txn, route, ballot, executeAt, deps, (ReadTxnData) null);
- if (registerCallback) node.send(to, send, callback);
- else node.send(to, send);
- }
+ }
+ else
+ {
+ Commit send = new Commit(kind, to, coordinates, all, txnId, txn, route, ballot, executeAt, deps, (ReadTxnData) null);
+ if (registerCallback) node.send(to, send, callback);
+ else node.send(to, send);
}
}
@@ -231,13 +249,13 @@ public class Commit extends TxnRequest.WithUnsynced<CommitOrReadNack>
@Override
public KeyHistory keyHistory()
{
- return KeyHistory.COMMANDS;
+ return KeyHistory.ASYNC;
}
@Override
- public void process()
+ public Cancellable submit()
{
- node.mapReduceConsumeLocal(this, txnId.epoch(), executeAt.epoch(), this);
+ return node.mapReduceConsumeLocal(this, txnId.epoch(), executeAt.epoch(), this);
}
// TODO (expected, efficiency, clarity): do not guard with synchronized; let mapReduceLocal decide how to enforce mutual exclusivity
@@ -268,7 +286,7 @@ public class Commit extends TxnRequest.WithUnsynced<CommitOrReadNack>
}
@Override
- public synchronized void accept(CommitOrReadNack reply, Throwable failure)
+ protected void acceptInternal(CommitOrReadNack reply, Throwable failure)
{
if (reply != null || failure != null)
node.reply(replyTo, replyContext, reply, failure);
diff --git a/accord-core/src/main/java/accord/messages/GetEphemeralReadDeps.java b/accord-core/src/main/java/accord/messages/GetEphemeralReadDeps.java
index 354989cf..14f18dd3 100644
--- a/accord-core/src/main/java/accord/messages/GetEphemeralReadDeps.java
+++ b/accord-core/src/main/java/accord/messages/GetEphemeralReadDeps.java
@@ -32,6 +32,7 @@ import accord.primitives.TxnId;
import accord.primitives.Unseekables;
import accord.topology.Topologies;
import accord.utils.Invariants;
+import accord.utils.async.Cancellable;
import static accord.messages.PreAccept.calculateDeps;
import static accord.primitives.EpochSupplier.constant;
@@ -61,9 +62,9 @@ public class GetEphemeralReadDeps extends TxnRequest.WithUnsynced<GetEphemeralRe
}
@Override
- public void process()
+ public Cancellable submit()
{
- node.mapReduceConsumeLocal(this, minEpoch, executionEpoch, this);
+ return node.mapReduceConsumeLocal(this, minEpoch, executionEpoch, this);
}
@Override
@@ -81,12 +82,6 @@ public class GetEphemeralReadDeps extends TxnRequest.WithUnsynced<GetEphemeralRe
return new GetEphemeralReadDepsOk(reply1.deps.with(reply2.deps), Math.max(reply1.latestEpoch, reply2.latestEpoch));
}
- @Override
- public void accept(GetEphemeralReadDepsOk result, Throwable failure)
- {
- node.reply(replyTo, replyContext, result, failure);
- }
-
@Override
public MessageType type()
{
@@ -117,7 +112,7 @@ public class GetEphemeralReadDeps extends TxnRequest.WithUnsynced<GetEphemeralRe
@Override
public KeyHistory keyHistory()
{
- return KeyHistory.COMMANDS;
+ return KeyHistory.SYNC;
}
public static class GetEphemeralReadDepsOk implements Reply
diff --git a/accord-core/src/main/java/accord/messages/GetMaxConflict.java b/accord-core/src/main/java/accord/messages/GetMaxConflict.java
index d21ebdeb..3b39a67d 100644
--- a/accord-core/src/main/java/accord/messages/GetMaxConflict.java
+++ b/accord-core/src/main/java/accord/messages/GetMaxConflict.java
@@ -30,6 +30,7 @@ import accord.primitives.TxnId;
import accord.primitives.Unseekables;
import accord.topology.Topologies;
import accord.utils.Invariants;
+import accord.utils.async.Cancellable;
public class GetMaxConflict extends TxnRequest.WithUnsynced<GetMaxConflict.GetMaxConflictOk>
{
@@ -56,9 +57,9 @@ public class GetMaxConflict extends TxnRequest.WithUnsynced<GetMaxConflict.GetMa
}
@Override
- public void process()
+ public Cancellable submit()
{
- node.mapReduceConsumeLocal(this, minEpoch, executionEpoch, this);
+ return node.mapReduceConsumeLocal(this, minEpoch, executionEpoch, this);
}
@Override
@@ -74,12 +75,6 @@ public class GetMaxConflict extends TxnRequest.WithUnsynced<GetMaxConflict.GetMa
return new GetMaxConflictOk(Timestamp.max(reply1.maxConflict, reply2.maxConflict), Math.max(reply1.latestEpoch, reply2.latestEpoch));
}
- @Override
- public void accept(GetMaxConflictOk result, Throwable failure)
- {
- node.reply(replyTo, replyContext, result, failure);
- }
-
@Override
public MessageType type()
{
diff --git a/accord-core/src/main/java/accord/messages/InformDurable.java b/accord-core/src/main/java/accord/messages/InformDurable.java
index d0e2ecfb..061849a9 100644
--- a/accord-core/src/main/java/accord/messages/InformDurable.java
+++ b/accord-core/src/main/java/accord/messages/InformDurable.java
@@ -31,6 +31,7 @@ import accord.primitives.Route;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.topology.Topologies;
+import accord.utils.async.Cancellable;
import static accord.local.PreLoadContext.contextFor;
import static accord.messages.SimpleReply.Ok;
@@ -63,11 +64,11 @@ public class InformDurable extends TxnRequest<Reply> implements PreLoadContext
}
@Override
- public void process()
+ public Cancellable submit()
{
// TODO (expected, efficiency): do not load from disk to perform this update
// TODO (expected, consider): do we need to send this to all epochs in between, or just execution epoch?
- node.mapReduceConsumeLocal(contextFor(txnId), scope, txnId.epoch(), executeAt != null ? executeAt.epoch() : txnId.epoch(), this);
+ return node.mapReduceConsumeLocal(contextFor(txnId), scope, txnId.epoch(), executeAt != null ? executeAt.epoch() : txnId.epoch(), this);
}
@Override
@@ -88,12 +89,6 @@ public class InformDurable extends TxnRequest<Reply> implements PreLoadContext
return Ok;
}
- @Override
- public void accept(Reply reply, Throwable failure)
- {
- node.reply(replyTo, replyContext, reply, failure);
- }
-
@Override
public String toString()
{
diff --git a/accord-core/src/main/java/accord/messages/PreAccept.java b/accord-core/src/main/java/accord/messages/PreAccept.java
index bcb01261..8a5fbf59 100644
--- a/accord-core/src/main/java/accord/messages/PreAccept.java
+++ b/accord-core/src/main/java/accord/messages/PreAccept.java
@@ -46,6 +46,7 @@ import accord.primitives.Unseekables;
import accord.topology.Shard;
import accord.topology.Topologies;
import accord.utils.Invariants;
+import accord.utils.async.Cancellable;
public class PreAccept extends WithUnsynced<PreAccept.PreAcceptReply>
{
@@ -98,13 +99,13 @@ public class PreAccept extends WithUnsynced<PreAccept.PreAcceptReply>
@Override
public KeyHistory keyHistory()
{
- return KeyHistory.COMMANDS;
+ return KeyHistory.SYNC;
}
@Override
- protected void process()
+ protected Cancellable submit()
{
- node.mapReduceConsumeLocal(this, minEpoch, acceptEpoch, this);
+ return node.mapReduceConsumeLocal(this, minEpoch, acceptEpoch, this);
}
@Override
@@ -158,12 +159,6 @@ public class PreAccept extends WithUnsynced<PreAccept.PreAcceptReply>
return new PreAcceptOk(txnId, witnessedAt, deps);
}
- @Override
- public void accept(PreAcceptReply reply, Throwable failure)
- {
- node.reply(replyTo, replyContext, reply, failure);
- }
-
@Override
public MessageType type()
{
diff --git a/accord-core/src/main/java/accord/messages/Propagate.java b/accord-core/src/main/java/accord/messages/Propagate.java
index a50f7deb..f872d08b 100644
--- a/accord-core/src/main/java/accord/messages/Propagate.java
+++ b/accord-core/src/main/java/accord/messages/Propagate.java
@@ -22,7 +22,6 @@ import accord.api.RoutingKey;
import accord.coordinate.FetchData.FetchResult;
import accord.local.Command;
import accord.local.Commands;
-import accord.local.KeyHistory;
import accord.local.Node;
import accord.local.PreLoadContext;
import accord.local.RedundantStatus;
@@ -168,19 +167,6 @@ public class Propagate implements PreLoadContext, MapReduceConsume<SafeCommandSt
return txnId;
}
- @Override
- public Unseekables<?> keys()
- {
- // TODO (required): confirm this includes Route
- return propagateTo;
- }
-
- @Override
- public KeyHistory keyHistory()
- {
- return KeyHistory.COMMANDS;
- }
-
@Override
public Void apply(SafeCommandStore safeStore)
{
diff --git a/accord-core/src/main/java/accord/messages/ReadData.java b/accord-core/src/main/java/accord/messages/ReadData.java
index 0d4423de..5e1d4635 100644
--- a/accord-core/src/main/java/accord/messages/ReadData.java
+++ b/accord-core/src/main/java/accord/messages/ReadData.java
@@ -18,7 +18,6 @@
package accord.messages;
-import java.util.BitSet;
import javax.annotation.Nullable;
import org.slf4j.Logger;
@@ -26,11 +25,13 @@ import org.slf4j.LoggerFactory;
import accord.api.Data;
import accord.api.LocalListeners;
-import accord.api.RequestTimeouts;
-import accord.api.RequestTimeouts.RegisteredTimeout;
+import accord.api.LocalListeners.Registered;
+import accord.api.Timeouts;
+import accord.api.Timeouts.RegisteredTimeout;
import accord.local.Command;
import accord.local.CommandStore;
import accord.local.Node;
+import accord.local.PreLoadContext;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.primitives.SaveStatus;
@@ -42,8 +43,11 @@ import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.topology.Topologies;
import accord.utils.Invariants;
+import accord.utils.MapReduceConsume;
import accord.utils.async.AsyncChain;
+import accord.utils.async.Cancellable;
import org.agrona.collections.Int2ObjectHashMap;
+import org.agrona.collections.IntHashSet;
import static accord.api.ProgressLog.BlockedUntil.CanApply;
import static accord.api.ProgressLog.BlockedUntil.HasStableDeps;
@@ -53,9 +57,9 @@ import static accord.messages.ReadData.CommitOrReadNack.Redundant;
import static accord.messages.TxnRequest.latestRelevantEpochIndex;
import static accord.primitives.Routables.Slice.Minimal;
import static accord.utils.Invariants.illegalState;
-import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.MICROSECONDS;
-public abstract class ReadData extends AbstractEpochRequest<ReadData.CommitOrReadNack> implements LocalListeners.ComplexListener, RequestTimeouts.Timeout
+public abstract class ReadData implements PreLoadContext, Request, MapReduceConsume<SafeCommandStore, ReadData.CommitOrReadNack>, LocalListeners.ComplexListener, Timeouts.Timeout
{
private static final Logger logger = LoggerFactory.getLogger(ReadData.class);
@@ -106,20 +110,26 @@ public abstract class ReadData extends AbstractEpochRequest<ReadData.CommitOrRea
}
// TODO (desired, cleanup): should this be a Route?
+ public final TxnId txnId;
public final Participants<?> readScope;
public final long executeAtEpoch;
private transient State state = State.PENDING; // TODO (low priority, semantics): respond with the Executed result we have stored?
private Data data;
- transient BitSet waitingOn, reading;
+ transient IntHashSet waitingOn, reading;
transient int waitingOnCount;
transient Ranges unavailable;
- Int2ObjectHashMap<LocalListeners.Registered> registrations = new Int2ObjectHashMap<>();
+ Int2ObjectHashMap<LocalListeners.Registered> listeners = new Int2ObjectHashMap<>();
+ Cancellable cancel;
RegisteredTimeout timeout;
+ protected transient Node node;
+ protected transient Node.Id replyTo;
+ protected transient ReplyContext replyContext;
+
public ReadData(Node.Id to, Topologies topologies, TxnId txnId, Participants<?> readScope, long executeAtEpoch)
{
- super(txnId);
+ this.txnId = txnId;
int startIndex = latestRelevantEpochIndex(to, topologies, readScope);
this.readScope = TxnRequest.computeScope(to, topologies, readScope, startIndex, Participants::slice, Participants::with);
this.executeAtEpoch = executeAtEpoch;
@@ -127,7 +137,7 @@ public abstract class ReadData extends AbstractEpochRequest<ReadData.CommitOrRea
protected ReadData(TxnId txnId, Participants<?> readScope, long executeAtEpoch)
{
- super(txnId);
+ this.txnId = txnId;
this.readScope = readScope;
this.executeAtEpoch = executeAtEpoch;
}
@@ -148,11 +158,32 @@ public abstract class ReadData extends AbstractEpochRequest<ReadData.CommitOrRea
}
@Override
- protected void process()
- {
- waitingOn = new BitSet();
- reading = new BitSet();
- node.mapReduceConsumeLocal(this, readScope, executeAtEpoch, executeAtEpoch, this);
+ public final void process(Node on, Node.Id replyTo, ReplyContext replyContext)
+ {
+ this.node = on;
+ this.replyTo = replyTo;
+ this.replyContext = replyContext;
+ waitingOn = new IntHashSet();
+ reading = new IntHashSet();
+ Cancellable cancel = node.mapReduceConsumeLocal(this, readScope, executeAtEpoch, executeAtEpoch, this);
+ long expiresAt = node.agent().expiresAt(replyContext, MICROSECONDS);
+ RegisteredTimeout timeout = expiresAt <= 0 ? null : node.timeouts().registerWithDelay(this, expiresAt, MICROSECONDS);
+ synchronized (this)
+ {
+ switch (state)
+ {
+ case PENDING:
+ this.cancel = cancel;
+ this.timeout = timeout;
+ return;
+ case OBSOLETE:
+ timeout = null;
+ case RETURNED:
+ cancel = null;
+ }
+ }
+ if (cancel != null) cancel.cancel();
+ if (timeout != null) timeout.cancel();
}
@Override
@@ -163,7 +194,6 @@ public abstract class ReadData extends AbstractEpochRequest<ReadData.CommitOrRea
: r1.compareTo(r2) >= 0 ? r1 : r2;
}
-
@Override
public CommitOrReadNack apply(SafeCommandStore safeStore)
{
@@ -188,8 +218,8 @@ public abstract class ReadData extends AbstractEpochRequest<ReadData.CommitOrRea
{
default: throw new AssertionError();
case WAIT:
- registrations.put(storeId, safeStore.register(txnId, this));
- waitingOn.set(storeId);
+ listeners.put(storeId, safeStore.register(txnId, this));
+ waitingOn.add(storeId);
++waitingOnCount;
int c = status.compareTo(SaveStatus.Stable);
@@ -202,10 +232,10 @@ public abstract class ReadData extends AbstractEpochRequest<ReadData.CommitOrRea
return Redundant;
case EXECUTE:
- registrations.put(storeId, safeStore.register(txnId, this));
- waitingOn.set(storeId);
+ listeners.put(storeId, safeStore.register(txnId, this));
+ waitingOn.add(storeId);
++waitingOnCount;
- reading.set(storeId);
+ reading.add(storeId);
}
}
@@ -246,15 +276,11 @@ public abstract class ReadData extends AbstractEpochRequest<ReadData.CommitOrRea
break;
case EXECUTE:
- if (reading.get(storeId))
+ if (!reading.add(storeId))
return true;
- if (!waitingOn.get(storeId))
- {
- waitingOn.set(storeId);
+ if (waitingOn.add(storeId))
++waitingOnCount;
- }
- reading.set(storeId);
execute = true;
}
}
@@ -321,8 +347,8 @@ public abstract class ReadData extends AbstractEpochRequest<ReadData.CommitOrRea
protected void readComplete(CommandStore commandStore, @Nullable Data result, @Nullable Ranges unavailable)
{
- LocalListeners.Registered cancel;
- Runnable clear;
+ Registered cancelSelf;
+ Cancellable clear;
synchronized(this)
{
if (state != State.PENDING)
@@ -333,10 +359,10 @@ public abstract class ReadData extends AbstractEpochRequest<ReadData.CommitOrRea
data = data == null ? result : data.merge(result);
int storeId = commandStore.id();
- cancel = registrations.remove(storeId);
+ cancelSelf = listeners.remove(storeId);
clear = onOneSuccessInternal(storeId, unavailable, false);
}
- cancel.cancel();
+ cancelSelf.cancel();
cleanup(clear);
}
@@ -345,12 +371,12 @@ public abstract class ReadData extends AbstractEpochRequest<ReadData.CommitOrRea
cleanup(onOneSuccessInternal(storeId, newUnavailable, registration));
}
- protected synchronized Runnable onOneSuccessInternal(int storeId, @Nullable Ranges newUnavailable, boolean registration)
+ protected synchronized Cancellable onOneSuccessInternal(int storeId, @Nullable Ranges newUnavailable, boolean registration)
{
if (storeId >= 0)
{
- Invariants.checkState(waitingOn.get(storeId), "Txn %s's reading not contain store %d; waitingOn=%s", txnId, storeId, waitingOn);
- waitingOn.clear(storeId);
+ boolean removed = waitingOn.remove(storeId);
+ Invariants.checkState(removed, "Txn %s's reading not contain store %d; waitingOn=%s", txnId, storeId, waitingOn);
}
if (newUnavailable != null && !newUnavailable.isEmpty())
@@ -364,15 +390,7 @@ public abstract class ReadData extends AbstractEpochRequest<ReadData.CommitOrRea
// and prevents races where we respond before dispatching all the required reads (if the reads are
// completing faster than the reads can be setup on all required shards)
if (--waitingOnCount >= 0)
- {
- if (registration)
- {
- // Time out reads to avoid doing extra work for requests that will have responses ignored
- long replyTimeout = node.agent().replyTimeout(replyContext, MILLISECONDS);
- timeout = node.requestTimeouts().register(this, replyTimeout, MILLISECONDS);
- }
return null;
- }
switch (state)
{
@@ -393,9 +411,11 @@ public abstract class ReadData extends AbstractEpochRequest<ReadData.CommitOrRea
}
}
+ protected void submitted() {}
+
boolean cancel()
{
- Runnable clear;
+ Cancellable clear;
synchronized (this)
{
if (state != State.PENDING)
@@ -408,45 +428,35 @@ public abstract class ReadData extends AbstractEpochRequest<ReadData.CommitOrRea
return true;
}
- void cleanup()
- {
- Runnable clear;
- synchronized (this)
- {
- clear = clearUnsafe();
- }
- cleanup(clear);
- }
-
- private static void cleanup(@Nullable Runnable clear)
- {
- if (clear != null)
- clear.run();
- }
-
- @Nullable Runnable clearUnsafe()
+ @Nullable Cancellable clearUnsafe()
{
Invariants.checkState(state != State.PENDING);
RegisteredTimeout cancelTimeout = timeout;
- Int2ObjectHashMap<LocalListeners.Registered> cancelRegistrations = registrations;
+ Int2ObjectHashMap<LocalListeners.Registered> cancelListeners = listeners;
timeout = null;
- registrations = null;
+ listeners = null;
waitingOn.clear();
reading.clear();
data = null;
unavailable = null;
- if (cancelRegistrations == null && cancelTimeout == null)
+ if (cancelListeners == null && cancelTimeout == null)
return null;
return () -> {
if (cancelTimeout != null)
cancelTimeout.cancel();
- if (cancelRegistrations != null)
- cancelRegistrations.forEach((i, r) -> r.cancel());
+ if (cancelListeners != null)
+ cancelListeners.forEach((i, r) -> r.cancel());
};
}
+ private static void cleanup(@Nullable Cancellable cancel)
+ {
+ if (cancel != null)
+ cancel.cancel();
+ }
+
public void timeout()
{
timeout = null;
diff --git a/accord-core/src/main/java/accord/messages/ReadEphemeralTxnData.java b/accord-core/src/main/java/accord/messages/ReadEphemeralTxnData.java
index d70d6f7c..e86a1805 100644
--- a/accord-core/src/main/java/accord/messages/ReadEphemeralTxnData.java
+++ b/accord-core/src/main/java/accord/messages/ReadEphemeralTxnData.java
@@ -23,6 +23,7 @@ import accord.api.Data;
import accord.local.Command;
import accord.local.Commands;
import accord.local.Node.Id;
+import accord.local.PreLoadContext;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.local.StoreParticipants;
@@ -38,7 +39,10 @@ import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.topology.Topologies;
+import accord.utils.Invariants;
+import org.agrona.collections.IntHashSet;
+import static accord.local.Commands.eraseEphemeralRead;
import static accord.messages.TxnRequest.computeScope;
import static accord.messages.TxnRequest.latestRelevantEpochIndex;
@@ -54,9 +58,9 @@ public class ReadEphemeralTxnData extends ReadData
private static final ExecuteOn EXECUTE_ON = new ExecuteOn(SaveStatus.ReadyToExecute, SaveStatus.Applied);
- public final @Nonnull PartialTxn partialTxn;
- public final @Nonnull PartialDeps partialDeps;
- public final @Nonnull FullRoute<?> route; // TODO (desired): should be unnecessary, only included to not breach Stable command validations
+ private PartialTxn partialTxn;
+ private PartialDeps partialDeps;
+ private FullRoute<?> route; // TODO (desired): should be unnecessary, only included to not breach Stable command validations
public ReadEphemeralTxnData(Id to, Topologies topologies, TxnId txnId, Participants<?> readScope, long executeAtEpoch, @Nonnull Txn txn, @Nonnull Deps deps, @Nonnull FullRoute<?> route)
{
@@ -71,6 +75,7 @@ public class ReadEphemeralTxnData extends ReadData
private ReadEphemeralTxnData(TxnId txnId, Participants<?> readScope, Route<?> scope, long executeAtEpoch, @Nonnull Txn txn, @Nonnull Deps deps, @Nonnull FullRoute<?> route)
{
super(txnId, readScope.intersecting(scope), executeAtEpoch);
+ Invariants.checkState(executeAtEpoch == txnId.epoch());
this.route = route;
this.partialTxn = txn.intersecting(scope, false);
this.partialDeps = deps.intersecting(scope);
@@ -79,6 +84,7 @@ public class ReadEphemeralTxnData extends ReadData
public ReadEphemeralTxnData(TxnId txnId, Participants<?> readScope, long executeAtEpoch, @Nonnull PartialTxn partialTxn, @Nonnull PartialDeps partialDeps, @Nonnull FullRoute<?> route)
{
super(txnId, readScope, executeAtEpoch);
+ Invariants.checkState(executeAtEpoch == txnId.epoch());
this.partialTxn = partialTxn;
this.partialDeps = partialDeps;
this.route = route;
@@ -87,10 +93,34 @@ public class ReadEphemeralTxnData extends ReadData
@Override
protected synchronized CommitOrReadNack apply(SafeCommandStore safeStore, SafeCommand safeCommand, StoreParticipants participants)
{
- Commands.ephemeralRead(safeStore, safeCommand, route, txnId, partialTxn, partialDeps, executeAtEpoch);
+ Commands.ephemeralRead(safeStore, safeCommand, route, txnId, partialTxn, partialDeps);
return super.apply(safeStore, safeCommand, participants);
}
+ public final PartialTxn partialTxn()
+ {
+ return partialTxn;
+ }
+
+ public final PartialDeps partialDeps()
+ {
+ return partialDeps;
+ }
+
+ public final FullRoute<?> route()
+ {
+ return route;
+ }
+
+ @Override
+ public void accept(CommitOrReadNack reply, Throwable failure)
+ {
+ super.accept(reply, failure);
+ partialTxn = null;
+ partialDeps = null;
+ route = null;
+ }
+
@Override
protected ExecuteOn executeOn()
{
@@ -137,6 +167,21 @@ public class ReadEphemeralTxnData extends ReadData
return 0;
}
+ @Override
+ public void timeout()
+ {
+ synchronized (this)
+ {
+ IntHashSet.IntIterator iter = waitingOn.iterator();
+ while (iter.hasNext())
+ {
+ node.commandStores().forId(iter.nextValue())
+ .execute(PreLoadContext.empty(), safeStore -> eraseEphemeralRead(safeStore, txnId))
+ .begin(node.agent());
+ }
+ }
+ super.timeout();
+ }
@Override
protected ReadOk constructReadOk(Ranges unavailable, Data data)
diff --git a/accord-core/src/main/java/accord/messages/ReadTxnData.java b/accord-core/src/main/java/accord/messages/ReadTxnData.java
index 59efd77f..421f89b4 100644
--- a/accord-core/src/main/java/accord/messages/ReadTxnData.java
+++ b/accord-core/src/main/java/accord/messages/ReadTxnData.java
@@ -25,7 +25,6 @@ import accord.topology.Topologies;
import static accord.primitives.SaveStatus.ReadyToExecute;
-// TODO (required, efficiency): dedup - can currently have infinite pending reads that will be executed independently
public class ReadTxnData extends ReadData
{
public static class SerializerSupport
diff --git a/accord-core/src/main/java/accord/messages/Request.java b/accord-core/src/main/java/accord/messages/Request.java
index 74d2cae4..9b319f58 100644
--- a/accord-core/src/main/java/accord/messages/Request.java
+++ b/accord-core/src/main/java/accord/messages/Request.java
@@ -25,5 +25,4 @@ public interface Request extends Message
{
default long waitForEpoch() { return 0; }
void process(Node on, Id from, ReplyContext replyContext);
-
}
diff --git a/accord-core/src/main/java/accord/messages/SetShardDurable.java b/accord-core/src/main/java/accord/messages/SetShardDurable.java
index 7a5d6e1c..4d07788b 100644
--- a/accord-core/src/main/java/accord/messages/SetShardDurable.java
+++ b/accord-core/src/main/java/accord/messages/SetShardDurable.java
@@ -23,10 +23,11 @@ import accord.local.SafeCommandStore;
import accord.primitives.AbstractRanges;
import accord.primitives.SyncPoint;
import accord.utils.MapReduceConsume;
+import accord.utils.async.Cancellable;
import static accord.messages.SimpleReply.Ok;
-public class SetShardDurable extends AbstractEpochRequest<SimpleReply>
+public class SetShardDurable extends AbstractRequest<SimpleReply>
implements Request, PreLoadContext, MapReduceConsume<SafeCommandStore, SimpleReply>
{
public final SyncPoint exclusiveSyncPoint;
@@ -38,13 +39,14 @@ public class SetShardDurable extends AbstractEpochRequest<SimpleReply>
}
@Override
- public void process()
+ public Cancellable submit()
{
node.markDurable(exclusiveSyncPoint.route.toRanges(), exclusiveSyncPoint.syncId, exclusiveSyncPoint.syncId)
.addCallback((success, fail) -> {
if (fail != null) node.reply(replyTo, replyContext, null, fail);
else node.mapReduceConsumeLocal(this, exclusiveSyncPoint.route, waitForEpoch(), waitForEpoch(), this);
});
+ return null;
}
@Override
@@ -61,7 +63,7 @@ public class SetShardDurable extends AbstractEpochRequest<SimpleReply>
}
@Override
- public void accept(SimpleReply ok, Throwable failure)
+ protected void acceptInternal(SimpleReply ok, Throwable failure)
{
}
diff --git a/accord-core/src/main/java/accord/messages/TxnRequest.java b/accord-core/src/main/java/accord/messages/TxnRequest.java
index e0fe7b24..363d3500 100644
--- a/accord-core/src/main/java/accord/messages/TxnRequest.java
+++ b/accord-core/src/main/java/accord/messages/TxnRequest.java
@@ -20,6 +20,8 @@ package accord.messages;
import java.util.function.BiFunction;
+import javax.annotation.Nullable;
+
import accord.local.Node;
import accord.local.Node.Id;
import accord.local.PreLoadContext;
@@ -35,12 +37,13 @@ import accord.topology.Topologies;
import accord.topology.Topology;
import accord.utils.Invariants;
import accord.utils.MapReduceConsume;
+import accord.utils.async.Cancellable;
import static accord.utils.Invariants.illegalArgument;
-public abstract class TxnRequest<R> implements Request, PreLoadContext, MapReduceConsume<SafeCommandStore, R>
+public abstract class TxnRequest<R extends Reply> extends AbstractRequest<R> implements Request, PreLoadContext, MapReduceConsume<SafeCommandStore, R>
{
- public static abstract class WithUnsynced<R> extends TxnRequest<R>
+ public static abstract class WithUnsynced<R extends Reply> extends TxnRequest<R>
{
public final long minEpoch; // TODO (low priority, clarity): can this just always be TxnId.epoch?
@@ -67,13 +70,8 @@ public abstract class TxnRequest<R> implements Request, PreLoadContext, MapReduc
}
}
- public final TxnId txnId;
public final Route<?> scope;
public final long waitForEpoch;
- // set on receive only
- protected transient Node node;
- protected transient Id replyTo;
- protected transient ReplyContext replyContext;
public ReplyContext replyContext()
{
@@ -92,8 +90,8 @@ public abstract class TxnRequest<R> implements Request, PreLoadContext, MapReduc
public TxnRequest(TxnId txnId, Route<?> scope, long waitForEpoch)
{
+ super(txnId);
Invariants.checkState(!scope.isEmpty());
- this.txnId = txnId;
this.scope = scope;
this.waitForEpoch = waitForEpoch;
}
@@ -118,16 +116,7 @@ public abstract class TxnRequest<R> implements Request, PreLoadContext, MapReduc
return waitForEpoch;
}
- @Override
- public void process(Node on, Id replyTo, ReplyContext replyContext)
- {
- this.node = on;
- this.replyTo = replyTo;
- this.replyContext = replyContext;
- process();
- }
-
- protected abstract void process();
+ protected abstract @Nullable Cancellable submit();
// finds the first topology index that intersects with the node
protected static int latestRelevantEpochIndex(Node.Id node, Topologies topologies, Routables<?> route)
diff --git a/accord-core/src/main/java/accord/primitives/Keys.java b/accord-core/src/main/java/accord/primitives/Keys.java
index 5d8bf130..abfd62b5 100644
--- a/accord-core/src/main/java/accord/primitives/Keys.java
+++ b/accord-core/src/main/java/accord/primitives/Keys.java
@@ -180,12 +180,12 @@ public class Keys extends AbstractKeys<Key> implements Seekables<Key, Keys>
public static Keys of(Key ... keys)
{
- return dedupSorted(sort(keys));
+ return ofSorted(sort(keys));
}
public static Keys copyOf(Key[] keys)
{
- return dedupSorted(sort(keys.clone()));
+ return ofSorted(sort(keys.clone()));
}
public static Keys ofUnique(Key ... keys)
@@ -265,6 +265,11 @@ public class Keys extends AbstractKeys<Key> implements Seekables<Key, Keys>
return ofUnique(keys.toArray(new Key[0]));
}
+ public static Keys ofSorted(Key ... keys)
+ {
+ return new Keys(SortedArrays.toUnique(keys));
+ }
+
public static Keys ofSortedUnique(Key ... keys)
{
if (!isSortedUnique(keys))
@@ -272,11 +277,6 @@ public class Keys extends AbstractKeys<Key> implements Seekables<Key, Keys>
return new Keys(keys);
}
- private static Keys dedupSorted(Key ... keys)
- {
- return new Keys(SortedArrays.toUnique(keys));
- }
-
public static Keys ofSortedUnique(Collection<? extends Key> keys)
{
return ofSortedUnique(keys.toArray(new Key[0]));
diff --git a/accord-core/src/main/java/accord/primitives/Participants.java b/accord-core/src/main/java/accord/primitives/Participants.java
index 4291ce71..84873470 100644
--- a/accord-core/src/main/java/accord/primitives/Participants.java
+++ b/accord-core/src/main/java/accord/primitives/Participants.java
@@ -45,6 +45,7 @@ public interface Participants<K extends Unseekable> extends Unseekables<K>
*/
static <K extends Unseekable> Participants<K> merge(Participants<K> left, Participants<K> right)
{
+ if (left == right) return left;
if (left == null) return right;
if (right == null) return left;
return left.with(right);
diff --git a/accord-core/src/main/java/accord/primitives/Route.java b/accord-core/src/main/java/accord/primitives/Route.java
index 856e694a..63d85808 100644
--- a/accord-core/src/main/java/accord/primitives/Route.java
+++ b/accord-core/src/main/java/accord/primitives/Route.java
@@ -169,6 +169,7 @@ public interface Route<K extends Unseekable> extends Participants<K>
static <T extends Unseekable> Route<T> merge(@Nullable Route<T> prefer, @Nullable Route<T> defer)
{
+ if (prefer == defer) return prefer;
if (defer == null) return prefer;
if (prefer == null) return defer;
return prefer.with(defer);
diff --git a/accord-core/src/main/java/accord/primitives/RoutingKeys.java b/accord-core/src/main/java/accord/primitives/RoutingKeys.java
index 32682f9a..c9f67cac 100644
--- a/accord-core/src/main/java/accord/primitives/RoutingKeys.java
+++ b/accord-core/src/main/java/accord/primitives/RoutingKeys.java
@@ -59,6 +59,13 @@ public class RoutingKeys extends AbstractUnseekableKeys implements Unseekables<R
return of(keys.toArray(EMPTY_KEYS_ARRAY));
}
+ public static RoutingKeys ofSortedUnique(Collection<? extends RoutingKey> keys)
+ {
+ if (keys.isEmpty())
+ return EMPTY;
+ return ofSortedUnique(keys.toArray(EMPTY_KEYS_ARRAY));
+ }
+
public static RoutingKeys ofSortedUnique(RoutingKey ... keys)
{
checkArgument(isSortedUnique(keys));
diff --git a/accord-core/src/main/java/accord/utils/ArrayBuffers.java b/accord-core/src/main/java/accord/utils/ArrayBuffers.java
index 455db381..33c6a3d6 100644
--- a/accord-core/src/main/java/accord/utils/ArrayBuffers.java
+++ b/accord-core/src/main/java/accord/utils/ArrayBuffers.java
@@ -27,6 +27,7 @@ import java.io.Closeable;
import java.lang.reflect.Array;
import java.util.AbstractList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.function.IntFunction;
import static accord.utils.Invariants.illegalState;
@@ -828,6 +829,17 @@ public class ArrayBuffers
return true;
}
+ @Override
+ public boolean addAll(Collection<? extends E> es)
+ {
+ int newSize = size + es.size();
+ if (newSize >= buffer.length)
+ buffer = cachedAny().resize(buffer, size, newSize);
+ for (E e : es)
+ buffer[size++] = e;
+ return true;
+ }
+
public void close()
{
if (buffer == null) return;
diff --git a/accord-core/src/main/java/accord/utils/IntrusiveLinkedList.java b/accord-core/src/main/java/accord/utils/IntrusiveLinkedList.java
index b9cefb40..f223fbea 100644
--- a/accord-core/src/main/java/accord/utils/IntrusiveLinkedList.java
+++ b/accord-core/src/main/java/accord/utils/IntrusiveLinkedList.java
@@ -64,11 +64,18 @@ public class IntrusiveLinkedList<O extends IntrusiveLinkedListNode> extends Intr
public O poll()
{
- if (isEmpty())
+ IntrusiveLinkedListNode next = this.next;
+ if (next == this)
return null;
+ next.remove();
+ return (O) next;
+ }
+ public O peek()
+ {
IntrusiveLinkedListNode next = this.next;
- next.remove();
+ if (next == this)
+ return null;
return (O) next;
}
@@ -80,7 +87,7 @@ public class IntrusiveLinkedList<O extends IntrusiveLinkedListNode> extends Intr
@Override
public Iterator<O> iterator()
{
- return new Iterator<O>()
+ return new Iterator<>()
{
IntrusiveLinkedListNode next = IntrusiveLinkedList.this.next;
diff --git a/accord-core/src/main/java/accord/utils/IntrusivePriorityHeap.java b/accord-core/src/main/java/accord/utils/IntrusivePriorityHeap.java
index 63c867e0..715f82b3 100644
--- a/accord-core/src/main/java/accord/utils/IntrusivePriorityHeap.java
+++ b/accord-core/src/main/java/accord/utils/IntrusivePriorityHeap.java
@@ -57,6 +57,7 @@ public abstract class IntrusivePriorityHeap<N extends IntrusivePriorityHeap.Node
*/
protected void append(N node)
{
+ Invariants.checkState(node.heapIndex < 0);
if (size == heap.length)
heap = Arrays.copyOf(heap, Math.max(size * 2, 8));
@@ -78,13 +79,19 @@ public abstract class IntrusivePriorityHeap<N extends IntrusivePriorityHeap.Node
else siftUp(node, index);
}
+ protected boolean contains(N node)
+ {
+ int i = node.heapIndex;
+ return i >= 0 && i < size && heap[i] == node;
+ }
+
/**
* remove; can be used as a simple list
*/
protected void remove(N node)
{
int i = node.heapIndex;
- Invariants.checkArgument(heap[i] == node);
+ Invariants.checkArgument(i >= 0 && i < heap.length && heap[i] == node);
if (size > 1)
{
N tail = (N) heap[--size];
@@ -137,12 +144,15 @@ public abstract class IntrusivePriorityHeap<N extends IntrusivePriorityHeap.Node
else siftDown(with, i);
}
- private void replaceHead()
+ protected void replaceHead()
{
--size;
--heapifiedSize;
if (size == 0)
+ {
+ heap[0] = null;
return;
+ }
N siftDown = (N) heap[size];
heap[size] = null;
@@ -152,7 +162,7 @@ public abstract class IntrusivePriorityHeap<N extends IntrusivePriorityHeap.Node
/**
* {@code i} is a free position in the heap, siftDown must be safely inserted at a position >= i
*/
- private void siftDown(N siftDown, int i)
+ protected void siftDown(N siftDown, int i)
{
while (true)
{
diff --git a/accord-core/src/main/java/accord/utils/Invariants.java b/accord-core/src/main/java/accord/utils/Invariants.java
index b9aa1ce8..d9a46e9d 100644
--- a/accord-core/src/main/java/accord/utils/Invariants.java
+++ b/accord-core/src/main/java/accord/utils/Invariants.java
@@ -18,6 +18,7 @@
package accord.utils;
+import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.Supplier;
import javax.annotation.Nullable;
@@ -87,6 +88,11 @@ public class Invariants
throw new IllegalArgumentException(msg);
}
+ public static IllegalArgumentException illegalArgument(String fmt, Object ... args)
+ {
+ throw new IllegalArgumentException(String.format(fmt, args));
+ }
+
private static IllegalArgumentException illegalArgument()
{
throw illegalArgument(null);
@@ -165,12 +171,37 @@ public class Invariants
throw illegalState(format(fmt, p1));
}
+ public static <P> void checkState(boolean condition, String fmt, @Nullable P p1, Function<? super P, ?> transformP)
+ {
+ if (!condition)
+ throw illegalState(format(fmt, transformP.apply(p1)));
+ }
+
public static void checkState(boolean condition, String fmt, @Nullable Object p1, @Nullable Object p2)
{
if (!condition)
throw illegalState(format(fmt, p1, p2));
}
+ public static <P> void checkState(boolean condition, String fmt, @Nullable Object p1, @Nullable P p2, Function<? super P, ?> transformP2)
+ {
+ if (!condition)
+ throw illegalState(format(fmt, p1, transformP2.apply(p2)));
+ }
+
+
+ public static void checkState(boolean condition, String fmt, @Nullable Object p1, @Nullable Object p2, @Nullable Object p3)
+ {
+ if (!condition)
+ throw illegalState(format(fmt, p1, p2, p3));
+ }
+
+ public static <P> void checkState(boolean condition, String fmt, @Nullable Object p1, @Nullable Object p2, @Nullable P p3, Function<? super P, Object> transformP3)
+ {
+ if (!condition)
+ throw illegalState(format(fmt, p1, p2, transformP3.apply(p3)));
+ }
+
public static void checkState(boolean condition, String fmt, Object... args)
{
if (!condition)
diff --git a/accord-core/src/main/java/accord/utils/LogGroupTimers.java b/accord-core/src/main/java/accord/utils/LogGroupTimers.java
index 9d83a87d..7bb75be2 100644
--- a/accord-core/src/main/java/accord/utils/LogGroupTimers.java
+++ b/accord-core/src/main/java/accord/utils/LogGroupTimers.java
@@ -103,6 +103,7 @@ public class LogGroupTimers<T extends LogGroupTimers.Timer>
if (contains(deadline))
return false;
+ timer.heapIndex = -1;
owner.addInternal(deadline, timer);
return true;
}
diff --git a/accord-core/src/main/java/accord/utils/MergeFewDisjointSortedListsCursor.java b/accord-core/src/main/java/accord/utils/MergeFewDisjointSortedListsCursor.java
index 91b5a77e..5a045c4f 100644
--- a/accord-core/src/main/java/accord/utils/MergeFewDisjointSortedListsCursor.java
+++ b/accord-core/src/main/java/accord/utils/MergeFewDisjointSortedListsCursor.java
@@ -17,7 +17,6 @@
*/
package accord.utils;
-import java.util.Arrays;
import javax.annotation.Nonnull;
public class MergeFewDisjointSortedListsCursor<T extends Comparable<? super T>> implements SortedCursor<T>
@@ -46,17 +45,30 @@ public class MergeFewDisjointSortedListsCursor<T extends Comparable<? super T>>
return this;
}
- private boolean find(Comparable<? super T> find)
+ private static final int FOUND = 1;
+ private static final int ADVANCED = 2;
+
+ private int find(Comparable<? super T> find)
{
- boolean found;
+ int result = 0;
int i = list.findNext(itemIdx, find);
- found = i >= 0;
+ if (i >= 0) result = FOUND;
if (i < 0) i = -1 - i;
- if ((itemIdx = i) < list.size())
- item = list.get(itemIdx = i);
- else
- item = null;
- return found;
+ if (i > itemIdx)
+ {
+ result |= ADVANCED;
+ if ((itemIdx = i) < list.size())
+ item = list.get(itemIdx = i);
+ else
+ item = null;
+ }
+ return result;
+ }
+
+ @Override
+ public String toString()
+ {
+ return list.toString();
}
public int compareTo(Candidate<T> that)
@@ -65,25 +77,16 @@ public class MergeFewDisjointSortedListsCursor<T extends Comparable<? super T>>
}
}
- protected final Candidate<T>[] heap;
-
- /** Number of non-exhausted iterators. */
- int size;
-
+ Candidate<T>[] heap;
+ int size = 0;
public MergeFewDisjointSortedListsCursor(int capacity)
{
- this.heap = (Candidate<T>[]) new Candidate[capacity];
+ heap = new Candidate[capacity];
}
public void add(SortedList<? extends T> list)
{
- Candidate<T> candidate = new Candidate<>(list);
- heap[size++] = candidate;
- }
-
- public void init()
- {
- Arrays.sort(heap, 0, size);
+ heap[size++] = new Candidate<>(list);
}
@Override
@@ -101,56 +104,74 @@ public class MergeFewDisjointSortedListsCursor<T extends Comparable<? super T>>
public void advance()
{
Candidate<T> sink = heap[0].advance();
- if (sink == null)
- {
- heap[0] = null;
- if (--size > 0)
- System.arraycopy(heap, 1, heap, 0, size);
- }
- else
- {
- int i = 1;
- for (; i < size && sink.compareTo(heap[i]) > 0; ++i)
- heap[i - 1] = heap[i];
- heap[i - 1] = sink;
- }
+ if (sink == null) replaceHead();
+ else siftDown(heap[0], 0);
+ }
+
+ public void init()
+ {
+ for (int i = size - 2 ; i >= 0 ; --i)
+ siftDown(heap[i], i);
}
@Override
public boolean find(Comparable<? super T> find)
{
- int i = 0;
- int removedCount = 0;
- boolean found = false;
- while (i < size)
+ if (size == 0)
+ return false;
+
+ Candidate<T> found = null;
+ while (true)
{
- int c = find.compareTo(heap[i].item);
- if (c <= 0)
+ Candidate<T> head = heap[0];
+ if (head == found)
+ return true;
+
+ int result = head.find(find);
+ if (0 != (result & Candidate.FOUND))
{
- found |= c == 0;
- if (removedCount > 0)
- System.arraycopy(heap, removedCount, heap, 0, size - removedCount);
- break;
+ Invariants.checkState(found == null, "%s and %s both contained %s", head, found, find);
+ found = head;
}
- Candidate<T> candidate = heap[i];
- found |= candidate.find(find);
- if (candidate.item == null)
+ if (0 == (result & Candidate.ADVANCED))
+ return found != null;
+
+ if (head.item == null)
{
- Invariants.checkState(candidate.itemIdx == candidate.list.size());
- ++removedCount;
+ Invariants.checkState(head != found);
+ replaceHead();
+ if (size == 0)
+ return false;
}
- else if (removedCount > 0)
+ else
{
- heap[i - removedCount] = candidate;
+ siftDown(head, 0);
}
- i++;
+ }
+ }
+
+ void replaceHead()
+ {
+ for (int i = 1 ; i < size ; ++i)
+ heap[i - 1] = heap[i];
+ heap[--size] = null;
+ }
+
+ private void siftDown(Candidate<T> node, int i)
+ {
+ int j = i;
+ while (++j < size)
+ {
+ if (node.compareTo(heap[j]) < 0)
+ break;
}
- while (removedCount-- > 0)
- heap[--size] = null;
+ if (--j == i)
+ return;
- Arrays.sort(heap, 0, size);
- return found;
+ for (; i < j ; ++i)
+ heap[i] = heap[i + 1];
+ heap[i] = node;
}
}
\ No newline at end of file
diff --git a/accord-core/src/main/java/accord/messages/Request.java b/accord-core/src/main/java/accord/utils/QuadConsumer.java
similarity index 77%
copy from accord-core/src/main/java/accord/messages/Request.java
copy to accord-core/src/main/java/accord/utils/QuadConsumer.java
index 74d2cae4..66c11155 100644
--- a/accord-core/src/main/java/accord/messages/Request.java
+++ b/accord-core/src/main/java/accord/utils/QuadConsumer.java
@@ -15,15 +15,9 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package accord.utils;
-package accord.messages;
-
-import accord.local.Node;
-import accord.local.Node.Id;
-
-public interface Request extends Message
+public interface QuadConsumer<P1, P2, P3, P4>
{
- default long waitForEpoch() { return 0; }
- void process(Node on, Id from, ReplyContext replyContext);
-
+ void accept(P1 p1, P2 p2, P3 p3, P4 p4);
}
diff --git a/accord-core/src/main/java/accord/messages/Request.java b/accord-core/src/main/java/accord/utils/QuintConsumer.java
similarity index 77%
copy from accord-core/src/main/java/accord/messages/Request.java
copy to accord-core/src/main/java/accord/utils/QuintConsumer.java
index 74d2cae4..3296b8a1 100644
--- a/accord-core/src/main/java/accord/messages/Request.java
+++ b/accord-core/src/main/java/accord/utils/QuintConsumer.java
@@ -15,15 +15,9 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package accord.utils;
-package accord.messages;
-
-import accord.local.Node;
-import accord.local.Node.Id;
-
-public interface Request extends Message
+public interface QuintConsumer<P1, P2, P3, P4, P5>
{
- default long waitForEpoch() { return 0; }
- void process(Node on, Id from, ReplyContext replyContext);
-
+ void accept(P1 p1, P2 p2, P3 p3, P4 p4, P5 p5);
}
diff --git a/accord-core/src/main/java/accord/utils/RelationMultiMap.java b/accord-core/src/main/java/accord/utils/RelationMultiMap.java
index feb42a0a..4fe7dd83 100644
--- a/accord-core/src/main/java/accord/utils/RelationMultiMap.java
+++ b/accord-core/src/main/java/accord/utils/RelationMultiMap.java
@@ -450,30 +450,35 @@ public class RelationMultiMap
int idIdx = i + startIndex;
int valueIdx = ids[idIdx];
valueIdx = SortedArrays.exponentialSearch(values, valueIdx, values.length, find);
- if (valueIdx < 0)
- valueIdx = -1 -valueIdx;
- return normalise(SortedArrays.exponentialSearch(ids, idIdx, endIndex, valueIdx));
+ boolean found = valueIdx >= 0;
+ if (!found) valueIdx = -1 -valueIdx;
+ return normalise(found, SortedArrays.exponentialSearch(ids, idIdx, endIndex, valueIdx));
}
public int findNext(int i, int valueIdx)
{
int idIdx = i + startIndex;
- return normalise(SortedArrays.exponentialSearch(ids, idIdx, endIndex, valueIdx));
+ return normalise(true, SortedArrays.exponentialSearch(ids, idIdx, endIndex, valueIdx));
}
@Override
public int find(Comparable<? super T> find)
{
int valueIdx = Arrays.binarySearch(values, 0, values.length, find);
- if (valueIdx < 0)
- valueIdx = -1 -valueIdx;
- // TODO (desired): use interpolation search
- return normalise(SortedArrays.exponentialSearch(ids, startIndex, endIndex, valueIdx));
+ boolean found = valueIdx >= 0;
+ if (!found) valueIdx = -1 -valueIdx;
+ // TODO (desired): use interpolation search (or binarySearch)
+ return normalise(found, Arrays.binarySearch(ids, startIndex, endIndex, valueIdx));
}
- private int normalise(int searchResult)
+ private int normalise(boolean foundValue, int searchResult)
{
- return searchResult >= 0 ? searchResult - startIndex : searchResult + startIndex;
+ if (searchResult < 0)
+ return searchResult + startIndex;
+
+ searchResult -= startIndex;
+ if (!foundValue) searchResult = -1 - searchResult;
+ return searchResult;
}
}
diff --git a/accord-core/src/main/java/accord/utils/SimpleBitSet.java b/accord-core/src/main/java/accord/utils/SimpleBitSet.java
index db61c1cc..4adad7ae 100644
--- a/accord-core/src/main/java/accord/utils/SimpleBitSet.java
+++ b/accord-core/src/main/java/accord/utils/SimpleBitSet.java
@@ -66,7 +66,7 @@ public class SimpleBitSet
public SimpleBitSet(SimpleBitSet copy, boolean share)
{
- bits = copy.bits.clone();
+ bits = share ? copy.bits : copy.bits.clone();
count = copy.count;
}
@@ -138,11 +138,9 @@ public class SimpleBitSet
{
int count = Long.bitCount(bits[fromIndex] & (-1L << (from & 63)));
for (int i = fromIndex + 1; i < toIndex ; ++i)
- {
- count += 64 - Long.bitCount(bits[i]);
- bits[i] = -1L;
- }
- count += Long.bitCount(bits[toIndex] & -1L >>> (64 - (to & 63)));
+ count += Long.bitCount(bits[i]);
+ if ((to & 63) != 0)
+ count += Long.bitCount(bits[toIndex] & -1L >>> (64 - (to & 63)));
return count;
}
}
@@ -300,7 +298,7 @@ public class SimpleBitSet
public final int nextSetBitBefore(int i, int exclBound, int ifNotFound)
{
- int result = nextSetBitInternal(i, upperLimitOf(exclBound), ifNotFound);
+ int result = nextSetBitInternal(i, upperIndexLimitOf(exclBound), ifNotFound);
return result < exclBound ? result : ifNotFound;
}
@@ -437,7 +435,7 @@ public class SimpleBitSet
return i >>> 6;
}
- private int upperLimitOf(int i)
+ private int upperIndexLimitOf(int i)
{
int index = (i + 63) >>> 6;
return Math.min(index, bits.length);
diff --git a/accord-core/src/main/java/accord/utils/SortedArrays.java b/accord-core/src/main/java/accord/utils/SortedArrays.java
index 1576bf48..f6d5e714 100644
--- a/accord-core/src/main/java/accord/utils/SortedArrays.java
+++ b/accord-core/src/main/java/accord/utils/SortedArrays.java
@@ -23,6 +23,7 @@ import java.util.AbstractList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Comparator;
+import java.util.List;
import java.util.function.IntBinaryOperator;
import java.util.function.IntFunction;
import java.util.function.Predicate;
@@ -109,6 +110,15 @@ public class SortedArrays
return isSubset(Comparable::compareTo, test.array, 0, test.array.length, array, 0, array.length);
}
+ public List<T> reverse()
+ {
+ return new AbstractList<T>()
+ {
+ @Override public T get(int index) { return array[array.length - (1 + index)]; }
+ @Override public int size() { return array.length; }
+ };
+ }
+
public T[] backingArrayUnsafe()
{
return array;
diff --git a/accord-core/src/main/java/accord/api/RequestTimeouts.java b/accord-core/src/main/java/accord/utils/UncheckedInterruptedException.java
similarity index 54%
rename from accord-core/src/main/java/accord/api/RequestTimeouts.java
rename to accord-core/src/main/java/accord/utils/UncheckedInterruptedException.java
index 83c3d2bc..ed48b016 100644
--- a/accord-core/src/main/java/accord/api/RequestTimeouts.java
+++ b/accord-core/src/main/java/accord/utils/UncheckedInterruptedException.java
@@ -16,23 +16,32 @@
* limitations under the License.
*/
-package accord.api;
+package accord.utils;
-import java.util.concurrent.TimeUnit;
-
-public interface RequestTimeouts
+public class UncheckedInterruptedException extends RuntimeException
{
- interface RegisteredTimeout
+ public UncheckedInterruptedException()
{
- void cancel();
+ super();
}
- interface Timeout
+ public UncheckedInterruptedException(String message)
{
- void timeout();
- int stripe();
+ super(message);
}
- RegisteredTimeout register(Timeout timeout, long delay, TimeUnit units);
- void maybeNotify();
+ public UncheckedInterruptedException(String message, InterruptedException cause)
+ {
+ super(message, cause);
+ }
+
+ public UncheckedInterruptedException(InterruptedException cause)
+ {
+ super(cause);
+ }
+
+ protected UncheckedInterruptedException(String message, InterruptedException cause, boolean enableSuppression, boolean writableStackTrace)
+ {
+ super(message, cause, enableSuppression, writableStackTrace);
+ }
}
diff --git a/accord-core/src/main/java/accord/utils/async/AsyncChain.java b/accord-core/src/main/java/accord/utils/async/AsyncChain.java
index 0899b268..e53c61f4 100644
--- a/accord-core/src/main/java/accord/utils/async/AsyncChain.java
+++ b/accord-core/src/main/java/accord/utils/async/AsyncChain.java
@@ -127,6 +127,8 @@ public interface AsyncChain<V>
/**
* Causes the chain to begin, starting all work required. This method must be called exactly once, not calling will
* not cause any work to start, and calling multiple times will be rejected.
+ *
+ * TODO (expected): Executor with cheap cancel() that cleansup task queue
*/
@Nullable Cancellable begin(BiConsumer<? super V, Throwable> callback);
diff --git a/accord-core/src/test/java/accord/Utils.java b/accord-core/src/test/java/accord/Utils.java
index 69df53a4..9ea7d858 100644
--- a/accord-core/src/test/java/accord/Utils.java
+++ b/accord-core/src/test/java/accord/Utils.java
@@ -36,7 +36,7 @@ import accord.api.Scheduler;
import accord.api.TopologySorter;
import accord.api.LocalConfig;
import accord.coordinate.CoordinationAdapter;
-import accord.impl.DefaultRequestTimeouts;
+import accord.impl.DefaultTimeouts;
import accord.impl.InMemoryCommandStores;
import accord.impl.IntKey;
import accord.impl.DefaultLocalListeners;
@@ -191,7 +191,7 @@ public class Utils
scheduler,
SizeOfIntersectionSorter.SUPPLIER,
DefaultRemoteListeners::new,
- DefaultRequestTimeouts::new,
+ DefaultTimeouts::new,
DefaultProgressLogs::new,
DefaultLocalListeners.Factory::new,
InMemoryCommandStores.Synchronized::new,
diff --git a/accord-core/src/test/java/accord/burn/BurnTest.java b/accord-core/src/test/java/accord/burn/BurnTest.java
index b7f6f51a..4678cec3 100644
--- a/accord-core/src/test/java/accord/burn/BurnTest.java
+++ b/accord-core/src/test/java/accord/burn/BurnTest.java
@@ -328,7 +328,7 @@ public class BurnTest
Supplier<LongSupplier> nowSupplier = () -> {
RandomSource forked = random.fork();
- // TODO (now): meta-randomise scale of clock drift
+ // TODO (expected): meta-randomise scale of clock drift
return FrequentLargeRange.builder(forked)
.ratio(1, 5)
.small(50, 5000, TimeUnit.MICROSECONDS)
diff --git a/accord-core/src/test/java/accord/coordinate/TopologyChangeTest.java b/accord-core/src/test/java/accord/coordinate/TopologyChangeTest.java
index ca6fe43f..8ed859d8 100644
--- a/accord-core/src/test/java/accord/coordinate/TopologyChangeTest.java
+++ b/accord-core/src/test/java/accord/coordinate/TopologyChangeTest.java
@@ -82,7 +82,7 @@ public class TopologyChangeTest
try
{
config.fetchTopologyForEpoch(2);
- getUninterruptibly(config.ackFor(topology2.epoch()).coordination);
+ getUninterruptibly(config.ackFor(topology2.epoch()).fastPath);
}
catch (ExecutionException e)
{
@@ -183,13 +183,13 @@ public class TopologyChangeTest
cluster.configServices(1, 2, 3, 4, 5).forEach(configService -> configService.reportTopology(topology2));
cluster.nodes(4).forEach(node -> {
MockConfigurationService configService = (MockConfigurationService) node.configService();
- getUncheckedTimeout(configService.ackFor(2).coordination, 5, TimeUnit.SECONDS);
+ getUncheckedTimeout(configService.ackFor(2).fastPath, 5, TimeUnit.SECONDS);
assertEpochRejection(node, keys, 1, false); // shouldn't have received the sync point preaccept
});
cluster.nodes(2, 3).forEach(node -> {
MockConfigurationService configService = (MockConfigurationService) node.configService();
- getUncheckedTimeout(configService.ackFor(2).coordination, 5, TimeUnit.SECONDS);
+ getUncheckedTimeout(configService.ackFor(2).fastPath, 5, TimeUnit.SECONDS);
assertEpochRejection(node, keys, 1, true);
});
@@ -200,14 +200,14 @@ public class TopologyChangeTest
cluster.configServices(1, 2, 3, 4, 5).forEach(configService -> configService.reportTopology(topology3));
cluster.nodes(2, 3).forEach(node -> {
MockConfigurationService configService = (MockConfigurationService) node.configService();
- getUncheckedTimeout(configService.ackFor(3).coordination, 5, TimeUnit.SECONDS);
+ getUncheckedTimeout(configService.ackFor(3).fastPath, 5, TimeUnit.SECONDS);
assertEpochRejection(node, keys, 2, true);
});
// node 4 shouldn't have up to date rejectBefore data
cluster.nodes(4).forEach(node -> {
MockConfigurationService configService = (MockConfigurationService) node.configService();
- getUncheckedTimeout(configService.ackFor(3).coordination, 5, TimeUnit.SECONDS);
+ getUncheckedTimeout(configService.ackFor(3).fastPath, 5, TimeUnit.SECONDS);
assertEpochRejection(node, keys, 1, false); // shouldn't have received the sync point preaccept
assertEpochRejection(node, keys, 2, false); // shouldn't have received the sync point preaccept
});
@@ -251,7 +251,7 @@ public class TopologyChangeTest
MockConfigurationService configService = (MockConfigurationService) node.configService();
try
{
- AsyncChains.getUninterruptibly(configService.ackFor(2).coordination, 5, TimeUnit.SECONDS);
+ AsyncChains.getUninterruptibly(configService.ackFor(2).fastPath, 5, TimeUnit.SECONDS);
Assertions.fail("Expected to timeout on node " + node.id());
}
catch (ExecutionException e)
diff --git a/accord-core/src/test/java/accord/impl/RemoteListenersTest.java b/accord-core/src/test/java/accord/impl/RemoteListenersTest.java
index eab9eb69..2b12039c 100644
--- a/accord-core/src/test/java/accord/impl/RemoteListenersTest.java
+++ b/accord-core/src/test/java/accord/impl/RemoteListenersTest.java
@@ -414,10 +414,13 @@ public class RemoteListenersTest
@Override public CommandStore commandStore() { return commandStore; }
@Override protected SafeCommand getInternal(TxnId txnId) { return null; }
- @Override protected SafeCommand getInternalIfLoadedAndInitialised(TxnId txnId) { return null; }
+ @Override protected SafeCommand ifLoadedAndInitialisedAndNotErasedInternal(TxnId txnId) { return null; }
@Override protected SafeCommandsForKey getInternal(RoutingKey key) { return null;}
- @Override protected SafeCommandsForKey getInternalIfLoadedAndInitialised(RoutingKey key) { return null;}
- @Override public boolean canExecuteWith(PreLoadContext context) { return false;}
+ @Override protected SafeCommandsForKey ifLoadedInternal(RoutingKey key) { return null;}
+ @Override public SafeTimestampsForKey timestampsForKey(RoutingKey key) { return null; }
+
+ @Override public PreLoadContext canExecute(PreLoadContext context) { return null;}
+ @Override public PreLoadContext context() { return null; }
@Override
public void upsertRedundantBefore(RedundantBefore addRedundantBefore)
diff --git a/accord-core/src/test/java/accord/impl/TestAgent.java b/accord-core/src/test/java/accord/impl/TestAgent.java
index 35c82a7b..80a5bf56 100644
--- a/accord-core/src/test/java/accord/impl/TestAgent.java
+++ b/accord-core/src/test/java/accord/impl/TestAgent.java
@@ -149,26 +149,26 @@ public class TestAgent implements Agent
}
@Override
- public long replyTimeout(ReplyContext replyContext, TimeUnit units)
+ public long attemptCoordinationDelay(Node node, SafeCommandStore safeStore, TxnId txnId, TimeUnit units, int retryCount)
{
return units.convert(1L, SECONDS);
}
@Override
- public long attemptCoordinationDelay(Node node, SafeCommandStore safeStore, TxnId txnId, TimeUnit units, int retryCount)
+ public long seekProgressDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, ProgressLog.BlockedUntil blockedUntil, TimeUnit units)
{
return units.convert(1L, SECONDS);
}
@Override
- public long seekProgressDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, ProgressLog.BlockedUntil blockedUntil, TimeUnit units)
+ public long retryAwaitTimeout(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, ProgressLog.BlockedUntil retrying, TimeUnit units)
{
return units.convert(1L, SECONDS);
}
@Override
- public long retryAwaitTimeout(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, ProgressLog.BlockedUntil retrying, TimeUnit units)
+ public long expiresAt(ReplyContext replyContext, TimeUnit unit)
{
- return units.convert(1L, SECONDS);
+ return 0;
}
}
diff --git a/accord-core/src/test/java/accord/impl/basic/Cluster.java b/accord-core/src/test/java/accord/impl/basic/Cluster.java
index fbe83707..56b7a970 100644
--- a/accord-core/src/test/java/accord/impl/basic/Cluster.java
+++ b/accord-core/src/test/java/accord/impl/basic/Cluster.java
@@ -66,7 +66,7 @@ import accord.coordinate.Truncated;
import accord.impl.DurabilityScheduling;
import accord.impl.DefaultLocalListeners;
import accord.impl.DefaultRemoteListeners;
-import accord.impl.DefaultRequestTimeouts;
+import accord.impl.DefaultTimeouts;
import accord.impl.InMemoryCommandStore.GlobalCommand;
import accord.impl.MessageListener;
import accord.impl.PrefixedIntHashKey;
@@ -480,7 +480,7 @@ public class Cluster implements Scheduler
Node node = new Node(id, messageSink, configService, TimeService.ofNonMonotonic(nowSupplier, MILLISECONDS),
() -> new ListStore(sinks, random, id), new ShardDistributor.EvenSplit<>(8, ignore -> new PrefixedIntHashKey.Splitter()),
nodeExecutor.agent(),
- randomSupplier.get(), sinks, SizeOfIntersectionSorter.SUPPLIER, DefaultRemoteListeners::new, DefaultRequestTimeouts::new,
+ randomSupplier.get(), sinks, SizeOfIntersectionSorter.SUPPLIER, DefaultRemoteListeners::new, DefaultTimeouts::new,
DefaultProgressLogs::new, DefaultLocalListeners.Factory::new, DelayedCommandStores.factory(sinks.pending, isLoadedCheck, journal), new CoordinationAdapter.DefaultFactory(),
DurableBefore.NOOP_PERSISTER, localConfig);
DurabilityScheduling durability = node.durabilityScheduling();
diff --git a/accord-core/src/test/java/accord/impl/basic/DelayedCommandStores.java b/accord-core/src/test/java/accord/impl/basic/DelayedCommandStores.java
index ed1df416..e5cf16d5 100644
--- a/accord-core/src/test/java/accord/impl/basic/DelayedCommandStores.java
+++ b/accord-core/src/test/java/accord/impl/basic/DelayedCommandStores.java
@@ -251,7 +251,10 @@ public class DelayedCommandStores extends InMemoryCommandStores.SingleThread
if (wasEmpty)
runNextTask();
task.begin(callback);
- return () -> pending.remove(task);
+ return () -> {
+ if (pending.peek() != task)
+ pending.remove(task);
+ };
}
};
}
diff --git a/accord-core/src/test/java/accord/impl/basic/Journal.java b/accord-core/src/test/java/accord/impl/basic/Journal.java
index ec0c1803..66052112 100644
--- a/accord-core/src/test/java/accord/impl/basic/Journal.java
+++ b/accord-core/src/test/java/accord/impl/basic/Journal.java
@@ -86,7 +86,7 @@ public class Journal
Command command = reconstruct(diffs, Reconstruct.Last).get(0);
if (command.status() == Truncated || command.status() == Invalidated)
continue; // Already truncated
- Cleanup cleanup = Cleanup.shouldCleanup(command, store.unsafeGetRedundantBefore(), store.durableBefore());
+ Cleanup cleanup = Cleanup.shouldCleanup(store.agent(), command, store.unsafeGetRedundantBefore(), store.durableBefore());
switch (cleanup)
{
case NO:
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 2f1d5712..bb1b04c8 100644
--- a/accord-core/src/test/java/accord/impl/list/ListAgent.java
+++ b/accord-core/src/test/java/accord/impl/list/ListAgent.java
@@ -149,12 +149,6 @@ public class ListAgent implements Agent
return new Txn.InMemory(kind, domain == Domain.Key ? Keys.EMPTY : Ranges.EMPTY, new ListRead(identity(), false, Keys.EMPTY, Keys.EMPTY), new ListQuery(NONE, Integer.MIN_VALUE, false), null);
}
- @Override
- public long replyTimeout(ReplyContext replyContext, TimeUnit units)
- {
- return units.convert(timeoutDelays.getAsInt(), MILLISECONDS);
- }
-
@Override
public long attemptCoordinationDelay(Node node, SafeCommandStore safeStore, TxnId txnId, TimeUnit units, int retryCount)
{
@@ -175,6 +169,14 @@ public class ListAgent implements Agent
return units.convert(retryDelay, SECONDS);
}
+ @Override
+ public long expiresAt(ReplyContext replyContext, TimeUnit unit)
+ {
+ // TODO (required): implement this (add it to the replyContext)
+ return -1;
+ }
+
+
public boolean collectMaxApplied()
{
// TODO (expected): randomise this to exercise both code paths
diff --git a/accord-core/src/test/java/accord/impl/mock/MockCluster.java b/accord-core/src/test/java/accord/impl/mock/MockCluster.java
index a7abceb7..c125f9e1 100644
--- a/accord-core/src/test/java/accord/impl/mock/MockCluster.java
+++ b/accord-core/src/test/java/accord/impl/mock/MockCluster.java
@@ -36,7 +36,7 @@ import accord.NetworkFilter;
import accord.api.MessageSink;
import accord.api.LocalConfig;
import accord.coordinate.CoordinationAdapter;
-import accord.impl.DefaultRequestTimeouts;
+import accord.impl.DefaultTimeouts;
import accord.impl.InMemoryCommandStores;
import accord.impl.IntKey;
import accord.impl.DefaultLocalListeners;
@@ -137,7 +137,7 @@ public class MockCluster implements Network, AutoCloseable, Iterable<Node>
new ThreadPoolScheduler(),
SizeOfIntersectionSorter.SUPPLIER,
DefaultRemoteListeners::new,
- DefaultRequestTimeouts::new,
+ DefaultTimeouts::new,
DefaultProgressLogs::new,
DefaultLocalListeners.Factory::new,
InMemoryCommandStores.SingleThread::new,
diff --git a/accord-core/src/test/java/accord/local/ImmutableCommandTest.java b/accord-core/src/test/java/accord/local/ImmutableCommandTest.java
index 57d508e8..b5958276 100644
--- a/accord-core/src/test/java/accord/local/ImmutableCommandTest.java
+++ b/accord-core/src/test/java/accord/local/ImmutableCommandTest.java
@@ -33,7 +33,7 @@ import accord.api.RoutingKey;
import accord.api.Scheduler;
import accord.api.TestableConfigurationService;
import accord.coordinate.CoordinationAdapter;
-import accord.impl.DefaultRequestTimeouts;
+import accord.impl.DefaultTimeouts;
import accord.impl.InMemoryCommandStore;
import accord.impl.InMemoryCommandStores;
import accord.impl.IntKey;
@@ -104,7 +104,7 @@ public class ImmutableCommandTest
Node node = new Node(id, null, new MockConfigurationService(null, (epoch, service) -> { }, storeSupport.local.get()),
clock,
() -> storeSupport.data, new ShardDistributor.EvenSplit(8, ignore -> new IntKey.Splitter()), new TestAgent(), new DefaultRandom(), Scheduler.NEVER_RUN_SCHEDULED,
- SizeOfIntersectionSorter.SUPPLIER, DefaultRemoteListeners::new, DefaultRequestTimeouts::new, ignore -> ignore2 -> new NoOpProgressLog(), DefaultLocalListeners.Factory::new,
+ SizeOfIntersectionSorter.SUPPLIER, DefaultRemoteListeners::new, DefaultTimeouts::new, ignore -> ignore2 -> new NoOpProgressLog(), DefaultLocalListeners.Factory::new,
InMemoryCommandStores.Synchronized::new,
new CoordinationAdapter.DefaultFactory(),
DurableBefore.NOOP_PERSISTER,
diff --git a/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java b/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java
index 73079a80..dfe37867 100644
--- a/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java
+++ b/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java
@@ -50,6 +50,7 @@ import accord.impl.DefaultLocalListeners;
import accord.impl.DefaultLocalListeners.DefaultNotifySink;
import accord.impl.DefaultRemoteListeners;
import accord.impl.IntKey;
+import accord.impl.SafeTimestampsForKey;
import accord.local.Command;
import accord.local.Command.AbstractCommand;
import accord.local.CommandStore;
@@ -718,7 +719,7 @@ public class CommandsForKeyTest
}
@Override
- protected SafeCommand getInternalIfLoadedAndInitialised(TxnId txnId)
+ protected SafeCommand ifLoadedAndInitialisedAndNotErasedInternal(TxnId txnId)
{
return getInternal(txnId);
}
@@ -756,7 +757,7 @@ public class CommandsForKeyTest
}
@Override
- public SafeCommand ifLoadedAndInitialised(TxnId txnId)
+ public SafeCommand ifLoadedAndInitialisedAndNotErased(TxnId txnId)
{
if (txnId.compareTo(cfk.current.prunedBefore()) < 0)
return null;
@@ -776,7 +777,7 @@ public class CommandsForKeyTest
}
@Override
- protected SafeCommandsForKey getInternalIfLoadedAndInitialised(RoutingKey key)
+ protected SafeCommandsForKey ifLoadedInternal(RoutingKey key)
{
if (key.equals(cfk.key()))
return cfk;
@@ -784,9 +785,21 @@ public class CommandsForKeyTest
}
@Override
- public boolean canExecuteWith(PreLoadContext context)
+ public SafeTimestampsForKey timestampsForKey(RoutingKey key)
{
- return true;
+ return null;
+ }
+
+ @Override
+ public PreLoadContext canExecute(PreLoadContext context)
+ {
+ return context;
+ }
+
+ @Override
+ public PreLoadContext context()
+ {
+ return null;
}
@Override
@@ -1044,25 +1057,25 @@ public class CommandsForKeyTest
}
@Override
- public long replyTimeout(ReplyContext replyContext, TimeUnit units)
+ public long attemptCoordinationDelay(Node node, SafeCommandStore safeStore, TxnId txnId, TimeUnit units, int retryCount)
{
- throw new UnsupportedOperationException();
+ return 0;
}
@Override
- public long attemptCoordinationDelay(Node node, SafeCommandStore safeStore, TxnId txnId, TimeUnit units, int retryCount)
+ public long seekProgressDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, BlockedUntil blockedUntil, TimeUnit units)
{
return 0;
}
@Override
- public long seekProgressDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, BlockedUntil blockedUntil, TimeUnit units)
+ public long retryAwaitTimeout(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, BlockedUntil retrying, TimeUnit units)
{
return 0;
}
@Override
- public long retryAwaitTimeout(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, BlockedUntil retrying, TimeUnit units)
+ public long expiresAt(ReplyContext replyContext, TimeUnit unit)
{
return 0;
}
diff --git a/accord-core/src/test/java/accord/messages/PreAcceptTest.java b/accord-core/src/test/java/accord/messages/PreAcceptTest.java
index a32c08fa..c07e8252 100644
--- a/accord-core/src/test/java/accord/messages/PreAcceptTest.java
+++ b/accord-core/src/test/java/accord/messages/PreAcceptTest.java
@@ -208,7 +208,7 @@ public class PreAcceptTest
messageSink.assertHistorySizes(0, 1);
Assertions.assertEquals(ID3, messageSink.responses.get(0).to);
PartialDeps expectedDeps = new PartialDeps(Ranges.of(range(0, 12)), KeyDeps.NONE, RangeDeps.NONE, KeyDeps.NONE);
- Timestamp expectedTs = Timestamp.fromValues(1, 110, ID1).withExtraFlags(txnId2.flags());
+ Timestamp expectedTs = Timestamp.fromValues(1, 110, ID1);
Assertions.assertEquals(new PreAccept.PreAcceptOk(txnId2, expectedTs, expectedDeps),
messageSink.responses.get(0).payload);
}
diff --git a/accord-core/src/test/java/accord/utils/LogGroupTimersTest.java b/accord-core/src/test/java/accord/utils/LogGroupTimersTest.java
index b5def35e..bb09b65b 100644
--- a/accord-core/src/test/java/accord/utils/LogGroupTimersTest.java
+++ b/accord-core/src/test/java/accord/utils/LogGroupTimersTest.java
@@ -152,7 +152,7 @@ public class LogGroupTimersTest
{
Timer first = canonical.first();
Assertions.assertTrue(test.wakeAt() <= first.deadline);
- Assertions.assertTrue(test.wakeAt() >= test.curEpoch);
+ Assertions.assertTrue(test.wakeAt() >= Math.min(test.curEpoch, test.buckets[test.bucketsStart].epoch));
}
if (rnd.decide(readChance))
@@ -162,7 +162,7 @@ public class LogGroupTimersTest
{
Timer first = canonical.first();
Assertions.assertTrue(test.wakeAt() <= first.deadline);
- Assertions.assertTrue(test.wakeAt() >= test.curEpoch);
+ Assertions.assertTrue(test.wakeAt() >= Math.min(test.curEpoch, test.buckets[test.bucketsStart].epoch));
}
}
diff --git a/accord-maelstrom/src/main/java/accord/maelstrom/Cluster.java b/accord-maelstrom/src/main/java/accord/maelstrom/Cluster.java
index 46d497c6..8fb34489 100644
--- a/accord-maelstrom/src/main/java/accord/maelstrom/Cluster.java
+++ b/accord-maelstrom/src/main/java/accord/maelstrom/Cluster.java
@@ -43,7 +43,7 @@ import accord.api.MessageSink;
import accord.api.Scheduler;
import accord.api.LocalConfig;
import accord.coordinate.CoordinationAdapter;
-import accord.impl.DefaultRequestTimeouts;
+import accord.impl.DefaultTimeouts;
import accord.impl.InMemoryCommandStores;
import accord.impl.DefaultLocalListeners;
import accord.impl.progresslog.DefaultProgressLogs;
@@ -345,7 +345,7 @@ public class Cluster implements Scheduler
TimeService.of(nowSupplier, elapsedWrapperFromNonMonotonicSource(TimeUnit.MICROSECONDS, nowSupplier)),
MaelstromStore::new, new ShardDistributor.EvenSplit(8, ignore -> new MaelstromKey.Splitter()),
MaelstromAgent.INSTANCE,
- randomSupplier.get(), sinks, SizeOfIntersectionSorter.SUPPLIER, DefaultRemoteListeners::new, DefaultRequestTimeouts::new,
+ randomSupplier.get(), sinks, SizeOfIntersectionSorter.SUPPLIER, DefaultRemoteListeners::new, DefaultTimeouts::new,
DefaultProgressLogs::new, DefaultLocalListeners.Factory::new, InMemoryCommandStores.SingleThread::new, new CoordinationAdapter.DefaultFactory(),
DurableBefore.NOOP_PERSISTER, localConfig));
}
diff --git a/accord-maelstrom/src/main/java/accord/maelstrom/MaelstromAgent.java b/accord-maelstrom/src/main/java/accord/maelstrom/MaelstromAgent.java
index 6b4bb3f7..06cdd3a8 100644
--- a/accord-maelstrom/src/main/java/accord/maelstrom/MaelstromAgent.java
+++ b/accord-maelstrom/src/main/java/accord/maelstrom/MaelstromAgent.java
@@ -121,9 +121,9 @@ public class MaelstromAgent implements Agent
}
@Override
- public long replyTimeout(ReplyContext replyContext, TimeUnit units)
+ public long expiresAt(ReplyContext replyContext, TimeUnit units)
{
- return units.convert(1L, SECONDS);
+ return -1;
}
@Override
diff --git a/accord-maelstrom/src/main/java/accord/maelstrom/Main.java b/accord-maelstrom/src/main/java/accord/maelstrom/Main.java
index 9b3cbe39..306313e1 100644
--- a/accord-maelstrom/src/main/java/accord/maelstrom/Main.java
+++ b/accord-maelstrom/src/main/java/accord/maelstrom/Main.java
@@ -35,7 +35,7 @@ import accord.api.Scheduler;
import accord.api.LocalConfig;
import accord.coordinate.CoordinationAdapter;
import accord.coordinate.Timeout;
-import accord.impl.DefaultRequestTimeouts;
+import accord.impl.DefaultTimeouts;
import accord.impl.InMemoryCommandStores;
import accord.impl.DefaultLocalListeners;
import accord.impl.progresslog.DefaultProgressLogs;
@@ -183,7 +183,7 @@ public class Main
TimeService.ofNonMonotonic(System::currentTimeMillis, TimeUnit.MILLISECONDS),
MaelstromStore::new, new ShardDistributor.EvenSplit(8, ignore -> new MaelstromKey.Splitter()),
MaelstromAgent.INSTANCE, new DefaultRandom(), scheduler, SizeOfIntersectionSorter.SUPPLIER,
- DefaultRemoteListeners::new, DefaultRequestTimeouts::new, DefaultProgressLogs::new, DefaultLocalListeners.Factory::new,
+ DefaultRemoteListeners::new, DefaultTimeouts::new, DefaultProgressLogs::new, DefaultLocalListeners.Factory::new,
InMemoryCommandStores.SingleThread::new, new CoordinationAdapter.DefaultFactory(),
DurableBefore.NOOP_PERSISTER, localConfig);
awaitUninterruptibly(on.unsafeStart());
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org