Bug 8494: Separate writing and completion threads
[controller.git] / opendaylight / md-sal / samples / clustering-test-app / provider / src / main / java / org / opendaylight / controller / clustering / it / provider / impl / WriteTransactionsHandler.java
1 /*
2  * Copyright (c) 2017 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.clustering.it.provider.impl;
10
11 import com.google.common.base.Preconditions;
12 import com.google.common.util.concurrent.Futures;
13 import com.google.common.util.concurrent.ListenableFuture;
14 import com.google.common.util.concurrent.SettableFuture;
15 import java.util.HashSet;
16 import java.util.LinkedHashSet;
17 import java.util.Set;
18 import java.util.SplittableRandom;
19 import java.util.concurrent.TimeUnit;
20 import java.util.concurrent.TimeoutException;
21 import org.opendaylight.controller.md.sal.common.api.data.AsyncTransaction;
22 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
23 import org.opendaylight.controller.md.sal.common.api.data.OptimisticLockFailedException;
24 import org.opendaylight.controller.md.sal.common.api.data.TransactionChain;
25 import org.opendaylight.controller.md.sal.common.api.data.TransactionChainListener;
26 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
27 import org.opendaylight.controller.md.sal.dom.api.DOMDataBroker;
28 import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
29 import org.opendaylight.controller.md.sal.dom.api.DOMTransactionChain;
30 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.WriteTransactionsInput;
31 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.WriteTransactionsOutput;
32 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.WriteTransactionsOutputBuilder;
33 import org.opendaylight.yangtools.yang.common.RpcError;
34 import org.opendaylight.yangtools.yang.common.RpcResult;
35 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
36 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
37 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
38 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
39 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
40 import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
41 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
42 import org.opendaylight.yangtools.yang.data.impl.schema.builder.api.CollectionNodeBuilder;
43 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
44 import org.slf4j.Logger;
45 import org.slf4j.LoggerFactory;
46
47 public abstract class WriteTransactionsHandler extends AbstractTransactionHandler {
48     private static final class Chained extends WriteTransactionsHandler implements TransactionChainListener {
49         private final SplittableRandom random = new SplittableRandom();
50         private final DOMTransactionChain transactionChain;
51
52         Chained(final DOMDataBroker dataBroker, final YangInstanceIdentifier idListItem,
53             final WriteTransactionsInput input) {
54             super(idListItem, input);
55             transactionChain = dataBroker.createTransactionChain(this);
56         }
57
58         @Override
59         DOMDataWriteTransaction createTransaction() {
60             return transactionChain.newWriteOnlyTransaction();
61         }
62
63         @Override
64         int nextInt(final int bound) {
65             return random.nextInt(bound);
66         }
67
68         @Override
69         public void onTransactionChainFailed(final TransactionChain<?, ?> chain,
70                 final AsyncTransaction<?, ?> transaction, final Throwable cause) {
71             // This is expected to happen frequently in isolation testing.
72             LOG.debug("Transaction chain failed.", cause);
73             // Do not return RPC here, rely on transaction failure to call runFailed.
74         }
75
76         @Override
77         public void onTransactionChainSuccessful(final TransactionChain<?, ?> chain) {
78             LOG.debug("Transaction chain closed successfully.");
79         }
80     }
81
82     private static final class Simple extends WriteTransactionsHandler {
83         private final LinkedHashSet<Integer> previousNumbers = new LinkedHashSet<>();
84         private final SplittableRandom random = new SplittableRandom();
85         private final DOMDataBroker dataBroker;
86
87         Simple(final DOMDataBroker dataBroker, final YangInstanceIdentifier idListItem,
88             final WriteTransactionsInput input) {
89             super(idListItem, input);
90             this.dataBroker = Preconditions.checkNotNull(dataBroker);
91         }
92
93         @Override
94         DOMDataWriteTransaction createTransaction() {
95             return dataBroker.newWriteOnlyTransaction();
96         }
97
98         @Override
99         int nextInt(final int bound) {
100             int nextInt;
101             do {
102                 nextInt = random.nextInt(bound);
103             } while (previousNumbers.contains(nextInt));
104
105             if (previousNumbers.size() > 100000) {
106                 previousNumbers.iterator().remove();
107             }
108             previousNumbers.add(nextInt);
109
110             return nextInt;
111         }
112     }
113
114     private static final Logger LOG = LoggerFactory.getLogger(WriteTransactionsHandler.class);
115
116     final SettableFuture<RpcResult<WriteTransactionsOutput>> completionFuture = SettableFuture.create();
117     private final Set<Integer> usedValues = new HashSet<>();
118     private final YangInstanceIdentifier idListItem;
119
120     private long insertTx = 0;
121     private long deleteTx = 0;
122
123     WriteTransactionsHandler(final YangInstanceIdentifier idListItem, final WriteTransactionsInput input) {
124         super(input);
125         this.idListItem = Preconditions.checkNotNull(idListItem);
126     }
127
128     public static ListenableFuture<RpcResult<WriteTransactionsOutput>> start(final DOMDataBroker domDataBroker,
129             final WriteTransactionsInput input) {
130         LOG.debug("Starting write-transactions.");
131
132         final String id = input.getId();
133         final MapEntryNode entry = ImmutableNodes.mapEntryBuilder(ID_INT, ID, id)
134                 .withChild(ImmutableNodes.mapNodeBuilder(ITEM).build())
135                 .build();
136         final YangInstanceIdentifier idListItem = ID_INT_YID.node(entry.getIdentifier());
137
138         final ContainerNode containerNode = ImmutableContainerNodeBuilder.create()
139                 .withNodeIdentifier(new NodeIdentifier(ID_INTS))
140                 .withChild(ImmutableNodes.mapNodeBuilder(ID_INT).build())
141                 .build();
142
143         DOMDataWriteTransaction tx = domDataBroker.newWriteOnlyTransaction();
144         // write only the top list
145         tx.merge(LogicalDatastoreType.CONFIGURATION, ID_INTS_YID, containerNode);
146         try {
147             tx.submit().checkedGet(INIT_TX_TIMEOUT_SECONDS, TimeUnit.SECONDS);
148         } catch (final OptimisticLockFailedException e) {
149             // when multiple write-transactions are executed concurrently we need to ignore this.
150             // If we get optimistic lock here it means id-ints already exists and we can continue.
151             LOG.debug("Got an optimistic lock when writing initial top level list element.", e);
152         } catch (final TransactionCommitFailedException | TimeoutException e) {
153             LOG.warn("Unable to ensure IdInts list for id: {} exists.", id, e);
154             return Futures.immediateFuture(RpcResultBuilder.<WriteTransactionsOutput>failed()
155                     .withError(RpcError.ErrorType.APPLICATION, "Unexpected-exception", e).build());
156         }
157
158         tx = domDataBroker.newWriteOnlyTransaction();
159         tx.merge(LogicalDatastoreType.CONFIGURATION, idListItem, entry);
160
161         try {
162             tx.submit().checkedGet(INIT_TX_TIMEOUT_SECONDS, TimeUnit.SECONDS);
163         } catch (final Exception e) {
164             LOG.warn("Unable to ensure IdInts list for id: {} exists.", id, e);
165             return Futures.immediateFuture(RpcResultBuilder.<WriteTransactionsOutput>failed()
166                     .withError(RpcError.ErrorType.APPLICATION, "Unexpected-exception", e).build());
167         }
168
169         LOG.debug("Filling the item list with initial values.");
170
171         final CollectionNodeBuilder<MapEntryNode, MapNode> mapBuilder = ImmutableNodes.mapNodeBuilder(ITEM);
172
173         final YangInstanceIdentifier itemListId = idListItem.node(ITEM);
174         tx = domDataBroker.newWriteOnlyTransaction();
175         tx.put(LogicalDatastoreType.CONFIGURATION, itemListId, mapBuilder.build());
176
177         try {
178             tx.submit().checkedGet(INIT_TX_TIMEOUT_SECONDS, TimeUnit.SECONDS);
179         } catch (final Exception e) {
180             LOG.warn("Unable to fill the initial item list.", e);
181             return Futures.immediateFuture(RpcResultBuilder.<WriteTransactionsOutput>failed()
182                     .withError(RpcError.ErrorType.APPLICATION, "Unexpected-exception", e).build());
183         }
184
185         final WriteTransactionsHandler handler;
186         if (input.isChainedTransactions()) {
187             handler = new Chained(domDataBroker, idListItem, input);
188         } else {
189             handler = new Simple(domDataBroker, idListItem, input);
190         }
191
192         handler.doStart();
193         return handler.completionFuture;
194     }
195
196     @Override
197     ListenableFuture<Void> execWrite(final long txId) {
198         final int i = nextInt(MAX_ITEM + 1);
199
200         final YangInstanceIdentifier entryId =
201                 idListItem.node(ITEM).node(new YangInstanceIdentifier.NodeIdentifierWithPredicates(ITEM, NUMBER, i));
202
203         final DOMDataWriteTransaction tx = createTransaction();
204
205         if (usedValues.contains(i)) {
206             LOG.debug("Deleting item: {}", i);
207             deleteTx++;
208             tx.delete(LogicalDatastoreType.CONFIGURATION, entryId);
209             usedValues.remove(i);
210
211         } else {
212             LOG.debug("Inserting item: {}", i);
213             insertTx++;
214             final MapEntryNode entry = ImmutableNodes.mapEntry(ITEM, NUMBER, i);
215             tx.put(LogicalDatastoreType.CONFIGURATION, entryId, entry);
216             usedValues.add(i);
217         }
218
219         return tx.submit();
220     }
221
222     @Override
223     void runFailed(final Throwable cause) {
224         completionFuture.set(RpcResultBuilder.<WriteTransactionsOutput>failed()
225             .withError(RpcError.ErrorType.APPLICATION, "Submit failed", cause).build());
226     }
227
228     @Override
229     void runSuccessful(final long allTx) {
230         final WriteTransactionsOutput output = new WriteTransactionsOutputBuilder()
231                 .setAllTx(allTx)
232                 .setInsertTx(insertTx)
233                 .setDeleteTx(deleteTx)
234                 .build();
235
236         completionFuture.set(RpcResultBuilder.<WriteTransactionsOutput>success()
237                 .withResult(output).build());
238     }
239
240     @Override
241     void runTimedOut(final Exception cause) {
242         completionFuture.set(RpcResultBuilder.<WriteTransactionsOutput>failed()
243             .withError(RpcError.ErrorType.APPLICATION,
244                     "Final submit was timed out by the test provider or was interrupted", cause).build());
245     }
246
247     abstract DOMDataWriteTransaction createTransaction();
248
249     abstract int nextInt(int bound);
250 }

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