openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/StackedOutboundQueueNoBarrier.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
   9 package org.opendaylight.openflowjava.protocol.impl.core.connection;
  10
  11 import com.google.common.util.concurrent.FutureCallback;
  12 import io.netty.channel.Channel;
  13 import javax.annotation.Nonnull;
  14 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowModInput;
  15 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
  16 import org.slf4j.Logger;
  17 import org.slf4j.LoggerFactory;
  18
  19 /**
  20  * Class is designed for stacking Statistics and propagate immediate response for all
  21  * another requests.
  22  */
  23 public class StackedOutboundQueueNoBarrier extends AbstractStackedOutboundQueue {
  24
  25     private static final Logger LOG = LoggerFactory.getLogger(StackedOutboundQueueNoBarrier.class);
  26
  27     StackedOutboundQueueNoBarrier(final AbstractOutboundQueueManager<?, ?> manager) {
  28         super(manager);
  29     }
  30
  31     /*
  32      * This method is expected to be called from multiple threads concurrently
  33      */
  34     @Override
  35     public void commitEntry(final Long xid, final OfHeader message, final FutureCallback<OfHeader> callback) {
  36         final OutboundQueueEntry entry = getEntry(xid);
  37
  38         if (message instanceof FlowModInput) {
  39             callback.onSuccess(null);
  40             entry.commit(message, null);
  41         } else {
  42             entry.commit(message, callback);
  43         }
  44
  45         LOG.trace("Queue {} committed XID {}", this, xid);
  46         manager.ensureFlushing();
  47     }
  48
  49     @Override
  50     int writeEntries(@Nonnull final Channel channel, final long now) {
  51         // Local cache
  52         StackedSegment segment = firstSegment;
  53         int entries = 0;
  54
  55         while (channel.isWritable()) {
  56             final OutboundQueueEntry entry = segment.getEntry(flushOffset);
  57             if (!entry.isCommitted()) {
  58                 LOG.debug("Queue {} XID {} segment {} offset {} not committed yet", this, segment.getBaseXid()
  59                         + flushOffset, segment, flushOffset);
  60                 break;
  61             }
  62
  63             LOG.trace("Queue {} flushing entry at offset {}", this, flushOffset);
  64             final OfHeader message = entry.takeMessage();
  65             flushOffset++;
  66             entries++;
  67
  68             if (message != null) {
  69                 manager.writeMessage(message, now);
  70             } else {
  71                 entry.complete(null);
  72             }
  73
  74             if (flushOffset >= StackedSegment.SEGMENT_SIZE) {
  75                 /*
  76                  * Slow path: purge the current segment unless it's the last one.
  77                  * If it is, we leave it for replacement when a new reservation
  78                  * is run on it.
  79                  * This costs us two slow paths, but hey, this should be very rare,
  80                  * so let's keep things simple.
  81                  */
  82                 synchronized (unflushedSegments) {
  83                     LOG.debug("Flush offset {} unflushed segments {}", flushOffset, unflushedSegments.size());
  84
  85                     // We may have raced ahead of reservation code and need to allocate a segment
  86                     ensureSegment(segment, flushOffset);
  87
  88                     // Remove the segment, update the firstSegment and reset flushOffset
  89                     final StackedSegment oldSegment = unflushedSegments.remove(0);
  90                     oldSegment.completeAll();
  91                     uncompletedSegments.remove(oldSegment);
  92                     oldSegment.recycle();
  93
  94                     // Reset the first segment and add it to the uncompleted list
  95                     segment = unflushedSegments.get(0);
  96                     uncompletedSegments.add(segment);
  97
  98                     // Update the shutdown offset
  99                     if (shutdownOffset != null) {
 100                         shutdownOffset -= StackedSegment.SEGMENT_SIZE;
 101                     }
 102
 103                     // Allow reservations back on the fast path by publishing the new first segment
 104                     firstSegment = segment;
 105
 106                     flushOffset = 0;
 107                     LOG.debug("Queue {} flush moved to segment {}", this, segment);
 108                 }
 109             }
 110         }
 111
 112         return entries;
 113     }
 114 }