diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index 4ac562ed..7044cf00 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -86,6 +86,9 @@ typedef struct adapter_callbacks { void (*update_im)(neu_adapter_t *adapter, const char *group, const char *tag, neu_dvalue_t value, neu_tag_meta_t *metas, int n_meta); + void (*update_im_f_m)(neu_adapter_t *adapter, const char *group, + const char *tag, neu_dvalue_t value, + neu_tag_meta_t *metas, int n_meta); void (*scan_tags_response)(neu_adapter_t *adapter, void *r, neu_resp_scan_tags_t *resp_scan); void (*test_read_tag_response)(neu_adapter_t *adapter, void *r, diff --git a/src/adapter/driver/cache.c b/src/adapter/driver/cache.c index 1fb18a3b..a4e8a84a 100644 --- a/src/adapter/driver/cache.c +++ b/src/adapter/driver/cache.c @@ -219,14 +219,15 @@ void *neu_driver_cache_get_trace(neu_driver_cache_t *cache, const char *group) return trace; } -void neu_driver_cache_update_change(neu_driver_cache_t *cache, +bool neu_driver_cache_update_change(neu_driver_cache_t *cache, const char *group, const char *tag, int64_t timestamp, neu_dvalue_t value, neu_tag_meta_t *metas, int n_meta, bool change) { - struct elem *elem = NULL; - tkey_t key = to_key(group, tag); + struct elem *elem = NULL; + tkey_t key = to_key(group, tag); + bool tag_changed = false; pthread_mutex_lock(&cache->mtx); HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem); @@ -241,6 +242,7 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, (sub_filter_err && elem->value.type != value.type && elem->value.type != NEU_TYPE_ERROR)) { elem->changed = true; + tag_changed = true; } else if (sub_filter_err && elem->value.type != value.type && elem->value.type == NEU_TYPE_ERROR) { switch (value.type) { @@ -264,17 +266,20 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, if (memcmp(&elem->value_old.value, &value.value, sizeof(value.value)) != 0) { elem->changed = true; + tag_changed = true; } break; case NEU_TYPE_BYTES: if (elem->value_old.value.bytes.length != value.value.bytes.length) { elem->changed = true; + tag_changed = true; } else { if (memcpy(elem->value_old.value.bytes.bytes, value.value.bytes.bytes, value.value.bytes.length) != 0) { elem->changed = true; + tag_changed = true; } } break; @@ -282,11 +287,13 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, if (elem->value_old.value.bools.length != value.value.bools.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp(elem->value_old.value.bools.bools, value.value.bools.bools, value.value.bools.length) != 0) { elem->changed = true; + tag_changed = true; } } break; @@ -294,11 +301,13 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, if (elem->value_old.value.i8s.length != value.value.i8s.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp(elem->value_old.value.i8s.i8s, value.value.i8s.i8s, value.value.i8s.length) != 0) { elem->changed = true; + tag_changed = true; } } break; @@ -306,11 +315,13 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, if (elem->value_old.value.u8s.length != value.value.u8s.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp(elem->value_old.value.u8s.u8s, value.value.u8s.u8s, value.value.u8s.length) != 0) { elem->changed = true; + tag_changed = true; } } break; @@ -318,12 +329,14 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, if (elem->value_old.value.i16s.length != value.value.i16s.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp(elem->value_old.value.i16s.i16s, value.value.i16s.i16s, value.value.i16s.length * sizeof(int16_t)) != 0) { elem->changed = true; + tag_changed = true; } } break; @@ -331,12 +344,14 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, if (elem->value_old.value.u16s.length != value.value.u16s.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp(elem->value_old.value.u16s.u16s, value.value.u16s.u16s, value.value.u16s.length * sizeof(uint16_t)) != 0) { elem->changed = true; + tag_changed = true; } } break; @@ -344,12 +359,14 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, if (elem->value_old.value.i32s.length != value.value.i32s.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp(elem->value_old.value.i32s.i32s, value.value.i32s.i32s, value.value.i32s.length * sizeof(int32_t)) != 0) { elem->changed = true; + tag_changed = true; } } break; @@ -357,12 +374,14 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, if (elem->value_old.value.u32s.length != value.value.u32s.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp(elem->value_old.value.u32s.u32s, value.value.u32s.u32s, value.value.u32s.length * sizeof(uint32_t)) != 0) { elem->changed = true; + tag_changed = true; } } break; @@ -370,12 +389,14 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, if (elem->value_old.value.i64s.length != value.value.i64s.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp(elem->value_old.value.i64s.i64s, value.value.i64s.i64s, value.value.i64s.length * sizeof(int64_t)) != 0) { elem->changed = true; + tag_changed = true; } } break; @@ -383,12 +404,14 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, if (elem->value_old.value.u64s.length != value.value.u64s.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp(elem->value_old.value.u64s.u64s, value.value.u64s.u64s, value.value.u64s.length * sizeof(uint64_t)) != 0) { elem->changed = true; + tag_changed = true; } } break; @@ -396,11 +419,13 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, if (elem->value_old.value.f32s.length != value.value.f32s.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp(elem->value_old.value.f32s.f32s, value.value.f32s.f32s, value.value.f32s.length * sizeof(float)) != 0) { elem->changed = true; + tag_changed = true; } } break; @@ -408,11 +433,13 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, if (elem->value_old.value.f64s.length != value.value.f64s.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp(elem->value_old.value.f64s.f64s, value.value.f64s.f64s, value.value.f64s.length * sizeof(double)) != 0) { elem->changed = true; + tag_changed = true; } } break; @@ -420,11 +447,13 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, if (elem->value_old.value.strs.length != value.value.strs.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp(elem->value_old.value.strs.strs, value.value.strs.strs, value.value.strs.length * sizeof(char *)) != 0) { elem->changed = true; + tag_changed = true; } } break; @@ -432,6 +461,7 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, if (json_equal(elem->value_old.value.json, value.value.json) != 0) { elem->changed = true; + tag_changed = true; } break; } @@ -439,11 +469,13 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, if (elem->value_old.value.ptr.length != value.value.ptr.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp(elem->value_old.value.ptr.ptr, value.value.ptr.ptr, value.value.ptr.length) != 0) { elem->changed = true; + tag_changed = true; } } @@ -453,10 +485,12 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, if (elem->value_old.precision == 0) { elem->changed = elem->value_old.value.f32 != value.value.f32; + tag_changed = elem->changed; } else { if (fabs(elem->value_old.value.f32 - value.value.f32) > pow(0.1, elem->value_old.precision)) { elem->changed = true; + tag_changed = true; } } break; @@ -464,10 +498,12 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, if (elem->value_old.precision == 0) { elem->changed = elem->value_old.value.d64 != value.value.d64; + tag_changed = elem->changed; } else { if (fabs(elem->value_old.value.d64 - value.value.d64) > pow(0.1, elem->value_old.precision)) { elem->changed = true; + tag_changed = true; } } @@ -497,17 +533,20 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, if (memcmp(&elem->value.value, &value.value, sizeof(value.value)) != 0) { elem->changed = true; + tag_changed = true; } break; case NEU_TYPE_BYTES: if (elem->value.value.bytes.length != value.value.bytes.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp(elem->value.value.bytes.bytes, value.value.bytes.bytes, value.value.bytes.length) != 0) { elem->changed = true; + tag_changed = true; } } break; @@ -515,140 +554,166 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, if (elem->value.value.bools.length != value.value.bools.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp(elem->value.value.bools.bools, value.value.bools.bools, value.value.bools.length) != 0) { elem->changed = true; + tag_changed = true; } } break; case NEU_TYPE_ARRAY_INT8: if (elem->value.value.i8s.length != value.value.i8s.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp(elem->value.value.i8s.i8s, value.value.i8s.i8s, value.value.i8s.length) != 0) { elem->changed = true; + tag_changed = true; } } break; case NEU_TYPE_ARRAY_UINT8: if (elem->value.value.u8s.length != value.value.u8s.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp(elem->value.value.u8s.u8s, value.value.u8s.u8s, value.value.u8s.length) != 0) { elem->changed = true; + tag_changed = true; } } break; case NEU_TYPE_ARRAY_INT16: if (elem->value.value.i16s.length != value.value.i16s.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp( elem->value.value.i16s.i16s, value.value.i16s.i16s, value.value.i16s.length * sizeof(int16_t)) != 0) { elem->changed = true; + tag_changed = true; } } break; case NEU_TYPE_ARRAY_UINT16: if (elem->value.value.u16s.length != value.value.u16s.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp( elem->value.value.u16s.u16s, value.value.u16s.u16s, value.value.u16s.length * sizeof(uint16_t)) != 0) { elem->changed = true; + tag_changed = true; } } break; case NEU_TYPE_ARRAY_INT32: if (elem->value.value.i32s.length != value.value.i32s.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp( elem->value.value.i32s.i32s, value.value.i32s.i32s, value.value.i32s.length * sizeof(int32_t)) != 0) { elem->changed = true; + tag_changed = true; } } break; case NEU_TYPE_ARRAY_UINT32: if (elem->value.value.u32s.length != value.value.u32s.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp( elem->value.value.u32s.u32s, value.value.u32s.u32s, value.value.u32s.length * sizeof(uint32_t)) != 0) { elem->changed = true; + tag_changed = true; } } break; case NEU_TYPE_ARRAY_INT64: if (elem->value.value.i64s.length != value.value.i64s.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp( elem->value.value.i64s.i64s, value.value.i64s.i64s, value.value.i64s.length * sizeof(int64_t)) != 0) { elem->changed = true; + tag_changed = true; } } break; case NEU_TYPE_ARRAY_UINT64: if (elem->value.value.u64s.length != value.value.u64s.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp( elem->value.value.u64s.u64s, value.value.u64s.u64s, value.value.u64s.length * sizeof(uint64_t)) != 0) { elem->changed = true; + tag_changed = true; } } break; case NEU_TYPE_ARRAY_FLOAT: if (elem->value.value.f32s.length != value.value.f32s.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp(elem->value.value.f32s.f32s, value.value.f32s.f32s, value.value.f32s.length * sizeof(float)) != 0) { elem->changed = true; + tag_changed = true; } } break; case NEU_TYPE_ARRAY_DOUBLE: if (elem->value.value.f64s.length != value.value.f64s.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp(elem->value.value.f64s.f64s, value.value.f64s.f64s, value.value.f64s.length * sizeof(double)) != 0) { elem->changed = true; + tag_changed = true; } } break; case NEU_TYPE_ARRAY_STRING: if (elem->value.value.strs.length != value.value.strs.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp(elem->value.value.strs.strs, value.value.strs.strs, value.value.strs.length * sizeof(char *)) != 0) { elem->changed = true; + tag_changed = true; } } break; case NEU_TYPE_PTR: { if (elem->value.value.ptr.length != value.value.ptr.length) { elem->changed = true; + tag_changed = true; } else { if (memcmp(elem->value.value.ptr.ptr, value.value.ptr.ptr, value.value.ptr.length) != 0) { elem->changed = true; + tag_changed = true; } } @@ -657,32 +722,38 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, case NEU_TYPE_CUSTOM: { if (json_equal(elem->value.value.json, value.value.json) != 0) { elem->changed = true; + tag_changed = true; } break; } case NEU_TYPE_FLOAT: if (elem->value.precision == 0) { elem->changed = elem->value.value.f32 != value.value.f32; + tag_changed = elem->changed; } else { if (fabs(elem->value.value.f32 - value.value.f32) > pow(0.1, elem->value.precision)) { elem->changed = true; + tag_changed = true; } } break; case NEU_TYPE_DOUBLE: if (elem->value.precision == 0) { elem->changed = elem->value.value.d64 != value.value.d64; + tag_changed = elem->changed; } else { if (fabs(elem->value.value.d64 - value.value.d64) > pow(0.1, elem->value.precision)) { elem->changed = true; + tag_changed = true; } } break; case NEU_TYPE_ERROR: elem->changed = true; + tag_changed = true; break; } } @@ -786,6 +857,7 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, } pthread_mutex_unlock(&cache->mtx); + return tag_changed; } void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group, diff --git a/src/adapter/driver/cache.h b/src/adapter/driver/cache.h index 3468a9a2..76c24442 100644 --- a/src/adapter/driver/cache.h +++ b/src/adapter/driver/cache.h @@ -35,7 +35,7 @@ void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group, const char *tag, int64_t timestamp, neu_dvalue_t value, neu_tag_meta_t *metas, int n_meta); -void neu_driver_cache_update_change(neu_driver_cache_t *cache, +bool neu_driver_cache_update_change(neu_driver_cache_t *cache, const char *group, const char *tag, int64_t timestamp, neu_dvalue_t value, neu_tag_meta_t *metas, int n_meta, diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 159c098e..672e49ee 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -113,6 +113,9 @@ static void update(neu_adapter_t *adapter, const char *group, const char *tag, static void update_im(neu_adapter_t *adapter, const char *group, const char *tag, neu_dvalue_t value, neu_tag_meta_t *metas, int n_meta); +static void update_im_f_m(neu_adapter_t *adapter, const char *group, + const char *tag, neu_dvalue_t value, + neu_tag_meta_t *metas, int n_meta); static void update_with_meta(neu_adapter_t *adapter, const char *group, const char *tag, neu_dvalue_t value, neu_tag_meta_t *metas, int n_meta); @@ -439,6 +442,107 @@ static void update_with_trace(neu_adapter_t *adapter, const char *group, } } +static void update_im_f_m(neu_adapter_t *adapter, const char *group, + const char *tag, neu_dvalue_t value, + neu_tag_meta_t *metas, int n_meta) +{ + neu_adapter_driver_t *driver = (neu_adapter_driver_t *) adapter; + + if (tag == NULL) { + nlog_warn("update_im_f_m tag is null"); + return; + } + + group_t *find = NULL; + + HASH_FIND_STR(driver->groups, group, find); + if (find == NULL) { + nlog_error( + "update_im_f_m group not found, driver: %s, group: %s, tag: %s", + driver->adapter.name, group, tag); + return; + } + + bool changed = neu_driver_cache_update_change(driver->cache, group, tag, + global_timestamp, value, + metas, n_meta, false); + driver->adapter.cb_funs.update_metric(&driver->adapter, + NEU_METRIC_TAG_READS_TOTAL, 1, NULL); + if (value.type == NEU_TYPE_ERROR) { + return; + } + + if (!changed) { + return; + } + + UT_array *tags = neu_adapter_driver_get_ptag(driver, group, tag); + if (tags == NULL) { + return; + } + neu_datatag_t *first = utarray_front(tags); + + if (first == NULL) { + utarray_free(tags); + nlog_debug("update immediately, driver: %s, " + "group: %s, tag: %s, type: %s, " + "timestamp: %" PRId64, + driver->adapter.name, group, tag, + neu_type_string(value.type), global_timestamp); + return; + } + + nlog_debug("update and report immediately, driver: %s, " + "group: %s, tag: %s, type: %s, " + "timestamp: %" PRId64, + driver->adapter.name, group, tag, neu_type_string(value.type), + global_timestamp); + + neu_reqresp_head_t header = { + .type = NEU_REQRESP_TRANS_DATA, + }; + neu_reqresp_trans_data_t *data = + calloc(1, sizeof(neu_reqresp_trans_data_t)); + + data->driver = strdup(driver->adapter.name); + data->group = strdup(group); + utarray_new(data->tags, neu_resp_tag_value_meta_icd()); + + read_report_group(global_timestamp, 0, + neu_adapter_get_tag_cache_type(&driver->adapter), + driver->cache, group, tags, data->tags); + + if (utarray_len(data->tags) > 0) { + + data->ctx = calloc(1, sizeof(neu_reqresp_trans_data_ctx_t)); + data->ctx->index = utarray_len(find->apps); + pthread_mutex_init(&data->ctx->mtx, NULL); + + if (utarray_len(find->apps) > 0) { + pthread_mutex_lock(&find->apps_mtx); + + utarray_foreach(find->apps, sub_app_t *, app) + { + if (driver->adapter.cb_funs.responseto( + &driver->adapter, &header, data, app->addr) != 0) { + neu_trans_data_free(data); + } + } + + pthread_mutex_unlock(&find->apps_mtx); + } else { + neu_trans_data_free(data); + } + } else { + utarray_free(data->tags); + free(data->group); + free(data->driver); + } + + utarray_free(tags); + free(data); +} + static void update_im(neu_adapter_t *adapter, const char *group, const char *tag, neu_dvalue_t value, neu_tag_meta_t *metas, int n_meta) @@ -580,6 +684,7 @@ neu_adapter_driver_t *neu_adapter_driver_create() driver->adapter.cb_funs.driver.fdown_open_response = fdown_open_response; driver->adapter.cb_funs.driver.fup_data_response = fup_data_response; driver->adapter.cb_funs.driver.update_im = update_im; + driver->adapter.cb_funs.driver.update_im_f_m = update_im_f_m; driver->adapter.cb_funs.driver.update_with_trace = update_with_trace; driver->adapter.cb_funs.driver.update_with_meta = update_with_meta; driver->adapter.cb_funs.driver.scan_tags_response = scan_tags_response;