Fixed page leak in Talloc().

lladd/constants.h

@@ -80,12 +80,12 @@ terms specified in this license.
 
 #define PAGE_SIZE 4096
 
-//#define MAX_BUFFER_SIZE 100003 
+#define MAX_BUFFER_SIZE 100003 
 /*#define MAX_BUFFER_SIZE 20029 */
 //#define MAX_BUFFER_SIZE 10007 
 //#define MAX_BUFFER_SIZE 5003
 //#define MAX_BUFFER_SIZE 2003 
-#define MAX_BUFFER_SIZE 4006 
+//#define MAX_BUFFER_SIZE 4006 
 /* #define MAX_BUFFER_SIZE 71  */
 /*#define MAX_BUFFER_SIZE 7 */
 

src/lladd/operations/alloc.c

@@ -187,6 +187,7 @@ compensated_function recordid Talloc(int xid, unsigned long size) {
       slottedPageInitialize(p);
     }    
     rid = TallocFromPageInternal(xid, p, size);
+    releasePage(p);
   } compensate_ret(NULLRID);
   return rid;
 }

src/lladd/operations/pageOperations.c

@@ -38,6 +38,8 @@ compensated_function int TpageSet(int xid, int pageid, byte * memAddr) {
   return 0;
 }
 
+/** @todo this should be dynamic. */
+#define TALLOC_PAGE_REGION_SIZE 128 // 512K
 
 /** 
     This calls loadPage and releasePage directly, and bypasses the
@@ -47,6 +49,12 @@ compensated_function void pageOperationsInit() {
 
   regionsInit();
 
+  boundary_tag t;
+  recordid rid = {0, 0, sizeof(boundary_tag)};
+  // Need to find a region with some free pages in it.
+  Tread(-1, rid, &t);
+  
+
   pthread_mutex_init(&pageAllocMutex, NULL);
 }
 

src/lladd/operations/regions.c

@@ -30,10 +30,10 @@ static void TreadBoundaryTag(int xid, unsigned int page, boundary_tag* tag) {
   //  printf("Reading boundary tag at %d\n", page);
   recordid rid = { page, 0, sizeof(boundary_tag) };
   Tread(xid, rid, tag);
-  Page * p = loadPage(xid, page);
+  //  Page * p = loadPage(xid, page);
   //  printf("regions.c: %d\n", *page_type_ptr(p)); fflush(NULL);
-  assert(*page_type_ptr(p) == BOUNDARY_TAG_PAGE);
+  //  assert(*page_type_ptr(p) == BOUNDARY_TAG_PAGE);
-  releasePage(p);
+  //  releasePage(p);
 }
 static void TsetBoundaryTag(int xid, unsigned int page, boundary_tag* tag) { 
   //  printf("Writing boundary tag at %d\n", page);
@@ -44,7 +44,6 @@ static void TsetBoundaryTag(int xid, unsigned int page, boundary_tag* tag) {
 void regionsInit() { 
   Page * p = loadPage(-1, 0);
   int pageType = *page_type_ptr(p);
-  releasePage(p);
   if(pageType != BOUNDARY_TAG_PAGE) {
     boundary_tag t;
     t.size = UINT32_MAX;
@@ -61,10 +60,9 @@ void regionsInit() {
     // after a crash.
     recordid rid = {0,0,sizeof(boundary_tag)};
 
-    Page * p = loadPage (-1, 0);
     operate_alloc_boundary_tag(0,p,0,rid,&t);
-    releasePage(p);
   }
+  releasePage(p);
 }
 
 pthread_mutex_t region_mutex = PTHREAD_MUTEX_INITIALIZER;

src/lladd/page/slotted.c

@@ -7,7 +7,7 @@
 #include <assert.h>
 
 static void really_do_ralloc(Page * page, recordid rid) ;
-
+size_t slottedFreespaceForSlot(Page * page, int slot);
 /**
    
 Move all of the records to the beginning of the page in order to 
@@ -92,6 +92,9 @@ void slottedCompact(Page * page) {
 	}
 
 	memcpy(page->memAddr, buffer, PAGE_SIZE);
+	
+	assert(slottedFreespaceForSlot(page, -1) || 1);
+
 }
 
 /**
@@ -128,7 +131,7 @@ void slottedPageDeInit() {
 void slottedPageInitialize(Page * page) {
   /*printf("Initializing page %d\n", page->id);
   fflush(NULL);  */
-  memset(page->memAddr, 0, PAGE_SIZE);
+  //  memset(page->memAddr, 0, PAGE_SIZE);
   *page_type_ptr(page) = SLOTTED_PAGE;
   *freespace_ptr(page) = 0;
   *numslots_ptr(page)  = 0;
@@ -144,10 +147,13 @@ size_t slottedFreespaceUnlocked(Page * page);
 size_t slottedFreespaceForSlot(Page * page, int slot) { 
   size_t slotOverhead;
 
-  if(slot >= 0 && slot < *numslots_ptr(page)) { 
+  if(slot == -1) { 
+    slotOverhead = (*freelist_ptr(page) == INVALID_SLOT) ? SLOTTED_PAGE_OVERHEAD_PER_RECORD : 0;
+  } else if(slot < *numslots_ptr(page)) { 
     slotOverhead = 0;
   } else { 
-    slotOverhead = SLOTTED_PAGE_OVERHEAD_PER_RECORD * (*numslots_ptr(page) - slot);
+    //    slotOverhead = SLOTTED_PAGE_OVERHEAD_PER_RECORD * (*numslots_ptr(page) - slot);
+    slotOverhead = SLOTTED_PAGE_OVERHEAD_PER_RECORD * ((slot+1) - *numslots_ptr(page));
   }
   // end_of_free_space points to the beginning of the slot header at the bottom of the page header.
   byte* end_of_free_space = (byte*)slot_length_ptr(page, (*numslots_ptr(page))-1); 
@@ -204,7 +210,7 @@ recordid slottedRawRalloc(Page * page, int size) {
   if(*freelist_ptr(page) != INVALID_SLOT) {
     rid.slot = *freelist_ptr(page);
     *freelist_ptr(page) = *slot_length_ptr(page, rid.slot);
-    *slot_length_ptr(page, rid.slot) = 0;
+    *slot_length_ptr(page, rid.slot) = INVALID_SLOT;
   }  
   
   really_do_ralloc(page, rid);
@@ -237,13 +243,6 @@ static void really_do_ralloc(Page * page, recordid rid) {
 
   short freeSpace;
   
-  /*  int isBlob = 0;
-
-  if(rid.size == BLOB_SLOT) {
-    isBlob = 1;
-    rid.size = sizeof(blob_record_t);
-    } */
-
   // Compact the page if we don't have enough room.
   if(slottedFreespaceForSlot(page, rid.slot) < physical_slot_length(rid.size)) {
     slottedCompact(page);
@@ -269,6 +268,8 @@ static void really_do_ralloc(Page * page, recordid rid) {
     }
     while(next != INVALID_SLOT && next != rid.slot) { 
       last = next;
+      short next_slot_ptr = *slot_ptr(page, next);
+      assert(next_slot_ptr == INVALID_SLOT);
       next = *slot_length_ptr(page, next);
     }
     if(next == rid.slot) { 
@@ -285,39 +286,50 @@ static void really_do_ralloc(Page * page, recordid rid) {
   // promote the reuse of free slot numbers, we go out of our way to make sure
   // that we put them in the list in increasing order.  (Note:  slottedCompact's 
   // correctness depends on this behavior!)
-  short lastFree = INVALID_SLOT;
-  while(*numslots_ptr(page) < rid.slot) {
-    int slot = *numslots_ptr(page);
-    short successor;
-    if(lastFree == INVALID_SLOT) { 
 
-      // The first time through, get our successor pointer from the 
-      // page's freelist pointer.
   
-      // @todo Grab this from the *end* of the freelist, since we 
+  if(rid.slot > *numslots_ptr(page)) { 
-      // know that each slot we are about to insert has a higher number
+    short lastSlot;
-      // than anything in the list.
+    short numSlots = *numslots_ptr(page);
+    if(*freelist_ptr(page) == INVALID_SLOT) { 
 
-      successor = *freelist_ptr(page);
+      *freelist_ptr(page) = numSlots;
-      *freelist_ptr(page) = slot;
+      lastSlot = numSlots;
+
+      *slot_ptr(page, lastSlot) = INVALID_SLOT;
+      // will set slot_length_ptr on next iteration.
+
+
+      (*numslots_ptr(page))++;
     } else {
-      // Put this page after the last page we inserted into the list
+      lastSlot = INVALID_SLOT;
-      successor = *slot_length_ptr(page, lastFree);
+      short next = *freelist_ptr(page);
-      *slot_length_ptr(page, lastFree) = slot;
+      while(next != INVALID_SLOT) { 
+	lastSlot = next;
+	next = *slot_length_ptr(page, lastSlot);
+	assert(lastSlot < *numslots_ptr(page));
+	assert(*slot_ptr(page, lastSlot) == INVALID_SLOT);
+      }
+      *slot_ptr(page, lastSlot) = INVALID_SLOT;
 
-      // Make sure that we didn't just find an allocated page on the free list.
-      assert(*slot_ptr(page, lastFree) == INVALID_SLOT);
     }
 
-    // Update the pointers in the new slot header.
+    // lastSlot now contains the tail of the free list.  We can start adding slots to the list starting at *numslots_ptr.
-    *slot_length_ptr(page, slot) = successor;
+    
-    *slot_ptr(page, slot) = INVALID_SLOT;
+    while(*numslots_ptr(page) < rid.slot) { 
-    (*numslots_ptr(page))++;
+      *slot_length_ptr(page, lastSlot) = *numslots_ptr(page);
-    lastFree = slot;
+      lastSlot = *numslots_ptr(page);
+      *slot_ptr(page, lastSlot) = INVALID_SLOT;
+      (*numslots_ptr(page))++;
+    }
+
+    // Terminate the end of the list.
+    *slot_length_ptr(page, lastSlot) = INVALID_SLOT;
+
   }
-  // Increment numslots_ptr if necessary.
+
   if(*numslots_ptr(page) == rid.slot) { 
-    (*numslots_ptr(page))++;
+    *numslots_ptr(page) = rid.slot+1;
   }
 
   DEBUG("Num slots %d\trid.slot %d\n", *numslots_ptr(page), rid.slot);
@@ -333,6 +345,7 @@ static void really_do_ralloc(Page * page, recordid rid) {
   //} else {
   *slot_length_ptr(page, rid.slot) = rid.size; 
   //} 
+  assert(slottedFreespaceForSlot(page, -1) || 1);
 
 }
 /**
@@ -398,8 +411,11 @@ recordid slottedPostRalloc(int xid, Page * page, lsn_t lsn, recordid rid) {
 
 	pageWriteLSN(xid, page, lsn);
 
+	assert(slottedFreespaceForSlot(page, -1) || 1);
+
 	writeunlock(page->rwlatch);
 
+
 	return rid;
 }
 
@@ -408,6 +424,7 @@ void slottedDeRalloc(int xid, Page * page, lsn_t lsn, recordid rid) {
   // readlock(page->rwlatch, 443);
   size_t oldFreeLen = slottedFreespaceUnlocked(page);
   *slot_ptr(page, rid.slot) =  INVALID_SLOT;
+  assert(*freelist_ptr(page) < *numslots_ptr(page));
   *slot_length_ptr(page, rid.slot) = *freelist_ptr(page); 
   *freelist_ptr(page) = rid.slot;  
   /*  *slot_length_ptr(page, rid.slot) = 0; */

test/lladd/check_linearHashNTA.c

@@ -133,7 +133,7 @@ START_TEST(linearHashNTAVariableSizetest)
   recordid hashHeader = ThashCreate(xid, VARIABLE_LENGTH, VARIABLE_LENGTH);
   recordid * val2;
   int i;
-  printf("\n"); fflush(stdout);
+  printf("\nstart test\n"); fflush(stdout);
   for(i = 0; i < NUM_ENTRIES; i++) {
     if(!(i % (NUM_ENTRIES/10))) {
       printf("."); fflush(stdout);
@@ -147,15 +147,19 @@ START_TEST(linearHashNTAVariableSizetest)
     ThashInsert(xid, hashHeader, (byte*)&i, sizeof(int), (byte*)&val, sizeof(recordid));
     val2 =0;
     int ret = ThashLookup(xid, hashHeader, (byte*)&i, sizeof(int), (byte**)&val2);
+
     assert(sizeof(recordid) == ret);
     assert(val2->page == i * NUM_ENTRIES);
     assert(val2->slot == val2->page * NUM_ENTRIES);
     assert(val2->size == val2->slot * NUM_ENTRIES);
     free(val2);
   }
+  printf("\nCalling commit"); fflush(NULL);
+
 
   Tcommit(xid);
   printf("\n"); fflush(stdout);
+  printf("committed\n"); fflush(NULL);
 
   xid = Tbegin();
   for(i = 0; i < NUM_ENTRIES; i+=10){
@@ -186,8 +190,10 @@ START_TEST(linearHashNTAVariableSizetest)
     assert(val2->size == val2->slot * NUM_ENTRIES);
     free(val2);
   }
+  printf("Begin shutdown.\n"); fflush(NULL);
   Tcommit(xid);
   Tdeinit();
+  printf("Test completed.\n"); fflush(NULL);
 } END_TEST