Skip to content

Commit

Permalink
插件化事务引擎API，增加一个MemoryTransactionEngine骨架代码用于演示
Browse files Browse the repository at this point in the history
  • Loading branch information
@codefollower
codefollower committed May 10, 2015
1 parent b347ad2 commit 778f5b5
Show file tree
Hide file tree
Showing 14 changed files with 514 additions and 77 deletions.
33 changes: 33 additions & 0 deletions lealone-common/src/main/java/org/lealone/transaction/Transaction.java
View file
Expand Up @@ -21,6 +21,39 @@

public interface Transaction {

/**
* The status of a closed transaction (committed or rolled back).
*/
public static final int STATUS_CLOSED = 0;

/**
* The status of an open transaction.
*/
public static final int STATUS_OPEN = 1;

/**
* The status of a prepared transaction.
*/
public static final int STATUS_PREPARED = 2;

/**
* The status of a transaction that is being committed, but possibly not
* yet finished. A transactions can go into this state when the store is
* closed while the transaction is committing. When opening a store,
* such transactions should be committed.
*/
public static final int STATUS_COMMITTING = 3;

String getName(); //用于2pc

void setName(String name); //用于2pc

void prepare(); //用于2pc

int getStatus();

void setStatus(int status);

//long getTransactionId();

//long getCommitTimestamp();
Expand Down
9 changes: 9 additions & 0 deletions lealone-common/src/main/java/org/lealone/transaction/TransactionEngine.java
View file
Expand Up @@ -17,10 +17,19 @@
*/
package org.lealone.transaction;

import java.util.List;
import java.util.Set;

public interface TransactionEngine {
Transaction beginTransaction(boolean autoCommit);

void close();

boolean isValid(String localTransactionName);

void init(Set<String> storageMapNames);

List<Transaction> getOpenTransactions();

<K, V> void removeMap(TransactionMap<K, V> map);
}
2 changes: 2 additions & 0 deletions lealone-common/src/main/java/org/lealone/transaction/TransactionMap.java
View file
Expand Up @@ -196,4 +196,6 @@ public interface TransactionMap<K, V> {
* @return the result
*/
public K lowerKey(K key);

public int getMapId();
}
1 change: 0 additions & 1 deletion lealone-mvdb/src/main/java/org/lealone/dbobject/index/MVPrimaryIndex.java
View file
Expand Up @@ -55,7 +55,6 @@ public class MVPrimaryIndex extends IndexBase {

private final MVTable mvTable;
private final String mapName;
//private final TransactionMap<Value, Value> dataMap;
private final TransactionMap<Value, Value> dataMap;
private long lastKey;
private int mainIndexColumn = -1;
Expand Down
103 changes: 62 additions & 41 deletions lealone-storage/engine/src/main/java/org/lealone/storage/MVStorageEngine.java
View file
Expand Up @@ -30,9 +30,9 @@
import org.lealone.mvstore.MVMap;
import org.lealone.mvstore.MVStore;
import org.lealone.mvstore.MVStoreTool;
import org.lealone.transaction.MVCCTransaction;
import org.lealone.transaction.MVCCTransactionEngine;
import org.lealone.transaction.MVCCTransactionMap;
import org.lealone.transaction.Transaction;
import org.lealone.transaction.TransactionEngine;
import org.lealone.transaction.TransactionMap;
import org.lealone.type.DataType;
import org.lealone.util.BitField;
Expand All @@ -46,37 +46,52 @@ public class MVStorageEngine extends StorageEngineBase implements TransactionSto
public static final String NAME = Constants.DEFAULT_STORAGE_ENGINE_NAME;

private StorageMap.Builder mapBuilder;

public MVStorageEngine() {
this(null);
}
private TransactionEngine transactionEngine;

//见StorageEngineManager.StorageEngineService中的注释
public MVStorageEngine(StorageMap.Builder mapBuilder) {
public MVStorageEngine() {
StorageEngineManager.registerStorageEngine(this);
setMapBuilder(mapBuilder);
}

@Override
public String getName() {
return NAME;
}

protected void setMapBuilder(StorageMap.Builder mapBuilder) {
public void setMapBuilder(StorageMap.Builder mapBuilder) {
this.mapBuilder = mapBuilder;
}

public StorageMap.Builder getMapBuilder() {
return mapBuilder;
}

public void setTransactionEngine(TransactionEngine transactionEngine) {
this.transactionEngine = transactionEngine;
}

public TransactionEngine getTransactionEngine() {
return transactionEngine;
}

@Override
public Table createTable(CreateTableData data) {
Database db = data.session.getDatabase();
Store store = stores.get(db.getName());
if (store == null) {
synchronized (stores) {
if (stores.get(db.getName()) == null) {
store = init(db, mapBuilder);

store = init(db, mapBuilder, transactionEngine);
stores.put(db.getName(), store);

if (mapBuilder == null)
mapBuilder = store.getMapBuilder();
if (transactionEngine == null)
transactionEngine = store.getTransactionEngine();

db.setStorageEngine(this);
db.setTransactionEngine(store.getTransactionEngine());
db.setTransactionEngine(transactionEngine);
db.setLobStorage(new LobStorageMap(db));
}
}
Expand Down Expand Up @@ -109,17 +124,13 @@ public static Store getStore(Database db) {
* @param db the database
* @return the store
*/
public static Store init(final Database db) {
return init(db, null);
}

public static Store init(final Database db, StorageMap.Builder mapBuilder) {
static Store init(final Database db, StorageMap.Builder mapBuilder, TransactionEngine transactionEngine) {
Store store = null;
byte[] key = db.getFileEncryptionKey();
String dbPath = db.getDatabasePath();
MVStore.Builder builder = new MVStore.Builder();
if (dbPath == null) {
store = new Store(db, builder);
store = new Store(db, builder, mapBuilder, transactionEngine);
} else {
String fileName = dbPath + Constants.SUFFIX_MV_FILE;
builder.pageSplitSize(db.getPageSize());
Expand Down Expand Up @@ -158,7 +169,7 @@ public void uncaughtException(Thread t, Throwable e) {

});
try {
store = new Store(db, builder, mapBuilder);
store = new Store(db, builder, mapBuilder, transactionEngine);
} catch (IllegalStateException e) {
int errorCode = DataUtils.getErrorCode(e.getMessage());
if (errorCode == DataUtils.ERROR_FILE_CORRUPT) {
Expand Down Expand Up @@ -195,22 +206,28 @@ public static class Store {
/**
* The transaction engine.
*/
private final MVCCTransactionEngine transactionEngine;
private final TransactionEngine transactionEngine;
private final StorageMap.Builder mapBuilder;

private long statisticsStart;

private int temporaryMapId;

public Store(Database db, MVStore.Builder builder) {
this(db, builder, null);
}

public Store(Database db, MVStore.Builder builder, StorageMap.Builder mapBuilder) {
public Store(Database db, MVStore.Builder builder, StorageMap.Builder mapBuilder,
TransactionEngine transactionEngine) {
this.store = builder.open();

if (mapBuilder == null)
mapBuilder = new MVMapBuilder(store);
this.transactionEngine = new MVCCTransactionEngine(new ValueDataType(null, db, null), mapBuilder,
DatabaseEngine.getHostAndPort());
this.mapBuilder = mapBuilder = new MVMapBuilder(store);
else
this.mapBuilder = mapBuilder;

if (transactionEngine == null)
this.transactionEngine = new MVCCTransactionEngine(new ValueDataType(null, db, null), mapBuilder,
DatabaseEngine.getHostAndPort());
else
this.transactionEngine = transactionEngine;

transactionEngine.init(store.getMapNames());
initTransactions();
}
Expand All @@ -219,10 +236,14 @@ public MVStore getStore() {
return store;
}

public MVCCTransactionEngine getTransactionEngine() {
public TransactionEngine getTransactionEngine() {
return transactionEngine;
}

public StorageMap.Builder getMapBuilder() {
return mapBuilder;
}

public HashMap<String, MVTable> getTables() {
return new HashMap<String, MVTable>(tableMap);
}
Expand Down Expand Up @@ -264,11 +285,11 @@ public void closeImmediately() {
* rollback all open transactions.
*/
public void initTransactions() {
List<MVCCTransaction> list = transactionEngine.getOpenTransactions();
for (MVCCTransaction t : list) {
if (t.getStatus() == MVCCTransaction.STATUS_COMMITTING) {
List<Transaction> list = transactionEngine.getOpenTransactions();
for (Transaction t : list) {
if (t.getStatus() == Transaction.STATUS_COMMITTING) {
t.commit();
} else if (t.getStatus() != MVCCTransaction.STATUS_PREPARED) {
} else if (t.getStatus() != Transaction.STATUS_PREPARED) {
t.rollback();
}
}
Expand All @@ -289,8 +310,8 @@ public void removeTemporaryMaps(BitField objectIds) {
if (!objectIds.get(id)) {
ValueDataType keyType = new ValueDataType(null, null, null);
ValueDataType valueType = new ValueDataType(null, null, null);
MVCCTransaction t = transactionEngine.beginTransaction(false);
MVCCTransactionMap<?, ?> m = t.openMap(mapName, keyType, valueType);
Transaction t = transactionEngine.beginTransaction(false);
TransactionMap<?, ?> m = t.openMap(mapName, keyType, valueType);
transactionEngine.removeMap(m);
t.commit();
}
Expand All @@ -314,17 +335,17 @@ public synchronized String nextTemporaryMapName() {
* @param transactionName the transaction name (may be null)
*/
public void prepareCommit(Session session, String transactionName) {
MVCCTransaction t = (MVCCTransaction) session.getTransaction();
Transaction t = session.getTransaction();
t.setName(transactionName);
t.prepare();
store.commit();
}

public ArrayList<InDoubtTransaction> getInDoubtTransactions() {
List<MVCCTransaction> list = transactionEngine.getOpenTransactions();
List<Transaction> list = transactionEngine.getOpenTransactions();
ArrayList<InDoubtTransaction> result = New.arrayList();
for (MVCCTransaction t : list) {
if (t.getStatus() == MVCCTransaction.STATUS_PREPARED) {
for (Transaction t : list) {
if (t.getStatus() == Transaction.STATUS_PREPARED) {
result.add(new MVInDoubtTransaction(store, t));
}
}
Expand Down Expand Up @@ -443,10 +464,10 @@ public Map<String, Integer> statisticsEnd() {
private static class MVInDoubtTransaction implements InDoubtTransaction {

private final MVStore store;
private final MVCCTransaction transaction;
private final Transaction transaction;
private int state = InDoubtTransaction.IN_DOUBT;

MVInDoubtTransaction(MVStore store, MVCCTransaction transaction) {
MVInDoubtTransaction(MVStore store, Transaction transaction) {
this.store = store;
this.transaction = transaction;
}
Expand Down Expand Up @@ -505,7 +526,7 @@ public String nextTemporaryMapName(Database db) {

@Override
public <K, V> TransactionMap<K, V> openMap(Session session, String name, DataType keyType, DataType valueType) {
return ((MVCCTransaction) session.getTransaction()).openMap(name, keyType, valueType);
return session.getTransaction().openMap(name, keyType, valueType);
}

}
2 changes: 1 addition & 1 deletion lealone-storage/engine/src/main/java/org/lealone/storage/WTStorageEngine.java
View file
Expand Up @@ -53,7 +53,7 @@ public Table createTable(CreateTableData data) {
conn = createConnection(dbName);
connections.put(dbName, conn);

Store store = init(db, new WTMapBuilder(conn.open_session(null)));
Store store = init(db, new WTMapBuilder(conn.open_session(null)), getTransactionEngine());
stores.put(db.getName(), store);
db.setStorageEngine(this);
db.setTransactionEngine(store.getTransactionEngine());
Expand Down
6 changes: 4 additions & 2 deletions lealone-test/src/test/java/org/lealone/test/storage/MemoryStorageEngine.java
View file
Expand Up @@ -52,7 +52,9 @@ public static void register() {
}

public MemoryStorageEngine() {
super(new MemoryMapBuilder());
super();
setMapBuilder(new MemoryMapBuilder());
//setTransactionEngine(new MemoryTransactionEngine());
}

@Override
Expand Down Expand Up @@ -124,7 +126,7 @@ public int compare(K k1, K k2) {

}

static class MemoryMap<K, V> extends ConcurrentSkipListMap<K, V> implements StorageMap<K, V> {
public static class MemoryMap<K, V> extends ConcurrentSkipListMap<K, V> implements StorageMap<K, V> {

private static final AtomicInteger counter = new AtomicInteger(0);

Expand Down
18 changes: 9 additions & 9 deletions ...one/test/transaction/TransactionTest.java → ...test/transaction/MVCCTransactionTest.java
View file
Expand Up @@ -18,33 +18,33 @@
package org.lealone.test.transaction;

import org.junit.Assert;
import org.junit.Test;
import org.lealone.storage.StorageMap;
import org.lealone.test.TestBase;
import org.lealone.test.storage.MemoryStorageEngine;
import org.lealone.transaction.MVCCTransaction;
import org.lealone.transaction.MVCCTransactionEngine;
import org.lealone.transaction.TransactionEngine;
import org.lealone.transaction.Transaction;
import org.lealone.transaction.TransactionMap;
import org.lealone.transaction.MVCCTransactionMap;
import org.lealone.type.ObjectDataType;

public class TransactionTest {
public static void main(String[] args) {
public class MVCCTransactionTest {
@Test
public void run() {
StorageMap.Builder mapBuilder = new MemoryStorageEngine.MemoryMapBuilder();
String hostAndPort = TestBase.getHost() + ":" + TestBase.getPort();
MVCCTransactionEngine e = new MVCCTransactionEngine(new ObjectDataType(), mapBuilder, hostAndPort);
e.init(null);
TransactionEngine te = e;

Transaction t = te.beginTransaction(false);
TransactionMap<String, String> map = t.openMap("test");
MVCCTransaction t = e.beginTransaction(false);
MVCCTransactionMap<String, String> map = t.openMap("test");
map.put("1", "a");
map.put("2", "b");
Assert.assertEquals("a", map.get("1"));
Assert.assertEquals("b", map.get("2"));

t.rollback();

t = te.beginTransaction(false);
t = e.beginTransaction(false);

Assert.assertNull(map.get("1"));
Assert.assertNull(map.get("2"));
Expand Down

0 comments on commit 778f5b5

Please sign in to comment.