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