typedef struct {
ushort ticket[1];
ushort serving[1];
+ ushort tid;
+ ushort dup;
} WOLock;
#define PHID 0x1
// 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
#else
SwitchToThread ();
#endif
+ lock->tid = tid;
}
void WriteORelease (WOLock *lock)
{
+ if( lock->dup ) {
+ lock->dup--;
+ return;
+ }
+
+ lock->tid = 0;
lock->serving[0]++;
}
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;
}
}
case BtLockAtomic:
WriteORelease (latch->atomic);
break;
+ case BtLockAtomic | BtLockRead:
+ WriteORelease (latch->atomic);
+ ReadRelease (latch->readwr);
+ break;
}
}
bt_unlockpage (bt, BtLockParent, set->latch);
bt_unpinlatch (set->latch);
- bt->found = 1;
return 0;
}
if( bt_fixfence (bt, set, lvl) )
return bt->err;
else
- return bt->found = found, 0;
+ return 0;
// do we need to collapse root?
if( bt_collapseroot (bt, set) )
return bt->err;
else
- return bt->found = found, 0;
+ return 0;
// delete empty page
set->latch->dirty = 1;
bt_unlockpage(bt, BtLockWrite, set->latch);
bt_unpinlatch (set->latch);
- return bt->found = found, 0;
+ return 0;
}
BtKey *bt_foundkey (BtDb *bt)
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
// 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
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));
// 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
// 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
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
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 one slot
-
- 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(bt, BtLockAccess, set->latch);
-
- bt_unlockpage(bt, BtLockRead, prevlatch);
- bt_unpinlatch (prevlatch);
-
- 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(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(bt, 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;
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;
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;
// 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(bt, 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;
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;
}
unsigned char value[BtId];
BtKey *key, *ptr, *key2;
BtLatchSet *latch;
-AtomicMod *locks;
+AtomicTxn *locks;
int result = 0;
BtSlot temp[1];
BtPage page;
uid right;
int type;
- locks = calloc (source->cnt + 1, sizeof(AtomicMod));
+ locks = calloc (source->cnt + 1, sizeof(AtomicTxn));
head = NULL;
tail = NULL;
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
+ 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;
// unlock last loadpage lock
- if( source->cnt > 1 )
+ if( source->cnt )
bt_unlockpage(bt, BtLockRead, set->latch);
// obtain write lock for each master page
}
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':
projects / btree / blobdiff