pipes-concurrency 2.0.2 → 2.0.3

raw patch · 2 files changed

+53/−48 lines, 2 filesdep ~stm

Dependency ranges changed: stm

Files

• pipes-concurrency.cabal +5/−5
• src/Pipes/Concurrent.hs +48/−43

@@ -1,5 +1,5 @@ Name: pipes-concurrency-Version: 2.0.2+Version: 2.0.3 Cabal-Version: >=1.8.0.2 Build-Type: Simple License: BSD3@@ -31,9 +31,9 @@ Library     Hs-Source-Dirs: src     Build-Depends:-        base         >= 4   && < 5  ,-        pipes        >= 4.0 && < 4.2,-        stm          >= 2.4 && < 2.5+        base         >= 4     && < 5  ,+        pipes        >= 4.0   && < 4.2,+        stm          >= 2.4.3 && < 2.5     Exposed-Modules:         Pipes.Concurrent,         Pipes.Concurrent.Tutorial@@ -47,5 +47,5 @@         base              >= 4     && < 5  ,         pipes             >= 4.0.0 && < 4.2,         pipes-concurrency >= 2.0.0 && < 2.1,-        stm               >= 2.4   && < 2.5,+        stm               >= 2.4.3 && < 2.5,         async             >= 2.0   && < 2.1

@@ -1,26 +1,6 @@ -- | Asynchronous communication between pipes -{-# LANGUAGE RankNTypes, Trustworthy #-}--{- 'unsafeIOToSTM' requires the Trustworthy annotation.--    I use 'unsafeIOToSTM' to touch IORefs to mark them as still alive. This-    action satisfies the necessary safety requirements because:--    * You can safely repeat it if the transaction rolls back--    * It does not acquire any resources--    * It does not leak any inconsistent view of memory to the outside world--    It appears to be unnecessary to read the IORef to keep it from being garbage-    collected, but I wanted to be absolutely certain since I cannot be sure that-    GHC won't optimize away the reference to the IORef.--    The other alternative was to make 'send' and 'recv' use the 'IO' monad-    instead of 'STM', but I felt that it was important to preserve the ability-    to combine them into larger transactions.--}+{-# LANGUAGE RankNTypes, Safe #-}  module Pipes.Concurrent (     -- * Inputs and Outputs@@ -35,6 +15,10 @@     spawn,     spawn',     Buffer(..),+    unbounded,+    bounded,+    latest,+    newest,      -- * Re-exports     -- $reexport@@ -46,12 +30,10 @@ import Control.Applicative (     Alternative(empty, (<|>)), Applicative(pure, (*>), (<*>)), (<*), (<$>) ) import Control.Concurrent (forkIO)-import Control.Concurrent.STM (atomically, STM)+import Control.Concurrent.STM (atomically, STM, mkWeakTVar, newTVarIO, readTVar) import qualified Control.Concurrent.STM as S-import Control.Monad (when)-import Data.IORef (newIORef, readIORef, mkWeakIORef)+import Control.Monad (when,void, MonadPlus(..)) import Data.Monoid (Monoid(mempty, mappend))-import GHC.Conc.Sync (unsafeIOToSTM) import Pipes (MonadIO(liftIO), yield, await, Producer', Consumer') import System.Mem (performGC) @@ -85,6 +67,10 @@             Nothing -> recv i             Just a  -> return (Just a) +instance MonadPlus Input where+    mzero = empty+    mplus = (<|>)+ instance Monoid (Input a) where     mempty = empty     mappend = (<|>)@@ -192,14 +178,14 @@     sealed <- S.newTVarIO False     let seal = S.writeTVar sealed True -    {- Use IORefs to keep track of whether the 'Input' or 'Output' has been+    {- Use weak TVars to keep track of whether the 'Input' or 'Output' has been        garbage collected.  Seal the mailbox when either of them becomes garbage        collected.     -}-    rSend <- newIORef ()-    mkWeakIORef rSend (S.atomically seal)-    rRecv <- newIORef ()-    mkWeakIORef rRecv (S.atomically seal)+    rSend <- newTVarIO ()+    void $ mkWeakTVar rSend (S.atomically seal)+    rRecv <- newTVarIO ()+    void $ mkWeakTVar rRecv (S.atomically seal)      let sendOrEnd a = do             b <- S.readTVar sealed@@ -212,30 +198,49 @@             b <- S.readTVar sealed             S.check b             return Nothing )-        _send a = sendOrEnd a <* unsafeIOToSTM (readIORef rSend)-        _recv   = readOrEnd   <* unsafeIOToSTM (readIORef rRecv)+        _send a = sendOrEnd a <* readTVar rSend+        _recv   = readOrEnd   <* readTVar rRecv     return (Output _send, Input _recv, seal) {-# INLINABLE spawn' #-}  -- | 'Buffer' specifies how to buffer messages stored within the mailbox data Buffer a-    -- | Store an 'Unbounded' number of messages in a FIFO queue     = Unbounded-    -- | Store a 'Bounded' number of messages, specified by the 'Int' argument     | Bounded Int-    -- | Store a 'Single' message (like @Bounded 1@, but more efficient)     | Single-    {-| Only store the 'Latest' message, beginning with an initial value--        'Latest' is never empty nor full.-    -}     | Latest a-    {-| Like @Bounded@, but 'send' never fails (the buffer is never full).-        Instead, old elements are discard to make room for new elements-    -}     | Newest Int-    -- | Like @Newest 1@, but more efficient     | New++{-# DEPRECATED Unbounded "Use `unbounded` instead" #-}+{-# DEPRECATED Bounded "Use `bounded` instead" #-}+{-# DEPRECATED Single "Use @`bounded` 1@ instead" #-}+{-# DEPRECATED Latest "Use `latest` instead" #-}+{-# DEPRECATED Newest "Use `newest` instead" #-}+{-# DEPRECATED New "Use @`newest` 1@ instead" #-}++-- | Store an unbounded number of messages in a FIFO queue+unbounded :: Buffer a+unbounded = Unbounded++-- | Store a bounded number of messages, specified by the 'Int' argument+bounded :: Int -> Buffer a+bounded 1 = Single+bounded n = Bounded n++{-| Only store the 'Latest' message, beginning with an initial value++    'Latest' is never empty nor full.+-}+latest :: a -> Buffer a+latest = Latest++{-| Like @Bounded@, but 'send' never fails (the buffer is never full).+    Instead, old elements are discard to make room for new elements+-}+newest :: Int -> Buffer a+newest 1 = New+newest n = Newest n  {- $reexport     @Control.Concurrent@ re-exports 'forkIO', although I recommend using the