Merge pull request #508 from fengzeroz/main

async queue & fix bug

CMakeLists.txt

@@ -162,7 +162,8 @@ set(NEURON_BASE_SOURCES
     src/utils/json.c
     src/utils/neu_jwt.c
     src/utils/file.c
-    src/utils/base64.c)
+    src/utils/base64.c
+    src/utils/async_queue.c)
 
 if(USE_NNG_MQTT)
   message("-- USE_NNG_MQTT is defined")
@@ -189,7 +190,7 @@ if(USE_NNG_MQTT)
   target_link_libraries(neuron-base mbedtls mbedx509 mbedcrypto nng)
 endif()
 
-target_link_libraries(neuron-base libzlog.so vector-static csptr jansson crypto ssl mqttc jwt
+target_link_libraries(neuron-base nng libzlog.so vector-static csptr jansson crypto ssl mqttc jwt
                       ${CMAKE_THREAD_LIBS_INIT})
 
 #copy file for run

include/neuron/type.h

@@ -348,10 +348,14 @@ static inline uint64_t neu_htonll(uint64_t u)
     uint64_t ret       = 0;
     uint8_t *ret_bytes = (uint8_t *) &ret;
 
-    ret_bytes[0] = bytes[3];
-    ret_bytes[1] = bytes[2];
-    ret_bytes[2] = bytes[1];
-    ret_bytes[3] = bytes[0];
+    ret_bytes[0] = bytes[7];
+    ret_bytes[1] = bytes[6];
+    ret_bytes[2] = bytes[5];
+    ret_bytes[3] = bytes[4];
+    ret_bytes[4] = bytes[3];
+    ret_bytes[5] = bytes[2];
+    ret_bytes[6] = bytes[1];
+    ret_bytes[7] = bytes[0];
 
     return ret;
 }

include/neuron/utils/async_queue.h

@@ -0,0 +1,45 @@
+/**
+ * NEURON IIoT System for Industry 4.0
+ * Copyright (C) 2020-2021 EMQ Technologies Co., Ltd All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+config_ **/
+
+#ifndef _NEU_ASYNC_QUEUE_H_
+#define _NEU_ASYNC_QUEUE_H_
+
+#include <stdbool.h>
+#include <stdint.h>
+
+typedef struct neu_async_queue neu_async_queue_t;
+
+typedef uint64_t (*neu_async_queue_key)(void *elem);
+typedef bool (*neu_async_queue_expire)(void *elem);
+typedef void (*neu_async_queue_free)(void *elem);
+typedef bool (*neu_async_queue_filter)(void *filter_elem, void *elem);
+
+neu_async_queue_t *neu_async_queue_new(neu_async_queue_key    key_fn,
+                                       neu_async_queue_expire expire_fn,
+                                       neu_async_queue_free   free_fn,
+                                       uint16_t               max_size);
+
+void neu_async_queue_destroy(neu_async_queue_t *q);
+void neu_async_queue_push(neu_async_queue_t *q, void *elem);
+int  neu_async_queue_pop(neu_async_queue_t *q, uint64_t key, void **elem);
+
+void neu_async_queue_remove(neu_async_queue_t *q, neu_async_queue_filter filter,
+                            void *filter_elem);
+
+#endif

include/neuron/utils/utextend.h

@@ -39,8 +39,19 @@ static inline UT_array *utarray_clone(UT_array *array)
     return result;
 }
 
-#define utarray_foreach(array, type, elem)               \
-    for (type elem = utarray_front(array); elem != NULL; \
-         elem      = utarray_next(array, elem))
+#define utarray_foreach(array, type, elem)                      \
+    for (type elem = (type) utarray_front(array); elem != NULL; \
+         elem      = (type) utarray_next(array, elem))
+
+#define DL_LAST(head) ((head) != NULL) ? (head)->prev : NULL;
+#define DL_PREV(el) ((el) != NULL) ? (el)->prev : NULL;
+
+#define DL_REVERSE_FOREACH(head, tail, el) \
+    (tail) = DL_LAST(head);                \
+    (el)   = (tail);                       \
+    do
+
+#define DL_REVERSE_FOREACH_END(head, tail, el) \
+    while (((el) = (el)->prev) != (head))
 
 #endif

src/adapter/driver/driver.c

@@ -118,6 +118,7 @@ static void update(neu_adapter_t *adapter, const char *name, neu_dvalue_t value)
     case NEU_TYPE_BYTES:
         break;
     default:
+        assert(value.type == NEU_TYPE_INT8);
         break;
     }
 

src/utils/async_queue.c

@@ -0,0 +1,163 @@
+/**
+ * NEURON IIoT System for Industry 4.0
+ * Copyright (C) 2020-2021 EMQ Technologies Co., Ltd All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+config_ **/
+
+#include <stdlib.h>
+
+#include <nng/nng.h>
+#include <nng/supplemental/util/platform.h>
+
+#include "utils/async_queue.h"
+#include "utils/utextend.h"
+
+typedef struct element {
+    uint32_t key;
+
+    void *data;
+
+    struct element *prev;
+    struct element *next;
+} element, element_list;
+
+struct neu_async_queue {
+    neu_async_queue_key    key_fn;
+    neu_async_queue_expire expire_fn;
+    neu_async_queue_free   free_fn;
+
+    nng_mtx *mtx;
+
+    uint16_t max;
+
+    element_list *list;
+};
+
+neu_async_queue_t *neu_async_queue_new(neu_async_queue_key    key_fn,
+                                       neu_async_queue_expire expire_fn,
+                                       neu_async_queue_free   free_fn,
+                                       uint16_t               max_size)
+{
+    neu_async_queue_t *q = calloc(1, sizeof(neu_async_queue_t));
+
+    q->key_fn    = key_fn;
+    q->expire_fn = expire_fn;
+    q->free_fn   = free_fn;
+
+    q->max = max_size;
+
+    nng_mtx_alloc(&q->mtx);
+
+    return q;
+}
+
+void neu_async_queue_destroy(neu_async_queue_t *q)
+{
+    element *elt = NULL, *tmp = NULL;
+
+    nng_mtx_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);
+
+    nng_mtx_free(q->mtx);
+    free(q);
+}
+
+void neu_async_queue_push(neu_async_queue_t *q, void *elem)
+{
+    element *elt   = NULL;
+    int      count = 0;
+
+    nng_mtx_lock(q->mtx);
+    DL_COUNT(q->list, elt, count);
+
+    if (count == q->max && count > 0) {
+        elt = q->list;
+        DL_DELETE(q->list, elt);
+        q->free_fn(elt->data);
+        free(elt);
+    }
+
+    elt       = calloc(1, sizeof(element));
+    elt->data = elem;
+    elt->key  = q->key_fn(elem);
+
+    DL_APPEND(q->list, elt);
+
+    nng_mtx_unlock(q->mtx);
+}
+
+int neu_async_queue_pop(neu_async_queue_t *q, uint64_t key, void **elem)
+{
+    element *elt = NULL, *first = NULL;
+    int      ret = -1;
+
+    nng_mtx_lock(q->mtx);
+    first = q->list;
+    elt   = DL_LAST(q->list);
+
+    while (elt != NULL) {
+        if (elt->key == key) {
+            DL_DELETE(q->list, elt);
+            *elem = elt->data;
+            free(elt);
+            ret = 0;
+            break;
+        }
+
+        if (first == elt) {
+            break;
+        }
+
+        if (q->expire_fn(elt->data)) {
+            element *tmp = DL_PREV(elt);
+
+            DL_DELETE(q->list, elt);
+            q->free_fn(elt->data);
+            free(elt);
+
+            elt = tmp;
+        } else {
+            elt = DL_PREV(elt);
+        }
+    }
+
+    nng_mtx_unlock(q->mtx);
+
+    return ret;
+}
+
+void neu_async_queue_remove(neu_async_queue_t *q, neu_async_queue_filter filter,
+                            void *filter_elem)
+{
+    element *el = NULL, *tmp = NULL;
+
+    nng_mtx_lock(q->mtx);
+    DL_FOREACH_SAFE(q->list, el, tmp)
+    {
+        if (filter(filter_elem, el->data)) {
+            DL_DELETE(q->list, el);
+            q->free_fn(el->data);
+            free(el);
+        }
+    }
+    nng_mtx_unlock(q->mtx);
+}