chp 1.0.1 → 1.0.2
raw patch · 14 files changed
+324/−30 lines, 14 filesdep +parallel
Dependencies added: parallel
Files
- Control/Concurrent/CHP.hs +2/−0
- Control/Concurrent/CHP/Alt.hs +9/−2
- Control/Concurrent/CHP/Arrow.hs +205/−0
- Control/Concurrent/CHP/Base.hs +23/−8
- Control/Concurrent/CHP/BroadcastChannels.hs +2/−0
- Control/Concurrent/CHP/CSP.hs +3/−1
- Control/Concurrent/CHP/Channels.hs +22/−1
- Control/Concurrent/CHP/Common.hs +9/−0
- Control/Concurrent/CHP/Console.hs +13/−13
- Control/Concurrent/CHP/Event.hs +7/−0
- Control/Concurrent/CHP/Monad.hs +3/−0
- Control/Concurrent/CHP/Parallel.hs +5/−3
- Control/Concurrent/CHP/Utils.hs +18/−0
- chp.cabal +3/−2
Control/Concurrent/CHP.hs view
@@ -30,6 +30,8 @@ -- | This module re-exports the core functionality of the CHP library. Other -- modules that you also may wish to import are: --+-- * "Control.Concurrent.CHP.Arrow"+-- -- * "Control.Concurrent.CHP.Buffers" -- -- * "Control.Concurrent.CHP.Common"
Control/Concurrent/CHP/Alt.hs view
@@ -204,7 +204,7 @@ -- The list of guards without any NestedGuards or StopGuards: flattenedGuards :: [(Int, Guard)]- flattenedGuards = (flatten $ zip [0..] $ map (fst . getAltable) items)+ flattenedGuards = (flatten $ zip [0..] $ map (fst . pullOutAltable) items) where flatten :: [(Int, Guard)] -> [(Int,Guard)] flatten [] = []@@ -230,6 +230,10 @@ storeChoice :: [Int] -> TraceT IO () storeChoice ns = modify (\(_, es) -> (ns, es)) + isBadGuard :: Guard -> Bool+ isBadGuard BadGuard = True+ isBadGuard _ = False+ -- Performs the select operation on all the guards. The choice is stored -- in the state ready to execute the bodies selectFromGuards :: TraceT IO ()@@ -237,6 +241,9 @@ | null eventGuards = do (_,ns) <- liftIO $ waitNormalGuards retry storeChoice ns+ | any isBadGuard wrappedGuards+ = liftIO $ do hPutStrLn stderr "ALTing not supported on given guard"+ ioError $ userError "ALTing not supported on given guard" | otherwise = do earliestReady <- liftIO $ atomically checkNormalGuards tv <- liftIO . atomically $ newTVar Nothing@@ -294,7 +301,7 @@ case st of ((g:gs), es) -> do put (gs, es)- snd $ getAltable (items !! g)+ snd $ pullOutAltable (items !! g) ([], _) -> liftIO $ do hPutStrLn stderr "ALTing not supported on given guard" ioError $ userError "ALTing not supported on given guard"
+ Control/Concurrent/CHP/Arrow.hs view
@@ -0,0 +1,205 @@+-- 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.+++-- | Provides an instance of Arrow for process pipelines. As described in+-- the original paper on arrows, they can be used to represent stream processing,+-- so CHP seemed like a possible fit for an arrow.+-- +-- Whether this is /actually/ an instance of Arrow depends on technicalities.+-- This can be demonstrated with the arrow law @arr id >>> f = f = f >>> arr+-- id@. Whether CHP satisfies this arrow law depends on the definition of+-- equality.+--+-- * If equality means that given the same input value, both arrows produce the+-- same corresponding output value, this is an arrow.+--+-- * If equality means you give the arrows the same single input and wait for the single output,+-- and the output is the same, this is an arrow.+--+-- * If equality means that you can feed the arrows lots of inputs (one after+-- the other) and the behaviour should be the same with regards to communication,+-- this is not an arrow.+--+-- The problem lies in the buffering inherent in arrows. Imagine if @f@ is+-- a single function. @f@ is effectively a buffer of one. You can feed it+-- a single value, but no more than that until you read its output. However,+-- if you have @arr id >>> f@, that can accept two inputs (one held by the+-- @arr id@ process and one held by @f@) before you must accept the output.+--+-- I am fairly confident that the arrow laws are satisfied for the+-- definition of equality that given the same single input, they will+-- produce the same single output. If you don't worry too much about the+-- behavioural difference, and just take arrows as another way to wire+-- together a certain class of process network, you should do fine.+--+-- All your processes should produce exactly one output per input, or else+-- you will find odd behaviour resulting.+-- +-- Added in version 1.0.2.+module Control.Concurrent.CHP.Arrow (ProcessPipeline, runPipeline) where++import Control.Arrow+import Control.Monad++import Control.Concurrent.CHP+import qualified Control.Concurrent.CHP.Common as CHP+import Control.Concurrent.CHP.Utils++-- | The type that is an instance of 'Arrow' for process pipelines. See 'runPipeline'.+data ProcessPipeline a b = ProcessPipeline+ { runPipeline :: Chanin a -> Chanout b -> CHP ()+ -- ^ Given a 'ProcessPipeline' (formed using its 'Arrow' instance) and+ -- the channels to plug into the ends of the pipeline, returns the process+ -- representing the pipeline.+ --+ -- The pipeline will run forever (until poisoned) and you must run it in+ -- parallel to whatever is feeding it the inputs and reading off the outputs.+ -- Imagine that you want a process pipeline that takes in a pair of numbers,+ -- doubles the first and adds one to the second. You could encode this+ -- in an arrow using:+ -- + -- > runPipeline (arr (*2) *** arr (+1))+ --+ -- Arrows are more useful where you already have processes written that+ -- process data and you want to easily wire them together. The arrow notation+ -- is probably easier for doing that than declaring all the channels yourself+ -- and composing everything in parallel.+ }++instance Arrow ProcessPipeline where+ arr = ProcessPipeline . CHP.map+ (ProcessPipeline p) >>> (ProcessPipeline q) = ProcessPipeline (p |->| q)++ first (ProcessPipeline p) = ProcessPipeline $ \in_ out -> do+ c <- newChannel+ c' <- newChannel+ d <- newChannel+ runParallel_+ [ CHP.split in_ (writer c) (writer d)+ , p (reader c) (writer c')+ , CHP.join (,) (reader c') (reader d) out+ ]++ second (ProcessPipeline p) = ProcessPipeline $ \in_ out -> do+ c <- newChannel+ c' <- newChannel+ d <- newChannel+ runParallel_+ [ CHP.split in_ (writer d) (writer c)+ , p (reader c) (writer c')+ , CHP.join (,) (reader d) (reader c') out+ ]++ (ProcessPipeline p) *** (ProcessPipeline q) = ProcessPipeline $ \in_ out -> do+ c <- newChannel+ c' <- newChannel+ d <- newChannel+ d' <- newChannel+ runParallel_+ [ CHP.split in_ (writer c) (writer d)+ , p (reader c) (writer c')+ , q (reader d) (writer d')+ , CHP.join (,) (reader c') (reader d') out+ ]++ (ProcessPipeline p) &&& (ProcessPipeline q) = ProcessPipeline $ \in_ out -> do+ c <- newChannel+ c' <- newChannel+ d <- newChannel+ d' <- newChannel+ runParallel_+ [ CHP.parDelta in_ [writer c, writer d]+ , p (reader c) (writer c')+ , q (reader d) (writer d')+ , CHP.join (,) (reader c') (reader d') out+ ]++instance ArrowChoice ProcessPipeline where+ left (ProcessPipeline p) = ProcessPipeline $ \in_ out -> do+ c <- oneToOneChannel+ d <- oneToOneChannel+ (forever $ do x <- readChannel in_+ case x of+ Left l -> do writeChannel (writer c) l+ l' <- readChannel (reader d)+ writeChannel out (Left l')+ Right r -> writeChannel out (Right r)+ ) <||> p (reader c) (writer d)+ return ()++ right (ProcessPipeline p) = ProcessPipeline $ \in_ out -> do+ c <- oneToOneChannel+ d <- oneToOneChannel+ (forever $ do x <- readChannel in_+ case x of+ Right r -> do writeChannel (writer c) r+ r' <- readChannel (reader d)+ writeChannel out (Right r')+ Left l -> writeChannel out (Left l)+ ) <||> p (reader c) (writer d)+ return ()++ (ProcessPipeline p) ||| (ProcessPipeline q)+ = ProcessPipeline $ \in_ out -> do+ c <- oneToOneChannel+ c' <- oneToOneChannel+ d <- oneToOneChannel+ d' <- oneToOneChannel+ runParallel_+ [ forever $ do x <- readChannel in_+ x' <- case x of+ Left l -> do writeChannel (writer c) l+ readChannel (reader c')+ Right r -> do writeChannel (writer d) r+ readChannel (reader d')+ writeChannel out x'+ , p (reader c) (writer c')+ , q (reader d) (writer d')+ ]++ (ProcessPipeline p) +++ (ProcessPipeline q)+ = ProcessPipeline $ \in_ out -> do+ c <- oneToOneChannel+ c' <- oneToOneChannel+ d <- oneToOneChannel+ d' <- oneToOneChannel+ runParallel_+ [ forever $ do x <- readChannel in_+ x' <- case x of+ Left l -> do writeChannel (writer c) l+ l' <- readChannel (reader c')+ return (Left l')+ Right r -> do writeChannel (writer d) r+ r' <- readChannel (reader d')+ return (Right r')+ writeChannel out x'+ , p (reader c) (writer c')+ , q (reader d) (writer d')+ ]
Control/Concurrent/CHP/Base.hs view
@@ -66,7 +66,7 @@ newtype CHP a = PoisonT (ErrorT PoisonError CHP' a) deriving (Monad, MonadIO) -data CHP' a = AltableT {+data CHP' a = AltableTRet a | AltableT { -- The guard, and body to execute after the guard getAltable :: (Guard, TraceT IO a), -- The body to execute without a guard@@ -101,11 +101,26 @@ class Poisonable c where -- | Poisons the given item. poison :: MonadCHP m => c -> m ()+ -- | Checks if the given item is poisoned. If it is, a poison exception+ -- will be thrown.+ --+ -- Added in version 1.0.2.+ checkForPoison :: MonadCHP m => c -> m () -- ========== -- Functions: -- ========== +pullOutStandard :: CHP' a -> TraceT IO a+pullOutStandard m = case m of+ AltableTRet x -> return x+ AltableT _ st -> st++pullOutAltable :: CHP' a -> (Guard, TraceT IO a)+pullOutAltable m = case m of+ AltableTRet x -> (badGuard, return x)+ AltableT alt _ -> alt+ liftTrace :: TraceT IO a -> CHP' a liftTrace m = AltableT (badGuard, m) m @@ -168,7 +183,7 @@ runCHPProgramWith :: TraceStore -> (TraceStore -> t) -> CHP a -> IO (Maybe a, t) runCHPProgramWith start f (PoisonT p)- = do (x, (_, t)) <- runStateT (liftM (either (const Nothing) Just) $ getStandard $ runErrorT p) ([], start)+ = do (x, (_, t)) <- runStateT (liftM (either (const Nothing) Just) $ pullOutStandard $ runErrorT p) ([], start) return (x, f t) runCHPProgramWith' :: SubTraceStore -> (ChannelLabels -> SubTraceStore -> IO t) -> CHP a -> IO (Maybe a, t)@@ -209,13 +224,13 @@ instance Monad CHP' where -- m :: AltableT g m a -- f :: a -> AltableT g m b- m >>= f = - let ~(grd, altBody) = getAltable m- nonAlt = getStandard m- altBody' = altBody >>= getStandard . f- nonAlt' = nonAlt >>= getStandard . f+ m >>= f = case m of+ AltableTRet x -> f x+ AltableT (grd, altBody) nonAlt ->+ let altBody' = altBody >>= pullOutStandard . f+ nonAlt' = nonAlt >>= pullOutStandard . f in AltableT (grd, altBody') nonAlt'- return x = AltableT (badGuard, return x) (return x)+ return x = AltableTRet x instance MonadIO CHP' where liftIO m = liftTrace (liftIO m)
Control/Concurrent/CHP/BroadcastChannels.hs view
@@ -118,9 +118,11 @@ instance Poisonable (BroadcastChanout a) where poison (BO (BC (b,_))) = poison $ Enrolled b+ checkForPoison (BO (BC (b,_))) = checkForPoison $ Enrolled b instance Poisonable (Enrolled BroadcastChanin a) where poison (Enrolled (BI (BC (b,_)))) = poison $ Enrolled b+ checkForPoison (Enrolled (BI (BC (b,_)))) = checkForPoison $ Enrolled b newBroadcastChannel :: CHP (BroadcastChannel a) newBroadcastChannel = dontWarnMe {- see above -} $ do
Control/Concurrent/CHP/CSP.hs view
@@ -78,7 +78,7 @@ = do x <- start tr <- liftPoison $ liftTrace get (y, tr') <- liftIO $ bracketOnError (return ()) (const errorEnd) $ const- $ runStateT (getStandard (wrapPoison $ body x)) tr+ $ runStateT (pullOutStandard (wrapPoison $ body x)) tr liftPoison $ liftTrace $ put tr' checkPoison y @@ -150,6 +150,8 @@ instance (Enum phase, Bounded phase, Eq phase) => Poisonable (Enrolled PhasedBarrier phase) where poison (Enrolled (Barrier (e,_))) = liftSTM $ Event.poisonEvent e+ checkForPoison (Enrolled (Barrier (e,_)))+ = liftCHP $ liftSTM (Event.checkEventForPoison e) >>= checkPoison -- | A wrapper (usually around a channel-end) indicating that the inner item -- is shared. Use the 'claim' function to use this type.
Control/Concurrent/CHP/Channels.hs view
@@ -46,7 +46,7 @@ -- which one you needed. module Control.Concurrent.CHP.Channels ( -- * Channel Creation- Chan, Channel(..), newChannelWithLabel, newChannelWR, newChannelRW, ChannelTuple(..),+ Chan, Channel(..), writeChannelStrict, newChannelWithLabel, newChannelWR, newChannelRW, ChannelTuple(..), newChannelList, newChannelListWithLabels, newChannelListWithStem, getChannelIdentifier, -- * Channel-Ends@@ -74,6 +74,7 @@ import Control.Monad import Control.Monad.STM import Control.Monad.Trans+import Control.Parallel.Strategies import Data.Maybe import Data.Unique @@ -110,11 +111,13 @@ startReadChannelC :: c a -> (Event, STM (WithPoison a)) endReadChannelC :: c a -> STM () poisonReadC :: c a -> IO ()+ checkPoisonReadC :: c a -> IO (WithPoison ()) class ChanoutC c a where startWriteChannelC :: c a -> (Event, STM (WithPoison ())) endWriteChannelC :: c a -> a -> STM () poisonWriteC :: c a -> IO ()+ checkPoisonWriteC :: c a -> IO (WithPoison ()) -- | A class used for allocating new channels, and getting the reading and -- writing ends. There is a bijective assocation between the channel, and@@ -164,6 +167,20 @@ -- Functions: -- ========== +-- | A helper function that uses the parallel strategies library (see the paper:+-- Algorithm + Strategy = Parallelism) to make sure that the value sent down+-- a channel is strictly evaluated by the sender before transmission.+--+-- This is useful when you want to write worker processes that evaluate data+-- and send it back to some "harvester" process. By default the values sent+-- back may be unevaluated, and thus the harvester might end up doing the evaluation.+-- If you use this function, the value is guaranteed to be completely evaluated+-- before sending.+--+-- Added in version 1.0.2.+writeChannelStrict :: (NFData a, WriteableChannel chanEnd) => chanEnd a -> a -> CHP ()+writeChannelStrict c x = (writeChannel c $| rnf) x+ chan :: Monad m => m (Unique, c a) -> (c a -> r a) -> (c a -> w a) -> m (Chan r w a) chan m r w = do (u, x) <- m return $ Chan u (r x) (w x)@@ -298,9 +315,11 @@ instance Poisonable (Chanin a) where poison (Chanin c) = liftIO $ poisonReadC c+ checkForPoison (Chanin c) = liftCHP $ liftIO (checkPoisonReadC c) >>= checkPoison instance Poisonable (Chanout a) where poison (Chanout c) = liftIO $ poisonWriteC c+ checkForPoison (Chanout c) = liftCHP $ liftIO (checkPoisonWriteC c) >>= checkPoison instance (Channel r w) => ChannelTuple (Chan r w a, Chan r w a) where newChannels = do c0 <- newChannel@@ -353,6 +372,7 @@ poisonReadC (STMChan (e,tv)) = liftSTM $ do poisonEvent e writeTVar tv PoisonItem+ checkPoisonReadC (STMChan (e,_)) = liftSTM $ checkEventForPoison e instance ChanoutC STMChannel a where startWriteChannelC (STMChan (e,tv))@@ -369,6 +389,7 @@ poisonWriteC (STMChan (e,tv)) = liftSTM $ do poisonEvent e writeTVar tv PoisonItem+ checkPoisonWriteC (STMChan (e,_)) = liftSTM $ checkEventForPoison e instance Channel Chanin Chanout where newChannel = chan stmChannel Chanin Chanout
Control/Concurrent/CHP/Common.hs view
@@ -159,6 +159,15 @@ writeChannel out $ f x y ) `onPoisonRethrow` (poison in0 >> poison in1 >> poison out) +-- | Forever reads a pair from its input channel, then in parallel sends out+-- the first and second parts of the pair on its output channels.+--+-- Added in version 1.0.2.+split :: Chanin (a, b) -> Chanout a -> Chanout b -> CHP ()+split in_ outA outB = (forever $ do+ (a, b) <- readChannel in_+ writeChannel outA a <||> writeChannel outB b+ ) `onPoisonRethrow` (poison in_ >> poison outA >> poison outB) -- | A sorter process. When it receives its first @Just x@ data item, it keeps -- it. When it receieves a second, it keeps the lowest of the two, and sends
Control/Concurrent/CHP/Console.hs view
@@ -32,7 +32,6 @@ import Control.Concurrent import Control.Concurrent.STM-import Control.Exception (Exception(..)) import qualified Control.Exception as C import Control.Monad import Control.Monad.Trans@@ -51,15 +50,9 @@ -- but it's good practice to poison them yourself when you finish. Only ever -- run one of these processes at a time, or undefined behaviour will result. ----- Note: getting the input handler to terminate is surpisingly difficult--- in Haskell. Currently it seems (version 6.8.2) that GHC compiling the--- end program with -threaded requires an extra character to be input--- before the input handler notices poison, whereas GHC compiled (not--- using -threaded) and GHCi act as intended. This is not a problem with--- the whole of the library, only with this consoleProcess. Unfortunately--- this means the one or two poison examples in the tutorial may not--- function correctly. I hope to resolve this problem in the next--- version.+-- When using this process, due to the way that the console handlers are terminated,+-- you may sometimes see a notice that a thread was killed. This is normal behaviour+-- (unfortunately). consoleProcess :: (ConsoleChans -> CHP ()) -> CHP () consoleProcess mainProc = do [cin, cout, cerr] <- replicateM 3 oneToOneChannel@@ -77,7 +70,8 @@ -- Poison won't do it if the handlers are blocked on input or -- output. Therefore we throw them an exception to "knock them -- off" their current action and make them exit.- liftIO $ mapM_ (flip throwTo BlockedIndefinitely) ids+ liftIO yield+ liftIO $ mapM_ killThread ids ] where getId :: TVar (Maybe a) -> CHP a@@ -91,8 +85,14 @@ inHandler :: TVar (Maybe ThreadId) -> Chanout Char -> CHP () inHandler tv c = do liftIO $ myThreadId >>= atomically . writeTVar tv . Just- (forever $ liftIO' getChar >>= writeChannel c)- `onPoisonTrap` (poison c)+ if rtsSupportsBoundThreads+ then (forever $ do ready <- liftIO $ hWaitForInput stdin 100+ checkForPoison c+ when ready $+ liftIO' getChar >>= writeChannel c)+ `onPoisonTrap` (poison c)+ else (forever $ liftIO' getChar >>= writeChannel c)+ `onPoisonTrap` (poison c) outHandler :: TVar (Maybe ThreadId) -> Handle -> Chanin Char -> CHP () outHandler tv h c
Control/Concurrent/CHP/Event.hs view
@@ -154,6 +154,13 @@ when (isNothing x) $ retractOffers [(tv, events)] return x +checkEventForPoison :: Event -> STM (WithPoison ())+checkEventForPoison (Event (_, tv))+ = do x <- readTVar tv+ case x of+ PoisonItem -> return PoisonItem+ _ -> return (NoPoison ())+ poisonEvent :: Event -> STM () poisonEvent (Event (_,tv)) = completeEvent True tv >> return ()
Control/Concurrent/CHP/Monad.hs view
@@ -122,6 +122,9 @@ -- 'waitFor' 0 '</>' 'waitFor' n will select the first guard for any value -- of n (including 0). It is not useful to use two waitFor guards in a -- single 'alt' anyway.+--+-- /NOTE:/ If you wish to use this as part of a choice, you must use @-threaded@+-- as a GHC compilation option (at least under 6.8.2). waitFor :: Int -> CHP () waitFor n = liftPoison $ AltableT (guardWaitFor n, return ()) (liftIO $ threadDelay n) -- TODO maybe fix the above lack of guarantees by keeping timeout guards explicit.
Control/Concurrent/CHP/Parallel.hs view
@@ -39,6 +39,7 @@ import Data.List import Data.Maybe import Data.Ord+import System.IO import Control.Concurrent.CHP.Base import Control.Concurrent.CHP.Traces.Base@@ -89,7 +90,8 @@ a, st)) -> IO (Maybe (Either st (a, st))) wrapProcess (PoisonT proc) unwrapInner = do let inner = runErrorT proc- x <- liftM Just (unwrapInner inner) `C.catch` const (return Nothing)+ x <- liftM Just (unwrapInner inner) `C.catch` (\x -> liftIO (hPutStrLn+ stderr $ "Thread terminated with: " ++ show x) >> return Nothing) case x of Nothing -> return Nothing Just (Left _, st) -> return $ Just $ Left st@@ -104,7 +106,7 @@ (_, trace) <- PoisonT $ lift $ liftTrace get blanks <- liftIO $ blankTraces trace (length processes) liftIO $ - mapM_ forkIO [do y <- wrapProcess p $ flip runStateT ([], btr) . getStandard+ mapM_ forkIO [do y <- wrapProcess p $ flip runStateT ([], btr) . pullOutStandard C.block $ atomically $ do ys <- readTVar c writeTVar c $ (case y of@@ -165,7 +167,7 @@ (_, trace) <- liftCHP $ PoisonT $ lift $ liftTrace get [blank] <- liftIO $ blankTraces trace 1 liftIO $ forkIO $ do- r <- wrapProcess p $ flip runStateT ([], blank) . getStandard+ r <- wrapProcess p $ flip runStateT ([], blank) . pullOutStandard C.block $ atomically $ do (poisonedAlready, n) <- readTVar b writeTVar b $ (poisonedAlready || isNothing r, n - 1)
Control/Concurrent/CHP/Utils.hs view
@@ -73,6 +73,24 @@ wireF :: (r a -> w a -> proc, Chan r w a) -> (w a, [proc]) -> (w a, [proc]) wireF (p, c) (w, ps) = (writer c, p (reader c) w : ps) +-- | A specialised version of 'wirePipeline'. Given a list of processes, composes+-- them into an ordered pipeline, that takes the channel-ends for the sticking+-- out ends of the pipeline and gives a process that returns a list of their+-- results. This is equivalent to 'wirePipeline', with the return value fed+-- to 'runParallel'.+--+-- Added in version 1.0.2.+pipeline :: [Chanin a -> Chanout a -> CHP b] -> Chanin a -> Chanout a -> CHP [b]+pipeline procs in_ out = wirePipeline procs in_ out >>= runParallel++-- | A specialised version of 'wireCycle'. Given a list of processes, composes+-- them into a cycle and runs them all in parallel. This is equivalent to+-- 'wireCycle' with the return value fed into 'runParallel'.+--+-- Added in version 1.0.2.+cycle :: [Chanin a -> Chanout a -> CHP b] -> CHP [b]+cycle procs = wireCycle procs >>= runParallel+ -- | Process composition. Given two processes, composes them into a pipeline, -- like function composition (but with an opposite ordering). The function -- is associative. Using wirePipeline will be more efficient than @foldl1
chp.cabal view
@@ -1,5 +1,5 @@ Name: chp-Version: 1.0.1+Version: 1.0.2 Synopsis: An implementation of concurrency ideas from Communicating Sequential Processes License: BSD3 License-file: LICENSE@@ -18,10 +18,11 @@ Category: Concurrency Build-Type: Simple-Build-Depends: base, containers, mtl, pretty, stm+Build-Depends: base, containers, mtl, parallel, pretty, stm Exposed-modules: Control.Concurrent.CHP Control.Concurrent.CHP.Alt+ Control.Concurrent.CHP.Arrow Control.Concurrent.CHP.Barriers Control.Concurrent.CHP.BroadcastChannels Control.Concurrent.CHP.Buffers