diff --git a/pipes-concurrency.cabal b/pipes-concurrency.cabal --- a/pipes-concurrency.cabal +++ b/pipes-concurrency.cabal @@ -1,6 +1,6 @@ Name: pipes-concurrency -Version: 2.0.12 -Cabal-Version: >=1.8.0.2 +Version: 2.0.13 +Cabal-Version: >=1.10 Build-Type: Simple License: BSD3 License-File: LICENSE @@ -35,13 +35,13 @@ async >= 2.0.0.0 && < 2.3 , contravariant >= 1.3.3 && < 1.6 , pipes >= 4.0 && < 4.4 , - semigroups < 0.19, stm >= 2.4.3 && < 2.6 , void >= 0.6 && < 1 Exposed-Modules: Pipes.Concurrent, Pipes.Concurrent.Tutorial GHC-Options: -O2 -Wall -Wcompat + Default-Language: Haskell2010 Test-Suite tests Type: exitcode-stdio-1.0 @@ -53,3 +53,4 @@ pipes-concurrency , stm >= 2.4.3 && < 2.6, async >= 2.0 && < 2.3 + Default-Language: Haskell2010 diff --git a/src/Pipes/Concurrent.hs b/src/Pipes/Concurrent.hs --- a/src/Pipes/Concurrent.hs +++ b/src/Pipes/Concurrent.hs @@ -11,6 +11,13 @@ fromInput, toOutput, + -- * Mailboxes + Mailbox(), + fromMailbox, + toMailbox, + send', + recv', + -- * Actors spawn, spawn', @@ -125,6 +132,43 @@ choose f i1 i2 = Output $ \a -> case f a of Left b -> send i1 b Right c -> send i2 c + +{-| Combines a source of values and a sink of values + + 'fromMailbox' uses 'Mailbox' as 'Pipes.Producer' + 'toMailbox' uses 'Mailbox' as 'Pipes.Consumer' + 'send\'' puts a value in the 'Mailbox' + 'recv\'' obtains a value from the 'Mailbox' +-} +type Mailbox a = (Output a, Input a) + +{-| Convert a 'Mailbox' to a 'Pipes.Producer' + + 'fromMailbox' terminates when the 'Mailbox' source of values is exhausted. +-} +fromMailbox :: (MonadIO m) => Mailbox a -> Producer' a m () +fromMailbox = fromInput . snd + +{-| Convert a 'Mailbox' to a 'Pipes.Consumer' + + 'toMailbox' terminates when the 'Mailbox' sink of values is exhausted. +-} +toMailbox :: (MonadIO m) => Mailbox a -> Consumer' a m () +toMailbox = toOutput . fst + +{-| Put a value in a 'Mailbox' + + 'send' returns 'False' if the 'Mailbox' sink is exhausted +-} +send' :: Mailbox a -> a -> STM Bool +send' = send . fst + +{-| Obtain a value from a 'Mailbox' + + 'recv' returns 'Nothing' if the 'Mailbox' source is exhausted +-} +recv' :: Mailbox a -> STM (Maybe a) +recv' = recv . snd {-| Convert an 'Output' to a 'Pipes.Consumer' diff --git a/src/Pipes/Concurrent/Tutorial.hs b/src/Pipes/Concurrent/Tutorial.hs --- a/src/Pipes/Concurrent/Tutorial.hs +++ b/src/Pipes/Concurrent/Tutorial.hs @@ -98,13 +98,13 @@ @ '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 + store. In this case we want our mailbox to store an 'unbounded' number of messages: > import Pipes.Concurrent > > main = do -> (output, input) <- spawn Unbounded +> (output, input) <- spawn unbounded > ... 'spawn' creates this mailbox in the background and then returns two values: @@ -166,7 +166,7 @@ Our final @main@ looks like this: > main = do -> (output, input) <- spawn Unbounded +> (output, input) <- spawn unbounded > > forkIO $ do runEffect $ lift user >~ toOutput output > performGC @@ -193,6 +193,44 @@ > $ -} +{- $mailbox + When we find it convenient we can use the 'Mailbox' type, which just + combines an 'Output' a with an 'Input' a together. + + So, we had the previous code: + +> 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 + + It could be changed into this equivalent implementation: + +> main = do +> mailbox <- spawn unbounded +> +> forkIO $ do runEffect $ lift user >~ toMailbox mailbox +> performGC +> +> forkIO $ do runEffect $ acidRain >-> toMailbox mailbox +> performGC +> +> runEffect $ fromMailbox mailbox >-> handler + + Behavior continues to be exactly the same, as expected. + + Real implementation can change, but we can think 'Mailbox' as the + following type: + +> type Mailbox a = (Output a, Input a) +-} + {- $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. @@ -218,7 +256,7 @@ > import Pipes.Concurrent > > main = do -> (output, input) <- spawn Unbounded +> (output, input) <- spawn unbounded > as <- forM [1..3] $ \i -> > async $ do runEffect $ fromInput input >-> worker i > performGC @@ -249,7 +287,7 @@ > user = P.stdinLn >-> P.takeWhile (/= "quit") > > main = do -> (output, input) <- spawn Unbounded +> (output, input) <- spawn unbounded > as <- forM [1..3] $ \i -> > async $ do runEffect $ fromInput input >-> worker i > performGC @@ -373,16 +411,16 @@ {- $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 + '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 + If we set the mailbox 'Buffer' to 'bounded 1', 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 +> (output, input) <- spawn (bounded 1) > as <- forM [1..3] $ \i -> > async $ do runEffect $ fromInput input >-> P.take 2 >-> worker i > performGC @@ -410,7 +448,7 @@ > Worker #3: Processed 4 > $ - Contrast this with an 'Unbounded' mailbox for the same program, which keeps + Contrast this with an 'unbounded' mailbox for the same program, which keeps accepting values until downstream finishes processing the first six values: > $ ./work @@ -441,12 +479,12 @@ > 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 + 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) +> (output, input) <- spawn (bounded 100) > ... > $ ./work @@ -480,8 +518,8 @@ > import Data.Monoid > > main = do -> (output1, input1) <- spawn Unbounded -> (output2, input2) <- spawn Unbounded +> (output1, input1) <- spawn unbounded +> (output2, input2) <- spawn unbounded > a1 <- async $ do > runEffect $ P.stdinLn >-> toOutput (output1 <> output2) > performGC @@ -541,21 +579,21 @@ 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': + 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) +> (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) + '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: @@ -568,18 +606,18 @@ > 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 + 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. - Another alternative is to use the 'Newest' mailbox, which is like a - 'Bounded' mailbox, except 'send' never blocks (the mailbox is never full). + Another alternative is to use the 'newest n' mailbox, which is like a + 'bounded' mailbox, except 'send' never blocks (the mailbox is never full). Instead, if there is no room 'send' will remove the oldest message from the mailbox to make room for a new message. - The 'New' mailbox is like the 'Newest' mailbox, except optimized for the - special case where you want to store a single message. You can use 'New' to + The 'newest 1' mailbox is like the 'newest n' mailbox, except optimized for the + special case where you want to store a single message. You can use 'newest 1' to read from a source that might potentially update rapidly, but still sleep if the source has no new values: @@ -590,7 +628,7 @@ > each [101..] -- More rapid updates > > main = do -> (output, input) <- spawn New +> (output, input) <- spawn (newest 1) > ... When the source goes quiet, the 'Input' will now block and wait, and will @@ -629,7 +667,7 @@ > > onLines' :: Producer String IO () > onLines' = do -> (output, input) <- lift $ spawn Single +> (output, input) <- lift $ spawn (bounded 1) > lift $ forkIO $ onLines (\str -> atomically $ send output str) > fromInput input > @@ -669,8 +707,8 @@ > import qualified Pipes.Prelude as P > > main = do -> (out1, in1) <- spawn Unbounded -> (out2, in2) <- spawn Unbounded +> (out1, in1) <- spawn unbounded +> (out2, in2) <- spawn unbounded > a1 <- async $ do > runEffect $ (each [1,2] >> fromInput in1) >-> toOutput out2 > performGC @@ -751,7 +789,7 @@ > Quit -> return () > >main = do -> (output, input) <- spawn Unbounded +> (output, input) <- spawn unbounded > > forkIO $ do runEffect $ lift user >~ toOutput output > performGC @@ -780,9 +818,9 @@ >user = P.stdinLn >-> P.takeWhile (/= "quit") > >main = do ->-- (output, input) <- spawn Unbounded ->-- (output, input) <- spawn Single -> (output, input) <- spawn (Bounded 100) +>-- (output, input) <- spawn unbounded +>-- (output, input) <- spawn (bounded 1) +> (output, input) <- spawn (bounded 100) > > as <- forM [1..3] $ \i -> >-- async $ do runEffect $ fromInput input >-> worker i @@ -816,7 +854,7 @@ > threadDelay 1000000 > >main = do -> (output, input) <- spawn (Latest 0) +> (output, input) <- spawn (latest 0) > a1 <- async $ do runEffect $ inputDevice >-> toOutput output > performGC > a2 <- async $ do runEffect $ fromInput input >-> P.take 5 >-> outputDevice @@ -837,7 +875,7 @@ > >onLines' :: Producer String IO () >onLines' = do -> (output, input) <- lift $ spawn Single +> (output, input) <- lift $ spawn (bounded 1) > lift $ forkIO $ onLines (\str -> atomically $ send output str) > fromInput input > diff --git a/tests/tests-main.hs b/tests/tests-main.hs --- a/tests/tests-main.hs +++ b/tests/tests-main.hs @@ -109,16 +109,16 @@ 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 (testSenderCloseDelayedSend New) "NewSenderClose" + runTest (testSenderClose unbounded) "UnboundedSenderClose" + runTest (testSenderClose $ bounded 3) "BoundedFilledSenderClose" + runTest (testSenderClose $ bounded 7) "BoundedNotFilledSenderClose" + runTest (testSenderClose $ bounded 1) "SingleSenderClose" + runTestExpectTimeout (testSenderCloseDelayedSend $ latest 42) "LatestSenderClose" + runTest (testSenderCloseDelayedSend (newest 1)) "NewSenderClose" -- - runTest (testReceiverClose Unbounded) "UnboundedReceiverClose" - runTest (testReceiverClose $ Bounded 3) "BoundedFilledReceiverClose" - runTest (testReceiverClose $ Bounded 7) "BoundedNotFilledReceiverClose" - runTest (testReceiverClose Single) "SingleReceiverClose" - runTest (testReceiverCloseDelayedReceive $ Latest 42) "LatestReceiverClose" - runTest (testReceiverClose New) "NewReceiverClose" + runTest (testReceiverClose unbounded) "UnboundedReceiverClose" + runTest (testReceiverClose $ bounded 3) "BoundedFilledReceiverClose" + runTest (testReceiverClose $ bounded 7) "BoundedNotFilledReceiverClose" + runTest (testReceiverClose $ bounded 1) "SingleReceiverClose" + runTest (testReceiverCloseDelayedReceive $ latest 42) "LatestReceiverClose" + runTest (testReceiverClose (newest 1)) "NewReceiverClose"