the logger understands special record types. (At least partially...); shortened blob tests, since the blob

implementation is fundamentally broken, and slow

src/lladd/logger/logEntry.c

@@ -42,6 +42,7 @@ terms specified in this license.
 
 #include <config.h>
 #include <lladd/common.h>
+#include <page.h> // For physical_slot_length()
 
 #include <lladd/transactional.h>
 
@@ -83,7 +84,7 @@ LogEntry * allocUpdateLogEntry(lsn_t prevLSN, int xid,
   int invertible = operationsTable[funcID].undo != NO_INVERSE;
   int whole_page_phys = operationsTable[funcID].undo == NO_INVERSE_WHOLE_PAGE;
   LogEntry * ret = malloc(sizeof(struct __raw_log_entry) + sizeof(UpdateLogEntry) + argSize +
-			  ((!invertible) ? rid.size : 0) + (whole_page_phys ? PAGE_SIZE : 0));
+			  ((!invertible) ? physical_slot_length(rid.size) : 0) + (whole_page_phys ? PAGE_SIZE : 0));
   ret->LSN = -1;
   ret->prevLSN = prevLSN;
   ret->xid = xid;
@@ -96,7 +97,7 @@ LogEntry * allocUpdateLogEntry(lsn_t prevLSN, int xid,
     memcpy((void*)getUpdateArgs(ret), args, argSize);
   } 
   if(!invertible) {
-    memcpy((void*)getUpdatePreImage(ret), preImage, rid.size);
+    memcpy((void*)getUpdatePreImage(ret), preImage, physical_slot_length(rid.size));
   }
   if(whole_page_phys) {
     memcpy((void*)getUpdatePreImage(ret), preImage, PAGE_SIZE);
@@ -129,7 +130,7 @@ long sizeofLogEntry(const LogEntry * log) {
     return sizeof(struct __raw_log_entry) + sizeof(CLRLogEntry);
   case UPDATELOG: 
     return sizeof(struct __raw_log_entry) + sizeof(UpdateLogEntry) + log->contents.update.argSize + 
-      ((operationsTable[log->contents.update.funcID].undo == NO_INVERSE) ? log->contents.update.rid.size : 0) +
+      ((operationsTable[log->contents.update.funcID].undo == NO_INVERSE) ? physical_slot_length(log->contents.update.rid.size) : 0) +
       ((operationsTable[log->contents.update.funcID].undo == NO_INVERSE_WHOLE_PAGE) ? PAGE_SIZE : 0) ;
   default:
     return sizeof(struct __raw_log_entry);

src/lladd/logger/logger2.c

@@ -258,7 +258,7 @@ LogEntry * LogUpdate(TransactionLog * l, Page * p, recordid rid, int operation,
 
 
   if(operationsTable[operation].sizeofData == SIZEOF_RECORD) {
-    argSize = rid.size;
+    argSize = physical_slot_length(rid.size);
   } else if(operationsTable[operation].sizeofData == SIZEIS_PAGEID) {
     argSize = rid.page;
  //   printf("argsize (page) %d, %d\n", argSize, sizeof(recordid) * 2 + sizeof(int) * 3);
@@ -267,8 +267,8 @@ LogEntry * LogUpdate(TransactionLog * l, Page * p, recordid rid, int operation,
   }
 
   if(operationsTable[operation].undo == NO_INVERSE) {
-    DEBUG("Creating %ld byte physical pre-image.\n", rid.size);
-    preImage = malloc(rid.size);
+    DEBUG("Creating %ld byte physical pre-image.\n", physical_slot_length(rid.size));
+    preImage = malloc(physical_slot_length(rid.size));
     if(!preImage) { perror("malloc"); abort(); }
     readRecord(l->xid, p, rid, preImage);
     DEBUG("got preimage");

src/lladd/page.h

@@ -183,7 +183,7 @@ struct Page_s {
 
 #define USABLE_SIZE_OF_PAGE (PAGE_SIZE - sizeof(lsn_t) - sizeof(int))
 
-
+#define physical_slot_length(size) ((size) >= 0 ? (size) : SLOT_TYPE_LENGTHS[-1*size])
 
 /**
  * initializes all the global variables needed by the functions

src/lladd/page/slotted.h

@@ -79,8 +79,6 @@ void slottedPageInitialize(Page * p);
 #define record_ptr(page, n)      bytes_from_start((page), *slot_ptr((page), (n)))
 #define isValidSlot(page, n)   ((*slot_ptr((page), (n)) == INVALID_SLOT) ? 0 : 1)
 
-#define physical_slot_length(size) ((size) >= 0 ? (size) : SLOT_TYPE_LENGTHS[-1*size])
-
 /**
  * allocate a record.  This must be done in two phases.  The first
  * phase reserves a slot, and produces a log entry.  The second phase

src/lladd/transactional2.c

@@ -216,6 +216,7 @@ compensated_function void Tupdate(int xid, recordid rid, const void *dat, int op
       } end;
     } 
   }
+
   /** @todo For logical undo logs, grabbing a lock makes no sense! */
   begin_action(releasePage, p) { 
     TupdateHelper(xid, rid, dat, op, p);

test/lladd/check_transactional2.c

@@ -50,7 +50,7 @@ terms specified in this license.
 #define THREAD_COUNT 25
 #define RECORDS_PER_THREAD 1000
 
-#define BLOBS_PER_THREAD 1000
+#define BLOBS_PER_THREAD 50
 
 void arraySet(int * array, int val) {
   int i;