1

a8/asiotcpclient.cc

@@ -60,6 +60,7 @@ namespace a8
 
     void AsioTcpClient::SendBuff(const char* buff, unsigned int bufflen)
     {
+        //a8::XPrintf("SendBuff bufflen:%d\n", {bufflen});
         if (bufflen > 0) {
             a8::SendQueueNode* p = (a8::SendQueueNode*)malloc(sizeof(a8::SendQueueNode));
             memset(p, 0, sizeof(SendQueueNode));
@@ -140,10 +141,11 @@ namespace a8
                      }
                      DoRead();
                  } else {
+                     a8::XPrintf("DoRead error %s\n", {ec.message()});
                      actived_ = false;
                      connected_ = false;
                      if (on_disconnect) {
-                         on_disconnect(this);
+                         on_disconnect(this, ec.value());
                      }
                  }
              });
@@ -158,7 +160,8 @@ namespace a8
             bot_node_ = nullptr;
             send_buffer_mutex_->unlock();
         }
-        if (work_node_) {
+        if (work_node_ && !sending_) {
+            sending_ = true;
             char* buf = work_node_->buff + work_node_->sent_bytes;
             int buf_len = work_node_->bufflen - work_node_->sent_bytes;
             asio::async_write
@@ -167,6 +170,8 @@ namespace a8
                  [this] (const asio::error_code& ec, std::size_t bytes_transferred)
                  {
                      if (!ec) {
+                         send_buffer_mutex_->lock();
+                         if (work_node_) {
                              work_node_->sent_bytes += bytes_transferred;
                              if (work_node_->sent_bytes >= work_node_->bufflen) {
                                  auto pdelnode = work_node_;
@@ -174,9 +179,21 @@ namespace a8
                                  free(pdelnode->buff);
                                  free((void*)pdelnode);
                              }
+                             if (!work_node_) {
+                                 sending_ = false;
+                             }
+                             send_buffer_mutex_->unlock();
                              DoSend();
+                             return;
+                         }
+                         send_buffer_mutex_->unlock();
                      } else {
-                         abort();
+                         a8::XPrintf("DoSend error %s\n", {ec.message()});
+                         actived_ = false;
+                         connected_ = false;
+                         if (on_disconnect) {
+                             on_disconnect(this, ec.value());
+                         }
                      }
                  });
 

a8/asiotcpclient.h

@@ -14,7 +14,7 @@ namespace a8
     public:
         std::function<void (a8::AsioTcpClient*, int)> on_error;
         std::function<void (a8::AsioTcpClient*)> on_connect;
-        std::function<void (a8::AsioTcpClient*)> on_disconnect;
+        std::function<void (a8::AsioTcpClient*, int)> on_disconnect;
         std::function<void (a8::AsioTcpClient*, char*, unsigned int)> on_socketread;
         AsioTcpClient(std::shared_ptr<asio::io_context> io_context,
                       const std::string& remote_ip,
@@ -47,6 +47,7 @@ namespace a8
         SendQueueNode *top_node_ = nullptr;
         SendQueueNode *bot_node_ = nullptr;
         volatile SendQueueNode *work_node_ = nullptr;
+        volatile bool sending_ = false;
         std::array<char, 1024 * 64> buffer_;
 
         void SetActive(bool active);

a8/perfmonitor.h

@@ -1,5 +1,7 @@
 #pragma once
 
+#include <atomic>
+
 #include <a8/singleton.h>
 
 namespace a8
@@ -20,6 +22,10 @@ namespace a8
         long long recv_eagain_times = 0;
         long long max_send_time = 0;
         long long max_recv_time = 0;
+        std::atomic<long long> server_send_bytes = {0};
+        std::atomic<long long> server_consume_bytes = {0};
+        std::atomic<long long> conn_send_bytes = {0};
+        std::atomic<long long> conn_consume_bytes = {0};
 
     };
 

a8/tcpclient.cc

@@ -13,6 +13,7 @@
 
 #include <a8/a8.h>
 #include <a8/tcpclient.h>
+#include <a8/perfmonitor.h>
 
 const int MAX_RECV_BUFFERSIZE = 1024 * 64;
 
@@ -65,6 +66,9 @@ namespace a8
     void TcpClient::SendBuff(const char* buff, unsigned int bufflen)
     {
         if (bufflen > 0) {
+#ifdef A8_PERFT
+            PerfMonitor::Instance()->conn_send_bytes += bufflen;
+#endif
             a8::SendQueueNode* p = (a8::SendQueueNode*)malloc(sizeof(a8::SendQueueNode));
             memset(p, 0, sizeof(SendQueueNode));
             p->buff = (char*)malloc(bufflen);
@@ -217,6 +221,9 @@ namespace a8
                                      currnode->buff + currnode->sent_bytes,
                                      currnode->bufflen - currnode->sent_bytes,
                                      0);
+#ifdef A8_PERFT
+                    PerfMonitor::Instance()->conn_consume_bytes += len;
+#endif
                     if (len > 0) {
                         currnode->sent_bytes += len;
                     } else {

a8/tcpsession.cc

@@ -68,6 +68,9 @@ namespace a8
             return;
         }
         if (bufflen > 0) {
+#ifdef A8_PERFT
+            PerfMonitor::Instance()->server_send_bytes += bufflen;
+#endif
             a8::SendQueueNode* p = (a8::SendQueueNode*)malloc(sizeof(a8::SendQueueNode));
             memset(p, 0, sizeof(a8::SendQueueNode));
             p->buff = (char*)malloc(bufflen);
@@ -299,6 +302,9 @@ namespace a8
                 }
             }
             work_node_->sent_bytes += sentbytes;
+#ifdef A8_PERFT
+            PerfMonitor::Instance()->server_consume_bytes += sentbytes;
+#endif
             if (work_node_->sent_bytes >= work_node_->bufflen) {
                 a8::SendQueueNode *pdelnode = work_node_;
                 work_node_ = work_node_->next;

a8/websocketclient.cc

@@ -66,10 +66,10 @@ namespace a8
                 socket->SendBuff(data.data(), data.size());
             };
         tcp_client_->on_disconnect =
-            [this] (a8::AsioTcpClient* socket)
+            [this] (a8::AsioTcpClient* socket, int err)
             {
                 if (on_disconnect) {
-                    on_disconnect(this);
+                    on_disconnect(this, err);
                 }
             };
         tcp_client_->on_socketread =

a8/websocketclient.h

@@ -19,7 +19,7 @@ namespace a8
 
         std::function<void (a8::WebSocketClient*, int)> on_error;
         std::function<void (a8::WebSocketClient*)> on_connect;
-        std::function<void (a8::WebSocketClient*)> on_disconnect;
+        std::function<void (a8::WebSocketClient*, int)> on_disconnect;
         std::function<void (char*, int&, unsigned int)> on_decode_userpacket;
 
         void Open();

a8/xtimer.cc

@@ -45,6 +45,7 @@ struct xtimer_list {
     TimerType_e timer_type;
     long long expires;
     int expire_time;
+    int deleting;
     struct xtvec_base *base;
 
     a8::TimerCb cb;
@@ -220,6 +221,10 @@ namespace a8
             if (!timer) {
                 abort();
             }
+            if (timer->deleting) {
+                abort();
+            }
+            timer->deleting = 1;
             if (base_->running_timer == timer) {
                 base_->running_timer = nullptr;
             }
@@ -227,6 +232,16 @@ namespace a8
             if (!list_empty(&timer->attach_entry)) {
                 list_del_init(&timer->attach_entry);
             }
+#if 1
+            while (!list_empty(&timer->destory_handle_list)) {
+                XTimerDestoryHandleNode* node = list_first_entry(&timer->destory_handle_list,
+                                                                 XTimerDestoryHandleNode,
+                                                                 entry);
+                list_del_init(&node->entry);
+                node->cb(timer);
+                delete node;
+            }
+#endif
             if (is_destory) {
                 if (timer->cb) {
                     timer->cb(TIMER_DESTORY_EVENT, nullptr);
@@ -237,6 +252,7 @@ namespace a8
                 }
             }
             timer->cb = nullptr;
+#if 0
             while (!list_empty(&timer->destory_handle_list)) {
                 XTimerDestoryHandleNode* node = list_first_entry(&timer->destory_handle_list,
                                                                  XTimerDestoryHandleNode,
@@ -245,6 +261,7 @@ namespace a8
                 node->cb(timer);
                 delete node;
             }
+#endif
             if (to_free_list) {
                 AddToFreeList(timer);
             } else {
@@ -273,22 +290,36 @@ namespace a8
 
         void ClearAttacher(Attacher* attacher)
         {
+            #if 1
+            while (!list_empty(&attacher->timer_list_)) {
+                xtimer_list* tmp_timer = list_first_entry(&attacher->timer_list_, struct xtimer_list, attach_entry);
+                InternalDelete(tmp_timer, false, true);
+            }
+            #else
             struct list_head* pos = nullptr;
             struct list_head* n = nullptr;
             list_for_each_safe(pos, n, &attacher->timer_list_) {
                 xtimer_list* tmp_timer = list_entry(pos, struct xtimer_list, attach_entry);
                 InternalDelete(tmp_timer, false, true);
             }
+            #endif
         }
 
         void DestoryAttacher(Attacher* attacher)
         {
+            #if 1
+            while (!list_empty(&attacher->timer_list_)) {
+                xtimer_list* tmp_timer = list_first_entry(&attacher->timer_list_, struct xtimer_list, attach_entry);
+                InternalDelete(tmp_timer, true, true);
+            }
+            #else
             struct list_head* pos = nullptr;
             struct list_head* n = nullptr;
             list_for_each_safe(pos, n, &attacher->timer_list_) {
                 xtimer_list* tmp_timer = list_entry(pos, struct xtimer_list, attach_entry);
                 InternalDelete(tmp_timer, true, true);
             }
+            #endif
         }
 
         void UpdateTimer()
@@ -390,11 +421,14 @@ namespace a8
         void ClearTimer()
         {
             auto free_timers =
-                [this] (list_head* head)
+                [this] (list_head* head, bool force_no_deleteing = false)
                 {
                     while (!list_empty(head)) {
                         struct xtimer_list *timer;
                         timer = list_first_entry(head, struct xtimer_list,entry);
+                        if (force_no_deleteing) {
+                            timer->deleting = 0;
+                        }
                         InternalDelete(timer, true, false);
                     }
                 };
@@ -407,7 +441,7 @@ namespace a8
             for (int j = 0; j < TVR_SIZE; j++) {
                 free_timers(base_->tv1.vec + j);
             }
-            free_timers(&base_->free_timer);
+            free_timers(&base_->free_timer, true);
         }
 
         void AddTimerDestoryHandle(std::weak_ptr<XTimerPtr>& timer_ptr,
@@ -431,6 +465,7 @@ namespace a8
             timer->expires = expires;
             timer->expire_time = expire_time;
             timer->base = base_;
+            timer->deleting = 0;
             timer->cb = std::move(cb);
         }