97816a55ccb5920e289eed12d665de8ca6a2ce35
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / ShardCommitCoordinator.java
1 /*
2  * Copyright (c) 2014 Brocade Communications 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.ActorRef;
11 import akka.actor.Status;
12 import akka.serialization.Serialization;
13 import com.google.common.annotations.VisibleForTesting;
14 import com.google.common.base.Preconditions;
15 import com.google.common.cache.Cache;
16 import com.google.common.cache.CacheBuilder;
17 import com.google.common.cache.RemovalCause;
18 import com.google.common.cache.RemovalListener;
19 import com.google.common.cache.RemovalNotification;
20 import java.util.LinkedList;
21 import java.util.Queue;
22 import java.util.concurrent.ExecutionException;
23 import java.util.concurrent.TimeUnit;
24 import org.opendaylight.controller.cluster.datastore.compat.BackwardsCompatibleThreePhaseCommitCohort;
25 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
26 import org.opendaylight.controller.cluster.datastore.messages.BatchedModificationsReply;
27 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
28 import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction;
29 import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction;
30 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
31 import org.opendaylight.controller.cluster.datastore.modification.Modification;
32 import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
33 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
34 import org.slf4j.Logger;
35
36 /**
37  * Coordinates commits for a shard ensuring only one concurrent 3-phase commit.
38  *
39  * @author Thomas Pantelis
40  */
41 public class ShardCommitCoordinator {
42
43     // Interface hook for unit tests to replace or decorate the DOMStoreThreePhaseCommitCohorts.
44     public interface CohortDecorator {
45         ShardDataTreeCohort decorate(String transactionID, ShardDataTreeCohort actual);
46     }
47
48     private final Cache<String, CohortEntry> cohortCache;
49
50     private CohortEntry currentCohortEntry;
51
52     private final ShardDataTree dataTree;
53
54     private final Queue<CohortEntry> queuedCohortEntries;
55
56     private int queueCapacity;
57
58     private final Logger log;
59
60     private final String name;
61
62     private final RemovalListener<String, CohortEntry> cacheRemovalListener =
63             new RemovalListener<String, CohortEntry>() {
64                 @Override
65                 public void onRemoval(RemovalNotification<String, CohortEntry> notification) {
66                     if(notification.getCause() == RemovalCause.EXPIRED) {
67                         log.warn("{}: Transaction {} was timed out of the cache", name, notification.getKey());
68                     }
69                 }
70             };
71
72     // This is a hook for unit tests to replace or decorate the DOMStoreThreePhaseCommitCohorts.
73     private CohortDecorator cohortDecorator;
74
75     private ReadyTransactionReply readyTransactionReply;
76
77     public ShardCommitCoordinator(ShardDataTree dataTree,
78             long cacheExpiryTimeoutInSec, int queueCapacity, ActorRef shardActor, Logger log, String name) {
79
80         this.queueCapacity = queueCapacity;
81         this.log = log;
82         this.name = name;
83         this.dataTree = Preconditions.checkNotNull(dataTree);
84
85         cohortCache = CacheBuilder.newBuilder().expireAfterAccess(cacheExpiryTimeoutInSec, TimeUnit.SECONDS).
86                 removalListener(cacheRemovalListener).build();
87
88         // We use a LinkedList here to avoid synchronization overhead with concurrent queue impls
89         // since this should only be accessed on the shard's dispatcher.
90         queuedCohortEntries = new LinkedList<>();
91     }
92
93     public void setQueueCapacity(int queueCapacity) {
94         this.queueCapacity = queueCapacity;
95     }
96
97     private ReadyTransactionReply readyTransactionReply(Shard shard) {
98         if(readyTransactionReply == null) {
99             readyTransactionReply = new ReadyTransactionReply(Serialization.serializedActorPath(shard.self()));
100         }
101
102         return readyTransactionReply;
103     }
104
105     /**
106      * This method is called to ready a transaction that was prepared by ShardTransaction actor. It caches
107      * the prepared cohort entry for the given transactions ID in preparation for the subsequent 3-phase commit.
108      */
109     public void handleForwardedReadyTransaction(ForwardedReadyTransaction ready, ActorRef sender, Shard shard) {
110         log.debug("{}: Readying transaction {}, client version {}", name,
111                 ready.getTransactionID(), ready.getTxnClientVersion());
112
113         CohortEntry cohortEntry = new CohortEntry(ready.getTransactionID(), ready.getCohort(),
114                 (MutableCompositeModification) ready.getModification());
115         cohortCache.put(ready.getTransactionID(), cohortEntry);
116
117         if(ready.getTxnClientVersion() < DataStoreVersions.LITHIUM_VERSION) {
118             // Return our actor path as we'll handle the three phase commit except if the Tx client
119             // version < Helium-1 version which means the Tx was initiated by a base Helium version node.
120             // In that case, the subsequent 3-phase commit messages won't contain the transactionId so to
121             // maintain backwards compatibility, we create a separate cohort actor to provide the compatible behavior.
122             ActorRef replyActorPath = shard.self();
123             if(ready.getTxnClientVersion() < DataStoreVersions.HELIUM_1_VERSION) {
124                 log.debug("{}: Creating BackwardsCompatibleThreePhaseCommitCohort", name);
125                 replyActorPath = shard.getContext().actorOf(BackwardsCompatibleThreePhaseCommitCohort.props(
126                         ready.getTransactionID()));
127             }
128
129             ReadyTransactionReply readyTransactionReply =
130                     new ReadyTransactionReply(Serialization.serializedActorPath(replyActorPath),
131                             ready.getTxnClientVersion());
132             sender.tell(ready.isReturnSerialized() ? readyTransactionReply.toSerializable() :
133                 readyTransactionReply, shard.self());
134         } else {
135             if(ready.isDoImmediateCommit()) {
136                 cohortEntry.setDoImmediateCommit(true);
137                 cohortEntry.setReplySender(sender);
138                 cohortEntry.setShard(shard);
139                 handleCanCommit(cohortEntry);
140             } else {
141                 // The caller does not want immediate commit - the 3-phase commit will be coordinated by the
142                 // front-end so send back a ReadyTransactionReply with our actor path.
143                 sender.tell(readyTransactionReply(shard), shard.self());
144             }
145         }
146     }
147
148     /**
149      * This method handles a BatchedModifications message for a transaction being prepared directly on the
150      * Shard actor instead of via a ShardTransaction actor. If there's no currently cached
151      * DOMStoreWriteTransaction, one is created. The batched modifications are applied to the write Tx. If
152      * the BatchedModifications is ready to commit then a DOMStoreThreePhaseCommitCohort is created.
153      *
154      * @param batched the BatchedModifications
155      * @param shardActor the transaction's shard actor
156      *
157      * @throws ExecutionException if an error occurs loading the cache
158      */
159     boolean handleBatchedModifications(BatchedModifications batched, ActorRef sender, Shard shard)
160             throws ExecutionException {
161         CohortEntry cohortEntry = cohortCache.getIfPresent(batched.getTransactionID());
162         if(cohortEntry == null) {
163             cohortEntry = new CohortEntry(batched.getTransactionID(),
164                     dataTree.newReadWriteTransaction(batched.getTransactionID(),
165                         batched.getTransactionChainID()));
166             cohortCache.put(batched.getTransactionID(), cohortEntry);
167         }
168
169         if(log.isDebugEnabled()) {
170             log.debug("{}: Applying {} batched modifications for Tx {}", name,
171                     batched.getModifications().size(), batched.getTransactionID());
172         }
173
174         cohortEntry.applyModifications(batched.getModifications());
175
176         if(batched.isReady()) {
177             if(log.isDebugEnabled()) {
178                 log.debug("{}: Readying Tx {}, client version {}", name,
179                         batched.getTransactionID(), batched.getVersion());
180             }
181
182             cohortEntry.ready(cohortDecorator, batched.isDoCommitOnReady());
183
184             if(batched.isDoCommitOnReady()) {
185                 cohortEntry.setReplySender(sender);
186                 cohortEntry.setShard(shard);
187                 handleCanCommit(cohortEntry);
188             } else {
189                 sender.tell(readyTransactionReply(shard), shard.self());
190             }
191         } else {
192             sender.tell(new BatchedModificationsReply(batched.getModifications().size()), shard.self());
193         }
194
195         return batched.isReady();
196     }
197
198     /**
199      * This method handles {@link ReadyLocalTransaction} message. All transaction modifications have
200      * been prepared beforehand by the sender and we just need to drive them through into the dataTree.
201      *
202      * @param message
203      * @param sender
204      * @param shard
205      */
206     void handleReadyLocalTransaction(ReadyLocalTransaction message, ActorRef sender, Shard shard) {
207         final ShardDataTreeCohort cohort = new SimpleShardDataTreeCohort(dataTree, message.getModification());
208         final CohortEntry cohortEntry = new CohortEntry(message.getTransactionID(), cohort);
209         cohortCache.put(message.getTransactionID(), cohortEntry);
210         cohortEntry.setDoImmediateCommit(message.isDoCommitOnReady());
211         log.debug("{}: Applying local modifications for Tx {}", name, message.getTransactionID());
212
213         if (message.isDoCommitOnReady()) {
214             cohortEntry.setReplySender(sender);
215             cohortEntry.setShard(shard);
216             handleCanCommit(cohortEntry);
217         } else {
218             sender.tell(readyTransactionReply(shard), shard.self());
219         }
220     }
221
222     private void handleCanCommit(CohortEntry cohortEntry) {
223         String transactionID = cohortEntry.getTransactionID();
224
225         if(log.isDebugEnabled()) {
226             log.debug("{}: Processing canCommit for transaction {} for shard {}",
227                     name, transactionID, cohortEntry.getShard().self().path());
228         }
229
230         if(currentCohortEntry != null) {
231             // There's already a Tx commit in progress - attempt to queue this entry to be
232             // committed after the current Tx completes.
233             log.debug("{}: Transaction {} is already in progress - queueing transaction {}",
234                     name, currentCohortEntry.getTransactionID(), transactionID);
235
236             if(queuedCohortEntries.size() < queueCapacity) {
237                 queuedCohortEntries.offer(cohortEntry);
238             } else {
239                 removeCohortEntry(transactionID);
240
241                 RuntimeException ex = new RuntimeException(
242                         String.format("%s: Could not enqueue transaction %s - the maximum commit queue"+
243                                       " capacity %d has been reached.",
244                                       name, transactionID, queueCapacity));
245                 log.error(ex.getMessage());
246                 cohortEntry.getReplySender().tell(new Status.Failure(ex), cohortEntry.getShard().self());
247             }
248         } else {
249             // No Tx commit currently in progress - make this the current entry and proceed with
250             // canCommit.
251             cohortEntry.updateLastAccessTime();
252             currentCohortEntry = cohortEntry;
253
254             doCanCommit(cohortEntry);
255         }
256     }
257
258     /**
259      * This method handles the canCommit phase for a transaction.
260      *
261      * @param canCommit the CanCommitTransaction message
262      * @param sender the actor that sent the message
263      * @param shard the transaction's shard actor
264      */
265     public void handleCanCommit(String transactionID, final ActorRef sender, final Shard shard) {
266         // Lookup the cohort entry that was cached previously (or should have been) by
267         // transactionReady (via the ForwardedReadyTransaction message).
268         final CohortEntry cohortEntry = cohortCache.getIfPresent(transactionID);
269         if(cohortEntry == null) {
270             // Either canCommit was invoked before ready(shouldn't happen)  or a long time passed
271             // between canCommit and ready and the entry was expired from the cache.
272             IllegalStateException ex = new IllegalStateException(
273                     String.format("%s: No cohort entry found for transaction %s", name, transactionID));
274             log.error(ex.getMessage());
275             sender.tell(new Status.Failure(ex), shard.self());
276             return;
277         }
278
279         cohortEntry.setReplySender(sender);
280         cohortEntry.setShard(shard);
281
282         handleCanCommit(cohortEntry);
283     }
284
285     private void doCanCommit(final CohortEntry cohortEntry) {
286
287         boolean canCommit = false;
288         try {
289             // We block on the future here so we don't have to worry about possibly accessing our
290             // state on a different thread outside of our dispatcher. Also, the data store
291             // currently uses a same thread executor anyway.
292             canCommit = cohortEntry.getCohort().canCommit().get();
293
294             if(cohortEntry.isDoImmediateCommit()) {
295                 if(canCommit) {
296                     doCommit(cohortEntry);
297                 } else {
298                     cohortEntry.getReplySender().tell(new Status.Failure(new TransactionCommitFailedException(
299                                 "Can Commit failed, no detailed cause available.")), cohortEntry.getShard().self());
300                 }
301             } else {
302                 cohortEntry.getReplySender().tell(
303                         canCommit ? CanCommitTransactionReply.YES.toSerializable() :
304                             CanCommitTransactionReply.NO.toSerializable(), cohortEntry.getShard().self());
305             }
306         } catch (Exception e) {
307             log.debug("{}: An exception occurred during canCommit: {}", name, e);
308
309             Throwable failure = e;
310             if(e instanceof ExecutionException) {
311                 failure = e.getCause();
312             }
313
314             cohortEntry.getReplySender().tell(new Status.Failure(failure), cohortEntry.getShard().self());
315         } finally {
316             if(!canCommit) {
317                 // Remove the entry from the cache now.
318                 currentTransactionComplete(cohortEntry.getTransactionID(), true);
319             }
320         }
321     }
322
323     private boolean doCommit(CohortEntry cohortEntry) {
324         log.debug("{}: Committing transaction {}", name, cohortEntry.getTransactionID());
325
326         boolean success = false;
327
328         // We perform the preCommit phase here atomically with the commit phase. This is an
329         // optimization to eliminate the overhead of an extra preCommit message. We lose front-end
330         // coordination of preCommit across shards in case of failure but preCommit should not
331         // normally fail since we ensure only one concurrent 3-phase commit.
332
333         try {
334             // We block on the future here so we don't have to worry about possibly accessing our
335             // state on a different thread outside of our dispatcher. Also, the data store
336             // currently uses a same thread executor anyway.
337             cohortEntry.getCohort().preCommit().get();
338
339             cohortEntry.getShard().continueCommit(cohortEntry);
340
341             cohortEntry.updateLastAccessTime();
342
343             success = true;
344         } catch (Exception e) {
345             log.error("{} An exception occurred while preCommitting transaction {}",
346                     name, cohortEntry.getTransactionID(), e);
347             cohortEntry.getReplySender().tell(new akka.actor.Status.Failure(e), cohortEntry.getShard().self());
348
349             currentTransactionComplete(cohortEntry.getTransactionID(), true);
350         }
351
352         return success;
353     }
354
355     boolean handleCommit(final String transactionID, final ActorRef sender, final Shard shard) {
356         // Get the current in-progress cohort entry in the commitCoordinator if it corresponds to
357         // this transaction.
358         final CohortEntry cohortEntry = getCohortEntryIfCurrent(transactionID);
359         if(cohortEntry == null) {
360             // We're not the current Tx - the Tx was likely expired b/c it took too long in
361             // between the canCommit and commit messages.
362             IllegalStateException ex = new IllegalStateException(
363                     String.format("%s: Cannot commit transaction %s - it is not the current transaction",
364                             name, transactionID));
365             log.error(ex.getMessage());
366             sender.tell(new akka.actor.Status.Failure(ex), shard.self());
367             return false;
368         }
369
370         return doCommit(cohortEntry);
371     }
372
373     /**
374      * Returns the cohort entry for the Tx commit currently in progress if the given transaction ID
375      * matches the current entry.
376      *
377      * @param transactionID the ID of the transaction
378      * @return the current CohortEntry or null if the given transaction ID does not match the
379      *         current entry.
380      */
381     public CohortEntry getCohortEntryIfCurrent(String transactionID) {
382         if(isCurrentTransaction(transactionID)) {
383             return currentCohortEntry;
384         }
385
386         return null;
387     }
388
389     public CohortEntry getCurrentCohortEntry() {
390         return currentCohortEntry;
391     }
392
393     public CohortEntry getAndRemoveCohortEntry(String transactionID) {
394         CohortEntry cohortEntry = cohortCache.getIfPresent(transactionID);
395         cohortCache.invalidate(transactionID);
396         return cohortEntry;
397     }
398
399     public void removeCohortEntry(String transactionID) {
400         cohortCache.invalidate(transactionID);
401     }
402
403     public boolean isCurrentTransaction(String transactionID) {
404         return currentCohortEntry != null &&
405                 currentCohortEntry.getTransactionID().equals(transactionID);
406     }
407
408     /**
409      * This method is called when a transaction is complete, successful or not. If the given
410      * given transaction ID matches the current in-progress transaction, the next cohort entry,
411      * if any, is dequeued and processed.
412      *
413      * @param transactionID the ID of the completed transaction
414      * @param removeCohortEntry if true the CohortEntry for the transaction is also removed from
415      *        the cache.
416      */
417     public void currentTransactionComplete(String transactionID, boolean removeCohortEntry) {
418         if(removeCohortEntry) {
419             removeCohortEntry(transactionID);
420         }
421
422         if(isCurrentTransaction(transactionID)) {
423             // Dequeue the next cohort entry waiting in the queue.
424             currentCohortEntry = queuedCohortEntries.poll();
425             if(currentCohortEntry != null) {
426                 currentCohortEntry.updateLastAccessTime();
427                 doCanCommit(currentCohortEntry);
428             }
429         }
430     }
431
432     @VisibleForTesting
433     void setCohortDecorator(CohortDecorator cohortDecorator) {
434         this.cohortDecorator = cohortDecorator;
435     }
436
437
438     static class CohortEntry {
439         private final String transactionID;
440         private ShardDataTreeCohort cohort;
441         private final ReadWriteShardDataTreeTransaction transaction;
442         private ActorRef replySender;
443         private Shard shard;
444         private long lastAccessTime;
445         private boolean doImmediateCommit;
446
447         CohortEntry(String transactionID, ReadWriteShardDataTreeTransaction transaction) {
448             this.transaction = Preconditions.checkNotNull(transaction);
449             this.transactionID = transactionID;
450         }
451
452         CohortEntry(String transactionID, ShardDataTreeCohort cohort,
453                 MutableCompositeModification compositeModification) {
454             this.transactionID = transactionID;
455             this.cohort = cohort;
456             this.transaction = null;
457         }
458
459         CohortEntry(String transactionID, ShardDataTreeCohort cohort) {
460             this.transactionID = transactionID;
461             this.cohort = cohort;
462             this.transaction = null;
463         }
464
465         void updateLastAccessTime() {
466             lastAccessTime = System.currentTimeMillis();
467         }
468
469         long getLastAccessTime() {
470             return lastAccessTime;
471         }
472
473         String getTransactionID() {
474             return transactionID;
475         }
476
477         ShardDataTreeCohort getCohort() {
478             return cohort;
479         }
480
481         void applyModifications(Iterable<Modification> modifications) {
482             for (Modification modification : modifications) {
483                 modification.apply(transaction.getSnapshot());
484             }
485         }
486
487         void ready(CohortDecorator cohortDecorator, boolean doImmediateCommit) {
488             Preconditions.checkState(cohort == null, "cohort was already set");
489
490             setDoImmediateCommit(doImmediateCommit);
491
492             cohort = transaction.ready();
493
494             if(cohortDecorator != null) {
495                 // Call the hook for unit tests.
496                 cohort = cohortDecorator.decorate(transactionID, cohort);
497             }
498         }
499
500         boolean isDoImmediateCommit() {
501             return doImmediateCommit;
502         }
503
504         void setDoImmediateCommit(boolean doImmediateCommit) {
505             this.doImmediateCommit = doImmediateCommit;
506         }
507
508         ActorRef getReplySender() {
509             return replySender;
510         }
511
512         void setReplySender(ActorRef replySender) {
513             this.replySender = replySender;
514         }
515
516         Shard getShard() {
517             return shard;
518         }
519
520         void setShard(Shard shard) {
521             this.shard = shard;
522         }
523     }
524 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.