X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FThreePhaseCommitCohort.java;h=df85bb136a93b51084676a39cb1d7b53b54b7037;hp=e6adfbee660669afd0889995c5e4ec0c755810e9;hb=c352d8958cf16d2eded7bb103b7be69546164ce5;hpb=9d166a0ee8e022ae9ec30fb251584d57f313c077 diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohort.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohort.java index e6adfbee66..df85bb136a 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohort.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohort.java @@ -11,11 +11,16 @@ package org.opendaylight.controller.cluster.datastore; import akka.actor.ActorRef; import akka.actor.PoisonPill; import akka.actor.Props; -import akka.actor.UntypedActor; import akka.event.Logging; import akka.event.LoggingAdapter; import akka.japi.Creator; + +import com.google.common.util.concurrent.FutureCallback; +import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; +import org.opendaylight.controller.cluster.common.actor.AbstractUntypedActor; + +import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats; import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction; import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply; import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction; @@ -27,47 +32,47 @@ import org.opendaylight.controller.cluster.datastore.messages.PreCommitTransacti import org.opendaylight.controller.cluster.datastore.modification.CompositeModification; import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort; -import java.util.concurrent.ExecutionException; - -public class ThreePhaseCommitCohort extends UntypedActor { +public class ThreePhaseCommitCohort extends AbstractUntypedActor { private final DOMStoreThreePhaseCommitCohort cohort; private final ActorRef shardActor; private final CompositeModification modification; + private final ShardStats shardStats; public ThreePhaseCommitCohort(DOMStoreThreePhaseCommitCohort cohort, - ActorRef shardActor, CompositeModification modification) { + ActorRef shardActor, CompositeModification modification, ShardStats shardStats) { this.cohort = cohort; this.shardActor = shardActor; this.modification = modification; + this.shardStats = shardStats; } private final LoggingAdapter log = Logging.getLogger(getContext().system(), this); public static Props props(final DOMStoreThreePhaseCommitCohort cohort, - final ActorRef shardActor, final CompositeModification modification) { - return Props.create(new Creator() { - @Override - public ThreePhaseCommitCohort create() throws Exception { - return new ThreePhaseCommitCohort(cohort, shardActor, - modification); - } - }); + final ActorRef shardActor, final CompositeModification modification, + ShardStats shardStats) { + return Props.create(new ThreePhaseCommitCohortCreator(cohort, shardActor, modification, + shardStats)); } @Override - public void onReceive(Object message) throws Exception { - log.debug("Received message {}", message); - - if (message instanceof CanCommitTransaction) { - canCommit((CanCommitTransaction) message); - } else if (message instanceof PreCommitTransaction) { - preCommit((PreCommitTransaction) message); - } else if (message instanceof CommitTransaction) { - commit((CommitTransaction) message); - } else if (message instanceof AbortTransaction) { - abort((AbortTransaction) message); + public void handleReceive(Object message) throws Exception { + if (message.getClass() + .equals(CanCommitTransaction.SERIALIZABLE_CLASS)) { + canCommit(new CanCommitTransaction()); + } else if (message.getClass() + .equals(PreCommitTransaction.SERIALIZABLE_CLASS)) { + preCommit(new PreCommitTransaction()); + } else if (message.getClass() + .equals(CommitTransaction.SERIALIZABLE_CLASS)) { + commit(new CommitTransaction()); + } else if (message.getClass() + .equals(AbortTransaction.SERIALIZABLE_CLASS)) { + abort(new AbortTransaction()); + } else { + unknownMessage(message); } } @@ -76,22 +81,29 @@ public class ThreePhaseCommitCohort extends UntypedActor { final ActorRef sender = getSender(); final ActorRef self = getSelf(); - future.addListener(new Runnable() { + Futures.addCallback(future, new FutureCallback() { @Override - public void run() { - try { - future.get(); - sender.tell(new AbortTransactionReply(), self); - } catch (InterruptedException | ExecutionException e) { - log.error(e, "An exception happened when aborting"); - } + public void onSuccess(Void v) { + shardStats.incrementAbortTransactionsCount(); + sender + .tell(new AbortTransactionReply().toSerializable(), + self); } - }, getContext().dispatcher()); + + @Override + public void onFailure(Throwable t) { + LOG.error(t, "An exception happened during abort"); + sender + .tell(new akka.actor.Status.Failure(t), self); + } + }); } private void commit(CommitTransaction message) { // Forward the commit to the shard - log.debug("Forward commit transaction to Shard {} ", shardActor); + if(log.isDebugEnabled()) { + log.debug("Forward commit transaction to Shard {} ", shardActor); + } shardActor.forward(new ForwardedCommitTransaction(cohort, modification), getContext()); @@ -103,18 +115,21 @@ public class ThreePhaseCommitCohort extends UntypedActor { final ListenableFuture future = cohort.preCommit(); final ActorRef sender = getSender(); final ActorRef self = getSelf(); + Futures.addCallback(future, new FutureCallback() { + @Override + public void onSuccess(Void v) { + sender + .tell(new PreCommitTransactionReply().toSerializable(), + self); + } - future.addListener(new Runnable() { @Override - public void run() { - try { - future.get(); - sender.tell(new PreCommitTransactionReply(), self); - } catch (InterruptedException | ExecutionException e) { - log.error(e, "An exception happened when preCommitting"); - } + public void onFailure(Throwable t) { + LOG.error(t, "An exception happened during pre-commit"); + sender + .tell(new akka.actor.Status.Failure(t), self); } - }, getContext().dispatcher()); + }); } @@ -122,18 +137,39 @@ public class ThreePhaseCommitCohort extends UntypedActor { final ListenableFuture future = cohort.canCommit(); final ActorRef sender = getSender(); final ActorRef self = getSelf(); + Futures.addCallback(future, new FutureCallback() { + @Override + public void onSuccess(Boolean canCommit) { + sender.tell(new CanCommitTransactionReply(canCommit) + .toSerializable(), self); + } - future.addListener(new Runnable() { @Override - public void run() { - try { - Boolean canCommit = future.get(); - sender.tell(new CanCommitTransactionReply(canCommit), self); - } catch (InterruptedException | ExecutionException e) { - log.error(e, "An exception happened when aborting"); - } + public void onFailure(Throwable t) { + LOG.error(t, "An exception happened during canCommit"); + sender + .tell(new akka.actor.Status.Failure(t), self); } - }, getContext().dispatcher()); + }); + } + private static class ThreePhaseCommitCohortCreator implements Creator { + final DOMStoreThreePhaseCommitCohort cohort; + final ActorRef shardActor; + final CompositeModification modification; + final ShardStats shardStats; + + ThreePhaseCommitCohortCreator(DOMStoreThreePhaseCommitCohort cohort, + ActorRef shardActor, CompositeModification modification, ShardStats shardStats) { + this.cohort = cohort; + this.shardActor = shardActor; + this.modification = modification; + this.shardStats = shardStats; + } + + @Override + public ThreePhaseCommitCohort create() throws Exception { + return new ThreePhaseCommitCohort(cohort, shardActor, modification, shardStats); + } } }