SQLite4  Check-in [ec6dda0ade]

#include <assert.h>
#include <stdio.h>
#include <stddef.h>

/* Magic values identifying WAL file header */
#define BT_WAL_MAGIC   0xBEE1CA62
#define BT_WAL_VERSION 0x00000001

/* Wrap the log around if there is a block of this many free frames at
** the start of the file.  */
#define BT_NWRAPLOG    100

typedef struct BtWalHdr BtWalHdr;
typedef struct BtShmHdr BtShmHdr;
typedef struct BtCkptHdr BtCkptHdr;
typedef struct BtFrameHdr BtFrameHdr;
typedef struct BtShm BtShm;

  u32 iFirstFrame;                /* First frame of log (numbered from 1) */

  u32 aCksum[2];                  /* Checksum of all prior fields */
};

/*
** WAL Frame header. All fields are stored in big-endian order.
**
** ctrl:
**   The most-significant-bit (BT_FRAME_COMMIT) of this field is set for
**   a commit frame and clear for all others. The other 31 bits contain
**   the frame number of the next frame in the log. 
*/
struct BtFrameHdr {
  u32 pgno;                       /* Page number of this frame */
  u32 ctrl;                       /* Next frame pointer and commit bit */
  u32 aCksum[2];                  /* Frame checksum */
};

#define BT_FRAME_COMMIT 0x80000000

/*
** Shared memory header. Shared memory begins with two copies of
** this structure. All fields are stored in machine byte-order.
*/
struct BtShmHdr {
  u32 aLog[6];                    /* First/last frames for each log region */
  int nSector;                    /* Sector size assumed for WAL file */
  int iHashSide;                  /* Hash table side for region (c) of log */
  u32 aFrameCksum[2];             /* Checksum of previous frame */
  u32 iNextFrame;                 /* Location to write next log frame to */
  int padding;                    /* So that this structure is 8-byte aligned */
  u32 aCksum[2];                  /* Object checksum */
};

/*
** A single instance of this structure follows the two BtShmHdr structures 
** in shared memory.
**

** Log handle used by bt_pager.c to access functionality implemented by
** this module. 
*/
struct BtLog {
  BtLock *pLock;                  /* Lock object associated with this log */
  bt_file *pFd;                   /* File handle open on WAL file */
  BtShmHdr snapshot;              /* Current snapshot of shm-header */

  int nShm;                       /* Size of apShm[] array */
  u8 **apShm;                     /* Array of mapped shared-memory blocks */
  int nWrapLog;                   /* Wrap if this many free frames at start */
};

typedef u16 ht_slot;

/*
** Number of entries in each hash table bar the first. 
*/

** returns the value that would be produced by intepreting the 4 bytes
** of the input value as a little-endian integer.
*/
#define BYTESWAP32(x) ( \
  (((x)&0x000000FF)<<24) + (((x)&0x0000FF00)<<8)  \
  + (((x)&0x00FF0000)>>8)  + (((x)&0xFF000000)>>24) \
)

static void btLogDebugTopology(char *zStr, u32 *aLog){
  fprintf(stderr, "%s: %d..%d  %d..%d  %d..%d\n", zStr,
      (int)aLog[0], (int)aLog[1], (int)aLog[2], 
      (int)aLog[3], (int)aLog[4], (int)aLog[5]
  );
  fflush(stderr);
}

/*
** Generate or extend an 8 byte checksum based on the data in
** array aByte[] and the initial values of aIn[0] and aIn[1] (or
** initial values of 0 and 0 if aIn==NULL).
**
** The checksum is written back into aOut[] before returning.

      btLogChecksum(1, aBuf, pgsz, aCksum, aCksum);
      if( aCksum[0]!=fhdr.aCksum[0] || aCksum[1]!=fhdr.aCksum[1] ) break;
    }
    if( rc==SQLITE4_OK ){
      rc = xFrame(pLog, pCtx, iFrame, &fhdr);
    }

    iFrame = (fhdr.ctrl & ~BT_FRAME_COMMIT);
  }

  return rc;
}

/*
** Locate the iHash'th hash table in shared memory. Return it.
*/
static int btLogFindHash(
  BtLog *pLog,                    /* Log handle */
  int iSide,                      /* Which set of hash slots to return */
  int iHash,                      /* Hash table (numbered from 0) to find */
  ht_slot **paHash,               /* OUT: Pointer to hash slots */
  u32 **paPgno,                   /* OUT: Pointer to page number array */
  u32 *piZero                     /* OUT: Frame associated with *paPgno[0] */
){
  int rc;                         /* Return code */

  assert( iSide==0 || iSide==1 );

  rc = btLogMapShm(pLog, iHash);
  if( rc==SQLITE4_OK ){
    u8 *aChunk = pLog->apShm[iHash];
    u32 *aPgno;
    u32 iZero;

    *paHash = (ht_slot*)&aChunk[iSide?HASHTABLE_OFFSET_1:HASHTABLE_OFFSET_2];
    if( iHash==0 ){
      aPgno = (u32*)&aChunk[sizeof(BtShm)];
      iZero = 1;
    }else{
      aPgno = (u32*)aChunk;
      iZero = 1 + HASHTABLE_NFRAME_ONE + (HASHTABLE_NFRAME * (iHash-1));
    }

*/
static int btLogHashInsert(BtLog *pLog, u32 pgno, u32 iFrame){
  int iHash;                      /* Index of hash table to update */
  int rc = SQLITE4_OK;            /* Return code */
  ht_slot *aHash;                 /* Hash slots */
  u32 *aPgno;                     /* Page array for updated hash table */
  u32 iZero;                      /* Zero-offset of updated hash table */
  int iSide = pLog->snapshot.iHashSide;

  assert( iFrame>=1 && pgno>=1 );

  /* Find the required hash table */
  iHash = btLogFrameHash(pLog, iFrame);
  rc = btLogFindHash(pLog, iSide, iHash, &aHash, &aPgno, &iZero);

  /* Update the hash table */
  if( rc==SQLITE4_OK ){
    int iSlot;
    int nCollide = HASHTABLE_NSLOT*2;
    aPgno[iFrame-iZero] = pgno;

    if( iFrame==iZero ){
      memset(aHash, 0, sizeof(ht_slot) * HASHTABLE_NSLOT);
    }

    for(iSlot=btLogHashKey(pLog,pgno); ; iSlot=btLogHashNext(pLog, iSlot)){
      if( aHash[iSlot]==0 ){
        aHash[iSlot] = (iFrame-iZero+1);
        break;
      }
      if( (nCollide--)==0 ) return btErrorBkpt(SQLITE4_CORRUPT);

** Remove everything following frame iFrame from the iHash'th hash table.
*/
static int btLogHashRollback(BtLog *pLog, int iHash, u32 iFrame){
  const int nPgno = (iHash==0 ? HASHTABLE_NFRAME_ONE : HASHTABLE_NFRAME);
  ht_slot *aHash;                 /* Hash slots */
  u32 *aPgno;                     /* Page array for updated hash table */
  u32 iZero;                      /* Zero-offset of updated hash table */
  int iSide = pLog->snapshot.iHashSide;
  int rc;

  rc = btLogFindHash(pLog, iSide, iHash, &aHash, &aPgno, &iZero);
  if( rc==SQLITE4_OK ){
    int i;
    ht_slot iMax;
    iMax = (iFrame - iZero) + 1;

    for(i=0; i<HASHTABLE_NSLOT; i++){
      if( aHash[i]>iMax ) aHash[i] = 0;
    }
    memset(&aPgno[iMax], 0, (nPgno-iMax)*sizeof(u32));
  }

  return rc;
}


/*
** Return true if log is completely empty (as it is if a file zero bytes
** in size has been opened or created).
*/
static int btLogIsEmpty(BtLog *pLog){
  return (pLog->snapshot.aLog[4]==0 && pLog->snapshot.iNextFrame==0);
}

typedef struct FrameRecoverCtx FrameRecoverCtx;
struct FrameRecoverCtx {
  u32 iLast;                      /* Frame containing last commit flag in log */
  u32 iNextFrame;                 /* Frame that follows frame iLast */
};

static int btLogRecoverFrame(
  BtLog *pLog,                    /* Log module handle */
  void *pCtx,                     /* woints to type u32 - pgno of last commit*/
  u32 iFrame,                     /* Frame number */
  BtFrameHdr *pHdr                /* Frame header */
){
  if( btLogIsEmpty(pLog) ){
    /* This is the first frame recovered. It is therefore both the first
    ** and last frame of log region (c).  */
    pLog->snapshot.aLog[4] = iFrame;
    pLog->snapshot.aLog[5] = iFrame;
  }else{
    u32 iExpect = pLog->snapshot.aLog[5]+1;
    if( iFrame==iExpect ){
      pLog->snapshot.aLog[5] = iFrame;
    }else if( iFrame<iExpect ){
      assert( iFrame==1 );
      assert( pLog->snapshot.aLog[0]==0 && pLog->snapshot.aLog[1]==0 );
      pLog->snapshot.aLog[0] = pLog->snapshot.aLog[4];
      pLog->snapshot.aLog[1] = pLog->snapshot.aLog[5];
      pLog->snapshot.aLog[4] = iFrame;
      pLog->snapshot.aLog[5] = iFrame;
      pLog->snapshot.iHashSide = (pLog->snapshot.iHashSide + 1) % 2;
    }else{
      assert( pLog->snapshot.aLog[2]==0 && pLog->snapshot.aLog[3]==0 );
      pLog->snapshot.aLog[2] = pLog->snapshot.aLog[4];
      pLog->snapshot.aLog[3] = pLog->snapshot.aLog[5];
      pLog->snapshot.aLog[4] = iFrame;
      pLog->snapshot.aLog[5] = iFrame;
    }
  }

  btLogHashInsert(pLog, pHdr->pgno, iFrame);
  if( pHdr->ctrl & BT_FRAME_COMMIT ){
    FrameRecoverCtx *pFRC = (FrameRecoverCtx*)pCtx;
    pFRC->iLast = iFrame;
    pFRC->iNextFrame = (pHdr->ctrl & ~BT_FRAME_COMMIT);
    memcpy(pLog->snapshot.aFrameCksum, pHdr->aCksum, sizeof(pHdr->aCksum));
  }

#if 0
  fprintf(stderr, "recovered frame=%d pgno=%d\n", iFrame, pHdr->pgno);
  fflush(stderr);
#endif
  return 0;
}

/*
** This function is called as part of log recovery. The log file has 
** already been scanned and the log topology (pLog->snapshot.aLog[])
** shared-memory hash tables have been populated with data corresponding
** to the entire set of valid frames recovered from the log file -
** including uncommitted frames. This function removes the uncommitted
** frames from the log topology and shared hash tables.
*/
static int btLogRollbackRecovery(BtLog *pLog, FrameRecoverCtx *pCtx){
  u32 iLast = pCtx->iLast;        /* Last committed frame in log file */
  u32 *aLog = pLog->snapshot.aLog;/* Log file topology */

  while( iLast<aLog[4] || iLast>aLog[5] ){
    if( aLog[2] ){
      aLog[5] = aLog[3];
      aLog[4] = aLog[2];
      if( aLog[0] && aLog[0]<aLog[4] ){
        aLog[3] = aLog[1];
        aLog[2] = aLog[0];
        aLog[0] = aLog[1] = 0;
      }else{
        aLog[2] = aLog[3] = 0;
      }
    }else{
      aLog[5] = aLog[1];
      aLog[4] = aLog[0];
      aLog[0] = aLog[1] = 0;
      pLog->snapshot.iHashSide = (pLog->snapshot.iHashSide + 1) % 2;
    }
  }

  aLog[5] = iLast;
  return btLogHashRollback(pLog, btLogFrameHash(pLog, iLast), iLast);
}

/*
** Run log recovery. In other words, read the log file from disk and 
** initialize the shared-memory accordingly.
*/
static int btLogRecover(BtLog *pLog){
  bt_env *pVfs = pLog->pLock->pVfs;

        pHdr = &hdr1;
        iSlot = 1;
      }
    }
    if( rc==SQLITE4_NOTFOUND ) rc = SQLITE4_OK;
  }

  /* If a header was successfully read from the file, attempt to 
  ** recover frames from the log file. */
  if( pHdr ){

    /* The following iterates through all readable frames in the log file.
    ** It populates pLog->snapshot.aLog[] with the log topology and the
    ** shared hash-tables with the pgno->frame mapping. The FrameRecoverCtx
    ** object is populated with the frame number and "next frame" pointer of
    ** the last commit-frame in the log (if any). Additionally, the
    ** pLog->snapshot.aFrameCksum[] variables are populated with the checksum
    ** beloging to the frame header of the last commit-frame in the log.  */
    FrameRecoverCtx ctx = {0, 0};
    rc = btLogTraverse(pLog, pHdr, btLogRecoverFrame, (void*)&ctx);

    if( rc==SQLITE4_OK ){
      if( ctx.iLast==0 ){
        /* No transactions recovered from the log file. */
        btLogZeroSnapshot(pLog);
      }else{
        /* One or more transactions were recovered from the log file. */
        BtShm *pShm = btLogShm(pLog);

        pShm->ckpt.iWalHdr = (iSlot<<2) + pHdr->iCnt;
        pShm->ckpt.iFirstRead = pHdr->iFirstFrame;
        pShm->ckpt.iFirstRecover = pHdr->iFirstFrame;
        rc = btLogRollbackRecovery(pLog, &ctx);
        pLog->snapshot.iNextFrame = ctx.iNextFrame;


      }
    }
  }

  return rc;
}

/*

  pLog = sqlite4_malloc(pEnv, sizeof(BtLog));
  if( pLog==0 ){
    rc = SQLITE4_NOMEM;
    goto open_out;
  }
  memset(pLog, 0, sizeof(BtLog));
  pLog->pLock = (BtLock*)pPager;
  pLog->nWrapLog = BT_NWRAPLOG;

  zWal = sqlite4BtPagerFilename(pPager, BT_PAGERFILE_LOG);
  rc = pVfs->xOpen(pEnv, pVfs, zWal, 0, &pLog->pFd);

  if( rc==SQLITE4_OK && bRecover ){
    rc = btLogMapShm(pLog, 0);
    if( rc==SQLITE4_OK ){

  /* Write the object to disk */
  iOff = iHdr * pLog->snapshot.nSector;
  rc = btLogWriteData(pLog, iOff, (u8*)pHdr, sizeof(BtWalHdr));

  return rc;
}

static int btLogHashSearch(
  BtLog *pLog,                    /* Log module handle */
  int iSide,                      /* 0 or 1 - the side of hash table to read */
  int iHash,                      /* Index of hash to query */
  u32 pgno,                       /* query for this page number */
  u32 *piFrame                    /* OUT: Frame number for matching entry */
){
  ht_slot *aHash;
  u32 *aPgno;
  u32 iZero;
  int rc;

  rc = btLogFindHash(pLog, iSide, iHash, &aHash, &aPgno, &iZero);
  if( rc==SQLITE4_OK ){
    int nCollide = HASHTABLE_NSLOT*2;
    int iSlot;
    u32 iFrame = 0;
    
    iSlot=btLogHashKey(pLog, pgno); 
    for( ; aHash[iSlot]; iSlot=btLogHashNext(pLog, iSlot)){

      rc = SQLITE4_NOTFOUND;
    }
  }

  return rc;
}


/*
** Attempt to read data for page pgno from the log file. If successful,
** the data is written into buffer aData[] (which must be at least as
** large as a database page). In this case SQLITE4_OK is returned.
**
** If the log does not contain any version of page pgno, SQLITE4_NOTFOUND
** is returned and the contents of buffer aData[] are not modified.
**
** If any other error occurs, an SQLite4 error code is returned. The final
** state of buffer aData[] is undefined in this case.
*/
int sqlite4BtLogRead(BtLog *pLog, u32 pgno, u8 *aData){
  const int pgsz = sqlite4BtPagerPagesize((BtPager*)(pLog->pLock));


  int rc = SQLITE4_NOTFOUND;
  u32 iFrame = 0;
  int i;

  /* Loop through regions (c), (b) and (a) of the log file. In that order. */
  for(i=2; i>=0; i--){
    u32 iLo = pLog->snapshot.aLog[i*2+0];
    u32 iHi = pLog->snapshot.aLog[i*2+1];
    int iSide;
    int iHash;
    int iHashLast;

    iHash = btLogFrameHash(pLog, iHi);
    iHashLast = btLogFrameHash(pLog, iLo);
    iSide = (pLog->snapshot.iHashSide + (i==0)) % 2;

    for( ; rc==SQLITE4_NOTFOUND && iHash>=iHashLast && iFrame==0; iHash--){
      rc = btLogHashSearch(pLog, iSide, iHash, pgno, &iFrame);
      if( rc==SQLITE4_OK && (iFrame<iLo || iFrame>iHi) ){
        rc = SQLITE4_NOTFOUND;
      }
    }
  }

  if( rc==SQLITE4_OK && iFrame!=0 ){
    bt_env *pVfs = pLog->pLock->pVfs;
    i64 iOff;
    assert( rc==SQLITE4_OK );
    iOff = btLogFrameOffset(pLog, pgsz, iFrame);

int sqlite4BtLogWrite(BtLog *pLog, u32 pgno, u8 *aData, int bCommit){
  const int pgsz = sqlite4BtPagerPagesize((BtPager*)(pLog->pLock));
  int rc = SQLITE4_OK;
  u32 iFrame;                     /* Write this frame (numbered from 1) */
  BtFrameHdr frame;               /* Header for new frame */
  u32 *a;                         /* Pointer to cksum of previous frame */
  i64 iOff;                       /* Offset of log file to write to */
  u32 iNextFrame;
  u32 *aLog = pLog->snapshot.aLog;

  /* Handle a special case - if the log file is completely empty then
  ** this writer must write the first header into the WAL file. */
  if( btLogIsEmpty(pLog) ){
    BtWalHdr hdr;
    memset(&hdr, 0, sizeof(BtWalHdr));

    hdr.iMagic = BT_WAL_MAGIC;
    hdr.iVersion = BT_WAL_VERSION;
    hdr.nSector = pLog->snapshot.nSector;
    hdr.nPgsz = pgsz;
    hdr.iSalt1 = 22;
    hdr.iSalt2 = 23;
    hdr.iFirstFrame = 1;

    rc = btLogWriteHeader(pLog, 0, &hdr);
    if( rc!=SQLITE4_OK ) return rc;

    pLog->snapshot.aFrameCksum[0] = hdr.iSalt1;
    pLog->snapshot.aFrameCksum[1] = hdr.iSalt2;
    pLog->snapshot.iNextFrame = 1;
  }

  /* Figure out the offset to write the current frame to. */
  iFrame = pLog->snapshot.iNextFrame;
  iOff = btLogFrameOffset(pLog, pgsz, iFrame);

  /* The current frame will be written to location pLog->snapshot.iNextFrame.
  ** This code determines where the following frame will be stored. There
  ** are three possibilities:
  **
  **   1) The next frame follows the current frame (this is the usual case).
  **   2) The next frame is frame 1 - the log wraps around.
  **   3) Following the current frame is a block of frames still in use.
  **      So the next frame will immediately follow this block.
  */
  iNextFrame = pLog->snapshot.iNextFrame + 1;
  if( iFrame!=1 
   && aLog[0]==0 && aLog[2]==0 
   && aLog[4]!=0 && aLog[4]>pLog->nWrapLog 
  ){
    /* Case 2) It is possible to wrap the log around */
    iNextFrame = 1;
  }else if( iNextFrame==aLog[0] ){
    /* Case 3) It is necessary to jump over some existing log. */
    iNextFrame = aLog[1]+1;
  }

  if( iNextFrame & 0x80000000 ){
    rc = SQLITE4_FULL;
  }else{

    /* Populate the frame header object. */
    memset(&frame, 0, sizeof(frame));
    frame.pgno = pgno;
    frame.ctrl = (bCommit ? BT_FRAME_COMMIT : 0) + iNextFrame;
    a = pLog->snapshot.aFrameCksum;
    btLogChecksum(1, (u8*)&frame, offsetof(BtFrameHdr,aCksum), a, frame.aCksum);
    btLogChecksum(1, aData, pgsz, frame.aCksum, frame.aCksum);

#if 0
    fprintf(stderr, "writing page %d at log offset %d (frame %d)\n", (int)pgno, (int)iOff, (int)iFrame);
    fflush(stderr);
#endif


    /* Write the frame header to the log file. */
    rc = btLogWriteData(pLog, iOff, (u8*)&frame, sizeof(frame));
  }
  pLog->snapshot.iNextFrame = iNextFrame;

  /* Write the frame contents to the log file. */
  if( rc==SQLITE4_OK ){
    rc = btLogWriteData(pLog, iOff+sizeof(frame), aData, pgsz);
  }

  /* Update the wal index hash tables with the (pgno -> iFrame) record. */
  if( rc==SQLITE4_OK ){
    if( iFrame==1 ){
      pLog->snapshot.iHashSide = (pLog->snapshot.iHashSide+1) %2;
    }
    rc = btLogHashInsert(pLog, pgno, iFrame);
  }

  /* Update the private copy of the shm-header */
  if( rc==SQLITE4_OK ){
btLogDebugTopology("log1", aLog);
    if( btLogIsEmpty(pLog) ){
      assert( iFrame==1 );
      aLog[4] = iFrame;
    }else if( iFrame==1 ){
      assert( aLog[0]==0 && aLog[1]==0 && aLog[2]==0 && aLog[3]==0 );
      aLog[0] = aLog[4];
      aLog[1] = aLog[5];
      aLog[4] = iFrame;
    }else if( iFrame!=aLog[5]+1 ){
      assert( iFrame>aLog[5] );
      assert( aLog[2]==0 && aLog[3]==0 );
      aLog[2] = aLog[4];
      aLog[3] = aLog[5];
      aLog[4] = iFrame;
    }

    aLog[5] = iFrame;
    memcpy(pLog->snapshot.aFrameCksum, frame.aCksum, sizeof(frame.aCksum));
btLogDebugTopology("log2", aLog);
  }

  /* If this is a COMMIT, also update the shared shm-header. */
  if( bCommit ){
    rc = btLogUpdateSharedHdr(pLog);
  }

    if( rc==SQLITE4_OK && memcmp(&shmhdr, &pLog->snapshot, sizeof(BtShmHdr)) ){
      sqlite4BtLockReaderUnlock(pLog->pLock);
      rc = SQLITE4_NOTFOUND;
    }
  }

  if( rc==SQLITE4_OK ){
    int iRegion;
    u32 *aLog = pLog->snapshot.aLog;
    BtShm *pShm = btLogShm(pLog);
    u32 iCkpt = pShm->ckpt.iFirstRecover;

fprintf(stderr, "iCkpt=%d\n", (int)iCkpt);
btLogDebugTopology("log3", aLog);

    for(iRegion=0; iRegion<3; iRegion++){
      if( aLog[iRegion*2] ){
        if( iCkpt>=aLog[iRegion*2] && iCkpt<=aLog[iRegion*2+1] ){
          aLog[iRegion*2] = iCkpt;
          break;
        }else{
          aLog[iRegion*2] = 0;
          aLog[iRegion*2+1] = 0;
        }
      }
    }

btLogDebugTopology("log4", aLog);
  }

  return rc;
}

int sqlite4BtLogSnapshotClose(BtLog *pLog){
  sqlite4BtLockReaderUnlock(pLog->pLock);

/*
** Parameters iFirst and iLast are frame numbers for frames that are part 
** of the current log. This function scans the wal-index from iFirst to
** iLast (inclusive) and records the set of page numbers that occur once.
** This set is sorted in ascending order and returned via the output 
** variables *paPgno and *pnPgno.

*/
static int btLogGatherPgno(
  BtLog *pLog,                    /* Log module handle */


  u32 **paPgno,                   /* OUT: s4_malloc'd array of sorted pgno */
  int *pnPgno,                    /* OUT: Number of entries in *paPgno */
  u32 *piLastFrame
){
  u32 *aLog = pLog->snapshot.aLog;/* Log file topology */
  u32 i;
  u32 *aPgno;                     /* Returned array */
  u32 *aSpace;                    /* Temporary space used by merge-sort */
  int nMax;
  int rc = SQLITE4_OK;
  int iRegion;

  /* Determine an upper limit on the number of distinct page numbers. This
  ** limit is used to allocate space for the returned array.  */
  nMax = 0;
  for(iRegion=0; iRegion<3; iRegion++){
    if( aLog[iRegion*2] ){
      nMax += 1 + aLog[iRegion*2+1] - aLog[iRegion*2+0];
    }
  }

  /* Allocate space to collect all page numbers. */
  aPgno = (u32*)sqlite4_malloc(pLog->pLock->pEnv, sizeof(u32)*nMax*2);
  if( aPgno==0 ) rc = btErrorBkpt(SQLITE4_NOMEM);
  aSpace = &aPgno[nMax];

  /* Copy the required page numbers into the allocated array */
  for(iRegion=0; iRegion<3; iRegion++){
    u32 iFirst = aLog[iRegion*2];
    u32 iLast = aLog[iRegion*2+1];
    if( iFirst ){
      for(i=iFirst; rc==SQLITE4_OK && i<=iLast; i++){
        int iHash = btLogFrameHash(pLog, i);
        u32 *aPage;
        ht_slot *aHash;
        u32 iZero;

        /* It doesn't matter which 'side' of the hash table is requested here,
        ** as only the page-number array, not the aHash[] table, will be used.
        ** And it is the same for both sides. Hence the constant 0 passed as
        ** the second argument to btLogFindHash().  */
        rc = btLogFindHash(pLog, 0, iHash, &aHash, &aPage, &iZero);
        if( rc==SQLITE4_OK ){
          aPgno[i-iFirst] = aPage[i-iZero];
        }
      }
      *piLastFrame = iLast;
    }
  }

  /* Sort the contents of the array in ascending order. This step also 
  ** eliminates any  duplicate page numbers. */
  if( rc==SQLITE4_OK ){
    btLogMergeSort(aPgno, &nMax, aSpace);

  /* Take the CHECKPOINTER lock. */
  rc = sqlite4BtLockCkpt(pLock);
  if( rc==SQLITE4_OK ){
    const int pgsz = sqlite4BtPagerPagesize((BtPager*)pLock);
    bt_env *pVfs = pLock->pVfs;
    bt_file *pFd = pLock->pFd;
    BtShm *pShm;                  /* Pointer to shared-memory region */

    u32 iLast;                    /* Last frame to checkpoint */
    BtFrameHdr fhdr;              /* Frame header of frame iLast */
    u32 *aPgno = 0;               /* Array of page numbers to checkpoint */
    int nPgno;                    /* Number of entries in aPgno[] */
    int i;                        /* Used to loop through aPgno[] */
    u8 *aBuf;                     /* Buffer to load page data into */
    u32 iFirstRead;               /* First frame not checkpointed */

    /* Allocate space to load log data into */
    aBuf = sqlite4_malloc(pLock->pEnv, pgsz);
    if( aBuf==0 ) rc = btErrorBkpt(SQLITE4_NOMEM);
    
    /* Figure out the set of page numbers stored in the part of the log 
    ** file being checkpointed. Remove any duplicates and sort them in 
    ** ascending order.  */
    if( rc==SQLITE4_OK ){



      rc = btLogGatherPgno(pLog, &aPgno, &nPgno, &iLast);
    }
    if( rc==SQLITE4_OK ){
      i64 iOff = btLogFrameOffset(pLog, pgsz, iLast);
      rc = btLogReadData(pLog, iOff, (u8*)&fhdr, sizeof(BtFrameHdr));
      iFirstRead = (fhdr.ctrl & ~BT_FRAME_COMMIT);
    }

    /* Copy data from the log file to the database file. */
    for(i=0; rc==SQLITE4_OK && i<nPgno; i++){
      u32 pgno = aPgno[i];
      rc = sqlite4BtLogRead(pLog, pgno, aBuf);
      if( rc==SQLITE4_OK ){
        i64 iOff = (i64)pgsz * (pgno-1);
        rc = pVfs->xWrite(pFd, iOff, aBuf, pgsz);
      }
    }

    /* Update the first field of the checkpoint-header. This tells readers
    ** that they need not consider anything that in the log before this
    ** point (since the data has already been copied into the database
    ** file).  */
    if( rc==SQLITE4_OK ){
      pShm = btLogShm(pLog);
      pShm->ckpt.iFirstRead = iFirstRead;
      pVfs->xShmBarrier(pLog->pFd);
    }

    /* Write a new header into the log file. This tells any future recovery
    ** where it should start reading the log. Once this new header is synced
    ** to disk, the space cleared by this checkpoint operation can be 
    ** reused.  */
    if( rc==SQLITE4_OK ){
      int iSlot = ((pShm->ckpt.iWalHdr >> 2) + 1) % 2;

      BtWalHdr hdr;


      memset(&hdr, 0, sizeof(BtWalHdr));
      hdr.iMagic = BT_WAL_MAGIC;
      hdr.iVersion = BT_WAL_VERSION;
      hdr.iCnt = (((pShm->ckpt.iWalHdr & 0x03) + 1) % 3);
      hdr.nSector = pLog->snapshot.nSector;
      hdr.nPgsz = pgsz;
      hdr.iFirstFrame = iFirstRead;




      hdr.iSalt1 = fhdr.aCksum[0];
      hdr.iSalt2 = fhdr.aCksum[1];
      rc = btLogWriteHeader(pLog, iSlot, &hdr);

      if( rc==SQLITE4_OK ){
        pShm->ckpt.iWalHdr = (iSlot<<2) + hdr.iCnt;
      }
    }

    /* Update the second field of the checkpoint header. This tells future
    ** writers that it is now safe to recycle pages before this point
    ** (assuming all live readers are cleared).  */
    if( rc==SQLITE4_OK ){
      pShm->ckpt.iFirstRecover = iFirstRead;
      pVfs->xShmBarrier(pLog->pFd);
    }

    /* Free the buffer and drop the checkpointer lock */
    sqlite4_free(pLock->pEnv, aBuf);
    sqlite4BtLockCkptUnlock(pLog->pLock);
  }
  return rc;
}