Add proof-of-concept PriorityWorker

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,27 @@ static void FinalizeDatabase (napi_env env, void* data, void* hint) {
   }
 }
 
+/**
+ * Base worker class for doing async work that must delay closing the database.
+ */
+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 +858,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,24 +881,24 @@ 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;
   }
 
-  virtual ~PutWorker () {
+  ~PutWorker () final {
     DisposeSliceBuffer(key_);
     DisposeSliceBuffer(value_);
   }
 
-  virtual void DoExecute () {
+  void DoExecute () final {
     SetStatus(database_->Put(options_, key_, value_));
   }
 

test/segfault-test.js

@@ -2,28 +2,30 @@ 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) {
-  t.plan(3)
-
-  const db = testCommon.factory()
-
-  db.open(function (err) {
-    t.ifError(err, 'no open error')
-
-    // The "sync" option seems to be a reliable way to trigger a segfault,
-    // but is not necessarily the cause of that segfault. More likely, it
-    // exposes a race condition that's already there.
-    db.put('foo', 'bar', { sync: true }, function (err) {
-      // We never get here, due to segfault.
-      t.ifError(err, 'no put error')
-    })
-
-    db.close(function (err) {
-      // We never get here, due to segfault.
-      t.ifError(err, 'no close error')
+for (let i = 0; i < 100; i++) {
+  test(`close() does not segfault if there is a pending write ${i}`, function (t) {
+    t.plan(3)
+
+    const db = testCommon.factory()
+
+    db.open(function (err) {
+      t.ifError(err, 'no open error')
+
+      // The "sync" option seems to be a reliable way to trigger a segfault,
+      // but is not necessarily the cause of that segfault. More likely, it
+      // exposes a race condition that's already there.
+      db.put('foo', 'bar', { sync: true }, function (err) {
+        // We never get here, due to segfault.
+        t.ifError(err, 'no put error')
+      })
+
+      db.close(function (err) {
+        // We never get here, due to segfault.
+        t.ifError(err, 'no close error')
+      })
     })
   })
-})
+}
 
 // See https://github.com/Level/leveldown/issues/134
 test('iterator() does not segfault if db is not open', function (t) {