chp 1.0.2 → 1.1.0
raw patch · 17 files changed
+1048/−234 lines, 17 files
Files
- Control/Concurrent/CHP/Alt.hs +218/−33
- Control/Concurrent/CHP/Arrow.hs +13/−4
- Control/Concurrent/CHP/Barriers.hs +10/−5
- Control/Concurrent/CHP/Base.hs +19/−8
- Control/Concurrent/CHP/BroadcastChannels.hs +6/−6
- Control/Concurrent/CHP/CSP.hs +16/−22
- Control/Concurrent/CHP/Channels.hs +53/−24
- Control/Concurrent/CHP/Common.hs +30/−10
- Control/Concurrent/CHP/Event.hs +612/−74
- Control/Concurrent/CHP/Guard.hs +3/−3
- Control/Concurrent/CHP/Monad.hs +8/−8
- Control/Concurrent/CHP/Parallel.hs +14/−6
- Control/Concurrent/CHP/Poison.hs +8/−1
- Control/Concurrent/CHP/ProcessId.hs +5/−2
- Control/Concurrent/CHP/Traces.hs +12/−2
- Control/Concurrent/CHP/Traces/Base.hs +19/−24
- chp.cabal +2/−2
Control/Concurrent/CHP/Alt.hs view
@@ -29,30 +29,39 @@ -- | A module containing the ALT constructs. An ALT (a term inherited from--- occam) is a choice between several events. In CHP, we say that an event+-- occam) is a choice between several alternate events. In CHP, we say that an event -- must support alting to be a valid choice. Events that /do/ support alting are: -- -- * 'Control.Concurrent.CHP.Monad.skip'+--+-- * 'Control.Concurrent.CHP.Monad.stop' -- -- * 'Control.Concurrent.CHP.Monad.waitFor' -- --- * Reading from a channel (including extended reads)+-- * Reading from a channel (including extended reads): that is, calls to 'Control.Concurrent.CHP.Channels.readChannel'+-- and 'Control.Concurrent.CHP.Channels.extReadChannel' -- --- * Writing to a channel+-- * Writing to a channel (including extended writes): that is, calls to 'Control.Concurrent.CHP.Channels.writeChannel'+-- and 'Control.Concurrent.CHP.Channels.extWriteChannel' -- --- * Synchronising on a barrier+-- * Synchronising on a barrier (using 'Control.Concurrent.CHP.Barriers.syncBarrier') -- --- * An alting construct (that is, you can nest alts)+-- * An alting construct (that is, you can nest alts) such as 'alt', 'priAlt' (or+-- the operator versions) -- --- * A sequential composition, if the first event supports alting+-- * A sequential composition, if the first event supports alting (i.e. is in this+-- list) --+-- * A call to 'every', which joins together several items (see the documentation+-- on 'every').+-- -- Examples of events that do /NOT/ support alting are: -- -- * Enrolling and resigning with a barrier -- -- * Poisoning a channel -- --- * Processes composed in parallel+-- * Processes composed in parallel (using 'runParallel', etc) -- -- * Any lifted IO event --@@ -84,30 +93,78 @@ -- * Check if a channel is ready; if so send, it on, otherwise return immediately: -- -- > (readChannel c >>= writeChannel d) </> skip-module Control.Concurrent.CHP.Alt (alt, (<->), priAlt, (</>)) where+--+-- Note that if you wait for a sequential composition:+--+-- > (readChannel c >>= writeChannel d) <-> (writeChannel e 6 >> readChannel f)+--+-- This only waits for the first action in both (reading from channel c, or writing+-- to channel e), not for all of the actions (as, for example, an STM transaction+-- would).+module Control.Concurrent.CHP.Alt (alt, (<->), priAlt, (</>), every, (<&>)) where import Control.Concurrent.STM import Control.Monad.State import Control.Monad.Trans import Data.List import Data.Maybe-import qualified Data.Set as Set import System.IO import Control.Concurrent.CHP.Base import Control.Concurrent.CHP.Event import Control.Concurrent.CHP.Guard+import Control.Concurrent.CHP.Parallel+import Control.Concurrent.CHP.Poison import Control.Concurrent.CHP.Traces.Base -- | An alt between several actions, with arbitrary priority. The first--- available action is chosen (with an arbitrary choice if many guards are+-- available action is chosen (with an arbitrary choice if many actions are -- available at the same time), its body run, and its value returned. alt :: [CHP a] -> CHP a alt = priAlt --- | An alt between several actions, with arbitrary priority. The first+-- | An alt between several actions, with descending priority. The first -- available action is chosen (biased towards actions nearest the beginning--- of the list), its body run, and its value returned. +-- of the list), its body run, and its value returned.+--+-- What priority means here is a difficult thing, and in some ways a historical+-- artifact. We can group the guards into three categories:+--+-- 1. synchronisation guards (reading from and writing to channels, and synchronising+-- on barriers)+--+-- 2. time-out guards (such as 'Control.Concurrent.CHP.Monad.waitFor')+--+-- 3. dummy guards ('Control.Concurrent.CHP.Monad.skip' and 'Control.Concurrent.CHP.Monad.stop')+--+-- There exists priority when comparing dummy guards to anything else. So for+-- example,+--+-- > priAlt [ skip, x ]+--+-- Will always select the first guard, whereas:+--+-- > priAlt [ x , skip ]+--+-- Is an effective way to poll and see if x is ready, otherwise the 'Control.Concurrent.CHP.Monad.skip' will+-- be chosen. However, there is no priority between synchronisation guards and+-- time-out guards. So the two lines:+--+-- > priAlt [ x, y ]+-- > priAlt [ y, x ]+--+-- May have the same or different behaviour (when x and y are not dummy guards),+-- there is no guarantee either way. The reason behind this is that if you ask+-- for:+--+-- > priAlt [ readChannel c, writeChannel d 6 ]+--+-- And the process at the other end is asking for:+--+-- > priAlt [ readChannel d, writeChannel c 8 ]+--+-- Whichever channel is chosen by both processes will not satisfy the priority+-- at one end (if such priority between channels was supported). priAlt :: [CHP a] -> CHP a priAlt items = (liftPoison $ priAlt' $ map wrapPoison items) >>= checkPoison @@ -128,7 +185,135 @@ infixl </> infixl <->+infixl <&> +-- | Runs all the given processes in parallel with each other, but only when the+-- choice at the beginning of each item is ready.+--+-- So for example, if you do:+--+-- > every [ readChannel c >>= writeChannel d, readChannel e >>= writeChannel f]+--+-- This will forward values from c and e to d and f respectively in parallel, but+-- only once both channels c and e are ready to be read from. So f cannot be written+-- to before c is read from (contrast this with what would happen if 'every' were+-- replaced with 'runParallel').+--+-- This behaviour can be somewhat useful, but 'every' is much more powerful when+-- used as part of an 'alt'. This code:+--+-- > alt [ every [ readChannel c, readChannel d]+-- > , every [ writeChannel e 6, writeChannel f 8] ]+--+-- Waits to either read from channels c and d, or to write to channels e and f.+--+-- The events involved can partially overlap, e.g.+--+-- > alt [ every [ readChannel a, readChannel b]+-- > , every [ readChannel a, writeChannel c 6] ]+-- +-- This will wait to either read from channels a and b, or to read from a and write+-- to c, whichever combination is ready first. If both are ready, the choice between+-- them will be arbitrary (just as with any other choices; see 'alt' for more details).+--+-- The sets can even be subsets of each other, such as:+--+-- > alt [ every [ readChannel a, readChannel b]+-- > , every [ readChannel a, readChannel b, readChannel b] ]+--+-- In this case there are no guarantees as to which choice will happen. Do not+-- assume it will be the smaller, and do not assume it will be the larger. +--+-- Be wary of what happens if a single event is included multiple times in the same 'every', as+-- this may not do what you expect (with or without choice). Consider:+-- +-- > every [ readChannel c >> writeChannel d 6+-- > , readChannel c >> writeChannel d 8 ]+--+-- What will happen is that the excecution will wait to read from c, but then it+-- will execute only one of the bodies (an arbitrary choice). In general, do not+-- rely on this behaviour, and instead try to avoid having the same events in an+-- 'every'. Also note that if you synchronise on a barrier twice in an 'every',+-- this will only count as one member enrolling, even if you have two enrolled+-- ends! For such a use, look at 'runParallel' instead.+--+-- Also note that this currently applies to both ends of channels, so that:+--+-- > every [ readChannel c, writeChannel c 2 ]+--+-- Will block indefinitely, rather than completing the communication.+--+-- Each item 'every' must support choice (and in fact+-- only a subset of the items supported by 'alt' are supported by 'every'). Currently the items+-- in the list passed to 'every' must be one of the following:+--+-- * A call to 'Control.Concurrent.CHP.Channels.readChannel' (or 'Control.Concurrent.CHP.Channels.extReadChannel').+-- +-- * A call to 'Control.Concurrent.CHP.Channels.writeChannel' (or 'Control.Concurrent.CHP.Channels.extWriteChannel').+--+-- * 'Control.Concurrent.CHP.Monad.skip', the always-ready guard.+--+-- * 'Control.Concurrent.CHP.Monad.stop', the never-ready guard (will cause the whole 'every' to never be ready,+-- since 'every' has to wait for all guards).+--+-- * A call to 'Control.Concurrent.CHP.Monad.syncBarrier'.+--+-- * A sequential composition where the first item is one of the things in this+-- list.+--+-- * A call to 'every' (you can nest 'every' blocks inside each other).+--+-- Timeouts (e.g. 'Control.Concurrent.CHP.Monad.waitFor') are currently not supported. You can always get another+-- process to synchronise on an event with you once a certain time has passed.+--+-- Note also that you cannot put an 'alt' inside an 'every'. So you cannot say:+--+-- > every [ readChannel c+-- > , alt [ readChannel d, readChannel e ] ]+--+-- To wait for c and (d or e) like this you must expand it out into (c and d) or+-- (c and e):+--+-- > alt [ every [ readChannel c, readChannel d]+-- > , every [ readChannel c, readChannel e] ]+--+-- As long as x meets the conditions laid out above, 'every' [x] will have the same+-- behaviour as x.+--+-- Added in version 1.1.0+every :: [CHP a] -> CHP [a]+every [] = liftPoison $ AltableT (SkipGuard [], return []) (return [])+every xs = liftPoison (AltableT (foldl1 merge $ map blankEvent gs, getEventPoison True) (return+ $ NoPoison False)) >>= checkPoison >>= \b -> if b then runParallel (map (unwrapPoison . liftTrace) bodies) else alt [every xs]+ where+ (gs, bodies) = unzip $ map (pullOutAltable . wrapPoison) xs++ blankEvent :: Guard -> Guard+ blankEvent (EventGuard _ rec act es) = EventGuard [] rec act es+ blankEvent g = g++ merge :: Guard -> Guard -> Guard+ merge (SkipGuard _) g = g+ merge g (SkipGuard _) = g+ merge StopGuard _ = StopGuard+ merge _ StopGuard = StopGuard+ merge (EventGuard _ recx actx esx) (EventGuard _ recy acty esy)+ = EventGuard [] (recx ++ recy) (actx >> acty) (esx ++ esy)+ merge _ _ = BadGuard "merging unsupported guards"++-- | A useful operator that acts like 'every'. The operator is associative and+-- commutative (see 'every' for notes on idempotence). When you have lots of things+-- to join with this operator, it's probably easier to use the 'every' function.+--+-- Added in version 1.1.0+(<&>) :: forall a b. CHP a -> CHP b -> CHP (a, b)+(<&>) a b = every [a >>= return . Left, b >>= return . Right] >>= return . merge+ where+ merge :: [Either a b] -> (a, b)+ merge [Left x, Right y] = (x, y)+ merge [Right y, Left x] = (x, y)+ merge _ = error "Invalid merge possibility in <&>"+ -- ALTing is implemented as follows in CHP. The CHP monad has [Int] in its -- state. When you choose between N events, you form one body, that pulls -- the next number from the head of the state and executes the body for the@@ -171,7 +356,8 @@ do g' <- g return $ do ns <- g' return (n : ns)- wrap (_, _) = BadGuard+ wrap (_, g@(BadGuard _)) = g+ wrap (_, _) = BadGuard "wrapped" -- Polls the available guards, but ignores timeout guards and alting barrier -- guards@@ -180,7 +366,7 @@ (map checkGuard flattenedGuards) ++ [return Nothing] where checkGuard :: (Int, Guard) -> STM (Maybe Int)- checkGuard (n, BadGuard) = return $ Just n+ checkGuard (n, BadGuard _) = return $ Just n checkGuard (n, SkipGuard {}) = return $ Just n checkGuard (_, _) = retry @@ -192,7 +378,7 @@ atomically $ foldl1 orElse (wrap True extra : map (wrap False) guards) where enable :: Guard -> IO (STM [Int])- enable (BadGuard) = return $ return []+ enable (BadGuard _) = return $ return [] enable (SkipGuard ns) = return $ return ns enable (TimeoutGuard g) = g enable _ = return retry -- This effectively ignores other guards@@ -214,11 +400,11 @@ g -> (n, g) : flatten xs -- The alting barrier guards:- eventGuards :: [(RecEvents, [Int], STM (), Event)]+ eventGuards :: [([RecordedIndivEvent], [Int], STM (), [Event])] eventGuards = [(rec,ns,act,ab) | EventGuard ns rec act ab <- wrappedGuards] -- We must use isPrefixOf, because things are added in the case of poison- findEventAssoc :: [Int] -> RecEvents+ findEventAssoc :: [Int] -> [RecordedIndivEvent] findEventAssoc x = case filter (\(_,y,_,_) -> y `isPrefixOf` x) eventGuards of [(rec,_,_,_)] -> rec _ -> error "Could not find associated event in alt, internal logic error" @@ -231,7 +417,7 @@ storeChoice ns = modify (\(_, es) -> (ns, es)) isBadGuard :: Guard -> Bool- isBadGuard BadGuard = True+ isBadGuard (BadGuard _) = True isBadGuard _ = False -- Performs the select operation on all the guards. The choice is stored@@ -242,8 +428,10 @@ = 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" + = let str = head [s | BadGuard s <- wrappedGuards]+ err = "ALTing not supported on given guard: " ++ str+ in liftIO $ do hPutStrLn stderr err+ ioError $ userError err | otherwise = do earliestReady <- liftIO $ atomically checkNormalGuards tv <- liftIO . atomically $ newTVar Nothing@@ -251,19 +439,16 @@ (_, tr) <- get mn <- liftIO . atomically $ do ret <- enableEvents tv pid- (maybe id take earliestReady eventGuards)+ (maybe id take earliestReady [(x,y,z) | (_,x,y,z)<-eventGuards]) (isNothing earliestReady)- case ret of- Just ((e,_), pids, _) ->- do recordEventLast e (Set.fromList pids) tr- return ret- Nothing -> return ret+ maybe (return ()) (\(_,es) -> recordEventLast (nub es) tr) ret+ return ret case (mn, earliestReady) of -- An event -- and we were the last person to arrive: -- The event must have been higher priority than any other -- ready guards- (Just (rec, _, ns), _) ->- do recordEvent (snd rec)+ (Just (ns, _), _) ->+ do recordEvent $ findEventAssoc ns storeChoice ns -- No events were ready, but there was an available normal -- guards. Re-run the normal guards; at least one will be ready@@ -275,15 +460,15 @@ (Nothing, Nothing) -> do (wasAltingBarrier, ns) <- liftIO $ waitNormalGuards $ waitAlting tv if wasAltingBarrier- then recordEvent (snd $ findEventAssoc ns) >> storeChoice ns -- It was a barrier, all done+ then recordEvent (findEventAssoc ns) >> storeChoice ns -- It was a barrier, all done else -- Another guard fired, but we must check in case -- we have meanwhile been committed to taking an -- event:- do mn' <- liftIO . atomically $ disableEvents tv (map fourth eventGuards)+ do mn' <- liftIO . atomically $ disableEvents tv (concatMap fourth eventGuards) case mn' of -- An event overrides our non-event choice:- Just bns -> recordEvent (snd $ findEventAssoc bns) >> storeChoice bns+ Just bns -> recordEvent (findEventAssoc bns) >> storeChoice bns -- Go with the original option, no events -- became ready: Nothing -> storeChoice ns@@ -303,6 +488,6 @@ do put (gs, es) snd $ pullOutAltable (items !! g) ([], _) -> liftIO $- do hPutStrLn stderr "ALTing not supported on given guard"- ioError $ userError "ALTing not supported on given guard"+ do hPutStrLn stderr "ALTing not supported on given guard (no index)"+ ioError $ userError "ALTing not supported on given guard (no index)"
Control/Concurrent/CHP/Arrow.hs view
@@ -59,11 +59,8 @@ -- 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+module Control.Concurrent.CHP.Arrow (ProcessPipeline, runPipeline, arrowProcess) where import Control.Arrow import Control.Monad@@ -92,6 +89,18 @@ -- is probably easier for doing that than declaring all the channels yourself -- and composing everything in parallel. }++-- | Adds a wrapper that forms this process into the right data type to be+-- part of an arrow.+--+-- Any process you apply this to should produce exactly one output per+-- input, or else you will find odd behaviour resulting (including deadlock).+-- So for example, /don't/ use @arrowProcess ('Control.Concurrent.CHP.Common.filter'+-- ...)@ or @arrowProcess 'Control.Concurrent.CHP.Common.stream'@+--+-- Added in version 1.1.0+arrowProcess :: (Chanin a -> Chanout b -> CHP ()) -> ProcessPipeline a b+arrowProcess = ProcessPipeline instance Arrow ProcessPipeline where arr = ProcessPipeline . CHP.map
Control/Concurrent/CHP/Barriers.hs view
@@ -63,7 +63,7 @@ -- Everyone is told the new phase once they complete a synchronisation, and -- may query the current phase for any barrier that they are currently enrolled -- on.-module Control.Concurrent.CHP.Barriers (Barrier, newBarrier, newBarrierWithLabel,+module Control.Concurrent.CHP.Barriers (Barrier, EnrolledBarrier, newBarrier, newBarrierWithLabel, PhasedBarrier, newPhasedBarrier, newPhasedBarrierWithLabel, currentPhase, waitForPhase, syncBarrier, getBarrierIdentifier) where @@ -82,10 +82,15 @@ -- standard barrier. type Barrier = PhasedBarrier () +-- | A useful type synonym for enrolled barriers with no phases+--+-- Added in 1.1.0+type EnrolledBarrier = Enrolled PhasedBarrier ()+ -- | Synchronises on the given barrier. You must be enrolled on a barrier in order -- to synchronise on it. Returns the new phase, following the synchronisation. syncBarrier :: (Enum phase, Bounded phase, Eq phase) => Enrolled PhasedBarrier phase -> CHP phase-syncBarrier = syncBarrierWith (recAs (Just . BarrierSync) (Just . BarrierSyncIndiv))+syncBarrier = syncBarrierWith (Just . BarrierSyncIndiv) -- | Finds out the current phase a barrier is on. currentPhase :: (Enum phase, Bounded phase, Eq phase) => Enrolled PhasedBarrier phase -> CHP phase@@ -112,7 +117,7 @@ -- cycle, so that the first synchronisation moves it on to the first newPhasedBarrier :: (Enum phase, Bounded phase, Eq phase) => phase -> CHP (PhasedBarrier phase) newPhasedBarrier ph = liftPoison $ liftTrace $ do- e <- liftIO $ newEvent 0+ e <- liftIO $ newEvent BarrierSync 0 tv <- liftIO $ atomically $ newTVar ph return $ Barrier (e, tv) @@ -126,7 +131,7 @@ -- using the given label newPhasedBarrierWithLabel :: (Enum phase, Bounded phase, Eq phase) => String -> phase -> CHP (PhasedBarrier phase) newPhasedBarrierWithLabel l ph = liftPoison $ liftTrace $ do- e <- liftIO $ newEvent 0+ e <- liftIO $ newEvent BarrierSync 0 labelEvent e l tv <- liftIO $ atomically $ newTVar ph return $ Barrier (e, tv)@@ -135,4 +140,4 @@ -- | Gets the identifier of a Barrier. Useful if you want to identify it in -- the trace later on. getBarrierIdentifier :: (Enum ph, Bounded ph, Eq ph) => PhasedBarrier ph -> Unique-getBarrierIdentifier (Barrier (Event (u,_),_)) = u+getBarrierIdentifier (Barrier (e,_)) = getEventUnique e
Control/Concurrent/CHP/Base.hs view
@@ -61,8 +61,10 @@ -- based on a single enroll call, undefined behaviour will result. newtype Enrolled b a = Enrolled (b a) --- | The central monad of the library. You can use 'runCHP' and 'runCHP_'--- to execute programs in this monad.+-- | The central monad of the library. You can use+-- 'Control.Concurrent.CHP.Monad.runCHP' and+-- 'Control.Concurrent.CHP.Monad.runCHP_' to execute programs in this+-- monad. newtype CHP a = PoisonT (ErrorT PoisonError CHP' a) deriving (Monad, MonadIO) @@ -76,8 +78,8 @@ -- Classes: -- ======== --- | A monad transformer class that is very similar to MonadIO. This can be--- useful if you want to add monad transformers (such as StateT, ReaderT) on+-- | A monad transformer class that is very similar to 'MonadIO'. This can be+-- useful if you want to add monad transformers (such as 'StateT', 'ReaderT') on -- top of the 'CHP' monad. class MonadIO m => MonadCHP m where liftCHP :: CHP a -> m a@@ -118,11 +120,11 @@ pullOutAltable :: CHP' a -> (Guard, TraceT IO a) pullOutAltable m = case m of- AltableTRet x -> (badGuard, return x)+ AltableTRet x -> (badGuard "return", return x) AltableT alt _ -> alt liftTrace :: TraceT IO a -> CHP' a-liftTrace m = AltableT (badGuard, m) m+liftTrace m = AltableT (badGuard "lifted action", m) m wrapPoison :: CHP a -> CHP' (WithPoison a) wrapPoison (PoisonT m) = (liftM $ either (const PoisonItem) NoPoison) $@@ -146,11 +148,11 @@ -- | Allows you to provide a handler for sections with poison. It is usually -- used in an infix form as follows: ----- > (readChannel c >>= writeChannel d) `onPoison` (poison c >> poison d)+-- > (readChannel c >>= writeChannel d) `onPoisonTrap` (poison c >> poison d) -- -- It handles the poison and does not rethrow it (unless your handler -- does so). If you want to rethrow (and actually, you'll find you usually--- do), use onPoisonRethrow+-- do), use 'onPoisonRethrow' onPoisonTrap :: CHP a -> CHP a -> CHP a onPoisonTrap (PoisonT body) (PoisonT handler) = PoisonT $ body `catchError` (const handler) @@ -197,6 +199,15 @@ l <- atomically $ readTVar tv t' <- f l t return (x, t')++getEventPoison :: a -> TraceT IO (WithPoison a)+getEventPoison x+ = do (ns,y) <- get+ case ns of+ [] -> return $ NoPoison x+ _ -> do put ([],y)+ return PoisonItem+ -- ========== -- Instances:
Control/Concurrent/CHP/BroadcastChannels.hs view
@@ -95,24 +95,24 @@ instance WriteableChannel BroadcastChanout where extWriteChannel (BO (BC (b, tv))) m- = do syncBarrierWith (recAs (Just . ChannelComm) (Just . ChannelWrite))+ = do syncBarrierWith (Just . ChannelWrite) $ Enrolled b m >>= liftIO . atomically . writeTVar tv- syncBarrierWith (recAs (const Nothing) (const Nothing))+ syncBarrierWith (const Nothing) $ Enrolled b- syncBarrierWith (recAs (const Nothing) (const Nothing))+ syncBarrierWith (const Nothing) $ Enrolled b return () instance ReadableChannel (Enrolled BroadcastChanin) where extReadChannel (Enrolled (BI (BC (b, tv)))) f- = do syncBarrierWith (recAs (Just . ChannelComm) (Just . ChannelRead))+ = do syncBarrierWith (Just . ChannelRead) $ Enrolled b- syncBarrierWith (recAs (const Nothing) (const Nothing))+ syncBarrierWith (const Nothing) $ Enrolled b x <- liftIO (atomically $ readTVar tv) y <- f x- syncBarrierWith (recAs (const Nothing) (const Nothing))+ syncBarrierWith (const Nothing) $ Enrolled b return y
Control/Concurrent/CHP/CSP.hs view
@@ -37,6 +37,8 @@ import Control.Monad.State import Control.Monad.Writer import Control.Monad.Trans+import Data.List+import Data.Maybe import Data.Unique import System.IO @@ -49,13 +51,9 @@ import Control.Concurrent.CHP.Poison import Control.Concurrent.CHP.Traces.Base -recAs :: (Unique -> Maybe RecordedEvent) -> (Unique -> Maybe RecordedIndivEvent) -> Unique- -> RecEvents-recAs f g u = (f u, g u)- -- First engages in event, then executes the body. The returned value is suitable -- for use in an alt-buildOnEventPoison :: (Unique -> RecEvents) -> Event.Event -> STM () -> CHP a -> CHP a+buildOnEventPoison :: (Unique -> Maybe RecordedIndivEvent) -> Event.Event -> STM () -> CHP a -> CHP a buildOnEventPoison rec e act body = liftPoison (AltableT (theGuard, getEventPoison True) (return $ NoPoison False))@@ -63,15 +61,7 @@ alt [liftPoison $ AltableT (theGuard, getEventPoison ()) (return $ NoPoison())] >>= checkPoison >> body where- theGuard = let Event.Event (u,_) = e in EventGuard [] (rec u) act e-- getEventPoison :: a -> TraceT IO (WithPoison a)- getEventPoison x- = do (ns,y) <- get- case ns of- [] -> return $ NoPoison x- _ -> do put ([],y)- return PoisonItem+ theGuard = EventGuard [] (maybeToList $ rec $ Event.getEventUnique e) act [e] scopeBlock :: CHP a -> (a -> CHP b) -> IO () -> CHP b scopeBlock start body errorEnd@@ -85,8 +75,8 @@ -- | Synchronises on the given barrier. You must be enrolled on a barrier in order -- to synchronise on it. Returns the new phase, following the synchronisation.-syncBarrierWith :: (Enum phase, Bounded phase, Eq phase) => (Unique -> RecEvents) ->- Enrolled PhasedBarrier phase -> CHP phase+syncBarrierWith :: (Enum phase, Bounded phase, Eq phase) => (Unique -> Maybe+ RecordedIndivEvent) -> Enrolled PhasedBarrier phase -> CHP phase syncBarrierWith rec (Enrolled (Barrier (e,tv))) = buildOnEventPoison rec e incPhase (liftIO $ atomically $ readTVar tv)@@ -97,7 +87,7 @@ -- | A phased barrier that is capable of being poisoned and throwing poison. -- You will need to enroll on it to do anything useful with it.--- For the phases you can use any type that satisfies Enum, Bounded and Eq.+-- For the phases you can use any type that satisfies 'Enum', 'Bounded' and 'Eq'. -- The phase increments every time the barrier completes. Incrementing consists -- of: @if p == maxBound then minBound else succ p@. Examples of things that -- make sense for phases:@@ -109,15 +99,15 @@ -- But don't forget that the count will wrap round when it reaches the end. -- You cannot use 'Integer' for a phase because it is unbounded. If you really -- want to have an infinitely increasing count, you can wrap 'Integer' in a newtype and--- provide a Bounded instance for it (with minBound and maxBound set to -1,+-- provide a 'Bounded' instance for it (with minBound and maxBound set to -1, -- if you start on 0). -- -- * A boolean. This implements a simple black-white barrier, where the state -- flips on each iteration. -- -- * A custom data type that has only constructors. For example, @data MyPhases--- = Discover | Plan | Move@. Haskell supports deriving Enum, Bounded and--- Eq automatically on such types.+-- = Discover | Plan | Move@. Haskell supports deriving 'Enum', 'Bounded' and+-- 'Eq' automatically on such types. newtype (Enum phase, Bounded phase, Eq phase) => PhasedBarrier phase = Barrier (Event.Event, TVar phase) @@ -126,11 +116,15 @@ enroll b@(Barrier (e,_)) f = do liftSTM (Event.enrollEvent e) >>= checkPoison x <- f $ Enrolled b- liftSTM (Event.resignEvent e) >>= checkPoison+ liftSTM (Event.resignEvent e) >>= checkPoison >>= (\es ->+ do (_,tr) <- liftPoison $ liftTrace get+ when (not $ null es) $ liftSTM $ recordEventLast (nub es) tr) return x resign (Enrolled (Barrier (e,_))) m- = do liftSTM (Event.resignEvent e) >>= checkPoison+ = do liftSTM (Event.resignEvent e) >>= checkPoison >>= (\es ->+ do (_,tr) <- liftPoison $ liftTrace get+ when (not $ null es) $ liftSTM $ recordEventLast (nub es) tr) x <- m liftSTM (Event.enrollEvent e) >>= checkPoison return x
Control/Concurrent/CHP/Channels.hs view
@@ -108,23 +108,35 @@ -- ======== class ChaninC c a where+ -- Start gets the event and the transaction that will wait for data. You+ -- sync on the event (possible extended write occurs) then wait for data startReadChannelC :: c a -> (Event, STM (WithPoison a))- endReadChannelC :: c a -> STM ()+ -- (extended read action goes here)+ -- Read releases the writer+ endReadChannelC :: c a -> STM (WithPoison ()) poisonReadC :: c a -> IO () checkPoisonReadC :: c a -> IO (WithPoison ()) class ChanoutC c a where+ -- Start checks for poison and gets the event: startWriteChannelC :: c a -> (Event, STM (WithPoison ()))- endWriteChannelC :: c a -> a -> STM ()+ -- (extended write action goes here)+ -- Send actually transmits the value:+ sendWriteChannelC :: c a -> a -> STM (WithPoison ())+ -- (extended read action goes here)+ -- 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 ())+ 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 -- its pair of end types. You can see the types in the list of instances below.--- Thus, newChannel may be used, and the compiler will infer which type of--- channel is required based on what end-types you get from reader and writer.--- Alternatively, if you explicitly type the return of newChannel, it will+-- Thus, 'newChannel' may be used, and the compiler will infer which type of+-- channel is required based on what end-types you get from 'reader' and 'writer'.+-- Alternatively, if you explicitly type the return of 'newChannel', it will -- be definite which ends you will use. If you do want to fix the type of -- the channel you are using when you allocate it, consider using one of the -- many 'oneToOneChannel'-like shorthand functions that fix the type.@@ -167,12 +179,15 @@ -- 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.+-- | A helper function that uses the parallel strategies library (see the+-- paper: \"Algorithm + Strategy = Parallelism\", P.W. Trinder et al, JFP+-- 8(1) 1998,+-- <http://www.macs.hw.ac.uk/~dsg/gph/papers/html/Strategies/strategies.html>)+-- 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+-- 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.@@ -193,7 +208,14 @@ 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 () + -- | Like 'newChannel' but also associates a label with that channel in a -- trace. You can use this function whether tracing is turned on or not, -- so if you ever use tracing, you should use this rather than 'newChannel'.@@ -246,9 +268,9 @@ stmChannel :: MonadIO m => m (Unique, STMChannel a) stmChannel = liftIO $- do e@(Event (u,_)) <- newEvent 2+ do e <- newEvent ChannelComm 2 c <- atomically $ newTVar $ NoPoison Nothing- return (u, STMChan (e,c))+ return (getEventUnique e, STMChan (e,c)) oneToOneChannel :: MonadCHP m => m (OneToOneChannel a) oneToOneChannel = newChannel@@ -282,7 +304,7 @@ instance ReadableChannel Chanin where readChannel (Chanin c) = let (e, m) = startReadChannelC c in- buildOnEventPoison (recAs (Just . ChannelComm) (Just . ChannelRead)) e (return ()) (liftSTM $+ buildOnEventPoison (Just . ChannelRead) e (return ()) (liftSTM $ do x <- m endReadChannelC c return x) >>= checkPoison@@ -290,7 +312,7 @@ extReadChannel (Chanin c) body = let (e, m) = startReadChannelC c in scopeBlock- (buildOnEventPoison (recAs (Just . ChannelComm) (Just . ChannelRead)) e (return ()) (liftSTM m) >>= checkPoison)+ (buildOnEventPoison (Just . ChannelRead) e (return ()) (liftSTM m) >>= checkPoison) (\val -> do x <- body val liftSTM $ endReadChannelC c return x)@@ -299,17 +321,19 @@ instance WriteableChannel Chanout where writeChannel (Chanout c) x = let (e, m) = startWriteChannelC c in- buildOnEventPoison (recAs (Just . ChannelComm) (Just . ChannelWrite)) e (return ())- (liftSTM $ do y <- m- endWriteChannelC c x- return y)- >>= checkPoison+ buildOnEventPoison (Just . ChannelWrite) e (return ())+ (liftM2 (++)+ (liftSTM $ sequence [m, sendWriteChannelC c x])+ (liftSTM $ sequence [endWriteChannelC c]))+ >>= checkPoison . mergeWithPoison extWriteChannel (Chanout c) body = let (e, m) = startWriteChannelC c in scopeBlock- (buildOnEventPoison (recAs (Just . ChannelComm) (Just . ChannelWrite))+ (buildOnEventPoison (Just . ChannelWrite) e (return ()) (liftSTM m) >>= checkPoison)- (const $ body >>= liftSTM . endWriteChannelC c)+ (const $ sequence [body >>= liftSTM . sendWriteChannelC c+ ,liftSTM (endWriteChannelC c)]+ >>= checkPoison . mergeWithPoison) (poisonWriteC c) @@ -367,8 +391,9 @@ endReadChannelC (STMChan (_,tv)) = do x <- readTVar tv case x of- PoisonItem -> return ()- NoPoison _ -> writeTVar tv $ NoPoison Nothing+ PoisonItem -> return PoisonItem+ NoPoison _ -> do writeTVar tv $ NoPoison Nothing+ return $ NoPoison () poisonReadC (STMChan (e,tv)) = liftSTM $ do poisonEvent e writeTVar tv PoisonItem@@ -380,11 +405,15 @@ case x of PoisonItem -> return PoisonItem NoPoison _ -> return $ NoPoison ())- endWriteChannelC (STMChan (_, tv)) val+ sendWriteChannelC (STMChan (_, tv)) val = do x <- readTVar tv case x of- PoisonItem -> return ()+ PoisonItem -> return PoisonItem NoPoison _ -> do writeTVar tv $ NoPoison $ Just val+ return $ NoPoison ()+ endWriteChannelC (STMChan (_, tv))+ = waitForNothingOrPoison tv+ poisonWriteC (STMChan (e,tv)) = liftSTM $ do poisonEvent e
Control/Concurrent/CHP/Common.hs view
@@ -47,6 +47,7 @@ module Control.Concurrent.CHP.Common where import Control.Monad+import Control.Parallel.Strategies import qualified Data.Traversable as Traversable import Prelude (Bool, Maybe(..), Enum, Ord, ($), (<), Int, otherwise, (.)) import qualified Prelude@@ -61,12 +62,21 @@ writeChannel out x ) `onPoisonRethrow` (poison in_ >> poison out) --- | Forever forwards the value onwards, in an extended rendezvous. This is--- like 'id' but does not add any buffering to your network.+-- | Forever forwards the value onwards. This is+-- like 'id' but does not add any buffering to your network, and its presence+-- is indetectable to the process either side. ----- extId is a unit of the associative operator '|->|'.+-- extId is a unit of the associative operator 'Control.Concurrent.CHP.Utils.|->|'.+--+-- The behaviour of this process was corrected in version 1.1.0 to work properly+-- when the reader of its output channel was offering choice. extId :: Chanin a -> Chanout a -> CHP ()-extId in_ out = tap in_ [out]+extId in_ out = do+ c <- oneToOneChannel+ forever $+ extReadChannel in_ (writeChannel (writer c))+ <&>+ extWriteChannel out (readChannel (reader c)) -- | A process that waits for an input, then sends it out on /all/ its output -- channels (in order) during an extended rendezvous. This is often used to send the@@ -76,7 +86,7 @@ -- value will not be sent to the other recipients until it does. The name -- of the process derives from the notion of a wire-tap, since the listener -- is hidden from the other processes (it does not visibly change the semantics--- for them).+-- for them -- except when the readers of the channels are offering a choice). tap :: Chanin a -> [Chanout a] -> CHP () tap in_ outs = (forever $ extReadChannel in_@@ -102,13 +112,22 @@ -- | Forever reads in a value, transforms it using the given function, and sends it -- out again. Note that the transformation is not applied strictly, so don't--- assume that this process will actually perform the computation.+-- assume that this process will actually perform the computation. If you+-- require a strict transformation, use 'map''. map :: (a -> b) -> Chanin a -> Chanout b -> CHP () map f in_ out = forever (readChannel in_ >>= (return . f) >>= writeChannel out) `onPoisonRethrow` (poison in_ >> poison out) +-- | Like 'map', but applies the transformation strictly before sending on+-- the value.+--+-- Added in version 1.1.0.+map' :: NFData b => (a -> b) -> Chanin a -> Chanout b -> CHP ()+map' f in_ out = forever (readChannel in_ >>= (return . f) >>= writeChannelStrict out)+ `onPoisonRethrow` (poison in_ >> poison out)+ -- | Forever reads in a value, and then based on applying the given function--- either discards it (if the function returns false) or sends it on (if+-- either discards it (if the function returns False) or sends it on (if -- the function returns True). filter :: (a -> Bool) -> Chanin a -> Chanout a -> CHP () filter f in_ out = forever (do@@ -116,9 +135,10 @@ when (f x) (writeChannel out x) ) `onPoisonRethrow` (poison in_ >> poison out) --- | Streams all items in a Traversable container out in the order given by--- 'Traversable.mapM' on the output channel (one at a time). Lists, Maybe,--- and Set are all instances of Traversable, so this can be used for all of+-- | Streams all items in a 'Data.Traversable.Traversable' container out+-- in the order given by 'Data.Traversable.mapM' on the output channel (one at+-- a time). Lists, 'Prelude.Maybe', and 'Data.Set.Set' are all instances+-- of 'Data.Traversable.Traversable', so this can be used for all of -- those. stream :: Traversable.Traversable t => Chanin (t a) -> Chanout a -> CHP () stream in_ out = (forever $ do
Control/Concurrent/CHP/Event.hs view
@@ -32,136 +32,674 @@ import Control.Concurrent.STM import Control.Monad+import Data.Function import Data.List+import qualified Data.Map as Map import Data.Maybe+import qualified Data.Set as Set import Data.Unique + import Control.Concurrent.CHP.Poison import Control.Concurrent.CHP.ProcessId +data RecordedEventType = ChannelComm | BarrierSync deriving (Eq, Ord, Show)+ -- Not really a CSP event, more like an enrollable poisonable alting barrier! newtype Event = Event ( Unique, -- Event identifier+ RecordedEventType, -- Event type for trace recording TVar (WithPoison (Int, -- Enrolled count [OfferSet]) -- A list of offer sets )) -type OfferSet = (TVar (Maybe [Int]) -- Variable to use to signal when committed- , [Int] -- Value to send when committed+instance Eq Event where+ (==) = (==) `on` getEventUnique++instance Ord Event where+ compare = compare `on` getEventUnique++-- For testing:+instance Show Event where+ show (Event (u, t, _tv)) = "Event " ++ show (hashUnique u,t)++getEventUnique :: Event -> Unique+getEventUnique (Event (u,_,_)) = u++getEventTVar :: Event -> TVar (WithPoison (Int, [OfferSet]))+getEventTVar (Event (_,_,tv)) = tv++getEventType :: Event -> RecordedEventType+getEventType (Event (_,t,_)) = t++newtype OfferSet = OfferSet (TVar (Maybe [Int]) -- Variable to use to signal when committed , ProcessId -- Id of the process making the offer- , [Event]) -- A list of all events currently offered+ , [([Int], Map.Map Event ())]) -- Value to send when committed+ -- A list of all sets of events currently offered -newEvent :: Int -> IO Event-newEvent n+instance Eq OfferSet where+ (==) = (==) `on` (\(OfferSet (tv,_,_)) -> tv)++ -- 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+ -- it too+ -- NoPoison False; event has been rejected by previous process, you cannot+ -- choose it++unionAll :: Ord k => [Map.Map k a] -> Map.Map k a+unionAll = foldl Map.union Map.empty++allEventsInOffer :: OfferSet -> Map.Map Event ()+allEventsInOffer (OfferSet (_, _, eventSets)) = unionAll (map snd eventSets)++search :: [OfferSet] -> Map.Map Event Bool -> Maybe (STM (), Map.Map Unique (RecordedEventType,+ Set.Set ProcessId), [(TVar (Maybe [Int]), [Event])])+search [] _ = Just (return (), Map.empty, [])+search (offer@(OfferSet (tv, pid, eventSets)) : offers) eventMap+ | Map.null mustChooseFromEventSets = tryAll eventSets+ | otherwise = tryAll filteredEventSets+ where+ allEventsInOfferMappedToFalse :: Map.Map Event Bool+ allEventsInOfferMappedToFalse = Map.map (const False) (allEventsInOffer offer)++ mustChooseFromEventSets :: Map.Map Event Bool+ mustChooseFromEventSets+ = (Map.filter id {- Keep all True events -} eventMap)+ `Map.intersection` allEventsInOfferMappedToFalse++ -- Only the offers containing all of the mustChooseFromEventSets+ filteredEventSets+ = [ off+ | off@(_,es) <- eventSets,+ Map.isSubmapOfBy (\_ _ -> True)+ mustChooseFromEventSets+ es+ ]++ mapdotall :: Ord k => (a -> Bool) -> Map.Map k a -> Bool+ mapdotall f = Map.fold (\x b -> f x && b) True++ -- All events in the maps in the first parameter will be mapped to True+ tryAll :: [([Int],Map.Map Event ())] ->+ Maybe (STM (), Map.Map Unique (RecordedEventType, Set.Set ProcessId),+ [(TVar (Maybe [Int]), [Event])])+ tryAll [] = Nothing+ tryAll ((ns, es):next)+ | not $ mapdotall id (eventMap `Map.intersection` es)+ -- Contains an already-rejected event (one that mapped to False), skip:+ -- Need to reject the other events too though -- well, at least put+ -- them in the appropriate map and pass them through. They will+ -- only be rejected if they are then not contained in the other chosen+ -- offer+ = tryAll next+ | otherwise = case search offers eventMap' of+ Nothing -> tryAll next+ Just (act, resolved, retract) -> Just (if null ns then act else writeTVar tv (Just ns) >> act, foldl+ (\m e -> Map.insertWith add (getEventUnique e) (getEventType+ e, Set.singleton pid) m) resolved (Map.keys es), if null ns then retract else+ (tv, Map.keys allEventsInOfferMappedToFalse) : retract)+ + where+ -- All events that features in other offers by this process, but not+ -- the current offer+ --+ -- It is very important here that union is left-biased for both unions. We don't want+ -- to overwrite poison with acceptance, or acceptance with rejection.+ eventMap'+ = (eventMap `Map.union` (Map.map (const True) es)) `Map.union` allEventsInOfferMappedToFalse+++ add (tx, pidsx) (_, pidsy) = (tx, pidsx `Set.union` pidsy)+ +data EventStatus = Fine | NotCompletable deriving (Eq, Show)++-- Given a list of offers that could possibly complete, check if any set+-- of offers can. If so, complete it (including all retractions and+-- notifications for each process), otherwise leave things untouched.+--+-- Takes an optional tvar identifier for the newest process to make an offer, the+-- list of all offer-sets that need to be considered (they will have come from+-- all events in a connected sub-graph), the map of relevant events to their status,+-- and returns the map of event-identifiers that did complete.+resolveOffers :: Maybe (TVar (Maybe [Int])) -> [OfferSet] -> Set.Set Event -> STM (Map.Map Unique (RecordedEventType,+ Set.Set ProcessId))+resolveOffers newTvid allOffers events+ = do let (offers', _) = trim (allOffers, events)+ (act, ret, retract) = fromMaybe (return (), Map.empty, []) $ search (map addNullOffer+ $ sortOffers offers') Map.empty+ act+ -- do the retractions for all involved processes once the choice is made:+ -- TODO optimise:+ retractOffers $ zip (map fst retract)+ (repeat $ unionAll $ map allEventsInOffer allOffers)+ return ret+ where+ -- 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 :: [([Int], Map.Map Event ())]+ nullOffer = [([],Map.empty)]++ -- SMallest offers first to minimise backtracking:+ sortOffers :: [OfferSet] -> [OfferSet]+ sortOffers xs = sortBy (compare `on` (\(OfferSet (_,_,es)) -> length es)) xs+ -- TODO put the newest process first again++-- Given a list of offer-sets, and a map of events already-looked-at to their status,+-- trims the offer-sets by removing any option in an offer-set that cannot possibly+-- complete. If this option includes any other events, any other options anywhere+-- that also feature these must be removed too. The function iterates until it+-- finds a fix-point.+trim :: ([OfferSet], Set.Set Event) -> ([OfferSet], Set.Set Event)+ -- Each iteration, we remove all offersets that reference events that can+ -- never be ready, and if the removing of any of those causes an event+ -- to never become ready, we remove those events too, then we'll go round+ -- 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')+ where+ trimOffer :: (Set.Set Event, Bool) -> OfferSet -> ((Set.Set Event, Bool), OfferSet)+ trimOffer (es, changed) (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+ -- 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:+ eventsNotCompletable = Map.keysSet $ + (unionAll $ map snd eventSetsToRemove)+ `Map.difference` (unionAll $ map snd eventSetsTrimmed)+ changed' = changed+ || not (null eventSetsToRemove)+ in ((es `Set.difference` eventsNotCompletable, changed'),+ OfferSet (tv, pid, eventSetsTrimmed))++-- Semantics of poison with waiting for multiple events is that if /any/ of+-- the events are poisoned, that whole offer will be available immediately+-- even if the other channels are not ready, and the body will throw a poison+-- exception rather than running any of the guards. This is because in situations+-- where for example you want to wait for two inputs (e.g. invisible process+-- or philosopher's forks) you usually want to forward poison from one onto+-- the other.+++-- Finds all the events that could be linked to the given one.+--+-- Given an event, spiders out and discovers all events (connected via mutual offers).+-- Returns the list of offer-sets found. It also+-- returns a set containing each connected completable event.+-- If any of the events are found to be poisoned, the associated STM action is+-- executed+discoverRelatedOffers :: [(STM (), Event)] -> STM (WithPoison ([OfferSet], Set.Set Event))+discoverRelatedOffers = discoverRelatedOffersAll $ NoPoison ([], Set.empty)+ where+ -- We need the supplied STM () actions for each event to take precedence over+ -- the default ones supplied later in the algorithm. So if, for example, the+ -- user supplies a,b and c in the list, but our usual depth-first search would+ -- lead a -> d -> c, we do not want to use the default event for c instead+ -- of the supplied one. Therefore we maintain the work list explicitly.+ + -- Nothing means that that event is poisoned (and thus always ready)+ discoverRelatedOffersAll :: WithPoison ([OfferSet], Set.Set Event)+ -> [(STM (), Event)]+ -> STM (WithPoison ([OfferSet], Set.Set Event))+ discoverRelatedOffersAll PoisonItem _ = return PoisonItem+ discoverRelatedOffersAll x [] = return x+ discoverRelatedOffersAll a@(NoPoison (accum, events)) ((act,e@(Event (_, _, tv))):next)+ -- Don't process the same event multiple times:+ | e `Set.member` events = discoverRelatedOffersAll a next+ | otherwise+ = do x <- readTVar tv+ case x of+ PoisonItem -> act >> return PoisonItem+ NoPoison (count, offers) ->+ let otherEvents = map allEventsInOffer offers in+ if length offers == count+ then -- It could be ready+ discoverRelatedOffersAll+ (NoPoison (accum ++ offers, Set.insert e events))+ (next ++ zip (repeat $ return ())+ (Map.keys $ unionAll otherEvents))+ else -- No way it could be ready, so ignore it:+ discoverRelatedOffersAll a next++-- Given an optional waiting-tvar from the newest process to offer (if any), and+-- an event, spiders out, discovers all the offers, then resolves them and returns+-- a map containing all the completed events, mapping the identifier to the event+-- type and the set of process identifiers that participated in the succesfully+-- completed events. The map will be empty if and only if no events were completed.+discoverAndResolve :: Either OfferSet Event -> STM (WithPoison (Map.Map Unique (RecordedEventType,Set.Set+ ProcessId)))+discoverAndResolve offOrEvent+ = do r <- discoverRelatedOffers $ case offOrEvent of+ Left off@(OfferSet (tv, _, nes)) ->+ let retract = retractOffers [(tv, allEventsInOffer off)] in+ concat [zip (repeat $ retract >> writeTVar tv (Just $ ns+ ++ [0])) (Map.keys es) | (ns, es) <- nes]+ Right e -> [(return (), e)]+ case r of+ PoisonItem -> return PoisonItem+ NoPoison (m, s) -> liftM NoPoison $ resolveOffers tvid (nub m) s+ where+ tvid = case offOrEvent of+ Left (OfferSet (tv, _, _)) -> Just tv+ _ -> Nothing++newEvent :: RecordedEventType -> Int -> IO Event+newEvent t n = do u <- newUnique atomically $ do tv <- newTVar (NoPoison (n, []))- return $ Event (u, tv)+ return $ Event (u, t, tv) enrollEvent :: Event -> STM (WithPoison ())-enrollEvent (Event (_, tv))- = do x <- readTVar tv+enrollEvent e+ = do x <- readTVar $ getEventTVar e case x of PoisonItem -> return PoisonItem NoPoison (count, offers) ->- do writeTVar tv $ NoPoison (count + 1, offers)+ do writeTVar (getEventTVar e) $ NoPoison (count + 1, offers) return $ NoPoison () -resignEvent :: Event -> STM (WithPoison ())-resignEvent (Event (_, tv))- = do x <- readTVar tv+-- If the event completes, we return details related to it:+resignEvent :: Event -> STM (WithPoison [((RecordedEventType, Unique), Set.Set+ ProcessId)])+resignEvent e+ = do x <- readTVar $ getEventTVar e case x of PoisonItem -> return PoisonItem NoPoison (count, offers) ->- do writeTVar tv $ NoPoison (count - 1, offers)+ do writeTVar (getEventTVar e) $ NoPoison (count - 1, offers) if (count - 1 == length offers)- then completeEvent False tv- else return $ NoPoison ()+ then liftM (fmap $ \mu -> [((r,u),pids) | (u,(r,pids)) <- Map.toList mu])+ $ discoverAndResolve $ Right e+ else return $ NoPoison [] -retractOffers :: [(TVar (Maybe [Int]), [Event])] -> STM ()+-- Given the list of identifiers paired with all the events that that process might+-- be engaged in, retracts all the offers during a transaction.+retractOffers :: [(TVar (Maybe [Int]), Map.Map Event ())] -> STM () retractOffers = mapM_ retractAll where- retractAll :: (TVar (Maybe [Int]), [Event]) -> STM ()- retractAll (tvid, evts) = mapM_ retract evts+ retractAll :: (TVar (Maybe [Int]), Map.Map Event ()) -> STM ()+ retractAll (tvid, evts) = mapM_ retract (Map.keys evts) where retract :: Event -> STM ()- retract (Event (_,tv))- = do x <- readTVar tv+ retract e+ = do x <- readTVar $ getEventTVar e case x of PoisonItem -> return () NoPoison (enrolled, offers) ->- let reducedOffers = filter (\(tvx,_,_,_) -> tvx /= tvid) offers in- writeTVar tv $ NoPoison (enrolled, reducedOffers)+ let reducedOffers = filter (\(OfferSet (tvx,_,_)) -> tvx /= tvid) offers in+ writeTVar (getEventTVar e) $ NoPoison (enrolled, reducedOffers) --- Takes True to poison the event, False to complete normally-completeEvent :: Bool -> TVar (WithPoison (Int, [OfferSet])) -> STM (WithPoison ())-completeEvent addPoison tv- = do x <- readTVar tv- case (x, addPoison) of- (PoisonItem, _) -> return PoisonItem- (NoPoison (_, offers), False) ->- do retractOffers $ [(tvw, events) | (tvw,_,_,events) <- offers]- sequence_ [writeTVar tvw (Just wx) | (tvw, wx, _,_) <- offers]- return $ NoPoison ()- (NoPoison (_, offers), True) ->- do retractOffers $ [(tvw, events) | (tvw,_,_,events) <- offers]- sequence_ [writeTVar tvw (Just $ wx ++ [0]) | (tvw, wx, _, _) <- offers]- writeTVar tv PoisonItem- return PoisonItem- +-- Simply adds the offers but doesn't check if that will complete an event:+-- Returns PoisonItem if any of the events were poisoned+makeOffers :: OfferSet -> STM (WithPoison ())+makeOffers offers+ = do let allEvents = Map.keys $ allEventsInOffer offers+ liftM mergeWithPoison $ mapM makeOffer allEvents+ where+ makeOffer :: Event -> STM (WithPoison ())+ makeOffer e+ = do x <- readTVar $ getEventTVar e+ case x of+ PoisonItem -> return PoisonItem+ NoPoison (count, prevOffers) ->+ do writeTVar (getEventTVar e) $ NoPoison (count, offers : prevOffers)+ return $ NoPoison ()+ -- Passed: True if allowed to commit to waiting -- Returns: True if committed, False otherwise-enableEvents :: forall a. TVar (Maybe [Int]) -> ProcessId -> [(a, [Int], STM- (), Event)] -> Bool -> STM (Maybe (a, [ProcessId], [Int]))+enableEvents :: TVar (Maybe [Int]) -> ProcessId -> [([Int], STM+ (), [Event])] -> Bool -> STM (Maybe ([Int], [((RecordedEventType, Unique), Set.Set+ ProcessId)])) enableEvents tvNotify pid events canCommitToWait- = do x <- checkAll- case (x, canCommitToWait) of- (Just (labels, pids, ns, act, Event (_,tv)), _) ->- act >> completeEvent False tv >>= \b ->- return $ Just (labels, pids, case b of {PoisonItem -> ns ++ [0]; _ -> ns})- (Nothing, False) -> return Nothing- (Nothing, True) ->- do sequence_ [do NoPoison (count, otherOffers) <- readTVar ab- writeTVar ab $ NoPoison (count,(tvNotify, ns, pid,- map fourth events):otherOffers)- | (_, ns, _, Event (_,ab)) <- events]+ = do let offer = OfferSet (tvNotify, pid, [(nid, Map.fromList (zip es (repeat ()))) | (nid, _, es) <- events])+ makeOffers offer+ pmu <- discoverAndResolve (Left offer)+ case (canCommitToWait, pmu) of+ (_, PoisonItem) -> do Just chosen <- readTVar tvNotify+ return $ Just (chosen, [])+ (True, NoPoison mu) | Map.null mu -> return Nothing+ (False, NoPoison mu) | Map.null mu ->+ do retractOffers [(tvNotify, Map.fromList $ zip es (repeat ())) | (_,_,es) <- events] return Nothing- where- fourth (_,_,_,c) = c- - checkAll :: STM (Maybe (a, [ProcessId], [Int], STM (), Event))- checkAll = do xs <- sequence [do x <- readTVar tv- return (x, evt)- | evt@(_,_,_,Event (_,tv)) <- events]- return $ fmap get $ find (ready . fst) xs-- where- get :: (WithPoison (Int, [OfferSet]), (a, [Int], STM (), Event))- -> (a, [ProcessId], [Int], STM (), Event)- get (PoisonItem,(l,ns,act,e)) = (l,[pid],ns,act,e)- get (NoPoison (_,offers),(l,ns,act,e)) = (l,pid : [p | (_,_,p,_) <- offers],ns,act,e)-- -- Sees if the barrier is ready *if* we commit to it too- ready :: WithPoison (Int, [OfferSet]) -> Bool- ready PoisonItem = True- ready (NoPoison (count, offers)) = count == 1 + length offers+ (_, NoPoison mu) -> -- Need to turn all the Unique ids back into the custom-typed+ -- parameter that the user gave in the list. We assume+ -- it will be present:+ do {- let y = mapMaybe (\(k,v) -> listToMaybe [(x,v) | (x,_,_,es) <- events,+ k `elem` map getEventUnique es]) $ Map.toList mu+ -}+ Just chosen <- readTVar tvNotify+ return $ Just (chosen, [((r,u),pids) | (u,(r,pids)) <- Map.toList mu]) disableEvents :: TVar (Maybe [Int]) -> [Event] -> STM (Maybe [Int]) disableEvents tv events = do x <- readTVar tv -- Since the transaction will be atomic, we know -- now that we can disable the barriers and nothing fired:- when (isNothing x) $ retractOffers [(tv, events)]+ when (isNothing x) $+ retractOffers [(tv, Map.fromList $ zip events (repeat ()))] return x checkEventForPoison :: Event -> STM (WithPoison ())-checkEventForPoison (Event (_, tv))- = do x <- readTVar tv+checkEventForPoison e+ = do x <- readTVar $ getEventTVar e case x of PoisonItem -> return PoisonItem _ -> return (NoPoison ()) poisonEvent :: Event -> STM ()-poisonEvent (Event (_,tv)) = completeEvent True tv >> return ()+poisonEvent e+ = do x <- readTVar $ getEventTVar e+ case x of+ PoisonItem -> return ()+ NoPoison (_, offers) ->+ do retractOffers [(tvw, unionAll $ map snd events)+ | OfferSet (tvw, _, events) <- offers]+ sequence_ [writeTVar tvw (Just $ pickInts events ++ [0])+ | OfferSet (tvw, _, events) <- offers]+ writeTVar (getEventTVar e) PoisonItem+ where+ pickInts :: [([Int], Map.Map Event ())] -> [Int]+ pickInts es = case filter ((e `Map.member`) . snd) es of+ [] -> [] -- Should never happen+ ((ns,_):_) -> ns --TODO document how if it's poisoned, 0 will be appended to the list++----------------------------------------------------------------------+----------------------------------------------------------------------+-- Testing:+----------------------------------------------------------------------+----------------------------------------------------------------------+++-- Tests if two lists have the same elements, but not necessarily in the same order:+(**==**) :: Eq a => [a] -> [a] -> Bool+a **==** b = (length a == length b) && (null $ a \\ b)++(**/=**) :: Eq a => [a] -> [a] -> Bool+a **/=** b = not $ a **==** b++testDiscover :: IO ()+testDiscover+ = 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 "Three channels, linked by two OR-offerers"+ [(NoPoison 2, False), (NoPoison 2, True), (NoPoison+ 2, False)]+ (zip (repeat True) [ [[0],[1]] , [[1],[2]] ]) [1,2]+ test "Three channels, linked by two AND-offerers"+ [(NoPoison 2, False), (NoPoison 2, True), (NoPoison+ 2, False)]+ (zip (repeat True) [ [[0,1]] , [[1,2]] ]) [0,1]+ test "Three barriers, one process offering all pairs"+ (replicate 3 (NoPoison 2, False))+ [(False,[ [0,1], [0,2], [1,2] ])] [0]+ -- Discovery on a poisoned event will not find offers associated with+ -- that event because they are not stored. The local offer is added+ -- in discoverAndResolve, not testDiscover, so for poison we expect+ -- to find nothing:+ test_Poison "Single poisoned event" [PoisonItem] [ [[0]] ] [0]+ test_Poison "Two poisoned events"+ [PoisonItem, PoisonItem]+ [ [[0,1]] ] [0,1]+ test_Poison "One poisoned, one non-poisoned event"+ [PoisonItem, NoPoison 1] [ [[0,1]] ] [0,1]+ where+ test :: String ->+ {- Events: -} [(WithPoison Int {-count -}, Bool {- Should be in set -})] ->+ {- Offers: -} [(Bool, [[Int] {- events -}])] -> {-Starting events: -} [Int] -> IO ()+ 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,+ present])+ act <- atomically $ discoverRelatedOffers+ $ zip (repeat $ return ()) $ map (events!!) startEvents+ case act of+ PoisonItem -> putStrLn $ testName ++ "Unexpected poison"+ NoPoison actualResult -> do+ when (fst expectedResult **/=** fst actualResult)+ $ putStrLn $ testName ++ " failed offers, exp: "+ ++ show (length $ fst expectedResult)+ ++ " got: " ++ show (length $ fst actualResult)+ when (snd expectedResult /= snd actualResult)+ $ putStrLn $ testName ++ " failed events "+ ++ "exp: " ++ show (snd expectedResult)+ ++ "but got: " ++ show (snd actualResult)+ test_Poison :: String ->+ {- Events: -} [WithPoison Int {-count -}] ->+ {- Offers: -} [[[Int] {- events -}]] -> {-Starting events: -} [Int] -> IO ()+ test_Poison testName eventCounts offerSets startEvents+ = do (events, _realOffers) <- makeTestEvents eventCounts offerSets+ act <- atomically $ discoverRelatedOffers+ $ zip (repeat $ return ()) (map (events!!) startEvents)+ case act of+ PoisonItem -> return ()+ NoPoison _ -> putStrLn $ testName ++ " expected poison but none"++++testTrim :: IO ()+testTrim+ = 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]+ test "Trim, simplified santa not complete"+ (replicate 4 (NoPoison 2, False))+ [ zip (repeat False) [[0,1,2],[0,1,3],[0,2,3],[1,2,3]], [(False, [0])],+ [(False, [1])]] [0]+ test "Trim, simplified santa complete"+ (replicate 3 (NoPoison 2, True) ++ [(NoPoison 2, False)])+ [ [(True,[0,1,2]),(False,[0,1,3]),(False,[0,2,3]),(False,[1,2,3])], [(True, [0])],+ [(True, [1])], [(True, [2])]] [0]+ where+ test :: String -> + {- Events: -} [(WithPoison Int {-count -}, Bool {- expected kept -})] ->+ {- Offers: -} [ [(Bool, [Int]) {- events -}]] -> {-Starting events:-} [Int] -> IO ()+ test testName eventCounts offerSets startEvents+ = do (events, realOffers) <- makeTestEvents (map fst eventCounts) (map (map snd) offerSets)+ let expectedResult' = NoPoison $+ ([OfferSet (tv,pid,[off | (m,off) <- zip [0..] offs, fst $ offerSets !! n !! m])+ | (n,OfferSet (tv,pid,offs)) <- zip [0..] realOffers]+ ,Set.fromList [events !! n+ | (n,(_count, present)) <- zip [0..] eventCounts,+ present])+ actualResult' <- liftM (fmap $ trim . (\(xs,y) -> (nub $ maybe id (:) (listToMaybe realOffers) xs, y)))+ $ 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"+ (NoPoison expectedResult, NoPoison actualResult)+ -> do+ when (fst expectedResult **/=** fst actualResult)+ $ putStrLn $ testName ++ " failed offers, exp: "+ ++ show (length $ fst expectedResult)+ ++ " got: " ++ show (length $ fst actualResult)+ when (snd expectedResult /= snd actualResult)+ $ putStrLn $ testName ++ " failed events, exp: "+ ++ show (snd expectedResult)+ ++ "but got: " ++ show (snd actualResult)++testPoison :: IO ()+testPoison = 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+ test "Poison, offered on two (OR)" [(NoPoison 2, PoisonItem), (NoPoison 2, NoPoison [])] [[[0],[1]]] 0+ where+ test :: String ->+ [(WithPoison Int {-count -}, WithPoison [Int] {- remaining offers -})] ->+ {- Offers: -} [[[Int] {- events -}]] -> Int {-Poison Event-} -> IO ()++ test testName eventCounts offerSets poisoned = do+ (events, realOffers) <- makeTestEvents (map fst eventCounts) offerSets+ atomically $ poisonEvent $ events !! poisoned+ -- Now we must check that the event is poisoned, and that all processes+ -- that were offering on that event have had their offers retracted (by+ -- checking that only the specified offers remain on each event)++ sequence_ [do x <- atomically $ readTVar $ getEventTVar $ events !! n+ case (expect, x) of+ (PoisonItem, PoisonItem) -> return ()+ (NoPoison _, PoisonItem) -> putStrLn $ testName +++ " expected no poison but found it"+ (PoisonItem, NoPoison _) -> putStrLn $ testName +++ " expected poison but found none"+ (NoPoison expOff, NoPoison (_, actOff)) ->+ when (map (realOffers !!) expOff **/=** actOff) $+ putStrLn $ testName ++ " offers did not match"+ | (n, (_, expect)) <- zip [0..] eventCounts]+++ +testAll :: IO ()+testAll = testDiscover >> testTrim >> testResolve >> testPoison++makeTestEvents ::+ {- Events: -} [WithPoison Int {-count -}] ->+ {- Offers: -} [[[Int] {- events -}]] -> IO ([Event], [OfferSet])+ -- Offers is a list of list of list of ints+ -- Outermost list is one-per-process+ -- Middle list is one-per-offer+ -- Inner list is a conjunction of events+makeTestEvents eventCounts offerSets+ = do events <- mapM (\n -> newEvent ChannelComm $ case n of+ NoPoison n' -> n'+ PoisonItem -> 0) eventCounts+ -- Poison all the events marked as poisoned:+ atomically $ sequence_ [writeTVar tv PoisonItem | (n,Event (_,_,tv)) <- zip [0..] events,+ eventCounts !! n == PoisonItem]+ realOffers <- sequence+ [ do tv <- atomically $ newTVar Nothing+ let pid = testProcessId processN+ offSub = [ ([processN, offerN],+ Map.fromList [ (events !! indivEvent, ())+ | indivEvent <- singleOffer])+ | (offerN, singleOffer) <- zip [0..] processOffers]+ off = OfferSet (tv, pid, offSub)+ mapM_ (\e -> atomically $ do+ x <- readTVar (getEventTVar e)+ case x of+ NoPoison (count, offs) ->+ writeTVar (getEventTVar e) $ NoPoison (count, off : offs)+ PoisonItem -> return ()+ ) (Map.keys $ unionAll $ map snd offSub)+ return off+ | (processN, processOffers) <- zip [0..] offerSets]+ return (events, realOffers)++testResolve :: IO ()+testResolve+ = 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]]]+ test "Two channels, two single offerers and one double"+ [(NoPoison 2, Left [(0,0),(2,0)]), (NoPoison 2, Left [(1,0),(2,0)])]+ [ [[0]], [[1]], [[0,1]] ]+ test "Two channels, two single offerers and one choosing"+ [(NoPoison 2, Left [(0,0),(2,0)]), (NoPoison 2, Right [1])]+ [ [[0]], [[1]], [[0],[1]] ]+ test "Three channels, both offering different pair"+ [(NoPoison 2, Right []), (NoPoison 2, Left [(0,1),(1,0)]), (NoPoison 2, Right [])]+ [ [[0],[1]] , [[1],[2]] ]+ -- This test is a bit hacky, given there are two valid results:+ test "Two channels, both could complete"+ [(NoPoison 2, Left [(0,0),(1,0)]), (NoPoison 2, Right [])]+ [ [[0],[1]] , [[0],[1]] ]+ test "Three channels, any could complete"+ [(NoPoison 2, Left [(0,0),(1,0)]), (NoPoison 2, Right [2]), (NoPoison 2,+ Right [2])]+ [ [[0],[1]] , [[0],[2]], [[1],[2]] ]+ test "Three channels, one guy offering three pairs, two single offerers"+ [(NoPoison 2, Left [(0,1),(1,0)]), (NoPoison 2, Right []), (NoPoison 2,+ Left [(0,1),(2,0)])]+ [ [[0,1],[0,2],[1,2]], [[0]], [[2]] ]+ test "Three channels, one guy offering three pairs, three single offerers"+ [(NoPoison 2, Left [(0,0),(1,0)]), (NoPoison 2, Left [(0,0),(2,0)]), (NoPoison 2,+ Right [3])]+ [ [[0,1],[0,2],[1,2]], [[0]], [[1]], [[2]] ]+ test "Four channels, one guy offering sets of three, three single offerers"+ [(NoPoison 2, Left [(0,0),(1,0)]), (NoPoison 2, Left [(0,0),(2,0)]),+ (NoPoison 2, Left [(0,0),(3,0)]), (NoPoison 2, Right [])]+ [ [[0,1,2],[0,1,3],[0,2,3],[1,2,3]], [[0]], [[1]], [[2]] ]+ test "Four channels, one guy offering sets of three, two single offerers"+ [(NoPoison 2, Right [1,0]), (NoPoison 2, Right [2,0]),+ (NoPoison 2, Right [0]), (NoPoison 2, Right [0])]+ [ [[0,1,2],[0,1,3],[0,2,3],[1,2,3]], [[0]], [[1]] ]+ -- test resolutions with poison:+ --+ test' "One event, poisoned"+ [(PoisonItem, Left [(0,0)])]+ [[[0]]] True+ test' "Two events, one poisoned"+ [(PoisonItem, Left [(0,0)]), (NoPoison 2, Left [(0,0)])]+ [[[0,1]]] True+ where+ test testName eventCounts offerSets = test' testName eventCounts offerSets False+ + test' :: String ->+ [(WithPoison Int {-count -},+ Either [(Int, Int)] {- success: expected process, offer indexes -}+ [Int] {- remaining offers -})] ->+ {- Offers: -} [[[Int] {- events -}]] -> Bool {-Poisoned-} -> IO ()++ test' testName eventCounts offerSets poisoned = do+ (events, realOffers) <- makeTestEvents (map fst eventCounts) offerSets++ actualResult <- atomically $ discoverAndResolve $ Left $ head realOffers+ let expectedResult = if poisoned then PoisonItem else NoPoison $+ Map.fromList [ (getEventUnique e, (ChannelComm,+ Set.fromList $ map (testProcessId . fst) is))+ | (e, Left is) <- zip events (map snd eventCounts)]+ when (expectedResult /= actualResult) $+ putStrLn $ 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)+ Just v -> when (v /= (if poisoned then (++[0]) else id) [pn, en]) $+ putStrLn $ testName ++ " wrong choice: " ++ show v ++ " exp: " ++ show+ [pn, en]+ return pn+ ) $ map snd eventCounts+ -- test the others are unchanged+ sequence_ [ let OfferSet (tv,_,_) = realOffers !! n in+ do x <- atomically $ readTVar tv+ case x of+ Nothing -> return ()+ Just _ -> putStrLn $ testName ++ " Unexpected win for process: " +++ show n ++ " " ++ show x+ | n <- [0 .. length offerSets - 1] \\ allFired]+ -- check events are blanked afterwards:+ sequence_ [ let e = events !! n+ expVal = case st of+ Left _ -> []+ Right ns -> map (realOffers !!) ns in do+ x <- atomically $ readTVar $ getEventTVar e+ when (x /= NoPoison (count, expVal)) $+ putStrLn $ testName ++ "Event " ++ show n +++ " not as expected after, exp: " ++ show (length expVal)+ ++ " act: " ++ (let NoPoison (_,act) = x in show (length act))+ | (n,(NoPoison count, st)) <- zip [0..] eventCounts]++ showStuff = show . fmap (map (\(u,x) -> (hashUnique u, x)) . Map.toList)
Control/Concurrent/CHP/Guard.hs view
@@ -40,16 +40,16 @@ data Guard = TimeoutGuard (IO (STM [Int])) | SkipGuard [Int] | StopGuard- | BadGuard+ | BadGuard String -- The STM item is an action to take in the same transaction as -- completing the event (before it is completed).- | EventGuard [Int] RecEvents (STM ()) Event+ | EventGuard [Int] [RecordedIndivEvent] (STM ()) [Event] | NestedGuards [Guard] skipGuard :: Guard skipGuard = SkipGuard [] -badGuard :: Guard+badGuard :: String -> Guard badGuard = BadGuard stopGuard :: Guard
Control/Concurrent/CHP/Monad.hs view
@@ -106,7 +106,7 @@ -- | Continues executing the loop if the given value is True. If the value -- is False, the loop is broken immediately, and control jumps back to the -- next action after the outer 'loop' statement. Thus you can build pre-condition,--- post-condition, and "mid-condition" loops, placing the condition wherever+-- post-condition, and \"mid-condition\" loops, placing the condition wherever -- you like. while :: Monad m => Bool -> LoopWhileT m () while b = LWT $ if b then (return $ Just ()) else return Nothing@@ -116,12 +116,12 @@ -- There is no guaranteed precision, but the wait will never complete in less -- time than the parameter given. -- --- Suitable for use in an 'alt', but note that waitFor 0 is not the same--- as skip. 'waitFor' 0 '</>' x will not always select the first guard,+-- Suitable for use in an 'Control.Concurrent.CHP.Alt.alt', but note that 'waitFor' 0 is not the same+-- as 'skip'. 'waitFor' 0 'Control.Concurrent.CHP.Alt.</>' x will not always select the first guard, -- depending on x. Included in this is the lack of guarantee that--- '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.+-- 'waitFor' 0 'Control.Concurrent.CHP.Alt.</>' '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 'Control.Concurrent.CHP.Alt.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).@@ -133,12 +133,12 @@ -- | The classic skip process\/guard. Does nothing, and is always ready. ----- Suitable for use in an 'alt'.+-- Suitable for use in an 'Control.Concurrent.CHP.Alt.alt'. skip :: CHP () skip = liftPoison $ AltableT (skipGuard, return ()) (return ()) -- | The stop guard. Its main use is that it is never ready in a choice, so--- can be used to mask out guards. If you actually execute stop, that process+-- can be used to mask out guards. If you actually execute 'stop', that process -- will do nothing more. Any parent process waiting for it to complete will -- wait forever. stop :: CHP ()
Control/Concurrent/CHP/Parallel.hs view
@@ -27,7 +27,7 @@ -- NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS -- SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. -module Control.Concurrent.CHP.Parallel (runParallel, runParallel_, (<||>),+module Control.Concurrent.CHP.Parallel (runParallel, runParallel_, (<||>), (<|*|>), ForkingT, forking, fork) where import Control.Concurrent@@ -50,7 +50,7 @@ -- -- In each case, the composition waits for all processes to finish, either -- successfully or with poison. At the end of this, if /any/ process--- exited with poison, the composition will "rethrow" this poison. If all+-- exited with poison, the composition will \"rethrow\" this poison. If all -- the processes completed successfully, the results will be returned. If -- you want to ignore poison from the sub-processes, use an empty poison -- handler and 'onPoisonTrap' with each branch.@@ -61,7 +61,7 @@ runParallel :: [CHP a] -> CHP [a] runParallel = runParallelPoison --- | A useful operator for performing a two process equivalent of runParallel+-- | A useful operator for performing a two process equivalent of 'runParallel' -- that gives the return values back as a pair rather than a list. This also -- allows the values to have different types (<||>) :: CHP a -> CHP b -> CHP (a, b)@@ -73,10 +73,16 @@ combine (Right y) (Left x) = return (x, y) -- An extra case to keep the compiler happy: combine _ _ = error "Impossible combination values in <|^|>"- +-- | An operator similar to '<||>' that discards the output (more like an operator+-- version of 'runParallel_').+--+-- Added in version 1.1.0.+(<|*|>) :: CHP a -> CHP b -> CHP ()+(<|*|>) p q = runParallel_ [p >> return (), q >> return ()]+ -- | Runs all the given processes in parallel and discards any output. Does--- not return until all the processes have completed. runParallel_ ps is effectively equivalent+-- not return until all the processes have completed. 'runParallel_' ps is effectively equivalent -- to 'runParallel' ps >> return (). runParallel_ :: [CHP a] -> CHP () runParallel_ procs = runParallel procs >> return ()@@ -84,6 +90,8 @@ -- We right associate to allow the liftM fst ((readResult) <||> runParallel_ -- workers) pattern infixr <||>+-- Doesn't really matter for this operator:+infixr <|*|> wrapProcess :: CHP a -> (CHP' (Either PoisonError a) -> IO (Either PoisonError@@ -155,7 +163,7 @@ -- | Forks off the given process. The process then runs in parallel with this--- code, until the end of the forking block, when all forked-off processes+-- code, until the end of the 'forking' block, when all forked-off processes -- are waited for. At that point, once all of the processes have finished, -- if any of them threw poison it is propagated. fork :: MonadCHP m => CHP () -> ForkingT m ()
Control/Concurrent/CHP/Poison.hs view
@@ -30,9 +30,16 @@ module Control.Concurrent.CHP.Poison where -- | A Maybe-like poison wrapper.-data WithPoison a = PoisonItem | NoPoison a deriving (Eq)+data WithPoison a = PoisonItem | NoPoison a deriving (Eq, Show) instance Functor WithPoison where fmap f (NoPoison x) = NoPoison $ f x fmap _ PoisonItem = PoisonItem +instance Monad WithPoison where+ return = NoPoison+ PoisonItem >>= _ = PoisonItem+ NoPoison x >>= f = f x++mergeWithPoison :: [WithPoison a] -> WithPoison ()+mergeWithPoison = sequence_
Control/Concurrent/CHP/ProcessId.hs view
@@ -34,15 +34,18 @@ -- in the tracing paper, which is provided by the pidLessThanOrEqual function -- below. -data ProcessIdPart = ParSeq Int Integer deriving (Eq, Ord)+data ProcessIdPart = ParSeq Int Integer deriving (Eq, Ord, Show) -newtype ProcessId = ProcessId [ProcessIdPart] deriving (Eq, Ord)+newtype ProcessId = ProcessId [ProcessIdPart] deriving (Eq, Ord, Show) rootProcessId :: ProcessId rootProcessId = ProcessId [ParSeq 0 0] emptyProcessId :: ProcessId emptyProcessId = ProcessId []++testProcessId :: Int -> ProcessId+testProcessId n = ProcessId [ParSeq n 0] -- We use here the property that the lists will either be equal, or different -- within their shared length. Therefore they will be different when zipped,
Control/Concurrent/CHP/Traces.hs view
@@ -48,19 +48,29 @@ -- It could hardly be easier. If you want more fine-grained control and examination -- of the trace, you can use the helper functions of the form 'runCHP_CSPTrace' -- that give back a data structure representing the trace, which you can then--- manipulate. The Doc used by the traces is from the 'Text.PrettyPrint.HughesPJ'+-- manipulate.+--+-- The Doc used by the traces is from the 'Text.PrettyPrint.HughesPJ' -- module that currently comes with GHC (I think).+--+-- For more details on the theory behind the tracing, the logic behind its+-- implementation, and example traces, see the paper \"Representation and Implementation+-- of CSP and VCR Traces\", N.C.C. Brown and M.L. Smith, CPA 2008. An online version can+-- be found at: <http://twistedsquare.com/Traces.pdf> module Control.Concurrent.CHP.Traces (module Control.Concurrent.CHP.Traces.CSP ,module Control.Concurrent.CHP.Traces.Structural ,module Control.Concurrent.CHP.Traces.TraceOff ,module Control.Concurrent.CHP.Traces.VCR- ,RecordedEvent(..)+ ,RecordedEvent+ ,ChannelLabels+ ,RecordedEventType(..) ,RecordedIndivEvent(..) ,Trace(..) ) where import Control.Concurrent.CHP.Base+import Control.Concurrent.CHP.Event import Control.Concurrent.CHP.Traces.Base import Control.Concurrent.CHP.Traces.CSP import Control.Concurrent.CHP.Traces.Structural
Control/Concurrent/CHP/Traces/Base.hs view
@@ -45,10 +45,7 @@ -- identifiers are per channel\/barrier, not per event. Currently, -- channels and barriers can never have the same Unique as each other, but -- do not rely on this behaviour.-data RecordedEvent =- ChannelComm Unique- | BarrierSync Unique- deriving (Eq, Ord)+type RecordedEvent = (RecordedEventType, Unique) -- | An individual record of an event, found in nested traces. Either a -- channel write or read, or a barrier synchronisation, each with a unique@@ -64,7 +61,7 @@ | BarrierSyncIndiv Unique deriving (Eq, Ord) -type RecEvents = (Maybe RecordedEvent, Maybe RecordedIndivEvent)+type RecEvents = ([RecordedEvent], [RecordedIndivEvent]) getName :: Unique -> State (Map.Map Unique String) String getName u = do m <- get@@ -75,8 +72,7 @@ return x nameEvent :: RecordedEvent -> State (Map.Map Unique String) String-nameEvent (ChannelComm c) = getName c-nameEvent (BarrierSync c) = getName c+nameEvent (_, c) = getName c nameIndivEvent :: RecordedIndivEvent -> State (Map.Map Unique String) String nameIndivEvent (ChannelWrite c) = do c' <- getName c@@ -116,34 +112,33 @@ deriving (Eq, Ord) -- | Records an event where you were the last person to engage in the event-recordEventLast :: Maybe RecordedEvent -> Set.Set ProcessId -> TraceStore -> STM ()-recordEventLast Nothing _ _ = return ()-recordEventLast (Just e) otherPids y+recordEventLast :: [(RecordedEvent, Set.Set ProcessId)] -> TraceStore -> STM ()+recordEventLast news y = case y of Trace (_,_,CSPTraceRev tv) ->- do t <- readTVar tv- writeTVar tv $! addRLE e t+ do t <- readTVar tv+ writeTVar tv $! foldl (flip addRLE) t (map fst news) Trace (pid, _, VCRTraceRev tv) -> do t <- readTVar tv- let pidSet = Set.insert pid otherPids+ let pidSet = (foldl Set.union (Set.singleton pid) $ map snd news)+ news' = map (\(a,b) -> (b,a)) news t' = case t of -- Trace previously empty:- [] -> [Set.singleton (pidSet, e)]+ [] -> [Set.fromList news'] (z:zs) | shouldMakeNewSetVCR pidSet z- -> Set.singleton (pidSet, e) : t+ -> Set.fromList news' : t | otherwise- -> Set.insert (pidSet, e) z : zs+ -> foldl (flip Set.insert) z news' : zs writeTVar tv $! t' _ -> return () -- | Records an event where you were one of the people involved-recordEvent :: Maybe RecordedIndivEvent -> TraceT IO ()-recordEvent Nothing = return ()-recordEvent (Just e)+recordEvent :: [RecordedIndivEvent] -> TraceT IO ()+recordEvent e = do (x,y) <- get- case (x, y, e) of- (as, Trace (pid,tvls,Hierarchy es),_) ->- put (as, Trace (pid, tvls, Hierarchy (addSeqEventH e es)))+ case (x, y) of+ (as, Trace (pid,tvls,Hierarchy es)) ->+ put (as, Trace (pid, tvls, Hierarchy (foldl (flip addSeqEventH) es e))) _ -> return () mergeSubProcessTraces :: [TraceStore] -> TraceT IO ()@@ -216,8 +211,8 @@ labelEvent :: Event -> String -> StateT (a, TraceStore) IO ()-labelEvent (Event (u,_)) l- = labelUnique u l+labelEvent e l+ = labelUnique (getEventUnique e) l labelUnique :: Unique -> String -> StateT (a, TraceStore) IO () labelUnique u l
chp.cabal view
@@ -1,5 +1,5 @@ Name: chp-Version: 1.0.2+Version: 1.1.0 Synopsis: An implementation of concurrency ideas from Communicating Sequential Processes License: BSD3 License-file: LICENSE@@ -52,4 +52,4 @@ FlexibleInstances UndecidableInstances GeneralizedNewtypeDeriving -GHC-Options: -Wall -threaded -O2+GHC-Options: -Wall -threaded