chp 1.1.0 → 1.1.1
raw patch · 8 files changed
+162/−8 lines, 8 files
Files
- Control/Concurrent/CHP/Arrow.hs +20/−1
- Control/Concurrent/CHP/Base.hs +6/−0
- Control/Concurrent/CHP/BroadcastChannels.hs +97/−1
- Control/Concurrent/CHP/Channels.hs +8/−3
- Control/Concurrent/CHP/Common.hs +16/−0
- Control/Concurrent/CHP/Console.hs +6/−0
- Control/Concurrent/CHP/Parallel.hs +6/−0
- chp.cabal +3/−3
Control/Concurrent/CHP/Arrow.hs view
@@ -62,7 +62,18 @@ -- Added in version 1.0.2. module Control.Concurrent.CHP.Arrow (ProcessPipeline, runPipeline, arrowProcess) where +-- I have got this module to work on GHC 6.8 and 6.10 by following the CPP-variant+-- instructions on this page: http://haskell.org/haskellwiki/Upgrading_packages++#if __GLASGOW_HASKELL__ >= 609+import Control.Category+import Prelude hiding ((.), id)+#endif+ import Control.Arrow+#if __GLASGOW_HASKELL__ < 610+ hiding (pure)+#endif import Control.Monad import Control.Concurrent.CHP@@ -102,9 +113,17 @@ arrowProcess :: (Chanin a -> Chanout b -> CHP ()) -> ProcessPipeline a b arrowProcess = ProcessPipeline +#if __GLASGOW_HASKELL__ >= 609+instance Category ProcessPipeline where+ (ProcessPipeline q) . (ProcessPipeline p) = ProcessPipeline (p |->| q)+ id = ProcessPipeline CHP.id+#endif+ instance Arrow ProcessPipeline where- arr = ProcessPipeline . CHP.map+#if __GLASGOW_HASKELL__ < 609 (ProcessPipeline p) >>> (ProcessPipeline q) = ProcessPipeline (p |->| q)+#endif+ arr = ProcessPipeline . CHP.map first (ProcessPipeline p) = ProcessPipeline $ \in_ out -> do c <- newChannel
Control/Concurrent/CHP/Base.hs view
@@ -34,7 +34,13 @@ module Control.Concurrent.CHP.Base where import Control.Concurrent.STM+#if __GLASGOW_HASKELL__ >= 609+-- I can't figure out the new Exception system in GHC 6.10 and how I catch all+-- exceptions (see GHC bug #2655), so I'm just going to use the old system:+import qualified Control.OldException as C+#else import qualified Control.Exception as C+#endif import Control.Monad.Error import Control.Monad.Reader import Control.Monad.State
Control/Concurrent/CHP/BroadcastChannels.hs view
@@ -38,11 +38,33 @@ -- 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).+--+-- This module also contains reduce channels (added in version 1.1.1). Because+-- in CHP channels must have the same type at both ends, we use the Monoid+-- type-class. It is important to be aware that the order of mappends will be+-- non-deterministic, and thus you should either use an mappend that is commutative+-- or code around this restruction.+--+-- For example, a common thing to do would be to use lists as the type for+-- reduce channels, make each writer write a single item list (but more is+-- possible), then use the list afterwards, but be aware that it is unordered.+-- If it is important to have an ordered list, make each writer write a pair+-- containing a (unique) index value and the real data, then sort by the index+-- value and discard it.+--+-- Since reduce channels were added after the initial library design, there+-- is a slight complication: it is not possible to use newChannel (and all+-- similar functions) with reduce channels because it is impossible to express+-- the Monoid constraint for the Channel instance. Instead, you must use manyToOneChannel+-- and manyToAnyChannel. module Control.Concurrent.CHP.BroadcastChannels (BroadcastChanin, BroadcastChanout,- OneToManyChannel, AnyToManyChannel, oneToManyChannel, anyToManyChannel) where+ OneToManyChannel, AnyToManyChannel, oneToManyChannel, anyToManyChannel, ReduceChanin,+ ReduceChanout, ManyToOneChannel, ManyToAnyChannel, manyToOneChannel, manyToAnyChannel)+ where import Control.Concurrent.STM import Control.Monad.Trans+import Data.Monoid import Control.Concurrent.CHP.Barriers import Control.Concurrent.CHP.Base@@ -151,3 +173,77 @@ anyToManyChannel :: CHP (AnyToManyChannel a) anyToManyChannel = newChannel++++newtype ReduceChannel a = GC (PhasedBarrier Phase, TVar a, (a -> a -> a, a))++-- | The reading end of a reduce channel.+newtype ReduceChanin a = GI (ReduceChannel a)++-- | The writing end of a reduce channel. You must enroll on it before+-- you can read from it or poison it.+newtype ReduceChanout a = GO (ReduceChannel a)++instance Enrollable ReduceChanout a where+ enroll c@(GO (GC (b,_,_))) f = enroll b (\eb -> waitForPhase Neutral eb >> f (Enrolled c))+ resign (Enrolled (GO (GC (b,_,_)))) m+ = do x <- resign (Enrolled b) m+ waitForPhase Neutral (Enrolled b)+ return x++instance WriteableChannel (Enrolled ReduceChanout) where+ extWriteChannel (Enrolled (GO (GC (b, tv, (f,_))))) m+ = do syncBarrierWith (Just . ChannelWrite)+ $ Enrolled b+ m >>= liftIO . atomically . \x -> readTVar tv >>= writeTVar tv . f x+ syncBarrierWith (const Nothing)+ $ Enrolled b+ syncBarrierWith (const Nothing)+ $ Enrolled b+ return ()++instance ReadableChannel ReduceChanin where+ extReadChannel (GI (GC (b, tv, (_, empty)))) f+ = do syncBarrierWith (Just . ChannelRead)+ $ Enrolled b+ syncBarrierWith (const Nothing)+ $ Enrolled b+ x <- liftIO (atomically $ readTVar tv)+ y <- f x+ liftIO (atomically $ writeTVar tv empty)+ syncBarrierWith (const Nothing)+ $ Enrolled b+ return y++instance Poisonable (Enrolled ReduceChanout a) where+ poison (Enrolled (GO (GC (b,_,_)))) = poison $ Enrolled b+ checkForPoison (Enrolled (GO (GC (b,_,_)))) = checkForPoison $ Enrolled b++instance Poisonable (ReduceChanin a) where+ poison (GI (GC (b,_,_))) = poison $ Enrolled b+ checkForPoison (GI (GC (b,_,_))) = checkForPoison $ Enrolled b++newReduceChannel :: Monoid a => CHP (ReduceChannel a)+newReduceChannel = dontWarnMe {- see above -} $ do+ do b@(Barrier (e,_)) <- newPhasedBarrier Neutral+ -- Writer is always enrolled:+ liftIO $ atomically $ enrollEvent e+ tv <- liftIO $ atomically $ newTVar mempty+ return $ GC (b, tv, (mappend, mempty))++type ManyToOneChannel = Chan ReduceChanin ReduceChanout+type ManyToAnyChannel = Chan (Shared ReduceChanin) ReduceChanout++manyToOneChannel :: Monoid a => CHP (ManyToOneChannel a)+manyToOneChannel = do+ c@(GC (b,_,_)) <- newReduceChannel+ return $ Chan (getBarrierIdentifier b) (GI c) (GO c)+++manyToAnyChannel :: Monoid a => CHP (ManyToAnyChannel a)+manyToAnyChannel = do+ m <- newMutex+ c <- manyToOneChannel+ return $ Chan (getChannelIdentifier c) (Shared (m, reader c)) (writer c)+
Control/Concurrent/CHP/Channels.hs view
@@ -91,12 +91,17 @@ -- ====== -- | A reading channel-end type that allows poison to be thrown-newtype Chanin a = Chanin (STMChannel a)+--+-- Eq instance added in version 1.1.1+newtype Chanin a = Chanin (STMChannel a) deriving Eq -- | A writing channel-end type that allows poison to be thrown-newtype Chanout a = Chanout (STMChannel a)+--+-- Eq instance added in version 1.1.1+newtype Chanout a = Chanout (STMChannel a) deriving Eq -newtype STMChannel a = STMChan (Event, TVar (WithPoison (Maybe a)))+newtype STMChannel a = STMChan (Event, TVar (WithPoison (Maybe a))) deriving+ Eq type OneToOneChannel = Chan Chanin Chanout type AnyToOneChannel = Chan (Chanin) (Shared Chanout)
Control/Concurrent/CHP/Common.hs view
@@ -222,3 +222,19 @@ | otherwise -> do writeChannel out newVal internal curVal +-- | A shared variable process. Given an initial value and two channels, it+-- continually offers to output its current value or read in a new one.+--+-- Added in version 1.1.1+valueStore :: a -> Chanin a -> Chanout a -> CHP ()+valueStore val input output = inner val+ where+ inner x = ((writeChannel output x >> return x) <-> readChannel input) >>= inner++-- | A shared variable process. The same as valueStore, but initially waits+-- to read its starting value before then offering to either output its current+-- value or read in a new one.+--+-- Added in version 1.1.1+valueStore' :: Chanin a -> Chanout a -> CHP ()+valueStore' input output = readChannel input >>= \x -> valueStore x input output
Control/Concurrent/CHP/Console.hs view
@@ -32,7 +32,13 @@ import Control.Concurrent import Control.Concurrent.STM+#if __GLASGOW_HASKELL__ >= 609+-- I can't figure out the new Exception system in GHC 6.10 and how I catch all+-- exceptions (see GHC bug #2655), so I'm just going to use the old system:+import qualified Control.OldException as C+#else import qualified Control.Exception as C+#endif import Control.Monad import Control.Monad.Trans import Data.Maybe
Control/Concurrent/CHP/Parallel.hs view
@@ -32,7 +32,13 @@ import Control.Concurrent import Control.Concurrent.STM+#if __GLASGOW_HASKELL__ >= 609+-- I can't figure out the new Exception system in GHC 6.10 and how I catch all+-- exceptions (see GHC bug #2655), so I'm just going to use the old system:+import qualified Control.OldException as C+#else import qualified Control.Exception as C+#endif import Control.Monad.Error import Control.Monad.Reader import Control.Monad.State
chp.cabal view
@@ -1,5 +1,5 @@ Name: chp-Version: 1.1.0+Version: 1.1.1 Synopsis: An implementation of concurrency ideas from Communicating Sequential Processes License: BSD3 License-file: LICENSE@@ -7,7 +7,7 @@ Maintainer: neil@twistedsquare.com Copyright: Copyright (c) 2008, University of Kent Stability: Provisional-Tested-with: GHC==6.8.2+Tested-with: GHC==6.8.2, GHC==6.10.1 Description: The Communicating Haskell Processes (CHP) library is an implementation of ideas from Hoare's Communicating Sequential Processes. More details and a tutorial can be@@ -50,6 +50,6 @@ Extensions: ScopedTypeVariables MultiParamTypeClasses FlexibleInstances UndecidableInstances- GeneralizedNewtypeDeriving+ GeneralizedNewtypeDeriving CPP GHC-Options: -Wall -threaded