Implemented 'compensations' for C. They're not proper compensations, but they're more flexible than pthread's mechanisms, as they allow the stack to be partially rolled up. (Although calling functions need to either check for errors and return manually, or wrap calls to error-producing callees in a begin_action{ }end_action block.

lladd/compensations.h

@@ -0,0 +1,121 @@
+#include <pthread.h>
+
+/** Rants about cpp:
+
+  There seems to be no way to add this syntax:
+  
+foo() {
+  lock * l = hashLookup(foo);
+ 
+  // stuff
+
+  compensate(l) {
+     lock(l);
+     // blah blah
+  } with {
+     unlock(l);
+  }
+
+  // more stuff
+}
+
+ => 
+
+foo() {
+  push_compensation_stack(lock_c_line_1231, l);
+
+  lock(l);
+
+  pop_compensation_stack();  // remove top stack entry and execute it.
+}
+
+void lock_c_line_1231(lock * l) {
+  unlock(l);
+}
+
+  (note that this syntax doesn't require closures!)
+
+  There are a few problems: 
+
+  1: 'compensate' and 'with' need to know the name of the
+  compensation's implementation function.
+
+  2: the 'with' block needs to move its code to the outside of the
+  enclosing function's scope, since nested functions cannot be called
+  after the function they are declared in returns.
+
+  You could try #defining a temporary variable, and reading from it in
+  the 'with' macro, but you seem to need a stack in order to support
+  that.
+
+  Here is the syntax that I've settled on:
+  
+  lock_t * l = foo();
+
+  begin_action(unlock, l) {
+     lock(l);
+     // ...
+  } compensate;
+
+  // Or:  (not recommended)
+
+  begin_action(unlock, l) {
+     lock(l);
+     // ...
+     unlock(l);
+  } end_action;
+  
+  // This is a useful variant, however:
+
+  lock(l);
+  // while loop , complicated stuff, etc
+   {
+     begin_action(unlock, l) {
+        // something that can cause an error.
+     } end_action;
+
+   }
+  unlock(l);
+
+  In all cases, an error can be raised using:
+
+  compensation_set_error(int i);
+
+  If an error is set, then each instance of begin_action and
+  end_action will cause the function to return, as appropriate.
+
+  Currently, nesting begin_actions within each other in the same
+  function will not work.  This could probably be partially fixed by
+  replacing return statements with 'break' statements, or a GOTO to
+  the proper enclosing end_action/compensate.  There may be a way to
+  #define/#undefine a variable in a way that would handle this
+  properly.
+
+  Also, begin_action(NULL, NULL) is supported, and is useful for
+  checking the return value of a called function.
+*/
+void compensations_init();
+void compensations_deinit();
+int compensation_error();
+void compensation_clear_error();
+void compensation_set_error(int code);
+
+#define begin_action(func, var)      \
+  if(compensation_error()) return;   \
+  do{                                \
+  void (*_func_)(void*);             \
+  pthread_cleanup_push(_func_=(void(*)(void*))(func), (void*)(var));\
+  do
+
+#define end_action                   \
+  while(0);                          \
+  pthread_cleanup_pop(_func_ && compensation_error());        \
+  if(compensation_error()) return;   \
+  } while(0)
+
+#define compensate                   \
+  while(0);                          \
+  pthread_cleanup_pop((int)_func_);  \
+  if(compensation_error()) return;   \
+  } while(0)
+

src/lladd/Makefile.am

@@ -7,7 +7,7 @@ liblladd_a_SOURCES=crc32.c common.c stats.c io.c bufferManager.c linkedlist.c op
                    pageFile.c pageCache.c page.c blobManager.c recovery2.c transactional2.c \
 		   lockManager.c \
 	           logger/logEntry.c logger/logWriter.c logger/logHandle.c logger/logger2.c \
-		   page/slotted.c page/header.c page/fixed.c \
+		   page/slotted.c page/header.c page/fixed.c  compensations.c \
 	           operations/pageOperations.c page/indirect.c operations/decrement.c  \
 		   operations/increment.c  operations/prepare.c  operations/set.c \
 		   operations/alloc.c operations/noop.c operations/instantSet.c \

src/lladd/compensations.c

@@ -0,0 +1,26 @@
+#include <lladd/compensations.h>
+#include <assert.h>
+pthread_key_t error_key;
+
+void compensations_init () {
+  int ret = pthread_key_create(&error_key, NULL);
+  assert(!ret); 
+  pthread_setspecific(error_key, NULL);
+}
+
+void compensations_deinit() {
+  pthread_key_delete(error_key);
+}
+
+int compensation_error() {
+  int error = (int) pthread_getspecific(error_key);
+  return error;
+}
+
+void compensation_clear_error() {
+  compensation_set_error(0);
+}
+
+void compensation_set_error(int error) {
+  pthread_setspecific(error_key, (void *)error);
+}

src/lladd/lockManager.c

@@ -1,5 +1,6 @@
 #include <pbl/pbl.h>
 #include <lladd/lockManager.h>
+#include <lladd/compensations.h>
 #include <sys/time.h>
 #include <time.h>
 #include <pthread.h>
@@ -127,9 +128,7 @@ int lockManagerReadLockHashed(int xid, byte * dat, int datLen) {
       if(wait_ret == ETIMEDOUT) {
 	ridLock->active--;
 	pthread_mutex_unlock(mut);
-
+	compensation_set_error(LLADD_DEADLOCK);
-	printf("Deadlock!\n"); fflush(stdout);
-	abort();
 	return LLADD_DEADLOCK;
       }
     } while(ridLock->writers);
@@ -187,6 +186,7 @@ int lockManagerWriteLockHashed(int xid, byte * dat, int datLen) {
     ts.tv_nsec = tv.tv_usec * 1000;
     if(tod_ret != 0) {
       perror("Could not get time of day");
+      compensation_set_error(LLADD_DEADLOCK);
       return LLADD_INTERNAL_ERROR;
     }
     while(ridLock->writers || (ridLock->readers - me)) { 
@@ -195,8 +195,7 @@ int lockManagerWriteLockHashed(int xid, byte * dat, int datLen) {
 	ridLock->waiting--;
 	ridLock->active--;
 	pthread_mutex_unlock(mut);
-	printf("Deadlock!\n"); fflush(stdout);
+	compensation_set_error(LLADD_DEADLOCK);
-	abort();
 	return LLADD_DEADLOCK;
       }
     }

src/lladd/page.c

@@ -85,7 +85,7 @@ terms specified in this license.
 #include "blobManager.h"
 #include <lladd/lockManager.h>
 #include "pageFile.h"
-
+#include <lladd/compensations.h>
 #include "page/slotted.h"
 #include "page/fixed.h"
 
@@ -106,13 +106,19 @@ Page pool[MAX_BUFFER_SIZE+1];
 void pageWriteLSN(int xid, Page * page, lsn_t lsn) {
   /* unlocked since we're only called by a function that holds the writelock. */
   /*  *(long *)(page->memAddr + START_OF_LSN) = page->LSN; */
-  if(globalLockManager.writeLockPage) { globalLockManager.writeLockPage(xid, page->id); }
+  
+  begin_action(NULL,NULL) {
+    if(globalLockManager.writeLockPage) { 
+      globalLockManager.writeLockPage(xid, page->id); 
+    }
+  } end_action;
 
   if(page->LSN < lsn) {
     page->LSN = lsn;
     *lsn_ptr(page) = page->LSN;
   } 
   page->dirty = 1;
+  return;
 }
 
 lsn_t pageReadLSN(const Page * page) {
@@ -253,6 +259,20 @@ Page *pageMalloc() {
 void writeRecord(int xid, Page * p, lsn_t lsn, recordid rid, const void *dat) {
 
   assert( (p->id == rid.page) && (p->memAddr != NULL) );	
+
+  
+  /*  writelock(p->rwlatch, 225);  // Need a writelock so that we can update the lsn. 
+  int lock_ret = pageWriteLSN(xid, p, lsn);
+  unlock(p->rwlatch);    
+  if(lock_ret) {
+    return lock_ret;
+    } */
+
+  begin_action((void(*)(void*))unlock, p->rwlatch) {
+    writelock(p->rwlatch, 225);
+    pageWriteLSN(xid, p, lsn);
+  } compensate;
+
   
   if(rid.size > BLOB_THRESHOLD_SIZE) {
     writeBlob(xid, p, lsn, rid, dat);
@@ -265,13 +285,9 @@ void writeRecord(int xid, Page * p, lsn_t lsn, recordid rid, const void *dat) {
   }
   assert( (p->id == rid.page) && (p->memAddr != NULL) );	
   
-  writelock(p->rwlatch, 225);  /* Need a writelock so that we can update the lsn. */
-  pageWriteLSN(xid, p, lsn);
-  unlock(p->rwlatch);    
-
 }
 
-void readRecord(int xid, Page * p, recordid rid, void *buf) {
+int readRecord(int xid, Page * p, recordid rid, void *buf) {
   assert(rid.page == p->id); 
   
   int page_type = *page_type_ptr(p);
@@ -288,10 +304,12 @@ void readRecord(int xid, Page * p, recordid rid, void *buf) {
     abort();
   }
   assert(rid.page == p->id); 
+
+  return 0;
 }
 
 
-void readRecordUnlocked(int xid, Page * p, recordid rid, void *buf) {
+int readRecordUnlocked(int xid, Page * p, recordid rid, void *buf) {
   assert(rid.page == p->id); 
   
   int page_type = *page_type_ptr(p);
@@ -309,15 +327,10 @@ void readRecordUnlocked(int xid, Page * p, recordid rid, void *buf) {
     abort();
   }
   assert(rid.page == p->id); 
+
+  return 0;
 }
-/** @todo getRecordType is a hack.  Instead, each record type should
-    implement code that decides whether a record exists, and returns its size
-    or -1.  Then, getRecordType coudl call that function directly depending on 
-	  page type, etc.
 
-	  A complementary function getRecordSize could return the size value.
-
-*/
 int getRecordTypeUnlocked(int xid, Page * p, recordid rid) {
   assert(rid.page == p->id);
   
@@ -349,7 +362,7 @@ int getRecordType(int xid, Page * p, recordid rid) {
 	unlock(p->rwlatch);
 	return ret;
 }
-/** @todo implemenet getRecordLength for blobs and fixed length pages. */
+/** @todo implement getRecordLength for blobs and fixed length pages. */
 int getRecordSize(int xid, Page * p, recordid rid) {
   readlock(p->rwlatch, 353);
   int ret = getRecordTypeUnlocked(xid, p, rid);
@@ -367,7 +380,22 @@ int getRecordSize(int xid, Page * p, recordid rid) {
 void writeRecordUnlocked(int xid, Page * p, lsn_t lsn, recordid rid, const void *dat) {
 
   assert( (p->id == rid.page) && (p->memAddr != NULL) );	
-  
+
+  /*  writelock(p->rwlatch, 225);  // Need a writelock so that we can update the lsn. 
+  int lock_error = pageWriteLSN(xid, p, lsn);
+  if(lock_error) {
+    unlock(p->rwlatch);
+    return lock_error;
+  }
+  unlock(p->rwlatch);    */
+
+
+  // Need a writelock so that we can update the lsn. 
+  begin_action(unlock, p->rwlatch) {
+    writelock(p->rwlatch, 225);
+    pageWriteLSN(xid, p, lsn);
+  } compensate;
+
   if(rid.size > BLOB_THRESHOLD_SIZE) {
     abort();
     writeBlob(xid, p, lsn, rid, dat);
@@ -380,8 +408,5 @@ void writeRecordUnlocked(int xid, Page * p, lsn_t lsn, recordid rid, const void
   }
   assert( (p->id == rid.page) && (p->memAddr != NULL) );	
   
-  writelock(p->rwlatch, 225);  /* Need a writelock so that we can update the lsn. */
-  pageWriteLSN(xid, p, lsn);
-  unlock(p->rwlatch);    
 
 }

src/lladd/page.h

@@ -261,6 +261,8 @@ lsn_t pageReadLSN(const Page * page);
  *
  * @param dat the new value of the record.
  *
+ * @return 0 on success, lladd error code on failure
+ *
  */
 void writeRecord(int xid, Page * page, lsn_t lsn, recordid rid, const void *dat); 
 /**
@@ -271,12 +273,13 @@ void writeRecordUnlocked(int xid, Page * page, lsn_t lsn, recordid rid, const vo
  * @param xid transaction ID
  * @param rid the record to be written
  * @param dat buffer for data
+ * @return 0 on success, lladd error code on failure
  */
-void readRecord(int xid, Page * page, recordid rid, void *dat);
+int readRecord(int xid, Page * page, recordid rid, void *dat);
 /**
  *  The same as readRecord, but does not obtain a latch.
  */
-void readRecordUnlocked(int xid, Page * p, recordid rid, void *buf);
+int readRecordUnlocked(int xid, Page * p, recordid rid, void *buf);
 /**
    Should be called when transaction xid commits.
 */
@@ -299,7 +302,7 @@ void  pageRealloc(Page * p, int id);
 /**
    obtains the type of the record pointed to by rid.  
 
-   @return UNINITIALIZED_RECORD, BLOB_RECORD, SLOTTED_RECORD, or FIXED_RECORD.
+   @return UNINITIALIZED_RECORD, BLOB_RECORD, SLOTTED_RECORD, FIXED_RECORD or an error code.
 */
 
 

src/lladd/transactional2.c

@@ -7,6 +7,7 @@
 #include "logger/logWriter.h"
 #include <lladd/bufferManager.h>
 #include <lladd/lockManager.h>
+#include <lladd/compensations.h>
 
 #include "page.h"
 #include <lladd/logger/logger2.h>
@@ -92,9 +93,11 @@ int Tinit() {
 	pageOperationsInit();
 	initNestedTopActions();
 	ThashInit();
+	
+	compensations_init();
 
-	//setupLockManagerCallbacksNil();
+	setupLockManagerCallbacksNil();
-	setupLockManagerCallbacksPage();
+	//setupLockManagerCallbacksPage();
 
 	InitiateRecovery();
 

test/lladd/Makefile.am

@@ -1,12 +1,12 @@
 INCLUDES = @CHECK_CFLAGS@
 if HAVE_CHECK
 ## Had to disable check_lht because lht needs to be rewritten.
-TESTS = check_logEntry check_logWriter check_page check_operations check_transactional2 check_recovery check_blobRecovery check_bufferManager check_indirect check_pageOperations check_linearHash  check_logicalLinearHash check_header check_linkedListNTA check_linearHashNTA check_pageOrientedList check_lockManager
+TESTS = check_logEntry check_logWriter check_page check_operations check_transactional2 check_recovery check_blobRecovery check_bufferManager check_indirect check_pageOperations check_linearHash  check_logicalLinearHash check_header check_linkedListNTA check_linearHashNTA check_pageOrientedList check_lockManager check_compensations
 #check_lladdhash
 else
 TESTS = 
 endif
 noinst_PROGRAMS = $(TESTS)
 LDADD = @CHECK_LIBS@ $(top_builddir)/src/lladd/liblladd.a $(top_builddir)/src/pbl/libpbl.a $(top_builddir)/src/libdfa/librw.a #-lefence
-CLEANFILES = check_lht.log check_logEntry.log storefile.txt logfile.txt blob0_file.txt blob1_file.txt check_blobRecovery.log check_logWriter.log check_operations.log check_recovery.log check_transactional2.log check_page.log check_bufferManager.log check_indirect.log check_bufferMananger.log check_lladdhash.log check_pageOperations.log check_linearhash.log check_linkedListNTA.log check_linearHashNTA.log check_pageOrientedListNTA.log check_lockManager.log
+CLEANFILES = check_lht.log check_logEntry.log storefile.txt logfile.txt blob0_file.txt blob1_file.txt check_blobRecovery.log check_logWriter.log check_operations.log check_recovery.log check_transactional2.log check_page.log check_bufferManager.log check_indirect.log check_bufferMananger.log check_lladdhash.log check_pageOperations.log check_linearhash.log check_linkedListNTA.log check_linearHashNTA.log check_pageOrientedListNTA.log check_lockManager.log check_compensations.log
 AM_CFLAGS= -g -Wall -pedantic -std=gnu99

test/lladd/check_compensations.c

@@ -0,0 +1,152 @@
+
+/*---
+This software is copyrighted by the Regents of the University of
+California, and other parties. The following terms apply to all files
+associated with the software unless explicitly disclaimed in
+individual files.
+
+The authors hereby grant permission to use, copy, modify, distribute,
+and license this software and its documentation for any purpose,
+provided that existing copyright notices are retained in all copies
+and that this notice is included verbatim in any distributions. No
+written agreement, license, or royalty fee is required for any of the
+authorized uses. Modifications to this software may be copyrighted by
+their authors and need not follow the licensing terms described here,
+provided that the new terms are clearly indicated on the first page of
+each file where they apply.
+
+IN NO EVENT SHALL THE AUTHORS OR DISTRIBUTORS BE LIABLE TO ANY PARTY
+FOR DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES
+ARISING OUT OF THE USE OF THIS SOFTWARE, ITS DOCUMENTATION, OR ANY
+DERIVATIVES THEREOF, EVEN IF THE AUTHORS HAVE BEEN ADVISED OF THE
+POSSIBILITY OF SUCH DAMAGE.
+
+THE AUTHORS AND DISTRIBUTORS SPECIFICALLY DISCLAIM ANY WARRANTIES,
+INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, AND
+NON-INFRINGEMENT. THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, AND
+THE AUTHORS AND DISTRIBUTORS HAVE NO OBLIGATION TO PROVIDE
+MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
+
+GOVERNMENT USE: If you are acquiring this software on behalf of the
+U.S. government, the Government shall have only "Restricted Rights" in
+the software and related documentation as defined in the Federal
+Acquisition Regulations (FARs) in Clause 52.227.19 (c) (2). If you are
+acquiring the software on behalf of the Department of Defense, the
+software shall be classified as "Commercial Computer Software" and the
+Government shall have only "Restricted Rights" as defined in Clause
+252.227-7013 (c) (1) of DFARs. Notwithstanding the foregoing, the
+authors grant the U.S. Government and others acting in its behalf
+permission to use and distribute the software in accordance with the
+terms specified in this license.
+---*/
+
+#include <config.h>
+#include <check.h>
+#include <malloc.h>
+#include <lladd/compensations.h>
+
+#include <assert.h>
+#include "../check_includes.h"
+
+
+#define LOG_NAME   "check_compensations.log"
+
+void decrement(void * j) {
+  int * i = (int*) j;
+  *i = *i-1;
+}
+
+void nested(int * i);
+void happy_nest(int * i);
+void nested2(int * i) {
+  begin_action(NULL, NULL) {
+    nested(i);
+  } compensate;
+  assert(0);
+}
+void nested3(int * i) {
+  begin_action(NULL, NULL) {
+    happy_nest(i);
+  } compensate;
+}
+
+void nested(int * i) {
+  begin_action(decrement, i) {
+    (*i)++;
+    compensation_set_error(1);
+    break; 
+    assert(0);
+  } end_action;
+  assert(0);
+}
+void happy_nest(int * i) {
+  begin_action(decrement, i) {
+    (*i)++;
+  } compensate;
+}
+
+/**
+   @test 
+
+*/
+START_TEST(compensationTest)
+{
+  compensations_init();
+  
+  int * i = malloc (sizeof(int));
+  *i = 0;
+  begin_action(decrement, i) {
+    (*i)++;
+    assert(*i == 1);
+  } compensate;
+
+  assert(*i == 0);
+
+  begin_action(decrement, i) {
+    (*i)++;
+    assert(*i == 1);
+  } end_action;
+
+  assert(*i == 1);
+
+  nested(i); // has no effect.
+  assert(compensation_error() == 1);
+  assert(*i == 1);
+
+  compensation_clear_error();
+
+  nested2(i); // has no effect.
+  assert(compensation_error() == 1);
+  assert(*i == 1);
+  compensation_clear_error();
+
+  nested3(i); // has no effect.
+  assert(compensation_error() == 0);
+  assert(*i == 1);
+
+  free(i);
+
+  compensations_deinit();
+
+}
+END_TEST
+
+Suite * check_suite(void) {
+  Suite *s = suite_create("compensations");
+  /* Begin a new test */
+  TCase *tc = tcase_create("simple_compensations");
+
+  /* Sub tests are added, one per line, here */
+
+  tcase_add_test(tc, compensationTest);
+
+  /* --------------------------------------------- */
+  
+  tcase_add_checked_fixture(tc, setup, teardown);
+
+  suite_add_tcase(s, tc);
+  return s;
+}
+
+#include "../check_setup.h"

test/lladd/check_page.c

@@ -487,3 +487,4 @@ Suite * check_suite(void) {
 }
 
 #include "../check_setup.h"
+