ConcurrentUtils 0.3.0.0 → 0.4.0.0

raw patch · 7 files changed

+457/−157 lines, 7 filesdep +binarydep +bytestringdep +networkdep −mtldep ~containers

Dependencies added: binary, bytestring, network, process

Dependencies removed: mtl

Dependency ranges changed: containers

Files

• ConcurrentUtils.cabal +6/−6
• Control/CUtils/Channel.hs +70/−0
• Control/CUtils/Conc.hs +13/−20
• Control/CUtils/DataParallel.hs +96/−129
• Control/CUtils/Deadlock.hs +8/−2
• Control/CUtils/NetChan.hs +256/−0
• Control/CUtils/Split.hs +8/−0

@@ -1,8 +1,8 @@--- Initial FChan.cabal generated by cabal init.  For further documentation,
---  see http://haskell.org/cabal/users-guide/
+-- Initial ConcurrentUtils.cabal generated by cabal init.  For further 
+-- documentation, see http://haskell.org/cabal/users-guide/
 
 name:                ConcurrentUtils
-version:             0.3.0.0
+version:             0.4.0.0
 synopsis:            Concurrent utilities
 -- description:         
 homepage:            http://alkalisoftware.net
@@ -16,6 +16,6 @@ cabal-version:       >=1.8
 
 library
-  exposed-modules:     Control.CUtils.FChan, Control.CUtils.Processes, Control.CUtils.AList, Control.CUtils.Deadlock, Control.CUtils.DataParallel, Control.CUtils.Conc
-  -- other-modules:
-  build-depends:     base >= 2 && <= 5, containers >= 0.4.0.0, parallel, array, mtl >= 2.0.1.0
+  exposed-modules:     Control.CUtils.Split, 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:       
+  build-depends:       base >=2 && <=5, process, network >=2.4, bytestring, binary, containers, array, parallel

@@ -0,0 +1,70 @@+{-# LANGUAGE FlexibleContexts #-}
+
+-- | A lock-free channel (queue) data structure.
+module Control.CUtils.Channel (Channel, newChannel, writeChannel, readChannel) where
+
+import Control.Concurrent.SampleVar
+import Control.Monad
+import Data.IORef
+import Data.Bits
+import Data.Word
+import Data.List
+import Data.Array.MArray
+
+count pred f = fst . head . dropWhile (not . pred . snd) . zip [0..] . iterate f
+
+nBits x = count (==0) (`shiftR` 1) x
+
+data Channel a t = Channel
+	Word32
+	(a Word32 t)
+	(IORef Word32){-maybe filled-}
+	(IORef Word32){-filled-}
+	(IORef Word32){-maybe empty-}
+	(IORef Word32){-empty-}
+	(SampleVar ()){-full lock-}
+	(SampleVar ()){-empty lock-}
+
+-- | Create a channel with a buffer at least as big as 'buffer'.
+newChannel :: (MArray a t IO) => Word32 -> IO (Channel a t)
+newChannel buffer = do
+	-- Adjust the buffer up to the next power of two.
+	let buffer' = shiftL 1 (nBits buffer)
+
+	-- Create array
+	a <- newArray_ (0, buffer' - 1)
+	
+	-- Create indices
+	mf <- newIORef 0
+	f <- newIORef 0
+	me <- newIORef 0
+	e <- newIORef 0
+
+	-- Create locks
+	fl <- newSampleVar ()
+	el <- newEmptySampleVar
+
+	return (Channel buffer' a mf f me e fl el)
+
+increment ref = atomicModifyIORef ref (\x -> (x + 1, x))
+
+alg buffer off a mx x my y lx ly f = do
+	mXN <- increment mx
+	let spin = do
+		yN <- readIORef y
+		if yN + off <= mXN then do
+				mYN <- readIORef my
+				when (yN == mYN) $ readSampleVar lx
+				spin
+			else do
+				val <- f a (mXN `mod` buffer)
+				increment x
+				writeSampleVar ly ()
+				return val
+	spin
+
+-- | Write into the channel, blocking when the buffer is full.
+writeChannel (Channel buffer a mf f me e fl el) x = alg buffer buffer a mf f me e fl el (\a i -> writeArray a i x)
+
+-- | 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

@@ -51,22 +51,22 @@ 			chanToList []
 	unless (null exslst) $ throwIO (ExceptionList exslst)
 
--- | Runs an associative folding function on the given list. 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. (a -> a -> IO a) -> Array Int a -> IO a
-assocFold f parm = do
+-- | 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 a))
+	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 f (parm ! x) (map (parm !) [x+1..y]) >>= writeArray ar i) $ zip [0..] (divideUp numCapabilities (rangeSize (bounds parm)))
+			innerExHandler $ foldM (\x -> f x . g . (parm !)) init [x..y] >>= writeArray ar i) $ zip [0..] (divideUp numCapabilities (rangeSize (bounds parm)))
 	getExceptions exs
-	(x:xs) <- getElems ar
-	foldM f x xs
+	ls <- getElems ar
+	foldM f init ls
 
 -- |
 concF_ :: (?seq :: Bool) => Int -> (Int -> IO ()) -> IO ()
@@ -94,18 +94,11 @@ 		writeArray res i x)
 	unsafeFreeze' res
 
--- | The next three functions take an implicit parameter ?seq. Set it to True
+-- | 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/,
--- good for speed). if you need all the actions to be executed concurrently,
+-- good for speed). If you need all the actions to be executed concurrently,
 -- set it to False.
 --
--- These functions promise O(m f(n)/c) time, provided:
---
---   * unsafeFreeze does a pointer cast (which it doesn't)
---
---   * green threads are created on the same OS thread as the creating
---     thread where possible
---
 -- n is the number of computations which are indexed from 0 to n - 1.
 concF n = partConcF (0, n - 1) (concF_ n)
 
@@ -113,11 +106,11 @@ conc mnds = partConcF (bounds mnds) (\f -> partConc_ f mnds) (mnds !)
 
 -- | Version of concF specialized for two computations.
-concP m m2 = liftM ((\[Left x, Right y] -> (x, y)) . elems)
+concP m m2 = let ?seq = False in liftM ((\[Left x, Right y] -> (x, y)) . elems)
 	$ concF 2 (\i -> if i == 0 then
-				liftM Left m
-			else
-				liftM Right m2)
+			liftM Left m
+		else
+			liftM Right m2)
 
 partOneOfF bnds mnds = do
 	thds <- newIORef []

@@ -1,161 +1,128 @@-{-# LANGUAGE GADTs, Rank2Types, StandaloneDeriving, ImplicitParams #-}
+{-# LANGUAGE ImplicitParams #-}
 -- | An implementation of nested data parallelism
-module Control.CUtils.DataParallel (ArrC, inject, project, newArray, A(Count, Index, Zip, Unzip, Concat, Map, Comp, Arr, Prod, Sum), optimize, eval) where
+module Control.CUtils.DataParallel where
 
-import Data.Array
+import Data.Array hiding (index)
 import Data.Tree
-import Data.Monoid (Any(Any))
-import Control.Category
-import Control.Arrow
-import Control.Monad.Writer (Writer, tell, runWriter)
-import Control.Monad
 import Control.CUtils.Conc
+import Control.Monad
 import System.IO.Unsafe
-import Prelude hiding (id, (.))
+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
 
-instance Functor ArrC where
-	fmap f (ArrC ar ls) = ArrC (fmap f ar) ls
-
+-- | Convenience for making an array from a list.
 newArray ls = listArray (0, length ls - 1) ls
 
-pairUp ls = zip ls (tail ls)
+mirror x = either Right Left x
 
-instance Show (t -> u) where
-	showsPrec _ _ = ("<FUNCTION>"++)
+pairUp ls = P.zip ls (tail ls)
 
--- | Constructors for caller's use
-data A t u where
-	Count :: A Int (ArrC Int)
-	Index :: A (ArrC t, Int) t
-	Zip :: A (ArrC t, ArrC u) (ArrC (t, u))
-	Unzip :: A (ArrC (t, u)) (ArrC t, ArrC u)
-	Concat :: A (ArrC (ArrC t)) (ArrC t)
-	Map :: A t u -> A (ArrC t) (ArrC u)
-	Comp :: A u v -> A t u -> A t v
-	Arr :: (t -> u) -> A t u
-	Prod :: A t u -> A v w -> A (t, v) (u, w)
-	Sum :: A t u -> A v w -> A (Either t v) (Either u w)
+-- | 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
 
-	-- Internal constructors
-	Id :: A t t
-	Pack :: A (ArrC (ArrC t)) (ArrC t)
-	Unpack :: A (ArrC t) (ArrC (ArrC t))
-	PackProd :: A (t, u) (ArrC (Either t u))
-	UnpackProd :: A (ArrC (Either t u)) (t, u)
-	PackSum1 :: A (Either t (ArrC u)) (ArrC (Either t u))
-	UnpackSum1 :: A (ArrC (Either t u)) (Either t (ArrC u))
-	PackSum2 :: A (Either (ArrC t) u) (ArrC (Either t u))
-	UnpackSum2 :: A (ArrC (Either t u)) (Either (ArrC t) u)
+{-# INLINE [0] count #-}
+count n = inject $ unsafePerformIO $ let ?seq = True in concF n (return $!)
 
-mirror ei = either Right Left ei
+{-# INLINE [0] index #-}
+index (ArrC ar _) i = ar ! i
 
-deriving instance Show (A t u)
+{-# 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))
 
-instance Category A where
-	id = arr id
-	(.) = Comp
+{-# 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
 
-instance Arrow A where
-	arr = Arr
-	(***) = Prod
-	first a = a *** arr id
-	second a = arr id *** a
+-- | Associative fold
+{-# INLINE [0] fold #-}
+fold f g init ar = unsafePerformIO $ assocFold (\y z -> return $! f y z) g init $ project ar
 
-instance ArrowChoice A where
-	(+++) = Sum
-	left a = a +++ arr id
-	right a = arr id +++ a
+-- | Control.Arrow substitutes
+{-# INLINE [0] first #-}
+first f (x, y) = (f x, y)
 
-reassociate :: A u v -> A t u -> A t v
-reassociate (Comp a a2) = reassociate a . reassociate a2
-reassociate x = (x .)
+{-# INLINE [0] second #-}
+second f (x, y) = (x, f y)
 
--- Optimizer step 1. Pushes indexes and concats to the right and separates maps/products/sums.
--- Once this is done, the result should be internal layers of only Maps.
-step :: A t u -> A t u
-step (Comp (Map (Comp a a2)) a3) = step (Map (step a)) . (Map a2 . a3)
-step (Comp (Map (Prod a a2)) a3) = Zip . ((Map a *** Map a2) . (Unzip . a3))
-step (Comp (Map a) a2) = step (Map (step a)) . a2
-step (Comp Index (Prod (Map a) a2)) = step a . (Index . second a2)
-step (Comp Index (Prod Count a)) = arr (\(i, j) -> if inRange (0, i - 1) j then j else error $ "DataParallel.eval: bad index: " ++ show j) . second a
-step (Comp Concat (Map (Map a))) = step (Map (step a)) . Concat
-step (Comp Concat (Map Concat)) = Concat . Concat
-step (Comp (Prod (Comp a a2) a3) a4) = step (Prod (step a) id) . (Prod a2 a3 . a4)
-step (Comp (Prod a (Comp a2 a3)) a4) = step (Prod id (step a2)) . (Prod a a3 . a4)
-step (Comp (Sum (Comp a a2) a3) a4) = step (Sum (step a) id) . (Sum a2 a3 . a4)
-step (Comp (Sum a (Comp a2 a3)) a4) = step (Sum id (step a2)) . (Sum a a3 . a4)
-step (Comp a (Comp a2 a3)) = case step (a . a2) of Comp a4 a5 -> a4 . step (a5 . a3)
-step a = a
+{-# INLINE [0] left #-}
+left f = either (Left . f) Right
 
--- Optimizer step 2. Replaces nested arrays with the packed representation.
--- The first two steps will be repeated, until there is only one layer of Maps.
-step2 :: A t u -> Writer Any (A t u)
-step2 (Map (Map a)) = tell (Any True) >> liftM ((Unpack .) . (. Pack) . Map) (step2 a)
-step2 (Prod a a2) = tell (Any True) >> liftM ((UnpackProd .) . (. PackProd)) (step2 (Map (Sum a a2)))
--- Sums create the possibility of recursion trees w/ variable depth.
-step2 (Sum a (Map a2)) = tell (Any True) >> liftM2 (\x y -> UnpackSum1 . Map (Sum x y) . PackSum1) (step2 a) (step2 a2)
-step2 (Sum (Map a) a2) = tell (Any True) >> liftM2 (\x y -> arr mirror . UnpackSum1 . Map (Sum y x) . PackSum1 . arr mirror) (step2 a) (step2 a2)
-step2 (Sum a a2) = liftM2 (+++) (step2 a) (step2 a2)
-step2 (Map a) = liftM Map (step2 a)
-step2 (Comp a a2) = liftM2 (.) (step2 a) (step2 a2)
-step2 a = return a
+{-# INLINE [0] right #-}
+right f = either Left (Right . f)
 
--- Optimizer step 3. Removes redundant packs and zips, combines maps/products/sums, pushes zips right.
-step3 :: A t u -> Maybe (A t u)
-step3 (Comp (Map a) (Comp (Map a2) a3)) = Just $ Map (repetition step3 (a . a2)) . a3
-step3 (Comp Zip (Prod (Map a) (Map a2))) = Just $ Map (repetition step3 (a *** a2)) . Zip
-step3 (Comp Zip (Prod Count Count)) = Just $ Map (arr (\x -> (x, x))) . (Count . arr (uncurry min))
-step3 (Comp Zip (Comp Unzip a)) = Just a
-step3 (Comp Pack (Comp Unpack a)) = Just a
-step3 (Comp PackProd (Comp UnpackProd a)) = Just a
-step3 (Comp PackSum1 (Comp UnpackSum1 a)) = Just a
-step3 (Comp PackSum2 (Comp UnpackSum2 a)) = Just a
-step3 (Comp (Sum a a2) (Sum a3 a4)) = Just $ repetition step3 (a . a3) +++ repetition step3 (a2 . a4)
-step3 (Comp a (Comp a2 a3)) = liftM (a .) (step3 (a2 . a3))
-step3 _ = Nothing
+{-# INLINE [0] and #-}
+and f g x = (f x, g x)
 
-repetition f x = maybe x (repetition f) (f 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) ++ [[]]))
 
-repetition2 f x = if b then repetition2 f y else y where
-	(y, Any b) = runWriter (f x)
+{-# 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
 
--- | Optimizes an arrow for parallel execution. The arrow can be optimized once, and the result saved for multiple computations. (The exact output of the optimizer is subject to change.)
---
---   The arrow must be finitely examinable.
-optimize = {-repetition step3 . -}repetition2 (liftM (`reassociate` arr id) . step2 . step) . (`reassociate` arr id)
+{-# INLINE [0] __packProd #-}
+__packProd (x, y) = inject $ newArray [Left x, Right y]
 
-eval0 :: (?seq :: Bool) => A t u -> t -> u
-eval0 Count n = inject $ unsafePerformIO $ concF n (return $!)
-eval0 Index (ArrC ar _, i) = ar ! i
-eval0 Zip (ArrC ar _, ArrC ar2 _) = inject $ unsafePerformIO $ concF (snd (bounds ar) `min` snd (bounds ar2))
-	(\i -> let x = ar ! i; y = ar2 ! i in x `seq` y `seq` return $! (x, y))
-eval0 Unzip ar = (fmap fst ar, fmap snd ar)
-eval0 Concat ar0 = ArrC ar [ Node (i + j) ls3 | Node i ls2 <- ls, Node j ls3 <- ls2 ] where ArrC ar ls = eval0 Pack ar0
-eval0 (Map a) (ArrC ar ls) = ArrC (unsafePerformIO $ conc $ fmap ((return $!) . eval0 a) ar) ls
-eval0 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) ++ [[]]))
-eval0 Unpack (ArrC ar ls) = inject $ newArray $ map
-	(\(Node i ls, Node j _) -> ArrC (ixmap (0, j-i-1) (+i) ar) ls)
-	(pairUp ls)
-eval0 PackProd (x, y) = inject $ newArray [Left x, Right y]
-eval0 UnpackProd ar = (let Left x = project ar ! 0 in x, let Right x = project ar ! 1 in x)
-eval0 PackSum1 (Left x) = inject (newArray [Left x])
-eval0 PackSum1 (Right ar) = fmap Right ar
-eval0 UnpackSum1 ar = either Left (\_ -> Right (fmap (\(Right x) -> x) ar)) (project ar ! 0)
-eval0 PackSum2 ei = fmap mirror $ eval0 PackSum1 $ mirror ei
-eval0 UnpackSum2 ar = mirror $ eval0 UnpackSum1 $ fmap mirror ar
-eval0 (Comp a a2) x = eval0 a $ eval0 a2 x
-eval0 (Arr f) x = f x
-eval0 (Prod a a2) (x, y) = b `seq` c `seq` (b, c) where b = eval0 a x; c = eval0 a2 y
-eval0 (Sum a a2) ei = either (Left . eval0 a) (Right . eval0 a2) ei
+{-# INLINE [0] __unpackProd #-}
+__unpackProd ar = (case project ar ! 0 of Left x -> x, case project ar ! 1 of Right x -> x)
 
--- | Evaluates arrows.
-eval a = let ?seq = True in eval0 a
+{-# 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
+  #-}
 

@@ -42,6 +42,14 @@ -- a thread.
 
 -- | The Res arrow.
+--
+--   Computations are built with these constructors (and the arrow
+--   interface). The implementation guarantees progress provided:
+--    * Pieces of the arrow that hold locks are finitely examinable,
+--    * threads are programmed to eventually release a lock they hold,
+--    * locks are the only source of deadlock,
+--    * and all locks are used only with the Acq and Rel ctors (which
+--    acquire and release a lock resp.).
 data Res t u where
 	Lift :: Kleisli IO t v -> Res v u -> Res t u
 	Acq :: MVar () -> Res t u -> Res t u -- acquire a lock
@@ -109,8 +117,6 @@ insert x y [] = [(x, y)]
 
 -- | Use this to run computations built in the Res arrow.
---   Pieces of the arrow that hold locks must be finitely examinable,
---   otherwise it doesn't terminate.
 run :: Res t u -> t -> IO u
 run (Lift k a) x = runKleisli k x >>= run a
 run (Acq m a) x = do

@@ -0,0 +1,256 @@+{-# LANGUAGE CPP, ScopedTypeVariables #-}
+
+-- | A channel module with transparent network communication.
+module Control.CUtils.NetChan (NetSend, NetRecv, localHost, newNetChan, newNetSend, newNetRecv, send, recv, recvSend, sendRecv, recvRecv, activateSend, activateRecv) where
+
+-- This module has a strategy for routing around dead nodes. See 'routeAround'.
+
+import System.IO
+import System.Process
+import Data.List (find, isPrefixOf, isInfixOf, (\\))
+import Network
+import Network.Socket (socketToHandle, SockAddr(..))
+import Network.BSD
+import Control.Concurrent
+import Control.Monad
+import Data.ByteString.Lazy (ByteString, hGet, hPut, length, fromChunks, append, empty)
+import qualified Data.ByteString as B
+import Data.Binary
+import qualified Data.Map as M
+import Data.Maybe
+import Data.Char
+import Data.IORef
+import Data.Bits
+import Control.Exception
+import System.IO.Unsafe
+import Prelude hiding (lookup, length, catch)
+
+import Control.CUtils.Split
+
+type Ident = ByteString
+
+{-# NOINLINE serverup #-}
+serverup = unsafePerformIO (newMVar False)
+
+{-# NOINLINE table #-}
+table :: MVar (M.Map Ident (ByteString -> IO ()))
+table = unsafePerformIO (newMVar (M.singleton empty (\_ -> return ())))
+
+data ChannelFibre t = ChannelFibre (MVar Bool) Handle
+
+data NetSend t = NetSend HostName Ident (MVar [HostName]) (MVar [ChannelFibre t])
+
+data NetRecv t = NetRecv Ident (NetSend t) (NetSend HostName) (Chan t)
+
+instance Eq (ChannelFibre t) where
+	ChannelFibre _ hdl == ChannelFibre _ hdl2 = hdl == hdl2
+
+instance Eq (NetSend t) where
+	NetSend _ ident _ _ == NetSend _ ident2 _ _ = ident == ident2
+
+instance Eq (NetRecv t) where
+	NetRecv ident _ _ _ == NetRecv ident2 _ _ _ = ident == ident2
+
+port = 2999
+
+getIPAddress :: String -> Word32
+getIPAddress ip = shiftL n4 24 .|. shiftL n3 16 .|. shiftL n2 8 .|. n1 where
+	[n1,n2,n3,n4] = map read $ split '.' ip
+
+-- Hack - just gets the local IP address
+localHost = liftM (drop 39 . head . dropWhile (not . isPrefixOf "   IPv4") . lines) $ readProcess "ipconfig" [] []
+
+-- The identifier of a channel is determined by the originating host and a host-unique serial number.
+identifier :: String -> Word32 -> Ident
+identifier ip entry = encode (entry, getIPAddress ip)
+
+--- Channel creation.
+
+-- | Creates a new channel, with receive and send ends.
+newNetChan :: (Binary t) => IO (NetRecv t, NetSend t)
+newNetChan = do
+	mp <- readMVar table
+	host <- localHost
+	let ident = identifier host (fromIntegral (M.size mp))
+	liftM2 (,) (__newNetRecv True Nothing ident) (__newNetSend True host ident)
+
+modifyIdent b ident = append (fromChunks [B.pack $ map (fromIntegral . ord) $ if b then "main" else "back"]) ident
+
+__emptyNetSend :: Bool -> NetSend HostName -> HostName -> Ident -> IO (NetSend t)
+__emptyNetSend b backDown hostName ident = do
+	let ident' = modifyIdent b ident
+
+	-- Create a back channel.
+	buffer <- newMVar []
+	-- Fill the buffer immediately, so this host gets the data before downstreams die.
+	if b then do
+			backR <- __newNetRecv False (Just backDown) ident
+			let loop = do
+				host <- recv backR
+				modifyMVar_ buffer (return . (host:))
+
+				loop
+			forkIO loop
+		else
+			return undefined
+
+	mvar <- newMVar []
+	return (NetSend hostName ident' buffer mvar)
+
+__addConnection s@(NetSend _ ident buffer mvar) hostName = do
+	mvar2 <- newMVar False
+
+	-- Open a TCPIP socket to send
+	hdl <- withSocketsDo $ connectTo hostName (PortNumber port)
+	hSetBuffering hdl (BlockBuffering (Just 1024))
+
+	-- Send identifier
+	hPut hdl ident
+
+	-- Send list of upstreams
+	upstreams <- readMVar buffer
+	let bs = encode (hostName : upstreams)
+	hPut hdl $ encode $ length bs
+	hPut hdl bs
+
+	hFlush hdl
+
+	modifyMVar_ mvar (return . (ChannelFibre mvar2 hdl:))
+
+__newNetSend b hostName ident = do
+	s <- if b then
+			__emptyNetSend False undefined "" ident
+		else
+			return undefined
+	s <- __emptyNetSend b s hostName ident
+	__addConnection s hostName
+	return s
+
+-- | Open a channel to another host
+newNetSend hostName = __newNetSend True hostName (identifier hostName 0)
+
+readLoop f hdl = do
+	n <- liftM decode (hGet hdl 8)
+	bs <- hGet hdl n
+	f bs
+	readLoop f hdl
+
+server socket = withSocketsDo $ do
+	-- Accept loop
+	let loop = do
+		(hdl, host, _) <- accept socket
+		ident <- hGet hdl 12
+		may <- liftM (M.lookup ident) $ readMVar table
+		maybe
+			(hPutStrLn stderr ("The host " ++ host ++ " used an invalid Ident: " ++ show ident))
+			(\f -> forkIO (withSocketsDo (readLoop f hdl)) >> return ())
+			may
+		loop
+
+	loop
+
+__newNetRecv :: (Binary t) => Bool -> Maybe (NetSend t) -> Ident -> IO (NetRecv t)
+__newNetRecv b may ident = do
+	chan <- newChan
+
+	-- Create a back channel
+	--
+	-- The downstream of the back channel is the upstream of the main channel.
+	backS <- if b then
+			__emptyNetSend False undefined "" ident
+		else
+			return undefined
+
+	downstream <- maybe
+		(__emptyNetSend b backS "" ident)
+		return
+		may
+
+	let ident' = modifyIdent b ident
+
+	gotUpstreams <- newIORef False
+	let listener bs = do
+		got <- readIORef gotUpstreams
+		if got then do
+				let x = decode bs
+				writeChan chan x
+
+				-- Send the value to downstream receive ends.
+				send downstream x
+			else do
+				writeIORef gotUpstreams True
+				let x:xs = decode bs
+				when b $ do
+					let NetSend _ _ buffer _ = backS
+					modifyMVar_ buffer (\_ -> return xs)
+					__addConnection backS x
+
+	-- Put a listener in the table.
+	modifyMVar_ table (return . M.insert ident' listener)
+
+	-- Start the server singleton
+	modifyMVar_ serverup (\b -> unless b (withSocketsDo $ listenOn (PortNumber port) >>= forkIO . server >> return ()) >> return True)
+
+	return (NetRecv ident' downstream backS chan)
+
+-- | Creates a receive end of this host's channel. Type unsafe!
+newNetRecv :: (Binary t) => IO (NetRecv t)
+newNetRecv = localHost >>= \host -> __newNetRecv True Nothing (identifier host 0)
+
+--- Send and receive.
+
+-- If send fails, route around the node.
+routeAround fib s@(NetSend _ ident buffer mvar) = do
+	hosts <- modifyMVar buffer (\ls -> return ([], ls))
+	mapM_ (__addConnection s) hosts
+	modifyMVar_ mvar (return . (\\[fib]))
+
+-- | Sends something on a channel.
+send :: (Binary t) => NetSend t -> t -> IO ()
+send snd@(NetSend _ ident _ mvar) x = readMVar mvar >>= mapM_ (\fib@(ChannelFibre mvar hdl) -> do
+	b <- modifyMVar mvar (\b -> let s = encode x in
+		s `seq` catch (hPut hdl (encode (length s)) >> hPut hdl s) (\(_ :: SomeException) -> routeAround fib snd >> send snd x)
+		>> return (True, b))
+	-- Buffering
+	unless b $ void $ forkIO $ do
+		threadDelay 100000
+		modifyMVar_ mvar (\_ -> return False)
+		catch (hFlush hdl) (\(_ :: SomeException) -> routeAround fib snd >> send snd x))
+
+-- | Receives something from a channel.
+recv (NetRecv _ _ _ chan) = readChan chan
+
+--- Sending and receiving channels.
+
+-- | Receives the send end of a channel, on a channel.
+recvSend r = recv r >>= activateSend
+
+-- | Sends the receive end of a channel, on a channel.
+sendRecv s@(NetSend hostName _ _ mvar) r@(NetRecv ident s2 backS _) = do
+	send s r
+
+	-- This node is now responsible for passing on messages to the destination(s).
+	__addConnection s2 hostName
+
+	-- Inform upstream of this
+	send backS hostName
+
+-- | Receives the receive end of a channel, on a channel.
+recvRecv r = recv r >>= activateRecv
+
+--- Channel data utilities.
+
+instance Binary (NetSend t) where
+	put (NetSend hostName ident _ _) = put hostName >> put ident
+	get = liftM2 (\x y -> NetSend x y undefined undefined) get get
+
+instance Binary (NetRecv t) where
+	put (NetRecv ident _ _ _) = put ident
+	get = liftM (\x -> NetRecv x undefined undefined undefined) get
+
+-- | 'get' produces channel ends with some data missing. Use these to make them usable.
+activateSend :: NetSend t -> IO (NetSend t)
+activateSend (NetSend hostName ident _ _) = __newNetSend True hostName ident
+
+activateRecv :: (Binary t) => NetRecv t -> IO (NetRecv t)
+activateRecv (NetRecv x _ _ _) = __newNetRecv True Nothing x

@@ -0,0 +1,8 @@+module Control.CUtils.Split where
+
+split x (y:ys)
+	| x == y	= [] : split x ys
+	| otherwise	=
+		let z:zs = split x ys in
+			(y : z) : zs
+split _ [] = [[]]