opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/AbstractTransactionHandler.java

   1 /*
   2  * Copyright (c) 2017 Pantheon Technologies, s.r.o. 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.clustering.it.provider.impl;
   9
  10 import com.google.common.base.Stopwatch;
  11 import com.google.common.util.concurrent.FluentFuture;
  12 import com.google.common.util.concurrent.FutureCallback;
  13 import com.google.common.util.concurrent.ListenableFuture;
  14 import com.google.common.util.concurrent.ThreadFactoryBuilder;
  15 import java.util.Collection;
  16 import java.util.Collections;
  17 import java.util.HashSet;
  18 import java.util.concurrent.ExecutionException;
  19 import java.util.concurrent.ScheduledExecutorService;
  20 import java.util.concurrent.ScheduledFuture;
  21 import java.util.concurrent.ScheduledThreadPoolExecutor;
  22 import java.util.concurrent.TimeUnit;
  23 import java.util.concurrent.TimeoutException;
  24 import java.util.concurrent.atomic.AtomicLong;
  25 import org.opendaylight.mdsal.common.api.CommitInfo;
  26 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.TransactionsParams;
  27 import org.opendaylight.yangtools.yang.common.QName;
  28 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  29 import org.slf4j.Logger;
  30 import org.slf4j.LoggerFactory;
  31
  32 abstract class AbstractTransactionHandler {
  33     private enum State {
  34         RUNNING,
  35         WAITING,
  36         SUCCESSFUL,
  37         FAILED,
  38     }
  39
  40     private static final Logger LOG = LoggerFactory.getLogger(AbstractTransactionHandler.class);
  41
  42     static final int SECOND_AS_NANO = 1_000_000_000;
  43     //2^20 as in the model
  44     static final int MAX_ITEM = 1048576;
  45
  46     static final QName ID_INTS =
  47             QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id-ints").intern();
  48     static final QName ID =
  49             QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id").intern();
  50     static final QName ITEM =
  51             QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "item").intern();
  52     static final QName NUMBER =
  53             QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "number").intern();
  54
  55     public static final QName ID_INT =
  56             QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id-int").intern();
  57     public static final YangInstanceIdentifier ID_INTS_YID = YangInstanceIdentifier.of(ID_INTS);
  58     public static final YangInstanceIdentifier ID_INT_YID = ID_INTS_YID.node(ID_INT).toOptimized();
  59
  60     static final long INIT_TX_TIMEOUT_SECONDS = 125;
  61
  62     private static final long DEAD_TIMEOUT_SECONDS = TimeUnit.MINUTES.toSeconds(15);
  63     private static final AtomicLong COUNTER = new AtomicLong();
  64
  65     /*
  66      * writingExecutor is a single thread executor. Only this thread will write to datastore,
  67      * incurring sleep penalties if backend is not responsive. This thread never changes, but reads State.
  68      * This thread only adds to futures set.
  69      */
  70     private final ScheduledExecutorService writingExecutor = newExecutorService("writing");
  71     /*
  72      * completingExecutor is a single thread executor. Only this thread writes to State.
  73      * This thread should never incur any sleep penalty, so RPC response should always come on time.
  74      * This thread only removes from futures set.
  75      */
  76     private final ScheduledExecutorService completingExecutor = newExecutorService("completing");
  77     private final Collection<ListenableFuture<?>> futures = Collections.synchronizedSet(new HashSet<>());
  78     private final Stopwatch stopwatch = Stopwatch.createUnstarted();
  79     private final long runtimeNanos;
  80     private final long delayNanos;
  81
  82     private ScheduledFuture<?> writingFuture;
  83     private ScheduledFuture<?> completingFuture;
  84     private final AtomicLong txCounter = new AtomicLong();
  85     private volatile State state;
  86
  87     AbstractTransactionHandler(final TransactionsParams params) {
  88         runtimeNanos = TimeUnit.SECONDS.toNanos(params.getSeconds().toJava());
  89         delayNanos = SECOND_AS_NANO / params.getTransactionsPerSecond().toJava();
  90     }
  91
  92     final synchronized void doStart() {
  93         // Setup state first...
  94         stopwatch.start();
  95         state = State.RUNNING;
  96
  97         writingFuture = writingExecutor.scheduleAtFixedRate(this::execute, 0, delayNanos, TimeUnit.NANOSECONDS);
  98     }
  99
 100     private void execute() {
 101         // Single volatile access
 102         final State local = state;
 103
 104         switch (local) {
 105             case FAILED:
 106                 // This could happen due to scheduling artifacts
 107                 break;
 108             case RUNNING:
 109                 runningExecute();
 110                 break;
 111             default:
 112                 throw new IllegalStateException("Unhandled state " + local);
 113         }
 114     }
 115
 116     private void runningExecute() {
 117         final long elapsed = stopwatch.elapsed(TimeUnit.NANOSECONDS);
 118         if (elapsed >= runtimeNanos) {
 119             LOG.debug("Reached maximum run time with {} outstanding futures", futures.size());
 120             completingExecutor.schedule(this::runtimeUp, 0, TimeUnit.SECONDS);
 121             return;
 122         }
 123
 124         // Not completed yet: create a transaction and hook it up
 125         final long txId = txCounter.incrementAndGet();
 126         final FluentFuture<? extends CommitInfo> execFuture = execWrite(txId);
 127         LOG.debug("New future #{} allocated", txId);
 128
 129         // Ordering is important: we need to add the future before hooking the callback
 130         futures.add(execFuture);
 131         execFuture.addCallback(new FutureCallback<CommitInfo>() {
 132             @Override
 133             public void onSuccess(final CommitInfo result) {
 134                 txSuccess(execFuture, txId);
 135             }
 136
 137             @Override
 138             public void onFailure(final Throwable cause) {
 139                 txFailure(execFuture, txId, cause);
 140             }
 141         }, completingExecutor);
 142     }
 143
 144     private void runtimeUp() {
 145         // checkSuccessful has two call sites, it is simpler to create completingFuture unconditionally.
 146         completingFuture = completingExecutor.schedule(this::checkComplete, DEAD_TIMEOUT_SECONDS, TimeUnit.SECONDS);
 147         if (!checkSuccessful()) {
 148             state = State.WAITING;
 149             writingFuture.cancel(false);
 150         }
 151     }
 152
 153     private boolean checkSuccessful() {
 154         if (futures.isEmpty()) {
 155             LOG.debug("Completed waiting for all futures");
 156             state = State.SUCCESSFUL;
 157             completingFuture.cancel(false);
 158             runSuccessful(txCounter.get());
 159             shutdownExecutors();
 160             return true;
 161         }
 162
 163         return false;
 164     }
 165
 166     final void txSuccess(final ListenableFuture<?> execFuture, final long txId) {
 167         LOG.debug("Future #{} completed successfully", txId);
 168         futures.remove(execFuture);
 169
 170         final State local = state;
 171         switch (local) {
 172             case FAILED:
 173             case RUNNING:
 174                 // No-op
 175                 break;
 176             case WAITING:
 177                 checkSuccessful();
 178                 break;
 179             default:
 180                 throw new IllegalStateException("Unhandled state " + local);
 181         }
 182     }
 183
 184     final void txFailure(final ListenableFuture<?> execFuture, final long txId, final Throwable cause) {
 185         LOG.error("Commit future failed for tx # {}", txId, cause);
 186         futures.remove(execFuture);
 187
 188         final State local = state;
 189         switch (local) {
 190             case FAILED:
 191                 // no-op
 192                 break;
 193             case RUNNING:
 194             case WAITING:
 195                 state = State.FAILED;
 196                 writingFuture.cancel(false);
 197                 runFailed(cause, txId);
 198                 shutdownExecutors();
 199                 break;
 200             default:
 201                 throw new IllegalStateException("Unhandled state " + local);
 202         }
 203     }
 204
 205     private void checkComplete() {
 206         final int size = futures.size();
 207         if (size == 0) {
 208             return;
 209         }
 210
 211         // Guards iteration against concurrent modification from callbacks
 212         synchronized (futures) {
 213             int offset = 0;
 214
 215             for (ListenableFuture<?> future : futures) {
 216                 try {
 217                     future.get(0, TimeUnit.NANOSECONDS);
 218                 } catch (final TimeoutException e) {
 219                     LOG.warn("Future #{}/{} not completed yet", offset, size);
 220                 } catch (final ExecutionException e) {
 221                     LOG.warn("Future #{}/{} failed", offset, size, e.getCause());
 222                 } catch (final InterruptedException e) {
 223                     LOG.warn("Interrupted while examining future #{}/{}", offset, size, e);
 224                 }
 225
 226                 ++offset;
 227             }
 228         }
 229
 230         state = State.FAILED;
 231         runTimedOut("Transactions did not finish in " + DEAD_TIMEOUT_SECONDS + " seconds");
 232         shutdownExecutors();
 233     }
 234
 235     private void shutdownExecutors() {
 236         writingExecutor.shutdown();
 237         completingExecutor.shutdown();
 238     }
 239
 240     abstract FluentFuture<? extends CommitInfo> execWrite(long txId);
 241
 242     abstract void runFailed(Throwable cause, long txId);
 243
 244     abstract void runSuccessful(long allTx);
 245
 246     abstract void runTimedOut(String cause);
 247
 248     private ScheduledExecutorService newExecutorService(final String kind) {
 249         final ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1, new ThreadFactoryBuilder()
 250             .setDaemon(true)
 251             .setNameFormat(getClass().getSimpleName() + "-" + kind + "-" + COUNTER.getAndIncrement() + "%d")
 252             .build());
 253         executor.setKeepAliveTime(15, TimeUnit.SECONDS);
 254         executor.allowCoreThreadTimeOut(true);
 255         return executor;
 256     }
 257 }