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

MAX_BUFFER_SIZE is now a runtime (pre-tinit) option; clean up some doxygen warnings

Browse source
This commit is contained in:
Sears Russell 2010-07-16 16:52:08 +00:00
parent e2ba421a53
commit 7a27e39e5b
10 changed files with 51 additions and 46 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
src/stasis/bufferManager/bufferHash.c
View file

@ -132,7 +132,7 @@ inline static int tryToWriteBackPage(stasis_buffer_manager_t *bm, pageid_t page)
inline static Page * getFreePage(stasis_buffer_manager_t *bm) {
stasis_buffer_hash_t * bh = bm->impl;
Page * ret;
if(bh->pageCount < MAX_BUFFER_SIZE) {
if(bh->pageCount < stasis_buffer_manager_size) {
ret = stasis_buffer_pool_malloc_page(bh->buffer_pool);
stasis_buffer_pool_free_page(bh->buffer_pool, ret,-1);
(*pagePendingPtr(ret)) = 0;
@ -219,7 +219,7 @@ static Page * bhGetCachedPage(stasis_buffer_manager_t* bm, int xid, const pageid
}
static Page * bhLoadPageImpl_helper(stasis_buffer_manager_t* bm, stasis_buffer_manager_handle_t* handle,
int xid, const pageid_t pageid, int uninitialized, pagetype_t type) {
int xid, const pageid_t pageid, int uninitialized, pagetype_t type) {
stasis_buffer_hash_t * bh = bm->impl;
DEBUG("load %lld (%d)\n", pageid, uninitialized);
@ -549,7 +549,7 @@ stasis_buffer_manager_t* stasis_buffer_manager_hash_open(stasis_page_handle_t *
bh->lru = lruFastInit(pageGetNode, pageSetNode, pagePinCountPtr, 0);
bh->cachedPages = LH_ENTRY(create)(MAX_BUFFER_SIZE);
bh->cachedPages = LH_ENTRY(create)(stasis_buffer_manager_size);
bh->pageCount = 0;

5
src/stasis/bufferManager/concurrentBufferManager.c
View file

@ -9,6 +9,7 @@
#include <stasis/replacementPolicy.h>
#include <stasis/bufferPool.h>
#include <stasis/pageHandle.h>
#include <stasis/flags.h>
#ifndef NO_CONCURRENT_LRU
#ifndef CONCURRENT_LRU
@ -356,9 +357,9 @@ stasis_buffer_manager_t* stasis_buffer_manager_concurrent_hash_open(stasis_page_
#else
ch->lru = replacementPolicyThreadsafeWrapperInit(lruFastInit(pageGetNode, pageSetNode, pagePinCountPtr, 0));
#endif
ch->ht = hashtable_init(MAX_BUFFER_SIZE * 4);
ch->ht = hashtable_init(stasis_buffer_manager_size * 4);
for(int i = 0; i < MAX_BUFFER_SIZE; i++) {
for(pageid_t i = 0; i < stasis_buffer_manager_size; i++) {
Page *p = stasis_buffer_pool_malloc_page(ch->buffer_pool);
stasis_buffer_pool_free_page(ch->buffer_pool, p,-1*i);
pageSetNode(p,0,0);

12
src/stasis/bufferManager/legacy/pageCache.c
View file

@ -6,6 +6,7 @@
writes, and locking.
*/
#include <stasis/common.h>
#include <stasis/flags.h>
#include <stasis/page.h>
#include <stasis/bufferManager.h>
@ -14,7 +15,7 @@
#include <assert.h>
#include <stdio.h>
static unsigned int bufferSize; /* < MAX_BUFFER_SIZE */
static pageid_t bufferSize;
static Page *repHead, *repMiddle, *repTail; /* replacement policy */
int cache_state;
@ -106,8 +107,7 @@ void cacheInsertPage (Page * ret) {
if(cache_state == FULL) {
middleInsert(ret);
} else {
if(bufferSize == MAX_BUFFER_SIZE/* - 1*/) { /* Set up page kick mechanism. */
int i;
if(bufferSize == stasis_buffer_manager_size/* - 1*/) { /* Set up page kick mechanism. */
Page *iter;
cache_state = FULL;
@ -120,9 +120,9 @@ void cacheInsertPage (Page * ret) {
* resulted in the best performance"
*/
repMiddle = repHead;
for( i = 0; i < MAX_BUFFER_SIZE / 3; i++ ) {
repMiddle->queue = 1;
repMiddle = repMiddle->next;
for(pageid_t i = 0; i < stasis_buffer_manager_size / 3; i++ ) {
repMiddle->queue = 1;
repMiddle = repMiddle->next;
}
for( iter = repMiddle; iter; iter = iter->next ) {

12
src/stasis/bufferPool.c
View file

@ -49,6 +49,8 @@ terms specified in this license.
*
*/
#include <stasis/common.h>
#include <stasis/flags.h>
#include <stasis/bufferPool.h>
#include <stasis/page.h>
#include <assert.h>
@ -70,7 +72,7 @@ stasis_buffer_pool_t* stasis_buffer_pool_init() {
#ifndef VALGRIND_MODE
byte * bufferSpace = calloc((MAX_BUFFER_SIZE + 2), PAGE_SIZE);
byte * bufferSpace = calloc((stasis_buffer_manager_size + 2), PAGE_SIZE);
assert(bufferSpace);
ret->addr_to_free = bufferSpace;
@ -83,9 +85,9 @@ stasis_buffer_pool_t* stasis_buffer_pool_init() {
// We need one dummy page for locking purposes,
// so this array has one extra page in it.
ret->pool = malloc(sizeof(ret->pool[0])*(MAX_BUFFER_SIZE+1));
ret->pool = malloc(sizeof(ret->pool[0])*(stasis_buffer_manager_size+1));
for(pageid_t i = 0; i < MAX_BUFFER_SIZE+1; i++) {
for(pageid_t i = 0; i < stasis_buffer_manager_size+1; i++) {
ret->pool[i].rwlatch = initlock();
ret->pool[i].loadlatch = initlock();
#ifndef VALGRIND_MODE
@ -100,7 +102,7 @@ stasis_buffer_pool_t* stasis_buffer_pool_init() {
}
void stasis_buffer_pool_deinit(stasis_buffer_pool_t * ret) {
for(pageid_t i = 0; i < MAX_BUFFER_SIZE+1; i++) {
for(pageid_t i = 0; i < stasis_buffer_manager_size+1; i++) {
deletelock(ret->pool[i].rwlatch);
deletelock(ret->pool[i].loadlatch);
#ifdef VALGRIND_MODE
@ -122,7 +124,7 @@ Page* stasis_buffer_pool_malloc_page(stasis_buffer_pool_t * ret) {
assert(!page->dirty);
(ret->nextPage)++;
/* There's a dummy page that we need to keep around, thus the +1 */
assert(ret->nextPage <= MAX_BUFFER_SIZE + 1);
assert(ret->nextPage <= stasis_buffer_manager_size + 1);
pthread_mutex_unlock(&ret->mut);

11
src/stasis/flags.c
View file

@ -17,6 +17,17 @@ stasis_buffer_manager_t * (*stasis_buffer_manager_factory)(stasis_log_t*, stasis
stasis_buffer_manager_t * (*stasis_buffer_manager_factory)(stasis_log_t*, stasis_dirty_page_table_t*) = stasis_buffer_manager_concurrent_hash_factory;
#endif
#ifdef STASIS_BUFFER_MANAGER_SIZE
pageid_t stasis_buffer_manager_size = STASIS_BUFFER_MANAGER_SIZE;
#else // STASIS_BUFFER_MANAGER_SIZE
#ifdef MAX_BUFFER_SIZE
pageid_t stasis_buffer_manager_size = MAX_BUFFER_SIZE;
#else // MAX_BUFFER_SIZE
pageid_t stasis_buffer_manager_size = 20029; // ~ 82MB
#endif // MAX_BUFFER_SIZE
#endif // STASIS_BUFFER_MANAGER_SIZE
#ifdef BUFFER_MANAGER_O_DIRECT
int bufferManagerO_DIRECT = BUFFER_MANAGER_O_DIRECT;
#else

2
stasis/bufferManager/legacy/pageCache.h
View file

@ -37,8 +37,6 @@ Page * cacheStalePage();
#define INITIAL 0
#define FULL 1
extern int cache_state;
#endif

11
stasis/constants.h
View file

@ -80,17 +80,6 @@ terms specified in this license.
#define PAGE_SIZE 4096
#ifndef MAX_BUFFER_SIZE
//#define MAX_BUFFER_SIZE 100003
#define MAX_BUFFER_SIZE 20029
//#define MAX_BUFFER_SIZE 10007
//#define MAX_BUFFER_SIZE 5003
//#define MAX_BUFFER_SIZE 2003
//#define MAX_BUFFER_SIZE 4006
/* #define MAX_BUFFER_SIZE 71 */
/*#define MAX_BUFFER_SIZE 7 */
#endif
#define BUFFER_MANAGER_HASH 1
#define BUFFER_MANAGER_MEM_ARRAY 2
#define BUFFER_MANAGER_DEPRECATED_HASH 3

4
stasis/flags.h
View file

@ -17,6 +17,8 @@
*/
extern stasis_buffer_manager_t* (*stasis_buffer_manager_factory)(stasis_log_t*, stasis_dirty_page_table_t*);
extern pageid_t stasis_buffer_manager_size;
/**
This determines which type of file handle the buffer manager will use.
@ -115,7 +117,7 @@ extern lsn_t stasis_log_file_write_buffer_size;
Set to 1 if segment based recovery is enabled. This disables some
optimizations that assume all operations are page based.
@TODO Stasis' segment implementation is a work in progress; therefore this is set to zero by default.
@todo Stasis' segment implementation is a work in progress; therefore this is set to zero by default.
*/
extern int stasis_segments_enabled;
#endif

4
stasis/transactional.h
View file

@ -763,8 +763,10 @@ int TgetTransactionFingerprint(int xid, stasis_transaction_fingerprint_t * fp);
/**
* Checks to see if a transaction is still active.
*
* @param xid The transaction id to be tested.
* @param fp The finger print of the transaction to be tested.
* @return true if the transaction is still running, false otherwise.
*
* @see TgetTransactionFingerprint()
*/
int TisActiveTransaction(stasis_transaction_fingerprint_t* fp);
/*

30
test/stasis/check_bufferManager.c
View file

@ -14,7 +14,7 @@
#ifdef LONG_TEST
#define THREAD_COUNT 100
#define NUM_PAGES ((MAX_BUFFER_SIZE * 3)/2) // Otherwise, we run out of disk cache, and it takes forever to complete...
#define NUM_PAGES ((stasis_buffer_manager_size * 3)/2) // Otherwise, we run out of disk cache, and it takes forever to complete...
#define PAGE_MULT 10 // This tells the system to only use every 10'th page, allowing us to quickly check >2 GB, >4 GB safeness.
#define READS_PER_THREAD (NUM_PAGES * 5)
@ -23,7 +23,7 @@
#else
#define THREAD_COUNT 25
#define NUM_PAGES ((MAX_BUFFER_SIZE * 3)/2)
#define NUM_PAGES ((stasis_buffer_manager_size * 3)/2)
#define PAGE_MULT 1000
#define READS_PER_THREAD (NUM_PAGES * 2)
@ -73,7 +73,7 @@ void * workerThread(void * p) {
int k = (int) (((double)NUM_PAGES)*rand()/(RAND_MAX+1.0));
Page * p;
if(! (i % (READS_PER_THREAD / 10)) ) {
printf("%d", i / (READS_PER_THREAD / 10)); fflush(NULL);
printf("%lld", i / ((unsigned long long)READS_PER_THREAD / 10)); fflush(NULL);
}
rid.page = PAGE_MULT * (k+1);
@ -109,7 +109,7 @@ void * workerThreadWriting(void * q) {
/* fflush(NULL); */
if(! (i % (RECORDS_PER_THREAD/10)) ) {
printf("A%d", i / (RECORDS_PER_THREAD/10)); fflush(NULL);
printf("A%lld", i / ((unsigned long long)RECORDS_PER_THREAD/10)); fflush(NULL);
}
@ -146,7 +146,7 @@ void * workerThreadWriting(void * q) {
releasePage(p);
if(! (i % (RECORDS_PER_THREAD/10)) ) {
printf("W%d", i / (RECORDS_PER_THREAD/10)); fflush(NULL);
printf("W%lld", i / ((unsigned long long)RECORDS_PER_THREAD/10)); fflush(NULL);
}
/* sched_yield(); */
@ -166,7 +166,7 @@ void * workerThreadWriting(void * q) {
releasePage(p);
if(! (i % (RECORDS_PER_THREAD/10))) {
printf("R%d", i / (RECORDS_PER_THREAD/10)); fflush(NULL);
printf("R%lld", i / ((unsigned long long)RECORDS_PER_THREAD/10)); fflush(NULL);
}
@ -256,9 +256,9 @@ START_TEST(pageThreadedWritersTest) {
Tdeinit();
}END_TEST
//#define PINNED_PAGE_COUNT (MAX_BUFFER_SIZE - 1)
#define PINNED_PAGE_COUNT ((MAX_BUFFER_SIZE - 1) / THREAD_COUNT)
#define MAX_PAGE_ID (MAX_BUFFER_SIZE * 2)
//#define PINNED_PAGE_COUNT (stasis_buffer_manager_size - 1)
#define PINNED_PAGE_COUNT ((stasis_buffer_manager_size - 1) / THREAD_COUNT)
#define MAX_PAGE_ID (stasis_buffer_manager_size * 2)
#ifdef LONG_TEST
#define BLIND_ITERS 100000
#else
@ -312,7 +312,7 @@ START_TEST(pageBlindThreadTest) {
printf("Spawning threads now.\n"); fflush(stdout);
printf("each thread pins %d pages (total: %d)\n", PINNED_PAGE_COUNT, PINNED_PAGE_COUNT*THREAD_COUNT);
printf("each thread pins %lld pages (total: %lld)\n", (unsigned long long)PINNED_PAGE_COUNT, (unsigned long long)PINNED_PAGE_COUNT*THREAD_COUNT);
pthread_t workers[THREAD_COUNT];
@ -332,15 +332,15 @@ static void stalePinTestImpl(stasis_buffer_manager_t * (*fact)(stasis_log_t*, st
Tinit();
Page * p[MAX_BUFFER_SIZE-1];
for(int i = 0; i < MAX_BUFFER_SIZE-2; i++) {
Page * p[stasis_buffer_manager_size-1];
for(int i = 0; i < stasis_buffer_manager_size-2; i++) {
p[i] = loadUninitializedPage(-1, i);
}
for(int i = 0; i < MAX_BUFFER_SIZE; i++) {
Page * foo = loadUninitializedPage(-1, i+MAX_BUFFER_SIZE);
for(int i = 0; i < stasis_buffer_manager_size; i++) {
Page * foo = loadUninitializedPage(-1, i+stasis_buffer_manager_size);
releasePage(foo);
}
for(int i = 0; i < MAX_BUFFER_SIZE-2; i++) {
for(int i = 0; i < stasis_buffer_manager_size-2; i++) {
releasePage(p[i]);
}
Tdeinit();