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
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 {
#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;
}
// 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:
// 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:
// 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
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;
}
// 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;
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
// 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
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
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;
}
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;
} 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;
}
else
return 0;
- bt_lockpage (BtLockWrite, right->latch);
+ bt_lockpage (bt, BtLockWrite, right->latch);
// cache copy of key to update
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
// 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;
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;
}
// 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;
}
} while( slot = bt_findnext (bt, set, slot) );
- bt_unlockpage (BtLockRead, set->latch);
+ bt_unlockpage (bt, BtLockRead, set->latch);
bt_unpinlatch (set->latch);
return ret;
}
// release and unpin root pages
- bt_unlockpage(BtLockWrite, root->latch);
+ bt_unlockpage(bt, BtLockWrite, root->latch);
bt_unpinlatch (root->latch);
bt_unpinlatch (right);
// 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
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;
}
node->dead = 0;
if( release ) {
- bt_unlockpage (BtLockWrite, set->latch);
+ bt_unlockpage (bt, BtLockWrite, set->latch);
bt_unpinlatch (set->latch);
}
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;
}
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;
}
uid page_no;
uint slot;
- // find level zero page
+ // find level one slot
if( !(slot = bt_loadpage (bt, set, key, len, 1, BtLockRead)) )
return 0;
// obtain read lock using lock chaining with Access mode
// release & unpin parent/left sibling page
- 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);
// 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);
+ bt_unlockpage(bt, BtLockRead, set->latch);
+ bt_lockpage(bt, BtLockAtomic, set->latch);
+ bt_lockpage(bt, BtLockRead, set->latch);
+ bt_unlockpage(bt, BtLockAccess, set->latch);
if( !set->page->kill )
if( slot = bt_findslot (set->page, key, len) )
return slot;
- bt_unlockpage(BtLockAtomic, set->latch);
+ bt_unlockpage(bt, BtLockAtomic, set->latch);
}
// slide right into next page
latch->split = set->latch->split;
set->latch->split = entry;
- bt_lockpage(BtLockWrite, latch);
+ bt_lockpage(bt, BtLockWrite, latch);
}
return bt->err = BTERR_atomic;
uid right_page_no;
BtKey *ptr;
- bt_lockpage(BtLockWrite, prev->latch);
+ bt_lockpage(bt, BtLockWrite, prev->latch);
// grab the right sibling
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
// 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
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);
// 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;
}
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);
+ bt_unlockpage(bt, BtLockRead, set->latch);
if( !slot )
if( slot = bt_atomicload(bt, set, key->key, key->len) )
// 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--;
// unlock last loadpage lock
if( source->cnt > 1 )
- bt_unlockpage(BtLockRead, set->latch);
+ bt_unlockpage(bt, BtLockRead, set->latch);
// obtain write lock for each master page
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
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;
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;
}
// 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;
}
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);
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;
}
// 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
}
if( !leaf->nounlock )
- bt_unlockpage (BtLockParent, set->latch);
+ bt_unlockpage (bt, BtLockParent, set->latch);
bt_unpinlatch (set->latch);
tail = leaf->next;
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;
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);
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;
bt->cursor_page = set->latch->page_no;
- bt_unlockpage(BtLockRead, set->latch);
+ bt_unlockpage(bt, BtLockRead, set->latch);
bt_unpinlatch (set->latch);
return slot;
}
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; )
cnt++;
}
- bt_unlockpage (BtLockRead, set->latch);
+ bt_unlockpage (bt, BtLockRead, set->latch);
bt_unpinlatch (set->latch);
} while( page_no = next );