X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fcds-access-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Faccess%2Fclient%2FTransmitQueue.java;h=da64ddf69f93af2a882cda25c261001ca8eb5510;hb=a8fa0eeb173402295e92a986022a61d9cad090d0;hp=24264eeedff77bca598e58d4d3f4f4a6ac996c34;hpb=dafc95d149bc62f101de37e94b9b5e3526d4e87b;p=controller.git diff --git a/opendaylight/md-sal/cds-access-client/src/main/java/org/opendaylight/controller/cluster/access/client/TransmitQueue.java b/opendaylight/md-sal/cds-access-client/src/main/java/org/opendaylight/controller/cluster/access/client/TransmitQueue.java index 24264eeedf..da64ddf69f 100644 --- a/opendaylight/md-sal/cds-access-client/src/main/java/org/opendaylight/controller/cluster/access/client/TransmitQueue.java +++ b/opendaylight/md-sal/cds-access-client/src/main/java/org/opendaylight/controller/cluster/access/client/TransmitQueue.java @@ -13,17 +13,22 @@ import com.google.common.base.Preconditions; import com.google.common.base.Verify; import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.util.ArrayDeque; +import java.util.ArrayList; import java.util.Collection; import java.util.Deque; import java.util.Iterator; +import java.util.List; import java.util.Optional; import java.util.Queue; import javax.annotation.concurrent.NotThreadSafe; import org.opendaylight.controller.cluster.access.concepts.Request; import org.opendaylight.controller.cluster.access.concepts.RequestEnvelope; -import org.opendaylight.controller.cluster.access.concepts.RequestException; import org.opendaylight.controller.cluster.access.concepts.Response; import org.opendaylight.controller.cluster.access.concepts.ResponseEnvelope; +import org.opendaylight.controller.cluster.access.concepts.RuntimeRequestException; +import org.opendaylight.controller.cluster.access.concepts.SliceableMessage; +import org.opendaylight.controller.cluster.messaging.MessageSlicer; +import org.opendaylight.controller.cluster.messaging.SliceOptions; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -70,19 +75,29 @@ abstract class TransmitQueue { } @Override - TransmittedConnectionEntry transmit(final ConnectionEntry entry, final long now) { + Optional transmit(final ConnectionEntry entry, final long now) { throw new UnsupportedOperationException("Attempted to transmit on a halted queue"); } + + @Override + void preComplete(ResponseEnvelope envelope) { + } } static final class Transmitting extends TransmitQueue { + private static final long NOT_SLICING = -1; + private final BackendInfo backend; + private final MessageSlicer messageSlicer; private long nextTxSequence; + private long currentSlicedEnvSequenceId = NOT_SLICING; // For ConnectedClientConnection. - Transmitting(final TransmitQueue oldQueue, final int targetDepth, final BackendInfo backend, final long now) { + Transmitting(final TransmitQueue oldQueue, final int targetDepth, final BackendInfo backend, final long now, + final MessageSlicer messageSlicer) { super(oldQueue, targetDepth, now); this.backend = Preconditions.checkNotNull(backend); + this.messageSlicer = Preconditions.checkNotNull(messageSlicer); } @Override @@ -91,14 +106,42 @@ abstract class TransmitQueue { } @Override - TransmittedConnectionEntry transmit(final ConnectionEntry entry, final long now) { - final RequestEnvelope env = new RequestEnvelope(entry.getRequest().toVersion(backend.getVersion()), + Optional transmit(final ConnectionEntry entry, final long now) { + // If we're currently slicing a message we can't send any subsequent requests until slicing completes to + // avoid an out-of-sequence request envelope failure on the backend. In this case we return an empty + // Optional to indicate the request was not transmitted. + if (currentSlicedEnvSequenceId >= 0) { + return Optional.empty(); + } + + final Request request = entry.getRequest(); + final RequestEnvelope env = new RequestEnvelope(request.toVersion(backend.getVersion()), backend.getSessionId(), nextTxSequence++); - final TransmittedConnectionEntry ret = new TransmittedConnectionEntry(entry, env.getSessionId(), - env.getTxSequence(), now); - backend.getActor().tell(env, ActorRef.noSender()); - return ret; + if (request instanceof SliceableMessage) { + if (messageSlicer.slice(SliceOptions.builder().identifier(request.getTarget()) + .message(env).replyTo(request.getReplyTo()).sendTo(backend.getActor()) + .onFailureCallback(t -> env.sendFailure(new RuntimeRequestException( + "Failed to slice request " + request, t), 0L)).build())) { + // The request was sliced so record the envelope sequence id to prevent transmitting + // subsequent requests until slicing completes. + currentSlicedEnvSequenceId = env.getTxSequence(); + } + } else { + backend.getActor().tell(env, ActorRef.noSender()); + } + + return Optional.of(new TransmittedConnectionEntry(entry, env.getSessionId(), + env.getTxSequence(), now)); + } + + @Override + void preComplete(ResponseEnvelope envelope) { + if (envelope.getTxSequence() == currentSlicedEnvSequenceId) { + // Slicing completed for the prior request - clear the cached sequence id field to enable subsequent + // requests to be transmitted. + currentSlicedEnvSequenceId = NOT_SLICING; + } } } @@ -163,6 +206,8 @@ abstract class TransmitQueue { // If a matching request was found, this will track a task was closed. final Optional complete(final ResponseEnvelope envelope, final long now) { + preComplete(envelope); + Optional maybeEntry = findMatchingEntry(inflight, envelope); if (maybeEntry == null) { LOG.debug("Request for {} not found in inflight queue, checking pending queue", envelope); @@ -193,37 +238,54 @@ abstract class TransmitQueue { private void transmitEntries(final int maxTransmit, final long now) { for (int i = 0; i < maxTransmit; ++i) { final ConnectionEntry e = pending.poll(); - if (e == null) { + if (e == null || !transmitEntry(e, now)) { LOG.debug("Queue {} transmitted {} requests", this, i); return; } - - transmitEntry(e, now); } LOG.debug("Queue {} transmitted {} requests", this, maxTransmit); } - private void transmitEntry(final ConnectionEntry entry, final long now) { + private boolean transmitEntry(final ConnectionEntry entry, final long now) { LOG.debug("Queue {} transmitting entry {}", this, entry); // We are not thread-safe and are supposed to be externally-guarded, // hence send-before-record should be fine. // This needs to be revisited if the external guards are lowered. - inflight.addLast(transmit(entry, now)); + final Optional maybeTransmitted = transmit(entry, now); + if (!maybeTransmitted.isPresent()) { + return false; + } + + inflight.addLast(maybeTransmitted.get()); + return true; } - /** - * Enqueue an entry, possibly also transmitting it. - * - * @return Delay to be forced on the calling thread, in nanoseconds. - */ - final long enqueue(final ConnectionEntry entry, final long now) { + final long enqueueOrForward(final ConnectionEntry entry, final long now) { if (successor != null) { // This call will pay the enqueuing price, hence the caller does not have to successor.forwardEntry(entry, now); return 0; } + return enqueue(entry, now); + } + + final void enqueueOrReplay(final ConnectionEntry entry, final long now) { + if (successor != null) { + successor.replayEntry(entry, now); + } else { + enqueue(entry, now); + } + } + + /** + * Enqueue an entry, possibly also transmitting it. + * + * @return Delay to be forced on the calling thread, in nanoseconds. + */ + private long enqueue(final ConnectionEntry entry, final long now) { + // XXX: we should place a guard against incorrect entry sequences: // entry.getEnqueueTicks() should have non-negative difference from the last entry present in the queues @@ -242,7 +304,11 @@ abstract class TransmitQueue { } if (pending.isEmpty()) { - transmitEntry(entry, now); + if (!transmitEntry(entry, now)) { + LOG.debug("Queue {} cannot transmit request {} - delaying it", this, entry.getRequest()); + pending.addLast(entry); + } + return delay; } @@ -256,7 +322,9 @@ abstract class TransmitQueue { */ abstract int canTransmitCount(int inflightSize); - abstract TransmittedConnectionEntry transmit(ConnectionEntry entry, long now); + abstract Optional transmit(ConnectionEntry entry, long now); + + abstract void preComplete(ResponseEnvelope envelope); final boolean isEmpty() { return inflight.isEmpty() && pending.isEmpty(); @@ -271,9 +339,13 @@ abstract class TransmitQueue { return pending.peek(); } - final void poison(final RequestException cause) { - poisonQueue(inflight, cause); - poisonQueue(pending, cause); + final List poison() { + final List entries = new ArrayList<>(inflight.size() + pending.size()); + entries.addAll(inflight); + inflight.clear(); + entries.addAll(pending); + pending.clear(); + return entries; } final void setForwarder(final ReconnectForwarder forwarder, final long now) { @@ -377,13 +449,4 @@ abstract class TransmitQueue { return null; } - - private static void poisonQueue(final Queue queue, final RequestException cause) { - for (ConnectionEntry e : queue) { - final Request request = e.getRequest(); - LOG.trace("Poisoning request {}", request, cause); - e.complete(request.toRequestFailure(cause)); - } - queue.clear(); - } }