text: /*
  ** 2005 December 14
  **
  ** The author disclaims copyright to this source code.  In place of
  ** a legal notice, here is a blessing:
  **
  **    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.
  **
  *************************************************************************
  **
  ** $Id: test_async.c,v 1.48 2008/09/26 20:02:50 drh Exp $
  **
  ** This file contains an example implementation of an asynchronous IO 
  ** backend for SQLite.
  **
  ** WHAT IS ASYNCHRONOUS I/O?
  **
  ** With asynchronous I/O, write requests are handled by a separate thread
  ** running in the background.  This means that the thread that initiates
  ** a database write does not have to wait for (sometimes slow) disk I/O
  ** to occur.  The write seems to happen very quickly, though in reality
  ** it is happening at its usual slow pace in the background.
  **
  ** Asynchronous I/O appears to give better responsiveness, but at a price.
  ** You lose the Durable property.  With the default I/O backend of SQLite,
  ** once a write completes, you know that the information you wrote is
  ** safely on disk.  With the asynchronous I/O, this is not the case.  If
  ** your program crashes or if a power loss occurs after the database
  ** write but before the asynchronous write thread has completed, then the
  ** database change might never make it to disk and the next user of the
  ** database might not see your change.
  **
  ** You lose Durability with asynchronous I/O, but you still retain the
  ** other parts of ACID:  Atomic,  Consistent, and Isolated.  Many
  ** appliations get along fine without the Durablity.
  **
  ** HOW IT WORKS
  **
  ** Asynchronous I/O works by creating a special SQLite "vfs" structure
  ** and registering it with sqlite3_vfs_register(). When files opened via 
  ** this vfs are written to (using sqlite3OsWrite()), the data is not 
  ** written directly to disk, but is placed in the "write-queue" to be
  ** handled by the background thread.
  **
  ** When files opened with the asynchronous vfs are read from 
  ** (using sqlite3OsRead()), the data is read from the file on 
  ** disk and the write-queue, so that from the point of view of
  ** the vfs reader the OsWrite() appears to have already completed.
  **
  ** The special vfs is registered (and unregistered) by calls to 
  ** function asyncEnable() (see below).
  **
  ** LIMITATIONS
  **
  ** This demonstration code is deliberately kept simple in order to keep
  ** the main ideas clear and easy to understand.  Real applications that
  ** want to do asynchronous I/O might want to add additional capabilities.
  ** For example, in this demonstration if writes are happening at a steady
  ** 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.
  **
  ** LOCKING + CONCURRENCY
  **
  ** Multiple connections from within a single process that use this
  ** implementation of asynchronous IO may access a single database
  ** file concurrently. From the point of view of the user, if all
  ** connections are from within a single process, there is no difference
  ** between the concurrency offered by "normal" SQLite and SQLite
  ** using the asynchronous backend.
  **
  ** If connections from within multiple database files may access the
  ** database file, the ENABLE_FILE_LOCKING symbol (see below) must be
  ** defined. If it is not defined, then no locks are established on 
  ** the database file. In this case, if multiple processes access 
  ** the database file, corruption will quickly result.
  **
  ** If ENABLE_FILE_LOCKING is defined (the default), then connections 
  ** from within multiple processes may access a single database file 
  ** without risking corruption. However concurrency is reduced as
  ** follows:
  **
  **   * When a connection using asynchronous IO begins a database
  **     transaction, the database is locked immediately. However the
  **     lock is not released until after all relevant operations
  **     in the write-queue have been flushed to disk. This means
  **     (for example) that the database may remain locked for some 
  **     time after a "COMMIT" or "ROLLBACK" is issued.
  **
  **   * If an application using asynchronous IO executes transactions
  **     in quick succession, other database users may be effectively
  **     locked out of the database. This is because when a BEGIN
  **     is executed, a database lock is established immediately. But
  **     when the corresponding COMMIT or ROLLBACK occurs, the lock
  **     is not released until the relevant part of the write-queue 
  **     has been flushed through. As a result, if a COMMIT is followed
  **     by a BEGIN before the write-queue is flushed through, the database 
  **     is never unlocked,preventing other processes from accessing 
  **     the database.
  **
  ** Defining ENABLE_FILE_LOCKING when using an NFS or other remote 
  ** file-system may slow things down, as synchronous round-trips to the 
  ** server may be required to establish database file locks.
  */
  #define ENABLE_FILE_LOCKING
  
  #ifndef SQLITE_AMALGAMATION
  # include "sqlite3.h"
  # include <assert.h>
  # include <string.h>
  #endif
  #include <tcl.h>
  
  /*
  ** This test uses pthreads and hence only works on unix and with
  ** a threadsafe build of SQLite.
  */
  #if SQLITE_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;
    va_start(ap, zFormat);
    z = sqlite3_vmprintf(zFormat, ap);
    va_end(ap);
    fprintf(stderr, "[%d] %s", (int)pthread_self(), z);
    sqlite3_free(z);
  }
  
  /*
  ** THREAD SAFETY NOTES
  **
  ** Basic rules:
  **
  **     * Both read and write access to the global write-op queue must be 
  **       protected by the async.queueMutex. As are the async.ioError and
  **       async.nFile variables.
  **
  **     * The async.pLock list and all AsyncLock and AsyncFileLock
  **       structures must be protected by the async.lockMutex mutex.
  **
  **     * The file handles from the underlying system are not assumed to 
  **       be thread safe.
  **
  **     * See the last two paragraphs under "The Writer Thread" for
  **       an assumption to do with file-handle synchronization by the Os.
  **
  ** Deadlock prevention:
  **
  **     There are three mutex used by the system: the "writer" mutex, 
  **     the "queue" mutex and the "lock" mutex. Rules are:
  **
  **     * It is illegal to block on the writer mutex when any other mutex
  **       are held, and 
  **
  **     * It is illegal to block on the queue mutex when the lock mutex
  **       is held.
  **
  **     i.e. mutex's must be grabbed in the order "writer", "queue", "lock".
  **
  ** File system operations (invoked by SQLite thread):
  **
  **     xOpen
  **     xDelete
  **     xFileExists
  **
  ** File handle operations (invoked by SQLite thread):
  **
  **         asyncWrite, asyncClose, asyncTruncate, asyncSync 
  **    
  **     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.
  **
  **    
  **         asyncRead, asyncFileSize.
  **
  **     Read operations. Both of these read from both the underlying file
  **     first then adjust their result based on pending writes in the 
  **     write-op queue.   So async.queueMutex is held for the duration
  **     of these operations to prevent other threads from changing the
  **     queue in mid operation.
  **    
  **
  **         asyncLock, asyncUnlock, asyncCheckReservedLock
  **    
  **     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.
  **
  **
  ** 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:
  **
  **         WHILE (write-op list is not empty)
  **             Do IO operation at head of write-op list
  **             Remove entry from head of write-op list
  **         END WHILE
  **
  **     The async.queueMutex is always held during the <write-op list is 
  **     not empty> test, and when the entry is removed from the head
  **     of the write-op list. Sometimes it is held for the interim
  **     period (while the IO is performed), and sometimes it is
  **     relinquished. It is relinquished if (a) the IO op is an
  **     ASYNC_CLOSE or (b) when the file handle was opened, two of
  **     the underlying systems handles were opened on the same
  **     file-system entry.
  **
  **     If condition (b) above is true, then one file-handle 
  **     (AsyncFile.pBaseRead) is used exclusively by sqlite threads to read the
  **     file, the other (AsyncFile.pBaseWrite) by sqlite3_async_flush() 
  **     threads to perform write() operations. This means that read 
  **     operations are not blocked by asynchronous writes (although 
  **     asynchronous writes may still be blocked by reads).
  **
  **     This assumes that the OS keeps two handles open on the same file
  **     properly in sync. That is, any read operation that starts after a
  **     write operation on the same file system entry has completed returns
  **     data consistent with the write. We also assume that if one thread 
  **     reads a file while another is writing it all bytes other than the
  **     ones actually being written contain valid data.
  **
  **     If the above assumptions are not true, set the preprocessor symbol
  **     SQLITE_ASYNC_TWO_FILEHANDLES to 0.
  */
  
  #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 the following structure.
  **
  ** Both async.ioError and async.nFile are protected by async.queueMutex.
  */
  static struct TestAsyncStaticData {
    pthread_mutex_t lockMutex;   /* For access to aLock hash table */
    pthread_mutex_t queueMutex;  /* Mutex for access to write operation queue */
    pthread_mutex_t writerMutex; /* Prevents multiple writer threads */
    pthread_cond_t queueSignal;  /* For waking up sleeping writer thread */
    pthread_cond_t emptySignal;  /* Notify when the write queue is empty */
    AsyncWrite *pQueueFirst;     /* Next write operation to be processed */
    AsyncWrite *pQueueLast;      /* Last write operation on the list */
    AsyncLock *pLock;            /* Linked list of all AsyncLock structures */
    volatile int ioDelay;             /* Extra delay between write operations */
    volatile int writerHaltWhenIdle;  /* Writer thread halts when queue empty */
    volatile int writerHaltNow;       /* Writer thread halts after next op */
    int ioError;                 /* True if an IO error has occured */
    int nFile;                   /* Number of open files (from sqlite pov) */
  } async = {
    PTHREAD_MUTEX_INITIALIZER,
    PTHREAD_MUTEX_INITIALIZER,
    PTHREAD_MUTEX_INITIALIZER,
    PTHREAD_COND_INITIALIZER,
    PTHREAD_COND_INITIALIZER,
  };
  
  /* Possible values of AsyncWrite.op */
  #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", "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 
  ** on the value of AsyncWrite.op:
  **
  ** ASYNC_NOOP:
  **     No values used.
  **
  ** ASYNC_WRITE:
  **     iOffset -> Offset in file to write to.
  **     nByte   -> Number of bytes of data to write (pointed to by zBuf).
  **
  ** ASYNC_SYNC:
  **     nByte   -> flags to pass to sqlite3OsSync().
  **
  ** ASYNC_TRUNCATE:
  **     iOffset -> Size to truncate file to.
  **     nByte   -> Unused.
  **
  ** ASYNC_CLOSE:
  **     iOffset -> Unused.
  **     nByte   -> Unused.
  **
  ** ASYNC_DELETE:
  **     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 {
    AsyncFileData *pFileData;    /* File to write data to or sync */
    int op;                      /* One of ASYNC_xxx etc. */
    sqlite_int64 iOffset;        /* See above */
    int nByte;          /* See above */
    char *zBuf;         /* Data to write to file (or NULL if op!=ASYNC_WRITE) */
    AsyncWrite *pNext;  /* Next write operation (to any file) */
  };
  
  /*
  ** An instance of this structure is created for each distinct open file 
  ** (i.e. if two handles are opened on the one file, only one of these
  ** structures is allocated) and stored in the async.aLock hash table. The
  ** keys for async.aLock are the full pathnames of the opened files.
  **
  ** AsyncLock.pList points to the head of a linked list of AsyncFileLock
  ** structures, one for each handle currently open on the file.
  **
  ** If the opened file is not a main-database (the SQLITE_OPEN_MAIN_DB is
  ** not passed to the sqlite3OsOpen() call), or if ENABLE_FILE_LOCKING is 
  ** not defined at compile time, variables AsyncLock.pFile and 
  ** AsyncLock.eLock are never used. Otherwise, pFile is a file handle
  ** opened on the file in question and used to obtain the file-system 
  ** locks required by database connections within this process.
  **
  ** See comments above the asyncLock() function for more details on 
  ** the implementation of database locking used by this backend.
  */
  struct AsyncLock {
    char *zFile;
    int nFile;
    sqlite3_file *pFile;
    int eLock;
    AsyncFileLock *pList;
    AsyncLock *pNext;           /* Next in linked list headed by async.pLock */
  };
  
  /*
  ** 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.
  */
  struct AsyncFileLock {
    int eLock;                /* Internally visible lock state (sqlite pov) */
    int eAsyncLock;           /* Lock-state with write-queue unlock */
    AsyncFileLock *pNext;
  };
  
  /* 
  ** 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
  ** pointed to by AsyncFile.pData, which is allocated as part of the
  ** sqlite3OsOpen() using sqlite3_malloc(). The reason for this is that the
  ** lifetime of the AsyncFile structure is ended by the caller after OsClose()
  ** is called, but the data in AsyncFileData may be required by the
  ** writer thread after that point.
  */
  struct AsyncFile {
    sqlite3_io_methods *pMethod;
    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 */
    AsyncFileLock lock;        /* Lock state for this handle */
    AsyncLock *pLock;          /* AsyncLock object for this file system entry */
    AsyncWrite close;
  };
  
  /*
  ** The following async_XXX functions are debugging wrappers around the
  ** corresponding pthread_XXX functions:
  **
  **     pthread_mutex_lock();
  **     pthread_mutex_unlock();
  **     pthread_mutex_trylock();
  **     pthread_cond_wait();
  **
  ** It is illegal to pass any mutex other than those stored in the
  ** following global variables of these functions.
  **
  **     async.queueMutex
  **     async.writerMutex
  **     async.lockMutex
  **
  ** If NDEBUG is defined, these wrappers do nothing except call the 
  ** corresponding pthreads function. If NDEBUG is not defined, then the
  ** following variables are used to store the thread-id (as returned
  ** by pthread_self()) currently holding the mutex, or 0 otherwise:
  **
  **     asyncdebug.queueMutexHolder
  **     asyncdebug.writerMutexHolder
  **     asyncdebug.lockMutexHolder
  **
  ** These variables are used by some assert() statements that verify
  ** the statements made in the "Deadlock Prevention" notes earlier
  ** in this file.
  */
  #ifndef NDEBUG
  
  static struct TestAsyncDebugData {
    pthread_t lockMutexHolder;
    pthread_t queueMutexHolder;
    pthread_t writerMutexHolder;
  } asyncdebug = {0, 0, 0};
  
  /*
  ** Wrapper around pthread_mutex_lock(). Checks that we have not violated
  ** the anti-deadlock rules (see "Deadlock prevention" above).
  */
  static int async_mutex_lock(pthread_mutex_t *pMutex){
    int iIdx;
    int rc;
    pthread_mutex_t *aMutex = (pthread_mutex_t *)(&async);
    pthread_t *aHolder = (pthread_t *)(&asyncdebug);
  
    /* The code in this 'ifndef NDEBUG' block depends on a certain alignment
     * of the variables in TestAsyncStaticData and TestAsyncDebugData. The
     * following assert() statements check that this has not been changed.
     *
     * Really, these only need to be run once at startup time.
     */
    assert(&(aMutex[0])==&async.lockMutex);
    assert(&(aMutex[1])==&async.queueMutex);
    assert(&(aMutex[2])==&async.writerMutex);
    assert(&(aHolder[0])==&asyncdebug.lockMutexHolder);
    assert(&(aHolder[1])==&asyncdebug.queueMutexHolder);
    assert(&(aHolder[2])==&asyncdebug.writerMutexHolder);
  
    assert( pthread_self()!=0 );
  
    for(iIdx=0; iIdx<3; iIdx++){
      if( pMutex==&aMutex[iIdx] ) break;
  
      /* This is the key assert(). Here we are checking that if the caller
       * is trying to block on async.writerMutex, neither of the other two
       * mutex are held. If the caller is trying to block on async.queueMutex,
       * lockMutex is not held.
       */
      assert(!pthread_equal(aHolder[iIdx], pthread_self()));
    }
    assert(iIdx<3);
  
    rc = pthread_mutex_lock(pMutex);
    if( rc==0 ){
      assert(aHolder[iIdx]==0);
      aHolder[iIdx] = pthread_self();
    }
    return rc;
  }
  
  /*
  ** Wrapper around pthread_mutex_unlock().
  */
  static int async_mutex_unlock(pthread_mutex_t *pMutex){
    int iIdx;
    int rc;
    pthread_mutex_t *aMutex = (pthread_mutex_t *)(&async);
    pthread_t *aHolder = (pthread_t *)(&asyncdebug);
  
    for(iIdx=0; iIdx<3; iIdx++){
      if( pMutex==&aMutex[iIdx] ) break;
    }
    assert(iIdx<3);
  
    assert(pthread_equal(aHolder[iIdx], pthread_self()));
    aHolder[iIdx] = 0;
    rc = pthread_mutex_unlock(pMutex);
    assert(rc==0);
  
    return 0;
  }
  
  /*
  ** Wrapper around pthread_mutex_trylock().
  */
  static int async_mutex_trylock(pthread_mutex_t *pMutex){
    int iIdx;
    int rc;
    pthread_mutex_t *aMutex = (pthread_mutex_t *)(&async);
    pthread_t *aHolder = (pthread_t *)(&asyncdebug);
  
    for(iIdx=0; iIdx<3; iIdx++){
      if( pMutex==&aMutex[iIdx] ) break;
    }
    assert(iIdx<3);
  
    rc = pthread_mutex_trylock(pMutex);
    if( rc==0 ){
      assert(aHolder[iIdx]==0);
      aHolder[iIdx] = pthread_self();
    }
    return rc;
  }
  
  /*
  ** Wrapper around pthread_cond_wait().
  */
  static int async_cond_wait(pthread_cond_t *pCond, pthread_mutex_t *pMutex){
    int iIdx;
    int rc;
    pthread_mutex_t *aMutex = (pthread_mutex_t *)(&async);
    pthread_t *aHolder = (pthread_t *)(&asyncdebug);
  
    for(iIdx=0; iIdx<3; iIdx++){
      if( pMutex==&aMutex[iIdx] ) break;
    }
    assert(iIdx<3);
  
    assert(pthread_equal(aHolder[iIdx],pthread_self()));
    aHolder[iIdx] = 0;
    rc = pthread_cond_wait(pCond, pMutex);
    if( rc==0 ){
      aHolder[iIdx] = pthread_self();
    }
    return rc;
  }
  
  /*
  ** Assert that the mutex is held by the current thread.
  */
  static void assert_mutex_is_held(pthread_mutex_t *pMutex){
    int iIdx;
    pthread_mutex_t *aMutex = (pthread_mutex_t *)(&async);
    pthread_t *aHolder = (pthread_t *)(&asyncdebug);
  
    for(iIdx=0; iIdx<3; iIdx++){
      if( pMutex==&aMutex[iIdx] ) break;
    }
    assert(iIdx<3);
    assert( aHolder[iIdx]==pthread_self() );
  }
  
  /* Call our async_XX wrappers instead of selected pthread_XX functions */
  #define pthread_mutex_lock    async_mutex_lock
  #define pthread_mutex_unlock  async_mutex_unlock
  #define pthread_mutex_trylock async_mutex_trylock
  #define pthread_cond_wait     async_cond_wait
  
  #else    /* if defined(NDEBUG) */
  
  #define assert_mutex_is_held(X)    /* A no-op when not debugging */
  
  #endif   /* !defined(NDEBUG) */
  
  /*
  ** 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.
  **
  ** Once an AsyncWrite structure has been added to the list, it becomes the
  ** property of the writer thread and must not be read or modified by the
  ** caller.  
  */
  static void addAsyncWrite(AsyncWrite *pWrite){
    /* We must hold the queue mutex in order to modify the queue pointers */
    pthread_mutex_lock(&async.queueMutex);
  
    /* Add the record to the end of the write-op queue */
    assert( !pWrite->pNext );
    if( async.pQueueLast ){
      assert( async.pQueueFirst );
      async.pQueueLast->pNext = pWrite;
    }else{
      async.pQueueFirst = pWrite;
    }
    async.pQueueLast = pWrite;
    ASYNC_TRACE(("PUSH %p (%s %s %d)\n", pWrite, azOpcodeName[pWrite->op],
           pWrite->pFileData ? pWrite->pFileData->zName : "-", pWrite->iOffset));
  
    if( pWrite->op==ASYNC_CLOSE ){
      async.nFile--;
    }
  
    /* Drop the queue mutex */
    pthread_mutex_unlock(&async.queueMutex);
  
    /* The writer thread might have been idle because there was nothing
    ** on the write-op queue for it to do.  So wake it up. */
    pthread_cond_signal(&async.queueSignal);
  }
  
  /*
  ** Increment async.nFile in a thread-safe manner.
  */
  static void incrOpenFileCount(){
    /* We must hold the queue mutex in order to modify async.nFile */
    pthread_mutex_lock(&async.queueMutex);
    if( async.nFile==0 ){
      async.ioError = SQLITE_OK;
    }
    async.nFile++;
    pthread_mutex_unlock(&async.queueMutex);
  }
  
  /*
  ** This is a utility function to allocate and populate a new AsyncWrite
  ** structure and insert it (via addAsyncWrite() ) into the global list.
  */
  static int addNewAsyncWrite(
    AsyncFileData *pFileData, 
    int op, 
    sqlite3_int64 iOffset, 
    int nByte,
    const char *zByte
  ){
    AsyncWrite *p;
    if( op!=ASYNC_CLOSE && async.ioError ){
      return async.ioError;
    }
    p = sqlite3_malloc(sizeof(AsyncWrite) + (zByte?nByte:0));
    if( !p ){
      /* The upper layer does not expect operations like OsWrite() to
      ** return SQLITE_NOMEM. This is partly because under normal conditions
      ** SQLite is required to do rollback without calling malloc(). So
      ** if malloc() fails here, treat it as an I/O error. The above
      ** layer knows how to handle that.
      */
      return SQLITE_IOERR;
    }
    p->op = op;
    p->iOffset = iOffset;
    p->nByte = nByte;
    p->pFileData = pFileData;
    p->pNext = 0;
    if( zByte ){
      p->zBuf = (char *)&p[1];
      memcpy(p->zBuf, zByte, nByte);
    }else{
      p->zBuf = 0;
    }
    addAsyncWrite(p);
    return SQLITE_OK;
  }
  
  /*
  ** Close the file. This just adds an entry to the write-op list, the file is
  ** not actually closed.
  */
  static int asyncClose(sqlite3_file *pFile){
    AsyncFileData *p = ((AsyncFile *)pFile)->pData;
  
    /* Unlock the file, if it is locked */
    pthread_mutex_lock(&async.lockMutex);
    p->lock.eLock = 0;
    pthread_mutex_unlock(&async.lockMutex);
  
    addAsyncWrite(&p->close);
    return SQLITE_OK;
  }
  
  /*
  ** Implementation of sqlite3OsWrite() for asynchronous files. Instead of 
  ** writing to the underlying file, this function adds an entry to the end of
  ** the global AsyncWrite list. Either SQLITE_OK or SQLITE_NOMEM may be
  ** returned.
  */
  static int asyncWrite(
    sqlite3_file *pFile, 
    const void *pBuf, 
    int amt, 
    sqlite3_int64 iOff
  ){
    AsyncFileData *p = ((AsyncFile *)pFile)->pData;
    return addNewAsyncWrite(p, ASYNC_WRITE, iOff, amt, pBuf);
  }
  
  /*
  ** Read data from the file. First we read from the filesystem, then adjust 
  ** the contents of the buffer based on ASYNC_WRITE operations in the 
  ** write-op queue.
  **
  ** This method holds the mutex from start to finish.
  */
  static int asyncRead(
    sqlite3_file *pFile, 
    void *zOut, 
    int iAmt, 
    sqlite3_int64 iOffset
  ){
    AsyncFileData *p = ((AsyncFile *)pFile)->pData;
    int rc = SQLITE_OK;
    sqlite3_int64 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 ){
      rc = async.ioError;
      goto asyncread_out;
    }
  
    if( pBase->pMethods ){
      rc = pBase->pMethods->xFileSize(pBase, &filesize);
      if( rc!=SQLITE_OK ){
        goto asyncread_out;
      }
      nRead = MIN(filesize - iOffset, iAmt);
      if( nRead>0 ){
        rc = pBase->pMethods->xRead(pBase, zOut, nRead, iOffset);
        ASYNC_TRACE(("READ %s %d bytes at %d\n", p->zName, nRead, iOffset));
      }
    }
  
    if( rc==SQLITE_OK ){
      AsyncWrite *pWrite;
      char *zName = p->zName;
  
      for(pWrite=async.pQueueFirst; pWrite; pWrite = pWrite->pNext){
        if( pWrite->op==ASYNC_WRITE && (
          (pWrite->pFileData==p) ||
          (zName && pWrite->pFileData->zName==zName)
        )){
          int iBeginOut = (pWrite->iOffset-iOffset);
          int iBeginIn = -iBeginOut;
          int nCopy;
  
          if( iBeginIn<0 ) iBeginIn = 0;
          if( iBeginOut<0 ) iBeginOut = 0;
          nCopy = MIN(pWrite->nByte-iBeginIn, iAmt-iBeginOut);
  
          if( nCopy>0 ){
            memcpy(&((char *)zOut)[iBeginOut], &pWrite->zBuf[iBeginIn], nCopy);
            ASYNC_TRACE(("OVERREAD %d bytes at %d\n", nCopy, iBeginOut+iOffset));
          }
        }
      }
    }
  
  asyncread_out:
    pthread_mutex_unlock(&async.queueMutex);
    return rc;
  }
  
  /*
  ** Truncate the file to nByte bytes in length. This just adds an entry to 
  ** the write-op list, no IO actually takes place.
  */
  static int asyncTruncate(sqlite3_file *pFile, sqlite3_int64 nByte){
    AsyncFileData *p = ((AsyncFile *)pFile)->pData;
    return addNewAsyncWrite(p, ASYNC_TRUNCATE, nByte, 0, 0);
  }
  
  /*
  ** Sync the file. This just adds an entry to the write-op list, the 
  ** sync() is done later by sqlite3_async_flush().
  */
  static int asyncSync(sqlite3_file *pFile, int flags){
    AsyncFileData *p = ((AsyncFile *)pFile)->pData;
    return addNewAsyncWrite(p, ASYNC_SYNC, 0, flags, 0);
  }
  
  /*
  ** Read the size of the file. First we read the size of the file system 
  ** entry, then adjust for any ASYNC_WRITE or ASYNC_TRUNCATE operations 
  ** currently in the write-op list. 
  **
  ** This method holds the mutex from start to finish.
  */
  int asyncFileSize(sqlite3_file *pFile, sqlite3_int64 *piSize){
    AsyncFileData *p = ((AsyncFile *)pFile)->pData;
    int rc = SQLITE_OK;
    sqlite3_int64 s = 0;
    sqlite3_file *pBase;
  
    pthread_mutex_lock(&async.queueMutex);
  
    /* Read the filesystem size from the base file. If pBaseRead is NULL, this
    ** means the file hasn't been opened yet. In this case all relevant data 
    ** must be in the write-op queue anyway, so we can omit reading from the
    ** file-system.
    */
    pBase = p->pBaseRead;
    if( pBase->pMethods ){
      rc = pBase->pMethods->xFileSize(pBase, &s);
    }
  
    if( rc==SQLITE_OK ){
      AsyncWrite *pWrite;
      for(pWrite=async.pQueueFirst; pWrite; pWrite = pWrite->pNext){
        if( pWrite->op==ASYNC_DELETE 
         && p->zName 
         && strcmp(p->zName, pWrite->zBuf)==0 
        ){
          s = 0;
        }else if( pWrite->pFileData && (
            (pWrite->pFileData==p) 
         || (p->zName && pWrite->pFileData->zName==p->zName) 
        )){
          switch( pWrite->op ){
            case ASYNC_WRITE:
              s = MAX(pWrite->iOffset + (sqlite3_int64)(pWrite->nByte), s);
              break;
            case ASYNC_TRUNCATE:
              s = MIN(s, pWrite->iOffset);
              break;
          }
        }
      }
      *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;
          assert(eRequired>=0 && eRequired<=SQLITE_LOCK_EXCLUSIVE);
        }
      }
  
      if( eRequired>pLock->eLock ){
        rc = pLock->pFile->pMethods->xLock(pLock->pFile, eRequired);
        if( rc==SQLITE_OK ){
          pLock->eLock = eRequired;
        }
      }
      else if( eRequired<pLock->eLock && eRequired<=SQLITE_LOCK_SHARED ){
        rc = pLock->pFile->pMethods->xUnlock(pLock->pFile, eRequired);
        if( rc==SQLITE_OK ){
          pLock->eLock = eRequired;
        }
      }
    }
  
    return rc;
  }
  
  /*
  ** Return the AsyncLock structure from the global async.pLock list 
  ** associated with the file-system entry identified by path zName 
  ** (a string of nName bytes). If no such structure exists, return 0.
  */
  static AsyncLock *findLock(const char *zName, int nName){
    AsyncLock *p = async.pLock;
    while( p && (p->nFile!=nName || memcmp(p->zFile, zName, nName)) ){
      p = p->pNext;
    }
    return p;
  }
  
  /*
  ** The following two methods - asyncLock() and asyncUnlock() - are used
  ** to obtain and release locks on database files opened with the
  ** asynchronous backend.
  */
  static int asyncLock(sqlite3_file *pFile, int eLock){
    int rc = SQLITE_OK;
    AsyncFileData *p = ((AsyncFile *)pFile)->pData;
  
    if( p->zName ){
      pthread_mutex_lock(&async.lockMutex);
      if( p->lock.eLock<eLock ){
        AsyncLock *pLock = p->pLock;
        AsyncFileLock *pIter;
        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;
          p->lock.eAsyncLock = MAX(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){
    int rc = SQLITE_OK;
    AsyncFileData *p = ((AsyncFile *)pFile)->pData;
    if( p->zName ){
      AsyncFileLock *pLock = &p->lock;
      pthread_mutex_lock(&async.lockMutex);
      pLock->eLock = MIN(pLock->eLock, eLock);
      pthread_mutex_unlock(&async.lockMutex);
      rc = addNewAsyncWrite(p, ASYNC_UNLOCK, 0, eLock, 0);
    }
    return rc;
  }
  
  /*
  ** 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 *pResOut){
    int ret = 0;
    AsyncFileLock *pIter;
    AsyncFileData *p = ((AsyncFile *)pFile)->pData;
  
    pthread_mutex_lock(&async.lockMutex);
    for(pIter=p->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));
    *pResOut = ret;
    return SQLITE_OK;
  }
  
  /* 
  ** sqlite3_file_control() implementation.
  */
  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.
  */
  static int asyncSectorSize(sqlite3_file *pFile){
    return 512;
  }
  static int asyncDeviceCharacteristics(sqlite3_file *pFile){
    return 0;
  }
  
  static int unlinkAsyncFile(AsyncFileData *pData){
    AsyncFileLock **ppIter;
    int rc = SQLITE_OK;
  
    if( pData->zName ){
      AsyncLock *pLock = pData->pLock;
      for(ppIter=&pLock->pList; *ppIter; ppIter=&((*ppIter)->pNext)){
        if( (*ppIter)==&pData->lock ){
          *ppIter = pData->lock.pNext;
          break;
        }
      }
      if( !pLock->pList ){
        AsyncLock **pp;
        if( pLock->pFile ){
          pLock->pFile->pMethods->xClose(pLock->pFile);
        }
        for(pp=&async.pLock; *pp!=pLock; pp=&((*pp)->pNext));
        *pp = pLock->pNext;
        sqlite3_free(pLock);
      }else{
        rc = getFileLock(pLock);
      }
    }
  
    return rc;
  }
  
  /*
  ** Open a file.
  */
  static int asyncOpen(
    sqlite3_vfs *pAsyncVfs,
    const char *zName,
    sqlite3_file *pFile,
    int flags,
    int *pOutFlags
  ){
    static sqlite3_io_methods async_methods = {
      1,                               /* iVersion */
      asyncClose,                      /* xClose */
      asyncRead,                       /* xRead */
      asyncWrite,                      /* xWrite */
      asyncTruncate,                   /* xTruncate */
      asyncSync,                       /* xSync */
      asyncFileSize,                   /* xFileSize */
      asyncLock,                       /* xLock */
      asyncUnlock,                     /* xUnlock */
      asyncCheckReservedLock,          /* xCheckReservedLock */
      asyncFileControl,                /* xFileControl */
      asyncSectorSize,                 /* xSectorSize */
      asyncDeviceCharacteristics       /* xDeviceCharacteristics */
    };
  
    sqlite3_vfs *pVfs = (sqlite3_vfs *)pAsyncVfs->pAppData;
    AsyncFile *p = (AsyncFile *)pFile;
    int nName = 0;
    int rc = SQLITE_OK;
    int nByte;
    AsyncFileData *pData;
    AsyncLock *pLock = 0;
    char *z;
    int isExclusive = (flags&SQLITE_OPEN_EXCLUSIVE);
  
    /* If zName is NULL, then the upper layer is requesting an anonymous file */
    if( zName ){
      nName = strlen(zName)+1;
    }
  
    nByte = (
      sizeof(AsyncFileData) +        /* AsyncFileData structure */
      2 * pVfs->szOsFile +           /* AsyncFileData.pBaseRead and pBaseWrite */
      nName                          /* AsyncFileData.zName */
    ); 
    z = sqlite3_malloc(nByte);
    if( !z ){
      return SQLITE_NOMEM;
    }
    memset(z, 0, nByte);
    pData = (AsyncFileData*)z;
    z += sizeof(pData[0]);
    pData->pBaseRead = (sqlite3_file*)z;
    z += pVfs->szOsFile;
    pData->pBaseWrite = (sqlite3_file*)z;
    pData->close.pFileData = pData;
    pData->close.op = ASYNC_CLOSE;
  
    if( zName ){
      z += pVfs->szOsFile;
      pData->zName = z;
      pData->nName = nName;
      memcpy(pData->zName, zName, nName);
    }
  
    if( !isExclusive ){
      rc = pVfs->xOpen(pVfs, zName, pData->pBaseRead, flags, pOutFlags);
      if( rc==SQLITE_OK && ((*pOutFlags)&SQLITE_OPEN_READWRITE) ){
        rc = pVfs->xOpen(pVfs, zName, pData->pBaseWrite, flags, 0);
      }
    }
  
    pthread_mutex_lock(&async.lockMutex);
  
    if( zName && rc==SQLITE_OK ){
      pLock = findLock(pData->zName, pData->nName);
      if( !pLock ){
        int nByte = pVfs->szOsFile + sizeof(AsyncLock) + pData->nName + 1; 
        pLock = (AsyncLock *)sqlite3_malloc(nByte);
        if( pLock ){
          memset(pLock, 0, nByte);
  #ifdef ENABLE_FILE_LOCKING
          if( flags&SQLITE_OPEN_MAIN_DB ){
            pLock->pFile = (sqlite3_file *)&pLock[1];
            rc = pVfs->xOpen(pVfs, zName, pLock->pFile, flags, 0);
            if( rc!=SQLITE_OK ){
              sqlite3_free(pLock);
              pLock = 0;
            }
          }
  #endif
          if( pLock ){
            pLock->nFile = pData->nName;
            pLock->zFile = &((char *)(&pLock[1]))[pVfs->szOsFile];
            memcpy(pLock->zFile, pData->zName, pLock->nFile);
            pLock->pNext = async.pLock;
            async.pLock = pLock;
          }
        }else{
          rc = SQLITE_NOMEM;
        }
      }
    }
  
    if( rc==SQLITE_OK ){
      p->pMethod = &async_methods;
      p->pData = pData;
  
      /* Link AsyncFileData.lock into the linked list of 
      ** AsyncFileLock structures for this file.
      */
      if( zName ){
        pData->lock.pNext = pLock->pList;
        pLock->pList = &pData->lock;
        pData->zName = pLock->zFile;
      }
    }else{
      if( pData->pBaseRead->pMethods ){
        pData->pBaseRead->pMethods->xClose(pData->pBaseRead);
      }
      if( pData->pBaseWrite->pMethods ){
        pData->pBaseWrite->pMethods->xClose(pData->pBaseWrite);
      }
      sqlite3_free(pData);
    }
  
    pthread_mutex_unlock(&async.lockMutex);
  
    if( rc==SQLITE_OK ){
      incrOpenFileCount();
      pData->pLock = pLock;
    }
  
    if( rc==SQLITE_OK && isExclusive ){
      rc = addNewAsyncWrite(pData, ASYNC_OPENEXCLUSIVE, (sqlite3_int64)flags,0,0);
      if( rc==SQLITE_OK ){
        if( pOutFlags ) *pOutFlags = flags;
      }else{
        pthread_mutex_lock(&async.lockMutex);
        unlinkAsyncFile(pData);
        pthread_mutex_unlock(&async.lockMutex);
        sqlite3_free(pData);
      }
    }
    return rc;
  }
  
  /*
  ** Implementation of sqlite3OsDelete. Add an entry to the end of the 
  ** write-op queue to perform the delete.
  */
  static int asyncDelete(sqlite3_vfs *pAsyncVfs, const char *z, int syncDir){
    return addNewAsyncWrite(0, ASYNC_DELETE, syncDir, strlen(z)+1, z);
  }
  
  /*
  ** Implementation of sqlite3OsAccess. This method holds the mutex from
  ** start to finish.
  */
  static int asyncAccess(
    sqlite3_vfs *pAsyncVfs, 
    const char *zName, 
    int flags,
    int *pResOut
  ){
    int rc;
    int ret;
    AsyncWrite *p;
    sqlite3_vfs *pVfs = (sqlite3_vfs *)pAsyncVfs->pAppData;
  
    assert(flags==SQLITE_ACCESS_READWRITE 
        || flags==SQLITE_ACCESS_READ 
        || flags==SQLITE_ACCESS_EXISTS 
    );
  
    pthread_mutex_lock(&async.queueMutex);
    rc = pVfs->xAccess(pVfs, zName, flags, &ret);
    if( rc==SQLITE_OK && flags==SQLITE_ACCESS_EXISTS ){
      for(p=async.pQueueFirst; p; p = p->pNext){
        if( p->op==ASYNC_DELETE && 0==strcmp(p->zBuf, zName) ){
          ret = 0;
        }else if( p->op==ASYNC_OPENEXCLUSIVE 
               && p->pFileData->zName
               && 0==strcmp(p->pFileData->zName, zName) 
        ){
          ret = 1;
        }
      }
    }
    ASYNC_TRACE(("ACCESS(%s): %s = %d\n", 
      flags==SQLITE_ACCESS_READWRITE?"read-write":
      flags==SQLITE_ACCESS_READ?"read":"exists"
      , zName, ret)
    );
    pthread_mutex_unlock(&async.queueMutex);
    *pResOut = ret;
    return rc;
  }
  
  /*
  ** Fill in zPathOut with the full path to the file identified by zPath.
  */
  static int asyncFullPathname(
    sqlite3_vfs *pAsyncVfs, 
    const char *zPath, 
    int nPathOut,
    char *zPathOut
  ){
    int rc;
    sqlite3_vfs *pVfs = (sqlite3_vfs *)pAsyncVfs->pAppData;
    rc = pVfs->xFullPathname(pVfs, zPath, nPathOut, zPathOut);
  
    /* Because of the way intra-process file locking works, this backend
    ** needs to return a canonical path. The following block assumes the
    ** file-system uses unix style paths. 
    */
    if( rc==SQLITE_OK ){
      int iIn;
      int iOut = 0;
      int nPathOut = strlen(zPathOut);
  
      for(iIn=0; iIn<nPathOut; iIn++){
  
        /* Replace any occurences of "//" with "/" */
        if( iIn<=(nPathOut-2) && zPathOut[iIn]=='/' && zPathOut[iIn+1]=='/'
        ){
          continue;
        }
  
        /* Replace any occurences of "/./" with "/" */
        if( iIn<=(nPathOut-3) 
         && zPathOut[iIn]=='/' && zPathOut[iIn+1]=='.' && zPathOut[iIn+2]=='/'
        ){
          iIn++;
          continue;
        }
  
        /* Replace any occurences of "<path-component>/../" with "" */
        if( iOut>0 && iIn<=(nPathOut-4) 
         && zPathOut[iIn]=='/' && zPathOut[iIn+1]=='.' 
         && zPathOut[iIn+2]=='.' && zPathOut[iIn+3]=='/'
        ){
          iIn += 3;
          iOut--;
          for( ; iOut>0 && zPathOut[iOut-1]!='/'; iOut--);
          continue;
        }
  
        zPathOut[iOut++] = zPathOut[iIn];
      }
      zPathOut[iOut] = '\0';
    }
  
    return rc;
  }
  static void *asyncDlOpen(sqlite3_vfs *pAsyncVfs, const char *zPath){
    sqlite3_vfs *pVfs = (sqlite3_vfs *)pAsyncVfs->pAppData;
    return pVfs->xDlOpen(pVfs, zPath);
  }
  static void asyncDlError(sqlite3_vfs *pAsyncVfs, int nByte, char *zErrMsg){
    sqlite3_vfs *pVfs = (sqlite3_vfs *)pAsyncVfs->pAppData;
    pVfs->xDlError(pVfs, nByte, zErrMsg);
  }
  static void *asyncDlSym(
    sqlite3_vfs *pAsyncVfs, 
    void *pHandle, 
    const char *zSymbol
  ){
    sqlite3_vfs *pVfs = (sqlite3_vfs *)pAsyncVfs->pAppData;
    return pVfs->xDlSym(pVfs, pHandle, zSymbol);
  }
  static void asyncDlClose(sqlite3_vfs *pAsyncVfs, void *pHandle){
    sqlite3_vfs *pVfs = (sqlite3_vfs *)pAsyncVfs->pAppData;
    pVfs->xDlClose(pVfs, pHandle);
  }
  static int asyncRandomness(sqlite3_vfs *pAsyncVfs, int nByte, char *zBufOut){
    sqlite3_vfs *pVfs = (sqlite3_vfs *)pAsyncVfs->pAppData;
    return pVfs->xRandomness(pVfs, nByte, zBufOut);
  }
  static int asyncSleep(sqlite3_vfs *pAsyncVfs, int nMicro){
    sqlite3_vfs *pVfs = (sqlite3_vfs *)pAsyncVfs->pAppData;
    return pVfs->xSleep(pVfs, nMicro);
  }
  static int asyncCurrentTime(sqlite3_vfs *pAsyncVfs, double *pTimeOut){
    sqlite3_vfs *pVfs = (sqlite3_vfs *)pAsyncVfs->pAppData;
    return pVfs->xCurrentTime(pVfs, pTimeOut);
  }
  
  static sqlite3_vfs async_vfs = {
    1,                    /* iVersion */
    sizeof(AsyncFile),    /* szOsFile */
    0,                    /* mxPathname */
    0,                    /* pNext */
    "async",              /* zName */
    0,                    /* pAppData */
    asyncOpen,            /* xOpen */
    asyncDelete,          /* xDelete */
    asyncAccess,          /* xAccess */
    asyncFullPathname,    /* xFullPathname */
    asyncDlOpen,          /* xDlOpen */
    asyncDlError,         /* xDlError */
    asyncDlSym,           /* xDlSym */
    asyncDlClose,         /* xDlClose */
    asyncRandomness,      /* xDlError */
    asyncSleep,           /* xDlSym */
    asyncCurrentTime      /* xDlClose */
  };
  
  /*
  ** Call this routine to enable or disable the
  ** asynchronous IO features implemented in this file. 
  **
  ** This routine is not even remotely threadsafe.  Do not call
  ** this routine while any SQLite database connections are open.
  */
  static void asyncEnable(int enable){
    if( enable ){
      if( !async_vfs.pAppData ){
        async_vfs.pAppData = (void *)sqlite3_vfs_find(0);
        async_vfs.mxPathname = ((sqlite3_vfs *)async_vfs.pAppData)->mxPathname;
        sqlite3_vfs_register(&async_vfs, 1);
      }
    }else{
      if( async_vfs.pAppData ){
        sqlite3_vfs_unregister(&async_vfs);
        async_vfs.pAppData = 0;
      }
    }
  }
  
  /* 
  ** This procedure runs in a separate thread, reading messages off of the
  ** write queue and processing them one by one.  
  **
  ** If async.writerHaltNow is true, then this procedure exits
  ** after processing a single message.
  **
  ** If async.writerHaltWhenIdle is true, then this procedure exits when
  ** the write queue is empty.
  **
  ** If both of the above variables are false, this procedure runs
  ** indefinately, waiting for operations to be added to the write queue
  ** and processing them in the order in which they arrive.
  **
  ** An artifical delay of async.ioDelay milliseconds is inserted before
  ** each write operation in order to simulate the effect of a slow disk.
  **
  ** Only one instance of this procedure may be running at a time.
  */
  static void *asyncWriterThread(void *pIsStarted){
    sqlite3_vfs *pVfs = (sqlite3_vfs *)(async_vfs.pAppData);
    AsyncWrite *p = 0;
    int rc = SQLITE_OK;
    int holdingMutex = 0;
  
    if( pthread_mutex_trylock(&async.writerMutex) ){
      return 0;
    }
    (*(int *)pIsStarted) = 1;
    while( async.writerHaltNow==0 ){
      int doNotFree = 0;
      sqlite3_file *pBase = 0;
  
      if( !holdingMutex ){
        pthread_mutex_lock(&async.queueMutex);
      }
      while( (p = async.pQueueFirst)==0 ){
        pthread_cond_broadcast(&async.emptySignal);
        if( async.writerHaltWhenIdle ){
          pthread_mutex_unlock(&async.queueMutex);
          break;
        }else{
          ASYNC_TRACE(("IDLE\n"));
          pthread_cond_wait(&async.queueSignal, &async.queueMutex);
          ASYNC_TRACE(("WAKEUP\n"));
        }
      }
      if( p==0 ) break;
      holdingMutex = 1;
  
      /* Right now this thread is holding the mutex on the write-op queue.
      ** Variable 'p' points to the first entry in the write-op queue. In
      ** the general case, we hold on to the mutex for the entire body of
      ** the loop. 
      **
      ** However in the cases enumerated below, we relinquish the mutex,
      ** perform the IO, and then re-request the mutex before removing 'p' from
      ** the head of the write-op queue. The idea is to increase concurrency with
      ** sqlite threads.
      **
      **     * An ASYNC_CLOSE operation.
      **     * An ASYNC_OPENEXCLUSIVE operation. For this one, we relinquish 
      **       the mutex, call the underlying xOpenExclusive() function, then
      **       re-aquire the mutex before seting the AsyncFile.pBaseRead 
      **       variable.
      **     * ASYNC_SYNC and ASYNC_WRITE operations, if 
      **       SQLITE_ASYNC_TWO_FILEHANDLES was set at compile time and two
      **       file-handles are open for the particular file being "synced".
      */
      if( async.ioError!=SQLITE_OK && p->op!=ASYNC_CLOSE ){
        p->op = ASYNC_NOOP;
      }
      if( p->pFileData ){
        pBase = p->pFileData->pBaseWrite;
        if( 
          p->op==ASYNC_CLOSE || 
          p->op==ASYNC_OPENEXCLUSIVE ||
          (pBase->pMethods && (p->op==ASYNC_SYNC || p->op==ASYNC_WRITE) ) 
        ){
          pthread_mutex_unlock(&async.queueMutex);
          holdingMutex = 0;
        }
        if( !pBase->pMethods ){
          pBase = p->pFileData->pBaseRead;
        }
      }
  
      switch( p->op ){
        case ASYNC_NOOP:
          break;
  
        case ASYNC_WRITE:
          assert( pBase );
          ASYNC_TRACE(("WRITE %s %d bytes at %d\n",
                  p->pFileData->zName, p->nByte, p->iOffset));
          rc = pBase->pMethods->xWrite(pBase, (void *)(p->zBuf), p->nByte, p->iOffset);
          break;
  
        case ASYNC_SYNC:
          assert( pBase );
          ASYNC_TRACE(("SYNC %s\n", p->pFileData->zName));
          rc = pBase->pMethods->xSync(pBase, p->nByte);
          break;
  
        case ASYNC_TRUNCATE:
          assert( pBase );
          ASYNC_TRACE(("TRUNCATE %s to %d bytes\n", 
                  p->pFileData->zName, p->iOffset));
          rc = pBase->pMethods->xTruncate(pBase, p->iOffset);
          break;
  
        case ASYNC_CLOSE: {
          AsyncFileData *pData = p->pFileData;
          ASYNC_TRACE(("CLOSE %s\n", p->pFileData->zName));
          if( pData->pBaseWrite->pMethods ){
            pData->pBaseWrite->pMethods->xClose(pData->pBaseWrite);
          }
          if( pData->pBaseRead->pMethods ){
            pData->pBaseRead->pMethods->xClose(pData->pBaseRead);
          }
  
          /* 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);
          rc = unlinkAsyncFile(pData);
          pthread_mutex_unlock(&async.lockMutex);
  
          if( !holdingMutex ){
            pthread_mutex_lock(&async.queueMutex);
            holdingMutex = 1;
          }
          assert_mutex_is_held(&async.queueMutex);
          async.pQueueFirst = p->pNext;
          sqlite3_free(pData);
          doNotFree = 1;
          break;
        }
  
        case ASYNC_UNLOCK: {
          AsyncFileData *pData = p->pFileData;
          int eLock = p->nByte;
          pthread_mutex_lock(&async.lockMutex);
          pData->lock.eAsyncLock = MIN(
              pData->lock.eAsyncLock, MAX(pData->lock.eLock, eLock)
          );
          assert(pData->lock.eAsyncLock>=pData->lock.eLock);
          rc = getFileLock(pData->pLock);
          pthread_mutex_unlock(&async.lockMutex);
          break;
        }
  
        case ASYNC_DELETE:
          ASYNC_TRACE(("DELETE %s\n", p->zBuf));
          rc = pVfs->xDelete(pVfs, p->zBuf, (int)p->iOffset);
          break;
  
        case ASYNC_OPENEXCLUSIVE: {
          int flags = (int)p->iOffset;
          AsyncFileData *pData = p->pFileData;
          ASYNC_TRACE(("OPEN %s flags=%d\n", p->zBuf, (int)p->iOffset));
          assert(pData->pBaseRead->pMethods==0 && pData->pBaseWrite->pMethods==0);
          rc = pVfs->xOpen(pVfs, pData->zName, pData->pBaseRead, flags, 0);
          assert( holdingMutex==0 );
          pthread_mutex_lock(&async.queueMutex);
          holdingMutex = 1;
          break;
        }
  
        default: assert(!"Illegal value for AsyncWrite.op");
      }
  
      /* If we didn't hang on to the mutex during the IO op, obtain it now
      ** so that the AsyncWrite structure can be safely removed from the 
      ** global write-op queue.
      */
      if( !holdingMutex ){
        pthread_mutex_lock(&async.queueMutex);
        holdingMutex = 1;
      }
      /* ASYNC_TRACE(("UNLINK %p\n", p)); */
      if( p==async.pQueueLast ){
        async.pQueueLast = 0;
      }
      if( !doNotFree ){
        assert_mutex_is_held(&async.queueMutex);
        async.pQueueFirst = p->pNext;
        sqlite3_free(p);
      }
      assert( holdingMutex );
  
      /* An IO error has occured. We cannot report the error back to the
      ** connection that requested the I/O since the error happened 
      ** asynchronously.  The connection has already moved on.  There 
      ** really is nobody to report the error to.
      **
      ** The file for which the error occured may have been a database or
      ** journal file. Regardless, none of the currently queued operations
      ** associated with the same database should now be performed. Nor should
      ** any subsequently requested IO on either a database or journal file 
      ** handle for the same database be accepted until the main database
      ** file handle has been closed and reopened.
      **
      ** Furthermore, no further IO should be queued or performed on any file
      ** handle associated with a database that may have been part of a 
      ** multi-file transaction that included the database associated with 
      ** the IO error (i.e. a database ATTACHed to the same handle at some 
      ** point in time).
      */
      if( rc!=SQLITE_OK ){
        async.ioError = rc;
      }
  
      if( async.ioError && !async.pQueueFirst ){
        pthread_mutex_lock(&async.lockMutex);
        if( 0==async.pLock ){
          async.ioError = SQLITE_OK;
        }
        pthread_mutex_unlock(&async.lockMutex);
      }
  
      /* Drop the queue mutex before continuing to the next write operation
      ** in order to give other threads a chance to work with the write queue.
      */
      if( !async.pQueueFirst || !async.ioError ){
        pthread_mutex_unlock(&async.queueMutex);
        holdingMutex = 0;
        if( async.ioDelay>0 ){
          pVfs->xSleep(pVfs, async.ioDelay);
        }else{
          sched_yield();
        }
      }
    }
    
    pthread_mutex_unlock(&async.writerMutex);
    return 0;
  }
  
  /**************************************************************************
  ** The remaining code defines a Tcl interface for testing the asynchronous
  ** IO implementation in this file.
  **
  ** To adapt the code to a non-TCL environment, delete or comment out
  ** the code that follows.
  */
  
  /*
  ** sqlite3async_enable ?YES/NO?
  **
  ** Enable or disable the asynchronous I/O backend.  This command is
  ** not thread-safe.  Do not call it while any database connections
  ** are open.
  */
  static int testAsyncEnable(
    void * clientData,
    Tcl_Interp *interp,
    int objc,
    Tcl_Obj *CONST objv[]
  ){
    if( objc!=1 && objc!=2 ){
      Tcl_WrongNumArgs(interp, 1, objv, "?YES/NO?");
      return TCL_ERROR;
    }
    if( objc==1 ){
      Tcl_SetObjResult(interp, Tcl_NewBooleanObj(async_vfs.pAppData!=0));
    }else{
      int en;
      if( Tcl_GetBooleanFromObj(interp, objv[1], &en) ) return TCL_ERROR;
      asyncEnable(en);
    }
    return TCL_OK;
  }
  
  /*
  ** sqlite3async_halt  "now"|"idle"|"never"
  **
  ** Set the conditions at which the writer thread will halt.
  */
  static int testAsyncHalt(
    void * clientData,
    Tcl_Interp *interp,
    int objc,
    Tcl_Obj *CONST objv[]
  ){
    const char *zCond;
    if( objc!=2 ){
      Tcl_WrongNumArgs(interp, 1, objv, "\"now\"|\"idle\"|\"never\"");
      return TCL_ERROR;
    }
    zCond = Tcl_GetString(objv[1]);
    if( strcmp(zCond, "now")==0 ){
      async.writerHaltNow = 1;
      pthread_cond_broadcast(&async.queueSignal);
    }else if( strcmp(zCond, "idle")==0 ){
      async.writerHaltWhenIdle = 1;
      async.writerHaltNow = 0;
      pthread_cond_broadcast(&async.queueSignal);
    }else if( strcmp(zCond, "never")==0 ){
      async.writerHaltWhenIdle = 0;
      async.writerHaltNow = 0;
    }else{
      Tcl_AppendResult(interp, 
        "should be one of: \"now\", \"idle\", or \"never\"", (char*)0);
      return TCL_ERROR;
    }
    return TCL_OK;
  }
  
  /*
  ** sqlite3async_delay ?MS?
  **
  ** Query or set the number of milliseconds of delay in the writer
  ** thread after each write operation.  The default is 0.  By increasing
  ** the memory delay we can simulate the effect of slow disk I/O.
  */
  static int testAsyncDelay(
    void * clientData,
    Tcl_Interp *interp,
    int objc,
    Tcl_Obj *CONST objv[]
  ){
    if( objc!=1 && objc!=2 ){
      Tcl_WrongNumArgs(interp, 1, objv, "?MS?");
      return TCL_ERROR;
    }
    if( objc==1 ){
      Tcl_SetObjResult(interp, Tcl_NewIntObj(async.ioDelay));
    }else{
      int ioDelay;
      if( Tcl_GetIntFromObj(interp, objv[1], &ioDelay) ) return TCL_ERROR;
      async.ioDelay = ioDelay;
    }
    return TCL_OK;
  }
  
  /*
  ** sqlite3async_start
  **
  ** Start a new writer thread.
  */
  static int testAsyncStart(
    void * clientData,
    Tcl_Interp *interp,
    int objc,
    Tcl_Obj *CONST objv[]
  ){
    pthread_t x;
    int rc;
    volatile int isStarted = 0;
    rc = pthread_create(&x, 0, asyncWriterThread, (void *)&isStarted);
    if( rc ){
      Tcl_AppendResult(interp, "failed to create the thread", 0);
      return TCL_ERROR;
    }
    pthread_detach(x);
    while( isStarted==0 ){
      sched_yield();
    }
    return TCL_OK;
  }
  
  /*
  ** sqlite3async_wait
  **
  ** Wait for the current writer thread to terminate.
  **
  ** If the current writer thread is set to run forever then this
  ** command would block forever.  To prevent that, an error is returned. 
  */
  static int testAsyncWait(
    void * clientData,
    Tcl_Interp *interp,
    int objc,
    Tcl_Obj *CONST objv[]
  ){
    int cnt = 10;
    if( async.writerHaltNow==0 && async.writerHaltWhenIdle==0 ){
      Tcl_AppendResult(interp, "would block forever", (char*)0);
      return TCL_ERROR;
    }
  
    while( cnt-- && !pthread_mutex_trylock(&async.writerMutex) ){
      pthread_mutex_unlock(&async.writerMutex);
      sched_yield();
    }
    if( cnt>=0 ){
      ASYNC_TRACE(("WAIT\n"));
      pthread_mutex_lock(&async.queueMutex);
      pthread_cond_broadcast(&async.queueSignal);
      pthread_mutex_unlock(&async.queueMutex);
      pthread_mutex_lock(&async.writerMutex);
      pthread_mutex_unlock(&async.writerMutex);
    }else{
      ASYNC_TRACE(("NO-WAIT\n"));
    }
    return TCL_OK;
  }
  
  
  #endif  /* SQLITE_OS_UNIX and SQLITE_THREADSAFE */
  
  /*
  ** This routine registers the custom TCL commands defined in this
  ** module.  This should be the only procedure visible from outside
  ** of this module.
  */
  int Sqlitetestasync_Init(Tcl_Interp *interp){
  #if SQLITE_OS_UNIX && SQLITE_THREADSAFE
    Tcl_CreateObjCommand(interp,"sqlite3async_enable",testAsyncEnable,0,0);
    Tcl_CreateObjCommand(interp,"sqlite3async_halt",testAsyncHalt,0,0);
    Tcl_CreateObjCommand(interp,"sqlite3async_delay",testAsyncDelay,0,0);
    Tcl_CreateObjCommand(interp,"sqlite3async_start",testAsyncStart,0,0);
    Tcl_CreateObjCommand(interp,"sqlite3async_wait",testAsyncWait,0,0);
    Tcl_LinkVar(interp, "sqlite3async_trace",
        (char*)&sqlite3async_trace, TCL_LINK_INT);
  #endif  /* SQLITE_OS_UNIX and SQLITE_THREADSAFE */
    return TCL_OK;
  }
  