more encapsulation changes

git-svn-id: svn+ssh://svn.corp.yahoo.com/yahoo/yrl/labs/pnuts/code/logstore@794 8dad8b1f-cf64-0410-95b6-bcf113ffbcfe

diskTreeComponent.cpp

@@ -54,24 +54,26 @@ void diskTreeComponent::list_regions(int xid, pageid_t *internal_node_region_len
 
 void diskTreeComponent::writes_done() {
   if(dp) {
+    stats->wrote_datapage(dp);
     dp->writes_done();
     delete dp;
     dp = 0;
   }
 }
 
-int diskTreeComponent::insertTuple(int xid, datatuple *t, mergeStats *stats)
+int diskTreeComponent::insertTuple(int xid, datatuple *t)
 {
   int ret = 0; // no error.
   if(dp==0) {
     dp = insertDataPage(xid, t);
-    stats->num_datapages_out++;
+    //    stats->num_datapages_out++;
   } else if(!dp->append(t)) {
-    stats->bytes_out += (PAGE_SIZE * dp->get_page_count());
+    //    stats->bytes_out += (PAGE_SIZE * dp->get_page_count());
+    stats->wrote_datapage(dp);
     dp->writes_done();
     delete dp;
     dp = insertDataPage(xid, t);
-    stats->num_datapages_out++;
+    //    stats->num_datapages_out++;
   }
   return ret;
 }
@@ -89,6 +91,7 @@ DataPage<datatuple>* diskTreeComponent::insertDataPage(int xid, datatuple *tuple
         if(!dp->append(tuple))
         {
             // the last datapage must have not wanted the tuple, and then this datapage figured out the region is full.
+            stats->wrote_datapage(dp);
             dp->writes_done();
             delete dp;
             dp = 0;

diskTreeComponent.h

@@ -17,15 +17,19 @@ class diskTreeComponent {
   class internalNodes;
   class iterator;
 
-  diskTreeComponent(int xid, pageid_t internal_region_size, pageid_t datapage_region_size, pageid_t datapage_size) :
+  diskTreeComponent(int xid, pageid_t internal_region_size, pageid_t datapage_region_size, pageid_t datapage_size,
+		    mergeStats* stats) :
     ltree(new diskTreeComponent::internalNodes(xid, internal_region_size, datapage_region_size, datapage_size)),
     dp(0),
-    datapage_size(datapage_size) {}
+    datapage_size(datapage_size),
+    stats(stats) {}
 
 
-  diskTreeComponent(int xid, recordid root, recordid internal_node_state, recordid datapage_state) :
+  diskTreeComponent(int xid, recordid root, recordid internal_node_state, recordid datapage_state, mergeStats* stats) :
     ltree(new diskTreeComponent::internalNodes(xid, root, internal_node_state, datapage_state)),
-    dp(0) {}
+    dp(0),
+    datapage_size(-1),
+    stats(stats) {}
 
   ~diskTreeComponent() {
     delete dp;
@@ -37,7 +41,7 @@ class diskTreeComponent {
   recordid get_internal_node_allocator_rid();
   internalNodes * get_internal_nodes() { return ltree; }
   datatuple* findTuple(int xid, datatuple::key_t key, size_t keySize);
-  int insertTuple(int xid, datatuple *t, mergeStats *stats);
+  int insertTuple(int xid, datatuple *t);
   void writes_done();
 
 
@@ -69,6 +73,7 @@ class diskTreeComponent {
   internalNodes * ltree;
   DataPage<datatuple>* dp;
   pageid_t datapage_size;
+  mergeStats *stats;
 
  public:
   class internalNodes{

logstore.cpp

@@ -78,12 +78,12 @@ recordid logtable<TUPLE>::allocTable(int xid)
 {
 
     table_rec = Talloc(xid, sizeof(tbl_header));
-    
+    mergeStats * stats = 0;
     //create the big tree
-    tree_c2 = new diskTreeComponent(xid, internal_region_size, datapage_region_size, datapage_size);
+    tree_c2 = new diskTreeComponent(xid, internal_region_size, datapage_region_size, datapage_size, stats);
 
     //create the small tree
-    tree_c1 = new diskTreeComponent(xid, internal_region_size, datapage_region_size, datapage_size);
+    tree_c1 = new diskTreeComponent(xid, internal_region_size, datapage_region_size, datapage_size, stats);
 
     update_persistent_header(xid);
 
@@ -93,8 +93,8 @@ template<class TUPLE>
 void logtable<TUPLE>::openTable(int xid, recordid rid) {
   table_rec = rid;
   Tread(xid, table_rec, &tbl_header);
-  tree_c2 = new diskTreeComponent(xid, tbl_header.c2_root, tbl_header.c2_state, tbl_header.c2_dp_state);
+  tree_c2 = new diskTreeComponent(xid, tbl_header.c2_root, tbl_header.c2_state, tbl_header.c2_dp_state, 0);
-  tree_c1 = new diskTreeComponent(xid, tbl_header.c1_root, tbl_header.c1_state, tbl_header.c1_dp_state);
+  tree_c1 = new diskTreeComponent(xid, tbl_header.c1_root, tbl_header.c1_state, tbl_header.c1_dp_state, 0);
 }
 template<class TUPLE>
 void logtable<TUPLE>::update_persistent_header(int xid) {

merger.cpp

@@ -284,7 +284,7 @@ void* memMergeThread(void*arg)
 
         
         //create a new tree
-        diskTreeComponent * c1_prime = new diskTreeComponent(xid,  a->internal_region_size, a->datapage_region_size, a->datapage_size);
+        diskTreeComponent * c1_prime = new diskTreeComponent(xid,  a->internal_region_size, a->datapage_region_size, a->datapage_size, &stats);
 
         //pthread_mutex_unlock(a->block_ready_mut);
         unlock(ltable->header_lock);
@@ -347,7 +347,7 @@ void* memMergeThread(void*arg)
           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));
+          ltable->set_tree_c1(new diskTreeComponent(xid, a->internal_region_size, a->datapage_region_size, a->datapage_size, &stats));
 
           pthread_cond_signal(a->out_block_ready_cond);
 
@@ -434,7 +434,7 @@ void *diskMergeThread(void*arg)
         diskTreeComponent::iterator *itrB = ltable->get_tree_c1_mergeable()->open_iterator();
 
         //create a new tree
-        diskTreeComponent * c2_prime = new diskTreeComponent(xid, a->internal_region_size, a->datapage_region_size, a->datapage_size);
+        diskTreeComponent * c2_prime = new diskTreeComponent(xid, a->internal_region_size, a->datapage_region_size, a->datapage_size, &stats);
 
         unlock(ltable->header_lock);
 
@@ -510,7 +510,7 @@ void merge_iterators(int xid,
         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);
+            scratch_tree->insertTuple(xid, t1);
             stats->wrote_tuple(t1);
             datatuple::freetuple(t1);
             //advance itrA
@@ -524,7 +524,7 @@ void merge_iterators(int xid,
 
             //insert merged tuple, drop deletes
             if(dropDeletes && !mtuple->isDelete()) {
-              scratch_tree->insertTuple(xid, mtuple, stats);
+              scratch_tree->insertTuple(xid, mtuple);
             }
             datatuple::freetuple(t1);
             t1 = itrA->next_callerFrees();  //advance itrA
@@ -536,7 +536,7 @@ void merge_iterators(int xid,
         else
         {
             //insert t2
-            scratch_tree->insertTuple(xid, t2, stats);
+            scratch_tree->insertTuple(xid, t2);
             // cannot free any tuples here; they may still be read through a lookup
         }
 
@@ -545,7 +545,7 @@ void merge_iterators(int xid,
     }
 
     while(t1 != 0) {// t1 is less than t2
-      scratch_tree->insertTuple(xid, t1, stats);
+      scratch_tree->insertTuple(xid, t1);
       stats->wrote_tuple(t1);
       datatuple::freetuple(t1);
 

test/check_logtable.cpp

@@ -30,13 +30,12 @@ void insertProbeIter(size_t NUM_ENTRIES)
 
     int xid = Tbegin();
 
-    logtable<datatuple> ltable(1000, 10000, 5);
-    recordid table_root = ltable.allocTable(xid);
-
     Tcommit(xid);
 
     xid = Tbegin();
-    diskTreeComponent *ltable_c1 = ltable.get_tree_c1();
+
+    mergeStats stats(1,0);
+    diskTreeComponent *ltable_c1 = new diskTreeComponent(xid, 1000, 10000, 5, &stats);
 
     std::vector<std::string> data_arr;
     std::vector<std::string> key_arr;
@@ -56,25 +55,17 @@ void insertProbeIter(size_t NUM_ENTRIES)
     
     printf("Stage 1: Writing %d keys\n", NUM_ENTRIES);
 
-    mergeStats *stats = (mergeStats*)calloc(sizeof(stats), 1);
-
     for(size_t i = 0; i < NUM_ENTRIES; i++)
     {
         //prepare the tuple
         datatuple* newtuple = datatuple::create(key_arr[i].c_str(), key_arr[i].length()+1, data_arr[i].c_str(), data_arr[i].length()+1);
 
-        stats->bytes_in_small += newtuple->byte_length();
+        ltable_c1->insertTuple(xid, newtuple);
-
-        ltable_c1->insertTuple(xid, newtuple, stats);
-
         datatuple::freetuple(newtuple);
     }
     printf("\nTREE STRUCTURE\n");
     ltable_c1->print_tree(xid);
 
-    printf("Total data set length: %lld\n", stats->bytes_in_small);
-    printf("Storage utilization: %.2f\n", (stats->bytes_in_small+.0) / (1.0* stats->bytes_out));
-    printf("Number of datapages: %lld\n", (long long)stats->num_datapages_out);
     printf("Writes complete.\n");
 
     ltable_c1->writes_done();