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

fixed rounding errors in statistics written to stdout; whitespace cleanup; more encapsulation

Browse source 
git-svn-id: svn+ssh://svn.corp.yahoo.com/yahoo/yrl/labs/pnuts/code/logstore@793 8dad8b1f-cf64-0410-95b6-bcf113ffbcfe
This commit is contained in:
sears 2010-04-28 21:29:15 +00:00
parent 151373d09a
commit ded9cc2585
1 changed files with 36 additions and 57 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

93
merger.cpp
View file

@ -204,7 +204,7 @@ void merge_iterators(int xid,
2: wait for c0_mergable
3: begin
4: merge c0_mergable and c1 into c1' # Blocks; tree must be consistent at this point
5: force c1' # Blocks
5: force c1' # Blocks
6: if c1' is too big # Blocks; tree must be consistent at this point.
7: c1_mergable = c1'
8: c1 = new_empty
@ -270,7 +270,7 @@ void* memMergeThread(void*arg)
break;
}
gettimeofday(&stats.start, 0);
stats.starting_merge();
// 3: Begin transaction
xid = Tbegin();
@ -317,7 +317,7 @@ void* memMergeThread(void*arg)
(a->max_size && new_c1_size > a->max_size );
if( signal_c2 )
{
DEBUG("mmt:\tsignaling C2 for merge\n");
DEBUG("mmt:\tsignaling C2 for merge\n");
DEBUG("mmt:\tnew_c1_size %.2f\tMAX_C0_SIZE %lld\ta->max_size %lld\t targetr %.2f \n", new_c1_size,
ltable->max_c0_size, a->max_size, target_R);
@ -343,17 +343,17 @@ void* memMergeThread(void*arg)
if( signal_c2 ) {
// 7: and perhaps c1_mergeable
ltable->set_tree_c1_mergeable(c1_prime);
// 7: and perhaps c1_mergeable
ltable->set_tree_c1_mergeable(c1_prime);
// 8: c1 = new empty.
ltable->set_tree_c1(new diskTreeComponent(xid, a->internal_region_size, a->datapage_region_size, a->datapage_size));
// 8: c1 = new empty.
ltable->set_tree_c1(new diskTreeComponent(xid, a->internal_region_size, a->datapage_region_size, a->datapage_size));
pthread_cond_signal(a->out_block_ready_cond);
pthread_cond_signal(a->out_block_ready_cond);
} else {
// 10: c1 = c1'
ltable->set_tree_c1(c1_prime);
// 10: c1 = c1'
ltable->set_tree_c1(c1_prime);
}
DEBUG("mmt:\tUpdated C1's position on disk to %lld\n",ltable->get_tree_c1()->get_root_rec().page);
@ -362,8 +362,8 @@ void* memMergeThread(void*arg)
Tcommit(xid);
unlock(ltable->header_lock);
gettimeofday(&stats.done, 0);
stats.finished_merge();
stats.pretty_print(stdout);
//TODO: get the freeing outside of the lock
@ -376,7 +376,7 @@ void* memMergeThread(void*arg)
void *diskMergeThread(void*arg)
{
int xid;// = Tbegin();
int xid;
merger_args * a = (merger_args*)(arg);
@ -423,9 +423,9 @@ void *diskMergeThread(void*arg)
break;
}
gettimeofday(&stats.start, 0);
stats.starting_merge();
// 3: begin
// 3: begin
xid = Tbegin();
// 4: do the merge.
@ -437,14 +437,14 @@ void *diskMergeThread(void*arg)
diskTreeComponent * c2_prime = new diskTreeComponent(xid, a->internal_region_size, a->datapage_region_size, a->datapage_size);
unlock(ltable->header_lock);
//do the merge
//do the merge
DEBUG("dmt:\tMerging:\n");
merge_iterators<typeof(*itrA),typeof(*itrB)>(xid, itrA, itrB, ltable, c2_prime, &stats, true);
delete itrA;
delete itrB;
delete itrB;
//5: force write the new region to disk
c2_prime->force(xid);
@ -480,15 +480,13 @@ void *diskMergeThread(void*arg)
unlock(ltable->header_lock);
gettimeofday(&stats.done, 0);
stats.finished_merge();
stats.pretty_print(stdout);
}
return 0;
}
template <class ITA, class ITB>
void merge_iterators(int xid,
ITA *itrA, //iterator on c1 or c2
@ -498,46 +496,32 @@ void merge_iterators(int xid,
bool dropDeletes // should be true iff this is biggest component
)
{
stats->bytes_out = 0;
stats->num_tuples_out = 0;
stats->bytes_in_small = 0;
stats->num_tuples_in_small = 0;
stats->bytes_in_large = 0;
stats->num_tuples_in_large = 0;
datatuple *t1 = itrA->next_callerFrees();
if(t1) {
stats->num_tuples_in_large++;
stats->bytes_in_large += t1->byte_length();
}
stats->read_tuple_from_large_component(t1);
datatuple *t2 = 0;
while( (t2=itrB->next_callerFrees()) != 0)
{
stats->num_tuples_in_small++;
stats->bytes_in_small += t2->byte_length();
{
stats->read_tuple_from_small_component(t2);
DEBUG("tuple\t%lld: keylen %d datalen %d\n",
ntuples, *(t2->keylen),*(t2->datalen) );
ntuples, *(t2->keylen),*(t2->datalen) );
while(t1 != 0 && datatuple::compare(t1->key(), t1->keylen(), t2->key(), t2->keylen()) < 0) // t1 is less than t2
{
//insert t1
scratch_tree->insertTuple(xid, t1, stats);
stats->wrote_tuple(t1);
datatuple::freetuple(t1);
stats->num_tuples_out++;
//advance itrA
t1 = itrA->next_callerFrees();
if(t1) {
stats->num_tuples_in_large++;
stats->bytes_in_large += t1->byte_length();
}
stats->read_tuple_from_large_component(t1);
}
if(t1 != 0 && datatuple::compare(t1->key(), t1->keylen(), t2->key(), t2->keylen()) == 0)
{
datatuple *mtuple = ltable->gettuplemerger()->merge(t1,t2);
//insert merged tuple, drop deletes
if(dropDeletes && !mtuple->isDelete()) {
scratch_tree->insertTuple(xid, mtuple, stats);
@ -545,36 +529,31 @@ void merge_iterators(int xid,
datatuple::freetuple(t1);
t1 = itrA->next_callerFrees(); //advance itrA
if(t1) {
stats->num_tuples_in_large++;
stats->bytes_in_large += t1->byte_length();
stats->read_tuple_from_large_component(t1);
}
datatuple::freetuple(mtuple);
}
else
{
{
//insert t2
scratch_tree->insertTuple(xid, t2, stats);
// cannot free any tuples here; they may still be read through a lookup
}
stats->wrote_tuple(t2);
datatuple::freetuple(t2);
stats->num_tuples_out++;
}
while(t1 != 0) {// t1 is less than t2
scratch_tree->insertTuple(xid, t1, stats);
stats->wrote_tuple(t1);
datatuple::freetuple(t1);
datatuple::freetuple(t1);
stats->num_tuples_out++;
//advance itrA
t1 = itrA->next_callerFrees();
if(t1) {
stats->num_tuples_in_large++;
stats->bytes_in_large += t1->byte_length();
}
//advance itrA
t1 = itrA->next_callerFrees();
stats->read_tuple_from_large_component(t1);
}
DEBUG("dpages: %d\tnpages: %d\tntuples: %d\n", dpages, npages, ntuples);
scratch_tree->writes_done();
}