1 /* 2 ** 2005 December 14 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 ** $Id: test_async.c,v 1.51 2009/03/25 14:24:42 drh Exp $ 14 ** 15 ** This file contains an example implementation of an asynchronous IO 16 ** backend for SQLite. 17 ** 18 ** WHAT IS ASYNCHRONOUS I/O? 19 ** 20 ** With asynchronous I/O, write requests are handled by a separate thread 21 ** running in the background. This means that the thread that initiates 22 ** a database write does not have to wait for (sometimes slow) disk I/O 23 ** to occur. The write seems to happen very quickly, though in reality 24 ** it is happening at its usual slow pace in the background. 25 ** 26 ** Asynchronous I/O appears to give better responsiveness, but at a price. 27 ** You lose the Durable property. With the default I/O backend of SQLite, 28 ** once a write completes, you know that the information you wrote is 29 ** safely on disk. With the asynchronous I/O, this is not the case. If 30 ** your program crashes or if a power loss occurs after the database 31 ** write but before the asynchronous write thread has completed, then the 32 ** database change might never make it to disk and the next user of the 33 ** database might not see your change. 34 ** 35 ** You lose Durability with asynchronous I/O, but you still retain the 36 ** other parts of ACID: Atomic, Consistent, and Isolated. Many 37 ** appliations get along fine without the Durablity. 38 ** 39 ** HOW IT WORKS 40 ** 41 ** Asynchronous I/O works by creating a special SQLite "vfs" structure 42 ** and registering it with sqlite3_vfs_register(). When files opened via 43 ** this vfs are written to (using sqlite3OsWrite()), the data is not 44 ** written directly to disk, but is placed in the "write-queue" to be 45 ** handled by the background thread. 46 ** 47 ** When files opened with the asynchronous vfs are read from 48 ** (using sqlite3OsRead()), the data is read from the file on 49 ** disk and the write-queue, so that from the point of view of 50 ** the vfs reader the OsWrite() appears to have already completed. 51 ** 52 ** The special vfs is registered (and unregistered) by calls to 53 ** function asyncEnable() (see below). 54 ** 55 ** LIMITATIONS 56 ** 57 ** This demonstration code is deliberately kept simple in order to keep 58 ** the main ideas clear and easy to understand. Real applications that 59 ** want to do asynchronous I/O might want to add additional capabilities. 60 ** For example, in this demonstration if writes are happening at a steady 61 ** stream that exceeds the I/O capability of the background writer thread, 62 ** the queue of pending write operations will grow without bound until we 63 ** run out of memory. Users of this technique may want to keep track of 64 ** the quantity of pending writes and stop accepting new write requests 65 ** when the buffer gets to be too big. 66 ** 67 ** LOCKING + CONCURRENCY 68 ** 69 ** Multiple connections from within a single process that use this 70 ** implementation of asynchronous IO may access a single database 71 ** file concurrently. From the point of view of the user, if all 72 ** connections are from within a single process, there is no difference 73 ** between the concurrency offered by "normal" SQLite and SQLite 74 ** using the asynchronous backend. 75 ** 76 ** If connections from within multiple database files may access the 77 ** database file, the ENABLE_FILE_LOCKING symbol (see below) must be 78 ** defined. If it is not defined, then no locks are established on 79 ** the database file. In this case, if multiple processes access 80 ** the database file, corruption will quickly result. 81 ** 82 ** If ENABLE_FILE_LOCKING is defined (the default), then connections 83 ** from within multiple processes may access a single database file 84 ** without risking corruption. However concurrency is reduced as 85 ** follows: 86 ** 87 ** * When a connection using asynchronous IO begins a database 88 ** transaction, the database is locked immediately. However the 89 ** lock is not released until after all relevant operations 90 ** in the write-queue have been flushed to disk. This means 91 ** (for example) that the database may remain locked for some 92 ** time after a "COMMIT" or "ROLLBACK" is issued. 93 ** 94 ** * If an application using asynchronous IO executes transactions 95 ** in quick succession, other database users may be effectively 96 ** locked out of the database. This is because when a BEGIN 97 ** is executed, a database lock is established immediately. But 98 ** when the corresponding COMMIT or ROLLBACK occurs, the lock 99 ** is not released until the relevant part of the write-queue 100 ** has been flushed through. As a result, if a COMMIT is followed 101 ** by a BEGIN before the write-queue is flushed through, the database 102 ** is never unlocked,preventing other processes from accessing 103 ** the database. 104 ** 105 ** Defining ENABLE_FILE_LOCKING when using an NFS or other remote 106 ** file-system may slow things down, as synchronous round-trips to the 107 ** server may be required to establish database file locks. 108 */ 109 #define ENABLE_FILE_LOCKING 110 111 #ifndef SQLITE_AMALGAMATION 112 # include "sqliteInt.h" 113 # include <assert.h> 114 # include <string.h> 115 #endif 116 #include <tcl.h> 117 118 /* 119 ** This test uses pthreads and hence only works on unix and with 120 ** a threadsafe build of SQLite. 121 */ 122 #if SQLITE_OS_UNIX && SQLITE_THREADSAFE 123 124 /* 125 ** This demo uses pthreads. If you do not have a pthreads implementation 126 ** for your operating system, you will need to recode the threading 127 ** logic. 128 */ 129 #include <pthread.h> 130 #include <sched.h> 131 132 /* Useful macros used in several places */ 133 #define MIN(x,y) ((x)<(y)?(x):(y)) 134 #define MAX(x,y) ((x)>(y)?(x):(y)) 135 136 /* Forward references */ 137 typedef struct AsyncWrite AsyncWrite; 138 typedef struct AsyncFile AsyncFile; 139 typedef struct AsyncFileData AsyncFileData; 140 typedef struct AsyncFileLock AsyncFileLock; 141 typedef struct AsyncLock AsyncLock; 142 143 /* Enable for debugging */ 144 static int sqlite3async_trace = 0; 145 # define ASYNC_TRACE(X) if( sqlite3async_trace ) asyncTrace X 146 static void asyncTrace(const char *zFormat, ...){ 147 char *z; 148 va_list ap; 149 va_start(ap, zFormat); 150 z = sqlite3_vmprintf(zFormat, ap); 151 va_end(ap); 152 fprintf(stderr, "[%d] %s", (int)pthread_self(), z); 153 sqlite3_free(z); 154 } 155 156 /* 157 ** THREAD SAFETY NOTES 158 ** 159 ** Basic rules: 160 ** 161 ** * Both read and write access to the global write-op queue must be 162 ** protected by the async.queueMutex. As are the async.ioError and 163 ** async.nFile variables. 164 ** 165 ** * The async.pLock list and all AsyncLock and AsyncFileLock 166 ** structures must be protected by the async.lockMutex mutex. 167 ** 168 ** * The file handles from the underlying system are not assumed to 169 ** be thread safe. 170 ** 171 ** * See the last two paragraphs under "The Writer Thread" for 172 ** an assumption to do with file-handle synchronization by the Os. 173 ** 174 ** Deadlock prevention: 175 ** 176 ** There are three mutex used by the system: the "writer" mutex, 177 ** the "queue" mutex and the "lock" mutex. Rules are: 178 ** 179 ** * It is illegal to block on the writer mutex when any other mutex 180 ** are held, and 181 ** 182 ** * It is illegal to block on the queue mutex when the lock mutex 183 ** is held. 184 ** 185 ** i.e. mutex's must be grabbed in the order "writer", "queue", "lock". 186 ** 187 ** File system operations (invoked by SQLite thread): 188 ** 189 ** xOpen 190 ** xDelete 191 ** xFileExists 192 ** 193 ** File handle operations (invoked by SQLite thread): 194 ** 195 ** asyncWrite, asyncClose, asyncTruncate, asyncSync 196 ** 197 ** The operations above add an entry to the global write-op list. They 198 ** prepare the entry, acquire the async.queueMutex momentarily while 199 ** list pointers are manipulated to insert the new entry, then release 200 ** the mutex and signal the writer thread to wake up in case it happens 201 ** to be asleep. 202 ** 203 ** 204 ** asyncRead, asyncFileSize. 205 ** 206 ** Read operations. Both of these read from both the underlying file 207 ** first then adjust their result based on pending writes in the 208 ** write-op queue. So async.queueMutex is held for the duration 209 ** of these operations to prevent other threads from changing the 210 ** queue in mid operation. 211 ** 212 ** 213 ** asyncLock, asyncUnlock, asyncCheckReservedLock 214 ** 215 ** These primitives implement in-process locking using a hash table 216 ** on the file name. Files are locked correctly for connections coming 217 ** from the same process. But other processes cannot see these locks 218 ** and will therefore not honor them. 219 ** 220 ** 221 ** The writer thread: 222 ** 223 ** The async.writerMutex is used to make sure only there is only 224 ** a single writer thread running at a time. 225 ** 226 ** Inside the writer thread is a loop that works like this: 227 ** 228 ** WHILE (write-op list is not empty) 229 ** Do IO operation at head of write-op list 230 ** Remove entry from head of write-op list 231 ** END WHILE 232 ** 233 ** The async.queueMutex is always held during the <write-op list is 234 ** not empty> test, and when the entry is removed from the head 235 ** of the write-op list. Sometimes it is held for the interim 236 ** period (while the IO is performed), and sometimes it is 237 ** relinquished. It is relinquished if (a) the IO op is an 238 ** ASYNC_CLOSE or (b) when the file handle was opened, two of 239 ** the underlying systems handles were opened on the same 240 ** file-system entry. 241 ** 242 ** If condition (b) above is true, then one file-handle 243 ** (AsyncFile.pBaseRead) is used exclusively by sqlite threads to read the 244 ** file, the other (AsyncFile.pBaseWrite) by sqlite3_async_flush() 245 ** threads to perform write() operations. This means that read 246 ** operations are not blocked by asynchronous writes (although 247 ** asynchronous writes may still be blocked by reads). 248 ** 249 ** This assumes that the OS keeps two handles open on the same file 250 ** properly in sync. That is, any read operation that starts after a 251 ** write operation on the same file system entry has completed returns 252 ** data consistent with the write. We also assume that if one thread 253 ** reads a file while another is writing it all bytes other than the 254 ** ones actually being written contain valid data. 255 ** 256 ** If the above assumptions are not true, set the preprocessor symbol 257 ** SQLITE_ASYNC_TWO_FILEHANDLES to 0. 258 */ 259 260 #ifndef SQLITE_ASYNC_TWO_FILEHANDLES 261 /* #define SQLITE_ASYNC_TWO_FILEHANDLES 0 */ 262 #define SQLITE_ASYNC_TWO_FILEHANDLES 1 263 #endif 264 265 /* 266 ** State information is held in the static variable "async" defined 267 ** as the following structure. 268 ** 269 ** Both async.ioError and async.nFile are protected by async.queueMutex. 270 */ 271 static struct TestAsyncStaticData { 272 pthread_mutex_t lockMutex; /* For access to aLock hash table */ 273 pthread_mutex_t queueMutex; /* Mutex for access to write operation queue */ 274 pthread_mutex_t writerMutex; /* Prevents multiple writer threads */ 275 pthread_cond_t queueSignal; /* For waking up sleeping writer thread */ 276 pthread_cond_t emptySignal; /* Notify when the write queue is empty */ 277 AsyncWrite *pQueueFirst; /* Next write operation to be processed */ 278 AsyncWrite *pQueueLast; /* Last write operation on the list */ 279 AsyncLock *pLock; /* Linked list of all AsyncLock structures */ 280 volatile int ioDelay; /* Extra delay between write operations */ 281 volatile int writerHaltWhenIdle; /* Writer thread halts when queue empty */ 282 volatile int writerHaltNow; /* Writer thread halts after next op */ 283 int ioError; /* True if an IO error has occurred */ 284 int nFile; /* Number of open files (from sqlite pov) */ 285 } async = { 286 PTHREAD_MUTEX_INITIALIZER, 287 PTHREAD_MUTEX_INITIALIZER, 288 PTHREAD_MUTEX_INITIALIZER, 289 PTHREAD_COND_INITIALIZER, 290 PTHREAD_COND_INITIALIZER, 291 }; 292 293 /* Possible values of AsyncWrite.op */ 294 #define ASYNC_NOOP 0 295 #define ASYNC_WRITE 1 296 #define ASYNC_SYNC 2 297 #define ASYNC_TRUNCATE 3 298 #define ASYNC_CLOSE 4 299 #define ASYNC_DELETE 5 300 #define ASYNC_OPENEXCLUSIVE 6 301 #define ASYNC_UNLOCK 7 302 303 /* Names of opcodes. Used for debugging only. 304 ** Make sure these stay in sync with the macros above! 305 */ 306 static const char *azOpcodeName[] = { 307 "NOOP", "WRITE", "SYNC", "TRUNCATE", "CLOSE", "DELETE", "OPENEX", "UNLOCK" 308 }; 309 310 /* 311 ** Entries on the write-op queue are instances of the AsyncWrite 312 ** structure, defined here. 313 ** 314 ** The interpretation of the iOffset and nByte variables varies depending 315 ** on the value of AsyncWrite.op: 316 ** 317 ** ASYNC_NOOP: 318 ** No values used. 319 ** 320 ** ASYNC_WRITE: 321 ** iOffset -> Offset in file to write to. 322 ** nByte -> Number of bytes of data to write (pointed to by zBuf). 323 ** 324 ** ASYNC_SYNC: 325 ** nByte -> flags to pass to sqlite3OsSync(). 326 ** 327 ** ASYNC_TRUNCATE: 328 ** iOffset -> Size to truncate file to. 329 ** nByte -> Unused. 330 ** 331 ** ASYNC_CLOSE: 332 ** iOffset -> Unused. 333 ** nByte -> Unused. 334 ** 335 ** ASYNC_DELETE: 336 ** iOffset -> Contains the "syncDir" flag. 337 ** nByte -> Number of bytes of zBuf points to (file name). 338 ** 339 ** ASYNC_OPENEXCLUSIVE: 340 ** iOffset -> Value of "delflag". 341 ** nByte -> Number of bytes of zBuf points to (file name). 342 ** 343 ** ASYNC_UNLOCK: 344 ** nByte -> Argument to sqlite3OsUnlock(). 345 ** 346 ** 347 ** For an ASYNC_WRITE operation, zBuf points to the data to write to the file. 348 ** This space is sqlite3_malloc()d along with the AsyncWrite structure in a 349 ** single blob, so is deleted when sqlite3_free() is called on the parent 350 ** structure. 351 */ 352 struct AsyncWrite { 353 AsyncFileData *pFileData; /* File to write data to or sync */ 354 int op; /* One of ASYNC_xxx etc. */ 355 sqlite_int64 iOffset; /* See above */ 356 int nByte; /* See above */ 357 char *zBuf; /* Data to write to file (or NULL if op!=ASYNC_WRITE) */ 358 AsyncWrite *pNext; /* Next write operation (to any file) */ 359 }; 360 361 /* 362 ** An instance of this structure is created for each distinct open file 363 ** (i.e. if two handles are opened on the one file, only one of these 364 ** structures is allocated) and stored in the async.aLock hash table. The 365 ** keys for async.aLock are the full pathnames of the opened files. 366 ** 367 ** AsyncLock.pList points to the head of a linked list of AsyncFileLock 368 ** structures, one for each handle currently open on the file. 369 ** 370 ** If the opened file is not a main-database (the SQLITE_OPEN_MAIN_DB is 371 ** not passed to the sqlite3OsOpen() call), or if ENABLE_FILE_LOCKING is 372 ** not defined at compile time, variables AsyncLock.pFile and 373 ** AsyncLock.eLock are never used. Otherwise, pFile is a file handle 374 ** opened on the file in question and used to obtain the file-system 375 ** locks required by database connections within this process. 376 ** 377 ** See comments above the asyncLock() function for more details on 378 ** the implementation of database locking used by this backend. 379 */ 380 struct AsyncLock { 381 char *zFile; 382 int nFile; 383 sqlite3_file *pFile; 384 int eLock; 385 AsyncFileLock *pList; 386 AsyncLock *pNext; /* Next in linked list headed by async.pLock */ 387 }; 388 389 /* 390 ** An instance of the following structure is allocated along with each 391 ** AsyncFileData structure (see AsyncFileData.lock), but is only used if the 392 ** file was opened with the SQLITE_OPEN_MAIN_DB. 393 */ 394 struct AsyncFileLock { 395 int eLock; /* Internally visible lock state (sqlite pov) */ 396 int eAsyncLock; /* Lock-state with write-queue unlock */ 397 AsyncFileLock *pNext; 398 }; 399 400 /* 401 ** The AsyncFile structure is a subclass of sqlite3_file used for 402 ** asynchronous IO. 403 ** 404 ** All of the actual data for the structure is stored in the structure 405 ** pointed to by AsyncFile.pData, which is allocated as part of the 406 ** sqlite3OsOpen() using sqlite3_malloc(). The reason for this is that the 407 ** lifetime of the AsyncFile structure is ended by the caller after OsClose() 408 ** is called, but the data in AsyncFileData may be required by the 409 ** writer thread after that point. 410 */ 411 struct AsyncFile { 412 sqlite3_io_methods *pMethod; 413 AsyncFileData *pData; 414 }; 415 struct AsyncFileData { 416 char *zName; /* Underlying OS filename - used for debugging */ 417 int nName; /* Number of characters in zName */ 418 sqlite3_file *pBaseRead; /* Read handle to the underlying Os file */ 419 sqlite3_file *pBaseWrite; /* Write handle to the underlying Os file */ 420 AsyncFileLock lock; /* Lock state for this handle */ 421 AsyncLock *pLock; /* AsyncLock object for this file system entry */ 422 AsyncWrite closeOp; /* Preallocated close operation */ 423 }; 424 425 /* 426 ** The following async_XXX functions are debugging wrappers around the 427 ** corresponding pthread_XXX functions: 428 ** 429 ** pthread_mutex_lock(); 430 ** pthread_mutex_unlock(); 431 ** pthread_mutex_trylock(); 432 ** pthread_cond_wait(); 433 ** 434 ** It is illegal to pass any mutex other than those stored in the 435 ** following global variables of these functions. 436 ** 437 ** async.queueMutex 438 ** async.writerMutex 439 ** async.lockMutex 440 ** 441 ** If NDEBUG is defined, these wrappers do nothing except call the 442 ** corresponding pthreads function. If NDEBUG is not defined, then the 443 ** following variables are used to store the thread-id (as returned 444 ** by pthread_self()) currently holding the mutex, or 0 otherwise: 445 ** 446 ** asyncdebug.queueMutexHolder 447 ** asyncdebug.writerMutexHolder 448 ** asyncdebug.lockMutexHolder 449 ** 450 ** These variables are used by some assert() statements that verify 451 ** the statements made in the "Deadlock Prevention" notes earlier 452 ** in this file. 453 */ 454 #ifndef NDEBUG 455 456 static struct TestAsyncDebugData { 457 pthread_t lockMutexHolder; 458 pthread_t queueMutexHolder; 459 pthread_t writerMutexHolder; 460 } asyncdebug = {0, 0, 0}; 461 462 /* 463 ** Wrapper around pthread_mutex_lock(). Checks that we have not violated 464 ** the anti-deadlock rules (see "Deadlock prevention" above). 465 */ 466 static int async_mutex_lock(pthread_mutex_t *pMutex){ 467 int iIdx; 468 int rc; 469 pthread_mutex_t *aMutex = (pthread_mutex_t *)(&async); 470 pthread_t *aHolder = (pthread_t *)(&asyncdebug); 471 472 /* The code in this 'ifndef NDEBUG' block depends on a certain alignment 473 * of the variables in TestAsyncStaticData and TestAsyncDebugData. The 474 * following assert() statements check that this has not been changed. 475 * 476 * Really, these only need to be run once at startup time. 477 */ 478 assert(&(aMutex[0])==&async.lockMutex); 479 assert(&(aMutex[1])==&async.queueMutex); 480 assert(&(aMutex[2])==&async.writerMutex); 481 assert(&(aHolder[0])==&asyncdebug.lockMutexHolder); 482 assert(&(aHolder[1])==&asyncdebug.queueMutexHolder); 483 assert(&(aHolder[2])==&asyncdebug.writerMutexHolder); 484 485 assert( pthread_self()!=0 ); 486 487 for(iIdx=0; iIdx<3; iIdx++){ 488 if( pMutex==&aMutex[iIdx] ) break; 489 490 /* This is the key assert(). Here we are checking that if the caller 491 * is trying to block on async.writerMutex, neither of the other two 492 * mutex are held. If the caller is trying to block on async.queueMutex, 493 * lockMutex is not held. 494 */ 495 assert(!pthread_equal(aHolder[iIdx], pthread_self())); 496 } 497 assert(iIdx<3); 498 499 rc = pthread_mutex_lock(pMutex); 500 if( rc==0 ){ 501 assert(aHolder[iIdx]==0); 502 aHolder[iIdx] = pthread_self(); 503 } 504 return rc; 505 } 506 507 /* 508 ** Wrapper around pthread_mutex_unlock(). 509 */ 510 static int async_mutex_unlock(pthread_mutex_t *pMutex){ 511 int iIdx; 512 int rc; 513 pthread_mutex_t *aMutex = (pthread_mutex_t *)(&async); 514 pthread_t *aHolder = (pthread_t *)(&asyncdebug); 515 516 for(iIdx=0; iIdx<3; iIdx++){ 517 if( pMutex==&aMutex[iIdx] ) break; 518 } 519 assert(iIdx<3); 520 521 assert(pthread_equal(aHolder[iIdx], pthread_self())); 522 aHolder[iIdx] = 0; 523 rc = pthread_mutex_unlock(pMutex); 524 assert(rc==0); 525 526 return 0; 527 } 528 529 /* 530 ** Wrapper around pthread_mutex_trylock(). 531 */ 532 static int async_mutex_trylock(pthread_mutex_t *pMutex){ 533 int iIdx; 534 int rc; 535 pthread_mutex_t *aMutex = (pthread_mutex_t *)(&async); 536 pthread_t *aHolder = (pthread_t *)(&asyncdebug); 537 538 for(iIdx=0; iIdx<3; iIdx++){ 539 if( pMutex==&aMutex[iIdx] ) break; 540 } 541 assert(iIdx<3); 542 543 rc = pthread_mutex_trylock(pMutex); 544 if( rc==0 ){ 545 assert(aHolder[iIdx]==0); 546 aHolder[iIdx] = pthread_self(); 547 } 548 return rc; 549 } 550 551 /* 552 ** Wrapper around pthread_cond_wait(). 553 */ 554 static int async_cond_wait(pthread_cond_t *pCond, pthread_mutex_t *pMutex){ 555 int iIdx; 556 int rc; 557 pthread_mutex_t *aMutex = (pthread_mutex_t *)(&async); 558 pthread_t *aHolder = (pthread_t *)(&asyncdebug); 559 560 for(iIdx=0; iIdx<3; iIdx++){ 561 if( pMutex==&aMutex[iIdx] ) break; 562 } 563 assert(iIdx<3); 564 565 assert(pthread_equal(aHolder[iIdx],pthread_self())); 566 aHolder[iIdx] = 0; 567 rc = pthread_cond_wait(pCond, pMutex); 568 if( rc==0 ){ 569 aHolder[iIdx] = pthread_self(); 570 } 571 return rc; 572 } 573 574 /* 575 ** Assert that the mutex is held by the current thread. 576 */ 577 static void assert_mutex_is_held(pthread_mutex_t *pMutex){ 578 int iIdx; 579 pthread_mutex_t *aMutex = (pthread_mutex_t *)(&async); 580 pthread_t *aHolder = (pthread_t *)(&asyncdebug); 581 582 for(iIdx=0; iIdx<3; iIdx++){ 583 if( pMutex==&aMutex[iIdx] ) break; 584 } 585 assert(iIdx<3); 586 assert( aHolder[iIdx]==pthread_self() ); 587 } 588 589 /* Call our async_XX wrappers instead of selected pthread_XX functions */ 590 #define pthread_mutex_lock async_mutex_lock 591 #define pthread_mutex_unlock async_mutex_unlock 592 #define pthread_mutex_trylock async_mutex_trylock 593 #define pthread_cond_wait async_cond_wait 594 595 #else /* if defined(NDEBUG) */ 596 597 #define assert_mutex_is_held(X) /* A no-op when not debugging */ 598 599 #endif /* !defined(NDEBUG) */ 600 601 /* 602 ** Add an entry to the end of the global write-op list. pWrite should point 603 ** to an AsyncWrite structure allocated using sqlite3_malloc(). The writer 604 ** thread will call sqlite3_free() to free the structure after the specified 605 ** operation has been completed. 606 ** 607 ** Once an AsyncWrite structure has been added to the list, it becomes the 608 ** property of the writer thread and must not be read or modified by the 609 ** caller. 610 */ 611 static void addAsyncWrite(AsyncWrite *pWrite){ 612 /* We must hold the queue mutex in order to modify the queue pointers */ 613 pthread_mutex_lock(&async.queueMutex); 614 615 /* Add the record to the end of the write-op queue */ 616 assert( !pWrite->pNext ); 617 if( async.pQueueLast ){ 618 assert( async.pQueueFirst ); 619 async.pQueueLast->pNext = pWrite; 620 }else{ 621 async.pQueueFirst = pWrite; 622 } 623 async.pQueueLast = pWrite; 624 ASYNC_TRACE(("PUSH %p (%s %s %d)\n", pWrite, azOpcodeName[pWrite->op], 625 pWrite->pFileData ? pWrite->pFileData->zName : "-", pWrite->iOffset)); 626 627 if( pWrite->op==ASYNC_CLOSE ){ 628 async.nFile--; 629 } 630 631 /* Drop the queue mutex */ 632 pthread_mutex_unlock(&async.queueMutex); 633 634 /* The writer thread might have been idle because there was nothing 635 ** on the write-op queue for it to do. So wake it up. */ 636 pthread_cond_signal(&async.queueSignal); 637 } 638 639 /* 640 ** Increment async.nFile in a thread-safe manner. 641 */ 642 static void incrOpenFileCount(){ 643 /* We must hold the queue mutex in order to modify async.nFile */ 644 pthread_mutex_lock(&async.queueMutex); 645 if( async.nFile==0 ){ 646 async.ioError = SQLITE_OK; 647 } 648 async.nFile++; 649 pthread_mutex_unlock(&async.queueMutex); 650 } 651 652 /* 653 ** This is a utility function to allocate and populate a new AsyncWrite 654 ** structure and insert it (via addAsyncWrite() ) into the global list. 655 */ 656 static int addNewAsyncWrite( 657 AsyncFileData *pFileData, 658 int op, 659 sqlite3_int64 iOffset, 660 int nByte, 661 const char *zByte 662 ){ 663 AsyncWrite *p; 664 if( op!=ASYNC_CLOSE && async.ioError ){ 665 return async.ioError; 666 } 667 p = sqlite3_malloc(sizeof(AsyncWrite) + (zByte?nByte:0)); 668 if( !p ){ 669 /* The upper layer does not expect operations like OsWrite() to 670 ** return SQLITE_NOMEM. This is partly because under normal conditions 671 ** SQLite is required to do rollback without calling malloc(). So 672 ** if malloc() fails here, treat it as an I/O error. The above 673 ** layer knows how to handle that. 674 */ 675 return SQLITE_IOERR; 676 } 677 p->op = op; 678 p->iOffset = iOffset; 679 p->nByte = nByte; 680 p->pFileData = pFileData; 681 p->pNext = 0; 682 if( zByte ){ 683 p->zBuf = (char *)&p[1]; 684 memcpy(p->zBuf, zByte, nByte); 685 }else{ 686 p->zBuf = 0; 687 } 688 addAsyncWrite(p); 689 return SQLITE_OK; 690 } 691 692 /* 693 ** Close the file. This just adds an entry to the write-op list, the file is 694 ** not actually closed. 695 */ 696 static int asyncClose(sqlite3_file *pFile){ 697 AsyncFileData *p = ((AsyncFile *)pFile)->pData; 698 699 /* Unlock the file, if it is locked */ 700 pthread_mutex_lock(&async.lockMutex); 701 p->lock.eLock = 0; 702 pthread_mutex_unlock(&async.lockMutex); 703 704 addAsyncWrite(&p->closeOp); 705 return SQLITE_OK; 706 } 707 708 /* 709 ** Implementation of sqlite3OsWrite() for asynchronous files. Instead of 710 ** writing to the underlying file, this function adds an entry to the end of 711 ** the global AsyncWrite list. Either SQLITE_OK or SQLITE_NOMEM may be 712 ** returned. 713 */ 714 static int asyncWrite( 715 sqlite3_file *pFile, 716 const void *pBuf, 717 int amt, 718 sqlite3_int64 iOff 719 ){ 720 AsyncFileData *p = ((AsyncFile *)pFile)->pData; 721 return addNewAsyncWrite(p, ASYNC_WRITE, iOff, amt, pBuf); 722 } 723 724 /* 725 ** Read data from the file. First we read from the filesystem, then adjust 726 ** the contents of the buffer based on ASYNC_WRITE operations in the 727 ** write-op queue. 728 ** 729 ** This method holds the mutex from start to finish. 730 */ 731 static int asyncRead( 732 sqlite3_file *pFile, 733 void *zOut, 734 int iAmt, 735 sqlite3_int64 iOffset 736 ){ 737 AsyncFileData *p = ((AsyncFile *)pFile)->pData; 738 int rc = SQLITE_OK; 739 sqlite3_int64 filesize; 740 int nRead; 741 sqlite3_file *pBase = p->pBaseRead; 742 743 /* Grab the write queue mutex for the duration of the call */ 744 pthread_mutex_lock(&async.queueMutex); 745 746 /* If an I/O error has previously occurred in this virtual file 747 ** system, then all subsequent operations fail. 748 */ 749 if( async.ioError!=SQLITE_OK ){ 750 rc = async.ioError; 751 goto asyncread_out; 752 } 753 754 if( pBase->pMethods ){ 755 rc = pBase->pMethods->xFileSize(pBase, &filesize); 756 if( rc!=SQLITE_OK ){ 757 goto asyncread_out; 758 } 759 nRead = MIN(filesize - iOffset, iAmt); 760 if( nRead>0 ){ 761 rc = pBase->pMethods->xRead(pBase, zOut, nRead, iOffset); 762 ASYNC_TRACE(("READ %s %d bytes at %d\n", p->zName, nRead, iOffset)); 763 } 764 } 765 766 if( rc==SQLITE_OK ){ 767 AsyncWrite *pWrite; 768 char *zName = p->zName; 769 770 for(pWrite=async.pQueueFirst; pWrite; pWrite = pWrite->pNext){ 771 if( pWrite->op==ASYNC_WRITE && ( 772 (pWrite->pFileData==p) || 773 (zName && pWrite->pFileData->zName==zName) 774 )){ 775 int iBeginOut = (pWrite->iOffset-iOffset); 776 int iBeginIn = -iBeginOut; 777 int nCopy; 778 779 if( iBeginIn<0 ) iBeginIn = 0; 780 if( iBeginOut<0 ) iBeginOut = 0; 781 nCopy = MIN(pWrite->nByte-iBeginIn, iAmt-iBeginOut); 782 783 if( nCopy>0 ){ 784 memcpy(&((char *)zOut)[iBeginOut], &pWrite->zBuf[iBeginIn], nCopy); 785 ASYNC_TRACE(("OVERREAD %d bytes at %d\n", nCopy, iBeginOut+iOffset)); 786 } 787 } 788 } 789 } 790 791 asyncread_out: 792 pthread_mutex_unlock(&async.queueMutex); 793 return rc; 794 } 795 796 /* 797 ** Truncate the file to nByte bytes in length. This just adds an entry to 798 ** the write-op list, no IO actually takes place. 799 */ 800 static int asyncTruncate(sqlite3_file *pFile, sqlite3_int64 nByte){ 801 AsyncFileData *p = ((AsyncFile *)pFile)->pData; 802 return addNewAsyncWrite(p, ASYNC_TRUNCATE, nByte, 0, 0); 803 } 804 805 /* 806 ** Sync the file. This just adds an entry to the write-op list, the 807 ** sync() is done later by sqlite3_async_flush(). 808 */ 809 static int asyncSync(sqlite3_file *pFile, int flags){ 810 AsyncFileData *p = ((AsyncFile *)pFile)->pData; 811 return addNewAsyncWrite(p, ASYNC_SYNC, 0, flags, 0); 812 } 813 814 /* 815 ** Read the size of the file. First we read the size of the file system 816 ** entry, then adjust for any ASYNC_WRITE or ASYNC_TRUNCATE operations 817 ** currently in the write-op list. 818 ** 819 ** This method holds the mutex from start to finish. 820 */ 821 int asyncFileSize(sqlite3_file *pFile, sqlite3_int64 *piSize){ 822 AsyncFileData *p = ((AsyncFile *)pFile)->pData; 823 int rc = SQLITE_OK; 824 sqlite3_int64 s = 0; 825 sqlite3_file *pBase; 826 827 pthread_mutex_lock(&async.queueMutex); 828 829 /* Read the filesystem size from the base file. If pBaseRead is NULL, this 830 ** means the file hasn't been opened yet. In this case all relevant data 831 ** must be in the write-op queue anyway, so we can omit reading from the 832 ** file-system. 833 */ 834 pBase = p->pBaseRead; 835 if( pBase->pMethods ){ 836 rc = pBase->pMethods->xFileSize(pBase, &s); 837 } 838 839 if( rc==SQLITE_OK ){ 840 AsyncWrite *pWrite; 841 for(pWrite=async.pQueueFirst; pWrite; pWrite = pWrite->pNext){ 842 if( pWrite->op==ASYNC_DELETE 843 && p->zName 844 && strcmp(p->zName, pWrite->zBuf)==0 845 ){ 846 s = 0; 847 }else if( pWrite->pFileData && ( 848 (pWrite->pFileData==p) 849 || (p->zName && pWrite->pFileData->zName==p->zName) 850 )){ 851 switch( pWrite->op ){ 852 case ASYNC_WRITE: 853 s = MAX(pWrite->iOffset + (sqlite3_int64)(pWrite->nByte), s); 854 break; 855 case ASYNC_TRUNCATE: 856 s = MIN(s, pWrite->iOffset); 857 break; 858 } 859 } 860 } 861 *piSize = s; 862 } 863 pthread_mutex_unlock(&async.queueMutex); 864 return rc; 865 } 866 867 /* 868 ** Lock or unlock the actual file-system entry. 869 */ 870 static int getFileLock(AsyncLock *pLock){ 871 int rc = SQLITE_OK; 872 AsyncFileLock *pIter; 873 int eRequired = 0; 874 875 if( pLock->pFile ){ 876 for(pIter=pLock->pList; pIter; pIter=pIter->pNext){ 877 assert(pIter->eAsyncLock>=pIter->eLock); 878 if( pIter->eAsyncLock>eRequired ){ 879 eRequired = pIter->eAsyncLock; 880 assert(eRequired>=0 && eRequired<=SQLITE_LOCK_EXCLUSIVE); 881 } 882 } 883 884 if( eRequired>pLock->eLock ){ 885 rc = pLock->pFile->pMethods->xLock(pLock->pFile, eRequired); 886 if( rc==SQLITE_OK ){ 887 pLock->eLock = eRequired; 888 } 889 } 890 else if( eRequired<pLock->eLock && eRequired<=SQLITE_LOCK_SHARED ){ 891 rc = pLock->pFile->pMethods->xUnlock(pLock->pFile, eRequired); 892 if( rc==SQLITE_OK ){ 893 pLock->eLock = eRequired; 894 } 895 } 896 } 897 898 return rc; 899 } 900 901 /* 902 ** Return the AsyncLock structure from the global async.pLock list 903 ** associated with the file-system entry identified by path zName 904 ** (a string of nName bytes). If no such structure exists, return 0. 905 */ 906 static AsyncLock *findLock(const char *zName, int nName){ 907 AsyncLock *p = async.pLock; 908 while( p && (p->nFile!=nName || memcmp(p->zFile, zName, nName)) ){ 909 p = p->pNext; 910 } 911 return p; 912 } 913 914 /* 915 ** The following two methods - asyncLock() and asyncUnlock() - are used 916 ** to obtain and release locks on database files opened with the 917 ** asynchronous backend. 918 */ 919 static int asyncLock(sqlite3_file *pFile, int eLock){ 920 int rc = SQLITE_OK; 921 AsyncFileData *p = ((AsyncFile *)pFile)->pData; 922 923 if( p->zName ){ 924 pthread_mutex_lock(&async.lockMutex); 925 if( p->lock.eLock<eLock ){ 926 AsyncLock *pLock = p->pLock; 927 AsyncFileLock *pIter; 928 assert(pLock && pLock->pList); 929 for(pIter=pLock->pList; pIter; pIter=pIter->pNext){ 930 if( pIter!=&p->lock && ( 931 (eLock==SQLITE_LOCK_EXCLUSIVE && pIter->eLock>=SQLITE_LOCK_SHARED) || 932 (eLock==SQLITE_LOCK_PENDING && pIter->eLock>=SQLITE_LOCK_RESERVED) || 933 (eLock==SQLITE_LOCK_RESERVED && pIter->eLock>=SQLITE_LOCK_RESERVED) || 934 (eLock==SQLITE_LOCK_SHARED && pIter->eLock>=SQLITE_LOCK_PENDING) 935 )){ 936 rc = SQLITE_BUSY; 937 } 938 } 939 if( rc==SQLITE_OK ){ 940 p->lock.eLock = eLock; 941 p->lock.eAsyncLock = MAX(p->lock.eAsyncLock, eLock); 942 } 943 assert(p->lock.eAsyncLock>=p->lock.eLock); 944 if( rc==SQLITE_OK ){ 945 rc = getFileLock(pLock); 946 } 947 } 948 pthread_mutex_unlock(&async.lockMutex); 949 } 950 951 ASYNC_TRACE(("LOCK %d (%s) rc=%d\n", eLock, p->zName, rc)); 952 return rc; 953 } 954 static int asyncUnlock(sqlite3_file *pFile, int eLock){ 955 int rc = SQLITE_OK; 956 AsyncFileData *p = ((AsyncFile *)pFile)->pData; 957 if( p->zName ){ 958 AsyncFileLock *pLock = &p->lock; 959 pthread_mutex_lock(&async.lockMutex); 960 pLock->eLock = MIN(pLock->eLock, eLock); 961 pthread_mutex_unlock(&async.lockMutex); 962 rc = addNewAsyncWrite(p, ASYNC_UNLOCK, 0, eLock, 0); 963 } 964 return rc; 965 } 966 967 /* 968 ** This function is called when the pager layer first opens a database file 969 ** and is checking for a hot-journal. 970 */ 971 static int asyncCheckReservedLock(sqlite3_file *pFile, int *pResOut){ 972 int ret = 0; 973 AsyncFileLock *pIter; 974 AsyncFileData *p = ((AsyncFile *)pFile)->pData; 975 976 pthread_mutex_lock(&async.lockMutex); 977 for(pIter=p->pLock->pList; pIter; pIter=pIter->pNext){ 978 if( pIter->eLock>=SQLITE_LOCK_RESERVED ){ 979 ret = 1; 980 } 981 } 982 pthread_mutex_unlock(&async.lockMutex); 983 984 ASYNC_TRACE(("CHECK-LOCK %d (%s)\n", ret, p->zName)); 985 *pResOut = ret; 986 return SQLITE_OK; 987 } 988 989 /* 990 ** sqlite3_file_control() implementation. 991 */ 992 static int asyncFileControl(sqlite3_file *id, int op, void *pArg){ 993 switch( op ){ 994 case SQLITE_FCNTL_LOCKSTATE: { 995 pthread_mutex_lock(&async.lockMutex); 996 *(int*)pArg = ((AsyncFile*)id)->pData->lock.eLock; 997 pthread_mutex_unlock(&async.lockMutex); 998 return SQLITE_OK; 999 } 1000 } 1001 return SQLITE_ERROR; 1002 } 1003 1004 /* 1005 ** Return the device characteristics and sector-size of the device. It 1006 ** is not tricky to implement these correctly, as this backend might 1007 ** not have an open file handle at this point. 1008 */ 1009 static int asyncSectorSize(sqlite3_file *pFile){ 1010 return 512; 1011 } 1012 static int asyncDeviceCharacteristics(sqlite3_file *pFile){ 1013 return 0; 1014 } 1015 1016 static int unlinkAsyncFile(AsyncFileData *pData){ 1017 AsyncFileLock **ppIter; 1018 int rc = SQLITE_OK; 1019 1020 if( pData->zName ){ 1021 AsyncLock *pLock = pData->pLock; 1022 for(ppIter=&pLock->pList; *ppIter; ppIter=&((*ppIter)->pNext)){ 1023 if( (*ppIter)==&pData->lock ){ 1024 *ppIter = pData->lock.pNext; 1025 break; 1026 } 1027 } 1028 if( !pLock->pList ){ 1029 AsyncLock **pp; 1030 if( pLock->pFile ){ 1031 pLock->pFile->pMethods->xClose(pLock->pFile); 1032 } 1033 for(pp=&async.pLock; *pp!=pLock; pp=&((*pp)->pNext)); 1034 *pp = pLock->pNext; 1035 sqlite3_free(pLock); 1036 }else{ 1037 rc = getFileLock(pLock); 1038 } 1039 } 1040 1041 return rc; 1042 } 1043 1044 /* 1045 ** Open a file. 1046 */ 1047 static int asyncOpen( 1048 sqlite3_vfs *pAsyncVfs, 1049 const char *zName, 1050 sqlite3_file *pFile, 1051 int flags, 1052 int *pOutFlags 1053 ){ 1054 static sqlite3_io_methods async_methods = { 1055 1, /* iVersion */ 1056 asyncClose, /* xClose */ 1057 asyncRead, /* xRead */ 1058 asyncWrite, /* xWrite */ 1059 asyncTruncate, /* xTruncate */ 1060 asyncSync, /* xSync */ 1061 asyncFileSize, /* xFileSize */ 1062 asyncLock, /* xLock */ 1063 asyncUnlock, /* xUnlock */ 1064 asyncCheckReservedLock, /* xCheckReservedLock */ 1065 asyncFileControl, /* xFileControl */ 1066 asyncSectorSize, /* xSectorSize */ 1067 asyncDeviceCharacteristics /* xDeviceCharacteristics */ 1068 }; 1069 1070 sqlite3_vfs *pVfs = (sqlite3_vfs *)pAsyncVfs->pAppData; 1071 AsyncFile *p = (AsyncFile *)pFile; 1072 int nName = 0; 1073 int rc = SQLITE_OK; 1074 int nByte; 1075 AsyncFileData *pData; 1076 AsyncLock *pLock = 0; 1077 char *z; 1078 int isExclusive = (flags&SQLITE_OPEN_EXCLUSIVE); 1079 1080 /* If zName is NULL, then the upper layer is requesting an anonymous file */ 1081 if( zName ){ 1082 nName = strlen(zName)+1; 1083 } 1084 1085 nByte = ( 1086 sizeof(AsyncFileData) + /* AsyncFileData structure */ 1087 2 * pVfs->szOsFile + /* AsyncFileData.pBaseRead and pBaseWrite */ 1088 nName /* AsyncFileData.zName */ 1089 ); 1090 z = sqlite3_malloc(nByte); 1091 if( !z ){ 1092 return SQLITE_NOMEM; 1093 } 1094 memset(z, 0, nByte); 1095 pData = (AsyncFileData*)z; 1096 z += sizeof(pData[0]); 1097 pData->pBaseRead = (sqlite3_file*)z; 1098 z += pVfs->szOsFile; 1099 pData->pBaseWrite = (sqlite3_file*)z; 1100 pData->closeOp.pFileData = pData; 1101 pData->closeOp.op = ASYNC_CLOSE; 1102 1103 if( zName ){ 1104 z += pVfs->szOsFile; 1105 pData->zName = z; 1106 pData->nName = nName; 1107 memcpy(pData->zName, zName, nName); 1108 } 1109 1110 if( !isExclusive ){ 1111 rc = pVfs->xOpen(pVfs, zName, pData->pBaseRead, flags, pOutFlags); 1112 if( rc==SQLITE_OK && ((*pOutFlags)&SQLITE_OPEN_READWRITE) ){ 1113 rc = pVfs->xOpen(pVfs, zName, pData->pBaseWrite, flags, 0); 1114 } 1115 } 1116 1117 pthread_mutex_lock(&async.lockMutex); 1118 1119 if( zName && rc==SQLITE_OK ){ 1120 pLock = findLock(pData->zName, pData->nName); 1121 if( !pLock ){ 1122 int nByte = pVfs->szOsFile + sizeof(AsyncLock) + pData->nName + 1; 1123 pLock = (AsyncLock *)sqlite3_malloc(nByte); 1124 if( pLock ){ 1125 memset(pLock, 0, nByte); 1126 #ifdef ENABLE_FILE_LOCKING 1127 if( flags&SQLITE_OPEN_MAIN_DB ){ 1128 pLock->pFile = (sqlite3_file *)&pLock[1]; 1129 rc = pVfs->xOpen(pVfs, zName, pLock->pFile, flags, 0); 1130 if( rc!=SQLITE_OK ){ 1131 sqlite3_free(pLock); 1132 pLock = 0; 1133 } 1134 } 1135 #endif 1136 if( pLock ){ 1137 pLock->nFile = pData->nName; 1138 pLock->zFile = &((char *)(&pLock[1]))[pVfs->szOsFile]; 1139 memcpy(pLock->zFile, pData->zName, pLock->nFile); 1140 pLock->pNext = async.pLock; 1141 async.pLock = pLock; 1142 } 1143 }else{ 1144 rc = SQLITE_NOMEM; 1145 } 1146 } 1147 } 1148 1149 if( rc==SQLITE_OK ){ 1150 p->pMethod = &async_methods; 1151 p->pData = pData; 1152 1153 /* Link AsyncFileData.lock into the linked list of 1154 ** AsyncFileLock structures for this file. 1155 */ 1156 if( zName ){ 1157 pData->lock.pNext = pLock->pList; 1158 pLock->pList = &pData->lock; 1159 pData->zName = pLock->zFile; 1160 } 1161 }else{ 1162 if( pData->pBaseRead->pMethods ){ 1163 pData->pBaseRead->pMethods->xClose(pData->pBaseRead); 1164 } 1165 if( pData->pBaseWrite->pMethods ){ 1166 pData->pBaseWrite->pMethods->xClose(pData->pBaseWrite); 1167 } 1168 sqlite3_free(pData); 1169 } 1170 1171 pthread_mutex_unlock(&async.lockMutex); 1172 1173 if( rc==SQLITE_OK ){ 1174 incrOpenFileCount(); 1175 pData->pLock = pLock; 1176 } 1177 1178 if( rc==SQLITE_OK && isExclusive ){ 1179 rc = addNewAsyncWrite(pData, ASYNC_OPENEXCLUSIVE, (sqlite3_int64)flags,0,0); 1180 if( rc==SQLITE_OK ){ 1181 if( pOutFlags ) *pOutFlags = flags; 1182 }else{ 1183 pthread_mutex_lock(&async.lockMutex); 1184 unlinkAsyncFile(pData); 1185 pthread_mutex_unlock(&async.lockMutex); 1186 sqlite3_free(pData); 1187 } 1188 } 1189 if( rc!=SQLITE_OK ){ 1190 p->pMethod = 0; 1191 } 1192 return rc; 1193 } 1194 1195 /* 1196 ** Implementation of sqlite3OsDelete. Add an entry to the end of the 1197 ** write-op queue to perform the delete. 1198 */ 1199 static int asyncDelete(sqlite3_vfs *pAsyncVfs, const char *z, int syncDir){ 1200 return addNewAsyncWrite(0, ASYNC_DELETE, syncDir, strlen(z)+1, z); 1201 } 1202 1203 /* 1204 ** Implementation of sqlite3OsAccess. This method holds the mutex from 1205 ** start to finish. 1206 */ 1207 static int asyncAccess( 1208 sqlite3_vfs *pAsyncVfs, 1209 const char *zName, 1210 int flags, 1211 int *pResOut 1212 ){ 1213 int rc; 1214 int ret; 1215 AsyncWrite *p; 1216 sqlite3_vfs *pVfs = (sqlite3_vfs *)pAsyncVfs->pAppData; 1217 1218 assert(flags==SQLITE_ACCESS_READWRITE 1219 || flags==SQLITE_ACCESS_READ 1220 || flags==SQLITE_ACCESS_EXISTS 1221 ); 1222 1223 pthread_mutex_lock(&async.queueMutex); 1224 rc = pVfs->xAccess(pVfs, zName, flags, &ret); 1225 if( rc==SQLITE_OK && flags==SQLITE_ACCESS_EXISTS ){ 1226 for(p=async.pQueueFirst; p; p = p->pNext){ 1227 if( p->op==ASYNC_DELETE && 0==strcmp(p->zBuf, zName) ){ 1228 ret = 0; 1229 }else if( p->op==ASYNC_OPENEXCLUSIVE 1230 && p->pFileData->zName 1231 && 0==strcmp(p->pFileData->zName, zName) 1232 ){ 1233 ret = 1; 1234 } 1235 } 1236 } 1237 ASYNC_TRACE(("ACCESS(%s): %s = %d\n", 1238 flags==SQLITE_ACCESS_READWRITE?"read-write": 1239 flags==SQLITE_ACCESS_READ?"read":"exists" 1240 , zName, ret) 1241 ); 1242 pthread_mutex_unlock(&async.queueMutex); 1243 *pResOut = ret; 1244 return rc; 1245 } 1246 1247 /* 1248 ** Fill in zPathOut with the full path to the file identified by zPath. 1249 */ 1250 static int asyncFullPathname( 1251 sqlite3_vfs *pAsyncVfs, 1252 const char *zPath, 1253 int nPathOut, 1254 char *zPathOut 1255 ){ 1256 int rc; 1257 sqlite3_vfs *pVfs = (sqlite3_vfs *)pAsyncVfs->pAppData; 1258 rc = pVfs->xFullPathname(pVfs, zPath, nPathOut, zPathOut); 1259 1260 /* Because of the way intra-process file locking works, this backend 1261 ** needs to return a canonical path. The following block assumes the 1262 ** file-system uses unix style paths. 1263 */ 1264 if( rc==SQLITE_OK ){ 1265 int iIn; 1266 int iOut = 0; 1267 int nPathOut = strlen(zPathOut); 1268 1269 for(iIn=0; iIn<nPathOut; iIn++){ 1270 1271 /* Replace any occurences of "//" with "/" */ 1272 if( iIn<=(nPathOut-2) && zPathOut[iIn]=='/' && zPathOut[iIn+1]=='/' 1273 ){ 1274 continue; 1275 } 1276 1277 /* Replace any occurences of "/./" with "/" */ 1278 if( iIn<=(nPathOut-3) 1279 && zPathOut[iIn]=='/' && zPathOut[iIn+1]=='.' && zPathOut[iIn+2]=='/' 1280 ){ 1281 iIn++; 1282 continue; 1283 } 1284 1285 /* Replace any occurences of "<path-component>/../" with "" */ 1286 if( iOut>0 && iIn<=(nPathOut-4) 1287 && zPathOut[iIn]=='/' && zPathOut[iIn+1]=='.' 1288 && zPathOut[iIn+2]=='.' && zPathOut[iIn+3]=='/' 1289 ){ 1290 iIn += 3; 1291 iOut--; 1292 for( ; iOut>0 && zPathOut[iOut-1]!='/'; iOut--); 1293 continue; 1294 } 1295 1296 zPathOut[iOut++] = zPathOut[iIn]; 1297 } 1298 zPathOut[iOut] = '\0'; 1299 } 1300 1301 return rc; 1302 } 1303 static void *asyncDlOpen(sqlite3_vfs *pAsyncVfs, const char *zPath){ 1304 sqlite3_vfs *pVfs = (sqlite3_vfs *)pAsyncVfs->pAppData; 1305 return pVfs->xDlOpen(pVfs, zPath); 1306 } 1307 static void asyncDlError(sqlite3_vfs *pAsyncVfs, int nByte, char *zErrMsg){ 1308 sqlite3_vfs *pVfs = (sqlite3_vfs *)pAsyncVfs->pAppData; 1309 pVfs->xDlError(pVfs, nByte, zErrMsg); 1310 } 1311 static void *asyncDlSym( 1312 sqlite3_vfs *pAsyncVfs, 1313 void *pHandle, 1314 const char *zSymbol 1315 ){ 1316 sqlite3_vfs *pVfs = (sqlite3_vfs *)pAsyncVfs->pAppData; 1317 return pVfs->xDlSym(pVfs, pHandle, zSymbol); 1318 } 1319 static void asyncDlClose(sqlite3_vfs *pAsyncVfs, void *pHandle){ 1320 sqlite3_vfs *pVfs = (sqlite3_vfs *)pAsyncVfs->pAppData; 1321 pVfs->xDlClose(pVfs, pHandle); 1322 } 1323 static int asyncRandomness(sqlite3_vfs *pAsyncVfs, int nByte, char *zBufOut){ 1324 sqlite3_vfs *pVfs = (sqlite3_vfs *)pAsyncVfs->pAppData; 1325 return pVfs->xRandomness(pVfs, nByte, zBufOut); 1326 } 1327 static int asyncSleep(sqlite3_vfs *pAsyncVfs, int nMicro){ 1328 sqlite3_vfs *pVfs = (sqlite3_vfs *)pAsyncVfs->pAppData; 1329 return pVfs->xSleep(pVfs, nMicro); 1330 } 1331 static int asyncCurrentTime(sqlite3_vfs *pAsyncVfs, double *pTimeOut){ 1332 sqlite3_vfs *pVfs = (sqlite3_vfs *)pAsyncVfs->pAppData; 1333 return pVfs->xCurrentTime(pVfs, pTimeOut); 1334 } 1335 1336 static sqlite3_vfs async_vfs = { 1337 1, /* iVersion */ 1338 sizeof(AsyncFile), /* szOsFile */ 1339 0, /* mxPathname */ 1340 0, /* pNext */ 1341 "async", /* zName */ 1342 0, /* pAppData */ 1343 asyncOpen, /* xOpen */ 1344 asyncDelete, /* xDelete */ 1345 asyncAccess, /* xAccess */ 1346 asyncFullPathname, /* xFullPathname */ 1347 asyncDlOpen, /* xDlOpen */ 1348 asyncDlError, /* xDlError */ 1349 asyncDlSym, /* xDlSym */ 1350 asyncDlClose, /* xDlClose */ 1351 asyncRandomness, /* xDlError */ 1352 asyncSleep, /* xDlSym */ 1353 asyncCurrentTime /* xDlClose */ 1354 }; 1355 1356 /* 1357 ** Call this routine to enable or disable the 1358 ** asynchronous IO features implemented in this file. 1359 ** 1360 ** This routine is not even remotely threadsafe. Do not call 1361 ** this routine while any SQLite database connections are open. 1362 */ 1363 static void asyncEnable(int enable){ 1364 if( enable ){ 1365 if( !async_vfs.pAppData ){ 1366 async_vfs.pAppData = (void *)sqlite3_vfs_find(0); 1367 async_vfs.mxPathname = ((sqlite3_vfs *)async_vfs.pAppData)->mxPathname; 1368 sqlite3_vfs_register(&async_vfs, 1); 1369 } 1370 }else{ 1371 if( async_vfs.pAppData ){ 1372 sqlite3_vfs_unregister(&async_vfs); 1373 async_vfs.pAppData = 0; 1374 } 1375 } 1376 } 1377 1378 /* 1379 ** This procedure runs in a separate thread, reading messages off of the 1380 ** write queue and processing them one by one. 1381 ** 1382 ** If async.writerHaltNow is true, then this procedure exits 1383 ** after processing a single message. 1384 ** 1385 ** If async.writerHaltWhenIdle is true, then this procedure exits when 1386 ** the write queue is empty. 1387 ** 1388 ** If both of the above variables are false, this procedure runs 1389 ** indefinately, waiting for operations to be added to the write queue 1390 ** and processing them in the order in which they arrive. 1391 ** 1392 ** An artifical delay of async.ioDelay milliseconds is inserted before 1393 ** each write operation in order to simulate the effect of a slow disk. 1394 ** 1395 ** Only one instance of this procedure may be running at a time. 1396 */ 1397 static void *asyncWriterThread(void *pIsStarted){ 1398 sqlite3_vfs *pVfs = (sqlite3_vfs *)(async_vfs.pAppData); 1399 AsyncWrite *p = 0; 1400 int rc = SQLITE_OK; 1401 int holdingMutex = 0; 1402 1403 if( pthread_mutex_trylock(&async.writerMutex) ){ 1404 return 0; 1405 } 1406 (*(int *)pIsStarted) = 1; 1407 while( async.writerHaltNow==0 ){ 1408 int doNotFree = 0; 1409 sqlite3_file *pBase = 0; 1410 1411 if( !holdingMutex ){ 1412 pthread_mutex_lock(&async.queueMutex); 1413 } 1414 while( (p = async.pQueueFirst)==0 ){ 1415 pthread_cond_broadcast(&async.emptySignal); 1416 if( async.writerHaltWhenIdle ){ 1417 pthread_mutex_unlock(&async.queueMutex); 1418 break; 1419 }else{ 1420 ASYNC_TRACE(("IDLE\n")); 1421 pthread_cond_wait(&async.queueSignal, &async.queueMutex); 1422 ASYNC_TRACE(("WAKEUP\n")); 1423 } 1424 } 1425 if( p==0 ) break; 1426 holdingMutex = 1; 1427 1428 /* Right now this thread is holding the mutex on the write-op queue. 1429 ** Variable 'p' points to the first entry in the write-op queue. In 1430 ** the general case, we hold on to the mutex for the entire body of 1431 ** the loop. 1432 ** 1433 ** However in the cases enumerated below, we relinquish the mutex, 1434 ** perform the IO, and then re-request the mutex before removing 'p' from 1435 ** the head of the write-op queue. The idea is to increase concurrency with 1436 ** sqlite threads. 1437 ** 1438 ** * An ASYNC_CLOSE operation. 1439 ** * An ASYNC_OPENEXCLUSIVE operation. For this one, we relinquish 1440 ** the mutex, call the underlying xOpenExclusive() function, then 1441 ** re-aquire the mutex before seting the AsyncFile.pBaseRead 1442 ** variable. 1443 ** * ASYNC_SYNC and ASYNC_WRITE operations, if 1444 ** SQLITE_ASYNC_TWO_FILEHANDLES was set at compile time and two 1445 ** file-handles are open for the particular file being "synced". 1446 */ 1447 if( async.ioError!=SQLITE_OK && p->op!=ASYNC_CLOSE ){ 1448 p->op = ASYNC_NOOP; 1449 } 1450 if( p->pFileData ){ 1451 pBase = p->pFileData->pBaseWrite; 1452 if( 1453 p->op==ASYNC_CLOSE || 1454 p->op==ASYNC_OPENEXCLUSIVE || 1455 (pBase->pMethods && (p->op==ASYNC_SYNC || p->op==ASYNC_WRITE) ) 1456 ){ 1457 pthread_mutex_unlock(&async.queueMutex); 1458 holdingMutex = 0; 1459 } 1460 if( !pBase->pMethods ){ 1461 pBase = p->pFileData->pBaseRead; 1462 } 1463 } 1464 1465 switch( p->op ){ 1466 case ASYNC_NOOP: 1467 break; 1468 1469 case ASYNC_WRITE: 1470 assert( pBase ); 1471 ASYNC_TRACE(("WRITE %s %d bytes at %d\n", 1472 p->pFileData->zName, p->nByte, p->iOffset)); 1473 rc = pBase->pMethods->xWrite(pBase, (void *)(p->zBuf), p->nByte, p->iOffset); 1474 break; 1475 1476 case ASYNC_SYNC: 1477 assert( pBase ); 1478 ASYNC_TRACE(("SYNC %s\n", p->pFileData->zName)); 1479 rc = pBase->pMethods->xSync(pBase, p->nByte); 1480 break; 1481 1482 case ASYNC_TRUNCATE: 1483 assert( pBase ); 1484 ASYNC_TRACE(("TRUNCATE %s to %d bytes\n", 1485 p->pFileData->zName, p->iOffset)); 1486 rc = pBase->pMethods->xTruncate(pBase, p->iOffset); 1487 break; 1488 1489 case ASYNC_CLOSE: { 1490 AsyncFileData *pData = p->pFileData; 1491 ASYNC_TRACE(("CLOSE %s\n", p->pFileData->zName)); 1492 if( pData->pBaseWrite->pMethods ){ 1493 pData->pBaseWrite->pMethods->xClose(pData->pBaseWrite); 1494 } 1495 if( pData->pBaseRead->pMethods ){ 1496 pData->pBaseRead->pMethods->xClose(pData->pBaseRead); 1497 } 1498 1499 /* Unlink AsyncFileData.lock from the linked list of AsyncFileLock 1500 ** structures for this file. Obtain the async.lockMutex mutex 1501 ** before doing so. 1502 */ 1503 pthread_mutex_lock(&async.lockMutex); 1504 rc = unlinkAsyncFile(pData); 1505 pthread_mutex_unlock(&async.lockMutex); 1506 1507 if( !holdingMutex ){ 1508 pthread_mutex_lock(&async.queueMutex); 1509 holdingMutex = 1; 1510 } 1511 assert_mutex_is_held(&async.queueMutex); 1512 async.pQueueFirst = p->pNext; 1513 sqlite3_free(pData); 1514 doNotFree = 1; 1515 break; 1516 } 1517 1518 case ASYNC_UNLOCK: { 1519 AsyncFileData *pData = p->pFileData; 1520 int eLock = p->nByte; 1521 pthread_mutex_lock(&async.lockMutex); 1522 pData->lock.eAsyncLock = MIN( 1523 pData->lock.eAsyncLock, MAX(pData->lock.eLock, eLock) 1524 ); 1525 assert(pData->lock.eAsyncLock>=pData->lock.eLock); 1526 rc = getFileLock(pData->pLock); 1527 pthread_mutex_unlock(&async.lockMutex); 1528 break; 1529 } 1530 1531 case ASYNC_DELETE: 1532 ASYNC_TRACE(("DELETE %s\n", p->zBuf)); 1533 rc = pVfs->xDelete(pVfs, p->zBuf, (int)p->iOffset); 1534 break; 1535 1536 case ASYNC_OPENEXCLUSIVE: { 1537 int flags = (int)p->iOffset; 1538 AsyncFileData *pData = p->pFileData; 1539 ASYNC_TRACE(("OPEN %s flags=%d\n", p->zBuf, (int)p->iOffset)); 1540 assert(pData->pBaseRead->pMethods==0 && pData->pBaseWrite->pMethods==0); 1541 rc = pVfs->xOpen(pVfs, pData->zName, pData->pBaseRead, flags, 0); 1542 assert( holdingMutex==0 ); 1543 pthread_mutex_lock(&async.queueMutex); 1544 holdingMutex = 1; 1545 break; 1546 } 1547 1548 default: assert(!"Illegal value for AsyncWrite.op"); 1549 } 1550 1551 /* If we didn't hang on to the mutex during the IO op, obtain it now 1552 ** so that the AsyncWrite structure can be safely removed from the 1553 ** global write-op queue. 1554 */ 1555 if( !holdingMutex ){ 1556 pthread_mutex_lock(&async.queueMutex); 1557 holdingMutex = 1; 1558 } 1559 /* ASYNC_TRACE(("UNLINK %p\n", p)); */ 1560 if( p==async.pQueueLast ){ 1561 async.pQueueLast = 0; 1562 } 1563 if( !doNotFree ){ 1564 assert_mutex_is_held(&async.queueMutex); 1565 async.pQueueFirst = p->pNext; 1566 sqlite3_free(p); 1567 } 1568 assert( holdingMutex ); 1569 1570 /* An IO error has occurred. We cannot report the error back to the 1571 ** connection that requested the I/O since the error happened 1572 ** asynchronously. The connection has already moved on. There 1573 ** really is nobody to report the error to. 1574 ** 1575 ** The file for which the error occurred may have been a database or 1576 ** journal file. Regardless, none of the currently queued operations 1577 ** associated with the same database should now be performed. Nor should 1578 ** any subsequently requested IO on either a database or journal file 1579 ** handle for the same database be accepted until the main database 1580 ** file handle has been closed and reopened. 1581 ** 1582 ** Furthermore, no further IO should be queued or performed on any file 1583 ** handle associated with a database that may have been part of a 1584 ** multi-file transaction that included the database associated with 1585 ** the IO error (i.e. a database ATTACHed to the same handle at some 1586 ** point in time). 1587 */ 1588 if( rc!=SQLITE_OK ){ 1589 async.ioError = rc; 1590 } 1591 1592 if( async.ioError && !async.pQueueFirst ){ 1593 pthread_mutex_lock(&async.lockMutex); 1594 if( 0==async.pLock ){ 1595 async.ioError = SQLITE_OK; 1596 } 1597 pthread_mutex_unlock(&async.lockMutex); 1598 } 1599 1600 /* Drop the queue mutex before continuing to the next write operation 1601 ** in order to give other threads a chance to work with the write queue. 1602 */ 1603 if( !async.pQueueFirst || !async.ioError ){ 1604 pthread_mutex_unlock(&async.queueMutex); 1605 holdingMutex = 0; 1606 if( async.ioDelay>0 ){ 1607 pVfs->xSleep(pVfs, async.ioDelay); 1608 }else{ 1609 sched_yield(); 1610 } 1611 } 1612 } 1613 1614 pthread_mutex_unlock(&async.writerMutex); 1615 return 0; 1616 } 1617 1618 /************************************************************************** 1619 ** The remaining code defines a Tcl interface for testing the asynchronous 1620 ** IO implementation in this file. 1621 ** 1622 ** To adapt the code to a non-TCL environment, delete or comment out 1623 ** the code that follows. 1624 */ 1625 1626 /* 1627 ** sqlite3async_enable ?YES/NO? 1628 ** 1629 ** Enable or disable the asynchronous I/O backend. This command is 1630 ** not thread-safe. Do not call it while any database connections 1631 ** are open. 1632 */ 1633 static int testAsyncEnable( 1634 void * clientData, 1635 Tcl_Interp *interp, 1636 int objc, 1637 Tcl_Obj *CONST objv[] 1638 ){ 1639 if( objc!=1 && objc!=2 ){ 1640 Tcl_WrongNumArgs(interp, 1, objv, "?YES/NO?"); 1641 return TCL_ERROR; 1642 } 1643 if( objc==1 ){ 1644 Tcl_SetObjResult(interp, Tcl_NewBooleanObj(async_vfs.pAppData!=0)); 1645 }else{ 1646 int en; 1647 if( Tcl_GetBooleanFromObj(interp, objv[1], &en) ) return TCL_ERROR; 1648 asyncEnable(en); 1649 } 1650 return TCL_OK; 1651 } 1652 1653 /* 1654 ** sqlite3async_halt "now"|"idle"|"never" 1655 ** 1656 ** Set the conditions at which the writer thread will halt. 1657 */ 1658 static int testAsyncHalt( 1659 void * clientData, 1660 Tcl_Interp *interp, 1661 int objc, 1662 Tcl_Obj *CONST objv[] 1663 ){ 1664 const char *zCond; 1665 if( objc!=2 ){ 1666 Tcl_WrongNumArgs(interp, 1, objv, "\"now\"|\"idle\"|\"never\""); 1667 return TCL_ERROR; 1668 } 1669 zCond = Tcl_GetString(objv[1]); 1670 if( strcmp(zCond, "now")==0 ){ 1671 async.writerHaltNow = 1; 1672 pthread_cond_broadcast(&async.queueSignal); 1673 }else if( strcmp(zCond, "idle")==0 ){ 1674 async.writerHaltWhenIdle = 1; 1675 async.writerHaltNow = 0; 1676 pthread_cond_broadcast(&async.queueSignal); 1677 }else if( strcmp(zCond, "never")==0 ){ 1678 async.writerHaltWhenIdle = 0; 1679 async.writerHaltNow = 0; 1680 }else{ 1681 Tcl_AppendResult(interp, 1682 "should be one of: \"now\", \"idle\", or \"never\"", (char*)0); 1683 return TCL_ERROR; 1684 } 1685 return TCL_OK; 1686 } 1687 1688 /* 1689 ** sqlite3async_delay ?MS? 1690 ** 1691 ** Query or set the number of milliseconds of delay in the writer 1692 ** thread after each write operation. The default is 0. By increasing 1693 ** the memory delay we can simulate the effect of slow disk I/O. 1694 */ 1695 static int testAsyncDelay( 1696 void * clientData, 1697 Tcl_Interp *interp, 1698 int objc, 1699 Tcl_Obj *CONST objv[] 1700 ){ 1701 if( objc!=1 && objc!=2 ){ 1702 Tcl_WrongNumArgs(interp, 1, objv, "?MS?"); 1703 return TCL_ERROR; 1704 } 1705 if( objc==1 ){ 1706 Tcl_SetObjResult(interp, Tcl_NewIntObj(async.ioDelay)); 1707 }else{ 1708 int ioDelay; 1709 if( Tcl_GetIntFromObj(interp, objv[1], &ioDelay) ) return TCL_ERROR; 1710 async.ioDelay = ioDelay; 1711 } 1712 return TCL_OK; 1713 } 1714 1715 /* 1716 ** sqlite3async_start 1717 ** 1718 ** Start a new writer thread. 1719 */ 1720 static int testAsyncStart( 1721 void * clientData, 1722 Tcl_Interp *interp, 1723 int objc, 1724 Tcl_Obj *CONST objv[] 1725 ){ 1726 pthread_t x; 1727 int rc; 1728 volatile int isStarted = 0; 1729 rc = pthread_create(&x, 0, asyncWriterThread, (void *)&isStarted); 1730 if( rc ){ 1731 Tcl_AppendResult(interp, "failed to create the thread", 0); 1732 return TCL_ERROR; 1733 } 1734 pthread_detach(x); 1735 while( isStarted==0 ){ 1736 sched_yield(); 1737 } 1738 return TCL_OK; 1739 } 1740 1741 /* 1742 ** sqlite3async_wait 1743 ** 1744 ** Wait for the current writer thread to terminate. 1745 ** 1746 ** If the current writer thread is set to run forever then this 1747 ** command would block forever. To prevent that, an error is returned. 1748 */ 1749 static int testAsyncWait( 1750 void * clientData, 1751 Tcl_Interp *interp, 1752 int objc, 1753 Tcl_Obj *CONST objv[] 1754 ){ 1755 int cnt = 10; 1756 if( async.writerHaltNow==0 && async.writerHaltWhenIdle==0 ){ 1757 Tcl_AppendResult(interp, "would block forever", (char*)0); 1758 return TCL_ERROR; 1759 } 1760 1761 while( cnt-- && !pthread_mutex_trylock(&async.writerMutex) ){ 1762 pthread_mutex_unlock(&async.writerMutex); 1763 sched_yield(); 1764 } 1765 if( cnt>=0 ){ 1766 ASYNC_TRACE(("WAIT\n")); 1767 pthread_mutex_lock(&async.queueMutex); 1768 pthread_cond_broadcast(&async.queueSignal); 1769 pthread_mutex_unlock(&async.queueMutex); 1770 pthread_mutex_lock(&async.writerMutex); 1771 pthread_mutex_unlock(&async.writerMutex); 1772 }else{ 1773 ASYNC_TRACE(("NO-WAIT\n")); 1774 } 1775 return TCL_OK; 1776 } 1777 1778 1779 #endif /* SQLITE_OS_UNIX and SQLITE_THREADSAFE */ 1780 1781 /* 1782 ** This routine registers the custom TCL commands defined in this 1783 ** module. This should be the only procedure visible from outside 1784 ** of this module. 1785 */ 1786 int Sqlitetestasync_Init(Tcl_Interp *interp){ 1787 #if SQLITE_OS_UNIX && SQLITE_THREADSAFE 1788 Tcl_CreateObjCommand(interp,"sqlite3async_enable",testAsyncEnable,0,0); 1789 Tcl_CreateObjCommand(interp,"sqlite3async_halt",testAsyncHalt,0,0); 1790 Tcl_CreateObjCommand(interp,"sqlite3async_delay",testAsyncDelay,0,0); 1791 Tcl_CreateObjCommand(interp,"sqlite3async_start",testAsyncStart,0,0); 1792 Tcl_CreateObjCommand(interp,"sqlite3async_wait",testAsyncWait,0,0); 1793 Tcl_LinkVar(interp, "sqlite3async_trace", 1794 (char*)&sqlite3async_trace, TCL_LINK_INT); 1795 #endif /* SQLITE_OS_UNIX and SQLITE_THREADSAFE */ 1796 return TCL_OK; 1797 } 1798