path: root/src/lua_plugin_manage.c

/*************************************************************************
    > File Name: lua_plugin_manage.c
    > Author:
    > Created Time: 2024-08
    > Encoding : UTF-8
 ************************************************************************/

/*************************************************************************
 * version
 * [ v0.1 ]
 * 08-05
 * 1. 实现函数
 *    void specific_instance_copy;
 *    void specific_instance_destory;
 *    int script_instance_init_byname;
 *    int script_instance_init_byrefid;
 *    void script_instance_clean;
 *    int script_execute; (暂未实现)
 *    int session_plugin_instance_init;
 *    void session_plugin_instance_destory;
 *    int plugin_env_instance_init;
 *    void plugin_env_instance_destory;
 *    int thread_state_instance_init;
 *    void thread_state_instance_destory;
 *    int thread_state_instance_load;
 *    int lua_plugin_manage_config_load;
 *    int lua_plugin_manage_thread_load;
 *
 * 08-06
 * 1. 实现函数
 *    int script_execute;
 *
 * 08-13
 * 1. 修改部分创建列表使用的结构
 * 2. 实现函数
 *    struct lua_plugin * search_plugin_by_id
 *    struct lua_message_mq * search_message_mq_by_id
 ************************************************************************/
#include "lua_plugin_manage_internal.h"

#include <time.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

struct lua_plugin_manage_schema *global_schema = NULL;

#if 0
void lua_message_mq_destory(void * elt)
{
    if (!elt) return;
    struct lua_message_mq * mq = (struct lua_message_mq *)elt;
    if ( mq->topic_name )
        free(mq->topic_name);
    return;
}
#endif
static UT_icd lua_message_mq_icd = {sizeof(struct lua_message_mq), NULL, NULL, NULL};

void lua_plugin_destory(void *elt)
{
    if (!elt)
        return;
    struct lua_plugin *plugin = (struct lua_plugin *)elt;
    if (plugin->sub_topic_array)
        utarray_free(plugin->sub_topic_array);
    return;
}
static UT_icd lua_plugin_icd = {sizeof(struct lua_plugin), NULL, NULL, lua_plugin_destory};

/*
 * Function: search_plugin_by_id
 * Input: | int | plugin_id | 需要查找plugin数据的id
 * Output:
 * Return: 查找得到的plugin结构
 * Description: 在global schema中根据plugin_id查找一个plugin
 */
struct lua_plugin *search_plugin_by_id(int plugin_id)
{
    for (int i = 0; i < global_schema->model_count; ++i)
    {
        struct lua_model *model = &global_schema->model[i];

        struct lua_plugin *plugin = NULL;
        while ((plugin = utarray_next(model->plugin_array, plugin)))
        {
            if (plugin->plugin_id == plugin_id)
                return plugin;
            else if (plugin->plugin_id > plugin_id)
                return NULL;
        }
    }
    return NULL;
}

/*
 * Function: search_message_mq_by_id
 * Input: | int | topic_id | 需要查找的topic_id
 * Output:
 * Return: 查找得到的message_mq结构
 * Description: 在global schema中根据topic_id查找一个message_mq
 */
struct lua_message_mq *search_message_mq_by_id(int topic_id)
{
    struct lua_message_mq *mq = NULL;
    while ((mq = utarray_next(global_schema->message_mq_array, mq)))
    {
        if (mq->topic_id == topic_id)
            return mq;
    }
    return NULL;
}

/*
 * Function: thread_state_init
 * Input: | int | thread_id | 创建状态机的线程号
 * Output:
 * Return: 创建完成的状态机实例指针
 * Description: 创建一个lua的状态机实例, 在每个线程中创建一个
 */
lua_State *thread_state_init(int thread_id)
{
    lua_State *new_state = luaL_newstate();
    if (__glibc_unlikely(!new_state))
        return NULL;

    luaL_openlibs(new_state);
    /* 绑定所有注册函数 */
    lua_cbinding_functions(new_state);
    /* 绑定所有全局变量 */
    lua_cbinding_datas(new_state);

    lua_pushinteger(new_state, thread_id);
    lua_setglobal(new_state, LUA_STATE_THREAD_ID_KEY);

    lua_settop(new_state, 0);

    return new_state;
}

/*
 * Function: thread_state_load_specific
 * Input: | lua_State *                  | state    | 进行初始化的状态机
 *        | struct lua_model *           | model    | 需要加载的模块, 模块数量与具体配置实例的数量一致
 *        | struct lua_config_specific * | specific | 加载的配置
 * Output:
 * Return: enum LUA_PLUGIN_RETURN
 * Description: 在状态机中加载一个具体配置, 并将加载完成后生成的引用值保存在model中, 加载过程中会使用model进行校验
 */
int thread_state_load_specific(
    lua_State *state,
    struct lua_model *model,
    struct lua_config_specific *specific)
{
    if (__glibc_unlikely(!state || !model || !specific))
        return PARAM_ERR;
    if (__glibc_unlikely(!specific->config_specific_file || !specific->config_specific_load_func || !specific->config_specific_unload_func))
        return PARAM_ERR;

    /* 加载lua文件 */
    if (access(specific->config_specific_file, F_OK))
        return STATE_LOAD_FILE_ERR;
    if (luaL_loadfile(state, specific->config_specific_file))
        return STATE_LOAD_FILE_ERR;
    if (lua_pcall(state, 0, 0, 0))
    {
        LOGERROR("load specific failed, file is %s, err is %s", specific->config_specific_file, lua_tostring(state, -1));
        lua_settop(state, 0);
        return STATE_LOAD_FILE_ERR;
    }
    lua_settop(state, 0);

    /* 创建plugin_env */
    lua_newtable(state);
    int env_ref_id = luaL_ref(state, LUA_REGISTRYINDEX);
    if (env_ref_id == LUA_REFNIL)
        return STATE_CREATE_ENV_FAIL;
    if ((model->model_mark & MODEL_MARK_INIT_DONE) && (model->private_env_ref != env_ref_id))
    {
        LOGERROR("load new model, create plugin env id is not same, %d, %d", model->private_env_ref, env_ref_id);
        return STATE_CREATE_ENV_FAIL;
    }

    /* 加载load函数 */
    lua_getglobal(state, specific->config_specific_load_func);
    if (lua_type(state, -1) != LUA_TFUNCTION)
    {
        luaL_unref(state, LUA_REGISTRYINDEX, env_ref_id);
        lua_settop(state, 0);
        return STATE_GET_LOAD_FAIL;
    }
    int load_ref_id = luaL_ref(state, LUA_REGISTRYINDEX);
    if (load_ref_id == LUA_REFNIL)
    {
        luaL_unref(state, LUA_REGISTRYINDEX, env_ref_id);
        lua_settop(state, 0);
        return STATE_GET_LOAD_FAIL;
    }
    if ((model->model_mark & MODEL_MARK_INIT_DONE) && (model->load_ref != load_ref_id))
    {
        LOGERROR("load new model, load ref id is not same, %d, %d", model->load_ref, load_ref_id);
        return STATE_GET_LOAD_FAIL;
    }

    /* 加载unload函数 */
    lua_getglobal(state, specific->config_specific_unload_func);
    if (lua_type(state, -1) != LUA_TFUNCTION)
    {
        luaL_unref(state, LUA_REGISTRYINDEX, env_ref_id);
        luaL_unref(state, LUA_REGISTRYINDEX, load_ref_id);
        lua_settop(state, 0);
        return STATE_GET_UNLOAD_FAIL;
    }
    int unload_ref_id = luaL_ref(state, LUA_REGISTRYINDEX);
    if (unload_ref_id == LUA_REFNIL)
    {
        luaL_unref(state, LUA_REGISTRYINDEX, env_ref_id);
        luaL_unref(state, LUA_REGISTRYINDEX, load_ref_id);
        lua_settop(state, 0);
        return STATE_GET_UNLOAD_FAIL;
    }
    if ((model->model_mark & MODEL_MARK_INIT_DONE) && (model->unload_ref != unload_ref_id))
    {
        LOGERROR("load new model, unload ref id is not same, %d, %d", model->unload_ref, unload_ref_id);
        return STATE_GET_UNLOAD_FAIL;
    }

    /* 首次加载时完成model初始化 */
    if (!(model->model_mark & MODEL_MARK_INIT_DONE))
    {
        model->private_env_ref = env_ref_id;
        model->load_ref = load_ref_id;
        model->unload_ref = unload_ref_id;
        model->model_mark |= MODEL_MARK_INIT_DONE;
    }

    lua_settop(state, 0);
    return SUCCESS;
}

/*
 * Function: thread_state_call_load
 * Input: | lua_State *                  | state    | 进行模块加载的状态机
 *        | struct lua_model *           | model    | 需要进行加载的lua模块
 *        | struct lua_config_specific * | specific | 加载模块的配置信息
 *        | struct stellar *             | st       | 与C插件管理保持一致, stellar作为参数传入
 * Output:
 * Return: enum LUA_PLUGIN_RETURN
 * Description: 在状态机中调用一个模块的load函数, 在状态机中完成该模块的加载
 */
int thread_state_call_load(
    lua_State *state,
    struct lua_model *model,
    struct lua_config_specific *specific,
    struct stellar *st)
{
    if (__glibc_unlikely(!state || !model || !specific || !st))
        return PARAM_ERR;

    if ((!(model->model_mark & MODEL_MARK_LOAD_DONE)) && (!model->plugin_array))
    {
        utarray_new(model->plugin_array, &lua_plugin_icd);
        model->model_mark |= MODEL_MARK_LOAD_DONE;
    }

    lua_rawgeti(state, LUA_REGISTRYINDEX, model->private_env_ref);
    lua_pushlightuserdata(state, model);
    /* 在该状态机对应模块的plugin_env中创建该结构对应的指针 */
    lua_setfield(state, -2, LUA_PLUGIN_ENV_DEFAULT_KEY);
    lua_settop(state, 0);
    luaL_loadfile(state, specific->config_specific_file);
    lua_pcall(state, 0, 0, 0);

    struct lua_cdata params[2];
    params[0].cdata_type = DATATYPE_POINTER;
    params[0].cdata_pointer = st;
    params[1].cdata_type = DATATYPE_TABLE;
    params[1].cdata_table = model->private_env_ref;

    return lua_chunk_execute(state, model->load_ref, 2, params, 0, NULL);
}

/*
 * Function: thread_state_call_unload
 * Input: | lua_State *        | 卸载模块的状态机
 *        | struct lua_model * | 需要卸载的模块
 * Output:
 * Return: enum LUA_PLUGIN_RETURN
 * Description: 在一个状态机中对某模块调用其卸载unload函数
 */
int thread_state_call_unload(
    lua_State *state,
    struct lua_model *model)
{
    if (__glibc_unlikely(!state || !model))
        return PARAM_ERR;

    struct lua_cdata param;
    param.cdata_type = DATATYPE_TABLE;
    param.cdata_table = model->private_env_ref;

    return lua_chunk_execute(state, model->unload_ref, 1, &param, 0, NULL);
}

/*
 * Function:
 * Input:
 * Output:
 * Return:
 * Description:
 */
struct lua_plugin_manage_schema *lua_plugin_manage_init(
    struct stellar *st,
    int specific_count,
    struct lua_config_specific *specific)
{
    if (__glibc_unlikely(!st))
        return NULL;
    if (__glibc_unlikely((specific_count && !specific) || (specific_count < 0)))
        return NULL;

    struct lua_plugin_manage_schema *new_schema = (struct lua_plugin_manage_schema *)calloc(1, sizeof(struct lua_plugin_manage_schema));
    if (__glibc_unlikely(!new_schema))
        return NULL;
    memset(new_schema, 0, sizeof(struct lua_plugin_manage_schema));
    global_schema = new_schema;

    new_schema->st = st;
    /* 可能运行过程中创建新的topic, 此处进行初始化 */
    new_schema->mq_count = 0;
    utarray_new(new_schema->message_mq_array, &lua_message_mq_icd);

    int thread_count = stellar_get_worker_thread_num(st);
    new_schema->state_count = thread_count;
    /* 为后续待创建的lua_State预分配内存 */
    new_schema->thread_state = (lua_State **)calloc(thread_count, sizeof(lua_State *));
    if (__glibc_unlikely(!new_schema->thread_state))
    {
        lua_plugin_manage_exit(new_schema);
        return NULL;
    }
    memset(new_schema->thread_state, 0, thread_count * sizeof(lua_State *));
    /* 为将要加载的模块预分配内存 */
    new_schema->model_count = specific_count;
    /* 如果没有配置, 也应该算是创建成功 */
    if (specific_count == 0)
        return new_schema;
    new_schema->model = (struct lua_model *)calloc(specific_count, sizeof(struct lua_model));
    if (__glibc_unlikely(!new_schema->model))
    {
        lua_plugin_manage_exit(new_schema);
        return NULL;
    }
    memset(new_schema->model, 0, specific_count * sizeof(struct lua_model));

    /* 依次初始化每个状态机 */
    for (int thread_index = 0; thread_index < thread_count; ++thread_index)
    {
        /* 创建状态机 */
        new_schema->thread_state[thread_index] = thread_state_init(thread_index);
        if (__glibc_unlikely(!new_schema->thread_state[thread_index]))
        {
            lua_plugin_manage_exit(new_schema);
            return NULL;
        }

        for (int spec_index = 0; spec_index < specific_count; ++spec_index)
        {
            /* 在状态机中加载specific中的配置 */
            int load_ret = thread_state_load_specific(new_schema->thread_state[thread_index],
                                                      &new_schema->model[spec_index],
                                                      &specific[spec_index]);
            if (load_ret)
            {
                LOGERROR("state load specific failed, ret is %d", load_ret);
                lua_plugin_manage_exit(new_schema);
                return NULL;
            }

            /* 调用该模块中的load函数 */
            int call_load_ret = thread_state_call_load(new_schema->thread_state[thread_index],
                                                       &new_schema->model[spec_index],
                                                       &specific[spec_index],
                                                       st);
            if (call_load_ret)
            {
                LOGERROR("call state load function failed, ret is %d", call_load_ret);
                lua_plugin_manage_exit(new_schema);
                return NULL;
            }
        }
    }
#ifdef LUAPLUGIN_BASIC_UNITTEST
    debug_lua_plugin_manage_schema(new_schema);
#endif

    /* 初始化全局状态统计 */
    global_plugin_statistics = (struct lua_plugin_statistics *)calloc((thread_count * global_max_plugin_id), sizeof(struct lua_plugin_statistics));
    if (!global_plugin_statistics)
    {
        LOGERROR("create plugin statistics failed");
        lua_plugin_manage_exit(new_schema);
        return NULL;
    }

    return new_schema;
}

/*
 * Function:
 * Input:
 * Output:
 * Return:
 * Description:
 */
int lua_plugin_manage_load_one_specific(
    struct lua_plugin_manage_schema *schema,
    struct lua_config_specific *specific)
{
    if (__glibc_unlikely(!schema || !specific))
        return -1;

    if (schema->model)
    {
        schema->model = (struct lua_model *)realloc(schema->model, (schema->model_count + 1) * sizeof(struct lua_model));
        schema->model_count += 1;
    }

    for (int thread_index = 0; thread_index < schema->state_count; ++thread_index)
    {
        /* 在状态机中加载specific中的配置 */
        int load_ret = thread_state_load_specific(schema->thread_state[thread_index],
                                                  &schema->model[schema->model_count - 1],
                                                  specific);
        if (load_ret)
        {
            LOGERROR("state load specific failed, ret is %d", load_ret);
            return -1;
        }

        /* 调用该模块中的load函数 */
        int call_load_ret = thread_state_call_load(schema->thread_state[thread_index],
                                                   &schema->model[schema->model_count - 1],
                                                   specific,
                                                   schema->st);
        if (call_load_ret)
        {
            LOGERROR("call state load function failed, ret is %d", call_load_ret);
            return -1;
        }
    }

    if (global_plugin_statistics)
        free(global_plugin_statistics);
    /* 初始化全局状态统计 */
    global_plugin_statistics = (struct lua_plugin_statistics *)calloc((schema->state_count * global_max_plugin_id), sizeof(struct lua_plugin_statistics));
    if (!global_plugin_statistics)
    {
        LOGERROR("create plugin statistics failed");
        return -1;
    }

    return 0;
}

/*
 * Function:
 * Input:
 * Output:
 * Return:
 * Description:
 */
void lua_plugin_manage_exit(struct lua_plugin_manage_schema *lua_plug_mgr)
{
    if (__glibc_unlikely(!lua_plug_mgr))
        return;

    /* 释放所有的状态机 */
    for (int state_index = 0; state_index < lua_plug_mgr->state_count; ++state_index)
    {
        for (int model_index = 0; model_index < lua_plug_mgr->model_count; ++model_index)
        {
            /* 在状态机中对每一个模块调用对应的卸载函数 */
            int call_unload_ret = thread_state_call_unload(lua_plug_mgr->thread_state[state_index],
                                                           &lua_plug_mgr->model[model_index]);
            if (call_unload_ret)
            {
                LOGERROR("call state unload function failed, ret is %d", call_unload_ret);
            }
        }
        /* 销毁状态机 */
        lua_close(lua_plug_mgr->thread_state[state_index]);
    }
    free(lua_plug_mgr->thread_state);

    /* 释放所有加载的模块 */
    for (int model_index = 0; model_index < lua_plug_mgr->model_count; ++model_index)
    {
        utarray_free(lua_plug_mgr->model[model_index].plugin_array);
    }
    free(lua_plug_mgr->model);

    if (lua_plug_mgr->message_mq_array)
        utarray_free(lua_plug_mgr->message_mq_array);

    free(lua_plug_mgr);
    return;
}

/*
 * Function:
 * Input:
 * Output:
 * Return:
 * Description: 获取某插件的运行成功及失败次数
 */
void lua_plugin_get_statistics(int plugin_id, int thread_id, int *new_success, int *new_fail, int *free_success, int *free_fail)
{
    int statistic_id = thread_id * global_max_plugin_id + plugin_id;
    if (new_success)
        *new_success = global_plugin_statistics[statistic_id].new_success_count;
    if (new_fail)
        *new_fail = global_plugin_statistics[statistic_id].new_failed_count;
    if (free_success)
        *free_success = global_plugin_statistics[statistic_id].free_success_count;
    if (free_fail)
        *free_fail = global_plugin_statistics[statistic_id].free_failed_count;
    return;
}

#ifdef LUAPLUGIN_BASIC_UNITTEST
void debug_lua_state_stack(lua_State *state, int mod, const char *message)
{
    int stackcount = lua_gettop(state);
    /* nothing in stack */
    if (!stackcount)
    {
        printf("debug stack, but stack is empty");
        return;
    }

    printf("\n***** begin to debug one lua stack *****\n");
    if (message)
        printf("debug here: %s\n", message);
    int i = stackcount;
    /* print variables one by one */
    for (; i > 0; --i)
    {
        /* adjust variables according to mod */
        if (mod)
            i = 0 - i;
        int type = lua_type(state, i);
        printf("stack[%d]: ", i);
        switch (type)
        {
        case LUA_TBOOLEAN:
            printf(lua_toboolean(state, i) ? "true\n" : "false\n");
            break;
        case LUA_TNUMBER:
            printf("%g\n", lua_tonumber(state, i));
            break;
        case LUA_TSTRING:
            printf("%s\n", lua_tostring(state, i));
            break;
        default:
            printf("this is not normal type, type is: %d[%s]\n", type, lua_typename(state, type));
            break;
        }
        /* adjust variables according to mod */
        if (mod)
            i = 0 - i;
    }
    printf("***** end of debug one lua stack *****\n\n");
}

void debug_lua_plugin_manage_schema(struct lua_plugin_manage_schema *schema)
{
    printf("\n***** begin to debug one lua schema *****\n");
    printf("schema.st is %p\n", schema->st);
    printf("schema state count is %d\n", schema->state_count);
    for (int i = 0; i < schema->state_count; ++i)
    {
        printf("schema state[%d]pointer is %p\n", i, schema->thread_state[i]);
    }
    printf("schema model count is %d\n", schema->model_count);
    for (int i = 0; i < schema->model_count; ++i)
    {
        printf("debug model[%d]\n", i);
        printf("array %p, load %d, unload %d, env %d, mark %04x, count %04x\n",
               schema->model[i].plugin_array, schema->model[i].load_ref, schema->model[i].unload_ref,
               schema->model[i].private_env_ref, schema->model[i].model_mark, schema->model[i].plugin_count);
        struct lua_plugin *plugin = NULL;
        while ((plugin = utarray_next(schema->model[i].plugin_array, plugin)))
        {
            printf("%d, %d, %d\n", plugin->plugin_id, plugin->ctx_new_ref, plugin->ctx_free_ref);
        }
    }

    printf("***** end of debug one lua schema *****\n\n");
}
#endif