openflowjava/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/AbstractStackedOutboundQueue.java

   1 /*
   2  * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.openflowjava.protocol.impl.core.connection;
   9
  10 import static com.google.common.base.Preconditions.checkArgument;
  11 import static com.google.common.base.Verify.verify;
  12 import static java.util.Objects.requireNonNull;
  13
  14 import com.google.common.util.concurrent.FutureCallback;
  15 import io.netty.channel.Channel;
  16 import java.util.ArrayList;
  17 import java.util.Iterator;
  18 import java.util.List;
  19 import java.util.concurrent.atomic.AtomicLongFieldUpdater;
  20 import org.checkerframework.checker.lock.qual.GuardedBy;
  21 import org.checkerframework.checker.lock.qual.Holding;
  22 import org.eclipse.jdt.annotation.NonNull;
  23 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueue;
  24 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueueException;
  25 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
  26 import org.slf4j.Logger;
  27 import org.slf4j.LoggerFactory;
  28
  29 abstract class AbstractStackedOutboundQueue implements OutboundQueue {
  30
  31     private static final Logger LOG = LoggerFactory.getLogger(AbstractStackedOutboundQueue.class);
  32     protected static final AtomicLongFieldUpdater<AbstractStackedOutboundQueue> LAST_XID_OFFSET_UPDATER =
  33             AtomicLongFieldUpdater.newUpdater(AbstractStackedOutboundQueue.class, "lastXid");
  34
  35     @GuardedBy("unflushedSegments")
  36     protected volatile StackedSegment firstSegment;
  37     @GuardedBy("unflushedSegments")
  38     protected final List<StackedSegment> unflushedSegments = new ArrayList<>(2);
  39     @GuardedBy("unflushedSegments")
  40     protected final List<StackedSegment> uncompletedSegments = new ArrayList<>(2);
  41
  42     private volatile long lastXid = -1;
  43     private volatile long allocatedXid = -1;
  44
  45     @GuardedBy("unflushedSegments")
  46     protected Integer shutdownOffset;
  47
  48     // Accessed from Netty only
  49     protected int flushOffset;
  50
  51     protected final AbstractOutboundQueueManager<?, ?> manager;
  52
  53     AbstractStackedOutboundQueue(final AbstractOutboundQueueManager<?, ?> manager) {
  54         this.manager = requireNonNull(manager);
  55         firstSegment = StackedSegment.create(0L);
  56         uncompletedSegments.add(firstSegment);
  57         unflushedSegments.add(firstSegment);
  58     }
  59
  60     @Override
  61     public void commitEntry(final Long xid, final OfHeader message, final FutureCallback<OfHeader> callback) {
  62         commitEntry(xid, message, callback, OutboundQueueEntry.DEFAULT_IS_COMPLETE);
  63     }
  64
  65     @Holding("unflushedSegments")
  66     protected void ensureSegment(final StackedSegment first, final int offset) {
  67         final int segmentOffset = offset / StackedSegment.SEGMENT_SIZE;
  68         LOG.debug("Queue {} slow offset {} maps to {} segments {}", this, offset, segmentOffset,
  69                 unflushedSegments.size());
  70
  71         for (int i = unflushedSegments.size(); i <= segmentOffset; ++i) {
  72             final StackedSegment newSegment = StackedSegment.create(first.getBaseXid()
  73                     + StackedSegment.SEGMENT_SIZE * (long)i);
  74             LOG.debug("Adding segment {}", newSegment);
  75             unflushedSegments.add(newSegment);
  76         }
  77
  78         allocatedXid = unflushedSegments.get(unflushedSegments.size() - 1).getEndXid();
  79     }
  80
  81     /*
  82      * This method is expected to be called from multiple threads concurrently.
  83      */
  84     @Override
  85     public Long reserveEntry() {
  86         final long xid = LAST_XID_OFFSET_UPDATER.incrementAndGet(this);
  87         final StackedSegment fastSegment = firstSegment;
  88
  89         if (xid >= fastSegment.getBaseXid() + StackedSegment.SEGMENT_SIZE) {
  90             if (xid >= allocatedXid) {
  91                 // Multiple segments, this a slow path
  92                 LOG.debug("Queue {} falling back to slow reservation for XID {}", this, xid);
  93
  94                 synchronized (unflushedSegments) {
  95                     LOG.debug("Queue {} executing slow reservation for XID {}", this, xid);
  96
  97                     // Shutdown was scheduled, need to fail the reservation
  98                     if (shutdownOffset != null) {
  99                         LOG.debug("Queue {} is being shutdown, failing reservation", this);
 100                         return null;
 101                     }
 102
 103                     // Ensure we have the appropriate segment for the specified XID
 104                     final StackedSegment slowSegment = firstSegment;
 105                     final int slowOffset = (int) (xid - slowSegment.getBaseXid());
 106                     verify(slowOffset >= 0);
 107
 108                     // Now, we let's see if we need to allocate a new segment
 109                     ensureSegment(slowSegment, slowOffset);
 110
 111                     LOG.debug("Queue {} slow reservation finished", this);
 112                 }
 113             } else {
 114                 LOG.debug("Queue {} XID {} is already backed", this, xid);
 115             }
 116         }
 117
 118         LOG.trace("Queue {} allocated XID {}", this, xid);
 119         return xid;
 120     }
 121
 122     /**
 123      * Write some entries from the queue to the channel. Guaranteed to run
 124      * in the corresponding EventLoop.
 125      *
 126      * @param channel Channel onto which we are writing
 127      * @param now time stamp
 128      * @return Number of entries written out
 129      */
 130     int writeEntries(@NonNull final Channel channel, final long now) {
 131         // Local cache
 132         StackedSegment segment = firstSegment;
 133         int entries = 0;
 134
 135         while (channel.isWritable()) {
 136             final OutboundQueueEntry entry = segment.getEntry(flushOffset);
 137             if (!entry.isCommitted()) {
 138                 LOG.debug("Queue {} XID {} segment {} offset {} not committed yet", this, segment.getBaseXid()
 139                         + flushOffset, segment, flushOffset);
 140                 break;
 141             }
 142
 143             LOG.trace("Queue {} flushing entry at offset {}", this, flushOffset);
 144             final OfHeader message = entry.takeMessage();
 145             flushOffset++;
 146             entries++;
 147
 148             if (message != null) {
 149                 manager.writeMessage(message, now);
 150             } else {
 151                 entry.complete(null);
 152             }
 153
 154             if (flushOffset >= StackedSegment.SEGMENT_SIZE) {
 155                 /*
 156                  * Slow path: purge the current segment unless it's the last one.
 157                  * If it is, we leave it for replacement when a new reservation
 158                  * is run on it.
 159                  *
 160                  * This costs us two slow paths, but hey, this should be very rare,
 161                  * so let's keep things simple.
 162                  */
 163                 synchronized (unflushedSegments) {
 164                     LOG.debug("Flush offset {} unflushed segments {}", flushOffset, unflushedSegments.size());
 165
 166                     // We may have raced ahead of reservation code and need to allocate a segment
 167                     ensureSegment(segment, flushOffset);
 168
 169                     // Remove the segment, update the firstSegment and reset flushOffset
 170                     final StackedSegment oldSegment = unflushedSegments.remove(0);
 171                     if (oldSegment.isComplete()) {
 172                         uncompletedSegments.remove(oldSegment);
 173                         oldSegment.recycle();
 174                     }
 175
 176                     // Reset the first segment and add it to the uncompleted list
 177                     segment = unflushedSegments.get(0);
 178                     uncompletedSegments.add(segment);
 179
 180                     // Update the shutdown offset
 181                     if (shutdownOffset != null) {
 182                         shutdownOffset -= StackedSegment.SEGMENT_SIZE;
 183                     }
 184
 185                     // Allow reservations back on the fast path by publishing the new first segment
 186                     firstSegment = segment;
 187
 188                     flushOffset = 0;
 189                     LOG.debug("Queue {} flush moved to segment {}", this, segment);
 190                 }
 191             }
 192         }
 193
 194         return entries;
 195     }
 196
 197     boolean pairRequest(final OfHeader message) {
 198         Iterator<StackedSegment> it = uncompletedSegments.iterator();
 199         while (it.hasNext()) {
 200             final StackedSegment queue = it.next();
 201             final OutboundQueueEntry entry = queue.pairRequest(message);
 202             if (entry == null) {
 203                 continue;
 204             }
 205
 206             LOG.trace("Queue {} accepted response {}", queue, message);
 207
 208             // This has been a barrier request, we need to flush all
 209             // previous queues
 210             if (entry.isBarrier() && uncompletedSegments.size() > 1) {
 211                 LOG.trace("Queue {} indicated request was a barrier", queue);
 212
 213                 it = uncompletedSegments.iterator();
 214                 while (it.hasNext()) {
 215                     final StackedSegment q = it.next();
 216
 217                     // We want to complete all queues before the current one, we will
 218                     // complete the current queue below
 219                     if (!queue.equals(q)) {
 220                         LOG.trace("Queue {} is implied finished", q);
 221                         q.completeAll();
 222                         it.remove();
 223                         q.recycle();
 224                     } else {
 225                         break;
 226                     }
 227                 }
 228             }
 229
 230             if (queue.isComplete()) {
 231                 LOG.trace("Queue {} is finished", queue);
 232                 it.remove();
 233                 queue.recycle();
 234             }
 235
 236             return true;
 237         }
 238
 239         LOG.debug("Failed to find completion for message {}", message);
 240         return false;
 241     }
 242
 243     boolean needsFlush() {
 244         // flushOffset always points to the first entry, which can be changed only
 245         // from Netty, so we are fine here.
 246         if (firstSegment.getBaseXid() + flushOffset > lastXid) {
 247             return false;
 248         }
 249
 250         if (shutdownOffset != null && flushOffset >= shutdownOffset) {
 251             return false;
 252         }
 253
 254         return firstSegment.getEntry(flushOffset).isCommitted();
 255     }
 256
 257     long startShutdown() {
 258         /*
 259          * We are dealing with a multi-threaded shutdown, as the user may still
 260          * be reserving entries in the queue. We are executing in a netty thread,
 261          * so neither flush nor barrier can be running, which is good news.
 262          * We will eat up all the slots in the queue here and mark the offset first
 263          * reserved offset and free up all the cached queues. We then schedule
 264          * the flush task, which will deal with the rest of the shutdown process.
 265          */
 266         synchronized (unflushedSegments) {
 267             // Increment the offset by the segment size, preventing fast path allocations,
 268             // since we are holding the slow path lock, any reservations will see the queue
 269             // in shutdown and fail accordingly.
 270             final long xid = LAST_XID_OFFSET_UPDATER.addAndGet(this, StackedSegment.SEGMENT_SIZE);
 271             shutdownOffset = (int) (xid - firstSegment.getBaseXid() - StackedSegment.SEGMENT_SIZE);
 272
 273             // Fails all uncompleted entries, because they will never be completed due to disconnected channel.
 274             return lockedFailSegments(uncompletedSegments.iterator());
 275         }
 276     }
 277
 278     /**
 279      * Checks if the shutdown is in final phase -> all allowed entries (number of entries < shutdownOffset) are flushed
 280      * and fails all not completed entries (if in final phase).
 281      *
 282      * @param channel netty channel
 283      * @return true if in final phase, false if a flush is needed
 284      */
 285     boolean finishShutdown(final Channel channel) {
 286         boolean needsFlush;
 287         synchronized (unflushedSegments) {
 288             // Fails all entries, that were flushed in shutdownOffset (became uncompleted)
 289             // - they will never be completed due to disconnected channel.
 290             lockedFailSegments(uncompletedSegments.iterator());
 291             // If no further flush is needed or we are not able to write to channel anymore, then we fail all unflushed
 292             // segments, so that each enqueued entry is reported as unsuccessful due to channel disconnection.
 293             // No further entries should be enqueued by this time.
 294             needsFlush = channel.isWritable() && needsFlush();
 295             if (!needsFlush) {
 296                 lockedFailSegments(unflushedSegments.iterator());
 297             }
 298         }
 299         return !needsFlush;
 300     }
 301
 302     protected OutboundQueueEntry getEntry(final Long xid) {
 303         final StackedSegment fastSegment = firstSegment;
 304         final long calcOffset = xid - fastSegment.getBaseXid();
 305         checkArgument(calcOffset >= 0, "Commit of XID %s does not match up with base XID %s",
 306                 xid, fastSegment.getBaseXid());
 307
 308         verify(calcOffset <= Integer.MAX_VALUE);
 309         final int fastOffset = (int) calcOffset;
 310
 311         if (fastOffset >= StackedSegment.SEGMENT_SIZE) {
 312             LOG.debug("Queue {} falling back to slow commit of XID {} at offset {}", this, xid, fastOffset);
 313
 314             final StackedSegment segment;
 315             final int slowOffset;
 316             synchronized (unflushedSegments) {
 317                 final StackedSegment slowSegment = firstSegment;
 318                 final long slowCalcOffset = xid - slowSegment.getBaseXid();
 319                 verify(slowCalcOffset >= 0 && slowCalcOffset <= Integer.MAX_VALUE);
 320                 slowOffset = (int) slowCalcOffset;
 321
 322                 LOG.debug("Queue {} recalculated offset of XID {} to {}", this, xid, slowOffset);
 323                 segment = unflushedSegments.get(slowOffset / StackedSegment.SEGMENT_SIZE);
 324             }
 325
 326             final int segOffset = slowOffset % StackedSegment.SEGMENT_SIZE;
 327             LOG.debug("Queue {} slow commit of XID {} completed at offset {} (segment {} offset {})", this,
 328                     xid, slowOffset, segment, segOffset);
 329             return segment.getEntry(segOffset);
 330         }
 331         return fastSegment.getEntry(fastOffset);
 332     }
 333
 334     /**
 335      * Fails not completed entries in segments and frees completed segments.
 336      *
 337      * @param iterator list of segments to be failed
 338      * @return number of failed entries
 339      */
 340     @GuardedBy("unflushedSegments")
 341     private long lockedFailSegments(Iterator<StackedSegment> iterator) {
 342         long entries = 0;
 343
 344         // Fail all queues
 345         while (iterator.hasNext()) {
 346             final StackedSegment segment = iterator.next();
 347
 348             entries += segment.failAll(OutboundQueueException.DEVICE_DISCONNECTED);
 349             if (segment.isComplete()) {
 350                 LOG.trace("Cleared segment {}", segment);
 351                 iterator.remove();
 352             }
 353         }
 354
 355         return entries;
 356     }
 357 }