svn commit: r644312 - /mina/trunk/core/src/main/java/org/apache/mina/common/AbstractPollingIoProcessor.java

[tr...@apache.org]

Author: trustin
Date: Thu Apr  3 06:36:54 2008
New Revision: 644312

URL: http://svn.apache.org/viewvc?rev=644312&view=rev
Log:
Fixed OOM in NIO transport due to cumulated interestOps requests

Modified:
    mina/trunk/core/src/main/java/org/apache/mina/common/AbstractPollingIoProcessor.java

Modified: mina/trunk/core/src/main/java/org/apache/mina/common/AbstractPollingIoProcessor.java
URL: http://svn.apache.org/viewvc/mina/trunk/core/src/main/java/org/apache/mina/common/AbstractPollingIoProcessor.java?rev=644312&r1=644311&r2=644312&view=diff
==============================================================================
--- mina/trunk/core/src/main/java/org/apache/mina/common/AbstractPollingIoProcessor.java (original)
+++ mina/trunk/core/src/main/java/org/apache/mina/common/AbstractPollingIoProcessor.java Thu Apr  3 06:36:54 2008
@@ -463,13 +463,13 @@
     }
 
     private void flush(long currentTime) {
-        for (; ;) {
-            T session = flushingSessions.poll();
-
-            if (session == null) {
-                break;
-            }
+        final T firstSession = flushingSessions.peek();
+        if (firstSession == null) {
+            return;
+        }
 
+        T session = flushingSessions.poll(); // the same one with firstSession
+        for (; ;) {
             session.setScheduledForFlush(false);
             SessionState state = state(session);
             switch (state) {
@@ -496,6 +496,12 @@
             default:
                 throw new IllegalStateException(String.valueOf(state));
             }
+
+            session = flushingSessions.peek();
+            if (session == null || session == firstSession) {
+                break;
+            }
+            session = flushingSessions.poll();
         }
     }
 
@@ -553,11 +559,17 @@
                     throw new IllegalStateException("Don't know how to handle message of type '" + message.getClass().getName() + "'.  Are you missing a protocol encoder?");
                 }
 
+                if (localWrittenBytes == 0) {
+                    // Kernel buffer is full.
+                    setInterestedInWrite(session, true);
+                    return false;
+                }
+
                 writtenBytes += localWrittenBytes;
 
-                if (localWrittenBytes == 0 || writtenBytes >= maxWrittenBytes) {
-                    // Kernel buffer is full or wrote too much.
-                    setInterestedInWrite(session, true);
+                if (writtenBytes >= maxWrittenBytes) {
+                    // Wrote too much
+                    scheduleFlush(session);
                     return false;
                 }
             } while (writtenBytes < maxWrittenBytes);