Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Additional fix for: https://issues.apache.org/jira/browse/AMQ-3775 

On remove of inner ListNode entries the Iterator remove was not properly setting the Next page Id for the previous node leading to
the iterator not being able to traverse the entire list correctly.  

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1328739 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Timothy A. Bish 2012-04-21 21:57:31 +00:00
parent fc4f01cf5d
commit f77b2525a5
2 changed files with 200 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

106
kahadb/src/main/java/org/apache/kahadb/index/ListNode.java
View File

@ -31,18 +31,19 @@ import org.apache.kahadb.util.Marshaller;
import org.apache.kahadb.util.VariableMarshaller; import org.apache.kahadb.util.VariableMarshaller;
/** /**
* The ListNode class represents a node in the List object graph. It is stored in * The ListNode class represents a node in the List object graph. It is stored
* one overflowing Page of a PageFile. * in one overflowing Page of a PageFile.
*/ */
public final class ListNode<Key,Value> { public final class ListNode<Key, Value> {
private final static boolean ADD_FIRST = true; private final static boolean ADD_FIRST = true;
private final static boolean ADD_LAST = false; private final static boolean ADD_LAST = false;
// The index that this node is part of. // The index that this node is part of.
private ListIndex<Key,Value> containingList; private ListIndex<Key, Value> containingList;
// The page associated with this node // The page associated with this node
private Page<ListNode<Key,Value>> page; private Page<ListNode<Key, Value>> page;
private LinkedNodeList<KeyValueEntry<Key, Value>> entries = new LinkedNodeList<KeyValueEntry<Key, Value>>() { private LinkedNodeList<KeyValueEntry<Key, Value>> entries = new LinkedNodeList<KeyValueEntry<Key, Value>>() {
@ -55,8 +56,8 @@ public final class ListNode<Key,Value> {
// The next page after this one. // The next page after this one.
private long next = ListIndex.NOT_SET; private long next = ListIndex.NOT_SET;
static final class KeyValueEntry<Key, Value> extends LinkedNode<KeyValueEntry<Key, Value>> implements Entry<Key, Value> static final class KeyValueEntry<Key, Value> extends LinkedNode<KeyValueEntry<Key, Value>> implements Entry<Key, Value> {
{
private final Key key; private final Key key;
private Value value; private Value value;
@ -85,30 +86,31 @@ public final class ListNode<Key,Value> {
} }
} }
private final class ListNodeIterator implements Iterator<ListNode<Key,Value>> { private final class ListNodeIterator implements Iterator<ListNode<Key, Value>> {
private final Transaction tx; private final Transaction tx;
private final ListIndex<Key,Value> index; private final ListIndex<Key, Value> index;
ListNode<Key,Value> nextEntry; ListNode<Key, Value> nextEntry;
private ListNodeIterator(Transaction tx, ListNode<Key,Value> current) { private ListNodeIterator(Transaction tx, ListNode<Key, Value> current) {
this.tx = tx; this.tx = tx;
nextEntry = current; nextEntry = current;
index = current.getContainingList(); index = current.getContainingList();
} }
public boolean hasNext() { public boolean hasNext() {
return nextEntry !=null; return nextEntry != null;
} }
public ListNode<Key,Value> next() { public ListNode<Key, Value> next() {
ListNode<Key,Value> current = nextEntry; ListNode<Key, Value> current = nextEntry;
if( current !=null ) { if (current != null) {
if (current.next != ListIndex.NOT_SET) { if (current.next != ListIndex.NOT_SET) {
try { try {
nextEntry = index.loadNode(tx, current.next); nextEntry = index.loadNode(tx, current.next);
} catch (IOException unexpected) { } catch (IOException unexpected) {
IllegalStateException e = new IllegalStateException("failed to load next: " + current.next + ", reason: " + unexpected.getLocalizedMessage()); IllegalStateException e = new IllegalStateException("failed to load next: " + current.next + ", reason: "
+ unexpected.getLocalizedMessage());
e.initCause(unexpected); e.initCause(unexpected);
throw e; throw e;
} }
@ -127,12 +129,12 @@ public final class ListNode<Key,Value> {
final class ListIterator implements Iterator<Entry<Key, Value>> { final class ListIterator implements Iterator<Entry<Key, Value>> {
private final Transaction tx; private final Transaction tx;
private final ListIndex<Key,Value> targetList; private final ListIndex<Key, Value> targetList;
ListNode<Key,Value> currentNode, previousNode; ListNode<Key, Value> currentNode, previousNode;
KeyValueEntry<Key, Value> nextEntry; KeyValueEntry<Key, Value> nextEntry;
KeyValueEntry<Key, Value> entryToRemove; KeyValueEntry<Key, Value> entryToRemove;
private ListIterator(Transaction tx, ListNode<Key,Value> current, long start) { private ListIterator(Transaction tx, ListNode<Key, Value> current, long start) {
this.tx = tx; this.tx = tx;
this.currentNode = current; this.currentNode = current;
this.targetList = current.getContainingList(); this.targetList = current.getContainingList();
@ -177,7 +179,7 @@ public final class ListNode<Key,Value> {
} }
public Entry<Key, Value> next() { public Entry<Key, Value> next() {
if( nextEntry !=null ) { if (nextEntry != null) {
entryToRemove = nextEntry; entryToRemove = nextEntry;
nextEntry = entryToRemove.getNext(); nextEntry = entryToRemove.getNext();
return entryToRemove; return entryToRemove;
@ -193,16 +195,15 @@ public final class ListNode<Key,Value> {
try { try {
entryToRemove.unlink(); entryToRemove.unlink();
entryToRemove = null; entryToRemove = null;
ListNode<Key,Value> toRemoveNode = null; ListNode<Key, Value> toRemoveNode = null;
if (currentNode.entries.isEmpty()) { if (currentNode.entries.isEmpty()) {
// may need to free this node // may need to free this node
if (currentNode.isHead() && currentNode.isTail()) { if (currentNode.isHead() && currentNode.isTail()) {
// store empty list // store empty list
} else if (currentNode.isHead()) { } else if (currentNode.isHead()) {
// merge next node into existing headNode // merge next node into existing headNode as we don't want to
// as we don't want to change our headPageId b/c // change our headPageId b/c that is our identity
// that is our identity ListNode<Key, Value> headNode = currentNode;
ListNode<Key,Value> headNode = currentNode;
nextEntry = getFromNextNode(); // will move currentNode nextEntry = getFromNextNode(); // will move currentNode
if (currentNode.isTail()) { if (currentNode.isTail()) {
@ -220,6 +221,10 @@ public final class ListNode<Key,Value> {
previousNode.setNext(ListIndex.NOT_SET); previousNode.setNext(ListIndex.NOT_SET);
previousNode.store(tx); previousNode.store(tx);
targetList.setTailPageId(previousNode.getPageId()); targetList.setTailPageId(previousNode.getPageId());
} else {
toRemoveNode = currentNode;
previousNode.setNext(toRemoveNode.getNext());
previousNode.store(tx);
} }
} }
targetList.onRemove(); targetList.onRemove();
@ -247,7 +252,7 @@ public final class ListNode<Key,Value> {
* @param <Key> * @param <Key>
* @param <Value> * @param <Value>
*/ */
static public final class NodeMarshaller<Key,Value> extends VariableMarshaller<ListNode<Key,Value>> { static public final class NodeMarshaller<Key, Value> extends VariableMarshaller<ListNode<Key, Value>> {
private final Marshaller<Key> keyMarshaller; private final Marshaller<Key> keyMarshaller;
private final Marshaller<Value> valueMarshaller; private final Marshaller<Value> valueMarshaller;
@ -256,10 +261,11 @@ public final class ListNode<Key,Value> {
this.valueMarshaller = valueMarshaller; this.valueMarshaller = valueMarshaller;
} }
public void writePayload(ListNode<Key,Value> node, DataOutput os) throws IOException { public void writePayload(ListNode<Key, Value> node, DataOutput os) throws IOException {
os.writeLong(node.next); os.writeLong(node.next);
short count = (short)node.entries.size(); // cast may truncate value... short count = (short) node.entries.size(); // cast may truncate
if( count != node.entries.size() ) { // value...
if (count != node.entries.size()) {
throw new IOException("short over flow, too many entries in list: " + node.entries.size()); throw new IOException("short over flow, too many entries in list: " + node.entries.size());
} }
@ -273,14 +279,12 @@ public final class ListNode<Key,Value> {
} }
@SuppressWarnings({ "unchecked", "rawtypes" }) @SuppressWarnings({ "unchecked", "rawtypes" })
public ListNode<Key,Value> readPayload(DataInput is) throws IOException { public ListNode<Key, Value> readPayload(DataInput is) throws IOException {
ListNode<Key,Value> node = new ListNode<Key,Value>(); ListNode<Key, Value> node = new ListNode<Key, Value>();
node.setNext(is.readLong()); node.setNext(is.readLong());
final short size = is.readShort(); final short size = is.readShort();
for (short i = 0; i < size; i++) { for (short i = 0; i < size; i++) {
node.entries.addLast( node.entries.addLast(new KeyValueEntry(keyMarshaller.readPayload(is), valueMarshaller.readPayload(is)));
new KeyValueEntry(keyMarshaller.readPayload(is),
valueMarshaller.readPayload(is)));
} }
return node; return node;
} }
@ -311,16 +315,16 @@ public final class ListNode<Key,Value> {
} else { } else {
getContainingList().storeNode(tx, this, false); getContainingList().storeNode(tx, this, false);
} }
} catch ( Transaction.PageOverflowIOException e ) { } catch (Transaction.PageOverflowIOException e) {
split(tx, ADD_FIRST); split(tx, ADD_FIRST);
} }
} }
private void store(Transaction tx, boolean addFirst) throws IOException { private void store(Transaction tx, boolean addFirst) throws IOException {
try { try {
// When we split to a node of one element we can span multiple // When we split to a node of one element we can span multiple pages for that
// pages for that entry, otherwise we keep the entries on one // entry, otherwise we keep the entries on one page to avoid fragmented reads
// page to avoid fragmented reads and segment the list traversal. // and segment the list traversal.
if (this.entries.size() == 1) { if (this.entries.size() == 1) {
getContainingList().storeNode(tx, this, true); getContainingList().storeNode(tx, this, true);
} else { } else {
@ -331,7 +335,7 @@ public final class ListNode<Key,Value> {
getContainingList().setTailPageId(getPageId()); getContainingList().setTailPageId(getPageId());
} }
} catch ( Transaction.PageOverflowIOException e ) { } catch (Transaction.PageOverflowIOException e) {
// If we get an overflow // If we get an overflow
split(tx, addFirst); split(tx, addFirst);
} }
@ -353,7 +357,7 @@ public final class ListNode<Key,Value> {
extension.setNext(this.getNext()); extension.setNext(this.getNext());
extension.store(tx, isAddFirst); extension.store(tx, isAddFirst);
this.setNext(extension.getPageId()); this.setNext(extension.getPageId());
} else { } else {
extension.setEntries(entries.getTail().getPrevious().splitAfter()); extension.setEntries(entries.getTail().getPrevious().splitAfter());
extension.store(tx, isAddFirst); extension.store(tx, isAddFirst);
getContainingList().setTailPageId(extension.getPageId()); getContainingList().setTailPageId(extension.getPageId());
@ -375,7 +379,7 @@ public final class ListNode<Key,Value> {
KeyValueEntry<Key, Value> nextEntry = entries.getTail(); KeyValueEntry<Key, Value> nextEntry = entries.getTail();
while (nextEntry != null) { while (nextEntry != null) {
if (nextEntry.getKey().equals(key)) { if (nextEntry.getKey().equals(key)) {
result = nextEntry.getValue(); result = nextEntry.getValue();
break; break;
} }
nextEntry = nextEntry.getPrevious(); nextEntry = nextEntry.getPrevious();
@ -383,23 +387,23 @@ public final class ListNode<Key,Value> {
return result; return result;
} }
public boolean isEmpty(final Transaction tx) { public boolean isEmpty(final Transaction tx) {
return entries.isEmpty(); return entries.isEmpty();
} }
public Entry<Key,Value> getFirst(Transaction tx) { public Entry<Key, Value> getFirst(Transaction tx) {
return entries.getHead(); return entries.getHead();
} }
public Entry<Key,Value> getLast(Transaction tx) { public Entry<Key, Value> getLast(Transaction tx) {
return entries.getTail(); return entries.getTail();
} }
public Iterator<Entry<Key,Value>> iterator(final Transaction tx, long pos) throws IOException { public Iterator<Entry<Key, Value>> iterator(final Transaction tx, long pos) throws IOException {
return new ListIterator(tx, this, pos); return new ListIterator(tx, this, pos);
} }
public Iterator<Entry<Key,Value>> iterator(final Transaction tx) throws IOException { public Iterator<Entry<Key, Value>> iterator(final Transaction tx) throws IOException {
return new ListIterator(tx, this, 0); return new ListIterator(tx, this, 0);
} }
@ -428,9 +432,9 @@ public final class ListNode<Key,Value> {
return found; return found;
} }
/////////////////////////////////////////////////////////////////// // /////////////////////////////////////////////////////////////////
// Implementation methods // Implementation methods
/////////////////////////////////////////////////////////////////// // /////////////////////////////////////////////////////////////////
public long getPageId() { public long getPageId() {
return page.getPageId(); return page.getPageId();
@ -456,7 +460,7 @@ public final class ListNode<Key,Value> {
this.containingList = list; this.containingList = list;
} }
public ListIndex<Key,Value> getContainingList() { public ListIndex<Key, Value> getContainingList() {
return containingList; return containingList;
} }
@ -474,8 +478,6 @@ public final class ListNode<Key,Value> {
@Override @Override
public String toString() { public String toString() {
return "[ListNode(" + (page != null ? page.getPageId() + "->" + next : "null") + ")[" + entries.size() + "]]"; return "[ListNode(" + (page != null ? page.getPageId() + "->" + next : "null") + ")[" + entries.size() + "]]";
} }
} }

146
kahadb/src/test/java/org/apache/kahadb/index/ListIndexTest.java
View File

@ -495,6 +495,152 @@ public class ListIndexTest extends IndexTestSupport {
} }
} }
public void testListLargeDataAddWithReverseRemove() throws Exception {
final int NUM_ITERATIONS = 100;
pf = new PageFile(directory, getClass().getName());
pf.setPageSize(4*1024);
pf.setEnablePageCaching(false);
pf.setWriteBatchSize(1);
pf.load();
tx = pf.tx();
long id = tx.allocate().getPageId();
ListIndex<String, SequenceSet> test = new ListIndex<String, SequenceSet>(pf, id);
test.setKeyMarshaller(StringMarshaller.INSTANCE);
test.setValueMarshaller(SequenceSet.Marshaller.INSTANCE);
test.load(tx);
tx.commit();
int expectedListEntries = 0;
int nextSequenceId = 0;
LOG.info("Loading up the ListIndex with "+NUM_ITERATIONS+" entires and sparsely populating the sequences.");
for (int i = 0; i < NUM_ITERATIONS; ++i) {
test.add(tx, String.valueOf(expectedListEntries++), new SequenceSet());
for (int j = 0; j < expectedListEntries; j++) {
SequenceSet sequenceSet = test.get(tx, String.valueOf(j));
int startSequenceId = nextSequenceId;
for (int ix = 0; ix < NUM_ITERATIONS; ix++) {
sequenceSet.add(nextSequenceId++);
test.put(tx, String.valueOf(j), sequenceSet);
}
sequenceSet = test.get(tx, String.valueOf(j));
for (int ix = 0; ix < NUM_ITERATIONS - 1; ix++) {
sequenceSet.remove(startSequenceId++);
test.put(tx, String.valueOf(j), sequenceSet);
}
}
}
LOG.info("Checking if Index has the expected number of entries.");
for (int i = 0; i < NUM_ITERATIONS; ++i) {
assertTrue("List should contain Key["+i+"]",test.containsKey(tx, String.valueOf(i)));
assertNotNull("List contents of Key["+i+"] should not be null", test.get(tx, String.valueOf(i)));
}
LOG.info("Index has the expected number of entries.");
assertEquals(expectedListEntries, test.size());
for (int i = NUM_ITERATIONS - 1; i >= 0; --i) {
LOG.debug("Size of ListIndex before removal of entry ["+i+"] is: " + test.size());
assertTrue("List should contain Key["+i+"]",test.containsKey(tx, String.valueOf(i)));
assertNotNull("List contents of Key["+i+"] should not be null", test.remove(tx, String.valueOf(i)));
LOG.debug("Size of ListIndex after removal of entry ["+i+"] is: " + test.size());
assertEquals(--expectedListEntries, test.size());
}
}
public void testListLargeDataAddAndNonSequentialRemove() throws Exception {
final int NUM_ITERATIONS = 100;
pf = new PageFile(directory, getClass().getName());
pf.setPageSize(4*1024);
pf.setEnablePageCaching(false);
pf.setWriteBatchSize(1);
pf.load();
tx = pf.tx();
long id = tx.allocate().getPageId();
ListIndex<String, SequenceSet> test = new ListIndex<String, SequenceSet>(pf, id);
test.setKeyMarshaller(StringMarshaller.INSTANCE);
test.setValueMarshaller(SequenceSet.Marshaller.INSTANCE);
test.load(tx);
tx.commit();
int expectedListEntries = 0;
int nextSequenceId = 0;
LOG.info("Loading up the ListIndex with "+NUM_ITERATIONS+" entires and sparsely populating the sequences.");
for (int i = 0; i < NUM_ITERATIONS; ++i) {
test.add(tx, String.valueOf(expectedListEntries++), new SequenceSet());
for (int j = 0; j < expectedListEntries; j++) {
SequenceSet sequenceSet = test.get(tx, String.valueOf(j));
int startSequenceId = nextSequenceId;
for (int ix = 0; ix < NUM_ITERATIONS; ix++) {
sequenceSet.add(nextSequenceId++);
test.put(tx, String.valueOf(j), sequenceSet);
}
sequenceSet = test.get(tx, String.valueOf(j));
for (int ix = 0; ix < NUM_ITERATIONS - 1; ix++) {
sequenceSet.remove(startSequenceId++);
test.put(tx, String.valueOf(j), sequenceSet);
}
}
}
LOG.info("Checking if Index has the expected number of entries.");
for (int i = 0; i < NUM_ITERATIONS; ++i) {
assertTrue("List should contain Key["+i+"]",test.containsKey(tx, String.valueOf(i)));
assertNotNull("List contents of Key["+i+"] should not be null", test.get(tx, String.valueOf(i)));
}
LOG.info("Index has the expected number of entries.");
assertEquals(expectedListEntries, test.size());
for (int i = 0; i < NUM_ITERATIONS; i += 2) {
LOG.debug("Size of ListIndex before removal of entry ["+i+"] is: " + test.size());
assertTrue("List should contain Key["+i+"]",test.containsKey(tx, String.valueOf(i)));
assertNotNull("List contents of Key["+i+"] should not be null", test.remove(tx, String.valueOf(i)));
LOG.debug("Size of ListIndex after removal of entry ["+i+"] is: " + test.size());
assertEquals(--expectedListEntries, test.size());
}
for (int i = NUM_ITERATIONS - 1; i > 0; i -= 2) {
LOG.debug("Size of ListIndex before removal of entry ["+i+"] is: " + test.size());
assertTrue("List should contain Key["+i+"]",test.containsKey(tx, String.valueOf(i)));
assertNotNull("List contents of Key["+i+"] should not be null", test.remove(tx, String.valueOf(i)));
LOG.debug("Size of ListIndex after removal of entry ["+i+"] is: " + test.size());
assertEquals(--expectedListEntries, test.size());
}
assertEquals(0, test.size());
}
static class HashSetStringMarshaller extends VariableMarshaller<HashSet<String>> { static class HashSetStringMarshaller extends VariableMarshaller<HashSet<String>> {
final static HashSetStringMarshaller INSTANCE = new HashSetStringMarshaller(); final static HashSetStringMarshaller INSTANCE = new HashSetStringMarshaller();