Skip to content

Commit

Permalink
workers: initial implementation
Browse files Browse the repository at this point in the history
  • Loading branch information
petkaantonov committed Sep 6, 2015
1 parent ba02bd0 commit ea143f7
Show file tree
Hide file tree
Showing 66 changed files with 4,625 additions and 486 deletions.
31 changes: 31 additions & 0 deletions LICENSE
Original file line number Diff line number Diff line change
Expand Up @@ -1041,4 +1041,35 @@ 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.

- src/producer-consumer-queue.h. The folly::ProducerConsumerQueue class is a
one-producer one-consumer queue with very low synchronization overhead.
ProducerConsumerQueue's license follows:
"""
Copyright 2015 Facebook, Inc.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

Significant changes made to the software:

- Removed Boost dependency
- Removed support for storing values directly
- Removed construction and destruction of the queue items feature
- Added initialization of all values to nullptr
- Made size a template parameter
- Crash instead of throw if malloc fails in constructor
- Changed namespace from folly to node
- Removed sizeGuess(), isFull(), isEmpty(), popFront() and frontPtr() methods
- Renamed write() to PushBack(), read() to PopFront()
- Added padding to fields
"""
6 changes: 6 additions & 0 deletions Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -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))
Expand Down
7 changes: 4 additions & 3 deletions common.gypi
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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',
Expand All @@ -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++
},
}],
],
Expand Down
216 changes: 216 additions & 0 deletions lib/worker.js
Original file line number Diff line number Diff line change
@@ -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;
17 changes: 16 additions & 1 deletion node.gyp
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,7 @@
'lib/internal/socket_list.js',
'lib/internal/repl.js',
'lib/internal/util.js',
'lib/worker.js'
],
},

Expand Down Expand Up @@ -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',
Expand All @@ -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',
Expand All @@ -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',
Expand All @@ -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',
Expand All @@ -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',
Expand All @@ -184,6 +193,11 @@
'V8_DEPRECATION_WARNINGS=1',
],

'xcode_settings': {
'OTHER_LDFLAGS': [
'-stdlib=libc++',
],
},

'conditions': [
[ 'node_tag!=""', {
Expand Down Expand Up @@ -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',
Expand Down
Loading

0 comments on commit ea143f7

Please sign in to comment.