Fix code style of async API

This commit is contained in:
ShadowNinja 2014-04-15 15:10:30 -04:00
parent 6ab3b4c838
commit f2b93b7bae
7 changed files with 130 additions and 111 deletions

@ -291,7 +291,7 @@ void GUIEngine::run()
else else
sleep_ms(25); sleep_ms(25);
m_script->Step(); m_script->step();
} }
} }
@ -563,7 +563,8 @@ void GUIEngine::stopSound(s32 handle)
} }
/******************************************************************************/ /******************************************************************************/
unsigned int GUIEngine::DoAsync(std::string serialized_fct, unsigned int GUIEngine::queueAsync(std::string serialized_func,
std::string serialized_params) { std::string serialized_params) {
return m_script->DoAsync(serialized_fct,serialized_params); return m_script->queueAsync(serialized_func, serialized_params);
} }

@ -167,7 +167,7 @@ public:
} }
/** pass async callback to scriptengine **/ /** pass async callback to scriptengine **/
unsigned int DoAsync(std::string serialized_fct,std::string serialized_params); unsigned int queueAsync(std::string serialized_fct,std::string serialized_params);
private: private:

@ -34,82 +34,102 @@ extern "C" {
/******************************************************************************/ /******************************************************************************/
AsyncEngine::AsyncEngine() : AsyncEngine::AsyncEngine() :
m_initDone(false), initDone(false),
m_JobIdCounter(0) jobIdCounter(0)
{ {
} }
/******************************************************************************/ /******************************************************************************/
AsyncEngine::~AsyncEngine() AsyncEngine::~AsyncEngine()
{ {
// Force kill all threads
for (std::vector<AsyncWorkerThread*>::iterator i = m_WorkerThreads.begin(); // Request all threads to stop
i != m_WorkerThreads.end(); i++) { for (std::vector<AsyncWorkerThread *>::iterator it = workerThreads.begin();
(*i)->Kill(); it != workerThreads.end(); it++) {
delete *i; (*it)->Stop();
} }
m_JobQueueMutex.Lock();
m_JobQueue.clear(); // Wake up all threads
m_JobQueueMutex.Unlock(); for (std::vector<AsyncWorkerThread *>::iterator it = workerThreads.begin();
m_WorkerThreads.clear(); it != workerThreads.end(); it++) {
jobQueueCounter.Post();
}
// Wait for threads to finish
for (std::vector<AsyncWorkerThread *>::iterator it = workerThreads.begin();
it != workerThreads.end(); it++) {
(*it)->Wait();
}
// Force kill all threads
for (std::vector<AsyncWorkerThread *>::iterator it = workerThreads.begin();
it != workerThreads.end(); it++) {
(*it)->Kill();
delete *it;
}
jobQueueMutex.Lock();
jobQueue.clear();
jobQueueMutex.Unlock();
workerThreads.clear();
} }
/******************************************************************************/ /******************************************************************************/
bool AsyncEngine::registerFunction(const char* name, lua_CFunction func) bool AsyncEngine::registerFunction(const char* name, lua_CFunction func)
{ {
if (m_initDone) { if (initDone) {
return false; return false;
} }
m_FunctionList[name] = func; functionList[name] = func;
return true; return true;
} }
/******************************************************************************/ /******************************************************************************/
void AsyncEngine::Initialize(unsigned int numEngines) void AsyncEngine::initialize(unsigned int numEngines)
{ {
m_initDone = true; initDone = true;
for (unsigned int i = 0; i < numEngines; i++) { for (unsigned int i = 0; i < numEngines; i++) {
AsyncWorkerThread* toAdd = new AsyncWorkerThread(this, i); AsyncWorkerThread *toAdd = new AsyncWorkerThread(this, i);
m_WorkerThreads.push_back(toAdd); workerThreads.push_back(toAdd);
toAdd->Start(); toAdd->Start();
} }
} }
/******************************************************************************/ /******************************************************************************/
unsigned int AsyncEngine::doAsyncJob(std::string func, std::string params) unsigned int AsyncEngine::queueAsyncJob(std::string func, std::string params)
{ {
m_JobQueueMutex.Lock(); jobQueueMutex.Lock();
LuaJobInfo toadd; LuaJobInfo toAdd;
toadd.JobId = m_JobIdCounter++; toAdd.id = jobIdCounter++;
toadd.serializedFunction = func; toAdd.serializedFunction = func;
toadd.serializedParams = params; toAdd.serializedParams = params;
m_JobQueue.push_back(toadd); jobQueue.push_back(toAdd);
m_JobQueueCounter.Post(); jobQueueCounter.Post();
m_JobQueueMutex.Unlock(); jobQueueMutex.Unlock();
return toadd.JobId; return toAdd.id;
} }
/******************************************************************************/ /******************************************************************************/
LuaJobInfo AsyncEngine::getJob() LuaJobInfo AsyncEngine::getJob()
{ {
m_JobQueueCounter.Wait(); jobQueueCounter.Wait();
m_JobQueueMutex.Lock(); jobQueueMutex.Lock();
LuaJobInfo retval; LuaJobInfo retval;
retval.valid = false; retval.valid = false;
if (m_JobQueue.size() != 0) { if (!jobQueue.empty()) {
retval = m_JobQueue.front(); retval = jobQueue.front();
jobQueue.pop_front();
retval.valid = true; retval.valid = true;
m_JobQueue.erase(m_JobQueue.begin());
} }
m_JobQueueMutex.Unlock(); jobQueueMutex.Unlock();
return retval; return retval;
} }
@ -117,19 +137,19 @@ LuaJobInfo AsyncEngine::getJob()
/******************************************************************************/ /******************************************************************************/
void AsyncEngine::putJobResult(LuaJobInfo result) void AsyncEngine::putJobResult(LuaJobInfo result)
{ {
m_ResultQueueMutex.Lock(); resultQueueMutex.Lock();
m_ResultQueue.push_back(result); resultQueue.push_back(result);
m_ResultQueueMutex.Unlock(); resultQueueMutex.Unlock();
} }
/******************************************************************************/ /******************************************************************************/
void AsyncEngine::Step(lua_State *L, int errorhandler) void AsyncEngine::step(lua_State *L, int errorhandler)
{ {
lua_getglobal(L, "engine"); lua_getglobal(L, "engine");
m_ResultQueueMutex.Lock(); resultQueueMutex.Lock();
while (!m_ResultQueue.empty()) { while (!resultQueue.empty()) {
LuaJobInfo jobdone = m_ResultQueue.front(); LuaJobInfo jobDone = resultQueue.front();
m_ResultQueue.erase(m_ResultQueue.begin()); resultQueue.pop_front();
lua_getfield(L, -1, "async_event_handler"); lua_getfield(L, -1, "async_event_handler");
@ -139,53 +159,54 @@ void AsyncEngine::Step(lua_State *L, int errorhandler)
luaL_checktype(L, -1, LUA_TFUNCTION); luaL_checktype(L, -1, LUA_TFUNCTION);
lua_pushinteger(L, jobdone.JobId); lua_pushinteger(L, jobDone.id);
lua_pushlstring(L, jobdone.serializedResult.c_str(), lua_pushlstring(L, jobDone.serializedResult.data(),
jobdone.serializedResult.length()); jobDone.serializedResult.size());
if (lua_pcall(L, 2, 0, errorhandler)) { if (lua_pcall(L, 2, 0, errorhandler)) {
script_error(L); script_error(L);
} }
} }
m_ResultQueueMutex.Unlock(); resultQueueMutex.Unlock();
lua_pop(L, 1); // Pop engine lua_pop(L, 1); // Pop engine
} }
/******************************************************************************/ /******************************************************************************/
void AsyncEngine::PushFinishedJobs(lua_State* L) { void AsyncEngine::pushFinishedJobs(lua_State* L) {
// Result Table // Result Table
m_ResultQueueMutex.Lock(); resultQueueMutex.Lock();
unsigned int index = 1; unsigned int index = 1;
lua_createtable(L, m_ResultQueue.size(), 0); lua_createtable(L, resultQueue.size(), 0);
int top = lua_gettop(L); int top = lua_gettop(L);
while (!m_ResultQueue.empty()) { while (!resultQueue.empty()) {
LuaJobInfo jobdone = m_ResultQueue.front(); LuaJobInfo jobDone = resultQueue.front();
m_ResultQueue.erase(m_ResultQueue.begin()); resultQueue.pop_front();
lua_createtable(L, 0, 2); // Pre-alocate space for two map fields lua_createtable(L, 0, 2); // Pre-allocate space for two map fields
int top_lvl2 = lua_gettop(L); int top_lvl2 = lua_gettop(L);
lua_pushstring(L, "jobid"); lua_pushstring(L, "jobid");
lua_pushnumber(L, jobdone.JobId); lua_pushnumber(L, jobDone.id);
lua_settable(L, top_lvl2); lua_settable(L, top_lvl2);
lua_pushstring(L, "retval"); lua_pushstring(L, "retval");
lua_pushlstring(L, jobdone.serializedResult.data(), lua_pushlstring(L, jobDone.serializedResult.data(),
jobdone.serializedResult.size()); jobDone.serializedResult.size());
lua_settable(L, top_lvl2); lua_settable(L, top_lvl2);
lua_rawseti(L, top, index++); lua_rawseti(L, top, index++);
} }
m_ResultQueueMutex.Unlock(); resultQueueMutex.Unlock();
} }
/******************************************************************************/ /******************************************************************************/
void AsyncEngine::PrepareEnvironment(lua_State* L, int top) { void AsyncEngine::prepareEnvironment(lua_State* L, int top)
for (std::map<std::string, lua_CFunction>::iterator it = m_FunctionList.begin(); {
it != m_FunctionList.end(); it++) { for (std::map<std::string, lua_CFunction>::iterator it = functionList.begin();
it != functionList.end(); it++) {
lua_pushstring(L, it->first.c_str()); lua_pushstring(L, it->first.c_str());
lua_pushcfunction(L, it->second); lua_pushcfunction(L, it->second);
lua_settable(L, top); lua_settable(L, top);
@ -196,8 +217,8 @@ void AsyncEngine::PrepareEnvironment(lua_State* L, int top) {
AsyncWorkerThread::AsyncWorkerThread(AsyncEngine* jobDispatcher, AsyncWorkerThread::AsyncWorkerThread(AsyncEngine* jobDispatcher,
unsigned int threadNum) : unsigned int threadNum) :
ScriptApiBase(), ScriptApiBase(),
m_JobDispatcher(jobDispatcher), jobDispatcher(jobDispatcher),
m_threadnum(threadNum) threadnum(threadNum)
{ {
lua_State *L = getStack(); lua_State *L = getStack();
@ -216,7 +237,7 @@ AsyncWorkerThread::AsyncWorkerThread(AsyncEngine* jobDispatcher,
(porting::path_share + DIR_DELIM + "builtin").c_str()); (porting::path_share + DIR_DELIM + "builtin").c_str());
lua_setglobal(L, "SCRIPTDIR"); lua_setglobal(L, "SCRIPTDIR");
m_JobDispatcher->PrepareEnvironment(L, top); jobDispatcher->prepareEnvironment(L, top);
} }
/******************************************************************************/ /******************************************************************************/
@ -232,7 +253,7 @@ void* AsyncWorkerThread::Thread()
// Register thread for error logging // Register thread for error logging
char number[21]; char number[21];
snprintf(number, sizeof(number), "%d", m_threadnum); snprintf(number, sizeof(number), "%d", threadnum);
log_register_thread(std::string("AsyncWorkerThread_") + number); log_register_thread(std::string("AsyncWorkerThread_") + number);
porting::setThreadName((std::string("AsyncWorkTh_") + number).c_str()); porting::setThreadName((std::string("AsyncWorkTh_") + number).c_str());
@ -251,7 +272,7 @@ void* AsyncWorkerThread::Thread()
// Main loop // Main loop
while (!StopRequested()) { while (!StopRequested()) {
// Wait for job // Wait for job
LuaJobInfo toProcess = m_JobDispatcher->getJob(); LuaJobInfo toProcess = jobDispatcher->getJob();
if (toProcess.valid == false || StopRequested()) { if (toProcess.valid == false || StopRequested()) {
continue; continue;
@ -293,7 +314,7 @@ void* AsyncWorkerThread::Thread()
lua_pop(L, 3); lua_pop(L, 3);
// Put job result // Put job result
m_JobDispatcher->putJobResult(toProcess); jobDispatcher->putJobResult(toProcess);
} }
log_deregister_thread(); log_deregister_thread();
return 0; return 0;

@ -17,10 +17,11 @@ with this program; if not, write to the Free Software Foundation, Inc.,
51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
*/ */
#ifndef L_ASYNC_EVENTS_H_ #ifndef CPP_API_ASYNC_EVENTS_HEADER
#define L_ASYNC_EVENTS_H_ #define CPP_API_ASYNC_EVENTS_HEADER
#include <vector> #include <vector>
#include <deque>
#include <map> #include <map>
#include "jthread/jthread.h" #include "jthread/jthread.h"
@ -45,7 +46,7 @@ struct LuaJobInfo {
// Result of function call // Result of function call
std::string serializedResult; std::string serializedResult;
// JobID used to identify a job and match it to callback // JobID used to identify a job and match it to callback
unsigned int JobId; unsigned int id;
bool valid; bool valid;
}; };
@ -61,13 +62,13 @@ public:
virtual ~AsyncWorkerThread(); virtual ~AsyncWorkerThread();
void* Thread(); void *Thread();
private: private:
AsyncEngine* m_JobDispatcher; AsyncEngine *jobDispatcher;
// Thread number. Used for debug output // Thread number. Used for debug output
unsigned int m_threadnum; unsigned int threadnum;
}; };
@ -89,15 +90,15 @@ public:
* Create async engine tasks and lock function registration * Create async engine tasks and lock function registration
* @param numEngines Number of async threads to be started * @param numEngines Number of async threads to be started
*/ */
void Initialize(unsigned int numEngines); void initialize(unsigned int numEngines);
/** /**
* queue/run a async job * Queue an async job
* @param func Serialized lua function * @param func Serialized lua function
* @param params Serialized parameters * @param params Serialized parameters
* @return jobid The job is queued * @return jobid The job is queued
*/ */
unsigned int doAsyncJob(std::string func, std::string params); unsigned int queueAsyncJob(std::string func, std::string params);
/** /**
* Engine step to process finished jobs * Engine step to process finished jobs
@ -105,13 +106,13 @@ public:
* @param L The Lua stack * @param L The Lua stack
* @param errorhandler Stack index of the Lua error handler * @param errorhandler Stack index of the Lua error handler
*/ */
void Step(lua_State *L, int errorhandler); void step(lua_State *L, int errorhandler);
/** /**
* Push a list of finished jobs onto the stack * Push a list of finished jobs onto the stack
* @param L The Lua stack * @param L The Lua stack
*/ */
void PushFinishedJobs(lua_State *L); void pushFinishedJobs(lua_State *L);
protected: protected:
/** /**
@ -134,38 +135,34 @@ protected:
* @param L Lua stack to initialize * @param L Lua stack to initialize
* @param top Stack position * @param top Stack position
*/ */
void PrepareEnvironment(lua_State* L, int top); void prepareEnvironment(lua_State* L, int top);
private: private:
// Variable locking the engine against further modification
// Stack index of error handler bool initDone;
int m_errorhandler;
// variable locking the engine against further modification
bool m_initDone;
// Internal store for registred functions // Internal store for registred functions
std::map<std::string, lua_CFunction> m_FunctionList; std::map<std::string, lua_CFunction> functionList;
// Internal counter to create job IDs // Internal counter to create job IDs
unsigned int m_JobIdCounter; unsigned int jobIdCounter;
// Mutex to protect job queue // Mutex to protect job queue
JMutex m_JobQueueMutex; JMutex jobQueueMutex;
// Job queue // Job queue
std::vector<LuaJobInfo> m_JobQueue; std::deque<LuaJobInfo> jobQueue;
// Mutex to protect result queue // Mutex to protect result queue
JMutex m_ResultQueueMutex; JMutex resultQueueMutex;
// Result queue // Result queue
std::vector<LuaJobInfo> m_ResultQueue; std::deque<LuaJobInfo> resultQueue;
// List of current worker threads // List of current worker threads
std::vector<AsyncWorkerThread*> m_WorkerThreads; std::vector<AsyncWorkerThread*> workerThreads;
// Counter semaphore for job dispatching // Counter semaphore for job dispatching
JSemaphore m_JobQueueCounter; JSemaphore jobQueueCounter;
}; };
#endif // L_ASYNC_EVENTS_H_ #endif // CPP_API_ASYNC_EVENTS_HEADER

@ -1045,7 +1045,7 @@ int ModApiMainMenu::l_do_async_callback(lua_State *L)
std::string serialized_func = std::string(serialized_func_raw, func_length); std::string serialized_func = std::string(serialized_func_raw, func_length);
std::string serialized_param = std::string(serialized_param_raw, param_length); std::string serialized_param = std::string(serialized_param_raw, param_length);
lua_pushinteger(L, engine->DoAsync(serialized_func, serialized_param)); lua_pushinteger(L, engine->queueAsync(serialized_func, serialized_param));
return 1; return 1;
} }

@ -30,7 +30,7 @@ extern "C" {
#include "lualib.h" #include "lualib.h"
} }
#define MAINMENU_ASYNC_THREADS 4 #define MAINMENU_NUM_ASYNC_THREADS 4
MainMenuScripting::MainMenuScripting(GUIEngine* guiengine) MainMenuScripting::MainMenuScripting(GUIEngine* guiengine)
@ -55,14 +55,14 @@ MainMenuScripting::MainMenuScripting(GUIEngine* guiengine)
lua_setglobal(L, "gamedata"); lua_setglobal(L, "gamedata");
// Initialize our lua_api modules // Initialize our lua_api modules
InitializeModApi(L, top); initializeModApi(L, top);
lua_pop(L, 1); lua_pop(L, 1);
infostream << "SCRIPTAPI: Initialized main menu modules" << std::endl; infostream << "SCRIPTAPI: Initialized main menu modules" << std::endl;
} }
/******************************************************************************/ /******************************************************************************/
void MainMenuScripting::InitializeModApi(lua_State *L, int top) void MainMenuScripting::initializeModApi(lua_State *L, int top)
{ {
// Initialize mod API modules // Initialize mod API modules
ModApiMainMenu::Initialize(L, top); ModApiMainMenu::Initialize(L, top);
@ -72,22 +72,22 @@ void MainMenuScripting::InitializeModApi(lua_State *L, int top)
LuaSettings::Register(L); LuaSettings::Register(L);
// Register functions to async environment // Register functions to async environment
ModApiMainMenu::InitializeAsync(m_AsyncEngine); ModApiMainMenu::InitializeAsync(asyncEngine);
ModApiUtil::InitializeAsync(m_AsyncEngine); ModApiUtil::InitializeAsync(asyncEngine);
// Initialize async environment // Initialize async environment
//TODO possibly make number of async threads configurable //TODO possibly make number of async threads configurable
m_AsyncEngine.Initialize(MAINMENU_ASYNC_THREADS); asyncEngine.initialize(MAINMENU_NUM_ASYNC_THREADS);
} }
/******************************************************************************/ /******************************************************************************/
void MainMenuScripting::Step() { void MainMenuScripting::step() {
m_AsyncEngine.Step(getStack(), m_errorhandler); asyncEngine.step(getStack(), m_errorhandler);
} }
/******************************************************************************/ /******************************************************************************/
unsigned int MainMenuScripting::DoAsync(std::string serialized_func, unsigned int MainMenuScripting::queueAsync(std::string serialized_func,
std::string serialized_param) { std::string serialized_param) {
return m_AsyncEngine.doAsyncJob(serialized_func, serialized_param); return asyncEngine.queueAsyncJob(serialized_func, serialized_param);
} }

@ -36,15 +36,15 @@ public:
MainMenuScripting(GUIEngine* guiengine); MainMenuScripting(GUIEngine* guiengine);
// Global step handler to pass back async events // Global step handler to pass back async events
void Step(); void step();
// Pass async events from engine to async threads // Pass async events from engine to async threads
unsigned int DoAsync(std::string serialized_func, unsigned int queueAsync(std::string serialized_func,
std::string serialized_params); std::string serialized_params);
private: private:
void InitializeModApi(lua_State *L, int top); void initializeModApi(lua_State *L, int top);
AsyncEngine m_AsyncEngine; AsyncEngine asyncEngine;
}; };