SQLite4  Check-in [9374c3a283]

    zStart = z;

    while( *z && *z!='=' ) z++;
    if( *z ){
      int eParam;
      int i;
      int iVal;
      int iMul = 1;
      int rc;
      char zParam[32];
      int nParam = z-zStart;
      if( nParam==0 || nParam>sizeof(zParam)-1 ) goto syntax_error;

      memcpy(zParam, zStart, nParam);
      zParam[nParam] = '\0';
      rc = testArgSelect(aParam, "param", zParam, &i);
      if( rc!=0 ) return rc;
      eParam = aParam[i].eParam;

      z++;
      zStart = z;
      while( *z>='0' && *z<='9' ) z++;
      if( *z=='k' || *z=='K' ){
        iMul = 1024;
        z++;
      }else if( *z=='M' || *z=='M' ){
        iMul = 1024 * 1024;
        z++;
      }
      nParam = z-zStart;
      if( nParam==0 || nParam>sizeof(zParam)-1 ) goto syntax_error;
      memcpy(zParam, zStart, nParam);
      zParam[nParam] = '\0';
      iVal = atoi(zParam) * iMul;

      if( eParam>0 ){
        if( bWorker || aParam[i].bWorker==0 ){
          lsm_config(db, eParam, &iVal);
        }
      }else{
        if( pLsm ){

**   lsmMCursorNext()
**   lsmMCursorPrev()
**   lsmMCursorFirst()
**   lsmMCursorLast()
**   lsmMCursorKey()
**   lsmMCursorValue()
**   lsmMCursorValid()
**
** iFree:
**   This variable is only used by cursors providing input data for a
**   new top-level segment. Such cursors only ever iterate forwards, not
**   backwards.
*/
struct MultiCursor {
  lsm_db *pDb;                    /* Connection that owns this cursor */
  MultiCursor *pNext;             /* Next cursor owned by connection pDb */
  int flags;                      /* Mask of CURSOR_XXX flags */

  int eType;                      /* Cache of current key type */

        lsmTreeCursorValue(pTreeCsr, &pVal, &nVal);
      }
      break;
    }

    case CURSOR_DATA_SYSTEM: {
      Snapshot *pWorker = pCsr->pDb->pWorker;
      if( pWorker && (pCsr->flags & CURSOR_FLUSH_FREELIST) ){
        int nEntry = pWorker->freelist.nEntry;
        if( pCsr->iFree < (nEntry*2) ){
          FreelistEntry *aEntry = pWorker->freelist.aEntry;
          int i = nEntry - 1 - (pCsr->iFree / 2);
          u32 iKey = 0;

          if( (pCsr->iFree % 2) ){
            eType = LSM_END_DELETE|LSM_SYSTEMKEY;
            iKey = aEntry[i].iBlk-1;
          }else if( aEntry[i].iId>=0 ){
            eType = LSM_INSERT|LSM_SYSTEMKEY;
            iKey = aEntry[i].iBlk;

            /* If the in-memory entry immediately before this one was a
             ** DELETE, and the block number is one greater than the current
             ** block number, mark this entry as an "end-delete-range". */
            if( i<(nEntry-1) 
                && aEntry[i+1].iBlk==aEntry[i].iBlk+1 && aEntry[i+1].iId<0
              ){
              eType |= LSM_END_DELETE;
            }

            /* If the in-memory entry immediately after this one is a
            ** DELETE, and the block number is one less than the current
            ** block number, mark this entry as an "start-delete-range". 
            ** Also increase iFree so that the next entry is not visited
            ** (since it has already been accounted for by setting this
            ** flag).  */
            if( i>0 
                && aEntry[i-1].iBlk==aEntry[i].iBlk-1 && aEntry[i-1].iId<0
              ){
              eType |= LSM_START_DELETE;
            }
          }else{
            eType = LSM_START_DELETE|LSM_SYSTEMKEY;
            iKey = aEntry[i].iBlk + 1;
          }

          pKey = pCsr->pSystemVal;
          nKey = 4;


          lsmPutU32(pKey, ~iKey);

        }
      }
      break;
    }

    default: {
      int iPtr = iKey - CURSOR_DATA_SEGMENT;

  if( peType ) *peType = eType;
  if( pnKey ) *pnKey = nKey;
  if( ppKey ) *ppKey = pKey;
}

static int sortedDbKeyCompare(
  MultiCursor *pCsr,
  int iLhsFlags, void *pLhsKey, int nLhsKey,
  int iRhsFlags, void *pRhsKey, int nRhsKey
){
  int (*xCmp)(void *, int, void *, int) = pCsr->pDb->xCmp;
  int res;

  /* Compare the keys, including the system flag. */
  res = sortedKeyCompare(xCmp, 
    rtTopic(iLhsFlags), pLhsKey, nLhsKey,
    rtTopic(iRhsFlags), pRhsKey, nRhsKey
  );

    iRes = i2;
  }else if( pKey2==0 ){
    iRes = i1;
  }else{
    int res;

    /* Compare the keys */
    res = sortedDbKeyCompare(pCsr,
        eType1, pKey1, nKey1, eType2, pKey2, nKey2
    );

    res = res * mul;
    if( res==0 ){
      iRes = (rtIsSeparator(eType1) ? i2 : i1);
    }else if( res<0 ){

        *pnVal = 0;
      }
      break;
    }

    case CURSOR_DATA_SYSTEM: {
      Snapshot *pWorker = pCsr->pDb->pWorker;
      if( pWorker 
       && (pCsr->iFree % 2)==0
       && pCsr->iFree < (pWorker->freelist.nEntry*2)
      ){
        int iEntry = pWorker->freelist.nEntry - 1 - (pCsr->iFree / 2);
        u8 *aVal = &((u8 *)(pCsr->pSystemVal))[4];
        lsmPutU64(aVal, pWorker->freelist.aEntry[iEntry].iId);
        *ppVal = aVal;
        *pnVal = 8;
      }
      break;
    }

  /* Check if this key has already been deleted by a range-delete */
  iKey = pCsr->aTree[1];
  if( (iKey>0 && (rdmask & lsmTreeCursorFlags(pCsr->apTreeCsr[0]))) 
   || (iKey>1 && (rdmask & lsmTreeCursorFlags(pCsr->apTreeCsr[1]))) 
  ){
    return 0;
  }
  if( iKey>CURSOR_DATA_SYSTEM && (pCsr->flags & CURSOR_FLUSH_FREELIST) ){
    int eType;
    multiCursorGetKey(pCsr, CURSOR_DATA_SYSTEM, &eType, 0, 0);
    if( rdmask & eType ) return 0;
  }

  for(i=CURSOR_DATA_SEGMENT; i<iKey; i++){
    int iPtr = i-CURSOR_DATA_SEGMENT;
    if( pCsr->aPtr[iPtr].pPg && (pCsr->aPtr[iPtr].eType & rdmask) ){
      return 0;
    }
  }

  /* Check the current key value. If it is not greater than (if bReverse==0)
  ** or less than (if bReverse!=0) the key currently cached in pCsr->key, 
  ** then the cursor has not yet been successfully advanced.  
  */
  multiCursorGetKey(pCsr, pCsr->aTree[1], &eNewType, &pNew, &nNew);
  if( pNew ){
    int typemask = (pCsr->flags & CURSOR_IGNORE_DELETE) ? ~(0) : LSM_SYSTEMKEY;
    int res = sortedDbKeyCompare(pCsr,
      eNewType & typemask, pNew, nNew, 
      pCsr->eType & typemask, pCsr->key.pData, pCsr->key.nData
    );

    if( (bReverse==0 && res<=0) || (bReverse!=0 && res>=0) ){
      return 0;
    }

    multiCursorCacheKey(pCsr, pRc);
    assert( pCsr->eType==eNewType );

    /* If this cursor is configured to skip deleted keys, and the current
    ** cursor points to a SORTED_DELETE entry, then the cursor has not been 
    ** successfully advanced.  
    **
    ** Similarly, if the cursor is configured to skip system keys and the
    ** current cursor points to a system key, it has not yet been advanced.
     */
    if( *pRc==LSM_OK && 0==mcursorLocationOk(pCsr, 0) ) return 0;
  }
  return 1;
}

static void flCsrAdvance(MultiCursor *pCsr){
  assert( pCsr->flags & CURSOR_FLUSH_FREELIST );
  if( pCsr->iFree % 2 ){
    pCsr->iFree++;
  }else{
    int nEntry = pCsr->pDb->pWorker->freelist.nEntry;
    FreelistEntry *aEntry = pCsr->pDb->pWorker->freelist.aEntry;

    int i = nEntry - 1 - (pCsr->iFree / 2);

    /* If the current entry is a delete and the "end-delete" key will not
    ** be attached to the next entry, increment iFree by 1 only. */
    if( aEntry[i].iId<0 ){
      while( 1 ){
        if( i==0 || aEntry[i-1].iBlk!=aEntry[i].iBlk-1 ){
          pCsr->iFree--;
          break;
        }
        if( aEntry[i-1].iId>=0 ) break;
        pCsr->iFree += 2;
        i--;
      }
    }
    pCsr->iFree += 2;
  }
}

static int multiCursorAdvance(MultiCursor *pCsr, int bReverse){
  int rc = LSM_OK;                /* Return Code */
  if( lsmMCursorValid(pCsr) ){
    do {
      int iKey = pCsr->aTree[1];

          rc = lsmTreeCursorPrev(pTreeCsr);
        }else{
          rc = lsmTreeCursorNext(pTreeCsr);
        }
      }else if( iKey==CURSOR_DATA_SYSTEM ){
        assert( pCsr->flags & CURSOR_FLUSH_FREELIST );
        assert( bReverse==0 );
        flCsrAdvance(pCsr);
      }else if( iKey==(CURSOR_DATA_SEGMENT+pCsr->nPtr) ){
        assert( bReverse==0 && pCsr->pBtCsr );
        rc = btreeCursorNext(pCsr->pBtCsr);
      }else{
        rc = segmentCursorAdvance(pCsr, iKey-CURSOR_DATA_SEGMENT, bReverse);
      }
      if( rc==LSM_OK ){

int lsmMCursorPrev(MultiCursor *pCsr){
  if( (pCsr->flags & CURSOR_PREV_OK)==0 ) return LSM_MISUSE_BKPT;
  return multiCursorAdvance(pCsr, 1);
}

int lsmMCursorKey(MultiCursor *pCsr, void **ppKey, int *pnKey){
  if( (pCsr->flags & CURSOR_SEEK_EQ) || pCsr->aTree==0 ){
    *pnKey = pCsr->key.nData;
    *ppKey = pCsr->key.pData;
  }else{
    int iKey = pCsr->aTree[1];

    if( iKey==CURSOR_DATA_TREE0 || iKey==CURSOR_DATA_TREE1 ){
      TreeCursor *pTreeCsr = pCsr->apTreeCsr[iKey-CURSOR_DATA_TREE0];

  return LSM_OK;
}

int lsmMCursorValue(MultiCursor *pCsr, void **ppVal, int *pnVal){
  void *pVal;
  int nVal;
  int rc;
  if( (pCsr->flags & CURSOR_SEEK_EQ) || pCsr->aTree==0 ){
    rc = LSM_OK;
    nVal = pCsr->val.nData;
    pVal = pCsr->val.pData;
  }else{

    assert( pCsr->aTree );
    assert( mcursorLocationOk(pCsr, (pCsr->flags & CURSOR_IGNORE_DELETE)) );

}

static int mergeWorkerWrite(
  MergeWorker *pMW,               /* Merge worker object to write into */
  int eType,                      /* One of SORTED_SEPARATOR, WRITE or DELETE */
  void *pKey, int nKey,           /* Key value */
  MultiCursor *pCsr,              /* Read value (if any) from here */
  int iVal,
  int iPtr                        /* Absolute value of page pointer, or 0 */
){
  int rc = LSM_OK;                /* Return code */
  Merge *pMerge;                  /* Persistent part of level merge state */
  int nHdr;                       /* Space required for this record header */
  Page *pPg;                      /* Page to write to */
  u8 *aData;                      /* Data buffer for page pWriter->pPage */

  **
  **     1) record type - 1 byte.
  **     2) Page-pointer-offset - 1 varint
  **     3) Key size - 1 varint
  **     4) Value size - 1 varint (only if LSM_INSERT flag is set)
  */
  if( rc==LSM_OK ){
    rc = multiCursorGetVal(pCsr, iVal, &pVal, &nVal);
  }
  if( rc==LSM_OK ){
    nHdr = 1 + lsmVarintLen32(iRPtr) + lsmVarintLen32(nKey);
    if( rtIsWrite(eType) ) nHdr += lsmVarintLen32(nVal);

    /* If the entire header will not fit on page pPg, or if page pPg is 
    ** marked read-only, advance to the next page of the output run. */

    if( rtIsWrite(eType) ) iOff += lsmVarintPut32(&aData[iOff], nVal);   /* 4 */
    pMerge->iOutputOff = iOff;

    /* Write the key and data into the segment. */
    assert( iFPtr==pageGetPtr(aData, nData) );
    rc = mergeWorkerData(pMW, 0, iFPtr+iRPtr, pKey, nKey);
    if( rc==LSM_OK && rtIsWrite(eType) ){
      if( rtTopic(eType)==0 ) rc = multiCursorGetVal(pCsr, iVal, &pVal, &nVal);
      if( rc==LSM_OK ){
        rc = mergeWorkerData(pMW, 0, iFPtr+iRPtr, pVal, nVal);
      }
    }
  }

  return rc;

** a merge operation. When this function is called, *piFlags contains the
** database entry flags for the current entry. The entry about to be written
** to the output.
**
** Note that this function only has to work for cursors configured to 
** iterate forwards (not backwards).
*/
static void mergeRangeDeletes(MultiCursor *pCsr, int *piVal, int *piFlags){
  int f = *piFlags;
  int iKey = pCsr->aTree[1];
  int i;

  assert( pCsr->flags & CURSOR_NEXT_OK );
  if( pCsr->flags & CURSOR_IGNORE_DELETE ){
    /* The ignore-delete flag is set when the output of the merge will form
    ** the oldest level in the database. In this case there is no point in
    ** retaining any range-delete flags.  */
    assert( (f & LSM_POINT_DELETE)==0 );
    f &= ~(LSM_START_DELETE|LSM_END_DELETE);
  }else{
#if 0
    if( iKey==0 ){
      int btreeflags = lsmTreeCursorFlags(pCsr->apTreeCsr[1]);
      if( btreeflags & LSM_END_DELETE ){
        f |= (LSM_START_DELETE|LSM_END_DELETE);
      }
    }

    for(i=LSM_MAX(0, iKey+1-CURSOR_DATA_SEGMENT); i<pCsr->nPtr; i++){
      SegmentPtr *pPtr = &pCsr->aPtr[i];
      if( pPtr->pPg && (pPtr->eType & LSM_END_DELETE) ){
        f |= (LSM_START_DELETE|LSM_END_DELETE);
      }
    }
#endif
    for(i=0; i<(CURSOR_DATA_SEGMENT + pCsr->nPtr); i++){
      if( i!=iKey ){
        int eType;
        void *pKey;
        int nKey;
        int res;
        multiCursorGetKey(pCsr, i, &eType, &pKey, &nKey);

        if( pKey ){
          res = sortedKeyCompare(pCsr->pDb->xCmp, 
              rtTopic(pCsr->eType), pCsr->key.pData, pCsr->key.nData,
              rtTopic(eType), pKey, nKey
              );
          assert( res<=0 );
          if( res==0 ){
            if( (f & (LSM_INSERT|LSM_POINT_DELETE))==0 ){
              if( eType & LSM_INSERT ){
                f |= LSM_INSERT;
                *piVal = i;
              }
              else if( eType & LSM_POINT_DELETE ){
                f |= LSM_POINT_DELETE;
              }
            }
            f |= (eType & (LSM_END_DELETE|LSM_START_DELETE));
          }

          if( i>iKey && (eType & LSM_END_DELETE) && res<0 ){
            f |= (LSM_END_DELETE|LSM_START_DELETE);
          }
        }
      }
    }

    assert( (f & LSM_INSERT)==0 || (f & LSM_POINT_DELETE)==0 );
    if( (f & LSM_START_DELETE) 
     && (f & LSM_END_DELETE) 
     && (f & LSM_POINT_DELETE )
    ){
      f = 0;
    }
  }

  *piFlags = f;
}

static int mergeWorkerStep(MergeWorker *pMW){
  lsm_db *pDb = pMW->pDb;       /* Database handle */
  MultiCursor *pCsr;            /* Cursor to read input data from */
  int rc = LSM_OK;              /* Return code */
  int eType;                    /* SORTED_SEPARATOR, WRITE or DELETE */
  void *pKey; int nKey;         /* Key */
  Segment *pSeg;                /* Output segment */
  Pgno iPtr;
  int iVal;

  pCsr = pMW->pCsr;
  pSeg = &pMW->pLevel->lhs;

  /* Pull the next record out of the source cursor. */
  lsmMCursorKey(pCsr, &pKey, &nKey);
  eType = pCsr->eType;

    if( pPtr->pPg
     && 0==pDb->xCmp(pPtr->pKey, pPtr->nKey, pKey, nKey)
    ){
      iPtr = pPtr->iPtr+pPtr->iPgPtr;
    }
  }

  iVal = pCsr->aTree[1];
  mergeRangeDeletes(pCsr, &iVal, &eType);

  if( eType!=0 ){
    if( pMW->aGobble ){
      int iGobble = pCsr->aTree[1] - CURSOR_DATA_SEGMENT;
      if( iGobble<pCsr->nPtr ){
        SegmentPtr *pGobble = &pCsr->aPtr[iGobble];
        if( (pGobble->flags & PGFTR_SKIP_THIS_FLAG)==0 ){
          pMW->aGobble[iGobble] = lsmFsPageNumber(pGobble->pPg);
        }
      }
    }

    /* If this is a separator key and we know that the output pointer has not
    ** changed, there is no point in writing an output record. Otherwise,
    ** proceed. */
    if( rtIsSeparator(eType)==0 || iPtr!=0 ){
      /* Write the record into the main run. */
      if( rc==LSM_OK ){
        rc = mergeWorkerWrite(pMW, eType, pKey, nKey, pCsr, iVal, iPtr);
      }
    }
  }

  /* Advance the cursor to the next input record (assuming one exists). */
  assert( lsmMCursorValid(pMW->pCsr) );
  if( rc==LSM_OK ) rc = lsmMCursorNext(pMW->pCsr);

  if( rc!=LSM_OK || pNew->lhs.iFirst==0 ){
    assert( rc!=LSM_OK || pDb->pWorker->freelist.nEntry==0 );
    lsmDbSnapshotSetLevel(pDb->pWorker, pNext);
    sortedFreeLevel(pDb->pEnv, pNew);
  }else{
    if( pDel ) pDel->iRoot = 0;

#if 1
    lsmSortedDumpStructure(pDb, pDb->pWorker, 1, 0, "new-toplevel");
#endif

    if( freelist.nEntry ){
      Freelist *p = &pDb->pWorker->freelist;
      lsmFree(pDb->pEnv, p->aEntry);
      memcpy(p, &freelist, sizeof(freelist));
      freelist.aEntry = 0;

      /* Clean up the MergeWorker object initialized above. If no error
      ** has occurred, invoke the work-hook to inform the application that
      ** the database structure has changed. */
      mergeWorkerShutdown(&mergeworker, &rc);
      if( rc==LSM_OK ) sortedInvokeWorkHook(pDb);

#if 1
      lsmSortedDumpStructure(pDb, pDb->pWorker, 1, 0, "work");
#endif
      assertBtreeOk(pDb, &pLevel->lhs);
      assertRunInOrder(pDb, &pLevel->lhs);

      /* If bFlush is true and the database is no longer considered "full",
      ** break out of the loop even if nRemaining is still greater than
      ** zero. The caller has an in-memory tree to flush to disk.  */

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

#include <tcl.h>
#include "lsm.h"
#include "sqlite4.h"
#include <assert.h>
#include <string.h>

extern int getDbPointer(Tcl_Interp *interp, const char *zA, sqlite4 **ppDb);
extern const char *sqlite4TestErrorName(int);

/*
** TCLCMD:    sqlite4_lsm_config DB DBNAME PARAM ...
*/
static int test_sqlite4_lsm_config(
  void * clientData,
  Tcl_Interp *interp,
  int objc,
  Tcl_Obj *CONST objv[]
){
  struct Switch {
    const char *zSwitch;

  }
  return TCL_OK;
}

/*
** TCLCMD:    sqlite4_lsm_info DB DBNAME PARAM
*/
static int test_sqlite4_lsm_info(
  void * clientData,
  Tcl_Interp *interp,
  int objc,
  Tcl_Obj *CONST objv[]
){
  struct Switch {
    const char *zSwitch;

  }
  return TCL_OK;
}

/*
** TCLCMD:    sqlite4_lsm_work DB DBNAME ?SWITCHES? ?N?
*/
static int test_sqlite4_lsm_work(
  void * clientData,
  Tcl_Interp *interp,
  int objc,
  Tcl_Obj *CONST objv[]
){
  struct Switch {
    const char *zSwitch;

  Tcl_SetObjResult(interp, Tcl_NewIntObj(nWork));
  return TCL_OK;
}

/*
** TCLCMD:    sqlite4_lsm_checkpoint DB DBNAME 
*/
static int test_sqlite4_lsm_checkpoint(
  void * clientData,
  Tcl_Interp *interp,
  int objc,
  Tcl_Obj *CONST objv[]
){
  const char *zDb;
  const char *zName;

  Tcl_ResetResult(interp);
  return TCL_OK;
}

/*
** TCLCMD:    sqlite4_lsm_flush DB DBNAME 
*/
static int test_sqlite4_lsm_flush(
  void * clientData,
  Tcl_Interp *interp,
  int objc,
  Tcl_Obj *CONST objv[]
){
  const char *zDb;
  const char *zName;

    Tcl_SetResult(interp, (char *)sqlite4TestErrorName(rc), TCL_STATIC);
    return TCL_ERROR;
  }

  Tcl_ResetResult(interp);
  return TCL_OK;
}

typedef struct TclLsmCursor TclLsmCursor;
typedef struct TclLsm TclLsm;

struct TclLsm {
  lsm_db *db;
};

struct TclLsmCursor {
  lsm_cursor *csr;
};

static int test_lsm_error(Tcl_Interp *interp, const char *zApi, int rc){
  char zMsg[64];
  if( rc==LSM_OK ){
    return TCL_OK;
  }

  sprintf(zMsg, "error in %s() - %d", zApi, rc);
  Tcl_ResetResult(interp);
  Tcl_AppendResult(interp, zMsg, 0);
  return TCL_ERROR;
}

static void test_lsm_cursor_del(void *ctx){
  TclLsmCursor *pCsr = (TclLsmCursor *)ctx;
  if( pCsr ){
    lsm_csr_close(pCsr->csr);
    ckfree((char *)pCsr);
  }
}

static void test_lsm_del(void *ctx){
  TclLsm *p = (TclLsm *)ctx;
  if( p ){
    lsm_close(p->db);
    ckfree((char *)p);
  }
}

/*
** Usage: CSR sub-command ...
*/
static int test_lsm_cursor_cmd(
  void * clientData,
  Tcl_Interp *interp,
  int objc,
  Tcl_Obj *CONST objv[]
){
  struct Subcmd {
    const char *zCmd;
    int nArg;
    const char *zUsage;
  } aCmd[] = {
    /* 0 */ {"close",      0, ""},
    /* 1 */ {"seek",       2, "KEY SEEK-TYPE"},
    /* 2 */ {"first",      0, ""},
    /* 3 */ {"last",       0, ""},
    /* 4 */ {"next",       0, ""},
    /* 5 */ {"prev",       0, ""},
    /* 6 */ {"key",        0, ""},
    /* 7 */ {"value",      0, ""},
    /* 8 */ {"valid",      0, ""},
    {0, 0, 0}
  };
  int iCmd;
  int rc;
  TclLsmCursor *pCsr = (TclLsmCursor *)clientData;

  rc = Tcl_GetIndexFromObjStruct(
      interp, objv[1], aCmd, sizeof(aCmd[0]), "sub-command", 0, &iCmd
  );
  if( rc!=TCL_OK ) return rc;
  if( aCmd[iCmd].nArg>=0 && objc!=(2 + aCmd[iCmd].nArg) ){
    Tcl_WrongNumArgs(interp, 2, objv, aCmd[iCmd].zUsage);
    return TCL_ERROR;
  }

  switch( iCmd ){

    case 0: assert( 0==strcmp(aCmd[0].zCmd, "close") ); {
      Tcl_DeleteCommand(interp, Tcl_GetStringFromObj(objv[0], 0));
      return TCL_OK;
    }

    case 1: assert( 0==strcmp(aCmd[1].zCmd, "seek") ); {
      struct Seekbias {
        const char *zBias;
        int eBias;
      } aBias[] = {
        {"eq",     LSM_SEEK_EQ},
        {"le",     LSM_SEEK_LE},
        {"lefast", LSM_SEEK_LEFAST},
        {"ge",     LSM_SEEK_GE},
        {0, 0}
      };
      int iBias;
      const char *zKey; int nKey;
      zKey = Tcl_GetStringFromObj(objv[2], &nKey);

      rc = Tcl_GetIndexFromObjStruct(
          interp, objv[3], aBias, sizeof(aBias[0]), "bias", 0, &iBias
      );
      if( rc!=TCL_OK ) return rc;

      rc = lsm_csr_seek(pCsr->csr, zKey, nKey, aBias[iBias].eBias);
      return test_lsm_error(interp, "lsm_seek", rc);
    }

    case 2: 
    case 3: 
    case 4: 
    case 5: {
      const char *zApi;

      assert( 0==strcmp(aCmd[2].zCmd, "first") );
      assert( 0==strcmp(aCmd[3].zCmd, "last") );
      assert( 0==strcmp(aCmd[4].zCmd, "next") );
      assert( 0==strcmp(aCmd[5].zCmd, "prev") );

      switch( iCmd ){
        case 2: rc = lsm_csr_first(pCsr->csr); zApi = "lsm_csr_first"; break;
        case 3: rc = lsm_csr_last(pCsr->csr);  zApi = "lsm_csr_last";  break;
        case 4: rc = lsm_csr_next(pCsr->csr);  zApi = "lsm_csr_next";  break;
        case 5: rc = lsm_csr_prev(pCsr->csr);  zApi = "lsm_csr_prev";  break;
      }

      return test_lsm_error(interp, zApi, rc);
    }

    case 6: assert( 0==strcmp(aCmd[6].zCmd, "key") ); {
      const void *pKey; int nKey;
      rc = lsm_csr_key(pCsr->csr, &pKey, &nKey);
      if( rc!=LSM_OK ) test_lsm_error(interp, "lsm_csr_key", rc);

      Tcl_SetObjResult(interp, Tcl_NewStringObj((const char *)pKey, nKey));
      return TCL_OK;
    }

    case 7: assert( 0==strcmp(aCmd[7].zCmd, "value") ); {
      const void *pVal; int nVal;
      rc = lsm_csr_value(pCsr->csr, &pVal, &nVal);
      if( rc!=LSM_OK ) test_lsm_error(interp, "lsm_csr_value", rc);

      Tcl_SetObjResult(interp, Tcl_NewStringObj((const char *)pVal, nVal));
      return TCL_OK;
    }

    case 8: assert( 0==strcmp(aCmd[8].zCmd, "valid") ); {
      int bValid = lsm_csr_valid(pCsr->csr);
      Tcl_SetObjResult(interp, Tcl_NewBooleanObj(bValid));
      return TCL_OK;
    }
  }

  Tcl_AppendResult(interp, "internal error", 0);
  return TCL_ERROR;
}

/*
** Usage: DB sub-command ...
*/
static int test_lsm_cmd(
  void * clientData,
  Tcl_Interp *interp,
  int objc,
  Tcl_Obj *CONST objv[]
){
  struct Subcmd {
    const char *zCmd;
    int nArg;
    const char *zUsage;
  } aCmd[] = {
    /* 0 */ {"close",        0, ""},
    /* 1 */ {"write",        2, "KEY VALUE"},
    /* 2 */ {"delete",       1, "KEY"},
    /* 3 */ {"delete_range", 2, "START-KEY END-KEY"},
    /* 4 */ {"begin",        1, "LEVEL"},
    /* 5 */ {"commit",       1, "LEVEL"},
    /* 6 */ {"rollback",     1, "LEVEL"},
    /* 7 */ {"csr_open",     1, "CSR"},
    /* 8 */ {"work",        -1, "NPAGE ?SWITCHES?"},
    {0, 0, 0}
  };
  int iCmd;
  int rc;
  TclLsm *p = (TclLsm *)clientData;

  if( objc<2 ){
    Tcl_WrongNumArgs(interp, 1, objv, "SUB-COMMAND ...");
    return TCL_ERROR;
  }

  rc = Tcl_GetIndexFromObjStruct(
      interp, objv[1], aCmd, sizeof(aCmd[0]), "sub-command", 0, &iCmd
  );
  if( rc!=TCL_OK ) return rc;
  if( aCmd[iCmd].nArg>=0 && objc!=(2 + aCmd[iCmd].nArg) ){
    Tcl_WrongNumArgs(interp, 2, objv, aCmd[iCmd].zUsage);
    return TCL_ERROR;
  }

  switch( iCmd ){

    case 0: assert( 0==strcmp(aCmd[0].zCmd, "close") ); {
      Tcl_DeleteCommand(interp, Tcl_GetStringFromObj(objv[0], 0));
      return TCL_OK;
    }

    case 1: assert( 0==strcmp(aCmd[1].zCmd, "write") ); {
      const char *zKey; int nKey;
      const char *zVal; int nVal;

      zKey = Tcl_GetStringFromObj(objv[2], &nKey);
      zVal = Tcl_GetStringFromObj(objv[3], &nVal);

      rc = lsm_write(p->db, zKey, nKey, zVal, nVal);
      return test_lsm_error(interp, "lsm_write", rc);
    }

    case 2: assert( 0==strcmp(aCmd[2].zCmd, "delete") ); {
      const char *zKey; int nKey;

      zKey = Tcl_GetStringFromObj(objv[2], &nKey);

      rc = lsm_delete(p->db, zKey, nKey);
      return test_lsm_error(interp, "lsm_delete", rc);
    }

    case 3: assert( 0==strcmp(aCmd[3].zCmd, "delete_range") ); {
      const char *zKey1; int nKey1;
      const char *zKey2; int nKey2;

      zKey1 = Tcl_GetStringFromObj(objv[2], &nKey1);
      zKey2 = Tcl_GetStringFromObj(objv[3], &nKey2);

      rc = lsm_delete_range(p->db, zKey1, nKey1, zKey2, nKey2);
      return test_lsm_error(interp, "lsm_delete_range", rc);
    }

    case 4: 
    case 5: 
    case 6: {
      const char *zApi;
      int iLevel;

      rc = Tcl_GetIntFromObj(interp, objv[2], &iLevel);
      if( rc!=TCL_OK ) return rc;

      assert( 0==strcmp(aCmd[4].zCmd, "begin") );
      assert( 0==strcmp(aCmd[5].zCmd, "commit") );
      assert( 0==strcmp(aCmd[6].zCmd, "rollback") );
      switch( iCmd ){
        case 4: rc = lsm_begin(p->db, iLevel); zApi = "lsm_begin"; break;
        case 5: rc = lsm_commit(p->db, iLevel); zApi = "lsm_commit"; break;
        case 6: rc = lsm_rollback(p->db, iLevel); zApi = "lsm_rollback"; break;
      }

      return test_lsm_error(interp, zApi, rc);
    }

    case 7: assert( 0==strcmp(aCmd[7].zCmd, "csr_open") ); {
      const char *zCsr = Tcl_GetString(objv[2]);
      TclLsmCursor *pCsr;

      pCsr = (TclLsmCursor *)ckalloc(sizeof(TclLsmCursor));
      rc = lsm_csr_open(p->db, &pCsr->csr);
      if( rc!=LSM_OK ){
        test_lsm_cursor_del(pCsr);
        return test_lsm_error(interp, "lsm_csr_open", rc);
      }

      Tcl_CreateObjCommand(
          interp, zCsr, test_lsm_cursor_cmd, 
          (ClientData)pCsr, test_lsm_cursor_del
      );
      Tcl_SetObjResult(interp, objv[2]);
      return TCL_OK;
    }

    case 8: assert( 0==strcmp(aCmd[8].zCmd, "work") ); {
      int nWork;
      int nWrite = 0;
      int flags = 0;
      int i;

      rc = Tcl_GetIntFromObj(interp, objv[2], &nWork);
      if( rc!=TCL_OK ) return rc;

      for(i=3; i<objc; i++){
        int iOpt;
        const char *azOpt[] = { "-optimize", "-flush", 0 };

        rc = Tcl_GetIndexFromObj(interp, objv[i], azOpt, "option", 0, &iOpt);
        if( rc!=TCL_OK ) return rc;

        if( iOpt==0 ) flags |= LSM_WORK_OPTIMIZE;
        if( iOpt==1 ) flags |= LSM_WORK_FLUSH;
      }

      rc = lsm_work(p->db, flags, nWork, &nWrite);
      if( rc!=LSM_OK ) return test_lsm_error(interp, "lsm_work", rc);
      Tcl_SetObjResult(interp, Tcl_NewIntObj(nWrite));
      return TCL_OK;
    }


    default:
      assert( 0 );
  }

  Tcl_AppendResult(interp, "internal error", 0);
  return TCL_ERROR;
}

static void xLog(void *pCtx, int rc, const char *z){
  (void)(rc);
  (void)(pCtx);
  fprintf(stderr, "%s\n", z);
  fflush(stderr);
}

/*
** Usage: lsm_open DB filename ?config?
*/
static int test_lsm_open(
  void * clientData,
  Tcl_Interp *interp,
  int objc,
  Tcl_Obj *CONST objv[]
){
  TclLsm *p;
  int rc;
  const char *zDb = 0;
  const char *zFile = 0;

  if( objc!=3 && objc!=4 ){
    Tcl_WrongNumArgs(interp, 1, objv, "DB FILENAME ?CONFIG?");
    return TCL_ERROR;
  }

  zDb = Tcl_GetString(objv[1]);
  zFile = Tcl_GetString(objv[2]);

  p = (TclLsm *)ckalloc(sizeof(TclLsm));
  rc = lsm_new(0, &p->db);
  if( rc!=LSM_OK ){
    test_lsm_del((void *)p);
    test_lsm_error(interp, "lsm_new", rc);
    return TCL_ERROR;
  }

  if( objc==4 ){
    struct Lsmconfig {
      const char *zOpt;
      int eOpt;
    } aConfig[] = {
      { "write_buffer",     LSM_CONFIG_WRITE_BUFFER },
      { "page_size",        LSM_CONFIG_PAGE_SIZE },
      { "block_size",       LSM_CONFIG_BLOCK_SIZE },
      { "safety",           LSM_CONFIG_SAFETY },
      { "autowork",         LSM_CONFIG_AUTOWORK },
      { "autocheckpoint",   LSM_CONFIG_AUTOCHECKPOINT },
      { "log_size",         LSM_CONFIG_LOG_SIZE },
      { "mmap",             LSM_CONFIG_MMAP },
      { "use_log",          LSM_CONFIG_USE_LOG },
      { "nmerge",           LSM_CONFIG_NMERGE },
      { "max_freelist",     LSM_CONFIG_MAX_FREELIST },
      { "multi_proc",       LSM_CONFIG_MULTIPLE_PROCESSES },
      { 0, 0 }
    };
    int nElem;
    int i;
    Tcl_Obj **apElem;

    rc = Tcl_ListObjGetElements(interp, objv[3], &nElem, &apElem);
    for(i=0; rc==TCL_OK && i<nElem; i+=2){
      int iOpt;
      rc = Tcl_GetIndexFromObjStruct(
          interp, apElem[i], aConfig, sizeof(aConfig[0]), "option", 0, &iOpt
      );
      if( rc==TCL_OK ){
        if( i==(nElem-1) ){
          Tcl_ResetResult(interp);
          Tcl_AppendResult(interp, "option \"", Tcl_GetString(apElem[i]), 
              "\" requires an argument", 0
          );
          rc = TCL_ERROR;
        }else{
          int iVal;
          rc = Tcl_GetIntFromObj(interp, apElem[i+1], &iVal);
          if( rc==TCL_OK ){
            lsm_config(p->db, aConfig[iOpt].eOpt, &iVal);
          }
        }
      }
    }
    if( rc!=TCL_OK ){ 
      test_lsm_del((void *)p);
      return rc;
    }
  }

  lsm_config_log(p->db, xLog, 0);

  rc = lsm_open(p->db, zFile);
  if( rc!=LSM_OK ){
    test_lsm_del((void *)p);
    test_lsm_error(interp, "lsm_open", rc);
    return TCL_ERROR;
  }

  Tcl_CreateObjCommand(interp, zDb, test_lsm_cmd, (ClientData)p, test_lsm_del);
  Tcl_SetObjResult(interp, objv[1]);
  return TCL_OK;
}

int SqlitetestLsm_Init(Tcl_Interp *interp){
  struct SyscallCmd {
    const char *zName;
    Tcl_ObjCmdProc *xCmd;
  } aCmd[] = {
    { "sqlite4_lsm_work",       test_sqlite4_lsm_work                },
    { "sqlite4_lsm_checkpoint", test_sqlite4_lsm_checkpoint          },
    { "sqlite4_lsm_flush",      test_sqlite4_lsm_flush               },
    { "sqlite4_lsm_info",       test_sqlite4_lsm_info                },
    { "sqlite4_lsm_config",     test_sqlite4_lsm_config              },
    { "lsm_open",               test_lsm_open                        },
  };
  int i;

  for(i=0; i<sizeof(aCmd)/sizeof(aCmd[0]); i++){
    Tcl_CreateObjCommand(interp, aCmd[i].zName, aCmd[i].xCmd, 0, 0);
  }
  return TCL_OK;
}

  set lRes [list]
  foreach {name sys} $lSys {
    set wt [list -w $nWrite -r $nRepeat -f $nFetch -system $sys]
    lappend lRes [exec_lsmtest_speed $nSec $wt]
    if {$sys != [lindex $lSys end]} {
      puts "Sleeping 20 seconds..."
      after 20000
    }
  }

  # Set up the header part of the gnuplot script.
  #
  set xmax 0
  foreach res $lRes {

  append script $data4

  append script "pause -1\n"
  exec_gnuplot_script $script $zPng
}

do_write_test x.png 600 50000 50000 20 {
  lsm-mt-1 "mmap=1 multi_proc=0 safety=0 threads=3 autowork=0 block_size=1M"

}

# lsm-mt    "mmap=1 multi_proc=0 threads=2 autowork=0 autocheckpoint=8192000"
# lsm-mt     "mmap=1 multi_proc=0 safety=1 threads=3 autowork=0"
# lsm-st     "mmap=1 multi_proc=0 safety=1 threads=1 autowork=1"
# lsm-mt     "mmap=1 multi_proc=0 safety=1 threads=3 autowork=0"
# lsm-mt     "mmap=1 multi_proc=0 safety=1 threads=3 autowork=0"