IO refactoring, to fix a bug

[vtjnash]

while attempting to fix the #12050 and #12829 rabbit holes, I accidentally intentionally refactored the IO types and removed a lot of unused nodes in the tree. see https://github.com/JuliaLang/julia/compare/jn/io_refactor?expand=1#diff-ac5d350530574f3f82c860d1b963bc0 for a diagram of the new tree.

[tkelman]

AsyncStream is used in IJulia and a few other packages, renaming it like this without a deprecation would be breaking. Can we do just the first commit here for now?

[vtjnash]

first commit doesn't pass tests without the second. AsyncStream has always been undefined, meaningless, and badly broken, so it's hard to say what a deprecation would look like. Maybe IO, or a Union of a random subset of the Stream objects?

looking briefly at the source in IJulia, that code is rather strange and it looks like it doesn't work anymore and maybe should just be a new custom IO subtype.

[JeffBezanson]

Agree with @tkelman --- would be really nice to have just the bug fix here.

[vtjnash]

the "bug" is that Base used "AsyncStream" inconsistently and incorrectly to mean various different things

[JeffBezanson]

Why is this change here? It's not clear that Array I/O should depend on reinterpret in so many cases. This causes the array to share data which makes some operations disallowed.

[vtjnash]

this was moved from https://github.com/JuliaLang/julia/pull/12839/files#diff-ac5d350530574f3f82c860d1b963bc0aL849

[vtjnash]

how about unsafe_read!(s::IO, p::Ptr, nb::UInt)?

[vtjnash]

also, this method was never used since it is shadowed by re-implementations on all the subtypes used in the testing https://codecov.io/github/JuliaLang/julia/base/io.jl?ref=e82ba0cea7a131b06716d81fbcbf0cfd5feb4f59#l-134

[JeffBezanson]

The following relatively small patch against the first commit gets the spawn test to pass. All it does is fix some mistaken method defs and typos.

diff --git a/base/io.jl b/base/io.jl
index 3d1905b..1203f12 100644
--- a/base/io.jl
+++ b/base/io.jl
@@ -303,3 +303,8 @@ function reset{T<:IO}(io::T)
 end

 ismarked(io::IO) = io.mark >= 0
+
+# Generic IO stubs
+
+lock(::IO) = nothing
+unlock(::IO) = nothing
diff --git a/base/stream.jl b/base/stream.jl
index 8d14f2f..0583b63 100644
--- a/base/stream.jl
+++ b/base/stream.jl
@@ -235,11 +235,11 @@ show(io::IO,stream::TTY) = print(io,"TTY(",uv_status_string(stream),", ",
     nb_available(stream.buffer)," bytes waiting)")

 function println(io::AsyncStream, xs...)
-    lock(io.lock)
+    lock(io)
     try
         invoke(println, Tuple{IO, map(typeof,xs)...}, io, xs...)
     finally
-        unlock(io.lock)
+        unlock(io)
     end
 end

@@ -543,20 +543,24 @@ show(io::IO,stream::Pipe) = print(io,
     uv_status_string(stream.in), " => ",
     uv_status_string(stream.out), ", ",
     nb_available(stream), " bytes waiting)")
-isreadable(io::AbstractPipe) = isreadable(io.out)
-iswritable(io::AbstractPipe) = iswritable(io.in)
-read{T<:AbstractPipe}(io::T, args...) = read(io.out, args...)
+write(io::AbstractPipe, byte::UInt8) = write(io.in, byte)
+write(io::AbstractPipe, bytes::Vector{UInt8}) = write(io.in, bytes)
 write{T<:AbstractPipe}(io::T, args...) = write(io.in, args...)
 write{S<:AbstractPipe,T}(io::S, a::Array{T}) = write(io.in, a)
 buffer_or_write(io::AbstractPipe, p::Ptr, n::Integer) = buffer_or_write(io.in, p, n)
-println{T<:AbstractPipe}(io::T, args...) = println(io.in, args...)
-flush(io::AbstractPipe) = flush(io.in)
 buffer_writes(io::AbstractPipe, args...) = buffer_writes(io.in, args...)
-read{T<:AbstractPipe}(io::AbstractPipe, args...) = read(io.out, args)
-readuntil{T<:AbstractPipe}(io::T, args...) = readuntil(io.out, args...)
+flush(io::AbstractPipe) = flush(io.in)
+
+read(io::AbstractPipe, byte::Type{UInt8}) = read(io.out, byte)
+read!(io::AbstractPipe, bytes::Vector{UInt8}) = read!(io.out, bytes)
+read{T<:AbstractPipe}(io::T, args...) = read(io.out, args...)
 read!{T<:AbstractPipe}(io::T, args...) = read!(io.out, args...)
+readuntil{T<:AbstractPipe}(io::T, args...) = readuntil(io.out, args...)
 readbytes(io::AbstractPipe) = readbytes(io.out)
 readavailable(io::AbstractPipe) = readavailable(io.out)
+
+isreadable(io::AbstractPipe) = isreadable(io.out)
+iswritable(io::AbstractPipe) = iswritable(io.in)
 isopen(io::AbstractPipe) = isopen(io.in) || isopen(io.out)
 close(io::AbstractPipe) = (close(io.in); close(io.out))
 wait_readnb(io::AbstractPipe, nb::Int) = wait_readnb(io.out, nb)
@@ -1167,6 +1171,8 @@ type BufferStream <: AsyncStream
     BufferStream() = new(PipeBuffer(), Condition(), Condition(), true, false, ReentrantLock())
 end

+lock(s::BufferStream) = lock(s.lock)
+unlock(s::BufferStream) = unlock(s.unlock)
 isopen(s::BufferStream) = s.is_open
 close(s::BufferStream) = (s.is_open = false; notify(s.r_c; all=true); notify(s.close_c; all=true); nothing)

diff --git a/test/spawn.jl b/test/spawn.jl
index 7b20d58..c6feee0 100644
--- a/test/spawn.jl
+++ b/test/spawn.jl
@@ -248,10 +248,10 @@ let bad = "bad\0name"
     @test_throws ArgumentError run(setenv(`echo hello`, "good"=>bad))
 end

-let out = Pipe()
+let out = Pipe(), echo = `$exename -f -e 'print(STDOUT, " 1\t", readall(STDIN))'`
     @test_throws ArgumentError write(out, "not open error")
-    open(`cat -n`, "w", out) do in1
-        open(`cat -n`, "w", out) do in2
+    open(echo, "w", out) do in1
+        open(echo, "w", out) do in2
             write(in1, 'h')
             write(in2, UInt8['w'])
             println(in1, "ello")
@@ -267,19 +267,18 @@ let out = Pipe()
     @test !iswritable(out)
     @test isopen(out)
     @test endswith(readuntil(out, '1'), '1')
-    @test read(out, UInt8) == '\n'
+    @test read(out, UInt8) == '\t'
     c = UInt8[0]
     @test c == read!(out, c)
-    @test nb_availble(out) == 0
-    wait_readnb(out, 1)
-    @test nb_availble(out) > 0
+    Base.wait_readnb(out, 1)
+    @test nb_available(out) > 0
     ln1 = readline(out)
     ln2 = readline(out)
     desc = readall(out)
     @test !isreadable(out)
     @test !iswritable(out)
     @test !isopen(out)
-    @test nb_availble(out) == 0
+    @test nb_available(out) == 0
     @test c == ['w']
     @test lstrip(ln2) == "1\thello\n"
     @test ln1 == "orld\n"

[vtjnash]

yes, i realize my test coverage of this change (especially the type checking and extensibility aspects) are not fully covered by the tests I added. if you want to skip the rest of the renaming in this commit (of AsyncStream -> LibuvStream), it would be nice if you could also write your own tests to show that it is covering other possible interactions.

[JeffBezanson]

You're moving the goalposts. Earlier, the story was that the second commit is needed to pass tests. If there are some additional tests that are also good to have, by all means let's write them, but it still seems to me the bug can be fixed more surgically.

[JeffBezanson]

Ok, I have a candidate smaller fix in #12858.

I do think this is a good redesign, and I agree the AsyncStream type is a problem. I just want to stage this differently.

[vtjnash]

rebased (i think), and ready to merge (as soon as CI passes)

[StefanKarpinski]

Is this non-breaking at this point?

[vtjnash]

It was always nonbreaking

[tkelman]

It initially just got rid of AsyncStream completely but at least now it has a (no-warning) deprecation for it. The deprecations should demarcate 0.4 vs 0.5 though.

[kshyatt]

Next time we do something like this (deprecate a type used all over the parallel code) we should probably update the docs within a month or two rather than 11...