diff --git a/BroadcastChan/Pipes.hs b/BroadcastChan/Pipes.hs new file mode 100644 --- /dev/null +++ b/BroadcastChan/Pipes.hs @@ -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 +-- 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 diff --git a/BroadcastChan/Pipes/Internal.hs b/BroadcastChan/Pipes/Internal.hs new file mode 100644 --- /dev/null +++ b/BroadcastChan/Pipes/Internal.hs @@ -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 diff --git a/BroadcastChan/Pipes/Throw.hs b/BroadcastChan/Pipes/Throw.hs new file mode 100644 --- /dev/null +++ b/BroadcastChan/Pipes/Throw.hs @@ -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 +-- 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 diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. 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/broadcast-chan-pipes.cabal b/broadcast-chan-pipes.cabal new file mode 100644 --- /dev/null +++ b/broadcast-chan-pipes.cabal @@ -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 +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 diff --git a/tests/PipeTest.hs b/tests/PipeTest.hs new file mode 100644 --- /dev/null +++ b/tests/PipeTest.hs @@ -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 + ]