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

msg q

Browse Source
This commit is contained in:
fengzero
2023-12-29 02:02:48 +00:00
parent 5737511f22
commit f1e4bed876
5 changed files with 204 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CMakeLists.txt
+1
View File
@@ -128,6 +128,7 @@ set(NEURON_SOURCES
src/core/node_manager.c
src/core/template_manager.c
src/core/storage.c
src/adapter/msg_q.c
src/adapter/storage.c
src/adapter/adapter.c
src/adapter/driver/cache.c
src/adapter/adapter.c
+47 -8
View File
@@ -39,11 +39,12 @@
#include "plugin.h"
#include "storage.h"
static int adapter_trans_data(enum neu_event_io_type type, int fd,
void *usr_data);
static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data);
static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
void *data);
static void *adapter_consumer(void *arg);
static int adapter_trans_data(enum neu_event_io_type type, int fd,
void *usr_data);
static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data);
static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
void *data);
static int adapter_response(neu_adapter_t *adapter, neu_reqresp_head_t *header,
void *data);
static int adapter_responseto(neu_adapter_t * adapter,
@@ -110,6 +111,27 @@ void neu_adapter_set_error(int error)
create_adapter_error = error;
}
static void *adapter_consumer(void *arg)
{
neu_adapter_t *adapter = (neu_adapter_t *) arg;
while (1) {
neu_msg_t * msg = NULL;
uint32_t n = adapter_msg_q_pop(adapter->msg_q, &msg);
neu_reqresp_head_t *header = neu_msg_get_header(msg);
nlog_debug("adapter(%s) recv msg from: %s %p, type: %s, %u",
adapter->name, header->sender, header->ctx,
neu_reqresp_type_string(header->type), n);
adapter->module->intf_funs->request(
adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
neu_msg_free(msg);
}
return NULL;
}
static inline zlog_category_t *get_log_category(const char *node)
{
char name[NEU_NODE_NAME_LEN] = { 0 };
@@ -181,6 +203,9 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load)
neu_adapter_driver_init((neu_adapter_driver_t *) adapter);
break;
case NEU_NA_TYPE_APP: {
adapter->msg_q = adapter_msg_q_new(adapter->name, 1024);
pthread_create(&adapter->consumer_tid, NULL, adapter_consumer,
(void *) adapter);
while (true) {
uint16_t port = neu_manager_get_port();
struct sockaddr_in local = {
@@ -596,11 +621,18 @@ static int adapter_trans_data(enum neu_event_io_type type, int fd,
return 0;
}
if (header->type == NEU_REQRESP_TRANS_DATA) {
if (adapter_msg_q_push(adapter->msg_q, msg) < 0) {
nlog_warn("adapter: %s trans data msg q is full, drop msg",
adapter->name);
neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
neu_msg_free(msg);
}
return 0;
}
adapter->module->intf_funs->request(
adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
if (header->type == NEU_REQRESP_TRANS_DATA) {
neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
}
neu_msg_free(msg);
return 0;
}
@@ -1256,6 +1288,13 @@ void neu_adapter_destroy(neu_adapter_t *adapter)
neu_node_metrics_free(adapter->metrics);
}
if (adapter->consumer_tid != 0) {
pthread_cancel(adapter->consumer_tid);
}
if (adapter->msg_q != NULL) {
adapter_msg_q_free(adapter->msg_q);
}
char *setting = NULL;
if (adapter_load_setting(adapter->name, &setting) != 0) {
remove_logs(adapter->name);
src/adapter/adapter_internal.h
+4
View File
@@ -26,6 +26,7 @@
#include "adapter_info.h"
#include "core/manager.h"
#include "msg_q.h"
struct neu_adapter {
char *name;
@@ -45,6 +46,9 @@ struct neu_adapter {
int control_fd;
int trans_data_fd;
adapter_msg_q_t *msg_q;
pthread_t consumer_tid;
uint16_t trans_data_port;
neu_events_t *events;
src/adapter/msg_q.c
+116
View File
@@ -0,0 +1,116 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2024 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.
**/
#include <pthread.h>
#include "utils/log.h"
#include "utils/utextend.h"
#include "msg_q.h"
struct item {
neu_msg_t * msg;
struct item *prev, *next;
};
struct adapter_msg_q {
struct item *list;
uint32_t max;
uint32_t current;
char * name;
pthread_mutex_t mtx;
pthread_cond_t cond;
};
adapter_msg_q_t *adapter_msg_q_new(const char *name, uint32_t size)
{
struct adapter_msg_q *q = calloc(1, sizeof(struct adapter_msg_q));
pthread_mutex_init(&q->mtx, NULL);
pthread_cond_init(&q->cond, NULL);
q->max = size;
q->name = strdup(name);
q->current = 0;
return q;
}
void adapter_msg_q_free(adapter_msg_q_t *q)
{
struct item *tmp = NULL, *elt = NULL;
nlog_warn("app: %s, drop %u msg", q->name, q->current);
pthread_mutex_destroy(&q->mtx);
pthread_cond_destroy(&q->cond);
DL_FOREACH_SAFE(q->list, elt, tmp)
{
DL_DELETE(q->list, elt);
neu_reqresp_head_t *header = neu_msg_get_header(elt->msg);
neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
neu_msg_free(elt->msg);
free(elt);
}
free(q->name);
free(q);
}
int adapter_msg_q_push(adapter_msg_q_t *q, neu_msg_t *msg)
{
int ret = -1;
pthread_mutex_lock(&q->mtx);
if (q->current < q->max) {
q->current += 1;
struct item *elt = calloc(1, sizeof(struct item));
elt->msg = msg;
DL_APPEND(q->list, elt);
ret = 0;
}
pthread_mutex_unlock(&q->mtx);
if (ret == -1) {
nlog_warn("app: %s, msg q is full, %u(%u)", q->name, q->current,
q->max);
} else {
pthread_cond_signal(&q->cond);
}
return ret;
}
uint32_t adapter_msg_q_pop(adapter_msg_q_t *q, neu_msg_t **p_data)
{
uint32_t ret = 0;
pthread_mutex_lock(&q->mtx);
while (q->current == 0) {
pthread_cond_wait(&q->cond, &q->mtx);
}
struct item *elt = DL_LAST(q->list);
if (elt != NULL) {
DL_DELETE(q->list, elt);
*p_data = elt->msg;
free(elt);
q->current -= 1;
ret = q->current;
}
pthread_mutex_unlock(&q->mtx);
return ret;
}
src/adapter/msg_q.h
+36
View File
@@ -0,0 +1,36 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2024 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.
**/
#ifndef ADAPTER_MSG_Q_H
#define ADAPTER_MSG_Q_H
#include <stdint.h>
#include "base/msg_internal.h"
#include "msg.h"
typedef struct adapter_msg_q adapter_msg_q_t;
adapter_msg_q_t *adapter_msg_q_new(const char *name, uint32_t size);
void adapter_msg_q_free(adapter_msg_q_t *q);
int adapter_msg_q_push(adapter_msg_q_t *q, neu_msg_t *msg);
uint32_t adapter_msg_q_pop(adapter_msg_q_t *q, neu_msg_t **p_data);
#endif