pipes-concurrency 1.1.0 → 1.2.0
raw patch · 5 files changed
+1086/−844 lines, 5 filesdep −transformersdep ~basesetup-changedPVP ok
version bump matches the API change (PVP)
Dependencies removed: transformers
Dependency ranges changed: base
API changes (from Hackage documentation)
- Control.Proxy.Concurrent: data Size
+ Control.Proxy.Concurrent: Latest :: a -> Buffer a
+ Control.Proxy.Concurrent: data Buffer a
+ Control.Proxy.Concurrent: instance Alternative Output
+ Control.Proxy.Concurrent: instance Applicative Output
+ Control.Proxy.Concurrent: instance Functor Output
+ Control.Proxy.Concurrent: instance Monad Output
+ Control.Proxy.Concurrent: instance Monoid (Input a)
- Control.Proxy.Concurrent: Bounded :: Int -> Size
+ Control.Proxy.Concurrent: Bounded :: Int -> Buffer a
- Control.Proxy.Concurrent: Single :: Size
+ Control.Proxy.Concurrent: Single :: Buffer a
- Control.Proxy.Concurrent: Unbounded :: Size
+ Control.Proxy.Concurrent: Unbounded :: Buffer a
- Control.Proxy.Concurrent: recvS :: Proxy p => Output a -> () -> Producer p a IO ()
+ Control.Proxy.Concurrent: recvS :: Proxy p => Output a -> r -> p x' x y' a IO r
- Control.Proxy.Concurrent: spawn :: Size -> IO (Input a, Output a)
+ Control.Proxy.Concurrent: spawn :: Buffer a -> IO (Input a, Output a)
Files
- Control/Proxy/Concurrent.hs +249/−205
- Control/Proxy/Concurrent/Tutorial.hs +772/−573
- LICENSE +24/−24
- Setup.hs +2/−2
- pipes-concurrency.cabal +39/−40
Control/Proxy/Concurrent.hs view
@@ -1,205 +1,249 @@--- | Asynchronous communication between proxies - -{-# LANGUAGE CPP #-} - -#if __GLASGOW_HASKELL__ >= 702 -{-# LANGUAGE Trustworthy #-} -#endif -{- 'unsafeIOToSTM' requires the Trustworthy annotation. - - I use 'unsafeIOToSTM' to touch an IORef to mark it 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, - Size(..), - 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 ((<|>), (<*), pure) -import Control.Concurrent (forkIO) -import Control.Concurrent.STM (atomically, STM) -import Control.Monad.Trans.Class (lift) -import qualified Control.Concurrent.STM as S -import qualified Control.Proxy as P -import Data.IORef (newIORef, readIORef, mkWeakIORef) -import GHC.Conc.Sync (unsafeIOToSTM) -import System.Mem (performGC) - -{-| Spawn a mailbox of the specified 'Size' that has an 'Input' and 'Output' end --} -spawn :: Size -> IO (Input a, Output a) -spawn size = do - (read, write) <- case size of - Bounded n -> do - q <- S.newTBQueueIO n - let read = do - ma <- S.readTBQueue q - case ma of - Nothing -> S.unGetTBQueue q ma - _ -> return () - return ma - return (read, S.writeTBQueue q) - Unbounded -> do - q <- S.newTQueueIO - let read = do - ma <- S.readTQueue q - case ma of - Nothing -> S.unGetTQueue q ma - _ -> return () - return ma - return (read, S.writeTQueue q) - Single -> do - m <- S.newEmptyTMVarIO - let read = do - ma <- S.takeTMVar m - case ma of - Nothing -> S.putTMVar m ma - _ -> return () - return ma - return (read, S.putTMVar m) - - {- Use an IORef to keep track of whether the 'Input' end has been garbage - collected and run a finalizer when the collection occurs - - The finalizer cannot anticipate how many listeners there are, so it only - writes a single 'Nothing' and trusts that the supplied 'read' action - will not consume the 'Nothing'. - - The 'write' must be protected with the "pure ()" fallback so that it does - not deadlock if the 'Output' end has also been garbage collected. - -} - rUp <- newIORef () - mkWeakIORef rUp (S.atomically $ write Nothing <|> pure ()) - - {- Use an IORef to keep track of whether the 'Output' end has been garbage - collected and run a finalizer when the collection occurs - -} - rDn <- newIORef () - done <- S.newTVarIO False - mkWeakIORef rDn (S.atomically $ S.writeTVar done True) - - let quit = do - b <- S.readTVar done - S.check b - return False - continue a = do - write (Just a) - return True - {- The '_send' action aborts 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. - -} - _send a = (quit <|> continue a) <* unsafeIOToSTM (readIORef rUp) - _recv = read <* unsafeIOToSTM (readIORef rDn) - return (Input _send , Output _recv) - -{-| 'Size' specifies how many messages to store in the mailbox before 'send' - blocks. --} -data Size - -- | Store an 'Unbounded' number of messages - = Unbounded - -- | Store a 'Bounded' number of messages specified by the 'Int' argument - | Bounded Int - -- | Store only a 'Single' message (like @Bounded 1@, but more efficient) - | Single - --- | 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 } - --- | 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) } - -{-| Writes all messages flowing \'@D@\'ownstream to the given 'Input' - - 'sendD' terminates when the corresponding 'Output' is garbage collected. --} -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 <- lift $ S.atomically $ send input a - if alive - then do - x2 <- P.respond a - loop x2 - else return () - -{-| Convert an 'Output' to a 'P.Producer' - - 'recvS' terminates when the corresponding 'Input' is garbage collected. --} -recvS :: (P.Proxy p) => Output a -> () -> P.Producer p a IO () -recvS output () = P.runIdentityP go - where - go = do - ma <- lift $ S.atomically $ recv output - case ma of - Nothing -> return () - Just a -> do - P.respond a - go - -{- $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'. --} +-- | Asynchronous communication between proxies++{-# LANGUAGE CPP #-}++#if __GLASGOW_HASKELL__ >= 702+{-# LANGUAGE Trustworthy #-}+#endif+{- 'unsafeIOToSTM' requires the Trustworthy annotation.++ I use 'unsafeIOToSTM' to touch an IORef to mark it 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+ let read = do+ ma <- S.readTBQueue q+ case ma of+ Nothing -> S.unGetTBQueue q ma+ _ -> return ()+ return ma+ return (read, S.writeTBQueue q)+ Unbounded -> do+ q <- S.newTQueueIO+ let read = do+ ma <- S.readTQueue q+ case ma of+ Nothing -> S.unGetTQueue q ma+ _ -> return ()+ return ma+ return (read, S.writeTQueue q)+ Single -> do+ m <- S.newEmptyTMVarIO+ let read = do+ ma <- S.takeTMVar m+ case ma of+ Nothing -> S.putTMVar m ma+ _ -> return ()+ return ma+ return (read, S.putTMVar m)+ Latest a -> do+ t <- S.newTVarIO a+ let write ma = case ma of+ Nothing -> return ()+ Just a -> S.writeTVar t a+ return (fmap Just (S.readTVar t), write)++ {- Use an IORef to keep track of whether the 'Input' end has been garbage+ collected and run a finalizer when the collection occurs++ The finalizer cannot anticipate how many listeners there are, so it only+ writes a single 'Nothing' and trusts that the supplied 'read' action+ will not consume the 'Nothing'.++ The 'write' must be protected with the "pure ()" fallback so that it does+ not deadlock if the 'Output' end has also been garbage collected.+ -}+ rUp <- newIORef ()+ mkWeakIORef rUp (S.atomically $ write Nothing <|> pure ())++ {- Use an IORef to keep track of whether the 'Output' end has been garbage+ collected and run a finalizer when the collection occurs+ -}+ rDn <- newIORef ()+ done <- S.newTVarIO False+ mkWeakIORef rDn (S.atomically $ S.writeTVar done True)++ let quit = do+ b <- S.readTVar done+ S.check b+ return False+ continue a = do+ write (Just a)+ return True+ {- The '_send' action aborts 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.+ -}+ _send a = (quit <|> continue a) <* unsafeIOToSTM (readIORef rUp)+ _recv = read <* unsafeIOToSTM (readIORef rDn)+ 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)++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 :: (P.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'.+-}
Control/Proxy/Concurrent/Tutorial.hs view
@@ -1,573 +1,772 @@-{-| 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 - - -- * Callbacks - -- $callback - - -- * Safety - -- $safety - - -- * Conclusion - -- $conclusion - - -- * Appendix - -- $appendix - ) where - -import Control.Proxy -import Control.Proxy.Concurrent - -{- $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, - - * implement 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 () -> 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 :: Size -> IO (Input a, Output a) - - 'spawn' takes a mailbox 'Size' as an argument, and we specify that we want - our mailbox to store an 'Unbounded' number of message. '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 - -> import Control.Proxy.Concurrent -> -> main = do -> (input, output) <- spawn Unbounded -> ... - - 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 -> x -> p x a x 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? - -> -- Each worker refuses to process more than two values -> 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 - -> $ ./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 'Size' 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 >-> 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 -> $ --} - -{- $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. --} - -{- $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. --} - -{- $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 () -> 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 -> 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) - -> -- 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 - --} +{-| 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+-}
LICENSE view
@@ -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.
Setup.hs view
@@ -1,2 +1,2 @@-import Distribution.Simple -main = defaultMain +import Distribution.Simple+main = defaultMain
pipes-concurrency.cabal view
@@ -1,40 +1,39 @@-Name: pipes-concurrency -Version: 1.1.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 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, - transformers >= 0.2.0.0 && < 0.4 - Exposed-Modules: - Control.Proxy.Concurrent, - Control.Proxy.Concurrent.Tutorial - GHC-Options: -O2 +Name: pipes-concurrency+Version: 1.2.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 "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