mpilquist/scodec — pull request #4 by pchiusano (2014-02-04 23:08:38+00:00)
Title: Support for streaming (lazy I/O) backed BitVectors

Added a new constructor to `BitVector`, `Suspend`, making it possible for `BitVector` to represent lazy, right associated cons lists, suitable for streaming applications. Also added `BitVector.sizeLessThan`, which only computes as much of `size` as is needed to return `true` or `false`. This is being used in `take` and `drop` to ensure that these don't do unnecessary forcing of streaming `BitVector` values.

Next, I added several functions:

* `BitVector.unfold`, a generic combinator for producing streaming bit vectors
* `BitVector.fromInputStream`, `BitVector.fromChannel`, and `BitVector.fromMmap`, for producing streaming bit vectors from `java.io` and `java.nio` sources. These just call `unfold`.

The implementations are pretty simple:

```Scala
/**
* Create a lazy `BitVector` by repeatedly extracting chunks from `S`.
* The returned `BitVector` will have the structure of a fully lazy
* right-associated cons list. Thus, `get`, `take`, and `drop` will
* be efficient when operating on the head of the list, but accessing
* later indices (for `takeRight`, say, or `get(size-1)` will require
* forcing the stream up to that point.
*
* Use `force` if you wish to convert the result to an in-memory strict
* `BitVector` backed by a balanced tree.
*/
def unfold[S](s: S)(f: S => Option[(BitVector, S)]): BitVector =
Suspend { () => f(s).map { case (h,t) => Append(h, unfold(t)(f)) }
.getOrElse { BitVector.empty } }

/**
* Produce a lazy `BitVector` from the given `InputStream`, using `chunkSizeInBytes`
* to control the number of bytes read in each chunk (defaulting to 4MB).
* This simply calls [[scodec.BitVector.unfold]] with a function to extract a series
* of flat byte arrays from the `InputStream`.
*
* This function does not handle closing the `InputStream` and has all the usual
* drawbacks of lazy I/O - `I/O` exceptions may be raised unexpectedly in pure code as
* chunks are forced, and it must memoize the results to prevent the underlying side
* effects from being observed. Streaming applications should take care to ensure
* that the head of the stream is not left on the stack, as this will cause the entire
* stream to be retained in memory.
*
* @param chunkSizeInBytes the number of bytes to read in each chunk
*/
def fromInputStream(in: java.io.InputStream, chunkSizeInBytes: Int = 4096 * 2): BitVector =
unfold(in) { in =>
val buf = new Array[Byte](chunkSizeInBytes)
val nRead = in.read(buf)
if (nRead == chunkSizeInBytes) Some((BitVector(buf), in))
else if (nRead == -1) None
else Some((BitVector(buf.take(nRead): Array[Byte]), in))
}

/**
* Produce a lazy `BitVector` from the given `ReadableByteChannel`, using `chunkSizeInBytes`
* to control the number of bytes read in each chunk (defaulting to 4k). This function
* does lazy I/O, see [[scodec.BitVector.fromInputStream]] for caveats.
*
* @param chunkSizeInBytes the number of bytes to read in each chunk
*/
def fromChannel(in: java.nio.channels.ReadableByteChannel, chunkSizeInBytes: Int = 4096 * 2): BitVector =
unfold(in) { in =>
val buf = java.nio.ByteBuffer.allocate(chunkSizeInBytes)
val nRead = in.read(buf)
buf.flip
if (nRead == chunkSizeInBytes) Some((BitVector.view(buf), in))
else if (nRead == -1) None
else Some((BitVector(ByteVector.view(ind => buf.get(ind), nRead)), in))
}

/**
* Produce a lazy `BitVector` from the given `FileChannel`, using `chunkSizeInBytes`
* to control the number of bytes read in each chunk (defaulting to 4MB). Unlike
* [[scodec.BitVector.fromChannel]], this memory-maps chunks in, rather than copying
* them explicitly.
*
* Behavior is unspecified if this function is used concurrently with the underlying
* file being written.
*
* @param chunkSizeInBytes the number of bytes to read in each chunk
*/
def fromMmap(in: java.nio.channels.FileChannel, chunkSizeInBytes: Int = 1024 * 1000 * 4): BitVector =
unfold(in -> 0L) { case (in,pos) =>
if (pos == in.size) None
else {
val bytesToRead = (in.size - pos) min chunkSizeInBytes.toLong
val buf = in.map(java.nio.channels.FileChannel.MapMode.READ_ONLY, pos, bytesToRead)
Some((BitVector.view(buf), (in -> (pos + bytesToRead))))
}
}

```

The `BitVector` scalacheck generators have been updated to occasionally produce `unfold`-based `BitVector`s and all tests pass.

Lastly, I have added an integration 'test', `StreamingBitVectorTest`. It is not really a proper test, more of an example that can be run as a sanity check. Try `sbt test:run` to execute it. Here is sample output from a run, which also shows throughput for processing a 408 MB file:

```
Simple integration test for streaming, lazy-I/O-backed `BitVector`.
When viewing output, we are looking for two things:

* Speed should be "reasonable", between 10-100 MB/s if touching
each byte, and up to 4 GB/s or so if we are just measuring time
to load the series of chunks.
* Memory usage should be bounded by the chunk size, not by overall
file size. That is, the garbage collector should be able to
reclaim previous chunks when the stream is traversed in a linear
fashion. Try running with -Xmx100m to verify this.

Processing file of size: 408.504702 MB
Touching each byte read: false
Max memory: 139.984896 MB
Total memory: 135.266304 MB
Free memory: 16.01076 MB
BitVector.fromInputStream took 5.59 seconds, 73 MB/s
Max memory: 139.984896 MB
Total memory: 135.266304 MB
Free memory: 19.906768 MB
BitVector.fromChannel took 0.461 seconds, 886 MB/s
Max memory: 139.984896 MB
Total memory: 135.266304 MB
Free memory: 26.234992 MB
BitVector.fromMmap took 0.089 seconds, 4590 MB/s
closing files... done
Processing file of size: 408.504702 MB
Touching each byte read: true
Max memory: 139.984896 MB
Total memory: 135.266304 MB
Free memory: 37.559176 MB
BitVector.fromInputStream took 8.97 seconds, 46 MB/s
Max memory: 139.984896 MB
Total memory: 135.266304 MB
Free memory: 30.380944 MB
BitVector.fromChannel took 7.346 seconds, 56 MB/s
Max memory: 139.984896 MB
Total memory: 135.266304 MB
Free memory: 27.25544 MB
BitVector.fromMmap took 7.382 seconds, 55 MB/s
closing files... done
```

Memory-mapping the file is by far the fastest if you are not planning to examine every byte, since the OS is just setting up the mapping without necessarily copying anything. If you are examining each byte, it seems to be kind of a wash, I suspect because you end up being CPU bound, though I haven't done any real profiling. One thing I have noticed is that memory mapping seems to be less sensitive to the choice of chunk size, which would make sense so long as your chunk size is bigger than a page. With standard `java.io`, I was seeing around 2x difference in performance depending on what chunk size I chose. It might be valuable later to do some real profiling and tuning of this later on. NB: chunk size is a parameter to the `from` methods, so users can experiment with this in their own code and figure out what works best for their use case.

I believe these are all the changes needed to the scodec core to support adding a safe, streaming parsing/serialization layer based on scalaz-stream!