forgot to add these...

lladd/truncation.h

@@ -0,0 +1,77 @@
+/*---
+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.
+---*/
+/**
+ * @file
+ *
+ * Implementation of log truncation for lladd.
+ * 
+ * $Id$
+ * 
+ */
+
+#include <lladd/transactional.h>
+
+#ifndef __LLADD_TRUNCATION_H
+#define __LLADD_TRUNCATION_H 1
+
+void dirtyPagesInit();
+void dirtyPagesDeinit();
+
+void dirtyPages_add(Page * p); 
+void dirtyPages_remove(Page * p);
+
+void truncationInit();
+void truncationDeinit();
+
+extern int lladd_enableAutoTruncation;
+
+/**
+   Spawn a periodic, demand-based log truncation thread.
+*/
+void autoTruncate();
+/**
+   Initiate a round of log truncation.
+*/
+int truncateNow();
+
+
+#endif

src/lladd/truncation.c

@@ -0,0 +1,155 @@
+#include <lladd/truncation.h>
+#include <pbl/pbl.h>
+#include <lladd/logger/logger2.h>
+#include "page.h"
+#include <assert.h>
+#include "pageFile.h"
+
+volatile static int initialized = 0;
+static int automaticallyTuncating = 0;
+static pthread_t truncationThread;
+
+static pblHashTable_t * dirtyPages = 0;
+static pthread_mutex_t dirtyPages_mutex = PTHREAD_MUTEX_INITIALIZER;
+
+int lladd_enableAutoTruncation = 1;
+#define TARGET_LOG_SIZE (1024 * 1024 * 50)
+#define TRUNCATE_INTERVAL 1
+#define MIN_INCREMENTAL_TRUNCATION (1024 * 1024 * 10)
+void dirtyPages_add(Page * p) {
+  pthread_mutex_lock(&dirtyPages_mutex);
+  if(!p->dirty) { 
+    p->dirty = 1;
+
+    assert(!pblHtLookup(dirtyPages, &(p->id), sizeof(int)));
+    pblHtInsert(dirtyPages, &(p->id), sizeof(int), (void*)p->LSN);
+  }
+  pthread_mutex_unlock(&dirtyPages_mutex);
+}
+ 
+void dirtyPages_remove(Page * p) { 
+  pthread_mutex_lock(&dirtyPages_mutex);
+  //  printf("Removing page %d\n", p->id);
+  //assert(pblHtLookup(dirtyPages, &(p->id), sizeof(int)));
+  //  printf("With lsn = %d\n", (lsn_t)pblHtCurrent(dirtyPages));
+  assert(!pblHtRemove(dirtyPages, &(p->id), sizeof(int)));
+  //assert(!pblHtLookup(dirtyPages, &(p->id), sizeof(int)));
+  pthread_mutex_unlock(&dirtyPages_mutex);
+}
+
+static lsn_t dirtyPages_minRecLSN() { 
+  lsn_t lsn = LogFlushedLSN ();
+  int* pageid;
+  pthread_mutex_lock(&dirtyPages_mutex);
+
+  for( pageid = (int*)pblHtFirst (dirtyPages); pageid; pageid = (int*)pblHtNext(dirtyPages)) { 
+    lsn_t thisLSN = (lsn_t) pblHtCurrent(dirtyPages);
+    //    printf("lsn = %d\n", thisLSN);
+    if(thisLSN < lsn) { 
+      lsn = thisLSN;
+    }
+  }
+  pthread_mutex_unlock(&dirtyPages_mutex);
+
+  return lsn;
+}
+
+static void dirtyPages_flush() { 
+  int * staleDirtyPages = malloc(sizeof(int) * MAX_BUFFER_SIZE);
+  int i;
+  for(i = 0; i < MAX_BUFFER_SIZE+1; i++) { 
+    staleDirtyPages[i] = -1;
+  }
+  Page* p = 0;
+  pthread_mutex_lock(&dirtyPages_mutex);
+  void* tmp;
+  i = 0;
+  
+  for(tmp = pblHtFirst(dirtyPages); tmp; tmp = pblHtNext(dirtyPages)) { 
+    staleDirtyPages[i] = *((int*) pblHtCurrentKey(dirtyPages));
+    i++;
+  }
+  assert(i <= MAX_BUFFER_SIZE);
+  pthread_mutex_unlock(&dirtyPages_mutex);
+
+  for(i = 0; i < MAX_BUFFER_SIZE && staleDirtyPages[i] != -1; i++) {
+    p = loadPage(-1, staleDirtyPages[i]);
+    //if(p->dirty) { 
+    pageWrite(p);
+      //      dirtyPages_remove(p);
+      //}
+    releasePage(p);
+  }
+
+}
+
+void dirtyPagesInit() { 
+  dirtyPages = pblHtCreate();
+}
+
+
+void dirtyPagesDeinit() { 
+  pblHtDelete(dirtyPages);
+  dirtyPages = 0;
+}
+void truncationInit() { 
+  initialized = 1;
+}
+
+void truncationDeinit() { 
+  initialized = 0;
+  if(automaticallyTuncating) {
+    void * ret = 0;
+    pthread_join(truncationThread, &ret);
+  }
+  automaticallyTuncating = 0;
+}
+
+static void* periodicTruncation(void * ignored) { 
+  while(initialized) { 
+    if(LogFlushedLSN() - LogTruncationPoint() > TARGET_LOG_SIZE) {
+      truncateNow();
+    }
+    // @todo TRUNCATE_INTERVAL should be dynamically set...
+    sleep(TRUNCATE_INTERVAL);
+  }
+  return (void*)0;
+}
+
+void autoTruncate() { 
+  assert(!automaticallyTuncating);
+  automaticallyTuncating = 1;
+  pthread_create(&truncationThread, 0, &periodicTruncation, 0);
+}
+
+
+int truncateNow() { 
+  lsn_t page_rec_lsn = dirtyPages_minRecLSN();
+  lsn_t xact_rec_lsn = transactions_minRecLSN();
+  lsn_t rec_lsn = page_rec_lsn < xact_rec_lsn ? page_rec_lsn : xact_rec_lsn;
+  lsn_t log_trunc = LogTruncationPoint();
+  if((rec_lsn - log_trunc) > MIN_INCREMENTAL_TRUNCATION) { 
+    printf("Truncating now. rec_lsn = %ld, log_trunc = %ld\n", rec_lsn, log_trunc);
+    LogTruncate(rec_lsn);
+    return 1;
+  } else { 
+    lsn_t flushed = LogFlushedLSN();
+    if(flushed - log_trunc > 2 * TARGET_LOG_SIZE) { 
+      printf("Flushing dirty buffers: rec_lsn = %ld log_trunc = %ld flushed = %ld\n", rec_lsn, log_trunc, flushed);
+      fflush(stdout);
+      dirtyPages_flush();
+      rec_lsn = dirtyPages_minRecLSN();
+      printf("Truncating to rec_lsn = %ld\n", rec_lsn);
+      fflush(stdout);
+      if(rec_lsn != flushed) {
+	LogTruncate(rec_lsn);
+	return 1;
+      } else {
+	return 0;
+      }
+
+    } else { 
+      return 0;
+    }
+  }
+}