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

refactor(metrics): add lock to avoid data race

Browse Source
This commit is contained in:
eeff
2024-02-05 17:17:24 +08:00
parent 31bd132398
commit 3d705aa772
3 changed files with 205 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/metrics.h
+176 -8
View File
@@ -26,11 +26,16 @@ extern "C" {
#include <stdint.h>
#include <pthread.h>
#include "define.h"
#include "type.h"
#include "utils/rolling_counter.h"
#include "utils/utextend.h"
#include "utils/uthash.h"
extern int64_t global_timestamp;
typedef enum {
NEU_METRIC_TYPE_COUNTER,
NEU_METRIC_TYPE_GAUAGE,
@@ -249,6 +254,7 @@ typedef struct {
// node metrics
typedef struct {
pthread_mutex_t lock; // spin lock
neu_node_type_e type; // node type
char * name; // node name
neu_metric_entry_t * entries; // node metric entries
@@ -285,6 +291,16 @@ typedef struct {
neu_metric_entry_t *registered_metrics;
} neu_metrics_t;
void neu_metrics_init();
void neu_metrics_add_node(const neu_adapter_t *adapter);
void neu_metrics_del_node(const neu_adapter_t *adapter);
int neu_metrics_register_entry(const char *name, const char *help,
neu_metric_type_e type);
void neu_metrics_unregister_entry(const char *name);
typedef void (*neu_metrics_cb_t)(const neu_metrics_t *metrics, void *data);
void neu_metrics_visist(neu_metrics_cb_t cb, void *data);
static inline const char *neu_metric_type_str(neu_metric_type_e type)
{
if (NEU_METRIC_TYPE_COUNTER == type) {
@@ -316,22 +332,41 @@ static inline void neu_group_metrics_free(neu_group_metrics_t *group_metrics)
HASH_ITER(hh, group_metrics->entries, e, tmp)
{
HASH_DEL(group_metrics->entries, e);
neu_metrics_unregister_entry(e->name);
neu_metric_entry_free(e);
}
free(group_metrics->name);
free(group_metrics);
}
static inline neu_node_metrics_t *
neu_node_metrics_new(neu_adapter_t *adapter, neu_node_type_e type, char *name)
{
neu_node_metrics_t *node_metrics =
(neu_node_metrics_t *) calloc(1, sizeof(*node_metrics));
if (NULL != node_metrics) {
pthread_mutex_init(&node_metrics->lock, NULL);
node_metrics->type = type;
node_metrics->name = name;
node_metrics->adapter = adapter;
// neu_metrics_add_node(adapter);
}
return node_metrics;
}
static inline void neu_node_metrics_free(neu_node_metrics_t *node_metrics)
{
if (NULL == node_metrics) {
return;
}
pthread_mutex_destroy(&node_metrics->lock);
neu_metric_entry_t *e = NULL, *etmp = NULL;
HASH_ITER(hh, node_metrics->entries, e, etmp)
{
HASH_DEL(node_metrics->entries, e);
neu_metrics_unregister_entry(e->name);
neu_metric_entry_free(e);
}
@@ -345,15 +380,148 @@ static inline void neu_node_metrics_free(neu_node_metrics_t *node_metrics)
free(node_metrics);
}
void neu_metrics_init();
void neu_metrics_add_node(const neu_adapter_t *adapter);
void neu_metrics_del_node(const neu_adapter_t *adapter);
int neu_metrics_register_entry(const char *name, const char *help,
neu_metric_type_e type);
void neu_metrics_unregister_entry(const char *name);
static inline int neu_node_metrics_add(neu_node_metrics_t *node_metrics,
const char *group_name, const char *name,
const char *help, neu_metric_type_e type,
uint64_t init)
{
int rv = 0;
typedef void (*neu_metrics_cb_t)(const neu_metrics_t *metrics, void *data);
void neu_metrics_visist(neu_metrics_cb_t cb, void *data);
if (0 > neu_metrics_register_entry(name, help, type)) {
return -1;
}
pthread_mutex_lock(&node_metrics->lock);
if (NULL == group_name) {
rv = neu_metric_entries_add(&node_metrics->entries, name, help, type,
init);
} else {
neu_group_metrics_t *group_metrics = NULL;
HASH_FIND_STR(node_metrics->group_metrics, group_name, group_metrics);
if (NULL != group_metrics) {
rv = neu_metric_entries_add(&group_metrics->entries, name, help,
type, init);
} else {
group_metrics =
(neu_group_metrics_t *) calloc(1, sizeof(*group_metrics));
if (NULL != group_metrics &&
NULL != (group_metrics->name = strdup(group_name))) {
HASH_ADD_STR(node_metrics->group_metrics, name, group_metrics);
rv = neu_metric_entries_add(&group_metrics->entries, name, help,
type, init);
} else {
free(group_metrics);
rv = -1;
}
}
}
pthread_mutex_unlock(&node_metrics->lock);
if (0 != rv) {
neu_metrics_unregister_entry(name);
}
return rv;
}
static inline int neu_node_metrics_update(neu_node_metrics_t *node_metrics,
const char * group,
const char *metric_name, uint64_t n)
{
neu_metric_entry_t *entry = NULL;
pthread_mutex_lock(&node_metrics->lock);
if (NULL == group) {
HASH_FIND_STR(node_metrics->entries, metric_name, entry);
} else if (NULL != node_metrics->group_metrics) {
neu_group_metrics_t *g = NULL;
HASH_FIND_STR(node_metrics->group_metrics, group, g);
if (NULL != g) {
HASH_FIND_STR(g->entries, metric_name, entry);
}
}
if (NULL == entry) {
pthread_mutex_unlock(&node_metrics->lock);
return -1;
}
if (NEU_METRIC_TYPE_COUNTER == entry->type) {
entry->value += n;
} else if (NEU_METRIC_TYPE_ROLLING_COUNTER == entry->type) {
entry->value =
neu_rolling_counter_inc(entry->rcnt, global_timestamp, n);
} else {
entry->value = n;
}
pthread_mutex_unlock(&node_metrics->lock);
return 0;
}
static inline void neu_node_metrics_reset(neu_node_metrics_t *node_metrics)
{
neu_metric_entry_t *entry = NULL;
pthread_mutex_lock(&node_metrics->lock);
HASH_LOOP(hh, node_metrics->entries, entry)
{
entry->value = 0;
if (NEU_METRIC_TYPE_ROLLING_COUNTER == entry->type) {
neu_rolling_counter_reset(entry->rcnt);
}
}
neu_group_metrics_t *g = NULL;
HASH_LOOP(hh, node_metrics->group_metrics, g)
{
HASH_LOOP(hh, g->entries, entry)
{
entry->value = 0;
if (NEU_METRIC_TYPE_ROLLING_COUNTER == entry->type) {
neu_rolling_counter_reset(entry->rcnt);
}
}
}
pthread_mutex_unlock(&node_metrics->lock);
}
static inline int
neu_node_metrics_update_group(neu_node_metrics_t *node_metrics,
const char * group_name,
const char * new_group_name)
{
int rv = -1;
neu_group_metrics_t *group_metrics = NULL;
pthread_mutex_lock(&node_metrics->lock);
HASH_FIND_STR(node_metrics->group_metrics, group_name, group_metrics);
if (NULL != group_metrics) {
char *name = strdup(new_group_name);
if (NULL != name) {
HASH_DEL(node_metrics->group_metrics, group_metrics);
free(group_metrics->name);
group_metrics->name = name;
HASH_ADD_STR(node_metrics->group_metrics, name, group_metrics);
rv = 0;
}
}
pthread_mutex_unlock(&node_metrics->lock);
return rv;
}
static inline void neu_node_metrics_del_group(neu_node_metrics_t *node_metrics,
const char * group_name)
{
neu_group_metrics_t *gm = NULL;
pthread_mutex_lock(&node_metrics->lock);
HASH_FIND_STR(node_metrics->group_metrics, group_name, gm);
if (NULL != gm) {
HASH_DEL(node_metrics->group_metrics, gm);
neu_group_metrics_free(gm);
}
pthread_mutex_unlock(&node_metrics->lock);
}
#ifdef __cplusplus
}
include/neuron/utils/rolling_counter.h
+9
View File
@@ -90,6 +90,15 @@ static inline uint64_t neu_rolling_counter_inc(neu_rolling_counter_t *counter,
return counter->val;
}
/** Reset the counter.
*/
static inline void neu_rolling_counter_reset(neu_rolling_counter_t *counter)
{
counter->val = 0;
counter->hd = 0;
memset(counter->counts, 0, counter->n * sizeof(counter->counts[0]));
}
/** Return the counter value.
*
* NOTE: may return stale value if the counter is not updated frequent enough.
src/adapter/adapter.c
+20 -129
View File
@@ -389,22 +389,19 @@ static int adapter_register_metric(neu_adapter_t *adapter, const char *name,
uint64_t init)
{
if (NULL == adapter->metrics) {
adapter->metrics = calloc(1, sizeof(*adapter->metrics));
adapter->metrics =
neu_node_metrics_new(adapter, adapter->module->type, adapter->name);
if (NULL == adapter->metrics) {
return -1;
}
adapter->metrics->type = adapter->module->type;
adapter->metrics->name = adapter->name;
adapter->metrics->adapter = adapter;
neu_metrics_add_node(adapter);
}
if (0 > neu_metric_entries_add(&adapter->metrics->entries, name, help, type,
init)) {
if (0 >
neu_node_metrics_add(adapter->metrics, NULL, name, help, type, init)) {
return -1;
}
neu_metrics_register_entry(name, help, type);
return 0;
}
@@ -412,50 +409,17 @@ static int adapter_update_metric(neu_adapter_t *adapter,
const char *metric_name, uint64_t n,
const char *group)
{
neu_metric_entry_t *entry = NULL;
if (NULL == adapter->metrics) {
return -1;
}
if (NULL == group) {
HASH_FIND_STR(adapter->metrics->entries, metric_name, entry);
} else if (NULL != adapter->metrics->group_metrics) {
neu_group_metrics_t *g = NULL;
HASH_FIND_STR(adapter->metrics->group_metrics, group, g);
if (NULL != g) {
HASH_FIND_STR(g->entries, metric_name, entry);
}
}
if (NULL == entry) {
return -1;
}
if (NEU_METRIC_TYPE_COUNTER == entry->type) {
entry->value += n;
} else if (NEU_METRIC_TYPE_ROLLING_COUNTER == entry->type) {
entry->value =
neu_rolling_counter_inc(entry->rcnt, global_timestamp, n);
} else {
entry->value = n;
}
return 0;
return neu_node_metrics_update(adapter->metrics, group, metric_name, n);
}
static void adapter_reset_metrics(neu_adapter_t *adapter)
static inline void adapter_reset_metrics(neu_adapter_t *adapter)
{
neu_metric_entry_t *entry = NULL;
if (NULL == adapter->metrics) {
return;
}
HASH_LOOP(hh, adapter->metrics->entries, entry) { entry->value = 0; }
neu_group_metrics_t *g = NULL;
HASH_LOOP(hh, adapter->metrics->group_metrics, g)
{
HASH_LOOP(hh, g->entries, entry) { entry->value = 0; }
if (NULL != adapter->metrics) {
neu_node_metrics_reset(adapter->metrics);
}
}
@@ -841,11 +805,13 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
neu_resp_get_node_state_t *resp =
(neu_resp_get_node_state_t *) &header[1];
neu_metric_entry_t *e = NULL;
if (NULL != adapter->metrics) {
pthread_mutex_lock(&adapter->metrics->lock);
neu_metric_entry_t *e = NULL;
HASH_FIND_STR(adapter->metrics->entries, NEU_METRIC_LAST_RTT_MS, e);
resp->rtt = NULL != e ? e->value : 0;
pthread_mutex_unlock(&adapter->metrics->lock);
}
resp->rtt = NULL != e ? e->value : 0;
resp->state = neu_adapter_get_state(adapter);
header->type = NEU_RESP_GET_NODE_STATE;
neu_msg_exchange(header);
@@ -1371,11 +1337,6 @@ void neu_adapter_destroy(neu_adapter_t *adapter)
if (NULL != adapter->metrics) {
neu_metrics_del_node(adapter);
neu_metric_entry_t *e = NULL;
HASH_LOOP(hh, adapter->metrics->entries, e)
{
neu_metrics_unregister_entry(e->name);
}
neu_node_metrics_free(adapter->metrics);
}
@@ -1570,113 +1531,43 @@ int neu_adapter_register_group_metric(neu_adapter_t *adapter,
const char *help, neu_metric_type_e type,
uint64_t init)
{
neu_group_metrics_t *group_metrics = NULL;
if (NULL == adapter->metrics) {
return -1;
}
if (0 > neu_metrics_register_entry(name, help, type)) {
return -1;
}
HASH_FIND_STR(adapter->metrics->group_metrics, group_name, group_metrics);
if (NULL == group_metrics) {
group_metrics = calloc(1, sizeof(*group_metrics));
if (NULL == group_metrics) {
return -1;
}
group_metrics->name = strdup(group_name);
if (NULL == group_metrics->name) {
free(group_metrics);
return -1;
}
HASH_ADD_STR(adapter->metrics->group_metrics, name, group_metrics);
}
if (0 > neu_metric_entries_add(&group_metrics->entries, name, help, type,
init)) {
return -1;
}
return 0;
return neu_node_metrics_add(adapter->metrics, group_name, name, help, type,
init);
}
int neu_adapter_update_group_metric(neu_adapter_t *adapter,
const char * group_name,
const char *metric_name, uint64_t n)
{
neu_metric_entry_t * entry = NULL;
neu_group_metrics_t *group_metrics = NULL;
if (NULL == adapter->metrics) {
return -1;
}
HASH_FIND_STR(adapter->metrics->group_metrics, group_name, group_metrics);
if (NULL == group_metrics) {
return -1;
}
HASH_FIND_STR(group_metrics->entries, metric_name, entry);
if (NULL == entry) {
return -1;
}
if (NEU_METRIC_TYPE_COUNTER == entry->type) {
entry->value += n;
} else if (NEU_METRIC_TYPE_ROLLING_COUNTER == entry->type) {
entry->value =
neu_rolling_counter_inc(entry->rcnt, global_timestamp, n);
} else {
entry->value = n;
}
return 0;
return neu_node_metrics_update(adapter->metrics, group_name, metric_name,
n);
}
int neu_adapter_metric_update_group_name(neu_adapter_t *adapter,
const char * group_name,
const char * new_group_name)
{
neu_group_metrics_t *group_metrics = NULL;
if (NULL == adapter->metrics) {
return -1;
}
HASH_FIND_STR(adapter->metrics->group_metrics, group_name, group_metrics);
if (NULL == group_metrics) {
return -1;
}
char *name = strdup(new_group_name);
if (NULL == name) {
return -1;
}
HASH_DEL(adapter->metrics->group_metrics, group_metrics);
free(group_metrics->name);
group_metrics->name = name;
HASH_ADD_STR(adapter->metrics->group_metrics, name, group_metrics);
return 0;
return neu_node_metrics_update_group(adapter->metrics, group_name,
new_group_name);
}
void neu_adapter_del_group_metrics(neu_adapter_t *adapter,
const char * group_name)
{
if (NULL == adapter->metrics) {
return;
}
neu_group_metrics_t *gm = NULL;
HASH_FIND_STR(adapter->metrics->group_metrics, group_name, gm);
if (NULL != gm) {
HASH_DEL(adapter->metrics->group_metrics, gm);
neu_metric_entry_t *e = NULL;
HASH_LOOP(hh, gm->entries, e) { neu_metrics_unregister_entry(e->name); }
neu_group_metrics_free(gm);
if (NULL != adapter->metrics) {
neu_node_metrics_del_group(adapter->metrics, group_name);
}
}