chp 1.3.1 → 1.3.2
raw patch · 6 files changed
+94/−57 lines, 6 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
+ Control.Concurrent.CHP.Arrow: arrStrict :: (NFData b) => (a -> b) -> ProcessPipeline a b
+ Control.Concurrent.CHP.Enroll: furtherEnroll :: (Enrollable b z) => Enrolled b z -> (Enrolled b z -> CHP a) -> CHP a
Files
- Control/Concurrent/CHP/Alt.hs +42/−36
- Control/Concurrent/CHP/Arrow.hs +11/−1
- Control/Concurrent/CHP/Enroll.hs +12/−1
- Control/Concurrent/CHP/Event.hs +19/−12
- Control/Concurrent/CHP/Parallel.hs +9/−6
- chp.cabal +1/−1
Control/Concurrent/CHP/Alt.hs view
@@ -354,7 +354,7 @@ priAlt' items -- Our guard is a nested guard of all our sub-guards. -- Our action-if-not-guard is to do the selection ourselves.- = AltableT flattenedGuards selectFromGuards+ = AltableT flattenedGuards (selectFromGuards flattenedGuards) where -- The list of guards without any NestedGuards or StopGuards: flattenedGuards :: Either String [(Guard, TraceT IO (WithPoison a))]@@ -363,22 +363,16 @@ altStuff :: Either String [(Guard, TraceT IO (WithPoison a))] altStuff = liftM concat $ mapM pullOutAltable items - -- The alting barrier guards:- eventGuards :: [Guard] -> [((Unique -> Integer) -> [RecordedIndivEvent Unique], Int, STM (), [Event])]- eventGuards guards = [(rec,n,act,ab) | (n, EventGuard rec act ab) <- zip [0..] guards]-- -- Performs the select operation on all the guards, and then executes the body- selectFromGuards :: TraceT IO (WithPoison a)- selectFromGuards = case flattenedGuards of- Left str ->+-- Performs the select operation on all the guards, and then executes the body+selectFromGuards :: (Either String [(Guard, TraceT IO (WithPoison a))]) -> TraceT IO (WithPoison a)+selectFromGuards (Left str) = let err = "ALTing not supported on given guard: " ++ str in liftIO $ do hPutStrLn stderr err ioError $ userError err-- Right both- | null (eventGuards $ map fst both) ->- join $ liftM snd $ liftIO $ waitNormalGuards both retry- | otherwise ->+selectFromGuards (Right both)+ | null (eventGuards $ map fst both) =+ join $ liftM snd $ liftIO $ waitNormalGuards both Nothing+ | otherwise = do let (guards, bodies) = unzip both earliestReady = findIndex isSkipGuard guards recordAndRun _ (Signal PoisonItem)@@ -390,14 +384,23 @@ justRun (Signal (NoPoison n)) = bodies !! n guardsAndSignal :: [(Guard, (SignalValue, Map.Map Unique Integer))] guardsAndSignal = zip guards (zip (map (Signal . NoPoison) [0..]) (repeat Map.empty))- tv <- liftIO . atomically $ newTVar Nothing+ tv <- liftIO $ newTVarIO Nothing pid <- getProcessId tr <- get mn <- liftIO . atomically $ do ret <- enableEvents tv pid- (maybe id take earliestReady [(Signal $ NoPoison x,y,z) | (_,x,y,z)<-eventGuards guards])+ (maybe id take earliestReady $ eventGuards guards) (isNothing earliestReady)- maybe (return ()) (\(_,es) -> recordEventLast (nub es) tr) ret+ maybe (return ())+ (\((sigVal,_),es) -> do+ recordEventLast (nub es) tr+ case sigVal of+ Signal PoisonItem -> return ()+ Signal (NoPoison n) ->+ let EventGuard _ act _ = guards !! n+ in act+ )+ ret return ret case (mn, earliestReady) of -- An event -- and we were the last person to arrive:@@ -407,19 +410,19 @@ -- No events were ready, but there was an available normal -- guards. Re-run the normal guards; at least one will be ready (Nothing, Just _) ->- join $ liftM snd $ liftIO $ waitNormalGuards both retry+ 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) -> do (wasAltingBarrier, (n, m)) <- liftIO $ waitNormalGuards- guardsAndSignal $ waitAlting tv+ guardsAndSignal $ Just $ waitAlting tv if wasAltingBarrier then recordAndRun m n -- It was a barrier, all done else -- Another guard fired, but we must check in case -- we have meanwhile been committed to taking an -- event:- do mn' <- liftIO . atomically $ disableEvents tv (concatMap fourth+ do mn' <- liftIO . atomically $ disableEvents tv (concatMap snd $ eventGuards guards) case mn' of -- An event overrides our non-event choice:@@ -427,26 +430,29 @@ -- Go with the original option, no events -- became ready: Nothing -> justRun n- where- waitAlting :: SignalVar -> STM (SignalValue, Map.Map Unique Integer)- waitAlting tv = do b <- readTVar tv- case b of- Nothing -> retry- Just ns -> return ns- fourth (_,_,_,c) = c - makeLookup :: Map.Map Unique Integer -> Unique -> Integer- makeLookup m u = fromMaybe (error "CHP: Unique not found in alt") $ Map.lookup u m+waitAlting :: SignalVar -> STM (SignalValue, Map.Map Unique Integer)+waitAlting tv = do b <- readTVar tv+ case b of+ Nothing -> retry+ Just ns -> return ns +makeLookup :: Map.Map Unique Integer -> Unique -> Integer+makeLookup m u = fromMaybe (error "CHP: Unique not found in alt") $ Map.lookup u m +-- The alting barrier guards:+eventGuards :: [Guard] -> [(SignalValue, [Event])]+eventGuards guards = zip (map (Signal . NoPoison) [0..]) [ab | EventGuard _ _ ab <- guards]++ -- Waits for one of the normal (non-alting barrier) guards to be ready, -- or the given transaction to complete-waitNormalGuards :: [(Guard, b)] -> STM b -> IO (Bool, b)+waitNormalGuards :: [(Guard, b)] -> Maybe (STM b) -> IO (Bool, b) waitNormalGuards guards extra- = do enabled <- mapM enable guards- atomically $ foldl1 orElse (liftM ((,) True) extra : enabled)+ = do enabled <- sequence $ mapMaybe enable guards+ atomically $ foldr orElse retry $ maybe id ((:) . liftM ((,) True)) extra $ enabled where- enable :: (Guard, b) -> IO (STM (Bool, b))- enable (SkipGuard, x) = return $ return (False, x)- enable (TimeoutGuard g, x) = liftM (>> return (False, x)) g- enable _ = return retry -- This effectively ignores other guards+ enable :: (Guard, b) -> Maybe (IO (STM (Bool, b)))+ enable (SkipGuard, x) = Just $ return $ return (False, x)+ enable (TimeoutGuard g, x) = Just $ liftM (>> return (False, x)) g+ enable _ = Nothing -- This effectively ignores other guards
Control/Concurrent/CHP/Arrow.hs view
@@ -60,7 +60,7 @@ -- together a certain class of process network, you should do fine. -- -- Added in version 1.0.2.-module Control.Concurrent.CHP.Arrow (ProcessPipeline, runPipeline, arrowProcess) where+module Control.Concurrent.CHP.Arrow (ProcessPipeline, runPipeline, arrowProcess, arrStrict) where -- I have got this module to work on GHC 6.8 and 6.10 by following the CPP-variant -- instructions on this page: http://haskell.org/haskellwiki/Upgrading_packages@@ -75,6 +75,7 @@ hiding (pure) #endif import Control.Monad+import Control.Parallel.Strategies import Control.Concurrent.CHP import qualified Control.Concurrent.CHP.Common as CHP@@ -112,6 +113,15 @@ -- Added in version 1.1.0 arrowProcess :: (Chanin a -> Chanout b -> CHP ()) -> ProcessPipeline a b arrowProcess = ProcessPipeline++-- | Like the arr function of the ProcessPipeline arrow instance, but fully evaluates+-- the result before sending it. If you are building process pipelines with arrows to+-- try and get some parallel speed-up, you should try this function instead of+-- arr itself.+-- +-- Added in version 1.3.2+arrStrict :: NFData b => (a -> b) -> ProcessPipeline a b+arrStrict = ProcessPipeline . CHP.map' #if __GLASGOW_HASKELL__ >= 609 instance Category ProcessPipeline where
Control/Concurrent/CHP/Enroll.hs view
@@ -30,7 +30,8 @@ -- | A module with support for things that are enrollable (barriers and broadcast -- channels). -module Control.Concurrent.CHP.Enroll (Enrolled, Enrollable(..), enrollPair, enrollList) where+module Control.Concurrent.CHP.Enroll (Enrolled, Enrollable(..), furtherEnroll,+ enrollPair, enrollList) where import Control.Concurrent.CHP.Base @@ -53,6 +54,16 @@ -- enroll on the barrier inside this, nesting enroll and resign blocks as -- much as you like resign :: Enrolled b z -> CHP a -> CHP a++-- | Just like enroll, but starts with an already enrolled item. During the+-- body, you are enrolled twice -- once from the original enrolled item (which+-- is still valid in the body) and once from the new enrolled item passed to+-- the inner function. Afterwards, you are enrolled once, on the original item+-- you had.+--+-- This function was added in version 1.3.2.+furtherEnroll :: Enrollable b z => Enrolled b z -> (Enrolled b z -> CHP a) -> CHP a+furtherEnroll (Enrolled x) = enroll x -- | Like 'enroll' but enrolls on the given pair of barriers enrollPair :: (Enrollable b p, Enrollable b' p') => (b p, b' p') -> ((Enrolled
Control/Concurrent/CHP/Event.hs view
@@ -116,9 +116,11 @@ -- choose it unionAll :: Ord k => [Map.Map k a] -> Map.Map k a-unionAll = foldl Map.union Map.empty+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 -> STM (WithPoison Integer)@@ -251,10 +253,12 @@ nullOffer :: [(SignalValue, Map.Map Event ())] nullOffer = [(nullSignalValue,Map.empty)] - -- SMallest offers first to minimise backtracking:- sortOffers :: [OfferSet] -> [OfferSet]- sortOffers xs = sortBy (compare `on` (\(OfferSet (_,_,es)) -> length es)) xs- -- TODO put the newest process first again+-- 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@@ -331,8 +335,12 @@ then -- It could be ready discoverRelatedOffersAll (NoPoison (accum ++ offers, Set.insert e events))- (next ++ zip (repeat $ return ())- (Map.keys $ unionAll otherEvents))+ -- 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 @@ -443,10 +451,9 @@ -- ^ Variable used to signal the process once a choice is made -> ProcessId -- ^ The id of the process making the choice- -> [(SignalValue, STM (), [Event])]+ -> [(SignalValue, [Event])] -- ^ The list of options. Each option has a signalvalue to return- -- if chosen, an STM action to execute at the same time as the- -- synchronisation, and a list of events (conjoined together).+ -- if chosen, and a list of events (conjoined together). -- So this list is the disjunction of conjunctions, with a little -- more information. -> Bool@@ -457,7 +464,7 @@ -- the offers. -> STM (Maybe ((SignalValue, Map.Map Unique Integer), [((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])+ = 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@@ -469,7 +476,7 @@ 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]+ 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
Control/Concurrent/CHP/Parallel.hs view
@@ -120,20 +120,25 @@ -- length and ordering of the input runParallelPoison :: forall a. [CHP a] -> CHP [a] runParallelPoison processes- = do c <- liftIO $ atomically $ newResultsVar+ = do (final, intermed) <- liftIO $ atomically $ do+ a <- newResultsVar+ b <- newResultsVar+ return (a, b) trace <- PoisonT $ lift $ liftTrace get blanks <- liftIO $ blankTraces trace (length processes) liftIO $ mapM_ forkIO [do y <- wrapProcess p $ flip runStateT btr . pullOutStandard C.block $ atomically $- do ys <- readTVar c- writeTVar c $ (case y of+ do ys <- readTVar intermed+ writeTVar+ (if length ys == length processes - 1 then final else intermed)+ $ (case y of Nothing -> (n, (Nothing, Nothing)) Just (Right (x,t)) -> (n, (Just x, Just t)) Just (Left t) -> (n, (Nothing, Just t)) ) : ys | (p, btr, n) <- zip3 processes blanks [0..]]- results <- liftIO $ atomically $ do xs <- readTVar c+ results <- liftIO $ atomically $ do xs <- readTVar final if length xs == length processes then return xs else retry@@ -143,8 +148,6 @@ where newResultsVar :: STM (TVar [(Integer, (Maybe a, Maybe TraceStore))]) newResultsVar = newTVar []---- TODO could make the parent only wake up once, with a bit of twiddling -- | A monad transformer used for introducing forking blocks. newtype (Monad m, MonadCHP m) => ForkingT m a = Forking (ReaderT (TVar (Bool,
chp.cabal view
@@ -1,5 +1,5 @@ Name: chp-Version: 1.3.1+Version: 1.3.2 Synopsis: An implementation of concurrency ideas from Communicating Sequential Processes License: BSD3 License-file: LICENSE