module Control.Concurrent.Parallel(parallel_) where

import GHC.Conc(numCapabilities)
import Control.Concurrent
import Control.Concurrent.Chan
import Control.Concurrent.MVar
import Control.Exception
import Control.Monad
import System.IO.Unsafe


{-# NOINLINE addParallel #-}
addParallel :: IO a -> IO ()
addParallel = unsafePerformIO $ do
    chan <- newChan
    mainThread <- myThreadId
    let err = throwTo mainThread $ ErrorCall "Control.Concurrent.Parallel: parallel thread died."
    replicateM_ (numCapabilities-1) $ forkIO $ finally (forever $ join $ readChan chan) err
    return $ writeChan chan

-- | Run the list of computations in parallel
--   Rule: No thread should get pre-empted (although not a guarantee)
parallel_ :: [IO a] -> IO ()
parallel_ xs | numCapabilities <= 1 = sequence_ xs
parallel_ [] = return ()
parallel_ [x] = x >> return ()
parallel_ (x:xs) = do
    ys <- mapM idempotent xs
    mapM_ addParallel ys
    sequence_ $ x : reverse ys


-- RULES:
-- The evaluation of act will happen at most once
-- If the resultant computation returns, the result will have happened exactly once
idempotent :: IO a -> IO (IO a)
idempotent act = do
    mvar <- newMVar Nothing
    return $ modifyMVar mvar f
    where
        f (Just x) = return (Just x, x)
        f Nothing = do
            v <- act
            return (Just v, v)