I've got something like this (it's an example from https://github.com/typelevel/fs2, with my additions, which I marked with comments):
import cats.effect.{Blocker, ExitCode, IO, IOApp, Resource}
import fs2.{io, text, Stream}
import java.nio.file.Paths
object Converter extends IOApp {
val converter: Stream[IO, Unit] = Stream.resource(Blocker[IO]).flatMap { blocker =>
def fahrenheitToCelsius(f: Double): Double =
(f - 32.0) * (5.0/9.0)
io.file.readAll[IO](Paths.get("testdata/fahrenheit.txt"), blocker, 4096)
.balanceAvailable // my addition
.map ( worker => // my addition
worker // my addition
.through(text.utf8Decode)
.through(text.lines)
.filter(s => !s.trim.isEmpty && !s.startsWith("//"))
.map(line => fahrenheitToCelsius(line.toDouble).toString)
.intersperse("\n")
.through(text.utf8Encode)
.through(io.file.writeAll(Paths.get("testdata/celsius.txt"), blocker))
) // my addition
.take(4).parJoinUnbounded // my addition
}
def run(args: List[String]): IO[ExitCode] =
converter.compile.drain.as(ExitCode.Success)
}
If fahrenheit.txt
is as big as eg. 300mb the execution of the original code takes several minutes. It appears that my code is not any faster. How can I improve its performance? There is plenty of unused CPU power when it runs, the disc is SSD, so I don't know why it's so slow. I'm not sure if I'm using balance
correctly.
The culprit is
text.utf8Encode
which unnecessarily emits one chunk per line. It's a huge overhead when there is a lot of short lines, like in the example (one temperature value per line, 108199750 lines). It was solved recently (Pull request: https://github.com/typelevel/fs2/pull/2096). Below I provide an inline solution, based on this PR (useful as long as someone uses the version without this fix):It's a difference between 2 minutes and possibly an hour or more, in this case...