X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsamples%2Fclustering-test-app%2Fprovider%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fclustering%2Fit%2Fprovider%2Fimpl%2FWriteTransactionsHandler.java;h=a026e6f2f3b00f8c536d099616acbff0962df356;hp=6c8750e6955f5c0592389a3ef5c4f3bc6662cff5;hb=640c1a8a610811c7d9b7d744d39fd37197fa3b63;hpb=20a414c60c02c61f07e67026ae850e3625b99216 diff --git a/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/WriteTransactionsHandler.java b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/WriteTransactionsHandler.java index 6c8750e695..a026e6f2f3 100644 --- a/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/WriteTransactionsHandler.java +++ b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/WriteTransactionsHandler.java @@ -8,7 +8,9 @@ package org.opendaylight.controller.clustering.it.provider.impl; +import com.google.common.base.Stopwatch; import com.google.common.util.concurrent.CheckedFuture; +import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.SettableFuture; @@ -59,57 +61,55 @@ public class WriteTransactionsHandler implements Runnable { private static final int MAX_ITEM = 1048576; private static final QName ID_INTS = - QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id-ints"); + QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id-ints").intern(); private static final QName ID_INT = - QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id-int"); + QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id-int").intern(); private static final QName ID = - QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id"); + QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id").intern(); private static final QName ITEM = - QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "item"); + QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "item").intern(); private static final QName NUMBER = - QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "number"); + QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "number").intern(); public static final YangInstanceIdentifier ID_INTS_YID = YangInstanceIdentifier.of(ID_INTS); - public static final YangInstanceIdentifier ID_INT_YID = ID_INTS_YID.node(ID_INT); + public static final YangInstanceIdentifier ID_INT_YID = ID_INTS_YID.node(ID_INT).toOptimized(); - private final DOMDataBroker domDataBroker; - private final Long timeToTake; - private final Long delay; - private final String id; private final WriteTransactionsInput input; private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); - private final ArrayList> futures = new ArrayList<>(); + private final List> futures = new ArrayList<>(); private final Set usedValues = new HashSet<>(); private RandomnessProvider random; private TxProvider txProvider; - private long startTime; + private final DOMDataBroker domDataBroker; + private final Long runtimeNanos; + private final Long delayNanos; + private final String id; + private SettableFuture> completionFuture; + private Stopwatch stopwatch; private long allTx = 0; private long insertTx = 0; private long deleteTx = 0; private ScheduledFuture scheduledFuture; - private YangInstanceIdentifier idListWithKey; + private YangInstanceIdentifier idListItem; public WriteTransactionsHandler(final DOMDataBroker domDataBroker, final WriteTransactionsInput input) { this.domDataBroker = domDataBroker; this.input = input; - timeToTake = input.getSeconds() * SECOND_AS_NANO; - delay = SECOND_AS_NANO / input.getTransactionsPerSecond(); + runtimeNanos = TimeUnit.SECONDS.toNanos(input.getSeconds()); + delayNanos = SECOND_AS_NANO / input.getTransactionsPerSecond(); id = input.getId(); } @Override public void run() { - final long current = System.nanoTime(); - - futures.add(execWrite()); - - maybeFinish(current); + futures.add(execWrite(futures.size())); + maybeFinish(); } public void start(final SettableFuture> settableFuture) { @@ -124,9 +124,9 @@ public class WriteTransactionsHandler implements Runnable { } if (ensureListExists(settableFuture) && fillInitialList(settableFuture)) { - startTime = System.nanoTime(); + stopwatch = Stopwatch.createStarted(); completionFuture = settableFuture; - scheduledFuture = executor.scheduleAtFixedRate(this, 0, delay, TimeUnit.NANOSECONDS); + scheduledFuture = executor.scheduleAtFixedRate(this, 0, delayNanos, TimeUnit.NANOSECONDS); } else { executor.shutdown(); } @@ -159,52 +159,46 @@ public class WriteTransactionsHandler implements Runnable { .withChild(ImmutableNodes.mapNodeBuilder(ITEM).build()) .build(); - idListWithKey = ID_INT_YID.node(entry.getIdentifier()); + idListItem = ID_INT_YID.node(entry.getIdentifier()); tx = txProvider.createTransaction(); - tx.merge(LogicalDatastoreType.CONFIGURATION, idListWithKey, entry); + tx.merge(LogicalDatastoreType.CONFIGURATION, idListItem, entry); try { tx.submit().checkedGet(125, TimeUnit.SECONDS); - } catch (final TransactionCommitFailedException | TimeoutException e) { + return true; + } catch (final Exception e) { LOG.warn("Unable to ensure IdInts list for id: {} exists.", id, e); settableFuture.set(RpcResultBuilder.failed() .withError(RpcError.ErrorType.APPLICATION, "Unexpected-exception", e).build()); return false; } - - return true; } private boolean fillInitialList(final SettableFuture> settableFuture) { LOG.debug("Filling the item list with initial values."); final CollectionNodeBuilder mapBuilder = ImmutableNodes.mapNodeBuilder(ITEM); - for (int i = 0; i < MAX_ITEM / 2; i++) { - usedValues.add(i); - mapBuilder.withChild(ImmutableNodes.mapEntry(ITEM, NUMBER, i)); - } - final YangInstanceIdentifier itemListId = idListWithKey.node(ITEM); + final YangInstanceIdentifier itemListId = idListItem.node(ITEM); final DOMDataWriteTransaction tx = txProvider.createTransaction(); tx.put(LogicalDatastoreType.CONFIGURATION, itemListId, mapBuilder.build()); try { tx.submit().checkedGet(125, TimeUnit.SECONDS); - } catch (final TransactionCommitFailedException | TimeoutException e) { + return true; + } catch (final Exception e) { LOG.warn("Unable to fill the initial item list.", e); settableFuture.set(RpcResultBuilder.failed() .withError(RpcError.ErrorType.APPLICATION, "Unexpected-exception", e).build()); return false; } - - return true; } - private CheckedFuture execWrite() { + private ListenableFuture execWrite(final int offset) { final int i = random.nextInt(MAX_ITEM + 1); final YangInstanceIdentifier entryId = - idListWithKey.node(ITEM).node(new YangInstanceIdentifier.NodeIdentifierWithPredicates(ITEM, NUMBER, i)); + idListItem.node(ITEM).node(new YangInstanceIdentifier.NodeIdentifierWithPredicates(ITEM, NUMBER, i)); final DOMDataWriteTransaction tx = txProvider.createTransaction(); allTx++; @@ -223,11 +217,27 @@ public class WriteTransactionsHandler implements Runnable { usedValues.add(i); } - return tx.submit(); + final ListenableFuture future = tx.submit(); + if (LOG.isDebugEnabled()) { + Futures.addCallback(future, new FutureCallback() { + @Override + public void onSuccess(final Void result) { + LOG.debug("Future #{} completed successfully", offset); + } + + @Override + public void onFailure(final Throwable cause) { + LOG.debug("Future #{} failed", offset, cause); + } + }); + } + + return future; } - private void maybeFinish(final long current) { - if ((current - startTime) > timeToTake) { + private void maybeFinish() { + final long elapsed = stopwatch.elapsed(TimeUnit.NANOSECONDS); + if (elapsed >= runtimeNanos) { LOG.debug("Reached max running time, waiting for futures to complete."); scheduledFuture.cancel(false); @@ -249,7 +259,32 @@ public class WriteTransactionsHandler implements Runnable { .withResult(output).build()); executor.shutdown(); - } catch (InterruptedException | ExecutionException | TimeoutException exception) { + } catch (final ExecutionException e) { + LOG.error("Write transactions failed.", e.getCause()); + + completionFuture.set(RpcResultBuilder.failed() + .withError(RpcError.ErrorType.APPLICATION, "Submit failed", e.getCause()).build()); + } catch (InterruptedException | TimeoutException e) { + LOG.error("Write transactions failed.", e); + + completionFuture.set(RpcResultBuilder.failed() + .withError(RpcError.ErrorType.APPLICATION, + "Final submit was timed out by the test provider or was interrupted", e).build()); + + for (int i = 0; i < futures.size(); i++) { + final ListenableFuture future = futures.get(i); + + try { + future.get(0, TimeUnit.NANOSECONDS); + } catch (final TimeoutException fe) { + LOG.warn("Future #{}/{} not completed yet", i, futures.size()); + } catch (final ExecutionException fe) { + LOG.warn("Future #{}/{} failed", i, futures.size(), e.getCause()); + } catch (final InterruptedException fe) { + LOG.warn("Interrupted while examining future #{}/{}", i, futures.size(), e); + } + } + } catch (Exception exception) { LOG.error("Write transactions failed.", exception); completionFuture.set(RpcResultBuilder.failed() .withError(RpcError.ErrorType.APPLICATION, "Unexpected-exception", exception).build());