]> pd.if.org Git - btree/blob - jaluta2.c
Introduce threadskv7 that includes atomic insert of a set of keys with values
[btree] / jaluta2.c
1 // jaluta's balanced B-Link tree algorithms
2 // 26 NOV 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
983                 // unlock page zero 
984
985                 if ( bt_unlockpage(bt, ALLOC_page, BtLockWrite) )
986                         return 0;
987
988                 return new_page;
989         }
990
991 #ifdef unix
992         if ( pwrite(bt->idx, page, bt->page_size, new_page << bt->page_bits) < bt->page_size )
993                 return bt->err = BTERR_wrt, 0;
994
995         // if writing first page of hash block, zero last page in the block
996
997         if ( !reuse && bt->hashmask > 0 && (new_page & bt->hashmask) == 0 )
998         {
999                 // use temp buffer to write zeros
1000                 memset(bt->temp, 0, bt->page_size);
1001                 if ( pwrite(bt->idx,bt->temp, bt->page_size, (new_page | bt->hashmask) << bt->page_bits) < bt->page_size )
1002                         return bt->err = BTERR_wrt, 0;
1003         }
1004 #else
1005         //      bring new page into page-cache and copy page.
1006         //      this will extend the file into the new pages.
1007
1008         if( !(pmap = (char*)bt_hashpage(bt, new_page & ~bt->hashmask)) )
1009                 return 0;
1010
1011         memcpy(pmap+((new_page & bt->hashmask) << bt->page_bits), page, bt->page_size);
1012 #endif
1013
1014         return new_page;
1015 }
1016
1017 //  find slot in given page for given key
1018
1019 int bt_findslot (BtPage page, unsigned char *key, uint len)
1020 {
1021 uint diff, higher = page->cnt, low = 1, slot;
1022 uint good = 0;
1023
1024         //      make last key an infinite fence value
1025
1026         if( !page->lvl || bt_getid (page->right) )
1027                 higher++;
1028         else
1029                 good++;
1030
1031         //      low is the next candidate, higher is already
1032         //      tested as .ge. the given key, loop ends when they meet
1033
1034         if( higher )
1035           while( diff = higher - low ) {
1036                 slot = low + ( diff >> 1 );
1037                 if( keycmp (keyptr(page, slot), key, len) < 0 )
1038                         low = slot + 1;
1039                 else
1040                         higher = slot, good++;
1041           }
1042
1043         //      return zero if key is beyond highkey value
1044         //      or page is empty
1045
1046         return good ? higher : 0;
1047 }
1048
1049 //  split full parent node
1050
1051 BTERR bt_splitparent (BtDb *bt, unsigned char *key, uint len)
1052 {
1053 uint cnt = 0, idx = 0, max, nxt = bt->page_size;
1054 uid parentpage = bt->parentpage, right;
1055 BtPage page = bt->parent;
1056 uid new_page;
1057 BtKey ptr;
1058
1059         //      upgrade parent latch to Xclusive
1060
1061         if( bt_lockpage (bt, bt->parentpage, BtLockUpgrade) )
1062                 return bt->err;
1063
1064         //  split higher half of keys to bt->frame
1065
1066         memset (bt->frame, 0, bt->page_size);
1067         max = (int)page->cnt;
1068         cnt = max / 2;
1069         idx = 0;
1070
1071         // link right sibling node into new right page
1072
1073         right = bt_getid (page->right);
1074         bt_putid(bt->frame->right, right);
1075
1076         //      record higher fence key in new right leaf page
1077
1078         if( bt->frame->fence = page->fence ) {
1079                 memcpy ((unsigned char *)bt->frame + bt->page_size - bt->frame->fence, (unsigned char *)(page) + bt->page_size - bt->frame->fence, bt->frame->fence);
1080                 nxt -= page->fence;
1081         }
1082
1083         while( cnt++ < max ) {
1084
1085                 // copy key, but not infinite values
1086
1087                 if( cnt < max || !page->lvl || right ) {
1088                         ptr = keyptr(page, cnt);
1089                         nxt -= ptr->len + 1;
1090                         memcpy ((unsigned char *)bt->frame + nxt, ptr, ptr->len + 1);
1091                 }
1092
1093                 // copy slot
1094
1095                 memcpy(slotptr(bt->frame,++idx)->id, slotptr(page,cnt)->id, BtId);
1096                 slotptr(bt->frame, idx)->tod = slotptr(page, cnt)->tod;
1097                 slotptr(bt->frame, idx)->off = nxt;
1098         }
1099
1100         bt->frame->bits = bt->page_bits;
1101         bt->frame->lvl = page->lvl;
1102         bt->frame->min = nxt;
1103         bt->frame->cnt = idx;
1104
1105         //      get new free page and write right frame to it.
1106
1107         if( !(new_page = bt_newpage(bt, bt->frame)) )
1108                 return bt->err;
1109
1110         //      update lower keys to continue in old page
1111
1112         memcpy (bt->frame, page, bt->page_size);
1113         memset (page+1, 0, bt->page_size - sizeof(*page));
1114         nxt = bt->page_size;
1115         max /= 2;
1116         cnt = 0;
1117         idx = 0;
1118
1119         //      record fence key in left leaf page
1120
1121         if( !page->lvl ) {
1122                 ptr = keyptr(bt->frame, max);
1123                 nxt -= ptr->len + 1;
1124                 memcpy ((unsigned char *)page + nxt, ptr, ptr->len + 1);
1125                 page->fence = ptr->len + 1;
1126         }
1127
1128         //  assemble page of smaller keys
1129         //      no infinite value to deal with
1130
1131         while( cnt++ < max ) {
1132                 ptr = keyptr(bt->frame, cnt);
1133                 nxt -= ptr->len + 1;
1134                 memcpy ((unsigned char *)page + nxt, ptr, ptr->len + 1);
1135                 memcpy(slotptr(page,++idx)->id, slotptr(bt->frame,cnt)->id, BtId);
1136                 slotptr(page, idx)->tod = slotptr(bt->frame, cnt)->tod;
1137                 slotptr(page, idx)->off = nxt;
1138         }
1139
1140         bt_putid(page->right, new_page);
1141         page->min = nxt;
1142         page->cnt = idx;
1143
1144         // update left node
1145
1146         if( bt_update(bt, page, parentpage) )
1147                 return bt->err;
1148
1149         //      decide to move to new right
1150         //      node or stay on left node
1151
1152         ptr = bt_highkey (bt, page);
1153
1154         if( keycmp (ptr, key, len) >= 0 )
1155                 return bt_unlockpage (bt, parentpage, BtLockUpgrade);
1156
1157         bt->parentpage = new_page;
1158
1159         if( bt_mappage (bt, &bt->parent, new_page) )
1160                 return bt->err;
1161         if( bt_lockpage (bt, new_page, BtLockUpdate) )
1162                 return bt->err;
1163         if( bt_unlockpage (bt, parentpage, BtLockUpgrade) )
1164                 return bt->err;
1165
1166         return bt_unlockpage (bt, parentpage, BtLockUpdate);
1167 }
1168
1169 //  add unlinked node key into parent
1170 //  childpage is existing record
1171 //      siblingpage is right record 
1172
1173 BTERR bt_parentlink (BtDb *bt, BtPage left, uid leftpage, BtKey rightkey, uid rightpage)
1174 {
1175 BtKey leftkey = bt_highkey (bt, left);
1176 BtPage page = bt->parent;
1177 uint slot, idx;
1178
1179         // upgrade parent latch to exclusive
1180
1181         if( bt_lockpage (bt, bt->parentpage, BtLockUpgrade) )
1182                 return bt->err;
1183
1184         // find the existing right high key in the parent
1185         // and fix the downlink to point to right page
1186
1187         if( rightkey ) {
1188           if( !(slot = bt_findslot (page, rightkey->key, rightkey->len)) )
1189                 return bt->err = BTERR_struct;
1190         } else
1191           slot = page->cnt;
1192
1193         bt_putid(slotptr(page,slot)->id, rightpage);
1194
1195         // calculate next available slot and copy left key onto page
1196
1197         page->min -= leftkey->len + 1; // reset lowest used offset
1198         memcpy ((unsigned char *)page + page->min, leftkey, leftkey->len + 1);
1199
1200         // now insert key into array before slot
1201
1202         idx = ++page->cnt;
1203
1204         while( idx > slot )
1205           *slotptr(page, idx) = *slotptr(page, idx -1), idx--;
1206
1207         bt_putid(slotptr(page,slot)->id, leftpage);
1208         slotptr(page, slot)->off = page->min;
1209
1210         if ( bt_update(bt, page, bt->parentpage) )
1211           return bt->err;
1212
1213         // downgrade parent page lock to BtLockUpdate
1214
1215         if( bt_unlockpage(bt, bt->parentpage, BtLockUpgrade) )
1216           return bt->err;
1217
1218         return 0;
1219 }
1220
1221 //      remove slot from parent
1222
1223 void bt_removeslot(BtDb *bt, uint slot)
1224 {
1225 uint nxt = bt->page_size, amt;
1226 BtPage page = bt->parent;
1227 uint cnt = 0, idx = 0;
1228 uint max = page->cnt;
1229 uid right;
1230 BtKey key;
1231
1232         memcpy (bt->frame, page, bt->page_size);
1233
1234         // skip page info and set rest of page to zero
1235         memset (page+1, 0, bt->page_size - sizeof(*page));
1236
1237         // copy fence key onto new page
1238         if( amt = page->fence ) {
1239                 nxt -= amt;
1240                 memcpy ((unsigned char *)page + nxt, (unsigned char *)(bt->frame) + nxt, amt);
1241         }
1242
1243         right = bt_getid (page->right);
1244
1245         while( cnt++ < max ) {
1246
1247                 // skip key to delete
1248
1249                 if( cnt == slot )
1250                         continue;
1251
1252                 // copy key, but not infinite value
1253
1254                 if( cnt < max || !page->lvl || right ) {
1255                         key = keyptr(bt->frame, cnt);
1256                         nxt -= key->len + 1;
1257                         memcpy ((unsigned char *)page + nxt, key, key->len + 1);
1258                 }
1259
1260                 // copy slot
1261                 memcpy(slotptr(page, ++idx)->id, slotptr(bt->frame, cnt)->id, BtId);
1262                 slotptr(page, idx)->tod = slotptr(bt->frame, cnt)->tod;
1263                 slotptr(page, idx)->off = nxt;
1264         }
1265         page->min = nxt;
1266         page->cnt = idx;
1267 }
1268
1269 //      unlink sibling node
1270
1271 BTERR bt_parentunlink (BtDb *bt, BtPage child, uid childpage)
1272 {
1273 BtKey parentkey = bt_highkey (bt, child);
1274 uint slot;
1275
1276         if( bt_lockpage (bt, bt->parentpage, BtLockUpgrade) )
1277                 return bt->err;
1278
1279         // delete child's slot from parent
1280
1281         if( slot = bt_findslot (bt->parent, parentkey->key, parentkey->len) )
1282                 bt_removeslot (bt, slot);
1283         else
1284                 return bt->err + BTERR_struct;
1285
1286         //      sibling now in the slot
1287
1288         bt_putid(slotptr(bt->parent,slot)->id, childpage);
1289
1290         if ( bt_update(bt, bt->parent, bt->parentpage) )
1291           return bt->err;
1292
1293         // unlock parent page completely
1294
1295         if( bt_unlockpage (bt, bt->parentpage, BtLockUpgrade) )
1296                 return bt->err;
1297
1298         return bt_unlockpage(bt, bt->parentpage, BtLockUpdate);
1299 }
1300
1301 //      merge right sibling page into child page
1302
1303 BTERR bt_mergepages (BtDb *bt, BtPage *right, uid rightpage)
1304 {
1305 BtPage *left = &bt->child;
1306 uint idx, amt;
1307 BtKey ptr;
1308
1309         if( bt_lockpage (bt, bt->childpage, BtLockUpgrade) )
1310                 return bt->err;
1311         if( bt_lockpage (bt, rightpage, BtLockUpgrade) )
1312                 return bt->err;
1313
1314         //      initialize empty frame
1315
1316         memset (bt->frame, 0, bt->page_size);
1317         *bt->frame = **right;
1318
1319         //      copy right fence key
1320
1321         if( amt = (*right)->fence )
1322                 memcpy ((unsigned char *)bt->frame + bt->page_size - amt, (unsigned char *)(*right) + bt->page_size - amt, amt);
1323
1324         bt->frame->min = bt->page_size - amt;
1325
1326         //      copy lowerkey key/values from left page
1327
1328         for( idx = 1; idx <= (*left)->cnt; idx++ ) {
1329                 ptr = keyptr(*left, idx);
1330                 bt->frame->min -= ptr->len + 1;
1331                 memcpy ((unsigned char *)bt->frame + bt->frame->min, ptr, ptr->len + 1);
1332                 memcpy (slotptr(bt->frame, idx)->id, slotptr(*left, idx)->id, BtId);
1333                 slotptr(bt->frame, idx)->tod = slotptr(*left, idx)->tod;
1334                 slotptr(bt->frame, idx)->off = bt->frame->min;
1335         }
1336
1337         bt->frame->cnt = (*left)->cnt;
1338
1339         //      copy higherkey key/values from right page
1340
1341         for( idx = 1; idx <= (*right)->cnt; idx++ ) {
1342
1343                 // copy key but not infinite value
1344
1345                 if( idx < (*right)->cnt || !(*right)->lvl || right ) {
1346                         ptr = keyptr(*right, idx);
1347                         bt->frame->min -= ptr->len + 1;
1348                         memcpy ((unsigned char *)bt->frame + bt->frame->min, ptr, ptr->len + 1);
1349                 }
1350
1351                 // copy slot
1352                 memcpy (slotptr(bt->frame, bt->frame->cnt + idx)->id, slotptr(*right, idx)->id, BtId);
1353                 slotptr(bt->frame, bt->frame->cnt + idx)->tod = slotptr(*right, idx)->tod;
1354                 slotptr(bt->frame, bt->frame->cnt + idx)->off = bt->frame->min;
1355         }
1356
1357         bt->frame->cnt += (*right)->cnt;
1358         memcpy (*left, bt->frame, bt->page_size);
1359
1360         if( bt_update (bt, *left, bt->childpage) )
1361                 return bt->err;
1362         if( bt_freepage (bt, *right, rightpage) )
1363                 return bt->err;
1364         if( bt_unlockpage (bt, rightpage, BtLockUpgrade) )
1365                 return bt->err;
1366         if( bt_unlockpage (bt, rightpage, BtLockUpdate) )
1367                 return bt->err;
1368
1369         return bt_unlockpage (bt, bt->childpage, BtLockUpgrade);
1370 }
1371
1372 //      redistribute right sibling page with child page
1373 //      switch child page to containing page
1374
1375 BTERR bt_redistribute (BtDb *bt, BtPage *right, uid rightpage, unsigned char *key, uint len)
1376 {
1377 uid siblingpage = bt_getid((*right)->right);
1378 BtPage *left = &bt->child;
1379 uint idx, cnt = 0, amt = 0;
1380 uint leftmax, rightmin;
1381 BtPage swap;
1382 BtKey ptr;
1383
1384         if( bt_lockpage (bt, bt->childpage, BtLockUpgrade) )
1385                 return bt->err;
1386         if( bt_lockpage (bt, rightpage, BtLockUpgrade) )
1387                 return bt->err;
1388
1389         //      initialize empty frames to contain redistributed pages
1390
1391         memset (bt->frame, 0, bt->page_size);   // new left page
1392         memset (bt->temp, 0, bt->page_size);    // new right page
1393
1394         *bt->frame = **left;
1395         *bt->temp = **right;
1396
1397         bt->frame->min = bt->page_size;
1398         bt->temp->min = bt->page_size;
1399         bt->frame->cnt = 0;
1400         bt->temp->cnt = 0;
1401
1402         //      find new left fence index
1403         //      and copy left fence key
1404
1405         if( (*left)->cnt > (*right)->cnt ) {
1406                 rightmin = 0;
1407                 leftmax = (*left)->cnt / 2;
1408                 if( !bt->frame->lvl ) {
1409                         ptr = keyptr(*left, leftmax);
1410                         bt->frame->fence = ptr->len + 1;
1411                         bt->frame->min -= ptr->len + 1;
1412                         memcpy ((unsigned char *)bt->frame + bt->frame->min, ptr, ptr->len + 1);
1413                 }
1414         } else {
1415                 leftmax = (*left)->cnt;
1416                 rightmin = (*right)->cnt / 2;
1417                 if( !bt->frame->lvl ) {
1418                         ptr = keyptr(*right, rightmin);
1419                         bt->frame->fence = ptr->len + 1;
1420                         bt->frame->min -= ptr->len + 1;
1421                         memcpy ((unsigned char *)bt->frame + bt->frame->min, ptr, ptr->len + 1);
1422                 }
1423         }
1424
1425         //      right fence stays the same, if any
1426
1427         if( amt = (*right)->fence ) {
1428                 bt->temp->min -= amt;
1429                 memcpy ((unsigned char *)bt->temp + bt->temp->min, (unsigned char *)(*right) + bt->temp->min, amt);
1430         }
1431
1432         //      copy first set of lowerkey key/values from left page
1433
1434         for( idx = 1; idx <= leftmax; idx++ ) {
1435                 bt->frame->cnt++;
1436                 ptr = keyptr(*left, idx);
1437                 bt->frame->min -= ptr->len + 1;
1438                 memcpy ((unsigned char *)bt->frame + bt->frame->min, ptr, ptr->len + 1);
1439                 memcpy (slotptr(bt->frame, idx)->id, slotptr(*left, idx)->id, BtId);
1440                 slotptr(bt->frame, idx)->tod = slotptr(*left, idx)->tod;
1441                 slotptr(bt->frame, idx)->off = bt->frame->min;
1442         }
1443
1444         //      copy remaining left page key/values from right page, if any
1445
1446         for( idx = 1; idx <= rightmin; idx++ ) {
1447                 bt->frame->cnt++;
1448                 ptr = keyptr(*right, idx);
1449                 bt->frame->min -= ptr->len + 1;
1450                 memcpy ((unsigned char *)bt->frame + bt->frame->min, ptr, ptr->len + 1);
1451                 memcpy (slotptr(bt->frame, bt->frame->cnt)->id, slotptr(*right, idx)->id, BtId);
1452                 slotptr(bt->frame, bt->frame->cnt)->tod = slotptr(*right, idx)->tod;
1453                 slotptr(bt->frame, bt->frame->cnt)->off = bt->frame->min;
1454         }
1455
1456         //      copy remaining left page key/values into new right page, if any
1457
1458         for( idx = leftmax; idx <= (*left)->cnt; idx++ ) {
1459                 bt->temp->cnt++;
1460                 ptr = keyptr(*left, idx);
1461                 bt->temp->min -= ptr->len + 1;
1462                 memcpy ((unsigned char *)bt->temp + bt->temp->min, ptr, ptr->len + 1);
1463                 memcpy (slotptr(bt->temp, bt->temp->cnt)->id, slotptr(*left, idx)->id, BtId);
1464                 slotptr(bt->temp, bt->temp->cnt)->tod = slotptr(*left, idx)->tod;
1465                 slotptr(bt->temp, bt->temp->cnt)->off = bt->temp->min;
1466         }
1467
1468         //      copy rest of higherkey key/values from right page
1469
1470         for( idx = rightmin; idx <= (*right)->cnt; idx++ ) {
1471
1472                 // copy key, but not infinite value
1473
1474                 if( idx < (*right)->cnt || !(*right)->lvl || siblingpage ) {
1475                         ptr = keyptr(*right, idx);
1476                         bt->temp->min -= ptr->len + 1;
1477                         memcpy ((unsigned char *)bt->temp + bt->temp->min, ptr, ptr->len + 1);
1478                 }
1479
1480                 //      copy slot
1481
1482                 bt->temp->cnt++;
1483                 memcpy (slotptr(bt->temp, bt->temp->cnt)->id, slotptr(*right, idx)->id, BtId);
1484                 slotptr(bt->temp, bt->temp->cnt)->tod = slotptr(*right, idx)->tod;
1485                 slotptr(bt->temp, bt->temp->cnt)->off = bt->temp->min;
1486         }
1487
1488         memcpy (*left, bt->frame, bt->page_size);
1489         memcpy (*right, bt->temp, bt->page_size);
1490
1491         if( bt_update (bt, *left, bt->childpage) )
1492                 return bt->err;
1493         if( bt_update (bt, *right, rightpage) )
1494                 return bt->err;
1495
1496         if( bt_unlockpage (bt, rightpage, BtLockUpgrade) )
1497                 return bt->err;
1498         if( bt_unlockpage (bt, rightpage, BtLockUpdate) )
1499                 return bt->err;
1500
1501         ptr = bt_highkey (bt, *left);
1502
1503         //      decide which page is the child page
1504         //      if leftkey >= our key, go with left
1505
1506         if( keycmp (ptr, key, len) >= 0 ) {
1507                 if( bt_unlockpage (bt, bt->childpage, BtLockUpgrade) )
1508                         return bt->err;
1509                 if( bt_unlockpage (bt, rightpage, BtLockUpgrade) )
1510                         return bt->err;
1511                 if( bt_unlockpage (bt, rightpage, BtLockUpdate) )
1512                         return bt->err;
1513         } else {
1514                 if( bt_unlockpage (bt, rightpage, BtLockUpgrade) )
1515                         return bt->err;
1516                 if( bt_unlockpage (bt, bt->childpage, BtLockUpgrade) )
1517                         return bt->err;
1518                 if( bt_unlockpage (bt, bt->childpage, BtLockUpdate) )
1519                         return bt->err;
1520                 swap = bt->child;
1521                 bt->child = *right;
1522                 *left = swap;
1523                 bt->childpage = rightpage;
1524         }
1525
1526         return 0;
1527 }
1528
1529 //      lower the root level by removing the child node
1530
1531 BTERR bt_lowerroot(BtDb *bt)
1532 {
1533         if( bt_lockpage (bt, ROOT_page, BtLockUpgrade) )
1534                 return bt->err;
1535
1536         if( bt_lockpage (bt, bt->childpage, BtLockUpgrade) )
1537                 return bt->err;
1538
1539         memcpy (bt->parent, bt->child, bt->page_size);
1540
1541         if( bt_update(bt, bt->parent, ROOT_page) )
1542                 return bt->err;
1543
1544         if( bt_freepage(bt, bt->child, bt->childpage) )
1545                 return bt->err;
1546
1547         if( bt_unlockpage (bt, bt->childpage, BtLockUpgrade) )
1548                 return bt->err;
1549
1550         if( bt_unlockpage (bt, bt->childpage, BtLockUpdate) )
1551                 return bt->err;
1552
1553         return bt_unlockpage (bt, ROOT_page, BtLockUpgrade);
1554 }
1555
1556 // split the root and raise the height of the btree
1557 //      return with parent page set to appropriate sibling
1558
1559 BTERR bt_raiseroot(BtDb *bt, uid sibling, unsigned char *key, uint len)
1560 {
1561 unsigned char lowerkey[256];
1562 uint nxt = bt->page_size;
1563 BtPage root = bt->parent;
1564 uid new_page;
1565 BtKey ptr;
1566
1567         //      upgrade root page lock to exclusive
1568
1569         if( bt_lockpage (bt, ROOT_page, BtLockUpgrade) )
1570                 return bt->err;
1571
1572         //  Obtain an empty page to use, and copy the current
1573         //  root (lower half) contents into it
1574
1575         if( !(new_page = bt_newpage(bt, root)) )
1576                 return bt->err;
1577
1578         if( bt_lockpage (bt, new_page, BtLockUpdate) )
1579                 return bt->err;
1580
1581         //      save high fence key for left page
1582
1583         ptr = bt_highkey(bt, root);
1584         memcpy (lowerkey, ptr, ptr->len + 1);
1585
1586         // preserve the page info at the bottom
1587         // and set rest to zero to initialize new root page
1588
1589         memset(root+1, 0, bt->page_size - sizeof(*root));
1590
1591         // insert left key in newroot page
1592
1593         nxt -= *lowerkey + 1;
1594         memcpy ((unsigned char *)root + nxt, lowerkey, *lowerkey + 1);
1595         bt_putid(slotptr(root, 1)->id, new_page);
1596         slotptr(root, 1)->off = nxt;
1597         
1598         // insert second (infinite) key on newroot page that's never examined
1599         // and increase the root level
1600
1601         bt_putid(slotptr(root, 2)->id, sibling);
1602         bt_putid(root->right, 0);
1603
1604         root->min = nxt;                // reset lowest used offset and key count
1605         root->fence = 0;
1606         root->cnt = 2;
1607         root->lvl++;
1608
1609         if( bt_update(bt, root, bt->parentpage) )
1610                 return bt->err;
1611
1612         if( bt_unlockpage(bt, ROOT_page, BtLockUpgrade) )
1613                 return bt->err;
1614
1615         if( bt_unlockpage(bt, ROOT_page, BtLockUpdate) )
1616                 return bt->err;
1617
1618         //      decide which root node to continue with
1619         //              sibling has upper keys, newpage the lower ones
1620
1621         if( keycmp((BtKey)lowerkey, key, len) < 0 ) {
1622                 bt->parentpage = sibling;               // go with the upper ones
1623
1624                 if( bt_unlockpage (bt, new_page, BtLockUpdate) )
1625                         return bt->err;
1626
1627                 return bt_mappage (bt, &bt->parent, sibling);
1628         }
1629
1630         bt->parentpage = new_page;                      // go with the lower ones
1631
1632         if( bt_unlockpage (bt, sibling, BtLockUpdate) )
1633                 return bt->err;
1634
1635         return bt_mappage (bt, &bt->parent, new_page);
1636 }
1637
1638 //      handle underflowing child node
1639
1640 BTERR bt_repairchild (BtDb *bt, uint parentslot, unsigned char *key, uint len)
1641 {
1642 BtKey parentkey = bt_slotkey(bt->parent, parentslot);
1643 BtKey fencekey = bt_highkey (bt, bt->child);
1644 BtKey highkey = bt_highkey(bt, bt->parent);
1645 BtKey siblingkey, siblingkey2;
1646 uid siblingpage, siblingpage2;
1647 uid swappage;
1648 BtPage swap;
1649 uint slot;
1650
1651   // high key is never NULL
1652   // fence key is null on right end
1653
1654   if( fencekey )
1655    if( !highkey || keycmp (fencekey, highkey->key, highkey->len) < 0 ) {
1656
1657         //  childpage is not rightmost child of parent page
1658
1659         siblingpage = bt_getid(bt->child->right);
1660
1661         if( bt_lockpage (bt, siblingpage, BtLockUpdate) )
1662           return bt->err;
1663
1664         if( bt_mappage (bt, &bt->sibling, siblingpage) )
1665           return bt->err;
1666
1667         if( !parentkey || keycmp (fencekey, parentkey->key, parentkey->len) < 0 ) {
1668
1669           // sibling is not linked in parent, so we can merge it
1670
1671           if( bt_unlockpage (bt, bt->parentpage, BtLockUpdate) )
1672                 return bt->err;
1673
1674           // calculate size of merged page by adding single child key to sibling
1675
1676           fencekey = keyptr(bt->child, 1);
1677
1678           if( bt->sibling->min < (bt->sibling->cnt + 1) * sizeof(BtSlot) + sizeof(*bt->sibling) + fencekey->len + 1)
1679             return bt_redistribute (bt, &bt->sibling, siblingpage, key, len);
1680           else
1681             return bt_mergepages (bt, &bt->sibling, siblingpage);
1682     }
1683
1684         //      sibling has a key in the parent node
1685         //      find its slot in parent
1686
1687         if( siblingkey = bt_highkey (bt, bt->sibling) )
1688           slot = bt_findslot (bt->parent, siblingkey->key, siblingkey->len);
1689         else
1690           slot = bt->parent->cnt;
1691
1692         parentkey = bt_slotkey (bt->parent, slot);
1693         siblingpage2 = bt_getid(bt->sibling->right);
1694
1695         if( siblingkey && parentkey && keycmp (siblingkey, parentkey->key, parentkey->len) < 0 ) {
1696
1697           //  sibling2 is not linked in P, its key is parentkey
1698           //  can parent support its insertion?
1699           //  if not, split parent node first.
1700
1701           if( bt->parent->min < (bt->parent->cnt + 1) * sizeof(BtSlot) + sizeof(*bt->parent) + parentkey->len + 1) {
1702                 if( bt_splitparent (bt, key, len) )
1703                   return bt->err;
1704
1705                 //  are we in the correct half of new parent nodes?
1706                 //  if not, restart the function.
1707
1708                 if( slot = bt_findslot (bt->parent, siblingkey->key, siblingkey->len) )
1709                   parentkey = keyptr(bt->parent, slot);
1710                 else
1711                   return BTERR_restart;
1712           }
1713
1714           //  link right of sibling into parent
1715
1716           if( bt_parentlink (bt, bt->sibling, siblingpage, parentkey, siblingpage2) )
1717                 return bt->err;
1718
1719           //  unlink sibling from parent
1720
1721           if( bt_parentunlink (bt, bt->child, bt->childpage) )
1722                 return bt->err;
1723
1724           fencekey = keyptr(bt->child, 1);
1725
1726           if( bt->sibling->min < (bt->sibling->cnt + 1) * sizeof(BtSlot) + sizeof(*bt->sibling) + fencekey->len + 1)
1727             return bt_redistribute (bt, &bt->sibling, siblingpage, key, len);
1728           else
1729             return bt_mergepages (bt, &bt->sibling, siblingpage);
1730         } else {
1731
1732           // unlink sibling from parent and merge
1733
1734           if( bt_parentunlink (bt, bt->child, bt->childpage) )
1735                 return bt->err;
1736
1737           fencekey = keyptr(bt->child, 1);
1738
1739           if( bt->sibling->min < (bt->sibling->cnt + 1) * sizeof(BtSlot) + sizeof(*bt->sibling) + fencekey->len + 1)
1740             return bt_redistribute (bt, &bt->sibling, siblingpage, key, len);
1741           else
1742             return bt_mergepages (bt, &bt->sibling, siblingpage);
1743         }
1744   }
1745
1746   //  child is rightmost key in the parent,
1747   //  work with nodes to left.
1748
1749   siblingpage = bt_getid(slotptr(bt->parent, parentslot - 1)->id);
1750   siblingkey = keyptr(bt->parent, parentslot - 1);
1751
1752   if( bt_unlockpage (bt, bt->childpage, BtLockUpdate) )
1753         return bt->err;
1754   if( bt_lockpage (bt, siblingpage, BtLockUpdate) )
1755         return bt->err;
1756   if( bt_mappage (bt, &bt->sibling, siblingpage) )
1757         return bt->err;
1758
1759   siblingpage2 = bt_getid(bt->sibling->right);
1760   siblingkey2 = bt_highkey (bt, bt->sibling);
1761
1762   if( bt_lockpage (bt, siblingpage2, BtLockUpdate) )
1763         return bt->err;
1764   if( bt_mappage (bt, &bt->sibling2, siblingpage2) )
1765         return bt->err;
1766
1767   if( keycmp (siblingkey, siblingkey2->key, siblingkey2->len) == 0 ) {
1768
1769         //      left sibling right (mapped into sibling2) is our child node
1770
1771         swap = bt->sibling2;
1772         bt->sibling2 = bt->child;
1773         bt->child = swap;
1774
1775         //  does child still need to merge/redistribute?
1776
1777         if( bt->child->cnt > 1 ) {
1778           if( bt_unlockpage (bt, bt->parentpage, BtLockUpdate) )
1779                 return bt->err;
1780
1781           return bt_unlockpage (bt, siblingpage, BtLockUpdate);
1782         }
1783         
1784         swap = bt->sibling;
1785         bt->sibling = bt->child;
1786         bt->child = swap;
1787
1788         bt->childpage = siblingpage;
1789         siblingpage = siblingpage2;
1790
1791         // unlink sibling node from parent and merge with child
1792
1793         if( bt_parentunlink (bt, bt->child, bt->childpage) )
1794                 return bt->err;
1795
1796         fencekey = keyptr(bt->sibling, 1);
1797
1798         if( bt->child->min < (bt->child->cnt + 1) * sizeof(BtSlot) + sizeof(*bt->sibling) + fencekey->len + 1)
1799           return bt_redistribute (bt, &bt->sibling, siblingpage, key, len);
1800         else
1801           return bt_mergepages (bt, &bt->sibling, siblingpage);
1802   }
1803
1804   //  currently unlinked sibling2 merges with child
1805
1806   fencekey = bt_highkey (bt, bt->sibling2);
1807
1808   if( bt->parent->min < (bt->parent->cnt + 1) * sizeof(BtSlot) + sizeof(*bt->parent) + fencekey->len + 1)
1809         if( bt_splitparent (bt, key, len) )
1810           return bt->err;
1811
1812   if( bt_parentlink (bt, bt->sibling, siblingpage, fencekey, siblingpage2) )
1813         return bt->err;
1814
1815   if( bt_unlockpage (bt, siblingpage, BtLockUpdate) )
1816           return bt->err;
1817   if( bt_lockpage (bt, bt->childpage, BtLockUpdate) )
1818           return bt->err;
1819   if( bt_mappage (bt, &bt->child, bt->childpage) )
1820         return bt->err;
1821
1822   if( bt->child->cnt > 1 ) {
1823     if( bt_unlockpage (bt, bt->parentpage, BtLockUpdate) )
1824           return bt->err;
1825     return bt_unlockpage (bt, siblingpage2, BtLockUpdate);
1826   }
1827
1828   // unlink child from parent node leaving immediate
1829   // left sibling2 to accept merge/redistribution
1830
1831   if( bt_parentunlink (bt, bt->sibling2, siblingpage2) )
1832         return bt->err;
1833
1834   swap = bt->sibling2;
1835   bt->sibling2 = bt->child;
1836   bt->child = swap;
1837   bt->childpage = siblingpage2;
1838
1839   fencekey = keyptr(bt->sibling2, 1);
1840
1841   if( bt->child->min < (bt->child->cnt + 1) * sizeof(BtSlot) + sizeof(*bt->child) + fencekey->len + 1) 
1842         return bt_redistribute (bt, &bt->sibling2, siblingpage2, key, len);
1843   else
1844         return bt_mergepages (bt, &bt->sibling2, siblingpage2);
1845 }
1846
1847 //  find and load leaf page for given key
1848 //      leave page BtLockShared, return with key's slot
1849
1850 int bt_loadpageread (BtDb *bt, unsigned char *key, uint len)
1851 {
1852 uid page_no = ROOT_page, prevpage = 0;
1853 uint slot, mode;
1854
1855   //  start at root of btree and drill down
1856
1857   do {
1858         bt->parentpage = page_no;
1859
1860         if( bt_lockpage(bt, bt->parentpage, BtLockShared) )
1861                 return 0;                                                                       
1862
1863         if( prevpage )
1864           if( bt_unlockpage(bt, prevpage, BtLockShared) )
1865                 return 0;
1866
1867         //      map/obtain page contents
1868
1869         if( bt_mappage (bt, &bt->parent, page_no) )
1870                 return 0;
1871
1872         //  find key on page at this level
1873         //      return if leaf page
1874
1875         if( (slot = bt_findslot (bt->parent, key, len)) ) {
1876           if( !bt->parent->lvl )
1877                 return slot;
1878
1879           // continue down to next level
1880
1881           page_no = bt_getid(slotptr(bt->parent, slot)->id);
1882         }
1883
1884         //  or slide right into next page
1885
1886         else
1887                 page_no = bt_getid(bt->parent->right);
1888
1889         prevpage = bt->parentpage;
1890   } while( page_no );
1891
1892   // return EOF on end of right chain
1893
1894   if( bt_unlockpage(bt, bt->parentpage, BtLockShared) )
1895         return 0;                                                                       
1896
1897   return 0;     // return EOF
1898 }
1899
1900 //  find and load leaf page for given key
1901 //      return w/slot # on leaf page
1902 //      leave page BtLockUpdate
1903
1904 int bt_loadpageupdate (BtDb *bt, unsigned char *key, uint len)
1905 {
1906 uid parentpage = 0, nextpage;
1907 BtKey fencekey, parentkey;
1908 uint slot, mode;
1909 BtPage swap;
1910
1911   //  start at root of btree and drill down
1912
1913   if( bt_lockpage(bt, ROOT_page, BtLockUpdate) )
1914         return 0;                                                                       
1915
1916   //    map/obtain page contents
1917
1918   if( bt_mappage (bt, &bt->parent, ROOT_page) )
1919         return 0;
1920
1921   bt->parentpage = ROOT_page;
1922
1923   do {
1924         // if root page, check for tree level growth
1925         // by existence of right pointer
1926
1927         if( bt->parentpage == ROOT_page )
1928           if( nextpage = bt_getid(bt->parent->right) ) {
1929                 if( bt_lockpage (bt, nextpage, BtLockUpdate) )
1930                   return 0;
1931
1932                 if( bt_raiseroot (bt, nextpage, key, len) )
1933                   return 0;
1934           }
1935
1936         //  find key on page at this level
1937         //      return if leaf page
1938
1939         slot = bt_findslot (bt->parent, key, len);
1940
1941         if( !bt->parent->lvl )
1942                 return slot;
1943
1944         // lock & map the child
1945
1946         bt->childpage = bt_getid(slotptr(bt->parent, slot)->id);
1947
1948         if( bt_lockpage(bt, bt->childpage, BtLockUpdate) )
1949                 return 0;                                                                       
1950
1951         if( bt_mappage (bt, &bt->child, bt->childpage) )
1952                 return 0;
1953
1954         //  check child for underflow
1955
1956         if( bt->parentpage == ROOT_page )
1957           if( bt->parent->cnt == 1 )
1958            if( !bt_getid(bt->child->right) ) {
1959                 if( bt_lowerroot (bt) )
1960                         return 0;
1961                 nextpage = ROOT_page;
1962                 continue;
1963            }
1964
1965         if( bt->child->cnt == 1 ) {
1966                 while( bt_repairchild (bt, slot, key, len) == BTERR_restart );
1967                 if( bt->err )
1968                         return 0;
1969
1970             swap = bt->child;
1971             bt->child = bt->parent;
1972             bt->parent = swap;
1973             nextpage = bt->childpage;
1974                 continue;
1975         }
1976
1977         fencekey = bt_highkey (bt, bt->child);
1978         parentkey = bt_slotkey(bt->parent, slot);
1979
1980         //  if right sibling is not linked,
1981         //      fix links in parent node
1982
1983         if( fencekey )
1984          if( !parentkey || keycmp (fencekey, parentkey->key, parentkey->len) < 0 ) {
1985
1986           if( bt->parent->min < (bt->parent->cnt + 1) * sizeof(BtSlot) + sizeof(*bt->parent) + fencekey->len + 1)
1987                 if( bt_splitparent (bt, key, len) ) {
1988                         return 0;
1989                 } else {
1990                         slot = bt_findslot (bt->parent, key, len);
1991                         parentkey = bt_slotkey(bt->parent, slot);
1992                 }
1993
1994           //  add key to parent for child page
1995           //    and fix downlink for childpage
1996
1997           nextpage = bt_getid(bt->child->right);
1998
1999           if( bt_parentlink (bt, bt->child, bt->childpage, parentkey, nextpage) )
2000                 return 0;
2001          }
2002
2003         //  unlock the parent page
2004
2005         if( bt_unlockpage (bt, bt->parentpage, BtLockUpdate) )
2006                 return 0;
2007
2008         //      is our key on this child page?
2009         //      is fencekey infinite, or .ge. our key
2010
2011         if( !fencekey || keycmp (fencekey, key, len) >= 0 ) {
2012           swap = bt->child;
2013           bt->child = bt->parent;
2014           bt->parent = swap;
2015           nextpage = bt->childpage;
2016           continue;
2017         }
2018
2019         //  otherwise slide right into next page
2020
2021         nextpage = bt_getid(bt->child->right);
2022
2023         if( bt_lockpage (bt, nextpage, BtLockUpdate) )
2024                 return 0;
2025
2026         if( bt_unlockpage (bt, bt->childpage, BtLockUpdate) )
2027                 return 0;
2028
2029         if( bt_mappage (bt, &bt->parent, nextpage) )
2030                 return 0;
2031
2032   } while( bt->parentpage = nextpage );
2033
2034   // return error on end of right chain
2035
2036   bt->err = BTERR_struct;
2037   return 0;     // return error
2038 }
2039
2040 //  find and delete key on leaf page
2041
2042 BTERR bt_deletekey (BtDb *bt, unsigned char *key, uint len)
2043 {
2044 uid page_no, right;
2045 uint slot, tod;
2046 BtKey ptr;
2047
2048         bt_resetpages (bt);
2049
2050         if( slot = bt_loadpageupdate (bt, key, len) )
2051                 ptr = keyptr(bt->parent, slot);
2052         else if( bt->err )
2053                 return bt->err;
2054
2055         // if key is found delete it, otherwise ignore request
2056
2057         if( slot && !keycmp (ptr, key, len) ) {
2058           if( bt_lockpage (bt, bt->parentpage, BtLockUpgrade) )
2059                 return bt->err;
2060
2061           bt_removeslot (bt, slot);
2062
2063           if( bt_update(bt, bt->parent, bt->parentpage) )
2064                 return bt->err;
2065
2066           if( bt_unlockpage (bt, bt->parentpage, BtLockUpgrade) )
2067                 return bt->err;
2068         }
2069
2070         return bt_unlockpage (bt, bt->parentpage, BtLockUpdate);
2071 }
2072
2073 //      find key in leaf page and return row-id
2074 //      or zero if key is not found.
2075
2076 uid bt_findkey (BtDb *bt, unsigned char *key, uint len)
2077 {
2078 uint  slot;
2079 BtKey ptr;
2080 uid id;
2081
2082         bt_resetpages (bt);
2083
2084         if( slot = bt_loadpageread (bt, key, len) )
2085                 ptr = keyptr(bt->parent, slot);
2086         else
2087                 return 0;
2088
2089         // if key exists, return row-id
2090         //      otherwise return 0
2091
2092         if( ptr->len == len && !memcmp (ptr->key, key, len) )
2093                 id = bt_getid(slotptr(bt->parent,slot)->id);
2094         else
2095                 id = 0;
2096
2097         if ( bt_unlockpage(bt, bt->parentpage, BtLockShared) )
2098                 return 0;
2099
2100         return id;
2101 }
2102
2103 //  Insert new key into the btree leaf page
2104
2105 BTERR bt_insertkey (BtDb *bt, unsigned char *key, uint len, uid id, uint tod)
2106 {
2107 uint slot, idx;
2108 BtKey ptr;
2109
2110         if( slot = bt_loadpageupdate (bt, key, len) )
2111                 ptr = keyptr(bt->parent, slot);
2112         else if( bt->err )
2113                 return bt->err;
2114
2115         if( bt->parent->lvl )
2116                 abort();
2117         if( bt->parent->lvl )
2118                 abort();
2119
2120
2121         // if key already exists, update id and return
2122
2123         if( slot )
2124           if( !keycmp (ptr, key, len) ) {
2125                 if( bt_lockpage (bt, bt->parentpage, BtLockUpgrade) )
2126                         return bt->err;
2127                 slotptr(bt->parent, slot)->tod = tod;
2128                 bt_putid(slotptr(bt->parent,slot)->id, id);
2129                 if ( bt_update(bt, bt->parent, bt->parentpage) )
2130                         return bt->err;
2131                 if( bt_unlockpage (bt, bt->parentpage, BtLockUpgrade) )
2132                         return bt->err;
2133                 return bt_unlockpage(bt, bt->parentpage, BtLockUpdate);
2134           }
2135
2136         // check if leaf page has enough space
2137
2138         if( bt->parent->min < (bt->parent->cnt + 1) * sizeof(BtSlot) + sizeof(*bt->parent) + len + 1) {
2139           if( bt_splitparent (bt, key, len) )
2140                 return bt->err;
2141
2142           slot = bt_findslot (bt->parent, key, len);
2143         }
2144
2145         // calculate next available slot and copy key into page
2146
2147         if( bt_lockpage (bt, bt->parentpage, BtLockUpgrade) )
2148                 return bt->err;
2149
2150         bt->parent->min -= len + 1; // reset lowest used offset
2151         ((unsigned char *)bt->parent)[bt->parent->min] = len;
2152         memcpy ((unsigned char *)bt->parent + bt->parent->min +1, key, len );
2153
2154         // now insert key into array before slot
2155
2156         idx = ++bt->parent->cnt;
2157
2158         if( slot )
2159          while( idx > slot )
2160           *slotptr(bt->parent, idx) = *slotptr(bt->parent, idx -1), idx--;
2161
2162         bt_putid(slotptr(bt->parent,idx)->id, id);
2163         slotptr(bt->parent, idx)->off = bt->parent->min;
2164         slotptr(bt->parent, idx)->tod = tod;
2165
2166         if ( bt_update(bt, bt->parent, bt->parentpage) )
2167           return bt->err;
2168
2169         if( bt_unlockpage (bt, bt->parentpage, BtLockUpgrade) )
2170                 return bt->err;
2171
2172         return bt_unlockpage(bt, bt->parentpage, BtLockUpdate);
2173 }
2174
2175 //  cache page of keys into cursor and return starting slot for given key
2176
2177 uint bt_startkey (BtDb *bt, unsigned char *key, uint len)
2178 {
2179 uint slot;
2180
2181         // cache page for retrieval
2182         if( slot = bt_loadpageread (bt, key, len) )
2183                 memcpy (bt->cursor, bt->parent, bt->page_size);
2184         bt->cursorpage = bt->parentpage;
2185         if ( bt_unlockpage(bt, bt->parentpage, BtLockShared) )
2186                 return 0;
2187
2188         return slot;
2189 }
2190
2191 //  return next slot for cursor page
2192 //  or slide cursor right into next page
2193
2194 uint bt_nextkey (BtDb *bt, uint slot)
2195 {
2196 off64_t right;
2197
2198   do {
2199         if( slot++ < bt->cursor->cnt )
2200                 return slot;
2201
2202         right = bt_getid(bt->cursor->right);
2203
2204         if( !right )
2205                 break;
2206
2207         bt->cursorpage = right;
2208
2209     if( bt_lockpage(bt, right,BtLockShared) )
2210                 return 0;
2211
2212         if( bt_mappage (bt, &bt->parent, right) )
2213                 break;
2214
2215         memcpy (bt->cursor, bt->parent, bt->page_size);
2216         if ( bt_unlockpage(bt, right, BtLockShared) )
2217                 return 0;
2218
2219         slot = 0;
2220   } while( 1 );
2221
2222   return bt->err = 0;
2223 }
2224
2225 BtKey bt_key(BtDb *bt, uint slot)
2226 {
2227         return keyptr(bt->cursor, slot);
2228 }
2229
2230 uid bt_uid(BtDb *bt, uint slot)
2231 {
2232         return bt_getid(slotptr(bt->cursor,slot)->id);
2233 }
2234
2235 uint bt_tod(BtDb *bt, uint slot)
2236 {
2237         return slotptr(bt->cursor,slot)->tod;
2238 }
2239
2240
2241 #ifdef STANDALONE
2242 //  standalone program to index file of keys
2243 //  then list them onto std-out
2244
2245 int main (int argc, char **argv)
2246 {
2247 uint slot, found = 0, line = 0, off = 0;
2248 int ch, cnt = 0, bits = 12;
2249 unsigned char key[256];
2250 clock_t done, start;
2251 time_t tod[1];
2252 uint scan = 0;
2253 uint len = 0;
2254 uint map = 0;
2255 BtKey ptr;
2256 BtDb *bt;
2257 FILE *in;
2258
2259         if( argc < 4 ) {
2260                 fprintf (stderr, "Usage: %s idx_file src_file Read/Write/Scan/Delete/Find [page_bits mapped_pool_pages start_line_number]", argv[0]);
2261                 exit(0);
2262         }
2263
2264         start = clock();
2265         time (tod);
2266
2267         if( argc > 4 )
2268                 bits = atoi(argv[4]);
2269
2270         if( argc > 5 )
2271                 map = atoi(argv[5]);
2272
2273         if( map > 65536 )
2274                 fprintf (stderr, "Warning: mapped_pool > 65536 pages\n");
2275
2276         if( argc > 6 )
2277                 off = atoi(argv[6]);
2278
2279         bt = bt_open ((argv[1]), BT_rw, bits, map);
2280
2281         if( !bt ) {
2282                 fprintf(stderr, "Index Open Error %s\n", argv[1]);
2283                 exit (1);
2284         }
2285
2286         switch(argv[3][0]| 0x20)
2287         {
2288         case 'w':
2289                 fprintf(stderr, "started indexing for %s\n", argv[2]);
2290                 if( argc > 2 && (in = fopen (argv[2], "rb")) )
2291                   while( ch = getc(in), ch != EOF )
2292                         if( ch == '\n' )
2293                         {
2294                           if( off )
2295                                 sprintf((char *)key+len, "%.9d", line + off), len += 9;
2296
2297                           if( bt_insertkey (bt, key, len, ++line, *tod) )
2298                                 fprintf(stderr, "Error %d Syserr %d Line: %d\n", bt->err, errno, line), exit(0);
2299                           len = 0;
2300                         }
2301                         else if( len < 245 )
2302                                 key[len++] = ch;
2303                 fprintf(stderr, "finished adding keys, %d \n", line);
2304                 break;
2305
2306         case 'd':
2307                 fprintf(stderr, "started deleting keys for %s\n", argv[2]);
2308                 if( argc > 2 && (in = fopen (argv[2], "rb")) )
2309                   while( ch = getc(in), ch != EOF )
2310                         if( ch == '\n' )
2311                         {
2312                           if( off )
2313                                 sprintf((char *)key+len, "%.9d", line + off), len += 9;
2314                           line++;
2315                           if( bt_deletekey (bt, key, len) )
2316                                 fprintf(stderr, "Error %d Line: %d\n", bt->err, line), exit(0);
2317                           len = 0;
2318                         }
2319                         else if( len < 245 )
2320                                 key[len++] = ch;
2321                 fprintf(stderr, "finished deleting keys, %d\n", line);
2322                 break;
2323
2324         case 'f':
2325                 fprintf(stderr, "started finding keys for %s\n", argv[2]);
2326                 if( argc > 2 && (in = fopen (argv[2], "rb")) )
2327                   while( ch = getc(in), ch != EOF )
2328                         if( ch == '\n' )
2329                         {
2330                           if( off )
2331                                 sprintf((char *)key+len, "%.9d", line + off), len += 9;
2332                           line++;
2333                           if( bt_findkey (bt, key, len) )
2334                                 found++;
2335                           else if( bt->err )
2336                                 fprintf(stderr, "Error %d Syserr %d Line: %d\n", bt->err, errno, line), exit(0);
2337                           len = 0;
2338                         }
2339                         else if( len < 245 )
2340                                 key[len++] = ch;
2341                 fprintf(stderr, "finished search of %d keys, found %d\n", line, found);
2342                 break;
2343
2344         case 's':
2345                 scan++;
2346                 break;
2347
2348         }
2349
2350         done = clock();
2351         fprintf(stderr, " Time to complete: %.2f seconds\n", (float)(done - start) / CLOCKS_PER_SEC);
2352
2353         cnt = 0;
2354         len = key[0] = 0;
2355
2356         fprintf(stderr, "started reading\n");
2357
2358         slot = bt_startkey (bt, key, len);
2359
2360         if( bt->err )
2361           fprintf(stderr, "Error %d in StartKey. Syserror: %d\n", bt->err, errno), exit(0);
2362
2363         if( slot-- )
2364          while( slot = bt_nextkey (bt, slot) )
2365           if( cnt++, scan ) {
2366                         ptr = bt_key(bt, slot);
2367                         fwrite (ptr->key, ptr->len, 1, stdout);
2368                         fputc ('\n', stdout);
2369           }
2370
2371         fprintf(stderr, " Total keys read %d\n", cnt);
2372 }
2373
2374 #endif  //STANDALONE