From bab2a8246983dce2b7b343f843433583355789a4 Mon Sep 17 00:00:00 2001 From: fengzero Date: Fri, 4 Nov 2022 06:51:37 +0000 Subject: [PATCH] improve(timestamp): remove update timestamp from adapter --- include/neuron/plugin.h | 3 ++- include/neuron/utils/time.h | 4 ++-- plugins/ekuiper/json_rw.c | 2 +- plugins/mqtt/command/heartbeat.c | 3 ++- plugins/mqtt/command/read_write.c | 2 +- src/adapter/adapter.c | 24 ------------------------ src/adapter/adapter_internal.h | 9 +++------ src/adapter/driver/driver.c | 16 ++++++++-------- src/base/neu_plugin_common.c | 2 -- src/core/manager.c | 20 ++++++++++++++++++++ src/core/manager_internal.h | 10 ++++++---- src/main.c | 2 ++ 12 files changed, 47 insertions(+), 50 deletions(-) diff --git a/include/neuron/plugin.h b/include/neuron/plugin.h index 62720469..c7f56cc8 100644 --- a/include/neuron/plugin.h +++ b/include/neuron/plugin.h @@ -34,6 +34,8 @@ extern "C" { #define NEURON_PLUGIN_VER_1_0 100 #define NEURON_PLUGIN_VER_2_0 200 +static volatile int64_t global_timestamp; + typedef struct neu_plugin_common { uint32_t magic; @@ -44,7 +46,6 @@ typedef struct neu_plugin_common { neu_node_link_state_e link_state; uint16_t tag_size; uint32_t tag_all_size; - int64_t timestamp; zlog_category_t *log; } neu_plugin_common_t; diff --git a/include/neuron/utils/time.h b/include/neuron/utils/time.h index b76de52a..82489816 100644 --- a/include/neuron/utils/time.h +++ b/include/neuron/utils/time.h @@ -27,11 +27,11 @@ extern "C" { #include #include -static inline uint64_t neu_time_ms() +static inline int64_t neu_time_ms() { struct timeval tv = { 0 }; gettimeofday(&tv, NULL); - return tv.tv_sec * 1000 + tv.tv_usec / 1000; + return (int64_t) tv.tv_sec * 1000 + (int64_t) tv.tv_usec / 1000; } #ifdef __cplusplus diff --git a/plugins/ekuiper/json_rw.c b/plugins/ekuiper/json_rw.c index ccd4901e..c217711c 100644 --- a/plugins/ekuiper/json_rw.c +++ b/plugins/ekuiper/json_rw.c @@ -198,7 +198,7 @@ int json_encode_read_resp(void *json_object, void *param) json_read_resp_header_t header = { .group_name = trans_data->group, .node_name = trans_data->driver, - .timestamp = plugin->common.timestamp }; + .timestamp = global_timestamp }; ret = json_encode_read_resp_header(json_object, &header); if (0 != ret) { diff --git a/plugins/mqtt/command/heartbeat.c b/plugins/mqtt/command/heartbeat.c index fffb900c..8c00173d 100644 --- a/plugins/mqtt/command/heartbeat.c +++ b/plugins/mqtt/command/heartbeat.c @@ -27,9 +27,10 @@ char *command_heartbeat_generate(neu_plugin_t *plugin, UT_array *states) { + (void) plugin; char * version = NEURON_VERSION; neu_json_states_head_t header = { .version = version, - .timpstamp = plugin->common.timestamp }; + .timpstamp = global_timestamp }; neu_json_states_t json = { 0 }; char * json_str = NULL; json.n_state = utarray_len(states); diff --git a/plugins/mqtt/command/read_write.c b/plugins/mqtt/command/read_write.c index eaf930f4..e27b17e2 100644 --- a/plugins/mqtt/command/read_write.c +++ b/plugins/mqtt/command/read_write.c @@ -169,7 +169,7 @@ char *command_rw_read_periodic_response(neu_plugin_t * plugin, char * json_str = NULL; neu_json_read_periodic_t header = { .group = (char *) data->group, .node = (char *) data->driver, - .timestamp = plugin->common.timestamp }; + .timestamp = global_timestamp }; neu_json_read_resp_t json = { 0 }; wrap_read_response_json(tags, len, &json); diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index f1d58582..f7a16d95 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -54,7 +54,6 @@ static int adapter_update_metric(neu_adapter_t *adapter, const char *group); inline static void reply(neu_adapter_t *adapter, neu_reqresp_head_t *header, void *data); -static int update_timestamp(void *usr_data); static int level_check(void *usr_data); static const adapter_callbacks_t callback_funs = { @@ -87,12 +86,6 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info) 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 = 10, - .cb = update_timestamp, - }; - neu_event_timer_param_t timer_level = { .second = 30, .millisecond = 0, @@ -179,8 +172,6 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info) timer_level.usr_data = (void *) adapter; adapter->timer_lev = neu_event_add_timer(adapter->events, timer_level); - 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); @@ -792,7 +783,6 @@ 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); neu_event_del_timer(adapter->events, adapter->timer_lev); nlog_info("Stop the adapter(%s)", adapter->name); @@ -1049,20 +1039,6 @@ static int level_check(void *usr_data) return 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 = (int64_t) tv.tv_sec * 1000 + (int64_t) tv.tv_usec / 1000; - adapter->timestamp = timestamp; - - neu_plugin_to_plugin_common(adapter->plugin)->timestamp = timestamp; - return 0; -} - void *neu_msg_gen(neu_reqresp_head_t *header, void *data) { nng_msg *msg = NULL; diff --git a/src/adapter/adapter_internal.h b/src/adapter/adapter_internal.h index e4edf4fd..27c9418b 100644 --- a/src/adapter/adapter_internal.h +++ b/src/adapter/adapter_internal.h @@ -34,8 +34,6 @@ struct neu_adapter { char *name; char *setting; - int64_t timestamp; - neu_node_running_state_e state; adapter_callbacks_t cb_funs; @@ -47,10 +45,9 @@ struct neu_adapter { nng_socket sock; nng_dialer dialer; - neu_events_t * events; - neu_event_io_t * nng_io; - neu_event_timer_t *timer; - int recv_fd; + neu_events_t * events; + neu_event_io_t *nng_io; + int recv_fd; neu_event_timer_t *timer_lev; int64_t timestamp_lev; diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 617ed030..37f6f4d3 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -97,7 +97,7 @@ static void update(neu_adapter_t *adapter, const char *group, const char *tag, utarray_foreach(tags, neu_datatag_t *, t) { neu_driver_cache_update(driver->cache, group, t->name, - driver->adapter.timestamp, value); + global_timestamp, value); } driver->adapter.cb_funs.update_metric(&driver->adapter, NEU_METRIC_TAG_READS_TOTAL, @@ -108,8 +108,8 @@ static void update(neu_adapter_t *adapter, const char *group, const char *tag, utarray_free(tags); } } else { - neu_driver_cache_update(driver->cache, group, tag, - driver->adapter.timestamp, value); + neu_driver_cache_update(driver->cache, group, tag, global_timestamp, + value); driver->adapter.cb_funs.update_metric( &driver->adapter, NEU_METRIC_TAG_READS_TOTAL, 1, NULL); driver->adapter.cb_funs.update_metric( @@ -119,7 +119,7 @@ static void update(neu_adapter_t *adapter, const char *group, const char *tag, nlog_debug( "update driver: %s, group: %s, tag: %s, type: %s, timestamp: %" PRId64, driver->adapter.name, group, tag, neu_type_string(value.type), - driver->adapter.timestamp); + global_timestamp); } neu_adapter_driver_t *neu_adapter_driver_create() @@ -212,7 +212,7 @@ void neu_adapter_driver_read_group(neu_adapter_driver_t *driver, } } else { - read_group(driver->adapter.timestamp, + read_group(global_timestamp, neu_group_get_interval(group) * NEU_DRIVER_TAG_CACHE_EXPIRE_TIME, driver->cache, cmd->group, tags, resp.tags); @@ -729,7 +729,7 @@ static int report_callback(void *usr_data) strcpy(data->driver, group->driver->adapter.name); strcpy(data->group, group->name); - data->n_tag = read_report_group(group->driver->adapter.timestamp, + data->n_tag = read_report_group(global_timestamp, neu_group_get_interval(group->group) * NEU_DRIVER_TAG_CACHE_EXPIRE_TIME, group->driver->cache, group->name, tags, @@ -802,12 +802,12 @@ static int read_callback(void *usr_data) } if (group->grp.tags != NULL) { - int64_t spend = group->driver->adapter.timestamp; + int64_t spend = global_timestamp; group->driver->adapter.module->intf_funs->driver.group_timer( group->driver->adapter.plugin, &group->grp); - spend = group->driver->adapter.timestamp - spend; + spend = global_timestamp - spend; nlog_info("%s-%s timer: %" PRId64, group->driver->adapter.name, group->name, spend); diff --git a/src/base/neu_plugin_common.c b/src/base/neu_plugin_common.c index 69d95bbe..c9920971 100644 --- a/src/base/neu_plugin_common.c +++ b/src/base/neu_plugin_common.c @@ -32,8 +32,6 @@ void neu_plugin_common_init(neu_plugin_common_t *common) common->magic = NEU_PLUGIN_MAGIC_NUMBER; common->link_state = NEU_NODE_LINK_STATE_DISCONNECTED; common->tag_size = 0; - common->timestamp = time(NULL); - common->timestamp *= 1000; // don't need much precision at init } bool neu_plugin_common_check(neu_plugin_t *plugin) diff --git a/src/core/manager.c b/src/core/manager.c index b8d0076b..1c1c4ecb 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -27,6 +27,7 @@ #include "event/event.h" #include "persist/persist.h" #include "utils/log.h" +#include "utils/time.h" #include "adapter.h" #include "adapter/adapter_internal.h" @@ -56,6 +57,7 @@ inline static void forward_msg_dup(neu_manager_t *manager, nng_msg *msg, inline static void forward_msg(neu_manager_t *manager, nng_msg *msg, const char *ndoe); static void start_static_adapter(neu_manager_t *manager, const char *name); +static int update_timestamp(void *usr_data); static int report_nodes_state(void *usr_data); static void start_single_adapter(neu_manager_t *manager, const char *name, const char *plugin_name, bool display); @@ -68,6 +70,12 @@ neu_manager_t *neu_manager_create() .usr_data = (void *) manager, .cb = manager_loop, }; + + neu_event_timer_param_t timestamp_timer_param = { + .second = 0, + .millisecond = 10, + .cb = update_timestamp, + }; neu_event_timer_param_t timer_param = { .second = 1, .millisecond = 0, @@ -130,6 +138,10 @@ neu_manager_t *neu_manager_create() timer_param.usr_data = (void *) manager; manager->timer = neu_event_add_timer(manager->events, timer_param); + timestamp_timer_param.usr_data = (void *) manager; + manager->timer_timestamp = + neu_event_add_timer(manager->events, timestamp_timer_param); + return manager; } @@ -142,6 +154,7 @@ void neu_manager_destroy(neu_manager_t *manager) neu_event_del_timer(manager->events, manager->timer); neu_event_del_timer(manager->events, manager->timer_lev); + neu_event_del_timer(manager->events, manager->timer_timestamp); strcpy(header.sender, "manager"); utarray_foreach(pipes, nng_pipe *, pipe) { @@ -756,3 +769,10 @@ static int report_nodes_state(void *usr_data) return 0; } + +static int update_timestamp(void *usr_data) +{ + (void) usr_data; + global_timestamp = neu_time_ms(); + return 0; +} \ No newline at end of file diff --git a/src/core/manager_internal.h b/src/core/manager_internal.h index e14d70ba..de37d43a 100644 --- a/src/core/manager_internal.h +++ b/src/core/manager_internal.h @@ -32,16 +32,18 @@ #include "subscribe.h" typedef struct neu_manager { - nng_socket socket; - neu_events_t * events; - neu_event_io_t * loop; - neu_event_timer_t *timer; + nng_socket socket; + neu_events_t * events; + neu_event_io_t *loop; neu_plugin_manager_t *plugin_manager; neu_node_manager_t * node_manager; neu_subscribe_mgr_t * subscribe_manager; neu_sub_msg_mgr_t * sub_msg_manager; + neu_event_timer_t *timer; + neu_event_timer_t *timer_timestamp; + neu_event_timer_t *timer_lev; int64_t timestamp_lev_manager; } neu_manager_t; diff --git a/src/main.c b/src/main.c index 90f74c94..6f6fd12c 100644 --- a/src/main.c +++ b/src/main.c @@ -24,6 +24,7 @@ #include "core/manager.h" #include "utils/log.h" +#include "utils/time.h" #include "argparse.h" #include "daemon.h" @@ -88,6 +89,7 @@ int main(int argc, char *argv[]) pid_t pid = 0; neu_cli_args_t args = { 0 }; + global_timestamp = neu_time_ms(); neu_cli_args_init(&args, argc, argv); disable_jwt = args.disable_auth;