GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-29 15:09:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #754 from gc87/mqtt-mem-cache

Browse Source 
Changed MQTT memory cache from per connection to per node
This commit is contained in:
fengzeroz
2022-09-06 13:42:37 +08:00
committed by GitHub
parent 7b96ec267b 18481a4faa
commit 3153b8a510
6 changed files with 109 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/utils/mem_cache.h
+2
View File
@@ -45,6 +45,8 @@ cache_item_t neu_mem_cache_earliest(neu_mem_cache_t *cache);
cache_item_t neu_mem_cache_latest(neu_mem_cache_t *cache);
void neu_mem_cache_used(neu_mem_cache_t *cache, size_t *used_bytes,
size_t *used_items);
void neu_mem_cache_resize(neu_mem_cache_t *cache, size_t new_bytes,
size_t new_items);
int neu_mem_cache_dump(neu_mem_cache_t *cache, cache_dump callback, void *ctx);
int neu_mem_cache_clear(neu_mem_cache_t *cache);
void neu_mem_cache_destroy(neu_mem_cache_t *cache);
plugins/mqtt/mqtt_plugin.c
+65 -75
View File
@@ -19,9 +19,6 @@
#include <assert.h>
#include <nng/nng.h>
#include <nng/supplemental/util/platform.h>
#include "command/command.h"
#include "mqtt_plugin.h"
#include "mqtt_util.h"
@@ -53,14 +50,10 @@ struct topic_pair {
};
struct mqtt_routine {
neu_plugin_t * plugin;
neu_mqtt_option_t option;
neu_mqtt_client_t client;
UT_array * topics;
neu_mem_cache_t * cache;
nng_mtx * mutex;
neu_events_t * events;
neu_event_timer_t *send;
neu_plugin_t * plugin;
neu_mqtt_option_t option;
neu_mqtt_client_t client;
UT_array * topics;
};
static char *topics_format(char *format, char *name)
@@ -171,46 +164,6 @@ static struct topic_pair *topics_find_type(UT_array *topics, int type)
return NULL;
}
static int mqtt_routine_cache_send(void *data)
{
mqtt_routine_t *routine = (mqtt_routine_t *) data;
// size_t used_bytes = 0;
// size_t used_items = 0;
// neu_mem_cache_used(routine->cache, &used_bytes, &used_items);
// plog_info(routine->plugin, "used bytes:%lu, used items:%lu", used_bytes,
// used_items);
neu_err_code_e rc = neu_mqtt_client_is_connected(routine->client);
if (0 != rc) {
return -1;
}
nng_mtx_lock(routine->mutex);
cache_item_t item = neu_mem_cache_earliest(routine->cache);
if (NULL == item.data) {
nng_mtx_unlock(routine->mutex);
return -1;
}
int type = TOPIC_TYPE_UPLOAD;
struct topic_pair *pair = topics_find_type(routine->topics, type);
char * json_str = item.data;
const char * topic = pair->topic_response;
const int qos = pair->qos_response;
neu_err_code_e error =
neu_mqtt_client_publish(routine->client, topic, qos,
(unsigned char *) json_str, strlen(json_str));
if (NEU_ERR_SUCCESS != error) {
plog_error(routine->plugin, "cache publish error code :%d, topoic:%s",
error, topic);
}
free(json_str);
nng_mtx_unlock(routine->mutex);
return 0;
}
static void mqtt_routine_response(const char *topic_name, size_t topic_len,
void *payload, const size_t len,
void *context)
@@ -299,21 +252,6 @@ static mqtt_routine_t *mqtt_routine_start(neu_plugin_t *plugin,
topics_generate(routine->topics, plugin, &routine->option);
topics_subscribe(routine->topics, routine->client);
// cache send
const size_t max_bytes = routine->option.cache * 1024 * 1024;
const size_t max_items = 0;
routine->cache = neu_mem_cache_create(max_bytes, max_items);
nng_mtx_alloc(&routine->mutex);
routine->events = neu_event_new();
neu_event_timer_param_t param = {
.second = 0,
.millisecond = 100,
.usr_data = routine,
.cb = mqtt_routine_cache_send,
};
routine->send = neu_event_add_timer(routine->events, param);
return routine;
}
@@ -321,12 +259,6 @@ static void mqtt_routine_stop(mqtt_routine_t *routine)
{
assert(NULL != routine);
// stop cache send timer
neu_event_del_timer(routine->events, routine->send);
neu_event_close(routine->events);
nng_mtx_free(routine->mutex);
neu_mem_cache_destroy(routine->cache);
// stop recevied
neu_mqtt_client_suspend(routine->client);
@@ -358,6 +290,8 @@ static int mqtt_plugin_close(neu_plugin_t *plugin)
return NEU_ERR_SUCCESS;
}
static void plugin_cache_init(neu_plugin_t *plugin);
static int mqtt_plugin_init(neu_plugin_t *plugin)
{
assert(NULL != plugin);
@@ -365,6 +299,7 @@ static int mqtt_plugin_init(neu_plugin_t *plugin)
plugin->routine = NULL;
plugin->running = false;
plugin->config = NULL;
plugin_cache_init(plugin);
const char *name = neu_plugin_module.module_name;
plog_info(plugin, "initialize plugin: %s", name);
@@ -418,11 +353,24 @@ static void plugin_config_save(neu_plugin_t *plugin, const char *config)
plugin->config = strdup(config);
}
static void plugin_cache_init(neu_plugin_t *plugin)
{
plugin->cache = neu_mem_cache_create(0, 0);
nng_mtx_alloc(&plugin->mutex);
}
static void plugin_cache_uninit(neu_plugin_t *plugin)
{
nng_mtx_free(plugin->mutex);
neu_mem_cache_destroy(plugin->cache);
}
static int mqtt_plugin_uninit(neu_plugin_t *plugin)
{
assert(NULL != plugin);
plugin_stop_running(plugin);
plugin_cache_uninit(plugin);
plugin_config_free(plugin);
const char *name = neu_plugin_module.module_name;
@@ -435,11 +383,17 @@ static int mqtt_plugin_config(neu_plugin_t *plugin, const char *config)
assert(NULL != plugin);
assert(NULL != config);
// Option setting
if (NEU_ERR_SUCCESS != mqtt_option_validate(plugin, config)) {
return NEU_ERR_NODE_SETTING_INVALID;
}
plugin_config_save(plugin, config);
size_t cache_size = mqtt_option_item_cache(plugin, config);
size_t max_bytes = cache_size * 1024 * 1024;
nng_mtx_lock(plugin->mutex);
neu_mem_cache_resize(plugin->cache, max_bytes, 0);
nng_mtx_unlock(plugin->mutex);
// Routine not running
if (!plugin->running) {
@@ -544,6 +498,35 @@ static void cache_string_release(void *data)
free(data);
}
static int cache_send(neu_plugin_t *plugin)
{
mqtt_routine_t *routine = plugin->routine;
nng_mtx_lock(plugin->mutex);
cache_item_t item = neu_mem_cache_earliest(plugin->cache);
if (NULL == item.data) {
nng_mtx_unlock(plugin->mutex);
return -1;
}
int type = TOPIC_TYPE_UPLOAD;
struct topic_pair *pair = topics_find_type(routine->topics, type);
char * json_str = item.data;
const char * topic = pair->topic_response;
const int qos = pair->qos_response;
neu_err_code_e error =
neu_mqtt_client_publish(routine->client, topic, qos,
(unsigned char *) json_str, strlen(json_str));
if (NEU_ERR_SUCCESS != error) {
plog_error(plugin, "cache publish error code :%d, topoic:%s", error,
topic);
}
free(json_str);
nng_mtx_unlock(plugin->mutex);
return 0;
}
static neu_err_code_e trans_data(neu_plugin_t *plugin, void *data)
{
neu_reqresp_trans_data_t *trans_data = data;
@@ -569,15 +552,22 @@ static neu_err_code_e trans_data(neu_plugin_t *plugin, void *data)
.release = cache_string_release,
};
nng_mtx_lock(routine->mutex);
rc = neu_mem_cache_add(routine->cache, &item);
nng_mtx_unlock(routine->mutex);
nng_mtx_lock(plugin->mutex);
rc = neu_mem_cache_add(plugin->cache, &item);
nng_mtx_unlock(plugin->mutex);
if (0 != rc) {
item.release(item.data);
}
return NEU_ERR_MQTT_CONNECT_FAILURE;
} else {
for (int i = 0; i < 100; i++) {
int r = cache_send(plugin);
if (-1 == r) {
break;
}
}
}
const char * topic = pair->topic_response;
plugins/mqtt/mqtt_plugin.h
+5
View File
@@ -26,6 +26,9 @@ extern "C" {
#include <stdlib.h>
#include <nng/nng.h>
#include <nng/supplemental/util/platform.h>
#include <neuron.h>
typedef struct mqtt_routine mqtt_routine_t;
@@ -35,6 +38,8 @@ struct neu_plugin {
bool running;
mqtt_routine_t * routine;
char * config;
neu_mem_cache_t * cache;
nng_mtx * mutex;
};
#ifdef __cplusplus
plugins/mqtt/mqtt_util.c
+15
View File
@@ -341,4 +341,19 @@ int mqtt_option_validate(neu_plugin_t *plugin, const char *config)
mqtt_option_uninit(plugin, &option);
return NEU_ERR_SUCCESS;
}
size_t mqtt_option_item_cache(neu_plugin_t *plugin, const char *config)
{
neu_mqtt_option_t option = { 0 };
int rc = mqtt_option_init(plugin, (char *) config, &option);
if (0 != rc) {
plog_error(plugin, "%s", mqtt_option_error(rc));
mqtt_option_uninit(plugin, &option);
return 64;
}
size_t cache_size = option.cache;
mqtt_option_uninit(plugin, &option);
return cache_size;
}
plugins/mqtt/mqtt_util.h
+2 -1
View File
@@ -32,7 +32,8 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config,
const char *mqtt_option_error(int error);
int mqtt_option_validate(neu_plugin_t *plugin, const char *config);
int mqtt_option_validate(neu_plugin_t *plugin, const char *config);
size_t mqtt_option_item_cache(neu_plugin_t *plugin, const char *config);
void mqtt_option_uninit(neu_plugin_t *plugin, neu_mqtt_option_t *option);
src/utils/mem_cache.c
+20 -6
View File
@@ -50,23 +50,26 @@ static bool queue_is_full(neu_mem_cache_t *cache, cache_item_t *item)
return false;
}
int neu_mem_cache_add(neu_mem_cache_t *cache, cache_item_t *item)
static void queue_reduce(neu_mem_cache_t *cache, cache_item_t *item)
{
assert(NULL != cache);
assert(NULL != item);
while (queue_is_full(cache, item)) {
cache_item_t remove_item = neu_mem_cache_earliest(cache);
if (NULL == remove_item.data) {
// size(item.data) > cache->max_bytes
return -1;
break;
}
if (NULL != remove_item.release && NULL != remove_item.data) {
remove_item.release(remove_item.data);
}
}
}
int neu_mem_cache_add(neu_mem_cache_t *cache, cache_item_t *item)
{
assert(NULL != cache);
assert(NULL != item);
queue_reduce(cache, item);
element *el = calloc(1, sizeof(element));
if (NULL == el) {
return -1;
@@ -126,6 +129,17 @@ void neu_mem_cache_used(neu_mem_cache_t *cache, size_t *bytes, size_t *msgs)
*msgs = cache->used_items;
}
void neu_mem_cache_resize(neu_mem_cache_t *cache, size_t new_bytes,
size_t new_items)
{
assert(NULL != cache);
cache->max_bytes = new_bytes;
cache->max_items = new_items;
cache_item_t item = { 0 };
queue_reduce(cache, &item);
}
int neu_mem_cache_dump(neu_mem_cache_t *cache, cache_dump callback, void *ctx)
{
assert(NULL != cache);