greg/stasis-aries-wal
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

add casts for buffer manager, replacement policy

Browse source
This commit is contained in:
Rusty Sears 2013-01-30 16:37:01 -08:00
parent 814614b0bb
commit 9503882489
6 changed files with 49 additions and 49 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
src/stasis/bufferManager.c
View file

@ -156,12 +156,12 @@ void stasis_buffer_manager_set_redo_mode(int in_redo) {
Page * loadPage(int xid, pageid_t pageid) {
// This lock is released at Tcommit()
if(globalLockManager.readLockPage) { globalLockManager.readLockPage(xid, pageid); }
stasis_buffer_manager_t * bm = stasis_runtime_buffer_manager();
stasis_buffer_manager_t * bm = (stasis_buffer_manager_t *)stasis_runtime_buffer_manager();
return bm->loadPageImpl(bm, 0, xid, pageid, UNKNOWN_TYPE_PAGE);
}
Page * loadPageOfType(int xid, pageid_t pageid, pagetype_t type) {
if(globalLockManager.readLockPage) { globalLockManager.readLockPage(xid, pageid); }
stasis_buffer_manager_t * bm = stasis_runtime_buffer_manager();
stasis_buffer_manager_t * bm = (stasis_buffer_manager_t *)stasis_runtime_buffer_manager();
Page *p = bm->loadPageImpl(bm, 0, xid, pageid, type);
return p;
}
@ -169,7 +169,7 @@ Page * loadUninitializedPage(int xid, pageid_t pageid) {
// This lock is released at Tcommit()
if(globalLockManager.readLockPage) { globalLockManager.readLockPage(xid, pageid); }
stasis_buffer_manager_t * bm = stasis_runtime_buffer_manager();
stasis_buffer_manager_t * bm = (stasis_buffer_manager_t *)stasis_runtime_buffer_manager();
if(bm->in_redo) {
return bm->loadPageImpl(bm, 0, xid, pageid, UNKNOWN_TYPE_PAGE);
} else {
@ -197,19 +197,19 @@ Page * loadPageForOperation(int xid, pageid_t pageid, int op) { //, int is_recov
}
void prefetchPages(pageid_t pageid, pageid_t count) {
stasis_buffer_manager_t * bm = stasis_runtime_buffer_manager();
stasis_buffer_manager_t * bm = (stasis_buffer_manager_t *)stasis_runtime_buffer_manager();
// This is just a performance hint; and is an optional method.
if(bm->prefetchPages != NULL) { bm->prefetchPages(bm, pageid, count); }
}
int preallocatePages(pageid_t pageid, pageid_t count) {
stasis_buffer_manager_t * bm = stasis_runtime_buffer_manager();
stasis_buffer_manager_t * bm = (stasis_buffer_manager_t *)stasis_runtime_buffer_manager();
// This is just a performance hint; and is an optional method.
if(stasis_buffer_manager_preallocate_mode == STASIS_BUFFER_MANAGER_PREALLOCATE_LEGACY) {
// Legacy option (ext3): pre-fill the stasis buffer manager with dirty zeros.
for(pageid_t i = 0; i < count; i++) {
Page * p = loadUninitializedPage(-1, pageid+i);
stasis_dirty_page_table_set_dirty(stasis_runtime_dirty_page_table(), p);
stasis_dirty_page_table_set_dirty((stasis_dirty_page_table_t *)stasis_runtime_dirty_page_table(), p);
releasePage(p);
}
} else if(stasis_buffer_manager_preallocate_mode == STASIS_BUFFER_MANAGER_PREALLOCATE_DEFAULT) {
@ -224,10 +224,10 @@ int preallocatePages(pageid_t pageid, pageid_t count) {
}
Page * getCachedPage(int xid, pageid_t pageid) {
stasis_buffer_manager_t * bm = stasis_runtime_buffer_manager();
stasis_buffer_manager_t * bm = (stasis_buffer_manager_t *)stasis_runtime_buffer_manager();
return bm->getCachedPageImpl(bm, xid, pageid);
}
void releasePage(Page * p) {
stasis_buffer_manager_t * bm = stasis_runtime_buffer_manager();
stasis_buffer_manager_t * bm = (stasis_buffer_manager_t *)stasis_runtime_buffer_manager();
bm->releasePageImpl(bm, p);
}

14
src/stasis/replacementPolicy/clock.c
View file

@ -36,15 +36,15 @@ typedef struct {
} stasis_replacement_policy_clock_t;
static void clockDeinit (struct replacementPolicy* impl) {
stasis_replacement_policy_clock_t * clock = impl->impl;
stasis_replacement_policy_clock_t * clock = (stasis_replacement_policy_clock_t *)impl->impl;
free(clock);
free(impl);
}
static void clockHit (struct replacementPolicy* impl, Page* page) {
page->next = (void*)1;
page->next = (Page*)1;
}
static Page* clockGetStale(struct replacementPolicy* impl) {
stasis_replacement_policy_clock_t * clock = impl->impl;
stasis_replacement_policy_clock_t * clock = (stasis_replacement_policy_clock_t *)impl->impl;
// NOTE: This just exists for legacy purposes, and is fundamantally not
// threadsafe. So, this doesn't mess with the __sync* stuff, or worry
@ -64,8 +64,8 @@ static Page* clockGetStale(struct replacementPolicy* impl) {
//clock->pages[ptr].next = (void*)-1;
return &clock->pages[ptr];
} else if(clock->pages[ptr].next == (void*)1){
clock->pages[ptr].next = (void*)0;
} else if(clock->pages[ptr].next == (Page*)1){
clock->pages[ptr].next = (Page*)0;
}
}
return NULL;
@ -79,7 +79,7 @@ static Page* clockRemove (struct replacementPolicy* impl, Page* page) {
}
}
static Page* clockGetStaleAndRemove (struct replacementPolicy* impl) {
stasis_replacement_policy_clock_t * clock = impl->impl;
stasis_replacement_policy_clock_t * clock = (stasis_replacement_policy_clock_t *)impl->impl;
for(uint64_t spin = 0; spin < 2*clock->page_count; spin++) {
uint64_t ptr = __sync_fetch_and_add(&clock->ptr,1) % clock->page_count;
@ -110,7 +110,7 @@ static Page* clockGetStaleAndRemove (struct replacementPolicy* impl) {
}
static void clockInsert (struct replacementPolicy* impl, Page* page) {
__sync_fetch_and_sub(&page->pinCount,1); // don't care about ordering of this line and next. pinCount is just a "message to ourselves"
page->next = (void*)1;
page->next = (Page*)1;
__sync_synchronize();
}

16
src/stasis/replacementPolicy/concurrentWrapper.c
View file

@ -35,7 +35,7 @@ static inline unsigned int bucket_hash(stasis_replacement_policy_concurrent_wrap
}
static void cwDeinit (struct replacementPolicy* impl) {
stasis_replacement_policy_concurrent_wrapper_t * rp = impl->impl;
stasis_replacement_policy_concurrent_wrapper_t * rp = (stasis_replacement_policy_concurrent_wrapper_t *)impl->impl;
for(int i = 0; i < rp->num_buckets; i++) {
rp->impl[i]->deinit(rp->impl[i]);
pthread_mutex_destroy(&rp->mut[i]);
@ -47,14 +47,14 @@ static void cwDeinit (struct replacementPolicy* impl) {
free(impl);
}
static void cwHit (struct replacementPolicy* impl, Page* page) {
stasis_replacement_policy_concurrent_wrapper_t * rp = impl->impl;
stasis_replacement_policy_concurrent_wrapper_t * rp = (stasis_replacement_policy_concurrent_wrapper_t *)impl->impl;
unsigned int bucket = bucket_hash(rp, page);
pthread_mutex_lock(&rp->mut[bucket]);
rp->impl[bucket]->hit(rp->impl[bucket], page);
pthread_mutex_unlock(&rp->mut[bucket]);
}
static Page* cwGetStaleHelper(struct replacementPolicy* impl, Page*(*func)(struct replacementPolicy*)) {
stasis_replacement_policy_concurrent_wrapper_t * rp = impl->impl;
stasis_replacement_policy_concurrent_wrapper_t * rp = (stasis_replacement_policy_concurrent_wrapper_t *)impl->impl;
intptr_t bucket = (intptr_t)pthread_getspecific(rp->next_bucket);
intptr_t oldbucket = bucket;
Page*ret = 0;
@ -120,23 +120,23 @@ static Page* cwGetStaleHelper(struct replacementPolicy* impl, Page*(*func)(struc
return ret;
}
static Page* cwGetStale(struct replacementPolicy* impl) {
stasis_replacement_policy_concurrent_wrapper_t * rp = impl->impl;
stasis_replacement_policy_concurrent_wrapper_t * rp = (stasis_replacement_policy_concurrent_wrapper_t *)impl->impl;
return cwGetStaleHelper(impl, rp->impl[0]->getStale);
}
static Page* cwGetStaleAndRemove(struct replacementPolicy* impl) {
stasis_replacement_policy_concurrent_wrapper_t * rp = impl->impl;
stasis_replacement_policy_concurrent_wrapper_t * rp = (stasis_replacement_policy_concurrent_wrapper_t *)impl->impl;
return cwGetStaleHelper(impl, rp->impl[0]->getStaleAndRemove);
}
static Page* cwRemove (struct replacementPolicy* impl, Page* page) {
stasis_replacement_policy_concurrent_wrapper_t * rp = impl->impl;
stasis_replacement_policy_concurrent_wrapper_t * rp = (stasis_replacement_policy_concurrent_wrapper_t *)impl->impl;
unsigned int bucket = bucket_hash(rp, page);
pthread_mutex_lock(&rp->mut[bucket]);
void *ret = rp->impl[bucket]->remove(rp->impl[bucket], page);
pthread_mutex_unlock(&rp->mut[bucket]);
return ret;
return (Page*)ret;
}
static void cwInsert (struct replacementPolicy* impl, Page* page) {
stasis_replacement_policy_concurrent_wrapper_t * rp = impl->impl;
stasis_replacement_policy_concurrent_wrapper_t * rp = (stasis_replacement_policy_concurrent_wrapper_t *)impl->impl;
unsigned int bucket = bucket_hash(rp, page);
pthread_mutex_lock(&rp->mut[bucket]);
rp->impl[bucket]->insert(rp->impl[bucket], page);

24
src/stasis/replacementPolicy/lru.c
View file

@ -23,8 +23,8 @@ typedef struct stasis_replacement_policy_lru_t {
} stasis_replacement_policy_lru_t;
static int stasis_replacement_policy_lru_entry_cmp(const void * ap, const void * bp, const void * ignored) {
const stasis_replacement_policy_lru_entry * a = ap;
const stasis_replacement_policy_lru_entry * b = bp;
const stasis_replacement_policy_lru_entry * a = (const stasis_replacement_policy_lru_entry *)ap;
const stasis_replacement_policy_lru_entry * b = (const stasis_replacement_policy_lru_entry *)bp;
return a->clock < b->clock ? -1
: a->clock == b->clock ? 0
@ -33,7 +33,7 @@ static int stasis_replacement_policy_lru_entry_cmp(const void * ap, const void *
static void stasis_replacement_policy_lru_deinit(replacementPolicy *r) {
//XXX free other stuff
stasis_replacement_policy_lru_t * l = r->impl;
stasis_replacement_policy_lru_t * l = (stasis_replacement_policy_lru_t*)r->impl;
LH_ENTRY(destroy)(l->hash);
RB_ENTRY(destroy)(l->lru);
@ -44,8 +44,8 @@ static void stasis_replacement_policy_lru_deinit(replacementPolicy *r) {
/** @todo handle clock wraps properly! */
static void stasis_replacement_policy_lru_hit(replacementPolicy *r, Page *p) {
stasis_replacement_policy_lru_t * l = r->impl;
stasis_replacement_policy_lru_entry * e = l->getNode(p, l->conf);
stasis_replacement_policy_lru_t * l = (stasis_replacement_policy_lru_t *)r->impl;
stasis_replacement_policy_lru_entry * e = (stasis_replacement_policy_lru_entry *)l->getNode(p, l->conf);
assert(e);
stasis_replacement_policy_lru_entry * old = (stasis_replacement_policy_lru_entry * ) RB_ENTRY(delete)(e, l->lru);
assert(e == old);
@ -55,28 +55,28 @@ static void stasis_replacement_policy_lru_hit(replacementPolicy *r, Page *p) {
assert(e == old);
}
static Page* stasis_replacement_policy_lru_get_stale(replacementPolicy* r) {
stasis_replacement_policy_lru_t * l = r->impl;
stasis_replacement_policy_lru_t * l = (stasis_replacement_policy_lru_t *)r->impl;
stasis_replacement_policy_lru_entry * e = (stasis_replacement_policy_lru_entry * ) rbmin(l->lru);
return e ? e->value : 0;
return (Page*)(e ? e->value : 0);
}
static Page* stasis_replacement_policy_lru_remove(replacementPolicy* r, Page *p) {
stasis_replacement_policy_lru_t * l = r->impl;
stasis_replacement_policy_lru_entry * e = l->getNode(p, l->conf);
stasis_replacement_policy_lru_t * l = (stasis_replacement_policy_lru_t *)r->impl;
stasis_replacement_policy_lru_entry * e = (stasis_replacement_policy_lru_entry *)l->getNode(p, l->conf);
assert(e);
stasis_replacement_policy_lru_entry * old = (stasis_replacement_policy_lru_entry *) RB_ENTRY(delete)(e, l->lru);
assert(old == e);
void * ret = e->value;
Page* ret = (Page*)e->value;
free(e);
return ret;
}
static Page* stasis_replacement_policy_lru_get_stale_and_remove(replacementPolicy* r) {
void* ret = stasis_replacement_policy_lru_get_stale(r);
Page* ret = (Page*)stasis_replacement_policy_lru_get_stale(r);
stasis_replacement_policy_lru_remove(r, ret);
return ret;
}
static void stasis_replacement_policy_lru_insert(replacementPolicy* r, Page* p) {
stasis_replacement_policy_lru_t * l = r->impl;
stasis_replacement_policy_lru_t * l = (stasis_replacement_policy_lru_t *)r->impl;
stasis_replacement_policy_lru_entry * e = stasis_alloc(stasis_replacement_policy_lru_entry);
e->value = p;
e->clock = l->now;

10
src/stasis/replacementPolicy/lruFast.c
View file

@ -44,7 +44,7 @@ static inline Page* llShift( List *list ) {
return ret;
}
static void stasis_lru_fast_hit(struct replacementPolicy * r, Page *p) {
lruFast *l = r->impl;
lruFast *l = (lruFast*)r->impl;
if( p->prev == NULL ) {
// ignore attempts to hit pages not in lru
return;
@ -53,7 +53,7 @@ static void stasis_lru_fast_hit(struct replacementPolicy * r, Page *p) {
llPush(&l->list, p);
}
static Page* stasis_lru_fast_getStale(struct replacementPolicy *r) {
lruFast *l = r->impl;
lruFast *l = (lruFast*)r->impl;
return llHead(&l->list);
}
static Page* stasis_lru_fast_remove(struct replacementPolicy* r, Page *p) {
@ -72,7 +72,7 @@ static Page* stasis_lru_fast_remove(struct replacementPolicy* r, Page *p) {
return ret;
}
static Page* stasis_lru_fast_getStaleAndRemove(struct replacementPolicy *r) {
lruFast * l = r->impl;
lruFast * l = (lruFast*)r->impl;
Page *ret = llShift(&l->list);
if(ret) {
assert(!ret->pinCount);
@ -81,7 +81,7 @@ static Page* stasis_lru_fast_getStaleAndRemove(struct replacementPolicy *r) {
return ret;
}
static void stasis_lru_fast_insert(struct replacementPolicy *r, Page *p) {
lruFast * l = r->impl;
lruFast * l = (lruFast*)r->impl;
p->pinCount--;
assert(p->pinCount >= 0);
if(stasis_buffer_manager_hint_writes_are_sequential &&
@ -105,7 +105,7 @@ static void stasis_lru_fast_insert(struct replacementPolicy *r, Page *p) {
}
}
static void stasis_lru_fast_deinit(struct replacementPolicy * r) {
lruFast * l = r->impl;
lruFast * l = (lruFast*)r->impl;
free(l);
free(r);
}

18
src/stasis/replacementPolicy/threadsafeWrapper.c
View file

@ -13,39 +13,39 @@ typedef struct {
} stasis_replacement_policy_threadsafe_wrapper_t;
static void tsDeinit (struct replacementPolicy* impl) {
stasis_replacement_policy_threadsafe_wrapper_t * rp = impl->impl;
stasis_replacement_policy_threadsafe_wrapper_t * rp = (stasis_replacement_policy_threadsafe_wrapper_t *)impl->impl;
rp->impl->deinit(rp->impl);
free(impl);
}
static void tsHit (struct replacementPolicy* impl, Page* page) {
stasis_replacement_policy_threadsafe_wrapper_t * rp = impl->impl;
stasis_replacement_policy_threadsafe_wrapper_t * rp = (stasis_replacement_policy_threadsafe_wrapper_t *)impl->impl;
pthread_mutex_lock(&rp->mut);
rp->impl->hit(rp->impl, page);
pthread_mutex_unlock(&rp->mut);
}
static Page* tsGetStale(struct replacementPolicy* impl) {
stasis_replacement_policy_threadsafe_wrapper_t * rp = impl->impl;
stasis_replacement_policy_threadsafe_wrapper_t * rp = (stasis_replacement_policy_threadsafe_wrapper_t *)impl->impl;
pthread_mutex_lock(&rp->mut);
void *ret = rp->impl->getStale(rp->impl);
Page *ret = rp->impl->getStale(rp->impl);
pthread_mutex_unlock(&rp->mut);
return ret;
}
static Page* tsRemove (struct replacementPolicy* impl, Page* page) {
stasis_replacement_policy_threadsafe_wrapper_t * rp = impl->impl;
stasis_replacement_policy_threadsafe_wrapper_t * rp = (stasis_replacement_policy_threadsafe_wrapper_t *)impl->impl;
pthread_mutex_lock(&rp->mut);
void *ret = rp->impl->remove(rp->impl, page);
Page *ret = rp->impl->remove(rp->impl, page);
pthread_mutex_unlock(&rp->mut);
return ret;
}
static Page* tsGetStaleAndRemove (struct replacementPolicy* impl) {
stasis_replacement_policy_threadsafe_wrapper_t * rp = impl->impl;
stasis_replacement_policy_threadsafe_wrapper_t * rp = (stasis_replacement_policy_threadsafe_wrapper_t *)impl->impl;
pthread_mutex_lock(&rp->mut);
void *ret = rp->impl->getStaleAndRemove(rp->impl);
Page *ret = rp->impl->getStaleAndRemove(rp->impl);
pthread_mutex_unlock(&rp->mut);
return ret;
}
static void tsInsert (struct replacementPolicy* impl, Page* page) {
stasis_replacement_policy_threadsafe_wrapper_t * rp = impl->impl;
stasis_replacement_policy_threadsafe_wrapper_t * rp = (stasis_replacement_policy_threadsafe_wrapper_t *)impl->impl;
pthread_mutex_lock(&rp->mut);
rp->impl->insert(rp->impl, page);
pthread_mutex_unlock(&rp->mut);