ParallelCountChangeRunner

package reductions

import org.scalameter._

object ParallelCountChangeRunner {

  @volatile var seqResult = 0

  @volatile var parResult = 0

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

  def main(args: Array[String]): Unit = {
    val amount = 250
    val coins = List(1, 2, 5, 10, 20, 50)
    val seqtime = standardConfig measure {
      seqResult = ParallelCountChange.countChange(amount, coins)
    }
    println(s"sequential result = $seqResult")
    println(s"sequential count time: $seqtime ms")

    def measureParallelCountChange(threshold: ParallelCountChange.Threshold): Unit = {
      val fjtime = standardConfig measure {
        parResult = ParallelCountChange.parCountChange(amount, coins, threshold)
      }
      println(s"parallel result = $parResult")
      println(s"parallel count time: $fjtime ms")
      println(s"speedup: ${seqtime / fjtime}")
    }

    measureParallelCountChange(ParallelCountChange.moneyThreshold(amount))
    measureParallelCountChange(ParallelCountChange.totalCoinsThreshold(coins.length))
    measureParallelCountChange(ParallelCountChange.combinedThreshold(amount, coins))
  }
}

object ParallelCountChange {

  /**
   * Returns the number of ways change can be made from the specified list of
   *  coins for the specified amount of money.
   */
  def countChange(money: Int, coins: List[Int]): Int = {
    //hint: we either decide to continue subtracting the next coin in the coins list from the money amount,
    //or we decide to drop the coin from the list of coins
    if (money < 0) 0
    else if (money == 0) 1
    else if (money > 0 && coins.isEmpty) 0
    else (
        countChange(money - coins.head, coins)
        +
        countChange(money, coins.tail)
        )

  }

  type Threshold = (Int, List[Int]) => Boolean

  /**
   * In parallel, counts the number of ways change can be made from the
   *  specified list of coins for the specified amount of money.
   */
  def parCountChange(money: Int, coins: List[Int], threshold: Threshold): Int = {
    if (threshold(money, coins) || coins.isEmpty || money <= 0) countChange(money, coins)
    else {
      val pair = parallel(
        parCountChange(money - coins.head, coins, threshold),
        parCountChange(money, coins.tail,threshold))
      pair._1 + pair._2
    }
  }

  /** Threshold heuristic based on the starting money. */
  def moneyThreshold(startingMoney: Int): Threshold =
    (money, coins) =>
      money <= (2 * startingMoney) / 3

  /** Threshold heuristic based on the total number of initial coins. */
  def totalCoinsThreshold(totalCoins: Int): Threshold = (money, coins) => coins.size <= (2 * totalCoins) / 3

  /** Threshold heuristic based on the starting money and the initial list of coins. */
  def combinedThreshold(startingMoney: Int, allCoins: List[Int]): Threshold = {
      (money,coins) => money * coins.size <= (startingMoney*allCoins.size)/2

      /* credit TomLous looks nice but yields a test error with message:
       * combinedThreshold should return true when the number of coins times money is
       * less than or equal to half of the initial number of coins times starting money */
    //(money,coins) => moneyThreshold(startingMoney)(money,coins) && totalCoinsThreshold(allCoins.size)(money,coins)

  }
}