edenmodules 1.1.0.1 → 1.2.0.0
raw patch · 6 files changed
+174/−27 lines, 6 filesdep ~containersnew-uploader
Dependency ranges changed: containers
Files
- Control/Parallel/Eden.hs +42/−13
- Control/Parallel/Eden/EdenConcHs.hs +6/−2
- Control/Parallel/Eden/Merge.hs +113/−0
- configure +9/−9
- configure.ac +1/−1
- edenmodules.cabal +3/−2
Control/Parallel/Eden.hs view
@@ -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
Control/Parallel/Eden/EdenConcHs.hs view
@@ -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
+ Control/Parallel/Eden/Merge.hs view
@@ -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
configure view
@@ -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 <eden@informatik.uni-marburg.de>. #@@ -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\\"
configure.ac view
@@ -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])
edenmodules.cabal view
@@ -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 }