add tests 9 --filter=[core]

Author: matyhtf

core-tests/src/server/server.cpp

@@ -3002,10 +3002,13 @@ TEST(server, send_timeout) {
 
     port->max_idle_time = 1;
 
+    String wbuf(2 * 1024 * 1024);
+    wbuf.append_random_bytes(2 * 1024 * 1024, false);
+
     ASSERT_EQ(serv.create(), SW_OK);
 
     thread t1;
-    serv.onStart = [&lock, &t1](Server *serv) {
+    serv.onStart = [&lock, &t1, &wbuf](Server *serv) {
         t1 = thread([=]() {
             swoole_signal_block_all();
 
@@ -3016,9 +3019,24 @@ TEST(server, send_timeout) {
             network::SyncClient c(SW_SOCK_TCP);
             c.connect(TEST_HOST, port->port);
             c.send(packet, strlen(packet));
-            char buf[1024];
-            EXPECT_EQ(c.recv(buf, sizeof(buf)), sizeof(buf));
-            EXPECT_EQ(c.recv(buf, sizeof(buf)), 0);
+
+            String rbuf(3 * 1024 * 1024);
+
+            auto rn = c.recv(rbuf.str, 1024);
+            EXPECT_EQ(rn, 1024);
+            rbuf.length += 1024;
+
+            sleep(2);
+
+            while (true) {
+                rn = c.recv(rbuf.str + rbuf.length, rbuf.size - rbuf.length);
+                if (rn <= 0) {
+                    break;
+                }
+                rbuf.length += rn;
+            }
+
+            EXPECT_MEMEQ(rbuf.str, wbuf.str, rbuf.length);
             c.close();
 
             kill(serv->gs->master_pid, SIGTERM);
@@ -3033,12 +3051,15 @@ TEST(server, send_timeout) {
         DEBUG() << "onWorkerStart: id=" << worker->id << "\n";
     };
 
-    serv.onReceive = [](Server *serv, RecvData *req) -> int {
+    serv.onReceive = [&wbuf](Server *serv, RecvData *req) -> int {
         auto sid = req->session_id();
-        auto buf = sw_tg_buffer();
-        buf->clear();
-        buf->append_random_bytes(1024, false);
-        EXPECT_TRUE(serv->send(sid, buf->str, buf->length));
+        auto conn = serv->get_connection_by_session_id(sid);
+
+        swoole_timer_del(conn->socket->recv_timer);
+        conn->socket->recv_timer = nullptr;
+        conn->socket->set_buffer_size(65536);
+
+        EXPECT_TRUE(serv->send(sid, wbuf.str, wbuf.length));
 
         test::counter_incr(0);
 

src/server/master.cc

@@ -35,6 +35,10 @@ TimerCallback Server::get_timeout_callback(ListenPort *port, Reactor *reactor, C
             return;
         }
         long ms = time<std::chrono::milliseconds>(true);
+        swoole_trace_log(SW_TRACE_SERVER,
+                         "timeout_callback, last_received_time=%f, last_sent_time=%f",
+                         conn->socket->last_received_time,
+                         conn->socket->last_sent_time);
         if (ms - conn->socket->last_received_time < port->max_idle_time &&
             ms - conn->socket->last_sent_time < port->max_idle_time) {
             return;
@@ -1513,6 +1517,10 @@ int Server::send_to_connection(SendData *_send) {
         _socket->send_timeout_ = port->max_idle_time;
         _socket->last_sent_time = time<std::chrono::milliseconds>(true);
         _socket->send_timer = swoole_timer_add((long) (port->max_idle_time * 1000), true, timeout_callback);
+        swoole_trace_log(SW_TRACE_SERVER,
+                         "added send_timer[id=%ld], port->max_idle_time=%f",
+                         _socket->send_timer->id,
+                         port->max_idle_time);
     }
 
     if (!_socket->isset_writable_event()) {

src/server/reactor_thread.cc

@@ -578,8 +578,9 @@ static int ReactorThread_onRead(Reactor *reactor, Event *event) {
 static int ReactorThread_onWrite(Reactor *reactor, Event *ev) {
     int ret;
     auto serv = static_cast<Server *>(reactor->ptr);
-    Socket *socket = ev->socket;
+    auto socket = ev->socket;
     int fd = ev->fd;
+    auto port = serv->get_port_by_fd(fd);
 
     if (serv->is_process_mode()) {
         assert(fd % serv->reactor_num == reactor->id);
@@ -639,16 +640,19 @@ static int ReactorThread_onWrite(Reactor *reactor, Event *ev) {
     }
 
     if (serv->onBufferEmpty && conn->high_watermark) {
-        ListenPort *port = serv->get_port_by_fd(fd);
         if (socket->get_out_buffer_length() <= port->buffer_low_watermark) {
             conn->high_watermark = 0;
             serv->notify(conn, SW_SERVER_EVENT_BUFFER_EMPTY);
         }
     }
 
     if (socket->send_timer) {
-        swoole_timer_del(socket->send_timer);
-        socket->send_timer = nullptr;
+        if (Buffer::empty(socket->out_buffer)) {
+            swoole_timer_del(socket->send_timer);
+            socket->send_timer = nullptr;
+        } else {
+            swoole_timer_delay(socket->send_timer, port->max_idle_time);
+        }
     }
 
     // remove EPOLLOUT event