[jboss-cvs] JBoss Messaging SVN: r7457 - branches/clebert_temp_expirement/src/main/org/jboss/messaging/core/journal/impl.

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

Author: clebert.suconic at jboss.com
Date: 2009-06-24 16:20:35 -0400 (Wed, 24 Jun 2009)
New Revision: 7457

Modified:
   branches/clebert_temp_expirement/src/main/org/jboss/messaging/core/journal/impl/JournalImpl.java
Log:
Fixes

Modified: branches/clebert_temp_expirement/src/main/org/jboss/messaging/core/journal/impl/JournalImpl.java
===================================================================
--- branches/clebert_temp_expirement/src/main/org/jboss/messaging/core/journal/impl/JournalImpl.java	2009-06-24 17:11:46 UTC (rev 7456)
+++ branches/clebert_temp_expirement/src/main/org/jboss/messaging/core/journal/impl/JournalImpl.java	2009-06-24 20:20:35 UTC (rev 7457)
@@ -732,11 +732,6 @@
 
       JournalTransaction tx = transactionInfos.remove(txID);
 
-      if (sync)
-      {
-         tx.syncPreviousFiles();
-      }
-
       readLockCompact.lock();
 
       try
@@ -2132,14 +2127,7 @@
    {
       // (I) First we get the summary of what we really have on the files now:
 
-      // FileID, NumberOfElements
-      Map<Integer, AtomicInteger> currentSummary = journalTransaction.getElementsSummary();
-      
-      AtomicInteger value = currentSummary.get(currentFile.getFileID());
-
-      Integer intValue = value == null ? 0 : value.intValue();
-
-      return intValue == numberOfRecords;
+      return journalTransaction.getCounter(currentFile) == numberOfRecords;
    }
 
    /**
@@ -2382,23 +2370,20 @@
 
             if (sync)
             {
-               // We already did sync previous files outside of the lock,
-               // but in a very rare occasion (maybe in a low speed disk)
-               // you could have a race where the currentFile changed between the last sync to the time the lock was
-               // acquired.
-               // So, we call the syncPreviousFiles again to guarantee data on disk.
-               // Even if there is data to be synced, this should be very fast since previous files were already
-               // scheduled to be closed.
-               // This is just verifying if previous files are already closed
+               // 99 % of the times this will be already synced, as previous files should be closed already.
+               // This is to have 100% guarantee the transaction will be persisted and no loss of information would happen
                tx.syncPreviousFiles();
             }
 
             if (completeTransaction)
             {
+               // Filling the number of records at the current file
                tx.fillNumberOfRecords(currentFile, bb);
             }
          }
 
+         
+         // Adding fileID
          bb.writerIndex(SIZE_BYTE);
 
          bb.writeInt(currentFile.getFileID());
@@ -2779,18 +2764,32 @@
 
       private TransactionCallback currentCallback;
 
-      // Map of file id to number of elements participating on the transaction
-      // in that file
-      // Used to verify completion on reload
-      private final Map<Integer, AtomicInteger> numberOfElementsPerFile = new HashMap<Integer, AtomicInteger>();
-
       private Map<JournalFile, TransactionCallback> callbackList;
+      
+      private JournalFile lastFile = null;
+      
+      private final AtomicInteger counter = new AtomicInteger();
 
-      public Map<Integer, AtomicInteger> getElementsSummary()
+      private AtomicInteger internalgetCounter(final JournalFile file)
       {
-         return numberOfElementsPerFile;
+         if (lastFile != file)
+         {
+            lastFile = file;
+            counter.set(0);
+         }
+         return counter;
       }
+      
+      public int getCounter(final JournalFile file)
+      {
+         return internalgetCounter(file).intValue();
+      }
 
+      public void incCounter(final JournalFile file)
+      {
+         internalgetCounter(file).incrementAndGet();
+      }
+      
       /**
        * @param currentFile
        * @param bb
@@ -2799,7 +2798,7 @@
       {
          bb.writerIndex(SIZE_BYTE + SIZE_INT + SIZE_LONG);
 
-         bb.writeInt(getCounter(currentFile).intValue());
+         bb.writeInt(getCounter(currentFile));
 
       }
 
@@ -2832,11 +2831,6 @@
          }
       }
 
-      public TransactionCallback getCurrentCallback()
-      {
-         return currentCallback;
-      }
-
       /**
        * @return
        */
@@ -2874,7 +2868,7 @@
 
       public void addPositive(final JournalFile file, final long id)
       {
-         getCounter(file).incrementAndGet();
+         incCounter(file);
 
          addFile(file);
 
@@ -2888,7 +2882,7 @@
 
       public void addNegative(final JournalFile file, final long id)
       {
-         getCounter(file).incrementAndGet();
+         incCounter(file);
 
          addFile(file);
 
@@ -3033,20 +3027,6 @@
             file.incPendingTransaction();
          }
       }
-
-      private AtomicInteger getCounter(final JournalFile file)
-      {
-         AtomicInteger value = numberOfElementsPerFile.get(file.getFileID());
-
-         if (value == null)
-         {
-            value = new AtomicInteger();
-            numberOfElementsPerFile.put(file.getFileID(), value);
-         }
-
-         return value;
-      }
-
    }
 
    private static class ByteArrayEncoding implements EncodingSupport