Fix checkstyle issues in module sal-dom-broker
[controller.git] / opendaylight / md-sal / sal-dom-broker / src / main / java / org / opendaylight / controller / md / sal / dom / broker / impl / CommitCoordinationTask.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
9 package org.opendaylight.controller.md.sal.dom.broker.impl;
10
11 import com.google.common.base.Preconditions;
12 import com.google.common.base.Throwables;
13 import com.google.common.util.concurrent.Futures;
14 import com.google.common.util.concurrent.ListenableFuture;
15 import java.util.Collection;
16 import java.util.concurrent.Callable;
17 import java.util.concurrent.ExecutionException;
18 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
19 import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
20 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
21 import org.opendaylight.yangtools.util.DurationStatisticsTracker;
22 import org.slf4j.Logger;
23 import org.slf4j.LoggerFactory;
24
25 /**
26  * Implementation of blocking three-phase commit-coordination tasks without
27  * support of cancellation.
28  */
29 final class CommitCoordinationTask implements Callable<Void> {
30     private enum Phase {
31         canCommit,
32         preCommit,
33         doCommit,
34     }
35
36     private static final Logger LOG = LoggerFactory.getLogger(CommitCoordinationTask.class);
37     private final Collection<DOMStoreThreePhaseCommitCohort> cohorts;
38     private final DurationStatisticsTracker commitStatTracker;
39     private final DOMDataWriteTransaction tx;
40
41     CommitCoordinationTask(final DOMDataWriteTransaction transaction,
42             final Collection<DOMStoreThreePhaseCommitCohort> cohorts,
43             final DurationStatisticsTracker commitStatTracker) {
44         this.tx = Preconditions.checkNotNull(transaction, "transaction must not be null");
45         this.cohorts = Preconditions.checkNotNull(cohorts, "cohorts must not be null");
46         this.commitStatTracker = commitStatTracker;
47     }
48
49     @Override
50     public Void call() throws TransactionCommitFailedException {
51         final long startTime = commitStatTracker != null ? System.nanoTime() : 0;
52
53         Phase phase = Phase.canCommit;
54
55         try {
56             LOG.debug("Transaction {}: canCommit Started", tx.getIdentifier());
57             canCommitBlocking();
58
59             phase = Phase.preCommit;
60             LOG.debug("Transaction {}: preCommit Started", tx.getIdentifier());
61             preCommitBlocking();
62
63             phase = Phase.doCommit;
64             LOG.debug("Transaction {}: doCommit Started", tx.getIdentifier());
65             commitBlocking();
66
67             LOG.debug("Transaction {}: doCommit completed", tx.getIdentifier());
68             return null;
69         } catch (final TransactionCommitFailedException e) {
70             LOG.warn("Tx: {} Error during phase {}, starting Abort", tx.getIdentifier(), phase, e);
71             abortBlocking(e);
72             throw e;
73         } finally {
74             if (commitStatTracker != null) {
75                 commitStatTracker.addDuration(System.nanoTime() - startTime);
76             }
77         }
78     }
79
80     /**
81      * Invokes canCommit on underlying cohorts and blocks till
82      * all results are returned.
83      *
84      * <p>
85      * Valid state transition is from SUBMITTED to CAN_COMMIT,
86      * if currentPhase is not SUBMITTED throws IllegalStateException.
87      *
88      * @throws TransactionCommitFailedException
89      *             If one of cohorts failed can Commit
90      *
91      */
92     private void canCommitBlocking() throws TransactionCommitFailedException {
93         for (final ListenableFuture<?> canCommit : canCommitAll()) {
94             try {
95                 final Boolean result = (Boolean)canCommit.get();
96                 if (result == null || !result) {
97                     throw new TransactionCommitFailedException("Can Commit failed, no detailed cause available.");
98                 }
99             } catch (InterruptedException | ExecutionException e) {
100                 throw TransactionCommitFailedExceptionMapper.CAN_COMMIT_ERROR_MAPPER.apply(e);
101             }
102         }
103     }
104
105     /**
106      * Invokes canCommit on underlying cohorts and returns composite future
107      * which will contains {@link Boolean#TRUE} only and only if
108      * all cohorts returned true.
109      *
110      * <p>
111      * Valid state transition is from SUBMITTED to CAN_COMMIT,
112      * if currentPhase is not SUBMITTED throws IllegalStateException.
113      *
114      * @return List of all cohorts futures from can commit phase.
115      *
116      */
117     private ListenableFuture<?>[] canCommitAll() {
118         final ListenableFuture<?>[] ops = new ListenableFuture<?>[cohorts.size()];
119         int index = 0;
120         for (final DOMStoreThreePhaseCommitCohort cohort : cohorts) {
121             ops[index++] = cohort.canCommit();
122         }
123         return ops;
124     }
125
126     /**
127      * Invokes preCommit on underlying cohorts and blocks till
128      * all results are returned.
129      *
130      * <p>
131      * Valid state transition is from CAN_COMMIT to PRE_COMMIT, if current
132      * state is not CAN_COMMIT
133      * throws IllegalStateException.
134      *
135      * @throws TransactionCommitFailedException
136      *             If one of cohorts failed preCommit
137      *
138      */
139     private void preCommitBlocking() throws TransactionCommitFailedException {
140         final ListenableFuture<?>[] preCommitFutures = preCommitAll();
141         try {
142             for (final ListenableFuture<?> future : preCommitFutures) {
143                 future.get();
144             }
145         } catch (InterruptedException | ExecutionException e) {
146             throw TransactionCommitFailedExceptionMapper.PRE_COMMIT_MAPPER.apply(e);
147         }
148     }
149
150     /**
151      * Invokes preCommit on underlying cohorts and returns future
152      * which will complete once all preCommit on cohorts completed or
153      * failed.
154      *
155      * <p>
156      * Valid state transition is from CAN_COMMIT to PRE_COMMIT, if current
157      * state is not CAN_COMMIT
158      * throws IllegalStateException.
159      *
160      * @return List of all cohorts futures from can commit phase.
161      *
162      */
163     private ListenableFuture<?>[] preCommitAll() {
164         final ListenableFuture<?>[] ops = new ListenableFuture<?>[cohorts.size()];
165         int index = 0;
166         for (final DOMStoreThreePhaseCommitCohort cohort : cohorts) {
167             ops[index++] = cohort.preCommit();
168         }
169         return ops;
170     }
171
172     /**
173      * Invokes commit on underlying cohorts and blocks till
174      * all results are returned.
175      *
176      * <p>
177      * Valid state transition is from PRE_COMMIT to COMMIT, if not throws
178      * IllegalStateException.
179      *
180      * @throws TransactionCommitFailedException
181      *             If one of cohorts failed preCommit
182      *
183      */
184     private void commitBlocking() throws TransactionCommitFailedException {
185         final ListenableFuture<?>[] commitFutures = commitAll();
186         try {
187             for (final ListenableFuture<?> future : commitFutures) {
188                 future.get();
189             }
190         } catch (InterruptedException | ExecutionException e) {
191             throw TransactionCommitFailedExceptionMapper.COMMIT_ERROR_MAPPER.apply(e);
192         }
193     }
194
195     /**
196      * Invokes commit on underlying cohorts and returns future which
197      * completes
198      * once all commits on cohorts are completed.
199      *
200      * <p>
201      * Valid state transition is from PRE_COMMIT to COMMIT, if not throws
202      * IllegalStateException
203      *
204      * @return List of all cohorts futures from can commit phase.
205      */
206     private ListenableFuture<?>[] commitAll() {
207         final ListenableFuture<?>[] ops = new ListenableFuture<?>[cohorts.size()];
208         int index = 0;
209         for (final DOMStoreThreePhaseCommitCohort cohort : cohorts) {
210             ops[index++] = cohort.commit();
211         }
212         return ops;
213     }
214
215     /**
216      * Aborts transaction.
217      *
218      * <p>
219      * Invokes {@link DOMStoreThreePhaseCommitCohort#abort()} on all
220      * cohorts, blocks
221      * for all results. If any of the abort failed throws
222      * IllegalStateException,
223      * which will contains originalCause as suppressed Exception.
224      *
225      * <p>
226      * If aborts we're successful throws supplied exception
227      *
228      * @param originalCause
229      *            Exception which should be used to fail transaction for
230      *            consumers of transaction
231      *            future and listeners of transaction failure.
232      * @param phase phase in which the problem ensued
233      * @throws TransactionCommitFailedException
234      *             on invocation of this method.
235      *             originalCa
236      * @throws IllegalStateException
237      *             if abort failed.
238      */
239     private void abortBlocking(
240             final TransactionCommitFailedException originalCause) throws TransactionCommitFailedException {
241         Exception cause = originalCause;
242         try {
243             abortAsyncAll().get();
244         } catch (InterruptedException | ExecutionException e) {
245             LOG.error("Tx: {} Error during Abort.", tx.getIdentifier(), e);
246             cause = new IllegalStateException("Abort failed.", e);
247             cause.addSuppressed(e);
248         }
249         Throwables.propagateIfPossible(cause, TransactionCommitFailedException.class);
250     }
251
252     /**
253      * Invokes abort on underlying cohorts and returns future which
254      * completes once all abort on cohorts are completed.
255      *
256      * @return Future which will complete once all cohorts completed
257      *         abort.
258      */
259     @SuppressWarnings({"unchecked", "rawtypes"})
260     private ListenableFuture<Void> abortAsyncAll() {
261
262         final ListenableFuture<?>[] ops = new ListenableFuture<?>[cohorts.size()];
263         int index = 0;
264         for (final DOMStoreThreePhaseCommitCohort cohort : cohorts) {
265             ops[index++] = cohort.abort();
266         }
267
268         /*
269          * We are returning all futures as list, not only succeeded ones in
270          * order to fail composite future if any of them failed.
271          * See Futures.allAsList for this description.
272          */
273         return (ListenableFuture) Futures.allAsList(ops);
274     }
275 }

©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.