X-Git-Url: https://pd.if.org/git/?a=blobdiff_plain;f=threadskv8.c;h=3f669b83d933d83d640cd3fe7079aacae2eb537f;hb=7a619addda2eab9c15567b62297a612b1d595f2c;hp=16ad09f024b44cef1ca8971a0e1b724bf52bd9e3;hpb=75e9a39cab3db94c792cbb16d56ed223ef5523a3;p=btree diff --git a/threadskv8.c b/threadskv8.c index 16ad09f..3f669b8 100644 --- a/threadskv8.c +++ b/threadskv8.c @@ -117,6 +117,8 @@ typedef struct { typedef struct { ushort ticket[1]; ushort serving[1]; + ushort tid; + ushort dup; } WOLock; #define PHID 0x1 @@ -275,6 +277,7 @@ typedef struct { uint latchtotal; // number of page latch entries uint latchhash; // number of latch hash table slots uint latchvictim; // next latch entry to examine + ushort thread_no[1]; // next thread number BtHashEntry *hashtable; // the buffer pool hash table entries BtLatchSet *latchsets; // mapped latch set from buffer pool unsigned char *pagepool; // mapped to the buffer pool pages @@ -294,6 +297,7 @@ typedef struct { int found; // last delete or insert was found int err; // last error int reads, writes; // number of reads and writes from the btree + ushort thread_no; // thread number } BtDb; typedef enum { @@ -409,9 +413,14 @@ uid bt_newdup (BtDb *bt) // Write-Only Queue Lock -void WriteOLock (WOLock *lock) +void WriteOLock (WOLock *lock, ushort tid) { ushort tix; + + if( lock->tid == tid ) { + lock->dup++; + return; + } #ifdef unix tix = __sync_fetch_and_add (lock->ticket, 1); #else @@ -425,10 +434,17 @@ ushort tix; #else SwitchToThread (); #endif + lock->tid = tid; } void WriteORelease (WOLock *lock) { + if( lock->dup ) { + lock->dup--; + return; + } + + lock->tid = 0; lock->serving[0]++; } @@ -1118,6 +1134,11 @@ BtDb *bt = malloc (sizeof(*bt)); #endif bt->frame = (BtPage)bt->mem; bt->cursor = (BtPage)(bt->mem + 1 * mgr->page_size); +#ifdef unix + bt->thread_no = __sync_fetch_and_add (mgr->thread_no, 1) + 1; +#else + bt->thread_no = _InterlockedIncrement16(mgr->thread_no, 1); +#endif return bt; } @@ -1145,7 +1166,7 @@ int ans; // place write, read, or parent lock on requested page_no. -void bt_lockpage(BtLock mode, BtLatchSet *latch) +void bt_lockpage(BtDb *bt, BtLock mode, BtLatchSet *latch) { switch( mode ) { case BtLockRead: @@ -1161,17 +1182,21 @@ void bt_lockpage(BtLock mode, BtLatchSet *latch) WriteLock (latch->access); break; case BtLockParent: - WriteOLock (latch->parent); + WriteOLock (latch->parent, bt->thread_no); break; case BtLockAtomic: - WriteOLock (latch->atomic); + WriteOLock (latch->atomic, bt->thread_no); + break; + case BtLockAtomic | BtLockRead: + WriteOLock (latch->atomic, bt->thread_no); + ReadLock (latch->readwr); break; } } // remove write, read, or parent lock on requested page -void bt_unlockpage(BtLock mode, BtLatchSet *latch) +void bt_unlockpage(BtDb *bt, BtLock mode, BtLatchSet *latch) { switch( mode ) { case BtLockRead: @@ -1192,6 +1217,10 @@ void bt_unlockpage(BtLock mode, BtLatchSet *latch) case BtLockAtomic: WriteORelease (latch->atomic); break; + case BtLockAtomic | BtLockRead: + WriteORelease (latch->atomic); + ReadRelease (latch->readwr); + break; } } @@ -1298,27 +1327,27 @@ uint mode, prevmode; // obtain access lock using lock chaining with Access mode if( page_no > ROOT_page ) - bt_lockpage(BtLockAccess, set->latch); + bt_lockpage(bt, BtLockAccess, set->latch); set->page = bt_mappage (bt, set->latch); // release & unpin parent or left sibling page if( prevpage ) { - bt_unlockpage(prevmode, prevlatch); + bt_unlockpage(bt, prevmode, prevlatch); bt_unpinlatch (prevlatch); prevpage = 0; } // obtain mode lock using lock chaining through AccessLock - bt_lockpage(mode, set->latch); + bt_lockpage(bt, mode, set->latch); if( set->page->free ) return bt->err = BTERR_struct, 0; if( page_no > ROOT_page ) - bt_unlockpage(BtLockAccess, set->latch); + bt_unlockpage(bt, BtLockAccess, set->latch); // re-read and re-lock root after determining actual level of root @@ -1329,7 +1358,7 @@ uint mode, prevmode; drill = set->page->lvl; if( lock != BtLockRead && drill == lvl ) { - bt_unlockpage(mode, set->latch); + bt_unlockpage(bt, mode, set->latch); bt_unpinlatch (set->latch); continue; } @@ -1342,10 +1371,8 @@ uint mode, prevmode; // find key on page at this level // and descend to requested level - if( set->page->kill ) - goto slideright; - - if( slot = bt_findslot (set->page, key, len) ) { + if( !set->page->kill ) + if( slot = bt_findslot (set->page, key, len) ) { if( drill == lvl ) return slot; @@ -1360,13 +1387,11 @@ uint mode, prevmode; page_no = bt_getid(valptr(set->page, slot)->value); drill--; continue; - } + } // or slide right into next page -slideright: page_no = bt_getid(set->page->right); - } while( page_no ); // return error on end of right chain @@ -1393,8 +1418,8 @@ void bt_freepage (BtDb *bt, BtPageSet *set) // unlock released page - bt_unlockpage (BtLockDelete, set->latch); - bt_unlockpage (BtLockWrite, set->latch); + bt_unlockpage (bt, BtLockDelete, set->latch); + bt_unlockpage (bt, BtLockWrite, set->latch); bt_unpinlatch (set->latch); // unlock allocation page @@ -1424,8 +1449,8 @@ uint idx; ptr = keyptr(set->page, set->page->cnt); memcpy (leftkey, ptr, ptr->len + sizeof(BtKey)); - bt_lockpage (BtLockParent, set->latch); - bt_unlockpage (BtLockWrite, set->latch); + bt_lockpage (bt, BtLockParent, set->latch); + bt_unlockpage (bt, BtLockWrite, set->latch); // insert new (now smaller) fence key @@ -1442,7 +1467,7 @@ uint idx; if( bt_deletekey (bt, ptr->key, ptr->len, lvl+1) ) return bt->err; - bt_unlockpage (BtLockParent, set->latch); + bt_unlockpage (bt, BtLockParent, set->latch); bt_unpinlatch(set->latch); return 0; } @@ -1470,8 +1495,8 @@ uint idx; else return bt->err; - bt_lockpage (BtLockDelete, child->latch); - bt_lockpage (BtLockWrite, child->latch); + bt_lockpage (bt, BtLockDelete, child->latch); + bt_lockpage (bt, BtLockWrite, child->latch); memcpy (root->page, child->page, bt->mgr->page_size); root->latch->dirty = 1; @@ -1480,7 +1505,7 @@ uint idx; } while( root->page->lvl > 1 && root->page->act == 1 ); - bt_unlockpage (BtLockWrite, root->latch); + bt_unlockpage (bt, BtLockWrite, root->latch); bt_unpinlatch (root->latch); return 0; } @@ -1513,7 +1538,7 @@ BtKey *ptr; else return 0; - bt_lockpage (BtLockWrite, right->latch); + bt_lockpage (bt, BtLockWrite, right->latch); // cache copy of key to update @@ -1536,11 +1561,11 @@ BtKey *ptr; right->latch->dirty = 1; right->page->kill = 1; - bt_lockpage (BtLockParent, right->latch); - bt_unlockpage (BtLockWrite, right->latch); + bt_lockpage (bt, BtLockParent, right->latch); + bt_unlockpage (bt, BtLockWrite, right->latch); - bt_lockpage (BtLockParent, set->latch); - bt_unlockpage (BtLockWrite, set->latch); + bt_lockpage (bt, BtLockParent, set->latch); + bt_unlockpage (bt, BtLockWrite, set->latch); // redirect higher key directly to our new node contents @@ -1559,14 +1584,13 @@ BtKey *ptr; // obtain delete and write locks to right node - bt_unlockpage (BtLockParent, right->latch); - bt_lockpage (BtLockDelete, right->latch); - bt_lockpage (BtLockWrite, right->latch); + bt_unlockpage (bt, BtLockParent, right->latch); + bt_lockpage (bt, BtLockDelete, right->latch); + bt_lockpage (bt, BtLockWrite, right->latch); bt_freepage (bt, right); - bt_unlockpage (BtLockParent, set->latch); + bt_unlockpage (bt, BtLockParent, set->latch); bt_unpinlatch (set->latch); - bt->found = 1; return 0; } @@ -1617,7 +1641,7 @@ BtVal *val; if( bt_fixfence (bt, set, lvl) ) return bt->err; else - return bt->found = found, 0; + return 0; // do we need to collapse root? @@ -1625,7 +1649,7 @@ BtVal *val; if( bt_collapseroot (bt, set) ) return bt->err; else - return bt->found = found, 0; + return 0; // delete empty page @@ -1633,9 +1657,9 @@ BtVal *val; return bt_deletepage (bt, set); set->latch->dirty = 1; - bt_unlockpage(BtLockWrite, set->latch); + bt_unlockpage(bt, BtLockWrite, set->latch); bt_unpinlatch (set->latch); - return bt->found = found, 0; + return 0; } BtKey *bt_foundkey (BtDb *bt) @@ -1665,13 +1689,13 @@ uid page_no; // obtain access lock using lock chaining with Access mode - bt_lockpage(BtLockAccess, set->latch); + bt_lockpage(bt, BtLockAccess, set->latch); - bt_unlockpage(BtLockRead, prevlatch); + bt_unlockpage(bt, BtLockRead, prevlatch); bt_unpinlatch (prevlatch); - bt_lockpage(BtLockRead, set->latch); - bt_unlockpage(BtLockAccess, set->latch); + bt_lockpage(bt, BtLockRead, set->latch); + bt_unlockpage(bt, BtLockAccess, set->latch); return 1; } @@ -1729,7 +1753,7 @@ BtVal *val; } while( slot = bt_findnext (bt, set, slot) ); - bt_unlockpage (BtLockRead, set->latch); + bt_unlockpage (bt, BtLockRead, set->latch); bt_unpinlatch (set->latch); return ret; } @@ -1774,7 +1798,9 @@ BtVal *val; while( cnt++ < max ) { if( cnt == slot ) newslot = idx + 2; - if( cnt < max && slotptr(bt->frame,cnt)->dead ) + + if( cnt < max || bt->frame->lvl ) + if( slotptr(bt->frame,cnt)->dead ) continue; // copy the value across @@ -1791,11 +1817,9 @@ BtVal *val; // make a librarian slot - if( idx ) { - slotptr(page, ++idx)->off = nxt; - slotptr(page, idx)->type = Librarian; - slotptr(page, idx)->dead = 1; - } + slotptr(page, ++idx)->off = nxt; + slotptr(page, idx)->type = Librarian; + slotptr(page, idx)->dead = 1; // set up the slot @@ -1885,7 +1909,7 @@ BtVal *val; // release and unpin root pages - bt_unlockpage(BtLockWrite, root->latch); + bt_unlockpage(bt, BtLockWrite, root->latch); bt_unpinlatch (root->latch); bt_unpinlatch (right); @@ -1915,8 +1939,10 @@ uint prev; idx = 0; while( cnt++ < max ) { - if( slotptr(set->page, cnt)->dead && cnt < max ) + if( cnt < max || set->page->lvl ) + if( slotptr(set->page, cnt)->dead ) continue; + src = valptr(set->page, cnt); nxt -= src->len + sizeof(BtVal); memcpy ((unsigned char *)bt->frame + nxt, src, src->len + sizeof(BtVal)); @@ -1928,11 +1954,9 @@ uint prev; // add librarian slot - if( idx ) { - slotptr(bt->frame, ++idx)->off = nxt; - slotptr(bt->frame, idx)->type = Librarian; - slotptr(bt->frame, idx)->dead = 1; - } + slotptr(bt->frame, ++idx)->off = nxt; + slotptr(bt->frame, idx)->type = Librarian; + slotptr(bt->frame, idx)->dead = 1; // add actual slot @@ -1987,11 +2011,9 @@ uint prev; // add librarian slot - if( idx ) { - slotptr(set->page, ++idx)->off = nxt; - slotptr(set->page, idx)->type = Librarian; - slotptr(set->page, idx)->dead = 1; - } + slotptr(set->page, ++idx)->off = nxt; + slotptr(set->page, idx)->type = Librarian; + slotptr(set->page, idx)->dead = 1; // add actual slot @@ -2034,10 +2056,10 @@ BtKey *ptr; // insert new fences in their parent pages - bt_lockpage (BtLockParent, right); + bt_lockpage (bt, BtLockParent, right); - bt_lockpage (BtLockParent, set->latch); - bt_unlockpage (BtLockWrite, set->latch); + bt_lockpage (bt, BtLockParent, set->latch); + bt_unlockpage (bt, BtLockWrite, set->latch); // insert new fence for reformulated left block of smaller keys @@ -2055,10 +2077,10 @@ BtKey *ptr; if( bt_insertkey (bt, ptr->key, ptr->len, lvl+1, value, BtId, 1) ) return bt->err; - bt_unlockpage (BtLockParent, set->latch); + bt_unlockpage (bt, BtLockParent, set->latch); bt_unpinlatch (set->latch); - bt_unlockpage (BtLockParent, right); + bt_unlockpage (bt, BtLockParent, right); bt_unpinlatch (right); return 0; } @@ -2131,7 +2153,7 @@ BtVal *val; node->dead = 0; if( release ) { - bt_unlockpage (BtLockWrite, set->latch); + bt_unlockpage (bt, BtLockWrite, set->latch); bt_unpinlatch (set->latch); } @@ -2216,7 +2238,7 @@ uint type; slotptr(set->page, slot)->dead = 0; val->len = vallen; memcpy (val->value, value, vallen); - bt_unlockpage(BtLockWrite, set->latch); + bt_unlockpage(bt, BtLockWrite, set->latch); bt_unpinlatch (set->latch); return 0; } @@ -2250,7 +2272,7 @@ uint type; ptr->len = keylen; slotptr(set->page, slot)->off = set->page->min; - bt_unlockpage(BtLockWrite, set->latch); + bt_unlockpage(bt, BtLockWrite, set->latch); bt_unpinlatch (set->latch); return 0; } @@ -2261,7 +2283,7 @@ typedef struct { uint entry; // latch table entry number uint slot:31; // page slot number uint reuse:1; // reused previous page -} AtomicMod; +} AtomicTxn; typedef struct { uid page_no; // page number for split leaf @@ -2272,83 +2294,10 @@ typedef struct { unsigned char leafkey[BT_keyarray]; } AtomicKey; -// find and load leaf page for given key -// leave page Atomic locked and Read locked. - -int bt_atomicload (BtDb *bt, BtPageSet *set, unsigned char *key, uint len) -{ -BtLatchSet *prevlatch; -uid page_no; -uint slot; - - // find level zero page - - if( !(slot = bt_loadpage (bt, set, key, len, 1, BtLockRead)) ) - return 0; - - // find next non-dead entry on this page - // it will be the fence key if nothing else - - while( slotptr(set->page, slot)->dead ) - if( slot++ < set->page->cnt ) - continue; - else - return bt->err = BTERR_struct, 0; - - page_no = bt_getid(valptr(set->page, slot)->value); - prevlatch = set->latch; - - while( page_no ) { - if( set->latch = bt_pinlatch (bt, page_no, 1) ) - set->page = bt_mappage (bt, set->latch); - else - return 0; - - // obtain read lock using lock chaining with Access mode - // release & unpin parent/left sibling page - - bt_lockpage(BtLockAccess, set->latch); - - bt_unlockpage(BtLockRead, prevlatch); - bt_unpinlatch (prevlatch); - - bt_lockpage(BtLockRead, set->latch); - bt_unlockpage(BtLockAccess, set->latch); - - // find key on page at this level - // and descend to requested level - - if( !set->page->kill ) - if( !bt_getid (set->page->right) || keycmp (keyptr(set->page, set->page->cnt), key, len) >= 0 ) { - bt_unlockpage(BtLockRead, set->latch); - bt_lockpage(BtLockAccess, set->latch); - bt_lockpage(BtLockAtomic, set->latch); - bt_lockpage(BtLockRead, set->latch); - bt_unlockpage(BtLockAccess, set->latch); - - if( !set->page->kill ) - if( slot = bt_findslot (set->page, key, len) ) - return slot; - - bt_unlockpage(BtLockAtomic, set->latch); - } - - // slide right into next page - - page_no = bt_getid(set->page->right); - prevlatch = set->latch; - } - - // return error on end of right chain - - bt->err = BTERR_struct; - return 0; // return error -} - // determine actual page where key is located // return slot number -uint bt_atomicpage (BtDb *bt, BtPage source, AtomicMod *locks, uint src, BtPageSet *set) +uint bt_atomicpage (BtDb *bt, BtPage source, AtomicTxn *locks, uint src, BtPageSet *set) { BtKey *key = keyptr(source,src); uint slot = locks[src].slot; @@ -2365,15 +2314,18 @@ uint entry; return slot; } - // is locks->reuse set? - // if so, find where our key - // is located on previous page or split pages + // is locks->reuse set? or was slot zeroed? + // if so, find where our key is located + // on current page or pages split on + // same page txn operations. do { set->latch = bt->mgr->latchsets + entry; set->page = bt_mappage (bt, set->latch); if( slot = bt_findslot(set->page, key->key, key->len) ) { + if( slotptr(set->page, slot)->type == Librarian ) + slot++; if( locks[src].reuse ) locks[src].entry = entry; return slot; @@ -2384,17 +2336,17 @@ uint entry; return 0; } -BTERR bt_atomicinsert (BtDb *bt, BtPage source, AtomicMod *locks, uint src) +BTERR bt_atomicinsert (BtDb *bt, BtPage source, AtomicTxn *locks, uint src) { BtKey *key = keyptr(source, src); BtVal *val = valptr(source, src); BtLatchSet *latch; BtPageSet set[1]; -uint entry; +uint entry, slot; - while( locks[src].slot = bt_atomicpage (bt, source, locks, src, set) ) { - if( locks[src].slot = bt_cleanpage(bt, set, key->len, locks[src].slot, val->len) ) - return bt_insertslot (bt, set, locks[src].slot, key->key, key->len, val->value, val->len, slotptr(source,src)->type, 0); + while( slot = bt_atomicpage (bt, source, locks, src, set) ) { + if( slot = bt_cleanpage(bt, set, key->len, slot, val->len) ) + return bt_insertslot (bt, set, slot, key->key, key->len, val->value, val->len, slotptr(source,src)->type, 0); if( entry = bt_splitpage (bt, set) ) latch = bt->mgr->latchsets + entry; @@ -2404,15 +2356,16 @@ uint entry; // splice right page into split chain // and WriteLock it. + bt_lockpage(bt, BtLockWrite, latch); latch->split = set->latch->split; set->latch->split = entry; - bt_lockpage(BtLockWrite, latch); + locks[src].slot = 0; } return bt->err = BTERR_atomic; } -BTERR bt_atomicdelete (BtDb *bt, BtPage source, AtomicMod *locks, uint src) +BTERR bt_atomicdelete (BtDb *bt, BtPage source, AtomicTxn *locks, uint src) { BtKey *key = keyptr(source, src); uint idx, entry, slot; @@ -2421,16 +2374,23 @@ BtKey *ptr; BtVal *val; if( slot = bt_atomicpage (bt, source, locks, src, set) ) + ptr = keyptr(set->page, slot); + else + return bt->err = BTERR_struct; + + if( !keycmp (ptr, key->key, key->len) ) + if( !slotptr(set->page, slot)->dead ) slotptr(set->page, slot)->dead = 1; + else + return 0; else - return bt->err = BTERR_struct; + return 0; - ptr = keyptr(set->page, slot); val = valptr(set->page, slot); - set->page->garbage += ptr->len + val->len + sizeof(BtKey) + sizeof(BtVal); set->latch->dirty = 1; set->page->act--; + bt->found++; return 0; } @@ -2448,7 +2408,7 @@ unsigned char value[BtId]; uid right_page_no; BtKey *ptr; - bt_lockpage(BtLockWrite, prev->latch); + bt_lockpage(bt, BtLockWrite, prev->latch); // grab the right sibling @@ -2457,8 +2417,8 @@ BtKey *ptr; else return bt->err; - bt_lockpage(BtLockAtomic, right->latch); - bt_lockpage(BtLockWrite, right->latch); + bt_lockpage(bt, BtLockAtomic, right->latch); + bt_lockpage(bt, BtLockWrite, right->latch); // and pull contents over empty page // while preserving master's left link @@ -2485,8 +2445,8 @@ BtKey *ptr; // now that master page is in good shape we can // remove its locks. - bt_unlockpage (BtLockAtomic, prev->latch); - bt_unlockpage (BtLockWrite, prev->latch); + bt_unlockpage (bt, BtLockAtomic, prev->latch); + bt_unlockpage (bt, BtLockWrite, prev->latch); // fix master's right sibling's left pointer // to remove scanner's poiner to the right page @@ -2495,11 +2455,11 @@ BtKey *ptr; if( temp->latch = bt_pinlatch (bt, right_page_no, 1) ) temp->page = bt_mappage (bt, temp->latch); - bt_lockpage (BtLockWrite, temp->latch); + bt_lockpage (bt, BtLockWrite, temp->latch); bt_putid (temp->page->left, prev->latch->page_no); temp->latch->dirty = 1; - bt_unlockpage (BtLockWrite, temp->latch); + bt_unlockpage (bt, BtLockWrite, temp->latch); bt_unpinlatch (temp->latch); } else { // master is now the far right page bt_spinwritelock (bt->mgr->lock); @@ -2510,10 +2470,10 @@ BtKey *ptr; // now that there are no pointers to the right page // we can delete it after the last read access occurs - bt_unlockpage (BtLockWrite, right->latch); - bt_unlockpage (BtLockAtomic, right->latch); - bt_lockpage (BtLockDelete, right->latch); - bt_lockpage (BtLockWrite, right->latch); + bt_unlockpage (bt, BtLockWrite, right->latch); + bt_unlockpage (bt, BtLockAtomic, right->latch); + bt_lockpage (bt, BtLockDelete, right->latch); + bt_lockpage (bt, BtLockWrite, right->latch); bt_freepage (bt, right); return 0; } @@ -2533,7 +2493,7 @@ BtPageSet set[1], prev[1]; unsigned char value[BtId]; BtKey *key, *ptr, *key2; BtLatchSet *latch; -AtomicMod *locks; +AtomicTxn *locks; int result = 0; BtSlot temp[1]; BtPage page; @@ -2541,7 +2501,7 @@ BtVal *val; uid right; int type; - locks = calloc (source->cnt + 1, sizeof(AtomicMod)); + locks = calloc (source->cnt + 1, sizeof(AtomicTxn)); head = NULL; tail = NULL; @@ -2577,11 +2537,11 @@ int type; if( samepage = src > 1 ) if( samepage = !bt_getid(set->page->right) || keycmp (keyptr(set->page, set->page->cnt), key->key, key->len) >= 0 ) slot = bt_findslot(set->page, key->key, key->len); - else // release read on previous page - bt_unlockpage(BtLockRead, set->latch); + else + bt_unlockpage(bt, BtLockRead, set->latch); if( !slot ) - if( slot = bt_atomicload(bt, set, key->key, key->len) ) + if( slot = bt_loadpage(bt, set, key->key, key->len, 0, BtLockRead | BtLockAtomic) ) set->latch->split = 0; else return -1; @@ -2610,13 +2570,13 @@ int type; // return constraint violation if key already exists - bt_unlockpage(BtLockRead, set->latch); + bt_unlockpage(bt, BtLockRead, set->latch); result = src; while( src ) { if( locks[src].entry ) { set->latch = bt->mgr->latchsets + locks[src].entry; - bt_unlockpage(BtLockAtomic, set->latch); + bt_unlockpage(bt, BtLockAtomic, set->latch); bt_unpinlatch (set->latch); } src--; @@ -2630,8 +2590,8 @@ int type; // unlock last loadpage lock - if( source->cnt > 1 ) - bt_unlockpage(BtLockRead, set->latch); + if( source->cnt ) + bt_unlockpage(bt, BtLockRead, set->latch); // obtain write lock for each master page @@ -2639,7 +2599,7 @@ int type; if( locks[src].reuse ) continue; else - bt_lockpage(BtLockWrite, bt->mgr->latchsets + locks[src].entry); + bt_lockpage(bt, BtLockWrite, bt->mgr->latchsets + locks[src].entry); // insert or delete each key // process any splits or merges @@ -2698,7 +2658,7 @@ int type; if( !prev->page->act ) { memcpy (set->page->left, prev->page->left, BtId); memcpy (prev->page, set->page, bt->mgr->page_size); - bt_lockpage (BtLockDelete, set->latch); + bt_lockpage (bt, BtLockDelete, set->latch); bt_freepage (bt, set); prev->latch->dirty = 1; @@ -2710,7 +2670,7 @@ int type; if( !set->page->act ) { memcpy (prev->page->right, set->page->right, BtId); prev->latch->split = set->latch->split; - bt_lockpage (BtLockDelete, set->latch); + bt_lockpage (bt, BtLockDelete, set->latch); bt_freepage (bt, set); continue; } @@ -2735,8 +2695,8 @@ int type; // splice in the left link into the split page bt_putid (set->page->left, prev->latch->page_no); - bt_lockpage(BtLockParent, prev->latch); - bt_unlockpage(BtLockWrite, prev->latch); + bt_lockpage(bt, BtLockParent, prev->latch); + bt_unlockpage(bt, BtLockWrite, prev->latch); *prev = *set; } @@ -2753,10 +2713,10 @@ int type; else return -1; - bt_lockpage (BtLockWrite, set->latch); + bt_lockpage (bt, BtLockWrite, set->latch); bt_putid (set->page->left, prev->latch->page_no); set->latch->dirty = 1; - bt_unlockpage (BtLockWrite, set->latch); + bt_unlockpage (bt, BtLockWrite, set->latch); bt_unpinlatch (set->latch); } else { // prev is rightmost page bt_spinwritelock (bt->mgr->lock); @@ -2781,20 +2741,20 @@ int type; tail = leaf; - bt_lockpage(BtLockParent, prev->latch); - bt_unlockpage(BtLockWrite, prev->latch); + bt_lockpage(bt, BtLockParent, prev->latch); + bt_unlockpage(bt, BtLockWrite, prev->latch); // remove atomic lock on master page - bt_unlockpage(BtLockAtomic, latch); + bt_unlockpage(bt, BtLockAtomic, latch); continue; } // finished if prev page occupied (either master or final split) if( prev->page->act ) { - bt_unlockpage(BtLockWrite, latch); - bt_unlockpage(BtLockAtomic, latch); + bt_unlockpage(bt, BtLockWrite, latch); + bt_unlockpage(bt, BtLockAtomic, latch); bt_unpinlatch(latch); continue; } @@ -2831,7 +2791,7 @@ int type; // leave atomic lock in place until // deletion completes in next phase. - bt_unlockpage(BtLockWrite, prev->latch); + bt_unlockpage(bt, BtLockWrite, prev->latch); } // add & delete keys for any pages split or merged during transaction @@ -2865,7 +2825,7 @@ int type; } if( !leaf->nounlock ) - bt_unlockpage (BtLockParent, set->latch); + bt_unlockpage (bt, BtLockParent, set->latch); bt_unpinlatch (set->latch); tail = leaf->next; @@ -2890,9 +2850,9 @@ BtPageSet set[1]; else return 0; - bt_lockpage(BtLockRead, set->latch); + bt_lockpage(bt, BtLockRead, set->latch); memcpy (bt->cursor, set->page, bt->mgr->page_size); - bt_unlockpage(BtLockRead, set->latch); + bt_unlockpage(bt, BtLockRead, set->latch); bt_unpinlatch (set->latch); bt->cursor_page = page_no; @@ -2923,9 +2883,9 @@ findourself: else return 0; - bt_lockpage(BtLockRead, set->latch); + bt_lockpage(bt, BtLockRead, set->latch); memcpy (bt->cursor, set->page, bt->mgr->page_size); - bt_unlockpage(BtLockRead, set->latch); + bt_unlockpage(bt, BtLockRead, set->latch); bt_unpinlatch (set->latch); next = bt_getid (bt->cursor->right); @@ -2971,11 +2931,11 @@ uid right; else return 0; - bt_lockpage(BtLockRead, set->latch); + bt_lockpage(bt, BtLockRead, set->latch); memcpy (bt->cursor, set->page, bt->mgr->page_size); - bt_unlockpage(BtLockRead, set->latch); + bt_unlockpage(bt, BtLockRead, set->latch); bt_unpinlatch (set->latch); slot = 0; @@ -3000,7 +2960,7 @@ uint slot; bt->cursor_page = set->latch->page_no; - bt_unlockpage(BtLockRead, set->latch); + bt_unlockpage(bt, BtLockRead, set->latch); bt_unpinlatch (set->latch); return slot; } @@ -3284,7 +3244,7 @@ FILE *in; } else if( len < BT_maxkey ) key[len++] = ch; - fprintf(stderr, "finished %s for %d keys: %d reads %d writes\n", args->infile, line, bt->reads, bt->writes); + fprintf(stderr, "finished %s for %d keys: %d reads %d writes %d found\n", args->infile, line, bt->reads, bt->writes, bt->found); break; case 'w': @@ -3329,7 +3289,7 @@ FILE *in; set->page = bt_mappage (bt, set->latch); else fprintf(stderr, "unable to obtain latch"), exit(1); - bt_lockpage (BtLockRead, set->latch); + bt_lockpage (bt, BtLockRead, set->latch); next = bt_getid (set->page->right); for( slot = 0; slot++ < set->page->cnt; ) @@ -3348,7 +3308,7 @@ FILE *in; cnt++; } - bt_unlockpage (BtLockRead, set->latch); + bt_unlockpage (bt, BtLockRead, set->latch); bt_unpinlatch (set->latch); } while( page_no = next );