streaming-sort (empty) → 0.1.0.0
raw patch · 7 files changed
+543/−0 lines, 7 filesdep +QuickCheckdep +basedep +binarysetup-changed
Dependencies added: QuickCheck, base, binary, bytestring, directory, exceptions, hspec, streaming, streaming-binary, streaming-bytestring, streaming-sort, streaming-with, transformers
Files
- ChangeLog.md +5/−0
- LICENSE +20/−0
- Setup.hs +2/−0
- src/Streaming/Sort.hs +310/−0
- src/Streaming/Sort/Lifted.hs +93/−0
- streaming-sort.cabal +55/−0
- test/Spec.hs +58/−0
+ ChangeLog.md view
@@ -0,0 +1,5 @@+# Revision history for streaming-sort++## 0.1.0.0 -- 2018-03-18++* First version. Released on an unsuspecting world.
+ LICENSE view
@@ -0,0 +1,20 @@+Copyright (c) 2018 Ivan Lazar Miljenovic++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.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ src/Streaming/Sort.hs view
@@ -0,0 +1,310 @@+{-# LANGUAGE MultiParamTypeClasses, RankNTypes, ScopedTypeVariables #-}++{- |+ Module : Streaming.Sort+ Description : Sorting values in a Stream+ Copyright : (c) Ivan Lazar Miljenovic+ License : MIT+ Maintainer : Ivan.Miljenovic@gmail.com+++ This module is designed to be imported qualified.++ -}+module Streaming.Sort+ ( -- * In-memory sorting+ -- $memory+ sort+ , sortBy+ , sortOn+ -- * File-based sorting+ -- $filesort+ , withFileSort+ , withFileSortBy+ -- ** Exceptions+ , SortException (..)+ -- ** Configuration+ , Config+ , defaultConfig+ -- $lenses+ , setConfig+ , chunkSize+ , maxFiles+ , useDirectory+ ) where++import Streaming (Of(..), Stream)+import qualified Streaming as S+import Streaming.Binary (decoded)+import qualified Streaming.Prelude as S+import Streaming.With++import Data.Binary (Binary, encode)+import qualified Data.ByteString.Lazy as BL+import qualified Data.ByteString.Streaming as BS++import Control.Exception (Exception(..), IOException,+ mapException)+import Control.Monad (join, void)+import Control.Monad.Catch (MonadMask, MonadThrow, finally,+ throwM)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.Trans.Class (lift)+import Data.Bool (bool)+import Data.Coerce (Coercible, coerce)+import Data.Function (on)+import Data.Functor.Identity (Identity(Identity), runIdentity)+import Data.Int (Int64)+import qualified Data.List as L+import Data.Maybe (catMaybes)+import System.Directory (doesDirectoryExist, getPermissions,+ removeFile, writable)+import System.IO (hClose, openBinaryTempFile)++--------------------------------------------------------------------------------++{- $memory++These functions require reading all the values from the Stream before+being able to sort them. As such, it is /highly/ recommended you only+use them for short Streams.++-}++-- | Sort the values based upon their 'Ord' instance.+sort :: (Monad m, Ord a) => Stream (Of a) m r -> Stream (Of a) m r+sort = sortBy compare++-- | Use the specified comparison function to sort the values.+sortBy :: (Monad m) => (a -> a -> Ordering) -> Stream (Of a) m r -> Stream (Of a) m r+sortBy cmp s = lift (S.toList s) >>= srt+ where+ srt (as :> r) = S.each (L.sortBy cmp as) >> return r++-- | Use the provided function to be able to compare values.+sortOn :: (Ord b, Monad m) => (a -> b) -> Stream (Of a) m r -> Stream (Of a) m r+sortOn f = S.map fst+ . sortBy (compare `on` snd)+ . S.map ((,) <*> f)++--------------------------------------------------------------------------------++{- $filesort++For large Streams it may not be possible to sort it entirely in+memory. As such, these functions work by sorting chunks of the Stream+and storing them in temporary files before merging them all together.++These functions may throw a 'SortException'.++-}++data Config = Config+ { _chunkSize :: !Int+ -- ^ Size of chunks to sort in-memory.+ , _maxFiles :: !Int+ -- ^ The maximum number of temporary files to be reading at any+ -- one time.+ , _useDirectory :: !(Maybe FilePath)+ -- ^ Where to store temporary files. Will be cleaned up+ -- afterwards. 'Nothing' indicates to use the system temporary+ -- directory.+ } deriving (Show)++-- | Default settings for sorting using external files:+--+-- * Have a chunk size of @1000@.+--+-- * No more than @100@ temporary files to be open at a time.+--+-- * Use the system temporary directory.+defaultConfig :: Config+defaultConfig = Config+ { _chunkSize = 1000+ , _maxFiles = 100+ , _useDirectory = Nothing+ }++{- $lenses++These functions are lenses compatible with the @lens@, @microlens@,+etc. libraries.++-}++-- type Lens s t a b = forall f. (Functor f) => (a -> f b) -> s -> f t++-- | A specialised variant of @set@ from lens, microlens, etc. defined+-- here in case you're not using one of those libraries.+setConfig :: (forall f. (Functor f) => (a -> f a) -> Config -> f Config)+ -> a -> Config -> Config+setConfig lens a = runIdentity #. lens (const (Identity a))+{-# INLINABLE setConfig #-}++(#.) :: (Coercible c b) => (b -> c) -> (a -> b) -> (a -> c)+(#.) _ = coerce (\x -> x :: b) :: forall a b. Coercible b a => a -> b++chunkSize :: (Functor f) => (Int -> f Int) -> Config -> f Config+chunkSize inj cfg = (\v -> cfg { _chunkSize = v}) <$> inj (_chunkSize cfg)+{-# INLINABLE chunkSize #-}++maxFiles :: (Functor f) => (Int -> f Int) -> Config -> f Config+maxFiles inj cfg = (\v -> cfg { _maxFiles = v}) <$> inj (_maxFiles cfg)+{-# INLINABLE maxFiles #-}++useDirectory :: (Functor f) => (Maybe FilePath -> f (Maybe FilePath)) -> Config -> f Config+useDirectory inj cfg = (\v -> cfg { _useDirectory = v}) <$> inj (_useDirectory cfg)+{-# INLINABLE useDirectory #-}++--------------------------------------------------------------------------------++-- | Use external files to temporarily store partially sorted (using+-- the comparison function) results (splitting into chunks of the+-- specified size if one is provided).+--+-- These files are stored inside the specified directory if+-- provided; if no such directory is provided then the system+-- temporary directory is used.+withFileSort :: (Ord a, Binary a, MonadMask m, MonadIO m, MonadThrow n, MonadIO n)+ => Config -> Stream (Of a) m v+ -> (Stream (Of a) n () -> m r) -> m r+withFileSort cfg = withFileSortBy cfg compare++-- | Use external files to temporarily store partially sorted (using+-- the comparison function) results (splitting into chunks of the+-- specified size if one is provided).+--+-- These files are stored inside the specified directory if+-- provided; if no such directory is provided then the system+-- temporary directory is used.+withFileSortBy :: (Binary a, MonadMask m, MonadIO m, MonadThrow n, MonadIO n)+ => Config -> (a -> a -> Ordering) -> Stream (Of a) m v+ -> (Stream (Of a) n () -> m r) -> m r+withFileSortBy cfg cmp str k = mapException SortIO . createDir $ \dir ->+ mergeAllFiles (_maxFiles cfg) dir cmp (initStream dir) k+ where+ createDir k' = liftIO (traverse checkDir (_useDirectory cfg)) -- :: m (Maybe (Maybe FilePath))+ -- If a directory is specified, make sure we can actually use it+ >>= (`getTmpDir` k') . join -- join is to get rid of double Maybe++ -- Make sure the directory exists and is writable.+ checkDir dir = do exists <- doesDirectoryExist dir+ canWrite <- writable <$> getPermissions dir+ return (bool Nothing (Just dir) (exists && canWrite))++ getTmpDir mdir = maybe withSystemTempDirectory withTempDirectory mdir "streaming-sort"++ initStream dir = initialSort (_chunkSize cfg) dir cmp str++-- | Do initial in-memory sorting, writing the results to disk.+initialSort :: (Binary a, MonadMask m, MonadIO m)+ => Int -> FilePath -> (a -> a -> Ordering)+ -> Stream (Of a) m r+ -> Stream (Of FilePath) m r+initialSort chnkSize dir cmp =+ S.mapped (writeSortedData dir)+ . S.maps (sortBy cmp)+ . S.chunksOf chnkSize++-- | Recursively merge files containing sorted data.+mergeAllFiles :: (Binary a, MonadMask m, MonadIO m, MonadThrow n, MonadIO n)+ => Int -> FilePath -> (a -> a -> Ordering)+ -> Stream (Of FilePath) m v+ -> (Stream (Of a) n () -> m r) -> m r+mergeAllFiles numFiles tmpDir cmp files k = go files+ where+ go = checkEmpty . S.mapped S.toList . S.chunksOf numFiles++ -- If no files, then evaluate the continuation with the empty Stream.+ checkEmpty chunks = S.uncons chunks >>= maybe (k (return ()))+ (uncurry checkSingleChunk)++ -- In a single chunk there's no need to write the contents back+ -- out to disk.+ checkSingleChunk ch chunks =+ S.uncons chunks >>= maybe (withFilesSort cmp ch k)+ (uncurry (withMultipleChunks ch))++ withMultipleChunks ch1 ch2 chunks = go (S.mapM sortAndWrite allChunks)+ where+ allChunks = S.yield ch1 >> S.yield ch2 >> chunks++ sortAndWrite fls = withFilesSort cmp fls (fmap S.fst' . writeSortedData tmpDir)++-- | Encode and write data to a new temporary file located within the+-- specified directory.+--+-- (Data not actually required to be sorted.)+writeSortedData :: (Binary a, MonadMask m, MonadIO m)+ => FilePath -> Stream (Of a) m r -> m (Of FilePath r)+writeSortedData tmpDir str = do fl <- liftIO newTmpFile+ (fl :>) <$> writeBinaryFile fl (encodeStream str)+ where+ newTmpFile = do (fl, h) <- openBinaryTempFile tmpDir "streaming-sort-chunk"+ hClose h -- Just want a new filename, not the actual Handle+ return fl++-- | Read in the specified files and merge the sorted streams.+withFilesSort :: (Binary a, MonadMask m, MonadIO m, MonadThrow n, MonadIO n)+ => (a -> a -> Ordering) -> [FilePath]+ -> (Stream (Of a) n () -> m r) -> m r+withFilesSort cmp files k = mergeContinuations readThenDelete+ files+ withMerged+ where+ withMerged = k . interleave cmp . map decodeStream++-- | A wrapper around 'withBinaryFileContents' that deletes the file+-- afterwards.+readThenDelete :: (MonadMask m, MonadIO m, MonadIO n) => FilePath+ -> (BS.ByteString n () -> m r) -> m r+readThenDelete fl k = withBinaryFileContents fl k `finally` liftIO (removeFile fl)++-- | Fold a list of continuations into one overall continuation.+mergeContinuations :: (Monad m) => (forall res. a -> (b -> m res) -> m res) -> [a] -> ([b] -> m r) -> m r+mergeContinuations toCont xs cont = go [] xs+ where+ go bs [] = cont bs+ go bs (a:as) = toCont a $ \b -> go (b:bs) as++-- | Merge multiple streams together using the provided comparison+-- function. Assumes provided streams are already sorted.+interleave :: (Monad m) => (a -> a -> Ordering) -> [Stream (Of a) m r] -> Stream (Of a) m ()+interleave cmp streams =+ go =<< lift (L.sortBy cmper . catMaybes <$> mapM S.uncons streams)+ where+ -- We take as input a list of @(a, Stream)@ tuples sorted on the @a@+ -- value to represent non-empty Streams.+ go [] = return ()+ go [(a,str)] = S.yield a >> void str -- Only stream left!+ go ((a,str):astrs') = do S.yield a+ -- Try to get the next element+ mastr' <- lift (S.uncons str)+ go (addBackIfNonEmpty mastr' astrs')++ -- If a Stream is non-empty, place it back in the correct position.+ addBackIfNonEmpty = maybe id (L.insertBy cmper)++ cmper = cmp `on` fst++-- | Streaming.Binary.encoded uses Builder under the hood, requiring IO.+encodeStream :: (Binary a, Monad m) => Stream (Of a) m r -> BS.ByteString m r+encodeStream = fromChunksLazy . S.map encode++-- | A wrapper around 'decoded' that throws an exception rather than+-- returning failure.+decodeStream :: (Binary a, MonadThrow m) => BS.ByteString m r -> Stream (Of a) m r+decodeStream bs = decoded bs >>= handleResult+ where+ handleResult (_, bytes, res) = either (lift . throwM . SortDecode bytes) return res++fromChunksLazy :: (Monad m) => Stream (Of BL.ByteString) m r -> BS.ByteString m r+fromChunksLazy = BS.fromChunks . S.concat . S.map BL.toChunks++--------------------------------------------------------------------------------++data SortException = SortIO IOException+ | SortDecode Int64 String+ deriving (Show)++instance Exception SortException
+ src/Streaming/Sort/Lifted.hs view
@@ -0,0 +1,93 @@+{-# LANGUAGE FlexibleContexts #-}++{- |+ Module : Streaming.Sort.Lifted+ Description : Lifted variants of "Streaming.Sort"+ Copyright : Ivan Lazar Miljenovic+ License : MIT+ Maintainer : Ivan.Miljenovic@gmail.com++++ -}+module Streaming.Sort.Lifted+ ( -- * In-memory sorting+ -- $memory+ sort+ , sortBy+ , sortOn+ -- * File-based sorting+ -- $filesort+ , withFileSort+ , withFileSortBy+ -- ** Exceptions+ , SortException (..)+ -- ** Configuration+ , Config+ , defaultConfig+ -- $lenses+ , setConfig+ , chunkSize+ , maxFiles+ , useDirectory+ ) where++import Streaming.Sort (Config, SortException(..), chunkSize,+ defaultConfig, maxFiles, setConfig, sort,+ sortBy, sortOn, useDirectory)+import qualified Streaming.Sort as SS++import Control.Monad.Catch (MonadMask, MonadThrow)+import Control.Monad.IO.Class (MonadIO)+import Data.Binary (Binary)+import Streaming.Prelude (Of, Stream)+import Streaming.With.Lifted (Withable(WithMonad), liftWith)++--------------------------------------------------------------------------------++{- $memory++These functions require reading all the values from the Stream before+being able to sort them. As such, it is /highly/ recommended you only+use them for short Streams.++-}++{- $filesort++For large Streams it may not be possible to sort it entirely in+memory. As such, these functions work by sorting chunks of the Stream+and storing them in temporary files before merging them all together.++These functions may throw a 'SortException'.++-}++-- | Use external files to temporarily store partially sorted (using+-- the comparison function) results (splitting into chunks of the+-- specified size if one is provided).+--+-- These files are stored inside the specified directory if+-- provided; if no such directory is provided then the system+-- temporary directory is used.+withFileSort :: (Ord a, Binary a+ , Withable w, MonadMask (WithMonad w), MonadIO (WithMonad w)+ , MonadThrow m, MonadIO m)+ => Config -> Stream (Of a) (WithMonad w) v+ -> w (Stream (Of a) m ())+withFileSort cfg str = liftWith (SS.withFileSort cfg str)++-- | Use external files to temporarily store partially sorted (using+-- the comparison function) results (splitting into chunks of the+-- specified size if one is provided).+--+-- These files are stored inside the specified directory if+-- provided; if no such directory is provided then the system+-- temporary directory is used.+withFileSortBy :: (Ord a, Binary a, Withable w+ , MonadMask (WithMonad w), MonadIO (WithMonad w)+ , MonadThrow m, MonadIO m)+ => Config -> (a -> a -> Ordering)+ -> Stream (Of a) (WithMonad w) v+ -> w (Stream (Of a) m ())+withFileSortBy cfg cmp str = liftWith (SS.withFileSortBy cfg cmp str)
+ streaming-sort.cabal view
@@ -0,0 +1,55 @@+name: streaming-sort+version: 0.1.0.0+synopsis: Sorting streams+description: Sort streaming values, using files for a cached merge-sort of long @Stream@s.+license: MIT+license-file: LICENSE+author: Ivan Lazar Miljenovic+maintainer: Ivan.Miljenovic@gmail.com+copyright: Ivan Lazar Miljenovic+category: Data, Streaming+build-type: Simple+extra-source-files: ChangeLog.md+cabal-version: >=1.10+tested-with: GHC == 7.10.2, GHC == 8.0.2, GHC == 8.2.2,+ -- GHC == 8.4.1,+ GHC == 8.5.*++source-repository head+ type: git+ location: https://github.com/haskell-streaming/streaming-sort.git++library+ exposed-modules: Streaming.Sort+ , Streaming.Sort.Lifted+ build-depends: base == 4.*+ , binary == 0.8.*+ , bytestring == 0.10.*+ , directory <= 1.4+ , exceptions >= 0.8 && < 0.11+ , streaming == 0.2.*+ , streaming-binary == 0.3.*+ , streaming-bytestring == 0.1.*+ , streaming-with == 0.2.*+ , transformers == 0.5.*+ hs-source-dirs: src+ default-language: Haskell2010++test-suite sorting-tests+ type: exitcode-stdio-1.0+ main-is: Spec.hs+ build-depends: streaming-sort+ , base+ , binary+ , directory+ , exceptions+ , streaming+ , streaming-with+ , transformers++ , QuickCheck >= 2.5 && < 2.12+ , hspec >= 2.1 && < 2.6+ hs-source-dirs: test+ default-language: Haskell2010++ ghc-options: -Wall
+ test/Spec.hs view
@@ -0,0 +1,58 @@+{-# LANGUAGE RankNTypes #-}+{- |+ Module : Main+ Description : Tests for sorting+ Copyright : (c) Ivan Lazar Miljenovic+ License : MIT+ Maintainer : Ivan.Miljenovic@gmail.com++++ -}+module Main (main) where++import Streaming.Sort (chunkSize, defaultConfig, setConfig, sort, withFileSort)++import Streaming.Prelude (Of, Stream)+import qualified Streaming.Prelude as S+import Streaming.With (withSystemTempDirectory)++import Control.Applicative (liftA2)+import Control.Monad.Catch (catchAll, throwM)+import Control.Monad.Trans.Class (lift)+import Data.Binary (Binary)+import Data.Functor.Identity (runIdentity)+import System.Directory (getDirectoryContents)++import qualified Data.List as L++import Test.Hspec (Expectation, describe, hspec, it, shouldReturn)+import Test.Hspec.QuickCheck (prop)+import Test.QuickCheck (Gen, Positive(Positive), arbitrary, forAllShrink,+ ioProperty, shrink)++--------------------------------------------------------------------------------++main :: IO ()+main = hspec $ do+ describe "in-memory" $+ prop "same as list-based" $+ forAllShrink (arbitrary :: Gen [Int]) shrink $+ liftA2 (==) L.sort (runIdentity . S.toList_ . sort . S.each)+ describe "file-based" $ do+ prop "same as list-based" $+ forAllShrink (arbitrary :: Gen [Int]) shrink $ \as (Positive cs) ->+ ioProperty $ withFileSort (setConfig chunkSize cs defaultConfig)+ (S.each as)+ (fmap (== L.sort as) . S.toList_)+ describe "cleans up temporary files" $ do+ it "on success" $+ fileSortCleanup (S.each [1::Int .. 10])+ it "on exception" $+ fileSortCleanup (S.each [1::Int .. 10] >> lift (throwM (userError "Tainted Producer")))++fileSortCleanup :: (Binary a, Ord a) => Stream (Of a) IO r -> Expectation+fileSortCleanup str = withSystemTempDirectory "test-streaming-sort-cleanup" $ \tmpDir -> do+ cnts0 <- getDirectoryContents tmpDir -- Should be [".", ".."]+ catchAll (withFileSort defaultConfig str S.effects) (const (return ()))+ (L.sort <$> getDirectoryContents tmpDir) `shouldReturn` L.sort cnts0