Merge pull request #200 from atom/aw/deferred-worker-start

Defer worker start

Author: smashwilson

src/binding.cpp

@@ -123,7 +123,7 @@ void configure(const Nan::FunctionCallbackInfo<Value> &info)
   }
 
   all->set_result(move(r));
-  all->fire_if_empty(true);
+  all->mark_ready();
 }
 
 void watch(const Nan::FunctionCallbackInfo<Value> &info)

src/hub.cpp

@@ -60,7 +60,6 @@ Hub::Hub() :
     report_uv_error(err);
   }
 
-  report_if_error(worker_thread.run());
   freeze();
 }
 
@@ -92,13 +91,13 @@ Result<> Hub::unwatch(ChannelID channel_id, unique_ptr<AsyncCallback> &&ack_call
 
   string root;
   shared_ptr<AllCallback> all = AllCallback::create(move(ack_callback));
+  unique_ptr<AsyncCallback> worker_cb = all->create_callback("@atom/watcher:hub.unwatch.worker");
+  unique_ptr<AsyncCallback> polling_cb = all->create_callback("@atom/worker:hub.unwatch.polling");
 
   Result<> r = ok_result();
-  r &= send_command(
-    worker_thread, CommandPayloadBuilder::remove(channel_id), all->create_callback("@atom/watcher:hub.unwatch.worker"));
-  r &= send_command(polling_thread,
-    CommandPayloadBuilder::remove(channel_id),
-    all->create_callback("@atom/worker:hub.unwatch.polling"));
+  r &= send_command(worker_thread, CommandPayloadBuilder::remove(channel_id), move(worker_cb));
+  r &= send_command(polling_thread, CommandPayloadBuilder::remove(channel_id), move(polling_cb));
+  all->mark_ready();
 
   auto maybe_event_callback = channel_callbacks.find(channel_id);
   if (maybe_event_callback == channel_callbacks.end()) {

src/nan/all_callback.cpp

@@ -34,6 +34,7 @@ shared_ptr<AllCallback> AllCallback::create(unique_ptr<AsyncCallback> &&done)
 
 AllCallback::AllCallback(unique_ptr<AsyncCallback> &&done) :
   done(move(done)),
+  ready{false},
   fired{false},
   total{0},
   remaining{0},
@@ -55,6 +56,12 @@ unique_ptr<AsyncCallback> AllCallback::create_callback(const char *async_name)
   return fn_callback(async_name, functions.front());
 }
 
+void AllCallback::mark_ready()
+{
+  ready = true;
+  fire_if_empty(true);
+}
+
 void AllCallback::set_result(Result<> &&r)
 {
   if (r.is_ok()) return;
@@ -70,6 +77,7 @@ void AllCallback::set_result(Result<> &&r)
 void AllCallback::fire_if_empty(bool sync)
 {
   if (remaining > 0) return;
+  if (!ready) return;
   if (fired) return;
   fired = true;
 

src/nan/all_callback.h

@@ -21,6 +21,8 @@ class AllCallback
 
   std::unique_ptr<AsyncCallback> create_callback(const char *async_name);
 
+  void mark_ready();
+
   void set_result(Result<> &&r);
 
   void fire_if_empty(bool sync);
@@ -36,6 +38,7 @@ class AllCallback
   void callback_complete(size_t callback_index, const Nan::FunctionCallbackInfo<v8::Value> &info);
 
   std::unique_ptr<AsyncCallback> done;
+  bool ready;
   bool fired;
   size_t total;
   size_t remaining;

src/worker/worker_thread.cpp

@@ -42,6 +42,26 @@ Result<> WorkerThread::body()
   return platform->listen();
 }
 
+Result<Thread::OfflineCommandOutcome> WorkerThread::handle_offline_command(const CommandPayload *payload)
+{
+  Result<OfflineCommandOutcome> r = Thread::handle_offline_command(payload);
+  if (r.is_error()) return r;
+
+  if (payload->get_action() == COMMAND_ADD) {
+    return ok_result(TRIGGER_RUN);
+  }
+
+  if (payload->get_action() == COMMAND_CACHE_SIZE) {
+    handle_cache_size_command(payload);
+  }
+
+  if (payload->get_action() == COMMAND_STATUS) {
+    handle_status_command(payload);
+  }
+
+  return ok_result(OFFLINE_ACK);
+}
+
 Result<Thread::CommandOutcome> WorkerThread::handle_add_command(const CommandPayload *payload)
 {
   Result<bool> r = platform->handle_add_command(

src/worker/worker_thread.h

@@ -30,6 +30,8 @@ class WorkerThread : public Thread
 
   Result<> body() override;
 
+  Result<OfflineCommandOutcome> handle_offline_command(const CommandPayload *payload) override;
+
   Result<CommandOutcome> handle_add_command(const CommandPayload *payload) override;
 
   Result<CommandOutcome> handle_remove_command(const CommandPayload *payload) override;

test/configuration.test.js

@@ -29,19 +29,29 @@ describe('configuration', function () {
     assert.match(contents, /FileLogger opened/)
   })
 
-  it('configures the worker thread logger', async function () {
-    await configure({ workerLog: fixture.workerLogFile })
-
-    const contents = await fs.readFile(fixture.workerLogFile)
-    assert.match(contents, /FileLogger opened/)
-  })
-
   it('fails if the main log file cannot be written', async function () {
     await assert.isRejected(configure({ mainLog: badPath }), /No such file or directory/)
   })
 
-  it('fails if the worker log file cannot be written', async function () {
-    await assert.isRejected(configure({ workerLog: badPath }), /No such file or directory/)
+  describe('for the worker thread', function () {
+    // There's currently no way to *stop* the worker thread, so we can't reliably test it in the stopped state.
+
+    describe("after it's started", function () {
+      it('configures the logger', async function () {
+        await fixture.watch([], { poll: false }, () => {})
+
+        await configure({ workerLog: fixture.workerLogFile })
+
+        const contents = await fs.readFile(fixture.workerLogFile)
+        assert.match(contents, /FileLogger opened/)
+      })
+
+      it('fails if the worker log file cannot be written', async function () {
+        await fixture.watch([], { poll: false }, () => {})
+
+        await assert.isRejected(configure({ workerLog: badPath }), /No such file or directory/)
+      })
+    })
   })
 
   describe('for the polling thread', function () {