X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FThreePhaseCommitCohort.java;h=5a6d0eca5c2a159963febc4ee9d6436f2e864a5d;hb=67d58d1ab50f3c3bbe19a96fb6f0d9d94211829f;hp=e6adfbee660669afd0889995c5e4ec0c755810e9;hpb=85a32885f57edce51d54021a8060be0e1f9aa45b;p=controller.git 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..5a6d0eca5c 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,15 @@ 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.datastore.jmx.mbeans.shard.ShardMBeanFactory; 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 +31,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 String shardName; public ThreePhaseCommitCohort(DOMStoreThreePhaseCommitCohort cohort, - ActorRef shardActor, CompositeModification modification) { + ActorRef shardActor, CompositeModification modification,String shardName) { this.cohort = cohort; this.shardActor = shardActor; this.modification = modification; + this.shardName = shardName; } 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, + String shardName) { + return Props.create(new ThreePhaseCommitCohortCreator(cohort, shardActor, modification, + shardName)); } @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,17 +80,22 @@ 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) { + ShardMBeanFactory.getShardStatsMBean(shardName).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) { @@ -103,18 +112,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 +134,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 String shardName; + + ThreePhaseCommitCohortCreator(DOMStoreThreePhaseCommitCohort cohort, + ActorRef shardActor, CompositeModification modification, String shardName) { + this.cohort = cohort; + this.shardActor = shardActor; + this.modification = modification; + this.shardName = shardName; + } + + @Override + public ThreePhaseCommitCohort create() throws Exception { + return new ThreePhaseCommitCohort(cohort, shardActor, modification, shardName); + } } }