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 java.util.Collection;
import java.util.Collections;
import java.util.concurrent.ScheduledFuture;
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;
* 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 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());
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);
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");
}
- abstract ListenableFuture<Void> execWrite(final long txId);
+ 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);
}