1 /* 2 ** 2012-01-23 3 ** 4 ** The author disclaims copyright to this source code. In place of 5 ** a legal notice, here is a blessing: 6 ** 7 ** May you do good and not evil. 8 ** May you find forgiveness for yourself and forgive others. 9 ** May you share freely, never taking more than you give. 10 ** 11 ************************************************************************* 12 ** 13 ** Utilities used to help multiple LSM clients to coexist within the 14 ** same process space. 15 */ 16 #include "lsmInt.h" 17 18 /* 19 ** Global data. All global variables used by code in this file are grouped 20 ** into the following structure instance. 21 ** 22 ** pDatabase: 23 ** Linked list of all Database objects allocated within this process. 24 ** This list may not be traversed without holding the global mutex (see 25 ** functions enterGlobalMutex() and leaveGlobalMutex()). 26 */ 27 static struct SharedData { 28 Database *pDatabase; /* Linked list of all Database objects */ 29 } gShared; 30 31 /* 32 ** Database structure. There is one such structure for each distinct 33 ** database accessed by this process. They are stored in the singly linked 34 ** list starting at global variable gShared.pDatabase. Database objects are 35 ** reference counted. Once the number of connections to the associated 36 ** database drops to zero, they are removed from the linked list and deleted. 37 ** 38 ** pFile: 39 ** In multi-process mode, this file descriptor is used to obtain locks 40 ** and to access shared-memory. In single process mode, its only job is 41 ** to hold the exclusive lock on the file. 42 ** 43 */ 44 struct Database { 45 /* Protected by the global mutex (enterGlobalMutex/leaveGlobalMutex): */ 46 char *zName; /* Canonical path to database file */ 47 int nName; /* strlen(zName) */ 48 int nDbRef; /* Number of associated lsm_db handles */ 49 Database *pDbNext; /* Next Database structure in global list */ 50 51 /* Protected by the local mutex (pClientMutex) */ 52 int bReadonly; /* True if Database.pFile is read-only */ 53 int bMultiProc; /* True if running in multi-process mode */ 54 lsm_file *pFile; /* Used for locks/shm in multi-proc mode */ 55 LsmFile *pLsmFile; /* List of deferred closes */ 56 lsm_mutex *pClientMutex; /* Protects the apShmChunk[] and pConn */ 57 int nShmChunk; /* Number of entries in apShmChunk[] array */ 58 void **apShmChunk; /* Array of "shared" memory regions */ 59 lsm_db *pConn; /* List of connections to this db. */ 60 }; 61 62 /* 63 ** Functions to enter and leave the global mutex. This mutex is used 64 ** to protect the global linked-list headed at gShared.pDatabase. 65 */ 66 static int enterGlobalMutex(lsm_env *pEnv){ 67 lsm_mutex *p; 68 int rc = lsmMutexStatic(pEnv, LSM_MUTEX_GLOBAL, &p); 69 if( rc==LSM_OK ) lsmMutexEnter(pEnv, p); 70 return rc; 71 } 72 static void leaveGlobalMutex(lsm_env *pEnv){ 73 lsm_mutex *p; 74 lsmMutexStatic(pEnv, LSM_MUTEX_GLOBAL, &p); 75 lsmMutexLeave(pEnv, p); 76 } 77 78 #ifdef LSM_DEBUG 79 static int holdingGlobalMutex(lsm_env *pEnv){ 80 lsm_mutex *p; 81 lsmMutexStatic(pEnv, LSM_MUTEX_GLOBAL, &p); 82 return lsmMutexHeld(pEnv, p); 83 } 84 #endif 85 86 #if 0 87 static void assertNotInFreelist(Freelist *p, int iBlk){ 88 int i; 89 for(i=0; i<p->nEntry; i++){ 90 assert( p->aEntry[i].iBlk!=iBlk ); 91 } 92 } 93 #else 94 # define assertNotInFreelist(x,y) 95 #endif 96 97 /* 98 ** Append an entry to the free-list. If (iId==-1), this is a delete. 99 */ 100 int freelistAppend(lsm_db *db, int iBlk, i64 iId){ 101 lsm_env *pEnv = db->pEnv; 102 Freelist *p; 103 int i; 104 105 assert( iId==-1 || iId>=0 ); 106 p = db->bUseFreelist ? db->pFreelist : &db->pWorker->freelist; 107 108 /* Extend the space allocated for the freelist, if required */ 109 assert( p->nAlloc>=p->nEntry ); 110 if( p->nAlloc==p->nEntry ){ 111 int nNew; 112 int nByte; 113 FreelistEntry *aNew; 114 115 nNew = (p->nAlloc==0 ? 4 : p->nAlloc*2); 116 nByte = sizeof(FreelistEntry) * nNew; 117 aNew = (FreelistEntry *)lsmRealloc(pEnv, p->aEntry, nByte); 118 if( !aNew ) return LSM_NOMEM_BKPT; 119 p->nAlloc = nNew; 120 p->aEntry = aNew; 121 } 122 123 for(i=0; i<p->nEntry; i++){ 124 assert( i==0 || p->aEntry[i].iBlk > p->aEntry[i-1].iBlk ); 125 if( p->aEntry[i].iBlk>=iBlk ) break; 126 } 127 128 if( i<p->nEntry && p->aEntry[i].iBlk==iBlk ){ 129 /* Clobber an existing entry */ 130 p->aEntry[i].iId = iId; 131 }else{ 132 /* Insert a new entry into the list */ 133 int nByte = sizeof(FreelistEntry)*(p->nEntry-i); 134 memmove(&p->aEntry[i+1], &p->aEntry[i], nByte); 135 p->aEntry[i].iBlk = iBlk; 136 p->aEntry[i].iId = iId; 137 p->nEntry++; 138 } 139 140 return LSM_OK; 141 } 142 143 /* 144 ** This function frees all resources held by the Database structure passed 145 ** as the only argument. 146 */ 147 static void freeDatabase(lsm_env *pEnv, Database *p){ 148 assert( holdingGlobalMutex(pEnv) ); 149 if( p ){ 150 /* Free the mutexes */ 151 lsmMutexDel(pEnv, p->pClientMutex); 152 153 if( p->pFile ){ 154 lsmEnvClose(pEnv, p->pFile); 155 } 156 157 /* Free the array of shm pointers */ 158 lsmFree(pEnv, p->apShmChunk); 159 160 /* Free the memory allocated for the Database struct itself */ 161 lsmFree(pEnv, p); 162 } 163 } 164 165 typedef struct DbTruncateCtx DbTruncateCtx; 166 struct DbTruncateCtx { 167 int nBlock; 168 i64 iInUse; 169 }; 170 171 static int dbTruncateCb(void *pCtx, int iBlk, i64 iSnapshot){ 172 DbTruncateCtx *p = (DbTruncateCtx *)pCtx; 173 if( iBlk!=p->nBlock || (p->iInUse>=0 && iSnapshot>=p->iInUse) ) return 1; 174 p->nBlock--; 175 return 0; 176 } 177 178 static int dbTruncate(lsm_db *pDb, i64 iInUse){ 179 int rc = LSM_OK; 180 #if 0 181 int i; 182 DbTruncateCtx ctx; 183 184 assert( pDb->pWorker ); 185 ctx.nBlock = pDb->pWorker->nBlock; 186 ctx.iInUse = iInUse; 187 188 rc = lsmWalkFreelist(pDb, 1, dbTruncateCb, (void *)&ctx); 189 for(i=ctx.nBlock+1; rc==LSM_OK && i<=pDb->pWorker->nBlock; i++){ 190 rc = freelistAppend(pDb, i, -1); 191 } 192 193 if( rc==LSM_OK ){ 194 #ifdef LSM_LOG_FREELIST 195 if( ctx.nBlock!=pDb->pWorker->nBlock ){ 196 lsmLogMessage(pDb, 0, 197 "dbTruncate(): truncated db to %d blocks",ctx.nBlock 198 ); 199 } 200 #endif 201 pDb->pWorker->nBlock = ctx.nBlock; 202 } 203 #endif 204 return rc; 205 } 206 207 208 /* 209 ** This function is called during database shutdown (when the number of 210 ** connections drops from one to zero). It truncates the database file 211 ** to as small a size as possible without truncating away any blocks that 212 ** contain data. 213 */ 214 static int dbTruncateFile(lsm_db *pDb){ 215 int rc; 216 217 assert( pDb->pWorker==0 ); 218 assert( lsmShmAssertLock(pDb, LSM_LOCK_DMS1, LSM_LOCK_EXCL) ); 219 rc = lsmCheckpointLoadWorker(pDb); 220 221 if( rc==LSM_OK ){ 222 DbTruncateCtx ctx; 223 224 /* Walk the database free-block-list in reverse order. Set ctx.nBlock 225 ** to the block number of the last block in the database that actually 226 ** contains data. */ 227 ctx.nBlock = pDb->pWorker->nBlock; 228 ctx.iInUse = -1; 229 rc = lsmWalkFreelist(pDb, 1, dbTruncateCb, (void *)&ctx); 230 231 /* If the last block that contains data is not already the last block in 232 ** the database file, truncate the database file so that it is. */ 233 if( rc==LSM_OK && ctx.nBlock!=pDb->pWorker->nBlock ){ 234 rc = lsmFsTruncateDb( 235 pDb->pFS, (i64)ctx.nBlock*lsmFsBlockSize(pDb->pFS) 236 ); 237 } 238 } 239 240 lsmFreeSnapshot(pDb->pEnv, pDb->pWorker); 241 pDb->pWorker = 0; 242 return rc; 243 } 244 245 static void doDbDisconnect(lsm_db *pDb){ 246 int rc; 247 248 if( pDb->bReadonly ){ 249 lsmShmLock(pDb, LSM_LOCK_DMS3, LSM_LOCK_UNLOCK, 0); 250 }else{ 251 /* Block for an exclusive lock on DMS1. This lock serializes all calls 252 ** to doDbConnect() and doDbDisconnect() across all processes. */ 253 rc = lsmShmLock(pDb, LSM_LOCK_DMS1, LSM_LOCK_EXCL, 1); 254 if( rc==LSM_OK ){ 255 256 lsmShmLock(pDb, LSM_LOCK_DMS2, LSM_LOCK_UNLOCK, 0); 257 258 /* Try an exclusive lock on DMS2. If successful, this is the last 259 ** connection to the database. In this case flush the contents of the 260 ** in-memory tree to disk and write a checkpoint. */ 261 rc = lsmShmTestLock(pDb, LSM_LOCK_DMS2, 1, LSM_LOCK_EXCL); 262 if( rc==LSM_OK ){ 263 rc = lsmShmTestLock(pDb, LSM_LOCK_CHECKPOINTER, 1, LSM_LOCK_EXCL); 264 } 265 if( rc==LSM_OK ){ 266 int bReadonly = 0; /* True if there exist read-only conns. */ 267 268 /* Flush the in-memory tree, if required. If there is data to flush, 269 ** this will create a new client snapshot in Database.pClient. The 270 ** checkpoint (serialization) of this snapshot may be written to disk 271 ** by the following block. 272 ** 273 ** There is no need to take a WRITER lock here. That there are no 274 ** other locks on DMS2 guarantees that there are no other read-write 275 ** connections at this time (and the lock on DMS1 guarantees that 276 ** no new ones may appear). 277 */ 278 rc = lsmTreeLoadHeader(pDb, 0); 279 if( rc==LSM_OK && (lsmTreeHasOld(pDb) || lsmTreeSize(pDb)>0) ){ 280 rc = lsmFlushTreeToDisk(pDb); 281 } 282 283 /* Now check if there are any read-only connections. If there are, 284 ** then do not truncate the db file or unlink the shared-memory 285 ** region. */ 286 if( rc==LSM_OK ){ 287 rc = lsmShmTestLock(pDb, LSM_LOCK_DMS3, 1, LSM_LOCK_EXCL); 288 if( rc==LSM_BUSY ){ 289 bReadonly = 1; 290 rc = LSM_OK; 291 } 292 } 293 294 /* Write a checkpoint to disk. */ 295 if( rc==LSM_OK ){ 296 rc = lsmCheckpointWrite(pDb, (bReadonly==0), 0); 297 } 298 299 /* If the checkpoint was written successfully, delete the log file 300 ** and, if possible, truncate the database file. */ 301 if( rc==LSM_OK ){ 302 int bRotrans = 0; 303 Database *p = pDb->pDatabase; 304 305 /* The log file may only be deleted if there are no clients 306 ** read-only clients running rotrans transactions. */ 307 rc = lsmDetectRoTrans(pDb, &bRotrans); 308 if( rc==LSM_OK && bRotrans==0 ){ 309 lsmFsCloseAndDeleteLog(pDb->pFS); 310 } 311 312 /* The database may only be truncated if there exist no read-only 313 ** clients - either connected or running rotrans transactions. */ 314 if( bReadonly==0 && bRotrans==0 ){ 315 dbTruncateFile(pDb); 316 if( p->pFile && p->bMultiProc ){ 317 lsmEnvShmUnmap(pDb->pEnv, p->pFile, 1); 318 } 319 } 320 } 321 } 322 } 323 324 if( pDb->iRwclient>=0 ){ 325 lsmShmLock(pDb, LSM_LOCK_RWCLIENT(pDb->iRwclient), LSM_LOCK_UNLOCK, 0); 326 pDb->iRwclient = -1; 327 } 328 329 lsmShmLock(pDb, LSM_LOCK_DMS1, LSM_LOCK_UNLOCK, 0); 330 } 331 pDb->pShmhdr = 0; 332 } 333 334 static int doDbConnect(lsm_db *pDb){ 335 const int nUsMax = 100000; /* Max value for nUs */ 336 int nUs = 1000; /* us to wait between DMS1 attempts */ 337 int rc; 338 339 /* Obtain a pointer to the shared-memory header */ 340 assert( pDb->pShmhdr==0 ); 341 assert( pDb->bReadonly==0 ); 342 rc = lsmShmCacheChunks(pDb, 1); 343 if( rc!=LSM_OK ) return rc; 344 pDb->pShmhdr = (ShmHeader *)pDb->apShm[0]; 345 346 /* Block for an exclusive lock on DMS1. This lock serializes all calls 347 ** to doDbConnect() and doDbDisconnect() across all processes. */ 348 while( 1 ){ 349 rc = lsmShmLock(pDb, LSM_LOCK_DMS1, LSM_LOCK_EXCL, 1); 350 if( rc!=LSM_BUSY ) break; 351 lsmEnvSleep(pDb->pEnv, nUs); 352 nUs = nUs * 2; 353 if( nUs>nUsMax ) nUs = nUsMax; 354 } 355 if( rc!=LSM_OK ){ 356 pDb->pShmhdr = 0; 357 return rc; 358 } 359 360 /* Try an exclusive lock on DMS2/DMS3. If successful, this is the first 361 ** and only connection to the database. In this case initialize the 362 ** shared-memory and run log file recovery. */ 363 assert( LSM_LOCK_DMS3==1+LSM_LOCK_DMS2 ); 364 rc = lsmShmTestLock(pDb, LSM_LOCK_DMS2, 2, LSM_LOCK_EXCL); 365 if( rc==LSM_OK ){ 366 memset(pDb->pShmhdr, 0, sizeof(ShmHeader)); 367 rc = lsmCheckpointRecover(pDb); 368 if( rc==LSM_OK ){ 369 rc = lsmLogRecover(pDb); 370 } 371 if( rc==LSM_OK ){ 372 ShmHeader *pShm = pDb->pShmhdr; 373 pShm->aReader[0].iLsmId = lsmCheckpointId(pShm->aSnap1, 0); 374 pShm->aReader[0].iTreeId = pDb->treehdr.iUsedShmid; 375 } 376 }else if( rc==LSM_BUSY ){ 377 rc = LSM_OK; 378 } 379 380 /* Take a shared lock on DMS2. In multi-process mode this lock "cannot" 381 ** fail, as connections may only hold an exclusive lock on DMS2 if they 382 ** first hold an exclusive lock on DMS1. And this connection is currently 383 ** holding the exclusive lock on DSM1. 384 ** 385 ** However, if some other connection has the database open in single-process 386 ** mode, this operation will fail. In this case, return the error to the 387 ** caller - the attempt to connect to the db has failed. 388 */ 389 if( rc==LSM_OK ){ 390 rc = lsmShmLock(pDb, LSM_LOCK_DMS2, LSM_LOCK_SHARED, 0); 391 } 392 393 /* If anything went wrong, unlock DMS2. Otherwise, try to take an exclusive 394 ** lock on one of the LSM_LOCK_RWCLIENT() locks. Unlock DMS1 in any case. */ 395 if( rc!=LSM_OK ){ 396 pDb->pShmhdr = 0; 397 }else{ 398 int i; 399 for(i=0; i<LSM_LOCK_NRWCLIENT; i++){ 400 int rc2 = lsmShmLock(pDb, LSM_LOCK_RWCLIENT(i), LSM_LOCK_EXCL, 0); 401 if( rc2==LSM_OK ) pDb->iRwclient = i; 402 if( rc2!=LSM_BUSY ){ 403 rc = rc2; 404 break; 405 } 406 } 407 } 408 lsmShmLock(pDb, LSM_LOCK_DMS1, LSM_LOCK_UNLOCK, 0); 409 410 return rc; 411 } 412 413 static int dbOpenSharedFd(lsm_env *pEnv, Database *p, int bRoOk){ 414 int rc; 415 416 rc = lsmEnvOpen(pEnv, p->zName, 0, &p->pFile); 417 if( rc==LSM_IOERR && bRoOk ){ 418 rc = lsmEnvOpen(pEnv, p->zName, LSM_OPEN_READONLY, &p->pFile); 419 p->bReadonly = 1; 420 } 421 422 return rc; 423 } 424 425 /* 426 ** Return a reference to the shared Database handle for the database 427 ** identified by canonical path zName. If this is the first connection to 428 ** the named database, a new Database object is allocated. Otherwise, a 429 ** pointer to an existing object is returned. 430 ** 431 ** If successful, *ppDatabase is set to point to the shared Database 432 ** structure and LSM_OK returned. Otherwise, *ppDatabase is set to NULL 433 ** and and LSM error code returned. 434 ** 435 ** Each successful call to this function should be (eventually) matched 436 ** by a call to lsmDbDatabaseRelease(). 437 */ 438 int lsmDbDatabaseConnect( 439 lsm_db *pDb, /* Database handle */ 440 const char *zName /* Full-path to db file */ 441 ){ 442 lsm_env *pEnv = pDb->pEnv; 443 int rc; /* Return code */ 444 Database *p = 0; /* Pointer returned via *ppDatabase */ 445 int nName = lsmStrlen(zName); 446 447 assert( pDb->pDatabase==0 ); 448 rc = enterGlobalMutex(pEnv); 449 if( rc==LSM_OK ){ 450 451 /* Search the global list for an existing object. TODO: Need something 452 ** better than the memcmp() below to figure out if a given Database 453 ** object represents the requested file. */ 454 for(p=gShared.pDatabase; p; p=p->pDbNext){ 455 if( nName==p->nName && 0==memcmp(zName, p->zName, nName) ) break; 456 } 457 458 /* If no suitable Database object was found, allocate a new one. */ 459 if( p==0 ){ 460 p = (Database *)lsmMallocZeroRc(pEnv, sizeof(Database)+nName+1, &rc); 461 462 /* If the allocation was successful, fill in other fields and 463 ** allocate the client mutex. */ 464 if( rc==LSM_OK ){ 465 p->bMultiProc = pDb->bMultiProc; 466 p->zName = (char *)&p[1]; 467 p->nName = nName; 468 memcpy((void *)p->zName, zName, nName+1); 469 rc = lsmMutexNew(pEnv, &p->pClientMutex); 470 } 471 472 /* If nothing has gone wrong so far, open the shared fd. And if that 473 ** succeeds and this connection requested single-process mode, 474 ** attempt to take the exclusive lock on DMS2. */ 475 if( rc==LSM_OK ){ 476 int bReadonly = (pDb->bReadonly && pDb->bMultiProc); 477 rc = dbOpenSharedFd(pDb->pEnv, p, bReadonly); 478 } 479 480 if( rc==LSM_OK && p->bMultiProc==0 ){ 481 /* Hold an exclusive lock DMS1 while grabbing DMS2. This ensures 482 ** that any ongoing call to doDbDisconnect() (even one in another 483 ** process) is finished before proceeding. */ 484 assert( p->bReadonly==0 ); 485 rc = lsmEnvLock(pDb->pEnv, p->pFile, LSM_LOCK_DMS1, LSM_LOCK_EXCL); 486 if( rc==LSM_OK ){ 487 rc = lsmEnvLock(pDb->pEnv, p->pFile, LSM_LOCK_DMS2, LSM_LOCK_EXCL); 488 lsmEnvLock(pDb->pEnv, p->pFile, LSM_LOCK_DMS1, LSM_LOCK_UNLOCK); 489 } 490 } 491 492 if( rc==LSM_OK ){ 493 p->pDbNext = gShared.pDatabase; 494 gShared.pDatabase = p; 495 }else{ 496 freeDatabase(pEnv, p); 497 p = 0; 498 } 499 } 500 501 if( p ){ 502 p->nDbRef++; 503 } 504 leaveGlobalMutex(pEnv); 505 506 if( p ){ 507 lsmMutexEnter(pDb->pEnv, p->pClientMutex); 508 pDb->pNext = p->pConn; 509 p->pConn = pDb; 510 lsmMutexLeave(pDb->pEnv, p->pClientMutex); 511 } 512 } 513 514 pDb->pDatabase = p; 515 if( rc==LSM_OK ){ 516 assert( p ); 517 rc = lsmFsOpen(pDb, zName, p->bReadonly); 518 } 519 520 /* If the db handle is read-write, then connect to the system now. Run 521 ** recovery as necessary. Or, if this is a read-only database handle, 522 ** defer attempting to connect to the system until a read-transaction 523 ** is opened. */ 524 if( pDb->bReadonly==0 ){ 525 if( rc==LSM_OK ){ 526 rc = lsmFsConfigure(pDb); 527 } 528 if( rc==LSM_OK ){ 529 rc = doDbConnect(pDb); 530 } 531 } 532 533 return rc; 534 } 535 536 static void dbDeferClose(lsm_db *pDb){ 537 if( pDb->pFS ){ 538 LsmFile *pLsmFile; 539 Database *p = pDb->pDatabase; 540 pLsmFile = lsmFsDeferClose(pDb->pFS); 541 pLsmFile->pNext = p->pLsmFile; 542 p->pLsmFile = pLsmFile; 543 } 544 } 545 546 LsmFile *lsmDbRecycleFd(lsm_db *db){ 547 LsmFile *pRet; 548 Database *p = db->pDatabase; 549 lsmMutexEnter(db->pEnv, p->pClientMutex); 550 if( (pRet = p->pLsmFile)!=0 ){ 551 p->pLsmFile = pRet->pNext; 552 } 553 lsmMutexLeave(db->pEnv, p->pClientMutex); 554 return pRet; 555 } 556 557 /* 558 ** Release a reference to a Database object obtained from 559 ** lsmDbDatabaseConnect(). There should be exactly one call to this function 560 ** for each successful call to Find(). 561 */ 562 void lsmDbDatabaseRelease(lsm_db *pDb){ 563 Database *p = pDb->pDatabase; 564 if( p ){ 565 lsm_db **ppDb; 566 567 if( pDb->pShmhdr ){ 568 doDbDisconnect(pDb); 569 } 570 571 lsmMutexEnter(pDb->pEnv, p->pClientMutex); 572 for(ppDb=&p->pConn; *ppDb!=pDb; ppDb=&((*ppDb)->pNext)); 573 *ppDb = pDb->pNext; 574 dbDeferClose(pDb); 575 lsmMutexLeave(pDb->pEnv, p->pClientMutex); 576 577 enterGlobalMutex(pDb->pEnv); 578 p->nDbRef--; 579 if( p->nDbRef==0 ){ 580 LsmFile *pIter; 581 LsmFile *pNext; 582 Database **pp; 583 584 /* Remove the Database structure from the linked list. */ 585 for(pp=&gShared.pDatabase; *pp!=p; pp=&((*pp)->pDbNext)); 586 *pp = p->pDbNext; 587 588 /* If they were allocated from the heap, free the shared memory chunks */ 589 if( p->bMultiProc==0 ){ 590 int i; 591 for(i=0; i<p->nShmChunk; i++){ 592 lsmFree(pDb->pEnv, p->apShmChunk[i]); 593 } 594 } 595 596 /* Close any outstanding file descriptors */ 597 for(pIter=p->pLsmFile; pIter; pIter=pNext){ 598 pNext = pIter->pNext; 599 lsmEnvClose(pDb->pEnv, pIter->pFile); 600 lsmFree(pDb->pEnv, pIter); 601 } 602 freeDatabase(pDb->pEnv, p); 603 } 604 leaveGlobalMutex(pDb->pEnv); 605 } 606 } 607 608 Level *lsmDbSnapshotLevel(Snapshot *pSnapshot){ 609 return pSnapshot->pLevel; 610 } 611 612 void lsmDbSnapshotSetLevel(Snapshot *pSnap, Level *pLevel){ 613 pSnap->pLevel = pLevel; 614 } 615 616 /* TODO: Shuffle things around to get rid of this */ 617 static int firstSnapshotInUse(lsm_db *, i64 *); 618 619 /* 620 ** Context object used by the lsmWalkFreelist() utility. 621 */ 622 typedef struct WalkFreelistCtx WalkFreelistCtx; 623 struct WalkFreelistCtx { 624 lsm_db *pDb; 625 int bReverse; 626 Freelist *pFreelist; 627 int iFree; 628 int (*xUsr)(void *, int, i64); /* User callback function */ 629 void *pUsrctx; /* User callback context */ 630 int bDone; /* Set to true after xUsr() returns true */ 631 }; 632 633 /* 634 ** Callback used by lsmWalkFreelist(). 635 */ 636 static int walkFreelistCb(void *pCtx, int iBlk, i64 iSnapshot){ 637 WalkFreelistCtx *p = (WalkFreelistCtx *)pCtx; 638 const int iDir = (p->bReverse ? -1 : 1); 639 Freelist *pFree = p->pFreelist; 640 641 assert( p->bDone==0 ); 642 if( pFree ){ 643 while( (p->iFree < pFree->nEntry) && p->iFree>=0 ){ 644 FreelistEntry *pEntry = &pFree->aEntry[p->iFree]; 645 if( (p->bReverse==0 && pEntry->iBlk>iBlk) 646 || (p->bReverse!=0 && pEntry->iBlk<iBlk) 647 ){ 648 break; 649 }else{ 650 p->iFree += iDir; 651 if( pEntry->iId>=0 652 && p->xUsr(p->pUsrctx, pEntry->iBlk, pEntry->iId) 653 ){ 654 p->bDone = 1; 655 return 1; 656 } 657 if( pEntry->iBlk==iBlk ) return 0; 658 } 659 } 660 } 661 662 if( p->xUsr(p->pUsrctx, iBlk, iSnapshot) ){ 663 p->bDone = 1; 664 return 1; 665 } 666 return 0; 667 } 668 669 /* 670 ** The database handle passed as the first argument must be the worker 671 ** connection. This function iterates through the contents of the current 672 ** free block list, invoking the supplied callback once for each list 673 ** element. 674 ** 675 ** The difference between this function and lsmSortedWalkFreelist() is 676 ** that lsmSortedWalkFreelist() only considers those free-list elements 677 ** stored within the LSM. This function also merges in any in-memory 678 ** elements. 679 */ 680 int lsmWalkFreelist( 681 lsm_db *pDb, /* Database handle (must be worker) */ 682 int bReverse, /* True to iterate from largest to smallest */ 683 int (*x)(void *, int, i64), /* Callback function */ 684 void *pCtx /* First argument to pass to callback */ 685 ){ 686 const int iDir = (bReverse ? -1 : 1); 687 int rc; 688 int iCtx; 689 690 WalkFreelistCtx ctx[2]; 691 692 ctx[0].pDb = pDb; 693 ctx[0].bReverse = bReverse; 694 ctx[0].pFreelist = &pDb->pWorker->freelist; 695 if( ctx[0].pFreelist && bReverse ){ 696 ctx[0].iFree = ctx[0].pFreelist->nEntry-1; 697 }else{ 698 ctx[0].iFree = 0; 699 } 700 ctx[0].xUsr = walkFreelistCb; 701 ctx[0].pUsrctx = (void *)&ctx[1]; 702 ctx[0].bDone = 0; 703 704 ctx[1].pDb = pDb; 705 ctx[1].bReverse = bReverse; 706 ctx[1].pFreelist = pDb->pFreelist; 707 if( ctx[1].pFreelist && bReverse ){ 708 ctx[1].iFree = ctx[1].pFreelist->nEntry-1; 709 }else{ 710 ctx[1].iFree = 0; 711 } 712 ctx[1].xUsr = x; 713 ctx[1].pUsrctx = pCtx; 714 ctx[1].bDone = 0; 715 716 rc = lsmSortedWalkFreelist(pDb, bReverse, walkFreelistCb, (void *)&ctx[0]); 717 718 if( ctx[0].bDone==0 ){ 719 for(iCtx=0; iCtx<2; iCtx++){ 720 int i; 721 WalkFreelistCtx *p = &ctx[iCtx]; 722 for(i=p->iFree; 723 p->pFreelist && rc==LSM_OK && i<p->pFreelist->nEntry && i>=0; 724 i += iDir 725 ){ 726 FreelistEntry *pEntry = &p->pFreelist->aEntry[i]; 727 if( pEntry->iId>=0 && p->xUsr(p->pUsrctx, pEntry->iBlk, pEntry->iId) ){ 728 return LSM_OK; 729 } 730 } 731 } 732 } 733 734 return rc; 735 } 736 737 738 typedef struct FindFreeblockCtx FindFreeblockCtx; 739 struct FindFreeblockCtx { 740 i64 iInUse; 741 int iRet; 742 int bNotOne; 743 }; 744 745 static int findFreeblockCb(void *pCtx, int iBlk, i64 iSnapshot){ 746 FindFreeblockCtx *p = (FindFreeblockCtx *)pCtx; 747 if( iSnapshot<p->iInUse && (iBlk!=1 || p->bNotOne==0) ){ 748 p->iRet = iBlk; 749 return 1; 750 } 751 return 0; 752 } 753 754 static int findFreeblock(lsm_db *pDb, i64 iInUse, int bNotOne, int *piRet){ 755 int rc; /* Return code */ 756 FindFreeblockCtx ctx; /* Context object */ 757 758 ctx.iInUse = iInUse; 759 ctx.iRet = 0; 760 ctx.bNotOne = bNotOne; 761 rc = lsmWalkFreelist(pDb, 0, findFreeblockCb, (void *)&ctx); 762 *piRet = ctx.iRet; 763 764 return rc; 765 } 766 767 /* 768 ** Allocate a new database file block to write data to, either by extending 769 ** the database file or by recycling a free-list entry. The worker snapshot 770 ** must be held in order to call this function. 771 ** 772 ** If successful, *piBlk is set to the block number allocated and LSM_OK is 773 ** returned. Otherwise, *piBlk is zeroed and an lsm error code returned. 774 */ 775 int lsmBlockAllocate(lsm_db *pDb, int iBefore, int *piBlk){ 776 Snapshot *p = pDb->pWorker; 777 int iRet = 0; /* Block number of allocated block */ 778 int rc = LSM_OK; 779 i64 iInUse = 0; /* Snapshot id still in use */ 780 i64 iSynced = 0; /* Snapshot id synced to disk */ 781 782 assert( p ); 783 784 #ifdef LSM_LOG_FREELIST 785 { 786 static int nCall = 0; 787 char *zFree = 0; 788 nCall++; 789 rc = lsmInfoFreelist(pDb, &zFree); 790 if( rc!=LSM_OK ) return rc; 791 lsmLogMessage(pDb, 0, "lsmBlockAllocate(): %d freelist: %s", nCall, zFree); 792 lsmFree(pDb->pEnv, zFree); 793 } 794 #endif 795 796 /* Set iInUse to the smallest snapshot id that is either: 797 ** 798 ** * Currently in use by a database client, 799 ** * May be used by a database client in the future, or 800 ** * Is the most recently checkpointed snapshot (i.e. the one that will 801 ** be used following recovery if a failure occurs at this point). 802 */ 803 rc = lsmCheckpointSynced(pDb, &iSynced, 0, 0); 804 if( rc==LSM_OK && iSynced==0 ) iSynced = p->iId; 805 iInUse = iSynced; 806 if( rc==LSM_OK && pDb->iReader>=0 ){ 807 assert( pDb->pClient ); 808 iInUse = LSM_MIN(iInUse, pDb->pClient->iId); 809 } 810 if( rc==LSM_OK ) rc = firstSnapshotInUse(pDb, &iInUse); 811 812 #ifdef LSM_LOG_FREELIST 813 { 814 lsmLogMessage(pDb, 0, "lsmBlockAllocate(): " 815 "snapshot-in-use: %lld (iSynced=%lld) (client-id=%lld)", 816 iInUse, iSynced, (pDb->iReader>=0 ? pDb->pClient->iId : 0) 817 ); 818 } 819 #endif 820 821 822 /* Unless there exists a read-only transaction (which prevents us from 823 ** recycling any blocks regardless, query the free block list for a 824 ** suitable block to reuse. 825 ** 826 ** It might seem more natural to check for a read-only transaction at 827 ** the start of this function. However, it is better do wait until after 828 ** the call to lsmCheckpointSynced() to do so. 829 */ 830 if( rc==LSM_OK ){ 831 int bRotrans; 832 rc = lsmDetectRoTrans(pDb, &bRotrans); 833 834 if( rc==LSM_OK && bRotrans==0 ){ 835 rc = findFreeblock(pDb, iInUse, (iBefore>0), &iRet); 836 } 837 } 838 839 if( iBefore>0 && (iRet<=0 || iRet>=iBefore) ){ 840 iRet = 0; 841 842 }else if( rc==LSM_OK ){ 843 /* If a block was found in the free block list, use it and remove it from 844 ** the list. Otherwise, if no suitable block was found, allocate one from 845 ** the end of the file. */ 846 if( iRet>0 ){ 847 #ifdef LSM_LOG_FREELIST 848 lsmLogMessage(pDb, 0, 849 "reusing block %d (snapshot-in-use=%lld)", iRet, iInUse); 850 #endif 851 rc = freelistAppend(pDb, iRet, -1); 852 if( rc==LSM_OK ){ 853 rc = dbTruncate(pDb, iInUse); 854 } 855 }else{ 856 iRet = ++(p->nBlock); 857 #ifdef LSM_LOG_FREELIST 858 lsmLogMessage(pDb, 0, "extending file to %d blocks", iRet); 859 #endif 860 } 861 } 862 863 assert( iBefore>0 || iRet>0 || rc!=LSM_OK ); 864 *piBlk = iRet; 865 return rc; 866 } 867 868 /* 869 ** Free a database block. The worker snapshot must be held in order to call 870 ** this function. 871 ** 872 ** If successful, LSM_OK is returned. Otherwise, an lsm error code (e.g. 873 ** LSM_NOMEM). 874 */ 875 int lsmBlockFree(lsm_db *pDb, int iBlk){ 876 Snapshot *p = pDb->pWorker; 877 assert( lsmShmAssertWorker(pDb) ); 878 879 #ifdef LSM_LOG_FREELIST 880 lsmLogMessage(pDb, LSM_OK, "lsmBlockFree(): Free block %d", iBlk); 881 #endif 882 883 return freelistAppend(pDb, iBlk, p->iId); 884 } 885 886 /* 887 ** Refree a database block. The worker snapshot must be held in order to call 888 ** this function. 889 ** 890 ** Refreeing is required when a block is allocated using lsmBlockAllocate() 891 ** but then not used. This function is used to push the block back onto 892 ** the freelist. Refreeing a block is different from freeing is, as a refreed 893 ** block may be reused immediately. Whereas a freed block can not be reused 894 ** until (at least) after the next checkpoint. 895 */ 896 int lsmBlockRefree(lsm_db *pDb, int iBlk){ 897 int rc = LSM_OK; /* Return code */ 898 899 #ifdef LSM_LOG_FREELIST 900 lsmLogMessage(pDb, LSM_OK, "lsmBlockRefree(): Refree block %d", iBlk); 901 #endif 902 903 rc = freelistAppend(pDb, iBlk, 0); 904 return rc; 905 } 906 907 /* 908 ** If required, copy a database checkpoint from shared memory into the 909 ** database itself. 910 ** 911 ** The WORKER lock must not be held when this is called. This is because 912 ** this function may indirectly call fsync(). And the WORKER lock should 913 ** not be held that long (in case it is required by a client flushing an 914 ** in-memory tree to disk). 915 */ 916 int lsmCheckpointWrite(lsm_db *pDb, int bTruncate, u32 *pnWrite){ 917 int rc; /* Return Code */ 918 u32 nWrite = 0; 919 920 assert( pDb->pWorker==0 ); 921 assert( 1 || pDb->pClient==0 ); 922 assert( lsmShmAssertLock(pDb, LSM_LOCK_WORKER, LSM_LOCK_UNLOCK) ); 923 924 rc = lsmShmLock(pDb, LSM_LOCK_CHECKPOINTER, LSM_LOCK_EXCL, 0); 925 if( rc!=LSM_OK ) return rc; 926 927 rc = lsmCheckpointLoad(pDb, 0); 928 if( rc==LSM_OK ){ 929 int nBlock = lsmCheckpointNBlock(pDb->aSnapshot); 930 ShmHeader *pShm = pDb->pShmhdr; 931 int bDone = 0; /* True if checkpoint is already stored */ 932 933 /* Check if this checkpoint has already been written to the database 934 ** file. If so, set variable bDone to true. */ 935 if( pShm->iMetaPage ){ 936 MetaPage *pPg; /* Meta page */ 937 u8 *aData; /* Meta-page data buffer */ 938 int nData; /* Size of aData[] in bytes */ 939 i64 iCkpt; /* Id of checkpoint just loaded */ 940 i64 iDisk; /* Id of checkpoint already stored in db */ 941 iCkpt = lsmCheckpointId(pDb->aSnapshot, 0); 942 rc = lsmFsMetaPageGet(pDb->pFS, 0, pShm->iMetaPage, &pPg); 943 if( rc==LSM_OK ){ 944 aData = lsmFsMetaPageData(pPg, &nData); 945 iDisk = lsmCheckpointId((u32 *)aData, 1); 946 nWrite = lsmCheckpointNWrite((u32 *)aData, 1); 947 lsmFsMetaPageRelease(pPg); 948 } 949 bDone = (iDisk>=iCkpt); 950 } 951 952 if( rc==LSM_OK && bDone==0 ){ 953 int iMeta = (pShm->iMetaPage % 2) + 1; 954 if( pDb->eSafety!=LSM_SAFETY_OFF ){ 955 rc = lsmFsSyncDb(pDb->pFS, nBlock); 956 } 957 if( rc==LSM_OK ) rc = lsmCheckpointStore(pDb, iMeta); 958 if( rc==LSM_OK && pDb->eSafety!=LSM_SAFETY_OFF){ 959 rc = lsmFsSyncDb(pDb->pFS, 0); 960 } 961 if( rc==LSM_OK ){ 962 pShm->iMetaPage = iMeta; 963 nWrite = lsmCheckpointNWrite(pDb->aSnapshot, 0) - nWrite; 964 } 965 #ifdef LSM_LOG_WORK 966 lsmLogMessage(pDb, 0, "finish checkpoint %d", 967 (int)lsmCheckpointId(pDb->aSnapshot, 0) 968 ); 969 #endif 970 } 971 972 if( rc==LSM_OK && bTruncate && nBlock>0 ){ 973 rc = lsmFsTruncateDb(pDb->pFS, (i64)nBlock*lsmFsBlockSize(pDb->pFS)); 974 } 975 } 976 977 lsmShmLock(pDb, LSM_LOCK_CHECKPOINTER, LSM_LOCK_UNLOCK, 0); 978 if( pnWrite && rc==LSM_OK ) *pnWrite = nWrite; 979 return rc; 980 } 981 982 int lsmBeginWork(lsm_db *pDb){ 983 int rc; 984 985 /* Attempt to take the WORKER lock */ 986 rc = lsmShmLock(pDb, LSM_LOCK_WORKER, LSM_LOCK_EXCL, 0); 987 988 /* Deserialize the current worker snapshot */ 989 if( rc==LSM_OK ){ 990 rc = lsmCheckpointLoadWorker(pDb); 991 } 992 return rc; 993 } 994 995 void lsmFreeSnapshot(lsm_env *pEnv, Snapshot *p){ 996 if( p ){ 997 lsmSortedFreeLevel(pEnv, p->pLevel); 998 lsmFree(pEnv, p->freelist.aEntry); 999 lsmFree(pEnv, p->redirect.a); 1000 lsmFree(pEnv, p); 1001 } 1002 } 1003 1004 /* 1005 ** Attempt to populate one of the read-lock slots to contain lock values 1006 ** iLsm/iShm. Or, if such a slot exists already, this function is a no-op. 1007 ** 1008 ** It is not an error if no slot can be populated because the write-lock 1009 ** cannot be obtained. If any other error occurs, return an LSM error code. 1010 ** Otherwise, LSM_OK. 1011 ** 1012 ** This function is called at various points to try to ensure that there 1013 ** always exists at least one read-lock slot that can be used by a read-only 1014 ** client. And so that, in the usual case, there is an "exact match" available 1015 ** whenever a read transaction is opened by any client. At present this 1016 ** function is called when: 1017 ** 1018 ** * A write transaction that called lsmTreeDiscardOld() is committed, and 1019 ** * Whenever the working snapshot is updated (i.e. lsmFinishWork()). 1020 */ 1021 static int dbSetReadLock(lsm_db *db, i64 iLsm, u32 iShm){ 1022 int rc = LSM_OK; 1023 ShmHeader *pShm = db->pShmhdr; 1024 int i; 1025 1026 /* Check if there is already a slot containing the required values. */ 1027 for(i=0; i<LSM_LOCK_NREADER; i++){ 1028 ShmReader *p = &pShm->aReader[i]; 1029 if( p->iLsmId==iLsm && p->iTreeId==iShm ) return LSM_OK; 1030 } 1031 1032 /* Iterate through all read-lock slots, attempting to take a write-lock 1033 ** on each of them. If a write-lock succeeds, populate the locked slot 1034 ** with the required values and break out of the loop. */ 1035 for(i=0; rc==LSM_OK && i<LSM_LOCK_NREADER; i++){ 1036 rc = lsmShmLock(db, LSM_LOCK_READER(i), LSM_LOCK_EXCL, 0); 1037 if( rc==LSM_BUSY ){ 1038 rc = LSM_OK; 1039 }else{ 1040 ShmReader *p = &pShm->aReader[i]; 1041 p->iLsmId = iLsm; 1042 p->iTreeId = iShm; 1043 lsmShmLock(db, LSM_LOCK_READER(i), LSM_LOCK_UNLOCK, 0); 1044 break; 1045 } 1046 } 1047 1048 return rc; 1049 } 1050 1051 /* 1052 ** Release the read-lock currently held by connection db. 1053 */ 1054 int dbReleaseReadlock(lsm_db *db){ 1055 int rc = LSM_OK; 1056 if( db->iReader>=0 ){ 1057 rc = lsmShmLock(db, LSM_LOCK_READER(db->iReader), LSM_LOCK_UNLOCK, 0); 1058 db->iReader = -1; 1059 } 1060 db->bRoTrans = 0; 1061 return rc; 1062 } 1063 1064 1065 /* 1066 ** Argument bFlush is true if the contents of the in-memory tree has just 1067 ** been flushed to disk. The significance of this is that once the snapshot 1068 ** created to hold the updated state of the database is synced to disk, log 1069 ** file space can be recycled. 1070 */ 1071 void lsmFinishWork(lsm_db *pDb, int bFlush, int *pRc){ 1072 int rc = *pRc; 1073 assert( rc!=0 || pDb->pWorker ); 1074 if( pDb->pWorker ){ 1075 /* If no error has occurred, serialize the worker snapshot and write 1076 ** it to shared memory. */ 1077 if( rc==LSM_OK ){ 1078 rc = lsmSaveWorker(pDb, bFlush); 1079 } 1080 1081 /* Assuming no error has occurred, update a read lock slot with the 1082 ** new snapshot id (see comments above function dbSetReadLock()). */ 1083 if( rc==LSM_OK ){ 1084 if( pDb->iReader<0 ){ 1085 rc = lsmTreeLoadHeader(pDb, 0); 1086 } 1087 if( rc==LSM_OK ){ 1088 rc = dbSetReadLock(pDb, pDb->pWorker->iId, pDb->treehdr.iUsedShmid); 1089 } 1090 } 1091 1092 /* Free the snapshot object. */ 1093 lsmFreeSnapshot(pDb->pEnv, pDb->pWorker); 1094 pDb->pWorker = 0; 1095 } 1096 1097 lsmShmLock(pDb, LSM_LOCK_WORKER, LSM_LOCK_UNLOCK, 0); 1098 *pRc = rc; 1099 } 1100 1101 /* 1102 ** Called when recovery is finished. 1103 */ 1104 int lsmFinishRecovery(lsm_db *pDb){ 1105 lsmTreeEndTransaction(pDb, 1); 1106 return LSM_OK; 1107 } 1108 1109 /* 1110 ** Check if the currently configured compression functions 1111 ** (LSM_CONFIG_SET_COMPRESSION) are compatible with a database that has its 1112 ** compression id set to iReq. Compression routines are compatible if iReq 1113 ** is zero (indicating the database is empty), or if it is equal to the 1114 ** compression id of the configured compression routines. 1115 ** 1116 ** If the check shows that the current compression are incompatible and there 1117 ** is a compression factory registered, give it a chance to install new 1118 ** compression routines. 1119 ** 1120 ** If, after any registered factory is invoked, the compression functions 1121 ** are still incompatible, return LSM_MISMATCH. Otherwise, LSM_OK. 1122 */ 1123 int lsmCheckCompressionId(lsm_db *pDb, u32 iReq){ 1124 if( iReq!=LSM_COMPRESSION_EMPTY && pDb->compress.iId!=iReq ){ 1125 if( pDb->factory.xFactory ){ 1126 pDb->bInFactory = 1; 1127 pDb->factory.xFactory(pDb->factory.pCtx, pDb, iReq); 1128 pDb->bInFactory = 0; 1129 } 1130 if( pDb->compress.iId!=iReq ){ 1131 /* Incompatible */ 1132 return LSM_MISMATCH; 1133 } 1134 } 1135 /* Compatible */ 1136 return LSM_OK; 1137 } 1138 1139 /* 1140 ** Begin a read transaction. This function is a no-op if the connection 1141 ** passed as the only argument already has an open read transaction. 1142 */ 1143 int lsmBeginReadTrans(lsm_db *pDb){ 1144 const int MAX_READLOCK_ATTEMPTS = 10; 1145 const int nMaxAttempt = (pDb->bRoTrans ? 1 : MAX_READLOCK_ATTEMPTS); 1146 1147 int rc = LSM_OK; /* Return code */ 1148 int iAttempt = 0; 1149 1150 assert( pDb->pWorker==0 ); 1151 1152 while( rc==LSM_OK && pDb->iReader<0 && (iAttempt++)<nMaxAttempt ){ 1153 int iTreehdr = 0; 1154 int iSnap = 0; 1155 assert( pDb->pCsr==0 && pDb->nTransOpen==0 ); 1156 1157 /* Load the in-memory tree header. */ 1158 rc = lsmTreeLoadHeader(pDb, &iTreehdr); 1159 1160 /* Load the database snapshot */ 1161 if( rc==LSM_OK ){ 1162 if( lsmCheckpointClientCacheOk(pDb)==0 ){ 1163 lsmFreeSnapshot(pDb->pEnv, pDb->pClient); 1164 pDb->pClient = 0; 1165 lsmMCursorFreeCache(pDb); 1166 lsmFsPurgeCache(pDb->pFS); 1167 rc = lsmCheckpointLoad(pDb, &iSnap); 1168 }else{ 1169 iSnap = 1; 1170 } 1171 } 1172 1173 /* Take a read-lock on the tree and snapshot just loaded. Then check 1174 ** that the shared-memory still contains the same values. If so, proceed. 1175 ** Otherwise, relinquish the read-lock and retry the whole procedure 1176 ** (starting with loading the in-memory tree header). */ 1177 if( rc==LSM_OK ){ 1178 u32 iShmMax = pDb->treehdr.iUsedShmid; 1179 u32 iShmMin = pDb->treehdr.iNextShmid+1-LSM_MAX_SHMCHUNKS; 1180 rc = lsmReadlock( 1181 pDb, lsmCheckpointId(pDb->aSnapshot, 0), iShmMin, iShmMax 1182 ); 1183 if( rc==LSM_OK ){ 1184 if( lsmTreeLoadHeaderOk(pDb, iTreehdr) 1185 && lsmCheckpointLoadOk(pDb, iSnap) 1186 ){ 1187 /* Read lock has been successfully obtained. Deserialize the 1188 ** checkpoint just loaded. TODO: This will be removed after 1189 ** lsm_sorted.c is changed to work directly from the serialized 1190 ** version of the snapshot. */ 1191 if( pDb->pClient==0 ){ 1192 rc = lsmCheckpointDeserialize(pDb, 0, pDb->aSnapshot,&pDb->pClient); 1193 } 1194 assert( (rc==LSM_OK)==(pDb->pClient!=0) ); 1195 assert( pDb->iReader>=0 ); 1196 1197 /* Check that the client has the right compression hooks loaded. 1198 ** If not, set rc to LSM_MISMATCH. */ 1199 if( rc==LSM_OK ){ 1200 rc = lsmCheckCompressionId(pDb, pDb->pClient->iCmpId); 1201 } 1202 }else{ 1203 rc = dbReleaseReadlock(pDb); 1204 } 1205 } 1206 1207 if( rc==LSM_BUSY ){ 1208 rc = LSM_OK; 1209 } 1210 } 1211 #if 0 1212 if( rc==LSM_OK && pDb->pClient ){ 1213 fprintf(stderr, 1214 "reading %p: snapshot:%d used-shmid:%d trans-id:%d iOldShmid=%d\n", 1215 (void *)pDb, 1216 (int)pDb->pClient->iId, (int)pDb->treehdr.iUsedShmid, 1217 (int)pDb->treehdr.root.iTransId, 1218 (int)pDb->treehdr.iOldShmid 1219 ); 1220 } 1221 #endif 1222 } 1223 1224 if( rc==LSM_OK ){ 1225 rc = lsmShmCacheChunks(pDb, pDb->treehdr.nChunk); 1226 } 1227 if( rc!=LSM_OK ){ 1228 dbReleaseReadlock(pDb); 1229 } 1230 if( pDb->pClient==0 && rc==LSM_OK ) rc = LSM_BUSY; 1231 return rc; 1232 } 1233 1234 /* 1235 ** This function is used by a read-write connection to determine if there 1236 ** are currently one or more read-only transactions open on the database 1237 ** (in this context a read-only transaction is one opened by a read-only 1238 ** connection on a non-live database). 1239 ** 1240 ** If no error occurs, LSM_OK is returned and *pbExists is set to true if 1241 ** some other connection has a read-only transaction open, or false 1242 ** otherwise. If an error occurs an LSM error code is returned and the final 1243 ** value of *pbExist is undefined. 1244 */ 1245 int lsmDetectRoTrans(lsm_db *db, int *pbExist){ 1246 int rc; 1247 1248 /* Only a read-write connection may use this function. */ 1249 assert( db->bReadonly==0 ); 1250 1251 rc = lsmShmTestLock(db, LSM_LOCK_ROTRANS, 1, LSM_LOCK_EXCL); 1252 if( rc==LSM_BUSY ){ 1253 *pbExist = 1; 1254 rc = LSM_OK; 1255 }else{ 1256 *pbExist = 0; 1257 } 1258 1259 return rc; 1260 } 1261 1262 /* 1263 ** db is a read-only database handle in the disconnected state. This function 1264 ** attempts to open a read-transaction on the database. This may involve 1265 ** connecting to the database system (opening shared memory etc.). 1266 */ 1267 int lsmBeginRoTrans(lsm_db *db){ 1268 int rc = LSM_OK; 1269 1270 assert( db->bReadonly && db->pShmhdr==0 ); 1271 assert( db->iReader<0 ); 1272 1273 if( db->bRoTrans==0 ){ 1274 1275 /* Attempt a shared-lock on DMS1. */ 1276 rc = lsmShmLock(db, LSM_LOCK_DMS1, LSM_LOCK_SHARED, 0); 1277 if( rc!=LSM_OK ) return rc; 1278 1279 rc = lsmShmTestLock( 1280 db, LSM_LOCK_RWCLIENT(0), LSM_LOCK_NREADER, LSM_LOCK_SHARED 1281 ); 1282 if( rc==LSM_OK ){ 1283 /* System is not live. Take a SHARED lock on the ROTRANS byte and 1284 ** release DMS1. Locking ROTRANS tells all read-write clients that they 1285 ** may not recycle any disk space from within the database or log files, 1286 ** as a read-only client may be using it. */ 1287 rc = lsmShmLock(db, LSM_LOCK_ROTRANS, LSM_LOCK_SHARED, 0); 1288 lsmShmLock(db, LSM_LOCK_DMS1, LSM_LOCK_UNLOCK, 0); 1289 1290 if( rc==LSM_OK ){ 1291 db->bRoTrans = 1; 1292 rc = lsmShmCacheChunks(db, 1); 1293 if( rc==LSM_OK ){ 1294 db->pShmhdr = (ShmHeader *)db->apShm[0]; 1295 memset(db->pShmhdr, 0, sizeof(ShmHeader)); 1296 rc = lsmCheckpointRecover(db); 1297 if( rc==LSM_OK ){ 1298 rc = lsmLogRecover(db); 1299 } 1300 } 1301 } 1302 }else if( rc==LSM_BUSY ){ 1303 /* System is live! */ 1304 rc = lsmShmLock(db, LSM_LOCK_DMS3, LSM_LOCK_SHARED, 0); 1305 lsmShmLock(db, LSM_LOCK_DMS1, LSM_LOCK_UNLOCK, 0); 1306 if( rc==LSM_OK ){ 1307 rc = lsmShmCacheChunks(db, 1); 1308 if( rc==LSM_OK ){ 1309 db->pShmhdr = (ShmHeader *)db->apShm[0]; 1310 } 1311 } 1312 } 1313 1314 if( rc==LSM_OK ){ 1315 rc = lsmBeginReadTrans(db); 1316 } 1317 } 1318 1319 return rc; 1320 } 1321 1322 /* 1323 ** Close the currently open read transaction. 1324 */ 1325 void lsmFinishReadTrans(lsm_db *pDb){ 1326 1327 /* Worker connections should not be closing read transactions. And 1328 ** read transactions should only be closed after all cursors and write 1329 ** transactions have been closed. Finally pClient should be non-NULL 1330 ** only iff pDb->iReader>=0. */ 1331 assert( pDb->pWorker==0 ); 1332 assert( pDb->pCsr==0 && pDb->nTransOpen==0 ); 1333 1334 if( pDb->bRoTrans ){ 1335 int i; 1336 for(i=0; i<pDb->nShm; i++){ 1337 lsmFree(pDb->pEnv, pDb->apShm[i]); 1338 } 1339 lsmFree(pDb->pEnv, pDb->apShm); 1340 pDb->apShm = 0; 1341 pDb->nShm = 0; 1342 pDb->pShmhdr = 0; 1343 1344 lsmShmLock(pDb, LSM_LOCK_ROTRANS, LSM_LOCK_UNLOCK, 0); 1345 } 1346 dbReleaseReadlock(pDb); 1347 } 1348 1349 /* 1350 ** Open a write transaction. 1351 */ 1352 int lsmBeginWriteTrans(lsm_db *pDb){ 1353 int rc = LSM_OK; /* Return code */ 1354 ShmHeader *pShm = pDb->pShmhdr; /* Shared memory header */ 1355 1356 assert( pDb->nTransOpen==0 ); 1357 assert( pDb->bDiscardOld==0 ); 1358 assert( pDb->bReadonly==0 ); 1359 1360 /* If there is no read-transaction open, open one now. */ 1361 if( pDb->iReader<0 ){ 1362 rc = lsmBeginReadTrans(pDb); 1363 } 1364 1365 /* Attempt to take the WRITER lock */ 1366 if( rc==LSM_OK ){ 1367 rc = lsmShmLock(pDb, LSM_LOCK_WRITER, LSM_LOCK_EXCL, 0); 1368 } 1369 1370 /* If the previous writer failed mid-transaction, run emergency rollback. */ 1371 if( rc==LSM_OK && pShm->bWriter ){ 1372 rc = lsmTreeRepair(pDb); 1373 if( rc==LSM_OK ) pShm->bWriter = 0; 1374 } 1375 1376 /* Check that this connection is currently reading from the most recent 1377 ** version of the database. If not, return LSM_BUSY. */ 1378 if( rc==LSM_OK && memcmp(&pShm->hdr1, &pDb->treehdr, sizeof(TreeHeader)) ){ 1379 rc = LSM_BUSY; 1380 } 1381 1382 if( rc==LSM_OK ){ 1383 rc = lsmLogBegin(pDb); 1384 } 1385 1386 /* If everything was successful, set the "transaction-in-progress" flag 1387 ** and return LSM_OK. Otherwise, if some error occurred, relinquish the 1388 ** WRITER lock and return an error code. */ 1389 if( rc==LSM_OK ){ 1390 TreeHeader *p = &pDb->treehdr; 1391 pShm->bWriter = 1; 1392 p->root.iTransId++; 1393 if( lsmTreeHasOld(pDb) && p->iOldLog==pDb->pClient->iLogOff ){ 1394 lsmTreeDiscardOld(pDb); 1395 pDb->bDiscardOld = 1; 1396 } 1397 }else{ 1398 lsmShmLock(pDb, LSM_LOCK_WRITER, LSM_LOCK_UNLOCK, 0); 1399 if( pDb->pCsr==0 ) lsmFinishReadTrans(pDb); 1400 } 1401 return rc; 1402 } 1403 1404 /* 1405 ** End the current write transaction. The connection is left with an open 1406 ** read transaction. It is an error to call this if there is no open write 1407 ** transaction. 1408 ** 1409 ** If the transaction was committed, then a commit record has already been 1410 ** written into the log file when this function is called. Or, if the 1411 ** transaction was rolled back, both the log file and in-memory tree 1412 ** structure have already been restored. In either case, this function 1413 ** merely releases locks and other resources held by the write-transaction. 1414 ** 1415 ** LSM_OK is returned if successful, or an LSM error code otherwise. 1416 */ 1417 int lsmFinishWriteTrans(lsm_db *pDb, int bCommit){ 1418 int rc = LSM_OK; 1419 int bFlush = 0; 1420 1421 lsmLogEnd(pDb, bCommit); 1422 if( rc==LSM_OK && bCommit && lsmTreeSize(pDb)>pDb->nTreeLimit ){ 1423 bFlush = 1; 1424 lsmTreeMakeOld(pDb); 1425 } 1426 lsmTreeEndTransaction(pDb, bCommit); 1427 1428 if( rc==LSM_OK ){ 1429 if( bFlush && pDb->bAutowork ){ 1430 rc = lsmSortedAutoWork(pDb, 1); 1431 }else if( bCommit && pDb->bDiscardOld ){ 1432 rc = dbSetReadLock(pDb, pDb->pClient->iId, pDb->treehdr.iUsedShmid); 1433 } 1434 } 1435 pDb->bDiscardOld = 0; 1436 lsmShmLock(pDb, LSM_LOCK_WRITER, LSM_LOCK_UNLOCK, 0); 1437 1438 if( bFlush && pDb->bAutowork==0 && pDb->xWork ){ 1439 pDb->xWork(pDb, pDb->pWorkCtx); 1440 } 1441 return rc; 1442 } 1443 1444 1445 /* 1446 ** Return non-zero if the caller is holding the client mutex. 1447 */ 1448 #ifdef LSM_DEBUG 1449 int lsmHoldingClientMutex(lsm_db *pDb){ 1450 return lsmMutexHeld(pDb->pEnv, pDb->pDatabase->pClientMutex); 1451 } 1452 #endif 1453 1454 static int slotIsUsable(ShmReader *p, i64 iLsm, u32 iShmMin, u32 iShmMax){ 1455 return( 1456 p->iLsmId && p->iLsmId<=iLsm 1457 && shm_sequence_ge(iShmMax, p->iTreeId) 1458 && shm_sequence_ge(p->iTreeId, iShmMin) 1459 ); 1460 } 1461 1462 /* 1463 ** Obtain a read-lock on database version identified by the combination 1464 ** of snapshot iLsm and tree iTree. Return LSM_OK if successful, or 1465 ** an LSM error code otherwise. 1466 */ 1467 int lsmReadlock(lsm_db *db, i64 iLsm, u32 iShmMin, u32 iShmMax){ 1468 int rc = LSM_OK; 1469 ShmHeader *pShm = db->pShmhdr; 1470 int i; 1471 1472 assert( db->iReader<0 ); 1473 assert( shm_sequence_ge(iShmMax, iShmMin) ); 1474 1475 /* This is a no-op if the read-only transaction flag is set. */ 1476 if( db->bRoTrans ){ 1477 db->iReader = 0; 1478 return LSM_OK; 1479 } 1480 1481 /* Search for an exact match. */ 1482 for(i=0; db->iReader<0 && rc==LSM_OK && i<LSM_LOCK_NREADER; i++){ 1483 ShmReader *p = &pShm->aReader[i]; 1484 if( p->iLsmId==iLsm && p->iTreeId==iShmMax ){ 1485 rc = lsmShmLock(db, LSM_LOCK_READER(i), LSM_LOCK_SHARED, 0); 1486 if( rc==LSM_OK && p->iLsmId==iLsm && p->iTreeId==iShmMax ){ 1487 db->iReader = i; 1488 }else if( rc==LSM_BUSY ){ 1489 rc = LSM_OK; 1490 } 1491 } 1492 } 1493 1494 /* Try to obtain a write-lock on each slot, in order. If successful, set 1495 ** the slot values to iLsm/iTree. */ 1496 for(i=0; db->iReader<0 && rc==LSM_OK && i<LSM_LOCK_NREADER; i++){ 1497 rc = lsmShmLock(db, LSM_LOCK_READER(i), LSM_LOCK_EXCL, 0); 1498 if( rc==LSM_BUSY ){ 1499 rc = LSM_OK; 1500 }else{ 1501 ShmReader *p = &pShm->aReader[i]; 1502 p->iLsmId = iLsm; 1503 p->iTreeId = iShmMax; 1504 rc = lsmShmLock(db, LSM_LOCK_READER(i), LSM_LOCK_SHARED, 0); 1505 assert( rc!=LSM_BUSY ); 1506 if( rc==LSM_OK ) db->iReader = i; 1507 } 1508 } 1509 1510 /* Search for any usable slot */ 1511 for(i=0; db->iReader<0 && rc==LSM_OK && i<LSM_LOCK_NREADER; i++){ 1512 ShmReader *p = &pShm->aReader[i]; 1513 if( slotIsUsable(p, iLsm, iShmMin, iShmMax) ){ 1514 rc = lsmShmLock(db, LSM_LOCK_READER(i), LSM_LOCK_SHARED, 0); 1515 if( rc==LSM_OK && slotIsUsable(p, iLsm, iShmMin, iShmMax) ){ 1516 db->iReader = i; 1517 }else if( rc==LSM_BUSY ){ 1518 rc = LSM_OK; 1519 } 1520 } 1521 } 1522 1523 if( rc==LSM_OK && db->iReader<0 ){ 1524 rc = LSM_BUSY; 1525 } 1526 return rc; 1527 } 1528 1529 /* 1530 ** This is used to check if there exists a read-lock locking a particular 1531 ** version of either the in-memory tree or database file. 1532 ** 1533 ** If iLsmId is non-zero, then it is a snapshot id. If there exists a 1534 ** read-lock using this snapshot or newer, set *pbInUse to true. Or, 1535 ** if there is no such read-lock, set it to false. 1536 ** 1537 ** Or, if iLsmId is zero, then iShmid is a shared-memory sequence id. 1538 ** Search for a read-lock using this sequence id or newer. etc. 1539 */ 1540 static int isInUse(lsm_db *db, i64 iLsmId, u32 iShmid, int *pbInUse){ 1541 ShmHeader *pShm = db->pShmhdr; 1542 int i; 1543 int rc = LSM_OK; 1544 1545 for(i=0; rc==LSM_OK && i<LSM_LOCK_NREADER; i++){ 1546 ShmReader *p = &pShm->aReader[i]; 1547 if( p->iLsmId ){ 1548 if( (iLsmId!=0 && p->iLsmId!=0 && iLsmId>=p->iLsmId) 1549 || (iLsmId==0 && shm_sequence_ge(p->iTreeId, iShmid)) 1550 ){ 1551 rc = lsmShmLock(db, LSM_LOCK_READER(i), LSM_LOCK_EXCL, 0); 1552 if( rc==LSM_OK ){ 1553 p->iLsmId = 0; 1554 lsmShmLock(db, LSM_LOCK_READER(i), LSM_LOCK_UNLOCK, 0); 1555 } 1556 } 1557 } 1558 } 1559 1560 if( rc==LSM_BUSY ){ 1561 *pbInUse = 1; 1562 return LSM_OK; 1563 } 1564 *pbInUse = 0; 1565 return rc; 1566 } 1567 1568 /* 1569 ** This function is called by worker connections to determine the smallest 1570 ** snapshot id that is currently in use by a database client. The worker 1571 ** connection uses this result to determine whether or not it is safe to 1572 ** recycle a database block. 1573 */ 1574 static int firstSnapshotInUse( 1575 lsm_db *db, /* Database handle */ 1576 i64 *piInUse /* IN/OUT: Smallest snapshot id in use */ 1577 ){ 1578 ShmHeader *pShm = db->pShmhdr; 1579 i64 iInUse = *piInUse; 1580 int i; 1581 1582 assert( iInUse>0 ); 1583 for(i=0; i<LSM_LOCK_NREADER; i++){ 1584 ShmReader *p = &pShm->aReader[i]; 1585 if( p->iLsmId ){ 1586 i64 iThis = p->iLsmId; 1587 if( iThis!=0 && iInUse>iThis ){ 1588 int rc = lsmShmLock(db, LSM_LOCK_READER(i), LSM_LOCK_EXCL, 0); 1589 if( rc==LSM_OK ){ 1590 p->iLsmId = 0; 1591 lsmShmLock(db, LSM_LOCK_READER(i), LSM_LOCK_UNLOCK, 0); 1592 }else if( rc==LSM_BUSY ){ 1593 iInUse = iThis; 1594 }else{ 1595 /* Some error other than LSM_BUSY. Return the error code to 1596 ** the caller in this case. */ 1597 return rc; 1598 } 1599 } 1600 } 1601 } 1602 1603 *piInUse = iInUse; 1604 return LSM_OK; 1605 } 1606 1607 int lsmTreeInUse(lsm_db *db, u32 iShmid, int *pbInUse){ 1608 if( db->treehdr.iUsedShmid==iShmid ){ 1609 *pbInUse = 1; 1610 return LSM_OK; 1611 } 1612 return isInUse(db, 0, iShmid, pbInUse); 1613 } 1614 1615 int lsmLsmInUse(lsm_db *db, i64 iLsmId, int *pbInUse){ 1616 if( db->pClient && db->pClient->iId<=iLsmId ){ 1617 *pbInUse = 1; 1618 return LSM_OK; 1619 } 1620 return isInUse(db, iLsmId, 0, pbInUse); 1621 } 1622 1623 /* 1624 ** This function may only be called after a successful call to 1625 ** lsmDbDatabaseConnect(). It returns true if the connection is in 1626 ** multi-process mode, or false otherwise. 1627 */ 1628 int lsmDbMultiProc(lsm_db *pDb){ 1629 return pDb->pDatabase && pDb->pDatabase->bMultiProc; 1630 } 1631 1632 1633 /************************************************************************* 1634 ************************************************************************** 1635 ************************************************************************** 1636 ************************************************************************** 1637 ************************************************************************** 1638 *************************************************************************/ 1639 1640 /* 1641 ** Ensure that database connection db has cached pointers to at least the 1642 ** first nChunk chunks of shared memory. 1643 */ 1644 int lsmShmCacheChunks(lsm_db *db, int nChunk){ 1645 int rc = LSM_OK; 1646 if( nChunk>db->nShm ){ 1647 static const int NINCR = 16; 1648 Database *p = db->pDatabase; 1649 lsm_env *pEnv = db->pEnv; 1650 int nAlloc; 1651 int i; 1652 1653 /* Ensure that the db->apShm[] array is large enough. If an attempt to 1654 ** allocate memory fails, return LSM_NOMEM immediately. The apShm[] array 1655 ** is always extended in multiples of 16 entries - so the actual allocated 1656 ** size can be inferred from nShm. */ 1657 nAlloc = ((db->nShm + NINCR - 1) / NINCR) * NINCR; 1658 while( nChunk>=nAlloc ){ 1659 void **apShm; 1660 nAlloc += NINCR; 1661 apShm = lsmRealloc(pEnv, db->apShm, sizeof(void*)*nAlloc); 1662 if( !apShm ) return LSM_NOMEM_BKPT; 1663 db->apShm = apShm; 1664 } 1665 1666 if( db->bRoTrans ){ 1667 for(i=db->nShm; rc==LSM_OK && i<nChunk; i++){ 1668 db->apShm[i] = lsmMallocZeroRc(pEnv, LSM_SHM_CHUNK_SIZE, &rc); 1669 db->nShm++; 1670 } 1671 1672 }else{ 1673 1674 /* Enter the client mutex */ 1675 lsmMutexEnter(pEnv, p->pClientMutex); 1676 1677 /* Extend the Database objects apShmChunk[] array if necessary. Using the 1678 ** same pattern as for the lsm_db.apShm[] array above. */ 1679 nAlloc = ((p->nShmChunk + NINCR - 1) / NINCR) * NINCR; 1680 while( nChunk>=nAlloc ){ 1681 void **apShm; 1682 nAlloc += NINCR; 1683 apShm = lsmRealloc(pEnv, p->apShmChunk, sizeof(void*)*nAlloc); 1684 if( !apShm ){ 1685 rc = LSM_NOMEM_BKPT; 1686 break; 1687 } 1688 p->apShmChunk = apShm; 1689 } 1690 1691 for(i=db->nShm; rc==LSM_OK && i<nChunk; i++){ 1692 if( i>=p->nShmChunk ){ 1693 void *pChunk = 0; 1694 if( p->bMultiProc==0 ){ 1695 /* Single process mode */ 1696 pChunk = lsmMallocZeroRc(pEnv, LSM_SHM_CHUNK_SIZE, &rc); 1697 }else{ 1698 /* Multi-process mode */ 1699 rc = lsmEnvShmMap(pEnv, p->pFile, i, LSM_SHM_CHUNK_SIZE, &pChunk); 1700 } 1701 if( rc==LSM_OK ){ 1702 p->apShmChunk[i] = pChunk; 1703 p->nShmChunk++; 1704 } 1705 } 1706 if( rc==LSM_OK ){ 1707 db->apShm[i] = p->apShmChunk[i]; 1708 db->nShm++; 1709 } 1710 } 1711 1712 /* Release the client mutex */ 1713 lsmMutexLeave(pEnv, p->pClientMutex); 1714 } 1715 } 1716 1717 return rc; 1718 } 1719 1720 static int lockSharedFile(lsm_env *pEnv, Database *p, int iLock, int eOp){ 1721 int rc = LSM_OK; 1722 if( p->bMultiProc ){ 1723 rc = lsmEnvLock(pEnv, p->pFile, iLock, eOp); 1724 } 1725 return rc; 1726 } 1727 1728 /* 1729 ** Test if it would be possible for connection db to obtain a lock of type 1730 ** eType on the nLock locks starting at iLock. If so, return LSM_OK. If it 1731 ** would not be possible to obtain the lock due to a lock held by another 1732 ** connection, return LSM_BUSY. If an IO or other error occurs (i.e. in the 1733 ** lsm_env.xTestLock function), return some other LSM error code. 1734 ** 1735 ** Note that this function never actually locks the database - it merely 1736 ** queries the system to see if there exists a lock that would prevent 1737 ** it from doing so. 1738 */ 1739 int lsmShmTestLock( 1740 lsm_db *db, 1741 int iLock, 1742 int nLock, 1743 int eOp 1744 ){ 1745 int rc = LSM_OK; 1746 lsm_db *pIter; 1747 Database *p = db->pDatabase; 1748 int i; 1749 u64 mask = 0; 1750 1751 for(i=iLock; i<(iLock+nLock); i++){ 1752 mask |= ((u64)1 << (iLock-1)); 1753 if( eOp==LSM_LOCK_EXCL ) mask |= ((u64)1 << (iLock+32-1)); 1754 } 1755 1756 lsmMutexEnter(db->pEnv, p->pClientMutex); 1757 for(pIter=p->pConn; pIter; pIter=pIter->pNext){ 1758 if( pIter!=db && (pIter->mLock & mask) ){ 1759 assert( pIter!=db ); 1760 break; 1761 } 1762 } 1763 1764 if( pIter ){ 1765 rc = LSM_BUSY; 1766 }else if( p->bMultiProc ){ 1767 rc = lsmEnvTestLock(db->pEnv, p->pFile, iLock, nLock, eOp); 1768 } 1769 1770 lsmMutexLeave(db->pEnv, p->pClientMutex); 1771 return rc; 1772 } 1773 1774 /* 1775 ** Attempt to obtain the lock identified by the iLock and bExcl parameters. 1776 ** If successful, return LSM_OK. If the lock cannot be obtained because 1777 ** there exists some other conflicting lock, return LSM_BUSY. If some other 1778 ** error occurs, return an LSM error code. 1779 ** 1780 ** Parameter iLock must be one of LSM_LOCK_WRITER, WORKER or CHECKPOINTER, 1781 ** or else a value returned by the LSM_LOCK_READER macro. 1782 */ 1783 int lsmShmLock( 1784 lsm_db *db, 1785 int iLock, 1786 int eOp, /* One of LSM_LOCK_UNLOCK, SHARED or EXCL */ 1787 int bBlock /* True for a blocking lock */ 1788 ){ 1789 lsm_db *pIter; 1790 const u64 me = ((u64)1 << (iLock-1)); 1791 const u64 ms = ((u64)1 << (iLock+32-1)); 1792 int rc = LSM_OK; 1793 Database *p = db->pDatabase; 1794 1795 assert( eOp!=LSM_LOCK_EXCL || p->bReadonly==0 ); 1796 assert( iLock>=1 && iLock<=LSM_LOCK_RWCLIENT(LSM_LOCK_NRWCLIENT-1) ); 1797 assert( LSM_LOCK_RWCLIENT(LSM_LOCK_NRWCLIENT-1)<=32 ); 1798 assert( eOp==LSM_LOCK_UNLOCK || eOp==LSM_LOCK_SHARED || eOp==LSM_LOCK_EXCL ); 1799 1800 /* Check for a no-op. Proceed only if this is not one of those. */ 1801 if( (eOp==LSM_LOCK_UNLOCK && (db->mLock & (me|ms))!=0) 1802 || (eOp==LSM_LOCK_SHARED && (db->mLock & (me|ms))!=ms) 1803 || (eOp==LSM_LOCK_EXCL && (db->mLock & me)==0) 1804 ){ 1805 int nExcl = 0; /* Number of connections holding EXCLUSIVE */ 1806 int nShared = 0; /* Number of connections holding SHARED */ 1807 lsmMutexEnter(db->pEnv, p->pClientMutex); 1808 1809 /* Figure out the locks currently held by this process on iLock, not 1810 ** including any held by connection db. */ 1811 for(pIter=p->pConn; pIter; pIter=pIter->pNext){ 1812 assert( (pIter->mLock & me)==0 || (pIter->mLock & ms)!=0 ); 1813 if( pIter!=db ){ 1814 if( pIter->mLock & me ){ 1815 nExcl++; 1816 }else if( pIter->mLock & ms ){ 1817 nShared++; 1818 } 1819 } 1820 } 1821 assert( nExcl==0 || nExcl==1 ); 1822 assert( nExcl==0 || nShared==0 ); 1823 assert( nExcl==0 || (db->mLock & (me|ms))==0 ); 1824 1825 switch( eOp ){ 1826 case LSM_LOCK_UNLOCK: 1827 if( nShared==0 ){ 1828 lockSharedFile(db->pEnv, p, iLock, LSM_LOCK_UNLOCK); 1829 } 1830 db->mLock &= ~(me|ms); 1831 break; 1832 1833 case LSM_LOCK_SHARED: 1834 if( nExcl ){ 1835 rc = LSM_BUSY; 1836 }else{ 1837 if( nShared==0 ){ 1838 rc = lockSharedFile(db->pEnv, p, iLock, LSM_LOCK_SHARED); 1839 } 1840 if( rc==LSM_OK ){ 1841 db->mLock |= ms; 1842 db->mLock &= ~me; 1843 } 1844 } 1845 break; 1846 1847 default: 1848 assert( eOp==LSM_LOCK_EXCL ); 1849 if( nExcl || nShared ){ 1850 rc = LSM_BUSY; 1851 }else{ 1852 rc = lockSharedFile(db->pEnv, p, iLock, LSM_LOCK_EXCL); 1853 if( rc==LSM_OK ){ 1854 db->mLock |= (me|ms); 1855 } 1856 } 1857 break; 1858 } 1859 1860 lsmMutexLeave(db->pEnv, p->pClientMutex); 1861 } 1862 1863 return rc; 1864 } 1865 1866 #ifdef LSM_DEBUG 1867 1868 int shmLockType(lsm_db *db, int iLock){ 1869 const u64 me = ((u64)1 << (iLock-1)); 1870 const u64 ms = ((u64)1 << (iLock+32-1)); 1871 1872 if( db->mLock & me ) return LSM_LOCK_EXCL; 1873 if( db->mLock & ms ) return LSM_LOCK_SHARED; 1874 return LSM_LOCK_UNLOCK; 1875 } 1876 1877 /* 1878 ** The arguments passed to this function are similar to those passed to 1879 ** the lsmShmLock() function. However, instead of obtaining a new lock 1880 ** this function returns true if the specified connection already holds 1881 ** (or does not hold) such a lock, depending on the value of eOp. As 1882 ** follows: 1883 ** 1884 ** (eOp==LSM_LOCK_UNLOCK) -> true if db has no lock on iLock 1885 ** (eOp==LSM_LOCK_SHARED) -> true if db has at least a SHARED lock on iLock. 1886 ** (eOp==LSM_LOCK_EXCL) -> true if db has an EXCLUSIVE lock on iLock. 1887 */ 1888 int lsmShmAssertLock(lsm_db *db, int iLock, int eOp){ 1889 int ret; 1890 int eHave; 1891 1892 assert( iLock>=1 && iLock<=LSM_LOCK_READER(LSM_LOCK_NREADER-1) ); 1893 assert( iLock<=16 ); 1894 assert( eOp==LSM_LOCK_UNLOCK || eOp==LSM_LOCK_SHARED || eOp==LSM_LOCK_EXCL ); 1895 1896 eHave = shmLockType(db, iLock); 1897 1898 switch( eOp ){ 1899 case LSM_LOCK_UNLOCK: 1900 ret = (eHave==LSM_LOCK_UNLOCK); 1901 break; 1902 case LSM_LOCK_SHARED: 1903 ret = (eHave!=LSM_LOCK_UNLOCK); 1904 break; 1905 case LSM_LOCK_EXCL: 1906 ret = (eHave==LSM_LOCK_EXCL); 1907 break; 1908 default: 1909 assert( !"bad eOp value passed to lsmShmAssertLock()" ); 1910 break; 1911 } 1912 1913 return ret; 1914 } 1915 1916 int lsmShmAssertWorker(lsm_db *db){ 1917 return lsmShmAssertLock(db, LSM_LOCK_WORKER, LSM_LOCK_EXCL) && db->pWorker; 1918 } 1919 1920 /* 1921 ** This function does not contribute to library functionality, and is not 1922 ** included in release builds. It is intended to be called from within 1923 ** an interactive debugger. 1924 ** 1925 ** When called, this function prints a single line of human readable output 1926 ** to stdout describing the locks currently held by the connection. For 1927 ** example: 1928 ** 1929 ** (gdb) call print_db_locks(pDb) 1930 ** (shared on dms2) (exclusive on writer) 1931 */ 1932 void print_db_locks(lsm_db *db){ 1933 int iLock; 1934 for(iLock=0; iLock<16; iLock++){ 1935 int bOne = 0; 1936 const char *azLock[] = {0, "shared", "exclusive"}; 1937 const char *azName[] = { 1938 0, "dms1", "dms2", "writer", "worker", "checkpointer", 1939 "reader0", "reader1", "reader2", "reader3", "reader4", "reader5" 1940 }; 1941 int eHave = shmLockType(db, iLock); 1942 if( azLock[eHave] ){ 1943 printf("%s(%s on %s)", (bOne?" ":""), azLock[eHave], azName[iLock]); 1944 bOne = 1; 1945 } 1946 } 1947 printf("\n"); 1948 } 1949 void print_all_db_locks(lsm_db *db){ 1950 lsm_db *p; 1951 for(p=db->pDatabase->pConn; p; p=p->pNext){ 1952 printf("%s connection %p ", ((p==db)?"*":""), p); 1953 print_db_locks(p); 1954 } 1955 } 1956 #endif 1957 1958 void lsmShmBarrier(lsm_db *db){ 1959 lsmEnvShmBarrier(db->pEnv); 1960 } 1961 1962 int lsm_checkpoint(lsm_db *pDb, int *pnKB){ 1963 int rc; /* Return code */ 1964 u32 nWrite = 0; /* Number of pages checkpointed */ 1965 1966 /* Attempt the checkpoint. If successful, nWrite is set to the number of 1967 ** pages written between this and the previous checkpoint. */ 1968 rc = lsmCheckpointWrite(pDb, 0, &nWrite); 1969 1970 /* If required, calculate the output variable (KB of data checkpointed). 1971 ** Set it to zero if an error occured. */ 1972 if( pnKB ){ 1973 int nKB = 0; 1974 if( rc==LSM_OK && nWrite ){ 1975 nKB = (((i64)nWrite * lsmFsPageSize(pDb->pFS)) + 1023) / 1024; 1976 } 1977 *pnKB = nKB; 1978 } 1979 1980 return rc; 1981 } 1982