opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/store/impl/InMemoryDOMDataStore.java

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   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
   7  */
   8 package org.opendaylight.controller.md.sal.dom.store.impl;
   9
  10 import static com.google.common.base.Preconditions.checkState;
  11
  12 import java.util.Collections;
  13 import java.util.concurrent.Callable;
  14 import java.util.concurrent.atomic.AtomicLong;
  15
  16 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
  17 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
  18 import org.opendaylight.controller.md.sal.dom.store.impl.SnapshotBackedWriteTransaction.TransactionReadyPrototype;
  19 import org.opendaylight.controller.md.sal.dom.store.impl.tree.DataPreconditionFailedException;
  20 import org.opendaylight.controller.md.sal.dom.store.impl.tree.DataTree;
  21 import org.opendaylight.controller.md.sal.dom.store.impl.tree.DataTreeCandidate;
  22 import org.opendaylight.controller.md.sal.dom.store.impl.tree.DataTreeModification;
  23 import org.opendaylight.controller.md.sal.dom.store.impl.tree.DataTreeSnapshot;
  24 import org.opendaylight.controller.md.sal.dom.store.impl.tree.ListenerTree;
  25 import org.opendaylight.controller.md.sal.dom.store.impl.tree.data.InMemoryDataTreeFactory;
  26 import org.opendaylight.controller.sal.core.spi.data.DOMStore;
  27 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
  28 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
  29 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  30 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
  31 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
  32 import org.opendaylight.yangtools.concepts.AbstractListenerRegistration;
  33 import org.opendaylight.yangtools.concepts.Identifiable;
  34 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  35 import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier;
  36 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  37 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  38 import org.opendaylight.yangtools.yang.model.api.SchemaContextListener;
  39 import org.slf4j.Logger;
  40 import org.slf4j.LoggerFactory;
  41
  42 import com.google.common.base.Optional;
  43 import com.google.common.base.Preconditions;
  44 import com.google.common.util.concurrent.Futures;
  45 import com.google.common.util.concurrent.ListenableFuture;
  46 import com.google.common.util.concurrent.ListeningExecutorService;
  47
  48 /**
  49  * In-memory DOM Data Store
  50  *
  51  * Implementation of {@link DOMStore} which uses {@link DataTree}
  52  * and other classes such as {@link SnapshotBackedWriteTransaction}.
  53  * {@link SnapshotBackedReadTransaction} and {@link ResolveDataChangeEventsTask}
  54  * to implement {@link DOMStore} contract.
  55  *
  56  */
  57 public class InMemoryDOMDataStore implements DOMStore, Identifiable<String>, SchemaContextListener, TransactionReadyPrototype {
  58     private static final Logger LOG = LoggerFactory.getLogger(InMemoryDOMDataStore.class);
  59     private final DataTree dataTree = InMemoryDataTreeFactory.getInstance().create();
  60     private final ListenerTree listenerTree = ListenerTree.create();
  61     private final AtomicLong txCounter = new AtomicLong(0);
  62     private final ListeningExecutorService executor;
  63     private final String name;
  64
  65     public InMemoryDOMDataStore(final String name, final ListeningExecutorService executor) {
  66         this.name = Preconditions.checkNotNull(name);
  67         this.executor = Preconditions.checkNotNull(executor);
  68     }
  69
  70     @Override
  71     public final String getIdentifier() {
  72         return name;
  73     }
  74
  75     @Override
  76     public DOMStoreReadTransaction newReadOnlyTransaction() {
  77         return new SnapshotBackedReadTransaction(nextIdentifier(), dataTree.takeSnapshot());
  78     }
  79
  80     @Override
  81     public DOMStoreReadWriteTransaction newReadWriteTransaction() {
  82         return new SnapshotBackedReadWriteTransaction(nextIdentifier(), dataTree.takeSnapshot(), this);
  83     }
  84
  85     @Override
  86     public DOMStoreWriteTransaction newWriteOnlyTransaction() {
  87         return new SnapshotBackedWriteTransaction(nextIdentifier(), dataTree.takeSnapshot(), this);
  88     }
  89
  90     @Override
  91     public DOMStoreTransactionChain createTransactionChain() {
  92         return new DOMStoreTransactionChainImpl();
  93     }
  94
  95     @Override
  96     public synchronized void onGlobalContextUpdated(final SchemaContext ctx) {
  97         dataTree.setSchemaContext(ctx);
  98     }
  99
 100     @Override
 101     public <L extends AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>> ListenerRegistration<L> registerChangeListener(
 102             final InstanceIdentifier path, final L listener, final DataChangeScope scope) {
 103
 104         /*
 105          * Make sure commit is not occurring right now. Listener has to be
 106          * registered and its state capture enqueued at a consistent point.
 107          *
 108          * FIXME: improve this to read-write lock, such that multiple listener
 109          * registrations can occur simultaneously
 110          */
 111         final DataChangeListenerRegistration<L> reg;
 112         synchronized (this) {
 113             LOG.debug("{}: Registering data change listener {} for {}", name, listener, path);
 114
 115             reg = listenerTree.registerDataChangeListener(path, listener, scope);
 116
 117             Optional<NormalizedNode<?, ?>> currentState = dataTree.takeSnapshot().readNode(path);
 118             if (currentState.isPresent()) {
 119                 final NormalizedNode<?, ?> data = currentState.get();
 120
 121                 final DOMImmutableDataChangeEvent event = DOMImmutableDataChangeEvent.builder(DataChangeScope.BASE) //
 122                         .setAfter(data) //
 123                         .addCreated(path, data) //
 124                         .build();
 125                 executor.submit(new ChangeListenerNotifyTask(Collections.singletonList(reg), event));
 126             }
 127         }
 128
 129         return new AbstractListenerRegistration<L>(listener) {
 130             @Override
 131             protected void removeRegistration() {
 132                 synchronized (InMemoryDOMDataStore.this) {
 133                     reg.close();
 134                 }
 135             }
 136         };
 137     }
 138
 139     @Override
 140     public synchronized DOMStoreThreePhaseCommitCohort ready(final SnapshotBackedWriteTransaction writeTx) {
 141         LOG.debug("Tx: {} is submitted. Modifications: {}", writeTx.getIdentifier(), writeTx.getMutatedView());
 142         return new ThreePhaseCommitImpl(writeTx);
 143     }
 144
 145     private Object nextIdentifier() {
 146         return name + "-" + txCounter.getAndIncrement();
 147     }
 148
 149     private class DOMStoreTransactionChainImpl implements DOMStoreTransactionChain, TransactionReadyPrototype {
 150
 151         private SnapshotBackedWriteTransaction previousOutstandingTx;
 152
 153         @Override
 154         public synchronized DOMStoreReadTransaction newReadOnlyTransaction() {
 155             final DataTreeSnapshot snapshot;
 156             if(previousOutstandingTx != null) {
 157                 checkState(previousOutstandingTx.isReady(), "Previous transaction in chain must be ready.");
 158                 snapshot = previousOutstandingTx.getMutatedView();
 159             } else {
 160                 snapshot = dataTree.takeSnapshot();
 161             }
 162             return new SnapshotBackedReadTransaction(nextIdentifier(), snapshot);
 163         }
 164
 165         @Override
 166         public synchronized DOMStoreReadWriteTransaction newReadWriteTransaction() {
 167             final DataTreeSnapshot snapshot;
 168             if(previousOutstandingTx != null) {
 169                 checkState(previousOutstandingTx.isReady(), "Previous transaction in chain must be ready.");
 170                 snapshot = previousOutstandingTx.getMutatedView();
 171             } else {
 172                 snapshot = dataTree.takeSnapshot().newModification();
 173             }
 174             SnapshotBackedReadWriteTransaction ret = new SnapshotBackedReadWriteTransaction(nextIdentifier(), snapshot,this);
 175             return ret;
 176         }
 177
 178         @Override
 179         public synchronized DOMStoreWriteTransaction newWriteOnlyTransaction() {
 180             final DataTreeSnapshot snapshot;
 181             if(previousOutstandingTx != null) {
 182                 checkState(previousOutstandingTx.isReady(), "Previous transaction in chain must be ready.");
 183                 snapshot = previousOutstandingTx.getMutatedView();
 184             } else {
 185                 snapshot = dataTree.takeSnapshot().newModification();
 186             }
 187             SnapshotBackedWriteTransaction ret =new SnapshotBackedWriteTransaction(nextIdentifier(), snapshot,this);
 188             return ret;
 189         }
 190
 191         @Override
 192         public DOMStoreThreePhaseCommitCohort ready(final SnapshotBackedWriteTransaction tx) {
 193             DOMStoreThreePhaseCommitCohort storeCohort = InMemoryDOMDataStore.this.ready(tx);
 194             // FIXME: We probably want to add Transaction Chain cohort
 195             return storeCohort;
 196         }
 197
 198         @Override
 199         public void close() {
 200             // TODO Auto-generated method stub
 201
 202         }
 203
 204     }
 205
 206     private class ThreePhaseCommitImpl implements DOMStoreThreePhaseCommitCohort {
 207
 208         private final SnapshotBackedWriteTransaction transaction;
 209         private final DataTreeModification modification;
 210
 211         private ResolveDataChangeEventsTask listenerResolver;
 212         private DataTreeCandidate candidate;
 213
 214         public ThreePhaseCommitImpl(final SnapshotBackedWriteTransaction writeTransaction) {
 215             this.transaction = writeTransaction;
 216             this.modification = transaction.getMutatedView();
 217         }
 218
 219         @Override
 220         public ListenableFuture<Boolean> canCommit() {
 221             return executor.submit(new Callable<Boolean>() {
 222                 @Override
 223                 public Boolean call() {
 224                     try {
 225                         dataTree.validate(modification);
 226                         LOG.debug("Store Transaction: {} can be committed", transaction.getIdentifier());
 227                         return true;
 228                     } catch (DataPreconditionFailedException e) {
 229                         LOG.warn("Store Tx: {} Data Precondition failed for {}.",transaction.getIdentifier(),e.getPath(),e);
 230                         return false;
 231                     }
 232                 }
 233             });
 234         }
 235
 236         @Override
 237         public ListenableFuture<Void> preCommit() {
 238             return executor.submit(new Callable<Void>() {
 239                 @Override
 240                 public Void call() {
 241                     candidate = dataTree.prepare(modification);
 242                     listenerResolver = ResolveDataChangeEventsTask.create(candidate, listenerTree);
 243                     return null;
 244                 }
 245             });
 246         }
 247
 248         @Override
 249         public ListenableFuture<Void> abort() {
 250             candidate = null;
 251             return Futures.immediateFuture(null);
 252         }
 253
 254         @Override
 255         public ListenableFuture<Void> commit() {
 256             checkState(candidate != null, "Proposed subtree must be computed");
 257
 258             /*
 259              * The commit has to occur atomically with regard to listener
 260              * registrations.
 261              */
 262             synchronized (this) {
 263                 dataTree.commit(candidate);
 264
 265                 for (ChangeListenerNotifyTask task : listenerResolver.call()) {
 266                     LOG.trace("Scheduling invocation of listeners: {}", task);
 267                     executor.submit(task);
 268                 }
 269             }
 270
 271             return Futures.immediateFuture(null);
 272         }
 273     }
 274 }