clean up handling of uninitialized pages (removes some special case code)

src/stasis/CMakeLists.txt

@@ -18,6 +18,7 @@ ADD_LIBRARY(stasis crc32.c redblack.c lhtable.c concurrentHash.c rw.c doubleLink
                    logger/logMemory.c
                    logger/reorderingHandle.c
                    logger/groupForce.c
+                   page/uninitialized.c
                    page/slotted.c
                    page/fixed.c 
                    page/raw.c

src/stasis/Makefile.am

@@ -13,6 +13,7 @@ libstasis_la_SOURCES=crc32.c redblack.c lhtable.c rw.c doubleLinkedList.c common
 		   logger/logMemory.c \
 		   logger/reorderingHandle.c \
 		   logger/groupForce.c \
+		   page/uninitialized.c \
 		   page/raw.c page/slotted.c page/lsnFree.c page/fixed.c page/segment.c compensations.c \
 	           operations/pageOperations.c operations/decrement.c  \
 		   operations/increment.c  operations/prepare.c  operations/set.c \

src/stasis/bufferManager/bufferHash.c

@@ -302,10 +302,8 @@ static Page * bhLoadPageImpl_helper(stasis_buffer_manager_t* bm, int xid, const
     pthread_mutex_lock(&bh->mut);
 
   } else {
-    memset(ret->memAddr,0,PAGE_SIZE);
-    *stasis_page_lsn_ptr(ret) = ret->LSN;
+    type = UNINITIALIZED_PAGE;
     assert(!ret->dirty);
-//    ret->dirty = 0;
     stasis_page_loaded(ret, type);
   }
   *pagePendingPtr(ret) = 0;

src/stasis/bufferManager/concurrentBufferManager.c

@@ -182,6 +182,7 @@ static Page * chLoadPageImpl_helper(stasis_buffer_manager_t* bm, int xid, const
     p->id = pageid;
     if(uninitialized) {
       type = UNINITIALIZED_PAGE;
+      stasis_page_loaded(p, UNINITIALIZED_PAGE);
     } else {
       ch->page_handle->read(ch->page_handle, p, type);
     }
@@ -193,7 +194,7 @@ static Page * chLoadPageImpl_helper(stasis_buffer_manager_t* bm, int xid, const
   if(first) { ch->lru->hit(ch->lru, p); }
   readlock(p->loadlatch, 0);
   hashtable_unlock(&h);
-
+  assert(p->id == pageid);
   return p;
 }
 static Page * chLoadPageImpl(stasis_buffer_manager_t *bm, int xid, const pageid_t pageid, pagetype_t type) {

src/stasis/page.c

@@ -76,6 +76,7 @@ terms specified in this license.
 #include <stasis/compensations.h>
 #include <stasis/page/slotted.h>
 #include <stasis/page/fixed.h>
+#include <stasis/page/uninitialized.h>
 #include <stasis/operations/arrayList.h>
 #include <stasis/bufferPool.h>
 #include <stasis/truncation.h>
@@ -113,6 +114,7 @@ void stasis_page_init(stasis_dirty_page_table_t * dpt) {
   stasis_page_slotted_init();
   fixedPageInit();
 
+  stasis_page_impl_register(stasis_page_uninitialized_impl());
   stasis_page_impl_register(stasis_page_slotted_impl());
   stasis_page_impl_register(fixedImpl());
   stasis_page_impl_register(stasis_page_boundary_tag_impl());
@@ -288,34 +290,24 @@ void stasis_record_compact_slotids(int xid, Page * p) {
 */
 void stasis_page_loaded(Page * p, pagetype_t type){
   p->pageType = (type == UNKNOWN_TYPE_PAGE) ? *stasis_page_type_ptr(p) : type;
-  if(p->pageType) {
-    assert(page_impls[p->pageType].page_type == p->pageType);
-    if (page_impls[p->pageType].pageLoaded) page_impls[p->pageType].pageLoaded(p);
-  } else {
-    p->LSN = *stasis_page_lsn_ptr(p);  // XXX kludge - shouldn't special-case UNINITIALIZED_PAGE
-  }
+  assert(page_impls[p->pageType].page_type == p->pageType);
+  if (page_impls[p->pageType].pageLoaded) page_impls[p->pageType].pageLoaded(p);
 }
 void stasis_page_flushed(Page * p){
 
   pagetype_t type = p->pageType;
-  if(type) {
-    assert(page_impls[type].page_type == type);
-    if(page_impls[type].has_header) {
-      *stasis_page_type_ptr(p)= type;
-      *stasis_page_lsn_ptr(p) = p->LSN;
-    }
-    if(page_impls[type].pageFlushed) page_impls[type].pageFlushed(p);
-  } else {
+
+  assert(page_impls[type].page_type == type);
+  if(page_impls[type].has_header) {
     *stasis_page_type_ptr(p)= type;
     *stasis_page_lsn_ptr(p) = p->LSN;
   }
+  if(page_impls[type].pageFlushed) page_impls[type].pageFlushed(p);
 }
 void stasis_page_cleanup(Page * p) {
   short type = p->pageType;
-  if(type) {
-    assert(page_impls[type].page_type == type);
-    if(page_impls[type].pageCleanup) page_impls[type].pageCleanup(p);
-  }
+  assert(page_impls[type].page_type == type);
+  if(page_impls[type].pageCleanup) page_impls[type].pageCleanup(p);
 }
 
 /// Generic block implementations

src/stasis/page/uninitialized.c

@@ -0,0 +1,54 @@
+/*
+ * uninitialized.c
+ *
+ *  Created on: Nov 9, 2009
+ *      Author: sears
+ */
+#include <stasis/page/uninitialized.h>
+
+static int notSupported(int xid, Page * p) { return 0; }
+
+static void uninitializedLoaded(Page *p) {
+  p->LSN = *stasis_page_lsn_cptr(p);
+}
+static void uninitializedFlushed(Page *p) {
+  *stasis_page_type_ptr(p)= p->pageType;
+  *stasis_page_lsn_ptr(p) = p->LSN;
+
+}
+static void uninitializedCleanup(Page *p) {
+
+}
+
+page_impl stasis_page_uninitialized_impl() {
+  static page_impl pi = {
+    UNINITIALIZED_PAGE,
+    0,
+    0, //read,
+    0, //write,
+    0,// readDone
+    0,// writeDone
+    0, //GetType,
+    0, //SetType,
+    0, //fixedGetLength,
+    0, //fixedFirst,
+    0, //fixedNext,
+    0, //fixedLast,
+    notSupported, // notSupported,
+    0,//stasis_block_first_default_impl,
+    0,//stasis_block_next_default_impl,
+    0,//stasis_block_done_default_impl,
+    0,//fixedFreespace,
+    0,//fixedCompact,
+    0,//fixedCompactSlotIds,
+    0,//fixedPreAlloc,
+    0,//fixedPostAlloc,
+    0,//fixedSplice,
+    0,//fixedFree,
+    0, // XXX dereference
+    uninitializedLoaded, // loaded
+    uninitializedFlushed, // flushed
+    uninitializedCleanup
+  };
+  return pi;
+}

stasis/page/uninitialized.h

@@ -0,0 +1,14 @@
+/*
+ * uninitialized.h
+ *
+ *  Created on: Nov 9, 2009
+ *      Author: sears
+ */
+
+#ifndef UNINITIALIZED_H_
+#define UNINITIALIZED_H_
+#include "../page.h"
+
+page_impl stasis_page_uninitialized_impl();
+
+#endif /* UNINITIALIZED_H_ */