chp 1.1.1 → 1.2.0

raw patch · 14 files changed

+1019/−77 lines, 14 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• Control/Concurrent/CHP.hs +6/−1
• Control/Concurrent/CHP/Actions.hs +122/−0
• Control/Concurrent/CHP/Barriers.hs +34/−13
• Control/Concurrent/CHP/BroadcastChannels.hs +30/−9
• Control/Concurrent/CHP/Buffers.hs +25/−1
• Control/Concurrent/CHP/CSP.hs +11/−13
• Control/Concurrent/CHP/Channels.hs +34/−21
• Control/Concurrent/CHP/Clocks.hs +622/−0
• Control/Concurrent/CHP/Common.hs +65/−6
• Control/Concurrent/CHP/Event.hs +8/−1
• Control/Concurrent/CHP/Poison.hs +16/−0
• Control/Concurrent/CHP/Traces/Base.hs +19/−7
• Control/Concurrent/CHP/Utils.hs +24/−4
• chp.cabal +3/−1

@@ -30,6 +30,8 @@ -- | This module re-exports the core functionality of the CHP library.  Other -- modules that you also may wish to import are: --+-- * "Control.Concurrent.CHP.Action"+--  -- * "Control.Concurrent.CHP.Arrow" -- -- * "Control.Concurrent.CHP.Buffers"@@ -50,16 +52,19 @@   module Control.Concurrent.CHP.Barriers,   module Control.Concurrent.CHP.BroadcastChannels,   module Control.Concurrent.CHP.Channels,+  module Control.Concurrent.CHP.Clocks,   module Control.Concurrent.CHP.Enroll,   module Control.Concurrent.CHP.Monad,-  module Control.Concurrent.CHP.Parallel,+  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.Clocks import Control.Concurrent.CHP.Enroll import Control.Concurrent.CHP.Monad import Control.Concurrent.CHP.Parallel+ 

@@ -0,0 +1,122 @@+-- Communicating Haskell Processes.+-- Copyright (c) 2009, 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 action wrappers around channel-ends.+--+-- In CHP, there are a variety of channel-ends.  Enrolled Chanin, Shared Chanout,+-- plain Chanin, and so on.  The difference between these ends can be important;+-- enrolled channel-ends can be resigned from, shared channel-ends need to be claimed+-- before use.  But sometimes you just want to ignore those differences and read+-- and write from the channel-end regardless of its type.  In particular, you want+-- to pass a channel-end to a process without the process worrying about its type.+--+-- Actions allow you to do this.  A send action is like a monadic function (@a+-- -> CHP()@ for sending an item, but can be poisoned too.  A recv action is like+-- something of type @CHP a@ that again can be poisoned.+module Control.Concurrent.CHP.Actions+  ( SendAction, RecvAction,+    sendAction, recvAction,+    makeSendAction, makeRecvAction,+    makeSendAction', makeRecvAction',+    makeCustomSendAction, makeCustomRecvAction,+    nullSendAction, nullRecvAction+  ) where++import Control.Concurrent.CHP+import Control.Monad++-- | A send action.  See 'sendAction'.  Note that it is poisonable.+newtype SendAction a = SendAction (a -> CHP (), CHP (), CHP ())+-- | A receive action.  See 'recvAction'.  Note that it is poisonable.+newtype RecvAction a = RecvAction (CHP a, CHP (), CHP ())++-- | Sends a data item using the given sendAction.  Whether this operation can+-- be used in a choice (see 'alt') is entirely dependent on whether the original+-- action could be used in an alt.  For all of CHP's channels, this is true, but+-- for your own custom send actions, probably not.+sendAction :: SendAction a -> a -> CHP ()+sendAction (SendAction (s, _, _)) = s++-- | Receives a data item using the given recvAction.  Whether this operation can+-- be used in a choice (see 'alt') is entirely dependent on whether the original+-- action could be used in an alt.  For all of CHP's channels, this is true, but+-- for your own custom receive actions, probably not.+recvAction :: RecvAction a -> CHP a+recvAction (RecvAction (s, _, _)) = s++instance Poisonable (SendAction c) where+  poison (SendAction (_,p,_)) = liftCHP p+  checkForPoison (SendAction (_,_,c)) = liftCHP c++instance Poisonable (RecvAction c) where+  poison (RecvAction (_,p,_)) = liftCHP p+  checkForPoison (RecvAction (_,_,c)) = liftCHP c++-- | Given a writing channel end, gives back the corresponding 'SendAction'.+makeSendAction :: (WriteableChannel w, Poisonable (w a)) => w a -> SendAction a+makeSendAction c = SendAction (writeChannel c, poison c, checkForPoison c)++-- | Like 'makeSendAction', but always applies the given function before sending+-- the item.+makeSendAction' :: (WriteableChannel w, Poisonable (w b)) =>+  w b -> (a -> b) -> SendAction a+makeSendAction' c f = SendAction (writeChannel c . f, poison c, checkForPoison c)++-- | Given a reading channel end, gives back the corresponding 'RecvAction'.+makeRecvAction :: (ReadableChannel r, Poisonable (r a)) => r a -> RecvAction a+makeRecvAction c = RecvAction (readChannel c, poison c, checkForPoison c)++-- | Like 'makeRecvAction', but always applies the given function after receiving+-- an item.+makeRecvAction' :: (ReadableChannel r, Poisonable (r a)) =>+  r a -> (a -> b) -> RecvAction b+makeRecvAction' c f = RecvAction (liftM f $ readChannel c, poison c, checkForPoison c)++-- | Creates a custom send operation.  The first parameter should perform the send,+-- the second parameter should poison your communication channel, and the third+-- parameter should check whether the communication channel is already poisoned.+--  Generally, you will want to use 'makeSendAction' instead of this function.+makeCustomSendAction :: (a -> CHP ()) -> CHP () -> CHP () -> SendAction a+makeCustomSendAction x y z = SendAction (x, y, z)++-- | Creates a custom receive operation.  The first parameter should perform the receive,+-- the second parameter should poison your communication channel, and the third+-- parameter should check whether the communication channel is already poisoned.+--  Generally, you will want to use 'makeRecvAction' instead of this function.+makeCustomRecvAction :: CHP a -> CHP () -> CHP () -> RecvAction a+makeCustomRecvAction x y z = RecvAction (x, y, z)++-- | Acts like a SendAction, but just discards the data.+nullSendAction :: SendAction a+nullSendAction = SendAction (const $ return (), return (), return ())++-- | Acts like a RecvAction, but always gives back the given data item.+nullRecvAction :: a -> RecvAction a+nullRecvAction x = RecvAction (return x, return (), return ())+

@@ -64,7 +64,8 @@ -- may query the current phase for any barrier that they are currently enrolled -- on. module Control.Concurrent.CHP.Barriers (Barrier, EnrolledBarrier, newBarrier, newBarrierWithLabel,-  PhasedBarrier, newPhasedBarrier, newPhasedBarrierWithLabel, currentPhase, waitForPhase,+  PhasedBarrier, newPhasedBarrier, newPhasedBarrierWithLabel, newPhasedBarrierCustomInc,+    newPhasedBarrierWithLabelCustomInc, currentPhase, waitForPhase,     syncBarrier, getBarrierIdentifier) where  import Control.Concurrent.STM@@ -89,21 +90,20 @@  -- | 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 :: Enrolled PhasedBarrier phase -> CHP phase syncBarrier = syncBarrierWith (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+currentPhase :: 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 ()+-- repeatedly until it /is/ in the given phase.+waitForPhase :: Eq phase => phase -> Enrolled PhasedBarrier phase -> CHP () waitForPhase ph b = do phCur <- currentPhase b                        when (ph /= phCur) $                          repeatUntil (== ph) (syncBarrier b)@@ -119,25 +119,46 @@ newPhasedBarrier ph = liftPoison $ liftTrace $ do   e <- liftIO $ newEvent BarrierSync 0   tv <- liftIO $ atomically $ newTVar ph-  return $ Barrier (e, tv)+  return $ Barrier (e, tv, \p -> if p == maxBound then minBound else succ p) +-- | Creates a new barrier with no processes enrolled, that will be on the+-- given phase, along with a custom function to increment the phase.  You can therefore+-- use this function with Integer as the inner type (and succ or (+1) as the incrementing+-- function) to get a barrier that never cycles.  You can also do things like supplying+-- (+2) as the incrementing function, or even using lists as the phase type to+-- do crazy things.+newPhasedBarrierCustomInc :: (phase -> phase) -> phase -> CHP (PhasedBarrier phase)+newPhasedBarrierCustomInc f ph = liftPoison $ liftTrace $ do+  e <- liftIO $ newEvent BarrierSync 0+  tv <- liftIO $ atomically $ newTVar ph+  return $ Barrier (e, tv, f) + -- | Creates a new barrier with no processes enrolled and labels it in traces--- using the given label+-- using the given label.  See 'newBarrier'. 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+-- using the given label.  See 'newPhasedBarrier'. newPhasedBarrierWithLabel :: (Enum phase, Bounded phase, Eq phase) => String -> phase -> CHP (PhasedBarrier phase) newPhasedBarrierWithLabel l ph = liftPoison $ liftTrace $ do   e <- liftIO $ newEvent BarrierSync 0   labelEvent e l   tv <- liftIO $ atomically $ newTVar ph-  return $ Barrier (e, tv)+  return $ Barrier (e, tv, \p -> if p == maxBound then minBound else succ p) +-- | Creates a new barrier with no processes enrolled and labels it in traces+-- using the given label.  See 'newPhasedBarrierCustomInc'.+newPhasedBarrierWithLabelCustomInc :: String -> (phase -> phase) -> phase -> CHP (PhasedBarrier phase)+newPhasedBarrierWithLabelCustomInc l f ph = liftPoison $ liftTrace $ do+  e <- liftIO $ newEvent BarrierSync 0+  labelEvent e l+  tv <- liftIO $ atomically $ newTVar ph+  return $ Barrier (e, tv, f) + -- | 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 (e,_)) = getEventUnique e+getBarrierIdentifier :: PhasedBarrier ph -> Unique+getBarrierIdentifier (Barrier (e, _, _)) = getEventUnique e

@@ -58,8 +58,10 @@ -- 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, ReduceChanin,-    ReduceChanout, ManyToOneChannel, ManyToAnyChannel, manyToOneChannel, manyToAnyChannel)+  OneToManyChannel, AnyToManyChannel, oneToManyChannel, anyToManyChannel,+    oneToManyChannelWithLabel, anyToManyChannelWithLabel, ReduceChanin,+    ReduceChanout, ManyToOneChannel, ManyToAnyChannel, manyToOneChannel,+    manyToAnyChannel, manyToOneChannelWithLabel, manyToAnyChannelWithLabel)       where  import Control.Concurrent.STM@@ -148,7 +150,7 @@  newBroadcastChannel :: CHP (BroadcastChannel a) newBroadcastChannel = dontWarnMe {- see above -} $ do-    do b@(Barrier (e,_)) <- newPhasedBarrier Neutral+    do b@(Barrier (e, _, _)) <- newPhasedBarrier Neutral        -- Writer is always enrolled:        liftIO $ atomically $ enrollEvent e        tv <- liftIO $ atomically $ newTVar undefined@@ -168,14 +170,22 @@ type OneToManyChannel = Chan BroadcastChanin BroadcastChanout type AnyToManyChannel = Chan BroadcastChanin (Shared BroadcastChanout) -oneToManyChannel :: CHP (OneToManyChannel a)+oneToManyChannel :: MonadCHP m => m (OneToManyChannel a) oneToManyChannel = newChannel -anyToManyChannel :: CHP (AnyToManyChannel a)+anyToManyChannel :: MonadCHP m => m (AnyToManyChannel a) anyToManyChannel = newChannel +-- | Added in version 1.2.0.+oneToManyChannelWithLabel :: MonadCHP m => String -> m (OneToManyChannel a)+oneToManyChannelWithLabel = newChannelWithLabel +-- | Added in version 1.2.0.+anyToManyChannelWithLabel :: MonadCHP m => String -> m (AnyToManyChannel a)+anyToManyChannelWithLabel = newChannelWithLabel ++ newtype ReduceChannel a = GC (PhasedBarrier Phase, TVar a, (a -> a -> a, a))  -- | The reading end of a reduce channel.@@ -226,7 +236,7 @@  newReduceChannel :: Monoid a => CHP (ReduceChannel a) newReduceChannel = dontWarnMe {- see above -} $ do-    do b@(Barrier (e,_)) <- newPhasedBarrier Neutral+    do b@(Barrier (e, _, _)) <- newPhasedBarrier Neutral        -- Writer is always enrolled:        liftIO $ atomically $ enrollEvent e        tv <- liftIO $ atomically $ newTVar mempty@@ -235,15 +245,26 @@ type ManyToOneChannel = Chan ReduceChanin ReduceChanout type ManyToAnyChannel = Chan (Shared ReduceChanin) ReduceChanout -manyToOneChannel :: Monoid a => CHP (ManyToOneChannel a)+manyToOneChannel :: (Monoid a, MonadCHP m) => m (ManyToOneChannel a) manyToOneChannel = do-    c@(GC (b,_,_)) <- newReduceChannel+    c@(GC (b,_,_)) <- liftCHP newReduceChannel     return $ Chan (getBarrierIdentifier b) (GI c) (GO c)  -manyToAnyChannel :: Monoid a => CHP (ManyToAnyChannel a)+manyToAnyChannel :: (Monoid a, MonadCHP m) => m (ManyToAnyChannel a) manyToAnyChannel = do     m <- newMutex     c <- manyToOneChannel     return $ Chan (getChannelIdentifier c) (Shared (m, reader c)) (writer c) +manyToOneChannelWithLabel :: (Monoid a, MonadCHP m) => String -> m (ManyToOneChannel a)+manyToOneChannelWithLabel l+  = do c <- manyToOneChannel+       liftCHP . liftPoison . liftTrace $ labelUnique (getChannelIdentifier c) l+       return c++manyToAnyChannelWithLabel :: (Monoid a, MonadCHP m) => String -> m (ManyToAnyChannel a)+manyToAnyChannelWithLabel l+  = do c <- manyToAnyChannel+       liftCHP . liftPoison . liftTrace $ labelUnique (getChannelIdentifier c) l+       return c

@@ -31,9 +31,12 @@ -- | 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)+module Control.Concurrent.CHP.Buffers (fifoBuffer, infiniteBuffer,+  accumulatingInfiniteBuffer, overflowingBuffer, overwritingBuffer)   where +import Control.Monad+import Data.Foldable import Data.Sequence (Seq, viewl, ViewL(..)) import qualified Data.Sequence as Seq @@ -71,6 +74,27 @@         takeIn = readChannel in_ >>= buff . addLast s         sendOut = do writeChannel out (seqHead s)                      buff (removeHead s)++-- | Acts like a FIFO buffer with unlimited capacity, but accumulates+-- sequential inputs into a list which it offers in a single output.  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 the empty list.  It always+-- accepts input.  Once it has sent out a value (or values) it removes them+-- from its internal storage.+--+-- Added in version 1.2.0.+accumulatingInfiniteBuffer :: forall a. Chanin a -> Chanout [a] -> CHP ()+accumulatingInfiniteBuffer in_ out+  = buff Seq.empty `onPoisonRethrow` (poison in_ >> poison out)+  where+    buff :: Seq a -> CHP ()+    buff s | Seq.null s = takeIn >>= buff+           | otherwise = (sendOut </> takeIn) >>= buff+      where+        takeIn = liftM (addLast s) $ readChannel in_ +        sendOut = do writeChannel out (toList s)+                     return Seq.empty+  -- | 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.

@@ -75,15 +75,15 @@  -- | 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 -> Maybe-  RecordedIndivEvent) -> Enrolled PhasedBarrier phase -> CHP phase-syncBarrierWith rec (Enrolled (Barrier (e,tv)))+syncBarrierWith :: (Unique -> Maybe RecordedIndivEvent)+  -> Enrolled PhasedBarrier phase -> CHP phase+syncBarrierWith rec (Enrolled (Barrier (e,tv, fph)))     = 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+                    writeTVar tv $ fph 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.@@ -108,12 +108,10 @@ -- * 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)+newtype PhasedBarrier phase = Barrier (Event.Event, TVar phase, phase -> phase) -instance (Enum phase, Bounded phase, Eq phase) => Enrollable PhasedBarrier phase-  where-  enroll b@(Barrier (e,_)) f+instance Enrollable PhasedBarrier phase where+  enroll b@(Barrier (e, _, _)) f     = do liftSTM (Event.enrollEvent e) >>= checkPoison          x <- f $ Enrolled b          liftSTM (Event.resignEvent e) >>= checkPoison >>= (\es ->@@ -121,7 +119,7 @@               when (not $ null es) $ liftSTM $ recordEventLast (nub es) tr)          return x -  resign (Enrolled (Barrier (e,_))) m+  resign (Enrolled (Barrier (e, _, _))) m     = do liftSTM (Event.resignEvent e) >>= checkPoison >>= (\es ->            do (_,tr) <- liftPoison $ liftTrace get               when (not $ null es) $ liftSTM $ recordEventLast (nub es) tr)@@ -141,10 +139,10 @@   -- | 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,_)))+instance Poisonable (Enrolled PhasedBarrier phase) where+  poison (Enrolled (Barrier (e, _, _)))     = liftSTM $ Event.poisonEvent e-  checkForPoison (Enrolled (Barrier (e,_)))+  checkForPoison (Enrolled (Barrier (e, _, _)))     = liftCHP $ liftSTM (Event.checkEventForPoison e) >>= checkPoison  -- | A wrapper (usually around a channel-end) indicating that the inner item

@@ -60,12 +60,12 @@   claim, Shared,    -- * Specific Channel Types-  -- | All the functions here are equivalent to newChannel, but typed.  So for+  -- | All the functions here are equivalent to newChannel (or newChannelWithLabel), but typed.  So for   -- example, @oneToOneChannel = newChannel :: MonadCHP m => m OneToOneChannel@.-  OneToOneChannel, oneToOneChannel,-  OneToAnyChannel, oneToAnyChannel,-  AnyToOneChannel, anyToOneChannel,-  AnyToAnyChannel, anyToAnyChannel+  OneToOneChannel, oneToOneChannel, oneToOneChannelWithLabel,+  OneToAnyChannel, oneToAnyChannel, oneToAnyChannelWithLabel,+  AnyToOneChannel, anyToOneChannel, anyToOneChannelWithLabel,+  AnyToAnyChannel, anyToAnyChannel, anyToAnyChannelWithLabel   )   where @@ -205,22 +205,6 @@ 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--waitForNothingOrPoison :: TVar (WithPoison (Maybe a)) -> STM (WithPoison ())-waitForNothingOrPoison tv = do x <- readTVar tv-                               case x of-                                 PoisonItem -> return PoisonItem-                                 NoPoison (Just _) -> retry-                                 NoPoison Nothing -> return $ NoPoison ()-- -- | 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'.@@ -277,9 +261,17 @@      c <- atomically $ newTVar $ NoPoison Nothing      return (getEventUnique e, STMChan (e,c)) +-- | A type-constrained version of newChannel. oneToOneChannel :: MonadCHP m => m (OneToOneChannel a) oneToOneChannel = newChannel +-- | A type-constrained version of newChannelWithLabel.+--+-- Added in version 1.2.0.+oneToOneChannelWithLabel :: MonadCHP m => String -> m (OneToOneChannel a)+oneToOneChannelWithLabel = newChannelWithLabel++ -- | 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@@ -293,14 +285,35 @@                  return x)        (releaseMutex lock) +-- | A type-constrained version of newChannel. anyToOneChannel :: MonadCHP m => m (AnyToOneChannel a) anyToOneChannel = newChannel +-- | A type-constrained version of newChannel. oneToAnyChannel :: MonadCHP m => m (OneToAnyChannel a) oneToAnyChannel = newChannel +-- | A type-constrained version of newChannel. anyToAnyChannel :: MonadCHP m => m (AnyToAnyChannel a) anyToAnyChannel = newChannel++-- | A type-constrained version of newChannelWithLabel.+--+-- Added in version 1.2.0.+anyToOneChannelWithLabel :: MonadCHP m => String -> m (AnyToOneChannel a)+anyToOneChannelWithLabel = newChannelWithLabel++-- | A type-constrained version of newChannelWithLabel.+--+-- Added in version 1.2.0.+oneToAnyChannelWithLabel :: MonadCHP m => String -> m (OneToAnyChannel a)+oneToAnyChannelWithLabel = newChannelWithLabel++-- | A type-constrained version of newChannelWithLabel.+--+-- Added in version 1.2.0.+anyToAnyChannelWithLabel :: MonadCHP m => String -> m (AnyToAnyChannel a)+anyToAnyChannelWithLabel = newChannelWithLabel  -- ========== -- Instances: 

@@ -0,0 +1,622 @@+-- Communicating Haskell Processes.+-- Copyright (c) 2009, 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.++-- | Clocks, based on an idea by Adam Sampson.+--+-- A clock is similar to a timer, but it is entirely concerned with logical+-- time, rather than any relation to actual time, and all clocks are entirely+-- independent.  A clock has the concept of enrollment, so at any time there+-- are N processes enrolled on the clock.  Each process may wait on the clock+-- for a specific time.  Once all N enrolled processes are waiting for a time+-- (giving a list of times Ts), the clock moves forward to the next time in+-- Ts.+--+-- Let's consider an example.  Three processes are enrolled on an Int clock.  The+-- current time of the clock is 0.  One process asks to wait for time 3, and blocks.+--  A second process asks to wait for time 5, and blocks.  Finally, the third process+-- waits for time 3.  At this point, the first and third process are free to proceed,+-- and the new clock time is 3.  The second process (waiting for time 5) stays+-- waiting until the two processes have returned and waited again.+--+-- There is also the option to wait for the next available time.  If our two enrolled+-- processes wait again, with the first waiting for time 7, and the third waiting for+-- the next available time, the second and third will wake up, with the new time+-- on the clock being 5 (the first stays waiting for time 7).+--+-- There is also the ability for time to wrap around.  If /all/ processes are+-- waiting for a time that is /before/ the current time (using less-than from+-- the Ord instance) or just the next available time, the earliest time of+-- those will be taken.  So if the current time is 26, and the processes are waiting+-- for 11, 21 and next available time, the first and third will wake up and the+-- new time will be 11.  This is particularly useful for using your own algebraic+-- type (@data Phase = PhaseA | PhaseB | PhaseC deriving (Eq, Ord)@).  If you want+-- to, you can use Integer and never use the time wrapping ability (see 'waitUnbounded').+--+-- What the units of a clock mean is entirely up to you.  The only requirement+-- is an Ord instance for comparing two times, to use the above rules.  The item+-- in a clock may be an Int, a Double, an Integer, or even types like Bool or Either,+-- your own types or newtypes, or things like lists!+--+-- The following rules apply to clocks:+--+-- * If there are no processes enrolled, the time never changes.+--+-- * Time never advances (and processes are never woken up) until all processes+-- enrolled on the clock have waited for a time (either the next available, or+-- a specific time).+-- +-- * If all processes enrolled wait for the next available time, they will not+-- wake up (until another process enrolls and asks for a specific time).  To make+-- sure that time advances, use the 'Control.Concurrent.CHP.Common.advanceTime'+-- process.+-- +-- * The clock always advances to the earliest (minimum) specific offer that is+-- stricly after the current time, unless:+-- +-- * If all processes that are waiting for specific times, ask for times that+-- are before the current time, the earliest (minimum) of these is taken, and+-- thus time effectively moves backwards (wraps around).  In this case, all+-- processes waiting for the next time will also be woken up.+--+-- * As a consequence of all of the above, if you wait and return being told the+-- current time, that time cannot change until you next wait, or if you resign+-- from the clock (temporarily or permanently).+--+-- * Note that waiting for clocks cannot be used as part of a choice+-- ('Control.Concurrent.CHP.Alt.alt' and 'Control.Concurrent.CHP.Alt.every').+-- The semantics of allowing this are unclear.  If a clock waits for time 5,+-- but later backs out, should it be possible for two other processes to+-- advance the time to 3 in the mean-time?  Due to this, clocks cannot be used+-- in a choice.  If you want to have a choice involving a time change, have a+-- process that waits for the next available time, then sends it down a+-- channel to the process making the choice.+--+-- Clocks are similar to phased barriers (indeed, both have an instance of+-- 'Waitable').  The fundamental differences are:+--+-- * A barrier can only move one phase at a time.  If you use barriers to skip+-- past several phases at once, this will be much less efficient than using a clock.+-- This is also true if not every process enrolled on a barrier wants to take action+-- every phase -- a clock allows those processes to remain sleeping, rather than+-- wake up only to sleep again,+-- * Barriers support choice, but clocks do not.  This means clocks are both+-- less powerful, but also faster than barriers.+-- * Barriers choose their next phase using their incrementing function.  Clocks+-- are more flexible, in that their next phase is chosen solely by looking at the+-- requests from the various processes.  Hence why Double is a suitable type for+-- a Clock time, but not for a PhasedBarrier phase.+-- +-- This whole module was first added in version 1.2.0.+module Control.Concurrent.CHP.Clocks (Clock, Waitable(..),+  waitUnbounded, newClock, newClockWithLabel) where++import Control.Concurrent.STM+import Control.Monad hiding (mapM, mapM_)+import Control.Monad.State (get)+import Control.Monad.Trans+-- Needed for testing:+--import Data.Maybe+import qualified Data.Sequence as Seq+import qualified Data.Set as Set+import Data.Traversable+import Data.Unique+import Prelude hiding (mapM, mapM_)++import Control.Concurrent.CHP.Barriers+import Control.Concurrent.CHP.Base+import qualified Control.Concurrent.CHP.Event as Event+import Control.Concurrent.CHP.Enroll+import Control.Concurrent.CHP.Poison+import Control.Concurrent.CHP.ProcessId+import Control.Concurrent.CHP.Traces.Base++-- | A type-class for things that you can on for a specific time\/phase.  The+-- instance for 'PhasedBarrier' repeatedly syncs until the specific phase is+-- reached.  Clock waits until the time is reached.+class Waitable c where+  -- | Given an enrolled item, waits for the specific time\/phase (if+  -- you pass a Just value) or the next available time\/phase. (if you+  -- pass Nothing).  The value returned is the new current time.  Note that+  -- waiting for the current time\/phase or a past time\/phase on a+  -- clock\/barrier will /not/ return immediately -- see the rules at the top+  -- of this module, and see 'waitUnbounded'.+  wait :: Ord t => Enrolled c t -> Maybe t -> CHP t+  -- | Gets the current time\/phase.+  getCurrentTime :: Ord t => Enrolled c t -> CHP t++instance Waitable PhasedBarrier where+  wait eb Nothing = syncBarrier eb+  -- If they ask for the current time, they will always go around again, so we+  -- sync once then wait for the next phase (which may require no further syncs)+  wait eb (Just ph) = syncBarrier eb >> waitForPhase ph eb >> return ph+  getCurrentTime = currentPhase++{- This was perhaps an instance too far:+instance Waitable BroadcastChanin where+  wait ec Nothing = readChannel ec+  wait ec (Just t) = do x <- readChannel ec+                        if x == t+                          then return x+                          else wait ec (Just t)+  getCurrentTime = readChannel+-}++-- | A clock that measures time using the given type.  Only Ord is required on+-- the type, so it can be all sorts.  Obvious possibilities are numeric types such+-- as Int and Double -- if you want monotonically-increasing time, see 'waitUnbounded'.+--  Other possibilities include your own algebraic types, if you want a clock that+-- cycles around a given set of phases.  If every process enrolled on the clock+-- always just waits for the next time, you may want to consider using a 'PhasedBarrier'.+--  If you want a clock that works in reverse or anything else strange, you can+-- always wrap your type in a newtype to give a custom Ord instance.+--+-- See the documentation at the beginning of this module for more information on+-- clocks.+newtype Ord time => Clock time+  = Clock (TVar (WithPoison (TimerData time)), Unique, time -> String)++-- | Normally, when waiting on a clock, if you wait for a time equal to (or earlier+-- than) the current time, you will block until the clock wraps around.  Sometimes,+-- however, you may want your clock to never wrap around (and use Integer as the+-- inner type, usually), and want to make sure that if a process waits for the+-- current time or earlier, it returns instantly.  That is what this function achieves.+--+-- Calling this function with Nothing has indentical behaviour to calling 'wait'+-- with Nothing.  If you wait for the current time or earlier, all of the other+-- processes waiting on the clock will remain blocked.  Processes who have asked+-- to wait for the current time will remain blocked -- it is generally not useful+-- to mix 'waitUnbounded' and 'wait' on the same clock.+waitUnbounded :: (Waitable c, Ord t) => Enrolled c t -> Maybe t -> CHP t+waitUnbounded clock Nothing = wait clock Nothing+waitUnbounded clock (Just waitT)+  = do realT <- getCurrentTime clock+       if waitT <= realT+         then return realT+         else wait clock (Just waitT)++modifyTVar :: TVar (WithPoison a) -> (a -> a) -> STM (WithPoison ())+modifyTVar tv f = do x <- readTVar tv+                     case x of+                       PoisonItem -> return PoisonItem+                       NoPoison y -> do writeTVar tv $ NoPoison $ f y+                                        return $ NoPoison ()++modifyTVar' :: TVar (WithPoison a) -> (a -> STM a) -> STM (WithPoison ())+modifyTVar' tv f = do x <- readTVar tv+                      case x of+                        PoisonItem -> return PoisonItem+                        NoPoison y -> do f y >>= writeTVar tv . NoPoison+                                         return $ NoPoison ()+++poisonTVar :: TVar (WithPoison a) -> STM ()+poisonTVar = flip writeTVar PoisonItem++-- Provides mapM_ for Traversable:+mapM_ :: (Traversable t, Monad m) => (a -> m b) -> t a -> m ()+mapM_ f x = mapM f x >> return ()++data TimerData time+  = TimerData {+      curTime :: time+     ,enrolledOnTimer :: Int+     -- A slightly more efficient way of knowing current offers:+     ,offeredOnTimer :: Int+      -- Offers are held, sorted by time.  We rely on the fact that for all x,+      -- Nothing < Just x+     ,timerOffersNext :: Maybe ([ProcessId], TVar (WithPoison (Maybe time)))+     ,timerOffersBefore :: [([ProcessId], (time, TVar (WithPoison (Maybe time))))]+     ,timerOffersAfter :: [([ProcessId], (time, TVar (WithPoison (Maybe time))))]+     ,timerEventPool :: Seq.Seq (TVar (WithPoison (Maybe time)))+     }+-- Uncomment these lines while testing:+--  deriving (Eq, Show)+--instance Show (TVar (WithPoison (Maybe a))) where show = const "<tv>"++emptyTimerData :: time -> TimerData time+emptyTimerData t = TimerData t 0 0 Nothing [] [] Seq.empty++enrollTimerData :: Maybe (TVar (WithPoison (Maybe time))) -> TimerData time -> TimerData time+enrollTimerData me td+  = td {enrolledOnTimer = enrolledOnTimer td + 1+       -- It's important that the event goes on the front, so that we don't re-use+       -- the event at the back until necessary:+       ,timerEventPool = maybe id (Seq.<|) me $ timerEventPool td}++resignTimerData :: Bool -> TimerData time -> TimerData time+resignTimerData removeOneFromPool td+  = td {enrolledOnTimer = enrolledOnTimer td - 1+       ,timerEventPool = case (Seq.viewl $ timerEventPool td, removeOneFromPool) of+         (_ Seq.:< es, True) -> es+         _ -> timerEventPool td}++poisonTimerData :: TimerData time -> STM ()+poisonTimerData td+  = do mapM_ poisonTVar $ map (snd . snd) (timerOffersAfter td)+       mapM_ poisonTVar $ map (snd . snd) (timerOffersBefore td)+       maybe (return ()) (poisonTVar . snd) (timerOffersNext td)+       mapM_ poisonTVar $ timerEventPool td++-- Gets the first spare event and makes sure the value is empty:+spareEvent :: Seq.Seq (TVar (WithPoison (Maybe a)))+  -> STM (TVar (WithPoison (Maybe a)), Seq.Seq (TVar (WithPoison (Maybe a))))+spareEvent evs = case Seq.viewl evs of+      (e Seq.:< es) -> do writeTVar e $ NoPoison Nothing+                          return (e, es)+      _ -> error "Event pool unexpectedly depleted"+++offerTimerData :: forall time. Ord time => ProcessId -> Maybe time -> TimerData time+  -> STM (TimerData time, TVar (WithPoison (Maybe time)))+offerTimerData pid Nothing td = case timerOffersNext td of+  Nothing -> do+    (e, pool) <- spareEvent $ timerEventPool td+    return (td { offeredOnTimer = offeredOnTimer td + 1+               , timerOffersNext = Just ([pid], e)+               , timerEventPool = pool+               }+           , e)+  Just (pids, e) -> return (td { offeredOnTimer = offeredOnTimer td + 1+                               , timerOffersNext = Just (pid:pids, e)+                               }+                           , e)+offerTimerData pid (Just t) td+  | t <= curTime td+    = do (newOffers, newPool, e) <- insert (timerOffersBefore td) (timerEventPool td)+         return (td { offeredOnTimer = offeredOnTimer td + 1+                    , timerOffersBefore = newOffers+                    , timerEventPool = newPool+                    }+                , e)+  | otherwise+    = do (newOffers, newPool, e) <- insert (timerOffersAfter td) (timerEventPool td)+         return (td { offeredOnTimer = offeredOnTimer td + 1+                    , timerOffersAfter = newOffers+                    , timerEventPool = newPool+                    }+                , e)++  where+    insert :: [([ProcessId], (time, TVar (WithPoison (Maybe a))))]+      -> Seq.Seq (TVar (WithPoison (Maybe a)))+      -> STM ( [([ProcessId], (time, TVar (WithPoison (Maybe a))))]+             , Seq.Seq (TVar (WithPoison (Maybe a)))+             , TVar (WithPoison (Maybe a)))+    insert [] pool = do (e, es) <- spareEvent pool+                        return ([([pid], (t, e))], es, e)+    insert (off@(pids, (toff, eoff)):offs) pool+      = case compare toff t of+          LT -> do (offs', es', e') <- insert offs pool+                   return (off:offs', es', e')+          GT -> do (e, es) <- spareEvent pool+                   return (([pid], (t, e)):off:offs, es, e)+          EQ -> return ((pid:pids, (toff, eoff)):offs, pool, eoff)+++instance Ord time =>+  Enrollable Clock time where+  enroll tim@(Clock (tv, u, sh)) f+    = do ev <- liftSTM $ newTVar (NoPoison Nothing)+         liftSTM (modifyTVar tv $ enrollTimerData $ Just ev)+           >>= checkPoison+         x <- f $ Enrolled tim+         ts <- liftPoison $ liftTrace $ liftM snd get+         liftSTM (modifyTVar' tv $ checkCompletion u sh ts . resignTimerData True)+           >>= checkPoison+         return x++  -- For temporary resignations, we don't touch the event pool+  resign (Enrolled (Clock (tv, u, sh))) m+    = do ts <- liftPoison $ liftTrace $ liftM snd get+         liftSTM (modifyTVar' tv (checkCompletion u sh ts . resignTimerData False))+           >>= checkPoison+         x <- m+         liftSTM (modifyTVar tv $ enrollTimerData Nothing)+           >>= checkPoison+         return x         ++checkCompletion :: Ord time => Unique -> (time -> String) -> TraceStore -> TimerData time -> STM (TimerData time)+checkCompletion u sh ts td+  | offeredOnTimer td == enrolledOnTimer td =+      case timerOffersAfter td of+        ((pids, (newT, ev)):rest) -> do+          writeTVar ev $ NoPoison $ Just newT+          maybe (return ()) (flip writeTVar (NoPoison $ Just newT) . snd) (timerOffersNext td)+          recordEventLast [((Event.ClockSync $ sh newT,u)+            , Set.fromList $ pids ++ maybe [] fst (timerOffersNext td))]+            ts+          return $ td { timerOffersAfter = rest+                      , offeredOnTimer =+                          offeredOnTimer td+                            - (length pids + maybe 0 (length . fst) (timerOffersNext td))+                      , curTime = newT+                      , timerOffersNext = Nothing+                      -- The event will only be re-used once we get to the+                      -- end of the list, and thus all the people are ready+                      -- to go again, so there shouldn't be any overlap involving+                      -- re-use+                      , timerEventPool =+                          maybe id (flip (Seq.|>) . snd) (timerOffersNext td)+                            $ timerEventPool td Seq.|> ev+                      }+        [] -> case timerOffersBefore td of+                ((pids, (newT, ev)):rest) -> do+                  writeTVar ev $ NoPoison $ Just newT+                  maybe (return ()) (flip writeTVar (NoPoison $ Just newT) . snd) (timerOffersNext td)+                  return $+                    td { timerOffersAfter = rest+                       , timerOffersBefore = []+                       , offeredOnTimer =+                           offeredOnTimer td+                             - (length pids + maybe 0 (length . fst) (timerOffersNext td))+                       , curTime = newT+                       , timerOffersNext = Nothing+                       -- The event will only be re-used once we get to the+                       -- end of the list, and thus all the people are ready+                       -- to go again, so there shouldn't be any overlap involving+                       -- re-use+                       , timerEventPool =+                          maybe id (flip (Seq.|>) . snd) (timerOffersNext td)+                            (timerEventPool td Seq.|> ev)+                       }+                [] -> return td -- Everyone waiting for the next time!+  | otherwise = return td++waitClock :: Ord time =>+  ProcessId -> TraceStore -> Enrolled Clock time -> Maybe time -> STM (STM (WithPoison time))+waitClock pid ts (Enrolled (Clock (tv, u, sh))) ph+  = do x <- readTVar tv+       case x of+         PoisonItem -> return $ return PoisonItem+         NoPoison td ->+              do (td', ev) <- offerTimerData pid ph td+                 checkCompletion u sh ts td' >>= writeTVar tv . NoPoison+                 return $ waitForJustOrPoison ev++-- | Creates a clock that starts at the given time.  The Show instance is needed+-- to display times in traces.+newClock :: (Ord time, Show time) => time -> CHP (Clock time)+newClock t = do tv <- liftSTM $ newTVar $ NoPoison $ emptyTimerData t+                u <- liftIO $ Event.newEventUnique+                return $ Clock (tv, u, show)++-- | Creates a clock that starts at the given time, and gives it the given+-- label in traces.  The Show instance is needed to display times in traces.+newClockWithLabel :: (Ord time, Show time) =>+  time -> String -> CHP (Clock time)+newClockWithLabel t l = do tv <- liftSTM $ newTVar $ NoPoison $ emptyTimerData t+                           u <- liftIO $ Event.newEventUnique+                           liftPoison $ liftTrace $ labelUnique u l+                           return $ Clock (tv, u, show)++instance Waitable Clock where+  getCurrentTime (Enrolled (Clock (tv, _, _)))+    = liftSTM (liftM (fmap curTime) $ readTVar tv) >>= checkPoison+  wait c@(Enrolled (Clock (_, u, sh))) mt+    = do ts <- liftPoison $ liftTrace $ liftM snd get+         pid <- liftPoison $ liftTrace $ getProcessId+         waitAct <- liftSTM $ waitClock pid ts c mt+         t <- liftSTM waitAct >>= checkPoison+         liftPoison $ liftTrace $ recordEvent [ClockSyncIndiv u $ sh t]+         return t++instance Ord time => Poisonable (Enrolled Clock time) where+  poison (Enrolled (Clock (tv,_,_)))+    = liftSTM $ do x <- readTVar tv+                   case x of+                     PoisonItem -> return ()+                     NoPoison td -> do poisonTimerData td+                                       writeTVar tv PoisonItem+  checkForPoison (Enrolled (Clock (tv,_,_)))+    = liftCHP $ liftSTM (readTVar tv) >>= checkPoison . fmap (const ())++{-+test_Clock :: IO ()+test_Clock+  = do let begin = emptyTimerData (7::Int)+       tv <- newTVar' $ NoPoison begin+       tv3 <- replicateM 3 $ newTVar' $ NoPoison Nothing+       let withTV f = atomically $ readTVar tv >>= \(NoPoison x) -> f x+           assertTVs vals = atomically (mapM readTVar tv3) >>=+             zipWithM assert1 (map (==) vals) . zip [0..]+           assert checks = atomically (readTVar tv)+             >>= zipWithM (assert1) checks . zip [0..] . repeat+           assert1 :: Show a => (a -> Bool) -> (Int, WithPoison a) -> IO ()+           assert1 f (n, NoPoison x)+             | f x = return ()+             | otherwise = putStrLn $ "Assertion " ++ show n ++ " failed: " ++ show x+           noComplete = do tvVals <- atomically $ mapM readTVar tv3+                           (td, td') <- atomically $ do+                             NoPoison td <- readTVar tv+                             td' <- checkCompletion (error "Unique") show NoTrace td+                             return (td, td')+                           tvVals' <- atomically $ mapM readTVar tv3+                           if td /= td' || tvVals /= tvVals'+                             then putStrLn "Completion unexpected!"+                             else return ()+           complete = atomically $ readTVar tv >>= \(NoPoison x) ->+             checkCompletion (error "Unique") show NoTrace x+               >>= writeTVar tv . NoPoison+       writeTVar' tv $ NoPoison $ foldr (enrollTimerData) begin (map Just tv3)+       assert [(== Seq.fromList tv3) . timerEventPool+              ,(== 3) . enrolledOnTimer+              ,(== 0) . offeredOnTimer+              ,(== 7) . curTime+              ,isNothing . timerOffersNext+              ,null . timerOffersBefore+              ,null . timerOffersAfter+              ]+       noComplete++       -- This sequence has two guys waiting for the next time, and one waiting+       -- for the next time after:+       withTV (offerTimerData (testProcessId 0) Nothing) >>= \(td, ev) ->+         do writeTVar' tv $ NoPoison td+            assert [(== Seq.fromList (tail tv3)) . timerEventPool+                   ,(== 3) . enrolledOnTimer+                   ,(== 1) . offeredOnTimer+                   ,(== 7) . curTime+                   ,(== Just ([testProcessId 0], head tv3)) . timerOffersNext+                   ,null . timerOffersBefore+                   ,null . timerOffersAfter+                   ,const $ head tv3 == ev+                   ]+       noComplete+       withTV (offerTimerData (testProcessId 1) Nothing) >>= \(td, ev) ->+         do writeTVar' tv $ NoPoison td+            assert [(== Seq.fromList (tail tv3)) . timerEventPool+                   ,(== 3) . enrolledOnTimer+                   ,(== 2) . offeredOnTimer+                   ,(== 7) . curTime+                   ,(== Just ([testProcessId 1, testProcessId 0], head tv3)) . timerOffersNext+                   ,null . timerOffersBefore+                   ,null . timerOffersAfter+                   ,const $ head tv3 == ev+                   ]+       noComplete+       withTV (offerTimerData (testProcessId 2) (Just 9)) >>= \(td, ev) ->+         do writeTVar' tv $ NoPoison td+            assert [(== Seq.fromList [last tv3]) . timerEventPool+                   ,(== 3) . enrolledOnTimer+                   ,(== 3) . offeredOnTimer+                   ,(== 7) . curTime+                   ,(== Just ([testProcessId 1, testProcessId 0], head tv3)) . timerOffersNext+                   ,null . timerOffersBefore+                   ,(== [([testProcessId 2], (9, sec tv3))]) . timerOffersAfter+                   ,const $ sec tv3 == ev+                   ]+       complete+       assertTVs [Just 9, Just 9, Nothing]+       readTVar' tv >>= \td ->+            assert [(== Seq.fromList (reverse tv3)) . timerEventPool+                   ,(== 3) . enrolledOnTimer+                   ,(== 0) . offeredOnTimer+                   ,(== 9) . curTime+                   ,isNothing . timerOffersNext+                   ,null . timerOffersBefore+                   ,null . timerOffersAfter+                   ]+       -- This sequence has one waiting before, one after and one before:+       withTV (offerTimerData (testProcessId 0) (Just 5)) >>= \(td, ev) ->+         do writeTVar' tv $ NoPoison td+            assert [(== Seq.fromList [sec tv3, head tv3]) . timerEventPool+                   ,(== 3) . enrolledOnTimer+                   ,(== 1) . offeredOnTimer+                   ,(== 9) . curTime+                   ,isNothing . timerOffersNext+                   ,(== [([testProcessId 0], (5, last tv3))]) . timerOffersBefore+                   ,null . timerOffersAfter+                   ,const $ last tv3 == ev+                   ]+       assertTVs [Just 9, Just 9, Nothing]+       noComplete+       withTV (offerTimerData (testProcessId 1) (Nothing)) >>= \(td, ev) ->+         do writeTVar' tv $ NoPoison td+            assert [(== Seq.fromList [head tv3]) . timerEventPool+                   ,(== 3) . enrolledOnTimer+                   ,(== 2) . offeredOnTimer+                   ,(== 9) . curTime+                   ,(== Just ([testProcessId 1], sec tv3)) . timerOffersNext+                   ,(== [([testProcessId 0], (5, last tv3))]) . timerOffersBefore+                   ,null . timerOffersAfter+                   ,const $ sec tv3 == ev+                   ]+       assertTVs $ [Just 9, Nothing, Nothing]+       noComplete+       withTV (offerTimerData (testProcessId 2) (Just 11)) >>= \(td, ev) ->+         do writeTVar' tv $ NoPoison td+            assert [(== Seq.fromList []) . timerEventPool+                   ,(== 3) . enrolledOnTimer+                   ,(== 3) . offeredOnTimer+                   ,(== 9) . curTime+                   ,(== Just ([testProcessId 1], sec tv3)) . timerOffersNext+                   ,(== [([testProcessId 0], (5, last tv3))]) . timerOffersBefore+                   ,(== [([testProcessId 2], (11, head tv3))]) . timerOffersAfter+                   ,const $ head tv3 == ev+                   ]+       assertTVs [Nothing, Nothing, Nothing]+       complete+       assertTVs [Just 11, Just 11, Nothing]+       readTVar' tv >>= \td ->+            assert [(== Seq.fromList [head tv3, sec tv3]) . timerEventPool+                   ,(== 3) . enrolledOnTimer+                   ,(== 1) . offeredOnTimer+                   ,(== 11) . curTime+                   ,isNothing . timerOffersNext+                   ,(== [([testProcessId 0], (5, last tv3))]) . timerOffersBefore+                   ,null . timerOffersAfter+                   ]+       -- This sequence has one joining in before on the same time, and one joining+       -- in before on the current time, which should count as before:+       noComplete+       withTV (offerTimerData (testProcessId 1) (Just 5)) >>= \(td, ev) ->+         do writeTVar' tv $ NoPoison td+            assert [(== Seq.fromList [head tv3, sec tv3]) . timerEventPool+                   ,(== 3) . enrolledOnTimer+                   ,(== 2) . offeredOnTimer+                   ,(== 11) . curTime+                   ,isNothing . timerOffersNext+                   ,(== [([testProcessId 1, testProcessId 0], (5, last tv3))]) . timerOffersBefore+                   ,null . timerOffersAfter+                   ,const $ last tv3 == ev+                   ]+       assertTVs [Just 11, Just 11, Nothing]+       noComplete+       withTV (offerTimerData (testProcessId 2) (Just 11)) >>= \(td, ev) ->+         do writeTVar' tv $ NoPoison td+            assert [(== Seq.fromList [sec tv3]) . timerEventPool+                   ,(== 3) . enrolledOnTimer+                   ,(== 3) . offeredOnTimer+                   ,(== 11) . curTime+                   ,isNothing . timerOffersNext+                   ,(== [([testProcessId 1, testProcessId 0], (5, last tv3))+                        ,([testProcessId 2], (11, head tv3))]) . timerOffersBefore+                   ,null . timerOffersAfter+                   ,const $ head tv3 == ev+                   ]+       assertTVs [Nothing, Just 11, Nothing]+       complete+       assertTVs [Nothing, Just 11, Just 5]+       readTVar' tv >>= \td ->+            assert [(== Seq.fromList [sec tv3, last tv3]) . timerEventPool+                   ,(== 3) . enrolledOnTimer+                   ,(== 1) . offeredOnTimer+                   ,(== 5) . curTime+                   ,isNothing . timerOffersNext+                   ,null . timerOffersBefore+                   ,(== [([testProcessId 2], (11, head tv3))]) . timerOffersAfter+                   ]+       return ()+  where+    sec (_:x:_) = x+    sec _ = error "sec"+    readTVar' = atomically . readTVar+    writeTVar' tv = atomically . writeTVar tv+    newTVar' = atomically . newTVar+-}

@@ -49,14 +49,15 @@ import Control.Monad import Control.Parallel.Strategies import qualified Data.Traversable as Traversable-import Prelude (Bool, Maybe(..), Enum, Ord, ($), (<), Int, otherwise, (.))+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 :: (ReadableChannel r, Poisonable (r a),+       WriteableChannel w, Poisonable (w a)) => r a -> w a -> CHP () id in_ out = (forever $   do x <- readChannel in_      writeChannel out x@@ -167,6 +168,34 @@ consume :: Chanin a -> CHP () consume c = (forever $ readChannel c) `onPoisonRethrow` poison c +-- | For the duration of the given process, acts as a consume process, but stops+-- when the given process stops.  Note that there could be a timing issue where+-- extra inputs are consumed at the end of the lifetime of the process.+-- Note also that while poison from the given process will be propagated on the+-- consumption channel, there is no mechanism to propagate poison from the consumption+-- channel into the given process.+--+-- Added in version 1.2.0.+consumeAlongside :: Chanin a -> CHP b -> CHP b+consumeAlongside in_ proc+  = do c <- oneToOneChannelWithLabel "consumeAlongside-Internal"+       (x,_) <- +         ((do x <- proc+              writeChannel (writer c) ()+              return x+          ) `onPoisonRethrow` poison (writer c))+         <||>+         (inner (reader c) `onPoisonRethrow` poison (reader c))+       return x+  where+    inner c = do cont <- alt+                   [readChannel c >> return False+                   ,readChannel in_ >> return True+                   ]+                 if cont+                   then inner c+                   else return ()+ -- | 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@@ -179,6 +208,13 @@   writeChannel out $ f x y   ) `onPoisonRethrow` (poison in0 >> poison in1 >> poison out) +-- | Forever reads a value from all its input channels in parallel, then joins+-- the values into a list in the same order as the channels, and sends them out again.+joinList :: [Chanin a] -> Chanout [a] -> CHP ()+joinList ins out = (forever $ runParallel [readChannel c | c <- ins] >>= writeChannel out+  ) `onPoisonRethrow` (poisonAll ins >> poison out)++ -- | Forever reads a pair from its input channel, then in parallel sends out -- the first and second parts of the pair on its output channels. --@@ -226,8 +262,14 @@ -- 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+--+-- Note that prior to version 1.2.0 (i.e. in version 1.1.1) there was a bug where+-- poison would not be propagated between the input and output.+valueStore :: (ReadableChannel r, Poisonable (r a),+               WriteableChannel w, Poisonable (w a)) =>+               a -> r a -> w a -> CHP ()+valueStore val input output+  = inner val `onPoisonRethrow` (poison input >> poison output)   where     inner x = ((writeChannel output x >> return x) <-> readChannel input) >>= inner @@ -236,5 +278,22 @@ -- 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+--+-- Note that prior to version 1.2.0 (i.e. in version 1.1.1) there was a bug where+-- poison would not be propagated between the input and output.+valueStore' :: (ReadableChannel r, Poisonable (r a),+               WriteableChannel w, Poisonable (w a)) => r a -> w a -> CHP ()+valueStore' input output+  = (readChannel input >>= \x -> valueStore x input output)+      `onPoisonRethrow` (poison input >> poison output)++-- | Continually waits for a specific time on the given clock, each time applying+-- the function to work out the next specific time to wait for.  The most common+-- thing to pass is Prelude.succ or (+1).+--+-- Added in version 1.2.0.+advanceTime :: (Waitable c, Ord t) => (t -> t) -> Enrolled c t -> CHP ()+advanceTime f c = do t <- getCurrentTime c+                     inner (f t)+  where+    inner t = wait c (Just t) >>= inner . f

@@ -43,7 +43,11 @@ import Control.Concurrent.CHP.Poison import Control.Concurrent.CHP.ProcessId -data RecordedEventType = ChannelComm | BarrierSync deriving (Eq, Ord, Show)+-- | ClockSync was added in version 1.2.0.+data RecordedEventType+  = ChannelComm+  | BarrierSync+  | ClockSync String deriving (Eq, Ord, Show)  -- Not really a CSP event, more like an enrollable poisonable alting barrier! newtype Event = Event (@@ -296,6 +300,9 @@   = do u <- newUnique        atomically $ do tv <- newTVar (NoPoison (n, []))                        return $ Event (u, t, tv)++newEventUnique :: IO Unique+newEventUnique = newUnique  enrollEvent :: Event -> STM (WithPoison ()) enrollEvent e

@@ -29,6 +29,8 @@  module Control.Concurrent.CHP.Poison where +import Control.Concurrent.STM+ -- | A Maybe-like poison wrapper. data WithPoison a = PoisonItem | NoPoison a deriving (Eq, Show) @@ -43,3 +45,17 @@  mergeWithPoison :: [WithPoison a] -> WithPoison () mergeWithPoison = sequence_++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++waitForNothingOrPoison :: TVar (WithPoison (Maybe a)) -> STM (WithPoison ())+waitForNothingOrPoison tv = do x <- readTVar tv+                               case x of+                                 PoisonItem -> return PoisonItem+                                 NoPoison (Just _) -> retry+                                 NoPoison Nothing -> return $ NoPoison ()

@@ -55,32 +55,44 @@ -- 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.+--+-- TimerSyncIndiv was added in version 1.2.0. data RecordedIndivEvent =    ChannelWrite Unique   | ChannelRead Unique   | BarrierSyncIndiv Unique+  | ClockSyncIndiv Unique String   deriving (Eq, Ord)  type RecEvents = ([RecordedEvent], [RecordedIndivEvent]) -getName :: Unique -> State (Map.Map Unique String) String-getName u = do m <- get+getName :: String -> Unique -> State (Map.Map Unique String) String+getName prefix u+          = do m <- get                case Map.lookup u m of                  Just x -> return x-                 Nothing -> let x = "c" ++ show (Map.size m) in+                 Nothing -> let x = prefix ++ show (Map.size m) in                             do put $ Map.insert u x m                                return x  nameEvent :: RecordedEvent -> State (Map.Map Unique String) String-nameEvent (_, c) = getName c+nameEvent (t, c) = liftM (++ suffix) $ getName prefix c+  where+    (prefix, suffix) = case t of+      ChannelComm -> ("_c","")+      BarrierSync -> ("_b","")+      ClockSync st -> ("_t", ':' : st)  nameIndivEvent :: RecordedIndivEvent -> State (Map.Map Unique String) String-nameIndivEvent (ChannelWrite c) = do c' <- getName c+nameIndivEvent (ChannelWrite c) = do c' <- getName "_c" c                                      return $ c' ++ "!"-nameIndivEvent (ChannelRead c) = do c' <- getName c+nameIndivEvent (ChannelRead c) = do c' <- getName "_c" c                                     return $ c' ++ "?"-nameIndivEvent (BarrierSyncIndiv c) = do c' <- getName c+nameIndivEvent (BarrierSyncIndiv c) = do c' <- getName "_b" c                                          return $ c' ++ "*"+nameIndivEvent (ClockSyncIndiv c t) = do c' <- getName "_t" c+                                         return $ c' ++ ":" ++ t+  ensureAllNamed :: Map.Map Unique String -> [RecordedEvent] -> Map.Map Unique String -- Quite hacky:

@@ -1,5 +1,5 @@ -- Communicating Haskell Processes.--- Copyright (c) 2008, University of Kent.+-- Copyright (c) 2008--2009, University of Kent. -- All rights reserved. --  -- Redistribution and use in source and binary forms, with or without@@ -95,12 +95,32 @@ -- 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 ()) ->+--+-- The type for this process became more specific in version 1.2.0.+(|->|) :: (a -> Chanout b ->  CHP ()) -> (Chanin b -> c -> CHP ()) ->   (a -> c -> CHP ())-(|->|) p q x y = do c <- newChannel+(|->|) p q x y = do c <- oneToOneChannel                     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 ()) ->+--+-- The type for this process became more specific in version 1.2.0.+(|<-|) :: (Chanin b -> c ->  CHP ()) -> (a -> Chanout b -> CHP ()) ->   (a -> c -> CHP ()) (|<-|) = flip (|->|)++-- | A function to use at the start of a pipeline you are chaining together with+-- the '(|->|)' operator.+-- Added in version 1.2.0.+(->|) :: (Chanout b -> CHP ()) -> (Chanin b -> c -> CHP ())+  -> (c -> CHP ())+(->|) p q x = do c <- oneToOneChannel+                 runParallel_ [p (writer c), q (reader c) x]++-- | A function to use at the end of a pipeline you are chaining together with+-- the '(|->|)' operator.+-- Added in version 1.2.0.+(|->) :: (a -> Chanout b -> CHP ()) -> (Chanin b -> CHP ())+  -> (a -> CHP ())+(|->) p q x = do c <- oneToOneChannel+                 runParallel_ [p x (writer c), q (reader c)]

@@ -1,5 +1,5 @@ Name:            chp-Version:         1.1.1+Version:         1.2.0 Synopsis:        An implementation of concurrency ideas from Communicating Sequential Processes License:         BSD3 License-file:    LICENSE@@ -21,12 +21,14 @@ Build-Depends:   base, containers, mtl, parallel, pretty, stm  Exposed-modules: Control.Concurrent.CHP+                 Control.Concurrent.CHP.Actions                  Control.Concurrent.CHP.Alt                  Control.Concurrent.CHP.Arrow                  Control.Concurrent.CHP.Barriers                  Control.Concurrent.CHP.BroadcastChannels                  Control.Concurrent.CHP.Buffers                  Control.Concurrent.CHP.Channels+                 Control.Concurrent.CHP.Clocks                  Control.Concurrent.CHP.Common                  Control.Concurrent.CHP.Console                  Control.Concurrent.CHP.Enroll