2 * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
4 * This program and the accompanying materials are made available under the
5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6 * and is available at http://www.eclipse.org/legal/epl-v10.html
8 package org.opendaylight.controller.md.sal.dom.store.impl;
10 import static com.google.common.base.Preconditions.checkNotNull;
11 import static com.google.common.base.Preconditions.checkState;
12 import static org.opendaylight.controller.md.sal.dom.store.impl.StoreUtils.increase;
14 import java.util.Collections;
15 import java.util.concurrent.Callable;
16 import java.util.concurrent.atomic.AtomicLong;
18 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
19 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
20 import org.opendaylight.controller.md.sal.dom.store.impl.tree.ListenerTree;
21 import org.opendaylight.controller.md.sal.dom.store.impl.tree.ModificationType;
22 import org.opendaylight.controller.md.sal.dom.store.impl.tree.NodeModification;
23 import org.opendaylight.controller.md.sal.dom.store.impl.tree.StoreMetadataNode;
24 import org.opendaylight.controller.sal.core.spi.data.DOMStore;
25 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
26 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
27 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
28 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransaction;
29 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
30 import org.opendaylight.yangtools.concepts.AbstractListenerRegistration;
31 import org.opendaylight.yangtools.concepts.Identifiable;
32 import org.opendaylight.yangtools.concepts.ListenerRegistration;
33 import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier;
34 import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier.PathArgument;
35 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
36 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
37 import org.opendaylight.yangtools.yang.model.api.SchemaContextListener;
38 import org.slf4j.Logger;
39 import org.slf4j.LoggerFactory;
41 import com.google.common.base.Objects;
42 import com.google.common.base.Objects.ToStringHelper;
43 import com.google.common.base.Optional;
44 import com.google.common.base.Preconditions;
45 import com.google.common.primitives.UnsignedLong;
46 import com.google.common.util.concurrent.Futures;
47 import com.google.common.util.concurrent.ListenableFuture;
48 import com.google.common.util.concurrent.ListeningExecutorService;
50 public class InMemoryDOMDataStore implements DOMStore, Identifiable<String>, SchemaContextListener {
52 private static final Logger LOG = LoggerFactory.getLogger(InMemoryDOMDataStore.class);
53 private static final InstanceIdentifier PUBLIC_ROOT_PATH = InstanceIdentifier.builder().build();
55 private final ListeningExecutorService executor;
56 private final String name;
57 private final AtomicLong txCounter = new AtomicLong(0);
58 private final ListenerTree listenerTree = ListenerTree.create();
59 private final DataTree dataTree = DataTree.create(null);
60 private ModificationApplyOperation operationTree = new AlwaysFailOperation();
62 public InMemoryDOMDataStore(final String name, final ListeningExecutorService executor) {
63 this.name = Preconditions.checkNotNull(name);
64 this.executor = Preconditions.checkNotNull(executor);
68 public final String getIdentifier() {
73 public DOMStoreReadTransaction newReadOnlyTransaction() {
74 return new SnapshotBackedReadTransaction(nextIdentifier(), dataTree.takeSnapshot());
78 public DOMStoreReadWriteTransaction newReadWriteTransaction() {
79 return new SnapshotBackedReadWriteTransaction(nextIdentifier(), dataTree.takeSnapshot(), this, operationTree);
83 public DOMStoreWriteTransaction newWriteOnlyTransaction() {
84 return new SnapshotBackedWriteTransaction(nextIdentifier(), dataTree.takeSnapshot(), this, operationTree);
88 public synchronized void onGlobalContextUpdated(final SchemaContext ctx) {
90 * Order of operations is important: dataTree may reject the context
91 * and creation of ModificationApplyOperation may fail. So pre-construct
92 * the operation, then update the data tree and then move the operation
95 final ModificationApplyOperation newOperationTree = SchemaAwareApplyOperationRoot.from(ctx);
96 dataTree.setSchemaContext(ctx);
97 operationTree = newOperationTree;
101 public <L extends AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>> ListenerRegistration<L> registerChangeListener(
102 final InstanceIdentifier path, final L listener, final DataChangeScope scope) {
105 * Make sure commit is not occurring right now. Listener has to be
106 * registered and its state capture enqueued at a consistent point.
108 * FIXME: improve this to read-write lock, such that multiple listener
109 * registrations can occur simultaneously
111 final DataChangeListenerRegistration<L> reg;
112 synchronized (this) {
113 LOG.debug("{}: Registering data change listener {} for {}", name, listener, path);
115 reg = listenerTree.registerDataChangeListener(path, listener, scope);
117 Optional<NormalizedNode<?, ?>> currentState = dataTree.takeSnapshot().readNode(path);
118 if (currentState.isPresent()) {
119 final NormalizedNode<?, ?> data = currentState.get();
121 final DOMImmutableDataChangeEvent event = DOMImmutableDataChangeEvent.builder(DataChangeScope.BASE) //
123 .addCreated(path, data) //
125 executor.submit(new ChangeListenerNotifyTask(Collections.singletonList(reg), event));
129 return new AbstractListenerRegistration<L>(listener) {
131 protected void removeRegistration() {
132 synchronized (InMemoryDOMDataStore.this) {
139 private synchronized DOMStoreThreePhaseCommitCohort submit(final SnapshotBackedWriteTransaction writeTx) {
140 LOG.debug("Tx: {} is submitted. Modifications: {}", writeTx.getIdentifier(), writeTx.getMutatedView());
141 return new ThreePhaseCommitImpl(writeTx);
144 private Object nextIdentifier() {
145 return name + "-" + txCounter.getAndIncrement();
148 private void commit(final DataTree.Snapshot currentSnapshot, final StoreMetadataNode newDataTree,
149 final ResolveDataChangeEventsTask listenerResolver) {
150 LOG.debug("Updating Store snaphot version: {} with version:{}", currentSnapshot, newDataTree.getSubtreeVersion());
152 if (LOG.isTraceEnabled()) {
153 LOG.trace("Data Tree is {}", StoreUtils.toStringTree(newDataTree.getData()));
157 * The commit has to occur atomically with regard to listener
160 synchronized (this) {
161 dataTree.commitSnapshot(currentSnapshot, newDataTree);
163 for (ChangeListenerNotifyTask task : listenerResolver.call()) {
164 LOG.trace("Scheduling invocation of listeners: {}", task);
165 executor.submit(task);
170 private static abstract class AbstractDOMStoreTransaction implements DOMStoreTransaction {
171 private final Object identifier;
173 protected AbstractDOMStoreTransaction(final Object identifier) {
174 this.identifier = identifier;
178 public final Object getIdentifier() {
183 public final String toString() {
184 return addToStringAttributes(Objects.toStringHelper(this)).toString();
188 * Add class-specific toString attributes.
190 * @param toStringHelper
191 * ToStringHelper instance
192 * @return ToStringHelper instance which was passed in
194 protected ToStringHelper addToStringAttributes(final ToStringHelper toStringHelper) {
195 return toStringHelper.add("id", identifier);
199 private static final class SnapshotBackedReadTransaction extends AbstractDOMStoreTransaction implements
200 DOMStoreReadTransaction {
201 private DataTree.Snapshot stableSnapshot;
203 public SnapshotBackedReadTransaction(final Object identifier, final DataTree.Snapshot snapshot) {
205 this.stableSnapshot = Preconditions.checkNotNull(snapshot);
206 LOG.debug("ReadOnly Tx: {} allocated with snapshot {}", identifier, snapshot);
210 public void close() {
211 LOG.debug("Store transaction: {} : Closed", getIdentifier());
212 stableSnapshot = null;
216 public ListenableFuture<Optional<NormalizedNode<?, ?>>> read(final InstanceIdentifier path) {
217 checkNotNull(path, "Path must not be null.");
218 checkState(stableSnapshot != null, "Transaction is closed");
219 return Futures.immediateFuture(stableSnapshot.readNode(path));
223 private static class SnapshotBackedWriteTransaction extends AbstractDOMStoreTransaction implements
224 DOMStoreWriteTransaction {
225 private MutableDataTree mutableTree;
226 private InMemoryDOMDataStore store;
227 private boolean ready = false;
229 public SnapshotBackedWriteTransaction(final Object identifier, final DataTree.Snapshot snapshot,
230 final InMemoryDOMDataStore store, final ModificationApplyOperation applyOper) {
232 mutableTree = MutableDataTree.from(snapshot, applyOper);
234 LOG.debug("Write Tx: {} allocated with snapshot {}", identifier, snapshot);
238 public void close() {
239 LOG.debug("Store transaction: {} : Closed", getIdentifier());
240 this.mutableTree = null;
245 public void write(final InstanceIdentifier path, final NormalizedNode<?, ?> data) {
248 LOG.trace("Tx: {} Write: {}:{}", getIdentifier(), path, data);
249 mutableTree.write(path, data);
250 // FIXME: Add checked exception
251 } catch (Exception e) {
252 LOG.error("Tx: {}, failed to write {}:{} in {}", getIdentifier(), path, data, mutableTree, e);
257 public void merge(final InstanceIdentifier path, final NormalizedNode<?, ?> data) {
260 LOG.trace("Tx: {} Merge: {}:{}", getIdentifier(), path, data);
261 mutableTree.merge(path, data);
262 // FIXME: Add checked exception
263 } catch (Exception e) {
264 LOG.error("Tx: {}, failed to write {}:{} in {}", getIdentifier(), path, data, mutableTree, e);
269 public void delete(final InstanceIdentifier path) {
272 LOG.trace("Tx: {} Delete: {}", getIdentifier(), path);
273 mutableTree.delete(path);
274 // FIXME: Add checked exception
275 } catch (Exception e) {
276 LOG.error("Tx: {}, failed to delete {} in {}", getIdentifier(), path, mutableTree, e);
280 protected final boolean isReady() {
284 protected final void checkNotReady() {
285 checkState(!ready, "Transaction %s is ready. No further modifications allowed.", getIdentifier());
289 public synchronized DOMStoreThreePhaseCommitCohort ready() {
290 checkState(!ready, "Transaction %s is already ready.", getIdentifier());
293 LOG.debug("Store transaction: {} : Ready", getIdentifier());
295 return store.submit(this);
298 protected MutableDataTree getMutatedView() {
303 protected ToStringHelper addToStringAttributes(final ToStringHelper toStringHelper) {
304 return toStringHelper.add("ready", isReady());
308 private static class SnapshotBackedReadWriteTransaction extends SnapshotBackedWriteTransaction implements
309 DOMStoreReadWriteTransaction {
311 protected SnapshotBackedReadWriteTransaction(final Object identifier, final DataTree.Snapshot snapshot,
312 final InMemoryDOMDataStore store, final ModificationApplyOperation applyOper) {
313 super(identifier, snapshot, store, applyOper);
317 public ListenableFuture<Optional<NormalizedNode<?, ?>>> read(final InstanceIdentifier path) {
318 LOG.trace("Tx: {} Read: {}", getIdentifier(), path);
320 return Futures.immediateFuture(getMutatedView().read(path));
321 } catch (Exception e) {
322 LOG.error("Tx: {} Failed Read of {}", getIdentifier(), path, e);
328 private class ThreePhaseCommitImpl implements DOMStoreThreePhaseCommitCohort {
330 private final SnapshotBackedWriteTransaction transaction;
331 private final NodeModification modification;
333 private DataTree.Snapshot storeSnapshot;
334 private Optional<StoreMetadataNode> proposedSubtree;
335 private ResolveDataChangeEventsTask listenerResolver;
337 public ThreePhaseCommitImpl(final SnapshotBackedWriteTransaction writeTransaction) {
338 this.transaction = writeTransaction;
339 this.modification = transaction.getMutatedView().getRootModification();
343 public ListenableFuture<Boolean> canCommit() {
344 final DataTree.Snapshot snapshotCapture = dataTree.takeSnapshot();
345 final ModificationApplyOperation snapshotOperation = operationTree;
347 return executor.submit(new Callable<Boolean>() {
350 public Boolean call() throws Exception {
351 Boolean applicable = false;
353 snapshotOperation.checkApplicable(PUBLIC_ROOT_PATH, modification,
354 Optional.of(snapshotCapture.getRootNode()));
356 } catch (DataPreconditionFailedException e) {
357 LOG.warn("Store Tx: {} Data Precondition failed for {}.",transaction.getIdentifier(),e.getPath(),e);
360 LOG.debug("Store Transaction: {} : canCommit : {}", transaction.getIdentifier(), applicable);
367 public ListenableFuture<Void> preCommit() {
368 storeSnapshot = dataTree.takeSnapshot();
369 if (modification.getModificationType() == ModificationType.UNMODIFIED) {
370 return Futures.immediateFuture(null);
372 return executor.submit(new Callable<Void>() {
375 public Void call() throws Exception {
376 StoreMetadataNode metadataTree = storeSnapshot.getRootNode();
378 proposedSubtree = operationTree.apply(modification, Optional.of(metadataTree),
379 increase(metadataTree.getSubtreeVersion()));
381 listenerResolver = ResolveDataChangeEventsTask.create() //
382 .setRootPath(PUBLIC_ROOT_PATH) //
383 .setBeforeRoot(Optional.of(metadataTree)) //
384 .setAfterRoot(proposedSubtree) //
385 .setModificationRoot(modification) //
386 .setListenerRoot(listenerTree);
394 public ListenableFuture<Void> abort() {
395 storeSnapshot = null;
396 proposedSubtree = null;
397 return Futures.<Void> immediateFuture(null);
401 public ListenableFuture<Void> commit() {
402 if (modification.getModificationType() == ModificationType.UNMODIFIED) {
403 return Futures.immediateFuture(null);
406 checkState(proposedSubtree != null, "Proposed subtree must be computed");
407 checkState(storeSnapshot != null, "Proposed subtree must be computed");
408 // return ImmediateFuture<>;
409 InMemoryDOMDataStore.this.commit(storeSnapshot, proposedSubtree.get(), listenerResolver);
410 return Futures.<Void> immediateFuture(null);
415 private static final class AlwaysFailOperation implements ModificationApplyOperation {
418 public Optional<StoreMetadataNode> apply(final NodeModification modification,
419 final Optional<StoreMetadataNode> storeMeta, final UnsignedLong subtreeVersion) {
420 throw new IllegalStateException("Schema Context is not available.");
424 public void checkApplicable(final InstanceIdentifier path,final NodeModification modification, final Optional<StoreMetadataNode> storeMetadata) {
425 throw new IllegalStateException("Schema Context is not available.");
429 public Optional<ModificationApplyOperation> getChild(final PathArgument child) {
430 throw new IllegalStateException("Schema Context is not available.");
434 public void verifyStructure(final NodeModification modification) throws IllegalArgumentException {
435 throw new IllegalStateException("Schema Context is not available.");