packages feed

broadcast-chan-conduit (empty) → 0.2.0

raw patch · 7 files changed

+296/−0 lines, 7 filesdep +basedep +broadcast-chandep +broadcast-chan-conduitsetup-changed

Dependencies added: base, broadcast-chan, broadcast-chan-conduit, broadcast-chan-tests, conduit, containers, resourcet, transformers, unliftio-core, void

Files

+ BroadcastChan/Conduit.hs view
@@ -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 <merijn@inconsistent.nl>+-- 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
+ BroadcastChan/Conduit/Internal.hs view
@@ -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
+ BroadcastChan/Conduit/Throw.hs view
@@ -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 <merijn@inconsistent.nl>+-- 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
+ LICENSE view
@@ -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.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ broadcast-chan-conduit.cabal view
@@ -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 <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:           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
+ tests/ConduitTest.hs view
@@ -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+    ]