diff --git a/Control/Concurrent/CHP/Alt.hs b/Control/Concurrent/CHP/Alt.hs --- a/Control/Concurrent/CHP/Alt.hs +++ b/Control/Concurrent/CHP/Alt.hs @@ -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' diff --git a/Control/Concurrent/CHP/Barriers.hs b/Control/Concurrent/CHP/Barriers.hs --- a/Control/Concurrent/CHP/Barriers.hs +++ b/Control/Concurrent/CHP/Barriers.hs @@ -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. diff --git a/Control/Concurrent/CHP/Base.hs b/Control/Concurrent/CHP/Base.hs --- a/Control/Concurrent/CHP/Base.hs +++ b/Control/Concurrent/CHP/Base.hs @@ -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 diff --git a/Control/Concurrent/CHP/CSP.hs b/Control/Concurrent/CHP/CSP.hs --- a/Control/Concurrent/CHP/CSP.hs +++ b/Control/Concurrent/CHP/CSP.hs @@ -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 diff --git a/Control/Concurrent/CHP/Channels/Base.hs b/Control/Concurrent/CHP/Channels/Base.hs --- a/Control/Concurrent/CHP/Channels/Base.hs +++ b/Control/Concurrent/CHP/Channels/Base.hs @@ -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 diff --git a/Control/Concurrent/CHP/Channels/BroadcastReduce.hs b/Control/Concurrent/CHP/Channels/BroadcastReduce.hs --- a/Control/Concurrent/CHP/Channels/BroadcastReduce.hs +++ b/Control/Concurrent/CHP/Channels/BroadcastReduce.hs @@ -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)) diff --git a/Control/Concurrent/CHP/Channels/Communication.hs b/Control/Concurrent/CHP/Channels/Communication.hs --- a/Control/Concurrent/CHP/Channels/Communication.hs +++ b/Control/Concurrent/CHP/Channels/Communication.hs @@ -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) diff --git a/Control/Concurrent/CHP/Channels/Creation.hs b/Control/Concurrent/CHP/Channels/Creation.hs --- a/Control/Concurrent/CHP/Channels/Creation.hs +++ b/Control/Concurrent/CHP/Channels/Creation.hs @@ -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 diff --git a/Control/Concurrent/CHP/Event.hs b/Control/Concurrent/CHP/Event.hs --- a/Control/Concurrent/CHP/Event.hs +++ b/Control/Concurrent/CHP/Event.hs @@ -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 diff --git a/Control/Concurrent/CHP/EventMap.hs b/Control/Concurrent/CHP/EventMap.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/EventMap.hs @@ -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 diff --git a/Control/Concurrent/CHP/EventSet.hs b/Control/Concurrent/CHP/EventSet.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/EventSet.hs @@ -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) diff --git a/Control/Concurrent/CHP/EventType.hs b/Control/Concurrent/CHP/EventType.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/EventType.hs @@ -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)) diff --git a/Control/Concurrent/CHP/Guard.hs b/Control/Concurrent/CHP/Guard.hs --- a/Control/Concurrent/CHP/Guard.hs +++ b/Control/Concurrent/CHP/Guard.hs @@ -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 diff --git a/Control/Concurrent/CHP/Monad.hs b/Control/Concurrent/CHP/Monad.hs --- a/Control/Concurrent/CHP/Monad.hs +++ b/Control/Concurrent/CHP/Monad.hs @@ -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 + diff --git a/Control/Concurrent/CHP/Parallel.hs b/Control/Concurrent/CHP/Parallel.hs --- a/Control/Concurrent/CHP/Parallel.hs +++ b/Control/Concurrent/CHP/Parallel.hs @@ -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 diff --git a/chp.cabal b/chp.cabal --- a/chp.cabal +++ b/chp.cabal @@ -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