Merge pull request #2530 from xmake-io/timeout

add timeout for service

14 files changed, 205 insertions, 80 deletions

diff --git a/xmake/core/base/socket.lua b/xmake/core/base/socket.lua
index 615f28a0c..b9da2e99a 100644
--- a/xmake/core/base/socket.lua
+++ b/xmake/core/base/socket.lua
@@ -318,6 +318,8 @@ function _instance:send(data, opt)
                 local events, waiterrs = _instance.wait(self, socket.EV_SEND, opt.timeout or -1)
                 if events == socket.EV_SEND then
                     wait = true
+                elseif events == 0 then
+                    os.raise("%s: send timeout!", self)
                 else
                     errors = waiterrs
                     break
@@ -385,6 +387,8 @@ function _instance:sendfile(file, opt)
                 local events, waiterrs = _instance.wait(self, socket.EV_SEND, opt.timeout or -1)
                 if events == socket.EV_SEND then
                     wait = true
+                elseif events == 0 then
+                    os.raise("%s: sendfile timeout!", self)
                 else
                     errors = waiterrs
                     break
@@ -451,6 +455,8 @@ function _instance:recv(buff, size, opt)
                 local events, waiterrs = _instance.wait(self, socket.EV_RECV, opt.timeout or -1)
                 if events == socket.EV_RECV then
                     wait = true
+                elseif events == 0 then
+                    os.raise("%s: recv timeout!", self)
                 else
                     data_or_errors = waiterrs
                     break
diff --git a/xmake/modules/private/service/client.lua b/xmake/modules/private/service/client.lua
index 437a10f8a..c3af668d8 100644
--- a/xmake/modules/private/service/client.lua
+++ b/xmake/modules/private/service/client.lua
@@ -22,12 +22,33 @@
 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._SEND_TIMEOUT = config.get("send_timeout") or -1
+    self._RECV_TIMEOUT = config.get("recv_timeout") or -1
+    self._CONNECT_TIMEOUT = config.get("connect_timeout") or -1
+end
+
+-- get send timeout
+function client:send_timeout()
+    return self._SEND_TIMEOUT
+end
+
+-- get recv timeout
+function client:recv_timeout()
+    return self._RECV_TIMEOUT
+end
+
+-- get connect timeout
+function client:connect_timeout()
+    return self._CONNECT_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..3c5156673 100644
--- a/xmake/modules/private/service/client_config.lua
+++ b/xmake/modules/private/service/client_config.lua
@@ -39,6 +39,9 @@ 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,
+        connect_timeout = -1,
         remote_build = {
             -- without authorization: "127.0.0.1:9691"
             -- with user authorization: "[email protected]:9691"
@@ -58,7 +61,6 @@ function _generate_configfile()
                 {connect = "127.0.0.1:9693", token = token}
             }
         }
-
     }
     save(configs)
 end
diff --git a/xmake/modules/private/service/distcc_build/client.lua b/xmake/modules/private/service/distcc_build/client.lua
index a34c17aa9..7520be46b 100644
--- a/xmake/modules/private/service/distcc_build/client.lua
+++ b/xmake/modules/private/service/distcc_build/client.lua
@@ -55,6 +55,11 @@ function distcc_build_client:init()
     else
         raise("we need enter a project directory with xmake.lua first!")
     end
+
+    -- init timeout
+    self._SEND_TIMEOUT = config.get("distcc_build.send_timeout") or config.get("send_timeout") or -1
+    self._RECV_TIMEOUT = config.get("distcc_build.recv_timeout") or config.get("recv_timeout") or -1
+    self._CONNECT_TIMEOUT = config.get("distcc_build.connect_timeout") or config.get("connect_timeout") or -1
 end
 
 -- get class
@@ -464,7 +469,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)
+            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(), connect_timeout = self:connect_timeout()})
             host_status.sessions[i] = session
             session:open()
             return session
@@ -531,14 +537,14 @@ function distcc_build_client:_connect_host(host)
     end
 
     -- do connect
-    local sock = assert(socket.connect(addr, port), "%s: server unreachable!", self)
+    local sock = assert(socket.connect(addr, port, {timeout = self:connect_timeout()}), "%s: server unreachable!", self)
     local session_id = self:_session_id(addr, port)
     local ok = false
     local errors
     local ncpu, njob
     print("%s: connect %s:%d ..", self, addr, port)
     if sock then
-        local stream = socket_stream(sock)
+        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
@@ -581,13 +587,13 @@ function distcc_build_client:_disconnect_host(host)
 
     -- do disconnect
     local token = host.token
-    local sock = socket.connect(addr, port)
+    local sock = socket.connect(addr, port, {timeout = self:connect_timeout()})
     local session_id = self:_session_id(addr, port)
     local errors
     local ok = false
     print("%s: disconnect %s:%d ..", self, addr, port)
     if sock then
-        local stream = socket_stream(sock)
+        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
@@ -632,13 +638,13 @@ function distcc_build_client:_clean_host(host)
 
     -- do clean
     local token = host.token
-    local sock = socket.connect(addr, port)
+    local sock = socket.connect(addr, port, {timeout = self:connect_timeout()})
     local session_id = self:_session_id(addr, port)
     local errors
     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, {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 1c3d37efb..d913878fa 100644
--- a/xmake/modules/private/service/distcc_build/client_session.lua
+++ b/xmake/modules/private/service/distcc_build/client_session.lua
@@ -35,12 +35,16 @@ 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._SEND_TIMEOUT = opt.send_timeout and opt.send_timeout or -1
+    self._RECV_TIMEOUT = opt.recv_timeout and opt.recv_timeout or -1
+    self._CONNECT_TIMEOUT = opt.connect_timeout and opt.connect_timeout or -1
 end
 
 -- get client session id
@@ -58,6 +62,21 @@ function client_session:client()
     return self._CLIENT
 end
 
+-- 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
+
+-- get connect timeout
+function client_session:connect_timeout()
+    return self._CONNECT_TIMEOUT
+end
+
 -- server unreachable?
 function client_session:is_unreachable()
     return self._UNREACHABLE
@@ -69,12 +88,12 @@ function client_session:stream()
     if stream == nil then
         local addr = self._ADDR
         local port = self._PORT
-        local sock = socket.connect(addr, port)
+        local sock = socket.connect(addr, port, {timeout = self:connect_timeout()})
         if not sock then
             self._UNREACHABLE = true
             raise("%s: server unreachable!", self)
         end
-        stream = socket_stream(sock)
+        stream = socket_stream(sock, {send_timeout = self:send_timeout(), recv_timeout = self:recv_timeout()})
         self._STREAM = stream
     end
     return stream
@@ -113,7 +132,7 @@ function client_session:compile(sourcefile, objectfile, cppfile, cppflags, opt)
     if stream:send_msg(message.new_compile(self:id(), toolname, toolkind, plat, arch, toolchain,
             cppflags, path.filename(sourcefile), {token = self:token(), cachekey = cachekey})) and
         stream:send_file(cppfile, {compress = os.filesize(cppfile) > 4096}) and stream:flush() then
-        local recv = stream:recv_file(objectfile)
+        local recv = stream:recv_file(objectfile, {timeout = -1})
         if recv ~= nil then
             local msg = stream:recv_msg()
             if msg then
@@ -144,8 +163,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..68e14de8f 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")
 
@@ -44,6 +43,10 @@ function distcc_build_server:init(daemon)
 
     -- init sessions
     self._SESSIONS = {}
+
+    -- init timeout
+    self._SEND_TIMEOUT = config.get("distcc_build.send_timeout") or config.get("send_timeout") or -1
+    self._RECV_TIMEOUT = config.get("distcc_build.recv_timeout") or config.get("recv_timeout") or -1
 end
 
 -- get class
diff --git a/xmake/modules/private/service/remote_build/client.lua b/xmake/modules/private/service/remote_build/client.lua
index 2ff66c845..95ad5110e 100644
--- a/xmake/modules/private/service/remote_build/client.lua
+++ b/xmake/modules/private/service/remote_build/client.lua
@@ -60,6 +60,11 @@ function remote_build_client:init()
     filesync:ignorefiles_add(".git/**")
     filesync:ignorefiles_add(".xmake/**")
     self._FILESYNC = filesync
+
+    -- init timeout
+    self._SEND_TIMEOUT = config.get("remote_build.send_timeout") or config.get("send_timeout") or -1
+    self._RECV_TIMEOUT = config.get("remote_build.recv_timeout") or config.get("recv_timeout") or -1
+    self._CONNECT_TIMEOUT = config.get("remote_build.connect_timeout") or config.get("connect_timeout") or -1
 end
 
 -- get class
@@ -92,13 +97,13 @@ function remote_build_client:connect()
     -- do connect
     local addr = self:addr()
     local port = self:port()
-    local sock = assert(socket.connect(addr, port), "%s: server unreachable!", self)
+    local sock = assert(socket.connect(addr, port, {timeout = self:connect_timeout()}), "%s: server unreachable!", self)
     local session_id = self:session_id()
     local ok = false
     local errors
     print("%s: connect %s:%d ..", self, addr, port)
     if sock then
-        local stream = socket_stream(sock)
+        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
@@ -140,13 +145,13 @@ function remote_build_client:disconnect()
     end
     local addr = self:addr()
     local port = self:port()
-    local sock = socket.connect(addr, port)
+    local sock = socket.connect(addr, port, {timeout = self:connect_timeout()})
     local session_id = self:session_id()
     local errors
     local ok = false
     print("%s: disconnect %s:%d ..", self, addr, port)
     if sock then
-        local stream = socket_stream(sock)
+        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 +186,7 @@ function remote_build_client:sync()
     assert(self:is_connected(), "%s: has been not connected!", self)
     local addr = self:addr()
     local port = self:port()
-    local sock = assert(socket.connect(addr, port), "%s: server unreachable!", self)
+    local sock = assert(socket.connect(addr, port, {timeout = self:connect_timeout()}), "%s: server unreachable!", self)
     local session_id = self:session_id()
     local errors
     local ok = false
@@ -190,7 +195,7 @@ function remote_build_client:sync()
     while sock do
 
         -- diff files
-        local stream = socket_stream(sock)
+        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
@@ -214,7 +219,7 @@ function remote_build_client:sync()
         end
 
         -- sync ok
-        local msg = stream:recv_msg()
+        local msg = stream:recv_msg({timeout = -1})
         if msg and msg:success() then
             vprint(msg:body())
             ok = true
@@ -235,14 +240,14 @@ function remote_build_client:clean()
     assert(self:is_connected(), "%s: has been not connected!", self)
     local addr = self:addr()
     local port = self:port()
-    local sock = assert(socket.connect(addr, port), "%s: server unreachable!", self)
+    local sock = assert(socket.connect(addr, port, {timeout = self:connect_timeout()}), "%s: server unreachable!", self)
     local session_id = self:session_id()
     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, {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()
+        local msg = stream:recv_msg({timeout = -1})
         if msg then
             vprint(msg:body())
             if msg:success() then
@@ -264,7 +269,7 @@ function remote_build_client:runcmd(program, argv)
     assert(self:is_connected(), "%s: has been not connected!", self)
     local addr = self:addr()
     local port = self:port()
-    local sock = assert(socket.connect(addr, port), "%s: server unreachable!", self)
+    local sock = assert(socket.connect(addr, port, {timeout = self:connect_timeout()}), "%s: server unreachable!", self)
     local session_id = self:session_id()
     local errors
     local ok = false
@@ -272,7 +277,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, {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"
@@ -280,7 +285,7 @@ function remote_build_client:runcmd(program, argv)
             scheduler.co_start(self._read_stdin, self, stream, stdin_opt)
         end)
         while true do
-            local msg = stream:recv_msg()
+            local msg = stream:recv_msg({timeout = -1})
             if msg then
                 if msg:is_data() then
                     local data = stream:recv(buff, msg:body().size)
@@ -412,7 +417,7 @@ function remote_build_client:_diff_files(stream)
     local result, errors
     cprint("Comparing ${bright}%d${clear} files ..", filecount)
     if stream:send_msg(message.new_diff(session_id, manifest, {token = self:token()}), {compress = true}) and stream:flush() then
-        local msg = stream:recv_msg()
+        local msg = stream:recv_msg({timeout = -1})
         if msg and msg:success() then
             result = msg:body().manifest
             if result then
diff --git a/xmake/modules/private/service/remote_build/server.lua b/xmake/modules/private/service/remote_build/server.lua
index 31ddf4b4c..52015c7f2 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")
 
@@ -44,6 +43,10 @@ function remote_build_server:init(daemon)
 
     -- init sessions
     self._SESSIONS = {}
+
+    -- init timeout
+    self._SEND_TIMEOUT = config.get("remote_build.send_timeout") or config.get("send_timeout") or -1
+    self._RECV_TIMEOUT = config.get("remote_build.recv_timeout") or config.get("recv_timeout") or -1
 end
 
 -- get class
diff --git a/xmake/modules/private/service/remote_build/server_session.lua b/xmake/modules/private/service/remote_build/server_session.lua
index 54a39392c..b56908456 100644
--- a/xmake/modules/private/service/remote_build/server_session.lua
+++ b/xmake/modules/private/service/remote_build/server_session.lua
@@ -336,7 +336,7 @@ end
 -- recv data from stream
 function server_session:_recv_data(buff)
     local stream = self:stream()
-    local msg = stream:recv_msg()
+    local msg = stream:recv_msg({timeout = -1})
     if msg and msg:is_data() then
         return stream:recv(buff, msg:body().size)
     end
diff --git a/xmake/modules/private/service/remote_cache/client.lua b/xmake/modules/private/service/remote_cache/client.lua
index 1a46cd34f..cd3d8d7d7 100644
--- a/xmake/modules/private/service/remote_cache/client.lua
+++ b/xmake/modules/private/service/remote_cache/client.lua
@@ -58,6 +58,11 @@ function remote_cache_client:init()
     -- init sockets
     self._FREESOCKS = {}
     self._OPENSOCKS = hashset.new()
+
+    -- init timeout
+    self._SEND_TIMEOUT = config.get("remote_cache.send_timeout") or config.get("send_timeout") or -1
+    self._RECV_TIMEOUT = config.get("remote_cache.recv_timeout") or config.get("recv_timeout") or -1
+    self._CONNECT_TIMEOUT = config.get("remote_cache.connect_timeout") or config.get("connect_timeout") or -1
 end
 
 -- get class
@@ -90,13 +95,13 @@ function remote_cache_client:connect()
     -- do connect
     local addr = self:addr()
     local port = self:port()
-    local sock = assert(socket.connect(addr, port), "%s: server unreachable!", self)
+    local sock = assert(socket.connect(addr, port, {timeout = self:connect_timeout()}), "%s: server unreachable!", self)
     local session_id = self:session_id()
     local ok = false
     local errors
     print("%s: connect %s:%d ..", self, addr, port)
     if sock then
-        local stream = socket_stream(sock)
+        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
@@ -133,13 +138,13 @@ function remote_cache_client:disconnect()
     end
     local addr = self:addr()
     local port = self:port()
-    local sock = socket.connect(addr, port)
+    local sock = socket.connect(addr, port, {timeout = self:connect_timeout()})
     local session_id = self:session_id()
     local errors
     local ok = false
     print("%s: disconnect %s:%d ..", self, addr, port)
     if sock then
-        local stream = socket_stream(sock)
+        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 +186,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, {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 +223,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, {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 +258,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, {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 +290,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, {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 +331,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, {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
@@ -436,7 +441,7 @@ function remote_cache_client:_sock_open()
 
     local addr = self:addr()
     local port = self:port()
-    local sock = socket.connect(addr, port)
+    local sock = socket.connect(addr, port, {timeout = self:connect_timeout()})
     if not sock then
         self._UNREACHABLE = true
         raise("%s: server unreachable!", self)
diff --git a/xmake/modules/private/service/remote_cache/server.lua b/xmake/modules/private/service/remote_cache/server.lua
index 9ee83eb5b..defb0992e 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")
 
@@ -44,6 +43,10 @@ function remote_cache_server:init(daemon)
 
     -- init sessions
     self._SESSIONS = {}
+
+    -- init timeout
+    self._SEND_TIMEOUT = config.get("remote_cache.send_timeout") or config.get("send_timeout") or -1
+    self._RECV_TIMEOUT = config.get("remote_cache.recv_timeout") or config.get("recv_timeout") or -1
 end
 
 -- get class
diff --git a/xmake/modules/private/service/server.lua b/xmake/modules/private/service/server.lua
index 204d6e70e..891942a03 100644
--- a/xmake/modules/private/service/server.lua
+++ b/xmake/modules/private/service/server.lua
@@ -43,6 +43,10 @@ function server:init(daemon)
     -- init known hosts
     local known_hosts = config.get("known_hosts")
     self:known_hosts_set(known_hosts)
+
+    -- init timeout
+    self._SEND_TIMEOUT = config.get("send_timeout") or -1
+    self._RECV_TIMEOUT = config.get("recv_timeout") or -1
 end
 
 -- is daemon?
@@ -78,6 +82,16 @@ function server:port()
     return self._PORT
 end
 
+-- 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
 function server:tokens()
     return self._TOKENS
@@ -181,11 +195,26 @@ 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, {send_timeout = self:send_timeout(), recv_timeout = self:recv_timeout()})
     while true do
-        local msg = stream:recv_object()
+        local msg = stream:recv_object({timeout = -1})
         if msg then
-            self:_HANDLER(stream, message(msg))
+            local ok = try
+            {
+                function ()
+                    self:_HANDLER(stream, message(msg))
+                    return true
+                end,
+                catch
+                {
+                    function (errors)
+                        vprint(errors)
+                    end
+                }
+            }
+            if not ok then
+                break
+            end
         else
             break
         end
diff --git a/xmake/modules/private/service/server_config.lua b/xmake/modules/private/service/server_config.lua
index ca007e8d7..d3590aeb5 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"),
@@ -60,7 +62,6 @@ function _generate_configfile()
                 }
             }
         }
-
     }
     save(configs)
 end
diff --git a/xmake/modules/private/service/stream.lua b/xmake/modules/private/service/stream.lua
index 561fea5f3..328cd2f10 100644
--- a/xmake/modules/private/service/stream.lua
+++ b/xmake/modules/private/service/stream.lua
@@ -36,13 +36,16 @@ 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._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
@@ -50,13 +53,24 @@ function stream:sock()
     return self._SOCK
 end
 
+-- 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
-function stream:flush()
+function stream:flush(opt)
+    opt = opt or {}
     local cache = self._WCACHE
     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})
+        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
@@ -67,7 +81,8 @@ function stream:flush()
 end
 
 -- send the given bytes (small data)
-function stream:send(data, start, last)
+function stream:send(data, start, last, opt)
+    opt = opt or {}
     start = start or 1
     last = last or data:size()
     local size = last + 1 - start
@@ -93,7 +108,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 = opt.timeout or self:send_timeout()})
     if real > 0 then
         -- copy left data to cache
         assert(size <= cache_maxn)
@@ -120,11 +135,11 @@ function stream:send_object(obj, opt)
 end
 
 -- send data header
-function stream:send_header(size, flags)
+function stream:send_header(size, flags, opt)
     local buff = self._BUFF
     buff:u32be_set(1, size)
     buff:u8_set(5, flags or 0)
-    return self:send(buff, 1, 5)
+    return self:send(buff, 1, 5, opt)
 end
 
 -- send data
@@ -137,13 +152,13 @@ function stream:send_data(data, opt)
     end
     local size = data:size()
     assert(size < STREAM_DATA_MAXN, "too large data size(%d)", size)
-    if self:send_header(size, flags) then
+    if self:send_header(size, flags, opt) then
         local send = 0
         local cache = self._WCACHE
         local cache_maxn = cache:size()
         while send < size do
             local left = math.min(cache_maxn, size - send)
-            if self:send(data, send + 1, send + left) then
+            if self:send(data, send + 1, send + left, opt) then
                 send = send + left
             else
                 break
@@ -162,7 +177,7 @@ end
 
 -- send empty data
 function stream:send_emptydata(opt)
-    return self:send_header(0)
+    return self:send_header(0, opt)
 end
 
 -- send file
@@ -201,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})
+        local send = sock:sendfile(file, {block = true, timeout = opt.timeout or self:send_timeout()})
         if send > 0 then
             ok = true
         end
@@ -232,7 +247,8 @@ function stream:send_files(filepaths, opt)
 end
 
 -- recv the given bytes
-function stream:recv(buff, size)
+function stream:recv(buff, size, opt)
+    opt = opt or {}
     assert(size <= buff:size(), "too large size(%d)", size)
 
     -- read data from cache first
@@ -280,9 +296,11 @@ 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
+            local ok = sock:wait(socket.EV_RECV, opt.timeout or self:recv_timeout())
+            if ok == socket.EV_RECV then
                 wait = true
             else
+                assert(ok ~= 0, "%s: recv timeout!", self)
                 break
             end
         else
@@ -292,16 +310,16 @@ function stream:recv(buff, size)
 end
 
 -- recv message
-function stream:recv_msg()
-    local body = self:recv_object()
+function stream:recv_msg(opt)
+    local body = self:recv_object(opt)
     if body then
         return message(body)
     end
 end
 
 -- recv object
-function stream:recv_object()
-    local str = self:recv_string()
+function stream:recv_object(opt)
+    local str = self:recv_string(opt)
     if str then
         local obj, errors = str:deserialize()
         if errors then
@@ -312,8 +330,8 @@ function stream:recv_object()
 end
 
 -- recv header
-function stream:recv_header()
-    local data = self:recv(self._BUFF, 5)
+function stream:recv_header(opt)
+    local data = self:recv(self._BUFF, 5, opt)
     if data then
         local size = data:u32be(1)
         local flags = data:u8(5)
@@ -322,14 +340,14 @@ function stream:recv_header()
 end
 
 -- recv data
-function stream:recv_data()
-    local size, flags = self:recv_header()
+function stream:recv_data(opt)
+    local size, flags = self:recv_header(opt)
     if size then
         local recv = 0
         assert(size < STREAM_DATA_MAXN, "too large data size(%d)", size)
         local buff = bytes(size)
         while recv < size do
-            local data = self:recv(buff:slice(recv + 1), size - recv)
+            local data = self:recv(buff:slice(recv + 1), size - recv, opt)
             if data then
                 recv = recv + data:size()
             else
@@ -346,16 +364,16 @@ function stream:recv_data()
 end
 
 -- recv string
-function stream:recv_string()
-    local data = self:recv_data()
+function stream:recv_string(opt)
+    local data = self:recv_data(opt)
     if data then
         return data:str()
     end
 end
 
 -- recv file
-function stream:recv_file(filepath)
-    local size, flags = self:recv_header()
+function stream:recv_file(filepath, opt)
+    local size, flags = self:recv_header(opt)
     if size then
         -- empty file? we just create an empty file
         if size == 0 then
@@ -366,13 +384,13 @@ function stream:recv_file(filepath)
         local result
         local tmpfile = os.tmpfile({ramdisk = false})
         if bit.band(flags, HEADER_FLAG_COMPRESS_LZ4) == HEADER_FLAG_COMPRESS_LZ4 then
-            result = self:_recv_compressed_file(lz4.decompress_stream(), tmpfile, size)
+            result = self:_recv_compressed_file(lz4.decompress_stream(), tmpfile, size, opt)
         else
             local buff = self._BUFF
             local recv = 0
             local file = io.open(tmpfile, "wb")
             while recv < size do
-                local data = self:recv(buff, math.min(buff:size(), size - recv))
+                local data = self:recv(buff, math.min(buff:size(), size - recv), opt)
                 if data then
                     file:write(data)
                     recv = recv + data:size()
@@ -392,10 +410,10 @@ function stream:recv_file(filepath)
 end
 
 -- recv files
-function stream:recv_files(filepaths)
+function stream:recv_files(filepaths, opt)
     local size, decompressed_size
     for _, filepath in ipairs(filepaths) do
-        local real, decompressed_real = self:recv_file(filepath)
+        local real, decompressed_real = self:recv_file(filepath, opt)
         if real then
             size = (size or 0) + real
             decompressed_size = (decompressed_size or 0) + decompressed_real
@@ -407,13 +425,13 @@ function stream:recv_files(filepaths)
 end
 
 -- recv compressed file
-function stream:_recv_compressed_file(lz4_stream, filepath, size)
+function stream:_recv_compressed_file(lz4_stream, filepath, size, opt)
     local buff = self._BUFF
     local recv = 0
     local file = io.open(filepath, "wb")
     local decompressed_size = 0
     while recv < size do
-        local data = self:recv(buff, math.min(buff:size(), size - recv))
+        local data = self:recv(buff, math.min(buff:size(), size - recv), opt)
         if data then
             local write = 0
             local writesize = data:size()
@@ -442,8 +460,12 @@ function stream:_recv_compressed_file(lz4_stream, filepath, size)
     end
 end
 
-function main(sock)
+function stream:__tostring()
+    return string.format("<stream: %s>", self:sock())
+end
+
+function main(sock, opt)
     local instance = stream()
-    instance:init(sock)
+    instance:init(sock, opt)
     return instance
 end