Slow performance compared to BufferedReader

[Alnaimi-]

I wanted to test the performance of FS2 and was somewhat surprised by how much vanilla Java and Scala functions outperformed FS2. I have attached a simple test of reading and counting the number of lines below.

Code to reproduce:

import cats.effect.Clock
import cats.effect.IO
import cats.effect.SyncIO
import cats.effect.kernel.Async
import cats.effect.kernel.Resource
import fs2.io.{file => fs2_file}

import java.nio.ByteBuffer
import java.nio.channels.FileChannel
import java.nio.file.Files
import java.nio.file.Path
import java.nio.file.StandardOpenOption
import java.util.stream.Collectors
import scala.concurrent.Await
import scala.concurrent.Future
import scala.concurrent.duration.Duration
import scala.concurrent.duration.FiniteDuration
import scala.util.Random
import scala.util.Using

class ReadFileTest extends munit.CatsEffectSuite {
  val n            = 5_000_000
  val C: Async[IO] = Clock[IO]

  override def munitTimeout: Duration = FiniteDuration(10, "min")

  val file: SyncIO[FunFixture[Path]] = ResourceFixture(Resource.make(C.timed(makeRandomCSV).flatMap {
    case (time, p) => IO(println(time)) *> IO(p)
  })(path => IO(Files.delete(path))))

  private def makeRandomCSV =
    IO {
      val r     = new Random()
      val value = Files.createTempFile("blergo_blargo", "read-test")

      // fill the file with random data
      Using(FileChannel.open(value, StandardOpenOption.APPEND)) { chan =>
        for (_ <- 0 until n) {
          val str = r.nextString(r.nextInt(256)).replaceAll("\n", "").replaceAll("\r", "").trim
          chan.write(
                  ByteBuffer.wrap(
                          s"${r.nextLong()},${r.nextBoolean()},$str,${r.nextGaussian()}\n"
                            .getBytes()
                  )
          )
        }
      }
      value
    }

  file.test(s"read and count $n lines in a csv file") { path =>
    // scala Source
    val scalaSource = C
      .timed {
        IO {
          val count = Using(scala.io.Source.fromFile(path.toFile)) { src =>
            Await.result(
                    Future
                      .sequence(src.getLines().grouped(1024).map { strings =>
                        Future(strings.map(_.split(",")).map(_ => 1).sum)
                      })
                      .map(_.sum),
                    FiniteDuration(10, "min")
            )
          }.get

          assertEquals(count, n)
        }
      }
      .map { case (t, v) => println(s"SCALA SOURCE ${t.toMillis}ms $v") }

    // java
    val javaStream  = C
      .timed {
        IO {
          Using(
                  Files
                    .newBufferedReader(path)
          ) { reader =>
            reader
              .lines()
              .parallel()
              .map(_.split(","))
              .map(_ => 1)
              .collect(Collectors.counting[Int]())
          }.get
        }
      }
      .map { case (t, v) => println(s"JAVA STREAM ${t.toMillis}ms $v") }

    val fs2Stream   = C
      .timed(
              fs2_file
                .Files[IO]
                .readUtf8Lines(fs2_file.Path.fromNioPath(path))
                .chunks
                .parEvalMapUnordered(16) { chunk =>
                  IO(chunk.map(_.split(",")).map(_ => 1).foldLeft(0)(_ + _))
                }
                .reduce(_ + _)
                .head
                .compile
                .toList
      )
      .map { case (t, v) => println(s"FS2 Stream ${t.toMillis}ms $v") }

    // Test performance a few times
    javaStream *> scalaSource *> fs2Stream *>
      javaStream *> scalaSource *> fs2Stream *>
      javaStream *> scalaSource *> fs2Stream *>
      javaStream *> scalaSource *> fs2Stream *>
      javaStream *> scalaSource *> fs2Stream
  }
}

For no. of lines in file = 5_000_000

JAVA STREAM 5353ms 5000000
SCALA SOURCE 5866ms ()
FS2 Stream 9036ms List(5000001)

This doesn't seem to scale any better for larger files either. Am I doing something wrong?

[nikiforo]

To count the number of lines I would use

def countLines(path: JPath): IO[Long] =
    Files[IO]
      .readUtf8Lines(Path.fromNioPath(path))
//      .prefetch
      .compile
      .count

If I wanted to count "," I wound use

  def count(path: JPath): IO[Int] =
    Files[IO]
      .readUtf8(Path.fromNioPath(path))
      .map(_.count(_ == ','))
//      .prefetch
      .compile
      .foldMonoid

In both versions I would measure execution with and without prefetch.

If I wanted to optimize reading from file I would also use def readAll(path: Path, chunkSize: Int, flags: Flags) and checked various ChunkSize values.

Also, Source.getLines() is not equivalent to Files[IO].readUtf8Lines. The latter reads only 64*1024 bytes in memory at any given moment(i.e. doesn't depend on file size) whereas the former reads ALL the file content into the memory.