greg/stasis-bLSM
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

evil? lucky? whitespace cleanup

Browse source 
git-svn-id: svn+ssh://svn.corp.yahoo.com/yahoo/yrl/labs/pnuts/code/logstore@666 8dad8b1f-cf64-0410-95b6-bcf113ffbcfe
This commit is contained in:
sears 2010-03-05 21:19:52 +00:00
parent b75ad3675c
commit 020f0faf17
2 changed files with 572 additions and 664 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

273
diskTreeComponent.cpp
View file

@ -61,16 +61,13 @@ void diskTreeComponent::init_stasis() {
void diskTreeComponent::deinit_stasis() { Tdeinit(); } void diskTreeComponent::deinit_stasis() { Tdeinit(); }
void diskTreeComponent::free_region_rid(int xid, recordid tree, void diskTreeComponent::free_region_rid(int xid, recordid tree,
diskTreeComponent_page_deallocator_t dealloc, void *allocator_state) diskTreeComponent_page_deallocator_t dealloc, void *allocator_state) {
{
dealloc(xid,allocator_state); dealloc(xid,allocator_state);
// XXX fishy shouldn't caller do this? // XXX fishy shouldn't caller do this?
Tdealloc(xid, *(recordid*)allocator_state); Tdealloc(xid, *(recordid*)allocator_state);
} }
void diskTreeComponent::dealloc_region_rid(int xid, recordid rid) {
void diskTreeComponent::dealloc_region_rid(int xid, recordid rid)
{
RegionAllocConf_t a; RegionAllocConf_t a;
Tread(xid,rid,&a); Tread(xid,rid,&a);
DEBUG("{%lld <- dealloc region arraylist}\n", a.regionList.page); DEBUG("{%lld <- dealloc region arraylist}\n", a.regionList.page);
@ -86,8 +83,7 @@ void diskTreeComponent::dealloc_region_rid(int xid, recordid rid)
} }
void diskTreeComponent::force_region_rid(int xid, recordid rid) void diskTreeComponent::force_region_rid(int xid, recordid rid) {
{
RegionAllocConf_t a; RegionAllocConf_t a;
Tread(xid,rid,&a); Tread(xid,rid,&a);
@ -96,19 +92,18 @@ void diskTreeComponent::force_region_rid(int xid, recordid rid)
a.regionList.slot = i; a.regionList.slot = i;
pageid_t pid; pageid_t pid;
Tread(xid,a.regionList,&pid); Tread(xid,a.regionList,&pid);
stasis_dirty_page_table_flush_range((stasis_dirty_page_table_t*)stasis_runtime_dirty_page_table(), pid, pid+a.regionSize); stasis_dirty_page_table_flush_range(
(stasis_dirty_page_table_t*)stasis_runtime_dirty_page_table(),
pid, pid+a.regionSize);
stasis_buffer_manager_t *bm = stasis_buffer_manager_t *bm =
(stasis_buffer_manager_t*)stasis_runtime_buffer_manager(); (stasis_buffer_manager_t*)stasis_runtime_buffer_manager();
bm->forcePageRange(bm, pid, pid+a.regionSize); bm->forcePageRange(bm, pid, pid+a.regionSize);
} }
} }
pageid_t diskTreeComponent::alloc_region(int xid, void *conf) {
pageid_t diskTreeComponent::alloc_region(int xid, void *conf)
{
RegionAllocConf_t* a = (RegionAllocConf_t*)conf; RegionAllocConf_t* a = (RegionAllocConf_t*)conf;
if(a->nextPage == a->endOfRegion) { if(a->nextPage == a->endOfRegion) {
if(a->regionList.size == -1) { if(a->regionList.size == -1) {
//DEBUG("nextPage: %lld\n", a->nextPage); //DEBUG("nextPage: %lld\n", a->nextPage);
@ -137,7 +132,6 @@ pageid_t diskTreeComponent::alloc_region(int xid, void *conf)
(a->nextPage)++; (a->nextPage)++;
DEBUG("tree %lld-%lld\n", (long long)ret, a->endOfRegion); DEBUG("tree %lld-%lld\n", (long long)ret, a->endOfRegion);
return ret; return ret;
} }
pageid_t diskTreeComponent::alloc_region_rid(int xid, void * ridp) { pageid_t diskTreeComponent::alloc_region_rid(int xid, void * ridp) {
@ -167,10 +161,7 @@ pageid_t * diskTreeComponent::list_region_rid(int xid, void *ridp, pageid_t * re
return ret; return ret;
} }
recordid diskTreeComponent::create(int xid) {
recordid diskTreeComponent::create(int xid)
{
tree_state = Talloc(xid,sizeof(RegionAllocConf_t)); tree_state = Talloc(xid,sizeof(RegionAllocConf_t));
@ -221,19 +212,15 @@ recordid diskTreeComponent::create(int xid)
* I guess this should never happen in rose, but what if? * I guess this should never happen in rose, but what if?
**/ **/
void diskTreeComponent::writeRecord(int xid, Page *p, recordid &rid, void diskTreeComponent::writeRecord(int xid, Page *p, recordid &rid,
const byte *data, size_t datalen) const byte *data, size_t datalen) {
{
byte *byte_arr = stasis_record_write_begin(xid, p, rid); byte *byte_arr = stasis_record_write_begin(xid, p, rid);
memcpy(byte_arr, data, datalen); //TODO: stasis write call memcpy(byte_arr, data, datalen); //TODO: stasis write call
stasis_record_write_done(xid, p, rid, byte_arr); stasis_record_write_done(xid, p, rid, byte_arr);
stasis_page_lsn_write(xid, p, get_lsn(xid)); stasis_page_lsn_write(xid, p, get_lsn(xid));
} }
void diskTreeComponent::writeNodeRecord(int xid, Page * p, recordid & rid, void diskTreeComponent::writeNodeRecord(int xid, Page * p, recordid & rid,
const byte *key, size_t keylen, pageid_t ptr) const byte *key, size_t keylen, pageid_t ptr) {
{
DEBUG("writenoderecord:\tp->id\t%lld\tkey:\t%s\tkeylen: %d\tval_page\t%lld\n", DEBUG("writenoderecord:\tp->id\t%lld\tkey:\t%s\tkeylen: %d\tval_page\t%lld\n",
p->id, datatuple::key_to_str(key).c_str(), keylen, ptr); p->id, datatuple::key_to_str(key).c_str(), keylen, ptr);
indexnode_rec *nr = (indexnode_rec*)stasis_record_write_begin(xid, p, rid); indexnode_rec *nr = (indexnode_rec*)stasis_record_write_begin(xid, p, rid);
@ -244,8 +231,7 @@ void diskTreeComponent::writeNodeRecord(int xid, Page * p, recordid & rid,
} }
void diskTreeComponent::writeRecord(int xid, Page *p, slotid_t slot, void diskTreeComponent::writeRecord(int xid, Page *p, slotid_t slot,
const byte *data, size_t datalen) const byte *data, size_t datalen) {
{
recordid rid; recordid rid;
rid.page = p->id; rid.page = p->id;
rid.slot = slot; rid.slot = slot;
@ -254,17 +240,16 @@ void diskTreeComponent::writeRecord(int xid, Page *p, slotid_t slot,
memcpy(byte_arr, data, datalen); //TODO: stasis write call memcpy(byte_arr, data, datalen); //TODO: stasis write call
stasis_record_write_done(xid, p, rid, byte_arr); stasis_record_write_done(xid, p, rid, byte_arr);
stasis_page_lsn_write(xid, p, get_lsn(xid)); stasis_page_lsn_write(xid, p, get_lsn(xid));
} }
const byte* diskTreeComponent::readRecord(int xid, Page * p, recordid &rid) const byte* diskTreeComponent::readRecord(int xid, Page * p, recordid &rid) {
{
const byte *nr = stasis_record_read_begin(xid,p,rid); // XXX API violation? const byte *nr = stasis_record_read_begin(xid,p,rid); // XXX API violation?
return nr; return nr;
} }
const byte* diskTreeComponent::readRecord(int xid, Page * p, slotid_t slot, int64_t size) const byte* diskTreeComponent::readRecord(int xid, Page * p, slotid_t slot, int64_t size) {
{
recordid rid; recordid rid;
rid.page = p->id; rid.page = p->id;
rid.slot = slot; rid.slot = slot;
@ -273,15 +258,13 @@ const byte* diskTreeComponent::readRecord(int xid, Page * p, slotid_t slot, int6
return nr; return nr;
} }
int32_t diskTreeComponent::readRecordLength(int xid, Page *p, slotid_t slot) int32_t diskTreeComponent::readRecordLength(int xid, Page *p, slotid_t slot) {
{
recordid rec = {p->id, slot, 0}; recordid rec = {p->id, slot, 0};
int32_t reclen = stasis_record_length_read(xid, p, rec); int32_t reclen = stasis_record_length_read(xid, p, rec);
return reclen; return reclen;
} }
void diskTreeComponent::initializeNodePage(int xid, Page *p) void diskTreeComponent::initializeNodePage(int xid, Page *p) {
{
stasis_page_slotted_initialize_page(p); stasis_page_slotted_initialize_page(p);
recordid reserved1 = stasis_record_alloc_begin(xid, p, sizeof(indexnode_rec)); recordid reserved1 = stasis_record_alloc_begin(xid, p, sizeof(indexnode_rec));
stasis_record_alloc_done(xid, p, reserved1); stasis_record_alloc_done(xid, p, reserved1);
@ -293,14 +276,11 @@ void diskTreeComponent::initializeNodePage(int xid, Page *p)
recordid diskTreeComponent::appendPage(int xid, recordid tree, pageid_t & rmLeafID, recordid diskTreeComponent::appendPage(int xid, recordid tree, pageid_t & rmLeafID,
const byte *key, size_t keySize, const byte *key, size_t keySize,
lsm_page_allocator_t allocator, void *allocator_state, lsm_page_allocator_t allocator, void *allocator_state,
long val_page) long val_page) {
{
Page *p = loadPage(xid, tree.page); Page *p = loadPage(xid, tree.page);
writelock(p->rwlatch, 0); writelock(p->rwlatch, 0);
//logtree_state *s = (logtree_state*)p->impl;
tree.slot = 0; tree.slot = 0;
//tree.size = sizeof(lsmTreeNodeRecord)+keySize;
const indexnode_rec *nr = (const indexnode_rec*)readRecord(xid, p , DEPTH, 0); const indexnode_rec *nr = (const indexnode_rec*)readRecord(xid, p , DEPTH, 0);
int64_t depth = *((int64_t*)nr); int64_t depth = *((int64_t*)nr);
@ -311,21 +291,18 @@ recordid diskTreeComponent::appendPage(int xid, recordid tree, pageid_t & rmLeaf
Page *lastLeaf; Page *lastLeaf;
if(rmLeafID != tree.page) if(rmLeafID != tree.page) {
{
lastLeaf= loadPage(xid, rmLeafID); lastLeaf= loadPage(xid, rmLeafID);
writelock(lastLeaf->rwlatch, 0); writelock(lastLeaf->rwlatch, 0);
} else } else {
lastLeaf = p; lastLeaf = p;
}
recordid ret = stasis_record_alloc_begin(xid, lastLeaf, recordid ret = stasis_record_alloc_begin(xid, lastLeaf,
sizeof(indexnode_rec)+keySize); sizeof(indexnode_rec)+keySize);
if(ret.size == INVALID_SLOT) if(ret.size == INVALID_SLOT) {
{ if(lastLeaf->id != p->id) {
if(lastLeaf->id != p->id)
{
assert(rmLeafID != tree.page); assert(rmLeafID != tree.page);
unlock(lastLeaf->rwlatch); unlock(lastLeaf->rwlatch);
releasePage(lastLeaf); // don't need that page anymore... releasePage(lastLeaf); // don't need that page anymore...
@ -341,8 +318,7 @@ recordid diskTreeComponent::appendPage(int xid, recordid tree, pageid_t & rmLeaf
rmLeafID == tree.page ? -1 : rmLeafID, rmLeafID == tree.page ? -1 : rmLeafID,
allocator, allocator_state); allocator, allocator_state);
if(ret.size == INVALID_SLOT) if(ret.size == INVALID_SLOT) {
{
DEBUG("Need to split root; depth = %d\n", depth); DEBUG("Need to split root; depth = %d\n", depth);
pageid_t child = allocator(xid, allocator_state); pageid_t child = allocator(xid, allocator_state);
@ -353,8 +329,7 @@ recordid diskTreeComponent::appendPage(int xid, recordid tree, pageid_t & rmLeaf
//creates a copy of the root page records in the //creates a copy of the root page records in the
//newly allocated child page //newly allocated child page
for(int i = FIRST_SLOT; i < *stasis_page_slotted_numslots_ptr(p); i++) for(int i = FIRST_SLOT; i < *stasis_page_slotted_numslots_ptr(p); i++) {
{
//read the record from the root page //read the record from the root page
const indexnode_rec *nr = (const indexnode_rec*)readRecord(xid,p,i,0); const indexnode_rec *nr = (const indexnode_rec*)readRecord(xid,p,i,0);
int reclen = readRecordLength(xid, p, i); int reclen = readRecordLength(xid, p, i);
@ -375,8 +350,7 @@ recordid diskTreeComponent::appendPage(int xid, recordid tree, pageid_t & rmLeaf
// with the last slot. so thats why i go backwards here. // with the last slot. so thats why i go backwards here.
DEBUG("slots %d (%d) keysize=%lld\n", (int)*stasis_page_slotted_numslots_ptr(p), (int)FIRST_SLOT+1, (long long int)keySize); DEBUG("slots %d (%d) keysize=%lld\n", (int)*stasis_page_slotted_numslots_ptr(p), (int)FIRST_SLOT+1, (long long int)keySize);
assert(*stasis_page_slotted_numslots_ptr(p) >= FIRST_SLOT+1); assert(*stasis_page_slotted_numslots_ptr(p) >= FIRST_SLOT+1);
for(int i = *stasis_page_slotted_numslots_ptr(p)-1; i>FIRST_SLOT; i--) for(int i = *stasis_page_slotted_numslots_ptr(p)-1; i>FIRST_SLOT; i--) {
{
assert(*stasis_page_slotted_numslots_ptr(p) > FIRST_SLOT+1); assert(*stasis_page_slotted_numslots_ptr(p) > FIRST_SLOT+1);
recordid tmp_rec= {p->id, i, INVALID_SIZE}; recordid tmp_rec= {p->id, i, INVALID_SIZE};
stasis_record_free(xid, p, tmp_rec); stasis_record_free(xid, p, tmp_rec);
@ -384,7 +358,6 @@ recordid diskTreeComponent::appendPage(int xid, recordid tree, pageid_t & rmLeaf
//TODO: could change with stasis_slotted_page_initialize(...); //TODO: could change with stasis_slotted_page_initialize(...);
// TODO: fsck? // TODO: fsck?
// stasis_page_slotted_initialize_page(p);
// reinsert first. // reinsert first.
recordid pFirstSlot = { p->id, FIRST_SLOT, readRecordLength(xid, p, FIRST_SLOT)}; recordid pFirstSlot = { p->id, FIRST_SLOT, readRecordLength(xid, p, FIRST_SLOT)};
@ -422,8 +395,7 @@ recordid diskTreeComponent::appendPage(int xid, recordid tree, pageid_t & rmLeaf
assert(ret.size != INVALID_SLOT); assert(ret.size != INVALID_SLOT);
} } else {
else {
DEBUG("Appended new internal node tree depth = %lld key = %s\n", DEBUG("Appended new internal node tree depth = %lld key = %s\n",
depth, datatuple::key_to_str(key).c_str()); depth, datatuple::key_to_str(key).c_str());
} }
@ -432,9 +404,7 @@ recordid diskTreeComponent::appendPage(int xid, recordid tree, pageid_t & rmLeaf
DEBUG("lastleaf is %lld\n", rmLeafID); DEBUG("lastleaf is %lld\n", rmLeafID);
} } else {
else
{
// write the new value to an existing page // write the new value to an existing page
DEBUG("Writing %s\t%d to existing page# %lld\n", datatuple::key_to_str(key).c_str(), DEBUG("Writing %s\t%d to existing page# %lld\n", datatuple::key_to_str(key).c_str(),
val_page, lastLeaf->id); val_page, lastLeaf->id);
@ -475,27 +445,23 @@ recordid diskTreeComponent::appendInternalNode(int xid, Page *p,
const byte *key, size_t key_len, const byte *key, size_t key_len,
pageid_t val_page, pageid_t lastLeaf, pageid_t val_page, pageid_t lastLeaf,
diskTreeComponent_page_allocator_t allocator, diskTreeComponent_page_allocator_t allocator,
void *allocator_state) void *allocator_state) {
{
// assert(*stasis_page_type_ptr(p) == LOGTREE_ROOT_PAGE ||
// *stasis_page_type_ptr(p) == SLOTTED_PAGE);
assert(p->pageType == LOGTREE_ROOT_PAGE || assert(p->pageType == LOGTREE_ROOT_PAGE ||
p->pageType == SLOTTED_PAGE); p->pageType == SLOTTED_PAGE);
DEBUG("appendInternalNode\tdepth %lldkeylen%d\tnumslots %d\n", depth, key_len, *stasis_page_slotted_numslots_ptr(p)); DEBUG("appendInternalNode\tdepth %lldkeylen%d\tnumslots %d\n", depth, key_len, *stasis_page_slotted_numslots_ptr(p));
if(!depth) recordid ret;
{
if(!depth) {
// leaf node. // leaf node.
recordid ret = stasis_record_alloc_begin(xid, p, sizeof(indexnode_rec)+key_len); ret = stasis_record_alloc_begin(xid, p, sizeof(indexnode_rec)+key_len);
if(ret.size != INVALID_SLOT) { if(ret.size != INVALID_SLOT) {
stasis_record_alloc_done(xid, p, ret); stasis_record_alloc_done(xid, p, ret);
writeNodeRecord(xid,p,ret,key,key_len,val_page); writeNodeRecord(xid,p,ret,key,key_len,val_page);
} }
return ret; } else {
}
else
{
// recurse // recurse
int slot = *stasis_page_slotted_numslots_ptr(p)-1;//*recordcount_ptr(p)-1; int slot = *stasis_page_slotted_numslots_ptr(p)-1;//*recordcount_ptr(p)-1;
@ -503,7 +469,6 @@ recordid diskTreeComponent::appendInternalNode(int xid, Page *p,
const indexnode_rec *nr = (const indexnode_rec*)readRecord(xid, p, slot, 0); const indexnode_rec *nr = (const indexnode_rec*)readRecord(xid, p, slot, 0);
pageid_t child_id = nr->ptr; pageid_t child_id = nr->ptr;
nr = 0; nr = 0;
recordid ret;
{ {
Page *child_page = loadPage(xid, child_id); Page *child_page = loadPage(xid, child_id);
writelock(child_page->rwlatch,0); writelock(child_page->rwlatch,0);
@ -514,8 +479,7 @@ recordid diskTreeComponent::appendInternalNode(int xid, Page *p,
releasePage(child_page); releasePage(child_page);
} }
if(ret.size == INVALID_SLOT) // subtree is full; split if(ret.size == INVALID_SLOT) { // subtree is full; split
{
ret = stasis_record_alloc_begin(xid, p, sizeof(indexnode_rec)+key_len); ret = stasis_record_alloc_begin(xid, p, sizeof(indexnode_rec)+key_len);
DEBUG("keylen %d\tnumslots %d for page id %lld ret.size %lld prv rec len %d\n", DEBUG("keylen %d\tnumslots %d for page id %lld ret.size %lld prv rec len %d\n",
key_len, key_len,
@ -523,8 +487,7 @@ recordid diskTreeComponent::appendInternalNode(int xid, Page *p,
p->id, p->id,
ret.size, ret.size,
readRecordLength(xid, p, slot)); readRecordLength(xid, p, slot));
if(ret.size != INVALID_SLOT) if(ret.size != INVALID_SLOT) {
{
stasis_record_alloc_done(xid, p, ret); stasis_record_alloc_done(xid, p, ret);
ret = buildPathToLeaf(xid, ret, p, depth, key, key_len, val_page, ret = buildPathToLeaf(xid, ret, p, depth, key, key_len, val_page,
lastLeaf, allocator, allocator_state); lastLeaf, allocator, allocator_state);
@ -538,16 +501,15 @@ recordid diskTreeComponent::appendInternalNode(int xid, Page *p,
} else { } else {
// we inserted the value in to a subtree rooted here. // we inserted the value in to a subtree rooted here.
} }
return ret;
} }
return ret;
} }
recordid diskTreeComponent::buildPathToLeaf(int xid, recordid root, Page *root_p, recordid diskTreeComponent::buildPathToLeaf(int xid, recordid root, Page *root_p,
int64_t depth, const byte *key, size_t key_len, int64_t depth, const byte *key, size_t key_len,
pageid_t val_page, pageid_t lastLeaf, pageid_t val_page, pageid_t lastLeaf,
diskTreeComponent_page_allocator_t allocator, diskTreeComponent_page_allocator_t allocator,
void *allocator_state) void *allocator_state) {
{
// root is the recordid on the root page that should point to the // root is the recordid on the root page that should point to the
// new subtree. // new subtree.
@ -580,7 +542,6 @@ recordid diskTreeComponent::buildPathToLeaf(int xid, recordid root, Page *root_p
// backward link.//these writes do not need alloc_begin as it is done in page initialization // backward link.//these writes do not need alloc_begin as it is done in page initialization
writeRecord(xid, child_p, PREV_LEAF, (byte*)(&lastLeaf), root_rec_size); writeRecord(xid, child_p, PREV_LEAF, (byte*)(&lastLeaf), root_rec_size);
//writeNodeRecord(xid,child_p,PREV_LEAF,dummy,key_len,lastLeaf);
// forward link (initialize to -1) // forward link (initialize to -1)
@ -600,8 +561,7 @@ recordid diskTreeComponent::buildPathToLeaf(int xid, recordid root, Page *root_p
unlock(child_p->rwlatch); unlock(child_p->rwlatch);
releasePage(child_p); releasePage(child_p);
if(lastLeaf != -1) if(lastLeaf != -1) {
{
// install forward link in previous page // install forward link in previous page
Page *lastLeafP = loadPage(xid, lastLeaf); Page *lastLeafP = loadPage(xid, lastLeaf);
writelock(lastLeafP->rwlatch,0); writelock(lastLeafP->rwlatch,0);
@ -619,21 +579,15 @@ recordid diskTreeComponent::buildPathToLeaf(int xid, recordid root, Page *root_p
} }
/** /**
* Traverse from the root of the page to the right most leaf (the one * Traverse from the root of the page to the right most leaf (the one
* with the higest base key value). * with the higest base key value).
**/ **/
pageid_t diskTreeComponent::findLastLeaf(int xid, Page *root, int64_t depth) pageid_t diskTreeComponent::findLastLeaf(int xid, Page *root, int64_t depth) {
{ if(!depth) {
if(!depth)
{
DEBUG("Found last leaf = %lld\n", root->id); DEBUG("Found last leaf = %lld\n", root->id);
return root->id; return root->id;
} } else {
else
{
const indexnode_rec *nr = (indexnode_rec*) readRecord(xid, root, const indexnode_rec *nr = (indexnode_rec*) readRecord(xid, root,
(*stasis_page_slotted_numslots_ptr(root))-1, 0); (*stasis_page_slotted_numslots_ptr(root))-1, 0);
pageid_t ret; pageid_t ret;
@ -653,12 +607,11 @@ pageid_t diskTreeComponent::findLastLeaf(int xid, Page *root, int64_t depth)
* Traverse from the root of the tree to the left most (lowest valued * Traverse from the root of the tree to the left most (lowest valued
* key) leaf. * key) leaf.
*/ */
pageid_t diskTreeComponent::findFirstLeaf(int xid, Page *root, int64_t depth) pageid_t diskTreeComponent::findFirstLeaf(int xid, Page *root, int64_t depth) {
{
if(!depth) //if depth is 0, then returns the id of the page if(!depth) { //if depth is 0, then returns the id of the page
return root->id; return root->id;
else } else {
{
const indexnode_rec *nr = (indexnode_rec*)readRecord(xid,root,FIRST_SLOT,0); const indexnode_rec *nr = (indexnode_rec*)readRecord(xid,root,FIRST_SLOT,0);
Page *p = loadPage(xid, nr->ptr); Page *p = loadPage(xid, nr->ptr);
readlock(p->rwlatch,0); readlock(p->rwlatch,0);
@ -670,15 +623,15 @@ pageid_t diskTreeComponent::findFirstLeaf(int xid, Page *root, int64_t depth)
} }
pageid_t diskTreeComponent::findPage(int xid, recordid tree, const byte *key, size_t keySize) pageid_t diskTreeComponent::findPage(int xid, recordid tree, const byte *key, size_t keySize) {
{
Page *p = loadPage(xid, tree.page); Page *p = loadPage(xid, tree.page);
readlock(p->rwlatch,0); readlock(p->rwlatch,0);
int64_t depth = *(int64_t*)readRecord(xid, p , DEPTH, 0); int64_t depth = *(int64_t*)readRecord(xid, p , DEPTH, 0);
recordid rid = lookup(xid, p, depth, key, keySize); recordid rid = lookup(xid, p, depth, key, keySize);
pageid_t ret = lookupLeafPageFromRid(xid,rid);//,keySize); pageid_t ret = lookupLeafPageFromRid(xid,rid);
unlock(p->rwlatch); unlock(p->rwlatch);
releasePage(p); releasePage(p);
@ -686,11 +639,10 @@ pageid_t diskTreeComponent::findPage(int xid, recordid tree, const byte *key, si
} }
pageid_t diskTreeComponent::lookupLeafPageFromRid(int xid, recordid rid) pageid_t diskTreeComponent::lookupLeafPageFromRid(int xid, recordid rid) {
{
pageid_t pid = -1; pageid_t pid = -1;
if(rid.page != NULLRID.page || rid.slot != NULLRID.slot) if(rid.page != NULLRID.page || rid.slot != NULLRID.slot) {
{
Page * p2 = loadPage(xid, rid.page); Page * p2 = loadPage(xid, rid.page);
readlock(p2->rwlatch,0); readlock(p2->rwlatch,0);
pid = ((const indexnode_rec*)(readRecord(xid,p2,rid.slot,0)))->ptr; pid = ((const indexnode_rec*)(readRecord(xid,p2,rid.slot,0)))->ptr;
@ -700,35 +652,33 @@ pageid_t diskTreeComponent::lookupLeafPageFromRid(int xid, recordid rid)
return pid; return pid;
} }
recordid diskTreeComponent::lookup(int xid, recordid diskTreeComponent::lookup(int xid,
Page *node, Page *node,
int64_t depth, int64_t depth,
const byte *key, size_t keySize ) const byte *key, size_t keySize ) {
{
//DEBUG("lookup: pid %lld\t depth %lld\n", node->id, depth);
if(*stasis_page_slotted_numslots_ptr(node) == FIRST_SLOT)
return NULLRID;
//DEBUG("lookup: pid %lld\t depth %lld\n", node->id, depth);
if(*stasis_page_slotted_numslots_ptr(node) == FIRST_SLOT) {
return NULLRID;
}
assert(*stasis_page_slotted_numslots_ptr(node) > FIRST_SLOT); assert(*stasis_page_slotted_numslots_ptr(node) > FIRST_SLOT);
// don't need to compare w/ first item in tree, since we need to position ourselves at the the max tree value <= key. // don't need to compare w/ first item in tree, since we need to position ourselves at the the max tree value <= key.
// positioning at FIRST_SLOT puts us "before" the first value // positioning at FIRST_SLOT puts us "before" the first value
int match = FIRST_SLOT; // (so match is now < key) int match = FIRST_SLOT; // (so match is now < key)
for(int i = FIRST_SLOT+1; i < *stasis_page_slotted_numslots_ptr(node); i++) for(int i = FIRST_SLOT+1; i < *stasis_page_slotted_numslots_ptr(node); i++) {
{
const indexnode_rec *rec = (const indexnode_rec*)readRecord(xid,node,i,0); const indexnode_rec *rec = (const indexnode_rec*)readRecord(xid,node,i,0);
int cmpval = datatuple::compare((datatuple::key_t) (rec+1), *stasis_page_slotted_slot_length_ptr(node, i)-sizeof(*rec), int cmpval = datatuple::compare((datatuple::key_t) (rec+1), *stasis_page_slotted_slot_length_ptr(node, i)-sizeof(*rec),
(datatuple::key_t) key, keySize); (datatuple::key_t) key, keySize);
if(cmpval>0) // key of current node is too big; there can be no matches under it.
break; // key of current node is too big; there can be no matches under it.
if(cmpval>0) break;
match = i; // only increment match after comparing with the current node. match = i; // only increment match after comparing with the current node.
} }
if(depth) {
if(depth)
{
pageid_t child_id = ((const indexnode_rec*)readRecord(xid,node,match,0))->ptr; pageid_t child_id = ((const indexnode_rec*)readRecord(xid,node,match,0))->ptr;
Page* child_page = loadPage(xid, child_id); Page* child_page = loadPage(xid, child_id);
readlock(child_page->rwlatch,0); readlock(child_page->rwlatch,0);
@ -736,17 +686,13 @@ recordid diskTreeComponent::lookup(int xid,
unlock(child_page->rwlatch); unlock(child_page->rwlatch);
releasePage(child_page); releasePage(child_page);
return ret; return ret;
} } else {
else
{
recordid ret = {node->id, match, keySize}; recordid ret = {node->id, match, keySize};
return ret; return ret;
} }
} }
void diskTreeComponent::print_tree(int xid) {
void diskTreeComponent::print_tree(int xid)
{
Page *p = loadPage(xid, root_rec.page); Page *p = loadPage(xid, root_rec.page);
readlock(p->rwlatch,0); readlock(p->rwlatch,0);
@ -761,60 +707,48 @@ void diskTreeComponent::print_tree(int xid)
} }
void diskTreeComponent::print_tree(int xid, pageid_t pid, int64_t depth) void diskTreeComponent::print_tree(int xid, pageid_t pid, int64_t depth) {
{
Page *node = loadPage(xid, pid); Page *node = loadPage(xid, pid);
readlock(node->rwlatch,0); readlock(node->rwlatch,0);
//const indexnode_rec *depth_nr = (const indexnode_rec*)readRecord(xid, p , DEPTH, 0);
printf("page_id:%lld\tnum_slots:%d\t\n", node->id, *stasis_page_slotted_numslots_ptr(node)); printf("page_id:%lld\tnum_slots:%d\t\n", node->id, *stasis_page_slotted_numslots_ptr(node));
if(*stasis_page_slotted_numslots_ptr(node) == FIRST_SLOT) if(*stasis_page_slotted_numslots_ptr(node) == FIRST_SLOT) {
return; return;
}
assert(*stasis_page_slotted_numslots_ptr(node) > FIRST_SLOT); assert(*stasis_page_slotted_numslots_ptr(node) > FIRST_SLOT);
if(depth) if(depth) {
{
printf("\tnot_leaf\n"); printf("\tnot_leaf\n");
for(int i = FIRST_SLOT; i < *stasis_page_slotted_numslots_ptr(node); i++) for(int i = FIRST_SLOT; i < *stasis_page_slotted_numslots_ptr(node); i++) {
{
const indexnode_rec *nr = (const indexnode_rec*)readRecord(xid,node,i,0); const indexnode_rec *nr = (const indexnode_rec*)readRecord(xid,node,i,0);
printf("\tchild_page_id:%lld\tkey:%s\n", nr->ptr, printf("\tchild_page_id:%lld\tkey:%s\n", nr->ptr,
datatuple::key_to_str((byte*)(nr+1)).c_str()); datatuple::key_to_str((byte*)(nr+1)).c_str());
} }
for(int i = FIRST_SLOT; i < *stasis_page_slotted_numslots_ptr(node); i++) for(int i = FIRST_SLOT; i < *stasis_page_slotted_numslots_ptr(node); i++) {
{
const indexnode_rec *nr = (const indexnode_rec*)readRecord(xid,node,i,0); const indexnode_rec *nr = (const indexnode_rec*)readRecord(xid,node,i,0);
print_tree(xid, nr->ptr, depth-1); print_tree(xid, nr->ptr, depth-1);
} }
} } else {
else
{
printf("\tis_leaf\t\n"); printf("\tis_leaf\t\n");
const indexnode_rec *nr = (const indexnode_rec*)readRecord(xid,node,FIRST_SLOT,0); const indexnode_rec *nr = (const indexnode_rec*)readRecord(xid,node,FIRST_SLOT,0);
printf("\tdata_page_id:%lld\tkey:%s\n", nr->ptr, printf("\tdata_page_id:%lld\tkey:%s\n", nr->ptr,
datatuple::key_to_str((byte*)(nr+1)).c_str()); datatuple::key_to_str((byte*)(nr+1)).c_str());
printf("\t...\n"); printf("\t...\n");
nr = (const indexnode_rec*)readRecord(xid,node,(*stasis_page_slotted_numslots_ptr(node))-1,0); nr = (const indexnode_rec*)readRecord(xid,node,(*stasis_page_slotted_numslots_ptr(node))-1,0);
printf("\tdata_page_id:%lld\tkey:%s\n", nr->ptr, printf("\tdata_page_id:%lld\tkey:%s\n", nr->ptr,
datatuple::key_to_str((byte*)(nr+1)).c_str()); datatuple::key_to_str((byte*)(nr+1)).c_str());
} }
unlock(node->rwlatch); unlock(node->rwlatch);
releasePage(node); releasePage(node);
} }
///////////////////////////////////////////////// /////////////////////////////////////////////////
@ -829,7 +763,6 @@ lladdIterator_t* diskTreeComponentIterator::open(int xid, recordid root)
Page *p = loadPage(xid,root.page); Page *p = loadPage(xid,root.page);
readlock(p->rwlatch,0); readlock(p->rwlatch,0);
//size_t keySize = getKeySize(xid,p);
DEBUG("ROOT_REC_SIZE %d\n", diskTreeComponent::root_rec_size); DEBUG("ROOT_REC_SIZE %d\n", diskTreeComponent::root_rec_size);
const byte * nr = diskTreeComponent::readRecord(xid,p, const byte * nr = diskTreeComponent::readRecord(xid,p,
diskTreeComponent::DEPTH, diskTreeComponent::DEPTH,
@ -838,26 +771,27 @@ lladdIterator_t* diskTreeComponentIterator::open(int xid, recordid root)
DEBUG("DEPTH = %lld\n", depth); DEBUG("DEPTH = %lld\n", depth);
pageid_t leafid = diskTreeComponent::findFirstLeaf(xid, p, depth); pageid_t leafid = diskTreeComponent::findFirstLeaf(xid, p, depth);
if(leafid != root.page) if(leafid != root.page) {
{
unlock(p->rwlatch); unlock(p->rwlatch);
releasePage(p); releasePage(p);
p = loadPage(xid,leafid); p = loadPage(xid,leafid);
readlock(p->rwlatch,0); readlock(p->rwlatch,0);
assert(depth != 0); assert(depth != 0);
} } else {
else
assert(depth == 0); assert(depth == 0);
}
diskTreeComponentIterator_t *impl = (diskTreeComponentIterator_t*)malloc(sizeof(diskTreeComponentIterator_t)); diskTreeComponentIterator_t *impl = (diskTreeComponentIterator_t*)malloc(sizeof(diskTreeComponentIterator_t));
impl->p = p; impl->p = p;
{ {
// Position just before the first slot. // Position just before the first slot.
// The first call to next() will increment us to the first slot, or return NULL. // The first call to next() will increment us to the first slot, or return NULL.
recordid rid = { p->id, diskTreeComponent::FIRST_SLOT-1, 0}; recordid rid = { p->id, diskTreeComponent::FIRST_SLOT-1, 0};
impl->current = rid; impl->current = rid;
} }
DEBUG("keysize = %d, slot = %d\n", keySize, impl->current.slot); DEBUG("keysize = %d, slot = %d\n", keySize, impl->current.slot);
impl->t = 0; impl->t = 0;
impl->justOnePage = (depth == 0); impl->justOnePage = (depth == 0);
@ -924,8 +858,7 @@ int diskTreeComponentIterator::next(int xid, lladdIterator_t *it)
impl->current = stasis_record_next(xid, impl->p, impl->current); impl->current = stasis_record_next(xid, impl->p, impl->current);
if(impl->current.size == INVALID_SLOT) if(impl->current.size == INVALID_SLOT) {
{
const indexnode_rec next_rec = *(const indexnode_rec*)diskTreeComponent::readRecord(xid,impl->p, const indexnode_rec next_rec = *(const indexnode_rec*)diskTreeComponent::readRecord(xid,impl->p,
diskTreeComponent::NEXT_LEAF, diskTreeComponent::NEXT_LEAF,
@ -936,8 +869,7 @@ int diskTreeComponentIterator::next(int xid, lladdIterator_t *it)
DEBUG("done with page %lld next = %lld\n", impl->p->id, next_rec.ptr); DEBUG("done with page %lld next = %lld\n", impl->p->id, next_rec.ptr);
if(next_rec.ptr != -1 && ! impl->justOnePage) if(next_rec.ptr != -1 && ! impl->justOnePage) {
{
impl->p = loadPage(xid, next_rec.ptr); impl->p = loadPage(xid, next_rec.ptr);
readlock(impl->p->rwlatch,0); readlock(impl->p->rwlatch,0);
impl->current.page = next_rec.ptr; impl->current.page = next_rec.ptr;
@ -949,49 +881,32 @@ int diskTreeComponentIterator::next(int xid, lladdIterator_t *it)
} }
} }
else
{
/*
assert(impl->current.size == keySize + sizeof(lsmTreeNodeRecord));
impl->current.size = keySize;
*/
}
if(impl->current.size != INVALID_SLOT) {
if(impl->current.size != INVALID_SLOT) if(impl->t != NULL) free(impl->t);
{
//size_t sz = sizeof(*impl->t) + impl->current.size;
if(impl->t != NULL)
free(impl->t);
impl->t = (indexnode_rec*)malloc(impl->current.size); impl->t = (indexnode_rec*)malloc(impl->current.size);
memcpy(impl->t, diskTreeComponent::readRecord(xid,impl->p,impl->current), impl->current.size); memcpy(impl->t, diskTreeComponent::readRecord(xid,impl->p,impl->current), impl->current.size);
return 1; return 1;
} } else {
else
{
assert(!impl->p); assert(!impl->p);
if(impl->t != NULL) if(impl->t != NULL) free(impl->t);
free(impl->t);
impl->t = 0; impl->t = 0;
return 0; return 0;
} }
} }
void diskTreeComponentIterator::close(int xid, lladdIterator_t *it) void diskTreeComponentIterator::close(int xid, lladdIterator_t *it) {
{
diskTreeComponentIterator_t *impl = (diskTreeComponentIterator_t*)it->impl; diskTreeComponentIterator_t *impl = (diskTreeComponentIterator_t*)it->impl;
if(impl->p)
{ if(impl->p) {
unlock(impl->p->rwlatch); unlock(impl->p->rwlatch);
releasePage(impl->p); releasePage(impl->p);
} }
if(impl->t) if(impl->t) free(impl->t);
{
free(impl->t);
}
free(impl); free(impl);
free(it); free(it);
} }

17
diskTreeComponent.h
View file

@ -58,7 +58,8 @@ private:
public: public:
static pageid_t alloc_region_rid(int xid, void * ridp); static pageid_t alloc_region_rid(int xid, void * ridp);
static void force_region_rid(int xid, recordid rid); static void force_region_rid(int xid, recordid rid);
static pageid_t*list_region_rid(int xid, void * ridp, pageid_t * region_len, pageid_t * region_count); static pageid_t*list_region_rid(int xid, void * ridp,
pageid_t * region_len, pageid_t * region_count);
static void dealloc_region_rid(int xid, recordid rid); static void dealloc_region_rid(int xid, recordid rid);
static void free_region_rid(int xid, recordid tree, static void free_region_rid(int xid, recordid tree,
diskTreeComponent_page_deallocator_t dealloc, diskTreeComponent_page_deallocator_t dealloc,
@ -139,13 +140,11 @@ private:
void print_tree(int xid, pageid_t pid, int64_t depth); void print_tree(int xid, pageid_t pid, int64_t depth);
private:
recordid tree_state; recordid tree_state;
recordid root_rec; recordid root_rec;
DataPage<datatuple>::RegionAllocator* region_alloc; DataPage<datatuple>::RegionAllocator* region_alloc;
}; };
@ -157,8 +156,7 @@ typedef struct {
} diskTreeComponentIterator_t; } diskTreeComponentIterator_t;
class diskTreeComponentIterator class diskTreeComponentIterator {
{
public: public:
static lladdIterator_t* open(int xid, recordid root); static lladdIterator_t* open(int xid, recordid root);
@ -167,16 +165,13 @@ public:
static void close(int xid, lladdIterator_t *it); static void close(int xid, lladdIterator_t *it);
static inline size_t key (int xid, lladdIterator_t *it, byte **key) static inline size_t key (int xid, lladdIterator_t *it, byte **key) {
{
diskTreeComponentIterator_t * impl = (diskTreeComponentIterator_t*)it->impl; diskTreeComponentIterator_t * impl = (diskTreeComponentIterator_t*)it->impl;
*key = (byte*)(impl->t+1); *key = (byte*)(impl->t+1);
return impl->current.size - sizeof(indexnode_rec); return impl->current.size - sizeof(indexnode_rec);
} }
static inline size_t value(int xid, lladdIterator_t *it, byte **value) {
static inline size_t value(int xid, lladdIterator_t *it, byte **value)
{
diskTreeComponentIterator_t * impl = (diskTreeComponentIterator_t*)it->impl; diskTreeComponentIterator_t * impl = (diskTreeComponentIterator_t*)it->impl;
*value = (byte*)&(impl->t->ptr); *value = (byte*)&(impl->t->ptr);
return sizeof(impl->t->ptr); return sizeof(impl->t->ptr);
@ -187,6 +182,4 @@ public:
}; };
#endif /* DISKTREECOMPONENT_H_ */ #endif /* DISKTREECOMPONENT_H_ */