oss/luanti
oss/luanti
1
0
Fork 0
mirror of https://github.com/luanti-org/luanti.git synced 2025-06-27 16:36:03 +00:00
Code Wiki Activity

Async environment for mods to do concurrent tasks (#11131)

Browse source
This commit is contained in:
sfan5 2022-05-02 20:55:04 +02:00
parent 663c936428
commit e7659883cc
38 changed files with 1646 additions and 48 deletions
Download patch file Download diff file Expand all files Collapse all files

183
src/script/cpp_api/s_async.cpp
View file

@ -21,9 +21,9 @@ with this program; if not, write to the Free Software Foundation, Inc.,
#include <cstdlib>
extern "C" {
#include "lua.h"
#include "lauxlib.h"
#include "lualib.h"
#include <lua.h>
#include <lauxlib.h>
#include <lualib.h>
}
#include "server.h"
@ -32,6 +32,7 @@ extern "C" {
#include "filesys.h"
#include "porting.h"
#include "common/c_internal.h"
#include "common/c_packer.h"
#include "lua_api/l_base.h"
/******************************************************************************/
@ -76,19 +77,34 @@ void AsyncEngine::initialize(unsigned int numEngines)
{
initDone = true;
for (unsigned int i = 0; i < numEngines; i++) {
AsyncWorkerThread *toAdd = new AsyncWorkerThread(this,
std::string("AsyncWorker-") + itos(i));
workerThreads.push_back(toAdd);
toAdd->start();
if (numEngines == 0) {
// Leave one core for the main thread and one for whatever else
autoscaleMaxWorkers = Thread::getNumberOfProcessors();
if (autoscaleMaxWorkers >= 2)
autoscaleMaxWorkers -= 2;
infostream << "AsyncEngine: using at most " << autoscaleMaxWorkers
<< " threads with automatic scaling" << std::endl;
addWorkerThread();
} else {
for (unsigned int i = 0; i < numEngines; i++)
addWorkerThread();
}
}
void AsyncEngine::addWorkerThread()
{
AsyncWorkerThread *toAdd = new AsyncWorkerThread(this,
std::string("AsyncWorker-") + itos(workerThreads.size()));
workerThreads.push_back(toAdd);
toAdd->start();
}
/******************************************************************************/
u32 AsyncEngine::queueAsyncJob(std::string &&func, std::string &&params,
const std::string &mod_origin)
{
jobQueueMutex.lock();
MutexAutoLock autolock(jobQueueMutex);
u32 jobId = jobIdCounter++;
jobQueue.emplace_back();
@ -99,7 +115,23 @@ u32 AsyncEngine::queueAsyncJob(std::string &&func, std::string &&params,
to_add.mod_origin = mod_origin;
jobQueueCounter.post();
jobQueueMutex.unlock();
return jobId;
}
u32 AsyncEngine::queueAsyncJob(std::string &&func, PackedValue *params,
const std::string &mod_origin)
{
MutexAutoLock autolock(jobQueueMutex);
u32 jobId = jobIdCounter++;
jobQueue.emplace_back();
auto &to_add = jobQueue.back();
to_add.id = jobId;
to_add.function = std::move(func);
to_add.params_ext.reset(params);
to_add.mod_origin = mod_origin;
jobQueueCounter.post();
return jobId;
}
@ -131,6 +163,12 @@ void AsyncEngine::putJobResult(LuaJobInfo &&result)
/******************************************************************************/
void AsyncEngine::step(lua_State *L)
{
stepJobResults(L);
stepAutoscale();
}
void AsyncEngine::stepJobResults(lua_State *L)
{
int error_handler = PUSH_ERROR_HANDLER(L);
lua_getglobal(L, "core");
@ -148,7 +186,10 @@ void AsyncEngine::step(lua_State *L)
luaL_checktype(L, -1, LUA_TFUNCTION);
lua_pushinteger(L, j.id);
lua_pushlstring(L, j.result.data(), j.result.size());
if (j.result_ext)
script_unpack(L, j.result_ext.get());
else
lua_pushlstring(L, j.result.data(), j.result.size());
// Call handler
const char *origin = j.mod_origin.empty() ? nullptr : j.mod_origin.c_str();
@ -161,12 +202,71 @@ void AsyncEngine::step(lua_State *L)
lua_pop(L, 2); // Pop core and error handler
}
void AsyncEngine::stepAutoscale()
{
if (workerThreads.size() >= autoscaleMaxWorkers)
return;
MutexAutoLock autolock(jobQueueMutex);
// 2) If the timer elapsed, check again
if (autoscaleTimer && porting::getTimeMs() >= autoscaleTimer) {
autoscaleTimer = 0;
// Determine overlap with previous snapshot
unsigned int n = 0;
for (const auto &it : jobQueue)
n += autoscaleSeenJobs.count(it.id);
autoscaleSeenJobs.clear();
infostream << "AsyncEngine: " << n << " jobs were still waiting after 1s" << std::endl;
// Start this many new threads
while (workerThreads.size() < autoscaleMaxWorkers && n > 0) {
addWorkerThread();
n--;
}
return;
}
// 1) Check if there's anything in the queue
if (!autoscaleTimer && !jobQueue.empty()) {
// Take a snapshot of all jobs we have seen
for (const auto &it : jobQueue)
autoscaleSeenJobs.emplace(it.id);
// and set a timer for 1 second
autoscaleTimer = porting::getTimeMs() + 1000;
}
}
/******************************************************************************/
void AsyncEngine::prepareEnvironment(lua_State* L, int top)
bool AsyncEngine::prepareEnvironment(lua_State* L, int top)
{
for (StateInitializer &stateInitializer : stateInitializers) {
stateInitializer(L, top);
}
auto *script = ModApiBase::getScriptApiBase(L);
try {
script->loadMod(Server::getBuiltinLuaPath() + DIR_DELIM + "init.lua",
BUILTIN_MOD_NAME);
} catch (const ModError &e) {
errorstream << "Execution of async base environment failed: "
<< e.what() << std::endl;
FATAL_ERROR("Execution of async base environment failed");
}
// Load per mod stuff
if (server) {
const auto &list = server->m_async_init_files;
try {
for (auto &it : list)
script->loadMod(it.second, it.first);
} catch (const ModError &e) {
errorstream << "Failed to load mod script inside async environment." << std::endl;
server->setAsyncFatalError(e.what());
return false;
}
}
return true;
}
/******************************************************************************/
@ -178,15 +278,25 @@ AsyncWorkerThread::AsyncWorkerThread(AsyncEngine* jobDispatcher,
{
lua_State *L = getStack();
if (jobDispatcher->server) {
setGameDef(jobDispatcher->server);
if (g_settings->getBool("secure.enable_security"))
initializeSecurity();
}
// Prepare job lua environment
lua_getglobal(L, "core");
int top = lua_gettop(L);
// Push builtin initialization type
lua_pushstring(L, "async");
lua_pushstring(L, jobDispatcher->server ? "async_game" : "async");
lua_setglobal(L, "INIT");
jobDispatcher->prepareEnvironment(L, top);
if (!jobDispatcher->prepareEnvironment(L, top)) {
// can't throw from here so we're stuck with this
isErrored = true;
}
}
/******************************************************************************/
@ -198,19 +308,20 @@ AsyncWorkerThread::~AsyncWorkerThread()
/******************************************************************************/
void* AsyncWorkerThread::run()
{
if (isErrored)
return nullptr;
lua_State *L = getStack();
try {
loadMod(getServer()->getBuiltinLuaPath() + DIR_DELIM + "init.lua",
BUILTIN_MOD_NAME);
} catch (const ModError &e) {
errorstream << "Execution of async base environment failed: "
<< e.what() << std::endl;
FATAL_ERROR("Execution of async base environment failed");
}
int error_handler = PUSH_ERROR_HANDLER(L);
auto report_error = [this] (const ModError &e) {
if (jobDispatcher->server)
jobDispatcher->server->setAsyncFatalError(e.what());
else
errorstream << e.what() << std::endl;
};
lua_getglobal(L, "core");
if (lua_isnil(L, -1)) {
FATAL_ERROR("Unable to find core within async environment!");
@ -223,6 +334,8 @@ void* AsyncWorkerThread::run()
if (!jobDispatcher->getJob(&j) || stopRequested())
continue;
const bool use_ext = !!j.params_ext;
lua_getfield(L, -1, "job_processor");
if (lua_isnil(L, -1))
FATAL_ERROR("Unable to get async job processor!");
@ -232,7 +345,10 @@ void* AsyncWorkerThread::run()
errorstream << "ASYNC WORKER: Unable to deserialize function" << std::endl;
lua_pushnil(L);
}
lua_pushlstring(L, j.params.data(), j.params.size());
if (use_ext)
script_unpack(L, j.params_ext.get());
else
lua_pushlstring(L, j.params.data(), j.params.size());
// Call it
setOriginDirect(j.mod_origin.empty() ? nullptr : j.mod_origin.c_str());
@ -241,19 +357,28 @@ void* AsyncWorkerThread::run()
try {
scriptError(result, "<async>");
} catch (const ModError &e) {
errorstream << e.what() << std::endl;
report_error(e);
}
} else {
// Fetch result
size_t length;
const char *retval = lua_tolstring(L, -1, &length);
j.result.assign(retval, length);
if (use_ext) {
try {
j.result_ext.reset(script_pack(L, -1));
} catch (const ModError &e) {
report_error(e);
result = LUA_ERRERR;
}
} else {
size_t length;
const char *retval = lua_tolstring(L, -1, &length);
j.result.assign(retval, length);
}
}
lua_pop(L, 1); // Pop retval
// Put job result
if (!j.result.empty())
if (result == 0)
jobDispatcher->putJobResult(std::move(j));
}

54
src/script/cpp_api/s_async.h
View file

@ -21,11 +21,15 @@ with this program; if not, write to the Free Software Foundation, Inc.,
#include <vector>
#include <deque>
#include <unordered_set>
#include <memory>
#include <lua.h>
#include "threading/semaphore.h"
#include "threading/thread.h"
#include "lua.h"
#include "common/c_packer.h"
#include "cpp_api/s_base.h"
#include "cpp_api/s_security.h"
// Forward declarations
class AsyncEngine;
@ -42,8 +46,12 @@ struct LuaJobInfo
std::string function;
// Parameter to be passed to function (serialized)
std::string params;
// Alternative parameters
std::unique_ptr<PackedValue> params_ext;
// Result of function call (serialized)
std::string result;
// Alternative result
std::unique_ptr<PackedValue> result_ext;
// Name of the mod who invoked this call
std::string mod_origin;
// JobID used to identify a job and match it to callback
@ -51,7 +59,8 @@ struct LuaJobInfo
};
// Asynchronous working environment
class AsyncWorkerThread : public Thread, virtual public ScriptApiBase {
class AsyncWorkerThread : public Thread,
virtual public ScriptApiBase, public ScriptApiSecurity {
friend class AsyncEngine;
public:
virtual ~AsyncWorkerThread();
@ -63,6 +72,7 @@ protected:
private:
AsyncEngine *jobDispatcher = nullptr;
bool isErrored = false;
};
// Asynchornous thread and job management
@ -71,6 +81,7 @@ class AsyncEngine {
typedef void (*StateInitializer)(lua_State *L, int top);
public:
AsyncEngine() = default;
AsyncEngine(Server *server) : server(server) {};
~AsyncEngine();
/**
@ -81,7 +92,7 @@ public:
/**
* Create async engine tasks and lock function registration
* @param numEngines Number of async threads to be started
* @param numEngines Number of worker threads, 0 for automatic scaling
*/
void initialize(unsigned int numEngines);
@ -94,9 +105,17 @@ public:
u32 queueAsyncJob(std::string &&func, std::string &&params,
const std::string &mod_origin = "");
/**
* Queue an async job
* @param func Serialized lua function
* @param params Serialized parameters (takes ownership!)
* @return ID of queued job
*/
u32 queueAsyncJob(std::string &&func, PackedValue *params,
const std::string &mod_origin = "");
/**
* Engine step to process finished jobs
* the engine step is one way to pass events back, PushFinishedJobs another
* @param L The Lua stack
*/
void step(lua_State *L);
@ -116,19 +135,44 @@ protected:
*/
void putJobResult(LuaJobInfo &&result);
/**
* Start an additional worker thread
*/
void addWorkerThread();
/**
* Process finished jobs callbacks
*/
void stepJobResults(lua_State *L);
/**
* Handle automatic scaling of worker threads
*/
void stepAutoscale();
/**
* Initialize environment with current registred functions
* this function adds all functions registred by registerFunction to the
* passed lua stack
* @param L Lua stack to initialize
* @param top Stack position
* @return false if a mod error ocurred
*/
void prepareEnvironment(lua_State* L, int top);
bool prepareEnvironment(lua_State* L, int top);
private:
// Variable locking the engine against further modification
bool initDone = false;
// Maximum number of worker threads for automatic scaling
// 0 if disabled
unsigned int autoscaleMaxWorkers = 0;
u64 autoscaleTimer = 0;
std::unordered_set<u32> autoscaleSeenJobs;
// Only set for the server async environment (duh)
Server *server = nullptr;
// Internal store for registred state initializers
std::vector<StateInitializer> stateInitializers;