Julia IO&进程通信：read「run(`cmd`)」后主进程阻塞/卡死，read后进程kill/close无效

[Argon]

背景&目标

开发一个程序交互接口：

1. 可使用Julia异步、后台启动一个命令/exe文件
2. 支持在运行时向程序异步写入命令、读取输出（文本IO）
3. 支持调用函数关闭程序，关闭程序进程
4. 支持对多个进程的并行、异步调度（e.g. 同时启动多个cmd shell，用户可以指定「向哪一个cmd shell读写输入/输出」）

问题

在「读取输出」的过程中，遇到「read没有输出导致主进程死循环阻塞」的问题

• 函数read/readline/readlines均无法识别「输出是否完整」
• 主进程阻塞后，除非使用Ctrl+C强行停止，代码无法自行执行完毕，后续对程序的IO操作无法继续执行

MWE

proc = open(`cmd`, write=true, read=true) # 后台异步打开一个cmd进程

println(proc, "echo 1") # 使用println输入指令

@show readline(proc) # 输出「"readline(proc) = 【cmd路径】> echo 1"」
@show readline(proc) # 输出「"readline(proc) = 1"」
@show readline(proc) # REPL/Debugger/.jl脚本 无响应：主进程阻塞，程序卡死

# 或：使用循环从指令中获取输入（及其应答）
for l in readlines(proc)
       @show l
       end
# 此后主进程阻塞，程序卡死 #

预期：执行前eof(proc) == true、执行时报错 或 执行后返回空值

实际：主进程阻塞，无法继续执行代码

在REPL运行上述代码直至卡顿后，使用Ctrl+C强制关闭的报错信息：

ERROR: InterruptException:
Stacktrace:
  [1] poptask(W::Base.IntrusiveLinkedListSynchronized{Task})
    @ Base .\task.jl:974
  [2] wait()
    @ Base .\task.jl:983
  [3] wait(c::Base.GenericCondition{Base.Threads.SpinLock}; first::Bool)
    @ Base .\condition.jl:130
  [4] wait
    @ .\condition.jl:125 [inlined]
  [5] readuntil(x::Base.PipeEndpoint, c::UInt8; keep::Bool)
    @ Base .\stream.jl:1014
  [6] readuntil(io::Base.Process, arg::UInt8; kw::Base.Pairs{Symbol, Bool, Tuple{Symbol}, NamedTuple{(:keep,), Tuple{Bool}}})
    @ Base .\io.jl:442
  [7] readuntil
    @ .\io.jl:442 [inlined]
  [8] readline(s::Base.Process; keep::Bool)
    @ Base .\io.jl:548
  [9] readline(s::Base.Process)
    @ Base .\io.jl:547
 [10] top-level scope
    @ REPL[28]:1
=#

已参考过的问题/话题

• 【中】请问Julia如何与外部程序互传数据？不是读写文件的那种
  • 使用"r+"与设置「write=true, read=true」等价：无法解决问题
• 【英】Run commands on externally running process - General Usage - Julia Programming Language — 在外部运行进程上运行命令
  • 运行read(pstdout)仍会遇到上述问题
• 【英】Processing output from long running process - New to Julia - Julia Programming Language — 处理长时间运行的进程的输出
  • 遍历eachline仍会遇到上述问题
• 【英】 How to continuously communicate with an external program? - General Usage - Julia Programming Language — 如何与外部程序持续通信？
  • 命令加上前缀stdbuf -oL 仍会遇到上述问题

[xgdgsc]

In Julia, how do I run an external program and process its output line by line?

你for前面为什么不异步

[Argon]

问题已解决，感谢你的提醒！

现在我把 readline 放入 while 循环中，并使用 @async 宏异步调用一个「钩子函数」，成功实现了预期目标。

下面是一个可以运作的代码示例：

"处理输出的函数（示例）"
function handle(message::String)::Nothing
    "Message '$message' handled!" |> println
end

"跟踪进程的钩子"
function hook(process::Base.Process)::Nothing
    while true # 这里的「true」可以换成其它「判断进程存活」的条件，但要响应还需要再println输入消息（使其在阻塞状态下恢复）
        process |> readline |> handle # 这里经测试用「read」无效（没有handle到数据）
    end
end

proc::Base.Process = open(`cmd`, "r+") # 正常以「读写」（r+、w+均可）权限打开程序（cmd换成powershell亦可）

@async hook(proc) # 【关键】使用@async异步宏读取程序，让其与主进程并行运行

# 异步读写示例

sleep(1) # 等待cmd加载：（输出）handle【初始化信息】


println(proc, "@echo off") # 写入示例：使用「println」（而非write）「打印」输入到进程（这一行powershell不适用）
println(proc, "echo 1") # 写入示例2：让程序echo「1」

sleep(1) # 程序响应输入：（输出）handle"1"，将打印「Message '1' handled!」

[Argon]

目前还需解决的问题：外部程序进程终止

针对cmd、Powershell外的「外部程序」而言，使用 open 打开为 Base.Process 后，暂时没有找到「在Julia主程序运行周期内结束外部程序」的方法：

• close(proc) 关闭无效：执行后阻塞主进程
• kill(proc) 关闭无效：在REPL窗口上方显示程序仍在运行

MWE

1. 以 cmd.exe 为例

Julia代码：

cmd = open(`cmd`, "w+") # 打开cmd

println(cmd, "echo 1") # 输入指令

@show readline(cmd)
@show readline(cmd)
@show readline(cmd)
@show readline(cmd) # echo 1
@show readline(cmd) # 1（读到这里停止，否则会堵塞主程序）

@show cmd kill(cmd) cmd close(cmd) cmd # 尝试kill&close
@show eof(cmd) cmd.in cmd.out

while !eof(cmd) # 一直读取到文件结束
    @show readline(cmd)
end
"eof!" |> println

@show cmd cmd.in cmd.out

预期：Process(`cmd`, ProcessExited(0))（程序退出）
实际：Process(`cmd`, ProcessSignaled(15))（未完全终止）

• 但 cmd.in、cmd.out 显示为close

指示信息：

cmd = Process(`cmd`, ProcessSignaled(15))
cmd.in = Base.PipeEndpoint(Base.Libc.WindowsRawSocket(0xffffffffffffffff) closed, 0 bytes waiting)
cmd.out = Base.PipeEndpoint(Base.Libc.WindowsRawSocket(0xffffffffffffffff) closed, 27 bytes waiting)

2. 读写外部程序的例子

示例程序：使用PyInstaller打包如下代码，生成 test.exe：

while True:
    print(input("input:"))

使用Julia读写该exe（假设在当前目录下）

function hook(process::Base.Process)::Nothing
    while !eof(process)
        message = process |> readline
        !isempty(message) && "Message '$message' handled!" |> println
    end
end

path = "test.exe"
test::Base.Process = open(`$path`, "r+")
@async hook(test)


sleep(1)

@show println(test.in, "My-Input-0")
@show println(test.in, "My-Input-1")
@show println(test.in, "My-Input-2")

sleep(1)

@show test test.in test.out

# 尝试用kill、close杀死进程
@show kill(test) close(test)

sleep(1)

@show test test.in test.out

输出：

test = Process(`'test.exe'`, ProcessRunning)
test.in = Base.PipeEndpoint(Base.Libc.WindowsRawSocket(0x00000000000002d4) open, 0 bytes waiting)
test.out = Base.PipeEndpoint(Base.Libc.WindowsRawSocket(0x00000000000002c8) open, 0 bytes waiting)
println(test.in, "My-Input-0") = nothing
println(test.in, "My-Input-1") = nothing
println(test.in, "My-Input-2") = nothing
Message 'input:My-Input-0' handled!
Message 'input:My-Input-1' handled!
Message 'input:My-Input-2' handled!
test = Process(`'test.exe'`, ProcessRunning)
test.in = Base.PipeEndpoint(Base.Libc.WindowsRawSocket(0x00000000000002d4) open, 0 bytes waiting)
test.out = Base.PipeEndpoint(Base.Libc.WindowsRawSocket(0x00000000000002c8) active, 6 bytes waiting)
test = Process(`'test.exe'`, ProcessSignaled(15))
test.in = Base.PipeEndpoint(Base.Libc.WindowsRawSocket(0x00000000000002d4) open, 0 bytes waiting)
test.out = Base.PipeEndpoint(Base.Libc.WindowsRawSocket(0x00000000000002c8) active, 6 bytes waiting)
kill(test) = nothing
Traceback (most recent call last):
  File "test.py", line 2, in <module>
close(test) = nothing
EOFError: EOF when reading a line
[18312] Failed to execute script 'test' due to unhandled exception!
【代码正常结束】

预期：Process(`test.exe`, ProcessExited(0))（程序退出）
实际：Process(`test.exe`, ProcessSignaled(15))（未完全终止）

• 并且 test.in、test.out也还是正常激活状态

[xgdgsc]

EOFError catch了exit行吗,while后面加 process_running(process) 呢

[Argon]

现在我将示例用的Python程序改为以下代码：

try:
    while True:
        print(input("input:"))
except EOFError:
    print("EOF! exiting...")
except BaseException as e:
    print(f"Error! {e}")
    exit()

并在同一目录下运行如下Julia代码（while 后的条件改为 !eof(process) && !process_exited(process) && process_running(process) ）：

function hook(process::Base.Process)::Nothing
    while !eof(process) && !process_exited(process) && process_running(process)
        message = process |> readline
        !isempty(message) && "Message '$message' handled!" |> println
    end
    println("async hook exited!")
end

path = raw"【这里替换为exe文件的父路径】\test.exe"

test::Base.Process = open(`$path`, "r+")

@show test test.in test.out

@async hook(test)

@show println(test.in, "My-Input-0")
@show println(test.in, "My-Input-1")
@show println(test.in, "My-Input-2")

sleep(1)

@show test test.in test.out

@show kill(test) close(test)

sleep(1)

@show test test.in test.out process_exited(test) process_running(test)

运行输出如下：

test = Process(`'********\test.exe'`, ProcessRunning)
test.in = Base.PipeEndpoint(Base.Libc.WindowsRawSocket(0x0000000000000300) open, 0 bytes waiting)
test.out = Base.PipeEndpoint(Base.Libc.WindowsRawSocket(0x00000000000002dc) open, 0 bytes waiting)
println(test.in, "My-Input-0") = nothing
println(test.in, "My-Input-1") = nothing
println(test.in, "My-Input-2") = nothing
Message 'input:My-Input-0' handled!
Message 'input:My-Input-1' handled!
Message 'input:My-Input-2' handled!
test = Process(`'********\test.exe'`, ProcessRunning)
test.in = Base.PipeEndpoint(Base.Libc.WindowsRawSocket(0x0000000000000300) open, 0 bytes waiting)
test.out = Base.PipeEndpoint(Base.Libc.WindowsRawSocket(0x00000000000002dc) active, 6 bytes waiting)
kill(test) = nothing
Message 'input:' handled!
Traceback (most recent call last):
async hook exited!  File "<string>", line 1, in <module>

OSError: [Errno 22] Invalid argument
close(test) = nothing
Exception ignored in: <_io.TextIOWrapper name='<stdout>' mode='w' encoding='gbk'>
OSError: [Errno 22] Invalid argument
【Julia程序退出】

可以看到，while 循环是正确退出的，但进程（Base.Process 对象）test依旧没有正常关闭（test = Process(`'********\test.exe'`, ProcessRunning)，依旧处于ProcessRunning状态）

所以，这样做仍然无法在Julia中做到对外部程序的完整进程管理，仍有可能在Julia程序运行后残留进程。

[xgdgsc]

我直接复制你的改了python路径好像运行是正常的? 你在非Windows系统试过没,比如WSL. 我在windows下测试也没问题

test = Process(`python r.py`, ProcessRunning)
test.in = Base.PipeEndpoint(RawFD(19) open, 0 bytes waiting)
test.out = Base.PipeEndpoint(RawFD(20) open, 0 bytes waiting)
println(test.in, "My-Input-0") = nothing
Message 'input:My-Input-0' handled!
println(test.in, "My-Input-1") = nothing
Message 'input:My-Input-1' handled!
println(test.in, "My-Input-2") = nothing
Message 'input:My-Input-2' handled!
test = Process(`python r.py`, ProcessRunning)
test.in = Base.PipeEndpoint(RawFD(19) open, 0 bytes waiting)
test.out = Base.PipeEndpoint(RawFD(20) active, 6 bytes waiting)
kill(test) = nothing
Message 'input:' handled!
async hook exited!
close(test) = nothing
test = Process(`python r.py`, ProcessSignaled(15))
test.in = Base.PipeEndpoint(RawFD(4294967295) closed, 0 bytes waiting)
test.out = Base.PipeEndpoint(RawFD(4294967295) closed, 0 bytes waiting)
process_exited(test) = true
process_running(test) = false

[Argon]

目前没在Windows以外的平台上测试过

现在测试上述代码，除了Process还是「ProcessSignaled(15)」外 ，已经没有问题了

使用 julia 源码.jl 启动，源码同#6，输出如下：

test = Process(`'test.exe'`, ProcessRunning)
test.in = Base.PipeEndpoint(Base.Libc.WindowsRawSocket(0x0000000000000324) open, 0 bytes waiting)
test.out = Base.PipeEndpoint(Base.Libc.WindowsRawSocket(0x0000000000000328) open, 0 bytes waiting)
println(test.in, "My-Input-0") = nothing
println(test.in, "My-Input-1") = nothing
println(test.in, "My-Input-2") = nothing
Message 'input:My-Input-0' handled!
Message 'input:My-Input-1' handled!
Message 'input:My-Input-2' handled!
test = Process(`'test.exe'`, ProcessRunning)
test.in = Base.PipeEndpoint(Base.Libc.WindowsRawSocket(0x0000000000000324) open, 0 bytes waiting)
test.out = Base.PipeEndpoint(Base.Libc.WindowsRawSocket(0x0000000000000328) active, 6 bytes waiting)
kill(test) = nothing
Message 'input:' handled!Traceback (most recent call last):

  File "<string>", line 1, in <module>
async hook exited!OSError: [Errno 22] Invalid argument

Exception ignored in: <_io.TextIOWrapper name='<stdout>' mode='w' encoding='gbk'>
close(test) = OSError: [Errno 22] Invalid argument
nothing
test = Process(`'test.exe'`, ProcessSignaled(15))
test.in = Base.PipeEndpoint(Base.Libc.WindowsRawSocket(0xffffffffffffffff) closed, 0 bytes waiting)
test.out = Base.PipeEndpoint(Base.Libc.WindowsRawSocket(0xffffffffffffffff) closed, 0 bytes waiting)
process_exited(test) = true
process_running(test) = false

若将源码中 test::Base.Process = open(`$path`, "r+") 换成

1. test::Base.Process = open(`python test.py`, "r+")
2. test::Base.Process = open(`powershell`, "r+") ; println(test.in, "python test.py") ; sleep(1)（加 sleep 以待powershell启动）
3. test::Base.Process = open(`powershell`, "r+") ; println(test.in, "$path") ; sleep(1)（加 sleep 以待powershell启动）

输出均一致。

即下列打开方式效果等同：

1. 直接用 Base.Cmd 调用程序
2. 直接用终端运行程序源码
3. 先用 open 打开命令行（cmd/powershell），再在命令行中运行程序/用终端运行程序源码