Got versioning iterator to work.

src/stasis/operations/lsmIterators.h

@@ -341,7 +341,7 @@ class versioningIterator {
     a_(a),
     aend_(aend),
     check_tombstone_(0),
-    tombstone_(0),
+    tombstone_(),
     off_(0)
   {}
   explicit versioningIterator(versioningIterator &i) :
@@ -352,11 +352,11 @@ class versioningIterator {
     off_(i.off_)
   {}
 
-  ROW& operator* () {
+  const ROW& operator* () {
     return *a_;
   }
   void seekEnd() {
-    a_ = aend_; // XXX good idea?
+    a_.seekEnd();// = aend_; // XXX good idea?
   }
   inline bool operator==(const versioningIterator &o) const {
     return a_ == o.a_;
@@ -372,7 +372,7 @@ class versioningIterator {
     } else {
       ++a_;
     }
-    if((*a_).tombstone()) {
+    if(a_ != aend_ && (*a_).tombstone()) {
       tombstone_.copyFrom(*a_);
       check_tombstone_ = 1;
     } else {

src/stasis/operations/lsmTable.h

@@ -19,11 +19,6 @@ namespace rose {
      dispatched), interface to the underlying primititves
   */
 
-  // Lower total work by perfomrming one merge at higher level
-  // for every FUDGE^2 merges at the immediately lower level.
-  // (Constrast to R, which controls the ratio of sizes of the trees.)
-  static const int FUDGE = 1;
-
   template<class PAGELAYOUT, class ITERA, class ITERB>
     struct merge_args {
       int worker_id;
@@ -69,7 +64,8 @@ namespace rose {
       rose::slot_index_t ret = mc->append(xid, *i);
 
       if(ret == rose::NOSPACE) {
-	p->dirty = 1;
+	dirtyPages_add(p);
+	//	p->dirty = 1;
 	mc->pack();
 	releasePage(p);
 	next_page = pageAlloc(xid,pageAllocState);
@@ -82,14 +78,29 @@ namespace rose {
       }
       (*inserted)++;
     }
-
-    p->dirty = 1;
+    dirtyPages_add(p);
+    //    p->dirty = 1;
     mc->pack();
     releasePage(p);
     return pageCount;
   }
 
 
+  // How many bytes of tuples can we afford to keep in RAM?
+  // this is just a guessed value... it seems about right based on
+  // experiments, but 450 bytes overhead per tuple is insane!
+  static const int RB_TREE_OVERHEAD = 400; // = 450;
+  static const pageid_t MEM_SIZE = 1000 * 1000 * 1000;
+  // How many pages should we try to fill with the first C1 merge?
+  static const int R = 3; // XXX set this as low as possible (for dynamic setting.  = sqrt(C2 size / C0 size))
+  static const pageid_t START_SIZE = MEM_SIZE * R /( PAGE_SIZE * 4); //10 * 1000; /*10 **/ //1000; // XXX 4 is fudge related to RB overhead.
+  // Lower total work by perfomrming one merge at higher level
+  // for every FUDGE^2 merges at the immediately lower level.
+  // (Constrast to R, which controls the ratio of sizes of the trees.)
+  static const int FUDGE = 1;
+
+
+
   /**
      ITERA is an iterator over the data structure that mergeThread creates (a lsm tree iterator).
      ITERB is an iterator over the data structures that mergeThread takes as input (lsm tree, or rb tree..)
@@ -155,11 +166,25 @@ namespace rose {
 	mEnd(taBegin, tbBegin, *taEnd, *tbEnd);
 
       mEnd.seekEnd();
+
+      versioningIterator<mergeIterator
+	<ITERA,ITERB,typename PAGELAYOUT::FMT::TUP>,
+	typename PAGELAYOUT::FMT::TUP> vBegin(mBegin,mEnd,0);
+
+      versioningIterator<mergeIterator
+	<ITERA,ITERB,typename PAGELAYOUT::FMT::TUP>,
+	typename PAGELAYOUT::FMT::TUP> vEnd(mBegin,mEnd,0);
+
+      vEnd.seekEnd();
+
       uint64_t insertedTuples;
 
       pageid_t mergedPages = compressData
+	<PAGELAYOUT,versioningIterator<mergeIterator<ITERA,ITERB,typename PAGELAYOUT::FMT::TUP>, typename PAGELAYOUT::FMT::TUP> >
+	(xid, &vBegin, &vEnd,tree->r_,a->pageAlloc,a->pageAllocState,&insertedTuples); 
+      /*      pageid_t mergedPages = compressData
 	<PAGELAYOUT,mergeIterator<ITERA,ITERB,typename PAGELAYOUT::FMT::TUP> >
-	(xid, &mBegin, &mEnd,tree->r_,a->pageAlloc,a->pageAllocState,&insertedTuples);
+	(xid, &mBegin, &mEnd,tree->r_,a->pageAlloc,a->pageAllocState,&insertedTuples);  */
 
       delete taEnd;
       delete tbEnd;
@@ -304,15 +329,6 @@ namespace rose {
       typename PAGELAYOUT::FMT::TUP> > * args2;
   };
 
-  // How many bytes of tuples can we afford to keep in RAM?
-  // this is just a guessed value... it seems about right based on
-  // experiments, but 450 bytes overhead per tuple is insane!
-  static const int RB_TREE_OVERHEAD = 450;
-  static const pageid_t MEM_SIZE = 800 * 1000 * 1000;
-  // How many pages should we try to fill with the first C1 merge?
-  static const pageid_t START_SIZE = /*10 **/ 1000;
-  static const int R = 10; // XXX set this as low as possible (for dynamic setting.  = sqrt(C2 size / C0 size))
-
   template<class PAGELAYOUT>
     lsmTableHandle <PAGELAYOUT> * TlsmTableStart(recordid& tree) {
     /// XXX xid for daemon processes?

src/stasis/page/compression/staticTuple.h

@@ -23,6 +23,9 @@ namespace rose {
   typedef TYPE8 TYP8;
   typedef TYPE9 TYP9;
 
+  inline bool tombstone() const {
+    return s.flag_ == TOMBSTONE;
+  }
 
   explicit inline StaticTuple() {
     s.flag_ = NORMAL; s.epoch_ = 0 ;
@@ -158,6 +161,21 @@ namespace rose {
     return 0;
   }
 
+  inline void copyFrom(const StaticTuple& t) {
+    s.flag_=t.s.flag_;
+    s.epoch_=t.s.epoch_;
+    if(0 < N) { s.cols0_ = t.s.cols0_; }
+    if(1 < N) { s.cols1_ = t.s.cols1_; }
+    if(2 < N) { s.cols2_ = t.s.cols2_; }
+    if(3 < N) { s.cols3_ = t.s.cols3_; }
+    if(4 < N) { s.cols4_ = t.s.cols4_; }
+    if(5 < N) { s.cols5_ = t.s.cols5_; }
+    if(6 < N) { s.cols6_ = t.s.cols6_; }
+    if(7 < N) { s.cols7_ = t.s.cols7_; }
+    if(8 < N) { s.cols8_ = t.s.cols8_; }
+    if(9 < N) { s.cols9_ = t.s.cols9_; }
+  }
+
   static void printSt(void const * const sp) {
     st const * const s = (st const * const)sp;
     printf("(");