Unbounded message queue

LICENSE

@@ -1041,4 +1041,3 @@ The externally maintained libraries used by Node.js are:
        on the database or the code.  Any person making a contribution to the
        database or code waives all rights to future claims in that
        contribution or in the TZ Database.
-  """

Makefile

@@ -182,6 +182,12 @@ test-timers:
 test-timers-clean:
 	$(MAKE) --directory=tools clean
 
+test-workers: all
+	$(PYTHON) tools/test.py --mode=release workers -J
+
+test-workers-debug: all
+	$(PYTHON) tools/test.py --mode=debug workers -J
+
 apidoc_sources = $(wildcard doc/api/*.markdown)
 apidocs = $(addprefix out/,$(apidoc_sources:.markdown=.html)) \
 		$(addprefix out/,$(apidoc_sources:.markdown=.json))

common.gypi

@@ -264,7 +264,7 @@
         'libraries': [ '-llog' ],
       }],
       ['OS=="mac"', {
-        'defines': ['_DARWIN_USE_64_BIT_INODE=1'],
+        'defines': ['_DARWIN_USE_64_BIT_INODE=1', 'NODE_OS_MACOSX'],
         'xcode_settings': {
           'ALWAYS_SEARCH_USER_PATHS': 'NO',
           'GCC_CW_ASM_SYNTAX': 'NO',                # No -fasm-blocks
@@ -275,7 +275,7 @@
           'GCC_ENABLE_PASCAL_STRINGS': 'NO',        # No -mpascal-strings
           'GCC_THREADSAFE_STATICS': 'NO',           # -fno-threadsafe-statics
           'PREBINDING': 'NO',                       # No -Wl,-prebind
-          'MACOSX_DEPLOYMENT_TARGET': '10.5',       # -mmacosx-version-min=10.5
+          'MACOSX_DEPLOYMENT_TARGET': '10.7',       # -mmacosx-version-min=10.7
           'USE_HEADERMAP': 'NO',
           'OTHER_CFLAGS': [
             '-fno-strict-aliasing',
@@ -302,7 +302,8 @@
           ['clang==1', {
             'xcode_settings': {
               'GCC_VERSION': 'com.apple.compilers.llvm.clang.1_0',
-              'CLANG_CXX_LANGUAGE_STANDARD': 'gnu++0x',  # -std=gnu++0x
+              'CLANG_CXX_LANGUAGE_STANDARD': 'c++11',  # -std=c++11
+              'CLANG_CXX_LIBRARY': 'libc++', #-stdlib=libc++
             },
           }],
         ],

lib/worker.js

@@ -0,0 +1,216 @@
+'use strict';
+
+if (!process.features.experimental_workers) {
+  throw new Error('Experimental workers are not enabled');
+}
+
+const util = require('util');
+const assert = require('assert');
+const EventEmitter = require('events');
+const WorkerContextBinding = process.binding('worker_context');
+const JSONStringify = function(value) {
+  if (value === undefined) value = null;
+  return JSON.stringify(value);
+};
+const JSONParse = JSON.parse;
+const EMPTY_ARRAY = [];
+
+const workerContextSymbol = Symbol('workerContext');
+const installEventsSymbol = Symbol('installEvents');
+const checkAliveSymbol = Symbol('checkAlive');
+const initSymbol = WorkerContextBinding.initSymbol;
+
+const builtinErrorTypes = new Map([
+  Error, SyntaxError, RangeError, URIError, TypeError, EvalError, ReferenceError
+].map(function(Type) {
+  return [Type.name, Type];
+}));
+
+const Worker = WorkerContextBinding.JsConstructor;
+util.inherits(Worker, EventEmitter);
+
+Worker.prototype[initSymbol] = function(entryModulePath, options) {
+  if (typeof entryModulePath !== 'string')
+    throw new TypeError('entryModulePath must be a string');
+  EventEmitter.call(this);
+  options = Object(options);
+  const keepAlive =
+      options.keepAlive === undefined ? true : !!options.keepAlive;
+  const evalCode = !!options.eval;
+  const userData = JSONStringify(options.data);
+  this[workerContextSymbol] =
+      new WorkerContextBinding.WorkerContext(entryModulePath, {
+        keepAlive: keepAlive,
+        userData: userData,
+        eval: evalCode
+      });
+  this[installEventsSymbol]();
+};
+
+Worker.prototype[installEventsSymbol] = function() {
+  const workerObject = this;
+  const workerContext = this[workerContextSymbol];
+
+  const onerror = function(payload) {
+    var ErrorConstructor = builtinErrorTypes.get(payload.builtinType);
+    if (typeof ErrorConstructor !== 'function')
+      ErrorConstructor = Error;
+    const error = new ErrorConstructor(payload.message);
+    error.stack = payload.stack;
+    util._extend(error, payload.additionalProperties);
+    workerObject.emit('error', error);
+  };
+
+  workerContext._onexit = function(exitCode) {
+    workerObject[workerContextSymbol] = null;
+    workerObject.emit('exit', exitCode);
+  };
+
+  workerContext._onmessage = function(payload, messageType) {
+    payload = JSONParse(payload);
+    switch (messageType) {
+      case WorkerContextBinding.USER:
+        return workerObject.emit('message', payload);
+      case WorkerContextBinding.EXCEPTION:
+        return onerror(payload);
+      default:
+        assert.fail('unreachable');
+    }
+  };
+};
+
+Worker.prototype[checkAliveSymbol] = function() {
+  if (!this[workerContextSymbol])
+    throw new RangeError('this worker has been terminated');
+};
+
+Worker.prototype.postMessage = function(payload) {
+  this[checkAliveSymbol]();
+  this[workerContextSymbol].postMessage(JSONStringify(payload),
+                                        EMPTY_ARRAY,
+                                        WorkerContextBinding.USER);
+};
+
+Worker.prototype.terminate = function(callback) {
+  this[checkAliveSymbol]();
+  const context = this[workerContextSymbol];
+  this[workerContextSymbol] = null;
+  if (typeof callback === 'function') {
+    this.once('exit', function(exitCode) {
+      callback(null, exitCode);
+    });
+  }
+  context.terminate();
+};
+
+Worker.prototype.ref = function() {
+  this[checkAliveSymbol]();
+  this[workerContextSymbol].ref();
+};
+
+Worker.prototype.unref = function() {
+  this[checkAliveSymbol]();
+  this[workerContextSymbol].unref();
+};
+
+if (process.isWorkerThread) {
+  const postMessage = function(payload, transferList, type) {
+    if (!Array.isArray(transferList))
+      throw new TypeError('transferList must be an array');
+
+    WorkerContextBinding.workerWrapper._postMessage(JSONStringify(payload),
+                                                    transferList,
+                                                    type);
+  };
+  const workerFatalError = function(er) {
+    const defaultStack = null;
+    const defaultMessage = '[toString() conversion failed]';
+    const defaultBuiltinType = 'Error';
+
+    var message = defaultMessage;
+    var builtinType = defaultBuiltinType;
+    var stack = defaultStack;
+    var additionalProperties = {};
+
+    // As this is a fatal error handler we cannot throw any new errors here
+    // but should still try to extract as much info as possible from the
+    // cause.
+    if (er instanceof Error) {
+      // .name can be a getter that throws.
+      try {
+        builtinType = er.name;
+      } catch (ignore) {}
+
+      if (typeof builtinType !== 'string')
+        builtinType = defaultBuiltinType;
+
+      // .stack can be a getter that throws.
+      try {
+        stack = er.stack;
+      } catch (ignore) {}
+
+      if (typeof stack !== 'string')
+        stack = defaultStack;
+
+      try {
+        // Get inherited enumerable properties.
+        // .name, .stack and .message are all non-enumerable
+        for (var key in er)
+          additionalProperties[key] = er[key];
+        // The message delivery must always succeed, otherwise the real cause
+        // of this fatal error is masked.
+        JSONStringify(additionalProperties);
+      } catch (e) {
+        additionalProperties = {};
+      }
+    }
+
+    try {
+      // .message can be a getter that throws or the call to toString may fail.
+      if (er instanceof Error) {
+        message = er.message;
+        if (typeof message !== 'string')
+          message = '' + er;
+      } else {
+        message = '' + er;
+      }
+    } catch (e) {
+      message = defaultMessage;
+    }
+
+    postMessage({
+      message: message,
+      stack: stack,
+      additionalProperties: additionalProperties,
+      builtinType: builtinType
+    }, EMPTY_ARRAY, WorkerContextBinding.EXCEPTION);
+  };
+
+  util._extend(Worker, EventEmitter.prototype);
+  EventEmitter.call(Worker);
+
+  WorkerContextBinding.workerWrapper._onmessage =
+      function(payload, messageType) {
+    payload = JSONParse(payload);
+    switch (messageType) {
+      case WorkerContextBinding.USER:
+        return Worker.emit('message', payload);
+      default:
+        assert.fail('unreachable');
+    }
+  };
+
+  Worker.postMessage = function(payload) {
+    postMessage(payload, EMPTY_ARRAY, WorkerContextBinding.USER);
+  };
+
+  Object.defineProperty(Worker, '_workerFatalError', {
+    configurable: true,
+    writable: false,
+    enumerable: false,
+    value: workerFatalError
+  });
+}
+
+
+module.exports = Worker;

node.gyp

@@ -73,6 +73,7 @@
       'lib/internal/socket_list.js',
       'lib/internal/repl.js',
       'lib/internal/util.js',
+      'lib/worker.js'
     ],
   },
 
@@ -118,6 +119,8 @@
         'src/node_watchdog.cc',
         'src/node_zlib.cc',
         'src/node_i18n.cc',
+        'src/notification-channel.cc',
+        'src/persistent-handle-cleanup.cc',
         'src/pipe_wrap.cc',
         'src/signal_wrap.cc',
         'src/spawn_sync.cc',
@@ -130,6 +133,7 @@
         'src/process_wrap.cc',
         'src/udp_wrap.cc',
         'src/uv.cc',
+        'src/worker.cc',
         # headers to make for a more pleasant IDE experience
         'src/async-wrap.h',
         'src/async-wrap-inl.h',
@@ -143,6 +147,7 @@
         'src/node.h',
         'src/node_buffer.h',
         'src/node_constants.h',
+        'src/node-contextify.h',
         'src/node_file.h',
         'src/node_http_parser.h',
         'src/node_internals.h',
@@ -152,7 +157,10 @@
         'src/node_watchdog.h',
         'src/node_wrap.h',
         'src/node_i18n.h',
+        'src/notification-channel.h',
+        'src/persistent-handle-cleanup.h',
         'src/pipe_wrap.h',
+        'src/producer-consumer-queue.h',
         'src/tty_wrap.h',
         'src/tcp_wrap.h',
         'src/udp_wrap.h',
@@ -166,6 +174,7 @@
         'src/util.h',
         'src/util-inl.h',
         'src/util.cc',
+        'src/worker.h',
         'deps/http_parser/http_parser.h',
         'deps/v8/include/v8.h',
         'deps/v8/include/v8-debug.h',
@@ -184,6 +193,11 @@
         'V8_DEPRECATION_WARNINGS=1',
       ],
 
+      'xcode_settings': {
+          'OTHER_LDFLAGS': [
+            '-stdlib=libc++',
+          ],
+      },
 
       'conditions': [
         [ 'node_tag!=""', {
@@ -641,7 +655,8 @@
       'dependencies': [
         'deps/gtest/gtest.gyp:gtest',
         'deps/v8/tools/gyp/v8.gyp:v8',
-        'deps/v8/tools/gyp/v8.gyp:v8_libplatform'
+        'deps/v8/tools/gyp/v8.gyp:v8_libplatform',
+        'deps/uv/uv.gyp:libuv',
       ],
       'include_dirs': [
         'src',

src/async-wrap.cc

@@ -162,6 +162,8 @@ void LoadAsyncWrapperInfo(Environment* env) {
 Local<Value> AsyncWrap::MakeCallback(const Local<Function> cb,
                                       int argc,
                                       Local<Value>* argv) {
+  if (!env()->CanCallIntoJs())
+    return Undefined(env()->isolate());
   CHECK(env()->context() == env()->isolate()->GetCurrentContext());
 
   Local<Object> context = object();
@@ -181,7 +183,7 @@ Local<Value> AsyncWrap::MakeCallback(const Local<Function> cb,
     }
   }
 
-  TryCatch try_catch;
+  TryCatch try_catch(env()->isolate());
   try_catch.SetVerbose(true);
 
   if (has_domain) {
@@ -196,6 +198,8 @@ Local<Value> AsyncWrap::MakeCallback(const Local<Function> cb,
   if (has_async_queue()) {
     try_catch.SetVerbose(false);
     env()->async_hooks_pre_function()->Call(context, 0, nullptr);
+    if (try_catch.HasTerminated())
+      return Undefined(env()->isolate());
     if (try_catch.HasCaught())
       FatalError("node::AsyncWrap::MakeCallback", "pre hook threw");
     try_catch.SetVerbose(true);
@@ -224,6 +228,8 @@ Local<Value> AsyncWrap::MakeCallback(const Local<Function> cb,
   if (has_async_queue()) {
     try_catch.SetVerbose(false);
     env()->async_hooks_post_function()->Call(context, 0, nullptr);
+    if (try_catch.HasTerminated())
+      return Undefined(env()->isolate());
     if (try_catch.HasCaught())
       FatalError("node::AsyncWrap::MakeCallback", "post hook threw");
     try_catch.SetVerbose(true);