stasis-bLSM/datapage.h

138 lines
3.6 KiB
C
Raw Normal View History

#ifndef _SIMPLE_DATA_PAGE_H_
#define _SIMPLE_DATA_PAGE_H_
#include <limits.h>
#include <stasis/page.h>
#include <stasis/constants.h>
#include "regionAllocator.h"
//#define CHECK_FOR_SCRIBBLING
template<class TUPLE>
class DataPage
{
public:
class iterator
{
private:
void scan_to_key(TUPLE * key) {
if(key) {
len_t old_off = read_offset_;
TUPLE * t = getnext();
while(t && TUPLE::compare(key->key(), key->keylen(), t->key(), t->keylen()) > 0) {
TUPLE::freetuple(t);
old_off = read_offset_;
t = getnext();
}
if(t) {
DEBUG("datapage opened at %s\n", t->key());
TUPLE::freetuple(t);
read_offset_ = old_off;
} else {
DEBUG("datapage key not found. Offset = %lld", read_offset_);
dp = NULL;
}
}
}
public:
iterator(DataPage *dp, TUPLE * key=NULL) : read_offset_(0), dp(dp) {
scan_to_key(key);
}
void operator=(const iterator &rhs)
{
this->read_offset_ = rhs.read_offset_;
this->dp = rhs.dp;
}
//returns the next tuple and also advances the iterator
TUPLE *getnext();
//advance the iterator by count tuples, i.e. skip over count tuples
// void advance(int xid, int count=1);
off_t read_offset_;
DataPage *dp;
};
public:
//to be used when reading an existing data page from disk
DataPage( int xid, pageid_t pid );
//to be used to create new data pages
DataPage( int xid, pageid_t page_count, RegionAllocator* alloc);
~DataPage() {
if(write_offset_ != -1) {
len_t dat_len = 0; // write terminating zero.
write_data((const byte*)&dat_len, sizeof(dat_len), false);
// if writing the zero fails, later reads will fail as well, and assume EOF.
}
}
bool append(TUPLE const * dat);
bool recordRead(typename TUPLE::key_t key, size_t keySize, TUPLE ** buf);
inline uint16_t recordCount();
iterator begin(){return iterator(this);}
pageid_t get_start_pid(){return first_page_;}
int get_page_count(){return page_count_;}
static void register_stasis_page_impl();
private:
// static pageid_t dp_alloc_region(int xid, void *conf, pageid_t count);
void initialize();
private:
static const uint16_t DATA_PAGE_HEADER_SIZE = sizeof(int32_t);
static const uint16_t DATA_PAGE_SIZE = USABLE_SIZE_OF_PAGE - DATA_PAGE_HEADER_SIZE;
typedef uint32_t len_t;
static inline int32_t* is_another_page_ptr(Page *p) {
return stasis_page_int32_ptr_from_start(p,0);
}
static inline byte * data_at_offset_ptr(Page *p, slotid_t offset) {
return ((byte*)(is_another_page_ptr(p)+1))+offset;
}
static inline len_t * length_at_offset_ptr(Page *p, slotid_t offset) {
return (len_t*)data_at_offset_ptr(p,offset);
}
inline recordid calc_chunk_from_offset(off_t offset) {
recordid ret;
ret.page = first_page_ + offset / DATA_PAGE_SIZE;
ret.slot = offset % DATA_PAGE_SIZE;
ret.size = DATA_PAGE_SIZE - ret.slot;
assert(ret.size);
return ret;
}
size_t write_bytes(const byte * buf, size_t remaining);
size_t read_bytes(byte * buf, off_t offset, size_t remaining);
bool write_data(const byte * buf, size_t len, bool init_next = true);
bool read_data(byte * buf, off_t offset, size_t len);
bool initialize_next_page();
void initialize_page(pageid_t pageid);
int xid_;
pageid_t page_count_;
const pageid_t initial_page_count_;
RegionAllocator *alloc_;
const pageid_t first_page_;
off_t write_offset_; // points to the next free byte (ignoring page boundaries)
};
#endif