1

a8/tcpsession2.cc

@@ -81,20 +81,46 @@ namespace a8
             memmove(p->buff, buff, bufflen);
             p->bufflen = bufflen;
             send_buffer_mutex_.lock();
-            if (bot_node_) {
+            if (sending_) {
-                bot_node_->next = p;
+                if (bot_node_) {
-                bot_node_ = p;
+                    bot_node_->next = p;
+                    bot_node_ = p;
+                } else {
+                    top_node_ = p;
+                    bot_node_ = p;
+                }
             } else {
-                top_node_ = p;
+                if (work_node_) {
-                bot_node_ = p;
+                    abort();
+                }
+                if (top_node_ || bot_node_) {
+                    abort();
+                }
+                int ret = DirectSend(p);
+                switch (ret) {
+                case 0:
+                case -2:
+                    {
+                        if (p->buff) {
+                            free(p->buff);
+                        }
+                        free(p);
+                    }
+                    break;
+                case -1:
+                    {
+                        if (top_node_ || bot_node_) {
+                            abort();
+                        }
+                        top_node_ = p;
+                        bot_node_ = p;
+                        NotifyEpollSend();
+                    }
+                    break;
+                default:
+                    abort();
+                }
             }
-            #if 1
-            NotifyEpollSend();
-            #else
-            if (!sending_) {
-                NotifyEpollSend();
-            }
-            #endif
             send_buffer_mutex_.unlock();
         }
     }
@@ -232,6 +258,7 @@ namespace a8
         if(socket_ == -1){
             return;
         }
+        ++epoll_out_times;
         if (!work_node_) {
             send_buffer_mutex_.lock();
             work_node_ = top_node_;
@@ -296,6 +323,7 @@ namespace a8
                 if (errno == EAGAIN || errno == EWOULDBLOCK) {
                     break;
                 } else {
+                    abort();
                     Close();
                     break;
                 }
@@ -306,18 +334,13 @@ namespace a8
                 work_node_ = work_node_->next;
                 if (!work_node_) {
                     send_buffer_mutex_.lock();
-                    if (!top_node_) {
+                    if (top_node_) {
                         work_node_ = top_node_;
                         top_node_ = nullptr;
                         bot_node_ = nullptr;
                     }
                     if (!work_node_) {
                         sending_ = false;
-                        struct epoll_event ev;
-                        ev.data.fd = socket_;
-                        ev.events = EPOLLIN | EPOLLET | EPOLLRDHUP;
-                        ev.data.ptr = this;
-                        ::epoll_ctl(epoll_fd, EPOLL_CTL_MOD, socket_, &ev);
                     }
                     send_buffer_mutex_.unlock();
                 }
@@ -339,5 +362,27 @@ namespace a8
         ::epoll_ctl(epoll_fd, EPOLL_CTL_MOD, socket_, &ev);
     }
 
+    int TcpSession::DirectSend(a8::SendQueueNode* p)
+    {
+        int sentbytes = ::send(socket_,
+                               p->buff,
+                               p->bufflen,
+                               0);
+        if (sentbytes >= p->bufflen) {
+            return 0;
+        }
+        if (sentbytes <= 0) {
+            if (errno == EAGAIN || errno == EWOULDBLOCK) {
+                return -1;
+            } else {
+                Close();
+                return -2;
+            }
+        } else {
+            p->sent_bytes += sentbytes;
+            return -1;
+        }
+    }
+
 }
 #endif

a8/tcpsession2.h

@@ -54,11 +54,14 @@ namespace a8
         void ClearSendBuff();
         void AsyncSend();
         void NotifyEpollSend();
+        int DirectSend(a8::SendQueueNode* node);
 
     protected:
         char *recv_buff_ = nullptr;
         int recv_bufflen_ = 0;
         int max_packet_len_ = 0;
+        volatile long long epoll_out_times = 0;
+        volatile long long epoll_in_times = 0;
 
     private:
         int socket_ = 0;