提交 162950e3 authored 作者: Thomas Mueller's avatar Thomas Mueller

A concurrent list, to replace the array list of old roots (work in progress)

上级 0dcbc5f7
...@@ -7,70 +7,51 @@ package org.h2.mvstore; ...@@ -7,70 +7,51 @@ package org.h2.mvstore;
import java.util.Iterator; import java.util.Iterator;
/** /**
* A very simple linked list that supports concurrent access. * A very simple linked list that supports concurrent access.
* Internally, it uses immutable objects.
* It uses recursion and is not meant for long lists.
* *
* @param <K> the key type * @param <K> the key type
*/ */
public class ConcurrentLinkedList<K> { public class ConcurrentLinkedList<K> {
volatile Entry<K> head; static final Entry<?> NULL = new Entry<Object>(null, null);
private volatile Entry<K> tail;
@SuppressWarnings("unchecked")
volatile Entry<K> head = (Entry<K>) NULL;
public K peekFirst() { public K peekFirst() {
Entry<K> x = head; Entry<K> x = head;
return x == null ? null : x.obj; return x.obj;
} }
public K peekLast() { public K peekLast() {
Entry<K> x = tail; Entry<K> x = head;
return x == null ? null : x.obj; while (x != NULL && x.next != NULL) {
} x = x.next;
public void add(K obj) {
Entry<K> x = new Entry<K>(obj);
Entry<K> t = tail;
if (t != null) {
t.next = x;
} }
tail = x; return x.obj;
if (head == null) {
head = x;
} }
public synchronized void add(K obj) {
head = Entry.append(head, obj);
} }
public boolean removeFirst(K obj) { public synchronized boolean removeFirst(K obj) {
Entry<K> x = head; if (head.obj != obj) {
if (x == null || x.obj != obj) {
return false; return false;
} }
if (head == tail) { head = head.next;
tail = x.next;
}
head = x.next;
return true; return true;
} }
public boolean removeLast(K obj) { public synchronized boolean removeLast(K obj) {
Entry<K> x = head; if (peekLast() != obj) {
if (x == null) {
return false; return false;
} }
Entry<K> prev = null; head = Entry.removeLast(head);
while (x.next != null) {
prev = x;
x = x.next;
}
if (x.obj != obj) {
return false;
}
if (prev != null) {
prev.next = null;
}
if (head == tail) {
head = prev;
}
tail = prev;
return true; return true;
} }
...@@ -81,7 +62,7 @@ public class ConcurrentLinkedList<K> { ...@@ -81,7 +62,7 @@ public class ConcurrentLinkedList<K> {
@Override @Override
public boolean hasNext() { public boolean hasNext() {
return current != null; return current != NULL;
} }
@Override @Override
...@@ -106,9 +87,27 @@ public class ConcurrentLinkedList<K> { ...@@ -106,9 +87,27 @@ public class ConcurrentLinkedList<K> {
final K obj; final K obj;
Entry<K> next; Entry<K> next;
Entry(K obj) { Entry(K obj, Entry<K> next) {
this.obj = obj; this.obj = obj;
this.next = next;
}
@SuppressWarnings("unchecked")
static <K> Entry<K> append(Entry<K> list, K obj) {
if (list == NULL) {
return new Entry<K>(obj, (Entry<K>) NULL);
} }
return new Entry<K>(list.obj, append(list.next, obj));
}
@SuppressWarnings("unchecked")
static <K> Entry<K> removeLast(Entry<K> list) {
if (list == NULL || list.next == NULL) {
return (Entry<K>) NULL;
}
return new Entry<K>(list.obj, removeLast(list.next));
}
} }
} }
/*
* Copyright 2004-2014 H2 Group. Multiple-Licensed under the MPL 2.0,
* and the EPL 1.0 (http://h2database.com/html/license.html).
* Initial Developer: H2 Group
*/
package org.h2.mvstore;
import java.util.Iterator;
/**
* A ring buffer that supports concurrent access.
*
* @param <K> the key type
*/
public class ConcurrentRing<K> {
K[] buffer;
volatile int readPos;
volatile int writePos;
@SuppressWarnings("unchecked")
public ConcurrentRing() {
buffer = (K[]) new Object[4];
}
public K peekFirst() {
return buffer[getIndex(readPos)];
}
public K peekLast() {
return buffer[getIndex(writePos - 1)];
}
public void add(K obj) {
buffer[getIndex(writePos)] = obj;
writePos++;
if (writePos - readPos >= buffer.length) {
// double the capacity
@SuppressWarnings("unchecked")
K[] b2 = (K[]) new Object[buffer.length * 2];
for (int i = readPos; i < writePos; i++) {
K x = buffer[getIndex(i)];
int i2 = i & b2.length - 1;
b2[i2] = x;
}
buffer = b2;
}
}
public boolean removeFirst(K obj) {
int p = readPos;
int idx = getIndex(p);
if (buffer[idx] != obj) {
return false;
}
buffer[idx] = null;
readPos = p + 1;
return true;
}
public boolean removeLast(K obj) {
int p = writePos;
int idx = getIndex(p - 1);
if (buffer[idx] != obj) {
return false;
}
buffer[idx] = null;
writePos = p - 1;
return true;
}
int getIndex(int pos) {
return pos & buffer.length - 1;
}
public Iterator<K> iterator() {
return new Iterator<K>() {
int pos = readPos;
@Override
public boolean hasNext() {
return pos != writePos;
}
@Override
public K next() {
return buffer[getIndex(pos++)];
}
@Override
public void remove() {
throw DataUtils.newUnsupportedOperationException("remove");
}
};
}
}
...@@ -8,7 +8,6 @@ package org.h2.mvstore; ...@@ -8,7 +8,6 @@ package org.h2.mvstore;
import java.util.AbstractList; import java.util.AbstractList;
import java.util.AbstractMap; import java.util.AbstractMap;
import java.util.AbstractSet; import java.util.AbstractSet;
import java.util.ArrayList;
import java.util.HashMap; import java.util.HashMap;
import java.util.Iterator; import java.util.Iterator;
import java.util.List; import java.util.List;
...@@ -54,8 +53,8 @@ public class MVMap<K, V> extends AbstractMap<K, V> ...@@ -54,8 +53,8 @@ public class MVMap<K, V> extends AbstractMap<K, V>
private final DataType keyType; private final DataType keyType;
private final DataType valueType; private final DataType valueType;
private ConcurrentRing<Page> oldRoots = private ConcurrentLinkedList<Page> oldRoots =
new ConcurrentRing<Page>(); new ConcurrentLinkedList<Page>();
private boolean closed; private boolean closed;
private boolean readOnly; private boolean readOnly;
......
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论