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 java.util.concurrent.atomic.AtomicLong;
23 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.TransactionsParams;
24 import org.opendaylight.yangtools.yang.common.QName;
25 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
26 import org.slf4j.Logger;
27 import org.slf4j.LoggerFactory;
29 abstract class AbstractTransactionHandler {
37 private static final Logger LOG = LoggerFactory.getLogger(AbstractTransactionHandler.class);
39 static final int SECOND_AS_NANO = 1_000_000_000;
40 //2^20 as in the model
41 static final int MAX_ITEM = 1048576;
43 static final QName ID_INTS =
44 QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id-ints").intern();
45 static final QName ID =
46 QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id").intern();
47 static final QName ITEM =
48 QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "item").intern();
49 static final QName NUMBER =
50 QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "number").intern();
52 public static final QName ID_INT =
53 QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id-int").intern();
54 public static final YangInstanceIdentifier ID_INTS_YID = YangInstanceIdentifier.of(ID_INTS);
55 public static final YangInstanceIdentifier ID_INT_YID = ID_INTS_YID.node(ID_INT).toOptimized();
57 static final long INIT_TX_TIMEOUT_SECONDS = 125;
59 private static final long DEAD_TIMEOUT_SECONDS = TimeUnit.MINUTES.toSeconds(15);
62 * writingExecutor is a single thread executor. Only this thread will write to datastore,
63 * incurring sleep penalties if backend is not responsive. This thread never changes, but reads State.
64 * This thread only adds to futures set.
66 private final ScheduledExecutorService writingExecutor = FinalizableScheduledExecutorService.newSingleThread();
68 * completingExecutor is a single thread executor. Only this thread writes to State.
69 * This thread should never incur any sleep penalty, so RPC response should always come on time.
70 * This thread only removes from futures set.
72 private final ScheduledExecutorService completingExecutor = FinalizableScheduledExecutorService.newSingleThread();
73 private final Collection<ListenableFuture<?>> futures = Collections.synchronizedSet(new HashSet<>());
74 private final Stopwatch stopwatch = Stopwatch.createUnstarted();
75 private final long runtimeNanos;
76 private final long delayNanos;
78 private ScheduledFuture<?> writingFuture;
79 private ScheduledFuture<?> completingFuture;
80 private final AtomicLong txCounter = new AtomicLong();
81 private volatile State state;
83 AbstractTransactionHandler(final TransactionsParams params) {
84 runtimeNanos = TimeUnit.SECONDS.toNanos(params.getSeconds());
85 delayNanos = SECOND_AS_NANO / params.getTransactionsPerSecond();
88 final synchronized void doStart() {
89 // Setup state first...
91 state = State.RUNNING;
93 writingFuture = writingExecutor.scheduleAtFixedRate(this::execute, 0, delayNanos, TimeUnit.NANOSECONDS);
96 private void execute() {
97 // Single volatile access
98 final State local = state;
102 // This could happen due to scheduling artifacts
108 throw new IllegalStateException("Unhandled state " + local);
112 private void runningExecute() {
113 final long elapsed = stopwatch.elapsed(TimeUnit.NANOSECONDS);
114 if (elapsed >= runtimeNanos) {
115 LOG.debug("Reached maximum run time with {} outstanding futures", futures.size());
116 completingExecutor.schedule(this::runtimeUp, 0, TimeUnit.SECONDS);
120 // Not completed yet: create a transaction and hook it up
121 final long txId = txCounter.incrementAndGet();
122 final ListenableFuture<?> execFuture = execWrite(txId);
123 LOG.debug("New future #{} allocated", txId);
125 // Ordering is important: we need to add the future before hooking the callback
126 futures.add(execFuture);
127 Futures.addCallback(execFuture, new FutureCallback<Object>() {
129 public void onSuccess(final Object result) {
130 txSuccess(execFuture, txId);
134 public void onFailure(final Throwable cause) {
135 txFailure(execFuture, txId, cause);
137 }, completingExecutor);
140 private void runtimeUp() {
141 // checkSuccessful has two call sites, it is simpler to create completingFuture unconditionally.
142 completingFuture = completingExecutor.schedule(this::checkComplete, DEAD_TIMEOUT_SECONDS, TimeUnit.SECONDS);
143 if (!checkSuccessful()) {
144 state = State.WAITING;
145 writingFuture.cancel(false);
149 private boolean checkSuccessful() {
150 if (futures.isEmpty()) {
151 LOG.debug("Completed waiting for all futures");
152 state = State.SUCCESSFUL;
153 completingFuture.cancel(false);
154 runSuccessful(txCounter.get());
161 final void txSuccess(final ListenableFuture<?> execFuture, final long txId) {
162 LOG.debug("Future #{} completed successfully", txId);
163 futures.remove(execFuture);
165 final State local = state;
175 throw new IllegalStateException("Unhandled state " + local);
179 final void txFailure(final ListenableFuture<?> execFuture, final long txId, final Throwable cause) {
180 LOG.error("Commit future failed for tx # {}", txId, cause);
181 futures.remove(execFuture);
183 final State local = state;
190 state = State.FAILED;
191 writingFuture.cancel(false);
192 runFailed(cause, txId);
195 throw new IllegalStateException("Unhandled state " + local);
199 private void checkComplete() {
200 final int size = futures.size();
205 // Guards iteration against concurrent modification from callbacks
206 synchronized (futures) {
209 for (ListenableFuture<?> future : futures) {
211 future.get(0, TimeUnit.NANOSECONDS);
212 } catch (final TimeoutException e) {
213 LOG.warn("Future #{}/{} not completed yet", offset, size);
214 } catch (final ExecutionException e) {
215 LOG.warn("Future #{}/{} failed", offset, size, e.getCause());
216 } catch (final InterruptedException e) {
217 LOG.warn("Interrupted while examining future #{}/{}", offset, size, e);
224 state = State.FAILED;
225 runTimedOut("Transactions did not finish in " + DEAD_TIMEOUT_SECONDS + " seconds");
228 abstract ListenableFuture<?> execWrite(long txId);
230 abstract void runFailed(Throwable cause, long txId);
232 abstract void runSuccessful(long allTx);
234 abstract void runTimedOut(String cause);