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

背景&目标

开发一个程序交互接口:

  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
=#

已参考过的问题/话题

你for前面为什么不异步

:white_check_mark:问题已解决,感谢你的提醒!

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

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

"处理输出的函数(示例)"
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!」

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

针对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.incmd.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.intest.out也还是正常激活状态

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

现在我将示例用的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程序运行后残留进程。

我直接复制你的改了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

目前没在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),再在命令行中运行程序/用终端运行程序源码