Added assertlock() and call it when page implementations assume a page has already been latched. (Fixed

associated bugs as well)

libdfa/rw.h

@@ -35,6 +35,8 @@ void readlock (rwl *lock, int d);
 int tryreadlock(rwl *lock, int d);
 void writelock (rwl *lock, int d);
 int trywritelock(rwl *lock, int d);
+/** aborts if called when no thread holds this latch. */
+void assertlocked(rwl *lock);
 void downgradelock(rwl * lock);
 void unlock(rwl * lock);
 /** @deprecated in favor of unlock() */

src/libdfa/rw.c

@@ -36,6 +36,10 @@ rwl *initlock (void)
   writelock(lock, d);
   }*/
 
+void assertlocked(rwl * lock) { 
+  assert(lock->writers || lock->readers);
+}
+
 void readlock (rwl *lock, int d)
 {
   /*  printf("reader %d\n", d);  

src/lladd/operations/alloc.c

@@ -202,7 +202,9 @@ compensated_function recordid Talloc(int xid, unsigned long size) {
     p = loadPage(xid, lastFreepage);
 
     while(slottedFreespace(p) < physical_slot_length(type)) { 
+      writelock(p->rwlatch,0);
       slottedCompact(p);
+      unlock(p->rwlatch);
       int newFreespace = slottedFreespace(p);
       if(newFreespace >= physical_slot_length(type)) { 
 	break;

src/lladd/operations/arrayList.c

@@ -75,6 +75,9 @@ static int operateAlloc(int xid, Page * p, lsn_t lsn, recordid rid, const void *
   int multiplier = tlp->multiplier;
   int size = tlp->size;
 
+  /* Allocing this page -> implicit lock, but latch to conformt to
+   fixedPage's interface. */
+  writelock(p->rwlatch, 0);
   fixedPageInitialize(p, sizeof(int), recordsPerPage(sizeof(int)));
 
 #define MAX_OFFSET_POSITION    3
@@ -91,7 +94,6 @@ static int operateAlloc(int xid, Page * p, lsn_t lsn, recordid rid, const void *
   firstDataPageRid.slot = 4;
   
   int firstDataPage = firstPage + 1;
-  /* Allocing this page -> implicit lock. */
   fixedWriteUnlocked(p, countRid,      (byte*)&count);
   fixedWriteUnlocked(p, multiplierRid, (byte*)&multiplier);
   fixedWriteUnlocked(p, firstDataPageRid, (byte*)&firstDataPage);
@@ -109,6 +111,7 @@ static int operateAlloc(int xid, Page * p, lsn_t lsn, recordid rid, const void *
   ret.page = firstPage;
   ret.slot = 0;        /* slot = # of slots in array... */
   ret.size = size;
+  unlock(p->rwlatch);
 
   return 0;
 }
@@ -196,7 +199,7 @@ compensated_function int TarrayListLength(int xid, recordid rid) {
 /*----------------------------------------------------------------------------*/
 
 recordid dereferenceArrayListRid(Page * p, int offset) {
-
+  readlock(p->rwlatch,0);
   TarrayListParameters tlp = pageToTLP(p);
 
   int rec_per_page = recordsPerPage((size_t)tlp.size);
@@ -215,6 +218,7 @@ recordid dereferenceArrayListRid(Page * p, int offset) {
 
   assert(pageRidSlot + FIRST_DATA_PAGE_OFFSET < fixedPageCount(p));
   thePage = *(int*)fixed_record_ptr(p, pageRidSlot + FIRST_DATA_PAGE_OFFSET); /*reading immutable record; don't need latch.*/
+  unlock(p->rwlatch);
 
   recordid rid;
   rid.page = thePage + blockPage;

src/lladd/operations/decrement.c

@@ -54,7 +54,6 @@ static int operate(int xid, Page * p, lsn_t lsn, recordid r, const void *d) {
 	recordRead(xid, p, r, (byte*)&i);
 	i--;
 	recordWrite(xid, p, lsn, r, (byte*)&i);
-
 	return 0;
 }
 

src/lladd/operations/increment.c

@@ -54,7 +54,6 @@ static int operate(int xid, Page * p, lsn_t lsn, recordid r, const void *d) {
 	recordRead(xid, p, r, (byte*)&i);
 	i++;
 	recordWrite(xid, p, lsn, r, (byte*)&i);
-
 	return 0;
 }
 

src/lladd/page.c

@@ -154,7 +154,7 @@ void recordWrite(int xid, Page * p, lsn_t lsn, recordid rid, const void *dat) {
   assert( (p->id == rid.page) && (p->memAddr != NULL) );	
   
 }
-
+static int recordReadWarnedAboutPageTypeKludge = 0;
 int recordRead(int xid, Page * p, recordid rid, void *buf) {
   assert(rid.page == p->id); 
   
@@ -164,8 +164,12 @@ int recordRead(int xid, Page * p, recordid rid, void *buf) {
     readBlob(xid, p, rid, buf);
   } else if(page_type == SLOTTED_PAGE || page_type == BOUNDARY_TAG_PAGE) {
     slottedRead(p, rid, buf);
-    /* @TODO !page_type can never be required if this code is correct... arraylist is broken!!*/
+    /* @TODO !page_type can never be required if this code is correct... arraylist is broken!! XXX */
   } else if(page_type == FIXED_PAGE || page_type==ARRAY_LIST_PAGE || !page_type) {
+    if(!page_type && ! recordReadWarnedAboutPageTypeKludge) {
+      recordReadWarnedAboutPageTypeKludge = 1;
+      printf("page.c: MAKING USE OF TERRIBLE KLUDGE AND IGNORING ASSERT FAILURE! FIX ARRAY LIST ASAP!!!\n");
+    }
     fixedRead(p, rid, buf);
   } else {
     abort();

src/lladd/page/fixed.c

@@ -18,11 +18,13 @@ void fixedPageInitialize(Page * page, size_t size, int count) {
 }
 
 short fixedPageCount(Page * page) {
+  assertlocked(page->rwlatch);
   assert(*page_type_ptr(page) == FIXED_PAGE || *page_type_ptr(page) == ARRAY_LIST_PAGE);
   return *recordcount_ptr(page);
 }
 
 short fixedPageRecordSize(Page * page) {
+  assertlocked(page->rwlatch);
   assert(*page_type_ptr(page) == FIXED_PAGE || *page_type_ptr(page) == ARRAY_LIST_PAGE);
   return *recordsize_ptr(page);
 }
@@ -53,7 +55,9 @@ recordid fixedRawRalloc(Page *page) {
   return fixedRawRallocMany(page, 1);
 }
 
+static int checkRidWarnedAboutUninitializedKludge = 0;
 static void checkRid(Page * page, recordid rid) {
+  assertlocked(page->rwlatch);
   if(*page_type_ptr(page)) {
     assert(*page_type_ptr(page) == FIXED_PAGE || *page_type_ptr(page) == ARRAY_LIST_PAGE);
     assert(page->id == rid.page);
@@ -62,11 +66,16 @@ static void checkRid(Page * page, recordid rid) {
     int recCount = *recordcount_ptr(page);
     assert(recCount  > rid.slot);          
   } else {
+     if(!checkRidWarnedAboutUninitializedKludge) {
+       checkRidWarnedAboutUninitializedKludge = 1;
+       printf("KLUDGE detected in checkRid. Fix it ASAP\n");
+     }
     fixedPageInitialize(page, rid.size, recordsPerPage(rid.size));
   }
 }
 
 void fixedReadUnlocked(Page * page, recordid rid, byte * buf) {
+  assertlocked(page->rwlatch);
   if(!memcpy(buf, fixed_record_ptr(page, rid.slot), rid.size)) {
     perror("memcpy");
     abort();
@@ -87,6 +96,7 @@ void fixedRead(Page * page, recordid rid, byte * buf) {
 }
 
 void fixedWriteUnlocked(Page * page, recordid rid, const byte *dat) {
+  assertlocked(page->rwlatch);
   checkRid(page,rid);
   if(!memcpy(fixed_record_ptr(page, rid.slot), dat, rid.size)) {
     perror("memcpy");

src/lladd/page/header.h

@@ -1,3 +1,9 @@
+/**
+   @file
+
+   @todo Is header.[ch] used for anything any more?  No one seems to call
+   into them.  Delete them?
+*/
 
 //void headerPageInitialize(Page * p);
 int headerPageInitialize();

src/lladd/page/indirect.c

@@ -13,39 +13,43 @@ void indirectInitialize(Page * p, int height) {
   *page_type_ptr(p) = INDIRECT_PAGE;
   memset(p->memAddr, INVALID_SLOT, ((size_t)level_ptr(p)) - ((size_t)p->memAddr));
 }
-/** @todo locking for dereferenceRID? */
+/** @todo Is locking for dereferenceRID really necessary? */
 compensated_function recordid dereferenceRID(int xid, recordid rid) {
-  Page * this;
+  Page * page;
   try_ret(NULLRID) {
-    this = loadPage(xid, rid.page);
+    page = loadPage(xid, rid.page);
+    readlock(page->rwlatch, 0);
   } end_ret(NULLRID);
   //  printf("a"); fflush(stdout);
   int offset = 0;
   int max_slot;
-  while(*page_type_ptr(this) == INDIRECT_PAGE) {
+  while(*page_type_ptr(page) == INDIRECT_PAGE) {
     int i = 0;
-    for(max_slot = *maxslot_ptr(this, i); ( max_slot + offset ) <= rid.slot; max_slot = *maxslot_ptr(this, i)) {
+    for(max_slot = *maxslot_ptr(page, i); ( max_slot + offset ) <= rid.slot; max_slot = *maxslot_ptr(page, i)) {
       i++;
       assert(max_slot != INVALID_SLOT);
     }
 
     if(i) {
-      offset += *maxslot_ptr(this, i - 1);
+      offset += *maxslot_ptr(page, i - 1);
     } /** else, the adjustment to the offset is zero */
     
-    int nextPage = *page_ptr(this, i);
+    int nextPage = *page_ptr(page, i);
 
-    releasePage(this);
+    unlock(page->rwlatch);
+    releasePage(page);
     try_ret(NULLRID) {
-      this = loadPage(xid, nextPage);
+      page = loadPage(xid, nextPage);
+      readlock(page->rwlatch, 0);
     } end_ret(NULLRID);
   }
   //  printf("b"); fflush(stdout);
   
-  rid.page = this->id;
+  rid.page = page->id;
   rid.slot -= offset;
 
-  releasePage(this);
+  unlock(page->rwlatch);
+  releasePage(page);
   //  printf("c"); fflush(stdout);
 
   return rid;
@@ -195,6 +199,7 @@ compensated_function int indirectPageRecordCount(int xid, recordid rid) {
   try_ret(-1){
     p = loadPage(xid, rid.page);
   }end_ret(-1);
+  readlock(p->rwlatch, 0);
   int i = 0;
   unsigned int ret;
   if(*page_type_ptr(p) == INDIRECT_PAGE) {
@@ -221,6 +226,8 @@ compensated_function int indirectPageRecordCount(int xid, recordid rid) {
     printf("Unknown page type in indirectPageRecordCount\n");
     abort();
   }
+
+  unlock(p->rwlatch);
   releasePage(p);
   return ret;
 }

src/lladd/page/raw.c

@@ -10,18 +10,21 @@ void rawPageInferMetadata(Page * p) {
 }
 
 byte* rawPageGetData(int xid, Page * p) {
+  assertlocked(p->rwlatch);
   return units_from_start_raw(byte, p, 0);
 }
 
 void  rawPageSetData(int xid, lsn_t lsn, Page * p) { 
-  writelock(p->rwlatch, 255);
+  assertlocked(p->rwlatch);
+  //  writelock(p->rwlatch, 255);
   rawPageWriteLSN(xid, p, lsn);
   dirtyPages_add(p);
-  unlock(p->rwlatch);
+  //  unlock(p->rwlatch);
   return;
 }
 
 lsn_t rawPageReadLSN(const Page * p) { 
+  assertlocked(p->rwlatch);
   // There are some potential optimizations here since the page
   // doesn't "really" have an LSN at all, but we need to be careful
   // about log truncation...
@@ -29,6 +32,7 @@ lsn_t rawPageReadLSN(const Page * p) {
 }
 
 void rawPageWriteLSN(int xid, Page * p, lsn_t lsn) { 
+  assertlocked(p->rwlatch);
   if(p->LSN < lsn) { p->LSN = lsn; }
 }
 

src/lladd/page/slotted.c

@@ -7,7 +7,7 @@
 static void really_do_ralloc(Page * page, recordid rid) ;
 size_t slottedFreespaceForSlot(Page * page, int slot);
 void fsckSlottedPage(const Page const * page) {
-
+  assertlocked(page->rwlatch);
 #ifdef SLOTTED_PAGE_SANITY_CHECKS
   Page dummy;
 
@@ -126,6 +126,7 @@ The caller of this function must have a writelock on the page.
 */
 
 void slottedCompact(Page * page) { 
+  assertlocked(page->rwlatch);
   Page bufPage; 
   byte buffer[PAGE_SIZE];
   bufPage.memAddr = buffer;
@@ -198,8 +199,6 @@ void slottedPageInitialize(Page * page) {
   *freespace_ptr(page) = 0;
   *numslots_ptr(page)  = 0;
   *freelist_ptr(page)  = INVALID_SLOT;
-  fsckSlottedPage(page);
-
 }
 size_t slottedFreespaceUnlocked(Page * page);
 
@@ -208,6 +207,7 @@ size_t slottedFreespaceUnlocked(Page * page);
     it takes the position of the new slot's header into account.
 */
 size_t slottedFreespaceForSlot(Page * page, int slot) { 
+  assertlocked(page->rwlatch);
   size_t slotOverhead;
 
   if(slot == INVALID_SLOT) { 
@@ -305,6 +305,7 @@ recordid slottedRawRalloc(Page * page, int size) {
   @param rid Recordid with 'internal' size.  The size should have already been translated to a type if necessary.
 */
 static void really_do_ralloc(Page * page, recordid rid) {
+  assertlocked(page->rwlatch);
 
   short freeSpace;
   
@@ -508,6 +509,7 @@ void slottedDeRalloc(int xid, Page * page, lsn_t lsn, recordid rid) {
 }
 
 void slottedReadUnlocked(Page * page, recordid rid, byte *buff) {
+  assertlocked(page->rwlatch);
   int slot_length;
 
   fsckSlottedPage(page);
@@ -556,6 +558,7 @@ void slottedWrite(Page * page, recordid rid, const byte *data) {
 
 }
 void slottedWriteUnlocked(Page * page, recordid rid, const byte *data) {
+  assertlocked(page->rwlatch);
   int slot_length;
   fsckSlottedPage(page);
 

src/lladd/transactional2.c

@@ -146,6 +146,7 @@ int Tinit() {
                                           NULL, 20, PAGE_SIZE * 1024, 1024);
         pageHandleOpen(pageFile);
 #else
+        printf("\nWarning: Using old I/O routines.\n");
         openPageFile();
 #endif // USE_PAGEFILE
 	bufInit(bufferManagerType);

test/lladd/check_bTree.c

@@ -206,6 +206,7 @@ void testFunctions(){
 
 
   // cleaning up
+
   releasePage(p1);
   Tcommit(xid);
 }
@@ -233,7 +234,7 @@ int SimpleExample(){
   int pageid1 = rid1.page;
   
   Page *  p1 = loadPage(xid, pageid1);
-
+  writelock(p1->rwlatch, 0);
   if(DEBUGP) {  printf("\n**** page->id  = %lld\n", p1->id);}
 
   /* check consistency between rid & page's values 
@@ -272,14 +273,15 @@ int SimpleExample(){
   recordid rid2 = rid1;
   rid2.slot = 0; 
 
+  // @todo This is a messy way to do this...
+  unlock(p1->rwlatch);
+ 
   fixedWrite(p1, rid2, b1);
   fixedRead(p1, rid2, b2);
   if (DEBUGP) { printf("\nb2** = %d\n",*((int *) b2));}
 
   //  initializeNewBTreeNode(p1, rid1); 
 
-
-
   releasePage(p1);
   Tcommit(xid);