chp 2.0.0 → 2.1.0
raw patch · 16 files changed
+1299/−911 lines, 16 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
+ Control.Concurrent.CHP.Barriers: barPriority :: BarOpts phase -> Int
+ Control.Concurrent.CHP.Barriers: newBarrierPri :: Int -> CHP Barrier
+ Control.Concurrent.CHP.Channels.Creation: chanOptsPriority :: ChanOpts a -> Int
- Control.Concurrent.CHP.Barriers: BarOpts :: (phase -> phase) -> (phase -> String) -> Maybe String -> BarOpts phase
+ Control.Concurrent.CHP.Barriers: BarOpts :: (phase -> phase) -> Int -> (phase -> String) -> Maybe String -> BarOpts phase
- Control.Concurrent.CHP.Channels.Creation: ChanOpts :: (a -> String) -> Maybe String -> ChanOpts a
+ Control.Concurrent.CHP.Channels.Creation: ChanOpts :: Int -> (a -> String) -> Maybe String -> ChanOpts a
Files
- Control/Concurrent/CHP/Alt.hs +9/−10
- Control/Concurrent/CHP/Barriers.hs +20/−13
- Control/Concurrent/CHP/Base.hs +0/−1
- Control/Concurrent/CHP/CSP.hs +0/−3
- Control/Concurrent/CHP/Channels/Base.hs +3/−3
- Control/Concurrent/CHP/Channels/BroadcastReduce.hs +4/−4
- Control/Concurrent/CHP/Channels/Communication.hs +1/−1
- Control/Concurrent/CHP/Channels/Creation.hs +15/−6
- Control/Concurrent/CHP/Event.hs +891/−862
- Control/Concurrent/CHP/EventMap.hs +100/−0
- Control/Concurrent/CHP/EventSet.hs +113/−0
- Control/Concurrent/CHP/EventType.hs +136/−0
- Control/Concurrent/CHP/Guard.hs +0/−1
- Control/Concurrent/CHP/Monad.hs +2/−4
- Control/Concurrent/CHP/Parallel.hs +1/−2
- chp.cabal +4/−1
Control/Concurrent/CHP/Alt.hs view
@@ -105,10 +105,9 @@ import Control.Applicative import Control.Arrow+import Control.Concurrent import Control.Concurrent.STM import Control.Monad.Reader-import Control.Monad.State-import Control.Monad.Trans import Data.List import qualified Data.Map as Map import Data.Maybe@@ -403,11 +402,12 @@ tv <- liftIO $ newTVarIO Nothing pid <- getProcessId tr <- ask+ tid <- liftIO myThreadId mn <- liftIO . atomically $ do- ret <- enableEvents tv pid+ ret <- enableEvents tv (tid, pid) (maybe id take earliestReady $ eventGuards guards) (isNothing earliestReady)- maybe (return ())+ either (const $ return ()) (\((sigVal,_),es) -> do recordEventLast (nub es) tr case sigVal of@@ -417,19 +417,19 @@ in actWhenLast act (Map.fromList $ map (snd *** Set.size) es) ) ret- return $ fmap (getRec . fst) ret+ return $ either Left (Right . getRec . fst) 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 r, _) -> recordAndRun r+ (Right r, _) -> recordAndRun r -- No events were ready, but there was an available normal -- guards. Re-run the normal guards; at least one will be ready- (Nothing, Just _) ->+ (Left _, Just _) -> join $ liftM snd $ liftIO $ waitNormalGuards both Nothing -- No events ready, no other guards ready either -- Events will have been enabled; wait for everything:- (Nothing, Nothing) ->+ (Left disable, Nothing) -> do (wasAltingBarrier, pr) <- liftIO $ waitNormalGuards guardsAndRec $ Just $ liftM getRec $ waitAlting tv if wasAltingBarrier@@ -438,8 +438,7 @@ -- Another guard fired, but we must check in case -- we have meanwhile been committed to taking an -- event:- do mn' <- liftIO . atomically $ disableEvents tv (concatMap snd- $ eventGuards guards)+ do mn' <- liftIO . atomically $ disable case mn' of -- An event overrides our non-event choice: Just pr' -> recordAndRun $ getRec pr'
Control/Concurrent/CHP/Barriers.hs view
@@ -63,19 +63,17 @@ -- 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, EnrolledBarrier, newBarrier, newBarrierWithLabel,+module Control.Concurrent.CHP.Barriers (Barrier, EnrolledBarrier, newBarrier, newBarrierPri, newBarrierWithLabel, PhasedBarrier, newPhasedBarrier, newPhasedBarrier', BarOpts(..), defaultIncPhase, defaultBarOpts, barLabel, currentPhase, waitForPhase, syncAndWaitForPhase, syncBarrier, getBarrierIdentifier) where import Control.Concurrent.STM import Control.Monad.State-import Control.Monad.Trans import Data.Unique import Control.Concurrent.CHP.Base import Control.Concurrent.CHP.CSP-import Control.Concurrent.CHP.Enroll import Control.Concurrent.CHP.Event import Control.Concurrent.CHP.Traces.Base @@ -126,7 +124,10 @@ -- Integer as the inner type to get a barrier that never cycles. You can also -- do things like supplying (+2) as the incrementing function, or even using -- lists as the phase type to do crazy things.- , barOptsShow :: phase -> String, barOptsLabel :: Maybe String }+ , barPriority :: Int+ -- ^ Added in version 2.1.0. See 'Control.Concurrent.CHP.Channels.Creation.ChanOpts'.+ , barOptsShow :: phase -> String+ , barOptsLabel :: Maybe String } -- | The default phase incrementing function. If the phase is already at 'maxBound', -- it sets it to 'minBound'; otherwise it uses 'succ' to increment the phase.@@ -140,41 +141,47 @@ -- -- Added in version 1.7.0. defaultBarOpts :: (Enum phase, Bounded phase, Eq phase) => BarOpts phase-defaultBarOpts = BarOpts defaultIncPhase (const "") Nothing+defaultBarOpts = BarOpts defaultIncPhase 0 (const "") Nothing -- | Uses the Show instance for showing the data in traces, and the given label. -- -- Added in version 1.7.0. barLabel :: (Enum phase, Bounded phase, Eq phase, Show phase) => String -> BarOpts phase-barLabel = BarOpts defaultIncPhase show . Just+barLabel = BarOpts defaultIncPhase 0 show . Just -- | Creates a new barrier with no processes enrolled newBarrier :: CHP Barrier-newBarrier = newPhasedBarrier' () $ BarOpts (const ()) (const "") Nothing+newBarrier = newBarrierPri 0 -newBarrierEvent :: (phase -> String) -> TVar phase -> IO Event-newBarrierEvent sh tv = newEvent (liftM (BarrierSync . sh) $ readTVar tv) 0+-- | Creates a new barrier with no processes enrolled and the given priority.+--+-- Added in version 2.1.0.+newBarrierPri :: Int -> CHP Barrier+newBarrierPri n = newPhasedBarrier' () $ BarOpts (const ()) n (const "") Nothing +newBarrierEvent :: (phase -> String) -> Int -> TVar phase -> IO Event+newBarrierEvent sh pri tv = newEventPri (liftM (BarrierSync . sh) $ readTVar tv) 0 pri+ -- | Creates a new barrier with no processes enrolled, that will be on the -- given phase. You will often want to pass in the last value in your phase -- cycle, so that the first synchronisation moves it on to the first -- -- The Show constraint was added in version 1.5.0 newPhasedBarrier :: (Enum phase, Bounded phase, Eq phase, Show phase) => phase -> CHP (PhasedBarrier phase)-newPhasedBarrier ph = newPhasedBarrier' ph $ BarOpts defaultIncPhase show Nothing+newPhasedBarrier ph = newPhasedBarrier' ph $ BarOpts defaultIncPhase 0 show Nothing -- | Like 'newPhasedBarrier' but allows you to customise the options. newPhasedBarrier' :: phase -> BarOpts phase -> CHP (PhasedBarrier phase)-newPhasedBarrier' ph (BarOpts incPh showPh label) = liftPoison $ liftTrace $ do+newPhasedBarrier' ph (BarOpts incPh pri showPh label) = liftPoison $ liftTrace $ do tv <- liftIO $ atomically $ newTVar ph- e <- liftIO $ newBarrierEvent showPh tv + e <- liftIO $ newBarrierEvent showPh pri tv maybe (return ()) (labelEvent e) label return $ Barrier (e, tv, incPh) -- | Creates a new barrier with no processes enrolled and labels it in traces -- using the given label. See 'newBarrier'. newBarrierWithLabel :: String -> CHP Barrier-newBarrierWithLabel = newPhasedBarrier' () . BarOpts (const ()) (const "") . Just+newBarrierWithLabel = newPhasedBarrier' () . BarOpts (const ()) 0 (const "") . Just -- | Gets the identifier of a Barrier. Useful if you want to identify it in -- the trace later on.
Control/Concurrent/CHP/Base.hs view
@@ -42,7 +42,6 @@ import Control.Monad.Reader import Control.Monad.State import Control.Monad.Writer-import Control.Monad.Trans import Data.Function (on) import qualified Data.Map as Map import Data.Unique
Control/Concurrent/CHP/CSP.hs view
@@ -35,12 +35,9 @@ import Control.Concurrent.STM import Control.Exception import Control.Monad.Reader-import Control.Monad.Writer-import Control.Monad.Trans import Data.List import qualified Data.Map as Map import Data.Unique-import System.IO import Control.Concurrent.CHP.Alt import Control.Concurrent.CHP.Base
Control/Concurrent/CHP/Channels/Base.hs view
@@ -108,10 +108,10 @@ checkForPoison (Chanout c) = liftCHP $ liftIO (checkPoisonWriteC c) >>= checkPoison -stmChannel :: MonadIO m => (a -> String) -> m (Unique, STMChannel a)-stmChannel sh = liftIO $+stmChannel :: MonadIO m => Int -> (a -> String) -> m (Unique, STMChannel a)+stmChannel pri sh = liftIO $ do c <- atomically $ newTVar $ NoPoison (Nothing, Nothing)- e <- newEvent (liftM (ChannelComm . maybe "" sh . getVal) $ readTVar c) 2+ e <- newEventPri (liftM (ChannelComm . maybe "" sh . getVal) $ readTVar c) 2 pri return (getEventUnique e, STMChan (e,c)) where getVal PoisonItem = Nothing
Control/Concurrent/CHP/Channels/BroadcastReduce.hs view
@@ -119,7 +119,7 @@ (x, r) <- m liftIO . atomically $ writeTVar tvSend $ Just x -- Must be two separate transactions:- liftIO . atomically $ readManyToOneTVar tvAck+ _ <- liftIO . atomically $ readManyToOneTVar tvAck return r instance ReadableChannel (Enrolled BroadcastChanin) where@@ -128,7 +128,7 @@ (resetManyToOneTVar tvAck . pred) $ Enrolled b x <- liftIO $ atomically $ readTVar tvSend >>= maybe retry return y <- f x- liftIO $ atomically $ writeManyToOneTVar pred tvAck+ _ <- liftIO $ atomically $ writeManyToOneTVar pred tvAck return y instance Poisonable (BroadcastChanout a) where@@ -143,7 +143,7 @@ newBroadcastChannel = do b@(Barrier (e, _, _)) <- newBarrier -- Writer is always enrolled:- liftIO $ atomically $ enrollEvent e+ _ <- liftIO $ atomically $ enrollEvent e tvSend <- liftIO $ atomically $ newTVar Nothing tvAck <- liftIO $ atomically $ newManyToOneTVar (== 0) (return 0) 0 return $ BC (b, tvSend, tvAck)@@ -247,7 +247,7 @@ newReduceChannel = do b@(Barrier (e, _, _)) <- newBarrier -- Writer is always enrolled:- liftIO $ atomically $ enrollEvent e+ _ <- liftIO $ atomically $ enrollEvent e mtv <- liftIO $ atomically $ newManyToOneTVar ((== 0) . fst) (return (0, Nothing)) (0, Nothing) return $ GC (b, mtv, (mappend, mempty))
Control/Concurrent/CHP/Channels/Communication.hs view
@@ -139,7 +139,7 @@ scopeBlock (buildOnEventPoison (wrapIndiv $ indivRecJust ChannelRead) e mempty (liftSTM m) >>= checkPoison) (\val -> do x <- body val- liftSTM $ endReadChannelC c+ _ <- liftSTM $ endReadChannelC c return x) (poisonReadC c)
Control/Concurrent/CHP/Channels/Creation.hs view
@@ -102,19 +102,28 @@ -- label (both only affect tracing). These options can be passed to newChannel'. -- -- Added in version 1.5.0.-data ChanOpts a = ChanOpts { chanOptsShow :: a -> String, chanOptsLabel :: Maybe String }+data ChanOpts a = ChanOpts {+ chanOptsPriority :: Int,+ -- ^ Added in version 2.1.0. Priority is per-event, static and system-wide.+ -- If it is possible at any given moment for a process to resolve a choice one+ -- of several ways, the channel/barrier with the highest priority is chosen.+ -- If the choice is a conjunction and all events in one conjunction are higher+ -- than all the events in the other, the higher one is chosen (otherwise no guarantees+ -- are made). The default is zero, and the range is the full range of Int (both+ -- positive and negative).+ chanOptsShow :: a -> String, chanOptsLabel :: Maybe String } -- | The default: don't show anything, don't label anything -- -- Added in version 1.5.0. defaultChanOpts :: ChanOpts a-defaultChanOpts = ChanOpts (const "") Nothing+defaultChanOpts = ChanOpts 0 (const "") Nothing -- | Uses the Show instance for showing the data in traces, and the given label. -- -- Added in version 1.5.0. chanLabel :: Show a => String -> ChanOpts a-chanLabel = ChanOpts show . Just+chanLabel = ChanOpts 0 show . Just -- | Allocates a new channel. Nothing need be done to -- destroy\/de-allocate the channel when it is no longer in use.@@ -161,13 +170,13 @@ -- to a pattern. Given a stem such as foo, it names the channels in the list -- foo0, foo1, foo2, etc. newChannelListWithStem :: (Channel r w, MonadCHP m) => Int -> String -> m [Chan r w a]-newChannelListWithStem n s = sequence [newChannel' $ ChanOpts (const "") (Just $ s ++ show i) | i <- [0 .. (n - 1)]]+newChannelListWithStem n s = sequence [newChannel' $ ChanOpts 0 (const "") (Just $ s ++ show i) | i <- [0 .. (n - 1)]] -- | A helper that is like 'newChannelList', but labels the channels with the -- given list. The number of channels returned is the same as the length of -- the list of labels newChannelListWithLabels :: (Channel r w, MonadCHP m) => [String] -> m [Chan r w a]-newChannelListWithLabels = mapM (newChannel' . ChanOpts (const "") . Just)+newChannelListWithLabels = mapM (newChannel' . ChanOpts 0 (const "") . Just) instance (Channel r w) => ChannelTuple (Chan r w a, Chan r w a) where newChannels = do c0 <- newChannel@@ -216,7 +225,7 @@ instance Channel Chanin Chanout where- newChannel' o = do c <- chan (stmChannel $ chanOptsShow o) Chanin Chanout+ newChannel' o = do c <- chan (stmChannel (chanOptsPriority o) (chanOptsShow o)) Chanin Chanout maybe (return ()) (labelChannel c) (chanOptsLabel o) return c sameChannel (Chanin x) (Chanout y) = x == y
Control/Concurrent/CHP/Event.hs view
@@ -1,865 +1,894 @@--- Communicating Haskell Processes.--- Copyright (c) 2008, University of Kent.--- All rights reserved.--- --- Redistribution and use in source and binary forms, with or without--- modification, are permitted provided that the following conditions are--- met:------ * Redistributions of source code must retain the above copyright--- notice, this list of conditions and the following disclaimer.--- * Redistributions in binary form must reproduce the above copyright--- notice, this list of conditions and the following disclaimer in the--- documentation and/or other materials provided with the distribution.--- * Neither the name of the University of Kent nor the names of its--- contributors may be used to endorse or promote products derived from--- this software without specific prior written permission.------ THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS--- IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,--- THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR--- PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR--- CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,--- EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,--- PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR--- PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF--- LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING--- NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS--- SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.----TODO document this (for internal purposes)-module Control.Concurrent.CHP.Event (RecordedEventType(..), Event, getEventUnique,- SignalVar, SignalValue(..), enableEvents, disableEvents,- newEvent, newEventUnique, enrollEvent, resignEvent, poisonEvent, checkEventForPoison,- getEventTypeVal-#ifdef CHP_TEST- , testAll-#endif- ) where--import Control.Arrow-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 qualified Data.Traversable as T-import Data.Unique-import Prelude hiding (seq)-#ifdef CHP_TEST-import Test.HUnit hiding (test)-#endif--import Control.Concurrent.CHP.Poison-import Control.Concurrent.CHP.ProcessId---- | The type of an event in the CSP and VCR traces.------ ClockSync was added in version 1.2.0.------ The extra parameter on ChannelComm and BarrierSync (which are the result of--- showing the value sent and phase ended respectively) was added in version 1.5.0.-data RecordedEventType- = ChannelComm String- | BarrierSync String- | ClockSync String deriving (Eq, Ord, Show)--getEventTypeVal :: RecordedEventType -> String-getEventTypeVal (ChannelComm s) = s-getEventTypeVal (BarrierSync s) = s-getEventTypeVal (ClockSync s) = s---- Not really a CSP event, more like an enrollable poisonable alting barrier!-newtype Event = Event (- Unique, -- Event identifier- STM RecordedEventType, -- Event type for trace recording- TVar (WithPoison- (Int, -- Enrolled count- Integer, -- Event sequence count- [OfferSet]) -- A list of offer sets- ))--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)--getEventUnique :: Event -> Unique-getEventUnique (Event (u,_,_)) = u--getEventTVar :: Event -> TVar (WithPoison (Int, Integer, [OfferSet]))-getEventTVar (Event (_,_,tv)) = tv--getEventType :: Event -> STM RecordedEventType-getEventType (Event (_,t,_)) = t---- The value used to pass information to a waiting process once one of their events--- has fired (and they have been committed to it). The Int is an index into their--- list of guards-newtype SignalValue = Signal (WithPoison Int)- deriving (Eq, Show)--type SignalVar = TVar (Maybe (SignalValue, Map.Map Unique (Integer, RecordedEventType)))--addPoison :: SignalValue -> SignalValue-addPoison = const $ Signal PoisonItem--nullSignalValue :: SignalValue-nullSignalValue = Signal $ NoPoison (-1)--isNullSignal :: SignalValue -> Bool-isNullSignal (Signal n) = n == NoPoison (-1)--newtype OfferSet = OfferSet (SignalVar -- Variable to use to signal when committed- , ProcessId -- Id of the process making the offer- , [((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- -- 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 [] = Map.empty-unionAll ms = foldl1 Map.union ms--allEventsInOffer :: OfferSet -> Map.Map Event ()-allEventsInOffer (OfferSet (_, _, [(_,es)])) = es-allEventsInOffer (OfferSet (_, _, eventSets)) = unionAll (map snd eventSets)--getAndIncCounter :: Event -> (a, b) -> STM (WithPoison (Integer, a))-getAndIncCounter e (r, _)- = 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, r)---- | 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,--- it just searches the offer-sets (which will have been discovered through STM)--- for completions.------ search performs a 2-dimensional traversal of the offers. The search function--- is called with a list of offer-sets. For the offer-set at the head, it calls--- tryAll. tryAll searches through each offer in the offer-set, seeing if it can--- be completed. If it can, it calls search on the remaining offer-sets. If this--- fails, it reverts to trying the other offers in the list. The map of events passed through--- relates to the previous things found in the search.-search :: [OfferSet]- -- ^ The collection of all the related offer-sets- -> Map.Map Event Bool- -- ^ This contains the events already decided upon in the search. If- -- an event maps to True, it means it was chosen by an earlier part of- -- the search, and thus future parts of the search /must/ have this event- -- in the chosen offer (if the process offers it at all -- if it doesn't,- -- 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, STM ())]- , Map.Map Event (STM RecordedEventType, Set.Set ProcessId)- )- -- ^ The list of tvars involved with the completion and the signal- -- value for them, and the map with information about the completed events.-search [] _ = Just ([], 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- ]-- -- Folds across a map, seeing if the given predicate holds for all values- -- in the map.- mapdotall :: Ord k => (a -> Bool) -> Map.Map k a -> Bool- mapdotall f = Map.fold (\x b -> f x && b) True-- and' :: Ord k => Map.Map k Bool -> Bool- and' = mapdotall id-- tryAll :: [((SignalValue, STM ()), Map.Map Event ())]- -> Maybe ( [(SignalVar, SignalValue, STM ())]- , Map.Map Event (STM RecordedEventType, Set.Set ProcessId)- )- tryAll [] = Nothing- tryAll ((ns, es):next)- | not $ and' (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) -> Just- (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)- )- - 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)- --- 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 SignalVar -> [OfferSet] -> Set.Set Event- -> STM (Map.Map Unique (RecordedEventType, Set.Set ProcessId))-resolveOffers newTvid allOffers events- = do let (offers', _) = trim (allOffers, events)- (act, ret) = fromMaybe ([], Map.empty) $- search (map addNullOffer $ sortOffers offers') Map.empty- -- The associated event-action must come first as that puts the values in the channels:- mapM_ (\(_, _, m) -> m) act- -- These values are then read by these on-completion bits:- ret' <- T.mapM (\(m,y) -> do x <- m- return (x, y)) ret- 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- -- do the retractions for all involved processes once the choice is made:- -- TODO optimise:- 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, STM ()), Map.Map Event ())]- nullOffer = [((nullSignalValue, return ()) ,Map.empty)]---- Smallest offers first to minimise backtracking:-sortOffers :: [OfferSet] -> [OfferSet]-sortOffers xs- | length xs > 2 = sortBy (compare `on` (\(OfferSet (_,_,es)) -> length es)) xs- | otherwise = 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- 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) 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)- | 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:- eventsNotCompletable = Map.keysSet $ - (unionAll $ map snd eventSetsToRemove)- `Map.difference` (unionAll $ map snd eventSetsTrimmed)- changed' = changed- || not (null eventSetsToRemove)- 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--- 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))- -- If the offers only have one event, must be this- -- one:- (if Map.size (unionAll otherEvents) == 1- then next- else next ++ zip (repeat $ return ())- (Map.keys $ unionAll otherEvents))- else -- No way it could be ready, so ignore it:- discoverRelatedOffersAll a next---- Given 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- -- ^ Either an OfferSet to spider out from, or a single- -- event. The latter case is for when we are resigning- -- from an event and need to check if that completes anything.- -> STM (WithPoison (Map.Map Unique (RecordedEventType, Set.Set ProcessId)))- -- ^ Gives back either poison, or a map from event identifiers- -- to information about the completed event. The map is- -- empty if no events were completed.-discoverAndResolve offOrEvent- = do r <- discoverRelatedOffers $ case offOrEvent of- Left off@(OfferSet (tv, _, nes)) ->- let retract = retractOffers [(tv, allEventsInOffer off)] in- concat [zip- -- This is the action to execute if an event is found to- -- be poisoned:- (repeat $ retract >> writeTVar tv (Just (addPoison ns, Map.empty)))- (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 :: STM RecordedEventType -> Int -> IO Event-newEvent t n- = do u <- newUnique- atomically $ do tv <- newTVar (NoPoison (n, 0, []))- return $ Event (u, t, tv)--newEventUnique :: IO Unique-newEventUnique = newUnique--enrollEvent :: Event -> STM (WithPoison ())-enrollEvent e- = do x <- readTVar $ getEventTVar e- case x of- PoisonItem -> return PoisonItem- NoPoison (count, seq, offers) ->- do writeTVar (getEventTVar e) $ NoPoison (count + 1, seq, offers)- return $ NoPoison ()---- 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, seq, offers) ->- do writeTVar (getEventTVar e) $ NoPoison (count - 1, seq, offers)- if count - 1 == length offers- then liftM (fmap $ \mu -> [((r,u),pids) | (u,(r,pids)) <- Map.toList mu])- $ discoverAndResolve $ Right e- else return $ NoPoison []---- Given the list of identifiers paired with all the events that that process might--- be engaged in, retracts all the offers that are associated with the given TVar;--- i.e. the TVar is used as an identifier for the process-retractOffers :: [(SignalVar, Map.Map Event ())] -> STM ()-retractOffers = mapM_ retractAll- where- retractAll :: (SignalVar, Map.Map Event ()) -> STM ()- retractAll (tvid, evts) = mapM_ retract (Map.keys evts)- where- retract :: Event -> STM ()- retract e- = do x <- readTVar $ getEventTVar e- case x of- PoisonItem -> return ()- NoPoison (enrolled, seq, offers) ->- let reducedOffers = filter (\(OfferSet (tvx,_,_)) -> tvx /= tvid) offers in- writeTVar (getEventTVar e) $ NoPoison (enrolled, seq, reducedOffers)---- 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- -- No need for nub, as having it come from a map guarantees there are no- -- duplicates in the list of events- 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, seq, prevOffers) ->- do writeTVar (getEventTVar e) $ NoPoison (count, seq, offers : prevOffers)- return $ NoPoison ()---- Returns Nothing if no events were ready. Returns Just with the signal value--- if an event was immediately available, followed by the information for each--- event involved in the synchronisation. If poison was encounted, this list will--- be empty.-enableEvents :: SignalVar- -- ^ Variable used to signal the process once a choice is made- -> ProcessId- -- ^ The id of the process making the choice- -> [((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- -- more information.- -> Bool- -- ^ True if it can commit to waiting. If there is an event- -- combination ready during the transaction, it will chosen regardless- -- of the value of this flag. However, if there no events ready,- -- passing True will leave the offers there, but False will retract- -- the offers.- -> STM (Maybe ((SignalValue, Map.Map Unique (Integer, RecordedEventType)), [((RecordedEventType, Unique), Set.Set ProcessId)]))-enableEvents tvNotify pid events canCommitToWait- = do let offer = OfferSet (tvNotify, pid, [(nid, Map.fromList (zip es (repeat ()))) | (nid, es) <- events])- -- First add our offer to all the events:- -- We don't check the result for poison, as discoverAndResolve will find- -- it anyway- makeOffers offer- -- Then spider out and see if anything can be resolved:- 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- (_, 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])---- | Given the variable used to signal the process, and the list of events that--- were involved in its offers, attempts to disable the events. If the variable--- has been signalled (i.e. has a Just value), that is returned and nothing is done, if the variable--- has not been signalled (i.e. is Nothing), the events are disabled and Nothing--- is returned.-disableEvents :: SignalVar -> [Event] -> STM (Maybe (SignalValue, Map.Map Unique (Integer,- RecordedEventType)))-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, Map.fromList $ zip events (repeat ()))]- return x--checkEventForPoison :: Event -> STM (WithPoison ())-checkEventForPoison e- = do x <- readTVar $ getEventTVar e- case x of- PoisonItem -> return PoisonItem- _ -> return (NoPoison ())--poisonEvent :: Event -> STM ()-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 (addPoison $ pickInts events, Map.empty))- | OfferSet (tvw, _, events) <- offers]- writeTVar (getEventTVar e) PoisonItem- where- pickInts :: [((SignalValue, STM ()), Map.Map Event ())] -> SignalValue- pickInts es = case filter ((e `Map.member`) . snd) es of- [] -> nullSignalValue -- Should never happen- (((ns,_),_):_) -> ns----TODO document how if it's poisoned, 0 will be appended to the list-------------------------------------------------------------------------------------------------------------------------------------------------- Testing:-----------------------------------------------------------------------------------------------------------------------------------------------#ifdef CHP_TEST---- 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 :: 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)]- (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 | ((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 -> assertFailure $ testName ++ "Unexpected poison"- NoPoison actualResult -> do- when (fst expectedResult **/=** fst actualResult)- $ assertFailure $ testName ++ " failed offers, exp: "- ++ show (length $ fst expectedResult)- ++ " got: " ++ show (length $ fst actualResult)- when (snd expectedResult /= snd actualResult)- $ assertFailure $ 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 _ -> assertFailure $ testName ++ " expected poison but none"----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]- 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, _) -> assertFailure $ testName ++ " expected poison but none found"- (_, PoisonItem) -> assertFailure $ testName ++ " unexpected poison"- (NoPoison expectedResult, NoPoison actualResult)- -> do- when (fst expectedResult **/=** fst actualResult)- $ assertFailure $ testName ++ " failed offers, exp: "- ++ show (length $ fst expectedResult)- ++ " got: " ++ show (length $ fst actualResult)- when (snd expectedResult /= snd actualResult)- $ assertFailure $ testName ++ " failed events, exp: "- ++ show (snd expectedResult)- ++ "but got: " ++ show (snd actualResult)--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- 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) -> assertFailure $ testName ++- " expected no poison but found it"- (PoisonItem, NoPoison _) -> assertFailure $ testName ++- " expected poison but found none"- (NoPoison expOff, NoPoison (_, _, actOff)) ->- when (map (realOffers !!) expOff **/=** actOff) $- assertFailure $ testName ++ " offers did not match"- | (n, (_, expect)) <- zip [0..] eventCounts]--- -testAll :: Test-testAll = TestList [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 (return $ 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- -- TODO test the STM actions too- offSub = [ ((Signal $ NoPoison (processN + offerN), return ()),- 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, s, offs) ->- writeTVar (getEventTVar e) $ NoPoison (count, s, off : offs)- PoisonItem -> return ()- ) (Map.keys $ unionAll $ map snd offSub)- return off- | (processN, processOffers) <- zip (map (*1000) [0..]) offerSets]- return (events, realOffers)--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]]]- 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 ->- -- List of events:- [(WithPoison Int {- enrolled 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 <- liftM (liftM (fmap snd)) $ atomically $ discoverAndResolve $ Left $ head realOffers- let expectedResult = if poisoned then PoisonItem else NoPoison $- Map.fromList [ (getEventUnique e,- Set.fromList $ map (testProcessId . (*1000) . fst) is)- | (e, Left is) <- zip events (map snd eventCounts)]- when (expectedResult /= actualResult) $- 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 -> assertFailure $ "Unexpected no-win for " ++ show (pn,en)- Just v -> when (fst v /= (if poisoned then addPoison else id)- (Signal $ NoPoison ((pn*1000)+en))) $- assertFailure $ testName ++ " wrong choice: " ++ " 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 _ -> assertFailure $ testName ++ " Unexpected win for process: " ++- show n- | n <- [0 .. length offerSets - 1] \\ allFired]- -- check events are blanked afterwards:- 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') -> return $ Just ((count, expVal), (c, e'))- _ -> do assertFailure $ testName ++ " unexpected poison"- return Nothing-{- NoPoison (c, _, e') | c == count && e' == expVal -> return ()- _ ->- assertFailure $ 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]- uncurry (assertEqual testName) (unzip $ catMaybes c)-- showStuff = show . fmap (map (first hashUnique) . Map.toList)++-- Communicating Haskell Processes.+-- Copyright (c) 2008, University of Kent.+-- All rights reserved.+-- +-- Redistribution and use in source and binary forms, with or without+-- modification, are permitted provided that the following conditions are+-- met:+--+-- * Redistributions of source code must retain the above copyright+-- notice, this list of conditions and the following disclaimer.+-- * Redistributions in binary form must reproduce the above copyright+-- notice, this list of conditions and the following disclaimer in the+-- documentation and/or other materials provided with the distribution.+-- * Neither the name of the University of Kent nor the names of its+-- contributors may be used to endorse or promote products derived from+-- this software without specific prior written permission.+--+-- THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS+-- IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,+-- THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR+-- PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR+-- CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,+-- EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,+-- PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR+-- PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF+-- LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING+-- NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS+-- SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.++--TODO document this (for internal purposes)+module Control.Concurrent.CHP.Event (RecordedEventType(..), Event, getEventUnique,+ SignalVar, SignalValue(..), enableEvents, disableEvents,+ newEvent, newEventPri, newEventUnique, enrollEvent, resignEvent, poisonEvent, checkEventForPoison,+ getEventTypeVal+#ifdef CHP_TEST+ , testAll+#endif+ ) where++import Control.Applicative+import Control.Arrow+import Control.Concurrent+import Control.Concurrent.STM hiding (always)+import Control.Concurrent.CHP.EventType+import Control.Monad+import Control.Monad.Reader+#ifdef CHP_TEST+import Control.Monad.State+#endif+import Data.Function+import Data.List hiding (or)+import Data.Ord+import qualified Control.Concurrent.CHP.EventMap as EventMap (empty, toList, unionWith)+import qualified Control.Concurrent.CHP.EventSet as EventSet (deleteOrFail,+#ifdef CHP_TEST+ empty,+#endif+ fromList, member, toList, union)+import qualified Control.Concurrent.CHP.EventMap as OfferSetMap (insert, keysSet, minViewWithKey, unionWithM, values)+import qualified Control.Concurrent.CHP.EventSet as OfferSetSet (delete, insert, intersection, null, toMap)+--import qualified Data.IntMap as IntMap+import qualified Data.Map as Map+import Data.Maybe+import qualified Data.Set as Set+import qualified Data.Traversable as T+import Data.Unique+import Prelude hiding (cos, or, seq)+#ifdef CHP_TEST+import Test.HUnit hiding (test, State)+#endif++import Control.Concurrent.CHP.Poison+import Control.Concurrent.CHP.ProcessId++type DiscoverM = WithPoisonMaybeT STM++type OfferSetMap v = [(OfferSet, v)]+type OfferSetSet = [OfferSet]++data SearchState = SS+ { visited :: OfferSetMap SearchResult, notVisited :: OfferSetMap [TrimmedOffer] }++data CurOfferSet = New OfferSet | Old OfferSet | Resigning++data TrimmedOffer = TrimmedOffer { pristineOffer :: Offer, trimmedEvents :: EventSet }++addResult :: OfferSet -> SearchResult -> SearchState -> SearchState+addResult os r (SS v nv) = SS (OfferSetMap.insert os r v) nv++-- First parameter is Left for original, (Just offerSet) if existing, Just+-- Nothing if resigning+checkEvent :: CurOfferSet -> SearchState -> Event -> DiscoverM SearchState+checkEvent cos ss e = do+ s <- lift $ readTVar $ getEventTVar e+ case s of+ PoisonItem -> WPMT $ return PoisonItem+ NoPoison (enrollCount, _, offers) ->+ let numOffers = length offers+ in if numOffers >= enrollCount || (numOffers >= enrollCount - 1 && isDefinitelyNew)+ then WPMT . return . NoPoison $ addFilter (deleteCur offers) e ss+ else backtrack+ where+ isDefinitelyNew = case cos of+ New _ -> True+ _ -> False++ deleteCur = case cos of+ Old os -> OfferSetSet.delete os+ _ -> id+ -- No need to delete new offerset as it won't be there++--TODO could merge this with checkEvent++-- The current offer-set will have been removed from the list:+addFilter :: OfferSetSet -> Event -> SearchState -> Maybe SearchState+addFilter allos e ss+ | OfferSetSet.null allos = Just ss+ | not . OfferSetSet.null $ OfferSetSet.intersection allos (OfferSetMap.keysSet $ visited ss) = Nothing+ | otherwise = SS (visited ss) <$> OfferSetMap.unionWithM merge (notVisited ss) (OfferSetSet.toMap getOffers allos)+ where+ nullNothing [] = Nothing+ nullNothing xs = Just xs++ -- Will only return Just if at least one given trimmed offer contains the event we're+ -- interested in; and if so it will return only those offers that contained+ -- the event -- but with that event removed from the offers+ mustHaveThenStrike :: [TrimmedOffer] -> Maybe [TrimmedOffer]+ mustHaveThenStrike = nullNothing . mapMaybe (\(TrimmedOffer p es) -> TrimmedOffer p <$> EventSet.deleteOrFail e es)++ merge :: Maybe [TrimmedOffer] -> Maybe [TrimmedOffer] -> Maybe [TrimmedOffer]+ merge (Just t) Nothing = Just t -- Nothing new here+ merge Nothing (Just t) = mustHaveThenStrike t -- We are new; insert filtered+ merge (Just t) (Just _) = mustHaveThenStrike t -- Have old; filter that one+ merge Nothing Nothing = error "Event.merge"+ +getOffers :: OfferSet -> [TrimmedOffer]+getOffers os = [TrimmedOffer o (eventsSet o) | o <- offersSet os]++ -- 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++--eventSet :: EventMap a -> EventSet+--eventSet = EventMap.map (const ()) -- IntMap.fromList . map (\(e, _) -> (hashUnique $ getEventUnique e, e)) . EventMap.toList++eventMap :: (Event -> a) -> EventSet -> EventMap a+eventMap f = map (\e -> (e, f e))++allEventsInOffer :: OfferSet -> EventSet+allEventsInOffer = foldl1 EventSet.union . map eventsSet . offersSet++getAndIncCounter :: Event -> (a, b) -> STM (WithPoison (Integer, a))+getAndIncCounter e (r, _)+ = 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, r)+++-- Here is how it all works. There are events+-- Each event is:+-- * poisoned+-- * not poisoned and has:+-- * an enrollment count EC+-- * a list of offer-sets.+-- Each offer-set represents a process, and is:+-- * a list of conjunctions (offers), where each conjunction/offer is:+-- * an event-set (no duplicates allowed!)+--+-- All the event-handling code is called for one of two reasons:+-- * resignation from an event. This may:+-- * cause that single event to complete+-- * making an offer-set. This may:+-- * cause one of the conjunctions/offers to complete+--+-- So, we begin with either an event or an offer-set:+-- * Offer-set. You examine each offer in turn:+-- * You have a set of events, S. All must be able to complete. Check each event:+-- * If the event is poisoned, we stop with poison+-- * If the event has less than EC-1 offer-sets, this whole offer cannot complete+-- * If the event has EC-1 offer-sets, it can complete providing all its offer-sets+-- can. For each offer-set:+-- * Filter the offer-sets to those that include the parent event. Then+-- recurse to the top-level and see if those offer-sets can be completed,+-- with the proviso that any-time you encounter an event from S, it only needs+-- its EC-1 offer-sets; all other events require EC offers. Also, do not+-- count any offers in the current offer-set but not the current offer;+-- or put another way, do not consider any events containing this offer-set+-- not in this offer.+-- ++type SearchResult = ( [(SignalVar, SignalValue, STM ())]+ , EventMap (STM RecordedEventType, Set.Set ProcessId)+ )+combineSearch :: [SearchResult] -> SearchResult+combineSearch [] = ([], EventMap.empty)+combineSearch rs = foldl1 f rs+ where+ f (xs, xm) (ys, ym) = (xs ++ ys, xm `combineMap` ym)+ combineMap = EventMap.unionWith (\(x, y) (_, z) -> (x, y `Set.union` z))++data WithPoisonMaybeT m a = WPMT { runWPMT :: m (WithPoison (Maybe a)) }+instance Monad m => Monad (WithPoisonMaybeT m) where+ return = WPMT . return . NoPoison . Just+ m >>= f = WPMT $ do+ x <- runWPMT m+ case x of+ PoisonItem -> return PoisonItem+ NoPoison Nothing -> return $ NoPoison Nothing+ NoPoison (Just y) -> runWPMT $ f y++instance Monad m => Functor (WithPoisonMaybeT m) where+ fmap f = WPMT . liftM (fmap (fmap f)) . runWPMT++instance MonadTrans WithPoisonMaybeT where+ lift = WPMT . liftM (NoPoison . Just)++instance (Monad m) => Applicative (WithPoisonMaybeT m) where+ pure = return+ (<*>) = ap++instance (Monad m) => Alternative (WithPoisonMaybeT m) where+ empty = WPMT $ return $ NoPoison Nothing+ (<|>) a b = WPMT $ runWPMT a >>= \x -> case x of+ PoisonItem -> return PoisonItem+ NoPoison Nothing -> runWPMT b+ y -> return y++backtrack :: Alternative f => f a+backtrack = empty++search :: (Alternative f, Monad f) => [f a] -> f a+search [] = empty+search xs = foldl1 (<|>) xs++searchWith :: (Alternative f, Monad f) => (a -> f b) -> [a] -> f b+searchWith = (search .) . map++searchOfferSet :: CurOfferSet -> [TrimmedOffer] -> SearchState -> DiscoverM SearchResult+searchOfferSet cos offers ss+ = searchWith searchOffer offers+ where+ searchOffer offer+ = do ss' <- foldM (checkEvent cos) ss (trimmedEvents offer)+ processNext $ case cos of+ New os -> addResult os ([(signalVar os, signalValue o, offerAction o)],+ eventMap (\e -> (getEventType e, Set.singleton $ processId os)) $ eventsSet o) ss'+ Resigning -> ss+ Old os -> addResult os ([(signalVar os, signalValue o, offerAction o >> retractOfferSet os)]+ , eventMap (\e -> (getEventType e, Set.singleton $ processId os)) (eventsSet o)) ss'+ where+ o = pristineOffer offer++searchOriginalOfferSet :: OfferSet -> DiscoverM SearchResult+searchOriginalOfferSet os = searchOfferSet (New os) (sortBy (flip $ comparing (getEventPriority . head . trimmedEvents)) $ getOffers os) (SS [] [])++processNext :: SearchState -> DiscoverM SearchResult+processNext s = case OfferSetMap.minViewWithKey (notVisited s) of+ -- All visited:+ Nothing -> return $ combineSearch (OfferSetMap.values $ visited s)+ -- At least one left:+ Just ((os, next), rest) -> searchOfferSet (Old os) next (s { notVisited = rest })+ +-- 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.+discoverAndResolve :: Either OfferSet Event+ -> STM (WithPoison (Map.Map Unique (RecordedEventType, Set.Set ProcessId)))+discoverAndResolve start = do+ r <- runWPMT $ either searchOriginalOfferSet+ (processNext <=< checkEvent Resigning (SS [] []))+ start+ case r of+ PoisonItem -> do either (flip writeTVar (Just (Signal PoisonItem, Map.empty)) . signalVar)+ (const $ return ()) start+ return PoisonItem+ NoPoison Nothing ->+ -- Must record our offers+ (const Map.empty <$>) <$> case start of+ Left offerSet -> makeAllOffers offerSet+ Right _ -> return $ NoPoison ()+ NoPoison (Just (actPossDup, ret)) ->+ do let act = nubBy ((==) `on` (\(var, _, _) -> var)) actPossDup+ -- The associated event-action must come first as that puts the values in the channels:+ mapM_ (\(_, _, m) -> m) act+ -- These values are then read by these on-completion bits:+ ret' <- mapM (\(k, (em, y)) -> do x <- em+ return (k, (x, y))) $ EventMap.toList ret+ NoPoison eventCounts <- liftM T.sequence . T.sequence $ map (\(k, v) -> liftM+ ((,) k) <$> getAndIncCounter k v) ret'+ let uniqCounts = Map.fromList $ map (first getEventUnique) eventCounts+ mapM_ (\(tv, x, _) -> writeTVar tv (Just (x, uniqCounts))) act++ return $ NoPoison (Map.fromAscList $ map (first getEventUnique) ret')+++-- 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.++newEventUnique :: IO Unique+newEventUnique = newUnique++enrollEvent :: Event -> STM (WithPoison ())+enrollEvent e+ = do x <- readTVar $ getEventTVar e+ case x of+ PoisonItem -> return PoisonItem+ NoPoison (count, seq, offers) ->+ do writeTVar (getEventTVar e) $ NoPoison (count + 1, seq, offers)+ return $ NoPoison ()++-- 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, seq, offers) ->+ do writeTVar (getEventTVar e) $ NoPoison (count - 1, seq, offers)+ if count - 1 == length offers+ then liftM (fmap $ \mu -> [((r,u),pids) | (u,(r,pids)) <- Map.toList mu])+ $ discoverAndResolve $ Right e+ else return $ NoPoison []++-- Given the list of identifiers paired with all the events that that process might+-- be engaged in, retracts all the offers that are associated with the given TVar;+-- i.e. the TVar is used as an identifier for the process+retractOffers :: [(OfferSet, EventSet)] -> STM ()+retractOffers = mapM_ retractAll+ where+ retractAll :: (OfferSet, EventSet) -> STM ()+ retractAll (offerSet, evts) = mapM_ retract (EventSet.toList evts)+ where+ retract :: Event -> STM ()+ retract e+ = do x <- readTVar $ getEventTVar e+ case x of+ PoisonItem -> return ()+ NoPoison (enrolled, seq, offers) ->+ let reducedOffers = OfferSetSet.delete offerSet offers in+ writeTVar (getEventTVar e) $ NoPoison (enrolled, seq, reducedOffers)++retractOfferSet :: OfferSet -> STM ()+retractOfferSet = retractOffers . (:[]) . (id &&& allEventsInOffer)+ ++-- Simply adds the offers but doesn't check if that will complete an event:+-- Returns PoisonItem if any of the events were poisoned+makeOffer :: OfferSet -> (Event -> STM (WithPoison ()))+makeOffer offers = makeOffer'+ where+ makeOffer' :: Event -> STM (WithPoison ())+ makeOffer' e+ = do x <- readTVar $ getEventTVar e+ case x of+ PoisonItem -> return PoisonItem+ NoPoison (count, seq, prevOffers) ->+ do writeTVar (getEventTVar e) $ NoPoison (count, seq, OfferSetSet.insert offers prevOffers)+ return $ NoPoison ()++makeAllOffers :: OfferSet -> STM (WithPoison ())+makeAllOffers offerSet+ = sequence_ <$> mapM (makeOffer offerSet) (EventSet.toList $ allEventsInOffer offerSet)++-- Returns Nothing if no events were ready. Returns Just with the signal value+-- if an event was immediately available, followed by the information for each+-- event involved in the synchronisation. If poison was encounted, this list will+-- be empty.+enableEvents :: SignalVar+ -- ^ Variable used to signal the process once a choice is made+ -> (ThreadId, ProcessId)+ -- ^ The id of the process making the choice+ -> [((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+ -- more information.+ -> Bool+ -- ^ True if it can commit to waiting. If there is an event+ -- combination ready during the transaction, it will chosen regardless+ -- of the value of this flag. However, if there no events ready,+ -- passing True will leave the offers there, but False will retract+ -- the offers.+ -> STM (Either+ (STM (Maybe (SignalValue, Map.Map Unique (Integer, RecordedEventType))))+ ((SignalValue, Map.Map Unique (Integer, RecordedEventType)), [((RecordedEventType, Unique), Set.Set ProcessId)])+ )+enableEvents tvNotify (tid, pid) events canCommitToWait+ = do let offer = makeOfferSet tvNotify pid tid [(nid, EventSet.fromList es) | (nid, es) <- events]+ -- Then spider out and see if anything can be resolved:+ pmu <- discoverAndResolve (Left offer)+ case (canCommitToWait, pmu) of+ (_, PoisonItem) -> return $ Right ((Signal PoisonItem, Map.empty), [])+ (True, NoPoison mu) | Map.null mu -> return $ Left $ disableEvents offer (concatMap snd events)+ (False, NoPoison mu) | Map.null mu ->+ do retractOffers [(offer, EventSet.fromList $ concatMap snd events)]+ return $ Left $ error "enableEvents"+ (_, 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 chosenItem <- readTVar tvNotify+ return $ Right (chosenItem, [((r,u),pids) | (u,(r,pids)) <- Map.toList mu])++-- | Given the variable used to signal the process, and the list of events that+-- were involved in its offers, attempts to disable the events. If the variable+-- has been signalled (i.e. has a Just value), that is returned and nothing is done, if the variable+-- has not been signalled (i.e. is Nothing), the events are disabled and Nothing+-- is returned.+disableEvents :: OfferSet -> [Event] -> STM (Maybe (SignalValue, Map.Map Unique (Integer,+ RecordedEventType)))+disableEvents offer events+ = do x <- readTVar $ signalVar offer+ -- Since the transaction will be atomic, we know+ -- now that we can disable the barriers and nothing fired:+ when (isNothing x) $+ retractOffers [(offer, EventSet.fromList events)]+ return x++checkEventForPoison :: Event -> STM (WithPoison ())+checkEventForPoison e+ = do x <- readTVar $ getEventTVar e+ case x of+ PoisonItem -> return PoisonItem+ _ -> return (NoPoison ())++poisonEvent :: Event -> STM ()+poisonEvent e+ = do x <- readTVar $ getEventTVar e+ case x of+ PoisonItem -> return ()+ NoPoison (_, _, offers) ->+ do retractOffers $ map (id &&& allEventsInOffer) offers+ sequence_ [writeTVar (signalVar o) (Just (addPoison $ pickInts (offersSet o), Map.empty))+ | o <- offers]+ writeTVar (getEventTVar e) PoisonItem+ where+ pickInts :: [Offer] -> SignalValue+ pickInts es = case filter ((e `EventSet.member`) . eventsSet) es of+ [] -> nullSignalValue -- Should never happen+ (o:_) -> signalValue o++----------------------------------------------------------------------+----------------------------------------------------------------------+-- Testing:+----------------------------------------------------------------------+----------------------------------------------------------------------+#ifdef CHP_TEST++unionAll :: [EventSet] -> EventSet+unionAll [] = EventSet.empty+unionAll ms = foldl1 EventSet.union ms+++-- 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++testPoison :: Test+testPoison = TestCase $ do+ test "Poison empty event" [(NoPoison $ EventInfo 2 0, PoisonItem)] [] 0+ test "Poison, single offerer" [(NoPoison $ EventInfo 2 0, PoisonItem)] [[[0]]] 0+ test "Poison, offered on two (AND)" [(NoPoison $ EventInfo 2 0, PoisonItem), (NoPoison $ EventInfo 2 0, NoPoison [])] [[[0,1]]] 0+ test "Poison, offered on two (OR)" [(NoPoison $ EventInfo 2 0, PoisonItem), (NoPoison $ EventInfo 2 0, NoPoison [])] [[[0],[1]]] 0+ where+ test :: String ->+ [(WithPoison EventInfo {-count -}, WithPoison [Int] {- remaining offers -})] ->+ {- Offers: -} [[[Int] {- events -}]] -> Int {-Poison Event-} -> IO ()++ test testName eventCounts offerSets poisoned = do+ (events, realOffers) <- makeTestEvents (map fst eventCounts) $+ map (map (flip (,) (return ()))) 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) -> assertFailure $ testName +++ " expected no poison but found it"+ (PoisonItem, NoPoison _) -> assertFailure $ testName +++ " expected poison but found none"+ (NoPoison expOff, NoPoison (_, _, actOff)) ->+ when (map (realOffers !!) expOff **/=** actOff) $+ assertFailure $ testName ++ " offers did not match"+ | (n, (_, expect)) <- zip [0..] eventCounts]+++ +testAll :: Test+testAll = TestList [{-testDiscover, testTrim, -}testResolve, testPoison]++makeTestEvents ::+ {- Events: -} [WithPoison EventInfo {-count -}] ->+ {- Offers: -} [[([Int] {- events -}, STM ())]] -> 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 (\x -> uncurry (newEventPri (return $ ChannelComm "")) $ case x of+ NoPoison (EventInfo n pri) -> (n, pri)+ PoisonItem -> (0, 0)) eventCounts+ -- Poison all the events marked as poisoned:+ atomically $ sequence_ [writeTVar (getEventTVar e) PoisonItem | (n, e) <- zip [0..] events, eventCounts !! n == PoisonItem]+ -- Nasty, but it's only for testing:+ tids <- replicateM (length offerSets) $ forkIO (threadDelay 1000000)+ realOffers <- sequence+ [ do tv <- atomically $ newTVar Nothing+ let pid = testProcessId processN+ -- TODO test the STM actions too+ offSub = [ ((Signal $ NoPoison (processN + offerN), act),+ EventSet.fromList (map (events !!) singleOffer))+ | (offerN, (singleOffer, act)) <- zip [0..] processOffers]+ off = makeOfferSet tv pid tid offSub+ when (processN /= 1000 * (length offerSets - 1)) $ mapM_ (\e -> atomically $ do+ x <- readTVar (getEventTVar e)+ case x of+ NoPoison (count, s, offs) ->+ writeTVar (getEventTVar e) $ NoPoison (count, s, OfferSetSet.insert off offs)+ PoisonItem -> return ()+ ) (EventSet.toList $ unionAll $ map snd offSub)+ return off+ | (tid, processN, processOffers) <- zip3 tids (map (*1000) [0..]) offerSets]+ return (events, realOffers)++data EventInfo = EventInfo {eventEnrolled :: Int, eventPriority :: Int}+ deriving (Eq, Show)++type CProcess = [CEvent] -- The list of conjunctions+newtype EventDSL a = EventDSL (State ([EventInfo], [CProcess]) a)+ deriving (Monad)++data ProcOrders = ProcOrders { procFinals :: [COffer]+ , procAll :: [COffer]+ }++runDSL :: EventDSL (ProcOrders, Outcome) ->+ [(([WithPoison EventInfo {- enrolled count -}],+ [[ Either [(Int, Int)] {- success: expected process, offer indexes -}+ [Int] {- remaining offers -}+ ]])+ ,{- Offers: -} [[[Int] {- events -}]])]+runDSL (EventDSL m)+ = let ((procOrders, Many outcomes), (evts, ps)) = runState m ([], [])+ orderings = [(h, procAll procOrders \\ [h]) | h <- procFinals procOrders]+ in+ [let conv p+ | p == cOffer new = length already+ | p < cOffer new = p+ | otherwise = p - 1+ in ((map NoPoison evts+ ,[let completing = nub $ concatMap cEvent [(ps !! p) !! i | (p, i) <- o]+ completers e = [(conv p, i) | (p, i) <- o, e `elem` cEvent ((ps !! p) !! i)]+ allCompleters = nub $ concatMap (map fst . completers) is+ is = [0..(length evts - 1)]+ in+ [if i `elem` completing+ then Left $ completers i+ else Right [conv j | (j, p) <- zip [0..] ps+ , conv j `notElem` allCompleters+ , i `elem` concatMap cEvent p]+ | i <- is ]+ | o <- outcomes]+ )+ , map (map cEvent . (ps !!) . cOffer) already ++ [map cEvent $ ps !! cOffer new] -- TODO iron this out later on+ )+ | (new, already) <- orderings]++evt :: Int -> EventDSL CEvent+evt n = evtNPri n 0++evtNPri :: Int -> Int -> EventDSL CEvent+evtNPri n pri = EventDSL $ do (evts, x) <- get+ put (evts ++ [EventInfo n pri], x)+ return $ CEvent [length evts]++newtype CEvent = CEvent {cEvent :: [Int]}+newtype COffer = COffer {cOffer :: Int}+ deriving Eq++offer :: [CEvent] -> EventDSL COffer+offer o = EventDSL $+ do (x, ps) <- get+ put (x, ps ++ [o])+ return $ COffer (length ps)++class Andable c where+ (&) :: c -> c -> c++instance Andable CEvent where+ (&) (CEvent a) (CEvent b) = CEvent (a ++ b)++-- Many is (process index, offer index)+data Outcome = Many [[(Int, Int)]]++(~>) :: COffer -> Int -> Outcome+(~>) (COffer p) i = Many [[(p, i)]]++instance Andable Outcome where+ (&) (Many [a]) (Many [b]) = Many [a++b]++(==>) :: [COffer] -> Outcome -> EventDSL (ProcOrders, Outcome)+(==>) finals o = EventDSL $ do+ (_, ps) <- get+ let allProcs = map COffer [0..(length ps - 1)]+ if null finals+ then return (ProcOrders allProcs allProcs, o)+ else return (ProcOrders finals allProcs, o)++none :: Outcome+none = Many [[]]++or :: Outcome -> Outcome -> Outcome+or (Many a) (Many b) = Many (a ++ b)++infix 0 ==>+infix 2 ~>+infixl 1 &++always = ([] ==>)++testResolve :: Test+testResolve = TestList $+ [ testD "Single offer on single event" $ do+ a <- evt 1+ p <- offer [a]+ always$ p ~> 0+ , testD "Not enough; one offer on two-party event" $ do+ a <- evt 2+ p <- offer [a]+ always$ none+ , testD "Not enough; two offers on three-party event" $ do+ a <- evt 3+ p <- offer [a]+ q <- offer [a]+ always$ none+ , testD "One channel, standard communication" $ do+ a <- evt 2+ p <- offer [a]+ q <- offer [a]+ always$ p ~> 0 & q ~> 0+ , testD "Two channels, two single offerers and one double" $ do+ a <- evt 2+ b <- evt 2+ p <- offer [a&b]+ q <- offer [a]+ r <- offer [b]+ always$ p ~> 0 & q ~> 0 & r ~> 0+ , testD "Two channels, two single offerers and one choosing" $ do+ a <- evt 2+ b <- evt 2+ p <- offer [a, b]+ q <- offer [a]+ r <- offer [b]+ [p] ==> (p ~> 0 & q ~> 0) `or` (p ~> 1 & r ~> 0)+ , testD "Two channels, both could complete" $ do+ [a, b] <- replicateM 2 $ evt 2+ [p, q] <- replicateM 2 $ offer [a, b]+ always$ (p ~> 0 & q ~> 0) `or` (p ~> 1 & q ~> 1)+ , testD "Two channels, both could complete, one pri" $ do+ [a, b] <- mapM (evtNPri 2) [0, 1]+ [p, q] <- sequence [offer [a, b], offer [b, a]]+ always$ (p ~> 1 & q ~> 0)+ , testD "Three channels, two could complete" $ do+ [a, b, c] <- replicateM 3 $ evt 2+ p <- offer [a, b, c]+ q <- offer [a]+ r <- offer [c]+ [p] ==> (p ~> 0 & q ~> 0) `or` (p ~> 2 & r ~> 0)+ , testD "Three channels, any could complete" $ do+ [a, b, c] <- replicateM 3 $ evt 2+ p <- offer [a, b, c]+ q <- offer [a]+ r <- offer [b]+ s <- offer [c]+ [p] ==> (p ~> 0 & q ~> 0) `or` (p ~> 1 & r ~> 0) `or` (p ~> 2 & s ~> 0)+ , testD "Three channels, both offering different overlapping pair" $ do+ [a, b, c] <- replicateM 3 $ evt 2+ p <- offer [a, b]+ q <- offer [b, c]+ always$ p ~> 1 & q ~> 0+ , testD "Three channels, one guy offering three pairs, two single offerers" $ do+ [a, b, c] <- replicateM 3 $ evt 2+ p <- offer [a&b, a&c, b&c]+ q <- offer [a]+ r <- offer [c]+ always$ p ~> 1 & q ~> 0 & r ~> 0+ , testD "Three channels, one guy offering three pairs, three single offerers" $ do+ [a, b, c] <- replicateM 3 $ evt 2+ p <- offer [a&b, b&c, a&c]+ q <- offer [a]+ r <- offer [b]+ s <- offer [c]+ [p] ==> (p ~> 0 & q ~> 0 & r ~> 0)+ `or` (p ~> 1 & r ~> 0 & s ~> 0)+ `or` (p ~> 2 & q ~> 0 & s ~> 0)+ , testD "Four channels, one guy offering sets of three, three single offerers" $ do+ [a, b, c,d ] <- replicateM 4 $ evt 2+ p <- offer [a&b&c, a&b&d, a&b&c, b&c&d]+ q <- offer [b]+ r <- offer [c]+ s <- offer [d]+ always$ p ~> 3 & q ~> 0 & r ~> 0 & s ~> 0+ , testD "Four channels, one guy offering sets of three, two single offerers" $ do+ [a, b, c,d ] <- replicateM 4 $ evt 2+ p <- offer [a&b&c, a&b&d, a&b&c, b&c&d]+ q <- offer [b]+ r <- offer [c]+ always$ none+ , testD "Four channels, one guy offering sets of three, one single offerer and one double" $ do+ [a, b, c,d ] <- replicateM 4 $ evt 2+ p <- offer [a&b&c, a&b&d, a&b&c, b&c&d]+ q <- offer [b&c]+ r <- offer [d]+ always$ p ~> 3 & q ~> 0 & r ~> 0+ , testD "Four channels, one guy offering sets of three, one single offerer and one on two" $ do+ [a, b, c,d ] <- replicateM 4 $ evt 2+ p <- offer [a&b&c, a&b&d, a&b&c, b&c&d]+ q <- offer [b, c]+ r <- offer [d]+ always$ none+ , testD "Links 1" $ do+ [a, b, c, d] <- replicateM 4 $ evt 2+ p <- offer [a&b]+ q <- offer [b&c&d]+ r <- offer [c, d]+ always$ none+ , testD "Links 2" $ do+ [a, b, c, d, e] <- replicateM 5 $ evt 2+ p <- offer [b]+ q <- offer [b&c&d&e]+ r <- offer [c, d]+ s <- offer [e]+ always$ none+ , testD "Links 3" $ do+ [a, b, c, d, e] <- replicateM 5 $ evt 2+ p <- offer [b]+ q <- offer [b&c&d&e]+ r <- offer [c&a, d&a]+ s <- offer [e]+ t <- offer [a]+ always$ none+ , testD "Ring 1" $ do+ [a, b, c, d] <- replicateM 4 $ evt 2+ p <- offer [a&b]+ q <- offer [b&c]+ r <- offer [c&d]+ s <- offer [d&a]+ always$ foldl1 (&) $ map (~> 0) [p, q, r, s]+ , testD "Ring 2" $ do+ [a, b, c, d] <- replicateM 4 $ evt 2+ p <- offer [a&b]+ q <- offer [b&c]+ r <- offer [c,d]+ s <- offer [d&a]+ always$ none+ , testD "Ring 3" $ do+ [a, a', b, c, d] <- replicateM 5 $ evt 2+ p <- offer [a&b, a']+ q <- offer [b&c]+ r <- offer [c&d]+ s <- offer [d&a']+ always$ none+ , testD "Pipeline 1" $ do+ [a,b,c,d,e,f] <- replicateM 6 $ evt 2+ p <- offer [a, b]+ q <- offer [a & c, b & c, b & d]+ r <- offer [d & e, d & f, c & e]+ s <- offer [f]+ always$ p ~> 1 & q ~> 2 & r ~> 1 & s ~> 0++ -- test resolutions with poison:+ --+ , test' "One event, poisoned" True+ ([PoisonItem], [[Left [(0,0)]]])+ [[[0]]]+ , test' "Two events, one poisoned" True+ ([PoisonItem, NoPoison $ EventInfo 2 0], [[Left [(0,0)], Left [(0,0)]]])+ [[[0,1]]]+ ]+ where+ testD testName = TestList . map (uncurry (test' testName False)) . runDSL++ test testName eventCounts offerSets = test' testName False (second (:[]) $+ unzip eventCounts) offerSets+ + test' :: String -> Bool {-Poisoned-} -> + -- List of events:+ ([WithPoison EventInfo] {- enrolled count -}+ ,[[Either [(Int, Int)] {- success: expected process, offer indexes -}+ [Int] {- remaining offers -}+ ] {- a single possibility, as long as the list of enroll counts -}+ ] {- the list of possibilities -}) ->+ {- Offers: -} [[[Int] {- events -}]] -> Test++ test' testName poisoned eventCounts offerSets = TestLabel testName $ TestCase $ do+ tv <- atomically $ newTVar Map.empty+ let add x = readTVar tv >>= (writeTVar tv . Map.insertWith (+) x 1)+ offerSets' = [ [ (offer, add (i, j))+ | offer <- offerSet | j <- [0..]]+ | offerSet <- offerSets | i <- [0..]]+ (events, realOffers) <- makeTestEvents (fst eventCounts) offerSets'++ actualResult <- liftM (liftM (fmap snd)) $ atomically $ discoverAndResolve $ Left $ last realOffers++ actionResult <- atomically $ readTVar tv++ let combinedActual = (,) actionResult <$> actualResult++ let expectedResults = if poisoned then [PoisonItem] else map NoPoison $+ [(Map.fromList $ zip (nub $ concat [x | Left x <- poss]) (repeat 1)+ ,Map.fromList [ (getEventUnique e,+ Set.fromList $ map (testProcessId . (*1000) . fst) is)+ | (e, Left is) <- zip events poss]+ )+ | poss <- snd eventCounts]+ when (combinedActual `notElem` expectedResults) $+ assertFailure $ testName ++ " failed on direct result/actions, expected one of: ["+ ++ intercalate "," (map showStuff expectedResults) ++ "] got: " ++ showStuff combinedActual+ ++ " (params: " ++ show offerSets ++ ")"++ vals <- mapM (atomically . readTVar . signalVar) realOffers+ let+ expAct = [+ [(unzip [(fst <$> (vals !! pn)+ ,Just $ (if poisoned then addPoison else id)+ (Signal $ NoPoison ((pn*1000)+en)))+ | (pn, en) <- exp]+ , map fst exp)+ | Left exp <- poss]+ | poss <- snd eventCounts]+ (poss, allFired) <- case findIndex (all (uncurry (==) . fst)) expAct of+ Nothing -> do assertFailure $ testName ++ "No possible firing outcomes matched"+ return $ error $ testName ++ "No possible firing outcomes matched"+ Just n -> return (snd eventCounts !! n, concatMap snd (expAct !! n))++ -- test the others are unchanged+ sequence_ [ let tv = signalVar $ realOffers !! n in+ do x <- atomically $ readTVar tv+ case x of+ Nothing -> return ()+ Just _ -> assertFailure $ testName ++ " Unexpected win for process: " +++ show n+ | n <- [0 .. length offerSets - 1] \\ allFired]+ -- check events are blanked afterwards:+ 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') -> return $ Just ((count, sort expVal), (EventInfo c (getEventPriority e), sort e'))+ _ -> do assertFailure $ testName ++ " unexpected poison"+ return Nothing+{- NoPoison (c, _, e') | c == count && e' == expVal -> return ()+ _ ->+ assertFailure $ 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) <- zip3 [0..] (fst eventCounts) poss]+ uncurry (assertEqual (testName ++ " not blanked " ++ show eventCounts+ ++ show offerSets)) (unzip $ catMaybes c)++ showStuff :: WithPoison (Map.Map (Int, Int) Int, Map.Map Unique (Set.Set ProcessId)) -> String+ showStuff = show . fmap (Map.toList *** (map (first hashUnique) . Map.toList)) #endif -- CHP_TEST
+ Control/Concurrent/CHP/EventMap.hs view
@@ -0,0 +1,100 @@+-- Communicating Haskell Processes.+-- Copyright (c) 2010, Neil Brown.+-- 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.++module Control.Concurrent.CHP.EventMap (empty, insert, keysSet, minViewWithKey, toList, unionWith, unionWithM, values) where++import Control.Concurrent.CHP.EventType+import Control.Monad (liftM)+import Data.Ord (comparing)+import Prelude hiding (lookup)++{-+instance Functor EventMap where+ fmap f = List.map (second f)++instance Foldable EventMap where+ foldr f x = foldr (f . snd) x++instance Traversable EventMap where+ traverse f = traverse (\(e, v) -> (,) e <$> f v)+-}++type ListMap k v = [(k, v)]++empty :: ListMap k v+empty = []++unionWith :: Ord k => (a -> a -> a) -> ListMap k a -> ListMap k a -> ListMap k a+{-# SPECIALISE unionWith :: (a -> a -> a) -> ListMap Event a -> ListMap Event a -> ListMap Event a #-}+unionWith f = union'+ where+ union' [] ys = ys+ union' xs [] = xs+ union' allxs@(x:xs) allys@(y:ys) = case comparing fst x y of+ LT -> x : union' xs allys+ EQ -> (fst x, f (snd x) (snd y)) : union' xs ys+ GT -> y : union' allxs ys++unionWithM :: (Ord k, Monad m) => (Maybe a -> Maybe b -> m c) ->+ ListMap k a -> ListMap k b -> m (ListMap k c)+{-# SPECIALISE unionWithM :: (Maybe a -> Maybe b -> Maybe c) ->+ ListMap OfferSet a -> ListMap OfferSet b -> Maybe (ListMap OfferSet c) #-}+unionWithM f = (sequence .) . union'+ where+ mapSM g = map (\(x, y) -> (,) x `liftM` g y)++ union' [] ys = mapSM (f Nothing . Just) ys+ union' xs [] = mapSM (flip f Nothing . Just) xs+ union' allxs@(x:xs) allys@(y:ys) = case compare (fst x) (fst y) of+ LT -> (,) (fst x) `liftM` f (Just $ snd x) Nothing : union' xs allys+ EQ -> ((,) (fst x) `liftM` f (Just $ snd x) (Just $ snd y)) : union' xs ys+ GT -> (,) (fst y) `liftM` f Nothing (Just $ snd y) : union' allxs ys++keysSet :: ListMap k a -> [k]+keysSet = map fst++toList :: ListMap k a -> [(k, a)]+toList = id++insert :: Ord k => k -> a -> ListMap k a -> ListMap k a+{-# SPECIALISE insert :: OfferSet -> a -> ListMap OfferSet a -> ListMap OfferSet a #-}+insert k v = insert'+ where+ insert' [] = [(k, v)]+ insert' allxs@(x:xs) = case compare k (fst x) of+ LT -> (k, v) : allxs+ EQ -> (k, v) : xs+ GT -> x : insert' xs++minViewWithKey :: ListMap k v -> Maybe ((k, v), ListMap k v)+minViewWithKey [] = Nothing+minViewWithKey (x:xs) = Just (x, xs)++values :: ListMap k v -> [v]+values = map snd
+ Control/Concurrent/CHP/EventSet.hs view
@@ -0,0 +1,113 @@+-- Communicating Haskell Processes.+-- Copyright (c) 2010, Neil Brown.+-- 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.++module Control.Concurrent.CHP.EventSet (delete, deleteOrFail, empty, fromList, insert, intersection, member, null, toList, toMap, union) where++import Control.Arrow ((&&&))+import Control.Concurrent.CHP.EventType+import qualified Data.List as List++type ListSet k = [k]++empty :: ListSet k+empty = []++toList :: ListSet k -> [k]+toList x = x++delete :: Ord k => k -> ListSet k -> ListSet k+{-# SPECIALISE delete :: Event -> ListSet Event -> ListSet Event #-}+{-# SPECIALISE delete :: OfferSet -> ListSet OfferSet -> ListSet OfferSet #-}+delete e = delete'+ where+ delete' [] = []+ delete' allxs@(x:xs) = case compare e x of+ LT -> allxs+ EQ -> xs+ GT -> x : delete' xs++-- If the element is present, returns Just the set without it+-- If the element is not present, returns Nothing+deleteOrFail :: Ord k => k -> ListSet k -> Maybe (ListSet k)+{-# SPECIALISE deleteOrFail :: Event -> ListSet Event -> Maybe (ListSet Event) #-}+deleteOrFail e = deleteOrFail'+ where+ deleteOrFail' [] = Nothing+ deleteOrFail' (x:xs) = case compare e x of+ LT -> Nothing+ EQ -> Just xs+ GT -> case deleteOrFail' xs of+ Just xs' -> Just (x : xs')+ Nothing -> Nothing++member :: Ord k => k -> ListSet k -> Bool+{-# SPECIALISE member :: Event -> ListSet Event -> Bool #-}+member e = member'+ where+ member' [] = False+ member' (x:xs) = case compare e x of+ LT -> False+ EQ -> True+ GT -> member' xs++insert :: Ord k => k -> ListSet k -> ListSet k+{-# SPECIALISE insert :: OfferSet -> ListSet OfferSet -> ListSet OfferSet #-}+insert k = insert'+ where+ insert' [] = [k]+ insert' allxs@(x:xs) = case compare k x of+ LT -> k : allxs+ EQ -> k : xs -- replace with new value+ GT -> x : insert' xs+++union :: Ord k => ListSet k -> ListSet k -> ListSet k+{-# SPECIALISE union :: ListSet Event -> ListSet Event -> ListSet Event #-}+union [] ys = ys+union xs [] = xs+union allxs@(x:xs) allys@(y:ys) = case compare x y of+ LT -> x : union xs allys+ EQ -> x : union xs ys -- left-bias+ GT -> y : union allxs ys++intersection :: Ord k => ListSet k -> ListSet k -> ListSet k+{-# SPECIALISE intersection :: ListSet OfferSet -> ListSet OfferSet -> ListSet OfferSet #-}+intersection [] _ = []+intersection _ [] = []+intersection allxs@(x:xs) allys@(y:ys) = case compare x y of+ LT -> intersection xs allys+ EQ -> x : intersection xs ys -- left-bias+ GT -> intersection allxs ys++fromList :: Ord k => [k] -> ListSet k+{-# SPECIALISE fromList :: [Event] -> ListSet Event #-}+fromList = List.sort++toMap :: (k -> v) -> [k] -> [(k, v)]+toMap f = map (id &&& f)
+ Control/Concurrent/CHP/EventType.hs view
@@ -0,0 +1,136 @@+-- Communicating Haskell Processes.+-- Copyright (c) 2010, University of Kent, Neil Brown.+-- 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.++module Control.Concurrent.CHP.EventType (+ Event, EventMap, EventSet, getEventTVar, getEventType, getEventTypeVal, getEventUnique, getEventPriority, newEvent, newEventPri,+ Offer(signalValue, offerAction, eventsSet),+ OfferSet(signalVar, offersSet, processId), makeOfferSet,+ RecordedEventType(..),+ SignalVar, SignalValue(..), addPoison, nullSignalValue, isNullSignal+ ) where++import Control.Arrow+import Data.Function (on)+import qualified Data.Map as Map+import Data.Unique+import Control.Concurrent+import Control.Concurrent.STM+import Control.Concurrent.CHP.Poison+import Control.Concurrent.CHP.ProcessId++type EventMap v = [(Event, v)]+type EventSet = [Event]+type OfferSetSet = [OfferSet]++-- | The type of an event in the CSP and VCR traces.+--+-- ClockSync was added in version 1.2.0.+--+-- The extra parameter on ChannelComm and BarrierSync (which are the result of+-- showing the value sent and phase ended respectively) was added in version 1.5.0.+data RecordedEventType+ = ChannelComm String+ | BarrierSync String+ | ClockSync String deriving (Eq, Ord, Show)++getEventTypeVal :: RecordedEventType -> String+getEventTypeVal (ChannelComm s) = s+getEventTypeVal (BarrierSync s) = s+getEventTypeVal (ClockSync s) = s++-- Not really a CSP event, more like an enrollable poisonable alting barrier!+data Event = Event {+ getEventUnique :: Unique, -- Event identifier+ getEventPriority :: Int, -- Priority+ getEventType :: STM RecordedEventType, -- Event type for trace recording+ getEventTVar :: TVar (WithPoison+ (Int, -- Enrolled count+ Integer, -- Event sequence count+ OfferSetSet) -- A list of offer sets+ )}++instance Eq Event where+ (==) = (==) `on` getEventUnique++instance Ord Event where+ compare = compare `on` getEventUnique++-- For testing:+instance Show Event where+ show e = "Event " ++ show (hashUnique $ getEventUnique e)++newEvent :: STM RecordedEventType -> Int -> IO Event+newEvent t n+ = do u <- newUnique+ atomically $ do tv <- newTVar (NoPoison (n, 0, []))+ return $ Event u 0 t tv++newEventPri :: STM RecordedEventType -> Int -> Int -> IO Event+newEventPri t n pri+ = do u <- newUnique+ atomically $ do tv <- newTVar (NoPoison (n, 0, []))+ return $ Event u pri t tv+++-- The value used to pass information to a waiting process once one of their events+-- has fired (and they have been committed to it). The Int is an index into their+-- list of guards+newtype SignalValue = Signal (WithPoison Int)+ deriving (Eq, Show)++type SignalVar = TVar (Maybe (SignalValue, Map.Map Unique (Integer, RecordedEventType)))++addPoison :: SignalValue -> SignalValue+addPoison = const $ Signal PoisonItem++nullSignalValue :: SignalValue+nullSignalValue = Signal $ NoPoison (-1)++isNullSignal :: SignalValue -> Bool+isNullSignal (Signal n) = n == NoPoison (-1)++data Offer = Offer {signalValue :: SignalValue, offerAction :: STM (), eventsSet :: EventSet}++data OfferSet = OfferSet { signalVar :: SignalVar -- Variable to use to signal when committed+ , threadId :: ThreadId+ , processId :: ProcessId -- Id of the process making the offer+ , offersSet :: [Offer]} -- Value to send when committed+ -- A list of all sets of events currently offered++instance Eq OfferSet where+ (==) = (==) `on` threadId++instance Ord OfferSet where+ compare = compare `on` threadId++instance Show OfferSet where+ show os = "OfferSet " ++ show (processId os, map (signalValue &&& eventsSet) $ offersSet os)++makeOfferSet :: SignalVar -> ProcessId -> ThreadId -> [((SignalValue, STM ()), EventSet)] -> OfferSet+makeOfferSet v pid tid = OfferSet v tid pid . map (uncurry (uncurry Offer))
Control/Concurrent/CHP/Guard.hs view
@@ -35,7 +35,6 @@ import qualified Data.Map as Map import Data.Monoid import Data.Unique-import System.IO import Control.Concurrent.CHP.Event import Control.Concurrent.CHP.Traces.Base
Control/Concurrent/CHP/Monad.hs view
@@ -42,8 +42,6 @@ import Control.Concurrent import Control.Monad.Reader-import Control.Monad.State-import Control.Monad.Trans import Data.Unique -- This module primarily re-exports the public definitions from@@ -131,5 +129,5 @@ -- process, which is not the desired behaviour. The only thing I can think -- to do is to repeatedly wait for a very long time. hang :: IO a- hang = do forever $ threadDelay maxBound- return undefined+ hang = forever $ threadDelay maxBound+
Control/Concurrent/CHP/Parallel.hs view
@@ -34,7 +34,6 @@ import Control.Concurrent.STM import qualified Control.Exception.Extensible as C import Control.Monad.Reader-import Control.Monad.State import Data.List import Data.Maybe import Data.Ord@@ -175,7 +174,7 @@ writeTVar b (pa, n + 1) trace <- liftCHP $ PoisonT $ lift $ liftTrace ask [blank] <- liftIO $ blankTraces trace 1- liftIO $ forkIO $ do+ _ <- liftIO $ forkIO $ do r <- wrapProcess p $ flip runReaderT blank . pullOutStandard C.block $ atomically $ do (poisonedAlready, n) <- readTVar b
chp.cabal view
@@ -1,5 +1,5 @@ Name: chp-Version: 2.0.0+Version: 2.1.0 Synopsis: An implementation of concurrency ideas from Communicating Sequential Processes License: BSD3 License-file: LICENSE@@ -48,6 +48,9 @@ Control.Concurrent.CHP.Channels.Base Control.Concurrent.CHP.CSP Control.Concurrent.CHP.Event+ Control.Concurrent.CHP.EventMap+ Control.Concurrent.CHP.EventSet+ Control.Concurrent.CHP.EventType Control.Concurrent.CHP.Guard Control.Concurrent.CHP.Mutex Control.Concurrent.CHP.Poison