Add PriorityWorker to defer closing until put is done

Author: vweevers

binding.cc

@@ -290,7 +290,7 @@ struct BaseWorker {
     delete self;
   }
 
-  void DoComplete () {
+  virtual void DoComplete () {
     if (status_.ok()) {
       return HandleOKCallback();
     }
@@ -309,7 +309,7 @@ struct BaseWorker {
     CallFunction(env_, callback, 1, &argv);
   }
 
-  void Queue () {
+  virtual void Queue () {
     napi_queue_async_work(env_, asyncWork_);
   }
 
@@ -333,6 +333,7 @@ struct Database {
       blockCache_(NULL),
       filterPolicy_(leveldb::NewBloomFilterPolicy(10)),
       currentIteratorId_(0),
+      priorityWork_(0),
       pendingCloseWorker_(NULL) {}
 
   ~Database () {
@@ -407,24 +408,39 @@ struct Database {
 
   void AttachIterator (uint32_t id, Iterator* iterator) {
     iterators_[id] = iterator;
+    IncrementPriorityWork();
   }
 
   void DetachIterator (uint32_t id) {
     iterators_.erase(id);
+    DecrementPriorityWork();
+  }
+
+  void IncrementPriorityWork () {
+    ++priorityWork_;
+  }
 
-    if (iterators_.empty() && pendingCloseWorker_ != NULL) {
+  void DecrementPriorityWork () {
+    if (--priorityWork_ == 0 && pendingCloseWorker_ != NULL) {
       pendingCloseWorker_->Queue();
       pendingCloseWorker_ = NULL;
     }
   }
 
+  bool HasPriorityWork () {
+    return priorityWork_ > 0;
+  }
+
   napi_env env_;
   leveldb::DB* db_;
   leveldb::Cache* blockCache_;
   const leveldb::FilterPolicy* filterPolicy_;
   uint32_t currentIteratorId_;
   BaseWorker *pendingCloseWorker_;
   std::map< uint32_t, Iterator * > iterators_;
+
+private:
+  uint32_t priorityWork_;
 };
 
 /**
@@ -436,6 +452,28 @@ static void FinalizeDatabase (napi_env env, void* data, void* hint) {
   }
 }
 
+/**
+ * Base worker class for doing async work that defers closing the database.
+ * @TODO Use this for Get-, Del-, Batch-, BatchWrite-, ApproximateSize- and CompactRangeWorker too.
+ */
+struct PriorityWorker : public BaseWorker {
+  PriorityWorker (napi_env env, Database* database, napi_value callback, const char* resourceName)
+    : BaseWorker(env, database, callback, resourceName) {
+  }
+
+  virtual ~PriorityWorker () {}
+
+  void Queue () final {
+    database_->IncrementPriorityWork();
+    BaseWorker::Queue();
+  }
+
+  void DoComplete () final {
+    database_->DecrementPriorityWork();
+    BaseWorker::DoComplete();
+  }
+};
+
 /**
  * Owns a leveldb iterator.
  */
@@ -821,7 +859,7 @@ NAPI_METHOD(db_close) {
   napi_value callback = argv[1];
   CloseWorker* worker = new CloseWorker(env, database, callback);
 
-  if (database->iterators_.empty()) {
+  if (!database->HasPriorityWork()) {
     worker->Queue();
     NAPI_RETURN_UNDEFINED();
   }
@@ -844,14 +882,14 @@ NAPI_METHOD(db_close) {
 /**
  * Worker class for putting key/value to the database
  */
-struct PutWorker : public BaseWorker {
+struct PutWorker : public PriorityWorker {
   PutWorker (napi_env env,
              Database* database,
              napi_value callback,
              leveldb::Slice key,
              leveldb::Slice value,
              bool sync)
-    : BaseWorker(env, database, callback, "leveldown.db.put"),
+    : PriorityWorker(env, database, callback, "leveldown.db.put"),
       key_(key), value_(value) {
     options_.sync = sync;
   }

test/segfault-test.js

@@ -2,7 +2,7 @@ const test = require('tape')
 const testCommon = require('./common')
 
 // Open issue: https://github.com/Level/leveldown/issues/157
-test.skip('close() does not segfault if there is a pending write', function (t) {
+test('close() does not segfault if there is a pending write', function (t) {
   t.plan(3)
 
   const db = testCommon.factory()