add recv/send timeout

9 files changed, 67 insertions, 38 deletions

diff --git a/xmake/modules/private/service/client.lua b/xmake/modules/private/service/client.lua
index 92330fec4..4eb18d1a1 100644
--- a/xmake/modules/private/service/client.lua
+++ b/xmake/modules/private/service/client.lua
@@ -31,12 +31,18 @@ local client = client or object()
 function client:init()
 
     -- init timeout
-    self._TIMEOUT = config.get("timeout") or -1
+    self._SEND_TIMEOUT = config.get("send_timeout") or -1
+    self._RECV_TIMEOUT = config.get("recv_timeout") or -1
 end
 
--- get timeout
-function client:timeout()
-    return self._TIMEOUT
+-- get send timeout
+function client:send_timeout()
+    return self._SEND_TIMEOUT
+end
+
+-- get recv timeout
+function client:recv_timeout()
+    return self._RECV_TIMEOUT
 end
 
 -- parse host address
diff --git a/xmake/modules/private/service/client_config.lua b/xmake/modules/private/service/client_config.lua
index b65ef8d13..fef687ef7 100644
--- a/xmake/modules/private/service/client_config.lua
+++ b/xmake/modules/private/service/client_config.lua
@@ -39,6 +39,8 @@ function _generate_configfile()
     local token = _get_local_server_token()
     print("generating the config file to %s ..", filepath)
     local configs = {
+        send_timeout = -1,
+        recv_timeout = -1,
         remote_build = {
             -- without authorization: "127.0.0.1:9691"
             -- with user authorization: "[email protected]:9691"
diff --git a/xmake/modules/private/service/distcc_build/client.lua b/xmake/modules/private/service/distcc_build/client.lua
index ad1ecf501..85d82a9c3 100644
--- a/xmake/modules/private/service/distcc_build/client.lua
+++ b/xmake/modules/private/service/distcc_build/client.lua
@@ -464,7 +464,8 @@ function distcc_build_client:_host_status_session_open(host_status)
     for i = 1, njob do
         local session = host_status.sessions[i]
         if not session then
-            session = client_session(self, host_status.session_id, host_status.token, host_status.addr, host_status.port, {timeout = self:timeout()})
+            session = client_session(self, host_status.session_id, host_status.token, host_status.addr, host_status.port,
+                {send_timeout = self:send_timeout(), recv_timeout = self:recv_timeout()})
             host_status.sessions[i] = session
             session:open()
             return session
@@ -538,7 +539,7 @@ function distcc_build_client:_connect_host(host)
     local ncpu, njob
     print("%s: connect %s:%d ..", self, addr, port)
     if sock then
-        local stream = socket_stream(sock, {timeout = self:timeout()})
+        local stream = socket_stream(sock, {send_timeout = self:send_timeout(), recv_timeout = self:recv_timeout()})
         if stream:send_msg(message.new_connect(session_id, {token = token})) and stream:flush() then
             local msg = stream:recv_msg()
             if msg then
@@ -587,7 +588,7 @@ function distcc_build_client:_disconnect_host(host)
     local ok = false
     print("%s: disconnect %s:%d ..", self, addr, port)
     if sock then
-        local stream = socket_stream(sock, {timeout = self:timeout()})
+        local stream = socket_stream(sock, {send_timeout = self:send_timeout(), recv_timeout = self:recv_timeout()})
         if stream:send_msg(message.new_disconnect(session_id, {token = token})) and stream:flush() then
             local msg = stream:recv_msg()
             if msg then
@@ -638,7 +639,7 @@ function distcc_build_client:_clean_host(host)
     local ok = false
     print("%s: clean files in %s:%d ..", self, addr, port)
     if sock then
-        local stream = socket_stream(sock, {timeout = self:timeout()})
+        local stream = socket_stream(sock, {send_timeout = self:send_timeout(), recv_timeout = self:recv_timeout()})
         if stream:send_msg(message.new_clean(session_id, {token = token})) and stream:flush() then
             local msg = stream:recv_msg()
             if msg then
diff --git a/xmake/modules/private/service/distcc_build/client_session.lua b/xmake/modules/private/service/distcc_build/client_session.lua
index 79d01d939..f4ff8c5ef 100644
--- a/xmake/modules/private/service/distcc_build/client_session.lua
+++ b/xmake/modules/private/service/distcc_build/client_session.lua
@@ -42,7 +42,8 @@ function client_session:init(client, session_id, token, addr, port, opt)
     self._PORT = port
     self._TOKEN = token
     self._CLIENT = client
-    self._TIMEOUT = opt.timeout and opt.timeout or -1
+    self._SEND_TIMEOUT = opt.send_timeout and opt.send_timeout or -1
+    self._RECV_TIMEOUT = opt.recv_timeout and opt.recv_timeout or -1
 end
 
 -- get client session id
@@ -60,9 +61,14 @@ function client_session:client()
     return self._CLIENT
 end
 
--- get timeout
-function client_session:timeout()
-    return self._TIMEOUT
+-- get send timeout
+function client_session:send_timeout()
+    return self._SEND_TIMEOUT
+end
+
+-- get recv timeout
+function client_session:recv_timeout()
+    return self._RECV_TIMEOUT
 end
 
 -- server unreachable?
@@ -81,7 +87,7 @@ function client_session:stream()
             self._UNREACHABLE = true
             raise("%s: server unreachable!", self)
         end
-        stream = socket_stream(sock, {timeout = self:timeout()})
+        stream = socket_stream(sock, {send_timeout = self:send_timeout(), recv_timeout = self:recv_timeout()})
         self._STREAM = stream
     end
     return stream
diff --git a/xmake/modules/private/service/remote_build/client.lua b/xmake/modules/private/service/remote_build/client.lua
index 993f1acf0..13018a970 100644
--- a/xmake/modules/private/service/remote_build/client.lua
+++ b/xmake/modules/private/service/remote_build/client.lua
@@ -98,7 +98,7 @@ function remote_build_client:connect()
     local errors
     print("%s: connect %s:%d ..", self, addr, port)
     if sock then
-        local stream = socket_stream(sock, {timeout = self:timeout()})
+        local stream = socket_stream(sock, {send_timeout = self:send_timeout(), recv_timeout = self:recv_timeout()})
         if stream:send_msg(message.new_connect(session_id, {token = token})) and stream:flush() then
             local msg = stream:recv_msg()
             if msg then
@@ -146,7 +146,7 @@ function remote_build_client:disconnect()
     local ok = false
     print("%s: disconnect %s:%d ..", self, addr, port)
     if sock then
-        local stream = socket_stream(sock, {timeout = self:timeout()})
+        local stream = socket_stream(sock, {send_timeout = self:send_timeout(), recv_timeout = self:recv_timeout()})
         if stream:send_msg(message.new_disconnect(session_id, {token = self:token()})) and stream:flush() then
             local msg = stream:recv_msg()
             if msg then
@@ -190,7 +190,7 @@ function remote_build_client:sync()
     while sock do
 
         -- diff files
-        local stream = socket_stream(sock, {timeout = self:timeout()})
+        local stream = socket_stream(sock, {send_timeout = self:send_timeout(), recv_timeout = self:recv_timeout()})
         diff_files, errors = self:_diff_files(stream)
         if not diff_files then
             break
@@ -240,7 +240,7 @@ function remote_build_client:clean()
     local errors
     local ok = false
     print("%s: clean files in %s:%d ..", self, addr, port)
-    local stream = socket_stream(sock, {timeout = self:timeout()})
+    local stream = socket_stream(sock, {send_timeout = self:send_timeout(), recv_timeout = self:recv_timeout()})
     if stream:send_msg(message.new_clean(session_id, {token = self:token()})) and stream:flush() then
         local msg = stream:recv_msg({timeout = -1})
         if msg then
@@ -272,7 +272,7 @@ function remote_build_client:runcmd(program, argv)
     local command = os.args(table.join(program, argv))
     local leftstr = ""
     cprint("%s: run ${bright}%s${clear} in %s:%d ..", self, command, addr, port)
-    local stream = socket_stream(sock, {timeout = self:timeout()})
+    local stream = socket_stream(sock, {send_timeout = self:send_timeout(), recv_timeout = self:recv_timeout()})
     if stream:send_msg(message.new_runcmd(session_id, program, argv, {token = self:token()})) and stream:flush() then
         local stdin_opt = {stop = false}
         local group_name = "remote_build/runcmd"
diff --git a/xmake/modules/private/service/remote_cache/client.lua b/xmake/modules/private/service/remote_cache/client.lua
index 02c9915f8..4c35b6411 100644
--- a/xmake/modules/private/service/remote_cache/client.lua
+++ b/xmake/modules/private/service/remote_cache/client.lua
@@ -96,7 +96,7 @@ function remote_cache_client:connect()
     local errors
     print("%s: connect %s:%d ..", self, addr, port)
     if sock then
-        local stream = socket_stream(sock, {timeout = self:timeout()})
+        local stream = socket_stream(sock, {send_timeout = self:send_timeout(), recv_timeout = self:recv_timeout()})
         if stream:send_msg(message.new_connect(session_id, {token = token})) and stream:flush() then
             local msg = stream:recv_msg()
             if msg then
@@ -139,7 +139,7 @@ function remote_cache_client:disconnect()
     local ok = false
     print("%s: disconnect %s:%d ..", self, addr, port)
     if sock then
-        local stream = socket_stream(sock, {timeout = self:timeout()})
+        local stream = socket_stream(sock, {send_timeout = self:send_timeout(), recv_timeout = self:recv_timeout()})
         if stream:send_msg(message.new_disconnect(session_id, {token = self:token()})) and stream:flush() then
             local msg = stream:recv_msg()
             if msg then
@@ -181,7 +181,7 @@ function remote_cache_client:pull(cachekey, cachefile)
     local exists = false
     local extrainfo
     dprint("%s: pull cache(%s) in %s:%d ..", self, cachekey, addr, port)
-    local stream = socket_stream(sock, {timeout = self:timeout()})
+    local stream = socket_stream(sock, {send_timeout = self:send_timeout(), recv_timeout = self:recv_timeout()})
     if stream:send_msg(message.new_pull(session_id, cachekey, {token = self:token()})) and stream:flush() then
         if stream:recv_file(cachefile) then
             local msg = stream:recv_msg()
@@ -218,7 +218,7 @@ function remote_cache_client:push(cachekey, cachefile, extrainfo)
     local errors
     local ok = false
     dprint("%s: push cache(%s) in %s:%d ..", self, cachekey, addr, port)
-    local stream = socket_stream(sock, {timeout = self:timeout()})
+    local stream = socket_stream(sock, {send_timeout = self:send_timeout(), recv_timeout = self:recv_timeout()})
     if stream:send_msg(message.new_push(session_id, cachekey, {token = self:token(), extrainfo = extrainfo})) and stream:flush() then
         if stream:send_file(cachefile, {compress = os.filesize(cachefile) > 4096}) then
             local msg = stream:recv_msg()
@@ -253,7 +253,7 @@ function remote_cache_client:cacheinfo(cachekey)
     local ok = false
     local cacheinfo
     dprint("%s: get cacheinfo(%s) in %s:%d ..", self, cachekey, addr, port)
-    local stream = socket_stream(sock, {timeout = self:timeout()})
+    local stream = socket_stream(sock, {send_timeout = self:send_timeout(), recv_timeout = self:recv_timeout()})
     if stream:send_msg(message.new_fileinfo(session_id, cachekey, {token = self:token()})) and stream:flush() then
         local msg = stream:recv_msg()
         if msg then
@@ -285,7 +285,7 @@ function remote_cache_client:existinfo()
     local errors
     local existinfo
     dprint("%s: get exist info in %s:%d ..", self, addr, port)
-    local stream = socket_stream(sock, {timeout = self:timeout()})
+    local stream = socket_stream(sock, {send_timeout = self:send_timeout(), recv_timeout = self:recv_timeout()})
     if stream:send_msg(message.new_existinfo(session_id, "objectfiles", {token = self:token()})) and stream:flush() then
         local data = stream:recv_data()
         if data then
@@ -326,7 +326,7 @@ function remote_cache_client:clean()
     local errors
     local ok = false
     print("%s: clean files in %s:%d ..", self, addr, port)
-    local stream = socket_stream(sock, {timeout = self:timeout()})
+    local stream = socket_stream(sock, {send_timeout = self:send_timeout(), recv_timeout = self:recv_timeout()})
     if stream:send_msg(message.new_clean(session_id, {token = self:token()})) and stream:flush() then
         local msg = stream:recv_msg()
         if msg then
diff --git a/xmake/modules/private/service/server.lua b/xmake/modules/private/service/server.lua
index 4dcbbf4ec..891942a03 100644
--- a/xmake/modules/private/service/server.lua
+++ b/xmake/modules/private/service/server.lua
@@ -45,7 +45,8 @@ function server:init(daemon)
     self:known_hosts_set(known_hosts)
 
     -- init timeout
-    self._TIMEOUT = config.get("timeout") or -1
+    self._SEND_TIMEOUT = config.get("send_timeout") or -1
+    self._RECV_TIMEOUT = config.get("recv_timeout") or -1
 end
 
 -- is daemon?
@@ -81,9 +82,14 @@ function server:port()
     return self._PORT
 end
 
--- get timeout
-function server:timeout()
-    return self._TIMEOUT
+-- get send timeout
+function server:send_timeout()
+    return self._SEND_TIMEOUT
+end
+
+-- get recv timeout
+function server:recv_timeout()
+    return self._RECV_TIMEOUT
 end
 
 -- get tokens
@@ -189,7 +195,7 @@ end
 -- handle session
 function server:_handle_session(sock)
     print("%s: %s: session connected", self, sock)
-    local stream = socket_stream(sock, {timeout = self:timeout()})
+    local stream = socket_stream(sock, {send_timeout = self:send_timeout(), recv_timeout = self:recv_timeout()})
     while true do
         local msg = stream:recv_object({timeout = -1})
         if msg then
diff --git a/xmake/modules/private/service/server_config.lua b/xmake/modules/private/service/server_config.lua
index ca007e8d7..5e6ec9c71 100644
--- a/xmake/modules/private/service/server_config.lua
+++ b/xmake/modules/private/service/server_config.lua
@@ -44,6 +44,8 @@ function _generate_configfile()
         tokens = {
             token
         },
+        send_timeout = -1,
+        recv_timeout = -1,
         remote_build = {
             listen = "0.0.0.0:9691",
             workdir = path.join(servicedir, "remote_build"),
diff --git a/xmake/modules/private/service/stream.lua b/xmake/modules/private/service/stream.lua
index 7b40e4b8b..328cd2f10 100644
--- a/xmake/modules/private/service/stream.lua
+++ b/xmake/modules/private/service/stream.lua
@@ -44,7 +44,8 @@ function stream:init(sock, opt)
     self._RCACHE_SIZE = 0
     self._WCACHE = bytes(8192)
     self._WCACHE_SIZE = 0
-    self._TIMEOUT = opt.timeout and opt.timeout or -1
+    self._SEND_TIMEOUT = opt.send_timeout and opt.send_timeout or -1
+    self._RECV_TIMEOUT = opt.recv_timeout and opt.recv_timeout or -1
 end
 
 -- get socket
@@ -52,9 +53,14 @@ function stream:sock()
     return self._SOCK
 end
 
--- get timeout
-function stream:timeout()
-    return self._TIMEOUT
+-- get send timeout
+function stream:send_timeout()
+    return self._SEND_TIMEOUT
+end
+
+-- get send timeout
+function stream:recv_timeout()
+    return self._RECV_TIMEOUT
 end
 
 -- flush data
@@ -64,7 +70,7 @@ function stream:flush(opt)
     local cache_size = self._WCACHE_SIZE
     if cache_size > 0 then
         local sock = self._SOCK
-        local real = sock:send(cache, {block = true, last = cache_size, timeout = opt.timeout or self:timeout()})
+        local real = sock:send(cache, {block = true, last = cache_size, timeout = opt.timeout or self:send_timeout()})
         if real > 0 then
             self._WCACHE_SIZE = 0
             return true
@@ -102,7 +108,7 @@ function stream:send(data, start, last, opt)
 
     -- send data to socket
     local sock = self._SOCK
-    local real = sock:send(cache, {block = true, timeout = opt.timeout or self:timeout()})
+    local real = sock:send(cache, {block = true, timeout = opt.timeout or self:send_timeout()})
     if real > 0 then
         -- copy left data to cache
         assert(size <= cache_maxn)
@@ -210,7 +216,7 @@ function stream:send_file(filepath, opt)
     local sock = self._SOCK
     local file = io.open(filepath, 'rb')
     if file then
-        local send = sock:sendfile(file, {block = true, timeout = opt.timeout or self:timeout()})
+        local send = sock:sendfile(file, {block = true, timeout = opt.timeout or self:send_timeout()})
         if send > 0 then
             ok = true
         end
@@ -290,7 +296,7 @@ function stream:recv(buff, size, opt)
             end
             wait = false
         elseif real == 0 and not wait then
-            local ok = sock:wait(socket.EV_RECV, opt.timeout or self:timeout())
+            local ok = sock:wait(socket.EV_RECV, opt.timeout or self:recv_timeout())
             if ok == socket.EV_RECV then
                 wait = true
             else