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%2FCohortEntry.java;h=de325279b7a1aa252ca403e7d8034b981486efb5;hb=118cd0216b0c6b0ec1a01689ec2025a13e090861;hp=073f0814c0834058ffc5d715d016605954913635;hpb=4d1709660b7af992d4c382a2a38debb5c7d64fb9;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/CohortEntry.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/CohortEntry.java index 073f0814c0..de325279b7 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/CohortEntry.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/CohortEntry.java @@ -7,82 +7,67 @@ */ package org.opendaylight.controller.cluster.datastore; +import static com.google.common.base.Preconditions.checkState; +import static java.util.Objects.requireNonNull; + import akka.actor.ActorRef; -import akka.util.Timeout; -import com.google.common.base.Preconditions; -import com.google.common.base.Stopwatch; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; +import com.google.common.primitives.UnsignedLong; +import com.google.common.util.concurrent.FutureCallback; +import java.util.List; +import java.util.Optional; +import java.util.SortedSet; import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier; import org.opendaylight.controller.cluster.datastore.ShardCommitCoordinator.CohortDecorator; import org.opendaylight.controller.cluster.datastore.modification.Modification; -import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate; -import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification; -import org.opendaylight.yangtools.yang.model.api.SchemaContext; -import scala.concurrent.duration.Duration; +import org.opendaylight.yangtools.yang.common.Empty; +import org.opendaylight.yangtools.yang.data.tree.api.DataTreeCandidate; +import org.opendaylight.yangtools.yang.data.tree.api.DataTreeModification; final class CohortEntry { - enum State { - PENDING, - CAN_COMMITTED, - PRE_COMMITTED, - COMMITTED, - ABORTED - } - - private static final Timeout COMMIT_STEP_TIMEOUT = new Timeout(Duration.create(5, TimeUnit.SECONDS)); - - private final Stopwatch lastAccessTimer = Stopwatch.createStarted(); private final ReadWriteShardDataTreeTransaction transaction; - private final TransactionIdentifier transactionID; - private final CompositeDataTreeCohort userCohorts; + private final TransactionIdentifier transactionId; private final short clientVersion; - private State state = State.PENDING; private RuntimeException lastBatchedModificationsException; private int totalBatchedModificationsReceived; + private int totalOperationsProcessed; private ShardDataTreeCohort cohort; private boolean doImmediateCommit; private ActorRef replySender; private Shard shard; - CohortEntry(TransactionIdentifier transactionID, ReadWriteShardDataTreeTransaction transaction, - DataTreeCohortActorRegistry cohortRegistry, SchemaContext schema, short clientVersion) { - this.transaction = Preconditions.checkNotNull(transaction); - this.transactionID = Preconditions.checkNotNull(transactionID); + private CohortEntry(final ReadWriteShardDataTreeTransaction transaction, final short clientVersion) { + cohort = null; + this.transaction = requireNonNull(transaction); + transactionId = transaction.getIdentifier(); this.clientVersion = clientVersion; - this.userCohorts = new CompositeDataTreeCohort(cohortRegistry, transactionID, schema, COMMIT_STEP_TIMEOUT); } - CohortEntry(TransactionIdentifier transactionID, ShardDataTreeCohort cohort, DataTreeCohortActorRegistry cohortRegistry, - SchemaContext schema, short clientVersion) { - this.transactionID = Preconditions.checkNotNull(transactionID); - this.cohort = cohort; - this.transaction = null; + private CohortEntry(final ShardDataTreeCohort cohort, final short clientVersion) { + this.cohort = requireNonNull(cohort); + transactionId = cohort.getIdentifier(); + transaction = null; this.clientVersion = clientVersion; - this.userCohorts = new CompositeDataTreeCohort(cohortRegistry, transactionID, schema, COMMIT_STEP_TIMEOUT); } - void updateLastAccessTime() { - lastAccessTimer.reset(); - lastAccessTimer.start(); + static CohortEntry createOpen(final ReadWriteShardDataTreeTransaction transaction, final short clientVersion) { + return new CohortEntry(transaction, clientVersion); } - TransactionIdentifier getTransactionID() { - return transactionID; + static CohortEntry createReady(final ShardDataTreeCohort cohort, final short clientVersion) { + return new CohortEntry(cohort, clientVersion); } - short getClientVersion() { - return clientVersion; + TransactionIdentifier getTransactionId() { + return transactionId; } - State getState() { - return state; + short getClientVersion() { + return clientVersion; } - DataTreeCandidate getCandidate() { - return cohort.getCandidate(); + boolean isFailed() { + return cohort != null && cohort.isFailed(); } DataTreeModification getDataTreeModification() { @@ -97,82 +82,70 @@ final class CohortEntry { return totalBatchedModificationsReceived; } + int getTotalOperationsProcessed() { + return totalOperationsProcessed; + } + RuntimeException getLastBatchedModificationsException() { return lastBatchedModificationsException; } - void applyModifications(Iterable modifications) { + @SuppressWarnings("checkstyle:IllegalCatch") + void applyModifications(final List modifications) { totalBatchedModificationsReceived++; - if(lastBatchedModificationsException == null) { + if (lastBatchedModificationsException == null) { + totalOperationsProcessed += modifications.size(); for (Modification modification : modifications) { - try { - modification.apply(transaction.getSnapshot()); - } catch (RuntimeException e) { - lastBatchedModificationsException = e; - throw e; - } + try { + modification.apply(transaction.getSnapshot()); + } catch (RuntimeException e) { + lastBatchedModificationsException = e; + throw e; + } } } } - boolean canCommit() throws InterruptedException, ExecutionException { - state = State.CAN_COMMITTED; - - // We block on the future here (and also preCommit(), commit(), abort()) so we don't have to worry - // about possibly accessing our state on a different thread outside of our dispatcher. - // TODO: the ShardDataTreeCohort returns immediate Futures anyway which begs the question - why - // bother even returning Futures from ShardDataTreeCohort if we have to treat them synchronously - // anyway?. The Futures are really a remnant from when we were using the InMemoryDataBroker. - return cohort.canCommit().get(); + void canCommit(final FutureCallback callback) { + cohort.canCommit(callback); } - - - void preCommit() throws InterruptedException, ExecutionException, TimeoutException { - state = State.PRE_COMMITTED; - cohort.preCommit().get(); - userCohorts.canCommit(cohort.getCandidate()); - userCohorts.preCommit(); + void preCommit(final FutureCallback callback) { + cohort.preCommit(callback); } - void commit() throws InterruptedException, ExecutionException, TimeoutException { - state = State.COMMITTED; - cohort.commit().get(); - userCohorts.commit(); + void commit(final FutureCallback callback) { + cohort.commit(callback); } - void abort() throws InterruptedException, ExecutionException, TimeoutException { - state = State.ABORTED; - cohort.abort().get(); - userCohorts.abort(); + void abort(final FutureCallback callback) { + cohort.abort(callback); } - void ready(CohortDecorator cohortDecorator, boolean doImmediateCommit) { - Preconditions.checkState(cohort == null, "cohort was already set"); + void ready(final Optional> participatingShardNames, final CohortDecorator cohortDecorator) { + checkState(cohort == null, "cohort was already set"); - setDoImmediateCommit(doImmediateCommit); + cohort = transaction.ready(participatingShardNames); - cohort = transaction.ready(); - - if(cohortDecorator != null) { + if (cohortDecorator != null) { // Call the hook for unit tests. - cohort = cohortDecorator.decorate(transactionID, cohort); + cohort = cohortDecorator.decorate(transactionId, cohort); } } - boolean isReadyToCommit() { - return replySender != null; + boolean isSealed() { + return cohort != null; } - boolean isExpired(long expireTimeInMillis) { - return lastAccessTimer.elapsed(TimeUnit.MILLISECONDS) >= expireTimeInMillis; + Optional> getParticipatingShardNames() { + return cohort != null ? cohort.getParticipatingShardNames() : Optional.empty(); } boolean isDoImmediateCommit() { return doImmediateCommit; } - void setDoImmediateCommit(boolean doImmediateCommit) { + void setDoImmediateCommit(final boolean doImmediateCommit) { this.doImmediateCommit = doImmediateCommit; } @@ -180,7 +153,7 @@ final class CohortEntry { return replySender; } - void setReplySender(ActorRef replySender) { + void setReplySender(final ActorRef replySender) { this.replySender = replySender; } @@ -188,20 +161,15 @@ final class CohortEntry { return shard; } - void setShard(Shard shard) { + void setShard(final Shard shard) { this.shard = shard; } - - boolean isAborted() { - return state == State.ABORTED; - } - @Override public String toString() { final StringBuilder builder = new StringBuilder(); - builder.append("CohortEntry [transactionID=").append(transactionID).append(", doImmediateCommit=") + builder.append("CohortEntry [transactionId=").append(transactionId).append(", doImmediateCommit=") .append(doImmediateCommit).append("]"); return builder.toString(); } -} \ No newline at end of file +}