diff --git a/include/neuron/neuron.h b/include/neuron/neuron.h index 35ada628..e3a87d5d 100644 --- a/include/neuron/neuron.h +++ b/include/neuron/neuron.h @@ -28,27 +28,34 @@ extern "C" { #include #include -#include "adapter.h" #include "errcodes.h" -#include "file.h" -#include "plugin.h" -#include "tag.h" + +#include "utils/async_queue.h" +#include "utils/protocol_buf.h" + #include "utils/base64.h" +#include "utils/log.h" +#include "utils/zlog.h" + +#include "utils/utarray.h" +#include "utils/utextend.h" +#include "utils/uthash.h" +#include "utils/utlist.h" #include "connection/mqtt_client_intf.h" #include "connection/neu_connection.h" #include "event/event.h" - #include "json/neu_json_param.h" -#include "utils/async_queue.h" -#include "utils/log.h" -#include "utils/protocol_buf.h" -#include "utils/utextend.h" - #include "define.h" + +#include "adapter.h" +#include "plugin.h" +#include "tag.h" #include "tag_sort.h" +#include "file.h" + #ifdef __cplusplus } #endif diff --git a/plugins/modbus/modbus_req.c b/plugins/modbus/modbus_req.c index 41483bab..2134ef7f 100644 --- a/plugins/modbus/modbus_req.c +++ b/plugins/modbus/modbus_req.c @@ -22,6 +22,7 @@ #include "modbus_req.h" struct modbus_group_data { + char * group; UT_array * tags; modbus_read_cmd_sort_t *cmd_sort; }; @@ -75,6 +76,7 @@ int modbus_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group, utarray_push_back(gd->tags, &p); } + gd->group = strdup(group->group_name); gd->cmd_sort = modbus_tag_sort(gd->tags, max_byte); } else { struct modbus_group_data *gd = @@ -251,8 +253,8 @@ int modbus_value_handle(void *ctx, uint8_t slave_id, uint16_t n_byte, } } - plugin->common.adapter_callbacks->driver.update(plugin->common.adapter, - (*p_tag)->name, dvalue); + plugin->common.adapter_callbacks->driver.update( + plugin->common.adapter, gd->group, (*p_tag)->name, dvalue); } return 0; } diff --git a/src/adapter/driver/cache.c b/src/adapter/driver/cache.c index fe282cb6..3b781795 100644 --- a/src/adapter/driver/cache.c +++ b/src/adapter/driver/cache.c @@ -18,9 +18,11 @@ **/ #include -#include #include +#include +#include + #include "utils/uthash.h" #include "define.h" @@ -28,6 +30,11 @@ #include "cache.h" +typedef struct { + char group[NEU_GROUP_NAME_LEN]; + char tag[NEU_TAG_NAME_LEN]; +} tkey_t; + struct elem { int32_t error; int64_t timestamp; @@ -35,21 +42,31 @@ struct elem { uint16_t n_byte; uint8_t *bytes; - char name[NEU_TAG_NAME_LEN]; + tkey_t key; UT_hash_handle hh; }; struct neu_driver_cache { - pthread_mutex_t mtx; + nng_mtx *mtx; struct elem *table; }; +inline static tkey_t to_key(const char *group, const char *tag) +{ + tkey_t key = { 0 }; + + strcpy(key.group, group); + strcpy(key.tag, tag); + + return key; +} + neu_driver_cache_t *neu_driver_cache_new() { neu_driver_cache_t *cache = calloc(1, sizeof(neu_driver_cache_t)); - pthread_mutex_init(&cache->mtx, NULL); + nng_mtx_alloc(&cache->mtx); return cache; } @@ -59,7 +76,7 @@ void neu_driver_cache_destroy(neu_driver_cache_t *cache) struct elem *elem = NULL; struct elem *tmp = NULL; - pthread_mutex_lock(&cache->mtx); + nng_mtx_lock(cache->mtx); HASH_ITER(hh, cache->table, elem, tmp) { HASH_DEL(cache->table, elem); @@ -68,48 +85,57 @@ void neu_driver_cache_destroy(neu_driver_cache_t *cache) } free(elem); } - pthread_mutex_unlock(&cache->mtx); + nng_mtx_unlock(cache->mtx); - pthread_mutex_destroy(&cache->mtx); + nng_mtx_free(cache->mtx); free(cache); } -void neu_driver_cache_error(neu_driver_cache_t *cache, const char *key, - int64_t timestamp, int32_t error) +void neu_driver_cache_error(neu_driver_cache_t *cache, const char *group, + const char *tag, int64_t timestamp, int32_t error) { struct elem *elem = NULL; + tkey_t key = to_key(group, tag); - pthread_mutex_lock(&cache->mtx); - HASH_FIND_STR(cache->table, key, elem); + nng_mtx_lock(cache->mtx); + HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem); if (elem == NULL) { elem = calloc(1, sizeof(struct elem)); - strncpy(elem->name, key, sizeof(elem->name)); - HASH_ADD_STR(cache->table, name, elem); + strcpy(elem->key.group, group); + strcpy(elem->key.tag, tag); + + HASH_ADD(hh, cache->table, key, sizeof(tkey_t), elem); } elem->timestamp = timestamp; elem->error = error; - pthread_mutex_unlock(&cache->mtx); + nng_mtx_unlock(cache->mtx); } -void neu_driver_cache_update(neu_driver_cache_t *cache, const char *key, - int64_t timestamp, uint16_t n_byte, uint8_t *bytes) +void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group, + const char *tag, int64_t timestamp, + uint16_t n_byte, uint8_t *bytes) { struct elem *elem = NULL; + tkey_t key = to_key(group, tag); - pthread_mutex_lock(&cache->mtx); - HASH_FIND_STR(cache->table, key, elem); + nng_mtx_lock(cache->mtx); + HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem); if (elem == NULL) { elem = calloc(1, sizeof(struct elem)); - strncpy(elem->name, key, sizeof(elem->name)); + + strcpy(elem->key.group, group); + strcpy(elem->key.tag, tag); + assert(n_byte != 0); assert(n_byte <= NEU_VALUE_SIZE); elem->n_byte = n_byte; elem->bytes = calloc(elem->n_byte, 1); - HASH_ADD_STR(cache->table, name, elem); + + HASH_ADD(hh, cache->table, key, sizeof(tkey_t), elem); } elem->error = 0; @@ -130,17 +156,18 @@ void neu_driver_cache_update(neu_driver_cache_t *cache, const char *key, memcpy(elem->bytes, bytes, elem->n_byte); - pthread_mutex_unlock(&cache->mtx); + nng_mtx_unlock(cache->mtx); } -int neu_driver_cache_get(neu_driver_cache_t *cache, const char *key, - neu_driver_cache_value_t *value) +int neu_driver_cache_get(neu_driver_cache_t *cache, const char *group, + const char *tag, neu_driver_cache_value_t *value) { struct elem *elem = NULL; int ret = -1; + tkey_t key = to_key(group, tag); - pthread_mutex_lock(&cache->mtx); - HASH_FIND_STR(cache->table, key, elem); + nng_mtx_lock(cache->mtx); + HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem); if (elem != NULL) { value->error = elem->error; @@ -151,17 +178,19 @@ int neu_driver_cache_get(neu_driver_cache_t *cache, const char *key, ret = 0; } - pthread_mutex_unlock(&cache->mtx); + nng_mtx_unlock(cache->mtx); return ret; } -void neu_driver_cache_del(neu_driver_cache_t *cache, const char *key) +void neu_driver_cache_del(neu_driver_cache_t *cache, const char *group, + const char *tag) { struct elem *elem = NULL; + tkey_t key = to_key(group, tag); - pthread_mutex_lock(&cache->mtx); - HASH_FIND_STR(cache->table, key, elem); + nng_mtx_lock(cache->mtx); + HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem); if (elem != NULL) { HASH_DEL(cache->table, elem); @@ -171,5 +200,5 @@ void neu_driver_cache_del(neu_driver_cache_t *cache, const char *key) free(elem); } - pthread_mutex_unlock(&cache->mtx); + nng_mtx_unlock(cache->mtx); } diff --git a/src/adapter/driver/cache.h b/src/adapter/driver/cache.h index 734caafa..4f5e4c63 100644 --- a/src/adapter/driver/cache.h +++ b/src/adapter/driver/cache.h @@ -29,12 +29,13 @@ typedef struct neu_driver_cache neu_driver_cache_t; neu_driver_cache_t *neu_driver_cache_new(); void neu_driver_cache_destroy(neu_driver_cache_t *cache); -void neu_driver_cache_error(neu_driver_cache_t *cache, const char *key, - int64_t timestamp, int32_t error); -void neu_driver_cache_update(neu_driver_cache_t *cache, const char *key, - int64_t timestamp, uint16_t n_byte, - uint8_t *bytes); -void neu_driver_cache_del(neu_driver_cache_t *cache, const char *key); +void neu_driver_cache_error(neu_driver_cache_t *cache, const char *group, + const char *tag, int64_t timestamp, int32_t error); +void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group, + const char *tag, int64_t timestamp, + uint16_t n_byte, uint8_t *bytes); +void neu_driver_cache_del(neu_driver_cache_t *cache, const char *group, + const char *tag); typedef struct { uint16_t n_byte; @@ -44,7 +45,7 @@ typedef struct { int64_t timestamp; } neu_driver_cache_value_t; -int neu_driver_cache_get(neu_driver_cache_t *cache, const char *key, - neu_driver_cache_value_t *value); +int neu_driver_cache_get(neu_driver_cache_t *cache, const char *group, + const char *tag, neu_driver_cache_value_t *value); #endif diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 5558a2d0..f8e1c602 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -61,12 +61,12 @@ 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, UT_array *tags, - neu_resp_tag_value_t *datas); +static void read_group(neu_driver_cache_t *cache, const char *group, + UT_array *tags, neu_resp_tag_value_t *datas); static UT_array *load_tags(UT_array *tags); static void free_tags(UT_array *tags); -static void update(neu_adapter_t *adapter, const char *name, - neu_dvalue_t value); +static void update(neu_adapter_t *adapter, const char *group, const char *tag, + neu_dvalue_t value); static void write_response(neu_adapter_t *adapter, void *r, neu_error error); static group_t *find_group(neu_adapter_driver_t *driver, const char *name); static void write_response(neu_adapter_t *adapter, void *r, neu_error error) @@ -79,7 +79,8 @@ static void write_response(neu_adapter_t *adapter, void *r, neu_error error) adapter->cb_funs.response(adapter, req, &nerror); } -static void update(neu_adapter_t *adapter, const char *name, neu_dvalue_t value) +static void update(neu_adapter_t *adapter, const char *group, const char *tag, + neu_dvalue_t value) { neu_adapter_driver_t *driver = (neu_adapter_driver_t *) adapter; uint16_t n_byte = 0; @@ -118,9 +119,9 @@ static void update(neu_adapter_t *adapter, const char *name, neu_dvalue_t value) } if (value.type == NEU_TYPE_ERROR) { - neu_driver_cache_error(driver->cache, name, 0, value.value.i32); + neu_driver_cache_error(driver->cache, group, tag, 0, value.value.i32); } else { - neu_driver_cache_update(driver->cache, name, 0, n_byte, + neu_driver_cache_update(driver->cache, group, tag, 0, n_byte, value.value.bytes); } } @@ -204,7 +205,7 @@ void neu_adapter_driver_read_group(neu_adapter_driver_t *driver, resp.n_tag = utarray_len(tags); resp.tags = calloc(1, utarray_len(tags) * sizeof(neu_resp_tag_value_t)); - read_group(driver->cache, tags, resp.tags); + read_group(driver->cache, cmd->group, tags, resp.tags); strcpy(resp.driver, cmd->driver); strcpy(resp.group, cmd->group); @@ -328,7 +329,7 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name) utarray_foreach(find->grp.tags, neu_datatag_t **, p_tag) { - neu_driver_cache_del(driver->cache, (*p_tag)->name); + neu_driver_cache_del(driver->cache, name, (*p_tag)->name); } free_tags(find->grp.tags); @@ -516,7 +517,8 @@ static int report_callback(void *usr_data) strcpy(data->driver, neu_adapter_get_name((neu_adapter_t *) group->driver)); strcpy(data->group, group->name); data->n_tag = utarray_len(tags); - read_group(group->driver->cache, tags, (neu_resp_tag_value_t *) &data[1]); + read_group(group->driver->cache, group->name, tags, + (neu_resp_tag_value_t *) &data[1]); group->driver->adapter.cb_funs.response(&group->driver->adapter, &header, data); @@ -538,7 +540,7 @@ static void group_change(void *arg, int64_t timestamp, UT_array *tags, utarray_foreach(group->grp.tags, neu_datatag_t **, p_tag) { - neu_driver_cache_del(group->driver->cache, (*p_tag)->name); + neu_driver_cache_del(group->driver->cache, group->name, (*p_tag)->name); } neu_plugin_group_t grp = { @@ -583,8 +585,8 @@ static int read_callback(void *usr_data) return 0; } -static void read_group(neu_driver_cache_t *cache, UT_array *tags, - neu_resp_tag_value_t *datas) +static void read_group(neu_driver_cache_t *cache, const char *group, + UT_array *tags, neu_resp_tag_value_t *datas) { utarray_foreach(tags, neu_datatag_t *, tag) { @@ -593,7 +595,7 @@ static void read_group(neu_driver_cache_t *cache, UT_array *tags, strcpy(datas[index].tag, tag->name); - if (neu_driver_cache_get(cache, tag->name, &value) != 0) { + if (neu_driver_cache_get(cache, group, tag->name, &value) != 0) { datas[index].value.type = NEU_TYPE_ERROR; datas[index].value.value.i32 = NEU_ERR_PLUGIN_READ_FAILURE; continue;