opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/legacy/sharded/adapter/ShardedDOMDataBrokerDelegatingReadWriteTransaction.java

   1 /*
   2  * Copyright (c) 2016 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.legacy.sharded.adapter;
   9
  10 import static com.google.common.base.Preconditions.checkState;
  11 import static java.util.Objects.requireNonNull;
  12
  13 import com.google.common.base.Optional;
  14 import com.google.common.collect.ImmutableMap;
  15 import com.google.common.collect.Lists;
  16 import com.google.common.util.concurrent.CheckedFuture;
  17 import com.google.common.util.concurrent.FluentFuture;
  18 import com.google.common.util.concurrent.FutureCallback;
  19 import com.google.common.util.concurrent.Futures;
  20 import com.google.common.util.concurrent.ListenableFuture;
  21 import com.google.common.util.concurrent.MoreExecutors;
  22 import com.google.common.util.concurrent.SettableFuture;
  23 import java.util.EnumMap;
  24 import java.util.LinkedList;
  25 import java.util.Map;
  26 import java.util.Queue;
  27 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  28 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  29 import org.opendaylight.controller.md.sal.dom.api.DOMDataReadOnlyTransaction;
  30 import org.opendaylight.controller.md.sal.dom.api.DOMDataReadWriteTransaction;
  31 import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
  32 import org.opendaylight.mdsal.common.api.CommitInfo;
  33 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  34 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  35 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
  36 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeConfiguration;
  37 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
  38 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot;
  39 import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory;
  40 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  41
  42 /**
  43  * Read/write transaction that delegates write and initial read to
  44  * {@link org.opendaylight.mdsal.dom.broker.ShardedDOMWriteTransactionAdapter}
  45  * and {@link org.opendaylight.mdsal.dom.broker.ShardedDOMReadTransactionAdapter}
  46  * respectively. These two in turn rely on shard aware implementation of
  47  * {@link org.opendaylight.mdsal.dom.api.DOMDataTreeService}.
  48  *
  49  * <p>
  50  * Since reading data distributed on different subshards is not guaranteed to
  51  * return all relevant data, best effort is to try to operate only on single
  52  * subtree in conceptual data tree. We define this subtree by first write
  53  * operation performed on transaction. All next read and write operations
  54  * should be performed just in this initial subtree.
  55  *
  56  */
  57 // FIXME explicitly enforce just one subtree requirement
  58 @Deprecated
  59 class ShardedDOMDataBrokerDelegatingReadWriteTransaction implements DOMDataReadWriteTransaction {
  60     private final DOMDataReadOnlyTransaction readTxDelegate;
  61     private final DOMDataWriteTransaction writeTxDelegate;
  62     private final Object txIdentifier;
  63     private final ImmutableMap<LogicalDatastoreType, Queue<Modification>> modificationHistoryMap;
  64     private final ImmutableMap<LogicalDatastoreType, DataTreeSnapshot> snapshotMap;
  65     private final Map<LogicalDatastoreType, ListenableFuture<Optional<NormalizedNode<?, ?>>>> initialReadMap;
  66     private YangInstanceIdentifier root = null;
  67
  68     ShardedDOMDataBrokerDelegatingReadWriteTransaction(final Object readWriteTxId, final SchemaContext ctx,
  69                                                               final DOMDataReadOnlyTransaction readTxDelegate,
  70                                                               final DOMDataWriteTransaction writeTxDelegate) {
  71         this.readTxDelegate = requireNonNull(readTxDelegate);
  72         this.writeTxDelegate = requireNonNull(writeTxDelegate);
  73         this.txIdentifier = requireNonNull(readWriteTxId);
  74         this.initialReadMap = new EnumMap<>(LogicalDatastoreType.class);
  75
  76         final InMemoryDataTreeFactory treeFactory = new InMemoryDataTreeFactory();
  77         final ImmutableMap.Builder<LogicalDatastoreType, DataTreeSnapshot> snapshotMapBuilder = ImmutableMap.builder();
  78         final ImmutableMap.Builder<LogicalDatastoreType, Queue<Modification>> modificationHistoryMapBuilder
  79                 = ImmutableMap.builder();
  80         for (final LogicalDatastoreType store : LogicalDatastoreType.values()) {
  81             final DataTree tree = treeFactory.create(treeConfigForStore(store));
  82             tree.setSchemaContext(ctx);
  83             snapshotMapBuilder.put(store, tree.takeSnapshot());
  84
  85             modificationHistoryMapBuilder.put(store, new LinkedList<>());
  86         }
  87
  88         modificationHistoryMap = modificationHistoryMapBuilder.build();
  89         snapshotMap = snapshotMapBuilder.build();
  90     }
  91
  92     @Override
  93     public boolean cancel() {
  94         readTxDelegate.close();
  95         return writeTxDelegate.cancel();
  96     }
  97
  98     @Override
  99     public void delete(final LogicalDatastoreType store, final YangInstanceIdentifier path) {
 100         if (root == null) {
 101             initialRead(path);
 102         }
 103
 104         modificationHistoryMap.get(store).add(new Modification(Modification.Operation.DELETE, path, null));
 105         writeTxDelegate.delete(store, path);
 106     }
 107
 108     @Override
 109     public FluentFuture<? extends CommitInfo> commit() {
 110         return writeTxDelegate.commit();
 111     }
 112
 113     @Override
 114     public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> read(final LogicalDatastoreType store,
 115                                                                                    final YangInstanceIdentifier path) {
 116         checkState(root != null,
 117                    "A modify operation (put, merge or delete) must be performed prior to a read operation");
 118         final SettableFuture<Optional<NormalizedNode<?, ?>>> readResult = SettableFuture.create();
 119         final Queue<Modification> currentHistory = Lists.newLinkedList(modificationHistoryMap.get(store));
 120         Futures.addCallback(initialReadMap.get(store), new FutureCallback<Optional<NormalizedNode<?, ?>>>() {
 121             @Override
 122             public void onSuccess(final Optional<NormalizedNode<?, ?>> result) {
 123                 final DataTreeModification mod = snapshotMap.get(store).newModification();
 124                 if (result.isPresent()) {
 125                     mod.write(path, result.get());
 126                 }
 127                 applyModificationHistoryToSnapshot(mod, currentHistory);
 128                 readResult.set(Optional.fromJavaUtil(mod.readNode(path)));
 129             }
 130
 131             @Override
 132             public void onFailure(final Throwable throwable) {
 133                 readResult.setException(throwable);
 134             }
 135         }, MoreExecutors.directExecutor());
 136
 137         return Futures.makeChecked(readResult, ReadFailedException.MAPPER);
 138     }
 139
 140     @Override
 141     public CheckedFuture<Boolean, ReadFailedException> exists(final LogicalDatastoreType store,
 142                                                               final YangInstanceIdentifier path) {
 143         checkState(root != null,
 144                    "A modify operation (put, merge or delete) must be performed prior to an exists operation");
 145         return Futures.makeChecked(Futures.transform(read(store, path), Optional::isPresent,
 146             MoreExecutors.directExecutor()), ReadFailedException.MAPPER);
 147     }
 148
 149     @Override
 150     public void put(final LogicalDatastoreType store, final YangInstanceIdentifier path,
 151                     final NormalizedNode<?, ?> data) {
 152         if (root == null) {
 153             initialRead(path);
 154         }
 155
 156         modificationHistoryMap.get(store).add(new Modification(Modification.Operation.WRITE, path, data));
 157         writeTxDelegate.put(store, path, data);
 158     }
 159
 160     @Override
 161     public void merge(final LogicalDatastoreType store, final YangInstanceIdentifier path,
 162                       final NormalizedNode<?, ?> data) {
 163         if (root == null) {
 164             initialRead(path);
 165         }
 166
 167         modificationHistoryMap.get(store).add(new Modification(Modification.Operation.MERGE, path, data));
 168         writeTxDelegate.merge(store, path, data);
 169     }
 170
 171     @Override
 172     public Object getIdentifier() {
 173         return txIdentifier;
 174     }
 175
 176     private void initialRead(final YangInstanceIdentifier path) {
 177         root = path;
 178
 179         for (final LogicalDatastoreType store : LogicalDatastoreType.values()) {
 180             initialReadMap.put(store, readTxDelegate.read(store, path));
 181         }
 182     }
 183
 184     private static DataTreeConfiguration treeConfigForStore(final LogicalDatastoreType store) {
 185         return store == LogicalDatastoreType.CONFIGURATION ? DataTreeConfiguration.DEFAULT_CONFIGURATION
 186                 : DataTreeConfiguration.DEFAULT_OPERATIONAL;
 187     }
 188
 189     private static void applyModificationHistoryToSnapshot(final DataTreeModification dataTreeModification,
 190                                                     final Queue<Modification> modificationHistory) {
 191         while (!modificationHistory.isEmpty()) {
 192             final Modification modification = modificationHistory.poll();
 193             switch (modification.getOperation()) {
 194                 case WRITE:
 195                     dataTreeModification.write(modification.getPath(), modification.getData());
 196                     break;
 197                 case MERGE:
 198                     dataTreeModification.merge(modification.getPath(), modification.getData());
 199                     break;
 200                 case DELETE:
 201                     dataTreeModification.delete(modification.getPath());
 202                     break;
 203                 default:
 204                     // NOOP
 205             }
 206         }
 207     }
 208
 209     static class Modification {
 210
 211         enum Operation {
 212             WRITE, MERGE, DELETE
 213         }
 214
 215         private final NormalizedNode<?, ?> data;
 216         private final YangInstanceIdentifier path;
 217         private final Operation operation;
 218
 219         Modification(final Operation operation, final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
 220             this.data = data;
 221             this.path = requireNonNull(path);
 222             this.operation = requireNonNull(operation);
 223         }
 224
 225         Operation getOperation() {
 226             return operation;
 227         }
 228
 229         YangInstanceIdentifier getPath() {
 230             return path;
 231         }
 232
 233         NormalizedNode<?, ?> getData() {
 234             return data;
 235         }
 236     }
 237 }