ext/lsm1/lsm-test/lsmtest_tdb3.c

   1
   2 #include "lsmtest_tdb.h"
   3 #include "lsm.h"
   4 #include "lsmtest.h"
   5
   6 #include <stdlib.h>
   7 #include <string.h>
   8 #include <assert.h>
   9 #ifndef _WIN32
  10 # include <unistd.h>
  11 #endif
  12 #include <stdio.h>
  13
  14 #ifndef _WIN32
  15 # include <sys/time.h>
  16 #endif
  17
  18 typedef struct LsmDb LsmDb;
  19 typedef struct LsmWorker LsmWorker;
  20 typedef struct LsmFile LsmFile;
  21
  22 #define LSMTEST_DFLT_MT_MAX_CKPT (8*1024)
  23 #define LSMTEST_DFLT_MT_MIN_CKPT (2*1024)
  24
  25 #ifdef LSM_MUTEX_PTHREADS
  26 #include <pthread.h>
  27
  28 #define LSMTEST_THREAD_CKPT      1
  29 #define LSMTEST_THREAD_WORKER    2
  30 #define LSMTEST_THREAD_WORKER_AC 3
  31
  32 /*
  33 ** There are several different types of worker threads that run in different
  34 ** test configurations, depending on the value of LsmWorker.eType.
  35 **
  36 **   1. Checkpointer.
  37 **   2. Worker with auto-checkpoint.
  38 **   3. Worker without auto-checkpoint.
  39 */
  40 struct LsmWorker {
  41   LsmDb *pDb;                     /* Main database structure */
  42   lsm_db *pWorker;                /* Worker database handle */
  43   pthread_t worker_thread;        /* Worker thread */
  44   pthread_cond_t worker_cond;     /* Condition var the worker waits on */
  45   pthread_mutex_t worker_mutex;   /* Mutex used with worker_cond */
  46   int bDoWork;                    /* Set to true by client when there is work */
  47   int worker_rc;                  /* Store error code here */
  48   int eType;                      /* LSMTEST_THREAD_XXX constant */
  49   int bBlock;
  50 };
  51 #else
  52 struct LsmWorker { int worker_rc; int bBlock; };
  53 #endif
  54
  55 static void mt_shutdown(LsmDb *);
  56
  57 lsm_env *tdb_lsm_env(void){
  58   static int bInit = 0;
  59   static lsm_env env;
  60   if( bInit==0 ){
  61     memcpy(&env, lsm_default_env(), sizeof(env));
  62     bInit = 1;
  63   }
  64   return &env;
  65 }
  66
  67 typedef struct FileSector FileSector;
  68 typedef struct FileData FileData;
  69
  70 struct FileSector {
  71   u8 *aOld;                       /* Old data for this sector */
  72 };
  73
  74 struct FileData {
  75   int nSector;                    /* Allocated size of apSector[] array */
  76   FileSector *aSector;            /* Array of file sectors */
  77 };
  78
  79 /*
  80 ** bPrepareCrash:
  81 **   If non-zero, the file wrappers maintain enough in-memory data to
  82 **   simulate the effect of a power-failure on the file-system (i.e. that
  83 **   unsynced sectors may be written, not written, or overwritten with
  84 **   arbitrary data when the crash occurs).
  85 **
  86 ** bCrashed:
  87 **   Set to true after a crash is simulated. Once this variable is true, all
  88 **   VFS methods other than xClose() return LSM_IOERR as soon as they are
  89 **   called (without affecting the contents of the file-system).
  90 **
  91 ** env:
  92 **   The environment object used by all lsm_db* handles opened by this
  93 **   object (i.e. LsmDb.db plus any worker connections). Variable env.pVfsCtx
  94 **   always points to the containing LsmDb structure.
  95 */
  96 struct LsmDb {
  97   TestDb base;                    /* Base class - methods table */
  98   lsm_env env;                    /* Environment used by connection db */
  99   char *zName;                    /* Database file name */
 100   lsm_db *db;                     /* LSM database handle */
 101
 102   lsm_cursor *pCsr;               /* Cursor held open during read transaction */
 103   void *pBuf;                     /* Buffer for tdb_fetch() output */
 104   int nBuf;                       /* Allocated (not used) size of pBuf */
 105
 106   /* Crash testing related state */
 107   int bCrashed;                   /* True once a crash has occurred */
 108   int nAutoCrash;                 /* Number of syncs until a crash */
 109   int bPrepareCrash;              /* True to store writes in memory */
 110
 111   /* Unsynced data (while crash testing) */
 112   int szSector;                   /* Assumed size of disk sectors (512B) */
 113   FileData aFile[2];              /* Database and log file data */
 114
 115   /* Other test instrumentation */
 116   int bNoRecovery;                /* If true, assume DMS2 is locked */
 117
 118   /* Work hook redirection */
 119   void (*xWork)(lsm_db *, void *);
 120   void *pWorkCtx;
 121
 122   /* IO logging hook */
 123   void (*xWriteHook)(void *, int, lsm_i64, int, int);
 124   void *pWriteCtx;
 125
 126   /* Worker threads (for lsm_mt) */
 127   int nMtMinCkpt;
 128   int nMtMaxCkpt;
 129   int eMode;
 130   int nWorker;
 131   LsmWorker *aWorker;
 132 };
 133
 134 #define LSMTEST_MODE_SINGLETHREAD    1
 135 #define LSMTEST_MODE_BACKGROUND_CKPT 2
 136 #define LSMTEST_MODE_BACKGROUND_WORK 3
 137 #define LSMTEST_MODE_BACKGROUND_BOTH 4
 138
 139 /*************************************************************************
 140 **************************************************************************
 141 ** Begin test VFS code.
 142 */
 143
 144 struct LsmFile {
 145   lsm_file *pReal;                /* Real underlying file */
 146   int bLog;                       /* True for log file. False for db file */
 147   LsmDb *pDb;                     /* Database handle that uses this file */
 148 };
 149
 150 static int testEnvFullpath(
 151   lsm_env *pEnv,                  /* Environment for current LsmDb */
 152   const char *zFile,              /* Relative path name */
 153   char *zOut,                     /* Output buffer */
 154   int *pnOut                      /* IN/OUT: Size of output buffer */
 155 ){
 156   lsm_env *pRealEnv = tdb_lsm_env();
 157   return pRealEnv->xFullpath(pRealEnv, zFile, zOut, pnOut);
 158 }
 159
 160 static int testEnvOpen(
 161   lsm_env *pEnv,                  /* Environment for current LsmDb */
 162   const char *zFile,              /* Name of file to open */
 163   int flags,
 164   lsm_file **ppFile               /* OUT: New file handle object */
 165 ){
 166   lsm_env *pRealEnv = tdb_lsm_env();
 167   LsmDb *pDb = (LsmDb *)pEnv->pVfsCtx;
 168   int rc;                         /* Return Code */
 169   LsmFile *pRet;                  /* The new file handle */
 170   int nFile;                      /* Length of string zFile in bytes */
 171
 172   nFile = strlen(zFile);
 173   pRet = (LsmFile *)testMalloc(sizeof(LsmFile));
 174   pRet->pDb = pDb;
 175   pRet->bLog = (nFile > 4 && 0==memcmp("-log", &zFile[nFile-4], 4));
 176
 177   rc = pRealEnv->xOpen(pRealEnv, zFile, flags, &pRet->pReal);
 178   if( rc!=LSM_OK ){
 179     testFree(pRet);
 180     pRet = 0;
 181   }
 182
 183   *ppFile = (lsm_file *)pRet;
 184   return rc;
 185 }
 186
 187 static int testEnvRead(lsm_file *pFile, lsm_i64 iOff, void *pData, int nData){
 188   lsm_env *pRealEnv = tdb_lsm_env();
 189   LsmFile *p = (LsmFile *)pFile;
 190   if( p->pDb->bCrashed ) return LSM_IOERR;
 191   return pRealEnv->xRead(p->pReal, iOff, pData, nData);
 192 }
 193
 194 static int testEnvWrite(lsm_file *pFile, lsm_i64 iOff, void *pData, int nData){
 195   lsm_env *pRealEnv = tdb_lsm_env();
 196   LsmFile *p = (LsmFile *)pFile;
 197   LsmDb *pDb = p->pDb;
 198
 199   if( pDb->bCrashed ) return LSM_IOERR;
 200
 201   if( pDb->bPrepareCrash ){
 202     FileData *pData2 = &pDb->aFile[p->bLog];
 203     int iFirst;
 204     int iLast;
 205     int iSector;
 206
 207     iFirst = (int)(iOff / pDb->szSector);
 208     iLast =  (int)((iOff + nData - 1) / pDb->szSector);
 209
 210     if( pData2->nSector<(iLast+1) ){
 211       int nNew = ( ((iLast + 1) + 63) / 64 ) * 64;
 212       assert( nNew>iLast );
 213       pData2->aSector = (FileSector *)testRealloc(
 214           pData2->aSector, nNew*sizeof(FileSector)
 215       );
 216       memset(&pData2->aSector[pData2->nSector],
 217           0, (nNew - pData2->nSector) * sizeof(FileSector)
 218       );
 219       pData2->nSector = nNew;
 220     }
 221
 222     for(iSector=iFirst; iSector<=iLast; iSector++){
 223       if( pData2->aSector[iSector].aOld==0 ){
 224         u8 *aOld = (u8 *)testMalloc(pDb->szSector);
 225         pRealEnv->xRead(
 226             p->pReal, (lsm_i64)iSector*pDb->szSector, aOld, pDb->szSector
 227         );
 228         pData2->aSector[iSector].aOld = aOld;
 229       }
 230     }
 231   }
 232
 233   if( pDb->xWriteHook ){
 234     int rc;
 235     int nUs;
 236     struct timeval t1;
 237     struct timeval t2;
 238
 239     gettimeofday(&t1, 0);
 240     assert( nData>0 );
 241     rc = pRealEnv->xWrite(p->pReal, iOff, pData, nData);
 242     gettimeofday(&t2, 0);
 243
 244     nUs = (t2.tv_sec - t1.tv_sec) * 1000000 + (t2.tv_usec - t1.tv_usec);
 245     pDb->xWriteHook(pDb->pWriteCtx, p->bLog, iOff, nData, nUs);
 246     return rc;
 247   }
 248
 249   return pRealEnv->xWrite(p->pReal, iOff, pData, nData);
 250 }
 251
 252 static void doSystemCrash(LsmDb *pDb);
 253
 254 static int testEnvSync(lsm_file *pFile){
 255   lsm_env *pRealEnv = tdb_lsm_env();
 256   LsmFile *p = (LsmFile *)pFile;
 257   LsmDb *pDb = p->pDb;
 258   FileData *pData = &pDb->aFile[p->bLog];
 259   int i;
 260
 261   if( pDb->bCrashed ) return LSM_IOERR;
 262
 263   if( pDb->nAutoCrash ){
 264     pDb->nAutoCrash--;
 265     if( pDb->nAutoCrash==0 ){
 266       doSystemCrash(pDb);
 267       pDb->bCrashed = 1;
 268       return LSM_IOERR;
 269     }
 270   }
 271
 272   if( pDb->bPrepareCrash ){
 273     for(i=0; i<pData->nSector; i++){
 274       testFree(pData->aSector[i].aOld);
 275       pData->aSector[i].aOld = 0;
 276     }
 277   }
 278
 279   if( pDb->xWriteHook ){
 280     int rc;
 281     int nUs;
 282     struct timeval t1;
 283     struct timeval t2;
 284
 285     gettimeofday(&t1, 0);
 286     rc = pRealEnv->xSync(p->pReal);
 287     gettimeofday(&t2, 0);
 288
 289     nUs = (t2.tv_sec - t1.tv_sec) * 1000000 + (t2.tv_usec - t1.tv_usec);
 290     pDb->xWriteHook(pDb->pWriteCtx, p->bLog, 0, 0, nUs);
 291     return rc;
 292   }
 293
 294   return pRealEnv->xSync(p->pReal);
 295 }
 296
 297 static int testEnvTruncate(lsm_file *pFile, lsm_i64 iOff){
 298   lsm_env *pRealEnv = tdb_lsm_env();
 299   LsmFile *p = (LsmFile *)pFile;
 300   if( p->pDb->bCrashed ) return LSM_IOERR;
 301   return pRealEnv->xTruncate(p->pReal, iOff);
 302 }
 303
 304 static int testEnvSectorSize(lsm_file *pFile){
 305   lsm_env *pRealEnv = tdb_lsm_env();
 306   LsmFile *p = (LsmFile *)pFile;
 307   return pRealEnv->xSectorSize(p->pReal);
 308 }
 309
 310 static int testEnvRemap(
 311   lsm_file *pFile,
 312   lsm_i64 iMin,
 313   void **ppOut,
 314   lsm_i64 *pnOut
 315 ){
 316   lsm_env *pRealEnv = tdb_lsm_env();
 317   LsmFile *p = (LsmFile *)pFile;
 318   return pRealEnv->xRemap(p->pReal, iMin, ppOut, pnOut);
 319 }
 320
 321 static int testEnvFileid(
 322   lsm_file *pFile,
 323   void *ppOut,
 324   int *pnOut
 325 ){
 326   lsm_env *pRealEnv = tdb_lsm_env();
 327   LsmFile *p = (LsmFile *)pFile;
 328   return pRealEnv->xFileid(p->pReal, ppOut, pnOut);
 329 }
 330
 331 static int testEnvClose(lsm_file *pFile){
 332   lsm_env *pRealEnv = tdb_lsm_env();
 333   LsmFile *p = (LsmFile *)pFile;
 334
 335   pRealEnv->xClose(p->pReal);
 336   testFree(p);
 337   return LSM_OK;
 338 }
 339
 340 static int testEnvUnlink(lsm_env *pEnv, const char *zFile){
 341   lsm_env *pRealEnv = tdb_lsm_env();
 342   unused_parameter(pEnv);
 343   return pRealEnv->xUnlink(pRealEnv, zFile);
 344 }
 345
 346 static int testEnvLock(lsm_file *pFile, int iLock, int eType){
 347   LsmFile *p = (LsmFile *)pFile;
 348   lsm_env *pRealEnv = tdb_lsm_env();
 349
 350   if( iLock==2 && eType==LSM_LOCK_EXCL && p->pDb->bNoRecovery ){
 351     return LSM_BUSY;
 352   }
 353   return pRealEnv->xLock(p->pReal, iLock, eType);
 354 }
 355
 356 static int testEnvTestLock(lsm_file *pFile, int iLock, int nLock, int eType){
 357   LsmFile *p = (LsmFile *)pFile;
 358   lsm_env *pRealEnv = tdb_lsm_env();
 359
 360   if( iLock==2 && eType==LSM_LOCK_EXCL && p->pDb->bNoRecovery ){
 361     return LSM_BUSY;
 362   }
 363   return pRealEnv->xTestLock(p->pReal, iLock, nLock, eType);
 364 }
 365
 366 static int testEnvShmMap(lsm_file *pFile, int iRegion, int sz, void **pp){
 367   LsmFile *p = (LsmFile *)pFile;
 368   lsm_env *pRealEnv = tdb_lsm_env();
 369   return pRealEnv->xShmMap(p->pReal, iRegion, sz, pp);
 370 }
 371
 372 static void testEnvShmBarrier(void){
 373 }
 374
 375 static int testEnvShmUnmap(lsm_file *pFile, int bDel){
 376   LsmFile *p = (LsmFile *)pFile;
 377   lsm_env *pRealEnv = tdb_lsm_env();
 378   return pRealEnv->xShmUnmap(p->pReal, bDel);
 379 }
 380
 381 static int testEnvSleep(lsm_env *pEnv, int us){
 382   lsm_env *pRealEnv = tdb_lsm_env();
 383   return pRealEnv->xSleep(pRealEnv, us);
 384 }
 385
 386 static void doSystemCrash(LsmDb *pDb){
 387   lsm_env *pEnv = tdb_lsm_env();
 388   int iFile;
 389   int iSeed = pDb->aFile[0].nSector + pDb->aFile[1].nSector;
 390
 391   char *zFile = pDb->zName;
 392   char *zFree = 0;
 393
 394   for(iFile=0; iFile<2; iFile++){
 395     lsm_file *pFile = 0;
 396     int i;
 397
 398     pEnv->xOpen(pEnv, zFile, 0, &pFile);
 399     for(i=0; i<pDb->aFile[iFile].nSector; i++){
 400       u8 *aOld = pDb->aFile[iFile].aSector[i].aOld;
 401       if( aOld ){
 402         int iOpt = testPrngValue(iSeed++) % 3;
 403         switch( iOpt ){
 404           case 0:
 405             break;
 406
 407           case 1:
 408             testPrngArray(iSeed++, (u32 *)aOld, pDb->szSector/4);
 409             /* Fall-through */
 410
 411           case 2:
 412             pEnv->xWrite(
 413                 pFile, (lsm_i64)i * pDb->szSector, aOld, pDb->szSector
 414             );
 415             break;
 416         }
 417         testFree(aOld);
 418         pDb->aFile[iFile].aSector[i].aOld = 0;
 419       }
 420     }
 421     pEnv->xClose(pFile);
 422     zFree = zFile = sqlite3_mprintf("%s-log", pDb->zName);
 423   }
 424
 425   sqlite3_free(zFree);
 426 }
 427 /*
 428 ** End test VFS code.
 429 **************************************************************************
 430 *************************************************************************/
 431
 432 /*************************************************************************
 433 **************************************************************************
 434 ** Begin test compression hooks.
 435 */
 436
 437 #ifdef HAVE_ZLIB
 438 #include <zlib.h>
 439
 440 static int testZipBound(void *pCtx, int nSrc){
 441   return compressBound(nSrc);
 442 }
 443
 444 static int testZipCompress(
 445   void *pCtx,                     /* Context pointer */
 446   char *aOut, int *pnOut,         /* OUT: Buffer containing compressed data */
 447   const char *aIn, int nIn        /* Buffer containing input data */
 448 ){
 449   uLongf n = *pnOut;              /* In/out buffer size for compress() */
 450   int rc;                         /* compress() return code */
 451
 452   rc = compress((Bytef*)aOut, &n, (Bytef*)aIn, nIn);
 453   *pnOut = n;
 454   return (rc==Z_OK ? 0 : LSM_ERROR);
 455 }
 456
 457 static int testZipUncompress(
 458   void *pCtx,                     /* Context pointer */
 459   char *aOut, int *pnOut,         /* OUT: Buffer containing uncompressed data */
 460   const char *aIn, int nIn        /* Buffer containing input data */
 461 ){
 462   uLongf n = *pnOut;              /* In/out buffer size for uncompress() */
 463   int rc;                         /* uncompress() return code */
 464
 465   rc = uncompress((Bytef*)aOut, &n, (Bytef*)aIn, nIn);
 466   *pnOut = n;
 467   return (rc==Z_OK ? 0 : LSM_ERROR);
 468 }
 469
 470 static int testConfigureCompression(lsm_db *pDb){
 471   static lsm_compress zip = {
 472     0,                            /* Context pointer (unused) */
 473     1,                            /* Id value */
 474     testZipBound,                 /* xBound method */
 475     testZipCompress,              /* xCompress method */
 476     testZipUncompress             /* xUncompress method */
 477   };
 478   return lsm_config(pDb, LSM_CONFIG_SET_COMPRESSION, &zip);
 479 }
 480 #endif /* ifdef HAVE_ZLIB */
 481
 482 /*
 483 ** End test compression hooks.
 484 **************************************************************************
 485 *************************************************************************/
 486
 487 static int test_lsm_close(TestDb *pTestDb){
 488   int i;
 489   int rc = LSM_OK;
 490   LsmDb *pDb = (LsmDb *)pTestDb;
 491
 492   lsm_csr_close(pDb->pCsr);
 493   lsm_close(pDb->db);
 494
 495   /* If this is a multi-threaded database, wait on the worker threads. */
 496   mt_shutdown(pDb);
 497   for(i=0; i<pDb->nWorker && rc==LSM_OK; i++){
 498     rc = pDb->aWorker[i].worker_rc;
 499   }
 500
 501   for(i=0; i<pDb->aFile[0].nSector; i++){
 502     testFree(pDb->aFile[0].aSector[i].aOld);
 503   }
 504   testFree(pDb->aFile[0].aSector);
 505   for(i=0; i<pDb->aFile[1].nSector; i++){
 506     testFree(pDb->aFile[1].aSector[i].aOld);
 507   }
 508   testFree(pDb->aFile[1].aSector);
 509
 510   memset(pDb, sizeof(LsmDb), 0x11);
 511   testFree((char *)pDb->pBuf);
 512   testFree((char *)pDb);
 513   return rc;
 514 }
 515
 516 static void mt_signal_worker(LsmDb*, int);
 517
 518 static int waitOnCheckpointer(LsmDb *pDb, lsm_db *db){
 519   int nSleep = 0;
 520   int nKB;
 521   int rc;
 522
 523   do {
 524     nKB = 0;
 525     rc = lsm_info(db, LSM_INFO_CHECKPOINT_SIZE, &nKB);
 526     if( rc!=LSM_OK || nKB<pDb->nMtMaxCkpt ) break;
 527 #ifdef LSM_MUTEX_PTHREADS
 528     mt_signal_worker(pDb,
 529         (pDb->eMode==LSMTEST_MODE_BACKGROUND_CKPT ? 0 : 1)
 530     );
 531 #endif
 532     usleep(5000);
 533     nSleep += 5;
 534   }while( 1 );
 535
 536 #if 0
 537     if( nSleep ) printf("# waitOnCheckpointer(): nSleep=%d\n", nSleep);
 538 #endif
 539
 540   return rc;
 541 }
 542
 543 static int waitOnWorker(LsmDb *pDb){
 544   int rc;
 545   int nLimit = -1;
 546   int nSleep = 0;
 547
 548   rc = lsm_config(pDb->db, LSM_CONFIG_AUTOFLUSH, &nLimit);
 549   do {
 550     int nOld, nNew, rc2;
 551     rc2 = lsm_info(pDb->db, LSM_INFO_TREE_SIZE, &nOld, &nNew);
 552     if( rc2!=LSM_OK ) return rc2;
 553     if( nOld==0 || nNew<(nLimit/2) ) break;
 554 #ifdef LSM_MUTEX_PTHREADS
 555     mt_signal_worker(pDb, 0);
 556 #endif
 557     usleep(5000);
 558     nSleep += 5;
 559   }while( 1 );
 560
 561 #if 0
 562   if( nSleep ) printf("# waitOnWorker(): nSleep=%d\n", nSleep);
 563 #endif
 564
 565   return rc;
 566 }
 567
 568 static int test_lsm_write(
 569   TestDb *pTestDb,
 570   void *pKey,
 571   int nKey,
 572   void *pVal,
 573   int nVal
 574 ){
 575   LsmDb *pDb = (LsmDb *)pTestDb;
 576   int rc = LSM_OK;
 577
 578   if( pDb->eMode==LSMTEST_MODE_BACKGROUND_CKPT ){
 579     rc = waitOnCheckpointer(pDb, pDb->db);
 580   }else if(
 581       pDb->eMode==LSMTEST_MODE_BACKGROUND_WORK
 582    || pDb->eMode==LSMTEST_MODE_BACKGROUND_BOTH
 583   ){
 584     rc = waitOnWorker(pDb);
 585   }
 586
 587   if( rc==LSM_OK ){
 588     rc = lsm_insert(pDb->db, pKey, nKey, pVal, nVal);
 589   }
 590   return rc;
 591 }
 592
 593 static int test_lsm_delete(TestDb *pTestDb, void *pKey, int nKey){
 594   LsmDb *pDb = (LsmDb *)pTestDb;
 595   return lsm_delete(pDb->db, pKey, nKey);
 596 }
 597
 598 static int test_lsm_delete_range(
 599   TestDb *pTestDb,
 600   void *pKey1, int nKey1,
 601   void *pKey2, int nKey2
 602 ){
 603   LsmDb *pDb = (LsmDb *)pTestDb;
 604   return lsm_delete_range(pDb->db, pKey1, nKey1, pKey2, nKey2);
 605 }
 606
 607 static int test_lsm_fetch(
 608   TestDb *pTestDb,
 609   void *pKey,
 610   int nKey,
 611   void **ppVal,
 612   int *pnVal
 613 ){
 614   int rc;
 615   LsmDb *pDb = (LsmDb *)pTestDb;
 616   lsm_cursor *csr;
 617
 618   if( pKey==0 ) return LSM_OK;
 619
 620   rc = lsm_csr_open(pDb->db, &csr);
 621   if( rc!=LSM_OK ) return rc;
 622
 623   rc = lsm_csr_seek(csr, pKey, nKey, LSM_SEEK_EQ);
 624   if( rc==LSM_OK ){
 625     if( lsm_csr_valid(csr) ){
 626       const void *pVal; int nVal;
 627       rc = lsm_csr_value(csr, &pVal, &nVal);
 628       if( nVal>pDb->nBuf ){
 629         testFree(pDb->pBuf);
 630         pDb->pBuf = testMalloc(nVal*2);
 631         pDb->nBuf = nVal*2;
 632       }
 633       memcpy(pDb->pBuf, pVal, nVal);
 634       *ppVal = pDb->pBuf;
 635       *pnVal = nVal;
 636     }else{
 637       *ppVal = 0;
 638       *pnVal = -1;
 639     }
 640   }
 641   lsm_csr_close(csr);
 642   return rc;
 643 }
 644
 645 static int test_lsm_scan(
 646   TestDb *pTestDb,
 647   void *pCtx,
 648   int bReverse,
 649   void *pFirst, int nFirst,
 650   void *pLast, int nLast,
 651   void (*xCallback)(void *, void *, int , void *, int)
 652 ){
 653   LsmDb *pDb = (LsmDb *)pTestDb;
 654   lsm_cursor *csr;
 655   int rc;
 656
 657   rc = lsm_csr_open(pDb->db, &csr);
 658   if( rc!=LSM_OK ) return rc;
 659
 660   if( bReverse ){
 661     if( pLast ){
 662       rc = lsm_csr_seek(csr, pLast, nLast, LSM_SEEK_LE);
 663     }else{
 664       rc = lsm_csr_last(csr);
 665     }
 666   }else{
 667     if( pFirst ){
 668       rc = lsm_csr_seek(csr, pFirst, nFirst, LSM_SEEK_GE);
 669     }else{
 670       rc = lsm_csr_first(csr);
 671     }
 672   }
 673
 674   while( rc==LSM_OK && lsm_csr_valid(csr) ){
 675     const void *pKey; int nKey;
 676     const void *pVal; int nVal;
 677     int cmp;
 678
 679     lsm_csr_key(csr, &pKey, &nKey);
 680     lsm_csr_value(csr, &pVal, &nVal);
 681
 682     if( bReverse && pFirst ){
 683       cmp = memcmp(pFirst, pKey, MIN(nKey, nFirst));
 684       if( cmp>0 || (cmp==0 && nFirst>nKey) ) break;
 685     }else if( bReverse==0 && pLast ){
 686       cmp = memcmp(pLast, pKey, MIN(nKey, nLast));
 687       if( cmp<0 || (cmp==0 && nLast<nKey) ) break;
 688     }
 689
 690     xCallback(pCtx, (void *)pKey, nKey, (void *)pVal, nVal);
 691
 692     if( bReverse ){
 693       rc = lsm_csr_prev(csr);
 694     }else{
 695       rc = lsm_csr_next(csr);
 696     }
 697   }
 698
 699   lsm_csr_close(csr);
 700   return rc;
 701 }
 702
 703 static int test_lsm_begin(TestDb *pTestDb, int iLevel){
 704   int rc = LSM_OK;
 705   LsmDb *pDb = (LsmDb *)pTestDb;
 706
 707   /* iLevel==0 is a no-op. */
 708   if( iLevel==0 ) return 0;
 709
 710   if( pDb->pCsr==0 ) rc = lsm_csr_open(pDb->db, &pDb->pCsr);
 711   if( rc==LSM_OK && iLevel>1 ){
 712     rc = lsm_begin(pDb->db, iLevel-1);
 713   }
 714
 715   return rc;
 716 }
 717 static int test_lsm_commit(TestDb *pTestDb, int iLevel){
 718   LsmDb *pDb = (LsmDb *)pTestDb;
 719
 720   /* If iLevel==0, close any open read transaction */
 721   if( iLevel==0 && pDb->pCsr ){
 722     lsm_csr_close(pDb->pCsr);
 723     pDb->pCsr = 0;
 724   }
 725
 726   /* If iLevel==0, close any open read transaction */
 727   return lsm_commit(pDb->db, MAX(0, iLevel-1));
 728 }
 729 static int test_lsm_rollback(TestDb *pTestDb, int iLevel){
 730   LsmDb *pDb = (LsmDb *)pTestDb;
 731
 732   /* If iLevel==0, close any open read transaction */
 733   if( iLevel==0 && pDb->pCsr ){
 734     lsm_csr_close(pDb->pCsr);
 735     pDb->pCsr = 0;
 736   }
 737
 738   return lsm_rollback(pDb->db, MAX(0, iLevel-1));
 739 }
 740
 741 /*
 742 ** A log message callback registered with lsm connections. Prints all
 743 ** messages to stderr.
 744 */
 745 static void xLog(void *pCtx, int rc, const char *z){
 746   unused_parameter(rc);
 747   /* fprintf(stderr, "lsm: rc=%d \"%s\"\n", rc, z); */
 748   if( pCtx ) fprintf(stderr, "%s: ", (char *)pCtx);
 749   fprintf(stderr, "%s\n", z);
 750   fflush(stderr);
 751 }
 752
 753 static void xWorkHook(lsm_db *db, void *pArg){
 754   LsmDb *p = (LsmDb *)pArg;
 755   if( p->xWork ) p->xWork(db, p->pWorkCtx);
 756 }
 757
 758 #define TEST_NO_RECOVERY -1
 759 #define TEST_COMPRESSION -3
 760
 761 #define TEST_MT_MODE     -2
 762 #define TEST_MT_MIN_CKPT -4
 763 #define TEST_MT_MAX_CKPT -5
 764
 765 int test_lsm_config_str(
 766   LsmDb *pLsm,
 767   lsm_db *db,
 768   int bWorker,
 769   const char *zStr,
 770   int *pnThread
 771 ){
 772   struct CfgParam {
 773     const char *zParam;
 774     int bWorker;
 775     int eParam;
 776   } aParam[] = {
 777     { "autoflush",        0, LSM_CONFIG_AUTOFLUSH },
 778     { "page_size",        0, LSM_CONFIG_PAGE_SIZE },
 779     { "block_size",       0, LSM_CONFIG_BLOCK_SIZE },
 780     { "safety",           0, LSM_CONFIG_SAFETY },
 781     { "autowork",         0, LSM_CONFIG_AUTOWORK },
 782     { "autocheckpoint",   0, LSM_CONFIG_AUTOCHECKPOINT },
 783     { "mmap",             0, LSM_CONFIG_MMAP },
 784     { "use_log",          0, LSM_CONFIG_USE_LOG },
 785     { "automerge",        0, LSM_CONFIG_AUTOMERGE },
 786     { "max_freelist",     0, LSM_CONFIG_MAX_FREELIST },
 787     { "multi_proc",       0, LSM_CONFIG_MULTIPLE_PROCESSES },
 788     { "worker_automerge", 1, LSM_CONFIG_AUTOMERGE },
 789     { "test_no_recovery", 0, TEST_NO_RECOVERY },
 790     { "bg_min_ckpt",      0, TEST_NO_RECOVERY },
 791
 792     { "mt_mode",          0, TEST_MT_MODE },
 793     { "mt_min_ckpt",      0, TEST_MT_MIN_CKPT },
 794     { "mt_max_ckpt",      0, TEST_MT_MAX_CKPT },
 795
 796 #ifdef HAVE_ZLIB
 797     { "compression",      0, TEST_COMPRESSION },
 798 #endif
 799     { 0, 0 }
 800   };
 801   const char *z = zStr;
 802   int nThread = 1;
 803
 804   if( zStr==0 ) return 0;
 805
 806   assert( db );
 807   while( z[0] ){
 808     const char *zStart;
 809
 810     /* Skip whitespace */
 811     while( *z==' ' ) z++;
 812     zStart = z;
 813
 814     while( *z && *z!='=' ) z++;
 815     if( *z ){
 816       int eParam;
 817       int i;
 818       int iVal;
 819       int iMul = 1;
 820       int rc;
 821       char zParam[32];
 822       int nParam = z-zStart;
 823       if( nParam==0 || nParam>sizeof(zParam)-1 ) goto syntax_error;
 824
 825       memcpy(zParam, zStart, nParam);
 826       zParam[nParam] = '\0';
 827       rc = testArgSelect(aParam, "param", zParam, &i);
 828       if( rc!=0 ) return rc;
 829       eParam = aParam[i].eParam;
 830
 831       z++;
 832       zStart = z;
 833       while( *z>='0' && *z<='9' ) z++;
 834       if( *z=='k' || *z=='K' ){
 835         iMul = 1;
 836         z++;
 837       }else if( *z=='M' || *z=='M' ){
 838         iMul = 1024;
 839         z++;
 840       }
 841       nParam = z-zStart;
 842       if( nParam==0 || nParam>sizeof(zParam)-1 ) goto syntax_error;
 843       memcpy(zParam, zStart, nParam);
 844       zParam[nParam] = '\0';
 845       iVal = atoi(zParam) * iMul;
 846
 847       if( eParam>0 ){
 848         if( bWorker || aParam[i].bWorker==0 ){
 849           lsm_config(db, eParam, &iVal);
 850         }
 851       }else{
 852         switch( eParam ){
 853           case TEST_NO_RECOVERY:
 854             if( pLsm ) pLsm->bNoRecovery = iVal;
 855             break;
 856           case TEST_MT_MODE:
 857             if( pLsm ) nThread = iVal;
 858             break;
 859           case TEST_MT_MIN_CKPT:
 860             if( pLsm && iVal>0 ) pLsm->nMtMinCkpt = iVal*1024;
 861             break;
 862           case TEST_MT_MAX_CKPT:
 863             if( pLsm && iVal>0 ) pLsm->nMtMaxCkpt = iVal*1024;
 864             break;
 865 #ifdef HAVE_ZLIB
 866           case TEST_COMPRESSION:
 867             testConfigureCompression(db);
 868             break;
 869 #endif
 870         }
 871       }
 872     }else if( z!=zStart ){
 873       goto syntax_error;
 874     }
 875   }
 876
 877   if( pnThread ) *pnThread = nThread;
 878   if( pLsm && pLsm->nMtMaxCkpt < pLsm->nMtMinCkpt ){
 879     pLsm->nMtMinCkpt = pLsm->nMtMaxCkpt;
 880   }
 881
 882   return 0;
 883  syntax_error:
 884   testPrintError("syntax error at: \"%s\"\n", z);
 885   return 1;
 886 }
 887
 888 int tdb_lsm_config_str(TestDb *pDb, const char *zStr){
 889   int rc = 0;
 890   if( tdb_lsm(pDb) ){
 891 #ifdef LSM_MUTEX_PTHREADS
 892     int i;
 893 #endif
 894     LsmDb *pLsm = (LsmDb *)pDb;
 895
 896     rc = test_lsm_config_str(pLsm, pLsm->db, 0, zStr, 0);
 897 #ifdef LSM_MUTEX_PTHREADS
 898     for(i=0; rc==0 && i<pLsm->nWorker; i++){
 899       rc = test_lsm_config_str(0, pLsm->aWorker[i].pWorker, 1, zStr, 0);
 900     }
 901 #endif
 902   }
 903   return rc;
 904 }
 905
 906 int tdb_lsm_configure(lsm_db *db, const char *zConfig){
 907   return test_lsm_config_str(0, db, 0, zConfig, 0);
 908 }
 909
 910 static int testLsmStartWorkers(LsmDb *, int, const char *, const char *);
 911
 912 static int testLsmOpen(
 913   const char *zCfg,
 914   const char *zFilename,
 915   int bClear,
 916   TestDb **ppDb
 917 ){
 918   static const DatabaseMethods LsmMethods = {
 919     test_lsm_close,
 920     test_lsm_write,
 921     test_lsm_delete,
 922     test_lsm_delete_range,
 923     test_lsm_fetch,
 924     test_lsm_scan,
 925     test_lsm_begin,
 926     test_lsm_commit,
 927     test_lsm_rollback
 928   };
 929
 930   int rc;
 931   int nFilename;
 932   LsmDb *pDb;
 933
 934   /* If the bClear flag is set, delete any existing database. */
 935   assert( zFilename);
 936   if( bClear ) testDeleteLsmdb(zFilename);
 937   nFilename = strlen(zFilename);
 938
 939   pDb = (LsmDb *)testMalloc(sizeof(LsmDb) + nFilename + 1);
 940   memset(pDb, 0, sizeof(LsmDb));
 941   pDb->base.pMethods = &LsmMethods;
 942   pDb->zName = (char *)&pDb[1];
 943   memcpy(pDb->zName, zFilename, nFilename + 1);
 944
 945   /* Default the sector size used for crash simulation to 512 bytes.
 946   ** Todo: There should be an OS method to obtain this value - just as
 947   ** there is in SQLite. For now, LSM assumes that it is smaller than
 948   ** the page size (default 4KB).
 949   */
 950   pDb->szSector = 256;
 951
 952   /* Default values for the mt_min_ckpt and mt_max_ckpt parameters. */
 953   pDb->nMtMinCkpt = LSMTEST_DFLT_MT_MIN_CKPT;
 954   pDb->nMtMaxCkpt = LSMTEST_DFLT_MT_MAX_CKPT;
 955
 956   memcpy(&pDb->env, tdb_lsm_env(), sizeof(lsm_env));
 957   pDb->env.pVfsCtx = (void *)pDb;
 958   pDb->env.xFullpath = testEnvFullpath;
 959   pDb->env.xOpen = testEnvOpen;
 960   pDb->env.xRead = testEnvRead;
 961   pDb->env.xWrite = testEnvWrite;
 962   pDb->env.xTruncate = testEnvTruncate;
 963   pDb->env.xSync = testEnvSync;
 964   pDb->env.xSectorSize = testEnvSectorSize;
 965   pDb->env.xRemap = testEnvRemap;
 966   pDb->env.xFileid = testEnvFileid;
 967   pDb->env.xClose = testEnvClose;
 968   pDb->env.xUnlink = testEnvUnlink;
 969   pDb->env.xLock = testEnvLock;
 970   pDb->env.xTestLock = testEnvTestLock;
 971   pDb->env.xShmBarrier = testEnvShmBarrier;
 972   pDb->env.xShmMap = testEnvShmMap;
 973   pDb->env.xShmUnmap = testEnvShmUnmap;
 974   pDb->env.xSleep = testEnvSleep;
 975
 976   rc = lsm_new(&pDb->env, &pDb->db);
 977   if( rc==LSM_OK ){
 978     int nThread = 1;
 979     lsm_config_log(pDb->db, xLog, 0);
 980     lsm_config_work_hook(pDb->db, xWorkHook, (void *)pDb);
 981
 982     rc = test_lsm_config_str(pDb, pDb->db, 0, zCfg, &nThread);
 983     if( rc==LSM_OK ) rc = lsm_open(pDb->db, zFilename);
 984
 985     pDb->eMode = nThread;
 986 #ifdef LSM_MUTEX_PTHREADS
 987     if( rc==LSM_OK && nThread>1 ){
 988       testLsmStartWorkers(pDb, nThread, zFilename, zCfg);
 989     }
 990 #endif
 991
 992     if( rc!=LSM_OK ){
 993       test_lsm_close((TestDb *)pDb);
 994       pDb = 0;
 995     }
 996   }
 997
 998   *ppDb = (TestDb *)pDb;
 999   return rc;
1000 }
1001
1002 int test_lsm_open(
1003   const char *zSpec,
1004   const char *zFilename,
1005   int bClear,
1006   TestDb **ppDb
1007 ){
1008   return testLsmOpen(zSpec, zFilename, bClear, ppDb);
1009 }
1010
1011 int test_lsm_small_open(
1012   const char *zSpec,
1013   const char *zFile,
1014   int bClear,
1015   TestDb **ppDb
1016 ){
1017   const char *zCfg = "page_size=256 block_size=64 mmap=1024";
1018   return testLsmOpen(zCfg, zFile, bClear, ppDb);
1019 }
1020
1021 int test_lsm_lomem_open(
1022   const char *zSpec,
1023   const char *zFilename,
1024   int bClear,
1025   TestDb **ppDb
1026 ){
1027     /* "max_freelist=4 autocheckpoint=32" */
1028   const char *zCfg =
1029     "page_size=256 block_size=64 autoflush=16 "
1030     "autocheckpoint=32"
1031     "mmap=0 "
1032   ;
1033   return testLsmOpen(zCfg, zFilename, bClear, ppDb);
1034 }
1035
1036 int test_lsm_lomem2_open(
1037   const char *zSpec,
1038   const char *zFilename,
1039   int bClear,
1040   TestDb **ppDb
1041 ){
1042     /* "max_freelist=4 autocheckpoint=32" */
1043   const char *zCfg =
1044     "page_size=512 block_size=64 autoflush=0 mmap=0 "
1045   ;
1046   return testLsmOpen(zCfg, zFilename, bClear, ppDb);
1047 }
1048
1049 int test_lsm_zip_open(
1050   const char *zSpec,
1051   const char *zFilename,
1052   int bClear,
1053   TestDb **ppDb
1054 ){
1055   const char *zCfg =
1056     "page_size=256 block_size=64 autoflush=16 "
1057     "autocheckpoint=32 compression=1 mmap=0 "
1058   ;
1059   return testLsmOpen(zCfg, zFilename, bClear, ppDb);
1060 }
1061
1062 lsm_db *tdb_lsm(TestDb *pDb){
1063   if( pDb->pMethods->xClose==test_lsm_close ){
1064     return ((LsmDb *)pDb)->db;
1065   }
1066   return 0;
1067 }
1068
1069 int tdb_lsm_multithread(TestDb *pDb){
1070   int ret = 0;
1071   if( tdb_lsm(pDb) ){
1072     ret = ((LsmDb*)pDb)->eMode!=LSMTEST_MODE_SINGLETHREAD;
1073   }
1074   return ret;
1075 }
1076
1077 void tdb_lsm_enable_log(TestDb *pDb, int bEnable){
1078   lsm_db *db = tdb_lsm(pDb);
1079   if( db ){
1080     lsm_config_log(db, (bEnable ? xLog : 0), (void *)"client");
1081   }
1082 }
1083
1084 void tdb_lsm_application_crash(TestDb *pDb){
1085   if( tdb_lsm(pDb) ){
1086     LsmDb *p = (LsmDb *)pDb;
1087     p->bCrashed = 1;
1088   }
1089 }
1090
1091 void tdb_lsm_prepare_system_crash(TestDb *pDb){
1092   if( tdb_lsm(pDb) ){
1093     LsmDb *p = (LsmDb *)pDb;
1094     p->bPrepareCrash = 1;
1095   }
1096 }
1097
1098 void tdb_lsm_system_crash(TestDb *pDb){
1099   if( tdb_lsm(pDb) ){
1100     LsmDb *p = (LsmDb *)pDb;
1101     p->bCrashed = 1;
1102     doSystemCrash(p);
1103   }
1104 }
1105
1106 void tdb_lsm_safety(TestDb *pDb, int eMode){
1107   assert( eMode==LSM_SAFETY_OFF
1108        || eMode==LSM_SAFETY_NORMAL
1109        || eMode==LSM_SAFETY_FULL
1110   );
1111   if( tdb_lsm(pDb) ){
1112     int iParam = eMode;
1113     LsmDb *p = (LsmDb *)pDb;
1114     lsm_config(p->db, LSM_CONFIG_SAFETY, &iParam);
1115   }
1116 }
1117
1118 void tdb_lsm_prepare_sync_crash(TestDb *pDb, int iSync){
1119   assert( iSync>0 );
1120   if( tdb_lsm(pDb) ){
1121     LsmDb *p = (LsmDb *)pDb;
1122     p->nAutoCrash = iSync;
1123     p->bPrepareCrash = 1;
1124   }
1125 }
1126
1127 void tdb_lsm_config_work_hook(
1128   TestDb *pDb,
1129   void (*xWork)(lsm_db *, void *),
1130   void *pWorkCtx
1131 ){
1132   if( tdb_lsm(pDb) ){
1133     LsmDb *p = (LsmDb *)pDb;
1134     p->xWork = xWork;
1135     p->pWorkCtx = pWorkCtx;
1136   }
1137 }
1138
1139 void tdb_lsm_write_hook(
1140   TestDb *pDb,
1141   void (*xWrite)(void *, int, lsm_i64, int, int),
1142   void *pWriteCtx
1143 ){
1144   if( tdb_lsm(pDb) ){
1145     LsmDb *p = (LsmDb *)pDb;
1146     p->xWriteHook = xWrite;
1147     p->pWriteCtx = pWriteCtx;
1148   }
1149 }
1150
1151 int tdb_lsm_open(const char *zCfg, const char *zDb, int bClear, TestDb **ppDb){
1152   return testLsmOpen(zCfg, zDb, bClear, ppDb);
1153 }
1154
1155 #ifdef LSM_MUTEX_PTHREADS
1156
1157 /*
1158 ** Signal worker thread iWorker that there may be work to do.
1159 */
1160 static void mt_signal_worker(LsmDb *pDb, int iWorker){
1161   LsmWorker *p = &pDb->aWorker[iWorker];
1162   pthread_mutex_lock(&p->worker_mutex);
1163   p->bDoWork = 1;
1164   pthread_cond_signal(&p->worker_cond);
1165   pthread_mutex_unlock(&p->worker_mutex);
1166 }
1167
1168 /*
1169 ** This routine is used as the main() for all worker threads.
1170 */
1171 static void *worker_main(void *pArg){
1172   LsmWorker *p = (LsmWorker *)pArg;
1173   lsm_db *pWorker;                /* Connection to access db through */
1174
1175   pthread_mutex_lock(&p->worker_mutex);
1176   while( (pWorker = p->pWorker) ){
1177     int rc = LSM_OK;
1178
1179     /* Do some work. If an error occurs, exit. */
1180
1181     pthread_mutex_unlock(&p->worker_mutex);
1182     if( p->eType==LSMTEST_THREAD_CKPT ){
1183       int nKB = 0;
1184       rc = lsm_info(pWorker, LSM_INFO_CHECKPOINT_SIZE, &nKB);
1185       if( rc==LSM_OK && nKB>=p->pDb->nMtMinCkpt ){
1186         rc = lsm_checkpoint(pWorker, 0);
1187       }
1188     }else{
1189       int nWrite;
1190       do {
1191
1192         if( p->eType==LSMTEST_THREAD_WORKER ){
1193           waitOnCheckpointer(p->pDb, pWorker);
1194         }
1195
1196         nWrite = 0;
1197         rc = lsm_work(pWorker, 0, 256, &nWrite);
1198
1199         if( p->eType==LSMTEST_THREAD_WORKER && nWrite ){
1200           mt_signal_worker(p->pDb, 1);
1201         }
1202       }while( nWrite && p->pWorker );
1203     }
1204     pthread_mutex_lock(&p->worker_mutex);
1205
1206     if( rc!=LSM_OK && rc!=LSM_BUSY ){
1207       p->worker_rc = rc;
1208       break;
1209     }
1210
1211     /* The thread will wake up when it is signaled either because another
1212     ** thread has created some work for this one or because the connection
1213     ** is being closed.  */
1214     if( p->pWorker && p->bDoWork==0 ){
1215       pthread_cond_wait(&p->worker_cond, &p->worker_mutex);
1216     }
1217     p->bDoWork = 0;
1218   }
1219   pthread_mutex_unlock(&p->worker_mutex);
1220
1221   return 0;
1222 }
1223
1224
1225 static void mt_stop_worker(LsmDb *pDb, int iWorker){
1226   LsmWorker *p = &pDb->aWorker[iWorker];
1227   if( p->pWorker ){
1228     void *pDummy;
1229     lsm_db *pWorker;
1230
1231     /* Signal the worker to stop */
1232     pthread_mutex_lock(&p->worker_mutex);
1233     pWorker = p->pWorker;
1234     p->pWorker = 0;
1235     pthread_cond_signal(&p->worker_cond);
1236     pthread_mutex_unlock(&p->worker_mutex);
1237
1238     /* Join the worker thread. */
1239     pthread_join(p->worker_thread, &pDummy);
1240
1241     /* Free resources allocated in mt_start_worker() */
1242     pthread_cond_destroy(&p->worker_cond);
1243     pthread_mutex_destroy(&p->worker_mutex);
1244     lsm_close(pWorker);
1245   }
1246 }
1247
1248 static void mt_shutdown(LsmDb *pDb){
1249   int i;
1250   for(i=0; i<pDb->nWorker; i++){
1251     mt_stop_worker(pDb, i);
1252   }
1253 }
1254
1255 /*
1256 ** This callback is invoked by LSM when the client database writes to
1257 ** the database file (i.e. to flush the contents of the in-memory tree).
1258 ** This implies there may be work to do on the database, so signal
1259 ** the worker threads.
1260 */
1261 static void mt_client_work_hook(lsm_db *db, void *pArg){
1262   LsmDb *pDb = (LsmDb *)pArg;     /* LsmDb database handle */
1263
1264   /* Invoke the user level work-hook, if any. */
1265   if( pDb->xWork ) pDb->xWork(db, pDb->pWorkCtx);
1266
1267   /* Wake up worker thread 0. */
1268   mt_signal_worker(pDb, 0);
1269 }
1270
1271 static void mt_worker_work_hook(lsm_db *db, void *pArg){
1272   LsmDb *pDb = (LsmDb *)pArg;     /* LsmDb database handle */
1273
1274   /* Invoke the user level work-hook, if any. */
1275   if( pDb->xWork ) pDb->xWork(db, pDb->pWorkCtx);
1276 }
1277
1278 /*
1279 ** Launch worker thread iWorker for database connection pDb.
1280 */
1281 static int mt_start_worker(
1282   LsmDb *pDb,                     /* Main database structure */
1283   int iWorker,                    /* Worker number to start */
1284   const char *zFilename,          /* File name of database to open */
1285   const char *zCfg,               /* Connection configuration string */
1286   int eType                       /* Type of worker thread */
1287 ){
1288   int rc = 0;                     /* Return code */
1289   LsmWorker *p;                   /* Object to initialize */
1290
1291   assert( iWorker<pDb->nWorker );
1292   assert( eType==LSMTEST_THREAD_CKPT
1293        || eType==LSMTEST_THREAD_WORKER
1294        || eType==LSMTEST_THREAD_WORKER_AC
1295   );
1296
1297   p = &pDb->aWorker[iWorker];
1298   p->eType = eType;
1299   p->pDb = pDb;
1300
1301   /* Open the worker connection */
1302   if( rc==0 ) rc = lsm_new(&pDb->env, &p->pWorker);
1303   if( zCfg ){
1304     test_lsm_config_str(pDb, p->pWorker, 1, zCfg, 0);
1305   }
1306   if( rc==0 ) rc = lsm_open(p->pWorker, zFilename);
1307   lsm_config_log(p->pWorker, xLog, (void *)"worker");
1308
1309   /* Configure the work-hook */
1310   if( rc==0 ){
1311     lsm_config_work_hook(p->pWorker, mt_worker_work_hook, (void *)pDb);
1312   }
1313
1314   if( eType==LSMTEST_THREAD_WORKER ){
1315     test_lsm_config_str(0, p->pWorker, 1, "autocheckpoint=0", 0);
1316   }
1317
1318   /* Kick off the worker thread. */
1319   if( rc==0 ) rc = pthread_cond_init(&p->worker_cond, 0);
1320   if( rc==0 ) rc = pthread_mutex_init(&p->worker_mutex, 0);
1321   if( rc==0 ) rc = pthread_create(&p->worker_thread, 0, worker_main, (void *)p);
1322
1323   return rc;
1324 }
1325
1326
1327 static int testLsmStartWorkers(
1328   LsmDb *pDb, int eModel, const char *zFilename, const char *zCfg
1329 ){
1330   int rc;
1331
1332   if( eModel<1 || eModel>4 ) return 1;
1333   if( eModel==1 ) return 0;
1334
1335   /* Configure a work-hook for the client connection. Worker 0 is signalled
1336   ** every time the users connection writes to the database.  */
1337   lsm_config_work_hook(pDb->db, mt_client_work_hook, (void *)pDb);
1338
1339   /* Allocate space for two worker connections. They may not both be
1340   ** used, but both are allocated.  */
1341   pDb->aWorker = (LsmWorker *)testMalloc(sizeof(LsmWorker) * 2);
1342   memset(pDb->aWorker, 0, sizeof(LsmWorker) * 2);
1343
1344   switch( eModel ){
1345     case LSMTEST_MODE_BACKGROUND_CKPT:
1346       pDb->nWorker = 1;
1347       test_lsm_config_str(0, pDb->db, 0, "autocheckpoint=0", 0);
1348       rc = mt_start_worker(pDb, 0, zFilename, zCfg, LSMTEST_THREAD_CKPT);
1349       break;
1350
1351     case LSMTEST_MODE_BACKGROUND_WORK:
1352       pDb->nWorker = 1;
1353       test_lsm_config_str(0, pDb->db, 0, "autowork=0", 0);
1354       rc = mt_start_worker(pDb, 0, zFilename, zCfg, LSMTEST_THREAD_WORKER_AC);
1355       break;
1356
1357     case LSMTEST_MODE_BACKGROUND_BOTH:
1358       pDb->nWorker = 2;
1359       test_lsm_config_str(0, pDb->db, 0, "autowork=0", 0);
1360       rc = mt_start_worker(pDb, 0, zFilename, zCfg, LSMTEST_THREAD_WORKER);
1361       if( rc==0 ){
1362         rc = mt_start_worker(pDb, 1, zFilename, zCfg, LSMTEST_THREAD_CKPT);
1363       }
1364       break;
1365   }
1366
1367   return rc;
1368 }
1369
1370
1371 int test_lsm_mt2(
1372   const char *zSpec,
1373   const char *zFilename,
1374   int bClear,
1375   TestDb **ppDb
1376 ){
1377   const char *zCfg = "mt_mode=2";
1378   return testLsmOpen(zCfg, zFilename, bClear, ppDb);
1379 }
1380
1381 int test_lsm_mt3(
1382   const char *zSpec,
1383   const char *zFilename,
1384   int bClear,
1385   TestDb **ppDb
1386 ){
1387   const char *zCfg = "mt_mode=4";
1388   return testLsmOpen(zCfg, zFilename, bClear, ppDb);
1389 }
1390
1391 #else
1392 static void mt_shutdown(LsmDb *pDb) {
1393   unused_parameter(pDb);
1394 }
1395 int test_lsm_mt(const char *zFilename, int bClear, TestDb **ppDb){
1396   unused_parameter(zFilename);
1397   unused_parameter(bClear);
1398   unused_parameter(ppDb);
1399   testPrintError("threads unavailable - recompile with LSM_MUTEX_PTHREADS\n");
1400   return 1;
1401 }
1402 #endif