diff options
Diffstat (limited to 'node_modules/webworker-threads/src/WebWorkerThreads.cc')
| -rw-r--r-- | node_modules/webworker-threads/src/WebWorkerThreads.cc | 955 |
1 files changed, 955 insertions, 0 deletions
diff --git a/node_modules/webworker-threads/src/WebWorkerThreads.cc b/node_modules/webworker-threads/src/WebWorkerThreads.cc new file mode 100644 index 0000000..7cf7746 --- /dev/null +++ b/node_modules/webworker-threads/src/WebWorkerThreads.cc @@ -0,0 +1,955 @@ +//2011-11 Proyectos Equis Ka, s.l., jorge@jorgechamorro.com +//WebWorkerThreads.cc + + +#include <v8.h> +#include <node.h> +#include <uv.h> +#include <string.h> +#include <stdio.h> +#include <stdlib.h> +#include <string> + +#if defined(__unix__) || defined(__POSIX__) || defined(__APPLE__) || defined(_AIX) +#define WWT_PTHREAD 1 +#include <pthread.h> +#include <unistd.h> +#ifndef uv_cond_t +#define uv_cond_signal(x) pthread_cond_signal(x) +#define uv_cond_init(x) pthread_cond_init(x, NULL) +#define uv_cond_wait(x,y) pthread_cond_wait(x, y) +typedef pthread_cond_t uv_cond_t; +#endif +#else +#define pthread_setcancelstate(x,y) NULL +#define pthread_setcanceltype(x,y) NULL +#endif + + +/* +static int debug_threads= 0; +static int debug_allocs= 0; +*/ + +#include "queues_a_gogo.cc" +#include "bson.cc" +#include "jslib.cc" + +//using namespace node; +using namespace v8; + +static Persistent<String> id_symbol; +static Persistent<ObjectTemplate> threadTemplate; +static bool useLocker; + +static typeQueue* freeJobsQueue= NULL; +static typeQueue* freeThreadsQueue= NULL; + +#define kThreadMagicCookie 0x99c0ffee +typedef struct { + uv_async_t async_watcher; //MUST be the first one + + long int id; + uv_thread_t thread; + volatile int sigkill; + + typeQueue inQueue; //Jobs to run + typeQueue outQueue; //Jobs done + + volatile int IDLE; + uv_cond_t IDLE_cv; + uv_mutex_t IDLE_mutex; + + Isolate* isolate; + Persistent<Context> context; + Persistent<Object> JSObject; + Persistent<Object> threadJSObject; + Persistent<Object> dispatchEvents; + + unsigned long threadMagicCookie; +} typeThread; + +enum jobTypes { + kJobTypeEval, + kJobTypeEvent, + kJobTypeEventSerialized +}; + +typedef struct { + int jobType; + Persistent<Object> cb; + union { + struct { + int length; + String::Utf8Value* eventName; + String::Utf8Value** argumentos; + } typeEvent; + struct { + int length; + String::Utf8Value* eventName; + char* buffer; + size_t bufferSize; + } typeEventSerialized; + struct { + int error; + int tiene_callBack; + int useStringObject; + String::Utf8Value* resultado; + union { + char* scriptText_CharPtr; + String::Utf8Value* scriptText_StringObject; + }; + } typeEval; + }; +} typeJob; + +/* + +cd deps/minifier/src +gcc minify.c -o minify +cat ../../../src/events.js | ./minify kEvents_js > ../../../src/kEvents_js +cat ../../../src/load.js | ./minify kLoad_js > ../../../src/kLoad_js +cat ../../../src/createPool.js | ./minify kCreatePool_js > ../../../src/kCreatePool_js +cat ../../../src/worker.js | ./minify kWorker_js > ../../../src/kWorker_js +cat ../../../src/thread_nextTick.js | ./minify kThread_nextTick_js > ../../../src/kThread_nextTick_js + +*/ + +#include "events.js.c" +#include "load.js.c" +#include "createPool.js.c" +#include "worker.js.c" +#include "thread_nextTick.js.c" +//#include "JASON.js.c" + +//node-waf configure uninstall distclean configure build install + + + + + + + + +static typeQueueItem* nuJobQueueItem (void) { + typeQueueItem* qitem= queue_pull(freeJobsQueue); + if (!qitem) { + qitem= nuItem(kItemTypePointer, calloc(1, sizeof(typeJob))); + } + return qitem; +} + + + + + + +static typeThread* isAThread (Handle<Object> receiver) { + typeThread* thread; + + if (receiver->IsObject()) { + if (receiver->InternalFieldCount() == 1) { + thread= (typeThread*) receiver->GetPointerFromInternalField(0); + if (thread && (thread->threadMagicCookie == kThreadMagicCookie)) { + return thread; + } + } + } + + return NULL; +} + + + + + + +static void pushToInQueue (typeQueueItem* qitem, typeThread* thread) { + uv_mutex_lock(&thread->IDLE_mutex); + queue_push(qitem, &thread->inQueue); + if (thread->IDLE) { + uv_cond_signal(&thread->IDLE_cv); + } + uv_mutex_unlock(&thread->IDLE_mutex); +} + + + + + + +static Handle<Value> Puts (const Arguments &args) { + //fprintf(stdout, "*** Puts BEGIN\n"); + + HandleScope scope; + int i= 0; + while (i < args.Length()) { + String::Utf8Value c_str(args[i]); + fputs(*c_str, stdout); + i++; + } + fflush(stdout); + + //fprintf(stdout, "*** Puts END\n"); + return Undefined(); +} + +static Handle<Value> Print (const Arguments &args) { + HandleScope scope; + int i= 0; + while (i < args.Length()) { + String::Utf8Value c_str(args[i]); + fputs(*c_str, stdout); + i++; + } + static char end = '\n'; + fputs(&end, stdout); + fflush(stdout); + + //fprintf(stdout, "*** Puts END\n"); + return Undefined(); +} + + + + +static void eventLoop (typeThread* thread); + +// A background thread +#ifdef WWT_PTHREAD +static void* aThread (void* arg) { +#else +static void aThread (void* arg) { +#endif + + int dummy; + pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &dummy); + pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, &dummy); + + typeThread* thread= (typeThread*) arg; + thread->isolate= Isolate::New(); + thread->isolate->SetData(thread); + + if (useLocker) { + //printf("**** USING LOCKER: YES\n"); + v8::Locker myLocker(thread->isolate); + //v8::Isolate::Scope isolate_scope(thread->isolate); + eventLoop(thread); + } + else { + //printf("**** USING LOCKER: NO\n"); + //v8::Isolate::Scope isolate_scope(thread->isolate); + eventLoop(thread); + } + thread->isolate->Exit(); + thread->isolate->Dispose(); + + // wake up callback + if (!(thread->inQueue.length)) uv_async_send(&thread->async_watcher); +#ifdef WWT_PTHREAD + return NULL; +#endif +} + + + +static Handle<Value> threadEmit (const Arguments &args); +static Handle<Value> postMessage (const Arguments &args); +static Handle<Value> postError (const Arguments &args); + + + +static void eventLoop (typeThread* thread) { + thread->isolate->Enter(); + thread->context= Context::New(); + thread->context->Enter(); + + { + HandleScope scope1; + + Local<Object> global= thread->context->Global(); + + Handle<Object> fs_obj = Object::New(); + JSObjFn(fs_obj, "readFileSync", readFileSync_); + global->Set(String::New("native_fs_"), fs_obj, attribute_ro_dd); + + Handle<Object> console_obj = Object::New(); + JSObjFn(console_obj, "log", console_log); + JSObjFn(console_obj, "error", console_error); + global->Set(String::New("console"), console_obj, attribute_ro_dd); + + global->Set(String::NewSymbol("self"), global); + global->Set(String::NewSymbol("global"), global); + + global->Set(String::NewSymbol("puts"), FunctionTemplate::New(Puts)->GetFunction()); + global->Set(String::NewSymbol("print"), FunctionTemplate::New(Print)->GetFunction()); + + global->Set(String::NewSymbol("postMessage"), FunctionTemplate::New(postMessage)->GetFunction()); + global->Set(String::NewSymbol("__postError"), FunctionTemplate::New(postError)->GetFunction()); + + Local<Object> threadObject= Object::New(); + global->Set(String::NewSymbol("thread"), threadObject); + + threadObject->Set(String::NewSymbol("id"), Number::New(thread->id)); + threadObject->Set(String::NewSymbol("emit"), FunctionTemplate::New(threadEmit)->GetFunction()); + Local<Object> dispatchEvents= Script::Compile(String::New(kEvents_js))->Run()->ToObject()->CallAsFunction(threadObject, 0, NULL)->ToObject(); + Local<Object> dispatchNextTicks= Script::Compile(String::New(kThread_nextTick_js))->Run()->ToObject(); + Local<Array> _ntq= (v8::Array*) *threadObject->Get(String::NewSymbol("_ntq")); + + Script::Compile(String::New(kLoad_js))->Run(); + + double nextTickQueueLength= 0; + long int ctr= 0; + + //SetFatalErrorHandler(FatalErrorCB); + + while (!thread->sigkill) { + typeJob* job; + typeQueueItem* qitem; + + { + HandleScope scope2; + TryCatch onError; + String::Utf8Value* str; + Local<String> source; + Local<Script> script; + Local<Value> resultado; + + + while ((qitem= queue_pull(&thread->inQueue))) { + + job= (typeJob*) qitem->asPtr; + + if ((++ctr) > 2e3) { + ctr= 0; + V8::IdleNotification(); + } + + if (job->jobType == kJobTypeEval) { + //Ejecutar un texto + + if (job->typeEval.useStringObject) { + str= job->typeEval.scriptText_StringObject; + source= String::New(**str, (*str).length()); + delete str; + } + else { + source= String::New(job->typeEval.scriptText_CharPtr); + free(job->typeEval.scriptText_CharPtr); + } + + script= Script::New(source); + + if (!onError.HasCaught()) resultado= script->Run(); + + if (job->typeEval.tiene_callBack) { + job->typeEval.error= onError.HasCaught() ? 1 : 0; + job->typeEval.resultado= new String::Utf8Value(job->typeEval.error ? onError.Exception() : resultado); + queue_push(qitem, &thread->outQueue); + // wake up callback + if (!(thread->inQueue.length)) uv_async_send(&thread->async_watcher); + } + else { + queue_push(qitem, freeJobsQueue); + } + + if (onError.HasCaught()) onError.Reset(); + } + else if (job->jobType == kJobTypeEvent) { + //Emitir evento. + + Local<Value> args[2]; + str= job->typeEvent.eventName; + args[0]= String::New(**str, (*str).length()); + delete str; + + Local<Array> array= Array::New(job->typeEvent.length); + args[1]= array; + + int i= 0; + while (i < job->typeEvent.length) { + str= job->typeEvent.argumentos[i]; + array->Set(i, String::New(**str, (*str).length())); + delete str; + i++; + } + + free(job->typeEvent.argumentos); + queue_push(qitem, freeJobsQueue); + dispatchEvents->CallAsFunction(global, 2, args); + } + else if (job->jobType == kJobTypeEventSerialized) { + Local<Value> args[2]; + str= job->typeEventSerialized.eventName; + args[0]= String::New(**str, (*str).length()); + delete str; + + int len = job->typeEventSerialized.length; + Local<Array> array= Array::New(len); + args[1]= array; + + { + BSON *bson = new BSON(); + char* data = job->typeEventSerialized.buffer; + size_t size = job->typeEventSerialized.bufferSize; + BSONDeserializer deserializer(bson, data, size); + Local<Object> result = deserializer.DeserializeDocument()->ToObject(); + int i = 0; do { array->Set(i, result->Get(i)); } while (++i < len); + free(data); + } + + queue_push(qitem, freeJobsQueue); + dispatchEvents->CallAsFunction(global, 2, args); + } + } + + if (_ntq->Length()) { + + if ((++ctr) > 2e3) { + ctr= 0; + V8::IdleNotification(); + } + + resultado= dispatchNextTicks->CallAsFunction(global, 0, NULL); + if (onError.HasCaught()) { + nextTickQueueLength= 1; + onError.Reset(); + } + else { + nextTickQueueLength= resultado->NumberValue(); + } + } + } + + if (nextTickQueueLength || thread->inQueue.length) continue; + if (thread->sigkill) break; + + uv_mutex_lock(&thread->IDLE_mutex); + if (!thread->inQueue.length) { + thread->IDLE= 1; + uv_cond_wait(&thread->IDLE_cv, &thread->IDLE_mutex); + thread->IDLE= 0; + } + uv_mutex_unlock(&thread->IDLE_mutex); + } + } + + thread->context.Dispose(); +} + + + + + + +static void destroyaThread (typeThread* thread) { + + thread->sigkill= 0; + //TODO: hay que vaciar las colas y destruir los trabajos antes de ponerlas a NULL + thread->inQueue.first= thread->inQueue.last= NULL; + thread->outQueue.first= thread->outQueue.last= NULL; + thread->JSObject->SetPointerInInternalField(0, NULL); + thread->JSObject.Dispose(); + + uv_unref((uv_handle_t*)&thread->async_watcher); + + if (freeThreadsQueue) { + queue_push(nuItem(kItemTypePointer, thread), freeThreadsQueue); + } + else { + free(thread); + } +} + + + + + + +// C callback that runs in the main nodejs thread. This is the one responsible for +// calling the thread's JS callback. +static void Callback (uv_async_t *watcher, int revents) { + typeThread* thread= (typeThread*) watcher; + + if (thread->sigkill) { + destroyaThread(thread); + return; + } + + HandleScope scope; + typeJob* job; + Local<Value> argv[2]; + Local<Value> null= Local<Value>::New(Null()); + typeQueueItem* qitem; + String::Utf8Value* str; + + TryCatch onError; + while ((qitem= queue_pull(&thread->outQueue))) { + job= (typeJob*) qitem->asPtr; + + if (job->jobType == kJobTypeEval) { + + if (job->typeEval.tiene_callBack) { + str= job->typeEval.resultado; + + if (job->typeEval.error) { + argv[0]= Exception::Error(String::New(**str, (*str).length())); + argv[1]= null; + } else { + argv[0]= null; + argv[1]= String::New(**str, (*str).length()); + } + job->cb->CallAsFunction(thread->JSObject, 2, argv); + job->cb.Dispose(); + job->typeEval.tiene_callBack= 0; + + delete str; + job->typeEval.resultado= NULL; + } + + queue_push(qitem, freeJobsQueue); + + if (onError.HasCaught()) { + if (thread->outQueue.first) { + uv_async_send(&thread->async_watcher); // wake up callback again + } + node::FatalException(onError); + return; + } + } + else if (job->jobType == kJobTypeEvent) { + + //fprintf(stdout, "*** Callback\n"); + + Local<Value> args[2]; + + str= job->typeEvent.eventName; + args[0]= String::New(**str, (*str).length()); + delete str; + + Local<Array> array= Array::New(job->typeEvent.length); + args[1]= array; + + int i= 0; + while (i < job->typeEvent.length) { + str= job->typeEvent.argumentos[i]; + array->Set(i, String::New(**str, (*str).length())); + delete str; + i++; + } + + free(job->typeEvent.argumentos); + queue_push(qitem, freeJobsQueue); + thread->dispatchEvents->CallAsFunction(thread->JSObject, 2, args); + } + else if (job->jobType == kJobTypeEventSerialized) { + Local<Value> args[2]; + + str= job->typeEventSerialized.eventName; + args[0]= String::New(**str, (*str).length()); + delete str; + + int len = job->typeEventSerialized.length; + Local<Array> array= Array::New(len); + args[1]= array; + + { + BSON *bson = new BSON(); + char* data = job->typeEventSerialized.buffer; + size_t size = job->typeEventSerialized.bufferSize; + BSONDeserializer deserializer(bson, data, size); + Local<Object> result = deserializer.DeserializeDocument()->ToObject(); + int i = 0; do { array->Set(i, result->Get(i)); } while (++i < len); + free(data); + } + + queue_push(qitem, freeJobsQueue); + thread->dispatchEvents->CallAsFunction(thread->JSObject, 2, args); + } + } +} + + + + + + +// unconditionally destroys a thread by brute force. +static Handle<Value> Destroy (const Arguments &args) { + HandleScope scope; + //TODO: Hay que comprobar que this en un objeto y que tiene hiddenRefTotypeThread_symbol y que no es nil + //TODO: Aquí habría que usar static void TerminateExecution(int thread_id); + //TODO: static void v8::V8::TerminateExecution ( Isolate * isolate= NULL ) [static] + + typeThread* thread= isAThread(args.This()); + if (!thread) { + return ThrowException(Exception::TypeError(String::New("thread.destroy(): the receiver must be a thread object"))); + } + + if (!thread->sigkill) { + //pthread_cancel(thread->thread); + thread->sigkill= 1; + uv_mutex_lock(&thread->IDLE_mutex); + if (thread->IDLE) { + uv_cond_signal(&thread->IDLE_cv); + } + uv_mutex_unlock(&thread->IDLE_mutex); + } + + return Undefined(); +} + + + + + + +// Eval: Pushes a job into the thread's ->inQueue. +static Handle<Value> Eval (const Arguments &args) { + HandleScope scope; + + if (!args.Length()) { + return ThrowException(Exception::TypeError(String::New("thread.eval(program [,callback]): missing arguments"))); + } + + typeThread* thread= isAThread(args.This()); + if (!thread) { + return ThrowException(Exception::TypeError(String::New("thread.eval(): the receiver must be a thread object"))); + } + + typeQueueItem* qitem= nuJobQueueItem(); + typeJob* job= (typeJob*) qitem->asPtr; + + job->typeEval.tiene_callBack= ((args.Length() > 1) && (args[1]->IsFunction())); + if (job->typeEval.tiene_callBack) { + job->cb= Persistent<Object>::New(args[1]->ToObject()); + } + job->typeEval.scriptText_StringObject= new String::Utf8Value(args[0]); + job->typeEval.useStringObject= 1; + job->jobType= kJobTypeEval; + + pushToInQueue(qitem, thread); + return scope.Close(args.This()); +} + + + + + +static char* readFile (Handle<String> path) { + v8::String::Utf8Value c_str(path); + FILE* fp= fopen(*c_str, "rb"); + if (!fp) { + fprintf(stderr, "Error opening the file %s\n", *c_str); + //@bruno: Shouldn't we throw, here ? + return NULL; + } + fseek(fp, 0, SEEK_END); + size_t len= ftell(fp); + rewind(fp); //fseek(fp, 0, SEEK_SET); + char *buf= (char*)malloc((len+1) * sizeof(char)); // +1 to get null terminated string + if (fread(buf, sizeof(char), len, fp) < len) { + fprintf(stderr, "Error reading the file %s\n", *c_str); + return NULL; + } + buf[len] = 0; + fclose(fp); + /* + printf("SOURCE:\n%s\n", buf); + fflush(stdout); + */ + return buf; +} + + + + + + +// Load: Loads from file and passes to Eval +static Handle<Value> Load (const Arguments &args) { + HandleScope scope; + + if (!args.Length()) { + return ThrowException(Exception::TypeError(String::New("thread.load(filename [,callback]): missing arguments"))); + } + + typeThread* thread= isAThread(args.This()); + if (!thread) { + return ThrowException(Exception::TypeError(String::New("thread.load(): the receiver must be a thread object"))); + } + + char* source= readFile(args[0]->ToString()); //@Bruno: here we don't know if the file was not found or if it was an empty file + if (!source) return scope.Close(args.This()); //@Bruno: even if source is empty, we should call the callback ? + + typeQueueItem* qitem= nuJobQueueItem(); + typeJob* job= (typeJob*) qitem->asPtr; + + job->typeEval.tiene_callBack= ((args.Length() > 1) && (args[1]->IsFunction())); + if (job->typeEval.tiene_callBack) { + job->cb= Persistent<Object>::New(args[1]->ToObject()); + } + job->typeEval.scriptText_CharPtr= source; + job->typeEval.useStringObject= 0; + job->jobType= kJobTypeEval; + + pushToInQueue(qitem, thread); + + return scope.Close(args.This()); +} + + + + + + +static Handle<Value> processEmit (const Arguments &args) { + HandleScope scope; + + //fprintf(stdout, "*** processEmit\n"); + + if (!args.Length()) return scope.Close(args.This()); + + typeThread* thread= isAThread(args.This()); + if (!thread) { + return ThrowException(Exception::TypeError(String::New("thread.emit(): the receiver must be a thread object"))); + } + + typeQueueItem* qitem= nuJobQueueItem(); + typeJob* job= (typeJob*) qitem->asPtr; + + job->jobType= kJobTypeEvent; + job->typeEvent.length= args.Length()- 1; + job->typeEvent.eventName= new String::Utf8Value(args[0]); + job->typeEvent.argumentos= (v8::String::Utf8Value**) malloc(job->typeEvent.length* sizeof(void*)); + + int i= 1; + do { + job->typeEvent.argumentos[i-1]= new String::Utf8Value(args[i]); + } while (++i <= job->typeEvent.length); + + pushToInQueue(qitem, thread); + + return scope.Close(args.This()); +} + +static Handle<Value> processEmitSerialized (const Arguments &args) { + HandleScope scope; + + //fprintf(stdout, "*** processEmit\n"); + int len = args.Length(); + + if (!len) return scope.Close(args.This()); + + typeThread* thread= isAThread(args.This()); + if (!thread) { + return ThrowException(Exception::TypeError(String::New("thread.emit(): the receiver must be a thread object"))); + } + + typeQueueItem* qitem= nuJobQueueItem(); + typeJob* job= (typeJob*) qitem->asPtr; + + job->jobType= kJobTypeEventSerialized; + job->typeEventSerialized.length= len-1; + job->typeEventSerialized.eventName= new String::Utf8Value(args[0]); + Local<Array> array= Array::New(len-1); + int i = 1; do { array->Set(i-1, args[i]); } while (++i < len); + + { + char* buffer; + BSON *bson = new BSON(); + size_t object_size; + Local<Object> object = bson->GetSerializeObject(array); + BSONSerializer<CountStream> counter(bson, false, false); + counter.SerializeDocument(object); + object_size = counter.GetSerializeSize(); + buffer = (char *)malloc(object_size); + BSONSerializer<DataStream> data(bson, false, false, buffer); + data.SerializeDocument(object); + job->typeEventSerialized.buffer= buffer; + job->typeEventSerialized.bufferSize= object_size; + } + + pushToInQueue(qitem, thread); + + return scope.Close(args.This()); +} + +#define POST_EVENT(eventname) { \ + HandleScope scope; \ + int len = args.Length(); \ + \ + if (!len) return scope.Close(args.This()); \ + \ + typeThread* thread= (typeThread*) Isolate::GetCurrent()->GetData(); \ + \ + typeQueueItem* qitem= nuJobQueueItem(); \ + typeJob* job= (typeJob*) qitem->asPtr; \ + \ + job->jobType= kJobTypeEventSerialized; \ + job->typeEventSerialized.eventName= new String::Utf8Value(String::New(eventname)); \ + job->typeEventSerialized.length= len; \ + \ + Local<Array> array= Array::New(len); \ + int i = 0; do { array->Set(i, args[i]); } while (++i < len); \ + \ + { \ + char* buffer; \ + BSON *bson = new BSON(); \ + size_t object_size; \ + Local<Object> object = bson->GetSerializeObject(array); \ + BSONSerializer<CountStream> counter(bson, false, false); \ + counter.SerializeDocument(object); \ + object_size = counter.GetSerializeSize(); \ + buffer = (char *)malloc(object_size); \ + BSONSerializer<DataStream> data(bson, false, false, buffer); \ + data.SerializeDocument(object); \ + job->typeEventSerialized.buffer= buffer; \ + job->typeEventSerialized.bufferSize= object_size; \ + } \ + \ + queue_push(qitem, &thread->outQueue); \ + if (!(thread->inQueue.length)) uv_async_send(&thread->async_watcher); \ + \ + return scope.Close(args.This()); \ +} + +static Handle<Value> postMessage (const Arguments &args) { + POST_EVENT("message"); +} + +static Handle<Value> postError (const Arguments &args) { + POST_EVENT("error"); +} + +static Handle<Value> threadEmit (const Arguments &args) { + HandleScope scope; + + //fprintf(stdout, "*** threadEmit\n"); + + if (!args.Length()) return scope.Close(args.This()); + + int i; + typeThread* thread= (typeThread*) Isolate::GetCurrent()->GetData(); + + typeQueueItem* qitem= nuJobQueueItem(); + typeJob* job= (typeJob*) qitem->asPtr; + + job->jobType= kJobTypeEvent; + job->typeEvent.length= args.Length()- 1; + job->typeEvent.eventName= new String::Utf8Value(args[0]); + job->typeEvent.argumentos= (v8::String::Utf8Value**) malloc(job->typeEvent.length* sizeof(void*)); + + i= 1; + do { + job->typeEvent.argumentos[i-1]= new String::Utf8Value(args[i]); + } while (++i <= job->typeEvent.length); + + queue_push(qitem, &thread->outQueue); + if (!(thread->inQueue.length)) uv_async_send(&thread->async_watcher); // wake up callback + + //fprintf(stdout, "*** threadEmit END\n"); + + return scope.Close(args.This()); +} + + + + + + + + +// Creates and launches a new isolate in a new background thread. +static Handle<Value> Create (const Arguments &args) { + HandleScope scope; + + typeThread* thread; + typeQueueItem* qitem= NULL; + qitem= queue_pull(freeThreadsQueue); + if (qitem) { + thread= (typeThread*) qitem->asPtr; + destroyItem(qitem); + } + else { + thread= (typeThread*) calloc(1, sizeof(typeThread)); + thread->threadMagicCookie= kThreadMagicCookie; + } + + static long int threadsCtr= 0; + thread->id= threadsCtr++; + + thread->JSObject= Persistent<Object>::New(threadTemplate->NewInstance()); + thread->JSObject->Set(id_symbol, Integer::New(thread->id)); + thread->JSObject->SetPointerInInternalField(0, thread); + Local<Value> dispatchEvents= Script::Compile(String::New(kEvents_js))->Run()->ToObject()->CallAsFunction(thread->JSObject, 0, NULL); + thread->dispatchEvents= Persistent<Object>::New(dispatchEvents->ToObject()); + + uv_async_init(uv_default_loop(), &thread->async_watcher, Callback); + uv_ref((uv_handle_t*)&thread->async_watcher); + + uv_cond_init(&thread->IDLE_cv); + uv_mutex_init(&thread->IDLE_mutex); + uv_mutex_init(&thread->inQueue.queueLock); + uv_mutex_init(&thread->outQueue.queueLock); + +#ifdef WWT_PTHREAD + pthread_attr_t attr; + pthread_attr_init(&attr); + pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); + int err= pthread_create(&thread->thread, &attr, aThread, thread); + pthread_attr_destroy(&attr); +#else + int err= uv_thread_create(&thread->thread, aThread, thread); +#endif + if (err) { + //Ha habido un error no se ha arrancado esta thread + destroyaThread(thread); + return ThrowException(Exception::TypeError(String::New("create(): error in pthread_create()"))); + } + + V8::AdjustAmountOfExternalAllocatedMemory(sizeof(typeThread)); //OJO V8 con V mayúscula. + return scope.Close(thread->JSObject); +} + + +#if NODE_MODULE_VERSION >= 0x000B +void Init (Handle<Object> target, Handle<Value> module) { +#else +void Init (Handle<Object> target) { +#endif + + initQueues(); + freeThreadsQueue= nuQueue(-3); + freeJobsQueue= nuQueue(-4); + + HandleScope scope; + + useLocker= v8::Locker::IsActive(); + + target->Set(String::NewSymbol("create"), FunctionTemplate::New(Create)->GetFunction()); + target->Set(String::NewSymbol("createPool"), Script::Compile(String::New(kCreatePool_js))->Run()->ToObject()); + target->Set(String::NewSymbol("Worker"), Script::Compile(String::New(kWorker_js))->Run()->ToObject()->CallAsFunction(target, 0, NULL)->ToObject()); + //target->Set(String::NewSymbol("JASON"), Script::Compile(String::New(kJASON_js))->Run()->ToObject()); + + id_symbol= Persistent<String>::New(String::NewSymbol("id")); + + threadTemplate= Persistent<ObjectTemplate>::New(ObjectTemplate::New()); + threadTemplate->SetInternalFieldCount(1); + threadTemplate->Set(id_symbol, Integer::New(0)); + threadTemplate->Set(String::NewSymbol("eval"), FunctionTemplate::New(Eval)); + threadTemplate->Set(String::NewSymbol("load"), FunctionTemplate::New(Load)); + threadTemplate->Set(String::NewSymbol("emit"), FunctionTemplate::New(processEmit)); + threadTemplate->Set(String::NewSymbol("emitSerialized"), FunctionTemplate::New(processEmitSerialized)); + threadTemplate->Set(String::NewSymbol("destroy"), FunctionTemplate::New(Destroy)); + +} + + + + +NODE_MODULE(WebWorkerThreads, Init) + +/* +gcc -E -I /Users/jorge/JAVASCRIPT/binarios/include/node -o /o.c /Users/jorge/JAVASCRIPT/threads_a_gogo/src/threads_a_gogo.cc && mate /o.c +*/ |