SQLite4  Check-in [d6bd08ca0e]

  if( pNew==0 ){
    rc = SQLITE4_NOMEM;
  }else{
    struct Config {
      const char *zParam;
      int eParam;
    } aConfig[] = {
      { "lsm_block_size", LSM_CONFIG_BLOCK_SIZE },
      { "lsm_multiple_processes", LSM_CONFIG_MULTIPLE_PROCESSES }
    };

    memset(pNew, 0, sizeof(KVLsm));
    pNew->base.pStoreVfunc = &kvlsmMethods;
    pNew->base.pEnv = pEnv;
    rc = lsm_new(0, &pNew->pDb);
    if( rc==SQLITE4_OK ){

/*
** Database structure. There is one such structure for each distinct 
** database accessed by this process. They are stored in the singly linked 
** list starting at global variable gShared.pDatabase. Database objects are 
** reference counted. Once the number of connections to the associated
** database drops to zero, they are removed from the linked list and deleted.
**
** pFile:
**   In multi-process mode, this file descriptor is used to obtain locks 
**   and to access shared-memory. In single process mode, its only job is
**   to hold the exclusive lock on the file.
**   
*/
struct Database {
  /* Protected by the global mutex (enterGlobalMutex/leaveGlobalMutex): */
  char *zName;                    /* Canonical path to database file */
  int nName;                      /* strlen(zName) */
  int nDbRef;                     /* Number of associated lsm_db handles */
  Database *pDbNext;              /* Next Database structure in global list */

  /* Protected by the local mutex (pClientMutex) */
  int bMultiProc;                 /* True if running in multi-process mode */
  lsm_file *pFile;                /* Used for locks/shm in multi-proc mode */
  LsmFile *pLsmFile;              /* List of deferred closes */
  lsm_mutex *pClientMutex;        /* Protects the apShmChunk[] and pConn */
  int nShmChunk;                  /* Number of entries in apShmChunk[] array */
  void **apShmChunk;              /* Array of "shared" memory regions */
  lsm_db *pConn;                  /* List of connections to this db. */
};

      /* If the checkpoint was written successfully, delete the log file
      ** and, if possible, truncate the database file.  */
      if( rc==LSM_OK ){
        Database *p = pDb->pDatabase;
        dbTruncateFile(pDb);
        lsmFsCloseAndDeleteLog(pDb->pFS);
        if( p->pFile && p->bMultiProc ) lsmEnvShmUnmap(pDb->pEnv, p->pFile, 1);
      }
    }
  }

  lsmShmLock(pDb, LSM_LOCK_DMS2, LSM_LOCK_UNLOCK, 0);
  lsmShmLock(pDb, LSM_LOCK_DMS1, LSM_LOCK_UNLOCK, 0);
  pDb->pShmhdr = 0;

    if( rc==LSM_OK ){
      rc = lsmLogRecover(pDb);
    }
  }else if( rc==LSM_BUSY ){
    rc = LSM_OK;
  }

  /* Take a shared lock on DMS2. In multi-process mode this lock "cannot" 
  ** fail, as connections may only hold an exclusive lock on DMS2 if they 
  ** first hold an exclusive lock on DMS1. And this connection is currently 
  ** holding the exclusive lock on DSM1. 
  **
  ** However, if some other connection has the database open in single-process
  ** mode, this operation will fail. In this case, return the error to the
  ** caller - the attempt to connect to the db has failed.
  */
  if( rc==LSM_OK ){
    rc = lsmShmLock(pDb, LSM_LOCK_DMS2, LSM_LOCK_SHARED, 0);

  }

  /* If anything went wrong, unlock DMS2. Unlock DMS1 in any case. */
  if( rc!=LSM_OK ){
    lsmShmLock(pDb, LSM_LOCK_DMS2, LSM_LOCK_UNLOCK, 0);
    pDb->pShmhdr = 0;
  }

  int nName = lsmStrlen(zName);

  assert( pDb->pDatabase==0 );
  rc = enterGlobalMutex(pEnv);
  if( rc==LSM_OK ){

    /* Search the global list for an existing object. TODO: Need something
    ** better than the memcmp() below to figure out if a given Database
    ** object represents the requested file.  */
    for(p=gShared.pDatabase; p; p=p->pDbNext){
      if( nName==p->nName && 0==memcmp(zName, p->zName, nName) ) break;
    }

    /* If no suitable Database object was found, allocate a new one. */
    if( p==0 ){
      p = (Database *)lsmMallocZeroRc(pEnv, sizeof(Database)+nName+1, &rc);

      /* If the allocation was successful, fill in other fields and
      ** allocate the client mutex. */ 
      if( rc==LSM_OK ){
        p->bMultiProc = pDb->bMultiProc;
        p->zName = (char *)&p[1];
        p->nName = nName;
        memcpy((void *)p->zName, zName, nName+1);
        rc = lsmMutexNew(pEnv, &p->pClientMutex);
      }

      /* If nothing has gone wrong so far, open the shared fd. And if that
      ** succeeds and this connection requested single-process mode, 
      ** attempt to take the exclusive lock on DMS2.  */
      if( rc==LSM_OK ){
        rc = lsmEnvOpen(pDb->pEnv, p->zName, &p->pFile);
      }
      if( rc==LSM_OK && p->bMultiProc==0 ){
        rc = lsmEnvLock(pDb->pEnv, p->pFile, LSM_LOCK_DMS2, LSM_LOCK_EXCL);
      }

      if( rc==LSM_OK ){
        p->pDbNext = gShared.pDatabase;
        gShared.pDatabase = p;
      }else{
        freeDatabase(pEnv, p);
        p = 0;

    if( pDb->pShmhdr ){
      doDbDisconnect(pDb);
    }

    lsmMutexEnter(pDb->pEnv, p->pClientMutex);
    for(ppDb=&p->pConn; *ppDb!=pDb; ppDb=&((*ppDb)->pNext));
    *ppDb = pDb->pNext;

    dbDeferClose(pDb);

    lsmMutexLeave(pDb->pEnv, p->pClientMutex);

    enterGlobalMutex(pDb->pEnv);
    p->nDbRef--;
    if( p->nDbRef==0 ){
      LsmFile *pIter;
      LsmFile *pNext;
      Database **pp;

      /* Remove the Database structure from the linked list. */
      for(pp=&gShared.pDatabase; *pp!=p; pp=&((*pp)->pDbNext));
      *pp = p->pDbNext;

      /* If they were allocated from the heap, free the shared memory chunks */
      if( p->bMultiProc==0 ){
        int i;
        for(i=0; i<p->nShmChunk; i++){
          lsmFree(pDb->pEnv, p->apShmChunk[i]);
        }
      }

      /* Close any outstanding file descriptors */

      for(pIter=p->pLsmFile; pIter; pIter=pNext){
        pNext = pIter->pNext;
        lsmEnvClose(pDb->pEnv, pIter->pFile);
        lsmFree(pDb->pEnv, pIter);

      }
      freeDatabase(pDb->pEnv, p);
    }
    leaveGlobalMutex(pDb->pEnv);
  }
}

/*
** This function may only be called after a successful call to
** lsmDbDatabaseConnect(). It returns true if the connection is in
** multi-process mode, or false otherwise.
*/
int lsmDbMultiProc(lsm_db *pDb){
  return pDb->pDatabase && pDb->pDatabase->bMultiProc;
}


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

      }
      p->apShmChunk = apShm;
    }

    for(i=db->nShm; rc==LSM_OK && i<nChunk; i++){
      if( i>=p->nShmChunk ){
        void *pChunk = 0;
        if( p->bMultiProc==0 ){
          /* Single process mode */
          pChunk = lsmMallocZeroRc(pEnv, LSM_SHM_CHUNK_SIZE, &rc);
        }else{
          /* Multi-process mode */
          rc = lsmEnvShmMap(pEnv, p->pFile, i, LSM_SHM_CHUNK_SIZE, &pChunk);
        }
        if( rc==LSM_OK ){

    /* Release the client mutex */
    lsmMutexLeave(pEnv, p->pClientMutex);
  }

  return rc;
}

static int lockSharedFile(lsm_env *pEnv, Database *p, int iLock, int eOp){
  int rc = LSM_OK;
  if( p->bMultiProc ){
    rc = lsmEnvLock(pEnv, p->pFile, iLock, eOp);
  }
  return rc;
}

/*
** Attempt to obtain the lock identified by the iLock and bExcl parameters.
** If successful, return LSM_OK. If the lock cannot be obtained because 
** there exists some other conflicting lock, return LSM_BUSY. If some other
** error occurs, return an LSM error code.
**

    assert( nExcl==0 || nExcl==1 );
    assert( nExcl==0 || nShared==0 );
    assert( nExcl==0 || (db->mLock & (me|ms))==0 );

    switch( eOp ){
      case LSM_LOCK_UNLOCK:
        if( nShared==0 ){
          lockSharedFile(db->pEnv, p, iLock, LSM_LOCK_UNLOCK);
        }
        db->mLock &= ~(me|ms);
        break;

      case LSM_LOCK_SHARED:
        if( nExcl ){
          rc = LSM_BUSY;
        }else{
          if( nShared==0 ){
            rc = lockSharedFile(db->pEnv, p, iLock, LSM_LOCK_SHARED);
          }
          db->mLock |= ms;
          db->mLock &= ~me;
        }
        break;

      default:
        assert( eOp==LSM_LOCK_EXCL );
        if( nExcl || nShared ){
          rc = LSM_BUSY;
        }else{
          rc = lockSharedFile(db->pEnv, p, iLock, LSM_LOCK_EXCL);
          db->mLock |= (me|ms);
        }
        break;
    }

    lsmMutexLeave(db->pEnv, p->pClientMutex);
  }

  <li> <p>
       Once sufficient data has been accumulated in an in-memory tree (by
       default "sufficient data" means 1MB, including data structure 
       overhead), it is marked as "old" and a new "live" in-memory tree 
       created. An old in-memory tree is immutable - new data is always
       inserted into the live tree. There may be at most one old tree
       in memory at a time.

  <li> <p>
       The contents of an old in-memory tree may be written into the 
       database file at any point. Once its contents have been written (or
       "flushed") to the database file, the in-memory tree may be discarded.
       Flushing an in-memory tree to the database file creates a new database
       "segment". A database segment is an immutable b-tree structure stored
       within the database file. A single database file may contain up to 64 
       segments.

  <li> <p>
       At any point, two or more existing segments within the database file
       may be merged together into a single segment. Once their contents has
       been merged into the new segment, the original segments may be 
       discarded.

  <li> <p>
       After the set of segments in a database file has been modified (either
       by flushing an in-memory tree to disk or by merging existing segments
       together), the changes may be made persistent by "checkpointing" the 
       database. Checkpointing involves updating the database file header and
       and (usually) syncing the contents of the database file to disk.
</ol>

<p>Steps 3 and 4 above are known as "working" on the database. Step 5 is
refered to as "checkpointing". By default, database connections perform work
and checkpoint operations periodically from within calls to API functions
<code>lsm_insert</code>, <code>lsm_delete</code>, <code>lsm_delete_range</code>
and <code>lsm_commit</code> (i.e. functions that write to the database).

    closing read and write transactions. 

    <p>This option can only be set before lsm_open() is called on the database
    connection.

    <p>If this option is set to false and there is already a connection to the
    database from another process when lsm_open() is called, the lsm_open()
    call fails with error code LSM_BUSY.


  <dt> <a href=lsmapi.wiki#LSM_CONFIG_SAFETY>LSM_CONFIG_SAFETY</a>
  <dd> <p style=margin-top:0>
    The effect of this option on <a href=#data_durability>data durability</a>
    is described above.

    <p>From a performance point of view, this option determines how often the

the nMerge argument set to 1 and the third parameter set to a negative value
(interpreted as - keep working until there is no more work to do). For 
example:

<verbatim>
  rc = lsm_work(db, 1, -1, 0);
</verbatim>

<p><span style=color:red>todo: the -1 as the 3rd argument above is currently
not supported</span>

<p>When optimizing the database as above, either the LSM_CONFIG_AUTOCHECKPOINT
parameter should be set to a non-zero value or lsm_checkpoint() should be
called periodically. Otherwise, no checkpoints will be performed, preventing
the library from reusing any space occupied by old segments even after their
content has been merged into the new segment. The result - a database file that
is optimized, except that it is up to twice as large as it otherwise would be.