belliottsmith commented on code in PR #113: URL: https://github.com/apache/cassandra-accord/pull/113#discussion_r1745960874
########## accord-core/src/main/java/accord/coordinate/AsynchronousAwait.java: ########## @@ -0,0 +1,140 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package accord.coordinate; + +import java.util.function.BiConsumer; +import javax.annotation.Nullable; + +import accord.api.ProgressLog.BlockedUntil; +import accord.coordinate.tracking.AwaitTracker; +import accord.coordinate.tracking.RequestStatus; +import accord.local.Commands; +import accord.local.Node; +import accord.local.Node.Id; +import accord.messages.Await; +import accord.messages.Await.AwaitOk; +import accord.messages.Callback; +import accord.primitives.Route; +import accord.primitives.TxnId; +import accord.primitives.Unseekables; +import accord.topology.Topologies; + +import static accord.coordinate.tracking.RequestStatus.Success; + +public class AsynchronousAwait implements Callback<AwaitOk> +{ + // TODO (desired, efficiency): this should collect the executeAt of any commit, and terminate as soon as one is found + // that is earlier than TxnId for the Txn we are recovering; if all commits we wait for + // are given earlier timestamps we can retry without restarting. + + public static class SynchronousResult + { + public final Unseekables<?> ready; + public final @Nullable Unseekables<?> notReady; + + public SynchronousResult(Unseekables<?> ready, @Nullable Unseekables<?> notReady) + { + this.ready = ready; + this.notReady = notReady; + } + } + + final TxnId txnId; + final Unseekables<?> contact; + final AwaitTracker tracker; + final int asynchronousCallbackId; + final BiConsumer<SynchronousResult, Throwable> synchronousCallback; + private boolean isDone; + private Throwable failure; + + public AsynchronousAwait(TxnId txnId, Unseekables<?> contact, AwaitTracker tracker, int asynchronousCallbackId, BiConsumer<SynchronousResult, Throwable> synchronousCallback) + { + this.txnId = txnId; + this.contact = contact; + this.tracker = tracker; + this.asynchronousCallbackId = asynchronousCallbackId; + this.synchronousCallback = synchronousCallback; + } + + /** + * we require a Route to contact so we can be sure a home shard recipient invokes {@link Commands#updateRouteOrParticipants}, + * notifying the progress log of a Route to determine it is the home shard. + */ + public static AsynchronousAwait awaitAny(Node node, Topologies topologies, TxnId txnId, Route<?> contact, BlockedUntil awaiting, int asynchronousCallbackId, BiConsumer<SynchronousResult, Throwable> synchronousCallback) + { + AwaitTracker tracker = new AwaitTracker(topologies); + AsynchronousAwait result = new AsynchronousAwait(txnId, contact, tracker, asynchronousCallbackId, synchronousCallback); + result.start(node, topologies, contact, awaiting); + return result; + } + + private void start(Node node, Topologies topologies, Route<?> route, BlockedUntil blockedUntil) + { + node.send(topologies.nodes(), to -> new Await(to, topologies, txnId, route, blockedUntil, asynchronousCallbackId), this); + } + + @Override + public synchronized void onSuccess(Id from, AwaitOk reply) + { + if (isDone) return; + + if (tracker.recordSuccess(from, reply == AwaitOk.Ready) == Success) + onSuccess(); + } + + @Override + public synchronized void onFailure(Id from, Throwable failure) + { + if (isDone) return; + + this.failure = FailureAccumulator.append(this.failure, failure); + RequestStatus status = tracker.recordFailure(from); + switch (status) + { + default: throw new AssertionError("Unhandled RequestStatus: " + status); + case NoChange: break; + case Success: + onSuccess(); + break; + case Failed: + isDone = true; + synchronousCallback.accept(null, this.failure); + } + } + + private void onSuccess() + { + isDone = true; Review Comment: That is intentional. You can't safely invoke the callback more than once. We could have a special case in all `onCallbackFailure` implementations where we invoke `unhandledException`. -- 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]
