Fix a race PingPongTransactionChain
[controller.git] / opendaylight / md-sal / sal-dom-broker / src / main / java / org / opendaylight / controller / md / sal / dom / broker / impl / PingPongTransactionChain.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 package org.opendaylight.controller.md.sal.dom.broker.impl;
9
10 import com.google.common.base.Optional;
11 import com.google.common.base.Preconditions;
12 import com.google.common.util.concurrent.CheckedFuture;
13 import com.google.common.util.concurrent.FutureCallback;
14 import com.google.common.util.concurrent.Futures;
15 import com.google.common.util.concurrent.ListenableFuture;
16 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
17 import javax.annotation.Nonnull;
18 import javax.annotation.concurrent.GuardedBy;
19 import org.opendaylight.controller.md.sal.common.api.TransactionStatus;
20 import org.opendaylight.controller.md.sal.common.api.data.AsyncTransaction;
21 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
22 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
23 import org.opendaylight.controller.md.sal.common.api.data.TransactionChain;
24 import org.opendaylight.controller.md.sal.common.api.data.TransactionChainListener;
25 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
26 import org.opendaylight.controller.md.sal.dom.api.DOMDataBroker;
27 import org.opendaylight.controller.md.sal.dom.api.DOMDataReadOnlyTransaction;
28 import org.opendaylight.controller.md.sal.dom.api.DOMDataReadWriteTransaction;
29 import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
30 import org.opendaylight.controller.md.sal.dom.api.DOMTransactionChain;
31 import org.opendaylight.controller.md.sal.dom.spi.ForwardingDOMDataReadWriteTransaction;
32 import org.opendaylight.yangtools.yang.common.RpcResult;
33 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
34 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
35 import org.slf4j.Logger;
36 import org.slf4j.LoggerFactory;
37
38 /**
39  * An implementation of {@link DOMTransactionChain}, which has a very specific
40  * behavior, which some users may find surprising. If keeps the general
41  * intent of the contract, but it makes sure there are never more than two
42  * transactions allocated at any given time: one of them is being committed,
43  * and while that is happening, the other one acts as the scratch pad. Once
44  * the committing transaction completes successfully, the scratch transaction
45  * is enqueued as soon as it is ready.
46  *
47  * This mode of operation means that there is no inherent isolation between
48  * the front-end transactions and transactions cannot be reasonably cancelled.
49  *
50  * It furthermore means that the transactions returned by {@link #newReadOnlyTransaction()}
51  * counts as an outstanding transaction and the user may not allocate multiple
52  * read-only transactions at the same time.
53  */
54 public final class PingPongTransactionChain implements DOMTransactionChain {
55     private static final Logger LOG = LoggerFactory.getLogger(PingPongTransactionChain.class);
56     private final DOMTransactionChain delegate;
57
58     @GuardedBy("this")
59     private boolean failed;
60
61     /**
62      * This updater is used to manipulate the "ready" transaction. We perform only atomic
63      * get-and-set on it.
64      */
65     private static final AtomicReferenceFieldUpdater<PingPongTransactionChain, PingPongTransaction> READY_UPDATER =
66             AtomicReferenceFieldUpdater.newUpdater(PingPongTransactionChain.class, PingPongTransaction.class, "readyTx");
67     private volatile PingPongTransaction readyTx;
68
69     /**
70      * This updater is used to manipulate the "locked" transaction. A locked transaction
71      * means we know that the user still holds a transaction and should at some point call
72      * us. We perform on compare-and-swap to ensure we properly detect when a user is
73      * attempting to allocated multiple transactions concurrently.
74      */
75     private static final AtomicReferenceFieldUpdater<PingPongTransactionChain, PingPongTransaction> LOCKED_UPDATER =
76             AtomicReferenceFieldUpdater.newUpdater(PingPongTransactionChain.class, PingPongTransaction.class, "lockedTx");
77     private volatile PingPongTransaction lockedTx;
78
79     /**
80      * This updater is used to manipulate the "inflight" transaction. There can be at most
81      * one of these at any given time. We perform only compare-and-swap on these.
82      */
83     private static final AtomicReferenceFieldUpdater<PingPongTransactionChain, PingPongTransaction> INFLIGHT_UPDATER =
84             AtomicReferenceFieldUpdater.newUpdater(PingPongTransactionChain.class, PingPongTransaction.class, "inflightTx");
85     private volatile PingPongTransaction inflightTx;
86
87     PingPongTransactionChain(final DOMDataBroker broker, final TransactionChainListener listener) {
88         this.delegate = broker.createTransactionChain(new TransactionChainListener() {
89             @Override
90             public void onTransactionChainFailed(final TransactionChain<?, ?> chain, final AsyncTransaction<?, ?> transaction, final Throwable cause) {
91                 LOG.debug("Delegate chain {} reported failure in {}", chain, transaction, cause);
92
93                 final DOMDataReadWriteTransaction frontend;
94                 final PingPongTransaction tx = inflightTx;
95                 if (tx == null) {
96                     LOG.warn("Transaction chain {} failed with no pending transactions", chain);
97                     frontend = null;
98                 } else {
99                     frontend = tx.getFrontendTransaction();
100                 }
101
102                 listener.onTransactionChainFailed(PingPongTransactionChain.this, frontend, cause);
103                 delegateFailed();
104             }
105
106             @Override
107             public void onTransactionChainSuccessful(final TransactionChain<?, ?> chain) {
108                 listener.onTransactionChainSuccessful(PingPongTransactionChain.this);
109             }
110         });
111     }
112
113     private synchronized void delegateFailed() {
114         failed = true;
115
116         /*
117          * If we do not have a locked transaction, we need to ensure that
118          * the backend transaction is cancelled. Otherwise we can defer
119          * until the user calls us.
120          */
121         if (lockedTx == null) {
122             processIfReady();
123         }
124     }
125
126     private synchronized PingPongTransaction slowAllocateTransaction() {
127         final DOMDataReadWriteTransaction delegateTx = delegate.newReadWriteTransaction();
128         final PingPongTransaction newTx = new PingPongTransaction(delegateTx);
129
130         if (!LOCKED_UPDATER.compareAndSet(this, null, newTx)) {
131             delegateTx.cancel();
132             throw new IllegalStateException(String.format("New transaction %s raced with transacion %s", newTx, lockedTx));
133         }
134
135         return newTx;
136     }
137
138     private PingPongTransaction allocateTransaction() {
139         // Step 1: acquire current state
140         final PingPongTransaction oldTx = READY_UPDATER.getAndSet(this, null);
141
142         // Slow path: allocate a delegate transaction
143         if (oldTx == null) {
144             return slowAllocateTransaction();
145         }
146
147         // Fast path: reuse current transaction. We will check
148         //            failures and similar on submit().
149         if (!LOCKED_UPDATER.compareAndSet(this, null, oldTx)) {
150             // Ouch. Delegate chain has not detected a duplicate
151             // transaction allocation. This is the best we can do.
152             oldTx.getTransaction().cancel();
153             throw new IllegalStateException(String.format("Reusable transaction %s raced with transaction %s", oldTx, lockedTx));
154         }
155
156         return oldTx;
157     }
158
159     /*
160      * This forces allocateTransaction() on a slow path, which has to happen after
161      * this method has completed executing.
162      */
163     @GuardedBy("this")
164     private void processIfReady() {
165         final PingPongTransaction tx = READY_UPDATER.getAndSet(this, null);
166         if (tx != null) {
167             processTransaction(tx);
168         }
169     }
170
171     /**
172      * Process a ready transaction. The caller needs to ensure that
173      * each transaction is seen only once by this method.
174      *
175      * @param tx Transaction which needs processing.
176      */
177     @GuardedBy("this")
178     private void processTransaction(final @Nonnull PingPongTransaction tx) {
179         if (failed) {
180             LOG.debug("Cancelling transaction {}", tx);
181             tx.getTransaction().cancel();
182             return;
183         }
184
185         LOG.debug("Submitting transaction {}", tx);
186         if (!INFLIGHT_UPDATER.compareAndSet(this, null, tx)) {
187             LOG.warn("Submitting transaction {} while {} is still running", tx, inflightTx);
188         }
189
190         Futures.addCallback(tx.getTransaction().submit(), new FutureCallback<Void>() {
191             @Override
192             public void onSuccess(final Void result) {
193                 transactionSuccessful(tx, result);
194             }
195
196             @Override
197             public void onFailure(final Throwable t) {
198                 transactionFailed(tx, t);
199             }
200         });
201     }
202
203     private void transactionSuccessful(final PingPongTransaction tx, final Void result) {
204         LOG.debug("Transaction {} completed successfully", tx);
205
206         final boolean success = INFLIGHT_UPDATER.compareAndSet(this, tx, null);
207         Preconditions.checkState(success, "Successful transaction %s while %s was submitted", tx, inflightTx);
208
209         synchronized (this) {
210             processIfReady();
211         }
212
213         // Can run unsynchronized
214         tx.onSuccess(result);
215     }
216
217     private void transactionFailed(final PingPongTransaction tx, final Throwable t) {
218         LOG.debug("Transaction {} failed", tx, t);
219
220         final boolean success = INFLIGHT_UPDATER.compareAndSet(this, tx, null);
221         Preconditions.checkState(success, "Failed transaction %s while %s was submitted", tx, inflightTx);
222
223         tx.onFailure(t);
224     }
225
226     private void readyTransaction(final @Nonnull PingPongTransaction tx) {
227         // First mark the transaction as not locked.
228         final boolean lockedMatch = LOCKED_UPDATER.compareAndSet(this, tx, null);
229         Preconditions.checkState(lockedMatch, "Attempted to submit transaction %s while we have %s", tx, lockedTx);
230         LOG.debug("Transaction {} unlocked", tx);
231
232         /*
233          * The transaction is ready. It will then be picked up by either next allocation,
234          * or a background transaction completion callback.
235          */
236         final boolean success = READY_UPDATER.compareAndSet(this, null, tx);
237         Preconditions.checkState(success, "Transaction %s collided on ready state", tx, readyTx);
238         LOG.debug("Transaction {} readied");
239
240         /*
241          * We do not see a transaction being in-flight, so we need to take care of dispatching
242          * the transaction to the backend. We are in the ready case, we cannot short-cut
243          * the checking of readyTx, as an in-flight transaction may have completed between us
244          * setting the field above and us checking.
245          */
246         if (inflightTx == null) {
247             synchronized (this) {
248                 processIfReady();
249             }
250         }
251     }
252
253     @Override
254     public void close() {
255         final PingPongTransaction notLocked = lockedTx;
256         Preconditions.checkState(notLocked == null, "Attempted to close chain with outstanding transaction %s", notLocked);
257
258         synchronized (this) {
259             processIfReady();
260             delegate.close();
261         }
262     }
263
264     @Override
265     public DOMDataReadOnlyTransaction newReadOnlyTransaction() {
266         final PingPongTransaction tx = allocateTransaction();
267
268         return new DOMDataReadOnlyTransaction() {
269             @Override
270             public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> read(final LogicalDatastoreType store,
271                     final YangInstanceIdentifier path) {
272                 return tx.getTransaction().read(store, path);
273             }
274
275             @Override
276             public CheckedFuture<Boolean, ReadFailedException> exists(final LogicalDatastoreType store,
277                     final YangInstanceIdentifier path) {
278                 return tx.getTransaction().exists(store, path);
279             }
280
281             @Override
282             public Object getIdentifier() {
283                 return tx.getTransaction().getIdentifier();
284             }
285
286             @Override
287             public void close() {
288                 readyTransaction(tx);
289             }
290         };
291     }
292
293     @Override
294     public DOMDataReadWriteTransaction newReadWriteTransaction() {
295         final PingPongTransaction tx = allocateTransaction();
296         final DOMDataReadWriteTransaction ret = new ForwardingDOMDataReadWriteTransaction() {
297             @Override
298             protected DOMDataReadWriteTransaction delegate() {
299                 return tx.getTransaction();
300             }
301
302             @Override
303             public CheckedFuture<Void, TransactionCommitFailedException> submit() {
304                 readyTransaction(tx);
305                 return tx.getSubmitFuture();
306             }
307
308             @Override
309             public ListenableFuture<RpcResult<TransactionStatus>> commit() {
310                 readyTransaction(tx);
311                 return tx.getCommitFuture();
312             }
313
314             @Override
315             public boolean cancel() {
316                 throw new UnsupportedOperationException("Transaction cancellation is not supported");
317             }
318         };
319
320         tx.recordFrontendTransaction(ret);
321         return ret;
322     }
323
324     @Override
325     public DOMDataWriteTransaction newWriteOnlyTransaction() {
326         return newReadWriteTransaction();
327     }
328 }

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