/*

** 2011-08-18

**

** The author disclaims copyright to this source code. In place of

** a legal notice, here is a blessing:

**

** May you do good and not evil.

** May you find forgiveness for yourself and forgive others.

** May you share freely, never taking more than you give.

**

*************************************************************************

**

** This file contains the implementation of an in-memory tree structure.

**

** Technically the tree is a B-tree of order 4 (in the Knuth sense - each

** node may have up to 4 children). Keys are stored within B-tree nodes by

** reference. This may be slightly slower than a conventional red-black

** tree, but it is simpler. It is also an easier structure to modify to

** create a version that supports nested transaction rollback.

**

** This tree does not currently support a delete operation. One is not

** required. When LSM deletes a key from a database, it inserts a DELETE

** marker into the data structure. As a result, although the value associated

** with a key stored in the in-memory tree structure may be modified, no

** keys are ever removed.

*/

/*

** MVCC NOTES

**

** The in-memory tree structure supports SQLite-style MVCC. This means

** that while one client is writing to the tree structure, other clients

** may still be querying an older snapshot of the tree.

**

** One way to implement this is to use an append-only b-tree. In this

** case instead of modifying nodes in-place, a copy of the node is made

** and the required modifications made to the copy. The parent of the

** node is then modified (to update the pointer so that it points to

** the new copy), which causes a copy of the parent to be made, and so on.

** This means that each time the tree is written to a new root node is

** created. A snapshot is identified by the root node that it uses.

**

** The problem with the above is that each time the tree is written to,

** a copy of the node structure modified and all of its ancestor nodes

** is made. This may prove excessive with large tree structures.

**

** To reduce this overhead, the data structure used for a tree node is

** designed so that it may be edited in place exactly once without

** affecting existing users. In other words, the node structure is capable

** of storing two separate versions of the node at the same time.

** When a node is to be edited, if the node structure already contains

** two versions, a copy is made as in the append-only approach. Or, if

** it only contains a single version, it is edited in place.

**

** This reduces the overhead so that, roughly, one new node structure

** must be allocated for each write (on top of those allocations that

** would have been required by a non-MVCC tree). Logic: Assume that at

** any time, 50% of nodes in the tree already contain 2 versions. When

** a new entry is written to a node, there is a 50% chance that a copy

** of the node will be required. And a 25% chance that a copy of its

** parent is required. And so on.

**

** ROLLBACK

**

** The in-memory tree also supports transaction and sub-transaction

** rollback. In order to rollback to point in time X, the following is

** necessary:

**

** 1. All memory allocated since X must be freed, and

** 2. All "v2" data adding to nodes that existed at X should be zeroed.

** 3. The root node must be restored to its X value.

**

** The Mempool object used to allocate memory for the tree supports

** operation (1) - see the lsmPoolMark() and lsmPoolRevert() functions.

**

** To support (2), all nodes that have v2 data are part of a singly linked

** list, sorted by the age of the v2 data (nodes that have had data added

** most recently are at the end of the list). So to zero all v2 data added

** since X, the linked list is traversed from the first node added following

** X onwards.

**

*/

#ifndef _LSM_INT_H

# include "lsmInt.h"

#endif

#include <string.h>

#define MAX_DEPTH 32

typedef struct TreeKey TreeKey;

typedef struct TreeNode TreeNode;

typedef struct TreeLeaf TreeLeaf;

typedef struct NodeVersion NodeVersion;

struct TreeOld {

u32 iShmid; /* Last shared-memory chunk in use by old */

u32 iRoot; /* Offset of root node in shm file */

u32 nHeight; /* Height of tree structure */

};

#if 0

/*

** assert() that a TreeKey.flags value is sane. Usage:

**

** assert( lsmAssertFlagsOk(pTreeKey->flags) );

*/

static int lsmAssertFlagsOk(u8 keyflags){

/* At least one flag must be set. Otherwise, what is this key doing? */

assert( keyflags!=0 );

/* The POINT_DELETE and INSERT flags cannot both be set. */

assert( (keyflags & LSM_POINT_DELETE)==0 || (keyflags & LSM_INSERT)==0 );

/* If both the START_DELETE and END_DELETE flags are set, then the INSERT

** flag must also be set. In other words - the three DELETE flags cannot

** all be set */

assert( (keyflags & LSM_END_DELETE)==0

|| (keyflags & LSM_START_DELETE)==0

|| (keyflags & LSM_POINT_DELETE)==0

);

return 1;

}

#endif

static int assert_delete_ranges_match(lsm_db *);

static int treeCountEntries(lsm_db *db);

/*

** Container for a key-value pair. Within the *-shm file, each key/value

** pair is stored in a single allocation (which may not actually be

** contiguous in memory). Layout is the TreeKey structure, followed by

** the nKey bytes of key blob, followed by the nValue bytes of value blob

** (if nValue is non-negative).

*/

struct TreeKey {

int nKey; /* Size of pKey in bytes */

int nValue; /* Size of pValue. Or negative. */

u8 flags; /* Various LSM_XXX flags */

};

#define TKV_KEY(p) ((void *)&(p)[1])

#define TKV_VAL(p) ((void *)(((u8 *)&(p)[1]) + (p)->nKey))

/*

** A single tree node. A node structure may contain up to 3 key/value

** pairs. Internal (non-leaf) nodes have up to 4 children.

**

** TODO: Update the format of this to be more compact. Get it working

** first though...

*/

struct TreeNode {

u32 aiKeyPtr[3]; /* Array of pointers to TreeKey objects */

/* The following fields are present for interior nodes only, not leaves. */

u32 aiChildPtr[4]; /* Array of pointers to child nodes */

/* The extra child pointer slot. */

u32 iV2; /* Transaction number of v2 */

u8 iV2Child; /* apChild[] entry replaced by pV2Ptr */

u32 iV2Ptr; /* Substitute pointer */

};

struct TreeLeaf {

u32 aiKeyPtr[3]; /* Array of pointers to TreeKey objects */

};

typedef struct TreeBlob TreeBlob;

struct TreeBlob {

int n;

u8 *a;

};

/*

** Cursor for searching a tree structure.

**

** If a cursor does not point to any element (a.k.a. EOF), then the

** TreeCursor.iNode variable is set to a negative value. Otherwise, the

** cursor currently points to key aiCell[iNode] on node apTreeNode[iNode].

**

** Entries in the apTreeNode[] and aiCell[] arrays contain the node and

** index of the TreeNode.apChild[] pointer followed to descend to the

** current element. Hence apTreeNode[0] always contains the root node of

** the tree.

*/

struct TreeCursor {

lsm_db *pDb; /* Database handle for this cursor */

TreeRoot *pRoot; /* Root node and height of tree to access */

int iNode; /* Cursor points at apTreeNode[iNode] */

TreeNode *apTreeNode[MAX_DEPTH];/* Current position in tree */

u8 aiCell[MAX_DEPTH]; /* Current position in tree */

TreeKey *pSave; /* Saved key */

TreeBlob blob; /* Dynamic storage for a key */

};

/*

** A value guaranteed to be larger than the largest possible transaction

** id (TreeHeader.iTransId).

*/

#define WORKING_VERSION (1<<30)

static int tblobGrow(lsm_db *pDb, TreeBlob *p, int n, int *pRc){

if( n>p->n ){

lsmFree(pDb->pEnv, p->a);

p->a = lsmMallocRc(pDb->pEnv, n, pRc);

p->n = n;

}

return (p->a==0);

}

static void tblobFree(lsm_db *pDb, TreeBlob *p){

lsmFree(pDb->pEnv, p->a);

}

/***********************************************************************

** Start of IntArray methods. */

/*

** Append value iVal to the contents of IntArray *p. Return LSM_OK if

** successful, or LSM_NOMEM if an OOM condition is encountered.

*/

static int intArrayAppend(lsm_env *pEnv, IntArray *p, u32 iVal){

assert( p->nArray<=p->nAlloc );

if( p->nArray>=p->nAlloc ){

u32 *aNew;

int nNew = p->nArray ? p->nArray*2 : 128;

aNew = lsmRealloc(pEnv, p->aArray, nNew*sizeof(u32));

if( !aNew ) return LSM_NOMEM_BKPT;

p->aArray = aNew;

p->nAlloc = nNew;

}

p->aArray[p->nArray++] = iVal;

return LSM_OK;

}

/*

** Zero the IntArray object.

*/

static void intArrayFree(lsm_env *pEnv, IntArray *p){

p->nArray = 0;

}

/*

** Return the number of entries currently in the int-array object.

*/

static int intArraySize(IntArray *p){

return p->nArray;

}

/*

** Return a copy of the iIdx'th entry in the int-array.

*/

static u32 intArrayEntry(IntArray *p, int iIdx){

return p->aArray[iIdx];

}

/*

** Truncate the int-array so that all but the first nVal values are

** discarded.

*/

static void intArrayTruncate(IntArray *p, int nVal){

p->nArray = nVal;

}

/* End of IntArray methods.

***********************************************************************/

static int treeKeycmp(void *p1, int n1, void *p2, int n2){

int res;

res = memcmp(p1, p2, LSM_MIN(n1, n2));

if( res==0 ) res = (n1-n2);

return res;

}

/*

** The pointer passed as the first argument points to an interior node,

** not a leaf. This function returns the offset of the iCell'th child

** sub-tree of the node.

*/

static u32 getChildPtr(TreeNode *p, int iVersion, int iCell){

assert( iVersion>=0 );

assert( iCell>=0 && iCell<=array_size(p->aiChildPtr) );

if( p->iV2 && p->iV2<=(u32)iVersion && iCell==p->iV2Child ) return p->iV2Ptr;

return p->aiChildPtr[iCell];

}

/*

** Given an offset within the *-shm file, return the associated chunk number.

*/

static int treeOffsetToChunk(u32 iOff){

assert( LSM_SHM_CHUNK_SIZE==(1<<15) );

return (int)(iOff>>15);

}

#define treeShmptrUnsafe(pDb, iPtr) \

(&((u8*)((pDb)->apShm[(iPtr)>>15]))[(iPtr) & (LSM_SHM_CHUNK_SIZE-1)])

/*

** Return a pointer to the mapped memory location associated with *-shm

** file offset iPtr.

*/

static void *treeShmptr(lsm_db *pDb, u32 iPtr){

assert( (iPtr>>15)<(u32)pDb->nShm );

assert( pDb->apShm[iPtr>>15] );

return iPtr ? treeShmptrUnsafe(pDb, iPtr) : 0;

}

static ShmChunk * treeShmChunk(lsm_db *pDb, int iChunk){

return (ShmChunk *)(pDb->apShm[iChunk]);

}

static ShmChunk * treeShmChunkRc(lsm_db *pDb, int iChunk, int *pRc){

assert( *pRc==LSM_OK );

if( iChunk<pDb->nShm || LSM_OK==(*pRc = lsmShmCacheChunks(pDb, iChunk+1)) ){

return (ShmChunk *)(pDb->apShm[iChunk]);

}

return 0;

}

#ifndef NDEBUG

static void assertIsWorkingChild(

lsm_db *db,

TreeNode *pNode,

TreeNode *pParent,

int iCell

){

TreeNode *p;

u32 iPtr = getChildPtr(pParent, WORKING_VERSION, iCell);

p = treeShmptr(db, iPtr);

assert( p==pNode );

}

#else

# define assertIsWorkingChild(w,x,y,z)

#endif

/* Values for the third argument to treeShmkey(). */

#define TKV_LOADKEY 1

#define TKV_LOADVAL 2

static TreeKey *treeShmkey(

lsm_db *pDb, /* Database handle */

u32 iPtr, /* Shmptr to TreeKey struct */

int eLoad, /* Either zero or a TREEKEY_LOADXXX value */

TreeBlob *pBlob, /* Used if dynamic memory is required */

int *pRc /* IN/OUT: Error code */

){

TreeKey *pRet;

assert( eLoad==TKV_LOADKEY || eLoad==TKV_LOADVAL );

pRet = (TreeKey *)treeShmptr(pDb, iPtr);

if( pRet ){

int nReq; /* Bytes of space required at pRet */

int nAvail; /* Bytes of space available at pRet */

nReq = sizeof(TreeKey) + pRet->nKey;

if( eLoad==TKV_LOADVAL && pRet->nValue>0 ){

nReq += pRet->nValue;

}

assert( LSM_SHM_CHUNK_SIZE==(1<<15) );

nAvail = LSM_SHM_CHUNK_SIZE - (iPtr & (LSM_SHM_CHUNK_SIZE-1));

if( nAvail<nReq ){

if( tblobGrow(pDb, pBlob, nReq, pRc)==0 ){

int nLoad = 0;

while( *pRc==LSM_OK ){

ShmChunk *pChunk;

void *p = treeShmptr(pDb, iPtr);

int n = LSM_MIN(nAvail, nReq-nLoad);

memcpy(&pBlob->a[nLoad], p, n);

nLoad += n;

if( nLoad==nReq ) break;

pChunk = treeShmChunk(pDb, treeOffsetToChunk(iPtr));

assert( pChunk );

iPtr = (pChunk->iNext * LSM_SHM_CHUNK_SIZE) + LSM_SHM_CHUNK_HDR;

nAvail = LSM_SHM_CHUNK_SIZE - LSM_SHM_CHUNK_HDR;

}

}

pRet = (TreeKey *)(pBlob->a);

}

}

return pRet;

}

#if defined(LSM_DEBUG) && defined(LSM_EXPENSIVE_ASSERT)

void assert_leaf_looks_ok(TreeNode *pNode){

assert( pNode->apKey[1] );

}

void assert_node_looks_ok(TreeNode *pNode, int nHeight){

if( pNode ){

assert( pNode->apKey[1] );

if( nHeight>1 ){

int i;

assert( getChildPtr(pNode, WORKING_VERSION, 1) );

assert( getChildPtr(pNode, WORKING_VERSION, 2) );

for(i=0; i<4; i++){

assert_node_looks_ok(getChildPtr(pNode, WORKING_VERSION, i), nHeight-1);

}

}

}

}

/*

** Run various assert() statements to check that the working-version of the

** tree is correct in the following respects:

**

** * todo...

*/

void assert_tree_looks_ok(int rc, Tree *pTree){

}

#else

# define assert_tree_looks_ok(x,y)

#endif

void lsmFlagsToString(int flags, char *zFlags){

zFlags[0] = (flags & LSM_END_DELETE) ? ']' : '.';

/* Only one of LSM_POINT_DELETE, LSM_INSERT and LSM_SEPARATOR should ever

** be set. If this is not true, write a '?' to the output. */

switch( flags & (LSM_POINT_DELETE|LSM_INSERT|LSM_SEPARATOR) ){

case 0: zFlags[1] = '.'; break;

case LSM_POINT_DELETE: zFlags[1] = '-'; break;

case LSM_INSERT: zFlags[1] = '+'; break;

case LSM_SEPARATOR: zFlags[1] = '^'; break;

default: zFlags[1] = '?'; break;

}

zFlags[2] = (flags & LSM_SYSTEMKEY) ? '*' : '.';

zFlags[3] = (flags & LSM_START_DELETE) ? '[' : '.';

zFlags[4] = '\0';

}

#ifdef LSM_DEBUG

/*

** Pointer pBlob points to a buffer containing a blob of binary data

** nBlob bytes long. Append the contents of this blob to *pStr, with

** each octet represented by a 2-digit hexadecimal number. For example,

** if the input blob is three bytes in size and contains {0x01, 0x44, 0xFF},

** then "0144ff" is appended to *pStr.

*/

static void lsmAppendStrBlob(LsmString *pStr, void *pBlob, int nBlob){

int i;

lsmStringExtend(pStr, nBlob*2);

if( pStr->nAlloc==0 ) return;

for(i=0; i<nBlob; i++){

u8 c = ((u8*)pBlob)[i];

if( c>='a' && c<='z' ){

pStr->z[pStr->n++] = c;

}else if( c!=0 || nBlob==1 || i!=(nBlob-1) ){

pStr->z[pStr->n++] = "0123456789abcdef"[(c>>4)&0xf];

pStr->z[pStr->n++] = "0123456789abcdef"[c&0xf];

}

}

pStr->z[pStr->n] = 0;

}

#if 0 /* NOT USED */

/*

** Append nIndent space (0x20) characters to string *pStr.

*/

static void lsmAppendIndent(LsmString *pStr, int nIndent){

int i;

lsmStringExtend(pStr, nIndent);

for(i=0; i<nIndent; i++) lsmStringAppend(pStr, " ", 1);

}

#endif

static void strAppendFlags(LsmString *pStr, u8 flags){

char zFlags[8];

lsmFlagsToString(flags, zFlags);

zFlags[4] = ':';

lsmStringAppend(pStr, zFlags, 5);

}

void dump_node_contents(

lsm_db *pDb,

u32 iNode, /* Print out the contents of this node */

char *zPath, /* Path from root to this node */

int nPath, /* Number of bytes in zPath */

int nHeight /* Height: (0==leaf) (1==parent-of-leaf) */

){

const char *zSpace = " ";

int i;

int rc = LSM_OK;

LsmString s;

TreeNode *pNode;

TreeBlob b = {0, 0};

pNode = (TreeNode *)treeShmptr(pDb, iNode);

if( nHeight==0 ){

/* Append the nIndent bytes of space to string s. */

lsmStringInit(&s, pDb->pEnv);

/* Append each key to string s. */

for(i=0; i<3; i++){

u32 iPtr = pNode->aiKeyPtr[i];

if( iPtr ){

TreeKey *pKey = treeShmkey(pDb, pNode->aiKeyPtr[i],TKV_LOADKEY, &b,&rc);

strAppendFlags(&s, pKey->flags);

lsmAppendStrBlob(&s, TKV_KEY(pKey), pKey->nKey);

lsmStringAppend(&s, " ", -1);

}

}

printf("% 6d %.*sleaf%.*s: %s\n",

iNode, nPath, zPath, 20-nPath-4, zSpace, s.z

);

lsmStringClear(&s);

}else{

for(i=0; i<4 && nHeight>0; i++){

u32 iPtr = getChildPtr(pNode, pDb->treehdr.root.iTransId, i);

zPath[nPath] = (char)(i+'0');

zPath[nPath+1] = '/';

if( iPtr ){

dump_node_contents(pDb, iPtr, zPath, nPath+2, nHeight-1);

}

if( i!=3 && pNode->aiKeyPtr[i] ){

TreeKey *pKey = treeShmkey(pDb, pNode->aiKeyPtr[i], TKV_LOADKEY,&b,&rc);

lsmStringInit(&s, pDb->pEnv);

strAppendFlags(&s, pKey->flags);

lsmAppendStrBlob(&s, TKV_KEY(pKey), pKey->nKey);

printf("% 6d %.*s%.*s: %s\n",

iNode, nPath+1, zPath, 20-nPath-1, zSpace, s.z);

lsmStringClear(&s);

}

}

}

tblobFree(pDb, &b);

}

void dump_tree_contents(lsm_db *pDb, const char *zCaption){

char zPath[64];

TreeRoot *p = &pDb->treehdr.root;

printf("\n%s\n", zCaption);

zPath[0] = '/';

if( p->iRoot ){

dump_node_contents(pDb, p->iRoot, zPath, 1, p->nHeight-1);

}

fflush(stdout);

}

#endif

/*

** Initialize a cursor object, the space for which has already been

** allocated.

*/

static void treeCursorInit(lsm_db *pDb, int bOld, TreeCursor *pCsr){

memset(pCsr, 0, sizeof(TreeCursor));

pCsr->pDb = pDb;

if( bOld ){

pCsr->pRoot = &pDb->treehdr.oldroot;

}else{

pCsr->pRoot = &pDb->treehdr.root;

}

pCsr->iNode = -1;

}

/*

** Return a pointer to the mapping of the TreeKey object that the cursor

** is pointing to.

*/

static TreeKey *csrGetKey(TreeCursor *pCsr, TreeBlob *pBlob, int *pRc){

TreeKey *pRet;

lsm_db *pDb = pCsr->pDb;

u32 iPtr = pCsr->apTreeNode[pCsr->iNode]->aiKeyPtr[pCsr->aiCell[pCsr->iNode]];

assert( iPtr );

pRet = (TreeKey*)treeShmptrUnsafe(pDb, iPtr);

if( !(pRet->flags & LSM_CONTIGUOUS) ){

pRet = treeShmkey(pDb, iPtr, TKV_LOADVAL, pBlob, pRc);

}

return pRet;

}

/*

** Save the current position of tree cursor pCsr.

*/

int lsmTreeCursorSave(TreeCursor *pCsr){

int rc = LSM_OK;

if( pCsr && pCsr->pSave==0 ){

int iNode = pCsr->iNode;

if( iNode>=0 ){

pCsr->pSave = csrGetKey(pCsr, &pCsr->blob, &rc);

}

pCsr->iNode = -1;

}

return rc;

}

/*

** Restore the position of a saved tree cursor.

*/

static int treeCursorRestore(TreeCursor *pCsr, int *pRes){

int rc = LSM_OK;

if( pCsr->pSave ){

TreeKey *pKey = pCsr->pSave;

pCsr->pSave = 0;

if( pRes ){

rc = lsmTreeCursorSeek(pCsr, TKV_KEY(pKey), pKey->nKey, pRes);

}

}

return rc;

}

/*

** Allocate nByte bytes of space within the *-shm file. If successful,

** return LSM_OK and set *piPtr to the offset within the file at which

** the allocated space is located.

*/

static u32 treeShmalloc(lsm_db *pDb, int bAlign, int nByte, int *pRc){

u32 iRet = 0;

if( *pRc==LSM_OK ){

const static int CHUNK_SIZE = LSM_SHM_CHUNK_SIZE;

const static int CHUNK_HDR = LSM_SHM_CHUNK_HDR;

u32 iWrite; /* Current write offset */

u32 iEof; /* End of current chunk */

int iChunk; /* Current chunk */

assert( nByte <= (CHUNK_SIZE-CHUNK_HDR) );

/* Check if there is enough space on the current chunk to fit the

** new allocation. If not, link in a new chunk and put the new

** allocation at the start of it. */

iWrite = pDb->treehdr.iWrite;

if( bAlign ){

iWrite = (iWrite + 3) & ~0x0003;

assert( (iWrite % 4)==0 );

}

assert( iWrite );

iChunk = treeOffsetToChunk(iWrite-1);

iEof = (iChunk+1) * CHUNK_SIZE;

assert( iEof>=iWrite && (iEof-iWrite)<(u32)CHUNK_SIZE );

if( (iWrite+nByte)>iEof ){

ShmChunk *pHdr; /* Header of chunk just finished (iChunk) */

ShmChunk *pFirst; /* Header of chunk treehdr.iFirst */

ShmChunk *pNext; /* Header of new chunk */

int iNext = 0; /* Next chunk */

int rc = LSM_OK;

pFirst = treeShmChunk(pDb, pDb->treehdr.iFirst);

assert( shm_sequence_ge(pDb->treehdr.iUsedShmid, pFirst->iShmid) );

assert( (pDb->treehdr.iNextShmid+1-pDb->treehdr.nChunk)==pFirst->iShmid );

/* Check if the chunk at the start of the linked list is still in

** use. If not, reuse it. If so, allocate a new chunk by appending

** to the *-shm file. */

if( pDb->treehdr.iUsedShmid!=pFirst->iShmid ){

int bInUse;

rc = lsmTreeInUse(pDb, pFirst->iShmid, &bInUse);

if( rc!=LSM_OK ){

*pRc = rc;

return 0;

}

if( bInUse==0 ){

iNext = pDb->treehdr.iFirst;

pDb->treehdr.iFirst = pFirst->iNext;

assert( pDb->treehdr.iFirst );

}

}

if( iNext==0 ) iNext = pDb->treehdr.nChunk++;

/* Set the header values for the new chunk */

pNext = treeShmChunkRc(pDb, iNext, &rc);

if( pNext ){

pNext->iNext = 0;

pNext->iShmid = (pDb->treehdr.iNextShmid++);

}else{

*pRc = rc;

return 0;

}

/* Set the header values for the chunk just finished */

pHdr = (ShmChunk *)treeShmptr(pDb, iChunk*CHUNK_SIZE);

pHdr->iNext = iNext;

/* Advance to the next chunk */

iWrite = iNext * CHUNK_SIZE + CHUNK_HDR;

}

/* Allocate space at iWrite. */

iRet = iWrite;

pDb->treehdr.iWrite = iWrite + nByte;

pDb->treehdr.root.nByte += nByte;

}

return iRet;

}

/*

** Allocate and zero nByte bytes of space within the *-shm file.

*/

static void *treeShmallocZero(lsm_db *pDb, int nByte, u32 *piPtr, int *pRc){

u32 iPtr;

void *p;

iPtr = treeShmalloc(pDb, 1, nByte, pRc);

p = treeShmptr(pDb, iPtr);

if( p ){

assert( *pRc==LSM_OK );

memset(p, 0, nByte);

*piPtr = iPtr;

}

return p;

}

static TreeNode *newTreeNode(lsm_db *pDb, u32 *piPtr, int *pRc){

return treeShmallocZero(pDb, sizeof(TreeNode), piPtr, pRc);

}

static TreeLeaf *newTreeLeaf(lsm_db *pDb, u32 *piPtr, int *pRc){

return treeShmallocZero(pDb, sizeof(TreeLeaf), piPtr, pRc);

}

static TreeKey *newTreeKey(

lsm_db *pDb,

u32 *piPtr,

void *pKey, int nKey, /* Key data */

void *pVal, int nVal, /* Value data (or nVal<0 for delete) */

int *pRc

){

TreeKey *p;

u32 iPtr;

u32 iEnd;

int nRem;

u8 *a;

int n;

/* Allocate space for the TreeKey structure itself */

*piPtr = iPtr = treeShmalloc(pDb, 1, sizeof(TreeKey), pRc);

p = treeShmptr(pDb, iPtr);

if( *pRc ) return 0;

p->nKey = nKey;

p->nValue = nVal;

/* Allocate and populate the space required for the key and value. */

n = nRem = nKey;

a = (u8 *)pKey;

while( a ){

while( nRem>0 ){

u8 *aAlloc;

int nAlloc;

u32 iWrite;

iWrite = (pDb->treehdr.iWrite & (LSM_SHM_CHUNK_SIZE-1));

iWrite = LSM_MAX(iWrite, LSM_SHM_CHUNK_HDR);

nAlloc = LSM_MIN((LSM_SHM_CHUNK_SIZE-iWrite), (u32)nRem);

aAlloc = treeShmptr(pDb, treeShmalloc(pDb, 0, nAlloc, pRc));

if( aAlloc==0 ) break;

memcpy(aAlloc, &a[n-nRem], nAlloc);

nRem -= nAlloc;

}

a = pVal;

n = nRem = nVal;

pVal = 0;

}

iEnd = iPtr + sizeof(TreeKey) + nKey + LSM_MAX(0, nVal);

if( (iPtr & ~(LSM_SHM_CHUNK_SIZE-1))!=(iEnd & ~(LSM_SHM_CHUNK_SIZE-1)) ){

p->flags = 0;

}else{

p->flags = LSM_CONTIGUOUS;

}

if( *pRc ) return 0;

#if 0

printf("store: %d %s\n", (int)iPtr, (char *)pKey);

#endif

return p;

}

static TreeNode *copyTreeNode(

lsm_db *pDb,

TreeNode *pOld,

u32 *piNew,

int *pRc

){

TreeNode *pNew;

pNew = newTreeNode(pDb, piNew, pRc);

if( pNew ){

memcpy(pNew->aiKeyPtr, pOld->aiKeyPtr, sizeof(pNew->aiKeyPtr));

memcpy(pNew->aiChildPtr, pOld->aiChildPtr, sizeof(pNew->aiChildPtr));

if( pOld->iV2 ) pNew->aiChildPtr[pOld->iV2Child] = pOld->iV2Ptr;

}

return pNew;

}

static TreeNode *copyTreeLeaf(

lsm_db *pDb,

TreeLeaf *pOld,

u32 *piNew,

int *pRc

){

TreeLeaf *pNew;

pNew = newTreeLeaf(pDb, piNew, pRc);

if( pNew ){

memcpy(pNew, pOld, sizeof(TreeLeaf));

}

return (TreeNode *)pNew;

}

/*

** The tree cursor passed as the second argument currently points to an

** internal node (not a leaf). Specifically, to a sub-tree pointer. This

** function replaces the sub-tree that the cursor currently points to

** with sub-tree pNew.

**

** The sub-tree may be replaced either by writing the "v2 data" on the

** internal node, or by allocating a new TreeNode structure and then

** calling this function on the parent of the internal node.

*/

static int treeUpdatePtr(lsm_db *pDb, TreeCursor *pCsr, u32 iNew){

int rc = LSM_OK;

if( pCsr->iNode<0 ){

/* iNew is the new root node */

pDb->treehdr.root.iRoot = iNew;

}else{

/* If this node already has version 2 content, allocate a copy and

** update the copy with the new pointer value. Otherwise, store the

** new pointer as v2 data within the current node structure. */

TreeNode *p; /* The node to be modified */

int iChildPtr; /* apChild[] entry to modify */

p = pCsr->apTreeNode[pCsr->iNode];

iChildPtr = pCsr->aiCell[pCsr->iNode];

if( p->iV2 ){

/* The "allocate new TreeNode" option */

u32 iCopy;

TreeNode *pCopy;

pCopy = copyTreeNode(pDb, p, &iCopy, &rc);

if( pCopy ){

assert( rc==LSM_OK );

pCopy->aiChildPtr[iChildPtr] = iNew;

pCsr->iNode--;

rc = treeUpdatePtr(pDb, pCsr, iCopy);

}

}else{

/* The "v2 data" option */

u32 iPtr;

assert( pDb->treehdr.root.iTransId>0 );

if( pCsr->iNode ){

iPtr = getChildPtr(

pCsr->apTreeNode[pCsr->iNode-1],

pDb->treehdr.root.iTransId, pCsr->aiCell[pCsr->iNode-1]

);

}else{

iPtr = pDb->treehdr.root.iRoot;

}

rc = intArrayAppend(pDb->pEnv, &pDb->rollback, iPtr);

if( rc==LSM_OK ){

p->iV2 = pDb->treehdr.root.iTransId;

p->iV2Child = (u8)iChildPtr;

p->iV2Ptr = iNew;

}

}

}

return rc;

}

/*

** Cursor pCsr points at a node that is part of pTree. This function

** inserts a new key and optionally child node pointer into that node.

**

** The position into which the new key and pointer are inserted is

** determined by the iSlot parameter. The new key will be inserted to

** the left of the key currently stored in apKey[iSlot]. Or, if iSlot is

** greater than the index of the rightmost key in the node.

**

** Pointer pLeftPtr points to a child tree that contains keys that are

** smaller than pTreeKey.

*/

static int treeInsert(

lsm_db *pDb, /* Database handle */

TreeCursor *pCsr, /* Cursor indicating path to insert at */

u32 iLeftPtr, /* Left child pointer */

u32 iTreeKey, /* Location of key to insert */

u32 iRightPtr, /* Right child pointer */

int iSlot /* Position to insert key into */

){

int rc = LSM_OK;

TreeNode *pNode = pCsr->apTreeNode[pCsr->iNode];

/* Check if the node is currently full. If so, split pNode in two and

** call this function recursively to add a key to the parent. Otherwise,

** insert the new key directly into pNode. */

assert( pNode->aiKeyPtr[1] );

if( pNode->aiKeyPtr[0] && pNode->aiKeyPtr[2] ){

u32 iLeft; TreeNode *pLeft; /* New left-hand sibling node */

u32 iRight; TreeNode *pRight; /* New right-hand sibling node */

pLeft = newTreeNode(pDb, &iLeft, &rc);

pRight = newTreeNode(pDb, &iRight, &rc);

if( rc ) return rc;

pLeft->aiChildPtr[1] = getChildPtr(pNode, WORKING_VERSION, 0);

pLeft->aiKeyPtr[1] = pNode->aiKeyPtr[0];

pLeft->aiChildPtr[2] = getChildPtr(pNode, WORKING_VERSION, 1);

pRight->aiChildPtr[1] = getChildPtr(pNode, WORKING_VERSION, 2);

pRight->aiKeyPtr[1] = pNode->aiKeyPtr[2];

pRight->aiChildPtr[2] = getChildPtr(pNode, WORKING_VERSION, 3);

if( pCsr->iNode==0 ){

/* pNode is the root of the tree. Grow the tree by one level. */

u32 iRoot; TreeNode *pRoot; /* New root node */

pRoot = newTreeNode(pDb, &iRoot, &rc);

pRoot->aiKeyPtr[1] = pNode->aiKeyPtr[1];

pRoot->aiChildPtr[1] = iLeft;

pRoot->aiChildPtr[2] = iRight;

pDb->treehdr.root.iRoot = iRoot;

pDb->treehdr.root.nHeight++;

}else{

pCsr->iNode--;

rc = treeInsert(pDb, pCsr,

iLeft, pNode->aiKeyPtr[1], iRight, pCsr->aiCell[pCsr->iNode]

);

}

assert( pLeft->iV2==0 );

assert( pRight->iV2==0 );

switch( iSlot ){

case 0:

pLeft->aiKeyPtr[0] = iTreeKey;

pLeft->aiChildPtr[0] = iLeftPtr;

if( iRightPtr ) pLeft->aiChildPtr[1] = iRightPtr;

break;

case 1:

pLeft->aiChildPtr[3] = (iRightPtr ? iRightPtr : pLeft->aiChildPtr[2]);

pLeft->aiKeyPtr[2] = iTreeKey;

pLeft->aiChildPtr[2] = iLeftPtr;

break;

case 2:

pRight->aiKeyPtr[0] = iTreeKey;

pRight->aiChildPtr[0] = iLeftPtr;

if( iRightPtr ) pRight->aiChildPtr[1] = iRightPtr;

break;

case 3:

pRight->aiChildPtr[3] = (iRightPtr ? iRightPtr : pRight->aiChildPtr[2]);

pRight->aiKeyPtr[2] = iTreeKey;

pRight->aiChildPtr[2] = iLeftPtr;

break;

}

}else{

TreeNode *pNew;

u32 *piKey;

u32 *piChild;

u32 iStore = 0;

u32 iNew = 0;

int i;

/* Allocate a new version of node pNode. */

pNew = newTreeNode(pDb, &iNew, &rc);

if( rc ) return rc;

piKey = pNew->aiKeyPtr;

piChild = pNew->aiChildPtr;

for(i=0; i<iSlot; i++){

if( pNode->aiKeyPtr[i] ){

*(piKey++) = pNode->aiKeyPtr[i];

*(piChild++) = getChildPtr(pNode, WORKING_VERSION, i);

}

}

*piKey++ = iTreeKey;

*piChild++ = iLeftPtr;

iStore = iRightPtr;

for(i=iSlot; i<3; i++){

if( pNode->aiKeyPtr[i] ){

*(piKey++) = pNode->aiKeyPtr[i];

*(piChild++) = iStore ? iStore : getChildPtr(pNode, WORKING_VERSION, i);

iStore = 0;

}

}