/* ** 2011-09-11 ** ** 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 code to read and write checkpoints. ** ** A checkpoint represents the database layout at a single point in time. ** It includes a log offset. When an existing database is opened, the ** current state is determined by reading the newest checkpoint and updating ** it with all committed transactions from the log that follow the specified ** offset. */ #include "lsmInt.h" /* ** CHECKPOINT BLOB FORMAT: ** ** A checkpoint blob is a series of unsigned 32-bit integers stored in ** big-endian byte order. As follows: ** ** Checkpoint header (see the CKPT_HDR_XXX #defines): ** ** 1. The checkpoint id MSW. ** 2. The checkpoint id LSW. ** 3. The number of integer values in the entire checkpoint, including ** the two checksum values. ** 4. The compression scheme id. ** 5. The total number of blocks in the database. ** 6. The block size. ** 7. The number of levels. ** 8. The nominal database page size. ** 9. The number of pages (in total) written to the database file. ** ** Log pointer: ** ** 1. The log offset MSW. ** 2. The log offset LSW. ** 3. Log checksum 0. ** 4. Log checksum 1. ** ** Note that the "log offset" is not the literal byte offset. Instead, ** it is the byte offset multiplied by 2, with least significant bit ** toggled each time the log pointer value is changed. This is to make ** sure that this field changes each time the log pointer is updated, ** even if the log file itself is disabled. See lsmTreeMakeOld(). ** ** See ckptExportLog() and ckptImportLog(). ** ** Append points: ** ** 8 integers (4 * 64-bit page numbers). See ckptExportAppendlist(). ** ** For each level in the database, a level record. Formatted as follows: ** ** 0. Age of the level (least significant 16-bits). And flags mask (most ** significant 16-bits). ** 1. The number of right-hand segments (nRight, possibly 0), ** 2. Segment record for left-hand segment (8 integers defined below), ** 3. Segment record for each right-hand segment (8 integers defined below), ** 4. If nRight>0, The number of segments involved in the merge ** 5. if nRight>0, Current nSkip value (see Merge structure defn.), ** 6. For each segment in the merge: ** 5a. Page number of next cell to read during merge (this field ** is 64-bits - 2 integers) ** 5b. Cell number of next cell to read during merge ** 7. Page containing current split-key (64-bits - 2 integers). ** 8. Cell within page containing current split-key. ** 9. Current pointer value (64-bits - 2 integers). ** ** The block redirect array: ** ** 1. Number of redirections (maximum LSM_MAX_BLOCK_REDIRECTS). ** 2. For each redirection: ** a. "from" block number ** b. "to" block number ** ** The in-memory freelist entries. Each entry is either an insert or a ** delete. The in-memory freelist is to the free-block-list as the ** in-memory tree is to the users database content. ** ** 1. Number of free-list entries stored in checkpoint header. ** 2. Number of free blocks (in total). ** 3. Total number of blocks freed during database lifetime. ** 4. For each entry: ** 2a. Block number of free block. ** 2b. A 64-bit integer (MSW followed by LSW). -1 for a delete entry, ** or the associated checkpoint id for an insert. ** ** The checksum: ** ** 1. Checksum value 1. ** 2. Checksum value 2. ** ** In the above, a segment record consists of the following four 64-bit ** fields (converted to 2 * u32 by storing the MSW followed by LSW): ** ** 1. First page of array, ** 2. Last page of array, ** 3. Root page of array (or 0), ** 4. Size of array in pages. */ /* ** LARGE NUMBERS OF LEVEL RECORDS: ** ** A limit on the number of rhs segments that may be present in the database ** file. Defining this limit ensures that all level records fit within ** the 4096 byte limit for checkpoint blobs. ** ** The number of right-hand-side segments in a database is counted as ** follows: ** ** * For each level in the database not undergoing a merge, add 1. ** ** * For each level in the database that is undergoing a merge, add ** the number of segments on the rhs of the level. ** ** A level record not undergoing a merge is 10 integers. A level record ** with nRhs rhs segments and (nRhs+1) input segments (i.e. including the ** separators from the next level) is (11*nRhs+20) integers. The maximum ** per right-hand-side level is therefore 21 integers. So the maximum ** size of all level records in a checkpoint is 21*40=820 integers. ** ** TODO: Before pointer values were changed from 32 to 64 bits, the above ** used to come to 420 bytes - leaving significant space for a free-list ** prefix. No more. To fix this, reduce the size of the level records in ** a db snapshot, and improve management of the free-list tail in ** lsm_sorted.c. */ #define LSM_MAX_RHS_SEGMENTS 40 /* ** LARGE NUMBERS OF FREELIST ENTRIES: ** ** There is also a limit (LSM_MAX_FREELIST_ENTRIES - defined in lsmInt.h) ** on the number of free-list entries stored in a checkpoint. Since each ** free-list entry consists of 3 integers, the maximum free-list size is ** 3*100=300 integers. Combined with the limit on rhs segments defined ** above, this ensures that a checkpoint always fits within a 4096 byte ** meta page. ** ** If the database contains more than 100 free blocks, the "overflow" flag ** in the checkpoint header is set and the remainder are stored in the ** system FREELIST entry in the LSM (along with user data). The value ** accompanying the FREELIST key in the LSM is, like a checkpoint, an array ** of 32-bit big-endian integers. As follows: ** ** For each entry: ** a. Block number of free block. ** b. MSW of associated checkpoint id. ** c. LSW of associated checkpoint id. ** ** The number of entries is not required - it is implied by the size of the ** value blob containing the integer array. ** ** Note that the limit defined by LSM_MAX_FREELIST_ENTRIES is a hard limit. ** The actual value used may be configured using LSM_CONFIG_MAX_FREELIST. */ /* ** The argument to this macro must be of type u32. On a little-endian ** architecture, it returns the u32 value that results from interpreting ** the 4 bytes as a big-endian value. On a big-endian architecture, it ** 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 const int one = 1; #define LSM_LITTLE_ENDIAN (*(u8 *)(&one)) /* Sizes, in integers, of various parts of the checkpoint. */ #define CKPT_HDR_SIZE 9 #define CKPT_LOGPTR_SIZE 4 #define CKPT_APPENDLIST_SIZE (LSM_APPLIST_SZ * 2) /* A #define to describe each integer in the checkpoint header. */ #define CKPT_HDR_ID_MSW 0 #define CKPT_HDR_ID_LSW 1 #define CKPT_HDR_NCKPT 2 #define CKPT_HDR_CMPID 3 #define CKPT_HDR_NBLOCK 4 #define CKPT_HDR_BLKSZ 5 #define CKPT_HDR_NLEVEL 6 #define CKPT_HDR_PGSZ 7 #define CKPT_HDR_NWRITE 8 #define CKPT_HDR_LO_MSW 9 #define CKPT_HDR_LO_LSW 10 #define CKPT_HDR_LO_CKSUM1 11 #define CKPT_HDR_LO_CKSUM2 12 typedef struct CkptBuffer CkptBuffer; /* ** Dynamic buffer used to accumulate data for a checkpoint. */ struct CkptBuffer { lsm_env *pEnv; int nAlloc; u32 *aCkpt; }; /* ** Calculate the checksum of the checkpoint specified by arguments aCkpt and ** nCkpt. Store the checksum in *piCksum1 and *piCksum2 before returning. ** ** The value of the nCkpt parameter includes the two checksum values at ** the end of the checkpoint. They are not used as inputs to the checksum ** calculation. The checksum is based on the array of (nCkpt-2) integers ** at aCkpt[]. */ static void ckptChecksum(u32 *aCkpt, u32 nCkpt, u32 *piCksum1, u32 *piCksum2){ u32 i; u32 cksum1 = 1; u32 cksum2 = 2; if( nCkpt % 2 ){ cksum1 += aCkpt[nCkpt-3] & 0x0000FFFF; cksum2 += aCkpt[nCkpt-3] & 0xFFFF0000; } for(i=0; (i+3)=p->nAlloc ){ int nNew = LSM_MAX(8, iIdx*2); p->aCkpt = (u32 *)lsmReallocOrFree(p->pEnv, p->aCkpt, nNew*sizeof(u32)); if( !p->aCkpt ){ *pRc = LSM_NOMEM_BKPT; return; } p->nAlloc = nNew; } p->aCkpt[iIdx] = iVal; } /* ** Argument aInt points to an array nInt elements in size. Switch the ** endian-ness of each element of the array. */ static void ckptChangeEndianness(u32 *aInt, int nInt){ if( LSM_LITTLE_ENDIAN ){ int i; for(i=0; iaCkpt, nCkpt+2, &aCksum[0], &aCksum[1]); ckptSetValue(p, nCkpt, aCksum[0], pRc); ckptSetValue(p, nCkpt+1, aCksum[1], pRc); } } static void ckptAppend64(CkptBuffer *p, int *piOut, i64 iVal, int *pRc){ int iOut = *piOut; ckptSetValue(p, iOut++, (iVal >> 32) & 0xFFFFFFFF, pRc); ckptSetValue(p, iOut++, (iVal & 0xFFFFFFFF), pRc); *piOut = iOut; } static i64 ckptRead64(u32 *a){ return (((i64)a[0]) << 32) + (i64)a[1]; } static i64 ckptGobble64(u32 *a, int *piIn){ int iIn = *piIn; *piIn += 2; return ckptRead64(&a[iIn]); } /* ** Append a 6-value segment record corresponding to pSeg to the checkpoint ** buffer passed as the third argument. */ static void ckptExportSegment( Segment *pSeg, CkptBuffer *p, int *piOut, int *pRc ){ ckptAppend64(p, piOut, pSeg->iFirst, pRc); ckptAppend64(p, piOut, pSeg->iLastPg, pRc); ckptAppend64(p, piOut, pSeg->iRoot, pRc); ckptAppend64(p, piOut, pSeg->nSize, pRc); } static void ckptExportLevel( Level *pLevel, /* Level object to serialize */ CkptBuffer *p, /* Append new level record to this ckpt */ int *piOut, /* IN/OUT: Size of checkpoint so far */ int *pRc /* IN/OUT: Error code */ ){ int iOut = *piOut; Merge *pMerge; pMerge = pLevel->pMerge; ckptSetValue(p, iOut++, (u32)pLevel->iAge + (u32)(pLevel->flags<<16), pRc); ckptSetValue(p, iOut++, pLevel->nRight, pRc); ckptExportSegment(&pLevel->lhs, p, &iOut, pRc); assert( (pLevel->nRight>0)==(pMerge!=0) ); if( pMerge ){ int i; for(i=0; inRight; i++){ ckptExportSegment(&pLevel->aRhs[i], p, &iOut, pRc); } assert( pMerge->nInput==pLevel->nRight || pMerge->nInput==pLevel->nRight+1 ); ckptSetValue(p, iOut++, pMerge->nInput, pRc); ckptSetValue(p, iOut++, pMerge->nSkip, pRc); for(i=0; inInput; i++){ ckptAppend64(p, &iOut, pMerge->aInput[i].iPg, pRc); ckptSetValue(p, iOut++, pMerge->aInput[i].iCell, pRc); } ckptAppend64(p, &iOut, pMerge->splitkey.iPg, pRc); ckptSetValue(p, iOut++, pMerge->splitkey.iCell, pRc); ckptAppend64(p, &iOut, pMerge->iCurrentPtr, pRc); } *piOut = iOut; } /* ** Populate the log offset fields of the checkpoint buffer. 4 values. */ static void ckptExportLog( lsm_db *pDb, int bFlush, CkptBuffer *p, int *piOut, int *pRc ){ int iOut = *piOut; assert( iOut==CKPT_HDR_LO_MSW ); if( bFlush ){ i64 iOff = pDb->treehdr.iOldLog; ckptAppend64(p, &iOut, iOff, pRc); ckptSetValue(p, iOut++, pDb->treehdr.oldcksum0, pRc); ckptSetValue(p, iOut++, pDb->treehdr.oldcksum1, pRc); }else{ for(; iOut<=CKPT_HDR_LO_CKSUM2; iOut++){ ckptSetValue(p, iOut, pDb->pShmhdr->aSnap2[iOut], pRc); } } assert( *pRc || iOut==CKPT_HDR_LO_CKSUM2+1 ); *piOut = iOut; } static void ckptExportAppendlist( lsm_db *db, /* Database connection */ CkptBuffer *p, /* Checkpoint buffer to write to */ int *piOut, /* IN/OUT: Offset within checkpoint buffer */ int *pRc /* IN/OUT: Error code */ ){ int i; LsmPgno *aiAppend = db->pWorker->aiAppend; for(i=0; ipFS; /* File system object */ Snapshot *pSnap = pDb->pWorker; /* Worker snapshot */ int nLevel = 0; /* Number of levels in checkpoint */ int iLevel; /* Used to count out nLevel levels */ int iOut = 0; /* Current offset in aCkpt[] */ Level *pLevel; /* Level iterator */ int i; /* Iterator used while serializing freelist */ CkptBuffer ckpt; /* Initialize the output buffer */ memset(&ckpt, 0, sizeof(CkptBuffer)); ckpt.pEnv = pDb->pEnv; iOut = CKPT_HDR_SIZE; /* Write the log offset into the checkpoint. */ ckptExportLog(pDb, bLog, &ckpt, &iOut, &rc); /* Write the append-point list */ ckptExportAppendlist(pDb, &ckpt, &iOut, &rc); /* Figure out how many levels will be written to the checkpoint. */ for(pLevel=lsmDbSnapshotLevel(pSnap); pLevel; pLevel=pLevel->pNext) nLevel++; /* Serialize nLevel levels. */ iLevel = 0; for(pLevel=lsmDbSnapshotLevel(pSnap); iLevelpNext){ ckptExportLevel(pLevel, &ckpt, &iOut, &rc); iLevel++; } /* Write the block-redirect list */ ckptSetValue(&ckpt, iOut++, pSnap->redirect.n, &rc); for(i=0; iredirect.n; i++){ ckptSetValue(&ckpt, iOut++, pSnap->redirect.a[i].iFrom, &rc); ckptSetValue(&ckpt, iOut++, pSnap->redirect.a[i].iTo, &rc); } /* Write the freelist */ assert( pSnap->freelist.nEntry<=pDb->nMaxFreelist ); if( rc==LSM_OK ){ int nFree = pSnap->freelist.nEntry; ckptSetValue(&ckpt, iOut++, nFree, &rc); for(i=0; ifreelist.aEntry[i]; ckptSetValue(&ckpt, iOut++, p->iBlk, &rc); ckptSetValue(&ckpt, iOut++, (p->iId >> 32) & 0xFFFFFFFF, &rc); ckptSetValue(&ckpt, iOut++, p->iId & 0xFFFFFFFF, &rc); } } /* Write the checkpoint header */ assert( iId>=0 ); assert( pSnap->iCmpId==pDb->compress.iId || pSnap->iCmpId==LSM_COMPRESSION_EMPTY ); ckptSetValue(&ckpt, CKPT_HDR_ID_MSW, (u32)(iId>>32), &rc); ckptSetValue(&ckpt, CKPT_HDR_ID_LSW, (u32)(iId&0xFFFFFFFF), &rc); ckptSetValue(&ckpt, CKPT_HDR_NCKPT, iOut+2, &rc); ckptSetValue(&ckpt, CKPT_HDR_CMPID, pDb->compress.iId, &rc); ckptSetValue(&ckpt, CKPT_HDR_NBLOCK, pSnap->nBlock, &rc); ckptSetValue(&ckpt, CKPT_HDR_BLKSZ, lsmFsBlockSize(pFS), &rc); ckptSetValue(&ckpt, CKPT_HDR_NLEVEL, nLevel, &rc); ckptSetValue(&ckpt, CKPT_HDR_PGSZ, lsmFsPageSize(pFS), &rc); ckptSetValue(&ckpt, CKPT_HDR_NWRITE, pSnap->nWrite, &rc); if( bCksum ){ ckptAddChecksum(&ckpt, iOut, &rc); }else{ ckptSetValue(&ckpt, iOut, 0, &rc); ckptSetValue(&ckpt, iOut+1, 0, &rc); } iOut += 2; assert( iOut<=1024 ); #ifdef LSM_LOG_FREELIST lsmLogMessage(pDb, rc, "ckptExportSnapshot(): id=%lld freelist: %d", iId, pSnap->freelist.nEntry ); for(i=0; ifreelist.nEntry; i++){ lsmLogMessage(pDb, rc, "ckptExportSnapshot(): iBlk=%d id=%lld", pSnap->freelist.aEntry[i].iBlk, pSnap->freelist.aEntry[i].iId ); } #endif *ppCkpt = (void *)ckpt.aCkpt; if( pnCkpt ) *pnCkpt = sizeof(u32)*iOut; return rc; } /* ** Helper function for ckptImport(). */ static void ckptNewSegment( u32 *aIn, int *piIn, Segment *pSegment /* Populate this structure */ ){ assert( pSegment->iFirst==0 && pSegment->iLastPg==0 ); assert( pSegment->nSize==0 && pSegment->iRoot==0 ); pSegment->iFirst = ckptGobble64(aIn, piIn); pSegment->iLastPg = ckptGobble64(aIn, piIn); pSegment->iRoot = ckptGobble64(aIn, piIn); pSegment->nSize = (int)ckptGobble64(aIn, piIn); assert( pSegment->iFirst ); } static int ckptSetupMerge(lsm_db *pDb, u32 *aInt, int *piIn, Level *pLevel){ Merge *pMerge; /* Allocated Merge object */ int nInput; /* Number of input segments in merge */ int iIn = *piIn; /* Next value to read from aInt[] */ int i; /* Iterator variable */ int nByte; /* Number of bytes to allocate */ /* Allocate the Merge object. If malloc() fails, return LSM_NOMEM. */ nInput = (int)aInt[iIn++]; nByte = sizeof(Merge) + sizeof(MergeInput) * nInput; pMerge = (Merge *)lsmMallocZero(pDb->pEnv, nByte); if( !pMerge ) return LSM_NOMEM_BKPT; pLevel->pMerge = pMerge; /* Populate the Merge object. */ pMerge->aInput = (MergeInput *)&pMerge[1]; pMerge->nInput = nInput; pMerge->iOutputOff = -1; pMerge->nSkip = (int)aInt[iIn++]; for(i=0; iaInput[i].iPg = ckptGobble64(aInt, &iIn); pMerge->aInput[i].iCell = (int)aInt[iIn++]; } pMerge->splitkey.iPg = ckptGobble64(aInt, &iIn); pMerge->splitkey.iCell = (int)aInt[iIn++]; pMerge->iCurrentPtr = ckptGobble64(aInt, &iIn); /* Set *piIn and return LSM_OK. */ *piIn = iIn; return LSM_OK; } static int ckptLoadLevels( lsm_db *pDb, u32 *aIn, int *piIn, int nLevel, Level **ppLevel ){ int i; int rc = LSM_OK; Level *pRet = 0; Level **ppNext; int iIn = *piIn; ppNext = &pRet; for(i=0; rc==LSM_OK && ipEnv, sizeof(Level), &rc); if( rc==LSM_OK ){ pLevel->iAge = (u16)(aIn[iIn] & 0x0000FFFF); pLevel->flags = (u16)((aIn[iIn]>>16) & 0x0000FFFF); iIn++; pLevel->nRight = aIn[iIn++]; if( pLevel->nRight ){ int nByte = sizeof(Segment) * pLevel->nRight; pLevel->aRhs = (Segment *)lsmMallocZeroRc(pDb->pEnv, nByte, &rc); } if( rc==LSM_OK ){ *ppNext = pLevel; ppNext = &pLevel->pNext; /* Allocate the main segment */ ckptNewSegment(aIn, &iIn, &pLevel->lhs); /* Allocate each of the right-hand segments, if any */ for(iRight=0; iRightnRight; iRight++){ ckptNewSegment(aIn, &iIn, &pLevel->aRhs[iRight]); } /* Set up the Merge object, if required */ if( pLevel->nRight>0 ){ rc = ckptSetupMerge(pDb, aIn, &iIn, pLevel); } } } } if( rc!=LSM_OK ){ /* An OOM must have occurred. Free any level structures allocated and ** return the error to the caller. */ lsmSortedFreeLevel(pDb->pEnv, pRet); pRet = 0; } *ppLevel = pRet; *piIn = iIn; return rc; } int lsmCheckpointLoadLevels(lsm_db *pDb, void *pVal, int nVal){ int rc = LSM_OK; if( nVal>0 ){ u32 *aIn; aIn = lsmMallocRc(pDb->pEnv, nVal, &rc); if( aIn ){ Level *pLevel = 0; Level *pParent; int nIn; int nLevel; int iIn = 1; memcpy(aIn, pVal, nVal); nIn = nVal / sizeof(u32); ckptChangeEndianness(aIn, nIn); nLevel = aIn[0]; rc = ckptLoadLevels(pDb, aIn, &iIn, nLevel, &pLevel); lsmFree(pDb->pEnv, aIn); assert( rc==LSM_OK || pLevel==0 ); if( rc==LSM_OK ){ pParent = lsmDbSnapshotLevel(pDb->pWorker); assert( pParent ); while( pParent->pNext ) pParent = pParent->pNext; pParent->pNext = pLevel; } } } return rc; } /* ** Return the data for the LEVELS record. ** ** The size of the checkpoint that can be stored in the database header ** must not exceed 1024 32-bit integers. Normally, it does not. However, ** if it does, part of the checkpoint must be stored in the LSM. This ** routine returns that part. */ int lsmCheckpointLevels( lsm_db *pDb, /* Database handle */ int nLevel, /* Number of levels to write to blob */ void **paVal, /* OUT: Pointer to LEVELS blob */ int *pnVal /* OUT: Size of LEVELS blob in bytes */ ){ Level *p; /* Used to iterate through levels */ int nAll= 0; int rc; int i; int iOut; CkptBuffer ckpt; assert( nLevel>0 ); for(p=lsmDbSnapshotLevel(pDb->pWorker); p; p=p->pNext) nAll++; assert( nAll>nLevel ); nAll -= nLevel; for(p=lsmDbSnapshotLevel(pDb->pWorker); p && nAll>0; p=p->pNext) nAll--; memset(&ckpt, 0, sizeof(CkptBuffer)); ckpt.pEnv = pDb->pEnv; ckptSetValue(&ckpt, 0, nLevel, &rc); iOut = 1; for(i=0; rc==LSM_OK && ipNext; } assert( rc!=LSM_OK || p==0 ); if( rc==LSM_OK ){ ckptChangeEndianness(ckpt.aCkpt, iOut); *paVal = (void *)ckpt.aCkpt; *pnVal = iOut * sizeof(u32); }else{ *pnVal = 0; *paVal = 0; } return rc; } /* ** Read the checkpoint id from meta-page pPg. */ static i64 ckptLoadId(MetaPage *pPg){ i64 ret = 0; if( pPg ){ int nData; u8 *aData = lsmFsMetaPageData(pPg, &nData); ret = (((i64)lsmGetU32(&aData[CKPT_HDR_ID_MSW*4])) << 32) + ((i64)lsmGetU32(&aData[CKPT_HDR_ID_LSW*4])); } return ret; } /* ** Return true if the buffer passed as an argument contains a valid ** checkpoint. */ static int ckptChecksumOk(u32 *aCkpt){ u32 nCkpt = aCkpt[CKPT_HDR_NCKPT]; u32 cksum1; u32 cksum2; if( nCkpt(LSM_META_RW_PAGE_SIZE)/sizeof(u32) ){ return 0; } ckptChecksum(aCkpt, nCkpt, &cksum1, &cksum2); return (cksum1==aCkpt[nCkpt-2] && cksum2==aCkpt[nCkpt-1]); } /* ** Attempt to load a checkpoint from meta page iMeta. ** ** This function is a no-op if *pRc is set to any value other than LSM_OK ** when it is called. If an error occurs, *pRc is set to an LSM error code ** before returning. ** ** If no error occurs and the checkpoint is successfully loaded, copy it to ** ShmHeader.aSnap1[] and ShmHeader.aSnap2[], and set ShmHeader.iMetaPage ** to indicate its origin. In this case return 1. Or, if the checkpoint ** cannot be loaded (because the checksum does not compute), return 0. */ static int ckptTryLoad(lsm_db *pDb, MetaPage *pPg, u32 iMeta, int *pRc){ int bLoaded = 0; /* Return value */ if( *pRc==LSM_OK ){ int rc = LSM_OK; /* Error code */ u32 *aCkpt = 0; /* Pointer to buffer containing checkpoint */ u32 nCkpt; /* Number of elements in aCkpt[] */ int nData; /* Bytes of data in aData[] */ u8 *aData; /* Meta page data */ aData = lsmFsMetaPageData(pPg, &nData); nCkpt = (u32)lsmGetU32(&aData[CKPT_HDR_NCKPT*sizeof(u32)]); if( nCkpt<=nData/sizeof(u32) && nCkpt>CKPT_HDR_NCKPT ){ aCkpt = (u32 *)lsmMallocRc(pDb->pEnv, nCkpt*sizeof(u32), &rc); } if( aCkpt ){ memcpy(aCkpt, aData, nCkpt*sizeof(u32)); ckptChangeEndianness(aCkpt, nCkpt); if( ckptChecksumOk(aCkpt) ){ ShmHeader *pShm = pDb->pShmhdr; memcpy(pShm->aSnap1, aCkpt, nCkpt*sizeof(u32)); memcpy(pShm->aSnap2, aCkpt, nCkpt*sizeof(u32)); memcpy(pDb->aSnapshot, aCkpt, nCkpt*sizeof(u32)); pShm->iMetaPage = iMeta; bLoaded = 1; } } lsmFree(pDb->pEnv, aCkpt); *pRc = rc; } return bLoaded; } /* ** Initialize the shared-memory header with an empty snapshot. This function ** is called when no valid snapshot can be found in the database header. */ static void ckptLoadEmpty(lsm_db *pDb){ u32 aCkpt[] = { 0, /* CKPT_HDR_ID_MSW */ 10, /* CKPT_HDR_ID_LSW */ 0, /* CKPT_HDR_NCKPT */ LSM_COMPRESSION_EMPTY, /* CKPT_HDR_CMPID */ 0, /* CKPT_HDR_NBLOCK */ 0, /* CKPT_HDR_BLKSZ */ 0, /* CKPT_HDR_NLEVEL */ 0, /* CKPT_HDR_PGSZ */ 0, /* CKPT_HDR_NWRITE */ 0, 0, 1234, 5678, /* The log pointer and initial checksum */ 0,0,0,0, 0,0,0,0, /* The append list */ 0, /* The redirected block list */ 0, /* The free block list */ 0, 0 /* Space for checksum values */ }; u32 nCkpt = array_size(aCkpt); ShmHeader *pShm = pDb->pShmhdr; aCkpt[CKPT_HDR_NCKPT] = nCkpt; aCkpt[CKPT_HDR_BLKSZ] = pDb->nDfltBlksz; aCkpt[CKPT_HDR_PGSZ] = pDb->nDfltPgsz; ckptChecksum(aCkpt, array_size(aCkpt), &aCkpt[nCkpt-2], &aCkpt[nCkpt-1]); memcpy(pShm->aSnap1, aCkpt, nCkpt*sizeof(u32)); memcpy(pShm->aSnap2, aCkpt, nCkpt*sizeof(u32)); memcpy(pDb->aSnapshot, aCkpt, nCkpt*sizeof(u32)); } /* ** This function is called as part of database recovery to initialize the ** ShmHeader.aSnap1[] and ShmHeader.aSnap2[] snapshots. */ int lsmCheckpointRecover(lsm_db *pDb){ int rc = LSM_OK; /* Return Code */ i64 iId1; /* Id of checkpoint on meta-page 1 */ i64 iId2; /* Id of checkpoint on meta-page 2 */ int bLoaded = 0; /* True once checkpoint has been loaded */ int cmp; /* True if (iId2>iId1) */ MetaPage *apPg[2] = {0, 0}; /* Meta-pages 1 and 2 */ rc = lsmFsMetaPageGet(pDb->pFS, 0, 1, &apPg[0]); if( rc==LSM_OK ) rc = lsmFsMetaPageGet(pDb->pFS, 0, 2, &apPg[1]); iId1 = ckptLoadId(apPg[0]); iId2 = ckptLoadId(apPg[1]); cmp = (iId2 > iId1); bLoaded = ckptTryLoad(pDb, apPg[cmp?1:0], (cmp?2:1), &rc); if( bLoaded==0 ){ bLoaded = ckptTryLoad(pDb, apPg[cmp?0:1], (cmp?1:2), &rc); } /* The database does not contain a valid checkpoint. Initialize the shared ** memory header with an empty checkpoint. */ if( bLoaded==0 ){ ckptLoadEmpty(pDb); } lsmFsMetaPageRelease(apPg[0]); lsmFsMetaPageRelease(apPg[1]); return rc; } /* ** Store the snapshot in pDb->aSnapshot[] in meta-page iMeta. */ int lsmCheckpointStore(lsm_db *pDb, int iMeta){ MetaPage *pPg = 0; int rc; assert( iMeta==1 || iMeta==2 ); rc = lsmFsMetaPageGet(pDb->pFS, 1, iMeta, &pPg); if( rc==LSM_OK ){ u8 *aData; int nData; int nCkpt; nCkpt = (int)pDb->aSnapshot[CKPT_HDR_NCKPT]; aData = lsmFsMetaPageData(pPg, &nData); memcpy(aData, pDb->aSnapshot, nCkpt*sizeof(u32)); ckptChangeEndianness((u32 *)aData, nCkpt); rc = lsmFsMetaPageRelease(pPg); } return rc; } /* ** Copy the current client snapshot from shared-memory to pDb->aSnapshot[]. */ int lsmCheckpointLoad(lsm_db *pDb, int *piRead){ int nRem = LSM_ATTEMPTS_BEFORE_PROTOCOL; ShmHeader *pShm = pDb->pShmhdr; while( (nRem--)>0 ){ int nInt; nInt = pShm->aSnap1[CKPT_HDR_NCKPT]; if( nInt<=(LSM_META_RW_PAGE_SIZE / sizeof(u32)) ){ memcpy(pDb->aSnapshot, pShm->aSnap1, nInt*sizeof(u32)); if( ckptChecksumOk(pDb->aSnapshot) ){ if( piRead ) *piRead = 1; return LSM_OK; } } nInt = pShm->aSnap2[CKPT_HDR_NCKPT]; if( nInt<=(LSM_META_RW_PAGE_SIZE / sizeof(u32)) ){ memcpy(pDb->aSnapshot, pShm->aSnap2, nInt*sizeof(u32)); if( ckptChecksumOk(pDb->aSnapshot) ){ if( piRead ) *piRead = 2; return LSM_OK; } } lsmShmBarrier(pDb); } return LSM_PROTOCOL_BKPT; } int lsmInfoCompressionId(lsm_db *db, u32 *piCmpId){ int rc; assert( db->pClient==0 && db->pWorker==0 ); rc = lsmCheckpointLoad(db, 0); if( rc==LSM_OK ){ *piCmpId = db->aSnapshot[CKPT_HDR_CMPID]; } return rc; } int lsmCheckpointLoadOk(lsm_db *pDb, int iSnap){ u32 *aShm; assert( iSnap==1 || iSnap==2 ); aShm = (iSnap==1) ? pDb->pShmhdr->aSnap1 : pDb->pShmhdr->aSnap2; return (lsmCheckpointId(pDb->aSnapshot, 0)==lsmCheckpointId(aShm, 0) ); } int lsmCheckpointClientCacheOk(lsm_db *pDb){ return ( pDb->pClient && pDb->pClient->iId==lsmCheckpointId(pDb->aSnapshot, 0) && pDb->pClient->iId==lsmCheckpointId(pDb->pShmhdr->aSnap1, 0) && pDb->pClient->iId==lsmCheckpointId(pDb->pShmhdr->aSnap2, 0) ); } int lsmCheckpointLoadWorker(lsm_db *pDb){ int rc; ShmHeader *pShm = pDb->pShmhdr; int nInt1; int nInt2; /* Must be holding the WORKER lock to do this. Or DMS2. */ assert( lsmShmAssertLock(pDb, LSM_LOCK_WORKER, LSM_LOCK_EXCL) || lsmShmAssertLock(pDb, LSM_LOCK_DMS1, LSM_LOCK_EXCL) ); /* Check that the two snapshots match. If not, repair them. */ nInt1 = pShm->aSnap1[CKPT_HDR_NCKPT]; nInt2 = pShm->aSnap2[CKPT_HDR_NCKPT]; if( nInt1!=nInt2 || memcmp(pShm->aSnap1, pShm->aSnap2, nInt2*sizeof(u32)) ){ if( ckptChecksumOk(pShm->aSnap1) ){ memcpy(pShm->aSnap2, pShm->aSnap1, sizeof(u32)*nInt1); }else if( ckptChecksumOk(pShm->aSnap2) ){ memcpy(pShm->aSnap1, pShm->aSnap2, sizeof(u32)*nInt2); }else{ return LSM_PROTOCOL_BKPT; } } rc = lsmCheckpointDeserialize(pDb, 1, pShm->aSnap1, &pDb->pWorker); if( pDb->pWorker ) pDb->pWorker->pDatabase = pDb->pDatabase; if( rc==LSM_OK ){ rc = lsmCheckCompressionId(pDb, pDb->pWorker->iCmpId); } #if 0 assert( rc!=LSM_OK || lsmFsIntegrityCheck(pDb) ); #endif return rc; } int lsmCheckpointDeserialize( lsm_db *pDb, int bInclFreelist, /* If true, deserialize free-list */ u32 *aCkpt, Snapshot **ppSnap ){ int rc = LSM_OK; Snapshot *pNew; pNew = (Snapshot *)lsmMallocZeroRc(pDb->pEnv, sizeof(Snapshot), &rc); if( rc==LSM_OK ){ Level *pLvl; int nFree; int i; int nLevel = (int)aCkpt[CKPT_HDR_NLEVEL]; int iIn = CKPT_HDR_SIZE + CKPT_APPENDLIST_SIZE + CKPT_LOGPTR_SIZE; pNew->iId = lsmCheckpointId(aCkpt, 0); pNew->nBlock = aCkpt[CKPT_HDR_NBLOCK]; pNew->nWrite = aCkpt[CKPT_HDR_NWRITE]; rc = ckptLoadLevels(pDb, aCkpt, &iIn, nLevel, &pNew->pLevel); pNew->iLogOff = lsmCheckpointLogOffset(aCkpt); pNew->iCmpId = aCkpt[CKPT_HDR_CMPID]; /* Make a copy of the append-list */ for(i=0; iaiAppend[i] = ckptRead64(a); } /* Read the block-redirect list */ pNew->redirect.n = aCkpt[iIn++]; if( pNew->redirect.n ){ pNew->redirect.a = lsmMallocZeroRc(pDb->pEnv, (sizeof(struct RedirectEntry) * LSM_MAX_BLOCK_REDIRECTS), &rc ); if( rc==LSM_OK ){ for(i=0; iredirect.n; i++){ pNew->redirect.a[i].iFrom = aCkpt[iIn++]; pNew->redirect.a[i].iTo = aCkpt[iIn++]; } } for(pLvl=pNew->pLevel; pLvl->pNext; pLvl=pLvl->pNext); if( pLvl->nRight ){ pLvl->aRhs[pLvl->nRight-1].pRedirect = &pNew->redirect; }else{ pLvl->lhs.pRedirect = &pNew->redirect; } } /* Copy the free-list */ if( rc==LSM_OK && bInclFreelist ){ nFree = aCkpt[iIn++]; if( nFree ){ pNew->freelist.aEntry = (FreelistEntry *)lsmMallocZeroRc( pDb->pEnv, sizeof(FreelistEntry)*nFree, &rc ); if( rc==LSM_OK ){ int j; for(j=0; jfreelist.aEntry[j]; p->iBlk = aCkpt[iIn++]; p->iId = ((i64)(aCkpt[iIn])<<32) + aCkpt[iIn+1]; iIn += 2; } pNew->freelist.nEntry = pNew->freelist.nAlloc = nFree; } } } } if( rc!=LSM_OK ){ lsmFreeSnapshot(pDb->pEnv, pNew); pNew = 0; } *ppSnap = pNew; return rc; } /* ** Connection pDb must be the worker connection in order to call this ** function. It returns true if the database already contains the maximum ** number of levels or false otherwise. ** ** This is used when flushing the in-memory tree to disk. If the database ** is already full, then the caller should invoke lsm_work() or similar ** until it is not full before creating a new level by flushing the in-memory ** tree to disk. Limiting the number of levels in the database ensures that ** the records describing them always fit within the checkpoint blob. */ int lsmDatabaseFull(lsm_db *pDb){ Level *p; int nRhs = 0; assert( lsmShmAssertLock(pDb, LSM_LOCK_WORKER, LSM_LOCK_EXCL) ); assert( pDb->pWorker ); for(p=pDb->pWorker->pLevel; p; p=p->pNext){ nRhs += (p->nRight ? p->nRight : 1); } return (nRhs >= LSM_MAX_RHS_SEGMENTS); } /* ** The connection passed as the only argument is currently the worker ** connection. Some work has been performed on the database by the connection, ** but no new snapshot has been written into shared memory. ** ** This function updates the shared-memory worker and client snapshots with ** the new snapshot produced by the work performed by pDb. ** ** If successful, LSM_OK is returned. Otherwise, if an error occurs, an LSM ** error code is returned. */ int lsmCheckpointSaveWorker(lsm_db *pDb, int bFlush){ Snapshot *pSnap = pDb->pWorker; ShmHeader *pShm = pDb->pShmhdr; void *p = 0; int n = 0; int rc; pSnap->iId++; rc = ckptExportSnapshot(pDb, bFlush, pSnap->iId, 1, &p, &n); if( rc!=LSM_OK ) return rc; assert( ckptChecksumOk((u32 *)p) ); assert( n<=LSM_META_RW_PAGE_SIZE ); memcpy(pShm->aSnap2, p, n); lsmShmBarrier(pDb); memcpy(pShm->aSnap1, p, n); lsmFree(pDb->pEnv, p); /* assert( lsmFsIntegrityCheck(pDb) ); */ return LSM_OK; } /* ** This function is used to determine the snapshot-id of the most recently ** checkpointed snapshot. Variable ShmHeader.iMetaPage indicates which of ** the two meta-pages said snapshot resides on (if any). ** ** If successful, this function loads the snapshot from the meta-page, ** verifies its checksum and sets *piId to the snapshot-id before returning ** LSM_OK. Or, if the checksum attempt fails, *piId is set to zero and ** LSM_OK returned. If an error occurs, an LSM error code is returned and ** the final value of *piId is undefined. */ int lsmCheckpointSynced(lsm_db *pDb, i64 *piId, i64 *piLog, u32 *pnWrite){ int rc = LSM_OK; MetaPage *pPg; u32 iMeta; iMeta = pDb->pShmhdr->iMetaPage; if( iMeta==1 || iMeta==2 ){ rc = lsmFsMetaPageGet(pDb->pFS, 0, iMeta, &pPg); if( rc==LSM_OK ){ int nCkpt; int nData; u8 *aData; aData = lsmFsMetaPageData(pPg, &nData); assert( nData==LSM_META_RW_PAGE_SIZE ); nCkpt = lsmGetU32(&aData[CKPT_HDR_NCKPT*sizeof(u32)]); if( nCkpt<(LSM_META_RW_PAGE_SIZE/sizeof(u32)) ){ u32 *aCopy = lsmMallocRc(pDb->pEnv, sizeof(u32) * nCkpt, &rc); if( aCopy ){ memcpy(aCopy, aData, nCkpt*sizeof(u32)); ckptChangeEndianness(aCopy, nCkpt); if( ckptChecksumOk(aCopy) ){ if( piId ) *piId = lsmCheckpointId(aCopy, 0); if( piLog ) *piLog = (lsmCheckpointLogOffset(aCopy) >> 1); if( pnWrite ) *pnWrite = aCopy[CKPT_HDR_NWRITE]; } lsmFree(pDb->pEnv, aCopy); } } lsmFsMetaPageRelease(pPg); } } if( (iMeta!=1 && iMeta!=2) || rc!=LSM_OK || pDb->pShmhdr->iMetaPage!=iMeta ){ if( piId ) *piId = 0; if( piLog ) *piLog = 0; if( pnWrite ) *pnWrite = 0; } return rc; } /* ** Return the checkpoint-id of the checkpoint array passed as the first ** argument to this function. If the second argument is true, then assume ** that the checkpoint is made up of 32-bit big-endian integers. If it ** is false, assume that the integers are in machine byte order. */ i64 lsmCheckpointId(u32 *aCkpt, int bDisk){ i64 iId; if( bDisk ){ u8 *aData = (u8 *)aCkpt; iId = (((i64)lsmGetU32(&aData[CKPT_HDR_ID_MSW*4])) << 32); iId += ((i64)lsmGetU32(&aData[CKPT_HDR_ID_LSW*4])); }else{ iId = ((i64)aCkpt[CKPT_HDR_ID_MSW] << 32) + (i64)aCkpt[CKPT_HDR_ID_LSW]; } return iId; } u32 lsmCheckpointNBlock(u32 *aCkpt){ return aCkpt[CKPT_HDR_NBLOCK]; } u32 lsmCheckpointNWrite(u32 *aCkpt, int bDisk){ if( bDisk ){ return lsmGetU32((u8 *)&aCkpt[CKPT_HDR_NWRITE]); }else{ return aCkpt[CKPT_HDR_NWRITE]; } } i64 lsmCheckpointLogOffset(u32 *aCkpt){ return ((i64)aCkpt[CKPT_HDR_LO_MSW] << 32) + (i64)aCkpt[CKPT_HDR_LO_LSW]; } int lsmCheckpointPgsz(u32 *aCkpt){ return (int)aCkpt[CKPT_HDR_PGSZ]; } int lsmCheckpointBlksz(u32 *aCkpt){ return (int)aCkpt[CKPT_HDR_BLKSZ]; } void lsmCheckpointLogoffset( u32 *aCkpt, DbLog *pLog ){ pLog->aRegion[2].iStart = (lsmCheckpointLogOffset(aCkpt) >> 1); pLog->cksum0 = aCkpt[CKPT_HDR_LO_CKSUM1]; pLog->cksum1 = aCkpt[CKPT_HDR_LO_CKSUM2]; pLog->iSnapshotId = lsmCheckpointId(aCkpt, 0); } void lsmCheckpointZeroLogoffset(lsm_db *pDb){ u32 nCkpt; nCkpt = pDb->aSnapshot[CKPT_HDR_NCKPT]; assert( nCkpt>CKPT_HDR_NCKPT ); assert( nCkpt==pDb->pShmhdr->aSnap1[CKPT_HDR_NCKPT] ); assert( 0==memcmp(pDb->aSnapshot, pDb->pShmhdr->aSnap1, nCkpt*sizeof(u32)) ); assert( 0==memcmp(pDb->aSnapshot, pDb->pShmhdr->aSnap2, nCkpt*sizeof(u32)) ); pDb->aSnapshot[CKPT_HDR_LO_MSW] = 0; pDb->aSnapshot[CKPT_HDR_LO_LSW] = 0; ckptChecksum(pDb->aSnapshot, nCkpt, &pDb->aSnapshot[nCkpt-2], &pDb->aSnapshot[nCkpt-1] ); memcpy(pDb->pShmhdr->aSnap1, pDb->aSnapshot, nCkpt*sizeof(u32)); memcpy(pDb->pShmhdr->aSnap2, pDb->aSnapshot, nCkpt*sizeof(u32)); } /* ** Set the output variable to the number of KB of data written into the ** database file since the most recent checkpoint. */ int lsmCheckpointSize(lsm_db *db, int *pnKB){ int rc = LSM_OK; u32 nSynced; /* Set nSynced to the number of pages that had been written when the ** database was last checkpointed. */ rc = lsmCheckpointSynced(db, 0, 0, &nSynced); if( rc==LSM_OK ){ u32 nPgsz = db->pShmhdr->aSnap1[CKPT_HDR_PGSZ]; u32 nWrite = db->pShmhdr->aSnap1[CKPT_HDR_NWRITE]; *pnKB = (int)(( ((i64)(nWrite - nSynced) * nPgsz) + 1023) / 1024); } return rc; }