packages feed

pipes-async (empty) → 0.1.0

raw patch · 5 files changed

+355/−0 lines, 5 filesdep +basedep +hspecdep +lifted-asyncsetup-changed

Dependencies added: base, hspec, lifted-async, lifted-base, monad-control, pipes, pipes-async, pipes-safe, stm

Files

+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2015, John Wiegley++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 John Wiegley 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.
+ Pipes/Async.hs view
@@ -0,0 +1,149 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE TypeFamilies #-}++module Pipes.Async where++import Control.Exception (AsyncException(..))+import Control.Concurrent.Lifted+import Control.Concurrent.Async.Lifted+import Control.Concurrent.STM+import Control.Monad (liftM)+import Control.Monad.Trans.Control+import Pipes+import Pipes.Internal+import Pipes.Safe++-- | A substitute for 'Pipes.>->' that executes both the upstream producer and+-- downstream consumer in separate threads (see '>&>' for an operator version,+-- with a default queue size of 16 slots). The reason separate threads are+-- used for both sides is so that the current thread (running 'runEffect' or+-- 'toListM', for example) can manage the bidirectional semantics for the+-- resulting Proxy. That is:+--+-- Upstream is executed in task A, downstream in task B, and 'runEffect' in+-- the parent thread. Tasks A and B are connected so that 'b' values produced+-- in A are immediately enqueued and available to B. 'runEffect' does not manage+-- passing 'b' values from A to B, as it normally would; rather they flow+-- directly through a 'TBQueue' side-channel.+--+-- If upstream should attempt to send an @a'@ value further upstream,+-- expecting an 'a' in return, this will block task A as 'runEffect' sends the+-- request further up the chain. Or, should downstream send a 'c' value+-- downstream and expect a @c'@, it will block task B as 'runEffect' sends the+-- response further down the chain.+--+-- If upstream exits, its result value is enqueued until downstream sees it,+-- at which point 'runEffect' terminates with this value. However, if+-- downstream should exit first, this result is communicated directly to+-- 'runEffect', which returns it immediately, canceling both threads. Thus,+-- execution lifetime is biased toward the downstream consumer, since it is+-- more likely that downstream will consume elements until there are none+-- left, than that upstream would produce elements while waiting for+-- downstream to terminate.+--+-- If an exception occurs in either upstream or downstream, it is re-thrown in+-- the 'runEffect' thread. Also, no matter what happens, both the upstream and+-- downstream threads are canceled at the conclusion of the enclosing+-- 'MonadSafe' block.+--+-- Note: Using '>&>' should be a drop-in replacement for 'Pipes.>->' anywhere+-- it is used, without changing the meaning of the pipeline; however, how the+-- composition is associated has an effect on the concurrency. For example, @a+-- >-> (b >&> c)@ causes 'b' and 'c' to be executed concurrently, with effects+-- from 'a' occuring in the parent thread (while 'b' blocks waiting on the+-- response). By contrast, @(a >-> b) >&> c@ executes @a >-> b@ and 'c'+-- concurrently, with nothing happening in the parent thread except to wait on+-- the final result. This will generally be faster since value passing through+-- 'MVar' will not be necessary. This is also the default interpretation of @a+-- >-> b >&> c@, since both operators left-associate at the same level.+--++buffer :: (MonadBaseControl IO m, MonadBaseControl IO (Base m),+           MonadSafe m, MonadIO m, MonadMask m)+       => Int                    -- ^ Number of slots in the bounded queue+       -> Proxy a' a () b m r    -- ^ Upstream producer+       -> Proxy () b c' c m r    -- ^ Downstream consumer+       -> Proxy a' a c' c m r+buffer sz ups downs = M $ do+    q   <- liftIO $ newTBQueueIO 3  -- control channel+    qeb <- liftIO $ newTBQueueIO sz -- bounded queue of 'b' values flowing+                                   -- from upstream to downstream+    me <- myThreadId+    hd <- spawn $ toDowns me q qeb+    hu <- spawn $ fromUps me q qeb++    mainLoop q $ \r -> do+        release hu+        release hd+        return $ Pure r+  where+    spawn f = do+        h <- async f+        link h+        register $ cancel h++    readQ q    = liftIO $ atomically $ readTBQueue q+    writeQ q x = liftIO $ atomically $ writeTBQueue q x++    mainLoop q done = loop+      where+        loop = readQ q >>= \case+            Left u -> case u of+                Request a' fa  -> return $ Request a' $ (>> M loop) . fa+                Respond _  _   -> error "Respond never comes from ups"+                M          _   -> error "M never comes from ups"+                Pure       r   -> done r+            Right d -> case d of+                Request _  _   -> error "Request never comes from downs"+                Respond c  fc' -> return $ Respond c $ (>> M loop) . fc'+                M          _   -> error "M never comes from downs"+                Pure       r   -> done r++    guarded :: (MonadBaseControl IO (Base m), MonadSafe m, MonadCatch m)+            => ThreadId+            -> ((Proxy a' a b' b m r -> m ()) -> Proxy a' a b' b m r -> m ())+            -> Proxy a' a b' b m r+            -> m ()+    guarded parent f = loop+      where+        loop p = f loop p+            `catch` (\e -> case e :: AsyncException of+                ThreadKilled -> return ()+                _ -> liftBase $ throwTo parent e)+            `catch` (\e -> liftBase $ throwTo parent (e :: SomeException))++    fromUps parent q qeb = flip (guarded parent) ups $ \loop -> \case+        Request a' fa  -> throughVar q (Left . Request a') fa >>= loop+        Respond b  fb' -> writeQ qeb (Right b) >> loop (fb' ())+        M       m      -> m >>= loop+        Pure    r      -> writeQ qeb (Left r) -- this enqueues exit++    toDowns parent q qeb = flip (guarded parent) downs $ \loop -> \case+        Request () fb  -> readQ qeb >>= \case+            Left r  -> writeQ q (Right (Pure r))+            Right b -> loop (fb b)+        Respond c  fc' -> throughVar q (Right . Respond c) fc' >>= loop+        M       m      -> m >>= loop+        Pure    r      -> writeQ q (Right (Pure r)) -- this causes exit++    throughVar q x f = do+        var <- newVar+        writeQ q $ x $ \v -> M $ do+            putVar var v+            return $ Pure $ error "(>> M loop) throws this value away"+        f `liftM` takeVar var+      where+        newVar     = liftIO newEmptyTMVarIO+        putVar v z = liftIO $ atomically $ putTMVar v z+        takeVar v  = liftIO $ atomically $ takeTMVar v++infixl 7 >&>+(>&>) :: (MonadBaseControl IO m, MonadBaseControl IO (Base m),+          MonadSafe m, MonadIO m, MonadMask m)+      => Proxy a' a () b m r -> Proxy () b c' c m r+      -> Proxy a' a c' c m r+(>&>) = buffer 16
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ pipes-async.cabal view
@@ -0,0 +1,56 @@+-- Initial pipes-async.cabal generated by cabal init.  For further +-- documentation, see http://haskell.org/cabal/users-guide/++name:                pipes-async+version:             0.1.0+synopsis:            A higher-level interface to using concurrency with pipes+description:         Provides combinators like '>&>' for easily adding concurrency.+homepage:            https://github.com/jwiegley/pipes-async+license:             BSD3+license-file:        LICENSE+author:              John Wiegley+maintainer:          johnw@newartisans.com+copyright:           Copyright (c) 2015, John Wiegley+category:            Control+build-type:          Simple+cabal-version:       >=1.10++library+  exposed-modules:     Pipes.Async+  ghc-options:         -Wall+  other-extensions:    +      FlexibleContexts+    , FlexibleInstances+    , MultiParamTypeClasses+    , UndecidableInstances+    , LambdaCase+    , TypeFamilies+  build-depends:  +      base          >=4.8 && <4.9+    , lifted-async  >=0.7 && <0.8+    , lifted-base   >=0.2 && <0.3+    , stm           >=2.4 && <2.5+    , monad-control >=1.0 && <1.1+    , pipes         >=4.1 && <4.2+    , pipes-safe    >=2.2.3 && <2.3+  default-language:    Haskell2010++test-suite test+  hs-source-dirs: . test+  default-language: Haskell2010+  main-is: Main.hs+  type: exitcode-stdio-1.0+  ghc-options: -threaded -with-rtsopts "-N2" -Wall+  build-depends:+      base          >=4.8 && <4.9+    , lifted-async  >=0.7 && <0.8+    , lifted-base   >=0.2 && <0.3+    , stm           >=2.4 && <2.5+    , monad-control >=1.0 && <1.1+    , pipes         >=4.1 && <4.2+    , pipes-safe    >=2.2 && <2.3+    , hspec         >=1.4+    , pipes-async+    -- , transformers+    -- , transformers-base+    -- , containers
+ test/Main.hs view
@@ -0,0 +1,118 @@+{-# OPTIONS_GHC -fno-warn-unused-do-bind #-}++module Main where++import           Control.Concurrent hiding (yield)+import           Control.Concurrent.STM+import           Control.Monad+import           Pipes+import           Pipes.Async+import qualified Pipes.Prelude as P+import           Pipes.Safe+import           Test.Hspec++segment1 :: Proxy x' x () Int (SafeT IO) ()+segment1 = do+    liftIO $ putStrLn "Starting segment 1"+    each [(1 :: Int)..3]++segment2 :: (Num b, MonadIO m) => TMVar String -> Proxy () b () b m ()+segment2 var = do+    liftIO $ putStrLn "Starting segment 2"+    replicateM_ 3 $ yield . (+1) =<< await+    -- P.map (+1)+    liftIO $ do+        putStrLn "atomically $ tryPutTMVar var \"sender\""+        atomically $ tryPutTMVar var "sender"+        putStrLn "Done sending"++segment3 :: (Num c, Show c, MonadIO m) => TMVar String -> Proxy () c () c m ()+segment3 var = do+    liftIO $ putStrLn "Starting segment 3"+    for cat $ \a -> do+        liftIO $ do+            putStrLn "Got a value from upstream"+            threadDelay 300000+            putStrLn "atomically $ tryPutTMVar var \"receiver\""+            atomically $ tryPutTMVar var "receiver"+            print a+        yield (a + 100)+    liftIO $ putStrLn "Done receiving"++segment4 :: Proxy () Int () Int (SafeT IO) b+segment4 = do+    liftIO $ putStrLn "Starting segment 4"+    P.map (+10)++main :: IO ()+main = hspec $ do+    describe "Sanity tests" $ do+        it "Works in the simplest case" $ do+            var' <- newEmptyTMVarIO+            ys <- runSafeT $ P.toListM $+                segment1 >-> (segment2 var' >-> segment3 var') >-> segment4+            ys `shouldBe` [112, 113, 114]++            var <- newEmptyTMVarIO+            xs <- runSafeT $ P.toListM $+                segment1 >-> (segment2 var >&> segment3 var) >-> segment4+            xs `shouldBe` [112, 113, 114]++            result <- atomically $ tryTakeTMVar var+            result `shouldBe` Just "sender"++        it "Behaves the same under composition (one)" $ do+            var' <- newEmptyTMVarIO+            ys <- runSafeT $ P.toListM $+                (segment1 >-> segment2 var') >-> segment3 var' >-> segment4+            ys `shouldBe` [112, 113, 114]++            var <- newEmptyTMVarIO+            xs <- runSafeT $ P.toListM $+                (segment1 >-> segment2 var) >&> segment3 var >-> segment4+            xs `shouldBe` [112, 113, 114]++            result <- atomically $ tryTakeTMVar var+            result `shouldBe` Just "sender"++        it "Behaves the same under composition (two)" $ do+            var' <- newEmptyTMVarIO+            ys <- runSafeT $ P.toListM $+                (segment1 >-> segment2 var' >-> segment3 var') >-> segment4+            ys `shouldBe` [112, 113, 114]++            var <- newEmptyTMVarIO+            xs <- runSafeT $ P.toListM $+                (segment1 >-> segment2 var >&> segment3 var) >-> segment4+            xs `shouldBe` [112, 113, 114]++            result <- atomically $ tryTakeTMVar var+            result `shouldBe` Just "sender"++        it "Behaves the same under composition (three)" $ do+            var' <- newEmptyTMVarIO+            ys <- runSafeT $ P.toListM $+                segment1 >-> segment2 var' >-> (segment3 var' >-> segment4)+            ys `shouldBe` [112, 113, 114]++            var <- newEmptyTMVarIO+            xs <- runSafeT $ P.toListM $+                segment1 >-> segment2 var >&> (segment3 var >-> segment4)+            xs `shouldBe` [112, 113, 114]++            result <- atomically $ tryTakeTMVar var+            result `shouldBe` Just "sender"++        it "Behaves the same under composition (four)" $ do+            var' <- newEmptyTMVarIO+            ys <- runSafeT $ P.toListM $+                segment1 >-> (segment2 var' >-> segment3 var' >-> segment4)+            ys `shouldBe` [112, 113, 114]++            var <- newEmptyTMVarIO+            xs <- runSafeT $ P.toListM $+                segment1 >-> (segment2 var >&> segment3 var >-> segment4)+            xs `shouldBe` [112, 113, 114]++            result <- atomically $ tryTakeTMVar var+            result `shouldBe` Just "sender"