chp 1.5.0 → 1.5.1

raw patch · 4 files changed

+50/−19 lines, 4 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• Control/Concurrent/CHP/Base.hs +20/−11
• Control/Concurrent/CHP/Channels/BroadcastReduce.hs +4/−4
• Control/Concurrent/CHP/Parallel.hs +25/−3
• chp.cabal +1/−1

@@ -242,17 +242,24 @@        t' <- f l        return (x, t') -newtype ManyToOneTVar a = ManyToOneTVar (TVar a, TVar (Maybe a))-  deriving Eq+data ManyToOneTVar a = ManyToOneTVar+  { mtoIsFinalValue :: a -> Bool+  , mtoReset :: STM a+  , mtoInter :: TVar a+  , mtoFinal :: TVar (Maybe a)+  } -newManyToOneTVar :: a -> STM (ManyToOneTVar a)-newManyToOneTVar x+instance Eq (ManyToOneTVar a) where+  (==) tvA tvB = mtoFinal tvA == mtoFinal tvB++newManyToOneTVar :: (a -> Bool) -> STM a -> a -> STM (ManyToOneTVar a)+newManyToOneTVar f r x   = do tvInter <- newTVar x-       tvFinal <- newTVar Nothing-       return $ ManyToOneTVar (tvInter, tvFinal)+       tvFinal <- newTVar $ if f x then Just x else Nothing+       return $ ManyToOneTVar f r tvInter tvFinal -writeManyToOneTVar :: (a -> Bool, STM a) -> (a -> a) -> ManyToOneTVar a -> STM a-writeManyToOneTVar (done, reset) f (ManyToOneTVar (tvInter, tvFinal))+writeManyToOneTVar :: (a -> a) -> ManyToOneTVar a -> STM a+writeManyToOneTVar f (ManyToOneTVar done reset tvInter tvFinal)   = do x <- readTVar tvInter        if done (f x)          then do writeTVar tvFinal $ Just $ f x@@ -261,14 +268,16 @@        return $ f x  readManyToOneTVar :: ManyToOneTVar a -> STM a-readManyToOneTVar (ManyToOneTVar (_tvInter, tvFinal))+readManyToOneTVar (ManyToOneTVar _done _reset _tvInter tvFinal)   = do x <- readTVar tvFinal >>= maybe retry return        writeTVar tvFinal Nothing        return x +-- If the value is final, it is stored as final! resetManyToOneTVar :: ManyToOneTVar a -> a -> STM ()-resetManyToOneTVar (ManyToOneTVar (tvInter, tvFinal)) x-  = writeTVar tvInter x >> writeTVar tvFinal Nothing+resetManyToOneTVar (ManyToOneTVar done reset tvInter tvFinal) x+  | done x = (reset >>= writeTVar tvInter) >> writeTVar tvFinal (Just x)+  | otherwise = writeTVar tvInter x >> writeTVar tvFinal Nothing  -- ========== -- Instances: 

@@ -122,7 +122,7 @@            (resetManyToOneTVar tvAck . pred) $ Enrolled b          x <- liftIO $ atomically $ readTVar tvSend >>= maybe retry return          y <- f x-         liftIO $ atomically $ writeManyToOneTVar ((== 0), return 0) pred tvAck+         liftIO $ atomically $ writeManyToOneTVar pred tvAck          return y  instance Poisonable (BroadcastChanout a) where@@ -139,7 +139,7 @@        -- Writer is always enrolled:        liftIO $ atomically $ enrollEvent e        tvSend <- liftIO $ atomically $ newTVar Nothing-       tvAck <- liftIO $ atomically $ newManyToOneTVar 0+       tvAck <- liftIO $ atomically $ newManyToOneTVar (== 0) (return 0) 0        return $ BC (b, tvSend, tvAck)  instance Channel BroadcastChanin BroadcastChanout where@@ -213,7 +213,7 @@          (_, Just (_, rtvb)) <- liftIO . atomically $ do            tvb <- newTVar False            let upd (n, mx) = (pred n, Just $ maybe (x, tvb) (first $ f x) mx)-           writeManyToOneTVar ((== 0) . fst, return (0, Nothing)) upd tv+           writeManyToOneTVar upd tv          -- Has to be two separate transactions          liftIO $ atomically $ readTVar rtvb >>= flip unless retry          return r@@ -241,7 +241,7 @@     do b@(Barrier (e, _, _)) <- newBarrier        -- Writer is always enrolled:        liftIO $ atomically $ enrollEvent e-       mtv <- liftIO $ atomically $ newManyToOneTVar (0, Nothing)+       mtv <- liftIO $ atomically $ newManyToOneTVar ((== 0) . fst) (return (0, Nothing)) (0, Nothing)        return $ GC (b, mtv, (mappend, mempty))  -- | The reduce channel version of sameChannel.

@@ -28,7 +28,7 @@ -- SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.  module Control.Concurrent.CHP.Parallel (runParallel, runParallel_, (<||>), (<|*|>),-  ForkingT, forking, fork) where+  runParMapM, runParMapM_, ForkingT, forking, fork) where  import Control.Concurrent import Control.Concurrent.STM@@ -56,9 +56,27 @@ -- | 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.+--+-- In version 1.5.0, a bug was introduced such that @runParallel []@ would deadlock;+-- this was fixed in version 1.5.1. runParallel :: [CHP a] -> CHP [a] runParallel = runParallelPoison +-- | A shorthand for applying mapM in parallel; really the composition of 'runParallel'+-- and 'map'.+--+-- Added in version 1.5.0.+runParMapM :: (a -> CHP b) -> [a] -> CHP [b]+runParMapM f = runParallel . map f++-- | A shorthand for applying mapM_ in parallel; really the composition of 'runParallel_'+-- and 'map'.+--+-- Added in version 1.5.0.+runParMapM_ :: (a -> CHP b) -> [a] -> CHP ()+runParMapM_ f = runParallel_ . map f++ -- | 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@@ -82,6 +100,9 @@ -- | 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 ().+--+-- In version 1.5.0, a bug was introduced such that @runParallel_ []@ would deadlock;+-- this was fixed in version 1.5.1. runParallel_ :: [CHP a] -> CHP () runParallel_ procs = runParallel procs >> return () @@ -96,13 +117,14 @@ -- length and ordering of the input runParallelPoison :: forall a. [CHP a] -> CHP [a] runParallelPoison processes-  = do resultVar <- liftIO $ atomically $ newManyToOneTVar []+  = do resultVar <- liftIO $ atomically $ newManyToOneTVar+         ((== length processes) . length) (return []) []        trace <- PoisonT $ lift $ liftTrace ask        blanks <- liftIO $ blankTraces trace (length processes)        liftIO $           mapM_ forkIO [do y <- wrapProcess p $ flip runReaderT btr . pullOutStandard                           C.block $ atomically $-                            writeManyToOneTVar ((== length processes) . length, return []) +                            writeManyToOneTVar                              ((:) (case y of                                  Nothing -> (n, Nothing)                                  Just (Right x) -> (n, Just x)

@@ -1,5 +1,5 @@ Name:            chp-Version:         1.5.0+Version:         1.5.1 Synopsis:        An implementation of concurrency ideas from Communicating Sequential Processes License:         BSD3 License-file:    LICENSE