diff --git a/Control/Concurrent/CHP.hs b/Control/Concurrent/CHP.hs --- a/Control/Concurrent/CHP.hs +++ b/Control/Concurrent/CHP.hs @@ -46,7 +46,8 @@ -- -- For an overview of the library, take a look at the CHP tutorial: -- available from --- the main CHP website: +-- the main CHP website: , or take +-- a look at the CHP blog: . module Control.Concurrent.CHP ( module Control.Concurrent.CHP.Alt, module Control.Concurrent.CHP.Barriers, diff --git a/Control/Concurrent/CHP/Alt.hs b/Control/Concurrent/CHP/Alt.hs --- a/Control/Concurrent/CHP/Alt.hs +++ b/Control/Concurrent/CHP/Alt.hs @@ -109,6 +109,7 @@ import Data.List import qualified Data.Map as Map import Data.Maybe +import Data.Monoid import Data.Unique import System.IO @@ -308,7 +309,7 @@ merge StopGuard _ = return StopGuard merge _ StopGuard = return StopGuard merge (EventGuard recx actx esx) (EventGuard recy acty esy) - = return $ EventGuard (\n -> recx n ++ recy n) (actx >> acty) (esx ++ esy) + = return $ EventGuard (\n -> recx n ++ recy n) (actx `mappend` acty) (esx ++ esy) merge _ _ = badGuard "merging unsupported guards" foldM1 :: Monad m => (b -> b -> m b) -> [b] -> m b @@ -398,7 +399,7 @@ Signal PoisonItem -> return () Signal (NoPoison n) -> let EventGuard _ act _ = guards !! n - in act + in actWhenLast act ) ret return ret @@ -441,8 +442,9 @@ makeLookup m u = fromMaybe (error "CHP: Unique not found in alt") $ Map.lookup u m -- The alting barrier guards: -eventGuards :: [Guard] -> [(SignalValue, [Event])] -eventGuards guards = zip (map (Signal . NoPoison) [0..]) [ab | EventGuard _ _ ab <- guards] +eventGuards :: [Guard] -> [((SignalValue, STM ()), [Event])] +eventGuards guards = [((Signal $ NoPoison n, actAlways acts), ab) + | (n, EventGuard _ acts ab) <- zip [0..] guards] -- Waits for one of the normal (non-alting barrier) guards to be ready, diff --git a/Control/Concurrent/CHP/Base.hs b/Control/Concurrent/CHP/Base.hs --- a/Control/Concurrent/CHP/Base.hs +++ b/Control/Concurrent/CHP/Base.hs @@ -61,7 +61,7 @@ -- you are enrolled on a barrier. Enrolled Barriers should never be passed -- to two (or more) processes running in parallel; if two processes synchronise -- based on a single enroll call, undefined behaviour will result. -newtype Enrolled b a = Enrolled (b a) +newtype Enrolled b a = Enrolled (b a) deriving (Eq) -- | The central monad of the library. You can use -- 'Control.Concurrent.CHP.Monad.runCHP' and diff --git a/Control/Concurrent/CHP/BroadcastChannels.hs b/Control/Concurrent/CHP/BroadcastChannels.hs --- a/Control/Concurrent/CHP/BroadcastChannels.hs +++ b/Control/Concurrent/CHP/BroadcastChannels.hs @@ -60,7 +60,7 @@ module Control.Concurrent.CHP.BroadcastChannels (BroadcastChanin, BroadcastChanout, OneToManyChannel, AnyToManyChannel, oneToManyChannel, anyToManyChannel, oneToManyChannelWithLabel, anyToManyChannelWithLabel, ReduceChanin, - ReduceChanout, ManyToOneChannel, ManyToAnyChannel, manyToOneChannel, + ReduceChanout, sameReduceChannel, ManyToOneChannel, ManyToAnyChannel, manyToOneChannel, manyToAnyChannel, manyToOneChannelWithLabel, manyToAnyChannelWithLabel) where @@ -101,14 +101,22 @@ dontWarnMe :: a -> a dontWarnMe = flip const [Agreement, Reading, Neutral] +-- | The Eq instance was added in version 1.4.0. newtype BroadcastChannel a = BC (PhasedBarrier Phase, TVar a) +instance Eq (BroadcastChannel a) where + (BC (_, tvX)) == (BC (_, tvY)) = tvX == tvY + -- | The reading end of a broadcast channel. You must enroll on it before -- you can read from it or poison it. -newtype BroadcastChanin a = BI (BroadcastChannel a) +-- +-- The Eq instance was added in version 1.4.0. +newtype BroadcastChanin a = BI (BroadcastChannel a) deriving (Eq) -- | The writing end of a broadcast channel. -newtype BroadcastChanout a = BO (BroadcastChannel a) +-- +-- The Eq instance was added in version 1.4.0. +newtype BroadcastChanout a = BO (BroadcastChannel a) deriving (Eq) instance Enrollable BroadcastChanin a where enroll c@(BI (BC (b,_))) f = enroll b (\eb -> waitForPhase Neutral eb >> f (Enrolled c)) @@ -118,15 +126,16 @@ return x instance WriteableChannel BroadcastChanout where - extWriteChannel (BO (BC (b, tv))) m + extWriteChannel' (BO (BC (b, tv))) m = do syncBarrierWith (indivRecJust ChannelWrite) $ Enrolled b - m >>= liftIO . atomically . writeTVar tv + (x, r) <- m + liftIO . atomically $ writeTVar tv x syncBarrierWith (const $ const Nothing) $ Enrolled b syncBarrierWith (const $ const Nothing) $ Enrolled b - return () + return r instance ReadableChannel (Enrolled BroadcastChanin) where extReadChannel (Enrolled (BI (BC (b, tv)))) f @@ -160,12 +169,14 @@ newChannel = liftCHP $ do c@(BC (b,_)) <- newBroadcastChannel return $ Chan (getBarrierIdentifier b) (BI c) (BO c) + sameChannel (BI x) (BO y) = x == y instance Channel BroadcastChanin (Shared BroadcastChanout) where newChannel = liftCHP $ do m <- newMutex c <- newChannel return $ Chan (getChannelIdentifier c) (reader c) (Shared (m, writer c)) + sameChannel (BI x) (Shared (_, BO y)) = x == y type OneToManyChannel = Chan BroadcastChanin BroadcastChanout type AnyToManyChannel = Chan BroadcastChanin (Shared BroadcastChanout) @@ -185,15 +196,22 @@ anyToManyChannelWithLabel = newChannelWithLabel - +-- | The Eq instance was added in version 1.4.0. newtype ReduceChannel a = GC (PhasedBarrier Phase, TVar a, (a -> a -> a, a)) +instance Eq (ReduceChannel a) where + (GC (_, tvX, _)) == (GC (_, tvY, _)) = tvX == tvY + -- | The reading end of a reduce channel. -newtype ReduceChanin a = GI (ReduceChannel a) +-- +-- The Eq instance was added in version 1.4.0. +newtype ReduceChanin a = GI (ReduceChannel a) deriving (Eq) -- | The writing end of a reduce channel. You must enroll on it before -- you can read from it or poison it. -newtype ReduceChanout a = GO (ReduceChannel a) +-- +-- The Eq instance was added in version 1.4.0. +newtype ReduceChanout a = GO (ReduceChannel a) deriving (Eq) instance Enrollable ReduceChanout a where enroll c@(GO (GC (b,_,_))) f = enroll b (\eb -> waitForPhase Neutral eb >> f (Enrolled c)) @@ -203,15 +221,16 @@ return x instance WriteableChannel (Enrolled ReduceChanout) where - extWriteChannel (Enrolled (GO (GC (b, tv, (f,_))))) m + extWriteChannel' (Enrolled (GO (GC (b, tv, (f,_))))) m = do syncBarrierWith (indivRecJust ChannelWrite) $ Enrolled b - m >>= liftIO . atomically . \x -> readTVar tv >>= writeTVar tv . f x + (x, r) <- m + liftIO . atomically $ readTVar tv >>= writeTVar tv . f x syncBarrierWith (const $ const Nothing) $ Enrolled b syncBarrierWith (const $ const Nothing) $ Enrolled b - return () + return r instance ReadableChannel ReduceChanin where extReadChannel (GI (GC (b, tv, (_, empty)))) f @@ -241,6 +260,12 @@ liftIO $ atomically $ enrollEvent e tv <- liftIO $ atomically $ newTVar mempty return $ GC (b, tv, (mappend, mempty)) + +-- | The reduce channel version of sameChannel. +-- +-- This function was added in version 1.4.0. +sameReduceChannel :: ReduceChanin a -> ReduceChanout a -> Bool +sameReduceChannel (GI x) (GO y) = x == y type ManyToOneChannel = Chan ReduceChanin ReduceChanout type ManyToAnyChannel = Chan (Shared ReduceChanin) ReduceChanout diff --git a/Control/Concurrent/CHP/CSP.hs b/Control/Concurrent/CHP/CSP.hs --- a/Control/Concurrent/CHP/CSP.hs +++ b/Control/Concurrent/CHP/CSP.hs @@ -52,7 +52,7 @@ -- First engages in event, then executes the body. The returned value is suitable -- for use in an alt -buildOnEventPoison :: (Unique -> (Unique -> Integer) -> Maybe (RecordedIndivEvent Unique)) -> Event.Event -> STM () -> CHP a -> CHP a +buildOnEventPoison :: (Unique -> (Unique -> Integer) -> Maybe (RecordedIndivEvent Unique)) -> Event.Event -> EventActions -> CHP a -> CHP a buildOnEventPoison rec e act body = liftPoison (AltableT (Right [(theGuard, return True)]) (return False)) @@ -76,7 +76,7 @@ syncBarrierWith :: (Unique -> (Unique -> Integer) -> Maybe (RecordedIndivEvent Unique)) -> Enrolled PhasedBarrier phase -> CHP phase syncBarrierWith rec (Enrolled (Barrier (e,tv, fph))) - = buildOnEventPoison rec e incPhase + = buildOnEventPoison rec e (EventActions incPhase (return ())) (liftIO $ atomically $ readTVar tv) where incPhase :: STM () diff --git a/Control/Concurrent/CHP/Channels.hs b/Control/Concurrent/CHP/Channels.hs --- a/Control/Concurrent/CHP/Channels.hs +++ b/Control/Concurrent/CHP/Channels.hs @@ -76,11 +76,13 @@ import Control.Monad.Trans import Control.Parallel.Strategies import Data.Maybe +import Data.Monoid import Data.Unique import Control.Concurrent.CHP.Base import Control.Concurrent.CHP.CSP import Control.Concurrent.CHP.Event +import Control.Concurrent.CHP.Guard import Control.Concurrent.CHP.Monad import Control.Concurrent.CHP.Mutex import Control.Concurrent.CHP.Poison @@ -100,7 +102,7 @@ -- Eq instance added in version 1.1.1 newtype Chanout a = Chanout (STMChannel a) deriving Eq -newtype STMChannel a = STMChan (Event, TVar (WithPoison (Maybe a))) deriving +newtype STMChannel a = STMChan (Event, TVar (WithPoison (Maybe a, Maybe ()))) deriving Eq type OneToOneChannel = Chan Chanin Chanout @@ -119,6 +121,10 @@ -- (extended read action goes here) -- Read releases the writer endReadChannelC :: c a -> STM (WithPoison ()) + + -- First action is to be done as part of the completion: + readChannelC :: c a -> (Event, STM (), STM (WithPoison a)) + poisonReadC :: c a -> IO () checkPoisonReadC :: c a -> IO (WithPoison ()) @@ -132,6 +138,9 @@ -- End waits for the reader to tell us we're done, must be done in a different -- transaction to the send endWriteChannelC :: c a -> STM (WithPoison ()) + + -- First action is to be done as part of the completion: + writeChannelC :: c a -> a -> (Event, STM (), STM (WithPoison ())) poisonWriteC :: c a -> IO () checkPoisonWriteC :: c a -> IO (WithPoison ()) @@ -150,6 +159,11 @@ -- destroy\/de-allocate the channel when it is no longer in use. newChannel :: MonadCHP m => m (Chan r w a) + -- | Determines if two channel-ends refer to the same channel. + -- + -- This function was added in version 1.4.0. + sameChannel :: r a -> w a -> Bool + -- | A class indicating that a channel can be read from. class ReadableChannel chanEnd where -- minimal implementation: extReadChannel -- | Reads from the given reading channel-end @@ -163,12 +177,19 @@ class WriteableChannel chanEnd where -- minimal implementation: extWriteChannel -- | Writes from the given writing channel-end writeChannel :: chanEnd a -> a -> CHP () - writeChannel c x = extWriteChannel c (return x) + writeChannel c x = extWriteChannel c (return x) >> return () -- | Starts the communication, then performs the given extended action, then - -- sends the result of that down the channel + -- sends the result of that down the channel. extWriteChannel :: chanEnd a -> CHP a -> CHP () + extWriteChannel c m = extWriteChannel' c (liftM (flip (,) ()) m) + -- | Like extWriteChannel, but allows a value to be returned from the inner action. + -- + -- This function was added in version 1.4.0. + extWriteChannel' :: chanEnd a -> CHP (a, b) -> CHP b + + -- | A helper class for easily creating several channels of the same type. -- The same type refers not only to what type the channel carries, but -- also to the type of channel (one-to-one no poison, one-to-any with @@ -258,7 +279,7 @@ stmChannel :: MonadIO m => m (Unique, STMChannel a) stmChannel = liftIO $ do e <- newEvent ChannelComm 2 - c <- atomically $ newTVar $ NoPoison Nothing + c <- atomically $ newTVar $ NoPoison (Nothing, Nothing) return (getEventUnique e, STMChan (e,c)) -- | A type-constrained version of newChannel. @@ -321,16 +342,15 @@ instance ReadableChannel Chanin where readChannel (Chanin c) - = let (e, m) = startReadChannelC c in - buildOnEventPoison (indivRecJust ChannelRead) e (return ()) (liftSTM $ - do x <- m - endReadChannelC c - return x) >>= checkPoison + = let (e, mdur, mafter) = readChannelC c in + buildOnEventPoison (indivRecJust ChannelRead) e + (EventActions (return ()) mdur) + (liftSTM mafter) >>= checkPoison extReadChannel (Chanin c) body = let (e, m) = startReadChannelC c in scopeBlock - (buildOnEventPoison (indivRecJust ChannelRead) e (return ()) (liftSTM m) >>= checkPoison) + (buildOnEventPoison (indivRecJust ChannelRead) e mempty (liftSTM m) >>= checkPoison) (\val -> do x <- body val liftSTM $ endReadChannelC c return x) @@ -338,20 +358,20 @@ instance WriteableChannel Chanout where writeChannel (Chanout c) x - = let (e, m) = startWriteChannelC c in - buildOnEventPoison (indivRecJust ChannelWrite) e (return ()) - (liftM2 (++) - (liftSTM $ sequence [m, sendWriteChannelC c x]) - (liftSTM $ sequence [endWriteChannelC c])) - >>= checkPoison . mergeWithPoison - extWriteChannel (Chanout c) body + = let (e, mdur, mafter) = writeChannelC c x in + buildOnEventPoison (indivRecJust ChannelWrite) e + (EventActions (return ()) mdur) (liftSTM mafter) + >>= checkPoison + extWriteChannel' (Chanout c) body = let (e, m) = startWriteChannelC c in scopeBlock (buildOnEventPoison (indivRecJust ChannelWrite) - e (return ()) (liftSTM m) >>= checkPoison) - (const $ sequence [body >>= liftSTM . sendWriteChannelC c - ,liftSTM (endWriteChannelC c)] - >>= checkPoison . mergeWithPoison) + e mempty (liftSTM m) >>= checkPoison) + (const $ do (x, r) <- body + sequence [liftSTM $ sendWriteChannelC c x + ,liftSTM (endWriteChannelC c)] + >>= checkPoison . mergeWithPoison + return r) (poisonWriteC c) @@ -404,14 +424,44 @@ -- Some of this is defensive programming -- the writer should never be able -- to discover poison in the channel variable, for example +consumeData :: TVar (WithPoison (Maybe a, Maybe ())) -> STM (WithPoison a) +consumeData tv = do d <- readTVar tv + case d of + PoisonItem -> return PoisonItem + NoPoison (Nothing, _) -> retry + NoPoison (Just x, a) -> do writeTVar tv $ NoPoison (Nothing, a) + return $ NoPoison x + +sendData :: TVar (WithPoison (Maybe a, Maybe ())) -> a -> STM (WithPoison ()) +sendData tv x = do y <- readTVar tv + case y of + PoisonItem -> return PoisonItem + NoPoison (Just _, _) -> error "CHP: Found data while sending data" + NoPoison (Nothing, a) -> do writeTVar tv $ NoPoison (Just x, a) + return $ NoPoison () + +consumeAck :: TVar (WithPoison (Maybe a, Maybe ())) -> STM (WithPoison ()) +consumeAck tv = do d <- readTVar tv + case d of + PoisonItem -> return PoisonItem + NoPoison (_, Nothing) -> retry + NoPoison (x, Just _) -> do writeTVar tv $ NoPoison (x, Nothing) + return $ NoPoison () + +sendAck :: TVar (WithPoison (Maybe a, Maybe ())) -> STM (WithPoison ()) +sendAck tv = do d <- readTVar tv + case d of + PoisonItem -> return PoisonItem + NoPoison (_, Just _) -> error "CHP: Found ack while placing ack!" + NoPoison (x, Nothing) -> do writeTVar tv $ NoPoison (x, Just ()) + return $ NoPoison () + instance ChaninC STMChannel a where - startReadChannelC (STMChan (e,tv)) = (e, waitForJustOrPoison tv) - endReadChannelC (STMChan (_,tv)) - = do x <- readTVar tv - case x of - PoisonItem -> return PoisonItem - NoPoison _ -> do writeTVar tv $ NoPoison Nothing - return $ NoPoison () + startReadChannelC (STMChan (e,tv)) = (e, consumeData tv) + endReadChannelC (STMChan (_,tv)) = sendAck tv + readChannelC (STMChan (e, tv)) + = (e, sendAck tv >> return (), consumeData tv) + poisonReadC (STMChan (e,tv)) = liftSTM $ do poisonEvent e writeTVar tv PoisonItem @@ -424,14 +474,12 @@ PoisonItem -> return PoisonItem NoPoison _ -> return $ NoPoison ()) sendWriteChannelC (STMChan (_, tv)) val - = do x <- readTVar tv - case x of - PoisonItem -> return PoisonItem - NoPoison _ -> do writeTVar tv $ NoPoison $ Just val - return $ NoPoison () + = sendData tv val endWriteChannelC (STMChan (_, tv)) - = waitForNothingOrPoison tv + = consumeAck tv + writeChannelC (STMChan (e, tv)) val + = (e, sendData tv val >> return (), consumeAck tv) poisonWriteC (STMChan (e,tv)) = liftSTM $ do poisonEvent e @@ -440,20 +488,23 @@ instance Channel Chanin Chanout where newChannel = chan stmChannel Chanin Chanout + sameChannel (Chanin x) (Chanout y) = x == y instance Channel (Shared Chanin) Chanout where newChannel = do m <- newMutex c <- newChannel return $ Chan (getChannelIdentifier c) (Shared (m, reader c)) (writer c) + sameChannel (Shared (_, Chanin x)) (Chanout y) = x == y instance Channel Chanin (Shared Chanout) where newChannel = do m <- newMutex c <- newChannel return $ Chan (getChannelIdentifier c) (reader c) (Shared (m, writer c)) + sameChannel (Chanin x) (Shared (_, Chanout y)) = x == y instance Channel (Shared Chanin) (Shared Chanout) where newChannel = do m <- newMutex m' <- newMutex c <- newChannel return $ Chan (getChannelIdentifier c) (Shared (m, reader c)) (Shared (m', writer c)) - + sameChannel (Shared (_, Chanin x)) (Shared (_, Chanout y)) = x == y diff --git a/Control/Concurrent/CHP/Clocks.hs b/Control/Concurrent/CHP/Clocks.hs --- a/Control/Concurrent/CHP/Clocks.hs +++ b/Control/Concurrent/CHP/Clocks.hs @@ -400,7 +400,7 @@ NoPoison td -> do (td', ev) <- offerTimerData pid ph td checkCompletion u sh ts td' >>= writeTVar tv . NoPoison - return $ waitForJustOrPoison ev + return $ waitForOrPoison id ev -- | Creates a clock that starts at the given time. The Show instance is needed -- to display times in traces. diff --git a/Control/Concurrent/CHP/Event.hs b/Control/Concurrent/CHP/Event.hs --- a/Control/Concurrent/CHP/Event.hs +++ b/Control/Concurrent/CHP/Event.hs @@ -33,6 +33,7 @@ newEvent, newEventUnique, enrollEvent, resignEvent, poisonEvent, checkEventForPoison, testAll) where +import Control.Arrow import Control.Concurrent.STM import Control.Monad import Data.Function @@ -43,6 +44,7 @@ import qualified Data.Traversable as T import Data.Unique import Prelude hiding (seq) +import Test.HUnit hiding (test) import Control.Concurrent.CHP.Poison @@ -102,12 +104,15 @@ newtype OfferSet = OfferSet (SignalVar -- Variable to use to signal when committed , ProcessId -- Id of the process making the offer - , [(SignalValue, Map.Map Event ())]) -- Value to send when committed + , [((SignalValue, STM ()), Map.Map Event ())]) -- Value to send when committed -- A list of all sets of events currently offered instance Eq OfferSet where (==) = (==) `on` (\(OfferSet (tv,_,_)) -> tv) +instance Show OfferSet where + show (OfferSet (_, pid, vs)) = "OfferSet " ++ show (pid, map (first fst) vs) + -- Each event in the map can have three possible values: -- PoisonItem; event is poisoned, can always be completed -- NoPoison True; event has been chosen by previous process, you must choose @@ -128,9 +133,9 @@ = do x <- readTVar (getEventTVar e) case x of PoisonItem -> return PoisonItem - NoPoison (a, n, c) -> do writeTVar (getEventTVar e) $ - NoPoison (a, succ n, c) - return $ NoPoison n + NoPoison (a, !n, c) -> do writeTVar (getEventTVar e) $ + NoPoison (a, succ n, c) + return $ NoPoison n -- | search is /not/ used for discovering offers. It is used for looking for possible -- resolutions to a collection of offer sets. It is pure; it performs no STM actions, @@ -153,7 +158,7 @@ -- it can be ignored). If an event maps to False, it was already ruled -- out by not being chosen in another part of the search, and it cannot -- be chosen by any future parts of the search. Should be empty when first called from the outside. - -> Maybe ( [(SignalVar, SignalValue)] + -> Maybe ( [(SignalVar, SignalValue, STM ())] , Map.Map Event (RecordedEventType, Set.Set ProcessId) ) -- ^ The list of tvars involved with the completion and the signal @@ -188,8 +193,8 @@ and' :: Ord k => Map.Map k Bool -> Bool and' = mapdotall id - tryAll :: [(SignalValue, Map.Map Event ())] - -> Maybe ( [(SignalVar, SignalValue)] + tryAll :: [((SignalValue, STM ()), Map.Map Event ())] + -> Maybe ( [(SignalVar, SignalValue, STM ())] , Map.Map Event (RecordedEventType, Set.Set ProcessId) ) tryAll [] = Nothing @@ -204,7 +209,7 @@ | otherwise = case search offers eventMap' of Nothing -> tryAll next Just (act, resolved) -> Just - (if isNullSignal ns then act else (tv, ns) : act + (if isNullSignal (fst ns) then act else (tv, fst ns, snd ns) : act , foldl (\m e -> Map.insertWith add e (getEventType e, Set.singleton pid) m) resolved (Map.keys es) @@ -238,20 +243,22 @@ search (map addNullOffer $ sortOffers offers') Map.empty eventCounts <- T.sequence $ Map.mapWithKey getAndIncCounter ret let NoPoison uniqCounts = T.sequence $ Map.mapKeysMonotonic getEventUnique eventCounts - mapM_ (\(tv, x) -> writeTVar tv (Just (x, uniqCounts))) act + mapM_ (\(tv, x, m) -> writeTVar tv (Just (x, uniqCounts)) >> m + ) act -- do the retractions for all involved processes once the choice is made: -- TODO optimise: - retractOffers $ zip (map fst act) + retractOffers $ zip (map fst3 act) (repeat $ unionAll $ map allEventsInOffer allOffers) return (Map.mapKeysMonotonic getEventUnique ret) where + fst3 (x, _, _) = x -- Don't add the null offer for the newest process, and null offer should be -- added to the end: addNullOffer :: OfferSet -> OfferSet addNullOffer (OfferSet (tv,y,zs)) = OfferSet (tv,y,if Just tv == newTvid then zs else zs++nullOffer) - nullOffer :: [(SignalValue, Map.Map Event ())] - nullOffer = [(nullSignalValue,Map.empty)] + nullOffer :: [((SignalValue, STM ()), Map.Map Event ())] + nullOffer = [((nullSignalValue, return ()) ,Map.empty)] -- Smallest offers first to minimise backtracking: sortOffers :: [OfferSet] -> [OfferSet] @@ -272,14 +279,16 @@ -- again (while finding the fix point) trim (offers, events) = let ((events', changed), offers') = mapAccumL trimOffer (events, False) offers - in (if changed then trim else id) (offers', events') + oe = (offers', events') + in if changed then trim oe else oe where trimOffer :: (Set.Set Event, Bool) -> OfferSet -> ((Set.Set Event, Bool), OfferSet) - trimOffer (es, changed) (OfferSet (tv, pid, eventSets)) + trimOffer (es, changed) o@(OfferSet (tv, pid, eventSets)) -- An offer is only retained if all the events are in the set of events -- that can possibly complete = let (eventSetsToRemove, eventSetsTrimmed) - = partition (\(_,x) -> not $ (Map.keysSet x) `Set.isSubsetOf` es) eventSets + | Set.size es == 1 = partition (\(_,x) -> Map.size x /= 1 || fst (Map.findMin x) /= Set.findMin es) eventSets + | otherwise = partition (\(_,x) -> not $ (Map.keysSet x) `Set.isSubsetOf` es) eventSets -- If any of the events to remove are not also in sets that will -- be kept, and the event is not poisoned, that event is no longer completable and should be -- removed from the set of events: @@ -288,7 +297,9 @@ `Map.difference` (unionAll $ map snd eventSetsTrimmed) changed' = changed || not (null eventSetsToRemove) - in ((es `Set.difference` eventsNotCompletable, changed'), + in if null eventSetsToRemove then ((es, changed), o) + else + ((es `Set.difference` eventsNotCompletable, changed'), OfferSet (tv, pid, eventSetsTrimmed)) -- Semantics of poison with waiting for multiple events is that if /any/ of @@ -366,7 +377,7 @@ -- be poisoned: (repeat $ retract >> writeTVar tv (Just (addPoison ns, Map.empty))) (Map.keys es) - | (ns, es) <- nes] + | ((ns,_), es) <- nes] Right e -> [(return (), e)] case r of PoisonItem -> return PoisonItem @@ -451,7 +462,7 @@ -- ^ Variable used to signal the process once a choice is made -> ProcessId -- ^ The id of the process making the choice - -> [(SignalValue, [Event])] + -> [((SignalValue, STM ()), [Event])] -- ^ The list of options. Each option has a signalvalue to return -- if chosen, and a list of events (conjoined together). -- So this list is the disjunction of conjunctions, with a little @@ -520,10 +531,10 @@ | OfferSet (tvw, _, events) <- offers] writeTVar (getEventTVar e) PoisonItem where - pickInts :: [(SignalValue, Map.Map Event ())] -> SignalValue + pickInts :: [((SignalValue, STM ()), Map.Map Event ())] -> SignalValue pickInts es = case filter ((e `Map.member`) . snd) es of [] -> nullSignalValue -- Should never happen - ((ns,_):_) -> ns + (((ns,_),_):_) -> ns --TODO document how if it's poisoned, 0 will be appended to the list @@ -541,14 +552,16 @@ (**/=**) :: Eq a => [a] -> [a] -> Bool a **/=** b = not $ a **==** b -testDiscover :: IO () -testDiscover - = do test "Empty discover" [(NoPoison 1, False)] [] [0] +testDiscover :: Test +testDiscover = TestCase $ + do test "Empty discover" [(NoPoison 1, False)] [] [0] test "Single full event" [(NoPoison 1, True)] [(True, [[0]])] [0] test "Two separate events A" [(NoPoison 1, True), (NoPoison 1, False)] [ (True, [[0]]), (False, [[1]]) ] [0] test "Two separate events B" [(NoPoison 1, False), (NoPoison 1, True)] [ (False, [[0]]), (True, [[1]]) ] [1] + test "Two separate events A, non-completable" [(NoPoison 2, False), (NoPoison 1, False)] + [ (False, [[0]]), (False, [[1]]) ] [0] test "Three channels, linked by two OR-offerers" [(NoPoison 2, False), (NoPoison 2, True), (NoPoison 2, False)] @@ -577,21 +590,21 @@ test testName eventCounts offerSets startEvents = do (events, realOffers) <- makeTestEvents (map fst eventCounts) (map snd offerSets) let expectedResult - = ([off | (n,off) <- zip [0..] realOffers, fst $ offerSets !! n] - ,Set.fromList [events !! n - | (n,(_count, present)) <- zip [0..] eventCounts, + = ([off | ((yes, _),off) <- zip offerSets realOffers, yes] + ,Set.fromList [e + | (e,(_count, present)) <- zip events eventCounts, present]) act <- atomically $ discoverRelatedOffers $ zip (repeat $ return ()) $ map (events!!) startEvents case act of - PoisonItem -> putStrLn $ testName ++ "Unexpected poison" + PoisonItem -> assertFailure $ testName ++ "Unexpected poison" NoPoison actualResult -> do when (fst expectedResult **/=** fst actualResult) - $ putStrLn $ testName ++ " failed offers, exp: " + $ assertFailure $ testName ++ " failed offers, exp: " ++ show (length $ fst expectedResult) ++ " got: " ++ show (length $ fst actualResult) when (snd expectedResult /= snd actualResult) - $ putStrLn $ testName ++ " failed events " + $ assertFailure $ testName ++ " failed events " ++ "exp: " ++ show (snd expectedResult) ++ "but got: " ++ show (snd actualResult) test_Poison :: String -> @@ -603,13 +616,13 @@ $ zip (repeat $ return ()) (map (events!!) startEvents) case act of PoisonItem -> return () - NoPoison _ -> putStrLn $ testName ++ " expected poison but none" + NoPoison _ -> assertFailure $ testName ++ " expected poison but none" -testTrim :: IO () -testTrim - = do test "Empty trim" [(NoPoison 1, False)] [] [0] +testTrim :: Test +testTrim = TestCase $ + do test "Empty trim" [(NoPoison 1, False)] [] [0] test "Trim, Three channels, linked by two OR-offerers" [(NoPoison 2, False), (NoPoison 2, True), (NoPoison 2, False)] [ [(False, [0]), (True, [1])] , [(True, [1]), (False, [2])] ] [1] @@ -637,21 +650,21 @@ $ atomically $ discoverRelatedOffers $ zip (repeat $ return ()) (map (events!!) startEvents) case (expectedResult', actualResult') of (PoisonItem, PoisonItem) -> return () - (PoisonItem, _) -> putStrLn $ testName ++ " expected poison but none found" - (_, PoisonItem) -> putStrLn $ testName ++ " unexpected poison" + (PoisonItem, _) -> assertFailure $ testName ++ " expected poison but none found" + (_, PoisonItem) -> assertFailure $ testName ++ " unexpected poison" (NoPoison expectedResult, NoPoison actualResult) -> do when (fst expectedResult **/=** fst actualResult) - $ putStrLn $ testName ++ " failed offers, exp: " + $ assertFailure $ testName ++ " failed offers, exp: " ++ show (length $ fst expectedResult) ++ " got: " ++ show (length $ fst actualResult) when (snd expectedResult /= snd actualResult) - $ putStrLn $ testName ++ " failed events, exp: " + $ assertFailure $ testName ++ " failed events, exp: " ++ show (snd expectedResult) ++ "but got: " ++ show (snd actualResult) -testPoison :: IO () -testPoison = do +testPoison :: Test +testPoison = TestCase $ do test "Poison empty event" [(NoPoison 2, PoisonItem)] [] 0 test "Poison, single offerer" [(NoPoison 2, PoisonItem)] [[[0]]] 0 test "Poison, offered on two (AND)" [(NoPoison 2, PoisonItem), (NoPoison 2, NoPoison [])] [[[0,1]]] 0 @@ -671,19 +684,19 @@ sequence_ [do x <- atomically $ readTVar $ getEventTVar $ events !! n case (expect, x) of (PoisonItem, PoisonItem) -> return () - (NoPoison _, PoisonItem) -> putStrLn $ testName ++ + (NoPoison _, PoisonItem) -> assertFailure $ testName ++ " expected no poison but found it" - (PoisonItem, NoPoison _) -> putStrLn $ testName ++ + (PoisonItem, NoPoison _) -> assertFailure $ testName ++ " expected poison but found none" (NoPoison expOff, NoPoison (_, _, actOff)) -> when (map (realOffers !!) expOff **/=** actOff) $ - putStrLn $ testName ++ " offers did not match" + assertFailure $ testName ++ " offers did not match" | (n, (_, expect)) <- zip [0..] eventCounts] -testAll :: IO () -testAll = testDiscover >> testTrim >> testResolve >> testPoison +testAll :: Test +testAll = TestList [testDiscover, testTrim, testResolve, testPoison] makeTestEvents :: {- Events: -} [WithPoison Int {-count -}] -> @@ -702,7 +715,8 @@ realOffers <- sequence [ do tv <- atomically $ newTVar Nothing let pid = testProcessId processN - offSub = [ (Signal $ NoPoison (processN + offerN), + -- TODO test the STM actions too + offSub = [ ((Signal $ NoPoison (processN + offerN), return ()), Map.fromList [ (events !! indivEvent, ()) | indivEvent <- singleOffer]) | (offerN, singleOffer) <- zip [0..] processOffers] @@ -718,9 +732,9 @@ | (processN, processOffers) <- zip (map (*1000) [0..]) offerSets] return (events, realOffers) -testResolve :: IO () -testResolve - = do test "Empty Resolve" [(NoPoison 0, Right [])] [[]] +testResolve :: Test +testResolve = TestCase $ + do test "Empty Resolve" [(NoPoison 0, Right [])] [[]] test "Single offer" [(NoPoison 1, Left [(0,0)])] [[[0]]] test "Not enough" [(NoPoison 2, Right [0])] [[[0]]] test "One channel" [(NoPoison 2, Left [(0,0),(1,0)])] [[[0]],[[0]]] @@ -769,7 +783,8 @@ test testName eventCounts offerSets = test' testName eventCounts offerSets False test' :: String -> - [(WithPoison Int {-count -}, + -- List of events: + [(WithPoison Int {- enrolled count -}, Either [(Int, Int)] {- success: expected process, offer indexes -} [Int] {- remaining offers -})] -> {- Offers: -} [[[Int] {- events -}]] -> Bool {-Poisoned-} -> IO () @@ -783,17 +798,17 @@ Set.fromList $ map (testProcessId . (*1000) . fst) is)) | (e, Left is) <- zip events (map snd eventCounts)] when (expectedResult /= actualResult) $ - putStrLn $ testName ++ " failed on direct result, expected: " + assertFailure $ testName ++ " failed on direct result, expected: " ++ showStuff expectedResult ++ " got: " ++ showStuff actualResult allFired <- liftM concat $ mapM (flip either (const $ return []) $ mapM $ \(pn, en) -> let OfferSet (tv,_,_) = realOffers !! pn in do x <- atomically $ readTVar tv case x of - Nothing -> putStrLn $ "Unexpected no-win for " ++ show (pn,en) + Nothing -> assertFailure $ "Unexpected no-win for " ++ show (pn,en) Just v -> when (fst v /= (if poisoned then addPoison else id) (Signal $ NoPoison ((pn*1000)+en))) $ - putStrLn $ testName ++ " wrong choice: " ++ " exp: " ++ show + assertFailure $ testName ++ " wrong choice: " ++ " exp: " ++ show (pn+en) return pn ) $ map snd eventCounts @@ -802,21 +817,26 @@ do x <- atomically $ readTVar tv case x of Nothing -> return () - Just _ -> putStrLn $ testName ++ " Unexpected win for process: " ++ + Just _ -> assertFailure $ testName ++ " Unexpected win for process: " ++ show n | n <- [0 .. length offerSets - 1] \\ allFired] -- check events are blanked afterwards: - sequence_ [ let e = events !! n + c <- sequence + [ let e = events !! n expVal = case st of Left _ -> [] Right ns -> map (realOffers !!) ns in do x <- atomically $ readTVar $ getEventTVar e case x of - NoPoison (c, _, e') | c == count && e' == expVal -> return () + NoPoison (c, _, e') -> return $ Just ((count, expVal), (c, e')) + _ -> do assertFailure $ testName ++ " unexpected poison" + return Nothing +{- NoPoison (c, _, e') | c == count && e' == expVal -> return () _ -> - putStrLn $ testName ++ "Event " ++ show n ++ + assertFailure $ testName ++ "Event " ++ show n ++ " not as expected after, exp: " ++ show (length expVal) - ++ " act: " ++ (let NoPoison (_,_,act) = x in show (length act)) + ++ " act: " ++ (let NoPoison (_,_,act) = x in show (length act))-} | (n,(NoPoison count, st)) <- zip [0..] eventCounts] + uncurry (assertEqual testName) (unzip $ catMaybes c) showStuff = show . fmap (map (\(u,x) -> (hashUnique u, x)) . Map.toList) diff --git a/Control/Concurrent/CHP/Guard.hs b/Control/Concurrent/CHP/Guard.hs --- a/Control/Concurrent/CHP/Guard.hs +++ b/Control/Concurrent/CHP/Guard.hs @@ -31,6 +31,7 @@ import Control.Concurrent.STM import Control.Monad.Trans +import Data.Monoid import Data.Unique import System.IO @@ -43,7 +44,14 @@ | StopGuard -- The STM item is an action to take in the same transaction as -- completing the event (before it is completed). - | EventGuard ((Unique -> Integer) -> [RecordedIndivEvent Unique]) (STM ()) [Event] + | EventGuard ((Unique -> Integer) -> [RecordedIndivEvent Unique]) EventActions [Event] + +data EventActions = EventActions { actWhenLast :: STM () + , actAlways :: STM () } + +instance Monoid EventActions where + mempty = EventActions (return ()) (return ()) + mappend (EventActions a a') (EventActions b b') = EventActions (a>>b) (a'>>b') skipGuard :: Guard skipGuard = SkipGuard diff --git a/Control/Concurrent/CHP/Poison.hs b/Control/Concurrent/CHP/Poison.hs --- a/Control/Concurrent/CHP/Poison.hs +++ b/Control/Concurrent/CHP/Poison.hs @@ -46,16 +46,13 @@ mergeWithPoison :: [WithPoison a] -> WithPoison () mergeWithPoison = sequence_ -waitForJustOrPoison :: TVar (WithPoison (Maybe a)) -> STM (WithPoison a) -waitForJustOrPoison tv = do x <- readTVar tv - case x of - PoisonItem -> return PoisonItem - NoPoison Nothing -> retry - NoPoison (Just y) -> return $ NoPoison y +waitForOrPoison :: (a -> Maybe b) -> TVar (WithPoison a) -> STM (WithPoison b) +waitForOrPoison f tv = do x <- readTVar tv + case fmap f x of + PoisonItem -> return PoisonItem + NoPoison Nothing -> retry + NoPoison (Just y) -> return $ NoPoison y -waitForNothingOrPoison :: TVar (WithPoison (Maybe a)) -> STM (WithPoison ()) -waitForNothingOrPoison tv = do x <- readTVar tv - case x of - PoisonItem -> return PoisonItem - NoPoison (Just _) -> retry - NoPoison Nothing -> return $ NoPoison () +flipMaybe :: Maybe a -> Maybe () +flipMaybe (Just _) = Nothing +flipMaybe Nothing = Just () diff --git a/Control/Concurrent/CHP/Test.hs b/Control/Concurrent/CHP/Test.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/Test.hs @@ -0,0 +1,131 @@ +-- Communicating Haskell Processes. +-- Copyright (c) 2009, 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. + +-- | A module containing some useful functions for testing CHP programs, both in +-- the QuickCheck 2 framework and using HUnit. +-- +-- This whole module was added in version 1.4.0. +module Control.Concurrent.CHP.Test where + +import Control.Monad +import Data.Maybe +import Test.HUnit (assertBool, Test(..)) +import Test.QuickCheck (Gen, Property) +import Test.QuickCheck.Monadic (assert, forAllM, monadicIO, run) + +import Control.Concurrent.CHP + +-- | Takes a CHP program that returns a Bool (True = test passed, False = test +-- failed) and forms it into a Property that QuickCheck can test. +-- +-- Note that if the program exits with poison, this is counted as a test failure. +propCHP :: CHP Bool -> Property +propCHP = monadicIO . (>>= assert . fromMaybe False) . run . runCHP + +-- | Tests a process that takes a single input and produces a single output, using +-- QuickCheck. +-- +-- The first parameter is a pure function that takes the input to the process, +-- the output the process gave back, and indicates whether this is okay (True = +-- test pass, False = test fail). The second parameter is the process to test, +-- and the third parameter is the thing to use to generate the inputs (passing 'arbitrary' +-- is the simplest thing to do). +-- +-- Here are a couple of example uses: +-- +-- > propCHPInOut (==) Common.id (arbitrary :: Gen Int) +-- +-- > propCHPInOut (const $ (< 0)) (Common.map (negate . abs)) (arbitrary :: Gen Int) +-- +-- The test starts the process afresh each time, and shuts it down after the single +-- output has been produced (by poisoning both its channels). Any poison from +-- the process being tested after it has produced its output is consequently ignored, +-- but poison instead of producing an output will cause a test failure. +-- If the process does not produce an output or poison (for example if you test +-- something like the Common.filter process), the test will deadlock. +propCHPInOut :: Show a => (a -> b -> Bool) -> (Chanin a -> Chanout b -> CHP ()) -> Gen a -> Property +propCHPInOut f p gen + = monadicIO $ forAllM gen $ \x -> run (runCHP $ + do c <- oneToOneChannel + d <- oneToOneChannel + (_,r) <- (p (reader c) (writer d) + `onPoisonTrap` (poison (reader c) >> poison (writer d))) + <||> ((do writeChannel (writer c) x + y <- readChannel (reader d) + poison (writer c) >> poison (reader d) + return $ f x y + ) `onPoisonTrap` return False) + return r) >>= assert . fromMaybe False + +-- | Takes a CHP program that returns a Bool (True = test passed, False = test +-- failed) and forms it into an HUnit test. +-- +-- Note that if the program exits with poison, this is counted as a test failure. +testCHP :: CHP Bool -> Test +testCHP = TestCase . (>>= assertBool "testCHP failure" . fromMaybe False) . runCHP + +-- | Tests a process that takes a single input and produces a single output, using +-- HUnit. +-- +-- The first parameter is a pure function that takes the input to the process, +-- the output the process gave back, and indicates whether this is okay (True = +-- test pass, False = test fail). The second parameter is the process to test, +-- and the third parameter is the input to send to the process. +-- +-- The intention is that you will either create several tests with the same first +-- two parameters or use a const function as the first parameter. So for example, +-- here is how you might test the identity process with several tests: +-- +-- > let check = testCHPInOut (==) Common.id +-- > in TestList [check 0, check 3, check undefined] +-- +-- Whereas here is how you could test a slightly different process: +-- +-- > let check = testCHPInOut (const $ (< 0)) (Common.map (negate . abs)) +-- > in TestList $ map check [-5..5] +-- +-- The test starts the process afresh each time, and shuts it down after the single +-- output has been produced (by poisoning both its channels). Any poison from +-- the process being tested after it has produced its output is consequently ignored, +-- but poison instead of producing an output will cause a test failure. +-- If the process does not produce an output or poison (for example if you test +-- something like the Common.filter process), the test will deadlock. +testCHPInOut :: (a -> b -> Bool) -> (Chanin a -> Chanout b -> CHP ()) -> a -> Test +testCHPInOut f p x + = testCHP $ do c <- oneToOneChannel + d <- oneToOneChannel + liftM snd $ (p (reader c) (writer d) + `onPoisonTrap` (poison (reader c) >> poison (writer d))) + <||> ((do writeChannel (writer c) x + y <- readChannel (reader d) + poison (writer c) >> poison (reader d) + return $ f x y + ) `onPoisonTrap` return False) + +-- TODO add some better HUnit facilities diff --git a/Control/Concurrent/CHP/Utils.hs b/Control/Concurrent/CHP/Utils.hs --- a/Control/Concurrent/CHP/Utils.hs +++ b/Control/Concurrent/CHP/Utils.hs @@ -28,6 +28,13 @@ -- SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. -- | A collection of useful functions to use with the library. +-- +-- The most useful operation is 'pipeline' which you can use to wire up a list +-- of processes into a line, and run them. The corresponding '|->|' operator is +-- a simple binary version that can be a little more concise. When the pipeline +-- has channels going in both directions rather than just one, 'dualPipeline' and/or +-- '|<->|' can be used. Several other variants on these functions are also provided, +-- including operators to use at the beginning and ends of pipelines. module Control.Concurrent.CHP.Utils where import Control.Monad @@ -43,6 +50,17 @@ wirePipeline procs (reader chan) (writer chan) -- return [p (reader $ chans !! i) (writer $ chans !! ((i + 1) `mod` n)) | (p, i) <- zip procs [0..]] +-- | Like wireCycle, but works with processes that connect with a channel in both +-- directions. +-- +-- This function was added in version 1.4.0. +wireDualCycle :: (Channel r w, Channel r' w') => + [(r a, w' b) -> (r' b, w a) -> proc] -> CHP [proc] +wireDualCycle procs + = do c <- newChannel + d <- newChannel + wireDualPipeline procs (reader c, writer d) (reader d, writer c) + -- | Wires the given processes up in a forward pipeline. The first process -- in the list is connected to the given reading channel-end (the first parameter) -- and the writing end of a new channel, A. The second process is wired up @@ -73,6 +91,25 @@ 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) +-- | Like wirePipeline, but works with processes that connect with a channel in both +-- directions. +-- +-- This function was added in version 1.4.0. +wireDualPipeline :: forall a b r w r' w' proc. (Channel r w, Channel r' w') => + [(r a, w' b) -> (r' b, w a) -> proc] -> (r a, w' b) -> (r' b, w a) -> CHP [proc] +wireDualPipeline [] _ _ = return [] +wireDualPipeline procs@(first:rest) in_ out + = do chans <- replicateM (n - 1) newChannel + chans' <- replicateM (n - 1) newChannel + return $ (\(w, ps) -> first in_ w : ps) + $ (foldr wireF (out, []) $ zip3 rest chans chans') + where + n = length procs + + wireF :: ((r a, w' b) -> (r' b, w a) -> proc, Chan r w a, Chan r' w' b) + -> ((r' b, w a), [proc]) -> ((r' b, w a), [proc]) + wireF (p, c, d) (w, ps) = ((reader d, writer c), p (reader c, writer d) 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 @@ -83,6 +120,14 @@ pipeline :: [Chanin a -> Chanout a -> CHP b] -> Chanin a -> Chanout a -> CHP [b] pipeline procs in_ out = wirePipeline procs in_ out >>= runParallel +-- | Like pipeline, but works with processes that connect with a channel in both +-- directions. +-- +-- This function was added in version 1.4.0. +dualPipeline :: [(Chanin a, Chanout b) -> (Chanin b, Chanout a) -> CHP c] + -> (Chanin a, Chanout b) -> (Chanin b, Chanout a) -> CHP [c] +dualPipeline p i o = wireDualPipeline p i o >>= 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'. @@ -91,6 +136,15 @@ cycle :: [Chanin a -> Chanout a -> CHP b] -> CHP [b] cycle procs = wireCycle procs >>= runParallel +-- | Like cycle, but works with processes that connect with a channel in both +-- directions. +-- +-- This function was added in version 1.4.0. +dualCycle :: [(Chanin a, Chanout b) -> (Chanin b, Chanout a) -> CHP c] + -> CHP [c] +dualCycle p = wireDualCycle p >>= 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 @@ -102,6 +156,18 @@ (|->|) p q x y = do c <- oneToOneChannel runParallel_ [p x (writer c), q (reader c) y] +-- | Process composition that works with processes that connect with a channel in both +-- directions. Like (|->|), but connects a channel in each direction. +-- +-- This function was added in version 1.4.0. + +(|<->|) :: (a -> (Chanin b, Chanout c) -> CHP ()) + -> ((Chanin c, Chanout b) -> d -> CHP ()) + -> (a -> d -> CHP ()) +(|<->|) p q x y = do c <- oneToOneChannel + d <- oneToOneChannel + runParallel_ [p x (reader d, writer c), q (reader c, writer d) y] + -- | The reversed version of the other operator. -- -- The type for this process became more specific in version 1.2.0. @@ -110,7 +176,7 @@ (|<-|) = flip (|->|) -- | A function to use at the start of a pipeline you are chaining together with --- the '(|->|)' operator. +-- the '|->|' operator. -- Added in version 1.2.0. (->|) :: (Chanout b -> CHP ()) -> (Chanin b -> c -> CHP ()) -> (c -> CHP ()) @@ -118,9 +184,28 @@ runParallel_ [p (writer c), q (reader c) x] -- | A function to use at the end of a pipeline you are chaining together with --- the '(|->|)' operator. +-- the '|->|' operator. -- Added in version 1.2.0. (|->) :: (a -> Chanout b -> CHP ()) -> (Chanin b -> CHP ()) -> (a -> CHP ()) (|->) p q x = do c <- oneToOneChannel runParallel_ [p x (writer c), q (reader c)] + +-- | A function to use at the start of a pipeline you are chaining together with +-- the '|<->|' operator. +-- Added in version 1.4.0. +(|<->) :: (a -> (Chanin b, Chanout c) -> CHP ()) -> ((Chanin c, Chanout b) -> CHP ()) + -> (a -> CHP ()) +(|<->) p q x = do c <- oneToOneChannel + d <- oneToOneChannel + runParallel_ [p x (reader d, writer c), q (reader c, writer d)] + +-- | A function to use at the end of a pipeline you are chaining together with +-- the '|<->|' operator. +-- Added in version 1.4.0. +(<->|) :: ((Chanin b, Chanout c) -> CHP ()) -> ((Chanin c, Chanout b) -> a -> CHP ()) + -> (a -> CHP ()) +(<->|) p q x = do c <- oneToOneChannel + d <- oneToOneChannel + runParallel_ [p (reader d, writer c), q (reader c, writer d) x] + diff --git a/chp.cabal b/chp.cabal --- a/chp.cabal +++ b/chp.cabal @@ -1,24 +1,26 @@ Name: chp -Version: 1.3.2 +Version: 1.4.0 Synopsis: An implementation of concurrency ideas from Communicating Sequential Processes License: BSD3 License-file: LICENSE Author: Neil Brown Maintainer: neil@twistedsquare.com -Copyright: Copyright (c) 2008, University of Kent +Copyright: Copyright (c) 2008--2009, University of Kent Stability: Provisional Tested-with: GHC==6.8.2, GHC==6.10.1 Description: The Communicating Haskell Processes (CHP) library is an - implementation of ideas from Hoare's Communicating - Sequential Processes. More details and a tutorial can be - found at its homepage: - - The library requires at least GHC 6.8.1. + implementation of message-passing concurrency ideas from + Hoare's Communicating Sequential Processes. More details and + a tutorial can be found at its homepage: + , and there is + also now a blog with examples of using the library: + . The library requires at + least GHC 6.8.1. Homepage: http://www.cs.kent.ac.uk/projects/ofa/chp/ Category: Concurrency Build-Type: Simple -Build-Depends: base >= 3 && < 5, extensible-exceptions >= 0.1.1.0, containers, mtl, parallel, pretty, stm +Build-Depends: base >= 3 && < 5, extensible-exceptions >= 0.1.1.0, containers, HUnit, mtl, parallel, pretty, QuickCheck >= 2, stm Exposed-modules: Control.Concurrent.CHP Control.Concurrent.CHP.Actions @@ -34,6 +36,7 @@ Control.Concurrent.CHP.Enroll Control.Concurrent.CHP.Monad Control.Concurrent.CHP.Parallel + Control.Concurrent.CHP.Test Control.Concurrent.CHP.Traces Control.Concurrent.CHP.Traces.CSP Control.Concurrent.CHP.Traces.Structural @@ -52,6 +55,6 @@ Extensions: ScopedTypeVariables MultiParamTypeClasses FlexibleInstances - GeneralizedNewtypeDeriving CPP + GeneralizedNewtypeDeriving CPP BangPatterns -GHC-Options: -Wall -threaded +GHC-Options: -Wall -auto-all