def readAtOffsets(
fileName: String,
offsetLengths: List[(Long, Int)]
): Stream[IO, Array[Byte]] = {
val cursor =
Blocker[IO].flatMap(ReadCursor.fromPath[IO](Paths.get(fileName), _))
Stream.resource(cursor).flatMap { cursor =>
Stream
.emits(offsetLengths)
.flatMap {
case (offset, length) =>
cursor
.seek(offset)
.readUntil(length, offset + length)
.void
.stream
.chunks
.map(_.toArray)
}
}
}
I did a benchmark vs the simple:
def readAtOffsetsSimple(fileName: String, offsetLengths: Seq[(Long, Int)]): Unit = {
val fc = FileChannel.open(Paths.get(fileName), StandardOpenOption.READ)
for {
(offset, length) <- offsetLengths
} yield {
val buf = ByteBuffer.allocate(length)
val len = fc.read(buf, offset)
if (len < 0) None
else Some(buf, len)
}
fc.close()
}The stream solution for 1 million random 100 byte reads in a 700 Mb file gave an average of just under 3 minutes. The simple version was about 4 seconds a factor of ~45
Another question, how best to debug stream applications. Obviously there is a lot of machinery under the hood (I stepped a simple stream all the way through). Sadly auto-stepping and reading a report doesn't exist on intellij (or does it via some plugin)?
When I've debugged systems like this before, I've used stepping filters to cut out all the machinery. I looks like breakpoints on the code portions also work, though that's clunky to step. Any suggestions?
One further question: Does fs2 support memoisation?
If I pass my stream elements through f(s) and if f is a pure function f(s) for a particular s should return the same value each time. Thus if f is a big computation, f only needs running once. If f has side effects but are captured, and it's okay to rerun the captured side effects each time, again, can that be memoised to avoid having to run f each time ?
val resource = Stream.resource(ReadCursor.fromPath[IO](Paths.get(fileName), blocker))
val outStream = offsetLengths.foldLeft(Stream.empty.covaryAll[IO, Array[Byte]]) { case (streamAcc, (offset, length)) =>
streamAcc ++ resource.flatMap {
_.seek(offset).readUntil(length, offset + length).void.stream.mapChunks { c => Chunk(c.toArray) }
}
}resource
streamAcc ++ Stream.resource(ReadCursor.fromPath...).flatMapreadPath program is happening on each iteration
debug and debugChunks on Stream, and we're in the process of adding tracing to IO, to help with the remaining cases
https://gist.github.com/ibanezn04/15429e56b9e19184c69b752b97eddd1d
@ibanezn04 this might help. It's an implementation of unzipping using 7zip because JDK fails on some newer ZIP files. Unfortunately 7zip needs random seek access, so you have to buffer the zip file either to memory or disk.
https://gist.github.com/arturaz/9a45a8b3807fc625e4feb006b3a94ab1
And I'm not sure, but maybe this issue could help functional-streams-for-scala/fs2#1822
NonProgressiveDecompressionException: buffer size 4096 is too small; gunzip cannot make progress
s.map(left).parZip(s.map(right)) ?
(left, right).product?
left.product(right)
@ val a: Int => String = (i: Int) => i.toString
a: Int => String = ammonite.$sess.cmd2$$$Lambda$1964/810898134@7c5f29c6
@ val b: Int => Int = (i: Int) => i * 2
b: Int => Int = ammonite.$sess.cmd3$$$Lambda$1973/33299582@483b7dc4
@ a.product(b)
res4: Int => (String, Int) = cats.instances.Function1Instances$$anon$7$$Lambda$1984/543938464@de63949
@ res4(3)
res5: (String, Int) = ("3", 6)