This commit breaks the build so that Jimmy and I can sync. Sorry. :(

lladd/consumer.h

@@ -8,7 +8,7 @@
 #define ARRAY_CONSUMER      1
 
 typedef struct {
-
+  int foo;
 } lladdConsumer_def_t;
 
 typedef struct {
@@ -19,7 +19,7 @@ typedef struct {
 /* call once per Tinit() call */
 void consumer_init();
 
-Tconsumer_close(int xid, lladdConsumer_t * it);
+void Tconsumer_close(int xid, lladdConsumer_t * it);
 /**
    
    @param xid Transaction id
@@ -35,4 +35,6 @@ int Tconsumer_push(int xid, lladdConsumer_t * it, byte * key, size_t keySize, by
 /* @see Tconsumer_push
    @return Error, or 'consumer full'
 */
-int Tconsumer_tryPush(int xid, ....);
+//int Tconsumer_tryPush(int xid, ....);
+
+#endif

lladd/iterator.h

@@ -64,7 +64,7 @@ int Titerator_next(int xid, lladdIterator_t * it);
 */
 int Titerator_key(int xid, lladdIterator_t * it, byte ** key);
 /** 
-    Analagour to lladdIterator_key.
+    Analagous to lladdIterator_key.
 
     @see lladdIterator_key.
 */

lladd/operations/linearHashNTA.h

@@ -113,6 +113,23 @@ int                   linearHashNTAIterator_next (int xid, void * it);
 int                   linearHashNTAIterator_key  (int xid, void * it, byte **key);
 int                   linearHashNTAIterator_value(int xid, void * it, byte **value);
 
+/** @todo these should be in linearHashNTA.c, but they've been moved
+    here so that multiplexer.c can (temoprarily) implement a
+    multiplexer for logical hash operations. */
+
+typedef struct {
+  recordid hashHeader;
+  int keySize;
+} linearHash_insert_arg;
+
+typedef struct {
+  recordid hashHeader;
+  int keySize;
+  int valueSize;
+} linearHash_remove_arg;
+
+
+
 //Support 16 entries by default.
 #define HASH_INIT_BITS 4
 #define HASH_FILL_FACTOR 0.7

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 iterator.c arrayCollection.c \
 	           logger/logEntry.c logger/logWriter.c logger/logHandle.c logger/logger2.c \
-		   logger/logMemory.c \
+		   logger/logMemory.c multiplexer.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 \

src/lladd/logger/logMemory.h

@@ -6,3 +6,19 @@ ringBufferLog_t * openLogRingBuffer(size_t size, lsn_t initialOffset);
 void closeLogRingBuffer(ringBufferLog_t * log);
 int ringBufferAppend(ringBufferLog_t * log, byte * dat, size_t size);
 int ringBufferTruncateRead(byte * buf, ringBufferLog_t * log,size_t size);
+
+// int writeLogEntry(LogEntry * e); (as implemented by ring buffer, but need to remember size of entry too
+
+// void syncLog() (block writers until log is empty / call sync log on consumers?)
+
+// lsn_t flushedLSN  return lsn as of last syncLog?
+
+// int truncateLog(lsn_t) (no-op?)
+
+// lsn_t firstLogEntry()  Unimplemented?  Or return first thing in current ring buffer?
+
+// void closeLogWriter() (syncLog + closeRingBuffer)
+
+// deleteLogWriter no-op
+
+// LogEntry* readLSNEntry(lsn_t) if LSN is the next available log entry, return it, else error.

src/lladd/operations/linearHashNTA.c

@@ -105,7 +105,7 @@ compensated_function void ThashDelete(int xid, recordid hash) {
 compensated_function static int __ThashInsert(int xid, recordid hashHeader, const byte* key, int keySize, const byte* value, int valueSize);
 compensated_function static int __ThashRemove(int xid, recordid hashHeader, const byte * key, int keySize);
 
-typedef struct {
+/*typedef struct {
   recordid hashHeader;
   int keySize;
 } linearHash_insert_arg;
@@ -114,7 +114,7 @@ typedef struct {
   recordid hashHeader;
   int keySize;
   int valueSize;
-} linearHash_remove_arg;
+  } linearHash_remove_arg;*/
 
 compensated_function static int operateInsert(int xid, Page *p,  lsn_t lsn, recordid rid, const void *dat) {
   const linearHash_remove_arg * args = dat;
@@ -185,10 +185,11 @@ compensated_function int ThashInsert(int xid, recordid hashHeader, const byte* k
     free(arg);
     ret = __ThashInsert(xid, hashHeader, key, keySize, value, valueSize);  
   } end_action_ret(compensation_error());
-  //  beg in_action_ret(pthread_mutex_unlock, &linear_hash_mutex, compensation_error()) {
-    TendNestedTopAction(xid, handle);
-    // } comp ensate_ret(compensation_error());
-    pthread_mutex_unlock(&linear_hash_mutex);
+
+  TendNestedTopAction(xid, handle);
+
+  pthread_mutex_unlock(&linear_hash_mutex);
+
   return ret;
 }
 compensated_function static int __ThashInsert(int xid, recordid hashHeader, const byte* key, int keySize, const byte* value, int valueSize) {