Author: chirino
Date: Thu Feb 5 18:07:57 2009
New Revision: 741212
URL: http://svn.apache.org/viewvc?rev=741212&view=rev
Log:
Changed the cleanup algorithm used in the KahaDB.. It should now be much
faster.
Fixed bug in the BTreeNode.visit() which was making us load most of the pages
instead of letting us filter down to just a few.
Modified:
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/VerifySteadyEnqueueRate.java
activemq/trunk/kahadb/src/main/java/org/apache/kahadb/index/BTreeNode.java
activemq/trunk/kahadb/src/main/java/org/apache/kahadb/journal/Journal.java
Modified:
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
URL:
http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java?rev=741212&r1=741211&r2=741212&view=diff
==============================================================================
---
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
(original)
+++
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
Thu Feb 5 18:07:57 2009
@@ -29,7 +29,9 @@
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
+import java.util.SortedSet;
import java.util.TreeMap;
+import java.util.TreeSet;
import java.util.Map.Entry;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -443,7 +445,6 @@
checkpointUpdate(tx, false);
}
});
- pageFile.flush();
closure.execute();
}
}
@@ -774,46 +775,82 @@
pageFile.flush();
if( cleanup ) {
- // Find empty journal files to remove.
- final HashSet<Integer> inUseFiles = new HashSet<Integer>();
+
+ final TreeSet<Integer> gcCandidateSet = new
TreeSet<Integer>(journal.getFileMap().keySet());
+
+ // Don't GC files under replication
+ if( journalFilesBeingReplicated!=null ) {
+ gcCandidateSet.removeAll(journalFilesBeingReplicated);
+ }
+
+ // Don't GC files after the first in progress tx
+ Location firstTxLocation = metadata.lastUpdate;
+ if( metadata.firstInProgressTransactionLocation!=null ) {
+ firstTxLocation = metadata.firstInProgressTransactionLocation;
+ }
+
+ if( firstTxLocation!=null ) {
+ while( !gcCandidateSet.isEmpty() ) {
+ Integer last = gcCandidateSet.last();
+ if( last >= firstTxLocation.getDataFileId() ) {
+ gcCandidateSet.remove(last);
+ } else {
+ break;
+ }
+ }
+ }
+
+ // Go through all the destinations to see if any of them can
remove GC candidates.
for (StoredDestination sd : storedDestinations.values()) {
+ if( gcCandidateSet.isEmpty() ) {
+ break;
+ }
// Use a visitor to cut down the number of pages that we load
sd.locationIndex.visit(tx, new BTreeVisitor<Location, Long>() {
int last=-1;
public boolean isInterestedInKeysBetween(Location first,
Location second) {
- if( second!=null ) {
- if( last+1 == second.getDataFileId() ) {
- last++;
- inUseFiles.add(last);
- }
- if( last == second.getDataFileId() ) {
- return false;
- }
- }
- return true;
+ if( first==null ) {
+ SortedSet<Integer> subset =
gcCandidateSet.headSet(second.getDataFileId()+1);
+ if( !subset.isEmpty() && subset.last() ==
second.getDataFileId() ) {
+ subset.remove(second.getDataFileId());
+ }
+ return
!subset.isEmpty();
+ } else if( second==null ) {
+ SortedSet<Integer> subset =
gcCandidateSet.tailSet(first.getDataFileId());
+ if( !subset.isEmpty() && subset.first() ==
first.getDataFileId() ) {
+ subset.remove(first.getDataFileId());
+ }
+ return
!subset.isEmpty();
+ } else {
+ SortedSet<Integer> subset =
gcCandidateSet.subSet(first.getDataFileId(), second.getDataFileId()+1);
+ if( !subset.isEmpty() && subset.first() ==
first.getDataFileId() ) {
+ subset.remove(first.getDataFileId());
+ }
+ if( !subset.isEmpty() && subset.last() ==
second.getDataFileId() ) {
+ subset.remove(second.getDataFileId());
+ }
+ return
!subset.isEmpty();
+ }
}
public void visit(List<Location> keys, List<Long> values) {
- for (int i = 0; i < keys.size(); i++) {
- if( last != keys.get(i).getDataFileId() ) {
- inUseFiles.add(keys.get(i).getDataFileId());
- last = keys.get(i).getDataFileId();
+ for (Location l : keys) {
+ int fileId = l.getDataFileId();
+ if( last != fileId ) {
+ gcCandidateSet.remove(fileId);
+ last = fileId;
}
- }
-
+ }
}
});
}
- inUseFiles.addAll(journalFilesBeingReplicated);
- Location l = metadata.lastUpdate;
- if( metadata.firstInProgressTransactionLocation!=null ) {
- l = metadata.firstInProgressTransactionLocation;
+
+ if( !gcCandidateSet.isEmpty() ) {
+ LOG.debug("Cleanup removing the data files:
"+gcCandidateSet);
+ journal.removeDataFiles(gcCandidateSet);
}
-
- LOG.debug("In use files: "+inUseFiles+", lastUpdate: "+l);
- journal.consolidateDataFilesNotIn(inUseFiles,
l==null?null:l.getDataFileId());
}
LOG.debug("Checkpoint done.");
Modified:
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/VerifySteadyEnqueueRate.java
URL:
http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/VerifySteadyEnqueueRate.java?rev=741212&r1=741211&r2=741212&view=diff
==============================================================================
---
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/VerifySteadyEnqueueRate.java
(original)
+++
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/VerifySteadyEnqueueRate.java
Thu Feb 5 18:07:57 2009
@@ -150,12 +150,12 @@
} else {
KahaDBStore kaha = new KahaDBStore();
kaha.setDirectory(new File("target/activemq-data/kahadb"));
- kaha.deleteAllMessages();
- kaha.setCleanupInterval(1000 * 60 * 60 * 60);
// The setEnableJournalDiskSyncs(false) setting is a little
dangerous right now, as I have not verified
// what happens if the index is updated but a journal update is
lost.
// Index is going to be in consistent, but can it be repaired?
kaha.setEnableJournalDiskSyncs(false);
+ // Using a bigger journal file size makes he take fewer spikes as
it is not switching files as often.
+ kaha.getJournal().setMaxFileLength(1024*1024*100);
kaha.getPageFile().setWriteBatchSize(100);
kaha.getPageFile().setEnableWriteThread(true);
broker.setPersistenceAdapter(kaha);
Modified:
activemq/trunk/kahadb/src/main/java/org/apache/kahadb/index/BTreeNode.java
URL:
http://svn.apache.org/viewvc/activemq/trunk/kahadb/src/main/java/org/apache/kahadb/index/BTreeNode.java?rev=741212&r1=741211&r2=741212&view=diff
==============================================================================
--- activemq/trunk/kahadb/src/main/java/org/apache/kahadb/index/BTreeNode.java
(original)
+++ activemq/trunk/kahadb/src/main/java/org/apache/kahadb/index/BTreeNode.java
Thu Feb 5 18:07:57 2009
@@ -524,7 +524,7 @@
}
Key key2 = null;
if( i!=this.children.length-1 ) {
- key1 = keys[i];
+ key2 = keys[i];
}
if( visitor.isInterestedInKeysBetween(key1, key2) ) {
BTreeNode<Key, Value> child = getChild(tx, i);
Modified:
activemq/trunk/kahadb/src/main/java/org/apache/kahadb/journal/Journal.java
URL:
http://svn.apache.org/viewvc/activemq/trunk/kahadb/src/main/java/org/apache/kahadb/journal/Journal.java?rev=741212&r1=741211&r2=741212&view=diff
==============================================================================
--- activemq/trunk/kahadb/src/main/java/org/apache/kahadb/journal/Journal.java
(original)
+++ activemq/trunk/kahadb/src/main/java/org/apache/kahadb/journal/Journal.java
Thu Feb 5 18:07:57 2009
@@ -16,17 +16,12 @@
*/
package org.apache.kahadb.journal;
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
import java.io.File;
import java.io.FilenameFilter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
@@ -42,7 +37,6 @@
import org.apache.kahadb.journal.DataFileAppender.WriteCommand;
import org.apache.kahadb.journal.DataFileAppender.WriteKey;
import org.apache.kahadb.util.ByteSequence;
-import org.apache.kahadb.util.IOHelper;
import org.apache.kahadb.util.LinkedNodeList;
import org.apache.kahadb.util.Scheduler;
@@ -299,18 +293,10 @@
return result;
}
- public synchronized void consolidateDataFilesNotIn(Set<Integer> inUse,
Integer lastFile) throws IOException {
- Set<Integer> unUsed = new HashSet<Integer>(fileMap.keySet());
- unUsed.removeAll(inUse);
-
- for (Integer key : unUsed) {
- // Don't remove files that come after the lastFile
- if (lastFile !=null && key >= lastFile ) {
- continue;
- }
+ public synchronized void removeDataFiles(Set<Integer> files) throws
IOException {
+ for (Integer key : files) {
DataFile dataFile = fileMap.get(key);
-
- // Can't remove the last file either.
+ // Can't remove the last file.
if( dataFile == dataFiles.getTail() ) {
continue;
}
