--- /dev/null
+/*
+ openisis - an open implementation of the CDS/ISIS database
+ Version 0.8.x (patchlevel see file Version)
+ Copyright (C) 2001-2003 by Erik Grziwotz, erik@openisis.org
+
+ This library is free software; you can redistribute it and/or
+ modify it under the terms of the GNU Lesser General Public
+ License as published by the Free Software Foundation; either
+ version 2.1 of the License, or (at your option) any later version.
+
+ This library is distributed in the hope that it will be useful,
+ but WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ Lesser General Public License for more details.
+
+ You should have received a copy of the GNU Lesser General Public
+ License along with this library; if not, write to the Free Software
+ Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+
+ see README for more information
+EOH */
+
+/*
+ $Id: lbt.c,v 1.35 2003/05/27 11:03:30 kripke Exp $
+ the btree.
+*/
+
+#include <stdlib.h>
+#include <string.h>
+
+/* special
+#if defined( __GNUC__ ) && defined ( alloca )
+#include <alloca.h>
+#endif
+*/
+#if defined( sparc ) || defined( __ppc__ )
+# define LDB_BIG_ENDIAN
+#endif
+
+#include "lio.h"
+#include "luti.h"
+#include "lbt.h"
+
+
+/*
+probably all relevant cases can be handled by the more
+efficient approach of storing recoded key a la strxfrm
+*/
+#define LBT_USECMP 0
+
+
+/* ************************************************************
+ private types
+*/
+
+enum {
+ /* little endian magic text */
+ LBT_OPEN = 'O' | 'P'<<8 | 'E'<<16 | 'N'<<24,
+ LBT_ISIS = 'I' | 'S'<<8 | 'I'<<16 | 'S'<<24
+};
+
+
+enum {
+ LBT_LEVELS = 16, /* max depth */
+ LBT_POSBITS = 13, /* bits for position within block */
+ LBT_POSMASK = 0x1fff,
+ LBT_POSMAX = LBT_POSMASK /* obviously :) */
+};
+
+typedef struct Dict { /* dictionary of block entries */
+ unsigned short pos; /* 13 bits position of entry, 2 bits val for leaves */
+ unsigned char vln; /* length of entry's value */
+ /* for inner nodes, this is the length of qualifying
+ second key segment (either 0 or bt->vsz <= 23).
+ for leaves, this, together with higher bits of pos,
+ is the number of values of bt->vsz each.
+ */
+ unsigned char kln; /* length of entry's key */
+/*
+ the following access macros are needed for leaves only
+*/
+#define LBT_POS( d ) (LBT_POSMASK & (d)->pos)
+#define LBT_VLN( d ) ((d)->vln | (~LBT_POSMASK & (d)->pos)>>(LBT_POSBITS-8))
+#define LBT_POSADD( d, a ) do { \
+ (d)->pos = (~LBT_POSMASK & (d)->pos) | (LBT_POSMASK & ((d)->pos + (a))); \
+ } while (0)
+#define LBT_VLNADD( d, a ) do { \
+ unsigned _vln = LBT_VLN( d ) + (a); \
+ (d)->pos = (LBT_POSMASK & (d)->pos) \
+ | (~LBT_POSMASK & ((_vln)<<(LBT_POSBITS-8))); \
+ (d)->vln = (unsigned char)(_vln); \
+ } while (0)
+#define LBT_PV( d, p, v ) do { \
+ (d)->pos = (LBT_POSMASK & (p)) | (~LBT_POSMASK & ((v)<<(LBT_POSBITS-8))); \
+ (d)->vln = (unsigned char)(v); \
+ } while (0)
+#define LBT_PVK( d, p, v, k ) do { \
+ LBT_PV( d, p, v ); \
+ (d)->kln = (unsigned char)(k); \
+ } while (0)
+} Dict;
+
+
+/* guard space around block */
+#ifndef NDEBUG
+# define LBT_GUARD 64
+/* template for guard space -- blank or pattern */
+static unsigned char guard[LBT_GUARD];
+#else
+# define LBT_GUARD 0
+#endif
+
+
+/*
+ TODO: check the possible performance gain in findkey
+ from separating disk block and administrative data
+ and aligning disk blocks on PAGE boundaries in memory
+*/
+typedef struct Block {
+#if LBT_GUARD
+ unsigned char grd[LBT_GUARD];
+#endif
+ unsigned char ird; /* is reading */
+ unsigned char wrd; /* #threads waiting for read */
+ unsigned char lck; /* is write locked */
+ unsigned char wlk; /* #threads waiting for write lock */
+#define LBT_BUSY( b ) (*(int*)&(b)->ird) /* any of 1st 4 bytes != 0 */
+ struct Block *nhs; /* next in hash chain */
+ struct Block *nru; /* next (more) recently used */
+ /* TODO: possibly double-link lru list, so we can unlink on every access ? */
+ /* the disk block -- from here on bsz bytes are on disk */
+#define LBT_BOFF ((char *)&((Block*)0)->num - (char*)0)
+#define LBT_BASE( b ) ((unsigned char *)&(b)->num)
+ unsigned num; /* number of this block, 0 is root */
+ unsigned nxt; /* right sibling, "OPEN" or "ISIS" for root */
+ /* typ,key,col are redundant in non-root blocks, maybe abused ... */
+ unsigned char typ; /* type: bsz, vsz, flags */
+ unsigned char key; /* max key length */
+ unsigned char col; /* collation */
+ unsigned char lev; /* level over bottom, 0 for leaves */
+ unsigned short ent; /* 10 bits # entries */
+#define LBT_ENT( b ) (b)->ent /* currently other bits arent abused */
+ unsigned short stk; /* 13 bits offset of keyspace stack */
+/* offset of dict RELATIVE to base */
+#define LBT_DOFF ((char *)&((Block*)0)->dct - (char*)&((Block*)0)->num)
+ Dict dct[1];
+} Block;
+
+/*
+ LAYOUT:
+ dct is an array of dictionary entries in increasing key order
+ stk is growing downwards from block end towards dict
+ position of entry i is b->stk <= LBT_POS(b->dct+i) < bt->bsz
+ decreasing on i (i.e. stack space is in reverse ordering)
+
+ at LBT_POS(b->dct+i), there are b->dct[i].kln bytes key
+ for a leaf entry, the key is followed by
+ LBT_VLN(b->dct+i) (between 0..1023 incl.) values
+ of bt->vsz (between 8..23 incl) each
+ for an inner entry, the key is followed by
+ 4 bytes block number and optionally bt->vsz qualifying value
+
+ SIZES:
+ block size >= 1024
+ block base overhead LBT_DOFF is 16 bytes
+ avail size >= 1008
+ Dict entry is 4 bytes
+ max. used stack space per entry is:
+ inner node: kln+4+bt->vsz <= 255+4+23 = 282
+ new leaf entry (one value): kln+bt->vsz <= 278
+ max total space for a new entry incl. dict is 286 < avail size/3
+*/
+
+typedef struct Chunk {
+ struct Chunk *nxt;
+} Chunk;
+
+
+enum {
+ /* number of levels for lru list */
+ LBT_LRULEV = sizeof(((Idx*)0)->lru)/sizeof(((Idx*)0)->lru[0]),
+ LBT_ENTBITS = 10, /* bits for # enries */
+ LBT_ENTMASK = 0x03ff,
+ LBT_ENTMAX = LBT_ENTMASK,
+ LBT_MATCH = 0x8000, /* entry matched */
+ LBT_ENTSIZE = 8 /* minimum cost of entry: 4 byte dict, 4 byte val */
+};
+
+
+typedef struct Batch {
+ Block *cur;
+ unsigned got; /* current for this key */
+ unsigned tot; /* total for this key */
+ unsigned let;
+ /* stats */
+ unsigned keys;
+ unsigned keyl;
+ unsigned maxl;
+ unsigned vals;
+ unsigned maxv;
+ unsigned span;
+ unsigned free;
+ /* buffer */
+ Key key;
+ Val val[LBT_ENTMAX];
+} Batch;
+
+
+typedef enum {
+ LBT_RD, /* read -- no lock */
+ LBT_WR, /* write */
+ LBT_EX /* exclusive */
+} lbt_get;
+
+
+typedef struct { /* structure filled by findkey */
+ Idx *bt; /* the tree */
+ Key *key; /* the key searched */
+ Block *b[LBT_LEVELS]; /* path to leaf, b[0] is leaf */
+ unsigned n[LBT_LEVELS]; /* block numbers of path */
+ unsigned short e[LBT_LEVELS]; /* entries in path */
+} FindKey;
+
+
+typedef struct {
+ unsigned get;
+ unsigned miss;
+ unsigned cmp;
+ unsigned key;
+} Stat;
+
+/* buffer to print a value to.
+ lowest 11 bytes are printed as 4 shorts + one int < 4*5+10 = 30 digits+5sep
+ higher 12 bytes are printed hex = 24 bytes.
+*/
+typedef char PrVal[64];
+
+
+/* ************************************************************
+ private data
+*/
+
+/*
+ stat access is not guarded by memory barrier.
+ may be wrong when running on multiple processors
+*/
+static Stat stat;
+
+
+/* ************************************************************
+ private functions
+*/
+
+#if defined( sparc )
+static unsigned GETINT ( void *m ) /* bus error safe version of *(int*) */
+{
+ unsigned l;
+ memcpy( &l, m, 4 );
+ return l;
+}
+static void SETINT ( void *m, unsigned i )
+{
+ memcpy( m, &i, 4 );
+}
+#else
+#define GETINT( m ) (*(unsigned*)(m))
+#define SETINT( m, i ) (*(unsigned*)(m) = (i))
+#endif
+
+static void mkval ( Val *v, Hit *ifp )
+{
+ unsigned char *c = v->byt;
+ switch ( v->len ) {
+ default: memset( c, 0, v->len-11 ); c += v->len-11;
+ case 11: *c++ = (unsigned char) (ifp->dbn >> 8);
+ case 10: *c++ = (unsigned char) ifp->dbn;
+ case 9: *c++ = (unsigned char) (ifp->mfn >> 24);
+ case 8: *c++ = (unsigned char) (ifp->mfn >> 16);
+ case 7: *c++ = (unsigned char) (ifp->mfn >> 8);
+ case 6: *c++ = (unsigned char) ifp->mfn;
+ case 5: *c++ = (unsigned char) (ifp->tag >> 8);
+ case 4: *c++ = (unsigned char) ifp->tag;
+ case 3: *c++ = (unsigned char) ifp->occ;
+ case 2: *c++ = (unsigned char) (ifp->pos >> 8);
+ case 1: *c++ = (unsigned char) ifp->pos;
+ case 0: ;
+ }
+} /* mkval */
+
+
+static Hit *rdval ( Hit *ifp, unsigned char *c, unsigned vsz )
+{
+ ifp->mfn = ifp->tag = ifp->occ = ifp->pos = ifp->dbn = 0;
+ switch ( vsz ) {
+ default: c += vsz-11;
+ case 11: ifp->dbn = (unsigned short)*c++ << 8;
+ case 10: ifp->dbn |= *c++;
+ case 9: ifp->mfn = (unsigned int)*c++ << 24;
+ case 8: ifp->mfn |= (unsigned int)*c++ << 16;
+ case 7: ifp->mfn |= (unsigned int)*c++ << 8;
+ case 6: ifp->mfn |= *c++;
+ case 5: ifp->tag = (unsigned short)*c++ << 8;
+ case 4: ifp->tag |= *c++;
+ case 3: ifp->occ = *c++;
+ case 2: ifp->pos = (unsigned short)*c++ << 8;
+ case 1: ifp->pos |= *c++;
+ case 0: ;
+ }
+ return ifp;
+} /* rdval */
+
+
+static char *prval ( PrVal buf, unsigned char *c, unsigned vsz )
+{
+ char *d = buf;
+ Hit h;
+ if ( ! vsz ) {
+ buf[0] = '#';
+ buf[1] = 0;
+ return buf;
+ }
+ if ( 23 < vsz )
+ vsz = 23;
+ if ( 11 < vsz ) {
+ while ( 11 < vsz-- ) {
+ *d++ = luti_hex[ *c >> 4 ];
+ *d++ = luti_hex[ *c++ & 0xf ];
+ }
+ *d++ = '.';
+ }
+ rdval( &h, c, vsz );
+ d += u2a( d, h.dbn ); *d++ = '.';
+ d += u2a( d, h.mfn ); *d++ = '.';
+ d += u2a( d, h.tag ); *d++ = '.';
+ d += u2a( d, h.occ ); *d++ = '.';
+ d += u2a( d, h.pos ); *d++ = '.';
+ *d = 0;
+ return buf;
+} /* prval */
+
+/* for PRVAL f->key->val */
+#define PRVAL( buf, v ) prval( buf, v.byt, v.len )
+
+
+/* ************************************************************
+ block I/O & cache mgmt
+*/
+
+static void iniblk ( Block *b, Idx *bt, unsigned num, unsigned char lev )
+{
+ memset( b, 0, LBT_BOFF + bt->bsz + LBT_GUARD );
+ b->num = num;
+ b->typ = bt->typ;
+ b->key = bt->key;
+ b->col = bt->col;
+ b->lev = lev;
+ b->stk = bt->bsz;
+} /* iniblk */
+
+
+static int chkblk ( Block *b, Idx *bt )
+{
+#if 0
+ (void)b; (void)bt;
+#else
+ int i;
+ unsigned short eod = LBT_DOFF + b->ent*sizeof(Dict), siz, pos, vln, key;
+ unsigned end = bt->bsz, tot = 0;
+ unsigned char *base = LBT_BASE(b);
+#if LBT_GUARD
+ if ( memcmp( guard, b->grd, LBT_GUARD ) )
+ return log_msg( ERR_TRASH, "OOPS! initial guard frobbed on blk %d", b->num );
+ if ( memcmp( guard, LBT_BASE(b)+bt->bsz, LBT_GUARD ) )
+ return log_msg( ERR_TRASH, "OOPS! initial guard frobbed on blk %d", b->num );
+#endif
+ if ( b->num > bt->len || (b->num && b->nxt > bt->len) )
+ return log_msg( ERR_TRASH,
+ "bad num %d / nxt %d on lev %d", b->num, b->nxt, b->lev );
+ if ( b->typ != bt->typ || b->key != bt->key || b->col != bt->col )
+ return log_msg( ERR_TRASH,
+ "bad sig 0x%02x,%d,%d in block %d want 0x%02x,%d,%d",
+ b->typ, b->key, b->col, b->num, bt->typ, bt->key, bt->col );
+ if ( b->ent > bt->bsz / LBT_ENTSIZE || b->stk > bt->bsz || b->stk < eod )
+ return log_msg( ERR_TRASH,
+ "bad ent/stk %d,%d in block %d eod %d sz %d",
+ b->ent, b->stk, b->num, eod, bt->bsz );
+ for ( i=0; i<b->ent; i++ ) {
+ Dict *d = &b->dct[i];
+ pos = LBT_POS( d );
+ vln = LBT_VLN( d );
+ key = d->kln;
+ siz = key + (b->lev ? vln+4u : vln * bt->vsz);
+ if ( pos < b->stk || end != (unsigned)pos + siz )
+ return log_msg( ERR_TRASH,
+ "bad pos/vln/key %d/%d/%d in entry %d of block %d"
+ " stack %d sz %d end %d expected %d",
+ pos, vln, key, i, b->num, b->stk, siz, pos+siz, end );
+ if ( b->lev ) {
+ unsigned lnk;
+ if ( vln && vln != +bt->vsz )
+ return log_msg( ERR_TRASH,
+ "bad vln %d in entry %d of block %d vsz %d",
+ vln, i, b->num, bt->vsz );
+ lnk = GETINT( base + pos + key + vln );
+ if ( lnk >= bt->len )
+ return log_msg( ERR_TRASH,
+ "bad link %u in entry %d of block %d",
+ lnk, i, b->num );
+ }
+#if 1 && ! defined( NDEBUG )
+ if ( i ) {
+ unsigned short op = LBT_POS( d-1 ), okl = d[-1].kln;
+ int mc = memcmp( base+pos, base+op, key<okl ? key : okl ), j;
+ if ( 0 < mc || (!mc && 0 < (mc = (int)key - (int)okl)) )
+ goto ok;
+ j = 0;
+ if ( !mc && b->lev && vln ) { /* compare values */
+ unsigned ovln = d[-1].vln;
+ if ( ! ovln || 0 < (j = memcmp( base+pos+key, base+op+okl, vln )) )
+ goto ok;
+ }
+ return log_msg( ERR_TRASH,
+ "bad key order entry %d of block %d lev %d"
+ " mc %d j %d key %d okl %d vln %d",
+ i, b->num, b->lev, mc, j, key, okl, vln
+ );
+ }
+ ok:
+#endif
+ tot += siz;
+ end = pos;
+ }
+ if ( tot > bt->bsz - b->stk )
+ return log_msg( ERR_TRASH,
+ "bad total entry size %d in block %d stack %d",
+ tot, b->num, b->stk );
+#ifndef NDEBUG
+ if ( LOG_DO( LOG_TRACE ) ) {
+ log_msg( LOG_TRACE,
+ "block %d lev %d sig 0x%02x,%d,%d ent %d eod %d stk %d",
+ b->num, b->lev, b->typ, b->key, b->col, b->ent, eod, b->stk
+ );
+ for ( i=0; i<2; i++ ) {
+ int j = i ? b->ent-1 : 0;
+ Dict *d = &b->dct[j];
+ if ( i >= b->ent ) /* if b->ent is 0 or 1, log no or one key */
+ break;
+ pos = LBT_POS( d );
+ vln = LBT_VLN( d );
+ key = d->kln;
+ end = pos + b->lev ? vln+4u : vln * bt->vsz;
+ /* start of value */
+ if ( b->lev )
+ log_msg( LOG_TRACE, "key %d '%.*s'/%d -> %d",
+ j, key, base+pos, vln, GETINT(base+pos+key+vln) );
+ else {
+ Hit h;
+ rdval( &h, base+pos+key, bt->vsz );
+ log_msg( LOG_TRACE, "key %d '%.*s'[%d] = %d[%d,%d,%d]",
+ j, key, base+pos, vln, h.mfn, h.tag, h.occ, h.pos );
+ }
+ }
+ }
+#endif
+#endif
+ return 0;
+} /* chkblk */
+
+
+static Block *addchunk ( Idx *bt, unsigned cnt )
+{
+ unsigned ibsz = LBT_BOFF + bt->bsz + LBT_GUARD; /* internal block size */
+ /* increase cnt so it's > 0 and add one extra to return */
+ char *mem = (char *)mAlloc( sizeof(Chunk) + (1 + ++cnt)*ibsz );
+ Chunk *chk = (Chunk *)mem;
+ if ( ! mem )
+ return 0;
+ /* chain the chunk */
+ chk->nxt = bt->mem;
+ bt->mem = chk;
+ mem += sizeof(Chunk);
+ bt->clen += cnt;
+ /* add cnt blocks to level 0 lru list */
+ if ( ! bt->mru[0] ) /* level 0 cache is empty */
+ bt->mru[0] = (Block *)mem;
+ for ( ; cnt--; mem += ibsz ) {
+ Block *b = (Block *)mem;
+ iniblk( b, bt, 0 , 0 ); /* initialise empty block */
+ b->nru = bt->lru[0];
+ bt->lru[0] = b;
+ }
+ iniblk( (Block *)mem, bt, 0 , 0 );
+ return (Block *)mem;
+} /* addchunk */
+
+
+static int readblk ( Block *b, Idx *bt, unsigned num )
+{
+ int got = lio_pread( &bt->fd, LBT_BASE(b), bt->bsz, num*bt->bsz );
+ log_msg( LOG_VERBOSE, "reading block %u got %u", num, got );
+ if ( bt->bsz != (unsigned)got )
+ return log_msg( LOG_SYSERR, "\twhen reading bt block %d", num );
+ if ( b->num != num )
+ return log_msg( ERR_TRASH, "bad num %d in block %d", b->num, num );
+ return chkblk( b, bt );
+} /* readblk */
+
+
+static void unlock ( Block *b )
+{
+ b->lck = 0;
+ if ( b->wlk )
+ (void)LIO_WAKE( b->num );
+}
+
+#define ULK( b ) do { \
+ if ( (b)->lck ) { (b)->lck = 0; if ( b->wlk ) unlock( b ); } \
+ } while (0)
+
+
+static int writeblk ( Block *b, Idx *bt, int ulk )
+{
+ int got;
+ if ( chkblk( b, bt ) ) /* don't write no shize */
+ return log_msg( ERR_TRASH, "\tbefore writing bt block %d", b->num );
+ got = lio_pwrite( &bt->fd, LBT_BASE(b), bt->bsz, b->num*bt->bsz );
+ log_msg( LOG_VERBOSE, "writing block %u got %u", b->num, got );
+ if ( ulk )
+ ULK( b );
+ if ( bt->bsz != (unsigned)got )
+ return log_msg( LOG_SYSERR, "\twhen writing bt block %d", b->num );
+ return 0;
+} /* writeblk */
+
+#define WRULK( b, bt ) writeblk( b, bt, 1 )
+
+
+#define NEWBLK( bt, lev, lock ) getblk( bt, bt->len, lev, lock )
+/**
+ get block #num.
+ 1) locate block #num in cache.
+ 2) if it is found:
+ if it is READING, or we want to lock and it is locked,
+ incr # waiters, wait for it.
+ 3) if it is not found:
+ locate a free cache entry, extending the cache if necessary,
+ set the num, set the READING flag, go read it.
+ after reading, if there are waiters, wake them up.
+ since we are the first to get a hold on the block,
+ we may write lock it, if we want.
+ 4) return the block.
+*/
+static Block *getblk ( Idx *bt, unsigned num,
+ unsigned char lev, lbt_get lock )
+{
+ Block *b;
+ stat.get++; /* dirty */
+ if ( !(stat.get & 0x3fff) ) { /* every 16K */
+ static unsigned lmiss;
+ log_msg( LOG_INFO,
+ "searched %i keys in %u blks cache %d %d%%"
+ " depth %u cmp/get/miss %u/%u/%u %d%% (%d %d%%)",
+ stat.key, bt->len, bt->clen, bt->clen*100 / bt->len,
+ bt->root->lev, stat.cmp, stat.get,
+ stat.miss, stat.miss*100 / stat.get,
+ (stat.miss - lmiss), (stat.miss - lmiss)*100 / 0x4000
+ );
+ lmiss = stat.miss;
+ }
+startover: /* only used on absurd congestion */
+ b = 0;
+ LOG_DBG( LOG_DEBUG, "get bt block %u %u %d", num, lev, lock );
+ if ( ! num )
+ b = bt->root;
+ else if ( num < bt->len ) {
+ b = bt->hash[ num % bt->hlen ];
+ while ( b && num != b->num )
+ b = b->nhs;
+ }
+ if ( b )
+ while ( b->ird || (lock && b->lck) ) {
+ int wrd = b->ird;
+ int wlk = lock && b->lck;
+ LOG_DBG( LOG_VERBOSE, "lck on block %u %u r%d l%d",
+ num, lev, b->ird, b->lck );
+ if ( LBT_EX <= lock ) {
+ log_msg( LOG_ERROR, "concurrent mod on EX lock" );
+ return 0;
+ }
+ if ( ! lio_lock ) {
+ log_msg( LOG_FATAL, "bad MT environment: concurrent mod, no lock" );
+ return 0;
+ }
+ if ( wrd && 0xff != b->wrd )
+ b->wrd++;
+ if ( wlk && 0xff != b->wlk )
+ b->wlk++;
+ (void)LIO_WAIT( num ); /* zzzzz .... sleep tight */
+ /* some hours later */
+ if ( num != b->num ) {
+ log_msg( LOG_WARN, "heavy congestion: block %d has gone", num );
+ goto startover;
+ }
+ if ( wrd && b->wrd )
+ b->wrd--;
+ if ( wlk && b->wlk )
+ b->wlk--;
+ }
+ else { /* not found or new block */
+ /* find free block */
+ int i = 0, llev, err = 0;
+ stat.miss++; /* dirty */
+ LOG_DBG( LOG_DEBUG, "bt block %u %u not found", num, lev );
+ for ( ; i < (int)LBT_LRULEV; i++ ) {
+ /* check lru list for level i */
+ Block **bp = &bt->lru[i];
+ while ( (b = *bp) && LBT_BUSY(b) )
+ bp = &b->nru;
+ if ( ! b ) /* try next level */
+ continue;
+ *bp = b->nru; /* dequeue */
+ if ( bt->mru[i] == b ) /* was the tail */
+ bt->mru[i] = 0;
+ break;
+ }
+ if ( ! b ) {
+ log_msg( LOG_WARN, "no free blocks in cache" );
+ if ( !(b = addchunk( bt, 64 )) )
+ return 0;
+ /* TODO: should we also extend the hash ? */
+ } else if ( b->num ) { /* got block -- should be hashed somewhere */
+ Block **bp = &bt->hash[ b->num % bt->hlen ];
+ while ( *bp && *bp != b )
+ bp = &(*bp)->nhs;
+ if ( *bp == b )
+ *bp = b->nhs; /* dehash */
+ /* else ... hmmm should have found it */
+ }
+ assert( ! LBT_BUSY( b ) );
+ iniblk( b, bt, num, lev );
+ /* hash it, so waiters can find it */
+ b->nhs = bt->hash[ b->num % bt->hlen ];
+ bt->hash[ b->num % bt->hlen ] = b;
+ if ( num == bt->len ) { /* extend */
+ b->lev = lev;
+ bt->len++;
+ /* TODO: should be configurable */
+ if ( bt->len > bt->clen << 5 && bt->clen < 0x4000 )
+ addchunk( bt, bt->clen / 4 );
+ } else {
+ b->ird = 1;
+ (void)LIO_RELE();
+ err = readblk( b, bt, num );
+ (void)LIO_LOCK();
+ b->ird = 0;
+ if ( ! err && lev != b->lev )
+ err = log_msg( ERR_TRASH, "bad lev %d in block %d want %d",
+ b->lev, b->num, lev );
+ if ( err ) {
+ Block **bp = &bt->hash[ num % bt->hlen ];
+ while ( *bp && *bp != b )
+ bp = &(*bp)->nhs;
+ if ( *bp == b )
+ *bp = b->nhs; /* dehash */
+ b->lev = 0; /* add to level 0 lru list */
+ b->num = 0; /* tell others that it isn't the expected block */
+ }
+ }
+ /* add to it's levels lru list */
+ llev = LBT_LRULEV > b->lev ? b->lev : LBT_LRULEV-1;
+ if ( bt->mru[ llev ] )
+ bt->mru[ llev ]->nru = b;
+ else
+ bt->lru[ llev ] = b;
+ (bt->mru[ llev ] = b)->nru = 0;
+ if ( b->wrd )
+ (void)LIO_WAKE( num );
+ if ( err ) {
+ b->wrd = b->wlk = 0; /* mark block unused */
+ return 0;
+ }
+ } /* got free block */
+ if ( LBT_WR == lock ) {
+ if ( b->lck ) {
+ log_msg( LOG_FATAL, "bt consi: unexpected lock on %u", num );
+ return 0;
+ }
+ b->lck = 1;
+ }
+ return b;
+} /* getblk */
+
+
+/** binsearch position for key.
+ position is either the index of the matching entry (0..b->ent-1)
+ or the index (0..b->ent) where key should be inserted.
+ @return position for key (0..b->ent), |ifyed with LBT_MATCH, if found
+*/
+static unsigned short keypos (
+#if LBT_USECMP
+ Idx *bt,
+#define KEYPOS( bt, b, key ) keypos( bt, b, key )
+#else
+#define KEYPOS( bt, b, key ) keypos( b, key )
+#endif
+ Block *b, Key *key )
+{
+ unsigned char *base = LBT_BASE( b );
+ unsigned char *kbt = key->byt;
+ unsigned char kln = key->len;
+ unsigned short l = 0, r = b->ent;
+ Dict *d = b->dct;
+ /* binsearch invariant:
+ key is < all entries i >= r (we do know relation at r)
+ key is > all entries i < l (don't know relation at l)
+ */
+ while ( l < r ) { /* invariant: l <= keypos < r */
+ unsigned short i = (l + r) >> 1; /* l <= i < r */
+ unsigned short pos = LBT_POS(&d[i]);
+ unsigned char cln = d[i].kln;
+ int cmp =
+#if LBT_USECMP
+ bt->cmp ? bt->cmp( kbt, base+pos, kln<cln ? kln : cln ) :
+#endif
+ memcmp( kbt, base+pos, kln<cln ? kln : cln );
+ stat.cmp++; /* dirty */
+ if ( ! cmp && !(cmp = (int)kln - cln) ) {
+ /* match on main key -- what about the value ? */
+ if ( ! b->lev /* ignore on leaves -- no duplicate keys here */
+ || (!d[i].vln && !key->val.len) /* no value */
+ )
+ return LBT_MATCH | i;
+ /* so we're on an inner node */
+ if ( ! d[i].vln ) /* common case */
+ cmp = 1;
+ else if ( ! key->val.len )
+ cmp = -1;
+ else {
+ unsigned char vln = LBT_VLN(&d[i]);
+ int c = key->val.len < d[i].vln ? key->val.len : vln;
+ if ( !(cmp = memcmp( key->val.byt, base+pos+cln, c ))
+ && !(cmp = (int) key->val.len - vln)
+ )
+ return LBT_MATCH | i;
+ }
+ }
+ if ( cmp > 0 )
+ l = i+1;
+ else
+ r = i;
+ }
+ return l; /* == r */
+} /* keypos */
+
+
+/** find position for (leaf) val.
+ search v in array b of n values of size sz
+ @return position for value (0..n), |ifyed with LBT_MATCH, if found
+*/
+static unsigned short valpos ( unsigned char *b,
+ unsigned short n, unsigned char sz, unsigned char *v )
+{
+ unsigned short l = 0, r = n;
+ /* binsearch invariant:
+ v is < all values i >= r (we do know relation at r)
+ v is > all values i < l (don't know relation at l)
+ */
+ while ( l < r ) { /* invariant: l <= valpos < r */
+ unsigned short i = (l + r) >> 1; /* l <= i < r */
+ int cmp = memcmp( v, b+i*sz, sz );
+ if ( ! cmp )
+ return LBT_MATCH | i;
+ if ( cmp > 0 )
+ l = i+1;
+ else
+ r = i;
+ }
+ return l; /* == r */
+} /* valpos */
+
+
+#ifndef NDEBUG
+static void TRACEKEY ( unsigned lev, FindKey *f )
+{
+ int i;
+ unsigned n = 0;
+ Hit h;
+ PrVal pv;
+
+ if ( LOG_DONT( lev ) )
+ return;
+ log_msg( lev, "key '%.*s' vln %d val %s",
+ f->key->len, f->key->byt, f->key->val.len,
+ PRVAL( pv, f->key->val ) );
+ if ( f->key->val.len ) {
+ rdval( &h, f->key->val.byt, f->key->val.len );
+ log_msg( lev, "\tdb %d mfn %d tag %d occ %d pos %d",
+ h.dbn, h.mfn, h.tag, h.occ, h.pos );
+ }
+ for ( i=LBT_LEVELS; i--; ) {
+ Block *b = f->b[i];
+ unsigned on = n, e;
+ Dict *d;
+ if ( ! b )
+ continue;
+ d = b->dct + (e = (LBT_ENTMASK & f->e[i]));
+ if ( i )
+ n = GETINT( LBT_BASE(b)+d->pos+d->kln+d->vln );
+ log_msg( lev, "lev %d n %d e 0x%x = %d %c '%.*s' val %d %s -> %d%s",
+ i, f->n[i], f->e[i], e, (LBT_MATCH & f->e[i]) ? 'M' : '-',
+ d->kln, LBT_BASE(b)+d->pos, d->vln,
+ prval( pv, LBT_BASE(b)+d->pos+d->kln, d->vln ),
+ n, on == b->num ? "" : " LY!"
+ );
+ if ( d->vln ) {
+ rdval( &h, LBT_BASE(b)+d->pos+d->kln, d->vln );
+ log_msg( lev, "\tdb %d mfn %d tag %d occ %d pos %d",
+ h.dbn, h.mfn, h.tag, h.occ, h.pos );
+ }
+ if ( e ) {
+ d--;
+ log_msg( lev, "\tleft '%.*s' val %d %s",
+ d->kln, LBT_BASE(b)+d->pos, d->vln,
+ prval( pv, LBT_BASE(b)+d->pos+d->kln, d->vln )
+ );
+ if ( d->vln ) {
+ rdval( &h, LBT_BASE(b)+d->pos+d->kln, d->vln );
+ log_msg( lev, "\tdb %d mfn %d tag %d occ %d pos %d",
+ h.dbn, h.mfn, h.tag, h.occ, h.pos );
+ }
+ }
+ }
+} /* TRACEKEY */
+#else
+#define TRACEKEY( l, f ) (void)(f)
+#endif
+
+
+static int findkey ( FindKey *f, lbt_get lock )
+{
+ int i;
+ Block *b = f->bt->root;
+ unsigned char *pkey = 0, pkln = 0, pvln = 0;
+ if ( b->lev >= LBT_LEVELS )
+ return log_msg( ERR_TRASH, "tree is too deep: %u", b->lev );
+ stat.key++;
+ for ( i=LBT_LEVELS; i--; ) { /* paranoia */
+ f->b[i] = 0;
+ f->n[i] = f->e[i] = 0;
+ }
+ for ( i=b->lev;; ) {
+ unsigned short e;
+ unsigned n;
+ Dict *d;
+#ifndef NDEBUG
+ chkblk( b, f->bt );
+#endif
+ f->n[i] = (f->b[i] = b)->num;
+ f->e[i] = e = KEYPOS( f->bt, b, f->key );
+ LOG_DBG( LOG_DEBUG, "keypos %x", e );
+#ifndef NDEBUG
+ if ( pkey && b->ent ) {
+ /* compare parent key against our first key
+ pkey must be equal for inner node and not greater for leave
+ */
+ int cmp; /* this - pkey >= 0 */
+ unsigned char *ent, ln;
+ d = b->dct;
+ ent = LBT_BASE(b)+LBT_POS(d);
+ ln = pkln < d->kln ? pkln : d->kln;
+ cmp = ln ? memcmp( ent, pkey, ln ) : 0;
+ if ( ! cmp )
+ cmp = (int)d->kln - (int)pkln;
+ if ( ! cmp ) {
+ if ( ! d->vln || ! pvln )
+ cmp = (int)d->vln - (int)pvln;
+ else
+ cmp = memcmp( ent+d->kln, pkey+pkln, f->bt->vsz );
+ }
+ if ( cmp && (i || 0 > cmp) ) {
+ Hit h;
+ PrVal pv;
+ TRACEKEY( ERR_TRASH, f );
+ if ( d->vln ) {
+ rdval( &h, ent+d->kln, d->vln );
+ log_msg( ERR_TRASH, "\tfirst val db %d mfn %d tag %d occ %d pos %d",
+ h.dbn, h.mfn, h.tag, h.occ, h.pos );
+ }
+ return log_msg( ERR_TRASH, "\tbad first '%.*s' val %d %s",
+ d->kln, ent, d->vln, prval( pv, ent+d->kln, d->vln ) );
+ }
+ }
+#endif
+ /*
+ here goes the Lehmann/Yao race detection:
+ If we ran on right boundary, follow right link
+ */
+ while ( !(LBT_MATCH & e) && b->ent == e && b->nxt && b->num ) {
+ Block *s;
+ unsigned se;
+ b->lck = LBT_WR; /* hold b while checking s */
+ s = getblk( f->bt, b->nxt, i, i ? LBT_RD : lock );
+ if ( i || ! lock )
+ ULK( b );
+ else
+ b->lck = lock;
+ se = KEYPOS( f->bt, s, f->key );
+ if ( !se ) { /* lower than first */
+ ULK( s );
+ break;
+ }
+ LOG_DBG( LOG_DEBUG, "LYao %d -> %d lev %d for '%.*s' got 0x%x",
+ b->num, s->num, i, f->key->len, f->key->byt, se );
+ /* trade b for s */
+ ULK( b );
+ f->n[i] = (f->b[i] = b = s)->num;
+ e = se;
+ }
+ f->e[i] = e;
+ if ( ! i )
+ return 0;
+ if ( !(LBT_MATCH & e) ) {
+ if ( e )
+ f->e[i] = --e;
+ else {
+ TRACEKEY( ERR_TRASH, f );
+ return log_msg( ERR_IDIOT,
+ "ran on left bound of block %u ent %u",
+ b->num, b->ent );
+ }
+ }
+ d = b->dct + (LBT_ENTMASK & e);
+ /* direct access ok in inner node */
+ pkey = LBT_BASE(b)+d->pos;
+ pkln = d->kln;
+ pvln = d->vln;
+ n = GETINT( pkey+pkln+pvln );
+ i--;
+ b = getblk( f->bt, n, i, i ? LBT_RD : lock );
+ }
+ return -1; /* go away */
+} /* findkey */
+
+
+static int mktree ( Idx *bt )
+{
+ Batch *ba = bt->bat;
+ unsigned char lev = 0; /* the level to be indexed */
+ unsigned left = 1; /* block on child level */
+ /* stats */
+ unsigned size = bt->len * bt->bsz;
+ unsigned sval = ba->vals * bt->vsz;
+ unsigned sove = bt->len * LBT_DOFF + sizeof(Dict)*(ba->keys+ba->span);
+ unsigned iblk = 0;
+ unsigned ikey = 0;
+ unsigned ikeyl = 0;
+ unsigned ipfx = 0;
+ unsigned ispn = 0;
+
+ log_msg( LOG_INFO,
+ "end batch:\n"
+ "\t#blks %u total len %u free %u (%.1f%%) overhead %u (%.1f%%)\n"
+ "\t#keys %u total len %u (%.1f%%) avg %.1f max %u\n"
+ "\t#vals %u total len %u (%.1f%%) avg %.1f max %u spans %u",
+ bt->len, size, ba->free, ba->free*100./size, sove, sove*100./size,
+ ba->keys, ba->keyl, ba->keyl*100./size, (float)ba->keyl/ba->keys, ba->maxl,
+ ba->vals, sval, sval*100./size, (float)ba->vals/ba->keys, ba->maxv, ba->span
+ );
+
+ /* build tree */
+ for (;;) { /* levels */
+ unsigned len = 0; /* current is n'th block on level */
+ Block *b = 0;
+ unsigned have = 0; /* room in current block b */
+ unsigned char*base = 0; /* of current block b */
+ Dict *d = 0; /* of current block b */
+ unsigned next = left; /* child */
+
+ left = bt->len;
+ /* get child c and create child pointer in block b */
+ while ( next ) {
+ unsigned char vln = 0;
+ unsigned char *key;
+ unsigned char kln;
+ unsigned need;
+ Block *c = getblk( bt, next, lev, LBT_EX );
+ if ( ! c )
+ return log_msg( ERR_NOMEM, "no child block" );
+ next = c->nxt;
+ if ( ! c->ent ) /* OOPS empty node */
+ continue;
+ key = LBT_BASE(c)+LBT_POS(c->dct);
+ kln = c->dct[0].kln;
+ if ( ! b ) /* the leftmost child on this level */
+ kln = 0; /* always 0 key on left */
+ else if ( lev )
+ vln = LBT_VLN(c->dct);
+ else if ( ba->key.len == kln /* compare key to last of prev. */
+ && !memcmp( ba->key.byt, key, kln ) /* which was saved in ba->key */
+ ) { /* key spans blocks */
+ LOG_DBG( LOG_DEBUG, "found span key '%.*s' child %u",
+ kln, key, c->num );
+ vln = bt->vsz; /* qualify with value */
+ } else {
+#if 1
+ unsigned char x = 0;
+ /* find prefix key */
+ if ( kln > ba->key.len )
+ kln = ba->key.len;
+ while ( x < kln && key[x] == ba->key.byt[x] )
+ x++;
+ if ( x < kln ) /* they differ at [x] */
+ kln = x+1;
+ else /* equal on the shorter length */
+ kln++;
+ if ( kln > c->dct[0].kln )
+ return log_msg( ERR_TRASH,
+ "bad key '%.*s' on level %u matching previous of length %u",
+ c->dct[0].kln, key, lev, ba->key.len );
+ LOG_DBG( LOG_DEBUG, "prefix on key '%.*s' child %u save %u",
+ kln, key, c->num, c->dct[0].kln - kln );
+ ipfx += c->dct[0].kln - kln;
+#endif
+ }
+ if ( vln )
+ ispn++;
+ ikeyl += kln;
+ need = sizeof(Dict) + kln + vln + 4;
+ if ( need > have ) {
+ if ( b ) {
+ log_msg( LOG_VERBOSE,
+ "level %u block %u is full %u ent stack %u",
+ lev, b->num, b->ent, b->stk );
+ b->nxt = bt->len; /* link */
+ b->lck = 0;
+ if ( WRULK( b, bt ) ) goto nowrite;
+ len++;
+ }
+ c->lck = LBT_EX; /* paranoia */
+ b = NEWBLK( bt, lev+1, LBT_EX );
+ c->lck = 0;
+ if ( ! b )
+ return log_msg( ERR_NOMEM, "no free block" );
+ if ( b->ent || b->stk != bt->bsz )
+ return log_msg( ERR_TRASH,
+ "new block %u is not empty: ent %u stk %u",
+ b->num, b->ent, b->stk );
+ base = LBT_BASE( b );
+ b->lck = LBT_EX; /* lock */
+ have = bt->bsz - LBT_DOFF; /* - ba->let; TODO: should we ??? */
+ d = b->dct;
+ len++;
+ iblk++;
+ }
+ SETINT( base+(b->stk -= 4), c->num );
+ d->kln = kln;
+ d->vln = vln;
+ if ( kln += vln ) /* key + consecutive qualifying value */
+ memcpy( base+(b->stk -= kln), key, kln );
+ d->pos = b->stk;
+ d++;
+ b->ent++;
+ ikey++;
+ have -= need;
+ /* save the last dance for me */
+ memcpy( ba->key.byt, LBT_BASE(c)+LBT_POS(&c->dct[c->ent-1]),
+ ba->key.len = c->dct[c->ent-1].kln );
+ }
+ if ( !len )
+ return log_msg( ERR_TRASH, "level %u was empty", lev );
+ /* close level */
+ log_msg( LOG_INFO, "closing level %u with %u blocks %u..%u",
+ lev, len, left, b->num );
+ b->lck = 0;
+ if ( 1 == len ) {
+ memcpy( bt->root, b, LBT_BOFF + bt->bsz );
+ b = bt->root;
+ b->num = 0;
+ }
+ if ( WRULK( b, bt ) ) goto nowrite;
+ lev++;
+ if ( 2 > len )
+ break;
+ }
+ log_msg( LOG_INFO,
+ "\tused %u inner blocks %u keys %u bytes avg %.1f\n"
+ "\t%u #spans (%.1f%%) %u pfx (- %.1f%%)",
+ iblk, ikey, ikeyl, (float)ikeyl/ikey,
+ ispn, ispn*100.*bt->vsz/ikeyl, ipfx, ipfx*100./(ipfx+ikeyl) );
+
+ mFree( bt->bat );
+ bt->bat = 0;
+ return 0;
+nowrite:
+ return log_msg( LOG_SYSERR, "\twhen writing batch block" );
+} /* mktree */
+
+
+
+/* ************************************************************
+ package data
+*/
+
+
+
+/* ************************************************************
+ package functions
+*/
+
+int lbt_batch ( Idx *bt, unsigned char pctfree )
+{
+ bt->bat = (Batch*)mAlloc( sizeof(Batch) );
+ if ( ! bt->bat )
+ return log_msg( ERR_NOMEM, "\twhen getting batch" );
+ memset( bt->bat, 0, sizeof(Batch) );
+ if ( pctfree > 50 )
+ pctfree = 50;
+ bt->bat->let = pctfree * bt->bsz / 100;
+ lio_trunc( &bt->fd, 0 );
+ iniblk( bt->root, bt, 0, 0 );
+ /* TODO: kill hash */
+ bt->len = 1;
+ bt->bat->cur = NEWBLK( bt, 0, LBT_EX );
+ return 0;
+} /* lbt_batch */
+
+
+int lbt_batchval ( Idx *bt, Key *key )
+{
+ int cmp = 0;
+ Batch *ba = bt->bat;
+ if ( key != &ba->key ) {
+ int l = key->len < ba->key.len ? key->len : ba->key.len; /* min */
+ if ( ! (cmp = memcmp(key->byt, ba->key.byt, l)) )
+ cmp = (int)key->len - (int)ba->key.len;
+ if ( 0 > cmp )
+ return log_msg( LOG_ERROR, "batch key '%.*s' < last '%.*s'",
+ key->len, key->byt, ba->key.len, ba->key.byt );
+ }
+ /* flush ? */
+ while ( ! key->val.len /* end */
+ || LBT_ENTMAX == ba->got /* full */
+ || cmp /* next key */
+ ) {
+ Block *b = ba->cur;
+ unsigned have = b->stk - LBT_DOFF - b->ent*sizeof(Dict) - ba->let;
+ unsigned need = sizeof(Dict) + ba->key.len;
+ unsigned want = need + ba->got * bt->vsz;
+ unsigned take = ba->got;
+ if ( have < want ) { /* you can't always get ... */
+ if ( have <= need )
+ take = 0;
+ else {
+ take = (have - need) / bt->vsz;
+ if ( 256 > take * bt->vsz )
+ take = 0;
+ }
+ }
+#if 0 /* prefix key turbo ??? -- doesn't seem to work all too well */
+ if ( have < (bt->bsz >> 1) ) { /* ready to waste half a block ? */
+ Dict *d = &b->dct[ b->ent-1 ];
+ unsigned char *last = LBT_BASE(b)+LBT_POS(d);
+ if ( ba->key.byt[0] != last[0]
+ || ba->key.byt[1] != last[1]
+ || ba->key.byt[2] != last[2]
+ )
+ take = 0;
+ }
+#endif
+ LOG_DBG( LOG_DEBUG, "batch '%.*s' got %u need %u want %u have %u take %u",
+ ba->key.len, ba->key.byt, ba->got, need, want, have, take );
+ /* we have enough room in b for take entries */
+ if ( take ) {
+ unsigned char *base = LBT_BASE( b );
+ unsigned char *s = base + b->stk;
+ int i = take;
+ Dict *d = &b->dct[ b->ent++ ];
+ while ( i-- )
+ memcpy( s -= bt->vsz, ba->val[i].byt, bt->vsz );
+ memcpy( s -= ba->key.len, ba->key.byt, ba->key.len );
+ b->stk = (unsigned short) (s - base);
+ LBT_PV( d, b->stk, take );
+ d->kln = ba->key.len;
+ if ( ba->got > take )
+ memcpy( ba->val, ba->val+take, sizeof(ba->val[0])*(ba->got - take) );
+ ba->got -= take;
+ }
+ if ( ! ba->got )
+ break;
+ if ( take ) /* key spans blocks */
+ ba->span++;
+ ba->free += b->stk - LBT_DOFF - b->ent*sizeof(Dict); /* wasted */
+ b->nxt = bt->len; /* link */
+ LOG_DBG( LOG_DEBUG, "batch: %u entries left, getting block %u",
+ ba->got, bt->len );
+ if ( WRULK( b, bt ) )
+ return log_msg( LOG_SYSERR, "\twhen writing batch block" );
+ /* so couldn't flush all, thus we need another block */
+ b = ba->cur = NEWBLK( bt, 0, LBT_EX );
+ /* will continue on end or keychange */
+ if ( b->ent || b->stk < bt->bsz )
+ return log_msg( ERR_TRASH, "new block %u is not empty: ent %u stk %u",
+ b->num, b->ent, b->stk );
+ LOG_DBG( LOG_VERBOSE, "new block %u has %u ent %u stk, will keep %u",
+ b->num, b->ent, b->stk, ba->let );
+ }
+ if ( cmp || ! key->val.len ) /* done with key: stats */
+ if ( ba->maxv < ba->tot )
+ ba->maxv = ba->tot;
+ if ( key->val.len ) {
+ if ( cmp ) { /* set new key */
+ if ( ba->got )
+ return log_msg( ERR_IDIOT, "batch not empty on new key" );
+ ba->tot = 0;
+ memcpy( ba->key.byt, key->byt, ba->key.len = key->len );
+ ba->keys++;
+ ba->keyl += key->len;
+ if ( ba->maxl < key->len )
+ ba->maxl = key->len;
+ }
+ /* add new val */
+ /* LOG_HEX( key->val.byt, bt->vsz ); */
+ memcpy( ba->val[ ba->got++ ].byt, key->val.byt, bt->vsz );
+ ba->vals++;
+ ba->tot++;
+ } else {
+ WRULK( ba->cur, bt );
+ mktree( bt );
+ }
+
+ return 0;
+} /* lbt_batchval */
+
+
+/* TODO: this function is an awful mess and needs complete rewrite, KR
+*/
+int lbt_add ( Idx *bt, Key *key )
+{
+ int ret = 0;
+ unsigned short e, i = 0;
+ Block *b, *s = 0; /* mainblock, sibling */
+ Dict *d;
+ unsigned char *base, *sbase, *src, *v, *stk;
+ FindKey f;
+ int hadkey, split = 0, insibling = 0;
+ unsigned need;
+ PrVal pv;
+
+ f.bt = bt;
+ f.key = key;
+ if ( bt->key && key->len > bt->key )
+ key->len = bt->key;
+ if ( findkey( &f, LBT_WR ) ) {
+ if ( f.b[0] )
+ ULK( f.b[0] );
+ b = 0;
+ LOG_OTO( done, ( ERR_TRASH, "could not find key" ) );
+ }
+ e = (LBT_ENTMASK & f.e[0]);
+ b = f.b[0];
+ base = LBT_BASE(b);
+ d = b->dct + e;
+ if ( ! (hadkey = LBT_MATCH & f.e[0]) ) /* new key */
+ need = key->len + sizeof(Dict) + bt->vsz;
+ else {
+ unsigned vln = LBT_VLN(d);
+ v = base+LBT_POS(d)+d->kln;
+ if ( 1 == vln ) { /* check stopword -- value entirely zeroed */
+ int l = bt->vsz;
+ unsigned char *C = v;
+ while ( l-- && !*C++ ) /* huh huh, he said "C++" */
+ ;
+ if ( -1 == l ) {
+ LOG_DBG( LOG_VERBOSE, "key '%.*s' is a stopword", key->len, key->byt );
+ goto done;
+ }
+ }
+ i = valpos( v, vln, bt->vsz, key->val.byt );
+ if ( LBT_MATCH & i ) {
+ LOG_DBG( LOG_DEBUG,
+ "adding key '%.*s'/%s: already there pos %u",
+ key->len, key->byt, PRVAL( pv, key->val ), i & ~LBT_MATCH );
+ goto done; /* value already there */
+ }
+ need = bt->vsz;
+ }
+ LOG_DBG( LOG_DEBUG,
+ "adding key '%.*s'/%s in block %d had %d ent %d stk %d need %d",
+ key->len, key->byt, PRVAL( pv, key->val ),
+ b->num, hadkey, b->ent, b->stk, need );
+
+ if ( need > b->stk - LBT_DOFF - b->ent*sizeof(Dict) ) {
+ /* now that's bad: not enough space
+ create new block s, move last entries there
+ starting from the end, entries (including the new one) are collected,
+ until their combined size reaches more than half a block.
+ If the combined size then is more than 2/3 a block,
+ and the last entry alone takes more than 1/3 a block, it is split.
+ TODO: check existing sibling for free space
+ */
+ unsigned half = (bt->bsz - LBT_DOFF) >> 1;
+ unsigned tot = 0;
+ unsigned mv = b->ent;
+
+ LOG_DBG( LOG_DEBUG, "splitting block %d need %d have %d (stk %d ent %d)",
+ b->num, need, b->stk - LBT_DOFF - b->ent*sizeof(Dict), b->stk, b->ent );
+ if ( need > (bt->bsz - LBT_DOFF) / 3 )
+ LOG_OTO( done, (ERR_IDIOT, "OOPS! excessive need for %d bytes", need) );
+ /* figure out split position */
+ if ( !hadkey && b->ent == e ) {
+ tot = need;
+ insibling = !0;
+ }
+ d = b->dct+mv;
+ while ( d--, mv-- ) {
+ unsigned sz = sizeof(Dict)+d->kln+LBT_VLN(d)*bt->vsz;
+ if ( half > (tot += sz) ) { /* sz fits */
+ if ( e == mv && !hadkey ) {
+ /*
+ check size of new entry.
+ In case of hadkey, we don't bother for the bt->vsz bytes.
+ */
+ if ( half < (tot + need) ) {
+ if ( (half - tot) > need/2 ) { /* more balanced if we move */
+ tot += need;
+ insibling = !0;
+ }
+ break;
+ }
+ tot += need;
+ insibling = !0;
+ }
+ if ( mv )
+ continue; /* else croak on ! mv below */
+ }
+ /* exceeded half -- maybe we move entry mv anyway */
+ if ( tot > (bt->bsz - LBT_DOFF)*2/3 ) {
+ if ( tot-sz < (bt->bsz - LBT_DOFF)/3 )
+ split = !0;
+ else { /* undo -- don't move this entry */
+ tot -= sz;
+ d++; mv++;
+ }
+ } else if ( ! mv )
+ LOG_OTO( done, (ERR_IDIOT, "OOPS! didn't find end tot %d bytes", tot) );
+ break;
+ } /* while mv-- */
+ if ( mv < e || (mv == e && hadkey && !split) )
+ insibling = !0;
+ /* now mv is the first entry to move */
+ s = NEWBLK( bt, 0, LBT_WR );
+ LOG_DBG( LOG_DEBUG,
+ "split: move %d of %d to new blk %d e %d tot %d insibling %d split %d",
+ mv, b->ent, s->num, e, tot, insibling, split );
+ s->nxt = b->nxt;
+ b->nxt = s->num;
+ sbase = LBT_BASE(s);
+ if ( split ) {
+ /* now total includes full size of entry mv.
+ if we included a new entry, it might even exceed bsz.
+ calculate sizes for both blocks w/o mv's values.
+ */
+ unsigned esz = sizeof(Dict)+d->kln; /* entry's base cost */
+ unsigned vln = LBT_VLN(d);
+ unsigned keep, diff, mvval, kpval, mvsiz, pos = LBT_POS(d);
+ /*
+ from currently used space, substract tot (including all
+ values and possibly new key "need"), but add entry base cost.
+ that should be the size remaining if we where moving
+ all of the values but keep an entry and key.
+ maybe slightly biased if need is bt->vsz (the hadkey case).
+ */
+ keep = bt->bsz - b->stk + b->ent*sizeof(Dict) + need - tot + esz;
+ tot -= vln * bt->vsz; /* undo entries */
+ diff = (keep > tot ? keep - tot : tot - keep) / bt->vsz;
+ if ( diff > vln )
+ LOG_OTO( done, (ERR_IDIOT,
+ "OOPS! got diff %d vln %d entry %d tot %d keep %d",
+ diff, vln, mv, tot, keep) );
+ mvval = (vln - diff)/2;
+ if ( keep > tot )
+ mvval += diff;
+ kpval = vln - mvval;
+ if ( ! mvval || ! kpval ) {
+ LOG_DBG( LOG_WARN, "bad mv/kpval %d/%d keep %d tot %d vln %d diff %d",
+ mvval, kpval, keep, tot, vln, diff );
+ if ( ! mvval )
+ kpval = vln - (mvval = 1);
+ else
+ mvval = vln - (kpval = 1);
+ }
+ tot += mvsiz = mvval*bt->vsz;
+ keep += kpval*bt->vsz;
+ src = base+pos;
+ LOG_DBG( LOG_DEBUG,
+ "split: entry %d '%.*s' mv %d of %d values, i %d keep %d tot %d",
+ mv, d->kln, src, mvval, vln, i, keep, tot );
+ if ( keep > (bt->bsz-LBT_DOFF) || tot > (bt->bsz-LBT_DOFF) )
+ LOG_OTO( done, (ERR_IDIOT,
+ "OOPS! got diff %d vln %d mvval %d entry %d tot %d keep %d",
+ diff, vln, mvval, mv, tot, keep) );
+ /* SIGH -- do it */
+ memcpy( sbase + (s->stk -= mvsiz),
+ src+d->kln+kpval*bt->vsz, mvsiz );
+ if ( e == mv && hadkey && i > kpval ) {
+ insibling = !0;
+ s->stk -= bt->vsz;
+ i -= kpval;
+ memmove( sbase+s->stk, sbase+s->stk+bt->vsz, i*bt->vsz );
+ memcpy( sbase+s->stk+i*bt->vsz, key->val.byt, bt->vsz );
+ mvval++;
+ }
+ memcpy( sbase + (s->stk -= d->kln), src, d->kln );
+ LBT_PVK( s->dct, s->stk, mvval, d->kln );
+ s->ent = 1;
+ memmove( src + mvsiz, src, d->kln+kpval*bt->vsz );
+ pos += mvsiz;
+ LBT_PV( d, pos, kpval );
+ d++; mv++;
+ }
+ /* now entries mv..b->ent are moved, mv > 0 */
+ b->stk = LBT_POS(b->dct+mv-1); /* pre-adjust stack */
+ if ( mv < e ) { /* move entries mv..e-1 */
+ /* used stack space is from pos of entry e-1 to end of entry mv
+ (might differ from pos of mv-1 due to split)
+ */
+ unsigned from = LBT_POS(b->dct+e-1), nent = e-mv,
+ sz = LBT_POS(b->dct+mv)+b->dct[mv].kln
+ +LBT_VLN(b->dct+mv)*bt->vsz - from;
+ int adj = (s->stk -= sz) - from; /* entry position adjustment */
+ memcpy( sbase + s->stk, base+from, sz );
+ memcpy( s->dct+s->ent, b->dct+mv, nent*sizeof(Dict) );
+ d = s->dct + s->ent;
+ s->ent += nent;
+ for ( ; nent--; d++ )
+ LBT_POSADD( d, adj );
+ }
+ if ( insibling && mv <= e ) { /* now mv or insert e, unless done on split */
+ d = b->dct + e;
+ if ( hadkey ) { /* mv existing entry e >= mv, add new val */
+ /* i is valpos */
+ unsigned vln = LBT_VLN(d);
+ src = base+LBT_POS(d)+d->kln;
+ if ( i < vln ) {
+ unsigned sz = (vln - i)*bt->vsz;
+ memcpy( sbase + (s->stk -= sz), src + i*bt->vsz, sz );
+ }
+ memcpy( sbase + (s->stk -= bt->vsz), key->val.byt, bt->vsz );
+ if ( i ) {
+ unsigned sz = i*bt->vsz;
+ memcpy( sbase + (s->stk -= sz), src, sz );
+ }
+ memcpy( sbase + (s->stk -= d->kln), base+LBT_POS(d), d->kln );
+ LBT_PVK( s->dct + s->ent, s->stk, vln+1, d->kln );
+ e++; /* e is moved */
+ } else {
+ memcpy( sbase + (s->stk -= bt->vsz), key->val.byt, bt->vsz );
+ memcpy( sbase + (s->stk -= key->len), key->byt, key->len );
+ LBT_PVK( s->dct + s->ent, s->stk, 1, key->len );
+ }
+ s->ent++;
+ } /* mv or insert e */
+ if ( e < b->ent && mv < b->ent ) { /* move entries max(e,mv) to end */
+ unsigned fst = mv < e ? e : mv;
+ /* used stack space is from pos of entry b->ent-1 (the original b->stk)
+ to end of entry fst (might differ from pos of fst-1 due to split)
+ */
+ unsigned from = LBT_POS(b->dct+b->ent-1), nent = b->ent-fst,
+ sz = LBT_POS(b->dct+fst)+b->dct[fst].kln
+ +LBT_VLN(b->dct+fst)*bt->vsz - from;
+ int adj = (s->stk -= sz) - from; /* entry position adjustment */
+ LOG_DBG( LOG_DEBUG, "split: entries %d..%d sz %d from %d to %d",
+ fst, b->ent-1, sz, from, s->stk );
+ memcpy( sbase + s->stk, base+from, sz );
+ memcpy( s->dct+s->ent, b->dct+fst, nent*sizeof(Dict) );
+ d = s->dct + s->ent;
+ s->ent += nent;
+ for ( ; nent--; d++ ) {
+#if !defined(NDEBUG)
+ unsigned p = LBT_POS(d);
+#endif
+ LBT_POSADD( d, adj );
+#if !defined(NDEBUG)
+ LOG_DBG( LOG_DEBUG, "new ent %d p/v/k %d/%d/%d o %d",
+ d - s->dct, LBT_POS(d), LBT_VLN(d), d->kln, p );
+#endif
+ }
+ }
+ b->ent = mv; /* kill entries */
+ LOG_DBG( LOG_DEBUG,
+ "split: blks %d/%d with %d/%d entries, s starts '%.*s'",
+ b->num, s->num, b->ent, s->ent, s->dct[0].kln, sbase+LBT_POS(s->dct) );
+ } /* new sibling */
+
+ if ( ! insibling ) {
+ /* now we do have enough space */
+ d = b->dct + e;
+ stk = base + b->stk; /* bottom of data stack */
+ if ( hadkey ) { /* insert value at i */
+ unsigned char *val = base + LBT_POS(d)+d->kln+i*bt->vsz;
+ need = bt->vsz;
+#if 0
+ LOG_DBG( LOG_DEBUG,
+ "in entry %d '%.*s' val %d: memmove( %p %p %d) val %p need %d",
+ e, d->kln, base+LBT_POS(d), i,
+ stk-need, stk, (val-stk)+need, val, need );
+#endif
+ memmove( stk - need, stk, (val - stk)+need );
+ memcpy( val - need, key->val.byt, bt->vsz );
+ LBT_VLNADD( d, 1 );
+ } else { /* insert key, value and dict entry */
+ /* take dict and data space from entry e */
+ unsigned char *nu;
+ need = key->len + bt->vsz;
+ if ( e == b->ent )
+ nu = base + b->stk - need;
+ else {
+ /* at end of next entry, i.e. end of block or beg of previous */
+ nu = base + LBT_POS(d) + d->kln + LBT_VLN(d)*bt->vsz;
+ memmove( stk - need, stk, nu - stk );
+ nu -= need;
+ memmove( d+1, d, (b->ent - e)*sizeof(Dict) );
+ }
+ memcpy( nu, key->byt, key->len );
+ memcpy( nu + key->len, key->val.byt, bt->vsz );
+ LBT_PV( d, nu-base, 1 );
+ d->kln = key->len;
+ d++;
+ e++;
+ b->ent++;
+ }
+ /*
+ need is set to the needed (i.e. additionally used) data space,
+ e is the first dict entry to redirect need bytes lower
+ */
+ b->stk -= need;
+ for ( ; e < b->ent; e++, d++ )
+ LBT_POSADD( d, -need );
+ } /* ! insibling */
+ while ( s ) { /* created sibling */
+ unsigned num = s->num;
+ unsigned char *e0 = LBT_BASE(s) + LBT_POS(s->dct);
+ unsigned l = s->lev + 1;
+ Block *ins;
+ Key k;
+
+ memcpy( k.byt, e0, k.len = s->dct[0].kln );
+ k.val.len = 0;
+ if ( s->lev ? s->dct[0].vln : split )
+ memcpy( k.val.byt, e0+k.len, k.val.len = bt->vsz );
+ LOG_DBG( LOG_DEBUG, "split: had %d/%d on lev %d key '%.*s' vln %d %s",
+ b->num, num, s->lev, k.len, k.byt, k.val.len, PRVAL( pv, k.val ) );
+
+ /* get parent */
+ if ( f.b[l]->num == f.n[l] && ! LBT_BUSY(f.b[l]) )
+ f.b[l]->lck = LBT_WR;
+ else
+ f.b[l] = getblk( bt, f.n[l], l, LBT_WR );
+ LOG_DBG( LOG_DEBUG,
+ "split: got parent %d lev %d", f.b[l]->num, f.b[l]->lev );
+ /* MT TODO: again L/Y link hunt */
+ /* ok, locked the parent -- release s and b */
+ if ( (ret = WRULK( s, bt )) || (ret = WRULK( b, bt )) )
+ break;
+ s = 0;
+ ins = b = f.b[l];
+ base = LBT_BASE(b);
+ e = KEYPOS( bt, b, &k );
+ LOG_DBG( LOG_DEBUG, "split: got parent pos %d", e );
+ if ( LBT_MATCH & e ) /* !??? */
+ LOG_OTO( done, (ERR_TRASH, "OOPS! found key in parent" ) );
+ if ( ! e ) {
+ TRACEKEY( ERR_TRASH, &f );
+ LOG_OTO( done, (ERR_TRASH, "OOPS! parent insert at 0 lev %d", l ) );
+ }
+ need = sizeof(Dict) + k.len + 4 + k.val.len;
+ if ( need > b->stk - LBT_DOFF - b->ent*sizeof(Dict) ) {
+ unsigned half = ((bt->bsz - b->stk)+b->ent*sizeof(Dict)+need) / 2;
+ unsigned mv = b->ent, tot = 0, nent, sz;
+ int adj;
+ if ( e == b->ent ) {
+ tot = need;
+ ins = 0;
+ }
+ for ( d = b->dct+mv; d--, mv--; ) {
+ sz = sizeof(Dict) + d->kln + 4 + d->vln;
+ if ( half > (tot += sz) ) {
+ if ( e == mv ) {
+ if ( half > tot+need || tot+need-half < half-tot ) {
+ tot += need;
+ ins = 0; /* new link goes to sibling */
+ } else
+ break;
+ if ( half <= tot )
+ break;
+ }
+ if ( mv )
+ continue;
+ }
+ if ( tot - half > sz/2 ) { /* don't move this */
+ tot -= sz;
+ d++; mv++;
+ } else if ( ! mv ) /* !??? */
+ mv = 1;
+ break;
+ }
+ /* split here a little bit simpler */
+ s = NEWBLK( bt, l, LBT_WR );
+ s->nxt = b->nxt;
+ b->nxt = s->num;
+ /* straight move entries mv..b->ent-1 to sibling */
+ sbase = LBT_BASE(s);
+ d = b->dct + mv;
+ sz = b->dct[mv-1].pos - b->stk;
+ nent = b->ent - mv;
+ LOG_DBG( LOG_DEBUG,
+ "split: parent lev %d split %d to new %d mv %d"
+ " '%.*s'/%s nent %d sz %d",
+ l, b->num, s->num, mv, d->kln, base+d->pos,
+ prval( pv, base+d->pos+d->kln, d->vln ), nent, sz );
+ memcpy( sbase + (s->stk -= sz), base + b->stk, sz );
+ adj = s->stk - b->stk;
+ b->stk += sz;
+ memcpy( s->dct, d, nent*sizeof(Dict) );
+ b->ent = mv;
+ s->ent = nent;
+ for ( d = s->dct; nent--; d++ )
+ d->pos += adj;
+ if ( ! ins ) {
+ ins = s;
+ e -= mv;
+ }
+ if ( (ret = chkblk( s, bt )) )
+ break;
+ }
+ /* now insert link in block ins at pos e */ {
+ unsigned isz = k.len + 4 + k.val.len, nent = ins->ent - e;
+ unsigned char *ibase = LBT_BASE(ins);
+ unsigned end = e ? ins->dct[e-1].pos : bt->bsz;
+ LOG_DBG( LOG_DEBUG,
+ "split: ins parent %d lev %d pos %d key '%.*s' -> %d vln %d %s",
+ ins->num, ins->lev, e, k.len, k.byt, num, k.val.len,
+ PRVAL( pv, k.val ) );
+ memmove( ibase + ins->stk - isz, ibase + ins->stk, end - ins->stk );
+ ins->stk -= isz;
+ memcpy( ibase + (end -= 4), &num, 4 );
+ if ( k.val.len )
+ memcpy( ibase + (end -= k.val.len), k.val.byt, k.val.len );
+ memcpy( ibase + (end -= k.len), k.byt, k.len );
+ for ( d = ins->dct+ins->ent; nent--; d-- ) {
+ *d = d[-1];
+ d->pos -= isz;
+ }
+ d->pos = end;
+ d->vln = k.val.len;
+ d->kln = k.len;
+ ins->ent++;
+ if ( (ret = chkblk( ins, bt )) )
+ break;
+ }
+ if ( ! s ) /* write b below or in next turn */
+ break;
+ if ( ! b->num ) { /* root split */
+ Block *bb = NEWBLK( bt, l, LBT_WR );
+ LOG_DBG( LOG_DEBUG, "root split!" );
+ /* cp all but the num */
+ memcpy( LBT_BASE(bb)+4, LBT_BASE(b)+4, bt->bsz-4 );
+ b->nxt = s->nxt; /* restore root's special nxt */
+ s->nxt = 0;
+ b->stk = bt->bsz; /* empty the root */
+ /* add two childs bb and s */
+ SETINT( base + (b->stk -= 4), bb->num );
+ LBT_PVK( b->dct, b->stk, 0, 0 );
+ SETINT( base + (b->stk -= 4), s->num );
+ e0 = LBT_BASE(s) + LBT_POS(s->dct);
+ if ( s->dct[0].vln )
+ memcpy( base + (b->stk -= bt->vsz), e0+s->dct[0].kln, bt->vsz );
+ memcpy( base + (b->stk -= s->dct[0].kln), e0, s->dct[0].kln );
+ LBT_PVK( b->dct+1, b->stk, s->dct[0].vln, s->dct[0].kln );
+ b->ent = 2;
+ b->lev++; /* increase tree depth */
+ if ( (ret = WRULK( s, bt )) || (ret = WRULK( bb, bt )) )
+ break;
+ s = 0;
+ break;
+ }
+ }
+ if ( b && !ret )
+ ret = WRULK( b, bt );
+done:
+ if ( s )
+ ULK( s );
+ if ( b )
+ ULK( b );
+ if ( !ret && b->lev && findkey( &f, LBT_RD ) )
+ ret = log_msg( ERR_TRASH, "could not find key after add" );
+ return ret;
+} /* lbt_add */
+
+
+int lbt_del ( Idx *bt, Key *key )
+{
+ int ret = 0;
+ unsigned short e, n, i, mv;
+ Block *b;
+ Dict *d;
+ unsigned char *base, *v;
+ FindKey f;
+ PrVal pv;
+
+ f.bt = bt;
+ f.key = key;
+ LOG_DBG( LOG_DEBUG, "deleting key '%.*s'/%s",
+ key->len, key->byt, PRVAL( pv, key->val ) );
+ if ( findkey( &f, LBT_WR ) )
+ LOG_OTO( done, ( ERR_TRASH, "could not find key" ) );
+ if ( !(LBT_MATCH & f.e[0]) ) /* key not found - ok */
+ goto done;
+ LOG_DBG( LOG_DEBUG, "key found" );
+ e = (LBT_ENTMASK & f.e[0]);
+ b = f.b[0];
+ d = b->dct + e;
+ base = LBT_BASE(b);
+ v = base+LBT_POS(d)+d->kln; /* start of values */
+ n = LBT_VLN(d);
+ if ( ! n )
+ goto done;
+ i = valpos( v, n, bt->vsz, key->val.byt );
+ LOG_DBG( LOG_DEBUG, "val %sfound pos %u of %u",
+ (LBT_MATCH & i) ? "" : "not ", i & ~LBT_MATCH, n );
+ if ( !(LBT_MATCH & i) ) /* val not found - ok */
+ goto done;
+ i &= ~LBT_MATCH;
+ LOG_DBG( LOG_DEBUG,
+ "del blk %d e %d i/n %d/%d", b->num, e, i, n );
+ if ( 1 < n ) {
+ v += i*bt->vsz; /* end of stack area to move: start of this value */
+ mv = bt->vsz; /* amount to move: one value size */
+ LBT_VLNADD( d, -1 );
+ } else { /* sole value -- rm whole entry */
+ v = base+LBT_POS(d); /* start of key */
+ mv = d->kln+bt->vsz;
+ for ( i=b->ent - e - 1; i--; d++ ) /* move the i entries after d down */
+ *d = d[1];
+ d = b->dct + e; /* reset */
+ b->ent--;
+ }
+ memmove( base+b->stk+mv, base+b->stk, v-(base+b->stk) );
+ b->stk += mv;
+ for ( ; e++ < b->ent; d++ )
+ LBT_POSADD( d, mv );
+ WRULK( b, bt );
+done:
+ if ( f.b[0] )
+ ULK( f.b[0] );
+ return ret;
+} /* lbt_del */
+
+
+
+int lbt_loop ( Idx *bt, DXLoop *l )
+{
+ int ret = 0;
+ unsigned i = 0;
+ int mode = IDXMODE & l->flg;
+ Block *b = 0;
+ Key *cmp = &l->key;
+ if ( ! cmp->len ) /* start at 1st leaf */
+ b = getblk( bt, 1, 0, LBT_RD );
+ else { /* locate key */
+ FindKey f;
+ f.bt = bt;
+ f.key = cmp;
+ if ( findkey( &f, LBT_RD ) )
+ LOG_OTO( done, ( ERR_TRASH, "could not find key" ) );
+ b = f.b[0];
+ i = (unsigned)(LBT_ENTMASK & f.e[0]);
+ if ( !(LBT_MATCH & f.e[0]) && IDXEQ == mode )
+ goto done;
+ }
+ if ( IDXUPTO <= mode )
+ cmp = &l->to;
+ for ( ;; i=0 ) { /* blocks */
+ unsigned char *base = LBT_BASE( b );
+ Dict *d;
+ if ( ! b || b->lev )
+ return -ERR_TRASH;
+ b->lck = LBT_EX; /* lock */
+ d = b->dct + i;
+ for ( ; i<b->ent; i++, d++ ) {
+ unsigned short pos = LBT_POS( d );
+ unsigned short vln = LBT_VLN( d );
+ unsigned short kln = d->kln;
+ unsigned short j;
+ unsigned char *v = base+pos+kln;
+ Key key;
+ Hit hit;
+ int diff = memcmp( cmp->byt, base+pos,
+ kln <= cmp->len ? kln : cmp->len );
+ switch ( mode ) {
+ case IDXPF:
+ if ( diff || kln < cmp->len )
+ goto moan;
+ break;
+ case IDXEQ:
+ if ( diff || kln != cmp->len )
+ goto done;
+ break;
+ case IDXUPTO:
+ if ( 0 >= diff )
+ goto done;
+ break;
+ case IDXINCL:
+ if ( 0 > diff || (!diff && kln > cmp->len) )
+ goto done;
+ break;
+ default: /* unused */
+ moan:
+ log_msg( LOG_INFO,
+ "loop ends on key '%.*s'(%d) %d %d",
+ kln, base+pos, kln, diff, cmp->len );
+ goto done;
+ }
+ memcpy( key.byt, base+pos, key.len = kln );
+ if ( l->cb )
+ for ( j=0; j<vln; j++, v += bt->vsz ) {
+ rdval( &hit, v, bt->vsz );
+ ret = l->cb( l->me, &key, &hit );
+ if ( ret )
+ goto done;
+ }
+ else { /* internal check */
+ FindKey f;
+ f.bt = bt;
+ f.key = &key;
+ memcpy( key.val.byt, v, key.val.len = bt->vsz );
+ if ( findkey( &f, LBT_RD ) )
+ LOG_OTO( done, ( ERR_TRASH, "could not find key" ) );
+ if ( !(LBT_MATCH & f.e[0])
+ || b != f.b[0]
+ || i != (unsigned)(LBT_ENTMASK & f.e[0])
+ ) {
+ log_msg( ERR_TRASH,
+ "OOPS! want %u(%p).%u got %u(%p).%u (match %x)",
+ b->num, b, i, f.b[0]->num, f.b[0], LBT_ENTMASK & f.e[0], f.e[0] );
+ rdval( &hit, v, bt->vsz );
+ log_msg( ERR_TRASH,
+ "\twhen looking for '%.*s'/%u.%u.%u.%u",
+ key.len, key.byt, hit.mfn, hit.tag, hit.occ, hit.pos );
+ if ( f.b[0]->ent <= (LBT_ENTMASK & f.e[0]) )
+ log_msg( ERR_TRASH, "\tI found nothing" );
+ else {
+ Dict *dd = f.b[0]->dct + (LBT_ENTMASK & f.e[0]);
+ unsigned char *bb = LBT_BASE(f.b[0])+LBT_POS(dd);
+ rdval( &hit, bb+dd->kln, bt->vsz );
+ log_msg( ERR_TRASH,
+ "\tI found '%.*s'/%u.%u.%u.%u",
+ dd->kln, bb, hit.mfn, hit.tag, hit.occ, hit.pos );
+ }
+ } else
+ log_msg( LOG_TRACE, "key ok" );
+ }
+ }
+ ULK( b );
+ if ( ! b->nxt || !(b = getblk(bt, b->nxt, 0, LBT_RD)) )
+ break;
+ }
+done:
+ if ( b )
+ ULK( b );
+ if ( l->cb )
+ l->cb( l->me, 0, 0 );
+ else
+ log_msg( LOG_INFO,
+ "checked %i keys in %u blks of %dK depth %u cmp/get/miss %u/%u/%u",
+ stat.key, bt->len, bt->bsz>>10, bt->root->lev, stat.cmp, stat.get, stat.miss );
+ return ret;
+} /* lbt_loop */
+
+
+/* mimik the plain old ldb_search */
+int lbt_search ( Idx *bt, Key *key, LdbPost *post, Rec *rec )
+{
+ int ret = 0;
+ unsigned i = 0;
+ int cont = rec && rec->len; /* continuing previous terms search */
+ char *stk = rec ? ((char*)rec + rec->bytes) : 0; /* term stack */
+ int prefix;
+ Block *b = 0;
+ Key start;
+ LdbP * const p = post->p; /* really just an alias */
+
+ /* check for prefix match */
+ if ( post ) {
+ if ( 8 != bt->vsz )
+ return log_msg( ERR_INVAL, "bad legacy search on vsz %d", bt->vsz );
+ prefix = LDB_PFX & post->mode;
+ } else {
+ if ( ! rec )
+ return log_msg( ERR_INVAL, "lbt_search needs rec or post" );
+ if ( (prefix = key->len && '$' == key->byt[key->len - 1]) )
+ key->len--;
+ }
+
+ if ( ! cont )
+ start = *key;
+ else {
+ memset( &start, 0, sizeof(start) );
+ start.len = 255 < rec->field[rec->len-1].len
+ ? 255 : rec->field[rec->len-1].len;
+ memcpy( start.byt, rec->field[rec->len-1].val, start.len );
+ rec->len = 0;
+ }
+
+ if ( ! start.len ) /* start at 1st leaf */
+ b = getblk( bt, 1, 0, LBT_RD );
+ else { /* locate key */
+ FindKey f;
+ f.bt = bt;
+ f.key = &start;
+ if ( findkey( &f, LBT_RD ) )
+ LOG_OTO( ouch, ( ERR_TRASH, "could not find key" ) );
+ b = f.b[0];
+ i = (unsigned)(LBT_ENTMASK & f.e[0]);
+ if ( !(LBT_MATCH & f.e[0]) && ! prefix )
+ goto done;
+ }
+
+ for ( ;; i=0 ) { /* blocks */
+ unsigned char *base = LBT_BASE( b );
+ Dict *d;
+ if ( ! b || b->lev )
+ return -ERR_TRASH;
+ b->lck = LBT_EX; /* lock */
+ d = b->dct + i;
+ for ( ; i<b->ent; i++, d++ ) { /* entries */
+ unsigned char *ent = base + LBT_POS( d ); /* start of key */
+ unsigned short kln = d->kln;
+ if ( memcmp( key->byt, ent, kln <= key->len ? kln : key->len )
+ || (prefix ? kln < key->len : kln != key->len )
+ )
+ goto done;
+ if ( rec ) { /* record term */
+ Field *f = rec->field + rec->len; /* field to assign */
+ /** append term to record, unless it's the same as previous */
+ if ( rec->len ) {
+ if ( kln == f[-1].len && !memcmp( f[-1].val, ent, kln ) )
+ continue;
+ } else if ( cont ) {
+ if ( kln == start.len && !memcmp( start.byt, ent, kln ) )
+ continue;
+ }
+ stk -= kln;
+ if ( stk < (char*)(f+1) ) /* stack reached field dict */
+ goto done;
+ memcpy( stk, ent, kln );
+ f->tag = 0;
+ f->val = stk;
+ f->len = kln;
+ rec->len++;
+ continue;
+ }
+ { /* more locals */
+ LdbP merge[1024];
+ unsigned short vln, k, m=0;
+ unsigned char *c;
+ int f = post->fil-1; /* highest pos to consider in given postings */
+ if ( !(vln = LBT_VLN( d )) )
+ continue;
+ ent += kln; /* set to start of values */
+ c = ent + 8*(vln-1); /* last value */
+ /* collect postings */
+ for ( k=vln; k--; c-=8 ) {
+ /* loop backwards (for the fun of it) postings in segment */
+ int prow, ptag, ppos;
+ LdbP e; /* the entry */
+ LdbP samerow; /* highest possible entry w/ same row as e */
+#if defined( LDB_BIG_ENDIAN )
+ /* the 8 bytes of a posting are BIG ENDIAN ! */
+ memcpy(e.bytes,c,8);
+#else
+ e.bytes[0] = c[7]; e.bytes[1] = c[6];
+ e.bytes[2] = c[5]; e.bytes[3] = c[4];
+ e.bytes[4] = c[3]; e.bytes[5] = c[2];
+ e.bytes[6] = c[1]; e.bytes[7] = c[0];
+#endif
+ prow = LDBP_ROW( &e );
+ ptag = LDBP_TAG( &e );
+ ppos = LDBP_POS( &e );
+ LOG_DBG( LOG_VERBOSE, "post %d.%hd pos %06x", prow, ptag, ppos );
+ if ( !prow )
+ continue;
+ if ( (post->cut && prow >= post->cut)
+ || (post->tag && post->tag != ptag)
+ )
+ continue;
+ if ( prow < post->skp ) /* quickly bail out on skip condition */
+ break;
+ LDBP_SETROWTOP( &samerow, &e ); /* for mfn comparison */
+ /* sweep down to postings for the same row as e ... */
+ while ( f >= 0 && LDBP_GT( p+f, &samerow ) )
+ f--;
+ if ( LDB_AND & post->mode ) {
+ int l;
+ /* loop postings for same row, mark all (that are near enough) */
+ LDBP_SETROWBOT( &samerow, &e ); /* for mfn comparison */
+ /* NOTE: postings for row are GT than bottom even if marked */
+ for ( l = f; l>=0 && LDBP_GT( p+l, &samerow ); l-- ) {
+ if ( post->near ) {
+ int dist;
+ if ( ptag != LDBP_TAG( p+l ) ) continue;
+ if ( LDB_NEAR_G != post->near ) {
+ dist = LDBP_POS( p+l ) - LDBP_POS( &e );
+ if ( dist < 0 ) dist = -dist;
+ if ( 0 < post->near
+ ? post->near < dist
+ : -post->near != dist /* exact $$$$ */
+ ) continue;
+ }
+ }
+ LDBP_SETMARK( p+l );
+ }
+ } else { /* OR mode */
+ int add;
+ if ( ! post->near ) /* add if row not found: ignore details */
+ add = 0 > f || prow > LDBP_ROW( p+f );
+ else { /* add if no exact match */
+ int l;
+ /* NOTE: we don't use mark bit in OR mode, do we ? */
+ for ( l = f; l>=0 && LDBP_GT( p+l, &e ); l-- )
+ ;
+ add = 0 > l || LDBP_GT( &e, p+l );
+ }
+ if ( add )
+ merge[ m++ ] = e;
+ }
+ } /* for postings in segment */
+ if ( m ) { /* merge in the merge buffer */
+ LdbP *mm = merge;
+ for ( k = post->fil += m; m && k--; ) {
+ LdbP src;
+ if ( k < m || LDBP_GT( mm, &p[k-m] ) ) {
+ src = *mm++;
+ m--;
+ LOG_DBG( LOG_DEBUG, "merging %d at %d", LDBP_ROW(&src), k );
+ } else
+ src = p[k-m];
+ if ( k < post->len )
+ p[k] = src;
+ else { /* set cut */
+ int row = LDBP_ROW( &src );
+ if ( row < post->cut || !post->cut )
+ post->cut = row;
+ }
+ }
+ if ( post->fil > post->len )
+ post->fil = post->len;
+ if ( post->cut ) /* postings for cut row are unreliable */
+ while ( post->fil && post->cut <= LDBP_ROW(p+post->fil-1) )
+ post->fil--;
+ } /* if ( m ) */
+ } /* more locals */
+ } /* for entries in block */
+ ULK( b );
+ if ( ! b->nxt || !(b = getblk(bt, b->nxt, 0, LBT_RD)) )
+ break;
+ }
+done:
+ if ( rec )
+ ret = rec->len;
+ouch:
+ if ( b )
+ ULK( b );
+ return ret;
+} /* lbt_search */
+
+
+int lbt_init ( Idx *bt )
+{
+ int ret = 0, blkbits, i;
+ Block *l;
+ int size = lio_size( bt->fd );
+
+ bt->root = 0;
+ bt->hash = 0;
+ bt->mem = 0;
+ for ( i=LBT_LRULEV; i--; )
+ bt->lru[i] = bt->mru[i] = 0;
+
+ if ( 0x3ff & size ) /* not on 1K boundary */
+ return log_msg( LOG_ERROR, "bt has bad size %dK + %d",
+ size >> 10, (int)size & 0x3ff );
+ if ( size ) { /* read root header */
+ Block b;
+ unsigned want = sizeof(b) - LBT_BOFF;
+ int got = lio_pread( &bt->fd, LBT_BASE(&b), want, 0 );
+ if ( want != (unsigned)got )
+ return log_msg( LOG_SYSERR, "could not get root head" );
+ if ( b.num )
+ return log_msg( ERR_TRASH, "root has num %d", b.num );
+ /*if ( LBT_ISIS != b.nxt ) log_msg( LOG_WARN, "magic is 0x%08x", b.nxt );*/
+ if ( bt->typ && bt->typ != b.typ )
+ log_msg( LOG_WARN, "typ want 0x02%x got 0x02%x", bt->typ, b.typ );
+ bt->typ = b.typ;
+ if ( bt->key && bt->key != b.key )
+ log_msg( LOG_WARN, "key want %d got %d", bt->key, b.key );
+ bt->key = b.key;
+ if ( bt->col && bt->col != b.col )
+ log_msg( LOG_WARN, "col want %d got %d", bt->col, b.col );
+ bt->col = b.col;
+ bt->dpt = b.lev;
+ /*
+ b.ent
+ b.stk
+ */
+ } else { /* new index */
+ bt->dpt = 1;
+ if ( ! bt->key )
+ bt->key = 30;
+ }
+
+ bt->vsz = 8+(0xf & bt->typ);
+ blkbits = 10 + (3 & (bt->typ >> 4));
+ bt->bsz = 1 << blkbits;
+ if ( size % bt->bsz )
+ return log_msg( ERR_TRASH, "size 0x%08x % blksize", size, bt->bsz );
+ bt->len = size >> blkbits;
+ if ( 2 > bt->len && !(LBT_WRITE & bt->flg) )
+ return log_msg( LOG_SYSERR, "attempt to read empty index" );
+
+ bt->hlen = 16 + bt->len / 32; /* initial cache size */
+ if ( bt->hlen < 1000 ) bt->hlen = 1000;
+ if ( !(bt->root = addchunk( bt, bt->hlen )) )
+ return log_msg( LOG_SYSERR, "\twhen getting bt cache" );
+ if ( bt->hlen < 256 ) /* the hash */
+ bt->hlen = 256;
+ if ( !(bt->hash = (Block**)mAlloc( bt->hlen * sizeof(Block*) )) )
+ return log_msg( LOG_SYSERR, "\twhen getting bt hash" );
+ memset( bt->hash, 0, bt->hlen * sizeof(Block*) );
+
+ if ( size && readblk( bt->root, bt, 0 ) )
+ return log_msg( ERR_TRASH, "could not read root" );
+ if ( ! bt->len ) {
+ /* init: link root to leaf */
+ SETINT( LBT_BASE(bt->root)+(bt->root->stk -= 4), 1 );
+ LBT_PVK( bt->root->dct, bt->root->stk, 0, 0 );
+ bt->root->ent = 1;
+ bt->len = 1;
+ }
+ if ( !(l = getblk( bt, 1, 0, LBT_EX )) )
+ return log_msg( ERR_TRASH, "could not read 1st leaf" );
+ if ( ! size ) {
+ log_msg( LOG_INFO, "creating index bsz %u ksz %u vsz %u",
+ bt->bsz, bt->key, bt->vsz );
+ bt->root->lev = 1;
+ if ( WRULK( bt->root, bt ) || WRULK( l, bt ) )
+ return log_msg( LOG_SYSERR, "\twhen writing base" );
+ }
+
+ return ret;
+} /* lbt_init */
+
+
+void lbt_close ( Idx *bt )
+{
+ Chunk *c, *n = bt->mem;
+ while ( (c = n) ) {
+ n = c->nxt;
+ mFree( c );
+ }
+ mFree( bt->hash );
+ lio_close( &bt->fd, LIO_INOUT );
+ memset( bt, 0, sizeof(*bt) );
+} /* lbt_close */
+
+
+/* ************************************************************
+ public functions
+*/
+
+
+void cXMkVal ( Idx *bt, Val *val, Hit *hit )
+{
+ if ( hit ) {
+ val->len = bt->vsz;
+ mkval( val, hit );
+ } else
+ val->len = 0;
+}
+
+int cXAdd ( Idx *bt, Key *key, Hit *hit )
+{
+ if ( bt->bat ) {
+ if ( ! key )
+ key = &bt->bat->key;
+ cXMkVal( bt, &key->val, hit );
+ return lbt_batchval( bt, key );
+ } else if ( !key )
+ return 0;
+ else if ( hit ) {
+ cXMkVal( bt, &key->val, hit );
+ if ( hit->dbn == 0xffff )
+ return lbt_del( bt, key );
+ }
+ return lbt_add( bt, key );
+} /* cXAdd */
projects / webpac / blobdiff