test_async.c

sqlite嵌入式数据库源码

  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(OsFile **pId){
  return addNewAsyncWrite((AsyncFile *)*pId, ASYNC_CLOSE, 0, 0, 0);
}

/*
** 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(OsFile *id, const void *pBuf, int amt){
  AsyncFile *pFile = (AsyncFile *)id;
  int rc = addNewAsyncWrite(pFile, ASYNC_WRITE, pFile->iOffset, amt, pBuf);
  pFile->iOffset += (i64)amt;
  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(OsFile *id, i64 nByte){
  return addNewAsyncWrite((AsyncFile *)id, ASYNC_TRUNCATE, nByte, 0, 0);
}

/*
** Open the directory identified by zName and associate it with the 
** specified file. This just adds an entry to the write-op list, the 
** directory is opened later by sqlite3_async_flush().
*/
static int asyncOpenDirectory(OsFile *id, const char *zName){
  AsyncFile *pFile = (AsyncFile *)id;
  return addNewAsyncWrite(pFile, ASYNC_OPENDIRECTORY, 0, strlen(zName)+1,zName);
}

/*
** 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(OsFile *id, int fullsync){
  return addNewAsyncWrite((AsyncFile *)id, ASYNC_SYNC, 0, fullsync, 0);
}

/*
** Set (or clear) the full-sync flag on the underlying file. This operation
** is queued and performed later by sqlite3_async_flush().
*/
static void asyncSetFullSync(OsFile *id, int value){
  addNewAsyncWrite((AsyncFile *)id, ASYNC_SETFULLSYNC, 0, value, 0);
}

/*
** 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(OsFile *id, void *obuf, int amt){
  int rc = SQLITE_OK;
  i64 filesize;
  int nRead;
  AsyncFile *pFile = (AsyncFile *)id;
  OsFile *pBase = pFile->pBaseRead;

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

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

  if( pBase ){
    rc = sqlite3OsFileSize(pBase, &filesize);
    if( rc!=SQLITE_OK ){
      goto asyncread_out;
    }
    rc = sqlite3OsSeek(pBase, pFile->iOffset);
    if( rc!=SQLITE_OK ){
      goto asyncread_out;
    }
    nRead = MIN(filesize - pFile->iOffset, amt);
    if( nRead>0 ){
      rc = sqlite3OsRead(pBase, obuf, nRead);
      TRACE(("READ %s %d bytes at %d\n", pFile->zName, nRead, pFile->iOffset));
    }
  }

  if( rc==SQLITE_OK ){
    AsyncWrite *p;
    i64 iOffset = pFile->iOffset;           /* Current seek offset */

    for(p=async.pQueueFirst; p; p = p->pNext){
      if( p->pFile==pFile && p->op==ASYNC_WRITE ){
        int iBeginOut = (p->iOffset - iOffset);
        int iBeginIn = -iBeginOut;
        int nCopy;

        if( iBeginIn<0 ) iBeginIn = 0;
        if( iBeginOut<0 ) iBeginOut = 0;
        nCopy = MIN(p->nByte-iBeginIn, amt-iBeginOut);

        if( nCopy>0 ){
          memcpy(&((char *)obuf)[iBeginOut], &p->zBuf[iBeginIn], nCopy);
          TRACE(("OVERREAD %d bytes at %d\n", nCopy, iBeginOut+iOffset));
        }
      }
    }

    pFile->iOffset += (i64)amt;
  }

asyncread_out:
  pthread_mutex_unlock(&async.queueMutex);
  return rc;
}

/*
** Seek to the specified offset. This just adjusts the AsyncFile.iOffset 
** variable - calling seek() on the underlying file is defered until the 
** next read() or write() operation. 
*/
static int asyncSeek(OsFile *id, i64 offset){
  AsyncFile *pFile = (AsyncFile *)id;
  pFile->iOffset = offset;
  return SQLITE_OK;
}

/*
** 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(OsFile *id, i64 *pSize){
  int rc = SQLITE_OK;
  i64 s = 0;
  OsFile *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 = ((AsyncFile *)id)->pBaseRead;
  if( pBase ){
    rc = sqlite3OsFileSize(pBase, &s);
  }

  if( rc==SQLITE_OK ){
    AsyncWrite *p;
    for(p=async.pQueueFirst; p; p = p->pNext){
      if( p->pFile==(AsyncFile *)id ){
        switch( p->op ){
          case ASYNC_WRITE:
            s = MAX(p->iOffset + (i64)(p->nByte), s);
            break;
          case ASYNC_TRUNCATE:
            s = MIN(s, p->iOffset);
            break;
        }
      }
    }
    *pSize = s;
  }
  pthread_mutex_unlock(&async.queueMutex);
  return rc;
}

/*
** Return the operating system file handle. This is only used for debugging 
** at the moment anyway.
*/
static int asyncFileHandle(OsFile *id){
  return sqlite3OsFileHandle(((AsyncFile *)id)->pBaseRead);
}

/*
** No disk locking is performed.  We keep track of locks locally in
** the async.aLock hash table.  Locking should appear to work the same
** as with standard (unmodified) SQLite as long as all connections 
** come from this one process.  Connections from external processes
** cannot see our internal hash table (obviously) and will thus not
** honor our locks.
*/
static int asyncLock(OsFile *id, int lockType){
  AsyncFile *pFile = (AsyncFile*)id;
  TRACE(("LOCK %d (%s)\n", lockType, pFile->zName));
  pthread_mutex_lock(&async.lockMutex);
  sqlite3HashInsert(&async.aLock, pFile->zName, pFile->nName, (void*)lockType);
  pthread_mutex_unlock(&async.lockMutex);
  return SQLITE_OK;
}
static int asyncUnlock(OsFile *id, int lockType){
  return asyncLock(id, lockType);
}

/*
** This function is called when the pager layer first opens a database file
** and is checking for a hot-journal.
*/
static int asyncCheckReservedLock(OsFile *id){
  AsyncFile *pFile = (AsyncFile*)id;
  int rc;
  pthread_mutex_lock(&async.lockMutex);
  rc = (int)sqlite3HashFind(&async.aLock, pFile->zName, pFile->nName);
  pthread_mutex_unlock(&async.lockMutex);
  TRACE(("CHECK-LOCK %d (%s)\n", rc, pFile->zName));
  return rc>SHARED_LOCK;
}

/* 
** This is broken. But sqlite3OsLockState() is only used for testing anyway.
*/
static int asyncLockState(OsFile *id){
  return SQLITE_OK;
}

/*
** The following variables hold pointers to the original versions of
** OS-layer interface routines that are overloaded in order to create
** the asynchronous I/O backend.
*/
static int (*xOrigOpenReadWrite)(const char*, OsFile**, int*) = 0;
static int (*xOrigOpenExclusive)(const char*, OsFile**, int) = 0;
static int (*xOrigOpenReadOnly)(const char*, OsFile**) = 0;
static int (*xOrigDelete)(const char*) = 0;
static int (*xOrigFileExists)(const char*) = 0;
static int (*xOrigSyncDirectory)(const char*) = 0;

/*
** This routine does most of the work of opening a file and building
** the OsFile structure.
*/
static int asyncOpenFile(
  const char *zName,     /* The name of the file to be opened */
  OsFile **pFile,        /* Put the OsFile structure here */
  OsFile *pBaseRead,     /* The real OsFile from the real I/O routine */
  int openForWriting     /* Open a second file handle for writing if true */
){
  int rc, i, n;
  AsyncFile *p;
  OsFile *pBaseWrite = 0;

  static IoMethod iomethod = {
    asyncClose,
    asyncOpenDirectory,
    asyncRead,
    asyncWrite,
    asyncSeek,
    asyncTruncate,
    asyncSync,
    asyncSetFullSync,
    asyncFileHandle,
    asyncFileSize,
    asyncLock,
    asyncUnlock,
    asyncLockState,
    asyncCheckReservedLock
  };

  if( openForWriting && SQLITE_ASYNC_TWO_FILEHANDLES ){
    int dummy;
    rc = xOrigOpenReadWrite(zName, &pBaseWrite, &dummy);
    if( rc!=SQLITE_OK ){
      goto error_out;
    }
  }

  n = strlen(zName);
  for(i=n-1; i>=0 && zName[i]!='/'; i--){}
  p = (AsyncFile *)sqlite3OsMalloc(sizeof(AsyncFile) + n - i);
  if( !p ){
    rc = SQLITE_NOMEM;
    goto error_out;
  }
  memset(p, 0, sizeof(AsyncFile));
  p->zName = (char*)&p[1];
  strcpy(p->zName, &zName[i+1]);
  p->nName = n - i;
  p->pMethod = &iomethod;
  p->pBaseRead = pBaseRead;
  p->pBaseWrite = pBaseWrite;
  
  *pFile = (OsFile *)p;
  return SQLITE_OK;

error_out:
  assert(!p);
  sqlite3OsClose(&pBaseRead);
  sqlite3OsClose(&pBaseWrite);
  *pFile = 0;
  return rc;
}

/*
** The async-IO backends implementation of the three functions used to open
** a file (xOpenExclusive, xOpenReadWrite and xOpenReadOnly). Most of the 
** work is done in function asyncOpenFile() - see above.
*/
static int asyncOpenExclusive(const char *z, OsFile **ppFile, int delFlag){
  int rc = asyncOpenFile(z, ppFile, 0, 0);
  if( rc==SQLITE_OK ){
    AsyncFile *pFile = (AsyncFile *)(*ppFile);
    int nByte = strlen(z)+1;
    i64 i = (i64)(delFlag);
    rc = addNewAsyncWrite(pFile, ASYNC_OPENEXCLUSIVE, i, nByte, z);
    if( rc!=SQLITE_OK ){
      sqlite3OsFree(pFile);
      *ppFile = 0;
    }
  }
  if( rc==SQLITE_OK ){
    incrOpenFileCount();
  }
  return rc;
}
static int asyncOpenReadOnly(const char *z, OsFile **ppFile){
  OsFile *pBase = 0;
  int rc = xOrigOpenReadOnly(z, &pBase);
  if( rc==SQLITE_OK ){
    rc = asyncOpenFile(z, ppFile, pBase, 0);
  }
  if( rc==SQLITE_OK ){
    incrOpenFileCount();
  }
  return rc;
}
static int asyncOpenReadWrite(const char *z, OsFile **ppFile, int *pReadOnly){
  OsFile *pBase = 0;
  int rc = xOrigOpenReadWrite(z, &pBase, pReadOnly);
  if( rc==SQLITE_OK ){
    rc = asyncOpenFile(z, ppFile, pBase, (*pReadOnly ? 0 : 1));
  }
  if( rc==SQLITE_OK ){
    incrOpenFileCount();
  }
  return rc;
}

/*
** Implementation of sqlite3OsDelete. Add an entry to the end of the 
** write-op queue to perform the delete.
*/
static int asyncDelete(const char *z){
  return addNewAsyncWrite(0, ASYNC_DELETE, 0, strlen(z)+1, z);
}

/*
** Implementation of sqlite3OsSyncDirectory. Add an entry to the end of the 
** write-op queue to perform the directory sync.
*/
static int asyncSyncDirectory(const char *z){
  return addNewAsyncWrite(0, ASYNC_SYNCDIRECTORY, 0, strlen(z)+1, z);
}

/*
** Implementation of sqlite3OsFileExists. Return true if file 'z' exists
** in the file system. 
**
** This method holds the mutex from start to finish.
*/
static int asyncFileExists(const char *z){
  int ret;
  AsyncWrite *p;

  pthread_mutex_lock(&async.queueMutex);

  /* See if the real file system contains the specified file.  */
  ret = xOrigFileExists(z);
  
  for(p=async.pQueueFirst; p; p = p->pNext){
    if( p->op==ASYNC_DELETE && 0==strcmp(p->zBuf, z) ){
      ret = 0;
    }else if( p->op==ASYNC_OPENEXCLUSIVE && 0==strcmp(p->zBuf, z) ){
      ret = 1;
    }
  }

  TRACE(("EXISTS: %s = %d\n", z, ret));
  pthread_mutex_unlock(&async.queueMutex);
  return ret;
}

/*
** 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 && xOrigOpenReadWrite==0 ){
    assert(sqlite3Os.xOpenReadWrite);
    sqlite3HashInit(&async.aLock, SQLITE_HASH_BINARY, 1);
    xOrigOpenReadWrite = sqlite3Os.xOpenReadWrite;
    xOrigOpenReadOnly = sqlite3Os.xOpenReadOnly;
    xOrigOpenExclusive = sqlite3Os.xOpenExclusive;
    xOrigDelete = sqlite3Os.xDelete;
    xOrigFileExists = sqlite3Os.xFileExists;
    xOrigSyncDirectory = sqlite3Os.xSyncDirectory;