added log modes to sequentialThroughput benchmark, added methods for concurrent appends to log api (and stubbed them out)

benchmarks/sequentialThroughput.c

@@ -6,7 +6,7 @@
 #include <stasis/bufferManager.h>
 #include <stasis/bufferManager/legacy/legacyBufferManager.h>
 #include <stasis/truncation.h>
-
+#include <stasis/logger/logger2.h>
 /*static stasis_handle_t * memory_factory(lsn_t off, lsn_t len, void * ignored) {
   stasis_handle_t * h = stasis_handle(open_memory)(off);
   //h = stasis_handle(open_debug)(h);
@@ -31,17 +31,26 @@ static inline long mb_to_page(long mb) {
   return (mb * 1024 * 1024) / PAGE_SIZE;
 }
 
+const char * usage = "./sequentialThroughput [--direct] [--page_count mb] [--stake mb]\n  [--deprecatedBM|--deprecatedFH|--log_safe_writes|--log_memory|--nb|--file|--pfile|--nb_pfile|--nb_file]\n";
+
 int main(int argc, char ** argv) {
   int direct = 0;
   int legacyBM = 0;
   int legacyFH = 0;
   int stake = 0;
+  int log_mode = 0;
   long page_count = mb_to_page(100);
 
   for(int i = 1; i < argc; i++) {
     if(!strcmp(argv[i], "--direct")) {
       direct = 1;
       bufferManagerO_DIRECT = 1;
+    } else if(!strcmp(argv[i], "--log_safe_writes")) {
+      stasis_log_type = LOG_TO_FILE;
+      log_mode = 1;
+    } else if(!strcmp(argv[i], "--log_memory")) {
+      stasis_log_type = LOG_TO_MEMORY;
+      log_mode = 1;
     } else if(!strcmp(argv[i], "--deprecatedBM")) {
       stasis_buffer_manager_factory = stasis_buffer_manager_deprecated_factory;
       legacyBM = 1;
@@ -71,7 +80,7 @@ int main(int argc, char ** argv) {
       i++;
       stake = mb_to_page(atoll(argv[i]));
     } else {
-      printf("Unknown argument: %s\n", argv[i]);
+      printf("Unknown argument: %s\nUsage: %s\n", argv[i], usage);
       return 1;
     }
   }
@@ -83,6 +92,24 @@ int main(int argc, char ** argv) {
 
   Tinit();
 
+  if(log_mode) {
+    lsn_t prevLSN = -1;
+    byte * arg = calloc(PAGE_SIZE, 1);
+    LogEntry * e = allocUpdateLogEntry(prevLSN, -1, OPERATION_NOOP,
+                                       0,
+                                       arg, PAGE_SIZE);
+    stasis_log_t * l = stasis_log();
+    for(long i = 0; i < page_count; i++) {
+      void * h;
+      LogEntry * e2 = l->reserve_entry(l, sizeofLogEntry(l, e), &h);
+      e->prevLSN = e->LSN;
+      e->LSN = -1;
+      memcpy(e2, e, sizeofLogEntry(l, e));
+      l->entry_done(l, e2, h);
+    }
+    freeLogEntry(e);
+    free(arg);
+  } else {
     if(stake) {
       Page * p = loadPage(-1, stake);
       writelock(p->rwlatch,0);
@@ -98,7 +125,7 @@ int main(int argc, char ** argv) {
       unlock(p->rwlatch);
       releasePage(p);
     }
-
+  }
   Tdeinit();
   return 0;
 }

src/stasis/logger/inMemoryLog.c

@@ -79,6 +79,17 @@ static int stasis_log_impl_in_memory_write_entry(stasis_log_t * log, LogEntry *e
   return 0;
 }
 
+LogEntry* stasis_log_impl_in_memory_reserve_entry(struct stasis_log_t* log, size_t sz, void **handle) {
+  // XXX need to assign LSN here.
+  return malloc(sz);
+}
+
+int stasis_log_impl_in_memory_entry_done(struct stasis_log_t* log, LogEntry* e, void * handle) {
+  int ret = stasis_log_impl_in_memory_write_entry(log, e);
+  free(e);
+  return ret;
+}
+
 static lsn_t stasis_log_impl_in_memory_first_unstable_lsn(stasis_log_t* log,
                                     stasis_log_force_mode_t mode) {
   stasis_log_impl_in_memory * impl = log->impl;
@@ -195,6 +206,8 @@ stasis_log_t* stasis_log_impl_in_memory_open() {
     stasis_log_impl_in_memory_set_truncation,
     stasis_log_impl_in_memory_sizeof_internal_entry,
     stasis_log_impl_in_memory_write_entry,
+    stasis_log_impl_in_memory_reserve_entry,
+    stasis_log_impl_in_memory_entry_done,
     stasis_log_impl_in_memory_read_entry,
     stasis_log_impl_in_memory_next_entry,
     stasis_log_impl_in_memory_first_unstable_lsn,

src/stasis/logger/safeWrites.c

@@ -367,6 +367,17 @@ static int writeLogEntry_LogWriter(stasis_log_t* log, LogEntry * e) {
   return ret;
 }
 
+LogEntry* reserveEntry_LogWriter(struct stasis_log_t* log, size_t sz, void **handle) {
+  // XXX need to assign LSN here
+  return malloc(sz);
+}
+
+int entryDone_LogWriter(struct stasis_log_t* log, LogEntry* e, void * handle) {
+  int ret = writeLogEntry_LogWriter(log, e);
+  free(e);
+  return ret;
+}
+
 static lsn_t sizeofInternalLogEntry_LogWriter(stasis_log_t * log,
                                               const LogEntry * e) {
   return sizeof(struct __raw_log_entry);
@@ -769,6 +780,8 @@ stasis_log_t* stasis_log_safe_writes_open(const char * filename,
     setTruncation_LogWriter,
     sizeofInternalLogEntry_LogWriter, // sizeof_internal_entry
     writeLogEntry_LogWriter,// write_entry
+    reserveEntry_LogWriter,
+    entryDone_LogWriter,
     readLSNEntry_LogWriter, // read_entry
     nextEntry_LogWriter,// next_entry
     flushedLSN_LogWriter, // first_unstable_lsn

stasis/logger/logger2.h

@@ -121,6 +121,10 @@ struct stasis_log_t {
   */
   int (*write_entry)(struct stasis_log_t* log, LogEntry * e);
 
+  LogEntry* (*reserve_entry)(struct stasis_log_t* log, size_t sz, void **handle);
+
+  int (*entry_done)(struct stasis_log_t* log, LogEntry* e, void * handle);
+
   /**
      Read a log entry, given its LSN.