diff --git a/Control/Parallel/Eden.hs b/Control/Parallel/Eden.hs --- a/Control/Parallel/Eden.hs +++ b/Control/Parallel/Eden.hs @@ -1,4 +1,4 @@ -{-# OPTIONS -XCPP #-} +{-# OPTIONS -XCPP -XGeneralizedNewtypeDeriving -XExistentialQuantification -XDeriveDataTypeable #-} ----------------------------------------------------------------------------- -- | -- Module : Control.Parallel.Eden @@ -106,7 +106,23 @@ (Strategy, using, r0, rseq, rdeepseq, seqList, seqFoldable) import Control.Parallel(pseq) +import Control.Parallel.Eden.Merge +import Control.Applicative(Applicative(..)) +import Control.Monad.Fix(MonadFix(..)) +-- support for re-throwing exceptions to the receiver of data +-- Exceptions will propagate through the process network, eventually +-- reaching the main process. +import qualified Control.Exception as E +import Data.Typeable + +data RemoteException = forall e . E.Exception e => RemoteException Int e + deriving Typeable +instance Show RemoteException where + show (RemoteException pe e) = show e ++ " (on node " ++ show pe ++ ")" + +instance E.Exception RemoteException + -------------------------- -- legacy code for Eden 5: @@ -126,14 +142,8 @@ --------------------parallel action monad-------------------------- -newtype PA a = PA { fromPA :: IO a } +newtype PA a = PA { fromPA :: IO a } deriving (Monad,MonadFix,Functor,Applicative) -instance Monad PA where - return b = PA $ return b - (PA ioX) >>= f = PA $ do - x <- ioX - fromPA $ f x - runPA :: PA a -> a runPA = unsafePerformIO . fromPA @@ -258,6 +268,7 @@ -> [a] -- ^Process inputs -> [b] -- ^Process outputs {-# NOINLINE spawnAt #-} +spawnAt [] ps is = spawnAt [0] ps is spawnAt pos ps is = runPA $ sequence @@ -304,10 +315,11 @@ -- | Non-deterministically @merge@s a list of lists (usually input streams) -- into a single list. The order of the output list is determined by the -- availability of the inner lists constructors. (Function merge is defined --- using Concurrent Haskell's function @nmergeIO@) +-- using a list merge function @nmergeIO_E@) (similar to nmergeIO from +-- Concurrent Haskell, but in a custom version). merge :: [[a]] -- ^ Input lists -> [a] -- ^ Nondeterministically merged output list -merge xss = unsafePerformIO (nmergeIO xss) +merge xss = unsafePerformIO (nmergeIO_E xss) -- | same as @ merge @ mergeProc :: [[a]] -- ^ Input lists @@ -457,7 +469,15 @@ -- Use the default implementation for @write@ and @createComm@ for instances of Trans. class NFData a => Trans a where write :: a -> IO () - write x = rdeepseq x `pseq` sendData Data x + -- exception handler propagates exceptions to receiver (ultimately main) + write x = (rdeepseq x `pseq` sendData Data x) + `E.catch` -- do not "redecorate" RemoteExceptions + (\e -> sendData Data (E.throw (e::RemoteException))) + `E.catch` -- all other exceptions are decorated with origin + -- (see RemoteException above) + (\e -> do pe <- ParPrim.selfPe + let ex = RemoteException pe (e::E.SomeException) + sendData Data (E.throw ex)) -- write x = unEval $ -- do x' <- rdeepseq x -- return (sendData Data x') @@ -480,6 +500,7 @@ instance Trans Bool instance Trans a => Trans (Maybe a) +instance (Trans a,Trans b) => Trans (Either a b) instance Trans () -- unit: no communication desired? BREAKS OLD PROGRAMS @@ -490,8 +511,16 @@ -- stream communication: instance (Trans a) => Trans [a] where write l@[] = sendData Data l - -- write (x:xs) = (rnf x `seq` sendData Stream x) >> - write (x:xs) = (write' x) >> write xs + -- exception handler propagates exceptions to receiver (ultimately main) + -- and closes the stream + write (x:xs) = ((write' x) >> write xs) + `E.catch` -- do not "redecorate" RemoteExceptions + (\e -> sendData Data (E.throw (e::RemoteException))) + `E.catch` -- all other exceptions are decorated with origin + -- (see RemoteException above) + (\e -> do pe <- ParPrim.selfPe + let ex = RemoteException pe (e::E.SomeException) + sendData Data (E.throw ex)) where write' x = rdeepseq x `pseq` sendData Stream x diff --git a/Control/Parallel/Eden/EdenConcHs.hs b/Control/Parallel/Eden/EdenConcHs.hs --- a/Control/Parallel/Eden/EdenConcHs.hs +++ b/Control/Parallel/Eden/EdenConcHs.hs @@ -109,6 +109,8 @@ import Control.Parallel(pseq) +import Control.Parallel.Eden.Merge + -------------------------- -- legacy code for Eden 5: @@ -306,10 +308,11 @@ -- | Non-deterministically @merge@s a list of lists (usually input streams) -- into a single list. The order of the output list is determined by the -- availability of the inner lists constructors. (Function merge is defined --- using Concurrent Haskell's function @nmergeIO@) +-- using a list merge function @nmergeIO_E@) (similar to nmergeIO from +-- Concurrent Haskell, but in a custom version). merge :: [[a]] -- ^ Input lists -> [a] -- ^ Nondeterministically merged output list -merge xss = unsafePerformIO (nmergeIO xss) +merge xss = unsafePerformIO (nmergeIO_E xss) -- | same as @ merge @ mergeProc :: [[a]] -- ^ Input lists @@ -482,6 +485,7 @@ instance Trans Bool instance Trans a => Trans (Maybe a) +instance (Trans a,Trans b) => Trans (Either a b) instance Trans () -- unit: no communication desired? BREAKS OLD PROGRAMS diff --git a/Control/Parallel/Eden/Merge.hs b/Control/Parallel/Eden/Merge.hs new file mode 100644 --- /dev/null +++ b/Control/Parallel/Eden/Merge.hs @@ -0,0 +1,113 @@ +{-# LANGUAGE CPP, BangPatterns, MagicHash, UnboxedTuples #-} +----------------------------------------------------------------------------- +-- | +-- Module : Control.Parallel.Eden.Merge +-- Copyright : (c) Philipps Universitaet Marburg 2012- +-- License : BSD-style (see the file LICENSE) +-- +-- Maintainer : eden@mathematik.uni-marburg.de +-- Stability : beta +-- Portability : not portable +-- +-- Provides an implementation of nmergeIO (as previously available from +-- Control.Concurrent in base package). +-- Eden language definition assumes merge functionality as a base construct. +-- +-- Eden Group Marburg ( http:\/\/www.mathematik.uni-marburg.de/~eden ) +-- +----------------------------------------------- + +module Control.Parallel.Eden.Merge ( + -- * Non-deterministic merge of lists into one result list. + + -- | Previously available in Concurrent Haskell, living here from GHC-7.6 on. + -- This function merges incoming lists non-deterministically into one + -- result. + -- Its implementation in this module follows the pattern of the + -- previous one from Control.Concurrent, but uses the old simple semaphore + -- implementation, and optimises the case of only one remaining list. + nmergeIO_E + ) +where + +#if __GLASGOW_HASKELL__ < 707 +import Control.Concurrent(nmergeIO) + +nmergeIO_E = nmergeIO + +#else + +import Control.Concurrent +import Control.Exception( mask_ ) +import System.IO.Unsafe + +-- simplified version of Control.Concurrent.QSem (as of GHC-7.6) +type QSem_E = MVar (Int,[MVar ()]) + +newQSem_E :: Int -> IO QSem_E +newQSem_E n | n >= 0 = newMVar (n,[]) + | otherwise = error "QSem: negative." + +-- This implementation is prone to losing Q signals in use cases +-- where waiting threads can be killed externally (not the case here). +qsem_P, qsem_Q :: QSem_E -> IO () +qsem_P sem = mask_ $ + do state <- takeMVar sem + case state of + (0,blocked) -> do b <- newEmptyMVar + putMVar sem (0, blocked ++ [b]) + takeMVar b + (n,[]) -> putMVar sem (n-1, []) + (n,other) -> error ("QSem: " ++ show (n, length other)) +qsem_Q sem = mask_ $ + do state <- takeMVar sem + case state of + (n,(blocker:blockers)) -> do putMVar sem (n,blockers) + putMVar blocker () + (n,[]) -> putMVar sem (n+1, []) + +nmergeIO_E :: [[a]] -> IO [a] +nmergeIO_E lss + = do let !len = length lss + tl_node <- newEmptyMVar + tl_list <- newMVar tl_node + sem <- newQSem_E 1 + count_var <- newMVar len + -- start filler threads, one per incoming list + mapM_ (forkIO . suckIO_E count_var (tl_list, sem) ) lss + -- deliver result (as soon as first item filled in) + val <- takeMVar tl_node + qsem_Q sem + return val + +suckIO_E :: MVar Int -> (MVar (MVar [a]),QSem_E) -> [a] -> IO () +suckIO_E count_var buff@(tl_list,sem) vs + = do count <- takeMVar count_var + if count == 1 + -- last writer, identify own list with result + then do node <- takeMVar tl_list + putMVar node vs + putMVar tl_list node + -- sync with other writers to share 1-space buffer + else do putMVar count_var count + case vs of -- will block until input available + -- on end of list, decrease counter + [] -> do count <- takeMVar count_var + if count == 1 + then do node <- takeMVar tl_list + putMVar node [] + putMVar tl_list node + else do putMVar count_var (count-1) + -- on new element, add to output list + (x:xs) -> do qsem_P sem + node <- takeMVar tl_list + next <- newEmptyMVar + next_val <- unsafeInterleaveIO + (do y <- takeMVar next + qsem_Q sem + return y) + putMVar node (x:next_val) + putMVar tl_list next + suckIO_E count_var buff xs + +#endif diff --git a/configure b/configure --- a/configure +++ b/configure @@ -1,6 +1,6 @@ #! /bin/sh # Guess values for system-dependent variables and create Makefiles. -# Generated by GNU Autoconf 2.68 for Eden modules for parallel Haskell 1.1.0.1. +# Generated by GNU Autoconf 2.68 for Eden modules for parallel Haskell 1.2.0.0. # # Report bugs to . # @@ -559,8 +559,8 @@ # Identity of this package. PACKAGE_NAME='Eden modules for parallel Haskell' PACKAGE_TARNAME='edenmodules' -PACKAGE_VERSION='1.1.0.1' -PACKAGE_STRING='Eden modules for parallel Haskell 1.1.0.1' +PACKAGE_VERSION='1.2.0.0' +PACKAGE_STRING='Eden modules for parallel Haskell 1.2.0.0' PACKAGE_BUGREPORT='eden@informatik.uni-marburg.de' PACKAGE_URL='' @@ -1158,7 +1158,7 @@ # Omit some internal or obsolete options to make the list less imposing. # This message is too long to be a string in the A/UX 3.1 sh. cat <<_ACEOF -\`configure' configures Eden modules for parallel Haskell 1.1.0.1 to adapt to many kinds of systems. +\`configure' configures Eden modules for parallel Haskell 1.2.0.0 to adapt to many kinds of systems. Usage: $0 [OPTION]... [VAR=VALUE]... @@ -1219,7 +1219,7 @@ if test -n "$ac_init_help"; then case $ac_init_help in - short | recursive ) echo "Configuration of Eden modules for parallel Haskell 1.1.0.1:";; + short | recursive ) echo "Configuration of Eden modules for parallel Haskell 1.2.0.0:";; esac cat <<\_ACEOF @@ -1291,7 +1291,7 @@ test -n "$ac_init_help" && exit $ac_status if $ac_init_version; then cat <<\_ACEOF -Eden modules for parallel Haskell configure 1.1.0.1 +Eden modules for parallel Haskell configure 1.2.0.0 generated by GNU Autoconf 2.68 Copyright (C) 2010 Free Software Foundation, Inc. @@ -1308,7 +1308,7 @@ This file contains any messages produced by compilers while running configure, to aid debugging if configure makes a mistake. -It was created by Eden modules for parallel Haskell $as_me 1.1.0.1, which was +It was created by Eden modules for parallel Haskell $as_me 1.2.0.0, which was generated by GNU Autoconf 2.68. Invocation command line was $ $0 $@ @@ -2314,7 +2314,7 @@ # report actual input values of CONFIG_FILES etc. instead of their # values after options handling. ac_log=" -This file was extended by Eden modules for parallel Haskell $as_me 1.1.0.1, which was +This file was extended by Eden modules for parallel Haskell $as_me 1.2.0.0, which was generated by GNU Autoconf 2.68. Invocation command line was CONFIG_FILES = $CONFIG_FILES @@ -2367,7 +2367,7 @@ cat >>$CONFIG_STATUS <<_ACEOF || ac_write_fail=1 ac_cs_config="`$as_echo "$ac_configure_args" | sed 's/^ //; s/[\\""\`\$]/\\\\&/g'`" ac_cs_version="\\ -Eden modules for parallel Haskell config.status 1.1.0.1 +Eden modules for parallel Haskell config.status 1.2.0.0 configured by $0, generated by GNU Autoconf 2.68, with options \\"\$ac_cs_config\\" diff --git a/configure.ac b/configure.ac --- a/configure.ac +++ b/configure.ac @@ -1,4 +1,4 @@ -AC_INIT([Eden modules for parallel Haskell], [1.1.0.1], [eden@informatik.uni-marburg.de], [edenmodules]) +AC_INIT([Eden modules for parallel Haskell], [1.2.0.0], [eden@informatik.uni-marburg.de], [edenmodules]) # Safety check: Ensure that we are in the correct source directory. AC_CONFIG_SRCDIR([edenmodules.cabal]) diff --git a/edenmodules.cabal b/edenmodules.cabal --- a/edenmodules.cabal +++ b/edenmodules.cabal @@ -1,5 +1,5 @@ name: edenmodules -version: 1.1.0.1 +version: 1.2.0.0 license: BSD3 license-file: LICENSE maintainer: eden@mathematik.uni-marburg.de @@ -47,9 +47,10 @@ Control.Parallel.Eden.ParPrimConcHs Control.Parallel.Eden.EdenConcHs Control.Parallel.Eden.Edi + Control.Parallel.Eden.Merge extensions: CPP build-depends: base >= 4 && < 5, - containers >= 0.3 && < 0.5, + containers >= 0.3 && < 0.6, deepseq >= 1.1 && < 1.4, parallel >= 3.0 && < 4.0 }