Merge "Remove SimpleDataTreeCandidate"
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / TransactionContextImpl.java
1 /*
2  * Copyright (c) 2015 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.cluster.datastore;
9
10 import akka.actor.ActorSelection;
11 import akka.dispatch.Mapper;
12 import akka.dispatch.OnComplete;
13 import com.google.common.base.Optional;
14 import com.google.common.collect.Lists;
15 import com.google.common.util.concurrent.SettableFuture;
16 import java.util.List;
17 import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier;
18 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
19 import org.opendaylight.controller.cluster.datastore.messages.BatchedModificationsReply;
20 import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
21 import org.opendaylight.controller.cluster.datastore.messages.DataExists;
22 import org.opendaylight.controller.cluster.datastore.messages.DataExistsReply;
23 import org.opendaylight.controller.cluster.datastore.messages.ReadData;
24 import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
25 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction;
26 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
27 import org.opendaylight.controller.cluster.datastore.messages.SerializableMessage;
28 import org.opendaylight.controller.cluster.datastore.modification.DeleteModification;
29 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
30 import org.opendaylight.controller.cluster.datastore.modification.Modification;
31 import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
32 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
33 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
34 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
35 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
36 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
37 import org.slf4j.Logger;
38 import org.slf4j.LoggerFactory;
39 import scala.concurrent.Future;
40
41 public class TransactionContextImpl extends AbstractTransactionContext {
42     private static final Logger LOG = LoggerFactory.getLogger(TransactionContextImpl.class);
43
44     private final String transactionChainId;
45     private final ActorContext actorContext;
46     private final ActorSelection actor;
47     private final boolean isTxActorLocal;
48     private final short remoteTransactionVersion;
49
50     private final OperationCompleter operationCompleter;
51     private BatchedModifications batchedModifications;
52
53     protected TransactionContextImpl(ActorSelection actor, TransactionIdentifier identifier,
54             String transactionChainId, ActorContext actorContext, SchemaContext schemaContext, boolean isTxActorLocal,
55             short remoteTransactionVersion, OperationCompleter operationCompleter) {
56         super(identifier);
57         this.actor = actor;
58         this.transactionChainId = transactionChainId;
59         this.actorContext = actorContext;
60         this.isTxActorLocal = isTxActorLocal;
61         this.remoteTransactionVersion = remoteTransactionVersion;
62         this.operationCompleter = operationCompleter;
63     }
64
65     private Future<Object> completeOperation(Future<Object> operationFuture){
66         operationFuture.onComplete(this.operationCompleter, actorContext.getClientDispatcher());
67         return operationFuture;
68     }
69
70
71     private ActorSelection getActor() {
72         return actor;
73     }
74
75     protected ActorContext getActorContext() {
76         return actorContext;
77     }
78
79     protected short getRemoteTransactionVersion() {
80         return remoteTransactionVersion;
81     }
82
83     protected Future<Object> executeOperationAsync(SerializableMessage msg) {
84         return completeOperation(actorContext.executeOperationAsync(getActor(), isTxActorLocal ? msg : msg.toSerializable()));
85     }
86
87     @Override
88     public void closeTransaction() {
89         LOG.debug("Tx {} closeTransaction called", getIdentifier());
90
91         actorContext.sendOperationAsync(getActor(), CloseTransaction.INSTANCE.toSerializable());
92     }
93
94     @Override
95     public Future<ActorSelection> readyTransaction() {
96         LOG.debug("Tx {} readyTransaction called with {} previous recorded operations pending",
97             getIdentifier(), recordedOperationCount());
98
99         // Send the remaining batched modifications if any.
100
101         sendAndRecordBatchedModifications();
102
103         // Send the ReadyTransaction message to the Tx actor.
104
105         Future<Object> readyReplyFuture = executeOperationAsync(ReadyTransaction.INSTANCE);
106
107         return combineRecordedOperationsFutures(readyReplyFuture);
108     }
109
110     protected Future<ActorSelection> combineRecordedOperationsFutures(final Future<Object> withLastReplyFuture) {
111         // Combine all the previously recorded put/merge/delete operation reply Futures and the
112         // ReadyTransactionReply Future into one Future. If any one fails then the combined
113         // Future will fail. We need all prior operations and the ready operation to succeed
114         // in order to attempt commit.
115
116         List<Future<Object>> futureList = Lists.newArrayListWithCapacity(recordedOperationCount() + 1);
117         copyRecordedOperationFutures(futureList);
118         futureList.add(withLastReplyFuture);
119
120         Future<Iterable<Object>> combinedFutures = akka.dispatch.Futures.sequence(futureList,
121                 actorContext.getClientDispatcher());
122
123         // Transform the combined Future into a Future that returns the cohort actor path from
124         // the ReadyTransactionReply. That's the end result of the ready operation.
125
126         return combinedFutures.transform(new Mapper<Iterable<Object>, ActorSelection>() {
127             @Override
128             public ActorSelection checkedApply(Iterable<Object> notUsed) {
129                 LOG.debug("Tx {} readyTransaction: pending recorded operations succeeded",
130                     getIdentifier());
131
132                 // At this point all the Futures succeeded and we need to extract the cohort
133                 // actor path from the ReadyTransactionReply. For the recorded operations, they
134                 // don't return any data so we're only interested that they completed
135                 // successfully. We could be paranoid and verify the correct reply types but
136                 // that really should never happen so it's not worth the overhead of
137                 // de-serializing each reply.
138
139                 // Note the Future get call here won't block as it's complete.
140                 Object serializedReadyReply = withLastReplyFuture.value().get().get();
141                 if (serializedReadyReply instanceof ReadyTransactionReply) {
142                     return actorContext.actorSelection(((ReadyTransactionReply)serializedReadyReply).getCohortPath());
143                 } else if(serializedReadyReply instanceof BatchedModificationsReply) {
144                     return actorContext.actorSelection(((BatchedModificationsReply)serializedReadyReply).getCohortPath());
145                 } else if(serializedReadyReply.getClass().equals(ReadyTransactionReply.SERIALIZABLE_CLASS)) {
146                     ReadyTransactionReply reply = ReadyTransactionReply.fromSerializable(serializedReadyReply);
147                     String cohortPath = deserializeCohortPath(reply.getCohortPath());
148                     return actorContext.actorSelection(cohortPath);
149                 } else {
150                     // Throwing an exception here will fail the Future.
151                     throw new IllegalArgumentException(String.format("%s: Invalid reply type %s",
152                         getIdentifier(), serializedReadyReply.getClass()));
153                 }
154             }
155         }, TransactionProxy.SAME_FAILURE_TRANSFORMER, actorContext.getClientDispatcher());
156     }
157
158     protected String deserializeCohortPath(String cohortPath) {
159         return cohortPath;
160     }
161
162     private void batchModification(Modification modification) {
163         if(batchedModifications == null) {
164             batchedModifications = new BatchedModifications(getIdentifier().toString(), remoteTransactionVersion,
165                     transactionChainId);
166         }
167
168         batchedModifications.addModification(modification);
169
170         if(batchedModifications.getModifications().size() >=
171                 actorContext.getDatastoreContext().getShardBatchedModificationCount()) {
172             sendAndRecordBatchedModifications();
173         }
174     }
175
176     private void sendAndRecordBatchedModifications() {
177         Future<Object> sentFuture = sendBatchedModifications();
178         if(sentFuture != null) {
179             recordOperationFuture(sentFuture);
180         }
181     }
182
183     protected Future<Object> sendBatchedModifications() {
184         return sendBatchedModifications(false);
185     }
186
187     protected Future<Object> sendBatchedModifications(boolean ready) {
188         Future<Object> sent = null;
189         if(batchedModifications != null) {
190             if(LOG.isDebugEnabled()) {
191                 LOG.debug("Tx {} sending {} batched modifications, ready: {}", getIdentifier(),
192                         batchedModifications.getModifications().size(), ready);
193             }
194
195             batchedModifications.setReady(ready);
196             sent = executeOperationAsync(batchedModifications);
197
198             batchedModifications = new BatchedModifications(getIdentifier().toString(), remoteTransactionVersion,
199                     transactionChainId);
200         }
201
202         return sent;
203     }
204
205     @Override
206     public void deleteData(YangInstanceIdentifier path) {
207         LOG.debug("Tx {} deleteData called path = {}", getIdentifier(), path);
208
209         batchModification(new DeleteModification(path));
210     }
211
212     @Override
213     public void mergeData(YangInstanceIdentifier path, NormalizedNode<?, ?> data) {
214         LOG.debug("Tx {} mergeData called path = {}", getIdentifier(), path);
215
216         batchModification(new MergeModification(path, data));
217     }
218
219     @Override
220     public void writeData(YangInstanceIdentifier path, NormalizedNode<?, ?> data) {
221         LOG.debug("Tx {} writeData called path = {}", getIdentifier(), path);
222
223         batchModification(new WriteModification(path, data));
224     }
225
226     @Override
227     public void readData(
228             final YangInstanceIdentifier path,final SettableFuture<Optional<NormalizedNode<?, ?>>> returnFuture ) {
229
230         LOG.debug("Tx {} readData called path = {}", getIdentifier(), path);
231
232         // Send the remaining batched modifications if any.
233
234         sendAndRecordBatchedModifications();
235
236         // If there were any previous recorded put/merge/delete operation reply Futures then we
237         // must wait for them to successfully complete. This is necessary to honor the read
238         // uncommitted semantics of the public API contract. If any one fails then fail the read.
239
240         if(recordedOperationCount() == 0) {
241             finishReadData(path, returnFuture);
242         } else {
243             LOG.debug("Tx {} readData: verifying {} previous recorded operations",
244                 getIdentifier(), recordedOperationCount());
245
246             // Note: we make a copy of recordedOperationFutures to be on the safe side in case
247             // Futures#sequence accesses the passed List on a different thread, as
248             // recordedOperationFutures is not synchronized.
249
250             Future<Iterable<Object>> combinedFutures = akka.dispatch.Futures.sequence(
251                 copyRecordedOperationFutures(), actorContext.getClientDispatcher());
252
253             OnComplete<Iterable<Object>> onComplete = new OnComplete<Iterable<Object>>() {
254                 @Override
255                 public void onComplete(Throwable failure, Iterable<Object> notUsed)
256                         throws Throwable {
257                     if(failure != null) {
258                         LOG.debug("Tx {} readData: a recorded operation failed: {}",
259                             getIdentifier(), failure);
260                         returnFuture.setException(new ReadFailedException(
261                                 "The read could not be performed because a previous put, merge,"
262                                 + "or delete operation failed", failure));
263                     } else {
264                         finishReadData(path, returnFuture);
265                     }
266                 }
267             };
268
269             combinedFutures.onComplete(onComplete, actorContext.getClientDispatcher());
270         }
271
272     }
273
274     private void finishReadData(final YangInstanceIdentifier path,
275             final SettableFuture<Optional<NormalizedNode<?, ?>>> returnFuture) {
276
277         LOG.debug("Tx {} finishReadData called path = {}", getIdentifier(), path);
278
279         OnComplete<Object> onComplete = new OnComplete<Object>() {
280             @Override
281             public void onComplete(Throwable failure, Object readResponse) throws Throwable {
282                 if(failure != null) {
283                     LOG.debug("Tx {} read operation failed: {}", getIdentifier(), failure);
284                     returnFuture.setException(new ReadFailedException(
285                             "Error reading data for path " + path, failure));
286
287                 } else {
288                     LOG.debug("Tx {} read operation succeeded", getIdentifier(), failure);
289
290                     if (readResponse instanceof ReadDataReply) {
291                         ReadDataReply reply = (ReadDataReply) readResponse;
292                         returnFuture.set(Optional.<NormalizedNode<?, ?>>fromNullable(reply.getNormalizedNode()));
293
294                     } else if (ReadDataReply.isSerializedType(readResponse)) {
295                         ReadDataReply reply = ReadDataReply.fromSerializable(readResponse);
296                         returnFuture.set(Optional.<NormalizedNode<?, ?>>fromNullable(reply.getNormalizedNode()));
297
298                     } else {
299                         returnFuture.setException(new ReadFailedException(
300                             "Invalid response reading data for path " + path));
301                     }
302                 }
303             }
304         };
305
306         Future<Object> readFuture = executeOperationAsync(new ReadData(path));
307
308         readFuture.onComplete(onComplete, actorContext.getClientDispatcher());
309     }
310
311     @Override
312     public void dataExists(final YangInstanceIdentifier path, final SettableFuture<Boolean> returnFuture) {
313
314         LOG.debug("Tx {} dataExists called path = {}", getIdentifier(), path);
315
316         // Send the remaining batched modifications if any.
317
318         sendAndRecordBatchedModifications();
319
320         // If there were any previous recorded put/merge/delete operation reply Futures then we
321         // must wait for them to successfully complete. This is necessary to honor the read
322         // uncommitted semantics of the public API contract. If any one fails then fail this
323         // request.
324
325         if(recordedOperationCount() == 0) {
326             finishDataExists(path, returnFuture);
327         } else {
328             LOG.debug("Tx {} dataExists: verifying {} previous recorded operations",
329                 getIdentifier(), recordedOperationCount());
330
331             // Note: we make a copy of recordedOperationFutures to be on the safe side in case
332             // Futures#sequence accesses the passed List on a different thread, as
333             // recordedOperationFutures is not synchronized.
334
335             Future<Iterable<Object>> combinedFutures = akka.dispatch.Futures.sequence(
336                     copyRecordedOperationFutures(),
337                     actorContext.getClientDispatcher());
338             OnComplete<Iterable<Object>> onComplete = new OnComplete<Iterable<Object>>() {
339                 @Override
340                 public void onComplete(Throwable failure, Iterable<Object> notUsed)
341                         throws Throwable {
342                     if(failure != null) {
343                         LOG.debug("Tx {} dataExists: a recorded operation failed: {}",
344                             getIdentifier(), failure);
345                         returnFuture.setException(new ReadFailedException(
346                                 "The data exists could not be performed because a previous "
347                                 + "put, merge, or delete operation failed", failure));
348                     } else {
349                         finishDataExists(path, returnFuture);
350                     }
351                 }
352             };
353
354             combinedFutures.onComplete(onComplete, actorContext.getClientDispatcher());
355         }
356     }
357
358     private void finishDataExists(final YangInstanceIdentifier path,
359             final SettableFuture<Boolean> returnFuture) {
360
361         LOG.debug("Tx {} finishDataExists called path = {}", getIdentifier(), path);
362
363         OnComplete<Object> onComplete = new OnComplete<Object>() {
364             @Override
365             public void onComplete(Throwable failure, Object response) throws Throwable {
366                 if(failure != null) {
367                     LOG.debug("Tx {} dataExists operation failed: {}", getIdentifier(), failure);
368                     returnFuture.setException(new ReadFailedException(
369                             "Error checking data exists for path " + path, failure));
370                 } else {
371                     LOG.debug("Tx {} dataExists operation succeeded", getIdentifier(), failure);
372
373                     if (response instanceof DataExistsReply) {
374                         returnFuture.set(Boolean.valueOf(((DataExistsReply) response).exists()));
375
376                     } else if (response.getClass().equals(DataExistsReply.SERIALIZABLE_CLASS)) {
377                         returnFuture.set(Boolean.valueOf(DataExistsReply.fromSerializable(response).exists()));
378
379                     } else {
380                         returnFuture.setException(new ReadFailedException(
381                                 "Invalid response checking exists for path " + path));
382                     }
383                 }
384             }
385         };
386
387         Future<Object> future = executeOperationAsync(new DataExists(path));
388
389         future.onComplete(onComplete, actorContext.getClientDispatcher());
390     }
391 }