potoki-core 2.2.8 → 2.2.8.2
raw patch · 10 files changed
+203/−137 lines, 10 files
Files
- library/Potoki/Core/Consume.hs +25/−3
- library/Potoki/Core/Fetch.hs +24/−22
- library/Potoki/Core/IO.hs +4/−3
- library/Potoki/Core/IO/Fetch.hs +7/−5
- library/Potoki/Core/Transform.hs +1/−0
- library/Potoki/Core/Transform/Basic.hs +64/−61
- library/Potoki/Core/Transform/ByteString.hs +23/−21
- library/Potoki/Core/Transform/Concurrency.hs +51/−20
- potoki-core.cabal +1/−1
- tests/Main.hs +3/−1
library/Potoki/Core/Consume.hs view
@@ -17,6 +17,7 @@ foldingInIO, execState, writeBytesToFile,+ writeBytesToFileWithoutBuffering, appendBytesToFile, deleteFiles, printBytes,@@ -206,11 +207,32 @@ -} {-# 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.
library/Potoki/Core/Fetch.hs view
@@ -214,36 +214,38 @@ {-# 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+ 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+ 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
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/Transform.hs view
@@ -32,6 +32,7 @@ A.parseNonEmptyLineBytesConcurrently, -- * Concurrency N.bufferize,+ N.bufferizeFlushing, N.concurrently, N.unsafeConcurrently, N.async,
library/Potoki/Core/Transform/Basic.hs view
@@ -34,39 +34,39 @@ drop amount = Transform $ \ (A.Fetch fetchIO) -> M.Acquire $ do countRef <- newIORef amount- return $ (, return ()) $ - A.Fetch $ fix $ \ doLoop -> do+ return $ (, return ()) $ A.Fetch $ let+ loop = do count <- readIORef countRef if count > 0 then do writeIORef countRef $! pred count- doLoop+ loop else fetchIO+ in loop {-# INLINE list #-} list :: Transform [a] a list = Transform $ \ (A.Fetch fetchListIO) -> M.Acquire $ do bufferRef <- newIORef []- return $ (, return ()) $ - A.Fetch $ 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+ return $ (, return ()) $ A.Fetch $ 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 {-# INLINABLE vector #-} vector :: Transform (Vector a) a@@ -74,8 +74,8 @@ Transform $ \ (A.Fetch fetchVectorIO) -> M.Acquire $ do indexRef <- newIORef 0 vectorRef <- newIORef mempty- return $ (, return ()) $ - A.Fetch $ fix $ \ doLoop -> do+ return $ (, return ()) $ A.Fetch $ let+ loop = do vectorVal <- readIORef vectorRef indexVal <- readIORef indexRef if indexVal < P.length vectorVal@@ -86,26 +86,27 @@ Just vectorVal' -> do writeIORef vectorRef vectorVal' writeIORef indexRef 0- doLoop+ loop Nothing -> 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@@ -138,7 +139,7 @@ mapInIOWithCounter handler = ioTransform $ do counter <- newIORef 0- return $ mapInIO $ \ a -> do+ return $ mapInIO $ \ !a -> do count <- atomicModifyIORef' counter (\ n -> (succ n, n)) handler count a @@ -191,29 +192,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
library/Potoki/Core/Transform/ByteString.hs view
@@ -65,25 +65,27 @@ lineList = Transform $ \ (A.Fetch fetchIO) -> M.Acquire $ do pokingRef <- newIORef mempty- return $ (, return ()) $ A.Fetch $ fetchIO >>= \ case- Just chunk -> case B.split 10 chunk of- head : tail -> do+ return $ (, return ()) $ A.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- 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 return (Just (bytes : []))+ if C.null poking+ then return Nothing+ else let+ !bytes = D.poking poking+ in return (Just (bytes : []))
library/Potoki/Core/Transform/Concurrency.hs view
@@ -8,6 +8,32 @@ import qualified Acquire.Acquire as M +bufferizeFlushing :: Int -> Transform input [input]+bufferizeFlushing maxSize =+ Transform $ \ (A.Fetch fetchIO) -> liftIO $ do+ buffer <- newTBQueueIO 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 size =@@ -15,13 +41,15 @@ buffer <- newTBQueueIO size activeVar <- newTVarIO True - forkIO $ fix $ \ loop -> do- fetchingResult <- fetchIO- case fetchingResult of- Just !element -> do- atomically $ writeTBQueue buffer element- loop- Nothing -> atomically $ writeTVar activeVar False+ 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 $ let readBuffer = Just <$> readTBQueue buffer@@ -79,14 +107,15 @@ replicateM_ workersAmount $ forkIO $ do (A.Fetch fetchIO, finalize) <- case syncTransformIO fetchIO of M.Acquire io -> io- fix $ \ loop -> do- fetchResult <- fetchIO- case fetchResult of- Just !result -> do- atomically (writeTBQueue chan result)- loop- Nothing -> atomically (modifyTVar' workersCounter pred)- finalize+ let+ loop = do+ fetchResult <- fetchIO+ case fetchResult of+ Just !result -> do+ atomically (writeTBQueue chan result)+ loop+ Nothing -> atomically (modifyTVar' workersCounter pred)+ in loop *> finalize return $ A.Fetch $ let readChan = Just <$> readTBQueue chan@@ -106,11 +135,13 @@ chan <- atomically newEmptyTMVar workersCounter <- atomically (newTVar workersAmount) - replicateM_ workersAmount $ forkIO $ fix $ \ loop -> do- fetchResult <- fetchIO- case fetchResult of- Just input -> atomically (putTMVar chan input) *> loop- Nothing -> atomically (modifyTVar' workersCounter pred)+ 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
potoki-core.cabal view
@@ -1,5 +1,5 @@ name: potoki-core-version: 2.2.8+version: 2.2.8.2 synopsis: Low-level components of "potoki" description: Provides everything required for building custom instances of
tests/Main.hs view
@@ -25,9 +25,11 @@ defaultMain $ testGroup "All tests" $ [- testCase "extractLines" $ do+ testCase "extractLinesWithoutTrail" $ do assertEqual "" ["ab", "", "cd"] =<< C.produceAndConsume (E.transform A.extractLinesWithoutTrail (E.list ["a", "b\n", "\nc", "d\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 ,