force_range is now being called after compressData()

benchmarks/roseTable.cpp

@@ -77,8 +77,8 @@ namespace rose {
     for(long int i = 0; i < INSERTS; i++) {
       getTuple<PAGELAYOUT>(i,t);
       TlsmTableInsert(h,t);
-      getTuple<PAGELAYOUT>(i,t);
+      //      getTuple<PAGELAYOUT>(i,t);
-      assert(TlsmTableFind(xid,h,t,s));
+      //      assert(TlsmTableFind(xid,h,t,s));
       count --;
       if(!count) {
 	count = COUNT;

src/stasis/io/pfile.c

@@ -385,8 +385,9 @@ static int pfile_force(stasis_handle_t *h) {
 static int pfile_force_range(stasis_handle_t *h, lsn_t start, lsn_t stop) {
   pfile_impl * impl = h->impl;
 #ifdef HAVE_SYNC_FILE_RANGE
-  printf("pfile_force_range calling sync_file_range\n");
+  printf("pfile_force_range calling sync_file_range %lld %lld\n",
-  int ret = sync_file_range(impl->fd, start-impl->start_pos, (stop-start),
+	 start-impl->start_pos, stop-start); fflush(stdout);
+  int ret = sync_file_range(impl->fd, start-impl->start_pos, stop-start,
 			      SYNC_FILE_RANGE_WAIT_BEFORE |
 			      SYNC_FILE_RANGE_WRITE |
 			      SYNC_FILE_RANGE_WAIT_AFTER);

src/stasis/operations/lsmTable.h

@@ -201,6 +201,9 @@ namespace rose {
 	<PAGELAYOUT,mergeIterator<ITERA,ITERB,typename PAGELAYOUT::FMT::TUP> >
 	(xid, &mBegin, &mEnd,tree->r_,a->pageAlloc,a->pageAllocState,&insertedTuples);  */
 
+      // XXX hardcodes tree type.
+      TlsmForce(xid,tree->r_,TlsmRegionForceRid,a->pageAllocState);
+
       delete taEnd;
       delete tbEnd;
 
@@ -489,6 +492,7 @@ namespace rose {
 
     return ret;
   }
+  // XXX this does not force the table to disk... it simply forces everything out of the in-memory tree.
   template<class PAGELAYOUT>
     void TlsmTableFlush(lsmTableHandle<PAGELAYOUT> *h) {
       pthread_mutex_lock(h->mut);

src/stasis/operations/lsmTree.c

@@ -38,6 +38,19 @@ pageid_t TlsmRegionAlloc(int xid, void *conf) {
   return ret;
 }
 
+void TlsmRegionForceRid(int xid, void *conf) {
+  recordid rid = *(recordid*)conf;
+  TlsmRegionAllocConf_t a;
+  Tread(xid,rid,&a);
+  //  TlsmRegionAllocConf_t* a = (TlsmRegionAllocConf_t*)conf;
+  for(int i = 0; i < a.regionCount; i++) {
+    a.regionList.slot = i;
+    pageid_t pid;
+    Tread(xid,a.regionList,&pid);
+    dirtyPages_flushRange(pid, pid+a.regionSize);
+    //    TregionDealloc(xid,pid);
+  }
+}
 void TlsmRegionDeallocRid(int xid, void *conf) {
   recordid rid = *(recordid*)conf;
   TlsmRegionAllocConf_t a;
@@ -602,6 +615,11 @@ recordid TlsmAppendPage(int xid, recordid tree,
 
   return ret;
 }
+void TlsmForce(int xid, recordid tree, lsm_page_forcer_t force,
+	       void *allocator_state) {
+  force(xid, allocator_state);
+}
+
 void TlsmFree(int xid, recordid tree, lsm_page_deallocator_t dealloc,
 	      void *allocator_state) {
   //  Tdealloc(xid,tree);

src/stasis/truncation.c

@@ -129,7 +129,7 @@ void dirtyPages_flushRange(pageid_t start, pageid_t stop) {
     releasePage(p);
   }
   free(staleDirtyPages);
-  forcePageRange(start,stop); // XXX
+  forcePageRange(start*PAGE_SIZE,stop*PAGE_SIZE);
 }
 void dirtyPagesInit() { 
   dirtyPages = pblHtCreate();

stasis/operations/lsmTree.h

@@ -26,12 +26,14 @@ typedef int(*lsm_comparator_t)(const void* a, const void* b);
 typedef void*(*lsm_page_initializer_t)(Page *, void *);
 typedef pageid_t(*lsm_page_allocator_t)(int, void *);
 typedef void(*lsm_page_deallocator_t)(int, void *);
+typedef void(*lsm_page_forcer_t)(int, void *);
 
 void lsmTreeRegisterComparator(int id, lsm_comparator_t i);
 void lsmTreeRegisterPageInitializer(int id, lsm_page_initializer_t i);
 
 pageid_t TlsmRegionAlloc(int xid, void *conf);
 pageid_t TlsmRegionAllocRid(int xid, void *conf);
+void TlsmRegionForceRid(int xid, void *conf);
 typedef struct {
   recordid regionList;
   pageid_t regionCount;
@@ -72,6 +74,8 @@ recordid TlsmAppendPage(int xid, recordid tree,
 			const byte *key,
 			lsm_page_allocator_t allocator, void *allocator_state,
 			long pageid);
+void TlsmForce(int xid, recordid tree, lsm_page_forcer_t force,
+	       void *allocator_state);
 void TlsmFree(int xid, recordid tree, lsm_page_deallocator_t dealloc,
 	 void *allocator_state);
 /**

stasis/truncation.h

@@ -68,9 +68,10 @@ void dirtyPages_remove(Page * p);
 int  dirtyPages_isDirty(Page * p);
 
 /**
-   @todo forceWriteRange has nothing to do with the dirty pages api, or truncation.
+   @todo flushRange's API sucks.  It should be two functions, "startRangeFlush" and "waitRangeFlushes" or something.
+   @todo flushRange has nothing to do with the dirty pages api, or truncation.
  */
-void dirtyPages_forceWriteRange(pageid_t start, pageid_t stop);
+void dirtyPages_flushRange(pageid_t start, pageid_t stop);
 
 void truncationInit();
 void truncationDeinit();