提交 e8f0a660 authored 作者: Thomas Mueller's avatar Thomas Mueller

MVStore: various changes (store temporary, store committed and rollback on close)

上级 cfb047bd
...@@ -72,7 +72,7 @@ public class Chunk { ...@@ -72,7 +72,7 @@ public class Chunk {
long version; long version;
/** /**
* When this chunk was created, in seconds after the store was created. * When this chunk was created, in milliseconds after the store was created.
*/ */
long time; long time;
......
...@@ -48,8 +48,10 @@ public class MVMap<K, V> extends AbstractMap<K, V> ...@@ -48,8 +48,10 @@ public class MVMap<K, V> extends AbstractMap<K, V>
private boolean closed; private boolean closed;
private boolean readOnly; private boolean readOnly;
/**
* This flag is set during a write operation to the tree.
*/
private volatile boolean writing; private volatile boolean writing;
private volatile int writeCount;
protected MVMap(DataType keyType, DataType valueType) { protected MVMap(DataType keyType, DataType valueType) {
this.keyType = keyType; this.keyType = keyType;
...@@ -918,26 +920,21 @@ public class MVMap<K, V> extends AbstractMap<K, V> ...@@ -918,26 +920,21 @@ public class MVMap<K, V> extends AbstractMap<K, V>
} }
/** /**
* This method is called after writing to the map. * This method is called after writing to the map (whether or not the write
* operation was successful).
*/ */
protected void afterWrite() { protected void afterWrite() {
writeCount++;
writing = false; writing = false;
} }
void waitUntilWritten(long version) { /**
if (root.getVersion() < version) { * If there is a concurrent update to the given version, wait until it is
// a write will create a new version * finished.
return; *
} * @param root the root page
// wait until writing is done, */
// but only for the current write operation protected void waitUntilWritten(Page root) {
// a bit like a spin lock while (writing && root == this.root) {
int w = writeCount;
while (writing) {
if (writeCount > w) {
return;
}
Thread.yield(); Thread.yield();
} }
} }
...@@ -967,7 +964,7 @@ public class MVMap<K, V> extends AbstractMap<K, V> ...@@ -967,7 +964,7 @@ public class MVMap<K, V> extends AbstractMap<K, V>
/** /**
* Remove the given page (make the space available). * Remove the given page (make the space available).
* *
* @param p the page * @param pos the position of the page to remove
*/ */
protected void removePage(long pos) { protected void removePage(long pos) {
store.removePage(this, pos); store.removePage(this, pos);
......
...@@ -52,7 +52,7 @@ public class MVMapConcurrent<K, V> extends MVMap<K, V> { ...@@ -52,7 +52,7 @@ public class MVMapConcurrent<K, V> extends MVMap<K, V> {
} }
} }
void waitUntilWritten(long version) { protected void waitUntilWritten(Page root) {
// no need to wait // no need to wait
} }
......
...@@ -43,16 +43,16 @@ H:3,... ...@@ -43,16 +43,16 @@ H:3,...
TODO: TODO:
- getTime: use milliseconds, not seconds (no division, finer granurality) - test new write / read algorithm for speed and errors
- naming: hasUnsavedChanges() versus store(): hasUnstoredChanges? - detect concurrent writes / reads in the MVMap
- test rollback of meta table: it is changed after save; could rollback be a problem? - maybe rename store to write
- async store: write test cases; should fail at freedChunks - document how committing, storing, and closing is coupled
- async store of current root is illegal, except with MVMapConcurrent - document temporary writes (to avoid out-of-memory)
- auto-store needs to be reverted on startup - store() should probably be store(false), and maybe rename to write
- auto-store synchronously if too many unstored pages (8 MB) - move setters to the builder, except for setRetainVersion, setReuseSpace,
- auto-store in background thread after 1-2 second by default and settings that are persistent (setStoreVersion)
- auto-store in background thread if more than 4 MB of unstored pages - update copyright
- auto-store: use notify to wake up background thread? - test meta table rollback: it is changed after save; could rollback break it?
- automated 'kill process' and 'power failure' test - automated 'kill process' and 'power failure' test
- mvcc with multiple transactions - mvcc with multiple transactions
- update checkstyle - update checkstyle
...@@ -94,6 +94,8 @@ TODO: ...@@ -94,6 +94,8 @@ TODO:
- implement an off-heap file system - implement an off-heap file system
- remove change cursor, or add support for writing to branches - remove change cursor, or add support for writing to branches
- support pluggable logging or remove log - support pluggable logging or remove log
- maybe add an optional finalizer and exit hook
to store committed changes
*/ */
...@@ -116,6 +118,9 @@ public class MVStore { ...@@ -116,6 +118,9 @@ public class MVStore {
private static final int FORMAT_WRITE = 1; private static final int FORMAT_WRITE = 1;
private static final int FORMAT_READ = 1; private static final int FORMAT_READ = 1;
/**
* Whether the store is closed.
*/
volatile boolean closed; volatile boolean closed;
private final String fileName; private final String fileName;
...@@ -172,6 +177,10 @@ public class MVStore { ...@@ -172,6 +177,10 @@ public class MVStore {
private final boolean compress; private final boolean compress;
private long currentVersion; private long currentVersion;
/**
* The version of the last stored chunk.
*/
private long lastStoredVersion; private long lastStoredVersion;
private int fileReadCount; private int fileReadCount;
private int fileWriteCount; private int fileWriteCount;
...@@ -179,12 +188,23 @@ public class MVStore { ...@@ -179,12 +188,23 @@ public class MVStore {
private int maxUnsavedPages; private int maxUnsavedPages;
/** /**
* The time the store was created, in seconds since 1970. * The time the store was created, in milliseconds since 1970.
*/ */
private long creationTime; private long creationTime;
private int retentionTime = 45; private int retentionTime = 45000;
private long lastStoreTime; private long lastStoreTime;
/**
* To which version to roll back when opening the store after a crash.
*/
private long lastCommittedVersion;
/**
* The earliest chunk to retain, if any.
*/
private Chunk retainChunk;
private Thread backgroundThread; private Thread backgroundThread;
/** /**
...@@ -194,6 +214,11 @@ public class MVStore { ...@@ -194,6 +214,11 @@ public class MVStore {
private volatile boolean metaChanged; private volatile boolean metaChanged;
/**
* The delay in milliseconds to automatically store changes.
*/
private int writeDelay = 1000;
MVStore(HashMap<String, Object> config) { MVStore(HashMap<String, Object> config) {
String f = (String) config.get("fileName"); String f = (String) config.get("fileName");
if (f != null && !f.startsWith("nio:")) { if (f != null && !f.startsWith("nio:")) {
...@@ -219,6 +244,8 @@ public class MVStore { ...@@ -219,6 +244,8 @@ public class MVStore {
mb = o == null ? 4 : (Integer) o; mb = o == null ? 4 : (Integer) o;
int writeBufferSize = mb * 1024 * 1024; int writeBufferSize = mb * 1024 * 1024;
maxUnsavedPages = writeBufferSize / pageSize; maxUnsavedPages = writeBufferSize / pageSize;
o = config.get("writeDelay");
writeDelay = o == null ? 1000 : (Integer) o;
} else { } else {
cache = null; cache = null;
filePassword = null; filePassword = null;
...@@ -421,9 +448,16 @@ public class MVStore { ...@@ -421,9 +448,16 @@ public class MVStore {
} }
} }
lastStoreTime = getTime(); lastStoreTime = getTime();
// if we use auto-save, also start the background thread String r = meta.get("rollbackOnOpen");
if (maxUnsavedPages > 0) { if (r != null) {
Writer w = new Writer(this); long rollback = Long.parseLong(r);
rollbackTo(rollback);
}
this.lastCommittedVersion = currentVersion;
// start the background thread if needed
if (writeDelay > 0) {
int sleep = Math.max(1, writeDelay / 10);
Writer w = new Writer(this, sleep);
Thread t = new Thread(w, "MVStore writer " + fileName); Thread t = new Thread(w, "MVStore writer " + fileName);
t.setDaemon(true); t.setDaemon(true);
t.start(); t.start();
...@@ -605,27 +639,44 @@ public class MVStore { ...@@ -605,27 +639,44 @@ public class MVStore {
} }
/** /**
* Close the file. Uncommitted changes are ignored, and all open maps are closed. * Close the file and the store. If there are any committed but unsaved
* changes, they are written to disk first. If any temporary data was
* written but not committed, this is rolled back. All open maps are closed.
*/ */
public void close() { public void close() {
close(true); close(true);
} }
private synchronized void close(boolean shrinkIfPossible) { private void close(boolean shrinkIfPossible) {
if (closed) {
return;
}
if (!readOnly) {
if (hasUnsavedChanges()) {
rollbackTo(lastCommittedVersion);
store(false);
}
}
closed = true; closed = true;
if (file == null) { if (file == null) {
return; return;
} }
// can not synchronize on this yet, because
// the thread also synchronized on this, which
// could result in a deadlock
if (backgroundThread != null) { if (backgroundThread != null) {
Thread t = backgroundThread; Thread t = backgroundThread;
backgroundThread = null; backgroundThread = null;
t.interrupt(); synchronized (this) {
notify();
}
try { try {
t.join(); t.join();
} catch (Exception e) { } catch (Exception e) {
// ignore // ignore
} }
} }
synchronized (this) {
try { try {
if (shrinkIfPossible) { if (shrinkIfPossible) {
shrinkFileIfPossible(0); shrinkFileIfPossible(0);
...@@ -650,6 +701,7 @@ public class MVStore { ...@@ -650,6 +701,7 @@ public class MVStore {
file = null; file = null;
} }
} }
}
/** /**
* Get the chunk for the given position. * Get the chunk for the given position.
...@@ -662,7 +714,7 @@ public class MVStore { ...@@ -662,7 +714,7 @@ public class MVStore {
} }
/** /**
* Increment the current version. * Increment the current version, without committing the changes.
* *
* @return the new version * @return the new version
*/ */
...@@ -670,6 +722,25 @@ public class MVStore { ...@@ -670,6 +722,25 @@ public class MVStore {
return ++currentVersion; return ++currentVersion;
} }
/**
* Commit the changes. This method marks the changes as committed and
* increments the version.
* <p>
* Unless the write delay is disabled, this method does not write to the
* file. Instead, data is written after the delay, manually by calling the
* store method, when the write buffer is full, or when closing the store.
*
* @return the new version
*/
public long commit() {
if (writeDelay == 0) {
return store(true);
}
long v = ++currentVersion;
lastCommittedVersion = v;
return v;
}
/** /**
* Commit all changes and persist them to disk. This method does nothing if * Commit all changes and persist them to disk. This method does nothing if
* there are no unsaved changes, otherwise it increments the current version * there are no unsaved changes, otherwise it increments the current version
...@@ -680,17 +751,22 @@ public class MVStore { ...@@ -680,17 +751,22 @@ public class MVStore {
* @return the new version (incremented if there were changes) * @return the new version (incremented if there were changes)
*/ */
public long store() { public long store() {
checkOpen();
return store(false); return store(false);
} }
/** /**
* Store changes. * Store changes. Changes that are marked as temporary are rolled back after
* a restart.
* *
* @param temp whether the changes should be rolled back after opening * @param temp whether the changes are only temporary (not committed), and
* should be rolled back after a crash
* @return the new version (incremented if there were changes) * @return the new version (incremented if there were changes)
*/ */
private synchronized long store(boolean temp) { private synchronized long store(boolean temp) {
checkOpen(); if (closed) {
return currentVersion;
}
if (currentStoreVersion >= 0) { if (currentStoreVersion >= 0) {
// store is possibly called within store, if the meta map changed // store is possibly called within store, if the meta map changed
return currentVersion; return currentVersion;
...@@ -705,10 +781,27 @@ public class MVStore { ...@@ -705,10 +781,27 @@ public class MVStore {
if (file == null) { if (file == null) {
return version; return version;
} }
long time = getTime(); long time = getTime();
lastStoreTime = time; lastStoreTime = time;
if (temp) {
meta.put("rollbackOnOpen", Long.toString(lastCommittedVersion));
// find the oldest chunk to retain
long minVersion = Long.MAX_VALUE;
Chunk minChunk = null;
for (Chunk c : chunks.values()) {
if (c.version < minVersion) {
minVersion = c.version;
minChunk = c;
}
}
retainChunk = minChunk;
} else {
lastCommittedVersion = version;
meta.remove("rollbackOnOpen");
retainChunk = null;
}
// the last chunk was not completely correct in the last store() // the last chunk was not completely correct in the last store()
// this needs to be updated now (it's better not to update right after // this needs to be updated now (it's better not to update right after
// storing, because that would modify the meta map again) // storing, because that would modify the meta map again)
...@@ -735,7 +828,9 @@ public class MVStore { ...@@ -735,7 +828,9 @@ public class MVStore {
if (m != meta) { if (m != meta) {
long v = m.getVersion(); long v = m.getVersion();
if (v >= 0 && m.getVersion() >= lastStoredVersion) { if (v >= 0 && m.getVersion() >= lastStoredVersion) {
changed.add(m.openVersion(storeVersion)); MVMap<?, ?> r = m.openVersion(storeVersion);
r.waitUntilWritten(r.getRoot());
changed.add(r);
} }
} }
} }
...@@ -747,10 +842,11 @@ public class MVStore { ...@@ -747,10 +842,11 @@ public class MVStore {
meta.put("root." + m.getId(), String.valueOf(Integer.MAX_VALUE)); meta.put("root." + m.getId(), String.valueOf(Integer.MAX_VALUE));
} }
} }
applyFreedChunks(storeVersion); applyFreedChunks(storeVersion);
ArrayList<Integer> removedChunks = New.arrayList(); ArrayList<Integer> removedChunks = New.arrayList();
// do it twice, because changing the meta table // do it twice, because changing the meta table
// could cause a chunk to get empty // could cause a chunk to become empty
for (int i = 0; i < 2; i++) { for (int i = 0; i < 2; i++) {
for (Chunk x : chunks.values()) { for (Chunk x : chunks.values()) {
if (x.maxLengthLive == 0 && canOverwriteChunk(x, time)) { if (x.maxLengthLive == 0 && canOverwriteChunk(x, time)) {
...@@ -848,11 +944,18 @@ public class MVStore { ...@@ -848,11 +944,18 @@ public class MVStore {
} }
private boolean canOverwriteChunk(Chunk c, long time) { private boolean canOverwriteChunk(Chunk c, long time) {
return c.time + retentionTime <= time; if (c.time + retentionTime > time) {
return false;
}
Chunk r = retainChunk;
if (r != null && c.version > r.version) {
return false;
}
return true;
} }
private long getTime() { private long getTime() {
return (System.currentTimeMillis() / 1000) - creationTime; return System.currentTimeMillis() - creationTime;
} }
private void applyFreedChunks(long storeVersion) { private void applyFreedChunks(long storeVersion) {
...@@ -1238,22 +1341,22 @@ public class MVStore { ...@@ -1238,22 +1341,22 @@ public class MVStore {
} }
/** /**
* How long to retain old, persisted chunks, in seconds. Chunks that are * How long to retain old, persisted chunks, in milliseconds. Chunks that
* older than this many seconds may be overwritten once they contain no live * are older may be overwritten once they contain no live data. The default
* data. The default is 45 seconds. It is assumed that a file system and * is 45000 (45 seconds). It is assumed that a file system and hard disk
* hard disk will flush all write buffers within this many seconds. Using a * will flush all write buffers within this time. Using a lower value might
* lower value might be dangerous, unless the file system and hard disk * be dangerous, unless the file system and hard disk flush the buffers
* flush the buffers earlier. To manually flush the buffers, use * earlier. To manually flush the buffers, use
* <code>MVStore.getFile().force(true)</code>, however please note that * <code>MVStore.getFile().force(true)</code>, however please note that
* according to various tests this does not always work as expected. * according to various tests this does not always work as expected.
* <p> * <p>
* This setting is not persisted. * This setting is not persisted.
* *
* @param seconds how many seconds to retain old chunks (0 to overwrite them * @param ms how many milliseconds to retain old chunks (0 to overwrite them
* as early as possible) * as early as possible)
*/ */
public void setRetentionTime(int seconds) { public void setRetentionTime(int ms) {
this.retentionTime = seconds; this.retentionTime = ms;
} }
/** /**
...@@ -1332,7 +1435,7 @@ public class MVStore { ...@@ -1332,7 +1435,7 @@ public class MVStore {
*/ */
void beforeWrite() { void beforeWrite() {
if (unsavedPageCount > maxUnsavedPages && maxUnsavedPages > 0) { if (unsavedPageCount > maxUnsavedPages && maxUnsavedPages > 0) {
store(); store(true);
} }
} }
...@@ -1364,12 +1467,27 @@ public class MVStore { ...@@ -1364,12 +1467,27 @@ public class MVStore {
* Revert to the beginning of the given version. All later changes (stored * Revert to the beginning of the given version. All later changes (stored
* or not) are forgotten. All maps that were created later are closed. A * or not) are forgotten. All maps that were created later are closed. A
* rollback to a version before the last stored version is immediately * rollback to a version before the last stored version is immediately
* persisted. * persisted. Rollback to version 0 means all data is removed.
* *
* @param version the version to revert to * @param version the version to revert to
*/ */
public synchronized void rollbackTo(long version) { public synchronized void rollbackTo(long version) {
checkOpen(); checkOpen();
if (version == 0) {
// special case: remove all data
for (MVMap<?, ?> m : maps.values()) {
m.close();
}
meta.clear();
chunks.clear();
maps.clear();
synchronized (freedChunks) {
freedChunks.clear();
}
currentVersion = version;
metaChanged = false;
return;
}
DataUtils.checkArgument( DataUtils.checkArgument(
isKnownVersion(version), isKnownVersion(version),
"Unknown version {0}", version); "Unknown version {0}", version);
...@@ -1422,7 +1540,6 @@ public class MVStore { ...@@ -1422,7 +1540,6 @@ public class MVStore {
} }
} }
} }
// this.lastStoredVersion = version - 1;
this.currentVersion = version; this.currentVersion = version;
} }
...@@ -1451,6 +1568,15 @@ public class MVStore { ...@@ -1451,6 +1568,15 @@ public class MVStore {
return currentVersion; return currentVersion;
} }
/**
* Get the last committed version.
*
* @return the version
*/
public long getCommittedVersion() {
return lastCommittedVersion;
}
/** /**
* Get the number of file write operations since this store was opened. * Get the number of file write operations since this store was opened.
* *
...@@ -1543,38 +1669,48 @@ public class MVStore { ...@@ -1543,38 +1669,48 @@ public class MVStore {
return DataUtils.parseMap(m).get("name"); return DataUtils.parseMap(m).get("name");
} }
void storeIfNeeded() { /**
* Store all unsaved changes, if there are any that are committed.
*/
void storeUnsaved() {
if (closed || unsavedPageCount == 0) { if (closed || unsavedPageCount == 0) {
return; return;
} }
if (lastCommittedVersion >= currentVersion) {
return;
}
long time = getTime(); long time = getTime();
if (time <= lastStoreTime + 1) { if (time <= lastStoreTime + writeDelay) {
return; return;
} }
store(); store(true);
} }
/** /**
* A background writer to automatically store changes every two seconds. * A background writer to automatically store changes from time to time.
*/ */
private static class Writer implements Runnable { private static class Writer implements Runnable {
private final MVStore store; private final MVStore store;
private final int sleep;
Writer(MVStore store) { Writer(MVStore store, int sleep) {
this.store = store; this.store = store;
this.sleep = sleep;
} }
@Override @Override
public void run() { public void run() {
while (!store.closed) { while (!store.closed) {
store.storeIfNeeded(); synchronized (store) {
try { try {
Thread.sleep(1000); store.wait(sleep);
} catch (InterruptedException e) { } catch (InterruptedException e) {
// ignore // ignore
} }
} }
store.storeUnsaved();
}
} }
} }
...@@ -1636,36 +1772,60 @@ public class MVStore { ...@@ -1636,36 +1772,60 @@ public class MVStore {
/** /**
* Set the read cache size in MB. The default is 16 MB. * Set the read cache size in MB. The default is 16 MB.
* *
* @param mb the cache size * @param mb the cache size in megabytes
* @return this * @return this
*/ */
public Builder cacheSizeMB(int mb) { public Builder cacheSize(int mb) {
return set("cacheSize", mb); return set("cacheSize", mb);
} }
/** /**
* Set the size of the write buffer in MB. The default is 4 MB. Changes * Compress data before writing using the LZF algorithm. This setting only
* are automatically stored if the buffer grows larger than this, and * affects writes; it is not necessary to enable compression when reading,
* after 2 seconds (whichever occurs earlier). * even if compression was enabled when writing.
*
* @return this
*/
public Builder compressData() {
return set("compress", 1);
}
/**
* Set the size of the write buffer, in MB (for file-based stores).
* Changes are automatically stored if the buffer grows larger than
* this. However, unless the changes are committed later on, they are
* rolled back when opening the store.
* <p> * <p>
* To disable automatically storing, set the buffer size to 0. * The default is 4 MB.
* <p>
* When the value is set to 0 or lower, data is never automatically
* stored.
* *
* @param mb the write buffer size * @param mb the write buffer size, in megabytes
* @return this * @return this
*/ */
public Builder writeBufferSizeMB(int mb) { public Builder writeBufferSize(int mb) {
return set("writeBufferSize", mb); return set("writeBufferSize", mb);
} }
/** /**
* Compress data before writing using the LZF algorithm. This setting only * Set the maximum delay in milliseconds to store committed changes (for
* affects writes; it is not necessary to enable compression when reading, * file-based stores).
* even if compression was enabled when writing. * <p>
* The default is 1000, meaning committed changes are stored after at
* most one second.
* <p>
* When the value is set to -1, committed changes are only written when
* calling the store method. When the value is set to 0, committed
* changes are immediately written on a commit, but please note this
* decreases performance and does still not guarantee the disk will
* actually write the data.
* *
* @param millis the maximum delay
* @return this * @return this
*/ */
public Builder compressData() { public Builder writeDelay(int millis) {
return set("compress", 1); return set("writeDelay", millis);
} }
/** /**
......
...@@ -42,6 +42,8 @@ public class TestMVStore extends TestBase { ...@@ -42,6 +42,8 @@ public class TestMVStore extends TestBase {
FileUtils.deleteRecursive(getBaseDir(), true); FileUtils.deleteRecursive(getBaseDir(), true);
FileUtils.createDirectories(getBaseDir()); FileUtils.createDirectories(getBaseDir());
testWriteBuffer();
testWriteDelay();
testEncryptedFile(); testEncryptedFile();
testFileFormatChange(); testFileFormatChange();
testRecreateMap(); testRecreateMap();
...@@ -77,6 +79,107 @@ public class TestMVStore extends TestBase { ...@@ -77,6 +79,107 @@ public class TestMVStore extends TestBase {
testSimple(); testSimple();
} }
private void testWriteBuffer() throws IOException {
String fileName = getBaseDir() + "/testAutoStoreBuffer.h3";
FileUtils.delete(fileName);
MVStore s;
MVMap<Integer, byte[]> m;
byte[] data = new byte[1000];
long lastSize = 0;
int len = 1000;
for (int bs = 0; bs <= 1; bs++) {
s = new MVStore.Builder().
fileName(fileName).
writeBufferSize(bs).
open();
m = s.openMap("data");
for (int i = 0; i < len; i++) {
m.put(i, data);
}
long size = s.getFile().size();
assertTrue("last:" + lastSize + " now: " + size, size > lastSize);
lastSize = size;
s.close();
}
s = new MVStore.Builder().
fileName(fileName).
open();
m = s.openMap("data");
assertFalse(m.containsKey(1));
m.put(1, data);
s.commit();
m.put(2, data);
s.close();
s = new MVStore.Builder().
fileName(fileName).
open();
m = s.openMap("data");
assertTrue(m.containsKey(1));
assertFalse(m.containsKey(2));
s.close();
FileUtils.delete(fileName);
}
private void testWriteDelay() throws InterruptedException {
String fileName = getBaseDir() + "/testUndoTempStore.h3";
FileUtils.delete(fileName);
MVStore s;
MVMap<Integer, String> m;
s = new MVStore.Builder().
writeDelay(1).
fileName(fileName).
open();
m = s.openMap("data");
m.put(1, "Hello");
s.store();
long v = s.getCurrentVersion();
m.put(2, "World");
Thread.sleep(5);
// must not store, as nothing has been committed yet
assertEquals(v, s.getCurrentVersion());
s.commit();
m.put(3, "!");
for (int i = 100; i > 0; i--) {
if (s.getCurrentVersion() > v) {
break;
}
if (i < 10) {
fail();
}
Thread.sleep(1);
}
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().
fileName(fileName).
open();
m = s.openMap("data");
assertEquals("Hello", m.get(1));
assertEquals("World", m.get(2));
assertFalse(m.containsKey(3));
s.close();
FileUtils.delete(fileName);
}
private void testEncryptedFile() { private void testEncryptedFile() {
String fileName = getBaseDir() + "/testEncryptedFile.h3"; String fileName = getBaseDir() + "/testEncryptedFile.h3";
FileUtils.delete(fileName); FileUtils.delete(fileName);
...@@ -175,6 +278,8 @@ public class TestMVStore extends TestBase { ...@@ -175,6 +278,8 @@ public class TestMVStore extends TestBase {
assertEquals("world", map.getName()); assertEquals("world", map.getName());
s.rollbackTo(old); s.rollbackTo(old);
assertEquals("hello", map.getName()); assertEquals("hello", map.getName());
s.rollbackTo(0);
assertTrue(map.isClosed());
s.close(); s.close();
} }
...@@ -211,7 +316,7 @@ public class TestMVStore extends TestBase { ...@@ -211,7 +316,7 @@ public class TestMVStore extends TestBase {
for (int cacheSize = 0; cacheSize <= 6; cacheSize += 4) { for (int cacheSize = 0; cacheSize <= 6; cacheSize += 4) {
s = new MVStore.Builder(). s = new MVStore.Builder().
fileName(fileName). fileName(fileName).
cacheSizeMB(1 + 3 * cacheSize).open(); cacheSize(1 + 3 * cacheSize).open();
map = s.openMap("test"); map = s.openMap("test");
for (int i = 0; i < 1024; i += 128) { for (int i = 0; i < 1024; i += 128) {
for (int j = 0; j < i; j++) { for (int j = 0; j < i; j++) {
...@@ -253,11 +358,11 @@ public class TestMVStore extends TestBase { ...@@ -253,11 +358,11 @@ public class TestMVStore extends TestBase {
private void testFileHeader() { private void testFileHeader() {
String fileName = getBaseDir() + "/testFileHeader.h3"; String fileName = getBaseDir() + "/testFileHeader.h3";
MVStore s = openStore(fileName); MVStore s = openStore(fileName);
long time = System.currentTimeMillis() / 1000; long time = System.currentTimeMillis();
assertEquals("3", s.getFileHeader().get("H")); assertEquals("3", s.getFileHeader().get("H"));
long creationTime = Long.parseLong(s.getFileHeader() long creationTime = Long.parseLong(s.getFileHeader()
.get("creationTime")); .get("creationTime"));
assertTrue(Math.abs(time - creationTime) < 5); assertTrue(Math.abs(time - creationTime) < 100);
s.getFileHeader().put("test", "123"); s.getFileHeader().put("test", "123");
MVMap<Integer, Integer> map = s.openMap("test"); MVMap<Integer, Integer> map = s.openMap("test");
map.put(10, 100); map.put(10, 100);
...@@ -274,6 +379,7 @@ public class TestMVStore extends TestBase { ...@@ -274,6 +379,7 @@ public class TestMVStore extends TestBase {
MVMap<Integer, Integer> map = s.openMap("test"); MVMap<Integer, Integer> map = s.openMap("test");
map.put(10, 100); map.put(10, 100);
FilePath f = FilePath.get(s.getFileName()); FilePath f = FilePath.get(s.getFileName());
s.store();
s.close(); s.close();
int blockSize = 4 * 1024; int blockSize = 4 * 1024;
// test corrupt file headers // test corrupt file headers
...@@ -299,6 +405,7 @@ public class TestMVStore extends TestBase { ...@@ -299,6 +405,7 @@ public class TestMVStore extends TestBase {
// header should be used // header should be used
s = openStore(fileName); s = openStore(fileName);
map = s.openMap("test"); map = s.openMap("test");
assertEquals(100, map.get(10).intValue());
s.close(); s.close();
} else { } else {
// both headers are corrupt // both headers are corrupt
...@@ -710,11 +817,11 @@ public class TestMVStore extends TestBase { ...@@ -710,11 +817,11 @@ public class TestMVStore extends TestBase {
FileUtils.delete(fileName); FileUtils.delete(fileName);
MVMap<String, String> meta; MVMap<String, String> meta;
MVStore s = openStore(fileName); MVStore s = openStore(fileName);
assertEquals(45, s.getRetentionTime()); assertEquals(45000, s.getRetentionTime());
s.setRetentionTime(0); s.setRetentionTime(0);
assertEquals(0, s.getRetentionTime()); assertEquals(0, s.getRetentionTime());
s.setRetentionTime(45); s.setRetentionTime(45000);
assertEquals(45, s.getRetentionTime()); assertEquals(45000, s.getRetentionTime());
assertEquals(0, s.getCurrentVersion()); assertEquals(0, s.getCurrentVersion());
assertFalse(s.hasUnsavedChanges()); assertFalse(s.hasUnsavedChanges());
MVMap<String, String> m = s.openMap("data"); MVMap<String, String> m = s.openMap("data");
......
/*
* Copyright 2004-2011 H2 Group. Multiple-Licensed under the H2 License,
* Version 1.0, and under the Eclipse Public License, Version 1.0
* (http://h2database.com/html/license.html).
* Initial Developer: H2 Group
*/
package org.h2.test.store;
import org.h2.test.TestBase;
/**
* Test using volatile fields to ensure we don't read from a version that is
* concurrently written to.
*/
public class TestSpinLock extends TestBase {
/**
* The version to use for writing.
*/
volatile int writeVersion;
/**
* The current data object.
*/
volatile Data data = new Data(0, null);
/**
* Run just this test.
*
* @param a ignored
*/
public static void main(String... a) throws Exception {
TestBase.createCaller().init().test();
}
@Override
public void test() throws Exception {
final TestSpinLock obj = new TestSpinLock();
Thread t = new Thread() {
public void run() {
while (!isInterrupted()) {
for (int i = 0; i < 10000; i++) {
Data d = obj.copyOnWrite();
obj.data = d;
d.write(i);
d.writing = false;
}
}
}
};
t.start();
try {
for (int i = 0; i < 100000; i++) {
Data d = obj.getImmutable();
int z = d.x + d.y;
if (z != 0) {
String error = i + " result: " + z + " now: " + d.x + " "
+ d.y;
System.out.println(error);
throw new Exception(error);
}
}
} finally {
t.interrupt();
t.join();
}
}
/**
* Clone the data object if necessary (if the write version is newer than
* the current version).
*
* @return the data object
*/
Data copyOnWrite() {
Data d = data;
d.writing = true;
int w = writeVersion;
if (w <= data.version) {
return d;
}
Data d2 = new Data(w, data);
d2.writing = true;
d.writing = false;
return d2;
}
/**
* Get an immutable copy of the data object.
*
* @return the immutable object
*/
private Data getImmutable() {
Data d = data;
++writeVersion;
// wait until writing is done,
// but only for the current write operation:
// a bit like a spin lock
while (d.writing) {
// Thread.yield() is not required, specially
// if there are multiple cores
// but getImmutable() doesn't
// need to be that fast actually
Thread.yield();
}
return d;
}
/**
* The data class - represents the root page.
*/
static class Data {
/**
* The version.
*/
final int version;
/**
* The values.
*/
int x, y;
/**
* Whether a write operation is in progress.
*/
volatile boolean writing;
/**
* Create a copy of the data.
*
* @param version the new version
* @param old the old data or null
*/
Data(int version, Data old) {
this.version = version;
if (old != null) {
this.x = old.x;
this.y = old.y;
}
}
/**
* Write to the fields in an unsynchronized way.
*
* @param value the new value
*/
void write(int value) {
this.x = value;
this.y = -value;
}
}
}
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论