replay_something

FakePTYs.jl

# This file is a part of Julia. License is MIT: https://julialang.org/license

module FakePTYs

if Sys.iswindows()
    pushfirst!(LOAD_PATH, Sys.STDLIB)
    using Sockets
    popfirst!(LOAD_PATH)
end


function open_fake_pty()
    @static if Sys.iswindows()
        # Fake being cygwin
        pid = string(getpid(), base=16, pad=16)
        pipename = """\\\\?\\pipe\\cygwin-$pid-pty10-abcdefg"""
        server = listen(pipename)
        pts = connect(pipename)
        @assert ccall(:jl_ispty, Cint, (Ptr{Cvoid},), pts.handle) == 1
        ptm = accept(server)
        close(server)
        # extract just the file descriptor
        fds = Libc.dup(Base._fd(pts))
        close(pts)
        pts = fds
        # convert pts handle to a TTY
        #fds = pts.handle
        #pts.status = Base.StatusClosed
        #pts.handle = C_NULL
        #pts = Base.TTY(fds, Base.StatusOpen)
    else
        O_RDWR = Base.Filesystem.JL_O_RDWR
        O_NOCTTY = Base.Filesystem.JL_O_NOCTTY

        fdm = ccall(:posix_openpt, Cint, (Cint,), O_RDWR | O_NOCTTY)
        fdm == -1 && error("Failed to open ptm")
        rc = ccall(:grantpt, Cint, (Cint,), fdm)
        rc != 0 && error("grantpt failed")
        rc = ccall(:unlockpt, Cint, (Cint,), fdm)
        rc != 0 && error("unlockpt")

        fds = ccall(:open, Cint, (Ptr{UInt8}, Cint),
            ccall(:ptsname, Ptr{UInt8}, (Cint,), fdm), O_RDWR | O_NOCTTY)

            pts = RawFD(fds)
        # pts = fdio(fds, true)
        # pts = Base.Filesystem.File(RawFD(fds))
        # pts = Base.TTY(RawFD(fds); readable = false)
        ptm = Base.TTY(RawFD(fdm))
    end
    return pts, ptm
end

function with_fake_pty(f)
    pts, ptm = open_fake_pty()
    try
        f(pts, ptm)
    finally
        close(ptm)
    end
    nothing
end

end

replay.jl

# This file is a part of Julia. License is MIT: https://julialang.org/license

include("FakePTYs.jl")
import .FakePTYs: open_fake_pty

CTRL_C = '\x03'
UP_ARROW = "\e[A"
DOWN_ARROW = "\e[B"
RIGHT_ARROW = "\e[C"
LEFT_ARROW = "\e[D"

repl_script = """
2+2
print("")
display([1])
display([1 2; 3 4])
@time 1+1
; pwd
$CTRL_C
xxxx = 3$LEFT_ARROW$LEFT_ARROW$(LEFT_ARROW)$(LEFT_ARROW)y
? reinterpret
using Ra\t$CTRL_C
\\alpha\t$CTRL_C
\e[200~paste here ;)\e[201~"$CTRL_C
$UP_ARROW$DOWN_ARROW$CTRL_C
123\b\b\b$CTRL_C
\b\b$CTRL_C
f(x) = x03
f(1,2)
[][1]
cd("complet_path\t\t$CTRL_C
st
$CTRL_C
\n
"""

julia_exepath() = joinpath(Sys.BINDIR::String, Base.julia_exename())

function generate_precompile_statements()
    start_time = time_ns()
    debug_output = stdout
    sysimg = Base.unsafe_string(Base.JLOptions().image_file)

    mktemp() do _, _
        # Collect statements from running a REPL process and replaying our REPL script
        pts, ptm = open_fake_pty()
        blackhole = Sys.isunix() ? "/dev/null" : "nul"
        p = withenv("JULIA_HISTORY" => blackhole,
            "JULIA_PROJECT" => nothing, # remove from environment
            "JULIA_LOAD_PATH" => Sys.iswindows() ? "@;@stdlib" : "@:@stdlib",
            "JULIA_PKG_PRECOMPILE_AUTO" => "0",
            "TERM" => "") do
            run(```$(julia_exepath()) -O0
                   --cpu-target=native --startup-file=no --color=yes
                   -e 'import REPL; REPL.Terminals.is_precompiling[] = true'
                   -i ```,
                pts, pts, pts; wait = false)
        end
        Base.close_stdio(pts)
        # Prepare a background process to copy output from process until `pts` is closed
        output_copy = Base.BufferStream()
        tee = @async try
            while !eof(ptm)
                l = readavailable(ptm)
                write(debug_output, l)
                Sys.iswindows() && (sleep(0.1); yield(); yield()) # workaround hang - probably a libuv issue?
                write(output_copy, l)
            end
            close(output_copy)
            close(ptm)
        catch ex
            close(output_copy)
            close(ptm)
            if !(ex isa Base.IOError && ex.code == Base.UV_EIO)
                rethrow() # ignore EIO on ptm after pts dies
            end
        end
        # wait for the definitive prompt before start writing to the TTY
        readuntil(output_copy, "julia>")
        sleep(0.1)
        readavailable(output_copy)
        # Input our script
        if true
            precompile_lines = split(repl_script::String, '\n'; keepempty = false)
            curr = 0
            for l in precompile_lines
                sleep(0.5)
                #=
                curr += 1
                print("\rGenerating REPL precompile statements... $curr/$(length(precompile_lines))")
                =#
                # consume any other output
                bytesavailable(output_copy) > 0 && readavailable(output_copy)
                # push our input
                #=
                write(debug_output, "\n#### inputting statement: ####\n$(repr(l))\n####\n")
                =#
                if endswith(l, "\x03")
                    write(ptm, l)
                else 
                    write(ptm, l, "\n")
                end
                readuntil(output_copy, "\n")
                # wait for the next prompt-like to appear
                # NOTE: this is rather inaccurate because the Pkg REPL mode is a special flower
                readuntil(output_copy, "\n")
                readuntil(output_copy, "> ")
            end
            println()
        end
        write(ptm, "exit()\n")
        wait(tee)
        success(p) || Base.pipeline_error(p)
        close(ptm)
    end
end

generate_precompile_statements()