potoki-core 2.1.0.1 → 2.3.4.1
raw patch · 24 files changed
Files
- Setup.hs +0/−2
- benchmark/Main.hs +25/−0
- library/Potoki/Core/Consume.hs +44/−11
- library/Potoki/Core/Fetch.hs +57/−29
- library/Potoki/Core/IO.hs +4/−3
- library/Potoki/Core/IO/Fetch.hs +7/−5
- library/Potoki/Core/Prelude.hs +31/−2
- library/Potoki/Core/Produce.hs +123/−5
- library/Potoki/Core/TextBuilder.hs +20/−0
- library/Potoki/Core/Transform.hs +21/−1
- library/Potoki/Core/Transform/Attoparsec.hs +35/−11
- library/Potoki/Core/Transform/Basic.hs +201/−79
- library/Potoki/Core/Transform/ByteString.hs +96/−15
- library/Potoki/Core/Transform/Concurrency.hs +174/−48
- library/Potoki/Core/Transform/Instances.hs +10/−10
- library/Potoki/Core/Transform/Scanner.hs +54/−0
- library/Potoki/Core/Transform/State.hs +4/−4
- potoki-core.cabal +62/−63
- test/Main.hs +264/−0
- test/Potoki.hs +309/−0
- test/Transform.hs +215/−0
- tests/Main.hs +0/−221
- tests/Potoki.hs +0/−279
- tests/Transform.hs +0/−178
− Setup.hs
@@ -1,2 +0,0 @@-import Distribution.Simple-main = defaultMain
+ benchmark/Main.hs view
@@ -0,0 +1,25 @@+module Main where++import Prelude+import Criterion.Main+import Potoki.Core.IO+import qualified Potoki.Core.Produce as Produce+import qualified Potoki.Core.Consume as Consume+import qualified Potoki.Core.Transform as Transform+++main =+ defaultMain $+ [+ bench "extractLinesConcurrently" $ whnfIO $ produceAndConsume+ (Produce.fileBytes "data/2.tsv")+ (right' (Consume.transform+ (Transform.extractLinesConcurrently numCapabilities)+ (Consume.count)))+ ,+ bench "extractLines" $ whnfIO $ produceAndConsume+ (Produce.fileBytes "data/2.tsv")+ (right' (Consume.transform+ (Transform.extractLines)+ (Consume.count)))+ ]
library/Potoki/Core/Consume.hs view
@@ -16,7 +16,9 @@ folding, foldingInIO, execState,+ writeBytesToStdout, writeBytesToFile,+ writeBytesToFileWithoutBuffering, appendBytesToFile, deleteFiles, printBytes,@@ -99,6 +101,10 @@ rightOutput <- takeMVar rightOutputVar return (leftOutput rightOutput) +unit :: Consume a ()+unit =+ Consume $ \ _ -> return ()+ {-# INLINABLE list #-} list :: Consume input [input] list =@@ -160,15 +166,16 @@ {-# INLINABLE count #-} count :: Consume input Int count =- Consume $ \ (A.Fetch fetchIO) -> build fetchIO 0- where- build fetchIO !acc =- fetchIO >>= \case- Nothing -> pure acc- Just _ -> build fetchIO (succ acc)+ Consume $ \ (A.Fetch fetchIO) -> let+ iterate !count = do+ fetchResult <- fetchIO+ case fetchResult of+ Just _ -> iterate (succ count)+ Nothing -> return count+ in iterate 0 {-# INLINABLE concat #-}-concat :: Monoid monoid => Consume monoid monoid+concat :: (Semigroup monoid, Monoid monoid) => Consume monoid monoid concat = Consume $ \ (A.Fetch fetchIO) -> build fetchIO mempty where@@ -197,6 +204,11 @@ printString = processInIO (putChar '\n') putStr +{-# INLINABLE writeBytesToStdout #-}+writeBytesToStdout :: Consume ByteString ()+writeBytesToStdout =+ processInIO (return ()) (C.hPut stdout)+ {-| Overwrite a file. @@ -205,13 +217,34 @@ -} {-# INLINABLE writeBytesToFile #-} writeBytesToFile :: FilePath -> Consume ByteString (Either IOException ())-writeBytesToFile path =+writeBytesToFile =+ writeBytesToFileWithBuffering (BlockBuffering Nothing)++{-|+Overwrite a file.++* Exception-free+* Automatic resource management+-}+{-# INLINABLE writeBytesToFileWithBuffering #-}+writeBytesToFileWithBuffering :: BufferMode -> FilePath -> Consume ByteString (Either IOException ())+writeBytesToFileWithBuffering bufferMode path = Consume $ \ fetch ->- try $ withFile path WriteMode $ \ handleVal ->+ try $ withFile path WriteMode $ \ handle -> do- L.fetchAndHandleAll fetch (return ()) (C.hPut handleVal)+ hSetBuffering handle bufferMode+ L.fetchAndHandleAll fetch (return ()) (C.hPut handle) {-|+A more efficient implementation than just writing to file without buffering.+It uses an explicit buffer of input chunks and flushes all the chunks that have been so far aggregated at once.+-}+{-# INLINABLE writeBytesToFileWithoutBuffering #-}+writeBytesToFileWithoutBuffering :: FilePath -> Consume ByteString (Either IOException ())+writeBytesToFileWithoutBuffering =+ transform (arr mconcat . B.bufferizeFlushing 64) . writeBytesToFileWithBuffering NoBuffering++{-| Append to a file. * Exception-free@@ -315,7 +348,7 @@ Execute a Consume concurrently and consume its results. -} {-# INLINABLE concurrently #-}-concurrently :: Int -> Consume a b -> Consume b c -> Consume a c+concurrently :: NFData b => Int -> Consume a b -> Consume b c -> Consume a c concurrently amount consume1 consume2 = transform (B.concurrently amount (J.consume consume1)) consume2
library/Potoki/Core/Fetch.hs view
@@ -22,6 +22,8 @@ finiteMVar, vector, handlingElements,+ lazyByteStringRef,+ enumUntil, ) where @@ -30,7 +32,8 @@ import qualified Data.Vector as C import qualified Data.ByteString as D import qualified Data.Text.IO as A-+import qualified Data.ByteString.Lazy as B+import qualified Data.ByteString.Lazy.Internal as B deriving instance Functor Fetch @@ -116,7 +119,7 @@ (!headVal) : (!tailVal) -> do writeIORef unsentListRef tailVal return $ Just headVal- _ -> do+ _ -> do writeIORef unsentListRef [] return Nothing @@ -126,7 +129,7 @@ Fetch $ do bothFetch <- bothFetchIO case bothFetch of- Nothing -> return Nothing+ Nothing -> return Nothing Just (!firstVal, !secondVal) -> do writeIORef cacheRef secondVal return $ Just firstVal@@ -137,7 +140,7 @@ Fetch $ do firstFetch <- firstFetchIO case firstFetch of- Nothing -> return Nothing+ Nothing -> return Nothing Just !firstFetched -> do secondCached <- readIORef cacheRef return $ Just (firstFetched, secondCached)@@ -148,7 +151,7 @@ Fetch $ do eitherFetch <- eitherFetchIO case eitherFetch of- Nothing -> return Nothing+ Nothing -> return Nothing Just input -> case input of Right rightInput -> return $ Just rightInput Left leftInput -> left2IO leftInput $> Nothing@@ -211,36 +214,40 @@ {-# INLINABLE mapFilter #-} mapFilter :: (input -> Maybe output) -> Fetch input -> Fetch output mapFilter mapping (Fetch fetchIO) =- Fetch $ - fix $ \ doLoop -> do - fetch <- fetchIO- case mapping <$> fetch of- Just (Just output) -> return (Just output)- Just Nothing -> doLoop- Nothing -> return Nothing+ {-# SCC "mapFilter" #-} + Fetch $ let+ loop = do + fetch <- fetchIO+ case mapping <$> fetch of+ Just (Just !output) -> return (Just output)+ Just Nothing -> loop+ Nothing -> return Nothing+ in loop {-# INLINABLE filter #-} filter :: (input -> Bool) -> Fetch input -> Fetch input filter predicate (Fetch fetchIO) =- Fetch $ - fix $ \ doLoop -> do - fetch <- fetchIO- case predicate <$> fetch of- Just True -> return fetch- Just False -> doLoop- Nothing -> return Nothing-+ Fetch $ let+ loop = do + fetch <- fetchIO+ case predicate <$> fetch of+ Just True -> return fetch+ Just False -> loop+ Nothing -> return Nothing+ in loop {-# INLINABLE just #-} just :: Fetch (Maybe element) -> Fetch element just (Fetch fetchIO) =- Fetch $ - fix $ \ doLoop -> do - fetch <- fetchIO- case fetch of- Just (Just element) -> return (Just element)- Just (Nothing) -> doLoop- Nothing -> return Nothing+ {-# SCC "just" #-} + Fetch $ let+ loop = do+ fetch <- fetchIO+ case fetch of+ Just (Just !element) -> return (Just element)+ Just (Nothing) -> loop+ Nothing -> return Nothing+ in loop {-# INLINABLE takeWhile #-} takeWhile :: (element -> Bool) -> Fetch element -> Fetch element@@ -278,6 +285,27 @@ Fetch $ do mbElement <- fetchIO case mbElement of- Just element -> xRay element $> mbElement+ Just !element -> xRay element $> mbElement Nothing -> return Nothing- ++{-# INLINE lazyByteStringRef #-}+lazyByteStringRef :: IORef B.ByteString -> Fetch ByteString+lazyByteStringRef ref =+ Fetch $ do+ lazyByteString <- readIORef ref+ case lazyByteString of+ B.Chunk chunk remainders -> do+ writeIORef ref remainders+ return (Just chunk)+ B.Empty -> return Nothing++{-# INLINE enumUntil #-}+enumUntil :: (Enum a, Ord a) => IORef a -> a -> Fetch a+enumUntil ref maxIndex =+ Fetch $ do+ index <- readIORef ref+ if index <= maxIndex+ then do+ writeIORef ref (succ index)+ return (Just index)+ else return Nothing
library/Potoki/Core/IO.hs view
@@ -23,12 +23,13 @@ produce :: Produce input -> forall x. IO x -> (input -> IO x) -> IO x produce (Produce produceAcquire) stop emit =- C.acquire produceAcquire $ \ (Fetch fetchIO) ->- fix $ \ doLoop -> do+ C.acquire produceAcquire $ \ (Fetch fetchIO) -> let+ loop = do fetch <- fetchIO case fetch of Nothing -> stop- Just element -> emit element >> doLoop+ Just element -> emit element >> loop+ in loop consume :: IO (Maybe input) -> Consume input output -> IO output consume fetchIO (Consume consumeIO) =
library/Potoki/Core/IO/Fetch.hs view
@@ -6,11 +6,13 @@ {-| Fetch all the elements running the provided handler on them -} fetchAndHandleAll :: Fetch element -> IO () -> (element -> IO ()) -> IO () fetchAndHandleAll (Fetch fetchIO) onEnd onElement =- fix $ \ doLoop -> do- fetch <- fetchIO- case fetch of- Nothing -> onEnd- Just element -> onElement element >> doLoop+ let+ loop = do+ fetch <- fetchIO+ case fetch of+ Nothing -> onEnd+ Just element -> onElement element >> loop+ in loop {-| Fetch and handle just one element -} fetchAndHandle :: Fetch element -> IO a -> (element -> IO a) -> IO a
library/Potoki/Core/Prelude.hs view
@@ -4,6 +4,7 @@ ioChunkSize, textString, unsnoc,+ mapLeft, ) where @@ -35,10 +36,11 @@ import Data.Ix as Exports import Data.List as Exports hiding (sortOn, isSubsequenceOf, uncons, concat, foldr, foldl1, maximum, minimum, product, sum, all, and, any, concatMap, elem, foldl, foldr1, notElem, or, find, maximumBy, minimumBy, mapAccumL, mapAccumR, foldl') import Data.Maybe as Exports-import Data.Monoid as Exports hiding (Last(..), First(..))+import Data.Monoid as Exports hiding (Last(..), First(..), (<>)) import Data.Ord as Exports import Data.Proxy as Exports import Data.Ratio as Exports+import Data.Semigroup as Exports import Data.STRef as Exports import Data.String as Exports import Data.Traversable as Exports@@ -50,7 +52,7 @@ import Foreign.ForeignPtr as Exports import Foreign.Ptr as Exports import Foreign.StablePtr as Exports-import Foreign.Storable as Exports+import Foreign.Storable as Exports hiding (sizeOf, alignment) import GHC.Conc as Exports hiding (withMVar, threadWaitWriteSTM, threadWaitWrite, threadWaitReadSTM, threadWaitRead) import GHC.Exts as Exports (lazy, inline, sortWith, groupWith) import GHC.Generics as Exports (Generic)@@ -101,10 +103,31 @@ ------------------------- import Data.Vector as Exports (Vector) +-- time+-------------------------+import Data.Time as Exports+ -- hashable ------------------------- import Data.Hashable as Exports (Hashable) +-- primitive+-------------------------+import Data.Primitive as Exports++-- deepseq+-------------------------+import Control.DeepSeq as Exports++-- deferred-folds+-------------------------+import DeferredFolds.Unfoldr as Exports (Unfoldr(..))++-- stm-chans+-------------------------+import Control.Concurrent.STM.TBMChan as Exports+import Control.Concurrent.STM.TMChan as Exports+ -------------------------------------------------------------------------------- import qualified Data.Text as A@@ -131,3 +154,9 @@ _ -> case process tailVal of (initVal, lastMaybe) -> (headVal : initVal, lastMaybe) _ -> ([], Nothing)++{-# INLINE mapLeft #-}+mapLeft :: (oldLeft -> newLeft) -> Either oldLeft right -> Either newLeft right+mapLeft f = \ case+ Left a -> Left (f a)+ Right b -> Right b
library/Potoki/Core/Produce.hs view
@@ -4,6 +4,7 @@ list, transform, vector,+ vectorWithIndices, hashMapRows, fileBytes, fileBytesAtOffset,@@ -12,16 +13,24 @@ directoryContents, finiteMVar, infiniteMVar,+ tbmChan,+ tmChan,+ lazyByteString,+ enumInRange,+ mergeOrdering,+ unfoldr, ) where -import Potoki.Core.Prelude+import Potoki.Core.Prelude hiding (unfoldr) import Potoki.Core.Types import qualified Potoki.Core.Fetch as A import qualified Data.HashMap.Strict as B import qualified Data.Vector as C+import qualified Data.Vector.Generic as GenericVector import qualified System.Directory as G import qualified Acquire.Acquire as M+import qualified Data.ByteString.Lazy as D deriving instance Functor Produce@@ -82,6 +91,13 @@ getElement return fetch +instance Semigroup (Produce a) where+ (<>) = (<|>)++instance Monoid (Produce a) where+ mempty = empty+ mappend = (<>)+ {-# INLINABLE list #-} list :: [input] -> Produce input list inputList =@@ -95,18 +111,31 @@ transformAcquire fetch {-# INLINE vector #-}-vector :: Vector input -> Produce input+vector :: GenericVector.Vector vector input => vector input -> Produce input vector vectorVal =- Produce $ M.Acquire $ do+ Produce $ liftIO $ do indexRef <- newIORef 0 let fetch = A.Fetch $ do indexVal <- readIORef indexRef writeIORef indexRef $! succ indexVal- return $ (C.!?) vectorVal indexVal- in return (fetch, return ())+ return $ (GenericVector.!?) vectorVal indexVal+ in return fetch +{-# INLINE vectorWithIndices #-}+vectorWithIndices :: GenericVector.Vector vector a => vector a -> Produce (Int, a)+vectorWithIndices vectorVal =+ Produce $ liftIO $ do+ indexRef <- newIORef 0+ let+ fetch =+ A.Fetch $ do+ indexVal <- readIORef indexRef+ writeIORef indexRef $! succ indexVal+ return $ fmap (indexVal,) $ (GenericVector.!?) vectorVal indexVal+ in return fetch+ {-# INLINE hashMapRows #-} hashMapRows :: HashMap a b -> Produce (a, b) hashMapRows =@@ -208,3 +237,92 @@ infiniteMVar :: MVar element -> Produce element infiniteMVar var = Produce $ M.Acquire (return (A.infiniteMVar var, return ()))++{-# INLINE tbmChan #-}+tbmChan :: TBMChan element -> Produce element+tbmChan chan = do+ Produce $ return $ Fetch $ atomically $ readTBMChan chan++{-# INLINE tmChan #-}+tmChan :: TMChan element -> Produce element+tmChan chan = do+ Produce $ return $ Fetch $ atomically $ readTMChan chan++{-# INLINE lazyByteString #-}+lazyByteString :: D.ByteString -> Produce ByteString+lazyByteString lbs =+ Produce $ M.Acquire $ do+ ref <- newIORef lbs+ return (A.lazyByteStringRef ref, return ())++{-# INLINE enumInRange #-}+enumInRange :: (Enum a, Ord a) => a -> a -> Produce a+enumInRange from to =+ Produce $ M.Acquire $ do+ ref <- newIORef from+ return (A.enumUntil ref to, return ())++{-|+Merge two ordered sequences into one+-}+mergeOrdering :: (a -> a -> Bool) -> Produce a -> Produce a -> Produce a+mergeOrdering compare (Produce produceLeft) (Produce produceRight) = Produce $ do+ Fetch fetchLeft <- produceLeft+ Fetch fetchRight <- produceRight+ leftCache <- liftIO $ newIORef Nothing+ rightCache <- liftIO $ newIORef Nothing+ return $ Fetch $ do+ cachedLeftMaybe <- readIORef leftCache+ case cachedLeftMaybe of+ Just left -> do+ fetchedRightMaybe <- fetchRight+ case fetchedRightMaybe of+ Just right -> if compare left right+ then do+ writeIORef leftCache Nothing+ writeIORef rightCache (Just right)+ return (Just left)+ else return (Just right)+ Nothing -> do+ writeIORef leftCache Nothing+ return (Just left)+ Nothing -> do+ fetchedLeftMaybe <- fetchLeft+ case fetchedLeftMaybe of+ Just left -> do+ cachedRightMaybe <- readIORef rightCache+ case cachedRightMaybe of+ Just right -> if compare left right+ then return (Just left)+ else do+ writeIORef leftCache (Just left)+ writeIORef rightCache Nothing+ return (Just right)+ Nothing -> do+ fetchedRightMaybe <- fetchRight+ case fetchedRightMaybe of+ Just right -> if compare left right+ then do+ writeIORef rightCache (Just right)+ return (Just left)+ else do+ writeIORef leftCache (Just left)+ return (Just right)+ Nothing -> return (Just left)+ Nothing -> do+ cachedRightMaybe <- readIORef rightCache+ case cachedRightMaybe of+ Just right -> do+ writeIORef rightCache Nothing+ return (Just right)+ Nothing -> fetchRight++unfoldr :: Unfoldr a -> Produce a+unfoldr (Unfoldr unfoldr) = Produce $ liftIO $ do+ fetchRef <- newIORef (return Nothing)+ writeIORef fetchRef $ let+ step !a fetchNext = do+ writeIORef fetchRef fetchNext+ return (Just a)+ in unfoldr step (return Nothing)+ return $ Fetch $ join $ readIORef fetchRef
+ library/Potoki/Core/TextBuilder.hs view
@@ -0,0 +1,20 @@+module Potoki.Core.TextBuilder+where++import Potoki.Core.Prelude+import Text.Builder+++count :: Int -> Builder+count = thousandSeparatedUnsignedDecimal ','++streamProgressMessage :: Int -> Int -> Builder+streamProgressMessage progress total =+ "Processed " <> count progress <> " elements (" <>+ count total <> " in total)"++indexationSummary :: Int -> Int -> Builder+indexationSummary userAmount uriAmount =+ "Got the following amounts of nodes:\n" <>+ "\tUsers: " <> count userAmount <> "\n" <>+ "\tUris: " <> count uriAmount
library/Potoki/Core/Transform.hs view
@@ -14,28 +14,47 @@ just, list, vector,+ batch, distinctBy, distinct, executeIO, mapInIO,+ reportProgress,+ handleProgressAndCountOnInterval,+ uniquify, -- * ByteString- module Potoki.Core.Transform.ByteString,+ builderChunks,+ extractLines,+ extractLinesWithoutTrail,+ extractLinesConcurrently, -- * State R.runState, R.execState, R.evalState, -- * Parsing+ B.scan, A.parseBytes, A.parseText,+ A.parseLineBytesConcurrently,+ A.parseNonEmptyLineBytesConcurrently, -- * Concurrency N.bufferize,+ N.bufferizeFlushing, N.concurrently,+ N.concurrentlyInOrder,+ N.unsafeConcurrently, N.async,+ N.concurrentlyWithBatching,+ N.concurrentlyInOrderWithBatching, -- * File IO deleteFile, appendBytesToFile, writeTextToFile, -- * Debugging+ count,+ mapInIOWithCounter,+ handleCount,+ handleCountOnInterval, traceWithCounter, ) where@@ -45,6 +64,7 @@ import Potoki.Core.Transform.FileIO import Potoki.Core.Transform.ByteString import qualified Potoki.Core.Transform.Attoparsec as A+import qualified Potoki.Core.Transform.Scanner as B import qualified Potoki.Core.Transform.Concurrency as N import qualified Potoki.Core.Transform.State as R
library/Potoki/Core/Transform/Attoparsec.hs view
@@ -3,29 +3,33 @@ import Potoki.Core.Prelude hiding (take, takeWhile, filter, drop) import Potoki.Core.Types+import Potoki.Core.Transform.Basic+import Potoki.Core.Transform.Concurrency+import Potoki.Core.Transform.ByteString import qualified Potoki.Core.Fetch as A import qualified Data.Attoparsec.ByteString as K import qualified Data.Attoparsec.Text as L import qualified Data.Attoparsec.Types as M import qualified Acquire.Acquire as N+import qualified Data.ByteString as ByteString {-# INLINE mapWithParseResult #-} mapWithParseResult :: forall input parsed. (Monoid input, Eq input) => (input -> M.IResult input parsed) -> Transform input (Either Text parsed) mapWithParseResult inputToResult = Transform $ \inputFetch -> N.Acquire $ do- unconsumedRef <- newIORef mempty- finishedRef <- newIORef False+ unconsumedRef <- newMutVar mempty+ finishedRef <- newMutVar False return (A.Fetch (fetchParsed inputFetch finishedRef unconsumedRef), return ()) where- fetchParsed :: A.Fetch input -> IORef Bool -> IORef input -> IO (Maybe (Either Text parsed))+ fetchParsed :: A.Fetch input -> MutVar RealWorld Bool -> MutVar RealWorld input -> IO (Maybe (Either Text parsed)) fetchParsed (A.Fetch inputFetchIO) finishedRef unconsumedRef = do- finished <- readIORef finishedRef+ finished <- readMutVar finishedRef if finished then return Nothing else do- unconsumed <- readIORef unconsumedRef+ unconsumed <- readMutVar unconsumedRef if unconsumed == mempty then inputFetchIO >>= \case@@ -35,7 +39,7 @@ then return Nothing else matchResult (inputToResult input) else do- writeIORef unconsumedRef mempty+ writeMutVar unconsumedRef mempty matchResult (inputToResult unconsumed) where matchResult :: M.IResult input parsed -> IO (Maybe (Either Text parsed))@@ -45,12 +49,12 @@ consumeVal inputToResultVal M.Done unconsumed parsed -> do- writeIORef unconsumedRef unconsumed+ writeMutVar unconsumedRef unconsumed return (Just (Right parsed)) M.Fail unconsumed contexts message -> do- writeIORef unconsumedRef unconsumed- writeIORef finishedRef True+ writeMutVar unconsumedRef unconsumed+ writeMutVar finishedRef True return (Just (Left resultMessage)) where resultMessage =@@ -60,10 +64,10 @@ consumeVal inputToResultVal' = inputFetchIO >>= \case Nothing -> do- writeIORef finishedRef True+ writeMutVar finishedRef True matchResult (inputToResultVal' mempty) Just input -> do- when (input == mempty) (writeIORef finishedRef True)+ when (input == mempty) (writeMutVar finishedRef True) matchResult (inputToResultVal' input) {-|@@ -81,3 +85,23 @@ parseText :: L.Parser parsed -> Transform Text (Either Text parsed) parseText parser = mapWithParseResult (L.parse parser)++{-|+Lift an Attoparsec ByteString parser to a transform,+which parses the lines concurrently.+-}+{-# INLINE parseLineBytesConcurrently #-}+parseLineBytesConcurrently :: NFData a => Int -> K.Parser a -> Transform ByteString (Either Text a)+parseLineBytesConcurrently concurrency parser =+ extractLines >>> bufferize concurrency >>>+ unsafeConcurrently concurrency (arr (mapLeft fromString . K.parseOnly parser))++{-|+Lift an Attoparsec ByteString parser to a transform,+which parses the lines concurrently.+-}+{-# INLINE parseNonEmptyLineBytesConcurrently #-}+parseNonEmptyLineBytesConcurrently :: NFData a => Int -> K.Parser a -> Transform ByteString (Either Text a)+parseNonEmptyLineBytesConcurrently concurrency parser =+ extractLinesWithoutTrail >>> filter (not . ByteString.null) >>> bufferize (concurrency * 2) >>>+ unsafeConcurrently concurrency (arr (mapLeft fromString . K.parseOnly parser))
library/Potoki/Core/Transform/Basic.hs view
@@ -7,6 +7,10 @@ import qualified Data.HashSet as C import qualified Data.Vector as P import qualified Acquire.Acquire as M+import qualified Data.Vector.Generic.Mutable as MutableGenericVector+import qualified Data.Vector.Generic as GenericVector+import qualified Text.Builder as TextBuilder+import qualified Potoki.Core.TextBuilder as TextBuilder {-# INLINE mapFilter #-}@@ -32,80 +36,131 @@ {-# INLINE drop #-} drop :: Int -> Transform input input drop amount =- Transform $ \ (A.Fetch fetchIO) -> M.Acquire $ do- countRef <- newIORef amount- return $ (, return ()) $ - A.Fetch $ fix $ \ doLoop -> do- count <- readIORef countRef- if count > 0- then do- writeIORef countRef $! pred count- doLoop- else fetchIO+ Transform $ \ (A.Fetch fetchIO) -> do+ countRef <- liftIO $ newIORef amount+ return $ Fetch $ do+ let+ loop = do+ count <- readIORef countRef+ fetch <- fetchIO+ case fetch of+ Just _ ->+ if count > 0+ then do+ writeIORef countRef $! pred count+ loop+ else (return fetch)+ Nothing -> return Nothing+ loop {-# INLINE list #-} list :: Transform [a] a list =- Transform $ \ (A.Fetch fetchListIO) -> M.Acquire $ do+ Transform $ \ (A.Fetch fetchListIO) -> liftIO $ do bufferRef <- newIORef []- return $ (, return ()) $ - A.Fetch $ do+ let+ fetchFromBufferOr whenEmpty = do buffer <- readIORef bufferRef case buffer of- headVal : tailVal -> do- writeIORef bufferRef tailVal- return (Just headVal)- _ ->- let- fetchElementIO = do- fetchListIO >>= \case- Nothing -> return Nothing- Just (headVal : tailVal) -> do- writeIORef bufferRef tailVal- return (Just headVal)- _ -> do- writeIORef bufferRef []- return Nothing- in fetchElementIO+ (!head) : tail -> do+ writeIORef bufferRef tail+ return (Just head)+ _ -> whenEmpty+ fetchFromSource = do+ fetchedList <- fetchListIO+ case fetchedList of+ Just list -> case list of+ (!head) : tail -> do+ writeIORef bufferRef tail+ return (Just head)+ _ -> do+ fetchFromSource+ Nothing -> return Nothing+ in return $ A.Fetch $ fetchFromBufferOr fetchFromSource {-# INLINABLE vector #-}-vector :: Transform (Vector a) a+vector :: GenericVector.Vector vector a => Transform (vector a) a vector =- Transform $ \ (A.Fetch fetchVectorIO) -> M.Acquire $ do+ Transform $ \ (Fetch fetchVectorIO) -> liftIO $ do indexRef <- newIORef 0- vectorRef <- newIORef mempty- return $ (, return ()) $ - A.Fetch $ fix $ \ doLoop -> do+ vectorRef <- newIORef GenericVector.empty+ return $ Fetch $ let+ loop = do vectorVal <- readIORef vectorRef indexVal <- readIORef indexRef- if indexVal < P.length vectorVal+ if indexVal < GenericVector.length vectorVal then do writeIORef indexRef (succ indexVal)- return (Just (P.unsafeIndex vectorVal indexVal))+ return (Just (GenericVector.unsafeIndex vectorVal indexVal)) else fetchVectorIO >>= \case Just vectorVal' -> do writeIORef vectorRef vectorVal' writeIORef indexRef 0- doLoop+ loop Nothing -> return Nothing+ in loop +{-|+Chunk the stream to vector batches of the given size.++It's useful in combination with 'concurrently' in cases where the lifted transform's iteration is too light.+Actually, there is a composed variation of 'concurrently', which utilizes it: 'concurrentlyWithBatching'.+-}+{-# INLINABLE batch #-}+batch :: GenericVector.Vector vector a => Int -> Transform a (vector a)+batch size = if size < 1+ then Transform $ const $ liftIO $ return $ empty+ else Transform $ \ (Fetch fetch) -> liftIO $ do+ mvec <- MutableGenericVector.new size+ cursor <- newIORef 0+ activeVar <- newIORef True+ return $ Fetch $ let+ loop = do+ active <- readIORef activeVar+ if active+ then do+ fetchingResult <- fetch+ case fetchingResult of+ Just !a -> do+ index <- readIORef cursor+ MutableGenericVector.unsafeWrite mvec index a+ let !nextIndex = succ index+ if nextIndex == size+ then do+ writeIORef cursor 0+ !vec <- GenericVector.freeze mvec+ return (Just vec)+ else do+ writeIORef cursor nextIndex+ loop+ Nothing -> do+ writeIORef activeVar False+ index <- readIORef cursor+ if index > 0+ then do+ !vec <- GenericVector.freeze (MutableGenericVector.unsafeSlice 0 index mvec)+ return (Just vec)+ else return Nothing+ else return Nothing+ in loop+ {-# INLINE distinctBy #-} distinctBy :: (Eq comparable, Hashable comparable) => (element -> comparable) -> Transform element element distinctBy f = Transform $ \ (A.Fetch fetch) -> M.Acquire $ do stateRef <- newIORef mempty- return $ (, return ()) $ - A.Fetch $ fix $ \ doLoop -> - fetch >>= \case - Nothing -> return Nothing- Just input -> do- let comparable = f input- !set <- readIORef stateRef- if C.member comparable set- then doLoop- else do- writeIORef stateRef $! C.insert comparable set- return (Just input)+ return $ (, return ()) $ A.Fetch $ let+ loop = fetch >>= \case+ Nothing -> return Nothing+ Just input -> do+ let comparable = f input+ !set <- readIORef stateRef+ if C.member comparable set+ then loop+ else do+ writeIORef stateRef $! C.insert comparable set+ return (Just input)+ in loop {-# INLINE distinct #-} distinct :: (Eq element, Hashable element) => Transform element element@@ -125,17 +180,47 @@ Transform acquire <- liftIO io acquire fetch +count :: Transform a Int+count = Transform $ \ (Fetch fetchIO) -> do+ counter <- liftIO (newIORef 0)+ return $ Fetch $ do+ result <- fetchIO+ case result of+ Just _ -> Just <$> atomicModifyIORef' counter (\ n -> (succ n, n))+ Nothing -> return Nothing++mapInIOWithCounter :: (Int -> a -> IO b) -> Transform a b+mapInIOWithCounter handler =+ ioTransform $ do+ counter <- newIORef 0+ return $ mapInIO $ \ !a -> do+ count <- atomicModifyIORef' counter (\ n -> (succ n, n))+ handler count a++handleCount :: (Int -> IO ()) -> Transform a a+handleCount handler = mapInIOWithCounter $ \ count a -> do+ handler count+ return a+ {-|+Provides for progress monitoring by means of periodic measurement.+-}+handleCountOnInterval :: NominalDiffTime -> (Int -> IO ()) -> Transform a a+handleCountOnInterval interval handler = ioTransform $ do+ nextTime <- addUTCTime interval <$> getCurrentTime+ nextTimeRef <- newIORef nextTime+ return $ handleCount $ \ count -> do+ nextTime <- readIORef nextTimeRef+ time <- getCurrentTime+ when (time >= nextTime) $ do+ writeIORef nextTimeRef (addUTCTime interval nextTime)+ handler count++{-| Useful for debugging -} traceWithCounter :: (Int -> String) -> Transform a a-traceWithCounter showFunc =- ioTransform $ do- counter <- newIORef 0- return $ mapInIO $ \ x -> do- n <- atomicModifyIORef' counter (\ n -> (succ n, n))- putStrLn (showFunc n)- return x+traceWithCounter shower = handleCount (putStrLn . shower) {-# INLINE consume #-} consume :: Consume input output -> Transform input output@@ -175,29 +260,31 @@ produce inputToProduce = Transform $ \ (Fetch inputFetchIO) -> do stateRef <- liftIO $ newIORef Nothing- return $ Fetch $ fix $ \ doLoop -> do- state <- readIORef stateRef- case state of- Just (Fetch outputFetchIO, kill) ->- do- outputFetchResult <- outputFetchIO- case outputFetchResult of- Just x -> return (Just x)- Nothing -> do- kill- writeIORef stateRef Nothing- doLoop- Nothing ->- do- inputFetchResult <- inputFetchIO- case inputFetchResult of- Just input -> do- case inputToProduce input of- Produce (Acquire produceIO) -> do- fetchAndKill <- produceIO- writeIORef stateRef (Just fetchAndKill)- doLoop- Nothing -> return Nothing+ return $ Fetch $ let+ loop = do+ state <- readIORef stateRef+ case state of+ Just (Fetch outputFetchIO, kill) ->+ do+ outputFetchResult <- outputFetchIO+ case outputFetchResult of+ Just x -> return (Just x)+ Nothing -> do+ kill+ writeIORef stateRef Nothing+ loop+ Nothing ->+ do+ inputFetchResult <- inputFetchIO+ case inputFetchResult of+ Just input -> do+ case inputToProduce input of+ Produce (Acquire produceIO) -> do+ fetchAndKill <- produceIO+ writeIORef stateRef (Just fetchAndKill)+ loop+ Nothing -> return Nothing+ in loop {-# INLINE mapFetch #-} mapFetch :: (Fetch a -> Fetch b) -> Transform a b@@ -228,4 +315,39 @@ fetchIO else return Nothing- ++reportProgress :: (Text -> IO ()) -> Transform a a+reportProgress log = handleProgressAndCountOnInterval 1 $ \ progress count ->+ log $ TextBuilder.run $ TextBuilder.streamProgressMessage progress count++handleProgressAndCountOnInterval :: NominalDiffTime -> (Int -> Int -> IO ()) -> Transform a a+handleProgressAndCountOnInterval interval handle = ioTransform $ do+ lastCountRef <- newIORef 0+ return $ handleCountOnInterval interval $ \ count -> do+ lastCount <- readIORef lastCountRef+ writeIORef lastCountRef count+ handle (count - lastCount) count++uniquify :: (Eq a) => Transform a a+uniquify = + Transform $ \ (Fetch fetchIO) -> M.Acquire $ do+ duplicateRef <- newIORef Nothing+ return $ (, return ()) $ A.Fetch $ let+ loop = do+ dupliacte <- readIORef duplicateRef+ fetch <- fetchIO+ case fetch of+ Nothing -> return Nothing+ Just elem -> do+ case dupliacte of+ Nothing -> do+ writeIORef duplicateRef fetch+ return fetch+ Just dupl -> do+ if elem == dupl + then+ loop+ else do+ writeIORef duplicateRef (Just elem)+ return fetch+ in loop
library/Potoki/Core/Transform/ByteString.hs view
@@ -4,6 +4,7 @@ import Potoki.Core.Prelude hiding (filter) import Potoki.Core.Transform.Basic import Potoki.Core.Transform.Instances ()+import Potoki.Core.Transform.Concurrency import Potoki.Core.Types import qualified Potoki.Core.Fetch as A import qualified Potoki.Core.Produce as H@@ -26,20 +27,13 @@ -} extractLines :: Transform ByteString ByteString extractLines =- lineList >>> filter (not . null) >>> list+ lineList >>> list where lineList =- Transform $ \ (A.Fetch fetchIO) -> M.Acquire $ do+ Transform $ \ (Fetch fetchIO) -> liftIO $ do stateRef <- newIORef Nothing- return $ (, return ()) $ A.Fetch $ fetchIO >>= \case- Nothing -> (do- state <- readIORef stateRef- case state of- Just poking -> do- writeIORef stateRef Nothing- return (Just [D.poking poking])- Nothing -> return Nothing)- Just chunk -> (+ return $ Fetch $ fetchIO >>= \case+ Just chunk -> case B.split 10 chunk of firstInput : tailVal -> do state <- readIORef stateRef@@ -47,12 +41,99 @@ newPoking = fold state <> C.bytes firstInput in case unsnoc tailVal of- Just (initVal, lastVal) ->+ Just (!initVal, !lastVal) -> do- writeIORef stateRef (Just (C.bytes lastVal))- return (Just (D.poking newPoking : initVal))+ writeIORef stateRef $! Just $! C.bytes lastVal+ let !bytes = D.poking newPoking+ return (Just (bytes : initVal)) Nothing -> do writeIORef stateRef (Just newPoking) return (Just [])- _ -> return (Just []))+ _ -> return (Just [])+ Nothing -> do+ state <- readIORef stateRef+ case state of+ Just poking -> do+ writeIORef stateRef Nothing+ return (Just [D.poking poking])+ Nothing -> return Nothing++extractLinesWithoutTrail :: Transform ByteString ByteString+extractLinesWithoutTrail =+ lineList >>> list+ where+ lineList =+ Transform $ \ (Fetch fetchIO) -> liftIO $ do+ pokingRef <- newIORef mempty+ return $ Fetch $ do+ fetchResult <- fetchIO+ case fetchResult of+ Just chunk -> case B.split 10 chunk of+ head : tail -> do+ poking <- readIORef pokingRef+ let newPoking = poking <> C.bytes head+ case unsnoc tail of+ Just (!init, last) -> do+ writeIORef pokingRef (C.bytes last)+ let+ !bytes = D.poking newPoking+ in return (Just (bytes : init))+ Nothing -> do+ writeIORef pokingRef newPoking+ return (Just [])+ _ -> return (Just [])+ Nothing -> do+ poking <- readIORef pokingRef+ if C.null poking+ then return Nothing+ else let+ !bytes = D.poking poking+ in do+ writeIORef pokingRef mempty+ return (Just [bytes])++extractLinesConcurrently :: Int -> Transform ByteString ByteString+extractLinesConcurrently concurrency =+ concurrentlyInOrderWithBatching 100 concurrency (arr (B.split 10)) >>> mergeLineChunks++mergeLineChunks :: Transform [ByteString] ByteString+mergeLineChunks = Transform $ \ (Fetch fetchChunkList) -> liftIO $ do+ cachedChunkListRef <- newIORef (Just [])+ incompleteChunkRef <- newIORef Nothing+ return $ Fetch $ let+ loop = do+ cachedChunkListIfAny <- readIORef cachedChunkListRef+ case cachedChunkListIfAny of+ Just cachedChunkList -> case cachedChunkList of+ (!a) : b : tail -> do+ writeIORef cachedChunkListRef (Just (b : tail))+ return (Just a)+ (!a) : _ -> do+ writeIORef cachedChunkListRef (Just [])+ writeIORef incompleteChunkRef (Just a)+ loop+ _ -> do+ newChunkListIfAny <- fetchChunkList+ incompleteChunkIfAny <- readIORef incompleteChunkRef+ case incompleteChunkIfAny of+ Just incompleteChunk -> case newChunkListIfAny of+ Just newChunkList -> case newChunkList of+ a : b : tail -> do+ completeChunk <- evaluate (incompleteChunk <> a)+ writeIORef cachedChunkListRef (Just (b : tail))+ writeIORef incompleteChunkRef Nothing+ return (Just completeChunk)+ a : _ -> do+ newIncompleteChunk <- evaluate (incompleteChunk <> a)+ writeIORef incompleteChunkRef (Just newIncompleteChunk)+ loop+ _ -> loop+ Nothing -> do+ writeIORef cachedChunkListRef Nothing+ return (Just incompleteChunk)+ Nothing -> do+ writeIORef cachedChunkListRef newChunkListIfAny+ loop+ Nothing -> return Nothing+ in loop
library/Potoki/Core/Transform/Concurrency.hs view
@@ -1,30 +1,67 @@ module Potoki.Core.Transform.Concurrency-(- bufferize,- concurrently,- async,-) where import Potoki.Core.Prelude hiding (take, takeWhile, filter) import Potoki.Core.Transform.Instances ()+import Potoki.Core.Transform.Basic import Potoki.Core.Types import qualified Potoki.Core.Fetch as A import qualified Acquire.Acquire as M-import qualified Control.Concurrent.Chan.Unagi.Bounded as B +bufferizeFlushing :: Int -> Transform input [input]+bufferizeFlushing maxSize =+ Transform $ \ (A.Fetch fetchIO) -> liftIO $ do+ buffer <- newTBQueueIO (fromIntegral maxSize)+ activeVar <- newTVarIO True++ forkIO $ let+ loop = do+ fetchingResult <- fetchIO+ case fetchingResult of+ Just !element -> do+ atomically $ writeTBQueue buffer element+ loop+ Nothing -> atomically $ writeTVar activeVar False+ in loop++ return $ Fetch $ atomically $ do+ batch <- flushTBQueue buffer+ if null batch+ then do+ active <- readTVar activeVar+ if active+ then retry+ else return Nothing+ else return (Just batch)+ {-# INLINE bufferize #-}-bufferize :: Int -> Transform element element+bufferize :: NFData element => Int -> Transform element element bufferize size =- Transform $ \ (A.Fetch fetch) -> M.Acquire $ do- (inChan, outChan) <- B.newChan size- forkIO $ fix $ \ doLoop ->- fetch >>= \case- Nothing -> B.writeChan inChan Nothing- Just !element -> B.writeChan inChan (Just element) >> doLoop- return $ (A.Fetch $ B.readChan outChan, return ())+ Transform $ \ (A.Fetch fetchIO) -> liftIO $ do+ buffer <- newTBQueueIO (fromIntegral size)+ activeVar <- newTVarIO True + forkIO $ let+ loop = do+ fetchingResult <- fetchIO+ case fetchingResult of+ Just element -> do+ forcedElement <- evaluate (force element)+ atomically $ writeTBQueue buffer forcedElement+ loop+ Nothing -> atomically $ writeTVar activeVar False+ in loop++ return $ Fetch $ let+ readBuffer = Just <$> readTBQueue buffer+ terminate = do+ active <- readTVar activeVar+ if active+ then empty+ else return Nothing+ in atomically (readBuffer <|> terminate)+ {-| Identity Transform, which ensures that the inputs are fetched synchronously. @@ -33,12 +70,12 @@ {-# INLINABLE sync #-} sync :: Transform a a sync =- Transform $ \ (A.Fetch fetch) -> M.Acquire $ do+ Transform $ \ (A.Fetch fetchIO) -> liftIO $ do activeVar <- newMVar True- return $ (, return ()) $ A.Fetch $ do+ return $ A.Fetch $ do active <- takeMVar activeVar if active- then fetch >>= \case+ then fetchIO >>= \ case Just !element -> do putMVar activeVar True return (Just element)@@ -54,48 +91,137 @@ The order of the outputs produced is indiscriminate. -} {-# INLINABLE concurrently #-}-concurrently :: Int -> Transform input output -> Transform input output+concurrently :: NFData output => Int -> Transform input output -> Transform input output concurrently workersAmount transform = if workersAmount == 1 then transform else sync >>>- concurrentlyUnsafe workersAmount transform+ unsafeConcurrently workersAmount transform -{-# INLINE concurrentlyUnsafe #-}-concurrentlyUnsafe :: Int -> Transform input output -> Transform input output-concurrentlyUnsafe workersAmount (Transform syncTransformIO) = - Transform $ \ fetch -> M.Acquire $ do- outChan <- newEmptyMVar+{-# INLINE unsafeConcurrently #-}+unsafeConcurrently :: NFData output => Int -> Transform input output -> Transform input output+unsafeConcurrently workersAmount (Transform syncTransformIO) = + Transform $ \ fetchIO -> liftIO $ do+ chan <- newTBQueueIO (fromIntegral (workersAmount * 2))+ workersCounter <- newTVarIO workersAmount+ replicateM_ workersAmount $ forkIO $ do- let runAcquire (M.Acquire io) = io- (A.Fetch fetchIO, _) <- runAcquire $ syncTransformIO fetch- fix $ \ doLoop -> fetchIO >>= \case- Nothing -> putMVar outChan Nothing- Just !result -> putMVar outChan (Just result) >> doLoop- activeWorkersAmountVar <- newMVar workersAmount- return $ (, return ()) $ A.Fetch $ fix $ \ doLoop' -> do- activeWorkersAmount <- takeMVar activeWorkersAmountVar- if activeWorkersAmount <= 0- then return Nothing- else do- fetchResult <- takeMVar outChan+ (A.Fetch fetchIO, finalize) <- case syncTransformIO fetchIO of M.Acquire io -> io+ let+ loop = do+ fetchResult <- fetchIO case fetchResult of Just result -> do- putMVar activeWorkersAmountVar activeWorkersAmount- return (Just result)- Nothing -> do- putMVar activeWorkersAmountVar (pred activeWorkersAmount)- doLoop'+ forcedResult <- evaluate (force result)+ atomically (writeTBQueue chan forcedResult)+ loop+ Nothing -> atomically (modifyTVar' workersCounter pred)+ in loop *> finalize + return $ A.Fetch $ let+ readChan = Just <$> readTBQueue chan+ terminate = do+ workersActive <- readTVar workersCounter+ if workersActive > 0+ then empty+ else return Nothing+ in atomically (readChan <|> terminate)++concurrentlyInOrder :: NFData b => Int -> Transform a b -> Transform a b+concurrentlyInOrder concurrency (Transform transform) = Transform $ \ (Fetch fetchA) -> liftIO $ do+ inputQueue <- newTBQueueIO (fromIntegral concurrency)+ outputSlotQueue <- newTQueueIO+ liveWorkersVar <- newTVarIO concurrency++ forkIO $ let+ loop = do+ fetchAResult <- fetchA+ case fetchAResult of+ Just a -> do+ atomically $ writeTBQueue inputQueue (Just a)+ loop+ Nothing -> atomically $ replicateM_ concurrency $ writeTBQueue inputQueue Nothing+ in loop++ replicateM_ concurrency $ forkIO $ do+ outputQueue <- newTQueueIO+ needsSwitchVar <- newTVarIO False++ let+ localizedFetchA = Fetch $ atomically $ do+ needsSwitch <- readTVar needsSwitchVar+ if needsSwitch+ then writeTQueue outputQueue Nothing+ else writeTVar needsSwitchVar True+ writeTQueue outputSlotQueue outputQueue+ readTBQueue inputQueue++ in do+ (Fetch fetchB, finalize) <- case transform localizedFetchA of M.Acquire io -> io+ let+ loop = do+ fetchBResult <- fetchB+ case fetchBResult of+ Just b -> do+ forcedB <- evaluate (force b)+ atomically $ writeTQueue outputQueue (Just forcedB)+ loop+ Nothing -> do+ atomically $ do+ writeTQueue outputQueue Nothing+ modifyTVar' liveWorkersVar pred+ finalize+ in loop++ return $ Fetch $ atomically $ fix $ \ loop -> mplus+ (do+ outputQueue <- peekTQueue outputSlotQueue+ bIfAny <- readTQueue outputQueue+ case bIfAny of+ Just b -> return (Just b)+ Nothing -> do+ readTQueue outputSlotQueue+ loop)+ (do+ liveWorkers <- readTVar liveWorkersVar+ guard (liveWorkers <= 0)+ return Nothing)+ {-| A transform, which fetches the inputs asynchronously on the specified number of threads. -}-async :: Int -> Transform input input+async :: NFData input => Int -> Transform input input async workersAmount = - Transform $ \ (A.Fetch fetchIO) -> M.Acquire $ do- chan <- newEmptyMVar - replicateM_ workersAmount $ forkIO $ fix $ \ _ -> do- fetchResult <- fetchIO- putMVar chan fetchResult- return (A.finiteMVar chan, return ())+ Transform $ \ (A.Fetch fetchIO) -> liftIO $ do+ chan <- atomically newEmptyTMVar+ workersCounter <- atomically (newTVar workersAmount)++ replicateM_ workersAmount $ forkIO $ let+ loop = do+ fetchResult <- fetchIO+ case fetchResult of+ Just input -> atomically (putTMVar chan input) *> loop+ Nothing -> atomically (modifyTVar' workersCounter pred)+ in loop++ return $ A.Fetch $ let+ readChan = Just <$> takeTMVar chan+ terminate = do+ workersActive <- readTVar workersCounter+ if workersActive > 0+ then empty+ else return Nothing+ in atomically (readChan <|> terminate)++concurrentlyWithBatching :: (NFData a, NFData b) => Int -> Int -> Transform a b -> Transform a b+concurrentlyWithBatching batching concurrency transform =+ batch @Vector batching >>> bufferize concurrency >>>+ unsafeConcurrently concurrency (vector >>> transform >>> batch @Vector batching) >>>+ vector++concurrentlyInOrderWithBatching :: (NFData b) => Int -> Int -> Transform a b -> Transform a b+concurrentlyInOrderWithBatching batching concurrency transform =+ batch @Vector batching >>> + concurrentlyInOrder concurrency (vector >>> transform >>> batch @Vector batching) >>>+ vector
library/Potoki/Core/Transform/Instances.hs view
@@ -23,16 +23,16 @@ fetchedLeftMaybeRef <- liftIO $ newIORef Nothing Fetch rightFetchIO <- rightTransformAcquire (A.rightHandlingLeft (writeIORef fetchedLeftMaybeRef . Just) inFetch) return $ Fetch $ do- rightFetch <- rightFetchIO- case rightFetch of- Nothing -> do- fetchedLeftMaybe <- readIORef fetchedLeftMaybeRef- case fetchedLeftMaybe of- Nothing -> return Nothing- Just fetchedLeft -> do- writeIORef fetchedLeftMaybeRef Nothing- return $ Just (Left fetchedLeft)- Just element -> return $ Just (Right element)+ rightFetch <- rightFetchIO+ case rightFetch of+ Nothing -> do+ fetchedLeftMaybe <- readIORef fetchedLeftMaybeRef+ case fetchedLeftMaybe of+ Nothing -> return Nothing+ Just fetchedLeft -> do+ writeIORef fetchedLeftMaybeRef Nothing+ return $ Just (Left fetchedLeft)+ Just element -> return $ Just (Right element) instance Strong Transform where first' (Transform firstTransformAcquire) =
+ library/Potoki/Core/Transform/Scanner.hs view
@@ -0,0 +1,54 @@+module Potoki.Core.Transform.Scanner+where++import Potoki.Core.Prelude hiding (take, takeWhile, filter, drop)+import Potoki.Core.Types+import Potoki.Core.Transform.Basic+import Scanner+++{-# INLINE scan #-}+scan :: Scanner a -> Transform ByteString (Either Text a)+scan scanner =+ Transform $ \ (Fetch fetchIo) -> liftIO $ do+ unconsumedRef <- newMutVar mempty+ finishedRef <- newMutVar False+ return $ Fetch $ let+ matchResult =+ \ case+ More inputToResultVal ->+ consumeVal inputToResultVal+ Done unconsumed parsed ->+ do+ writeMutVar unconsumedRef unconsumed+ return (Just (Right parsed))+ Fail unconsumed message ->+ do+ writeMutVar unconsumedRef unconsumed+ writeMutVar finishedRef True+ return (Just (Left (fromString message)))+ consumeVal inputToResultVal' =+ fetchIo >>= \ case+ Nothing -> do+ writeMutVar finishedRef True+ matchResult (inputToResultVal' mempty)+ Just input -> do+ when (input == mempty) (writeMutVar finishedRef True)+ matchResult (inputToResultVal' input)+ in do+ finished <- readMutVar finishedRef+ if finished+ then return Nothing+ else do+ unconsumed <- readMutVar unconsumedRef+ if unconsumed == mempty+ then+ fetchIo >>= \ case+ Nothing -> return Nothing+ Just input -> do+ if input == mempty+ then return Nothing+ else matchResult (Scanner.scan scanner input)+ else do+ writeMutVar unconsumedRef mempty+ matchResult (Scanner.scan scanner unconsumed)
library/Potoki/Core/Transform/State.hs view
@@ -15,7 +15,7 @@ the "list" transform. -} {-# INLINE runState #-}-runState :: (input -> O.State state output) -> state -> Transform input (state, output)+runState :: (input -> O.State state output) -> state -> Transform input (output, state) runState stateFn initialState = Transform $ \ (A.Fetch fetchIO) -> M.Acquire $ do stateRef <- newIORef initialState@@ -26,15 +26,15 @@ case O.runState (stateFn input) currentState of (output, newState) -> do writeIORef stateRef newState- return (Just (newState, output))+ return (Just (output, newState)) Nothing -> return Nothing {-# INLINE evalState #-} evalState :: (input -> O.State state output) -> state -> Transform input output evalState stateFn initialState =- runState stateFn initialState >>> arr snd+ rmap fst (runState stateFn initialState) {-# INLINE execState #-} execState :: (input -> O.State state output) -> state -> Transform input state execState stateFn initialState =- runState stateFn initialState >>> arr fst+ rmap snd (runState stateFn initialState)
potoki-core.cabal view
@@ -1,47 +1,31 @@-name:- potoki-core-version:- 2.1.0.1-synopsis:- Low-level components of "potoki"+name: potoki-core+version: 2.3.4.1+synopsis: Low-level components of "potoki" description: Provides everything required for building custom instances of the \"potoki\" abstractions. Consider this library to be the Internals modules of \"potoki\".-category:- Streaming-homepage:- https://github.com/metrix-ai/potoki-core -bug-reports:- https://github.com/metrix-ai/potoki-core/issues -author:- Nikita Volkov <nikita.y.volkov@mail.ru>-maintainer:- Metrix.AI Ninjas <ninjas@metrix.ai>-copyright:- (c) 2017, Metrix.AI-license:- MIT-license-file:- LICENSE-build-type:- Simple-cabal-version:- >=1.10+category: Streaming+homepage: https://github.com/metrix-ai/potoki-core+bug-reports: https://github.com/metrix-ai/potoki-core/issues+author: Nikita Volkov <nikita.y.volkov@mail.ru>+maintainer: Metrix.AI Tech Team <tech@metrix.ai>+copyright: (c) 2017, Metrix.AI+license: MIT+license-file: LICENSE+build-type: Simple+cabal-version: >=1.10+tested-with: GHC ==8.4.2, GHC ==8.6.* + source-repository head- type:- git- location:- git://github.com/metrix-ai/potoki-core.git+ type: git+ location: git://github.com/metrix-ai/potoki-core.git library- hs-source-dirs:- library- default-extensions:- Arrows, BangPatterns, ConstraintKinds, DataKinds, DefaultSignatures, DeriveDataTypeable, DeriveFoldable, DeriveFunctor, DeriveGeneric, DeriveTraversable, EmptyDataDecls, FlexibleContexts, FlexibleInstances, FunctionalDependencies, GADTs, GeneralizedNewtypeDeriving, InstanceSigs, LambdaCase, LiberalTypeSynonyms, MagicHash, MultiParamTypeClasses, MultiWayIf, NoImplicitPrelude, NoMonomorphismRestriction, OverloadedStrings, PatternGuards, ParallelListComp, QuasiQuotes, RankNTypes, RecordWildCards, ScopedTypeVariables, StandaloneDeriving, TemplateHaskell, TupleSections, TypeFamilies, TypeOperators, UnboxedTuples- default-language:- Haskell2010+ hs-source-dirs: library+ default-extensions: Arrows, BangPatterns, ConstraintKinds, DataKinds, DefaultSignatures, DeriveDataTypeable, DeriveFoldable, DeriveFunctor, DeriveGeneric, DeriveTraversable, EmptyDataDecls, FlexibleContexts, FlexibleInstances, FunctionalDependencies, GADTs, GeneralizedNewtypeDeriving, InstanceSigs, LambdaCase, LiberalTypeSynonyms, MagicHash, MultiParamTypeClasses, MultiWayIf, NoImplicitPrelude, NoMonomorphismRestriction, OverloadedStrings, PatternGuards, ParallelListComp, QuasiQuotes, RankNTypes, RecordWildCards, ScopedTypeVariables, StandaloneDeriving, TemplateHaskell, TupleSections, TypeApplications, TypeFamilies, TypeOperators, UnboxedTuples+ default-language: Haskell2010 exposed-modules: Potoki.Core.Produce Potoki.Core.Fetch@@ -49,6 +33,7 @@ Potoki.Core.IO Potoki.Core.Transform other-modules:+ Potoki.Core.TextBuilder Potoki.Core.Types Potoki.Core.Prelude Potoki.Core.Transform.Attoparsec@@ -57,52 +42,66 @@ Potoki.Core.Transform.Concurrency Potoki.Core.Transform.FileIO Potoki.Core.Transform.State+ Potoki.Core.Transform.Scanner Potoki.Core.Transform.Instances Potoki.Core.IO.Fetch build-depends: acquire >=0.2 && <0.3,- base >=4.7 && <5,- profunctors >=5.2 && <6,- stm >=2.4 && <3, attoparsec >=0.13 && <0.15,+ base >=4.9 && <5, bytestring ==0.10.*,+ deepseq >=1.4 && <2,+ deferred-folds >=0.9.7.1 && <0.10, directory >=1.3 && <2, foldl >=1.3 && <2, hashable >=1 && <2,+ primitive >=0.6.4 && <0.7,+ profunctors >=5.2 && <6, ptr >=0.16.2 && <0.17,+ scanner >=0.3 && <0.4,+ stm >=2.5 && <2.6,+ stm-chans >=3 && <3.1, text >=1 && <2,+ text-builder >=0.6.3 && <0.7,+ time >=1.5 && <2, transformers >=0.5 && <0.6,- unagi-chan >=0.4 && <0.5, unordered-containers >=0.2 && <0.3, vector >=0.12 && <0.13 -test-suite tests- type:- exitcode-stdio-1.0- hs-source-dirs:- tests- main-is:- Main.hs+test-suite test+ type: exitcode-stdio-1.0+ hs-source-dirs: test+ main-is: Main.hs+ default-extensions: Arrows, BangPatterns, ConstraintKinds, DataKinds, DefaultSignatures, DeriveDataTypeable, DeriveFoldable, DeriveFunctor, DeriveGeneric, DeriveTraversable, EmptyDataDecls, FlexibleContexts, FlexibleInstances, FunctionalDependencies, GADTs, GeneralizedNewtypeDeriving, InstanceSigs, LambdaCase, LiberalTypeSynonyms, MagicHash, MultiParamTypeClasses, MultiWayIf, NoImplicitPrelude, NoMonomorphismRestriction, OverloadedStrings, PatternGuards, ParallelListComp, QuasiQuotes, RankNTypes, RecordWildCards, ScopedTypeVariables, StandaloneDeriving, TemplateHaskell, TupleSections, TypeApplications, TypeFamilies, TypeOperators, UnboxedTuples+ default-language: Haskell2010+ ghc-options: -threaded "-with-rtsopts=-N" other-modules: Potoki Transform- default-extensions:- Arrows, BangPatterns, ConstraintKinds, DataKinds, DefaultSignatures, DeriveDataTypeable, DeriveFoldable, DeriveFunctor, DeriveGeneric, DeriveTraversable, EmptyDataDecls, FlexibleContexts, FlexibleInstances, FunctionalDependencies, GADTs, GeneralizedNewtypeDeriving, InstanceSigs, LambdaCase, LiberalTypeSynonyms, MagicHash, MultiParamTypeClasses, MultiWayIf, NoImplicitPrelude, NoMonomorphismRestriction, OverloadedStrings, PatternGuards, ParallelListComp, QuasiQuotes, RankNTypes, RecordWildCards, ScopedTypeVariables, StandaloneDeriving, TemplateHaskell, TupleSections, TypeFamilies, TypeOperators, UnboxedTuples- default-language:- Haskell2010 build-depends:- --- potoki-core,- -- testing:- attoparsec, acquire >=0.2 && <0.3,- tasty >=1.0.1 && <1.2,- tasty-quickcheck >=0.10 && <0.11,- tasty-hunit >=0.10 && <0.11,- quickcheck-instances >=0.3.11 && <0.4,+ attoparsec,+ deferred-folds,+ foldl >=1.3.7 && <2,+ ilist >=0.3.1.0 && <0.4,+ split >=0.2.3.3 && <0.3,+ potoki-core, QuickCheck >=2.8.1 && <3,+ quickcheck-instances >=0.3.11 && <0.4, random >=1.1 && <2,- foldl >=1.3.7 && <1.4,- ilist >=0.3.1.0 && <0.4,- --- rerebase >=1.1 && <2+ rerebase >=1.1 && <2,+ tasty >=1.0.1 && <1.2,+ tasty-hunit >=0.10 && <0.11,+ tasty-quickcheck >=0.10 && <0.11++benchmark benchmark+ type: exitcode-stdio-1.0+ hs-source-dirs: benchmark+ default-extensions: Arrows, BangPatterns, ConstraintKinds, DataKinds, DefaultSignatures, DeriveDataTypeable, DeriveFoldable, DeriveFunctor, DeriveGeneric, DeriveTraversable, EmptyDataDecls, FlexibleContexts, FlexibleInstances, FunctionalDependencies, GADTs, GeneralizedNewtypeDeriving, InstanceSigs, LambdaCase, LiberalTypeSynonyms, MagicHash, MultiParamTypeClasses, MultiWayIf, NoImplicitPrelude, NoMonomorphismRestriction, OverloadedStrings, PatternGuards, ParallelListComp, QuasiQuotes, RankNTypes, RecordWildCards, ScopedTypeVariables, StandaloneDeriving, TemplateHaskell, TupleSections, TypeApplications, TypeFamilies, TypeOperators, UnboxedTuples+ default-language: Haskell2010+ ghc-options: -O2 -threaded "-with-rtsopts=-N -A64M"+ main-is: Main.hs+ build-depends:+ criterion >=1.5.1 && <2,+ potoki-core,+ rerebase >=1 && <2
+ test/Main.hs view
@@ -0,0 +1,264 @@+module Main where++import Prelude hiding (first, second, choose)+import Control.Arrow+import Test.QuickCheck.Instances+import Test.QuickCheck.Monadic as M+import Test.Tasty+import Test.Tasty.Runners+import Test.Tasty.HUnit+import Test.Tasty.QuickCheck+import qualified Potoki.Core.IO as C+import qualified Potoki.Core.Consume as D+import qualified Potoki.Core.Transform as A+import qualified Potoki.Core.Produce as E+import qualified Potoki.Core.Fetch as Fe+import qualified Data.Attoparsec.ByteString.Char8 as B+import qualified Data.ByteString as F+import qualified Data.ByteString.Char8 as I+import qualified Data.Vector as G+import qualified System.Random as H+import qualified Acquire.Acquire as Ac+import qualified DeferredFolds.Unfoldr as Unfoldr+import Potoki+import Transform++main =+ defaultMain $+ testGroup "All tests" $+ [+ testProperty "unfoldr" $ \ (list :: [Int]) ->+ list ===+ unsafePerformIO (C.produceAndConsume (E.unfoldr (Unfoldr.foldable list)) (D.list))+ ,+ testGroup "lines extraction props" $ let+ chunksGen = listOf $ do+ byteList <- listOf (frequency [(1, pure 10), (20, choose (97, 122))])+ return (F.pack byteList)+ in [+ testProperty "extractLines" $ forAll chunksGen $ \ chunks ->+ F.split 10 (F.concat chunks) ===+ unsafePerformIO (C.produceAndConsume (E.transform A.extractLines (E.list chunks)) D.list)+ ,+ testProperty "extractLinesWithoutTrail" $ forAll chunksGen $ \ chunks ->+ I.lines (F.concat chunks) ===+ unsafePerformIO (C.produceAndConsume (E.transform A.extractLinesWithoutTrail (E.list chunks)) D.list)+ ,+ testProperty "extractLinesConcurrently" $ forAll chunksGen $ \ chunks ->+ F.split 10 (F.concat chunks) ===+ unsafePerformIO (C.produceAndConsume (E.transform (A.extractLinesConcurrently numCapabilities) (E.list chunks)) D.list)+ ]+ ,+ testCase "extractLinesWithoutTrail" $ do+ assertEqual "" ["ab", "", "c", "d"] =<<+ C.produceAndConsume (E.transform A.extractLinesWithoutTrail (E.list ["a", "b\n\nc", "\nd\n"])) D.list+ ,+ testCase "extractLines" $ do+ assertEqual "" ["ab", "", "cd", ""] =<<+ C.produceAndConsume (E.transform A.extractLines (E.list ["a", "b\n", "\nc", "d\n"])) D.list+ ,+ testProperty "list to list" $ \ (list :: [Int]) ->+ list === unsafePerformIO (C.produceAndConsume (E.list list) D.list)+ ,+ testProperty "consecutive consumers" $ \ (list :: [Int], amount) ->+ list === unsafePerformIO (C.produceAndConsume (E.list list) ((++) <$> D.transform (A.take amount) D.list <*> D.list))+ ,+ potoki+ ,+ transform+ ,+ resourceChecker+ ,+ testCase "sync resource checker" $ do+ resourceVar <- newIORef Initial+ let produce = syncCheckResource resourceVar+ C.produceAndConsume produce D.sum+ fin <- readIORef resourceVar+ assertEqual "" Released fin+ ,+ testCase "async resource checker" $ do+ resourceVar <- newTVarIO Initial+ let produce = asyncCheckResource resourceVar+ potokiThreadId <- forkIO $ do+ C.produceAndConsume produce D.sum+ return ()+ atomically $ do+ resource <- readTVar resourceVar+ guard $ resource == Acquired+ killThread potokiThreadId+ atomically $ do+ resource <- readTVar resourceVar+ guard $ resource == Released+ ,+ testProperty "Produce.transform resource checker" $ \ (list :: [Int]) ->+ let prod = E.list list+ in monadicIO $ do+ check <- run $ do+ resourceVar1 <- newIORef Initial+ res <- C.produceAndConsume (E.transform (checkTransform resourceVar1) prod) D.sum+ readIORef resourceVar1+ M.assert $ check == Released+ ,+ testProperty "Consume.transform resource checker" $ \ (list :: [Int]) ->+ let prod = E.list list+ in monadicIO $ do+ check <- run $ do+ resourceVar1 <- newIORef Initial+ res <- C.produceAndConsume prod (D.transform (checkTransform resourceVar1) D.sum)+ readIORef resourceVar1+ M.assert $ check == Released+ ,+ testCase "Transform.produce resource checker #1" $ do+ resourceVar <- newIORef Initial+ let prod = checkProduce resourceVar (/= Released) 100+ res1 <- C.produceAndConsume (E.transform (A.produce intToProduce) prod) D.sum+ res2 <- C.produceAndConsume prod (D.transform (A.produce intToProduce) D.sum)+ fin <- readIORef resourceVar+ assertEqual "" Released fin+ assertEqual "" res1 res2+ ,+ testCase "Transform.produce resource checker #2" $ do+ resourceVar <- newIORef Initial+ let prod = checkProduce resourceVar (/= Released) 100+ res1 <- C.produceAndConsume (E.transform (A.take 5 >>> A.produce intToProduce) prod) D.sum+ res2 <- C.produceAndConsume prod (D.transform (A.take 5 >>> A.produce intToProduce) D.sum)+ fin <- readIORef resourceVar+ assertEqual "" Released fin+ assertEqual "" res1 res2+ ]++resourceChecker :: TestTree+resourceChecker =+ testGroup "produce1 >>= produce2" $+ [+ testCase "Check produce binding" $ do+ resourceVar1 <- newIORef Initial+ resourceVar2 <- newIORef Initial+ let prod1 = checkProduce resourceVar1 (const True) 100+ prod2 = \x -> checkProduce resourceVar2 (/= Released) x+ res <- C.produceAndConsume (prod1 >>= prod2) D.sum+ fin <- readIORef resourceVar1+ assertEqual "" Released fin+ ,+ testProperty "Bind for produce" $ \ (list :: [Int]) ->+ let check = list >>= (enumFromTo 0)+ prod1 = E.list list+ prod2 = \x -> E.list $ enumFromTo 0 x+ in monadicIO $ do+ res <- run $ C.produceAndConsume (prod1 >>= prod2) D.list+ M.assert (check == res)+ ,+ testProperty "liftIO for Produce. Consume0" $ \ (_ :: Int) ->+ monadicIO $ do+ check <- run $ do+ checkVar <- newIORef False+ let prod = liftIO $ writeIORef checkVar True+ C.produceAndConsume prod someThing+ readIORef checkVar+ M.assert $ check == False+ ,+ testProperty "liftIO for Produce. ConsumeN" $ \ (n :: Int) ->+ monadicIO $ do+ let prod = liftIO (return n)+ len <- run (C.produceAndConsume prod D.count)+ M.assert (len == 1)+ ,+ testCase "Check mergeOrdering" $ do+ resourceVar1 <- newIORef Initial+ resourceVar2 <- newIORef Initial+ let prod1 = checkProduce resourceVar1 (/= Released) 100+ prod2 = checkProduce resourceVar2 (/= Released) 50+ res <- C.produceAndConsume (E.mergeOrdering (\a b -> a <= b) prod1 prod2) D.list + fin1 <- readIORef resourceVar1+ fin2 <- readIORef resourceVar2+ assertEqual "" Released fin1+ assertEqual "" Released fin2+ ]++intToProduce :: Int -> E.Produce Int+intToProduce a = E.Produce . Ac.Acquire $ do+ stVar <- newIORef 0+ return $ flip (,) (return ()) $ Fe.Fetch $ do+ n <- readIORef stVar+ if n >= a + 1 then (return Nothing)+ else do+ writeIORef stVar $! n + 1+ return (Just n)++someThing :: D.Consume input Int+someThing = D.Consume $ \ (Fe.Fetch _) -> return 0++single :: Foldable f => f a -> Maybe a+single = join . (foldr' f Nothing)+ where+ f x Nothing = Just $ Just x+ f _ _ = Just Nothing++data Resource+ = Initial+ | Acquired+ | Released+ | AcquiredImproperly+ | ReleasedImproperly+ deriving (Show, Eq)++checkTransform :: IORef Resource -> A.Transform Int Int+checkTransform resourceVar = A.Transform $ \ fetchIO -> Ac.Acquire $ do+ writeIORef resourceVar Acquired+ return $ (,) (plusFetch fetchIO) $ do+ res <- readIORef resourceVar+ case res of+ Acquired -> writeIORef resourceVar Released+ _ -> writeIORef resourceVar ReleasedImproperly++plusFetch :: Fe.Fetch Int -> Fe.Fetch Int+plusFetch (Fe.Fetch fetchIO) = Fe.Fetch $ do+ fetch <- fetchIO+ return $ fmap succ fetch++checkProduce :: IORef Resource -> (Resource -> Bool) -> Int -> E.Produce Int+checkProduce resourceVar f k = E.Produce . Ac.Acquire $ do+ res <- readIORef resourceVar+ if f res && res /= Initial+ then do+ return $ (,) (Fe.Fetch $ return Nothing) $ writeIORef resourceVar AcquiredImproperly+ else do+ writeIORef resourceVar Acquired+ stVar <- newIORef 0+ let fetch = do+ n <- readIORef stVar+ if n >= k then return (Nothing)+ else do+ writeIORef stVar $! n + 1+ return (Just n)+ return $ (,) (Fe.Fetch fetch) $ do+ res <- readIORef resourceVar+ case res of+ Acquired -> writeIORef resourceVar Released+ _ -> writeIORef resourceVar ReleasedImproperly++asyncCheckResource :: TVar Resource -> E.Produce Int+asyncCheckResource resourceVar = E.Produce . Ac.Acquire $ do+ atomically $ writeTVar resourceVar Acquired+ stVar <- newIORef 0+ let fetch = do+ n <- readIORef stVar+ writeIORef stVar $! n + 1+ return (Just n)+ return (Fe.Fetch fetch, atomically $ writeTVar resourceVar Released)++syncCheckResource :: IORef Resource -> E.Produce Int+syncCheckResource resourceVar = E.Produce . Ac.Acquire $ do+ writeIORef resourceVar Acquired+ stVar <- newIORef 0+ let fetch = do+ n <- readIORef stVar+ if n >= 1000 then return (Nothing)+ else do+ writeIORef stVar $! n + 1+ return (Just n)+ return $ (,) (Fe.Fetch fetch) $ do+ res <- readIORef resourceVar+ case res of+ Acquired -> writeIORef resourceVar Released+ _ -> writeIORef resourceVar ReleasedImproperly
+ test/Potoki.hs view
@@ -0,0 +1,309 @@+module Potoki where++import Prelude hiding (first, second)+import Control.Arrow+import Test.QuickCheck.Instances+import Test.QuickCheck.Monadic as M+import Test.Tasty+import Test.Tasty.Runners+import Test.Tasty.HUnit+import Test.Tasty.QuickCheck+import qualified Control.Foldl as Fl+import qualified Potoki.Core.IO as C+import qualified Potoki.Core.Consume as D+import qualified Potoki.Core.Transform as A+import qualified Potoki.Core.Produce as E+import qualified Data.Attoparsec.ByteString.Char8 as B+import qualified Data.ByteString as F+import qualified Data.Vector as G+import qualified System.Random as H+import Data.List.Index (indexed)++potoki :: TestTree+potoki =+ testGroup "All tests for potoki's end-users functions" $+ [+ testCase "vector to list" $ do+ result <- C.produceAndConsume (E.vector (G.fromList [1,2,3])) (D.list)+ assertEqual "" [1,2,3] result+ ,+ testCase "just" $ do+ result <- C.produceAndConsume (E.list [Just 1, Nothing, Just 2]) (D.transform A.just D.list)+ assertEqual "" [1,2] result+ ,+ testCase "transform,consume,take" $ do+ let+ transform = A.consume (D.transform (A.take 3) D.list)+ consume = D.transform transform D.list+ produceAndConsume list = C.produceAndConsume (E.list list) (consume)+ assertEqual "" [[1,2,3], [4,5,6], [7,8]] =<< produceAndConsume [1,2,3,4,5,6,7,8]+ assertEqual "" [[1,2,3], [4,5,6], [7,8,9]] =<< produceAndConsume [1,2,3,4,5,6,7,8,9]+ assertEqual "" [] =<< produceAndConsume ([] :: [Int])+ ,+ testCase "File reading" $ do+ let produce =+ E.transform (arr (either (const Nothing) Just) >>> A.just) $+ E.fileBytes "samples/1"+ result <- C.produceAndConsume produce (fmap F.length D.concat)+ assertEqual "" 17400 result+ ,+ testCase "mergeOrdering produce testCase" $ do+ let+ consume = D.list+ produceAndConsume list1 list2 = C.produceAndConsume (E.mergeOrdering (\a b -> a <= b) (E.list list1) (E.list list2)) (consume)+ assertEqual "" [1,2,3,4,5,6,7,8,9,10] =<< produceAndConsume [1,3,5,7,9] [2,4,6,8,10]+ assertEqual "" [1,2,3,4,5,6,7,8,9,10] =<< produceAndConsume [2,4,6,8,10] [1,3,5,7,9]+ assertEqual "" [1,2,2,3,3,4,5,6,7,8,8,9,9,10] =<< produceAndConsume [1,2,3,8,9,10] [2,3,4,5,6,7,8,9]+ assertEqual "" [1,2,3,4,5,6,7,8,9,10] =<< produceAndConsume [1,2,3] [4,5,6,7,8,9,10]+ assertEqual "" [1,2,3,4,5,6,7,8,9,10] =<< produceAndConsume [4,5,6] [1,2,3,7,8,9,10]+ assertEqual "" [1,2,3,4,5,6,7,8,9,10] =<< produceAndConsume [1,2,3,7,8,9,10] [4,5,6]+ assertEqual "" [1,2,3,4,5,6,7,8,9,10] =<< produceAndConsume [1,2,3,4,5,6,7,8,9,10] []+ assertEqual "" [] =<< produceAndConsume ([] :: [Int]) ([] :: [Int]) + ,+ testProperty "mergeOrdering produce" $ \ (l1 :: [Int], l2 :: [Int]) -> + let + list1 = sort l1+ list2 = sort l2+ list = sort $ l1 ++ l2+ in list === unsafePerformIO (C.produceAndConsume (E.mergeOrdering (\a b -> a <= b) (E.list list1) (E.list list2)) D.list)+ ,+ transformPotoki+ ,+ parsingPotoki+ ,+ consumePotoki+ ]+++transformPotoki :: TestTree+transformPotoki =+ testGroup "Transform" $+ [+ testCase "Order" $ do+ let+ list = [Left 1, Left 2, Right 'z', Left 2, Right 'a', Left 1, Right 'b', Left 0, Right 'x', Left 4, Left 3]+ transform = left (A.consume (D.transform (A.take 2) D.sum))+ result <- C.produceAndConsume (E.list list) (D.transform transform D.list)+ assertEqual "" [Left 3, Right 'z', Left 2, Right 'a', Left 1, Right 'b', Left 0, Right 'x', Left 7] result+ ,+ testCase "Interrupted order" $ do+ let+ list = [Left 1, Left 2, Right 'a']+ transform = left (A.consume (D.transform (A.take 3) D.sum))+ result <- C.produceAndConsume (E.list list) (D.transform transform D.list)+ assertEqual "" [Left 3, Right 'a'] result+ ,+ testCase "Distinct" $ do+ let+ list = [1,2,3,2,3,2,1,4,1] :: [Int]+ 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] :: [Int]+ produce = E.list list+ transform =+ A.concurrently 12 $+ arr (\ x -> H.randomRIO (0, 100) >>= threadDelay >> return x) >>>+ A.executeIO+ consume = D.transform transform D.list+ result <- C.produceAndConsume produce consume+ assertBool "Is dispersed" (list /= result)+ assertEqual "Contains no duplicates" 0 (length result - length (nub result))+ assertEqual "Equals the original once sorted" list (sort result)+ ,+ testProperty "Line" $ \ chunks ->+ let+ expected =+ mconcat chunks+ actual =+ unsafePerformIO (C.produceAndConsume produce consume)+ where+ produce =+ E.list chunks+ consume =+ rmap (mconcat . intersperse "\n") $+ D.transform A.extractLines D.list+ in expected === actual+ ,+ testProperty "takeWhile" $ \ (list :: [Int]) ->+ let listPart = takeWhile odd list+ in monadicIO $ do+ let prod = E.list list+ res <- run (C.produceAndTransformAndConsume prod (A.takeWhile odd) D.list)+ M.assert (res == listPart)+ ,+ testProperty "mapFilter" $ \ (list :: [Int]) ->+ let in2MaybeOut input =+ if input `mod` 4 == 0+ then Just $ input `mod` 4+ else Nothing+ filteredList = map fromJust . filter (/= Nothing) . map in2MaybeOut $ list+ in monadicIO $ do+ let prod = E.list list+ res <- run (C.produceAndTransformAndConsume prod (A.mapFilter in2MaybeOut) D.list)+ M.assert (res == filteredList)+ ,+ testProperty "filter" $ \ (list :: [Int]) ->+ let filteredList = filter even list+ in monadicIO $ do+ let prod = E.list list+ res <- run (C.produceAndTransformAndConsume prod (A.filter even) D.list)+ M.assert (res == filteredList)+ ,+ testCase "uniquify" $ do+ let list = [1,1,2,3,4,5,5,5,6,7,8,9,10]+ res <- C.produceAndTransformAndConsume (E.list list) (A.uniquify) D.list+ assertEqual "" [1,2,3,4,5,6,7,8,9,10] res+ ,+ testProperty "uniquify unit test" $ \ (list :: [Int]) ->+ let sortList = sort list+ uniquifyList = nub sortList+ in uniquifyList === unsafePerformIO (C.produceAndTransformAndConsume (E.list sortList) (A.uniquify) D.list)+ ]++parsingPotoki :: TestTree+parsingPotoki =+ testGroup "Parsing" $+ [+ testCase "Sample 1" $ do+ let parser = B.double <* B.char ','+ transform = arr (either (const Nothing) Just) >>> A.just >>> A.parseBytes parser+ produce = E.transform transform (E.fileBytes "samples/1")+ result <- C.produceAndConsume produce D.count+ assertEqual "" 4350 result+ ,+ testCase "Sample 1 greedy" $ do+ let parser = B.sepBy B.double (B.char ',')+ transform = arr (either (const Nothing) Just) >>> A.just >>> A.parseBytes parser+ produce = E.transform transform (E.fileBytes "samples/1")+ result <- C.produceAndConsume produce D.list+ assertEqual "" [Right 4350] (fmap (fmap length) result)+ ,+ testCase "Split chunk" $+ let+ produce = E.list ["1", "2", "3"]+ parser = B.anyChar+ transform = A.parseBytes parser >>> arr (either (const Nothing) Just) >>> A.just+ consume = D.transform transform D.count+ in do+ assertEqual "" 3 =<< C.produceAndConsume produce consume+ ]++consumePotoki =+ testGroup "Consume" $+ [+ testProperty "count" $ \ (list :: [Int]) ->+ let n = length list+ in monadicIO $ do+ let prod = E.list list+ len <- run (C.produceAndConsume prod D.count)+ M.assert (len == n)+ ,+ testProperty "sum" $ \ (list :: [Int]) ->+ let n = sum list+ in monadicIO $ do+ let prod = E.list list+ len <- run (C.produceAndConsume prod D.sum)+ M.assert (len == n)+ ,+ testProperty "head" $ \ (list :: [Int]) ->+ let el = if null list then Nothing else (Just (head list))+ in monadicIO $ do+ let prod = E.list list+ he <- run (C.produceAndConsume prod D.head)+ M.assert (he == el)+ ,+ testProperty "last" $ \ (list :: [Int]) ->+ let el = if null list then Nothing else (Just (last list))+ in monadicIO $ do+ let prod = E.list list+ he <- run (C.produceAndConsume prod D.last)+ M.assert (he == el)+ ,+ testProperty "list" $ \ (list :: [Int]) ->+ monadicIO $ do+ let prod = E.list list+ res <- run (C.produceAndConsume prod D.list)+ M.assert (res == list)+ ,+ testProperty "reverseList" $ \ (list :: [Int]) ->+ let revList = reverse list+ in monadicIO $ do+ let prod = E.list list+ res <- run (C.produceAndConsume prod D.reverseList)+ M.assert (res == revList)+ ,+ testProperty "vector" $ \ (list :: [Int]) ->+ let vec = G.fromList list+ in monadicIO $ do+ let prod = E.list list+ res <- run (C.produceAndConsume prod D.vector)+ M.assert (res == vec)+ ,+ testProperty "concat" $ \ (list :: [[Int]]) ->+ let con = concat list+ in monadicIO $ do+ let prod = E.list list+ res <- run (C.produceAndConsume prod D.concat)+ M.assert (res == con)+ ,+ testProperty "fold" $ \ (list :: [Int], fun :: (Fun (Int, Int) Int), first :: Int) ->+ let f = applyFun2 fun+ fol = foldl' f first list+ in monadicIO $ do+ let prod = E.list list+ res <- run (C.produceAndConsume prod (D.fold (Fl.Fold f first id)))+ M.assert (res == fol)+ ,+ testProperty "foldInIO" $ \ (list :: [Int]) ->+ let fin = sum list+ in monadicIO $ do+ let prod = E.list list+ res <- run $ do+ sumVar <- newIORef 0+ (C.produceAndConsume prod (D.foldInIO (Fl.FoldM (\_ a -> modifyIORef' sumVar (a+)) (pure ()) (\_ -> readIORef sumVar)) ) )+ M.assert (res == fin)+ ,+ testProperty "folding" $ \ (list :: [Int], fun :: (Fun (Int, Int) Int), first :: Int) ->+ let f = applyFun2 fun+ fol = foldl' f first list+ in monadicIO $ do+ let prod = E.list list+ res <- run (C.produceAndConsume prod (D.folding (Fl.Fold f first id) (D.sum) ))+ a <- run (C.produceAndConsume prod D.sum)+ M.assert (res == (fol, a))+ ,+ testProperty "foldingInIO" $ \ (list :: [Int]) ->+ let fol = sum list+ in monadicIO $ do+ let prod = E.list list+ res <- run $ do+ sumVar <- newIORef 0+ (C.produceAndConsume prod (D.foldingInIO (Fl.FoldM (\_ a -> modifyIORef' sumVar (a+)) (pure ()) (\_ -> readIORef sumVar)) (D.sum) ))+ a <- run (C.produceAndConsume prod D.sum)+ M.assert (res == (fol, a))+ ,+ testProperty "execState" $ \ (list :: [Int]) ->+ let f = modify' . (+)+ resL = sum list+ in monadicIO $ do+ let prod = E.list list+ res <- run (C.produceAndConsume prod (D.execState f 0))+ M.assert (res == resL)+ ,+ testProperty "Choice consume right'" $ \ (list :: [Either Bool Int]) ->+ let n = sum <$> sequence list+ in monadicIO $ do+ let prod = E.list list+ len <- run (C.produceAndConsume prod $ right' D.sum)+ M.assert (len == n)+ ]+
+ test/Transform.hs view
@@ -0,0 +1,215 @@+module Transform where++import Prelude hiding (first, second, choose)+import Control.Arrow+import Test.QuickCheck.Instances+import Test.Tasty+import Test.Tasty.Runners+import Test.Tasty.HUnit+import Test.Tasty.QuickCheck+import qualified Potoki.Core.IO as C+import qualified Potoki.Core.Consume as D+import qualified Potoki.Core.Transform as A+import qualified Potoki.Core.Produce as E+import qualified Data.Attoparsec.ByteString.Char8 as B+import qualified Data.ByteString as F+import qualified Data.Vector as G+import qualified Data.List.Split as SplitList+import qualified System.Random as H++transform :: TestTree+transform =+ testGroup "Transform" $+ [+ testGroup "concurrentlyInOrder unit tests" $ let+ testList (list :: [[Int]]) =+ testCase (show list) $ do+ actual <- C.produceAndTransformAndConsume (E.list list) (A.concurrentlyInOrder 4 A.list) D.list+ assertEqual "" (concat list) actual+ in [+ testList [[]] ,+ testList [[],[]],+ testList [[0, 1, 2], []],+ testList [[0, 1, 2], [3]],+ testList [[0,3,7,-2],[],[],[-8,7,-8,7,3,-4,1,4],[-4,1],[5,-2,3,-2,-5,6,-7],[-8,6,7,1,8,-8,5]]+ ]+ ,+ testProperty "concurrentlyInOrder" $ \ (list :: [[Int]]) ->+ concat list ===+ unsafePerformIO (C.produceAndTransformAndConsume (E.list list) (A.concurrentlyInOrder 4 A.list) D.list)+ ,+ testProperty "list" $ \ (list :: [[Int]]) -> + concat list ===+ unsafePerformIO (C.produceAndTransformAndConsume (E.list list) A.list D.list)+ ,+ testProperty "drop" $ \ (list :: [Int], n :: Int) -> + (drop n list) ===+ unsafePerformIO (C.produceAndTransformAndConsume (E.list list) (A.drop n) D.list)+ ,+ testProperty "Applying chunksOf to list has the same effect as the \"batch\" transform" $ let+ gen = do+ list <- listOf (choose (0, 1000 :: Int))+ batchSize <- frequency [(1000, choose (1, 3)), (100, choose (4, 100)), (1, pure 0)]+ traceShowM (list, batchSize)+ return (list, batchSize)+ in forAll gen $ \ (list, batchSize) -> let+ listChunks = if batchSize < 1 then [] else SplitList.chunksOf batchSize list+ potokiChunks = unsafePerformIO $ C.produceAndTransformAndConsume (E.list list) (rmap toList (A.batch @Vector batchSize)) D.list+ in listChunks === potokiChunks+ ,+ transformProduce+ ,+ transformChoice+ ,+ transformArrowLaws+ ]++transformProduce =+ testGroup "Produce" $+ [+ testCase "1" $ do+ let+ list = [1, 2, 3] :: [Int]+ result <- C.produceAndTransformAndConsume+ (E.list list)+ (A.produce (E.list . \ n -> flip replicate n n))+ (D.list)+ assertEqual "" [1, 2, 2, 3, 3, 3] result+ ,+ testCase "2" $ do+ let+ list = [1, 2, 3] :: [Int]+ result <- C.produceAndTransformAndConsume+ (E.list list)+ (A.produce (E.list . \ n -> [(n, n)]))+ (D.list)+ assertEqual "" [(1, 1), (2, 2), (3, 3)] result+ ,+ testCase "3" $ do+ let+ list = [1, 2, 3] :: [Int]+ result <- C.produceAndTransformAndConsume+ (E.list list)+ (A.produce (E.list . \ n -> [n, n]))+ (D.list)+ assertEqual "" [1, 1, 2, 2, 3, 3] result+ ]++transformChoice =+ testGroup "Choice" $+ [+ testCase "1" $ do+ let+ list = [Left 1, Left 2, Right 'z', Left 2, Right 'a', Left 1, Right 'b', Left 0, Right 'x', Left 4, Left 3]+ transform = left' id+ result <- C.produceAndTransformAndConsume (E.list list) transform D.list+ assertEqual "" [Left 1, Left 2, Right 'z', Left 2, Right 'a', Left 1, Right 'b', Left 0, Right 'x', Left 4, Left 3] result+ ,+ testCase "2" $ do+ let+ list = [Left 1, Left 2, Right 'z', Right 'a', Right 'b', Left 0, Right 'x', Left 4, Left 3]+ transform = right' (A.consume D.list)+ result <- C.produceAndTransformAndConsume (E.list list) transform D.list+ assertEqual "" [Left 1, Left 2, Right "zab", Left 0, Right "x", Left 4, Left 3] result+ ,+ testCase "3" $ do+ let+ list = [Left 4, Right 'z', Right 'a', Left 3, Right 'b', Left 0, Left 1, Right 'x', Left 4, Left 3]+ transform = left' (A.consume D.list)+ result <- C.produceAndTransformAndConsume (E.list list) transform D.list+ assertEqual "" [Left [4], Right 'z', Right 'a', Left [3], Right 'b', Left [0, 1], Right 'x', Left [4, 3]] result+ ]++transformArrowLaws =+ testGroup "Arrow laws"+ [+ testGroup "Strong"+ [+ testCase "1" $ do+ let+ input = [(1,'a'),(2,'b'),(3,'c'),(4,'d')]+ transform = first transform1+ result <- C.produceAndTransformAndConsume (E.list input) transform D.list+ assertEqual "" [(6,'c'),(4,'d')] result+ ,+ testCase "Lack of elements" $ do+ let+ input = [(1,'a'),(2,'b')]+ transform = first transform1+ result <- C.produceAndTransformAndConsume (E.list input) transform D.list+ assertEqual "" [(3,'b')] result+ ]+ ,+ transformProperty "arr id = id"+ (arr id :: A.Transform Int Int)+ id+ ,+ transformProperty "arr (f >>> g) = arr f >>> arr g"+ (arr (f >>> g))+ (arr f >>> arr g)+ ,+ transformProperty "first (arr f) = arr (first f)"+ (first (arr f) :: A.Transform (Int, Char) (Int, Char))+ (arr (first f))+ ,+ transformProperty "first (f >>> g) = first f >>> first g"+ (first (transform1 >>> transform2) :: A.Transform (Int, Char) (Int, Char))+ (first (transform1) >>> first (transform2))+ ,+ transformProperty "first f >>> arr fst = arr fst >>> f"+ (first transform1 >>> arr fst :: A.Transform (Int, Char) Int)+ (arr fst >>> transform1)+ ,+ transformProperty "first f >>> arr (id *** g) = arr (id *** g) >>> first f"+ (first transform1 >>> arr (id *** g))+ (arr (id *** g) >>> first transform1)+ ,+ transformProperty "first (first f) >>> arr assoc = arr assoc >>> first f"+ (first (first transform1) >>> arr assoc :: A.Transform ((Int, Char), Double) (Int, (Char, Double)))+ (arr assoc >>> first transform1)+ ,+ transformProperty "left (arr f) = arr (left f)"+ (left (arr f) :: A.Transform (Either Int Char) (Either Int Char))+ (arr (left f))+ ,+ transformProperty "left (f >>> g) = left f >>> left g"+ (left (transform1 >>> transform2) :: A.Transform (Either Int Char) (Either Int Char))+ (left (transform1) >>> left (transform2))+ ,+ transformProperty "f >>> arr Left = arr Left >>> left f"+ (transform1 >>> arr Left :: A.Transform Int (Either Int Char))+ (arr Left >>> left transform1)+ ,+ transformProperty "left f >>> arr (id +++ g) = arr (id +++ g) >>> left f"+ (left transform1 >>> arr (id +++ g))+ (arr (id +++ g) >>> left transform1)+ ,+ transformProperty "left (left f) >>> arr assocsum = arr assocsum >>> left f"+ (left (left transform1) >>> arr assocsum :: A.Transform (Either (Either Int Char) Double) (Either Int (Either Char Double)))+ (arr assocsum >>> left transform1)+ ,+ transformProperty "left (left (arr f)) >>> arr assocsum = arr assocsum >>> left (arr f)"+ (left (left (arr f)) >>> arr assocsum :: A.Transform (Either (Either Int Char) Double) (Either Int (Either Char Double)))+ (arr assocsum >>> left (arr f))+ ]+ where+ f = (+24) :: Int -> Int+ g = (*3) :: Int -> Int+ transform1 = A.consume (D.transform (A.take 3) D.sum) :: A.Transform Int Int+ transform2 = A.consume (D.transform (A.take 4) D.sum) :: A.Transform Int Int+ assoc ((a,b),c) = (a,(b,c))+ assocsum (Left (Left x)) = Left x+ assocsum (Left (Right y)) = Right (Left y)+ assocsum (Right z) = Right (Right z)++transformProperty ::+ (Arbitrary input, Show input, Eq output, Show output) =>+ String -> A.Transform input output -> A.Transform input output -> TestTree+transformProperty name leftTransform rightTransform =+ testProperty name property+ where+ property list =+ transform leftTransform === transform rightTransform+ where+ transform transform =+ unsafePerformIO (C.produceAndTransformAndConsume (E.list list) transform D.list)
− tests/Main.hs
@@ -1,221 +0,0 @@-module Main where--import Prelude hiding (first, second)-import Control.Arrow-import Test.QuickCheck.Instances-import Test.QuickCheck.Monadic as M-import Test.Tasty-import Test.Tasty.Runners-import Test.Tasty.HUnit-import Test.Tasty.QuickCheck-import qualified Potoki.Core.IO as C-import qualified Potoki.Core.Consume as D-import qualified Potoki.Core.Transform as A-import qualified Potoki.Core.Produce as E-import qualified Potoki.Core.Fetch as Fe-import qualified Data.Attoparsec.ByteString.Char8 as B-import qualified Data.ByteString as F-import qualified Data.Vector as G-import qualified System.Random as H-import qualified Acquire.Acquire as Ac-import Potoki-import Transform--main =- defaultMain $- testGroup "All tests" $- [- testProperty "list to list" $ \ (list :: [Int]) ->- list === unsafePerformIO (C.produceAndConsume (E.list list) D.list)- ,- testProperty "consecutive consumers" $ \ (list :: [Int], amount) ->- list === unsafePerformIO (C.produceAndConsume (E.list list) ((++) <$> D.transform (A.take amount) D.list <*> D.list))- ,- potoki- ,- transform- ,- resourceChecker- ,- testCase "sync resource checker" $ do- resourceVar <- newIORef Initial- let produce = syncCheckResource resourceVar- C.produceAndConsume produce D.sum- fin <- readIORef resourceVar- assertEqual "" Released fin- ,- testCase "async resource checker" $ do- resourceVar <- newTVarIO Initial- let produce = asyncCheckResource resourceVar- potokiThreadId <- forkIO $ do- C.produceAndConsume produce D.sum- return ()- atomically $ do- resource <- readTVar resourceVar- guard $ resource == Acquired- killThread potokiThreadId- atomically $ do- resource <- readTVar resourceVar- guard $ resource == Released- ,- testProperty "Produce.transform resource checker" $ \ (list :: [Int]) ->- let prod = E.list list- in monadicIO $ do- check <- run $ do- resourceVar1 <- newIORef Initial- res <- C.produceAndConsume (E.transform (checkTransform resourceVar1) prod) D.sum- readIORef resourceVar1- M.assert $ check == Released- ,- testProperty "Consume.transform resource checker" $ \ (list :: [Int]) ->- let prod = E.list list- in monadicIO $ do- check <- run $ do- resourceVar1 <- newIORef Initial- res <- C.produceAndConsume prod (D.transform (checkTransform resourceVar1) D.sum)- readIORef resourceVar1- M.assert $ check == Released- ,- testCase "Transform.produce resource checker #1" $ do- resourceVar <- newIORef Initial- let prod = checkProduce resourceVar (/= Released) 100- res1 <- C.produceAndConsume (E.transform (A.produce intToProduce) prod) D.sum- res2 <- C.produceAndConsume prod (D.transform (A.produce intToProduce) D.sum)- fin <- readIORef resourceVar- assertEqual "" Released fin- assertEqual "" res1 res2- ,- testCase "Transform.produce resource checker #2" $ do- resourceVar <- newIORef Initial- let prod = checkProduce resourceVar (/= Released) 100- res1 <- C.produceAndConsume (E.transform (A.take 5 >>> A.produce intToProduce) prod) D.sum- res2 <- C.produceAndConsume prod (D.transform (A.take 5 >>> A.produce intToProduce) D.sum)- fin <- readIORef resourceVar- assertEqual "" Released fin- assertEqual "" res1 res2- ]--resourceChecker :: TestTree-resourceChecker =- testGroup "produce1 >>= produce2" $- [- testCase "Check produce binding" $ do- resourceVar1 <- newIORef Initial- resourceVar2 <- newIORef Initial- let prod1 = checkProduce resourceVar1 (const True) 100- prod2 = \x -> checkProduce resourceVar2 (/= Released) x- res <- C.produceAndConsume (prod1 >>= prod2) D.sum- fin <- readIORef resourceVar1- assertEqual "" Released fin- ,- testProperty "Bind for produce" $ \ (list :: [Int]) ->- let check = list >>= (enumFromTo 0)- prod1 = E.list list- prod2 = \x -> E.list $ enumFromTo 0 x- in monadicIO $ do- res <- run $ C.produceAndConsume (prod1 >>= prod2) D.list- M.assert (check == res)- ,- testProperty "liftIO for Produce. Consume0" $ \ (_ :: Int) ->- monadicIO $ do- check <- run $ do- checkVar <- newIORef False- let prod = liftIO $ writeIORef checkVar True- C.produceAndConsume prod someThing- readIORef checkVar- M.assert $ check == False- ,- testProperty "liftIO for Produce. ConsumeN" $ \ (n :: Int) ->- monadicIO $ do- let prod = liftIO (return n)- len <- run (C.produceAndConsume prod D.count)- M.assert (len == 1)- ]--intToProduce :: Int -> E.Produce Int-intToProduce a = E.Produce . Ac.Acquire $ do- stVar <- newIORef 0- return $ flip (,) (return ()) $ Fe.Fetch $ do- n <- readIORef stVar- if n >= a + 1 then (return Nothing)- else do- writeIORef stVar $! n + 1- return (Just n)--someThing :: D.Consume input Int-someThing = D.Consume $ \ (Fe.Fetch _) -> return 0--single :: Foldable f => f a -> Maybe a-single = join . (foldr' f Nothing)- where- f x Nothing = Just $ Just x- f _ _ = Just Nothing--data Resource- = Initial- | Acquired- | Released- | AcquiredImproperly- | ReleasedImproperly- deriving (Show, Eq)--checkTransform :: IORef Resource -> A.Transform Int Int-checkTransform resourceVar = A.Transform $ \ fetchIO -> Ac.Acquire $ do- writeIORef resourceVar Acquired- return $ (,) (plusFetch fetchIO) $ do- res <- readIORef resourceVar- case res of- Acquired -> writeIORef resourceVar Released- _ -> writeIORef resourceVar ReleasedImproperly--plusFetch :: Fe.Fetch Int -> Fe.Fetch Int-plusFetch (Fe.Fetch fetchIO) = Fe.Fetch $ do- fetch <- fetchIO- return $ fmap succ fetch--checkProduce :: IORef Resource -> (Resource -> Bool) -> Int -> E.Produce Int-checkProduce resourceVar f k = E.Produce . Ac.Acquire $ do- res <- readIORef resourceVar- if f res && res /= Initial- then do- return $ (,) (Fe.Fetch $ return Nothing) $ writeIORef resourceVar AcquiredImproperly- else do- writeIORef resourceVar Acquired- stVar <- newIORef 0- let fetch = do- n <- readIORef stVar- if n >= k then return (Nothing)- else do- writeIORef stVar $! n + 1- return (Just n)- return $ (,) (Fe.Fetch fetch) $ do- res <- readIORef resourceVar- case res of- Acquired -> writeIORef resourceVar Released- _ -> writeIORef resourceVar ReleasedImproperly--asyncCheckResource :: TVar Resource -> E.Produce Int-asyncCheckResource resourceVar = E.Produce . Ac.Acquire $ do- atomically $ writeTVar resourceVar Acquired- stVar <- newIORef 0- let fetch = do- n <- readIORef stVar- writeIORef stVar $! n + 1- return (Just n)- return (Fe.Fetch fetch, atomically $ writeTVar resourceVar Released)--syncCheckResource :: IORef Resource -> E.Produce Int-syncCheckResource resourceVar = E.Produce . Ac.Acquire $ do- writeIORef resourceVar Acquired- stVar <- newIORef 0- let fetch = do- n <- readIORef stVar- if n >= 1000 then return (Nothing)- else do- writeIORef stVar $! n + 1- return (Just n)- return $ (,) (Fe.Fetch fetch) $ do- res <- readIORef resourceVar- case res of- Acquired -> writeIORef resourceVar Released- _ -> writeIORef resourceVar ReleasedImproperly
− tests/Potoki.hs
@@ -1,279 +0,0 @@-module Potoki where--import Prelude hiding (first, second)-import Control.Arrow-import Test.QuickCheck.Instances-import Test.QuickCheck.Monadic as M-import Test.Tasty-import Test.Tasty.Runners-import Test.Tasty.HUnit-import Test.Tasty.QuickCheck-import qualified Control.Foldl as Fl-import qualified Potoki.Core.IO as C-import qualified Potoki.Core.Consume as D-import qualified Potoki.Core.Transform as A-import qualified Potoki.Core.Produce as E-import qualified Data.Attoparsec.ByteString.Char8 as B-import qualified Data.ByteString as F-import qualified Data.Vector as G-import qualified System.Random as H-import Data.List.Index (indexed)--potoki :: TestTree-potoki =- testGroup "All tests for potoki's end-users functions" $- [- testCase "vector to list" $ do- result <- C.produceAndConsume (E.vector (G.fromList [1,2,3])) (D.list)- assertEqual "" [1,2,3] result- ,- testCase "just" $ do- result <- C.produceAndConsume (E.list [Just 1, Nothing, Just 2]) (D.transform A.just D.list)- assertEqual "" [1,2] result- ,- testCase "transform,consume,take" $ do- let- transform = A.consume (D.transform (A.take 3) D.list)- consume = D.transform transform D.list- produceAndConsume list = C.produceAndConsume (E.list list) (consume)- assertEqual "" [[1,2,3], [4,5,6], [7,8]] =<< produceAndConsume [1,2,3,4,5,6,7,8]- assertEqual "" [[1,2,3], [4,5,6], [7,8,9]] =<< produceAndConsume [1,2,3,4,5,6,7,8,9]- assertEqual "" [] =<< produceAndConsume ([] :: [Int])- ,- testCase "File reading" $ do- let produce =- E.transform (arr (either (const Nothing) Just) >>> A.just) $- E.fileBytes "samples/1"- result <- C.produceAndConsume produce (fmap F.length D.concat)- assertEqual "" 17400 result- ,- transformPotoki- ,- parsingPotoki- ,- consumePotoki- ]---transformPotoki :: TestTree-transformPotoki =- testGroup "Transform" $- [- testCase "Order" $ do- let- list = [Left 1, Left 2, Right 'z', Left 2, Right 'a', Left 1, Right 'b', Left 0, Right 'x', Left 4, Left 3]- transform = left (A.consume (D.transform (A.take 2) D.sum))- result <- C.produceAndConsume (E.list list) (D.transform transform D.list)- assertEqual "" [Left 3, Right 'z', Left 2, Right 'a', Left 1, Right 'b', Left 0, Right 'x', Left 7] result- ,- testCase "Interrupted order" $ do- let- list = [Left 1, Left 2, Right 'a']- transform = left (A.consume (D.transform (A.take 3) D.sum))- result <- C.produceAndConsume (E.list list) (D.transform transform D.list)- assertEqual "" [Left 3, Right 'a'] result- ,- testCase "Distinct" $ do- let- list = [1,2,3,2,3,2,1,4,1] :: [Int]- 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]- produce = E.list list- transform =- A.concurrently 12 $- arr (\ x -> H.randomRIO (0, 100) >>= threadDelay >> return x) >>>- A.executeIO- consume = D.transform transform D.list- result <- C.produceAndConsume produce consume- assertBool "Is dispersed" (list /= result)- assertEqual "Contains no duplicates" 0 (length result - length (nub result))- assertEqual "Equals the original once sorted" list (sort result)- ,- testProperty "Line" $ \ chunks ->- let- expected =- mconcat chunks- actual =- unsafePerformIO (C.produceAndConsume produce consume)- where- produce =- E.list chunks- consume =- rmap (mconcat . intersperse "\n") $- D.transform A.extractLines D.list- in expected === actual- ,- testProperty "takeWhile" $ \ (list :: [Int]) ->- let listPart = takeWhile odd list- in monadicIO $ do- let prod = E.list list- res <- run (C.produceAndTransformAndConsume prod (A.takeWhile odd) D.list)- M.assert (res == listPart)- ,- testProperty "mapFilter" $ \ (list :: [Int]) ->- let in2MaybeOut input =- if input `mod` 4 == 0- then Just $ input `mod` 4- else Nothing- filteredList = map fromJust . filter (/= Nothing) . map in2MaybeOut $ list- in monadicIO $ do- let prod = E.list list- res <- run (C.produceAndTransformAndConsume prod (A.mapFilter in2MaybeOut) D.list)- M.assert (res == filteredList)- ,- testProperty "filter" $ \ (list :: [Int]) ->- let filteredList = filter even list- in monadicIO $ do- let prod = E.list list- res <- run (C.produceAndTransformAndConsume prod (A.filter even) D.list)- M.assert (res == filteredList)- ]--parsingPotoki :: TestTree-parsingPotoki =- testGroup "Parsing" $- [- testCase "Sample 1" $ do- let parser = B.double <* B.char ','- transform = arr (either (const Nothing) Just) >>> A.just >>> A.parseBytes parser- produce = E.transform transform (E.fileBytes "samples/1")- result <- C.produceAndConsume produce D.count- assertEqual "" 4350 result- ,- testCase "Sample 1 greedy" $ do- let parser = B.sepBy B.double (B.char ',')- transform = arr (either (const Nothing) Just) >>> A.just >>> A.parseBytes parser- produce = E.transform transform (E.fileBytes "samples/1")- result <- C.produceAndConsume produce D.list- assertEqual "" [Right 4350] (fmap (fmap length) result)- ,- testCase "Split chunk" $- let- produce = E.list ["1", "2", "3"]- parser = B.anyChar- transform = A.parseBytes parser >>> arr (either (const Nothing) Just) >>> A.just- consume = D.transform transform D.count- in do- assertEqual "" 3 =<< C.produceAndConsume produce consume- ]--consumePotoki =- testGroup "Consume" $- [- testProperty "count" $ \ (list :: [Int]) ->- let n = length list- in monadicIO $ do- let prod = E.list list- len <- run (C.produceAndConsume prod D.count)- M.assert (len == n)- ,- testProperty "sum" $ \ (list :: [Int]) ->- let n = sum list- in monadicIO $ do- let prod = E.list list- len <- run (C.produceAndConsume prod D.sum)- M.assert (len == n)- ,- testProperty "head" $ \ (list :: [Int]) ->- let el = if null list then Nothing else (Just (head list))- in monadicIO $ do- let prod = E.list list- he <- run (C.produceAndConsume prod D.head)- M.assert (he == el)- ,- testProperty "last" $ \ (list :: [Int]) ->- let el = if null list then Nothing else (Just (last list))- in monadicIO $ do- let prod = E.list list- he <- run (C.produceAndConsume prod D.last)- M.assert (he == el)- ,- testProperty "list" $ \ (list :: [Int]) ->- monadicIO $ do- let prod = E.list list- res <- run (C.produceAndConsume prod D.list)- M.assert (res == list)- ,- testProperty "reverseList" $ \ (list :: [Int]) ->- let revList = reverse list- in monadicIO $ do- let prod = E.list list- res <- run (C.produceAndConsume prod D.reverseList)- M.assert (res == revList)- ,- testProperty "vector" $ \ (list :: [Int]) ->- let vec = G.fromList list- in monadicIO $ do- let prod = E.list list- res <- run (C.produceAndConsume prod D.vector)- M.assert (res == vec)- ,- testProperty "concat" $ \ (list :: [[Int]]) ->- let con = concat list- in monadicIO $ do- let prod = E.list list- res <- run (C.produceAndConsume prod D.concat)- M.assert (res == con)- ,- testProperty "fold" $ \ (list :: [Int], fun :: (Fun (Int, Int) Int), first :: Int) ->- let f = applyFun2 fun- fol = foldl' f first list- in monadicIO $ do- let prod = E.list list- res <- run (C.produceAndConsume prod (D.fold (Fl.Fold f first id)))- M.assert (res == fol)- ,- testProperty "foldInIO" $ \ (list :: [Int]) ->- let fin = sum list- in monadicIO $ do- let prod = E.list list- res <- run $ do- sumVar <- newIORef 0- (C.produceAndConsume prod (D.foldInIO (Fl.FoldM (\_ a -> modifyIORef' sumVar (a+)) (pure ()) (\_ -> readIORef sumVar)) ) )- M.assert (res == fin)- ,- testProperty "folding" $ \ (list :: [Int], fun :: (Fun (Int, Int) Int), first :: Int) ->- let f = applyFun2 fun- fol = foldl' f first list- in monadicIO $ do- let prod = E.list list- res <- run (C.produceAndConsume prod (D.folding (Fl.Fold f first id) (D.sum) ))- a <- run (C.produceAndConsume prod D.sum)- M.assert (res == (fol, a))- ,- testProperty "foldingInIO" $ \ (list :: [Int]) ->- let fol = sum list- in monadicIO $ do- let prod = E.list list- res <- run $ do- sumVar <- newIORef 0- (C.produceAndConsume prod (D.foldingInIO (Fl.FoldM (\_ a -> modifyIORef' sumVar (a+)) (pure ()) (\_ -> readIORef sumVar)) (D.sum) ))- a <- run (C.produceAndConsume prod D.sum)- M.assert (res == (fol, a))- ,- testProperty "execState" $ \ (list :: [Int]) ->- let f = modify' . (+)- resL = sum list- in monadicIO $ do- let prod = E.list list- res <- run (C.produceAndConsume prod (D.execState f 0))- M.assert (res == resL)- ,- testProperty "Choice consume right'" $ \ (list :: [Either Bool Int]) ->- let n = sum <$> sequence list- in monadicIO $ do- let prod = E.list list- len <- run (C.produceAndConsume prod $ right' D.sum)- M.assert (len == n)- ]-
− tests/Transform.hs
@@ -1,178 +0,0 @@-module Transform where--import Prelude hiding (first, second)-import Control.Arrow-import Test.QuickCheck.Instances-import Test.Tasty-import Test.Tasty.Runners-import Test.Tasty.HUnit-import Test.Tasty.QuickCheck-import qualified Potoki.Core.IO as C-import qualified Potoki.Core.Consume as D-import qualified Potoki.Core.Transform as A-import qualified Potoki.Core.Produce as E-import qualified Data.Attoparsec.ByteString.Char8 as B-import qualified Data.ByteString as F-import qualified Data.Vector as G-import qualified System.Random as H--transform :: TestTree-transform =- testGroup "Transform" $- [- transformProduce- ,- transformChoice- ,- transformArrowLaws- ]--transformProduce =- testGroup "Produce" $- [- testCase "1" $ do- let- list = [1, 2, 3] :: [Int]- result <- C.produceAndTransformAndConsume- (E.list list)- (A.produce (E.list . \ n -> flip replicate n n))- (D.list)- assertEqual "" [1, 2, 2, 3, 3, 3] result- ,- testCase "2" $ do- let- list = [1, 2, 3] :: [Int]- result <- C.produceAndTransformAndConsume- (E.list list)- (A.produce (E.list . \ n -> [(n, n)]))- (D.list)- assertEqual "" [(1, 1), (2, 2), (3, 3)] result- ,- testCase "3" $ do- let- list = [1, 2, 3] :: [Int]- result <- C.produceAndTransformAndConsume- (E.list list)- (A.produce (E.list . \ n -> [n, n]))- (D.list)- assertEqual "" [1, 1, 2, 2, 3, 3] result- ]--transformChoice =- testGroup "Choice" $- [- testCase "1" $ do- let- list = [Left 1, Left 2, Right 'z', Left 2, Right 'a', Left 1, Right 'b', Left 0, Right 'x', Left 4, Left 3]- transform = left' id- result <- C.produceAndTransformAndConsume (E.list list) transform D.list- assertEqual "" [Left 1, Left 2, Right 'z', Left 2, Right 'a', Left 1, Right 'b', Left 0, Right 'x', Left 4, Left 3] result- ,- testCase "2" $ do- let- list = [Left 1, Left 2, Right 'z', Right 'a', Right 'b', Left 0, Right 'x', Left 4, Left 3]- transform = right' (A.consume D.list)- result <- C.produceAndTransformAndConsume (E.list list) transform D.list- assertEqual "" [Left 1, Left 2, Right "zab", Left 0, Right "x", Left 4, Left 3] result- ,- testCase "3" $ do- let- list = [Left 4, Right 'z', Right 'a', Left 3, Right 'b', Left 0, Left 1, Right 'x', Left 4, Left 3]- transform = left' (A.consume D.list)- result <- C.produceAndTransformAndConsume (E.list list) transform D.list- assertEqual "" [Left [4], Right 'z', Right 'a', Left [3], Right 'b', Left [0, 1], Right 'x', Left [4, 3]] result- ]--transformArrowLaws =- testGroup "Arrow laws"- [- testGroup "Strong"- [- testCase "1" $ do- let- input = [(1,'a'),(2,'b'),(3,'c'),(4,'d')]- transform = first transform1- result <- C.produceAndTransformAndConsume (E.list input) transform D.list- assertEqual "" [(6,'c'),(4,'d')] result- ,- testCase "Lack of elements" $ do- let- input = [(1,'a'),(2,'b')]- transform = first transform1- result <- C.produceAndTransformAndConsume (E.list input) transform D.list- assertEqual "" [(3,'b')] result- ]- ,- transformProperty "arr id = id"- (arr id :: A.Transform Int Int)- id- ,- transformProperty "arr (f >>> g) = arr f >>> arr g"- (arr (f >>> g))- (arr f >>> arr g)- ,- transformProperty "first (arr f) = arr (first f)"- (first (arr f) :: A.Transform (Int, Char) (Int, Char))- (arr (first f))- ,- transformProperty "first (f >>> g) = first f >>> first g"- (first (transform1 >>> transform2) :: A.Transform (Int, Char) (Int, Char))- (first (transform1) >>> first (transform2))- ,- transformProperty "first f >>> arr fst = arr fst >>> f"- (first transform1 >>> arr fst :: A.Transform (Int, Char) Int)- (arr fst >>> transform1)- ,- transformProperty "first f >>> arr (id *** g) = arr (id *** g) >>> first f"- (first transform1 >>> arr (id *** g))- (arr (id *** g) >>> first transform1)- ,- transformProperty "first (first f) >>> arr assoc = arr assoc >>> first f"- (first (first transform1) >>> arr assoc :: A.Transform ((Int, Char), Double) (Int, (Char, Double)))- (arr assoc >>> first transform1)- ,- transformProperty "left (arr f) = arr (left f)"- (left (arr f) :: A.Transform (Either Int Char) (Either Int Char))- (arr (left f))- ,- transformProperty "left (f >>> g) = left f >>> left g"- (left (transform1 >>> transform2) :: A.Transform (Either Int Char) (Either Int Char))- (left (transform1) >>> left (transform2))- ,- transformProperty "f >>> arr Left = arr Left >>> left f"- (transform1 >>> arr Left :: A.Transform Int (Either Int Char))- (arr Left >>> left transform1)- ,- transformProperty "left f >>> arr (id +++ g) = arr (id +++ g) >>> left f"- (left transform1 >>> arr (id +++ g))- (arr (id +++ g) >>> left transform1)- ,- transformProperty "left (left f) >>> arr assocsum = arr assocsum >>> left f"- (left (left transform1) >>> arr assocsum :: A.Transform (Either (Either Int Char) Double) (Either Int (Either Char Double)))- (arr assocsum >>> left transform1)- ,- transformProperty "left (left (arr f)) >>> arr assocsum = arr assocsum >>> left (arr f)"- (left (left (arr f)) >>> arr assocsum :: A.Transform (Either (Either Int Char) Double) (Either Int (Either Char Double)))- (arr assocsum >>> left (arr f))- ]- where- f = (+24) :: Int -> Int- g = (*3) :: Int -> Int- transform1 = A.consume (D.transform (A.take 3) D.sum) :: A.Transform Int Int- transform2 = A.consume (D.transform (A.take 4) D.sum) :: A.Transform Int Int- assoc ((a,b),c) = (a,(b,c))- assocsum (Left (Left x)) = Left x- assocsum (Left (Right y)) = Right (Left y)- assocsum (Right z) = Right (Right z)--transformProperty ::- (Arbitrary input, Show input, Eq output, Show output) =>- String -> A.Transform input output -> A.Transform input output -> TestTree-transformProperty name leftTransform rightTransform =- testProperty name property- where- property list =- transform leftTransform === transform rightTransform- where- transform transform =- unsafePerformIO (C.produceAndTransformAndConsume (E.list list) transform D.list)