Add TarrayListDealloc() to arrayList. Remove one region allocation from each array list alloc. Add unit test for dealloc.

Also, removed some old compensations cruft.

src/stasis/operations/arrayList.c

@@ -141,12 +141,13 @@ recordid stasis_array_list_dereference_recordid(int xid, Page * p, int offset) {
 
 /*----------------------------------------------------------------------------*/
 
-compensated_function recordid TarrayListAlloc(int xid, pageid_t count, int multiplier, int recordSize) {
+recordid TarrayListAlloc(int xid, pageid_t count, int multiplier, int recordSize) {
 
   pageid_t firstPage;
-  try_ret(NULLRID) {
   firstPage = TpageAllocMany(xid, count+1);
-  } end_ret(NULLRID);
+
+  if(firstPage == INVALID_PAGE) { return NULLRID; }
+
   array_list_parameter_t alp;
 
   alp.firstPage = firstPage;
@@ -160,13 +161,26 @@ compensated_function recordid TarrayListAlloc(int xid, pageid_t count, int multi
   rid.page = firstPage;
   rid.slot = 0; /* number of slots in array (maxOffset + 1) */
   rid.size = recordSize;
-  try_ret(NULLRID) {
   Tupdate(xid, firstPage, &alp, sizeof(alp), OPERATION_ARRAY_LIST_HEADER_INIT);
-  } end_ret(NULLRID);
+  TinitializeFixedPage(xid, firstPage+1, alp.size);
 
   return rid;
 }
 
+void TarrayListDealloc(int xid, recordid rid) {
+  Page * p = loadPage(xid, rid.page);
+  array_list_parameter_t alp = array_list_read_parameter(xid, p);
+  slotid_t n = array_list_get_block_containing_offset(alp, alp.maxOffset, NULL) + 1;
+  for(slotid_t i = 1; i < n; i++) {  // block 0 points to p->id + 1.
+    pageid_t pid;
+    rid.slot = i + FIRST_DATA_PAGE_OFFSET;
+    stasis_record_read(xid, p, rid, (byte*)&pid);
+    TpageDeallocMany(xid, pid);
+  }
+  releasePage(p);
+  TpageDeallocMany(xid, rid.page);
+}
+
 /** @todo locking for arrayList... this isn't pressing since currently
     the only thing that calls arraylist (the hashtable
     implementations) serialize bucket list operations anyway...
@@ -188,7 +202,7 @@ compensated_function int TarrayListExtend(int xid, recordid rid, int slots) {
 
   int lastCurrentBlock; // just a slot on a page
   if(alp.maxOffset == -1) {
-    lastCurrentBlock = -1;
+    lastCurrentBlock = 0;
   } else{
     lastCurrentBlock = array_list_get_block_containing_offset(alp, alp.maxOffset, NULL);
   }
@@ -237,7 +251,7 @@ compensated_function int TarrayListExtend(int xid, recordid rid, int slots) {
 
 }
 
-compensated_function int TarrayListLength(int xid, recordid rid) {
+int TarrayListLength(int xid, recordid rid) {
  Page * p = loadPage(xid, rid.page);
  readlock(p->rwlatch, 0);
  array_list_parameter_t alp

src/stasis/operations/pageOperations.c

@@ -110,10 +110,12 @@ pageid_t TfixedPageAlloc(int xid, int size) {
   return page;
 }
 
-compensated_function pageid_t TpageAllocMany(int xid, int count) {
+pageid_t TpageAllocMany(int xid, int count) {
   return TregionAlloc(xid, count, STORAGE_MANAGER_NAIVE_PAGE_ALLOC);
 }
-
+void TpageDeallocMany(int xid, pageid_t pid) {
+  TregionDealloc(xid, pid);
+}
 int TpageGetType(int xid, pageid_t page) {
   Page * p = loadPage(xid, page);
   int ret = p->pageType;

src/stasis/operations/regions.c

@@ -162,7 +162,6 @@ void regionsInit(stasis_log_t *log) {
   holding_mutex = 0;
   releasePage(p);
 }
-const static int FIRST_TAG = 0;
 int TregionNextBoundaryTag(int xid, pageid_t* pid, boundary_tag * tag, int type) {
   pthread_mutex_lock(&region_mutex);
   assert(0 == holding_mutex);
@@ -184,8 +183,6 @@ int TregionNextBoundaryTag(int xid, pageid_t* pid, boundary_tag * tag, int type)
         break;
       }
     }
-  } else {
-    ret = 0;
   }
   holding_mutex = 0;
   pthread_mutex_unlock(&region_mutex);

src/stasis/transactional2.c

@@ -138,7 +138,7 @@ int Tbegin() {
   }
 }
 
-compensated_function void Tupdate(int xid, pageid_t page,
+void Tupdate(int xid, pageid_t page,
 					       const void * dat, size_t datlen, int op) {
   assert(stasis_initted);
   assert(page != INVALID_PAGE);
@@ -147,11 +147,9 @@ compensated_function void Tupdate(int xid, pageid_t page,
   assert(xact);
   Page * p = loadPageForOperation(xid, page, op);
 
-  try {
   if(globalLockManager.writeLockPage && p) {
     globalLockManager.writeLockPage(xid, page);
   }
-  } end;
 
   if(p) writelock(p->rwlatch,0);
 
@@ -225,12 +223,12 @@ void TreorderableWritebackUpdate(int xid, void* hp,
   pthread_mutex_unlock(&h->mut);
   free(e);
 }
-compensated_function void TupdateStr(int xid, pageid_t page,
+void TupdateStr(int xid, pageid_t page,
                                      const char *dat, size_t datlen, int op) {
   Tupdate(xid, page, dat, datlen, op);
 }
 
-compensated_function void TreadStr(int xid, recordid rid, char * dat) {
+void TreadStr(int xid, recordid rid, char * dat) {
   Tread(xid, rid, dat);
 }
 

stasis/operations/arrayList.h

@@ -92,7 +92,9 @@ terms specified in this license.
 
 */
 
-compensated_function recordid TarrayListAlloc(int xid, pageid_t numPages, int multiplier, int recordSize);
+recordid TarrayListAlloc(int xid, pageid_t numPages, int multiplier, int recordSize);
+void TarrayListDealloc(int xid, recordid rid);
+
 /**
    Extend the ArrayList in place.
 

stasis/operations/pageOperations.h

@@ -59,16 +59,17 @@ terms specified in this license.
 #define __PAGE_OPERATIONS_H__
 #include <stasis/operations.h>
 
-compensated_function pageid_t TpageAlloc(int xid);
-compensated_function pageid_t TfixedPageAlloc(int xid, int size);
-compensated_function pageid_t TpageAllocMany(int xid, int count);
-compensated_function int TpageDealloc(int xid, pageid_t page);
-compensated_function int TpageSet(int xid, pageid_t page, const void* dat);
-compensated_function int TpageSetRange(int xid, pageid_t page, int offset, const void* dat, int len);
-compensated_function int TpageGet(int xid, pageid_t page, void* buf);
+pageid_t TpageAlloc(int xid);
+pageid_t TfixedPageAlloc(int xid, int size);
+pageid_t TpageAllocMany(int xid, int count);
+int TpageDealloc(int xid, pageid_t page);
+void TpageDeallocMany(int xid, pageid_t page);
+int TpageSet(int xid, pageid_t page, const void* dat);
+int TpageSetRange(int xid, pageid_t page, int offset, const void* dat, int len);
+int TpageGet(int xid, pageid_t page, void* buf);
 
-compensated_function void TinitializeSlottedPage(int xid, pageid_t page);
-compensated_function void TinitializeFixedPage(int xid, pageid_t page,
+void TinitializeSlottedPage(int xid, pageid_t page);
+void TinitializeFixedPage(int xid, pageid_t page,
 					       int slotLength);
 
 int TpageGetType(int xid, pageid_t page);

stasis/operations/regions.h

@@ -35,8 +35,24 @@ void TregionForce(int xid, pageid_t pid);
 
 /** Currently, this function is O(n) in the number of regions, so be careful! */
 void TregionFindNthActive(int xid, pageid_t n, pageid_t * firstPage, pageid_t * size);
-
+/**
+ * Read the active boundary tag that follows the given region.
+ * Inactive boundary tags (ones that are locked by ongoing transactions, or that point
+ * to free space) will be skipped.
+ *
+ * @param xid The transaction reading the boundary tag
+ * @param pid A pointer to the current pageid.
+ * @param tag A pointer to a buffer that will hold the next boundary tag.
+ * @param allocationManager The allocation manager whose tags should be returned, or 0 if all active tags should be returned.
+ * @return 0 on failure, true on success
+ */
 int TregionNextBoundaryTag(int xid, pageid_t*pid, boundary_tag *tag, int allocationManager);
+/** Read the boundary tag marking the first page of some region.  If pid = ROOT_RECORD.page, this will return the first boundary tag.
+ *
+ * @param xid The transaction examining the boundary tag
+ * @param pid The first page in the region (ie: the one returned by TregionAlloc().  In the current implementation, the boundary tag lives on the page before this one.
+ * @param tag A buffer that the tag will be read into.
+ */
 int TregionReadBoundaryTag(int xid, pageid_t pid, boundary_tag *tag);
 
 stasis_operation_impl stasis_op_impl_boundary_tag_alloc();

stasis/transactional.h

@@ -581,12 +581,12 @@ int Tbegin(void);
  *
  * @see operations.h For an overview of the operations API
  */
-compensated_function void Tupdate(int xid, pageid_t page,
+void Tupdate(int xid, pageid_t page,
 				  const void *dat, size_t datlen, int op);
 /**
    @deprecated Only exists to work around swig/python limitations.
  */
-compensated_function void TupdateStr(int xid, pageid_t page,
+void TupdateStr(int xid, pageid_t page,
                                      const char *dat, size_t datlen, int op);
 
 void TreorderableUpdate(int xid, void * h, pageid_t page,

test/stasis/check_operations.c

@@ -543,6 +543,8 @@ START_TEST(operation_array_list) {
     Tread(xid, rid2, &j);
     assert(i == j);
   }
+
+  TarrayListDealloc(xid, rid);
   Tcommit(xid);
 
   Tdeinit();