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

Merge pull request #582 from fengzeroz/main

Browse Source 
feat(driver): the value of tag will expire after 5X group interval
This commit is contained in:
fengzeroz
2022-06-23 13:03:49 +08:00
committed by GitHub
parent ea23c41e42 dcd34e0b87
commit a8dc4bd54c
8 changed files with 88 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ft/resource/error.resource
+2 -1
View File
@@ -54,4 +54,5 @@ ${NEU_ERR_PLUGIN_WRITE_FAILURE} 3001
${NEU_ERR_PLUGIN_DISCONNECTED} 3002
${NEU_ERR_PLUGIN_TAG_NOT_ALLOW_READ} 3003
${NEU_ERR_PLUGIN_TAG_NOT_ALLOW_WRITE} 3004
${NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH} 3007
${NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH} 3007
${NEU_ERR_PLUGIN_TAG_EXPIRED} 3008
include/neuron/define.h
+1
View File
@@ -29,6 +29,7 @@
#define NEU_PLUGIN_NAME_LEN 32
#define NEU_PLUGIN_LIBRARY_LEN 32
#define NEU_PLUGIN_DESCRIPTION_LEN 256
#define NEU_DRIVER_TAG_CACHE_EXPIRE_TIME 5
typedef enum neu_plugin_kind {
NEU_PLUGIN_KIND_STATIC = 0,
include/neuron/errcodes.h
+1
View File
@@ -85,6 +85,7 @@ typedef enum {
NEU_ERR_PLUGIN_TAG_NOT_ALLOW_READ = 3003,
NEU_ERR_PLUGIN_TAG_NOT_ALLOW_WRITE = 3004,
NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH = 3007,
NEU_ERR_PLUGIN_TAG_EXPIRED = 3008,
NEU_ERR_MQTT_FAILURE = 4000,
NEU_ERR_MQTT_NO_CERTFILESET = 4001,
include/neuron/plugin.h
+1
View File
@@ -42,6 +42,7 @@ typedef struct neu_plugin_common {
neu_node_link_state_e link_state;
uint16_t tag_size;
int64_t timestamp;
zlog_category_t *log;
} neu_plugin_common_t;
src/adapter/adapter.c
+28 -3
View File
@@ -23,6 +23,7 @@
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/time.h>
#include <unistd.h>
#include <nng/nng.h>
@@ -46,6 +47,7 @@ static int adapter_response(neu_adapter_t *adapter, neu_reqresp_head_t *header,
void *data);
inline static void reply(neu_adapter_t *adapter, neu_reqresp_head_t *header,
void *data);
static int update_timestamp(void *usr_data);
static const adapter_callbacks_t callback_funs = {
.command = adapter_command,
@@ -54,9 +56,14 @@ static const adapter_callbacks_t callback_funs = {
neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool start)
{
int rv = 0;
neu_adapter_t * adapter = NULL;
neu_event_io_param_t param = { 0 };
int rv = 0;
neu_adapter_t * adapter = NULL;
neu_event_io_param_t param = { 0 };
neu_event_timer_param_t timer_param = {
.second = 0,
.millisecond = 100,
.cb = update_timestamp,
};
switch (info->module->type) {
case NEU_NA_TYPE_DRIVER:
@@ -126,6 +133,9 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool start)
rv = nng_dial(adapter->sock, neu_manager_get_url(), &adapter->dialer, 0);
assert(rv == 0);
timer_param.usr_data = (void *) adapter;
adapter->timer = neu_event_add_timer(adapter->events, timer_param);
nlog_info("Success to create adapter: %s", adapter->name);
return adapter;
@@ -604,6 +614,7 @@ int neu_adapter_uninit(neu_adapter_t *adapter)
neu_adapter_driver_destroy((neu_adapter_driver_t *) adapter);
}
neu_event_del_timer(adapter->events, adapter->timer);
adapter->module->intf_funs->close(adapter->plugin);
nlog_info("Stop the adapter(%s)", adapter->name);
@@ -744,4 +755,18 @@ inline static void reply(neu_adapter_t *adapter, neu_reqresp_head_t *header,
nng_msg *msg = neu_msg_gen(header, data);
nng_sendmsg(adapter->sock, msg, 0);
}
static int update_timestamp(void *usr_data)
{
neu_adapter_t *adapter = (neu_adapter_t *) usr_data;
struct timeval tv = { 0 };
uint64_t timestamp = 0;
gettimeofday(&tv, NULL);
timestamp = tv.tv_sec * 1000 + tv.tv_usec / 1000;
adapter->timestamp = timestamp;
neu_plugin_to_plugin_common(adapter->plugin)->timestamp = timestamp;
return 0;
}
src/adapter/adapter_internal.h
+6 -3
View File
@@ -34,6 +34,8 @@ struct neu_adapter {
char *name;
char *setting;
int64_t timestamp;
neu_node_running_state_e state;
neu_persister_t * persister;
@@ -46,9 +48,10 @@ struct neu_adapter {
nng_socket sock;
nng_dialer dialer;
neu_events_t * events;
neu_event_io_t *nng_io;
int recv_fd;
neu_events_t * events;
neu_event_io_t * nng_io;
neu_event_timer_t *timer;
int recv_fd;
};
typedef void (*adapter_handler)(neu_adapter_t * adapter,
src/adapter/driver/cache.c
+10 -9
View File
@@ -53,9 +53,9 @@ struct neu_driver_cache {
};
static void update_tag_error(neu_driver_cache_t *cache, const char *group,
const char *tag, int error);
const char *tag, int64_t timestamp, int error);
static void update_group_error(neu_driver_cache_t *cache, const char *group,
int error);
int64_t timestamp, int error);
inline static tkey_t to_key(const char *group, const char *tag)
{
@@ -100,11 +100,10 @@ void neu_driver_cache_destroy(neu_driver_cache_t *cache)
void neu_driver_cache_error(neu_driver_cache_t *cache, const char *group,
const char *tag, int64_t timestamp, int32_t error)
{
(void) timestamp;
if (tag == NULL) {
update_group_error(cache, group, error);
update_group_error(cache, group, timestamp, error);
} else {
update_tag_error(cache, group, tag, error);
update_tag_error(cache, group, tag, timestamp, error);
}
}
@@ -198,7 +197,7 @@ void neu_driver_cache_del(neu_driver_cache_t *cache, const char *group,
}
static void update_tag_error(neu_driver_cache_t *cache, const char *group,
const char *tag, int error)
const char *tag, int64_t timestamp, int error)
{
struct elem *elem = NULL;
tkey_t key = to_key(group, tag);
@@ -213,13 +212,14 @@ static void update_tag_error(neu_driver_cache_t *cache, const char *group,
HASH_ADD(hh, cache->table, key, sizeof(tkey_t), elem);
}
elem->error = error;
elem->error = error;
elem->timestamp = timestamp;
nng_mtx_unlock(cache->mtx);
}
static void update_group_error(neu_driver_cache_t *cache, const char *group,
int error)
int64_t timestamp, int error)
{
struct elem *elem = NULL, *tmp = NULL;
@@ -227,7 +227,8 @@ static void update_group_error(neu_driver_cache_t *cache, const char *group,
HASH_ITER(hh, cache->table, elem, tmp)
{
if (strcmp(elem->key.group, group) == 0) {
elem->error = error;
elem->timestamp = timestamp;
elem->error = error;
}
}
src/adapter/driver/driver.c
+39 -18
View File
@@ -61,7 +61,8 @@ struct neu_adapter_driver {
static int report_callback(void *usr_data);
static int read_callback(void *usr_data);
static void read_group(neu_driver_cache_t *cache, const char *group,
static void read_group(int64_t timestamp, int64_t timeout,
neu_driver_cache_t *cache, const char *group,
UT_array *tags, neu_resp_tag_value_t *datas);
static void update(neu_adapter_t *adapter, const char *group, const char *tag,
neu_dvalue_t value);
@@ -119,9 +120,11 @@ static void update(neu_adapter_t *adapter, const char *group, const char *tag,
}
if (value.type == NEU_TYPE_ERROR) {
neu_driver_cache_error(driver->cache, group, tag, 0, value.value.i32);
neu_driver_cache_error(driver->cache, group, tag,
driver->adapter.timestamp, value.value.i32);
} else {
neu_driver_cache_update(driver->cache, group, tag, 0, n_byte,
neu_driver_cache_update(driver->cache, group, tag,
driver->adapter.timestamp, n_byte,
value.value.bytes);
}
}
@@ -190,21 +193,30 @@ int neu_adapter_driver_uninit(neu_adapter_driver_t *driver)
void neu_adapter_driver_read_group(neu_adapter_driver_t *driver,
neu_reqresp_head_t * req)
{
if (driver->adapter.state != NEU_NODE_RUNNING_STATE_RUNNING) {
driver->adapter.cb_funs.driver.write_response(&driver->adapter, req,
NEU_ERR_NODE_NOT_RUNNING);
return;
}
neu_req_read_group_t *cmd = (neu_req_read_group_t *) &req[1];
neu_resp_read_group_t resp = { 0 };
UT_array * tags =
neu_group_get_read_tag(find_group(driver, cmd->group)->group);
neu_req_read_group_t *cmd = (neu_req_read_group_t *) &req[1];
neu_resp_read_group_t resp = { 0 };
neu_group_t * group = find_group(driver, cmd->group)->group;
UT_array * tags = neu_group_get_read_tag(group);
resp.n_tag = utarray_len(tags);
resp.tags = calloc(1, utarray_len(tags) * sizeof(neu_resp_tag_value_t));
read_group(driver->cache, cmd->group, tags, resp.tags);
if (driver->adapter.state != NEU_NODE_RUNNING_STATE_RUNNING) {
utarray_foreach(tags, neu_datatag_t *, tag)
{
int index = utarray_eltidx(tags, tag);
strcpy(resp.tags[index].tag, tag->name);
resp.tags[index].value.type = NEU_TYPE_ERROR;
resp.tags[index].value.value.i32 = NEU_ERR_NODE_NOT_RUNNING;
}
} else {
read_group(driver->adapter.timestamp,
neu_group_get_interval(group) *
NEU_DRIVER_TAG_CACHE_EXPIRE_TIME,
driver->cache, cmd->group, tags, resp.tags);
}
strcpy(resp.driver, cmd->driver);
strcpy(resp.group, cmd->group);
@@ -517,7 +529,10 @@ static int report_callback(void *usr_data)
strcpy(data->driver, group->driver->adapter.name);
strcpy(data->group, group->name);
data->n_tag = utarray_len(tags);
read_group(group->driver->cache, group->name, tags,
read_group(group->driver->adapter.timestamp,
neu_group_get_interval(group->group) *
NEU_DRIVER_TAG_CACHE_EXPIRE_TIME,
group->driver->cache, group->name, tags,
(neu_resp_tag_value_t *) &data[1]);
group->driver->adapter.cb_funs.response(&group->driver->adapter, &header,
@@ -580,7 +595,8 @@ static int read_callback(void *usr_data)
return 0;
}
static void read_group(neu_driver_cache_t *cache, const char *group,
static void read_group(int64_t timestamp, int64_t timeout,
neu_driver_cache_t *cache, const char *group,
UT_array *tags, neu_resp_tag_value_t *datas)
{
utarray_foreach(tags, neu_datatag_t *, tag)
@@ -602,7 +618,12 @@ static void read_group(neu_driver_cache_t *cache, const char *group,
continue;
}
datas[index].value.type = tag->type;
datas[index].value.value = value.value;
if ((timestamp - value.timestamp) > timeout) {
datas[index].value.type = NEU_TYPE_ERROR;
datas[index].value.value.i32 = NEU_ERR_PLUGIN_TAG_EXPIRED;
} else {
datas[index].value.type = tag->type;
datas[index].value.value = value.value;
}
}
}