potoki 0.7 → 0.7.2
raw patch · 5 files changed
+115/−78 lines, 5 files
Files
- library/Potoki/Consume.hs +10/−0
- library/Potoki/Transform.hs +14/−77
- library/Potoki/Transform/Concurrency.hs +83/−0
- potoki.cabal +2/−1
- tests/Main.hs +6/−0
library/Potoki/Consume.hs view
@@ -18,6 +18,7 @@ printString, parseBytes, parseText,+ concurrently, ) where @@ -34,6 +35,7 @@ import qualified Data.Text.IO as K import qualified Control.Foldl as D import qualified System.Directory as G+import qualified Potoki.Transform.Concurrency as B {-# INLINABLE transform #-}@@ -177,3 +179,11 @@ parseText :: F.Parser output -> Consume Text (Either Text output) parseText = runParseResult . F.parse++{-|+Execute a Consume concurrently and consume its results.+-}+{-# INLINABLE concurrently #-}+concurrently :: Int -> Consume a b -> Consume b c -> Consume a c+concurrently amount consume1 consume2 =+ transform (B.concurrently amount (J.consume consume1)) consume2
library/Potoki/Transform.hs view
@@ -11,6 +11,7 @@ mapFilter, filter, just,+ distinctBy, distinct, builderChunks, executeIO,@@ -19,8 +20,8 @@ parseBytes, parseText, -- * Concurrency- bufferize,- concurrently,+ N.bufferize,+ N.concurrently, -- * File IO deleteFile, appendBytesToFile,@@ -42,6 +43,7 @@ import qualified Data.ByteString as J import qualified System.Directory as I import qualified Control.Concurrent.Chan.Unagi.Bounded as B+import qualified Potoki.Transform.Concurrency as N {-# INLINE mapFilter #-}@@ -64,76 +66,6 @@ takeWhile predicate = Transform (pure . A.takeWhile predicate) -{-# INLINE bufferize #-}-bufferize :: Int -> Transform element element-bufferize size =- Transform $ \ (A.Fetch fetch) -> do- (inChan, outChan) <- B.newChan size- forkIO $ fix $ \ loop ->- join $ fetch- (B.writeChan inChan Nothing)- (\ !element -> B.writeChan inChan (Just element) >> loop)- return $ A.Fetch $ \ nil just -> fmap (maybe nil just) (B.readChan outChan)--{-|-Identity Transform, which ensures that the inputs are fetched synchronously.--Useful for concurrent transforms.--}-{-# INLINABLE sync #-}-sync :: Transform a a-sync =- Transform $ \ (A.Fetch fetch) -> do- activeVar <- newMVar True- return $ A.Fetch $ \ nil just -> do- active <- takeMVar activeVar- if active- then join $ fetch- (do- putMVar activeVar False- return nil)- (\ !element -> do- putMVar activeVar True- return (just element))- else do- putMVar activeVar False- return nil--{-|-Execute the transform on the specified amount of threads.-The order of the outputs produced is indiscriminate.--}-{-# INLINABLE concurrently #-}-concurrently :: Int -> Transform input output -> Transform input output-concurrently workersAmount transform =- sync >>>- concurrentlyUnsafe workersAmount transform--{-# INLINE concurrentlyUnsafe #-}-concurrentlyUnsafe :: Int -> Transform input output -> Transform input output-concurrentlyUnsafe workersAmount (Transform syncTransformIO) =- Transform $ \ fetch -> do- outChan <- newEmptyMVar- replicateM_ workersAmount $ forkIO $ do- A.Fetch fetchIO <- syncTransformIO fetch- fix $ \ loop -> join $ fetchIO- (putMVar outChan Nothing)- (\ !result -> putMVar outChan (Just result) >> loop)- activeWorkersAmountVar <- newMVar workersAmount- return $ A.Fetch $ \ nil just -> fix $ \ loop -> do- activeWorkersAmount <- takeMVar activeWorkersAmountVar- if activeWorkersAmount <= 0- then return nil- else do- fetchResult <- takeMVar outChan- case fetchResult of- Just result -> do- putMVar activeWorkersAmountVar activeWorkersAmount- return (just result)- Nothing -> do- putMVar activeWorkersAmountVar (pred activeWorkersAmount)- loop- {-# INLINE mapWithParseResult #-} mapWithParseResult :: forall input parsed. (Monoid input, Eq input) => (input -> M.IResult input parsed) -> Transform input (Either Text parsed) mapWithParseResult inputToResult =@@ -226,18 +158,23 @@ withFile path AppendMode $ \ handle -> J.hPut handle bytes -{-# INLINE distinct #-}-distinct :: (Eq element, Hashable element) => Transform element element-distinct =+{-# INLINE distinctBy #-}+distinctBy :: (Eq comparable, Hashable comparable) => (element -> comparable) -> Transform element element+distinctBy f = Transform $ \ (A.Fetch fetch) -> do stateRef <- newIORef mempty return $ A.Fetch $ \ nil just -> fix $ \ loop -> join $ fetch (return nil) $ \ !input -> do+ let comparable = f input !set <- readIORef stateRef- if C.member input set+ if C.member comparable set then loop else do- writeIORef stateRef $! C.insert input set+ writeIORef stateRef $! C.insert comparable set return (just input)++{-# INLINE distinct #-}+distinct :: (Eq element, Hashable element) => Transform element element+distinct = distinctBy id {-# INLINE builderChunks #-} builderChunks :: Transform E.Builder ByteString
+ library/Potoki/Transform/Concurrency.hs view
@@ -0,0 +1,83 @@+module Potoki.Transform.Concurrency+(+ bufferize,+ concurrently,+)+where++import Potoki.Prelude hiding (take, takeWhile, filter)+import Potoki.Core.Transform+import qualified Potoki.Fetch as A+import qualified Potoki.Core.Fetch as A+import qualified Control.Concurrent.Chan.Unagi.Bounded as B+++{-# INLINE bufferize #-}+bufferize :: Int -> Transform element element+bufferize size =+ Transform $ \ (A.Fetch fetch) -> do+ (inChan, outChan) <- B.newChan size+ forkIO $ fix $ \ loop ->+ join $ fetch+ (B.writeChan inChan Nothing)+ (\ !element -> B.writeChan inChan (Just element) >> loop)+ return $ A.Fetch $ \ nil just -> fmap (maybe nil just) (B.readChan outChan)++{-|+Identity Transform, which ensures that the inputs are fetched synchronously.++Useful for concurrent transforms.+-}+{-# INLINABLE sync #-}+sync :: Transform a a+sync =+ Transform $ \ (A.Fetch fetch) -> do+ activeVar <- newMVar True+ return $ A.Fetch $ \ nil just -> do+ active <- takeMVar activeVar+ if active+ then join $ fetch+ (do+ putMVar activeVar False+ return nil)+ (\ !element -> do+ putMVar activeVar True+ return (just element))+ else do+ putMVar activeVar False+ return nil++{-|+Execute the transform on the specified amount of threads.+The order of the outputs produced is indiscriminate.+-}+{-# INLINABLE concurrently #-}+concurrently :: Int -> Transform input output -> Transform input output+concurrently workersAmount transform =+ sync >>>+ concurrentlyUnsafe workersAmount transform++{-# INLINE concurrentlyUnsafe #-}+concurrentlyUnsafe :: Int -> Transform input output -> Transform input output+concurrentlyUnsafe workersAmount (Transform syncTransformIO) =+ Transform $ \ fetch -> do+ outChan <- newEmptyMVar+ replicateM_ workersAmount $ forkIO $ do+ A.Fetch fetchIO <- syncTransformIO fetch+ fix $ \ loop -> join $ fetchIO+ (putMVar outChan Nothing)+ (\ !result -> putMVar outChan (Just result) >> loop)+ activeWorkersAmountVar <- newMVar workersAmount+ return $ A.Fetch $ \ nil just -> fix $ \ loop -> do+ activeWorkersAmount <- takeMVar activeWorkersAmountVar+ if activeWorkersAmount <= 0+ then return nil+ else do+ fetchResult <- takeMVar outChan+ case fetchResult of+ Just result -> do+ putMVar activeWorkersAmountVar activeWorkersAmount+ return (just result)+ Nothing -> do+ putMVar activeWorkersAmountVar (pred activeWorkersAmount)+ loop
potoki.cabal view
@@ -1,7 +1,7 @@ name: potoki version:- 0.7+ 0.7.2 synopsis: Simple streaming in IO description:@@ -72,6 +72,7 @@ other-modules: Potoki.Fetch Potoki.Prelude+ Potoki.Transform.Concurrency build-depends: attoparsec >=0.13 && <0.15, base >=4.7 && <5,
tests/Main.hs view
@@ -74,6 +74,12 @@ result <- C.produceAndConsume (E.list list) (D.transform A.distinct D.list) assertEqual "" [1,2,3,4] result ,+ testCase "Distinct By" $ do+ let+ list = [(1, ""),(2, ""),(3, ""),(2, ""),(3, ""),(2, ""),(1, ""),(4, ""),(1, "")] :: [(Int, String)]+ result <- C.produceAndConsume (E.list list) (D.transform (A.distinctBy fst) D.list)+ assertEqual "" [(1, ""),(2, ""),(3, ""),(4, "")] result+ , testCase "Concurrently" $ do let list = [1..20000]