-- Communicating Haskell Processes.
-- Copyright (c) 2008, University of Kent.
-- All rights reserved.
-- 
-- Redistribution and use in source and binary forms, with or without
-- modification, are permitted provided that the following conditions are
-- met:
--
--  * Redistributions of source code must retain the above copyright
--    notice, this list of conditions and the following disclaimer.
--  * Redistributions in binary form must reproduce the above copyright
--    notice, this list of conditions and the following disclaimer in the
--    documentation and/or other materials provided with the distribution.
--  * Neither the name of the University of Kent nor the names of its
--    contributors may be used to endorse or promote products derived from
--    this software without specific prior written permission.
--
-- THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
-- IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
-- THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
-- PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
-- CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
-- EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
-- PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
-- PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
-- LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
-- NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
-- SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

module Control.Concurrent.CHP.Parallel (runParallel, runParallel_, (<||>), (<|*|>),
  runParMapM, runParMapM_, ForkingT, forking, fork) where

import Control.Concurrent
import Control.Concurrent.STM
import qualified Control.Exception.Extensible as C
import Control.Monad.Reader
import Control.Monad.State
import Data.List
import Data.Maybe
import Data.Ord

import Control.Concurrent.CHP.Base
import Control.Concurrent.CHP.Traces.Base

-- | This type-class supports parallel composition of processes.  You may use
-- the 'runParallel' function to run a list of processes, or the '<||>' operator
-- to run just a pair of processes.
-- 
-- In each case, the composition waits for all processes to finish, either
-- successfully or with poison.  At the end of this, if /any/ process
-- exited with poison, the composition will \"rethrow\" this poison.  If all
-- the processes completed successfully, the results will be returned.  If
-- you want to ignore poison from the sub-processes, use an empty poison
-- handler and 'onPoisonTrap' with each branch.

-- | Runs the given list of processes in parallel, and then returns a list
-- of the results, in the same order as the processes given.  Will only return
-- when all the inner processes have completed.
--
-- In version 1.5.0, a bug was introduced such that @runParallel []@ would deadlock;
-- this was fixed in version 1.5.1.
runParallel :: [CHP a] -> CHP [a]
runParallel = runParallelPoison

-- | A shorthand for applying mapM in parallel; really the composition of 'runParallel'
-- and 'map'.
--
-- Added in version 1.5.0.
runParMapM :: (a -> CHP b) -> [a] -> CHP [b]
runParMapM f = runParallel . map f

-- | A shorthand for applying mapM_ in parallel; really the composition of 'runParallel_'
-- and 'map'.
--
-- Added in version 1.5.0.
runParMapM_ :: (a -> CHP b) -> [a] -> CHP ()
runParMapM_ f = runParallel_ . map f


-- | A useful operator for performing a two process equivalent of 'runParallel'
-- that gives the return values back as a pair rather than a list.  This also
-- allows the values to have different types
(<||>) :: CHP a -> CHP b -> CHP (a, b)
(<||>) p q = do [x, y] <- runParallel [liftM Left p, liftM Right q]
                combine x y
    where
      combine :: Monad m => Either a b -> Either a b -> m (a, b)
      combine (Left x) (Right y) = return (x, y)
      combine (Right y) (Left x) = return (x, y)
      -- An extra case to keep the compiler happy:
      combine _ _ = error "Impossible combination values in <|^|>"

-- | An operator similar to '<||>' that discards the output (more like an operator
-- version of 'runParallel_').
--
-- Added in version 1.1.0.
(<|*|>) :: CHP a -> CHP b -> CHP ()
(<|*|>) p q = runParallel_ [p >> return (), q >> return ()]

-- | Runs all the given processes in parallel and discards any output.  Does
-- not return until all the processes have completed.  'runParallel_' ps is effectively equivalent
-- to 'runParallel' ps >> return ().
--
-- In version 1.5.0, a bug was introduced such that @runParallel_ []@ would deadlock;
-- this was fixed in version 1.5.1.
runParallel_ :: [CHP a] -> CHP ()
runParallel_ procs = runParallel procs >> return ()

-- We right associate to allow the liftM fst ((readResult) <||> runParallel_
-- workers) pattern
infixr <||>
-- Doesn't really matter for this operator:
infixr <|*|>

-- | Runs all the processes in parallel and returns their results once they
-- have all finished.  The length and ordering of the results reflects the
-- length and ordering of the input
runParallelPoison :: forall a. [CHP a] -> CHP [a]
runParallelPoison processes
  = do resultVar <- liftIO $ atomically $ newManyToOneTVar
         ((== length processes) . length) (return []) []
       trace <- PoisonT $ lift $ liftTrace ask
       blanks <- liftIO $ blankTraces trace (length processes)
       liftIO $ 
         mapM_ forkIO [do y <- wrapProcess p $ flip runReaderT btr . pullOutStandard
                          C.block $ atomically $
                            writeManyToOneTVar
                             ((:) (case y of
                                 Nothing -> (n, Nothing)
                                 Just (Right x) -> (n, Just x)
                                 Just (Left _) -> (n, Nothing)
                                 )) resultVar
                             >> return ()
                      | (p, btr, n) <- zip3 processes blanks [(0::Int)..]]
       results <- liftIO $ atomically $ readManyToOneTVar resultVar
       let sortedResults = map snd $ sortBy (comparing fst) results
       PoisonT $ lift $ liftTrace $ mergeSubProcessTraces blanks
       mapM (maybe throwPoison return) sortedResults

-- | A monad transformer used for introducing forking blocks.
newtype (Monad m, MonadCHP m) => ForkingT m a = Forking (ReaderT (TVar (Bool,
  Int)) m a)
  deriving (Monad, MonadIO, MonadTrans, MonadCHP)

-- TODO in future, get forking working with structural traces

-- | Executes a forking block.  Processes may be forked off inside (using the
-- 'fork' function).  When the block completes, it waits for all the forked
-- off processes to complete before returning the output, as long as none of
-- the processes terminated with uncaught poison.  If they did, the poison
-- is propagated (rethrown).
forking :: MonadCHP m => ForkingT m a -> m a
-- Like with parallel, this could probably be made a little more efficient
forking (Forking m) = do b <- liftIO $ atomically $ newTVar (False, 0)
                         output <- runReaderT m b
                         p <- liftIO $ atomically $ do
                           (p,n) <- readTVar b
                           if n == 0
                             then return p
                             else retry
                         if p
                           then liftCHP throwPoison
                           else return output
                         

-- | Forks off the given process.  The process then runs in parallel with this
-- code, until the end of the 'forking' block, when all forked-off processes
-- are waited for.  At that point, once all of the processes have finished,
-- if any of them threw poison it is propagated.
fork :: MonadCHP m => CHP () -> ForkingT m ()
fork p = Forking $
           do b <- ask
              liftIO $ atomically $ do
                (pa, n) <- readTVar b
                writeTVar b (pa, n + 1)
              trace <- liftCHP $ PoisonT $ lift $ liftTrace ask
              [blank] <- liftIO $ blankTraces trace 1
              liftIO $ forkIO $ do
                r <- wrapProcess p $ flip runReaderT blank . pullOutStandard
                C.block $ atomically $ do
                  (poisonedAlready, n) <- readTVar b
                  writeTVar b $ (poisonedAlready || isNothing r, n - 1)
              return ()