diff --git a/.gitignore b/.gitignore new file mode 100644 --- /dev/null +++ b/.gitignore @@ -0,0 +1,22 @@ +dist +dist-* +cabal-dev +*.o +*.hi +*.chi +*.chs.h +*.dyn_o +*.dyn_hi +.hpc +.hsenv +.cabal-sandbox/ +cabal.sandbox.config +*.prof +*.aux +*.hp +*.eventlog +.stack-work/ +cabal.project.local +cabal.project.local~ +.HTF/ +.ghc.environment.* diff --git a/.travis.yml b/.travis.yml new file mode 100644 --- /dev/null +++ b/.travis.yml @@ -0,0 +1,3 @@ +language: haskell +ghc: + - "8.4.1" diff --git a/ChangeLog.md b/ChangeLog.md new file mode 100644 --- /dev/null +++ b/ChangeLog.md diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2018 Daniel Díaz Carrete + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,65 @@ +# streaming-bracketed + +[![Build Status](https://travis-ci.org/danidiaz/streaming-bracketed.svg?branch=master)](https://travis-ci.org/danidiaz/streaming-bracketed) + +## What's this? + +A resource management "decorator" for the `Stream` type from +[streaming](http://hackage.haskell.org/package/streaming). + +The idea is that the `Bracketed` type represents a `Stream` which might have +some finalizers that will be triggered when we reach a given point in the +stream. + +By being careful about how we lift operations to work on `Bracketed` streams, +we can ensure that finalizers are promptly called even with operations like +`take`. + +`Bracketed` streams are ultimately consumed by using a continuation. + +## Differences with resourcet + +[resourcet](http://hackage.haskell.org/package/resourcet) is a widely used +library for resource handling. It provides a monad transformer over IO that +keeps track of registered resources and ensures proper cleanup. + +The main differences with the present library are: + +- This library only works on `Stream`s from streaming. + +- `Bracketed` sits above the streaming monad, not below like `ResourceT`. + +- This library aims to provide smarter handling of stream functions like + `take`, without too much hassle. + +- In this library finalizer scopes are nested, unlike `ResourceT` which allows + arbitrary interleavings. + +## Doubts + +- Lifting functions like `splitAt` might cause problems if we try to use the + rest of the stream. + +## Motivation + +From the +[CHANGELOG](http://hackage.haskell.org/package/streaming-0.2.1.0/changelog) of +the [streaming](http://hackage.haskell.org/package/streaming) package: + +> Remove bracketStream, MonadCatch instance, and everything dealing with +> ResourceT. All of these things of sort of broken for Stream since there is no +> guarantee of linear consumption (functions like take can prevent finalizers +> from running). + +[One Github issue](https://github.com/haskell-streaming/streaming/issues/52). + +[Another one](https://github.com/haskell-streaming/streaming-with/issues/2). + +[Streaming libs exercise/challenge](https://twitter.com/DiazCarrete/status/1016073374458671104): + +> Given a list [(Filepath,Int,Int)] of files and line ranges, create a stream +> of lines belonging to the concatenated ranges. + +> Prompt release of file handles is required. resource-handling monads and +> "withXXX"-style functions are allowed. + diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/library/Streaming/Bracketed.hs b/library/Streaming/Bracketed.hs new file mode 100644 --- /dev/null +++ b/library/Streaming/Bracketed.hs @@ -0,0 +1,58 @@ +{-| A resource management decorator for `Stream`s. + + Resource-allocating 'Stream's are lifted to values of type `Bracketed` that + can be combined as such, using an API that is more restricted than + that of the original 'Stream's and ensures prompt deallocation of + resources. + + Values of type `Bracketed` can later be run by supplying a + 'Stream'-consumer continuation to the 'with' function. + +>>> :{ + do -- boring setup stuff for a two-line text file + path <- ( "streaming-bracketed-doctest.txt") <$> getTemporaryDirectory + exists <- doesPathExist path + when exists (removeFile path) + withFile path WriteMode (for_ ["aaa","bbb"] . hPutStrLn) + -- end of setup + let lineStream = R.linesFromFile utf8 nativeNewlineMode path + lines :> () <- R.with (R.over_ (S.take 1) lineStream *> R.over (S.map (map succ)) lineStream) + S.toList + return lines + :} +["aaa","bbb","ccc"] + +-} +module Streaming.Bracketed ( + -- * Bracketed + Bracketed + -- * Lifting streams + , clear + , bracketed + -- * Consuming bracketed streams with continuations + , with + , with_ + -- * Transforming bracketed streams + , over + , over_ + , for + -- * Reading text files + , linesFromFile + , concatRanges + ) where + +import Streaming +import Streaming.Bracketed.Internal + +{- $setup + +>>> import Data.Foldable +>>> import Control.Monad +>>> import System.IO +>>> import System.FilePath +>>> import System.Directory +>>> import Streaming +>>> import qualified Streaming.Prelude as S +>>> import qualified Streaming.Bracketed as R + +-} diff --git a/library/Streaming/Bracketed/Internal.hs b/library/Streaming/Bracketed/Internal.hs new file mode 100644 --- /dev/null +++ b/library/Streaming/Bracketed/Internal.hs @@ -0,0 +1,143 @@ +{-# LANGUAGE DeriveFunctor #-} +{-# LANGUAGE RankNTypes #-} +module Streaming.Bracketed.Internal where + +import Data.Foldable +import Data.Bifunctor +import Data.IORef +import Control.Exception + +import Streaming +import qualified Streaming.Prelude as S + +import System.IO + +newtype Bracketed a r = + Bracketed { runBracketed :: IORef Finstack -> Stream (Of a) IO r } + deriving Functor + +-- | `first` maps over the yielded elements. +instance Bifunctor Bracketed where + first f (Bracketed b) = Bracketed (S.map f . b) + second = fmap + +instance Applicative (Bracketed a) where + pure = Bracketed . const . pure + Bracketed b <*> Bracketed b' = Bracketed (\finref -> + b finref <*> b' finref) + +instance Monad (Bracketed a) where + return = pure + Bracketed b >>= f = Bracketed (\finref -> + do r <- b finref + let Bracketed b' = f r + b' finref) + +instance MonadIO (Bracketed a) where + liftIO action = Bracketed (\_ -> liftIO action) + +-- | A stack of finalizers, accompanied by its length. +-- +-- Finalizers at the head of the list correspond to deeper levels of nesting. +data Finstack = Finstack !Int [IO ()] + +-- | Lift a `Stream` that doesn't perform allocation to a `Bracketed`. +clear :: Stream (Of x) IO r -> Bracketed x r +clear stream = Bracketed (const stream) + +-- | Lift a `Stream` that requires resource allocation to a `Bracketed`. +bracketed :: IO a -> (a -> IO ()) -> (a -> Stream (Of x) IO r) -> Bracketed x r +bracketed allocate finalize stream = Bracketed (\finref -> + let open = do + a <- allocate + Finstack size0 fins <- readIORef finref + writeIORef finref (Finstack (succ size0) (finalize a : fins)) + pure (size0,a) + in do (size0,a) <- liftIO (mask (\_ -> open)) + r <- stream a + liftIO (mask (\_ -> reset size0 finref)) + pure r) + +-- | Consume a `Bracketed` stream, exhausting it. +with :: Bracketed a r -> (forall x. Stream (Of a) IO x -> IO (Of b x)) -> IO (Of b r) +with (Bracketed b) f = + Control.Exception.bracket (newIORef (Finstack 0 [])) + (reset 0) + (f . b) + +-- | Consume a `Bracketed` stream, possibly wihout exhausting it. +-- +-- Finalizers lying in unconsumed parts of the stream will not be executed +-- until the callback returns, so better not tarry too long if you want +-- prompt finalization. +with_ :: Bracketed a r -> (Stream (Of a) IO r -> IO b) -> IO b +with_ (Bracketed b) f = + Control.Exception.bracket (newIORef (Finstack 0 [])) + (reset 0) + (f . b) + +-- | Apply to the underlying stream a transformation that preserves the return value. +over :: (forall x. Stream (Of a) IO x -> Stream (Of b) IO x) -> Bracketed a r -> Bracketed b r +over transform (Bracketed b) = Bracketed (transform . b) + +-- | Apply to the underlying stream a transformation that might not preserve +-- the return value. +over_ :: (Stream (Of a) IO r -> Stream (Of b) IO r') -> Bracketed a r -> Bracketed b r' +over_ transform (Bracketed b) = Bracketed (\finref -> + let level = do + Finstack size _ <- readIORef finref + pure size + in do size0 <- liftIO level + r <- transform (b finref) + liftIO (mask (\_ -> reset size0 finref)) + pure r) + +-- | Replaces each element of a stream with an associated stream. +-- +-- Can be useful for traversing hierachical structures. +for :: Bracketed a r -> (a -> Bracketed b x) -> Bracketed b r +for (Bracketed b) f = + Bracketed (\fins -> S.for (b fins) (flip (runBracketed . f) fins)) + +-- | Executes all finalizers that lie above a certain level. +reset :: Int -> IORef Finstack -> IO () +reset size0 finref = + do Finstack size fins <- readIORef finref + let (pending,fins') = splitAt (size - size0) fins + writeIORef finref (Finstack size0 fins') + foldr finally (pure ()) pending + +-- | A bracketed stream of all the lines in a text file. +-- +-- This is adequate for simple use cases. For more advanced ones where +-- efficiency and memory usage are important, it's better to use a packed +-- text representation like the one provided by the @text@ package. +linesFromFile :: TextEncoding -> NewlineMode -> FilePath -> Bracketed String () +linesFromFile encoding newlineMode path = bracketed + (openFile path ReadMode) + hClose + (\h -> do + liftIO (hSetEncoding h encoding) + liftIO (hSetNewlineMode h newlineMode) + S.untilRight + (do + eof <- hIsEOF h + if eof then Right <$> pure () else Left <$> hGetLine h + ) + ) + +-- | Given a list of text files and line ranges, create a stream of lines +-- belonging to the concatenated ranges. +concatRanges + :: TextEncoding + -> NewlineMode + -> [(FilePath, Int, Int)] + -> Bracketed String () +concatRanges encoding newlineMode ranges = + let streamRange (path, start, end) = + over_ (S.take (end - start)) . over (S.drop start) $ linesFromFile + encoding + newlineMode + path + in traverse_ streamRange ranges + diff --git a/streaming-bracketed.cabal b/streaming-bracketed.cabal new file mode 100644 --- /dev/null +++ b/streaming-bracketed.cabal @@ -0,0 +1,79 @@ +name: streaming-bracketed +version: 0.1.0.0 +synopsis: A resource management decorator for "streaming". +description: This package provides a decorator for the Stream type from + the "streaming" package, that lets you perform bracket-like + operations that allocate and deallocate resources used + by the stream. + . + By carefully managing the operations that are lifted to + the bracketed streams, we can ensure that finalizers are + promptly called even with operations like "take", which do + not consume the whole stream. +license: MIT +license-file: LICENSE +author: daniel +maintainer: diaz_carrete@yahoo.com +category: Data,Streaming +build-type: Simple +extra-source-files: + LICENSE + README.md + ChangeLog.md + .travis.yml + .gitignore +cabal-version: 2 + +library + exposed-modules: + Streaming.Bracketed + Streaming.Bracketed.Internal + build-depends: + base >= 4.10 && <5, + streaming >= 0.2.0.0 && < 0.3 + hs-source-dirs: library + default-language: Haskell2010 + +test-suite doctests + type: + exitcode-stdio-1.0 + ghc-options: + -Wall -threaded + hs-source-dirs: + tests + main-is: + doctests.hs + build-depends: + base >= 4.10 && < 5, + streaming >= 0.2.0.0 && < 0.3, + doctest >= 0.16.0, + directory >= 1.3.1.0, + filepath >= 1.3 + default-language: + Haskell2010 + +test-suite tests + type: + exitcode-stdio-1.0 + ghc-options: + -Wall -threaded + hs-source-dirs: + tests + main-is: + tests.hs + build-depends: + base >= 4.10 && < 5, + tasty >= 0.10.1.1, + tasty-hunit >= 0.9.2, + streaming >= 0.2.0.0 && < 0.3, + streaming-commons >= 0.2.1.0, + directory >= 1.3.1.0, + filepath >= 1.3, + containers >= 0.5.0.1, + streaming-bracketed + default-language: + Haskell2010 + +source-repository head + type: git + location: https://github.com/danidiaz/streaming-bracketed.git diff --git a/tests/doctests.hs b/tests/doctests.hs new file mode 100644 --- /dev/null +++ b/tests/doctests.hs @@ -0,0 +1,7 @@ +module Main where + +import Test.DocTest + +main :: IO () +main = doctest + ["-ilibrary","library/Streaming/Bracketed.hs"] diff --git a/tests/tests.hs b/tests/tests.hs new file mode 100644 --- /dev/null +++ b/tests/tests.hs @@ -0,0 +1,277 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE ViewPatterns #-} +module Main where + +import Data.Monoid +import Data.Foldable +import Data.List +import Data.List.NonEmpty ( NonEmpty(..) ) +import qualified Data.List.NonEmpty +import qualified Data.Set +import Data.Tree +import Test.Tasty +import Test.Tasty.HUnit ( testCase + , Assertion + , assertEqual + , assertBool + , assertFailure + ) + +import Data.IORef +import Control.Monad +import Control.Exception + +import System.IO +import System.FilePath +import System.Directory + +import Streaming +import qualified Streaming.Prelude as S +import qualified Streaming.Bracketed as R + +import qualified Data.Streaming.Filesystem as FS -- streaming-commons + +main :: IO () +main = defaultMain tests + +tests :: TestTree +tests = testGroup + "All" + [ testCase "bracketed" testBracket + , testCase "over" testOver + , testCase "over_" testOver_ + , testCase "exception" testException + , testCase "for" testFor + , testCase "forException" testForException + , testCase "forCleanupException" testForCleanupException + , testCase "forTake" testForTake + , testGroup + "file stuff" + [ testCase "testDirTraversal" testDirTraversal + , testCase "challenge" testChallenge + ] + ] + +testBracket :: Assertion +testBracket = + do ref <- newIORef "" + let b = R.bracketed (modifyIORef' ref ('x':)) + (\_ -> modifyIORef' ref ('y':)) + (\_ -> S.each "abcd") + () :> () <- R.with b (\stream -> + S.foldM (\() c -> modifyIORef' ref (c:)) (pure ()) pure stream) + res <- reverse <$> readIORef ref + assertEqual "stream results" "xabcdy" res + + +testOver :: Assertion +testOver = + do ref <- newIORef "" + let b = R.bracketed (modifyIORef' ref ('x':)) + (\_ -> modifyIORef' ref ('y':)) + (\_ -> S.each "abcd") + b' = R.over (\stream -> S.yield 'u' *> stream <* S.yield 'v') b + () :> () <- R.with b' (\stream -> + S.foldM (\() c -> modifyIORef' ref (c:)) (pure ()) pure stream) + res <- reverse <$> readIORef ref + assertEqual "stream results" "uxabcdyv" res + +testOver_ :: Assertion +testOver_ = + do ref <- newIORef "" + let b = R.bracketed (modifyIORef' ref ('x':)) + (\_ -> modifyIORef' ref ('y':)) + (\_ -> S.each "abcd") + b' = R.over_ (S.take 2) b + b'' = R.over (\stream -> S.yield 'u' *> stream <* S.yield 'v') b' + h = R.bracketed (modifyIORef' ref ('h':)) + (\_ -> modifyIORef' ref ('l':)) + (\_ -> S.each "ijk") + () :> () <- R.with (b'' *> h) (\stream -> + S.foldM (\() c -> modifyIORef' ref (c:)) (pure ()) pure stream) + res <- reverse <$> readIORef ref + assertEqual "stream results" "uxabyvhijkl" res + +testException :: Assertion +testException = + do ref <- newIORef "" + let b = R.bracketed (modifyIORef' ref ('x':)) + (\_ -> modifyIORef' ref ('y':)) + (\_ -> S.yield 'a' *> S.yield 'b' *> liftIO (fail "oops")) + _ :: Either IOException (Of () ()) <- try (R.with b (\stream -> + S.foldM (\() c -> modifyIORef' ref (c:)) (pure ()) pure stream)) + res <- reverse <$> readIORef ref + assertEqual "stream results" "xaby" res + +testFor :: Assertion +testFor = + do ref <- newIORef "" + let b = R.bracketed (modifyIORef' ref ('x':)) + (\_ -> modifyIORef' ref ('y':)) + (\_ -> S.each "ab") + f _ = R.bracketed (modifyIORef' ref ('u':)) + (\_ -> modifyIORef' ref ('v':)) + (\_ -> S.each "ij") + () :> () <- R.with (R.for b f) (\stream -> + S.foldM (\() c -> modifyIORef' ref (c:)) (pure ()) pure stream) + res <- reverse <$> readIORef ref + assertEqual "stream results" "xuijvuijvy" res + +testForException :: Assertion +testForException = + do ref <- newIORef "" + let b = R.bracketed (modifyIORef' ref ('x':)) + (\_ -> modifyIORef' ref ('y':)) + (\_ -> S.each "ab") + f _ = R.bracketed (modifyIORef' ref ('u':)) + (\_ -> modifyIORef' ref ('v':)) + (\_ -> S.yield 'i' *> liftIO (fail "oops")) + _ :: Either IOException (Of () ()) <- try (R.with (R.for b f) (\stream -> + S.foldM (\() c -> modifyIORef' ref (c:)) (pure ()) pure stream)) + res <- reverse <$> readIORef ref + assertEqual "stream results" "xuivy" res + + +testForCleanupException :: Assertion +testForCleanupException = + do ref <- newIORef "" + let b = R.bracketed (modifyIORef' ref ('x':)) + (\_ -> modifyIORef' ref ('y':)) + (\_ -> S.each "ab") + f _ = R.bracketed (modifyIORef' ref ('u':)) + (\_ -> modifyIORef' ref ('v':) *> fail "finoops") + (\_ -> S.each "ij") + _ :: Either IOException (Of () ()) <- try (R.with (R.for b f) (\stream -> + S.foldM (\() c -> modifyIORef' ref (c:)) (pure ()) pure stream)) + res <- reverse <$> readIORef ref + assertEqual "stream results" "xuijvy" res + + +testForTake :: Assertion +testForTake = + do ref <- newIORef "" + let b = R.bracketed (modifyIORef' ref ('x':)) + (\_ -> modifyIORef' ref ('y':)) + (\_ -> S.each "ab") + f _ = R.bracketed (modifyIORef' ref ('u':)) + (\_ -> modifyIORef' ref ('v':)) + (\_ -> S.each "ij") + () :> () <- R.with (R.over_ (S.take 3) (forever (R.for b f))) (\stream -> + S.foldM (\() c -> modifyIORef' ref (c:)) (pure ()) pure stream) + res <- reverse <$> readIORef ref + assertEqual "stream results" "xuijvuivy" res + +directoryTree :: Tree (FilePath, [FilePath]) +directoryTree = Node + ("a", ["file1", "file2"]) + [ Node + ("aa", ["file3", "file4"]) + [ Node ("aaa", ["file5"]) + [Node ("aaaa", ["file6"]) [], Node ("aaab", ["file7", "file8"]) []] + ] + , Node + ("ab", ["file9", "file10"]) + [ Node + ("aba", ["file11"]) + [ Node ("abaa", ["file12"]) [] + , Node ("abab", ["file13", "file14"]) [] + , Node ("abac", ["file15"]) [] + , Node ("abad", ["file16", "file17"]) [] + ] + , Node ("abb", ["file18"]) [] + ] + , Node ("ac", []) [] + , Node ("ad", []) [Node ("ada", []) []] + ] + +-- | Annotate each node with the list of all its ancestors. The root node will +-- be at the end of the list. +inherit :: Tree a -> Tree (NonEmpty a) +inherit tree = foldTree algebra tree [] where + algebra :: a -> [[a] -> Tree (NonEmpty a)] -> [a] -> Tree (NonEmpty a) + algebra a fs as = Node (a:|as) (fs <*> [a:as]) + +expectedPaths :: Tree (FilePath, [FilePath]) -> [FilePath] +expectedPaths tree = + let alg (dir, filenames) pathsBelow = + dir : ((dir ) <$> filenames ++ mconcat pathsBelow) + in foldTree alg tree + +createHierarchy :: Tree (FilePath, [FilePath]) -> FilePath -> IO () +createHierarchy = + let alg (dir, filenames) downwards (( dir) -> base) = do + createDirectory base + let filepaths = (base ) <$> filenames + for_ filepaths (\path -> withFile path WriteMode (\_ -> pure ())) + for_ downwards ($ base) + in foldTree alg + +traverseDirectory :: FilePath -> R.Bracketed FilePath () +traverseDirectory dir = do + let maybeToEither = maybe (Right ()) Left + stream = R.bracketed + (FS.openDirStream dir) + FS.closeDirStream + (\dirStream -> + S.untilRight $ maybeToEither <$> FS.readDirStream dirStream + ) + R.for + stream + (\filename -> do + let filepath = dir filename + R.clear (S.yield filepath) + filetype <- liftIO $ FS.getFileType filepath + case filetype of + FS.FTDirectory -> traverseDirectory filepath + _ -> pure () + ) + + +testDirTraversal :: Assertion +testDirTraversal = do + -- http://hackage.haskell.org/package/directory-1.3.3.0/docs/System-Directory.html + -- http://hackage.haskell.org/package/filepath-1.4.2.1/docs/System-FilePath-Posix.html + let baseDir = "__3hgal34_streaming_bracketed_testTreeTraversal_" + testDir <- fmap ( baseDir) getTemporaryDirectory + do + testDirExists <- doesPathExist testDir + when testDirExists (removePathForcibly testDir) + createDirectory testDir + createHierarchy directoryTree testDir + absolute :> _ <- R.with (traverseDirectory testDir) S.toList + let relative = makeRelative testDir <$> absolute + assertEqual "path sets" + (Data.Set.fromList (expectedPaths directoryTree)) + (Data.Set.fromList relative) + +testChallenge :: Assertion +testChallenge = do + -- https://twitter.com/DiazCarrete/status/1016073374458671104 + let baseDir = "__3hgal34_streaming_bracketed_testChallenge_" + testDir <- fmap ( baseDir) getTemporaryDirectory + let testFile = "challenge.txt" + testPath = testDir testFile + do + testDirExists <- doesPathExist testDir + when testDirExists (removePathForcibly testDir) + createDirectory testDir + withFile + testPath + WriteMode + (\h -> do + hPutStrLn h "aaa" + hPutStrLn h "bbb" + hPutStrLn h "bbb" + hPutStrLn h "ccc" + hPutStrLn h "ccc" + ) + ranges :> _ <- R.with + (R.concatRanges utf8 + nativeNewlineMode + [(testPath, 4, 5), (testPath, 1, 3), (testPath, 0, 1)] + ) + S.toList + assertEqual "ranges" ["ccc", "bbb", "bbb", "aaa"] ranges + +