From b44f8b17b32430b04a8ca39895b71219f730e7fb Mon Sep 17 00:00:00 2001
From: Sears Russell <sears@cs.berkeley.edu>
Date: Tue, 14 Jul 2009 07:46:47 +0000
Subject: [PATCH] Add getCachedPage() call.  This allows dirtyPageTable to
 write back pages with out accidentally reading them back in from disk with
 loadPage().  This should improve performance and allow loadPageOfType() to be
 used safely.

---
 src/stasis/bufferManager.c                    |  7 +++---
 src/stasis/bufferManager/bufferHash.c         | 22 +++++++++++++++++++
 .../legacy/legacyBufferManager.c              |  1 +
 src/stasis/bufferManager/pageArray.c          |  1 +
 src/stasis/dirtyPageTable.c                   | 16 +++++++++-----
 stasis/bufferManager.h                        |  7 ++++++
 6 files changed, 45 insertions(+), 9 deletions(-)

diff --git a/src/stasis/bufferManager.c b/src/stasis/bufferManager.c
index de9b61c..dd328f6 100644
--- a/src/stasis/bufferManager.c
+++ b/src/stasis/bufferManager.c
@@ -164,6 +164,7 @@ compensated_function void  __profile_releasePage(Page * p) {
 
 Page * (*loadPageImpl)(int xid, pageid_t pageid, pagetype_t type) = 0;
 Page * (*loadUninitPageImpl)(int xid, pageid_t pageid) = 0;
+Page * (*getCachedPageImpl)(int xid, pageid_t pageid) = 0;
 void   (*releasePageImpl)(Page * p) = 0;
 void (*writeBackPage)(Page * p) = 0;
 void (*forcePages)() = 0;
@@ -175,7 +176,6 @@ Page * loadPage(int xid, pageid_t pageid) {
   // This lock is released at Tcommit()
   if(globalLockManager.readLockPage) { globalLockManager.readLockPage(xid, pageid); }
   return loadPageImpl(xid, pageid, UNKNOWN_TYPE_PAGE);
-
 }
 Page * loadPageOfType(int xid, pageid_t pageid, pagetype_t type) {
   if(globalLockManager.readLockPage) { globalLockManager.readLockPage(xid, pageid); }
@@ -186,9 +186,10 @@ Page * loadUninitializedPage(int xid, pageid_t pageid) {
   if(globalLockManager.readLockPage) { globalLockManager.readLockPage(xid, pageid); }
 
   return loadUninitPageImpl(xid, pageid);
-
 }
-
+Page * getCachedPage(int xid, pageid_t pageid) {
+  return getCachedPageImpl(xid, pageid);
+}
 void releasePage(Page * p) {
   releasePageImpl(p);
 }
diff --git a/src/stasis/bufferManager/bufferHash.c b/src/stasis/bufferManager/bufferHash.c
index a8980eb..a6450b4 100644
--- a/src/stasis/bufferManager/bufferHash.c
+++ b/src/stasis/bufferManager/bufferHash.c
@@ -190,6 +190,27 @@ static void * writeBackWorker(void * ignored) {
   return 0;
 }
 
+static Page * bhGetCachedPage(int xid, const pageid_t pageid) {
+  pthread_mutex_lock(&mut);
+  // Is the page in cache?
+  Page * ret = LH_ENTRY(find)(cachedPages, &pageid, sizeof(pageid));
+  if(ret) {
+    checkPageState(ret);
+    if(!*pagePendingPtr(ret)) {
+      // good
+      if(!*pagePinCountPtr(ret) ) {
+        // Then ret is in lru (otherwise it would be pending, or not cached); remove it.
+        lru->remove(lru, ret);
+      }
+      (*pagePinCountPtr(ret))++;
+    } else {
+      ret = 0;
+    }
+  }
+  pthread_mutex_unlock(&mut);
+  return ret;
+}
+
 static Page * bhLoadPageImpl_helper(int xid, const pageid_t pageid, int uninitialized, pagetype_t type) {
 
   // Note:  Calls to loadlatch in this function violate lock order, but
@@ -397,6 +418,7 @@ void stasis_buffer_manager_hash_open(stasis_page_handle_t * h) {
 
   loadPageImpl = bhLoadPageImpl;
   loadUninitPageImpl = bhLoadUninitPageImpl;
+  getCachedPageImpl = bhGetCachedPage;
   releasePageImpl = bhReleasePage;
   writeBackPage = bhWriteBackPage;
   forcePages = bhForcePages;
diff --git a/src/stasis/bufferManager/legacy/legacyBufferManager.c b/src/stasis/bufferManager/legacy/legacyBufferManager.c
index bc003fc..b73cbc4 100644
--- a/src/stasis/bufferManager/legacy/legacyBufferManager.c
+++ b/src/stasis/bufferManager/legacy/legacyBufferManager.c
@@ -45,6 +45,7 @@ int stasis_buffer_manager_deprecated_open(stasis_page_handle_t * ph) {
     releasePageImpl = bufManReleasePage;
     loadPageImpl = bufManLoadPage;
     loadUninitPageImpl = bufManLoadUninitPage;
+    getCachedPageImpl = bufManLoadPage; // Since this code is deprecated, loadPage is "good enough" though it breaks segments.
     writeBackPage = pageWrite_legacyWrapper;
     forcePages = forcePageFile_legacyWrapper;
     forcePageRange = forceRangePageFile_legacyWrapper;
diff --git a/src/stasis/bufferManager/pageArray.c b/src/stasis/bufferManager/pageArray.c
index 8d7c245..2168dc6 100644
--- a/src/stasis/bufferManager/pageArray.c
+++ b/src/stasis/bufferManager/pageArray.c
@@ -62,6 +62,7 @@ void stasis_buffer_manager_mem_array_open () {
 
   releasePageImpl = paReleasePage;
   loadPageImpl = paLoadPage;
+  getCachedPageImpl = paLoadPage;
   writeBackPage = paWriteBackPage;
   forcePages = paForcePages;
   stasis_buffer_manager_close = paBufDeinit;
diff --git a/src/stasis/dirtyPageTable.c b/src/stasis/dirtyPageTable.c
index ee333da..6892ad5 100644
--- a/src/stasis/dirtyPageTable.c
+++ b/src/stasis/dirtyPageTable.c
@@ -91,9 +91,11 @@ void stasis_dirty_page_table_flush(stasis_dirty_page_table_t * dirtyPages) {
   pthread_mutex_unlock(&dirtyPages->mutex);
 
   for(i = 0; i < MAX_BUFFER_SIZE && staleDirtyPages[i] != -1; i++) {
-    p = loadPage(-1, staleDirtyPages[i]);
-    writeBackPage(p);
-    releasePage(p);
+    p = getCachedPage(-1, staleDirtyPages[i]);
+    if(p) {
+      writeBackPage(p);
+      releasePage(p);
+    }
   }
   free(staleDirtyPages);
 }
@@ -117,9 +119,11 @@ void stasis_dirty_page_table_flush_range(stasis_dirty_page_table_t * dirtyPages,
   pthread_mutex_unlock(&dirtyPages->mutex);
 
   for(i = 0; i < MAX_BUFFER_SIZE && staleDirtyPages[i] != -1; i++) {
-    p = loadPage(-1, staleDirtyPages[i]);
-    writeBackPage(p);
-    releasePage(p);
+    p = getCachedPage(-1, staleDirtyPages[i]);
+    if(p) {
+      writeBackPage(p);
+      releasePage(p);
+    }
   }
   free(staleDirtyPages);
   forcePageRange(start*PAGE_SIZE,stop*PAGE_SIZE);
diff --git a/stasis/bufferManager.h b/stasis/bufferManager.h
index 54abb38..4c93097 100644
--- a/stasis/bufferManager.h
+++ b/stasis/bufferManager.h
@@ -97,6 +97,7 @@ Page * loadPageOfType(int xid, pageid_t pageid, pagetype_t type);
 
 Page * loadUninitializedPage(int xid, pageid_t pageid);
 
+Page * getCachedPage(int xid, const pageid_t pageid);
 
 /**
     This is the function pointer that stasis_buffer_manager_open sets in order to
@@ -104,6 +105,12 @@ Page * loadUninitializedPage(int xid, pageid_t pageid);
 */
 extern Page * (*loadPageImpl)(int xid, pageid_t pageid, pagetype_t type);
 extern Page * (*loadUninitPageImpl)(int xid, pageid_t pageid);
+/**
+    Get a page from cache.  This function should never block on I/O.
+
+    @return a pointer to the page, or NULL if the page is not in cache, or is being read from disk.
+ */
+extern Page * (*getCachedPageImpl)(int xid, pageid_t pageid);
 /**
    loadPage aquires a lock when it is called, effectively pinning it
    in memory.  releasePage releases this lock.