SQLite4  Check-in [93d9ff7c12]

}

/*
** A log message callback registered with lsm connections. Prints all 
** messages to stderr.
*/
static void xLog(void *pCtx, int rc, const char *z){

  unused_parameter(rc);
  /* fprintf(stderr, "lsm: rc=%d \"%s\"\n", rc, z); */
  if( pCtx ) fprintf(stderr, "%s: ", (char *)pCtx);
  fprintf(stderr, "%s\n", z);
  fflush(stderr);

}

static void xWorkHook(lsm_db *db, void *pArg){
  LsmDb *p = (LsmDb *)pArg;

  }
  return 0;
}

void tdb_lsm_enable_log(TestDb *pDb, int bEnable){
  lsm_db *db = tdb_lsm(pDb);
  if( db ){
    LsmDb *p = (LsmDb *)pDb;
    int i;
    lsm_config_log(db, (bEnable ? xLog : 0), (void *)"client");
  }
}

void tdb_lsm_application_crash(TestDb *pDb){
  if( tdb_lsm(pDb) ){
    LsmDb *p = (LsmDb *)pDb;
    p->bCrashed = 1;

}

int tdb_lsm_open(const char *zCfg, const char *zDb, int bClear, TestDb **ppDb){
  return testLsmOpen(zCfg, zDb, bClear, ppDb);
}

#ifdef LSM_MUTEX_PTHREADS

/*
** Signal worker thread iWorker that there may be work to do.
*/
static void mt_signal_worker(LsmDb *pDb, int iWorker){
  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);
}

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 nWrite = 0;
    int rc;

    /* Do some work. If an error occurs, exit. */
    pthread_mutex_unlock(&p->worker_mutex);
    rc = lsm_work(pWorker, p->lsm_work_flags, p->lsm_work_npage, &nWrite);
/*    printf("# worked %d units\n", nWrite); */
    pthread_mutex_lock(&p->worker_mutex);
    if( rc!=LSM_OK && rc!=LSM_BUSY ){
      p->worker_rc = rc;
      break;
    }

    if( nWrite && (p->lsm_work_flags & LSM_WORK_CHECKPOINT)==0 ){
      mt_signal_worker(p->pDb, 1);
    }

    /* If the call to lsm_work() indicates that there is nothing more
    ** to do at this point, wait on the condition variable. The thread will
    ** wake up when it is signaled either because the client thread has
    ** flushed an in-memory tree into the db file or when the connection
    ** is being closed.  */
    if( nWrite==0 ){
      if( p->pWorker && p->bDoWork==0 ){
        pthread_cond_wait(&p->worker_cond, &p->worker_mutex);
      }
      p->bDoWork = 0;
    }
  }
  pthread_mutex_unlock(&p->worker_mutex);
  /* printf("# worker EXIT\n"); */
  
  return 0;
}














static void mt_stop_worker(LsmDb *pDb, int iWorker){
  LsmWorker *p = &pDb->aWorker[iWorker];
  if( p->pWorker ){
    void *pDummy;
    lsm_db *pWorker;

static void mt_client_work_hook(lsm_db *db, void *pArg){
  LsmDb *pDb = (LsmDb *)pArg;     /* LsmDb database handle */
  int i;                          /* Iterator variable */

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

  /* printf("# signalling worker threads\n"); */
  /* Signal each worker thread */
  for(i=0; i<pDb->nWorker; i++){
    mt_signal_worker(pDb, i);
  }
}

static void mt_worker_work_hook(lsm_db *db, void *pArg){

  LsmWorker *p;                   /* Object to initialize */

  assert( iWorker<pDb->nWorker );

  p = &pDb->aWorker[iWorker];
  p->lsm_work_flags = flags;
  p->lsm_work_npage = nPage;
  p->pDb = pDb;

  /* Open the worker connection */
  if( rc==0 ) rc = lsm_new(&pDb->env, &p->pWorker);
  if( rc==0 ) rc = lsm_open(p->pWorker, zFilename);
lsm_config_log(p->pWorker, xLog, (void *)"worker");

  /* Configure the work-hook */
  if( rc==0 ){
    lsm_config_work_hook(p->pWorker, mt_worker_work_hook, (void *)pDb);
  }

  /* Kick off the worker thread. */

static int testLsmStartWorkers(
  LsmDb *pDb, int nWorker, const char *zFilename, const char *zCfg
){
  int rc;
  int bAutowork = 0;
  assert( nWorker==1 || nWorker==2 );


  /* Configure a work-hook for the client connection. */


  lsm_config_work_hook(pDb->db, mt_client_work_hook, (void *)pDb);

  pDb->aWorker = (LsmWorker *)testMalloc(sizeof(LsmWorker) * nWorker);
  memset(pDb->aWorker, 0, sizeof(LsmWorker) * nWorker);
  pDb->nWorker = nWorker;

  if( nWorker==1 ){
    int flags = LSM_WORK_CHECKPOINT|LSM_WORK_FLUSH;


    rc = mt_start_worker(pDb, 0, zFilename, flags, 2048);


  }else{
    rc = mt_start_worker(pDb, 0, zFilename, LSM_WORK_FLUSH, 1024);
    if( rc==LSM_OK ){
      rc = mt_start_worker(pDb, 1, zFilename, LSM_WORK_CHECKPOINT, 0);
    }
  }

  return rc;
}


static int test_lsm_mt(

**     The third argument should be of type (char **). The location pointed
**     to is populated with a pointer to a nul-terminated string containing
**     the string representation of a Tcl data-structure. The returned 
**     string should be eventually freed by the caller using lsm_free().
**
**     The Tcl structure returned is a list of six integers that describe
**     the current structure of the log file.
**
**   LSM_INFO_FREELIST
**     The third argument should be of type (char **). The location pointed
**     to is populated with a pointer to a nul-terminated string containing
**     the string representation of a Tcl data-structure. The returned 
**     string should be eventually freed by the caller using lsm_free().
**
**     The Tcl structure returned is a list containing one element for each
**     free block in the database. The element itself consists of two 
**     integers - the block number and the id of the snapshot that freed it.
*/
#define LSM_INFO_NWRITE           1
#define LSM_INFO_NREAD            2
#define LSM_INFO_DB_STRUCTURE     3
#define LSM_INFO_LOG_STRUCTURE    4
#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


/* 
** Open and close transactions and nested transactions.
**
** lsm_begin():
**   Used to open transactions and sub-transactions. A successful call to 

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);
int lsmInfoFreelist(lsm_db *, char **pzVal);


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

int lsmDbDatabaseConnect(lsm_db*, const char *);

}

void lsmAppendSegmentList(LsmString *pStr, char *zPre, Segment *pSeg){
  lsmStringAppendf(pStr, "%s{%d %d %d %d}", zPre, 
        pSeg->iFirst, pSeg->iLast, pSeg->iRoot, pSeg->nSize
  );
}

static int infoGetWorker(lsm_db *pDb, Snapshot **pp, int *pbUnlock){
  int rc = LSM_OK;

  assert( *pbUnlock==0 );
  if( !pDb->pWorker ){
    rc = lsmBeginWork(pDb);
    if( rc!=LSM_OK ) return rc;
    *pbUnlock = 1;
  }
  *pp = pDb->pWorker;
  return rc;
}

static void infoFreeWorker(lsm_db *pDb, int bUnlock){
  if( bUnlock ){
    int rcdummy = LSM_BUSY;
    lsmFinishWork(pDb, 0, 0, &rcdummy);
  }
}

int lsmStructList(
  lsm_db *pDb,                    /* Database handle */
  char **pzOut                    /* OUT: Nul-terminated string (tcl list) */
){
  Level *pTopLevel = 0;           /* Top level of snapshot to report on */
  int rc = LSM_OK;
  Level *p;
  LsmString s;
  Snapshot *pWorker;              /* Worker snapshot */
  int bUnlock = 0;

  /* Obtain the worker snapshot */

  rc = infoGetWorker(pDb, &pWorker, &bUnlock);

  if( rc!=LSM_OK ) return rc;




  /* Format the contents of the snapshot as text */
  pTopLevel = lsmDbSnapshotLevel(pWorker);
  lsmStringInit(&s, pDb->pEnv);
  for(p=pTopLevel; rc==LSM_OK && p; p=p->pNext){
    int i;
    lsmStringAppendf(&s, "%s{", (s.n ? " " : ""));
    lsmAppendSegmentList(&s, "", &p->lhs);
    for(i=0; rc==LSM_OK && i<p->nRight; i++){
      lsmAppendSegmentList(&s, " ", &p->aRhs[i]);
    }
    lsmStringAppend(&s, "}", 1);
  }
  rc = s.n>=0 ? LSM_OK : LSM_NOMEM;

  /* Release the snapshot and return */
  infoFreeWorker(pDb, bUnlock);
  *pzOut = s.z;
  return rc;
}

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

  /* Obtain the worker snapshot */
  rc = infoGetWorker(pDb, &pWorker, &bUnlock);
  if( rc!=LSM_OK ) return rc;

  lsmStringInit(&s, pDb->pEnv);
  lsmStringAppendf(&s, "%d+%d",pWorker->freelist.nEntry,pWorker->nFreelistOvfl);
  for(i=0; i<pWorker->freelist.nEntry; i++){
    FreelistEntry *p = &pWorker->freelist.aEntry[i];
    lsmStringAppendf(&s, " {%d %d}", p->iBlk, (int)p->iId);
  }
  rc = s.n>=0 ? LSM_OK : LSM_NOMEM;

  /* Release the snapshot and return */
  infoFreeWorker(pDb, bUnlock);
  *pzOut = s.z;
  return rc;
}

int lsm_info(lsm_db *pDb, int eParam, ...){
  int rc = LSM_OK;
  va_list ap;

    }

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

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

    default:
      rc = LSM_MISUSE;
      break;
  }

  va_end(ap);

  ** db and checkpoint the latest snapshot.
  **
  ** Ignore any LSM_BUSY errors that occur during these operations. If
  ** LSM_BUSY does occur, it means some other connection is already working
  ** on flushing the in-memory tree or checkpointing the database. 
  */
  assert( rc!=LSM_BUSY);
  if( rc==LSM_OK ){
    if( nFlush>1 || (nFlush && pDb->bAutowork) ){
      rc = lsmFlushToDisk(pDb);
      if( rc==LSM_OK && pDb->bAutowork ){
        rc = lsmCheckpointWrite(pDb);
      }
    }else if( nFlush && pDb->xWork ){
      pDb->xWork(pDb, pDb->pWorkCtx);
    }
  }
  if( rc==LSM_BUSY ) rc = LSM_OK;

  return rc;
}

    /* The "is in use" bit */
    rc = lsmLsmInUse(pDb, iFree, &bInUse);

    /* The "has been checkpointed" bit */
    if( rc==LSM_OK && bInUse==0 ){
      i64 iId = 0;
      rc = lsmCheckpointSynced(pDb, &iId, 0);
      if( rc!=LSM_OK || iId<iFree ) bInUse = 2;
    }

    if( rc==LSM_OK && bInUse==0 ){
      iRet = pFree->aEntry[0].iBlk;
      flRemoveEntry0(pFree);
      assert( iRet!=0 );
    }
#if 0
    lsmLogMessage(
        pDb, 0, "%d reusing block %d%s", (iRet==0 ? "not " : "")
        pFree->aEntry[0].iBlk, 
        bInUse==0 ? "" : bInUse==1 ? " (client)" : " (unsynced)"
    );
#endif
  }

  /* If no block was allocated from the free-list, allocate one at the
  ** end of the file. */
  if( rc==LSM_OK && iRet==0 ){
    iRet = ++pDb->pWorker->nBlock;
  }

        lsmFsMetaPageRelease(pPg);
      }
      bDone = (iDisk>=iCkpt);
    }

    if( rc==LSM_OK && bDone==0 ){
      int iMeta = (pShm->iMetaPage % 2) + 1;
#if 0
  lsmLogMessage(pDb, 0, "starting checkpoint");
#endif
      if( pDb->eSafety!=LSM_SAFETY_OFF ){
        rc = lsmFsSyncDb(pDb->pFS);
      }
      if( rc==LSM_OK ) rc = lsmCheckpointStore(pDb, iMeta);
      if( rc==LSM_OK && pDb->eSafety!=LSM_SAFETY_OFF){
        rc = lsmFsSyncDb(pDb->pFS);
      }
      if( rc==LSM_OK ) pShm->iMetaPage = iMeta;
#if 0
  lsmLogMessage(pDb, 0, "finish checkpoint %d", 
      (int)lsmCheckpointId(pDb->aSnapshot, 0)
  );
#endif
    }
  }

  lsmShmLock(pDb, LSM_LOCK_CHECKPOINTER, LSM_LOCK_UNLOCK, 0);
  return rc;
}

    ** that the shared-memory still contains the same values. If so, proceed.
    ** Otherwise, relinquish the read-lock and retry the whole procedure
    ** (starting with loading the in-memory tree header).  */
    if( rc==LSM_OK ){
      ShmHeader *pShm = pDb->pShmhdr;
      u32 iShmMax = pDb->treehdr.iUsedShmid;
      u32 iShmMin = pDb->treehdr.iNextShmid+1-pDb->treehdr.nChunk;
      rc = lsmReadlock(
          pDb, lsmCheckpointId(pDb->aSnapshot, 0), iShmMin, iShmMax
      );
      if( rc==LSM_OK ){
        if( lsmTreeLoadHeaderOk(pDb, iTreehdr)
         && lsmCheckpointLoadOk(pDb, iSnap)
        ){
          /* Read lock has been successfully obtained. Deserialize the 
          ** checkpoint just loaded. TODO: This will be removed after 
          ** lsm_sorted.c is changed to work directly from the serialized

  ShmHeader *pShm = db->pShmhdr;
  int i;
  int rc = LSM_OK;

  for(i=0; rc==LSM_OK && i<LSM_LOCK_NREADER; i++){
    ShmReader *p = &pShm->aReader[i];
    if( p->iLsmId ){
      if( (iLsmId!=0 && p->iLsmId!=0 && iLsmId>=p->iLsmId) 
       || (iLsmId==0 && shm_sequence_ge(p->iTreeId, iShmid))
      ){
        rc = lsmShmLock(db, LSM_LOCK_READER(i), LSM_LOCK_EXCL, 0);
        if( rc==LSM_OK ){
          p->iLsmId = 0;
          lsmShmLock(db, LSM_LOCK_READER(i), LSM_LOCK_UNLOCK, 0);
        }

  rc = sortedNewToplevel(pDb, 1, pnOvfl);
  assert( rc!=LSM_OK || lsmFsIntegrityCheck(pDb) );

#if 0
  lsmSortedDumpStructure(pDb, pDb->pWorker, 1, 0, "tree flush");
#endif
#if 0
  lsmLogMessage(pDb, rc, "flushed tree to disk");
#endif
  return rc;
}

/*
** The nMerge levels in the LSM beginning with pLevel consist of a
** left-hand-side segment only. Replace these levels with a single new
** level consisting of a new empty segment on the left-hand-side and the

int lsmSortedAutoWork(
  lsm_db *pDb,                    /* Database handle */
  int nUnit                       /* Pages of data written to in-memory tree */
){
  int rc;                         /* Return code */
  int nRemaining;                 /* Units of work to do before returning */
  int nDepth;                     /* Current height of tree (longest path) */
  int nWrite;                     /* Pages written */
  Level *pLevel;                  /* Used to iterate through levels */

  assert( lsmFsIntegrityCheck(pDb) );
  assert( pDb->pWorker );

  /* Determine how many units of work to do before returning. One unit of
  ** work is achieved by writing one page (~4KB) of merged data.  */
  nRemaining = nDepth = 0;
  for(pLevel=lsmDbSnapshotLevel(pDb->pWorker); pLevel; pLevel=pLevel->pNext){
    /* nDepth += LSM_MAX(1, pLevel->nRight); */
    nDepth += 1;
  }
  nRemaining = nUnit * nDepth;

  rc = sortedWork(pDb, nRemaining, 0, &nWrite);
#if 0
  lsmLogMessage(pDb, 0, "auto-work: %d pages", nWrite);
#endif
  return rc;
}

/*
** Perform work to merge database segments together.
*/
int lsm_work(lsm_db *pDb, int flags, int nPage, int *pnWrite){

    bFinishWork = 1;
  }

  /* If the FLUSH flag is set, try to flush the contents of the in-memory
  ** tree to disk.  */
  if( rc==LSM_OK && ((flags & LSM_WORK_FLUSH)) ){
    rc = lsmTreeLoadHeader(pDb, 0);
    if( rc==LSM_OK 
     && pDb->treehdr.iOldShmid 
     && pDb->treehdr.iOldLog!=pDb->pWorker->iLogOff 
    ){
      rc = lsmSortedFlushTree(pDb, &nOvfl);
      bFlush = 1;
    }
  }

  /* If nPage is greater than zero, do some merging. */
  if( rc==LSM_OK && nPage>0 ){
    int bOptimize = ((flags & LSM_WORK_OPTIMIZE) ? 1 : 0);
    rc = sortedWork(pDb, nPage, bOptimize, &nWrite);
    if( rc==LSM_OK && nWrite ){
#if 0
  {
    char *z = 0;
    lsmInfoFreelist(pDb, &z);
    lsmLogMessage(pDb, 0, "work: %d pages", nWrite);
    lsmLogMessage(pDb, 0, "freelist: %s", z);
    lsm_free(lsm_get_env(pDb), z);
  }
#endif
      rc = lsmSortedFlushDb(pDb);
      if( rc==LSM_OK && lsmCheckpointOverflowRequired(pDb) ){
        nOvfl = -1;
        rc = sortedNewToplevel(pDb, 0, &nOvfl);
      }
    }
  }

#!/bin/sh
# \
exec tclsh "$0" "$@"


proc exec_lsmtest_speed {nSec spec} {
  set fd [open [list |lsmtest speed2 {*}$spec]]
  set res [list]

  puts "lsmtest speed2 $spec"

  set initial [clock seconds]

  while {![eof $fd] && ($nSec==0 || ([clock second]-$initial)<$nSec)} { 
    set line [gets $fd]
    puts $line
    if {[string range $line 0 0]=="#"} continue
    if {[llength $line]==0} continue

    incr nMs $msInsert
    append ret "$nIns [expr $nIns*1000.0/$nMs]\n"
  }
  append ret "end\n"
  set ret
}

proc make_dataset {res iRes nWrite nShift nOp} {
  set ret ""
  foreach row $res {
    set i [lindex $row 0]
    set j [lindex $row [expr $iRes+1]]
    set x [expr $i*$nWrite + $nShift]
    append ret "$x [expr int($nOp * 1000.0 / $j)]\n"
  }
  append ret "end\n"
  set ret
}

proc do_write_test {zPng nSec nWrite nFetch nRepeat lSys} {

  set data3 ""

  foreach {name sys} $lSys res $lRes col $cols {
    foreach {c1 c2} $col {}

    if {$plot1 != ""} { set plot1 ", $plot1" }
    set plot1 "\"-\" ti \"$name writes/sec\" with boxes fs solid lc rgb \"$c1\"$plot1"
    set data1 "[make_dataset $res 0 $nWrite $nShift $nWrite] $data1"

    set plot3 ",\"-\" ti \"$name cumulative writes/sec\" with lines lc rgb \"$c2\" lw 2 $plot3"
    set data3 "[make_totalset $res $nWrite] $data3"

    if {$nFetch>0} {
      set new ", \"-\" ti \"$name fetches/sec\" axis x1y2 with points lw 3 lc rgb \"$c2\""
      set plot2 "$new $plot2"
      set data2 "[make_dataset $res 1 $nWrite $nWrite $nFetch] $data2"
    }

    incr nShift [expr $nWrite/4]
  }
  append script "plot $plot1 $plot2 $plot3\n"
  append script $data1
  append script $data2
  append script $data3

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

do_write_test x.png 60 10000 20000 1000 {
  LSM   "mmap=1 multi_proc=0 safety=1 threads=3 autowork=0 worker_nmerge=2" 
}