2 * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
4 * This program and the accompanying materials are made available under the
5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6 * and is available at http://www.eclipse.org/legal/epl-v10.html
9 package org.opendaylight.controller.cluster.datastore;
11 import akka.actor.ActorSelection;
12 import akka.dispatch.OnComplete;
13 import com.google.common.base.Preconditions;
14 import com.google.common.base.Supplier;
15 import com.google.common.collect.Lists;
16 import com.google.common.util.concurrent.FutureCallback;
17 import com.google.common.util.concurrent.Futures;
18 import com.google.common.util.concurrent.ListenableFuture;
19 import com.google.common.util.concurrent.SettableFuture;
20 import java.util.ArrayList;
21 import java.util.Iterator;
22 import java.util.List;
23 import java.util.concurrent.atomic.AtomicInteger;
24 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
25 import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction;
26 import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply;
27 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction;
28 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
29 import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
30 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
31 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
32 import org.slf4j.Logger;
33 import org.slf4j.LoggerFactory;
34 import scala.concurrent.Future;
37 * ThreePhaseCommitCohortProxy represents a set of remote cohort proxies.
39 public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort<ActorSelection> {
41 private static final Logger LOG = LoggerFactory.getLogger(ThreePhaseCommitCohortProxy.class);
43 private static final MessageSupplier COMMIT_MESSAGE_SUPPLIER = new MessageSupplier() {
45 public Object newMessage(TransactionIdentifier transactionId, short version) {
46 return new CommitTransaction(transactionId, version).toSerializable();
50 public boolean isSerializedReplyType(Object reply) {
51 return CommitTransactionReply.isSerializedType(reply);
55 private static final MessageSupplier ABORT_MESSAGE_SUPPLIER = new MessageSupplier() {
57 public Object newMessage(TransactionIdentifier transactionId, short version) {
58 return new AbortTransaction(transactionId, version).toSerializable();
62 public boolean isSerializedReplyType(Object reply) {
63 return AbortTransactionReply.isSerializedType(reply);
67 private final ActorContext actorContext;
68 private final List<CohortInfo> cohorts;
69 private final SettableFuture<Void> cohortsResolvedFuture = SettableFuture.create();
70 private final TransactionIdentifier transactionId;
71 private volatile OperationCallback commitOperationCallback;
73 public ThreePhaseCommitCohortProxy(ActorContext actorContext, List<CohortInfo> cohorts,
74 TransactionIdentifier transactionId) {
75 this.actorContext = actorContext;
76 this.cohorts = cohorts;
77 this.transactionId = Preconditions.checkNotNull(transactionId);
79 if (cohorts.isEmpty()) {
80 cohortsResolvedFuture.set(null);
84 private ListenableFuture<Void> resolveCohorts() {
85 if (cohortsResolvedFuture.isDone()) {
86 return cohortsResolvedFuture;
89 final AtomicInteger completed = new AtomicInteger(cohorts.size());
90 final Object lock = new Object();
91 for (final CohortInfo info: cohorts) {
92 info.getActorFuture().onComplete(new OnComplete<ActorSelection>() {
94 public void onComplete(Throwable failure, ActorSelection actor) {
96 boolean done = completed.decrementAndGet() == 0;
97 if (failure != null) {
98 LOG.debug("Tx {}: a cohort Future failed", transactionId, failure);
99 cohortsResolvedFuture.setException(failure);
100 } else if (!cohortsResolvedFuture.isDone()) {
101 LOG.debug("Tx {}: cohort actor {} resolved", transactionId, actor);
103 info.setResolvedActor(actor);
105 LOG.debug("Tx {}: successfully resolved all cohort actors", transactionId);
106 cohortsResolvedFuture.set(null);
111 }, actorContext.getClientDispatcher());
114 return cohortsResolvedFuture;
118 public ListenableFuture<Boolean> canCommit() {
119 LOG.debug("Tx {} canCommit", transactionId);
121 final SettableFuture<Boolean> returnFuture = SettableFuture.create();
123 // The first phase of canCommit is to gather the list of cohort actor paths that will
124 // participate in the commit. buildCohortPathsList combines the cohort path Futures into
125 // one Future which we wait on asynchronously here. The cohort actor paths are
126 // extracted from ReadyTransactionReply messages by the Futures that were obtained earlier
127 // and passed to us from upstream processing. If any one fails then we'll fail canCommit.
129 Futures.addCallback(resolveCohorts(), new FutureCallback<Void>() {
131 public void onSuccess(Void notUsed) {
132 finishCanCommit(returnFuture);
136 public void onFailure(Throwable failure) {
137 returnFuture.setException(failure);
144 private void finishCanCommit(final SettableFuture<Boolean> returnFuture) {
145 LOG.debug("Tx {} finishCanCommit", transactionId);
147 // For empty transactions return immediately
148 if (cohorts.size() == 0) {
149 LOG.debug("Tx {}: canCommit returning result true", transactionId);
150 returnFuture.set(Boolean.TRUE);
154 commitOperationCallback = new TransactionRateLimitingCallback(actorContext);
155 commitOperationCallback.run();
157 final Iterator<CohortInfo> iterator = cohorts.iterator();
159 final OnComplete<Object> onComplete = new OnComplete<Object>() {
161 public void onComplete(Throwable failure, Object response) {
162 if (failure != null) {
163 LOG.debug("Tx {}: a canCommit cohort Future failed", transactionId, failure);
165 returnFuture.setException(failure);
166 commitOperationCallback.failure();
170 // Only the first call to pause takes effect - subsequent calls before resume are no-ops. So
171 // this means we'll only time the first transaction canCommit which should be fine.
172 commitOperationCallback.pause();
174 boolean result = true;
175 if (CanCommitTransactionReply.isSerializedType(response)) {
176 CanCommitTransactionReply reply = CanCommitTransactionReply.fromSerializable(response);
178 LOG.debug("Tx {}: received {}", transactionId, response);
180 if (!reply.getCanCommit()) {
184 LOG.error("Unexpected response type {}", response.getClass());
185 returnFuture.setException(new IllegalArgumentException(
186 String.format("Unexpected response type %s", response.getClass())));
190 if (iterator.hasNext() && result) {
191 sendCanCommitTransaction(iterator.next(), this);
193 LOG.debug("Tx {}: canCommit returning result: {}", transactionId, result);
194 returnFuture.set(Boolean.valueOf(result));
200 sendCanCommitTransaction(iterator.next(), onComplete);
203 private void sendCanCommitTransaction(CohortInfo toCohortInfo, OnComplete<Object> onComplete) {
204 CanCommitTransaction message = new CanCommitTransaction(transactionId, toCohortInfo.getActorVersion());
206 LOG.debug("Tx {}: sending {} to {}", transactionId, message, toCohortInfo.getResolvedActor());
208 Future<Object> future = actorContext.executeOperationAsync(toCohortInfo.getResolvedActor(),
209 message.toSerializable(), actorContext.getTransactionCommitOperationTimeout());
210 future.onComplete(onComplete, actorContext.getClientDispatcher());
213 private Future<Iterable<Object>> invokeCohorts(MessageSupplier messageSupplier) {
214 List<Future<Object>> futureList = Lists.newArrayListWithCapacity(cohorts.size());
215 for (CohortInfo cohort : cohorts) {
216 Object message = messageSupplier.newMessage(transactionId, cohort.getActorVersion());
218 LOG.debug("Tx {}: Sending {} to cohort {}", transactionId, message , cohort);
220 futureList.add(actorContext.executeOperationAsync(cohort.getResolvedActor(), message,
221 actorContext.getTransactionCommitOperationTimeout()));
224 return akka.dispatch.Futures.sequence(futureList, actorContext.getClientDispatcher());
228 public ListenableFuture<Void> preCommit() {
229 // We don't need to do anything here - preCommit is done atomically with the commit phase
231 return IMMEDIATE_VOID_SUCCESS;
235 public ListenableFuture<Void> abort() {
236 // Note - we pass false for propagateException. In the front-end data broker, this method
237 // is called when one of the 3 phases fails with an exception. We'd rather have that
238 // original exception propagated to the client. If our abort fails and we propagate the
239 // exception then that exception will supersede and suppress the original exception. But
240 // it's the original exception that is the root cause and of more interest to the client.
242 return voidOperation("abort", ABORT_MESSAGE_SUPPLIER,
243 AbortTransactionReply.class, false, OperationCallback.NO_OP_CALLBACK);
247 public ListenableFuture<Void> commit() {
248 OperationCallback operationCallback = commitOperationCallback != null ? commitOperationCallback :
249 OperationCallback.NO_OP_CALLBACK;
251 return voidOperation("commit", COMMIT_MESSAGE_SUPPLIER,
252 CommitTransactionReply.class, true, operationCallback);
255 @SuppressWarnings("checkstyle:IllegalCatch")
256 private static boolean successfulFuture(ListenableFuture<Void> future) {
257 if (!future.isDone()) {
264 } catch (Exception e) {
269 private ListenableFuture<Void> voidOperation(final String operationName,
270 final MessageSupplier messageSupplier, final Class<?> expectedResponseClass,
271 final boolean propagateException, final OperationCallback callback) {
272 LOG.debug("Tx {} {}", transactionId, operationName);
274 final SettableFuture<Void> returnFuture = SettableFuture.create();
276 // The cohort actor list should already be built at this point by the canCommit phase but,
277 // if not for some reason, we'll try to build it here.
279 ListenableFuture<Void> future = resolveCohorts();
280 if (successfulFuture(future)) {
281 finishVoidOperation(operationName, messageSupplier, expectedResponseClass, propagateException,
282 returnFuture, callback);
284 Futures.addCallback(future, new FutureCallback<Void>() {
286 public void onSuccess(Void notUsed) {
287 finishVoidOperation(operationName, messageSupplier, expectedResponseClass,
288 propagateException, returnFuture, callback);
292 public void onFailure(Throwable failure) {
293 LOG.debug("Tx {}: a {} cohort path Future failed: {}", transactionId, operationName, failure);
295 if (propagateException) {
296 returnFuture.setException(failure);
298 returnFuture.set(null);
307 private void finishVoidOperation(final String operationName, MessageSupplier messageSupplier,
308 final Class<?> expectedResponseClass, final boolean propagateException,
309 final SettableFuture<Void> returnFuture, final OperationCallback callback) {
310 LOG.debug("Tx {} finish {}", transactionId, operationName);
314 Future<Iterable<Object>> combinedFuture = invokeCohorts(messageSupplier);
316 combinedFuture.onComplete(new OnComplete<Iterable<Object>>() {
318 public void onComplete(Throwable failure, Iterable<Object> responses) throws Throwable {
319 Throwable exceptionToPropagate = failure;
320 if (exceptionToPropagate == null) {
321 for (Object response: responses) {
322 if (!response.getClass().equals(expectedResponseClass)) {
323 exceptionToPropagate = new IllegalArgumentException(
324 String.format("Unexpected response type %s", response.getClass()));
330 if (exceptionToPropagate != null) {
331 LOG.debug("Tx {}: a {} cohort Future failed", transactionId, operationName, exceptionToPropagate);
332 if (propagateException) {
333 // We don't log the exception here to avoid redundant logging since we're
334 // propagating to the caller in MD-SAL core who will log it.
335 returnFuture.setException(exceptionToPropagate);
337 // Since the caller doesn't want us to propagate the exception we'll also
338 // not log it normally. But it's usually not good to totally silence
339 // exceptions so we'll log it to debug level.
340 returnFuture.set(null);
345 LOG.debug("Tx {}: {} succeeded", transactionId, operationName);
347 returnFuture.set(null);
352 }, actorContext.getClientDispatcher());
356 List<Future<ActorSelection>> getCohortFutures() {
357 List<Future<ActorSelection>> cohortFutures = new ArrayList<>(cohorts.size());
358 for (CohortInfo info: cohorts) {
359 cohortFutures.add(info.getActorFuture());
362 return cohortFutures;
365 static class CohortInfo {
366 private final Future<ActorSelection> actorFuture;
367 private volatile ActorSelection resolvedActor;
368 private final Supplier<Short> actorVersionSupplier;
370 CohortInfo(Future<ActorSelection> actorFuture, Supplier<Short> actorVersionSupplier) {
371 this.actorFuture = actorFuture;
372 this.actorVersionSupplier = actorVersionSupplier;
375 Future<ActorSelection> getActorFuture() {
379 ActorSelection getResolvedActor() {
380 return resolvedActor;
383 void setResolvedActor(ActorSelection resolvedActor) {
384 this.resolvedActor = resolvedActor;
387 short getActorVersion() {
388 Preconditions.checkState(resolvedActor != null,
389 "getActorVersion cannot be called until the actor is resolved");
390 return actorVersionSupplier.get();
394 private interface MessageSupplier {
395 Object newMessage(TransactionIdentifier transactionId, short version);
397 boolean isSerializedReplyType(Object reply);