SQLite4  Check-in [21db5f73f5]

#include "lsmtest_tdb.h"
#include "lsm.h"

#include "lsmtest.h"

#include <stdlib.h>
#include <string.h>
#include <assert.h>
#include <unistd.h>
#include <stdio.h>

#include <sys/time.h>

typedef struct LsmDb LsmDb;
typedef struct LsmWorker LsmWorker;
typedef struct LsmFile LsmFile;

#ifdef LSM_MUTEX_PTHREADS
#include <pthread.h>

#define LSMTEST_THREAD_CKPT      1
#define LSMTEST_THREAD_WORKER    2
#define LSMTEST_THREAD_WORKER_AC 3

/*
** There are several different types of worker threads that run in different
** test configurations, depending on the value of LsmWorker.eType.
**
**   1. Checkpointer.
**   2. Worker with auto-checkpoint.
**   3. Worker without auto-checkpoint.
*/
struct LsmWorker {
  LsmDb *pDb;                     /* Main database structure */
  lsm_db *pWorker;                /* Worker database handle */
  pthread_t worker_thread;        /* Worker thread */
  pthread_cond_t worker_cond;     /* Condition var the worker waits on */
  pthread_mutex_t worker_mutex;   /* Mutex used with worker_cond */
  int bDoWork;                    /* Set to true by client when there is work */
  int worker_rc;                  /* Store error code here */
  int eType;                      /* LSMTEST_THREAD_XXX constant */


  int bBlock;
};
#else
struct LsmWorker { int worker_rc; int bBlock; };
#endif

static void mt_shutdown(LsmDb *);

lsm_env *tdb_lsm_env(void){
  static int bInit = 0;
  static lsm_env env;

  return rc;
}

static int test_lsm_write(
  TestDb *pTestDb, 
  void *pKey, 
  int nKey, 
  void *pVal,
  int nVal
){
  LsmDb *pDb = (LsmDb *)pTestDb;

  int nSleep = 0;
  while( pDb->aWorker && pDb->aWorker[0].bBlock ){
    usleep(1000);
    nSleep++;
  }
#if 0
  if( nSleep ) printf("nSleep=%d\n", nSleep);
  if( pDb->aWorker ){
    int nLimit = -1;
    int nSleep = 0;
    lsm_config(pDb->db, LSM_CONFIG_AUTOFLUSH, &nLimit);
    do {
      int bOld, nNew, rc;
      rc = lsm_info(pDb->db, LSM_INFO_TREE_SIZE, &bOld, &nNew);
      if( rc!=LSM_OK ) return rc;
      if( bOld==0 || nNew<nLimit ) break;
      usleep(1000);
      nSleep += 1;
    }while( 1 );
#if 0
    if( nSleep ) printf("nSleep=%d\n", nSleep);
#endif
  }
#endif

  return lsm_insert(pDb->db, pKey, nKey, pVal, nVal);
}

static int test_lsm_delete(TestDb *pTestDb, void *pKey, int nKey){
  LsmDb *pDb = (LsmDb *)pTestDb;
  return lsm_delete(pDb->db, pKey, nKey);

    lsm_config_log(pDb->db, xLog, 0);
    lsm_config_work_hook(pDb->db, xWorkHook, (void *)pDb);

    rc = test_lsm_config_str(pDb, pDb->db, 0, zCfg, &nThread);
    if( rc==LSM_OK ) rc = lsm_open(pDb->db, zFilename);

#ifdef LSM_MUTEX_PTHREADS
    if( rc==LSM_OK && nThread>1 ){
      testLsmStartWorkers(pDb, nThread, zFilename, zCfg);
    }
#endif

    if( rc!=LSM_OK ){
      test_lsm_close((TestDb *)pDb);
      pDb = 0;
    }

  }
  return 0;
}

void tdb_lsm_enable_log(TestDb *pDb, int bEnable){
  lsm_db *db = tdb_lsm(pDb);
  if( db ){


    lsm_config_log(db, (bEnable ? xLog : 0), (void *)"client");
  }
}

void tdb_lsm_application_crash(TestDb *pDb){
  if( tdb_lsm(pDb) ){
    LsmDb *p = (LsmDb *)pDb;

  LsmWorker *p = &pDb->aWorker[iWorker];
  pthread_mutex_lock(&p->worker_mutex);
  p->bDoWork = 1;
  pthread_cond_signal(&p->worker_cond);
  pthread_mutex_unlock(&p->worker_mutex);
}

/*
** This routine is used as the main() for all worker threads.
*/
static void *worker_main(void *pArg){
  LsmWorker *p = (LsmWorker *)pArg;
  lsm_db *pWorker;                /* Connection to access db through */

  pthread_mutex_lock(&p->worker_mutex);
  while( (pWorker = p->pWorker) ){
    int rc = LSM_OK;
    int nCkpt = -1;

    /* Do some work. If an error occurs, exit. */
    pthread_mutex_unlock(&p->worker_mutex);

    if( p->eType==LSMTEST_THREAD_CKPT ){
      int nByte = 0;
      rc = lsm_info(pWorker, LSM_INFO_CHECKPOINT_SIZE, &nByte);
      if( rc==LSM_OK && nByte>=(2*1024*1024) ){
        if( nByte>(8*1024*1024) ) p->bBlock = 1;
        rc = lsm_checkpoint(pWorker, 0);
        p->bBlock = 0;
      }
    }else{
      int nWrite = 0;             /* Pages written by lsm_work() call */
      int nAuto = -1;             /* Configured AUTOCHECKPOINT value */
      int nLimit = -1;            /* Configured AUTOFLUSH value */

      lsm_config(pWorker, LSM_CONFIG_AUTOFLUSH, &nLimit);
      lsm_config(pWorker, LSM_CONFIG_AUTOCHECKPOINT, &nAuto);
      do {

        lsm_info(pWorker, LSM_INFO_CHECKPOINT_SIZE, &nCkpt);
#if 0
        int nSleep = 0;
        while( nAuto==0 && nCkpt>(nLimit*4) ){
          usleep(1000);
          mt_signal_worker(p->pDb, 1);
          nSleep++;
          lsm_info(pWorker, LSM_INFO_CHECKPOINT_SIZE, &nCkpt);
        }

          if( nSleep ) printf("nLimit=%d nSleep=%d (worker)\n", nLimit, nSleep);
#endif

        rc = lsm_work(pWorker, 0, 256, &nWrite);
        if( p->eType==LSMTEST_THREAD_WORKER_AC && nWrite && rc==LSM_OK ){
          mt_signal_worker(p->pDb, 1);
        }
      }while( nWrite && p->pWorker );
    }
    pthread_mutex_lock(&p->worker_mutex);

    if( rc!=LSM_OK && rc!=LSM_BUSY ){
      p->worker_rc = rc;
      break;

** This callback is invoked by LSM when the client database writes to
** the database file (i.e. to flush the contents of the in-memory tree).
** This implies there may be work to do on the database, so signal
** the worker threads.
*/
static void mt_client_work_hook(lsm_db *db, void *pArg){
  LsmDb *pDb = (LsmDb *)pArg;     /* LsmDb database handle */


  /* Invoke the user level work-hook, if any. */
  if( pDb->xWork ) pDb->xWork(db, pDb->pWorkCtx);

  /* Signal the lsm_work() thread */
  mt_signal_worker(pDb, 0);
}

/*
** Launch worker thread iWorker for database connection pDb.
*/
static int mt_start_worker(
  LsmDb *pDb,                     /* Main database structure */
  int iWorker,                    /* Worker number to start */
  const char *zFilename,          /* File name of database to open */
  const char *zCfg,               /* Connection configuration string */
  int eType                       /* Type of worker thread */


){
  int rc = 0;                     /* Return code */
  LsmWorker *p;                   /* Object to initialize */

  assert( iWorker<pDb->nWorker );
  assert( eType==LSMTEST_THREAD_CKPT 
       || eType==LSMTEST_THREAD_WORKER 
       || eType==LSMTEST_THREAD_WORKER_AC 
  );

  p = &pDb->aWorker[iWorker];


  p->eType = eType;
  p->pDb = pDb;

  /* Open the worker connection */
  if( rc==0 ) rc = lsm_new(&pDb->env, &p->pWorker);
  if( zCfg ){
    test_lsm_config_str(pDb, p->pWorker, 1, zCfg, 0);
  }

  if( rc==0 ) rc = pthread_create(&p->worker_thread, 0, worker_main, (void *)p);

  return rc;
}


static int testLsmStartWorkers(
  LsmDb *pDb, int eModel, const char *zFilename, const char *zCfg
){
  int rc;

  if( eModel<1 || eModel>4 ) return 1;
  if( eModel==1 ) return 0;

  /* Configure a work-hook for the client connection. Worker 0 is signalled
  ** every time the users connection writes to the database.  */
  lsm_config_work_hook(pDb->db, mt_client_work_hook, (void *)pDb);

  /* Allocate space for two worker connections. They may not both be
  ** used, but both are allocated.  */
  pDb->aWorker = (LsmWorker *)testMalloc(sizeof(LsmWorker) * 2);
  memset(pDb->aWorker, 0, sizeof(LsmWorker) * 2);


  switch( eModel ){
    case 2:
      pDb->nWorker = 1;
      test_lsm_config_str(0, pDb->db, 0, "autocheckpoint=0", 0);
      rc = mt_start_worker(pDb, 0, zFilename, zCfg, LSMTEST_THREAD_CKPT);
      break;


    case 3:
      pDb->nWorker = 2;
      assert( 0 );
      break;

    case 4:
      pDb->nWorker = 2;
      assert( 0 );
      break;
  }

  return rc;
}


int test_lsm_mt2(const char *zFilename, int bClear, TestDb **ppDb){

#define LSM_INFO_ARRAY_STRUCTURE  5
#define LSM_INFO_PAGE_ASCII_DUMP  6
#define LSM_INFO_PAGE_HEX_DUMP    7
#define LSM_INFO_FREELIST         8
#define LSM_INFO_ARRAY_PAGES      9
#define LSM_INFO_CHECKPOINT_SIZE 10
#define LSM_INFO_TREE_SIZE       11

#define LSM_INFO_FREELIST_SIZE   12


/* 
** CAPI: Opening and Closing Write Transactions
**
** These functions are used to open and close transactions and nested 
** sub-transactions.

int lsmLogCommit(lsm_db *);
void lsmLogEnd(lsm_db *pDb, int bCommit);
void lsmLogTell(lsm_db *, LogMark *);
void lsmLogSeek(lsm_db *, LogMark *);

int lsmLogRecover(lsm_db *);
int lsmInfoLogStructure(lsm_db *pDb, char **pzVal);



/**************************************************************************
** Functions from file "lsm_shared.c".
*/

int lsmDbDatabaseConnect(lsm_db*, const char *);

**     2. The checkpoint id LSW.
**     3. The number of integer values in the entire checkpoint, including 
**        the two checksum values.
**     4. The total number of blocks in the database.
**     5. The block size.
**     6. The number of levels.
**     7. The nominal database page size.
**     8. The number of pages (in total) written to the database file.
**
**   Log pointer:
**
**     1. The log offset MSW.
**     2. The log offset LSW.
**     3. Log checksum 0.
**     4. Log checksum 1.

**     9. Current pointer value (64-bits - 2 integers).
**
**   The in-memory freelist entries. Each entry is either an insert or a
**   delete. The in-memory freelist is to the free-block-list as the
**   in-memory tree is to the users database content.
**
**     1. Number of free-list entries stored in checkpoint header.
**     2. Number of free blocks (in total).
**     3. Total number of blocks freed during database lifetime.
**     4. For each entry:
**        2a. Block number of free block.
**        2b. A 64-bit integer (MSW followed by LSW). -1 for a delete entry,
**            or the associated checkpoint id for an insert.
**
**   The checksum:
**
**     1. Checksum value 1.

  int rc = LSM_OK;

  assert( iPg>=fsFirstPageOnBlock(pFS, 1) );
  *ppPg = 0;

  assert( pFS->bUseMmap==0 || pFS->pCompress==0 );
  if( pFS->bUseMmap ){
    Page *pTest;
    i64 iEnd = (i64)iPg * pFS->nPagesize;
    fsGrowMapping(pFS, iEnd, &rc);
    if( rc!=LSM_OK ) return rc;

    p = 0;
    for(pTest=pFS->pWaiting; pTest; pTest=pTest->pNextWaiting){
      if( pTest->iPg==iPg ){
        p = pTest;
        p->nRef++;
        *ppPg = p;
        return LSM_OK;
      }
    }
    if( pFS->pFree ){
      p = pFS->pFree;
      pFS->pFree = p->pHashNext;
      assert( p->nRef==0 );
    }else{
      p = lsmMallocZeroRc(pFS->pEnv, sizeof(Page), &rc);
      if( rc ) return rc;

        ** lsmFsPagePersist() to write an out-of-order page. Instead a page 
        ** number is assigned here so that the page data will be appended
        ** to the current segment.
        */
        Page **pp;
        int iPrev = 0;
        int iNext = 0;


        assert( pPg->pSeg->iFirst );
        assert( pPg->flags & PAGE_FREE );
        assert( (pPg->flags & PAGE_HASPREV)==0 );
        assert( pPg->nData==pFS->nPagesize-4 );

        rc = fsAppendPage(pFS, pPg->pSeg, &pPg->iPg, &iPrev, &iNext);
        if( rc!=LSM_OK ) return rc;

        if( pFS->bUseMmap==0 ){
          int iHash = fsHashKey(pFS->nHash, pPg->iPg);
          pPg->pHashNext = pFS->apHash[iHash];
          pFS->apHash[iHash] = pPg;
          assert( pPg->pHashNext==0 || pPg->pHashNext->iPg!=pPg->iPg );
        }

        if( iPrev ){
          assert( iNext==0 );
          memmove(&pPg->aData[4], pPg->aData, pPg->nData);
          lsmPutU32(pPg->aData, iPrev);
          pPg->flags |= PAGE_HASPREV;
          pPg->aData += 4;

  int i;
  int rc;
  Freelist freelist = {0, 0, 0};
  u8 *aUsed;
  Level *pLevel;
  Snapshot *pWorker = pDb->pWorker;
  int nBlock = pWorker->nBlock;

#if 0 
  static int nCall = 0;
  nCall++;
  printf("%d calls\n", nCall);
#endif

  aUsed = lsmMallocZero(pDb->pEnv, nBlock);
  if( aUsed==0 ){
    /* Malloc has failed. Since this function is only called within debug
    ** builds, this probably means the user is running an OOM injection test.
    ** Regardless, it will not be possible to run the integrity-check at this
    ** time, so assume the database is Ok and return non-zero. */

static int infoFreelistCb(void *pCtx, int iBlk, i64 iSnapshot){
  LsmString *pStr = (LsmString *)pCtx;
  lsmStringAppendf(pStr, "%s{%d %lld}", (pStr->n?" ":""), iBlk, iSnapshot);
  return 0;
}

static int infoFreelist(lsm_db *pDb, char **pzOut){
  Snapshot *pWorker;              /* Worker snapshot */
  int bUnlock = 0;
  LsmString s;
  int i;
  int rc;

  /* Obtain the worker snapshot */

    *pzOut = s.z;
  }

  /* Release the snapshot and return */
  infoFreeWorker(pDb, bUnlock);
  return rc;
}

static int infoFreelistSize(lsm_db *pDb, int *pnFree, int *pnWaiting){
}

static int infoTreeSize(lsm_db *db, int *pnOld, int *pnNew){
  ShmHeader *pShm = db->pShmhdr;
  TreeHeader *p = &pShm->hdr1;

  /* The following code suffers from two race conditions, as it accesses and
  ** trusts the contents of shared memory without verifying checksums:

      char **pzVal = va_arg(ap, char **);
      rc = lsmInfoLogStructure(pDb, pzVal);
      break;
    }

    case LSM_INFO_FREELIST: {
      char **pzVal = va_arg(ap, char **);
      rc = infoFreelist(pDb, pzVal);
      break;
    }

    case LSM_INFO_CHECKPOINT_SIZE: {
      int *pnByte = va_arg(ap, int *);
      rc = lsmCheckpointSize(pDb, pnByte);
      break;

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

  off_t iSz;
  int prc;
  PosixFile *p = (PosixFile *)pFile;
  struct stat buf;

  if( p->pMap ){
    munmap(p->pMap, p->nMap);
    *ppOut = p->pMap = 0;
    *pnOut = p->nMap = 0;
  }

  memset(&buf, 0, sizeof(buf));
  prc = fstat(p->fd, &buf);
  if( prc!=0 ) return LSM_IOERR_BKPT;
  iSz = buf.st_size;
  if( iSz<iMin ){

  append script $data3
  append script $data4

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

do_write_test x.png 100 50000 0 20 {
  lsm-mt "threads=2 multi_proc=0"
  leveldb leveldb
}


  #lsm "mmap=1 multi_proc=0 page_size=4096 block_size=2097152 autocheckpoint=4194000"
  #lsm-mt    "mmap=1 multi_proc=0 threads=2 autowork=0 autocheckpoint=4196000"

# lsm     "safety=1 multi_proc=0"