add stats (#30)

Author: bicycle1885

docs/src/examples.md

@@ -160,6 +160,35 @@ read(GzipDecompressionStream(stream, stop_on_end=true))  #> the content of foo.t
 read(GzipDecompressionStream(stream, stop_on_end=true))  #> the content of bar.txt
 ```
 
+Check I/O statistics
+--------------------
+
+`TranscodingStreams.stats` returns a snapshot of the I/O statistics. For
+example, the following function shows progress of decompression to the standard
+error:
+```julia
+using CodecZlib
+
+function decompress(input, output)
+    buffer = Vector{UInt8}(16 * 1024)
+    while !eof(input)
+        n = min(nb_available(input), length(buffer))
+        unsafe_read(input, pointer(buffer), n)
+        unsafe_write(output, pointer(buffer), n)
+        stats = TranscodingStreams.stats(input)
+        print(STDERR, "\rin: $(stats.in), out: $(stats.out)")
+    end
+    println(STDERR)
+end
+
+input = GzipDecompressionStream(open("foobar.txt.gz"))
+output = IOBuffer()
+decompress(input, output)
+```
+
+`stats.in` is the number of bytes supplied to the stream and `stats.out` is the
+number of bytes consumed out of the stream.
+
 Transcode data in one shot
 --------------------------
 

docs/src/references.md

@@ -17,6 +17,14 @@ TranscodingStreams.unread
 TranscodingStreams.unsafe_unread
 ```
 
+Statistics
+----------
+
+```@docs
+TranscodingStreams.Stats
+TranscodingStreams.stats
+```
+
 Codec
 -----
 

src/buffer.jl

@@ -89,6 +89,18 @@ function supplied!(buf::Buffer, n::Integer)
     return buf
 end
 
+function consumed2!(buf::Buffer, n::Integer)
+    buf.bufferpos += n
+    buf.total += n
+    return buf
+end
+
+function supplied2!(buf::Buffer, n::Integer)
+    buf.marginpos += n
+    buf.total += n
+    return buf
+end
+
 function readbyte!(buf::Buffer)
     b = buf.data[buf.bufferpos]
     consumed!(buf, 1)

src/noop.jl

@@ -114,6 +114,30 @@ function Base.transcode(::Noop, data::Vector{UInt8})
 end
 
 
+# Stats
+# -----
+
+function stats(stream::NoopStream)
+    state = stream.state
+    mode = state.mode
+    @checkmode (:idle, :read, :write)
+    buffer = state.buffer1
+    @assert buffer == stream.state.buffer2
+    if mode == :idle
+        consumed = supplied = 0
+    elseif mode == :read
+        supplied = buffer.total
+        consumed = supplied - buffersize(buffer)
+    elseif mode == :write
+        supplied = buffer.total + buffersize(buffer)
+        consumed = buffer.total
+    else
+        assert(false)
+    end
+    return Stats(consumed, supplied, supplied, supplied)
+end
+
+
 # Buffering
 # ---------
 #
@@ -133,29 +157,28 @@ function fillbuffer(stream::NoopStream)
         makemargin!(buffer, 1)
         nfilled += readdata!(stream.stream, buffer)
     end
+    buffer.total += nfilled
     return nfilled
 end
 
-function flushbuffer(stream::NoopStream)
+function flushbuffer(stream::NoopStream, all::Bool=false)
     changemode!(stream, :write)
     buffer = stream.state.buffer1
     @assert buffer === stream.state.buffer2
-    nflushed = writedata!(stream.stream, buffer)
-    makemargin!(buffer, 0)
-    return nflushed
-end
-
-function flushbufferall(stream::NoopStream)
-    changemode!(stream, :write)
-    buffer = stream.state.buffer1
-    bufsize = buffersize(buffer)
-    while buffersize(buffer) > 0
-        writedata!(stream.stream, buffer)
+    nflushed::Int = 0
+    if all
+        while buffersize(buffer) > 0
+            nflushed += writedata!(stream.stream, buffer)
+        end
+    else
+        nflushed += writedata!(stream.stream, buffer)
+        makemargin!(buffer, 0)
     end
-    return bufsize
+    buffer.total += nflushed
+    return nflushed
 end
 
 function flushuntilend(stream::NoopStream)
-    writedata!(stream.stream, stream.state.buffer1)
+    stream.state.buffer1.total += writedata!(stream.stream, stream.state.buffer1)
     return
 end

src/stream.jl

@@ -431,31 +431,64 @@ function Base.flush(stream::TranscodingStream)
 end
 
 
-# Utils
+# Stats
 # -----
 
-function total_in(stream::TranscodingStream)::Int64
-    checkmode(stream)
-    state = stream.state
-    if state.mode == :read
-        return state.buffer2.total
-    elseif state.mode == :write
-        return state.buffer1.total
-    else
-        return zero(Int64)
-    end
+"""
+I/O statistics.
+
+Its object has four fields:
+- `in`: the number of bytes supplied into the stream
+- `out`: the number of bytes consumed out of the stream
+- `transcoded_in`: the number of bytes transcoded from the input buffer
+- `transcoded_out`: the number of bytes transcoded to the output buffer
+
+Note that, since the transcoding stream does buffering, `in` is `transcoded_in +
+{size of buffered data}` and `out` is `transcoded_out - {size of buffered
+data}`.
+"""
+struct Stats
+    in::Int64
+    out::Int64
+    transcoded_in::Int64
+    transcoded_out::Int64
 end
 
-function total_out(stream::TranscodingStream)::Int64
-    checkmode(stream)
+function Base.show(io::IO, stats::Stats)
+    println(io, summary(stats), ':')
+    println(io, "  in: ", stats.in)
+    println(io, "  out: ", stats.out)
+    println(io, "  transcoded_in: ", stats.transcoded_in)
+      print(io, "  transcoded_out: ", stats.transcoded_out)
+end
+
+"""
+    stats(stream::TranscodingStream)
+
+Create an I/O statistics object of `stream`.
+"""
+function stats(stream::TranscodingStream)
     state = stream.state
-    if state.mode == :read
-        return state.buffer1.total
-    elseif state.mode == :write
-        return state.buffer2.total
+    mode = state.mode
+    @checkmode (:idle, :read, :write)
+    buffer1 = state.buffer1
+    buffer2 = state.buffer2
+    if mode == :idle
+        transcoded_in = transcoded_out = in = out = 0
+    elseif mode == :read
+        transcoded_in = buffer2.total
+        transcoded_out = buffer1.total
+        in = transcoded_in + buffersize(buffer2)
+        out = transcoded_out - buffersize(buffer1)
+    elseif mode == :write
+        transcoded_in = buffer1.total
+        transcoded_out = buffer2.total
+        in = transcoded_in + buffersize(buffer1)
+        out = transcoded_out - buffersize(buffer2)
     else
-        return zero(Int64)
+        assert(false)
     end
+    return Stats(in, out, transcoded_in, transcoded_out)
 end
 
 
@@ -537,8 +570,8 @@ function callprocess(stream::TranscodingStream, inbuf::Buffer, outbuf::Buffer)
     input = buffermem(inbuf)
     makemargin!(outbuf, minoutsize(stream.codec, input))
     Δin, Δout, state.code = process(stream.codec, input, marginmem(outbuf), state.error)
-    consumed!(inbuf, Δin)
-    supplied!(outbuf, Δout)
+    consumed2!(inbuf, Δin)
+    supplied2!(outbuf, Δout)
     if state.code == :error
         changemode!(stream, :panic)
     elseif state.code == :ok && Δin == Δout == 0

test/runtests.jl

@@ -30,6 +30,21 @@ using Base.Test
     @test mem.size == sizeof(data)
 end
 
+@testset "Stats" begin
+    stats = TranscodingStreams.Stats(1,2,3,4)
+    @test repr(stats) ==
+    """
+    TranscodingStreams.Stats:
+      in: 1
+      out: 2
+      transcoded_in: 3
+      transcoded_out: 4"""
+    @test stats.in == 1
+    @test stats.out == 2
+    @test stats.transcoded_in == 3
+    @test stats.transcoded_out == 4
+end
+
 @testset "TranscodingStream" begin
     @test TranscodingStreams.splitkwargs(
         [(:foo, 1), (:bar, true), (:baz, :ok)], (:foo,)) ==
@@ -200,24 +215,22 @@ end
     @test s1.state.buffer1 === s2.state.buffer1 === s3.state.buffer1 ===
           s1.state.buffer2 === s2.state.buffer2 === s3.state.buffer2
 
-    #= FIXME: restore these tests
     stream = TranscodingStream(Noop(), IOBuffer(b"foobar"))
-    @test TranscodingStreams.total_in(stream) === Int64(0)
-    @test TranscodingStreams.total_out(stream) === Int64(0)
+    @test TranscodingStreams.stats(stream).in === Int64(0)
+    @test TranscodingStreams.stats(stream).out === Int64(0)
     read(stream)
-    @test TranscodingStreams.total_in(stream) === Int64(6)
-    @test TranscodingStreams.total_out(stream) === Int64(6)
+    @test TranscodingStreams.stats(stream).in === Int64(6)
+    @test TranscodingStreams.stats(stream).out === Int64(6)
     close(stream)
 
     stream = TranscodingStream(Noop(), IOBuffer())
-    @test TranscodingStreams.total_in(stream) === Int64(0)
-    @test TranscodingStreams.total_out(stream) === Int64(0)
+    @test TranscodingStreams.stats(stream).in === Int64(0)
+    @test TranscodingStreams.stats(stream).out === Int64(0)
     write(stream, b"foobar")
     flush(stream)
-    @test TranscodingStreams.total_in(stream) === Int64(6)
-    @test TranscodingStreams.total_out(stream) === Int64(6)
+    @test TranscodingStreams.stats(stream).in === Int64(6)
+    @test TranscodingStreams.stats(stream).out === Int64(6)
     close(stream)
-    =#
 
     stream = NoopStream(IOBuffer("foobar"))
     @test nb_available(stream) === 0
@@ -420,3 +433,44 @@ end
         @test read(stream) == b"abracadabra"
     end
 end
+
+@testset "stats" begin
+    size = filesize(joinpath(dirname(@__FILE__), "abra.gzip"))
+    stream = CodecZlib.GzipDecompressionStream(open(joinpath(dirname(@__FILE__), "abra.gzip")))
+    stats = TranscodingStreams.stats(stream)
+    @test stats.in == 0
+    @test stats.out == 0
+    @test stats.transcoded_in == 0
+    @test stats.transcoded_out == 0
+    read(stream, UInt8)
+    stats = TranscodingStreams.stats(stream)
+    @test stats.in == size
+    @test stats.out == 1
+    @test stats.transcoded_in == size
+    @test stats.transcoded_out == 11
+    close(stream)
+    @test_throws ArgumentError TranscodingStreams.stats(stream)
+
+    buf = IOBuffer()
+    stream = CodecZlib.GzipCompressionStream(buf)
+    stats = TranscodingStreams.stats(stream)
+    @test stats.in == 0
+    @test stats.out == 0
+    @test stats.transcoded_in == 0
+    @test stats.transcoded_out == 0
+    write(stream, b"abracadabra")
+    stats = TranscodingStreams.stats(stream)
+    @test stats.in == 11
+    @test stats.out == 0
+    @test stats.transcoded_in == 0
+    @test stats.transcoded_out == 0
+    write(stream, TranscodingStreams.TOKEN_END)
+    flush(stream)
+    stats = TranscodingStreams.stats(stream)
+    @test stats.in == 11
+    @test stats.out == position(buf)
+    @test stats.transcoded_in == 11
+    @test stats.transcoded_out == position(buf)
+    close(stream)
+    @test_throws ArgumentError TranscodingStreams.stats(stream)
+end