Initial release commit

src/lua_platform.cpp

@@ -0,0 +1,191 @@
+#include "lua_platform.h"
+#include "platform_agnostic.h"
+#include "dynarray.hpp"
+#include "mutex"
+#include <chrono>
+#include "simple_string.h"
+
+int luaopen_platform(lua_State* L)
+{
+    lua_newtable(L);
+
+    lua_pushcfunction(L, lua_platform_exec);
+    lua_setfield(L, -2, "exec");
+    lua_pushcfunction(L, lua_platform_exec_parallel);
+    lua_setfield(L, -2, "exec_parallel");
+
+    lua_pushvalue(L, -1);
+    lua_setfield(L, LUA_GLOBALSINDEX, "platform");
+    return 1;
+}
+
+int lua_platform_exec(lua_State* L)
+{
+    luaL_argcheck(L, lua_isstring(L, 1), 1, "Expected first argument to be a string");
+    luaL_argcheck(L, lua_istable(L, 2), 2, "Expected second argument to be a table of strings");
+    std::string cmd(lua_tostring(L, 1));
+    size_t nargs = lua_objlen(L, 2);
+    for (size_t i = 0; i < nargs; ++i)
+    {
+        lua_rawgeti(L, -1, i + 1);
+        if (!lua_isstring(L, -1))
+            return luaL_error(L, "Expected all values in argument table to be a string. Index %d was a %s", i, lua_typename(L, lua_type(L, -1)));
+        cmd.append(" ").append(lua_tostring(L, -1));
+    }
+
+    FILE* f = popen(cmd.c_str(), "r");
+    char buf[64];
+    std::string result;
+    while (fgets(buf, sizeof(buf), f) != NULL)
+        result.append(buf);
+
+    lua_pushinteger(L, pclose(f));
+    lua_pushlstring(L, result.c_str(), result.size());
+    return 2;
+}
+
+
+struct command_stack
+{
+    std::string base_cmd;
+    dynarray<dynarray<std::string>> commands;
+    std::recursive_mutex mux;
+
+    size_t num_cmds(void) const noexcept
+    { return commands.size(); }
+    bool is_empty(void) const noexcept
+    { return commands.size() == 0; }
+    std::pair<dynarray<std::string>, size_t> get_command(void) noexcept
+    {
+        std::scoped_lock lock(mux);
+        const size_t sz = commands.size();
+        return std::make_pair(std::move(commands.pop_back()),sz-1);
+    }
+};
+struct result_stack
+{
+    struct result
+    {
+        std::string program_output;
+        int program_ret;
+    };
+    
+    std::recursive_mutex mux;
+    dynarray<result> results;
+
+    void add_result(const size_t idx, const std::string& out, const int ret)
+    {
+        std::scoped_lock lock(mux);
+        new (results.data() + idx) result(out, ret);
+    }
+};
+
+
+void worker_thread(std::stop_token token, const std::string& base_cmd, command_stack& in, result_stack& out)
+{
+    char buf[64];
+    while (!token.stop_requested())
+    {
+        std::string cmd = base_cmd;
+        if (in.is_empty()) return;
+        auto [arg, idx] = in.get_command();
+        // Get a command
+        
+        for (const std::string& a : arg)
+            cmd.append(" ").append(a);
+
+        // Run the command
+        FILE* f = popen(cmd.c_str(), "r");
+        
+        // Capture the commands stdout
+        std::string pout = "";
+        memset(buf, '\0', sizeof(buf));
+        while (fgets(buf, sizeof(buf), f) != NULL)
+        {
+            using namespace std::chrono_literals;
+            std::this_thread::sleep_for(10us);
+            pout.append(buf);
+        }
+
+        // Await the program to exit
+        int rc = pclose(f);
+        out.add_result(idx, pout, rc);
+    }
+}
+
+int lua_platform_exec_parallel(lua_State* L)
+{
+    luaL_argcheck(L, lua_type(L, 1) == LUA_TSTRING, 1, "Expected first argument to be a string representing a command");
+    luaL_argcheck(L, lua_type(L, 2) == LUA_TNUMBER, 2, "Expected second argument to be an integer representing the maximum number of parallel jobs to start");
+    luaL_argcheck(L, lua_type(L, 3) == LUA_TTABLE, 3, "Expected third argument to be a table of table of strings representing the arguments");
+    std::string base_cmd(lua_tostring(L, 1));
+    const size_t max_parallel = lua_tointeger(L, 2);
+    const size_t num_cmds = lua_objlen(L, 3);
+
+    command_stack in;
+    in.base_cmd = base_cmd;
+    result_stack out;
+    for (size_t i = 0; i < num_cmds; ++i)
+        out.results.push_back(result_stack::result{"Not Executed",-1});
+
+    // Generate the command stack from lua arguments
+    for (size_t i = 0; i < num_cmds; ++i)
+    {
+        lua_rawgeti(L, 3, i + 1);
+        if (lua_istable(L, -1))
+        {
+            const size_t num_args = lua_objlen(L, -1);
+            in.commands.push_back(dynarray<std::string>());
+            dynarray<std::string>& back = in.commands.back();
+            for (size_t j = 0; j < num_args; ++j)
+            {
+                lua_rawgeti(L, -1, j + 1);
+                if (!lua_isstring(L, -1))
+                    return luaL_error(L, "Expected program argument to be a string. args[%d][%d] was a %s, not a string!\n", i, j, lua_typename(L, lua_type(L, -1)));
+                std::string str(lua_tostring(L, -1), lua_objlen(L, -1));
+                back.push_back(std::move(str));
+                lua_pop(L, 1);
+            }
+        }
+        else if (lua_type(L, -1) == LUA_TSTRING)
+        {
+            in.commands.push_back(dynarray<std::string>());
+            in.commands.back().push_back(lua_tostring(L, -1));
+        }
+        else
+            return luaL_error(L, "Expected a table of program arguments. args[%d] was a %s, not a table!\n", i, lua_typename(L, lua_type(L, -1)));
+        lua_pop(L, 1);
+    }
+
+    // Dispatch worker threads
+    std::vector<std::jthread> workers;
+    for (size_t i = 0; i < max_parallel; ++i)
+        workers.push_back(std::jthread(worker_thread, 
+            std::ref(base_cmd),
+            std::ref(in),
+            std::ref(out)
+        ));
+
+    using namespace std::chrono_literals;
+    while (!in.is_empty())
+        std::this_thread::sleep_for(1ms);
+
+    for (size_t i = 0; i < workers.size(); ++i)
+        workers[i].request_stop();
+    for (size_t i = 0; i < workers.size(); ++i)
+        workers[i].join();
+
+    // Join thread results together and push them to lua stack
+    lua_newtable(L);    // rc
+    lua_newtable(L);    // stdout
+    for (size_t i = 0; i < out.results.size(); ++i)
+    {
+        const result_stack::result& x = out.results[i];
+        lua_pushinteger(L, x.program_ret);
+        lua_rawseti(L, -3, i+1);
+        lua_pushlstring(L, x.program_output.c_str(), x.program_output.size());
+        lua_rawseti(L, -2, i + 1);
+    }
+
+    return 2;
+}