pd.if.org Git - btree/blob - jaluta2.c

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