make start_reading/stop_reading automatic exactly when needed. fix #1925, fix #10655 (closes #11530)

this also makes the read throttle more intelligent so that it doesn't
get tripped up by wait_nb requests for more than READ_BUFFER_SZ bytes

Author: vtjnash

base/LineEdit.jl

@@ -1569,7 +1569,6 @@ function prompt!(term, prompt, s = init_state(term, prompt))
     raw!(term, true)
     enable_bracketed_paste(term)
     try
-        Base.start_reading(term)
         activate(prompt, s, term)
         while true
             map = keymap(s, prompt)
@@ -1585,14 +1584,11 @@ function prompt!(term, prompt, s = init_state(term, prompt))
                 state = :done
             end
             if state == :abort
-                Base.stop_reading(term)
                 return buffer(s), false, false
             elseif state == :done
-                Base.stop_reading(term)
                 return buffer(s), true, false
             elseif state == :suspend
                 @unix_only begin
-                    Base.stop_reading(term)
                     return buffer(s), true, true
                 end
             else

base/REPL.jl

@@ -800,11 +800,9 @@ function setup_interface(repl::LineEditREPL; hascolor = repl.hascolor, extra_rep
                     LineEdit.commit_line(s)
                     # This is slightly ugly but ok for now
                     terminal = LineEdit.terminal(s)
-                    Base.stop_reading(terminal)
                     raw!(terminal, false) && disable_bracketed_paste(terminal)
                     LineEdit.mode(s).on_done(s, LineEdit.buffer(s), true)
                     raw!(terminal, true) && enable_bracketed_paste(terminal)
-                    Base.start_reading(terminal)
                 else
                     break
                 end

base/iobuffer.jl

@@ -52,9 +52,6 @@ show(io::IO, b::AbstractIOBuffer) = print(io, "IOBuffer(data=UInt8[...], ",
                                       "ptr=",      b.ptr, ", ",
                                       "mark=",     b.mark, ")")
 
-is_maxsize_unlimited(io::AbstractIOBuffer) = (io.maxsize == typemax(Int))
-maxsize(io::AbstractIOBuffer) = io.maxsize
-
 read!(from::AbstractIOBuffer, a::Array) = read_sub(from, a, 1, length(a))
 
 function read_sub{T}(from::AbstractIOBuffer, a::AbstractArray{T}, offs, nel)

base/multi.jl

@@ -811,7 +811,6 @@ end
 function process_messages(r_stream::TCPSocket, w_stream::TCPSocket; kwargs...)
     @schedule begin
         disable_nagle(r_stream)
-        Base.start_reading(r_stream)
         wait_connected(r_stream)
         if r_stream != w_stream
             disable_nagle(w_stream)

base/process.jl

@@ -434,7 +434,6 @@ eachline(cmd::AbstractCmd) = eachline(cmd, DevNull)
 function open(cmds::AbstractCmd, mode::AbstractString="r", stdio::AsyncStream=DevNull)
     if mode == "r"
         processes = @tmp_rpipe out tmp spawn(false, cmds, (stdio,tmp,STDERR))
-        start_reading(out)
         (out, processes)
     elseif mode == "w"
         processes = @tmp_wpipe tmp inpipe spawn(false, cmds, (tmp,stdio,STDERR))
@@ -467,8 +466,7 @@ end
 function readbytes(cmd::AbstractCmd, stdin::AsyncStream=DevNull)
     (out,pc) = open(cmd, "r", stdin)
     !success(pc) && pipeline_error(pc)
-    wait_close(out)
-    return takebuf_array(out.buffer)
+    return readbytes(out)
 end
 
 function readall(cmd::AbstractCmd, stdin::AsyncStream=DevNull)

base/socket.jl

@@ -263,6 +263,7 @@ type TCPSocket <: Socket
     closenotify::Condition
     sendbuf::Nullable{IOBuffer}
     lock::ReentrantLock
+    throttle::Int
 
     TCPSocket(handle) = new(
         handle,
@@ -273,7 +274,8 @@ type TCPSocket <: Socket
         false, Condition(),
         false, Condition(),
         nothing,
-        ReentrantLock()
+        ReentrantLock(),
+        DEFAULT_READ_BUFFER_SZ
     )
 end
 function TCPSocket()

base/stream.jl

@@ -49,6 +49,8 @@ function eof(s::AsyncStream)
     !isopen(s) && nb_available(s.buffer)<=0
 end
 
+const DEFAULT_READ_BUFFER_SZ = 10485760           # 10 MB
+
 const StatusUninit      = 0 # handle is allocated, but not initialized
 const StatusInit        = 1 # handle is valid, but not connected/active
 const StatusConnecting  = 2 # handle is in process of connecting
@@ -104,6 +106,7 @@ type Pipe <: AsyncStream
     closenotify::Condition
     sendbuf::Nullable{IOBuffer}
     lock::ReentrantLock
+    throttle::Int
 
     Pipe(handle) = new(
         handle,
@@ -113,7 +116,8 @@ type Pipe <: AsyncStream
         false,Condition(),
         false,Condition(),
         false,Condition(),
-        nothing, ReentrantLock())
+        nothing, ReentrantLock(),
+        DEFAULT_READ_BUFFER_SZ)
 end
 function Pipe()
     handle = Libc.malloc(_sizeof_uv_named_pipe)
@@ -182,6 +186,7 @@ type TTY <: AsyncStream
     closenotify::Condition
     sendbuf::Nullable{IOBuffer}
     lock::ReentrantLock
+    throttle::Int
     @windows_only ispty::Bool
     function TTY(handle)
         tty = new(
@@ -191,7 +196,8 @@ type TTY <: AsyncStream
             PipeBuffer(),
             false,Condition(),
             false,Condition(),
-            nothing, ReentrantLock())
+            nothing, ReentrantLock(),
+            DEFAULT_READ_BUFFER_SZ)
         @windows_only tty.ispty = ccall(:jl_ispty, Cint, (Ptr{Void},), handle)!=0
         tty
     end
@@ -269,7 +275,7 @@ function init_stdio(handle)
     end
 end
 
-function stream_wait(x,c...)
+function stream_wait(x, c...) # for x::LibuvObject
     preserve_handle(x)
     try
         return wait(c...)
@@ -307,26 +313,51 @@ end
 function wait_connected(x)
     check_open(x)
     while x.status == StatusConnecting
-        stream_wait(x,x.connectnotify)
+        stream_wait(x, x.connectnotify)
         check_open(x)
     end
 end
 
 function wait_readbyte(x::AsyncStream, c::UInt8)
-    while isopen(x) && search(x.buffer,c) <= 0
-        start_reading(x)
-        stream_wait(x,x.readnotify)
+    preserve_handle(x)
+    try
+        while isopen(x) && search(x.buffer,c) <= 0
+            start_reading(x) # ensure we are reading
+            wait(x.readnotify)
+        end
+    finally
+        if isempty(x.readnotify.waitq)
+            stop_reading(x) # stop reading iff there are currently no other read clients of the stream
+        end
+        unpreserve_handle(x)
     end
 end
 
 function wait_readnb(x::AsyncStream, nb::Int)
-    while isopen(x) && nb_available(x.buffer) < nb
-        start_reading(x)
-        stream_wait(x,x.readnotify)
+    oldthrottle = x.throttle
+    preserve_handle(x)
+    try
+        while isopen(x) && nb_available(x.buffer) < nb
+            x.throttle = max(nb, x.throttle)
+            start_reading(x) # ensure we are reading
+            wait(x.readnotify)
+        end
+    finally
+        if oldthrottle <= x.throttle <= nb
+            x.throttle = oldthrottle
+        end
+        if isempty(x.readnotify.waitq)
+            stop_reading(x) # stop reading iff there are currently no other read clients of the stream
+        end
+        unpreserve_handle(x)
     end
 end
 
-wait_close(x) = if isopen(x) stream_wait(x,x.closenotify); end
+function wait_close(x::AsyncStream)
+    if isopen(x)
+        stream_wait(x, x.closenotify)
+    end
+end
 
 #from `connect`
 function _uv_hook_connectcb(sock::AsyncStream, status::Int32)
@@ -367,8 +398,7 @@ function alloc_request(buffer::IOBuffer, recommended_size::UInt)
 end
 function _uv_hook_alloc_buf(stream::AsyncStream, recommended_size::UInt)
     (buf,size) = alloc_request(stream.buffer, recommended_size)
-    @assert size>0 # because libuv requires this (TODO: possibly stop reading too if it fails)
-    (buf,UInt(size))
+    return (buf,UInt(size))
 end
 
 function notify_filled(buffer::IOBuffer, nread::Int, base::Ptr{Void}, len::UInt)
@@ -394,36 +424,39 @@ function notify_filled(stream::AsyncStream, nread::Int)
     end
 end
 
-const READ_BUFFER_SZ=10485760           # 10 MB
 function _uv_hook_readcb(stream::AsyncStream, nread::Int, base::Ptr{Void}, len::UInt)
     if nread < 0
-        if nread != UV_EOF
-            # This is a fatal connectin error. Shutdown requests as per the usual
-            # close function won't work and libuv will fail with an assertion failure
-            ccall(:jl_forceclose_uv,Void,(Ptr{Void},),stream.handle)
-            notify_error(stream.readnotify, UVError("readcb",nread))
-        else
+        if nread == UV_ENOBUFS && len == 0
+            # remind the client that stream.buffer is full
+            notify(stream.readnotify)
+        elseif nread == UV_EOF
             if isa(stream,TTY)
-                stream.status = StatusEOF
+                stream.status = StatusEOF # libuv called stop_reading already
                 notify(stream.readnotify)
                 notify(stream.closenotify)
             else
                 close(stream)
             end
+        else
+            # This is a fatal connection error. Shutdown requests as per the usual
+            # close function won't work and libuv will fail with an assertion failure
+            ccall(:jl_forceclose_uv,Void,(Ptr{Void},),stream.handle)
+            notify_error(stream.readnotify, UVError("readcb",nread))
         end
     else
         notify_filled(stream.buffer, nread, base, len)
         notify_filled(stream, nread)
         notify(stream.readnotify)
     end
 
-    # Stop reading when
-    # 1) when we have an infinite buffer, and we have accumulated a lot of unread data OR
+    # Stop background reading when
+    # 1) we have accumulated a lot of unread data OR
     # 2) we have an alternate buffer that has reached its limit.
-    if (is_maxsize_unlimited(stream.buffer) && (nb_available(stream.buffer) > READ_BUFFER_SZ )) ||
-       (nb_available(stream.buffer) == stream.buffer.maxsize)
+    if (nb_available(stream.buffer) >= stream.throttle) ||
+       (nb_available(stream.buffer) >= stream.buffer.maxsize)
         stop_reading(stream)
     end
+    nothing
 end
 
 reseteof(x::IO) = nothing
@@ -532,7 +565,7 @@ function sleep(sec::Real)
     end)
     start_timer(timer, float(sec), 0)
     try
-        stream_wait(timer,w)
+        stream_wait(timer, w)
     finally
         stop_timer(timer)
     end
@@ -640,15 +673,15 @@ function start_reading(stream::AsyncStream)
     end
 end
 function start_reading(stream::AsyncStream, cb::Function)
-    start_reading(stream)
+    failure = start_reading(stream)
     stream.readcb = cb
     nread = nb_available(stream.buffer)
     if nread > 0
         notify_filled(stream, nread)
     end
-    nothing
+    return failure_code
 end
-start_reading(stream::AsyncStream, cb::Bool) = (start_reading(stream); stream.readcb = cb; nothing)
+start_reading(stream::AsyncStream, cb::Bool) = (failure_code = start_reading(stream); stream.readcb = cb; return failure_code)
 
 function stop_reading(stream::AsyncStream)
     if stream.status == StatusActive
@@ -662,10 +695,9 @@ function stop_reading(stream::AsyncStream)
     end
 end
 
-function readall(stream::AsyncStream)
-    start_reading(stream)
-    wait_close(stream)
-    return takebuf_string(stream.buffer)
+function readbytes(stream::AsyncStream)
+    wait_readnb(stream, typemax(Int))
+    return takebuf_array(stream.buffer)
 end
 
 function read!{T}(s::AsyncStream, a::Array{T})
@@ -687,16 +719,22 @@ function read!(s::AsyncStream, a::Vector{UInt8})
     end
 
     if nb <= SZ_UNBUFFERED_IO # Under this limit we are OK with copying the array from the stream's buffer
-        wait_readnb(s,nb)
+        wait_readnb(s, nb)
         read!(sbuf, a)
     else
-        stop_reading(s) # Just playing it safe, since we are going to switch buffers.
-        newbuf = PipeBuffer(a, nb)
-        newbuf.size = 0
-        s.buffer = newbuf
-        write(newbuf, sbuf)
-        wait_readnb(s,nb)
-        s.buffer = sbuf
+        try
+            stop_reading(s) # Just playing it safe, since we are going to switch buffers.
+            newbuf = PipeBuffer(a, #=maxsize=# nb)
+            newbuf.size = 0 # reset the write pointer to the beginning
+            s.buffer = newbuf
+            write(newbuf, sbuf)
+            wait_readnb(s, nb)
+        finally
+            s.buffer = sbuf
+            if !isempty(s.readnotify.waitq)
+                start_reading(x) # resume reading iff there are currently other read clients of the stream
+            end
+        end
     end
     return a
 end
@@ -711,8 +749,8 @@ end
 function read(this::AsyncStream,::Type{UInt8})
     buf = this.buffer
     @assert buf.seekable == false
-    wait_readnb(this,1)
-    read(buf,UInt8)
+    wait_readnb(this, 1)
+    read(buf, UInt8)
 end
 
 readline(this::AsyncStream) = readuntil(this, '\n')
@@ -776,11 +814,11 @@ function buffer_or_write(s::AsyncStream, p::Ptr, n::Integer)
     end
 
     totb = nb_available(buf) + n
-    if totb < maxsize(buf)
+    if totb < buf.maxsize
         nb = write(buf, p, n)
     else
         flush(s)
-        if n > maxsize(buf)
+        if n > buf.maxsize
             nb = uv_write(s, p, n)
         else
             nb = write(buf, p, n)
@@ -988,19 +1026,10 @@ function wait_readnb(s::BufferStream, nb::Int)
     while isopen(s) && nb_available(s.buffer) < nb
         wait(s.r_c)
     end
-
-    (nb_available(s.buffer) < nb) && error("closed BufferStream")
-end
-
-function eof(s::BufferStream)
-    wait_readnb(s,1)
-    !isopen(s) && nb_available(s.buffer)<=0
 end
 
 show(io::IO, s::BufferStream) = print(io,"BufferStream() bytes waiting:",nb_available(s.buffer),", isopen:", s.is_open)
 
-nb_available(s::BufferStream) = nb_available(s.buffer)
-
 function wait_readbyte(s::BufferStream, c::UInt8)
     while isopen(s) && search(s.buffer,c) <= 0
         wait(s.r_c)

test/repl.jl

@@ -16,7 +16,7 @@ function fake_repl()
     Base.link_pipe(stderr_read,true,stderr_write,true)
 
     repl = Base.REPL.LineEditREPL(TestHelpers.FakeTerminal(stdin_read, stdout_write, stderr_write))
-    stdin_write, stdout_read, stdout_read, repl
+    stdin_write, stdout_read, stderr_read, repl
 end
 
 # Writing ^C to the repl will cause sigint, so let's not die on that
@@ -28,7 +28,7 @@ ccall(:jl_exit_on_sigint, Void, (Cint,), 0)
 # this should make sure nothing crashes without depending on how exactly the control
 # characters are being used.
 if @unix? true : (Base.windows_version() >= Base.WINDOWS_VISTA_VER)
-stdin_write, stdout_read, stdout_read, repl = fake_repl()
+stdin_write, stdout_read, stderr_read, repl = fake_repl()
 
 repl.specialdisplay = Base.REPL.REPLDisplay(repl)
 repl.history_file = false
@@ -261,17 +261,19 @@ master = Base.TTY(RawFD(fdm); readable = true)
 nENV = copy(ENV)
 nENV["TERM"] = "dumb"
 p = spawn(setenv(`$exename --startup-file=no --quiet`,nENV),slave,slave,slave)
-Base.start_reading(master)
-Base.wait_readnb(master,1)
+output = readuntil(master,"julia> ")
+if ccall(:jl_running_on_valgrind,Cint,()) == 0
+    # If --trace-children=yes is passed to valgrind, we will get a
+    # valgrind banner here, not just the prompt.
+    @test output == "julia> "
+end
 write(master,"1\nquit()\n")
 
 wait(p)
-
-ccall(:close,Cint,(Cint,),fds)
-output = readall(master)
-if ccall(:jl_running_on_valgrind,Cint,()) == 0
-    @test output == "julia> 1\r\nquit()\r\n1\r\n\r\njulia> "
-end
+output = readuntil(master,' ')
+@test output == "1\r\nquit()\r\n1\r\n\r\njulia> "
+@test nb_available(master) == 0
+ccall(:close,Cint,(Cint,),fds) # XXX: this causes the kernel to throw away all unread data on the pty
 close(master)
 
 end