SQLite4
Check-in [3e1ecb95c9]
Not logged in

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

Overview
Comment:Fix memory leaks. Add the LSM_CONFIG_MAX_FREELIST parameter to make testing free-list overflow easier.
Downloads: Tarball | ZIP archive | SQL archive
Timelines: family | ancestors | descendants | both | multi-process
Files: files | file ages | folders
SHA1: 3e1ecb95c9e680f5533d889b4eda95150a8d364a
User & Date: dan 2012-08-30 18:01:26
Context
2012-08-30
20:01
Remove dead code. Run "lomem" tests with max-freelist set to 4. check-in: d6c6889249 user: dan tags: multi-process
18:01
Fix memory leaks. Add the LSM_CONFIG_MAX_FREELIST parameter to make testing free-list overflow easier. check-in: 3e1ecb95c9 user: dan tags: multi-process
2012-08-29
19:45
Fix some of the problems with very large free-block lists. check-in: dc91e55841 user: dan tags: multi-process
Changes
Hide Diffs Unified Diffs Ignore Whitespace Patch

Changes to lsm-test/lsmtest_tdb3.c.

572
573
574
575
576
577
578

579
580
581
582
583
584
585
...
726
727
728
729
730
731
732

733
734
735
736
737
738
739
740
    { "block_size",     0, LSM_CONFIG_BLOCK_SIZE },
    { "safety",         0, LSM_CONFIG_SAFETY },
    { "autowork",       0, LSM_CONFIG_AUTOWORK },
    { "log_size",       0, LSM_CONFIG_LOG_SIZE },
    { "mmap",           0, LSM_CONFIG_MMAP },
    { "use_log",        0, LSM_CONFIG_USE_LOG },
    { "nmerge",         0, LSM_CONFIG_NMERGE },

    { "worker_nmerge",  1, LSM_CONFIG_NMERGE },
    { 0, 0 }
  };
  const char *z = zStr;

  while( z[0] && pDb ){
    const char *zStart;
................................................................................
}

int test_lsm_lomem_open(
  const char *zFilename, 
  int bClear, 
  TestDb **ppDb
){

  const char *zCfg = "page_size=256 block_size=65536 write_buffer=16384";
  return testLsmOpen(zCfg, zFilename, bClear, ppDb);
}

lsm_db *tdb_lsm(TestDb *pDb){
  if( pDb->pMethods->xClose==test_lsm_close ){
    return ((LsmDb *)pDb)->db;
  }







>







 







>
|







572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
...
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
    { "block_size",     0, LSM_CONFIG_BLOCK_SIZE },
    { "safety",         0, LSM_CONFIG_SAFETY },
    { "autowork",       0, LSM_CONFIG_AUTOWORK },
    { "log_size",       0, LSM_CONFIG_LOG_SIZE },
    { "mmap",           0, LSM_CONFIG_MMAP },
    { "use_log",        0, LSM_CONFIG_USE_LOG },
    { "nmerge",         0, LSM_CONFIG_NMERGE },
    { "max_freelist",   0, LSM_CONFIG_MAX_FREELIST },
    { "worker_nmerge",  1, LSM_CONFIG_NMERGE },
    { 0, 0 }
  };
  const char *z = zStr;

  while( z[0] && pDb ){
    const char *zStart;
................................................................................
}

int test_lsm_lomem_open(
  const char *zFilename, 
  int bClear, 
  TestDb **ppDb
){
  const char *zCfg = 
    "page_size=256 block_size=65536 write_buffer=16384 max_freelist=4";
  return testLsmOpen(zCfg, zFilename, bClear, ppDb);
}

lsm_db *tdb_lsm(TestDb *pDb){
  if( pDb->pMethods->xClose==test_lsm_close ){
    return ((LsmDb *)pDb)->db;
  }

Changes to src/lsm.h.

163
164
165
166
167
168
169









170
171
172
173
174
175
176
177
178
179

180
181
182
183
184
185
186
**   LSM_CONFIG_USE_LOG
**     A read/write boolean parameter. True (the default) to use the log
**     file normally. False otherwise.
**
**   LSM_CONFIG_NMERGE
**     A read/write integer parameter. The minimum number of segments to
**     merge together at a time. Default value 4.









*/
#define LSM_CONFIG_WRITE_BUFFER  1
#define LSM_CONFIG_PAGE_SIZE     2
#define LSM_CONFIG_SAFETY        3
#define LSM_CONFIG_BLOCK_SIZE    4
#define LSM_CONFIG_AUTOWORK      5
#define LSM_CONFIG_LOG_SIZE      6
#define LSM_CONFIG_MMAP          7
#define LSM_CONFIG_USE_LOG       8
#define LSM_CONFIG_NMERGE        9


#define LSM_SAFETY_OFF    0
#define LSM_SAFETY_NORMAL 1
#define LSM_SAFETY_FULL   2


/*







>
>
>
>
>
>
>
>
>










>







163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
**   LSM_CONFIG_USE_LOG
**     A read/write boolean parameter. True (the default) to use the log
**     file normally. False otherwise.
**
**   LSM_CONFIG_NMERGE
**     A read/write integer parameter. The minimum number of segments to
**     merge together at a time. Default value 4.
**
**   LSM_CONFIG_MAX_FREELIST
**     A read/write integer parameter. The maximum number of free-list 
**     entries that are stored in a database checkpoint (the others are
**     stored elsewhere in the database).
**
**     There is no reason for an application to configure or query this
**     parameter. It is only present because configuring a small value
**     makes certain parts of the lsm code easier to test.
*/
#define LSM_CONFIG_WRITE_BUFFER  1
#define LSM_CONFIG_PAGE_SIZE     2
#define LSM_CONFIG_SAFETY        3
#define LSM_CONFIG_BLOCK_SIZE    4
#define LSM_CONFIG_AUTOWORK      5
#define LSM_CONFIG_LOG_SIZE      6
#define LSM_CONFIG_MMAP          7
#define LSM_CONFIG_USE_LOG       8
#define LSM_CONFIG_NMERGE        9
#define LSM_CONFIG_MAX_FREELIST 10

#define LSM_SAFETY_OFF    0
#define LSM_SAFETY_NORMAL 1
#define LSM_SAFETY_FULL   2


/*

Changes to src/lsmInt.h.

137
138
139
140
141
142
143






144
145
146
147
148
149
150
...
262
263
264
265
266
267
268

269
270
271
272
273
274
275
...
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470

/* Lock definitions */
#define LSM_LOCK_WRITER       1
#define LSM_LOCK_WORKER       2
#define LSM_LOCK_CHECKPOINTER 3
#define LSM_LOCK_READER(i)    ((i) + LSM_LOCK_CHECKPOINTER + 1)








/*
** A string that can grow by appending.
*/
struct LsmString {
  lsm_env *pEnv;              /* Run-time environment */
  int n;                      /* Size of string.  -1 indicates error */
................................................................................
  int bAutowork;                  /* Configured by LSM_CONFIG_AUTOWORK */
  int nTreeLimit;                 /* Configured by LSM_CONFIG_WRITE_BUFFER */
  int nMerge;                     /* Configured by LSM_CONFIG_NMERGE */
  int nLogSz;                     /* Configured by LSM_CONFIG_LOG_SIZE */
  int bUseLog;                    /* Configured by LSM_CONFIG_USE_LOG */
  int nDfltPgsz;                  /* Configured by LSM_CONFIG_PAGE_SIZE */
  int nDfltBlksz;                 /* Configured by LSM_CONFIG_BLOCK_SIZE */


  /* Sub-system handles */
  FileSystem *pFS;                /* On-disk portion of database */
  Database *pDatabase;            /* Database shared data */

  /* Client transaction context */
  Snapshot *pClient;              /* Client snapshot (non-NULL in read trans) */
................................................................................
};
#define LSM_INITIAL_SNAPSHOT_ID 11

/*
** Functions from file "lsm_ckpt.c".
*/
int lsmCheckpointWrite(lsm_db *);
int lsmCheckpointExport(lsm_db *, int, int, i64, int, void **, int *);
int lsmCheckpointLevels(lsm_db *, int, void **, int *);
int lsmCheckpointLoadLevels(lsm_db *pDb, void *pVal, int nVal);
int lsmCheckpointOverflow(lsm_db *pDb, void **, int *, int *);

int lsmCheckpointRecover(lsm_db *);
int lsmCheckpointDeserialize(lsm_db *, int, u32 *, Snapshot **);








>
>
>
>
>
>







 







>







 







<







137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
...
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
...
463
464
465
466
467
468
469

470
471
472
473
474
475
476

/* Lock definitions */
#define LSM_LOCK_WRITER       1
#define LSM_LOCK_WORKER       2
#define LSM_LOCK_CHECKPOINTER 3
#define LSM_LOCK_READER(i)    ((i) + LSM_LOCK_CHECKPOINTER + 1)

/*
** Hard limit on the number of free-list entries that may be stored in 
** a checkpoint (the remainder are stored as a system record in the LSM).
** See also LSM_CONFIG_MAX_FREELIST.
*/
#define LSM_MAX_FREELIST_ENTRIES 100

/*
** A string that can grow by appending.
*/
struct LsmString {
  lsm_env *pEnv;              /* Run-time environment */
  int n;                      /* Size of string.  -1 indicates error */
................................................................................
  int bAutowork;                  /* Configured by LSM_CONFIG_AUTOWORK */
  int nTreeLimit;                 /* Configured by LSM_CONFIG_WRITE_BUFFER */
  int nMerge;                     /* Configured by LSM_CONFIG_NMERGE */
  int nLogSz;                     /* Configured by LSM_CONFIG_LOG_SIZE */
  int bUseLog;                    /* Configured by LSM_CONFIG_USE_LOG */
  int nDfltPgsz;                  /* Configured by LSM_CONFIG_PAGE_SIZE */
  int nDfltBlksz;                 /* Configured by LSM_CONFIG_BLOCK_SIZE */
  int nMaxFreelist;               /* Configured by LSM_CONFIG_MAX_FREELIST */

  /* Sub-system handles */
  FileSystem *pFS;                /* On-disk portion of database */
  Database *pDatabase;            /* Database shared data */

  /* Client transaction context */
  Snapshot *pClient;              /* Client snapshot (non-NULL in read trans) */
................................................................................
};
#define LSM_INITIAL_SNAPSHOT_ID 11

/*
** Functions from file "lsm_ckpt.c".
*/
int lsmCheckpointWrite(lsm_db *);

int lsmCheckpointLevels(lsm_db *, int, void **, int *);
int lsmCheckpointLoadLevels(lsm_db *pDb, void *pVal, int nVal);
int lsmCheckpointOverflow(lsm_db *pDb, void **, int *, int *);

int lsmCheckpointRecover(lsm_db *);
int lsmCheckpointDeserialize(lsm_db *, int, u32 *, Snapshot **);

Changes to src/lsm_ckpt.c.

109
110
111
112
113
114
115

116
117
118
119
120
121
122
123
124
125
...
128
129
130
131
132
133
134



135
136
137
138
139
140
141
142
143
...
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
...
370
371
372
373
374
375
376








377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
...
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
...
431
432
433
434
435
436
437






438
439
440
441
442
443
444
...
656
657
658
659
660
661
662






663
664
665
666
667
668
669
670
...
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
....
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
** size of all level records in a checkpoint is 12*40=480 integers.
*/
#define LSM_MAX_RHS_SEGMENTS 40

/*
** LARGE NUMBERS OF FREELIST ENTRIES:
**

** A limit 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
................................................................................
**     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.



*/
#define LSM_MAX_FREELIST_ENTRIES 100

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

  for(i=0; i<CKPT_APPENDLIST_SIZE; i++){
    ckptSetValue(p, iOut++, aiAppend[i], pRc);
  }
  *piOut = iOut;
};

int lsmCheckpointExport( 
  lsm_db *pDb,                    /* Connection handle */
  int nOvfl,                      /* Number of free-list entries in LSM */
  int bLog,                       /* True to update log-offset fields */
  i64 iId,                        /* Checkpoint id */
  int bCksum,                     /* If true, include checksums */
  void **ppCkpt,                  /* OUT: Buffer containing checkpoint */
  int *pnCkpt                     /* OUT: Size of checkpoint in bytes */
................................................................................
  int nAll = 0;                   /* Number of levels in db */
  int nHdrLevel = 0;              /* Number of levels in checkpoint */
  int iLevel;                     /* Used to count out nHdrLevel 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;

  /* If nOvfl is negative, copy the value from the previous worker snaphsot. */
  if( nOvfl<0 ){
    nOvfl = (int)(pDb->pShmhdr->aWorker[CKPT_HDR_OVFL]);
  }

  /* 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); iLevel<nHdrLevel; pLevel=pLevel->pNext){
    ckptExportLevel(pLevel, &ckpt, &iOut, &rc);
    iLevel++;
  }

  /* Write the freelist */
  if( rc==LSM_OK ){
    int nFree = (pSnap->freelist.nEntry - nOvfl);
    ckptSetValue(&ckpt, iOut++, nFree, &rc);
    for(i=0; i<nFree; i++){
      FreelistEntry *p = &pSnap->freelist.aEntry[i];
      ckptSetValue(&ckpt, iOut++, p->iBlk, &rc);
      ckptSetValue(&ckpt, iOut++, (p->iId >> 32) & 0xFFFFFFFF, &rc);
      ckptSetValue(&ckpt, iOut++, p->iId & 0xFFFFFFFF, &rc);
    }
................................................................................
  }else{
    ckptSetValue(&ckpt, iOut, 0, &rc);
    ckptSetValue(&ckpt, iOut+1, 0, &rc);
  }
  iOut += 2;
  assert( iOut<=1024 );







  *ppCkpt = (void *)ckpt.aCkpt;
  if( pnCkpt ) *pnCkpt = sizeof(u32)*iOut;
  return rc;
}


/*
................................................................................
  int nRet;
  Snapshot *p = pDb->pWorker;

  assert( lsmShmAssertWorker(pDb) );
  assert( (pnVal==0)==(ppVal==0) );
  assert( pnOvfl );







  nRet = p->freelist.nEntry - LSM_MAX_FREELIST_ENTRIES;
  if( nRet<=0 ){
    nRet = 0;
    if( ppVal ){
      *pnVal = 0;
      *ppVal = 0;
    }
  }else if( ppVal ){
................................................................................
  if( ckptChecksumOk(pShm->aWorker)==0 ){
    int nInt = (int)pShm->aClient[CKPT_HDR_NCKPT];
    memcpy(pShm->aWorker, pShm->aClient, nInt*sizeof(u32));
    if( ckptChecksumOk(pShm->aWorker)==0 ) return LSM_CORRUPT_BKPT;
  }

  rc = lsmCheckpointDeserialize(pDb, 1, pShm->aWorker, &pDb->pWorker);
  if( rc==LSM_OK && pDb->pWorker->nFreelistOvfl ){
    rc = lsmCheckpointLoadOverflow(pDb, &pDb->pWorker->freelist);
    pDb->pWorker->nFreelistOvfl = 0;
  }

  assert( rc!=LSM_OK || lsmFsIntegrityCheck(pDb) );
  return rc;
}

int lsmCheckpointDeserialize(
  lsm_db *pDb, 
  int bInclFreelist,              /* If true, deserialize free-list */
................................................................................
int lsmCheckpointSaveWorker(lsm_db *pDb, int bFlush, int nOvfl){
  Snapshot *pSnap = pDb->pWorker;
  ShmHeader *pShm = pDb->pShmhdr;
  void *p = 0;
  int n = 0;
  int rc;

  rc = lsmCheckpointExport(pDb, nOvfl, bFlush, pSnap->iId+1, 1, &p, &n);
  if( rc!=LSM_OK ) return rc;
  assert( ckptChecksumOk((u32 *)p) );

  assert( n<=LSM_META_PAGE_SIZE );
  memcpy(pShm->aWorker, p, n);
  lsmShmBarrier(pDb);
  memcpy(pShm->aClient, p, n);







>
|
|
|







 







>
>
>

<







 







|







 







>
>
>
>
>
>
>
>






<
<
<
<
<







 







<







 







>
>
>
>
>
>







 







>
>
>
>
>
>
|







 







<
<
<
<
<







 







|







109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
...
129
130
131
132
133
134
135
136
137
138
139

140
141
142
143
144
145
146
...
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
...
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393





394
395
396
397
398
399
400
...
407
408
409
410
411
412
413

414
415
416
417
418
419
420
...
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
...
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
...
956
957
958
959
960
961
962





963
964
965
966
967
968
969
....
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
** size of all level records in a checkpoint is 12*40=480 integers.
*/
#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
................................................................................
**     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.
................................................................................

  for(i=0; i<CKPT_APPENDLIST_SIZE; i++){
    ckptSetValue(p, iOut++, aiAppend[i], pRc);
  }
  *piOut = iOut;
};

static int ckptExportSnapshot( 
  lsm_db *pDb,                    /* Connection handle */
  int nOvfl,                      /* Number of free-list entries in LSM */
  int bLog,                       /* True to update log-offset fields */
  i64 iId,                        /* Checkpoint id */
  int bCksum,                     /* If true, include checksums */
  void **ppCkpt,                  /* OUT: Buffer containing checkpoint */
  int *pnCkpt                     /* OUT: Size of checkpoint in bytes */
................................................................................
  int nAll = 0;                   /* Number of levels in db */
  int nHdrLevel = 0;              /* Number of levels in checkpoint */
  int iLevel;                     /* Used to count out nHdrLevel levels */
  int iOut = 0;                   /* Current offset in aCkpt[] */
  Level *pLevel;                  /* Level iterator */
  int i;                          /* Iterator used while serializing freelist */
  CkptBuffer ckpt;
  int nFree;
 
  nFree = pSnap->freelist.nEntry;
  if( nOvfl>=0 ){
    nFree -=  nOvfl;
  }else{
    nOvfl = pDb->pShmhdr->aWorker[CKPT_HDR_OVFL];
  }

  /* 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); iLevel<nHdrLevel; pLevel=pLevel->pNext){
    ckptExportLevel(pLevel, &ckpt, &iOut, &rc);
    iLevel++;
  }

  /* Write the freelist */
  if( rc==LSM_OK ){

    ckptSetValue(&ckpt, iOut++, nFree, &rc);
    for(i=0; i<nFree; i++){
      FreelistEntry *p = &pSnap->freelist.aEntry[i];
      ckptSetValue(&ckpt, iOut++, p->iBlk, &rc);
      ckptSetValue(&ckpt, iOut++, (p->iId >> 32) & 0xFFFFFFFF, &rc);
      ckptSetValue(&ckpt, iOut++, p->iId & 0xFFFFFFFF, &rc);
    }
................................................................................
  }else{
    ckptSetValue(&ckpt, iOut, 0, &rc);
    ckptSetValue(&ckpt, iOut+1, 0, &rc);
  }
  iOut += 2;
  assert( iOut<=1024 );

#if 0
  lsmLogMessage(pDb, rc, 
      "ckptExportSnapshot(): id=%d freelist: %d/%d", (int)iId, nFree, nOvfl
  );
#endif

  *ppCkpt = (void *)ckpt.aCkpt;
  if( pnCkpt ) *pnCkpt = sizeof(u32)*iOut;
  return rc;
}


/*
................................................................................
  int nRet;
  Snapshot *p = pDb->pWorker;

  assert( lsmShmAssertWorker(pDb) );
  assert( (pnVal==0)==(ppVal==0) );
  assert( pnOvfl );

  if( ppVal && p->nFreelistOvfl ){
    rc = lsmCheckpointLoadOverflow(pDb, &p->freelist);
    if( rc!=LSM_OK ) return rc;
    p->nFreelistOvfl = 0;
  }

  nRet = p->freelist.nEntry - pDb->nMaxFreelist;
  if( nRet<=0 ){
    nRet = 0;
    if( ppVal ){
      *pnVal = 0;
      *ppVal = 0;
    }
  }else if( ppVal ){
................................................................................
  if( ckptChecksumOk(pShm->aWorker)==0 ){
    int nInt = (int)pShm->aClient[CKPT_HDR_NCKPT];
    memcpy(pShm->aWorker, pShm->aClient, nInt*sizeof(u32));
    if( ckptChecksumOk(pShm->aWorker)==0 ) return LSM_CORRUPT_BKPT;
  }

  rc = lsmCheckpointDeserialize(pDb, 1, pShm->aWorker, &pDb->pWorker);





  assert( rc!=LSM_OK || lsmFsIntegrityCheck(pDb) );
  return rc;
}

int lsmCheckpointDeserialize(
  lsm_db *pDb, 
  int bInclFreelist,              /* If true, deserialize free-list */
................................................................................
int lsmCheckpointSaveWorker(lsm_db *pDb, int bFlush, int nOvfl){
  Snapshot *pSnap = pDb->pWorker;
  ShmHeader *pShm = pDb->pShmhdr;
  void *p = 0;
  int n = 0;
  int rc;

  rc = ckptExportSnapshot(pDb, nOvfl, bFlush, pSnap->iId+1, 1, &p, &n);
  if( rc!=LSM_OK ) return rc;
  assert( ckptChecksumOk((u32 *)p) );

  assert( n<=LSM_META_PAGE_SIZE );
  memcpy(pShm->aWorker, p, n);
  lsmShmBarrier(pDb);
  memcpy(pShm->aClient, p, n);

Changes to src/lsm_main.c.

77
78
79
80
81
82
83

84
85
86
87
88
89
90
...
425
426
427
428
429
430
431









432
433
434
435
436
437
438
  pDb->bAutowork = 1;
  pDb->eSafety = LSM_SAFETY_NORMAL;
  pDb->xCmp = xCmp;
  pDb->nLogSz = LSM_DEFAULT_LOG_SIZE;
  pDb->nDfltPgsz = LSM_PAGE_SIZE;
  pDb->nDfltBlksz = LSM_BLOCK_SIZE;
  pDb->nMerge = LSM_DEFAULT_NMERGE;

  pDb->bUseLog = 1;
  pDb->iReader = -1;
  return LSM_OK;
}

lsm_env *lsm_get_env(lsm_db *pDb){
  assert( pDb->pEnv );
................................................................................

    case LSM_CONFIG_NMERGE: {
      int *piVal = va_arg(ap, int *);
      if( *piVal>1 ) pDb->nMerge = *piVal;
      *piVal = pDb->nMerge;
      break;
    }










    default:
      rc = LSM_MISUSE;
      break;
  }

  va_end(ap);







>







 







>
>
>
>
>
>
>
>
>







77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
...
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
  pDb->bAutowork = 1;
  pDb->eSafety = LSM_SAFETY_NORMAL;
  pDb->xCmp = xCmp;
  pDb->nLogSz = LSM_DEFAULT_LOG_SIZE;
  pDb->nDfltPgsz = LSM_PAGE_SIZE;
  pDb->nDfltBlksz = LSM_BLOCK_SIZE;
  pDb->nMerge = LSM_DEFAULT_NMERGE;
  pDb->nMaxFreelist = LSM_MAX_FREELIST_ENTRIES;
  pDb->bUseLog = 1;
  pDb->iReader = -1;
  return LSM_OK;
}

lsm_env *lsm_get_env(lsm_db *pDb){
  assert( pDb->pEnv );
................................................................................

    case LSM_CONFIG_NMERGE: {
      int *piVal = va_arg(ap, int *);
      if( *piVal>1 ) pDb->nMerge = *piVal;
      *piVal = pDb->nMerge;
      break;
    }

    case LSM_CONFIG_MAX_FREELIST: {
      int *piVal = va_arg(ap, int *);
      if( *piVal>=2 && *piVal<=LSM_MAX_FREELIST_ENTRIES ){
        pDb->nMaxFreelist = *piVal;
      }
      *piVal = pDb->nMaxFreelist;
      break;
    }

    default:
      rc = LSM_MISUSE;
      break;
  }

  va_end(ap);

Changes to src/lsm_sorted.c.

1558
1559
1560
1561
1562
1563
1564


1565
1566
1567
1568
1569
1570
1571
1572
1573
....
3632
3633
3634
3635
3636
3637
3638
3639
3640
3641
3642
3643
3644
3645
3646
....
4050
4051
4052
4053
4054
4055
4056

4057
4058

4059

4060
4061
4062
4063
4064
4065
4066
){
  int iRet;
  if( pLeft->pPg==0 ){
    iRet = 1;
  }else if( pRight->pPg==0 ){
    iRet = 0;
  }else{


    int res = pCsr->xCmp(pLeft->pKey, pLeft->nKey, pRight->pKey, pRight->nKey);

    if( res==0 || (res<0 && bLargest==0) || (res>0 && bLargest) ){
      iRet = 0;
    }else{
      iRet = 1;
    }
  }
  return iRet;
................................................................................
    Level *p = pLevel;
    Level **pp;
    pNew->nRight = nMerge;
    pNew->iAge = pLevel->iAge+1;
    for(i=0; i<nMerge; i++){
      pNext = p->pNext;
      pNew->aRhs[i] = p->lhs;
      lsmFree(pDb->pEnv, p);
      p = pNext;
    }

    /* Replace the old levels with the new. */
    pTopLevel = lsmDbSnapshotLevel(pDb->pWorker);
    pNew->pNext = p;
    for(pp=&pTopLevel; *pp!=pLevel; pp=&((*pp)->pNext));
................................................................................
    assert( pDb->pWorker==0 );
    rc = lsmBeginWork(pDb);
    if( rc==LSM_OK ){
      rc = sortedWork(pDb, nPage, bOptimize, &nWrite);
    }

    if( rc==LSM_OK && nWrite ){

      lsmCheckpointOverflow(pDb, 0, 0, &nOvfl);
      rc = lsmSortedFlushDb(pDb);

      if( rc==LSM_OK && nOvfl ) rc = sortedNewToplevel(pDb, 0, &nOvfl);

    }

    if( nWrite ){
      lsmFinishWork(pDb, 0, nOvfl, &rc);
    }else{
      int rcdummy = LSM_BUSY;
      lsmFinishWork(pDb, 0, 0, &rcdummy);







>
>
|
|







 







|







 







>
|

>
|
>







1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
....
3634
3635
3636
3637
3638
3639
3640
3641
3642
3643
3644
3645
3646
3647
3648
....
4052
4053
4054
4055
4056
4057
4058
4059
4060
4061
4062
4063
4064
4065
4066
4067
4068
4069
4070
4071
){
  int iRet;
  if( pLeft->pPg==0 ){
    iRet = 1;
  }else if( pRight->pPg==0 ){
    iRet = 0;
  }else{
    int res = rtTopic(pLeft->eType) - rtTopic(pRight->eType);
    if( res==0 ){
      res = pCsr->xCmp(pLeft->pKey, pLeft->nKey, pRight->pKey, pRight->nKey);
    }
    if( res==0 || (res<0 && bLargest==0) || (res>0 && bLargest) ){
      iRet = 0;
    }else{
      iRet = 1;
    }
  }
  return iRet;
................................................................................
    Level *p = pLevel;
    Level **pp;
    pNew->nRight = nMerge;
    pNew->iAge = pLevel->iAge+1;
    for(i=0; i<nMerge; i++){
      pNext = p->pNext;
      pNew->aRhs[i] = p->lhs;
      sortedFreeLevel(pDb->pEnv, p);
      p = pNext;
    }

    /* Replace the old levels with the new. */
    pTopLevel = lsmDbSnapshotLevel(pDb->pWorker);
    pNew->pNext = p;
    for(pp=&pTopLevel; *pp!=pLevel; pp=&((*pp)->pNext));
................................................................................
    assert( pDb->pWorker==0 );
    rc = lsmBeginWork(pDb);
    if( rc==LSM_OK ){
      rc = sortedWork(pDb, nPage, bOptimize, &nWrite);
    }

    if( rc==LSM_OK && nWrite ){
      int nExpectOvfl = 0;
      lsmCheckpointOverflow(pDb, 0, 0, &nExpectOvfl);
      rc = lsmSortedFlushDb(pDb);
      if( rc==LSM_OK && nExpectOvfl ){
        rc = sortedNewToplevel(pDb, 0, &nOvfl);
      }
    }

    if( nWrite ){
      lsmFinishWork(pDb, 0, nOvfl, &rc);
    }else{
      int rcdummy = LSM_BUSY;
      lsmFinishWork(pDb, 0, 0, &rcdummy);