chp-1.1.0: Control/Concurrent/CHP/Alt.hs
-- 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.
-- | A module containing the ALT constructs. An ALT (a term inherited from
-- occam) is a choice between several alternate events. In CHP, we say that an event
-- must support alting to be a valid choice. Events that /do/ support alting are:
--
-- * 'Control.Concurrent.CHP.Monad.skip'
--
-- * 'Control.Concurrent.CHP.Monad.stop'
--
-- * 'Control.Concurrent.CHP.Monad.waitFor'
--
-- * Reading from a channel (including extended reads): that is, calls to 'Control.Concurrent.CHP.Channels.readChannel'
-- and 'Control.Concurrent.CHP.Channels.extReadChannel'
--
-- * Writing to a channel (including extended writes): that is, calls to 'Control.Concurrent.CHP.Channels.writeChannel'
-- and 'Control.Concurrent.CHP.Channels.extWriteChannel'
--
-- * Synchronising on a barrier (using 'Control.Concurrent.CHP.Barriers.syncBarrier')
--
-- * An alting construct (that is, you can nest alts) such as 'alt', 'priAlt' (or
-- the operator versions)
--
-- * A sequential composition, if the first event supports alting (i.e. is in this
-- list)
--
-- * A call to 'every', which joins together several items (see the documentation
-- on 'every').
--
-- Examples of events that do /NOT/ support alting are:
--
-- * Enrolling and resigning with a barrier
--
-- * Poisoning a channel
--
-- * Processes composed in parallel (using 'runParallel', etc)
--
-- * Any lifted IO event
--
-- * Creating channels, barriers, etc
--
-- * Claiming a shared channel (yet...)
--
-- It is not easily possible to represent this at the type level (while still
-- making CHP easy to use). Therefore it is left to you to not try to alt
-- over something that does not support it. Given how much of the library
-- does support alting, that should hopefully be straightforward.
--
-- Here are some examples of using alting:
--
-- * Wait for an integer channel, or 1 second to elapse:
--
-- > liftM Just (readChannel c) <-> (waitFor 1000000 >> return Nothing)
--
-- * Check if a channel is ready, otherwise return immediately. Note that you must use the
-- alt operator with priority here, otherwise your skip guard might be chosen,
-- even when the channel is ready.
--
-- > liftM Just (readChannel c) </> (skip >> return Nothing)
--
-- * Wait for input from one of two (identically typed) channels
--
-- > readChannel c0 <-> readChannel c1
--
-- * Check if a channel is ready; if so send, it on, otherwise return immediately:
--
-- > (readChannel c >>= writeChannel d) </> skip
--
-- Note that if you wait for a sequential composition:
--
-- > (readChannel c >>= writeChannel d) <-> (writeChannel e 6 >> readChannel f)
--
-- This only waits for the first action in both (reading from channel c, or writing
-- to channel e), not for all of the actions (as, for example, an STM transaction
-- would).
module Control.Concurrent.CHP.Alt (alt, (<->), priAlt, (</>), every, (<&>)) where
import Control.Concurrent.STM
import Control.Monad.State
import Control.Monad.Trans
import Data.List
import Data.Maybe
import System.IO
import Control.Concurrent.CHP.Base
import Control.Concurrent.CHP.Event
import Control.Concurrent.CHP.Guard
import Control.Concurrent.CHP.Parallel
import Control.Concurrent.CHP.Poison
import Control.Concurrent.CHP.Traces.Base
-- | An alt between several actions, with arbitrary priority. The first
-- available action is chosen (with an arbitrary choice if many actions are
-- available at the same time), its body run, and its value returned.
alt :: [CHP a] -> CHP a
alt = priAlt
-- | An alt between several actions, with descending priority. The first
-- available action is chosen (biased towards actions nearest the beginning
-- of the list), its body run, and its value returned.
--
-- What priority means here is a difficult thing, and in some ways a historical
-- artifact. We can group the guards into three categories:
--
-- 1. synchronisation guards (reading from and writing to channels, and synchronising
-- on barriers)
--
-- 2. time-out guards (such as 'Control.Concurrent.CHP.Monad.waitFor')
--
-- 3. dummy guards ('Control.Concurrent.CHP.Monad.skip' and 'Control.Concurrent.CHP.Monad.stop')
--
-- There exists priority when comparing dummy guards to anything else. So for
-- example,
--
-- > priAlt [ skip, x ]
--
-- Will always select the first guard, whereas:
--
-- > priAlt [ x , skip ]
--
-- Is an effective way to poll and see if x is ready, otherwise the 'Control.Concurrent.CHP.Monad.skip' will
-- be chosen. However, there is no priority between synchronisation guards and
-- time-out guards. So the two lines:
--
-- > priAlt [ x, y ]
-- > priAlt [ y, x ]
--
-- May have the same or different behaviour (when x and y are not dummy guards),
-- there is no guarantee either way. The reason behind this is that if you ask
-- for:
--
-- > priAlt [ readChannel c, writeChannel d 6 ]
--
-- And the process at the other end is asking for:
--
-- > priAlt [ readChannel d, writeChannel c 8 ]
--
-- Whichever channel is chosen by both processes will not satisfy the priority
-- at one end (if such priority between channels was supported).
priAlt :: [CHP a] -> CHP a
priAlt items = (liftPoison $ priAlt' $ map wrapPoison items) >>= checkPoison
-- | A useful operator to perform an 'alt'. This operator is associative,
-- and has arbitrary priority. When you have lots of guards, it is probably easier
-- to use the 'alt' function. 'alt' /may/ be more efficent than
-- foldl1 (\<-\>)
(<->) :: CHP a -> CHP a -> CHP a
(<->) a b = alt [a,b]
-- | A useful operator to perform a 'priAlt'. This operator is
-- associative, and has descending priority (that is, it is
-- left-biased). When you have lots of actions, it is probably easier
-- to use the 'priAlt' function. 'priAlt' /may/ be more efficent than
-- foldl1 (\<\/\>)
(</>) :: CHP a -> CHP a -> CHP a
(</>) a b = priAlt [a,b]
infixl </>
infixl <->
infixl <&>
-- | Runs all the given processes in parallel with each other, but only when the
-- choice at the beginning of each item is ready.
--
-- So for example, if you do:
--
-- > every [ readChannel c >>= writeChannel d, readChannel e >>= writeChannel f]
--
-- This will forward values from c and e to d and f respectively in parallel, but
-- only once both channels c and e are ready to be read from. So f cannot be written
-- to before c is read from (contrast this with what would happen if 'every' were
-- replaced with 'runParallel').
--
-- This behaviour can be somewhat useful, but 'every' is much more powerful when
-- used as part of an 'alt'. This code:
--
-- > alt [ every [ readChannel c, readChannel d]
-- > , every [ writeChannel e 6, writeChannel f 8] ]
--
-- Waits to either read from channels c and d, or to write to channels e and f.
--
-- The events involved can partially overlap, e.g.
--
-- > alt [ every [ readChannel a, readChannel b]
-- > , every [ readChannel a, writeChannel c 6] ]
--
-- This will wait to either read from channels a and b, or to read from a and write
-- to c, whichever combination is ready first. If both are ready, the choice between
-- them will be arbitrary (just as with any other choices; see 'alt' for more details).
--
-- The sets can even be subsets of each other, such as:
--
-- > alt [ every [ readChannel a, readChannel b]
-- > , every [ readChannel a, readChannel b, readChannel b] ]
--
-- In this case there are no guarantees as to which choice will happen. Do not
-- assume it will be the smaller, and do not assume it will be the larger.
--
-- Be wary of what happens if a single event is included multiple times in the same 'every', as
-- this may not do what you expect (with or without choice). Consider:
--
-- > every [ readChannel c >> writeChannel d 6
-- > , readChannel c >> writeChannel d 8 ]
--
-- What will happen is that the excecution will wait to read from c, but then it
-- will execute only one of the bodies (an arbitrary choice). In general, do not
-- rely on this behaviour, and instead try to avoid having the same events in an
-- 'every'. Also note that if you synchronise on a barrier twice in an 'every',
-- this will only count as one member enrolling, even if you have two enrolled
-- ends! For such a use, look at 'runParallel' instead.
--
-- Also note that this currently applies to both ends of channels, so that:
--
-- > every [ readChannel c, writeChannel c 2 ]
--
-- Will block indefinitely, rather than completing the communication.
--
-- Each item 'every' must support choice (and in fact
-- only a subset of the items supported by 'alt' are supported by 'every'). Currently the items
-- in the list passed to 'every' must be one of the following:
--
-- * A call to 'Control.Concurrent.CHP.Channels.readChannel' (or 'Control.Concurrent.CHP.Channels.extReadChannel').
--
-- * A call to 'Control.Concurrent.CHP.Channels.writeChannel' (or 'Control.Concurrent.CHP.Channels.extWriteChannel').
--
-- * 'Control.Concurrent.CHP.Monad.skip', the always-ready guard.
--
-- * 'Control.Concurrent.CHP.Monad.stop', the never-ready guard (will cause the whole 'every' to never be ready,
-- since 'every' has to wait for all guards).
--
-- * A call to 'Control.Concurrent.CHP.Monad.syncBarrier'.
--
-- * A sequential composition where the first item is one of the things in this
-- list.
--
-- * A call to 'every' (you can nest 'every' blocks inside each other).
--
-- Timeouts (e.g. 'Control.Concurrent.CHP.Monad.waitFor') are currently not supported. You can always get another
-- process to synchronise on an event with you once a certain time has passed.
--
-- Note also that you cannot put an 'alt' inside an 'every'. So you cannot say:
--
-- > every [ readChannel c
-- > , alt [ readChannel d, readChannel e ] ]
--
-- To wait for c and (d or e) like this you must expand it out into (c and d) or
-- (c and e):
--
-- > alt [ every [ readChannel c, readChannel d]
-- > , every [ readChannel c, readChannel e] ]
--
-- As long as x meets the conditions laid out above, 'every' [x] will have the same
-- behaviour as x.
--
-- Added in version 1.1.0
every :: [CHP a] -> CHP [a]
every [] = liftPoison $ AltableT (SkipGuard [], return []) (return [])
every xs = liftPoison (AltableT (foldl1 merge $ map blankEvent gs, getEventPoison True) (return
$ NoPoison False)) >>= checkPoison >>= \b -> if b then runParallel (map (unwrapPoison . liftTrace) bodies) else alt [every xs]
where
(gs, bodies) = unzip $ map (pullOutAltable . wrapPoison) xs
blankEvent :: Guard -> Guard
blankEvent (EventGuard _ rec act es) = EventGuard [] rec act es
blankEvent g = g
merge :: Guard -> Guard -> Guard
merge (SkipGuard _) g = g
merge g (SkipGuard _) = g
merge StopGuard _ = StopGuard
merge _ StopGuard = StopGuard
merge (EventGuard _ recx actx esx) (EventGuard _ recy acty esy)
= EventGuard [] (recx ++ recy) (actx >> acty) (esx ++ esy)
merge _ _ = BadGuard "merging unsupported guards"
-- | A useful operator that acts like 'every'. The operator is associative and
-- commutative (see 'every' for notes on idempotence). When you have lots of things
-- to join with this operator, it's probably easier to use the 'every' function.
--
-- Added in version 1.1.0
(<&>) :: forall a b. CHP a -> CHP b -> CHP (a, b)
(<&>) a b = every [a >>= return . Left, b >>= return . Right] >>= return . merge
where
merge :: [Either a b] -> (a, b)
merge [Left x, Right y] = (x, y)
merge [Right y, Left x] = (x, y)
merge _ = error "Invalid merge possibility in <&>"
-- ALTing is implemented as follows in CHP. The CHP monad has [Int] in its
-- state. When you choose between N events, you form one body, that pulls
-- the next number from the head of the state and executes the body for the
-- event corresponding to that index. Nested ALTs prepend to the list.
-- So for example, if you choose between:
--
-- > (a <-> b) <-> c
--
-- The overall number corresponding to a is [0,0], b is [0,1], c is [1]. The
-- outer choice peels off the head of the list. On 1 it executes c; on 0 it
-- descends to the nested choice, which takes the next number in the list and
-- executes a or b given 0 or 1 respectively.
--
-- If an event is poisoned, an integer (of arbitrary value) is /appended/ to
-- the list. Thus when an event-based guard is executed, if the list in the
-- state is non-empty, it knows it has been poisoned.
--
-- I did try implementing this in a more functional manner, making each event
-- in the monad take [Int] and return the body, rather than using state. However,
-- I had some memory efficiency problems so I went with the state-monad-based
-- approach instead.
priAlt' :: forall a. [CHP' a] -> CHP' a
priAlt' items
-- Our guard is a nested guard of all our sub-guards.
-- Our action-if-not-guard is to do the selection ourselves.
-- Our body is to read the numbered list, strip one off and follow the path,
-- ignoring the action-if-not-guard of the chosen body
= AltableT (NestedGuards $ wrappedGuards
,executeNumberedBody)
(selectFromGuards >> executeNumberedBody)
where
wrappedGuards :: [Guard]
wrappedGuards = map wrap flattenedGuards
where
wrap :: (Int, Guard) -> Guard
wrap (n, SkipGuard ns) = SkipGuard $ n : ns
wrap (n, EventGuard ns e act ab) = EventGuard (n:ns) e act ab
wrap (n, TimeoutGuard g) = TimeoutGuard $
do g' <- g
return $ do ns <- g'
return (n : ns)
wrap (_, g@(BadGuard _)) = g
wrap (_, _) = BadGuard "wrapped"
-- Polls the available guards, but ignores timeout guards and alting barrier
-- guards
checkNormalGuards :: STM (Maybe Int)
checkNormalGuards = foldl1 orElse $
(map checkGuard flattenedGuards) ++ [return Nothing]
where
checkGuard :: (Int, Guard) -> STM (Maybe Int)
checkGuard (n, BadGuard _) = return $ Just n
checkGuard (n, SkipGuard {}) = return $ Just n
checkGuard (_, _) = retry
-- Waits for one of the normal (non-alting barrier) guards to be ready,
-- or the given transaction to complete
waitNormalGuards :: STM [Int] -> IO (Bool, [Int])
waitNormalGuards extra
= do guards <- mapM enable wrappedGuards
atomically $ foldl1 orElse (wrap True extra : map (wrap False) guards)
where
enable :: Guard -> IO (STM [Int])
enable (BadGuard _) = return $ return []
enable (SkipGuard ns) = return $ return ns
enable (TimeoutGuard g) = g
enable _ = return retry -- This effectively ignores other guards
wrap :: Bool -> STM [Int] -> STM (Bool, [Int])
wrap b m = do x <- m
return (b, x)
-- The list of guards without any NestedGuards or StopGuards:
flattenedGuards :: [(Int, Guard)]
flattenedGuards = (flatten $ zip [0..] $ map (fst . pullOutAltable) items)
where
flatten :: [(Int, Guard)] -> [(Int,Guard)]
flatten [] = []
flatten ((n,x):xs) = case x of
NestedGuards gs -> flatten $ zip (repeat n) gs ++ xs
StopGuard -> flatten xs
g -> (n, g) : flatten xs
-- The alting barrier guards:
eventGuards :: [([RecordedIndivEvent], [Int], STM (), [Event])]
eventGuards = [(rec,ns,act,ab) | EventGuard ns rec act ab <- wrappedGuards]
-- We must use isPrefixOf, because things are added in the case of poison
findEventAssoc :: [Int] -> [RecordedIndivEvent]
findEventAssoc x = case filter (\(_,y,_,_) -> y `isPrefixOf` x) eventGuards of
[(rec,_,_,_)] -> rec
_ -> error "Could not find associated event in alt, internal logic error"
-- Stores a list of ints in the state
storeChoice :: [Int] -> TraceT IO ()
storeChoice ns = modify (\(_, es) -> (ns, es))
isBadGuard :: Guard -> Bool
isBadGuard (BadGuard _) = True
isBadGuard _ = False
-- Performs the select operation on all the guards. The choice is stored
-- in the state ready to execute the bodies
selectFromGuards :: TraceT IO ()
selectFromGuards
| null eventGuards
= do (_,ns) <- liftIO $ waitNormalGuards retry
storeChoice ns
| any isBadGuard wrappedGuards
= let str = head [s | BadGuard s <- wrappedGuards]
err = "ALTing not supported on given guard: " ++ str
in liftIO $ do hPutStrLn stderr err
ioError $ userError err
| otherwise
= do earliestReady <- liftIO $ atomically checkNormalGuards
tv <- liftIO . atomically $ newTVar Nothing
pid <- getProcessId
(_, tr) <- get
mn <- liftIO . atomically $ do
ret <- enableEvents tv pid
(maybe id take earliestReady [(x,y,z) | (_,x,y,z)<-eventGuards])
(isNothing earliestReady)
maybe (return ()) (\(_,es) -> recordEventLast (nub es) tr) ret
return ret
case (mn, earliestReady) of
-- An event -- and we were the last person to arrive:
-- The event must have been higher priority than any other
-- ready guards
(Just (ns, _), _) ->
do recordEvent $ findEventAssoc ns
storeChoice ns
-- No events were ready, but there was an available normal
-- guards. Re-run the normal guards; at least one will be ready
(Nothing, Just _) ->
do (_, ns) <- liftIO $ waitNormalGuards retry
storeChoice ns
-- No events ready, no other guards ready either
-- Events will have been enabled; wait for everything:
(Nothing, Nothing) ->
do (wasAltingBarrier, ns) <- liftIO $ waitNormalGuards $ waitAlting tv
if wasAltingBarrier
then recordEvent (findEventAssoc ns) >> storeChoice ns -- It was a barrier, all done
else
-- Another guard fired, but we must check in case
-- we have meanwhile been committed to taking an
-- event:
do mn' <- liftIO . atomically $ disableEvents tv (concatMap fourth eventGuards)
case mn' of
-- An event overrides our non-event choice:
Just bns -> recordEvent (findEventAssoc bns) >> storeChoice bns
-- Go with the original option, no events
-- became ready:
Nothing -> storeChoice ns
where
waitAlting :: TVar (Maybe [Int]) -> STM [Int]
waitAlting tv = do b <- readTVar tv
case b of
Nothing -> retry
Just ns -> return ns
fourth (_,_,_,c) = c
executeNumberedBody :: TraceT IO a
executeNumberedBody
= do st <- get
case st of
((g:gs), es) ->
do put (gs, es)
snd $ pullOutAltable (items !! g)
([], _) -> liftIO $
do hPutStrLn stderr "ALTing not supported on given guard (no index)"
ioError $ userError "ALTing not supported on given guard (no index)"