Fix shard commit deadlock
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / TransactionProxy.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.cluster.datastore;
9
10 import akka.actor.ActorSelection;
11 import com.google.common.annotations.VisibleForTesting;
12 import com.google.common.base.Function;
13 import com.google.common.base.Optional;
14 import com.google.common.base.Preconditions;
15 import com.google.common.base.Supplier;
16 import com.google.common.collect.Iterables;
17 import com.google.common.util.concurrent.CheckedFuture;
18 import com.google.common.util.concurrent.Futures;
19 import com.google.common.util.concurrent.ListenableFuture;
20 import com.google.common.util.concurrent.MoreExecutors;
21 import com.google.common.util.concurrent.SettableFuture;
22 import java.util.ArrayList;
23 import java.util.Collection;
24 import java.util.List;
25 import java.util.Map;
26 import java.util.Map.Entry;
27 import java.util.Set;
28 import java.util.TreeMap;
29 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
30 import org.opendaylight.controller.cluster.datastore.messages.AbstractRead;
31 import org.opendaylight.controller.cluster.datastore.messages.DataExists;
32 import org.opendaylight.controller.cluster.datastore.messages.ReadData;
33 import org.opendaylight.controller.cluster.datastore.modification.AbstractModification;
34 import org.opendaylight.controller.cluster.datastore.modification.DeleteModification;
35 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
36 import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
37 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
38 import org.opendaylight.controller.cluster.datastore.utils.NormalizedNodeAggregator;
39 import org.opendaylight.mdsal.common.api.MappingCheckedFuture;
40 import org.opendaylight.mdsal.common.api.ReadFailedException;
41 import org.opendaylight.mdsal.dom.spi.store.AbstractDOMStoreTransaction;
42 import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
43 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
44 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
45 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException;
46 import org.slf4j.Logger;
47 import org.slf4j.LoggerFactory;
48 import scala.concurrent.Future;
49 import scala.concurrent.Promise;
50
51 /**
52  * A transaction potentially spanning multiple backend shards.
53  */
54 public class TransactionProxy extends AbstractDOMStoreTransaction<TransactionIdentifier>
55         implements DOMStoreReadWriteTransaction {
56     private enum TransactionState {
57         OPEN,
58         READY,
59         CLOSED,
60     }
61
62     private static final Logger LOG = LoggerFactory.getLogger(TransactionProxy.class);
63
64     // Global lock used for transactions spanning multiple shards - synchronizes sending of the ready messages
65     // for atomicity to avoid potential deadlock with concurrent transactions spanning the same shards as outlined
66     // in the following scenario:
67     //
68     //  - Tx1 sends ready message to shard A
69     //  - Tx2 sends ready message to shard A
70     //  - Tx2 sends ready message to shard B
71     //  - Tx1 sends ready message to shard B
72     //
73     // This scenario results in deadlock: after Tx1 canCommits to shard A, it can't proceed with shard B until Tx2
74     // completes as Tx2 was readied first on shard B. However Tx2 cannot make progress because it's waiting to canCommit
75     // on shard A which is blocked by Tx1.
76     //
77     // The global lock avoids this as it forces the ready messages to be sent in a predictable order:
78     //
79     //  - Tx1 sends ready message to shard A
80     //  - Tx1 sends ready message to shard B
81     //  - Tx2 sends ready message to shard A
82     //  - Tx2 sends ready message to shard B
83     //
84     private static final Object GLOBAL_TX_READY_LOCK = new Object();
85
86     private final Map<String, TransactionContextWrapper> txContextWrappers = new TreeMap<>();
87     private final AbstractTransactionContextFactory<?> txContextFactory;
88     private final TransactionType type;
89     private TransactionState state = TransactionState.OPEN;
90
91     @VisibleForTesting
92     public TransactionProxy(final AbstractTransactionContextFactory<?> txContextFactory, final TransactionType type) {
93         super(txContextFactory.nextIdentifier(), txContextFactory.getActorContext().getDatastoreContext()
94                 .isTransactionDebugContextEnabled());
95         this.txContextFactory = txContextFactory;
96         this.type = Preconditions.checkNotNull(type);
97
98         LOG.debug("New {} Tx - {}", type, getIdentifier());
99     }
100
101     @Override
102     public CheckedFuture<Boolean, ReadFailedException> exists(final YangInstanceIdentifier path) {
103         return executeRead(shardNameFromIdentifier(path), new DataExists(path, DataStoreVersions.CURRENT_VERSION));
104     }
105
106     private <T> CheckedFuture<T, ReadFailedException> executeRead(final String shardName,
107             final AbstractRead<T> readCmd) {
108         Preconditions.checkState(type != TransactionType.WRITE_ONLY,
109                 "Reads from write-only transactions are not allowed");
110
111         LOG.trace("Tx {} {} {}", getIdentifier(), readCmd.getClass().getSimpleName(), readCmd.getPath());
112
113         final SettableFuture<T> proxyFuture = SettableFuture.create();
114         TransactionContextWrapper contextWrapper = getContextWrapper(shardName);
115         contextWrapper.maybeExecuteTransactionOperation(new TransactionOperation() {
116             @Override
117             public void invoke(final TransactionContext transactionContext, final Boolean havePermit) {
118                 transactionContext.executeRead(readCmd, proxyFuture, havePermit);
119             }
120         });
121
122         return MappingCheckedFuture.create(proxyFuture, ReadFailedException.MAPPER);
123     }
124
125     @Override
126     public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> read(final YangInstanceIdentifier path) {
127         Preconditions.checkState(type != TransactionType.WRITE_ONLY,
128                 "Reads from write-only transactions are not allowed");
129         Preconditions.checkNotNull(path, "path should not be null");
130
131         LOG.trace("Tx {} read {}", getIdentifier(), path);
132         return path.isEmpty() ? readAllData() :  singleShardRead(shardNameFromIdentifier(path), path);
133     }
134
135     private CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> singleShardRead(
136             final String shardName, final YangInstanceIdentifier path) {
137         return executeRead(shardName, new ReadData(path, DataStoreVersions.CURRENT_VERSION));
138     }
139
140     private CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> readAllData() {
141         final Set<String> allShardNames = txContextFactory.getActorContext().getConfiguration().getAllShardNames();
142         final Collection<CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException>> futures =
143                 new ArrayList<>(allShardNames.size());
144
145         for (String shardName : allShardNames) {
146             futures.add(singleShardRead(shardName, YangInstanceIdentifier.EMPTY));
147         }
148
149         final ListenableFuture<List<Optional<NormalizedNode<?, ?>>>> listFuture = Futures.allAsList(futures);
150         final ListenableFuture<Optional<NormalizedNode<?, ?>>> aggregateFuture;
151
152         aggregateFuture = Futures.transform(listFuture,
153             (Function<List<Optional<NormalizedNode<?, ?>>>, Optional<NormalizedNode<?, ?>>>) input -> {
154                 try {
155                     return NormalizedNodeAggregator.aggregate(YangInstanceIdentifier.EMPTY, input,
156                             txContextFactory.getActorContext().getSchemaContext(),
157                             txContextFactory.getActorContext().getDatastoreContext().getLogicalStoreType());
158                 } catch (DataValidationFailedException e) {
159                     throw new IllegalArgumentException("Failed to aggregate", e);
160                 }
161             }, MoreExecutors.directExecutor());
162
163         return MappingCheckedFuture.create(aggregateFuture, ReadFailedException.MAPPER);
164     }
165
166     @Override
167     public void delete(final YangInstanceIdentifier path) {
168         executeModification(new DeleteModification(path));
169     }
170
171     @Override
172     public void merge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
173         executeModification(new MergeModification(path, data));
174     }
175
176     @Override
177     public void write(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
178         executeModification(new WriteModification(path, data));
179     }
180
181     private void executeModification(final AbstractModification modification) {
182         checkModificationState();
183
184         LOG.trace("Tx {} executeModification {} {}", getIdentifier(), modification.getClass().getSimpleName(),
185                 modification.getPath());
186
187         TransactionContextWrapper contextWrapper = getContextWrapper(modification.getPath());
188         contextWrapper.maybeExecuteTransactionOperation(new TransactionOperation() {
189             @Override
190             protected void invoke(final TransactionContext transactionContext, final Boolean havePermit) {
191                 transactionContext.executeModification(modification, havePermit);
192             }
193         });
194     }
195
196     private void checkModificationState() {
197         Preconditions.checkState(type != TransactionType.READ_ONLY,
198                 "Modification operation on read-only transaction is not allowed");
199         Preconditions.checkState(state == TransactionState.OPEN,
200                 "Transaction is sealed - further modifications are not allowed");
201     }
202
203     private boolean seal(final TransactionState newState) {
204         if (state == TransactionState.OPEN) {
205             state = newState;
206             return true;
207         } else {
208             return false;
209         }
210     }
211
212     @Override
213     public final void close() {
214         if (!seal(TransactionState.CLOSED)) {
215             Preconditions.checkState(state == TransactionState.CLOSED, "Transaction %s is ready, it cannot be closed",
216                 getIdentifier());
217             // Idempotent no-op as per AutoCloseable recommendation
218             return;
219         }
220
221         for (TransactionContextWrapper contextWrapper : txContextWrappers.values()) {
222             contextWrapper.maybeExecuteTransactionOperation(new TransactionOperation() {
223                 @Override
224                 public void invoke(final TransactionContext transactionContext, final Boolean havePermit) {
225                     transactionContext.closeTransaction();
226                 }
227             });
228         }
229
230
231         txContextWrappers.clear();
232     }
233
234     @Override
235     public final AbstractThreePhaseCommitCohort<?> ready() {
236         Preconditions.checkState(type != TransactionType.READ_ONLY, "Read-only transactions cannot be readied");
237
238         final boolean success = seal(TransactionState.READY);
239         Preconditions.checkState(success, "Transaction %s is %s, it cannot be readied", getIdentifier(), state);
240
241         LOG.debug("Tx {} Readying {} components for commit", getIdentifier(), txContextWrappers.size());
242
243         final AbstractThreePhaseCommitCohort<?> ret;
244         switch (txContextWrappers.size()) {
245             case 0:
246                 ret = NoOpDOMStoreThreePhaseCommitCohort.INSTANCE;
247                 break;
248             case 1:
249                 final Entry<String, TransactionContextWrapper> e = Iterables.getOnlyElement(
250                         txContextWrappers.entrySet());
251                 ret = createSingleCommitCohort(e.getKey(), e.getValue());
252                 break;
253             default:
254                 ret = createMultiCommitCohort(txContextWrappers.entrySet());
255         }
256
257         txContextFactory.onTransactionReady(getIdentifier(), ret.getCohortFutures());
258
259         final Throwable debugContext = getDebugContext();
260         return debugContext == null ? ret : new DebugThreePhaseCommitCohort(getIdentifier(), ret, debugContext);
261     }
262
263     @SuppressWarnings({ "rawtypes", "unchecked" })
264     private AbstractThreePhaseCommitCohort<?> createSingleCommitCohort(final String shardName,
265             final TransactionContextWrapper contextWrapper) {
266
267         LOG.debug("Tx {} Readying transaction for shard {}", getIdentifier(), shardName);
268
269         final OperationCallback.Reference operationCallbackRef =
270                 new OperationCallback.Reference(OperationCallback.NO_OP_CALLBACK);
271
272         final TransactionContext transactionContext = contextWrapper.getTransactionContext();
273         final Future future;
274         if (transactionContext == null) {
275             final Promise promise = akka.dispatch.Futures.promise();
276             contextWrapper.maybeExecuteTransactionOperation(new TransactionOperation() {
277                 @Override
278                 public void invoke(final TransactionContext newTransactionContext, final Boolean havePermit) {
279                     promise.completeWith(getDirectCommitFuture(newTransactionContext, operationCallbackRef,
280                         havePermit));
281                 }
282             });
283             future = promise.future();
284         } else {
285             // avoid the creation of a promise and a TransactionOperation
286             future = getDirectCommitFuture(transactionContext, operationCallbackRef, null);
287         }
288
289         return new SingleCommitCohortProxy(txContextFactory.getActorContext(), future, getIdentifier(),
290             operationCallbackRef);
291     }
292
293     private Future<?> getDirectCommitFuture(final TransactionContext transactionContext,
294             final OperationCallback.Reference operationCallbackRef, final Boolean havePermit) {
295         TransactionRateLimitingCallback rateLimitingCallback = new TransactionRateLimitingCallback(
296                 txContextFactory.getActorContext());
297         operationCallbackRef.set(rateLimitingCallback);
298         rateLimitingCallback.run();
299         return transactionContext.directCommit(havePermit);
300     }
301
302     private AbstractThreePhaseCommitCohort<ActorSelection> createMultiCommitCohort(
303             final Set<Entry<String, TransactionContextWrapper>> txContextWrapperEntries) {
304
305         final List<ThreePhaseCommitCohortProxy.CohortInfo> cohorts = new ArrayList<>(txContextWrapperEntries.size());
306
307         synchronized (GLOBAL_TX_READY_LOCK) {
308             for (Entry<String, TransactionContextWrapper> e : txContextWrapperEntries) {
309                 LOG.debug("Tx {} Readying transaction for shard {}", getIdentifier(), e.getKey());
310
311                 final TransactionContextWrapper wrapper = e.getValue();
312
313                 // The remote tx version is obtained the via TransactionContext which may not be available yet so
314                 // we pass a Supplier to dynamically obtain it. Once the ready Future is resolved the
315                 // TransactionContext is available.
316                 Supplier<Short> txVersionSupplier = () -> wrapper.getTransactionContext().getTransactionVersion();
317
318                 cohorts.add(new ThreePhaseCommitCohortProxy.CohortInfo(wrapper.readyTransaction(), txVersionSupplier));
319             }
320         }
321
322         return new ThreePhaseCommitCohortProxy(txContextFactory.getActorContext(), cohorts, getIdentifier());
323     }
324
325     private String shardNameFromIdentifier(final YangInstanceIdentifier path) {
326         return txContextFactory.getActorContext().getShardStrategyFactory().getStrategy(path).findShard(path);
327     }
328
329     private TransactionContextWrapper getContextWrapper(final YangInstanceIdentifier path) {
330         return getContextWrapper(shardNameFromIdentifier(path));
331     }
332
333     private TransactionContextWrapper getContextWrapper(final String shardName) {
334         final TransactionContextWrapper existing = txContextWrappers.get(shardName);
335         if (existing != null) {
336             return existing;
337         }
338
339         final TransactionContextWrapper fresh = txContextFactory.newTransactionContextWrapper(this, shardName);
340         txContextWrappers.put(shardName, fresh);
341         return fresh;
342     }
343
344     TransactionType getType() {
345         return type;
346     }
347
348     boolean isReady() {
349         return state != TransactionState.OPEN;
350     }
351
352     ActorContext getActorContext() {
353         return txContextFactory.getActorContext();
354     }
355 }

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