X-Git-Url: https://pd.if.org/git/?p=btree;a=blobdiff_plain;f=fosterbtreee.c;h=89be3c0c9bfc560385d30afc3194d493e26f38b7;hp=8d18fed9353ec1b77c51bbe2e97b34eee81b1f27;hb=0756819e460acb958eeb073f4d5ca9ddf66ec6c7;hpb=81d755dfab3eb03bf41b888a1e64629c3e396854 diff --git a/fosterbtreee.c b/fosterbtreee.c index 8d18fed..89be3c0 100644 --- a/fosterbtreee.c +++ b/fosterbtreee.c @@ -1,5 +1,5 @@ // foster btree version e -// 30 DEC 2013 +// 29 JAN 2014 // author: karl malbrain, malbrain@cal.berkeley.edu @@ -97,7 +97,7 @@ typedef enum{ // the tod field from the key. // Keys are marked dead, but remain on the page until -// it cleanup is called. The fence key (highest key) for +// cleanup is called. The fence key (highest key) for // the page is always present, even after cleanup. typedef struct { @@ -121,13 +121,12 @@ typedef struct { // by the BtSlot array of keys. typedef struct Page { - uint cnt; // count of keys in page - uint act; // count of active keys - uint min; // next key offset - uint foster; // count of foster children + volatile uint cnt; // count of keys in page + volatile uint act; // count of active keys + volatile uint min; // next key offset + volatile uint foster; // count of foster children unsigned char bits; // page size in bits - unsigned char lvl:6; // level of page - unsigned char kill:1; // page is being deleted + unsigned char lvl:7; // level of page unsigned char dirty:1; // page needs to be cleaned unsigned char right[BtId]; // page number to right } *BtPage; @@ -159,8 +158,7 @@ typedef struct { volatile ushort slot; // Latch table entry at head of chain } BtHashEntry; -// latch table lock structure -// implements a fair read-write lock +// latch manager table structure typedef struct { #ifdef unix @@ -219,20 +217,19 @@ typedef struct { uint mode; // read-write mode #ifdef unix int idx; - char *pooladvise; // bit maps for pool page advisements #else HANDLE idx; #endif ushort poolcnt; // highest page pool node in use ushort poolmax; // highest page pool node allocated - ushort poolmask; // total size of pages in mmap segment - 1 + ushort poolmask; // total number of pages in mmap segment - 1 ushort hashsize; // size of Hash Table for pool entries ushort evicted; // last evicted hash table slot ushort *hash; // hash table of pool entries BtPool *pool; // memory pool page segments BtSpinLatch *latch; // latches for pool hash slots BtLatchMgr *latchmgr; // mapped latch page from allocation page - BtLatchSet *latchset; // first mapped latch set from latch pages + BtLatchSet *latchsets; // mapped latch set from latch pages #ifndef unix HANDLE halloc; // allocation and latch table handle #endif @@ -240,14 +237,17 @@ typedef struct { typedef struct { BtMgr *mgr; // buffer manager for thread - BtPage temp; // temporary frame buffer (memory mapped/file IO) 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 foster; // last search was to foster child + int found; // last delete was found int err; // last error } BtDb; @@ -265,8 +265,8 @@ 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, uid id, uint tod); -extern BTERR bt_deletekey (BtDb *bt, unsigned char *key, uint len, uint lvl); +extern BTERR bt_insertkey (BtDb *bt, unsigned char *key, uint len, uid id, uint tod, uint lvl); +extern BTERR bt_deletekey (BtDb *bt, unsigned char *key, uint len); extern uid bt_findkey (BtDb *bt, unsigned char *key, uint len); extern uint bt_startkey (BtDb *bt, unsigned char *key, uint len); extern uint bt_nextkey (BtDb *bt, uint slot); @@ -275,6 +275,11 @@ extern uint bt_nextkey (BtDb *bt, uint slot); extern BtMgr *bt_mgr (char *name, uint mode, uint bits, uint poolsize, uint segsize, uint hashsize); void bt_mgrclose (BtMgr *mgr); +// internal functions +BTERR bt_splitpage (BtDb *bt, BtPage page, BtPool *pool, BtLatchSet *set, uid page_no); +uint bt_cleanpage(BtDb *bt, BtPage page, uint amt, uint slot); +BTERR bt_mergeleft (BtDb *bt, BtPage page, BtPool *pool, BtLatchSet *set, uid page_no, uint lvl); + // Helper functions to return cursor slot values extern BtKey bt_key (BtDb *bt, uint slot); @@ -397,30 +402,30 @@ ushort prev; void bt_spinwritelock(BtSpinLatch *latch) { -ushort prev; - do { -#ifdef unix +#ifdef unix while( __sync_fetch_and_or((ushort *)latch, Mutex | Pending) & Mutex ) sched_yield(); #else while( _InterlockedOr16((ushort *)latch, Mutex | Pending) & Mutex ) SwitchToThread(); #endif - if( prev = !(latch->share | latch->exclusive) ) + if( !(latch->share | latch->exclusive) ) { #ifdef unix __sync_fetch_and_or((ushort *)latch, Write); + __sync_fetch_and_and ((ushort *)latch, ~(Mutex | Pending)); #else _InterlockedOr16((ushort *)latch, Write); + _InterlockedAnd16((ushort *)latch, ~(Mutex | Pending)); #endif + return; + } #ifdef unix - __sync_fetch_and_and ((ushort *)latch, ~(Mutex | Pending)); + __sync_fetch_and_and ((ushort *)latch, ~Mutex); #else - _InterlockedAnd16((ushort *)latch, ~(Mutex | Pending)); + _InterlockedAnd16((ushort *)latch, ~Mutex); #endif - if( prev ) - return; #ifdef unix sched_yield(); #else @@ -484,11 +489,17 @@ void bt_spinreleaseread(BtSpinLatch *latch) #endif } -void bt_initlockset (BtLatchSet *set) +void bt_initlockset (BtLatchSet *set, int reuse) { #ifdef unix pthread_rwlockattr_t rwattr[1]; + if( reuse ) { + pthread_rwlock_destroy (set->readwr->lock); + pthread_rwlock_destroy (set->access->lock); + pthread_rwlock_destroy (set->parent->lock); + } + pthread_rwlockattr_init (rwattr); pthread_rwlockattr_setkind_np (rwattr, PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP); pthread_rwlockattr_setpshared (rwattr, PTHREAD_PROCESS_SHARED); @@ -508,10 +519,10 @@ pthread_rwlockattr_t rwattr[1]; void bt_latchlink (BtDb *bt, ushort hashidx, ushort victim, uid page_no) { -BtLatchSet *set = bt->mgr->latchset + victim; +BtLatchSet *set = bt->mgr->latchsets + victim; if( set->next = bt->mgr->latchmgr->table[hashidx].slot ) - bt->mgr->latchset[set->next].prev = victim; + bt->mgr->latchsets[set->next].prev = victim; bt->mgr->latchmgr->table[hashidx].slot = victim; set->page_no = page_no; @@ -519,10 +530,19 @@ BtLatchSet *set = bt->mgr->latchset + victim; set->prev = 0; } +void bt_unpinlatch (BtLatchSet *set) +{ +#ifdef unix + __sync_fetch_and_add(&set->pin, -1); +#else + _InterlockedDecrement16 (&set->pin); +#endif +} + // find existing latchset or inspire new one // return with latchset pinned -BtLatchSet *bt_bindlatch (BtDb *bt, uid page_no, int incr) +BtLatchSet *bt_pinlatch (BtDb *bt, uid page_no) { ushort hashidx = page_no % bt->mgr->latchmgr->latchhash; ushort slot, avail = 0, victim, idx; @@ -534,12 +554,12 @@ BtLatchSet *set; if( slot = bt->mgr->latchmgr->table[hashidx].slot ) do { - set = bt->mgr->latchset + slot; + set = bt->mgr->latchsets + slot; if( page_no == set->page_no ) break; } while( slot = set->next ); - if( slot && incr ) { + if( slot ) { #ifdef unix __sync_fetch_and_add(&set->pin, 1); #else @@ -558,7 +578,7 @@ BtLatchSet *set; if( slot = bt->mgr->latchmgr->table[hashidx].slot ) do { - set = bt->mgr->latchset + slot; + set = bt->mgr->latchsets + slot; if( page_no == set->page_no ) break; if( !set->pin && !avail ) @@ -568,12 +588,11 @@ BtLatchSet *set; // found our entry, or take over an unpinned one if( slot || (slot = avail) ) { - set = bt->mgr->latchset + slot; - if( incr ) + set = bt->mgr->latchsets + slot; #ifdef unix - __sync_fetch_and_add(&set->pin, 1); + __sync_fetch_and_add(&set->pin, 1); #else - _InterlockedIncrement16 (&set->pin); + _InterlockedIncrement16 (&set->pin); #endif set->page_no = page_no; bt_spinreleasewrite(bt->mgr->latchmgr->table[hashidx].latch); @@ -588,14 +607,13 @@ BtLatchSet *set; #endif if( victim < bt->mgr->latchmgr->latchtotal ) { - set = bt->mgr->latchset + victim; - if( incr ) + set = bt->mgr->latchsets + victim; #ifdef unix - __sync_fetch_and_add(&set->pin, 1); + __sync_fetch_and_add(&set->pin, 1); #else - _InterlockedIncrement16 (&set->pin); + _InterlockedIncrement16 (&set->pin); #endif - bt_initlockset (set); + bt_initlockset (set, 0); bt_latchlink (bt, hashidx, victim, page_no); bt_spinreleasewrite (bt->mgr->latchmgr->table[hashidx].latch); return set; @@ -617,7 +635,7 @@ BtLatchSet *set; // we don't use slot zero if( victim %= bt->mgr->latchmgr->latchtotal ) - set = bt->mgr->latchset + victim; + set = bt->mgr->latchsets + victim; else continue; @@ -647,22 +665,20 @@ BtLatchSet *set; // unlink our available victim from its hash chain if( set->prev ) - bt->mgr->latchset[set->prev].next = set->next; + bt->mgr->latchsets[set->prev].next = set->next; else bt->mgr->latchmgr->table[idx].slot = set->next; if( set->next ) - bt->mgr->latchset[set->next].prev = set->prev; + bt->mgr->latchsets[set->next].prev = set->prev; bt_spinreleasewrite (bt->mgr->latchmgr->table[idx].latch); - - if( incr ) #ifdef unix - __sync_fetch_and_add(&set->pin, 1); + __sync_fetch_and_add(&set->pin, 1); #else - _InterlockedIncrement16 (&set->pin); + _InterlockedIncrement16 (&set->pin); #endif - + bt_initlockset (set, 1); bt_latchlink (bt, hashidx, victim, page_no); bt_spinreleasewrite (bt->mgr->latchmgr->table[hashidx].latch); bt_spinreleasewrite (set->busy); @@ -692,12 +708,19 @@ uint slot; #endif } +#ifdef unix + munmap (mgr->latchsets, mgr->latchmgr->nlatchpage * mgr->page_size); + munmap (mgr->latchmgr, mgr->page_size); +#else + FlushViewOfFile(mgr->latchmgr, 0); + UnmapViewOfFile(mgr->latchmgr); + CloseHandle(mgr->halloc); +#endif #ifdef unix close (mgr->idx); free (mgr->pool); free (mgr->hash); free (mgr->latch); - free (mgr->pooladvise); free (mgr); #else FlushFileBuffers(mgr->idx); @@ -733,7 +756,6 @@ BtMgr *bt_mgr (char *name, uint mode, uint bits, uint poolmax, uint segsize, uin uint lvl, attr, cacheblk, last, slot, idx; uint nlatchpage, latchhash; BtLatchMgr *latchmgr; -int lockmode; off64_t size; uint amt[1]; BtMgr* mgr; @@ -757,19 +779,8 @@ SYSTEM_INFO sysinfo[1]; #ifdef unix mgr = calloc (1, sizeof(BtMgr)); - switch (mode & 0x7fff) - { - case BT_rw: - mgr->idx = open ((char*)name, O_RDWR | O_CREAT, 0666); - lockmode = 1; - break; + mgr->idx = open ((char*)name, O_RDWR | O_CREAT, 0666); - case BT_ro: - default: - mgr->idx = open ((char*)name, O_RDONLY); - lockmode = 0; - break; - } if( mgr->idx == -1 ) return free(mgr), NULL; @@ -778,19 +789,8 @@ SYSTEM_INFO sysinfo[1]; #else mgr = GlobalAlloc (GMEM_FIXED|GMEM_ZEROINIT, sizeof(BtMgr)); attr = FILE_ATTRIBUTE_NORMAL; - switch (mode & 0x7fff) - { - case BT_rw: - mgr->idx = CreateFile(name, GENERIC_READ| GENERIC_WRITE, FILE_SHARE_READ|FILE_SHARE_WRITE, NULL, OPEN_ALWAYS, attr, NULL); - lockmode = 1; - break; + mgr->idx = CreateFile(name, GENERIC_READ| GENERIC_WRITE, FILE_SHARE_READ|FILE_SHARE_WRITE, NULL, OPEN_ALWAYS, attr, NULL); - case BT_ro: - default: - mgr->idx = CreateFile(name, GENERIC_READ, FILE_SHARE_READ|FILE_SHARE_WRITE, NULL, OPEN_EXISTING, attr, NULL); - lockmode = 0; - break; - } if( mgr->idx == INVALID_HANDLE_VALUE ) return GlobalFree(mgr), NULL; @@ -811,7 +811,7 @@ SYSTEM_INFO sysinfo[1]; else return free(mgr), free(latchmgr), NULL; } else if( mode == BT_ro ) - return bt_mgrclose (mgr), NULL; + return free(latchmgr), free (mgr), NULL; #else latchmgr = VirtualAlloc(NULL, BT_maxpage, MEM_COMMIT, PAGE_READWRITE); size = GetFileSize(mgr->idx, amt); @@ -853,7 +853,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)); @@ -887,7 +886,7 @@ SYSTEM_INFO sysinfo[1]; #ifdef unix if( write (mgr->idx, latchmgr, mgr->page_size) < mgr->page_size ) - return bt_mgrclose (mgr), NULL; + return free(latchmgr), bt_mgrclose (mgr), NULL; #else if( !WriteFile (mgr->idx, (char *)latchmgr, mgr->page_size, amt, NULL) ) return bt_mgrclose (mgr), NULL; @@ -943,25 +942,25 @@ SYSTEM_INFO sysinfo[1]; mgrlatch: #ifdef unix - flag = PROT_READ | ( mgr->mode == BT_ro ? 0 : PROT_WRITE ); + flag = PROT_READ | PROT_WRITE; mgr->latchmgr = mmap (0, mgr->page_size, flag, MAP_SHARED, mgr->idx, ALLOC_page * mgr->page_size); if( mgr->latchmgr == MAP_FAILED ) return bt_mgrclose (mgr), NULL; - mgr->latchset = (BtLatchSet *)mmap (0, mgr->latchmgr->nlatchpage * mgr->page_size, flag, MAP_SHARED, mgr->idx, LATCH_page * mgr->page_size); - if( mgr->latchset == MAP_FAILED ) + mgr->latchsets = (BtLatchSet *)mmap (0, mgr->latchmgr->nlatchpage * mgr->page_size, flag, MAP_SHARED, mgr->idx, LATCH_page * mgr->page_size); + if( mgr->latchsets == MAP_FAILED ) return bt_mgrclose (mgr), NULL; #else - flag = ( mgr->mode == BT_ro ? PAGE_READONLY : PAGE_READWRITE ); + flag = PAGE_READWRITE; mgr->halloc = CreateFileMapping(mgr->idx, NULL, flag, 0, (BT_latchtable / (mgr->page_size / sizeof(BtLatchSet)) + 1 + LATCH_page) * mgr->page_size, NULL); if( !mgr->halloc ) return bt_mgrclose (mgr), NULL; - flag = ( mgr->mode == BT_ro ? FILE_MAP_READ : FILE_MAP_WRITE ); + flag = FILE_MAP_WRITE; mgr->latchmgr = MapViewOfFile(mgr->halloc, flag, 0, 0, (BT_latchtable / (mgr->page_size / sizeof(BtLatchSet)) + 1 + LATCH_page) * mgr->page_size); if( !mgr->latchmgr ) return GetLastError(), bt_mgrclose (mgr), NULL; - mgr->latchset = (void *)((char *)mgr->latchmgr + LATCH_page * mgr->page_size); + mgr->latchsets = (void *)((char *)mgr->latchmgr + LATCH_page * mgr->page_size); #endif #ifdef unix @@ -989,6 +988,8 @@ BtDb *bt = malloc (sizeof(*bt)); bt->frame = (BtPage)bt->mem; bt->zero = (BtPage)(bt->mem + 1 * mgr->page_size); bt->cursor = (BtPage)(bt->mem + 2 * mgr->page_size); + + memset(bt->zero, 0, mgr->page_size); return bt; } @@ -1161,8 +1162,6 @@ int flag; pool->map = mmap (0, (bt->mgr->poolmask+1) << bt->mgr->page_bits, flag, MAP_SHARED, bt->mgr->idx, off); 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); @@ -1177,10 +1176,32 @@ int flag; return bt->err = 0; } +// calculate page within pool + +BtPage bt_page (BtDb *bt, BtPool *pool, uid page_no) +{ +uint subpage = (uint)(page_no & bt->mgr->poolmask); // page within mapping +BtPage page; + + page = (BtPage)(pool->map + (subpage << bt->mgr->page_bits)); + return page; +} + +// release pool pin + +void bt_unpinpool (BtPool *pool) +{ +#ifdef unix + __sync_fetch_and_add(&pool->pin, -1); +#else + _InterlockedDecrement16 (&pool->pin); +#endif +} + // find or place requested page in segment-pool // return pool table entry, incrementing pin -BtPool *bt_pinpage(BtDb *bt, uid page_no) +BtPool *bt_pinpool(BtDb *bt, uid page_no) { BtPool *pool, *node, *next; uint slot, idx, victim; @@ -1322,41 +1343,10 @@ BtLatchSet *set; } // place write, read, or parent lock on requested page_no. -// pin to buffer pool and return page pointer +// pin to buffer pool and return latchset pointer -BTERR bt_lockpage(BtDb *bt, uid page_no, BtLock mode, BtPage *pageptr) +void bt_lockpage(BtLock mode, BtLatchSet *set) { -BtLatchSet *set; -BtPool *pool; -uint subpage; -BtPage page; - - // find/create maping in pool table - // and pin our pool slot - - if( pool = bt_pinpage(bt, page_no) ) - subpage = (uint)(page_no & bt->mgr->poolmask); // page within mapping - else - return bt->err; - - if( !(set = bt_bindlatch (bt, page_no, 1)) ) - return bt->err; - - page = (BtPage)(pool->map + (subpage << bt->mgr->page_bits)); - -#ifdef unix - { - uint idx = subpage / 8; - uint bit = subpage % 8; - - if( mode == BtLockRead || mode == BtLockWrite ) - 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 - switch( mode ) { case BtLockRead: bt_readlock (set->readwr); @@ -1373,39 +1363,13 @@ BtPage page; case BtLockParent: bt_writelock (set->parent); break; - default: - return bt->err = BTERR_lock; } - - if( pageptr ) - *pageptr = page; - - return bt->err = 0; } // remove write, read, or parent lock on requested page_no. -BTERR bt_unlockpage(BtDb *bt, uid page_no, BtLock mode) +void bt_unlockpage(BtLock mode, BtLatchSet *set) { -BtLatchSet *set; -BtPool *pool; -uint idx; - - // since page is pinned - // it should still be in the buffer pool - // and is in no danger of being a victim for reuse - - if( !(set = bt_bindlatch (bt, page_no, 0)) ) - return bt->err = BTERR_latch; - - idx = (uint)(page_no >> bt->mgr->seg_bits) % bt->mgr->hashsize; - bt_spinreadlock (&bt->mgr->latch[idx]); - - if( !(pool = bt_findpool(bt, page_no, idx)) ) - return bt->err = BTERR_hash; - - bt_spinreleaseread (&bt->mgr->latch[idx]); - switch( mode ) { case BtLockRead: bt_releaseread (set->readwr); @@ -1422,65 +1386,15 @@ uint idx; case BtLockParent: bt_releasewrite (set->parent); break; - default: - return bt->err = BTERR_lock; } - -#ifdef unix - __sync_fetch_and_add(&pool->pin, -1); - __sync_fetch_and_add (&set->pin, -1); -#else - _InterlockedDecrement16 (&pool->pin); - _InterlockedDecrement16 (&set->pin); -#endif - return bt->err = 0; -} - -// deallocate a deleted page -// place on free chain out of allocator page -// fence key must already be removed from parent - -BTERR bt_freepage(BtDb *bt, uid page_no) -{ - // obtain delete lock on deleted page - - if( bt_lockpage(bt, page_no, BtLockDelete, NULL) ) - return bt->err; - - // obtain write lock on deleted page - - if( bt_lockpage(bt, page_no, BtLockWrite, &bt->temp) ) - return bt->err; - - // lock allocation page - - bt_spinwritelock(bt->mgr->latchmgr->lock); - - // store free chain in allocation page second right - bt_putid(bt->temp->right, bt_getid(bt->mgr->latchmgr->alloc[1].right)); - bt_putid(bt->mgr->latchmgr->alloc[1].right, page_no); - - // unlock page zero - - bt_spinreleasewrite(bt->mgr->latchmgr->lock); - - // remove write lock on deleted node - - if( bt_unlockpage(bt, page_no, BtLockWrite) ) - return bt->err; - - // remove delete lock on deleted node - - if( bt_unlockpage(bt, page_no, BtLockDelete) ) - return bt->err; - - return 0; } // allocate a new page and write page into it uid bt_newpage(BtDb *bt, BtPage page) { +BtLatchSet *set; +BtPool *pool; uid new_page; BtPage pmap; int reuse; @@ -1493,11 +1407,12 @@ int reuse; // else allocate empty page if( new_page = bt_getid(bt->mgr->latchmgr->alloc[1].right) ) { - if( bt_lockpage (bt, new_page, BtLockWrite, &bt->temp) ) - return 0; - bt_putid(bt->mgr->latchmgr->alloc[1].right, bt_getid(bt->temp->right)); - if( bt_unlockpage (bt, new_page, BtLockWrite) ) + if( pool = bt_pinpool (bt, new_page) ) + pmap = bt_page (bt, pool, new_page); + else return 0; + bt_putid(bt->mgr->latchmgr->alloc[1].right, bt_getid(pmap->right)); + bt_unpinpool (pool); reuse = 1; } else { new_page = bt_getid(bt->mgr->latchmgr->alloc->right); @@ -1505,33 +1420,39 @@ 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 ) - 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 ) { // 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 ) return bt->err = BTERR_wrt, 0; } + + // unlock allocation latch + + bt_spinreleasewrite(bt->mgr->latchmgr->lock); + + 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; + #else + // unlock allocation latch + + bt_spinreleasewrite(bt->mgr->latchmgr->lock); + // bring new page into pool and copy page. // this will extend the file into the new pages. + // NB -- no latch required - if( bt_lockpage(bt, new_page, BtLockWrite, &pmap) ) + if( pool = bt_pinpool (bt, new_page) ) + pmap = bt_page (bt, pool, new_page); + else return 0; memcpy(pmap, page, bt->mgr->page_size); - - if( bt_unlockpage (bt, new_page, BtLockWrite) ) - return 0; + bt_unpinpool (pool); #endif - // unlock allocation latch and return new page no - - bt_spinreleasewrite(bt->mgr->latchmgr->lock); return new_page; } @@ -1540,29 +1461,47 @@ int reuse; int bt_findslot (BtDb *bt, unsigned char *key, uint len) { uint diff, higher = bt->page->cnt, low = 1, slot; +uint good = 0; + + // if no right link + // make stopper key an infinite fence value + // by setting the good flag + + if( bt_getid (bt->page->right) ) + higher++; + else + good++; + + // low is the next candidate. + // loop ends when they meet - // low is the lowest candidate, higher is already - // tested as .ge. the given key, loop ends when they meet + // if good, higher is already + // tested as .ge. the given key. while( diff = higher - low ) { slot = low + ( diff >> 1 ); if( keycmp (keyptr(bt->page, slot), key, len) < 0 ) low = slot + 1; else - higher = slot; + higher = slot, good++; } - return higher; + // return zero if key is on right link page + + 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) +uint bt_loadpage (BtDb *bt, unsigned char *key, uint len, uint lvl, BtLock lock) { uid page_no = ROOT_page, prevpage = 0; +BtLatchSet *set, *prevset; uint drill = 0xff, slot; uint mode, prevmode; +BtPool *prevpool; +int foster = 0; // start at root of btree and drill down @@ -1570,84 +1509,104 @@ uint mode, prevmode; // determine lock mode of drill level mode = (lock == BtLockWrite) && (drill == lvl) ? BtLockWrite : BtLockRead; + // obtain latch set for this page + + bt->set = bt_pinlatch (bt, page_no); bt->page_no = page_no; + // pin page contents + + if( bt->pool = bt_pinpool (bt, page_no) ) + bt->page = bt_page (bt, bt->pool, page_no); + else + return 0; + // obtain access lock using lock chaining with Access mode if( page_no > ROOT_page ) - if( bt_lockpage(bt, page_no, BtLockAccess, NULL) ) - return 0; + bt_lockpage(BtLockAccess, bt->set); - // now unlock our (possibly foster) parent + // now unlock and unpin our (possibly foster) parent - if( prevpage ) - if( bt_unlockpage(bt, prevpage, prevmode) ) - return 0; - else - prevpage = 0; + if( prevpage ) { + bt_unlockpage(prevmode, prevset); + bt_unpinlatch (prevset); + bt_unpinpool (prevpool); + prevpage = 0; + } // obtain read lock using lock chaining - // and pin page contents - if( bt_lockpage(bt, page_no, mode, &bt->page) ) - return 0; + bt_lockpage(mode, bt->set); if( page_no > ROOT_page ) - if( bt_unlockpage(bt, page_no, BtLockAccess) ) - return 0; + bt_unlockpage(BtLockAccess, bt->set); // re-read and re-lock root after determining actual level of root - if( bt->page_no == ROOT_page ) + if( page_no == ROOT_page ) if( bt->page->lvl != drill) { drill = bt->page->lvl; - if( lock == BtLockWrite && drill == lvl ) - if( bt_unlockpage(bt, page_no, mode) ) - return 0; - else - continue; + if( lock == BtLockWrite && drill == lvl ) { + bt_unlockpage(mode, bt->set); + bt_unpinlatch (bt->set); + bt_unpinpool (bt->pool); + continue; + } } - prevpage = bt->page_no; - prevmode = mode; - - // if page is being deleted, - // move back to preceeding page - - if( bt->page->kill ) { - page_no = bt_getid (bt->page->right); - continue; - } - // find key on page at this level - // and descend to requested level + // and either descend to requested level + // or return key slot - slot = bt_findslot (bt, key, len); + if( slot = bt_findslot (bt, key, len) ) { + // is this slot < foster child area + // on the requested level? - // is this slot a foster child? + // if so, return actual slot even if dead - if( slot <= bt->page->cnt - bt->page->foster ) - if( drill == lvl ) - return slot; + if( slot <= bt->page->cnt - bt->page->foster ) + if( drill == lvl ) + return bt->foster = foster, slot; + + // find next active slot + // note: foster children are never dead - while( slotptr(bt->page, slot)->dead ) - if( slot++ < bt->page->cnt ) + while( slotptr(bt->page, slot)->dead ) + if( slot++ < bt->page->cnt ) continue; - else + else { + // we are waiting for fence key posting + page_no = bt_getid(bt->page->right); goto slideright; + } + + // is this slot < foster child area + // if so, drill to next level - if( slot <= bt->page->cnt - bt->page->foster ) - drill--; + if( slot <= bt->page->cnt - bt->page->foster ) + foster = 0, drill--; + else + foster = 1; - // continue down / right using overlapping locks - // to protect pages being killed or split. + // continue right onto foster child + // or down to next level. - page_no = bt_getid(slotptr(bt->page, slot)->id); - continue; + page_no = bt_getid(slotptr(bt->page, slot)->id); + + // or slide right into next page + + } else { + page_no = bt_getid(bt->page->right); + foster = 1; + } slideright: - page_no = bt_getid(bt->page->right); + prevpage = bt->page_no; + prevpool = bt->pool; + prevset = bt->set; + prevmode = mode; } while( page_no ); @@ -1657,108 +1616,386 @@ slideright: return 0; // return error } -// find and delete key on page by marking delete flag bit -// when page becomes empty, delete it from the btree +// remove empty page from the B-tree +// by pulling our right node left over ourselves + +// call with bt->page, etc, set to page's locked parent +// returns with page locked. -BTERR bt_deletekey (BtDb *bt, unsigned char *key, uint len, uint lvl) +BTERR bt_mergeright (BtDb *bt, BtPage page, BtPool *pool, BtLatchSet *set, uid page_no, uint lvl, uint slot) { -unsigned char leftkey[256], rightkey[256]; -uid page_no, right; -uint slot, tod; +BtLatchSet *rset, *pset, *rpset; +BtPool *rpool, *ppool, *rppool; +BtPage rpage, ppage, rppage; +uid right, parent, rparent; BtKey ptr; +uint idx; - if( slot = bt_loadpage (bt, key, len, lvl, BtLockWrite) ) - ptr = keyptr(bt->page, slot); + // cache node's parent page + + parent = bt->page_no; + ppage = bt->page; + ppool = bt->pool; + pset = bt->set; + + // lock and map our right page + // note that it cannot be our foster child + // since the our node is empty + // and it cannot be NULL because of the stopper + // in the last right page + + bt_lockpage (BtLockWrite, set); + + // if we aren't dead yet + + if( page->act ) + goto rmergexit; + + if( right = bt_getid (page->right) ) + if( rpool = bt_pinpool (bt, right) ) + rpage = bt_page (bt, rpool, right); + else + return bt->err; else + return bt->err = BTERR_struct; + + rset = bt_pinlatch (bt, right); + + // find our right neighbor + + if( ppage->act > 1 ) { + for( idx = slot; idx++ < ppage->cnt; ) + if( !slotptr(ppage, idx)->dead ) + break; + + if( idx > ppage->cnt ) + return bt->err = BTERR_struct; + + // redirect right neighbor in parent to left node + + bt_putid(slotptr(ppage,idx)->id, page_no); + } + + // if parent has only our deleted page, e.g. no right neighbor + // prepare to merge parent itself + + if( ppage->act == 1 ) { + if( rparent = bt_getid (ppage->right) ) + if( rppool = bt_pinpool (bt, rparent) ) + rppage = bt_page (bt, rppool, rparent); + else return bt->err; + else + return bt->err = BTERR_struct; - // if key is found delete it, otherwise ignore request + rpset = bt_pinlatch (bt, rparent); + bt_lockpage (BtLockWrite, rpset); + + // find our right neighbor on right parent page - if( !keycmp (ptr, key, len) ) - if( slotptr(bt->page, slot)->dead == 0 ) { - slotptr(bt->page,slot)->dead = 1; - if( slot < bt->page->cnt ) - bt->page->dirty = 1; - bt->page->act--; + for( idx = 0; idx++ < rppage->cnt; ) + if( !slotptr(rppage, idx)->dead ) { + bt_putid (slotptr(rppage, idx)->id, page_no); + break; } - // return if page is not empty, or it has no right sibling + if( idx > rppage->cnt ) + return bt->err = BTERR_struct; + } - right = bt_getid(bt->page->right); - page_no = bt->page_no; + // now that there are no more pointers to our right node + // we can wait for delete lock on it - if( !right || bt->page->act ) - return bt_unlockpage(bt, page_no, BtLockWrite); + bt_lockpage(BtLockDelete, rset); + bt_lockpage(BtLockWrite, rset); - // obtain Parent lock over write lock + // pull contents of right page into our empty page - if( bt_lockpage(bt, page_no, BtLockParent, NULL) ) - return bt->err; + memcpy (page, rpage, bt->mgr->page_size); + + // ready to release right parent lock + // now that we have a new page in place - // cache copy of key to delete + if( ppage->act == 1 ) { + bt_unlockpage (BtLockWrite, rpset); + bt_unpinlatch (rpset); + bt_unpinpool (rppool); + } - ptr = keyptr(bt->page, bt->page->cnt); - memcpy(leftkey, ptr, ptr->len + 1); + // add killed right block to free chain + // lock latch mgr - // lock and map right page + bt_spinwritelock(bt->mgr->latchmgr->lock); - if( bt_lockpage(bt, right, BtLockWrite, &bt->temp) ) - return bt->err; + // store free chain in allocation page second right - // pull contents of next page into current empty page - memcpy (bt->page, bt->temp, bt->mgr->page_size); + bt_putid(rpage->right, bt_getid(bt->mgr->latchmgr->alloc[1].right)); + bt_putid(bt->mgr->latchmgr->alloc[1].right, right); - // cache copy of key to update - ptr = keyptr(bt->temp, bt->temp->cnt); - memcpy(rightkey, ptr, ptr->len + 1); + // unlock latch mgr and right page - // Mark right page as deleted and point it to left page - // until we can post updates at higher level. + bt_unlockpage(BtLockDelete, rset); + bt_unlockpage(BtLockWrite, rset); + bt_unpinlatch (rset); + bt_unpinpool (rpool); - bt_putid(bt->temp->right, page_no); - bt->temp->kill = 1; - bt->temp->cnt = 0; + bt_spinreleasewrite(bt->mgr->latchmgr->lock); - if( bt_unlockpage(bt, right, BtLockWrite) ) - return bt->err; - if( bt_unlockpage(bt, page_no, BtLockWrite) ) + // delete our obsolete fence key from our parent + + slotptr(ppage, slot)->dead = 1; + ppage->dirty = 1; + + // if our parent now empty + // remove it from the tree + + if( ppage->act-- == 1 ) + if( bt_mergeleft (bt, ppage, ppool, pset, parent, lvl+1) ) return bt->err; - // delete old lower key to consolidated node +rmergexit: + bt_unlockpage (BtLockWrite, pset); + bt_unpinlatch (pset); + bt_unpinpool (ppool); + + bt->found = 1; + return bt->err = 0; +} + +// remove empty page from the B-tree +// try merging left first. If no left +// sibling, then merge right. + +// call with page loaded and locked, +// return with page locked. - if( bt_deletekey (bt, leftkey + 1, *leftkey, lvl + 1) ) +BTERR bt_mergeleft (BtDb *bt, BtPage page, BtPool *pool, BtLatchSet *set, uid page_no, uint lvl) +{ +unsigned char fencekey[256], postkey[256]; +uint slot, idx, postfence = 0; +BtLatchSet *lset, *pset; +BtPool *lpool, *ppool; +BtPage lpage, ppage; +uid left, parent; +BtKey ptr; + + ptr = keyptr(page, page->cnt); + memcpy(fencekey, ptr, ptr->len + 1); + bt_unlockpage (BtLockWrite, set); + + // load and lock our parent + +retry: + if( !(slot = bt_loadpage (bt, fencekey+1, *fencekey, lvl+1, BtLockWrite)) ) return bt->err; - // redirect higher key directly to consolidated node + parent = bt->page_no; + ppage = bt->page; + ppool = bt->pool; + pset = bt->set; - if( slot = bt_loadpage (bt, rightkey+1, *rightkey, lvl+1, BtLockWrite) ) - ptr = keyptr(bt->page, slot); + // wait until we are not a foster child + + if( bt->foster ) { + bt_unlockpage (BtLockWrite, pset); + bt_unpinlatch (pset); + bt_unpinpool (ppool); +#ifdef unix + sched_yield(); +#else + SwitchToThread(); +#endif + goto retry; + } + + // find our left neighbor in our parent page + + for( idx = slot; --idx; ) + if( !slotptr(ppage, idx)->dead ) + break; + + // if no left neighbor, do right merge + + if( !idx ) + return bt_mergeright (bt, page, pool, set, page_no, lvl, slot); + + // lock and map our left neighbor's page + + left = bt_getid (slotptr(ppage, idx)->id); + + if( lpool = bt_pinpool (bt, left) ) + lpage = bt_page (bt, lpool, left); else return bt->err; - // since key already exists, update id + lset = bt_pinlatch (bt, left); + bt_lockpage(BtLockWrite, lset); - if( keycmp (ptr, rightkey+1, *rightkey) ) - return bt->err = BTERR_struct; + // wait until foster sibling is in our parent - slotptr(bt->page, slot)->dead = 0; - bt_putid(slotptr(bt->page,slot)->id, page_no); + if( bt_getid (lpage->right) != page_no ) { + bt_unlockpage (BtLockWrite, pset); + bt_unpinlatch (pset); + bt_unpinpool (ppool); + bt_unlockpage (BtLockWrite, lset); + bt_unpinlatch (lset); + bt_unpinpool (lpool); +#ifdef linux + sched_yield(); +#else + SwitchToThread(); +#endif + goto retry; + } - if( bt_unlockpage(bt, bt->page_no, BtLockWrite) ) - return bt->err; + // since our page will have no more pointers to it, + // obtain Delete lock and wait for write locks to clear - // obtain write lock and - // add right block to free chain + bt_lockpage(BtLockDelete, set); + bt_lockpage(BtLockWrite, set); + + // if we aren't dead yet, + // get ready for exit + + if( page->act ) { + bt_unlockpage(BtLockDelete, set); + bt_unlockpage(BtLockWrite, lset); + bt_unpinlatch (lset); + bt_unpinpool (lpool); + goto lmergexit; + } - if( bt_freepage (bt, right) ) + // are we are the fence key for our parent? + // if so, grab our old fence key + + if( postfence = slot == ppage->cnt ) { + ptr = keyptr (ppage, ppage->cnt); + memcpy(fencekey, ptr, ptr->len + 1); + memset(slotptr(ppage, ppage->cnt), 0, sizeof(BtSlot)); + + // clear out other dead slots + + while( --ppage->cnt ) + if( slotptr(ppage, ppage->cnt)->dead ) + memset(slotptr(ppage, ppage->cnt), 0, sizeof(BtSlot)); + else + break; + + ptr = keyptr (ppage, ppage->cnt); + memcpy(postkey, ptr, ptr->len + 1); + } else + slotptr(ppage,slot)->dead = 1; + + ppage->dirty = 1; + ppage->act--; + + // push our right neighbor pointer to our left + + memcpy (lpage->right, page->right, BtId); + + // add ourselves to free chain + // lock latch mgr + + bt_spinwritelock(bt->mgr->latchmgr->lock); + + // store free chain in allocation page second right + bt_putid(page->right, bt_getid(bt->mgr->latchmgr->alloc[1].right)); + bt_putid(bt->mgr->latchmgr->alloc[1].right, page_no); + + // unlock latch mgr and pages + + bt_spinreleasewrite(bt->mgr->latchmgr->lock); + bt_unlockpage(BtLockWrite, lset); + bt_unpinlatch (lset); + bt_unpinpool (lpool); + + // release our node's delete lock + + bt_unlockpage(BtLockDelete, set); + +lmergexit: + bt_unlockpage (BtLockWrite, pset); + bt_unpinpool (ppool); + + // do we need to post parent's fence key in its parent? + + if( !postfence || parent == ROOT_page ) { + bt_unpinlatch (pset); + bt->found = 1; + return bt->err = 0; + } + + // interlock parent fence post + + bt_lockpage (BtLockParent, pset); + + // load parent's parent page +posttry: + if( !(slot = bt_loadpage (bt, fencekey+1, *fencekey, lvl+2, BtLockWrite)) ) + return bt->err; + + if( !(slot = bt_cleanpage (bt, bt->page, *fencekey, slot)) ) + if( bt_splitpage (bt, bt->page, bt->pool, bt->set, bt->page_no) ) return bt->err; + else + goto posttry; + + page = bt->page; + + page->min -= *postkey + 1; + ((unsigned char *)page)[page->min] = *postkey; + memcpy ((unsigned char *)page + page->min +1, postkey + 1, *postkey ); + slotptr(page, slot)->off = page->min; + + bt_unlockpage (BtLockParent, pset); + bt_unpinlatch (pset); + + bt_unlockpage (BtLockWrite, bt->set); + bt_unpinlatch (bt->set); + bt_unpinpool (bt->pool); + + bt->found = 1; + return bt->err = 0; +} + +// find and delete key on page by marking delete flag bit +// if page becomes empty, delete it from the btree - // remove ParentModify lock +BTERR bt_deletekey (BtDb *bt, unsigned char *key, uint len) +{ +BtLatchSet *set; +BtPool *pool; +BtPage page; +uid page_no; +BtKey ptr; +uint slot; - if( bt_unlockpage(bt, page_no, BtLockParent) ) + if( !(slot = bt_loadpage (bt, key, len, 0, BtLockWrite)) ) return bt->err; - - return 0; + + page_no = bt->page_no; + page = bt->page; + pool = bt->pool; + set = bt->set; + + // if key is found delete it, otherwise ignore request + + ptr = keyptr(page, slot); + + if( bt->found = !keycmp (ptr, key, len) ) + if( bt->found = slotptr(page, slot)->dead == 0 ) { + slotptr(page,slot)->dead = 1; + if( slot < page->cnt ) + page->dirty = 1; + if( !--page->act ) + if( bt_mergeleft (bt, page, pool, set, page_no, 0) ) + return bt->err; + } + + bt_unlockpage(BtLockWrite, set); + bt_unpinlatch (set); + bt_unpinpool (pool); + return bt->err = 0; } // find key in leaf level and return row-id @@ -1777,32 +2014,32 @@ uid id; // if key exists, return row-id // otherwise return 0 - if( ptr->len == len && !memcmp (ptr->key, key, len) ) + if( slot <= bt->page->cnt && !keycmp (ptr, key, len) ) id = bt_getid(slotptr(bt->page,slot)->id); else id = 0; - if( bt_unlockpage (bt, bt->page_no, BtLockRead) ) - return 0; - + bt_unlockpage (BtLockRead, bt->set); + bt_unpinlatch (bt->set); + bt_unpinpool (bt->pool); return id; } // check page for space available, // clean if necessary and return // 0 - page needs splitting -// 1 - go ahead +// >0 new slot value -uint bt_cleanpage(BtDb *bt, uint amt) +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 = max; BtKey key; if( page->min >= (max+1) * sizeof(BtSlot) + sizeof(*page) + amt + 1 ) - return 1; + return slot; // skip cleanup if nothing to reclaim @@ -1819,12 +2056,19 @@ BtKey key; // try cleaning up page first + // always leave fence key in the array + // otherwise, remove deleted key + + // note: foster children are never dead + while( cnt++ < max ) { - // always leave fence key and foster children in list - if( cnt < max - page->foster && slotptr(bt->frame,cnt)->dead ) + if( cnt == slot ) + newslot = idx + 1; + if( cnt < max && slotptr(bt->frame,cnt)->dead ) continue; // copy key + key = keyptr(bt->frame, cnt); nxt -= key->len + 1; memcpy ((unsigned char *)page + nxt, key, key->len + 1); @@ -1843,7 +2087,7 @@ BtKey key; // see if page has enough space now, or does it need splitting? if( page->min >= (idx+1) * sizeof(BtSlot) + sizeof(*page) + amt + 1 ) - return 1; + return newslot; return 0; } @@ -1851,32 +2095,33 @@ BtKey key; // add key to current page // page must already be writelocked -void bt_addkeytopage (BtDb *bt, uint slot, unsigned char *key, uint len, uid id, uint tod) +void bt_addkeytopage (BtDb *bt, BtPage page, uint slot, unsigned char *key, uint len, uid id, uint tod) { -BtPage page = bt->page; uint idx; - // calculate next available slot and copy key into page + // find next available dead slot and copy key onto page + // note that foster children on the page are never dead - page->min -= len + 1; - ((unsigned char *)page)[page->min] = len; - memcpy ((unsigned char *)page + page->min +1, key, len ); + // look for next hole, but stay back from the fence key for( idx = slot; idx < page->cnt; idx++ ) if( slotptr(page, idx)->dead ) break; - // now insert key into array before slot - // preserving the fence slot - if( idx == page->cnt ) idx++, page->cnt++; page->act++; + // now insert key into array before slot + while( idx > slot ) *slotptr(page, idx) = *slotptr(page, idx -1), idx--; + page->min -= len + 1; + ((unsigned char *)page)[page->min] = len; + memcpy ((unsigned char *)page + page->min +1, key, len ); + bt_putid(slotptr(page,slot)->id, id); slotptr(page, slot)->off = page->min; slotptr(page, slot)->tod = tod; @@ -1899,9 +2144,11 @@ BtKey key; // contents into it from the root. Strip foster child key. // (it's the stopper key) + memset (slotptr(root, root->cnt), 0, sizeof(BtSlot)); + root->dirty = 1; + root->foster--; root->act--; root->cnt--; - root->foster--; // Save left fence key. @@ -1942,66 +2189,65 @@ BtKey key; root->act = 2; root->lvl++; - // release root (bt->page) + // release and unpin root (bt->page) - return bt_unlockpage(bt, ROOT_page, BtLockWrite); + bt_unlockpage(BtLockWrite, bt->set); + bt_unpinlatch (bt->set); + bt_unpinpool (bt->pool); + return 0; } // split already locked full node -// in current page variables -// return unlocked. +// return unlocked and unpinned. -BTERR bt_splitpage (BtDb *bt) +BTERR bt_splitpage (BtDb *bt, BtPage page, BtPool *pool, BtLatchSet *set, uid page_no) { uint slot, cnt, idx, max, nxt = bt->mgr->page_size; unsigned char fencekey[256]; -uid page_no = bt->page_no; -BtPage page = bt->page; uint tod = time(NULL); uint lvl = page->lvl; -uid new_page, right; +uid new_page; BtKey key; - // initialize frame buffer + // initialize frame buffer for right node memset (bt->frame, 0, bt->mgr->page_size); max = page->cnt - page->foster; - tod = (uint)time(NULL); cnt = max / 2; idx = 0; // split higher half of keys to bt->frame - // leaving foster children in the left node. + // leaving old foster children in the left node, + // and adding a new foster child there. while( cnt++ < max ) { key = keyptr(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) ) + bt->frame->act++; slotptr(bt->frame, idx)->tod = slotptr(page, cnt)->tod; slotptr(bt->frame, idx)->off = nxt; - bt->frame->act++; } - // transfer right link node + // transfer right link node to new right node - if( page_no > ROOT_page ) { - right = bt_getid (page->right); - bt_putid(bt->frame->right, right); - } + if( page_no > ROOT_page ) + memcpy (bt->frame->right, page->right, BtId); bt->frame->bits = bt->mgr->page_bits; bt->frame->min = nxt; bt->frame->cnt = idx; bt->frame->lvl = lvl; - // get new free page and write frame to it. + // get new free page and write right frame to it. if( !(new_page = bt_newpage(bt, bt->frame)) ) return bt->err; - // remember fence key for new page to add - // as foster child + // remember fence key for new right page to add + // as foster child to the left node key = keyptr(bt->frame, idx); memcpy (fencekey, key, key->len + 1); @@ -2011,6 +2257,7 @@ BtKey key; memcpy (bt->frame, page, bt->mgr->page_size); memset (page+1, 0, bt->mgr->page_size - sizeof(*page)); nxt = bt->mgr->page_size; + page->dirty = 0; page->act = 0; cnt = 0; idx = 0; @@ -2023,22 +2270,26 @@ BtKey key; nxt -= key->len + 1; memcpy ((unsigned char *)page + nxt, key, key->len + 1); 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->act++; } - // insert new foster child at beginning of the current foster children + // insert new foster child for right page in queue + // before any of the current foster children nxt -= *fencekey + 1; memcpy ((unsigned char *)page + nxt, fencekey, *fencekey + 1); + bt_putid (slotptr(page,++idx)->id, new_page); slotptr(page, idx)->tod = tod; slotptr(page, idx)->off = nxt; page->foster++; page->act++; - // continue with old foster child keys if any + // continue with old foster child keys + // note that none will be dead cnt = bt->frame->cnt - bt->frame->foster; @@ -2066,109 +2317,77 @@ BtKey key; // release wr lock on our page - if( bt_unlockpage (bt, page_no, BtLockWrite) ) - return bt->err; + bt_unlockpage (BtLockWrite, set); // obtain ParentModification lock for current page - // to fix fence key and highest foster child on page + // to fix new fence key and oldest foster child on page - if( bt_lockpage (bt, page_no, BtLockParent, NULL) ) - return bt->err; + bt_lockpage (BtLockParent, set); - // get our highest foster child key to find in parent node + // get our new fence key to insert in parent node - if( bt_lockpage (bt, page_no, BtLockRead, &page) ) - return bt->err; + bt_lockpage (BtLockRead, set); - key = keyptr(page, page->cnt); + key = keyptr(page, page->cnt-1); memcpy (fencekey, key, key->len+1); - if( bt_unlockpage (bt, page_no, BtLockRead) ) - return bt->err; - - // update our parent -try_again: - - do { - slot = bt_loadpage (bt, fencekey + 1, *fencekey, lvl + 1, BtLockWrite); + bt_unlockpage (BtLockRead, set); - if( !slot ) + if( bt_insertkey (bt, fencekey + 1, *fencekey, page_no, tod, lvl + 1) ) return bt->err; - // check if parent page has enough space for any possible key - - if( bt_cleanpage (bt, 256) ) - break; - - if( bt_splitpage (bt) ) - return bt->err; - } while( 1 ); - - // see if we are still a foster child from another node + // lock our page for writing - if( bt_getid (slotptr(bt->page, slot)->id) != page_no ) { - if( bt_unlockpage (bt, bt->page_no, BtLockWrite) ) - return bt->err; -#ifdef unix - sched_yield(); -#else - SwitchToThread(); -#endif - goto try_again; - } + bt_lockpage (BtLockRead, set); - // wait until readers from parent get their locks - // on our page + // switch old parent key from us to our oldest foster child - if( bt_lockpage (bt, page_no, BtLockDelete, NULL) ) - return bt->err; + key = keyptr(page, page->cnt); + memcpy (fencekey, key, key->len+1); - // lock our page for writing + new_page = bt_getid (slotptr(page, page->cnt)->id); + bt_unlockpage (BtLockRead, set); - if( bt_lockpage (bt, page_no, BtLockWrite, &page) ) + if( bt_insertkey (bt, fencekey + 1, *fencekey, new_page, tod, lvl + 1) ) return bt->err; - // switch parent fence key to foster child - - if( slotptr(page, page->cnt)->dead ) - slotptr(bt->page, slot)->dead = 1; - else - bt_putid (slotptr(bt->page, slot)->id, bt_getid(slotptr(page, page->cnt)->id)); - - // remove highest foster child from our page + // now that it has its own parent pointer, + // remove oldest foster child from our page + bt_lockpage (BtLockWrite, set); + memset (slotptr(page, page->cnt), 0, sizeof(BtSlot)); + page->dirty = 1; + page->foster--; page->cnt--; page->act--; - page->foster--; - page->dirty = 1; - key = keyptr(page, page->cnt); - // add our new fence key for foster child to our parent + bt_unlockpage (BtLockParent, set); - bt_addkeytopage (bt, slot, key->key, key->len, page_no, tod); + // if this emptied page, + // undo the foster child - if( bt_unlockpage (bt, bt->page_no, BtLockWrite) ) + if( !page->act ) + if( bt_mergeleft (bt, page, pool, set, page_no, lvl) ) return bt->err; - if( bt_unlockpage (bt, page_no, BtLockDelete) ) - return bt->err; + // unlock and unpin - if( bt_unlockpage (bt, page_no, BtLockWrite) ) - return bt->err; - - return bt_unlockpage (bt, page_no, BtLockParent); + bt_unlockpage (BtLockWrite, set); + bt_unpinlatch (set); + bt_unpinpool (pool); + return 0; } // Insert new key into the btree at leaf level. -BTERR bt_insertkey (BtDb *bt, unsigned char *key, uint len, uid id, uint tod) +BTERR bt_insertkey (BtDb *bt, unsigned char *key, uint len, uid id, uint tod, uint lvl) { uint slot, idx; BtPage page; BtKey ptr; while( 1 ) { - if( slot = bt_loadpage (bt, key, len, 0, BtLockWrite) ) + if( slot = bt_loadpage (bt, key, len, lvl, BtLockWrite) ) ptr = keyptr(bt->page, slot); else { @@ -2182,24 +2401,32 @@ BtKey ptr; page = bt->page; if( !keycmp (ptr, key, len) ) { + if( slotptr(page, slot)->dead ) + page->act++; slotptr(page, slot)->dead = 0; slotptr(page, slot)->tod = tod; bt_putid(slotptr(page,slot)->id, id); - return bt_unlockpage(bt, bt->page_no, BtLockWrite); + bt_unlockpage(BtLockWrite, bt->set); + bt_unpinlatch (bt->set); + bt_unpinpool (bt->pool); + return bt->err; } // check if page has enough space - if( bt_cleanpage (bt, len) ) + if( slot = bt_cleanpage (bt, bt->page, len, slot) ) break; - if( bt_splitpage (bt) ) + if( bt_splitpage (bt, bt->page, bt->pool, bt->set, bt->page_no) ) return bt->err; } - bt_addkeytopage (bt, slot, key, len, id, tod); + bt_addkeytopage (bt, bt->page, slot, key, len, id, tod); - return bt_unlockpage (bt, bt->page_no, BtLockWrite); + bt_unlockpage (BtLockWrite, bt->set); + bt_unpinlatch (bt->set); + bt_unpinpool (bt->pool); + return 0; } // cache page of keys into cursor and return starting slot for given key @@ -2211,10 +2438,12 @@ 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; - if ( bt_unlockpage(bt, bt->page_no, BtLockRead) ) - return 0; + bt_unlockpage(BtLockRead, bt->set); + bt_unpinlatch (bt->set); + bt_unpinpool (bt->pool); return slot; } @@ -2223,6 +2452,8 @@ uint slot; uint bt_nextkey (BtDb *bt, uint slot) { +BtLatchSet *set; +BtPool *pool; BtPage page; uid right; @@ -2240,15 +2471,19 @@ uid right; break; bt->cursor_page = right; - - if( bt_lockpage(bt, right, BtLockRead, &page) ) + if( pool = bt_pinpool (bt, right) ) + page = bt_page (bt, pool, right); + else return 0; - memcpy (bt->cursor, page, bt->mgr->page_size); + set = bt_pinlatch (bt, right); + bt_lockpage(BtLockRead, set); - if ( bt_unlockpage(bt, right, BtLockRead) ) - return 0; + memcpy (bt->cursor, page, bt->mgr->page_size); + bt_unlockpage(BtLockRead, set); + bt_unpinlatch (set); + bt_unpinpool (pool); slot = 0; } while( 1 ); @@ -2273,6 +2508,54 @@ uint bt_tod(BtDb *bt, uint slot) #ifdef STANDALONE +void bt_latchaudit (BtDb *bt) +{ +ushort idx, hashidx; +BtLatchSet *set; +BtPool *pool; +BtPage page; +uid page_no; + +#ifdef unix + for( idx = 1; idx < bt->mgr->latchmgr->latchdeployed; idx++ ) { + set = bt->mgr->latchsets + idx; + if( set->pin ) { + fprintf(stderr, "latchset %d pinned\n", idx); + set->pin = 0; + } + } + + for( hashidx = 0; hashidx < bt->mgr->latchmgr->latchhash; hashidx++ ) { + if( idx = bt->mgr->latchmgr->table[hashidx].slot ) do { + set = bt->mgr->latchsets + idx; + if( set->hash != hashidx ) + fprintf(stderr, "latchset %d wrong hashidx\n", idx); + if( set->pin ) + fprintf(stderr, "latchset %d pinned\n", idx); + } while( idx = set->next ); + } + page_no = LEAF_page; + + while( page_no ) { + fprintf(stderr, "page: %.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); + } + + 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); + } +#endif +} + typedef struct { char type, idx; char *infile; @@ -2294,7 +2577,9 @@ uid next, page_no = LEAF_page; // start on first page of leaves unsigned char key[256]; ThreadArg *args = arg; int ch, len = 0, slot; +BtLatchSet *set; time_t tod[1]; +BtPool *pool; BtPage page; BtKey ptr; BtDb *bt; @@ -2305,6 +2590,12 @@ FILE *in; switch(args->type | 0x20) { + case 'a': + fprintf(stderr, "started latch mgr audit\n"); + bt_latchaudit (bt); + fprintf(stderr, "finished latch mgr audit\n"); + break; + case 'w': fprintf(stderr, "started indexing for %s\n", args->infile); if( in = fopen (args->infile, "rb") ) @@ -2319,7 +2610,7 @@ FILE *in; else if( args->num ) sprintf((char *)key+len, "%.9d", line + args->idx * args->num), len += 9; - if( bt_insertkey (bt, key, len, line, *tod) ) + if( bt_insertkey (bt, key, len, line, *tod, 0) ) fprintf(stderr, "Error %d Line: %d\n", bt->err, line), exit(0); len = 0; } @@ -2341,7 +2632,7 @@ FILE *in; else if( args->num ) sprintf((char *)key+len, "%.9d", line + args->idx * args->num), len += 9; - if( bt_deletekey (bt, key, len, 0) ) + if( bt_deletekey (bt, key, len) ) fprintf(stderr, "Error %d Line: %d\n", bt->err, line), exit(0); len = 0; } @@ -2396,10 +2687,17 @@ FILE *in; fprintf(stderr, "started reading\n"); do { - bt_lockpage (bt, page_no, BtLockRead, &page); + if( pool = bt_pinpool (bt, page_no) ) + page = bt_page (bt, pool, page_no); + else + break; + set = bt_pinlatch (bt, page_no); + bt_lockpage (BtLockRead, set); cnt += page->act; next = bt_getid (page->right); - bt_unlockpage (bt, page_no, BtLockRead); + bt_unlockpage (BtLockRead, set); + bt_unpinlatch (set); + bt_unpinpool (pool); } while( page_no = next ); cnt--; // remove stopper key