Merge pull request #897 from fengzeroz/main

connection: statics of bytes sent and recvied

include/neuron/connection/neu_connection.h

@@ -111,6 +111,11 @@ typedef struct neu_conn_param {
 
 typedef struct neu_conn neu_conn_t;
 
+typedef struct neu_conn_state {
+    uint64_t send_bytes;
+    uint64_t recv_bytes;
+} neu_conn_state_t;
+
 /**
  * @brief create a new connection.
  *
@@ -166,6 +171,14 @@ void neu_conn_stop(neu_conn_t *conn);
  */
 void neu_conn_start(neu_conn_t *conn);
 
+/**
+ * @brief get state of connection
+ *
+ *
+ * @param[in] conn
+ * @return returns state.
+ */
+neu_conn_state_t neu_conn_state(neu_conn_t *conn);
 /**
  * @brief Receive connection requests from clients.
  *

plugins/modbus/modbus_req.c

@@ -50,15 +50,8 @@ int modbus_send_msg(void *ctx, uint16_t n_byte, uint8_t *bytes)
 {
     neu_plugin_t *plugin = (neu_plugin_t *) ctx;
 
-    neu_adapter_update_metric_cb_t update_metric =
-        plugin->common.adapter_callbacks->update_metric;
-
     plog_send_protocol(plugin, bytes, n_byte);
-    int ret = neu_conn_send(plugin->conn, bytes, n_byte);
-    if (ret > 0) {
-        update_metric(plugin->common.adapter, NEU_METRIC_SEND_BYTES, ret, NULL);
-    }
-    return ret;
+    return neu_conn_send(plugin->conn, bytes, n_byte);
 }
 
 int modbus_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group,
@@ -66,9 +59,9 @@ int modbus_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group,
 {
     neu_adapter_update_metric_cb_t update_metric =
         plugin->common.adapter_callbacks->update_metric;
-
-    struct modbus_group_data *gd  = NULL;
-    int64_t                   rtt = NEU_METRIC_LAST_RTT_MS_MAX;
+    struct modbus_group_data *gd    = NULL;
+    neu_conn_state_t          state = { 0 };
+    int64_t                   rtt   = NEU_METRIC_LAST_RTT_MS_MAX;
 
     if (group->user_data == NULL) {
         gd = calloc(1, sizeof(struct modbus_group_data));
@@ -112,6 +105,12 @@ int modbus_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group,
         }
     }
 
+    state = neu_conn_state(plugin->conn);
+
+    update_metric(plugin->common.adapter, NEU_METRIC_SEND_BYTES,
+                  state.send_bytes, NULL);
+    update_metric(plugin->common.adapter, NEU_METRIC_RECV_BYTES,
+                  state.recv_bytes, NULL);
     update_metric(plugin->common.adapter, NEU_METRIC_LAST_RTT_MS, rtt, NULL);
     update_metric(plugin->common.adapter, NEU_METRIC_GROUP_LAST_SEND_MSGS,
                   gd->cmd_sort->n_cmd, group->group_name);
@@ -319,13 +318,9 @@ static int process_protocol_buf(neu_plugin_t *plugin, uint16_t response_size)
     neu_protocol_unpack_buf_t pbuf     = { 0 };
     ssize_t                   ret      = 0;
 
-    neu_adapter_update_metric_cb_t update_metric =
-        plugin->common.adapter_callbacks->update_metric;
-
     ret = neu_conn_recv(plugin->conn, recv_buf, response_size);
 
     if (ret > 0) {
-        update_metric(plugin->common.adapter, NEU_METRIC_RECV_BYTES, ret, NULL);
         neu_protocol_unpack_buf_init(&pbuf, recv_buf, ret);
         plog_recv_protocol(plugin, recv_buf, ret);
         ret = modbus_stack_recv(plugin->stack, &pbuf);

src/connection/connection.c

@@ -53,6 +53,8 @@ struct neu_conn {
     int  fd;
     bool block;
 
+    neu_conn_state_t state;
+
     struct {
         struct tcp_client *clients;
         int                n_client;
@@ -120,7 +122,6 @@ void neu_conn_start(neu_conn_t *conn)
 neu_conn_t *neu_conn_reconfig(neu_conn_t *conn, neu_conn_param_t *param)
 {
     pthread_mutex_lock(&conn->mtx);
-    (void) param;
 
     conn_disconnect(conn);
     conn_free_param(conn);
@@ -129,6 +130,9 @@ neu_conn_t *neu_conn_reconfig(neu_conn_t *conn, neu_conn_param_t *param)
     conn_init_param(conn, param);
     conn_tcp_server_listen(conn);
 
+    conn->state.recv_bytes = 0;
+    conn->state.send_bytes = 0;
+
     pthread_mutex_unlock(&conn->mtx);
 
     return conn;
@@ -150,6 +154,11 @@ void neu_conn_destory(neu_conn_t *conn)
     free(conn);
 }
 
+neu_conn_state_t neu_conn_state(neu_conn_t *conn)
+{
+    return conn->state;
+}
+
 int neu_conn_tcp_server_accept(neu_conn_t *conn)
 {
     struct sockaddr_in client     = { 0 };
@@ -234,6 +243,9 @@ ssize_t neu_conn_tcp_server_send(neu_conn_t *conn, int fd, uint8_t *buf,
 
     conn_tcp_server_listen(conn);
     ret = send(fd, buf, len, MSG_NOSIGNAL | MSG_DONTWAIT);
+    if (ret > 0) {
+        conn->state.send_bytes += ret;
+    }
 
     pthread_mutex_unlock(&conn->mtx);
 
@@ -257,6 +269,10 @@ ssize_t neu_conn_tcp_server_recv(neu_conn_t *conn, int fd, uint8_t *buf,
         ret = recv(fd, buf, len, 0);
     }
 
+    if (ret > 0) {
+        conn->state.recv_bytes += ret;
+    }
+
     pthread_mutex_unlock(&conn->mtx);
 
     return ret;
@@ -310,6 +326,10 @@ ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len)
         }
     }
 
+    if (ret > 0) {
+        conn->state.send_bytes += ret;
+    }
+
     pthread_mutex_unlock(&conn->mtx);
 
     return ret;
@@ -370,6 +390,10 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
 
     pthread_mutex_unlock(&conn->mtx);
 
+    if (ret > 0) {
+        conn->state.recv_bytes += ret;
+    }
+
     return ret;
 }