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

remove more template stuff

Browse source 
git-svn-id: svn+ssh://svn.corp.yahoo.com/yahoo/yrl/labs/pnuts/code/logstore@2668 8dad8b1f-cf64-0410-95b6-bcf113ffbcfe
This commit is contained in:
sears 2011-06-09 01:16:55 +00:00
parent ef4d360b68
commit 6f82ae268a
22 changed files with 142 additions and 163 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
logserver.cpp
View file

@ -19,7 +19,7 @@
#include <fcntl.h>
void *serverLoop(void *args);
void logserver::startserver(logtable<datatuple> *ltable)
void logserver::startserver(logtable *ltable)
{
sys_alive = true;
this->ltable = ltable;

6
logserver.h
View file

@ -39,7 +39,7 @@ struct pthread_data {
int *workitem; //id of the socket to work
logtable<datatuple> *ltable;
logtable *ltable;
bool *sys_alive;
#ifdef STATS_ENABLED
@ -97,7 +97,7 @@ public:
delete qlock;
}
void startserver(logtable<datatuple> *ltable);
void startserver(logtable *ltable);
void stopserver();
@ -126,7 +126,7 @@ private:
int * self_pipe; // write a byte to self_pipe[1] to wake up select().
std::vector<pthread_item *> th_list; // list of threads
logtable<datatuple> *ltable;
logtable *ltable;
#ifdef STATS_ENABLED
int num_reqs;

76
logstore.cpp
View file

@ -22,8 +22,7 @@ static inline double tv_to_double(struct timeval tv)
// LOG TABLE IMPLEMENTATION
/////////////////////////////////////////////////////////////////
template<class TUPLE>
logtable<TUPLE>::logtable(int log_mode, pageid_t max_c0_size, pageid_t internal_region_size, pageid_t datapage_region_size, pageid_t datapage_size)
logtable::logtable(int log_mode, pageid_t max_c0_size, pageid_t internal_region_size, pageid_t datapage_region_size, pageid_t datapage_size)
{
recovering = true;
this->max_c0_size = max_c0_size;
@ -68,8 +67,7 @@ logtable<TUPLE>::logtable(int log_mode, pageid_t max_c0_size, pageid_t internal_
stasis_log_file_permissions);
}
template<class TUPLE>
logtable<TUPLE>::~logtable()
logtable::~logtable()
{
delete merge_mgr; // shuts down pretty print thread.
@ -94,8 +92,7 @@ logtable<TUPLE>::~logtable()
delete tmerger;
}
template<class TUPLE>
void logtable<TUPLE>::init_stasis() {
void logtable::init_stasis() {
DataPage::register_stasis_page_impl();
stasis_buffer_manager_hint_writes_are_sequential = 1;
@ -103,11 +100,9 @@ void logtable<TUPLE>::init_stasis() {
}
template<class TUPLE>
void logtable<TUPLE>::deinit_stasis() { Tdeinit(); }
void logtable::deinit_stasis() { Tdeinit(); }
template<class TUPLE>
recordid logtable<TUPLE>::allocTable(int xid)
recordid logtable::allocTable(int xid)
{
table_rec = Talloc(xid, sizeof(tbl_header));
mergeStats * stats = 0;
@ -128,8 +123,8 @@ recordid logtable<TUPLE>::allocTable(int xid)
return table_rec;
}
template<class TUPLE>
void logtable<TUPLE>::openTable(int xid, recordid rid) {
void logtable::openTable(int xid, recordid rid) {
table_rec = rid;
Tread(xid, table_rec, &tbl_header);
tree_c2 = new diskTreeComponent(xid, tbl_header.c2_root, tbl_header.c2_state, tbl_header.c2_dp_state, 0);
@ -143,16 +138,14 @@ void logtable<TUPLE>::openTable(int xid, recordid rid) {
}
template<class TUPLE>
void logtable<TUPLE>::logUpdate(datatuple * tup) {
void logtable::logUpdate(datatuple * tup) {
byte * buf = tup->to_bytes();
LogEntry * e = stasis_log_write_update(log_file, 0, INVALID_PAGE, 0/*Page**/, 0/*op*/, buf, tup->byte_length());
log_file->write_entry_done(log_file,e);
free(buf);
}
template<class TUPLE>
void logtable<TUPLE>::replayLog() {
void logtable::replayLog() {
lsn_t start = tbl_header.log_trunc;
LogHandle * lh = start ? getLSNHandle(log_file, start) : getLogHandle(log_file);
const LogEntry * e;
@ -173,13 +166,12 @@ void logtable<TUPLE>::replayLog() {
}
template<class TUPLE>
lsn_t logtable<TUPLE>::get_log_offset() {
lsn_t logtable::get_log_offset() {
if(recovering || !log_mode) { return INVALID_LSN; }
return log_file->next_available_lsn(log_file);
}
template<class TUPLE>
void logtable<TUPLE>::truncate_log() {
void logtable::truncate_log() {
if(recovering) {
printf("Not truncating log until recovery is complete.\n");
} else {
@ -190,8 +182,7 @@ void logtable<TUPLE>::truncate_log() {
}
}
template<class TUPLE>
void logtable<TUPLE>::update_persistent_header(int xid, lsn_t trunc_lsn) {
void logtable::update_persistent_header(int xid, lsn_t trunc_lsn) {
tbl_header.c2_root = tree_c2->get_root_rid();
tbl_header.c2_dp_state = tree_c2->get_datapage_allocator_rid();
@ -210,8 +201,7 @@ void logtable<TUPLE>::update_persistent_header(int xid, lsn_t trunc_lsn) {
Tset(xid, table_rec, &tbl_header);
}
template<class TUPLE>
void logtable<TUPLE>::flushTable()
void logtable::flushTable()
{
struct timeval start_tv, stop_tv;
double start, stop;
@ -269,8 +259,7 @@ void logtable<TUPLE>::flushTable()
c0_flushing = false;
}
template<class TUPLE>
datatuple * logtable<TUPLE>::findTuple(int xid, const datatuple::key_t key, size_t keySize)
datatuple * logtable::findTuple(int xid, const datatuple::key_t key, size_t keySize)
{
//prepare a search tuple
datatuple *search_tuple = datatuple::create(key, keySize);
@ -450,8 +439,7 @@ datatuple * logtable<TUPLE>::findTuple(int xid, const datatuple::key_t key, size
* returns the first record found with the matching key
* (not to be used together with diffs)
**/
template<class TUPLE>
datatuple * logtable<TUPLE>::findTuple_first(int xid, datatuple::key_t key, size_t keySize)
datatuple * logtable::findTuple_first(int xid, datatuple::key_t key, size_t keySize)
{
//prepare a search tuple
datatuple * search_tuple = datatuple::create(key, keySize);
@ -545,9 +533,7 @@ datatuple * logtable<TUPLE>::findTuple_first(int xid, datatuple::key_t key, size
}
template<class TUPLE>
datatuple * logtable<TUPLE>::insertTupleHelper(datatuple *tuple)
datatuple * logtable::insertTupleHelper(datatuple *tuple)
{
bool need_free = false;
if(!tuple->isDelete() && expiry != 0) {
@ -560,10 +546,10 @@ datatuple * logtable<TUPLE>::insertTupleHelper(datatuple *tuple)
memcpy(newkey, tuple->strippedkey(), kl);
newkey[kl] = 0;
memcpy(newkey+kl+1, &ts, ts_sz);
TUPLE * old = tuple;
datatuple * old = tuple;
tuple = datatuple::create(newkey, kl+ 1+ ts_sz, tuple->data(), tuple->datalen());
assert(tuple->strippedkeylen() == old->strippedkeylen());
assert(!TUPLE::compare_obj(tuple, old));
assert(!datatuple::compare_obj(tuple, old));
free(newkey);
need_free = true;
} //find the previous tuple with same key in the memtree if exists
@ -590,12 +576,12 @@ datatuple * logtable<TUPLE>::insertTupleHelper(datatuple *tuple)
tree_c0->insert(t);
}
if(need_free) { TUPLE::freetuple(tuple); }
if(need_free) { datatuple::freetuple(tuple); }
return pre_t;
}
template<class TUPLE>
void logtable<TUPLE>::insertManyTuples(datatuple ** tuples, int tuple_count) {
void logtable::insertManyTuples(datatuple ** tuples, int tuple_count) {
for(int i = 0; i < tuple_count; i++) {
merge_mgr->read_tuple_from_small_component(0, tuples[i]);
}
@ -625,8 +611,7 @@ void logtable<TUPLE>::insertManyTuples(datatuple ** tuples, int tuple_count) {
merge_mgr->read_tuple_from_large_component(0, num_old_tups, sum_old_tup_lens);
}
template<class TUPLE>
void logtable<TUPLE>::insertTuple(datatuple *tuple)
void logtable::insertTuple(datatuple *tuple)
{
if(log_mode && !recovering) {
logUpdate(tuple);
@ -653,8 +638,7 @@ void logtable<TUPLE>::insertTuple(datatuple *tuple)
DEBUG("tree size %d tuples %lld bytes.\n", tsize, tree_bytes);
}
template<class TUPLE>
bool logtable<TUPLE>::testAndSetTuple(datatuple *tuple, datatuple *tuple2)
bool logtable::testAndSetTuple(datatuple *tuple, datatuple *tuple2)
{
bool succ = false;
static pthread_mutex_t test_and_set_mut = PTHREAD_MUTEX_INITIALIZER;
@ -682,12 +666,11 @@ bool logtable<TUPLE>::testAndSetTuple(datatuple *tuple, datatuple *tuple2)
return succ;
}
template<class TUPLE>
void logtable<TUPLE>::registerIterator(iterator * it) {
void logtable::registerIterator(iterator * it) {
its.push_back(it);
}
template<class TUPLE>
void logtable<TUPLE>::forgetIterator(iterator * it) {
void logtable::forgetIterator(iterator * it) {
for(unsigned int i = 0; i < its.size(); i++) {
if(its[i] == it) {
its.erase(its.begin()+i);
@ -695,11 +678,10 @@ void logtable<TUPLE>::forgetIterator(iterator * it) {
}
}
}
template<class TUPLE>
void logtable<TUPLE>::bump_epoch() {
void logtable::bump_epoch() {
epoch++;
for(unsigned int i = 0; i < its.size(); i++) {
its[i]->invalidate();
}
}
template class logtable<datatuple>;

59
logstore.h
View file

@ -17,7 +17,6 @@
class logtable_mergedata;
template<class TUPLE>
class logtable {
public:
@ -205,11 +204,11 @@ public:
template<class ITRA, class ITRN>
class mergeManyIterator {
public:
explicit mergeManyIterator(ITRA* a, ITRN** iters, int num_iters, TUPLE*(*merge)(const TUPLE*,const TUPLE*), int (*cmp)(const TUPLE*,const TUPLE*)) :
explicit mergeManyIterator(ITRA* a, ITRN** iters, int num_iters, datatuple*(*merge)(const datatuple*,const datatuple*), int (*cmp)(const datatuple*,const datatuple*)) :
num_iters_(num_iters+1),
first_iter_(a),
iters_((ITRN**)malloc(sizeof(*iters_) * num_iters)), // exactly the number passed in
current_((TUPLE**)malloc(sizeof(*current_) * (num_iters_))), // one more than was passed in
current_((datatuple**)malloc(sizeof(*current_) * (num_iters_))), // one more than was passed in
last_iter_(-1),
cmp_(cmp),
merge_(merge),
@ -225,7 +224,7 @@ public:
delete(first_iter_);
for(int i = 0; i < num_iters_; i++) {
if(i != last_iter_) {
if(current_[i]) TUPLE::freetuple(current_[i]);
if(current_[i]) datatuple::freetuple(current_[i]);
}
}
for(int i = 1; i < num_iters_; i++) {
@ -235,16 +234,16 @@ public:
free(iters_);
free(dups);
}
TUPLE * peek() {
TUPLE * ret = next_callerFrees();
datatuple * peek() {
datatuple * ret = next_callerFrees();
last_iter_ = -1; // don't advance iterator on next peek() or getnext() call.
return ret;
}
TUPLE * next_callerFrees() {
datatuple * next_callerFrees() {
int num_dups = 0;
if(last_iter_ != -1) {
// get the value after the one we just returned to the user
//TUPLE::freetuple(current_[last_iter_]); // should never be null
//datatuple::freetuple(current_[last_iter_]); // should never be null
if(last_iter_ == 0) {
current_[last_iter_] = first_iter_->next_callerFrees();
} else if(last_iter_ != -1){
@ -272,7 +271,7 @@ public:
}
}
}
TUPLE * ret;
datatuple * ret;
if(!merge_) {
ret = current_[min];
} else {
@ -281,7 +280,7 @@ public:
}
// advance the iterators that match the tuple we're returning.
for(int i = 0; i < num_dups; i++) {
TUPLE::freetuple(current_[dups[i]]); // should never be null
datatuple::freetuple(current_[dups[i]]); // should never be null
current_[dups[i]] = iters_[dups[i]-1]->next_callerFrees();
}
last_iter_ = min; // mark the min iter to be advance at the next invocation of next(). This saves us a copy in the non-merging case.
@ -292,12 +291,12 @@ public:
int num_iters_;
ITRA * first_iter_;
ITRN ** iters_;
TUPLE ** current_;
datatuple ** current_;
int last_iter_;
int (*cmp_)(const TUPLE*,const TUPLE*);
TUPLE*(*merge_)(const TUPLE*,const TUPLE*);
int (*cmp_)(const datatuple*,const datatuple*);
datatuple*(*merge_)(const datatuple*,const datatuple*);
// temporary variables initiaized once for effiency
int * dups;
@ -323,7 +322,7 @@ public:
// rwlc_unlock(ltable->header_mut);
}
explicit iterator(logtable* ltable,TUPLE *key)
explicit iterator(logtable* ltable,datatuple *key)
: ltable(ltable),
epoch(ltable->get_epoch()),
merge_it_(NULL),
@ -346,16 +345,16 @@ public:
ltable->forgetIterator(this);
invalidate();
pthread_mutex_unlock(&ltable->rb_mut);
if(last_returned) TUPLE::freetuple(last_returned);
if(last_returned) datatuple::freetuple(last_returned);
rwlc_unlock(ltable->header_mut);
}
private:
TUPLE * getnextHelper() {
datatuple * getnextHelper() {
// rwlc_readlock(ltable->header_mut);
revalidate();
TUPLE * tmp = merge_it_->next_callerFrees();
datatuple * tmp = merge_it_->next_callerFrees();
if(last_returned && tmp) {
int res = TUPLE::compare(last_returned->strippedkey(), last_returned->strippedkeylen(), tmp->strippedkey(), tmp->strippedkeylen());
int res = datatuple::compare(last_returned->strippedkey(), last_returned->strippedkeylen(), tmp->strippedkey(), tmp->strippedkeylen());
if(res >= 0) {
int al = last_returned->strippedkeylen();
char * a =(char*)malloc(al + 1);
@ -372,21 +371,21 @@ public:
}
if(last_returned) {
TUPLE::freetuple(last_returned);
datatuple::freetuple(last_returned);
}
last_returned = tmp;
// rwlc_unlock(ltable->header_mut);
return last_returned;
}
public:
TUPLE * getnextIncludingTombstones() {
TUPLE * ret = getnextHelper();
datatuple * getnextIncludingTombstones() {
datatuple * ret = getnextHelper();
ret = ret ? ret->create_copy() : NULL;
return ret;
}
TUPLE * getnext() {
TUPLE * ret;
datatuple * getnext() {
datatuple * ret;
while((ret = getnextHelper()) && ret->isDelete()) { } // getNextHelper handles its own memory.
ret = ret ? ret->create_copy() : NULL; // XXX hate making copy! Caller should not manage our memory.
return ret;
@ -423,8 +422,8 @@ public:
merge_it_t* merge_it_;
TUPLE * last_returned;
TUPLE * key;
datatuple * last_returned;
datatuple * key;
bool valid;
int reval_count;
static const int reval_period = 100;
@ -475,13 +474,13 @@ public:
disk_it[3] = ltable->get_tree_c2()->open_iterator(t);
inner_merge_it_t * inner_merge_it =
new inner_merge_it_t(c0_it, c0_mergeable_it, 1, NULL, TUPLE::compare_obj);
merge_it_ = new merge_it_t(inner_merge_it, disk_it, 4, NULL, TUPLE::compare_obj); // XXX Hardcodes comparator, and does not handle merges
new inner_merge_it_t(c0_it, c0_mergeable_it, 1, NULL, datatuple::compare_obj);
merge_it_ = new merge_it_t(inner_merge_it, disk_it, 4, NULL, datatuple::compare_obj); // XXX Hardcodes comparator, and does not handle merges
if(last_returned) {
TUPLE * junk = merge_it_->peek();
if(junk && !TUPLE::compare(junk->strippedkey(), junk->strippedkeylen(), last_returned->strippedkey(), last_returned->strippedkeylen())) {
datatuple * junk = merge_it_->peek();
if(junk && !datatuple::compare(junk->strippedkey(), junk->strippedkeylen(), last_returned->strippedkey(), last_returned->strippedkeylen())) {
// we already returned junk
TUPLE::freetuple(merge_it_->next_callerFrees());
datatuple::freetuple(merge_it_->next_callerFrees());
}
}
valid = true;

6
mergeManager.cpp
View file

@ -499,7 +499,7 @@ void mergeManager::init_helper(void) {
pthread_create(&pp_thread, 0, merge_manager_pretty_print_thread, (void*)this);
}
mergeManager::mergeManager(logtable<datatuple> *ltable):
mergeManager::mergeManager(logtable *ltable):
UPDATE_PROGRESS_PERIOD(0.005),
ltable(ltable) {
c0 = new mergeStats(0, ltable ? ltable->max_c0_size : 10000000);
@ -507,7 +507,7 @@ mergeManager::mergeManager(logtable<datatuple> *ltable):
c2 = new mergeStats(2, 0);
init_helper();
}
mergeManager::mergeManager(logtable<datatuple> *ltable, int xid, recordid rid):
mergeManager::mergeManager(logtable *ltable, int xid, recordid rid):
UPDATE_PROGRESS_PERIOD(0.005),
ltable(ltable) {
marshalled_header h;
@ -537,7 +537,7 @@ void mergeManager::marshal(int xid, recordid rid) {
void mergeManager::pretty_print(FILE * out) {
#if EXTENDED_STATS
logtable<datatuple> * lt = (logtable<datatuple>*)ltable;
logtable * lt = ltable;
bool have_c0 = false;
bool have_c0m = false;
bool have_c1 = false;

7
mergeManager.h
View file

@ -15,7 +15,6 @@
#include <stdio.h>
#include <datatuple.h>
template<class TUPLE>
class logtable;
class mergeStats;
@ -37,8 +36,8 @@ public:
uint64_t long_tv(struct timeval& tv) {
return (1000000ULL * (uint64_t)tv.tv_sec) + ((uint64_t)tv.tv_usec);
}
mergeManager(logtable<datatuple> *ltable);
mergeManager(logtable<datatuple> *ltable, int xid, recordid rid);
mergeManager(logtable *ltable);
mergeManager(logtable *ltable, int xid, recordid rid);
void marshal(int xid, recordid rid);
recordid talloc(int xid);
~mergeManager();
@ -94,7 +93,7 @@ private:
*
* TODO: remove mergeManager->ltable?
*/
logtable<datatuple>* ltable;
logtable* ltable;
mergeStats * c0; /// Per-tree component statistics for c0 and c0_mergeable (the latter should always be null...)
mergeStats * c1; /// Per-tree component statistics for c1 and c1_mergeable.
mergeStats * c2; /// Per-tree component statistics for c2.

10
merger.cpp
View file

@ -12,7 +12,7 @@ static void* diskMerge_thr(void* arg) {
return ((merge_scheduler*)arg)->diskMergeThread();
}
merge_scheduler::merge_scheduler(logtable<datatuple> *ltable) : ltable_(ltable), MIN_R(3.0) { }
merge_scheduler::merge_scheduler(logtable *ltable) : ltable_(ltable), MIN_R(3.0) { }
merge_scheduler::~merge_scheduler() { }
void merge_scheduler::shutdown() {
@ -26,7 +26,7 @@ void merge_scheduler::start() {
pthread_create(&disk_merge_thread_, 0, diskMerge_thr, this);
}
bool insert_filter(logtable<datatuple> * ltable, datatuple * t, bool dropDeletes) {
bool insert_filter(logtable * ltable, datatuple * t, bool dropDeletes) {
if(t->isDelete()) {
if(dropDeletes || ! ltable->mightBeAfterMemMerge(t)) {
return false;
@ -41,7 +41,7 @@ template <class ITA, class ITB>
void merge_iterators(int xid, diskTreeComponent * forceMe,
ITA *itrA,
ITB *itrB,
logtable<datatuple> *ltable,
logtable *ltable,
diskTreeComponent *scratch_tree,
mergeStats * stats,
bool dropDeletes);
@ -339,7 +339,7 @@ static void periodically_force(int xid, int *i, diskTreeComponent * forceMe, sta
}
}
static int garbage_collect(logtable<datatuple> * ltable_, datatuple ** garbage, int garbage_len, int next_garbage, bool force = false) {
static int garbage_collect(logtable * ltable_, datatuple ** garbage, int garbage_len, int next_garbage, bool force = false) {
if(next_garbage == garbage_len || force) {
pthread_mutex_lock(&ltable_->rb_mut);
for(int i = 0; i < next_garbage; i++) {
@ -375,7 +375,7 @@ void merge_iterators(int xid,
diskTreeComponent * forceMe,
ITA *itrA, //iterator on c1 or c2
ITB *itrB, //iterator on c0 or c1, respectively
logtable<datatuple> *ltable,
logtable *ltable,
diskTreeComponent *scratch_tree, mergeStats * stats,
bool dropDeletes // should be true iff this is biggest component
)

5
merger.h
View file

@ -2,7 +2,6 @@
#define _MERGER_H_
#include "logstore.h"
#include "datatuple.h"
#include <stasis/common.h>
#undef try
@ -10,7 +9,7 @@
class merge_scheduler {
public:
merge_scheduler(logtable<datatuple> * ltable);
merge_scheduler(logtable * ltable);
~merge_scheduler();
void start();
@ -22,7 +21,7 @@ public:
private:
pthread_t mem_merge_thread_;
pthread_t disk_merge_thread_;
logtable<datatuple> * ltable_;
logtable * ltable_;
const double MIN_R;
};

6
newserver.cpp
View file

@ -51,14 +51,14 @@ int main(int argc, char *argv[])
}
}
logtable<datatuple>::init_stasis();
logtable::init_stasis();
int xid = Tbegin();
recordid table_root = ROOT_RECORD;
{
logtable<datatuple> ltable(log_mode, c0_size);
logtable ltable(log_mode, c0_size);
ltable.expiry = expiry_delta;
if(TrecordType(xid, ROOT_RECORD) == INVALID_SLOT) {
@ -91,7 +91,7 @@ int main(int argc, char *argv[])
printf("Deinitializing stasis...\n");
fflush(stdout);
}
logtable<datatuple>::deinit_stasis();
logtable::deinit_stasis();
printf("Shutdown complete\n");
}

36
requestDispatch.cpp
View file

@ -8,21 +8,21 @@
#include "regionAllocator.h"
template<class HANDLE>
inline int requestDispatch<HANDLE>::op_insert(logtable<datatuple> * ltable, HANDLE fd, datatuple * tuple) {
inline int requestDispatch<HANDLE>::op_insert(logtable * ltable, HANDLE fd, datatuple * tuple) {
//insert/update/delete
ltable->insertTuple(tuple);
//step 4: send response
return writeoptosocket(fd, LOGSTORE_RESPONSE_SUCCESS);
}
template<class HANDLE>
inline int requestDispatch<HANDLE>::op_test_and_set(logtable<datatuple> * ltable, HANDLE fd, datatuple * tuple, datatuple * tuple2) {
inline int requestDispatch<HANDLE>::op_test_and_set(logtable * ltable, HANDLE fd, datatuple * tuple, datatuple * tuple2) {
//insert/update/delete
bool succ = ltable->testAndSetTuple(tuple, tuple2);
//step 4: send response
return writeoptosocket(fd, succ ? LOGSTORE_RESPONSE_SUCCESS : LOGSTORE_RESPONSE_FAIL);
}
template<class HANDLE>
inline int requestDispatch<HANDLE>::op_bulk_insert(logtable<datatuple> *ltable, HANDLE fd) {
inline int requestDispatch<HANDLE>::op_bulk_insert(logtable *ltable, HANDLE fd) {
int err = writeoptosocket(fd, LOGSTORE_RESPONSE_RECEIVING_TUPLES);
datatuple ** tups = (datatuple **) malloc(sizeof(tups[0]) * 100);
int tups_size = 100;
@ -46,7 +46,7 @@ inline int requestDispatch<HANDLE>::op_bulk_insert(logtable<datatuple> *ltable,
return err;
}
template<class HANDLE>
inline int requestDispatch<HANDLE>::op_find(logtable<datatuple> * ltable, HANDLE fd, datatuple * tuple) {
inline int requestDispatch<HANDLE>::op_find(logtable * ltable, HANDLE fd, datatuple * tuple) {
//find the tuple
datatuple *dt = ltable->findTuple_first(-1, tuple->strippedkey(), tuple->strippedkeylen());
@ -91,12 +91,12 @@ inline int requestDispatch<HANDLE>::op_find(logtable<datatuple> * ltable, HANDLE
return err;
}
template<class HANDLE>
inline int requestDispatch<HANDLE>::op_scan(logtable<datatuple> * ltable, HANDLE fd, datatuple * tuple, datatuple * tuple2, size_t limit) {
inline int requestDispatch<HANDLE>::op_scan(logtable * ltable, HANDLE fd, datatuple * tuple, datatuple * tuple2, size_t limit) {
size_t count = 0;
int err = writeoptosocket(fd, LOGSTORE_RESPONSE_SENDING_TUPLES);
if(!err) {
logtable<datatuple>::iterator * itr = new logtable<datatuple>::iterator(ltable, tuple);
logtable::iterator * itr = new logtable::iterator(ltable, tuple);
datatuple * t;
while(!err && (t = itr->getnext())) {
if(tuple2) { // are we at the end of range?
@ -116,17 +116,17 @@ inline int requestDispatch<HANDLE>::op_scan(logtable<datatuple> * ltable, HANDLE
return err;
}
template<class HANDLE>
inline int requestDispatch<HANDLE>::op_flush(logtable<datatuple> * ltable, HANDLE fd) {
inline int requestDispatch<HANDLE>::op_flush(logtable * ltable, HANDLE fd) {
ltable->flushTable();
return writeoptosocket(fd, LOGSTORE_RESPONSE_SUCCESS);
}
template<class HANDLE>
inline int requestDispatch<HANDLE>::op_shutdown(logtable<datatuple> * ltable, HANDLE fd) {
inline int requestDispatch<HANDLE>::op_shutdown(logtable * ltable, HANDLE fd) {
ltable->accepting_new_requests = false;
return writeoptosocket(fd, LOGSTORE_RESPONSE_SUCCESS);
}
template<class HANDLE>
inline int requestDispatch<HANDLE>::op_stat_space_usage(logtable<datatuple> * ltable, HANDLE fd) {
inline int requestDispatch<HANDLE>::op_stat_space_usage(logtable * ltable, HANDLE fd) {
int xid = Tbegin();
@ -224,13 +224,13 @@ inline int requestDispatch<HANDLE>::op_stat_space_usage(logtable<datatuple> * lt
return err;
}
template<class HANDLE>
inline int requestDispatch<HANDLE>::op_stat_perf_report(logtable<datatuple> * ltable, HANDLE fd) {
inline int requestDispatch<HANDLE>::op_stat_perf_report(logtable * ltable, HANDLE fd) {
}
template<class HANDLE>
inline int requestDispatch<HANDLE>::op_stat_histogram(logtable<datatuple> * ltable, HANDLE fd, size_t limit) {
inline int requestDispatch<HANDLE>::op_stat_histogram(logtable * ltable, HANDLE fd, size_t limit) {
if(limit < 3) {
return writeoptosocket(fd, LOGSTORE_PROTOCOL_ERROR);
@ -288,7 +288,7 @@ inline int requestDispatch<HANDLE>::op_stat_histogram(logtable<datatuple> * ltab
return err;
}
template<class HANDLE>
inline int requestDispatch<HANDLE>::op_dbg_blockmap(logtable<datatuple> * ltable, HANDLE fd) {
inline int requestDispatch<HANDLE>::op_dbg_blockmap(logtable * ltable, HANDLE fd) {
// produce a list of stasis regions
int xid = Tbegin();
@ -392,8 +392,8 @@ inline int requestDispatch<HANDLE>::op_dbg_blockmap(logtable<datatuple> * ltable
}
template<class HANDLE>
inline int requestDispatch<HANDLE>::op_dbg_drop_database(logtable<datatuple> * ltable, HANDLE fd) {
logtable<datatuple>::iterator * itr = new logtable<datatuple>::iterator(ltable);
inline int requestDispatch<HANDLE>::op_dbg_drop_database(logtable * ltable, HANDLE fd) {
logtable::iterator * itr = new logtable::iterator(ltable);
datatuple * del;
fprintf(stderr, "DROPPING DATABASE...\n");
long long n = 0;
@ -418,11 +418,11 @@ inline int requestDispatch<HANDLE>::op_dbg_drop_database(logtable<datatuple> * l
return writeoptosocket(fd, LOGSTORE_RESPONSE_SUCCESS);
}
template<class HANDLE>
inline int requestDispatch<HANDLE>::op_dbg_noop(logtable<datatuple> * ltable, HANDLE fd) {
inline int requestDispatch<HANDLE>::op_dbg_noop(logtable * ltable, HANDLE fd) {
return writeoptosocket(fd, LOGSTORE_RESPONSE_SUCCESS);
}
template<class HANDLE>
inline int requestDispatch<HANDLE>::op_dbg_set_log_mode(logtable<datatuple> * ltable, HANDLE fd, datatuple * tuple) {
inline int requestDispatch<HANDLE>::op_dbg_set_log_mode(logtable * ltable, HANDLE fd, datatuple * tuple) {
if(tuple->rawkeylen() != sizeof(int)) {
abort();
return writeoptosocket(fd, LOGSTORE_PROTOCOL_ERROR);
@ -434,7 +434,7 @@ inline int requestDispatch<HANDLE>::op_dbg_set_log_mode(logtable<datatuple> * lt
}
}
template<class HANDLE>
int requestDispatch<HANDLE>::dispatch_request(HANDLE f, logtable<datatuple>*ltable) {
int requestDispatch<HANDLE>::dispatch_request(HANDLE f, logtable *ltable) {
//step 1: read the opcode
network_op_t opcode = readopfromsocket(f, LOGSTORE_CLIENT_REQUEST);
if(opcode == LOGSTORE_CONN_CLOSED_ERROR) {
@ -471,7 +471,7 @@ int requestDispatch<HANDLE>::dispatch_request(HANDLE f, logtable<datatuple>*ltab
}
template<class HANDLE>
int requestDispatch<HANDLE>::dispatch_request(network_op_t opcode, datatuple * tuple, datatuple * tuple2, logtable<datatuple> * ltable, HANDLE fd) {
int requestDispatch<HANDLE>::dispatch_request(network_op_t opcode, datatuple * tuple, datatuple * tuple2, logtable * ltable, HANDLE fd) {
int err = 0;
#if 0
if(tuple) {

32
requestDispatch.h
View file

@ -13,23 +13,23 @@
template<class HANDLE>
class requestDispatch {
private:
static inline int op_insert(logtable<datatuple> * ltable, HANDLE fd, datatuple * tuple);
static inline int op_test_and_set(logtable<datatuple> * ltable, HANDLE fd, datatuple * tuple, datatuple * tuple2);
static inline int op_find(logtable<datatuple> * ltable, HANDLE fd, datatuple * tuple);
static inline int op_scan(logtable<datatuple> * ltable, HANDLE fd, datatuple * tuple, datatuple * tuple2, size_t limit);
static inline int op_bulk_insert(logtable<datatuple> * ltable, HANDLE fd);
static inline int op_flush(logtable<datatuple> * ltable, HANDLE fd);
static inline int op_shutdown(logtable<datatuple> * ltable, HANDLE fd);
static inline int op_stat_space_usage(logtable<datatuple> * ltable, HANDLE fd);
static inline int op_stat_perf_report(logtable<datatuple> * ltable, HANDLE fd);
static inline int op_stat_histogram(logtable<datatuple> * ltable, HANDLE fd, size_t limit);
static inline int op_dbg_blockmap(logtable<datatuple> * ltable, HANDLE fd);
static inline int op_dbg_drop_database(logtable<datatuple> * ltable, HANDLE fd);
static inline int op_dbg_noop(logtable<datatuple> * ltable, HANDLE fd);
static inline int op_dbg_set_log_mode(logtable<datatuple> * ltable, HANDLE fd, datatuple * tuple);
static inline int op_insert(logtable * ltable, HANDLE fd, datatuple * tuple);
static inline int op_test_and_set(logtable * ltable, HANDLE fd, datatuple * tuple, datatuple * tuple2);
static inline int op_find(logtable * ltable, HANDLE fd, datatuple * tuple);
static inline int op_scan(logtable * ltable, HANDLE fd, datatuple * tuple, datatuple * tuple2, size_t limit);
static inline int op_bulk_insert(logtable * ltable, HANDLE fd);
static inline int op_flush(logtable * ltable, HANDLE fd);
static inline int op_shutdown(logtable * ltable, HANDLE fd);
static inline int op_stat_space_usage(logtable * ltable, HANDLE fd);
static inline int op_stat_perf_report(logtable * ltable, HANDLE fd);
static inline int op_stat_histogram(logtable * ltable, HANDLE fd, size_t limit);
static inline int op_dbg_blockmap(logtable * ltable, HANDLE fd);
static inline int op_dbg_drop_database(logtable * ltable, HANDLE fd);
static inline int op_dbg_noop(logtable * ltable, HANDLE fd);
static inline int op_dbg_set_log_mode(logtable * ltable, HANDLE fd, datatuple * tuple);
public:
static int dispatch_request(HANDLE f, logtable<datatuple> * ltable);
static int dispatch_request(network_op_t opcode, datatuple * tuple, datatuple * tuple2, logtable<datatuple> * ltable, HANDLE fd);
static int dispatch_request(HANDLE f, logtable * ltable);
static int dispatch_request(network_op_t opcode, datatuple * tuple, datatuple * tuple2, logtable * ltable, HANDLE fd);
};
#endif /* REQUESTDISPATCH_H_ */

6
server.cpp
View file

@ -37,7 +37,7 @@ void terminate (int param)
printf("Deinitializing stasis...\n");
fflush(stdout);
logtable<datatuple>::deinit_stasis();
logtable::deinit_stasis();
exit(0);
}
@ -50,7 +50,7 @@ int main(int argc, char *argv[])
prev_fn = signal (SIGINT,terminate);
logtable<datatuple>::init_stasis();
logtable::init_stasis();
int xid = Tbegin();
@ -68,7 +68,7 @@ int main(int argc, char *argv[])
printf("note: running w/ 2GB c0 for benchmarking"); // XXX build a separate test server and deployment server?
}
logtable<datatuple> ltable(c0_size);
logtable ltable(c0_size);
recordid table_root = ROOT_RECORD;
if(TrecordType(xid, ROOT_RECORD) == INVALID_SLOT) {

2
simpleServer.cpp
View file

@ -53,7 +53,7 @@ void * simpleServer::worker(int self) {
}
}
simpleServer::simpleServer(logtable<datatuple> * ltable, int max_threads, int port):
simpleServer::simpleServer(logtable * ltable, int max_threads, int port):
ltable(ltable),
port(port),
max_threads(max_threads),

4
simpleServer.h
View file

@ -15,11 +15,11 @@ public:
static const int DEFAULT_PORT = 32432;
static const int DEFAULT_THREADS = 1000;
simpleServer(logtable<datatuple> * ltable, int max_threads = DEFAULT_THREADS, int port = DEFAULT_PORT);
simpleServer(logtable * ltable, int max_threads = DEFAULT_THREADS, int port = DEFAULT_PORT);
bool acceptLoop();
~simpleServer();
private:
logtable<datatuple>* ltable;
logtable* ltable;
int port;
int max_threads;
int * thread_fd;

8
test/check_datapage.cpp
View file

@ -28,7 +28,7 @@ void insertWithConcurrentReads(size_t NUM_ENTRIES) {
sync();
logtable<datatuple>::init_stasis();
logtable::init_stasis();
int xid = Tbegin();
@ -120,7 +120,7 @@ void insertWithConcurrentReads(size_t NUM_ENTRIES) {
Tcommit(xid);
logtable<datatuple>::deinit_stasis();
logtable::deinit_stasis();
}
void insertProbeIter(size_t NUM_ENTRIES)
@ -131,7 +131,7 @@ void insertProbeIter(size_t NUM_ENTRIES)
sync();
logtable<datatuple>::init_stasis();
logtable::init_stasis();
int xid = Tbegin();
@ -219,7 +219,7 @@ void insertProbeIter(size_t NUM_ENTRIES)
Tcommit(xid);
logtable<datatuple>::deinit_stasis();
logtable::deinit_stasis();
}

6
test/check_gen.cpp
View file

@ -15,11 +15,11 @@ int main(int argc, char **argv)
sync();
logtable<datatuple>::init_stasis();
logtable::init_stasis();
int xid = Tbegin();
logtable<datatuple> *ltable = new logtable<datatuple>(1000, 10000, 5);
logtable *ltable = new logtable(1000, 10000, 5);
recordid table_root = ltable->allocTable(xid);
@ -34,7 +34,7 @@ int main(int argc, char **argv)
delete ro_alloc;
Tcommit(xid);
delete ltable;
logtable<datatuple>::deinit_stasis();
logtable::deinit_stasis();

4
test/check_logtable.cpp
View file

@ -28,7 +28,7 @@ void insertProbeIter(size_t NUM_ENTRIES)
sync();
logtable<datatuple>::init_stasis();
logtable::init_stasis();
int xid = Tbegin();
@ -116,7 +116,7 @@ void insertProbeIter(size_t NUM_ENTRIES)
printf("Random Reads completed.\n");
Tcommit(xid);
logtable<datatuple>::deinit_stasis();
logtable::deinit_stasis();
}
/** @test

4
test/check_logtree.cpp
View file

@ -36,7 +36,7 @@ void insertProbeIter_str(int NUM_ENTRIES)
system("rm -rf stasis_log/");
sync();
logtable<datatuple>::init_stasis();
logtable::init_stasis();
int xid = Tbegin();
@ -138,7 +138,7 @@ void insertProbeIter_str(int NUM_ENTRIES)
delete it;
delete ro_alloc;
Tcommit(xid);
logtable<datatuple>::deinit_stasis();
logtable::deinit_stasis();
}

6
test/check_merge.cpp
View file

@ -26,7 +26,7 @@ void insertProbeIter(size_t NUM_ENTRIES)
unlink("logfile.txt");
system("rm -rf stasis_log/");
logtable<datatuple>::init_stasis();
logtable::init_stasis();
//data generation
std::vector<std::string> * data_arr = new std::vector<std::string>;
@ -51,7 +51,7 @@ void insertProbeIter(size_t NUM_ENTRIES)
int xid = Tbegin();
logtable<datatuple> * ltable = new logtable<datatuple>(10 * 1024 * 1024, 1000, 10000, 5);
logtable * ltable = new logtable(10 * 1024 * 1024, 1000, 10000, 5);
merge_scheduler mscheduler(ltable);
recordid table_root = ltable->allocTable(xid);
@ -149,7 +149,7 @@ void insertProbeIter(size_t NUM_ENTRIES)
Tcommit(xid);
delete ltable;
logtable<datatuple>::deinit_stasis();
logtable::deinit_stasis();
}

6
test/check_mergelarge.cpp
View file

@ -26,7 +26,7 @@ void insertProbeIter(size_t NUM_ENTRIES)
unlink("logfile.txt");
system("rm -rf stasis_log/");
logtable<datatuple>::init_stasis();
logtable::init_stasis();
//data generation
// std::vector<std::string> * data_arr = new std::vector<std::string>;
@ -45,7 +45,7 @@ void insertProbeIter(size_t NUM_ENTRIES)
int xid = Tbegin();
logtable<datatuple> *ltable = new logtable<datatuple>(10*1024*1024, 1000, 10000, 100);
logtable *ltable = new logtable(10*1024*1024, 1000, 10000, 100);
merge_scheduler mscheduler(ltable);
recordid table_root = ltable->allocTable(xid);
@ -92,7 +92,7 @@ void insertProbeIter(size_t NUM_ENTRIES)
printf("merge threads finished.\n");
gettimeofday(&stop_tv,0);
printf("run time: %6.1f\n", (tv_to_double(stop_tv) - tv_to_double(start_tv)));
logtable<datatuple>::deinit_stasis();
logtable::deinit_stasis();
}

6
test/check_mergetuple.cpp
View file

@ -28,7 +28,7 @@ void insertProbeIter(size_t NUM_ENTRIES)
sync();
logtable<datatuple>::init_stasis();
logtable::init_stasis();
double delete_freq = .05;
double update_freq = .15;
@ -103,7 +103,7 @@ void insertProbeIter(size_t NUM_ENTRIES)
int xid = Tbegin();
logtable<datatuple> *ltable = new logtable<datatuple>(10 * 1024 * 1024, 1000, 1000, 40);
logtable *ltable = new logtable(10 * 1024 * 1024, 1000, 1000, 40);
merge_scheduler mscheduler(ltable);
recordid table_root = ltable->allocTable(xid);
@ -248,7 +248,7 @@ void insertProbeIter(size_t NUM_ENTRIES)
Tcommit(xid);
delete ltable;
logtable<datatuple>::deinit_stasis();
logtable::deinit_stasis();
}

8
test/check_testAndSet.cpp
View file

@ -28,7 +28,7 @@
unsigned char vals[NUM_THREADS];
logtable<datatuple> * ltable;
logtable * ltable;
int myucharcmp(const void * ap, const void * bp) {
unsigned char a = *(unsigned char*)ap;
@ -59,10 +59,10 @@ void insertProbeIter(size_t NUM_ENTRIES)
unlink("logfile.txt");
system("rm -rf stasis_log/");
logtable<datatuple>::init_stasis();
logtable::init_stasis();
int xid = Tbegin();
ltable = new logtable<datatuple>(10 * 1024 * 1024, 1000, 10000, 5);
ltable = new logtable(10 * 1024 * 1024, 1000, 10000, 5);
merge_scheduler mscheduler(ltable);
@ -92,7 +92,7 @@ void insertProbeIter(size_t NUM_ENTRIES)
mscheduler.shutdown();
delete ltable;
logtable<datatuple>::deinit_stasis();
logtable::deinit_stasis();
printf("\npass\n");
}