object ParCollectionFun extends App {

  import scala.collection.par._
  import scala.util.Random

  import Scheduler.Implicits.global
//  import Scheduler.Implicits.sequential


  def seq_sum(n: Int) = (0 until n).reduce(_ + _)

  def par_sum(n: Int) = (0 until n).toPar.reduce(_ + _)

  val n = 15000000

  def time[A](a: => A) = {
    val now = System.nanoTime
    val result = a
    val micros = (System.nanoTime - now) / 1000
      println("%f seconds".format(micros/1.0e6))
    result
  }

  def fib(n: Int): Long = if (n<=2) 1 else fib(n-1) + fib(n-2)

  def sumFib: Long = {
    val nvalus = (1 until 47).toArray


    val fibValus = nvalus.toPar.map(fib)

    fibValus.reduce (_ + _)

  }

  def sumFibFuture: Long = {
    import scala.concurrent.{Future, Await}
    import scala.concurrent.duration._
    import scala.concurrent.ExecutionContext.Implicits.global

    val nvalus = (1 to 47).toSeq

    val fibValus = Future.sequence(nvalus map { case n => Future { fib(n) } })
    val sumFuture = fibValus.map {case vals => vals.sum}

    Await.result(sumFuture, Duration.Inf)
  }

  def seq_qsort(xs: Vector[Int]): Vector[Int] = {
    def qsort_(xs: Vector[Int]): Vector[Int] =
      if (xs.length <= 1000) xs.sorted else {
        val pi = xs.hashCode.abs % xs.length
        val pivot = xs(pi)

        val less = xs.filter { case elt => elt < pivot }
        val eq = xs.filter { case elt => elt == pivot }
        val greater = xs.filter { case elt => elt > pivot }

        qsort_(less) ++ eq ++ qsort_(greater)
      }

    qsort_(xs)
  }

  def par_qsort_future(xs: Vector[Int]): Vector[Int] = {
    import scala.concurrent.{Future, Await}
    import scala.concurrent.duration._
    import scala.concurrent.ExecutionContext.Implicits.global
    def qsort_(xs: Vector[Int]): Vector[Int] =
      if (xs.length <= 10000) seq_qsort(xs) else {
        val pi = xs.hashCode.abs % xs.length
        val pivot = xs(pi)

        val less = Future { qsort_(xs.filter { case elt => elt < pivot }) }
        val eq = xs.filter { case elt => elt == pivot }
        val greater = Future { qsort_(xs.filter { case elt => elt > pivot }) }

        val sorted = for { lt <- less
          ge <- greater } yield lt ++ eq ++ ge

        Await.result(sorted, Duration.Inf)
      }

    qsort_(xs)
  }


  def par_qsort(xs: Vector[Int]): Vector[Int] = {
    def qsort_(xs: Array[Int]): Array[Int] =
      if (xs.length <= 10000) seq_qsort(xs.toVector).toArray  else {
        val pi = xs.hashCode.abs % xs.length
        val pivot = xs(pi)

        val less = xs.filter { case elt => elt < pivot }
        val eq = xs.filter { case elt => elt == pivot }
        val greater = xs.filter { case elt => elt > pivot }


        val ans = Array(less, greater).map {case toSort => qsort_(toSort)}

        ans(0) ++ eq ++ ans(1)
      }

    qsort_(xs.toArray).toVector
  }

  val a = time { seq_sum(n) }
  println(s"seq: $a")

  val b = time { par_sum(n) }
  println(s"par: $b")

  println("creating a permutation")
  val in = time { Random.shuffle((1 to 10000000).toVector) }

  println("sorting: seq")
  time {seq_qsort(in)}
  println("sorting: future")
  time {par_qsort_future(in)}
  println("sorting: blitz")
  time {par_qsort(in)}

//  val c = time { sumFib }
//  println(s"sumFib: $c")

//  val d = time { sumFibFuture }
//  println(s"sumFibFuture: $d")


}