Add batching of non-isolated transaction in ShardedDOMDataTreeProducer
[mdsal.git] / dom / mdsal-dom-inmemory-datastore / src / main / java / org / opendaylight / mdsal / dom / store / inmemory / InmemoryDOMDataTreeShardWriteTransaction.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
9 package org.opendaylight.mdsal.dom.store.inmemory;
10
11 import com.google.common.base.Optional;
12 import com.google.common.base.Preconditions;
13 import com.google.common.util.concurrent.CheckedFuture;
14 import com.google.common.util.concurrent.ListenableFuture;
15 import com.google.common.util.concurrent.ListeningExecutorService;
16 import java.util.ArrayList;
17 import java.util.Iterator;
18 import java.util.Map.Entry;
19 import org.opendaylight.mdsal.common.api.ReadFailedException;
20 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
21 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteCursor;
22 import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort;
23 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
24 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
25 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
26 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
27 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
28 import org.slf4j.Logger;
29 import org.slf4j.LoggerFactory;
30
31 class InmemoryDOMDataTreeShardWriteTransaction implements DOMDataTreeShardWriteTransaction {
32
33     private static final Logger LOG = LoggerFactory.getLogger(InmemoryDOMDataTreeShardWriteTransaction.class);
34
35     private enum SimpleCursorOperation {
36         MERGE {
37             @Override
38             void applyOnLeaf(final DOMDataTreeWriteCursor cursor, final PathArgument child,
39                     final NormalizedNode<?, ?> data) {
40                 cursor.merge(child, data);
41             }
42         },
43         DELETE {
44             @Override
45             void applyOnLeaf(final DOMDataTreeWriteCursor cursor, final PathArgument child,
46                     final NormalizedNode<?, ?> data) {
47                 cursor.delete(child);
48             }
49         },
50         WRITE {
51             @Override
52             void applyOnLeaf(final DOMDataTreeWriteCursor cursor, final PathArgument child,
53                     final NormalizedNode<?, ?> data) {
54                 cursor.write(child, data);
55             }
56         };
57
58         abstract void applyOnLeaf(DOMDataTreeWriteCursor cursor, PathArgument child, NormalizedNode<?, ?> data);
59
60         void apply(final DOMDataTreeWriteCursor cursor, final YangInstanceIdentifier path,
61                 final NormalizedNode<?, ?> data) {
62             int enterCount = 0;
63             final Iterator<PathArgument> it = path.getPathArguments().iterator();
64             while (it.hasNext()) {
65                 final PathArgument currentArg = it.next();
66                 if (it.hasNext()) {
67                     // We need to enter one level deeper, we are not at leaf (modified) node
68                     cursor.enter(currentArg);
69                     enterCount++;
70                 } else {
71                     applyOnLeaf(cursor, currentArg, data);
72                 }
73             }
74             cursor.exit(enterCount);
75         }
76     }
77
78     private final ArrayList<DOMStoreThreePhaseCommitCohort> cohorts = new ArrayList<>();
79     private final InMemoryDOMDataTreeShardChangePublisher changePublisher;
80     private final ShardDataModification modification;
81     private final ListeningExecutorService executor;
82     private final DataTree rootShardDataTree;
83
84     private DataTreeModification rootModification = null;
85     private DOMDataTreeWriteCursor cursor;
86     private boolean finished = false;
87
88     InmemoryDOMDataTreeShardWriteTransaction(final ShardDataModification root,
89                                              final DataTree rootShardDataTree,
90                                              final InMemoryDOMDataTreeShardChangePublisher changePublisher,
91                                              final ListeningExecutorService executor) {
92         this.modification = Preconditions.checkNotNull(root);
93         this.rootShardDataTree = Preconditions.checkNotNull(rootShardDataTree);
94         this.changePublisher = Preconditions.checkNotNull(changePublisher);
95         this.executor = executor;
96     }
97
98     private DOMDataTreeWriteCursor getCursor() {
99         if (cursor == null) {
100             cursor = new ShardDataModificationCursor(modification, this);
101         }
102         return cursor;
103     }
104
105     void delete(final YangInstanceIdentifier path) {
106         final YangInstanceIdentifier relativePath = toRelative(path);
107         Preconditions.checkArgument(!YangInstanceIdentifier.EMPTY.equals(relativePath),
108                 "Deletion of shard root is not allowed");
109         SimpleCursorOperation.DELETE.apply(getCursor(), relativePath , null);
110     }
111
112     void merge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
113         SimpleCursorOperation.MERGE.apply(getCursor(), toRelative(path), data);
114     }
115
116     void write(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
117         SimpleCursorOperation.DELETE.apply(getCursor(), toRelative(path), data);
118     }
119
120     private YangInstanceIdentifier toRelative(final YangInstanceIdentifier path) {
121         final Optional<YangInstanceIdentifier> relative =
122                 path.relativeTo(modification.getPrefix().getRootIdentifier());
123         Preconditions.checkArgument(relative.isPresent());
124         return relative.get();
125     }
126
127     public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> read(final YangInstanceIdentifier path) {
128         throw new UnsupportedOperationException("Not implemented yet");
129     }
130
131     public CheckedFuture<Boolean, ReadFailedException> exists(final YangInstanceIdentifier path) {
132         throw new UnsupportedOperationException("Not implemented yet");
133     }
134
135     @Override
136     public void close() {
137         Preconditions.checkState(!finished, "Attempting to close an already finished transaction.");
138         modification.closeTransactions();
139         if (cursor != null) {
140             cursor.close();
141         }
142         finished = true;
143     }
144
145     void cursorClosed() {
146         Preconditions.checkNotNull(cursor);
147         modification.closeCursor();
148         cursor = null;
149     }
150
151     public boolean isFinished() {
152         return finished;
153     }
154
155     @Override
156     public void ready() {
157         Preconditions.checkState(!finished, "Attempting to ready an already finished transaction.");
158         Preconditions.checkState(cursor == null, "Attempting to ready a transaction that has an open cursor.");
159         Preconditions.checkNotNull(modification, "Attempting to ready an empty transaction.");
160
161         LOG.debug("Readying open transaction on shard {}", modification.getPrefix());
162         rootModification = modification.seal();
163
164         cohorts.add(new InMemoryDOMDataTreeShardThreePhaseCommitCohort(
165                 rootShardDataTree, rootModification, changePublisher));
166         for (final Entry<DOMDataTreeIdentifier, ForeignShardModificationContext> entry :
167                 modification.getChildShards().entrySet()) {
168             cohorts.add(new ForeignShardThreePhaseCommitCohort(entry.getKey(), entry.getValue()));
169         }
170         finished = true;
171     }
172
173     @Override
174     public ListenableFuture<Void> submit() {
175         LOG.debug("Submitting open transaction on shard {}", modification.getPrefix());
176
177         Preconditions.checkNotNull(cohorts);
178         Preconditions.checkState(!cohorts.isEmpty(), "Transaction was not readied yet.");
179
180         final ListenableFuture<Void> submit = executor.submit(new ShardSubmitCoordinationTask(
181                 modification.getPrefix(), cohorts));
182
183         return submit;
184     }
185
186     @Override
187     public ListenableFuture<Boolean> validate() {
188         LOG.debug("CanCommit on open transaction on shard {}", modification.getPrefix());
189
190         final ListenableFuture<Boolean> submit = executor.submit(new ShardCanCommitCoordinationTask(
191                 modification.getPrefix(), cohorts));
192         return submit;
193     }
194
195     @Override
196     public ListenableFuture<Void> prepare() {
197         LOG.debug("PreCommit on open transaction on shard {}", modification.getPrefix());
198
199         final ListenableFuture<Void> submit = executor.submit(new ShardPreCommitCoordinationTask(
200                 modification.getPrefix(), cohorts));
201         return submit;
202     }
203
204     @Override
205     public ListenableFuture<Void> commit() {
206         LOG.debug("Commit open transaction on shard {}", modification.getPrefix());
207
208         final ListenableFuture<Void> submit = executor.submit(new ShardCommitCoordinationTask(
209                 modification.getPrefix(), cohorts));
210         return submit;
211     }
212
213     @Override
214     public DOMDataTreeWriteCursor createCursor(final DOMDataTreeIdentifier prefix) {
215         Preconditions.checkState(!finished, "Transaction is finished/closed already.");
216         Preconditions.checkState(cursor == null, "Previous cursor wasn't closed");
217         final DOMDataTreeWriteCursor ret = getCursor();
218         final YangInstanceIdentifier relativePath = toRelative(prefix.getRootIdentifier());
219         ret.enter(relativePath.getPathArguments());
220         return ret;
221     }
222 }