ConcurrentUtils 0.4.2.0 → 0.4.4.0

raw patch · 9 files changed

+534/−202 lines, 9 filesdep +MonadRandomdep +list-extrasdep +monads-tfdep ~basePVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: MonadRandom, list-extras, monads-tf

Dependency ranges changed: base

API changes (from Hackage documentation)

Files

• ConcurrentUtils.cabal +4/−4
• Control/CUtils/AList.hs +15/−4
• Control/CUtils/Channel.hs +14/−1
• Control/CUtils/Conc.hs +72/−38
• Control/CUtils/DataParallel.hs +281/−128
• Control/CUtils/Deadlock.hs +64/−15
• Control/CUtils/FChan.hs +42/−12
• Control/CUtils/StrictArrow.hs +14/−0
• Data/BellmanFord.hs +28/−0

@@ -2,7 +2,7 @@ -- documentation, see http://haskell.org/cabal/users-guide/
 
 name:                ConcurrentUtils
-version:             0.4.2.0
+version:             0.4.4.0
 synopsis:            Concurrent utilities
 -- description:         
 homepage:            http://alkalisoftware.net
@@ -16,6 +16,6 @@ cabal-version:       >=1.8
 
 library
-  exposed-modules:     Control.CUtils.Processes, Control.CUtils.NetChan, Control.CUtils.FChan, Control.CUtils.Deadlock, Control.CUtils.DataParallel, Control.CUtils.Conc, Control.CUtils.Channel, Control.CUtils.AList
-  other-modules:       Control.CUtils.Split   
-  build-depends:       base >=4 && <=5, process, network >=2.4, bytestring, binary, containers, array, parallel, cryptohash >=0.11.6, RSA >=2.1.0, crypto-random >=0.0.8, securemem >= 0.1.7, reexport-crypto-random, tagged >= 0.7.3
+  exposed-modules:     Control.CUtils.Processes, Control.CUtils.NetChan, Control.CUtils.FChan, Control.CUtils.Deadlock, Control.CUtils.Conc, Control.CUtils.Channel, Control.CUtils.AList, Control.CUtils.StrictArrow, Data.BellmanFord, Control.CUtils.DataParallel
+  other-modules:       Control.CUtils.Split
+  build-depends:       base >=4.4.0.0 && <=5, process, network >=2.4, bytestring, binary, containers, array, parallel, cryptohash >=0.11.6, RSA >=2.1.0, crypto-random >=0.0.8, securemem >= 0.1.7, reexport-crypto-random, tagged >= 0.7.3, MonadRandom, monads-tf, list-extras

@@ -1,3 +1,5 @@+{-# LANGUAGE Trustworthy, DeriveDataTypeable #-}
+
 -- | Lists suitable for parallel execution (taken from Hackage's monad-par package). (For converting to regular lists, there is the toList function in Data.Foldable.)
 module Control.CUtils.AList (AList(..), filterAList, assocFold, monoid, lenAList, findAList, concatAList) where
 
@@ -7,23 +9,32 @@ import Data.Monoid
 import Data.Foldable (Foldable, foldMap)
 import Data.Traversable (Traversable, traverse, foldMapDefault)
+import Data.Data
 
-data AList t = Append (AList t) (AList t) | List [t] deriving (Eq, Ord, Show)
+data AList t = Append (AList t) (AList t) | List [t] deriving (Eq, Ord, Show, Typeable, Data)
 
 instance Monad AList where
 	return x = List [x]
-	Append ls ls2 >>= f = ((ls >>= f) `par` (ls2 >>= f)) `seq` Append (ls >>= f) (ls2 >>= f)
+	Append ls ls2 >>= f = ((ls >>= f) `par` (ls2 >>= f)) `pseq` Append (ls >>= f) (ls2 >>= f)
 	List ls >>= f = foldr mplus mzero (map f ls)
 
 instance MonadPlus AList where
 	mzero = List []
-	mplus m n = (m `par` n) `seq` case (m, n) of
+	mplus m n = (m `par` n) `pseq` case (m, n) of
 		(List [x], List xs) -> List (x:xs)
 		(List [x], Append y z) -> Append (mplus m y) z
 		(List [], n) -> n
 		(m, List []) -> m
 		_ -> Append m n
 
+instance Applicative AList where
+	pure = return	
+	(<*>) = ap
+
+instance Alternative AList where
+	empty = mzero
+	(<|>) = mplus
+
 instance Functor AList where
 	fmap f m = m >>= return . f
 
@@ -40,7 +51,7 @@ noNils (Append m n) = noNils m `mplus` noNils n
 noNils ls = ls
 
-assocFold0 f (Append ls ls2) = (assocFold0 f ls `par` assocFold0 f ls2) `seq` f (assocFold0 f ls) (assocFold0 f ls2)
+assocFold0 f (Append ls ls2) = (assocFold0 f ls `par` assocFold0 f ls2) `pseq` f (assocFold0 f ls) (assocFold0 f ls2)
 assocFold0 f (List ls) = foldl1 f ls
 
 -- | Folds the AList with a function, that must be associative. This allows parallelism to be introduced.

@@ -1,7 +1,7 @@ {-# LANGUAGE FlexibleContexts #-}
 
 -- | A lock-free channel (queue) data structure.
-module Control.CUtils.Channel (Channel, newChannel, writeChannel, readChannel) where
+module Control.CUtils.Channel (Channel, newChannel, writeChannel, readChannel, tryReadChannel) where
 
 import Control.Concurrent.MVar
 import Control.Monad
@@ -68,3 +68,16 @@ 
 -- | Read from the channel, blocking when the buffer is empty.
 readChannel (Channel buffer a mf f me e fl el) = alg buffer 0 a me e mf f el fl readArray
+
+-- |
+tryReadChannel (Channel buffer a _ f me e fl _) = do
+	meN <- increment me
+	fN <- readIORef f
+	if fN <= meN then do
+			increment e
+			return Nothing
+		else do
+			val <- readArray a (meN `mod` buffer)
+			increment e
+			tryPutMVar fl ()
+			return (Just val)

@@ -1,6 +1,6 @@-{-# LANGUAGE ScopedTypeVariables, DeriveDataTypeable, ImplicitParams #-}
+{-# LANGUAGE Trustworthy, ScopedTypeVariables, DeriveDataTypeable, ImplicitParams #-}
 -- | A module of concurrent higher order functions.
-module Control.CUtils.Conc (ExceptionList(..), ConcException(..), assocFold, concF_, conc_, concF, conc, concP, oneOfF, oneOf) where
+module Control.CUtils.Conc (ExceptionList(..), ConcException(..), assocFold, Concurrent(..), concF_, concF, conc_, conc, concP, progressConcF, oneOfF, oneOf) where
 
 import Prelude hiding (catch)
 import Control.Exception
@@ -10,10 +10,12 @@ import GHC.Conc
 import Data.Array.IO (newArray_, readArray, writeArray, getElems, IOArray)
 import Data.Array
--- import Data.Array.Unsafe
+import Data.Array.Unsafe
 import Data.Array.MArray
 import Data.IORef
 import Control.Monad
+import Control.Arrow
+import System.IO.Unsafe
 
 -- | For exceptions caused by caller code.
 data ExceptionList = ExceptionList [SomeException] deriving (Show, Typeable)
@@ -51,41 +53,73 @@ 			chanToList []
 	unless (null exslst) $ throwIO (ExceptionList exslst)
 
--- | Runs an associative folding function on the given array. Note: this function only spawns enough threads to make effective use of the /capabilities/. Any two list elements may be processed sequentially or concurrently. To get parallelism, you have to set the numCapabilities value, e.g. using GHC's +RTS -N flag.
-assocFold :: forall a b. (b -> b -> IO b) -> (a -> b) -> b -> Array Int a -> IO b
-assocFold f g init parm = do
-	let (lo, hi) = bounds parm
-	when (lo > hi) $ error "Conc.assocFold: empty list"
-	exs <- newChan
-	ar <- (newArray_ (0, (rangeSize (bounds parm) `min` numCapabilities) - 1) :: IO (IOArray Int b))
-	let
-		rtnException ex = writeChan exs (Just ex) >> return undefined
-		innerExHandler m = catch m rtnException
-		outerExHandler m = catch m (\(_ :: SomeException) -> rtnException (toException ConcException)) in
-		outerExHandler $ simpleConc_ $ map (\(i, (x, y)) ->
-			innerExHandler $ foldM (\x -> f x . g . (parm !)) init [x..y] >>= writeArray ar i) $ zip [0..] (divideUp numCapabilities (rangeSize (bounds parm)))
-	getExceptions exs
-	ls <- getElems ar
-	foldM f init ls
+-- | A type class of arrows that support some form of concurrency.
+class Concurrent a where
+	-- | Runs an associative folding function on the given array.
+	--   Note: this function only spawns enough threads to make effective use of the /capabilities/.
+	--   Any two list elements may be processed sequentially or concurrently. To get parallelism,
+	--   you have to set the numCapabilities value, e.g. using GHC's +RTS -N flag.
+	arr_assocFold :: a (b, b) b -> (c -> b) -> a (b, Array Int c) b
+	-- | The first parameter is the number of computations which are indexed from 0 to n - 1.
+	arr_concF_ :: (?seq :: Bool) => a (t, Int) () -> a (t, Int) ()
+	arr_concF :: (?seq :: Bool) => a (u, Int) t -> a (u, Int) (Array Int t)
+	arr_oneOfF :: a (u, Int) b -> a (u, Int) b
 
+instance Concurrent (Kleisli IO) where
+	arr_assocFold f g = Kleisli $ \(init, parm) -> do
+		let (lo, hi) = bounds parm
+		when (lo > hi) $ error "Conc.arr_assocFold: empty list"
+		exs <- newChan
+		caps <- getNumCapabilities
+		-- With unlimited caps, you can do sqrt(n) folds on each thread, then sqrt(n) to fold the results (O(sqrt(n)f(n)) time).
+		let effectiveCaps = ceiling (sqrt (fromIntegral (rangeSize (bounds parm)))) `min` caps
+		ar <- (newArray_ (0, effectiveCaps - 1) :: IO (IOArray Int b))
+		let
+			rtnException ex = writeChan exs (Just ex) >> return undefined
+			innerExHandler m = catch m rtnException
+			outerExHandler m = catch m (\(_ :: SomeException) -> rtnException (toException ConcException)) in
+			outerExHandler $ simpleConc_ $ map (\(i, (x, y)) ->
+				innerExHandler $ foldM (\x -> runKleisli f . (,) x . g . (parm !)) init [x..y] >>= writeArray ar i) $ zip [0..] (divideUp effectiveCaps (rangeSize $ bounds parm))
+		getExceptions exs
+		ls <- getElems ar
+		foldM (curry (runKleisli f)) init ls
+	arr_concF_ mnds = Kleisli $ \(parm, n) -> do
+		exs <- newChan
+		caps <- getNumCapabilities
+		let
+			rtnException ex = writeChan exs (Just ex) >> return undefined
+			innerExHandler m = catch m rtnException
+			outerExHandler m = catch m (\(_ :: SomeException) -> rtnException (toException ConcException)) in
+			outerExHandler $
+			simpleConc_ $ map (\(x, y) -> outerExHandler $ (if ?seq then sequence_ else simpleConc_) $ map (innerExHandler . runKleisli mnds . (,) parm) [x..y-1]) $ divideUp caps n
+		getExceptions exs
+	arr_concF mnds = Kleisli $ \(parm, n) -> partConcF (0, n - 1) (concF_ n) (runKleisli mnds . (,) parm)
+	arr_oneOfF mnds = Kleisli $ \(parm, n) -> partOneOfF (0, n - 1) (runKleisli mnds . (,) parm)
+
+-- '->' has no effects, but one can compute its results in parallel anyway (pointlessly,
+-- in the case of 'arr_concF_').
+instance Concurrent (->) where
+	arr_assocFold f g x = unsafePerformIO $ assocFold (\x y -> return $! f (x, y)) g x
+	arr_concF_ _ = arr (const ())
+	arr_concF mnds (parm, n) = let ?seq = True in unsafePerformIO $ concF n ((return $!) . mnds . (,) parm)
+	arr_oneOfF mnds (parm, n) = unsafePerformIO $ oneOfF n ((return $!) . mnds . (,) parm)
+
 -- |
-concF_ :: (?seq :: Bool) => Int -> (Int -> IO ()) -> IO ()
-concF_ n mnds = do
-	exs <- newChan
-	let
-		rtnException ex = writeChan exs (Just ex) >> return undefined
-		innerExHandler m = catch m rtnException
-		outerExHandler m = catch m (\(_ :: SomeException) -> rtnException (toException ConcException)) in
-		outerExHandler $ simpleConc_ $ map (\(x, y) -> outerExHandler $ (if ?seq then sequence_ else simpleConc_) $ map (innerExHandler . mnds) [x..y-1]) $ divideUp numCapabilities n
-	getExceptions exs
+assocFold f g = runKleisli (arr_assocFold (Kleisli (uncurry f)) g)
 
 partConc_ f mnds = concF_ (rangeSize (bounds mnds)) $ f . (+ fst (bounds mnds))
 
 -- |
+concF_ n mnds = runKleisli (arr_concF_ (Kleisli (mnds . snd))) ((), n)
+
+-- |
+concF n mnds = runKleisli (arr_concF (Kleisli (mnds . snd))) ((), n)
+
+-- |
 conc_ mnds = partConc_ (mnds !) mnds
 
 unsafeFreeze' :: IOArray Int e -> IO (Array Int e)
-unsafeFreeze' = freeze
+unsafeFreeze' = unsafeFreeze
 
 partConcF bnds f mnds = do
 	res <- newArray_ bnds
@@ -94,15 +128,12 @@ 		writeArray res i x)
 	unsafeFreeze' res
 
--- | The next two functions take an implicit parameter ?seq. Set it to True
--- if you want to only spawn threads for the capabilities (same as /assocFold/,
+-- | The next function takes an implicit parameter ?seq. Set it to True
+-- if you want to only spawn threads for the capabilities (same as /assocFold/;
 -- good for speed). If you need all the actions to be executed concurrently,
 -- set it to False.
---
--- n is the number of computations which are indexed from 0 to n - 1.
-concF n = partConcF (0, n - 1) (concF_ n)
 
--- | Runs several computations concurrently, and returns their results as an array. Waits for all threads to end before returning.
+-- Runs several computations concurrently, and returns their results as an array. Waits for all threads to end before returning.
 conc mnds = partConcF (bounds mnds) (\f -> partConc_ f mnds) (mnds !)
 
 -- | Version of concF specialized for two computations.
@@ -112,6 +143,11 @@ 		else
 			liftM Right m2)
 
+progressConcF n f = do
+	res <- concF n (\i -> f i >>= \x -> when (i * 80 `mod` n == 0) (putChar '|') >> return x)
+	putStrLn ""
+	return res
+
 partOneOfF bnds mnds = do
 	thds <- newIORef []
 	chn <- newChan
@@ -129,9 +165,7 @@ 			chanToList 0 [])
 		(catch (readIORef thds >>= mapM_ killThread) (\(_ :: SomeException) -> throwIO ConcException))
 
--- |
-oneOfF :: Int -> (Int -> IO a) -> IO a
-oneOfF n = partOneOfF (0, n - 1)
+oneOfF n mnds = runKleisli (arr_oneOfF (Kleisli (mnds . snd))) ((), n)
 
 -- | Runs several computations in parallel, and returns one of their results (terminating the other computations).
 oneOf :: Array Int (IO a) -> IO a

@@ -1,128 +1,281 @@-{-# LANGUAGE ImplicitParams #-}
--- | An implementation of nested data parallelism
-module Control.CUtils.DataParallel where
-
-import Data.Array hiding (index)
-import Data.Tree
-import Control.CUtils.Conc
-import Control.Monad
-import System.IO.Unsafe
-import Prelude hiding (zip, concat, and)
-import qualified Prelude as P
-
--- | The array interface
-data ArrC t = ArrC !(Array Int t) !(Forest Int)
-
--- | Inject a basic array into the ArrC type.
-inject ar = ArrC ar [Node 0 [], Node (uncurry subtract (bounds ar) + 1) []]
-
--- | Get a basic array out.
-project (ArrC ar _) = ar
-
--- | Convenience for making an array from a list.
-newArray ls = listArray (0, length ls - 1) ls
-
-mirror x = either Right Left x
-
-pairUp ls = P.zip ls (tail ls)
-
--- | Programs involving these array operations are optimized
---   by a set of rules when GHC's -O option is set. Use +RTS -N to get parallelism.
-{-# INLINE [0] mp #-}
-mp f (ArrC ar ls) = ArrC (unsafePerformIO $ let ?seq = True in conc $ fmap ((return $!) . f) ar) ls
-
-{-# INLINE [0] count #-}
-count n = inject $ unsafePerformIO $ let ?seq = True in concF n (return $!)
-
-{-# INLINE [0] index #-}
-index (ArrC ar _) i = ar ! i
-
-{-# INLINE [0] zip #-}
-zip (ArrC ar _) (ArrC ar2 _) = inject $ unsafePerformIO $ let ?seq = True in concF (snd (bounds ar) `min` snd (bounds ar2))
-	(\i -> let x = ar ! i; y = ar2 ! i in x `seq` y `seq` return $! (x, y))
-
-{-# INLINE [0] concat #-}
-concat ar0 = ArrC ar [ Node (i + j) ls3 | Node i ls2 <- ls, Node j ls3 <- ls2 ]
-	where ArrC ar ls = __pack ar0
-
--- | Associative fold
-{-# INLINE [0] fold #-}
-fold f g init ar = unsafePerformIO $ assocFold (\y z -> return $! f y z) g init $ project ar
-
--- | Control.Arrow substitutes
-{-# INLINE [0] first #-}
-first f (x, y) = (f x, y)
-
-{-# INLINE [0] second #-}
-second f (x, y) = (x, f y)
-
-{-# INLINE [0] left #-}
-left f = either (Left . f) Right
-
-{-# INLINE [0] right #-}
-right f = either Left (Right . f)
-
-{-# INLINE [0] and #-}
-and f g x = (f x, g x)
-
--- | Internals
-{-# INLINE [0] __pack #-}
-__pack (ArrC ar ls) = ArrC (newArray $ concatMap (elems . project) $ elems ar)
-	(zipWith Node (scanl (\i (ArrC ar _) -> i + rangeSize (bounds ar)) 0 $ elems ar)
-		(map (\(ArrC _ ls) -> ls) (elems ar) ++ [[]]))
-
-{-# INLINE [0] __unpack #-}
-__unpack (ArrC ar ls) = inject $ unsafePerformIO $ let ?seq = True in conc $ fmap
-	(\(Node i ls, Node j _) -> liftM (\ar -> ArrC ar ls) $ concF (j-i) $ \k -> return $! ar ! (k+i))
-	$ newArray $ pairUp ls
-
-{-# INLINE [0] __packProd #-}
-__packProd (x, y) = inject $ newArray [Left x, Right y]
-
-{-# INLINE [0] __unpackProd #-}
-__unpackProd ar = (case project ar ! 0 of Left x -> x, case project ar ! 1 of Right x -> x)
-
-{-# INLINE [0] __packSum1 #-}
-__packSum1 (Left x) = inject (newArray [Left x])
-__packSum1 (Right ar) = mp Right ar
-
-{-# INLINE [0] __unpackSum1 #-}
-__unpackSum1 ar = either Left (\_ -> Right (mp (\(Right x) -> x) ar)) (project ar ! 0)
-
-{-# INLINE [0] __packSum2 #-}
-__packSum2 x = mp mirror (__packSum1 (mirror x))
-
-{-# INLINE [0] __unpackSum2 #-}
-__unpackSum2 x = mirror (__unpackSum1 (mp mirror x))
-
-{-# RULES
-
-"packMap" [2] forall f x. mp (mp f) x = __unpack (mp f (__pack x))
-"packProd" [2] forall f x. first f x = __unpackProd (mp (left f) (__packProd x))
-"packProd2" [2] forall f x. second f x = __unpackProd (mp (right f) (__packProd x))
-"packSum" [2] forall f x. right (mp f) x = __unpackSum1 (mp (right f) (__packSum1 x))
-"packSum2" [2] forall f x. left (mp f) x = __unpackSum2 (mp (left f) (__packSum2 x))
-
-"sepMapComp" [2] forall f g x. mp (f . g) x = mp f (mp g x)
-"sepMapProd" [2] forall f ar. mp (and f id) ar = zip (mp f ar) ar
-"sepMapProd2" [2] forall f ar. mp (and id f) ar = zip ar (mp f ar)
-"sepSum" [2] forall f g x. left (f . g) x = left f (left g x)
-"sepSum2" [2] forall f g x. right (f . g) x = right f (right g x)
-
-"combMapComp" [1] forall f g x. mp f (mp g x) = mp (f . g) x
-"combMapProd" [1] forall f ar. zip (mp f ar) ar = mp (and f id) ar
-"combMapProd2" [1] forall f ar. zip ar (mp f ar) = mp (and id f) ar
-"combSum" [1] forall f g x. left f (left g x) = left (f . g) x
-"combSum2" [1] forall f g x. right f (right g x) = right (f . g) x
-"unpackPack" [1] forall x. __pack (__unpack x) = x
-"unpackProd" [1] forall x. __packProd (__unpackProd x) = x
-"unpackSum" [1] forall x. __packSum1 (__unpackSum1 x) = x
-"unpackSum2" [1] forall x. __packSum2 (__unpackSum2 x) = x
-
-"zip" [1] forall f x y. mp (\y -> f (fst y)) (zip x y) = mp f x
-"zip2" [1] forall f x y. mp (\y -> f (snd y)) (zip x y) = mp f y
-"index" [1] forall f ar i. index (mp f ar) i = f (index ar i)
-"concatConcat" [1] forall x. concat (mp concat x) = concat (concat x)
-"fold" [1] forall f g h x y. fold f g x (mp h y) = fold f (g . h) x y
-  #-}
-
+{-# LANGUAGE Safe, GADTs, Rank2Types, ImplicitParams, Arrows #-}+-- | An implementation of nested data parallelism (due to Simon Peyton Jones et al)+module Control.CUtils.DataParallel (+-- * Flattenable arrays+ArrC, newArray, inject, project,+-- * The arrows and associated operations+Structural, A, unA, mapA', liftA, countA, indexA, zipA, unzipA, concatA, eval,+-- * Examples+nQueens, sorting, permute) where++import Data.Array+import Data.List+import Data.Monoid (Any(Any))+import Control.Category+import Control.Arrow+import Control.Monad.Writer (Writer, tell, runWriter)+import Control.Monad.Identity+import Control.Monad+import Control.CUtils.Conc+import Control.CUtils.StrictArrow+import Prelude hiding (id, (.))++data Tree t = Node !t !(Array Int (Tree t))++data ArrC t = ArrC !(Array Int t) !(Array Int (Tree Int))++newArray ls = listArray (0, length ls - 1) ls++inject ar = ArrC (ixmap (0, uncurry subtract (bounds ar)) (subtract (fst (bounds ar))) ar) (newArray [Node 0 (newArray []), Node (uncurry subtract (bounds ar) + 1) (newArray [])])++project (ArrC ar _) = ar++instance Functor ArrC where+	fmap f (ArrC ar fr) = ArrC (fmap f ar) fr++instance Show (t -> u) where+	showsPrec _ _ = ("<FUNCTION>"++)++data Structural a t u where+	Map :: Structural a t u -> Structural a (ArrC t) (ArrC u)+	Comp :: Structural a u v -> Structural a t u -> Structural a t v+	Id :: Structural a t t+	Product :: Structural a t u -> Structural a v w -> Structural a (t, v) (u, w)+	Lift :: a t u -> Structural a t u+	Count :: Structural a (t, Int) (ArrC (t, Int))+	Index :: Structural a (ArrC t, Int) t+	Zip :: Structural a (ArrC t, ArrC u) (ArrC (t, u))+	Unzip :: Structural a (ArrC (t, u)) (ArrC t, ArrC u)+	ClearMarks :: Structural a (ArrC t) (ArrC t)+	Separate :: Structural a (Either t u) (ArrC t, ArrC u)+	Combine :: Structural a (ArrC t, ArrC u) (Either t u)+	Pack :: Structural a (ArrC (ArrC t)) (ArrC t)+	Unpack :: Structural a (ArrC t) (ArrC (ArrC t))++-- | The 'A' arrow includes a set of primitives that may be executed concurrently.+--   Programs are incrementally optimized as they are put together. A program may be+--   optimized once, and the result saved for repeated use.+--+-- Notes:+--+--   * The exact output of the optimizer is subject to change.+--+--   * The program must be a finite data structure, or optimization will diverge.+data A a t u = A (forall v. Structural a v t -> Structural a v u)++data Equal t u = (t ~ u) => Equal++reassociate :: (Category a) => Structural a u v -> Either (Equal t u) (Structural a t u) -> Structural a t v+reassociate (Comp a Id) = reassociate a+reassociate (Comp a a2) = reassociate a . Right . reassociate a2+reassociate a = either (\Equal -> a) (a.)++-- | Obtain a 'Structural' program from an 'A' program.+unA (A f) = f id++-- | Obtain a 'Structural' program but postcompose with another program. +unA' :: A a u v -> Structural a t u -> Structural a t v+unA' (A f) = f++mapA' :: (ArrowChoice a) => A a t u -> A a (ArrC t) (ArrC u)+mapA' (A f) = mapA (f id)++liftA :: (Category a) => a t u -> A a t u+liftA a = A (\a2 -> case a2 of+	Comp (Lift a2) a3 -> Comp (Lift (a . a2)) a3+	_ -> Lift a . a2)++pack :: (Category a) => A a (ArrC (ArrC t)) (ArrC t)+pack = A (\a -> case a of+	Comp (Map (Comp (Map a) a2)) a3 -> Map a . unA' pack (Map a2 . a3)+	Comp (Map (Map a)) a2 -> Map a . unA' pack a2+	Comp (Map (Comp Pack a)) a2 -> unA' pack (unA' pack (Map a . a2))+	Comp (Map Pack) a2 -> unA' pack (unA' pack a2)+	Comp Unpack a2 -> a2+	_ -> Pack . a)++flatten :: Structural a t u -> Bool+flatten (Comp a a2) = flatten a || flatten a2+flatten Id = False+flatten Unpack = False+flatten Pack = False+flatten Zip = False+flatten Unzip = False+flatten Separate = False+flatten Combine = False+flatten _ = True++-- | Mapping is the primary way of constructing nested data parallel programs.+--   It applies an (arrow) transformation to each element of an array+--   uniformly. A form of flattening transformation is applied to nested+--   maps (following the NESL paper). The flattening transformation converts+--   two levels of 'Map' into one level.+mapA :: (ArrowChoice a) => Structural a t u -> A a (ArrC t) (ArrC u)+mapA (Map a) | flatten a = A (\a2 -> case a2 of+	Comp Unpack a3 -> Unpack . unA' (mapA a) a3+	_ -> Unpack . unA' (mapA a) (unA' pack a2))+mapA (Comp a a2) = mapA a . mapA a2+mapA Id = id+mapA (Product a a2) = zipA . (mapA a *** mapA a2) . unzipA+mapA Unpack = A (\a -> case a of+	Comp Unpack a -> Unpack . (Unpack . a)+	Comp (Map (Comp Pack a)) a2 -> Map a . a2+	Comp (Map Pack) a -> a+	_ -> Map Unpack . a)+mapA Count = A (Comp Unpack) . arr (\(ArrC ar fr) -> ArrC+	(newArray $ concatMap (\(x, n) -> map ((,) x) [0..n-1]) $ elems ar)+	(newArray $ zipWith Node (scanl (\n (_, m) -> n + m) 0 $ elems ar)+		(map (\(_, n) -> newArray [Node 0 (newArray []), Node n (newArray [])]) (elems ar) ++ [newArray []])))+mapA a = A (\a2 -> case a2 of+	Comp (Map a2) a3 -> Comp (Map (reassociate a (Right a2))) a3+	_ -> Comp (Map a) a2)++instance (Category a) => Category (A a) where+	id = A (\a -> a)+	A f . A g = A (f . g)++instance (ArrowChoice a) => Arrow (A a) where+	arr = liftA . arr+	A f *** A g = A (\a -> case a of+		Comp (Product a2 a3) a4 -> Product (f a2) (g a3) . a4+		_ -> Product (f id) (g id) . a)+	first a = a *** id+	second a = id *** a++instance (ArrowChoice a) => ArrowChoice (A a) where+	a +++ a2 = A (\a3 -> case a3 of+		Comp Combine a3 -> Combine . unA' (mapA (unA a) *** mapA (unA a2)) a3+		_ -> Combine . unA' (mapA (unA a) *** mapA (unA a2)) (Separate . a3))+	left a = a +++ id+	right a = id +++ a++instance Show (Structural a t u) where+	showsPrec prec (Map a) = ("Map " ++) . showParen (prec==11) (showsPrec 11 a)+	showsPrec _ (Comp a a2) = showsPrec 11 a . (" . "++) . showsPrec 11 a2+	showsPrec prec (Product a a2) = showParen (prec>=3) (showsPrec 3 a . (" *** "++) . showsPrec 3 a2)+	showsPrec _ Count = ("Count"++)+	showsPrec _ Index = ("Index"++)+	showsPrec _ Zip = ("Zip"++)+	showsPrec _ Unzip = ("Unzip"++)+	showsPrec _ ClearMarks = ("Clr"++)+	showsPrec _ Pack = ("Pk"++)+	showsPrec _ Unpack = ("Unpk"++)+	showsPrec _ Separate = ("Sep"++)+	showsPrec _ Combine = ("Comb"++)+	showsPrec _ _ = ("_"++)++instance (Category a) => Category (Structural a) where+	id = Id+	(.) = Comp++mirror ei = either Right Left ei++-- | Supplies an array of a repeated value paired with the index of each element.+countA = A (Comp Count)++-- | Access one index of an array.+indexA = A (Comp Index)++-- | An operation analogous to 'zip'.+zipA :: (Category a) => A a (ArrC t, ArrC u) (ArrC (t, u))+zipA = A (\a -> case a of+	Comp Unzip a2 -> a2+	Comp (Product (Map a) (Map a2)) a3 -> Map (Product a a2) . unA' zipA a3+	Comp (Product (Map a) Id) a3 -> Map (Product a Id) . unA' zipA a3+	Comp (Product Id (Map a2)) a3 -> Map (Product Id a2) . unA' zipA a3+	_ -> Zip . a)++-- | 'unzipA' and 'zipA' are inverses.+unzipA :: (Category a) => A a (ArrC (t, u)) (ArrC t, ArrC u)+unzipA = A (\a -> case a of+	Comp Zip a2 -> a2+	Comp (Map (Product a2 a3)) a4 -> Product (Map a2) (Map a3) . unA' unzipA a4+	_ -> Unzip . a)++concatA :: (Category a) => A a (ArrC (ArrC t)) (ArrC t)+concatA = A (Comp ClearMarks) . pack++forcePair (x, y) = x `seq` y `seq` (x, y)++-- | An evaluator for 'Structural' arrows.+eval0 :: (Concurrent a, Strict a, ArrowChoice a, ?seq :: Bool) => Structural a t u -> a t u+eval0 Count = arr_concF id >>> arr inject+eval0 Index = arr (\(ArrC ar _, i) -> ar ! i)+eval0 Zip = arr (\pr@(ArrC ar _, ArrC ar2 _) -> (pr, (snd (bounds ar) `min` snd (bounds ar2)) + 1))+	>>> arr_concF (arr (\((ArrC ar _, ArrC ar2 _), i) ->+	forcePair (ar ! i, ar2 ! i)))+	>>> arr inject+eval0 Unzip = arr (\ar -> (fmap fst ar, fmap snd ar))+eval0 ClearMarks =+	arr (\(ArrC ar fr) ->+		ArrC ar (newArray [ Node (i + j) fr3 | Node i fr2 <- elems fr, Node j fr3 <- elems fr2 ]))+eval0 (Map a) = (arr (\(ArrC ar _) -> (ar, uncurry subtract (bounds ar) + 1)) >>> arr_concF (arr (uncurry (!)) >>> eval0 a)) &&& arr (\(ArrC _ fr) -> fr) >>> arr (uncurry ArrC)+eval0 Pack = arr (\(ArrC ar _) -> ArrC (newArray $ concatMap (elems . project) $ elems ar)+	(newArray $ zipWith Node (scanl (\i (ArrC ar _) -> i + rangeSize (bounds ar)) 0 $ elems ar)+		(map (\(ArrC _ fr) -> fr) (elems ar) ++ [newArray []])))+eval0 Unpack = arr (\ arc@(ArrC _ fr) -> (arc, uncurry subtract (bounds fr))) >>> arr_concF (arr (\(ArrC ar fr, index) ->+	let+		Node i fr2 = fr ! index+		Node j _ = fr ! (index + 1) in+	ArrC (ixmap (0, j-i-1) (+i) ar) fr2))+	>>> arr inject+eval0 Separate = arr (\ei -> ((,) $! either (\x -> inject $ newArray [x]) (\_ -> inject $ newArray []) ei) $! either (\_ -> inject $ newArray []) (\x -> inject $ newArray [x]) ei)+eval0 Combine = arr (\(ar, ar2) -> let+	a1 = project ar+	a2 = project ar2 in+	if uncurry subtract (bounds (project ar)) == 0 then Left $! a1 ! 0 else Right $! a2 ! 0)+eval0 (Comp a a2) = force (eval0 a) . eval0 a2+eval0 Id = id+eval0 (Lift a) = a+eval0 (Product a a2) = arr forcePair . force (second (eval0 a2)) . arr forcePair . first (eval0 a)++-- | Evaluates arrows.+--+-- Notes:+--+--   * Effects are supported, but with much weaker semantics than the Kleisli arrows+--   of the monad. In particular, the 'Map' and '***' operations are allowed to be parallelized,+--   but on the other hand parallelism is not guaranteed.++eval a = let ?seq = True in eval0 a++instance (Concurrent a, Strict a, ArrowChoice a, ArrowApply a) => ArrowApply (A a) where+	app = first (arr (eval . unA)) >>> liftA app++-------------------------------++checkThreats n positions = n `elem` positions -- Check if there is a piece on the row+	|| n `elem` zipWith (-) positions [1..] -- ... the diagonal+	|| n `elem` zipWith (+) positions [1..] -- ... or the other diagonal++checkThreats2 positions = or [ checkThreats n tl | n:tl <- tails positions ]++nQueensImpl :: Int -> Int -> A (->) [Int] (ArrC [Int])+nQueensImpl _ n | n <= 0 = arr (\soln -> if checkThreats2 soln then inject (newArray []) else inject (newArray [soln]))+nQueensImpl m n = arr (\partialSoln -> (partialSoln, m)) >>> countA >>>+	mapA'+		(arr (uncurry (flip (:))) >>> nQueensImpl m (pred n))+	>>> concatA++nQueens n = arr (\() -> []) >>> nQueensImpl n n++-------------------------------++sorting :: (Ord t) => Int -> A (->) (ArrC t) (ArrC t)+sorting depth | depth <= 0 = arr (inject . newArray . sort . elems . project)+sorting depth = arr (\x -> if uncurry subtract (bounds (project x)) <= 0 then Left x else Right x)+	>>> id+		||| (arr (\ar -> let+			x:xs = elems (project ar)+			(bef, aft) = partition (<x) xs in+			((inject (newArray bef), inject (newArray aft)), x))+			>>> first (s *** s)+			>>> arr (\((bef, aft), x) -> inject (newArray (elems (project bef) ++ x : elems (project aft)))))+	where s = sorting (pred depth) -- Memoize the answer+-- In order to make this recursive function a finite structure, there is a depth limit+-- parameter, beyond which the standard 'sort' takes over.++-------------------------------++permute :: A (->) (ArrC Int) (ArrC Int)+permute = arr (\ar -> (ar, uncurry subtract (bounds (project ar)) + 1)) >>> countA >>> mapA' indexA

@@ -1,4 +1,4 @@-{-# LANGUAGE GADTs, ParallelListComp #-}
+{-# LANGUAGE Trustworthy, GADTs, ParallelListComp, Arrows, ImplicitParams, ScopedTypeVariables #-}
 
 -- | Automatic deadlock prevention.
 --
@@ -6,18 +6,25 @@ -- back or aborted in general.
 --
 -- Instead, I prevent deadlocks before they happen.
-module Control.CUtils.Deadlock (Res(Lift, Acq, Rel, Fork, Plus, Id), run, lft) where
+module Control.CUtils.Deadlock (Res(Lift, Acq, Rel, Fork, Plus, Id), liftK, lft, acq, rel, fork, run) where
 
 import Control.Category
 import Control.Arrow
 import Control.Monad
 import Data.Map (Map)
 import Data.List (inits, tails, elemIndex, deleteBy)
+import Data.Array.IO (IOArray, newArray_, writeArray)
+import Data.Array.Unsafe
+import Data.Array
 import Data.Maybe
 import Data.Function (on)
+import Data.BellmanFord
 import qualified Data.Map as M
 import System.IO.Unsafe
+import Unsafe.Coerce
 import Control.Concurrent
+import Control.CUtils.Conc
+import Control.CUtils.StrictArrow
 import Prelude hiding (id, (.))
 
 -- | The typical sequence that produces a deadlock is as follows:
@@ -51,6 +58,16 @@ 	Plus :: Res t v -> Res u v -> Res (Either t u) v -- choice
 	Id :: Res t t
 
+liftK f = Lift (Kleisli f) Id
+
+lft m = Lift (Kleisli (const m))
+
+acq m = Acq m Id
+
+rel m = Rel m Id
+
+fork a = Fork a
+
 instance Category Res where
 	id = Id
 	a . Lift k a2 = Lift k (a . a2)
@@ -71,6 +88,35 @@ instance ArrowChoice Res where
 	left a = Plus (arr Left . a) (arr Right)
 
+unsafeFreeze' :: IOArray Int e -> IO (Array Int e)
+unsafeFreeze' = unsafeFreeze
+
+instance Concurrent Res where
+	-- This is a cheesy reimplementation of the stuff in .Conc, so that we
+	-- can use (and examine) the 'Fork' primitive.
+	arr_concF_ mnds = proc (parm, n) -> do
+		sem <- Lift (Kleisli $ const $ newQSem 0) Id -< ()
+		recurse -< (sem, parm, n)
+		Lift (Kleisli $ \(sem, n) -> sequence_ (replicate n (waitQSem sem))) Id -< (sem, n) where
+		recurse = proc (sem, parm, n) -> if n <= 0 then
+				returnA -< ()
+			else
+				(Fork (proc (sem, parm, n) -> do
+					mnds -< (parm, n)
+					Lift (Kleisli signalQSem) Id -< sem)
+					recurse) -< (sem, parm, pred n)
+	arr_concF mnds = proc (parm, n) -> do
+		ar <- Lift (Kleisli newArray_) Id -< (0, n-1)
+		arr_concF_ (proc ((ar, parm), n) -> do
+			x <- mnds -< (parm, n)
+			Lift (Kleisli $ \(n, x, ar) -> writeArray ar n x) Id -< (n, x, ar))
+			-< ((ar, parm), n)
+		Lift (Kleisli unsafeFreeze') Id -< ar
+	arr_oneOfF mnds = let ?seq = False in arr_concF mnds >>> arr (! 0)
+
+instance Strict Res where
+	force = Lift (force id)
+
 -- For each thread, we need to track what resources it currently
 -- holds, and for each resource, the resources it may
 -- potentially acquire while holding that resource.
@@ -82,18 +128,15 @@ -- A hazard is a HOLD-ACQUIRE cycle among threads.
 -- We generate all sequences looking for a cycle.
 
-selects ls = [ (y, xs ++ ys) | xs <- inits ls | y:ys <- tails ls ]
+instance Ord (MVar t) where
+	x <= y = (unsafeCoerce x :: Int) <= (unsafeCoerce y :: Int)
 
-generateSequences ls lock = if null ls then
-		return []
-	else do
-		((t, m), xs) <- selects ls
-		lock' <- maybe [] id $ lookup lock m
-		liftM (lock':) $ generateSequences xs lock'
+instance Show (MVar t) where
+	show x = show (unsafeCoerce x :: Int)
 
 -- If there is a hazard, returns a /guard/ for the hazard, i.e.
 -- a lock which avoids the hazard.
-hazard mp m = msum $ map (\(x:xs) -> guard (m `elem` xs) >> return x) $ generateSequences (M.assocs mp) m
+hazard mp m = cycles (M.fromList $ concatMap (\ls -> concatMap (\(x, ls2) -> map (\y -> ((x, y), ())) ls2) ls) $ M.elems mp) m
 
 -- This is the static analysis bit.
 acquired :: Res t u -> MVar () -> [MVar ()]
@@ -130,7 +173,9 @@ 		-- to this thread until the hazard has passed.
 		(\m' -> do
 			putMVar resource mp
-			run (Acq m' $ Rel m' $ Acq m a) x)
+			takeMVar m'
+			putMVar m' ()
+			run (Acq m a) x)
 		may
 run (Rel m a) x = do
 	putMVar m ()
@@ -141,13 +186,17 @@ run (Plus a a2) ei = either (run a) (run a2) ei
 run Id x = return x
 
-lft m = Lift $ Kleisli $ \x -> m >> return x
-
 -- This implements the example above, using the primitives of this library.
 test = do
 	m1 <- newMVar ()
 	m2 <- newMVar ()
-	run (Fork (lft (print "Thd1 done") Id . Rel m1 Id . Rel m2 Id . Acq m1 Id . lft (threadDelay 1000000) Id . Acq m2 Id)
-		(lft (print "Thd2 done") Id . Rel m1 Id . Rel m2 Id . Acq m2 Id . lft (threadDelay 1000000) Id . Acq m1 Id))
+	run (fork (lft (print "Thd1 done") Id . rel m1 . rel m2 . acq m1 . lft (threadDelay 1000000) Id . acq m2)
+		(lft (print "Thd2 done") Id . rel m1 . rel m2 . acq m2 . lft (threadDelay 1000000) Id . acq m1))
 		()
 
+test2 = do
+	m1 <- newMVar ()
+	m2 <- newMVar ()
+	let waltz = rel m1 >>> acq m1 >>> rel m2 >>> acq m2 >>> lft (threadDelay 500000) Id . waltz
+	run (fork (acq m1 >>> acq m2 >>> waltz) (lft (threadDelay 1000000) Id >>> acq m2 >>> lft (print "Done") Id >>> rel m2))
+		()

@@ -1,39 +1,59 @@-{-# LANGUAGE DeriveDataTypeable #-}
+{-# LANGUAGE Trustworthy, DeriveDataTypeable #-}
 
 -- | Functional channels
 -- | A channel data type which allows consumers to hold references to different points in a stream at the same time. Elements of a channel are kept alive only so long as there are references pointing before those elements. And producers on a channel are kept alive only so long as there are consumers.
-module Control.CUtils.FChan (Chan, DoneReadingException(..), takeChan, newChan, makeConsumer) where
+module Control.CUtils.FChan (Chan, listToChan, chanContents, DoneReadingException(..), takeChan, tryTakeChan, newChan, makeConsumer, dupChan) where
 
 import Control.Concurrent.MVar
+import Control.Concurrent (forkIO)
+import Control.Monad
 import Control.Exception
 import System.Mem.Weak
 import Data.Typeable
+import Data.IORef
 
-newtype Chan t = Chan (MVar (t, Chan t))
+import System.IO.Unsafe
 
--- | Thrown by the writer function when the garbage collector detects that no one will read it.
+newtype Chan t = Chan {-# NOUNPACK #-} (MVar (t, Chan t))
+
+-- | Construct a channel from a list.
+{-# NOINLINE listToChan #-}
+listToChan :: [t] -> Chan t
+listToChan (x:xs) = Chan (unsafePerformIO (newMVar (x, listToChan xs)))
+listToChan [] = Chan (unsafePerformIO newEmptyMVar)
+-- Referential transparency is preserved because a means of adding
+-- to a channel is not available unless explicitly provided.
+
+-- | Thrown by the writer function.
 data DoneReadingException = DoneReadingException deriving (Typeable, Show)
 
 instance Exception DoneReadingException
 
-addChan vr x = modifyMVar_ vr (\weak -> do
-	may <- deRefWeak weak
+addChan :: MVar (Chan t) -> t -> IO ()
+addChan vr x = modifyMVar_ vr (\chn -> do
+	may <- return (Just chn)
 	case may of
-		Just vr2 -> do
+		Just (Chan vr2) -> do
 			vr' <- newEmptyMVar
-			putMVar vr2 (x, Chan vr')
-			mkWeak vr' vr' Nothing
+			let chn' = Chan vr'
+			putMVar vr2 (x, chn')
+			-- mkWeak chn' chn' Nothing
+			return chn'
 		Nothing -> throwIO DoneReadingException)
 
 -- | Take the first element from a channel, and a channel representing the remainder of the output.
 takeChan (Chan vr) = readMVar vr
 
+tryTakeChan (Chan vr) = tryReadMVar vr
+
 -- | Create a new channel. The first return value is a function that can be used to add values to the channel. The second return value is the channel itself.
 newChan = do
 	vr <- newEmptyMVar
-	weak <- mkWeak vr vr Nothing
-	vr2 <- newMVar weak
-	return (addChan vr2, Chan vr)
+	vr2 <- newEmptyMVar
+	let chn = Chan vr
+	-- weak <- mkWeak chn chn Nothing
+	putMVar vr2 chn
+	return (addChan vr2, chn)
 
 -- | The first return value is a thunk that returns values from the channel successively, starting from the position of the parameter channel. The second thunk can be used to retrieve the position of the channel after all the reads made using the first thunk.
 makeConsumer chn = do
@@ -42,4 +62,14 @@ 		(x, chn2) <- takeChan chn
 		return (chn2, x)),
 		readMVar vr2)
+
+chanContents :: Chan t -> IO [t]
+chanContents chn = tryTakeChan chn >>= maybe
+	(return [])
+	(\(x, xs) -> liftM (x:) (chanContents xs))
+
+-- | Create a channel which is initially empty, but accumulates new elements.
+dupChan chn = tryTakeChan chn >>= maybe
+	(return chn)
+	(dupChan . snd)
 

@@ -0,0 +1,14 @@+module Control.CUtils.StrictArrow where
+
+import Control.Arrow
+
+-- | Arrows possessing a strictness effect
+class Strict a where
+	force :: a t u -> a t u
+
+instance Strict (->) where
+	force f x = x `seq` f x
+
+instance (Monad m) => Strict (Kleisli m) where
+	force a = Kleisli (\x -> x `seq` runKleisli a x)
+

@@ -0,0 +1,28 @@+module Data.BellmanFord where
+
+import qualified Data.Map as M
+import Data.Maybe
+import Control.Monad
+import Data.List.Extras.Argmax
+
+-- | Edge relaxation.
+relaxEdge nodeWeights edgeWeights x (k, weight) = argmin snd ((k, weight) : map (\y -> (y, maybe (1/0) (\z -> snd (fromJust (M.lookup y nodeWeights)) + z) (M.lookup (y, x) edgeWeights))) (M.keys nodeWeights))
+
+-- | The Bellman-Ford shortest path algorithm.
+bellmanFord :: (Ord a) => M.Map (a, a) Double -> a -> M.Map a (a, Double)
+bellmanFord gr x = foldl (\weights _ -> M.mapWithKey (relaxEdge weights gr) weights) weights [1..M.size gr+1] where
+	keys = map fst (M.keys gr) ++ map snd (M.keys gr)
+	weights = M.insert x (x, 0) $ M.fromList $ map (\x -> (x, (x, 1/0))) keys
+
+retrievePath :: (Ord a) => M.Map a (a, Double) -> a -> a -> [a]
+retrievePath mp a a2 | a == a2 = [a]
+retrievePath mp a a2 = retrievePath mp a (fst (fromJust (M.lookup a2 mp))) ++ [a2]
+
+-- | Cycle finding
+cycles :: (Ord a) => M.Map (a, a) () -> a -> Maybe a
+cycles gr x = do
+	(y, z) <- M.lookup x weights
+	guard (round z < 0)
+	return y where
+	gr' = fmap (const (-1)) gr
+	weights = bellmanFord gr' x