packages feed

conduit-algorithms (empty) → 0.0.1.0

raw patch · 9 files changed

+455/−0 lines, 9 filesdep +HUnitdep +asyncdep +basesetup-changed

Dependencies added: HUnit, async, base, bytestring, bzlib-conduit, conduit, conduit-combinators, conduit-extra, containers, deepseq, directory, filepath, mtl, resourcet, stm, stm-chans, stm-conduit, test-framework, test-framework-hunit, test-framework-th, transformers, unix

Files

+ COPYING view
@@ -0,0 +1,23 @@+Copyright (c) 2017+Luis Pedro Coelho <luis@luispedro.org>++Permission is hereby granted, free of charge, to any person+obtaining a copy of this software and associated documentation+files (the "Software"), to deal in the Software without+restriction, including without limitation the rights to use,+copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the+Software is furnished to do so, subject to the following+conditions:++The above copyright notice and this permission notice shall be+included in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES+OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT+HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,+WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING+FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR+OTHER DEALINGS IN THE SOFTWARE.
+ ChangeLog view
@@ -0,0 +1,3 @@+Version 0.0.1.0 2017-08-07 by luispedro+	* First version. Generic code from NGLess and internal projects, extracted+	for wider convenience
+ Data/Conduit/Algorithms.hs view
@@ -0,0 +1,77 @@+{-# LANGUAGE Rank2Types #-}+module Data.Conduit.Algorithms+    ( uniqueOnC+    , uniqueC+    , mergeC+    ) where++import qualified Data.Conduit as C+import qualified Data.Conduit.Combinators as CC+import qualified Data.Set as S+import           Control.Monad.Trans.Class (lift)++import           Data.Conduit.Algorithms.Utils (awaitJust)+++-- | Unique conduit.+--+-- 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).+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)+-- | See 'uniqueOnC'+uniqueC :: (Ord a, Monad m) => C.Conduit a m a+uniqueC = uniqueOnC id++-- | Merge a list of sorted sources+--+-- See 'mergeC2'+mergeC :: (Ord a, Monad m) => [C.Source m a] -> C.Source m a+mergeC [] = return ()+mergeC [s] = s+mergeC [a,b] = mergeC2 a b+mergeC args = let (a,b) = split2 args in mergeC2 (mergeC a) (mergeC b)+    where+        split2 :: [a] -> ([a],[a])+        split2 [] = ([], [])+        split2 [a] = ([a], [])+        split2 [a,b] = ([a], [b])+        split2 (x:y:rs) = let (xs,ys) = split2 rs in (x:xs, y:ys)++-- | 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 s1 s2 = do+        (c1', e1) <- lift $ s1 C.$$+ CC.head+        (c2', e2) <- lift $ s2 C.$$+ CC.head+        continue c1' c2' e1 e2+    where+        continue :: (Monad m, Ord a) => C.ResumableSource m a -> C.ResumableSource m a -> Maybe a -> Maybe a -> C.Source m a+        continue _ _ Nothing Nothing = return ()+        continue _ c Nothing e = continue c undefined e Nothing+        continue c _ (Just e) Nothing = do+            C.yield e+            yieldAll c+        continue c1 c2 je1@(Just e1) je2@(Just e2)+            | compare e1 e2 == GT = continue c2 c1 je2 je1+            | otherwise = do+                C.yield e1+                (c1', e1') <- lift $ c1 C.$$++ CC.head+                continue c1' c2 e1' (Just e2)+        yieldAll :: (Monad m) => C.ResumableSource m a -> C.Source m a+        yieldAll rs = do+            (rs',v) <- lift $ rs C.$$++ CC.head+            case v of+                Just v' -> do+                    C.yield v'+                    yieldAll rs'+                Nothing -> return ()
+ Data/Conduit/Algorithms/Async.hs view
@@ -0,0 +1,188 @@+{- Copyright 2013-2017 Luis Pedro Coelho+ - License: MIT -}+{-# LANGUAGE ScopedTypeVariables, FlexibleContexts, CPP #-}++module Data.Conduit.Algorithms.Async+    ( conduitPossiblyCompressedFile+    , asyncMapC+    , asyncMapEitherC+    , asyncGzipTo+    , asyncGzipToFile+    , asyncGzipFrom+    , asyncGzipFromFile+    ) where++-- | Higher level async processing interfaces++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 maxSize can be queued up at any one time).+-- Results are evaluated to normal form (not WHNF!) to ensure that the+-- computation is fully evaluated before being yielded to the next conduit.+asyncMapC :: forall a m b . (MonadIO m, NFData b) => Int -> (a -> b) -> C.Conduit a m b+asyncMapC maxSize 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 == maxSize = 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).+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 maxSize f = asyncMapC maxSize 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 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.+--+-- 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+
+ Data/Conduit/Algorithms/Tests.hs view
@@ -0,0 +1,46 @@+{- Copyright 2017 Luis Pedro Coelho+ - License: MIT+ -}+{-# LANGUAGE TemplateHaskell, QuasiQuotes, FlexibleContexts, OverloadedStrings #-}+module Main where++import Test.Framework.TH+import Test.HUnit+import Test.Framework.Providers.HUnit++import qualified Data.ByteString as B+import qualified Data.Conduit as C+import qualified Data.Conduit.Combinators as CC+import           Data.Conduit ((.|))++import qualified Data.Conduit.Algorithms as CAlg+import qualified Data.Conduit.Algorithms.Utils as CAlg+import qualified Data.Conduit.Algorithms.Async as CAlg+import           System.Directory (removeFile)++main :: IO ()+main = $(defaultMainGenerator)++testingFileNameGZ :: FilePath+testingFileNameGZ = "file_just_for_testing_delete_me_please.gz"++extract c = C.runConduitPure (c .| CC.sinkList)++extractIO c = C.runConduitRes (c .| CC.sinkList)++case_uniqueC = extract (CC.yieldMany [1,2,3,1,1,2,3] .| CAlg.uniqueC) @=? [1,2,3 :: Int]+case_mergeC = extract (CAlg.mergeC [CC.yieldMany [0, 2, 4], CC.yieldMany [1,3,4,5]]) @=? [0,1,2,3,4,4,5 :: Int]+case_groupC = extract (CC.yieldMany [0..10] .| CAlg.groupC 3) @=? [[0,1,2], [3,4,5], [6,7,8], [9, 10 :: Int]]++case_asyncMap :: IO ()+case_asyncMap = do+    vals <- extractIO (CC.yieldMany [0..10] .| CAlg.asyncMapC 3 (+ (1:: Int)))+    (vals @=? [1..11])++case_asyncGzip :: IO ()+case_asyncGzip = do+    C.runConduitRes (CC.yieldMany ["Hello", " ", "World"] .| CAlg.asyncGzipToFile testingFileNameGZ)+    r <- B.concat <$> (extractIO (CAlg.asyncGzipFromFile testingFileNameGZ))+    r @=? "Hello World"+    removeFile testingFileNameGZ+
+ Data/Conduit/Algorithms/Utils.hs view
@@ -0,0 +1,26 @@+{- Copyright 2013-2017 Luis Pedro Coelho+ - License: MIT -}+module Data.Conduit.Algorithms.Utils+    ( awaitJust+    , groupC+    ) where++import qualified Data.Conduit as C+import           Data.Maybe (maybe)+import           Control.Monad (unless)++-- | 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++-- | groupC yields the input as groups of 'n' elements. If the input is not a+-- multiple of 'n', the last element will be incomplete+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)+
+ README.md view
@@ -0,0 +1,11 @@+# 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.++License: MIT++## Author+Luis Pedro Coelho | [Email](mailto:luis@luispedro.org) | [Twitter](https://twitter.com/luispedrocoelho)
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ conduit-algorithms.cabal view
@@ -0,0 +1,79 @@+name:               conduit-algorithms+version:            0.0.1.0+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++library+  default-language: Haskell2010+  default-extensions:  BangPatterns, OverloadedStrings, LambdaCase, TupleSections+  exposed-modules: Data.Conduit.Algorithms+                   Data.Conduit.Algorithms.Utils+                   Data.Conduit.Algorithms.Async+  ghc-options: -Wall+  build-depends:+    base > 4.8 && < 5,+    async,+    bytestring,+    bzlib-conduit,+    conduit >= 1.0,+    conduit-combinators,+    conduit-extra,+    containers,+    deepseq,+    directory,+    filepath,+    mtl,+    resourcet,+    stm,+    stm-chans,+    stm-conduit >= 2.7,+    transformers,+    unix++Test-Suite catest+  default-language: Haskell2010+  default-extensions:  BangPatterns, OverloadedStrings, LambdaCase, TupleSections+  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+  ghc-options: -Wall+  hs-source-dirs: .+  build-depends:+    base > 4.8 && < 5,+    async,+    bytestring,+    bzlib-conduit,+    conduit >= 1.0,+    conduit-combinators,+    conduit-extra,+    containers,+    deepseq,+    directory,+    filepath,+    mtl,+    resourcet,+    stm,+    stm-chans,+    stm-conduit >= 2.7,+    transformers,+    unix,+    HUnit,+    test-framework,+    test-framework-hunit,+    test-framework-th++source-repository head+  type: git+  location: https://github.com/luispedro/conduit-algorithms