diff --git a/src/adapter/driver/cache.c b/src/adapter/driver/cache.c index 171fd831..fa4fe395 100644 --- a/src/adapter/driver/cache.c +++ b/src/adapter/driver/cache.c @@ -19,11 +19,9 @@ #include #include +#include #include -#include -#include - #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); } diff --git a/src/event/event_linux.c b/src/event/event_linux.c index 0a1959c0..70764bf8 100644 --- a/src/event/event_linux.c +++ b/src/event/event_linux.c @@ -26,9 +26,6 @@ #include #include -#include -#include - #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; } diff --git a/src/utils/async_queue.c b/src/utils/async_queue.c index 5c06b4b7..52e226c8 100644 --- a/src/utils/async_queue.c +++ b/src/utils/async_queue.c @@ -17,11 +17,9 @@ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. config_ **/ +#include #include -#include -#include - #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); } \ No newline at end of file