name polishing for consistency

mlin · mlin · commit 828ef5357cee · 2020-08-11T23:27:35.000-10:00
diff --git a/README.md b/README.md
@@ -117,14 +117,15 @@ Some parameters are controlled from the file URI's query string opening the data
 |   | URI query parameters | PRAGMA |
 | -- | -- | -- |
 | writing/compression | <ul><li>level</li><li>threads</li><li>outer_page_size</li><li>outer_unsafe</li></ul> | <ul><li>page_size</li><li>auto_vacuum</li><li>journal_mode</li></ul> |
-| reading/decompression | <ul><li>outer_cache_size</li></ul> | <ul><li>cache_size</li></ul> |
+| reading/decompression | <ul><li>outer_cache_size</li><li>noprefetch</li></ul> | <ul><li>cache_size</li></ul> |
 
 * **&level=3**: Zstandard compression level for newly written pages (-7 to 22)
-* **&threads=1**: background compression threads (-1 to match host processors)
+* **&threads=1**: thread budget for page compression & prefetching (-1 to match host processors)
 * **&outer_page_size=4096**: page size for the newly-created outer database; suggest doubling the (inner) page_size, to reduce space overhead from packing the compressed inner pages
 * **&outer_unsafe=false**: set true to speed up bulk load by disabling transaction safety for outer database (app crash easily causes corruption)
 
 * **&outer_cache_size=-2000**: page cache size for outer database, in [PRAGMA cache_size](https://www.sqlite.org/pragma.html#pragma_cache_size) units. Limited effect if on SSD.
+* **&noprefetch=false**: set true to disable background prefetching/decompression even if threads>1. Prefetching is counterproductive if page size is too small or CPU cycles are scarce.
 
 * **PRAGMA page_size=4096**: uncompressed [page size](https://www.sqlite.org/pragma.html#pragma_page_size) for the newly-created inner database. Larger pages are more compressible, but increase [read/write amplification](http://smalldatum.blogspot.com/2015/11/read-write-space-amplification-pick-2_23.html). YMMV but 8 or 16 KiB have been working well for us.
 * **PRAGMA auto_vacuum=NONE**: set to FULL or INCREMENTAL on a newly-created database if you expect its size to fluctuate over time, so that the file will [shrink to fit](https://www.sqlite.org/pragma.html#pragma_auto_vacuum). (The outer database auto-vacuums when it's closed.)
diff --git a/src/SQLiteNestedVFS.h b/src/SQLiteNestedVFS.h
@@ -8,7 +8,7 @@
  * is useless, but subclasses can easily layer in compression and/or encryption codecs.
  *
  * Uses thread pools so that (i) page encoding and writeback occur on background threads, and (ii)
- * page reads and decoding can be "prefetched" in the background during detected sequential scans.
+ * pages can be prefetched and decoded on background threads during detected sequential scans.
  * These schemes are each a bit complex but at least they're separate: when asked to read a page,
  * we first wait for any outstanding writes to finish, and vice-versa.
  */
@@ -120,7 +120,7 @@ class InnerDatabaseFile : public SQLiteVFS::File {
 
     int FileSize(sqlite3_int64 *pSize) override {
         try {
-            FinishUpserts();
+            UpsertBarrier();
             *pSize = DetectPageSize() * DetectPageCount();
             return SQLITE_OK;
         } catch (std::exception &exn) {
@@ -144,7 +144,7 @@ class InnerDatabaseFile : public SQLiteVFS::File {
     // 5. Little thread synchronization overhead is affordable, as Zstandard decompression of a
     //    SQLite page (4-64 KiB) takes only 1-10 microseconds -- about 10x time to memcpy.
     // Subclasses may in turn subclass this to add their own logic to SeekCursor and DecodePage.
-    struct PageFetchJob {
+    struct FetchJob {
         enum State { NEW, QUEUE, WIP, DONE };
         // NEW = idle, QUEUE = ready & waiting, WIP = work in progress, DONE = complete or error
         // QUEUE=>WIP is the only contentious transition; background threads and the main thread
@@ -176,15 +176,15 @@ class InnerDatabaseFile : public SQLiteVFS::File {
         std::chrono::nanoseconds t_seek = std::chrono::nanoseconds::zero(),
                                  t_decode = std::chrono::nanoseconds::zero();
 
-        PageFetchJob(InnerDatabaseFile &that)
+        FetchJob(InnerDatabaseFile &that)
             : cursor(*(that.outer_db_), "SELECT pageno, data, meta1, meta2 FROM " +
                                             that.inner_db_pages_table_ +
                                             " WHERE pageno >= ? ORDER BY pageno"),
               page_size(that.page_size_) {
             PutState(State::NEW);
         }
 
-        virtual ~PageFetchJob() {}
+        virtual ~FetchJob() {}
 
         inline void *EffectiveDest() noexcept {
             return dest ? dest : (decodebuf.resize(page_size), decodebuf.data());
@@ -286,12 +286,12 @@ class InnerDatabaseFile : public SQLiteVFS::File {
     };
 
     // Override me!
-    virtual std::unique_ptr<PageFetchJob> NewPageFetchJob() {
-        return std::unique_ptr<PageFetchJob>(new PageFetchJob(*this));
+    virtual std::unique_ptr<FetchJob> NewFetchJob() {
+        return std::unique_ptr<FetchJob>(new FetchJob(*this));
     }
 
     const size_t MAX_FETCH_CURSORS = 4;
-    std::vector<std::unique_ptr<PageFetchJob>> fetch_jobs_;
+    std::vector<std::unique_ptr<FetchJob>> fetch_jobs_;
     ThreadPool fetch_thread_pool_;
     std::mutex seek_lock_; // serializes outer db interactions among fetch background threads
     std::atomic<bool> seek_interrupt_; // broadcast that main thread wants seek_lock_
@@ -300,15 +300,15 @@ class InnerDatabaseFile : public SQLiteVFS::File {
     sqlite3_int64 longest_read_ = 0;
 
     void *BackgroundFetchJob(void *ctx) noexcept {
-        PageFetchJob *job = (PageFetchJob *)ctx;
+        FetchJob *job = (FetchJob *)ctx;
         std::unique_lock<std::mutex> seek_lock(seek_lock_);
         while (seek_interrupt_.load(std::memory_order_relaxed)) {
             // yield to main thread
             seek_lock.unlock();
             std::this_thread::yield();
             seek_lock.lock();
         }
-        if (!job->TransitionState(PageFetchJob::State::QUEUE, PageFetchJob::State::WIP)) {
+        if (!job->TransitionState(FetchJob::State::QUEUE, FetchJob::State::WIP)) {
             return nullptr; // they took our job!!!!
         }
         job->Execute(&seek_lock);
@@ -324,16 +324,15 @@ class InnerDatabaseFile : public SQLiteVFS::File {
         bool can_prefetch = fetch_thread_pool_.MaxThreads() > 1;
 
         // Is there already a background job to prefetch the desired page?
-        PageFetchJob *job = nullptr;
+        FetchJob *job = nullptr;
         bool foreground = false;
         if (can_prefetch) {
             for (auto job_i = fetch_jobs_.begin(); job_i != fetch_jobs_.end(); job_i++) {
                 if ((*job_i)->pageno == pageno) {
-                    assert((*job_i)->GetState() > PageFetchJob::State::NEW);
+                    assert((*job_i)->GetState() > FetchJob::State::NEW);
                     job = job_i->get();
                     // If yes & it hasn't yet started, race to run it here in the foreground
-                    foreground =
-                        job->TransitionState(PageFetchJob::State::QUEUE, PageFetchJob::State::WIP);
+                    foreground = job->TransitionState(FetchJob::State::QUEUE, FetchJob::State::WIP);
                     break;
                 }
             }
@@ -343,7 +342,7 @@ class InnerDatabaseFile : public SQLiteVFS::File {
         if (!job) {
             for (auto job_i = fetch_jobs_.begin(); job_i != fetch_jobs_.end(); job_i++) {
                 auto st = (*job_i)->GetState();
-                if (st == PageFetchJob::State::NEW || st == PageFetchJob::State::DONE) {
+                if (st == FetchJob::State::NEW || st == FetchJob::State::DONE) {
                     if ((*job_i)->cursor_pageno + 1 == pageno) {
                         job = job_i->get();
                         break;
@@ -357,17 +356,17 @@ class InnerDatabaseFile : public SQLiteVFS::File {
             if (!job || (fetch_jobs_.size() < MAX_FETCH_CURSORS &&
                          job->cursor_pageno + 1 != pageno && job->cursor_pageno)) {
                 assert(fetch_jobs_.size() < MAX_FETCH_CURSORS);
-                fetch_jobs_.push_back(NewPageFetchJob());
+                fetch_jobs_.push_back(NewFetchJob());
                 job = fetch_jobs_.back().get();
             }
 #ifndef NDEBUG
-            if (job->GetState() == PageFetchJob::State::DONE && job->cursor_pageno) {
+            if (job->GetState() == FetchJob::State::DONE && job->cursor_pageno) {
                 ++prefetch_wasted_;
             }
 #endif
             job->Renew();
             job->pageno = pageno;
-            job->PutState(PageFetchJob::State::WIP);
+            job->PutState(FetchJob::State::WIP);
             foreground = true;
         }
 
@@ -386,10 +385,10 @@ class InnerDatabaseFile : public SQLiteVFS::File {
             } else {
                 job->Execute();
             }
-            assert(job->GetState() == PageFetchJob::State::DONE);
+            assert(job->GetState() == FetchJob::State::DONE);
         } else {
             // Semi-busy-wait for background job to finish
-            while (job->GetState() < PageFetchJob::State::DONE) {
+            while (job->GetState() < FetchJob::State::DONE) {
                 std::this_thread::yield();
             }
 #ifndef NDEBUG
@@ -400,6 +399,7 @@ class InnerDatabaseFile : public SQLiteVFS::File {
         if (!job->errmsg.empty()) {
             _DBG << job->errmsg << _EOL;
             job->Renew();
+            PrefetchBarrier();
             throw SQLite::Exception(job->errmsg, SQLITE_IOERR_READ);
         }
 
@@ -422,15 +422,15 @@ class InnerDatabaseFile : public SQLiteVFS::File {
             for (auto job_i = fetch_jobs_.begin(); job_i != fetch_jobs_.end(); job_i++) {
                 if (job_i->get() != job) {
                     auto st = (*job_i)->GetState();
-                    if (st == PageFetchJob::State::QUEUE || st == PageFetchJob::State::WIP) {
+                    if (st == FetchJob::State::QUEUE || st == FetchJob::State::WIP) {
                         ++active_jobs;
                     }
                 }
             }
             // always leave at least one slot free for a non-sequential read to use
             if (active_jobs + 2 <= fetch_thread_pool_.MaxThreads()) {
                 job->pageno = pageno_hint;
-                job->PutState(PageFetchJob::State::QUEUE);
+                job->PutState(FetchJob::State::QUEUE);
                 fetch_thread_pool_.Enqueue(
                     job, [this](void *job) { return this->BackgroundFetchJob(job); }, nullptr);
             } else {
@@ -467,7 +467,7 @@ class InnerDatabaseFile : public SQLiteVFS::File {
             if (iAmt == 0) {
                 return SQLITE_OK;
             }
-            FinishUpserts();
+            UpsertBarrier();
             if (!DetectPageSize()) { // file is empty
                 memset(zBuf, 0, iAmt);
                 return iAmt > 0 ? SQLITE_IOERR_SHORT_READ : SQLITE_OK;
@@ -532,7 +532,7 @@ class InnerDatabaseFile : public SQLiteVFS::File {
                 seek_interrupt_.store(false, std::memory_order_relaxed);
             }
             for (auto &job : fetch_jobs_) {
-                if (job->TransitionState(PageFetchJob::State::QUEUE, PageFetchJob::State::NEW)) {
+                if (job->TransitionState(FetchJob::State::QUEUE, FetchJob::State::NEW)) {
                     job->Renew();
                 }
             }
@@ -542,8 +542,8 @@ class InnerDatabaseFile : public SQLiteVFS::File {
         }
         // wipe all prefetch cursors
         for (auto &job : fetch_jobs_) {
-            assert(job->GetState() == PageFetchJob::State::NEW ||
-                   job->GetState() == PageFetchJob::State::DONE);
+            assert(job->GetState() == FetchJob::State::NEW ||
+                   job->GetState() == FetchJob::State::DONE);
             job->Renew();
             job->ResetCursor();
         }
@@ -614,7 +614,7 @@ class InnerDatabaseFile : public SQLiteVFS::File {
     std::vector<std::unique_ptr<EncodeJob>> encode_job_pool_;
     std::mutex encode_job_pool_mutex_;
 
-    ThreadPool thread_pool_;
+    ThreadPool upsert_thread_pool_;
     std::string upsert_errmsg_;
 
     // enqueue page encoding+upsert on thread pool
@@ -637,7 +637,7 @@ class InnerDatabaseFile : public SQLiteVFS::File {
             assert(job->insert ? pageno == page_count_ + 1 : pageno <= page_count_);
             page_count_ += job->insert ? 1 : 0;
             // use ThreadPool to run encoding jobs in parallel and upsert jobs serially
-            thread_pool_.Enqueue(
+            upsert_thread_pool_.Enqueue(
                 job.release(),
                 [](void *job) noexcept {
                     ((EncodeJob *)job)->Execute();
@@ -651,8 +651,8 @@ class InnerDatabaseFile : public SQLiteVFS::File {
             throw;
         }
 
-        if (thread_pool_.MaxThreads() == 1) {
-            FinishUpserts();
+        if (upsert_thread_pool_.MaxThreads() == 1) {
+            UpsertBarrier();
         }
     }
 
@@ -695,8 +695,8 @@ class InnerDatabaseFile : public SQLiteVFS::File {
     }
 
     // wait for background upserts to complete + raise any error message
-    void FinishUpserts(bool ignore_error = false) {
-        thread_pool_.Barrier();
+    void UpsertBarrier(bool ignore_error = false) {
+        upsert_thread_pool_.Barrier();
         if (!ignore_error && !upsert_errmsg_.empty()) {
             throw SQLite::Exception(upsert_errmsg_, SQLITE_IOERR_WRITE);
         }
@@ -747,7 +747,7 @@ class InnerDatabaseFile : public SQLiteVFS::File {
             return SQLITE_OK;
         } catch (std::exception &exn) {
             _DBG << exn.what() << _EOL;
-            FinishUpserts(true);
+            UpsertBarrier(true);
             page_count_ = 0; // to be redetected
             return SQLITE_IOERR_WRITE;
         }
@@ -757,7 +757,7 @@ class InnerDatabaseFile : public SQLiteVFS::File {
         assert(!read_only_);
         try {
             PrefetchBarrier();
-            FinishUpserts();
+            UpsertBarrier();
             if (!DetectPageSize()) {
                 return size == 0 ? SQLITE_OK : SQLITE_IOERR_TRUNCATE;
             }
@@ -790,7 +790,7 @@ class InnerDatabaseFile : public SQLiteVFS::File {
     int Sync(int flags) override {
         assert(!read_only_);
         try {
-            FinishUpserts();
+            UpsertBarrier();
         } catch (std::exception &exn) {
             _DBG << exn.what() << _EOL;
             page_count_ = 0; // to be redetected
@@ -861,7 +861,7 @@ class InnerDatabaseFile : public SQLiteVFS::File {
             PrefetchBarrier();
             if (!read_only_) {
                 int rc;
-                if ((rc = Sync(0)) != SQLITE_OK) { // includes FinishUpserts
+                if ((rc = Sync(0)) != SQLITE_OK) { // includes UpsertBarrier
                     return rc;
                 }
                 outer_db_->exec("PRAGMA incremental_vacuum");
@@ -925,7 +925,7 @@ class InnerDatabaseFile : public SQLiteVFS::File {
           // MAX(pageno) instead of COUNT(pageno) because the latter would trigger table scan
           select_page_count_(*outer_db_,
                              "SELECT IFNULL(MAX(pageno), 0) FROM " + inner_db_pages_table_),
-          thread_pool_(threads, threads * 3),
+          upsert_thread_pool_(threads, threads * 3),
           fetch_thread_pool_(std::min(noprefetch ? 1 : threads, MAX_FETCH_CURSORS),
                              MAX_FETCH_CURSORS),
           seek_interrupt_(false) {
diff --git a/src/zstd_vfs.h b/src/zstd_vfs.h
@@ -110,15 +110,15 @@ class ZstdInnerDatabaseFile : public SQLiteNested::InnerDatabaseFile {
         return dict_cache_[dict_id];
     }
 
-    struct ZstdPageFetchJob : public super::PageFetchJob {
+    struct ZstdFetchJob : public super::FetchJob {
         ZSTD_DCtx *dctx = nullptr;
         const ZSTD_DDict *ddict = nullptr;
         bool plain = false; // uncompressed page
         ZstdInnerDatabaseFile *that;
 
-        ZstdPageFetchJob(ZstdInnerDatabaseFile &that_) : super::PageFetchJob(that_), that(&that_) {}
+        ZstdFetchJob(ZstdInnerDatabaseFile &that_) : super::FetchJob(that_), that(&that_) {}
 
-        ~ZstdPageFetchJob() {
+        ~ZstdFetchJob() {
             if (dctx) {
                 ZSTD_freeDCtx(dctx);
             }
@@ -128,7 +128,7 @@ class ZstdInnerDatabaseFile : public SQLiteNested::InnerDatabaseFile {
         // dictionary ready for use. This has to be done in this serialized method since it may
         // need to load it from the outer db.
         void SeekCursor() override {
-            super::PageFetchJob::SeekCursor();
+            super::FetchJob::SeekCursor();
 #ifndef NDEBUG
             auto t0 = std::chrono::high_resolution_clock::now();
 #endif
@@ -157,7 +157,7 @@ class ZstdInnerDatabaseFile : public SQLiteNested::InnerDatabaseFile {
             auto t0 = std::chrono::high_resolution_clock::now();
 #endif
             if (plain) { // uncompressed page
-                return super::PageFetchJob::DecodePage();
+                return super::FetchJob::DecodePage();
             }
             if (!dctx) {
                 dctx = ZSTD_createDCtx();
@@ -181,8 +181,8 @@ class ZstdInnerDatabaseFile : public SQLiteNested::InnerDatabaseFile {
         }
     };
 
-    std::unique_ptr<super::PageFetchJob> NewPageFetchJob() override {
-        return std::unique_ptr<super::PageFetchJob>(new ZstdPageFetchJob(*this));
+    std::unique_ptr<super::FetchJob> NewFetchJob() override {
+        return std::unique_ptr<super::FetchJob>(new ZstdFetchJob(*this));
     }
 
     // dict currently used for compression of new pages
@@ -214,7 +214,7 @@ class ZstdInnerDatabaseFile : public SQLiteNested::InnerDatabaseFile {
 
         if (cur_dict_ < 0) {
             PrefetchBarrier();
-            FinishUpserts();
+            UpsertBarrier();
             // start off with the newest dict stored in the database, if any
             if (!last_dict_id_) {
                 last_dict_id_.reset(
@@ -235,7 +235,7 @@ class ZstdInnerDatabaseFile : public SQLiteNested::InnerDatabaseFile {
             std::max(page_count_, cur_dict_pages_written_) >= 3 * cur_dict_page_count_) {
             // time to create a new dict...
             PrefetchBarrier();
-            FinishUpserts();
+            UpsertBarrier();
             auto t0 = std::chrono::high_resolution_clock::now();
 
             // read random pages
