Commit 00c7c686 authored by Philipp Götze's avatar Philipp Götze
Browse files

Minor adaptions to table implementations

parent be97a7a6
......@@ -214,6 +214,10 @@ class BOCCTable : public BaseTable,
tblID = sCtx.registerState(this->shared_from_this());
}
TableID getID() const {
return tblID;
}
void transactionBegin(const TransactionID& txnID) {
sCtx.txCntW++;
writeSet.txnID = txnID;
......@@ -227,23 +231,29 @@ class BOCCTable : public BaseTable,
Errc s = Errc::SUCCESS;
thisState = Status::Commit;
if(otherState == Status::Commit) {
if (otherState == Status::Commit) {
this->saveWriteSet();
sCtx.regStates[otherID]->saveWriteSet();
s = this->transactionCommit(txnID);
if (s != Errc::SUCCESS) return s;
s = sCtx.regStates[otherID]->transactionCommit(txnID);
sCtx.setLastCTS(0, txnID);
sCtx.removeTx(txnID);
}
return s;
}
Errc transactionCommit(const TransactionID& txnID) {
void saveWriteSet() {
const auto valTS = sCtx.getNewTS();
/// Save writeSet
dQLock.lockExclusive();
auto &ws = committedWSs.emplace_front(valTS, DTS_INF, writeSet.set.size());
for (const auto& e : writeSet.set)
ws.keys.emplace(e.first);
dQLock.unlockExclusive();
}
Errc transactionCommit(const TransactionID& txnID) {
/// Actual insert to table
#ifdef USE_NVM_TABLES
auto pop = pmem::obj::pool_by_pptr(tbl.q);
......@@ -257,22 +267,23 @@ class BOCCTable : public BaseTable,
tbl.insert(std::move(e.first), std::move(e.second));
}
#endif
auto &ws = committedWSs.front();
ws.endTS = sCtx.getNewTS(); ///< note end of transaction writing in write set
writeSet.clean();
/// Cleanup old write sets [remove all where EndTS < oldest active tx]
// if (committedWSs.size() > 100) {
auto oldestTx = sCtx.getOldestActiveTx();
if (oldestTx == txnID) oldestTx = ws.endTS;
for(auto it = committedWSs.cbegin(); it != committedWSs.cend(); ++it) {
if(it->endTS <= oldestTx) {
dQLock.lockExclusive();
committedWSs.erase(it, committedWSs.cend());
dQLock.unlockExclusive();
return Errc::SUCCESS;
}
// if (committedWSs.size() > 100) {
auto oldestTx = sCtx.getOldestActiveTx();
if (oldestTx == txnID) oldestTx = ws.valTS;
for(auto it = committedWSs.cbegin(); it != committedWSs.cend(); ++it) {
if(it->endTS <= oldestTx) {
dQLock.lockExclusive();
committedWSs.erase(it, committedWSs.cend());
dQLock.unlockExclusive();
return Errc::SUCCESS;
}
// }
}
// }
return Errc::SUCCESS;
}
......@@ -301,7 +312,6 @@ class BOCCTable : public BaseTable,
}
}
dQLock.unlockShared();
return Errc::SUCCESS;
}
......
......@@ -48,8 +48,9 @@
namespace pfabric {
constexpr auto BRANCHSIZE = 32;
constexpr auto LEAFSIZE = 16;
constexpr auto BRANCHSIZE = 30; ///< 1024 Byte
//constexpr auto LEAFSIZE = 24; ///< 1024 Byte
constexpr auto LEAFSIZE = 17; ///< for MVCC 8 - 1024, 17 - 2048
using pmem::obj::delete_persistent;
using pmem::obj::make_persistent;
......
......@@ -179,6 +179,10 @@ class S2PLTable : public BaseTable,
tblID = sCtx.registerState(this->shared_from_this());
}
TableID getID() const {
return tblID;
}
void transactionBegin(const TransactionID& txnID) {
sCtx.txCntW++;
/*
......@@ -200,8 +204,7 @@ class S2PLTable : public BaseTable,
s = this->transactionCommit(txnID);
if (s != Errc::SUCCESS) return s;
s = sCtx.regStates[otherID]->transactionCommit(txnID);
//transaction::commit();
//delete tx;
sCtx.setLastCTS(0, txnID);
sCtx.removeTx(txnID);
}
return s;
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment