]> pd.if.org Git - btree/blob - jaluta.c
Add finalized threadskv10g LSM btree.
[btree] / jaluta.c
1 // jaluta's balanced B-Link tree algorithms
2 // 26 APR 2013
3
4 // author: karl malbrain, malbrain@cal.berkeley.edu
5
6 /*
7 This work, including the source code, documentation
8 and related data, is placed into the public domain.
9
10 The orginal author is Karl Malbrain.
11
12 THIS SOFTWARE IS PROVIDED AS-IS WITHOUT WARRANTY
13 OF ANY KIND, NOT EVEN THE IMPLIED WARRANTY OF
14 MERCHANTABILITY. THE AUTHOR OF THIS SOFTWARE,
15 ASSUMES _NO_ RESPONSIBILITY FOR ANY CONSEQUENCE
16 RESULTING FROM THE USE, MODIFICATION, OR
17 REDISTRIBUTION OF THIS SOFTWARE.
18 */
19
20 // Please see the project home page for documentation
21 // http://code.google.com/p/high-concurrency-btree
22
23 #define _FILE_OFFSET_BITS 64
24 #define _LARGEFILE64_SOURCE
25
26 #ifdef linux
27 #define _GNU_SOURCE
28 #endif
29
30 #ifdef unix
31 #include <unistd.h>
32 #include <stdio.h>
33 #include <stdlib.h>
34 #include <time.h>
35 #include <fcntl.h>
36 #include <sys/mman.h>
37 #include <errno.h>
38 #else
39 #define WIN32_LEAN_AND_MEAN
40 #include <windows.h>
41 #include <stdio.h>
42 #include <stdlib.h>
43 #include <time.h>
44 #include <fcntl.h>
45 #endif
46
47 #include <memory.h>
48 #include <string.h>
49
50 typedef unsigned long long      uid;
51
52 #ifndef unix
53 typedef unsigned long long      off64_t;
54 typedef unsigned short          ushort;
55 typedef unsigned int            uint;
56 #endif
57
58 #define BT_ro 0x6f72    // ro
59 #define BT_rw 0x7772    // rw
60 #define BT_fl 0x6c66    // fl
61
62 #define BT_maxbits              24                                      // maximum page size in bits
63 #define BT_minbits              9                                       // minimum page size in bits
64 #define BT_minpage              (1 << BT_minbits)       // minimum page size
65
66 #define BT_hashsize             512             // size of hash index for page cache
67 #define BT_hashprime    8191    // prime number for hashing
68
69 typedef enum{
70         BtLockShared    = 1,
71         BtLockUpdate    = 2,
72         BtLockXclusive  = 3,
73         BtLockUpgrade   = 4,
74 }BtLock;
75
76 //      Define the length of the page and key pointers
77
78 #define BtId 6
79
80 //      Page key slot definition.
81
82 //      If BT_maxbits is 16 or less, you can save 4 bytes
83 //      for each key stored by making the first two uints
84 //      into ushorts.  You can also save 4 bytes by removing
85 //      the tod field from the key.
86
87 typedef struct {
88         uint off;                               // page offset for key start
89         uint tod;                               // time-stamp for key
90         unsigned char id[BtId]; // id associated with key
91 } BtSlot;
92
93 //      The key structure occupies space at the upper end of
94 //      each page.  It's a length byte followed by up to
95 //      255 value bytes.
96
97 typedef struct {
98         unsigned char len;
99         unsigned char key[255];
100 } *BtKey;
101
102 //      The first part of an index page.
103 //      It is immediately followed
104 //      by the BtSlot array of keys.
105
106 typedef struct {
107         uint cnt;                                       // count of keys in page
108         uint min;                                       // next key offset
109         unsigned char lvl:3;            // level of page
110         unsigned char bits:5;           // page size in bits
111         unsigned char fence;            // len of fence key at top of page
112         unsigned char right[BtId];      // page number to right
113         BtSlot slots[0];                        // page slots
114 } *BtPage;
115
116 //      The memory mapping hash table entry
117
118 typedef struct {
119         BtPage page;            // mapped page pointer
120         uid  page_no;           // mapped page number
121         void *lruprev;          // least recently used previous cache block
122         void *lrunext;          // lru next cache block
123         void *hashprev;         // previous cache block for the same hash idx
124         void *hashnext;         // next cache block for the same hash idx
125 #ifndef unix
126         HANDLE hmap;
127 #endif
128 }BtHash;
129
130 //      The object structure for Btree access
131
132 typedef struct {
133         uint page_size;         // each page size       
134         uint page_bits;         // each page size in bits       
135         uid parentpage;         // current parent page number   
136         uid cursorpage;         // current cursor page number   
137         uid childpage;          // current child page number    
138         int  err;
139         uint mode;                      // read-write mode
140         uint mapped_io;         // use memory mapping
141         BtPage temp;            // temporary frame buffer (memory mapped/file IO)
142         BtPage alloc;           // frame buffer for alloc page ( page 0 )
143         BtPage cursor;          // cached frame for start/next (never mapped)
144         BtPage frame;           // spare frame for the page split (never mapped)
145         BtPage parent;          // current parent page
146         BtPage child;           // current child page
147         BtPage sibling;         // current sibling page
148         BtPage sibling2;        // current sibling2 page
149 #ifdef unix
150         int idx;
151 #else
152         HANDLE idx;
153 #endif
154         unsigned char *mem;                     // frame, cursor, page memory buffer
155         int nodecnt;                            // highest page cache node in use
156         int nodemax;                            // highest page cache node allocated
157         int hashmask;                           // number of hash headers in cache - 1
158         BtHash *lrufirst;                       // lru list head
159         BtHash *lrulast;                        // lru list tail
160         ushort cache[BT_hashsize];      // hash index for cache
161         BtHash nodes[1];                        // page cache follows
162 } BtDb;
163
164 typedef enum {
165 BTERR_ok = 0,
166 BTERR_struct,
167 BTERR_ovflw,
168 BTERR_lock,
169 BTERR_map,
170 BTERR_wrt,
171 BTERR_hash,
172 BTERR_restart
173 } BTERR;
174
175 // B-Tree functions
176 extern void bt_close (BtDb *bt);
177 extern BtDb *bt_open (char *name, uint mode, uint bits, uint cacheblk);
178 extern BTERR  bt_insertkey (BtDb *bt, unsigned char *key, uint len, uid id, uint tod);
179 extern BTERR  bt_deletekey (BtDb *bt, unsigned char *key, uint len);
180 extern uid bt_findkey    (BtDb *bt, unsigned char *key, uint len);
181 extern uint bt_startkey  (BtDb *bt, unsigned char *key, uint len);
182 extern uint bt_nextkey   (BtDb *bt, uint slot);
183
184 //  Helper functions to return slot values
185
186 extern BtKey bt_key (BtDb *bt, uint slot);
187 extern uid bt_uid (BtDb *bt, uint slot);
188 extern uint bt_tod (BtDb *bt, uint slot);
189
190 //  BTree page number constants
191 #define ALLOC_page              0
192 #define ROOT_page               1
193
194 //  The page is allocated from low and hi ends.
195 //  The key offsets and row-id's are allocated
196 //  from the bottom, while the text of the key
197 //  is allocated from the top.  When the two
198 //  areas meet, the page overflowns.
199
200 //  A key consists of a length byte, two bytes of
201 //  index number (0 - 65535), and up to 253 bytes
202 //  of key value.  Duplicate keys are discarded.
203 //  Associated with each key is a 48 bit row-id.
204
205 //  The b-tree root is always located at page 1.
206 //      The first leaf page of level zero is always
207 //      located on page 2.
208
209 //      The b-tree pages at each level are linked
210 //      with next page to right to facilitate
211 //      cursors and provide for concurrency.
212
213 //      When to root page overflows, it is split in two and
214 //      the tree height is raised by a new root at page
215 //      one with two keys.
216
217 //  Groups of pages from the btree are optionally
218 //  cached with memory mapping. A hash table is used to keep
219 //  track of the cached pages.  This behaviour is controlled
220 //  by the cache block size parameter to bt_open.
221
222 //  To achieve maximum concurrency one page is locked at a time
223 //  as the tree is traversed to find leaf key in question. The right
224 //  page numbers are used in cases where the page is being split,
225 //      or consolidated.
226
227 //  Page 0 is dedicated to lock for new page extensions,
228 //      and chains empty pages together for reuse.
229
230 //      Empty nodes are chained together through the ALLOC page and reused.
231
232 //      A special open mode of BT_fl is provided to safely access files on
233 //      WIN32 networks. WIN32 network operations should not use memory mapping.
234 //      This WIN32 mode sets FILE_FLAG_NOBUFFERING and FILE_FLAG_WRITETHROUGH
235 //      to prevent local caching of network file contents.
236
237 //      Access macros to address slot and key values from the page
238
239 #define slotptr(page, slot) ((page)->slots + slot - 1)
240 #define keyptr(page, slot) ((BtKey)((unsigned char*)(page) + slotptr(page, slot)->off))
241
242 void bt_putid(unsigned char *dest, uid id)
243 {
244 int i = BtId;
245
246         while( i-- )
247                 dest[i] = (unsigned char)id, id >>= 8;
248 }
249
250 uid bt_getid(unsigned char *src)
251 {
252 uid id = 0;
253 int i;
254
255         for( i = 0; i < BtId; i++ )
256                 id <<= 8, id |= *src++; 
257
258         return id;
259 }
260
261 // place requested latch on requested page_no.
262 //              the Shared latch is a read lock over segment 0
263 //              the Update latch is a write lock over segment 1
264 //              the Xclusive latch is a write lock over segment 0 & 1
265 //              the Upgrade latch upgrades Update to Xclusive
266
267 BTERR bt_lockpage(BtDb *bt, uid page_no, BtLock mode)
268 {
269 off64_t off = page_no << bt->page_bits;
270 uint len = sizeof(*bt->parent);
271 uint type;
272 #ifdef unix
273 int flag = PROT_READ | ( bt->mode == BT_ro ? 0 : PROT_WRITE );
274 struct flock lock[1];
275 #else
276 uint flags = 0;
277 OVERLAPPED ovl[1];
278 #endif
279
280         switch( mode ) {
281         case BtLockShared:      // lock segment 0 w/read lock
282                 type = 0;
283                 break;
284
285         case BtLockUpdate:      // lock segment 1 w/write lock
286                 off += sizeof(*bt->parent);
287                 type = 1;
288                 break;
289
290         case BtLockXclusive:// lock both segments w/write lock
291                 len +=  sizeof(*bt->parent);
292                 type = 1;
293                 break;
294
295         case BtLockUpgrade:     // lock segment 0 w/write lock
296                 type = 1;
297                 break;
298         }
299
300 #ifdef unix
301         memset (lock, 0, sizeof(lock));
302
303         lock->l_start = off;
304         lock->l_type = type ? F_WRLCK : F_RDLCK;
305         lock->l_len = len;
306         lock->l_whence = 0;
307
308         if( fcntl (bt->idx, F_SETLKW, lock) < 0 )
309                 return bt->err = BTERR_lock;
310
311         return 0;
312 #else
313         memset (ovl, 0, sizeof(ovl));
314         ovl->OffsetHigh = (uint)(off >> 32);
315         ovl->Offset = (uint)off;
316
317         //      use large offsets to
318         //      simulate advisory locking
319
320         ovl->OffsetHigh |= 0x80000000;
321
322         if( type = 1 )
323                 flags |= LOCKFILE_EXCLUSIVE_LOCK;
324
325         if( LockFileEx (bt->idx, flags, 0, len, 0L, ovl) )
326                 return bt->err = 0;
327
328         return bt->err = BTERR_lock;
329 #endif 
330 }
331
332 // remove lock on requested page_no.
333
334 BTERR bt_unlockpage(BtDb *bt, uid page_no, BtLock mode)
335 {
336 off64_t off = page_no << bt->page_bits;
337 uint len = sizeof(*bt->parent);
338 #ifdef unix
339 struct flock lock[1];
340 #else
341 OVERLAPPED ovl[1];
342 #endif
343
344         switch( mode ) {
345         case BtLockShared:      // unlock segment 0
346                 break;
347
348         case BtLockUpdate:      // unlock segment 1
349                 off += sizeof(*bt->parent);
350                 break;
351
352         case BtLockXclusive:// unlock both segments
353                 len += sizeof(*bt->parent);
354                 break;
355
356         case BtLockUpgrade:     // unlock segment 0
357                 break;
358         }
359
360 #ifdef unix
361         memset (lock, 0, sizeof(lock));
362
363         lock->l_start = off;
364         lock->l_type = F_UNLCK;
365         lock->l_len = len;
366         lock->l_whence = 0;
367
368         if( fcntl (bt->idx, F_SETLK, lock) < 0 )
369                 return bt->err = BTERR_lock;
370 #else
371         memset (ovl, 0, sizeof(ovl));
372         ovl->OffsetHigh = (uint)(off >> 32);
373         ovl->Offset = (uint)off;
374
375         //      use large offsets to
376         //      simulate advisory locking
377
378         ovl->OffsetHigh |= 0x80000000;
379
380         if( !UnlockFileEx (bt->idx, 0, len, 0, ovl) )
381                 return GetLastError(), bt->err = BTERR_lock;
382 #endif
383
384         return bt->err = 0;
385 }
386
387 //      close and release memory
388
389 void bt_close (BtDb *bt)
390 {
391 BtHash *hash;
392 #ifdef unix
393         // release mapped pages
394
395         if( hash = bt->lrufirst )
396                 do munmap (hash->page, (bt->hashmask+1) << bt->page_bits);
397                 while(hash = hash->lrunext);
398
399         if ( bt->mem )
400                 free (bt->mem);
401         close (bt->idx);
402         free (bt);
403 #else
404         if( hash = bt->lrufirst )
405           do
406           {
407                 FlushViewOfFile(hash->page, 0);
408                 UnmapViewOfFile(hash->page);
409                 CloseHandle(hash->hmap);
410           } while(hash = hash->lrunext);
411
412         if ( bt->mem)
413                 VirtualFree (bt->mem, 0, MEM_RELEASE);
414         FlushFileBuffers(bt->idx);
415         CloseHandle(bt->idx);
416         GlobalFree (bt);
417 #endif
418 }
419
420 //  open/create new btree
421 //      call with file_name, BT_openmode, bits in page size (e.g. 16),
422 //              size of mapped page cache (e.g. 8192) or zero for no mapping.
423
424 BtDb *bt_open (char *name, uint mode, uint bits, uint nodemax)
425 {
426 BtLock lockmode = BtLockXclusive;
427 uint lvl, attr, cacheblk;
428 BtPage alloc;
429 off64_t size;
430 uint amt[1];
431 BtKey key;
432 BtDb* bt;
433
434 #ifndef unix
435 SYSTEM_INFO sysinfo[1];
436 #endif
437
438 #ifdef unix
439         bt = malloc (sizeof(BtDb) + nodemax * sizeof(BtHash));
440         memset (bt, 0, sizeof(BtDb));
441
442         switch (mode & 0x7fff)
443         {
444         case BT_fl:
445         case BT_rw:
446                 bt->idx = open ((char*)name, O_RDWR | O_CREAT, 0666);
447                 break;
448
449         case BT_ro:
450         default:
451                 bt->idx = open ((char*)name, O_RDONLY);
452                 lockmode = BtLockShared;
453                 break;
454         }
455         if( bt->idx == -1 )
456                 return free(bt), NULL;
457         
458         if( nodemax )
459                 cacheblk = 4096;        // page size for unix
460         else
461                 cacheblk = 0;
462
463 #else
464         bt = GlobalAlloc (GMEM_FIXED|GMEM_ZEROINIT, sizeof(BtDb) + nodemax * sizeof(BtHash));
465         attr = FILE_ATTRIBUTE_NORMAL;
466         switch (mode & 0x7fff)
467         {
468         case BT_fl:
469                 attr |= FILE_FLAG_WRITE_THROUGH | FILE_FLAG_NO_BUFFERING;
470
471         case BT_rw:
472                 bt->idx = CreateFile(name, GENERIC_READ| GENERIC_WRITE, FILE_SHARE_READ|FILE_SHARE_WRITE, NULL, OPEN_ALWAYS, attr, NULL);
473                 break;
474
475         case BT_ro:
476         default:
477                 bt->idx = CreateFile(name, GENERIC_READ, FILE_SHARE_READ|FILE_SHARE_WRITE, NULL, OPEN_EXISTING, attr, NULL);
478                 lockmode = BtLockShared;
479                 break;
480         }
481         if( bt->idx == INVALID_HANDLE_VALUE )
482                 return GlobalFree(bt), NULL;
483
484         // normalize cacheblk to multiple of sysinfo->dwAllocationGranularity
485         GetSystemInfo(sysinfo);
486
487         if( nodemax )
488                 cacheblk = sysinfo->dwAllocationGranularity;
489         else
490                 cacheblk = 0;
491 #endif
492
493         // determine sanity of page size
494
495         if( bits > BT_maxbits )
496                 bits = BT_maxbits;
497         else if( bits < BT_minbits )
498                 bits = BT_minbits;
499
500         if ( bt_lockpage(bt, ALLOC_page, lockmode) )
501                 return bt_close (bt), NULL;
502
503 #ifdef unix
504         *amt = 0;
505
506         // read minimum page size to get root info
507
508         if( size = lseek (bt->idx, 0L, 2) ) {
509                 alloc = malloc (BT_minpage);
510                 pread(bt->idx, alloc, BT_minpage, 0);
511                 bits = alloc->bits;
512                 free (alloc);
513         } else if( mode == BT_ro )
514                 return bt_close (bt), NULL;
515 #else
516         size = GetFileSize(bt->idx, amt);
517
518         if( size || *amt ) {
519                 alloc = VirtualAlloc(NULL, BT_minpage, MEM_COMMIT, PAGE_READWRITE);
520                 if( !ReadFile(bt->idx, (char *)alloc, BT_minpage, amt, NULL) )
521                         return bt_close (bt), NULL;
522                 bits = alloc->bits;
523                 VirtualFree (alloc, 0, MEM_RELEASE);
524         } else if( mode == BT_ro )
525                 return bt_close (bt), NULL;
526 #endif
527
528         bt->page_size = 1 << bits;
529         bt->page_bits = bits;
530
531         bt->nodemax = nodemax;
532         bt->mode = mode;
533
534         // setup cache mapping
535
536         if( cacheblk ) {
537                 if( cacheblk < bt->page_size )
538                         cacheblk = bt->page_size;
539
540                 bt->hashmask = (cacheblk >> bits) - 1;
541                 bt->mapped_io = 1;
542         }
543
544 #ifdef unix
545         bt->mem = malloc (8 *bt->page_size);
546 #else
547         bt->mem = VirtualAlloc(NULL, 8 * bt->page_size, MEM_COMMIT, PAGE_READWRITE);
548 #endif
549         bt->frame = (BtPage)bt->mem;
550         bt->cursor = (BtPage)(bt->mem + bt->page_size);
551         bt->alloc = (BtPage)(bt->mem + 2 * bt->page_size);
552         bt->parent = (BtPage)(bt->mem + 3 * bt->page_size);
553         bt->child = (BtPage)(bt->mem + 4 * bt->page_size);
554         bt->temp = (BtPage)(bt->mem + 5 * bt->page_size);
555         bt->sibling = (BtPage)(bt->mem + 6 * bt->page_size);
556         bt->sibling2 = (BtPage)(bt->mem + 7 * bt->page_size);
557
558         if( size || *amt ) {
559                 if ( bt_unlockpage(bt, ALLOC_page, lockmode) )
560                         return bt_close (bt), NULL;
561
562                 return bt;
563         }
564
565         // initialize an empty b-tree with alloc page & root page
566
567         memset (bt->alloc, 0, bt->page_size);
568         bt_putid(bt->alloc->right, ROOT_page + 1);
569         bt->alloc->bits = bt->page_bits;
570
571 #ifdef unix
572         if( write (bt->idx, bt->alloc, bt->page_size) < bt->page_size )
573                 return bt_close (bt), NULL;
574 #else
575         if( !WriteFile (bt->idx, (char *)bt->alloc, bt->page_size, amt, NULL) )
576                 return bt_close (bt), NULL;
577
578         if( *amt < bt->page_size )
579                 return bt_close (bt), NULL;
580 #endif
581
582         //      write root page
583
584         memset (bt->frame, 0, bt->page_size);
585         bt->frame->bits = bt->page_bits;
586
587         bt->frame->min = bt->page_size;
588 #ifdef unix
589         if( write (bt->idx, bt->frame, bt->page_size) < bt->page_size )
590                 return bt_close (bt), NULL;
591 #else
592         if( !WriteFile (bt->idx, (char *)bt->frame, bt->page_size, amt, NULL) )
593                 return bt_close (bt), NULL;
594
595         if( *amt < bt->page_size )
596                 return bt_close (bt), NULL;
597 #endif
598
599         // create initial empty page area by writing last page of first
600         // cache area (other pages are zeroed by O/S)
601
602         if( bt->mapped_io && bt->hashmask > 2 ) {
603                 memset(bt->frame, 0, bt->page_size);
604
605 #ifdef unix
606                 pwrite(bt->idx, bt->frame, bt->page_size, bt->hashmask << bt->page_bits);
607 #else
608                 SetFilePointer (bt->idx, bt->hashmask << bt->page_bits, NULL, FILE_BEGIN);
609                 if( !WriteFile (bt->idx, (char *)bt->frame, bt->page_size, amt, NULL) )
610                         return bt_close (bt), NULL;
611                 if( *amt < bt->page_size )
612                         return bt_close (bt), NULL;
613 #endif
614         }
615
616         if( bt_unlockpage(bt, ALLOC_page, lockmode) )
617                 return bt_close (bt), NULL;
618
619         return bt;
620 }
621
622 //  reset parent/child page pointers
623
624 void bt_resetpages (BtDb *bt)
625 {
626         if( bt->mapped_io )
627                 return;
628
629         bt->frame = (BtPage)bt->mem;
630         bt->cursor = (BtPage)(bt->mem + bt->page_size);
631         bt->alloc = (BtPage)(bt->mem + 2 * bt->page_size);
632         bt->parent = (BtPage)(bt->mem + 3 * bt->page_size);
633         bt->child = (BtPage)(bt->mem + 4 * bt->page_size);
634         bt->temp = (BtPage)(bt->mem + 5 * bt->page_size);
635         bt->sibling = (BtPage)(bt->mem + 6 * bt->page_size);
636         bt->sibling2 = (BtPage)(bt->mem + 7 * bt->page_size);
637 }
638
639 //      return pointer to high key
640 //      or NULL if infinite value
641
642 BtKey bt_highkey (BtDb *bt, BtPage page)
643 {
644         if( page->lvl )
645           if( bt_getid (page->right) )
646                 return keyptr(page, page->cnt);
647           else
648                 return NULL;
649
650         if( bt_getid (page->right) )
651                 return ((BtKey)((unsigned char*)(page) + bt->page_size - page->fence));
652
653         return NULL;
654 }
655
656 //      return pointer to slot key in index page
657 //      or NULL if infinite value
658
659 BtKey bt_slotkey (BtPage page, uint slot)
660 {
661         if( slot < page->cnt || bt_getid (page->right) )
662                 return keyptr(page, slot);
663         else
664                 return NULL;
665 }
666
667 //  compare two keys, returning > 0, = 0, or < 0
668 //  as the comparison value
669
670 int keycmp (BtKey key1, unsigned char *key2, uint len2)
671 {
672 uint len1 = key1->len;
673 int ans;
674
675         if( ans = memcmp (key1->key, key2, len1 > len2 ? len2 : len1) )
676                 return ans;
677
678         if( len1 > len2 )
679                 return 1;
680         if( len1 < len2 )
681                 return -1;
682
683         return 0;
684 }
685
686 //  Update current page of btree by writing file contents
687 //      or flushing mapped area to disk.
688
689 BTERR bt_update (BtDb *bt, BtPage page, uid page_no)
690 {
691 off64_t off = page_no << bt->page_bits;
692
693 #ifdef unix
694     if ( !bt->mapped_io )
695          if ( pwrite(bt->idx, page, bt->page_size, off) != bt->page_size )
696                  return bt->err = BTERR_wrt;
697 #else
698 uint amt[1];
699         if ( !bt->mapped_io )
700         {
701                 SetFilePointer (bt->idx, (long)off, (long*)(&off)+1, FILE_BEGIN);
702                 if( !WriteFile (bt->idx, (char *)page, bt->page_size, amt, NULL) )
703                         return GetLastError(), bt->err = BTERR_wrt;
704
705                 if( *amt < bt->page_size )
706                         return GetLastError(), bt->err = BTERR_wrt;
707         } 
708         else if ( bt->mode == BT_fl ) {
709                         FlushViewOfFile(page, bt->page_size);
710                         FlushFileBuffers(bt->idx);
711         }
712 #endif
713         return 0;
714 }
715
716 // find page in cache 
717
718 BtHash *bt_findhash(BtDb *bt, uid page_no)
719 {
720 BtHash *hash;
721 uint idx;
722
723         // compute cache block first page and hash idx 
724
725         page_no &= ~bt->hashmask;
726         idx = (uint)(page_no * BT_hashprime % BT_hashsize);
727
728         if( bt->cache[idx] ) 
729                 hash = bt->nodes + bt->cache[idx];
730         else
731                 return NULL;
732
733         do if( hash->page_no == page_no )
734                  break;
735         while(hash = hash->hashnext );
736
737         return hash;
738 }
739
740 // add page cache entry to hash index
741
742 void bt_linkhash(BtDb *bt, BtHash *node, uid page_no)
743 {
744 uint idx = (uint)((page_no & ~bt->hashmask) * BT_hashprime % BT_hashsize);
745 BtHash *hash;
746
747         if( bt->cache[idx] ) {
748                 node->hashnext = hash = bt->nodes + bt->cache[idx];
749                 hash->hashprev = node;
750         }
751
752         node->hashprev = NULL;
753         bt->cache[idx] = (ushort)(node - bt->nodes);
754 }
755
756 // remove cache entry from hash table
757
758 void bt_unlinkhash(BtDb *bt, BtHash *node)
759 {
760 uint idx = (uint)((node->page_no & ~bt->hashmask) * BT_hashprime % BT_hashsize);
761 BtHash *hash;
762
763         // unlink node
764         if( hash = node->hashprev )
765                 hash->hashnext = node->hashnext;
766         else if( hash = node->hashnext )
767                 bt->cache[idx] = (ushort)(hash - bt->nodes);
768         else
769                 bt->cache[idx] = 0;
770
771         if( hash = node->hashnext )
772                 hash->hashprev = node->hashprev;
773 }
774
775 // add cache page to lru chain and map pages
776
777 BtPage bt_linklru(BtDb *bt, BtHash *hash, uid page_no)
778 {
779 int flag;
780 off64_t off = (page_no & ~bt->hashmask) << bt->page_bits;
781 off64_t limit = off + ((bt->hashmask+1) << bt->page_bits);
782 BtHash *node;
783
784         memset(hash, 0, sizeof(BtHash));
785         hash->page_no = (page_no & ~bt->hashmask);
786         bt_linkhash(bt, hash, page_no);
787
788         if( node = hash->lrunext = bt->lrufirst )
789                 node->lruprev = hash;
790         else
791                 bt->lrulast = hash;
792
793         bt->lrufirst = hash;
794
795 #ifdef unix
796         flag = PROT_READ | ( bt->mode == BT_ro ? 0 : PROT_WRITE );
797         hash->page = (BtPage)mmap (0, (bt->hashmask+1) << bt->page_bits, flag, MAP_SHARED, bt->idx, off);
798         if( (long long int)hash->page == -1LL )
799                 return bt->err = BTERR_map, (BtPage)NULL;
800
801 #else
802         flag = ( bt->mode == BT_ro ? PAGE_READONLY : PAGE_READWRITE );
803         hash->hmap = CreateFileMapping(bt->idx, NULL, flag,     (DWORD)(limit >> 32), (DWORD)limit, NULL);
804         if( !hash->hmap )
805                 return bt->err = BTERR_map, NULL;
806
807         flag = ( bt->mode == BT_ro ? FILE_MAP_READ : FILE_MAP_WRITE );
808         hash->page = MapViewOfFile(hash->hmap, flag, (DWORD)(off >> 32), (DWORD)off, (bt->hashmask+1) << bt->page_bits);
809         if( !hash->page )
810                 return bt->err = BTERR_map, NULL;
811 #endif
812
813         return (BtPage)((char*)hash->page + ((uint)(page_no & bt->hashmask) << bt->page_bits));
814 }
815
816 //      find or place requested page in page-cache
817 //      return memory address where page is located.
818
819 BtPage bt_hashpage(BtDb *bt, uid page_no)
820 {
821 BtHash *hash, *node, *next;
822 BtPage page;
823
824         // find page in cache and move to top of lru list  
825
826         if( hash = bt_findhash(bt, page_no) ) {
827                 page = (BtPage)((char*)hash->page + ((uint)(page_no & bt->hashmask) << bt->page_bits));
828                 // swap node in lru list
829                 if( node = hash->lruprev ) {
830                         if( next = node->lrunext = hash->lrunext )
831                                 next->lruprev = node;
832                         else
833                                 bt->lrulast = node;
834
835                         if( next = hash->lrunext = bt->lrufirst )
836                                 next->lruprev = hash;
837                         else
838                                 return bt->err = BTERR_hash, (BtPage)NULL;
839
840                         hash->lruprev = NULL;
841                         bt->lrufirst = hash;
842                 }
843                 return page;
844         }
845
846         // map pages and add to cache entry
847
848         if( bt->nodecnt < bt->nodemax ) {
849                 hash = bt->nodes + ++bt->nodecnt;
850                 return bt_linklru(bt, hash, page_no);
851         }
852
853         // hash table is already full, replace last lru entry from the cache
854
855         if( hash = bt->lrulast ) {
856                 // unlink from lru list
857                 if( node = bt->lrulast = hash->lruprev )
858                         node->lrunext = NULL;
859                 else
860                         return bt->err = BTERR_hash, (BtPage)NULL;
861
862 #ifdef unix
863                 munmap (hash->page, (bt->hashmask+1) << bt->page_bits);
864 #else
865                 FlushViewOfFile(hash->page, 0);
866                 UnmapViewOfFile(hash->page);
867                 CloseHandle(hash->hmap);
868 #endif
869                 // unlink from hash table
870
871                 bt_unlinkhash(bt, hash);
872
873                 // map and add to cache
874
875                 return bt_linklru(bt, hash, page_no);
876         }
877
878         return bt->err = BTERR_hash, (BtPage)NULL;
879 }
880
881 //  map a btree page onto current page
882
883 BTERR bt_mappage (BtDb *bt, BtPage *page, uid page_no)
884 {
885 off64_t off = page_no << bt->page_bits;
886 #ifndef unix
887 int amt[1];
888 #endif
889         
890         if( bt->mapped_io ) {
891                 bt->err = 0;
892                 *page = bt_hashpage(bt, page_no);
893                 return bt->err;
894         }
895 #ifdef unix
896         if ( pread(bt->idx, *page, bt->page_size, off) < bt->page_size )
897                 return bt->err = BTERR_map;
898 #else
899         SetFilePointer (bt->idx, (long)off, (long*)(&off)+1, FILE_BEGIN);
900
901         if( !ReadFile(bt->idx, *page, bt->page_size, amt, NULL) )
902                 return bt->err = BTERR_map;
903
904         if( *amt <  bt->page_size )
905                 return bt->err = BTERR_map;
906 #endif
907         return 0;
908 }
909
910 //      deallocate a deleted page 
911 //      place on free chain out of allocator page
912 //      page must already be BtLockXclusive and mapped
913
914 BTERR bt_freepage (BtDb *bt, BtPage page, uid page_no)
915 {
916         //      lock allocation page
917
918         if ( bt_lockpage(bt, ALLOC_page, BtLockUpdate) )
919                 return bt->err;
920
921         if( bt_mappage (bt, &bt->alloc, ALLOC_page) )
922                 return bt->err;
923
924         //      store chain in second right
925         bt_putid(page->right, bt_getid(bt->alloc[1].right));
926         bt_putid(bt->alloc[1].right, page_no);
927
928         if( bt_update(bt, bt->alloc, ALLOC_page) )
929                 return bt->err;
930         if( bt_update(bt, page, page_no) )
931                 return bt->err;
932
933         // unlock page zero 
934
935         if( bt_unlockpage(bt, ALLOC_page, BtLockUpdate) )
936                 return bt->err;
937
938         return 0;
939 }
940
941 //      allocate a new page and write page into it
942
943 uid bt_newpage(BtDb *bt, BtPage page)
944 {
945 uid new_page;
946 char *pmap;
947 int reuse;
948
949         // lock page zero
950
951         if ( bt_lockpage(bt, ALLOC_page, BtLockUpdate) )
952                 return 0;
953
954         if( bt_mappage (bt, &bt->alloc, ALLOC_page) )
955                 return 0;
956
957         // use empty chain first
958         // else allocate empty page
959
960         if( new_page = bt_getid(bt->alloc[1].right) ) {
961                 if( bt_mappage (bt, &bt->temp, new_page) )
962                         return 0;       // don't unlock on error
963                 bt_putid(bt->alloc[1].right, bt_getid(bt->temp->right));
964                 reuse = 1;
965         } else {
966                 new_page = bt_getid(bt->alloc->right);
967                 bt_putid(bt->alloc->right, new_page+1);
968                 reuse = 0;
969         }
970
971         if( bt_update(bt, bt->alloc, ALLOC_page) )
972                 return 0;       // don't unlock on error
973
974         // unlock page zero 
975
976         if ( bt_unlockpage(bt, ALLOC_page, BtLockUpdate) )
977                 return 0;
978
979         if( !bt->mapped_io ) {
980                 if( bt_update(bt, page, new_page) )
981                         return 0;       //don't unlock on error
982                 if ( bt_unlockpage(bt, ALLOC_page, BtLockWrite) )
983                         return 0;
984
985                 return new_page;
986         }
987
988 #ifdef unix
989         if ( pwrite(bt->idx, page, bt->page_size, new_page << bt->page_bits) < bt->page_size )
990                 return bt->err = BTERR_wrt, 0;
991
992         // if writing first page of hash block, zero last page in the block
993
994         if ( !reuse && bt->hashmask > 0 && (new_page & bt->hashmask) == 0 )
995         {
996                 // use temp buffer to write zeros
997                 memset(bt->temp, 0, bt->page_size);
998                 if ( pwrite(bt->idx,bt->temp, bt->page_size, (new_page | bt->hashmask) << bt->page_bits) < bt->page_size )
999                         return bt->err = BTERR_wrt, 0;
1000         }
1001 #else
1002         //      bring new page into page-cache and copy page.
1003         //      this will extend the file into the new pages.
1004
1005         if( !(pmap = (char*)bt_hashpage(bt, new_page & ~bt->hashmask)) )
1006                 return 0;
1007
1008         memcpy(pmap+((new_page & bt->hashmask) << bt->page_bits), page, bt->page_size);
1009 #endif
1010
1011         return new_page;
1012 }
1013
1014 //  find slot in given page for given key
1015
1016 int bt_findslot (BtPage page, unsigned char *key, uint len)
1017 {
1018 uint diff, higher = page->cnt, low = 1, slot;
1019 uint good = 0;
1020
1021         //      make last key an infinite fence value
1022
1023         if( !page->lvl || bt_getid (page->right) )
1024                 higher++;
1025         else
1026                 good++;
1027
1028         //      low is the next candidate, higher is already
1029         //      tested as .ge. the given key, loop ends when they meet
1030
1031         if( higher )
1032           while( diff = higher - low ) {
1033                 slot = low + ( diff >> 1 );
1034                 if( keycmp (keyptr(page, slot), key, len) < 0 )
1035                         low = slot + 1;
1036                 else
1037                         higher = slot, good++;
1038           }
1039
1040         //      return zero if key is beyond highkey value
1041         //      or page is empty
1042
1043         return good ? higher : 0;
1044 }
1045
1046 //  split full parent node
1047
1048 BTERR bt_splitparent (BtDb *bt, unsigned char *key, uint len)
1049 {
1050 uint cnt = 0, idx = 0, max, nxt = bt->page_size;
1051 uid parentpage = bt->parentpage, right;
1052 BtPage page = bt->parent;
1053 uid new_page;
1054 BtKey ptr;
1055
1056         //      upgrade parent latch to Xclusive
1057
1058         if( bt_lockpage (bt, bt->parentpage, BtLockUpgrade) )
1059                 return bt->err;
1060
1061         //  split higher half of keys to bt->frame
1062
1063         memset (bt->frame, 0, bt->page_size);
1064         max = (int)page->cnt;
1065         cnt = max / 2;
1066         idx = 0;
1067
1068         // link right sibling node into new right page
1069
1070         right = bt_getid (page->right);
1071         bt_putid(bt->frame->right, right);
1072
1073         //      record higher fence key in new right leaf page
1074
1075         if( bt->frame->fence = page->fence ) {
1076                 memcpy ((unsigned char *)bt->frame + bt->page_size - bt->frame->fence, (unsigned char *)(page) + bt->page_size - bt->frame->fence, bt->frame->fence);
1077                 nxt -= page->fence;
1078         }
1079
1080         while( cnt++ < max ) {
1081
1082                 // copy key, but not infinite values
1083
1084                 if( cnt < max || !page->lvl || right ) {
1085                         ptr = keyptr(page, cnt);
1086                         nxt -= ptr->len + 1;
1087                         memcpy ((unsigned char *)bt->frame + nxt, ptr, ptr->len + 1);
1088                 }
1089
1090                 // copy slot
1091
1092                 memcpy(slotptr(bt->frame,++idx)->id, slotptr(page,cnt)->id, BtId);
1093                 slotptr(bt->frame, idx)->tod = slotptr(page, cnt)->tod;
1094                 slotptr(bt->frame, idx)->off = nxt;
1095         }
1096
1097         bt->frame->bits = bt->page_bits;
1098         bt->frame->lvl = page->lvl;
1099         bt->frame->min = nxt;
1100         bt->frame->cnt = idx;
1101
1102         //      get new free page and write right frame to it.
1103
1104         if( !(new_page = bt_newpage(bt, bt->frame)) )
1105                 return bt->err;
1106
1107         //      update lower keys to continue in old page
1108
1109         memcpy (bt->frame, page, bt->page_size);
1110         memset (page+1, 0, bt->page_size - sizeof(*page));
1111         nxt = bt->page_size;
1112         max /= 2;
1113         cnt = 0;
1114         idx = 0;
1115
1116         //      record fence key in left leaf page
1117
1118         if( !page->lvl ) {
1119                 ptr = keyptr(bt->frame, max);
1120                 nxt -= ptr->len + 1;
1121                 memcpy ((unsigned char *)page + nxt, ptr, ptr->len + 1);
1122                 page->fence = ptr->len + 1;
1123         }
1124
1125         //  assemble page of smaller keys
1126         //      no infinite value to deal with
1127
1128         while( cnt++ < max ) {
1129                 ptr = keyptr(bt->frame, cnt);
1130                 nxt -= ptr->len + 1;
1131                 memcpy ((unsigned char *)page + nxt, ptr, ptr->len + 1);
1132                 memcpy(slotptr(page,++idx)->id, slotptr(bt->frame,cnt)->id, BtId);
1133                 slotptr(page, idx)->tod = slotptr(bt->frame, cnt)->tod;
1134                 slotptr(page, idx)->off = nxt;
1135         }
1136
1137         bt_putid(page->right, new_page);
1138         page->min = nxt;
1139         page->cnt = idx;
1140
1141         // update left node
1142
1143         if( bt_update(bt, page, parentpage) )
1144                 return bt->err;
1145
1146         //      decide to move to new right
1147         //      node or stay on left node
1148
1149         ptr = bt_highkey (bt, page);
1150
1151         if( keycmp (ptr, key, len) >= 0 )
1152                 return bt_unlockpage (bt, parentpage, BtLockUpgrade);
1153
1154         bt->parentpage = new_page;
1155
1156         if( bt_mappage (bt, &bt->parent, new_page) )
1157                 return bt->err;
1158         if( bt_lockpage (bt, new_page, BtLockUpdate) )
1159                 return bt->err;
1160         if( bt_unlockpage (bt, parentpage, BtLockUpgrade) )
1161                 return bt->err;
1162
1163         return bt_unlockpage (bt, parentpage, BtLockUpdate);
1164 }
1165
1166 //  add unlinked node key into parent
1167 //  childpage is existing record
1168 //      siblingpage is right record 
1169
1170 BTERR bt_parentlink (BtDb *bt, BtPage left, uid leftpage, BtKey rightkey, uid rightpage)
1171 {
1172 BtKey leftkey = bt_highkey (bt, left);
1173 BtPage page = bt->parent;
1174 uint slot, idx;
1175
1176         // upgrade parent latch to exclusive
1177
1178         if( bt_lockpage (bt, bt->parentpage, BtLockUpgrade) )
1179                 return bt->err;
1180
1181         // find the existing right high key in the parent
1182         // and fix the downlink to point to right page
1183
1184         if( rightkey ) {
1185           if( !(slot = bt_findslot (page, rightkey->key, rightkey->len)) )
1186                 return bt->err = BTERR_struct;
1187         } else
1188           slot = page->cnt;
1189
1190         bt_putid(slotptr(page,slot)->id, rightpage);
1191
1192         // calculate next available slot and copy left key onto page
1193
1194         page->min -= leftkey->len + 1; // reset lowest used offset
1195         memcpy ((unsigned char *)page + page->min, leftkey, leftkey->len + 1);
1196
1197         // now insert key into array before slot
1198
1199         idx = ++page->cnt;
1200
1201         while( idx > slot )
1202           *slotptr(page, idx) = *slotptr(page, idx -1), idx--;
1203
1204         bt_putid(slotptr(page,slot)->id, leftpage);
1205         slotptr(page, slot)->off = page->min;
1206
1207         if ( bt_update(bt, page, bt->parentpage) )
1208           return bt->err;
1209
1210         // downgrade parent page lock to BtLockUpdate
1211
1212         if( bt_unlockpage(bt, bt->parentpage, BtLockUpgrade) )
1213           return bt->err;
1214
1215         return 0;
1216 }
1217
1218 //      remove slot from parent
1219
1220 void bt_removeslot(BtDb *bt, uint slot)
1221 {
1222 uint nxt = bt->page_size, amt;
1223 BtPage page = bt->parent;
1224 uint cnt = 0, idx = 0;
1225 uint max = page->cnt;
1226 uid right;
1227 BtKey key;
1228
1229         memcpy (bt->frame, page, bt->page_size);
1230
1231         // skip page info and set rest of page to zero
1232         memset (page+1, 0, bt->page_size - sizeof(*page));
1233
1234         // copy fence key onto new page
1235         if( amt = page->fence ) {
1236                 nxt -= amt;
1237                 memcpy ((unsigned char *)page + nxt, (unsigned char *)(bt->frame) + nxt, amt);
1238         }
1239
1240         right = bt_getid (page->right);
1241
1242         while( cnt++ < max ) {
1243
1244                 // skip key to delete
1245
1246                 if( cnt == slot )
1247                         continue;
1248
1249                 // copy key, but not infinite value
1250
1251                 if( cnt < max || !page->lvl || right ) {
1252                         key = keyptr(bt->frame, cnt);
1253                         nxt -= key->len + 1;
1254                         memcpy ((unsigned char *)page + nxt, key, key->len + 1);
1255                 }
1256
1257                 // copy slot
1258                 memcpy(slotptr(page, ++idx)->id, slotptr(bt->frame, cnt)->id, BtId);
1259                 slotptr(page, idx)->tod = slotptr(bt->frame, cnt)->tod;
1260                 slotptr(page, idx)->off = nxt;
1261         }
1262         page->min = nxt;
1263         page->cnt = idx;
1264 }
1265
1266 //      unlink sibling node
1267
1268 BTERR bt_parentunlink (BtDb *bt, BtPage child, uid childpage)
1269 {
1270 BtKey parentkey = bt_highkey (bt, child);
1271 uint slot;
1272
1273         if( bt_lockpage (bt, bt->parentpage, BtLockUpgrade) )
1274                 return bt->err;
1275
1276         // delete child's slot from parent
1277
1278         if( slot = bt_findslot (bt->parent, parentkey->key, parentkey->len) )
1279                 bt_removeslot (bt, slot);
1280         else
1281                 return bt->err + BTERR_struct;
1282
1283         //      sibling now in the slot
1284
1285         bt_putid(slotptr(bt->parent,slot)->id, childpage);
1286
1287         if ( bt_update(bt, bt->parent, bt->parentpage) )
1288           return bt->err;
1289
1290         // unlock parent page completely
1291
1292         if( bt_unlockpage (bt, bt->parentpage, BtLockUpgrade) )
1293                 return bt->err;
1294
1295         return bt_unlockpage(bt, bt->parentpage, BtLockUpdate);
1296 }
1297
1298 //      merge right sibling page into child page
1299
1300 BTERR bt_mergepages (BtDb *bt, BtPage *right, uid rightpage)
1301 {
1302 BtPage *left = &bt->child;
1303 uint idx, amt;
1304 BtKey ptr;
1305
1306         if( bt_lockpage (bt, bt->childpage, BtLockUpgrade) )
1307                 return bt->err;
1308         if( bt_lockpage (bt, rightpage, BtLockUpgrade) )
1309                 return bt->err;
1310
1311         //      initialize empty frame
1312
1313         memset (bt->frame, 0, bt->page_size);
1314         *bt->frame = **right;
1315
1316         //      copy right fence key
1317
1318         if( amt = (*right)->fence )
1319                 memcpy ((unsigned char *)bt->frame + bt->page_size - amt, (unsigned char *)(*right) + bt->page_size - amt, amt);
1320
1321         bt->frame->min = bt->page_size - amt;
1322
1323         //      copy lowerkey key/values from left page
1324
1325         for( idx = 1; idx <= (*left)->cnt; idx++ ) {
1326                 ptr = keyptr(*left, idx);
1327                 bt->frame->min -= ptr->len + 1;
1328                 memcpy ((unsigned char *)bt->frame + bt->frame->min, ptr, ptr->len + 1);
1329                 memcpy (slotptr(bt->frame, idx)->id, slotptr(*left, idx)->id, BtId);
1330                 slotptr(bt->frame, idx)->tod = slotptr(*left, idx)->tod;
1331                 slotptr(bt->frame, idx)->off = bt->frame->min;
1332         }
1333
1334         bt->frame->cnt = (*left)->cnt;
1335
1336         //      copy higherkey key/values from right page
1337
1338         for( idx = 1; idx <= (*right)->cnt; idx++ ) {
1339
1340                 // copy key but not infinite value
1341
1342                 if( idx < (*right)->cnt || !(*right)->lvl || right ) {
1343                         ptr = keyptr(*right, idx);
1344                         bt->frame->min -= ptr->len + 1;
1345                         memcpy ((unsigned char *)bt->frame + bt->frame->min, ptr, ptr->len + 1);
1346                 }
1347
1348                 // copy slot
1349                 memcpy (slotptr(bt->frame, bt->frame->cnt + idx)->id, slotptr(*right, idx)->id, BtId);
1350                 slotptr(bt->frame, bt->frame->cnt + idx)->tod = slotptr(*right, idx)->tod;
1351                 slotptr(bt->frame, bt->frame->cnt + idx)->off = bt->frame->min;
1352         }
1353
1354         bt->frame->cnt += (*right)->cnt;
1355         memcpy (*left, bt->frame, bt->page_size);
1356
1357         if( bt_update (bt, *left, bt->childpage) )
1358                 return bt->err;
1359         if( bt_freepage (bt, *right, rightpage) )
1360                 return bt->err;
1361         if( bt_unlockpage (bt, rightpage, BtLockUpgrade) )
1362                 return bt->err;
1363         if( bt_unlockpage (bt, rightpage, BtLockUpdate) )
1364                 return bt->err;
1365
1366         return bt_unlockpage (bt, bt->childpage, BtLockUpgrade);
1367 }
1368
1369 //      redistribute right sibling page with child page
1370 //      switch child page to containing page
1371
1372 BTERR bt_redistribute (BtDb *bt, BtPage *right, uid rightpage, unsigned char *key, uint len)
1373 {
1374 uid siblingpage = bt_getid((*right)->right);
1375 BtPage *left = &bt->child;
1376 uint idx, cnt = 0, amt = 0;
1377 uint leftmax, rightmin;
1378 BtPage swap;
1379 BtKey ptr;
1380
1381         if( bt_lockpage (bt, bt->childpage, BtLockUpgrade) )
1382                 return bt->err;
1383         if( bt_lockpage (bt, rightpage, BtLockUpgrade) )
1384                 return bt->err;
1385
1386         //      initialize empty frames to contain redistributed pages
1387
1388         memset (bt->frame, 0, bt->page_size);   // new left page
1389         memset (bt->temp, 0, bt->page_size);    // new right page
1390
1391         *bt->frame = **left;
1392         *bt->temp = **right;
1393
1394         bt->frame->min = bt->page_size;
1395         bt->temp->min = bt->page_size;
1396         bt->frame->cnt = 0;
1397         bt->temp->cnt = 0;
1398
1399         //      find new left fence index
1400         //      and copy left fence key
1401
1402         if( (*left)->cnt > (*right)->cnt ) {
1403                 rightmin = 0;
1404                 leftmax = (*left)->cnt / 2;
1405                 if( !bt->frame->lvl ) {
1406                         ptr = keyptr(*left, leftmax);
1407                         bt->frame->fence = ptr->len + 1;
1408                         bt->frame->min -= ptr->len + 1;
1409                         memcpy ((unsigned char *)bt->frame + bt->frame->min, ptr, ptr->len + 1);
1410                 }
1411         } else {
1412                 leftmax = (*left)->cnt;
1413                 rightmin = (*right)->cnt / 2;
1414                 if( !bt->frame->lvl ) {
1415                         ptr = keyptr(*right, rightmin);
1416                         bt->frame->fence = ptr->len + 1;
1417                         bt->frame->min -= ptr->len + 1;
1418                         memcpy ((unsigned char *)bt->frame + bt->frame->min, ptr, ptr->len + 1);
1419                 }
1420         }
1421
1422         //      right fence stays the same, if any
1423
1424         if( amt = (*right)->fence ) {
1425                 bt->temp->min -= amt;
1426                 memcpy ((unsigned char *)bt->temp + bt->temp->min, (unsigned char *)(*right) + bt->temp->min, amt);
1427         }
1428
1429         //      copy first set of lowerkey key/values from left page
1430
1431         for( idx = 1; idx <= leftmax; idx++ ) {
1432                 bt->frame->cnt++;
1433                 ptr = keyptr(*left, idx);
1434                 bt->frame->min -= ptr->len + 1;
1435                 memcpy ((unsigned char *)bt->frame + bt->frame->min, ptr, ptr->len + 1);
1436                 memcpy (slotptr(bt->frame, idx)->id, slotptr(*left, idx)->id, BtId);
1437                 slotptr(bt->frame, idx)->tod = slotptr(*left, idx)->tod;
1438                 slotptr(bt->frame, idx)->off = bt->frame->min;
1439         }
1440
1441         //      copy remaining left page key/values from right page, if any
1442
1443         for( idx = 1; idx <= rightmin; idx++ ) {
1444                 bt->frame->cnt++;
1445                 ptr = keyptr(*right, idx);
1446                 bt->frame->min -= ptr->len + 1;
1447                 memcpy ((unsigned char *)bt->frame + bt->frame->min, ptr, ptr->len + 1);
1448                 memcpy (slotptr(bt->frame, bt->frame->cnt)->id, slotptr(*right, idx)->id, BtId);
1449                 slotptr(bt->frame, bt->frame->cnt)->tod = slotptr(*right, idx)->tod;
1450                 slotptr(bt->frame, bt->frame->cnt)->off = bt->frame->min;
1451         }
1452
1453         //      copy remaining left page key/values into new right page, if any
1454
1455         for( idx = leftmax; idx <= (*left)->cnt; idx++ ) {
1456                 bt->temp->cnt++;
1457                 ptr = keyptr(*left, idx);
1458                 bt->temp->min -= ptr->len + 1;
1459                 memcpy ((unsigned char *)bt->temp + bt->temp->min, ptr, ptr->len + 1);
1460                 memcpy (slotptr(bt->temp, bt->temp->cnt)->id, slotptr(*left, idx)->id, BtId);
1461                 slotptr(bt->temp, bt->temp->cnt)->tod = slotptr(*left, idx)->tod;
1462                 slotptr(bt->temp, bt->temp->cnt)->off = bt->temp->min;
1463         }
1464
1465         //      copy rest of higherkey key/values from right page
1466
1467         for( idx = rightmin; idx <= (*right)->cnt; idx++ ) {
1468
1469                 // copy key, but not infinite value
1470
1471                 if( idx < (*right)->cnt || !(*right)->lvl || siblingpage ) {
1472                         ptr = keyptr(*right, idx);
1473                         bt->temp->min -= ptr->len + 1;
1474                         memcpy ((unsigned char *)bt->temp + bt->temp->min, ptr, ptr->len + 1);
1475                 }
1476
1477                 //      copy slot
1478
1479                 bt->temp->cnt++;
1480                 memcpy (slotptr(bt->temp, bt->temp->cnt)->id, slotptr(*right, idx)->id, BtId);
1481                 slotptr(bt->temp, bt->temp->cnt)->tod = slotptr(*right, idx)->tod;
1482                 slotptr(bt->temp, bt->temp->cnt)->off = bt->temp->min;
1483         }
1484
1485         memcpy (*left, bt->frame, bt->page_size);
1486         memcpy (*right, bt->temp, bt->page_size);
1487
1488         if( bt_update (bt, *left, bt->childpage) )
1489                 return bt->err;
1490         if( bt_update (bt, *right, rightpage) )
1491                 return bt->err;
1492
1493         if( bt_unlockpage (bt, rightpage, BtLockUpgrade) )
1494                 return bt->err;
1495         if( bt_unlockpage (bt, rightpage, BtLockUpdate) )
1496                 return bt->err;
1497
1498         ptr = bt_highkey (bt, *left);
1499
1500         //      decide which page is the child page
1501         //      if leftkey >= our key, go with left
1502
1503         if( keycmp (ptr, key, len) >= 0 ) {
1504                 if( bt_unlockpage (bt, bt->childpage, BtLockUpgrade) )
1505                         return bt->err;
1506                 if( bt_unlockpage (bt, rightpage, BtLockUpgrade) )
1507                         return bt->err;
1508                 if( bt_unlockpage (bt, rightpage, BtLockUpdate) )
1509                         return bt->err;
1510         } else {
1511                 if( bt_unlockpage (bt, rightpage, BtLockUpgrade) )
1512                         return bt->err;
1513                 if( bt_unlockpage (bt, bt->childpage, BtLockUpgrade) )
1514                         return bt->err;
1515                 if( bt_unlockpage (bt, bt->childpage, BtLockUpdate) )
1516                         return bt->err;
1517                 swap = bt->child;
1518                 bt->child = *right;
1519                 *left = swap;
1520                 bt->childpage = rightpage;
1521         }
1522
1523         return 0;
1524 }
1525
1526 //      lower the root level by removing the child node
1527
1528 BTERR bt_lowerroot(BtDb *bt)
1529 {
1530         if( bt_lockpage (bt, ROOT_page, BtLockUpgrade) )
1531                 return bt->err;
1532
1533         if( bt_lockpage (bt, bt->childpage, BtLockUpgrade) )
1534                 return bt->err;
1535
1536         memcpy (bt->parent, bt->child, bt->page_size);
1537
1538         if( bt_update(bt, bt->parent, ROOT_page) )
1539                 return bt->err;
1540
1541         if( bt_freepage(bt, bt->child, bt->childpage) )
1542                 return bt->err;
1543
1544         if( bt_unlockpage (bt, bt->childpage, BtLockUpgrade) )
1545                 return bt->err;
1546
1547         if( bt_unlockpage (bt, bt->childpage, BtLockUpdate) )
1548                 return bt->err;
1549
1550         return bt_unlockpage (bt, ROOT_page, BtLockUpgrade);
1551 }
1552
1553 // split the root and raise the height of the btree
1554 //      return with parent page set to appropriate sibling
1555
1556 BTERR bt_raiseroot(BtDb *bt, uid sibling, unsigned char *key, uint len)
1557 {
1558 unsigned char lowerkey[256];
1559 uint nxt = bt->page_size;
1560 BtPage root = bt->parent;
1561 uid new_page;
1562 BtKey ptr;
1563
1564         //      upgrade root page lock to exclusive
1565
1566         if( bt_lockpage (bt, ROOT_page, BtLockUpgrade) )
1567                 return bt->err;
1568
1569         //  Obtain an empty page to use, and copy the current
1570         //  root (lower half) contents into it
1571
1572         if( !(new_page = bt_newpage(bt, root)) )
1573                 return bt->err;
1574
1575         if( bt_lockpage (bt, new_page, BtLockUpdate) )
1576                 return bt->err;
1577
1578         //      save high fence key for left page
1579
1580         ptr = bt_highkey(bt, root);
1581         memcpy (lowerkey, ptr, ptr->len + 1);
1582
1583         // preserve the page info at the bottom
1584         // and set rest to zero to initialize new root page
1585
1586         memset(root+1, 0, bt->page_size - sizeof(*root));
1587
1588         // insert left key in newroot page
1589
1590         nxt -= *lowerkey + 1;
1591         memcpy ((unsigned char *)root + nxt, lowerkey, *lowerkey + 1);
1592         bt_putid(slotptr(root, 1)->id, new_page);
1593         slotptr(root, 1)->off = nxt;
1594         
1595         // insert second (infinite) key on newroot page that's never examined
1596         // and increase the root level
1597
1598         bt_putid(slotptr(root, 2)->id, sibling);
1599         bt_putid(root->right, 0);
1600
1601         root->min = nxt;                // reset lowest used offset and key count
1602         root->fence = 0;
1603         root->cnt = 2;
1604         root->lvl++;
1605
1606         if( bt_update(bt, root, bt->parentpage) )
1607                 return bt->err;
1608
1609         if( bt_unlockpage(bt, ROOT_page, BtLockUpgrade) )
1610                 return bt->err;
1611
1612         if( bt_unlockpage(bt, ROOT_page, BtLockUpdate) )
1613                 return bt->err;
1614
1615         //      decide which root node to continue with
1616         //              sibling has upper keys, newpage the lower ones
1617
1618         if( keycmp((BtKey)lowerkey, key, len) < 0 ) {
1619                 bt->parentpage = sibling;               // go with the upper ones
1620
1621                 if( bt_unlockpage (bt, new_page, BtLockUpdate) )
1622                         return bt->err;
1623
1624                 return bt_mappage (bt, &bt->parent, sibling);
1625         }
1626
1627         bt->parentpage = new_page;                      // go with the lower ones
1628
1629         if( bt_unlockpage (bt, sibling, BtLockUpdate) )
1630                 return bt->err;
1631
1632         return bt_mappage (bt, &bt->parent, new_page);
1633 }
1634
1635 //      handle underflowing child node
1636
1637 BTERR bt_repairchild (BtDb *bt, uint parentslot, unsigned char *key, uint len)
1638 {
1639 BtKey parentkey = bt_slotkey(bt->parent, parentslot);
1640 BtKey fencekey = bt_highkey (bt, bt->child);
1641 BtKey highkey = bt_highkey(bt, bt->parent);
1642 BtKey siblingkey, siblingkey2;
1643 uid siblingpage, siblingpage2;
1644 uid swappage;
1645 BtPage swap;
1646 uint slot;
1647
1648   // high key is never NULL
1649   // fence key is null on right end
1650
1651   if( fencekey )
1652    if( !highkey || keycmp (fencekey, highkey->key, highkey->len) < 0 ) {
1653
1654         //  childpage is not rightmost child of parent page
1655
1656         siblingpage = bt_getid(bt->child->right);
1657
1658         if( bt_lockpage (bt, siblingpage, BtLockUpdate) )
1659           return bt->err;
1660
1661         if( bt_mappage (bt, &bt->sibling, siblingpage) )
1662           return bt->err;
1663
1664         if( !parentkey || keycmp (fencekey, parentkey->key, parentkey->len) < 0 ) {
1665
1666           // sibling is not linked in parent, so we can merge it
1667
1668           if( bt_unlockpage (bt, bt->parentpage, BtLockUpdate) )
1669                 return bt->err;
1670
1671           // calculate size of merged page by adding single child key to sibling
1672
1673           fencekey = keyptr(bt->child, 1);
1674
1675           if( bt->sibling->min < (bt->sibling->cnt + 1) * sizeof(BtSlot) + sizeof(*bt->sibling) + fencekey->len + 1)
1676             return bt_redistribute (bt, &bt->sibling, siblingpage, key, len);
1677           else
1678             return bt_mergepages (bt, &bt->sibling, siblingpage);
1679     }
1680
1681         //      sibling has a key in the parent node
1682         //      find its slot in parent
1683
1684         if( siblingkey = bt_highkey (bt, bt->sibling) )
1685           slot = bt_findslot (bt->parent, siblingkey->key, siblingkey->len);
1686         else
1687           slot = bt->parent->cnt;
1688
1689         parentkey = bt_slotkey (bt->parent, slot);
1690         siblingpage2 = bt_getid(bt->sibling->right);
1691
1692         if( siblingkey && parentkey && keycmp (siblingkey, parentkey->key, parentkey->len) < 0 ) {
1693
1694           //  sibling2 is not linked in P, its key is parentkey
1695           //  can parent support its insertion?
1696           //  if not, split parent node first.
1697
1698           if( bt->parent->min < (bt->parent->cnt + 1) * sizeof(BtSlot) + sizeof(*bt->parent) + parentkey->len + 1) {
1699                 if( bt_splitparent (bt, key, len) )
1700                   return bt->err;
1701
1702                 //  are we in the correct half of new parent nodes?
1703                 //  if not, restart the function.
1704
1705                 if( slot = bt_findslot (bt->parent, siblingkey->key, siblingkey->len) )
1706                   parentkey = keyptr(bt->parent, slot);
1707                 else
1708                   return BTERR_restart;
1709           }
1710
1711           //  link right of sibling into parent
1712
1713           if( bt_parentlink (bt, bt->sibling, siblingpage, parentkey, siblingpage2) )
1714                 return bt->err;
1715
1716           //  unlink sibling from parent
1717
1718           if( bt_parentunlink (bt, bt->child, bt->childpage) )
1719                 return bt->err;
1720
1721           fencekey = keyptr(bt->child, 1);
1722
1723           if( bt->sibling->min < (bt->sibling->cnt + 1) * sizeof(BtSlot) + sizeof(*bt->sibling) + fencekey->len + 1)
1724             return bt_redistribute (bt, &bt->sibling, siblingpage, key, len);
1725           else
1726             return bt_mergepages (bt, &bt->sibling, siblingpage);
1727         } else {
1728
1729           // unlink sibling from parent and merge
1730
1731           if( bt_parentunlink (bt, bt->child, bt->childpage) )
1732                 return bt->err;
1733
1734           fencekey = keyptr(bt->child, 1);
1735
1736           if( bt->sibling->min < (bt->sibling->cnt + 1) * sizeof(BtSlot) + sizeof(*bt->sibling) + fencekey->len + 1)
1737             return bt_redistribute (bt, &bt->sibling, siblingpage, key, len);
1738           else
1739             return bt_mergepages (bt, &bt->sibling, siblingpage);
1740         }
1741   }
1742
1743   //  child is rightmost key in the parent,
1744   //  work with nodes to left.
1745
1746   siblingpage = bt_getid(slotptr(bt->parent, parentslot - 1)->id);
1747   siblingkey = keyptr(bt->parent, parentslot - 1);
1748
1749   if( bt_unlockpage (bt, bt->childpage, BtLockUpdate) )
1750         return bt->err;
1751   if( bt_lockpage (bt, siblingpage, BtLockUpdate) )
1752         return bt->err;
1753   if( bt_mappage (bt, &bt->sibling, siblingpage) )
1754         return bt->err;
1755
1756   siblingpage2 = bt_getid(bt->sibling->right);
1757   siblingkey2 = bt_highkey (bt, bt->sibling);
1758
1759   if( bt_lockpage (bt, siblingpage2, BtLockUpdate) )
1760         return bt->err;
1761   if( bt_mappage (bt, &bt->sibling2, siblingpage2) )
1762         return bt->err;
1763
1764   if( keycmp (siblingkey, siblingkey2->key, siblingkey2->len) == 0 ) {
1765
1766         //      left sibling right (mapped into sibling2) is our child node
1767
1768         swap = bt->sibling2;
1769         bt->sibling2 = bt->child;
1770         bt->child = swap;
1771
1772         //  does child still need to merge/redistribute?
1773
1774         if( bt->child->cnt > 1 ) {
1775           if( bt_unlockpage (bt, bt->parentpage, BtLockUpdate) )
1776                 return bt->err;
1777
1778           return bt_unlockpage (bt, siblingpage, BtLockUpdate);
1779         }
1780         
1781         swap = bt->sibling;
1782         bt->sibling = bt->child;
1783         bt->child = swap;
1784
1785         bt->childpage = siblingpage;
1786         siblingpage = siblingpage2;
1787
1788         // unlink sibling node from parent and merge with child
1789
1790         if( bt_parentunlink (bt, bt->child, bt->childpage) )
1791                 return bt->err;
1792
1793         fencekey = keyptr(bt->sibling, 1);
1794
1795         if( bt->child->min < (bt->child->cnt + 1) * sizeof(BtSlot) + sizeof(*bt->sibling) + fencekey->len + 1)
1796           return bt_redistribute (bt, &bt->sibling, siblingpage, key, len);
1797         else
1798           return bt_mergepages (bt, &bt->sibling, siblingpage);
1799   }
1800
1801   //  currently unlinked sibling2 merges with child
1802
1803   fencekey = bt_highkey (bt, bt->sibling2);
1804
1805   if( bt->parent->min < (bt->parent->cnt + 1) * sizeof(BtSlot) + sizeof(*bt->parent) + fencekey->len + 1)
1806         if( bt_splitparent (bt, key, len) )
1807           return bt->err;
1808
1809   if( bt_parentlink (bt, bt->sibling, siblingpage, fencekey, siblingpage2) )
1810         return bt->err;
1811
1812   if( bt_unlockpage (bt, siblingpage, BtLockUpdate) )
1813           return bt->err;
1814   if( bt_lockpage (bt, bt->childpage, BtLockUpdate) )
1815           return bt->err;
1816   if( bt_mappage (bt, &bt->child, bt->childpage) )
1817         return bt->err;
1818
1819   if( bt->child->cnt > 1 ) {
1820     if( bt_unlockpage (bt, bt->parentpage, BtLockUpdate) )
1821           return bt->err;
1822     return bt_unlockpage (bt, siblingpage2, BtLockUpdate);
1823   }
1824
1825   // unlink child from parent node leaving immediate
1826   // left sibling2 to accept merge/redistribution
1827
1828   if( bt_parentunlink (bt, bt->sibling2, siblingpage2) )
1829         return bt->err;
1830
1831   swap = bt->sibling2;
1832   bt->sibling2 = bt->child;
1833   bt->child = swap;
1834   bt->childpage = siblingpage2;
1835
1836   fencekey = keyptr(bt->sibling2, 1);
1837
1838   if( bt->child->min < (bt->child->cnt + 1) * sizeof(BtSlot) + sizeof(*bt->child) + fencekey->len + 1) 
1839         return bt_redistribute (bt, &bt->sibling2, siblingpage2, key, len);
1840   else
1841         return bt_mergepages (bt, &bt->sibling2, siblingpage2);
1842 }
1843
1844 //  find and load leaf page for given key
1845 //      leave page BtLockShared, return with key's slot
1846
1847 int bt_loadpageread (BtDb *bt, unsigned char *key, uint len)
1848 {
1849 uid page_no = ROOT_page, prevpage = 0;
1850 uint slot, mode;
1851
1852   //  start at root of btree and drill down
1853
1854   do {
1855         bt->parentpage = page_no;
1856
1857         if( bt_lockpage(bt, bt->parentpage, BtLockShared) )
1858                 return 0;                                                                       
1859
1860         if( prevpage )
1861           if( bt_unlockpage(bt, prevpage, BtLockShared) )
1862                 return 0;
1863
1864         //      map/obtain page contents
1865
1866         if( bt_mappage (bt, &bt->parent, page_no) )
1867                 return 0;
1868
1869         //  find key on page at this level
1870         //      return if leaf page
1871
1872         if( (slot = bt_findslot (bt->parent, key, len)) ) {
1873           if( !bt->parent->lvl )
1874                 return slot;
1875
1876           // continue down to next level
1877
1878           page_no = bt_getid(slotptr(bt->parent, slot)->id);
1879         }
1880
1881         //  or slide right into next page
1882
1883         else
1884                 page_no = bt_getid(bt->parent->right);
1885
1886         prevpage = bt->parentpage;
1887   } while( page_no );
1888
1889   // return EOF on end of right chain
1890
1891   if( bt_unlockpage(bt, bt->parentpage, BtLockShared) )
1892         return 0;                                                                       
1893
1894   return 0;     // return EOF
1895 }
1896
1897 //  find and load leaf page for given key
1898 //      return w/slot # on leaf page
1899 //      leave page BtLockUpdate
1900
1901 int bt_loadpageupdate (BtDb *bt, unsigned char *key, uint len)
1902 {
1903 uid parentpage = 0, nextpage;
1904 BtKey fencekey, parentkey;
1905 uint slot, mode;
1906 BtPage swap;
1907
1908   //  start at root of btree and drill down
1909
1910   if( bt_lockpage(bt, ROOT_page, BtLockUpdate) )
1911         return 0;                                                                       
1912
1913   //    map/obtain page contents
1914
1915   if( bt_mappage (bt, &bt->parent, ROOT_page) )
1916         return 0;
1917
1918   bt->parentpage = ROOT_page;
1919
1920   do {
1921         // if root page, check for tree level growth
1922         // by existence of right pointer
1923
1924         if( bt->parentpage == ROOT_page )
1925           if( nextpage = bt_getid(bt->parent->right) ) {
1926                 if( bt_lockpage (bt, nextpage, BtLockUpdate) )
1927                   return 0;
1928
1929                 if( bt_raiseroot (bt, nextpage, key, len) )
1930                   return 0;
1931           }
1932
1933         //  find key on page at this level
1934         //      return if leaf page
1935
1936         slot = bt_findslot (bt->parent, key, len);
1937
1938         if( !bt->parent->lvl )
1939                 return slot;
1940
1941         // lock & map the child
1942
1943         bt->childpage = bt_getid(slotptr(bt->parent, slot)->id);
1944
1945         if( bt_lockpage(bt, bt->childpage, BtLockUpdate) )
1946                 return 0;                                                                       
1947
1948         if( bt_mappage (bt, &bt->child, bt->childpage) )
1949                 return 0;
1950
1951         //  check child for underflow
1952
1953         if( bt->parentpage == ROOT_page )
1954           if( bt->parent->cnt == 1 )
1955            if( !bt_getid(bt->child->right) ) {
1956                 if( bt_lowerroot (bt) )
1957                         return 0;
1958                 nextpage = ROOT_page;
1959                 continue;
1960            }
1961
1962         if( bt->child->cnt == 1 ) {
1963                 while( bt_repairchild (bt, slot, key, len) == BTERR_restart );
1964                 if( bt->err )
1965                         return 0;
1966
1967             swap = bt->child;
1968             bt->child = bt->parent;
1969             bt->parent = swap;
1970             nextpage = bt->childpage;
1971                 continue;
1972         }
1973
1974         fencekey = bt_highkey (bt, bt->child);
1975         parentkey = bt_slotkey(bt->parent, slot);
1976
1977         //  if right sibling is not linked,
1978         //      fix links in parent node
1979
1980         if( fencekey )
1981          if( !parentkey || keycmp (fencekey, parentkey->key, parentkey->len) < 0 ) {
1982
1983           if( bt->parent->min < (bt->parent->cnt + 1) * sizeof(BtSlot) + sizeof(*bt->parent) + fencekey->len + 1)
1984                 if( bt_splitparent (bt, key, len) ) {
1985                         return 0;
1986                 } else {
1987                         slot = bt_findslot (bt->parent, key, len);
1988                         parentkey = bt_slotkey(bt->parent, slot);
1989                 }
1990
1991           //  add key to parent for child page
1992           //    and fix downlink for childpage
1993
1994           nextpage = bt_getid(bt->child->right);
1995
1996           if( bt_parentlink (bt, bt->child, bt->childpage, parentkey, nextpage) )
1997                 return 0;
1998          }
1999
2000         //  unlock the parent page
2001
2002         if( bt_unlockpage (bt, bt->parentpage, BtLockUpdate) )
2003                 return 0;
2004
2005         //      is our key on this child page?
2006         //      is fencekey infinite, or .ge. our key
2007
2008         if( !fencekey || keycmp (fencekey, key, len) >= 0 ) {
2009           swap = bt->child;
2010           bt->child = bt->parent;
2011           bt->parent = swap;
2012           nextpage = bt->childpage;
2013           continue;
2014         }
2015
2016         //  otherwise slide right into next page
2017
2018         nextpage = bt_getid(bt->child->right);
2019
2020         if( bt_lockpage (bt, nextpage, BtLockUpdate) )
2021                 return 0;
2022
2023         if( bt_unlockpage (bt, bt->childpage, BtLockUpdate) )
2024                 return 0;
2025
2026         if( bt_mappage (bt, &bt->parent, nextpage) )
2027                 return 0;
2028
2029   } while( bt->parentpage = nextpage );
2030
2031   // return error on end of right chain
2032
2033   bt->err = BTERR_struct;
2034   return 0;     // return error
2035 }
2036
2037 //  find and delete key on leaf page
2038
2039 BTERR bt_deletekey (BtDb *bt, unsigned char *key, uint len)
2040 {
2041 uid page_no, right;
2042 uint slot, tod;
2043 BtKey ptr;
2044
2045         bt_resetpages (bt);
2046
2047         if( slot = bt_loadpageupdate (bt, key, len) )
2048                 ptr = keyptr(bt->parent, slot);
2049         else if( bt->err )
2050                 return bt->err;
2051
2052         // if key is found delete it, otherwise ignore request
2053
2054         if( slot && !keycmp (ptr, key, len) ) {
2055           if( bt_lockpage (bt, bt->parentpage, BtLockUpgrade) )
2056                 return bt->err;
2057
2058           bt_removeslot (bt, slot);
2059
2060           if( bt_update(bt, bt->parent, bt->parentpage) )
2061                 return bt->err;
2062
2063           if( bt_unlockpage (bt, bt->parentpage, BtLockUpgrade) )
2064                 return bt->err;
2065         }
2066
2067         return bt_unlockpage (bt, bt->parentpage, BtLockUpdate);
2068 }
2069
2070 //      find key in leaf page and return row-id
2071 //      or zero if key is not found.
2072
2073 uid bt_findkey (BtDb *bt, unsigned char *key, uint len)
2074 {
2075 uint  slot;
2076 BtKey ptr;
2077 uid id;
2078
2079         bt_resetpages (bt);
2080
2081         if( slot = bt_loadpageread (bt, key, len) )
2082                 ptr = keyptr(bt->parent, slot);
2083         else
2084                 return 0;
2085
2086         // if key exists, return row-id
2087         //      otherwise return 0
2088
2089         if( ptr->len == len && !memcmp (ptr->key, key, len) )
2090                 id = bt_getid(slotptr(bt->parent,slot)->id);
2091         else
2092                 id = 0;
2093
2094         if ( bt_unlockpage(bt, bt->parentpage, BtLockShared) )
2095                 return 0;
2096
2097         return id;
2098 }
2099
2100 //  Insert new key into the btree leaf page
2101
2102 BTERR bt_insertkey (BtDb *bt, unsigned char *key, uint len, uid id, uint tod)
2103 {
2104 uint slot, idx;
2105 BtKey ptr;
2106
2107         if( slot = bt_loadpageupdate (bt, key, len) )
2108                 ptr = keyptr(bt->parent, slot);
2109         else if( bt->err )
2110                 return bt->err;
2111
2112         if( bt->parent->lvl )
2113                 abort();
2114         if( bt->parent->lvl )
2115                 abort();
2116
2117
2118         // if key already exists, update id and return
2119
2120         if( slot )
2121           if( !keycmp (ptr, key, len) ) {
2122                 if( bt_lockpage (bt, bt->parentpage, BtLockUpgrade) )
2123                         return bt->err;
2124                 slotptr(bt->parent, slot)->tod = tod;
2125                 bt_putid(slotptr(bt->parent,slot)->id, id);
2126                 if ( bt_update(bt, bt->parent, bt->parentpage) )
2127                         return bt->err;
2128                 if( bt_unlockpage (bt, bt->parentpage, BtLockUpgrade) )
2129                         return bt->err;
2130                 return bt_unlockpage(bt, bt->parentpage, BtLockUpdate);
2131           }
2132
2133         // check if leaf page has enough space
2134
2135         if( bt->parent->min < (bt->parent->cnt + 1) * sizeof(BtSlot) + sizeof(*bt->parent) + len + 1) {
2136           if( bt_splitparent (bt, key, len) )
2137                 return bt->err;
2138
2139           slot = bt_findslot (bt->parent, key, len);
2140         }
2141
2142         // calculate next available slot and copy key into page
2143
2144         if( bt_lockpage (bt, bt->parentpage, BtLockUpgrade) )
2145                 return bt->err;
2146
2147         bt->parent->min -= len + 1; // reset lowest used offset
2148         ((unsigned char *)bt->parent)[bt->parent->min] = len;
2149         memcpy ((unsigned char *)bt->parent + bt->parent->min +1, key, len );
2150
2151         // now insert key into array before slot
2152
2153         idx = ++bt->parent->cnt;
2154
2155         if( slot )
2156          while( idx > slot )
2157           *slotptr(bt->parent, idx) = *slotptr(bt->parent, idx -1), idx--;
2158
2159         bt_putid(slotptr(bt->parent,idx)->id, id);
2160         slotptr(bt->parent, idx)->off = bt->parent->min;
2161         slotptr(bt->parent, idx)->tod = tod;
2162
2163         if ( bt_update(bt, bt->parent, bt->parentpage) )
2164           return bt->err;
2165
2166         if( bt_unlockpage (bt, bt->parentpage, BtLockUpgrade) )
2167                 return bt->err;
2168
2169         return bt_unlockpage(bt, bt->parentpage, BtLockUpdate);
2170 }
2171
2172 //  cache page of keys into cursor and return starting slot for given key
2173
2174 uint bt_startkey (BtDb *bt, unsigned char *key, uint len)
2175 {
2176 uint slot;
2177
2178         // cache page for retrieval
2179         if( slot = bt_loadpageread (bt, key, len) )
2180                 memcpy (bt->cursor, bt->parent, bt->page_size);
2181         bt->cursorpage = bt->parentpage;
2182         if ( bt_unlockpage(bt, bt->parentpage, BtLockShared) )
2183                 return 0;
2184
2185         return slot;
2186 }
2187
2188 //  return next slot for cursor page
2189 //  or slide cursor right into next page
2190
2191 uint bt_nextkey (BtDb *bt, uint slot)
2192 {
2193 off64_t right;
2194
2195   do {
2196         if( slot++ < bt->cursor->cnt )
2197                 return slot;
2198
2199         right = bt_getid(bt->cursor->right);
2200
2201         if( !right )
2202                 break;
2203
2204         bt->cursorpage = right;
2205
2206     if( bt_lockpage(bt, right,BtLockShared) )
2207                 return 0;
2208
2209         if( bt_mappage (bt, &bt->parent, right) )
2210                 break;
2211
2212         memcpy (bt->cursor, bt->parent, bt->page_size);
2213         if ( bt_unlockpage(bt, right, BtLockShared) )
2214                 return 0;
2215
2216         slot = 0;
2217   } while( 1 );
2218
2219   return bt->err = 0;
2220 }
2221
2222 BtKey bt_key(BtDb *bt, uint slot)
2223 {
2224         return keyptr(bt->cursor, slot);
2225 }
2226
2227 uid bt_uid(BtDb *bt, uint slot)
2228 {
2229         return bt_getid(slotptr(bt->cursor,slot)->id);
2230 }
2231
2232 uint bt_tod(BtDb *bt, uint slot)
2233 {
2234         return slotptr(bt->cursor,slot)->tod;
2235 }
2236
2237
2238 #ifdef STANDALONE
2239 //  standalone program to index file of keys
2240 //  then list them onto std-out
2241
2242 int main (int argc, char **argv)
2243 {
2244 uint slot, found = 0, line = 0, off = 0;
2245 int ch, cnt = 0, bits = 12;
2246 unsigned char key[256];
2247 clock_t done, start;
2248 time_t tod[1];
2249 uint scan = 0;
2250 uint len = 0;
2251 uint map = 0;
2252 BtKey ptr;
2253 BtDb *bt;
2254 FILE *in;
2255
2256         if( argc < 4 ) {
2257                 fprintf (stderr, "Usage: %s idx_file src_file Read/Write/Scan/Delete/Find [page_bits mapped_pool_pages start_line_number]", argv[0]);
2258                 exit(0);
2259         }
2260
2261         start = clock();
2262         time (tod);
2263
2264         if( argc > 4 )
2265                 bits = atoi(argv[4]);
2266
2267         if( argc > 5 )
2268                 map = atoi(argv[5]);
2269
2270         if( map > 65536 )
2271                 fprintf (stderr, "Warning: mapped_pool > 65536 pages\n");
2272
2273         if( argc > 6 )
2274                 off = atoi(argv[6]);
2275
2276         bt = bt_open ((argv[1]), BT_rw, bits, map);
2277
2278         if( !bt ) {
2279                 fprintf(stderr, "Index Open Error %s\n", argv[1]);
2280                 exit (1);
2281         }
2282
2283         switch(argv[3][0]| 0x20)
2284         {
2285         case 'w':
2286                 fprintf(stderr, "started indexing for %s\n", argv[2]);
2287                 if( argc > 2 && (in = fopen (argv[2], "rb")) )
2288                   while( ch = getc(in), ch != EOF )
2289                         if( ch == '\n' )
2290                         {
2291                           if( off )
2292                                 sprintf((char *)key+len, "%.9d", line + off), len += 9;
2293
2294                           if( bt_insertkey (bt, key, len, ++line, *tod) )
2295                                 fprintf(stderr, "Error %d Syserr %d Line: %d\n", bt->err, errno, line), exit(0);
2296                           len = 0;
2297                         }
2298                         else if( len < 245 )
2299                                 key[len++] = ch;
2300                 fprintf(stderr, "finished adding keys, %d \n", line);
2301                 break;
2302
2303         case 'd':
2304                 fprintf(stderr, "started deleting keys for %s\n", argv[2]);
2305                 if( argc > 2 && (in = fopen (argv[2], "rb")) )
2306                   while( ch = getc(in), ch != EOF )
2307                         if( ch == '\n' )
2308                         {
2309                           if( off )
2310                                 sprintf((char *)key+len, "%.9d", line + off), len += 9;
2311                           line++;
2312                           if( bt_deletekey (bt, key, len) )
2313                                 fprintf(stderr, "Error %d Line: %d\n", bt->err, line), exit(0);
2314                           len = 0;
2315                         }
2316                         else if( len < 245 )
2317                                 key[len++] = ch;
2318                 fprintf(stderr, "finished deleting keys, %d\n", line);
2319                 break;
2320
2321         case 'f':
2322                 fprintf(stderr, "started finding keys for %s\n", argv[2]);
2323                 if( argc > 2 && (in = fopen (argv[2], "rb")) )
2324                   while( ch = getc(in), ch != EOF )
2325                         if( ch == '\n' )
2326                         {
2327                           if( off )
2328                                 sprintf((char *)key+len, "%.9d", line + off), len += 9;
2329                           line++;
2330                           if( bt_findkey (bt, key, len) )
2331                                 found++;
2332                           else if( bt->err )
2333                                 fprintf(stderr, "Error %d Syserr %d Line: %d\n", bt->err, errno, line), exit(0);
2334                           len = 0;
2335                         }
2336                         else if( len < 245 )
2337                                 key[len++] = ch;
2338                 fprintf(stderr, "finished search of %d keys, found %d\n", line, found);
2339                 break;
2340
2341         case 's':
2342                 scan++;
2343                 break;
2344
2345         }
2346
2347         done = clock();
2348         fprintf(stderr, " Time to complete: %.2f seconds\n", (float)(done - start) / CLOCKS_PER_SEC);
2349
2350         cnt = 0;
2351         len = key[0] = 0;
2352
2353         fprintf(stderr, "started reading\n");
2354
2355         slot = bt_startkey (bt, key, len);
2356
2357         if( bt->err )
2358           fprintf(stderr, "Error %d in StartKey. Syserror: %d\n", bt->err, errno), exit(0);
2359
2360         if( slot-- )
2361          while( slot = bt_nextkey (bt, slot) )
2362           if( cnt++, scan ) {
2363                         ptr = bt_key(bt, slot);
2364                         fwrite (ptr->key, ptr->len, 1, stdout);
2365                         fputc ('\n', stdout);
2366           }
2367
2368         fprintf(stderr, " Total keys read %d\n", cnt);
2369 }
2370
2371 #endif  //STANDALONE