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