X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-dom-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fmd%2Fsal%2Fdom%2Fstore%2Fimpl%2FInMemoryDOMDataStore.java;h=00df6580effda8da33e80a648653935049010c47;hp=4d5b8ee906a07640f5be07bfb8a3173b57a66bb9;hb=617941ac35b64881556f7311a8a296f4b60b2935;hpb=c4940d6fa1f6928b7189afb2dc6964fb2f2cbae2 diff --git a/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/store/impl/InMemoryDOMDataStore.java b/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/store/impl/InMemoryDOMDataStore.java index 4d5b8ee906..00df6580ef 100644 --- a/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/store/impl/InMemoryDOMDataStore.java +++ b/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/store/impl/InMemoryDOMDataStore.java @@ -9,116 +9,172 @@ package org.opendaylight.controller.md.sal.dom.store.impl; import static com.google.common.base.Preconditions.checkNotNull; import static com.google.common.base.Preconditions.checkState; -import static org.opendaylight.controller.md.sal.dom.store.impl.StoreUtils.increase; +import java.util.Collections; import java.util.concurrent.Callable; import java.util.concurrent.atomic.AtomicLong; import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener; -import org.opendaylight.controller.md.sal.dom.store.impl.tree.NodeModification; -import org.opendaylight.controller.md.sal.dom.store.impl.tree.StoreMetadataNode; +import org.opendaylight.controller.md.sal.dom.store.impl.tree.DataPreconditionFailedException; +import org.opendaylight.controller.md.sal.dom.store.impl.tree.DataTree; +import org.opendaylight.controller.md.sal.dom.store.impl.tree.DataTreeCandidate; +import org.opendaylight.controller.md.sal.dom.store.impl.tree.DataTreeModification; +import org.opendaylight.controller.md.sal.dom.store.impl.tree.DataTreeSnapshot; +import org.opendaylight.controller.md.sal.dom.store.impl.tree.ListenerTree; +import org.opendaylight.controller.md.sal.dom.store.impl.tree.data.InMemoryDataTreeFactory; import org.opendaylight.controller.sal.core.spi.data.DOMStore; import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction; import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction; import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort; +import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransaction; import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction; +import org.opendaylight.yangtools.concepts.AbstractListenerRegistration; import org.opendaylight.yangtools.concepts.Identifiable; import org.opendaylight.yangtools.concepts.ListenerRegistration; import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier; -import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier.PathArgument; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; -import org.opendaylight.yangtools.yang.data.impl.schema.NormalizedNodeUtils; import org.opendaylight.yangtools.yang.model.api.SchemaContext; import org.opendaylight.yangtools.yang.model.api.SchemaContextListener; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.google.common.base.Objects; +import com.google.common.base.Objects.ToStringHelper; import com.google.common.base.Optional; -import com.google.common.primitives.UnsignedLong; +import com.google.common.base.Preconditions; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListeningExecutorService; public class InMemoryDOMDataStore implements DOMStore, Identifiable, SchemaContextListener { - private static final Logger LOG = LoggerFactory.getLogger(InMemoryDOMDataStore.class); - + private final DataTree dataTree = InMemoryDataTreeFactory.getInstance().create(); + private final ListenerTree listenerTree = ListenerTree.create(); private final AtomicLong txCounter = new AtomicLong(0); - - private DataAndMetadataSnapshot snapshot; - private ModificationApplyOperation operation; - private final ListeningExecutorService executor; private final String name; - private SchemaContext schemaContext; - public InMemoryDOMDataStore(final String name, final ListeningExecutorService executor) { - this.executor = executor; - this.name = name; - this.operation = new AllwaysFailOperation(); - this.snapshot = DataAndMetadataSnapshot.createEmpty(); + this.name = Preconditions.checkNotNull(name); + this.executor = Preconditions.checkNotNull(executor); } @Override - public String getIdentifier() { + public final String getIdentifier() { return name; } @Override public DOMStoreReadTransaction newReadOnlyTransaction() { - return new SnapshotBackedReadTransaction(nextIdentifier(), snapshot); + return new SnapshotBackedReadTransaction(nextIdentifier(), dataTree.takeSnapshot()); } @Override public DOMStoreReadWriteTransaction newReadWriteTransaction() { - return new SnapshotBackedReadWriteTransaction(nextIdentifier(), snapshot, this, operation); + return new SnapshotBackedReadWriteTransaction(nextIdentifier(), dataTree.takeSnapshot(), this); } @Override public DOMStoreWriteTransaction newWriteOnlyTransaction() { - return new SnaphostBackedWriteTransaction(nextIdentifier(), snapshot, this, operation); + return new SnapshotBackedWriteTransaction(nextIdentifier(), dataTree.takeSnapshot(), this); } @Override public synchronized void onGlobalContextUpdated(final SchemaContext ctx) { - operation = SchemaAwareApplyOperationRoot.from(ctx); - schemaContext = ctx; + dataTree.setSchemaContext(ctx); } @Override public >> ListenerRegistration registerChangeListener( final InstanceIdentifier path, final L listener, final DataChangeScope scope) { - return null; + + /* + * Make sure commit is not occurring right now. Listener has to be + * registered and its state capture enqueued at a consistent point. + * + * FIXME: improve this to read-write lock, such that multiple listener + * registrations can occur simultaneously + */ + final DataChangeListenerRegistration reg; + synchronized (this) { + LOG.debug("{}: Registering data change listener {} for {}", name, listener, path); + + reg = listenerTree.registerDataChangeListener(path, listener, scope); + + Optional> currentState = dataTree.takeSnapshot().readNode(path); + if (currentState.isPresent()) { + final NormalizedNode data = currentState.get(); + + final DOMImmutableDataChangeEvent event = DOMImmutableDataChangeEvent.builder(DataChangeScope.BASE) // + .setAfter(data) // + .addCreated(path, data) // + .build(); + executor.submit(new ChangeListenerNotifyTask(Collections.singletonList(reg), event)); + } + } + + return new AbstractListenerRegistration(listener) { + @Override + protected void removeRegistration() { + synchronized (InMemoryDOMDataStore.this) { + reg.close(); + } + } + }; } - private synchronized DOMStoreThreePhaseCommitCohort submit( - final SnaphostBackedWriteTransaction snaphostBackedWriteTransaction) { - return new ThreePhaseCommitImpl(snaphostBackedWriteTransaction); + private synchronized DOMStoreThreePhaseCommitCohort submit(final SnapshotBackedWriteTransaction writeTx) { + LOG.debug("Tx: {} is submitted. Modifications: {}", writeTx.getIdentifier(), writeTx.getMutatedView()); + return new ThreePhaseCommitImpl(writeTx); } private Object nextIdentifier() { return name + "-" + txCounter.getAndIncrement(); } - private static class SnapshotBackedReadTransaction implements DOMStoreReadTransaction { - - private DataAndMetadataSnapshot stableSnapshot; + private static abstract class AbstractDOMStoreTransaction implements DOMStoreTransaction { private final Object identifier; - public SnapshotBackedReadTransaction(final Object identifier, final DataAndMetadataSnapshot snapshot) { + protected AbstractDOMStoreTransaction(final Object identifier) { this.identifier = identifier; - this.stableSnapshot = snapshot; } @Override - public Object getIdentifier() { + public final Object getIdentifier() { return identifier; } + @Override + public final String toString() { + return addToStringAttributes(Objects.toStringHelper(this)).toString(); + } + + /** + * Add class-specific toString attributes. + * + * @param toStringHelper + * ToStringHelper instance + * @return ToStringHelper instance which was passed in + */ + protected ToStringHelper addToStringAttributes(final ToStringHelper toStringHelper) { + return toStringHelper.add("id", identifier); + } + } + + private static final class SnapshotBackedReadTransaction extends AbstractDOMStoreTransaction implements + DOMStoreReadTransaction { + private DataTreeSnapshot stableSnapshot; + + public SnapshotBackedReadTransaction(final Object identifier, final DataTreeSnapshot snapshot) { + super(identifier); + this.stableSnapshot = Preconditions.checkNotNull(snapshot); + LOG.debug("ReadOnly Tx: {} allocated with snapshot {}", identifier, snapshot); + } + @Override public void close() { + LOG.debug("Store transaction: {} : Closed", getIdentifier()); stableSnapshot = null; } @@ -126,38 +182,27 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch public ListenableFuture>> read(final InstanceIdentifier path) { checkNotNull(path, "Path must not be null."); checkState(stableSnapshot != null, "Transaction is closed"); - return Futures.immediateFuture(NormalizedNodeUtils.findNode(stableSnapshot.getDataTree(), path)); - } - - @Override - public String toString() { - return "SnapshotBackedReadTransaction [id =" + identifier + "]"; + return Futures.immediateFuture(stableSnapshot.readNode(path)); } - } - private static class SnaphostBackedWriteTransaction implements DOMStoreWriteTransaction { - - private MutableDataTree mutableTree; - private final Object identifier; + private static class SnapshotBackedWriteTransaction extends AbstractDOMStoreTransaction implements + DOMStoreWriteTransaction { + private DataTreeModification mutableTree; private InMemoryDOMDataStore store; - private boolean ready = false; - public SnaphostBackedWriteTransaction(final Object identifier, final DataAndMetadataSnapshot snapshot, - final InMemoryDOMDataStore store, final ModificationApplyOperation applyOper) { - this.identifier = identifier; - mutableTree = MutableDataTree.from(snapshot, applyOper); + public SnapshotBackedWriteTransaction(final Object identifier, final DataTreeSnapshot snapshot, + final InMemoryDOMDataStore store) { + super(identifier); + mutableTree = snapshot.newModification(); this.store = store; - } - - @Override - public Object getIdentifier() { - return identifier; + LOG.debug("Write Tx: {} allocated with snapshot {}", identifier, snapshot); } @Override public void close() { + LOG.debug("Store transaction: {} : Closed", getIdentifier()); this.mutableTree = null; this.store = null; } @@ -165,100 +210,124 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch @Override public void write(final InstanceIdentifier path, final NormalizedNode data) { checkNotReady(); - mutableTree.write(path, data); + try { + LOG.trace("Tx: {} Write: {}:{}", getIdentifier(), path, data); + mutableTree.write(path, data); + // FIXME: Add checked exception + } catch (Exception e) { + LOG.error("Tx: {}, failed to write {}:{} in {}", getIdentifier(), path, data, mutableTree, e); + } + } + + @Override + public void merge(final InstanceIdentifier path, final NormalizedNode data) { + checkNotReady(); + try { + LOG.trace("Tx: {} Merge: {}:{}", getIdentifier(), path, data); + mutableTree.merge(path, data); + // FIXME: Add checked exception + } catch (Exception e) { + LOG.error("Tx: {}, failed to write {}:{} in {}", getIdentifier(), path, data, mutableTree, e); + } } @Override public void delete(final InstanceIdentifier path) { checkNotReady(); - mutableTree.delete(path); + try { + LOG.trace("Tx: {} Delete: {}", getIdentifier(), path); + mutableTree.delete(path); + // FIXME: Add checked exception + } catch (Exception e) { + LOG.error("Tx: {}, failed to delete {} in {}", getIdentifier(), path, mutableTree, e); + } } - protected boolean isReady() { + protected final boolean isReady() { return ready; } - protected void checkNotReady() { - checkState(!ready, "Transaction is ready. No further modifications allowed."); + protected final void checkNotReady() { + checkState(!ready, "Transaction %s is ready. No further modifications allowed.", getIdentifier()); } @Override public synchronized DOMStoreThreePhaseCommitCohort ready() { + checkState(!ready, "Transaction %s is already ready.", getIdentifier()); ready = true; - LOG.debug("Store transaction: {} : Ready",getIdentifier()); + + LOG.debug("Store transaction: {} : Ready", getIdentifier()); mutableTree.seal(); return store.submit(this); } - protected MutableDataTree getMutatedView() { + protected DataTreeModification getMutatedView() { return mutableTree; } @Override - public String toString() { - return "SnaphostBackedWriteTransaction [id=" + getIdentifier() + ", ready=" + isReady() + "]"; + protected ToStringHelper addToStringAttributes(final ToStringHelper toStringHelper) { + return toStringHelper.add("ready", isReady()); } - } - private static class SnapshotBackedReadWriteTransaction extends SnaphostBackedWriteTransaction implements - DOMStoreReadWriteTransaction { + private static class SnapshotBackedReadWriteTransaction extends SnapshotBackedWriteTransaction implements + DOMStoreReadWriteTransaction { - protected SnapshotBackedReadWriteTransaction(final Object identifier, final DataAndMetadataSnapshot snapshot, - final InMemoryDOMDataStore store, final ModificationApplyOperation applyOper) { - super(identifier, snapshot, store, applyOper); + protected SnapshotBackedReadWriteTransaction(final Object identifier, final DataTreeSnapshot snapshot, + final InMemoryDOMDataStore store) { + super(identifier, snapshot, store); } @Override public ListenableFuture>> read(final InstanceIdentifier path) { - return Futures.immediateFuture(getMutatedView().read(path)); - } - - @Override - public String toString() { - return "SnapshotBackedReadWriteTransaction [id=" + getIdentifier() + ", ready=" + isReady() + "]"; + LOG.trace("Tx: {} Read: {}", getIdentifier(), path); + try { + return Futures.immediateFuture(getMutatedView().readNode(path)); + } catch (Exception e) { + LOG.error("Tx: {} Failed Read of {}", getIdentifier(), path, e); + throw e; + } } - } private class ThreePhaseCommitImpl implements DOMStoreThreePhaseCommitCohort { - private final SnaphostBackedWriteTransaction transaction; - private final NodeModification modification; + private final SnapshotBackedWriteTransaction transaction; + private final DataTreeModification modification; - private DataAndMetadataSnapshot storeSnapshot; - private Optional proposedSubtree; + private ResolveDataChangeEventsTask listenerResolver; + private DataTreeCandidate candidate; - public ThreePhaseCommitImpl(final SnaphostBackedWriteTransaction writeTransaction) { + public ThreePhaseCommitImpl(final SnapshotBackedWriteTransaction writeTransaction) { this.transaction = writeTransaction; - this.modification = transaction.getMutatedView().getRootModification(); + this.modification = transaction.getMutatedView(); } @Override public ListenableFuture canCommit() { - final DataAndMetadataSnapshot snapshotCapture = snapshot; - final ModificationApplyOperation snapshotOperation = operation; - return executor.submit(new Callable() { - @Override - public Boolean call() throws Exception { - boolean applicable = snapshotOperation.isApplicable(modification, Optional.of(snapshotCapture.getMetadataTree())); - LOG.debug("Store Transcation: {} : canCommit : {}",transaction.getIdentifier(),applicable); - return applicable; + public Boolean call() { + try { + dataTree.validate(modification); + LOG.debug("Store Transaction: {} can be committed", transaction.getIdentifier()); + return true; + } catch (DataPreconditionFailedException e) { + LOG.warn("Store Tx: {} Data Precondition failed for {}.",transaction.getIdentifier(),e.getPath(),e); + return false; + } } }); } @Override public ListenableFuture preCommit() { - storeSnapshot = snapshot; return executor.submit(new Callable() { - @Override - public Void call() throws Exception { - StoreMetadataNode metadataTree = storeSnapshot.getMetadataTree(); - proposedSubtree = operation.apply(modification, Optional.of(metadataTree),increase(metadataTree.getSubtreeVersion())); + public Void call() { + candidate = dataTree.prepare(modification); + listenerResolver = ResolveDataChangeEventsTask.create(candidate, listenerTree); return null; } }); @@ -266,52 +335,28 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch @Override public ListenableFuture abort() { - storeSnapshot = null; - proposedSubtree = null; - return Futures. immediateFuture(null); + candidate = null; + return Futures.immediateFuture(null); } @Override public ListenableFuture commit() { - checkState(proposedSubtree != null); - checkState(storeSnapshot != null); - // return ImmediateFuture<>; - InMemoryDOMDataStore.this.commit(storeSnapshot, proposedSubtree); - return Futures. immediateFuture(null); - } - - } - - private synchronized void commit(final DataAndMetadataSnapshot storeSnapshot, - final Optional proposedSubtree) { - //LOG.info("Updating Store snaphot."); - checkState(snapshot == storeSnapshot, "Store snapshot and transaction snapshot differs"); - snapshot = DataAndMetadataSnapshot.builder().setMetadataTree(proposedSubtree.get()) - .setSchemaContext(schemaContext).build(); - } - - private class AllwaysFailOperation implements ModificationApplyOperation { - - @Override - public Optional apply(final NodeModification modification, - final Optional storeMeta,final UnsignedLong subtreeVersion) { - throw new IllegalStateException("Schema Context is not available."); - } - - @Override - public boolean isApplicable(final NodeModification modification, final Optional storeMetadata) { - throw new IllegalStateException("Schema Context is not available."); - } - - @Override - public Optional getChild(final PathArgument child) { - throw new IllegalStateException("Schema Context is not available."); - } + checkState(candidate != null, "Proposed subtree must be computed"); + + /* + * The commit has to occur atomically with regard to listener + * registrations. + */ + synchronized (this) { + dataTree.commit(candidate); + + for (ChangeListenerNotifyTask task : listenerResolver.call()) { + LOG.trace("Scheduling invocation of listeners: {}", task); + executor.submit(task); + } + } - @Override - public void verifyStructure(final NodeModification modification) throws IllegalArgumentException { - throw new IllegalStateException("Schema Context is not available."); + return Futures.immediateFuture(null); } - } }