src/script/cpp_api/s_async.h

   1 /*
   2 Minetest
   3 Copyright (C) 2013 sapier, <sapier AT gmx DOT net>
   4
   5 This program is free software; you can redistribute it and/or modify
   6 it under the terms of the GNU Lesser General Public License as published by
   7 the Free Software Foundation; either version 2.1 of the License, or
   8 (at your option) any later version.
   9
  10 This program is distributed in the hope that it will be useful,
  11 but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 GNU Lesser General Public License for more details.
  14
  15 You should have received a copy of the GNU Lesser General Public License along
  16 with this program; if not, write to the Free Software Foundation, Inc.,
  17 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  18 */
  19
  20 #pragma once
  21
  22 #include <vector>
  23 #include <deque>
  24 #include <map>
  25
  26 #include "threading/semaphore.h"
  27 #include "threading/thread.h"
  28 #include "lua.h"
  29 #include "cpp_api/s_base.h"
  30
  31 // Forward declarations
  32 class AsyncEngine;
  33
  34
  35 // Declarations
  36
  37 // Data required to queue a job
  38 struct LuaJobInfo
  39 {
  40         LuaJobInfo() = default;
  41
  42         // Function to be called in async environment
  43         std::string serializedFunction = "";
  44         // Parameter to be passed to function
  45         std::string serializedParams = "";
  46         // Result of function call
  47         std::string serializedResult = "";
  48         // JobID used to identify a job and match it to callback
  49         unsigned int id = 0;
  50
  51         bool valid = false;
  52 };
  53
  54 // Asynchronous working environment
  55 class AsyncWorkerThread : public Thread, public ScriptApiBase {
  56 public:
  57         AsyncWorkerThread(AsyncEngine* jobDispatcher, const std::string &name);
  58         virtual ~AsyncWorkerThread();
  59
  60         void *run();
  61
  62 private:
  63         AsyncEngine *jobDispatcher = nullptr;
  64 };
  65
  66 // Asynchornous thread and job management
  67 class AsyncEngine {
  68         friend class AsyncWorkerThread;
  69         typedef void (*StateInitializer)(lua_State *L, int top);
  70 public:
  71         AsyncEngine() = default;
  72         ~AsyncEngine();
  73
  74         /**
  75          * Register function to be called on new states
  76          * @param func C function to be called
  77          */
  78         void registerStateInitializer(StateInitializer func);
  79
  80         /**
  81          * Create async engine tasks and lock function registration
  82          * @param numEngines Number of async threads to be started
  83          */
  84         void initialize(unsigned int numEngines);
  85
  86         /**
  87          * Queue an async job
  88          * @param func Serialized lua function
  89          * @param params Serialized parameters
  90          * @return jobid The job is queued
  91          */
  92         unsigned int queueAsyncJob(const std::string &func, const std::string &params);
  93
  94         /**
  95          * Engine step to process finished jobs
  96          *   the engine step is one way to pass events back, PushFinishedJobs another
  97          * @param L The Lua stack
  98          */
  99         void step(lua_State *L);
 100
 101         /**
 102          * Push a list of finished jobs onto the stack
 103          * @param L The Lua stack
 104          */
 105         void pushFinishedJobs(lua_State *L);
 106
 107 protected:
 108         /**
 109          * Get a Job from queue to be processed
 110          *  this function blocks until a job is ready
 111          * @return a job to be processed
 112          */
 113         LuaJobInfo getJob();
 114
 115         /**
 116          * Put a Job result back to result queue
 117          * @param result result of completed job
 118          */
 119         void putJobResult(const LuaJobInfo &result);
 120
 121         /**
 122          * Initialize environment with current registred functions
 123          *  this function adds all functions registred by registerFunction to the
 124          *  passed lua stack
 125          * @param L Lua stack to initialize
 126          * @param top Stack position
 127          */
 128         void prepareEnvironment(lua_State* L, int top);
 129
 130 private:
 131         // Variable locking the engine against further modification
 132         bool initDone = false;
 133
 134         // Internal store for registred state initializers
 135         std::vector<StateInitializer> stateInitializers;
 136
 137         // Internal counter to create job IDs
 138         unsigned int jobIdCounter = 0;
 139
 140         // Mutex to protect job queue
 141         std::mutex jobQueueMutex;
 142
 143         // Job queue
 144         std::deque<LuaJobInfo> jobQueue;
 145
 146         // Mutex to protect result queue
 147         std::mutex resultQueueMutex;
 148         // Result queue
 149         std::deque<LuaJobInfo> resultQueue;
 150
 151         // List of current worker threads
 152         std::vector<AsyncWorkerThread*> workerThreads;
 153
 154         // Counter semaphore for job dispatching
 155         Semaphore jobQueueCounter;
 156 };