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=b7543410cd1a63128ac2da7ffde001cc9b3d778f;hpb=b74c6012092e47430a8f4d6f4ddeb1d3e2b1b7df;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 b7543410cd..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; @@ -54,28 +59,45 @@ import org.slf4j.LoggerFactory; @NotThreadSafe abstract class TransmitQueue { static final class Halted extends TransmitQueue { + // For ConnectingClientConnection. Halted(final int targetDepth) { super(targetDepth); } + // For ReconnectingClientConnection. + Halted(final TransmitQueue oldQueue, final long now) { + super(oldQueue, now); + } + @Override int canTransmitCount(final int inflightSize) { return 0; } @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; - Transmitting(final int targetDepth, final BackendInfo backend) { - super(targetDepth); + // For ConnectedClientConnection. + 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 @@ -84,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; + } } } @@ -99,13 +149,37 @@ abstract class TransmitQueue { private final Deque inflight = new ArrayDeque<>(); private final Deque pending = new ArrayDeque<>(); - private final ProgressTracker tracker; + private final AveragingProgressTracker tracker; // Cannot be just ProgressTracker as we are inheriting limits. private ReconnectForwarder successor; + /** + * Construct initial transmitting queue. + */ TransmitQueue(final int targetDepth) { tracker = new AveragingProgressTracker(targetDepth); } + /** + * Construct new transmitting queue while inheriting timing data from the previous transmit queue instance. + */ + TransmitQueue(final TransmitQueue oldQueue, final int targetDepth, final long now) { + tracker = new AveragingProgressTracker(oldQueue.tracker, targetDepth, now); + } + + /** + * Construct new transmitting queue while inheriting timing and size data from the previous transmit queue instance. + */ + TransmitQueue(final TransmitQueue oldQueue, final long now) { + tracker = new AveragingProgressTracker(oldQueue.tracker, now); + } + + /** + * Cancel the accumulated sum of delays as we expect the new backend to work now. + */ + void cancelDebt(final long now) { + tracker.cancelDebt(now); + } + /** * Drain the contents of the connection into a list. This will leave the queue empty and allow further entries * to be added to it during replay. When we set the successor all entries enqueued between when this methods @@ -132,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); @@ -162,23 +238,45 @@ 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) { - LOG.debug("Queue {} transmitting entry {}", entry); + 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; + } + + 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); + } } /** @@ -186,11 +284,7 @@ abstract class TransmitQueue { * * @return Delay to be forced on the calling thread, in nanoseconds. */ - final long enqueue(final ConnectionEntry entry, final long now) { - if (successor != null) { - successor.forwardEntry(entry, now); - return 0; - } + 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 @@ -210,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; } @@ -224,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(); @@ -239,13 +339,17 @@ 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) { - Verify.verify(successor == null, "Successor {} already set on connection {}", successor, this); + Verify.verify(successor == null, "Successor %s already set on connection %s", successor, this); successor = Preconditions.checkNotNull(forwarder); LOG.debug("Connection {} superseded by {}, splicing queue", this, successor); @@ -257,14 +361,14 @@ abstract class TransmitQueue { int count = 0; ConnectionEntry entry = inflight.poll(); while (entry != null) { - successor.forwardEntry(entry, now); + successor.replayEntry(entry, now); entry = inflight.poll(); count++; } entry = pending.poll(); while (entry != null) { - successor.forwardEntry(entry, now); + successor.replayEntry(entry, now); entry = pending.poll(); count++; } @@ -345,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(); - } }