diff --git a/Control/Concurrent/CHP.hs b/Control/Concurrent/CHP.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP.hs @@ -0,0 +1,59 @@ +-- 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. + +-- | This module re-exports the core functionality of the CHP library. Other +-- modules that you also may wish to import are: +-- +-- * "Control.Concurrent.CHP.Buffers" +-- +-- * "Control.Concurrent.CHP.Common" +-- +-- * "Control.Concurrent.CHP.Console" +-- +-- * "Control.Concurrent.CHP.Traces" +-- +-- * "Control.Concurrent.CHP.Utils" +module Control.Concurrent.CHP ( + module Control.Concurrent.CHP.Alt, + module Control.Concurrent.CHP.Barriers, + module Control.Concurrent.CHP.BroadcastChannels, + module Control.Concurrent.CHP.Channels, + module Control.Concurrent.CHP.Enroll, + module Control.Concurrent.CHP.Monad, + module Control.Concurrent.CHP.Parallel, + ) where + +import Control.Concurrent.CHP.Alt +import Control.Concurrent.CHP.Barriers +import Control.Concurrent.CHP.BroadcastChannels +import Control.Concurrent.CHP.Channels +import Control.Concurrent.CHP.Enroll +import Control.Concurrent.CHP.Monad +import Control.Concurrent.CHP.Parallel + diff --git a/Control/Concurrent/CHP/Alt.hs b/Control/Concurrent/CHP/Alt.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/Alt.hs @@ -0,0 +1,295 @@ +-- 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 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.waitFor' +-- +-- * Reading from a channel (including extended reads) +-- +-- * Writing to a channel +-- +-- * Synchronising on a barrier +-- +-- * An alting construct (that is, you can nest alts) +-- +-- * A sequential composition, if the first event supports alting +-- +-- Examples of events that do /NOT/ support alting are: +-- +-- * Enrolling and resigning with a barrier +-- +-- * Poisoning a channel +-- +-- * Processes composed in parallel +-- +-- * 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 +module Control.Concurrent.CHP.Alt (alt, (<->), priAlt, ()) where + +import Control.Concurrent.STM +import Control.Monad.State +import Control.Monad.Trans +import Data.List +import Data.Maybe +import qualified Data.Set as Set +import System.IO + +import Control.Concurrent.CHP.Base +import Control.Concurrent.CHP.Event +import Control.Concurrent.CHP.Guard +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 guards 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 arbitrary priority. The first +-- available action is chosen (biased towards actions nearest the beginning +-- of the list), its body run, and its value returned. +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 <-> + +-- 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 (_, _) = BadGuard + + -- 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 . getAltable) 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 :: [(RecEvents, [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] -> RecEvents + 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)) + + -- 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 + | otherwise + = do earliestReady <- liftIO $ atomically checkNormalGuards + tv <- liftIO . atomically $ newTVar Nothing + pid <- getProcessId + mn <- liftIO . atomically $ enableEvents tv pid + (maybe id take earliestReady eventGuards) + (isNothing earliestReady) + 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 (rec, pids, ns), _) -> + recordEventLast (fst rec) (Set.fromList pids) + >> recordEvent (snd rec) + >> 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 (snd $ 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 (map fourth eventGuards) + case mn' of + -- An event overrides our non-event choice: + Just bns -> recordEvent (snd $ 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 $ getAltable (items !! g) + ([], _) -> liftIO $ + do hPutStrLn stderr "ALTing not supported on given guard" + ioError $ userError "ALTing not supported on given guard" + diff --git a/Control/Concurrent/CHP/Barriers.hs b/Control/Concurrent/CHP/Barriers.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/Barriers.hs @@ -0,0 +1,138 @@ +-- 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 barriers. +-- +-- A barrier is a synchronisation primitive. When N processes are enrolled +-- on a barrier, all N must synchronise on the barrier before any synchronisations +-- may complete, at which point they all complete. That is, when a single +-- process synchronises on a barrier, it must then wait until all the other +-- enrolled processes also synchronise before it can finish. +-- +-- Only processes enrolled on a barrier may synchronise on it. Enrolled barriers +-- should not be passed around between processes, or used twice in a parallel +-- composition. Instead, each process should enroll on the barrier itself. +-- +-- Barriers support choice (alting). This can lead to a lot of non-determinism +-- and some confusion. Consider these two processes, both enrolled on barriers a and b: +-- +-- > (sync a <-> sync b) +-- > (sync b <-> sync a) +-- +-- Which barrier completes is determined by the run-time, and will be an arbitrary +-- choice. This is even the case when priority is involved: +-- +-- > (sync a sync b) +-- > (sync b sync a) +-- +-- Clearly there is no way to resolve this to satisfy both priorities; the +-- run-time will end up choosing. +-- +-- Barrier poison can be detected when syncing, enrolling or resigning. You +-- may only poison a barrier that you are currently enrolled on. +-- +-- Barriers can also support phases. The idea behind a phased barrier is that +-- a barrier is always on a certain phase P. Whenever a barrier successfully +-- completes, the phase is incremented (but it does not have to be an integer). +-- 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, newBarrier, newBarrierWithLabel, + PhasedBarrier, newPhasedBarrier, newPhasedBarrierWithLabel, currentPhase, waitForPhase, + 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 + +-- | A special case of the PhasedBarrier that has no useful phases, i.e. a +-- standard barrier. +type Barrier = PhasedBarrier () + +-- | Synchronises on the given barrier. You must be enrolled on a barrier in order +-- to synchronise on it. Returns the new phase, following the synchronisation. +syncBarrier :: (Enum phase, Bounded phase, Eq phase) => Enrolled PhasedBarrier phase -> CHP phase +syncBarrier = syncBarrierWith (recAs (Just . BarrierSync) (Just . BarrierSyncIndiv)) + +-- | Finds out the current phase a barrier is on. +currentPhase :: (Enum phase, Bounded phase, Eq phase) => Enrolled PhasedBarrier phase -> CHP phase +currentPhase (Enrolled (Barrier (_,tv))) = liftIO $ atomically $ readTVar tv + +repeatUntil :: (Monad m, Eq a) => (a -> Bool) -> m a -> m () +repeatUntil target comp = do x <- comp + unless (target x) $ repeatUntil target comp + +-- | If the barrier is not in the given phase, synchronises on the barrier +-- repeatedly until it /is/ in the given phase +waitForPhase :: (Enum phase, Bounded phase, Eq phase) => + phase -> Enrolled PhasedBarrier phase -> CHP () +waitForPhase ph b = do phCur <- currentPhase b + when (ph /= phCur) $ + repeatUntil (== ph) (syncBarrier b) + +-- | Creates a new barrier with no processes enrolled +newBarrier :: CHP Barrier +newBarrier = newPhasedBarrier () + +-- | 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 +newPhasedBarrier :: (Enum phase, Bounded phase, Eq phase) => phase -> CHP (PhasedBarrier phase) +newPhasedBarrier ph = liftPoison $ liftTrace $ do + e <- liftIO $ newEvent 0 + tv <- liftIO $ atomically $ newTVar ph + return $ Barrier (e, tv) + + +-- | Creates a new barrier with no processes enrolled and labels it in traces +-- using the given label +newBarrierWithLabel :: String -> CHP Barrier +newBarrierWithLabel l = newPhasedBarrierWithLabel l () + +-- | Creates a new barrier with no processes enrolled and labels it in traces +-- using the given label +newPhasedBarrierWithLabel :: (Enum phase, Bounded phase, Eq phase) => String -> phase -> CHP (PhasedBarrier phase) +newPhasedBarrierWithLabel l ph = liftPoison $ liftTrace $ do + e <- liftIO $ newEvent 0 + labelEvent e l + tv <- liftIO $ atomically $ newTVar ph + return $ Barrier (e, tv) + + +-- | Gets the identifier of a Barrier. Useful if you want to identify it in +-- the trace later on. +getBarrierIdentifier :: (Enum ph, Bounded ph, Eq ph) => PhasedBarrier ph -> Unique +getBarrierIdentifier (Barrier (Event (u,_),_)) = u diff --git a/Control/Concurrent/CHP/Base.hs b/Control/Concurrent/CHP/Base.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/Base.hs @@ -0,0 +1,221 @@ +-- 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 various definitions relating to the CSP\/CSPPoison +-- monads, and poison. Not publicly visible. +module Control.Concurrent.CHP.Base where + +import Control.Concurrent.STM +import qualified Control.Exception as C +import Control.Monad.Error +import Control.Monad.Reader +import Control.Monad.State +import Control.Monad.Writer +import Control.Monad.Trans +import qualified Data.Map as Map +import qualified Text.PrettyPrint.HughesPJ + +import Control.Concurrent.CHP.Guard +import Control.Concurrent.CHP.Poison +import Control.Concurrent.CHP.ProcessId +import Control.Concurrent.CHP.Traces.Base + +-- ====== +-- Types: +-- ====== + +-- | An internal dummy type for poison errors +newtype PoisonError = PoisonError () + +-- | An enrolled wrapper for barriers that shows at the type-level whether +-- you are enrolled on a barrier. Enrolled Barriers should never be passed +-- to two (or more) processes running in parallel; if two processes synchronise +-- based on a single enroll call, undefined behaviour will result. +newtype Enrolled b a = Enrolled (b a) + +-- | The central monad of the library. You can use 'runCHP' and 'runCHP_' +-- to execute programs in this monad. +newtype CHP a = PoisonT (ErrorT PoisonError CHP' a) + deriving (Monad, MonadIO) + +data CHP' a = AltableT { + -- The guard, and body to execute after the guard + getAltable :: (Guard, TraceT IO a), + -- The body to execute without a guard + getStandard :: TraceT IO a } + +-- ======== +-- Classes: +-- ======== + +-- | A monad transformer class that is very similar to MonadIO. This can be +-- useful if you want to add monad transformers (such as StateT, ReaderT) on +-- top of the 'CHP' monad. +class MonadIO m => MonadCHP m where + liftCHP :: CHP a -> m a + +-- | A class representing a type of trace. The main useful function is 'runCHPAndTrace', +-- but because its type is only determined by its return type, you may wish +-- to use the already-typed functions offered in each trace module -- see the +-- modules in "Control.Concurrent.CHP.Traces". +class Show t => Trace t where + -- | Runs the given CHP program, and returns its return value and the trace. + -- The return value is a Maybe type because the process may have exited + -- due to uncaught poison. In that case Nothing is return as the result. + runCHPAndTrace :: CHP a -> IO (Maybe a, t) + -- | The empty trace. + emptyTrace :: t + -- | Pretty-prints the given trace using the "Text.PrettyPrint.HughesPJ" + -- module. + prettyPrint :: t -> Text.PrettyPrint.HughesPJ.Doc + +-- | A class indicating that something is poisonable. +class Poisonable c where + -- | Poisons the given item. + poison :: MonadCHP m => c -> m () + +-- ========== +-- Functions: +-- ========== + +liftTrace :: TraceT IO a -> CHP' a +liftTrace m = AltableT (badGuard, m) m + +wrapPoison :: CHP a -> CHP' (WithPoison a) +wrapPoison (PoisonT m) = (liftM $ either (const PoisonItem) NoPoison) $ + runErrorT m + +unwrapPoison :: CHP' (WithPoison a) -> CHP a +unwrapPoison m = PoisonT (lift m) >>= checkPoison + +-- | Checks for poison, and either returns the value, or throws a poison exception +checkPoison :: WithPoison a -> CHP a +checkPoison (NoPoison x) = return x +checkPoison PoisonItem = PoisonT $ throwError $ PoisonError () + +liftPoison :: CHP' a -> CHP a +liftPoison = PoisonT . lift + +-- | Throws a poison exception. +throwPoison :: CHP a +throwPoison = checkPoison PoisonItem + +-- | Allows you to provide a handler for sections with poison. It is usually +-- used in an infix form as follows: +-- +-- > (readChannel c >>= writeChannel d) `onPoison` (poison c >> poison d) +-- +-- It handles the poison and does not rethrow it (unless your handler +-- does so). If you want to rethrow (and actually, you'll find you usually +-- do), use onPoisonRethrow +onPoisonTrap :: CHP a -> CHP a -> CHP a +onPoisonTrap (PoisonT body) (PoisonT handler) = PoisonT $ body `catchError` (const handler) + +-- | Like 'onPoisonTrap', this function allows you to provide a handler +-- for poison. The difference with this function is that even if the +-- poison handler does not throw, the poison exception will always be +-- re-thrown after the handler anyway. That is, the following lines of +-- code all have identical behaviour: +-- +-- > foo +-- > foo `onPoisonRethrow` throwPoison +-- > foo `onPoisonRethrow` return () +onPoisonRethrow :: CHP a -> CHP () -> CHP a +onPoisonRethrow (PoisonT body) (PoisonT handler) = PoisonT $ + body `catchError` (\err -> handler >> throwError err) + +-- | Poisons all the given items. A handy shortcut for @mapM_ poison@. +poisonAll :: (Poisonable c, MonadCHP m) => [c] -> m () +poisonAll ps = mapM_ poison ps + + +liftSTM :: MonadIO m => STM a -> m a +liftSTM = liftIO . atomically + +getProcessId :: TraceT IO ProcessId +getProcessId = do (_, x) <- get + case x of + Trace (pid,_,_) -> return pid + NoTrace -> return emptyProcessId + +runCHPProgramWith :: TraceStore -> (TraceStore -> t) -> CHP a -> IO (Maybe a, t) +runCHPProgramWith start f (PoisonT p) + = do (x, (_, t)) <- runStateT (liftM (either (const Nothing) Just) $ getStandard $ runErrorT p) ([], start) + return (x, f t) + +runCHPProgramWith' :: SubTraceStore -> (ChannelLabels -> SubTraceStore -> IO t) -> CHP a -> IO (Maybe a, t) +runCHPProgramWith' subStart f p + = do tv <- atomically $ newTVar Map.empty + (x, Trace (_,_,t)) <- runCHPProgramWith + (Trace (rootProcessId, tv, subStart)) + id p + `C.catch` const (return (Nothing, + Trace (undefined, undefined, subStart))) + l <- atomically $ readTVar tv + t' <- f l t + return (x, t') + +-- ========== +-- Instances: +-- ========== + +instance Error PoisonError where + noMsg = PoisonError () + +instance (Error e, MonadCHP m) => MonadCHP (ErrorT e m) where + liftCHP = lift . liftCHP + +instance MonadCHP m => MonadCHP (ReaderT r m) where + liftCHP = lift . liftCHP + +instance MonadCHP m => MonadCHP (StateT s m) where + liftCHP = lift . liftCHP + +instance (Monoid w, MonadCHP m) => MonadCHP (WriterT w m) where + liftCHP = lift . liftCHP + +instance MonadCHP CHP where + liftCHP = id + +-- The monad is lazy, and very similar to the writer monad +instance Monad CHP' where + -- m :: AltableT g m a + -- f :: a -> AltableT g m b + m >>= f = + let ~(grd, altBody) = getAltable m + nonAlt = getStandard m + altBody' = altBody >>= getStandard . f + nonAlt' = nonAlt >>= getStandard . f + in AltableT (grd, altBody') nonAlt' + return x = AltableT (badGuard, return x) (return x) + +instance MonadIO CHP' where + liftIO m = liftTrace (liftIO m) diff --git a/Control/Concurrent/CHP/BroadcastChannels.hs b/Control/Concurrent/CHP/BroadcastChannels.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/BroadcastChannels.hs @@ -0,0 +1,151 @@ +-- 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 broadcast channels (one-to-many). Whereas a one-to-any +-- channel features one writer sending a /single/ value to /one/ (of many) readers, a +-- one-to-many channel features one writer sending the /same/ value to /many/ +-- readers. So a one-to-any channel involves claiming the channel-end to ensure +-- exclusivity, but a one-to-many channel involves enrolling on the channel-end +-- (subscribing) before it can engage in communication. +-- +-- A communication on a one-to-many channel only takes place when the writer +-- and all readers currently enrolled agree to communicate. What happens when +-- the writer wants to communicate and no readers are enrolled is undefined +-- (the writer may block, or may communicate happily to no-one). +module Control.Concurrent.CHP.BroadcastChannels (BroadcastChanin, BroadcastChanout, + OneToManyChannel, AnyToManyChannel, oneToManyChannel, anyToManyChannel) where + +import Control.Concurrent.STM +import Control.Monad.Trans + +import Control.Concurrent.CHP.Barriers +import Control.Concurrent.CHP.Base +import Control.Concurrent.CHP.Channels +import Control.Concurrent.CHP.CSP +import Control.Concurrent.CHP.Enroll +import Control.Concurrent.CHP.Event +import Control.Concurrent.CHP.Mutex +import Control.Concurrent.CHP.Traces.Base + + +-- The general pattern of a broadcast channel is as follows: +-- SYNC -> Agreement; the readers indicate they are all willing to read, and the +-- writer indicates it is ready to write. Either side may ALT. +-- +-- After this synchronisation, the writer can write his data to the TVar, possibly +-- following an extended action +-- +-- SYNC -> Reading; everyone syncs (no-one ALTs) to move to the reading phase +-- +-- After this synchronisation, the readers can all read the data from the TVar, +-- and possibly complete an extended action +-- +-- SYNC -> Neutral; everyone syncs (no-one ALTs) to indicate one communication +-- cycle has finished. After this the writer may proceed on their way (the +-- main reason for needing a third sync). + +-- There used to be a warning that the first two constructors are never used, but they +-- do need to be there for the Enum and Bounded instances... +data Phase = Agreement | Reading | Neutral deriving (Enum, Bounded, Eq) +-- So I constructed this horrendous hack to suppress the warning: +dontWarnMe :: a -> a +dontWarnMe = flip const [Agreement, Reading, Neutral] + +newtype BroadcastChannel a = BC (PhasedBarrier Phase, TVar a) + +-- | The reading end of a broadcast channel. You must enroll on it before +-- you can read from it or poison it. +newtype BroadcastChanin a = BI (BroadcastChannel a) + +-- | The writing end of a broadcast channel. +newtype BroadcastChanout a = BO (BroadcastChannel a) + +instance Enrollable BroadcastChanin a where + enroll c@(BI (BC (b,_))) f = enroll b (\eb -> waitForPhase Neutral eb >> f (Enrolled c)) + resign (Enrolled (BI (BC (b,_)))) m + = do x <- resign (Enrolled b) m + waitForPhase Neutral (Enrolled b) + return x + +instance WriteableChannel BroadcastChanout where + extWriteChannel (BO (BC (b, tv))) m + = do syncBarrierWith (recAs (Just . ChannelComm) (Just . ChannelWrite)) + $ Enrolled b + m >>= liftIO . atomically . writeTVar tv + syncBarrierWith (recAs (const Nothing) (const Nothing)) + $ Enrolled b + syncBarrierWith (recAs (const Nothing) (const Nothing)) + $ Enrolled b + return () + +instance ReadableChannel (Enrolled BroadcastChanin) where + extReadChannel (Enrolled (BI (BC (b, tv)))) f + = do syncBarrierWith (recAs (Just . ChannelComm) (Just . ChannelRead)) + $ Enrolled b + syncBarrierWith (recAs (const Nothing) (const Nothing)) + $ Enrolled b + x <- liftIO (atomically $ readTVar tv) + y <- f x + syncBarrierWith (recAs (const Nothing) (const Nothing)) + $ Enrolled b + return y + +instance Poisonable (BroadcastChanout a) where + poison (BO (BC (b,_))) = poison $ Enrolled b + +instance Poisonable (Enrolled BroadcastChanin a) where + poison (Enrolled (BI (BC (b,_)))) = poison $ Enrolled b + +newBroadcastChannel :: CHP (BroadcastChannel a) +newBroadcastChannel = dontWarnMe {- see above -} $ do + do b@(Barrier (e,_)) <- newPhasedBarrier Neutral + -- Writer is always enrolled: + liftIO $ atomically $ enrollEvent e + tv <- liftIO $ atomically $ newTVar undefined + return $ BC (b, tv) + +instance Channel BroadcastChanin BroadcastChanout where + newChannel = liftCHP $ do + c@(BC (b,_)) <- newBroadcastChannel + return $ Chan (getBarrierIdentifier b) (BI c) (BO c) + +instance Channel BroadcastChanin (Shared BroadcastChanout) where + newChannel = liftCHP $ do + m <- newMutex + c <- newChannel + return $ Chan (getChannelIdentifier c) (reader c) (Shared (m, writer c)) + +type OneToManyChannel = Chan BroadcastChanin BroadcastChanout +type AnyToManyChannel = Chan BroadcastChanin (Shared BroadcastChanout) + +oneToManyChannel :: CHP (OneToManyChannel a) +oneToManyChannel = newChannel + +anyToManyChannel :: CHP (AnyToManyChannel a) +anyToManyChannel = newChannel diff --git a/Control/Concurrent/CHP/Buffers.hs b/Control/Concurrent/CHP/Buffers.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/Buffers.hs @@ -0,0 +1,121 @@ +-- 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. + + +-- | Various processes that act like buffers. Poisoning either end of a buffer +-- process is immediately passed on to the other side, in contrast to C++CSP2 +-- and JCSP. +module Control.Concurrent.CHP.Buffers (fifoBuffer, infiniteBuffer, overflowingBuffer, overwritingBuffer) + where + +import Data.Sequence (Seq, viewl, ViewL(..)) +import qualified Data.Sequence as Seq + +import Control.Concurrent.CHP +import qualified Control.Concurrent.CHP.Common as Common + +-- | Acts like a limited capacity FIFO buffer of the given size. When it is +-- full it accepts no input, and when it is empty it offers no output. +fifoBuffer :: forall a. Int -> Chanin a -> Chanout a -> CHP () +fifoBuffer n in_ out + | n < 0 = return () + | n == 0 = Common.id in_ out + | otherwise = fifo Seq.empty `onPoisonRethrow` (poison in_ >> poison out) + where + fifo :: Seq a -> CHP () + fifo s | Seq.null s = takeIn + | Seq.length s == n = sendOut + | otherwise = takeIn <-> sendOut + where + takeIn = readChannel in_ >>= fifo . addLast s + sendOut = do writeChannel out (seqHead s) + fifo (removeHead s) + +-- | Acts like a FIFO buffer with unlimited capacity. Use with caution; make +-- sure you do not let the buffer grow so large that it eats up all your memory. +-- When it is empty, it offers no output. It always accepts input. +infiniteBuffer :: forall a. Chanin a -> Chanout a -> CHP () +infiniteBuffer in_ out + = buff Seq.empty `onPoisonRethrow` (poison in_ >> poison out) + where + buff :: Seq a -> CHP () + buff s | Seq.null s = takeIn + | otherwise = sendOut takeIn + where + takeIn = readChannel in_ >>= buff . addLast s + sendOut = do writeChannel out (seqHead s) + buff (removeHead s) + +-- | Acts like a FIFO buffer of limited capacity, except that when it is full, +-- it always accepts input and discards it. When it is empty, it does not offer output. +overflowingBuffer :: forall a. Int -> Chanin a -> Chanout a -> CHP () +overflowingBuffer n in_ out + | n < 0 = return () + | n == 0 = Common.id in_ out + | otherwise = flow Seq.empty `onPoisonRethrow` (poison in_ >> poison out) + where + flow :: Seq a -> CHP () + flow s | Seq.null s = takeIn + | Seq.length s == n = sendOut <-> dropItem + | otherwise = takeIn <-> sendOut + where + takeIn = readChannel in_ >>= flow . addLast s + dropItem = readChannel in_ >> flow s + sendOut = do writeChannel out (seqHead s) + flow (removeHead s) + +-- | Acts like a FIFO buffer of limited capacity, except that when it is full, +-- it always accepts input and pushes out the oldest item in the buffer. When +-- it is empty, it does not offer output. +overwritingBuffer :: forall a. Int -> Chanin a -> Chanout a -> CHP () +overwritingBuffer n in_ out + | n < 0 = return () + | n == 0 = Common.id in_ out + | otherwise = over Seq.empty `onPoisonRethrow` (poison in_ >> poison out) + where + over :: Seq a -> CHP () + over s | Seq.null s = takeIn + | Seq.length s == n = sendOut <-> takeInOver + | otherwise = takeIn <-> sendOut + where + takeIn = readChannel in_ >>= over . addLast s + takeInOver = readChannel in_ >>= over . removeHead . addLast s + sendOut = do writeChannel out (seqHead s) + over (removeHead s) + +seqHead :: Seq a -> a +seqHead s = case viewl s of + EmptyL -> error "Internal code logic error in buffer" + x :< _ -> x + +removeHead :: Seq a -> Seq a +removeHead = Seq.drop 1 + +addLast :: Seq a -> a -> Seq a +addLast = (Seq.|>) diff --git a/Control/Concurrent/CHP/CSP.hs b/Control/Concurrent/CHP/CSP.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/CSP.hs @@ -0,0 +1,157 @@ +-- 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 a few miscellaneous items that can't go in Control.Concurrent.CHP.Base +-- because they would form a cyclic module link. Not publicly visible. +-- TODO rename this module. +module Control.Concurrent.CHP.CSP where + +import Control.Concurrent.STM +import Control.Exception +import Control.Monad.State +import Control.Monad.Writer +import Control.Monad.Trans +import Data.Unique +import System.IO + +import Control.Concurrent.CHP.Alt +import Control.Concurrent.CHP.Base +import qualified Control.Concurrent.CHP.Event as Event +import Control.Concurrent.CHP.Enroll +import Control.Concurrent.CHP.Guard +import Control.Concurrent.CHP.Mutex +import Control.Concurrent.CHP.Poison +import Control.Concurrent.CHP.Traces.Base + +recAs :: (Unique -> Maybe RecordedEvent) -> (Unique -> Maybe RecordedIndivEvent) -> Unique + -> RecEvents +recAs f g u = (f u, g u) + +-- First engages in event, then executes the body. The returned value is suitable +-- for use in an alt +buildOnEventPoison :: (Unique -> RecEvents) -> Event.Event -> STM () -> CHP a -> CHP a +buildOnEventPoison rec e act body + = liftPoison (AltableT (theGuard, getEventPoison True) + (return $ NoPoison False)) + >>= checkPoison >>= \b -> if b then body else + alt [liftPoison $ AltableT (theGuard, getEventPoison ()) (return $ NoPoison())] >>= + checkPoison >> body + where + theGuard = let Event.Event (u,_) = e in EventGuard [] (rec u) act e + + getEventPoison :: a -> TraceT IO (WithPoison a) + getEventPoison x + = do (ns,y) <- get + case ns of + [] -> return $ NoPoison x + _ -> do put ([],y) + return PoisonItem + +scopeBlock :: CHP a -> (a -> CHP b) -> IO () -> CHP b +scopeBlock start body errorEnd + = do x <- start + tr <- liftPoison $ liftTrace get + (y, tr') <- liftIO $ bracketOnError (return ()) (const errorEnd) $ const + $ runStateT (getStandard (wrapPoison $ body x)) tr + liftPoison $ liftTrace $ put tr' + checkPoison y + + +-- | Synchronises on the given barrier. You must be enrolled on a barrier in order +-- to synchronise on it. Returns the new phase, following the synchronisation. +syncBarrierWith :: (Enum phase, Bounded phase, Eq phase) => (Unique -> RecEvents) -> + Enrolled PhasedBarrier phase -> CHP phase +syncBarrierWith rec (Enrolled (Barrier (e,tv))) + = buildOnEventPoison rec e incPhase + (liftIO $ atomically $ readTVar tv) + where + incPhase :: STM () + incPhase = do p <- readTVar tv + writeTVar tv $ if p == maxBound then minBound else succ p + +-- | A phased barrier that is capable of being poisoned and throwing poison. +-- You will need to enroll on it to do anything useful with it. +-- For the phases you can use any type that satisfies Enum, Bounded and Eq. +-- The phase increments every time the barrier completes. Incrementing consists +-- of: @if p == maxBound then minBound else succ p@. Examples of things that +-- make sense for phases: +-- +-- * The () type (see the 'Barrier' type). This effectively has a single repeating +-- phase, and acts like a non-phased barrier. +-- +-- * A bounded integer type. This increments the count every time the barrier completes. +-- But don't forget that the count will wrap round when it reaches the end. +-- You cannot use 'Integer' for a phase because it is unbounded. If you really +-- want to have an infinitely increasing count, you can wrap 'Integer' in a newtype and +-- provide a Bounded instance for it (with minBound and maxBound set to -1, +-- if you start on 0). +-- +-- * A boolean. This implements a simple black-white barrier, where the state +-- flips on each iteration. +-- +-- * A custom data type that has only constructors. For example, @data MyPhases +-- = Discover | Plan | Move@. Haskell supports deriving Enum, Bounded and +-- Eq automatically on such types. +newtype (Enum phase, Bounded phase, Eq phase) => + PhasedBarrier phase = Barrier (Event.Event, TVar phase) + +instance (Enum phase, Bounded phase, Eq phase) => Enrollable PhasedBarrier phase + where + enroll b@(Barrier (e,_)) f + = do liftSTM (Event.enrollEvent e) >>= checkPoison + x <- f $ Enrolled b + liftSTM (Event.resignEvent e) >>= checkPoison + return x + + resign (Enrolled (Barrier (e,_))) m + = do liftSTM (Event.resignEvent e) >>= checkPoison + x <- m + liftSTM (Event.enrollEvent e) >>= checkPoison + return x + + +-- | A channel type, that can be used to get the ends of the channel via 'reader' +-- and 'writer' +data Chan r w a = Chan { + -- | Gets the channel's identifier. Useful if you need to be able to identify + -- a channel in the trace later on. + getChannelIdentifier :: Unique, + -- | Gets the reading end of a channel from its 'Chan' type. + reader :: r a, + -- | Gets the writing end of a channel from its 'Chan' type. + writer :: w a} + +instance (Enum phase, Bounded phase, Eq phase) => Poisonable (Enrolled PhasedBarrier phase) where + poison (Enrolled (Barrier (e,_))) + = liftSTM $ Event.poisonEvent e + +-- | A wrapper (usually around a channel-end) indicating that the inner item +-- is shared. Use the 'claim' function to use this type. +data Shared c a = Shared (Mutex, c a) + diff --git a/Control/Concurrent/CHP/Channels.hs b/Control/Concurrent/CHP/Channels.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/Channels.hs @@ -0,0 +1,391 @@ +-- 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. + + +-- | The module containing all the different types of channels in CHP. Unlike +-- JCSP and C++CSP2, CHP does not offer buffered channels directly (see the +-- "Control.Concurrent.CHP.Buffers" module). There are four different channel types, effectively +-- all possible combinations of: +-- +-- * Shared reader vs non-shared reader +-- +-- * Shared writer vs non-shared writer +-- +-- For most applications you probably want just 'OneToOneChannel'. +-- +-- It is possible for the type system to infer which channel you want when +-- you use 'newChannel'. If the types of the ends are known by the type system, +-- the channel-type can be inferred. So you can usually just write 'newChannel', +-- and depending on how you use the channel, the type system will figure out +-- which one you needed. +module Control.Concurrent.CHP.Channels ( + -- * Channel Creation + Chan, Channel(..), newChannelWithLabel, newChannelWR, newChannelRW, ChannelTuple(..), + newChannelList, newChannelListWithLabels, newChannelListWithStem, + getChannelIdentifier, + -- * Channel-Ends + Chanin, Chanout, + reader, writer, readers, writers, + + -- * Reading and Writing with Channels + ReadableChannel(..), WriteableChannel(..), + + -- * Shared Channels + claim, Shared, + + -- * Specific Channel Types + -- | All the functions here are equivalent to newChannel, but typed. So for + -- example, @oneToOneChannel = newChannel :: MonadCHP m => m OneToOneChannel@. + OneToOneChannel, oneToOneChannel, + OneToAnyChannel, oneToAnyChannel, + AnyToOneChannel, anyToOneChannel, + AnyToAnyChannel, anyToAnyChannel + ) + where + + +import Control.Concurrent.STM.TVar +import Control.Monad +import Control.Monad.STM +import Control.Monad.Trans +import Data.Maybe +import Data.Unique + +import Control.Concurrent.CHP.Base +import Control.Concurrent.CHP.CSP +import Control.Concurrent.CHP.Event +import Control.Concurrent.CHP.Monad +import Control.Concurrent.CHP.Mutex +import Control.Concurrent.CHP.Poison +import Control.Concurrent.CHP.Traces.Base + +-- ====== +-- Types: +-- ====== + +-- | A reading channel-end type that allows poison to be thrown +newtype Chanin a = Chanin (STMChannel a) + +-- | A writing channel-end type that allows poison to be thrown +newtype Chanout a = Chanout (STMChannel a) + +newtype STMChannel a = STMChan (Event, TVar (WithPoison (Maybe a))) + +type OneToOneChannel = Chan Chanin Chanout +type AnyToOneChannel = Chan (Chanin) (Shared Chanout) +type OneToAnyChannel = Chan (Shared Chanin) (Chanout) +type AnyToAnyChannel = Chan (Shared Chanin) (Shared Chanout) + +-- ======== +-- Classes: +-- ======== + +class ChaninC c a where + startReadChannelC :: c a -> (Event, STM (WithPoison a)) + endReadChannelC :: c a -> STM () + poisonReadC :: c a -> IO () + +class ChanoutC c a where + startWriteChannelC :: c a -> (Event, STM (WithPoison ())) + endWriteChannelC :: c a -> a -> STM () + poisonWriteC :: c a -> IO () + +-- | A class used for allocating new channels, and getting the reading and +-- writing ends. There is a bijective assocation between the channel, and +-- its pair of end types. You can see the types in the list of instances below. +-- Thus, newChannel may be used, and the compiler will infer which type of +-- channel is required based on what end-types you get from reader and writer. +-- Alternatively, if you explicitly type the return of newChannel, it will +-- be definite which ends you will use. If you do want to fix the type of +-- the channel you are using when you allocate it, consider using one of the +-- many 'oneToOneChannel'-like shorthand functions that fix the type. +class Channel r w where + -- | Allocates a new channel. Nothing need be done to + -- destroy\/de-allocate the channel when it is no longer in use. + newChannel :: MonadCHP m => m (Chan r w a) + +-- | A class indicating that a channel can be read from. +class ReadableChannel chanEnd where -- minimal implementation: extReadChannel + -- | Reads from the given reading channel-end + readChannel :: chanEnd a -> CHP a + readChannel c = extReadChannel c return + -- | Performs an extended read from the channel, performing the given action + -- before freeing the writer + extReadChannel :: chanEnd a -> (a -> CHP b) -> CHP b + +-- | A class indicating that a channel can be written to. +class WriteableChannel chanEnd where -- minimal implementation: extWriteChannel + -- | Writes from the given writing channel-end + writeChannel :: chanEnd a -> a -> CHP () + writeChannel c x = extWriteChannel c (return x) + + -- | Starts the communication, then performs the given extended action, then + -- sends the result of that down the channel + extWriteChannel :: chanEnd a -> CHP a -> CHP () + +-- | A helper class for easily creating several channels of the same type. +-- The same type refers not only to what type the channel carries, but +-- also to the type of channel (one-to-one no poison, one-to-any with +-- poison, etc). You can write code like this: +-- +-- > (a, b, c, d, e) <- newChannels +-- +-- To create five channels of the same type. +class ChannelTuple t where + newChannels :: MonadCHP m => m t + +-- ========== +-- Functions: +-- ========== + +chan :: Monad m => m (Unique, c a) -> (c a -> r a) -> (c a -> w a) -> m (Chan r w a) +chan m r w = do (u, x) <- m + return $ Chan u (r x) (w x) + + +waitForJustOrPoison :: TVar (WithPoison (Maybe a)) -> STM (WithPoison a) +waitForJustOrPoison tv = do x <- readTVar tv + case x of + PoisonItem -> return PoisonItem + NoPoison Nothing -> retry + NoPoison (Just y) -> return $ NoPoison y + + +-- | Like 'newChannel' but also associates a label with that channel in a +-- trace. You can use this function whether tracing is turned on or not, +-- so if you ever use tracing, you should use this rather than 'newChannel'. +newChannelWithLabel :: (Channel r w, MonadCHP m) => String -> m (Chan r w a) +newChannelWithLabel l + = do c <- newChannel + liftCHP . liftPoison . liftTrace $ labelUnique (getChannelIdentifier c) l + return c + + +-- | A helper that is like 'newChannel' but returns the reading and writing +-- end of the channels directly. +newChannelRW :: (Channel r w, MonadCHP m) => m (r a, w a) +newChannelRW = do c <- newChannel + return (reader c, writer c) + +-- | A helper that is like 'newChannel' but returns the writing and reading +-- end of the channels directly. +newChannelWR :: (Channel r w, MonadCHP m) => m (w a, r a) +newChannelWR = do c <- newChannel + return (writer c, reader c) + +-- | Creates a list of channels of the same type with the given length. If +-- you need to access some channels by index, use this function. Otherwise +-- you may find using 'newChannels' to be easier. +newChannelList :: (Channel r w, MonadCHP m) => Int -> m [Chan r w a] +newChannelList n = replicateM n newChannel + +-- | A helper that is like 'newChannelList', but labels the channels according +-- 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 [newChannelWithLabel (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 newChannelWithLabel + +-- | Gets all the reading ends of a list of channels. A shorthand for @map +-- reader@. +readers :: [Chan r w a] -> [r a] +readers = map reader + +-- | Gets all the writing ends of a list of channels. A shorthand for @map +-- writer@. +writers :: [Chan r w a] -> [w a] +writers = map writer + +stmChannel :: MonadIO m => m (Unique, STMChannel a) +stmChannel = liftIO $ + do e@(Event (u,_)) <- newEvent 2 + c <- atomically $ newTVar $ NoPoison Nothing + return (u, STMChan (e,c)) + +oneToOneChannel :: MonadCHP m => m (OneToOneChannel a) +oneToOneChannel = newChannel + +-- | Claims the given channel-end, executes the given block, then releases +-- the channel-end and returns the output value. If poison or an IO +-- exception is thrown inside the block, the channel is released and the +-- poison\/exception re-thrown. +claim :: Shared c a -> (c a -> CHP b) -> CHP b +claim (Shared (lock, c)) body + = scopeBlock + (claimMutex lock >> return c) + (\y -> do x <- body y + liftIO $ releaseMutex lock + return x) + (releaseMutex lock) + +anyToOneChannel :: MonadCHP m => m (AnyToOneChannel a) +anyToOneChannel = newChannel + +oneToAnyChannel :: MonadCHP m => m (OneToAnyChannel a) +oneToAnyChannel = newChannel + +anyToAnyChannel :: MonadCHP m => m (AnyToAnyChannel a) +anyToAnyChannel = newChannel + +-- ========== +-- Instances: +-- ========== + +instance ReadableChannel Chanin where + readChannel (Chanin c) + = let (e, m) = startReadChannelC c in + buildOnEventPoison (recAs (Just . ChannelComm) (Just . ChannelRead)) e (return ()) (liftSTM $ + do x <- m + endReadChannelC c + return x) >>= checkPoison + + extReadChannel (Chanin c) body + = let (e, m) = startReadChannelC c in + scopeBlock + (buildOnEventPoison (recAs (Just . ChannelComm) (Just . ChannelRead)) e (return ()) (liftSTM m) >>= checkPoison) + (\val -> do x <- body val + liftSTM $ endReadChannelC c + return x) + (poisonReadC c) + +instance WriteableChannel Chanout where + writeChannel (Chanout c) x + = let (e, m) = startWriteChannelC c in + buildOnEventPoison (recAs (Just . ChannelComm) (Just . ChannelWrite)) e (return ()) + (liftSTM $ do y <- m + endWriteChannelC c x + return y) + >>= checkPoison + extWriteChannel (Chanout c) body + = let (e, m) = startWriteChannelC c in + scopeBlock + (buildOnEventPoison (recAs (Just . ChannelComm) (Just . ChannelWrite)) + e (return ()) (liftSTM m) >>= checkPoison) + (const $ body >>= liftSTM . endWriteChannelC c) + (poisonWriteC c) + + +instance Poisonable (Chanin a) where + poison (Chanin c) = liftIO $ poisonReadC c + +instance Poisonable (Chanout a) where + poison (Chanout c) = liftIO $ poisonWriteC c + +instance (Channel r w) => ChannelTuple (Chan r w a, Chan r w a) where + newChannels = do c0 <- newChannel + c1 <- newChannel + return (c0, c1) + +instance (Channel r w) => ChannelTuple (Chan r w a, Chan r w a, Chan r w a) where + newChannels = do c0 <- newChannel + c1 <- newChannel + c2 <- newChannel + return (c0, c1, c2) + +instance (Channel r w) => ChannelTuple (Chan r w a, Chan r w a, Chan r w a, + Chan r w a) where + newChannels = do c0 <- newChannel + c1 <- newChannel + c2 <- newChannel + c3 <- newChannel + return (c0, c1, c2, c3) + +instance (Channel r w) => ChannelTuple (Chan r w a, Chan r w a, Chan r w a, + Chan r w a, Chan r w a) where + newChannels = do c0 <- newChannel + c1 <- newChannel + c2 <- newChannel + c3 <- newChannel + c4 <- newChannel + return (c0, c1, c2, c3, c4) + +instance (Channel r w) => ChannelTuple (Chan r w a, Chan r w a, Chan r w a, + Chan r w a, Chan r w a, Chan r w a) where + newChannels = do c0 <- newChannel + c1 <- newChannel + c2 <- newChannel + c3 <- newChannel + c4 <- newChannel + c5 <- newChannel + return (c0, c1, c2, c3, c4, c5) + +-- Some of this is defensive programming -- the writer should never be able +-- to discover poison in the channel variable, for example + +instance ChaninC STMChannel a where + startReadChannelC (STMChan (e,tv)) = (e, waitForJustOrPoison tv) + endReadChannelC (STMChan (_,tv)) + = do x <- readTVar tv + case x of + PoisonItem -> return () + NoPoison _ -> writeTVar tv $ NoPoison Nothing + poisonReadC (STMChan (e,tv)) + = liftSTM $ do poisonEvent e + writeTVar tv PoisonItem + +instance ChanoutC STMChannel a where + startWriteChannelC (STMChan (e,tv)) + = (e, do x <- readTVar tv + case x of + PoisonItem -> return PoisonItem + NoPoison _ -> return $ NoPoison ()) + endWriteChannelC (STMChan (_, tv)) val + = do x <- readTVar tv + case x of + PoisonItem -> return () + NoPoison _ -> do writeTVar tv $ NoPoison $ Just val + + poisonWriteC (STMChan (e,tv)) + = liftSTM $ do poisonEvent e + writeTVar tv PoisonItem + +instance Channel Chanin Chanout where + newChannel = chan stmChannel Chanin Chanout + +instance Channel (Shared Chanin) Chanout where + newChannel = do m <- newMutex + c <- newChannel + return $ Chan (getChannelIdentifier c) (Shared (m, reader c)) (writer c) + +instance Channel Chanin (Shared Chanout) where + newChannel = do m <- newMutex + c <- newChannel + return $ Chan (getChannelIdentifier c) (reader c) (Shared (m, writer c)) + +instance Channel (Shared Chanin) (Shared Chanout) where + newChannel = do m <- newMutex + m' <- newMutex + c <- newChannel + return $ Chan (getChannelIdentifier c) (Shared (m, reader c)) (Shared (m', writer c)) + diff --git a/Control/Concurrent/CHP/Common.hs b/Control/Concurrent/CHP/Common.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/Common.hs @@ -0,0 +1,195 @@ +-- 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 collection of useful common processes that are useful when plumbing +-- together a process network. All the processes here rethrow poison when +-- it is encountered, as this gives the user maximum flexibility (they can +-- let it propagate it, or ignore it). +-- +-- The names here overlap with standard Prelude names. This is +-- deliberate, as the processes act in a similar manner to the +-- corresponding Prelude versions. It is expected that you will do +-- something like: +-- +-- > import qualified Control.Concurrent.CHP.Common as Common +-- +-- or: +-- +-- > import qualified Control.Concurrent.CHP.Common as CHP +-- +-- to circumvent this problem. +module Control.Concurrent.CHP.Common where + +import Control.Monad +import qualified Data.Traversable as Traversable +import Prelude (Bool, Maybe(..), Enum, Ord, ($), (<), Int, otherwise, (.)) +import qualified Prelude + +import Control.Concurrent.CHP + +-- | Forever forwards the value onwards, unchanged. Adding this to your process +-- network effectively adds a single-place buffer. +id :: Chanin a -> Chanout a -> CHP () +id in_ out = (forever $ + do x <- readChannel in_ + writeChannel out x + ) `onPoisonRethrow` (poison in_ >> poison out) + +-- | Forever forwards the value onwards, in an extended rendezvous. This is +-- like 'id' but does not add any buffering to your network. +-- +-- extId is a unit of the associative operator '|->|'. +extId :: Chanin a -> Chanout a -> CHP () +extId in_ out = tap in_ [out] + +-- | A process that waits for an input, then sends it out on /all/ its output +-- channels (in order) during an extended rendezvous. This is often used to send the +-- output on to both the normal recipient (without introducing buffering) and +-- also to a listener process that wants to examine the value. If the listener +-- process is first in the list, and does not take the input immediately, the +-- value will not be sent to the other recipients until it does. The name +-- of the process derives from the notion of a wire-tap, since the listener +-- is hidden from the other processes (it does not visibly change the semantics +-- for them). +tap :: Chanin a -> [Chanout a] -> CHP () +tap in_ outs = (forever $ + extReadChannel in_ + (\x -> mapM_ (Prelude.flip writeChannel x) outs) + ) `onPoisonRethrow` (poison in_ >> poisonAll outs) + +-- | Sends out a single value first (the prefix) then behaves like id. +prefix :: a -> Chanin a -> Chanout a -> CHP () +prefix x in_ out = (writeChannel out x >> id in_ out) + `onPoisonRethrow` (poison in_ >> poison out) + +-- | Forever reads in a value, and then sends out its successor (using 'Prelude.succ'). +succ :: Enum a => Chanin a -> Chanout a -> CHP () +succ = map Prelude.succ + +-- | Reads in a value, and sends it out in parallel on all the given output +-- channels. +parDelta :: Chanin a -> [Chanout a] -> CHP () +parDelta in_ outs = (forever $ + do x <- readChannel in_ + runParallel_ $ Prelude.map (Prelude.flip writeChannel x) outs + ) `onPoisonRethrow` (poison in_ >> mapM_ poison outs) + +-- | Forever reads in a value, transforms it using the given function, and sends it +-- out again. Note that the transformation is not applied strictly, so don't +-- assume that this process will actually perform the computation. +map :: (a -> b) -> Chanin a -> Chanout b -> CHP () +map f in_ out = forever (readChannel in_ >>= (return . f) >>= writeChannel out) + `onPoisonRethrow` (poison in_ >> poison out) + +-- | Forever reads in a value, and then based on applying the given function +-- either discards it (if the function returns false) or sends it on (if +-- the function returns True). +filter :: (a -> Bool) -> Chanin a -> Chanout a -> CHP () +filter f in_ out = forever (do + x <- readChannel in_ + when (f x) (writeChannel out x) + ) `onPoisonRethrow` (poison in_ >> poison out) + +-- | Streams all items in a Traversable container out in the order given by +-- 'Traversable.mapM' on the output channel (one at a time). Lists, Maybe, +-- and Set are all instances of Traversable, so this can be used for all of +-- those. +stream :: Traversable.Traversable t => Chanin (t a) -> Chanout a -> CHP () +stream in_ out = (forever $ do + xs <- readChannel in_ + Traversable.mapM (writeChannel out) xs) + `onPoisonRethrow` (poison in_ >> poison out) + +-- | Forever waits for input from one of its many channels and sends it +-- out again on the output channel. +merger :: [Chanin a] -> Chanout a -> CHP () +merger ins out = (forever $ alt (Prelude.map readChannel ins) >>= writeChannel out) + `onPoisonRethrow` (poisonAll ins >> poison out) + +-- | Sends out the specified value on the given channel the specified number +-- of times, then finishes. +replicate :: Int -> a -> Chanout a -> CHP () +replicate n x c = replicateM_ n (writeChannel c x) `onPoisonRethrow` poison c + +-- | Forever sends out the same value on the given channel, until poisoned. +-- Similar to the white-hole processes in some other frameworks. +repeat :: a -> Chanout a -> CHP () +repeat x c = (forever $ writeChannel c x) `onPoisonRethrow` poison c + +-- | Forever reads values from the channel and discards them, until poisoned. +-- Similar to the black-hole processes in some other frameworks. +consume :: Chanin a -> CHP () +consume c = (forever $ readChannel c) `onPoisonRethrow` poison c + +-- | Forever reads a value from both its input channels in parallel, then joins +-- the two values using the given function and sends them out again. For example, +-- @join (,) c d@ will pair the values read from @c@ and @d@ and send out the +-- pair on the output channel, whereas @join (&&)@ will send out the conjunction +-- of two boolean values, @join (==)@ will read two values and output whether +-- they are equal or not, etc. +join :: (a -> b -> c) -> Chanin a -> Chanin b -> Chanout c -> CHP () +join f in0 in1 out = (forever $ do + (x,y) <- readChannel in0 <||> readChannel in1 + writeChannel out $ f x y + ) `onPoisonRethrow` (poison in0 >> poison in1 >> poison out) + + +-- | A sorter process. When it receives its first @Just x@ data item, it keeps +-- it. When it receieves a second, it keeps the lowest of the two, and sends +-- out the other one. When it receives Nothing, it sends out its data value, +-- then sends Nothing too. The overall effect when chaining these things together +-- is a sorting pump. You inject all the values with Just, then send in a +-- single Nothing to get the results out (in reverse order). +sorter :: Ord a => Chanin (Maybe a) -> Chanout (Maybe a) -> CHP () +sorter = sorter' (<) + +-- | Like sorter, but with a custom comparison method. You should pass in +-- the equivalent of less-than: (<). +sorter' :: forall a. (a -> a -> Bool) -> Chanin (Maybe a) -> Chanout (Maybe a) -> CHP () +sorter' lt in_ out = internal Nothing `onPoisonRethrow` (poison in_ >> poison out) + where + internal :: Maybe a -> CHP () + internal curVal = do newVal <- readChannel in_ + case (curVal, newVal) of + -- Flush, but we're empty: + (Nothing, Nothing) -> do writeChannel out newVal + internal curVal + -- Flush: + (Just _, Nothing) -> do writeChannel out curVal + writeChannel out newVal + internal curVal + -- New value, we were empty: + (Nothing, Just _) -> internal newVal + -- New value, we had one already: + (Just cur, Just new) + | new `lt` cur -> do writeChannel out curVal + internal newVal + | otherwise -> do writeChannel out newVal + internal curVal + diff --git a/Control/Concurrent/CHP/Console.hs b/Control/Concurrent/CHP/Console.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/Console.hs @@ -0,0 +1,101 @@ +-- 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. + +-- | Contains a process for easily using stdin, stdout and stderr as channels. +module Control.Concurrent.CHP.Console where + +import Control.Concurrent +import Control.Concurrent.STM +import Control.Exception (Exception(..)) +import qualified Control.Exception as C +import Control.Monad +import Control.Monad.Trans +import Data.Maybe +import System.IO + +import Control.Concurrent.CHP + +-- | A set of channels to be given to the process to run, containing channels +-- for stdin, stdout and stderr. +data ConsoleChans = ConsoleChans { cStdin :: Chanin Char, cStdout :: Chanout + Char, cStderr :: Chanout Char } + +-- | A function for running the given CHP process that wants console channels. +-- When your program finishes, the console channels are automatically poisoned, +-- but it's good practice to poison them yourself when you finish. Only ever +-- run one of these processes at a time, or undefined behaviour will result. +-- +-- Note: getting the input handler to terminate is surpisingly difficult +-- in Haskell. Currently it seems (version 6.8.2) that GHC compiling the +-- end program with -threaded requires an extra character to be input +-- before the input handler notices poison, whereas GHC compiled (not +-- using -threaded) and GHCi act as intended. This is not a problem with +-- the whole of the library, only with this consoleProcess. Unfortunately +-- this means the one or two poison examples in the tutorial may not +-- function correctly. I hope to resolve this problem in the next +-- version. +consoleProcess :: (ConsoleChans -> CHP ()) -> CHP () +consoleProcess mainProc + = do [cin, cout, cerr] <- replicateM 3 oneToOneChannel + tvs@[tvinId, tvoutId, tverrId] <- liftIO $ atomically $ replicateM 3 $ newTVar Nothing + runParallel_ + [ inHandler tvinId (writer cin) + , outHandler tvoutId stdout (reader cout) + , outHandler tverrId stderr (reader cerr) + , do ids <- mapM getId tvs + (mainProc $ ConsoleChans (reader cin) (writer cout) (writer cerr)) + `onPoisonTrap` (return ()) + poison (reader cin) + poison (writer cout) + poison (writer cerr) + -- Poison won't do it if the handlers are blocked on input or + -- output. Therefore we throw them an exception to "knock them + -- off" their current action and make them exit. + liftIO $ mapM_ (flip throwTo BlockedIndefinitely) ids + ] + where + getId :: TVar (Maybe a) -> CHP a + getId tv = liftIO $ atomically $ readTVar tv >>= maybe retry return + + -- Like liftIO, but turns any caught exceptions into throwing poison + liftIO' :: IO a -> CHP a + liftIO' m = liftIO (liftM Just m `C.catch` const (return Nothing)) + >>= maybe throwPoison return + + inHandler :: TVar (Maybe ThreadId) -> Chanout Char -> CHP () + inHandler tv c + = do liftIO $ myThreadId >>= atomically . writeTVar tv . Just + (forever $ liftIO' getChar >>= writeChannel c) + `onPoisonTrap` (poison c) + + outHandler :: TVar (Maybe ThreadId) -> Handle -> Chanin Char -> CHP () + outHandler tv h c + = do liftIO $ myThreadId >>= atomically . writeTVar tv . Just + (forever $ readChannel c >>= liftIO' . hPutChar h) + `onPoisonTrap` (poison c) diff --git a/Control/Concurrent/CHP/Enroll.hs b/Control/Concurrent/CHP/Enroll.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/Enroll.hs @@ -0,0 +1,66 @@ +-- 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 with support for things that are enrollable (barriers and broadcast +-- channels). + +module Control.Concurrent.CHP.Enroll (Enrolled, Enrollable(..), enrollPair, enrollList) where + +import Control.Concurrent.CHP.Base + +class Enrollable b z where + -- | Enrolls on the given barrier, then executes the given function (passing + -- it in the enrolled barrier) and when that function body finishes, resigns + -- from the barrier and returns the result. If a poison or IO exception is thrown + -- in the inside block, the barrier will be correctly resigned from, and + -- the exception rethrown. + -- + -- Do not attempt to return the + -- enrolled barrier out of the inner function and use it afterwards. + enroll :: b z -> (Enrolled b z -> CHP a) -> CHP a + + -- | Resigns from a barrier, then executes the given action, and when the + -- action finishes, re-enrolls on the barrier and continues. This function + -- is designed for use from /inside/ the body of the 'enroll' function, to + -- temporarily resign from a barrier, do some things, then re-enroll. Do + -- not use the enrolled barrier inside the resign block. However, you may + -- enroll on the barrier inside this, nesting enroll and resign blocks as + -- much as you like + resign :: Enrolled b z -> CHP a -> CHP a + +-- | Like 'enroll' but enrolls on the given pair of barriers +enrollPair :: (Enrollable b p, Enrollable b' p') => (b p, b' p') -> ((Enrolled + b p, Enrolled b' p') -> CHP a) -> CHP a +enrollPair (b0,b1) f = enroll b0 $ \eb0 -> enroll b1 $ \eb1 -> f (eb0, eb1) + +-- | Like 'enroll' but enrolls on the given list of barriers +enrollList :: Enrollable b p => [b p] -> ([Enrolled b p] -> CHP a) -> CHP a +enrollList [] f = f [] +enrollList (b:bs) f = enroll b $ \eb -> enrollList bs $ \ebs -> f (eb:ebs) + diff --git a/Control/Concurrent/CHP/Event.hs b/Control/Concurrent/CHP/Event.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/Event.hs @@ -0,0 +1,160 @@ +-- 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 where + +import Control.Concurrent.STM +import Control.Monad +import Data.List +import Data.Maybe +import Data.Unique + +import Control.Concurrent.CHP.Poison +import Control.Concurrent.CHP.ProcessId + +-- Not really a CSP event, more like an enrollable poisonable alting barrier! +newtype Event = Event ( + Unique, -- Event identifier + TVar (WithPoison + (Int, -- Enrolled count + [OfferSet]) -- A list of offer sets + )) + +type OfferSet = (TVar (Maybe [Int]) -- Variable to use to signal when committed + , [Int] -- Value to send when committed + , ProcessId -- Id of the process making the offer + , [Event]) -- A list of all events currently offered + +newEvent :: Int -> IO Event +newEvent n + = do u <- newUnique + atomically $ do tv <- newTVar (NoPoison (n, [])) + return $ Event (u, tv) + +enrollEvent :: Event -> STM (WithPoison ()) +enrollEvent (Event (_, tv)) + = do x <- readTVar tv + case x of + PoisonItem -> return PoisonItem + NoPoison (count, offers) -> + do writeTVar tv $ NoPoison (count + 1, offers) + return $ NoPoison () + +resignEvent :: Event -> STM (WithPoison ()) +resignEvent (Event (_, tv)) + = do x <- readTVar tv + case x of + PoisonItem -> return PoisonItem + NoPoison (count, offers) -> + do writeTVar tv $ NoPoison (count - 1, offers) + if (count - 1 == length offers) + then completeEvent False tv + else return $ NoPoison () + +retractOffers :: [(TVar (Maybe [Int]), [Event])] -> STM () +retractOffers = mapM_ retractAll + where + retractAll :: (TVar (Maybe [Int]), [Event]) -> STM () + retractAll (tvid, evts) = mapM_ retract evts + where + retract :: Event -> STM () + retract (Event (_,tv)) + = do x <- readTVar tv + case x of + PoisonItem -> return () + NoPoison (enrolled, offers) -> + let reducedOffers = filter (\(tvx,_,_,_) -> tvx /= tvid) offers in + writeTVar tv $ NoPoison (enrolled, reducedOffers) + +-- Takes True to poison the event, False to complete normally +completeEvent :: Bool -> TVar (WithPoison (Int, [OfferSet])) -> STM (WithPoison ()) +completeEvent addPoison tv + = do x <- readTVar tv + case (x, addPoison) of + (PoisonItem, _) -> return PoisonItem + (NoPoison (_, offers), False) -> + do retractOffers $ [(tvw, events) | (tvw,_,_,events) <- offers] + sequence_ [writeTVar tvw (Just wx) | (tvw, wx, _,_) <- offers] + return $ NoPoison () + (NoPoison (_, offers), True) -> + do retractOffers $ [(tvw, events) | (tvw,_,_,events) <- offers] + sequence_ [writeTVar tvw (Just $ wx ++ [0]) | (tvw, wx, _, _) <- offers] + writeTVar tv PoisonItem + return PoisonItem + +-- Passed: True if allowed to commit to waiting +-- Returns: True if committed, False otherwise +enableEvents :: forall a. TVar (Maybe [Int]) -> ProcessId -> [(a, [Int], STM + (), Event)] -> Bool -> STM (Maybe (a, [ProcessId], [Int])) +enableEvents tvNotify pid events canCommitToWait + = do x <- checkAll + case (x, canCommitToWait) of + (Just (labels, pids, ns, act, Event (_,tv)), _) -> + act >> completeEvent False tv >>= \b -> + return $ Just (labels, pids, case b of {PoisonItem -> ns ++ [0]; _ -> ns}) + (Nothing, False) -> return Nothing + (Nothing, True) -> + do sequence_ [do NoPoison (count, otherOffers) <- readTVar ab + writeTVar ab $ NoPoison (count,(tvNotify, ns, pid, + map fourth events):otherOffers) + | (_, ns, _, Event (_,ab)) <- events] + return Nothing + where + fourth (_,_,_,c) = c + + checkAll :: STM (Maybe (a, [ProcessId], [Int], STM (), Event)) + checkAll = do xs <- sequence [do x <- readTVar tv + return (x, evt) + | evt@(_,_,_,Event (_,tv)) <- events] + return $ fmap get $ find (ready . fst) xs + + where + get :: (WithPoison (Int, [OfferSet]), (a, [Int], STM (), Event)) + -> (a, [ProcessId], [Int], STM (), Event) + get (PoisonItem,(l,ns,act,e)) = (l,[pid],ns,act,e) + get (NoPoison (_,offers),(l,ns,act,e)) = (l,pid : [p | (_,_,p,_) <- offers],ns,act,e) + + -- Sees if the barrier is ready *if* we commit to it too + ready :: WithPoison (Int, [OfferSet]) -> Bool + ready PoisonItem = True + ready (NoPoison (count, offers)) = count == 1 + length offers + +disableEvents :: TVar (Maybe [Int]) -> [Event] -> STM (Maybe [Int]) +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, events)] + return x + +poisonEvent :: Event -> STM () +poisonEvent (Event (_,tv)) = completeEvent True tv >> return () + +--TODO document how if it's poisoned, 0 will be appended to the list diff --git a/Control/Concurrent/CHP/Guard.hs b/Control/Concurrent/CHP/Guard.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/Guard.hs @@ -0,0 +1,68 @@ +-- 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. + +module Control.Concurrent.CHP.Guard where + +import Control.Concurrent.STM +import Control.Monad.Trans +import System.IO + +import Control.Concurrent.CHP.Event +import Control.Concurrent.CHP.Traces.Base + +-- Setup (giving transaction) +data Guard = TimeoutGuard (IO (STM [Int])) + | SkipGuard [Int] + | StopGuard + | BadGuard + -- The STM item is an action to take in the same transaction as + -- completing the event (before it is completed). + | EventGuard [Int] RecEvents (STM ()) Event + | NestedGuards [Guard] + +skipGuard :: Guard +skipGuard = SkipGuard [] + +badGuard :: Guard +badGuard = BadGuard + +stopGuard :: Guard +stopGuard = StopGuard + +-- Microseconds +guardWaitFor :: Int -> Guard +guardWaitFor n + = TimeoutGuard $ + do signalDone <- liftIO $ registerDelay n + return $ do b <- readTVar signalDone + if b == False + then retry + else return [] + + diff --git a/Control/Concurrent/CHP/Monad.hs b/Control/Concurrent/CHP/Monad.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/Monad.hs @@ -0,0 +1,151 @@ +-- 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. + +-- | This module contains all the central monads in the CHP library. +module Control.Concurrent.CHP.Monad + ( + -- * CHP Monad + CHP, MonadCHP(..), runCHP, runCHP_, + + onPoisonTrap, onPoisonRethrow, throwPoison, Poisonable(..), poisonAll, + + -- * LoopWhileT Monad + LoopWhileT, loop, while, + + -- * Primitive actions + skip, stop, waitFor + ) where + +import Control.Concurrent +import Control.Monad.Error +import Control.Monad.State +import Control.Monad.Trans + +-- This module primarily re-exports the public definitions from +-- Control.Concurrent.CHP.{Base,CSP,Poison}: + +import Control.Concurrent.CHP.Base +import Control.Concurrent.CHP.Guard +import Control.Concurrent.CHP.Traces.TraceOff + +-- | Runs a CHP program. You should use this once, at the top-level of your +-- program. Do not ever use this function twice in parallel and attempt to +-- communicate between those processes using channels. Instead, run this function +-- once and use it to spawn off the parallel processes that you need. +runCHP :: CHP a -> IO (Maybe a) +runCHP = liftM fst . (runCHPAndTrace :: CHP a -> IO (Maybe a, TraceOff)) + +-- | Runs a CHP program. Like 'runCHP' but discards the output. +runCHP_ :: CHP a -> IO () +runCHP_ p = runCHP p >> return () + +-- | A monad transformer for easier looping. This is independent of the +-- CHP aspects, but has all the right type-classes defined for it to make +-- it easy to use with the CHP library. +newtype Monad m => LoopWhileT m a = LWT { getLoop :: m (Maybe a) } + +instance Monad m => Monad (LoopWhileT m) where + -- m :: RW (Maybe (m a)) + -- f :: a -> RW (Maybe (m b)) + m >>= f = LWT $ do x <- getLoop m + case x of + Nothing -> return Nothing + Just m' -> getLoop $ f m' + return x = LWT $ return $ Just x + +instance MonadTrans LoopWhileT where + lift m = LWT $ m >>= return . Just + +instance MonadIO m => MonadIO (LoopWhileT m) where + liftIO = lift . liftIO + +instance MonadCHP m => MonadCHP (LoopWhileT m) where + liftCHP = lift . liftCHP + +instance MonadError e m => MonadError e (LoopWhileT m) where + throwError e = lift $ throwError e + catchError m h = LWT $ catchError (getLoop m) (getLoop . h) + +--TODO instances for all the other monad transformers + +-- | Runs the given action in a loop, executing it repeatedly until a 'while' +-- statement inside it has a False condition. If you use 'loop' without 'while', +-- the effect is the same as 'forever'. +loop :: Monad m => LoopWhileT m a -> m () +loop l = do x <- getLoop l + case x of + Nothing -> return () + Just _ -> loop l + +-- | Continues executing the loop if the given value is True. If the value +-- is False, the loop is broken immediately, and control jumps back to the +-- next action after the outer 'loop' statement. Thus you can build pre-condition, +-- post-condition, and "mid-condition" loops, placing the condition wherever +-- you like. +while :: Monad m => Bool -> LoopWhileT m () +while b = LWT $ if b then (return $ Just ()) else return Nothing + + +-- | Waits for the specified number of microseconds (millionths of a second). +-- There is no guaranteed precision, but the wait will never complete in less +-- time than the parameter given. +-- +-- Suitable for use in an 'alt', but note that waitFor 0 is not the same +-- as skip. 'waitFor' 0 '' x will not always select the first guard, +-- depending on x. Included in this is the lack of guarantee that +-- 'waitFor' 0 '' 'waitFor' n will select the first guard for any value +-- of n (including 0). It is not useful to use two waitFor guards in a +-- single 'alt' anyway. +waitFor :: Int -> CHP () +waitFor n = liftPoison $ AltableT (guardWaitFor n, return ()) (liftIO $ threadDelay n) +-- TODO maybe fix the above lack of guarantees by keeping timeout guards explicit. + +-- TODO add waitUntil + +-- | The classic skip process\/guard. Does nothing, and is always ready. +-- +-- Suitable for use in an 'alt'. +skip :: CHP () +skip = liftPoison $ AltableT (skipGuard, return ()) (return ()) + +-- | The stop guard. Its main use is that it is never ready in a choice, so +-- can be used to mask out guards. If you actually execute stop, that process +-- will do nothing more. Any parent process waiting for it to complete will +-- wait forever. +stop :: CHP () +stop = liftPoison $ AltableT (stopGuard, liftIO hang) (liftIO hang) + where + -- Strangely, I can't work out a good way to actually implement stop. + -- If you wait on a variable that will never be ready, GHC will wake + -- you up with an exception. If you loop doing that, you'll burn the + -- CPU. Throwing an exception would be caught and terminate the + -- 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 () + hang = forever $ threadDelay maxBound diff --git a/Control/Concurrent/CHP/Mutex.hs b/Control/Concurrent/CHP/Mutex.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/Mutex.hs @@ -0,0 +1,44 @@ +-- 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. + +module Control.Concurrent.CHP.Mutex where + +import Control.Concurrent +import Control.Monad.Trans + +type Mutex = MVar () + +claimMutex :: MonadIO m => Mutex -> m () +claimMutex m = liftIO $ takeMVar m + +newMutex :: MonadIO m => m Mutex +newMutex = liftIO $ newMVar () + +releaseMutex :: MonadIO m => Mutex -> m () +releaseMutex m = liftIO $ putMVar m () diff --git a/Control/Concurrent/CHP/Parallel.hs b/Control/Concurrent/CHP/Parallel.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/Parallel.hs @@ -0,0 +1,173 @@ +-- 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. + +module Control.Concurrent.CHP.Parallel (runParallel, runParallel_, (<||>), + ForkingT, forking, fork) where + +import Control.Concurrent +import Control.Concurrent.STM +import qualified Control.Exception as C +import Control.Monad.Error +import Control.Monad.Reader +import Control.Monad.State +import Data.List +import Data.Maybe +import Data.Ord + +import Control.Concurrent.CHP.Base +import Control.Concurrent.CHP.Traces.Base + +-- | This type-class supports parallel composition of processes. You may use +-- the 'runParallel' function to run a list of processes, or the '<||>' operator +-- to run just a pair of processes. +-- +-- In each case, the composition waits for all processes to finish, either +-- successfully or with poison. At the end of this, if /any/ process +-- exited with poison, the composition will "rethrow" this poison. If all +-- the processes completed successfully, the results will be returned. If +-- you want to ignore poison from the sub-processes, use an empty poison +-- handler and 'onPoisonTrap' with each branch. + +-- | Runs the given list of processes in parallel, and then returns a list +-- of the results, in the same order as the processes given. Will only return +-- when all the inner processes have completed. +runParallel :: [CHP a] -> CHP [a] +runParallel = runParallelPoison + +-- | A useful operator for performing a two process equivalent of runParallel +-- that gives the return values back as a pair rather than a list. This also +-- allows the values to have different types +(<||>) :: CHP a -> CHP b -> CHP (a, b) +(<||>) p q = do [x, y] <- runParallel [liftM Left p, liftM Right q] + combine x y + where + combine :: Monad m => Either a b -> Either a b -> m (a, b) + combine (Left x) (Right y) = return (x, y) + combine (Right y) (Left x) = return (x, y) + -- An extra case to keep the compiler happy: + combine _ _ = error "Impossible combination values in <|^|>" + + +-- | Runs all the given processes in parallel and discards any output. Does +-- not return until all the processes have completed. runParallel_ ps is effectively equivalent +-- to 'runParallel' ps >> return (). +runParallel_ :: [CHP a] -> CHP () +runParallel_ procs = runParallel procs >> return () + +-- We right associate to allow the liftM fst ((readResult) <||> runParallel_ +-- workers) pattern +infixr <||> + + +wrapProcess :: CHP a -> (CHP' (Either PoisonError a) -> IO (Either PoisonError + a, st)) -> IO (Maybe (Either st (a, st))) +wrapProcess (PoisonT proc) unwrapInner + = do let inner = runErrorT proc + x <- liftM Just (unwrapInner inner) `C.catch` const (return Nothing) + case x of + Nothing -> return Nothing + Just (Left _, st) -> return $ Just $ Left st + Just (Right y, st) -> return $ Just $ Right (y, st) + +-- | Runs all the processes in parallel and returns their results once they +-- have all finished. The length and ordering of the results reflects the +-- length and ordering of the input +runParallelPoison :: forall a. [CHP a] -> CHP [a] +runParallelPoison processes + = do c <- liftIO $ atomically $ newResultsVar + (_, trace) <- PoisonT $ lift $ liftTrace get + blanks <- liftIO $ blankTraces trace (length processes) + liftIO $ + mapM_ forkIO [do y <- wrapProcess p $ flip runStateT ([], btr) . getStandard + C.block $ atomically $ + do ys <- readTVar c + writeTVar c $ (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 + if length xs == length processes + then return xs + else retry + let sortedResults = map snd $ sortBy (comparing fst) results + PoisonT $ lift $ liftTrace $ mergeSubProcessTraces (mapMaybe snd sortedResults) + mapM (maybe throwPoison return . fst) sortedResults + 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, + Int)) m a) + deriving (Monad, MonadIO, MonadTrans, MonadCHP) + +-- TODO in future, get forking working with structural traces + +-- | Executes a forking block. Processes may be forked off inside (using the +-- 'fork' function). When the block completes, it waits for all the forked +-- off processes to complete before returning the output, as long as none of +-- the processes terminated with uncaught poison. If they did, the poison +-- is propagated (rethrown). +forking :: MonadCHP m => ForkingT m a -> m a +-- Like with parallel, this could probably be made a little more efficient +forking (Forking m) = do b <- liftIO $ atomically $ newTVar (False, 0) + output <- runReaderT m b + p <- liftIO $ atomically $ do + (p,n) <- readTVar b + if n == 0 + then return p + else retry + if p + then liftCHP throwPoison + else return output + + +-- | Forks off the given process. The process then runs in parallel with this +-- code, until the end of the forking block, when all forked-off processes +-- are waited for. At that point, once all of the processes have finished, +-- if any of them threw poison it is propagated. +fork :: MonadCHP m => CHP () -> ForkingT m () +fork p = Forking $ + do b <- ask + liftIO $ atomically $ do + (pa, n) <- readTVar b + writeTVar b (pa, n + 1) + (_, trace) <- liftCHP $ PoisonT $ lift $ liftTrace get + [blank] <- liftIO $ blankTraces trace 1 + liftIO $ forkIO $ do + r <- wrapProcess p $ flip runStateT ([], blank) . getStandard + C.block $ atomically $ do + (poisonedAlready, n) <- readTVar b + writeTVar b $ (poisonedAlready || isNothing r, n - 1) + return () + diff --git a/Control/Concurrent/CHP/Poison.hs b/Control/Concurrent/CHP/Poison.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/Poison.hs @@ -0,0 +1,38 @@ +-- 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. + +module Control.Concurrent.CHP.Poison where + +-- | A Maybe-like poison wrapper. +data WithPoison a = PoisonItem | NoPoison a deriving (Eq) + +instance Functor WithPoison where + fmap f (NoPoison x) = NoPoison $ f x + fmap _ PoisonItem = PoisonItem + diff --git a/Control/Concurrent/CHP/ProcessId.hs b/Control/Concurrent/CHP/ProcessId.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/ProcessId.hs @@ -0,0 +1,59 @@ +-- 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. + +module Control.Concurrent.CHP.ProcessId where + +-- We derive Ord for these two types so that they can be used in sets. However, +-- the Ord instance should not be confused with the partial ordering detailed +-- in the tracing paper, which is provided by the pidLessThanOrEqual function +-- below. + +data ProcessIdPart = ParSeq Int Integer deriving (Eq, Ord) + +newtype ProcessId = ProcessId [ProcessIdPart] deriving (Eq, Ord) + +rootProcessId :: ProcessId +rootProcessId = ProcessId [ParSeq 0 0] + +emptyProcessId :: ProcessId +emptyProcessId = ProcessId [] + +-- We use here the property that the lists will either be equal, or different +-- within their shared length. Therefore they will be different when zipped, +-- or if they're identical that means the originals were identical +pidLessThanOrEqual :: ProcessId -> ProcessId -> Bool +pidLessThanOrEqual (ProcessId pida) (ProcessId pidb) = cmp (zip pida pidb) + where + cmp :: ([(ProcessIdPart, ProcessIdPart)]) -> Bool + cmp [] = True -- Equal + cmp ((ParSeq pa sa, ParSeq pb sb):xs) + | pa /= pb = False -- No ordering + | sa < sb = True -- Less than + | sa > sb = False -- Greater than + | otherwise = cmp xs --Both parts equal, keep going diff --git a/Control/Concurrent/CHP/Traces.hs b/Control/Concurrent/CHP/Traces.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/Traces.hs @@ -0,0 +1,68 @@ +-- 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 that re-exports all the parts of the library related to tracing. +-- +-- The idea of tracing is to record the concurrent events (i.e. channel communications +-- and barrier synchronisations) that occur during a run of the program. You +-- can think of it as automatically turning on a lot of debug logging. +-- +-- Typically, at the top-level of your program you should have: +-- +-- > main :: IO () +-- > main = runCHP myMainProcess +-- +-- To turn on the tracing mechanism of your choice (for example, CSP-style tracing) +-- to automatically print out the trace after completion of your program, just +-- use the appropriate helper function: +-- +-- > main :: IO () +-- > main = runCHP_CSPTraceAndPrint myMainProcess +-- +-- It could hardly be easier. If you want more fine-grained control and examination +-- of the trace, you can use the helper functions of the form 'runCHP_CSPTrace' +-- that give back a data structure representing the trace, which you can then +-- manipulate. The Doc used by the traces is from the 'Text.PrettyPrint.HughesPJ' +-- module that currently comes with GHC (I think). +module Control.Concurrent.CHP.Traces + (module Control.Concurrent.CHP.Traces.CSP + ,module Control.Concurrent.CHP.Traces.Structural + ,module Control.Concurrent.CHP.Traces.TraceOff + ,module Control.Concurrent.CHP.Traces.VCR + ,RecordedEvent(..) + ,RecordedIndivEvent(..) + ,Trace(..) + ) where + +import Control.Concurrent.CHP.Base +import Control.Concurrent.CHP.Traces.Base +import Control.Concurrent.CHP.Traces.CSP +import Control.Concurrent.CHP.Traces.Structural +import Control.Concurrent.CHP.Traces.TraceOff +import Control.Concurrent.CHP.Traces.VCR diff --git a/Control/Concurrent/CHP/Traces/Base.hs b/Control/Concurrent/CHP/Traces/Base.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/Traces/Base.hs @@ -0,0 +1,253 @@ +-- 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. + +module Control.Concurrent.CHP.Traces.Base where + +import Control.Concurrent.STM +import Control.Monad.State +import Data.List +import qualified Data.Map as Map +import qualified Data.Set as Set +import Data.Unique + +import Control.Concurrent.CHP.Event +import Control.Concurrent.CHP.ProcessId + +-- | A globally recorded event, found in CSP and VCR traces. Either a +-- channel communication (with a unique identifier of the channel) or a +-- barrier synchronisation (with a unique identifier of the barrier). The +-- identifiers are per channel\/barrier, not per event. Currently, +-- channels and barriers can never have the same Unique as each other, but +-- do not rely on this behaviour. +data RecordedEvent = + ChannelComm Unique + | BarrierSync Unique + deriving (Eq, Ord) + +-- | An individual record of an event, found in nested traces. Either a +-- channel write or read, or a barrier synchronisation, each with a unique +-- identifier for the barrier\/channel. The event will be recorded by +-- everyone involved, and a ChannelWrite will have the same channel +-- identifier as the corresponding channel-read. The identifiers are per +-- channel\/barrier, not per event. Currently, channels and barriers can +-- never have the same Unique as each other, but do not rely on this +-- behaviour. +data RecordedIndivEvent = + ChannelWrite Unique + | ChannelRead Unique + | BarrierSyncIndiv Unique + deriving (Eq, Ord) + +type RecEvents = (Maybe RecordedEvent, Maybe RecordedIndivEvent) + +getName :: Unique -> State (Map.Map Unique String) String +getName u = do m <- get + case Map.lookup u m of + Just x -> return x + Nothing -> let x = "c" ++ show (Map.size m) in + do put $ Map.insert u x m + return x + +nameEvent :: RecordedEvent -> State (Map.Map Unique String) String +nameEvent (ChannelComm c) = getName c +nameEvent (BarrierSync c) = getName c + +nameIndivEvent :: RecordedIndivEvent -> State (Map.Map Unique String) String +nameIndivEvent (ChannelWrite c) = do c' <- getName c + return $ c' ++ "!" +nameIndivEvent (ChannelRead c) = do c' <- getName c + return $ c' ++ "?" +nameIndivEvent (BarrierSyncIndiv c) = do c' <- getName c + return $ c' ++ "*" + +ensureAllNamed :: Map.Map Unique String -> [RecordedEvent] -> Map.Map Unique String +-- Quite hacky: +ensureAllNamed m es = execState (mapM_ nameEvent es) m + +ensureAllNamedIndiv :: Map.Map Unique String -> [RecordedIndivEvent] -> Map.Map Unique String +-- Quite hacky: +ensureAllNamedIndiv m es = execState (mapM_ nameIndivEvent es) m + + +-- The list of integers is for alting +type TraceT = StateT ([Int], TraceStore) + +data TraceStore = + NoTrace + | Trace (ProcessId, TVar ChannelLabels, SubTraceStore) + +type ChannelLabels = Map.Map Unique String + +data SubTraceStore = + Hierarchy (Structured RecordedIndivEvent) + | CSPTraceRev (TVar [(Int, [RecordedEvent])]) + | VCRTraceRev (TVar [Set.Set (Set.Set ProcessId, RecordedEvent)]) + +data Ord a => Structured a = + StrEvent a + | Par [Structured a] + | RevSeq [(Int, [Structured a])] + deriving (Eq, Ord) + +-- | Records an event where you were the last person to engage in the event +recordEventLast :: Maybe RecordedEvent -> Set.Set ProcessId -> TraceT IO () +recordEventLast Nothing _ = return () +recordEventLast (Just e) otherPids + = do (x,y) <- get + case (x, y) of + ( _, Trace (_,_,CSPTraceRev tv)) -> liftIO $ atomically $ + do t <- readTVar tv + writeTVar tv $! addRLE e t + (_, Trace (pid, _, VCRTraceRev tv)) -> liftIO $ atomically $ do + t <- readTVar tv + let pidSet = Set.insert pid otherPids + t' = case t of + -- Trace previously empty: + [] -> [Set.singleton (pidSet, e)] + (z:zs) | shouldMakeNewSetVCR pidSet z + -> Set.singleton (pidSet, e) : t + | otherwise + -> Set.insert (pidSet, e) z : zs + writeTVar tv $! t' + _ -> return () + +-- | Records an event where you were one of the people involved +recordEvent :: Maybe RecordedIndivEvent -> TraceT IO () +recordEvent Nothing = return () +recordEvent (Just e) + = do (x,y) <- get + case (x, y, e) of + (as, Trace (pid,tvls,Hierarchy es),_) -> + put (as, Trace (pid, tvls, Hierarchy (addSeqEventH e es))) + _ -> return () + +mergeSubProcessTraces :: [TraceStore] -> TraceT IO () +mergeSubProcessTraces ts + = do s <- get + case s of + (as, Trace (pid, tvls, Hierarchy es)) -> + put (as, Trace (pid, tvls, Hierarchy $ addParEventsH ts' es)) + where ts' = [t | Trace (_,_,Hierarchy t) <- ts] + _ -> return () + + +shouldMakeNewSetVCR :: Set.Set ProcessId -> Set.Set (Set.Set ProcessId, RecordedEvent) + -> Bool +shouldMakeNewSetVCR newIds existingSet + = exists existingSet $ \(bigP,_) -> exists bigP $ \p -> exists newIds $ \q -> + p `pidLessThanOrEqual` q + where + -- Like the any function (flipped), but for sets: + exists :: Ord a => Set.Set a -> (a -> Bool) -> Bool + exists s f = not . Set.null $ Set.filter f s + + +compress :: (Eq a, Ord a) => Structured a -> Structured a +compress (RevSeq ((1,s):(n,s'):ss)) + | n == 1 && (s `isPrefixOf` s') = compress $ RevSeq $ (2,s):(1,drop (length s) s'):ss + | s == s' = compress $ RevSeq $ (n+1,s'):ss +compress x = x + + +addParEventsH :: (Eq a, Ord a) => [Structured a] -> Structured a -> Structured a +addParEventsH es t = let n = es in case t of + StrEvent _ -> RevSeq [(1, [Par n, t])] + Par _ -> RevSeq [(1, [Par n, t])] + RevSeq ((1,s):ss) -> compress $ RevSeq $ (1,Par n : s) : ss + RevSeq ss -> compress $ RevSeq $ (1, [Par n]):ss + +addSeqEventH :: (Eq a, Ord a) => a -> Structured a -> Structured a +addSeqEventH e (StrEvent e') = RevSeq [(1,[StrEvent e, StrEvent e'])] +addSeqEventH e (Par p) = RevSeq [(1,[StrEvent e, Par p])] +addSeqEventH e (RevSeq ((1,s):ss)) + | (StrEvent e) `notElem` s = compress $ RevSeq $ (1,StrEvent e:s):ss +addSeqEventH e (RevSeq ss) = compress $ RevSeq $ (1,[StrEvent e]):ss + + +addRLE :: Eq a => a -> [(Int,[a])] -> [(Int,[a])] +-- Single event in most recent list: +addRLE x ((n,[e]):nes) + -- If we have the same event, bump up the count: + | x == e = (n+1,[e]):nes +-- Only one list thus far +addRLE x allEs@[(1,es)] + -- If we are the same as the most recent event, break it off and aggregate: + | x == head es = [(2, [x]), (1, tail es)] + -- If we're already in that list, start a new one: + | x `elem` es = (1,[x]):allEs + -- Otherwise join the queue! + | otherwise = [(1,x:es)] +-- Most recent list has count of 1 +addRLE x allEs@((1,es):(n,es'):nes) + -- If adding us to the most recent list forms an identical list to the one + -- before that, aggregate + | x == head es' && es == tail es' = (n+1,es'):nes + -- If we're already in that list, start a new one: + | x `elem` es = (1,[x]):allEs + -- Otherwise, join the most recent list + | otherwise = (1,x:es):(n,es'):nes +-- If no special cases hold, start a new list: +addRLE x nes = (1,[x]):nes + + +labelEvent :: Event -> String -> StateT (a, TraceStore) IO () +labelEvent (Event (u,_)) l + = labelUnique u l + +labelUnique :: Unique -> String -> StateT (a, TraceStore) IO () +labelUnique u l + = do (_,t) <- get + case t of + NoTrace -> return () + Trace (_,tvls,_) -> add tvls + where + add :: TVar (Map.Map Unique String) -> StateT (a, TraceStore) IO () + add tv = liftIO $ atomically $ do + m <- readTVar tv + writeTVar tv $ Map.insert u l m + + +blankTraces :: TraceStore -> Int -> IO [TraceStore] +blankTraces NoTrace n = return $ replicate n NoTrace +blankTraces (Trace (pid, tvls, subT)) n = + return [Trace (newId, tvls, newSubT) | newId <- newIds] + where + newIds :: [ProcessId] + newIds = let ProcessId parts = pid in + [ProcessId $ parts ++ [ParSeq i 0] | i <- [0 .. (n - 1)]] + + newSubT :: SubTraceStore + newSubT = case subT of + Hierarchy {} -> Hierarchy $ RevSeq [] + _ -> subT + + + + + diff --git a/Control/Concurrent/CHP/Traces/CSP.hs b/Control/Concurrent/CHP/Traces/CSP.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/Traces/CSP.hs @@ -0,0 +1,76 @@ +-- 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. + +-- | This module contains support for CSP-style tracing. A CSP trace is simply +-- a flat list of events in the order in which they occurred. +module Control.Concurrent.CHP.Traces.CSP (CSPTrace(..), runCHP_CSPTrace, runCHP_CSPTraceAndPrint) where + +import Control.Concurrent.STM +import Control.Monad.State +import qualified Data.Map as Map +import Text.PrettyPrint.HughesPJ + +import Control.Concurrent.CHP.Base +import Control.Concurrent.CHP.Traces.Base + +-- | A classic CSP trace. It is simply the channel labels, and a list of recorded +-- events in sequence -- the head of the list is the first (oldest) event. +newtype CSPTrace = CSPTrace (ChannelLabels, [RecordedEvent]) + +instance Show CSPTrace where + show = renderStyle (Style OneLineMode 1 1) . prettyPrint + +instance Trace CSPTrace where + emptyTrace = CSPTrace (Map.empty, []) + runCHPAndTrace p = do tv <- atomically $ newTVar [] + runCHPProgramWith' (CSPTraceRev tv) toPublic p + + prettyPrint (CSPTrace (labels, events)) + = char '<' <+> (sep $ punctuate (char ',') $ map (text . nameCSP labels') events) <+> char '>' + where + labels' = ensureAllNamed labels events + +toPublic :: ChannelLabels -> SubTraceStore -> IO CSPTrace +toPublic l (CSPTraceRev tv) + = do list <- atomically $ readTVar tv + return $ CSPTrace (l, concatMap (\(n,es) -> concat $ replicate n $ reverse es) $ reverse list) +toPublic _ _ = error "Error in CSP trace -- tracing type got switched" + +nameCSP :: ChannelLabels -> RecordedEvent -> String +nameCSP m = flip evalState m . nameEvent + + +runCHP_CSPTrace :: CHP a -> IO (Maybe a, CSPTrace) +runCHP_CSPTrace = runCHPAndTrace + +runCHP_CSPTraceAndPrint :: CHP a -> IO () +runCHP_CSPTraceAndPrint p = do (_, tr) <- runCHP_CSPTrace p + putStrLn $ show $ tr + + diff --git a/Control/Concurrent/CHP/Traces/Structural.hs b/Control/Concurrent/CHP/Traces/Structural.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/Traces/Structural.hs @@ -0,0 +1,130 @@ +-- 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. + +-- | This module contains support for structural traces. Structural +-- traces reflect the parallel composition of your program. Effectively, +-- each process records its own local trace. Parallel traces are then +-- merged as parallel compositions, and you end up with a big tree of +-- sequentially and parallel-composed traces. Note that in this tracing +-- style, unlike CSP and VCR, events are recorded by /every/ process +-- involved in them, not just once per occurrence. +module Control.Concurrent.CHP.Traces.Structural (StructuralTrace(..), EventHierarchy(..), runCHP_StructuralTrace, runCHP_StructuralTraceAndPrint, + getAllEventsInHierarchy) where + +import Control.Monad.State +import Data.List +import qualified Data.Map as Map +import Data.Maybe +import Text.PrettyPrint.HughesPJ + +import Control.Concurrent.CHP.Base +import Control.Concurrent.CHP.Traces.Base + +-- | A data type representing a hierarchy of events. The count on the StructuralSequence +-- count is a replicator count for that list of sequential items. +data EventHierarchy a = + SingleEvent a + | StructuralSequence Int [EventHierarchy a] + | StructuralParallel [EventHierarchy a] + +instance Functor EventHierarchy where + fmap f (SingleEvent x) = SingleEvent $ f x + fmap f (StructuralSequence n es) = StructuralSequence n $ map (fmap f) es + fmap f (StructuralParallel es) = StructuralParallel $ map (fmap f) es + +-- | Flattens the events into a list. The resulting list may contain duplicates, and it +-- should not be assumed that the order relates in any way to the original +-- hierarchy. +getAllEventsInHierarchy :: EventHierarchy a -> [a] +getAllEventsInHierarchy (SingleEvent e) = [e] +getAllEventsInHierarchy (StructuralSequence _ es) = concatMap getAllEventsInHierarchy es +getAllEventsInHierarchy (StructuralParallel es) = concatMap getAllEventsInHierarchy es + + +-- | A nested (or hierarchical) trace. The trace is an event hierarchy, wrapped +-- in a Maybe type to allow for representation of the empty trace (Nothing). +newtype StructuralTrace = StructuralTrace (ChannelLabels, Maybe (EventHierarchy RecordedIndivEvent)) + +instance Show StructuralTrace where + show = renderStyle (Style OneLineMode 1 1) . prettyPrint + +instance Trace StructuralTrace where + emptyTrace = StructuralTrace (Map.empty, Nothing) + runCHPAndTrace p = runCHPProgramWith' (Hierarchy $ RevSeq []) toPublic p + + prettyPrint (StructuralTrace (_,Nothing)) = empty + prettyPrint (StructuralTrace (labels, Just h)) + = pp $ fmap (flip evalState labels' . nameIndivEvent) h + where + labels' = ensureAllNamedIndiv labels $ + getAllEventsInHierarchy h + + pp :: EventHierarchy String -> Doc + pp (SingleEvent x) = text x + pp (StructuralSequence 1 es) + = sep $ intersperse (text "->") $ map pp es + pp (StructuralSequence n es) + = int n <> char '*' <> (parens $ sep $ + intersperse (text "->") $ map pp es) + pp (StructuralParallel es) + = parens $ sep $ intersperse (text "||") $ map pp es + +toPublic :: ChannelLabels -> SubTraceStore -> IO StructuralTrace +toPublic l (Hierarchy h) + = return $ StructuralTrace (l, conv h) + where + nonEmptyListToMaybe :: ([a] -> b) -> [a] -> Maybe b + nonEmptyListToMaybe _ [] = Nothing + nonEmptyListToMaybe f xs = Just $ f xs + + mapToMaybe :: ([b] -> c) -> (a -> Maybe b) -> [a] -> Maybe c + mapToMaybe f g = nonEmptyListToMaybe f . mapMaybe g + + conv :: Ord a => Structured a -> Maybe (EventHierarchy a) + conv (StrEvent x) = Just $ SingleEvent x + conv (Par es) = mapToMaybe StructuralParallel conv es + conv (RevSeq []) = Nothing + conv (RevSeq [(0, _)]) = Nothing + conv (RevSeq [(n, ss)]) = mapToMaybe (StructuralSequence n) conv (reverse ss) + conv (RevSeq es) = trans + where + rev = reverse es + trans = mapToMaybe (StructuralSequence 1) (\(n,s) -> mapToMaybe (StructuralSequence n) conv $ + reverse s) rev +toPublic _ _ = error "Error in Structural trace -- tracing type got switched" + +runCHP_StructuralTrace :: CHP a -> IO (Maybe a, StructuralTrace) +runCHP_StructuralTrace = runCHPAndTrace + +runCHP_StructuralTraceAndPrint :: CHP a -> IO () +runCHP_StructuralTraceAndPrint p + = do (_, tr) <- runCHP_StructuralTrace p + putStrLn $ show tr + + diff --git a/Control/Concurrent/CHP/Traces/TraceOff.hs b/Control/Concurrent/CHP/Traces/TraceOff.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/Traces/TraceOff.hs @@ -0,0 +1,51 @@ +-- 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. + +-- This module contains a trace-type that does not record anything. This is +-- generally not needed (just use runCSP without tracing) but is included in +-- the library for completeness. +module Control.Concurrent.CHP.Traces.TraceOff (TraceOff) where + +import Control.Concurrent.CHP.Base +import Control.Concurrent.CHP.Traces.Base + +import Text.PrettyPrint.HughesPJ + +-- | A trace type that does not record anything. +newtype TraceOff = TraceOff () + +instance Show TraceOff where + show = const "" + +instance Trace TraceOff where + runCHPAndTrace = runCHPProgramWith NoTrace (const $ TraceOff ()) + emptyTrace = TraceOff () + prettyPrint = const empty + + diff --git a/Control/Concurrent/CHP/Traces/VCR.hs b/Control/Concurrent/CHP/Traces/VCR.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/Traces/VCR.hs @@ -0,0 +1,86 @@ +-- 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 for recording View Centric Reasoning (VCR) traces. A view centric +-- reasnoning trace is a list of sets of events. Each set contains independent +-- events that have no causal relationship between them. Hopefully we will +-- publish a paper explaining all this in detail soon. +module Control.Concurrent.CHP.Traces.VCR (VCRTrace(..), runCHP_VCRTrace, runCHP_VCRTraceAndPrint) where + +import Control.Concurrent.STM +import Control.Monad.State +import qualified Data.Map as Map +import qualified Data.Set as Set +import Text.PrettyPrint.HughesPJ + +import Control.Concurrent.CHP.Base +import Control.Concurrent.CHP.Traces.Base + +-- | A VCR (View-Centric Reasoning) trace. It is the channel labels, +-- accompanied by a sequential list of sets of recorded events. Each of +-- the sets is a set of independent events. The set at the head of the +-- list is the first-recorded (oldest). +newtype VCRTrace = VCRTrace (ChannelLabels, [Set.Set RecordedEvent]) + +instance Show VCRTrace where + show = renderStyle (Style OneLineMode 1 1) . prettyPrint + +instance Trace VCRTrace where + emptyTrace = VCRTrace (Map.empty, []) + runCHPAndTrace p = do tv <- atomically $ newTVar [] + runCHPProgramWith' (VCRTraceRev tv) toPublic p + + prettyPrint (VCRTrace (labels, eventSets)) + = char '<' <+> (sep $ punctuate (char ',') $ map (braces . sep . punctuate (char ',')) ropes) <+> char '>' + where + labels' = ensureAllNamed labels (concatMap Set.toList eventSets) + es = map (nameVCR labels') eventSets + + ropes :: [[Doc]] + ropes = map (map text . Set.toList) es + + +toPublic :: ChannelLabels -> SubTraceStore -> IO VCRTrace +toPublic l (VCRTraceRev tv) + = do setList <- atomically $ readTVar tv + return $ VCRTrace (l, map (Set.map snd) $ reverse setList) +toPublic _ _ = error "Error in VCR trace -- tracing type got switched" + +nameVCR :: ChannelLabels -> Set.Set RecordedEvent -> Set.Set String +nameVCR m x = Set.fromList $ evalState (mapM nameEvent $ Set.toList x) m + + +runCHP_VCRTrace :: CHP a -> IO (Maybe a, VCRTrace) +runCHP_VCRTrace = runCHPAndTrace + +runCHP_VCRTraceAndPrint :: CHP a -> IO () +runCHP_VCRTraceAndPrint p = do (_, tr) <- runCHP_VCRTrace p + putStrLn $ show tr + + diff --git a/Control/Concurrent/CHP/Utils.hs b/Control/Concurrent/CHP/Utils.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/CHP/Utils.hs @@ -0,0 +1,88 @@ +-- 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 collection of useful functions to use with the library. +module Control.Concurrent.CHP.Utils where + +import Control.Monad + +import Control.Concurrent.CHP + +-- | Wires given processes up in a forward cycle. That is, the first process +-- writes to the second, and receives from the last. It returns the list of +-- wired-up processes, which you will almost certainly want to run in parallel. +wireCycle :: Channel r w => [r a -> w a -> proc] -> CHP [proc] +wireCycle procs + = do chan <- newChannel + wirePipeline procs (reader chan) (writer chan) + -- return [p (reader $ chans !! i) (writer $ chans !! ((i + 1) `mod` n)) | (p, i) <- zip procs [0..]] + +-- | Wires the given processes up in a forward pipeline. The first process +-- in the list is connected to the given reading channel-end (the first parameter) +-- and the writing end of a new channel, A. The second process is wired up +-- to the reading end of A, and the writing end of the next new channel, B. +-- This proceeds all the way to the end of the list, until the final process +-- is wired to the reading end of Z (if you have 27 processes in the list, +-- and therefore 26 channels in the middle of them) and the second parameter. +-- The list of wired-up processes is returned, which you can then run in parallel. +wirePipeline :: forall a r w proc. Channel r w => [r a -> w a -> proc] -> r a -> w a + -> CHP [proc] +wirePipeline [] _ _ = return [] +wirePipeline procs in_ out + = do chans <- replicateM (n - 1) newChannel + -- return $ map (wire chans) $ zip procs [0..] + return $ (\(w, ps) -> head procs in_ w : ps) $ (foldr wireF (out, []) $ zip (tail procs) chans) + where + n = length procs + + -- One way of doing it: + {- + wire :: [OneToOneChannel a] -> (Chanin a -> Chanout a -> CSProcess, Int) -> CSProcess + wire cs (p, i) + | i == 0 = p in_ (writer $ cs !! 0) + | i == n - 1 = p (reader $ cs !! i) out + | otherwise = p (reader $ cs !! i) (writer $ cs !! (i + 1)) + -} + -- A way without indexing (possibly a bit more efficient): + wireF :: (r a -> w a -> proc, Chan r w a) -> (w a, [proc]) -> (w a, [proc]) + wireF (p, c) (w, ps) = (writer c, p (reader c) w : ps) + +-- | Process composition. Given two processes, composes them into a pipeline, +-- like function composition (but with an opposite ordering). The function +-- is associative. Using wirePipeline will be more efficient than @foldl1 +-- (|->|)@ for more than two processes. +(|->|) :: Channel r w => (a -> w b -> CHP ()) -> (r b -> c -> CHP ()) -> + (a -> c -> CHP ()) +(|->|) p q x y = do c <- newChannel + runParallel_ [p x (writer c), q (reader c) y] + +-- | The reversed version of the other operator. +(|<-|) :: Channel r w => (r b -> c -> CHP ()) -> (a -> w b -> CHP ()) -> + (a -> c -> CHP ()) +(|<-|) = flip (|->|) diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,27 @@ +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. diff --git a/Setup.lhs b/Setup.lhs new file mode 100644 --- /dev/null +++ b/Setup.lhs @@ -0,0 +1,5 @@ +#! /usr/bin/env runhaskell + +> import Distribution.Simple +> main = defaultMain + diff --git a/chp.cabal b/chp.cabal new file mode 100644 --- /dev/null +++ b/chp.cabal @@ -0,0 +1,48 @@ +Name: chp +Version: 1.0.0 +Synopsis: Communicating Haskell Processes: an implementation of concurrency ideas from Communicating Sequential Processes +License: BSD3 +License-file: LICENSE +Author: Neil Brown +Maintainer: neil@twistedsquare.com +Copyright: Copyright (c) 2008, University of Kent +Stability: Provisional +Tested-with: GHC==6.8.2 +Description: Requires at least GHC 6.8.1 +Category: Concurrency + +Build-Type: Simple +Build-Depends: base, containers, mtl, pretty, stm + +Exposed-modules: Control.Concurrent.CHP + Control.Concurrent.CHP.Alt + Control.Concurrent.CHP.Barriers + Control.Concurrent.CHP.BroadcastChannels + Control.Concurrent.CHP.Buffers + Control.Concurrent.CHP.Channels + Control.Concurrent.CHP.Common + Control.Concurrent.CHP.Console + Control.Concurrent.CHP.Enroll + Control.Concurrent.CHP.Monad + Control.Concurrent.CHP.Parallel + Control.Concurrent.CHP.Traces + Control.Concurrent.CHP.Traces.CSP + Control.Concurrent.CHP.Traces.Structural + Control.Concurrent.CHP.Traces.TraceOff + Control.Concurrent.CHP.Traces.VCR + Control.Concurrent.CHP.Utils + +Other-modules: Control.Concurrent.CHP.Base + Control.Concurrent.CHP.CSP + Control.Concurrent.CHP.Event + Control.Concurrent.CHP.Guard + Control.Concurrent.CHP.Mutex + Control.Concurrent.CHP.Poison + Control.Concurrent.CHP.ProcessId + Control.Concurrent.CHP.Traces.Base + +Extensions: ScopedTypeVariables MultiParamTypeClasses + FlexibleInstances UndecidableInstances + GeneralizedNewtypeDeriving + +GHC-Options: -Wall -threaded -O2