belliottsmith commented on code in PR #45:
URL: https://github.com/apache/cassandra-accord/pull/45#discussion_r1188339989
##########
accord-core/src/main/java/accord/local/CommandStore.java:
##########
@@ -154,4 +257,219 @@ public static void checkNotInStore()
if (store != null)
throw new IllegalStateException("Expected to not be running in a
CommandStore, but running in " + store);
}
+
+ /**
+ * Defer submitting the work until we have wired up any changes to
topology in memory, then first submit the work
+ * to setup any state in the command store, and finally submit the
distributed work to bootstrap the data locally.
+ * So, the outer future's success is sufficient for the topology to be
acknowledged, and the inner future for the
+ * bootstrap to be complete.
+ */
+ Supplier<EpochReady> bootstrapper(Node node, Ranges newRanges, long epoch)
+ {
+ return () -> {
+ AsyncResult<EpochReady> metadata = submit(empty(), safeStore -> {
+ Bootstrap bootstrap = new Bootstrap(node, this, epoch,
newRanges);
+ bootstraps.add(bootstrap);
+ bootstrap.start(safeStore);
+ return new EpochReady(epoch, null, bootstrap.coordination,
bootstrap.data, bootstrap.reads);
+ }).beginAsResult();
+
+ return new EpochReady(epoch, metadata.<Void>map(ignore ->
null).beginAsResult(),
+ metadata.flatMap(e -> e.coordination).beginAsResult(),
+ metadata.flatMap(e -> e.data).beginAsResult(),
+ metadata.flatMap(e -> e.reads).beginAsResult());
+ };
+ }
+
+ /**
+ * Defer submitting the work until we have wired up any changes to
topology in memory, then first submit the work
+ * to setup any state in the command store, and finally submit the
distributed work to bootstrap the data locally.
+ * So, the outer future's success is sufficient for the topology to be
acknowledged, and the inner future for the
+ * bootstrap to be complete.
+ */
+ Supplier<EpochReady> sync(Node node, Ranges ranges, long epoch)
+ {
+ return () -> {
+ AsyncResults.SettableResult<Void> whenDone = new
AsyncResults.SettableResult<>();
+ fetchMajorityDeps(whenDone, node, epoch, ranges);
+ return new EpochReady(epoch, DONE, whenDone, whenDone, whenDone);
+ };
+ }
+
+ private void fetchMajorityDeps(AsyncResults.SettableResult<Void>
coordination, Node node, long epoch, Ranges ranges)
+ {
+ TxnId id = TxnId.fromValues(epoch - 1, 0, node.id());
+ Timestamp before = Timestamp.minForEpoch(epoch);
+ FullRoute<?> route = node.computeRoute(id, ranges);
+ CollectDeps.withDeps(node, id, route, ranges, before, (deps, fail) -> {
+ if (fail != null)
+ {
+ fetchMajorityDeps(coordination, node, epoch, ranges);
+ }
+ else
+ {
+ execute(contextFor(null, deps.txnIds()), safeStore -> {
+ registerHistoricalTransactions(deps);
Review Comment:
Oh wait, maybe I misunderstood your query as the context is the registering
if "historical transactions" (may be we can name it better - we're registering
transactions for use in `calculateDependencies`)
But their fast or slow path outcomes should not be relevant here, we're just
propagating deps before fully deprecating the old epoch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
