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

This reverts commit 5863b48.

Author: amitmurthy

base/LineEdit.jl

@@ -1625,6 +1625,7 @@ 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, term)
         while true
             map = keymap(s, prompt)
@@ -1640,11 +1641,14 @@ 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

@@ -824,9 +824,11 @@ 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

@@ -53,6 +53,9 @@ show(io::IO, b::AbstractIOBuffer) = print(io, "IOBuffer(data=UInt8[...], ",
                                       "mark=",     b.mark, ")")
 
 read!(from::AbstractIOBuffer, a::Vector{UInt8}) = read_sub(from, a, 1, length(a))
+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

@@ -946,6 +946,7 @@ process_messages(r_stream::TCPSocket, w_stream::TCPSocket) = @schedule process_t
 
 function process_tcp_streams(r_stream::TCPSocket, w_stream::TCPSocket)
         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

@@ -546,6 +546,7 @@ function open(cmds::AbstractCmd, mode::AbstractString="r", other::Redirectable=D
         in = other
         out = io = Pipe()
         processes = spawn(cmds, (in,out,STDERR))
+        start_reading(io)
         close(out.in)
     elseif mode == "w"
         in = io = Pipe()

base/socket.jl

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

base/stream.jl

@@ -91,8 +91,6 @@ function eof(s::LibuvStream)
     !isopen(s) && nb_available(s)<=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
@@ -148,7 +146,6 @@ type PipeEndpoint <: LibuvStream
     closenotify::Condition
     sendbuf::Nullable{IOBuffer}
     lock::ReentrantLock
-    throttle::Int
 
     PipeEndpoint(handle::Ptr{Void} = C_NULL) = new(
         handle,
@@ -158,8 +155,7 @@ type PipeEndpoint <: LibuvStream
         false,Condition(),
         false,Condition(),
         false,Condition(),
-        nothing, ReentrantLock(),
-        DEFAULT_READ_BUFFER_SZ)
+        nothing, ReentrantLock())
 end
 
 type PipeServer <: LibuvServer
@@ -202,7 +198,6 @@ type TTY <: LibuvStream
     closenotify::Condition
     sendbuf::Nullable{IOBuffer}
     lock::ReentrantLock
-    throttle::Int
     @windows_only ispty::Bool
     function TTY(handle)
         tty = new(
@@ -212,8 +207,7 @@ type TTY <: LibuvStream
             PipeBuffer(),
             false,Condition(),
             false,Condition(),
-            nothing, ReentrantLock(),
-            DEFAULT_READ_BUFFER_SZ)
+            nothing, ReentrantLock())
         @windows_only tty.ispty = ccall(:jl_ispty, Cint, (Ptr{Void},), handle)!=0
         tty
     end
@@ -289,6 +283,15 @@ function init_stdio(handle::Ptr{Void})
     end
 end
 
+function stream_wait(x,c...)
+    preserve_handle(x)
+    try
+        return wait(c...)
+    finally
+        unpreserve_handle(x)
+    end
+end
+
 function reinit_stdio()
     global uv_jl_asynccb       = cfunction(uv_asynccb, Void, (Ptr{Void},))
     global uv_jl_timercb       = cfunction(uv_timercb, Void, (Ptr{Void},))
@@ -324,51 +327,27 @@ end
 function wait_connected(x::Union{LibuvStream, LibuvServer})
     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::LibuvStream, c::UInt8)
-    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)
+    while isopen(x) && search(x.buffer,c) <= 0
+        start_reading(x)
+        stream_wait(x,x.readnotify)
     end
 end
 
 function wait_readnb(x::LibuvStream, nb::Int)
-    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)
+    while isopen(x) && nb_available(x.buffer) < nb
+        start_reading(x)
+        stream_wait(x,x.readnotify)
     end
 end
 
-function wait_close(x::Union{LibuvStream, LibuvServer})
-    if isopen(x)
-        stream_wait(x, x.closenotify)
-    end
-end
+wait_close(x) = if isopen(x) stream_wait(x,x.closenotify); end
 
 function close(stream::Union{LibuvStream, LibuvServer})
     if isopen(stream) && stream.status != StatusClosing
@@ -503,6 +482,7 @@ function notify_filled(stream::LibuvStream, nread::Int)
     end
 end
 
+const READ_BUFFER_SZ=10485760           # 10 MB
 function uv_readcb(handle::Ptr{Void}, nread::Cssize_t, buf::Ptr{Void})
     stream = @handle_as handle LibuvStream
     nread = Int(nread)
@@ -533,11 +513,11 @@ function uv_readcb(handle::Ptr{Void}, nread::Cssize_t, buf::Ptr{Void})
         notify(stream.readnotify)
     end
 
-    # Stop background reading when
-    # 1) we have accumulated a lot of unread data OR
+    # Stop reading when
+    # 1) when we have an infinite buffer, and we have accumulated a lot of unread data OR
     # 2) we have an alternate buffer that has reached its limit.
-    if (nb_available(stream.buffer) >= stream.throttle) ||
-       (nb_available(stream.buffer) >= stream.buffer.maxsize)
+    if (is_maxsize_unlimited(stream.buffer) && (nb_available(stream.buffer) > READ_BUFFER_SZ )) ||
+       (nb_available(stream.buffer) == stream.buffer.maxsize)
         stop_reading(stream)
     end
     nothing
@@ -855,7 +835,7 @@ function start_reading(stream::LibuvStream, cb::Function)
     if nread > 0
         notify_filled(stream, nread)
     end
-    return failure_code
+    nothing
 end
 
 function start_reading(stream::LibuvStream, cb::Bool)
@@ -892,22 +872,16 @@ function read!(s::LibuvStream, a::Array{UInt8, 1})
     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
-        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
+        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
     end
     return a
 end
@@ -916,7 +890,8 @@ function read(this::LibuvStream, ::Type{UInt8})
     wait_readnb(this, 1)
     buf = this.buffer
     @assert buf.seekable == false
-    read(buf, UInt8)
+    wait_readnb(this,1)
+    read(buf,UInt8)
 end
 
 function readavailable(this::LibuvStream)
@@ -964,11 +939,11 @@ function buffer_or_write(s::LibuvStream, p::Ptr, n::Integer)
 
     buf = get(s.sendbuf)
     totb = nb_available(buf) + n
-    if totb < buf.maxsize
+    if totb < maxsize(buf)
         nb = write(buf, p, n)
     else
         flush(s)
-        if n > buf.maxsize
+        if n > maxsize(buf)
             nb = uv_write(s, p, n)
         else
             nb = write(buf, p, n)
@@ -1187,10 +1162,19 @@ 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

@@ -17,7 +17,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, stderr_read, repl
+    stdin_write, stdout_read, stdout_read, repl
 end
 
 # Writing ^C to the repl will cause sigint, so let's not die on that
@@ -29,7 +29,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, stderr_read, repl = fake_repl()
+stdin_write, stdout_read, stdout_read, repl = fake_repl()
 
 repl.specialdisplay = Base.REPL.REPLDisplay(repl)
 repl.history_file = false
@@ -367,19 +367,17 @@ 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)
-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
+Base.start_reading(master)
+Base.wait_readnb(master,1)
 write(master,"1\nquit()\n")
 
 wait(p)
-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
+
+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
 close(master)
 
 end