X-Git-Url: https://pd.if.org/git/?a=blobdiff_plain;f=threads2j.c;h=27212ef96602fb8a18bbd29d04117f34d209a106;hb=1a0821acfbaaf782e68983c15ac7adfc811d81bb;hp=ce186a60b2e52fb3c1150bfd03b081b5f02a6575;hpb=471ecef14bab98d48dad385a2d9c3d4f361aa6c6;p=btree diff --git a/threads2j.c b/threads2j.c index ce186a6..27212ef 100644 --- a/threads2j.c +++ b/threads2j.c @@ -101,15 +101,15 @@ enum { // share is count of read accessors // grant write lock when share == 0 -typedef struct { - volatile unsigned char mutex; // 1 = busy - volatile unsigned char write:1; // 1 = exclusive - volatile unsigned char readwait:1; // readers are waiting - volatile unsigned char writewait:1; // writers are waiting - volatile unsigned char filler:5; - volatile ushort share; // count of readers holding locks - volatile ushort rcnt; // count of waiting readers - volatile ushort wcnt; // count of waiting writers +volatile typedef struct { + unsigned char mutex[1]; // 1 = busy + unsigned char write:1; // 1 = exclusive + unsigned char readwait:1; // readers are waiting + unsigned char writewait:1; // writers are waiting + unsigned char filler:5; + ushort share; // count of readers holding locks + ushort rcnt; // count of waiting readers + ushort wcnt; // count of waiting writers } BtLatch; // Define the length of the page and key pointers @@ -153,10 +153,9 @@ typedef struct BtPage_ { uint min; // next key offset unsigned char bits:7; // page size in bits unsigned char free:1; // page is on free list - unsigned char lvl:5; // level of page + unsigned char lvl:6; // level of page unsigned char kill:1; // page is being deleted unsigned char dirty:1; // page has deleted keys - unsigned char posted:1; // page fence has posted unsigned char right[BtId]; // page number to right } *BtPage; @@ -385,7 +384,7 @@ uint prev; while( 1 ) { // obtain latch mutex - while( __sync_lock_test_and_set(&latch->mutex, 1) ) + while( __sync_lock_test_and_set(latch->mutex, 1) ) sched_yield(); if( decr ) @@ -398,7 +397,7 @@ uint prev; latch->readwait = 1; latch->rcnt++; prev = *(uint *)latch & ~1; - __sync_lock_release (&latch->mutex); + __sync_lock_release (latch->mutex); sys_futex( (uint *)latch, FUTEX_WAIT_BITSET | private, prev, NULL, NULL, QueRd ); decr = 1; continue; @@ -409,7 +408,7 @@ uint prev; latch->readwait = 0; latch->share++; - __sync_lock_release (&latch->mutex); + __sync_lock_release (latch->mutex); return; } } @@ -426,7 +425,7 @@ uint prev; while( 1 ) { // obtain latch mutex - while( __sync_lock_test_and_set(&latch->mutex, 1) ) + while( __sync_lock_test_and_set(latch->mutex, 1) ) sched_yield(); if( decr ) @@ -438,7 +437,7 @@ uint prev; latch->writewait = 1; latch->wcnt++; prev = *(uint *)latch & ~1; - __sync_lock_release (&latch->mutex); + __sync_lock_release (latch->mutex); sys_futex( (uint *)latch, FUTEX_WAIT_BITSET | private, prev, NULL, NULL, QueWr ); decr = 1; continue; @@ -451,7 +450,7 @@ uint prev; latch->writewait = 0; latch->write = 1; - __sync_lock_release (&latch->mutex); + __sync_lock_release (latch->mutex); return; } } @@ -468,7 +467,7 @@ int ans; // try for mutex, // abandon request if not taken - if( __sync_lock_test_and_set(&latch->mutex, 1) ) + if( __sync_lock_test_and_set(latch->mutex, 1) ) return 0; // see if write mode is available @@ -480,7 +479,7 @@ int ans; // release latch mutex - __sync_lock_release (&latch->mutex); + __sync_lock_release (latch->mutex); return ans; } @@ -493,7 +492,7 @@ void bt_spinreleasewrite(BtLatch *latch, int private) // obtain latch mutex - while( __sync_lock_test_and_set(&latch->mutex, 1) ) + while( __sync_lock_test_and_set(latch->mutex, 1) ) sched_yield(); latch->write = 0; @@ -510,7 +509,7 @@ void bt_spinreleasewrite(BtLatch *latch, int private) // release latch mutex wakexit: - __sync_lock_release (&latch->mutex); + __sync_lock_release (latch->mutex); } // decrement reader count @@ -522,19 +521,19 @@ void bt_spinreleaseread(BtLatch *latch, int private) // obtain latch mutex - while( __sync_lock_test_and_set(&latch->mutex, 1) ) + while( __sync_lock_test_and_set(latch->mutex, 1) ) sched_yield(); latch->share--; - // wake waiting writers + // wake one waiting writer if( !latch->share && latch->wcnt ) sys_futex( (uint *)latch, FUTEX_WAKE_BITSET | private, 1, NULL, NULL, QueWr ); // release latch mutex - __sync_lock_release (&latch->mutex); + __sync_lock_release (latch->mutex); } // link latch table entry into latch hash table @@ -1926,11 +1925,11 @@ uid left; BTERR bt_splitpage (BtDb *bt, BtPageSet *set) { uint cnt = 0, idx = 0, max, nxt = bt->mgr->page_size; -unsigned char fencekey[256]; +unsigned char fencekey[256], rightkey[256]; uint lvl = set->page->lvl; -uid right; -BtKey key; +BtPageSet right[1]; uint prev; +BtKey key; // split higher half of keys to bt->frame @@ -1951,6 +1950,10 @@ uint prev; slotptr(bt->frame, idx)->off = nxt; } + // remember existing fence key for new page to the right + + memcpy (rightkey, key, key->len + 1); + bt->frame->bits = bt->mgr->page_bits; bt->frame->min = nxt; bt->frame->cnt = idx; @@ -1963,7 +1966,7 @@ uint prev; // get new free page and write higher keys to it. - if( !(right = bt_newpage(bt, bt->frame)) ) + if( !(right->page_no = bt_newpage(bt, bt->frame)) ) return bt->err; // update lower keys to continue in old page @@ -1971,7 +1974,6 @@ uint prev; memcpy (bt->frame, set->page, bt->mgr->page_size); memset (set->page+1, 0, bt->mgr->page_size - sizeof(*set->page)); nxt = bt->mgr->page_size; - set->page->posted = 0; set->page->dirty = 0; set->page->act = 0; cnt = 0; @@ -1992,65 +1994,42 @@ uint prev; // remember fence key for smaller page memcpy(fencekey, key, key->len + 1); - bt_putid(set->page->right, right); + + bt_putid(set->page->right, right->page_no); set->page->min = nxt; set->page->cnt = idx; // if current page is the root page, split it if( set->page_no == ROOT_page ) - return bt_splitroot (bt, set, fencekey, right); - - bt_unlockpage (BtLockWrite, set->latch); - right = 0; + return bt_splitroot (bt, set, fencekey, right->page_no); // insert new fences in their parent pages - while( 1 ) { - bt_lockpage (BtLockParent, set->latch); - bt_lockpage (BtLockWrite, set->latch); - - key = keyptr (set->page, set->page->cnt); - memcpy (fencekey, key, key->len + 1); - prev = set->page->posted; - - if( right && prev ) { - bt_unlockpage (BtLockParent, set->latch); - bt_unlockpage (BtLockWrite, set->latch); - bt_unpinlatch (set->latch); - bt_unpinpool (set->pool); - return 0; - } - - right = bt_getid (set->page->right); - set->page->posted = 1; + right->latch = bt_pinlatch (bt, right->page_no); + bt_lockpage (BtLockParent, right->latch); - bt_unlockpage (BtLockWrite, set->latch); + bt_lockpage (BtLockParent, set->latch); + bt_unlockpage (BtLockWrite, set->latch); - // insert new fence for reformulated left block of smaller keys + // insert new fence for reformulated left block of smaller keys - if( !prev ) - if( bt_insertkey (bt, fencekey+1, *fencekey, lvl+1, set->page_no, time(NULL)) ) - return bt->err; - - bt_unlockpage (BtLockParent, set->latch); - bt_unpinlatch (set->latch); - bt_unpinpool (set->pool); + if( bt_insertkey (bt, fencekey+1, *fencekey, lvl+1, set->page_no, time(NULL)) ) + return bt->err; - if( !(set->page_no = right) ) - break; + // switch fence for right block of larger keys to new right page - set->latch = bt_pinlatch (bt, right); + if( bt_insertkey (bt, rightkey+1, *rightkey, lvl+1, right->page_no, time(NULL)) ) + return bt->err; - if( set->pool = bt_pinpool (bt, right) ) - set->page = bt_page (bt, set->pool, right); - else - return bt->err; - } + bt_unlockpage (BtLockParent, set->latch); + bt_unpinlatch (set->latch); + bt_unpinpool (set->pool); + bt_unlockpage (BtLockParent, right->latch); + bt_unpinlatch (right->latch); return 0; } - // Insert new key into the btree at given level. BTERR bt_insertkey (BtDb *bt, unsigned char *key, uint len, uint lvl, uid id, uint tod)