SQLite

** stream that exceeds the I/O capability of the background writer thread,
** the queue of pending write operations will grow without bound until we
** run out of memory.  Users of this technique may want to keep track of
** the quantity of pending writes and stop accepting new write requests
** when the buffer gets to be too big.
*/

/* 
** If this symbol is defined, then file-system locks are obtained as
** required. This slows things down, but allows multiple processes
** to access the database concurrently.
*/
#define ENABLE_FILE_LOCKING

#include "sqliteInt.h"
#include <tcl.h>

/*
** This test uses pthreads and hence only works on unix and with
** a threadsafe build of SQLite.
*/
#if OS_UNIX && SQLITE_THREADSAFE


/*
** This demo uses pthreads.  If you do not have a pthreads implementation
** for your operating system, you will need to recode the threading 
** logic.
*/
#include <pthread.h>
#include <sched.h>

/* Useful macros used in several places */
#define MIN(x,y) ((x)<(y)?(x):(y))
#define MAX(x,y) ((x)>(y)?(x):(y))

/* Forward references */
typedef struct AsyncWrite AsyncWrite;
typedef struct AsyncFile AsyncFile;
typedef struct AsyncFileData AsyncFileData;
typedef struct AsyncFileLock AsyncFileLock;
typedef struct AsyncLock AsyncLock;

/* Enable for debugging */
static int sqlite3async_trace = 0;
# define ASYNC_TRACE(X) if( sqlite3async_trace ) asyncTrace X
static void asyncTrace(const char *zFormat, ...){
  char *z;
  va_list ap;

**     xOpenXXX (three versions)
**     xDelete
**     xFileExists
**     xSyncDirectory
**
** File handle operations (invoked by SQLite thread):
**
**         asyncWrite, asyncClose, asyncTruncate, asyncSync, 
**         asyncWrite, asyncClose, asyncTruncate, asyncSync 
**         asyncSetFullSync, asyncOpenDirectory.
**    
**     The operations above add an entry to the global write-op list. They
**     prepare the entry, acquire the async.queueMutex momentarily while
**     list pointers are  manipulated to insert the new entry, then release
**     the mutex and signal the writer thread to wake up in case it happens
**     to be asleep.
**

**    
**     These primitives implement in-process locking using a hash table
**     on the file name.  Files are locked correctly for connections coming
**     from the same process.  But other processes cannot see these locks
**     and will therefore not honor them.
**
**
**         asyncFileHandle.
**    
**     The sqlite3OsFileHandle() function is currently only used when 
**     debugging the pager module. Unless sqlite3OsClose() is called on the
**     file (shouldn't be possible for other reasons), the underlying 
**     implementations are safe to call without grabbing any mutex. So we just
**     go ahead and call it no matter what any other threads are doing.
**
**    
**         asyncSeek.
**
**     Calling this method just manipulates the AsyncFile.iOffset variable. 
**     Since this variable is never accessed by writer thread, this
**     function does not require the mutex.  Actual calls to OsSeek() take 
**     place just before OsWrite() or OsRead(), which are always protected by 
**     the mutex.
**
** The writer thread:
**
**     The async.writerMutex is used to make sure only there is only
**     a single writer thread running at a time.
**
**     Inside the writer thread is a loop that works like this:
**

#ifndef SQLITE_ASYNC_TWO_FILEHANDLES
/* #define SQLITE_ASYNC_TWO_FILEHANDLES 0 */
#define SQLITE_ASYNC_TWO_FILEHANDLES 1
#endif

/*
** State information is held in the static variable "async" defined
** as follows:
** as the following structure.
**
** Both async.ioError and async.nFile are protected by async.queueMutex.
*/
static struct TestAsyncStaticData {
  pthread_mutex_t queueMutex;  /* Mutex for access to write operation queue */
  pthread_mutex_t writerMutex; /* Prevents multiple writer threads */
  pthread_mutex_t lockMutex;   /* For access to aLock hash table */
  pthread_cond_t queueSignal;  /* For waking up sleeping writer thread */
  pthread_cond_t emptySignal;  /* Notify when the write queue is empty */

#define ASYNC_NOOP          0
#define ASYNC_WRITE         1
#define ASYNC_SYNC          2
#define ASYNC_TRUNCATE      3
#define ASYNC_CLOSE         4
#define ASYNC_DELETE        5
#define ASYNC_OPENEXCLUSIVE 6
#define ASYNC_UNLOCK        7

/* Names of opcodes.  Used for debugging only.
** Make sure these stay in sync with the macros above!
*/
static const char *azOpcodeName[] = {
  "NOOP", "WRITE", "SYNC", "TRUNCATE", "CLOSE", "DELETE", "OPENEX"
  "NOOP", "WRITE", "SYNC", "TRUNCATE", "CLOSE", "DELETE", "OPENEX", "UNLOCK"
};

/*
** Entries on the write-op queue are instances of the AsyncWrite
** structure, defined here.
**
** The interpretation of the iOffset and nByte variables varies depending 

**     iOffset -> Contains the "syncDir" flag.
**     nByte   -> Number of bytes of zBuf points to (file name).
**
** ASYNC_OPENEXCLUSIVE:
**     iOffset -> Value of "delflag".
**     nByte   -> Number of bytes of zBuf points to (file name).
**
** ASYNC_UNLOCK:
**     nByte   -> Argument to sqlite3OsUnlock().
**
**
** For an ASYNC_WRITE operation, zBuf points to the data to write to the file. 
** This space is sqlite3_malloc()d along with the AsyncWrite structure in a
** single blob, so is deleted when sqlite3_free() is called on the parent 
** structure.
*/
struct AsyncWrite {

/*
** An instance of the following structure is allocated along with each
** AsyncFileData structure (see AsyncFileData.lock), but is only used if the
** file was opened with the SQLITE_OPEN_MAIN_DB.
**
** The global async.aLock[] hash table maps from database file-name to a
** linked-list of AsyncLock structures corresponding to handles opened on the
** file. The AsyncLock structures are linked into the list when the file is
** opened and removed when it is closed. Mutex async.lockMutex must be held
** before accessing any AsyncLock structure or the async.aLock[] table.
** linked-list of AsyncFileLock structures corresponding to handles opened on
** the file. The AsyncFileLock structures are linked into the list when the
** file is opened and removed when it is closed. Mutex async.lockMutex must be
** held before accessing any AsyncFileLock structure or the async.aLock[]
** table.
*/
typedef struct AsyncLock AsyncLock;
struct AsyncFileLock {
  int eLock;                /* Internally visible lock state (sqlite pov) */
  int eAsyncLock;           /* Lock-state with write-queue unlock */
  AsyncFileLock *pNext;
};

struct AsyncLock {
  sqlite3_file *pFile;
  int eLock;
  AsyncLock *pNext;
  AsyncFileLock *pList;
};

/* 
** The AsyncFile structure is a subclass of sqlite3_file used for 
** asynchronous IO. 
**
** All of the actual data for the structure is stored in the structure

  AsyncFileData *pData;
};
struct AsyncFileData {
  char *zName;               /* Underlying OS filename - used for debugging */
  int nName;                 /* Number of characters in zName */
  sqlite3_file *pBaseRead;   /* Read handle to the underlying Os file */
  sqlite3_file *pBaseWrite;  /* Write handle to the underlying Os file */
  AsyncLock lock;
  AsyncFileLock lock;
};

/*
** Add an entry to the end of the global write-op list. pWrite should point 
** to an AsyncWrite structure allocated using sqlite3_malloc().  The writer
** thread will call sqlite3_free() to free the structure after the specified
** operation has been completed.

*/
static int asyncRead(sqlite3_file *pFile, void *zOut, int iAmt, i64 iOffset){
  AsyncFileData *p = ((AsyncFile *)pFile)->pData;
  int rc = SQLITE_OK;
  i64 filesize;
  int nRead;
  sqlite3_file *pBase = p->pBaseRead;

  /* Grab the write queue mutex for the duration of the call */
  pthread_mutex_lock(&async.queueMutex);

  /* If an I/O error has previously occurred in this virtual file 
  ** system, then all subsequent operations fail.
  */
  if( async.ioError!=SQLITE_OK ){
    return async.ioError;
    rc = async.ioError;
    goto asyncread_out;
  }

  /* Grab the write queue mutex for the duration of the call */
  pthread_mutex_lock(&async.queueMutex);

  if( pBase->pMethods ){
    rc = sqlite3OsFileSize(pBase, &filesize);
    if( rc!=SQLITE_OK ){
      goto asyncread_out;
    }
    nRead = MIN(filesize - iOffset, iAmt);
    if( nRead>0 ){

      }
    }
    *piSize = s;
  }
  pthread_mutex_unlock(&async.queueMutex);
  return rc;
}

/*
** Lock or unlock the actual file-system entry.
*/
static int getFileLock(AsyncLock *pLock){
  int rc = SQLITE_OK;
  AsyncFileLock *pIter;
  int eRequired = 0;

  if( pLock->pFile ){
    for(pIter=pLock->pList; pIter; pIter=pIter->pNext){
      assert(pIter->eAsyncLock>=pIter->eLock);
      if( pIter->eAsyncLock>eRequired ){
        eRequired = pIter->eAsyncLock;
      }
    }
    if( eRequired>pLock->eLock ){
      rc = sqlite3OsLock(pLock->pFile, eRequired);
    }else if(eRequired<pLock->eLock){
      rc = sqlite3OsUnlock(pLock->pFile, eRequired);
    }
    if( rc==SQLITE_OK ){
      pLock->eLock = eRequired;
    }
  }

  return rc;
}

/*
** No disk locking is performed.  We keep track of locks locally in
** the async.aLock hash table.  Locking should appear to work the same
** as with standard (unmodified) SQLite as long as all connections 
** come from this one process.  Connections from external processes
** cannot see our internal hash table (obviously) and will thus not
** honor our locks.
*/
static int asyncLock(sqlite3_file *pFile, int eLock){
  int rc = SQLITE_OK;
  AsyncFileData *p = ((AsyncFile *)pFile)->pData;

  pthread_mutex_lock(&async.lockMutex);
  if( p->lock.eLock<eLock ){
    AsyncLock *pLock;
    AsyncFileLock *pIter;
    pLock = (AsyncLock *)sqlite3HashFind(&async.aLock, p->zName, p->nName);
    assert(pLock);
    for(/*no-op*/; pLock; pLock=pLock->pNext){
      if( pLock!=&p->lock && (
        (eLock==SQLITE_LOCK_EXCLUSIVE && pLock->eLock>=SQLITE_LOCK_SHARED) ||
        (eLock==SQLITE_LOCK_PENDING && pLock->eLock>=SQLITE_LOCK_RESERVED) ||
        (eLock==SQLITE_LOCK_RESERVED && pLock->eLock>=SQLITE_LOCK_RESERVED) ||
        (eLock==SQLITE_LOCK_SHARED && pLock->eLock>=SQLITE_LOCK_PENDING)
    assert(pLock && pLock->pList);
    for(pIter=pLock->pList; pIter; pIter=pIter->pNext){
      if( pIter!=&p->lock && (
        (eLock==SQLITE_LOCK_EXCLUSIVE && pIter->eLock>=SQLITE_LOCK_SHARED) ||
        (eLock==SQLITE_LOCK_PENDING && pIter->eLock>=SQLITE_LOCK_RESERVED) ||
        (eLock==SQLITE_LOCK_RESERVED && pIter->eLock>=SQLITE_LOCK_RESERVED) ||
        (eLock==SQLITE_LOCK_SHARED && pIter->eLock>=SQLITE_LOCK_PENDING)
      )){
        rc = SQLITE_BUSY;
      }
    }
    if( rc==SQLITE_OK ){
      p->lock.eLock = eLock;
      if( eLock>p->lock.eAsyncLock ){
        p->lock.eAsyncLock = eLock;
      }
    }
    assert(p->lock.eAsyncLock>=p->lock.eLock);
    if( rc==SQLITE_OK ){
      rc = getFileLock(pLock);
    }
  }
  pthread_mutex_unlock(&async.lockMutex);

  ASYNC_TRACE(("LOCK %d (%s) rc=%d\n", eLock, p->zName, rc));
  return rc;
}
static int asyncUnlock(sqlite3_file *pFile, int eLock){
  AsyncFileData *p = ((AsyncFile *)pFile)->pData;
  AsyncLock *pLock = &p->lock;
  AsyncFileLock *pLock = &p->lock;
  pthread_mutex_lock(&async.lockMutex);
  if( pLock->eLock>eLock ){
    pLock->eLock = eLock;
  }
  pthread_mutex_unlock(&async.lockMutex);
  return SQLITE_OK;
  return addNewAsyncWrite(p, ASYNC_UNLOCK, 0, eLock, 0);
}

/*
** This function is called when the pager layer first opens a database file
** and is checking for a hot-journal.
*/
static int asyncCheckReservedLock(sqlite3_file *pFile){
  int ret = 0;
  AsyncFileLock *pIter;
  AsyncLock *pLock;
  AsyncFileData *p = ((AsyncFile *)pFile)->pData;

  pthread_mutex_lock(&async.lockMutex);
  pLock = (AsyncLock *)sqlite3HashFind(&async.aLock, p->zName, p->nName);
  for(/*no-op*/; pLock; pLock=pLock->pNext){
    if( pLock->eLock>=SQLITE_LOCK_RESERVED ){
  for(pIter=pLock->pList; pIter; pIter=pIter->pNext){
    if( pIter->eLock>=SQLITE_LOCK_RESERVED ){
      ret = 1;
    }
  }
  pthread_mutex_unlock(&async.lockMutex);

  ASYNC_TRACE(("CHECK-LOCK %d (%s)\n", ret, p->zName));
  return ret;
}

/* 
** This is a no-op, as the asynchronous backend does not support locking.
*/
static int asyncFileControl(sqlite3_file *id, int op, void *pArg){
  switch( op ){
    case SQLITE_FCNTL_LOCKSTATE: {
      pthread_mutex_lock(&async.lockMutex);
      *(int*)pArg = ((AsyncFile*)id)->pData->lock.eLock;
      pthread_mutex_unlock(&async.lockMutex);
      return SQLITE_OK;
    }
  }
  return SQLITE_ERROR;
}

/* 
** Return the device characteristics and sector-size of the device. It
** is not tricky to implement these correctly, as this backend might 
** not have an open file handle at this point.

  sqlite3_vfs *pVfs = (sqlite3_vfs *)pAsyncVfs->pAppData;
  AsyncFile *p = (AsyncFile *)pFile;
  int nName = strlen(zName)+1;
  int rc;
  int nByte;
  AsyncFileData *pData;

  AsyncLock *pLock = 0;

  nByte = (
    sizeof(AsyncFileData) +        /* AsyncFileData structure */
    2 * pVfs->szOsFile +           /* AsyncFileData.pBaseRead and pBaseWrite */
    nName                          /* AsyncFileData.zName */
  ); 
  pData = sqlite3_malloc(nByte);
  if( !pData ){

    if( pOutFlags ) *pOutFlags = flags;
  }else{
    rc = sqlite3OsOpen(pVfs, zName, pData->pBaseRead, flags, pOutFlags);
    if( rc==SQLITE_OK && ((*pOutFlags)&SQLITE_OPEN_READWRITE) ){
      rc = sqlite3OsOpen(pVfs, zName, pData->pBaseWrite, flags, 0);
    }
  }

  pthread_mutex_lock(&async.lockMutex);

  if( rc==SQLITE_OK ){
    pLock = sqlite3HashFind(&async.aLock, pData->zName, pData->nName);
    if( !pLock ){
      pLock = sqlite3MallocZero(pVfs->szOsFile + sizeof(AsyncLock));
      if( pLock ){
#ifdef ENABLE_FILE_LOCKING
        if( flags&SQLITE_OPEN_MAIN_DB ){
          pLock->pFile = (sqlite3_file *)&pLock[1];
          rc = sqlite3OsOpen(pVfs, zName, pLock->pFile, flags, 0);
          if( rc!=SQLITE_OK ){
            sqlite3_free(pLock);
            pLock = 0;
          }
        }
#endif
        sqlite3HashInsert(
          &async.aLock, pData->zName, pData->nName, (void *)pLock
        );
      }else{
        rc = SQLITE_NOMEM;
      }
    }
  }

  if( rc==SQLITE_OK ){
    HashElem *pElem;
    p->pMethod = &async_methods;
    p->pData = pData;
    incrOpenFileCount();

    /* Link AsyncFileData.lock into the linked list of AsyncLock structures
    ** for this file. Obtain the async.lockMutex mutex before doing so.
    /* Link AsyncFileData.lock into the linked list of 
    ** AsyncFileLock structures for this file.
    */
    AsyncLock *pNext;
    pthread_mutex_lock(&async.lockMutex);
    pNext = sqlite3HashInsert(
        &async.aLock, pData->zName, pData->nName, (void *)&pData->lock
    );
    pData->lock.pNext = pNext;
    pData->lock.pNext = pLock->pList;
    pLock->pList = &pData->lock;

    pElem = sqlite3HashFindElem(&async.aLock, pData->zName, pData->nName);
    pData->zName = (char *)sqliteHashKey(pElem);
    pthread_mutex_unlock(&async.lockMutex);
  }else{
    sqlite3OsClose(pData->pBaseRead);
    sqlite3OsClose(pData->pBaseWrite);
    sqlite3_free(pData);
  }

  pthread_mutex_unlock(&async.lockMutex);
  return rc;
}

/*
** Implementation of sqlite3OsDelete. Add an entry to the end of the 
** write-op queue to perform the delete.
*/

        ASYNC_TRACE(("TRUNCATE %s to %d bytes\n", 
                p->pFileData->zName, p->iOffset));
        rc = sqlite3OsTruncate(pBase, p->iOffset);
        break;

      case ASYNC_CLOSE: {
        AsyncLock *pLock;
        AsyncFileLock **ppIter;
        AsyncFileData *pData = p->pFileData;
        ASYNC_TRACE(("CLOSE %s\n", p->pFileData->zName));
        sqlite3OsClose(pData->pBaseWrite);
        sqlite3OsClose(pData->pBaseRead);

        /* Unlink AsyncFileData.lock from the linked list of AsyncLock 
        /* Unlink AsyncFileData.lock from the linked list of AsyncFileLock 
        ** structures for this file. Obtain the async.lockMutex mutex 
        ** before doing so.
        */
        pthread_mutex_lock(&async.lockMutex);
        pLock = sqlite3HashFind(&async.aLock, pData->zName, pData->nName);
        for(ppIter=&pLock->pList; *ppIter; ppIter=&((*ppIter)->pNext)){
        if( pLock==&pData->lock ){
          sqlite3HashInsert(
              &async.aLock, pData->zName, pData->nName, (void *)pLock->pNext 
          if( (*ppIter)==&pData->lock ){
            *ppIter = (*ppIter)->pNext;
            break;
          }
        }
        if( !pLock->pList ){
          if( pLock->pFile ) sqlite3OsClose(pLock->pFile);
          sqlite3_free(pLock);
          sqlite3HashInsert(&async.aLock, pData->zName, pData->nName, 0);
          );
        }else{
          for( ; pLock && pLock->pNext!=&pData->lock; pLock=pLock->pNext);
          if( pLock ){
          rc = getFileLock(pLock);
            pLock->pNext = pData->lock.pNext;
          }
        }
        pthread_mutex_unlock(&async.lockMutex);

        sqlite3_free(pData);
        break;
      }

      case ASYNC_UNLOCK: {
        AsyncLock *pLock;
        AsyncFileData *pData = p->pFileData;
        int eLock = p->nByte;
        pthread_mutex_lock(&async.lockMutex);
        if( pData->lock.eAsyncLock>eLock ){
          if( pData->lock.eLock>eLock ){
            pData->lock.eAsyncLock = pData->lock.eLock;
          }else{
            pData->lock.eAsyncLock = eLock;
          }
        }
        assert(pData->lock.eAsyncLock>=pData->lock.eLock);
        pLock = sqlite3HashFind(&async.aLock, pData->zName, pData->nName);
        rc = getFileLock(pLock);
        pthread_mutex_unlock(&async.lockMutex);
        break;
      }

      case ASYNC_DELETE:
        ASYNC_TRACE(("DELETE %s\n", p->zBuf));
        rc = sqlite3OsDelete(pVfs, p->zBuf, (int)p->iOffset);
        break;

      case ASYNC_OPENEXCLUSIVE: {

#
#    May you do good and not evil.
#    May you find forgiveness for yourself and forgive others.
#    May you share freely, never taking more than you give.
#
#***********************************************************************
# This file runs all tests.
#
# $Id: async.test,v 1.8 2007/09/04 14:31:47 danielk1977 Exp $
# $Id: async.test,v 1.9 2007/09/04 18:28:44 danielk1977 Exp $


if {[catch {sqlite3async_enable}]} {
  # The async logic is not built into this system
  return
}

  select4.test
  insert.test
  insert2.test
  insert3.test
  trans.test
  lock.test
  lock3.test
  lock2.test
}
# set INCLUDE lock4.test

# Enable asynchronous IO.
sqlite3async_enable 1

rename do_test really_do_test
proc do_test {name args} {
  uplevel really_do_test async_io-$name $args