Skip to content

H
h2database
提交 86846013 authored 作者: Andrei Tokar's avatar Andrei Tokar
浏览文件
操作

batch

上级 aad6a418
隐藏空白字符变更
内嵌 并排
正在显示 包含 214 行增加 和 20 行删除
h2/src/main/org/h2/mvstore/MVMap.java
浏览文件 @ 86846013
......@@ -45,6 +45,11 @@ public class MVMap<K, V> extends AbstractMap<K, V>
private final long createVersion;
private final DataType keyType;
private final DataType valueType;
private final int keysPerPage;
private final boolean singleWriter;
private final K keysBuffer[];
private final V valuesBuffer[];
/**
* Whether the map is closed. Volatile so we don't accidentally write to a
......@@ -66,7 +71,9 @@ public class MVMap<K, V> extends AbstractMap<K, V>
(DataType) config.get("val"),
DataUtils.readHexInt(config, "id", 0),
DataUtils.readHexLong(config, "createVersion", 0),
new AtomicReference<RootReference>()
new AtomicReference<RootReference>(),
((MVStore) config.get("store")).getKeysPerPage(),
config.containsKey("singleWriter") && (Boolean) config.get("singleWriter")
);
setInitialRoot(createEmptyLeaf(), store.getCurrentVersion());
}
......@@ -74,23 +81,29 @@ public class MVMap<K, V> extends AbstractMap<K, V>
// constructor for cloneIt()
protected MVMap(MVMap<K, V> source) {
this(source.store, source.keyType, source.valueType, source.id, source.createVersion,
new AtomicReference<>(source.root.get()));
new AtomicReference<>(source.root.get()), source.keysPerPage, source.singleWriter);
}
// meta map constructor
MVMap(MVStore store) {
this(store, StringDataType.INSTANCE,StringDataType.INSTANCE, 0, 0, new AtomicReference<RootReference>());
this(store, StringDataType.INSTANCE,StringDataType.INSTANCE, 0, 0, new AtomicReference<RootReference>(),
store.getKeysPerPage(), false);
setInitialRoot(createEmptyLeaf(), store.getCurrentVersion());
}
@SuppressWarnings("unchecked")
private MVMap(MVStore store, DataType keyType, DataType valueType, int id, long createVersion,
AtomicReference<RootReference> root) {
AtomicReference<RootReference> root, int keysPerPage, boolean singleWriter) {
this.store = store;
this.id = id;
this.createVersion = createVersion;
this.keyType = keyType;
this.valueType = valueType;
this.root = root;
this.keysPerPage = keysPerPage;
this.keysBuffer = singleWriter ? (K[]) new Object[keysPerPage] : null;
this.valuesBuffer = singleWriter ? (V[]) new Object[keysPerPage] : null;
this.singleWriter = singleWriter;
}
protected MVMap<K, V> cloneIt() {
......@@ -1037,6 +1050,10 @@ public class MVMap<K, V> extends AbstractMap<K, V>
return getVersion() > version;
}
public boolean isSingleWriter() {
return singleWriter;
}
/**
* Get the child page count for this page. This is to allow another map
* implementation to override the default, in case the last child is not to
......@@ -1091,6 +1108,7 @@ public class MVMap<K, V> extends AbstractMap<K, V>
}
return rootReference;
}
rootReference = flushAppendBuffer(rootReference);
RootReference updatedRootReference = new RootReference(rootReference, writeVersion, ++attempt);
if(root.compareAndSet(rootReference, updatedRootReference)) {
removeUnusedOldVersions(updatedRootReference);
......@@ -1148,6 +1166,135 @@ public class MVMap<K, V> extends AbstractMap<K, V>
return target;
}
public RootReference flushAppendBuffer() {
return flushAppendBuffer(null);
}
private RootReference flushAppendBuffer(RootReference rootReference) {
int attempt = 0;
while(true) {
if (rootReference == null) {
rootReference = getRoot();
}
byte keyCount = rootReference.appendCounter;
if (keyCount == 0) {
break;
}
Page page = Page.create(this,
createAndFillStorage(keyCount, keysBuffer),
createAndFillStorage(keyCount, valuesBuffer),
null, keyCount, 0);
rootReference = appendLeafPage(rootReference, page, ++attempt);
if (rootReference != null) {
break;
}
}
assert rootReference.appendCounter == 0;
return rootReference;
}
private RootReference appendLeafPage(RootReference rootReference, Page split, int attempt) {
CursorPos pos = rootReference.root.getAppendCursorPos(null);
assert split.map == this;
assert pos != null;
assert split.getKeyCount() > 0;
Object key = split.getKey(0);
assert pos.index < 0 : pos.index;
int index = -pos.index - 1;
assert index == pos.page.getKeyCount() : index + " != " + pos.page.getKeyCount();
Page p = pos.page;
pos = pos.parent;
CursorPos tip = pos;
int unsavedMemory = 0;
while (true) {
if (pos == null) {
if (p.getKeyCount() == 0) {
p = split;
} else {
Object keys[] = new Object[] { key };
Page.PageReference children[] = new Page.PageReference[store.getKeysPerPage() + 1];
children[0] = new Page.PageReference(p);
children[1] = new Page.PageReference(split);
p = Page.create(this, keys, null, children, p.getTotalCount() + split.getTotalCount(), 0);
}
break;
}
Page c = p;
p = pos.page;
index = pos.index;
pos = pos.parent;
p = p.copy();
p.setChild(index, split);
p.insertNode(index, key, c);
int keyCount;
if ((keyCount = p.getKeyCount()) <= store.getKeysPerPage() && (p.getMemory() < store.getMaxPageSize() || keyCount <= (p.isLeaf() ? 1 : 2))) {
break;
}
int at = keyCount - 2;
key = p.getKey(at);
split = p.split(at);
unsavedMemory += p.getMemory() + split.getMemory();
}
unsavedMemory += p.getMemory();
while (pos != null) {
Page c = p;
p = pos.page;
p = p.copy();
p.setChild(pos.index, c);
unsavedMemory += p.getMemory();
pos = pos.parent;
}
RootReference updatedRootReference = new RootReference(rootReference, p, ++attempt);
if(root.compareAndSet(rootReference, updatedRootReference)) {
while (tip != null) {
tip.page.removePage();
tip = tip.parent;
}
if (store.getFileStore() != null) {
store.registerUnsavedPage(unsavedMemory);
}
return updatedRootReference;
}
return null;
}
public void append(K key, V value) {
int attempt = 0;
boolean success = false;
while(!success) {
RootReference rootReference = getRoot();
if (rootReference.appendCounter >= keysPerPage) {
rootReference = flushAppendBuffer(rootReference);
assert rootReference.appendCounter < keysPerPage;
}
keysBuffer[rootReference.appendCounter] = key;
valuesBuffer[rootReference.appendCounter] = value;
RootReference updatedRootReference = new RootReference(rootReference, (byte) (rootReference.appendCounter + 1), ++attempt);
success = root.compareAndSet(rootReference, updatedRootReference);
}
}
public void trimLast() {
int attempt = 0;
boolean success;
do {
RootReference rootReference = getRoot();
if (rootReference.appendCounter > 0) {
RootReference updatedRootReference = new RootReference(rootReference, (byte) (rootReference.appendCounter - 1), ++attempt);
success = root.compareAndSet(rootReference, updatedRootReference);
} else {
assert rootReference.root.getKeyCount() > 0;
Page lastLeaf = rootReference.root.getAppendCursorPos(null).page;
assert lastLeaf.isLeaf();
assert lastLeaf.getKeyCount() > 0;
Object key = lastLeaf.getKey(lastLeaf.getKeyCount() - 1);
success = remove(key) != null;
assert success;
}
} while(!success);
}
@Override
public final String toString() {
return asString(null);
......@@ -1179,6 +1326,10 @@ public class MVMap<K, V> extends AbstractMap<K, V>
* Counter for attempted root updates.
*/
public final long updateAttemptCounter;
/**
* Size of the occupied part of the append buffer.
*/
public final byte appendCounter;
RootReference(Page root, long version, RootReference previous,
long updateCounter, long updateAttemptCounter,
......@@ -1189,6 +1340,7 @@ public class MVMap<K, V> extends AbstractMap<K, V>
this.updateCounter = updateCounter;
this.updateAttemptCounter = updateAttemptCounter;
this.lockedForUpdate = lockedForUpdate;
this.appendCounter = 0;
}
// This one is used for locking
......@@ -1199,6 +1351,7 @@ public class MVMap<K, V> extends AbstractMap<K, V>
this.updateCounter = r.updateCounter;
this.updateAttemptCounter = r.updateAttemptCounter;
this.lockedForUpdate = true;
this.appendCounter = 0;
}
// This one is used for unlocking
......@@ -1209,6 +1362,7 @@ public class MVMap<K, V> extends AbstractMap<K, V>
this.updateCounter = r.updateCounter + 1;
this.updateAttemptCounter = r.updateAttemptCounter + attempt;
this.lockedForUpdate = false;
this.appendCounter = 0;
}
// This one is used for version change
......@@ -1224,6 +1378,7 @@ public class MVMap<K, V> extends AbstractMap<K, V>
this.updateCounter = r.updateCounter + 1;
this.updateAttemptCounter = r.updateAttemptCounter + attempt;
this.lockedForUpdate = r.lockedForUpdate;
this.appendCounter = r.appendCounter;
}
// This one is used for r/o snapshots
......@@ -1234,6 +1389,18 @@ public class MVMap<K, V> extends AbstractMap<K, V>
this.updateCounter = 1;
this.updateAttemptCounter = 1;
this.lockedForUpdate = false;
this.appendCounter = 0;
}
// This one is used for append buffer maintance
RootReference(RootReference r, byte appendCounter, int attempt) {
this.root = r.root;
this.version = r.version;
this.previous = r.previous;
this.updateCounter = r.updateCounter + 1;
this.updateAttemptCounter = r.updateAttemptCounter + attempt;
this.lockedForUpdate = r.lockedForUpdate;
this.appendCounter = appendCounter;
}
@Override
......@@ -1357,6 +1524,7 @@ public class MVMap<K, V> extends AbstractMap<K, V>
* @param <V> the value type
*/
public static class Builder<K, V> extends BasicBuilder<MVMap<K, V>, K, V> {
private boolean singleWriter;
public Builder() {}
......@@ -1372,8 +1540,14 @@ public class MVMap<K, V> extends AbstractMap<K, V>
return this;
}
public Builder<K,V> singleWriter() {
singleWriter = true;
return this;
}
@Override
protected MVMap<K, V> create(Map<String, Object> config) {
config.put("singleWriter", singleWriter);
Object type = config.get("type");
if(type == null || type.equals("rtree")) {
return new MVMap<>(config);
......@@ -1678,6 +1852,12 @@ public class MVMap<K, V> extends AbstractMap<K, V>
return new CursorPos(p, p.binarySearch(key), pos);
}
private static Object[] createAndFillStorage(int count, Object dataBuffer[]) {
Object storage[] = new Object[count];
System.arraycopy(dataBuffer, 0, storage, 0, count);
return storage;
}
private static final class EqualsDecisionMaker<V> extends DecisionMaker<V> {
private final DataType dataType;
private final V expectedValue;
......
h2/src/main/org/h2/mvstore/Page.java
浏览文件 @ 86846013
......@@ -836,6 +836,8 @@ public abstract class Page implements Cloneable
}
}
public abstract CursorPos getAppendCursorPos(CursorPos cursorPos);
public abstract void removeAllRecursive();
private Object[] createKeyStorage(int size)
......@@ -1083,6 +1085,13 @@ public abstract class Page implements Cloneable
removePage();
}
@Override
public CursorPos getAppendCursorPos(CursorPos cursorPos) {
int keyCount = getKeyCount();
Page childPage = getChildPage(keyCount);
return childPage.getAppendCursorPos(new CursorPos(this, keyCount, cursorPos));
}
@Override
protected void readPayLoad(ByteBuffer buff) {
int keyCount = getKeyCount();
......@@ -1322,6 +1331,12 @@ public abstract class Page implements Cloneable
removePage();
}
@Override
public CursorPos getAppendCursorPos(CursorPos cursorPos) {
int keyCount = getKeyCount();
return new CursorPos(this, -keyCount - 1, cursorPos);
}
@Override
protected void readPayLoad(ByteBuffer buff) {
int keyCount = getKeyCount();
......
h2/src/main/org/h2/mvstore/tx/Transaction.java
浏览文件 @ 86846013
......@@ -300,7 +300,7 @@ public class Transaction {
}
int currentStatus = getStatus(currentState);
checkOpen(currentStatus);
store.removeUndoLogRecord(transactionId, logId);
store.removeUndoLogRecord(transactionId);
}
/**
......
h2/src/main/org/h2/mvstore/tx/TransactionMap.java
浏览文件 @ 86846013
......@@ -88,9 +88,9 @@ public class TransactionMap<K, V> {
for (int i = opentransactions.nextSetBit(0); i >= 0; i = opentransactions.nextSetBit(i+1)) {
MVMap<Long, Object[]> undoLog = store.undoLogs[i];
if (undoLog != null) {
MVMap.RootReference rootReference = undoLog.getRoot();
MVMap.RootReference rootReference = undoLog.flushAppendBuffer();
undoLogRootReferences[i] = rootReference;
undoLogSize += rootReference.root.getTotalCount();
undoLogSize += rootReference.root.getTotalCount() + rootReference.appendCounter;
}
}
} while(committingTransactions != store.committingTransactions.get() ||
......
h2/src/main/org/h2/mvstore/tx/TransactionStore.java
浏览文件 @ 86846013
......@@ -16,6 +16,7 @@ import org.h2.mvstore.Cursor;
import org.h2.mvstore.DataUtils;
import org.h2.mvstore.MVMap;
import org.h2.mvstore.MVStore;
import org.h2.mvstore.Page;
import org.h2.mvstore.WriteBuffer;
import org.h2.mvstore.type.DataType;
import org.h2.mvstore.type.ObjectDataType;
......@@ -132,7 +133,9 @@ public class TransactionStore {
ArrayType undoLogValueType = new ArrayType(new DataType[]{
new ObjectDataType(), dataType, oldValueType
});
undoLogBuilder = new MVMap.Builder<Long, Object[]>().valueType(undoLogValueType);
undoLogBuilder = new MVMap.Builder<Long, Object[]>()
.singleWriter()
.valueType(undoLogValueType);
}
/**
......@@ -381,24 +384,16 @@ public class TransactionStore {
"is still open: {0}",
transactionId);
}
undoLog.put(undoKey, undoLogRecord);
undoLog.append(undoKey, undoLogRecord);
return undoKey;
}
/**
* Remove an undo log entry.
* @param transactionId id of the transaction
* @param logId sequential number of the log record within transaction
*/
public void removeUndoLogRecord(int transactionId, long logId) {
Long undoKey = getOperationId(transactionId, logId);
Object[] old = undoLogs[transactionId].remove(undoKey);
if (old == null) {
throw DataUtils.newIllegalStateException(
DataUtils.ERROR_TRANSACTION_ILLEGAL_STATE,
"Transaction {0} was concurrently rolled back",
transactionId);
}
void removeUndoLogRecord(int transactionId) {
undoLogs[transactionId].trimLast();
}
/**
......@@ -432,7 +427,9 @@ public class TransactionStore {
store.renameMap(undoLog, getUndoLogName(true, transactionId));
}
try {
Cursor<Long, Object[]> cursor = undoLog.cursor(null);
MVMap.RootReference rootReference = undoLog.flushAppendBuffer();
Page rootPage = rootReference.root;
Cursor<Long, Object[]> cursor = new Cursor<>(rootPage, null);
while (cursor.hasNext()) {
Long undoKey = cursor.next();
Object[] op = cursor.getValue();
......@@ -587,6 +584,7 @@ public class TransactionStore {
void rollbackTo(Transaction t, long maxLogId, long toLogId) {
int transactionId = t.getId();
MVMap<Long, Object[]> undoLog = undoLogs[transactionId];
undoLog.flushAppendBuffer();
RollbackDecisionMaker decisionMaker = new RollbackDecisionMaker(this, transactionId, toLogId, t.listener);
for (long logId = maxLogId - 1; logId >= toLogId; logId--) {
Long undoKey = getOperationId(transactionId, logId);
......@@ -608,6 +606,7 @@ public class TransactionStore {
final long toLogId) {
final MVMap<Long, Object[]> undoLog = undoLogs[t.getId()];
undoLog.flushAppendBuffer();
return new Iterator<Change>() {
private long logId = maxLogId - 1;
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论