/*

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

 }