BUG-3219: do to park thread in flush

If we encounter an entry which has been reserved but not committed
during flush operation, do not wait for it to get committed, but bail
immediately.

Change-Id: I04a3ca1de4447bba9b513d85601c83fd31d115a9
Signed-off-by: Robert Varga <rovarga@cisco.com>

diff --git a/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/OutboundQueueImpl.java b/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/OutboundQueueImpl.java
index 5d22435dddf8aaa3f410e22c53c148c9c58bf1f9..f133082ff76e41ff8455dcc5a607ef4b7bf7c15b 100644 (file)
--- a/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/OutboundQueueImpl.java
+++ b/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/OutboundQueueImpl.java
@@ -10,7 +10,6 @@ package org.opendaylight.openflowjava.protocol.impl.core.connection;
 import com.google.common.base.Preconditions;
 import com.google.common.util.concurrent.FutureCallback;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
-import java.util.concurrent.locks.LockSupport;
 import javax.annotation.Nonnull;
 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueue;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
@@ -23,7 +22,6 @@ final class OutboundQueueImpl implements OutboundQueue {
             AtomicIntegerFieldUpdater.newUpdater(OutboundQueueImpl.class, "reserveOffset");
     private static final AtomicIntegerFieldUpdater<OutboundQueueImpl> BARRIER_OFFSET_UPDATER =
             AtomicIntegerFieldUpdater.newUpdater(OutboundQueueImpl.class, "barrierOffset");
-    private static final long FLUSH_RETRY_NANOS = 1L;
     private final OutboundQueueManager<?> manager;
     private final OutboundQueueEntry[] queue;
     private final long baseXid;
@@ -181,26 +179,23 @@ final class OutboundQueueImpl implements OutboundQueue {
         for (;;) {
             // No message ready
             if (isEmpty()) {
-                LOG.debug("Flush offset {} is uptodate with reserved", flushOffset);
+                LOG.trace("Flushed all reserved entries up to ", flushOffset);
                 return null;
             }
 
-            boolean retry = true;
-            while (!queue[flushOffset].isCommitted()) {
-                if (!retry) {
-                    LOG.debug("Offset {} not ready yet, giving up", flushOffset);
-                    return null;
-                }
-
-                LOG.debug("Offset {} not ready yet, retrying", flushOffset);
-                LockSupport.parkNanos(FLUSH_RETRY_NANOS);
-                retry = false;
+            final OutboundQueueEntry entry = queue[flushOffset];
+            if (!entry.isCommitted()) {
+                LOG.trace("Request at offset {} not ready yet, giving up", flushOffset);
+                return null;
             }
 
-            final OfHeader msg = queue[flushOffset++].getMessage();
+            final OfHeader msg = entry.getMessage();
+            flushOffset++;
             if (msg != null) {
                 return msg;
             }
+
+            LOG.trace("Null message, skipping to offset {}", flushOffset);
         }
     }