lineOfSight

package reductions

import org.scalameter._
import common._
import scala.collection.par.Conc.Prepend

object LineOfSightRunner {

  val standardConfig = config(
    Key.exec.minWarmupRuns -> 40,
    Key.exec.maxWarmupRuns -> 80,
    Key.exec.benchRuns -> 100,
    Key.verbose -> true) withWarmer (new Warmer.Default)

  def main(args: Array[String]) {
    val length = 10000000
    val input = (0 until length).map(_ % 100 * 1.0f).toArray
    val output = new Array[Float](length + 1)
    val seqtime = standardConfig measure {
      LineOfSight.lineOfSight(input, output)
    }
    println(s"sequential time: $seqtime ms")

    val partime = standardConfig measure {
      LineOfSight.parLineOfSight(input, output, 10000)
    }
    println(s"parallel time: $partime ms")
    println(s"speedup: ${seqtime / partime}")
  }
}

object LineOfSight {

  def max(a: Float, b: Float): Float = if (a > b) a else b

  def lineOfSight(input: Array[Float], output: Array[Float]): Unit = {
    /*Using scanLeft
        val out = input.scanLeft(0f)((a, b) => max(a, b / max(1, (input.indexOf(b)))))
          var i = 0
          var j = 0
          while (i < input.length) {
            output.update(i, out(j + 1))
            i += 1
            j += 1
          }
    */

    /*or re-implementing scanLeft*/
    output(0) = 0f
    var a = 0f
    var i = 0
    while (i < input.length - 1) {
      a = max(a, input(i + 1) / max(1, (input.indexOf(input(i + 1)))))
      println(a)
      i += 1
      output(i) = a
    }

    /* elegant solution from Tom Lou thanks
     * input.zipWithIndex.foreach{
      case (xs, 0) => output(0) = 0
      case (xs, i) => output(i) = Math.max(xs / i, output(i-1))
    }
     */
  }

  sealed abstract class Tree {
    def maxPrevious: Float
  }

  case class Node(left: Tree, right: Tree) extends Tree {
    val maxPrevious = max(left.maxPrevious, right.maxPrevious)
  }

  case class Leaf(from: Int, until: Int, maxPrevious: Float) extends Tree

  /**
   * Traverses the specified part of the array and returns the maximum angle.
   */
  def upsweepSequential(input: Array[Float], from: Int, until: Int): Float = {
    /* elegant solution thanks Tom Lou */
    (from until until).foldLeft(0f)((currentMax, i) => max(input(i) / i, currentMax))
  }

  /**
   * Traverses the part of the array starting at `from` and until `end`, and
   *  returns the reduction tree for that part of the array.
   *
   *  The reduction tree is a `Leaf` if the length of the specified part of the
   *  array is smaller or equal to `threshold`, and a `Node` otherwise.
   *  If the specified part of the array is longer than `threshold`, then the
   *  work is divided and done recursively in parallel.
   */
  def upsweep(input: Array[Float], from: Int, end: Int, threshold: Int): Tree = {
    if (end - from <= threshold)
      Leaf(from, end, upsweepSequential(input, from, end))
    else {
      val mid = from + (end - from) / 2
      val (left, right) = parallel(
        upsweep(input, from, mid, threshold),
        upsweep(input, mid, end, threshold))
      Node(left, right)
    }
  }

  /**
   * Traverses the part of the `input` array starting at `from` and until
   *  `until`, and computes the maximum angle for each entry of the output array,
   *  given the `startingAngle`.
   */
  def downsweepSequential(input: Array[Float], output: Array[Float], startingAngle: Float,
    from: Int, until: Int): Unit = {
    // thanks Tom Lou
    def maxAngle(start: Int, end: Int, previousMax: Float) {
      if (start < end) {
        output(start) = max(previousMax, input(start) / start)
        maxAngle(start + 1, end, output(start))
      }
    }
    maxAngle(from, until, startingAngle)
  }

  /**
   * Pushes the maximum angle in the prefix of the array to each leaf of the
   *  reduction `tree` in parallel, and then calls `downsweepSequential` to write
   *  the `output` angles.
   */
  def downsweep(input: Array[Float], output: Array[Float], startingAngle: Float, tree: Tree): Unit = {
    tree match {
      case Leaf(from, until, _) => downsweepSequential(input, output, startingAngle, from, until)
      case Node(left, right) => {
        parallel(
          downsweep(input, output, startingAngle, left),
          downsweep(input, output, max(startingAngle, left.maxPrevious), right))
      }
    }
  }

  /** Compute the line-of-sight in parallel. */
  def parLineOfSight(input: Array[Float], output: Array[Float], threshold: Int): Unit = {
    val tree = upsweep(input, 0, input.length, threshold)
    downsweep(input, output, 0f, tree)
  }
}

/* FYI AUTO GRADER OUTPUT. Your solution achieved a testing score of 137 out of 150.

[Test Description] parLineOfSight should invoke the parallel construct 30 times (15 times during upsweep and 15 times during downsweep) for an array of size 17, with threshold 1
[Observed Error] 32 did not equal 30 (The number of parallel calls should be 30)
[Lost Points] 3

[Test Description] parLineOfSight should correctly compute the output for threshold 3
[Observed Error] List(NaN, 9.0, 9.0, 9.0, 9.0, 9.0) did not equal List(0.0, 9.0, 9.0, 9.0, 9.0, 9.0)
[Lost Points] 2

[Test Description] parLineOfSight should call parallel constuct 6 times, where the last two parallel constructs should update the 4 sections of the array (1 until 5), (5 until 9), (9 until 13), (13 until 17), respectively
[Observed Error] success was false [During the execution of first part of 5th call to parallel construct, the indices 1 until 5 of the output array was not correctly updated]
[Lost Points] 6

[Test Description] parLineOfSight should correctly compute the output for threshold 2
[Observed Error] List(NaN, 7.0, 7.0, 11.0, 12.0) did not equal List(0.0, 7.0, 7.0, 11.0, 12.0)
[Lost Points] 2

*/