X-Git-Url: https://pd.if.org/git/?p=btree;a=blobdiff_plain;f=threads2i.c;h=f1d0269b89776928b209bef823aa65c40c4f91ae;hp=25feba3b5ef345f7b97ff78cdf177cbe4835d56e;hb=HEAD;hpb=418c6dcf655db4593f628193de63d486156a72e9 diff --git a/threads2i.c b/threads2i.c index 25feba3..f1d0269 100644 --- a/threads2i.c +++ b/threads2i.c @@ -1,5 +1,6 @@ // btree version threads2i sched_yield version -// 24 JAN 2014 +// with reworked bt_deletekey code +// 17 FEB 2014 // author: karl malbrain, malbrain@cal.berkeley.edu @@ -49,6 +50,7 @@ REDISTRIBUTION OF THIS SOFTWARE. #include #include +#include typedef unsigned long long uid; @@ -85,24 +87,17 @@ typedef enum{ BtLockParent } BtLock; -// mode & definition for latch implementation - -enum { - Mutex = 1, - Write = 2, - Pending = 4, - Share = 8 -} LockMode; +// definition for latch implementation // exclusive is set for write access // share is count of read accessors // grant write lock when share == 0 -typedef struct { - volatile ushort mutex:1; - volatile ushort exclusive:1; - volatile ushort pending:1; - volatile ushort share:13; +volatile typedef struct { + unsigned char mutex[1]; + unsigned char exclusive:1; + unsigned char pending:1; + ushort share; } BtSpinLatch; // hash table entries @@ -117,7 +112,7 @@ typedef struct { typedef struct { BtSpinLatch readwr[1]; // read/write page lock BtSpinLatch access[1]; // Access Intent/Page delete - BtSpinLatch parent[1]; // adoption of foster children + BtSpinLatch parent[1]; // Posting of fence key in parent BtSpinLatch busy[1]; // slot is being moved between chains volatile ushort next; // next entry in hash table chain volatile ushort prev; // prev entry in hash table chain @@ -149,7 +144,7 @@ typedef struct { } BtSlot; // The key structure occupies space at the upper end of -// each page. It's a length byte followed by the value +// each page. It's a length byte followed by the key // bytes. typedef struct { @@ -161,23 +156,21 @@ typedef struct { // It is immediately followed // by the BtSlot array of keys. -typedef struct Page { - BtLatchSet latch[1]; // Set of three latches +typedef struct BtPage_ { uint cnt; // count of keys in page uint act; // count of active keys uint min; // next key offset - unsigned char bits; // page size in bits + unsigned char bits:7; // page size in bits + unsigned char free:1; // page is on free chain unsigned char lvl:6; // level of page unsigned char kill:1; // page is being deleted unsigned char dirty:1; // page has deleted keys unsigned char right[BtId]; // page number to right - BtSlot table[0]; // array of key slots } *BtPage; // The memory mapping pool table buffer manager entry typedef struct { - unsigned long long int lru; // number of times accessed uid basepage; // mapped base page number char *map; // mapped memory pointer ushort slot; // slot index in this array @@ -185,14 +178,25 @@ typedef struct { void *hashprev; // previous pool entry for the same hash idx void *hashnext; // next pool entry for the same hash idx #ifndef unix - HANDLE hmap; + HANDLE hmap; // Windows memory mapping handle #endif } BtPool; +#define CLOCK_bit 0x8000 // bit in pool->pin + +// The loadpage interface object + +typedef struct { + uid page_no; // current page number + BtPage page; // current page pointer + BtPool *pool; // current page pool + BtLatchSet *latch; // current page latch set +} BtPageSet; + // structure for latch manager on ALLOC_page typedef struct { - struct Page alloc[2]; // next & free page_nos in right ptr + struct BtPage_ alloc[2]; // next & free page_nos in right ptr BtSpinLatch lock[1]; // allocation area lite latch ushort latchdeployed; // highest number of latch entries deployed ushort nlatchpage; // number of latch pages at BT_latch @@ -210,7 +214,6 @@ typedef struct { uint seg_bits; // seg size in pages in bits uint mode; // read-write mode #ifdef unix - char *pooladvise; // bit maps for pool page advisements int idx; #else HANDLE idx; @@ -235,11 +238,7 @@ typedef struct { BtPage cursor; // cached frame for start/next (never mapped) BtPage frame; // spare frame for the page split (never mapped) BtPage zero; // page frame for zeroes at end of file - BtPage page; // current page - uid page_no; // current page number uid cursor_page; // current cursor page number - BtLatchSet *set; // current page latch set - BtPool *pool; // current page pool unsigned char *mem; // frame, cursor, page memory buffer int found; // last delete or insert was found int err; // last error @@ -258,7 +257,7 @@ typedef enum { // B-Tree functions extern void bt_close (BtDb *bt); extern BtDb *bt_open (BtMgr *mgr); -extern BTERR bt_insertkey (BtDb *bt, unsigned char *key, uint len, uint lvl, uid id, uint tod); +extern BTERR bt_insertkey (BtDb *bt, unsigned char *key, uint len, uint lvl, uid id, uint tod); extern BTERR bt_deletekey (BtDb *bt, unsigned char *key, uint len, uint lvl); extern uid bt_findkey (BtDb *bt, unsigned char *key, uint len); extern uint bt_startkey (BtDb *bt, unsigned char *key, uint len); @@ -293,7 +292,8 @@ extern uint bt_tod (BtDb *bt, uint slot); // A key consists of a length byte, two bytes of // index number (0 - 65534), and up to 253 bytes // of key value. Duplicate keys are discarded. -// Associated with each key is a 48 bit row-id. +// Associated with each key is a 48 bit row-id, +// or any other value desired. // The b-tree root is always located at page 1. // The first leaf page of level zero is always @@ -308,8 +308,8 @@ extern uint bt_tod (BtDb *bt, uint slot); // one with two keys. // Deleted keys are marked with a dead bit until -// page cleanup The fence key for a node is always -// present, even after deletion and cleanup. +// page cleanup. The fence key for a node is +// always present // Groups of pages called segments from the btree are optionally // cached with a memory mapped pool. A hash table is used to keep @@ -324,14 +324,15 @@ extern uint bt_tod (BtDb *bt, uint slot); // Page 0 is dedicated to lock for new page extensions, // and chains empty pages together for reuse. -// The ParentModification lock on a node is obtained to prevent resplitting -// or deleting a node before its fence is posted into its upper level. +// The ParentModification lock on a node is obtained to serialize posting +// or changing the fence key for a node. // Empty pages are chained together through the ALLOC page and reused. // Access macros to address slot and key values from the page +// Page slots use 1 based indexing. -#define slotptr(page, slot) (page->table + slot-1) +#define slotptr(page, slot) (((BtSlot *)(page+1)) + (slot-1)) #define keyptr(page, slot) ((BtKey)((unsigned char*)(page) + slotptr(page, slot)->off)) void bt_putid(unsigned char *dest, uid id) @@ -365,29 +366,26 @@ ushort prev; do { // obtain latch mutex #ifdef unix - if( __sync_fetch_and_or((ushort *)latch, Mutex) & Mutex ) + if( __sync_lock_test_and_set(latch->mutex, 1) ) continue; #else - if( prev = _InterlockedOr16((ushort *)latch, Mutex) & Mutex ) + if( _InterlockedExchange8(latch->mutex, 1) ) continue; #endif // see if exclusive request is granted or pending if( prev = !(latch->exclusive | latch->pending) ) -#ifdef unix - __sync_fetch_and_add((ushort *)latch, Share); -#else - _InterlockedExchangeAdd16 ((ushort *)latch, Share); -#endif + latch->share++; #ifdef unix - __sync_fetch_and_and ((ushort *)latch, ~Mutex); + *latch->mutex = 0; #else - _InterlockedAnd16((ushort *)latch, ~Mutex); + _InterlockedExchange8(latch->mutex, 0); #endif if( prev ) return; + #ifdef unix } while( sched_yield(), 1 ); #else @@ -399,31 +397,27 @@ ushort prev; void bt_spinwritelock(BtSpinLatch *latch) { +uint prev; + do { #ifdef unix - if( __sync_fetch_and_or((ushort *)latch, Mutex | Pending) & Mutex ) + if( __sync_lock_test_and_set(latch->mutex, 1) ) continue; #else - if( _InterlockedOr16((ushort *)latch, Mutex | Pending) & Mutex ) + if( _InterlockedExchange8(latch->mutex, 1) ) continue; #endif - if( !(latch->share | latch->exclusive) ) { + if( prev = !(latch->share | latch->exclusive) ) + latch->exclusive = 1, latch->pending = 0; + else + latch->pending = 1; #ifdef unix - __sync_fetch_and_or((ushort *)latch, Write); - __sync_fetch_and_and ((ushort *)latch, ~(Mutex | Pending)); + *latch->mutex = 0; #else - _InterlockedOr16((ushort *)latch, Write); - _InterlockedAnd16((ushort *)latch, ~(Mutex | Pending)); + _InterlockedExchange8(latch->mutex, 0); #endif + if( prev ) return; - } - -#ifdef unix - __sync_fetch_and_and ((ushort *)latch, ~Mutex); -#else - _InterlockedAnd16((ushort *)latch, ~Mutex); -#endif - #ifdef unix } while( sched_yield(), 1 ); #else @@ -438,30 +432,26 @@ void bt_spinwritelock(BtSpinLatch *latch) int bt_spinwritetry(BtSpinLatch *latch) { -ushort prev; +uint prev; #ifdef unix - if( prev = __sync_fetch_and_or((ushort *)latch, Mutex), prev & Mutex ) + if( __sync_lock_test_and_set(latch->mutex, 1) ) return 0; #else - if( prev = _InterlockedOr16((ushort *)latch, Mutex), prev & Mutex ) + if( _InterlockedExchange8(latch->mutex, 1) ) return 0; #endif // take write access if all bits are clear - if( !prev ) -#ifdef unix - __sync_fetch_and_or ((ushort *)latch, Write); -#else - _InterlockedOr16((ushort *)latch, Write); -#endif + if( prev = !(latch->exclusive | latch->share) ) + latch->exclusive = 1; #ifdef unix - __sync_fetch_and_and ((ushort *)latch, ~Mutex); + *latch->mutex = 0; #else - _InterlockedAnd16((ushort *)latch, ~Mutex); + _InterlockedExchange8(latch->mutex, 0); #endif - return !prev; + return prev; } // clear write mode @@ -469,9 +459,17 @@ ushort prev; void bt_spinreleasewrite(BtSpinLatch *latch) { #ifdef unix - __sync_fetch_and_and ((ushort *)latch, ~Write); + while( __sync_lock_test_and_set(latch->mutex, 1) ) + sched_yield(); #else - _InterlockedAnd16((ushort *)latch, ~Write); + while( _InterlockedExchange8(latch->mutex, 1) ) + SwitchToThread(); +#endif + latch->exclusive = 0; +#ifdef unix + *latch->mutex = 0; +#else + _InterlockedExchange8(latch->mutex, 0); #endif } @@ -480,9 +478,17 @@ void bt_spinreleasewrite(BtSpinLatch *latch) void bt_spinreleaseread(BtSpinLatch *latch) { #ifdef unix - __sync_fetch_and_add((ushort *)latch, -Share); + while( __sync_lock_test_and_set(latch->mutex, 1) ) + sched_yield(); +#else + while( _InterlockedExchange8(latch->mutex, 1) ) + SwitchToThread(); +#endif + latch->share--; +#ifdef unix + *latch->mutex = 0; #else - _InterlockedExchangeAdd16 ((ushort *)latch, -Share); + _InterlockedExchange8(latch->mutex, 0); #endif } @@ -691,15 +697,14 @@ uint slot; close (mgr->idx); free (mgr->pool); free (mgr->hash); - free (mgr->latch); - free (mgr->pooladvise); + free ((void *)mgr->latch); free (mgr); #else FlushFileBuffers(mgr->idx); CloseHandle(mgr->idx); GlobalFree (mgr->pool); GlobalFree (mgr->hash); - GlobalFree (mgr->latch); + GlobalFree ((void *)mgr->latch); GlobalFree (mgr); #endif } @@ -709,10 +714,10 @@ uint slot; void bt_close (BtDb *bt) { #ifdef unix - if ( bt->mem ) + if( bt->mem ) free (bt->mem); #else - if ( bt->mem) + if( bt->mem) VirtualFree (bt->mem, 0, MEM_RELEASE); #endif free (bt); @@ -825,7 +830,6 @@ SYSTEM_INFO sysinfo[1]; mgr->pool = calloc (poolmax, sizeof(BtPool)); mgr->hash = calloc (hashsize, sizeof(ushort)); mgr->latch = calloc (hashsize, sizeof(BtSpinLatch)); - mgr->pooladvise = calloc (poolmax, (mgr->poolmask + 8) / 8); #else mgr->pool = GlobalAlloc (GMEM_FIXED|GMEM_ZEROINIT, poolmax * sizeof(BtPool)); mgr->hash = GlobalAlloc (GMEM_FIXED|GMEM_ZEROINIT, hashsize * sizeof(ushort)); @@ -875,7 +879,7 @@ SYSTEM_INFO sysinfo[1]; slotptr(latchmgr->alloc, 1)->off = mgr->page_size - 3; bt_putid(slotptr(latchmgr->alloc, 1)->id, lvl ? MIN_lvl - lvl + 1 : 0); // next(lower) page number key = keyptr(latchmgr->alloc, 1); - key->len = 2; // create stopper key + key->len = 2; // create stopper key key->key[0] = 0xff; key->key[1] = 0xff; latchmgr->alloc->min = mgr->page_size - 3; @@ -1024,7 +1028,7 @@ uint slot; pool->hashprev = pool->hashnext = NULL; pool->basepage = page_no & ~bt->mgr->poolmask; - pool->lru = 1; + pool->pin = CLOCK_bit + 1; if( slot = bt->mgr->hash[idx] ) { node = bt->mgr->pool + slot; @@ -1035,32 +1039,6 @@ uint slot; bt->mgr->hash[idx] = pool->slot; } -// find best segment to evict from buffer pool - -BtPool *bt_findlru (BtDb *bt, uint hashslot) -{ -unsigned long long int target = ~0LL; -BtPool *pool = NULL, *node; - - if( !hashslot ) - return NULL; - - node = bt->mgr->pool + hashslot; - - // scan pool entries under hash table slot - - do { - if( node->pin ) - continue; - if( node->lru > target ) - continue; - target = node->lru; - pool = node; - } while( node = node->hashnext ); - - return pool; -} - // map new buffer pool segment to virtual memory BTERR bt_mapsegment(BtDb *bt, BtPool *pool, uid page_no) @@ -1075,8 +1053,6 @@ int flag; if( pool->map == MAP_FAILED ) return bt->err = BTERR_map; - // clear out madvise issued bits - memset (bt->mgr->pooladvise + pool->slot * ((bt->mgr->poolmask + 8) / 8), 0, (bt->mgr->poolmask + 8)/8); #else flag = ( bt->mgr->mode == BT_ro ? PAGE_READONLY : PAGE_READWRITE ); pool->hmap = CreateFileMapping(bt->mgr->idx, NULL, flag, (DWORD)(limit >> 32), (DWORD)limit, NULL); @@ -1099,17 +1075,6 @@ uint subpage = (uint)(page_no & bt->mgr->poolmask); // page within mapping BtPage page; page = (BtPage)(pool->map + (subpage << bt->mgr->page_bits)); -#ifdef unix - { - uint idx = subpage / 8; - uint bit = subpage % 8; - - if( ~((bt->mgr->pooladvise + pool->slot * ((bt->mgr->poolmask + 8)/8))[idx] >> bit) & 1 ) { - madvise (page, bt->mgr->page_size, MADV_WILLNEED); - (bt->mgr->pooladvise + pool->slot * ((bt->mgr->poolmask + 8)/8))[idx] |= 1 << bit; - } - } -#endif return page; } @@ -1129,42 +1094,25 @@ void bt_unpinpool (BtPool *pool) BtPool *bt_pinpool(BtDb *bt, uid page_no) { +uint slot, hashidx, idx, victim; BtPool *pool, *node, *next; -uint slot, idx, victim; // lock hash table chain - idx = (uint)(page_no >> bt->mgr->seg_bits) % bt->mgr->hashsize; - bt_spinreadlock (&bt->mgr->latch[idx]); + hashidx = (uint)(page_no >> bt->mgr->seg_bits) % bt->mgr->hashsize; + bt_spinwritelock (&bt->mgr->latch[hashidx]); // look up in hash table - if( pool = bt_findpool(bt, page_no, idx) ) { -#ifdef unix - __sync_fetch_and_add(&pool->pin, 1); -#else - _InterlockedIncrement16 (&pool->pin); -#endif - bt_spinreleaseread (&bt->mgr->latch[idx]); - pool->lru++; - return pool; - } - - // upgrade to write lock - - bt_spinreleaseread (&bt->mgr->latch[idx]); - bt_spinwritelock (&bt->mgr->latch[idx]); - - // try to find page in pool with write lock - - if( pool = bt_findpool(bt, page_no, idx) ) { + if( pool = bt_findpool(bt, page_no, hashidx) ) { #ifdef unix + __sync_fetch_and_or(&pool->pin, CLOCK_bit); __sync_fetch_and_add(&pool->pin, 1); #else + _InterlockedOr16 (&pool->pin, CLOCK_bit); _InterlockedIncrement16 (&pool->pin); #endif - bt_spinreleasewrite (&bt->mgr->latch[idx]); - pool->lru++; + bt_spinreleasewrite (&bt->mgr->latch[hashidx]); return pool; } @@ -1184,13 +1132,8 @@ uint slot, idx, victim; if( bt_mapsegment(bt, pool, page_no) ) return NULL; - bt_linkhash(bt, pool, page_no, idx); -#ifdef unix - __sync_fetch_and_add(&pool->pin, 1); -#else - _InterlockedIncrement16 (&pool->pin); -#endif - bt_spinreleasewrite (&bt->mgr->latch[idx]); + bt_linkhash(bt, pool, page_no, hashidx); + bt_spinreleasewrite (&bt->mgr->latch[hashidx]); return pool; } @@ -1209,20 +1152,30 @@ uint slot, idx, victim; #else victim = _InterlockedIncrement (&bt->mgr->evicted) - 1; #endif - victim %= bt->mgr->hashsize; + victim %= bt->mgr->poolmax; + pool = bt->mgr->pool + victim; + idx = (uint)(pool->basepage >> bt->mgr->seg_bits) % bt->mgr->hashsize; + + if( !victim ) + continue; // try to get write lock // skip entry if not obtained - if( !bt_spinwritetry (&bt->mgr->latch[victim]) ) + if( !bt_spinwritetry (&bt->mgr->latch[idx]) ) continue; - // if pool entry is empty - // or any pages are pinned - // skip this entry + // skip this entry if + // page is pinned + // or clock bit is set - if( !(pool = bt_findlru(bt, bt->mgr->hash[victim])) ) { - bt_spinreleasewrite (&bt->mgr->latch[victim]); + if( pool->pin ) { +#ifdef unix + __sync_fetch_and_and(&pool->pin, ~CLOCK_bit); +#else + _InterlockedAnd16 (&pool->pin, ~CLOCK_bit); +#endif + bt_spinreleasewrite (&bt->mgr->latch[idx]); continue; } @@ -1231,14 +1184,14 @@ uint slot, idx, victim; if( node = pool->hashprev ) node->hashnext = pool->hashnext; else if( node = pool->hashnext ) - bt->mgr->hash[victim] = node->slot; + bt->mgr->hash[idx] = node->slot; else - bt->mgr->hash[victim] = 0; + bt->mgr->hash[idx] = 0; if( node = pool->hashnext ) node->hashprev = pool->hashprev; - bt_spinreleasewrite (&bt->mgr->latch[victim]); + bt_spinreleasewrite (&bt->mgr->latch[idx]); // remove old file mapping #ifdef unix @@ -1256,13 +1209,8 @@ uint slot, idx, victim; if( bt_mapsegment(bt, pool, page_no) ) return NULL; - bt_linkhash(bt, pool, page_no, idx); -#ifdef unix - __sync_fetch_and_add(&pool->pin, 1); -#else - _InterlockedIncrement16 (&pool->pin); -#endif - bt_spinreleasewrite (&bt->mgr->latch[idx]); + bt_linkhash(bt, pool, page_no, hashidx); + bt_spinreleasewrite (&bt->mgr->latch[hashidx]); return pool; } } @@ -1317,10 +1265,8 @@ void bt_unlockpage(BtLock mode, BtLatchSet *set) uid bt_newpage(BtDb *bt, BtPage page) { -BtLatchSet *set; -BtPool *pool; +BtPageSet set[1]; uid new_page; -BtPage pmap; int reuse; // lock allocation page @@ -1331,12 +1277,13 @@ int reuse; // else allocate empty page if( new_page = bt_getid(bt->mgr->latchmgr->alloc[1].right) ) { - if( pool = bt_pinpool (bt, new_page) ) - pmap = bt_page (bt, pool, new_page); + if( set->pool = bt_pinpool (bt, new_page) ) + set->page = bt_page (bt, set->pool, new_page); else return 0; - bt_putid(bt->mgr->latchmgr->alloc[1].right, bt_getid(pmap->right)); - bt_unpinpool (pool); + + bt_putid(bt->mgr->latchmgr->alloc[1].right, bt_getid(set->page->right)); + bt_unpinpool (set->pool); reuse = 1; } else { new_page = bt_getid(bt->mgr->latchmgr->alloc->right); @@ -1344,29 +1291,28 @@ int reuse; reuse = 0; } #ifdef unix - if ( pwrite(bt->mgr->idx, page, bt->mgr->page_size, new_page << bt->mgr->page_bits) < bt->mgr->page_size ) + if( pwrite(bt->mgr->idx, page, bt->mgr->page_size, new_page << bt->mgr->page_bits) < bt->mgr->page_size ) return bt->err = BTERR_wrt, 0; // if writing first page of pool block, zero last page in the block - if ( !reuse && bt->mgr->poolmask > 0 && (new_page & bt->mgr->poolmask) == 0 ) + if( !reuse && bt->mgr->poolmask > 0 && (new_page & bt->mgr->poolmask) == 0 ) { // use zero buffer to write zeros - memset(bt->zero, 0, bt->mgr->page_size); - if ( pwrite(bt->mgr->idx,bt->zero, bt->mgr->page_size, (new_page | bt->mgr->poolmask) << bt->mgr->page_bits) < bt->mgr->page_size ) + if( pwrite(bt->mgr->idx,bt->zero, bt->mgr->page_size, (new_page | bt->mgr->poolmask) << bt->mgr->page_bits) < bt->mgr->page_size ) return bt->err = BTERR_wrt, 0; } #else // bring new page into pool and copy page. // this will extend the file into the new pages. - if( pool = bt_pinpool (bt, new_page) ) - pmap = bt_page (bt, pool, new_page); + if( set->pool = bt_pinpool (bt, new_page) ) + set->page = bt_page (bt, set->pool, new_page); else return 0; - memcpy(pmap, page, bt->mgr->page_size); - bt_unpinpool (pool); + memcpy(set->page, page, bt->mgr->page_size); + bt_unpinpool (set->pool); #endif // unlock allocation latch and return new page no @@ -1376,24 +1322,27 @@ int reuse; // find slot in page for given key at a given level -int bt_findslot (BtDb *bt, unsigned char *key, uint len) +int bt_findslot (BtPageSet *set, unsigned char *key, uint len) { -uint diff, higher = bt->page->cnt, low = 1, slot; +uint diff, higher = set->page->cnt, low = 1, slot; uint good = 0; - // make stopper key an infinite fence value + // make stopper key an infinite fence value - if( bt_getid (bt->page->right) ) + if( bt_getid (set->page->right) ) higher++; else good++; - // low is the next candidate, higher is already - // tested as .ge. the given key, loop ends when they meet + // low is the lowest candidate. + // loop ends when they meet + + // higher is already + // tested as .ge. the passed key. while( diff = higher - low ) { slot = low + ( diff >> 1 ); - if( keycmp (keyptr(bt->page, slot), key, len) < 0 ) + if( keycmp (keyptr(set->page, slot), key, len) < 0 ) low = slot + 1; else higher = slot, good++; @@ -1401,108 +1350,105 @@ uint good = 0; // return zero if key is on right link page - return good ? higher : 0; + return good ? higher : 0; } // find and load page at given level for given key // leave page rd or wr locked as requested -int bt_loadpage (BtDb *bt, unsigned char *key, uint len, uint lvl, uint lock) +int bt_loadpage (BtDb *bt, BtPageSet *set, unsigned char *key, uint len, uint lvl, BtLock lock) { uid page_no = ROOT_page, prevpage = 0; -BtLatchSet *set, *prevset; uint drill = 0xff, slot; +BtLatchSet *prevlatch; uint mode, prevmode; BtPool *prevpool; // start at root of btree and drill down - bt->set = NULL; - do { // determine lock mode of drill level - mode = (lock == BtLockWrite) && (drill == lvl) ? BtLockWrite : BtLockRead; + mode = (drill == lvl) ? lock : BtLockRead; - bt->set = bt_pinlatch (bt, page_no); - bt->page_no = page_no; + set->latch = bt_pinlatch (bt, page_no); + set->page_no = page_no; // pin page contents - if( bt->pool = bt_pinpool (bt, page_no) ) - bt->page = bt_page (bt, bt->pool, page_no); + if( set->pool = bt_pinpool (bt, page_no) ) + set->page = bt_page (bt, set->pool, page_no); else return 0; // obtain access lock using lock chaining with Access mode if( page_no > ROOT_page ) - bt_lockpage(BtLockAccess, bt->set); + bt_lockpage(BtLockAccess, set->latch); // release & unpin parent page if( prevpage ) { - bt_unlockpage(prevmode, prevset); - bt_unpinlatch (prevset); + bt_unlockpage(prevmode, prevlatch); + bt_unpinlatch (prevlatch); bt_unpinpool (prevpool); prevpage = 0; } // obtain read lock using lock chaining - bt_lockpage(mode, bt->set); + bt_lockpage(mode, set->latch); + + if( set->page->free ) + return bt->err = BTERR_struct, 0; if( page_no > ROOT_page ) - bt_unlockpage(BtLockAccess, bt->set); + bt_unlockpage(BtLockAccess, set->latch); // re-read and re-lock root after determining actual level of root - if( bt->page->lvl != drill) { - if ( bt->page_no != ROOT_page ) + if( set->page->lvl != drill) { + if( set->page_no != ROOT_page ) return bt->err = BTERR_struct, 0; - drill = bt->page->lvl; + drill = set->page->lvl; - if( lock == BtLockWrite && drill == lvl ) { - bt_unlockpage(mode, bt->set); - bt_unpinlatch (bt->set); - bt_unpinpool (bt->pool); + if( lock != BtLockRead && drill == lvl ) { + bt_unlockpage(mode, set->latch); + bt_unpinlatch (set->latch); + bt_unpinpool (set->pool); continue; } } + prevpage = set->page_no; + prevlatch = set->latch; + prevpool = set->pool; + prevmode = mode; + // find key on page at this level // and descend to requested level - if( !bt->page->kill && (slot = bt_findslot (bt, key, len)) ) { + if( !set->page->kill ) + if( slot = bt_findslot (set, key, len) ) { if( drill == lvl ) return slot; - while( slotptr(bt->page, slot)->dead ) - if( slot++ < bt->page->cnt ) + while( slotptr(set->page, slot)->dead ) + if( slot++ < set->page->cnt ) continue; - else { - page_no = bt_getid(bt->page->right); + else goto slideright; - } - page_no = bt_getid(slotptr(bt->page, slot)->id); + page_no = bt_getid(slotptr(set->page, slot)->id); drill--; - } + continue; + } // or slide right into next page - // (slide left from deleted page) - - else - page_no = bt_getid(bt->page->right); - - // continue down / right using overlapping locks - // to protect pages being killed or split. slideright: - prevpage = bt->page_no; - prevpool = bt->pool; - prevset = bt->set; - prevmode = mode; + page_no = bt_getid(set->page->right); + } while( page_no ); // return error on end of right chain @@ -1511,118 +1457,235 @@ slideright: return 0; // return error } +// return page to free list +// page must be delete & write locked + +void bt_freepage (BtDb *bt, BtPageSet *set) +{ + // lock allocation page + + bt_spinwritelock (bt->mgr->latchmgr->lock); + + // store chain in second right + bt_putid(set->page->right, bt_getid(bt->mgr->latchmgr->alloc[1].right)); + bt_putid(bt->mgr->latchmgr->alloc[1].right, set->page_no); + set->page->free = 1; + + // unlock released page + + bt_unlockpage (BtLockDelete, set->latch); + bt_unlockpage (BtLockWrite, set->latch); + bt_unpinlatch (set->latch); + bt_unpinpool (set->pool); + + // unlock allocation page + + bt_spinreleasewrite (bt->mgr->latchmgr->lock); +} + +// a fence key was deleted from a page +// push new fence value upwards + +BTERR bt_fixfence (BtDb *bt, BtPageSet *set, uint lvl) +{ +unsigned char leftkey[256], rightkey[256]; +uid page_no; +BtKey ptr; + + // remove the old fence value + + ptr = keyptr(set->page, set->page->cnt); + memcpy (rightkey, ptr, ptr->len + 1); + + memset (slotptr(set->page, set->page->cnt--), 0, sizeof(BtSlot)); + set->page->dirty = 1; + + ptr = keyptr(set->page, set->page->cnt); + memcpy (leftkey, ptr, ptr->len + 1); + page_no = set->page_no; + + bt_lockpage (BtLockParent, set->latch); + bt_unlockpage (BtLockWrite, set->latch); + + // insert new (now smaller) fence key + + if( bt_insertkey (bt, leftkey+1, *leftkey, lvl+1, page_no, time(NULL)) ) + return bt->err; + + // now delete old fence key + + if( bt_deletekey (bt, rightkey+1, *rightkey, lvl+1) ) + return bt->err; + + bt_unlockpage (BtLockParent, set->latch); + bt_unpinlatch(set->latch); + bt_unpinpool (set->pool); + return 0; +} + +// root has a single child +// collapse a level from the tree + +BTERR bt_collapseroot (BtDb *bt, BtPageSet *root) +{ +BtPageSet child[1]; +uint idx; + + // find the child entry and promote as new root contents + + do { + for( idx = 0; idx++ < root->page->cnt; ) + if( !slotptr(root->page, idx)->dead ) + break; + + child->page_no = bt_getid (slotptr(root->page, idx)->id); + + child->latch = bt_pinlatch (bt, child->page_no); + bt_lockpage (BtLockDelete, child->latch); + bt_lockpage (BtLockWrite, child->latch); + + if( child->pool = bt_pinpool (bt, child->page_no) ) + child->page = bt_page (bt, child->pool, child->page_no); + else + return bt->err; + + memcpy (root->page, child->page, bt->mgr->page_size); + bt_freepage (bt, child); + + } while( root->page->lvl > 1 && root->page->act == 1 ); + + bt_unlockpage (BtLockWrite, root->latch); + bt_unpinlatch (root->latch); + bt_unpinpool (root->pool); + return 0; +} + // find and delete key on page by marking delete flag bit -// when page becomes empty, delete it +// if page becomes empty, delete it from the btree BTERR bt_deletekey (BtDb *bt, unsigned char *key, uint len, uint lvl) { -unsigned char lowerkey[256], higherkey[256]; -BtLatchSet *rset, *set; -BtPool *pool, *rpool; -uid page_no, right; -uint slot, tod; -BtPage rpage; +unsigned char lowerfence[256], higherfence[256]; +uint slot, idx, dirty = 0, fence, found; +BtPageSet set[1], right[1]; BtKey ptr; - if( slot = bt_loadpage (bt, key, len, lvl, BtLockWrite) ) - ptr = keyptr(bt->page, slot); + if( slot = bt_loadpage (bt, set, key, len, lvl, BtLockWrite) ) + ptr = keyptr(set->page, slot); else return bt->err; + // are we deleting a fence slot? + + fence = slot == set->page->cnt; + // if key is found delete it, otherwise ignore request - if( bt->found = !keycmp (ptr, key, len) ) - if( bt->found = slotptr(bt->page, slot)->dead == 0 ) { - slotptr(bt->page,slot)->dead = 1; - if( slot < bt->page->cnt ) - bt->page->dirty = 1; - bt->page->act--; - } + if( found = !keycmp (ptr, key, len) ) + if( found = slotptr(set->page, slot)->dead == 0 ) { + dirty = slotptr(set->page, slot)->dead = 1; + set->page->dirty = 1; + set->page->act--; - // return if page is not empty, or it has no right sibling + // collapse empty slots - right = bt_getid(bt->page->right); - page_no = bt->page_no; - pool = bt->pool; - set = bt->set; + while( idx = set->page->cnt - 1 ) + if( slotptr(set->page, idx)->dead ) { + *slotptr(set->page, idx) = *slotptr(set->page, idx + 1); + memset (slotptr(set->page, set->page->cnt--), 0, sizeof(BtSlot)); + } else + break; + } + + // did we delete a fence key in an upper level? - if( !right || bt->page->act ) { - bt_unlockpage(BtLockWrite, set); - bt_unpinlatch (set); - bt_unpinpool (pool); + if( dirty && lvl && set->page->act && fence ) + if( bt_fixfence (bt, set, lvl) ) return bt->err; - } + else + return bt->found = found, 0; - // obtain Parent lock over write lock + // is this a collapsed root? - bt_lockpage(BtLockParent, set); + if( lvl > 1 && set->page_no == ROOT_page && set->page->act == 1 ) + if( bt_collapseroot (bt, set) ) + return bt->err; + else + return bt->found = found, 0; - // keep copy of key to delete + // return if page is not empty - ptr = keyptr(bt->page, bt->page->cnt); - memcpy(lowerkey, ptr, ptr->len + 1); + if( set->page->act ) { + bt_unlockpage(BtLockWrite, set->latch); + bt_unpinlatch (set->latch); + bt_unpinpool (set->pool); + return bt->found = found, 0; + } - // lock and map right page + // cache copy of fence key + // to post in parent - if( rpool = bt_pinpool (bt, right) ) - rpage = bt_page (bt, rpool, right); - else - return bt->err; + ptr = keyptr(set->page, set->page->cnt); + memcpy (lowerfence, ptr, ptr->len + 1); - rset = bt_pinlatch (bt, right); - bt_lockpage(BtLockWrite, rset); + // obtain lock on right page - // pull contents of next page into current empty page + right->page_no = bt_getid(set->page->right); + right->latch = bt_pinlatch (bt, right->page_no); + bt_lockpage (BtLockWrite, right->latch); - memcpy (bt->page, rpage, bt->mgr->page_size); + // pin page contents - // keep copy of key to update + if( right->pool = bt_pinpool (bt, right->page_no) ) + right->page = bt_page (bt, right->pool, right->page_no); + else + return 0; - ptr = keyptr(rpage, rpage->cnt); - memcpy(higherkey, ptr, ptr->len + 1); + if( right->page->kill ) + return bt->err = BTERR_struct; - // Mark right page as deleted and point it to left page - // until we can post updates at higher level. + // pull contents of right peer into our empty page - bt_putid(rpage->right, page_no); - rpage->kill = 1; - rpage->cnt = 0; + memcpy (set->page, right->page, bt->mgr->page_size); - bt_unlockpage(BtLockWrite, rset); - bt_unlockpage(BtLockWrite, set); + // cache copy of key to update - // delete old lower key to consolidated node + ptr = keyptr(right->page, right->page->cnt); + memcpy (higherfence, ptr, ptr->len + 1); - if( bt_deletekey (bt, lowerkey + 1, *lowerkey, lvl + 1) ) - return bt->err; + // mark right page deleted and point it to left page + // until we can post parent updates - // redirect higher key directly to consolidated node + bt_putid (right->page->right, set->page_no); + right->page->kill = 1; - tod = (uint)time(NULL); + bt_lockpage (BtLockParent, right->latch); + bt_unlockpage (BtLockWrite, right->latch); - if( bt_insertkey (bt, higherkey+1, *higherkey, lvl + 1, page_no, tod) ) - return bt->err; + bt_lockpage (BtLockParent, set->latch); + bt_unlockpage (BtLockWrite, set->latch); - // add killed right block to free chain - // lock latch mgr + // redirect higher key directly to our new node contents - bt_spinwritelock(bt->mgr->latchmgr->lock); + if( bt_insertkey (bt, higherfence+1, *higherfence, lvl+1, set->page_no, time(NULL)) ) + return bt->err; - // store free chain in allocation page second right - bt_putid(rpage->right, bt_getid(bt->mgr->latchmgr->alloc[1].right)); - bt_putid(bt->mgr->latchmgr->alloc[1].right, right); + // delete old lower key to our node - // unlock latch mgr and unpin right page + if( bt_deletekey (bt, lowerfence+1, *lowerfence, lvl+1) ) + return bt->err; - bt_spinreleasewrite(bt->mgr->latchmgr->lock); - bt_unpinlatch (rset); - bt_unpinpool (rpool); + // obtain delete and write locks to right node - // remove ParentModify lock + bt_unlockpage (BtLockParent, right->latch); + bt_lockpage (BtLockDelete, right->latch); + bt_lockpage (BtLockWrite, right->latch); + bt_freepage (bt, right); - bt_unlockpage(BtLockParent, set); - bt_unpinlatch (set); - bt_unpinpool (pool); + bt_unlockpage (BtLockParent, set->latch); + bt_unpinlatch (set->latch); + bt_unpinpool (set->pool); + bt->found = found; return 0; } @@ -1630,41 +1693,40 @@ BtKey ptr; uid bt_findkey (BtDb *bt, unsigned char *key, uint len) { +BtPageSet set[1]; uint slot; +uid id = 0; BtKey ptr; -uid id; - if( slot = bt_loadpage (bt, key, len, 0, BtLockRead) ) - ptr = keyptr(bt->page, slot); + if( slot = bt_loadpage (bt, set, key, len, 0, BtLockRead) ) + ptr = keyptr(set->page, slot); else return 0; // if key exists, return row-id // otherwise return 0 - if( ptr->len == len && !memcmp (ptr->key, key, len) ) - id = bt_getid(slotptr(bt->page,slot)->id); - else - id = 0; + if( slot <= set->page->cnt ) + if( !keycmp (ptr, key, len) ) + id = bt_getid(slotptr(set->page,slot)->id); - bt_unlockpage (BtLockRead, bt->set); - bt_unpinlatch (bt->set); - bt_unpinpool (bt->pool); + bt_unlockpage (BtLockRead, set->latch); + bt_unpinlatch (set->latch); + bt_unpinpool (set->pool); return id; } // check page for space available, // clean if necessary and return -// =0 - page needs splitting -// >0 - go ahead at returned slot +// 0 - page needs splitting +// >0 new slot value -uint bt_cleanpage(BtDb *bt, uint amt, uint slot) +uint bt_cleanpage(BtDb *bt, BtPage page, uint amt, uint slot) { uint nxt = bt->mgr->page_size; -BtPage page = bt->page; uint cnt = 0, idx = 0; uint max = page->cnt; -uint newslot; +uint newslot = max; BtKey key; if( page->min >= (max+1) * sizeof(BtSlot) + sizeof(*page) + amt + 1 ) @@ -1683,29 +1745,35 @@ BtKey key; page->dirty = 0; page->act = 0; - // always leave fence key in list + // try cleaning up page first + // by removing deleted keys while( cnt++ < max ) { if( cnt == slot ) newslot = idx + 1; - else if( cnt < max && slotptr(bt->frame,cnt)->dead ) + if( cnt < max && slotptr(bt->frame,cnt)->dead ) continue; - // copy key + // copy the key across + key = keyptr(bt->frame, cnt); nxt -= key->len + 1; memcpy ((unsigned char *)page + nxt, key, key->len + 1); // copy slot + memcpy(slotptr(page, ++idx)->id, slotptr(bt->frame, cnt)->id, BtId); if( !(slotptr(page, idx)->dead = slotptr(bt->frame, cnt)->dead) ) page->act++; slotptr(page, idx)->tod = slotptr(bt->frame, cnt)->tod; slotptr(page, idx)->off = nxt; } + page->min = nxt; page->cnt = idx; + // see if page has enough space now, or does it need splitting? + if( page->min >= (idx+1) * sizeof(BtSlot) + sizeof(*page) + amt + 1 ) return newslot; @@ -1714,93 +1782,87 @@ BtKey key; // split the root and raise the height of the btree -BTERR bt_splitroot(BtDb *bt, unsigned char *newkey, unsigned char *oldkey, uid page_no2) +BTERR bt_splitroot(BtDb *bt, BtPageSet *root, unsigned char *leftkey, uid page_no2) { uint nxt = bt->mgr->page_size; -BtPage root = bt->page; -uid new_page; +uid left; // Obtain an empty page to use, and copy the current - // root contents into it which is the lower half of - // the old root. + // root contents into it, e.g. lower keys - if( !(new_page = bt_newpage(bt, root)) ) + if( !(left = bt_newpage(bt, root->page)) ) return bt->err; // preserve the page info at the bottom - // and set rest to zero + // of higher keys and set rest to zero - memset(root+1, 0, bt->mgr->page_size - sizeof(*root)); + memset(root->page+1, 0, bt->mgr->page_size - sizeof(*root->page)); - // insert first key on newroot page + // insert lower keys page fence key on newroot page as first key - nxt -= *newkey + 1; - memcpy ((unsigned char *)root + nxt, newkey, *newkey + 1); - bt_putid(slotptr(root, 1)->id, new_page); - slotptr(root, 1)->off = nxt; + nxt -= *leftkey + 1; + memcpy ((unsigned char *)root->page + nxt, leftkey, *leftkey + 1); + bt_putid(slotptr(root->page, 1)->id, left); + slotptr(root->page, 1)->off = nxt; - // insert second key on newroot page + // insert stopper key on newroot page // and increase the root height - nxt -= *oldkey + 1; - memcpy ((unsigned char *)root + nxt, oldkey, *oldkey + 1); - bt_putid(slotptr(root, 2)->id, page_no2); - slotptr(root, 2)->off = nxt; + nxt -= 3; + ((unsigned char *)root->page)[nxt] = 2; + ((unsigned char *)root->page)[nxt+1] = 0xff; + ((unsigned char *)root->page)[nxt+2] = 0xff; + bt_putid(slotptr(root->page, 2)->id, page_no2); + slotptr(root->page, 2)->off = nxt; - bt_putid(root->right, 0); - root->min = nxt; // reset lowest used offset and key count - root->cnt = 2; - root->act = 2; - root->lvl++; + bt_putid(root->page->right, 0); + root->page->min = nxt; // reset lowest used offset and key count + root->page->cnt = 2; + root->page->act = 2; + root->page->lvl++; - // release and unpin root (bt->page) + // release and unpin root - bt_unlockpage(BtLockWrite, bt->set); - bt_unpinlatch (bt->set); - bt_unpinpool (bt->pool); + bt_unlockpage(BtLockWrite, root->latch); + bt_unpinlatch (root->latch); + bt_unpinpool (root->pool); return 0; } // split already locked full node // return unlocked. -BTERR bt_splitpage (BtDb *bt) +BTERR bt_splitpage (BtDb *bt, BtPageSet *set) { uint cnt = 0, idx = 0, max, nxt = bt->mgr->page_size; -unsigned char oldkey[256], lowerkey[256]; -uid page_no = bt->page_no, right; -BtLatchSet *nset, *set = bt->set; -BtPool *pool = bt->pool; -BtPage page = bt->page; -uint lvl = page->lvl; -uid new_page; +unsigned char fencekey[256], rightkey[256]; +uint lvl = set->page->lvl; +BtPageSet right[1]; +uint prev; BtKey key; -uint tod; // split higher half of keys to bt->frame - // the last key (fence key) might be dead - - tod = (uint)time(NULL); memset (bt->frame, 0, bt->mgr->page_size); - max = (int)page->cnt; + max = set->page->cnt; cnt = max / 2; idx = 0; while( cnt++ < max ) { - key = keyptr(page, cnt); + key = keyptr(set->page, cnt); nxt -= key->len + 1; memcpy ((unsigned char *)bt->frame + nxt, key, key->len + 1); - memcpy(slotptr(bt->frame,++idx)->id, slotptr(page,cnt)->id, BtId); - if( !(slotptr(bt->frame, idx)->dead = slotptr(page, cnt)->dead) ) + + memcpy(slotptr(bt->frame,++idx)->id, slotptr(set->page,cnt)->id, BtId); + if( !(slotptr(bt->frame, idx)->dead = slotptr(set->page, cnt)->dead) ) bt->frame->act++; - slotptr(bt->frame, idx)->tod = slotptr(page, cnt)->tod; + slotptr(bt->frame, idx)->tod = slotptr(set->page, cnt)->tod; slotptr(bt->frame, idx)->off = nxt; } // remember existing fence key for new page to the right - memcpy (oldkey, key, key->len + 1); + memcpy (rightkey, key, key->len + 1); bt->frame->bits = bt->mgr->page_bits; bt->frame->min = nxt; @@ -1809,170 +1871,166 @@ uint tod; // link right node - if( page_no > ROOT_page ) { - right = bt_getid (page->right); - bt_putid(bt->frame->right, right); - } + if( set->page_no > ROOT_page ) + memcpy (bt->frame->right, set->page->right, BtId); - // get new free page and write frame to it. + // get new free page and write higher keys to it. - if( !(new_page = bt_newpage(bt, bt->frame)) ) + if( !(right->page_no = bt_newpage(bt, bt->frame)) ) return bt->err; // update lower keys to continue in old page - memcpy (bt->frame, page, bt->mgr->page_size); - memset (page+1, 0, bt->mgr->page_size - sizeof(*page)); + memcpy (bt->frame, set->page, bt->mgr->page_size); + memset (set->page+1, 0, bt->mgr->page_size - sizeof(*set->page)); nxt = bt->mgr->page_size; - page->act = 0; + set->page->dirty = 0; + set->page->act = 0; cnt = 0; idx = 0; // assemble page of smaller keys - // (they're all active keys) while( cnt++ < max / 2 ) { key = keyptr(bt->frame, cnt); nxt -= key->len + 1; - memcpy ((unsigned char *)page + nxt, key, key->len + 1); - memcpy(slotptr(page,++idx)->id, slotptr(bt->frame,cnt)->id, BtId); - slotptr(page, idx)->tod = slotptr(bt->frame, cnt)->tod; - slotptr(page, idx)->off = nxt; - page->act++; + memcpy ((unsigned char *)set->page + nxt, key, key->len + 1); + memcpy(slotptr(set->page,++idx)->id, slotptr(bt->frame,cnt)->id, BtId); + slotptr(set->page, idx)->tod = slotptr(bt->frame, cnt)->tod; + slotptr(set->page, idx)->off = nxt; + set->page->act++; } - // remember fence key for old page + // remember fence key for smaller page - memcpy(lowerkey, key, key->len + 1); - bt_putid(page->right, new_page); - page->min = nxt; - page->cnt = idx; - - // if current page is the root page, split it + memcpy(fencekey, key, key->len + 1); - if( page_no == ROOT_page ) - return bt_splitroot (bt, lowerkey, oldkey, new_page); + bt_putid(set->page->right, right->page_no); + set->page->min = nxt; + set->page->cnt = idx; - // obtain Parent/Write locks - // for left and right node pages + // if current page is the root page, split it - nset = bt_pinlatch (bt, new_page); + if( set->page_no == ROOT_page ) + return bt_splitroot (bt, set, fencekey, right->page_no); - bt_lockpage (BtLockParent, nset); - bt_lockpage (BtLockParent, set); + // insert new fences in their parent pages - // release wr lock on left page - // (keep the SMO in sequence) + right->latch = bt_pinlatch (bt, right->page_no); + bt_lockpage (BtLockParent, right->latch); - bt_unlockpage (BtLockWrite, set); + bt_lockpage (BtLockParent, set->latch); + bt_unlockpage (BtLockWrite, set->latch); - // insert new fence for reformulated left block + // insert new fence for reformulated left block of smaller keys - if( bt_insertkey (bt, lowerkey+1, *lowerkey, lvl + 1, page_no, tod) ) + if( bt_insertkey (bt, fencekey+1, *fencekey, lvl+1, set->page_no, time(NULL)) ) return bt->err; - // fix old fence for newly allocated right block page + // switch fence for right block of larger keys to new right page - if( bt_insertkey (bt, oldkey+1, *oldkey, lvl + 1, new_page, tod) ) + if( bt_insertkey (bt, rightkey+1, *rightkey, lvl+1, right->page_no, time(NULL)) ) return bt->err; - // release Parent locks + bt_unlockpage (BtLockParent, set->latch); + bt_unpinlatch (set->latch); + bt_unpinpool (set->pool); - bt_unlockpage (BtLockParent, nset); - bt_unlockpage (BtLockParent, set); - bt_unpinlatch (nset); - bt_unpinlatch (set); - bt_unpinpool (pool); + bt_unlockpage (BtLockParent, right->latch); + bt_unpinlatch (right->latch); return 0; } - -// Insert new key into the btree at requested level. -// Level zero pages are leaf pages. Page is unlocked at exit. +// Insert new key into the btree at given level. BTERR bt_insertkey (BtDb *bt, unsigned char *key, uint len, uint lvl, uid id, uint tod) { +BtPageSet set[1]; uint slot, idx; -BtPage page; BtKey ptr; - while( 1 ) { - if( slot = bt_loadpage (bt, key, len, lvl, BtLockWrite) ) - ptr = keyptr(bt->page, slot); - else - { - if ( !bt->err ) - bt->err = BTERR_ovflw; - return bt->err; - } - - // if key already exists, update id and return - - page = bt->page; + while( 1 ) { + if( slot = bt_loadpage (bt, set, key, len, lvl, BtLockWrite) ) + ptr = keyptr(set->page, slot); + else + { + if( !bt->err ) + bt->err = BTERR_ovflw; + return bt->err; + } - if( bt->found = !keycmp (ptr, key, len) ) { - slotptr(page, slot)->dead = 0; - slotptr(page, slot)->tod = tod; - bt_putid(slotptr(page,slot)->id, id); - bt_unlockpage(BtLockWrite, bt->set); - bt_unpinlatch(bt->set); - bt_unpinpool (bt->pool); - return bt->err; - } + // if key already exists, update id and return + + if( !keycmp (ptr, key, len) ) { + if( slotptr(set->page, slot)->dead ) + set->page->act++; + slotptr(set->page, slot)->dead = 0; + slotptr(set->page, slot)->tod = tod; + bt_putid(slotptr(set->page,slot)->id, id); + bt_unlockpage(BtLockWrite, set->latch); + bt_unpinlatch (set->latch); + bt_unpinpool (set->pool); + return 0; + } - // check if page has enough space + // check if page has enough space - if( slot = bt_cleanpage (bt, len, slot) ) - break; + if( slot = bt_cleanpage (bt, set->page, len, slot) ) + break; - if( bt_splitpage (bt) ) - return bt->err; - } + if( bt_splitpage (bt, set) ) + return bt->err; + } - // calculate next available slot and copy key into page + // calculate next available slot and copy key into page - page->min -= len + 1; // reset lowest used offset - ((unsigned char *)page)[page->min] = len; - memcpy ((unsigned char *)page + page->min +1, key, len ); + set->page->min -= len + 1; // reset lowest used offset + ((unsigned char *)set->page)[set->page->min] = len; + memcpy ((unsigned char *)set->page + set->page->min +1, key, len ); - for( idx = slot; idx < page->cnt; idx++ ) - if( slotptr(page, idx)->dead ) + for( idx = slot; idx < set->page->cnt; idx++ ) + if( slotptr(set->page, idx)->dead ) break; - // now insert key into array before slot - // preserving the fence slot + // now insert key into array before slot - if( idx == page->cnt ) - idx++, page->cnt++; + if( idx == set->page->cnt ) + idx++, set->page->cnt++; - page->act++; + set->page->act++; - while( idx > slot ) - *slotptr(page, idx) = *slotptr(page, idx -1), idx--; + while( idx > slot ) + *slotptr(set->page, idx) = *slotptr(set->page, idx -1), idx--; - bt_putid(slotptr(page,slot)->id, id); - slotptr(page, slot)->off = page->min; - slotptr(page, slot)->tod = tod; - slotptr(page, slot)->dead = 0; + bt_putid(slotptr(set->page,slot)->id, id); + slotptr(set->page, slot)->off = set->page->min; + slotptr(set->page, slot)->tod = tod; + slotptr(set->page, slot)->dead = 0; - bt_unlockpage (BtLockWrite, bt->set); - bt_unpinlatch (bt->set); - bt_unpinpool (bt->pool); - return 0; + bt_unlockpage (BtLockWrite, set->latch); + bt_unpinlatch (set->latch); + bt_unpinpool (set->pool); + return 0; } // cache page of keys into cursor and return starting slot for given key uint bt_startkey (BtDb *bt, unsigned char *key, uint len) { +BtPageSet set[1]; uint slot; // cache page for retrieval - if( slot = bt_loadpage (bt, key, len, 0, BtLockRead) ) - memcpy (bt->cursor, bt->page, bt->mgr->page_size); - bt->cursor_page = bt->page_no; - bt_unlockpage(BtLockRead, bt->set); - bt_unpinlatch (bt->set); - bt_unpinpool (bt->pool); + + if( slot = bt_loadpage (bt, set, key, len, 0, BtLockRead) ) + memcpy (bt->cursor, set->page, bt->mgr->page_size); + else + return 0; + + bt->cursor_page = set->page_no; + + bt_unlockpage(BtLockRead, set->latch); + bt_unpinlatch (set->latch); + bt_unpinpool (set->pool); return slot; } @@ -1981,16 +2039,16 @@ uint slot; uint bt_nextkey (BtDb *bt, uint slot) { -BtPool *pool; -BtPage page; +BtPageSet set[1]; uid right; do { right = bt_getid(bt->cursor->right); + while( slot++ < bt->cursor->cnt ) if( slotptr(bt->cursor,slot)->dead ) continue; - else if( right || (slot < bt->cursor->cnt)) + else if( right || (slot < bt->cursor->cnt) ) // skip infinite stopper return slot; else break; @@ -2000,20 +2058,21 @@ uid right; bt->cursor_page = right; - if( pool = bt_pinpool (bt, right) ) - page = bt_page (bt, pool, right); + if( set->pool = bt_pinpool (bt, right) ) + set->page = bt_page (bt, set->pool, right); else return 0; - bt->set = bt_pinlatch (bt, right); - bt_lockpage(BtLockRead, bt->set); + set->latch = bt_pinlatch (bt, right); + bt_lockpage(BtLockRead, set->latch); - memcpy (bt->cursor, page, bt->mgr->page_size); + memcpy (bt->cursor, set->page, bt->mgr->page_size); - bt_unlockpage(BtLockRead, bt->set); - bt_unpinlatch (bt->set); - bt_unpinpool (pool); + bt_unlockpage(BtLockRead, set->latch); + bt_unpinlatch (set->latch); + bt_unpinpool (set->pool); slot = 0; + } while( 1 ); return bt->err = 0; @@ -2036,50 +2095,132 @@ uint bt_tod(BtDb *bt, uint slot) #ifdef STANDALONE +#ifndef unix +double getCpuTime(int type) +{ +FILETIME crtime[1]; +FILETIME xittime[1]; +FILETIME systime[1]; +FILETIME usrtime[1]; +SYSTEMTIME timeconv[1]; +double ans = 0; + + memset (timeconv, 0, sizeof(SYSTEMTIME)); + + switch( type ) { + case 0: + GetSystemTimeAsFileTime (xittime); + FileTimeToSystemTime (xittime, timeconv); + ans = (double)timeconv->wDayOfWeek * 3600 * 24; + break; + case 1: + GetProcessTimes (GetCurrentProcess(), crtime, xittime, systime, usrtime); + FileTimeToSystemTime (usrtime, timeconv); + break; + case 2: + GetProcessTimes (GetCurrentProcess(), crtime, xittime, systime, usrtime); + FileTimeToSystemTime (systime, timeconv); + break; + } + + ans += (double)timeconv->wHour * 3600; + ans += (double)timeconv->wMinute * 60; + ans += (double)timeconv->wSecond; + ans += (double)timeconv->wMilliseconds / 1000; + return ans; +} +#else +#include +#include + +double getCpuTime(int type) +{ +struct rusage used[1]; +struct timeval tv[1]; + + switch( type ) { + case 0: + gettimeofday(tv, NULL); + return (double)tv->tv_sec + (double)tv->tv_usec / 1000000; + + case 1: + getrusage(RUSAGE_SELF, used); + return (double)used->ru_utime.tv_sec + (double)used->ru_utime.tv_usec / 1000000; + + case 2: + getrusage(RUSAGE_SELF, used); + return (double)used->ru_stime.tv_sec + (double)used->ru_stime.tv_usec / 1000000; + } + + return 0; +} +#endif + void bt_latchaudit (BtDb *bt) { ushort idx, hashidx; -BtLatchSet *set; -BtPool *pool; -BtPage page; -uid page_no; +uid next, page_no; +BtLatchSet *latch; +BtKey ptr; #ifdef unix - for( idx = 1; idx < bt->mgr->latchmgr->latchdeployed; idx++ ) { - set = bt->mgr->latchsets + idx; - if( *(ushort *)set->readwr || *(ushort *)set->access || *(ushort *)set->parent ) { - fprintf(stderr, "latchset %d locked for page %6x\n", idx, set->page_no); - *(ushort *)set->readwr = 0; - *(ushort *)set->access = 0; - *(ushort *)set->parent = 0; - } - if( set->pin ) { - fprintf(stderr, "latchset %d pinned\n", idx); - set->pin = 0; + if( *(uint *)(bt->mgr->latchmgr->lock) ) + fprintf(stderr, "Alloc page locked\n"); + *(uint *)(bt->mgr->latchmgr->lock) = 0; + + for( idx = 1; idx <= bt->mgr->latchmgr->latchdeployed; idx++ ) { + latch = bt->mgr->latchsets + idx; + if( *(uint *)latch->readwr ) + fprintf(stderr, "latchset %d rwlocked for page %.8x\n", idx, latch->page_no); + *(uint *)latch->readwr = 0; + + if( *(uint *)latch->access ) + fprintf(stderr, "latchset %d accesslocked for page %.8x\n", idx, latch->page_no); + *(uint *)latch->access = 0; + + if( *(uint *)latch->parent ) + fprintf(stderr, "latchset %d parentlocked for page %.8x\n", idx, latch->page_no); + *(uint *)latch->parent = 0; + + if( latch->pin ) { + fprintf(stderr, "latchset %d pinned for page %.8x\n", idx, latch->page_no); + latch->pin = 0; } } for( hashidx = 0; hashidx < bt->mgr->latchmgr->latchhash; hashidx++ ) { - if( *(uint *)bt->mgr->latchmgr->table[hashidx].latch ) - fprintf(stderr, "latchmgr locked\n"); + if( *(uint *)(bt->mgr->latchmgr->table[hashidx].latch) ) + fprintf(stderr, "hash entry %d locked\n", hashidx); + + *(uint *)(bt->mgr->latchmgr->table[hashidx].latch) = 0; + if( idx = bt->mgr->latchmgr->table[hashidx].slot ) do { - set = bt->mgr->latchsets + idx; - if( *(uint *)set->readwr || *(ushort *)set->access || *(ushort *)set->parent ) - fprintf(stderr, "latchset %d locked\n", idx); - if( set->hash != hashidx ) + latch = bt->mgr->latchsets + idx; + if( *(uint *)latch->busy ) + fprintf(stderr, "latchset %d busylocked for page %.8x\n", idx, latch->page_no); + *(uint *)latch->busy = 0; + if( latch->hash != hashidx ) fprintf(stderr, "latchset %d wrong hashidx\n", idx); - if( set->pin ) - fprintf(stderr, "latchset %d pinned\n", idx); - } while( idx = set->next ); + if( latch->pin ) + fprintf(stderr, "latchset %d pinned for page %.8x\n", idx, latch->page_no); + } while( idx = latch->next ); } - page_no = bt_getid(bt->mgr->latchmgr->alloc[1].right); - - while( page_no ) { - fprintf(stderr, "free: %.6x\n", (uint)page_no); - pool = bt_pinpool (bt, page_no); - page = bt_page (bt, pool, page_no); - page_no = bt_getid(page->right); - bt_unpinpool (pool); + + next = bt->mgr->latchmgr->nlatchpage + LATCH_page; + page_no = LEAF_page; + + while( page_no < bt_getid(bt->mgr->latchmgr->alloc->right) ) { + pread (bt->mgr->idx, bt->frame, bt->mgr->page_size, page_no << bt->mgr->page_bits); + if( !bt->frame->free ) + for( idx = 0; idx++ < bt->frame->cnt - 1; ) { + ptr = keyptr(bt->frame, idx+1); + if( keycmp (keyptr(bt->frame, idx), ptr->key, ptr->len) >= 0 ) + fprintf(stderr, "page %.8x idx %.2x out of order\n", page_no, idx); + } + + if( page_no > LEAF_page ) + next = page_no + 1; + page_no = next; } #endif } @@ -2105,9 +2246,8 @@ uid next, page_no = LEAF_page; // start on first page of leaves unsigned char key[256]; ThreadArg *args = arg; int ch, len = 0, slot; +BtPageSet set[1]; time_t tod[1]; -BtPool *pool; -BtPage page; BtKey ptr; BtDb *bt; FILE *in; @@ -2193,40 +2333,61 @@ FILE *in; break; case 's': - len = key[0] = 0; - - fprintf(stderr, "started reading\n"); - - if( slot = bt_startkey (bt, key, len) ) - slot--; - else - fprintf(stderr, "Error %d in StartKey. Syserror: %d\n", bt->err, errno), exit(0); - - while( slot = bt_nextkey (bt, slot) ) { - ptr = bt_key(bt, slot); - fwrite (ptr->key, ptr->len, 1, stdout); - fputc ('\n', stdout); - } + fprintf(stderr, "started scanning\n"); + do { + if( set->pool = bt_pinpool (bt, page_no) ) + set->page = bt_page (bt, set->pool, page_no); + else + break; + set->latch = bt_pinlatch (bt, page_no); + bt_lockpage (BtLockRead, set->latch); + next = bt_getid (set->page->right); + cnt += set->page->act; + + for( slot = 0; slot++ < set->page->cnt; ) + if( next || slot < set->page->cnt ) + if( !slotptr(set->page, slot)->dead ) { + ptr = keyptr(set->page, slot); + fwrite (ptr->key, ptr->len, 1, stdout); + fputc ('\n', stdout); + } + + bt_unlockpage (BtLockRead, set->latch); + bt_unpinlatch (set->latch); + bt_unpinpool (set->pool); + } while( page_no = next ); + cnt--; // remove stopper key + fprintf(stderr, " Total keys read %d\n", cnt); break; case 'c': - fprintf(stderr, "started reading\n"); + fprintf(stderr, "started counting\n"); + next = bt->mgr->latchmgr->nlatchpage + LATCH_page; + page_no = LEAF_page; - do { - if( bt->pool = bt_pinpool (bt, page_no) ) - page = bt_page (bt, bt->pool, page_no); - else - break; - bt->set = bt_pinlatch (bt, page_no); - bt_lockpage (BtLockRead, bt->set); - cnt += page->act; - next = bt_getid (page->right); - bt_unlockpage (BtLockRead, bt->set); - bt_unpinlatch (bt->set); - bt_unpinpool (bt->pool); - } while( page_no = next ); + while( page_no < bt_getid(bt->mgr->latchmgr->alloc->right) ) { + uid off = page_no << bt->mgr->page_bits; +#ifdef unix + pread (bt->mgr->idx, bt->frame, bt->mgr->page_size, off); +#else + DWORD amt[1]; + SetFilePointer (bt->mgr->idx, (long)off, (long*)(&off)+1, FILE_BEGIN); + + if( !ReadFile(bt->mgr->idx, bt->frame, bt->mgr->page_size, amt, NULL)) + return bt->err = BTERR_map; + + if( *amt < bt->mgr->page_size ) + return bt->err = BTERR_map; +#endif + if( !bt->frame->free && !bt->frame->lvl ) + cnt += bt->frame->act; + if( page_no > LEAF_page ) + next = page_no + 1; + page_no = next; + } + cnt--; // remove stopper key fprintf(stderr, " Total keys read %d\n", cnt); break; @@ -2246,16 +2407,15 @@ int main (int argc, char **argv) { int idx, cnt, len, slot, err; int segsize, bits = 16; +double start, stop; #ifdef unix pthread_t *threads; -timer start, stop; #else -time_t start[1], stop[1]; HANDLE *threads; #endif -double real_time; ThreadArg *args; uint poolsize = 0; +float elapsed; int num = 0; char key[1]; BtMgr *mgr; @@ -2272,11 +2432,7 @@ BtDb *bt; exit(0); } -#ifdef unix - gettimeofday(&start, NULL); -#else - time(start); -#endif + start = getCpuTime(0); if( argc > 3 ) bits = atoi(argv[3]); @@ -2334,18 +2490,20 @@ BtDb *bt; #ifdef unix for( idx = 0; idx < cnt; idx++ ) pthread_join (threads[idx], NULL); - gettimeofday(&stop, NULL); - real_time = 1000.0 * ( stop.tv_sec - start.tv_sec ) + 0.001 * (stop.tv_usec - start.tv_usec ); #else WaitForMultipleObjects (cnt, threads, TRUE, INFINITE); for( idx = 0; idx < cnt; idx++ ) CloseHandle(threads[idx]); - time (stop); - real_time = 1000 * (*stop - *start); #endif - fprintf(stderr, " Time to complete: %.2f seconds\n", real_time/1000); + elapsed = getCpuTime(0) - start; + fprintf(stderr, " real %dm%.3fs\n", (int)(elapsed/60), elapsed - (int)(elapsed/60)*60); + elapsed = getCpuTime(1); + fprintf(stderr, " user %dm%.3fs\n", (int)(elapsed/60), elapsed - (int)(elapsed/60)*60); + elapsed = getCpuTime(2); + fprintf(stderr, " sys %dm%.3fs\n", (int)(elapsed/60), elapsed - (int)(elapsed/60)*60); + bt_mgrclose (mgr); }