TestDb *testOpen(const char *zSystem, int, int *pRc);
void testReopen(TestDb **ppDb, int *pRc);
void testClose(TestDb **ppDb);

void testFetch(TestDb *, void *, int, void *, int, int *);
void testWrite(TestDb *, void *, int, void *, int, int *);
void testDelete(TestDb *, void *, int, int *);
void testDeleteRange(TestDb *, void *, int, void *, int, int *);
void testWriteStr(TestDb *, const char *, const char *zVal, int *pRc);
void testFetchStr(TestDb *, const char *, const char *, int *pRc);

void testBegin(TestDb *pDb, int iTrans, int *pRc);
void testCommit(TestDb *pDb, int iTrans, int *pRc);

void test_failed(void);

void testDeleteDatasource(TestDb *, Datasource *, int, int *);
void testDeleteDatasourceRange(TestDb *, Datasource *, int, int, int *);


/* test1.c */
void test_data_1(const char *, const char *, int *pRc);
void test_data_2(const char *, const char *, int *pRc);
void test_data_3(const char *, const char *, int *pRc);
void testDbContents(TestDb *, Datasource *, int, int, int, int, int, int *);
void testCaseProgress(int, int, int, int *);
int testCaseNDot(void);

typedef struct CksumDb CksumDb;
CksumDb *testCksumArrayNew(Datasource *, int, int, int);
char *testCksumArrayGet(CksumDb *, int);

    char *zName = getName2(zSystem, &aTest[i]);
    if( testCaseBegin(pRc, zPattern, "%s", zName) ){
      doDataTest2(zSystem, &aTest[i], pRc);
    }
    testFree(zName);
  }
}

/*************************************************************************
** Test case data3.*
*/

typedef struct Datatest3 Datatest3;
struct Datatest3 {
  int nRange;                     /* Keys are between 1 and this value, incl. */
  int nIter;                      /* Number of iterations */
  int nWrite;                     /* Number of writes per iteration */
  int nDelete;                    /* Number of deletes per iteration */

  int nValMin;                    /* Minimum value size for writes */
  int nValMax;                    /* Maximum value size for writes */
};

void testPutU32(u8 *aBuf, u32 iVal){
  aBuf[0] = (iVal >> 24) & 0xFF;
  aBuf[1] = (iVal >> 16) & 0xFF;
  aBuf[2] = (iVal >>  8) & 0xFF;
  aBuf[3] = (iVal >>  0) & 0xFF;
}

void dt3PutKey(u8 *aBuf, int iKey){
  assert( iKey<100000 && iKey>=0 );
  sprintf((char *)aBuf, "%.5d", iKey);
}

static void doDataTest3(
  const char *zSystem,            /* Database system to test */
  Datatest3 *p,                   /* Structure containing test parameters */
  int *pRc                        /* OUT: Error code */
){
  int iDot = 0;
  int rc = *pRc;
  TestDb *pDb;
  u8 *abPresent;                  /* Array of boolean */
  char *aVal;                     /* Buffer to hold values */
  int i;
  u32 iSeq = 10;                  /* prng counter */

  abPresent = (u8 *)testMalloc(p->nRange+1);
  aVal = (char *)testMalloc(p->nValMax+1);
  pDb = testOpen(zSystem, 1, &rc);

  for(i=0; i<p->nIter && rc==0; i++){
    int ii;

    testCaseProgress(i, p->nIter, testCaseNDot(), &iDot);

    /* Perform nWrite inserts */
    for(ii=0; ii<p->nWrite; ii++){
      u8 aKey[6];
      u32 iKey;
      int nVal;

      iKey = (testPrngValue(iSeq++) % p->nRange) + 1;
      nVal = (testPrngValue(iSeq++) % (p->nValMax - p->nValMin)) + p->nValMin;
      testPrngString(testPrngValue(iSeq++), aVal, nVal);
      dt3PutKey(aKey, iKey);

      testWrite(pDb, aKey, sizeof(aKey)-1, aVal, nVal, &rc);
      abPresent[iKey] = 1;
    }

    /* Perform nDelete deletes */
    for(ii=0; ii<p->nDelete; ii++){
      u8 aKey1[6];
      u8 aKey2[6];
      u32 iKey;

      iKey = (testPrngValue(iSeq++) % p->nRange) + 1;
      dt3PutKey(aKey1, iKey-1);
      dt3PutKey(aKey2, iKey+1);

      testDeleteRange(pDb, aKey1, sizeof(aKey1)-1, aKey2, sizeof(aKey2)-1, &rc);
      abPresent[iKey] = 0;
    }

    testReopen(&pDb, &rc);

    for(ii=1; rc==0 && ii<=p->nRange; ii++){
      int nDbVal;
      void *pDbVal;
      u8 aKey[6];
      int dbrc;

      dt3PutKey(aKey, ii);
      dbrc = tdb_fetch(pDb, aKey, sizeof(aKey)-1, &pDbVal, &nDbVal);
      testCompareInt(0, dbrc, &rc);

      if( abPresent[ii] ){
        testCompareInt(1, (nDbVal>0), &rc);
      }else{
        testCompareInt(1, (nDbVal<0), &rc);
      }
    }
  }

  testClose(&pDb);
  testCaseFinish(rc);
  *pRc = rc;
}

static char *getName3(const char *zSystem, Datatest3 *p){
  return testMallocPrintf("data3.%s.%d.%d.%d.%d.(%d..%d)",
      zSystem, p->nRange, p->nIter, p->nWrite, p->nDelete, 
      p->nValMin, p->nValMax
  );
}

void test_data_3(
  const char *zSystem,            /* Database system name */
  const char *zPattern,           /* Run test cases that match this pattern */
  int *pRc                        /* IN/OUT: Error code */
){
  Datatest3 aTest[] = {
    /* nRange, nIter, nWrite, nDelete, nValMin, nValMax */
    {  100,    1000,  5,      5,       50,      100 },
    {  100,    1000,  2,      2,        5,       10 },
  };

  int i;

  for(i=0; *pRc==LSM_OK && i<ArraySize(aTest); i++){
    char *zName = getName3(zSystem, &aTest[i]);
    if( testCaseBegin(pRc, zPattern, "%s", zName) ){
      doDataTest3(zSystem, &aTest[i], pRc);
    }
    testFree(zName);
  }
}

  }

  for(j=0; tdb_system_name(j); j++){
    rc = 0;

    test_data_1(tdb_system_name(j), zPattern, &rc);
    test_data_2(tdb_system_name(j), zPattern, &rc);
    test_data_3(tdb_system_name(j), zPattern, &rc);
    test_rollback(tdb_system_name(j), zPattern, &rc);
    test_mc(tdb_system_name(j), zPattern, &rc);
    test_mt(tdb_system_name(j), zPattern, &rc);

    if( rc ) nFail++;
  }

    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 ){

**   can be found. This is usually used to optimize the database by 
**   merging the whole thing into one big array.
*/
int lsm_work(lsm_db *pDb, int flags, int nPage, int *pnWrite);

#define LSM_WORK_FLUSH           0x00000001
#define LSM_WORK_OPTIMIZE        0x00000002

int lsm_flush(lsm_db *pDb);

/*
** Attempt to checkpoint the current database snapshot. Return an LSM
** error code if an error occurs or LSM_OK otherwise.
**
** If the current snapshot has already been checkpointed, calling this 
** function is a no-op. In this case if pnByte is not NULL, *pnByte is

/* 
** Functions from file "lsm_tree.c".
*/
int lsmTreeNew(lsm_env *, int (*)(void *, int, void *, int), Tree **ppTree);
void lsmTreeRelease(lsm_env *, Tree *);

int lsmTreeInit(lsm_db *);
int lsmTreeRepair(lsm_db *);

void lsmTreeMakeOld(lsm_db *pDb);
void lsmTreeDiscardOld(lsm_db *pDb);
int lsmTreeHasOld(lsm_db *pDb);

          rc = lsmEnvWrite(pFS->pEnv, pFS->fdDb, iWrite-4, aPtr, sizeof(aPtr));
          if( nRem>0 ) iApp = iWrite;
        }
      }else{
        /* The next block is already allocated. */
        assert( nRem>0 );
        rc = fsBlockNext(pFS, fsPageToBlock(pFS, iApp), &iBlk);
        iRet = iApp = fsFirstPageOnBlock(pFS, iBlk);
      }

      /* Write the remaining data into the new block */
      if( rc==LSM_OK && nRem>0 ){
        rc = lsmEnvWrite(pFS->pEnv, pFS->fdDb, iApp, &aData[nWrite], nRem);
        iApp += nRem;
      }

**   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==iKey+1 && aEntry[i+1].iId<0 ){
              eType |= LSM_END_DELETE;
            }

          }else{
            eType = LSM_START_DELETE|LSM_SYSTEMKEY;
            iKey = aEntry[i].iBlk + 1;
          }

          /* If the in-memory entry immediately after this one is a
          ** DELETE, and the block number is one less than the current
          ** key, mark this entry as an "start-delete-range".  */
          if( i>0 && aEntry[i-1].iBlk==iKey-1 && aEntry[i-1].iId<0 ){
            eType |= LSM_START_DELETE;
          }

          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
  );

  /* If a key has the LSM_START_DELETE flag set, but not the LSM_INSERT or
  ** LSM_POINT_DELETE flags, it is considered a delta larger. This prevents
  ** the beginning of an open-ended set from masking a database entry or
  ** delete at a lower level.  */
  if( res==0 && (pCsr->flags & CURSOR_IGNORE_DELETE) ){
    const int m = LSM_POINT_DELETE|LSM_INSERT|LSM_END_DELETE |LSM_START_DELETE;
    int iDel1 = 0;
    int iDel2 = 0;

    if( LSM_START_DELETE==(iLhsFlags & m) ) iDel1 = +1;
    if( LSM_END_DELETE  ==(iLhsFlags & m) ) iDel1 = -1;
    if( LSM_START_DELETE==(iRhsFlags & m) ) iDel2 = +1;
    if( LSM_END_DELETE  ==(iRhsFlags & m) ) iDel2 = -1;

    res = (iDel1 - iDel2);
  }

  return res;
}

static void multiCursorDoCompare(MultiCursor *pCsr, int iOut, int bReverse){

    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 ){
      /* The two keys are identical. Normally, this means that the key from
      ** the newer run clobbers the old. However, if the newer key is a
      ** separator key, or a range-delete-boundary only, do not allow it
      ** to clobber an older entry.  */
      int nc1 = (eType1 & (LSM_INSERT|LSM_POINT_DELETE))==0;
      int nc2 = (eType2 & (LSM_INSERT|LSM_POINT_DELETE))==0;
      iRes = (nc1 > nc2) ? i2 : i1;
    }else if( res<0 ){
      iRes = i1;
    }else{
      iRes = i2;
    }
  }

        *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;
    }

  }
  if( (pCsr->flags & CURSOR_IGNORE_SYSTEM) && rtTopic(eType)!=0 ){
    return 0;
  }

  /* Check if this key has already been deleted by a range-delete */
  iKey = pCsr->aTree[1];
  for(i=0; i<iKey; i++){
    int csrflags;
    multiCursorGetKey(pCsr, i, &csrflags, 0, 0);
    if( (rdmask & csrflags) ){
      const int SD_ED = (LSM_START_DELETE|LSM_END_DELETE);
      if( (csrflags & SD_ED)==SD_ED 
       || (pCsr->flags & CURSOR_IGNORE_DELETE)==0
      ){
        void *pKey; int nKey;
        multiCursorGetKey(pCsr, i, 0, &pKey, &nKey);
        if( 0==sortedKeyCompare(pCsr->pDb->xCmp,
              rtTopic(eType), pCsr->key.pData, pCsr->key.nData,
              rtTopic(csrflags), pKey, nKey
        )){
          continue;
        }
      }
      return 0;
    }
  }

#if 0
  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;
    }
  }
#endif

  return 1;
}

static int multiCursorEnd(MultiCursor *pCsr, int bLast){
  int rc = LSM_OK;
  int i;

  /* 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)) );

  return rc;
}

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 */
  void *pVal, int nVal,           /* Value value */
  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 */
  int nData;                      /* Size of buffer aData[] in bytes */
  int nRec;                       /* Number of records on page pPg */
  int iFPtr;                      /* Value of pointer in footer of pPg */
  int iRPtr = 0;                  /* Value of pointer written into record */
  int iOff;                       /* Current write offset within page pPg */
  Segment *pSeg;                  /* Segment being written */
  int flags = 0;                  /* If != 0, flags value for page footer */
  int bFirst = 0;                 /* True for first key of output run */



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

  if( pSeg->iFirst==0 && pMW->pPage==0 ){
    rc = mergeWorkerFirstPage(pMW);
    bFirst = 1;

  ** The header space is:
  **
  **     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 ){
    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. */
    iOff = pMerge->iOutputOff;
    if( iOff<0 || pPg==0 || iOff+nHdr > SEGMENT_EOF(nData, nRec+1) ){
      iFPtr = *pMW->pCsr->pPrevMergePtr;
      iRPtr = iPtr - iFPtr;
      iOff = 0;
      nRec = 0;
      rc = mergeWorkerNextPage(pMW, iFPtr);
      pPg = pMW->pPage;
    }
  }

    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( 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{
    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 ){
            if( f & (LSM_INSERT|LSM_POINT_DELETE) ){
              f |= (LSM_END_DELETE|LSM_START_DELETE);
            }else{
              f = 0;
            }
            break;
          }
        }
      }
    }

    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( rc==LSM_OK && (rtIsSeparator(eType)==0 || iPtr!=0) ){
      /* Write the record into the main run. */
      void *pVal; int nVal;
      rc = multiCursorGetVal(pCsr, iVal, &pVal, &nVal);
      if( pVal && rc==LSM_OK ){
        assert( nVal>=0 );
        rc = sortedBlobSet(pDb->pEnv, &pCsr->val, pVal, nVal);
        pVal = pCsr->val.pData;
      }
      if( rc==LSM_OK ){
        rc = mergeWorkerWrite(pMW, eType, pKey, nKey, pVal, nVal, iPtr);
      }
    }
  }

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

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

#if 0
    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 0
      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.  */

  /* This function may not be called if pDb has an open read or write
  ** transaction. Return LSM_MISUSE if an application attempts this.  */
  if( pDb->nTransOpen || pDb->pCsr ) return LSM_MISUSE_BKPT;

  return doLsmWork(pDb, flags, nPage, pnWrite);
}

int lsm_flush(lsm_db *db){
  int rc;

  if( db->nTransOpen>0 || db->pCsr ){
    rc = LSM_MISUSE_BKPT;
  }else{
    rc = lsmBeginWriteTrans(db);
    if( rc==LSM_OK ){
      lsmFlushTreeToDisk(db);
      lsmTreeDiscardOld(db);
      lsmTreeMakeOld(db);
      lsmTreeDiscardOld(db);
    }

    if( rc==LSM_OK ){
      rc = lsmFinishWriteTrans(db, 1);
    }else{
      lsmFinishWriteTrans(db, 0);
    }
    lsmFinishReadTrans(db);
  }

  return rc;
}

/*
** This function is called in auto-work mode to perform merging work on
** the data structure. It performs enough merging work to prevent the
** height of the tree from growing indefinitely assuming that roughly
** nUnit database pages worth of data have been written to the database
** (i.e. the in-memory tree) since the last call.

      rc = treeUpdatePtr(pDb, pCsr, iNew);
    }
  }

  return rc;
}












void lsmTreeMakeOld(lsm_db *pDb){
  if( pDb->treehdr.iOldShmid==0 ){
    pDb->treehdr.iOldLog = pDb->treehdr.log.aRegion[2].iEnd;
    pDb->treehdr.oldcksum0 = pDb->treehdr.log.cksum0;
    pDb->treehdr.oldcksum1 = pDb->treehdr.log.cksum1;
    pDb->treehdr.iOldShmid = pDb->treehdr.iNextShmid-1;
    memcpy(&pDb->treehdr.oldroot, &pDb->treehdr.root, sizeof(TreeRoot));

# 2012 November 02
#
# 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.
#
#***********************************************************************
#
set testdir [file dirname $argv0]
source $testdir/tester.tcl
set testprefix lsm1
db close


proc reopen {{bClear 0}} {
  catch {db close}
  if {$bClear} { forcedelete test.db }
  lsm_open db test.db {mmap 0 nmerge 2 autowork 0}
}

proc contents {} {
  db csr_open csr
  set res [list]
  for {csr first} {[csr valid]} {csr next} {
    lappend res [list [csr key] [csr value]]
  }
  csr close
  set res
}

proc fetch {key} {
  db csr_open csr
  csr seek $key eq
  set val [csr value]
  csr close
  set val
}

proc dbwrite {list} {
  foreach {k v} $list {
    db write $k $v
  }
}

proc do_contents_test {tn res} {
  set con [contents]
  set res2 [list]
  foreach r $res {lappend res2 $r}
  uplevel do_test $tn [list [list set {} $con]] [list $res2]
}

if 1 {

do_test 1.1 {
  reopen
  db write abc def
  db close
} {}

do_test 1.2 {
  reopen
  db csr_open csr
  csr seek abc eq
} {}

do_test 1.3 { 
  list [csr valid] [csr key] [csr value] 
} {1 abc def}

do_test 1.4 { 
  db delete abc
  csr seek abc eq
  csr valid
} {0}

do_test 1.5 { csr close } {}
do_test 1.6 { db close  } {}


do_test 2.1 {
  forcedelete test.db
  reopen
  db write aaa one
  db write bbb two
  db write ccc three
  db write ddd four
  db write eee five
  db write fff six
  reopen
  db delete_range a bbb
  reopen
  db work 10 
} {1}

do_contents_test 2.2 { {bbb two} {ccc three} {ddd four} {eee five} {fff six} }


#-------------------------------------------------------------------------

# The following populates the db with a single age=1 segment, containing
# the six keys inserted below.
do_test 3.1 {
  reopen 1
  db write aaa one
  db write ddd four
  db write fff six
  reopen
  db write bbb two
  db write ccc three
  db write eee five
  reopen
  db work 10
} {1}

do_test 3.2 {
  db write bx seven
  reopen
  db delete_range aaa bx
  reopen
  db work 10
} {2}

do_contents_test 3.3 { 
  {aaa one} {bx seven} {ccc three} {ddd four} {eee five} {fff six}
}

do_test 3.4 { fetch ddd } four

#-------------------------------------------------------------------------
#
do_test 4.1 {
  reopen 1
  dbwrite { 222 helloworld }
  db flush
  db delete_range 111 222
  db delete_range 222 333
  db flush
  contents
} {{222 helloworld}}

do_test 4.2 { fetch 222 } helloworld

#-------------------------------------------------------------------------
#
do_test 5.1 {
  reopen 1

  dbwrite { 10 ten    }  ; db flush
  dbwrite { 20 twenty }  ; db flush
  db work 10

  dbwrite { 30 thirty }  ; db flush
  dbwrite { 40 forty  }  ; db flush
  db work 10

  db delete_range 11 29  ; db flush
  db delete_range 20 39  ; db flush
  db work 10

  contents
} {{10 ten} {40 forty}}

do_test 5.2 {
  reopen 1
  db config {nmerge 4}

  dbwrite { 10 ten    }  ; db flush
  dbwrite { 20 twenty }  ; db flush
  dbwrite { 30 thirty }  ; db flush
  dbwrite { 40 forty  }  ; db flush
  db work 10

  db delete_range 10 17  ; db flush
  dbwrite {17 seventeen} ; db flush
  db delete_range 10 17  ; db flush

  db config {nmerge 3}
  db work 10

  contents
} {{10 ten} {17 seventeen} {20 twenty} {30 thirty} {40 forty}}

}

do_test 5.3 {
  reopen 1
  db config {nmerge 4}

  dbwrite { 10 ten    }  ; db flush
  dbwrite { 20 twenty }  ; db flush
  dbwrite { 30 thirty }  ; db flush
  dbwrite { 40 forty  }  ; db flush
  db work 10

  db delete_range 10 17  ; db flush
  db delete_range 12 19  ; db flush
  dbwrite {17 seventeen} ; db flush

  db config {nmerge 3}
  db work 10

  contents
} {{10 ten} {17 seventeen} {20 twenty} {30 thirty} {40 forty}}

finish_test

#
lappend ::testsuitelist xxx

test_suite "src4" -prefix "" -description {
} -files {
  simple.test simple2.test
  log3.test 
  lsm1.test
  csr1.test
  ckpt1.test
  mc1.test

  aggerror.test
  attach.test
  autoindex1.test

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

#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;
}

static int testConfigureLsm(Tcl_Interp *interp, lsm_db *db, Tcl_Obj *pObj){
  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;
  int rc;

  rc = Tcl_ListObjGetElements(interp, pObj, &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(db, aConfig[iOpt].eOpt, &iVal);
        }
      }
    }
  }

  return rc;
}

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?"},
    /*  9 */ {"flush",        0, ""},
    /* 10 */ {"config",       1, "LIST"},
    {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;
    }

    case 9: assert( 0==strcmp(aCmd[9].zCmd, "flush") ); {
      rc = lsm_flush(p->db);
      return test_lsm_error(interp, "lsm_flush", rc);
    }

    case 10: assert( 0==strcmp(aCmd[10].zCmd, "config") ); {
      return testConfigureLsm(interp, p->db, objv[2]);
    }

    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 ){
    rc = testConfigureLsm(interp, p->db, objv[3]);
    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"