Multithread

Sources/worker.cpp

@@ -8,30 +8,42 @@
 #include <kinc/log.h>
 #include <kinc/threads/thread.h>
 #include <kinc/threads/mutex.h>
+#include <kinc/threads/event.h>
 #include <kinc/io/filereader.h>
 #include <kinc/system.h>
 
+#include <v8-value-serializer.h>
+#include <v8-wasm.h>
+#include <v8-array-buffer.h>
+#include <atomic>
+#include <memory>
+
 using namespace v8;
 
 namespace {
 	const uint32_t worker_data_slot = 0;
 
+	struct TransferData;
+
 	struct WorkerMessage {
-		char* message;
-		int length;
+		uint8_t* data;
+		size_t size;
+		TransferData* transferData;
 	};
 
 	struct MessageQueue {
 		std::vector<WorkerMessage> messages;
 		kinc_mutex_t messageMutex;
+		kinc_event_t event;
 		Global<Function> messageFunc;
+		const char* label = "?";
 	};
 
 	struct WorkerMessagePort {
 		MessageQueue workerMessages;
 		MessageQueue ownerMessages;
 
-		bool isTerminated = false;
+		std::atomic<bool> isTerminated{false};
 	};
 
 	struct WorkerData {
@@ -53,6 +65,7 @@ namespace {
 		double interval;
 		double nextCallTime;
 		int id;
+		bool isTimeout;
 	};
 
 	struct IntervalFunctions {
@@ -60,6 +73,70 @@ namespace {
 		int latestId;
 	};
 
+	struct TransferData {
+		std::vector<std::shared_ptr<BackingStore>> sharedArrayBuffers;
+		std::vector<CompiledWasmModule> wasmModules;
+		std::unique_ptr<SharedValueConveyor> sharedValueDispatch;
+	};
+
+	class SerializerDelegate : public ValueSerializer::Delegate {
+	public:
+		TransferData* transferData;
+
+		SerializerDelegate(TransferData* data) : transferData(data) {}
+
+		void ThrowDataCloneError(Local<String> message) override {
+			Isolate* isolate = Isolate::GetCurrent();
+			isolate->ThrowException(message);
+		}
+
+		Maybe<uint32_t> GetSharedArrayBufferId(Isolate* isolate, Local<SharedArrayBuffer> shared_array_buffer) override {
+			auto backingStore = shared_array_buffer->GetBackingStore();
+			uint32_t id = (uint32_t)transferData->sharedArrayBuffers.size();
+			transferData->sharedArrayBuffers.push_back(backingStore);
+			return Just(id);
+		}
+
+		Maybe<uint32_t> GetWasmModuleTransferId(Isolate* isolate, Local<WasmModuleObject> module) override {
+			CompiledWasmModule compiled = module->GetCompiledModule();
+			uint32_t id = (uint32_t)transferData->wasmModules.size();
+			transferData->wasmModules.push_back(compiled);
+			return Just(id);
+		}
+
+		bool AdoptSharedValueConveyor(Isolate* isolate, SharedValueConveyor&& conveyor) override {
+			transferData->sharedValueDispatch = std::make_unique<SharedValueConveyor>(std::move(conveyor));
+			return true;
+		}
+	};
+
+	class DeserializerDelegate : public ValueDeserializer::Delegate {
+	public:
+		TransferData* transferData;
+
+		DeserializerDelegate(TransferData* data) : transferData(data) {}
+
+		MaybeLocal<SharedArrayBuffer> GetSharedArrayBufferFromId(Isolate* isolate, uint32_t clone_id) override {
+			if (clone_id >= transferData->sharedArrayBuffers.size()) {
+				return MaybeLocal<SharedArrayBuffer>();
+			}
+			return SharedArrayBuffer::New(isolate, transferData->sharedArrayBuffers[clone_id]);
+		}
+
+		MaybeLocal<WasmModuleObject> GetWasmModuleFromId(Isolate* isolate, uint32_t transfer_id) override {
+			if (transfer_id >= transferData->wasmModules.size()) {
+				return MaybeLocal<WasmModuleObject>();
+			}
+			return WasmModuleObject::FromCompiledModule(isolate, transferData->wasmModules[transfer_id]);
+		}
+
+		const SharedValueConveyor* GetSharedValueConveyor(Isolate* isolate) override {
+			return transferData->sharedValueDispatch.get();
+		}
+	};
+
+	void handle_exception(Isolate* isolate, TryCatch* try_catch);
+
 	void set_onmessage(MessageQueue* messageQueue, Isolate* isolate, Local<Value> callback) {
 		if (callback->IsFunction()) {
 			messageQueue->messageFunc.Reset(isolate, Local<Function>::Cast(callback));
@@ -73,20 +150,53 @@ namespace {
 	}
 
 	void post_message(MessageQueue* messageQueue, Isolate* isolate, Local<Value> messageObject) {
-		Local<String> result = JSON::Stringify(isolate->GetCurrentContext(), messageObject).ToLocalChecked();
+		Local<Context> context = isolate->GetCurrentContext();
+		TransferData* transferData = new TransferData();
+
+		SerializerDelegate delegate(transferData);
+		ValueSerializer serializer(isolate, &delegate);
+		serializer.WriteHeader();
+
+		TryCatch try_catch(isolate);
+		Maybe<bool> writeResult = serializer.WriteValue(context, messageObject);
+		if (writeResult.IsNothing()) {
+			if (try_catch.HasCaught()) {
+				isolate->ThrowException(try_catch.Exception());
+			} else {
+				isolate->ThrowException(String::NewFromUtf8(isolate, "postMessage: failed to serialize message").ToLocalChecked());
+			}
+			delete transferData;
+			return;
+		}
+
+		std::pair<uint8_t*, size_t> released = serializer.Release();
 
 		WorkerMessage message;
-		message.length = result->Length();
-		message.message = new char[message.length];
-		result->WriteUtf8(isolate, message.message, message.length);
+		message.data = released.first;
+		message.size = released.second;
+		message.transferData = transferData;
 
 		kinc_mutex_lock(&messageQueue->messageMutex);
 		messageQueue->messages.push_back(message);
 		kinc_mutex_unlock(&messageQueue->messageMutex);
+		kinc_event_signal(&messageQueue->event);
 	}
 
 	void handle_message_queue(Isolate* isolate, const Global<Context>& context, MessageQueue* messageQueue) {
+		kinc_mutex_lock(&messageQueue->messageMutex);
+		if (messageQueue->messages.empty()) {
+			kinc_mutex_unlock(&messageQueue->messageMutex);
+			return;
+		}
+		std::vector<WorkerMessage> messages;
+		messages.swap(messageQueue->messages);
+		kinc_mutex_unlock(&messageQueue->messageMutex);
+
 		if (messageQueue->messageFunc.IsEmpty()) {
+			for (WorkerMessage& message : messages) {
+				free(message.data);
+				delete message.transferData;
+			}
 			return;
 		}
 
@@ -94,29 +204,63 @@ namespace {
 		Local<Context> current_context = Local<Context>::New(isolate, context);
 		Context::Scope context_scope(current_context);
 
-		kinc_mutex_lock(&messageQueue->messageMutex);
+		for (WorkerMessage& message : messages) {
+			DeserializerDelegate delegate(message.transferData);
+			ValueDeserializer deserializer(isolate, message.data, message.size, &delegate);
+
+			Maybe<bool> headerResult = deserializer.ReadHeader(current_context);
+			if (headerResult.IsNothing()) {
+				free(message.data);
+				delete message.transferData;
+				continue;
+			}
+
+			MaybeLocal<Value> value = deserializer.ReadValue(current_context);
+
+			free(message.data);
+
+			if (value.IsEmpty()) {
+				delete message.transferData;
+				continue;
+			}
 
-		for (WorkerMessage& message : messageQueue->messages) {
-			MaybeLocal<Value> value = JSON::Parse(current_context, String::NewFromUtf8(isolate, message.message, NewStringType::kNormal, message.length).ToLocalChecked());
-			delete[] message.message;
 			Local<Object> object = Object::New(isolate);
 			object->Set(current_context, String::NewFromUtf8(isolate, "data").ToLocalChecked(), value.ToLocalChecked());
 			Local<Value> args[] = { object };
 
 			Local<Function> message_func = messageQueue->messageFunc.Get(isolate);
-			message_func->Call(current_context, current_context->Global(), 1, args);
+			TryCatch try_catch(isolate);
+			Local<Value> callResult;
+			if (!message_func->Call(current_context, current_context->Global(), 1, args).ToLocal(&callResult)) {
+				handle_exception(isolate, &try_catch);
+			} 
+			delete message.transferData;
 		}
+	}
 
-		messageQueue->messages.clear();
-
-		kinc_mutex_unlock(&messageQueue->messageMutex);
+	void worker_console_log(const FunctionCallbackInfo<Value>& args) {
+		Isolate* isolate = args.GetIsolate();
+		HandleScope scope(isolate);
+		std::string output;
+		for (int i = 0; i < args.Length(); ++i) {
+			if (i > 0) output += " ";
+			String::Utf8Value utf8(isolate, args[i]);
+			output += *utf8;
+		}
+		kinc_log(KINC_LOG_LEVEL_INFO, "%s", output.c_str());
 	}
 
 	void handle_exception(Isolate* isolate, TryCatch* try_catch) {
 		MaybeLocal<Value> trace = try_catch->StackTrace(isolate->GetCurrentContext());
 		if (trace.IsEmpty()) {
-			v8::String::Utf8Value stack_trace(isolate, try_catch->Message()->Get());
-			kinc_log(KINC_LOG_LEVEL_ERROR, "uncaught exception %s", *stack_trace);
+			MaybeLocal<v8::Message> message = try_catch->Message();
+			if (message.IsEmpty()) {
+				kinc_log(KINC_LOG_LEVEL_ERROR, "uncaught exception (no message available)");
+			}
+			else {
+				v8::String::Utf8Value stack_trace(isolate, message.ToLocalChecked()->Get());
+				kinc_log(KINC_LOG_LEVEL_ERROR, "uncaught exception %s", *stack_trace);
+			}
 		}
 		else {
 			v8::String::Utf8Value stack_trace(isolate, trace.ToLocalChecked());
@@ -130,9 +274,6 @@ namespace {
 		if (args.Length() < 1) {
 			return;
 		}
-		if (args.Length() > 1) {
-			kinc_log(KINC_LOG_LEVEL_WARNING, "RunT workers only support 1 argument for postMessage");
-		}
 
 		Local<External> external = Local<External>::Cast(args.Data());
 		WorkerMessagePort* messagePort = (WorkerMessagePort*)external->Value();
@@ -182,41 +323,60 @@ namespace {
 		set_onmessage(&messagePort->ownerMessages, isolate, args[1]);
 	}
 
-	void worker_set_interval(const FunctionCallbackInfo<Value>& args) {
-		Isolate* isolate = args.GetIsolate();
-		HandleScope scope(isolate);
-
-		Local<External> external = Local<External>::Cast(args.Data());
-		IntervalFunctions* interval_functions = (IntervalFunctions*)external->Value();
-
+	int add_timer(IntervalFunctions* interval_functions, Isolate* isolate, const FunctionCallbackInfo<Value>& args, bool is_timeout) {
+		if (args.Length() < 1 || !args[0]->IsFunction()) {
+			isolate->ThrowException(String::NewFromUtf8(isolate, "Timer callback must be a function").ToLocalChecked());
+			return -1;
+		}
 		Local<Function> function = Local<Function>::Cast(args[0]);
 
-		double frequency;
+		double delay;
 		if (args.Length() > 1) {
-			frequency = Local<Number>::Cast(args[1])->Value();
-			frequency /= 1000.0;
+			if (!args[1]->IsNumber()) {
+				isolate->ThrowException(String::NewFromUtf8(isolate, "Timer delay must be a number").ToLocalChecked());
+				return -1;
+			}
+			delay = Local<Number>::Cast(args[1])->Value() / 1000.0;
 		}
 		else {
-			frequency = 0.016;
+			delay = is_timeout ? 0.0 : 0.016;
 		}
 
 		interval_functions->functions.emplace_back();
-		IntervalFunction& interval_function = interval_functions->functions[interval_functions->functions.size() - 1];
+		IntervalFunction& interval_function = interval_functions->functions.back();
 		interval_function.function.Reset(isolate, function);
-		interval_function.interval = frequency;
-		interval_function.nextCallTime = kinc_time() + frequency;
+		interval_function.interval = delay;
+		interval_function.nextCallTime = kinc_time() + delay;
 		interval_function.id = interval_functions->latestId;
+		interval_function.isTimeout = is_timeout;
 
 		interval_functions->latestId++;
-
-		args.GetReturnValue().Set(interval_function.id);
+		return interval_function.id;
 	}
 
-	void worker_clear_interval(const FunctionCallbackInfo<Value>& args) {
+	void worker_set_interval(const FunctionCallbackInfo<Value>& args) {
+		HandleScope scope(args.GetIsolate());
+		IntervalFunctions* interval_functions = (IntervalFunctions*)Local<External>::Cast(args.Data())->Value();
+		args.GetReturnValue().Set(add_timer(interval_functions, args.GetIsolate(), args, false));
+	}
+
+	void worker_set_timeout(const FunctionCallbackInfo<Value>& args) {
+		HandleScope scope(args.GetIsolate());
+		IntervalFunctions* interval_functions = (IntervalFunctions*)Local<External>::Cast(args.Data())->Value();
+		args.GetReturnValue().Set(add_timer(interval_functions, args.GetIsolate(), args, true));
+	}
+
+	void worker_clear_timer(const FunctionCallbackInfo<Value>& args) {
+		Isolate* isolate = args.GetIsolate();
+		HandleScope scope(isolate);
+		if (args.Length() < 1 || !args[0]->IsInt32()) {
+			return;
+		}
+
 		Local<External> external = Local<External>::Cast(args.Data());
 		IntervalFunctions* intervalFunctions = (IntervalFunctions*)external->Value();
 
-		int id = Local<Int32>::Cast(args[0])->Value();
+		int id = args[0]->Int32Value(isolate->GetCurrentContext()).ToChecked();
 
 		for (std::vector<IntervalFunction>::iterator it = intervalFunctions->functions.begin(); it != intervalFunctions->functions.end(); ++it) {
 			if (it->id == id) {
@@ -232,8 +392,10 @@ namespace {
 		WorkerMessagePort* message_port = worker_data->messagePort;
 
 		Isolate::CreateParams create_params;
-		create_params.array_buffer_allocator = v8::ArrayBuffer::Allocator::NewDefaultAllocator();
+		auto array_buffer_allocator = v8::ArrayBuffer::Allocator::NewDefaultAllocator();
+		create_params.array_buffer_allocator = array_buffer_allocator;
 		Isolate* isolate = Isolate::New(create_params);
+		{
 		Locker locker(isolate);
 		Isolate::Scope isolate_scope(isolate);
 		HandleScope handle_scope(isolate);
@@ -250,13 +412,34 @@ namespace {
 		Local<External> extern_interval_functions = External::New(isolate, &interval_functions);
 
 		global->Set(String::NewFromUtf8(isolate, "setInterval").ToLocalChecked(), FunctionTemplate::New(isolate, worker_set_interval, extern_interval_functions));
-		global->Set(String::NewFromUtf8(isolate, "clearInterval").ToLocalChecked(), FunctionTemplate::New(isolate, worker_clear_interval, extern_interval_functions));
+		global->Set(String::NewFromUtf8(isolate, "clearInterval").ToLocalChecked(), FunctionTemplate::New(isolate, worker_clear_timer, extern_interval_functions));
+		global->Set(String::NewFromUtf8(isolate, "setTimeout").ToLocalChecked(), FunctionTemplate::New(isolate, worker_set_timeout, extern_interval_functions));
+		global->Set(String::NewFromUtf8(isolate, "clearTimeout").ToLocalChecked(), FunctionTemplate::New(isolate, worker_clear_timer, extern_interval_functions));
+
+		Local<ObjectTemplate> console_template = ObjectTemplate::New(isolate);
+		console_template->Set(String::NewFromUtf8(isolate, "log").ToLocalChecked(), FunctionTemplate::New(isolate, worker_console_log));
+		console_template->Set(String::NewFromUtf8(isolate, "warn").ToLocalChecked(), FunctionTemplate::New(isolate, worker_console_log));
+		console_template->Set(String::NewFromUtf8(isolate, "error").ToLocalChecked(), FunctionTemplate::New(isolate, worker_console_log));
+		console_template->Set(String::NewFromUtf8(isolate, "info").ToLocalChecked(), FunctionTemplate::New(isolate, worker_console_log));
+		console_template->Set(String::NewFromUtf8(isolate, "debug").ToLocalChecked(), FunctionTemplate::New(isolate, worker_console_log));
+		global->Set(String::NewFromUtf8(isolate, "console").ToLocalChecked(), console_template);
 
 		Local<Context> context = Context::New(isolate, nullptr, global);
 		Global<Context> global_context;
 		global_context.Reset(isolate, context);
 		Context::Scope context_scope(context);
 
+		Local<Object> global_obj = context->Global();
+		global_obj->Set(context, String::NewFromUtf8(isolate, "self").ToLocalChecked(), global_obj).ToChecked();
+		global_obj->Set(context, String::NewFromUtf8(isolate, "WorkerGlobalScope").ToLocalChecked(), global_obj).ToChecked();
+		global_obj->Set(context, String::NewFromUtf8(isolate, "globalThis").ToLocalChecked(), global_obj).ToChecked();
+
+		Local<Object> navigator = Object::New(isolate);
+		navigator->Set(context, String::NewFromUtf8(isolate, "hardwareConcurrency").ToLocalChecked(), Integer::New(isolate, kinc_hardware_threads())).ToChecked();
+		navigator->Set(context, String::NewFromUtf8(isolate, "userAgent").ToLocalChecked(), String::NewFromUtf8(isolate, "RunT/1.0").ToLocalChecked()).ToChecked();
+		navigator->Set(context, String::NewFromUtf8(isolate, "language").ToLocalChecked(), String::NewFromUtf8(isolate, "en-US").ToLocalChecked()).ToChecked();
+		global_obj->Set(context, String::NewFromUtf8(isolate, "navigator").ToLocalChecked(), navigator).ToChecked();
+
 		bind_worker_class(isolate, global_context);
 
 		kinc_file_reader_t reader;
@@ -274,46 +457,105 @@ namespace {
 		Local<String> source = String::NewFromUtf8(isolate, code, NewStringType::kNormal).ToLocalChecked();
 
 		TryCatch try_catch(isolate);
-		Local<Script> compiled_script = Script::Compile(isolate->GetCurrentContext(), source).ToLocalChecked();
-
-		Local<Value> result;
-		if (!compiled_script->Run(context).ToLocal(&result)) {
+		Local<Script> compiled_script;
+		if (!Script::Compile(isolate->GetCurrentContext(), source).ToLocal(&compiled_script)) {
 			handle_exception(isolate, &try_catch);
 		}
+		else {
+			Local<Value> result;
+			if (!compiled_script->Run(context).ToLocal(&result)) {
+				handle_exception(isolate, &try_catch);
+			}
+			MicrotasksScope::PerformCheckpoint(isolate);
+		}
 
-		while (!message_port->isTerminated) {
+		std::vector<int> due_ids;
+		due_ids.reserve(8);
+		while (!message_port->isTerminated.load()) {
+			HandleScope loop_scope(isolate);
 			double time = kinc_time();
-			for (int i = 0; i < interval_functions.functions.size(); ++i) {
-				if (interval_functions.functions[i].nextCallTime <= time) {
-					TryCatch try_catch(isolate);
-					Local<v8::Function> func = Local<v8::Function>::New(isolate, interval_functions.functions[i].function);
-					Local<Value> result;
+			bool had_work = false;
+			due_ids.clear();
+			for (const IntervalFunction& fn : interval_functions.functions) {
+				if (fn.nextCallTime <= time) {
+					due_ids.push_back(fn.id);
+				}
+			}
 
-					if (!func->Call(context, context->Global(), 0, NULL).ToLocal(&result)) {
-						handle_exception(isolate, &try_catch);
+			for (int due_id : due_ids) {
+				int idx = -1;
+				for (int j = 0; j < (int)interval_functions.functions.size(); ++j) {
+					if (interval_functions.functions[j].id == due_id) {
+						idx = j;
+						break;
 					}
+				}
+				if (idx < 0) {
+					continue;
+				}
 
-					interval_functions.functions[i].nextCallTime = time + interval_functions.functions[i].interval;
+				had_work = true;
+				IntervalFunction& fn = interval_functions.functions[idx];
+				Local<v8::Function> func = Local<v8::Function>::New(isolate, fn.function);
+
+				if (fn.isTimeout) {
+					fn.function.Reset();
+					interval_functions.functions.erase(interval_functions.functions.begin() + idx);
+				}
+				else {
+					fn.nextCallTime = time + fn.interval;
+				}
+
+				TryCatch try_catch(isolate);
+				Local<Value> result;
+				if (!func->Call(context, context->Global(), 0, NULL).ToLocal(&result)) {
+					handle_exception(isolate, &try_catch);
 				}
 			}
 
 			handle_message_queue(isolate, global_context, &message_port->ownerMessages);
+			MicrotasksScope::PerformCheckpoint(isolate);
 
-			handle_worker_messages(isolate, global_context);
+			if (!had_work && !message_port->isTerminated.load()) {
+				double timeout = 1.0;
+				for (const IntervalFunction& fn : interval_functions.functions) {
+					double dt = fn.nextCallTime - time;
+					if (dt < timeout) {
+						timeout = dt;
+					}
+				}
+				if (timeout < 0.0) {
+					timeout = 0.0;
+				}
+				kinc_event_try_to_wait(&message_port->ownerMessages.event, timeout);
+			}
 		}
 
+
 		ContextData* context_data = (ContextData*)isolate->GetData(worker_data_slot);
 
 		for (int i = 0; i < context_data->workers.size(); ++i) {
 			WorkerMessagePort* workerPort = context_data->workers[i].workerMessagePort;
-			workerPort->isTerminated = true;
+			workerPort->isTerminated.store(true);
+		
+
+			kinc_event_signal(&workerPort->ownerMessages.event);
+
 			kinc_thread_wait_and_destroy(context_data->workers[i].workerThread);
 
 			delete context_data->workers[i].workerThread;
-			kinc_mutex_destroy(&context_data->workers[i].workerMessagePort->ownerMessages.messageMutex);
-			kinc_mutex_destroy(&context_data->workers[i].workerMessagePort->workerMessages.messageMutex);
+			for (auto& msg : workerPort->ownerMessages.messages) {
+				free(msg.data);
+				delete msg.transferData;
+			}
+			for (auto& msg : workerPort->workerMessages.messages) {
+				free(msg.data);
+				delete msg.transferData;
+			}
 			kinc_mutex_destroy(&workerPort->ownerMessages.messageMutex);
 			kinc_mutex_destroy(&workerPort->workerMessages.messageMutex);
+			kinc_event_destroy(&workerPort->ownerMessages.event);
+			kinc_event_destroy(&workerPort->workerMessages.event);
 			delete workerPort;
 		}
 
@@ -321,6 +563,9 @@ namespace {
 		delete worker_data;
 		delete context_data;
 		isolate->TerminateExecution();
+		}
+		isolate->Dispose();
+		delete array_buffer_allocator;
 	}
 
 	void owner_onmessage_get(Local<String> property,const PropertyCallbackInfo<Value>& info) {
@@ -373,9 +618,6 @@ namespace {
 		if (args.Length() < 1) {
 			return;
 		}
-		if (args.Length() > 1) {
-			kinc_log(KINC_LOG_LEVEL_WARNING, "RunT workers only support 1 argument for postMessage");
-		}
 
 		Local<External> external = Local<External>::Cast(args.This()->GetInternalField(0));
 		WorkerMessagePort* messagePort = (WorkerMessagePort*)external->Value();
@@ -390,7 +632,8 @@ namespace {
 		Local<External> external = Local<External>::Cast(args.This()->GetInternalField(0));
 		WorkerMessagePort* messagePort = (WorkerMessagePort*)external->Value();
 
-		messagePort->isTerminated = true;
+		messagePort->isTerminated.store(true);
+		kinc_event_signal(&messagePort->ownerMessages.event);
 
 		ContextData* context_data = (ContextData*)isolate->GetData(worker_data_slot);
 		for (std::vector<OwnedWorker>::iterator it = context_data->workers.begin(); it != context_data->workers.end(); ++it) {
@@ -401,8 +644,18 @@ namespace {
 			}
 		}
 
+		for (auto& msg : messagePort->ownerMessages.messages) {
+			free(msg.data);
+			delete msg.transferData;
+		}
+		for (auto& msg : messagePort->workerMessages.messages) {
+			free(msg.data);
+			delete msg.transferData;
+		}
 		kinc_mutex_destroy(&messagePort->ownerMessages.messageMutex);
 		kinc_mutex_destroy(&messagePort->workerMessages.messageMutex);
+		kinc_event_destroy(&messagePort->ownerMessages.event);
+		kinc_event_destroy(&messagePort->workerMessages.event);
 		delete messagePort;
 
 		args.This()->SetInternalField(0, External::New(isolate, nullptr));
@@ -421,13 +674,13 @@ namespace {
 			return;
 		}
 
-		if (args.Length() > 1) {
-			kinc_log(KINC_LOG_LEVEL_WARNING, "RunT only supports one argument for worker constructor, ignoring extra arguments");
-		}
-
 		WorkerMessagePort* messagePort = new WorkerMessagePort;
+		messagePort->ownerMessages.label = "ownerMessages(main->worker)";
+		messagePort->workerMessages.label = "workerMessages(worker->main)";
 		kinc_mutex_init(&messagePort->ownerMessages.messageMutex);
 		kinc_mutex_init(&messagePort->workerMessages.messageMutex);
+		kinc_event_init(&messagePort->ownerMessages.event, true);
+		kinc_event_init(&messagePort->workerMessages.event, true);
 
 		Local<Context> context = isolate->GetCurrentContext();
 
@@ -452,14 +705,22 @@ namespace {
 		worker_data->fileName[length] = 0;
 
 		// Check if file exists before we create thread
-		kinc_file_reader fileCheck;
+		kinc_file_reader_t fileCheck;
 		if (!kinc_file_reader_open(&fileCheck, worker_data->fileName, KINC_FILE_TYPE_ASSET)) {
 			const char* errorFmt = "Worker constructor: file %s does not exist";
 			int length = snprintf(nullptr, 0, errorFmt, worker_data->fileName);
 			char* errorMessage = (char*)malloc(length + 1);
-			snprintf(errorMessage, length, errorFmt, worker_data->fileName);
+			snprintf(errorMessage, length + 1, errorFmt, worker_data->fileName);
 			isolate->ThrowException(String::NewFromUtf8(isolate, errorMessage).ToLocalChecked());
 			free(errorMessage);
+			delete owned_worker.workerThread;
+			delete worker_data;
+			kinc_mutex_destroy(&messagePort->ownerMessages.messageMutex);
+			kinc_mutex_destroy(&messagePort->workerMessages.messageMutex);
+			kinc_event_destroy(&messagePort->ownerMessages.event);
+			kinc_event_destroy(&messagePort->workerMessages.event);
+			delete messagePort;
+			args.This()->SetInternalField(0, External::New(isolate, nullptr));
 			return;
 		}
 		kinc_file_reader_close(&fileCheck);