Merge pull request #3878 from stevengj/readbytes

new readbytes and readbytes! functions

base/exports.jl

@@ -1074,6 +1074,8 @@ export
     position,
     read,
     readall,
+    readbytes,
+    readbytes!,
     readchomp,
     readdir,
     readline,

base/fs.jl

@@ -147,9 +147,12 @@ function read(f::File, ::Type{Uint8})
     return uint8(ret)
 end
 
-function read{T}(f::File, a::Array{T})
+function read{T}(f::File, a::Array{T}, nel=length(a))
+    if nel < 0 || nel > length(a)
+        throw(BoundsError())
+    end
     if isbits(T)
-        nb = length(a)*sizeof(T)
+        nb = nel*sizeof(T)
         ret = ccall(:jl_fs_read, Int32, (Int32, Ptr{Void}, Csize_t),
                     f.handle, a, nb)
         uv_error("write",ret == -1)
@@ -159,8 +162,21 @@ function read{T}(f::File, a::Array{T})
     a
 end
 
+nb_available(f::File) = filesize(f) - position(f)
+
+function readbytes!(f::File, b::Array{Uint8}, nb=length(b))
+    nr = min(nb, nb_available(f))
+    if length(b) < nr
+        resize!(b, nr)
+    end
+    read(f, b, nr)
+    return nr
+end
+readbytes(io::File) = read(io, Array(Uint8, nb_available(io)))
+readbytes(io::File, nb) = read(io, Array(Uint8, min(nb, nb_available(io))))
+
 function readbytes(f::File)
-    a = Array(Uint8, filesize(f) - position(f))
+    a = Array(Uint8, nb_available(f))
     read(f,a)
     a
 end

base/io.jl

@@ -165,17 +165,33 @@ function readuntil{T}(s::IO, delim::T)
 end
 
 readline(s::IO) = readuntil(s, '\n')
+readchomp(x) = chomp!(readall(x))
 
-function readall(s::IO)
-    out = IOBuffer()
-    while !eof(s)
+# read up to nb bytes into nb, returning # bytes read
+function readbytes!(s::IO, b::AbstractArray{Uint8}, nb=length(b))
+    olb = lb = length(b)
+    nr = 0
+    while !eof(s) && nr < nb
         a = read(s, Uint8)
-        write(out, a)
+        nr += 1
+        if nr > lb
+            lb = nr * 2
+            resize!(b, lb)
+        end
+        b[nr] = a
     end
-    takebuf_string(out)
+    if lb > olb
+        resize!(b, nr) # shrink to just contain input data if was resized
+    end
+    return nr
 end
 
-readchomp(x) = chomp!(readall(x))
+# read up to nb bytes from s, returning a Vector{Uint8} of bytes read.
+function readbytes(s::IO, nb=typemax(Int))
+    b = Array(Uint8, min(nb, 65536))
+    nr = readbytes!(s, b, nb)
+    resize!(b, nr)
+end
 
 function readall(s::IO)
     b = readbytes(s)
@@ -414,21 +430,22 @@ function readuntil(s::IOStream, delim::Uint8)
     ccall(:jl_readuntil, Array{Uint8,1}, (Ptr{Void}, Uint8), s.ios, delim)
 end
 
-function readbytes(s::IOStream)
-    n = 65536
-    b = Array(Uint8, n)
-    p = 1
-    while true
-        nr = int(ccall(:ios_readall, Uint,
-                       (Ptr{Void}, Ptr{Void}, Uint), s.ios, pointer(b,p), n))
-        if eof(s)
-            resize!(b, p+nr-1)
-            break
+function readbytes!(s::IOStream, b::Array{Uint8}, nb=length(b))
+    olb = lb = length(b)
+    nr = 0
+    while !eof(s) && nr < nb
+        if lb < nr+1
+            lb = max(65536, (nr+1) * 2)
+            resize!(b, lb)
         end
-        p += nr
-        resize!(b, p+n-1)
+        nr += int(ccall(:ios_readall, Uint,
+                        (Ptr{Void}, Ptr{Void}, Uint),
+                        s.ios, pointer(b, nr+1), lb - nr))
+    end
+    if lb > olb
+        resize!(b, nr) # shrink to just contain input data if was resized
     end
-    b
+    return nr
 end
 
 # based on code by Glen Hertz

base/iobuffer.jl

@@ -38,19 +38,28 @@ IOBuffer(readable::Bool,writable::Bool) = IOBuffer(Uint8[],readable,writable)
 IOBuffer() = IOBuffer(Uint8[], true, true)
 IOBuffer(maxsize::Int) = (x=IOBuffer(Array(Uint8,maxsize),true,true,maxsize); x.size=0; x)
 
-function read{T}(from::IOBuffer, a::Array{T})
-    if !from.readable error("read failed") end
-    if isbits(T)
-        nb = length(a)*sizeof(T)
-        if nb > nb_available(from)
-            throw(EOFError())
-        end
-        ccall(:memcpy, Ptr{Void}, (Ptr{Void}, Ptr{Void}, Uint), a, pointer(from.data,from.ptr), nb)
-        from.ptr += nb
-        return a
-    else
+read(from::IOBuffer, a::Array) = read_sub(from, a, 1, length(a))
+
+function read_sub{T}(from::IOBuffer, a::Array{T}, offs, nel)
+    if offs+nel-1 > length(a) || offs < 1 || nel < 0
+        throw(BoundsError())
+    end
+    if !isbits(T)
         error("Read from IOBuffer only supports bits types or arrays of bits types; got "*string(T)*".")
     end
+    read(from, pointer(a, offs), nel*sizeof(T))
+    return a
+end
+
+read(from::IOBuffer, p::Ptr, nb::Integer) = read(from, p, int(nb))
+function read(from::IOBuffer, p::Ptr, nb::Int)
+    if !from.readable error("read failed") end
+    if nb > nb_available(from)
+        throw(EOFError())
+    end
+    ccall(:memcpy, Ptr{Void}, (Ptr{Void}, Ptr{Void}, Uint), p, pointer(from.data,from.ptr), nb)
+    from.ptr += nb
+    p
 end
 
 function read(from::IOBuffer, ::Type{Uint8})
@@ -216,8 +225,17 @@ end
 
 write(to::IOBuffer, p::Ptr) = write(to, convert(Uint, p))
 
-readbytes(io::IOBuffer,nb::Integer) = bytestring(read(io, Array(Uint8, nb)))
-readall(io::IOBuffer) = readbytes(io,nb_available(io))
+function readbytes!(io::IOBuffer, b::Array{Uint8}, nb=length(b))
+    nr = min(nb, nb_available(io))
+    if length(b) < nr
+        resize!(b, nr)
+    end
+    read_sub(io, b, 1, nr)
+    return nr
+end
+readbytes(io::IOBuffer) = read(io, Array(Uint8, nb_available(io)))
+readbytes(io::IOBuffer, nb) = read(io, Array(Uint8, min(nb, nb_available(io))))
+
 function search(buf::IOBuffer, delim)
     p = pointer(buf.data, buf.ptr)
     q = ccall(:memchr,Ptr{Uint8},(Ptr{Uint8},Int32,Csize_t),p,delim,nb_available(buf))

base/multi.jl

@@ -1006,6 +1006,10 @@ function create_worker(privhost, port, pubhost, stream, config)
                 if nread>0
                     try
                         line = readbytes(stream.buffer, nread)
+                        if length(line) < nread
+                            println(STDERR,"\tTruncated reply from worker $(wrker.id):\t",err)
+                            return false
+                        end
                         print("\tFrom worker $(wrker.id):\t",line)
                     catch err
                         println(STDERR,"\tError parsing reply from worker $(wrker.id):\t",err)

doc/helpdb.jl

@@ -1667,6 +1667,26 @@
 
 "),
 
+("I/O","Base","readbytes!","readbytes!(stream, b::Vector{Uint8}, nb=length(b))
+
+   Read up to nb bytes from the stream into b, returning the
+   number of bytes read (increasing the size of b as needed).
+
+"),
+
+("I/O","Base","readbytes","readbytes(stream, nb=typemax(Int))
+
+   Read at most nb bytes from the stream, returning a
+   Vector{Uint8} of the bytes read.
+
+"),
+
+("Text I/O","Base","readall","readall(stream)
+
+   Read the entire contents of an I/O stream as a string.
+
+"),
+
 ("Text I/O","Base","readline","readline(stream)
 
    Read a single line of text, including a trailing newline character

doc/stdlib/base.rst

@@ -970,6 +970,16 @@ I/O
 
    Read a series of values of the given type from a stream, in canonical binary representation. ``dims`` is either a tuple or a series of integer arguments specifying the size of ``Array`` to return.
 
+.. function:: readbytes!(stream, b::Vector{Uint8}, nb=length(b))
+
+   Read at most nb bytes from the stream into b, returning the
+   number of bytes read (increasing the size of b as needed).
+
+.. function:: readbytes(stream, nb=typemax(Int))
+
+   Read at most nb bytes from the stream, returning a
+   Vector{Uint8} of the bytes read.
+
 .. function:: position(s)
 
    Get the current position of a stream.