SQLite4  Check-in [3d1dacff87]

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


      rc = lsm_checkpoint(pWorker, 0);

    }else{
      int nWrite = 0;             /* Pages written by lsm_work() call */
      int nAuto = -1;             /* Configured AUTOCHECKPOINT value */
      int nLimit = -1;            /* Configured WRITE_BUFFER value */

      lsm_config(pWorker, LSM_CONFIG_WRITE_BUFFER, &nLimit);
      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);

}

/*
** Close the log file. Then delete it from the file-system. This function
** is called during database shutdown only.
*/
int lsmFsCloseAndDeleteLog(FileSystem *pFS){
  char *zDel;

  if( pFS->fdLog ){
    lsmEnvClose(pFS->pEnv, pFS->fdLog );
    pFS->fdLog = 0;
  }

  zDel = lsmMallocPrintf(pFS->pEnv, "%s-log", pFS->zDb);
  if( zDel ){

    lsmEnvUnlink(pFS->pEnv, zDel);
    lsmFree(pFS->pEnv, zDel);


  }
  return LSM_OK;
}

/*
** Given that there are currently nHash slots in the hash table, return 
** the hash key for file iFile, page iPg.

    ** connection to the database. In this case flush the contents of the
    ** in-memory tree to disk and write a checkpoint.  */
    rc = lsmShmLock(pDb, LSM_LOCK_DMS2, LSM_LOCK_EXCL, 0);
    if( rc==LSM_OK ){
      /* Flush the in-memory tree, if required. If there is data to flush,
      ** this will create a new client snapshot in Database.pClient. The
      ** checkpoint (serialization) of this snapshot may be written to disk
      ** by the following block.  
      **
      ** There is no need to mess around with WRITER locks or anything at
      ** this point. The lock on DMS2 guarantees that pDb has exclusive
      ** access to the db at this point.
      */
      rc = lsmTreeLoadHeader(pDb, 0);
      if( rc==LSM_OK && (lsmTreeHasOld(pDb) || lsmTreeSize(pDb)>0) ){




        rc = lsmFlushTreeToDisk(pDb);


      }

      /* Write a checkpoint to disk. */
      if( rc==LSM_OK ){
        rc = lsmCheckpointWrite(pDb, 0);
      }

      /* If the checkpoint was written successfully, delete the log file */
      if( rc==LSM_OK ){


        Database *p = pDb->pDatabase;
        lsmFsCloseAndDeleteLog(pDb->pFS);
        if( p->pFile ) lsmEnvShmUnmap(pDb->pEnv, p->pFile, 1);
      }
    }
  }

      if( rc==LSM_OK ) pCsr->nSegCsr++;
    }
  }

  return rc;
}

#define TREE_NONE 0
#define TREE_OLD  1
#define TREE_BOTH 2

/*
** Parameter eTree must be set to TREE_NONE, OLD or BOTH.



*/
static int multiCursorNew(
  lsm_db *pDb,                    /* Database handle */
  Snapshot *pSnap,                /* Snapshot to use for this cursor */
  int eTree,                      /* One of the TREE_XXX values above */
  int bUserOnly,                  /* If true, ignore all system data */
  MultiCursor **ppCsr             /* OUT: Allocated cursor */
){
  int rc = LSM_OK;                /* Return Code */
  MultiCursor *pCsr = *ppCsr;     /* Allocated multi-cursor */

  assert( eTree==TREE_NONE || eTree==TREE_OLD || eTree==TREE_BOTH );

  if( pCsr==0 ){
    pCsr = (MultiCursor *)lsmMallocZeroRc(pDb->pEnv, sizeof(MultiCursor), &rc);
    if( pCsr ){
      pCsr->pNext = pDb->pCsr;
      pDb->pCsr = pCsr;
      if( bUserOnly ) pCsr->flags |= CURSOR_IGNORE_SYSTEM;
      pCsr->pDb = pDb;
      pCsr->pSnap = pSnap;
      pCsr->xCmp = pDb->xCmp;
    }
  }

  /* Add a tree cursor on the 'old' tree, if required. */
  if( rc==LSM_OK 
   && eTree!=TREE_NONE 

   && lsmTreeHasOld(pDb)

   && pDb->treehdr.iOldLog!=pSnap->iLogOff
  ){
    rc = lsmTreeCursorNew(pDb, 1, &pCsr->apTreeCsr[1]);
  }




  /* Add a tree cursor on the 'current' tree, if required. */
  if( rc==LSM_OK && eTree==TREE_BOTH ){
    rc = lsmTreeCursorNew(pDb, 0, &pCsr->apTreeCsr[0]);
  }

  if( rc!=LSM_OK ){
    lsmMCursorClose(pCsr);
    pCsr = 0;
  }
  *ppCsr = pCsr;
  return rc;
}

  MultiCursor *pCsr = *ppCsr;     /* Allocated multi-cursor */
  Level *p;                       /* Level iterator */
  Snapshot *pSnap;                /* Snapshot to use for cursor */

  pSnap = (bSystem ? pDb->pWorker : pDb->pClient);
  assert( pSnap );

  rc = multiCursorNew(pDb, pSnap, 
      (bSystem ? TREE_NONE : TREE_BOTH), !bSystem, &pCsr
  );

  multiCursorIgnoreDelete(pCsr);
  for(p=lsmDbSnapshotLevel(pSnap); p && rc==LSM_OK; p=p->pNext){
    rc = multiCursorAddLevel(pCsr, p, MULTICURSOR_ADDLEVEL_ALL);
  }

  if( rc!=LSM_OK ){
    lsmMCursorClose(pCsr);

  if( pDb->xWork ){
    pDb->xWork(pDb, pDb->pWorkCtx);
  }
}

static int sortedNewToplevel(
  lsm_db *pDb,                    /* Connection handle */
  int eTree,                      /* One of the TREE_XXX constants */
  int *pnOvfl,                    /* OUT: Number of free-list entries stored */
  int *pnWrite                    /* OUT: Number of database pages written */
){
  int rc = LSM_OK;                /* Return Code */
  MultiCursor *pCsr = 0;
  Level *pNext = 0;               /* The current top level */
  Level *pNew;                    /* The new level itself */

  pNext = lsmDbSnapshotLevel(pDb->pWorker);
  pNew = (Level *)lsmMallocZeroRc(pDb->pEnv, sizeof(Level), &rc);

  /* Create a cursor to gather the data required by the new segment. The new
  ** segment contains everything in the tree and pointers to the next segment
  ** in the database (if any).  */
  if( rc==LSM_OK ){
    rc = multiCursorNew(pDb, pDb->pWorker, eTree, 0, &pCsr);
    if( rc==LSM_OK ){
      pNew->pNext = pNext;
      lsmDbSnapshotSetLevel(pDb->pWorker, pNew);
    }
    if( rc==LSM_OK ){
      if( pNext ){
        assert( pNext->pMerge==0 || pNext->nRight>0 );

  **   1. Records from LHS of each of the nMerge levels being merged.
  **   2. Separators from either the last level being merged, or the
  **      separators attached to the LHS of the following level, or neither.
  **
  ** If the new level is the lowest (oldest) in the db, discard any
  ** delete keys. Key annihilation.
  */
  rc = multiCursorNew(pDb, pDb->pWorker, TREE_NONE, 0, &pCsr);
  if( rc==LSM_OK ){
    rc = multiCursorAddLevel(pCsr, pLevel, MULTICURSOR_ADDLEVEL_RHS);
  }
  if( rc==LSM_OK && pLevel->pNext ){
    if( pMerge->nInput > pLevel->nRight ){
      Level *pNext = pLevel->pNext;
      rc = multiCursorAddLevel(pCsr, pNext, MULTICURSOR_ADDLEVEL_LHS_SEP);

        nRem -= nPg;
        assert( rc!=LSM_OK || nRem<=0 || !sortedDbIsFull(pDb) );
        bToplevel = 1;
      }

      if( rc==LSM_OK && nRem>0 ){
        int nPg = 0;
        rc = sortedNewToplevel(pDb, TREE_OLD, &nOvfl, &nPg);
        nRem -= nPg;
        if( rc==LSM_OK && pDb->nTransOpen>0 ){
          lsmTreeDiscardOld(pDb);
        }
        bFlush = 1;
        bToplevel = 0;
      }

  if( rc==LSM_OK && bToplevel && lsmCheckpointOverflowRequired(pDb) ){
    while( rc==LSM_OK && sortedDbIsFull(pDb) ){
      int nPg = 0;
      rc = sortedWork(pDb, 16, 0, 1, &nPg);
    }
    if( rc==LSM_OK && lsmCheckpointOverflowRequired(pDb) ){
      rc = sortedNewToplevel(pDb, TREE_NONE, &nOvfl, 0);
    }
  }

  if( rc==LSM_OK && (nRem!=nMax) ){
    rc = lsmSortedFlushDb(pDb);
    lsmFinishWork(pDb, bFlush, nOvfl, &rc);
  }else{

#if 0
  lsmLogMessage(pDb, 0, "auto-work: %d pages", nWrite);
#endif

  return rc;
}

/*
** This function is only called during system shutdown. The contents of
** any in-memory trees present (old or current) are written out to disk.
*/
int lsmFlushTreeToDisk(lsm_db *pDb){
  int rc;
  int nOvfl = 0;

  rc = lsmBeginWork(pDb);
  while( rc==LSM_OK && sortedDbIsFull(pDb) ){
    rc = sortedWork(pDb, 256, 0, 1, 0);
  }

  if( rc==LSM_OK ){
    rc = sortedNewToplevel(pDb, TREE_BOTH, &nOvfl, 0);

  }
  lsmFinishWork(pDb, 1, nOvfl, &rc);
  return rc;
}

/*
** Return a string representation of the segment passed as the only argument.
** Space for the returned string is allocated using lsmMalloc(), and should
** be freed by the caller using lsmFree().

  append script $data2
  append script $data3

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

do_write_test x.png 120 50000 0 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"