Added TdurabilityLevel().

Test cases now pass with LOG_TO_MEMORY and BUFFER_MANAGER_MEM_ARRAY

lladd/bufferManager.h

@@ -115,7 +115,13 @@ extern int bufferManagerType;
  * @param pageid ID of the page you want to load
  * @return fully formed Page type
  */
-extern Page * (*loadPage)(int xid, int pageid);
+Page * loadPage(int xid, int pageid);
+
+/** 
+    This is the function pointer that bufInit sets in order to
+    override loadPage.
+*/
+extern Page * (*loadPageImpl)(int xid, int pageid);
 /**
    loadPage aquires a lock when it is called, effectively pinning it
    in memory.  releasePage releases this lock.

lladd/constants.h

@@ -98,6 +98,10 @@ terms specified in this license.
 #define LOG_TO_FILE   0
 #define LOG_TO_MEMORY 1
 
+#define VOLATILE   0
+#define PERSISTENT 1
+#define DURABLE    2
+
 #define MAX_TRANSACTIONS 1000
 
 /** Operation types */

lladd/transactional.h

@@ -435,6 +435,17 @@ int TisActiveTransaction(int xid);
 */
 lsn_t transactions_minRecLSN();
 
+
+/**
+   Report Stasis' current durablity guarantees.
+
+   @return VOLATILE if the data will be lost after Tdeinit(), or a
+   crash, PERSISTENT if the data will be written back to disk after
+   Tdeinit(), but may be corrupted at crash, or DURABLE if Stasis will
+   apply committed transactions, and roll back active transactions
+   after a crash.
+*/
+int TdurabilityLevel();
 END_C_DECLS
 
 #endif

src/lladd/bufferManager.c

@@ -122,7 +122,7 @@ static void bufManSimulateBufferManagerCrash();
 static int bufManBufInit() {
 
     releasePage = bufManReleasePage;
-    loadPage = bufManLoadPage;
+    loadPageImpl = bufManLoadPage;
     writeBackPage = pageWrite;
     forcePages = forcePageFile;
     bufDeinit = bufManBufDeinit; 
@@ -469,11 +469,6 @@ compensated_function void  __profile_releasePage(Page * p) {
 
 static compensated_function Page *bufManLoadPage(int xid, int pageid) {
 
-  try_ret(NULL) {
-    if(globalLockManager.readLockPage) { globalLockManager.readLockPage(xid, pageid); }
-  } end_ret(NULL);
-
-
   Page * ret = pthread_getspecific(lastPage);
 
   if(ret && ret->id == pageid) { 
@@ -503,13 +498,23 @@ static compensated_function Page *bufManLoadPage(int xid, int pageid) {
   return ret;
 }
 
-Page * (*loadPage)(int xid, int pageid) = 0;
+Page * (*loadPageImpl)(int xid, int pageid) = 0;
 void   (*releasePage)(Page * p) = 0;
 void (*writeBackPage)(Page * p) = 0;
 void (*forcePages)() = 0;
 void   (*bufDeinit)()  = 0;
 void   (*simulateBufferManagerCrash)()  = 0;
 
+Page * loadPage(int xid, int pageid) { 
+  try_ret(NULL) {
+    // This lock is released at Tcommit()
+    if(globalLockManager.readLockPage) { globalLockManager.readLockPage(xid, pageid); }
+  } end_ret(NULL);
+
+  return loadPageImpl(xid, pageid);
+
+}
+
 int bufInit(int type) { 
   bufferManagerType = type;
   static int lastType = 0;

src/lladd/bufferManager/pageArray.c

@@ -12,6 +12,7 @@ static int pageCount;
 static pthread_mutex_t pageArray_mut = PTHREAD_MUTEX_INITIALIZER;
 
 static Page * paLoadPage(int xid, int pageid) {
+
   pthread_mutex_lock(&pageArray_mut);
   if(pageid >= pageCount) { 
     pageMap = realloc(pageMap, (1+pageid) * sizeof(Page*));
@@ -58,7 +59,7 @@ static void paBufDeinit() {
 void paBufInit () { 
 
   releasePage = paReleasePage;
-  loadPage = paLoadPage;
+  loadPageImpl = paLoadPage;
   writeBackPage = paWriteBackPage;
   forcePages = paForcePages;
   bufDeinit = paBufDeinit; 

src/lladd/transactional2.c

@@ -378,3 +378,12 @@ int TisActiveTransaction(int xid) {
   return ret;
 }
 
+int TdurabilityLevel() {
+  if(bufferManagerType == BUFFER_MANAGER_MEM_ARRAY) { 
+    return VOLATILE;
+  } else if(loggerType == LOG_TO_MEMORY) { 
+    return PERSISTENT;
+  } else { 
+    return DURABLE;
+  }
+}

test/lladd/check_header.c

@@ -69,6 +69,8 @@ START_TEST (header_test) {
 
   Tdeinit();
 
+  if(TdurabilityLevel() == VOLATILE) { return ; }
+
   Tinit();
 
   xid = Tbegin();

test/lladd/check_linearHash.c

@@ -238,6 +238,7 @@ START_TEST(transactionalLinearHashTest)
 	
   Tabort(xid);
   Tdeinit();
+  if(TdurabilityLevel() == VOLATILE) return;  
   Tinit();
   xid = Tbegin();
   ThashOpen(xid, hashRoot, sizeof(int), sizeof(recordid));

test/lladd/check_operations.c

@@ -424,9 +424,10 @@ START_TEST(operation_instant_set) {
   Tread(xid, rid, &three);
   assert(two == three);
   Tcommit(xid);
- 
   Tdeinit();
  
+  if(TdurabilityLevel() == VOLATILE) { return; } 
+
   Tinit();
 
   xid = Tbegin();

test/lladd/check_page.c

@@ -202,8 +202,6 @@ static void* worker_thread(void * arg_ptr) {
 START_TEST(pageNoThreadTest)
 {
   Page * p;
-  /*    p->id = 0;*/
-
 
   pthread_mutex_init(&lsn_mutex, NULL);
   
@@ -214,12 +212,15 @@ START_TEST(pageNoThreadTest)
   slottedPageInitialize(p);
   worker_thread(p);
 
-  unlock(p->loadlatch);
   p->LSN = 0;
   *lsn_ptr(p) = p->LSN;
 
+  releasePage(p);
+
   Tdeinit();
 
+  pthread_mutex_destroy(&lsn_mutex);
+
 }
 END_TEST
 
@@ -316,6 +317,7 @@ START_TEST(pageNoThreadMultPageTest)
 
 
   Tdeinit();
+  pthread_mutex_destroy(&lsn_mutex);
 
 }
 END_TEST
@@ -357,6 +359,9 @@ START_TEST(pageThreadTest) {
 
   Tdeinit();
 
+  pthread_mutex_destroy(&lsn_mutex);
+  pthread_mutex_destroy(&random_mutex);
+
 } END_TEST
 
 
@@ -384,6 +389,8 @@ START_TEST(fixedPageThreadTest) {
   *lsn_ptr(p) = p->LSN;
   releasePage(p);
   Tdeinit();
+  pthread_mutex_destroy(&lsn_mutex);
+  pthread_mutex_destroy(&random_mutex);
 } END_TEST
 
 START_TEST(pageCheckSlotTypeTest) {

test/lladd/check_regions.c

@@ -329,6 +329,8 @@ START_TEST(regions_recoveryTest) {
 
   Tdeinit();
 
+  if(TdurabilityLevel() == VOLATILE) { return; }
+
   Tinit();
   
   int xid = Tbegin();