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%2FDataTreeCohortActorRegistry.java;h=3ff6a9f0e61ba90e7e17a9f2a6b0bf34a7a3a818;hb=3859df9beca8f13f1ff2b2744ed3470a1715bec3;hp=ec7e2ee61b93876094f168b6ced9661113a020c7;hpb=1d3c54640b9fff649fe8d0f57e20d56f8f936cc1;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DataTreeCohortActorRegistry.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DataTreeCohortActorRegistry.java index ec7e2ee61b..3ff6a9f0e6 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DataTreeCohortActorRegistry.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DataTreeCohortActorRegistry.java @@ -5,26 +5,29 @@ * terms of the Eclipse Public License v1.0 which accompanies this distribution, * and is available at http://www.eclipse.org/legal/epl-v10.html */ - package org.opendaylight.controller.cluster.datastore; +import static java.util.Objects.requireNonNull; + import akka.actor.ActorRef; import akka.actor.PoisonPill; import akka.actor.Status; -import com.google.common.base.Preconditions; +import akka.util.Timeout; +import com.google.common.collect.ArrayListMultimap; +import com.google.common.collect.Multimap; import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; import java.util.List; import java.util.Map; -import javax.annotation.concurrent.NotThreadSafe; -import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.CanCommit; -import org.opendaylight.controller.md.sal.dom.spi.AbstractRegistrationTree; -import org.opendaylight.controller.md.sal.dom.spi.RegistrationTreeNode; -import org.opendaylight.controller.md.sal.dom.spi.RegistrationTreeSnapshot; +import java.util.concurrent.Executor; +import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier; import org.opendaylight.mdsal.common.api.LogicalDatastoreType; import org.opendaylight.mdsal.dom.api.DOMDataTreeCandidate; import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier; +import org.opendaylight.mdsal.dom.spi.AbstractRegistrationTree; +import org.opendaylight.mdsal.dom.spi.RegistrationTreeNode; +import org.opendaylight.mdsal.dom.spi.RegistrationTreeSnapshot; import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument; import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate; @@ -36,18 +39,21 @@ import org.slf4j.LoggerFactory; /** * Registry of user commit cohorts, which is responsible for handling registration and calculation - * of affected cohorts based on {@link DataTreeCandidate}. + * of affected cohorts based on {@link DataTreeCandidate}. This class is NOT thread-safe. * */ -@NotThreadSafe class DataTreeCohortActorRegistry extends AbstractRegistrationTree { private static final Logger LOG = LoggerFactory.getLogger(DataTreeCohortActorRegistry.class); private final Map> cohortToNode = new HashMap<>(); + Collection getCohortActors() { + return new ArrayList<>(cohortToNode.keySet()); + } - void registerCohort(ActorRef sender, RegisterCohort cohort) { + @SuppressWarnings("checkstyle:IllegalCatch") + void registerCohort(final ActorRef sender, final RegisterCohort cohort) { takeLock(); try { final ActorRef cohortRef = cohort.getCohort(); @@ -64,7 +70,7 @@ class DataTreeCohortActorRegistry extends AbstractRegistrationTree { sender.tell(new Status.Success(null), ActorRef.noSender()); } - void removeCommitCohort(ActorRef sender, RemoveCohort message) { + void removeCommitCohort(final ActorRef sender, final RemoveCohort message) { final ActorRef cohort = message.getCohort(); final RegistrationTreeNode node = cohortToNode.get(cohort); if (node != null) { @@ -75,14 +81,14 @@ class DataTreeCohortActorRegistry extends AbstractRegistrationTree { cohort.tell(PoisonPill.getInstance(), cohort); } - Collection createCanCommitMessages(String txId, DataTreeCandidate candidate, - SchemaContext schema) { + List createCanCommitMessages(final TransactionIdentifier txId, + final DataTreeCandidate candidate, final SchemaContext schema) { try (RegistrationTreeSnapshot cohorts = takeSnapshot()) { return new CanCommitMessageBuilder(txId, candidate, schema).perform(cohorts.getRootNode()); } } - void process(ActorRef sender, CohortRegistryCommand message) { + void process(final ActorRef sender, final CohortRegistryCommand message) { if (message instanceof RegisterCohort) { registerCohort(sender, (RegisterCohort) message); } else if (message instanceof RemoveCohort) { @@ -90,12 +96,11 @@ class DataTreeCohortActorRegistry extends AbstractRegistrationTree { } } - static abstract class CohortRegistryCommand { - + abstract static class CohortRegistryCommand { private final ActorRef cohort; - CohortRegistryCommand(ActorRef cohort) { - this.cohort = Preconditions.checkNotNull(cohort); + CohortRegistryCommand(final ActorRef cohort) { + this.cohort = requireNonNull(cohort); } ActorRef getCohort() { @@ -104,45 +109,39 @@ class DataTreeCohortActorRegistry extends AbstractRegistrationTree { } static class RegisterCohort extends CohortRegistryCommand { - private final DOMDataTreeIdentifier path; - RegisterCohort(DOMDataTreeIdentifier path, ActorRef cohort) { + RegisterCohort(final DOMDataTreeIdentifier path, final ActorRef cohort) { super(cohort); this.path = path; - } public DOMDataTreeIdentifier getPath() { return path; } - } static class RemoveCohort extends CohortRegistryCommand { - - RemoveCohort(ActorRef cohort) { + RemoveCohort(final ActorRef cohort) { super(cohort); } - } private static class CanCommitMessageBuilder { - - private final String txId; + private final Multimap actorToCandidates = ArrayListMultimap.create(); + private final TransactionIdentifier txId; private final DataTreeCandidate candidate; private final SchemaContext schema; - private final Collection messages = - new ArrayList<>(); - CanCommitMessageBuilder(String txId, DataTreeCandidate candidate, SchemaContext schema) { - this.txId = Preconditions.checkNotNull(txId); - this.candidate = Preconditions.checkNotNull(candidate); + CanCommitMessageBuilder(final TransactionIdentifier txId, final DataTreeCandidate candidate, + final SchemaContext schema) { + this.txId = requireNonNull(txId); + this.candidate = requireNonNull(candidate); this.schema = schema; } - private void lookupAndCreateCanCommits(List args, int offset, - RegistrationTreeNode node) { + private void lookupAndCreateCanCommits(final List args, final int offset, + final RegistrationTreeNode node) { if (args.size() != offset) { final PathArgument arg = args.get(offset); @@ -158,8 +157,8 @@ class DataTreeCohortActorRegistry extends AbstractRegistrationTree { } } - private void lookupAndCreateCanCommits(YangInstanceIdentifier path, RegistrationTreeNode regNode, - DataTreeCandidateNode candNode) { + private void lookupAndCreateCanCommits(final YangInstanceIdentifier path, + final RegistrationTreeNode regNode, final DataTreeCandidateNode candNode) { if (candNode.getModificationType() == ModificationType.UNMODIFIED) { LOG.debug("Skipping unmodified candidate {}", path); return; @@ -185,24 +184,34 @@ class DataTreeCohortActorRegistry extends AbstractRegistrationTree { } } - private void createCanCommits(Collection regs, YangInstanceIdentifier path, - DataTreeCandidateNode node) { - final DOMDataTreeCandidate candidate = DOMDataTreeCandidateTO.create(treeIdentifier(path), node); + private void createCanCommits(final Collection regs, final YangInstanceIdentifier path, + final DataTreeCandidateNode node) { + final DOMDataTreeCandidate domCandidate = DOMDataTreeCandidateTO.create(treeIdentifier(path), node); for (final ActorRef reg : regs) { - final CanCommit message = new DataTreeCohortActor.CanCommit(txId, candidate, schema, reg); - messages.add(message); + actorToCandidates.put(reg, domCandidate); } } - private static DOMDataTreeIdentifier treeIdentifier(YangInstanceIdentifier path) { + private static DOMDataTreeIdentifier treeIdentifier(final YangInstanceIdentifier path) { return new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, path); } - private Collection perform(RegistrationTreeNode rootNode) { + List perform(final RegistrationTreeNode rootNode) { final List toLookup = candidate.getRootPath().getPathArguments(); lookupAndCreateCanCommits(toLookup, 0, rootNode); + + final Map> mapView = actorToCandidates.asMap(); + final List messages = new ArrayList<>(mapView.size()); + for (Map.Entry> entry: mapView.entrySet()) { + messages.add(new DataTreeCohortActor.CanCommit(txId, entry.getValue(), schema, entry.getKey())); + } + return messages; } } + CompositeDataTreeCohort createCohort(final SchemaContext schemaContext, final TransactionIdentifier txId, + final Executor callbackExecutor, final Timeout commitStepTimeout) { + return new CompositeDataTreeCohort(this, txId, schemaContext, callbackExecutor, commitStepTimeout); + } }