MultiThreading

Multi-Threading

In NPL, each thread is explicitly created and explicitly named. NPL thread is persistent and reused during the lifetime of a NPL process.

In NPL, multi-threading is handled in the same way as networking communication. In other words, activating scripts in other local threads is virtually the same as calling scripts running on another computer. You simply communicate with remote script file with the NPL.activate in the same way for both local threads and remote computers. The only difference is that the target script url is different.

For example, to activate a script on local thread A, one can use

NPL.activate("(A)helloworld.lua", {});

To activate the script in a remote computer B's C thread, one can use

NPL.activate("(C)B:helloworld.lua", {});

For more information, please see file activation

Creating NPL Worker Thread

NPL worker thread must be created, before messages to it can be processed by script running in that thread.

NPL.activate("(A)helloworld.lua", {}); does not automatically create thread A. You need to call following code to create NPL runtime on thread A.

NPL.CreateRuntimeState("A", 0):Start();

After that, messages sent to (A)helloworld.lua will be processed in a real system-level thread called A.

Example Project

Now let us create a real multi-threaded application with just a single file script/test/TestMultithread.lua. The application print helloworld in 5 threads simultaneously.

NPL.load("(gl)script/ide/commonlib.lua");

local function Start()
   for i=1, 5 do
      local thead_name = "T"..i;
      NPL.CreateRuntimeState(thead_name, 0):Start();
      NPL.activate(format("(%s)script/test/TestMultithread.lua", thead_name), {
	   text = "hello world", 
	   sleep_time = math.random()*5,
      });
   end
end

local isStarted;
local function activate()
   if(msg and msg.text) then
      -- sleep random seconds to simulate heavy task
      ParaEngine.Sleep(msg.sleep_time);
      LOG.std(nil, "info", "MultiThread", "%s from thread %s", msg.text,  __rts__:GetName());
   elseif(not isStarted) then
      -- initialize on first call 
      isStarted = true;
      Start();
   end
end

NPL.this(activate);

To run above file, use

NPL.activate("(gl)script/test/TestMultithread.lua");

or from command line

npl script/test/TestMultithread.lua

The output will be something like below

2016-03-16 6:22:00 PM|T1|info|MultiThread|hello world from thread T1
2016-03-16 6:22:01 PM|T3|info|MultiThread|hello world from thread T3
2016-03-16 6:22:03 PM|T2|info|MultiThread|hello world from thread T2
2016-03-16 6:22:03 PM|T5|info|MultiThread|hello world from thread T5
2016-03-16 6:22:04 PM|T4|info|MultiThread|hello world from thread T4

Advanced Examples

Following NPL modules utilize multiple local threads.

• script/ide/System/Database/TableDatabase.lua: a data base server that routes all requests via IO thread to distribute load to a number of worker threads.
• script/apps/DBServer/DBServer.lua: a database server, each thread for processing SQL logics, a thread monitor is used to find the most free thread to route any sql query.

Helper class: Async RPC

rpc class provides a way to easily create a function to be run in any local threads using virtual NPL files internally.

This is similar to NPL.activate() and NPL.this(function() end), but without the need to explicitly use a file.

NPL.load("(gl)script/ide/System/Concurrent/rpc.lua");
local rpc = commonlib.gettable("System.Concurrent.Async.rpc");
-- the third parameter is usually a file that contains the rpc init code (__rts__:GetField("filename", "") will do that)
-- you can also explicitly specify a central API file where all RPC are defined, such as "XXX_API.lua"
rpc:new():init("Test.testRPC", function(self, msg) 
	LOG.std(nil, "info", "category", msg);
	msg.output=true; 
	ParaEngine.Sleep(1);
	return msg; 
end, __rts__:GetField("filename", ""))

Test.testRPC:MakePublic();

-- now we can invoke it anywhere in any thread or remote address.
Test.testRPC("", {"input"}, function(err, msg) 
	assert(msg.output == true and msg[1] == "input")
	echo(msg);
end);

-- time out in 500ms, this will automatically create `worker1` NPL thread
Test.testRPC("(worker1)", {"input"}, function(err, msg) 
	assert(err == "timeout" and msg==nil)
	echo(err);
end, 500);

by default, rpc can only be called from threads in the current NPL process. in order to expose the API via NPL tcp protocol, one needs to call rpc:MakePublic() as in above example. Test.testRPC can be invoked in any computer, provided the class is defined and nid is specified.

The first parameter to Test.testRPC is RPC address. if nil, it is current NPL thread. it can be thread name like "(worker1)" if NPL thread worker1 is not created, it will be automatically created. Because NPL thread is reused, it is good practice to use only limited number of NPL threads per process. For complete format, please see NPL.activate function.

For example, if you want to randomly distribute workload to 5 local threads, simple do this:

-- this will automatically create `workerX` NPL threads
for i=1,10 do
   Test.testRPC(format("(worker%d)", math.random(1,5)), {"input"}, function(err, msg) 
      echo(msg);
   end, 5000);
end

Because NPL threads are not destroyed when request is finished, one can reuse those workerX threads for large number of requests.

Helper class: AsyncTask

Async task provide an easy API to run small tasks in NPL worker threads. By default, there are 2 NPL worker threads in total. It will serialize function to string and send to worker threads. It will serialize and send function between threads only once per function. Code needs to be static because we will cache all past task functions in memory.

NPL.load("(gl)script/ide/System/Concurrent/AsyncTask.lua");
local AsyncTask = commonlib.gettable("System.Concurrent.AsyncTask");

-- input function should not use any up-values, default timeout is 10000 milliseconds
AsyncTask.RunTask(function(param1, param2)
	echo({"from worker thread", param1, param2})
	return "OK"
end, "param1", "param2"):OnFinish(function(result)
	echo("result is "..result)
end)

local task = AsyncTask.CreateTask(function(param1, param2)
	echo({"from worker thread", param1, param2})
	return "OK"
end, "param1", "param2")
-- force running on a given thread index or name
task:Run("MyThreadName"):OnFinish(function(result)
	echo("result is "..result)
end):OnError(function(errMsg, msgType)
	if(msgType == "timeout") then
		echo("task timed out")
	else
		echo("error: "..errMsg)
	end)
end)

Other Options

By design, threading should be avoided to simplify software design and debugging. In addition to real threads, it is usually preferred to use architecture to avoid using thread at all.

• Timer/callbacks/events/signals are good candidates for asynchronous tasks in a single thread. With NPL.activate, it even allows you to switch implementation without changing any code; and you can boost performance or debug code in a single thread more easily.
• Coroutine is a lua language feature, which is also supported by NPL. In short, it uses a single thread to simulate multiple virtual threads, allowing you to share all data in the same thread without using locks, but still allowing the developer to yield CPU resource to other virtual threads at any time. The interactive debugging module in NPL is implemented with coroutines. Please see script/ide/Debugger/IPCDebugger.lua for details.