-- |
-- Module      :  MergeSort
-- Copyright   :  (c) Philipps Universitaet Marburg 2004-2015
-- License     :  BSD-style (see the file LICENSE)
-- 
-- Maintainer  :  eden@mathematik.uni-marburg.de
-- Stability   :  beta
-- Portability :  not portable
--
-- The following Haskell module implements parallel mergesort with Eden.
--
-- Depends on the Eden Compiler.
--
-- Eden Project

{- Case study: merge sort

-- Arguments: 1. skeleton to be used 
--               (options: seq, mapRed, odisDC, trivPar)   
--            2. length of input list  (random numbers)
--            3. chunk size for result lists
  Output is suppressed
-}

module Main where

import System.Environment
import Control.Seq
import Control.Parallel (pseq)
import Control.Parallel.Eden
import Control.Parallel.Eden.Auxiliary
import Control.Parallel.Eden.DivConq 
import Control.Parallel.Eden.Map
import System.Random
import Data.List
import ParRed 

main :: IO ()
main = do ins <- getArgs 
          if length ins < 2 then print usage else 
           do let (v:a:xs) = ins
              let len      = read a :: Int
              let chunksize = if null xs then 1000 else read (head xs) :: Int
              let seed = if null xs || null (tail xs) then 42 else read (head (tail xs)) :: Int
              let rs = randomlist len seed 
              putStrLn (rnf rs `pseq` rnf (ms v rs len chunksize noPe) `pseq` "Done")
-- use this line, if you want to print the result list
--              putStrLn (show (ms v rs len chunksize seed))

usage :: String
usage =  "Use 2-4 parameters: version (seq/parMap/disDC/flatDC/parRed/psrs), list length, " ++ 
         "chunk size (optional, default=1000), seed (optional, default=42)"

-- generating list of random numbers 
randomlist :: Int -> Int -> [Int]
randomlist n seed = take n $ randoms $ mkStdGen seed




ms :: String -> [Int] -> Int -> Int -> Int -> [Int]
-- sequential mergeSort
ms "seq" xs _ _ _ =  mergeSort xs
-- parMap with chunking
ms "parMap" xs n d p  
  = mergeAll $ map concat $ parMap (chunk d.mergeSort.concat) (map (chunk d) (unshuffle (p-1) xs))
-- divide and conquer: distributed expansion 
ms "disDC" xs n d p
--  = concat $ disDC 2 [2..p] triv solve split combine (chunk d xs) -- does not work with ghc-7.6.2
  = concat $ dcNTickets_c 2 [2..p] triv solve split combine (chunk d xs)
  where
   split    =  unshuffle 2   
   threshold = n `div` p
   triv   xss = length (concat xss) < threshold -- null xs || null (tail xs)  
   solve  xss = (chunk d) . mergeSort .concat $ xss        -- chunk d xs
   combine _ (b1:b2:_) = sortMergeChunk d b1 b2
-- divide and conquer: flat expansion with parMap skeleton
ms "flatDC" xs n d p 
  = concat $ flatDC parMap depth triv solve split combine (chunk d xs)
  where
   depth    = floor ((log (fromIntegral p)) / log 2) :: Int
   split     = unshuffle 2   
   threshold = n `div` p
   triv  xss = length (concat xss)  < threshold -- null xs || null (tail xs)  
   solve xss = (chunk d) . mergeSort . concat $ xss 
   combine _ (b1:b2:_) = sortMergeChunk d b1 b2
-- using parallel reduction and skeleton composition
ms "parRed" xs n d p  = concat . fetch . 
                        (parRedAt [1..p] (sortMergeChunk d) []) . 
                        (parMapAt [1..p] (release.(chunk d).mergeSort.concat)) 
                        $ (map (chunk d) (unshuffle p xs))  
-- PSRS
ms "psrs" xs _ d p = psrs (p-1) d xs 


  
  
-- sequential mergeSort function
mergeSort      :: Ord a => [a] -> [a]
mergeSort xs@(_:_:_)   =  sortMerge (mergeSort xs1)
                                    (mergeSort xs2)
  where [xs1,xs2] = unshuffle 2 xs
mergeSort []   =  []
mergeSort [x]  =  [x]

-- merging two sorted lists  
sortMerge :: Ord a => [a] -> [a] -> [a]
sortMerge xlist@(x:xs) ylist@(y:ys)
   | x <= y = x : sortMerge xs ylist
   | x >  y = y : sortMerge xlist ys
sortMerge []           ylist        = ylist
sortMerge xlist        []           = xlist


-- merge list of sorted lists
mergeAll ::  Ord a => [[a]] -> [a]
mergeAll [xs] = xs
mergeAll xss  = mergeAll (mergePairs xss)

mergePairs :: Ord a => [[a]] -> [[a]]
mergePairs (xs1:xs2:xss) = (ys : zss)
  where ys  = sortMerge xs1 xs2
        zss = mergePairs xss 
mergePairs xs = xs

-- sortMerge for chunked sorted lists
sortMergeChunk :: Ord a => Int -> [[a]] -> [[a]] -> [[a]]
sortMergeChunk size xss yss = chunk size (sortMerge (concat xss) (concat yss))

-- Box definition, used to suppress streaming by putting an input list into a box
newtype Box a = Box {unbox :: a} 
instance NFData a => NFData (Box a)
 where rnf (Box x) = rnf x 
instance Trans a => Trans (Box a) 
 


-- PSRS   
psrs :: (Trans a, Ord a) => Int -> Int -> [a] -> [a]
psrs p d xs = concat . concat $ results 
 where 
 -- samples :: [Box [a]], rdys :: [Rd ([ [a]])] 
 (samples, rdys)    = unzip $ parMapAt [2..p+1] (\ css -> let ys = sort (concat css) 
                                                             in (Box $ getSamples p ys, release (chunk d ys)))   
                                              (map (chunk d)  (unshuffle p xs))
 globalSamples      = getGlobalSamples p $ mergeAll $ map unbox samples 
 -- partitions :: [[Rd (Box [a])]]
 partitions         = --rnf globalSamples `pseq` 
                       parMapAt [2..p+1]  (\ (handle, (Box pivots)) 
                                          -> (releaseAll $ map Box $ decompose pivots (concat (fetch handle))))  
                                        (zip rdys (replicate p (Box globalSamples)))
 parts              = transpose partitions
 results            = parMapAt [2..p+1] ((chunk d) . mergeAll . map unbox . fetchAll) parts    
   
   
getGlobalSamples :: Trans a => Int -> [a] -> [a] 
getGlobalSamples p xs = takeEach p (drop (p+ (div p 2)-1) xs) 

getSamples :: Int -> [a] -> [a]
getSamples p xs =  takeEach step xs
   where 
     len  = length xs
     step = len `div` p  + (if (len `mod` p == 0) then 0 else 1)   
   
decompose  :: (Ord a) => [a] -> [a] -> [[a]] 
decompose (p:ps) ys = xs : decompose ps zs
     where (xs,zs) = break (> p) ys
decompose [] ys = [ys]