s_async.h

Go to the documentation of this file.

// Luanti
// SPDX-License-Identifier: LGPL-2.1-or-later
// Copyright (C) 2013 sapier, <sapier AT gmx DOT net>
 
#pragma once
 
#include <vector>
#include <deque>
#include <unordered_set>
#include <memory>
 
#include <lua.h>
#include "threading/semaphore.h"
#include "threading/thread.h"
#include "common/c_packer.h"
#include "cpp_api/s_base.h"
#include "cpp_api/s_security.h"
 
// Forward declarations
class AsyncEngine;
 
 
// Declarations
 
// Data required to queue a job
struct LuaJobInfo
{
        LuaJobInfo() = default;
        LuaJobInfo(std::string &&func, std::string &&params, const std::string &mod_origin = "") :
                function(func), params(params), mod_origin(mod_origin) {}
        LuaJobInfo(std::string &&func, PackedValue *params, const std::string &mod_origin = "") :
                function(func), mod_origin(mod_origin) {
                params_ext.reset(params);
        }
 
        // Function to be called in async environment (from string.dump)
        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
        u32 id;
};
 
// Asynchronous working environment
class AsyncWorkerThread : public Thread,
        virtual public ScriptApiBase, public ScriptApiSecurity {
        friend class AsyncEngine;
public:
        virtual ~AsyncWorkerThread();
 
        void *run() override;
 
protected:
        AsyncWorkerThread(AsyncEngine* jobDispatcher, const std::string &name);
 
        bool checkPathInternal(const std::string &abs_path, bool write_required,
                bool *write_allowed) override;
 
private:
        AsyncEngine *jobDispatcher = nullptr;
        bool isErrored = false;
};
 
// Asynchornous thread and job management
class AsyncEngine {
        friend class AsyncWorkerThread;
        typedef void (*StateInitializer)(lua_State *L, int top);
public:
        AsyncEngine() = default;
        AsyncEngine(Server *server) : server(server) {};
        ~AsyncEngine();
 
        void registerStateInitializer(StateInitializer func);
 
        void initialize(unsigned int numEngines);
 
        u32 queueAsyncJob(std::string &&func, std::string &&params,
                        const std::string &mod_origin = "");
 
        u32 queueAsyncJob(std::string &&func, PackedValue *params,
                        const std::string &mod_origin = "");
 
        bool cancelAsyncJob(u32 id);
 
        void step(lua_State *L);
 
        unsigned int getThreadingCapacity() const {
                return MYMAX(workerThreads.size(), autoscaleMaxWorkers);
        }
 
protected:
        bool getJob(LuaJobInfo *job);
 
        u32 queueAsyncJob(LuaJobInfo &&job);
 
        void putJobResult(LuaJobInfo &&result);
 
        void addWorkerThread();
 
        void stepJobResults(lua_State *L);
 
        void stepAutoscale();
 
        void stepStuckWarning();
 
        bool prepareEnvironment(lua_State* L, int top);
 
private:
        template <typename T>
        inline void snapshotJobs(T &to)
        {
                for (const auto &it : jobQueue)
                        to.emplace(it.id);
        }
        template <typename T>
        inline size_t compareJobs(const T &from)
        {
                size_t overlap = 0;
                for (const auto &it : jobQueue)
                        overlap += from.count(it.id);
                return overlap;
        }
 
        // 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;
 
        u64 stuckTimer = 0;
        std::unordered_set<u32> stuckSeenJobs;
 
        // Only set for the server async environment (duh)
        Server *server = nullptr;
 
        // Internal store for registred state initializers
        std::vector<StateInitializer> stateInitializers;
 
        // Internal counter to create job IDs
        u32 jobIdCounter = 0;
 
        // Mutex to protect job queue
        std::mutex jobQueueMutex;
        // Job queue
        std::deque<LuaJobInfo> jobQueue;
 
        // Mutex to protect result queue
        std::mutex resultQueueMutex;
        // Result queue
        std::deque<LuaJobInfo> resultQueue;
 
        // List of current worker threads
        std::vector<AsyncWorkerThread*> workerThreads;
 
        // Counter semaphore for job dispatching
        Semaphore jobQueueCounter;
};
 
class ScriptApiAsync:
        virtual public ScriptApiBase
{
public:
        ScriptApiAsync(Server *server): asyncEngine(server) {}
 
        virtual void initAsync() = 0;
        void stepAsync();
 
        u32 queueAsync(std::string &&serialized_func,
                        PackedValue *param, const std::string &mod_origin);
        bool cancelAsync(u32 id);
        unsigned int getThreadingCapacity() const {
                return asyncEngine.getThreadingCapacity();
        }
 
protected:
        AsyncEngine asyncEngine;
};