conduit-algorithms 0.0.6.1 → 0.0.7.0

raw patch · 7 files changed

+99/−469 lines, 7 filesdep +lzma-conduitdep +vectordep ~conduit-combinatorsPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: lzma-conduit, vector

Dependency ranges changed: conduit-combinators

API changes (from Hackage documentation)

Files

• ChangeLog +8/−0
• Data/Conduit/Algorithms.hs +0/−130
• Data/Conduit/Algorithms/Async.hs +0/−213
• Data/Conduit/Algorithms/Tests.hs +22/−1
• Data/Conduit/Algorithms/Utils.hs +0/−61
• README.md +0/−1
• conduit-algorithms.cabal +69/−63

@@ -1,3 +1,11 @@+Version 0.0.7.0 2018-01-17 by luispedro+	* Add unorderedAsyncMapC+	* Add Data.Conduit.Algorithms.Async.ByteString module+	* Add Data.Conduit.Algorithms.Storable++Version 0.0.6.1 2017-11-01 by luispedro+	* Remove Unix dependency (compile on Windows)+ Version 0.0.6.0 2017-10-09 by luispedro 	* More efficient mergeC conduit 

@@ -1,130 +0,0 @@-{-|-Module      : Data.Conduit.Algorithms-Copyright   : 2013-2017 Luis Pedro Coelho-License     : MIT-Maintainer  : luis@luispedro.org--Simple algorithms packaged as Conduits--}---{-# LANGUAGE Rank2Types #-}-module Data.Conduit.Algorithms-    ( uniqueOnC-    , uniqueC-    , removeRepeatsC-    , mergeC-    , mergeC2-    ) where--import qualified Data.Conduit as C-import qualified Data.Conduit.Internal as CI-import qualified Data.Set as S-import           Data.List (sortBy, insertBy)-import           Control.Monad.Trans.Class (lift)--import           Data.Conduit.Algorithms.Utils (awaitJust)----- | Unique conduit.------ For each element, it checks its key (using the @a -> b@ key function) and--- yields it if it has not seen it before.------ Note that this conduit /does not/ assume that the input is sorted. Instead--- it uses a 'Data.Set' to store previously seen elements. Thus, memory usage--- is O(N) and time is O(N log N). If the input is sorted, you can use--- 'removeRepeatsC'-uniqueOnC :: (Ord b, Monad m) => (a -> b) -> C.Conduit a m a-uniqueOnC f = checkU (S.empty :: S.Set b)-    where-        checkU cur = awaitJust $ \val ->-                        if f val `S.member` cur-                            then checkU cur-                            else do-                                C.yield val-                                checkU (S.insert (f val) cur)--- | Unique conduit------ See 'uniqueOnC' and 'removeRepeatsC'-uniqueC :: (Ord a, Monad m) => C.Conduit a m a-uniqueC = uniqueOnC id---- | Removes repeated elements------ @---  yieldMany [0, 0, 1, 1, 1, 2, 2, 0] .| removeRepeatsC .| consume--- @------ is equivalent to @[0, 1, 2, 0]@------ See 'uniqueC' and 'uniqueOnC'-removeRepeatsC :: (Eq a, Monad m) => C.Conduit a m a-removeRepeatsC = awaitJust removeRepeatsC'-    where-        removeRepeatsC' prev = C.await >>= \case-                                        Nothing -> C.yield prev-                                        Just next-                                            | next == prev -> removeRepeatsC' prev-                                            | otherwise -> do-                                                        C.yield prev-                                                        removeRepeatsC' next----- | Merge a list of sorted sources to produce a single (sorted) source------ This takes a list of sorted sources and produces a 'C.Source' which outputs--- all elements in sorted order.------ See 'mergeC2'-mergeC :: (Ord a, Monad m) => [C.Source m a] -> C.Source m a-mergeC [a] = a-mergeC [a,b] = mergeC2 a b-mergeC cs = CI.ConduitM $ \rest -> let-        go [] = rest ()-        go allc@(CI.HaveOutput c_next _ v:larger) =-            CI.HaveOutput (norm1 c_next >>= go . insert1 larger) (finalizeAll allc) v-        go _ = error "This situation should have been impossible (mergeC/compareHO)"-        insert1 larger CI.Done{} = larger-        insert1 larger c = insertBy compareHO c larger-        norm1 :: Monad m => CI.Pipe () i o () m () -> CI.Pipe () i o () m (CI.Pipe () i o () m ())-        norm1 c@CI.HaveOutput{} = return c-        norm1 c@CI.Done{} = return c-        norm1 (CI.PipeM p) = lift p >>= norm1-        norm1 (CI.NeedInput _ next) = norm1 (next ())-        norm1 (CI.Leftover next ()) = norm1 next-        isHO CI.HaveOutput{} = True-        isHO _ = False-        compareHO (CI.HaveOutput _ _ a) (CI.HaveOutput _ _ b) = compare a b-        compareHO _ _ = error "This situation should have been impossible (mergeC/compareHO)"-        finalizeAll [] = return ()-        finalizeAll (CI.HaveOutput _ f _ : larger) = f >> finalizeAll larger-        finalizeAll (_ :larger) = finalizeAll larger-    in do-        let st = map (($ CI.Done) . CI.unConduitM) cs-        st' <- mapM norm1 st-        go . sortBy compareHO . filter isHO $ st'----- | Take two sorted sources and merge them.------ See 'mergeC'-mergeC2 :: (Ord a, Monad m) => C.Source m a -> C.Source m a -> C.Source m a-mergeC2 (CI.ConduitM s1) (CI.ConduitM s2) = CI.ConduitM $ \rest -> let-        go right@(CI.HaveOutput s1' f1 v1) left@(CI.HaveOutput s2' f2 v2)-            | v1 <= v2 = CI.HaveOutput (go s1' left) (f1 >> f2) v1-            | otherwise = CI.HaveOutput (go right s2') (f1 >> f2) v2-        go right@CI.Done{} (CI.HaveOutput s f v) = CI.HaveOutput (go right s) f v-        go (CI.HaveOutput s f v) left@CI.Done{}  = CI.HaveOutput (go s left)  f v-        go CI.Done{} CI.Done{} = rest ()-        go (CI.PipeM p) left = do-            next <- lift p-            go next left-        go right (CI.PipeM p) = do-            next <- lift p-            go right next-        go (CI.NeedInput _ next) left = go (next ()) left-        go right (CI.NeedInput _ next) = go right (next ())-        go (CI.Leftover next ()) left = go next left-        go right (CI.Leftover next ()) = go right next-    in go (s1 CI.Done) (s2 CI.Done)

@@ -1,213 +0,0 @@-{-|-Module      : Data.Conduit.Algorithms.Async-Copyright   : 2013-2017 Luis Pedro Coelho-License     : MIT-Maintainer  : luis@luispedro.org--Higher level async processing interfaces.--}-{-# LANGUAGE ScopedTypeVariables, FlexibleContexts, CPP #-}--module Data.Conduit.Algorithms.Async-    ( conduitPossiblyCompressedFile-    , asyncMapC-    , asyncMapEitherC-    , asyncGzipTo-    , asyncGzipToFile-    , asyncGzipFrom-    , asyncGzipFromFile-    ) where---import qualified Data.ByteString as B-import qualified Control.Concurrent.Async as A-import qualified Control.Concurrent.STM.TBQueue as TQ-import           Control.Concurrent.STM (atomically)--import qualified Data.Conduit.Combinators as C-import qualified Data.Conduit.Async as CA-import qualified Data.Conduit.TQueue as CA-import qualified Data.Conduit.List as CL-import qualified Data.Conduit.Zlib as CZ-#ifndef WINDOWS--- bzlib cannot compile on Windows (as of 2016/07/05)-import qualified Data.Conduit.BZlib as CZ-#endif-import qualified Data.Conduit as C-import           Data.Conduit ((.|))--import qualified Data.Sequence as Seq-import           Data.Sequence ((|>), ViewL(..))-import           Control.Monad (forM_)-import           Control.Monad.IO.Class (MonadIO, liftIO)-import           Control.Monad.Error.Class (MonadError(..))-import           Control.Monad.Trans.Resource (MonadResource, MonadBaseControl)-import           Control.Exception (evaluate)-import           Control.DeepSeq-import           System.IO-import           Data.List (isSuffixOf)-import           Data.Conduit.Algorithms.Utils (awaitJust)------ | This is like 'Data.Conduit.List.map', except that each element is processed--- in a separate thread (up to 'maxThreads' can be queued up at any one time).--- Results are evaluated to normal form (not weak-head normal form!, i.e., the--- structure is deeply evaluated) to ensure that the computation is fully--- evaluated in the worker thread.------ Note that there is some overhead in threading. It is often a good idea to--- build larger chunks of input before passing it to 'asyncMapC' to amortize--- the costs. That is, when @f@ is not a lot of work, instead of @asyncMapC f@,--- it is sometimes better to do------ @---    CC.conduitVector 4096 .| asyncMapC (V.map f) .| CC.concat--- @------ where @CC@ refers to 'Data.Conduit.Combinators'-asyncMapC :: forall a m b . (MonadIO m, NFData b) =>-                    Int -- ^ Maximum number of worker threads-                    -> (a -> b) -- ^ Function to execute-                    -> C.Conduit a m b-asyncMapC maxThreads f = initLoop (0 :: Int) (Seq.empty :: Seq.Seq (A.Async b))-    where-        initLoop :: Int -> Seq.Seq (A.Async b) -> C.Conduit a m b-        initLoop size q-            | size == maxThreads = loop q-            | otherwise = C.await >>= \case-                Nothing -> yAll q-                Just v -> do-                        v' <- sched v-                        initLoop (size + 1) (q |> v')-        sched :: a -> C.ConduitM a b m (A.Async b)-        sched v = liftIO . A.async . evaluate . force $ f v--        -- | yield all-        yAll :: Seq.Seq (A.Async b) -> C.Conduit a m b-        yAll q = case Seq.viewl q of-            EmptyL -> return ()-            v :< rest -> (liftIO (A.wait v) >>= yieldOrCleanup rest) >> yAll rest--        loop :: Seq.Seq (A.Async b) -> C.Conduit a m b-        loop q = C.await >>= \case-                Nothing -> yAll q-                Just v -> do-                    v' <- sched v-                    case Seq.viewl q of-                        (r :< rest) -> do-                            yieldOrCleanup rest =<< liftIO (A.wait r)-                            loop (rest |> v')-                        _ -> error "should never happen"-        cleanup :: Seq.Seq (A.Async b) -> m ()-        cleanup q = liftIO $ forM_ q A.cancel-        yieldOrCleanup q = flip C.yieldOr (cleanup q)----- | 'asyncMapC' with error handling. The inner function can now return an--- error (as a 'Left'). When the first error is seen, it 'throwError's in the--- main monad. Note that 'f' may be evaluated for arguments beyond the first--- error (as some threads may be running in the background and already--- processing elements after the first error).------ See 'asyncMapC'-asyncMapEitherC :: forall a m b e . (MonadIO m, NFData b, NFData e, MonadError e m) => Int -> (a -> Either e b) -> C.Conduit a m b-asyncMapEitherC maxThreads f = asyncMapC maxThreads f .| (C.awaitForever $ \case-                                Right v -> C.yield v-                                Left err -> throwError err)----- | concatenates input into larger chunks and yields it. Its indended use is--- to build up larger blocks from smaller ones so that they can be sent across--- thread barriers with little overhead.------ the chunkSize parameter is a hint, not an exact element. In particular,--- larger chunks are not split up and smaller chunks can be yielded too.-bsConcatTo :: MonadIO m => Int -- ^ chunk hint-                            -> C.Conduit B.ByteString m [B.ByteString]-bsConcatTo chunkSize = awaitJust start-    where-        start v-            | B.length v >= chunkSize = C.yield [v] >> bsConcatTo chunkSize-            | otherwise = continue [v] (B.length v)-        continue chunks s = C.await >>= \case-            Nothing -> C.yield chunks-            Just v-                | B.length v + s > chunkSize -> C.yield chunks >> start v-                | otherwise -> continue (v:chunks) (s + B.length v)--untilNothing :: forall m i. (Monad m) => C.Conduit (Maybe i) m i-untilNothing = C.await >>= \case-    Just (Just val) -> do-        C.yield val-        untilNothing-    _ -> return ()---- | A simple sink which performs gzip compression in a separate thread and--- writes the results to `h`.------ See also 'asyncGzipToFile'-asyncGzipTo :: forall m. (MonadIO m, MonadBaseControl IO m) => Handle -> C.Sink B.ByteString m ()-asyncGzipTo h = do-    let drain q = liftIO . C.runConduit $-                CA.sourceTBQueue q-                    .| untilNothing-                    .| CL.map (B.concat . reverse)-                    .| CZ.gzip-                    .| C.sinkHandle h-    bsConcatTo ((2 :: Int) ^ (15 :: Int))-        .| CA.drainTo 8 drain---- | Compresses the output and writes to the given file with compression being--- performed in a separate thread.------ See also 'asyncGzipTo'-asyncGzipToFile :: forall m. (MonadResource m, MonadBaseControl IO m) => FilePath -> C.Sink B.ByteString m ()-asyncGzipToFile fname = C.bracketP-    (openFile fname WriteMode)-    hClose-    asyncGzipTo---- | A source which produces the ungzipped content from the the given handle.--- Note that this "reads ahead" so if you do not use all the input, the Handle--- will probably be left at an undefined position in the file.------ See also 'asyncGzipFromFile'-asyncGzipFrom :: forall m. (MonadIO m, MonadResource m, MonadBaseControl IO m) => Handle -> C.Source m B.ByteString-asyncGzipFrom h = do-    let prod q = liftIO $ do-                    C.runConduit $-                        C.sourceHandle h-                            .| CZ.multiple CZ.ungzip-                            .| CL.map Just-                            .| CA.sinkTBQueue q-                    atomically (TQ.writeTBQueue q Nothing)-    CA.gatherFrom 8 prod-        .| untilNothing---- | Open and read a gzip file with the uncompression being performed in a--- separate thread.------ See also 'asyncGzipFrom'-asyncGzipFromFile :: forall m. (MonadResource m, MonadBaseControl IO m) => FilePath -> C.Source m B.ByteString-asyncGzipFromFile fname = C.bracketP-    (openFile fname ReadMode)-    hClose-    asyncGzipFrom---- | If the filename indicates a gzipped file (or, on Unix, also a bz2 file),--- then it reads it and uncompresses it.------ On Windows, attempting to read from a bzip2 file, results in 'error'.------ For the case of gzip, 'asyncGzipFromFile' is used.-conduitPossiblyCompressedFile :: (MonadBaseControl IO m, MonadResource m) => FilePath -> C.Source m B.ByteString-conduitPossiblyCompressedFile fname-    | ".gz" `isSuffixOf` fname = asyncGzipFromFile fname-#ifndef WINDOWS-    | ".bz2" `isSuffixOf` fname = C.sourceFile fname .| CZ.bunzip2-#else-    | ".bz2" `isSuffixOf` fname = error "bzip2 decompression is not available on Windows"-#endif-    | otherwise = C.sourceFile fname-

@@ -1,4 +1,4 @@-{- Copyright 2017 Luis Pedro Coelho+{- Copyright 2017-2018 Luis Pedro Coelho  - License: MIT  -} {-# LANGUAGE TemplateHaskell, QuasiQuotes, FlexibleContexts, OverloadedStrings #-}@@ -14,14 +14,17 @@ import qualified Data.Conduit.Combinators as CC import qualified Data.Conduit.Binary as CB import qualified Data.Conduit.List as CL+import qualified Data.Vector.Storable as VS import           Data.Conduit ((.|)) import           Data.List (sort) import           System.Directory (removeFile) import           Control.Monad (forM_)  import qualified Data.Conduit.Algorithms as CAlg+import qualified Data.Conduit.Algorithms.Storable as CAlg import qualified Data.Conduit.Algorithms.Utils as CAlg import qualified Data.Conduit.Algorithms.Async as CAlg+import qualified Data.Conduit.Algorithms.Async.ByteString as CAlg  main :: IO () main = $(defaultMainGenerator)@@ -103,6 +106,11 @@     vals <- extractIO (CC.yieldMany [0..10] .| CAlg.asyncMapC 3 (+ (1:: Int)))     (vals @?= [1..11]) +case_unorderedAsyncMapC :: IO ()+case_unorderedAsyncMapC = do+    vals <- extractIO (CC.yieldMany [0..10] .| CAlg.unorderedAsyncMapC 3 (+ (1:: Int)))+    (sort vals @?= [1..11])+ case_asyncGzip :: IO () case_asyncGzip = do     C.runConduitRes (CC.yieldMany ["Hello", " ", "World"] .| CAlg.asyncGzipToFile testingFileNameGZ)@@ -126,3 +134,16 @@             .| CL.map (read . B8.unpack)     removeFile testingFileNameGZ     removeFile testingFileNameGZ2++case_asyncFilterLines = do+    vals <- extractIO (CC.yieldMany ["This is\nMy data\nBut"," sometimes","\nit is split,\n","in weird ways."] .| CAlg.asyncFilterLinesC 2 (B8.notElem ','))+    (vals @?= ["This is", "My data", "But sometimes", "in weird ways."])++case_asyncFilterLinesAllTrue = do+    vals <- extractIO (CC.yieldMany ["This is\nMy data\nBut"," sometimes","\nit is split,\n","in weird ways."] .| CAlg.asyncFilterLinesC 2 (const True))+    (vals @?= ["This is", "My data", "But sometimes", "it is split,", "in weird ways."])++case_storableVector = do+    let v = VS.fromList [0:: Int, 1, 2, 4, 6, 12]+    vals <- extractIO (CC.yieldMany [v,v,v] .| CAlg.writeStorableV .| CAlg.readStorableV 3)+    (VS.concat vals @=? VS.concat [v,v,v])

@@ -1,61 +0,0 @@-{-|-Module      : Data.Conduit.Algorithms.Utils-Copyright   : 2013-2017 Luis Pedro Coelho-License     : MIT-Maintainer  : luis@luispedro.org--A few miscellaneous conduit utils--}-module Data.Conduit.Algorithms.Utils-    ( awaitJust-    , enumerateC-    , groupC-    ) where--import qualified Data.Conduit as C-import           Data.Maybe (maybe)-import           Control.Monad (unless)---- | Act on the next input (do nothing if no input). @awaitJust f@ is equivalent to--------- @ do---      next <- C.await---      case next of---          Just val -> f val---          Nothing -> return ()--- @------ This is a simple utility adapted from--- http://neilmitchell.blogspot.de/2015/07/thoughts-on-conduits.html-awaitJust :: Monad m => (a -> C.Conduit a m b) -> C.Conduit a m b-awaitJust f = C.await >>= maybe (return ()) f---- Conduit analogue to Python's enumerate function-enumerateC :: Monad m => C.Conduit a m (Int, a)-enumerateC = enumerateC' 0-    where-        enumerateC' !i = awaitJust $ \v -> do-                                        C.yield (i, v)-                                        enumerateC' (i + 1)---- | groupC yields the input as groups of 'n' elements. If the input is not a--- multiple of 'n', the last element will be incomplete------ Example:------ @---      CC.yieldMany [0..10] .| groupC 3 .| CC.consumeList--- @------ results in @[ [0,1,2], [3,4,5], [6,7,8], [9, 10] ]@------ This function is deprecated; use 'Data.Conduit.List.chunksOf'-groupC :: (Monad m) => Int -> C.Conduit a m [a]-groupC n = loop n []-    where-        loop 0 ps = C.yield (reverse ps) >> loop n []-        loop c ps = C.await >>= \case-            Nothing -> unless (null ps) $ C.yield (reverse ps)-            Just p -> loop (c-1) (p:ps)-

@@ -6,7 +6,6 @@ [![Travis](https://api.travis-ci.org/luispedro/conduit-algorithms.png)](https://travis-ci.org/luispedro/conduit-algorithms) Some conduit-based algorithms. - Much of this code was originally part of [NGLess](http://ngless.embl.de) and has been in production use for years. However, it can be of generic use. 

@@ -1,72 +1,78 @@-name:               conduit-algorithms-version:            0.0.6.1-synopsis:           Conduit-based algorithms-description:        Algorithms on Conduits, including higher level asynchronous-                    processing and some other utilities.-category:           Conduit-author:             Luis Pedro Coelho-maintainer:         Luis Pedro Coelho-license:            MIT-license-file:       COPYING-cabal-version:      >= 1.10-build-type:         Simple-bug-reports:        https://github.com/luispedro/conduit-algorithms/issues-extra-source-files: README.md ChangeLog+-- This file has been generated from package.yaml by hpack version 0.18.1.+--+-- see: https://github.com/sol/hpack +name:           conduit-algorithms+version:        0.0.7.0+synopsis:       Conduit-based algorithms+description:    Algorithms on Conduits, including higher level asynchronous processing and some other utilities.+category:       Conduit+homepage:       https://github.com/luispedro/conduit-algorithms#readme+bug-reports:    https://github.com/luispedro/conduit-algorithms/issues+author:         Luis Pedro Coelho+maintainer:     Luis Pedro Coelho+license:        MIT+license-file:   COPYING+build-type:     Simple+cabal-version:  >= 1.10++extra-source-files:+    ChangeLog+    README.md++source-repository head+  type: git+  location: https://github.com/luispedro/conduit-algorithms+ library-  default-language: Haskell2010-  default-extensions:  BangPatterns, OverloadedStrings, LambdaCase, TupleSections-  exposed-modules: Data.Conduit.Algorithms-                   Data.Conduit.Algorithms.Utils-                   Data.Conduit.Algorithms.Async+  default-extensions: BangPatterns OverloadedStrings LambdaCase TupleSections   ghc-options: -Wall   build-depends:-    base > 4.8 && < 5,-    async,-    bytestring,-    bzlib-conduit,-    conduit >= 1.0,-    conduit-combinators,-    conduit-extra,-    containers,-    deepseq,-    mtl,-    resourcet,-    stm,-    stm-conduit >= 2.7,-    transformers--Test-Suite catest+      base > 4.8 && < 5+    , async+    , bytestring+    , bzlib-conduit+    , conduit >= 1.0+    , conduit-combinators >= 1.1.2+    , conduit-extra+    , containers+    , deepseq+    , lzma-conduit+    , mtl+    , resourcet+    , stm+    , stm-conduit >= 2.7+    , vector+    , transformers+  other-modules:+      Paths_conduit_algorithms   default-language: Haskell2010-  default-extensions:  BangPatterns, OverloadedStrings, LambdaCase, TupleSections++test-suite conduit-algorithms-test   type: exitcode-stdio-1.0   main-is: Data/Conduit/Algorithms/Tests.hs-  other-modules: Data.Conduit.Algorithms-                 Data.Conduit.Algorithms.Utils-                 Data.Conduit.Algorithms.Async+  default-extensions: BangPatterns OverloadedStrings LambdaCase TupleSections   ghc-options: -Wall-  hs-source-dirs: .   build-depends:-    base > 4.8 && < 5,-    async,-    bytestring,-    bzlib-conduit,-    conduit >= 1.0,-    conduit-combinators,-    conduit-extra,-    containers,-    directory,-    deepseq,-    mtl,-    resourcet,-    stm,-    stm-conduit >= 2.7,-    transformers,-    HUnit,-    test-framework,-    test-framework-hunit,-    test-framework-th--source-repository head-  type: git-  location: https://github.com/luispedro/conduit-algorithms+      base > 4.8 && < 5+    , async+    , bytestring+    , bzlib-conduit+    , conduit >= 1.0+    , conduit-combinators >= 1.1.2+    , conduit-extra+    , containers+    , deepseq+    , lzma-conduit+    , mtl+    , resourcet+    , stm+    , stm-conduit >= 2.7+    , vector+    , transformers+    , directory+    , HUnit+    , test-framework+    , test-framework-hunit+    , test-framework-th+  default-language: Haskell2010