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 @@ -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 diff --git a/Control/Concurrent/CHP/Arrow.hs b/Control/Concurrent/CHP/Arrow.hs --- a/Control/Concurrent/CHP/Arrow.hs +++ b/Control/Concurrent/CHP/Arrow.hs @@ -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 diff --git a/Control/Concurrent/CHP/Enroll.hs b/Control/Concurrent/CHP/Enroll.hs --- a/Control/Concurrent/CHP/Enroll.hs +++ b/Control/Concurrent/CHP/Enroll.hs @@ -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 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 @@ -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 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 @@ -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, diff --git a/chp.cabal b/chp.cabal --- a/chp.cabal +++ b/chp.cabal @@ -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