SQLite

/*
** This file contains the implementation of a log file used in 
** "journal_mode=wal" mode.
*/

/*
** LOG FILE FORMAT
**
** A log file consists of a header followed by zero or more log frames.
** The log header is 12 bytes in size and consists of the following three
** big-endian 32-bit unsigned integer values:
**
**     0: Database page size,
**     4: Randomly selected salt value 1,
**     8: Randomly selected salt value 2.
**
** Immediately following the log header are zero or more log frames. Each
** frame itself consists of a 16-byte header followed by a <page-size> bytes
** of page data. The header is broken into 4 big-endian 32-bit unsigned 
** integer values, as follows:
**
**     0: Page number.
**     4: For commit records, the size of the database image in pages 
**        after the commit. For all other records, zero.
**     8: Checksum value 1.
**    12: Checksum value 2.
*/

/* 
** LOG SUMMARY FORMAT
**
** TODO.

  int nData;                      /* Size of aData allocation/mapping */
  u32 *aData;                     /* File body */
};


/*
** The four lockable regions associated with each log-summary. A connection
** may take either a SHARED or EXCLUSIVE lock on each. An ORed combination
** of the following bitmasks is passed as the second argument to the
** logLockRegion() function.
*/
#define LOG_REGION_A 0x01
#define LOG_REGION_B 0x02
#define LOG_REGION_C 0x04
#define LOG_REGION_D 0x08

#define LOG_LOCK_MUTEX  12
#define LOG_LOCK_DMH    13
#define LOG_LOCK_REGION 14

/*
** A single instance of this structure is allocated as part of each 
** connection to a database log. All structures associated with the 
** same log file are linked together into a list using LogLock.pNext
** starting at LogSummary.pLock.
**

** Unmap the log-summary mapping and close the file-descriptor. If
** the isTruncate argument is non-zero, truncate the log-summary file
** region to zero bytes.
**
** Regardless of the value of isTruncate, close the file-descriptor
** opened on the log-summary file.
*/
static int logSummaryUnmap(LogSummary *pSummary, int isUnlink){
  int rc = SQLITE_OK;
  if( pSummary->aData ){
    assert( pSummary->fd>0 );
    munmap(pSummary->aData, pSummary->nData);
    pSummary->aData = 0;
    if( isUnlink ){
      char *zFile = sqlite3_mprintf("%s-summary", pSummary->zPath);
      if( !zFile ){
        rc = SQLITE_NOMEM;
      }
      unlink(zFile);
      sqlite3_free(zFile);
    }
  }
  if( pSummary->fd>0 ){
    close(pSummary->fd);
    pSummary->fd = -1;
  }
  return rc;

  }

finished:
  logSummaryWriteHdr(pSummary, &hdr);
  return rc;
}

/*
** Values for the third parameter to logLockRegion().
*/
#define LOG_UNLOCK  0
#define LOG_RDLOCK  1
#define LOG_WRLOCK  2
#define LOG_WRLOCKW 3

static int logLockFd(LogSummary *pSummary, int iStart, int nByte, int op){
  int aType[4] = { 
    F_UNLCK,                    /* LOG_UNLOCK */
    F_RDLCK,                    /* LOG_RDLOCK */
    F_WRLCK,                    /* LOG_WRLOCK */
    F_WRLCK                     /* LOG_WRLOCKW */
  };
  int aOp[4] = { 
    F_SETLK,                    /* LOG_UNLOCK */
    F_SETLK,                    /* LOG_RDLOCK */
    F_SETLK,                    /* LOG_WRLOCK */
    F_SETLKW                    /* LOG_WRLOCKW */
  };

  struct flock f;               /* Locking operation */
  int rc;                       /* Value returned by fcntl() */

  assert( ArraySize(aType)==ArraySize(aOp) );
  assert( op>=0 && op<ArraySize(aType) );

  memset(&f, 0, sizeof(f));
  f.l_type = aType[op];
  f.l_whence = SEEK_SET;
  f.l_start = iStart;
  f.l_len = nByte;
  rc = fcntl(pSummary->fd, aOp[op], &f);
  return (rc==0) ? SQLITE_OK : SQLITE_BUSY;
}

static int logLockRegion(Log *pLog, u32 mRegion, int op){
  LogSummary *pSummary = pLog->pSummary;
  LogLock *p;                     /* Used to iterate through in-process locks */
  u32 mOther;                     /* Locks held by other connections */
  u32 mNew;                       /* New mask for pLog */

  assert( 
       /* Writer lock operations */
          (op==LOG_WRLOCK && mRegion==(LOG_REGION_C|LOG_REGION_D))
       || (op==LOG_UNLOCK && mRegion==(LOG_REGION_C|LOG_REGION_D))

       /* Normal reader lock operations */
       || (op==LOG_RDLOCK && mRegion==(LOG_REGION_A|LOG_REGION_B))
       || (op==LOG_UNLOCK && mRegion==(LOG_REGION_A))
       || (op==LOG_UNLOCK && mRegion==(LOG_REGION_B))

       /* Region D reader lock operations */
       || (op==LOG_RDLOCK && mRegion==(LOG_REGION_D))
       || (op==LOG_RDLOCK && mRegion==(LOG_REGION_A))
       || (op==LOG_UNLOCK && mRegion==(LOG_REGION_D))

       /* Checkpointer lock operations */
       || (op==LOG_WRLOCK && mRegion==(LOG_REGION_B|LOG_REGION_C))
       || (op==LOG_WRLOCK && mRegion==(LOG_REGION_A))
       || (op==LOG_UNLOCK && mRegion==(LOG_REGION_B|LOG_REGION_C))
       || (op==LOG_UNLOCK && mRegion==(LOG_REGION_A|LOG_REGION_B|LOG_REGION_C))
  );

  /* Assert that a connection never tries to go from an EXCLUSIVE to a 
  ** SHARED lock on a region. Moving from SHARED to EXCLUSIVE sometimes
  ** happens though (when a region D reader upgrades to a writer).
  */
  assert( op!=LOG_RDLOCK || 0==(pLog->lock.mLock & (mRegion<<8)) );

  sqlite3_mutex_enter(pSummary->mutex);

  /* Calculate a mask of logs held by all connections in this process apart
  ** from this one. The least significant byte of the mask contains a mask
  ** of the SHARED logs held. The next least significant byte of the mask
  ** indicates the EXCLUSIVE locks held. For example, to test if some other
  ** connection is holding a SHARED lock on region A, or an EXCLUSIVE lock
  ** on region C, do:
  **
  **   hasSharedOnA    = (mOther & (LOG_REGION_A<<0));
  **   hasExclusiveOnC = (mOther & (LOG_REGION_C<<8));
  **
  ** In all masks, if the bit in the EXCLUSIVE byte mask is set, so is the 
  ** corresponding bit in the SHARED mask.
  */
  mOther = 0;
  for(p=pSummary->pLock; p; p=p->pNext){
    assert( (p->mLock & (p->mLock<<8))==(p->mLock&0x0000FF00) );
    if( p!=&pLog->lock ){
      mOther |= p->mLock;
    }
  }

  /* If this call is to lock a region (not to unlock one), test if locks held
  ** by any other connection in this process prevent the new locks from
  ** begin granted. If so, exit the summary mutex and return SQLITE_BUSY.
  */
  if( op && (mOther & (mRegion << (op==LOG_RDLOCK ? 8 : 0))) ){
    sqlite3_mutex_leave(pSummary->mutex);
    return SQLITE_BUSY;
  }

  /* Figure out the new log mask for this connection. */
  switch( op ){
    case LOG_UNLOCK: 
      mNew = (pLog->lock.mLock & ~(mRegion|(mRegion<<8)));
      break;
    case LOG_RDLOCK:
      mNew = (pLog->lock.mLock | mRegion);
      break;
    default:
      assert( op==LOG_WRLOCK );
      mNew = (pLog->lock.mLock | (mRegion<<8) | mRegion);
      break;
  }

  /* Now modify the locks held on the log-summary file descriptor. This
  ** file descriptor is shared by all log connections in this process. 
  ** Therefore:
  **
  **   + If one or more log connections in this process hold a SHARED lock
  **     on a region, the file-descriptor should hold a SHARED lock on
  **     the file region.
  **
  **   + If a log connection in this process holds an EXCLUSIVE lock on a
  **     region, the file-descriptor should also hold an EXCLUSIVE lock on
  **     the region in question.
  **
  ** If this is an LOG_UNLOCK operation, only regions for which no other
  ** connection holds a lock should actually be unlocked. And if this
  ** is a LOG_RDLOCK operation and other connections already hold all
  ** the required SHARED locks, then no system call is required.
  */
  if( op==LOG_UNLOCK ){
    mRegion = (mRegion & ~mOther);
  }
  if( (op==LOG_WRLOCK)
   || (op==LOG_UNLOCK && mRegion) 
   || (op==LOG_RDLOCK && (mOther&mRegion)!=mRegion)
  ){
    struct LockMap {
      int iStart;                 /* Byte offset to start locking operation */
      int iLen;                   /* Length field for locking operation */
    } aMap[] = {
      /* 0000 */ {0, 0},                    /* 0001 */ {4+LOG_LOCK_REGION, 1}, 
      /* 0010 */ {3+LOG_LOCK_REGION, 1},    /* 0011 */ {3+LOG_LOCK_REGION, 2},
      /* 0100 */ {2+LOG_LOCK_REGION, 1},    /* 0101 */ {0, 0}, 
      /* 0110 */ {2+LOG_LOCK_REGION, 2},    /* 0111 */ {2+LOG_LOCK_REGION, 3},
      /* 1000 */ {1+LOG_LOCK_REGION, 1},    /* 1001 */ {0, 0}, 
      /* 1010 */ {0, 0},                    /* 1011 */ {0, 0},
      /* 1100 */ {1+LOG_LOCK_REGION, 2},    /* 1101 */ {0, 0}, 
      /* 1110 */ {0, 0},                    /* 1111 */ {0, 0}
    };
    int rc;                       /* Return code of logLockFd() */

    assert( mRegion<ArraySize(aMap) && aMap[mRegion].iStart!=0 );

    rc = logLockFd(pSummary, aMap[mRegion].iStart, aMap[mRegion].iLen, op);
    if( rc!=0 ){
      sqlite3_mutex_leave(pSummary->mutex);
      return rc;
    }
  }

  pLog->lock.mLock = mNew;
  sqlite3_mutex_leave(pSummary->mutex);
  return SQLITE_OK;
}

static int logLockDMH(LogSummary *pSummary, int eLock){
  assert( eLock==LOG_RDLOCK || eLock==LOG_WRLOCK );
  return logLockFd(pSummary, LOG_LOCK_DMH, 1, eLock);
}

static int logLockMutex(LogSummary *pSummary, int eLock){
  assert( eLock==LOG_WRLOCKW || eLock==LOG_UNLOCK );
  logLockFd(pSummary, LOG_LOCK_MUTEX, 1, eLock);
  return SQLITE_OK;
}



/*
** This function intializes the connection to the log-summary identified
** by struct pSummary.
*/
static int logSummaryInit(
  LogSummary *pSummary,           /* Log summary object to initialize */
  sqlite3_file *pFd               /* File descriptor open on log file */
){
  int rc;                         /* Return Code */
  char *zFile;                    /* File name for summary file */

  assert( pSummary->fd<0 );
  assert( pSummary->aData==0 );
  assert( pSummary->nRef>0 );
  assert( pSummary->zPath );

  /* Open a file descriptor on the summary file. */
  zFile = sqlite3_mprintf("%s-summary", pSummary->zPath);
  if( !zFile ){
    return SQLITE_NOMEM;
  }
  pSummary->fd = open(zFile, O_RDWR|O_CREAT, S_IWUSR|S_IRUSR);
  sqlite3_free(zFile);
  if( pSummary->fd<0 ){
    return SQLITE_IOERR;
  }

  /* Grab an exclusive lock the summary file. Then mmap() it. 
  **
  ** TODO: This code needs to be enhanced to support a growable mapping. 
  ** For now, just make the mapping very large to start with. The 
  ** pages should not be allocated until they are first accessed anyhow,
  ** so using a large mapping consumes no more resources than a smaller
  ** one would.
  */
  assert( sqlite3_mutex_held(pSummary->mutex) );
  rc = logLockMutex(pSummary, LOG_WRLOCKW);
  if( rc!=SQLITE_OK ) return rc;
  rc = logSummaryMap(pSummary, 512*1024);
  if( rc!=SQLITE_OK ) goto out;

  /* Try to obtain an EXCLUSIVE lock on the dead-mans-hand region. If this

  ** is possible, the contents of the log-summary file (if any) may not
  ** be trusted. Zero the log-summary header before continuing.



  */
  rc = logLockDMH(pSummary, LOG_WRLOCK);


  if( rc==SQLITE_OK ){

    memset(pSummary->aData, 0, (LOGSUMMARY_HDR_NFIELD+2)*sizeof(u32) );
  }
  rc = logLockDMH(pSummary, LOG_RDLOCK);
  if( rc!=SQLITE_OK ){
    return SQLITE_IOERR;
  }

 out:
  logLockMutex(pSummary, LOG_UNLOCK);
  return rc;
}

/* 
** Open a connection to the log file associated with database zDb. The
** database file does not actually have to exist. zDb is used only to
** figure out the name of the log file to open. If the log file does not 
** exist it is created by this call.
**
** A SHARED lock should be held on the database file when this function
** is called. The purpose of this SHARED lock is to prevent any other
** client from unlinking the log or log-summary file. If another process
** were to do this just after this client opened one of these files, the
** system would be badly broken.
*/
int sqlite3LogOpen(
  sqlite3_vfs *pVfs,              /* vfs module to open log file with */
  const char *zDb,                /* Name of database file */
  Log **ppLog                     /* OUT: Allocated Log handle */
){
  int rc = SQLITE_OK;             /* Return Code */
  Log *pRet;                      /* Object to allocate and return */
  LogSummary *pSummary = 0;       /* Summary object */
  sqlite3_mutex *mutex = 0;       /* LOG_SUMMARY_MUTEX mutex */
  int flags;                      /* Flags passed to OsOpen() */
  char *zWal = 0;                 /* Path to WAL file */
  int nWal;                       /* Length of zWal in bytes */


  assert( zDb );


  /* Allocate an instance of struct Log to return. */
  *ppLog = 0;
  pRet = (Log *)sqlite3MallocZero(sizeof(Log) + pVfs->szOsFile);
  if( !pRet ) goto out;
  pRet->pVfs = pVfs;
  pRet->pFd = (sqlite3_file *)&pRet[1];
  pRet->sync_flags = SQLITE_SYNC_NORMAL;

  /* Normalize the path name. */

  /* Open file handle on the log file. */
  rc = sqlite3OsOpen(pVfs, pSummary->zPath, pRet->pFd, flags, &flags);
  if( rc!=SQLITE_OK ) goto out;

  /* Object pSummary is shared between all connections to the database made
  ** by this process. So at this point it may or may not be connected to
  ** the log-summary. If it is not, connect it.

  */
  sqlite3_mutex_enter(pSummary->mutex);
  mutex = pSummary->mutex;
  if( pSummary->fd<0 ){
    rc = logSummaryInit(pSummary, pRet->pFd);


  }

  pRet->lock.pNext = pSummary->pLock;
  pSummary->pLock = &pRet->lock;

 out:
  sqlite3_mutex_leave(mutex);

    ** reference to the log summary object in this process, the object will
    ** be freed. If this is also the last connection to the database, then
    ** checkpoint the database and truncate the log and log-summary files
    ** to zero bytes in size.
    **/
    pSummary->nRef--;
    if( pSummary->nRef==0 ){
      int rc;
      LogSummary **pp;
      for(pp=&pLogSummary; *pp!=pSummary; pp=&(*pp)->pNext);
      *pp = (*pp)->pNext;

      sqlite3_mutex_leave(mutex);

      rc = sqlite3OsLock(pFd, SQLITE_LOCK_EXCLUSIVE);
      if( rc==SQLITE_OK ){



        /* This is the last connection to the database (including other
        ** processes). Do three things:
        **
        **   1. Checkpoint the db.
        **   2. Truncate the log file.
        **   3. Unlink the log-summary file.
        */
        rc = logCheckpoint(pLog, pFd, zBuf);
        if( rc==SQLITE_OK ){
          rc = sqlite3OsDelete(pLog->pVfs, pSummary->zPath, 0);
        }

        logSummaryUnmap(pSummary, 1);
      }else{
        if( rc==SQLITE_BUSY ){
          rc = SQLITE_OK;
        }
        logSummaryUnmap(pSummary, 0);



      }
      sqlite3OsUnlock(pFd, SQLITE_LOCK_NONE);






      sqlite3_mutex_free(pSummary->mutex);
      sqlite3_free(pSummary);
    }else{
      sqlite3_mutex_leave(mutex);
    }



    /* Close the connection to the log file and free the Log handle. */
    sqlite3OsClose(pLog->pFd);
    sqlite3_free(pLog);
  }
  return rc;
}

** hosted in this address space.
*/
static int logEnterMutex(Log *pLog){
  LogSummary *pSummary = pLog->pSummary;
  int rc;

  sqlite3_mutex_enter(pSummary->mutex);
  rc = logLockMutex(pSummary, LOG_WRLOCKW);
  if( rc!=SQLITE_OK ){
    sqlite3_mutex_leave(pSummary->mutex);
  }
  return rc;
}
static void logLeaveMutex(Log *pLog){
  LogSummary *pSummary = pLog->pSummary;















































































  logLockMutex(pSummary, LOG_UNLOCK);





























































  sqlite3_mutex_leave(pSummary->mutex);







}

/*
** Try to read the log-summary header. Attempt to verify the header
** checksum. If the checksum can be verified, copy the log-summary
** header into structure pLog->hdr. If the contents of pLog->hdr are
** modified by this and pChanged is not NULL, set *pChanged to 1. 

  /* The changeCountDone flag is always set for temp-files */
  assert( pPager->tempFile==0 || pPager->changeCountDone );

  return 1;
}
#endif


/*
** Return true if it is necessary to write page *pPg into the sub-journal.
** A page needs to be written into the sub-journal if there exists one
** or more open savepoints for which:
**
**   * The page-number is less than or equal to PagerSavepoint.nOrig, and

      testcase( rc==SQLITE_NOMEM );
      assert( rc==SQLITE_OK || rc==SQLITE_NOMEM );
    }
  }
  return rc;
}














/*
** Return true if this pager uses a write-ahead log instead of the usual
** rollback journal. Otherwise false.
*/
static int pagerUseLog(Pager *pPager){
  return (pPager->pLog!=0);
}

    ** until the change-counter check fails in PagerSharedLock().
    ** Clearing the page size cache here is being conservative.
    */
    pPager->dbSizeValid = 0;

    if( pagerUseLog(pPager) ){
      sqlite3LogCloseSnapshot(pPager->pLog);
    }else{
      rc = osUnlock(pPager->fd, NO_LOCK);
    }
    if( rc ){
      pPager->errCode = rc;
    }
    IOTRACE(("UNLOCK %p\n", pPager))

    /* If Pager.errCode is set, the contents of the pager cache cannot be
    ** trusted. Now that the pager file is unlocked, the contents of the

        }
      }
    }
  }

  return rc;
}

/*
** Open a connection to the write-ahead log file for pager pPager. If
** the log connection is already open, this function is a no-op.
*/
static int pagerOpenLog(Pager *pPager){
  if( !pPager->pLog ){
    int rc;                       /* Return code */

    /* Before opening the log file, obtain a SHARED lock on the database
    ** file. This lock will not be released until after the log file
    ** connection has been closed. The purpose of this lock is to stop
    ** any other process from unlinking the log or log-summary files while
    ** this connection still has them open. An EXCLUSIVE lock on the
    ** database file is required to unlink either of those two files.
    */
    assert( pPager->state==PAGER_UNLOCK );
    rc = pager_wait_on_lock(pPager, SHARED_LOCK);
    if( rc!=SQLITE_OK ){
      assert( pPager->state==PAGER_UNLOCK );
      return pager_error(pPager, rc);
    }
    assert( pPager->state>=SHARED_LOCK );

    /* Open the connection to the log file. If this operation fails, 
    ** (e.g. due to malloc() failure), unlock the database file and 
    ** return an error code.
    */
    rc = sqlite3LogOpen(pPager->pVfs, pPager->zFilename, &pPager->pLog);
    if( rc!=SQLITE_OK ){
      osUnlock(pPager->fd, SQLITE_LOCK_NONE);
      pPager->state = PAGER_UNLOCK;
      return rc;
    }
  }else{
    /* If the log file was already open, check that the pager is still holding
    ** the required SHARED lock on the database file. 
    */
#ifdef SQLITE_DEBUG
    int locktype;
    sqlite3OsFileControl(pPager->fd, SQLITE_FCNTL_LOCKSTATE, &locktype);
    assert( locktype==SQLITE_LOCK_SHARED );
#endif
  }

  return SQLITE_OK;
}


/*
** This function is called to obtain a shared lock on the database file.
** It is illegal to call sqlite3PagerAcquire() until after this function
** has been successfully called. If a shared-lock is already held when
** this function is called, it is a no-op.
**

    if( isOpen(pPager->jfd) || pPager->zJournal ){
      isErrorReset = 1;
    }
    pPager->errCode = SQLITE_OK;
    pager_reset(pPager);
  }


  if( pPager->journalMode==PAGER_JOURNALMODE_WAL ){
    int changed = 0;              /* True if the cache must be flushed */

    /* Open the log file, if it is not already open. */
    rc = pagerOpenLog(pPager);
    if( rc!=SQLITE_OK ){
      return rc;
    }

    /* Open a log snapshot to read from. */
    rc = sqlite3LogOpenSnapshot(pPager->pLog, &changed);
    if( rc==SQLITE_OK ){
      int dummy;
      if( changed ){
        pager_reset(pPager);
        assert( pPager->errCode || pPager->dbSizeValid==0 );
      }

      rc = sqlite3PagerPagecount(pPager, &dummy);
    }
  }else if( pPager->state==PAGER_UNLOCK || isErrorReset ){
    sqlite3_vfs * const pVfs = pPager->pVfs;
    int isHotJournal = 0;
    assert( !MEMDB );
    assert( sqlite3PcacheRefCount(pPager->pPCache)==0 );
    if( pPager->noReadlock ){

    assert( (PAGER_JOURNALMODE_MEMORY & 1)==0 );
    assert( (PAGER_JOURNALMODE_OFF & 1)==0 );
    if( (pPager->journalMode & 1)==1 && (eMode & 1)==0
         && !pPager->exclusiveMode ){
      sqlite3OsDelete(pPager->pVfs, pPager->zJournal, 0);
    }

    /* Switching into WAL mode can only take place when no 
    ** locks are held on the database file. 
    */
    if( eMode==PAGER_JOURNALMODE_WAL && pPager->state!=PAGER_UNLOCK ){





      return (int)pPager->journalMode;
    }

    pPager->journalMode = (u8)eMode;
  }
  return (int)pPager->journalMode;
}

/*
** Get/set the size-limit used for persistent journal files.

  [lindex $args 0] eval { PRAGMA journal_mode = wal }
}

proc log_file_size {nFrame pgsz} {
  expr {12 + ($pgsz+16)*$nFrame}
}

proc log_deleted {logfile} {
  return [expr [file exists $logfile]==0]
}

#
# These are 'warm-body' tests used while developing the WAL code. They
# serve to prove that a few really simple cases work:
#
# wal-1.*: Read and write the database.
# wal-2.*: Test MVCC with one reader, one writer.
# wal-3.*: Test transaction rollback.

  execsql {
    COMMIT;
    SELECT * FROM t2;
  }
} {1 2 3 4}
db close


foreach sector {512 4096} {
  sqlite3_simulate_device -sectorsize $sector
  foreach pgsz {512 1024 2048 4096} {
    file delete -force test.db test.db-wal
    do_test wal-6.$sector.$pgsz.1 {
      sqlite3_wal db test.db -vfs devsym
      execsql "
        PRAGMA page_size = $pgsz ;
      "
      execsql "
        CREATE TABLE t1(a, b);
        INSERT INTO t1 VALUES(1, 2);
      "
      db close
      file size test.db
    } [expr $pgsz*2]
  
    do_test wal-6.$sector.$pgsz.2 {
      log_deleted test.db-wal
    } {1}
  }
}

do_test wal-7.1 {
  file delete -force test.db test.db-wal
  sqlite3_wal db test.db
  execsql {

do_test wal-11.8 {
  execsql { PRAGMA checkpoint }
  list [expr [file size test.db]/1024] [file size test.db-wal]
} [list 37 [log_file_size 42 1024]]
do_test wal-11.9 {
  db close
  sqlite3_wal db test.db
  list [expr [file size test.db]/1024] [log_deleted test.db-wal]
} {37 1}

do_test wal-11.10 {
  execsql {
    PRAGMA cache_size = 10;
    BEGIN;
      INSERT INTO t1 SELECT randomblob(900) FROM t1;   -- 32
      SELECT count(*) FROM t1;

source $testdir/tester.tcl
if {[run_thread_tests]==0} { finish_test ; return }

do_test walthread-1.1 {
  execsql {
    PRAGMA journal_mode = WAL;
    PRAGMA lock_status;
    CREATE TABLE t1(x PRIMARY KEY);
    PRAGMA lock_status;
    INSERT INTO t1 VALUES(randomblob(100));
    INSERT INTO t1 VALUES(randomblob(100));
    INSERT INTO t1 SELECT md5sum(x) FROM t1;
  }
} {wal main unlocked temp closed main shared temp closed}
do_test walthread-1.2 {
  execsql {
    SELECT (SELECT count(*) FROM t1), (
      SELECT md5sum(x) FROM t1 WHERE oid != (SELECT max(oid) FROM t1)
    ) == (
      SELECT x FROM t1 WHERE oid = (SELECT max(oid) FROM t1)
    )
  }
} {3 1}
do_test walthread-1.3 {
  execsql { PRAGMA integrity_check } 
} {ok}
do_test walthread-1.4 {
  execsql { PRAGMA lock_status } 
} {main shared temp unknown}

#--------------------------------------------------------------------------
# Start N threads. Each thread performs both read and write transactions.
# Each read transaction consists of:
#
#   1) Reading the md5sum of all but the last table row,
#   2) Running integrity check.