pipes-concurrency 1.2.1 → 2.0.0

raw patch · 8 files changed

+1245/−1065 lines, 8 filesdep +asyncdep +pipes-concurrencydep ~basedep ~pipessetup-changedPVP ok

version bump matches the API change (PVP)

Dependencies added: async, pipes-concurrency

Dependency ranges changed: base, pipes

API changes (from Hackage documentation)

Files

• Control/Proxy/Concurrent.hs +0/−228
• Control/Proxy/Concurrent/Tutorial.hs +0/−772
• LICENSE +24/−24
• Setup.hs +2/−2
• pipes-concurrency.cabal +51/−39
• src/Pipes/Concurrent.hs +234/−0
• src/Pipes/Concurrent/Tutorial.hs +812/−0
• tests/tests-main.hs +122/−0

@@ -1,228 +0,0 @@--- | Asynchronous communication between proxies
-
-{-# LANGUAGE CPP #-}
-
-#if __GLASGOW_HASKELL__ >= 702
-{-# LANGUAGE Trustworthy #-}
-#endif
-{- '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.
--}
-
-module Control.Proxy.Concurrent (
-    -- * Spawn mailboxes
-    spawn,
-    Buffer(..),
-    Input,
-    Output,
-
-    -- * Send and receive messages
-    send,
-    recv,
-
-    -- * Proxy utilities
-    sendD,
-    recvS,
-
-    -- * Re-exports
-    -- $reexport
-    module Control.Concurrent,
-    module Control.Concurrent.STM,
-    module System.Mem
-    ) where
-
-import Control.Applicative (
-    Alternative(empty, (<|>)), Applicative(pure, (<*>)), (<*), (<$>) )
-import Control.Concurrent (forkIO)
-import Control.Concurrent.STM (atomically, STM)
-import qualified Control.Concurrent.STM as S
-import qualified Control.Proxy as P
-import Data.IORef (newIORef, readIORef, mkWeakIORef)
-import Data.Monoid (Monoid(mempty, mappend))
-import GHC.Conc.Sync (unsafeIOToSTM)
-import System.Mem (performGC)
-
-{-| Spawn a mailbox that has an 'Input' and 'Output' end, using the specified
-    'Buffer' to store messages
--}
-spawn :: Buffer a -> IO (Input a, Output a)
-spawn buffer = do
-    (read, write) <- case buffer of
-        Bounded n -> do
-            q <- S.newTBQueueIO n
-            return (S.readTBQueue q, S.writeTBQueue q)
-        Unbounded -> do
-            q <- S.newTQueueIO
-            return (S.readTQueue q, S.writeTQueue q)
-        Single    -> do
-            m <- S.newEmptyTMVarIO
-            return (S.takeTMVar m, S.putTMVar m)
-        Latest a  -> do
-            t <- S.newTVarIO a
-            return (S.readTVar t, S.writeTVar t)
-
-    {- Use an IORef to keep track of whether the 'Input' end has been garbage
-       collected and run a finalizer when the collection occurs
-    -}
-    rSend    <- newIORef ()
-    doneSend <- S.newTVarIO False
-    mkWeakIORef rSend (S.atomically $ S.writeTVar doneSend True)
-
-    {- Use an IORef to keep track of whether the 'Output' end has been garbage
-       collected and run a finalizer when the collection occurs
-    -}
-    rRecv    <- newIORef ()
-    doneRecv <- S.newTVarIO False
-    mkWeakIORef rRecv (S.atomically $ S.writeTVar doneRecv True)
-
-    let sendOrEnd a = do
-            b <- S.readTVar doneRecv
-            if b
-                then return False
-                else do
-                    write a
-                    return True
-        {- The '_send' action aborts without writing a value to the 'Buffer' if
-           the 'Output' has been garbage collected, since there is no point
-           wasting memory if nothing can empty the mailbox.  This protects
-           against careless users not checking send's return value, especially
-           if they use a mailbox of 'Unbounded' size.
-        -}
-        readOrEnd = (Just <$> read) <|> (do
-            b <- S.readTVar doneSend
-            S.check b
-            return Nothing )
-        _send a = sendOrEnd a <* unsafeIOToSTM (readIORef rSend)
-        _recv   = readOrEnd   <* unsafeIOToSTM (readIORef rRecv)
-    return (Input _send, Output _recv)
-{-# INLINABLE spawn #-}
-
-{-| 'Buffer' specifies how to store messages sent to the 'Input' end until the
-    'Output' receives them.
--}
-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
-    {-| Store the 'Latest' message, beginning with an initial value
-
-        'Latest' is never empty nor full.
-    -}
-    | Latest a
-
--- | Accepts messages for the mailbox
-newtype Input a = Input {
-    {-| Send a message to the mailbox
-
-        * Fails and returns 'False' if the mailbox's 'Output' has been garbage
-          collected (even if the mailbox is not full), otherwise it:
-
-        * Retries if the mailbox is full, or:
-
-        * Succeeds if the mailbox is not full and returns 'True'.
-    -}
-    send :: a -> S.STM Bool }
-
-instance Monoid (Input a) where
-    mempty  = Input (\_ -> return False)
-    mappend i1 i2 = Input (\a -> (||) <$> send i1 a <*> send i2 a)
-
--- | Retrieves messages from the mailbox
-newtype Output a = Output {
-    {-| Receive a message from the mailbox
-
-        * Succeeds and returns a 'Just' if the mailbox is not empty, otherwise
-          it:
-
-        * Retries if mailbox's 'Input' has not been garbage collected, or:
-
-        * Fails if the mailbox's 'Input' has been garbage collected and returns
-          'Nothing'.
-    -}
-    recv :: S.STM (Maybe a) }
-
-instance Functor Output where
-    fmap f m = Output (fmap (fmap f) (recv m))
-
-instance Applicative Output where
-    pure r    = Output (pure (pure r))
-    mf <*> mx = Output ((<*>) <$> recv mf <*> recv mx)
-
-instance Monad Output where
-    return r = Output (return (return r))
-    m >>= f  = Output $ do
-        ma <- recv m
-        case ma of
-            Nothing -> return Nothing
-            Just a  -> recv (f a)
-
--- Deriving 'Alternative'
-instance Alternative Output where
-    empty   = Output empty
-    x <|> y = Output (recv x <|> recv y)
-
-{-| Writes all messages flowing \'@D@\'ownstream to the given 'Input'
-
-    'sendD' terminates when the corresponding 'Output' is garbage collected.
-
-> sendD :: (Proxy p) => Input a -> () -> Pipe p a a IO ()
--}
-sendD :: (P.Proxy p) => Input a -> x -> p x a x a IO ()
-sendD input = P.runIdentityK loop
-  where
-    loop x = do
-        a <- P.request x
-        alive <- P.lift $ S.atomically $ send input a
-        if alive
-            then do
-                x2 <- P.respond a
-                loop x2
-            else return ()
-{-# INLINABLE sendD #-}
-
-{-| Convert an 'Output' to a 'P.Producer'
-
-    'recvS' terminates when the 'Buffer' is empty and the corresponding 'Input'
-    is garbage collected.
-
-> recvS :: (Proxy p) => Output a -> () -> Producer p a IO ()
--}
-recvS :: (P.Proxy p) => Output a -> r -> p x' x y' a IO r
-recvS output r = P.runIdentityP go
-  where
-    go = do
-        ma <- P.lift $ S.atomically $ recv output
-        case ma of
-            Nothing -> return r
-            Just a  -> do
-                P.respond a
-                go
-{-# INLINABLE recvS #-}
-
-{- $reexport
-    @Control.Concurrent@ re-exports 'forkIO', although I recommend using the
-    @async@ library instead.
-
-    @Control.Concurrent.STM@ re-exports 'atomically' and 'STM'.
-
-    @System.Mem@ re-exports 'performGC'.
--}

@@ -1,772 +0,0 @@-{-| This module provides a tutorial for the @pipes-concurrency@ library.
-
-    This tutorial assumes that you have read the @pipes@ tutorial in
-    @Control.Proxy.Tutorial@.
-
-    I've condensed all the code examples into self-contained code listings in
-    the Appendix section that you can use to follow along.
--}
-
-module Control.Proxy.Concurrent.Tutorial (
-    -- * Introduction
-    -- $intro
-
-    -- * Work Stealing
-    -- $steal
-
-    -- * Termination
-    -- $termination
-
-    -- * Mailbox Sizes
-    -- $mailbox
-
-    -- * Broadcasts
-    -- $broadcast
-
-    -- * Updates
-    -- $updates
-
-    -- * Callbacks
-    -- $callback
-
-    -- * Safety
-    -- $safety
-
-    -- * Conclusion
-    -- $conclusion
-
-    -- * Appendix
-    -- $appendix
-    ) where
-
-import Control.Proxy
-import Control.Proxy.Concurrent
-import Data.Monoid
-
-{- $intro
-    The @pipes-concurrency@ library provides a simple interface for
-    communicating between concurrent pipelines.  Use this library if you want
-    to:
-
-    * merge multiple streams into a single stream,
-
-    * stream data from a callback \/ continuation,
-
-    * broadcast data,
-
-    * build a work-stealing setup, or
-
-    * implement basic functional reactive programming (FRP).
-
-    For example, let's say that we design a simple game with a single unit's
-    health as the global state.  We'll define an event handler that modifies the
-    unit's health in response to events:
-
-> import Control.Monad
-> import Control.Proxy
-> import Control.Proxy.Trans.Maybe
-> import Control.Proxy.Trans.State
-> 
-> -- The game events
-> data Event = Harm Integer | Heal Integer | Quit
-> 
-> -- The game state
-> type Health = Integer
-> 
-> handler :: (Proxy p) => () -> Consumer (StateP Health (MaybeP p)) Event IO r
-> handler () = forever $ do
->     event <- request ()
->     case event of
->         Harm n -> modify (subtract n)
->         Heal n -> modify (+        n)
->         Quit   -> mzero
->     health <- get
->     lift $ putStrLn $ "Health = " ++ show health
-
-    However, we have two concurrent event sources that we wish to hook up to our
-    event handler.  One translates user input to game events:
-
-> user :: (Proxy p) => () -> Producer p Event IO r
-> user () = runIdentityP $ forever $ do
->     command <- lift getLine
->     case command of
->         "potion" -> respond (Heal 10)
->         "quit"   -> respond  Quit
->         _        -> lift $ putStrLn "Invalid command"
-
-    ... while the other creates inclement weather:
-
-> import Control.Concurrent
->
-> acidRain :: (Proxy p) => () -> Producer p Event IO r
-> acidRain () = runIdentityP $ forever $ do
->     respond (Harm 1)
->     lift $ threadDelay 2000000
-
-    To merge these sources, we 'spawn' a new FIFO mailbox which we will use to
-    merge the two streams of asynchronous events:
-
-> spawn :: Buffer a -> IO (Input a, Output a)
-
-    'spawn' takes a mailbox 'Buffer' as an argument, and we will specify that we
-    want our mailbox to store an 'Unbounded' number of messages:
-
-> import Control.Proxy.Concurrent
->
-> main = do
->     (input, output) <- spawn Unbounded
->     ...
-
-   'spawn' creates this mailbox in the background and then returns two values:
-
-    * an @(Input a)@ that we use to add messages of type @a@ to the mailbox
-
-    * an @(Output a)@ that we use to consume messages of type @a@ from the
-      mailbox
-
-    We will be streaming @Event@s through our mailbox, so our @input@ has type
-    @(Input Event)@ and our @output@ has type @(Output Event)@.
-
-    To stream @Event@s into the mailbox , we use 'sendD', which writes values to
-    the mailbox's 'Input' end:
-
-> sendD :: (Proxy p) => Input a -> () -> Pipe p a a IO ()
-
-    We can concurrently forward multiple streams to the same 'Input', which
-    asynchronously merges their messages into the same mailbox:
-
->     ...
->     forkIO $ do runProxy $ acidRain >-> sendD input
->                 performGC  -- I'll explain 'performGC' below
->     forkIO $ do runProxy $ user     >-> sendD input
->                 performGC
->     ...
-
-    To stream @Event@s out of the mailbox, we use 'recvS', which reads values
-    from the mailbox's 'Output' end:
-
-> recvS :: (Proxy p) => Output a -> () -> Producer p a IO ()
-
-    We will forward our merged stream to our @handler@ so that it can listen to
-    both @Event@ sources:
-
->     ...
->     runProxy $ runMaybeK $ evalStateK 100 $ recvS output >-> handler
-
-    Our final @main@ becomes:
-
-> main = do
->     (input, output) <- spawn Unbounded
->     forkIO $ do runProxy $ acidRain >-> sendD input
->                 performGC
->     forkIO $ do runProxy $ user     >-> sendD input
->                 performGC
->     runProxy $ runMaybeK $ evalStateK 100 $ recvS output >-> handler
-
-    ... and when we run it we get the desired concurrent behavior:
-
-> $ ./game
-> Health = 99
-> Health = 98
-> potion<Enter>
-> Health = 108
-> Health = 107
-> Health = 106
-> potion<Enter>
-> Health = 116
-> Health = 115
-> quit<Enter>
-> $
--}
-
-{- $steal
-    You can also have multiple pipes reading from the same mailbox.  Messages
-    get split between listening pipes on a first-come first-serve basis.
-
-    For example, we'll define a \"worker\" that takes a one-second break each
-    time it receives a new job:
-
-> import Control.Concurrent
-> import Control.Monad
-> import Control.Proxy
-> 
-> worker :: (Proxy p, Show a) => Int -> () -> Consumer p a IO r
-> worker i () = runIdentityP $ forever $ do
->     a <- request ()
->     lift $ threadDelay 1000000  -- 1 second
->     lift $ putStrLn $ "Worker #" ++ show i ++ ": Processed " ++ show a
-
-    Fortunately, these workers are cheap, so we can assign several of them to
-    the same job:
-
-> import Control.Concurrent.Async
-> import Control.Proxy.Concurrent
-> 
-> main = do
->     (input, output) <- spawn Unbounded
->     as <- forM [1..3] $ \i ->
->           async $ do runProxy $ recvS output >-> worker i
->                      performGC
->     a  <- async $ do runProxy $ fromListS [1..10] >-> sendD input
->                      performGC
->     mapM_ wait (a:as)
-
-    The above example uses @Control.Concurrent.Async@ from the @async@ package
-    to fork each thread and wait for all of them to terminate:
-
-> $ ./work
-> Worker #2: Processed 3
-> Worker #1: Processed 2
-> Worker #3: Processed 1
-> Worker #3: Processed 6
-> Worker #1: Processed 5
-> Worker #2: Processed 4
-> Worker #2: Processed 9
-> Worker #1: Processed 8
-> Worker #3: Processed 7
-> Worker #2: Processed 10
-> $
-
-    What if we replace 'fromListS' with a different source that reads lines from
-    user input until the user types \"quit\":
-
-> user :: (Proxy p) => () -> Producer p String IO ()
-> user = stdinS >-> takeWhileD (/= "quit")
-> 
-> main = do
->     (input, output) <- spawn Unbounded
->     as <- forM [1..3] $ \i ->
->           async $ do runProxy $ recvS output >-> worker i
->                      performGC
->     a  <- async $ do runProxy $ user >-> sendD input
->                      performGC
->     mapM_ wait (a:as)
-
-    This still produces the correct behavior:
-
-> $ ./work
-> Test<Enter>
-> Worker #1: Processed "Test"
-> Apple<Enter>
-> Worker #2: Processed "Apple"
-> 42<Enter>
-> Worker #3: Processed "42"
-> A<Enter>
-> B<Enter>
-> C<Enter>
-> Worker #1: Processed "A"
-> Worker #2: Processed "B"
-> Worker #3: Processed "C"
-> quit<Enter>
-> $
--}
-
-{- $termination
-
-    Wait...  How do the workers know when to stop listening for data?  After
-    all, anything that has a reference to 'Input' could potentially add more
-    data to the mailbox.
-
-    It turns out that 'recvS' is smart and only terminates when the upstream
-    'Input' is garbage collected.  'recvS' builds on top of the more primitive
-    'recv' command, which returns a 'Nothing' when the 'Input' is garbage
-    collected:
-
-> recv :: Output a -> STM (Maybe a)
-
-    Otherwise, 'recv' blocks if the mailbox is empty since it assumes that if
-    the 'Input' has not been garbage collected then somebody might still produce
-    more data.
-
-    Does it work the other way around?  What happens if the workers go on strike
-    before processing the entire data set?
-
->     ...
->     as <- forM [1..3] $ \i ->
->           -- Each worker refuses to process more than two values
->           async $ do runProxy $ recvS output >-> takeB_ 2 >-> worker i
->                      performGC
->     ...
-
-    Let's find out:
-
-> $ ./work
-> How<Enter>
-> Worker #1: Processed "How"
-> many<Enter>
-> roads<Enter>
-> Worker #2: Processed "many"
-> Worker #3: Processed "roads"
-> must<Enter>
-> a<Enter>
-> man<Enter>
-> Worker #1: Processed "must"
-> Worker #2: Processed "a"
-> Worker #3: Processed "man"
-> walk<Enter>
-> $
-
-    'sendD' similarly shuts down when the 'Output' is garbage collected,
-    preventing the user from submitting new values.  'sendD' builds on top of
-    the more primitive 'send' command, which returns a 'False' when the 'Output'
-    is garbage collected:
-
-> send :: Input a -> a -> STM Bool
-
-    Otherwise, 'send' blocks if the mailbox is full, since it assumes that if
-    the 'Output' has not been garbage collected then somebody could still
-    consume a value from the mailbox, making room for a new value.
-
-    This is why we have to insert 'performGC' calls whenever we release a
-    reference to either the 'Input' or 'Output'.  Without these calls we cannot
-    guarantee that the garbage collector will trigger and notify the opposing
-    end if the last reference was released.  If you forget to insert a
-    'performGC' call then termination will delay until the next garbage
-    collection cycle.
--}
-
-{- $mailbox
-    So far we haven't observed 'send' blocking because we only 'spawn'ed
-    'Unbounded' mailboxes.  However, we can control the size of the mailbox to
-    tune the coupling between the 'Input' and the 'Output' ends.
-
-    If we set the mailbox 'Buffer' to 'Single', then the mailbox holds exactly
-    one message, forcing synchronization between 'send's and 'recv's.  Let's
-    observe this by sending an infinite stream of values, logging all values to
-    'stdout':
-
-> main = do
->     (input, output) <- spawn Single
->     as <- forM [1..3] $ \i ->
->           async $ do runProxy $ recvS output >-> takeB_ 2 >-> worker i
->                      performGC
->     a  <- async $ do runProxy $ enumFromS 1 >-> printD >-> sendD input
->                      performGC
->     mapM_ wait (a:as)
-
-    The 7th value gets stuck in the mailbox, and the 8th value blocks because
-    the mailbox never clears the 7th value:
-
-> $ ./work
-> 1
-> 2
-> 3
-> 4
-> 5
-> Worker #3: Processed 3
-> Worker #2: Processed 2
-> Worker #1: Processed 1
-> 6
-> 7
-> 8
-> Worker #1: Processed 6
-> Worker #2: Processed 5
-> Worker #3: Processed 4
-> $
-
-    Contrast this with an 'Unbounded' mailbox for the same program, which keeps
-    accepting values until downstream finishes processing the first six values:
-
-> $ ./work
-> 1
-> 2
-> 3
-> 4
-> 5
-> 6
-> 7
-> 8
-> 9
-> ...
-> 487887
-> 487888
-> Worker #3: Processed 3
-> Worker #2: Processed 2
-> Worker #1: Processed 1
-> 487889
-> 487890
-> ...
-> 969188
-> 969189
-> Worker #1: Processed 6
-> Worker #2: Processed 5
-> Worker #3: Processed 4
-> 969190
-> 969191
-> $
-
-    You can also choose something in between by using a 'Bounded' mailbox which
-    caps the mailbox size to a fixed value.  Use 'Bounded' when you want mostly
-    loose coupling but still want to guarantee bounded memory usage:
-
-> main = do
->     (input, output) <- spawn (Bounded 100)
->     ...
-
-> $ ./work
-> ...
-> 103
-> 104
-> Worker #3: Processed 3
-> Worker #2: Processed 2
-> Worker #1: Processed 1
-> 105
-> 106
-> 107
-> Worker #1: Processed 6
-> Worker #2: Processed 5
-> Worker #3: Processed 4
-> $
--}
-
-{- $broadcast
-    You can also broadcast data to multiple listeners instead of dividing up the
-    data.  Just use the 'Monoid' instance for 'Input' to combine multiple
-    'Input' ends together into a single broadcast 'Input':
-
-> import Control.Monad
-> import Control.Concurrent.Async
-> import Control.Proxy
-> import Control.Proxy.Concurrent
-> import Data.Monoid
-> 
-> main = do
->     (input1, output1) <- spawn Unbounded
->     (input2, output2) <- spawn Unbounded
->     a1 <- async $ do
->         runProxy $ stdinS >-> sendD (input1 <> input2)
->         performGC
->     as <- forM [output1, output2] $ \output -> async $ do
->         runProxy $ recvS output >-> takeB_ 2 >-> stdoutD
->         performGC
->     mapM_ wait (a1:as)
-
-    In the above example, 'stdinS' will broadcast user input to both mailboxes,
-    and each mailbox forwards its values to 'stdoutD', echoing the message to
-    standard output:
-
-> $ ./broadcast
-> ABC<Enter>
-> ABC
-> ABC
-> DEF<Enter>
-> DEF
-> DEF
-> GHI<Enter>
-> $ 
-
-    The combined 'Input' stays alive as long as any of the original 'Input's
-    remains alive.  In the above example, 'sendD' terminates on the third 'send'
-    attempt because it detects that both listeners died after receiving two
-    messages.
-
-    Use 'mconcat' to broadcast to a list of 'Input's, but keep in mind that you
-    will incur a performance price if you combine thousands of 'Input's or more
-    because they will create a very large 'STM' transaction.  You can improve
-    performance for very large broadcasts if you sacrifice atomicity and
-    manually combine multiple 'send' actions in 'IO' instead of 'STM'.
--}
-
-{- $updates
-    Sometimes you don't want to handle every single event.  For example, you
-    might have an input and output device (like a mouse and a monitor) where the
-    input device updates at a different pace than the output device
-
-> import Control.Concurrent
-> import Control.Proxy
-> 
-> -- Fast input updates
-> inputDevice :: (Monad m, Proxy p) => () -> Producer p Integer m r
-> inputDevice = enumFromS 1
-> 
-> -- Slow output updates
-> outputDevice :: (Proxy p) => () -> Consumer p Integer IO r
-> outputDevice () = runIdentityP $ forever $ do
->     n <- request ()
->     lift $ do
->         print n
->         threadDelay 1000000
-
-    In this scenario you don't want to enforce a one-to-one correspondence
-    between input device updates and output device updates because you don't
-    want either end to block waiting for the other end.  Instead, you just need
-    the output device to consult the 'Latest' value received from the 'Input':
-
-> import Control.Concurrent.Async
-> import Control.Proxy.Concurrent
->
-> main = do
->     (input, output) <- spawn (Latest 0)
->     a1 <- async $ do
->         runProxy $ inputDevice >-> sendD input
->         performGC
->     a2 <- async $ do
->         runProxy $ recvS output >-> takeB_ 5 >-> outputDevice
->         performGC
->     mapM_ wait [a1, a2]
-
-    'Latest' selects a mailbox that always stores exactly one value.  The
-    'Latest' constructor takes a single argument (@0@, in the above example)
-    specifying the starting value to store in the mailbox.  'send' overrides the
-    currently stored value and 'recv' peeks at the latest stored value without
-    consuming it.  In the above example the @outputDevice@ periodically peeks at    the latest value stashed inside the mailbox:
-
-> $ ./peek
-> 5
-> 752452
-> 1502636
-> 2248278
-> 2997705
-> $
-
-    A 'Latest' mailbox is never empty because it begins with a default value and
-    'recv' never removes the value from the mailbox.  A 'Latest' mailbox is also
-    never full because 'send' always succeeds, overwriting the previously stored
-    value.
--}
-
-{- $callback
-    @pipes-concurrency@ also solves the common problem of getting data out of a
-    callback-based framework into @pipes@.
-
-    For example, suppose that we have the following callback-based function:
-
-> import Control.Monad
-> 
-> onLines :: (String -> IO a) -> IO b
-> onLines callback = forever $ do
->     str <- getLine
->     callback str
-
-    We can use 'send' to free the data from the callback and then we can
-    retrieve the data on the outside using 'recvS':
-
-> import Control.Proxy
-> import Control.Proxy.Concurrent
-> 
-> onLines' :: (Proxy p) => () -> Producer p String IO ()
-> onLines' () = runIdentityP $ do
->     (input, output) <- lift $ spawn Single
->     lift $ forkIO $ onLines (\str -> atomically $ send input str)
->     recvS output ()
-> 
-> main = runProxy $ onLines' >-> takeWhileD (/= "quit") >-> stdoutD
-
-    Now we can stream from the callback as if it were an ordinary 'Producer':
-
-> $ ./callback
-> Test<Enter>
-> Test
-> Apple<Enter>
-> Apple
-> quit<Enter>
-> $
-
--}
-
-{- $safety
-    @pipes-concurrency@ avoids deadlocks because 'send' and 'recv' always
-    cleanly return before triggering a deadlock.  This behavior works even in
-    complicated scenarios like:
-
-    * cyclic graphs of connected mailboxes,
-
-    * multiple readers and multiple writers to the same mailbox, and
-
-    * dynamically adding or garbage collecting mailboxes.
-
-    The following example shows how @pipes-concurrency@ will do the right thing
-    even in the case of cycles:
-
-> import Control.Concurrent.Async
-> import Control.Proxy
-> import Control.Proxy.Concurrent
-> 
-> main = do
->     (in1, out1) <- spawn Unbounded
->     (in2, out2) <- spawn Unbounded
->     a1 <- async $ do runProxy $ (fromListS [1,2] >=> recvS out1) >-> sendD in2
->                      performGC
->     a2 <- async $ do runProxy $ recvS out2 >-> printD >-> takeB_ 6 >-> sendD in1
->                      performGC
->     mapM_ wait [a1, a2]
-
-    The above program jump-starts a cyclic chain with two input values and
-    terminates one branch of the cycle after six values flow through.  Both
-    branches correctly terminate and get garbage collected without triggering
-    deadlocks when 'takeB_' finishes:
-
-> $ ./cycle
-> 1
-> 2
-> 1
-> 2
-> 1
-> 2
-> $
-
--}
-
-{- $conclusion
-    @pipes-concurrency@ adds an asynchronous dimension to @pipes@.  This
-    promotes a natural division of labor for concurrent programs:
-
-    * Fork one pipeline per deterministic behavior
-
-    * Communicate between concurrent pipelines using @pipes-concurrency@
-
-    This promotes an actor-style approach to concurrent programming where
-    pipelines behave like processes and mailboxes behave like ... mailboxes.
-
-    You can ask questions about @pipes-concurrency@ and other @pipes@ libraries
-    on the official @pipes@ mailing list at
-    <mailto:haskell-pipes@googlegroups.com>.
--}
-
-{- $appendix
-    I've provided the full code for the above examples here so you can easily
-    try them out:
-
-> -- game.hs
->
-> import Control.Concurrent
-> import Control.Monad
-> import Control.Proxy
-> import Control.Proxy.Concurrent
-> import Control.Proxy.Trans.Maybe
-> import Control.Proxy.Trans.State
-> 
-> -- The game events
-> data Event = Harm Integer | Heal Integer | Quit
-> 
-> -- The game state
-> type Health = Integer
-> 
-> handler :: (Proxy p) => () -> Consumer (StateP Health (MaybeP p)) Event IO r
-> handler () = forever $ do
->     event <- request ()
->     case event of
->         Harm n -> modify (subtract n)
->         Heal n -> modify (+        n)
->         Quit   -> mzero
->     health <- get
->     lift $ putStrLn $ "Health = " ++ show health
->
-> user :: (Proxy p) => () -> Producer p Event IO r
-> user () = runIdentityP $ forever $ do
->     command <- lift getLine
->     case command of
->         "potion" -> respond (Heal 10)
->         "quit"   -> respond  Quit
->         _        -> lift $ putStrLn "Invalid command"
->
-> acidRain :: (Proxy p) => () -> Producer p Event IO r
-> acidRain () = runIdentityP $ forever $ do
->     respond (Harm 1)
->     lift $ threadDelay 2000000
->
-> main = do
->     (input, output) <- spawn Unbounded
->     forkIO $ do runProxy $ acidRain >-> sendD input
->                 performGC  -- I'll explain 'performGC' below
->     forkIO $ do runProxy $ user     >-> sendD input
->                 performGC
->     runProxy $ runMaybeK $ evalStateK 100 $ recvS output >-> handler
-
-> -- work.hs
-> 
-> import Control.Concurrent
-> import Control.Monad
-> import Control.Proxy
-> import Control.Concurrent.Async
-> import Control.Proxy.Concurrent
-> 
-> worker :: (Proxy p, Show a) => Int -> () -> Consumer p a IO r
-> worker i () = runIdentityP $ forever $ do
->     a <- request ()
->     lift $ threadDelay 1000000  -- 1 second
->     lift $ putStrLn $ "Worker #" ++ show i ++ ": Processed " ++ show a
-> {-
-> worker :: (Proxy p, Show a) => Int -> () -> Consumer p a IO ()
-> worker i () = runIdentityP $ replicateM_ 2 $ do
->     a <- request ()
->     lift $ threadDelay 1000000
->     lift $ putStrLn $ "Worker #" ++ show i ++ ": Processed " ++ show a
-> -}
->
-> user :: (Proxy p) => () -> Producer p String IO ()
-> user = stdinS >-> takeWhileD (/= "quit")
-> 
-> main = do
->     (input, output) <- spawn Unbounded
-> --  (input, output) <- spawn Single
-> --  (input, output) <- spawn (Bounded 100)
->     as <- forM [1..3] $ \i ->
->           async $ do runProxy $ recvS output >-> worker i
-> --        async $ do runProxy $ recvS output >-> takeB_ 2 >-> worker i
->                      performGC
->     a  <- async $ do runProxy $ fromListS [1..10]      >-> sendD input
-> --  a  <- async $ do runProxy $ user                   >-> sendD input
-> --  a  <- async $ do runProxy $ enumFromS 1 >-> printD >-> sendD input
->                      performGC
->     mapM_ wait (a:as)
-
-> -- broadcast.hs
->
-> import Control.Monad
-> import Control.Concurrent.Async
-> import Control.Proxy
-> import Control.Proxy.Concurrent
-> import Data.Monoid
-> 
-> main = do
->     (input1, output1) <- spawn Unbounded
->     (input2, output2) <- spawn Unbounded
->     a1 <- async $ do
->         runProxy $ stdinS >-> sendD (input1 <> input2)
->         performGC
->     as <- forM [output1, output2] $ \output -> async $ do
->         runProxy $ recvS output >-> takeB_ 2 >-> stdoutD
->         performGC
->     mapM_ wait (a1:as)
-
-> -- peek.hs
-> 
-> import Control.Concurrent
-> import Control.Concurrent.Async
-> import Control.Proxy
-> import Control.Proxy.Concurrent
-> 
-> inputDevice :: (Monad m, Proxy p) => () -> Producer p Integer m r
-> inputDevice = enumFromS 1
-> 
-> outputDevice :: (Proxy p) => () -> Consumer p Integer IO r
-> outputDevice () = runIdentityP $ forever $ do
->     n <- request ()
->     lift $ do
->         print n
->         threadDelay 1000000
->
-> main = do
->     (input, output) <- spawn (Latest 0)
->     a1 <- async $ do
->         runProxy $ inputDevice >-> sendD input
->         performGC
->     a2 <- async $ do
->         runProxy $ recvS output >-> takeB_ 5 >-> outputDevice
->         performGC
->     mapM_ wait [a1, a2]
-
-> -- callback.hs
-> 
-> import Control.Proxy
-> import Control.Proxy.Concurrent
-> 
-> onLines' :: (Proxy p) => () -> Producer p String IO ()
-> onLines' () = runIdentityP $ do
->     (input, output) <- lift $ spawn Single
->     lift $ forkIO $ onLines (\str -> atomically $ send input str)
->     recvS output ()
-> 
-> main = runProxy $ onLines' >-> takeWhileD (/= "quit) >-> stdoutD
--}

@@ -1,24 +1,24 @@-Copyright (c) 2013 Gabriel Gonzalez
-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 Gabriel Gonzalez nor the names of other 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.
+Copyright (c) 2013 Gabriel Gonzalez+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 Gabriel Gonzalez nor the names of other 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.

@@ -1,2 +1,2 @@-import Distribution.Simple
-main = defaultMain
+import Distribution.Simple+main = defaultMain

@@ -1,39 +1,51 @@-Name: pipes-concurrency
-Version: 1.2.1
-Cabal-Version: >=1.8.0.2
-Build-Type: Simple
-License: BSD3
-License-File: LICENSE
-Copyright: 2013 Gabriel Gonzalez
-Author: Gabriel Gonzalez
-Maintainer: Gabriel439@gmail.com
-Bug-Reports: https://github.com/Gabriel439/Haskell-Pipes-Concurrency-Library/issues
-Synopsis: Concurrency for the pipes ecosystem
-Description: This library provides light-weight concurrency primitives for
-  pipes, with the following features:
-  .
-  * /Simple API/: Use only five functions
-  .
-  * /Deadlock Safety/: Automatically avoid concurrency deadlocks
-  .
-  * /Flexibility/: Build many-to-many and cyclic communication topologies
-  .
-  * /Dynamic Graphs/: Add or remove readers and writers at any time
-  .
-  Import "Control.Proxy.Concurrent" to use the library.
-  .
-  Read "Control.Proxy.Concurrent.Tutorial" for an tutorial.
-Category: Control, Pipes, Proxies, Concurrency
-Source-Repository head
-    Type: git
-    Location: https://github.com/Gabriel439/Haskell-Pipes-Concurrency-Library
-
-Library
-    Build-Depends:
-        base         >= 4       && < 5  ,
-        pipes        >= 3.0     && < 3.4,
-        stm          >= 2.4     && < 2.5
-    Exposed-Modules:
-        Control.Proxy.Concurrent,
-        Control.Proxy.Concurrent.Tutorial
-    GHC-Options: -O2
+Name: pipes-concurrency+Version: 2.0.0+Cabal-Version: >=1.8.0.2+Build-Type: Simple+License: BSD3+License-File: LICENSE+Copyright: 2013 Gabriel Gonzalez+Author: Gabriel Gonzalez+Maintainer: Gabriel439@gmail.com+Bug-Reports: https://github.com/Gabriel439/Haskell-Pipes-Concurrency-Library/issues+Synopsis: Concurrency for the pipes ecosystem+Description: This library provides light-weight concurrency primitives for+  pipes, with the following features:+  .+  * /Simple API/: Use only five functions+  .+  * /Deadlock Safety/: Automatically avoid concurrency deadlocks+  .+  * /Flexibility/: Build many-to-many and cyclic communication topologies+  .+  * /Dynamic Graphs/: Add or remove readers and writers at any time+  .+  Import "Pipes.Concurrent" to use the library.+  .+  Read "Pipes.Concurrent.Tutorial" for a tutorial.+Category: Control, Pipes, Concurrency+Source-Repository head+    Type: git+    Location: https://github.com/Gabriel439/Haskell-Pipes-Concurrency-Library++Library+    Hs-Source-Dirs: src+    Build-Depends:+        base         >= 4   && < 5  ,+        pipes        >= 4.0 && < 4.1,+        stm          >= 2.4 && < 2.5+    Exposed-Modules:+        Pipes.Concurrent,+        Pipes.Concurrent.Tutorial+    GHC-Options: -O2++Test-Suite tests+    Type: exitcode-stdio-1.0+    Main-Is: tests-main.hs+    HS-Source-Dirs: tests .+    Build-Depends:+        base              >= 4     && < 5  ,+        pipes             >= 4.0.0 && < 4.1,+        pipes-concurrency >= 2.0.0 && < 4.1,+        stm               >= 2.4   && < 2.5,+        async             >= 2.0   && < 2.1

@@ -0,0 +1,234 @@+-- | Asynchronous communication between pipes++{-# LANGUAGE CPP, RankNTypes#-}++#if __GLASGOW_HASKELL__ >= 702+{-# LANGUAGE Trustworthy #-}+#endif+{- '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.+-}++module Pipes.Concurrent (+    -- * Inputs and Outputs+    Input(..),+    Output(..),++    -- * Pipe utilities+    fromInput,+    toOutput,++    -- * Actors+    spawn,+    spawn',+    Buffer(..),++    -- * Re-exports+    -- $reexport+    module Control.Concurrent,+    module Control.Concurrent.STM,+    module System.Mem+    ) where++import Control.Applicative (+    Alternative(empty, (<|>)), Applicative(pure, (<*>)), (<*), (<$>) )+import Control.Concurrent (forkIO)+import Control.Concurrent.STM (atomically, STM)+import qualified Control.Concurrent.STM as S+import Control.Monad (when)+import Data.IORef (newIORef, readIORef, mkWeakIORef)+import Data.Monoid (Monoid(mempty, mappend))+import GHC.Conc.Sync (unsafeIOToSTM)+import Pipes (MonadIO(liftIO), yield, await, Producer', Consumer')+import System.Mem (performGC)++{-| An exhaustible source of values++    'recv' returns 'Nothing' if the source is exhausted+-}+newtype Input a = Input {+    recv :: S.STM (Maybe a) }++instance Functor Input where+    fmap f m = Input (fmap (fmap f) (recv m))++instance Applicative Input where+    pure r    = Input (pure (pure r))+    mf <*> mx = Input ((<*>) <$> recv mf <*> recv mx)++instance Monad Input where+    return r = Input (return (return r))+    m >>= f  = Input $ do+        ma <- recv m+        case ma of+            Nothing -> return Nothing+            Just a  -> recv (f a)++-- Deriving 'Alternative'+instance Alternative Input where+    empty   = Input empty+    x <|> y = Input (recv x <|> recv y)++instance Monoid (Input a) where+    mempty = empty+    mappend = (<|>)++{-| An exhaustible sink of values++    'send' returns 'False' if the sink is exhausted+-}+newtype Output a = Output {+    send :: a -> S.STM Bool }++instance Monoid (Output a) where+    mempty  = Output (\_ -> return False)+    mappend i1 i2 = Output (\a -> (||) <$> send i1 a <*> send i2 a)++{-| Convert an 'Output' to a 'Pipes.Consumer'++    'toOutput' terminates when the 'Output' is exhausted.+-}+toOutput :: (MonadIO m) => Output a -> Consumer' a m ()+toOutput output = loop+  where+    loop = do+        a     <- await+        alive <- liftIO $ S.atomically $ send output a+        when alive loop+{-# INLINABLE toOutput #-}++{-| Convert an 'Input' to a 'Pipes.Producer'++    'fromInput' terminates when the 'Input' is exhausted.+-}+fromInput :: (MonadIO m) => Input a -> Producer' a m ()+fromInput input = loop+  where+    loop = do+        ma <- liftIO $ S.atomically $ recv input+        case ma of+            Nothing -> return ()+            Just a  -> do+                yield a+                loop+{-# INLINABLE fromInput #-}++{-| Spawn a mailbox using the specified 'Buffer' to store messages++    Using 'send' on the 'Output'++        * fails and returns 'False' if the mailbox is sealed, otherwise it:++        * retries if the mailbox is full, or:++        * adds a message to the mailbox and returns 'True'.++    Using 'recv' on the 'Input':++        * retrieves a message from the mailbox wrapped in 'Just' if the mailbox+          is not empty, otherwise it:++        * retries if the mailbox is not sealed, or:++        * fails and returns 'Nothing'.++    If either the 'Input' or 'Output' is garbage collected the mailbox will+    become sealed.+-}+spawn :: Buffer a -> IO (Output a, Input a)+spawn buffer = fmap simplify (spawn' buffer)+  where+    simplify (output, input, _) = (output, input)+{-# INLINABLE spawn #-}++{-| Like 'spawn', but also returns an action to manually @seal@ the mailbox+    early:++> (output, input, seal) <- spawn' buffer+> ...++    Use the @seal@ action to allow early cleanup of readers and writers to the+    mailbox without waiting for the next garbage collection cycle.+-}+spawn' :: Buffer a -> IO (Output a, Input a, STM ())+spawn' buffer = do+    (read, write) <- case buffer of+        Bounded n -> do+            q <- S.newTBQueueIO n+            return (S.readTBQueue q, S.writeTBQueue q)+        Unbounded -> do+            q <- S.newTQueueIO+            return (S.readTQueue q, S.writeTQueue q)+        Single    -> do+            m <- S.newEmptyTMVarIO+            return (S.takeTMVar m, S.putTMVar m)+        Latest a  -> do+            t <- S.newTVarIO a+            return (S.readTVar t, S.writeTVar t)++    sealed <- S.newTVarIO False+    let seal = S.writeTVar sealed True++    {- Use IORefs 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)++    let sendOrEnd a = do+            b <- S.readTVar sealed+            if b+                then return False+                else do+                    write a+                    return True+        readOrEnd = (Just <$> read) <|> (do+            b <- S.readTVar sealed+            S.check b+            return Nothing )+        _send a = sendOrEnd a <* unsafeIOToSTM (readIORef rSend)+        _recv   = readOrEnd   <* unsafeIOToSTM (readIORef 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++{- $reexport+    @Control.Concurrent@ re-exports 'forkIO', although I recommend using the+    @async@ library instead.++    @Control.Concurrent.STM@ re-exports 'atomically' and 'STM'.++    @System.Mem@ re-exports 'performGC'.+-}

@@ -0,0 +1,812 @@+{-| This module provides a tutorial for the @pipes-concurrency@ library.++    This tutorial assumes that you have read the @pipes@ tutorial in+    @Pipes.Tutorial@.++    I've condensed all the code examples into self-contained code listings in+    the Appendix section that you can use to follow along.+-}++module Pipes.Concurrent.Tutorial (+    -- * Introduction+    -- $intro++    -- * Work Stealing+    -- $steal++    -- * Termination+    -- $termination++    -- * Mailbox Sizes+    -- $mailbox++    -- * Broadcasts+    -- $broadcast++    -- * Updates+    -- $updates++    -- * Callbacks+    -- $callback++    -- * Safety+    -- $safety++    -- * Conclusion+    -- $conclusion++    -- * Appendix+    -- $appendix+    ) where++import Control.Concurrent+import Control.Monad+import Pipes+import Pipes.Concurrent+import qualified Pipes.Prelude as P+import Data.Monoid++{- $intro+    The @pipes-concurrency@ library provides a simple interface for+    communicating between concurrent pipelines.  Use this library if you want+    to:++    * merge multiple streams into a single stream,++    * stream data from a callback \/ continuation,++    * broadcast data,++    * build a work-stealing setup, or++    * implement basic functional reactive programming (FRP).++    For example, let's say that we want to design a simple game with two+    concurrent sources of game @Event@s.++    One source translates user input to game events:++> -- The game events+> data Event = Harm Integer | Heal Integer | Quit deriving (Show)+>+> user :: IO Event+> user = do+>     command <- getLine+>     case command of+>         "potion" -> return (Heal 10)+>         "quit"   -> return  Quit+>         _        -> do+>             putStrLn "Invalid command"+>             user  -- Try again++    ... while the other creates inclement weather:++> import Control.Concurrent (threadDelay)+> import Control.Monad (forever)+> import Pipes+>+> acidRain :: Producer Event IO r+> acidRain = forever $ do+>     lift $ threadDelay 2000000  -- Wait 2 seconds+>     yield (Harm 1)++    We can asynchronously merge these two separate sources of @Event@s into a+    single stream by 'spawn'ing a first-in-first-out (FIFO) mailbox:++@+ 'spawn' :: 'Buffer' a -> 'IO' ('Output' a, 'Input' a)+@++    'spawn' takes a 'Buffer' as an argument which specifies how many messages to+    store.  In this case we want our mailbox to store an 'Unbounded' number of+    messages:++> import Pipes.Concurrent+> +> main = do+>     (output, input) <- spawn Unbounded+>     ...++   'spawn' creates this mailbox in the background and then returns two values:++    * an @(Output a)@ that we use to add messages of type @a@ to the mailbox++    * an @(Input a)@ that we use to consume messages of type @a@ from the+      mailbox++    We will be streaming @Event@s through our mailbox, so our @output@ has type+    @(Output Event)@ and our @input@ has type @(Input Event)@.++    To stream @Event@s into the mailbox , we use 'toOutput', which writes values+    to the mailbox's 'Output' end:++@+ 'toOutput' :: ('MonadIO' m) => 'Output' a -> 'Consumer' a m ()+@++    We can concurrently forward multiple streams to the same 'Output', which+    asynchronously merges their messages into the same mailbox:++>     ...+>     forkIO $ do runEffect $ lift user >~  toOutput output+>                 performGC  -- I'll explain 'performGC' below+> +>     forkIO $ do runEffect $ acidRain  >-> toOutput output+>                 performGC+>     ...++    To stream @Event@s out of the mailbox, we use 'fromInput', which streams+    values from the mailbox's 'Input' end using a 'Producer':++@+ 'fromInput' :: ('MonadIO' m) => 'Input' a -> 'Producer' a m ()+@++    For this example we'll build a 'Consumer' to handle this stream of @Event@s,+    that either harms or heals our intrepid adventurer depending on which+    @Event@ we receive:++> handler :: Consumer Event IO ()+> handler = loop 100+>   where+>     loop health = do+>         lift $ putStrLn $ "Health = " ++ show health+>         event <- await+>         case event of+>             Harm n -> loop (health - n)+>             Heal n -> loop (health + n)+>             Quit   -> return ()++    Now we can just connect our @Event@ 'Producer' to our @Event@ 'Consumer'+    using ('>->'):++>     ...+>     runEffect $ fromInput input >-> handler++    Our final @main@ looks like this:++> main = do+>     (output, input) <- spawn Unbounded+>+>     forkIO $ do runEffect $ lift user >~  toOutput output+>                 performGC  +>+>     forkIO $ do runEffect $ acidRain  >-> toOutput output+>                 performGC+>+>     runEffect $ fromInput input >-> handler++    ... and when we run it we get the desired concurrent behavior:++> $ ./game+> Health = 100+> Health = 99+> Health = 98+> potion<Enter>+> Health = 108+> Health = 107+> Health = 106+> potion<Enter>+> Health = 116+> Health = 115+> quit<Enter>+> $+-}++{- $steal+    You can also have multiple pipes reading from the same mailbox.  Messages+    get split between listening pipes on a first-come first-serve basis.++    For example, we'll define a \"worker\" that takes a one-second break each+    time it receives a new job:++> import Control.Concurrent (threadDelay)+> import Control.Monad+> import Pipes+> +> worker :: (Show a) => Int -> Consumer a IO r+> worker i = forever $ do+>     a <- await+>     lift $ threadDelay 1000000  -- 1 second+>     lift $ putStrLn $ "Worker #" ++ show i ++ ": Processed " ++ show a++    Fortunately, these workers are cheap, so we can assign several of them to+    the same job:++> import Control.Concurrent.Async+> import qualified Pipes.Prelude as P+> import Pipes.Concurrent+> +> main = do+>     (output, input) <- spawn Unbounded+>     as <- forM [1..3] $ \i ->+>           async $ do runEffect $ fromInput input  >-> worker i+>                      performGC+>     a  <- async $ do runEffect $ each [1..10] >-> toOutput output+>                      performGC+>     mapM_ wait (a:as)++    The above example uses @Control.Concurrent.Async@ from the @async@ package+    to fork each thread and wait for all of them to terminate:++> $ ./work+> Worker #2: Processed 3+> Worker #1: Processed 2+> Worker #3: Processed 1+> Worker #3: Processed 6+> Worker #1: Processed 5+> Worker #2: Processed 4+> Worker #2: Processed 9+> Worker #1: Processed 8+> Worker #3: Processed 7+> Worker #2: Processed 10+> $++    What if we replace 'each' with a different source that reads lines from user+    input until the user types \"quit\":++> user :: Producer String IO ()+> user = P.stdinLn >-> P.takeWhile (/= "quit")+> +> main = do+>     (output, input) <- spawn Unbounded+>     as <- forM [1..3] $ \i ->+>           async $ do runEffect $ fromInput input >-> worker i+>                      performGC+>     a  <- async $ do runEffect $ user >-> toOutput output+>                      performGC+>     mapM_ wait (a:as)++    This still produces the correct behavior:++> $ ./work+> Test<Enter>+> Worker #1: Processed "Test"+> Apple<Enter>+> Worker #2: Processed "Apple"+> 42<Enter>+> Worker #3: Processed "42"+> A<Enter>+> B<Enter>+> C<Enter>+> Worker #1: Processed "A"+> Worker #2: Processed "B"+> Worker #3: Processed "C"+> quit<Enter>+> $+-}++{- $termination++    Wait...  How do the workers know when to stop listening for data?  After+    all, anything that has a reference to 'Output' could potentially add more+    data to the mailbox.++    It turns out that 'spawn' is smart and instruments the 'Input' to+    terminate when the 'Output' is garbage collected.  'fromInput' builds on top+    of the more primitive 'recv' command, which returns a 'Nothing' when the+    'Input' terminates:++@+ 'recv' :: 'Input' a -> 'STM' ('Maybe' a)+@++    Otherwise, 'recv' will block if the mailbox is empty since if the 'Output'+    has not been garbage collected then somebody might still produce more data.++    Does it work the other way around?  What happens if the workers go on strike+    before processing the entire data set?++>     ...+>     as <- forM [1..3] $ \i ->+>           -- Each worker refuses to process more than two values+>           async $ do runEffect $ fromInput input >-> P.take 2 >-> worker i+>                      performGC+>     ...++    Let's find out:++> $ ./work+> How<Enter>+> Worker #1: Processed "How"+> many<Enter>+> roads<Enter>+> Worker #2: Processed "many"+> Worker #3: Processed "roads"+> must<Enter>+> a<Enter>+> man<Enter>+> Worker #1: Processed "must"+> Worker #2: Processed "a"+> Worker #3: Processed "man"+> walk<Enter>+> $++    'spawn' tells the 'Output' to similarly terminate when the 'Input' is+    garbage collected, preventing the user from submitting new values.+    'toOutput' builds on top of the more primitive 'send' command, which returns+    a 'False' when the 'Output' terminates:++@+ 'send' :: 'Output' a -> a -> 'STM' 'Bool'+@++    Otherwise, 'send' will blocks if the mailbox is full, since if the 'Input'+    has not been garbage collected then somebody could still consume a value+    from the mailbox, making room for a new value.++    This is why we have to insert 'performGC' calls whenever we release a+    reference to either the 'Output' or 'Input'.  Without these calls we cannot+    guarantee that the garbage collector will trigger and notify the opposing+    end if the last reference was released.++    There are two ways to avoid using 'performGC'.  First, you can omit the+    'performGC' call, which is safe and preferable for long-running programs.+    This simply delays garbage collecting mailboxes until the next garbage+    collection cycle.++    Second, you can use the 'spawn'' command, which returns a third @seal@+    action:++> (output, input, seal) <- spawn' buffer+> ...++    Use this to @seal@ the mailbox so that it cannot receive new messages.  This+    allows both readers and writers to shut down early without relying on+    garbage collection:++    * writers will shut down immediately because they can no longer write to the+      mailbox++    * readers will shut down when the mailbox goes empty because they know that+      no new data will arrive++    For simplicity, this tutorial will continue to use `performGC` since all+    the examples are short-lived programs that do not build up a large heap.+    However, when the heap grows large you want to avoid `performGC` and+    consider using one of the above two alternatives instead.++    Note only 'Input's and 'Output's specifically built using 'spawn' or+    'spawn'' make use of the garbage collector.  If you build your own custom+    'Input's and 'Output's then you do not need to use 'performGC' at all.+-}++{- $mailbox+    So far we haven't observed 'send' blocking because we only 'spawn'ed+    'Unbounded' mailboxes.  However, we can control the size of the mailbox to+    tune the coupling between the 'Output' and the 'Input' ends.++    If we set the mailbox 'Buffer' to 'Single', then the mailbox holds exactly+    one message, forcing synchronization between 'send's and 'recv's.  Let's+    observe this by sending an infinite stream of values, logging all values to+    the console:++> main = do+>     (output, input) <- spawn Single+>     as <- forM [1..3] $ \i ->+>           async $ do runEffect $ fromInput input >-> P.take 2 >-> worker i+>                      performGC+>     a  <- async $ do runEffect $ each [1..] >-> P.chain print >-> toOutput output+>                      performGC+>     mapM_ wait (a:as)++    The 7th value gets stuck in the mailbox, and the 8th value blocks because+    the mailbox never clears the 7th value:++> $ ./work+> 1+> 2+> 3+> 4+> 5+> Worker #3: Processed 3+> Worker #2: Processed 2+> Worker #1: Processed 1+> 6+> 7+> 8+> Worker #1: Processed 6+> Worker #2: Processed 5+> Worker #3: Processed 4+> $++    Contrast this with an 'Unbounded' mailbox for the same program, which keeps+    accepting values until downstream finishes processing the first six values:++> $ ./work+> 1+> 2+> 3+> 4+> 5+> 6+> 7+> 8+> 9+> ...+> 487887+> 487888+> Worker #3: Processed 3+> Worker #2: Processed 2+> Worker #1: Processed 1+> 487889+> 487890+> ...+> 969188+> 969189+> Worker #1: Processed 6+> Worker #2: Processed 5+> Worker #3: Processed 4+> 969190+> 969191+> $++    You can also choose something in between by using a 'Bounded' mailbox which+    caps the mailbox size to a fixed value.  Use 'Bounded' when you want mostly+    loose coupling but still want to guarantee bounded memory usage:++> main = do+>     (output, input) <- spawn (Bounded 100)+>     ...++> $ ./work+> ...+> 103+> 104+> Worker #3: Processed 3+> Worker #2: Processed 2+> Worker #1: Processed 1+> 105+> 106+> 107+> Worker #1: Processed 6+> Worker #2: Processed 5+> Worker #3: Processed 4+> $+-}++{- $broadcast+    You can also broadcast data to multiple listeners instead of dividing up the+    data.  Just use the 'Monoid' instance for 'Output' to combine multiple+    'Output' ends together into a single broadcast 'Output':++> -- broadcast.hs+>+> import Control.Monad+> import Control.Concurrent.Async+> import Pipes+> import Pipes.Concurrent+> import qualified Pipes.Prelude as P+> import Data.Monoid+> +> main = do+>     (output1, input1) <- spawn Unbounded+>     (output2, input2) <- spawn Unbounded+>     a1 <- async $ do+>         runEffect $ P.stdinLn >-> toOutput (output1 <> output2)+>         performGC+>     as <- forM [input1, input2] $ \input -> async $ do+>         runEffect $ fromInput input >-> P.take 2 >-> P.stdoutLn+>         performGC+>     mapM_ wait (a1:as)++    In the above example, 'P.stdinLn' will broadcast user input to both+    mailboxes, and each mailbox forwards its values to 'P.stdoutLn', echoing the+    message to standard output:++> $ ./broadcast+> ABC<Enter>+> ABC+> ABC+> DEF<Enter>+> DEF+> DEF+> GHI<Enter>+> $ ++    The combined 'Output' stays alive as long as any of the original 'Output's+    remains alive.  In the above example, 'toOutput' terminates on the third+    'send' attempt because it detects that both listeners died after receiving+    two messages.++    Use 'mconcat' to broadcast to a list of 'Output's, but keep in mind that you+    will incur a performance price if you combine thousands of 'Output's or more+    because they will create a very large 'STM' transaction.  You can improve+    performance for very large broadcasts if you sacrifice atomicity and+    manually combine multiple 'send' actions in 'IO' instead of 'STM'.+-}++{- $updates+    Sometimes you don't want to handle every single event.  For example, you+    might have an input and output device (like a mouse and a monitor) where the+    input device updates at a different pace than the output device++> import Control.Concurrent (threadDelay)+> import Control.Monad+> import Pipes+> import qualified Pipes.Prelude as P+> +> -- Fast input updates+> inputDevice :: (Monad m) => Producer Integer m ()+> inputDevice = each [1..]+> +> -- Slow output updates+> outputDevice :: Consumer Integer IO r+> outputDevice = forever $ do+>     n <- await+>     lift $ do+>         print n+>         threadDelay 1000000++    In this scenario you don't want to enforce a one-to-one correspondence+    between input device updates and output device updates because you don't+    want either end to block waiting for the other end.  Instead, you just need+    the output device to consult the 'Latest' value received from the 'Input':++> import Control.Concurrent.Async+> import Pipes.Concurrent+> +> main = do+>     (output, input) <- spawn (Latest 0)+>     a1 <- async $ do runEffect $ inputDevice >-> toOutput output+>                      performGC+>     a2 <- async $ do runEffect $ fromInput input >-> P.take 5 >-> outputDevice+>                      performGC+>     mapM_ wait [a1, a2]++    'Latest' selects a mailbox that always stores exactly one value.  The+    'Latest' constructor takes a single argument (@0@, in the above example)+    specifying the starting value to store in the mailbox.  'send' overrides the+    currently stored value and 'recv' peeks at the latest stored value without+    consuming it.  In the above example the @outputDevice@ periodically peeks at    the latest value stashed inside the mailbox:++> $ ./peek+> 7+> 2626943+> 5303844+> 7983519+> 10604940+> $++    A 'Latest' mailbox is never empty because it begins with a default value and+    'recv' never removes the value from the mailbox.  A 'Latest' mailbox is also+    never full because 'send' always succeeds, overwriting the previously stored+    value.+-}++{- $callback+    @pipes-concurrency@ also solves the common problem of getting data out of a+    callback-based framework into @pipes@.++    For example, suppose that we have the following callback-based function:++> import Control.Monad+> +> onLines :: (String -> IO a) -> IO b+> onLines callback = forever $ do+>     str <- getLine+>     callback str++    We can use 'send' to free the data from the callback and then we can+    retrieve the data on the outside using 'fromInput':++> import Pipes+> import Pipes.Concurrent+> import qualified Pipes.Prelude as P+> +> onLines' :: Producer String IO ()+> onLines' = do+>     (output, input) <- lift $ spawn Single+>     lift $ forkIO $ onLines (\str -> atomically $ send output str)+>     fromInput input+> +> main = runEffect $ onLines' >-> P.takeWhile (/= "quit") >-> P.stdoutLn++    Now we can stream from the callback as if it were an ordinary 'Producer':++> $ ./callback+> Test<Enter>+> Test+> Apple<Enter>+> Apple+> quit<Enter>+> $++-}++{- $safety+    @pipes-concurrency@ avoids deadlocks because 'send' and 'recv' always+    cleanly return before triggering a deadlock.  This behavior works even in+    complicated scenarios like:++    * cyclic graphs of connected mailboxes,++    * multiple readers and multiple writers to the same mailbox, and++    * dynamically adding or garbage collecting mailboxes.++    The following example shows how @pipes-concurrency@ will do the right thing+    even in the case of cycles:++> -- cycle.hs+>+> import Control.Concurrent.Async+> import Pipes+> import Pipes.Concurrent+> import qualified Pipes.Prelude as P+> +> main = do+>     (out1, in1) <- spawn Unbounded+>     (out2, in2) <- spawn Unbounded+>     a1 <- async $ do+>         runEffect $ (each [1,2] >> fromInput in1) >-> toOutput out2+>         performGC+>     a2 <- async $ do+>         runEffect $ fromInput in2 >-> P.chain print >-> P.take 6 >-> toOutput out1+>         performGC+>     mapM_ wait [a1, a2]++    The above program jump-starts a cyclic chain with two input values and+    terminates one branch of the cycle after six values flow through.  Both+    branches correctly terminate and get garbage collected without triggering+    deadlocks when 'takeB_' finishes:++> $ ./cycle+> 1+> 2+> 1+> 2+> 1+> 2+> $++-}++{- $conclusion+    @pipes-concurrency@ adds an asynchronous dimension to @pipes@.  This+    promotes a natural division of labor for concurrent programs:++    * Fork one pipeline per deterministic behavior++    * Communicate between concurrent pipelines using @pipes-concurrency@++    This promotes an actor-style approach to concurrent programming where+    pipelines behave like processes and mailboxes behave like ... mailboxes.++    You can ask questions about @pipes-concurrency@ and other @pipes@ libraries+    on the official @pipes@ mailing list at+    <mailto:haskell-pipes@googlegroups.com>.+-}++{- $appendix+    I've provided the full code for the above examples here so you can easily+    try them out:++>-- game.hs+>+>import Control.Concurrent (threadDelay)+>import Control.Monad (forever)+>import Pipes+>import Pipes.Concurrent+>+>data Event = Harm Integer | Heal Integer | Quit deriving (Show)+>+>user :: IO Event+>user = do+>    command <- getLine+>    case command of+>        "potion" -> return (Heal 10)+>        "quit"   -> return  Quit+>        _        -> do+>            putStrLn "Invalid command"+>            user+>+>acidRain :: Producer Event IO r+>acidRain = forever $ do+>    lift $ threadDelay 2000000  -- Wait 2 seconds+>    yield (Harm 1)+>+>handler :: Consumer Event IO ()+>handler = loop 100+>  where+>    loop health = do+>        lift $ putStrLn $ "Health = " ++ show health+>        event <- await+>        case event of+>            Harm n -> loop (health - n)+>            Heal n -> loop (health + n)+>            Quit   -> return ()+>+>main = do+>    (output, input) <- spawn Unbounded+>+>    forkIO $ do runEffect $ lift user >~  toOutput output+>                performGC+>+>    forkIO $ do runEffect $ acidRain  >-> toOutput output+>                performGC+>+>    runEffect $ fromInput input >-> handler++>-- work.hs+>+>import Control.Concurrent (threadDelay)+>import Control.Concurrent.Async+>import Control.Monad+>import Pipes+>import Pipes.Concurrent+>import qualified Pipes.Prelude as P+>+>worker :: (Show a) => Int -> Consumer a IO r+>worker i = forever $ do+>    a <- await+>    lift $ threadDelay 1000000  -- 1 second+>    lift $ putStrLn $ "Worker #" ++ show i ++ ": Processed " ++ show a+>+>user :: Producer String IO ()+>user = P.stdinLn >-> P.takeWhile (/= "quit")+>+>main = do+>--  (output, input) <- spawn Unbounded+>--  (output, input) <- spawn Single+>    (output, input) <- spawn (Bounded 100)+>+>    as <- forM [1..3] $ \i ->+>--        async $ do runEffect $ fromInput input  >-> worker i+>          async $ do runEffect $ fromInput input  >-> P.take 2 >-> worker i+>                     performGC+>+>--  a  <- async $ do runEffect $ each [1..10]                 >-> toOutput output+>--  a  <- async $ do runEffect $ user                         >-> toOutput output+>    a  <- async $ do runEffect $ each [1..] >-> P.chain print >-> toOutput output+>                     performGC+>+>    mapM_ wait (a:as)++>-- peek.hs+>+>import Control.Concurrent (threadDelay)+>import Control.Concurrent.Async+>import Control.Monad+>import Pipes+>import Pipes.Concurrent+>import qualified Pipes.Prelude as P+>+>inputDevice :: (Monad m) => Producer Integer m ()+>inputDevice = each [1..]+>+>outputDevice :: Consumer Integer IO r+>outputDevice = forever $ do+>    n <- await+>    lift $ do+>        print n+>        threadDelay 1000000+>+>main = do+>    (output, input) <- spawn (Latest 0)+>    a1 <- async $ do runEffect $ inputDevice >-> toOutput output+>                     performGC+>    a2 <- async $ do runEffect $ fromInput input >-> P.take 5 >-> outputDevice+>                     performGC+>    mapM_ wait [a1, a2]++>-- callback.hs+>+>import Control.Monad+>import Pipes+>import Pipes.Concurrent+>import qualified Pipes.Prelude as P+>+>onLines :: (String -> IO a) -> IO b+>onLines callback = forever $ do+>    str <- getLine+>    callback str+>+>onLines' :: Producer String IO ()+>onLines' = do+>    (output, input) <- lift $ spawn Single+>    lift $ forkIO $ onLines (\str -> atomically $ send output str)+>    fromInput input+>+>main = runEffect $ onLines' >-> P.takeWhile (/= "quit") >-> P.stdoutLn+-}

@@ -0,0 +1,122 @@+module Main ( main ) where++import Control.Concurrent hiding (yield)+import Control.Concurrent.Async+import Control.Monad (forever)+import Pipes+import Pipes.Concurrent+import qualified Pipes.Prelude as P+import System.Exit+import System.IO+import System.Timeout++defaultTimeout :: Int+defaultTimeout = 100000         -- 0.1 s++labelPrint :: (Show a) => String -> Consumer a IO r+labelPrint label = forever $ do+  a <- await+  lift $ putStrLn $ label ++ ": " ++ show a++testSenderClose :: Buffer Int -> IO ()+testSenderClose buffer = do+    (output, input) <- spawn buffer+    t1 <- async $ do+        runEffect $ each [1..5] >-> toOutput output+        performGC+    t2 <- async $ do+        runEffect $   fromInput input+                  >-> P.chain (\_ -> threadDelay 1000)+                  >-> P.print+        performGC+    wait t1+    wait t2++testSenderCloseDelayedSend :: Buffer Int -> IO ()+testSenderCloseDelayedSend buffer = do+    (output, input) <- spawn buffer+    t1 <- async $ do+        runEffect $   each [1..5]+                  >-> P.tee (toOutput output)+                  >-> for cat (\_ -> lift $ threadDelay 2000)+        performGC+    t2 <- async $ do+        runEffect $   fromInput input+                  >-> P.chain (\_ -> threadDelay 1000)+                  >-> P.print+        performGC+    wait t1+    wait t2++testReceiverClose :: Buffer Int -> IO ()+testReceiverClose buffer = do+    (output, input) <- spawn buffer+    t1 <- async $ do+        runEffect $   each [1..]+                  >-> P.tee (toOutput output)+                  >-> P.chain (\_ -> threadDelay 1000)+                  >-> P.print+        performGC+    t2 <- async $ do+        runEffect $ for (fromInput input >-> P.take 10) discard+        performGC+    wait t1+    wait t2++testReceiverCloseDelayedReceive :: Buffer Int -> IO ()+testReceiverCloseDelayedReceive buffer = do+    (output, input) <- spawn buffer+    t1 <- async $ do+        runEffect $   each [1..]+                  >-> P.tee (toOutput output)+                  >-> P.chain (\_ -> threadDelay 1000)+                  >-> labelPrint "Send"+        performGC+    t2 <- async $ do+        runEffect $   fromInput input+                  >-> P.take 10+                  >-> P.chain (\_ -> threadDelay 800)+                  >-> labelPrint "Recv"+        performGC+    wait t1+    wait t2++runTest :: IO () -> String -> IO ()+runTest test name = do+    putStrLn $ "Starting test: " ++ name+    hFlush stdout+    result <- timeout defaultTimeout test+    case result of+        Nothing -> do putStrLn $ "Test " ++ name ++ " timed out. Aborting."+                      exitFailure+        Just _  -> do putStrLn $ "Test " ++ name ++ " finished."+    hFlush stdout++runTestExpectTimeout :: IO () -> String -> IO ()+runTestExpectTimeout test name = do+    putStrLn $ "Starting test: " ++ name+    hFlush stdout+    result <- timeout defaultTimeout test+    case result of+        Nothing -> putStrLn $ "Test " ++ name ++ " timed out as expected."+        Just _  -> do+            putStrLn $+                   "Test "+                ++ name+                ++ " finished, but a timeout was expected. Aborting."+            exitFailure+    hFlush stdout++main :: IO ()+main = do+    runTest (testSenderClose Unbounded) "UnboundedSenderClose"+    runTest (testSenderClose $ Bounded 3) "BoundedFilledSenderClose"+    runTest (testSenderClose $ Bounded 7) "BoundedNotFilledSenderClose"+    runTest (testSenderClose Single) "SingleSenderClose"+    runTestExpectTimeout (testSenderCloseDelayedSend $ Latest 42) "LatestSenderClose"+    --+    runTest (testReceiverClose Unbounded) "UnboundedReceiverClose"+    runTest (testReceiverClose $ Bounded 3) "BoundedFilledReceiverClose"+    runTest (testReceiverClose $ Bounded 7) "BoundedNotFilledReceiverClose"+    runTest (testReceiverClose Single) "SingleReceiverClose"+    runTest (testReceiverCloseDelayedReceive $ Latest 42) "LatestReceiverClose"