package org.opendaylight.controller.cluster.databroker.actors.dds;
import com.google.common.base.Optional;
-import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.CheckedFuture;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
@Override
void doDelete(final YangInstanceIdentifier path) {
- appendModification(new TransactionDelete(path));
+ appendModification(new TransactionDelete(path), Optional.absent());
}
@Override
void doMerge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
- appendModification(new TransactionMerge(path, data));
+ appendModification(new TransactionMerge(path, data), Optional.absent());
}
@Override
void doWrite(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
- appendModification(new TransactionWrite(path, data));
+ appendModification(new TransactionWrite(path, data), Optional.absent());
}
private <T> CheckedFuture<T, ReadFailedException> sendReadRequest(final AbstractReadTransactionRequest<?> request,
}
}
+ private void ensureFlushedBuider(final Optional<Long> enqueuedTicks) {
+ if (builderBusy) {
+ flushBuilder(enqueuedTicks);
+ }
+ }
+
private void flushBuilder() {
+ flushBuilder(Optional.absent());
+ }
+
+ private void flushBuilder(final Optional<Long> enqueuedTicks) {
final ModifyTransactionRequest request = builder.build();
builderBusy = false;
- sendModification(request);
+ sendModification(request, enqueuedTicks);
}
- private void sendModification(final TransactionRequest<?> request) {
- sendRequest(request, response -> completeModify(request, response));
- }
-
- @Override
- void handleForwardedLocalRequest(final AbstractLocalTransactionRequest<?> request,
- final Consumer<Response<?, ?>> callback) {
- if (request instanceof CommitLocalTransactionRequest) {
- replayLocalCommitRequest((CommitLocalTransactionRequest) request, callback);
- } else if (request instanceof AbortLocalTransactionRequest) {
- sendRequest(abortRequest(), callback);
+ private void sendModification(final TransactionRequest<?> request, final Optional<Long> enqueuedTicks) {
+ if (enqueuedTicks.isPresent()) {
+ enqueueRequest(request, response -> completeModify(request, response), enqueuedTicks.get().longValue());
} else {
- throw new IllegalStateException("Unhandled request " + request);
+ sendRequest(request, response -> completeModify(request, response));
}
}
- private void replayLocalCommitRequest(final CommitLocalTransactionRequest request,
- final Consumer<Response<?, ?>> callback) {
- final DataTreeModification mod = request.getModification();
- mod.applyToCursor(new AbstractDataTreeModificationCursor() {
- @Override
- public void write(final PathArgument child, final NormalizedNode<?, ?> data) {
- doWrite(current().node(child), data);
- }
-
- @Override
- public void merge(final PathArgument child, final NormalizedNode<?, ?> data) {
- doMerge(current().node(child), data);
- }
-
- @Override
- public void delete(final PathArgument child) {
- doDelete(current().node(child));
- }
- });
-
- sendRequest(commitRequest(request.isCoordinated()), callback);
- }
-
- @Override
- void handleForwardedRemoteRequest(final TransactionRequest<?> request,
- final @Nullable Consumer<Response<?, ?>> callback) {
- nextSequence();
-
- if (callback == null) {
- sendModification(request);
- return;
- }
-
- /*
- * FindBugs is utterly stupid, as it does not recognize the fact that we have checked for null
- * and reports NP_PARAMETER_MUST_BE_NONNULL_BUT_MARKED_AS_NULLABLE in the lambda below.
- */
- final Consumer<Response<?, ?>> findBugsIsStupid = callback;
-
- // FIXME: do not use sendRequest() once we have throttling in place, as we have already waited the
- // period required to get into the queue.
- sendRequest(request, response -> {
- findBugsIsStupid.accept(Preconditions.checkNotNull(response));
- completeModify(request, response);
- });
+ private void appendModification(final TransactionModification modification) {
+ appendModification(modification, Optional.absent());
}
- private void appendModification(final TransactionModification modification) {
+ private void appendModification(final TransactionModification modification, final Optional<Long> enqueuedTicks) {
if (operationFailure == null) {
ensureInitializedBuilder();
builder.addModification(modification);
if (builder.size() >= REQUEST_MAX_MODIFICATIONS) {
- flushBuilder();
+ flushBuilder(enqueuedTicks);
}
} else {
LOG.debug("Transaction {} failed, not attempting further transactions", getIdentifier());
if (builderBusy) {
final ModifyTransactionRequest request = builder.build();
builderBusy = false;
- successor.handleForwardedRemoteRequest(request, null);
+ forwardToSuccessor(successor, request, null);
}
}
if (maybeProto.isPresent()) {
ensureSealed();
+ final TransactionRequest<?> tmp;
switch (maybeProto.get()) {
case ABORT:
- sendRequest(abortRequest(), callback);
+ tmp = abortRequest();
+ sendRequest(tmp, resp -> {
+ completeModify(tmp, resp);
+ callback.accept(resp);
+ });
break;
case SIMPLE:
- sendRequest(commitRequest(false), callback);
+ tmp = commitRequest(false);
+ sendRequest(tmp, resp -> {
+ completeModify(tmp, resp);
+ callback.accept(resp);
+ });
break;
case THREE_PHASE:
- sendRequest(commitRequest(true), callback);
+ tmp = commitRequest(true);
+ sendRequest(tmp, resp -> {
+ recordSuccessfulRequest(tmp);
+ callback.accept(resp);
+ });
break;
case READY:
//no op
} else if (request instanceof ReadTransactionRequest) {
ensureFlushedBuider();
sendRequest(new ReadTransactionRequest(getIdentifier(), nextSequence(), localActor(),
- ((ReadTransactionRequest) request).getPath(), isSnapshotOnly()), callback);
+ ((ReadTransactionRequest) request).getPath(), isSnapshotOnly()), resp -> {
+ recordFinishedRequest();
+ callback.accept(resp);
+ });
} else if (request instanceof ExistsTransactionRequest) {
ensureFlushedBuider();
sendRequest(new ExistsTransactionRequest(getIdentifier(), nextSequence(), localActor(),
- ((ExistsTransactionRequest) request).getPath(), isSnapshotOnly()), callback);
+ ((ExistsTransactionRequest) request).getPath(), isSnapshotOnly()), resp -> {
+ recordFinishedRequest();
+ callback.accept(resp);
+ });
} else if (request instanceof TransactionPreCommitRequest) {
ensureFlushedBuider();
- sendRequest(new TransactionPreCommitRequest(getIdentifier(), nextSequence(), localActor()), callback);
+ final TransactionRequest<?> tmp = new TransactionPreCommitRequest(getIdentifier(), nextSequence(),
+ localActor());
+ sendRequest(tmp, resp -> {
+ recordSuccessfulRequest(tmp);
+ callback.accept(resp);
+ });
} else if (request instanceof TransactionDoCommitRequest) {
ensureFlushedBuider();
sendRequest(new TransactionDoCommitRequest(getIdentifier(), nextSequence(), localActor()), callback);
ensureFlushedBuider();
sendAbort(callback);
} else if (request instanceof TransactionPurgeRequest) {
- purge();
+ sendPurge();
} else {
throw new IllegalArgumentException("Unhandled request {}" + request);
}
final Consumer<Response<?, ?>> callback) {
successor.handleForwardedRemoteRequest(request, callback);
}
+
+ @Override
+ void handleReplayedLocalRequest(final AbstractLocalTransactionRequest<?> request,
+ final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
+ if (request instanceof CommitLocalTransactionRequest) {
+ replayLocalCommitRequest((CommitLocalTransactionRequest) request, callback, enqueuedTicks);
+ } else if (request instanceof AbortLocalTransactionRequest) {
+ enqueueRequest(abortRequest(), callback, enqueuedTicks);
+ } else {
+ throw new IllegalStateException("Unhandled request " + request);
+ }
+ }
+
+ private void replayLocalCommitRequest(final CommitLocalTransactionRequest request,
+ final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
+ final DataTreeModification mod = request.getModification();
+ final Optional<Long> optTicks = Optional.of(Long.valueOf(enqueuedTicks));
+
+ mod.applyToCursor(new AbstractDataTreeModificationCursor() {
+ @Override
+ public void write(final PathArgument child, final NormalizedNode<?, ?> data) {
+ appendModification(new TransactionWrite(current().node(child), data), optTicks);
+ }
+
+ @Override
+ public void merge(final PathArgument child, final NormalizedNode<?, ?> data) {
+ appendModification(new TransactionMerge(current().node(child), data), optTicks);
+ }
+
+ @Override
+ public void delete(final PathArgument child) {
+ appendModification(new TransactionDelete(current().node(child)), optTicks);
+ }
+ });
+
+ enqueueRequest(commitRequest(request.isCoordinated()), callback, enqueuedTicks);
+ }
+
+ @Override
+ void handleReplayedRemoteRequest(final TransactionRequest<?> request,
+ final @Nullable Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
+ final Consumer<Response<?, ?>> cb = callback != null ? callback : resp -> { };
+ final Optional<Long> optTicks = Optional.of(Long.valueOf(enqueuedTicks));
+
+ if (request instanceof ModifyTransactionRequest) {
+ final ModifyTransactionRequest req = (ModifyTransactionRequest) request;
+ for (TransactionModification mod : req.getModifications()) {
+ appendModification(mod, optTicks);
+ }
+
+ final java.util.Optional<PersistenceProtocol> maybeProto = req.getPersistenceProtocol();
+ if (maybeProto.isPresent()) {
+ ensureSealed();
+
+ final TransactionRequest<?> tmp;
+ switch (maybeProto.get()) {
+ case ABORT:
+ tmp = abortRequest();
+ enqueueRequest(tmp, resp -> {
+ completeModify(tmp, resp);
+ cb.accept(resp);
+ }, enqueuedTicks);
+ break;
+ case SIMPLE:
+ tmp = commitRequest(false);
+ enqueueRequest(tmp, resp -> {
+ completeModify(tmp, resp);
+ cb.accept(resp);
+ }, enqueuedTicks);
+ break;
+ case THREE_PHASE:
+ tmp = commitRequest(true);
+ enqueueRequest(tmp, resp -> {
+ recordSuccessfulRequest(tmp);
+ cb.accept(resp);
+ }, enqueuedTicks);
+ break;
+ case READY:
+ //no op
+ break;
+ default:
+ throw new IllegalArgumentException("Unhandled protocol " + maybeProto.get());
+ }
+ }
+ } else if (request instanceof ReadTransactionRequest) {
+ ensureFlushedBuider(optTicks);
+ enqueueRequest(new ReadTransactionRequest(getIdentifier(), nextSequence(), localActor(),
+ ((ReadTransactionRequest) request).getPath(), isSnapshotOnly()), resp -> {
+ recordFinishedRequest();
+ cb.accept(resp);
+ }, enqueuedTicks);
+ } else if (request instanceof ExistsTransactionRequest) {
+ ensureFlushedBuider(optTicks);
+ enqueueRequest(new ExistsTransactionRequest(getIdentifier(), nextSequence(), localActor(),
+ ((ExistsTransactionRequest) request).getPath(), isSnapshotOnly()), resp -> {
+ recordFinishedRequest();
+ cb.accept(resp);
+ }, enqueuedTicks);
+ } else if (request instanceof TransactionPreCommitRequest) {
+ ensureFlushedBuider(optTicks);
+ final TransactionRequest<?> tmp = new TransactionPreCommitRequest(getIdentifier(), nextSequence(),
+ localActor());
+ enqueueRequest(tmp, resp -> {
+ recordSuccessfulRequest(tmp);
+ cb.accept(resp);
+ }, enqueuedTicks);
+ } else if (request instanceof TransactionDoCommitRequest) {
+ ensureFlushedBuider(optTicks);
+ enqueueRequest(new TransactionDoCommitRequest(getIdentifier(), nextSequence(), localActor()), callback,
+ enqueuedTicks);
+ } else if (request instanceof TransactionAbortRequest) {
+ ensureFlushedBuider(optTicks);
+ enqueueAbort(callback, enqueuedTicks);
+ } else if (request instanceof TransactionPurgeRequest) {
+ enqueuePurge(enqueuedTicks);
+ } else {
+ throw new IllegalArgumentException("Unhandled request {}" + request);
+ }
+ }
}
Code Review / controller.git / blobdiff