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=e3ae5dac7b7950f3fc300a3189bc83922df24033;hp=61baf1ab64421e04f76d52ec684709ca33f38d25;hb=5834673a28941e470173b5215f56cebdf0cfa2f0;hpb=81aa5072801e6453306e296b91dba3dbeeaf046d 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 61baf1ab64..e3ae5dac7b 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 @@ -9,12 +9,18 @@ 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; @@ -26,103 +32,142 @@ 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{ - private final DOMStoreThreePhaseCommitCohort cohort; - private final ActorRef shardActor; - private final CompositeModification modification; - - public ThreePhaseCommitCohort(DOMStoreThreePhaseCommitCohort cohort, ActorRef shardActor, CompositeModification modification) { - this.cohort = cohort; - this.shardActor = shardActor; - this.modification = modification; - } - - 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); - } - }); - } - - @Override - public void onReceive(Object message) throws Exception { - 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 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, ShardStats shardStats) { + + this.cohort = cohort; + this.shardActor = shardActor; + this.modification = modification; + this.shardStats = shardStats; } - } - - private void abort(AbortTransaction message) { - final ListenableFuture future = cohort.abort(); - final ActorRef sender = getSender(); - final ActorRef self = getSelf(); - - future.addListener(new Runnable() { - @Override - public void run() { - try { - future.get(); - sender.tell(new AbortTransactionReply(), self); - } catch (InterruptedException | ExecutionException e) { - log.error(e, "An exception happened when aborting"); - } - } - }, getContext().dispatcher()); - } - - private void commit(CommitTransaction message) { - // Forward the commit to the shard - log.info("Commit transaction now + " + shardActor); - shardActor.forward(new ForwardedCommitTransaction(cohort, modification), getContext()); - - } - - private void preCommit(PreCommitTransaction message) { - final ListenableFuture future = cohort.preCommit(); - final ActorRef sender = getSender(); - final ActorRef self = getSelf(); - - 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"); + + private final LoggingAdapter log = + Logging.getLogger(getContext().system(), this); + + public static Props props(final DOMStoreThreePhaseCommitCohort cohort, + final ActorRef shardActor, final CompositeModification modification, + ShardStats shardStats) { + return Props.create(new ThreePhaseCommitCohortCreator(cohort, shardActor, modification, + shardStats)); + } + + @Override + 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); } - } - }, getContext().dispatcher()); - - } - - private void canCommit(CanCommitTransaction message) { - final ListenableFuture future = cohort.canCommit(); - final ActorRef sender = getSender(); - final ActorRef self = getSelf(); - - 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"); + } + + private void abort(AbortTransaction message) { + final ListenableFuture future = cohort.abort(); + final ActorRef sender = getSender(); + final ActorRef self = getSelf(); + + Futures.addCallback(future, new FutureCallback() { + @Override + public void onSuccess(Void v) { + shardStats.incrementAbortTransactionsCount(); + sender + .tell(new AbortTransactionReply().toSerializable(), + self); + } + + @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); + shardActor.forward(new ForwardedCommitTransaction(cohort, modification), + getContext()); + + getContext().parent().tell(PoisonPill.getInstance(), getSelf()); + + } + + private void preCommit(PreCommitTransaction message) { + 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); + } + + @Override + public void onFailure(Throwable t) { + LOG.error(t, "An exception happened during pre-commit"); + sender + .tell(new akka.actor.Status.Failure(t), self); + } + }); + + } + + private void canCommit(CanCommitTransaction message) { + 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); + } + + @Override + public void onFailure(Throwable t) { + LOG.error(t, "An exception happened during canCommit"); + sender + .tell(new akka.actor.Status.Failure(t), self); + } + }); + } + + 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; } - } - }, getContext().dispatcher()); - } + @Override + public ThreePhaseCommitCohort create() throws Exception { + return new ThreePhaseCommitCohort(cohort, shardActor, modification, shardStats); + } + } }