diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/Pipes/Async.hs b/Pipes/Async.hs new file mode 100644 --- /dev/null +++ b/Pipes/Async.hs @@ -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 diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/pipes-async.cabal b/pipes-async.cabal new file mode 100644 --- /dev/null +++ b/pipes-async.cabal @@ -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 diff --git a/test/Main.hs b/test/Main.hs new file mode 100644 --- /dev/null +++ b/test/Main.hs @@ -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"