SQLite4  Check-in [e16b04ca69]

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 nLimit = -1;
    int nCkpt = -1;

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





    if( p->bCkpt ){
      lsm_ckpt_size(pWorker, &nCkpt);
      if( nCkpt>(nLimit*2) ){
        rc = lsm_checkpoint(pWorker, 0);



      }
    }else{
      int nWrite = 0;
      int nAuto = -1;
      lsm_config(pWorker, LSM_CONFIG_AUTOCHECKPOINT, &nAuto);
      do {
        int nSleep = 0;
        lsm_ckpt_size(pWorker, &nCkpt);
        while( nAuto==0 && nCkpt>(nLimit*4) ){

          usleep(1000);
          mt_signal_worker(p->pDb, 1);
          nSleep++;
          lsm_ckpt_size(pWorker, &nCkpt);
        }
#if 0
          if( nSleep ) printf("nLimit=%d nSleep=%d (worker)\n", nLimit, nSleep);
#endif



        rc = lsm_work(pWorker, p->lsm_work_flags, p->lsm_work_npage, &nWrite);
        if( nAuto==0 && nWrite && rc==LSM_OK ) mt_signal_worker(p->pDb, 1);
      }while( nWrite );



    }
    pthread_mutex_lock(&p->worker_mutex);

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





    /* 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( p->pWorker && p->bDoWork==0 ){
      pthread_cond_wait(&p->worker_cond, &p->worker_mutex);
    }
    p->bDoWork = 0;

  }
  pthread_mutex_unlock(&p->worker_mutex);
  
  return 0;
}

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


  /* Signal the lsm_work() thread */

  mt_signal_worker(pDb, 0);

}

static void mt_worker_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);

  /* 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 = 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. */

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

  if( nWorker==1 ){
    int flags = LSM_WORK_FLUSH;
    rc = mt_start_worker(pDb, 0, zFilename, zCfg, flags, 256, 0);
  }else{
    int flags = LSM_WORK_FLUSH;
    rc = mt_start_worker(pDb, 0, zFilename, zCfg, flags, 256, 0);
    if( rc==LSM_OK ){
      rc = mt_start_worker(pDb, 1, zFilename, zCfg, 0, 0, 1);
    }
  }

  return rc;
}

    }

    if( rc==LSM_OK && bInUse==0 ){
      iRet = pFree->aEntry[0].iBlk;
      flRemoveEntry0(pFree);
      assert( iRet!=0 );
    }
#ifdef LSM_LOG_BLOCKS
    lsmLogMessage(
        pDb, 0, "%s 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;
#ifdef LSM_LOG_BLOCKS
    lsmLogMessage(pDb, 0, "extending file to %d blocks", iRet);
#endif
  }

  *piBlk = iRet;
  return LSM_OK;
}

/*

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


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

  puts -nonewline "./lsmtest speed2"
  foreach s $spec {
    if {[llength $s]==1} {
      puts -nonewline " $s"
    } else {
      puts -nonewline " \"$s\""
    }
  }
  puts ""

  set initial [clock seconds]

  while {![eof $fd] && ($nSec==0 || ([clock second]-$initial)<$nSec)} { 
    set line [gets $fd]
    set sz 0
    catch { set sz [file size testdb.lsm] }
    puts "$line ([expr $sz/1024])"
    if {[string range $line 0 0]=="#"} continue
    if {[llength $line]==0} continue
    lappend res $line
  }
  catch { close $fd }
  set res
}

  append script $data2
  append script $data3

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

do_write_test x.png 120 50000 50000 30 {
  lsm-mt    "mmap=1 multi_proc=0 threads=3 autowork=0 autocheckpoint=0"
  leveldb   leveldb
}
#  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"
# LevelDB leveldb
# lsm-st     "mmap=1 multi_proc=0 safety=1 threads=1 autowork=1"
# LevelDB leveldb
# SQLite sqlite3