improve pipe mode

8 files changed, 41 insertions, 12 deletions

diff --git a/core/src/xmake/io/pipe_open.c b/core/src/xmake/io/pipe_open.c
index e40451641..599c3f53d 100644
--- a/core/src/xmake/io/pipe_open.c
+++ b/core/src/xmake/io/pipe_open.c
@@ -51,6 +51,9 @@ tb_int_t xm_io_pipe_open(lua_State* lua)
     tb_size_t mode = TB_PIPE_MODE_RO;
     if (modestr[0] == 'w') mode = TB_PIPE_MODE_WO;
 
+    // set block mode
+    if (modestr[1] == 'B') mode |= TB_PIPE_MODE_BLOCK;
+
     // get buffer size
     tb_size_t buffsize = (tb_size_t)luaL_checknumber(lua, 3);
 
diff --git a/core/src/xmake/io/pipe_openpair.c b/core/src/xmake/io/pipe_openpair.c
index 3e76c94e7..cb7d2c761 100644
--- a/core/src/xmake/io/pipe_openpair.c
+++ b/core/src/xmake/io/pipe_openpair.c
@@ -35,19 +35,28 @@
  */
 
 /*
- * io.pipe_openpair(buffsize)
+ * io.pipe_openpair(mode, buffsize)
  */
 tb_int_t xm_io_pipe_openpair(lua_State* lua)
 {
     // check
     tb_assert_and_check_return_val(lua, 0);
 
+    // get pipe mode
+    tb_char_t const* modestr = luaL_optstring(lua, 1, "AA");
+    tb_assert_and_check_return_val(modestr, 0);
+
+    // init mode
+    tb_size_t mode[2] = {0};
+    if (modestr[0] == 'B') mode[0] |= TB_PIPE_MODE_BLOCK;
+    if (modestr[1] == 'B') mode[1] |= TB_PIPE_MODE_BLOCK;
+
     // get buffer size
-    tb_size_t buffsize = (tb_size_t)luaL_checknumber(lua, 1);
+    tb_size_t buffsize = (tb_size_t)luaL_checknumber(lua, 2);
 
     // init pipe
     tb_pipe_file_ref_t pipefile[2];
-    if (tb_pipe_file_init_pair(pipefile, tb_null, buffsize))
+    if (tb_pipe_file_init_pair(pipefile, mode, buffsize))
     {
         xm_lua_pushpointer(lua, (tb_pointer_t)pipefile[0]);
         xm_lua_pushpointer(lua, (tb_pointer_t)pipefile[1]);
diff --git a/tests/modules/pipe/pipe_pair.lua b/tests/modules/pipe/pipe_pair.lua
index f6b2cb5b1..da38f81ff 100644
--- a/tests/modules/pipe/pipe_pair.lua
+++ b/tests/modules/pipe/pipe_pair.lua
@@ -3,7 +3,7 @@ import("core.base.bytes")
 
 function main()
     local buff = bytes(8192)
-    local rpipe, wpipe = pipe.openpair(4096)
+    local rpipe, wpipe = pipe.openpair()
     wpipe:write("hello xmake!", {block = true})
     local read, data = rpipe:read(buff, 13)
     if read > 0 and data then
diff --git a/tests/modules/pipe/sched_pipe_pair.lua b/tests/modules/pipe/sched_pipe_pair.lua
index be64d0011..840df0d2b 100644
--- a/tests/modules/pipe/sched_pipe_pair.lua
+++ b/tests/modules/pipe/sched_pipe_pair.lua
@@ -26,7 +26,7 @@ function _session_write(id, pipefile)
 end
 
 function _session(id)
-    local rpipe, wpipe = pipe.openpair(256)
+    local rpipe, wpipe = pipe.openpair()
     scheduler.co_start(_session_read, id, rpipe)
     scheduler.co_start(_session_write, id, wpipe)
 end
diff --git a/tests/modules/process/sched_process_pipe.lua b/tests/modules/process/sched_process_pipe.lua
index d399c3a3b..eb36c6c20 100644
--- a/tests/modules/process/sched_process_pipe.lua
+++ b/tests/modules/process/sched_process_pipe.lua
@@ -29,7 +29,7 @@ function _session_read_pipe(id, rpipeopt)
 end
 
 function _session(id, program, ...)
-    local rpipe, wpipe = pipe.openpair(10)
+    local rpipe, wpipe = pipe.openpair()
     local rpipeopt = {rpipe = rpipe, stop = false}
     scheduler.co_start(_session_read_pipe, id, rpipeopt)
     local proc = process.openv(program, table.pack(...), {stdout = wpipe})
diff --git a/xmake/core/base/pipe.lua b/xmake/core/base/pipe.lua
index affade303..d3a5e3a87 100644
--- a/xmake/core/base/pipe.lua
+++ b/xmake/core/base/pipe.lua
@@ -305,6 +305,8 @@ end
 --    pipe:close()
 -- end
 --
+-- mode: "r", "w", "rB" (block), "wB" (block), "rA" (non-block), "wA" (non-block)
+--
 function pipe.open(name, mode, buffsize)
 
     -- open named pipe
@@ -322,10 +324,17 @@ end
 -- rpipe:read(...)
 -- wpipe:write(...)
 --
-function pipe.openpair(buffsize)
+-- mode:
+--
+-- "BB": read block/write block
+-- "BA": read block/write non-block
+-- "AB": read non-block/write block
+-- "AA": read non-block/write non-block (default)
+--
+function pipe.openpair(mode, buffsize)
 
     -- open anonymous pipe pair
-    local rpipefile, wpipefile, errors = io.pipe_openpair(buffsize or 0)
+    local rpipefile, wpipefile, errors = io.pipe_openpair(mode, buffsize or 0)
     if rpipefile and wpipefile then
         return _instance.new(rpipefile), _instance.new(wpipefile)
     else
diff --git a/xmake/core/sandbox/modules/import/core/base/pipe.lua b/xmake/core/sandbox/modules/import/core/base/pipe.lua
index b3603d0ad..1f7e12a30 100644
--- a/xmake/core/sandbox/modules/import/core/base/pipe.lua
+++ b/xmake/core/sandbox/modules/import/core/base/pipe.lua
@@ -104,8 +104,8 @@ function sandbox_core_base_pipe.open(name, mode, buffsize)
 end
 
 -- open a anonymous pipe pair
-function sandbox_core_base_pipe.openpair(buffsize)
-    local rpipefile, wpipefile, errors = pipe.openpair(buffsize)
+function sandbox_core_base_pipe.openpair(mode, buffsize)
+    local rpipefile, wpipefile, errors = pipe.openpair(mode, buffsize)
     if not rpipefile or not wpipefile then
         raise(errors)
     end
diff --git a/xmake/modules/private/service/remote_build/session.lua b/xmake/modules/private/service/remote_build/session.lua
index 04f053e68..b78ddf8c4 100644
--- a/xmake/modules/private/service/remote_build/session.lua
+++ b/xmake/modules/private/service/remote_build/session.lua
@@ -91,13 +91,21 @@ function session:runcmd(respmsg)
     local program = body.program
     local argv = body.argv
     vprint("%s: run command(%s) ..", self, os.args(table.join(program, argv)))
-    local stdin_rpipe, stdin_wpipe = pipe.openpair(256)
+
+    -- init pipes
+    local stdin_rpipe, stdin_wpipe = pipe.openpair("BA") -- rpipe (block)
     local stdin_wpipeopt = {wpipe = stdin_wpipe, stop = false}
-    local stdout_rpipe, stdout_wpipe = pipe.openpair(256)
+    local stdout_rpipe, stdout_wpipe = pipe.openpair()
     local stdout_rpipeopt = {rpipe = stdout_rpipe, stop = false}
+
+    -- read and write pipe
     scheduler.co_start(self._write_pipe, self, stdin_wpipeopt)
     scheduler.co_start(self._read_pipe, self, stdout_rpipeopt)
+
+    -- run program
     os.execv(program, argv, {curdir = self:sourcedir(), stdout = stdout_wpipe, stdin = stdin_rpipe})
+
+    -- stop it
     stdin_wpipeopt.stop = true
     stdin_wpipe:close()
     stdout_rpipeopt.stop = true