add timeout

10 files changed, 58 insertions, 30 deletions

diff --git a/xmake/modules/private/service/client.lua b/xmake/modules/private/service/client.lua
index 437a10f8a..92330fec4 100644
--- a/xmake/modules/private/service/client.lua
+++ b/xmake/modules/private/service/client.lua
@@ -22,12 +22,21 @@
 import("core.base.object")
 import("core.base.socket")
 import("core.base.scheduler")
+import("private.service.client_config", {alias = "config"})
 
 -- define module
 local client = client or object()
 
 -- init client
 function client:init()
+
+    -- init timeout
+    self._TIMEOUT = config.get("timeout") or -1
+end
+
+-- get timeout
+function client:timeout()
+    return self._TIMEOUT
 end
 
 -- parse host address
diff --git a/xmake/modules/private/service/distcc_build/client.lua b/xmake/modules/private/service/distcc_build/client.lua
index a34c17aa9..ad1ecf501 100644
--- a/xmake/modules/private/service/distcc_build/client.lua
+++ b/xmake/modules/private/service/distcc_build/client.lua
@@ -464,7 +464,7 @@ 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)
+            session = client_session(self, host_status.session_id, host_status.token, host_status.addr, host_status.port, {timeout = self:timeout()})
             host_status.sessions[i] = session
             session:open()
             return session
@@ -538,7 +538,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)
+        local stream = socket_stream(sock, {timeout = self: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 +587,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)
+        local stream = socket_stream(sock, {timeout = self: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 +638,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)
+        local stream = socket_stream(sock, {timeout = self: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 1c3d37efb..7d6f4e06e 100644
--- a/xmake/modules/private/service/distcc_build/client_session.lua
+++ b/xmake/modules/private/service/distcc_build/client_session.lua
@@ -35,12 +35,14 @@ import("private.service.stream", {alias = "socket_stream"})
 local client_session = client_session or object()
 
 -- init client session
-function client_session:init(client, session_id, token, addr, port)
+function client_session:init(client, session_id, token, addr, port, opt)
+    opt = opt or {}
     self._ID = session_id
     self._ADDR = addr
     self._PORT = port
     self._TOKEN = token
     self._CLIENT = client
+    self._TIMEOUT = opt.timeout and opt.timeout or -1
 end
 
 -- get client session id
@@ -58,6 +60,11 @@ function client_session:client()
     return self._CLIENT
 end
 
+-- get timeout
+function client_session:timeout()
+    return self._TIMEOUT
+end
+
 -- server unreachable?
 function client_session:is_unreachable()
     return self._UNREACHABLE
@@ -74,7 +81,7 @@ function client_session:stream()
             self._UNREACHABLE = true
             raise("%s: server unreachable!", self)
         end
-        stream = socket_stream(sock)
+        stream = socket_stream(sock, {timeout = self:timeout()})
         self._STREAM = stream
     end
     return stream
@@ -144,8 +151,8 @@ function client_session:__tostring()
     return string.format("<session %s>", self:id())
 end
 
-function main(client, session_id, token, addr, port)
+function main(client, session_id, token, addr, port, opt)
     local instance = client_session()
-    instance:init(client, session_id, token, addr, port)
+    instance:init(client, session_id, token, addr, port, opt)
     return instance
 end
diff --git a/xmake/modules/private/service/distcc_build/server.lua b/xmake/modules/private/service/distcc_build/server.lua
index 7f73820f2..0207761a1 100644
--- a/xmake/modules/private/service/distcc_build/server.lua
+++ b/xmake/modules/private/service/distcc_build/server.lua
@@ -23,7 +23,6 @@ import("core.base.global")
 import("private.service.server_config", {alias = "config"})
 import("private.service.message")
 import("private.service.server")
-import("private.service.stream", {alias = "socket_stream"})
 import("private.service.distcc_build.server_session")
 import("lib.detect.find_tool")
 
diff --git a/xmake/modules/private/service/remote_build/client.lua b/xmake/modules/private/service/remote_build/client.lua
index 2ff66c845..ef52ba8fd 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)
+        local stream = socket_stream(sock, {timeout = self: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)
+        local stream = socket_stream(sock, {timeout = self: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)
+        local stream = socket_stream(sock, {timeout = self: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)
+    local stream = socket_stream(sock, {timeout = self: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
@@ -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)
+    local stream = socket_stream(sock, {timeout = self: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_build/server.lua b/xmake/modules/private/service/remote_build/server.lua
index 31ddf4b4c..42e0e401c 100644
--- a/xmake/modules/private/service/remote_build/server.lua
+++ b/xmake/modules/private/service/remote_build/server.lua
@@ -23,7 +23,6 @@ import("core.base.global")
 import("private.service.server_config", {alias = "config"})
 import("private.service.message")
 import("private.service.server")
-import("private.service.stream", {alias = "socket_stream"})
 import("private.service.remote_build.server_session")
 import("lib.detect.find_tool")
 
diff --git a/xmake/modules/private/service/remote_cache/client.lua b/xmake/modules/private/service/remote_cache/client.lua
index 1a46cd34f..02c9915f8 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)
+        local stream = socket_stream(sock, {timeout = self: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)
+        local stream = socket_stream(sock, {timeout = self: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)
+    local stream = socket_stream(sock, {timeout = self: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)
+    local stream = socket_stream(sock, {timeout = self: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)
+    local stream = socket_stream(sock, {timeout = self: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)
+    local stream = socket_stream(sock, {timeout = self: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)
+    local stream = socket_stream(sock, {timeout = self: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/remote_cache/server.lua b/xmake/modules/private/service/remote_cache/server.lua
index 9ee83eb5b..ec4a31f67 100644
--- a/xmake/modules/private/service/remote_cache/server.lua
+++ b/xmake/modules/private/service/remote_cache/server.lua
@@ -23,7 +23,6 @@ import("core.base.global")
 import("private.service.server_config", {alias = "config"})
 import("private.service.message")
 import("private.service.server")
-import("private.service.stream", {alias = "socket_stream"})
 import("private.service.remote_cache.server_session")
 import("lib.detect.find_tool")
 
diff --git a/xmake/modules/private/service/server.lua b/xmake/modules/private/service/server.lua
index 204d6e70e..4b40373ff 100644
--- a/xmake/modules/private/service/server.lua
+++ b/xmake/modules/private/service/server.lua
@@ -43,6 +43,9 @@ function server:init(daemon)
     -- init known hosts
     local known_hosts = config.get("known_hosts")
     self:known_hosts_set(known_hosts)
+
+    -- init timeout
+    self._TIMEOUT = config.get("timeout") or -1
 end
 
 -- is daemon?
@@ -78,6 +81,11 @@ function server:port()
     return self._PORT
 end
 
+-- get timeout
+function server:timeout()
+    return self._TIMEOUT
+end
+
 -- get tokens
 function server:tokens()
     return self._TOKENS
@@ -181,7 +189,7 @@ end
 -- handle session
 function server:_handle_session(sock)
     print("%s: %s: session connected", self, sock)
-    local stream = socket_stream(sock)
+    local stream = socket_stream(sock, {timeout = self:timeout()})
     while true do
         local msg = stream:recv_object()
         if msg then
diff --git a/xmake/modules/private/service/stream.lua b/xmake/modules/private/service/stream.lua
index 561fea5f3..9037bdd8b 100644
--- a/xmake/modules/private/service/stream.lua
+++ b/xmake/modules/private/service/stream.lua
@@ -36,13 +36,15 @@ local STREAM_DATA_MAXN = 10 * 1024 * 1024
 local HEADER_FLAG_COMPRESS_LZ4 = 1
 
 -- init stream
-function stream:init(sock)
+function stream:init(sock, opt)
+    opt = opt or {}
     self._SOCK = sock
     self._BUFF = bytes(65536)
     self._RCACHE = bytes(8192)
     self._RCACHE_SIZE = 0
     self._WCACHE = bytes(8192)
     self._WCACHE_SIZE = 0
+    self._TIMEOUT = opt.timeout and opt.timeout or -1
 end
 
 -- get socket
@@ -50,6 +52,11 @@ function stream:sock()
     return self._SOCK
 end
 
+-- get timeout
+function stream:timeout()
+    return self._TIMEOUT
+end
+
 -- flush data
 function stream:flush()
     local cache = self._WCACHE
@@ -93,7 +100,7 @@ function stream:send(data, start, last)
 
     -- send data to socket
     local sock = self._SOCK
-    local real = sock:send(cache, {block = true})
+    local real = sock:send(cache, {block = true, timeout = self:timeout()})
     if real > 0 then
         -- copy left data to cache
         assert(size <= cache_maxn)
@@ -201,7 +208,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})
+        local send = sock:sendfile(file, {block = true, timeout = self:timeout()})
         if send > 0 then
             ok = true
         end
@@ -280,7 +287,7 @@ function stream:recv(buff, size)
             end
             wait = false
         elseif real == 0 and not wait then
-            if sock:wait(socket.EV_RECV, -1) == socket.EV_RECV then
+            if sock:wait(socket.EV_RECV, self:timeout()) == socket.EV_RECV then
                 wait = true
             else
                 break
@@ -442,8 +449,8 @@ function stream:_recv_compressed_file(lz4_stream, filepath, size)
     end
 end
 
-function main(sock)
+function main(sock, opt)
     local instance = stream()
-    instance:init(sock)
+    instance:init(sock, opt)
     return instance
 end