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

chore: replace nng_mtx with pthread_mtx

Browse Source
This commit is contained in:
fengzero
2023-10-10 02:36:51 +00:00
parent 15fcbe2ae3
commit 0df2a94b3f
3 changed files with 41 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/adapter/driver/cache.c
+20 -22
View File
@@ -19,11 +19,9 @@
#include <assert.h>
#include <math.h>
#include <pthread.h>
#include <string.h>
#include <nng/nng.h>
#include <nng/supplemental/util/platform.h>
#include "utils/uthash.h"
#include "define.h"
@@ -49,7 +47,7 @@ struct elem {
};
struct neu_driver_cache {
nng_mtx *mtx;
pthread_mutex_t mtx;
struct elem *table;
};
@@ -71,7 +69,7 @@ neu_driver_cache_t *neu_driver_cache_new()
{
neu_driver_cache_t *cache = calloc(1, sizeof(neu_driver_cache_t));
nng_mtx_alloc(&cache->mtx);
pthread_mutex_init(&cache->mtx, NULL);
return cache;
}
@@ -81,7 +79,7 @@ void neu_driver_cache_destroy(neu_driver_cache_t *cache)
struct elem *elem = NULL;
struct elem *tmp = NULL;
nng_mtx_lock(cache->mtx);
pthread_mutex_lock(&cache->mtx);
HASH_ITER(hh, cache->table, elem, tmp)
{
HASH_DEL(cache->table, elem);
@@ -93,9 +91,9 @@ void neu_driver_cache_destroy(neu_driver_cache_t *cache)
}
free(elem);
}
nng_mtx_unlock(cache->mtx);
pthread_mutex_unlock(&cache->mtx);
nng_mtx_free(cache->mtx);
pthread_mutex_destroy(&cache->mtx);
free(cache);
}
@@ -112,7 +110,7 @@ void neu_driver_cache_add(neu_driver_cache_t *cache, const char *group,
struct elem *elem = NULL;
tkey_t key = to_key(group, tag);
nng_mtx_lock(cache->mtx);
pthread_mutex_lock(&cache->mtx);
HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem);
if (elem == NULL) {
@@ -128,7 +126,7 @@ void neu_driver_cache_add(neu_driver_cache_t *cache, const char *group,
elem->changed = false;
elem->value = value;
nng_mtx_unlock(cache->mtx);
pthread_mutex_unlock(&cache->mtx);
}
void neu_driver_cache_update_change(neu_driver_cache_t *cache,
@@ -140,7 +138,7 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
struct elem *elem = NULL;
tkey_t key = to_key(group, tag);
nng_mtx_lock(cache->mtx);
pthread_mutex_lock(&cache->mtx);
HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem);
if (elem != NULL) {
elem->timestamp = timestamp;
@@ -242,7 +240,7 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
}
}
nng_mtx_unlock(cache->mtx);
pthread_mutex_unlock(&cache->mtx);
}
void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group,
@@ -262,7 +260,7 @@ int neu_driver_cache_meta_get(neu_driver_cache_t *cache, const char *group,
int ret = -1;
tkey_t key = to_key(group, tag);
nng_mtx_lock(cache->mtx);
pthread_mutex_lock(&cache->mtx);
HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem);
if (elem != NULL) {
@@ -330,7 +328,7 @@ int neu_driver_cache_meta_get(neu_driver_cache_t *cache, const char *group,
ret = 0;
}
nng_mtx_unlock(cache->mtx);
pthread_mutex_unlock(&cache->mtx);
return ret;
}
@@ -344,7 +342,7 @@ int neu_driver_cache_meta_get_changed(neu_driver_cache_t *cache,
int ret = -1;
tkey_t key = to_key(group, tag);
nng_mtx_lock(cache->mtx);
pthread_mutex_lock(&cache->mtx);
HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem);
if (elem != NULL && elem->changed) {
@@ -415,7 +413,7 @@ int neu_driver_cache_meta_get_changed(neu_driver_cache_t *cache,
ret = 0;
}
nng_mtx_unlock(cache->mtx);
pthread_mutex_unlock(&cache->mtx);
return ret;
}
@@ -427,7 +425,7 @@ int neu_driver_cache_get(neu_driver_cache_t *cache, const char *group,
int ret = -1;
tkey_t key = to_key(group, tag);
nng_mtx_lock(cache->mtx);
pthread_mutex_lock(&cache->mtx);
HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem);
if (elem != NULL) {
@@ -492,7 +490,7 @@ int neu_driver_cache_get(neu_driver_cache_t *cache, const char *group,
ret = 0;
}
nng_mtx_unlock(cache->mtx);
pthread_mutex_unlock(&cache->mtx);
return ret;
}
@@ -505,7 +503,7 @@ int neu_driver_cache_get_changed(neu_driver_cache_t *cache, const char *group,
int ret = -1;
tkey_t key = to_key(group, tag);
nng_mtx_lock(cache->mtx);
pthread_mutex_lock(&cache->mtx);
HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem);
if (elem != NULL && elem->changed) {
@@ -572,7 +570,7 @@ int neu_driver_cache_get_changed(neu_driver_cache_t *cache, const char *group,
ret = 0;
}
nng_mtx_unlock(cache->mtx);
pthread_mutex_unlock(&cache->mtx);
return ret;
}
@@ -583,7 +581,7 @@ void neu_driver_cache_del(neu_driver_cache_t *cache, const char *group,
struct elem *elem = NULL;
tkey_t key = to_key(group, tag);
nng_mtx_lock(cache->mtx);
pthread_mutex_lock(&cache->mtx);
HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem);
if (elem != NULL) {
@@ -597,5 +595,5 @@ void neu_driver_cache_del(neu_driver_cache_t *cache, const char *group,
free(elem);
}
nng_mtx_unlock(cache->mtx);
pthread_mutex_unlock(&cache->mtx);
}
src/event/event_linux.c
+7 -10
View File
@@ -26,9 +26,6 @@
#include <string.h>
#include <unistd.h>
#include <nng/nng.h>
#include <nng/supplemental/util/platform.h>
#include "event/event.h"
#include "utils/log.h"
@@ -41,7 +38,7 @@ struct neu_event_timer {
struct event_data * event_data;
struct itimerspec value;
neu_event_timer_type_e type;
nng_mtx * mtx;
pthread_mutex_t mtx;
bool stop;
};
@@ -132,7 +129,7 @@ static void *event_loop(void *arg)
switch (data->type) {
case TIMER:
nng_mtx_lock(data->ctx.timer.mtx);
pthread_mutex_lock(&data->ctx.timer.mtx);
if ((event.events & EPOLLIN) == EPOLLIN) {
uint64_t t;
@@ -152,7 +149,7 @@ static void *event_loop(void *arg)
}
}
nng_mtx_unlock(data->ctx.timer.mtx);
pthread_mutex_unlock(&data->ctx.timer.mtx);
break;
case IO:
if ((event.events & EPOLLHUP) == EPOLLHUP) {
@@ -238,7 +235,7 @@ neu_event_timer_t *neu_event_add_timer(neu_events_t * events,
timer_ctx->fd = timer_fd;
timer_ctx->type = timer.type;
timer_ctx->stop = false;
nng_mtx_alloc(&timer_ctx->mtx);
pthread_mutex_init(&timer_ctx->mtx, NULL);
ret = epoll_ctl(events->epoll_fd, EPOLL_CTL_ADD, timer_fd, &event);
@@ -260,11 +257,11 @@ int neu_event_del_timer(neu_events_t *events, neu_event_timer_t *timer)
timer->stop = true;
epoll_ctl(events->epoll_fd, EPOLL_CTL_DEL, timer->fd, NULL);
nng_mtx_lock(timer->mtx);
pthread_mutex_lock(&timer->mtx);
close(timer->fd);
nng_mtx_unlock(timer->mtx);
pthread_mutex_unlock(&timer->mtx);
nng_mtx_free(timer->mtx);
pthread_mutex_destroy(&timer->mtx);
free_event(events, timer->event_data->index);
return 0;
}
src/utils/async_queue.c
+14 -16
View File
@@ -17,11 +17,9 @@
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
config_ **/
#include <pthread.h>
#include <stdlib.h>
#include <nng/nng.h>
#include <nng/supplemental/util/platform.h>
#include "utils/async_queue.h"
#include "utils/utextend.h"
@@ -39,7 +37,7 @@ struct neu_async_queue {
neu_async_queue_expire expire_fn;
neu_async_queue_free free_fn;
nng_mtx *mtx;
pthread_mutex_t mtx;
uint16_t max;
@@ -59,7 +57,7 @@ neu_async_queue_t *neu_async_queue_new(neu_async_queue_key key_fn,
q->max = max_size;
nng_mtx_alloc(&q->mtx);
pthread_mutex_init(&q->mtx, NULL);
return q;
}
@@ -68,16 +66,16 @@ void neu_async_queue_destroy(neu_async_queue_t *q)
{
element *elt = NULL, *tmp = NULL;
nng_mtx_lock(q->mtx);
pthread_mutex_lock(&q->mtx);
DL_FOREACH_SAFE(q->list, elt, tmp)
{
q->free_fn(elt->data);
DL_DELETE(q->list, elt);
free(elt);
}
nng_mtx_unlock(q->mtx);
pthread_mutex_unlock(&q->mtx);
nng_mtx_free(q->mtx);
pthread_mutex_destroy(&q->mtx);
free(q);
}
@@ -86,7 +84,7 @@ void neu_async_queue_push(neu_async_queue_t *q, void *elem)
element *elt = NULL;
int count = 0;
nng_mtx_lock(q->mtx);
pthread_mutex_lock(&q->mtx);
DL_COUNT(q->list, elt, count);
if (count == q->max && count > 0) {
@@ -102,7 +100,7 @@ void neu_async_queue_push(neu_async_queue_t *q, void *elem)
DL_APPEND(q->list, elt);
nng_mtx_unlock(q->mtx);
pthread_mutex_unlock(&q->mtx);
}
int neu_async_queue_pop(neu_async_queue_t *q, uint64_t key, void **elem)
@@ -110,7 +108,7 @@ int neu_async_queue_pop(neu_async_queue_t *q, uint64_t key, void **elem)
element *elt = NULL, *tmp = NULL;
int ret = -1;
nng_mtx_lock(q->mtx);
pthread_mutex_lock(&q->mtx);
DL_FOREACH_SAFE(q->list, elt, tmp)
{
if (elt->key == key) {
@@ -127,7 +125,7 @@ int neu_async_queue_pop(neu_async_queue_t *q, uint64_t key, void **elem)
}
}
}
nng_mtx_unlock(q->mtx);
pthread_mutex_unlock(&q->mtx);
return ret;
}
@@ -137,7 +135,7 @@ void neu_async_queue_remove(neu_async_queue_t *q, neu_async_queue_filter filter,
{
element *el = NULL, *tmp = NULL;
nng_mtx_lock(q->mtx);
pthread_mutex_lock(&q->mtx);
DL_FOREACH_SAFE(q->list, el, tmp)
{
if (filter(filter_elem, el->data)) {
@@ -146,19 +144,19 @@ void neu_async_queue_remove(neu_async_queue_t *q, neu_async_queue_filter filter,
free(el);
}
}
nng_mtx_unlock(q->mtx);
pthread_mutex_unlock(&q->mtx);
}
void neu_async_queue_clean(neu_async_queue_t *q)
{
element *el = NULL, *tmp = NULL;
nng_mtx_lock(q->mtx);
pthread_mutex_lock(&q->mtx);
DL_FOREACH_SAFE(q->list, el, tmp)
{
DL_DELETE(q->list, el);
q->free_fn(el->data);
free(el);
}
nng_mtx_unlock(q->mtx);
pthread_mutex_unlock(&q->mtx);
}