socket send timeout

include/swoole.h

@@ -2365,6 +2365,7 @@ typedef struct
      */
     uint32_t socket_buffer_size;
     swArray *socket_array;
+    double socket_send_timeout;
 
     swServer *serv;
 

include/swoole_config.h

@@ -39,13 +39,18 @@
 #define SW_MALLOC_TRIM_PAD         0
 #define SW_USE_MONOTONIC_TIME      1
 
+#define SW_MAX_SOCKETS_DEFAULT     1024
+
 #define SW_SOCKET_OVERFLOW_WAIT    100
 #define SW_SOCKET_MAX_DEFAULT      65536
 #if defined(__MACH__) || defined(__FreeBSD__)
 #define SW_SOCKET_BUFFER_SIZE      262144
 #else
 #define SW_SOCKET_BUFFER_SIZE      8388608
 #endif
+#define SW_SOCKET_SEND_TIMEOUT     1.0
+#define SW_SOCKET_ARRAY_INIT_SIZE  1024
+
 #define SW_SYSTEMD_FDS_START       3
 
 #define SW_GLOBAL_MEMORY_PAGESIZE  (2*1024*1024) // global memory page
@@ -61,6 +66,7 @@
 #define SW_CLIENT_CONNECT_TIMEOUT  0.5
 #define SW_CLIENT_MAX_PORT         65535
 
+
 // !!!Don't modify.----------------------------------------------------------
 #ifdef __MACH__
 #define SW_IPC_MAX_SIZE            2048  // MacOS

src/core/base.c

@@ -94,25 +94,26 @@ void swoole_init(void)
         exit(1);
     }
 
-    SwooleG.max_sockets = 1024;
+    SwooleG.max_sockets = SW_MAX_SOCKETS_DEFAULT;
     struct rlimit rlmt;
     if (getrlimit(RLIMIT_NOFILE, &rlmt) < 0)
     {
         swSysWarn("getrlimit() failed");
     }
     else
     {
-        SwooleG.max_sockets = MAX((uint32_t) rlmt.rlim_cur, 1024);
+        SwooleG.max_sockets = MAX((uint32_t) rlmt.rlim_cur, SW_MAX_SOCKETS_DEFAULT);
         SwooleG.max_sockets = MIN((uint32_t) rlmt.rlim_cur, SW_SESSION_LIST_SIZE);
     }
 
     SwooleG.socket_buffer_size = SW_SOCKET_BUFFER_SIZE;
-    SwooleG.socket_array = swArray_new(1024, sizeof(swSocket));
+    SwooleG.socket_array = swArray_new(SW_SOCKET_ARRAY_INIT_SIZE, sizeof(swSocket));
     if (!SwooleG.socket_array)
     {
         swSysWarn("[Core] Fatal Error: socket array memory allocation failure");
         exit(1);
     }
+    SwooleG.socket_send_timeout = SW_SOCKET_SEND_TIMEOUT;
 
     SwooleTG.buffer_stack = swString_new(SW_STACK_BUFFER_SIZE);
     if (SwooleTG.buffer_stack == NULL)

src/core/socket.c

@@ -296,22 +296,16 @@ ssize_t swSocket_sendto_blocking(int fd, const void *__buf, size_t __n, int flag
         {
             break;
         }
-        else
+        if (errno == EINTR)
         {
-            if (errno == EINTR)
-            {
-                continue;
-            }
-            else if (swConnection_error(errno) == SW_WAIT)
-            {
-                swSocket_wait(fd, 1000, SW_EVENT_WRITE);
-                continue;
-            }
-            else
-            {
-                break;
-            }
+            continue;
         }
+        if (swConnection_error(errno) == SW_WAIT
+                && swSocket_wait(fd, (int) (SwooleG.socket_send_timeout * 1000), SW_EVENT_WRITE) == SW_OK)
+        {
+            continue;
+        }
+        break;
     }
 
     return n;

swoole_async_coro.cc

@@ -93,6 +93,14 @@ PHP_FUNCTION(swoole_async_set)
             SwooleG.socket_buffer_size = INT_MAX;
         }
     }
+    if (php_swoole_array_get_value(vht, "socket_send_timeout", ztmp))
+    {
+        SwooleG.socket_send_timeout = zval_get_double(ztmp);
+        if (SwooleG.socket_send_timeout <= 0 || SwooleG.socket_send_timeout > INT_MAX)
+        {
+            SwooleG.socket_send_timeout = INT_MAX;
+        }
+    }
     if (php_swoole_array_get_value(vht, "log_level", ztmp))
     {
         zend_long level = zval_get_long(ztmp);

swoole_client.cc

@@ -982,7 +982,10 @@ static PHP_METHOD(swoole_client, sendto)
         swoole_set_object(ZEND_THIS, cli);
     }
 
-    int ret;
+    double ori_timeout = SwooleG.socket_send_timeout;
+    SwooleG.socket_send_timeout = cli->timeout;
+
+    int ret = -1;
     if (cli->type == SW_SOCK_UDP)
     {
         ret = swSocket_udp_sendto(cli->socket->fd, ip, port, data, len);
@@ -998,8 +1001,8 @@ static PHP_METHOD(swoole_client, sendto)
     else
     {
         php_swoole_fatal_error(E_WARNING, "only supports SWOOLE_SOCK_(UDP/UDP6/UNIX_DGRAM)");
-        RETURN_FALSE;
     }
+    SwooleG.socket_send_timeout = ori_timeout;
     SW_CHECK_RETURN(ret);
 }
 

tests/swoole_client_sync/udg_send_timeout.phpt

@@ -0,0 +1,25 @@
+--TEST--
+swoole_client_sync: udg send timeout
+--SKIPIF--
+<?php require __DIR__ . '/../include/skipif.inc'; ?>
+--FILE--
+<?php
+require __DIR__ . '/../include/bootstrap.php';
+const N = 65507;
+define("SOCKET_FILE", __DIR__.'/server.sock');
+$socket = stream_socket_server("udg://".SOCKET_FILE, $errno, $errstr, STREAM_SERVER_BIND);
+
+$client = new swoole_client(SWOOLE_SOCK_UNIX_DGRAM);
+$client->connect(SOCKET_FILE, 0, 0.3);
+$s = microtime(true);
+
+while (true) {
+    $re = $ret = $client->sendto(SOCKET_FILE, 0, str_repeat('B', N));
+    if ($re == false) {
+        break;
+    }
+}
+unlink(SOCKET_FILE);
+Assert::lessThan(microtime(true) - $s, 0.8);
+?>
+--EXPECT--

tests/swoole_server/sendto_timeout.phpt

@@ -0,0 +1,50 @@
+--TEST--
+swoole_server: sendto timeout
+--SKIPIF--
+<?php require __DIR__ . '/../include/skipif.inc'; ?>
+--FILE--
+<?php
+require __DIR__ . '/../include/bootstrap.php';
+
+const N = 65507;
+
+define('SOCK_FILE', __DIR__.'/server.sock');
+swoole_async_set(['socket_send_timeout' => 0.5]);
+$pm = new SwooleTest\ProcessManager;
+
+$pm->parentFunc = function ($pid) use ($pm) {
+    $client = new Swoole\Client(SWOOLE_SOCK_UNIX_DGRAM, SWOOLE_SOCK_SYNC);
+    if (!$client->connect(SOCK_FILE, 0, 0.5)) {
+        exit("connect failed\n");
+    }
+    $client->send(str_repeat('A', N));
+    $s = microtime(true);
+    $pm->wait();
+    Assert::lessThan(microtime(true) - $s, 0.6);
+    $data = $client->recv();
+    Assert::same(strlen($data), N);
+    $pm->kill();
+};
+
+$pm->childFunc = function () use ($pm) {
+    $serv = new swoole_server(SOCK_FILE, 0, SWOOLE_BASE, SWOOLE_SOCK_UNIX_DGRAM);
+    $serv->set(['worker_num' => 1, 'log_file' => '/dev/null']);
+    $serv->on("workerStart", function ($serv) use ($pm) {
+        $pm->wakeup();
+    });
+    $serv->on('packet', function ($serv, $data, $client) use ($pm) {
+        while (true) {
+            $re = $serv->send($client['address'], str_repeat('B', strlen($data)));
+            if ($re == false) {
+                break;
+            }
+        }
+        $pm->wakeup();
+    });
+    $serv->start();
+};
+
+$pm->childFirst();
+$pm->run();
+?>
+--EXPECT--