提交 2ea52348 authored 作者: Thomas Mueller's avatar Thomas Mueller

MVStore: use a separate write version for every map, so that the meta map…

MVStore: use a separate write version for every map, so that the meta map version matches the version of the other maps.
上级 bae98769
...@@ -38,6 +38,11 @@ public class MVMap<K, V> extends AbstractMap<K, V> ...@@ -38,6 +38,11 @@ public class MVMap<K, V> extends AbstractMap<K, V>
* The current root page (may not be null). * The current root page (may not be null).
*/ */
protected volatile Page root; protected volatile Page root;
/**
* The version used for writing.
*/
protected long writeVersion;
private int id; private int id;
private long createVersion; private long createVersion;
...@@ -70,6 +75,7 @@ public class MVMap<K, V> extends AbstractMap<K, V> ...@@ -70,6 +75,7 @@ public class MVMap<K, V> extends AbstractMap<K, V>
this.id = Integer.parseInt(config.get("id")); this.id = Integer.parseInt(config.get("id"));
String x = config.get("createVersion"); String x = config.get("createVersion");
this.createVersion = x == null ? 0 : Long.parseLong(x); this.createVersion = x == null ? 0 : Long.parseLong(x);
this.writeVersion = store.getCurrentVersion();
} }
/** /**
...@@ -100,10 +106,10 @@ public class MVMap<K, V> extends AbstractMap<K, V> ...@@ -100,10 +106,10 @@ public class MVMap<K, V> extends AbstractMap<K, V>
DataUtils.checkArgument(value != null, "The value may not be null"); DataUtils.checkArgument(value != null, "The value may not be null");
beforeWrite(); beforeWrite();
try { try {
long writeVersion = store.getCurrentVersion(); long v = writeVersion;
Page p = copyOnWrite(root, writeVersion); Page p = copyOnWrite(root, v);
p = splitRootIfNeeded(p, writeVersion); p = splitRootIfNeeded(p, v);
Object result = put(p, writeVersion, key, value); Object result = put(p, v, key, value);
newRoot(p); newRoot(p);
return (V) result; return (V) result;
} finally { } finally {
...@@ -496,7 +502,7 @@ public class MVMap<K, V> extends AbstractMap<K, V> ...@@ -496,7 +502,7 @@ public class MVMap<K, V> extends AbstractMap<K, V>
beforeWrite(); beforeWrite();
try { try {
root.removeAllRecursive(); root.removeAllRecursive();
newRoot(Page.createEmpty(this, store.getCurrentVersion())); newRoot(Page.createEmpty(this, writeVersion));
} finally { } finally {
afterWrite(); afterWrite();
} }
...@@ -544,10 +550,10 @@ public class MVMap<K, V> extends AbstractMap<K, V> ...@@ -544,10 +550,10 @@ public class MVMap<K, V> extends AbstractMap<K, V>
public V remove(Object key) { public V remove(Object key) {
beforeWrite(); beforeWrite();
try { try {
long writeVersion = store.getCurrentVersion(); long v = writeVersion;
Page p = copyOnWrite(root, writeVersion); Page p = copyOnWrite(root, v);
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
V result = (V) remove(p, writeVersion, key); V result = (V) remove(p, v, key);
newRoot(p); newRoot(p);
return result; return result;
} finally { } finally {
...@@ -1014,7 +1020,7 @@ public class MVMap<K, V> extends AbstractMap<K, V> ...@@ -1014,7 +1020,7 @@ public class MVMap<K, V> extends AbstractMap<K, V>
// need to copy because it can change // need to copy because it can change
Page r = root; Page r = root;
if (version >= r.getVersion() && if (version >= r.getVersion() &&
(version == store.getCurrentVersion() || (version == writeVersion ||
r.getVersion() >= 0 || r.getVersion() >= 0 ||
version <= createVersion || version <= createVersion ||
store.getFile() == null)) { store.getFile() == null)) {
...@@ -1128,6 +1134,10 @@ public class MVMap<K, V> extends AbstractMap<K, V> ...@@ -1128,6 +1134,10 @@ public class MVMap<K, V> extends AbstractMap<K, V>
afterWrite(); afterWrite();
} }
} }
void setWriteVersion(long writeVersion) {
this.writeVersion = writeVersion;
}
@Override @Override
public String toString() { public String toString() {
......
...@@ -46,11 +46,11 @@ public class MVMapConcurrent<K, V> extends MVMap<K, V> { ...@@ -46,11 +46,11 @@ public class MVMapConcurrent<K, V> extends MVMap<K, V> {
// even if the result is the same, we still update the value // even if the result is the same, we still update the value
// (otherwise compact doesn't work) // (otherwise compact doesn't work)
get(key); get(key);
long writeVersion = store.getCurrentVersion(); long v = writeVersion;
synchronized (this) { synchronized (this) {
Page p = copyOnWrite(root, writeVersion); Page p = copyOnWrite(root, v);
p = splitRootIfNeeded(p, writeVersion); p = splitRootIfNeeded(p, v);
V result = (V) put(p, writeVersion, key, value); V result = (V) put(p, v, key, value);
newRoot(p); newRoot(p);
return result; return result;
} }
...@@ -73,10 +73,10 @@ public class MVMapConcurrent<K, V> extends MVMap<K, V> { ...@@ -73,10 +73,10 @@ public class MVMapConcurrent<K, V> extends MVMap<K, V> {
if (result == null) { if (result == null) {
return null; return null;
} }
long writeVersion = store.getCurrentVersion(); long v = writeVersion;
synchronized (this) { synchronized (this) {
Page p = copyOnWrite(root, writeVersion); Page p = copyOnWrite(root, v);
result = (V) remove(p, writeVersion, key); result = (V) remove(p, v, key);
newRoot(p); newRoot(p);
} }
return result; return result;
......
...@@ -913,6 +913,11 @@ public class TransactionStore { ...@@ -913,6 +913,11 @@ public class TransactionStore {
newValue.value = value; newValue.value = value;
if (current == null) { if (current == null) {
// a new value // a new value
int todo;
// either write the log before the data (and handle this case in rollback)
// or ensure/document concurrent commits are not allowed
VersionedValue old = map.putIfAbsent(key, newValue); VersionedValue old = map.putIfAbsent(key, newValue);
if (old == null) { if (old == null) {
transaction.log(opType, mapId, key, current); transaction.log(opType, mapId, key, current);
......
...@@ -223,30 +223,30 @@ public class MVRTreeMap<V> extends MVMap<SpatialKey, V> { ...@@ -223,30 +223,30 @@ public class MVRTreeMap<V> extends MVMap<SpatialKey, V> {
private Object putOrAdd(SpatialKey key, V value, boolean alwaysAdd) { private Object putOrAdd(SpatialKey key, V value, boolean alwaysAdd) {
beforeWrite(); beforeWrite();
try { try {
long writeVersion = store.getCurrentVersion(); long v = writeVersion;
Page p = copyOnWrite(root, writeVersion); Page p = copyOnWrite(root, v);
Object result; Object result;
if (alwaysAdd || get(key) == null) { if (alwaysAdd || get(key) == null) {
if (p.getMemory() > store.getPageSize() && p.getKeyCount() > 1) { if (p.getMemory() > store.getPageSize() && p.getKeyCount() > 1) {
// only possible if this is the root, else we would have split earlier // only possible if this is the root, else we would have split earlier
// (this requires maxPageSize is fixed) // (this requires maxPageSize is fixed)
long totalCount = p.getTotalCount(); long totalCount = p.getTotalCount();
Page split = split(p, writeVersion); Page split = split(p, v);
Object k1 = getBounds(p); Object k1 = getBounds(p);
Object k2 = getBounds(split); Object k2 = getBounds(split);
Object[] keys = { k1, k2 }; Object[] keys = { k1, k2 };
long[] children = { p.getPos(), split.getPos(), 0 }; long[] children = { p.getPos(), split.getPos(), 0 };
Page[] childrenPages = { p, split, null }; Page[] childrenPages = { p, split, null };
long[] counts = { p.getTotalCount(), split.getTotalCount(), 0 }; long[] counts = { p.getTotalCount(), split.getTotalCount(), 0 };
p = Page.create(this, writeVersion, 2, p = Page.create(this, v, 2,
keys, null, children, childrenPages, counts, keys, null, children, childrenPages, counts,
totalCount, 0, 0); totalCount, 0, 0);
// now p is a node; continues // now p is a node; continues
} }
add(p, writeVersion, key, value); add(p, v, key, value);
result = null; result = null;
} else { } else {
result = set(p, writeVersion, key, value); result = set(p, v, key, value);
} }
newRoot(p); newRoot(p);
return result; return result;
......
...@@ -123,7 +123,7 @@ public class TestMVStore extends TestBase { ...@@ -123,7 +123,7 @@ public class TestMVStore extends TestBase {
assertTrue(e != null); assertTrue(e != null);
assertEquals(DataUtils.ERROR_WRITING_FAILED, DataUtils.getErrorCode(e.getMessage())); assertEquals(DataUtils.ERROR_WRITING_FAILED, DataUtils.getErrorCode(e.getMessage()));
s.close(); s.closeImmediately();
FileUtils.delete(fileName); FileUtils.delete(fileName);
} }
...@@ -237,7 +237,14 @@ public class TestMVStore extends TestBase { ...@@ -237,7 +237,14 @@ public class TestMVStore extends TestBase {
m.put(2, "World"); m.put(2, "World");
Thread.sleep(5); Thread.sleep(5);
// must not store, as nothing has been committed yet // must not store, as nothing has been committed yet
assertEquals(v, s.getCurrentVersion()); s.closeImmediately();
s = new MVStore.Builder().
writeDelay(1).
fileName(fileName).
open();
m = s.openMap("data");
assertEquals(null, m.get(2));
m.put(2, "World");
s.commit(); s.commit();
m.put(3, "!"); m.put(3, "!");
...@@ -251,20 +258,7 @@ public class TestMVStore extends TestBase { ...@@ -251,20 +258,7 @@ public class TestMVStore extends TestBase {
Thread.sleep(1); Thread.sleep(1);
} }
s.close(); s.close();
s = new MVStore.Builder().
fileName(fileName).
open();
m = s.openMap("data");
assertEquals("Hello", m.get(1));
assertEquals("World", m.get(2));
assertFalse(m.containsKey(3));
String data = new String(new char[1000]).replace((char) 0, 'x');
for (int i = 0; i < 1000; i++) {
m.put(i, data);
}
s.close();
s = new MVStore.Builder(). s = new MVStore.Builder().
fileName(fileName). fileName(fileName).
open(); open();
...@@ -472,13 +466,21 @@ public class TestMVStore extends TestBase { ...@@ -472,13 +466,21 @@ public class TestMVStore extends TestBase {
s.close(); s.close();
} }
private void testFileHeaderCorruption() throws IOException { private void testFileHeaderCorruption() throws Exception {
String fileName = getBaseDir() + "/testFileHeader.h3"; String fileName = getBaseDir() + "/testFileHeader.h3";
MVStore s = openStore(fileName); MVStore s = openStore(fileName);
s.setRetentionTime(10);
MVMap<Integer, Integer> map = s.openMap("test"); MVMap<Integer, Integer> map = s.openMap("test");
for (int i = 0; i < 5; i++) {
s.setStoreVersion(i);
s.store();
}
// ensure the oldest chunks can be overwritten
Thread.sleep(11);
s.compact(50);
map.put(10, 100); map.put(10, 100);
FilePath f = FilePath.get(s.getFileName());
s.store(); s.store();
FilePath f = FilePath.get(s.getFileName());
s.close(); s.close();
int blockSize = 4 * 1024; int blockSize = 4 * 1024;
// test corrupt file headers // test corrupt file headers
...@@ -652,13 +654,13 @@ public class TestMVStore extends TestBase { ...@@ -652,13 +654,13 @@ public class TestMVStore extends TestBase {
s.setStoreVersion(1); s.setStoreVersion(1);
s.close(); s.close();
s = MVStore.open(fileName); s = MVStore.open(fileName);
assertEquals(0, s.getCurrentVersion()); assertEquals(1, s.getCurrentVersion());
assertEquals(0, s.getStoreVersion()); assertEquals(0, s.getStoreVersion());
s.setStoreVersion(1); s.setStoreVersion(1);
s.store(); s.store();
s.close(); s.close();
s = MVStore.open(fileName); s = MVStore.open(fileName);
assertEquals(1, s.getCurrentVersion()); assertEquals(2, s.getCurrentVersion());
assertEquals(1, s.getStoreVersion()); assertEquals(1, s.getStoreVersion());
s.close(); s.close();
} }
...@@ -698,6 +700,7 @@ public class TestMVStore extends TestBase { ...@@ -698,6 +700,7 @@ public class TestMVStore extends TestBase {
map.put(new Object[1], new Object[]{1, "2"}); map.put(new Object[1], new Object[]{1, "2"});
s.store(); s.store();
s.close(); s.close();
s = new MVStore.Builder().fileName(fileName).open(); s = new MVStore.Builder().fileName(fileName).open();
map = s.openMap("test"); map = s.openMap("test");
assertEquals("Hello", map.get(1).toString()); assertEquals("Hello", map.get(1).toString());
...@@ -938,6 +941,8 @@ public class TestMVStore extends TestBase { ...@@ -938,6 +941,8 @@ public class TestMVStore extends TestBase {
s.rollbackTo(1); s.rollbackTo(1);
assertEquals(1, s.getCurrentVersion()); assertEquals(1, s.getCurrentVersion());
assertEquals("Hello", m.get("1")); assertEquals("Hello", m.get("1"));
// so a new version is created
m.put("1", "Hello");
long v2 = s.store(); long v2 = s.store();
assertEquals(2, v2); assertEquals(2, v2);
......
...@@ -441,7 +441,7 @@ public class TestTransactionStore extends TestBase { ...@@ -441,7 +441,7 @@ public class TestTransactionStore extends TestBase {
return; return;
} }
statements.get(0).execute( statements.get(0).execute(
"drop table if exists test"); "drop table if exists test cascade");
statements.get(0).execute( statements.get(0).execute(
"create table test(id int primary key, name varchar(255))"); "create table test(id int primary key, name varchar(255))");
......
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论