[jboss-cvs] JBoss Messaging SVN: r5165 - branches/Branch_Chunk_Clebert/src/main/org/jboss/messaging/core/persistence/impl/journal.

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

Author: clebert.suconic at jboss.com
Date: 2008-10-21 18:38:02 -0400 (Tue, 21 Oct 2008)
New Revision: 5165

Modified:
   branches/Branch_Chunk_Clebert/src/main/org/jboss/messaging/core/persistence/impl/journal/JournalServerLargeMessageImpl.java
   branches/Branch_Chunk_Clebert/src/main/org/jboss/messaging/core/persistence/impl/journal/JournalStorageManager.java
Log:
Deleting files from an executor instead

Modified: branches/Branch_Chunk_Clebert/src/main/org/jboss/messaging/core/persistence/impl/journal/JournalServerLargeMessageImpl.java
===================================================================
--- branches/Branch_Chunk_Clebert/src/main/org/jboss/messaging/core/persistence/impl/journal/JournalServerLargeMessageImpl.java	2008-10-21 17:07:40 UTC (rev 5164)
+++ branches/Branch_Chunk_Clebert/src/main/org/jboss/messaging/core/persistence/impl/journal/JournalServerLargeMessageImpl.java	2008-10-21 22:38:02 UTC (rev 5165)
@@ -69,7 +69,7 @@
     */
    public synchronized void addBytes(final byte[] bytes) throws Exception
    {
-      testFile();
+      validateFile();
 
       if (!file.isOpen())
       {
@@ -85,7 +85,7 @@
    @Override
    public synchronized void encodeBody(final MessagingBuffer bufferOut, final int start, final int size)
    {
-      testFile();
+      validateFile();
 
       try
       {
@@ -118,7 +118,7 @@
    @Override
    public synchronized int getBodySize()
    {
-      testFile();
+      validateFile();
 
       try
       {
@@ -178,17 +178,7 @@
 
    public void deleteFile() throws MessagingException
    {
-
-      // TODO: This should use an executor somewhere...
-      // We can't take the risk of blocking the queue when lots of ServerLargeMessages are being deleted
-      try
-      {
-         file.delete();
-      }
-      catch (Exception e)
-      {
-         throw new MessagingException(MessagingException.INTERNAL_ERROR, e.getMessage(), e);
-      }
+      this.storageManager.deleteFile(file);
    }
 
    @Override
@@ -210,7 +200,9 @@
 
    // Protected -----------------------------------------------------
 
-   protected void testFile()
+   // Private -------------------------------------------------------
+
+   private void validateFile()
    {
       if (file == null)
       {
@@ -223,7 +215,6 @@
       }
    }
 
-   // Private -------------------------------------------------------
 
    // Inner classes -------------------------------------------------
 

Modified: branches/Branch_Chunk_Clebert/src/main/org/jboss/messaging/core/persistence/impl/journal/JournalStorageManager.java
===================================================================
--- branches/Branch_Chunk_Clebert/src/main/org/jboss/messaging/core/persistence/impl/journal/JournalStorageManager.java	2008-10-21 17:07:40 UTC (rev 5164)
+++ branches/Branch_Chunk_Clebert/src/main/org/jboss/messaging/core/persistence/impl/journal/JournalStorageManager.java	2008-10-21 22:38:02 UTC (rev 5165)
@@ -34,6 +34,9 @@
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 
 import javax.transaction.xa.Xid;
@@ -75,6 +78,7 @@
 import org.jboss.messaging.core.transaction.Transaction;
 import org.jboss.messaging.core.transaction.impl.TransactionImpl;
 import org.jboss.messaging.util.IDGenerator;
+import org.jboss.messaging.util.JBMThreadFactory;
 import org.jboss.messaging.util.SimpleString;
 import org.jboss.messaging.util.TimeAndCounterIDGenerator;
 
@@ -130,9 +134,14 @@
    private final ConcurrentMap<SimpleString, Long> destinationIDMap = new ConcurrentHashMap<SimpleString, Long>();
 
    private volatile boolean started;
-
+   
+   private final ExecutorService executor;
+   
+   
    public JournalStorageManager(final Configuration config)
    {
+      this.executor = Executors.newCachedThreadPool(new JBMThreadFactory("JBM-journal-storage-manager"));
+      
       if (config.getJournalType() != JournalType.NIO && config.getJournalType() != JournalType.ASYNCIO)
       {
          throw new IllegalArgumentException("Only NIO and AsyncIO are supported journals");
@@ -206,6 +215,7 @@
                                 final Journal bindingsJournal,
                                 final SequentialFileFactory largeMessagesFactory)
    {
+      this.executor = Executors.newCachedThreadPool(new JBMThreadFactory("JBM-journal-storage-manager"));
       this.messageJournal = messageJournal;
       this.bindingsJournal = bindingsJournal;
       this.largeMessagesFactory = largeMessagesFactory;
@@ -683,11 +693,15 @@
       {
          return;
       }
+      
+      executor.shutdown();
 
       bindingsJournal.stop();
 
       messageJournal.stop();
 
+      executor.awaitTermination(60, TimeUnit.SECONDS);
+
       started = false;
    }
 
@@ -709,7 +723,28 @@
    }
 
    // Package protected ---------------------------------------------
+   
+   // This should be accessed from this package only
+   void deleteFile(final SequentialFile file)
+   {
+      this.executor.execute(new Runnable() {
 
+         public void run()
+         {
+            try
+            {
+               System.out.println("Deleting file " + file);
+               file.delete();
+            }
+            catch (Exception e)
+            {
+               log.warn(e.getMessage(), e);
+            }
+         }
+         
+      });
+   }
+
    /**
     * @param messageID
     * @return