X-Git-Url: https://pd.if.org/git/?p=btree;a=blobdiff_plain;f=threadskv10.c;h=194757b25f2aa192b379b01511ef567338dd76ad;hp=c524a48ae50ac7d3d92cbdc6c399347a9ecf6822;hb=392e5f08cc164c87e56153aa78a740f93325750e;hpb=7fe45d207c827702c6f61be4104afce5a4dee3d0

diff --git a/threadskv10.c b/threadskv10.c
index c524a48..194757b 100644
--- a/threadskv10.c
+++ b/threadskv10.c
@@ -1,4 +1,4 @@
-// btree version threadskv10 FUTEX version
+// btree version threadskv10 futex version
 //	with reworked bt_deletekey code,
 //	phase-fair reader writer lock,
 //	librarian page split code,
@@ -7,9 +7,9 @@
 //	traditional buffer pool manager
 //	ACID batched key-value updates
 //	redo log for failure recovery
-//	and dual B-trees for write optimization
+//	and LSM B-trees for write optimization
 
-// 09 OCT 2014
+// 15 OCT 2014
 
 // author: karl malbrain, malbrain@cal.berkeley.edu
 
@@ -110,6 +110,19 @@ typedef enum{
 	BtLockAtomic = 32
 } BtLock;
 
+typedef struct {
+  union {
+	struct {
+	  volatile ushort xlock[1];		// one writer has exclusive lock
+	  volatile ushort wrt[1];		// count of other writers waiting
+	} bits[1];
+	uint value[1];
+  };
+} BtMutexLatch;
+
+#define XCL 1
+#define WRT 65536
+
 //	definition for phase-fair reader/writer lock implementation
 
 typedef struct {
@@ -117,16 +130,20 @@ typedef struct {
 	volatile ushort rout[1];
 	volatile ushort ticket[1];
 	volatile ushort serving[1];
-	ushort tid;
-	ushort dup;
 } RWLock;
 
-//	write only lock
+//	write only reentrant lock
 
 typedef struct {
-	volatile uint exclusive[1];
-	ushort tid;
-	ushort dup;
+  BtMutexLatch xcl[1];
+  union {
+	struct {
+		volatile ushort tid[1];
+		volatile ushort dup[1];
+	} bits[1];
+	uint value[1];
+  };
+  volatile uint waiters[1];
 } WOLock;
 
 #define PHID 0x1
@@ -134,16 +151,6 @@ typedef struct {
 #define MASK 0x3
 #define RINC 0x4
 
-//	lite weight mutex
-
-// exclusive is set for write access
-
-typedef struct {
-	volatile uint exclusive[1];
-} BtMutexLatch;
-
-#define XCL 1
-
 //	mode & definition for lite latch implementation
 
 enum {
@@ -167,12 +174,11 @@ typedef struct {
 	WOLock parent[1];		// Posting of fence key in parent
 	WOLock atomic[1];		// Atomic update in progress
 	uint split;				// right split page atomic insert
-	uint entry;				// entry slot in latch table
 	uint next;				// next entry in hash table chain
 	uint prev;				// prev entry in hash table chain
 	ushort pin;				// number of accessing threads
-	unsigned char dirty;	// page in cache is dirty (atomic set)
-	unsigned char avail;	// page is an available entry
+	unsigned char dirty;	// page in cache is dirty (atomic setable)
+	unsigned char promote;	// page in cache is dirty (atomic setable)
 	BtMutexLatch modify[1];	// modify entry lite latch
 } BtLatchSet;
 
@@ -203,8 +209,7 @@ typedef enum {
 	Unique,
 	Librarian,
 	Duplicate,
-	Delete,
-	Update
+	Delete
 } BtSlotType;
 
 typedef struct {
@@ -239,7 +244,7 @@ typedef struct {
 
 //	note that this structure size
 //	must be a multiple of 8 bytes
-//	in order to place dups correctly.
+//	in order to place PageZero correctly.
 
 typedef struct BtPage_ {
 	uint cnt;					// count of keys in page
@@ -267,9 +272,10 @@ typedef struct {
 //	structure for latch manager on ALLOC_page
 
 typedef struct {
-	struct BtPage_ alloc[1];	// next page_no in right ptr
-	unsigned long long dups[1];	// global duplicate key uniqueifier
-	unsigned char chain[BtId];	// head of free page_nos chain
+	struct BtPage_ alloc[1];		// next page_no in right ptr
+	unsigned char freechain[BtId];	// head of free page_nos chain
+	unsigned long long activepages;	// number of active pages
+	uint redopages;					// number of redo pages in file
 } BtPageZero;
 
 //	The object structure for Btree access
@@ -288,18 +294,16 @@ typedef struct {
 	unsigned char *pagepool;	// mapped to the buffer pool pages
 	unsigned char *redobuff;	// mapped recovery buffer pointer
 	logseqno lsn, flushlsn;		// current & first lsn flushed
-	BtMutexLatch dump[1];		// redo dump lite latch
 	BtMutexLatch redo[1];		// redo area lite latch
 	BtMutexLatch lock[1];		// allocation area lite latch
+	BtMutexLatch maps[1];		// mapping segments lite latch
 	ushort thread_no[1];		// next thread number
 	uint nlatchpage;			// number of latch pages at BT_latch
 	uint latchtotal;			// number of page latch entries
 	uint latchhash;				// number of latch hash table slots
 	uint latchvictim;			// next latch entry to examine
-	uint latchavail;			// next available latch entry
-	uint availlock[1];			// latch available commitments
-	uint available;				// number of available latches
-	uint redopages;				// size of recovery buff in pages
+	uint latchpromote;			// next latch entry to promote
+	uint redolast;				// last msync size of recovery buff
 	uint redoend;				// eof/end element in recovery buff
 	int err;					// last error
 	int line;					// last error line no
@@ -309,6 +313,8 @@ typedef struct {
 	HANDLE halloc;				// allocation handle
 	HANDLE hpool;				// buffer pool handle
 #endif
+	uint segments;				// number of memory mapped segments
+	unsigned char *pages[64000];// memory mapped segments of b-tree
 } BtMgr;
 
 typedef struct {
@@ -319,6 +325,15 @@ typedef struct {
 	unsigned char key[BT_keyarray];	// last found complete key
 } BtDb;
 
+//	atomic txn structures
+
+typedef struct {
+	logseqno reqlsn;	// redo log seq no required
+	uint entry;			// latch table entry number
+	uint slot:31;		// page slot number
+	uint reuse:1;		// reused previous page
+} AtomicTxn;
+
 //	Catastrophic errors
 
 typedef enum {
@@ -330,8 +345,7 @@ typedef enum {
 	BTERR_read,
 	BTERR_wrt,
 	BTERR_atomic,
-	BTERR_recovery,
-	BTERR_avail
+	BTERR_recovery
 } BTERR;
 
 #define CLOCK_bit 0x8000
@@ -367,7 +381,7 @@ extern BTERR bt_writepage (BtMgr *mgr, BtPage page, uid page_no);
 extern BTERR bt_readpage (BtMgr *mgr, BtPage page, uid page_no);
 extern void bt_lockpage(BtLock mode, BtLatchSet *latch, ushort thread_no);
 extern void bt_unlockpage(BtLock mode, BtLatchSet *latch);
-extern BTERR bt_insertkey (BtMgr *mgr, unsigned char *key, uint len, uint lvl, void *value, uint vallen, uint update, ushort thread_no);
+extern BTERR bt_insertkey (BtMgr *mgr, unsigned char *key, uint len, uint lvl, void *value, uint vallen, BtSlotType type, ushort thread_no);
 extern BTERR  bt_deletekey (BtMgr *mgr, unsigned char *key, uint len, uint lvl, ushort thread_no);
 
 extern int bt_findkey (BtDb *db, unsigned char *key, uint keylen, unsigned char *value, uint valmax);
@@ -383,6 +397,10 @@ extern void bt_mgrclose (BtMgr *mgr);
 extern logseqno bt_newredo (BtMgr *mgr, BTRM type, int lvl, BtKey *key, BtVal *val, ushort thread_no);
 extern logseqno bt_txnredo (BtMgr *mgr, BtPage page, ushort thread_no);
 
+//	atomic transaction functions
+BTERR bt_atomicexec(BtMgr *mgr, BtPage source, logseqno lsn, int lsm, ushort thread_no);
+BTERR bt_txnpromote (BtDb *bt);
+
 //  The page is allocated from low and hi ends.
 //  The key slots are allocated from the bottom,
 //	while the text and value of the key
@@ -452,56 +470,137 @@ int i;
 	return id;
 }
 
-uid bt_newdup (BtMgr *mgr)
+//	lite weight spin lock Latch Manager
+
+int sys_futex(void *addr1, int op, int val1, struct timespec *timeout, void *addr2, int val3)
+{
+	return syscall(SYS_futex, addr1, op, val1, timeout, addr2, val3);
+}
+
+void bt_mutexlock(BtMutexLatch *latch)
 {
-#ifdef unix
-	return __sync_fetch_and_add (mgr->pagezero->dups, 1) + 1;
-#else
-	return _InterlockedIncrement64(mgr->pagezero->dups, 1);
-#endif
+BtMutexLatch prev[1];
+uint slept = 0;
+
+  while( 1 ) {
+	*prev->value = __sync_fetch_and_or(latch->value, XCL);
+
+	if( !*prev->bits->xlock ) {			// did we set XCL?
+	    if( slept )
+		  __sync_fetch_and_sub(latch->value, WRT);
+		return;
+	}
+
+	if( !slept ) {
+		*prev->bits->wrt += 1;
+		__sync_fetch_and_add(latch->value, WRT);
+	}
+
+	sys_futex (latch->value, FUTEX_WAIT_BITSET_PRIVATE, *prev->value, NULL, NULL, QueWr);
+	slept = 1;
+  }
+}
+
+//	try to obtain write lock
+
+//	return 1 if obtained,
+//		0 otherwise
+
+int bt_mutextry(BtMutexLatch *latch)
+{
+BtMutexLatch prev[1];
+
+	*prev->value = __sync_fetch_and_or(latch->value, XCL);
+
+	//	take write access if exclusive bit was clear
+
+	return !*prev->bits->xlock;
+}
+
+//	clear write mode
+
+void bt_releasemutex(BtMutexLatch *latch)
+{
+BtMutexLatch prev[1];
+
+	*prev->value = __sync_fetch_and_and(latch->value, ~XCL);
+
+	if( *prev->bits->wrt )
+	  sys_futex( latch->value, FUTEX_WAKE_BITSET_PRIVATE, 1, NULL, NULL, QueWr );
 }
 
-//	Write-Only Queue Lock
+//	Write-Only Reentrant Lock
 
 void WriteOLock (WOLock *lock, ushort tid)
 {
-uint prev;
+uint prev, waited = 0;
+
+  while( 1 ) {
+	bt_mutexlock(lock->xcl);
 
-	if( lock->tid == tid ) {
-		lock->dup++;
+	if( waited )
+		*lock->waiters -= 1;
+
+	if( *lock->bits->tid == tid ) {
+		*lock->bits->dup += 1;
+		bt_releasemutex(lock->xcl);
 		return;
 	}
-
-	while( 1 ) {
-#ifdef unix
-	  prev = __sync_fetch_and_or (lock->exclusive, 1);
-#else
-	  prev = _InterlockedExchangeOr (lock->exclusive, 1);
-#endif
-	  if( !(prev & XCL) ) {
-		lock->tid = tid;
+	if( !*lock->bits->tid ) {
+		*lock->bits->tid = tid;
+		bt_releasemutex(lock->xcl);
 		return;
-	  }
-#ifdef unix
-	  sys_futex( (void *)lock->exclusive, FUTEX_WAIT_BITSET, prev, NULL, NULL, QueWr );
-#else
-	  SwitchToThread ();
-#endif
 	}
+
+	waited = 1;
+	*lock->waiters += 1;
+	prev = *lock->value;
+
+	bt_releasemutex(lock->xcl);
+	
+	sys_futex( lock->value, FUTEX_WAIT_BITSET_PRIVATE, prev, NULL, NULL, QueWr );
+  }
 }
 
 void WriteORelease (WOLock *lock)
 {
-	if( lock->dup ) {
-		lock->dup--;
-		return;
+  bt_mutexlock(lock->xcl);
+
+  if( *lock->bits->dup ) {
+	*lock->bits->dup -= 1;
+    bt_releasemutex(lock->xcl);
+	return;
+  }
+
+  *lock->bits->tid = 0;
+
+  if( *lock->waiters )
+	sys_futex( lock->value, FUTEX_WAKE_BITSET_PRIVATE, 32768, NULL, NULL, QueWr );
+  bt_releasemutex(lock->xcl);
+}
+
+//	clear lock of holders and waiters
+
+ClearWOLock (WOLock *lock)
+{
+  while( 1 ) {
+	bt_mutexlock(lock->xcl);
+
+	if( *lock->waiters ) {
+		bt_releasemutex(lock->xcl);
+		sched_yield();
+		continue;
 	}
 
-	*lock->exclusive = 0;
-	lock->tid = 0;
-#ifdef linux
-	sys_futex( (void *)lock->exclusive, FUTEX_WAKE_BITSET, 1, NULL, NULL, QueWr );
-#endif
+	if( *lock->bits->tid ) {
+		bt_releasemutex(lock->xcl);
+		sched_yield();
+		continue;
+	}
+
+	bt_releasemutex(lock->xcl);
+	return;
+  }
 }
 
 //	Phase-Fair reader/writer lock implementation
@@ -510,10 +609,6 @@ void WriteLock (RWLock *lock, ushort tid)
 {
 ushort w, r, tix;
 
-	if( lock->tid == tid ) {
-		lock->dup++;
-		return;
-	}
 #ifdef unix
 	tix = __sync_fetch_and_add (lock->ticket, 1);
 #else
@@ -540,17 +635,10 @@ ushort w, r, tix;
 #else
 		SwitchToThread();
 #endif
-	lock->tid = tid;
 }
 
 void WriteRelease (RWLock *lock)
 {
-	if( lock->dup ) {
-		lock->dup--;
-		return;
-	}
-
-	lock->tid = 0;
 #ifdef unix
 	__sync_fetch_and_and (lock->rin, ~MASK);
 #else
@@ -566,12 +654,6 @@ int ReadTry (RWLock *lock, ushort tid)
 {
 ushort w;
 
-	//  OK if write lock already held by same thread
-
-	if( lock->tid == tid ) {
-		lock->dup++;
-		return 1;
-	}
 #ifdef unix
 	w = __sync_fetch_and_add (lock->rin, RINC) & MASK;
 #else
@@ -593,10 +675,7 @@ ushort w;
 void ReadLock (RWLock *lock, ushort tid)
 {
 ushort w;
-	if( lock->tid == tid ) {
-		lock->dup++;
-		return;
-	}
+
 #ifdef unix
 	w = __sync_fetch_and_add (lock->rin, RINC) & MASK;
 #else
@@ -613,11 +692,6 @@ ushort w;
 
 void ReadRelease (RWLock *lock)
 {
-	if( lock->dup ) {
-		lock->dup--;
-		return;
-	}
-
 #ifdef unix
 	__sync_fetch_and_add (lock->rout, RINC);
 #else
@@ -625,74 +699,18 @@ void ReadRelease (RWLock *lock)
 #endif
 }
 
-//	lite weight spin lock Latch Manager
-
-int sys_futex(void *addr1, int op, int val1, struct timespec *timeout, void *addr2, int val3)
-{
-	return syscall(SYS_futex, addr1, op, val1, timeout, addr2, val3);
-}
-
-void bt_mutexlock(BtMutexLatch *latch)
-{
-uint prev;
-
-  while( 1 ) {
-#ifdef  unix
-	prev = __sync_fetch_and_or(latch->exclusive, XCL);
-#else
-	prev = _InterlockedOr(latch->exclusive, XCL);
-#endif
-	if( !(prev & XCL) )
-		return;
-#ifdef  unix
-	sys_futex( (void *)latch->exclusive, FUTEX_WAIT_BITSET, prev, NULL, NULL, QueWr );
-#else
-	SwitchToThread();
-#endif
-  }
-}
-
-//	try to obtain write lock
-
-//	return 1 if obtained,
-//		0 otherwise
-
-int bt_mutextry(BtMutexLatch *latch)
-{
-uint prev;
-
-#ifdef  unix
-	prev = __sync_fetch_and_or(latch->exclusive, XCL);
-#else
-	prev = _InterlockedOr(latch->exclusive, XCL);
-#endif
-	//	take write access if exclusive bit is clear
-
-	return !(prev & XCL);
-}
-
-//	clear write mode
-
-void bt_releasemutex(BtMutexLatch *latch)
-{
-	*latch->exclusive = 0;
-#ifdef unix
-	sys_futex( (void *)latch->exclusive, FUTEX_WAKE_BITSET, 1, NULL, NULL, QueWr );
-#endif
-}
-
 //	recovery manager -- flush dirty pages
 
 void bt_flushlsn (BtMgr *mgr, ushort thread_no)
 {
 uint cnt3 = 0, cnt2 = 0, cnt = 0;
+uint entry, segment;
 BtLatchSet *latch;
 BtPage page;
-uint entry;
 
   //	flush dirty pool pages to the btree
 
-fprintf(stderr, "Start flushlsn\n");
+fprintf(stderr, "Start flushlsn  ");
   for( entry = 1; entry < mgr->latchtotal; entry++ ) {
 	page = (BtPage)(((uid)entry << mgr->page_bits) + mgr->pagepool);
 	latch = mgr->latchsets + entry;
@@ -703,14 +721,17 @@ fprintf(stderr, "Start flushlsn\n");
 	  bt_writepage(mgr, page, latch->page_no);
 	  latch->dirty = 0, cnt++;
 	}
-if( latch->avail )
-cnt3++;
 if( latch->pin & ~CLOCK_bit )
 cnt2++;
 	bt_unlockpage(BtLockRead, latch);
 	bt_releasemutex (latch->modify);
   }
-fprintf(stderr, "End flushlsn %d pages  %d pinned  %d available\n", cnt, cnt2, cnt3);
+fprintf(stderr, "End flushlsn %d pages  %d pinned\n", cnt, cnt2);
+fprintf(stderr, "begin sync");
+	for( segment = 0; segment < mgr->segments; segment++ )
+	  if( msync (mgr->pages[segment], (uid)65536 << mgr->page_bits, MS_SYNC) < 0 )
+		fprintf(stderr, "msync error %d line %d\n", errno, __LINE__);
+fprintf(stderr, " end sync\n");
 }
 
 //	recovery manager -- process current recovery buff on startup
@@ -723,8 +744,6 @@ uint offset = 0;
 BtKey *key;
 BtVal *val;
 
-  pread (mgr->idx, mgr->redobuff, mgr->redopages << mgr->page_size, REDO_page << mgr->page_size);
-
   hdr = (BtLogHdr *)mgr->redobuff;
   mgr->flushlsn = hdr->lsn;
 
@@ -746,45 +765,15 @@ BtVal *val;
   }
 }
 
-//	recovery manager -- dump current recovery buff & flush dirty pages
-//		in preparation for next recovery buffer.
-
-BTERR bt_dumpredo (BtMgr *mgr)
-{
-BtLogHdr *eof;
-fprintf(stderr, "Flush pages ");
-
-	eof = (BtLogHdr *)(mgr->redobuff + mgr->redoend);
-	memset (eof, 0, sizeof(BtLogHdr));
-
-	//  flush pages written at beginning of this redo buffer
-	//	then write the redo buffer out to disk
-
-	fdatasync (mgr->idx);
-
-fprintf(stderr, "Dump ReDo: %d bytes\n", mgr->redoend);
-	pwrite (mgr->idx, mgr->redobuff, mgr->redoend + sizeof(BtLogHdr), REDO_page << mgr->page_bits);
-
-	sync_file_range (mgr->idx, REDO_page << mgr->page_bits, mgr->redoend + sizeof(BtLogHdr), SYNC_FILE_RANGE_WAIT_AFTER);
-
-	mgr->flushlsn = mgr->lsn;
-	mgr->redoend = 0;
-
-	eof = (BtLogHdr *)(mgr->redobuff);
-	memset (eof, 0, sizeof(BtLogHdr));
-	eof->lsn = mgr->lsn;
-	return 0;
-}
-
 //	recovery manager -- append new entry to recovery log
-//	flush to disk when it overflows.
+//	flush dirty pages to disk when it overflows.
 
 logseqno bt_newredo (BtMgr *mgr, BTRM type, int lvl, BtKey *key, BtVal *val, ushort thread_no)
 {
-uint size = mgr->page_size * mgr->redopages - sizeof(BtLogHdr);
+uint size = mgr->page_size * mgr->pagezero->redopages - sizeof(BtLogHdr);
 uint amt = sizeof(BtLogHdr);
 BtLogHdr *hdr, *eof;
-uint flush;
+uint last, end;
 
 	bt_mutexlock (mgr->redo);
 
@@ -794,11 +783,13 @@ uint flush;
 	//	see if new entry fits in buffer
 	//	flush and reset if it doesn't
 
-	if( flush = amt > size - mgr->redoend ) {
-	  bt_mutexlock (mgr->dump);
-
-	  if( bt_dumpredo (mgr) )
-		return 0;
+	if( amt > size - mgr->redoend ) {
+	  mgr->flushlsn = mgr->lsn;
+	  if( msync (mgr->redobuff + (mgr->redolast & ~0xfff), mgr->redoend - (mgr->redolast & ~0xfff) + sizeof(BtLogHdr), MS_SYNC) < 0 )
+		fprintf(stderr, "msync error %d line %d\n", errno, __LINE__);
+	  mgr->redolast = 0;
+	  mgr->redoend = 0;
+	  bt_flushlsn(mgr, thread_no);
 	}
 
 	//	fill in new entry & either eof or end block
@@ -822,26 +813,33 @@ uint flush;
 	  memcpy ((unsigned char *)(hdr + 1) + key->len + sizeof(BtKey), val, val->len + sizeof(BtVal));
 	}
 
-	bt_releasemutex(mgr->redo);
+	eof = (BtLogHdr *)(mgr->redobuff + mgr->redoend);
+	memset (eof, 0, sizeof(BtLogHdr));
+	eof->lsn = mgr->lsn;
 
-	if( flush ) {
-	  bt_flushlsn (mgr, thread_no);
-	  bt_releasemutex(mgr->dump);
-	}
+	last = mgr->redolast & ~0xfff;
+	end = mgr->redoend;
 
+	if( end - last + sizeof(BtLogHdr) >= 32768 )
+	  if( msync (mgr->redobuff + last, end - last + sizeof(BtLogHdr), MS_SYNC) < 0 )
+		fprintf(stderr, "msync error %d line %d\n", errno, __LINE__);
+	  else
+		mgr->redolast = end;
+
+	bt_releasemutex(mgr->redo);
 	return hdr->lsn;
 }
 
 //	recovery manager -- append transaction to recovery log
-//	flush to disk when it overflows.
+//	flush dirty pages to disk when it overflows.
 
 logseqno bt_txnredo (BtMgr *mgr, BtPage source, ushort thread_no)
 {
-uint size = mgr->page_size * mgr->redopages - sizeof(BtLogHdr);
+uint size = mgr->page_size * mgr->pagezero->redopages - sizeof(BtLogHdr);
 uint amt = 0, src, type;
 BtLogHdr *hdr, *eof;
+uint last, end;
 logseqno lsn;
-uint flush;
 BtKey *key;
 BtVal *val;
 
@@ -859,11 +857,13 @@ BtVal *val;
 	//	see if new entry fits in buffer
 	//	flush and reset if it doesn't
 
-	if( flush = amt > size - mgr->redoend ) {
-	  bt_mutexlock (mgr->dump);
-
-	  if( bt_dumpredo (mgr) )
-		return 0;
+	if( amt > size - mgr->redoend ) {
+	  mgr->flushlsn = mgr->lsn;
+	  if( msync (mgr->redobuff + (mgr->redolast & ~0xfff), mgr->redoend - (mgr->redolast & ~0xfff) + sizeof(BtLogHdr), MS_SYNC) < 0 )
+		fprintf(stderr, "msync error %d line %d\n", errno, __LINE__);
+	  mgr->redolast = 0;
+	  mgr->redoend = 0;
+	  bt_flushlsn (mgr, thread_no);
 	}
 
 	//	assign new lsn to transaction
@@ -886,9 +886,6 @@ BtVal *val;
 	  case Delete:
 		type = BTRM_del;
 		break;
-	  case Update:
-		type = BTRM_upd;
-		break;
 	  }
 
 	  amt = key->len + val->len + sizeof(BtKey) + sizeof(BtVal);
@@ -910,47 +907,70 @@ BtVal *val;
 
 	eof = (BtLogHdr *)(mgr->redobuff + mgr->redoend);
 	memset (eof, 0, sizeof(BtLogHdr));
+	eof->lsn = lsn;
 
-	bt_releasemutex(mgr->redo);
+	last = mgr->redolast & ~0xfff;
+	end = mgr->redoend;
 
-	if( flush ) {
-	  bt_flushlsn (mgr, thread_no);
-	  bt_releasemutex(mgr->dump);
-	}
+	if( end - last + sizeof(BtLogHdr) >= 32768 )
+	  if( msync (mgr->redobuff + last, end - last + sizeof(BtLogHdr), MS_SYNC) < 0 )
+		fprintf(stderr, "msync error %d line %d\n", errno, __LINE__);
+	  else
+		mgr->redolast = end;
 
+	bt_releasemutex(mgr->redo);
 	return lsn;
 }
 
+//	sync a single btree page to disk
+
+BTERR bt_syncpage (BtMgr *mgr, BtPage page, BtLatchSet *latch)
+{
+uint segment = latch->page_no >> 16;
+BtPage perm;
+
+	if( bt_writepage (mgr, page, latch->page_no) )
+		return mgr->err;
+
+	perm = (BtPage)(mgr->pages[segment] + ((latch->page_no & 0xffff) << mgr->page_bits));
+
+	if( msync (perm, mgr->page_size, MS_SYNC) < 0 )
+	  fprintf(stderr, "msync error %d line %d\n", errno, __LINE__);
+
+	latch->dirty = 0;
+	return 0;
+}
+
 //	read page into buffer pool from permanent location in Btree file
 
 BTERR bt_readpage (BtMgr *mgr, BtPage page, uid page_no)
 {
-off64_t off = page_no << mgr->page_bits;
+int flag = PROT_READ | PROT_WRITE;
+uint segment = page_no >> 16;
+BtPage perm;
 
-#ifdef unix
-	if( pread (mgr->idx, page, mgr->page_size, page_no << mgr->page_bits) < mgr->page_size ) {
-		fprintf (stderr, "Unable to read page %d errno = %d\n", page_no, errno);
-		return mgr->err = BTERR_read;
+  while( 1 ) {
+	if( segment < mgr->segments ) {
+	  perm = (BtPage)(mgr->pages[segment] + ((page_no & 0xffff) << mgr->page_bits));
+if( perm->page_no != page_no )
+abort();
+		memcpy (page, perm, mgr->page_size);
+		mgr->reads++;
+		return 0;
 	}
-#else
-OVERLAPPED ovl[1];
-uint amt[1];
 
-	memset (ovl, 0, sizeof(OVERLAPPED));
-	ovl->Offset = off;
-	ovl->OffsetHigh = off >> 32;
+	bt_mutexlock (mgr->maps);
 
-	if( !ReadFile(mgr->idx, page, mgr->page_size, amt, ovl)) {
-		fprintf (stderr, "Unable to read page %d GetLastError = %d\n", page_no, GetLastError());
-		return BTERR_read;
-	}
-	if( *amt <  mgr->page_size ) {
-		fprintf (stderr, "Unable to read page %.8x GetLastError = %d\n", page_no, GetLastError());
-		return BTERR_read;
+	if( segment < mgr->segments ) {
+		bt_releasemutex (mgr->maps);
+		continue;
 	}
-#endif
-	mgr->reads++;
-	return 0;
+
+	mgr->pages[mgr->segments] = mmap (0, (uid)65536 << mgr->page_bits, flag, MAP_SHARED, mgr->idx, mgr->segments << (mgr->page_bits + 16));
+	mgr->segments++;
+
+	bt_releasemutex (mgr->maps);
+  }
 }
 
 //	write page to permanent location in Btree file
@@ -958,29 +978,31 @@ uint amt[1];
 
 BTERR bt_writepage (BtMgr *mgr, BtPage page, uid page_no)
 {
-off64_t off = page_no << mgr->page_bits;
+int flag = PROT_READ | PROT_WRITE;
+uint segment = page_no >> 16;
+BtPage perm;
 
-	page->page_no = page_no;
-
-#ifdef unix
-	if( pwrite(mgr->idx, page, mgr->page_size, off) < mgr->page_size )
-		return BTERR_wrt;
-#else
-OVERLAPPED ovl[1];
-uint amt[1];
+  while( 1 ) {
+	if( segment < mgr->segments ) {
+	  perm = (BtPage)(mgr->pages[segment] + ((page_no & 0xffff) << mgr->page_bits));
+if( page_no > LEAF_page && perm->page_no != page_no)
+abort();
+		memcpy (perm, page, mgr->page_size);
+		mgr->writes++;
+		return 0;
+	}
 
-	memset (ovl, 0, sizeof(OVERLAPPED));
-	ovl->Offset = off;
-	ovl->OffsetHigh = off >> 32;
+	bt_mutexlock (mgr->maps);
 
-	if( !WriteFile(mgr->idx, page, mgr->page_size, amt, ovl) )
-		return BTERR_wrt;
+	if( segment < mgr->segments ) {
+		bt_releasemutex (mgr->maps);
+		continue;
+	}
 
-	if( *amt <  mgr->page_size )
-		return BTERR_wrt;
-#endif
-	mgr->writes++;
-	return 0;
+	mgr->pages[mgr->segments] = mmap (0, (uid)65536 << mgr->page_bits, flag, MAP_SHARED, mgr->idx, mgr->segments << (mgr->page_bits + 16));
+	bt_releasemutex (mgr->maps);
+	mgr->segments++;
+  }
 }
 
 //	set CLOCK bit in latch
@@ -992,30 +1014,15 @@ void bt_unpinlatch (BtMgr *mgr, BtLatchSet *latch)
 	latch->pin |= CLOCK_bit;
 	latch->pin--;
 
-	// if not doing redo recovery
-	//	set latch available
-
-	if( mgr->redopages )
-	  if( !(latch->pin & ~CLOCK_bit) ) {
-		latch->avail = 1;
-#ifdef unix
-		__sync_fetch_and_add (&mgr->available, 1);
-#else
-		_InterlockedIncrement(&mgr->available);
-#endif
-	}
-
 	bt_releasemutex(latch->modify);
 }
 
 //  return the btree cached page address
-//	if page is dirty and has not yet been
-//	flushed to disk for the current redo
-//	recovery buffer, write it out.
 
 BtPage bt_mappage (BtMgr *mgr, BtLatchSet *latch)
 {
-BtPage page = (BtPage)(((uid)latch->entry << mgr->page_bits) + mgr->pagepool);
+uid entry = latch - mgr->latchsets;
+BtPage page = (BtPage)((entry << mgr->page_bits) + mgr->pagepool);
 
   return page;
 }
@@ -1023,24 +1030,16 @@ BtPage page = (BtPage)(((uid)latch->entry << mgr->page_bits) + mgr->pagepool);
 //  return next available latch entry
 //	  and with latch entry locked
 
-uint bt_availnext (BtMgr *mgr, ushort thread_id)
+uint bt_availnext (BtMgr *mgr)
 {
 BtLatchSet *latch;
 uint entry;
 
   while( 1 ) {
-
-  //  flush dirty pages if none are available
-  //	and we aren't doing redo recovery
-
-  if( !mgr->redopages )
-	if( !mgr->available )
-	  bt_flushlsn (mgr, thread_id);
-
 #ifdef unix
-	entry = __sync_fetch_and_add (&mgr->latchavail, 1) + 1;
+	entry = __sync_fetch_and_add (&mgr->latchvictim, 1) + 1;
 #else
-	entry = _InterlockedIncrement (&mgr->latchavail);
+	entry = _InterlockedIncrement (&mgr->latchvictim);
 #endif
 	entry %= mgr->latchtotal;
 
@@ -1049,24 +1048,26 @@ uint entry;
 
 	latch = mgr->latchsets + entry;
 
-	if( !latch->avail )
+	if( !bt_mutextry(latch->modify) )
 		continue;
 
-	bt_mutexlock(latch->modify);
+	//  return this entry if it is not pinned
 
-	if( !latch->avail ) {
-		bt_releasemutex(latch->modify);
-		continue;
-	}
+	if( !latch->pin )
+		return entry;
+
+	//	if the CLOCK bit is set
+	//	reset it to zero.
 
-	return entry;
+	latch->pin &= ~CLOCK_bit;
+	bt_releasemutex(latch->modify);
   }
 }
 
-//	find available latchset
+//	pin page in buffer pool
 //	return with latchset pinned
 
-BtLatchSet *bt_pinlatch (BtMgr *mgr, uid page_no, BtPage loadit, ushort thread_id)
+BtLatchSet *bt_pinlatch (BtMgr *mgr, uid page_no, BtPage contents, ushort thread_id)
 {
 uint hashidx = page_no % mgr->latchhash;
 BtLatchSet *latch;
@@ -1085,34 +1086,25 @@ BtPage page;
   } while( entry = latch->next );
 
   //  found our entry: increment pin
-  //  remove from available status
 
   if( entry ) {
 	latch = mgr->latchsets + entry;
 	bt_mutexlock(latch->modify);
-	if( latch->avail )
-#ifdef unix
-	  __sync_fetch_and_add (&mgr->available, -1);
-#else
-	  _InterlockedDecrement(&mgr->available);
-#endif
-	latch->avail = 0;
 	latch->pin |= CLOCK_bit;
 	latch->pin++;
-
+if(contents)
+abort();
 	bt_releasemutex(latch->modify);
 	bt_releasemutex(mgr->hashtable[hashidx].latch);
 	return latch;
   }
 
-  //  find and reuse entry from available set
+  //  find and reuse unpinned entry
 
 trynext:
 
-  if( entry = bt_availnext (mgr, thread_id) )
-  	latch = mgr->latchsets + entry;
-  else
-	return mgr->line = __LINE__, mgr->err = BTERR_avail, NULL;
+  entry = bt_availnext (mgr);
+  latch = mgr->latchsets + entry;
 
   idx = latch->page_no % mgr->latchhash;
 
@@ -1140,28 +1132,22 @@ trynext:
     bt_releasemutex (mgr->hashtable[idx].latch);
    }
 
-  //  remove available status
-
-  latch->avail = 0;
-#ifdef unix
-  __sync_fetch_and_add (&mgr->available, -1);
-#else
-  _InterlockedDecrement(&mgr->available);
-#endif
   page = (BtPage)(((uid)entry << mgr->page_bits) + mgr->pagepool);
 
+  //  update permanent page area in btree from buffer pool
+  //	no read-lock is required since page is not pinned.
+
   if( latch->dirty )
 	if( mgr->err = bt_writepage (mgr, page, latch->page_no) )
 	  return mgr->line = __LINE__, NULL;
 	else
 	  latch->dirty = 0;
 
-  if( loadit ) {
-	memcpy (page, loadit, mgr->page_size);
+  if( contents ) {
+	memcpy (page, contents, mgr->page_size);
 	latch->dirty = 1;
-  } else
-	if( bt_readpage (mgr, page, page_no) )
-	  return mgr->line = __LINE__, NULL;
+  } else if( bt_readpage (mgr, page, page_no) )
+	return mgr->line = __LINE__, NULL;
 
   //  link page as head of hash table chain
   //  if this is a never before used entry,
@@ -1182,7 +1168,6 @@ trynext:
 
   latch->pin = CLOCK_bit | 1;
   latch->page_no = page_no;
-  latch->entry = entry;
   latch->split = 0;
 
   bt_releasemutex (latch->modify);
@@ -1206,8 +1191,6 @@ uint slot;
 	if( mgr->redoend ) {
 		eof = (BtLogHdr *)(mgr->redobuff + mgr->redoend);
 		memset (eof, 0, sizeof(BtLogHdr));
-
-		pwrite (mgr->idx, mgr->redobuff, mgr->redoend + sizeof(BtLogHdr), REDO_page << mgr->page_bits);
 	}
 
 	//	write remaining dirty pool pages to the btree
@@ -1222,24 +1205,22 @@ uint slot;
 		}
 	}
 
-	//	flush last batch to disk and clear
-	//	redo recovery buffer on disk.
+	//	clear redo recovery buffer on disk.
 
-	fdatasync (mgr->idx);
-
-	if( mgr->redopages ) {
+	if( mgr->pagezero->redopages ) {
 		eof = (BtLogHdr *)mgr->redobuff;
 		memset (eof, 0, sizeof(BtLogHdr));
 		eof->lsn = mgr->lsn;
-
-		pwrite (mgr->idx, mgr->redobuff, sizeof(BtLogHdr), REDO_page << mgr->page_bits);
-
-		sync_file_range (mgr->idx, REDO_page << mgr->page_bits, sizeof(BtLogHdr), SYNC_FILE_RANGE_WAIT_AFTER);
+		if( msync (mgr->redobuff, 4096, MS_SYNC) < 0 )
+		  fprintf(stderr, "msync error %d line %d\n", errno, __LINE__);
 	}
 
 	fprintf(stderr, "%d buffer pool pages flushed\n", num);
 
 #ifdef unix
+	while( mgr->segments )
+		munmap (mgr->pages[--mgr->segments], (uid)65536 << mgr->page_bits);
+
 	munmap (mgr->pagepool, (uid)mgr->nlatchpage << mgr->page_bits);
 	munmap (mgr->pagezero, mgr->page_size);
 #else
@@ -1250,7 +1231,6 @@ uint slot;
 	CloseHandle(mgr->hpool);
 #endif
 #ifdef unix
-	free (mgr->redobuff);
 	close (mgr->idx);
 	free (mgr);
 #else
@@ -1368,13 +1348,18 @@ BtVal *val;
 	// and page(s) of latches and page pool cache
 
 	memset (pagezero, 0, 1 << bits);
+	pagezero->alloc->lvl = MIN_lvl - 1;
 	pagezero->alloc->bits = mgr->page_bits;
-	bt_putid(pagezero->alloc->right, redopages + MIN_lvl+1);
+	pagezero->redopages = redopages;
+
+	bt_putid(pagezero->alloc->right, pagezero->redopages + MIN_lvl+1);
+	pagezero->activepages = 2;
 
 	//  initialize left-most LEAF page in
-	//	alloc->left.
+	//	alloc->left and count of active leaf pages.
 
 	bt_putid (pagezero->alloc->left, LEAF_page);
+	ftruncate (mgr->idx, (REDO_page + pagezero->redopages) << mgr->page_bits);
 
 	if( bt_writepage (mgr, pagezero->alloc, 0) ) {
 		fprintf (stderr, "Unable to create btree page zero\n");
@@ -1385,7 +1370,8 @@ BtVal *val;
 	pagezero->alloc->bits = mgr->page_bits;
 
 	for( lvl=MIN_lvl; lvl--; ) {
-		slotptr(pagezero->alloc, 1)->off = mgr->page_size - 3 - (lvl ? BtId + sizeof(BtVal): sizeof(BtVal));
+	BtSlot *node = slotptr(pagezero->alloc, 1);
+		node->off = mgr->page_size - 3 - (lvl ? BtId + sizeof(BtVal): sizeof(BtVal));
 		key = keyptr(pagezero->alloc, 1);
 		key->len = 2;		// create stopper key
 		key->key[0] = 0xff;
@@ -1396,13 +1382,14 @@ BtVal *val;
 		val->len = lvl ? BtId : 0;
 		memcpy (val->value, value, val->len);
 
-		pagezero->alloc->min = slotptr(pagezero->alloc, 1)->off;
+		pagezero->alloc->min = node->off;
 		pagezero->alloc->lvl = lvl;
 		pagezero->alloc->cnt = 1;
 		pagezero->alloc->act = 1;
+		pagezero->alloc->page_no = MIN_lvl - lvl;
 
 		if( bt_writepage (mgr, pagezero->alloc, MIN_lvl - lvl) ) {
-			fprintf (stderr, "Unable to create btree page zero\n");
+			fprintf (stderr, "Unable to create btree page\n");
 			return bt_mgrclose (mgr), NULL;
 		}
 	}
@@ -1414,23 +1401,28 @@ mgrlatch:
 	VirtualFree (pagezero, 0, MEM_RELEASE);
 #endif
 #ifdef unix
-	// mlock the pagezero page
+	// mlock the first segment of 64K pages
 
 	flag = PROT_READ | PROT_WRITE;
-	mgr->pagezero = mmap (0, mgr->page_size, flag, MAP_SHARED, mgr->idx, ALLOC_page << mgr->page_bits);
-	if( mgr->pagezero == MAP_FAILED ) {
-		fprintf (stderr, "Unable to mmap btree page zero, error = %d\n", errno);
+	mgr->pages[0] = mmap (0, (uid)65536 << mgr->page_bits, flag, MAP_SHARED, mgr->idx, 0);
+	mgr->segments = 1;
+
+	if( mgr->pages[0] == MAP_FAILED ) {
+		fprintf (stderr, "Unable to mmap first btree segment, error = %d\n", errno);
 		return bt_mgrclose (mgr), NULL;
 	}
+
+	mgr->pagezero = (BtPageZero *)mgr->pages[0];
 	mlock (mgr->pagezero, mgr->page_size);
 
+	mgr->redobuff = mgr->pages[0] + REDO_page * mgr->page_size;
+	mlock (mgr->redobuff, mgr->pagezero->redopages << mgr->page_bits);
+
 	mgr->pagepool = mmap (0, (uid)mgr->nlatchpage << mgr->page_bits, flag, MAP_ANONYMOUS | MAP_SHARED, -1, 0);
 	if( mgr->pagepool == MAP_FAILED ) {
 		fprintf (stderr, "Unable to mmap anonymous buffer pool pages, error = %d\n", errno);
 		return bt_mgrclose (mgr), NULL;
 	}
-	if( mgr->redopages = redopages )
-		mgr->redobuff = valloc (redopages * mgr->page_size);
 #else
 	flag = PAGE_READWRITE;
 	mgr->halloc = CreateFileMapping(mgr->idx, NULL, flag, 0, mgr->page_size, NULL);
@@ -1460,22 +1452,12 @@ mgrlatch:
 		fprintf (stderr, "Unable to map buffer pool, error = %d\n", GetLastError());
 		return bt_mgrclose (mgr), NULL;
 	}
-	if( mgr->redopages = redopages )
-		mgr->redobuff = VirtualAlloc (NULL, redopages * mgr->page_size | MEM_COMMIT, PAGE_READWRITE);
 #endif
 
 	mgr->latchsets = (BtLatchSet *)(mgr->pagepool + ((uid)mgr->latchtotal << mgr->page_bits));
 	mgr->hashtable = (BtHashEntry *)(mgr->latchsets + mgr->latchtotal);
 	mgr->latchhash = (mgr->pagepool + ((uid)mgr->nlatchpage << mgr->page_bits) - (unsigned char *)mgr->hashtable) / sizeof(BtHashEntry);
 
-	//	mark all pool entries as available
-
-	for( idx = 1; idx < mgr->latchtotal; idx++ ) {
-		latch = mgr->latchsets + idx;
-		latch->avail = 1;
-		mgr->available++;
-	}
-
 	return mgr;
 }
 
@@ -1551,6 +1533,10 @@ void bt_lockpage(BtLock mode, BtLatchSet *latch, ushort thread_no)
 		WriteOLock (latch->atomic, thread_no);
 		ReadLock (latch->readwr, thread_no);
 		break;
+	case BtLockAtomic | BtLockWrite:
+		WriteOLock (latch->atomic, thread_no);
+		WriteLock (latch->readwr, thread_no);
+		break;
 	}
 }
 
@@ -1581,6 +1567,10 @@ void bt_unlockpage(BtLock mode, BtLatchSet *latch)
 		WriteORelease (latch->atomic);
 		ReadRelease (latch->readwr);
 		break;
+	case BtLockAtomic | BtLockWrite:
+		WriteORelease (latch->atomic);
+		WriteRelease (latch->readwr);
+		break;
 	}
 }
 
@@ -1597,19 +1587,32 @@ int blk;
 	bt_mutexlock(mgr->lock);
 
 	// use empty chain first
-	// else allocate empty page
+	// else allocate new page
 
-	if( page_no = bt_getid(mgr->pagezero->chain) ) {
+	if( page_no = bt_getid(mgr->pagezero->freechain) ) {
 		if( set->latch = bt_pinlatch (mgr, page_no, NULL, thread_id) )
 			set->page = bt_mappage (mgr, set->latch);
 		else
-			return mgr->err = BTERR_struct, mgr->line = __LINE__, -1;
+			return mgr->line = __LINE__, mgr->err = BTERR_struct;
 
-		bt_putid(mgr->pagezero->chain, bt_getid(set->page->right));
-		bt_releasemutex(mgr->lock);
+		mgr->pagezero->activepages++;
+		bt_putid(mgr->pagezero->freechain, bt_getid(set->page->right));
 
-		memcpy (set->page, contents, mgr->page_size);
+		//  the page is currently free and this
+		//  will keep bt_txnpromote out.
+
+		//	contents will replace this bit
+		//  and pin will keep bt_txnpromote out
+
+		contents->page_no = page_no;
 		set->latch->dirty = 1;
+
+		memcpy (set->page, contents, mgr->page_size);
+
+		if( msync (mgr->pagezero, mgr->page_size, MS_SYNC) < 0 )
+		  fprintf(stderr, "msync error %d line %d\n", errno, __LINE__);
+
+		bt_releasemutex(mgr->lock);
 		return 0;
 	}
 
@@ -1619,16 +1622,27 @@ int blk;
 	// unlock allocation latch and
 	//	extend file into new page.
 
+	mgr->pagezero->activepages++;
+	if( msync (mgr->pagezero, mgr->page_size, MS_SYNC) < 0 )
+	  fprintf(stderr, "msync error %d line %d\n", errno, __LINE__);
 	bt_releasemutex(mgr->lock);
 
-	//	don't load cache from btree page
+	//	keep bt_txnpromote out of this page
+
+	contents->free = 1;
+	contents->page_no = page_no;
+	pwrite (mgr->idx, contents, mgr->page_size, page_no << mgr->page_bits);
+
+	//	don't load cache from btree page, load it from contents
 
 	if( set->latch = bt_pinlatch (mgr, page_no, contents, thread_id) )
 		set->page = bt_mappage (mgr, set->latch);
 	else
 		return mgr->err;
 
-	set->latch->dirty = 1;
+	// now pin will keep bt_txnpromote out
+
+	set->page->free = 0;
 	return 0;
 }
 
@@ -1670,10 +1684,11 @@ uint good = 0;
 
 int bt_loadpage (BtMgr *mgr, BtPageSet *set, unsigned char *key, uint len, uint lvl, BtLock lock, ushort thread_no)
 {
-uid page_no = ROOT_page, prevpage = 0;
+uid page_no = ROOT_page, prevpage_no = 0;
 uint drill = 0xff, slot;
 BtLatchSet *prevlatch;
 uint mode, prevmode;
+BtPage prevpage;
 BtVal *val;
 
   //  start at root of btree and drill down
@@ -1691,13 +1706,15 @@ BtVal *val;
 	  bt_lockpage(BtLockAccess, set->latch, thread_no);
 
 	set->page = bt_mappage (mgr, set->latch);
+if( set->latch->promote )
+abort();
 
 	//	release & unpin parent or left sibling page
 
-	if( prevpage ) {
+	if( prevpage_no ) {
 	  bt_unlockpage(prevmode, prevlatch);
 	  bt_unpinlatch (mgr, prevlatch);
-	  prevpage = 0;
+	  prevpage_no = 0;
 	}
 
  	// obtain mode lock using lock chaining through AccessLock
@@ -1725,8 +1742,9 @@ BtVal *val;
 		}
 	}
 
-	prevpage = set->latch->page_no;
+	prevpage_no = set->latch->page_no;
 	prevlatch = set->latch;
+	prevpage = set->page;
 	prevmode = mode;
 
 	//  find key on page at this level
@@ -1769,6 +1787,7 @@ BtVal *val;
 
 //	return page to free list
 //	page must be delete & write locked
+//	and have no keys pointing to it.
 
 void bt_freepage (BtMgr *mgr, BtPageSet *set)
 {
@@ -1778,19 +1797,25 @@ void bt_freepage (BtMgr *mgr, BtPageSet *set)
 
 	//	store chain
 
-	memcpy(set->page->right, mgr->pagezero->chain, BtId);
-	bt_putid(mgr->pagezero->chain, set->latch->page_no);
+	memcpy(set->page->right, mgr->pagezero->freechain, BtId);
+	bt_putid(mgr->pagezero->freechain, set->latch->page_no);
+	set->latch->promote = 0;
 	set->latch->dirty = 1;
 	set->page->free = 1;
 
+	// decrement active page count
+
+	mgr->pagezero->activepages--;
+
+	if( msync (mgr->pagezero, mgr->page_size, MS_SYNC) < 0 )
+	  fprintf(stderr, "msync error %d line %d\n", errno, __LINE__);
+
 	// unlock released page
+	// and unlock allocation page
 
 	bt_unlockpage (BtLockDelete, set->latch);
 	bt_unlockpage (BtLockWrite, set->latch);
 	bt_unpinlatch (mgr, set->latch);
-
-	// unlock allocation page
-
 	bt_releasemutex (mgr->lock);
 }
 
@@ -1824,7 +1849,7 @@ uint idx;
 	bt_putid (value, set->latch->page_no);
 	ptr = (BtKey*)leftkey;
 
-	if( bt_insertkey (mgr, ptr->key, ptr->len, lvl+1, value, BtId, 1, thread_no) )
+	if( bt_insertkey (mgr, ptr->key, ptr->len, lvl+1, value, BtId, Unique, thread_no) )
 	  return mgr->err;
 
 	//	now delete old fence key
@@ -1885,9 +1910,11 @@ uint idx;
 
 //  delete a page and manage keys
 //  call with page writelocked
-//	returns with page unpinned
 
-BTERR bt_deletepage (BtMgr *mgr, BtPageSet *set, ushort thread_no)
+//	returns with page removed
+//	from the page pool.
+
+BTERR bt_deletepage (BtMgr *mgr, BtPageSet *set, ushort thread_no, int delkey)
 {
 unsigned char lowerfence[BT_keyarray], higherfence[BT_keyarray];
 unsigned char value[BtId];
@@ -1897,7 +1924,7 @@ uid page_no;
 BtKey *ptr;
 
 	//	cache copy of fence key
-	//	to post in parent
+	//	to remove in parent
 
 	ptr = keyptr(set->page, set->page->cnt);
 	memcpy (lowerfence, ptr, ptr->len + sizeof(BtKey));
@@ -1924,6 +1951,7 @@ BtKey *ptr;
 	// pull contents of right peer into our empty page
 
 	memcpy (set->page, right->page, mgr->page_size);
+	set->page->page_no = set->latch->page_no;
 	set->latch->dirty = 1;
 
 	// mark right page deleted and point it to left page
@@ -1945,14 +1973,15 @@ BtKey *ptr;
 	bt_putid (value, set->latch->page_no);
 	ptr = (BtKey*)higherfence;
 
-	if( bt_insertkey (mgr, ptr->key, ptr->len, lvl+1, value, BtId, 1, thread_no) )
+	if( bt_insertkey (mgr, ptr->key, ptr->len, lvl+1, value, BtId, Unique, thread_no) )
 	  return mgr->err;
 
 	//	delete old lower key to our node
 
 	ptr = (BtKey*)lowerfence;
 
-	if( bt_deletekey (mgr, ptr->key, ptr->len, lvl+1, thread_no) )
+	if( delkey )
+	 if( bt_deletekey (mgr, ptr->key, ptr->len, lvl+1, thread_no) )
 	  return mgr->err;
 
 	//	obtain delete and write locks to right node
@@ -1974,33 +2003,43 @@ BTERR bt_deletekey (BtMgr *mgr, unsigned char *key, uint len, uint lvl, ushort t
 {
 uint slot, idx, found, fence;
 BtPageSet set[1];
+BtSlot *node;
 BtKey *ptr;
 BtVal *val;
 
-	if( slot = bt_loadpage (mgr, set, key, len, lvl, BtLockWrite, thread_no) )
+	if( slot = bt_loadpage (mgr, set, key, len, lvl, BtLockWrite, thread_no) ) {
+		node = slotptr(set->page, slot);
 		ptr = keyptr(set->page, slot);
-	else
+	} else
 		return mgr->err;
 
 	// if librarian slot, advance to real slot
 
-	if( slotptr(set->page, slot)->type == Librarian )
+	if( node->type == Librarian ) {
 		ptr = keyptr(set->page, ++slot);
+		node = slotptr(set->page, slot);
+	}
 
 	fence = slot == set->page->cnt;
 
-	// if key is found delete it, otherwise ignore request
+	// delete the key, ignore request if already dead
 
 	if( found = !keycmp (ptr, key, len) )
-	  if( found = slotptr(set->page, slot)->dead == 0 ) {
+	  if( found = node->dead == 0 ) {
 		val = valptr(set->page,slot);
-		slotptr(set->page, slot)->dead = 1;
  		set->page->garbage += ptr->len + val->len + sizeof(BtKey) + sizeof(BtVal);
  		set->page->act--;
 
+		//  mark node type as delete
+
+		node->type = Delete;
+		node->dead = 1;
+
 		// collapse empty slots beneath the fence
+		// on interiour nodes
 
-		while( idx = set->page->cnt - 1 )
+		if( lvl )
+		 while( idx = set->page->cnt - 1 )
 		  if( slotptr(set->page, idx)->dead ) {
 			*slotptr(set->page, idx) = *slotptr(set->page, idx + 1);
 			memset (slotptr(set->page, set->page->cnt--), 0, sizeof(BtSlot));
@@ -2008,9 +2047,12 @@ BtVal *val;
 			break;
 	  }
 
+	if( !found )
+		return 0;
+
 	//	did we delete a fence key in an upper level?
 
-	if( found && lvl && set->page->act && fence )
+	if( lvl && set->page->act && fence )
 	  if( bt_fixfence (mgr, set, lvl, thread_no) )
 		return mgr->err;
 	  else
@@ -2018,7 +2060,7 @@ BtVal *val;
 
 	//	do we need to collapse root?
 
-	if( lvl > 1 && set->latch->page_no == ROOT_page && set->page->act == 1 )
+	if( set->latch->page_no == ROOT_page && set->page->act == 1 )
 	  if( bt_collapseroot (mgr, set, thread_no) )
 		return mgr->err;
 	  else
@@ -2027,7 +2069,7 @@ BtVal *val;
 	//	delete empty page
 
  	if( !set->page->act )
-		return bt_deletepage (mgr, set, thread_no);
+	  return bt_deletepage (mgr, set, thread_no, 1);
 
 	set->latch->dirty = 1;
 	bt_unlockpage(BtLockWrite, set->latch);
@@ -2221,9 +2263,13 @@ uint nxt = mgr->page_size;
 unsigned char value[BtId];
 BtPageSet left[1];
 uid left_page_no;
+BtPage frame;
 BtKey *ptr;
 BtVal *val;
 
+	frame = malloc (mgr->page_size);
+	memcpy (frame, root->page, mgr->page_size);
+
 	//	save left page fence key for new root
 
 	ptr = keyptr(root->page, root->page->cnt);
@@ -2232,11 +2278,12 @@ BtVal *val;
 	//  Obtain an empty page to use, and copy the current
 	//  root contents into it, e.g. lower keys
 
-	if( bt_newpage(mgr, left, root->page, page_no) )
+	if( bt_newpage(mgr, left, frame, page_no) )
 		return mgr->err;
 
 	left_page_no = left->latch->page_no;
 	bt_unpinlatch (mgr, left->latch);
+	free (frame);
 
 	// preserve the page info at the bottom
 	// of higher keys and set rest to zero
@@ -2278,6 +2325,8 @@ BtVal *val;
 	root->page->act = 2;
 	root->page->lvl++;
 
+	mgr->pagezero->alloc->lvl = root->page->lvl;
+
 	// release and unpin root pages
 
 	bt_unlockpage(BtLockWrite, root->latch);
@@ -2290,7 +2339,7 @@ BtVal *val;
 //  split already locked full node
 //	leave it locked.
 //	return pool entry for new right
-//	page, unlocked
+//	page, pinned & unlocked
 
 uint bt_splitpage (BtMgr *mgr, BtPageSet *set, ushort thread_no)
 {
@@ -2399,13 +2448,14 @@ uint prev;
 	bt_putid(set->page->right, right->latch->page_no);
 	set->page->min = nxt;
 	set->page->cnt = idx;
+	free(frame);
 
-	return right->latch->entry;
+	return right->latch - mgr->latchsets;
 }
 
 //	fix keys for newly split page
-//	call with page locked, return
-//	unlocked
+//	call with both pages pinned & locked
+//  return unlocked and unpinned
 
 BTERR bt_splitkeys (BtMgr *mgr, BtPageSet *set, BtLatchSet *right, ushort thread_no)
 {
@@ -2440,7 +2490,7 @@ BtKey *ptr;
 	bt_putid (value, set->latch->page_no);
 	ptr = (BtKey *)leftkey;
 
-	if( bt_insertkey (mgr, ptr->key, ptr->len, lvl+1, value, BtId, 1, thread_no) )
+	if( bt_insertkey (mgr, ptr->key, ptr->len, lvl+1, value, BtId, Unique, thread_no) )
 		return mgr->err;
 
 	// switch fence for right block of larger keys to new right page
@@ -2448,7 +2498,7 @@ BtKey *ptr;
 	bt_putid (value, right->page_no);
 	ptr = (BtKey *)rightkey;
 
-	if( bt_insertkey (mgr, ptr->key, ptr->len, lvl+1, value, BtId, 1, thread_no) )
+	if( bt_insertkey (mgr, ptr->key, ptr->len, lvl+1, value, BtId, Unique, thread_no) )
 		return mgr->err;
 
 	bt_unlockpage (BtLockParent, set->latch);
@@ -2469,6 +2519,7 @@ uint idx, librarian;
 BtSlot *node;
 BtKey *ptr;
 BtVal *val;
+int rate;
 
 	//	if found slot > desired slot and previous slot
 	//	is a librarian slot, use it
@@ -2497,27 +2548,52 @@ BtVal *val;
 	  if( slotptr(set->page, idx)->dead )
 		break;
 
-	// now insert key into array before slot
+	// now insert key into array before slot,
+	//	adding as many librarian slots as
+	//	makes sense.
 
-	if( idx == set->page->cnt )
-		idx += 2, set->page->cnt += 2, librarian = 2;
-	else
-		librarian = 1;
+	if( idx == set->page->cnt ) {
+	int avail = 4 * set->page->min / 5 - sizeof(*set->page) - ++set->page->cnt * sizeof(BtSlot);
 
-	set->latch->dirty = 1;
-	set->page->act++;
+		librarian = ++idx - slot;
+		avail /= sizeof(BtSlot);
 
-	while( idx > slot + librarian - 1 )
-		*slotptr(set->page, idx) = *slotptr(set->page, idx - librarian), idx--;
+		if( avail < 0 )
+			avail = 0;
 
-	//	add librarian slot
+		if( librarian > avail )
+			librarian = avail;
 
-	if( librarian > 1 ) {
-		node = slotptr(set->page, slot++);
-		node->off = set->page->min;
-		node->type = Librarian;
-		node->dead = 1;
+		if( librarian ) {
+			rate = (idx - slot) / librarian;
+			set->page->cnt += librarian;
+			idx += librarian;
+		} else
+			rate = 0;
+	} else
+		librarian = 0, rate = 0;
+
+	while( idx > slot ) {
+		//  transfer slot
+		*slotptr(set->page, idx) = *slotptr(set->page, idx-librarian-1);
+		idx--;
+
+		//	add librarian slot per rate
+
+		if( librarian )
+		 if( (idx - slot + 1)/2 <= librarian * rate ) {
+//		  if( rate && !(idx % rate) ) {
+			node = slotptr(set->page, idx--);
+			node->off = node[1].off;
+			node->type = Librarian;
+			node->dead = 1;
+			librarian--;
+		  }
 	}
+if(librarian)
+abort();
+	set->latch->dirty = 1;
+	set->page->act++;
 
 	//	fill in new slot
 
@@ -2537,37 +2613,19 @@ BtVal *val;
 //  Insert new key into the btree at given level.
 //	either add a new key or update/add an existing one
 
-BTERR bt_insertkey (BtMgr *mgr, unsigned char *key, uint keylen, uint lvl, void *value, uint vallen, uint unique, ushort thread_no)
+BTERR bt_insertkey (BtMgr *mgr, unsigned char *key, uint keylen, uint lvl, void *value, uint vallen, BtSlotType type, ushort thread_no)
 {
-unsigned char newkey[BT_keyarray];
 uint slot, idx, len, entry;
 BtPageSet set[1];
-BtKey *ptr, *ins;
-uid sequence;
+BtSlot *node;
+BtKey *ptr;
 BtVal *val;
-uint type;
-
-  // set up the key we're working on
-
-  ins = (BtKey*)newkey;
-  memcpy (ins->key, key, keylen);
-  ins->len = keylen;
-
-  // is this a non-unique index value?
-
-  if( unique )
-	type = Unique;
-  else {
-	type = Duplicate;
-	sequence = bt_newdup (mgr);
-	bt_putid (ins->key + ins->len + sizeof(BtKey), sequence);
-	ins->len += BtId;
-  }
 
   while( 1 ) { // find the page and slot for the current key
-	if( slot = bt_loadpage (mgr, set, ins->key, ins->len, lvl, BtLockWrite, thread_no) )
+	if( slot = bt_loadpage (mgr, set, key, keylen, lvl, BtLockWrite, thread_no) ) {
+		node = slotptr(set->page, slot);
 		ptr = keyptr(set->page, slot);
-	else {
+	} else {
 		if( !mgr->err )
 			mgr->line = __LINE__, mgr->err = BTERR_ovflw;
 		return mgr->err;
@@ -2575,496 +2633,228 @@ uint type;
 
 	// if librarian slot == found slot, advance to real slot
 
-	if( slotptr(set->page, slot)->type == Librarian )
-	  if( !keycmp (ptr, key, keylen) )
+	if( node->type == Librarian )
+	  if( !keycmp (ptr, key, keylen) ) {
 		ptr = keyptr(set->page, ++slot);
+		node = slotptr(set->page, slot);
+	  }
 
-	len = ptr->len;
-
-	if( slotptr(set->page, slot)->type == Duplicate )
-		len -= BtId;
-
-	//  if inserting a duplicate key or unique key
+	//  if inserting a duplicate key or unique
+	//	key that doesn't exist on the page,
 	//	check for adequate space on the page
 	//	and insert the new key before slot.
 
-	if( unique && (len != ins->len || memcmp (ptr->key, ins->key, ins->len)) || !unique ) {
-	  if( !(slot = bt_cleanpage (mgr, set, ins->len, slot, vallen)) )
-		if( !(entry = bt_splitpage (mgr, set, thread_no)) )
-		  return mgr->err;
-		else if( bt_splitkeys (mgr, set, mgr->latchsets + entry, thread_no) )
-		  return mgr->err;
-		else
-		  continue;
-
-	  return bt_insertslot (mgr, set, slot, ins->key, ins->len, value, vallen, type, 1);
-	}
+	switch( type ) {
+	case Unique:
+	case Duplicate:
+	  if( keycmp (ptr, key, keylen) )
+	   if( slot = bt_cleanpage (mgr, set, keylen, slot, vallen) )
+	    return bt_insertslot (mgr, set, slot, key, keylen, value, vallen, type, 1);
+	   else if( !(entry = bt_splitpage (mgr, set, thread_no)) )
+		return mgr->err;
+	   else if( bt_splitkeys (mgr, set, mgr->latchsets + entry, thread_no) )
+		return mgr->err;
+	   else
+		continue;
 
-	// if key already exists, update value and return
+	  // if key already exists, update value and return
 
-	val = valptr(set->page, slot);
+	  val = valptr(set->page, slot);
 
-	if( val->len >= vallen ) {
+	  if( val->len >= vallen ) {
 		if( slotptr(set->page, slot)->dead )
 			set->page->act++;
+		node->type = type;
+		node->dead = 0;
+
 		set->page->garbage += val->len - vallen;
 		set->latch->dirty = 1;
-		slotptr(set->page, slot)->dead = 0;
 		val->len = vallen;
 		memcpy (val->value, value, vallen);
 		bt_unlockpage(BtLockWrite, set->latch);
 		bt_unpinlatch (mgr, set->latch);
 		return 0;
-	}
-
-	//	new update value doesn't fit in existing value area
-
-	if( !slotptr(set->page, slot)->dead )
-		set->page->garbage += val->len + ptr->len + sizeof(BtKey) + sizeof(BtVal);
-	else {
-		slotptr(set->page, slot)->dead = 0;
-		set->page->act++;
-	}
-
-	if( !(slot = bt_cleanpage (mgr, set, keylen, slot, vallen)) )
-	  if( !(entry = bt_splitpage (mgr, set, thread_no)) )
-		return mgr->err;
-	  else if( bt_splitkeys (mgr, set, mgr->latchsets + entry, thread_no) )
-		return mgr->err;
-	  else
-		continue;
-
-	set->page->min -= vallen + sizeof(BtVal);
-	val = (BtVal*)((unsigned char *)set->page + set->page->min);
-	memcpy (val->value, value, vallen);
-	val->len = vallen;
-
-	set->latch->dirty = 1;
-	set->page->min -= keylen + sizeof(BtKey);
-	ptr = (BtKey*)((unsigned char *)set->page + set->page->min);
-	memcpy (ptr->key, key, keylen);
-	ptr->len = keylen;
-	
-	slotptr(set->page, slot)->off = set->page->min;
-	bt_unlockpage(BtLockWrite, set->latch);
-	bt_unpinlatch (mgr, set->latch);
-	return 0;
-  }
-  return 0;
-}
-
-typedef struct {
-	logseqno reqlsn;	// redo log seq no required
-	logseqno lsn;		// redo log sequence number
-	uint entry;			// latch table entry number
-	uint slot:31;		// page slot number
-	uint reuse:1;		// reused previous page
-} AtomicTxn;
-
-typedef struct {
-	uid page_no;		// page number for split leaf
-	void *next;			// next key to insert
-	uint entry:29;		// latch table entry number
-	uint type:2;		// 0 == insert, 1 == delete, 2 == free
-	uint nounlock:1;	// don't unlock ParentModification
-	unsigned char leafkey[BT_keyarray];
-} AtomicKey;
-
-//	determine actual page where key is located
-//  return slot number
-
-uint bt_atomicpage (BtMgr *mgr, BtPage source, AtomicTxn *locks, uint src, BtPageSet *set)
-{
-BtKey *key = keyptr(source,src);
-uint slot = locks[src].slot;
-uint entry;
-
-	if( src > 1 && locks[src].reuse )
-	  entry = locks[src-1].entry, slot = 0;
-	else
-	  entry = locks[src].entry;
-
-	if( slot ) {
-		set->latch = mgr->latchsets + entry;
-		set->page = bt_mappage (mgr, set->latch);
-		return slot;
-	}
-
-	//	is locks->reuse set? or was slot zeroed?
-	//	if so, find where our key is located 
-	//	on current page or pages split on
-	//	same page txn operations.
-
-	do {
-		set->latch = mgr->latchsets + entry;
-		set->page = bt_mappage (mgr, set->latch);
-
-		if( slot = bt_findslot(set->page, key->key, key->len) ) {
-		  if( slotptr(set->page, slot)->type == Librarian )
-			slot++;
-		  if( locks[src].reuse )
-			locks[src].entry = entry;
-		  return slot;
-		}
-	} while( entry = set->latch->split );
-
-	mgr->line = __LINE__, mgr->err = BTERR_atomic;
-	return 0;
-}
-
-BTERR bt_atomicinsert (BtMgr *mgr, BtPage source, AtomicTxn *locks, uint src, ushort thread_no)
-{
-BtKey *key = keyptr(source, src);
-BtVal *val = valptr(source, src);
-BtLatchSet *latch;
-BtPageSet set[1];
-uint entry, slot;
-
-  while( slot = bt_atomicpage (mgr, source, locks, src, set) ) {
-	if( slot = bt_cleanpage(mgr, set, key->len, slot, val->len) ) {
-	  if( bt_insertslot (mgr, set, slot, key->key, key->len, val->value, val->len, slotptr(source,src)->type, 0) )
-		return mgr->err;
-	  set->page->lsn = locks[src].lsn;
-	  return 0;
-	}
-
-	if( entry = bt_splitpage (mgr, set, thread_no) )
-	  latch = mgr->latchsets + entry;
-	else
-	  return mgr->err;
-
-	//	splice right page into split chain
-	//	and WriteLock it.
-
-	bt_lockpage(BtLockWrite, latch, thread_no);
-	latch->split = set->latch->split;
-	set->latch->split = entry;
-	locks[src].slot = 0;
-  }
-
-  return mgr->line = __LINE__, mgr->err = BTERR_atomic;
-}
-
-BTERR bt_atomicdelete (BtMgr *mgr, BtPage source, AtomicTxn *locks, uint src, ushort thread_no)
-{
-BtKey *key = keyptr(source, src);
-BtPageSet set[1];
-uint idx, slot;
-BtKey *ptr;
-BtVal *val;
-
-	if( slot = bt_atomicpage (mgr, source, locks, src, set) )
-	  ptr = keyptr(set->page, slot);
-	else
-	  return mgr->line = __LINE__, mgr->err = BTERR_struct;
-
-	if( !keycmp (ptr, key->key, key->len) )
-	  if( !slotptr(set->page, slot)->dead )
-		slotptr(set->page, slot)->dead = 1;
-	  else
-		return 0;
-	else
-		return 0;
-
-	val = valptr(set->page, slot);
-	set->page->garbage += ptr->len + val->len + sizeof(BtKey) + sizeof(BtVal);
-	set->latch->dirty = 1;
-	set->page->lsn = locks[src].lsn;
- 	set->page->act--;
-	mgr->found++;
-	return 0;
-}
-
-//	delete an empty master page for a transaction
-
-//	note that the far right page never empties because
-//	it always contains (at least) the infinite stopper key
-//	and that all pages that don't contain any keys are
-//	deleted, or are being held under Atomic lock.
-
-BTERR bt_atomicfree (BtMgr *mgr, BtPageSet *prev, ushort thread_no)
-{
-BtPageSet right[1], temp[1];
-unsigned char value[BtId];
-uid right_page_no;
-BtKey *ptr;
-
-	bt_lockpage(BtLockWrite, prev->latch, thread_no);
-
-	//	grab the right sibling
-
-	if( right->latch = bt_pinlatch(mgr, bt_getid (prev->page->right), NULL, thread_no) )
-		right->page = bt_mappage (mgr, right->latch);
-	else
-		return mgr->err;
-
-	bt_lockpage(BtLockAtomic, right->latch, thread_no);
-	bt_lockpage(BtLockWrite, right->latch, thread_no);
-
-	//	and pull contents over empty page
-	//	while preserving master's left link
-
-	memcpy (right->page->left, prev->page->left, BtId);
-	memcpy (prev->page, right->page, mgr->page_size);
-
-	//	forward seekers to old right sibling
-	//	to new page location in set
-
-	bt_putid (right->page->right, prev->latch->page_no);
-	right->latch->dirty = 1;
-	right->page->kill = 1;
-
-	//	remove pointer to right page for searchers by
-	//	changing right fence key to point to the master page
-
-	ptr = keyptr(right->page,right->page->cnt);
-	bt_putid (value, prev->latch->page_no);
-
-	if( bt_insertkey (mgr, ptr->key, ptr->len, 1, value, BtId, 1, thread_no) )
-		return mgr->err;
-
-	//  now that master page is in good shape we can
-	//	remove its locks.
-
-	bt_unlockpage (BtLockAtomic, prev->latch);
-	bt_unlockpage (BtLockWrite, prev->latch);
-
-	//  fix master's right sibling's left pointer
-	//	to remove scanner's poiner to the right page
-
-	if( right_page_no = bt_getid (prev->page->right) ) {
-	  if( temp->latch = bt_pinlatch (mgr, right_page_no, NULL, thread_no) )
-		temp->page = bt_mappage (mgr, temp->latch);
-
-	  bt_lockpage (BtLockWrite, temp->latch, thread_no);
-	  bt_putid (temp->page->left, prev->latch->page_no);
-	  temp->latch->dirty = 1;
-
-	  bt_unlockpage (BtLockWrite, temp->latch);
-	  bt_unpinlatch (mgr, temp->latch);
-	} else {	// master is now the far right page
-	  bt_mutexlock (mgr->lock);
-	  bt_putid (mgr->pagezero->alloc->left, prev->latch->page_no);
-	  bt_releasemutex(mgr->lock);
-	}
-
-	//	now that there are no pointers to the right page
-	//	we can delete it after the last read access occurs
-
-	bt_unlockpage (BtLockWrite, right->latch);
-	bt_unlockpage (BtLockAtomic, right->latch);
-	bt_lockpage (BtLockDelete, right->latch, thread_no);
-	bt_lockpage (BtLockWrite, right->latch, thread_no);
-	bt_freepage (mgr, right);
-	return 0;
-}
-
-//	find and add the next available latch entry
-//	to the queue
-
-BTERR bt_txnavaillatch (BtDb *bt)
-{
-BtLatchSet *latch;
-uint startattempt;
-uint cnt, entry;
-uint hashidx;
-BtPage page;
-
-  //  find and reuse previous entry on victim
-
-  startattempt = bt->mgr->latchvictim;
-
-  while( 1 ) {
-#ifdef unix
-	entry = __sync_fetch_and_add(&bt->mgr->latchvictim, 1);
-#else
-	entry = _InterlockedIncrement (&bt->mgr->latchvictim) - 1;
-#endif
-	//	skip entry if it has outstanding pins
-
-	entry %= bt->mgr->latchtotal;
-
-	if( !entry )
-		continue;
-
-	//	only go around one time before
-	//	flushing redo recovery buffer,
-	//	and the buffer pool to free up entries.
-
-	if( bt->mgr->redopages )
-	  if( bt->mgr->latchvictim - startattempt > bt->mgr->latchtotal ) {
-	    if( bt_mutextry (bt->mgr->dump) ) {
-		 if( bt_dumpredo (bt->mgr) )
-		  return bt->mgr->err;
-		 bt_flushlsn (bt->mgr, bt->thread_no);
-		// synchronize the various threads running into this condition
-		//	so that only one thread does the dump and flush
-		} else
-  			bt_mutexlock(bt->mgr->dump);
-
-  		startattempt = bt->mgr->latchvictim;
-		bt_releasemutex(bt->mgr->dump);
-	  }
-
-	latch = bt->mgr->latchsets + entry;
-
-	if( latch->avail )
-	  continue;
-
-	bt_mutexlock(latch->modify);
-
-	//	skip if already an available entry
-
-	if( latch->avail ) {
-	  bt_releasemutex(latch->modify);
-	  continue;
-	}
-
-	//  skip this entry if it is pinned
-	//	if the CLOCK bit is set
-	//	reset it to zero.
+	  }
 
-	if( latch->pin ) {
-	  latch->pin &= ~CLOCK_bit;
-	  bt_releasemutex(latch->modify);
-	  continue;
-	}
+	  //  new update value doesn't fit in existing value area
+	  //	make sure page has room
 
-	page = (BtPage)(((uid)entry << bt->mgr->page_bits) + bt->mgr->pagepool);
+	  if( !node->dead )
+		set->page->garbage += val->len + ptr->len + sizeof(BtKey) + sizeof(BtVal);
+	  else
+		set->page->act++;
 
-	//	if dirty page has lsn >= last redo recovery buffer
-	//	then hold page in the buffer pool until next redo
-	//	recovery buffer is being written to disk.
+	  node->type = type;
+	  node->dead = 0;
 
-	if( latch->dirty )
-	  if( page->lsn >= bt->mgr->flushlsn ) {
-	    bt_releasemutex(latch->modify);
+	  if( !(slot = bt_cleanpage (mgr, set, keylen, slot, vallen)) )
+	   if( !(entry = bt_splitpage (mgr, set, thread_no)) )
+		return mgr->err;
+	   else if( bt_splitkeys (mgr, set, mgr->latchsets + entry, thread_no) )
+		return mgr->err;
+	   else
 		continue;
-	  }
 
-	//	entry is now available
-#ifdef unix
-	__sync_fetch_and_add (&bt->mgr->available, 1);
-#else
-	_InterlockedIncrement(&bt->mgr->available);
-#endif
-	latch->avail = 1;
-	bt_releasemutex(latch->modify);
-	return 0;
-  }
-}
+	  //  copy key and value onto page and update slot
 
-//	release available latch requests
+	  set->page->min -= vallen + sizeof(BtVal);
+	  val = (BtVal*)((unsigned char *)set->page + set->page->min);
+	  memcpy (val->value, value, vallen);
+	  val->len = vallen;
 
-void bt_txnavailrelease (BtDb *bt, uint count)
-{
-#ifdef unix
-	__sync_fetch_and_add(bt->mgr->availlock, -count);
-#else
-	_InterlockedAdd(bt->mgr->availlock, -count);
-#endif
+	  set->latch->dirty = 1;
+	  set->page->min -= keylen + sizeof(BtKey);
+	  ptr = (BtKey*)((unsigned char *)set->page + set->page->min);
+	  memcpy (ptr->key, key, keylen);
+	  ptr->len = keylen;
+	
+	  node->off = set->page->min;
+	  bt_unlockpage(BtLockWrite, set->latch);
+	  bt_unpinlatch (mgr, set->latch);
+	  return 0;
+    }
+  }
+  return 0;
 }
 
-//	promote page of keys from first btree
-//  into main btree
+//	determine actual page where key is located
+//  return slot number
 
-BTERR bt_txnavailmain (BtDb *bt)
+uint bt_atomicpage (BtMgr *mgr, BtPage source, AtomicTxn *locks, uint src, BtPageSet *set)
 {
-BtLatchSet *latch;
+BtKey *key = keyptr(source,src);
+uint slot = locks[src].slot;
 uint entry;
 
-  while( 1 ) {
-#ifdef unix
-	entry = __sync_fetch_and_add(&bt->mgr->latchvictim, 1);
-#else
-	entry = _InterlockedIncrement (&bt->mgr->latchvictim) - 1;
-#endif
-	//	skip entry if it has outstanding pins
+	if( src > 1 && locks[src].reuse )
+	  entry = locks[src-1].entry, slot = 0;
+	else
+	  entry = locks[src].entry;
 
-	entry %= bt->mgr->latchtotal;
+	if( slot ) {
+		set->latch = mgr->latchsets + entry;
+		set->page = bt_mappage (mgr, set->latch);
+		return slot;
+	}
 
-	if( !entry )
-		continue;
+	//	is locks->reuse set? or was slot zeroed?
+	//	if so, find where our key is located 
+	//	on current page or pages split on
+	//	same page txn operations.
 
-	latch = bt->mgr->latchsets + entry;
+	do {
+		set->latch = mgr->latchsets + entry;
+		set->page = bt_mappage (mgr, set->latch);
 
-	if( latch->avail )
-	  continue;
+		if( slot = bt_findslot(set->page, key->key, key->len) ) {
+		  if( slotptr(set->page, slot)->type == Librarian )
+			slot++;
+		  if( locks[src].reuse )
+			locks[src].entry = entry;
+		  return slot;
+		}
+	} while( entry = set->latch->split );
 
-	bt_mutexlock(latch->modify);
+	mgr->line = __LINE__, mgr->err = BTERR_atomic;
+	return 0;
+}
 
-	//	skip if already an available entry
+BTERR bt_atomicinsert (BtMgr *mgr, BtPage source, AtomicTxn *locks, uint src, ushort thread_no, logseqno lsn)
+{
+BtKey *key = keyptr(source, src);
+BtVal *val = valptr(source, src);
+BtLatchSet *latch;
+BtPageSet set[1];
+uint entry, slot;
 
-	if( latch->avail ) {
-	  bt_releasemutex(latch->modify);
-	  continue;
+  while( slot = bt_atomicpage (mgr, source, locks, src, set) ) {
+	if( slot = bt_cleanpage(mgr, set, key->len, slot, val->len) ) {
+	  if( bt_insertslot (mgr, set, slot, key->key, key->len, val->value, val->len, slotptr(source,src)->type, 0) )
+		return mgr->err;
+	  set->page->lsn = lsn;
+	  return 0;
 	}
 
-	//  skip this entry if it is pinned
-	//	if the CLOCK bit is set
-	//	reset it to zero.
+	//  split page
 
-	if( latch->pin ) {
-	  latch->pin &= ~CLOCK_bit;
-	  bt_releasemutex(latch->modify);
-	  continue;
-	}
+	if( entry = bt_splitpage (mgr, set, thread_no) )
+	  latch = mgr->latchsets + entry;
+	else
+	  return mgr->err;
+
+	//	splice right page into split chain
+	//	and WriteLock it
 
+	bt_lockpage(BtLockWrite, latch, thread_no);
+	latch->split = set->latch->split;
+	set->latch->split = entry;
+	locks[src].slot = 0;
   }
+
+  return mgr->line = __LINE__, mgr->err = BTERR_atomic;
 }
 
-//	commit available pool entries
-//	find available entries as required
+//	perform delete from smaller btree
+//  insert a delete slot if not found there
 
-BTERR bt_txnavailrequest (BtDb *bt, uint count)
+BTERR bt_atomicdelete (BtMgr *mgr, BtPage source, AtomicTxn *locks, uint src, ushort thread_no, logseqno lsn)
 {
-#ifdef unix
-	__sync_fetch_and_add(bt->mgr->availlock, count);
-#else
-	_InterlockedAdd(bt->mgr->availlock, count);
-#endif
+BtKey *key = keyptr(source, src);
+BtPageSet set[1];
+uint idx, slot;
+BtSlot *node;
+BtKey *ptr;
+BtVal *val;
 
-	//	find another available pool entry
+	if( slot = bt_atomicpage (mgr, source, locks, src, set) ) {
+	  node = slotptr(set->page, slot);
+	  ptr = keyptr(set->page, slot);
+	  val = valptr(set->page, slot);
+	} else
+	  return mgr->line = __LINE__, mgr->err = BTERR_struct;
 
-	while( *bt->mgr->availlock > bt->mgr->available )
-	  if( bt->mgr->redopages )
-		bt_txnavaillatch (bt);
-	  else
-		if( bt_txnavailmain (bt) )
-			return bt->mgr->err;
+	//  if slot is not found, insert a delete slot
+
+	if( keycmp (ptr, key->key, key->len) )
+	  return bt_insertslot (mgr, set, slot, key->key, key->len, NULL, 0, Delete, 0);
+
+	//	if node is already dead,
+	//	ignore the request.
+
+	if( node->dead )
+		return 0;
+
+	set->page->garbage += ptr->len + val->len + sizeof(BtKey) + sizeof(BtVal);
+	set->latch->dirty = 1;
+	set->page->lsn = lsn;
+ 	set->page->act--;
+
+	node->dead = 0;
+	__sync_fetch_and_add(&mgr->found, 1);
+	return 0;
 }
 
-//	atomic modification of a batch of keys.
+int qsortcmp (BtSlot *slot1, BtSlot *slot2, BtPage page)
+{
+BtKey *key1 = (BtKey *)((char *)page + slot1->off);
+BtKey *key2 = (BtKey *)((char *)page + slot2->off);
 
-//	return -1 if BTERR is set
-//	otherwise return slot number
-//	causing the key constraint violation
-//	or zero on successful completion.
+	return keycmp (key1, key2->key, key2->len);
+}
+//	atomic modification of a batch of keys.
 
-int bt_atomictxn (BtDb *bt, BtPage source)
+BTERR bt_atomictxn (BtDb *bt, BtPage source)
 {
-uint src, idx, slot, samepage, entry, avail, que = 0;
-AtomicKey *head, *tail, *leaf;
-BtPageSet set[1], prev[1];
-unsigned char value[BtId];
+uint src, idx, slot, samepage, entry, que = 0;
 BtKey *key, *ptr, *key2;
-BtLatchSet *latch;
-AtomicTxn *locks;
 int result = 0;
 BtSlot temp[1];
 logseqno lsn;
-BtPage page;
-BtVal *val;
-uid right;
 int type;
 
-  locks = calloc (source->cnt + 1, sizeof(AtomicTxn));
-  head = NULL;
-  tail = NULL;
-
   // stable sort the list of keys into order to
   //	prevent deadlocks between threads.
-
+/*
   for( src = 1; src++ < source->cnt; ) {
 	*temp = *slotptr(source,src);
 	key = keyptr (source,src);
@@ -3078,14 +2868,48 @@ int type;
 		break;
 	}
   }
+*/
+	qsort_r (slotptr(source,1), source->cnt, sizeof(BtSlot), (__compar_d_fn_t)qsortcmp, source);
+  //  add entries to redo log
 
-  // reserve enough buffer pool entries
+  if( bt->mgr->pagezero->redopages )
+	lsn = bt_txnredo (bt->mgr, source, bt->thread_no);
+  else
+	lsn = 0;
 
-  avail = source->cnt * 3 + bt->mgr->pagezero->alloc->lvl + 1;
+  //  perform the individual actions in the transaction
 
-  if( bt_txnavailrequest (bt, avail) )
+  if( bt_atomicexec (bt->mgr, source, lsn, 0, bt->thread_no) )
 	return bt->mgr->err;
 
+  // if number of active pages
+  // is greater than the buffer pool
+  // promote page into larger btree
+
+  if( bt->main )
+   while( bt->mgr->pagezero->activepages > bt->mgr->latchtotal - 10 )
+	if( bt_txnpromote (bt) )
+	  return bt->mgr->err;
+
+  // return success
+
+  return 0;
+}
+
+BTERR bt_atomicexec(BtMgr *mgr, BtPage source, logseqno lsn, int lsm, ushort thread_no)
+{
+uint src, idx, slot, samepage, entry, que = 0;
+BtPageSet set[1], prev[1];
+unsigned char value[BtId];
+BtLatchSet *latch;
+AtomicTxn *locks;
+BtKey *key, *ptr;
+BtPage page;
+BtVal *val;
+uid right;
+
+  locks = calloc (source->cnt + 1, sizeof(AtomicTxn));
+
   // Load the leaf page for each key
   // group same page references with reuse bit
   // and determine any constraint violations
@@ -3101,14 +2925,12 @@ int type;
 	if( samepage = src > 1 )
 	  if( samepage = !bt_getid(set->page->right) || keycmp (keyptr(set->page, set->page->cnt), key->key, key->len) >= 0 )
 		slot = bt_findslot(set->page, key->key, key->len);
-	  else
-	 	bt_unlockpage(BtLockRead, set->latch); 
 
 	if( !slot )
-	  if( slot = bt_loadpage(bt->mgr, set, key->key, key->len, 0, BtLockRead | BtLockAtomic, bt->thread_no) )
+	  if( slot = bt_loadpage(mgr, set, key->key, key->len, 0, BtLockAtomic, thread_no) )
 		set->latch->split = 0;
 	  else
-		goto atomicerr;
+  	 	return mgr->err;
 
 	if( slotptr(set->page, slot)->type == Librarian )
 	  ptr = keyptr(set->page, ++slot);
@@ -3116,7 +2938,7 @@ int type;
 	  ptr = keyptr(set->page, slot);
 
 	if( !samepage ) {
-	  locks[src].entry = set->latch->entry;
+	  locks[src].entry = set->latch - mgr->latchsets;
 	  locks[src].slot = slot;
 	  locks[src].reuse = 0;
 	} else {
@@ -3128,66 +2950,20 @@ int type;
 	//	capture current lsn for master page
 
 	locks[src].reqlsn = set->page->lsn;
-
-	//	perform constraint checks
-
-	switch( slotptr(source, src)->type ) {
-	case Duplicate:
-	case Unique:
-	  if( !slotptr(set->page, slot)->dead )
-	   if( slot < set->page->cnt || bt_getid (set->page->right) )
-	    if( !keycmp (ptr, key->key, key->len) ) {
-
-		  // return constraint violation if key already exists
-
-		  bt_unlockpage(BtLockRead, set->latch);
-		  result = src;
-
-		  while( src ) {
-			if( locks[src].entry ) {
-			  set->latch = bt->mgr->latchsets + locks[src].entry;
-			  bt_unlockpage(BtLockAtomic, set->latch);
-			  bt_unpinlatch (bt->mgr, set->latch);
-			}
-			src--;
-		  }
-		  free (locks);
-		  return result;
-		}
-	  break;
-	}
   }
 
-  //  unlock last loadpage lock
-
-  if( source->cnt )
-	bt_unlockpage(BtLockRead, set->latch);
-
-  //  and add entries to redo log
-
-  if( bt->mgr->redopages )
-   if( lsn = bt_txnredo (bt->mgr, source, bt->thread_no) )
-    for( src = 0; src++ < source->cnt; )
-	 locks[src].lsn = lsn;
-    else
-	 goto atomicerr;
-
   //  obtain write lock for each master page
 
   for( src = 0; src++ < source->cnt; ) {
 	if( locks[src].reuse )
 	  continue;
-	else
-	  bt_lockpage(BtLockWrite, bt->mgr->latchsets + locks[src].entry, bt->thread_no);
+
+	set->latch = mgr->latchsets + locks[src].entry;
+	bt_lockpage (BtLockWrite, set->latch, thread_no);
   }
 
   // insert or delete each key
   // process any splits or merges
-  // release Write & Atomic latches
-  // set ParentModifications and build
-  // queue of keys to insert for split pages
-  // or delete for deleted pages.
-
   // run through txn list backwards
 
   samepage = source->cnt + 1;
@@ -3203,232 +2979,247 @@ int type;
 	for( idx = src; idx < samepage; idx++ )
 	 switch( slotptr(source,idx)->type ) {
 	 case Delete:
-	  if( bt_atomicdelete (bt->mgr, source, locks, idx, bt->thread_no) )
-		goto atomicerr;
+	  if( bt_atomicdelete (mgr, source, locks, idx, thread_no, lsn) )
+  	 	return mgr->err;
 	  break;
 
 	case Duplicate:
 	case Unique:
-	  if( bt_atomicinsert (bt->mgr, source, locks, idx, bt->thread_no) )
-		goto atomicerr;
+	  if( bt_atomicinsert (mgr, source, locks, idx, thread_no, lsn) )
+  	 	return mgr->err;
+	  break;
+
+	default:
+  	  bt_atomicpage (mgr, source, locks, idx, set);
 	  break;
 	}
 
 	//	after the same page operations have finished,
 	//  process master page for splits or deletion.
 
-	latch = prev->latch = bt->mgr->latchsets + locks[src].entry;
-	prev->page = bt_mappage (bt->mgr, prev->latch);
+	latch = prev->latch = mgr->latchsets + locks[src].entry;
+	prev->page = bt_mappage (mgr, prev->latch);
 	samepage = src;
 
 	//  pick-up all splits from master page
-	//	each one is already WriteLocked.
+	//	each one is already pinned & WriteLocked.
 
-	entry = prev->latch->split;
-
-	while( entry ) {
-	  set->latch = bt->mgr->latchsets + entry;
-	  set->page = bt_mappage (bt->mgr, set->latch);
-	  entry = set->latch->split;
+	if( entry = prev->latch->split ) do {
+	  set->latch = mgr->latchsets + entry;
+	  set->page = bt_mappage (mgr, set->latch);
 
 	  // delete empty master page by undoing its split
 	  //  (this is potentially another empty page)
-	  //  note that there are no new left pointers yet
+	  //  note that there are no pointers to it yet
 
 	  if( !prev->page->act ) {
 		memcpy (set->page->left, prev->page->left, BtId);
-		memcpy (prev->page, set->page, bt->mgr->page_size);
-		bt_lockpage (BtLockDelete, set->latch, bt->thread_no);
-		bt_freepage (bt->mgr, set);
-
+		memcpy (prev->page, set->page, mgr->page_size);
+		bt_lockpage (BtLockDelete, set->latch, thread_no);
 		prev->latch->split = set->latch->split;
 		prev->latch->dirty = 1;
+		bt_freepage (mgr, set);
 		continue;
 	  }
 
-	  // remove empty page from the split chain
-	  // and return it to the free list.
+	  // remove empty split page from the split chain
+	  // and return it to the free list. No other
+	  // thread has its page number yet.
 
 	  if( !set->page->act ) {
 		memcpy (prev->page->right, set->page->right, BtId);
 		prev->latch->split = set->latch->split;
-		bt_lockpage (BtLockDelete, set->latch, bt->thread_no);
-		bt_freepage (bt->mgr, set);
+
+		bt_lockpage (BtLockDelete, set->latch, thread_no);
+		bt_freepage (mgr, set);
 		continue;
 	  }
 
-	  //  schedule prev fence key update
+	  //  update prev's fence key
 
 	  ptr = keyptr(prev->page,prev->page->cnt);
-	  leaf = calloc (sizeof(AtomicKey), 1), que++;
-
-	  memcpy (leaf->leafkey, ptr, ptr->len + sizeof(BtKey));
-	  leaf->page_no = prev->latch->page_no;
-	  leaf->entry = prev->latch->entry;
-	  leaf->type = 0;
-
-	  if( tail )
-		tail->next = leaf;
-	  else
-		head = leaf;
+	  bt_putid (value, prev->latch->page_no);
 
-	  tail = leaf;
+	  if( bt_insertkey (mgr, ptr->key, ptr->len, 1, value, BtId, Unique, thread_no) )
+		return mgr->err;
 
 	  // splice in the left link into the split page
 
 	  bt_putid (set->page->left, prev->latch->page_no);
-	  bt_lockpage(BtLockParent, prev->latch, bt->thread_no);
+
+	  if( lsm )
+		bt_syncpage (mgr, prev->page, prev->latch);
+
+	  // page is unpinned below to avoid bt_txnpromote
+
 	  bt_unlockpage(BtLockWrite, prev->latch);
 	  *prev = *set;
-	}
+	} while( entry = prev->latch->split );
 
 	//  update left pointer in next right page from last split page
-	//	(if all splits were reversed, latch->split == 0)
+	//	(if all splits were reversed or none occurred, latch->split == 0)
 
 	if( latch->split ) {
 	  //  fix left pointer in master's original (now split)
 	  //  far right sibling or set rightmost page in page zero
 
 	  if( right = bt_getid (prev->page->right) ) {
-		if( set->latch = bt_pinlatch (bt->mgr, right, NULL, bt->thread_no) )
-	  	  set->page = bt_mappage (bt->mgr, set->latch);
+		if( set->latch = bt_pinlatch (mgr, right, NULL, thread_no) )
+	  	  set->page = bt_mappage (mgr, set->latch);
 	 	else
-		  goto atomicerr;
+  	 	  return mgr->err;
 
-	    bt_lockpage (BtLockWrite, set->latch, bt->thread_no);
+	    bt_lockpage (BtLockWrite, set->latch, thread_no);
 	    bt_putid (set->page->left, prev->latch->page_no);
 		set->latch->dirty = 1;
+
 	    bt_unlockpage (BtLockWrite, set->latch);
-		bt_unpinlatch (bt->mgr, set->latch);
+		bt_unpinlatch (mgr, set->latch);
 	  } else {	// prev is rightmost page
-	    bt_mutexlock (bt->mgr->lock);
-		bt_putid (bt->mgr->pagezero->alloc->left, prev->latch->page_no);
-	    bt_releasemutex(bt->mgr->lock);
+	    bt_mutexlock (mgr->lock);
+		bt_putid (mgr->pagezero->alloc->left, prev->latch->page_no);
+	    bt_releasemutex(mgr->lock);
 	  }
 
 	  //  Process last page split in chain
+	  //  by switching the key from the master
+	  //  page to the last split.
 
 	  ptr = keyptr(prev->page,prev->page->cnt);
-	  leaf = calloc (sizeof(AtomicKey), 1), que++;
-
-	  memcpy (leaf->leafkey, ptr, ptr->len + sizeof(BtKey));
-	  leaf->page_no = prev->latch->page_no;
-	  leaf->entry = prev->latch->entry;
-	  leaf->type = 0;
-  
-	  if( tail )
-		tail->next = leaf;
-	  else
-		head = leaf;
+	  bt_putid (value, prev->latch->page_no);
 
-	  tail = leaf;
+	  if( bt_insertkey (mgr, ptr->key, ptr->len, 1, value, BtId, Unique, thread_no) )
+		return mgr->err;
 
-	  bt_lockpage(BtLockParent, prev->latch, bt->thread_no);
 	  bt_unlockpage(BtLockWrite, prev->latch);
 
-	  //  remove atomic lock on master page
+	  if( lsm )
+		bt_syncpage (mgr, prev->page, prev->latch);
 
 	  bt_unlockpage(BtLockAtomic, latch);
+	  bt_unpinlatch(mgr, latch);
+
+	  // go through the list of splits and
+	  // release the latch pins
+
+	  while( entry = latch->split ) {
+	  	latch = mgr->latchsets + entry;
+		bt_unpinlatch(mgr, latch);
+	  }
+
 	  continue;
 	}
 
-	//  finished if prev page occupied (either master or final split)
+	//	since there are no splits, we're
+	//  finished if master page occupied
 
 	if( prev->page->act ) {
-	  bt_unlockpage(BtLockWrite, latch);
-	  bt_unlockpage(BtLockAtomic, latch);
-	  bt_unpinlatch(bt->mgr, latch);
+	  bt_unlockpage(BtLockAtomic, prev->latch);
+	  bt_unlockpage(BtLockWrite, prev->latch);
+
+	  if( lsm )
+		bt_syncpage (mgr, prev->page, prev->latch);
+
+	  bt_unpinlatch(mgr, prev->latch);
 	  continue;
 	}
 
 	// any and all splits were reversed, and the
 	// master page located in prev is empty, delete it
-	// by pulling over master's right sibling.
 
-	// Remove empty master's fence key
+	if( bt_deletepage (mgr, prev, thread_no, 1) )
+		return mgr->err;
+  }
 
-	ptr = keyptr(prev->page,prev->page->cnt);
+  free (locks);
+  return 0;
+}
 
-	if( bt_deletekey (bt->mgr, ptr->key, ptr->len, 1, bt->thread_no) )
-		goto atomicerr;
+//  promote a page into the larger btree
 
-	//	perform the remainder of the delete
-	//	from the FIFO queue
+BTERR bt_txnpromote (BtDb *bt)
+{
+uint entry, slot, idx;
+BtPageSet set[1];
+BtSlot *node;
+BtKey *ptr;
+BtVal *val;
 
-	leaf = calloc (sizeof(AtomicKey), 1), que++;
+  while( 1 ) {
+#ifdef unix
+	entry = __sync_fetch_and_add(&bt->mgr->latchpromote, 1);
+#else
+	entry = _InterlockedIncrement (&bt->mgr->latchpromote) - 1;
+#endif
+	entry %= bt->mgr->latchtotal;
 
-	memcpy (leaf->leafkey, ptr, ptr->len + sizeof(BtKey));
-	leaf->page_no = prev->latch->page_no;
-	leaf->entry = prev->latch->entry;
-	leaf->nounlock = 1;
-	leaf->type = 2;
-  
-	if( tail )
-	  tail->next = leaf;
-	else
-	  head = leaf;
+	if( !entry )
+		continue;
 
-	tail = leaf;
+	set->latch = bt->mgr->latchsets + entry;
 
-	//	leave atomic lock in place until
-	//	deletion completes in next phase.
+	if( !bt_mutextry(set->latch->modify) )
+		continue;
 
-	bt_unlockpage(BtLockWrite, prev->latch);
-  }
+	//  skip this entry if it is pinned
 
-  bt_txnavailrelease (bt, avail);
+	if( set->latch->pin & ~CLOCK_bit ) {
+	  bt_releasemutex(set->latch->modify);
+	  continue;
+	}
 
-  que *= bt->mgr->pagezero->alloc->lvl;
+	set->page = bt_mappage (bt->mgr, set->latch);
 
-  if( bt_txnavailrequest (bt, que) )
-	return bt->mgr->err;
+	// entry never used or has no right sibling
 
-  //  add & delete keys for any pages split or merged during transaction
+	if( !set->latch->page_no || !bt_getid (set->page->right) ) {
+	  bt_releasemutex(set->latch->modify);
+	  continue;
+	}
 
-  if( leaf = head )
-    do {
-	  set->latch = bt->mgr->latchsets + leaf->entry;
-	  set->page = bt_mappage (bt->mgr, set->latch);
+	// entry interiour node or being killed or promoted
 
-	  bt_putid (value, leaf->page_no);
-	  ptr = (BtKey *)leaf->leafkey;
+	if( set->page->lvl || set->page->free || set->page->kill ) {
+	  bt_releasemutex(set->latch->modify);
+	  continue;
+	}
 
-	  switch( leaf->type ) {
-	  case 0:	// insert key
-	    if( bt_insertkey (bt->mgr, ptr->key, ptr->len, 1, value, BtId, 1, bt->thread_no) )
-		  goto atomicerr;
+	//  pin the page for our useage
 
-		break;
+	set->latch->pin++;
+	set->latch->promote = 1;
+	bt_releasemutex(set->latch->modify);
 
-	  case 1:	// delete key
-		if( bt_deletekey (bt->mgr, ptr->key, ptr->len, 1, bt->thread_no) )
-		  goto atomicerr;
+	bt_lockpage (BtLockWrite, set->latch, bt->thread_no);
 
-		break;
+	//  remove the key for the page
+	//	and wait for other threads to fade away
 
-	  case 2:	// free page
-		if( bt_atomicfree (bt->mgr, set, bt->thread_no) )
-		  goto atomicerr;
+	ptr = keyptr(set->page, set->page->cnt);
 
-		break;
-	  }
+	if( bt_deletekey (bt->mgr, ptr->key, ptr->len, 1, bt->thread_no) )
+		return bt->mgr->err;
 
-	  if( !leaf->nounlock )
-	    bt_unlockpage (BtLockParent, set->latch);
+	bt_unlockpage (BtLockWrite, set->latch);
+while( (set->latch->pin & ~CLOCK_bit) > 1 )
+sched_yield();
+	bt_lockpage (BtLockDelete, set->latch, bt->thread_no);
+	bt_lockpage (BtLockAtomic, set->latch, bt->thread_no);
+	bt_lockpage (BtLockWrite, set->latch, bt->thread_no);
 
-	  bt_unpinlatch (bt->mgr, set->latch);
-	  tail = leaf->next;
-	  free (leaf);
-	} while( leaf = tail );
+	// transfer slots in our selected page to larger btree
+if( !(set->latch->page_no % 100) )
+fprintf(stderr, "Promote page %d, %d keys\n", set->latch->page_no, set->page->act);
 
-  bt_txnavailrelease (bt, que);
+	if( bt_atomicexec (bt->main, set->page, 0, bt->mgr->pagezero->redopages ? 1 : 0, bt->thread_no) )
+		return bt->main->err;
 
-  // return success
+	//  now delete the page
 
-  free (locks);
-  return 0;
-atomicerr:
-  return -1;
+	bt_unlockpage (BtLockDelete, set->latch);
+	bt_unlockpage (BtLockAtomic, set->latch);
+	return bt_deletepage (bt->mgr, set, bt->thread_no, 0);
+  }
 }
 
 //	set cursor to highest slot on highest page
@@ -3628,14 +3419,14 @@ uint entry = 0;
 		if( *latch->access->rin & MASK )
 			fprintf(stderr, "latchset %d accesslocked for page %d\n", entry, latch->page_no);
 
-		if( *latch->parent->exclusive )
-			fprintf(stderr, "latchset %d parentlocked for page %d\n", entry, latch->page_no);
+//		if( *latch->parent->xcl->value )
+//			fprintf(stderr, "latchset %d parentlocked for page %d\n", entry, latch->page_no);
 
-		if( *latch->atomic->exclusive )
-			fprintf(stderr, "latchset %d atomiclocked for page %d\n", entry, latch->page_no);
+//		if( *latch->atomic->xcl->value )
+//			fprintf(stderr, "latchset %d atomiclocked for page %d\n", entry, latch->page_no);
 
-		if( *latch->modify->exclusive )
-			fprintf(stderr, "latchset %d modifylocked for page %d\n", entry, latch->page_no);
+//		if( *latch->modify->value )
+//			fprintf(stderr, "latchset %d modifylocked for page %d\n", entry, latch->page_no);
 
 		if( latch->pin & ~CLOCK_bit )
 			fprintf(stderr, "latchset %d pinned %d times for page %d\n", entry, latch->pin & ~CLOCK_bit, latch->page_no);
@@ -3709,7 +3500,7 @@ FILE *in;
 			  line++;
 
 			  if( !args->num ) {
-			    if( bt_insertkey (bt->mgr, key, 10, 0, key + 10, len - 10, 1, bt->thread_no) )
+			    if( bt_insertkey (bt->mgr, key, 10, 0, key + 10, len - 10, Unique, bt->thread_no) )
 				  fprintf(stderr, "Error %d Line: %d source: %d\n", bt->mgr->err, bt->mgr->line, line), exit(0);
 			    len = 0;
 				continue;
@@ -3753,7 +3544,7 @@ FILE *in;
 			{
 			  line++;
 
-			  if( bt_insertkey (bt->mgr, key, len, 0, NULL, 0, 1, bt->thread_no) )
+			  if( bt_insertkey (bt->mgr, key, len, 0, NULL, 0, Unique, bt->thread_no) )
 				fprintf(stderr, "Error %d Line: %d source: %d\n", bt->mgr->err, bt->mgr->line, line), exit(0);
 			  len = 0;
 			}
@@ -3808,7 +3599,6 @@ FILE *in;
 				cnt++;
 			   }
 
-			set->latch->avail = 1;
 			bt_unlockpage (BtLockRead, set->latch);
 			bt_unpinlatch (bt->mgr, set->latch);
 	  	} while( page_no = next );
@@ -3844,7 +3634,7 @@ FILE *in;
 		posix_fadvise( bt->mgr->idx, 0, 0, POSIX_FADV_SEQUENTIAL);
 #endif
 		fprintf(stderr, "started counting\n");
-		page_no = LEAF_page;
+		next = REDO_page + bt->mgr->pagezero->redopages;
 
 		while( page_no < bt_getid(bt->mgr->pagezero->alloc->right) ) {
 			if( bt_readpage (bt->mgr, bt->cursor, page_no) )
@@ -3853,7 +3643,8 @@ FILE *in;
 			if( !bt->cursor->free && !bt->cursor->lvl )
 				cnt += bt->cursor->act;
 
-			page_no++;
+			bt->mgr->reads++;
+			page_no = next++;
 		}
 		
 	  	cnt--;	// remove stopper key
@@ -3861,7 +3652,6 @@ FILE *in;
 		break;
 	}
 
-	fprintf(stderr, "%d reads %d writes %d found\n", bt->mgr->reads, bt->mgr->writes, bt->mgr->found);
 	bt_close (bt);
 #ifdef unix
 	return NULL;
@@ -3883,8 +3673,8 @@ pthread_t *threads;
 HANDLE *threads;
 #endif
 ThreadArg *args;
+uint redopages = 0;
 uint poolsize = 0;
-uint recovery = 0;
 uint mainpool = 0;
 uint mainbits = 0;
 float elapsed;
@@ -3924,7 +3714,10 @@ BtKey *ptr;
 		num = atoi(argv[6]);
 
 	if( argc > 7 )
-		recovery = atoi(argv[7]);
+		redopages = atoi(argv[7]);
+
+	if( redopages + REDO_page > 65535 )
+		fprintf (stderr, "Warning: Recovery buffer too large\n");
 
 	if( argc > 8 )
 		mainbits = atoi(argv[8]);
@@ -3940,19 +3733,22 @@ BtKey *ptr;
 #endif
 	args = malloc ((cnt + 1) * sizeof(ThreadArg));
 
-	mgr = bt_mgr (argv[1], bits, poolsize, recovery);
+	mgr = bt_mgr (argv[1], bits, poolsize, redopages);
 
 	if( !mgr ) {
 		fprintf(stderr, "Index Open Error %s\n", argv[1]);
 		exit (1);
 	}
 
-	main = bt_mgr (argv[2], mainbits, mainpool, 0);
+	if( mainbits ) {
+		main = bt_mgr (argv[2], mainbits, mainpool, 0);
 
-	if( !main ) {
-		fprintf(stderr, "Index Open Error %s\n", argv[2]);
-		exit (1);
-	}
+		if( !main ) {
+			fprintf(stderr, "Index Open Error %s\n", argv[2]);
+			exit (1);
+		}
+	} else
+		main = NULL;
 
 	//	fire off threads
 
@@ -3993,9 +3789,14 @@ BtKey *ptr;
 		CloseHandle(threads[idx]);
 #endif
 	bt_poolaudit(mgr);
-	bt_poolaudit(main);
 
-	bt_mgrclose (main);
+	if( main )
+		bt_poolaudit(main);
+
+	fprintf(stderr, "%d reads %d writes %d found\n", mgr->reads, mgr->writes, mgr->found);
+
+	if( main )
+		bt_mgrclose (main);
 	bt_mgrclose (mgr);
 
 	elapsed = getCpuTime(0) - start;