SQLite4
Check-in [ec6dda0ade]
Not logged in

Many hyperlinks are disabled.
Use anonymous login to enable hyperlinks.

Overview
Comment:Begin adding code to make the log file circular.
Downloads: Tarball | ZIP archive | SQL archive
Timelines: family | ancestors | descendants | both | trunk
Files: files | file ages | folders
SHA1: ec6dda0ade61f1511404fb0f4145d8ccbc0a5eb0
User & Date: dan 2013-10-26 19:58:34
Context
2013-10-28
10:47
Fix some problems with circular logs. check-in: 4b36a0245f user: dan tags: trunk
2013-10-26
19:58
Begin adding code to make the log file circular. check-in: ec6dda0ade user: dan tags: trunk
12:44
Use sqlite4_malloc instead of malloc in test fixture for num check-in: 3f6924e784 user: peterreid tags: trunk
Changes
Hide Diffs Unified Diffs Ignore Whitespace Patch

Changes to src/bt_log.c.

17
18
19
20
21
22
23




24
25
26
27
28
29
30
..
47
48
49
50
51
52
53





54
55
56
57
58
59
60
..
63
64
65
66
67
68
69

70
71


72
73
74
75
76
77
78
...
105
106
107
108
109
110
111
112
113
114

115
116
117
118
119
120
121
...
151
152
153
154
155
156
157








158
159
160
161
162
163
164
...
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349

350
351
352
353
354
355


356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
...
404
405
406
407
408
409
410

411
412
413
414
415
416
417
418
419
420
421
422




423
424
425
426
427
428
429
...
437
438
439
440
441
442
443

444
445
446
447
448
449
450
451
452
453
...
460
461
462
463
464
465
466
467
468






469
470
471
472
473
474
475


























476
477

478

479
480
481
482
483
484
485
486
487



































488
489
490
491
492
493
494
...
521
522
523
524
525
526
527
528
529
530
531









532

533





534
535
536

537

538
539
540

541
542
543
544
545
546
547
...
561
562
563
564
565
566
567

568
569
570
571
572
573
574
...
627
628
629
630
631
632
633
634






635
636
637
638
639
640
641
642
643
644
645
646
647
...
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680

681








682
683


684
685




686
687
688
689
690
691
692
...
707
708
709
710
711
712
713


714
715
716
717
718
719
720
...
727
728
729
730
731
732
733

734
735
736

737
738
739

























740
741
742
743
744
745
746
747
748
749
750
751
752
753
754

755




756
757
758
759
760
761



762
763
764
765
766
767
768
769
770











771


772

773
774
775
776
777
778
779
...
833
834
835
836
837
838
839


840
841
842







843




844




845
846
847
848
849
850
851
...
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919

920

921
922
923
924
925










926
927
928
929
930
931
932




933

934
935
936





937
938
939



940
941
942
943
944
945
946
...
970
971
972
973
974
975
976
977
978

979
980
981
982

983
984
985
986
987
988
989
990
991
992
993
994
995


996
997
998
999
1000
1001
1002
....
1006
1007
1008
1009
1010
1011
1012

1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
#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





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.





*/
struct BtFrameHdr {
  u32 pgno;                       /* Page number of this frame */
  u32 ctrl;                       /* Next frame pointer and commit bit */
  u32 aCksum[2];                  /* Frame checksum */
};

................................................................................
/*
** 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 */

  u32 aFrameCksum[2];             /* Checksum of previous frame */
  int padding;


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

};

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) \
)









/*
** 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++;
  }

  return rc;
}

/*
** Locate the iHash'th hash table in shared memory. Return it.
*/
static int btLogFindHash(
  BtLog *pLog,                    /* Log handle */

  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;



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

    *paHash = (ht_slot*)&aChunk[HASHTABLE_OFFSET_1];
    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 */


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

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

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





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

  rc = btLogFindHash(pLog, 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;
................................................................................


/*
** 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);
}







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 */
){


























  btLogHashInsert(pLog, pHdr->pgno, iFrame);
  if( pHdr->ctrl & BT_FRAME_COMMIT ){

    *(u32*)pCtx = iFrame;

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




































/*
** 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, scan the log file
  ** and populate the shared-memory hash tables.  */
  if( pHdr ){
    u32 iCommit = 0;









    rc = btLogTraverse(pLog, pHdr, btLogRecoverFrame, (void*)&iCommit);

    if( rc==SQLITE4_OK && iCommit>0 ){





      BtShm *pShm = btLogShm(pLog);
      pShm->ckpt.iFirstRead = 1;
      pShm->ckpt.iWalHdr = (iSlot<<2) + pHdr->iCnt;

      pShm->ckpt.iFirstRecover = 1;

      pLog->snapshot.aLog[4] = 1;
      pLog->snapshot.aLog[5] = iCommit;
      rc = btLogHashRollback(pLog, btLogFrameHash(pLog, iCommit), iCommit);

    }
  }

  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;


  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, int iHash, u32 pgno, u32 *piFrame){






  ht_slot *aHash;
  u32 *aPgno;
  u32 iZero;
  int rc;

  rc = btLogFindHash(pLog, 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 iHash;
  int iHashLast;
  int rc = SQLITE4_NOTFOUND;
  u32 iFrame = 0;










  iHash = btLogFrameHash(pLog, pLog->snapshot.aLog[5]);
  iHashLast = btLogFrameHash(pLog, pLog->snapshot.aLog[4]);


  for( ; rc==SQLITE4_NOTFOUND && iHash>=iHashLast && iFrame==0; iHash--){
    rc = btLogHashSearch(pLog, iHash, pgno, &iFrame);




  }

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



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

  }

  /* Figure out where exactly to write the new data */

  iFrame = pLog->snapshot.aLog[5] + 1;
  iOff = btLogFrameOffset(pLog, pgsz, iFrame);


























  /* Populate the frame header object. */
  memset(&frame, 0, sizeof(frame));
  frame.pgno = pgno;
  frame.ctrl = (bCommit ? BT_FRAME_COMMIT : 0);
  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 header and page record to the log file. */

  rc = btLogWriteData(pLog, iOff, (u8*)&frame, sizeof(frame));




  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 ){



    rc = btLogHashInsert(pLog, pgno, iFrame);
  }

  /* Update the private copy of the shm-header */
  if( rc==SQLITE4_OK ){
    pLog->snapshot.aLog[5] = iFrame;
    if( btLogIsEmpty(pLog) ){
      assert( iFrame==1 );
      pLog->snapshot.aLog[4] = iFrame;











    }


    memcpy(pLog->snapshot.aFrameCksum, frame.aCksum, sizeof(frame.aCksum));

  }

  /* 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 ){


    BtShm *pShm = btLogShm(pLog);
    u32 iCkpt = pShm->ckpt.iFirstRead;
    if( iCkpt>pLog->snapshot.aLog[4] ){







      pLog->snapshot.aLog[4] = iCkpt;




    }




  }

  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 iFirst,                     /* First frame to gather pgnos from */
  u32 iLast,                      /* Last frame to gather pgnos from */
  u32 **paPgno,                   /* OUT: s4_malloc'd array of sorted pgno */
  int *pnPgno                     /* OUT: Number of entries in *paPgno */

){

  u32 i;
  u32 *aPgno;                     /* Returned array */
  u32 *aSpace;                    /* Temporary space used by merge-sort */
  int nMax = (iLast - iFirst) + 1;
  int rc = SQLITE4_OK;











  /* 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(i=iFirst; rc==SQLITE4_OK && i<=iLast; i++){

    u32 *aPage;
    ht_slot *aHash;
    u32 iZero;





    rc = btLogFindHash(pLog, btLogFrameHash(pLog, i), &aHash, &aPage, &iZero);
    if( rc==SQLITE4_OK ){
      aPgno[i-iFirst] = aPage[i-iZero];



    }
  }

  /* 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 iFirst;                   /* First frame to checkpoint */
    u32 iLast;                    /* Last frame to checkpoint */

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


    /* 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 ){
      pShm = btLogShm(pLog);
      iFirst = pShm->ckpt.iFirstRead;
      iLast = pLog->snapshot.aLog[5];
      rc = btLogGatherPgno(pLog, iFirst, iLast, &aPgno, &nPgno);


    }

    /* 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 ){
................................................................................
    }

    /* 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->ckpt.iFirstRead = iLast+1;
      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;
      i64 iOff;
      BtWalHdr hdr;
      BtFrameHdr fhdr;

      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 = iLast+1;

      iOff = btLogFrameOffset(pLog, pgsz, iLast);
      rc = btLogReadData(pLog, iOff, (u8*)&fhdr, sizeof(BtFrameHdr));
      if( rc==SQLITE4_OK ){
        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 = iLast+1;
      pVfs->xShmBarrier(pLog->pFd);
    }

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









>
>
>
>







 







>
>
>
>
>







 







>

<
>
>







 







<


>







 







>
>
>
>
>
>
>
>







 







|










>





|
>
>







|







 







>





|






>
>
>
>







 







>


|







 







|

>
>
>
>
>
>







>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>


>
|
>









>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>







 







|
|

<
>
>
>
>
>
>
>
>
>
|
>
|
>
>
>
>
>
|
<
|
>
|
>
|
<
<
>







 







>







 







|
>
>
>
>
>
>





|







 







<













<
<


>

>
>
>
>
>
>
>
>
|
|
>
>
|
|
>
>
>
>







 







>
>







 







>


<
>
|


>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
|
|
|
|
|
|
|


|
|


<
<
>
|
>
>
>
>






>
>
>





|


|
>
>
>
>
>
>
>
>
>
>
>

>
>

>







 







>
>

|
<
>
>
>
>
>
>
>
|
>
>
>
>
|
>
>
>
>







 







<



<
<

|
>

>



|

>
>
>
>
>
>
>
>
>
>







>
>
>
>
|
>
|
|
|
>
>
>
>
>
|
|
|
>
>
>







 







<

>




>









|
|
|
|
>
>







 







>
|









<

<







|

<
<
<
|
|
|
<









|











17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
..
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
..
72
73
74
75
76
77
78
79
80

81
82
83
84
85
86
87
88
89
...
116
117
118
119
120
121
122

123
124
125
126
127
128
129
130
131
132
...
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
...
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
...
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
...
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
...
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
...
618
619
620
621
622
623
624
625
626
627

628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645

646
647
648
649
650


651
652
653
654
655
656
657
658
...
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
...
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
...
774
775
776
777
778
779
780

781
782
783
784
785
786
787
788
789
790
791
792
793


794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
...
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
...
859
860
861
862
863
864
865
866
867
868

869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910


911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
....
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021

1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
....
1099
1100
1101
1102
1103
1104
1105

1106
1107
1108


1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
....
1186
1187
1188
1189
1190
1191
1192

1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
....
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242

1243

1244
1245
1246
1247
1248
1249
1250
1251
1252



1253
1254
1255

1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
#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 */
};

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


/*
** 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.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 ){
................................................................................
    }

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