2 * Copyright (c) 2017 Pantheon Technologies, s.r.o. 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
8 package org.opendaylight.controller.clustering.it.provider.impl;
10 import com.google.common.base.Stopwatch;
11 import com.google.common.util.concurrent.FutureCallback;
12 import com.google.common.util.concurrent.Futures;
13 import com.google.common.util.concurrent.ListenableFuture;
14 import java.util.Collection;
15 import java.util.HashSet;
16 import java.util.concurrent.ExecutionException;
17 import java.util.concurrent.ScheduledExecutorService;
18 import java.util.concurrent.ScheduledFuture;
19 import java.util.concurrent.TimeUnit;
20 import java.util.concurrent.TimeoutException;
21 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.TransactionsParams;
22 import org.opendaylight.yangtools.yang.common.QName;
23 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
24 import org.slf4j.Logger;
25 import org.slf4j.LoggerFactory;
27 abstract class AbstractTransactionHandler {
35 private static final Logger LOG = LoggerFactory.getLogger(AbstractTransactionHandler.class);
37 static final int SECOND_AS_NANO = 1_000_000_000;
38 //2^20 as in the model
39 static final int MAX_ITEM = 1048576;
41 static final QName ID_INTS =
42 QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id-ints").intern();
43 static final QName ID =
44 QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id").intern();
45 static final QName ITEM =
46 QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "item").intern();
47 static final QName NUMBER =
48 QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "number").intern();
50 public static final QName ID_INT =
51 QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id-int").intern();
52 public static final YangInstanceIdentifier ID_INTS_YID = YangInstanceIdentifier.of(ID_INTS);
53 public static final YangInstanceIdentifier ID_INT_YID = ID_INTS_YID.node(ID_INT).toOptimized();
55 static final long INIT_TX_TIMEOUT_SECONDS = 125;
57 private static final long DEAD_TIMEOUT_SECONDS = TimeUnit.MINUTES.toSeconds(15);
59 private final ScheduledExecutorService executor = FinalizableScheduledExecutorService.newSingleThread();
60 private final Collection<ListenableFuture<Void>> futures = new HashSet<>();
61 private final Stopwatch stopwatch = Stopwatch.createUnstarted();
62 private final long runtimeNanos;
63 private final long delayNanos;
65 private ScheduledFuture<?> scheduledFuture;
66 private long txCounter;
69 AbstractTransactionHandler(final TransactionsParams params) {
70 runtimeNanos = TimeUnit.SECONDS.toNanos(params.getSeconds());
71 delayNanos = SECOND_AS_NANO / params.getTransactionsPerSecond();
74 final synchronized void doStart() {
75 scheduledFuture = executor.scheduleAtFixedRate(this::execute, 0, delayNanos, TimeUnit.NANOSECONDS);
77 state = State.RUNNING;
80 private synchronized void execute() {
83 // This could happen due to scheduling artifacts
89 throw new IllegalStateException("Unhandled state " + state);
93 private void runningExecute() {
94 final long elapsed = stopwatch.elapsed(TimeUnit.NANOSECONDS);
95 if (elapsed >= runtimeNanos) {
96 LOG.debug("Reached maximum run time with {} outstanding futures", futures.size());
97 if (!checkSuccessful()) {
98 state = State.WAITING;
99 scheduledFuture.cancel(false);
100 scheduledFuture = executor.schedule(this::checkComplete, DEAD_TIMEOUT_SECONDS, TimeUnit.SECONDS);
106 // Not completed yet: create a transaction and hook it up
107 final long txId = txCounter++;
108 final ListenableFuture<Void> execFuture = execWrite(txId);
109 LOG.debug("New future #{} allocated", txId);
111 // Ordering is important: we need to add the future before hooking the callback
112 futures.add(execFuture);
113 Futures.addCallback(execFuture, new FutureCallback<Void>() {
115 public void onSuccess(final Void result) {
116 txSuccess(execFuture, txId);
120 public void onFailure(final Throwable cause) {
121 txFailure(execFuture, txId, cause);
126 final void txSuccess(final ListenableFuture<Void> execFuture, final long txId) {
127 LOG.debug("Future #{} completed successfully", txId);
128 futures.remove(execFuture);
139 throw new IllegalStateException("Unhandled state " + state);
143 final void txFailure(final ListenableFuture<Void> execFuture, final long txId, final Throwable cause) {
144 LOG.debug("Future #{} failed", txId, cause);
145 futures.remove(execFuture);
153 state = State.FAILED;
154 scheduledFuture.cancel(false);
158 throw new IllegalStateException("Unhandled state " + state);
162 private void checkComplete() {
163 final int size = futures.size();
169 for (ListenableFuture<Void> future : futures) {
171 future.get(0, TimeUnit.NANOSECONDS);
172 } catch (final TimeoutException e) {
173 LOG.warn("Future #{}/{} not completed yet", offset, size);
174 } catch (final ExecutionException e) {
175 LOG.warn("Future #{}/{} failed", offset, size, e.getCause());
176 } catch (final InterruptedException e) {
177 LOG.warn("Interrupted while examining future #{}/{}", offset, size, e);
183 state = State.FAILED;
184 runTimedOut(new TimeoutException("Collection did not finish in " + DEAD_TIMEOUT_SECONDS + " seconds"));
187 private boolean checkSuccessful() {
188 if (futures.isEmpty()) {
189 LOG.debug("Completed waiting for all futures");
190 state = State.SUCCESSFUL;
191 scheduledFuture.cancel(false);
192 runSuccessful(txCounter);
199 abstract ListenableFuture<Void> execWrite(final long txId);
201 abstract void runFailed(Throwable cause);
203 abstract void runSuccessful(long allTx);
205 abstract void runTimedOut(Exception cause);