broadcast-chan-pipes (empty) → 0.2.0

raw patch · 7 files changed

+277/−0 lines, 7 filesdep +basedep +broadcast-chandep +broadcast-chan-pipessetup-changed

Dependencies added: base, broadcast-chan, broadcast-chan-pipes, broadcast-chan-tests, containers, foldl, pipes, pipes-safe

Files

• BroadcastChan/Pipes.hs +40/−0
• BroadcastChan/Pipes/Internal.hs +67/−0
• BroadcastChan/Pipes/Throw.hs +40/−0
• LICENSE +30/−0
• Setup.hs +2/−0
• broadcast-chan-pipes.cabal +63/−0
• tests/PipeTest.hs +35/−0

@@ -0,0 +1,40 @@+{-# LANGUAGE Safe #-}+-------------------------------------------------------------------------------+-- |+-- Module      :  BroadcastChan.Pipes+-- Copyright   :  (C) 2014-2018 Merijn Verstraaten+-- License     :  BSD-style (see the file LICENSE)+-- Maintainer  :  Merijn Verstraaten <merijn@inconsistent.nl>+-- Stability   :  experimental+-- Portability :  haha+--+-- This module is identical to "BroadcastChan", but replaces the parallel+-- processing operations with functions for creating producers and effects that+-- process in parallel.+-------------------------------------------------------------------------------+module BroadcastChan.Pipes+    ( Action(..)+    , Handler(..)+    , parMapM+    , parMapM_+    -- * Re-exports from "BroadcastChan"+    -- ** Datatypes+    , BroadcastChan+    , Direction(..)+    , In+    , Out+    -- ** Construction+    , newBroadcastChan+    , newBChanListener+    -- ** Basic Operations+    , closeBChan+    , isClosedBChan+    , getBChanContents+    -- ** Foldl combinators+    -- | Combinators for use with Tekmo's @foldl@ package.+    , foldBChan+    , foldBChanM+    ) where++import BroadcastChan hiding (parMapM_)+import BroadcastChan.Pipes.Internal

@@ -0,0 +1,67 @@+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE Safe #-}+{-# LANGUAGE ScopedTypeVariables #-}+module BroadcastChan.Pipes.Internal (parMapM, parMapM_) where++import Control.Monad ((>=>), replicateM)+import Data.Foldable (traverse_)+import Pipes+import qualified Pipes.Prelude as P+import Pipes.Safe (MonadSafe)+import qualified Pipes.Safe as Safe++import BroadcastChan.Extra+    (BracketOnError(..), Handler, runParallel, runParallel_)++bracketOnError :: MonadSafe m => IO a -> (a -> IO b) -> m c -> m c+bracketOnError alloc clean =+  Safe.bracketOnError (liftIO alloc) (liftIO . clean) . const++-- | Create a producer that processes its inputs in parallel.+--+-- This function does __NOT__ guarantee that input elements are processed or+-- output in a deterministic order!+parMapM+    :: forall a b m+     . MonadSafe m+    => Handler IO a+    -- ^ Exception handler+    -> Int+    -- ^ Number of parallel threads to use+    -> (a -> IO b)+    -- ^ Function to run in parallel+    -> Producer a m ()+    -- ^ Input producer+    -> Producer b m ()+parMapM hndl i f prod = do+    Bracket{allocate,cleanup,action} <- runParallel (Left yield) hndl i f body+    bracketOnError allocate cleanup action+  where+    body :: (a -> m ()) -> (a -> m (Maybe b)) -> Producer b m ()+    body buffer process = prod >-> work+      where+        work :: Pipe a b m ()+        work = do+            replicateM i (await >>= lift . buffer)+            for cat $ lift . process >=> traverse_ yield++-- | Create an Effect that processes its inputs in parallel.+--+-- This function does __NOT__ guarantee that input elements are processed or+-- output in a deterministic order!+parMapM_+    :: MonadSafe m+    => Handler IO a+    -- ^ Exception handler+    -> Int+    -- ^ Number of parallel threads to use+    -> (a -> IO ())+    -- ^ Function to run in parallel+    -> Producer a m r+    -- ^ Input producer+    -> Effect m r+parMapM_ hndl i f prod = do+    Bracket{allocate,cleanup,action} <- runParallel_ hndl i f workProd+    bracketOnError allocate cleanup action+  where+    workProd buffer = prod >-> P.mapM_ buffer

@@ -0,0 +1,40 @@+{-# LANGUAGE Safe #-}+-------------------------------------------------------------------------------+-- |+-- Module      :  BroadcastChan.Pipes.Throw+-- Copyright   :  (C) 2014-2018 Merijn Verstraaten+-- License     :  BSD-style (see the file LICENSE)+-- Maintainer  :  Merijn Verstraaten <merijn@inconsistent.nl>+-- Stability   :  experimental+-- Portability :  haha+--+-- This module is identical to "BroadcastChan.Throw", but replaces the parallel+-- processing operations with functions for creating producers and effects that+-- process in parallel.+-------------------------------------------------------------------------------+module BroadcastChan.Pipes.Throw+    ( Action(..)+    , Handler(..)+    , parMapM+    , parMapM_+    -- * Re-exports from "BroadcastChan.Throw"+    -- ** Datatypes+    , BroadcastChan+    , Direction(..)+    , In+    , Out+    -- ** Construction+    , newBroadcastChan+    , newBChanListener+    -- ** Basic Operations+    , closeBChan+    , isClosedBChan+    , getBChanContents+    -- ** Foldl combinators+    -- | Combinators for use with Tekmo's @foldl@ package.+    , foldBChan+    , foldBChanM+    ) where++import BroadcastChan.Throw hiding (parMapM_)+import BroadcastChan.Pipes.Internal

@@ -0,0 +1,30 @@+Copyright (c) 2013-2017, Merijn Verstraaten++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 Merijn Verstraaten 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.

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

@@ -0,0 +1,63 @@+Name:               broadcast-chan-pipes+Version:            0.2.0++Homepage:           https://github.com/merijn/broadcast-chan+Bug-Reports:        https://github.com/merijn/broadcast-chan/issues++Author:             Merijn Verstraaten+Maintainer:         Merijn Verstraaten <merijn@inconsistent.nl>+Copyright:          Copyright © 2014-2018 Merijn Verstraaten++License:            BSD3+License-File:       LICENSE++Category:           System+Cabal-Version:      >= 1.10+Build-Type:         Simple+Tested-With:        GHC == 7.6.3, GHC == 7.8.4, GHC == 7.10.3, GHC == 8.0.2,+                    GHC == 8.2.2, GHC == 8.4.3, GHC == 8.6.1++Synopsis:           Pipes-based parallel streaming code for broadcast-chan++Description:+    __WARNING:__ While the code in this library should be fairly stable and+    production, the API is something I'm still working on. API changes will+    follow the PVP, but __expect__ breaking API changes in future versions!++Library+  Default-Language:     Haskell2010+  GHC-Options:          -Wall -O2 -fno-warn-unused-do-bind+  Exposed-Modules:      BroadcastChan.Pipes+                        BroadcastChan.Pipes.Throw+  Other-Modules:        BroadcastChan.Pipes.Internal++  Other-Extensions:     NamedFieldPuns+                        Safe+                        ScopedTypeVariables++  Build-Depends:        base >= 4.6 && < 5+               ,        broadcast-chan == 0.2.0.*+               ,        pipes >= 4.1.6 && < 4.4+               ,        pipes-safe == 2.2.*++Test-Suite pipes+  Default-Language:     Haskell2010+  Type:                 exitcode-stdio-1.0+  Main-Is:              PipeTest.hs+  GHC-Options:          -Wall -fno-warn-unused-do-bind+  Hs-Source-Dirs:       tests+  Build-Depends:        base+               ,        broadcast-chan-pipes+               ,        broadcast-chan-tests+               ,        containers >= 0.4 && < 0.6+               ,        foldl >= 1.0.4 && < 1.5+               ,        pipes >= 4.1.6 && < 4.4+               ,        pipes-safe == 2.2.*++Source-Repository head+  Type:     git+  Location: ssh://github.com:merijn/broadcast-chan.git++Source-Repository head+  Type:     mercurial+  Location: https://bitbucket.org/merijnv/broadcast-chan

@@ -0,0 +1,35 @@+import Control.Foldl (purely, set)+import Data.Set (Set)+import Pipes+import qualified Pipes.Prelude as P+import Pipes.Safe (runSafeT)++import BroadcastChan.Pipes+import BroadcastChan.Test++sequentialSink :: [a] -> (a -> IO b) -> IO ()+sequentialSink inputs f = runSafeT . runEffect $+    each inputs >-> P.mapM_ (liftIO . void. f)++parallelSink :: Handler IO a -> [a] -> (a -> IO b) -> Int -> IO ()+parallelSink hnd inputs f n =+  runSafeT . runEffect $ parMapM_ handler n (void . f) $ each inputs+  where+    handler = mapHandler liftIO hnd++sequentialFold :: Ord b => [a] -> (a -> IO b) -> IO (Set b)+sequentialFold inputs f = runSafeT $ purely P.fold set $+    each inputs >-> P.mapM (liftIO . f)++parallelFold+    :: Ord b => Handler IO a -> [a] -> (a -> IO b) -> Int -> IO (Set b)+parallelFold hnd inputs f n =+  runSafeT . purely P.fold set . parMapM handler n f $ each inputs+  where+    handler = mapHandler liftIO hnd++main :: IO ()+main = runTests "pipes" $+    [ genStreamTests "sink" sequentialSink parallelSink+    , genStreamTests "fold" sequentialFold parallelFold+    ]