2 * Copyright (c) 2017 Cisco Systems, Inc. and others. All rights reserved.
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
9 package org.opendaylight.controller.clustering.it.provider.impl;
11 import com.google.common.util.concurrent.CheckedFuture;
12 import com.google.common.util.concurrent.FutureCallback;
13 import com.google.common.util.concurrent.Futures;
14 import com.google.common.util.concurrent.ListenableFuture;
15 import com.google.common.util.concurrent.SettableFuture;
16 import java.util.ArrayList;
17 import java.util.Collections;
18 import java.util.HashSet;
19 import java.util.List;
21 import java.util.SplittableRandom;
22 import java.util.concurrent.Executors;
23 import java.util.concurrent.ScheduledExecutorService;
24 import java.util.concurrent.ScheduledFuture;
25 import java.util.concurrent.TimeUnit;
26 import javax.annotation.Nullable;
27 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
28 import org.opendaylight.mdsal.common.api.TransactionCommitFailedException;
29 import org.opendaylight.mdsal.dom.api.DOMDataTreeCursorAwareTransaction;
30 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
31 import org.opendaylight.mdsal.dom.api.DOMDataTreeProducer;
32 import org.opendaylight.mdsal.dom.api.DOMDataTreeProducerException;
33 import org.opendaylight.mdsal.dom.api.DOMDataTreeService;
34 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteCursor;
35 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.ProduceTransactionsInput;
36 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.ProduceTransactionsOutput;
37 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.ProduceTransactionsOutputBuilder;
38 import org.opendaylight.yangtools.yang.common.QName;
39 import org.opendaylight.yangtools.yang.common.RpcError;
40 import org.opendaylight.yangtools.yang.common.RpcResult;
41 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
42 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
43 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
44 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifierWithPredicates;
45 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
46 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
47 import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
48 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
49 import org.opendaylight.yangtools.yang.data.impl.schema.builder.api.CollectionNodeBuilder;
50 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
51 import org.slf4j.Logger;
52 import org.slf4j.LoggerFactory;
54 public class ProduceTransactionsHandler implements Runnable {
56 private static final Logger LOG = LoggerFactory.getLogger(ProduceTransactionsHandler.class);
57 private static final int SECOND_AS_NANO = 1000000000;
58 //2^20 as in the model
59 private static final int MAX_ITEM = 1048576;
61 static final QName ID_INTS =
62 QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id-ints");
63 static final QName ID_INT =
64 QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id-int");
65 static final QName ID =
66 QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id");
67 static final QName ITEM =
68 QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "item");
69 private static final QName NUMBER =
70 QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "number");
72 public static final YangInstanceIdentifier ID_INTS_YID = YangInstanceIdentifier.of(ID_INTS);
73 public static final YangInstanceIdentifier ID_INT_YID = ID_INTS_YID.node(ID_INT);
75 private final DOMDataTreeService domDataTreeService;
77 private final long timeToTake;
78 private final long delay;
79 private final String id;
81 private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
82 private final ArrayList<CheckedFuture<Void, TransactionCommitFailedException>> futures = new ArrayList<>();
83 private final Set<Integer> usedValues = new HashSet<>();
84 private final SplittableRandom random = new SplittableRandom();
86 private long startTime;
87 private SettableFuture<RpcResult<ProduceTransactionsOutput>> completionFuture;
89 private long allTx = 0;
90 private long insertTx = 0;
91 private long deleteTx = 0;
92 private ScheduledFuture<?> scheduledFuture;
93 private DOMDataTreeProducer itemProducer;
94 private YangInstanceIdentifier idListWithKey;
96 public ProduceTransactionsHandler(final DOMDataTreeService domDataTreeService,
97 final ProduceTransactionsInput input) {
99 this.domDataTreeService = domDataTreeService;
101 timeToTake = input.getSeconds() * SECOND_AS_NANO;
102 delay = SECOND_AS_NANO / input.getTransactionsPerSecond();
108 final long current = System.nanoTime();
110 futures.add(execWrite());
112 maybeFinish(current);
115 public void start(final SettableFuture<RpcResult<ProduceTransactionsOutput>> settableFuture) {
116 completionFuture = settableFuture;
118 if (fillInitialList(completionFuture)) {
119 startTime = System.nanoTime();
120 scheduledFuture = executor.scheduleAtFixedRate(this, 0, delay, TimeUnit.NANOSECONDS);
126 private boolean fillInitialList(final SettableFuture<RpcResult<ProduceTransactionsOutput>> settableFuture) {
127 LOG.debug("Filling the item list with initial values.");
129 final CollectionNodeBuilder<MapEntryNode, MapNode> mapBuilder = ImmutableNodes.mapNodeBuilder(ITEM);
130 for (int i = 0; i < MAX_ITEM / 2; i++) {
132 mapBuilder.withChild(ImmutableNodes.mapEntry(ITEM, NUMBER, i));
135 idListWithKey = ID_INT_YID.node(new NodeIdentifierWithPredicates(ID_INT, ID, id));
137 itemProducer = domDataTreeService.createProducer(
138 Collections.singleton(new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, idListWithKey)));
140 final DOMDataTreeCursorAwareTransaction tx = itemProducer.createTransaction(false);
141 final DOMDataTreeWriteCursor cursor =
142 tx.createCursor(new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, idListWithKey));
144 final MapNode list = mapBuilder.build();
145 cursor.write(list.getIdentifier(), list);
149 tx.submit().checkedGet();
150 } catch (final TransactionCommitFailedException e) {
151 LOG.warn("Unable to fill the initial item list.", e);
152 settableFuture.set(RpcResultBuilder.<ProduceTransactionsOutput>failed()
153 .withError(RpcError.ErrorType.APPLICATION, "Unexpected-exception", e).build());
160 private CheckedFuture<Void, TransactionCommitFailedException> execWrite() {
161 final int i = random.nextInt(MAX_ITEM + 1);
163 final YangInstanceIdentifier entryId =
164 idListWithKey.node(ITEM).node(new NodeIdentifierWithPredicates(ITEM, NUMBER, i));
166 final DOMDataTreeCursorAwareTransaction tx = itemProducer.createTransaction(false);
167 final DOMDataTreeWriteCursor cursor = tx.createCursor(
168 new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, idListWithKey.node(ITEM)));
171 if (usedValues.contains(i)) {
172 LOG.debug("Deleting item: {}", i);
174 cursor.delete(entryId.getLastPathArgument());
175 usedValues.remove(i);
178 LOG.debug("Inserting item: {}", i);
180 final MapEntryNode entry = ImmutableNodes.mapEntry(ITEM, NUMBER, i);
181 cursor.write(entryId.getLastPathArgument(), entry);
190 private void maybeFinish(final long current) {
191 if ((current - startTime) > timeToTake) {
192 LOG.debug("Reached max running time, waiting for futures to complete.");
193 scheduledFuture.cancel(false);
195 final ListenableFuture<List<Void>> allFutures = Futures.allAsList(futures);
197 Futures.addCallback(allFutures, new FutureCallback<List<Void>>() {
199 public void onSuccess(@Nullable final List<Void> result) {
200 LOG.debug("All futures completed successfully.");
202 final ProduceTransactionsOutput output = new ProduceTransactionsOutputBuilder()
204 .setInsertTx(insertTx)
205 .setDeleteTx(deleteTx)
209 itemProducer.close();
210 } catch (final DOMDataTreeProducerException e) {
211 LOG.warn("Failure while closing item producer.", e);
214 completionFuture.set(RpcResultBuilder.<ProduceTransactionsOutput>success()
215 .withResult(output).build());
221 public void onFailure(final Throwable t) {
222 LOG.error("Write transactions failed.", t);
223 completionFuture.set(RpcResultBuilder.<ProduceTransactionsOutput>failed()
224 .withError(RpcError.ErrorType.APPLICATION, "Unexpected-exception", t).build());