Implement shared buffers (#22)

Author: bicycle1885

docs/make.jl

@@ -5,7 +5,7 @@ makedocs(
     format=:html,
     sitename="TranscodingStreams.jl",
     modules=[TranscodingStreams],
-    pages=["index.md", "examples.md", "references.md"],
+    pages=["index.md", "examples.md", "references.md", "devnotes.md"],
     assets=["assets/custom.css"])
 
 deploydocs(

docs/src/assets/modes.dot

@@ -0,0 +1,14 @@
+digraph modes {
+    "idle" -> "read";
+    "idle" -> "write";
+    "idle" -> "close";
+    "idle" -> "panic";
+
+    "read" -> "read";
+    "read" -> "close";
+    "read" -> "panic";
+
+    "write" -> "write";
+    "write" -> "close";
+    "write" -> "panic";
+}

docs/src/assets/modes.svg

@@ -0,0 +1,85 @@
+Developer's Notes
+=================
+
+These notes are not for end users but rather for developers who are interested
+in the design of the package.
+
+
+`TranscodingStream` type
+------------------------
+
+`TranscodingStream{C,S}` (defined in src/stream.jl) has three fields:
+- `codec`: data codec (`<:C where C<:Codec`)
+- `stream`: data stream (`<:S where S<:IO`)
+- `state`: current state (`<:State`).
+
+A codec will be implemented by package developers and only a special codec
+`Noop` is defined in this package.  A stream can be any object that implements
+at least `Base.isopen`, `Base.eof`, `Base.close`, `Base.nb_available`,
+`Base.unsafe_read`, and `Base.unsafe_write`.  All mutable fields are delegated
+to `state` and hence the stream type itself is immutable.
+
+A stream has two buffers in the `state` field. These are used to store
+pre-transcoded and transcoded data in the stream. The stream passes references
+of these two buffers to the codec when processing data. The following diagram
+illustrates the flow of data:
+
+    When reading data (`state.mode == :read`):
+      user <--- |state.buffer1| <--- <stream.codec> <--- |state.buffer2| <--- stream
+
+    When writing data (`state.mode == :write`):
+      user ---> |state.buffer1| ---> <stream.codec> ---> |state.buffer2| ---> stream
+
+In the read mode, a user pull out data from `state.buffer1` and pre-transcoded
+data are filled in `state.buffer2`. In the write mode, a user will push data
+into `state.buffer1` and transcoded data are filled in `state.buffer2`. The
+default buffer size is 16KiB for each.
+
+`State` (defined in src/state.jl) has five fields:
+- `mode`: current stream mode (`<:Symbol`)
+- `code`: return code of the last codec's method call (`<:Symbol`)
+- `error`: exception returned by the codec (`<:Error`)
+- `buffer1`: data buffer that is closer to the user (`<:Buffer`)
+- `buffer2`: data buffer that is farther to the user (`<:Buffer`)
+
+The `mode` field may be one of the following value:
+- `:idle` : initial and intermediate mode, no buffered data
+- `:read` : ready to read data, data may be buffered
+- `:write`: ready to write data, data may be buffered
+- `:close`: closed, no buffered data
+- `:panic`: an exception has been thrown in codec, data may be buffered but we
+            cannot do anything
+
+The initial mode is `:idle` and mode transition happens as shown in the
+following diagram:
+![Mode transition](./assets/modes.svg)
+
+The mode transition should happen in the `changemode!(stream, newmode)` function
+in src/stream.jl. Trying an undefined transition will thrown an exception.
+
+A transition happens based on actions (or function calls) of the user or return
+code of the codec. For example, calling `read(stream)` will change the mode from
+`:init` to `:read` and then calling `close(stream)` will change the mode from
+`:read` to `:close`. When data processing fails in the codec, a codec will
+return `:error` and the stream will result in `:panic`.
+
+
+Shared buffers
+--------------
+
+Adjacent transcoding streams may share their buffers. This will reduce memory
+allocation and eliminate data copy between buffers.
+
+`readdata!(input::IO, output::Buffer)` and `writedata!(output::IO,
+input::Buffer)` do the actual work of read/write data from/to the underlying
+stream. These methods have a special pass for shared buffers.
+
+
+`Noop` codec
+------------
+
+`Noop` (`NoopStream`) is a codec that does *nothing*. It works as a buffering
+layer on top of the underlying stream. Since `NoopStream` does not need to have
+two distinct buffers, `buffer1` and `buffer2` in the `State` object are shared
+and some specialized methods are defined for the type. All of these are defined
+in src/noop.jl.

docs/src/devnotes.md

@@ -77,16 +77,27 @@ function buffermem(buf::Buffer)
     return Memory(bufferptr(buf), buffersize(buf))
 end
 
+# Notify that `n` bytes are consumed from `buf`.
+function consumed!(buf::Buffer, n::Integer)
+    buf.bufferpos += n
+    return buf
+end
+
+# Notify that `n` bytes are supplied to `buf`.
+function supplied!(buf::Buffer, n::Integer)
+    buf.marginpos += n
+    return buf
+end
+
 function readbyte!(buf::Buffer)
     b = buf.data[buf.bufferpos]
-    buf.bufferpos += 1
+    consumed!(buf, 1)
     return b
 end
 
 function writebyte!(buf::Buffer, b::UInt8)
     buf.data[buf.marginpos] = b
-    buf.marginpos += 1
-    buf.total += 1
+    supplied!(buf, 1)
     return 1
 end
 
@@ -102,6 +113,10 @@ function marginmem(buf::Buffer)
     return Memory(marginptr(buf), marginsize(buf))
 end
 
+function ismarked(buf::Buffer)
+    return buf.markpos != 0
+end
+
 function mark!(buf::Buffer)
     return buf.markpos = buf.bufferpos
 end
@@ -160,11 +175,12 @@ function emptybuffer!(buf::Buffer)
     return buf
 end
 
+# Skip `n` bytes in the buffer.
 function skipbuffer!(buf::Buffer, n::Integer)
     if n > buffersize(buf)
         throw(ArgumentError("too large skip size"))
     end
-    buf.bufferpos += n
+    consumed!(buf, n)
     return buf
 end
 
@@ -176,12 +192,6 @@ function initbuffer!(buf::Buffer)
     return buf
 end
 
-# Copy marked data.
-function copymarked(buf::Buffer)
-    @assert buf.markpos > 0
-    return buf.data[buf.markpos:buf.marginpos-1]
-end
-
 # Take the ownership of the marked data.
 function takemarked!(buf::Buffer)
     @assert buf.markpos > 0
@@ -191,52 +201,43 @@ function takemarked!(buf::Buffer)
     return resize!(buf.data, sz)
 end
 
-# Insert data to the current buffer.
-function insertdata!(buf::Buffer, data::Ptr{UInt8}, nbytes::Integer)
+# Copy data from `data` to `buf`.
+function copydata!(buf::Buffer, data::Ptr{UInt8}, nbytes::Integer)
     makemargin!(buf, nbytes)
-    copy!(buf.data, buf.bufferpos + nbytes, buf.data, buf.bufferpos, buffersize(buf))
-    unsafe_copy!(bufferptr(buf), data, nbytes)
-    buf.marginpos += nbytes
+    unsafe_copy!(marginptr(buf), data, nbytes)
+    supplied!(buf, nbytes)
     return buf
 end
 
-# Read as much data as possbile from `input` to the margin of `output`.
-# This function will not block if `input` has buffered data.
-function readdata!(input::IO, output::Buffer)
-    nread::Int = 0
-    navail = nb_available(input)
-    if navail == 0 && marginsize(output) > 0 && !eof(input)
-        nread += writebyte!(output, read(input, UInt8))
-        navail = nb_available(input)
-    end
-    n = min(navail, marginsize(output))
-    Base.unsafe_read(input, marginptr(output), n)
-    output.marginpos += n
-    nread += n
-    output.total += nread
-    return nread
+# Copy data from `buf` to `data`.
+function copydata!(data::Ptr{UInt8}, buf::Buffer, nbytes::Integer)
+    # NOTE: It's caller's responsibility to ensure that the buffer has at least
+    # nbytes.
+    @assert buffersize(buf) ≥ nbytes
+    unsafe_copy!(data, bufferptr(buf), nbytes)
+    consumed!(buf, nbytes)
+    return data
 end
 
-# Read data from `buf` to `dst`.
-function readdata!(buf::Buffer, dst::Vector{UInt8}, dpos::Integer, sz::Integer)
-    copy!(dst, dpos, buf.data, buf.bufferpos, sz)
-    buf.bufferpos += sz
-    return dst
-end
-
-# Write all data to `output` from the buffer of `input`.
-function writebuffer!(output::IO, input::Buffer)
-    while buffersize(input) > 0
-        input.bufferpos += Base.unsafe_write(output, bufferptr(input), buffersize(input))
-    end
+# Insert data to the current buffer.
+function insertdata!(buf::Buffer, data::Ptr{UInt8}, nbytes::Integer)
+    makemargin!(buf, nbytes)
+    copy!(buf.data, buf.bufferpos + nbytes, buf.data, buf.bufferpos, buffersize(buf))
+    unsafe_copy!(bufferptr(buf), data, nbytes)
+    supplied!(buf, nbytes)
+    return buf
 end
 
 # Find the first occurrence of a specific byte.
 function findbyte(buf::Buffer, byte::UInt8)
-    ptr = ccall(:memchr, Ptr{Void}, (Ptr{Void}, Cint, Csize_t), pointer(buf.data, buf.bufferpos), byte, buffersize(buf))
-    if ptr == C_NULL
-        return 0
+    p = ccall(
+        :memchr,
+        Ptr{UInt8},
+        (Ptr{UInt8}, Cint, Csize_t),
+        pointer(buf.data, buf.bufferpos), byte, buffersize(buf))
+    if p == C_NULL
+        return marginptr(buf)
     else
-        return Int(ptr - pointer(buf.data, buf.bufferpos)) + buf.bufferpos
+        return p
     end
 end