/*
* sophia database
* sphia.org
*
* Copyright (c) Dmitry Simonenko
* BSD License
*/
#include <sp.h>
static inline void
sp_pageopen(spc *c, uint32_t page)
{
assert(page < c->s->s.count);
sppage *p = c->s->s.i[page];
spepoch *e = p->epoch;
c->pi = page;
c->p = p;
c->ph = (sppageh*)(e->db.map + p->offset);
/* validate header */
assert(c->ph->id > 0);
c->pvi = 0 ;
c->pv = (spvh*)((char*)c->ph + sizeof(sppageh));
}
static inline void sp_pageclose(spc *c) {
c->p = NULL;
c->ph = NULL;
c->pv = NULL;
}
static inline void
sp_pagesetlast(spc *c)
{
c->pvi = c->ph->count - 1;
c->pv = (spvh*)((char*)c->ph + sizeof(sppageh) + c->ph->bsize * c->pvi);
}
static inline int
sp_pageseek(spc *c, char *rkey, int size, int *minp, int *midp, int *maxp)
{
int min = 0;
int max = c->ph->count - 1;
int mid = 0;
while (max >= min)
{
mid = min + ((max - min) >> 1);
spvh *a = (spvh*)((char*)c->ph + sizeof(sppageh) + c->ph->bsize * mid);
register int rc =
c->s->env->cmp(a->key, a->size, rkey, size, c->s->env->cmparg);
switch (rc) {
case -1: min = mid + 1;
continue;
case 1: max = mid - 1;
continue;
default:
*minp = min;
*midp = mid;
*maxp = max;
return 1;
}
}
*minp = min;
*midp = mid;
*maxp = max;
return 0;
}
static inline int
sp_pagesetlte(spc *c, char *rkey, int size)
{
int mid, min, max;
int match;
int eq = 0;
if (sp_pageseek(c, rkey, size, &min, &mid, &max)) {
eq = 1;
match = mid;
} else {
match = min;
}
if (match >= c->ph->count)
match = c->ph->count; /* skip */
c->pv = (spvh*)((char*)c->ph + sizeof(sppageh) + c->ph->bsize * match);
c->pvi = match;
return eq;
}
static inline int
sp_pagesetgte(spc *c, char *rkey, int size)
{
int mid, min, max;
int match;
int eq = 0;
if (sp_pageseek(c, rkey, size, &min, &mid, &max)) {
eq = 1;
match = mid;
} else {
match = max;
}
if (match < 0)
match = c->ph->count; /* skip */
else
if (match >= c->ph->count)
match = c->ph->count - 1;
c->pv = (spvh*)((char*)c->ph + sizeof(sppageh) + c->ph->bsize * match);
c->pvi = match;
return eq;
}
static inline void
sp_pagenext(spc *c)
{
c->pvi++;
c->pv = (spvh*)((char*)c->pv + c->ph->bsize);
if (spunlikely(c->pvi == c->ph->count)) {
sp_pageclose(c);
uint32_t next_page = c->pi + 1;
if (splikely(next_page < c->s->s.count))
sp_pageopen(c, next_page);
}
}
static inline void
sp_pageprev(spc *c)
{
c->pvi--;
c->pv = (spvh*)((char*)c->pv - c->ph->bsize);
if (spunlikely(c->pvi < 0)) {
sp_pageclose(c);
int prev_page = c->pi - 1;
if (splikely(prev_page >= 0)) {
sp_pageopen(c, prev_page);
sp_pagesetlast(c);
}
}
}
static inline void sp_first(spc *c)
{
sp_iopen(&c->i0, c->s->i);
if (! c->s->iskip)
sp_iopen(&c->i1, sp_ipair(c->s));
if (c->s->txn == SPTMS)
sp_iopen(&c->itxn, &c->s->itxn);
if (spunlikely(c->s->s.count == 0))
return;
sp_pageopen(c, 0);
}
static inline int
sp_lte(spc *c, char *rkey, int size)
{
/* do lte search on all indexes for the key */
int eq = sp_ilte(c->s->i, &c->i0, rkey, size);
if (! c->s->iskip)
eq = eq + sp_ilte(sp_ipair(c->s), &c->i1, rkey, size);
if (c->s->txn == SPTMS)
eq = eq + sp_ilte(&c->s->itxn, &c->itxn, rkey, size);
/* match page for the key */
if (spunlikely(c->s->s.count == 0))
return eq;
/* open page and do lte search of key */
uint32_t idx = 0;
sppage *p spunused;
p = sp_catroute(&c->s->s, rkey, size, &idx);
assert(p != NULL);
sp_pageopen(c, idx);
eq = eq + sp_pagesetlte(c, rkey, size);
if (spunlikely(c->pvi == c->ph->count))
sp_pageclose(c);
return eq;
}
static inline void sp_last(spc *c)
{
sp_iopen(&c->i0, c->s->i);
sp_ilast(&c->i0);
if (! c->s->iskip) {
sp_iopen(&c->i1, sp_ipair(c->s));
sp_ilast(&c->i1);
}
if (c->s->txn == SPTMS) {
sp_iopen(&c->itxn, &c->s->itxn);
sp_ilast(&c->itxn);
}
if (spunlikely(c->s->s.count == 0))
return;
sp_pageopen(c, c->s->s.count-1);
sp_pagesetlast(c);
}
static inline int
sp_gte(spc *c, char *rkey, int size)
{
/* do gte search the index for the key */
int eq = sp_igte(c->s->i, &c->i0, rkey, size);
if (! c->s->iskip)
eq = eq + sp_igte(sp_ipair(c->s), &c->i1, rkey, size);
if (c->s->txn == SPTMS)
eq = eq + sp_igte(&c->s->itxn, &c->itxn, rkey, size);
/* match page for the key */
if (spunlikely(c->s->s.count == 0))
return eq;
/* open page and do gte search of key */
uint32_t idx = 0;
sppage *p spunused;
p = sp_catroute(&c->s->s, rkey, size, &idx);
assert(p != NULL);
sp_pageopen(c, idx);
eq = eq + sp_pagesetgte(c, rkey, size);
if (spunlikely(c->pvi == c->ph->count))
sp_pageclose(c);
return eq;
}
void sp_cursoropen(spc *c, sp *s, sporder o, char *rkey, int size)
{
/* lock all */
sp_glock(s);
c->m = SPMCUR;
c->o = o;
c->s = s;
c->mask = SPCNONE;
c->p = NULL;
c->ph = NULL;
c->pi = 0;
c->pvi = 0;
c->pv = NULL;
sp_iinv(&c->s->itxn, &c->itxn);
sp_iinv(c->s->i, &c->i0);
sp_iinv(sp_ipair(c->s), &c->i1);
c->r.type = SPREFNONE;
switch (o) {
case SPGTE:
case SPGT:
if (rkey) {
/* init first key and skip on gt */
if (sp_lte(c, rkey, size) && o == SPGT)
sp_iterate(c);
return;
}
sp_first(c);
break;
case SPLTE:
case SPLT:
if (rkey) {
/* init last key and skip on lt */
if (sp_gte(c, rkey, size) && o == SPLT)
sp_iterate(c);
return;
}
sp_last(c);
break;
}
}
void sp_cursorclose(spc *c)
{
/* unlock all */
sp_gunlock(c->s);
}
static inline void sp_reset(spc *c) {
c->mask = SPCNONE;
c->r.type = SPREFNONE;
}
static inline int sp_next(spc *c)
{
/* apply last iteration skip mask */
if (c->mask & SPCITXN)
sp_inext(&c->itxn);
if (c->mask & SPCI0)
sp_inext(&c->i0);
if (c->mask & SPCI1)
sp_inext(&c->i1);
if (c->mask & SPCP) {
assert(c->p != NULL);
assert(c->r.v.vh == c->pv);
sp_pagenext(c);
}
sp_reset(c);
/* do priority multi-index LTE search */
int rc;
int dup = SPCNONE;
int src = SPCNONE;
spv *v = NULL;
struct {
spv *v;
int mask;
} vv[3] = {
{ sp_ival(&c->itxn), SPCITXN },
{ sp_ival(&c->i0), SPCI0 },
{ sp_ival(&c->i1), SPCI1 }
};
/* end of iteration */
if (spunlikely(vv[0].v == NULL &&
vv[1].v == NULL &&
vv[2].v == NULL && c->pv == NULL)) {
return 0;
}
int i;
for (i = 0; i < 3; i++) {
if (vv[i].v == NULL)
continue;
if (v == NULL) {
src = vv[i].mask;
v = vv[i].v;
continue;
}
rc = c->s->env->cmp(vv[i].v->key, vv[i].v->size, v->key, v->size,
c->s->env->cmparg);
switch (rc) {
case 0: dup |= vv[i].mask;
break;
case -1:
src = vv[i].mask;
v = vv[i].v;
break;
}
}
/* no page key */
if (c->pv == NULL)
goto index;
/* no index key */
if (v == NULL)
goto page;
/* compare in-memory and on-disk key */
rc = c->s->env->cmp(v->key, v->size, c->pv->key, c->pv->size,
c->s->env->cmparg);
switch (rc) {
case 0: dup |= SPCP;
case -1: goto index;
case 1: goto page;
}
/* unreach */
assert(0);
return 1;
index:
c->mask = dup|src;
c->r.type = SPREFM;
c->r.v.v = v;
return 1;
page:
c->mask = dup|SPCP;
c->r.type = SPREFD;
c->r.v.vh = c->pv;
return 1;
}
static inline int sp_prev(spc *c)
{
/* apply last iteration skip mask */
if (c->mask & SPCITXN)
sp_iprev(&c->itxn);
if (c->mask & SPCI0)
sp_iprev(&c->i0);
if (c->mask & SPCI1)
sp_iprev(&c->i1);
if (c->mask & SPCP) {
assert(c->p != NULL);
assert(c->r.v.vh == c->pv);
sp_pageprev(c);
}
sp_reset(c);
/* do priority multi-index LTE search */
int rc;
int dup = SPCNONE;
int src = SPCNONE;
spv *v = NULL;
struct {
spv *v;
int mask;
} vv[3] = {
{ sp_ival(&c->itxn), SPCITXN },
{ sp_ival(&c->i0), SPCI0 },
{ sp_ival(&c->i1), SPCI1 }
};
/* end of iteration */
if (spunlikely(vv[0].v == NULL &&
vv[1].v == NULL &&
vv[2].v == NULL && c->pv == NULL)) {
return 0;
}
int i;
for (i = 0; i < 3; i++) {
if (vv[i].v == NULL)
continue;
if (v == NULL) {
src = vv[i].mask;
v = vv[i].v;
continue;
}
rc = c->s->env->cmp(vv[i].v->key, vv[i].v->size, v->key, v->size,
c->s->env->cmparg);
switch (rc) {
case 0: dup |= vv[i].mask;
break;
case 1:
src = vv[i].mask;
v = vv[i].v;
break;
}
}
/* no page key */
if (c->pv == NULL)
goto index;
/* no index key */
if (v == NULL)
goto page;
/* compare in-memory and on-disk key */
rc = c->s->env->cmp(v->key, v->size, c->pv->key, c->pv->size,
c->s->env->cmparg);
switch (rc) {
case 0: dup |= SPCP;
case 1: goto index;
case -1: goto page;
}
/* unreach */
assert(0);
return 1;
index:
c->mask = dup|src;
c->r.type = SPREFM;
c->r.v.v = v;
return 1;
page:
c->mask = dup|SPCP;
c->r.type = SPREFD;
c->r.v.vh = c->pv;
return 1;
}
int sp_iterate(spc *c) {
int rc = 0;
do {
switch (c->o) {
case SPGTE:
case SPGT: rc = sp_next(c);
break;
case SPLTE:
case SPLT: rc = sp_prev(c);
break;
}
} while (rc > 0 && sp_refisdel(&c->r));
return rc;
}
static inline int
sp_matchi(sp *s, spi *i, void *key, size_t size, void **v, size_t *vsize)
{
spv *a = sp_igetraw(i, key, size);
if (splikely(a)) {
if (a->flags & SPDEL)
return 0;
if (v) {
*vsize = sp_vvsize(a);
*v = NULL;
if (*vsize > 0) {
*v = sp_memdup(s, sp_vv(a), sp_vvsize(a));
if (spunlikely(*v == NULL))
return sp_e(s, SPEOOM, "failed to allocate key");
}
}
return 1;
}
return 0;
}
int sp_match(sp *s, void *k, size_t ksize, void **v, size_t *vsize)
{
/*
* if multi-stmt transaction is active, try to search the key
* in the transaction findex first */
int rc;
if (s->txn == SPTMS) {
rc = sp_matchi(s, &s->itxn, k, ksize, v, vsize);
if (rc == -1 || rc == 1)
return rc;
}
/* match both in-memory index'es for the key */
sp_lock(&s->locki);
rc = sp_matchi(s, s->i, k, ksize, v, vsize);
if (rc == -1 || rc == 1) {
sp_unlock(&s->locki);
return rc;
}
/* skip the second index if it is been truncating at the
* somement, all updates are on disk pages */
if (! s->iskip) {
rc = sp_matchi(s, sp_ipair(s), k, ksize, v, vsize);
if (rc == -1 || rc == 1) {
sp_unlock(&s->locki);
return rc;
}
}
sp_unlock(&s->locki);
/* match the page */
sp_lock(&s->locks);
uint32_t page = 0;
sppage *p = sp_catfind(&s->s, k, ksize, &page);
if (p == NULL) {
sp_unlock(&s->locks);
return 0;
}
/* match the key in the page */
spepoch *e = (spepoch*)p->epoch;
sp_lock(&e->lock);
sppageh *ph = (sppageh*)(e->db.map + p->offset);
rc = 0;
register uint32_t min = 0;
register uint32_t max = ph->count - 1;
spvh *match = NULL;
while (max >= min) {
register uint32_t mid = min + ((max - min) >> 1);
register spvh *v =
(spvh*)(e->db.map + p->offset + sizeof(sppageh) +
ph->bsize * mid);
switch (s->env->cmp(v->key, v->size, k, ksize, s->env->cmparg)) {
case -1: min = mid + 1;
continue;
case 1: max = mid - 1;
continue;
default:
match = v;
goto done;
}
}
done:
if (match == NULL)
goto ret;
if (v) {
*vsize = match->vsize;
*v = NULL;
if (match->vsize > 0) {
*v = sp_memdup(s, e->db.map + p->offset + match->voffset, match->vsize);
if (spunlikely(*v == NULL)) {
sp_e(s, SPEOOM, "failed to allocate key");
rc = -1;
goto ret;
}
}
}
rc = 1;
ret:
sp_unlock(&e->lock);
sp_unlock(&s->locks);
return rc;
}