[jboss-cvs] JBoss Messaging SVN: r7104 - branches/Branch_JBM2_Perf_Clebert/src/main/org/jboss/messaging/core/asyncio/impl.

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

Author: clebert.suconic at jboss.com
Date: 2009-05-28 11:48:51 -0400 (Thu, 28 May 2009)
New Revision: 7104

Modified:
   branches/Branch_JBM2_Perf_Clebert/src/main/org/jboss/messaging/core/asyncio/impl/AsynchronousFileImpl.java
Log:
Improvements on semaphore on writing stuff

Modified: branches/Branch_JBM2_Perf_Clebert/src/main/org/jboss/messaging/core/asyncio/impl/AsynchronousFileImpl.java
===================================================================
--- branches/Branch_JBM2_Perf_Clebert/src/main/org/jboss/messaging/core/asyncio/impl/AsynchronousFileImpl.java	2009-05-28 15:47:06 UTC (rev 7103)
+++ branches/Branch_JBM2_Perf_Clebert/src/main/org/jboss/messaging/core/asyncio/impl/AsynchronousFileImpl.java	2009-05-28 15:48:51 UTC (rev 7104)
@@ -138,6 +138,8 @@
    private int maxIO;
 
    private final Lock writeLock = new ReentrantReadWriteLock().writeLock();
+   
+   private final VariableLatch pendingWrites = new VariableLatch();
 
    private Semaphore writeSemaphore;
 
@@ -223,11 +225,16 @@
       try
       {
 
+         while (!pendingWrites.waitCompletion(60000))
+         {
+            log.warn("Couldn't get lock after 60 seconds on closing AsynchronousFileImpl::" + this.fileName);
+         }
+         
          while (!writeSemaphore.tryAcquire(maxIO, 60, TimeUnit.SECONDS))
          {
-            log.warn("Couldn't acquire lock after 60 seconds on AIO",
-                     new Exception("Warning: Couldn't acquire lock after 60 seconds on AIO"));
+            log.warn("Couldn't get lock after 60 seconds on closing AsynchronousFileImpl::" + this.fileName);
          }
+
          writeSemaphore = null;
          if (poller != null)
          {
@@ -263,7 +270,8 @@
       {
          startPoller();
       }
-      writeSemaphore.acquireUninterruptibly();
+      
+      pendingWrites.up();
 
       if (writeExecutor != null)
       {
@@ -271,6 +279,8 @@
          {
             public void run()
             {
+               writeSemaphore.acquireUninterruptibly();
+
                try
                {
                   write(handler, position, size, directByteBuffer, aioCallback);
@@ -288,6 +298,8 @@
       }
       else
       {
+         writeSemaphore.acquireUninterruptibly();
+
          try
          {
             write(handler, position, size, directByteBuffer, aioCallback);
@@ -314,6 +326,7 @@
       {
          startPoller();
       }
+      pendingWrites.up();
       writeSemaphore.acquireUninterruptibly();
       try
       {
@@ -323,12 +336,14 @@
       {
          // Release only if an exception happened
          writeSemaphore.release();
+         pendingWrites.down();
          throw e;
       }
       catch (RuntimeException e)
       {
          // Release only if an exception happened
          writeSemaphore.release();
+         pendingWrites.down();
          throw e;
       }
    }
@@ -399,6 +414,7 @@
    private void callbackDone(final AIOCallback callback, final ByteBuffer buffer)
    {
       writeSemaphore.release();
+      pendingWrites.down();
       callback.done();
       if (bufferCallback != null)
       {
@@ -412,6 +428,7 @@
    {
       log.warn("CallbackError: " + errorMessage);
       writeSemaphore.release();
+      pendingWrites.down();
       callback.onError(errorCode, errorMessage);
    }