Improving performance of fs2 stream involving file transformation

Question

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.

Answer 1

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):

import cats.effect.{Blocker, ExitCode, IO, IOApp, Resource}
import fs2.{io, text, Stream, Pipe, Chunk}
import java.nio.file.Paths
import java.nio.charset.Charset

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)

    def betterUtf8Encode[F[_]]: Pipe[F, String, Byte] =
      _.mapChunks(c => c.flatMap(s => Chunk.bytes(s.getBytes(Charset.forName("UTF-8")))))

    io.file.readAll[IO](Paths.get("testdata/fahrenheit.txt"), blocker, 4096)
      .through(text.utf8Decode)
      .through(text.lines)
      .filter(s => !s.trim.isEmpty && !s.startsWith("//"))
      .map(line => fahrenheitToCelsius(line.toDouble).toString)
      .intersperse("\n")
      // .through(text.utf8Encode) // didn't finish, could be an hour
      .through(betterUtf8Encode) // 2 minutes
      .through(io.file.writeAll(Paths.get("testdata/celsius.txt"), blocker))
  }

  def run(args: List[String]): IO[ExitCode] =
    converter.compile.drain.as(ExitCode.Success)
}

It's a difference between 2 minutes and possibly an hour or more, in this case...