[jboss-cvs] JBoss Messaging SVN: r4245 - trunk/src/main/org/jboss/messaging/core/remoting/impl/mina.

[jboss-cvs-commits at lists.jboss.org]

Author: trustin
Date: 2008-05-19 20:31:41 -0400 (Mon, 19 May 2008)
New Revision: 4245

Modified:
   trunk/src/main/org/jboss/messaging/core/remoting/impl/mina/MinaHandler.java
Log:
Fix for dead lock while limiting write queue size in MinaHandler - used traffic throttling when required.

Modified: trunk/src/main/org/jboss/messaging/core/remoting/impl/mina/MinaHandler.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/mina/MinaHandler.java	2008-05-20 00:17:53 UTC (rev 4244)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/mina/MinaHandler.java	2008-05-20 00:31:41 UTC (rev 4245)
@@ -193,6 +193,11 @@
          {
             session.setAttribute(BLOCKED, Boolean.FALSE);
 
+            // Resume read operation if it's suspended.
+            if (!session.getTrafficMask().isReadable()) {
+               session.resumeRead();
+            }
+
             //Note that we need to notify all since there may be more than one thread waiting on this
             //E.g. the response from a blocking acknowledge and a delivery
             notifyAll();
@@ -245,19 +250,38 @@
    {
       PacketReturner returner;
 
+      final Thread ioThread = Thread.currentThread();
+
       if (packet.getResponseTargetID() != Packet.NO_ID_SET)
       {
          returner = new PacketReturner()
          {
             public void send(Packet p) throws Exception
             {
-               try
+               // Make sure we don't block the I/O thread.
+               if (Thread.currentThread() != ioThread)
                {
-                  checkWrite(session);
+                  try
+                  {
+                     checkWrite(session);
+                  }
+                  catch (Exception e)
+                  {
+                     log.error("Failed to acquire sem", e);
+                  }
                }
-               catch (Exception e)
+               else
                {
-                  log.error("Failed to acquire sem", e);
+                  // If we are running this in the I/O thread, we should not block, because
+                  // blocking will prevent the decrement of the scheduledWriteBytes() property.
+                  // Instead, we should disable the read operation temporarily so we don't
+                  // need to handle incoming messages which causes write queue overflow.
+
+                  if (session.getScheduledWriteBytes() >= bytesHigh)
+                  {
+                     // Will resume on messageSent()
+                     session.suspendRead();
+                  }
                }
 
                dispatcher.callFilters(p);