]> pd.if.org Git - btree/blob - threads2h.c
Add finalized threadskv10g LSM btree.
[btree] / threads2h.c
1 // btree version threads2h pthread rw lock/SRW version
2 //      with fixed bt_deletekey code
3 // 17 FEB 2014
4
5 // author: karl malbrain, malbrain@cal.berkeley.edu
6
7 /*
8 This work, including the source code, documentation
9 and related data, is placed into the public domain.
10
11 The orginal author is Karl Malbrain.
12
13 THIS SOFTWARE IS PROVIDED AS-IS WITHOUT WARRANTY
14 OF ANY KIND, NOT EVEN THE IMPLIED WARRANTY OF
15 MERCHANTABILITY. THE AUTHOR OF THIS SOFTWARE,
16 ASSUMES _NO_ RESPONSIBILITY FOR ANY CONSEQUENCE
17 RESULTING FROM THE USE, MODIFICATION, OR
18 REDISTRIBUTION OF THIS SOFTWARE.
19 */
20
21 // Please see the project home page for documentation
22 // code.google.com/p/high-concurrency-btree
23
24 #define _FILE_OFFSET_BITS 64
25 #define _LARGEFILE64_SOURCE
26
27 #ifdef linux
28 #define _GNU_SOURCE
29 #endif
30
31 #ifdef unix
32 #include <unistd.h>
33 #include <stdio.h>
34 #include <stdlib.h>
35 #include <fcntl.h>
36 #include <sys/time.h>
37 #include <sys/mman.h>
38 #include <errno.h>
39 #include <pthread.h>
40 #else
41 #define WIN32_LEAN_AND_MEAN
42 #include <windows.h>
43 #include <stdio.h>
44 #include <stdlib.h>
45 #include <time.h>
46 #include <fcntl.h>
47 #include <process.h>
48 #include <intrin.h>
49 #endif
50
51 #include <memory.h>
52 #include <string.h>
53 #include <stddef.h>
54
55 typedef unsigned long long      uid;
56
57 #ifndef unix
58 typedef unsigned long long      off64_t;
59 typedef unsigned short          ushort;
60 typedef unsigned int            uint;
61 #endif
62
63 #define BT_latchtable   128                                     // number of latch manager slots
64
65 #define BT_ro 0x6f72    // ro
66 #define BT_rw 0x7772    // rw
67
68 #define BT_maxbits              24                                      // maximum page size in bits
69 #define BT_minbits              9                                       // minimum page size in bits
70 #define BT_minpage              (1 << BT_minbits)       // minimum page size
71 #define BT_maxpage              (1 << BT_maxbits)       // maximum page size
72
73 /*
74 There are five lock types for each node in three independent sets: 
75 1. (set 1) AccessIntent: Sharable. Going to Read the node. Incompatible with NodeDelete. 
76 2. (set 1) NodeDelete: Exclusive. About to release the node. Incompatible with AccessIntent. 
77 3. (set 2) ReadLock: Sharable. Read the node. Incompatible with WriteLock. 
78 4. (set 2) WriteLock: Exclusive. Modify the node. Incompatible with ReadLock and other WriteLocks. 
79 5. (set 3) ParentModification: Exclusive. Change the node's parent keys. Incompatible with another ParentModification. 
80 */
81
82 typedef enum{
83         BtLockAccess,
84         BtLockDelete,
85         BtLockRead,
86         BtLockWrite,
87         BtLockParent
88 } BtLock;
89
90 //      mode & definition for latch implementation
91
92 // exclusive is set for write access
93 // share is count of read accessors
94 // grant write lock when share == 0
95
96 volatile typedef struct {
97         unsigned char mutex[1];
98         unsigned char exclusive:1;
99         unsigned char pending:1;
100         ushort share;
101 } BtSpinLatch;
102
103 //  hash table entries
104
105 typedef struct {
106         BtSpinLatch latch[1];
107         volatile ushort slot;           // Latch table entry at head of chain
108 } BtHashEntry;
109
110 //      latch manager table structure
111
112 typedef struct {
113 #ifdef unix
114         pthread_rwlock_t lock[1];
115 #else
116         SRWLOCK srw[1];
117 #endif
118 } BtLatch;
119
120 typedef struct {
121         BtLatch readwr[1];                      // read/write page lock
122         BtLatch access[1];                      // Access Intent/Page delete
123         BtLatch parent[1];                      // Posting of fence key in parent
124         BtSpinLatch busy[1];            // slot is being moved between chains
125         volatile ushort next;           // next entry in hash table chain
126         volatile ushort prev;           // prev entry in hash table chain
127         volatile ushort pin;            // number of outstanding locks
128         volatile ushort hash;           // hash slot entry is under
129         volatile uid page_no;           // latch set page number
130 } BtLatchSet;
131
132 //      Define the length of the page and key pointers
133
134 #define BtId 6
135
136 //      Page key slot definition.
137
138 //      If BT_maxbits is 15 or less, you can save 4 bytes
139 //      for each key stored by making the first two uints
140 //      into ushorts.  You can also save 4 bytes by removing
141 //      the tod field from the key.
142
143 //      Keys are marked dead, but remain on the page until
144 //      it cleanup is called. The fence key (highest key) for
145 //      the page is always present, even after cleanup.
146
147 typedef struct {
148         uint off:BT_maxbits;            // page offset for key start
149         uint dead:1;                            // set for deleted key
150         uint tod;                                       // time-stamp for key
151         unsigned char id[BtId];         // id associated with key
152 } BtSlot;
153
154 //      The key structure occupies space at the upper end of
155 //      each page.  It's a length byte followed by the value
156 //      bytes.
157
158 typedef struct {
159         unsigned char len;
160         unsigned char key[1];
161 } *BtKey;
162
163 //      The first part of an index page.
164 //      It is immediately followed
165 //      by the BtSlot array of keys.
166
167 typedef struct BtPage_ {
168         uint cnt;                                       // count of keys in page
169         uint act;                                       // count of active keys
170         uint min;                                       // next key offset
171         unsigned char bits:7;           // page size in bits
172         unsigned char free:1;           // page is on free list
173         unsigned char lvl:6;            // level of page
174         unsigned char kill:1;           // page is being killed
175         unsigned char dirty:1;          // page has deleted keys
176         unsigned char right[BtId];      // page number to right
177 } *BtPage;
178
179 //      The memory mapping pool table buffer manager entry
180
181 typedef struct {
182         uid  basepage;                          // mapped base page number
183         char *map;                                      // mapped memory pointer
184         ushort slot;                            // slot index in this array
185         ushort pin;                                     // mapped page pin counter
186         void *hashprev;                         // previous pool entry for the same hash idx
187         void *hashnext;                         // next pool entry for the same hash idx
188 #ifndef unix
189         HANDLE hmap;                            // Windows memory mapping handle
190 #endif
191 } BtPool;
192
193 #define CLOCK_bit 0x8000                // bit in pool->pin
194
195 //  The loadpage interface object
196
197 typedef struct {
198         uid page_no;            // current page number
199         BtPage page;            // current page pointer
200         BtPool *pool;           // current page pool
201         BtLatchSet *latch;      // current page latch set
202 } BtPageSet;
203
204 //      structure for latch manager on ALLOC_page
205
206 typedef struct {
207         struct BtPage_ alloc[2];        // next & free page_nos in right ptr
208         BtSpinLatch lock[1];            // allocation area lite latch
209         ushort latchdeployed;           // highest number of latch entries deployed
210         ushort nlatchpage;                      // number of latch pages at BT_latch
211         ushort latchtotal;                      // number of page latch entries
212         ushort latchhash;                       // number of latch hash table slots
213         ushort latchvictim;                     // next latch entry to examine
214         BtHashEntry table[0];           // the hash table
215 } BtLatchMgr;
216
217 //      The object structure for Btree access
218
219 typedef struct {
220         uint page_size;                         // page size    
221         uint page_bits;                         // page size in bits    
222         uint seg_bits;                          // seg size in pages in bits
223         uint mode;                                      // read-write mode
224 #ifdef unix
225         int idx;
226 #else
227         HANDLE idx;
228 #endif
229         ushort poolcnt;                         // highest page pool node in use
230         ushort poolmax;                         // highest page pool node allocated
231         ushort poolmask;                        // total number of pages in mmap segment - 1
232         ushort hashsize;                        // size of Hash Table for pool entries
233         volatile uint evicted;          // last evicted pool table slot
234         ushort *hash;                           // pool index for hash entries
235         BtSpinLatch *latch;                     // latches for hash table slots
236         BtLatchMgr *latchmgr;           // mapped latch page from allocation page
237         BtLatchSet *latchsets;          // mapped latch set from latch pages
238         BtPool *pool;                           // memory pool page segments
239 #ifndef unix
240         HANDLE halloc;                          // allocation and latch table handle
241 #endif
242 } BtMgr;
243
244 typedef struct {
245         BtMgr *mgr;                     // buffer manager for thread
246         BtPage cursor;          // cached frame for start/next (never mapped)
247         BtPage frame;           // spare frame for the page split (never mapped)
248         BtPage zero;            // page frame for zeroes at end of file
249         uid cursor_page;        // current cursor page number   
250         unsigned char *mem;     // frame, cursor, page memory buffer
251         int found;                      // last delete or insert was found
252         int err;                        // last error
253 } BtDb;
254
255 typedef enum {
256         BTERR_ok = 0,
257         BTERR_struct,
258         BTERR_ovflw,
259         BTERR_lock,
260         BTERR_map,
261         BTERR_wrt,
262         BTERR_hash
263 } BTERR;
264
265 // B-Tree functions
266 extern void bt_close (BtDb *bt);
267 extern BtDb *bt_open (BtMgr *mgr);
268 extern BTERR bt_insertkey (BtDb *bt, unsigned char *key, uint len, uint lvl, uid id, uint tod);
269 extern BTERR  bt_deletekey (BtDb *bt, unsigned char *key, uint len, uint lvl);
270 extern uid bt_findkey    (BtDb *bt, unsigned char *key, uint len);
271 extern uint bt_startkey  (BtDb *bt, unsigned char *key, uint len);
272 extern uint bt_nextkey   (BtDb *bt, uint slot);
273
274 //      manager functions
275 extern BtMgr *bt_mgr (char *name, uint mode, uint bits, uint poolsize, uint segsize, uint hashsize);
276 void bt_mgrclose (BtMgr *mgr);
277
278 //  Helper functions to return slot values
279
280 extern BtKey bt_key (BtDb *bt, uint slot);
281 extern uid bt_uid (BtDb *bt, uint slot);
282 extern uint bt_tod (BtDb *bt, uint slot);
283
284 //  BTree page number constants
285 #define ALLOC_page              0       // allocation & lock manager hash table
286 #define ROOT_page               1       // root of the btree
287 #define LEAF_page               2       // first page of leaves
288 #define LATCH_page              3       // pages for lock manager
289
290 //      Number of levels to create in a new BTree
291
292 #define MIN_lvl                 2
293
294 //  The page is allocated from low and hi ends.
295 //  The key offsets and row-id's are allocated
296 //  from the bottom, while the text of the key
297 //  is allocated from the top.  When the two
298 //  areas meet, the page is split into two.
299
300 //  A key consists of a length byte, two bytes of
301 //  index number (0 - 65534), and up to 253 bytes
302 //  of key value.  Duplicate keys are discarded.
303 //  Associated with each key is a 48 bit row-id.
304
305 //  The b-tree root is always located at page 1.
306 //      The first leaf page of level zero is always
307 //      located on page 2.
308
309 //      The b-tree pages are linked with next
310 //      pointers to facilitate enumerators,
311 //      and provide for concurrency.
312
313 //      When to root page fills, it is split in two and
314 //      the tree height is raised by a new root at page
315 //      one with two keys.
316
317 //      Deleted keys are marked with a dead bit until
318 //      page cleanup The fence key for a node is
319 //      present in a special array.
320
321 //  Groups of pages called segments from the btree are optionally
322 //  cached with a memory mapped pool. A hash table is used to keep
323 //  track of the cached segments.  This behaviour is controlled
324 //  by the cache block size parameter to bt_open.
325
326 //  To achieve maximum concurrency one page is locked at a time
327 //  as the tree is traversed to find leaf key in question. The right
328 //  page numbers are used in cases where the page is being split,
329 //      or consolidated.
330
331 //  Page 0 is dedicated to lock for new page extensions,
332 //      and chains empty pages together for reuse.
333
334 //      The ParentModification lock on a node is obtained to serialize posting
335 //      or changing the fence key for a node.
336
337 //      Empty pages are chained together through the ALLOC page and reused.
338
339 //      Access macros to address slot and key values from the page.
340 //      Page slots use 1 based indexing.
341
342 #define slotptr(page, slot) (((BtSlot *)(page+1)) + (slot-1))
343 #define keyptr(page, slot) ((BtKey)((unsigned char*)(page) + slotptr(page, slot)->off))
344
345 void bt_putid(unsigned char *dest, uid id)
346 {
347 int i = BtId;
348
349         while( i-- )
350                 dest[i] = (unsigned char)id, id >>= 8;
351 }
352
353 uid bt_getid(unsigned char *src)
354 {
355 uid id = 0;
356 int i;
357
358         for( i = 0; i < BtId; i++ )
359                 id <<= 8, id |= *src++; 
360
361         return id;
362 }
363
364 //      Latch Manager
365
366 //      wait until write lock mode is clear
367 //      and add 1 to the share count
368
369 void bt_spinreadlock(BtSpinLatch *latch)
370 {
371 ushort prev;
372
373   do {
374         //      obtain latch mutex
375 #ifdef unix
376         if( __sync_lock_test_and_set(latch->mutex, 1) )
377                 continue;
378 #else
379         if( _InterlockedExchange8(latch->mutex, 1) )
380                 continue;
381 #endif
382         //  see if exclusive request is granted or pending
383
384         if( prev = !(latch->exclusive | latch->pending) )
385                 latch->share++;
386
387 #ifdef unix
388         __sync_lock_release (latch->mutex);
389 #else
390         _InterlockedExchange8(latch->mutex, 0);
391 #endif
392
393         if( prev )
394                 return;
395
396 #ifdef  unix
397   } while( sched_yield(), 1 );
398 #else
399   } while( SwitchToThread(), 1 );
400 #endif
401 }
402
403 //      wait for other read and write latches to relinquish
404
405 void bt_spinwritelock(BtSpinLatch *latch)
406 {
407 uint prev;
408
409   do {
410 #ifdef  unix
411         if( __sync_lock_test_and_set(latch->mutex, 1) )
412                 continue;
413 #else
414         if( _InterlockedExchange8(latch->mutex, 1) )
415                 continue;
416 #endif
417         if( prev = !(latch->share | latch->exclusive) )
418                 latch->exclusive = 1, latch->pending = 0;
419         else
420                 latch->pending = 1;
421 #ifdef unix
422         __sync_lock_release (latch->mutex);
423 #else
424         _InterlockedExchange8(latch->mutex, 0);
425 #endif
426         if( prev )
427                 return;
428 #ifdef  unix
429   } while( sched_yield(), 1 );
430 #else
431   } while( SwitchToThread(), 1 );
432 #endif
433 }
434
435 //      try to obtain write lock
436
437 //      return 1 if obtained,
438 //              0 otherwise
439
440 int bt_spinwritetry(BtSpinLatch *latch)
441 {
442 uint prev;
443
444 #ifdef unix
445         if( __sync_lock_test_and_set(latch->mutex, 1) )
446                 return 0;
447 #else
448         if( _InterlockedExchange8(latch->mutex, 1) )
449                 return 0;
450 #endif
451         //      take write access if all bits are clear
452
453         if( prev = !(latch->exclusive | latch->share) )
454                 latch->exclusive = 1;
455
456 #ifdef unix
457         __sync_lock_release (latch->mutex);
458 #else
459         _InterlockedExchange8(latch->mutex, 0);
460 #endif
461         return prev;
462 }
463
464 //      clear write mode
465
466 void bt_spinreleasewrite(BtSpinLatch *latch)
467 {
468         //      obtain latch mutex
469 #ifdef unix
470         while( __sync_lock_test_and_set(latch->mutex, 1) )
471                 sched_yield();
472 #else
473         while( _InterlockedExchange8(latch->mutex, 1) )
474                 SwitchToThread();
475 #endif
476         latch->exclusive = 0;
477 #ifdef unix
478         __sync_lock_release (latch->mutex);
479 #else
480         _InterlockedExchange8(latch->mutex, 0);
481 #endif
482 }
483
484 //      decrement reader count
485
486 void bt_spinreleaseread(BtSpinLatch *latch)
487 {
488 #ifdef unix
489         while( __sync_lock_test_and_set(latch->mutex, 1) )
490                 sched_yield();
491 #else
492         while( _InterlockedExchange8(latch->mutex, 1) )
493                 SwitchToThread();
494 #endif
495         latch->share--;
496 #ifdef unix
497         __sync_lock_release (latch->mutex);
498 #else
499         _InterlockedExchange8(latch->mutex, 0);
500 #endif
501 }
502
503 void bt_readlock(BtLatch *latch)
504 {
505 #ifdef unix
506         pthread_rwlock_rdlock (latch->lock);
507 #else
508         AcquireSRWLockShared (latch->srw);
509 #endif
510 }
511
512 //      wait for other read and write latches to relinquish
513
514 void bt_writelock(BtLatch *latch)
515 {
516 #ifdef unix
517         pthread_rwlock_wrlock (latch->lock);
518 #else
519         AcquireSRWLockExclusive (latch->srw);
520 #endif
521 }
522
523 //      try to obtain write lock
524
525 //      return 1 if obtained,
526 //              0 if already write or read locked
527
528 int bt_writetry(BtLatch *latch)
529 {
530 int result = 0;
531
532 #ifdef unix
533         result = !pthread_rwlock_trywrlock (latch->lock);
534 #else
535         result = TryAcquireSRWLockExclusive (latch->srw);
536 #endif
537         return result;
538 }
539
540 //      clear write mode
541
542 void bt_releasewrite(BtLatch *latch)
543 {
544 #ifdef unix
545         pthread_rwlock_unlock (latch->lock);
546 #else
547         ReleaseSRWLockExclusive (latch->srw);
548 #endif
549 }
550
551 //      decrement reader count
552
553 void bt_releaseread(BtLatch *latch)
554 {
555 #ifdef unix
556         pthread_rwlock_unlock (latch->lock);
557 #else
558         ReleaseSRWLockShared (latch->srw);
559 #endif
560 }
561
562 void bt_initlockset (BtLatchSet *set)
563 {
564 #ifdef unix
565 pthread_rwlockattr_t rwattr[1];
566
567         pthread_rwlockattr_init (rwattr);
568         pthread_rwlockattr_setpshared (rwattr, PTHREAD_PROCESS_SHARED);
569
570         pthread_rwlock_init (set->readwr->lock, rwattr);
571         pthread_rwlock_init (set->access->lock, rwattr);
572         pthread_rwlock_init (set->parent->lock, rwattr);
573         pthread_rwlockattr_destroy (rwattr);
574 #else
575         InitializeSRWLock (set->readwr->srw);
576         InitializeSRWLock (set->access->srw);
577         InitializeSRWLock (set->parent->srw);
578 #endif
579 }
580
581 //      link latch table entry into latch hash table
582
583 void bt_latchlink (BtDb *bt, ushort hashidx, ushort victim, uid page_no)
584 {
585 BtLatchSet *set = bt->mgr->latchsets + victim;
586
587         if( set->next = bt->mgr->latchmgr->table[hashidx].slot )
588                 bt->mgr->latchsets[set->next].prev = victim;
589
590         bt->mgr->latchmgr->table[hashidx].slot = victim;
591         set->page_no = page_no;
592         set->hash = hashidx;
593         set->prev = 0;
594 }
595
596 //      release latch pin
597
598 void bt_unpinlatch (BtLatchSet *set)
599 {
600 #ifdef unix
601         __sync_fetch_and_add(&set->pin, -1);
602 #else
603         _InterlockedDecrement16 (&set->pin);
604 #endif
605 }
606
607 //      find existing latchset or inspire new one
608 //      return with latchset pinned
609
610 BtLatchSet *bt_pinlatch (BtDb *bt, uid page_no)
611 {
612 ushort hashidx = page_no % bt->mgr->latchmgr->latchhash;
613 ushort slot, avail = 0, victim, idx;
614 BtLatchSet *set;
615
616   //  try to find existing latch table entry for this page
617
618         //  obtain read lock on hash table entry
619
620         bt_spinreadlock(bt->mgr->latchmgr->table[hashidx].latch);
621
622         if( slot = bt->mgr->latchmgr->table[hashidx].slot ) do
623         {
624                 set = bt->mgr->latchsets + slot;
625                 if( page_no == set->page_no )
626                         break;
627         } while( slot = set->next );
628
629         if( slot ) {
630 #ifdef unix
631                 __sync_fetch_and_add(&set->pin, 1);
632 #else
633                 _InterlockedIncrement16 (&set->pin);
634 #endif
635         }
636
637     bt_spinreleaseread (bt->mgr->latchmgr->table[hashidx].latch);
638
639         if( slot )
640                 return set;
641
642   //  try again, this time with write lock
643
644   bt_spinwritelock(bt->mgr->latchmgr->table[hashidx].latch);
645
646   if( slot = bt->mgr->latchmgr->table[hashidx].slot ) do
647   {
648         set = bt->mgr->latchsets + slot;
649         if( page_no == set->page_no )
650                 break;
651         if( !set->pin && !avail )
652                 avail = slot;
653   } while( slot = set->next );
654
655   //  found our entry, or take over an unpinned one
656
657   if( slot || (slot = avail) ) {
658         set = bt->mgr->latchsets + slot;
659 #ifdef unix
660         __sync_fetch_and_add(&set->pin, 1);
661 #else
662         _InterlockedIncrement16 (&set->pin);
663 #endif
664         set->page_no = page_no;
665         bt_spinreleasewrite(bt->mgr->latchmgr->table[hashidx].latch);
666         return set;
667   }
668
669         //  see if there are any unused entries
670 #ifdef unix
671         victim = __sync_fetch_and_add (&bt->mgr->latchmgr->latchdeployed, 1) + 1;
672 #else
673         victim = _InterlockedIncrement16 (&bt->mgr->latchmgr->latchdeployed);
674 #endif
675
676         if( victim < bt->mgr->latchmgr->latchtotal ) {
677                 set = bt->mgr->latchsets + victim;
678 #ifdef unix
679                 __sync_fetch_and_add(&set->pin, 1);
680 #else
681                 _InterlockedIncrement16 (&set->pin);
682 #endif
683                 bt_initlockset (set);
684                 bt_latchlink (bt, hashidx, victim, page_no);
685                 bt_spinreleasewrite (bt->mgr->latchmgr->table[hashidx].latch);
686                 return set;
687         }
688
689 #ifdef unix
690         victim = __sync_fetch_and_add (&bt->mgr->latchmgr->latchdeployed, -1);
691 #else
692         victim = _InterlockedDecrement16 (&bt->mgr->latchmgr->latchdeployed);
693 #endif
694   //  find and reuse previous lock entry
695
696   while( 1 ) {
697 #ifdef unix
698         victim = __sync_fetch_and_add(&bt->mgr->latchmgr->latchvictim, 1);
699 #else
700         victim = _InterlockedIncrement16 (&bt->mgr->latchmgr->latchvictim) - 1;
701 #endif
702         //      we don't use slot zero
703
704         if( victim %= bt->mgr->latchmgr->latchtotal )
705                 set = bt->mgr->latchsets + victim;
706         else
707                 continue;
708
709         //      take control of our slot
710         //      from other threads
711
712         if( set->pin || !bt_spinwritetry (set->busy) )
713                 continue;
714
715         idx = set->hash;
716
717         // try to get write lock on hash chain
718         //      skip entry if not obtained
719         //      or has outstanding locks
720
721         if( !bt_spinwritetry (bt->mgr->latchmgr->table[idx].latch) ) {
722                 bt_spinreleasewrite (set->busy);
723                 continue;
724         }
725
726         if( set->pin ) {
727                 bt_spinreleasewrite (set->busy);
728                 bt_spinreleasewrite (bt->mgr->latchmgr->table[idx].latch);
729                 continue;
730         }
731
732         //  unlink our available victim from its hash chain
733
734         if( set->prev )
735                 bt->mgr->latchsets[set->prev].next = set->next;
736         else
737                 bt->mgr->latchmgr->table[idx].slot = set->next;
738
739         if( set->next )
740                 bt->mgr->latchsets[set->next].prev = set->prev;
741
742         bt_spinreleasewrite (bt->mgr->latchmgr->table[idx].latch);
743 #ifdef unix
744         __sync_fetch_and_add(&set->pin, 1);
745 #else
746         _InterlockedIncrement16 (&set->pin);
747 #endif
748         bt_latchlink (bt, hashidx, victim, page_no);
749         bt_spinreleasewrite (bt->mgr->latchmgr->table[hashidx].latch);
750         bt_spinreleasewrite (set->busy);
751         return set;
752   }
753 }
754
755 void bt_mgrclose (BtMgr *mgr)
756 {
757 BtPool *pool;
758 uint slot;
759
760         // release mapped pages
761         //      note that slot zero is never used
762
763         for( slot = 1; slot < mgr->poolmax; slot++ ) {
764                 pool = mgr->pool + slot;
765                 if( pool->slot )
766 #ifdef unix
767                         munmap (pool->map, (mgr->poolmask+1) << mgr->page_bits);
768 #else
769                 {
770                         FlushViewOfFile(pool->map, 0);
771                         UnmapViewOfFile(pool->map);
772                         CloseHandle(pool->hmap);
773                 }
774 #endif
775         }
776
777 #ifdef unix
778         munmap (mgr->latchsets, mgr->latchmgr->nlatchpage * mgr->page_size);
779         munmap (mgr->latchmgr, mgr->page_size);
780 #else
781         FlushViewOfFile(mgr->latchmgr, 0);
782         UnmapViewOfFile(mgr->latchmgr);
783         CloseHandle(mgr->halloc);
784 #endif
785 #ifdef unix
786         close (mgr->idx);
787         free (mgr->pool);
788         free (mgr->hash);
789         free ((void *)mgr->latch);
790         free (mgr);
791 #else
792         FlushFileBuffers(mgr->idx);
793         CloseHandle(mgr->idx);
794         GlobalFree (mgr->pool);
795         GlobalFree (mgr->hash);
796         GlobalFree ((void *)mgr->latch);
797         GlobalFree (mgr);
798 #endif
799 }
800
801 //      close and release memory
802
803 void bt_close (BtDb *bt)
804 {
805 #ifdef unix
806         if( bt->mem )
807                 free (bt->mem);
808 #else
809         if( bt->mem)
810                 VirtualFree (bt->mem, 0, MEM_RELEASE);
811 #endif
812         free (bt);
813 }
814
815 //  open/create new btree buffer manager
816
817 //      call with file_name, BT_openmode, bits in page size (e.g. 16),
818 //              size of mapped page pool (e.g. 8192)
819
820 BtMgr *bt_mgr (char *name, uint mode, uint bits, uint poolmax, uint segsize, uint hashsize)
821 {
822 uint lvl, attr, cacheblk, last, slot, idx;
823 uint nlatchpage, latchhash;
824 BtLatchMgr *latchmgr;
825 off64_t size;
826 uint amt[1];
827 BtMgr* mgr;
828 BtKey key;
829 int flag;
830
831 #ifndef unix
832 SYSTEM_INFO sysinfo[1];
833 #endif
834
835         // determine sanity of page size and buffer pool
836
837         if( bits > BT_maxbits )
838                 bits = BT_maxbits;
839         else if( bits < BT_minbits )
840                 bits = BT_minbits;
841
842         if( !poolmax )
843                 return NULL;    // must have buffer pool
844
845 #ifdef unix
846         mgr = calloc (1, sizeof(BtMgr));
847
848         mgr->idx = open ((char*)name, O_RDWR | O_CREAT, 0666);
849
850         if( mgr->idx == -1 )
851                 return free(mgr), NULL;
852         
853         cacheblk = 4096;        // minimum mmap segment size for unix
854
855 #else
856         mgr = GlobalAlloc (GMEM_FIXED|GMEM_ZEROINIT, sizeof(BtMgr));
857         attr = FILE_ATTRIBUTE_NORMAL;
858         mgr->idx = CreateFile(name, GENERIC_READ| GENERIC_WRITE, FILE_SHARE_READ|FILE_SHARE_WRITE, NULL, OPEN_ALWAYS, attr, NULL);
859
860         if( mgr->idx == INVALID_HANDLE_VALUE )
861                 return GlobalFree(mgr), NULL;
862
863         // normalize cacheblk to multiple of sysinfo->dwAllocationGranularity
864         GetSystemInfo(sysinfo);
865         cacheblk = sysinfo->dwAllocationGranularity;
866 #endif
867
868 #ifdef unix
869         latchmgr = malloc (BT_maxpage);
870         *amt = 0;
871
872         // read minimum page size to get root info
873
874         if( size = lseek (mgr->idx, 0L, 2) ) {
875                 if( pread(mgr->idx, latchmgr, BT_minpage, 0) == BT_minpage )
876                         bits = latchmgr->alloc->bits;
877                 else
878                         return free(mgr), free(latchmgr), NULL;
879         } else if( mode == BT_ro )
880                 return free(latchmgr), bt_mgrclose (mgr), NULL;
881 #else
882         latchmgr = VirtualAlloc(NULL, BT_maxpage, MEM_COMMIT, PAGE_READWRITE);
883         size = GetFileSize(mgr->idx, amt);
884
885         if( size || *amt ) {
886                 if( !ReadFile(mgr->idx, (char *)latchmgr, BT_minpage, amt, NULL) )
887                         return bt_mgrclose (mgr), NULL;
888                 bits = latchmgr->alloc->bits;
889         } else if( mode == BT_ro )
890                 return bt_mgrclose (mgr), NULL;
891 #endif
892
893         mgr->page_size = 1 << bits;
894         mgr->page_bits = bits;
895
896         mgr->poolmax = poolmax;
897         mgr->mode = mode;
898
899         if( cacheblk < mgr->page_size )
900                 cacheblk = mgr->page_size;
901
902         //  mask for partial memmaps
903
904         mgr->poolmask = (cacheblk >> bits) - 1;
905
906         //      see if requested size of pages per memmap is greater
907
908         if( (1 << segsize) > mgr->poolmask )
909                 mgr->poolmask = (1 << segsize) - 1;
910
911         mgr->seg_bits = 0;
912
913         while( (1 << mgr->seg_bits) <= mgr->poolmask )
914                 mgr->seg_bits++;
915
916         mgr->hashsize = hashsize;
917
918 #ifdef unix
919         mgr->pool = calloc (poolmax, sizeof(BtPool));
920         mgr->hash = calloc (hashsize, sizeof(ushort));
921         mgr->latch = calloc (hashsize, sizeof(BtSpinLatch));
922 #else
923         mgr->pool = GlobalAlloc (GMEM_FIXED|GMEM_ZEROINIT, poolmax * sizeof(BtPool));
924         mgr->hash = GlobalAlloc (GMEM_FIXED|GMEM_ZEROINIT, hashsize * sizeof(ushort));
925         mgr->latch = GlobalAlloc (GMEM_FIXED|GMEM_ZEROINIT, hashsize * sizeof(BtSpinLatch));
926 #endif
927
928         if( size || *amt )
929                 goto mgrlatch;
930
931         // initialize an empty b-tree with latch page, root page, page of leaves
932         // and page(s) of latches
933
934         memset (latchmgr, 0, 1 << bits);
935         nlatchpage = BT_latchtable / (mgr->page_size / sizeof(BtLatchSet)) + 1; 
936         bt_putid(latchmgr->alloc->right, MIN_lvl+1+nlatchpage);
937         latchmgr->alloc->bits = mgr->page_bits;
938
939         latchmgr->nlatchpage = nlatchpage;
940         latchmgr->latchtotal = nlatchpage * (mgr->page_size / sizeof(BtLatchSet));
941
942         //  initialize latch manager
943
944         latchhash = (mgr->page_size - sizeof(BtLatchMgr)) / sizeof(BtHashEntry);
945
946         //      size of hash table = total number of latchsets
947
948         if( latchhash > latchmgr->latchtotal )
949                 latchhash = latchmgr->latchtotal;
950
951         latchmgr->latchhash = latchhash;
952
953 #ifdef unix
954         if( write (mgr->idx, latchmgr, mgr->page_size) < mgr->page_size )
955                 return bt_mgrclose (mgr), NULL;
956 #else
957         if( !WriteFile (mgr->idx, (char *)latchmgr, mgr->page_size, amt, NULL) )
958                 return bt_mgrclose (mgr), NULL;
959
960         if( *amt < mgr->page_size )
961                 return bt_mgrclose (mgr), NULL;
962 #endif
963
964         memset (latchmgr, 0, 1 << bits);
965         latchmgr->alloc->bits = mgr->page_bits;
966
967         for( lvl=MIN_lvl; lvl--; ) {
968                 slotptr(latchmgr->alloc, 1)->off = mgr->page_size - 3;
969                 bt_putid(slotptr(latchmgr->alloc, 1)->id, lvl ? MIN_lvl - lvl + 1 : 0);         // next(lower) page number
970                 key = keyptr(latchmgr->alloc, 1);
971                 key->len = 2;           // create stopper key
972                 key->key[0] = 0xff;
973                 key->key[1] = 0xff;
974                 latchmgr->alloc->min = mgr->page_size - 3;
975                 latchmgr->alloc->lvl = lvl;
976                 latchmgr->alloc->cnt = 1;
977                 latchmgr->alloc->act = 1;
978 #ifdef unix
979                 if( write (mgr->idx, latchmgr, mgr->page_size) < mgr->page_size )
980                         return bt_mgrclose (mgr), NULL;
981 #else
982                 if( !WriteFile (mgr->idx, (char *)latchmgr, mgr->page_size, amt, NULL) )
983                         return bt_mgrclose (mgr), NULL;
984
985                 if( *amt < mgr->page_size )
986                         return bt_mgrclose (mgr), NULL;
987 #endif
988         }
989
990         // clear out latch manager locks
991         //      and rest of pages to round out segment
992
993         memset(latchmgr, 0, mgr->page_size);
994         last = MIN_lvl + 1;
995
996         while( last <= ((MIN_lvl + 1 + nlatchpage) | mgr->poolmask) ) {
997 #ifdef unix
998                 pwrite(mgr->idx, latchmgr, mgr->page_size, last << mgr->page_bits);
999 #else
1000                 SetFilePointer (mgr->idx, last << mgr->page_bits, NULL, FILE_BEGIN);
1001                 if( !WriteFile (mgr->idx, (char *)latchmgr, mgr->page_size, amt, NULL) )
1002                         return bt_mgrclose (mgr), NULL;
1003                 if( *amt < mgr->page_size )
1004                         return bt_mgrclose (mgr), NULL;
1005 #endif
1006                 last++;
1007         }
1008
1009 mgrlatch:
1010 #ifdef unix
1011         flag = PROT_READ | PROT_WRITE;
1012         mgr->latchmgr = mmap (0, mgr->page_size, flag, MAP_SHARED, mgr->idx, ALLOC_page * mgr->page_size);
1013         if( mgr->latchmgr == MAP_FAILED )
1014                 return bt_mgrclose (mgr), NULL;
1015         mgr->latchsets = (BtLatchSet *)mmap (0, mgr->latchmgr->nlatchpage * mgr->page_size, flag, MAP_SHARED, mgr->idx, LATCH_page * mgr->page_size);
1016         if( mgr->latchsets == MAP_FAILED )
1017                 return bt_mgrclose (mgr), NULL;
1018 #else
1019         flag = PAGE_READWRITE;
1020         mgr->halloc = CreateFileMapping(mgr->idx, NULL, flag, 0, (BT_latchtable / (mgr->page_size / sizeof(BtLatchSet)) + 1 + LATCH_page) * mgr->page_size, NULL);
1021         if( !mgr->halloc )
1022                 return bt_mgrclose (mgr), NULL;
1023
1024         flag = FILE_MAP_WRITE;
1025         mgr->latchmgr = MapViewOfFile(mgr->halloc, flag, 0, 0, (BT_latchtable / (mgr->page_size / sizeof(BtLatchSet)) + 1 + LATCH_page) * mgr->page_size);
1026         if( !mgr->latchmgr )
1027                 return GetLastError(), bt_mgrclose (mgr), NULL;
1028
1029         mgr->latchsets = (void *)((char *)mgr->latchmgr + LATCH_page * mgr->page_size);
1030 #endif
1031
1032 #ifdef unix
1033         free (latchmgr);
1034 #else
1035         VirtualFree (latchmgr, 0, MEM_RELEASE);
1036 #endif
1037         return mgr;
1038 }
1039
1040 //      open BTree access method
1041 //      based on buffer manager
1042
1043 BtDb *bt_open (BtMgr *mgr)
1044 {
1045 BtDb *bt = malloc (sizeof(*bt));
1046
1047         memset (bt, 0, sizeof(*bt));
1048         bt->mgr = mgr;
1049 #ifdef unix
1050         bt->mem = malloc (3 *mgr->page_size);
1051 #else
1052         bt->mem = VirtualAlloc(NULL, 3 * mgr->page_size, MEM_COMMIT, PAGE_READWRITE);
1053 #endif
1054         bt->frame = (BtPage)bt->mem;
1055         bt->zero = (BtPage)(bt->mem + 1 * mgr->page_size);
1056         bt->cursor = (BtPage)(bt->mem + 2 * mgr->page_size);
1057
1058         memset (bt->zero, 0, mgr->page_size);
1059         return bt;
1060 }
1061
1062 //  compare two keys, returning > 0, = 0, or < 0
1063 //  as the comparison value
1064
1065 int keycmp (BtKey key1, unsigned char *key2, uint len2)
1066 {
1067 uint len1 = key1->len;
1068 int ans;
1069
1070         if( ans = memcmp (key1->key, key2, len1 > len2 ? len2 : len1) )
1071                 return ans;
1072
1073         if( len1 > len2 )
1074                 return 1;
1075         if( len1 < len2 )
1076                 return -1;
1077
1078         return 0;
1079 }
1080
1081 //      Buffer Pool mgr
1082
1083 // find segment in pool
1084 // must be called with hashslot idx locked
1085 //      return NULL if not there
1086 //      otherwise return node
1087
1088 BtPool *bt_findpool(BtDb *bt, uid page_no, uint idx)
1089 {
1090 BtPool *pool;
1091 uint slot;
1092
1093         // compute start of hash chain in pool
1094
1095         if( slot = bt->mgr->hash[idx] ) 
1096                 pool = bt->mgr->pool + slot;
1097         else
1098                 return NULL;
1099
1100         page_no &= ~bt->mgr->poolmask;
1101
1102         while( pool->basepage != page_no )
1103           if( pool = pool->hashnext )
1104                 continue;
1105           else
1106                 return NULL;
1107
1108         return pool;
1109 }
1110
1111 // add segment to hash table
1112
1113 void bt_linkhash(BtDb *bt, BtPool *pool, uid page_no, int idx)
1114 {
1115 BtPool *node;
1116 uint slot;
1117
1118         pool->hashprev = pool->hashnext = NULL;
1119         pool->basepage = page_no & ~bt->mgr->poolmask;
1120         pool->pin = CLOCK_bit + 1;
1121
1122         if( slot = bt->mgr->hash[idx] ) {
1123                 node = bt->mgr->pool + slot;
1124                 pool->hashnext = node;
1125                 node->hashprev = pool;
1126         }
1127
1128         bt->mgr->hash[idx] = pool->slot;
1129 }
1130
1131 //  map new buffer pool segment to virtual memory
1132
1133 BTERR bt_mapsegment(BtDb *bt, BtPool *pool, uid page_no)
1134 {
1135 off64_t off = (page_no & ~bt->mgr->poolmask) << bt->mgr->page_bits;
1136 off64_t limit = off + ((bt->mgr->poolmask+1) << bt->mgr->page_bits);
1137 int flag;
1138
1139 #ifdef unix
1140         flag = PROT_READ | ( bt->mgr->mode == BT_ro ? 0 : PROT_WRITE );
1141         pool->map = mmap (0, (bt->mgr->poolmask+1) << bt->mgr->page_bits, flag, MAP_SHARED, bt->mgr->idx, off);
1142         if( pool->map == MAP_FAILED )
1143                 return bt->err = BTERR_map;
1144 #else
1145         flag = ( bt->mgr->mode == BT_ro ? PAGE_READONLY : PAGE_READWRITE );
1146         pool->hmap = CreateFileMapping(bt->mgr->idx, NULL, flag, (DWORD)(limit >> 32), (DWORD)limit, NULL);
1147         if( !pool->hmap )
1148                 return bt->err = BTERR_map;
1149
1150         flag = ( bt->mgr->mode == BT_ro ? FILE_MAP_READ : FILE_MAP_WRITE );
1151         pool->map = MapViewOfFile(pool->hmap, flag, (DWORD)(off >> 32), (DWORD)off, (bt->mgr->poolmask+1) << bt->mgr->page_bits);
1152         if( !pool->map )
1153                 return bt->err = BTERR_map;
1154 #endif
1155         return bt->err = 0;
1156 }
1157
1158 //      calculate page within pool
1159
1160 BtPage bt_page (BtDb *bt, BtPool *pool, uid page_no)
1161 {
1162 uint subpage = (uint)(page_no & bt->mgr->poolmask); // page within mapping
1163 BtPage page;
1164
1165         page = (BtPage)(pool->map + (subpage << bt->mgr->page_bits));
1166         return page;
1167 }
1168
1169 //  release pool pin
1170
1171 void bt_unpinpool (BtPool *pool)
1172 {
1173 #ifdef unix
1174         __sync_fetch_and_add(&pool->pin, -1);
1175 #else
1176         _InterlockedDecrement16 (&pool->pin);
1177 #endif
1178 }
1179
1180 //      find or place requested page in segment-pool
1181 //      return pool table entry, incrementing pin
1182
1183 BtPool *bt_pinpool(BtDb *bt, uid page_no)
1184 {
1185 uint slot, hashidx, idx, victim;
1186 BtPool *pool, *node, *next;
1187
1188         //      lock hash table chain
1189
1190         hashidx = (uint)(page_no >> bt->mgr->seg_bits) % bt->mgr->hashsize;
1191         bt_spinwritelock (&bt->mgr->latch[hashidx]);
1192
1193         //      look up in hash table
1194
1195         if( pool = bt_findpool(bt, page_no, hashidx) ) {
1196 #ifdef unix
1197                 __sync_fetch_and_or(&pool->pin, CLOCK_bit);
1198                 __sync_fetch_and_add(&pool->pin, 1);
1199 #else
1200                 _InterlockedOr16 (&pool->pin, CLOCK_bit);
1201                 _InterlockedIncrement16 (&pool->pin);
1202 #endif
1203                 bt_spinreleasewrite (&bt->mgr->latch[hashidx]);
1204                 return pool;
1205         }
1206
1207         // allocate a new pool node
1208         // and add to hash table
1209
1210 #ifdef unix
1211         slot = __sync_fetch_and_add(&bt->mgr->poolcnt, 1);
1212 #else
1213         slot = _InterlockedIncrement16 (&bt->mgr->poolcnt) - 1;
1214 #endif
1215
1216         if( ++slot < bt->mgr->poolmax ) {
1217                 pool = bt->mgr->pool + slot;
1218                 pool->slot = slot;
1219
1220                 if( bt_mapsegment(bt, pool, page_no) )
1221                         return NULL;
1222
1223                 bt_linkhash(bt, pool, page_no, hashidx);
1224                 bt_spinreleasewrite (&bt->mgr->latch[hashidx]);
1225                 return pool;
1226         }
1227
1228         // pool table is full
1229         //      find best pool entry to evict
1230
1231 #ifdef unix
1232         __sync_fetch_and_add(&bt->mgr->poolcnt, -1);
1233 #else
1234         _InterlockedDecrement16 (&bt->mgr->poolcnt);
1235 #endif
1236
1237         while( 1 ) {
1238 #ifdef unix
1239                 victim = __sync_fetch_and_add(&bt->mgr->evicted, 1);
1240 #else
1241                 victim = _InterlockedIncrement (&bt->mgr->evicted) - 1;
1242 #endif
1243                 victim %= bt->mgr->poolmax;
1244                 pool = bt->mgr->pool + victim;
1245                 idx = (uint)(pool->basepage >> bt->mgr->seg_bits) % bt->mgr->hashsize;
1246
1247                 if( !victim )
1248                         continue;
1249
1250                 // try to get write lock
1251                 //      skip entry if not obtained
1252
1253                 if( !bt_spinwritetry (&bt->mgr->latch[idx]) )
1254                         continue;
1255
1256                 //      skip this entry if
1257                 //      page is pinned
1258                 //  or clock bit is set
1259
1260                 if( pool->pin ) {
1261 #ifdef unix
1262                         __sync_fetch_and_and(&pool->pin, ~CLOCK_bit);
1263 #else
1264                         _InterlockedAnd16 (&pool->pin, ~CLOCK_bit);
1265 #endif
1266                         bt_spinreleasewrite (&bt->mgr->latch[idx]);
1267                         continue;
1268                 }
1269
1270                 // unlink victim pool node from hash table
1271
1272                 if( node = pool->hashprev )
1273                         node->hashnext = pool->hashnext;
1274                 else if( node = pool->hashnext )
1275                         bt->mgr->hash[idx] = node->slot;
1276                 else
1277                         bt->mgr->hash[idx] = 0;
1278
1279                 if( node = pool->hashnext )
1280                         node->hashprev = pool->hashprev;
1281
1282                 bt_spinreleasewrite (&bt->mgr->latch[idx]);
1283
1284                 //      remove old file mapping
1285 #ifdef unix
1286                 munmap (pool->map, (bt->mgr->poolmask+1) << bt->mgr->page_bits);
1287 #else
1288                 FlushViewOfFile(pool->map, 0);
1289                 UnmapViewOfFile(pool->map);
1290                 CloseHandle(pool->hmap);
1291 #endif
1292                 pool->map = NULL;
1293
1294                 //  create new pool mapping
1295                 //  and link into hash table
1296
1297                 if( bt_mapsegment(bt, pool, page_no) )
1298                         return NULL;
1299
1300                 bt_linkhash(bt, pool, page_no, hashidx);
1301                 bt_spinreleasewrite (&bt->mgr->latch[hashidx]);
1302                 return pool;
1303         }
1304 }
1305
1306 // place write, read, or parent lock on requested page_no.
1307
1308 void bt_lockpage(BtLock mode, BtLatchSet *set)
1309 {
1310         switch( mode ) {
1311         case BtLockRead:
1312                 bt_readlock (set->readwr);
1313                 break;
1314         case BtLockWrite:
1315                 bt_writelock (set->readwr);
1316                 break;
1317         case BtLockAccess:
1318                 bt_readlock (set->access);
1319                 break;
1320         case BtLockDelete:
1321                 bt_writelock (set->access);
1322                 break;
1323         case BtLockParent:
1324                 bt_writelock (set->parent);
1325                 break;
1326         }
1327 }
1328
1329 // remove write, read, or parent lock on requested page
1330
1331 void bt_unlockpage(BtLock mode, BtLatchSet *set)
1332 {
1333         switch( mode ) {
1334         case BtLockRead:
1335                 bt_releaseread (set->readwr);
1336                 break;
1337         case BtLockWrite:
1338                 bt_releasewrite (set->readwr);
1339                 break;
1340         case BtLockAccess:
1341                 bt_releaseread (set->access);
1342                 break;
1343         case BtLockDelete:
1344                 bt_releasewrite (set->access);
1345                 break;
1346         case BtLockParent:
1347                 bt_releasewrite (set->parent);
1348                 break;
1349         }
1350 }
1351
1352 //      allocate a new page and write page into it
1353
1354 uid bt_newpage(BtDb *bt, BtPage page)
1355 {
1356 BtPageSet set[1];
1357 uid new_page;
1358 int reuse;
1359
1360         //      lock allocation page
1361
1362         bt_spinwritelock(bt->mgr->latchmgr->lock);
1363
1364         // use empty chain first
1365         // else allocate empty page
1366
1367         if( new_page = bt_getid(bt->mgr->latchmgr->alloc[1].right) ) {
1368                 if( set->pool = bt_pinpool (bt, new_page) )
1369                         set->page = bt_page (bt, set->pool, new_page);
1370                 else
1371                         return 0;
1372
1373                 bt_putid(bt->mgr->latchmgr->alloc[1].right, bt_getid(set->page->right));
1374                 bt_unpinpool (set->pool);
1375                 reuse = 1;
1376         } else {
1377                 new_page = bt_getid(bt->mgr->latchmgr->alloc->right);
1378                 bt_putid(bt->mgr->latchmgr->alloc->right, new_page+1);
1379                 reuse = 0;
1380         }
1381 #ifdef unix
1382         if( pwrite(bt->mgr->idx, page, bt->mgr->page_size, new_page << bt->mgr->page_bits) < bt->mgr->page_size )
1383                 return bt->err = BTERR_wrt, 0;
1384
1385         // if writing first page of pool block, zero last page in the block
1386
1387         if( !reuse && bt->mgr->poolmask > 0 && (new_page & bt->mgr->poolmask) == 0 )
1388         {
1389                 // use zero buffer to write zeros
1390                 if( pwrite(bt->mgr->idx,bt->zero, bt->mgr->page_size, (new_page | bt->mgr->poolmask) << bt->mgr->page_bits) < bt->mgr->page_size )
1391                         return bt->err = BTERR_wrt, 0;
1392         }
1393 #else
1394         //      bring new page into pool and copy page.
1395         //      this will extend the file into the new pages.
1396
1397         if( set->pool = bt_pinpool (bt, new_page) )
1398                 set->page = bt_page (bt, set->pool, new_page);
1399         else
1400                 return 0;
1401
1402         memcpy(set->page, page, bt->mgr->page_size);
1403         bt_unpinpool (set->pool);
1404 #endif
1405         // unlock allocation latch and return new page no
1406
1407         bt_spinreleasewrite(bt->mgr->latchmgr->lock);
1408         return new_page;
1409 }
1410
1411 //  find slot in page for given key at a given level
1412
1413 int bt_findslot (BtPageSet *set, unsigned char *key, uint len)
1414 {
1415 uint diff, higher = set->page->cnt, low = 1, slot;
1416 uint good = 0;
1417
1418         //        make stopper key an infinite fence value
1419
1420         if( bt_getid (set->page->right) )
1421                 higher++;
1422         else
1423                 good++;
1424
1425         //      low is the lowest candidate.
1426         //  loop ends when they meet
1427
1428         //  higher is already
1429         //      tested as .ge. the passed key.
1430
1431         while( diff = higher - low ) {
1432                 slot = low + ( diff >> 1 );
1433                 if( keycmp (keyptr(set->page, slot), key, len) < 0 )
1434                         low = slot + 1;
1435                 else
1436                         higher = slot, good++;
1437         }
1438
1439         //      return zero if key is on right link page
1440
1441         return good ? higher : 0;
1442 }
1443
1444 //  find and load page at given level for given key
1445 //      leave page rd or wr locked as requested
1446
1447 int bt_loadpage (BtDb *bt, BtPageSet *set, unsigned char *key, uint len, uint lvl, BtLock lock)
1448 {
1449 uid page_no = ROOT_page, prevpage = 0;
1450 uint drill = 0xff, slot;
1451 BtLatchSet *prevlatch;
1452 uint mode, prevmode;
1453 BtPool *prevpool;
1454
1455   //  start at root of btree and drill down
1456
1457   do {
1458         // determine lock mode of drill level
1459         mode = (drill == lvl) ? lock : BtLockRead; 
1460
1461         set->latch = bt_pinlatch (bt, page_no);
1462         set->page_no = page_no;
1463
1464         // pin page contents
1465
1466         if( set->pool = bt_pinpool (bt, page_no) )
1467                 set->page = bt_page (bt, set->pool, page_no);
1468         else
1469                 return 0;
1470
1471         // obtain access lock using lock chaining with Access mode
1472
1473         if( page_no > ROOT_page )
1474           bt_lockpage(BtLockAccess, set->latch);
1475
1476         //      release & unpin parent page
1477
1478         if( prevpage ) {
1479           bt_unlockpage(prevmode, prevlatch);
1480           bt_unpinlatch (prevlatch);
1481           bt_unpinpool (prevpool);
1482           prevpage = 0;
1483         }
1484
1485         // obtain read lock using lock chaining
1486
1487         bt_lockpage(mode, set->latch);
1488
1489         if( set->page->free )
1490                 return bt->err = BTERR_struct, 0;
1491
1492         if( page_no > ROOT_page )
1493           bt_unlockpage(BtLockAccess, set->latch);
1494
1495         // re-read and re-lock root after determining actual level of root
1496
1497         if( set->page->lvl != drill) {
1498                 if( set->page_no != ROOT_page )
1499                         return bt->err = BTERR_struct, 0;
1500                         
1501                 drill = set->page->lvl;
1502
1503                 if( lock != BtLockRead && drill == lvl ) {
1504                   bt_unlockpage(mode, set->latch);
1505                   bt_unpinlatch (set->latch);
1506                   bt_unpinpool (set->pool);
1507                   continue;
1508                 }
1509         }
1510
1511         prevpage = set->page_no;
1512         prevlatch = set->latch;
1513         prevpool = set->pool;
1514         prevmode = mode;
1515
1516         //  find key on page at this level
1517         //  and descend to requested level
1518
1519         if( !set->page->kill )
1520          if( slot = bt_findslot (set, key, len) ) {
1521           if( drill == lvl )
1522                 return slot;
1523
1524           while( slotptr(set->page, slot)->dead )
1525                 if( slot++ < set->page->cnt )
1526                         continue;
1527                 else
1528                         goto slideright;
1529
1530           page_no = bt_getid(slotptr(set->page, slot)->id);
1531           drill--;
1532           continue;
1533          }
1534
1535         //  or slide right into next page
1536
1537 slideright:
1538         page_no = bt_getid(set->page->right);
1539
1540   } while( page_no );
1541
1542   // return error on end of right chain
1543
1544   bt->err = BTERR_struct;
1545   return 0;     // return error
1546 }
1547
1548 //      return page to free list
1549 //      page must be delete & write locked
1550
1551 void bt_freepage (BtDb *bt, BtPageSet *set)
1552 {
1553         //      lock allocation page
1554
1555         bt_spinwritelock (bt->mgr->latchmgr->lock);
1556
1557         //      store chain in second right
1558         bt_putid(set->page->right, bt_getid(bt->mgr->latchmgr->alloc[1].right));
1559         bt_putid(bt->mgr->latchmgr->alloc[1].right, set->page_no);
1560         set->page->free = 1;
1561
1562         // unlock released page
1563
1564         bt_unlockpage (BtLockDelete, set->latch);
1565         bt_unlockpage (BtLockWrite, set->latch);
1566         bt_unpinlatch (set->latch);
1567         bt_unpinpool (set->pool);
1568
1569         // unlock allocation page
1570
1571         bt_spinreleasewrite (bt->mgr->latchmgr->lock);
1572 }
1573
1574 //      a fence key was deleted from a page
1575 //      push new fence value upwards
1576
1577 BTERR bt_fixfence (BtDb *bt, BtPageSet *set, uint lvl)
1578 {
1579 unsigned char leftkey[256], rightkey[256];
1580 uid page_no;
1581 BtKey ptr;
1582
1583         //      remove the old fence value
1584
1585         ptr = keyptr(set->page, set->page->cnt);
1586         memcpy (rightkey, ptr, ptr->len + 1);
1587
1588         memset (slotptr(set->page, set->page->cnt--), 0, sizeof(BtSlot));
1589         set->page->dirty = 1;
1590
1591         ptr = keyptr(set->page, set->page->cnt);
1592         memcpy (leftkey, ptr, ptr->len + 1);
1593         page_no = set->page_no;
1594
1595         bt_lockpage (BtLockParent, set->latch);
1596         bt_unlockpage (BtLockWrite, set->latch);
1597
1598         //      insert new (now smaller) fence key
1599
1600         if( bt_insertkey (bt, leftkey+1, *leftkey, lvl+1, page_no, time(NULL)) )
1601           return bt->err;
1602
1603         //      now delete old fence key
1604
1605         if( bt_deletekey (bt, rightkey+1, *rightkey, lvl+1) )
1606                 return bt->err;
1607
1608         bt_unlockpage (BtLockParent, set->latch);
1609         bt_unpinlatch(set->latch);
1610         bt_unpinpool (set->pool);
1611         return 0;
1612 }
1613
1614 //      root has a single child
1615 //      collapse a level from the tree
1616
1617 BTERR bt_collapseroot (BtDb *bt, BtPageSet *root)
1618 {
1619 BtPageSet child[1];
1620 uint idx;
1621
1622   // find the child entry and promote as new root contents
1623
1624   do {
1625         for( idx = 0; idx++ < root->page->cnt; )
1626           if( !slotptr(root->page, idx)->dead )
1627                 break;
1628
1629         child->page_no = bt_getid (slotptr(root->page, idx)->id);
1630
1631         child->latch = bt_pinlatch (bt, child->page_no);
1632         bt_lockpage (BtLockDelete, child->latch);
1633         bt_lockpage (BtLockWrite, child->latch);
1634
1635         if( child->pool = bt_pinpool (bt, child->page_no) )
1636                 child->page = bt_page (bt, child->pool, child->page_no);
1637         else
1638                 return bt->err;
1639
1640         memcpy (root->page, child->page, bt->mgr->page_size);
1641         bt_freepage (bt, child);
1642
1643   } while( root->page->lvl > 1 && root->page->act == 1 );
1644
1645   bt_unlockpage (BtLockWrite, root->latch);
1646   bt_unpinlatch (root->latch);
1647   bt_unpinpool (root->pool);
1648   return 0;
1649 }
1650
1651 //  find and delete key on page by marking delete flag bit
1652 //  if page becomes empty, delete it from the btree
1653
1654 BTERR bt_deletekey (BtDb *bt, unsigned char *key, uint len, uint lvl)
1655 {
1656 unsigned char lowerfence[256], higherfence[256];
1657 uint slot, idx, dirty = 0, fence, found;
1658 BtPageSet set[1], right[1];
1659 BtKey ptr;
1660
1661         if( slot = bt_loadpage (bt, set, key, len, lvl, BtLockWrite) )
1662                 ptr = keyptr(set->page, slot);
1663         else
1664                 return bt->err;
1665
1666         //      are we deleting a fence slot?
1667
1668         fence = slot == set->page->cnt;
1669
1670         // if key is found delete it, otherwise ignore request
1671
1672         if( found = !keycmp (ptr, key, len) )
1673           if( found = slotptr(set->page, slot)->dead == 0 ) {
1674                 dirty = slotptr(set->page, slot)->dead = 1;
1675                 set->page->dirty = 1;
1676                 set->page->act--;
1677
1678                 // collapse empty slots
1679
1680                 while( idx = set->page->cnt - 1 )
1681                   if( slotptr(set->page, idx)->dead ) {
1682                         *slotptr(set->page, idx) = *slotptr(set->page, idx + 1);
1683                         memset (slotptr(set->page, set->page->cnt--), 0, sizeof(BtSlot));
1684                   } else
1685                         break;
1686                   }
1687
1688         //      did we delete a fence key in an upper level?
1689
1690         if( dirty && lvl && set->page->act && fence )
1691           if( bt_fixfence (bt, set, lvl) )
1692                 return bt->err;
1693           else
1694                 return bt->found = found, 0;
1695
1696         //      is this a collapsed root?
1697
1698         if( lvl > 1 && set->page_no == ROOT_page && set->page->act == 1 )
1699           if( bt_collapseroot (bt, set) )
1700                 return bt->err;
1701           else
1702                 return bt->found = found, 0;
1703
1704         //      return if page is not empty
1705
1706         if( set->page->act ) {
1707                 bt_unlockpage(BtLockWrite, set->latch);
1708                 bt_unpinlatch (set->latch);
1709                 bt_unpinpool (set->pool);
1710                 return bt->found = found, 0;
1711         }
1712
1713         //      cache copy of fence key
1714         //      to post in parent
1715
1716         ptr = keyptr(set->page, set->page->cnt);
1717         memcpy (lowerfence, ptr, ptr->len + 1);
1718
1719         //      obtain lock on right page
1720
1721         right->page_no = bt_getid(set->page->right);
1722         right->latch = bt_pinlatch (bt, right->page_no);
1723         bt_lockpage (BtLockWrite, right->latch);
1724
1725         // pin page contents
1726
1727         if( right->pool = bt_pinpool (bt, right->page_no) )
1728                 right->page = bt_page (bt, right->pool, right->page_no);
1729         else
1730                 return 0;
1731
1732         if( right->page->kill )
1733                 return bt->err = BTERR_struct;
1734
1735         // pull contents of right peer into our empty page
1736
1737         memcpy (set->page, right->page, bt->mgr->page_size);
1738
1739         // cache copy of key to update
1740
1741         ptr = keyptr(right->page, right->page->cnt);
1742         memcpy (higherfence, ptr, ptr->len + 1);
1743
1744         // mark right page deleted and point it to left page
1745         //      until we can post parent updates
1746
1747         bt_putid (right->page->right, set->page_no);
1748         right->page->kill = 1;
1749
1750         bt_lockpage (BtLockParent, right->latch);
1751         bt_unlockpage (BtLockWrite, right->latch);
1752
1753         bt_lockpage (BtLockParent, set->latch);
1754         bt_unlockpage (BtLockWrite, set->latch);
1755
1756         // redirect higher key directly to our new node contents
1757
1758         if( bt_insertkey (bt, higherfence+1, *higherfence, lvl+1, set->page_no, time(NULL)) )
1759           return bt->err;
1760
1761         //      delete old lower key to our node
1762
1763         if( bt_deletekey (bt, lowerfence+1, *lowerfence, lvl+1) )
1764           return bt->err;
1765
1766         //      obtain delete and write locks to right node
1767
1768         bt_unlockpage (BtLockParent, right->latch);
1769         bt_lockpage (BtLockDelete, right->latch);
1770         bt_lockpage (BtLockWrite, right->latch);
1771         bt_freepage (bt, right);
1772
1773         bt_unlockpage (BtLockParent, set->latch);
1774         bt_unpinlatch (set->latch);
1775         bt_unpinpool (set->pool);
1776         bt->found = found;
1777         return 0;
1778 }
1779
1780 //      find key in leaf level and return row-id
1781
1782 uid bt_findkey (BtDb *bt, unsigned char *key, uint len)
1783 {
1784 BtPageSet set[1];
1785 uint  slot;
1786 uid id = 0;
1787 BtKey ptr;
1788
1789         if( slot = bt_loadpage (bt, set, key, len, 0, BtLockRead) )
1790                 ptr = keyptr(set->page, slot);
1791         else
1792                 return 0;
1793
1794         // if key exists, return row-id
1795         //      otherwise return 0
1796
1797         if( slot <= set->page->cnt )
1798           if( !keycmp (ptr, key, len) )
1799                 id = bt_getid(slotptr(set->page,slot)->id);
1800
1801         bt_unlockpage (BtLockRead, set->latch);
1802         bt_unpinlatch (set->latch);
1803         bt_unpinpool (set->pool);
1804         return id;
1805 }
1806
1807 //      check page for space available,
1808 //      clean if necessary and return
1809 //      0 - page needs splitting
1810 //      >0  new slot value
1811
1812 uint bt_cleanpage(BtDb *bt, BtPage page, uint amt, uint slot)
1813 {
1814 uint nxt = bt->mgr->page_size;
1815 uint cnt = 0, idx = 0;
1816 uint max = page->cnt;
1817 uint newslot = max;
1818 BtKey key;
1819
1820         if( page->min >= (max+1) * sizeof(BtSlot) + sizeof(*page) + amt + 1 )
1821                 return slot;
1822
1823         //      skip cleanup if nothing to reclaim
1824
1825         if( !page->dirty )
1826                 return 0;
1827
1828         memcpy (bt->frame, page, bt->mgr->page_size);
1829
1830         // skip page info and set rest of page to zero
1831
1832         memset (page+1, 0, bt->mgr->page_size - sizeof(*page));
1833         page->dirty = 0;
1834         page->act = 0;
1835
1836         // try cleaning up page first
1837         // by removing deleted keys
1838
1839         while( cnt++ < max ) {
1840                 if( cnt == slot )
1841                         newslot = idx + 1;
1842                 if( cnt < max && slotptr(bt->frame,cnt)->dead )
1843                         continue;
1844
1845                 // copy the key across
1846
1847                 key = keyptr(bt->frame, cnt);
1848                 nxt -= key->len + 1;
1849                 memcpy ((unsigned char *)page + nxt, key, key->len + 1);
1850
1851                 // copy slot
1852
1853                 memcpy(slotptr(page, ++idx)->id, slotptr(bt->frame, cnt)->id, BtId);
1854                 if( !(slotptr(page, idx)->dead = slotptr(bt->frame, cnt)->dead) )
1855                         page->act++;
1856                 slotptr(page, idx)->tod = slotptr(bt->frame, cnt)->tod;
1857                 slotptr(page, idx)->off = nxt;
1858         }
1859
1860         page->min = nxt;
1861         page->cnt = idx;
1862
1863         //      see if page has enough space now, or does it need splitting?
1864
1865         if( page->min >= (idx+1) * sizeof(BtSlot) + sizeof(*page) + amt + 1 )
1866                 return newslot;
1867
1868         return 0;
1869 }
1870
1871 // split the root and raise the height of the btree
1872
1873 BTERR bt_splitroot(BtDb *bt, BtPageSet *root, unsigned char *leftkey, uid page_no2)
1874 {
1875 uint nxt = bt->mgr->page_size;
1876 uid left;
1877
1878         //  Obtain an empty page to use, and copy the current
1879         //  root contents into it, e.g. lower keys
1880
1881         if( !(left = bt_newpage(bt, root->page)) )
1882                 return bt->err;
1883
1884         // preserve the page info at the bottom
1885         // of higher keys and set rest to zero
1886
1887         memset(root->page+1, 0, bt->mgr->page_size - sizeof(*root->page));
1888
1889         // insert lower keys page fence key on newroot page as first key
1890
1891         nxt -= *leftkey + 1;
1892         memcpy ((unsigned char *)root->page + nxt, leftkey, *leftkey + 1);
1893         bt_putid(slotptr(root->page, 1)->id, left);
1894         slotptr(root->page, 1)->off = nxt;
1895         
1896         // insert stopper key on newroot page
1897         // and increase the root height
1898
1899         nxt -= 3;
1900         ((unsigned char *)root->page)[nxt] = 2;
1901         ((unsigned char *)root->page)[nxt+1] = 0xff;
1902         ((unsigned char *)root->page)[nxt+2] = 0xff;
1903         bt_putid(slotptr(root->page, 2)->id, page_no2);
1904         slotptr(root->page, 2)->off = nxt;
1905
1906         bt_putid(root->page->right, 0);
1907         root->page->min = nxt;          // reset lowest used offset and key count
1908         root->page->cnt = 2;
1909         root->page->act = 2;
1910         root->page->lvl++;
1911
1912         // release and unpin root
1913
1914         bt_unlockpage(BtLockWrite, root->latch);
1915         bt_unpinlatch (root->latch);
1916         bt_unpinpool (root->pool);
1917         return 0;
1918 }
1919
1920 //  split already locked full node
1921 //      return unlocked.
1922
1923 BTERR bt_splitpage (BtDb *bt, BtPageSet *set)
1924 {
1925 uint cnt = 0, idx = 0, max, nxt = bt->mgr->page_size;
1926 unsigned char fencekey[256], rightkey[256];
1927 uint lvl = set->page->lvl;
1928 BtPageSet right[1];
1929 uint prev;
1930 BtKey key;
1931
1932         //  split higher half of keys to bt->frame
1933
1934         memset (bt->frame, 0, bt->mgr->page_size);
1935         max = set->page->cnt;
1936         cnt = max / 2;
1937         idx = 0;
1938
1939         while( cnt++ < max ) {
1940                 key = keyptr(set->page, cnt);
1941                 nxt -= key->len + 1;
1942                 memcpy ((unsigned char *)bt->frame + nxt, key, key->len + 1);
1943
1944                 memcpy(slotptr(bt->frame,++idx)->id, slotptr(set->page,cnt)->id, BtId);
1945                 if( !(slotptr(bt->frame, idx)->dead = slotptr(set->page, cnt)->dead) )
1946                         bt->frame->act++;
1947                 slotptr(bt->frame, idx)->tod = slotptr(set->page, cnt)->tod;
1948                 slotptr(bt->frame, idx)->off = nxt;
1949         }
1950
1951         // remember existing fence key for new page to the right
1952
1953         memcpy (rightkey, key, key->len + 1);
1954
1955         bt->frame->bits = bt->mgr->page_bits;
1956         bt->frame->min = nxt;
1957         bt->frame->cnt = idx;
1958         bt->frame->lvl = lvl;
1959
1960         // link right node
1961
1962         if( set->page_no > ROOT_page )
1963                 memcpy (bt->frame->right, set->page->right, BtId);
1964
1965         //      get new free page and write higher keys to it.
1966
1967         if( !(right->page_no = bt_newpage(bt, bt->frame)) )
1968                 return bt->err;
1969
1970         //      update lower keys to continue in old page
1971
1972         memcpy (bt->frame, set->page, bt->mgr->page_size);
1973         memset (set->page+1, 0, bt->mgr->page_size - sizeof(*set->page));
1974         nxt = bt->mgr->page_size;
1975         set->page->dirty = 0;
1976         set->page->act = 0;
1977         cnt = 0;
1978         idx = 0;
1979
1980         //  assemble page of smaller keys
1981
1982         while( cnt++ < max / 2 ) {
1983                 key = keyptr(bt->frame, cnt);
1984                 nxt -= key->len + 1;
1985                 memcpy ((unsigned char *)set->page + nxt, key, key->len + 1);
1986                 memcpy(slotptr(set->page,++idx)->id, slotptr(bt->frame,cnt)->id, BtId);
1987                 slotptr(set->page, idx)->tod = slotptr(bt->frame, cnt)->tod;
1988                 slotptr(set->page, idx)->off = nxt;
1989                 set->page->act++;
1990         }
1991
1992         // remember fence key for smaller page
1993
1994         memcpy(fencekey, key, key->len + 1);
1995
1996         bt_putid(set->page->right, right->page_no);
1997         set->page->min = nxt;
1998         set->page->cnt = idx;
1999
2000         // if current page is the root page, split it
2001
2002         if( set->page_no == ROOT_page )
2003                 return bt_splitroot (bt, set, fencekey, right->page_no);
2004
2005         // insert new fences in their parent pages
2006
2007         right->latch = bt_pinlatch (bt, right->page_no);
2008         bt_lockpage (BtLockParent, right->latch);
2009
2010         bt_lockpage (BtLockParent, set->latch);
2011         bt_unlockpage (BtLockWrite, set->latch);
2012
2013         // insert new fence for reformulated left block of smaller keys
2014
2015         if( bt_insertkey (bt, fencekey+1, *fencekey, lvl+1, set->page_no, time(NULL)) )
2016                 return bt->err;
2017
2018         // switch fence for right block of larger keys to new right page
2019
2020         if( bt_insertkey (bt, rightkey+1, *rightkey, lvl+1, right->page_no, time(NULL)) )
2021                 return bt->err;
2022
2023         bt_unlockpage (BtLockParent, set->latch);
2024         bt_unpinlatch (set->latch);
2025         bt_unpinpool (set->pool);
2026
2027         bt_unlockpage (BtLockParent, right->latch);
2028         bt_unpinlatch (right->latch);
2029         return 0;
2030 }
2031
2032 //  Insert new key into the btree at given level.
2033
2034 BTERR bt_insertkey (BtDb *bt, unsigned char *key, uint len, uint lvl, uid id, uint tod)
2035 {
2036 BtPageSet set[1];
2037 uint slot, idx;
2038 BtKey ptr;
2039
2040         while( 1 ) {
2041                 if( slot = bt_loadpage (bt, set, key, len, lvl, BtLockWrite) )
2042                         ptr = keyptr(set->page, slot);
2043                 else
2044                 {
2045                         if( !bt->err )
2046                                 bt->err = BTERR_ovflw;
2047                         return bt->err;
2048                 }
2049
2050                 // if key already exists, update id and return
2051
2052                 if( !keycmp (ptr, key, len) ) {
2053                         if( slotptr(set->page, slot)->dead )
2054                                 set->page->act++;
2055                         slotptr(set->page, slot)->dead = 0;
2056                         slotptr(set->page, slot)->tod = tod;
2057                         bt_putid(slotptr(set->page,slot)->id, id);
2058                         bt_unlockpage(BtLockWrite, set->latch);
2059                         bt_unpinlatch (set->latch);
2060                         bt_unpinpool (set->pool);
2061                         return 0;
2062                 }
2063
2064                 // check if page has enough space
2065
2066                 if( slot = bt_cleanpage (bt, set->page, len, slot) )
2067                         break;
2068
2069                 if( bt_splitpage (bt, set) )
2070                         return bt->err;
2071         }
2072
2073         // calculate next available slot and copy key into page
2074
2075         set->page->min -= len + 1; // reset lowest used offset
2076         ((unsigned char *)set->page)[set->page->min] = len;
2077         memcpy ((unsigned char *)set->page + set->page->min +1, key, len );
2078
2079         for( idx = slot; idx < set->page->cnt; idx++ )
2080           if( slotptr(set->page, idx)->dead )
2081                 break;
2082
2083         // now insert key into array before slot
2084
2085         if( idx == set->page->cnt )
2086                 idx++, set->page->cnt++;
2087
2088         set->page->act++;
2089
2090         while( idx > slot )
2091                 *slotptr(set->page, idx) = *slotptr(set->page, idx -1), idx--;
2092
2093         bt_putid(slotptr(set->page,slot)->id, id);
2094         slotptr(set->page, slot)->off = set->page->min;
2095         slotptr(set->page, slot)->tod = tod;
2096         slotptr(set->page, slot)->dead = 0;
2097
2098         bt_unlockpage (BtLockWrite, set->latch);
2099         bt_unpinlatch (set->latch);
2100         bt_unpinpool (set->pool);
2101         return 0;
2102 }
2103
2104 //  cache page of keys into cursor and return starting slot for given key
2105
2106 uint bt_startkey (BtDb *bt, unsigned char *key, uint len)
2107 {
2108 BtPageSet set[1];
2109 uint slot;
2110
2111         // cache page for retrieval
2112
2113         if( slot = bt_loadpage (bt, set, key, len, 0, BtLockRead) )
2114           memcpy (bt->cursor, set->page, bt->mgr->page_size);
2115         else
2116           return 0;
2117
2118         bt->cursor_page = set->page_no;
2119
2120         bt_unlockpage(BtLockRead, set->latch);
2121         bt_unpinlatch (set->latch);
2122         bt_unpinpool (set->pool);
2123         return slot;
2124 }
2125
2126 //  return next slot for cursor page
2127 //  or slide cursor right into next page
2128
2129 uint bt_nextkey (BtDb *bt, uint slot)
2130 {
2131 BtPageSet set[1];
2132 uid right;
2133
2134   do {
2135         right = bt_getid(bt->cursor->right);
2136
2137         while( slot++ < bt->cursor->cnt )
2138           if( slotptr(bt->cursor,slot)->dead )
2139                 continue;
2140           else if( right || (slot < bt->cursor->cnt) ) // skip infinite stopper
2141                 return slot;
2142           else
2143                 break;
2144
2145         if( !right )
2146                 break;
2147
2148         bt->cursor_page = right;
2149
2150         if( set->pool = bt_pinpool (bt, right) )
2151                 set->page = bt_page (bt, set->pool, right);
2152         else
2153                 return 0;
2154
2155         set->latch = bt_pinlatch (bt, right);
2156     bt_lockpage(BtLockRead, set->latch);
2157
2158         memcpy (bt->cursor, set->page, bt->mgr->page_size);
2159
2160         bt_unlockpage(BtLockRead, set->latch);
2161         bt_unpinlatch (set->latch);
2162         bt_unpinpool (set->pool);
2163         slot = 0;
2164
2165   } while( 1 );
2166
2167   return bt->err = 0;
2168 }
2169
2170 BtKey bt_key(BtDb *bt, uint slot)
2171 {
2172         return keyptr(bt->cursor, slot);
2173 }
2174
2175 uid bt_uid(BtDb *bt, uint slot)
2176 {
2177         return bt_getid(slotptr(bt->cursor,slot)->id);
2178 }
2179
2180 uint bt_tod(BtDb *bt, uint slot)
2181 {
2182         return slotptr(bt->cursor,slot)->tod;
2183 }
2184
2185
2186 #ifdef STANDALONE
2187
2188 #ifndef unix
2189 double getCpuTime(int type)
2190 {
2191 FILETIME crtime[1];
2192 FILETIME xittime[1];
2193 FILETIME systime[1];
2194 FILETIME usrtime[1];
2195 SYSTEMTIME timeconv[1];
2196 double ans = 0;
2197
2198         memset (timeconv, 0, sizeof(SYSTEMTIME));
2199
2200         switch( type ) {
2201         case 0:
2202                 GetSystemTimeAsFileTime (xittime);
2203                 FileTimeToSystemTime (xittime, timeconv);
2204                 ans = (double)timeconv->wDayOfWeek * 3600 * 24;
2205                 break;
2206         case 1:
2207                 GetProcessTimes (GetCurrentProcess(), crtime, xittime, systime, usrtime);
2208                 FileTimeToSystemTime (usrtime, timeconv);
2209                 break;
2210         case 2:
2211                 GetProcessTimes (GetCurrentProcess(), crtime, xittime, systime, usrtime);
2212                 FileTimeToSystemTime (systime, timeconv);
2213                 break;
2214         }
2215
2216         ans += (double)timeconv->wHour * 3600;
2217         ans += (double)timeconv->wMinute * 60;
2218         ans += (double)timeconv->wSecond;
2219         ans += (double)timeconv->wMilliseconds / 1000;
2220         return ans;
2221 }
2222 #else
2223 #include <time.h>
2224 #include <sys/resource.h>
2225
2226 double getCpuTime(int type)
2227 {
2228 struct rusage used[1];
2229 struct timeval tv[1];
2230
2231         switch( type ) {
2232         case 0:
2233                 gettimeofday(tv, NULL);
2234                 return (double)tv->tv_sec + (double)tv->tv_usec / 1000000;
2235
2236         case 1:
2237                 getrusage(RUSAGE_SELF, used);
2238                 return (double)used->ru_utime.tv_sec + (double)used->ru_utime.tv_usec / 1000000;
2239
2240         case 2:
2241                 getrusage(RUSAGE_SELF, used);
2242                 return (double)used->ru_stime.tv_sec + (double)used->ru_stime.tv_usec / 1000000;
2243         }
2244
2245         return 0;
2246 }
2247 #endif
2248
2249 void bt_latchaudit (BtDb *bt)
2250 {
2251 ushort idx, hashidx;
2252 uid next, page_no;
2253 BtPageSet set[1];
2254 BtKey ptr;
2255
2256 #ifdef unix
2257         for( idx = 1; idx <= bt->mgr->latchmgr->latchdeployed; idx++ ) {
2258                 set->latch = bt->mgr->latchsets + idx;
2259                 if( set->latch->pin ) {
2260                         fprintf(stderr, "latchset %d pinned for page %.6x\n", idx, set->latch->page_no);
2261                         set->latch->pin = 0;
2262                 }
2263         }
2264
2265         for( hashidx = 0; hashidx < bt->mgr->latchmgr->latchhash; hashidx++ ) {
2266           if( idx = bt->mgr->latchmgr->table[hashidx].slot ) do {
2267                 set->latch = bt->mgr->latchsets + idx;
2268                 if( set->latch->hash != hashidx )
2269                         fprintf(stderr, "latchset %d wrong hashidx\n", idx);
2270                 if( set->latch->pin )
2271                         fprintf(stderr, "latchset %d pinned for page %.8x\n", idx, set->latch->page_no);
2272           } while( idx = set->latch->next );
2273         }
2274
2275         next = bt->mgr->latchmgr->nlatchpage + LATCH_page;
2276         page_no = LEAF_page;
2277
2278         while( page_no < bt_getid(bt->mgr->latchmgr->alloc->right) ) {
2279                 pread (bt->mgr->idx, bt->frame, bt->mgr->page_size, page_no << bt->mgr->page_bits);
2280                 if( !bt->frame->free )
2281                  for( idx = 0; idx++ < bt->frame->cnt - 1; ) {
2282                   ptr = keyptr(bt->frame, idx+1);
2283                   if( keycmp (keyptr(bt->frame, idx), ptr->key, ptr->len) >= 0 )
2284                         fprintf(stderr, "page %.8x idx %.2x out of order\n", page_no, idx);
2285                  }
2286
2287                 if( page_no > LEAF_page )
2288                         next = page_no + 1;
2289                 page_no = next;
2290         }
2291 #endif
2292 }
2293
2294 typedef struct {
2295         char type, idx;
2296         char *infile;
2297         BtMgr *mgr;
2298         int num;
2299 } ThreadArg;
2300
2301 //  standalone program to index file of keys
2302 //  then list them onto std-out
2303
2304 #ifdef unix
2305 void *index_file (void *arg)
2306 #else
2307 uint __stdcall index_file (void *arg)
2308 #endif
2309 {
2310 int line = 0, found = 0, cnt = 0;
2311 uid next, page_no = LEAF_page;  // start on first page of leaves
2312 unsigned char key[256];
2313 ThreadArg *args = arg;
2314 int ch, len = 0, slot;
2315 BtPageSet set[1];
2316 time_t tod[1];
2317 BtKey ptr;
2318 BtDb *bt;
2319 FILE *in;
2320
2321         bt = bt_open (args->mgr);
2322         time (tod);
2323
2324         switch(args->type | 0x20)
2325         {
2326         case 'a':
2327                 fprintf(stderr, "started latch mgr audit\n");
2328                 bt_latchaudit (bt);
2329                 fprintf(stderr, "finished latch mgr audit\n");
2330                 break;
2331
2332         case 'w':
2333                 fprintf(stderr, "started indexing for %s\n", args->infile);
2334                 if( in = fopen (args->infile, "rb") )
2335                   while( ch = getc(in), ch != EOF )
2336                         if( ch == '\n' )
2337                         {
2338                           line++;
2339
2340                           if( args->num == 1 )
2341                                 sprintf((char *)key+len, "%.9d", 1000000000 - line), len += 9;
2342
2343                           else if( args->num )
2344                                 sprintf((char *)key+len, "%.9d", line + args->idx * args->num), len += 9;
2345
2346                           if( bt_insertkey (bt, key, len, 0, line, *tod) )
2347                                 fprintf(stderr, "Error %d Line: %d\n", bt->err, line), exit(0);
2348                           len = 0;
2349                         }
2350                         else if( len < 255 )
2351                                 key[len++] = ch;
2352                 fprintf(stderr, "finished %s for %d keys\n", args->infile, line);
2353                 break;
2354
2355         case 'd':
2356                 fprintf(stderr, "started deleting keys for %s\n", args->infile);
2357                 if( in = fopen (args->infile, "rb") )
2358                   while( ch = getc(in), ch != EOF )
2359                         if( ch == '\n' )
2360                         {
2361                           line++;
2362                           if( args->num == 1 )
2363                                 sprintf((char *)key+len, "%.9d", 1000000000 - line), len += 9;
2364
2365                           else if( args->num )
2366                                 sprintf((char *)key+len, "%.9d", line + args->idx * args->num), len += 9;
2367
2368                           if( bt_deletekey (bt, key, len, 0) )
2369                                 fprintf(stderr, "Error %d Line: %d\n", bt->err, line), exit(0);
2370                           len = 0;
2371                         }
2372                         else if( len < 255 )
2373                                 key[len++] = ch;
2374                 fprintf(stderr, "finished %s for keys, %d \n", args->infile, line);
2375                 break;
2376
2377         case 'f':
2378                 fprintf(stderr, "started finding keys for %s\n", args->infile);
2379                 if( in = fopen (args->infile, "rb") )
2380                   while( ch = getc(in), ch != EOF )
2381                         if( ch == '\n' )
2382                         {
2383                           line++;
2384                           if( args->num == 1 )
2385                                 sprintf((char *)key+len, "%.9d", 1000000000 - line), len += 9;
2386
2387                           else if( args->num )
2388                                 sprintf((char *)key+len, "%.9d", line + args->idx * args->num), len += 9;
2389
2390                           if( bt_findkey (bt, key, len) )
2391                                 found++;
2392                           else if( bt->err )
2393                                 fprintf(stderr, "Error %d Syserr %d Line: %d\n", bt->err, errno, line), exit(0);
2394                           len = 0;
2395                         }
2396                         else if( len < 255 )
2397                                 key[len++] = ch;
2398                 fprintf(stderr, "finished %s for %d keys, found %d\n", args->infile, line, found);
2399                 break;
2400
2401         case 's':
2402                 fprintf(stderr, "started scanning\n");
2403                 do {
2404                         if( set->pool = bt_pinpool (bt, page_no) )
2405                                 set->page = bt_page (bt, set->pool, page_no);
2406                         else
2407                                 break;
2408                         set->latch = bt_pinlatch (bt, page_no);
2409                         bt_lockpage (BtLockRead, set->latch);
2410                         next = bt_getid (set->page->right);
2411                         cnt += set->page->act;
2412
2413                         for( slot = 0; slot++ < set->page->cnt; )
2414                          if( next || slot < set->page->cnt )
2415                           if( !slotptr(set->page, slot)->dead ) {
2416                                 ptr = keyptr(set->page, slot);
2417                                 fwrite (ptr->key, ptr->len, 1, stdout);
2418                                 fputc ('\n', stdout);
2419                           }
2420
2421                         bt_unlockpage (BtLockRead, set->latch);
2422                         bt_unpinlatch (set->latch);
2423                         bt_unpinpool (set->pool);
2424                 } while( page_no = next );
2425
2426                 cnt--;  // remove stopper key
2427                 fprintf(stderr, " Total keys read %d\n", cnt);
2428                 break;
2429
2430         case 'c':
2431                 fprintf(stderr, "started counting\n");
2432                 next = bt->mgr->latchmgr->nlatchpage + LATCH_page;
2433                 page_no = LEAF_page;
2434
2435                 while( page_no < bt_getid(bt->mgr->latchmgr->alloc->right) ) {
2436                 uid off = page_no << bt->mgr->page_bits;
2437 #ifdef unix
2438                   pread (bt->mgr->idx, bt->frame, bt->mgr->page_size, off);
2439 #else
2440                 DWORD amt[1];
2441
2442                   SetFilePointer (bt->mgr->idx, (long)off, (long*)(&off)+1, FILE_BEGIN);
2443
2444                   if( !ReadFile(bt->mgr->idx, bt->frame, bt->mgr->page_size, amt, NULL))
2445                         return bt->err = BTERR_map;
2446
2447                   if( *amt <  bt->mgr->page_size )
2448                         return bt->err = BTERR_map;
2449 #endif
2450                         if( !bt->frame->free && !bt->frame->lvl )
2451                                 cnt += bt->frame->act;
2452                         if( page_no > LEAF_page )
2453                                 next = page_no + 1;
2454                         page_no = next;
2455                 }
2456                 
2457                 cnt--;  // remove stopper key
2458                 fprintf(stderr, " Total keys read %d\n", cnt);
2459                 break;
2460         }
2461
2462         bt_close (bt);
2463 #ifdef unix
2464         return NULL;
2465 #else
2466         return 0;
2467 #endif
2468 }
2469
2470 typedef struct timeval timer;
2471
2472 int main (int argc, char **argv)
2473 {
2474 int idx, cnt, len, slot, err;
2475 int segsize, bits = 16;
2476 double start, stop;
2477 #ifdef unix
2478 pthread_t *threads;
2479 #else
2480 HANDLE *threads;
2481 #endif
2482 ThreadArg *args;
2483 uint poolsize = 0;
2484 float elapsed;
2485 int num = 0;
2486 char key[1];
2487 BtMgr *mgr;
2488 BtKey ptr;
2489 BtDb *bt;
2490
2491         if( argc < 3 ) {
2492                 fprintf (stderr, "Usage: %s idx_file Read/Write/Scan/Delete/Find [page_bits mapped_segments seg_bits line_numbers src_file1 src_file2 ... ]\n", argv[0]);
2493                 fprintf (stderr, "  where page_bits is the page size in bits\n");
2494                 fprintf (stderr, "  mapped_segments is the number of mmap segments in buffer pool\n");
2495                 fprintf (stderr, "  seg_bits is the size of individual segments in buffer pool in pages in bits\n");
2496                 fprintf (stderr, "  line_numbers = 1 to append line numbers to keys\n");
2497                 fprintf (stderr, "  src_file1 thru src_filen are files of keys separated by newline\n");
2498                 exit(0);
2499         }
2500
2501         start = getCpuTime(0);
2502
2503         if( argc > 3 )
2504                 bits = atoi(argv[3]);
2505
2506         if( argc > 4 )
2507                 poolsize = atoi(argv[4]);
2508
2509         if( !poolsize )
2510                 fprintf (stderr, "Warning: no mapped_pool\n");
2511
2512         if( poolsize > 65535 )
2513                 fprintf (stderr, "Warning: mapped_pool > 65535 segments\n");
2514
2515         if( argc > 5 )
2516                 segsize = atoi(argv[5]);
2517         else
2518                 segsize = 4;    // 16 pages per mmap segment
2519
2520         if( argc > 6 )
2521                 num = atoi(argv[6]);
2522
2523         cnt = argc - 7;
2524 #ifdef unix
2525         threads = malloc (cnt * sizeof(pthread_t));
2526 #else
2527         threads = GlobalAlloc (GMEM_FIXED|GMEM_ZEROINIT, cnt * sizeof(HANDLE));
2528 #endif
2529         args = malloc (cnt * sizeof(ThreadArg));
2530
2531         mgr = bt_mgr ((argv[1]), BT_rw, bits, poolsize, segsize, poolsize / 8);
2532
2533         if( !mgr ) {
2534                 fprintf(stderr, "Index Open Error %s\n", argv[1]);
2535                 exit (1);
2536         }
2537
2538         //      fire off threads
2539
2540         for( idx = 0; idx < cnt; idx++ ) {
2541                 args[idx].infile = argv[idx + 7];
2542                 args[idx].type = argv[2][0];
2543                 args[idx].mgr = mgr;
2544                 args[idx].num = num;
2545                 args[idx].idx = idx;
2546 #ifdef unix
2547                 if( err = pthread_create (threads + idx, NULL, index_file, args + idx) )
2548                         fprintf(stderr, "Error creating thread %d\n", err);
2549 #else
2550                 threads[idx] = (HANDLE)_beginthreadex(NULL, 65536, index_file, args + idx, 0, NULL);
2551 #endif
2552         }
2553
2554         //      wait for termination
2555
2556 #ifdef unix
2557         for( idx = 0; idx < cnt; idx++ )
2558                 pthread_join (threads[idx], NULL);
2559 #else
2560         WaitForMultipleObjects (cnt, threads, TRUE, INFINITE);
2561
2562         for( idx = 0; idx < cnt; idx++ )
2563                 CloseHandle(threads[idx]);
2564
2565 #endif
2566         elapsed = getCpuTime(0) - start;
2567         fprintf(stderr, " real %dm%.3fs\n", (int)(elapsed/60), elapsed - (int)(elapsed/60)*60);
2568         elapsed = getCpuTime(1);
2569         fprintf(stderr, " user %dm%.3fs\n", (int)(elapsed/60), elapsed - (int)(elapsed/60)*60);
2570         elapsed = getCpuTime(2);
2571         fprintf(stderr, " sys  %dm%.3fs\n", (int)(elapsed/60), elapsed - (int)(elapsed/60)*60);
2572
2573         bt_mgrclose (mgr);
2574 }
2575
2576 #endif  //STANDALONE