Skip to content

H
h2database
提交 8a5415bb authored 作者: Thomas Mueller's avatar Thomas Mueller
浏览文件
操作

MVStore: concurrency problems have been fixed.

上级 8022f479
显示空白字符变更
内嵌 并排
正在显示 包含 180 行增加 和 101 行删除
h2/src/docsrc/html/changelog.html
浏览文件 @ 8a5415bb
......@@ -19,6 +19,7 @@ Change Log
<h2>Next Version (unreleased)</h2>
<ul><li>Issue 73: MySQL compatibility: support REPLACE, patch by Cemo Koc.
</li><li>MVStore: concurrency problems have been fixed.
</li></ul>
<h2>Version 1.3.174 (2013-10-19)</h2>
......
h2/src/installer/release.txt
浏览文件 @ 8a5415bb
......@@ -28,4 +28,5 @@ Add to http://freecode.com/
Add to http://twitter.com
- tweet: add @geospatialnews for the new geometry type and disk spatial index
Close bugs: http://code.google.com/p/h2database/issues/list
Update statistics
h2/src/main/org/h2/mvstore/DataUtils.java
浏览文件 @ 8a5415bb
......@@ -396,7 +396,7 @@ public class DataUtils {
}
throw newIllegalStateException(
ERROR_READING_FAILED,
"Reading from {0} failed; file length {1} read length {1} at {2}",
"Reading from {0} failed; file length {1} read length {2} at {3}",
file, size, dst.remaining(), pos, e);
}
}
......
h2/src/main/org/h2/mvstore/MVMap.java
浏览文件 @ 8a5415bb
......@@ -214,7 +214,6 @@ public class MVMap<K, V> extends AbstractMap<K, V>
*/
@SuppressWarnings("unchecked")
public K getKey(long index) {
checkOpen();
if (index < 0 || index >= size()) {
return null;
}
......@@ -285,7 +284,6 @@ public class MVMap<K, V> extends AbstractMap<K, V>
* @return the index
*/
public long getKeyIndex(K key) {
checkOpen();
if (size() == 0) {
return -1;
}
......@@ -319,7 +317,6 @@ public class MVMap<K, V> extends AbstractMap<K, V>
*/
@SuppressWarnings("unchecked")
protected K getFirstLast(boolean first) {
checkOpen();
if (size() == 0) {
return null;
}
......@@ -383,7 +380,6 @@ public class MVMap<K, V> extends AbstractMap<K, V>
* @return the key, or null if no such key exists
*/
protected K getMinMax(K key, boolean min, boolean excluding) {
checkOpen();
return getMinMax(root, key, min, excluding);
}
......@@ -429,7 +425,6 @@ public class MVMap<K, V> extends AbstractMap<K, V>
@Override
@SuppressWarnings("unchecked")
public V get(Object key) {
checkOpen();
return (V) binarySearch(root, key);
}
......@@ -514,7 +509,7 @@ public class MVMap<K, V> extends AbstractMap<K, V>
* Remove all entries, and close the map.
*/
public void removeMap() {
checkOpen();
int todoMoveToMVStore;
if (this == store.getMetaMap()) {
return;
}
......@@ -712,7 +707,6 @@ public class MVMap<K, V> extends AbstractMap<K, V>
} else {
list.add(root);
}
store.markChanged(this);
}
root = newRoot;
}
......@@ -775,7 +769,6 @@ public class MVMap<K, V> extends AbstractMap<K, V>
* @return the iterator
*/
public Cursor<K> keyIterator(K from) {
checkOpen();
return new Cursor<K>(this, root, from);
}
......@@ -790,7 +783,6 @@ public class MVMap<K, V> extends AbstractMap<K, V>
@Override
public Set<K> keySet() {
checkOpen();
final MVMap<K, V> map = this;
final Page root = this.root;
return new AbstractSet<K>() {
......@@ -831,7 +823,7 @@ public class MVMap<K, V> extends AbstractMap<K, V>
return store.getMapName(id);
}
public MVStore getStore() {
MVStore getStore() {
return store;
}
......@@ -897,18 +889,6 @@ public class MVMap<K, V> extends AbstractMap<K, V>
return readOnly;
}
/**
* Check whether the map is open.
*
* @throws IllegalStateException if the map is closed
*/
protected void checkOpen() {
if (closed) {
throw DataUtils.newIllegalStateException(
DataUtils.ERROR_CLOSED, "This map is closed");
}
}
/**
* This method is called before writing to the map. The default
* implementation checks whether writing is allowed, and tries
......@@ -918,8 +898,11 @@ public class MVMap<K, V> extends AbstractMap<K, V>
* or if another thread is concurrently writing
*/
protected void beforeWrite() {
if (closed) {
throw DataUtils.newIllegalStateException(
DataUtils.ERROR_CLOSED, "This map is closed");
}
if (readOnly) {
checkOpen();
throw DataUtils.newUnsupportedOperationException(
"This map is read-only");
}
......@@ -991,14 +974,12 @@ public class MVMap<K, V> extends AbstractMap<K, V>
* @return the number of entries
*/
public long sizeAsLong() {
checkOpen();
return root.getTotalCount();
}
@Override
public boolean isEmpty() {
// could also use (sizeAsLong() == 0)
checkOpen();
return root.isLeaf() && root.getKeyCount() == 0;
}
......@@ -1061,7 +1042,7 @@ public class MVMap<K, V> extends AbstractMap<K, V>
*
* @return the opened map
*/
protected MVMap<K, V> openReadOnly() {
MVMap<K, V> openReadOnly() {
MVMap<K, V> m = new MVMap<K, V>(keyType, valueType);
m.readOnly = true;
HashMap<String, String> config = New.hashMap();
......@@ -1140,6 +1121,7 @@ public class MVMap<K, V> extends AbstractMap<K, V>
* @param newMapName the name name
*/
public void renameMap(String newMapName) {
int todoMoveToMVStore;
beforeWrite();
try {
store.renameMap(this, newMapName);
......
h2/src/main/org/h2/mvstore/MVStore.java
浏览文件 @ 8a5415bb
......@@ -47,7 +47,7 @@ Documentation
TestMVStoreDataLoss
MVTableEngine:
- use StreamStore
- use StreamStore to avoid deadlocks
- when the MVStore was enabled before, use it again
(probably by checking existence of the mvstore file)
- not use the .h2.db file
......@@ -56,7 +56,6 @@ MVTableEngine:
TransactionStore:
MVStore:
- rename FilePathCrypt to FilePathCipher
- automated 'kill process' and 'power failure' test
- update checkstyle
- auto-compact from time to time and on close
......@@ -107,11 +106,14 @@ MVStore:
- storage that splits database into multiple files,
to speed up compact and allow using trim
(by truncating / deleting empty files)
- add new feature to file systems that avoid copying data
(reads should return a ByteBuffer, not write into one)
- add new feature to the file system API to avoid copying data
(reads that returns a ByteBuffer instead of writing into one)
for memory mapped files and off-heap storage
- do we need to store a dummy chunk entry in the chunk itself?
currently yes, as some fields are not set in the chunk header
- off-heap LIRS cache (the LIRS cache should use a map factory)
- support log structured merge style operations (blind writes)
using one map per level plus bloom filter
- have a strict call order MVStore -> MVMap -> Page -> FileStore
*/
......@@ -207,6 +209,7 @@ public class MVStore {
private int unsavedPageCount;
private int unsavedPageCountMax;
private boolean storeNeeded;
/**
* The time the store was created, in milliseconds since 1970.
......@@ -427,9 +430,9 @@ public class MVStore {
c.put("createVersion", Long.toString(currentVersion));
map = builder.create();
map.init(this, c);
markMetaChanged();
meta.put("map." + id, map.asString(name));
meta.put("name." + name, Integer.toString(id));
markMetaChanged();
root = 0;
}
map.setRootPos(root, -1);
......@@ -439,8 +442,11 @@ public class MVStore {
/**
* Get the metadata map. This data is for informational purposes only. The
* data is subject to change in future versions. The data should not be
* modified (doing so may corrupt the store).
* data is subject to change in future versions.
* <p>
* The data should not be modified (doing so may corrupt the store). Writing
* to it is not always detected as a modification, so that changes to it
* might not be stored.
* <p>
* It contains the following entries:
*
......@@ -502,21 +508,10 @@ public class MVStore {
return meta.containsKey("name." + name);
}
/**
* Mark a map as changed (containing unsaved changes).
*
* @param map the map
*/
void markChanged(MVMap<?, ?> map) {
if (map == meta) {
metaChanged = true;
}
}
private void markMetaChanged() {
// changes in the metadata alone are usually not detected, as the meta
// map is changed after storing
markChanged(meta);
metaChanged = true;
}
private void readMeta() {
......@@ -892,6 +887,10 @@ public class MVStore {
for (MVMap<?, ?> m : list) {
m.setWriteVersion(version);
long v = m.getVersion();
if (m.getCreateVersion() > storeVersion) {
// the map was created after storing started
continue;
}
if (v >= 0 && v >= lastStoredVersion) {
m.waitUntilWritten(storeVersion);
MVMap<?, ?> r = m.openVersion(storeVersion);
......@@ -1203,6 +1202,7 @@ public class MVStore {
}
for (Chunk c : free) {
chunks.remove(c.id);
markMetaChanged();
meta.remove("chunk." + c.id);
int length = MathUtils.roundUpInt(c.length, BLOCK_SIZE) + BLOCK_SIZE;
fileStore.free(c.start, length);
......@@ -1237,6 +1237,7 @@ public class MVStore {
buff.position(0);
fileStore.writeFully(end, buff.getBuffer());
releaseWriteBuffer(buff);
markMetaChanged();
meta.put("chunk." + c.id, c.asString());
}
boolean oldReuse = reuseSpace;
......@@ -1273,6 +1274,7 @@ public class MVStore {
buff.position(0);
fileStore.writeFully(pos, buff.getBuffer());
releaseWriteBuffer(buff);
markMetaChanged();
meta.put("chunk." + c.id, c.asString());
}
......@@ -1501,11 +1503,15 @@ public class MVStore {
}
private void registerFreePage(long version, int chunkId, long maxLengthLive, int pageCount) {
HashMap<Integer, Chunk>freed = freedPageSpace.get(version);
HashMap<Integer, Chunk> freed = freedPageSpace.get(version);
if (freed == null) {
freed = New.hashMap();
freedPageSpace.put(version, freed);
HashMap<Integer, Chunk> f2 = freedPageSpace.putIfAbsent(version, freed);
if (f2 != null) {
freed = f2;
}
}
synchronized (freed) {
Chunk f = freed.get(chunkId);
if (f == null) {
f = new Chunk(chunkId);
......@@ -1514,6 +1520,7 @@ public class MVStore {
f.maxLengthLive -= maxLengthLive;
f.pageCountLive -= pageCount;
}
}
Compressor getCompressor() {
return compressor;
......@@ -1679,17 +1686,17 @@ public class MVStore {
*/
void registerUnsavedPage() {
unsavedPageCount++;
if (unsavedPageCount > unsavedPageCountMax && unsavedPageCountMax > 0) {
storeNeeded = true;
}
}
/**
* This method is called before writing to a map.
*/
void beforeWrite() {
if (currentStoreVersion >= 0) {
// store is possibly called within store, if the meta map changed
return;
}
if (unsavedPageCount > unsavedPageCountMax && unsavedPageCountMax > 0) {
if (storeNeeded) {
storeNeeded = false;
store(true);
}
}
......@@ -1819,6 +1826,7 @@ public class MVStore {
// rollback might have rolled back the stored chunk metadata as well
Chunk c = chunks.get(lastChunkId - 1);
if (c != null) {
markMetaChanged();
meta.put("chunk." + c.id, c.asString());
}
currentVersion = version;
......
h2/src/main/org/h2/mvstore/rtree/MVRTreeMap.java
浏览文件 @ 8a5415bb
......@@ -52,7 +52,6 @@ public class MVRTreeMap<V> extends MVMap<SpatialKey, V> {
@Override
@SuppressWarnings("unchecked")
public V get(Object key) {
checkOpen();
return (V) get(root, key);
}
......@@ -63,7 +62,6 @@ public class MVRTreeMap<V> extends MVMap<SpatialKey, V> {
* @return the iterator
*/
public RTreeCursor findIntersectingKeys(SpatialKey x) {
checkOpen();
return new RTreeCursor(root, x) {
@Override
protected boolean check(boolean leaf, SpatialKey key, SpatialKey test) {
......@@ -79,7 +77,6 @@ public class MVRTreeMap<V> extends MVMap<SpatialKey, V> {
* @return the iterator
*/
public RTreeCursor findContainedKeys(SpatialKey x) {
checkOpen();
return new RTreeCursor(root, x) {
@Override
protected boolean check(boolean leaf, SpatialKey key, SpatialKey test) {
......
h2/src/test/org/h2/test/TestAll.java
浏览文件 @ 8a5415bb
......@@ -233,7 +233,7 @@ java org.h2.test.TestAll timer
*/
;
private static final boolean MV_STORE = false;
private static final boolean MV_STORE = true;
/**
* If the test should run with many rows.
......@@ -470,7 +470,7 @@ kill -9 `jps -l | grep "org.h2.test." | cut -d " " -f 1`
prof.interval = 1;
prof.startCollecting();
if (test.mvStore) {
TestPerformance.main("-init", "-db", "9", "-size", "1000");
TestPerformance.main("-init", "-db", "9", "-size", "10000");
} else {
TestPerformance.main("-init", "-db", "1");
}
......@@ -481,7 +481,7 @@ kill -9 `jps -l | grep "org.h2.test." | cut -d " " -f 1`
prof.depth = 16;
prof.interval = 1;
prof.startCollecting();
TestPerformance.main("-init", "-db", "1", "-size", "1000");
TestPerformance.main("-init", "-db", "1", "-size", "10000");
prof.stopCollecting();
System.out.println(prof.getTop(3));
}
......
h2/src/test/org/h2/test/store/TestConcurrent.java
浏览文件 @ 8a5415bb
......@@ -10,6 +10,7 @@ import java.io.ByteArrayOutputStream;
import java.io.FileOutputStream;
import java.io.InputStream;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.ConcurrentModificationException;
import java.util.Iterator;
import java.util.Map;
......@@ -23,6 +24,7 @@ import org.h2.mvstore.MVStore;
import org.h2.store.fs.FileChannelInputStream;
import org.h2.store.fs.FileUtils;
import org.h2.test.TestBase;
import org.h2.upgrade.v1_1.util.New;
import org.h2.util.Task;
/**
......@@ -41,9 +43,12 @@ public class TestConcurrent extends TestMVStore {
@Override
public void test() throws Exception {
FileUtils.deleteRecursive(getBaseDir(), true);
FileUtils.createDirectories(getBaseDir());
FileUtils.deleteRecursive("memFS:", false);
testConcurrentFree();
testConcurrentStoreAndRemoveMap();
testConcurrentStoreAndClose();
testConcurrentOnlineBackup();
......@@ -53,65 +58,149 @@ public class TestConcurrent extends TestMVStore {
testConcurrentRead();
}
private void testConcurrentFree() throws InterruptedException {
String fileName = "memFS:testConcurrentFree.h3";
for (int test = 0; test < 10; test++) {
FileUtils.delete(fileName);
final MVStore s1 = new MVStore.Builder().
fileName(fileName).writeDelay(-1).open();
s1.setRetentionTime(0);
final int count = 200;
for (int i = 0; i < count; i++) {
MVMap<Integer, Integer> m = s1.openMap("d" + i);
m.put(1, 1);
if (i % 2 == 0) {
s1.store();
}
}
s1.store();
s1.close();
final MVStore s = new MVStore.Builder().
fileName(fileName).writeDelay(-1).open();
s.setRetentionTime(0);
final ArrayList<MVMap<Integer, Integer>> list = New.arrayList();
for (int i = 0; i < count; i++) {
MVMap<Integer, Integer> m = s.openMap("d" + i);
list.add(m);
}
final AtomicInteger counter = new AtomicInteger();
Task task = new Task() {
@Override
public void call() throws Exception {
while (!stop) {
int x = counter.getAndIncrement();
if (x >= count) {
break;
}
MVMap<Integer, Integer> m = list.get(x);
m.clear();
m.removeMap();
}
}
};
task.execute();
Thread.sleep(1);
while (true) {
int x = counter.getAndIncrement();
if (x >= count) {
break;
}
MVMap<Integer, Integer> m = list.get(x);
m.clear();
m.removeMap();
if (x % 5 == 0) {
s.incrementVersion();
}
}
task.get();
s.store();
MVMap<String, String> meta = s.getMetaMap();
int chunkCount = 0;
for (String k : meta.keyList()) {
if (k.startsWith("chunk.")) {
chunkCount++;
}
}
assertEquals(1, chunkCount);
s.close();
}
}
private void testConcurrentStoreAndRemoveMap() throws InterruptedException {
String fileName = getBaseDir() + "/testConcurrentStoreAndRemoveMap.h3";
String fileName = "memFS:testConcurrentStoreAndRemoveMap.h3";
final MVStore s = openStore(fileName);
int count = 100;
int count = 200;
for (int i = 0; i < count; i++) {
MVMap<Integer, Integer> m = s.openMap("d" + i);
m.put(1, 1);
}
final AtomicInteger counter = new AtomicInteger();
Task task = new Task() {
@Override
public void call() throws Exception {
while (!stop) {
counter.incrementAndGet();
s.store();
}
}
};
task.execute();
Thread.sleep(1);
for (int i = 0; i < count; i++) {
for (int i = 0; i < count || counter.get() < count; i++) {
MVMap<Integer, Integer> m = s.openMap("d" + i);
m.put(1, 10);
m.removeMap();
if (task.isFinished()) {
break;
}
}
task.get();
s.close();
}
private void testConcurrentStoreAndClose() throws InterruptedException {
String fileName = getBaseDir() + "/testConcurrentStoreAndClose.h3";
String fileName = "memFS:testConcurrentStoreAndClose";
for (int i = 0; i < 10; i++) {
FileUtils.delete(fileName);
final MVStore s = openStore(fileName);
final AtomicInteger counter = new AtomicInteger();
Task task = new Task() {
@Override
public void call() throws Exception {
int x = 0;
while (!stop) {
s.setStoreVersion(x++);
s.setStoreVersion(counter.incrementAndGet());
s.store();
}
}
};
task.execute();
while (counter.get() < 5) {
Thread.sleep(1);
}
try {
s.close();
// sometimes closing works, in which case
// storing fails at some point
Thread.sleep(100);
// storing must fail at some point (not necessarily
// immediately)
for (int x = counter.get(), y = x; x <= y + 2; x++) {
Thread.sleep(1);
}
Exception e = task.getException();
assertEquals(DataUtils.ERROR_CLOSED,
DataUtils.getErrorCode(e.getMessage()));
} catch (IllegalStateException e) {
// sometimes storing works, in which case
// closing fails
// closing must fail
assertEquals(DataUtils.ERROR_WRITING_FAILED,
DataUtils.getErrorCode(e.getMessage()));
task.get();
}
s.close();
}
}
/**
* Test the concurrent map implementation.
......@@ -177,7 +266,7 @@ public class TestConcurrent extends TestMVStore {
@Override
public void call() throws Exception {
while (!stop) {
for (int i = 0; i < 20; i++) {
for (int i = 0; i < 10; i++) {
map.put(i, new byte[100 * r.nextInt(100)]);
}
s.store();
......@@ -187,7 +276,7 @@ public class TestConcurrent extends TestMVStore {
if (len > 1024 * 1024) {
// slow down writing a lot
Thread.sleep(200);
} else if (len > 1024 * 100) {
} else if (len > 20 * 1024) {
// slow down writing
Thread.sleep(20);
}
......
h2/src/test/org/h2/test/store/TestMVStore.java
浏览文件 @ 8a5415bb
......@@ -547,6 +547,7 @@ public class TestMVStore extends TestBase {
MVMap<Integer, String> map;
s = new MVStore.Builder().
fileName(fileName).
writeDelay(-1).
compressData().open();
map = s.openMap("test");
// add 10 MB of data
......@@ -556,7 +557,7 @@ public class TestMVStore extends TestBase {
s.store();
s.close();
int[] expectedReadsForCacheSize = {
3405, 2590, 1924, 1440, 1108, 956, 918
3405, 2590, 1924, 1440, 1103, 956, 918
};
for (int cacheSize = 0; cacheSize <= 6; cacheSize += 4) {
s = new MVStore.Builder().
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论