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.Collections;
16 import java.util.HashSet;
17 import java.util.concurrent.ExecutionException;
18 import java.util.concurrent.ScheduledExecutorService;
19 import java.util.concurrent.ScheduledFuture;
20 import java.util.concurrent.TimeUnit;
21 import java.util.concurrent.TimeoutException;
22 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.TransactionsParams;
23 import org.opendaylight.yangtools.yang.common.QName;
24 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
25 import org.slf4j.Logger;
26 import org.slf4j.LoggerFactory;
28 abstract class AbstractTransactionHandler {
36 private static final Logger LOG = LoggerFactory.getLogger(AbstractTransactionHandler.class);
38 static final int SECOND_AS_NANO = 1_000_000_000;
39 //2^20 as in the model
40 static final int MAX_ITEM = 1048576;
42 static final QName ID_INTS =
43 QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id-ints").intern();
44 static final QName ID =
45 QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id").intern();
46 static final QName ITEM =
47 QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "item").intern();
48 static final QName NUMBER =
49 QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "number").intern();
51 public static final QName ID_INT =
52 QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id-int").intern();
53 public static final YangInstanceIdentifier ID_INTS_YID = YangInstanceIdentifier.of(ID_INTS);
54 public static final YangInstanceIdentifier ID_INT_YID = ID_INTS_YID.node(ID_INT).toOptimized();
56 static final long INIT_TX_TIMEOUT_SECONDS = 125;
58 private static final long DEAD_TIMEOUT_SECONDS = TimeUnit.MINUTES.toSeconds(15);
61 * writingExecutor is a single thread executor. Only this thread will write to datastore,
62 * incurring sleep penalties if backend is not responsive. This thread never changes, but reads State.
63 * This thread only adds to futures set.
65 private final ScheduledExecutorService writingExecutor = FinalizableScheduledExecutorService.newSingleThread();
67 * completingExecutor is a single thread executor. Only this thread writes to State.
68 * This thread should never incur any sleep penalty, so RPC response should always come on time.
69 * This thread only removes from futures set.
71 private final ScheduledExecutorService completingExecutor = FinalizableScheduledExecutorService.newSingleThread();
72 private final Collection<ListenableFuture<Void>> futures = Collections.synchronizedSet(new HashSet<>());
73 private final Stopwatch stopwatch = Stopwatch.createUnstarted();
74 private final long runtimeNanos;
75 private final long delayNanos;
77 private ScheduledFuture<?> writingFuture;
78 private ScheduledFuture<?> completingFuture;
79 private long txCounter;
80 private volatile State state;
82 AbstractTransactionHandler(final TransactionsParams params) {
83 runtimeNanos = TimeUnit.SECONDS.toNanos(params.getSeconds());
84 delayNanos = SECOND_AS_NANO / params.getTransactionsPerSecond();
87 final synchronized void doStart() {
88 // Setup state first...
90 state = State.RUNNING;
92 writingFuture = writingExecutor.scheduleAtFixedRate(this::execute, 0, delayNanos, TimeUnit.NANOSECONDS);
95 private void execute() {
96 // Single volatile access
97 final State local = state;
101 // This could happen due to scheduling artifacts
107 throw new IllegalStateException("Unhandled state " + local);
111 private void runningExecute() {
112 final long elapsed = stopwatch.elapsed(TimeUnit.NANOSECONDS);
113 if (elapsed >= runtimeNanos) {
114 LOG.debug("Reached maximum run time with {} outstanding futures", futures.size());
115 completingExecutor.schedule(this::runtimeUp, 0, TimeUnit.SECONDS);
119 // Not completed yet: create a transaction and hook it up
120 final long txId = txCounter++;
121 final ListenableFuture<Void> execFuture = execWrite(txId);
122 LOG.debug("New future #{} allocated", txId);
124 // Ordering is important: we need to add the future before hooking the callback
125 futures.add(execFuture);
126 Futures.addCallback(execFuture, new FutureCallback<Void>() {
128 public void onSuccess(final Void result) {
129 txSuccess(execFuture, txId);
133 public void onFailure(final Throwable cause) {
134 txFailure(execFuture, txId, cause);
136 }, completingExecutor);
139 private void runtimeUp() {
140 // checkSuccessful has two call sites, it is simpler to create completingFuture unconditionally.
141 completingFuture = completingExecutor.schedule(this::checkComplete, DEAD_TIMEOUT_SECONDS, TimeUnit.SECONDS);
142 if (!checkSuccessful()) {
143 state = State.WAITING;
144 writingFuture.cancel(false);
148 private boolean checkSuccessful() {
149 if (futures.isEmpty()) {
150 LOG.debug("Completed waiting for all futures");
151 state = State.SUCCESSFUL;
152 completingFuture.cancel(false);
153 runSuccessful(txCounter);
160 final void txSuccess(final ListenableFuture<Void> execFuture, final long txId) {
161 LOG.debug("Future #{} completed successfully", txId);
162 futures.remove(execFuture);
164 final State local = state;
174 throw new IllegalStateException("Unhandled state " + local);
178 final void txFailure(final ListenableFuture<Void> execFuture, final long txId, final Throwable cause) {
179 LOG.debug("Future #{} failed", txId, cause);
180 futures.remove(execFuture);
182 final State local = state;
189 state = State.FAILED;
190 writingFuture.cancel(false);
194 throw new IllegalStateException("Unhandled state " + local);
198 private void checkComplete() {
199 final int size = futures.size();
204 // Guards iteration against concurrent modification from callbacks
205 synchronized (futures) {
208 for (ListenableFuture<Void> future : futures) {
210 future.get(0, TimeUnit.NANOSECONDS);
211 } catch (final TimeoutException e) {
212 LOG.warn("Future #{}/{} not completed yet", offset, size);
213 } catch (final ExecutionException e) {
214 LOG.warn("Future #{}/{} failed", offset, size, e.getCause());
215 } catch (final InterruptedException e) {
216 LOG.warn("Interrupted while examining future #{}/{}", offset, size, e);
223 state = State.FAILED;
224 runTimedOut(new TimeoutException("Collection did not finish in " + DEAD_TIMEOUT_SECONDS + " seconds"));
227 abstract ListenableFuture<Void> execWrite(long txId);
229 abstract void runFailed(Throwable cause);
231 abstract void runSuccessful(long allTx);
233 abstract void runTimedOut(Exception cause);