--- /dev/null
+/*
+ * Copyright (c) 2017 Pantheon Technologies, s.r.o. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.clustering.it.provider.impl;
+
+import com.google.common.base.Stopwatch;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
+import java.util.Queue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import javax.annotation.concurrent.GuardedBy;
+import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.TransactionsParams;
+import org.opendaylight.yangtools.yang.common.QName;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+abstract class AbstractTransactionHandler {
+ private abstract static class Phase {
+ abstract void txSuccess(ListenableFuture<Void> execFuture, long txId);
+
+ abstract void txFailure(ListenableFuture<Void> execFuture, long txId, Throwable cause);
+ }
+
+ private static final class Running extends Phase {
+ private final Queue<ListenableFuture<Void>> futures = new ArrayDeque<>();
+ private Throwable failure;
+
+ void addFuture(final ListenableFuture<Void> execFuture) {
+ futures.add(execFuture);
+ }
+
+ @Override
+ void txSuccess(final ListenableFuture<Void> execFuture, final long txId) {
+ futures.remove(execFuture);
+ }
+
+ @Override
+ void txFailure(final ListenableFuture<Void> execFuture, final long txId, final Throwable cause) {
+ futures.remove(execFuture);
+ if (failure != null) {
+ failure = cause;
+ }
+ }
+
+ Optional<Throwable> getFailure() {
+ return Optional.ofNullable(failure);
+ }
+ }
+
+ private final class Collecting extends Phase {
+ private final List<ListenableFuture<Void>> futures;
+ private boolean done;
+
+ Collecting(final Collection<ListenableFuture<Void>> futures) {
+ this.futures = new ArrayList<>(futures);
+ }
+
+ @Override
+ void txSuccess(final ListenableFuture<Void> execFuture, final long txId) {
+ futures.remove(execFuture);
+ if (futures.isEmpty() && !done) {
+ LOG.debug("All futures completed successfully.");
+ runSuccessful(txCounter);
+ }
+ }
+
+ @Override
+ void txFailure(final ListenableFuture<Void> execFuture, final long txId, final Throwable cause) {
+ futures.remove(execFuture);
+ done = true;
+ runFailed(cause);
+ }
+ }
+
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractTransactionHandler.class);
+
+ static final int SECOND_AS_NANO = 1000000000;
+ //2^20 as in the model
+ static final int MAX_ITEM = 1048576;
+
+ static final QName ID_INTS =
+ QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id-ints").intern();
+ static final QName ID =
+ QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id").intern();
+ static final QName ITEM =
+ QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "item").intern();
+ static final QName NUMBER =
+ QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "number").intern();
+
+ public static final QName ID_INT =
+ QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id-int").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).toOptimized();
+
+ static final long INIT_TX_TIMEOUT_SECONDS = 125;
+
+ private static final long DEAD_TIMEOUT_SECONDS = TimeUnit.MINUTES.toSeconds(5);
+
+ private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
+ private final Stopwatch stopwatch = Stopwatch.createStarted();
+ private final long runtimeNanos;
+ private final long delayNanos;
+
+ private ScheduledFuture<?> scheduledFuture;
+ private long txCounter;
+ @GuardedBy("this")
+ private Phase phase;
+
+ AbstractTransactionHandler(final TransactionsParams params) {
+ runtimeNanos = TimeUnit.SECONDS.toNanos(params.getSeconds());
+ delayNanos = SECOND_AS_NANO / params.getTransactionsPerSecond();
+ }
+
+ final synchronized void doStart() {
+ phase = new Running();
+ scheduledFuture = executor.scheduleAtFixedRate(this::execute, 0, delayNanos, TimeUnit.NANOSECONDS);
+ }
+
+ private void execute() {
+ final long elapsed = stopwatch.elapsed(TimeUnit.NANOSECONDS);
+ if (elapsed < runtimeNanos) {
+ // Not completed yet: create a transaction and hook it up
+ final long txId = txCounter++;
+ final ListenableFuture<Void> execFuture = execWrite(txId);
+
+ // Ordering is important: we need to add the future before hooking the callback
+ synchronized (this) {
+ ((Running) phase).addFuture(execFuture);
+ }
+ Futures.addCallback(execFuture, new FutureCallback<Void>() {
+ @Override
+ public void onSuccess(final Void result) {
+ txSuccess(execFuture, txId);
+ }
+
+ @Override
+ public void onFailure(final Throwable cause) {
+ txFailure(execFuture, txId, cause);
+ }
+ });
+ } else {
+ startCollection();
+ }
+ }
+
+ private synchronized void startCollection() {
+ scheduledFuture.cancel(false);
+
+ final Running running = (Running) phase;
+ final Optional<Throwable> failure = running.getFailure();
+ if (failure.isPresent()) {
+ executor.shutdown();
+ runFailed(failure.get());
+ return;
+ }
+
+ LOG.debug("Reached maximum run time with {} outstanding futures", running.futures.size());
+ if (running.futures.isEmpty()) {
+ executor.shutdown();
+ runSuccessful(txCounter);
+ return;
+ }
+
+ phase = new Collecting(running.futures);
+ executor.schedule(this::checkCollection, DEAD_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ executor.shutdown();
+ }
+
+ final synchronized void txSuccess(final ListenableFuture<Void> execFuture, final long txId) {
+ LOG.debug("Future #{} completed successfully", txId);
+ phase.txSuccess(execFuture, txId);
+ }
+
+ final synchronized void txFailure(final ListenableFuture<Void> execFuture, final long txId, final Throwable cause) {
+ LOG.debug("Future #{} failed", txId, cause);
+ phase.txFailure(execFuture, txId, cause);
+ }
+
+ private synchronized void checkCollection() {
+ final Collecting collecting = (Collecting) phase;
+ if (!collecting.done) {
+ final int size = collecting.futures.size();
+ for (int i = 0; i < size; i++) {
+ final ListenableFuture<Void> future = collecting.futures.get(i);
+
+ try {
+ future.get(0, TimeUnit.NANOSECONDS);
+ } catch (final TimeoutException e) {
+ LOG.warn("Future #{}/{} not completed yet", i, size);
+ } catch (final ExecutionException e) {
+ LOG.warn("Future #{}/{} failed", i, size, e.getCause());
+ } catch (final InterruptedException e) {
+ LOG.warn("Interrupted while examining future #{}/{}", i, size, e);
+ }
+ }
+
+ runTimedOut(new TimeoutException("Collection did not finish in " + DEAD_TIMEOUT_SECONDS + " seconds"));
+ }
+ }
+
+ abstract ListenableFuture<Void> execWrite(final long txId);
+
+ abstract void runFailed(Throwable cause);
+
+ abstract void runSuccessful(long allTx);
+
+ abstract void runTimedOut(Exception cause);
+}
Code Review / controller.git / blobdiff