longeststreak.scala

trait MapReduce[E, R] {
  type S
  def map(e: E): S
  def reduce(s1: S, s2: S): S
  def finish(s: S): R
}

def linear[E, R](seq: Array[E], mr: MapReduce[E, R]): R =
  mr.finish(seq.map(mr.map).reduceLeft(mr.reduce))

def divideAndConquer[E, R](seq: Seq[E], mr: MapReduce[E, R]): R = {
  def inner(seq: Seq[E]): mr.S =
    if (seq.size == 1)
      mr.map(seq(0))
    else if (seq.size < 5)
      seq.map(mr.map).reduceLeft(mr.reduce)
    else {
      val pivot = seq.size / 2
      val (s1, s2) = seq.splitAt(pivot)
      mr.reduce(inner(s1), inner(s2))
    }
  mr.finish(inner(seq))
}      

/**
 * A parallelizable version of an algorithm to compute the length
 * of the longest streak of elements satisfying a certain predicate.
 */
def longestStreakLength[E](pred: E => Boolean): MapReduce[E, Int] =
  new MapReduce[E, Int] {
    /*
     * State is either a tuple of 
     * (front, max, end) where front/end are the numbers
     * of matching elements found at the front/end and max
     * is the current maximum. Or it is just the number of elements
     * if it is just one long streak.
     */
    type S = Either[(Int, Int, Int), Int]
    def map(e: E): S = if (pred(e)) Right(1) else Left((0, 0, 0))
    def finish(s: S): Int = s match {
      case Left((_, max, _)) => max
      case Right(max) => max
    }
    def reduce(s1: S, s2: S): S = (s1, s2) match {
      case (Right(x), Right(y)) => Right(x + y)
      case (Left((v1, m1, h1)), Right(y)) => Left(v1, Math.max(m1, h1 + y), h1 + y)
      case (Right(x), Left((v2, m2, h2))) => Left(x + v2, Math.max(m2, x + v2), h2)
      case (Left((v1, m1, h1)), Left((v2, m2, h2))) => Left(v1,Math.max(Math.max(m1, m2), h1 + v2), h2)
    }
  }