package org.opendaylight.controller.clustering.it.provider.impl;
import com.google.common.base.Stopwatch;
+import com.google.common.util.concurrent.FluentFuture;
import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import org.opendaylight.mdsal.common.api.CommitInfo;
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;
static final long INIT_TX_TIMEOUT_SECONDS = 125;
private static final long DEAD_TIMEOUT_SECONDS = TimeUnit.MINUTES.toSeconds(15);
+ private static final AtomicLong COUNTER = new AtomicLong();
/*
* writingExecutor is a single thread executor. Only this thread will write to datastore,
* incurring sleep penalties if backend is not responsive. This thread never changes, but reads State.
* This thread only adds to futures set.
*/
- private final ScheduledExecutorService writingExecutor = FinalizableScheduledExecutorService.newSingleThread();
+ private final ScheduledExecutorService writingExecutor = newExecutorService("writing");
/*
* completingExecutor is a single thread executor. Only this thread writes to State.
* This thread should never incur any sleep penalty, so RPC response should always come on time.
* This thread only removes from futures set.
*/
- private final ScheduledExecutorService completingExecutor = FinalizableScheduledExecutorService.newSingleThread();
- private final Collection<ListenableFuture<Void>> futures = Collections.synchronizedSet(new HashSet<>());
+ private final ScheduledExecutorService completingExecutor = newExecutorService("completing");
+ private final Collection<ListenableFuture<?>> futures = Collections.synchronizedSet(new HashSet<>());
private final Stopwatch stopwatch = Stopwatch.createUnstarted();
private final long runtimeNanos;
private final long delayNanos;
private ScheduledFuture<?> writingFuture;
private ScheduledFuture<?> completingFuture;
- private long txCounter;
+ private final AtomicLong txCounter = new AtomicLong();
private volatile State state;
AbstractTransactionHandler(final TransactionsParams params) {
- runtimeNanos = TimeUnit.SECONDS.toNanos(params.getSeconds());
- delayNanos = SECOND_AS_NANO / params.getTransactionsPerSecond();
+ runtimeNanos = TimeUnit.SECONDS.toNanos(params.getSeconds().toJava());
+ delayNanos = SECOND_AS_NANO / params.getTransactionsPerSecond().toJava();
}
final synchronized void doStart() {
}
// Not completed yet: create a transaction and hook it up
- final long txId = txCounter++;
- final ListenableFuture<Void> execFuture = execWrite(txId);
+ final long txId = txCounter.incrementAndGet();
+ final FluentFuture<? extends CommitInfo> execFuture = execWrite(txId);
LOG.debug("New future #{} allocated", txId);
// Ordering is important: we need to add the future before hooking the callback
futures.add(execFuture);
- Futures.addCallback(execFuture, new FutureCallback<Void>() {
+ execFuture.addCallback(new FutureCallback<CommitInfo>() {
@Override
- public void onSuccess(final Void result) {
+ public void onSuccess(final CommitInfo result) {
txSuccess(execFuture, txId);
}
LOG.debug("Completed waiting for all futures");
state = State.SUCCESSFUL;
completingFuture.cancel(false);
- runSuccessful(txCounter);
+ runSuccessful(txCounter.get());
+ shutdownExecutors();
return true;
}
return false;
}
- final void txSuccess(final ListenableFuture<Void> execFuture, final long txId) {
+ final void txSuccess(final ListenableFuture<?> execFuture, final long txId) {
LOG.debug("Future #{} completed successfully", txId);
futures.remove(execFuture);
}
}
- final void txFailure(final ListenableFuture<Void> execFuture, final long txId, final Throwable cause) {
- LOG.debug("Future #{} failed", txId, cause);
+ final void txFailure(final ListenableFuture<?> execFuture, final long txId, final Throwable cause) {
+ LOG.error("Commit future failed for tx # {}", txId, cause);
futures.remove(execFuture);
final State local = state;
case WAITING:
state = State.FAILED;
writingFuture.cancel(false);
- runFailed(cause);
+ runFailed(cause, txId);
+ shutdownExecutors();
break;
default:
throw new IllegalStateException("Unhandled state " + local);
synchronized (futures) {
int offset = 0;
- for (ListenableFuture<Void> future : futures) {
+ for (ListenableFuture<?> future : futures) {
try {
future.get(0, TimeUnit.NANOSECONDS);
} catch (final TimeoutException e) {
}
state = State.FAILED;
- runTimedOut(new TimeoutException("Collection did not finish in " + DEAD_TIMEOUT_SECONDS + " seconds"));
+ runTimedOut("Transactions did not finish in " + DEAD_TIMEOUT_SECONDS + " seconds");
+ shutdownExecutors();
}
- abstract ListenableFuture<Void> execWrite(final long txId);
+ private void shutdownExecutors() {
+ writingExecutor.shutdown();
+ completingExecutor.shutdown();
+ }
+
+ abstract FluentFuture<? extends CommitInfo> execWrite(long txId);
- abstract void runFailed(Throwable cause);
+ abstract void runFailed(Throwable cause, long txId);
abstract void runSuccessful(long allTx);
- abstract void runTimedOut(Exception cause);
+ abstract void runTimedOut(String cause);
+
+ private ScheduledExecutorService newExecutorService(final String kind) {
+ final ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1, new ThreadFactoryBuilder()
+ .setDaemon(true)
+ .setNameFormat(getClass().getSimpleName() + "-" + kind + "-" + COUNTER.getAndIncrement() + "%d")
+ .build());
+ executor.setKeepAliveTime(15, TimeUnit.SECONDS);
+ executor.allowCoreThreadTimeOut(true);
+ return executor;
+ }
}
Code Review / controller.git / blobdiff