Adding upstream version 3.45.1.upstream/3.45.1

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 1156 insertions, 0 deletions

diff --git a/ext/lsm1/lsm_log.c b/ext/lsm1/lsm_log.c
new file mode 100644
index 0000000..a66e40b
--- /dev/null
+++ b/ext/lsm1/lsm_log.c
@@ -0,0 +1,1156 @@
+/*
+** 2011-08-13
+**
+** The author disclaims copyright to this source code.  In place of
+** a legal notice, here is a blessing:
+**
+**    May you do good and not evil.
+**    May you find forgiveness for yourself and forgive others.
+**    May you share freely, never taking more than you give.
+**
+*************************************************************************
+**
+** This file contains the implementation of LSM database logging. Logging
+** has one purpose in LSM - to make transactions durable.
+**
+** When data is written to an LSM database, it is initially stored in an
+** in-memory tree structure. Since this structure is in volatile memory,
+** if a power failure or application crash occurs it may be lost. To
+** prevent loss of data in this case, each time a record is written to the
+** in-memory tree an equivalent record is appended to the log on disk.
+** If a power failure or application crash does occur, data can be recovered
+** by reading the log.
+**
+** A log file consists of the following types of records representing data
+** written into the database:
+**
+**   LOG_WRITE:  A key-value pair written to the database.
+**   LOG_DELETE: A delete key issued to the database.
+**   LOG_COMMIT: A transaction commit.
+**
+** And the following types of records for ancillary purposes..
+**
+**   LOG_EOF:    A record indicating the end of a log file.
+**   LOG_PAD1:   A single byte padding record.
+**   LOG_PAD2:   An N byte padding record (N>1).
+**   LOG_JUMP:   A pointer to another offset within the log file.
+**
+** Each transaction written to the log contains one or more LOG_WRITE and/or
+** LOG_DELETE records, followed by a LOG_COMMIT record. The LOG_COMMIT record
+** contains an 8-byte checksum based on all previous data written to the
+** log file.
+**
+** LOG CHECKSUMS & RECOVERY
+**
+**   Checksums are found in two types of log records: LOG_COMMIT and
+**   LOG_CKSUM records. In order to recover content from a log, a client
+**   reads each record from the start of the log, calculating a checksum as
+**   it does. Each time a LOG_COMMIT or LOG_CKSUM is encountered, the 
+**   recovery process verifies that the checksum stored in the log 
+**   matches the calculated checksum. If it does not, the recovery process
+**   can stop reading the log.
+**
+**   If a recovery process reads records (other than COMMIT or CKSUM) 
+**   consisting of at least LSM_CKSUM_MAXDATA bytes, then the next record in
+**   the log must be either a LOG_CKSUM or LOG_COMMIT record. If it is
+**   not, the recovery process also stops reading the log.
+**
+**   To recover the log file, it must be read twice. The first time to 
+**   determine the location of the last valid commit record. And the second
+**   time to load data into the in-memory tree.
+**
+**   Todo: Surely there is a better way...
+**
+** LOG WRAPPING
+**
+**   If the log file were never deleted or wrapped, it would be possible to
+**   read it from start to end each time is required recovery (i.e each time
+**   the number of database clients changes from 0 to 1). Effectively reading
+**   the entire history of the database each time. This would quickly become 
+**   inefficient. Additionally, since the log file would grow without bound,
+**   it wastes storage space.
+**
+**   Instead, part of each checkpoint written into the database file contains 
+**   a log offset (and other information required to read the log starting at
+**   at this offset) at which to begin recovery. Offset $O.
+**
+**   Once a checkpoint has been written and synced into the database file, it
+**   is guaranteed that no recovery process will need to read any data before
+**   offset $O of the log file. It is therefore safe to begin overwriting
+**   any data that occurs before offset $O.
+**
+**   This implementation separates the log into three regions mapped into
+**   the log file - regions 0, 1 and 2. During recovery, regions are read
+**   in ascending order (i.e. 0, then 1, then 2). Each region is zero or
+**   more bytes in size.
+**
+**     |---1---|..|--0--|.|--2--|....
+**
+**   New records are always appended to the end of region 2.
+**
+**   Initially (when it is empty), all three regions are zero bytes in size.
+**   Each of them are located at the beginning of the file. As records are
+**   added to the log, region 2 grows, so that the log consists of a zero
+**   byte region 1, followed by a zero byte region 0, followed by an N byte
+**   region 2. After one or more checkpoints have been written to disk, 
+**   the start point of region 2 is moved to $O. For example:
+**
+**     A) ||.........|--2--|....
+**   
+**   (both regions 0 and 1 are 0 bytes in size at offset 0).
+**
+**   Eventually, the log wraps around to write new records into the start.
+**   At this point, region 2 is renamed to region 0. Region 0 is renamed
+**   to region 2. After appending a few records to the new region 2, the
+**   log file looks like this:
+**
+**     B) ||--2--|...|--0--|....
+**
+**   (region 1 is still 0 bytes in size, located at offset 0).
+**
+**   Any checkpoints made at this point may reduce the size of region 0.
+**   However, if they do not, and region 2 expands so that it is about to
+**   overwrite the start of region 0, then region 2 is renamed to region 1,
+**   and a new region 2 created at the end of the file following the existing
+**   region 0.
+**
+**     C) |---1---|..|--0--|.|-2-|
+**
+**   In this state records are appended to region 2 until checkpoints have
+**   contracted regions 0 AND 1 UNTil they are both zero bytes in size. They 
+**   are then shifted to the start of the log file, leaving the system in 
+**   the equivalent of state A above.
+**
+**   Alternatively, state B may transition directly to state A if the size
+**   of region 0 is reduced to zero bytes before region 2 threatens to 
+**   encroach upon it.
+**
+** LOG_PAD1 & LOG_PAD2 RECORDS
+**
+**   PAD1 and PAD2 records may appear in a log file at any point. They allow
+**   a process writing the log file align the beginning of transactions with 
+**   the beginning of disk sectors, which increases robustness.
+**
+** RECORD FORMATS:
+**
+**   LOG_EOF:    * A single 0x00 byte.
+**
+**   LOG_PAD1:   * A single 0x01 byte.
+**
+**   LOG_PAD2:   * A single 0x02 byte, followed by
+**               * The number of unused bytes (N) as a varint,
+**               * An N byte block of unused space.
+**
+**   LOG_COMMIT: * A single 0x03 byte.
+**               * An 8-byte checksum.
+**
+**   LOG_JUMP:   * A single 0x04 byte.
+**               * Absolute file offset to jump to, encoded as a varint.
+**
+**   LOG_WRITE:  * A single 0x06 or 0x07 byte, 
+**               * The number of bytes in the key, encoded as a varint, 
+**               * The number of bytes in the value, encoded as a varint, 
+**               * If the first byte was 0x07, an 8 byte checksum.
+**               * The key data,
+**               * The value data.
+**
+**   LOG_DELETE: * A single 0x08 or 0x09 byte, 
+**               * The number of bytes in the key, encoded as a varint, 
+**               * If the first byte was 0x09, an 8 byte checksum.
+**               * The key data.
+**
+**   Varints are as described in lsm_varint.c (SQLite 4 format).
+**
+** CHECKSUMS:
+**
+**   The checksum is calculated using two 32-bit unsigned integers, s0 and
+**   s1. The initial value for both is 42. It is updated each time a record
+**   is written into the log file by treating the encoded (binary) record as 
+**   an array of 32-bit little-endian integers. Then, if x[] is the integer
+**   array, updating the checksum accumulators as follows:
+**
+**     for i from 0 to n-1 step 2:
+**       s0 += x[i] + s1;
+**       s1 += x[i+1] + s0;
+**     endfor
+**
+**   If the record is not an even multiple of 8-bytes in size it is padded
+**   with zeroes to make it so before the checksum is updated.
+**
+**   The checksum stored in a COMMIT, WRITE or DELETE is based on all bytes
+**   up to the start of the 8-byte checksum itself, including the COMMIT,
+**   WRITE or DELETE fields that appear before the checksum in the record.
+**
+** VARINT FORMAT
+**
+** See lsm_varint.c.
+*/
+
+#ifndef _LSM_INT_H
+# include "lsmInt.h"
+#endif
+
+/* Log record types */
+#define LSM_LOG_EOF          0x00
+#define LSM_LOG_PAD1         0x01
+#define LSM_LOG_PAD2         0x02
+#define LSM_LOG_COMMIT       0x03
+#define LSM_LOG_JUMP         0x04
+
+#define LSM_LOG_WRITE        0x06
+#define LSM_LOG_WRITE_CKSUM  0x07
+
+#define LSM_LOG_DELETE       0x08
+#define LSM_LOG_DELETE_CKSUM 0x09
+
+#define LSM_LOG_DRANGE       0x0A
+#define LSM_LOG_DRANGE_CKSUM 0x0B
+
+/* Require a checksum every 32KB. */
+#define LSM_CKSUM_MAXDATA (32*1024)
+
+/* Do not wrap a log file smaller than this in bytes. */
+#define LSM_MIN_LOGWRAP      (128*1024)
+
+/*
+** szSector:
+**   Commit records must be aligned to end on szSector boundaries. If
+**   the safety-mode is set to NORMAL or OFF, this value is 1. Otherwise,
+**   if the safety-mode is set to FULL, it is the size of the file-system
+**   sectors as reported by lsmFsSectorSize().
+*/
+struct LogWriter {
+  u32 cksum0;                     /* Checksum 0 at offset iOff */
+  u32 cksum1;                     /* Checksum 1 at offset iOff */
+  int iCksumBuf;                  /* Bytes of buf that have been checksummed */
+  i64 iOff;                       /* Offset at start of buffer buf */
+  int szSector;                   /* Sector size for this transaction */
+  LogRegion jump;                 /* Avoid writing to this region */
+  i64 iRegion1End;                /* End of first region written by trans */
+  i64 iRegion2Start;              /* Start of second regions written by trans */
+  LsmString buf;                  /* Buffer containing data not yet written */
+};
+
+/*
+** Return the result of interpreting the first 4 bytes in buffer aIn as 
+** a 32-bit unsigned little-endian integer.
+*/
+static u32 getU32le(u8 *aIn){
+  return ((u32)aIn[3] << 24) 
+       + ((u32)aIn[2] << 16) 
+       + ((u32)aIn[1] << 8) 
+       + ((u32)aIn[0]);
+}
+
+
+/*
+** This function is the same as logCksum(), except that pointer "a" need
+** not be aligned to an 8-byte boundary or padded with zero bytes. This
+** version is slower, but sometimes more convenient to use.
+*/
+static void logCksumUnaligned(
+  char *z,                        /* Input buffer */
+  int n,                          /* Size of input buffer in bytes */
+  u32 *pCksum0,                   /* IN/OUT: Checksum value 1 */
+  u32 *pCksum1                    /* IN/OUT: Checksum value 2 */
+){
+  u8 *a = (u8 *)z;
+  u32 cksum0 = *pCksum0;
+  u32 cksum1 = *pCksum1;
+  int nIn = (n/8) * 8;
+  int i;
+
+  assert( n>0 );
+  for(i=0; i<nIn; i+=8){
+    cksum0 += getU32le(&a[i]) + cksum1;
+    cksum1 += getU32le(&a[i+4]) + cksum0;
+  }
+
+  if( nIn!=n ){
+    u8 aBuf[8] = {0, 0, 0, 0, 0, 0, 0, 0};
+    assert( (n-nIn)<8 && n>nIn );
+    memcpy(aBuf, &a[nIn], n-nIn);
+    cksum0 += getU32le(aBuf) + cksum1;
+    cksum1 += getU32le(&aBuf[4]) + cksum0;
+  }
+
+  *pCksum0 = cksum0;
+  *pCksum1 = cksum1;
+}
+
+/*
+** Update pLog->cksum0 and pLog->cksum1 so that the first nBuf bytes in the 
+** write buffer (pLog->buf) are included in the checksum.
+*/
+static void logUpdateCksum(LogWriter *pLog, int nBuf){
+  assert( (pLog->iCksumBuf % 8)==0 );
+  assert( pLog->iCksumBuf<=nBuf );
+  assert( (nBuf % 8)==0 || nBuf==pLog->buf.n );
+  if( nBuf>pLog->iCksumBuf ){
+    logCksumUnaligned(
+        &pLog->buf.z[pLog->iCksumBuf], nBuf-pLog->iCksumBuf, 
+        &pLog->cksum0, &pLog->cksum1
+    );
+  }
+  pLog->iCksumBuf = nBuf;
+}
+
+static i64 firstByteOnSector(LogWriter *pLog, i64 iOff){
+  return (iOff / pLog->szSector) * pLog->szSector;
+}
+static i64 lastByteOnSector(LogWriter *pLog, i64 iOff){
+  return firstByteOnSector(pLog, iOff) + pLog->szSector - 1;
+}
+
+/*
+** If possible, reclaim log file space. Log file space is reclaimed after
+** a snapshot that points to the same data in the database file is synced
+** into the db header.
+*/
+static int logReclaimSpace(lsm_db *pDb){
+  int rc;
+  int iMeta;
+  int bRotrans;                   /* True if there exists some ro-trans */
+
+  /* Test if there exists some other connection with a read-only transaction
+  ** open. If there does, then log file space may not be reclaimed.  */
+  rc = lsmDetectRoTrans(pDb, &bRotrans);
+  if( rc!=LSM_OK || bRotrans ) return rc;
+
+  iMeta = (int)pDb->pShmhdr->iMetaPage;
+  if( iMeta==1 || iMeta==2 ){
+    DbLog *pLog = &pDb->treehdr.log;
+    i64 iSyncedId;
+
+    /* Read the snapshot-id of the snapshot stored on meta-page iMeta. Note
+    ** that in theory, the value read is untrustworthy (due to a race 
+    ** condition - see comments above lsmFsReadSyncedId()). So it is only 
+    ** ever used to conclude that no log space can be reclaimed. If it seems
+    ** to indicate that it may be possible to reclaim log space, a
+    ** second call to lsmCheckpointSynced() (which does return trustworthy
+    ** values) is made below to confirm.  */
+    rc = lsmFsReadSyncedId(pDb, iMeta, &iSyncedId);
+
+    if( rc==LSM_OK && pLog->iSnapshotId!=iSyncedId ){
+      i64 iSnapshotId = 0;
+      i64 iOff = 0;
+      rc = lsmCheckpointSynced(pDb, &iSnapshotId, &iOff, 0);
+      if( rc==LSM_OK && pLog->iSnapshotId<iSnapshotId ){
+        int iRegion;
+        for(iRegion=0; iRegion<3; iRegion++){
+          LogRegion *p = &pLog->aRegion[iRegion];
+          if( iOff>=p->iStart && iOff<=p->iEnd ) break;
+          p->iStart = 0;
+          p->iEnd = 0;
+        }
+        assert( iRegion<3 );
+        pLog->aRegion[iRegion].iStart = iOff;
+        pLog->iSnapshotId = iSnapshotId;
+      }
+    }
+  }
+  return rc;
+}
+
+/*
+** This function is called when a write-transaction is first opened. It
+** is assumed that the caller is holding the client-mutex when it is 
+** called.
+**
+** Before returning, this function allocates the LogWriter object that
+** will be used to write to the log file during the write transaction.
+** LSM_OK is returned if no error occurs, otherwise an LSM error code.
+*/
+int lsmLogBegin(lsm_db *pDb){
+  int rc = LSM_OK;
+  LogWriter *pNew;
+  LogRegion *aReg;
+
+  if( pDb->bUseLog==0 ) return LSM_OK;
+
+  /* If the log file has not yet been opened, open it now. Also allocate
+  ** the LogWriter structure, if it has not already been allocated.  */
+  rc = lsmFsOpenLog(pDb, 0);
+  if( pDb->pLogWriter==0 ){
+    pNew = lsmMallocZeroRc(pDb->pEnv, sizeof(LogWriter), &rc);
+    if( pNew ){
+      lsmStringInit(&pNew->buf, pDb->pEnv);
+      rc = lsmStringExtend(&pNew->buf, 2);
+    }
+    pDb->pLogWriter = pNew;
+  }else{
+    pNew = pDb->pLogWriter;
+    assert( (u8 *)(&pNew[1])==(u8 *)(&((&pNew->buf)[1])) );
+    memset(pNew, 0, ((u8 *)&pNew->buf) - (u8 *)pNew);
+    pNew->buf.n = 0;
+  }
+
+  if( rc==LSM_OK ){
+    /* The following call detects whether or not a new snapshot has been 
+    ** synced into the database file. If so, it updates the contents of
+    ** the pDb->treehdr.log structure to reclaim any space in the log
+    ** file that is no longer required. 
+    **
+    ** TODO: Calling this every transaction is overkill. And since the 
+    ** call has to read and checksum a snapshot from the database file,
+    ** it is expensive. It would be better to figure out a way so that
+    ** this is only called occasionally - say for every 32KB written to 
+    ** the log file.
+    */
+    rc = logReclaimSpace(pDb);
+  }
+  if( rc!=LSM_OK ){
+    lsmLogClose(pDb);
+    return rc;
+  }
+
+  /* Set the effective sector-size for this transaction. Sectors are assumed
+  ** to be one byte in size if the safety-mode is OFF or NORMAL, or as
+  ** reported by lsmFsSectorSize if it is FULL.  */
+  if( pDb->eSafety==LSM_SAFETY_FULL ){
+    pNew->szSector = lsmFsSectorSize(pDb->pFS);
+    assert( pNew->szSector>0 );
+  }else{
+    pNew->szSector = 1;
+  }
+
+  /* There are now three scenarios:
+  **
+  **   1) Regions 0 and 1 are both zero bytes in size and region 2 begins
+  **      at a file offset greater than LSM_MIN_LOGWRAP. In this case, wrap
+  **      around to the start and write data into the start of the log file. 
+  **
+  **   2) Region 1 is zero bytes in size and region 2 occurs earlier in the 
+  **      file than region 0. In this case, append data to region 2, but
+  **      remember to jump over region 1 if required.
+  **
+  **   3) Region 2 is the last in the file. Append to it.
+  */
+  aReg = &pDb->treehdr.log.aRegion[0];
+
+  assert( aReg[0].iEnd==0 || aReg[0].iEnd>aReg[0].iStart );
+  assert( aReg[1].iEnd==0 || aReg[1].iEnd>aReg[1].iStart );
+
+  pNew->cksum0 = pDb->treehdr.log.cksum0;
+  pNew->cksum1 = pDb->treehdr.log.cksum1;
+
+  if( aReg[0].iEnd==0 && aReg[1].iEnd==0 && aReg[2].iStart>=LSM_MIN_LOGWRAP ){
+    /* Case 1. Wrap around to the start of the file. Write an LSM_LOG_JUMP 
+    ** into the log file in this case. Pad it out to 8 bytes using a PAD2
+    ** record so that the checksums can be updated immediately.  */
+    u8 aJump[] = { 
+      LSM_LOG_PAD2, 0x04, 0x00, 0x00, 0x00, 0x00, LSM_LOG_JUMP, 0x00 
+    };
+
+    lsmStringBinAppend(&pNew->buf, aJump, sizeof(aJump));
+    logUpdateCksum(pNew, pNew->buf.n);
+    rc = lsmFsWriteLog(pDb->pFS, aReg[2].iEnd, &pNew->buf);
+    pNew->iCksumBuf = pNew->buf.n = 0;
+
+    aReg[2].iEnd += 8;
+    pNew->jump = aReg[0] = aReg[2];
+    aReg[2].iStart = aReg[2].iEnd = 0;
+  }else if( aReg[1].iEnd==0 && aReg[2].iEnd<aReg[0].iEnd ){
+    /* Case 2. */
+    pNew->iOff = aReg[2].iEnd;
+    pNew->jump = aReg[0];
+  }else{
+    /* Case 3. */
+    assert( aReg[2].iStart>=aReg[0].iEnd && aReg[2].iStart>=aReg[1].iEnd );
+    pNew->iOff = aReg[2].iEnd;
+  }
+
+  if( pNew->jump.iStart ){
+    i64 iRound;
+    assert( pNew->jump.iStart>pNew->iOff );
+
+    iRound = firstByteOnSector(pNew, pNew->jump.iStart);
+    if( iRound>pNew->iOff ) pNew->jump.iStart = iRound;
+    pNew->jump.iEnd = lastByteOnSector(pNew, pNew->jump.iEnd);
+  }
+
+  assert( pDb->pLogWriter==pNew );
+  return rc;
+}
+
+/*
+** This function is called when a write-transaction is being closed.
+** Parameter bCommit is true if the transaction is being committed,
+** or false otherwise. The caller must hold the client-mutex to call
+** this function.
+**
+** A call to this function deletes the LogWriter object allocated by
+** lsmLogBegin(). If the transaction is being committed, the shared state
+** in *pLog is updated before returning.
+*/
+void lsmLogEnd(lsm_db *pDb, int bCommit){
+  DbLog *pLog;
+  LogWriter *p;
+  p = pDb->pLogWriter;
+
+  if( p==0 ) return;
+  pLog = &pDb->treehdr.log;
+
+  if( bCommit ){
+    pLog->aRegion[2].iEnd = p->iOff;
+    pLog->cksum0 = p->cksum0;
+    pLog->cksum1 = p->cksum1;
+    if( p->iRegion1End ){
+      /* This happens when the transaction had to jump over some other
+      ** part of the log.  */
+      assert( pLog->aRegion[1].iEnd==0 );
+      assert( pLog->aRegion[2].iStart<p->iRegion1End );
+      pLog->aRegion[1].iStart = pLog->aRegion[2].iStart;
+      pLog->aRegion[1].iEnd = p->iRegion1End;
+      pLog->aRegion[2].iStart = p->iRegion2Start;
+    }
+  }
+}
+
+static int jumpIfRequired(
+  lsm_db *pDb,
+  LogWriter *pLog,
+  int nReq,
+  int *pbJump
+){
+  /* Determine if it is necessary to add an LSM_LOG_JUMP to jump over the
+  ** jump region before writing the LSM_LOG_WRITE or DELETE record. This
+  ** is necessary if there is insufficient room between the current offset
+  ** and the jump region to fit the new WRITE/DELETE record and the largest
+  ** possible JUMP record with up to 7 bytes of padding (a total of 17 
+  ** bytes).  */
+  if( (pLog->jump.iStart > (pLog->iOff + pLog->buf.n))
+   && (pLog->jump.iStart < (pLog->iOff + pLog->buf.n + (nReq + 17))) 
+  ){
+    int rc;                       /* Return code */
+    i64 iJump;                    /* Offset to jump to */
+    u8 aJump[10];                 /* Encoded jump record */
+    int nJump;                    /* Valid bytes in aJump[] */
+    int nPad;                     /* Bytes of padding required */
+
+    /* Serialize the JUMP record */
+    iJump = pLog->jump.iEnd+1;
+    aJump[0] = LSM_LOG_JUMP;
+    nJump = 1 + lsmVarintPut64(&aJump[1], iJump);
+
+    /* Adding padding to the contents of the buffer so that it will be a 
+    ** multiple of 8 bytes in size after the JUMP record is appended. This
+    ** is not strictly required, it just makes the keeping the running 
+    ** checksum up to date in this file a little simpler.  */
+    nPad = (pLog->buf.n + nJump) % 8;
+    if( nPad ){
+      u8 aPad[7] = {0,0,0,0,0,0,0};
+      nPad = 8-nPad;
+      if( nPad==1 ){
+        aPad[0] = LSM_LOG_PAD1;
+      }else{
+        aPad[0] = LSM_LOG_PAD2;
+        aPad[1] = (u8)(nPad-2);
+      }
+      rc = lsmStringBinAppend(&pLog->buf, aPad, nPad);
+      if( rc!=LSM_OK ) return rc;
+    }
+
+    /* Append the JUMP record to the buffer. Then flush the buffer to disk
+    ** and update the checksums. The next write to the log file (assuming
+    ** there is no transaction rollback) will be to offset iJump (just past
+    ** the jump region).  */
+    rc = lsmStringBinAppend(&pLog->buf, aJump, nJump);
+    if( rc!=LSM_OK ) return rc;
+    assert( (pLog->buf.n % 8)==0 );
+    rc = lsmFsWriteLog(pDb->pFS, pLog->iOff, &pLog->buf);
+    if( rc!=LSM_OK ) return rc;
+    logUpdateCksum(pLog, pLog->buf.n);
+    pLog->iRegion1End = (pLog->iOff + pLog->buf.n);
+    pLog->iRegion2Start = iJump;
+    pLog->iOff = iJump;
+    pLog->iCksumBuf = pLog->buf.n = 0;
+    if( pbJump ) *pbJump = 1;
+  }
+
+  return LSM_OK;
+}
+
+static int logCksumAndFlush(lsm_db *pDb){
+  int rc;                         /* Return code */
+  LogWriter *pLog = pDb->pLogWriter;
+
+  /* Calculate the checksum value. Append it to the buffer. */
+  logUpdateCksum(pLog, pLog->buf.n);
+  lsmPutU32((u8 *)&pLog->buf.z[pLog->buf.n], pLog->cksum0);
+  pLog->buf.n += 4;
+  lsmPutU32((u8 *)&pLog->buf.z[pLog->buf.n], pLog->cksum1);
+  pLog->buf.n += 4;
+
+  /* Write the contents of the buffer to disk. */
+  rc = lsmFsWriteLog(pDb->pFS, pLog->iOff, &pLog->buf);
+  pLog->iOff += pLog->buf.n;
+  pLog->iCksumBuf = pLog->buf.n = 0;
+
+  return rc;
+}
+
+/*
+** Write the contents of the log-buffer to disk. Then write either a CKSUM
+** or COMMIT record, depending on the value of parameter eType.
+*/
+static int logFlush(lsm_db *pDb, int eType){
+  int rc;
+  int nReq;
+  LogWriter *pLog = pDb->pLogWriter;
+  
+  assert( eType==LSM_LOG_COMMIT );
+  assert( pLog );
+
+  /* Commit record is always 9 bytes in size. */
+  nReq = 9;
+  if( eType==LSM_LOG_COMMIT && pLog->szSector>1 ) nReq += pLog->szSector + 17;
+  rc = jumpIfRequired(pDb, pLog, nReq, 0);
+
+  /* If this is a COMMIT, add padding to the log so that the COMMIT record
+  ** is aligned against the end of a disk sector. In other words, add padding
+  ** so that the first byte following the COMMIT record lies on a different
+  ** sector.  */
+  if( eType==LSM_LOG_COMMIT && pLog->szSector>1 ){
+    int nPad;                     /* Bytes of padding to add */
+
+    /* Determine the value of nPad. */
+    nPad = ((pLog->iOff + pLog->buf.n + 9) % pLog->szSector);
+    if( nPad ) nPad = pLog->szSector - nPad;
+    rc = lsmStringExtend(&pLog->buf, nPad);
+    if( rc!=LSM_OK ) return rc;
+
+    while( nPad ){
+      if( nPad==1 ){
+        pLog->buf.z[pLog->buf.n++] = LSM_LOG_PAD1;
+        nPad = 0;
+      }else{
+        int n = LSM_MIN(200, nPad-2);
+        pLog->buf.z[pLog->buf.n++] = LSM_LOG_PAD2;
+        pLog->buf.z[pLog->buf.n++] = (char)n;
+        nPad -= 2;
+        memset(&pLog->buf.z[pLog->buf.n], 0x2B, n);
+        pLog->buf.n += n;
+        nPad -= n;
+      }
+    }
+  }
+
+  /* Make sure there is room in the log-buffer to add the CKSUM or COMMIT
+  ** record. Then add the first byte of it.  */
+  rc = lsmStringExtend(&pLog->buf, 9);
+  if( rc!=LSM_OK ) return rc;
+  pLog->buf.z[pLog->buf.n++] = (char)eType;
+  memset(&pLog->buf.z[pLog->buf.n], 0, 8);
+
+  rc = logCksumAndFlush(pDb);
+
+  /* If this is a commit and synchronous=full, sync the log to disk. */
+  if( rc==LSM_OK && eType==LSM_LOG_COMMIT && pDb->eSafety==LSM_SAFETY_FULL ){
+    rc = lsmFsSyncLog(pDb->pFS);
+  }
+  return rc;
+}
+
+/*
+** Append an LSM_LOG_WRITE (if nVal>=0) or LSM_LOG_DELETE (if nVal<0) 
+** record to the database log.
+*/
+int lsmLogWrite(
+  lsm_db *pDb,                    /* Database handle */
+  int eType,
+  void *pKey, int nKey,           /* Database key to write to log */
+  void *pVal, int nVal            /* Database value (or nVal<0) to write */
+){
+  int rc = LSM_OK;
+  LogWriter *pLog;                /* Log object to write to */
+  int nReq;                       /* Bytes of space required in log */
+  int bCksum = 0;                 /* True to embed a checksum in this record */
+
+  assert( eType==LSM_WRITE || eType==LSM_DELETE || eType==LSM_DRANGE );
+  assert( LSM_LOG_WRITE==LSM_WRITE );
+  assert( LSM_LOG_DELETE==LSM_DELETE );
+  assert( LSM_LOG_DRANGE==LSM_DRANGE );
+  assert( (eType==LSM_LOG_DELETE)==(nVal<0) );
+
+  if( pDb->bUseLog==0 ) return LSM_OK;
+  pLog = pDb->pLogWriter;
+
+  /* Determine how many bytes of space are required, assuming that a checksum
+  ** will be embedded in this record (even though it may not be).  */
+  nReq = 1 + lsmVarintLen32(nKey) + 8 + nKey;
+  if( eType!=LSM_LOG_DELETE ) nReq += lsmVarintLen32(nVal) + nVal;
+
+  /* Jump over the jump region if required. Set bCksum to true to tell the
+  ** code below to include a checksum in the record if either (a) writing
+  ** this record would mean that more than LSM_CKSUM_MAXDATA bytes of data
+  ** have been written to the log since the last checksum, or (b) the jump
+  ** is taken.  */
+  rc = jumpIfRequired(pDb, pLog, nReq, &bCksum);
+  if( (pLog->buf.n+nReq) > LSM_CKSUM_MAXDATA ) bCksum = 1;
+
+  if( rc==LSM_OK ){
+    rc = lsmStringExtend(&pLog->buf, nReq);
+  }
+  if( rc==LSM_OK ){
+    u8 *a = (u8 *)&pLog->buf.z[pLog->buf.n];
+    
+    /* Write the record header - the type byte followed by either 1 (for
+    ** DELETE) or 2 (for WRITE) varints.  */
+    assert( LSM_LOG_WRITE_CKSUM == (LSM_LOG_WRITE | 0x0001) );
+    assert( LSM_LOG_DELETE_CKSUM == (LSM_LOG_DELETE | 0x0001) );
+    assert( LSM_LOG_DRANGE_CKSUM == (LSM_LOG_DRANGE | 0x0001) );
+    *(a++) = (u8)eType | (u8)bCksum;
+    a += lsmVarintPut32(a, nKey);
+    if( eType!=LSM_LOG_DELETE ) a += lsmVarintPut32(a, nVal);
+
+    if( bCksum ){
+      pLog->buf.n = (a - (u8 *)pLog->buf.z);
+      rc = logCksumAndFlush(pDb);
+      a = (u8 *)&pLog->buf.z[pLog->buf.n];
+    }
+
+    memcpy(a, pKey, nKey);
+    a += nKey;
+    if( eType!=LSM_LOG_DELETE ){
+      memcpy(a, pVal, nVal);
+      a += nVal;
+    }
+    pLog->buf.n = a - (u8 *)pLog->buf.z;
+    assert( pLog->buf.n<=pLog->buf.nAlloc );
+  }
+
+  return rc;
+}
+
+/*
+** Append an LSM_LOG_COMMIT record to the database log.
+*/
+int lsmLogCommit(lsm_db *pDb){
+  if( pDb->bUseLog==0 ) return LSM_OK;
+  return logFlush(pDb, LSM_LOG_COMMIT);
+}
+
+/*
+** Store the current offset and other checksum related information in the
+** structure *pMark. Later, *pMark can be passed to lsmLogSeek() to "rewind"
+** the LogWriter object to the current log file offset. This is used when
+** rolling back savepoint transactions.
+*/
+void lsmLogTell(
+  lsm_db *pDb,                    /* Database handle */
+  LogMark *pMark                  /* Populate this object with current offset */
+){
+  LogWriter *pLog;
+  int nCksum;
+
+  if( pDb->bUseLog==0 ) return;
+  pLog = pDb->pLogWriter;
+  nCksum = pLog->buf.n & 0xFFFFFFF8;
+  logUpdateCksum(pLog, nCksum);
+  assert( pLog->iCksumBuf==nCksum );
+  pMark->nBuf = pLog->buf.n - nCksum;
+  memcpy(pMark->aBuf, &pLog->buf.z[nCksum], pMark->nBuf);
+
+  pMark->iOff = pLog->iOff + pLog->buf.n;
+  pMark->cksum0 = pLog->cksum0;
+  pMark->cksum1 = pLog->cksum1;
+}
+
+/*
+** Seek (rewind) back to the log file offset stored by an ealier call to
+** lsmLogTell() in *pMark.
+*/
+void lsmLogSeek(
+  lsm_db *pDb,                    /* Database handle */
+  LogMark *pMark                  /* Object containing log offset to seek to */
+){
+  LogWriter *pLog;
+
+  if( pDb->bUseLog==0 ) return;
+  pLog = pDb->pLogWriter;
+
+  assert( pMark->iOff<=pLog->iOff+pLog->buf.n );
+  if( (pMark->iOff & 0xFFFFFFF8)>=pLog->iOff ){
+    pLog->buf.n = (int)(pMark->iOff - pLog->iOff);
+    pLog->iCksumBuf = (pLog->buf.n & 0xFFFFFFF8);
+  }else{
+    pLog->buf.n = pMark->nBuf;
+    memcpy(pLog->buf.z, pMark->aBuf, pMark->nBuf);
+    pLog->iCksumBuf = 0;
+    pLog->iOff = pMark->iOff - pMark->nBuf;
+  }
+  pLog->cksum0 = pMark->cksum0;
+  pLog->cksum1 = pMark->cksum1;
+
+  if( pMark->iOff > pLog->iRegion1End ) pLog->iRegion1End = 0;
+  if( pMark->iOff > pLog->iRegion2Start ) pLog->iRegion2Start = 0;
+}
+
+/*
+** This function does the work for an lsm_info(LOG_STRUCTURE) request.
+*/
+int lsmInfoLogStructure(lsm_db *pDb, char **pzVal){
+  int rc = LSM_OK;
+  char *zVal = 0;
+
+  /* If there is no read or write transaction open, read the latest 
+  ** tree-header from shared-memory to report on. If necessary, update
+  ** it based on the contents of the database header.  
+  **
+  ** No locks are taken here - these are passive read operations only.
+  */
+  if( pDb->pCsr==0 && pDb->nTransOpen==0 ){
+    rc = lsmTreeLoadHeader(pDb, 0);
+    if( rc==LSM_OK ) rc = logReclaimSpace(pDb);
+  }
+
+  if( rc==LSM_OK ){
+    DbLog *pLog = &pDb->treehdr.log;
+    zVal = lsmMallocPrintf(pDb->pEnv, 
+        "%d %d %d %d %d %d", 
+        (int)pLog->aRegion[0].iStart, (int)pLog->aRegion[0].iEnd,
+        (int)pLog->aRegion[1].iStart, (int)pLog->aRegion[1].iEnd,
+        (int)pLog->aRegion[2].iStart, (int)pLog->aRegion[2].iEnd
+    );
+    if( !zVal ) rc = LSM_NOMEM_BKPT;
+  }
+
+  *pzVal = zVal;
+  return rc;
+}
+
+/*************************************************************************
+** Begin code for log recovery.
+*/
+
+typedef struct LogReader LogReader;
+struct LogReader {
+  FileSystem *pFS;                /* File system to read from */
+  i64 iOff;                       /* File offset at end of buf content */
+  int iBuf;                       /* Current read offset in buf */
+  LsmString buf;                  /* Buffer containing file content */
+
+  int iCksumBuf;                  /* Offset in buf corresponding to cksum[01] */
+  u32 cksum0;                     /* Checksum 0 at offset iCksumBuf */
+  u32 cksum1;                     /* Checksum 1 at offset iCksumBuf */
+};
+
+static void logReaderBlob(
+  LogReader *p,                   /* Log reader object */
+  LsmString *pBuf,                /* Dynamic storage, if required */
+  int nBlob,                      /* Number of bytes to read */
+  u8 **ppBlob,                    /* OUT: Pointer to blob read */
+  int *pRc                        /* IN/OUT: Error code */
+){
+  static const int LOG_READ_SIZE = 512;
+  int rc = *pRc;                  /* Return code */
+  int nReq = nBlob;               /* Bytes required */
+
+  while( rc==LSM_OK && nReq>0 ){
+    int nAvail;                   /* Bytes of data available in p->buf */
+    if( p->buf.n==p->iBuf ){
+      int nCksum;                 /* Total bytes requiring checksum */
+      int nCarry = 0;             /* Total bytes requiring checksum */
+
+      nCksum = p->iBuf - p->iCksumBuf;
+      if( nCksum>0 ){
+        nCarry = nCksum % 8;
+        nCksum = ((nCksum / 8) * 8);
+        if( nCksum>0 ){
+          logCksumUnaligned(
+              &p->buf.z[p->iCksumBuf], nCksum, &p->cksum0, &p->cksum1
+          );
+        }
+      }
+      if( nCarry>0 ) memcpy(p->buf.z, &p->buf.z[p->iBuf-nCarry], nCarry);
+      p->buf.n = nCarry;
+      p->iBuf = nCarry;
+
+      rc = lsmFsReadLog(p->pFS, p->iOff, LOG_READ_SIZE, &p->buf);
+      if( rc!=LSM_OK ) break;
+      p->iCksumBuf = 0;
+      p->iOff += LOG_READ_SIZE;
+    }
+
+    nAvail = p->buf.n - p->iBuf;
+    if( ppBlob && nReq==nBlob && nBlob<=nAvail ){
+      *ppBlob = (u8 *)&p->buf.z[p->iBuf];
+      p->iBuf += nBlob;
+      nReq = 0;
+    }else{
+      int nCopy = LSM_MIN(nAvail, nReq);
+      if( nBlob==nReq ){
+        pBuf->n = 0;
+      }
+      rc = lsmStringBinAppend(pBuf, (u8 *)&p->buf.z[p->iBuf], nCopy);
+      nReq -= nCopy;
+      p->iBuf += nCopy;
+      if( nReq==0 && ppBlob ){
+        *ppBlob = (u8*)pBuf->z;
+      }
+    }
+  }
+
+  *pRc = rc;
+}
+
+static void logReaderVarint(
+  LogReader *p, 
+  LsmString *pBuf,
+  int *piVal,                     /* OUT: Value read from log */
+  int *pRc                        /* IN/OUT: Error code */
+){
+  if( *pRc==LSM_OK ){
+    u8 *aVarint;
+    if( p->buf.n==p->iBuf ){
+      logReaderBlob(p, 0, 10, &aVarint, pRc);
+      if( LSM_OK==*pRc ) p->iBuf -= (10 - lsmVarintGet32(aVarint, piVal));
+    }else{
+      logReaderBlob(p, pBuf, lsmVarintSize(p->buf.z[p->iBuf]), &aVarint, pRc);
+      if( LSM_OK==*pRc ) lsmVarintGet32(aVarint, piVal);
+    }
+  }
+}
+
+static void logReaderByte(LogReader *p, u8 *pByte, int *pRc){
+  u8 *pPtr = 0;
+  logReaderBlob(p, 0, 1, &pPtr, pRc);
+  if( pPtr ) *pByte = *pPtr;
+}
+
+static void logReaderCksum(LogReader *p, LsmString *pBuf, int *pbEof, int *pRc){
+  if( *pRc==LSM_OK ){
+    u8 *pPtr = 0;
+    u32 cksum0, cksum1;
+    int nCksum = p->iBuf - p->iCksumBuf;
+
+    /* Update in-memory (expected) checksums */
+    assert( nCksum>=0 );
+    logCksumUnaligned(&p->buf.z[p->iCksumBuf], nCksum, &p->cksum0, &p->cksum1);
+    p->iCksumBuf = p->iBuf + 8;
+    logReaderBlob(p, pBuf, 8, &pPtr, pRc);
+    assert( pPtr || *pRc );
+
+    /* Read the checksums from the log file. Set *pbEof if they do not match. */
+    if( pPtr ){
+      cksum0 = lsmGetU32(pPtr);
+      cksum1 = lsmGetU32(&pPtr[4]);
+      *pbEof = (cksum0!=p->cksum0 || cksum1!=p->cksum1);
+      p->iCksumBuf = p->iBuf;
+    }
+  }
+}
+
+static void logReaderInit(
+  lsm_db *pDb,                    /* Database handle */
+  DbLog *pLog,                    /* Log object associated with pDb */
+  int bInitBuf,                   /* True if p->buf is uninitialized */
+  LogReader *p                    /* Initialize this LogReader object */
+){
+  p->pFS = pDb->pFS;
+  p->iOff = pLog->aRegion[2].iStart;
+  p->cksum0 = pLog->cksum0;
+  p->cksum1 = pLog->cksum1;
+  if( bInitBuf ){ lsmStringInit(&p->buf, pDb->pEnv); }
+  p->buf.n = 0;
+  p->iCksumBuf = 0;
+  p->iBuf = 0;
+}
+
+/*
+** This function is called after reading the header of a LOG_DELETE or
+** LOG_WRITE record. Parameter nByte is the total size of the key and
+** value that follow the header just read. Return true if the size and
+** position of the record indicate that it should contain a checksum.
+*/
+static int logRequireCksum(LogReader *p, int nByte){
+  return ((p->iBuf + nByte - p->iCksumBuf) > LSM_CKSUM_MAXDATA);
+}
+
+/*
+** Recover the contents of the log file.
+*/
+int lsmLogRecover(lsm_db *pDb){
+  LsmString buf1;                 /* Key buffer */
+  LsmString buf2;                 /* Value buffer */
+  LogReader reader;               /* Log reader object */
+  int rc = LSM_OK;                /* Return code */
+  int nCommit = 0;                /* Number of transactions to recover */
+  int iPass;
+  int nJump = 0;                  /* Number of LSM_LOG_JUMP records in pass 0 */
+  DbLog *pLog;
+  int bOpen;
+
+  rc = lsmFsOpenLog(pDb, &bOpen);
+  if( rc!=LSM_OK ) return rc;
+
+  rc = lsmTreeInit(pDb);
+  if( rc!=LSM_OK ) return rc;
+
+  pLog = &pDb->treehdr.log;
+  lsmCheckpointLogoffset(pDb->pShmhdr->aSnap2, pLog);
+
+  logReaderInit(pDb, pLog, 1, &reader);
+  lsmStringInit(&buf1, pDb->pEnv);
+  lsmStringInit(&buf2, pDb->pEnv);
+
+  /* The outer for() loop runs at most twice. The first iteration is to 
+  ** count the number of committed transactions in the log. The second 
+  ** iterates through those transactions and updates the in-memory tree 
+  ** structure with their contents.  */
+  if( bOpen ){
+    for(iPass=0; iPass<2 && rc==LSM_OK; iPass++){
+      int bEof = 0;
+
+      while( rc==LSM_OK && !bEof ){
+        u8 eType = 0;
+        logReaderByte(&reader, &eType, &rc);
+
+        switch( eType ){
+          case LSM_LOG_PAD1:
+            break;
+
+          case LSM_LOG_PAD2: {
+            int nPad;
+            logReaderVarint(&reader, &buf1, &nPad, &rc);
+            logReaderBlob(&reader, &buf1, nPad, 0, &rc);
+            break;
+          }
+
+          case LSM_LOG_DRANGE:
+          case LSM_LOG_DRANGE_CKSUM:
+          case LSM_LOG_WRITE:
+          case LSM_LOG_WRITE_CKSUM: {
+            int nKey;
+            int nVal;
+            u8 *aVal;
+            logReaderVarint(&reader, &buf1, &nKey, &rc);
+            logReaderVarint(&reader, &buf2, &nVal, &rc);
+
+            if( eType==LSM_LOG_WRITE_CKSUM || eType==LSM_LOG_DRANGE_CKSUM ){
+              logReaderCksum(&reader, &buf1, &bEof, &rc);
+            }else{
+              bEof = logRequireCksum(&reader, nKey+nVal);
+            }
+            if( bEof ) break;
+
+            logReaderBlob(&reader, &buf1, nKey, 0, &rc);
+            logReaderBlob(&reader, &buf2, nVal, &aVal, &rc);
+            if( iPass==1 && rc==LSM_OK ){ 
+              if( eType==LSM_LOG_WRITE || eType==LSM_LOG_WRITE_CKSUM ){
+                rc = lsmTreeInsert(pDb, (u8 *)buf1.z, nKey, aVal, nVal);
+              }else{
+                rc = lsmTreeDelete(pDb, (u8 *)buf1.z, nKey, aVal, nVal);
+              }
+            }
+            break;
+          }
+
+          case LSM_LOG_DELETE:
+          case LSM_LOG_DELETE_CKSUM: {
+            int nKey; u8 *aKey;
+            logReaderVarint(&reader, &buf1, &nKey, &rc);
+
+            if( eType==LSM_LOG_DELETE_CKSUM ){
+              logReaderCksum(&reader, &buf1, &bEof, &rc);
+            }else{
+              bEof = logRequireCksum(&reader, nKey);
+            }
+            if( bEof ) break;
+
+            logReaderBlob(&reader, &buf1, nKey, &aKey, &rc);
+            if( iPass==1 && rc==LSM_OK ){ 
+              rc = lsmTreeInsert(pDb, aKey, nKey, NULL, -1);
+            }
+            break;
+          }
+
+          case LSM_LOG_COMMIT:
+            logReaderCksum(&reader, &buf1, &bEof, &rc);
+            if( bEof==0 ){
+              nCommit++;
+              assert( nCommit>0 || iPass==1 );
+              if( nCommit==0 ) bEof = 1;
+            }
+            break;
+
+          case LSM_LOG_JUMP: {
+            int iOff = 0;
+            logReaderVarint(&reader, &buf1, &iOff, &rc);
+            if( rc==LSM_OK ){
+              if( iPass==1 ){
+                if( pLog->aRegion[2].iStart==0 ){
+                  assert( pLog->aRegion[1].iStart==0 );
+                  pLog->aRegion[1].iEnd = reader.iOff;
+                }else{
+                  assert( pLog->aRegion[0].iStart==0 );
+                  pLog->aRegion[0].iStart = pLog->aRegion[2].iStart;
+                  pLog->aRegion[0].iEnd = reader.iOff-reader.buf.n+reader.iBuf;
+                }
+                pLog->aRegion[2].iStart = iOff;
+              }else{
+                if( (nJump++)==2 ){
+                  bEof = 1;
+                }
+              }
+
+              reader.iOff = iOff;
+              reader.buf.n = reader.iBuf;
+            }
+            break;
+          }
+
+          default:
+            /* Including LSM_LOG_EOF */
+            bEof = 1;
+            break;
+        }
+      }
+
+      if( rc==LSM_OK && iPass==0 ){
+        if( nCommit==0 ){
+          if( pLog->aRegion[2].iStart==0 ){
+            iPass = 1;
+          }else{
+            pLog->aRegion[2].iStart = 0;
+            iPass = -1;
+            lsmCheckpointZeroLogoffset(pDb);
+          }
+        }
+        logReaderInit(pDb, pLog, 0, &reader);
+        nCommit = nCommit * -1;
+      }
+    }
+  }
+
+  /* Initialize DbLog object */
+  if( rc==LSM_OK ){
+    pLog->aRegion[2].iEnd = reader.iOff - reader.buf.n + reader.iBuf;
+    pLog->cksum0 = reader.cksum0;
+    pLog->cksum1 = reader.cksum1;
+  }
+
+  if( rc==LSM_OK ){
+    rc = lsmFinishRecovery(pDb);
+  }else{
+    lsmFinishRecovery(pDb);
+  }
+
+  if( pDb->bRoTrans ){
+    lsmFsCloseLog(pDb);
+  }
+
+  lsmStringClear(&buf1);
+  lsmStringClear(&buf2);
+  lsmStringClear(&reader.buf);
+  return rc;
+}
+
+void lsmLogClose(lsm_db *db){
+  if( db->pLogWriter ){
+    lsmFree(db->pEnv, db->pLogWriter->buf.z);
+    lsmFree(db->pEnv, db->pLogWriter);
+    db->pLogWriter = 0;
+  }
+}