Fix Segmentation Fault Crash (parcel-bundler#40)

* create repro

* debug config

* add locking mechanism

* tweaks

* cleanup

* Delete crash.log

* Delete tasks.json

* up the stress

* kind of fixed it

* remove segfault-handler for alpine

.gitignore

@@ -4,4 +4,6 @@ prebuilds/
 node_modules/
 token.txt
 .DS_Store
-.idea
+.idea
+stress-test
+crash.log

.vscode/launch.json

@@ -0,0 +1,14 @@
+{
+  "version": "0.2.0",
+  "configurations": [
+    {
+      "type": "lldb",
+      "request": "launch",
+      "name": "Launch Program",
+      "preLaunchTask": "npm: rebuild",
+      "program": "node",
+      "args": ["${workspaceFolder}/stress-test.js"],
+      "cwd": "${workspaceFolder}"
+    }
+  ]
+}

package.json

@@ -27,8 +27,9 @@
     "prebuild": "prebuildify --napi --strip --tag-libc -t 10.0.0",
     "format": "prettier --write \"./**/*.{js,json,md}\"",
     "install": "node-gyp-build",
-    "rebuild": "node-gyp rebuild --debug",
-    "test": "mocha"
+    "rebuild": "node-gyp rebuild -j 8 --debug --verbose",
+    "test": "mocha",
+    "stress-test": "node ./stress-test.js"
   },
   "engines": {
     "node": ">= 10.0.0"
@@ -53,8 +54,8 @@
     "fs-extra": "^7.0.1",
     "husky": "^4.2.3",
     "mocha": "^6.0.2",
-    "prettier": "^1.19.1",
-    "prebuildify": "^4.0.0"
+    "prebuildify": "^4.0.0",
+    "prettier": "^1.19.1"
   },
   "binary": {
     "napi_versions": [

src/Event.hh

@@ -27,51 +27,57 @@ struct Event {
 class EventList {
 public:
   void create(std::string path) {
-    Event *event = update(path);
+    std::lock_guard<std::mutex> l(mMutex);
+    Event *event = internalUpdate(path);
     event->isCreated = true;
   }
 
   Event *update(std::string path) {
-    auto found = mEvents.find(path);
-    if (found == mEvents.end()) {
-      auto it = mEvents.emplace(path, Event(path));
-      return &it.first->second;
-    }
-
-    return &found->second;
+    std::lock_guard<std::mutex> l(mMutex);
+    return internalUpdate(path);
   }
 
   void remove(std::string path) {
-    Event *event = update(path);
+    std::lock_guard<std::mutex> l(mMutex);
+    Event *event = internalUpdate(path);
     if (event->isCreated) {
       mEvents.erase(path);
     } else {
       event->isDeleted = true;
     }
   }
 
-  void clear() {
-    mEvents.clear();
-  }
-
   size_t size() {
+    std::lock_guard<std::mutex> l(mMutex);
     return mEvents.size();
   }
 
-  Value toJS(const Env& env) {
-    EscapableHandleScope scope(env);
-    Array arr = Array::New(env, mEvents.size());
-    size_t i = 0;
-
-    for (auto it = mEvents.begin(); it != mEvents.end(); it++) {
-      arr.Set(i++, it->second.toJS(env));
+  std::vector<Event> getEvents() {
+    std::lock_guard<std::mutex> l(mMutex);
+    std::vector<Event> eventsCloneVector;
+    for(auto it = mEvents.begin(); it != mEvents.end(); ++it) {
+      eventsCloneVector.push_back(it->second);
     }
+    return eventsCloneVector;
+  }
 
-    return scope.Escape(arr);
+  void clear() {
+    std::lock_guard<std::mutex> l(mMutex);
+    mEvents.clear();
   }
 
 private:
+  mutable std::mutex mMutex;
   std::map<std::string, Event> mEvents;
+  Event *internalUpdate(std::string path) {
+    auto found = mEvents.find(path);
+    if (found == mEvents.end()) {
+      auto it = mEvents.emplace(path, Event(path));
+      return &it.first->second;
+    }
+
+    return &found->second;
+  }
 };
 
 #endif

src/Watcher.cc

@@ -37,7 +37,7 @@ void removeShared(Watcher *watcher) {
   }
 }
 
-Watcher::Watcher(std::string dir, std::unordered_set<std::string> ignore) 
+Watcher::Watcher(std::string dir, std::unordered_set<std::string> ignore)
   : mDir(dir),
     mIgnore(ignore),
     mWatched(false),
@@ -61,7 +61,7 @@ void Watcher::wait() {
 void Watcher::notify() {
   std::unique_lock<std::mutex> lk(mMutex);
   mCond.notify_all();
-  
+
   if (mCallbacks.size() > 0 && mEvents.size() > 0) {
     mDebounce->trigger();
   }
@@ -79,19 +79,36 @@ void Watcher::notifyError(std::exception &err) {
 }
 
 void Watcher::triggerCallbacks() {
+  std::lock_guard<std::mutex> l(mCallbackEventsMutex);
   if (mCallbacks.size() > 0 && (mEvents.size() > 0 || mError.size() > 0)) {
     if (mCallingCallbacks) {
       mCallbackSignal.wait();
       mCallbackSignal.reset();
     }
 
-    mCallbackEvents = mEvents;
+    mCallbackEvents = mEvents.getEvents();
     mEvents.clear();
 
     uv_async_send(mAsync);
   }
 }
 
+Value Watcher::callbackEventsToJS(const Env& env) {
+  std::lock_guard<std::mutex> l(mCallbackEventsMutex);
+  EscapableHandleScope scope(env);
+  Array arr = Array::New(env, mCallbackEvents.size());
+  size_t currentEventIndex = 0;
+  for (auto eventIterator = mCallbackEvents.begin(); eventIterator != mCallbackEvents.end(); eventIterator++) {
+    arr.Set(currentEventIndex++, eventIterator->toJS(env));
+  }
+  return scope.Escape(arr);
+}
+
+// TODO: Doesn't this need some kind of locking?
+void Watcher::clearCallbacks() {
+  mCallbacks.clear();
+}
+
 void Watcher::fireCallbacks(uv_async_t *handle) {
   Watcher *watcher = (Watcher *)handle->data;
   watcher->mCallingCallbacks = true;
@@ -101,7 +118,8 @@ void Watcher::fireCallbacks(uv_async_t *handle) {
     auto it = watcher->mCallbacksIterator;
     HandleScope scope(it->Env());
     auto err = watcher->mError.size() > 0 ? Error::New(it->Env(), watcher->mError).Value() : it->Env().Null();
-    auto events = watcher->mCallbackEvents.toJS(it->Env());
+    auto events = watcher->callbackEventsToJS(it->Env());
+
     it->MakeCallback(it->Env().Global(), std::initializer_list<napi_value>{err, events});
 
     // If the iterator was changed, then the callback trigged an unwatch.
@@ -115,7 +133,7 @@ void Watcher::fireCallbacks(uv_async_t *handle) {
   watcher->mCallingCallbacks = false;
 
   if (watcher->mError.size() > 0) {
-    watcher->mCallbacks.clear();
+    watcher->clearCallbacks();
   }
 
   if (watcher->mCallbacks.size() == 0) {
@@ -150,7 +168,7 @@ bool Watcher::unwatch(Function callback) {
       break;
     }
   }
-  
+
   if (removed && mCallbacks.size() == 0) {
     unref();
     return true;

src/Watcher.hh

@@ -39,16 +39,19 @@ struct Watcher {
 
 private:
   std::mutex mMutex;
+  std::mutex mCallbackEventsMutex;
   std::condition_variable mCond;
   uv_async_t *mAsync;
   std::set<FunctionReference> mCallbacks;
   std::set<FunctionReference>::iterator mCallbacksIterator;
   bool mCallingCallbacks;
-  EventList mCallbackEvents;
+  std::vector<Event> mCallbackEvents;
   std::shared_ptr<Debounce> mDebounce;
   Signal mCallbackSignal;
   std::string mError;
 
+  Value callbackEventsToJS(const Env& env);
+  void clearCallbacks();
   void triggerCallbacks();
   static void fireCallbacks(uv_async_t *handle);
   static void onClose(uv_handle_t *handle);

src/binding.cc

@@ -40,7 +40,7 @@ std::shared_ptr<Backend> getBackend(Env env, Value opts) {
 
 class WriteSnapshotRunner : public PromiseRunner {
 public:
-  WriteSnapshotRunner(Env env, Value dir, Value snap, Value opts) 
+  WriteSnapshotRunner(Env env, Value dir, Value snap, Value opts)
     : PromiseRunner(env),
       snapshotPath(std::string(snap.As<String>().Utf8Value().c_str())) {
     watcher = Watcher::getShared(
@@ -67,7 +67,7 @@ class WriteSnapshotRunner : public PromiseRunner {
 
 class GetEventsSinceRunner : public PromiseRunner {
 public:
-  GetEventsSinceRunner(Env env, Value dir, Value snap, Value opts) 
+  GetEventsSinceRunner(Env env, Value dir, Value snap, Value opts)
     : PromiseRunner(env),
       snapshotPath(std::string(snap.As<String>().Utf8Value().c_str())) {
     watcher = std::make_shared<Watcher>(
@@ -92,7 +92,13 @@ class GetEventsSinceRunner : public PromiseRunner {
   }
 
   Value getResult() override {
-    return watcher->mEvents.toJS(env);
+    std::vector<Event> events = watcher->mEvents.getEvents();
+    Array eventsArray = Array::New(env, events.size());
+    size_t i = 0;
+    for (auto it = events.begin(); it != events.end(); it++) {
+      eventsArray.Set(i++, it->toJS(env));
+    }
+    return eventsArray;
   }
 };
 
@@ -130,7 +136,7 @@ class SubscribeRunner : public PromiseRunner {
 public:
   SubscribeRunner(Env env, Value dir, Value fn, Value opts) : PromiseRunner(env) {
     watcher = Watcher::getShared(
-      std::string(dir.As<String>().Utf8Value().c_str()), 
+      std::string(dir.As<String>().Utf8Value().c_str()),
       getIgnore(env, opts)
     );
 
@@ -153,7 +159,7 @@ class UnsubscribeRunner : public PromiseRunner {
 public:
   UnsubscribeRunner(Env env, Value dir, Value fn, Value opts) : PromiseRunner(env) {
     watcher = Watcher::getShared(
-      std::string(dir.As<String>().Utf8Value().c_str()), 
+      std::string(dir.As<String>().Utf8Value().c_str()),
       getIgnore(env, opts)
     );
 

stress-test.js

@@ -0,0 +1,47 @@
+const watcher = require('./');
+const path = require('path');
+const fs = require('fs-extra');
+// const SegfaultHandler = require('segfault-handler');
+
+// SegfaultHandler.registerHandler('crash.log');
+
+let backend = [];
+if (process.platform === 'darwin') {
+  backends = ['fs-events'];
+} else if (process.platform === 'linux') {
+  backends = ['inotify'];
+} else if (process.platform === 'win32') {
+  backends = ['windows'];
+}
+
+async function run(backend) {
+  console.log('Run Stress Test For:', backend);
+
+  console.log('Initialising...');
+  let stressTestDir = path.join(__dirname, 'stress-test');
+  await fs.ensureDir(stressTestDir);
+  await watcher.subscribe(stressTestDir, () => {}, {
+    backend,
+  });
+
+  let currentRunId = 0;
+  while (true) {
+    console.log('Running stress test:', currentRunId);
+    await fs.remove(stressTestDir);
+    await Promise.all(
+      new Array(100).fill('').map(async (_, dirName) => {
+        let dir = path.join(stressTestDir, dirName.toString(10));
+        for (let filename = 0; filename < 250; filename++) {
+          let filepath = path.join(dir, filename.toString(10));
+          await fs.outputFile(filepath, '');
+        }
+        await fs.remove(dir);
+      }),
+    );
+    currentRunId++;
+  }
+}
+
+for (let backend of backends) {
+  run(backend);
+}