opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/DOMDataBrokerImpl.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.broker.impl;
   9
  10 import static com.google.common.base.Preconditions.checkNotNull;
  11 import static com.google.common.base.Preconditions.checkState;
  12
  13 import java.util.Collections;
  14 import java.util.List;
  15 import java.util.Map.Entry;
  16 import java.util.concurrent.Callable;
  17 import java.util.concurrent.ExecutionException;
  18 import java.util.concurrent.atomic.AtomicLong;
  19
  20 import org.opendaylight.controller.md.sal.common.api.TransactionStatus;
  21 import org.opendaylight.controller.md.sal.common.api.data.AsyncTransaction;
  22 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  23 import org.opendaylight.controller.md.sal.dom.api.DOMDataBroker;
  24 import org.opendaylight.controller.md.sal.dom.api.DOMDataChangeListener;
  25 import org.opendaylight.controller.md.sal.dom.api.DOMDataReadTransaction;
  26 import org.opendaylight.controller.md.sal.dom.api.DOMDataReadWriteTransaction;
  27 import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
  28 import org.opendaylight.controller.sal.common.util.Rpcs;
  29 import org.opendaylight.controller.sal.core.spi.data.DOMStore;
  30 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
  31 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
  32 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  33 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransaction;
  34 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
  35 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  36 import org.opendaylight.yangtools.yang.common.RpcError;
  37 import org.opendaylight.yangtools.yang.common.RpcResult;
  38 import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier;
  39 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  40 import org.slf4j.Logger;
  41 import org.slf4j.LoggerFactory;
  42
  43 import com.google.common.base.Function;
  44 import com.google.common.base.Optional;
  45 import com.google.common.collect.ImmutableList;
  46 import com.google.common.collect.ImmutableList.Builder;
  47 import com.google.common.collect.ImmutableMap;
  48 import com.google.common.util.concurrent.Futures;
  49 import com.google.common.util.concurrent.ListenableFuture;
  50 import com.google.common.util.concurrent.ListeningExecutorService;
  51
  52 public class DOMDataBrokerImpl implements DOMDataBroker, AutoCloseable {
  53
  54     private static final Logger LOG = LoggerFactory.getLogger(DOMDataBrokerImpl.class);
  55     private static final Logger COORDINATOR_LOG = LoggerFactory.getLogger(CommitCoordination.class);
  56     private final ImmutableMap<LogicalDatastoreType, DOMStore> datastores;
  57     private final ListeningExecutorService executor;
  58     private final AtomicLong txNum = new AtomicLong();
  59
  60     public DOMDataBrokerImpl(final ImmutableMap<LogicalDatastoreType, DOMStore> datastores,
  61             final ListeningExecutorService executor) {
  62         super();
  63         this.datastores = datastores;
  64         this.executor = executor;
  65     }
  66
  67     private static final Function<Iterable<Boolean>, Boolean> AND_FUNCTION = new Function<Iterable<Boolean>, Boolean>() {
  68
  69         @Override
  70         public Boolean apply(final Iterable<Boolean> input) {
  71
  72             for (Boolean value : input) {
  73                 if (value == false) {
  74                     return Boolean.FALSE;
  75                 }
  76             }
  77             return Boolean.TRUE;
  78         }
  79     };
  80
  81     @Override
  82     public DOMDataReadTransaction newReadOnlyTransaction() {
  83         ImmutableMap.Builder<LogicalDatastoreType, DOMStoreReadTransaction> builder = ImmutableMap.builder();
  84         for (Entry<LogicalDatastoreType, DOMStore> store : datastores.entrySet()) {
  85             builder.put(store.getKey(), store.getValue().newReadOnlyTransaction());
  86         }
  87         return new ReadOnlyTransactionImpl(newTransactionIdentifier(), builder.build());
  88     }
  89
  90     private Object newTransactionIdentifier() {
  91         return "DOM-" + txNum.getAndIncrement();
  92     }
  93
  94     @Override
  95     public DOMDataReadWriteTransaction newReadWriteTransaction() {
  96         ImmutableMap.Builder<LogicalDatastoreType, DOMStoreReadWriteTransaction> builder = ImmutableMap.builder();
  97         for (Entry<LogicalDatastoreType, DOMStore> store : datastores.entrySet()) {
  98             builder.put(store.getKey(), store.getValue().newReadWriteTransaction());
  99         }
 100         return new ReadWriteTransactionImpl(newTransactionIdentifier(), builder.build(), this);
 101     }
 102
 103     @Override
 104     public DOMDataWriteTransaction newWriteOnlyTransaction() {
 105         ImmutableMap.Builder<LogicalDatastoreType, DOMStoreWriteTransaction> builder = ImmutableMap.builder();
 106         for (Entry<LogicalDatastoreType, DOMStore> store : datastores.entrySet()) {
 107             builder.put(store.getKey(), store.getValue().newWriteOnlyTransaction());
 108         }
 109         return new WriteTransactionImpl<DOMStoreWriteTransaction>(newTransactionIdentifier(), builder.build(), this);
 110     }
 111
 112     @Override
 113     public ListenerRegistration<DOMDataChangeListener> registerDataChangeListener(final LogicalDatastoreType store,
 114             final InstanceIdentifier path, final DOMDataChangeListener listener, final DataChangeScope triggeringScope) {
 115
 116         DOMStore potentialStore = datastores.get(store);
 117         checkState(potentialStore != null, "Requested logical data store is not available.");
 118         return potentialStore.registerChangeListener(path, listener, triggeringScope);
 119     }
 120
 121     private ListenableFuture<RpcResult<TransactionStatus>> submit(
 122             final WriteTransactionImpl<? extends DOMStoreWriteTransaction> transaction) {
 123         LOG.debug("Tx: {} is submitted for execution.", transaction.getIdentifier());
 124         return executor.submit(new CommitCoordination(transaction));
 125     }
 126
 127     private abstract static class AbstractCompositeTransaction<K, T extends DOMStoreTransaction> implements
 128             AsyncTransaction<InstanceIdentifier, NormalizedNode<?, ?>> {
 129
 130         private final ImmutableMap<K, T> backingTxs;
 131         private final Object identifier;
 132
 133         protected AbstractCompositeTransaction(final Object identifier, final ImmutableMap<K, T> backingTxs) {
 134             this.identifier = checkNotNull(identifier, "Identifier should not be null");
 135             this.backingTxs = checkNotNull(backingTxs, "Backing transactions should not be null");
 136         }
 137
 138         protected T getSubtransaction(final K key) {
 139             return backingTxs.get(key);
 140         }
 141
 142         public Iterable<T> getSubtransactions() {
 143             return backingTxs.values();
 144         }
 145
 146         @Override
 147         public Object getIdentifier() {
 148             return identifier;
 149         }
 150
 151         @Override
 152         public void close() {
 153             try {
 154                 for (T subtransaction : backingTxs.values()) {
 155                     subtransaction.close();
 156                 }
 157             } catch (Exception e) {
 158                 throw new IllegalStateException("Uncaught exception occured during closing transaction.", e);
 159             }
 160         }
 161
 162     }
 163
 164     private static class ReadOnlyTransactionImpl extends
 165             AbstractCompositeTransaction<LogicalDatastoreType, DOMStoreReadTransaction> implements
 166             DOMDataReadTransaction {
 167
 168         protected ReadOnlyTransactionImpl(final Object identifier,
 169                 final ImmutableMap<LogicalDatastoreType, DOMStoreReadTransaction> backingTxs) {
 170             super(identifier, backingTxs);
 171         }
 172
 173         @Override
 174         public ListenableFuture<Optional<NormalizedNode<?, ?>>> read(final LogicalDatastoreType store,
 175                 final InstanceIdentifier path) {
 176             return getSubtransaction(store).read(path);
 177         }
 178
 179     }
 180
 181     private static class WriteTransactionImpl<T extends DOMStoreWriteTransaction> extends
 182             AbstractCompositeTransaction<LogicalDatastoreType, T> implements DOMDataWriteTransaction {
 183
 184         private final DOMDataBrokerImpl broker;
 185         private ImmutableList<DOMStoreThreePhaseCommitCohort> cohorts;
 186
 187         protected WriteTransactionImpl(final Object identifier, final ImmutableMap<LogicalDatastoreType, T> backingTxs,
 188                 final DOMDataBrokerImpl broker) {
 189             super(identifier, backingTxs);
 190             this.broker = broker;
 191         }
 192
 193         public synchronized Iterable<DOMStoreThreePhaseCommitCohort> ready() {
 194             checkState(cohorts == null, "Transaction was already marked as ready.");
 195             ImmutableList.Builder<DOMStoreThreePhaseCommitCohort> cohortsBuilder = ImmutableList.builder();
 196             for (DOMStoreWriteTransaction subTx : getSubtransactions()) {
 197                 cohortsBuilder.add(subTx.ready());
 198             }
 199             cohorts = cohortsBuilder.build();
 200             return cohorts;
 201         }
 202
 203         protected ImmutableList<DOMStoreThreePhaseCommitCohort> getCohorts() {
 204             return cohorts;
 205         }
 206
 207         @Override
 208         public void put(final LogicalDatastoreType store, final InstanceIdentifier path, final NormalizedNode<?, ?> data) {
 209             getSubtransaction(store).write(path, data);
 210         }
 211
 212         @Override
 213         public void delete(final LogicalDatastoreType store, final InstanceIdentifier path) {
 214             getSubtransaction(store).delete(path);
 215         }
 216
 217         @Override
 218         public void merge(final LogicalDatastoreType store, final InstanceIdentifier path,
 219                 final NormalizedNode<?, ?> data) {
 220             getSubtransaction(store).merge(path,data);
 221         }
 222
 223         @Override
 224         public void cancel() {
 225             // TODO Auto-generated method stub
 226
 227         }
 228
 229         @Override
 230         public ListenableFuture<RpcResult<TransactionStatus>> commit() {
 231
 232             ready();
 233             return broker.submit(this);
 234         }
 235
 236     }
 237
 238     private static class ReadWriteTransactionImpl extends WriteTransactionImpl<DOMStoreReadWriteTransaction> implements
 239             DOMDataReadWriteTransaction {
 240
 241         protected ReadWriteTransactionImpl(final Object identifier,
 242                 final ImmutableMap<LogicalDatastoreType, DOMStoreReadWriteTransaction> backingTxs,
 243                 final DOMDataBrokerImpl broker) {
 244             // super(identifier, backingTxs);
 245             super(identifier, backingTxs, broker);
 246         }
 247
 248         @Override
 249         public ListenableFuture<Optional<NormalizedNode<?, ?>>> read(final LogicalDatastoreType store,
 250                 final InstanceIdentifier path) {
 251             return getSubtransaction(store).read(path);
 252         }
 253     }
 254
 255     private final class CommitCoordination implements Callable<RpcResult<TransactionStatus>> {
 256
 257         private final WriteTransactionImpl<? extends DOMStoreWriteTransaction> transaction;
 258
 259         public CommitCoordination(final WriteTransactionImpl<? extends DOMStoreWriteTransaction> transaction) {
 260             this.transaction = transaction;
 261         }
 262
 263         @Override
 264         public RpcResult<TransactionStatus> call() throws Exception {
 265
 266             try {
 267                 Boolean canCommit = canCommit().get();
 268
 269                 if (canCommit) {
 270                     try {
 271                         preCommit().get();
 272                         try {
 273                             commit().get();
 274                             COORDINATOR_LOG.debug("Tx: {} Is commited.", transaction.getIdentifier());
 275                             return Rpcs.getRpcResult(true, TransactionStatus.COMMITED,
 276                                     Collections.<RpcError> emptySet());
 277
 278                         } catch (InterruptedException | ExecutionException e) {
 279                             COORDINATOR_LOG.error("Tx: {} Error during commit", transaction.getIdentifier(), e);
 280                         }
 281
 282                     } catch (InterruptedException | ExecutionException e) {
 283                         COORDINATOR_LOG.warn("Tx: {} Error during preCommit, starting Abort",
 284                                 transaction.getIdentifier(), e);
 285                     }
 286                 } else {
 287                     COORDINATOR_LOG.info("Tx: {} Did not pass canCommit phase.", transaction.getIdentifier());
 288                     abort().get();
 289                 }
 290             } catch (InterruptedException | ExecutionException e) {
 291                 COORDINATOR_LOG.warn("Tx: {} Error during canCommit, starting Abort", transaction.getIdentifier(), e);
 292
 293             }
 294             try {
 295                 abort().get();
 296             } catch (InterruptedException | ExecutionException e) {
 297                 COORDINATOR_LOG.error("Tx: {} Error during abort", transaction.getIdentifier(), e);
 298             }
 299             return Rpcs.getRpcResult(false, TransactionStatus.FAILED, Collections.<RpcError> emptySet());
 300         }
 301
 302         public ListenableFuture<Void> preCommit() {
 303             COORDINATOR_LOG.debug("Transaction {}: PreCommit Started ", transaction.getIdentifier());
 304             Builder<ListenableFuture<Void>> ops = ImmutableList.builder();
 305             for (DOMStoreThreePhaseCommitCohort cohort : transaction.getCohorts()) {
 306                 ops.add(cohort.preCommit());
 307             }
 308             return (ListenableFuture) Futures.allAsList(ops.build());
 309         }
 310
 311         public ListenableFuture<Void> commit() {
 312             COORDINATOR_LOG.debug("Transaction {}: Commit Started ", transaction.getIdentifier());
 313             Builder<ListenableFuture<Void>> ops = ImmutableList.builder();
 314             for (DOMStoreThreePhaseCommitCohort cohort : transaction.getCohorts()) {
 315                 ops.add(cohort.commit());
 316             }
 317             return (ListenableFuture) Futures.allAsList(ops.build());
 318         }
 319
 320         public ListenableFuture<Boolean> canCommit() {
 321             COORDINATOR_LOG.debug("Transaction {}: CanCommit Started ", transaction.getIdentifier());
 322             Builder<ListenableFuture<Boolean>> canCommitOperations = ImmutableList.builder();
 323             for (DOMStoreThreePhaseCommitCohort cohort : transaction.getCohorts()) {
 324                 canCommitOperations.add(cohort.canCommit());
 325             }
 326             ListenableFuture<List<Boolean>> allCanCommits = Futures.allAsList(canCommitOperations.build());
 327             return Futures.transform(allCanCommits, AND_FUNCTION);
 328         }
 329
 330         public ListenableFuture<Void> abort() {
 331             COORDINATOR_LOG.debug("Transaction {}: Abort Started ", transaction.getIdentifier());
 332             Builder<ListenableFuture<Void>> ops = ImmutableList.builder();
 333             for (DOMStoreThreePhaseCommitCohort cohort : transaction.getCohorts()) {
 334                 ops.add(cohort.abort());
 335             }
 336             return (ListenableFuture) Futures.allAsList(ops.build());
 337         };
 338
 339     }
 340
 341     @Override
 342     public void close() throws Exception {
 343
 344     }
 345
 346 }