Some informal scalaz-stream performance testing code.

profile.scala

package scalaz.stream

import scalaz.concurrent.Task
import scalaz.syntax.traverse._
import scalaz.std.list._
import Process.Process1

import collection.immutable.{IndexedSeq, Vector}

object profile extends App {
  
  def time(block: => Any): Double = {
    // for (i <- 1 to 1600) block // trigger JIT on block
    var goodSample = false
    var start = 0L; var stop = 0L;
    var N = 3
    while (!goodSample) {
      start = System.currentTimeMillis
      for (i <- 0 until N) block
      stop = System.currentTimeMillis
      if (stop - start < 5000) N = N*2 // require at least 5 seconds for a decent sample
      else goodSample = true
    }
    val perOp = (stop.toDouble - start.toDouble) / N
    perOp / 1000
  }

  def printTime(label: String)(block: => Any): Unit = 
    println(label + ": " + time(block))

  val N = 10000

  val nums: Array[Double] = 
    Array.tabulate(N)(_.toDouble)

  val numsVec: Vector[Double] = 
    Vector(nums: _*) 

  val ranges: Vector[(Int,Int)] = 
    Vector() ++ (0 to N by 2000).sliding(2).map { case Seq(a,b) => (a,b) }

  val chunkedNumsVec: Vector[(Array[Double], Int, Int)] = 
    ranges.map { case (start, stop) => (nums, start, stop) }

  val slicedNumsVec: Vector[IndexedSeq[Double]] = 
    ranges.map { case (start, stop) => Slice(nums, start, stop) }

  case class Slice[@specialized A](
      overall: Array[A], 
      start: Int, 
      stop: Int) extends IndexedSeq[A] {
    def apply(i: Int): A = overall(start + i) 
    def length = stop - start
  }

  def sumChunk(vs: IndexedSeq[Double]): Double = vs match {
    case Slice(vs: Array[Double], i, j) => sum(vs, i, j)
    case _ => sumOf(vs)
  }

  def chunkedReduce[A](v: Vector[IndexedSeq[A]])(f: IndexedSeq[A] => A): A =
    f(v map f)

  def sum(ds: Array[Double], start: Int, stop: Int): Double = {
    var total = 0.0
    var i: Int = 0
    while (i < stop) { total += ds(i); i += 1 }
    total
  }

  def sumOf(ds: TraversableOnce[Double]): Double = {
    var total = 0.0
    ds.foreach { d => total += d }
    total
  }
  
  println("N: " + nums.length)

  printTime("unboxed-in-memory") { 
    sum(nums, 0, nums.length) 
  }
  printTime("boxed-in-memory") { 
    numsVec.sum 
  }
  printTime("chunked-boxed-in-memory") { 
    sumOf(numsVec) 
  }
  printTime("deep-chunked-boxed-in-memory") { 
    chunkedReduce(slicedNumsVec)(sumChunk) 
  }
  printTime("naive-streaming-overhead") { 
    Process.fill(N)(1).run.run 
  }

  printTime("chunked-streaming") { 
    Process.ranges(0, N, 20000).
            map { case (i,j) => Slice(nums, i, j) }.
            map(sumChunk).
            pipe(process1.sum[Double]).
            run.run
  }
}