Bug 6646 Fix infinite reschedule of flush

 - sometimes (on disconnect) there can be still some unflushed segments
   but they are not able to be flushed if channel is not writable anymore
   and we get to infinite loop of flushing (but not writing)

Change-Id: I74cac21b4635e22f5b8d63f3a602f40796108059
Signed-off-by: Andrej Leitner <andrej.leitner@pantheon.tech>

diff --git a/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/AbstractOutboundQueueManager.java b/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/AbstractOutboundQueueManager.java
index 34df0170b4f34692f9225f757585d0ffbf846613..bec266bb5b5abcf727fc4eb0b3a9ab2cec9bf98a 100644 (file)
--- a/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/AbstractOutboundQueueManager.java
+++ b/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/AbstractOutboundQueueManager.java
@@ -173,7 +173,7 @@ abstract class AbstractOutboundQueueManager<T extends OutboundQueueHandler, O ex
         // Then we start queue shutdown, start counting written messages (so that we don't keep sending messages
         // indefinitely) and failing not completed entries.
         shuttingDown = true;
-        final long entries = currentQueue.startShutdown(ctx.channel());
+        final long entries = currentQueue.startShutdown();
         LOG.debug("Cleared {} queue entries from channel {}", entries, ctx.channel());
 
         // Finally, we schedule flush task that will take care of unflushed entries. We also cover the case,
@@ -301,7 +301,7 @@ abstract class AbstractOutboundQueueManager<T extends OutboundQueueHandler, O ex
             rescheduleFlush();
         } else {
             close();
-            if (currentQueue.finishShutdown()) {
+            if (currentQueue.finishShutdown(parent.getChannel())) {
                LOG.debug("Channel {} shutdown complete", parent.getChannel());
             } else {
                LOG.trace("Channel {} current queue not completely flushed yet", parent.getChannel());

diff --git a/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/AbstractStackedOutboundQueue.java b/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/AbstractStackedOutboundQueue.java
index 759633354024f573697de62308fd944715dfbc76..16106a1a4ee8a17ff5621e02ebe40b7e5f2263da 100644 (file)
--- a/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/AbstractStackedOutboundQueue.java
+++ b/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/AbstractStackedOutboundQueue.java
@@ -244,7 +244,7 @@ abstract class AbstractStackedOutboundQueue implements OutboundQueue {
         return firstSegment.getEntry(flushOffset).isCommitted();
     }
 
-    long startShutdown(final Channel channel) {
+    long startShutdown() {
         /*
          * We are dealing with a multi-threaded shutdown, as the user may still
          * be reserving entries in the queue. We are executing in a netty thread,
@@ -268,18 +268,19 @@ abstract class AbstractStackedOutboundQueue implements OutboundQueue {
     /**
      * Checks if the shutdown is in final phase -> all allowed entries (number of entries < shutdownOffset) are flushed
      * and fails all not completed entries (if in final phase)
+     * @param channel netty channel
      * @return true if in final phase, false if a flush is needed
      */
-    boolean finishShutdown() {
+    boolean finishShutdown(final Channel channel) {
         boolean needsFlush;
         synchronized (unflushedSegments) {
             // Fails all entries, that were flushed in shutdownOffset (became uncompleted)
             // - they will never be completed due to disconnected channel.
             lockedFailSegments(uncompletedSegments.iterator());
-            // If no further flush is needed, than we fail all unflushed segments, so that each enqueued entry
-            // is reported as unsuccessful due to channel disconnection. No further entries should be enqueued
-            // by this time.
-            needsFlush = needsFlush();
+            // If no further flush is needed or we are not able to write to channel anymore, then we fail all unflushed
+            // segments, so that each enqueued entry is reported as unsuccessful due to channel disconnection.
+            // No further entries should be enqueued by this time.
+            needsFlush = channel.isWritable() && needsFlush();
             if (!needsFlush) {
                 lockedFailSegments(unflushedSegments.iterator());
             }