diff --git a/BroadcastChan/Conduit.hs b/BroadcastChan/Conduit.hs new file mode 100644 --- /dev/null +++ b/BroadcastChan/Conduit.hs @@ -0,0 +1,40 @@ +{-# LANGUAGE Safe #-} +------------------------------------------------------------------------------- +-- | +-- Module : BroadcastChan.Conduit +-- 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 conduits and sinks that +-- process in parallel. +------------------------------------------------------------------------------- +module BroadcastChan.Conduit + ( 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.Conduit.Internal diff --git a/BroadcastChan/Conduit/Internal.hs b/BroadcastChan/Conduit/Internal.hs new file mode 100644 --- /dev/null +++ b/BroadcastChan/Conduit/Internal.hs @@ -0,0 +1,79 @@ +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE Trustworthy #-} +{-# LANGUAGE ScopedTypeVariables #-} +module BroadcastChan.Conduit.Internal (parMapM, parMapM_) where + +import Control.Monad ((>=>)) +import Control.Monad.Trans.Resource (MonadResource) +import Control.Monad.Trans.Class (lift) +import Control.Monad.IO.Unlift (MonadUnliftIO(askUnliftIO), UnliftIO(..)) +import Data.Acquire + (ReleaseType(ReleaseException), allocateAcquire, mkAcquireType) +import Data.Conduit +import qualified Data.Conduit.List as C +import Data.Foldable (traverse_) +import Data.Void (Void) + +import BroadcastChan.Extra + (BracketOnError(..), Handler, mapHandler, runParallel, runParallel_) + +bracketOnError :: MonadResource m => IO a -> (a -> IO ()) -> m r -> m r +bracketOnError alloc clean work = + allocateAcquire (mkAcquireType alloc cleanup) >>= const work + where + cleanup x ReleaseException = clean x + cleanup _ _ = return () + +-- | Create a conduit that processes inputs in parallel. +-- +-- This function does __NOT__ guarantee that input elements are processed or +-- output in a deterministic order! +parMapM + :: (MonadResource m, MonadUnliftIO m) + => Handler m a + -- ^ Exception handler + -> Int + -- ^ Number of parallel threads to use + -> (a -> m b) + -- ^ Function to run in parallel + -> ConduitM a b m () +parMapM hnd threads workFun = do + UnliftIO runInIO <- lift askUnliftIO + + Bracket{allocate,cleanup,action} <- runParallel + (Left yield) + (mapHandler runInIO hnd) + threads + (runInIO . workFun) + body + + bracketOnError allocate cleanup action + where + body :: Monad m => (a -> m ()) -> (a -> m (Maybe b)) -> ConduitM a b m () + body buffer process = do + C.isolate threads .| C.mapM_ buffer + awaitForever $ lift . process >=> traverse_ yield + +-- | Create a conduit sink that consumes inputs in parallel. +-- +-- This function does __NOT__ guarantee that input elements are processed or +-- output in a deterministic order! +parMapM_ + :: (MonadResource m, MonadUnliftIO m) + => Handler m a + -- ^ Exception handler + -> Int + -- ^ Number of parallel threads to use + -> (a -> m ()) + -- ^ Function to run in parallel + -> ConduitM a Void m () +parMapM_ hnd threads workFun = do + UnliftIO runInIO <- lift askUnliftIO + + Bracket{allocate,cleanup,action} <- runParallel_ + (mapHandler runInIO hnd) + threads + (runInIO . workFun) + C.mapM_ + + bracketOnError allocate cleanup action diff --git a/BroadcastChan/Conduit/Throw.hs b/BroadcastChan/Conduit/Throw.hs new file mode 100644 --- /dev/null +++ b/BroadcastChan/Conduit/Throw.hs @@ -0,0 +1,40 @@ +{-# LANGUAGE Safe #-} +------------------------------------------------------------------------------- +-- | +-- Module : BroadcastChan.Conduit.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 conduits and sinks that +-- process in parallel. +------------------------------------------------------------------------------- +module BroadcastChan.Conduit.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.Conduit.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-conduit.cabal b/broadcast-chan-conduit.cabal new file mode 100644 --- /dev/null +++ b/broadcast-chan-conduit.cabal @@ -0,0 +1,68 @@ +Name: broadcast-chan-conduit +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: Conduit-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.Conduit + BroadcastChan.Conduit.Throw + Other-Modules: BroadcastChan.Conduit.Internal + + Other-Extensions: NamedFieldPuns + Safe + ScopedTypeVariables + Trustworthy + + Build-Depends: base >= 4.6 && < 5 + , broadcast-chan == 0.2.0.* + , conduit >= 1.2 && < 1.4 + , resourcet >= 1.1 && < 1.3 + , transformers >= 0.2 && < 0.6 + , unliftio-core >= 0.1 && < 0.2 + + if impl(ghc < 7.10) + Build-Depends: void >= 0.1 && < 0.8 + +Test-Suite conduit + Default-Language: Haskell2010 + Type: exitcode-stdio-1.0 + Main-Is: ConduitTest.hs + GHC-Options: -Wall -fno-warn-unused-do-bind -threaded + Hs-Source-Dirs: tests + + Build-Depends: base + , broadcast-chan-conduit + , broadcast-chan-tests + , containers >= 0.4 && < 0.6 + , conduit >= 1.2 && < 1.4 + +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/ConduitTest.hs b/tests/ConduitTest.hs new file mode 100644 --- /dev/null +++ b/tests/ConduitTest.hs @@ -0,0 +1,37 @@ +import Control.Monad (void) +import Data.Set (Set) +import qualified Data.Set as S +import Data.Conduit +import qualified Data.Conduit.List as C + +import BroadcastChan.Conduit +import BroadcastChan.Test + +sequentialSink :: [a] -> (a -> IO b) -> IO () +sequentialSink inputs f = + runConduitRes $ C.sourceList inputs .| C.mapM_ (liftIO . void . f) + +parallelSink :: Handler IO a -> [a] -> (a -> IO b) -> Int -> IO () +parallelSink hnd inputs f n = runConduitRes $ + C.sourceList inputs .| parMapM_ handler n (liftIO . void . f) + where + handler = mapHandler liftIO hnd + +sequentialFold :: Ord b => [a] -> (a -> IO b) -> IO (Set b) +sequentialFold inputs f = runConduitRes $ + C.sourceList inputs .| C.mapM (liftIO . f) .| C.foldMap S.singleton + +parallelFold + :: Ord b => Handler IO a -> [a] -> (a -> IO b) -> Int -> IO (Set b) +parallelFold hnd inputs f n = runConduitRes $ + C.sourceList inputs + .| parMapM handler n (liftIO . f) + .| C.foldMap S.singleton + where + handler = mapHandler liftIO hnd + +main :: IO () +main = runTests "conduit" $ + [ genStreamTests "sink" sequentialSink parallelSink + , genStreamTests "fold" sequentialFold parallelFold + ]