提交 5f5eb72d authored 作者: Thomas Mueller's avatar Thomas Mueller

MVStore: simpler API

上级 7cb0c93e
...@@ -847,7 +847,7 @@ public class MVMap<K, V> extends AbstractMap<K, V> ...@@ -847,7 +847,7 @@ public class MVMap<K, V> extends AbstractMap<K, V>
* Forget those old versions that are no longer needed. * Forget those old versions that are no longer needed.
*/ */
void removeUnusedOldVersions() { void removeUnusedOldVersions() {
long oldest = store.getRetainOrStoreVersion(); long oldest = store.getOldestVersionToKeep();
if (oldest == -1) { if (oldest == -1) {
return; return;
} }
......
...@@ -96,7 +96,7 @@ public class MVSecondaryIndex extends BaseIndex { ...@@ -96,7 +96,7 @@ public class MVSecondaryIndex extends BaseIndex {
} }
} }
try { try {
map.put(array, ValueLong.get(0)); map.put(array, ValueNull.INSTANCE);
} catch (IllegalStateException e) { } catch (IllegalStateException e) {
throw DbException.get(ErrorCode.CONCURRENT_UPDATE_1, table.getName()); throw DbException.get(ErrorCode.CONCURRENT_UPDATE_1, table.getName());
} }
......
...@@ -58,6 +58,7 @@ public class MVSpatialIndex extends BaseIndex implements SpatialIndex { ...@@ -58,6 +58,7 @@ public class MVSpatialIndex extends BaseIndex implements SpatialIndex {
/** /**
* Constructor. * Constructor.
* *
* @param db the database
* @param table the table instance * @param table the table instance
* @param id the index id * @param id the index id
* @param indexName the index name * @param indexName the index name
......
...@@ -168,13 +168,13 @@ public class MVTableEngine implements TableEngine { ...@@ -168,13 +168,13 @@ public class MVTableEngine implements TableEngine {
/** /**
* Store all pending changes. * Store all pending changes.
*/ */
public void store() { public void flush() {
FileStore s = store.getFileStore(); FileStore s = store.getFileStore();
if (s == null || s.isReadOnly()) { if (s == null || s.isReadOnly()) {
return; return;
} }
if (!store.compact(50)) { if (!store.compact(50)) {
store.store(); store.commit();
} }
} }
...@@ -213,9 +213,7 @@ public class MVTableEngine implements TableEngine { ...@@ -213,9 +213,7 @@ public class MVTableEngine implements TableEngine {
Transaction t = session.getTransaction(); Transaction t = session.getTransaction();
t.setName(transactionName); t.setName(transactionName);
t.prepare(); t.prepare();
if (store.getFileStore() != null) { store.commit();
store.store();
}
} }
public ArrayList<InDoubtTransaction> getInDoubtTransactions() { public ArrayList<InDoubtTransaction> getInDoubtTransactions() {
...@@ -245,7 +243,7 @@ public class MVTableEngine implements TableEngine { ...@@ -245,7 +243,7 @@ public class MVTableEngine implements TableEngine {
* Force the changes to disk. * Force the changes to disk.
*/ */
public void sync() { public void sync() {
store(); flush();
store.sync(); store.sync();
} }
...@@ -316,9 +314,7 @@ public class MVTableEngine implements TableEngine { ...@@ -316,9 +314,7 @@ public class MVTableEngine implements TableEngine {
} else { } else {
transaction.rollback(); transaction.rollback();
} }
if (store.getFileStore() != null) { store.commit();
store.store();
}
this.state = state; this.state = state;
} }
......
...@@ -170,9 +170,7 @@ public class TransactionStore { ...@@ -170,9 +170,7 @@ public class TransactionStore {
public synchronized void close() { public synchronized void close() {
// to avoid losing transaction ids // to avoid losing transaction ids
settings.put(LAST_TRANSACTION_ID, "" + lastTransactionId); settings.put(LAST_TRANSACTION_ID, "" + lastTransactionId);
if (store.getFileStore() != null) { store.commit();
store.store();
}
} }
/** /**
...@@ -192,9 +190,7 @@ public class TransactionStore { ...@@ -192,9 +190,7 @@ public class TransactionStore {
private void commitIfNeeded() { private void commitIfNeeded() {
if (store.getUnsavedPageCount() > MAX_UNSAVED_PAGES) { if (store.getUnsavedPageCount() > MAX_UNSAVED_PAGES) {
if (store.getFileStore() != null) { store.commit();
store.store();
}
} }
} }
...@@ -354,10 +350,8 @@ public class TransactionStore { ...@@ -354,10 +350,8 @@ public class TransactionStore {
if (t.getId() == firstOpenTransaction) { if (t.getId() == firstOpenTransaction) {
firstOpenTransaction = -1; firstOpenTransaction = -1;
} }
if (store.getWriteDelay() == 0) { if (store.getAutoCommitDelay() == 0) {
if (store.getFileStore() != null) { store.commit();
store.store();
}
return; return;
} }
// to avoid having to store the transaction log, // to avoid having to store the transaction log,
...@@ -365,10 +359,10 @@ public class TransactionStore { ...@@ -365,10 +359,10 @@ public class TransactionStore {
// and if there have been many changes, store them now // and if there have been many changes, store them now
if (undoLog.isEmpty()) { if (undoLog.isEmpty()) {
int unsaved = store.getUnsavedPageCount(); int unsaved = store.getUnsavedPageCount();
int max = store.getUnsavedPageCountMax(); int max = store.getAutoCommitPageCount();
// save at 3/4 capacity // save at 3/4 capacity
if (unsaved * 4 > max * 3) { if (unsaved * 4 > max * 3) {
store.store(); store.commit();
} }
} }
} }
...@@ -659,6 +653,14 @@ public class TransactionStore { ...@@ -659,6 +653,14 @@ public class TransactionStore {
return new TransactionMap<K, V>(this, map, mapId); return new TransactionMap<K, V>(this, map, mapId);
} }
/**
* Open the transactional version of the given map.
*
* @param <K> the key type
* @param <V> the value type
* @param map the base map
* @return the transactional map
*/
public <K, V> TransactionMap<K, V> openMap(MVMap<K, VersionedValue> map) { public <K, V> TransactionMap<K, V> openMap(MVMap<K, VersionedValue> map) {
checkNotClosed(); checkNotClosed();
int mapId = map.getId(); int mapId = map.getId();
...@@ -729,6 +731,8 @@ public class TransactionStore { ...@@ -729,6 +731,8 @@ public class TransactionStore {
/** /**
* Remove the map. * Remove the map.
*
* @param map the map
*/ */
public <K, V> void removeMap(TransactionMap<K, V> map) { public <K, V> void removeMap(TransactionMap<K, V> map) {
store.store.removeMap(map.map); store.store.removeMap(map.map);
......
...@@ -62,20 +62,19 @@ public class TestConcurrent extends TestMVStore { ...@@ -62,20 +62,19 @@ public class TestConcurrent extends TestMVStore {
for (int test = 0; test < 10; test++) { for (int test = 0; test < 10; test++) {
FileUtils.delete(fileName); FileUtils.delete(fileName);
final MVStore s1 = new MVStore.Builder(). final MVStore s1 = new MVStore.Builder().
fileName(fileName).writeDelay(-1).open(); fileName(fileName).autoCommitDisabled().open();
s1.setRetentionTime(0); s1.setRetentionTime(0);
final int count = 200; final int count = 200;
for (int i = 0; i < count; i++) { for (int i = 0; i < count; i++) {
MVMap<Integer, Integer> m = s1.openMap("d" + i); MVMap<Integer, Integer> m = s1.openMap("d" + i);
m.put(1, 1); m.put(1, 1);
if (i % 2 == 0) { if (i % 2 == 0) {
s1.store(); s1.commit();
} }
} }
s1.store();
s1.close(); s1.close();
final MVStore s = new MVStore.Builder(). final MVStore s = new MVStore.Builder().
fileName(fileName).writeDelay(-1).open(); fileName(fileName).autoCommitDisabled().open();
s.setRetentionTime(0); s.setRetentionTime(0);
final ArrayList<MVMap<Integer, Integer>> list = New.arrayList(); final ArrayList<MVMap<Integer, Integer>> list = New.arrayList();
for (int i = 0; i < count; i++) { for (int i = 0; i < count; i++) {
...@@ -113,7 +112,7 @@ public class TestConcurrent extends TestMVStore { ...@@ -113,7 +112,7 @@ public class TestConcurrent extends TestMVStore {
} }
} }
task.get(); task.get();
s.store(); s.commit();
MVMap<String, String> meta = s.getMetaMap(); MVMap<String, String> meta = s.getMetaMap();
int chunkCount = 0; int chunkCount = 0;
...@@ -142,7 +141,7 @@ public class TestConcurrent extends TestMVStore { ...@@ -142,7 +141,7 @@ public class TestConcurrent extends TestMVStore {
public void call() throws Exception { public void call() throws Exception {
while (!stop) { while (!stop) {
counter.incrementAndGet(); counter.incrementAndGet();
s.store(); s.commit();
} }
} }
}; };
...@@ -172,7 +171,7 @@ public class TestConcurrent extends TestMVStore { ...@@ -172,7 +171,7 @@ public class TestConcurrent extends TestMVStore {
public void call() throws Exception { public void call() throws Exception {
while (!stop) { while (!stop) {
s.setStoreVersion(counter.incrementAndGet()); s.setStoreVersion(counter.incrementAndGet());
s.store(); s.commit();
} }
} }
}; };
...@@ -270,9 +269,9 @@ public class TestConcurrent extends TestMVStore { ...@@ -270,9 +269,9 @@ public class TestConcurrent extends TestMVStore {
for (int i = 0; i < 10; i++) { for (int i = 0; i < 10; i++) {
map.put(i, new byte[100 * r.nextInt(100)]); map.put(i, new byte[100 * r.nextInt(100)]);
} }
s.store(); s.commit();
map.clear(); map.clear();
s.store(); s.commit();
long len = s.getFileStore().size(); long len = s.getFileStore().size();
if (len > 1024 * 1024) { if (len > 1024 * 1024) {
// slow down writing a lot // slow down writing a lot
...@@ -322,6 +321,7 @@ public class TestConcurrent extends TestMVStore { ...@@ -322,6 +321,7 @@ public class TestConcurrent extends TestMVStore {
private void testConcurrentIterate() { private void testConcurrentIterate() {
MVStore s = new MVStore.Builder().pageSplitSize(3).open(); MVStore s = new MVStore.Builder().pageSplitSize(3).open();
s.setVersionsToKeep(100);
final MVMap<Integer, Integer> map = s.openMap("test"); final MVMap<Integer, Integer> map = s.openMap("test");
final int len = 10; final int len = 10;
final Random r = new Random(); final Random r = new Random();
...@@ -343,7 +343,6 @@ public class TestConcurrent extends TestMVStore { ...@@ -343,7 +343,6 @@ public class TestConcurrent extends TestMVStore {
Iterator<Integer> it = map.keyIterator(r.nextInt(len)); Iterator<Integer> it = map.keyIterator(r.nextInt(len));
long old = s.getCurrentVersion(); long old = s.getCurrentVersion();
s.commit(); s.commit();
s.setRetainVersion(old - 100);
while (map.getVersion() == old) { while (map.getVersion() == old) {
Thread.yield(); Thread.yield();
} }
......
...@@ -84,9 +84,8 @@ public class TestKillProcessWhileWriting extends TestBase { ...@@ -84,9 +84,8 @@ public class TestKillProcessWhileWriting extends TestBase {
s = new MVStore.Builder(). s = new MVStore.Builder().
fileName(fileName). fileName(fileName).
pageSplitSize(50). pageSplitSize(50).
writeDelay(0). autoCommitDisabled().
open(); open();
s.setWriteDelay(0);
m = s.openMap("data"); m = s.openMap("data");
Random r = new Random(seed); Random r = new Random(seed);
int op = 0; int op = 0;
...@@ -107,7 +106,7 @@ public class TestKillProcessWhileWriting extends TestBase { ...@@ -107,7 +106,7 @@ public class TestKillProcessWhileWriting extends TestBase {
m.remove(k); m.remove(k);
break; break;
case 6: case 6:
s.store(); s.commit();
break; break;
case 7: case 7:
s.compact(80); s.compact(80);
...@@ -120,12 +119,12 @@ public class TestKillProcessWhileWriting extends TestBase { ...@@ -120,12 +119,12 @@ public class TestKillProcessWhileWriting extends TestBase {
s = new MVStore.Builder(). s = new MVStore.Builder().
fileName(fileName). fileName(fileName).
pageSplitSize(50). pageSplitSize(50).
writeDelay(0).open(); autoCommitDisabled().
open();
m = s.openMap("data"); m = s.openMap("data");
break; break;
} }
} }
s.store();
s.close(); s.close();
return 0; return 0;
} catch (Exception e) { } catch (Exception e) {
......
...@@ -119,7 +119,7 @@ public class TestMVRTree extends TestMVStore { ...@@ -119,7 +119,7 @@ public class TestMVRTree extends TestMVStore {
SpatialKey k = new SpatialKey(i, x - p, x + p, y - p, y + p); SpatialKey k = new SpatialKey(i, x - p, x + p, y - p, y + p);
r.add(k, "" + i); r.add(k, "" + i);
if (i > 0 && (i % len / 10) == 0) { if (i > 0 && (i % len / 10) == 0) {
s.store(); s.commit();
} }
if (i > 0 && (i % 10000) == 0) { if (i > 0 && (i % 10000) == 0) {
render(r, getBaseDir() + "/test.png"); render(r, getBaseDir() + "/test.png");
...@@ -127,7 +127,6 @@ public class TestMVRTree extends TestMVStore { ...@@ -127,7 +127,6 @@ public class TestMVRTree extends TestMVStore {
} }
// System.out.println(prof.getTop(5)); // System.out.println(prof.getTop(5));
// System.out.println("add: " + (System.currentTimeMillis() - t)); // System.out.println("add: " + (System.currentTimeMillis() - t));
s.store();
s.close(); s.close();
s = openStore(fileName); s = openStore(fileName);
r = s.openMap("data", r = s.openMap("data",
......
...@@ -92,7 +92,7 @@ public class TestRandomMapOps extends TestBase { ...@@ -92,7 +92,7 @@ public class TestRandomMapOps extends TestBase {
for (; op < size; op++) { for (; op < size; op++) {
int k = r.nextInt(100); int k = r.nextInt(100);
byte[] v = new byte[r.nextInt(10) * 10]; byte[] v = new byte[r.nextInt(10) * 10];
int type = r.nextInt(13); int type = r.nextInt(12);
switch (type) { switch (type) {
case 0: case 0:
case 1: case 1:
...@@ -109,23 +109,19 @@ public class TestRandomMapOps extends TestBase { ...@@ -109,23 +109,19 @@ public class TestRandomMapOps extends TestBase {
map.remove(k); map.remove(k);
break; break;
case 6: case 6:
log(op, k, v, "s.store()");
s.store();
break;
case 7:
log(op, k, v, "s.compact(90)"); log(op, k, v, "s.compact(90)");
s.compact(90); s.compact(90);
break; break;
case 8: case 7:
log(op, k, v, "m.clear()"); log(op, k, v, "m.clear()");
m.clear(); m.clear();
map.clear(); map.clear();
break; break;
case 9: case 8:
log(op, k, v, "s.commit()"); log(op, k, v, "s.commit()");
s.commit(); s.commit();
break; break;
case 10: case 9:
log(op, k, v, "s.commit()"); log(op, k, v, "s.commit()");
s.commit(); s.commit();
log(op, k, v, "s.close()"); log(op, k, v, "s.close()");
...@@ -135,13 +131,13 @@ public class TestRandomMapOps extends TestBase { ...@@ -135,13 +131,13 @@ public class TestRandomMapOps extends TestBase {
log(op, k, v, "m = s.openMap(\"data\")"); log(op, k, v, "m = s.openMap(\"data\")");
m = s.openMap("data"); m = s.openMap("data");
break; break;
case 11: case 10:
log(op, k, v, "s.commit()"); log(op, k, v, "s.commit()");
s.commit(); s.commit();
log(op, k, v, "s.compactMoveChunks()"); log(op, k, v, "s.compactMoveChunks()");
s.compactMoveChunks(); s.compactMoveChunks();
break; break;
case 12: case 11:
log(op, k, v, "m.getKeyIndex({0})"); log(op, k, v, "m.getKeyIndex({0})");
ArrayList<Integer> keyList = new ArrayList<Integer>(map.keySet()); ArrayList<Integer> keyList = new ArrayList<Integer>(map.keySet());
int index = Collections.binarySearch(keyList, k, null); int index = Collections.binarySearch(keyList, k, null);
...@@ -165,13 +161,12 @@ public class TestRandomMapOps extends TestBase { ...@@ -165,13 +161,12 @@ public class TestRandomMapOps extends TestBase {
assertEquals(map.lastKey(), m.lastKey()); assertEquals(map.lastKey(), m.lastKey());
} }
} }
s.store();
s.close(); s.close();
} }
private static MVStore openStore(String fileName) { private static MVStore openStore(String fileName) {
MVStore s = new MVStore.Builder().fileName(fileName). MVStore s = new MVStore.Builder().fileName(fileName).
pageSplitSize(50).writeDelay(0).open(); pageSplitSize(50).autoCommitDisabled().open();
s.setRetentionTime(0); s.setRetentionTime(0);
return s; return s;
} }
......
...@@ -65,7 +65,7 @@ public class TestStreamStore extends TestBase { ...@@ -65,7 +65,7 @@ public class TestStreamStore extends TestBase {
for (int i = 0; i < 100; i++) { for (int i = 0; i < 100; i++) {
streamStore.put(new RandomStream(size, i)); streamStore.put(new RandomStream(size, i));
} }
s.store(); s.commit();
MVMap<Long, byte[]> map = s.openMap("data"); MVMap<Long, byte[]> map = s.openMap("data");
assertTrue("size: " + map.size(), map.sizeAsLong() >= 100); assertTrue("size: " + map.size(), map.sizeAsLong() >= 100);
s.close(); s.close();
...@@ -77,7 +77,7 @@ public class TestStreamStore extends TestBase { ...@@ -77,7 +77,7 @@ public class TestStreamStore extends TestBase {
for (int i = 0; i < 100; i++) { for (int i = 0; i < 100; i++) {
streamStore.put(new RandomStream(size, -i)); streamStore.put(new RandomStream(size, -i));
} }
s.store(); s.commit();
long readCount = s.getFileStore().getReadCount(); long readCount = s.getFileStore().getReadCount();
// the read count should be low because new blocks // the read count should be low because new blocks
// are appended at the end (not between existing blocks) // are appended at the end (not between existing blocks)
...@@ -92,7 +92,7 @@ public class TestStreamStore extends TestBase { ...@@ -92,7 +92,7 @@ public class TestStreamStore extends TestBase {
return new StreamStore(map) { return new StreamStore(map) {
@Override @Override
protected void onStore(int len) { protected void onStore(int len) {
if (s.getUnsavedPageCount() > s.getUnsavedPageCountMax() / 2) { if (s.getUnsavedPageCount() > s.getAutoCommitPageCount() / 2) {
s.commit(); s.commit();
} }
} }
...@@ -111,12 +111,11 @@ public class TestStreamStore extends TestBase { ...@@ -111,12 +111,11 @@ public class TestStreamStore extends TestBase {
@Override @Override
protected void onStore(int len) { protected void onStore(int len) {
count.incrementAndGet(); count.incrementAndGet();
s.store(); s.commit();
} }
}; };
long size = 1 * 1024 * 1024; long size = 1 * 1024 * 1024;
streamStore.put(new RandomStream(size, 0)); streamStore.put(new RandomStream(size, 0));
s.store();
s.close(); s.close();
assertEquals(4, count.get()); assertEquals(4, count.get());
} }
......
...@@ -86,7 +86,7 @@ public class TestTransactionStore extends TestBase { ...@@ -86,7 +86,7 @@ public class TestTransactionStore extends TestBase {
for (int i = 0; !stop; i++) { for (int i = 0; !stop; i++) {
state.set(i); state.set(i);
other.put(i, value); other.put(i, value);
store.store(); store.commit();
} }
} }
}; };
......
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论