diff --git a/Data/Repa/Flow.hs b/Data/Repa/Flow.hs --- a/Data/Repa/Flow.hs +++ b/Data/Repa/Flow.hs @@ -10,9 +10,10 @@ -- @ -- > import Data.Repa.Array as A -- > import Data.Repa.Flow as F +-- > import Data.Repa.Flow.IO as F -- > import Data.Repa.Flow.Auto.Debug as F -- --- > ws <- fromFiles' [\"\/usr\/share\/dict\/words\", \"\/usr\/share\/dict\/cracklib-small\"] sourceLines +-- > ws <- fromFiles [\"\/usr\/share\/dict\/words\", \"\/usr\/share\/dict\/cracklib-small\"] sourceLines -- @ -- -- Show the first few elements of the first chunk of the first file. @@ -71,7 +72,7 @@ -- so open a file for each stream: -- -- @ --- > out <- toFiles' ["out1.txt", "out2.txt"] sinkLines +-- > out <- toFiles ["out1.txt", "out2.txt"] sinkLines -- @ -- -- Note that the @ws@ and @up@ we used before were bundles of stream @@ -123,19 +124,31 @@ , drainP -- * Conversion + -- ** List conversion , fromList, fromLists , toList1, toLists1 + -- ** Array conversion + , fromArray, fromArrays + , toArray1, toArrays1 + -- * Finalizers , finalize_i, finalize_o -- * Flow Operators + -- ** Replicating + , replicates_i + -- ** Mapping - -- | If you want to work on a chunk at a time then use - -- `Data.Repa.Flow.Generic.map_i` and - -- `Data.Repa.Flow.Generic.map_o` from "Data.Repa.Flow.Generic". + -- | If you want to work on a chunk at a time then use + -- `Data.Repa.Flow.Generic.map_i` and + -- `Data.Repa.Flow.Generic.map_o` from "Data.Repa.Flow.Generic". , map_i, map_o + , zipWith_i + -- | Higher arity zipWith functions. + , module Data.Repa.Flow.Auto.ZipWith + -- ** Connecting , dup_oo , dup_io @@ -147,12 +160,20 @@ , trigger_o -- ** Ignorance - , discard_o , ignore_o + , abandon_o -- ** Splitting , head_i + -- ** Concatenation + , concat_i + + -- ** Selecting + , select_i, select_o + , discard_i, discard_o + , mask_i, mask_o + -- ** Grouping , groups_i , groupsBy_i @@ -164,33 +185,11 @@ -- *** Segmented , folds_i, FoldsDict - , foldGroupsBy_i, FoldGroupsDict - - -- * Flow I/O - , defaultChunkSize - - -- ** Buckets - , module Data.Repa.Flow.IO.Bucket - - -- ** Sourcing - , sourceCSV - , sourceTSV - , sourceRecords - , sourceLines - , sourceChars - , sourceBytes - , sourcePacked - - -- ** Sinking - , sinkChars - , sinkLines - , sinkBytes - , sinkPacked - ) + , foldGroupsBy_i, FoldGroupsDict) where import Data.Repa.Flow.Auto import Data.Repa.Flow.Auto.Debug -import Data.Repa.Flow.Auto.IO +import Data.Repa.Flow.Auto.ZipWith import Data.Repa.Flow.IO.Bucket import Data.Repa.Flow.States diff --git a/Data/Repa/Flow/Auto.hs b/Data/Repa/Flow/Auto.hs --- a/Data/Repa/Flow/Auto.hs +++ b/Data/Repa/Flow/Auto.hs @@ -12,20 +12,36 @@ , sinksArity -- * Conversion + -- ** List conversion , fromList, fromLists , toList1, toLists1 + -- ** Array conversion + , fromArray, fromArrays + , toArray1, toArrays1 + -- * Evaluation , drainS , drainP + , consumeS -- * Flow Operators + -- ** Replicating + , replicates_i + -- ** Mapping -- | If you want to work on a chunk at a time then use -- `Data.Repa.Flow.Generic.map_i` and -- `Data.Repa.Flow.Generic.map_o` from "Data.Repa.Flow.Generic". , map_i, map_o + , zipWith_i + -- ** Processing + , process_i + + -- | Higher arity zipWith functions. + , module Data.Repa.Flow.Auto.ZipWith + -- ** Connecting , dup_oo , dup_io @@ -37,12 +53,20 @@ , trigger_o -- ** Ignorance - , discard_o , ignore_o + , abandon_o -- ** Splitting , head_i + -- ** Concatenation + , concat_i + + -- ** Selecting + , select_i, select_o + , discard_i, discard_o + , mask_i, mask_o + -- ** Grouping , groups_i , groupsBy_i @@ -60,14 +84,15 @@ , finalize_i, finalize_o ) where +import Data.Repa.Flow.Auto.ZipWith +import Data.Repa.Flow.Auto.Select +import Data.Repa.Flow.Auto.Base import Data.Repa.Array.Auto hiding (fromList, fromLists) import Data.Repa.Array.Material.Auto (A(..), Name(..)) -import Data.Repa.Fusion.Unpack as A import qualified Data.Repa.Array.Meta.Window as A import qualified Data.Repa.Array.Material as A -import qualified Data.Repa.Array.Generic.Target as A import qualified Data.Repa.Array.Generic as A import qualified Data.Repa.Flow.Chunked as C hiding (next) import qualified Data.Repa.Flow.Generic as G hiding (next) @@ -75,31 +100,6 @@ #include "repa-flow.h" --- | A bundle of stream sources, where the elements of the stream --- are chunked into arrays. -type Sources a = C.Sources Int IO A a - - --- | A bundle of stream sinks, where the elements of the stream --- are chunked into arrays. --- -type Sinks a = C.Sinks Int IO A a - - --- | Yield the number of streams in the bundle. -sourcesArity :: Sources a -> Int -sourcesArity = G.sourcesArity - - --- | Yield the number of streams in the bundle. -sinksArity :: Sinks a -> Int -sinksArity = G.sinksArity - - --- | Shorthand for common type classes. -type Flow a = (C.Flow Int IO A a, A.Windowable A a) - - -- Evaluation ----------------------------------------------------------------- -- | Pull all available values from the sources and push them to the sinks. -- Streams in the bundle are processed sequentially, from first to last. @@ -124,43 +124,13 @@ {-# INLINE drainP #-} --- Conversion ----------------------------------------------------------------- --- | Given an arity and a list of elements, yield sources that each produce all --- the elements. --- --- * All elements are stuffed into a single chunk, and each stream is given --- the same chunk. +-- | Pull all available values from the sources and pass them to the +-- given action. -- -fromList :: Build a t - => Int -> [a] -> IO (Sources a) -fromList xs = C.fromList A xs -{-# INLINE fromList #-} - - --- | Like `fromLists_i` but take a list of lists. Each each of the inner --- lists is packed into a single chunk. -fromLists :: Build a t - => Int -> [[a]] -> IO (Sources a) -fromLists xss = C.fromLists A xss -{-# INLINE fromLists #-} - - --- | Drain a single source from a bundle into a list of elements. -toList1 :: Build a t - => Int -> Sources a -> IO [a] -toList1 ix s - | ix >= G.sourcesArity s = return [] - | otherwise = C.toList1 ix s -{-# INLINE toList1 #-} - - --- | Drain a single source from a bundle into a list of chunks. -toLists1 :: Build a t - => Int -> Sources a -> IO [[a]] -toLists1 ix s - | ix >= G.sourcesArity s = return [] - | otherwise = C.toLists1 ix s -{-# INLINE toLists1 #-} +consumeS :: A.Bulk A a + => Sources a -> (Int -> a -> IO ()) -> IO () +consumeS = C.consumeS +{-# INLINE consumeS #-} -- Finalizers ----------------------------------------------------------------- @@ -207,21 +177,66 @@ {-# INLINE finalize_o #-} +-- Replicating ---------------------------------------------------------------- +-- | Segmented replicate. +replicates_i + :: (Flow (Int, a), Build a) + => Sources (Int, a) -- ^ Source of segment lengths and values. + -> IO (Sources a) +replicates_i = C.replicates_i A +{-# INLINE replicates_i #-} + + -- Mapping -------------------------------------------------------------------- -- | Apply a function to all elements pulled from some sources. -map_i :: (Flow a, Build b bt) +map_i :: (Flow a, Build b) => (a -> b) -> Sources a -> IO (Sources b) map_i f s = C.smap_i (\_ x -> f x) s {-# INLINE map_i #-} -- | Apply a function to all elements pushed to some sinks. -map_o :: (Flow a, Build b bt) +map_o :: (Flow a, Build b) => (a -> b) -> Sinks b -> IO (Sinks a) map_o f s = C.smap_o (\_ x -> f x) s {-# INLINE map_o #-} +-- | Combine corresponding elements of two sources with the given function. +zipWith_i + :: (Flow a, Flow b, Build c) + => (a -> b -> c) + -> Sources a -> Sources b + -> IO (Sources c) +zipWith_i f sa sb + = C.szipWith_ii A (\_ a b -> f a b) sa sb +{-# INLINE zipWith_i #-} + + +-- Processing ----------------------------------------------------------------- +-- | Apply a generic stream process to a bundle of sources. +process_i + :: ( Flow a, Flow b, Build b) + => (s -> a -> (s, Array b)) -- ^ Worker function. + -> s -- ^ Initial state. + -> Sources a -- ^ Input sources. + -> IO (Sources b) + +process_i = C.process_i +{-# INLINE process_i #-} + + +-- Concatenation -------------------------------------------------------------- +-- | Concatenate a flow of arrays into a flow of the elements. +concat_i + :: (Flow a, Build a) + => Sources (Array a) + -> IO (Sources a) +concat_i ss + = G.map_i (A.concat A) ss +{-# INLINE concat_i #-} + + -- Connecting ----------------------------------------------------------------- -- | Send the same data to two consumers. -- @@ -303,31 +318,30 @@ -- Ignorance ------------------------------------------------------------------ -- | Create a bundle of sinks of the given arity that drop all data on the --- floor. +-- floor. -- --- * The sinks is strict in the *chunks*, so they are demanded before being --- discarded. -- * Haskell debugging thunks attached to the chunks will be -- demanded, but thunks attached to elements may not be -- depending on -- whether the chunk representation is strict in the elements. -- -discard_o :: Int -> IO (Sinks a) -discard_o = G.discard_o -{-# INLINE discard_o #-} +ignore_o :: Int -> IO (Sinks a) +ignore_o = G.ignore_o +{-# INLINE ignore_o #-} -- | Create a bundle of sinks of the given arity that drop all data on the --- floor. +-- floor. -- --- * As opposed to `discard_o` the sinks are non-strict in the chunks. +-- * As opposed to `ignore_o` the sinks are non-strict in the chunks. -- * Haskell debugging thunks attached to the chunks will *not* be -- demanded. -- -ignore_o :: Int -> IO (Sinks a) -ignore_o = G.ignore_o -{-# INLINE ignore_o #-} +abandon_o :: Int -> IO (Sinks a) +abandon_o = G.abandon_o +{-# INLINE abandon_o #-} + -- Splitting ------------------------------------------------------------------ -- | Given a source index and a length, split the a list of that -- length from the front of the source. Yields a new source for the @@ -357,7 +371,7 @@ -- @ -- groups_i - :: (GroupsDict a u1 u2, Eq a) + :: (GroupsDict a, Eq a) => Sources a -- ^ Input elements. -> IO (Sources (a, Int)) -- ^ Starting element and length of groups. @@ -369,7 +383,7 @@ -- | Like `groupsBy`, but take a function to determine whether two consecutive -- values should be in the same group. groupsBy_i - :: GroupsDict a u1 u2 + :: GroupsDict a => (a -> a -> Bool) -- ^ Fn to check if consecutive elements -- are in the same group. -> Sources a -- ^ Input elements. @@ -382,15 +396,15 @@ -- | Dictionaries needed to perform a grouping. -type GroupsDict a u1 u2 - = C.GroupsDict Int IO A A u1 A u2 a +type GroupsDict a + = C.GroupsDict Int IO A A A a -- Folding -------------------------------------------------------------------- -- | Fold all the elements of each stream in a bundle, one stream after the -- other, returning an array of fold results. foldlS - :: (Flow b, Build a at) + :: (Flow b, Build a) => (a -> b -> a) -- ^ Combining funtion. -> a -- ^ Starting value. -> Sources b -- ^ Input elements to fold. @@ -404,7 +418,7 @@ -- | Fold all the elements of each stream in a bundle, one stream after the -- other, returning an array of fold results. foldlAllS - :: (Flow b) + :: Flow b => (a -> b -> a) -- ^ Combining funtion. -> a -- ^ Starting value. -> Sources b -- ^ Input elements to fold. @@ -439,7 +453,7 @@ -- [(\'a\',10),(\'b\',600),(\'c\',1),(\'d\',1),(\'e\',1)] -- @ -- -folds_i :: FoldsDict n a b u1 u2 u3 u4 +folds_i :: FoldsDict n a b => (a -> b -> b) -- ^ Worker function. -> b -- ^ Initial state when folding each segment. -> Sources (n, Int) -- ^ Segment lengths. @@ -452,8 +466,8 @@ {-# INLINE folds_i #-} -- | Dictionaries needed to perform a segmented fold. -type FoldsDict n a b u1 u2 u3 u4 - = C.FoldsDict Int IO A u1 A u2 A u3 A u4 n a b +type FoldsDict n a b + = C.FoldsDict Int IO A A A A n a b -- | Combination of `groupsBy_i` and `folds_i`. We determine the the segment @@ -475,7 +489,7 @@ -- @ -- foldGroupsBy_i - :: (FoldGroupsDict n a b u1 u2) + :: (FoldGroupsDict n a b) => (n -> n -> Bool) -- ^ Fn to check if consecutive elements -- are in the same group. -> (a -> b -> b) -- ^ Worker function for the fold. @@ -490,10 +504,8 @@ {-# INLINE foldGroupsBy_i #-} -type FoldGroupsDict n a b u1 u2 +type FoldGroupsDict n a b = ( A.BulkI A n , A.Material A a , A.Material A n - , A.Material A b - , Unpack (A.Buffer A n) u1 - , Unpack (A.Buffer A b) u2) + , A.Material A b) diff --git a/Data/Repa/Flow/Auto/Base.hs b/Data/Repa/Flow/Auto/Base.hs new file mode 100644 --- /dev/null +++ b/Data/Repa/Flow/Auto/Base.hs @@ -0,0 +1,161 @@ + +module Data.Repa.Flow.Auto.Base + ( Sources + , Sinks + , Flow + , sourcesArity + , sinksArity + + -- * Conversion + -- ** List conversion + , fromList, fromLists + , toList1, toLists1 + + -- ** Array conversion + , fromArray, fromArrays + , toArray1, toArrays1) +where +import Data.Repa.Array.Auto + hiding (fromList, fromLists) +import Data.Repa.Array.Material.Auto (A(..), Name(..)) +import qualified Data.Repa.Flow.Chunked as C +import qualified Data.Repa.Flow.Generic as G +import qualified Data.Repa.Array.Meta.Window as A +import qualified Data.Repa.Array.Auto as A +#include "repa-flow.h" + + +-- | A bundle of stream sources, where the elements of the stream +-- are chunked into arrays. +type Sources a = C.Sources Int IO A a + + +-- | A bundle of stream sinks, where the elements of the stream +-- are chunked into arrays. +-- +type Sinks a = C.Sinks Int IO A a + + +-- | Shorthand for common type classes. +type Flow a = (C.Flow Int IO A a, A.Windowable A a) + + +-- | Yield the number of streams in the bundle. +sourcesArity :: Sources a -> Int +sourcesArity = G.sourcesArity +{-# INLINE_FLOW sourcesArity #-} + + +-- | Yield the number of streams in the bundle. +sinksArity :: Sinks a -> Int +sinksArity = G.sinksArity +{-# INLINE_FLOW sinksArity #-} + + +-- Conversion ----------------------------------------------------------------- +-- | Given an arity and a list of elements, +-- yield sources that each produce all the elements. +-- +-- * All elements are stuffed into a single chunk, +-- and each stream is given the same chunk. +-- +fromList :: Build a + => Int -> [a] -> IO (Sources a) +fromList xs = C.fromList A xs +{-# INLINE_FLOW fromList #-} + + +-- | Like `fromList` but take a list of lists. +-- Each each of the inner lists is packed into a single chunk. +fromLists :: Build a + => Int -> [[a]] -> IO (Sources a) +fromLists xss = C.fromLists A xss +{-# INLINE_FLOW fromLists #-} + + +-- | Drain a single source from a bundle into a list of elements. +-- +-- * If the index does not specify a valid stream then the result will +-- be empty. +-- +toList1 :: Build a + => Int -> Sources a -> IO [a] +toList1 ix s + | ix < 0 || ix >= G.sourcesArity s + = return [] + + | otherwise + = C.toList1 ix s +{-# INLINE_FLOW toList1 #-} + + +-- | Drain a single source from a bundle into a list of chunks. +-- +-- * If the index does not specify a valid stream then the result will +-- be empty. +-- +toLists1 :: Build a + => Int -> Sources a -> IO [[a]] +toLists1 ix s + | ix < 0 || ix >= G.sourcesArity s + = return [] + + | otherwise + = C.toLists1 ix s +{-# INLINE_FLOW toLists1 #-} + + +------------------------------------------------------------------------------- +-- | Given an arity and an array of elements, +-- yield sources that each produce all the elements. +-- +-- * All elements are stuffed into a single chunk, +-- and each stream is given the same chunk. +-- +fromArray :: Build a + => Int -> Array a -> IO (Sources a) +fromArray n arr = G.fromList n [arr] +{-# INLINE_FLOW fromArray #-} + + +-- | Like `fromArray` but take an array of arrays. +-- Each of the inner arrays is packed into a single chunk. +fromArrays :: (Elem a, Build a) + => Int -> Array (Array a) -> IO (Sources a) +fromArrays n arrs = G.fromList n (A.toList arrs) +{-# INLINE_FLOW fromArrays #-} + + +-- | Drain a single source from a bundle into an array of elements. +-- +-- * If the index does not specify a valid stream then the result will +-- be empty. +-- +toArray1 :: (Elem a, Build a) + => Int -> Sources a -> IO (Array a) +toArray1 ix ss + | ix < 0 || ix >= G.sourcesArity ss + = return $ A.fromList [] + + | otherwise + = do chunks <- G.toList1 ix ss + return $ A.concat $ A.fromList chunks +{-# INLINE_FLOW toArray1 #-} + + +-- | Drain a single source from a bundle into an array of elements. +-- +-- * If the index does not specify a valid stream then the result will +-- be empty. +-- +toArrays1 :: (Elem a, Build a) + => Int -> Sources a -> IO (Array (Array a)) +toArrays1 ix ss + | ix < 0 || ix >= G.sourcesArity ss + = return $ A.fromList [] + + | otherwise + = do chunks <- G.toList1 ix ss + return $ A.fromList chunks +{-# INLINE_FLOW toArrays1 #-} + diff --git a/Data/Repa/Flow/Auto/Debug.hs b/Data/Repa/Flow/Auto/Debug.hs --- a/Data/Repa/Flow/Auto/Debug.hs +++ b/Data/Repa/Flow/Auto/Debug.hs @@ -1,4 +1,4 @@ -{-# LANGUAGE OverlappingInstances, TypeSynonymInstances, FlexibleInstances #-} +{-# LANGUAGE TypeSynonymInstances, FlexibleInstances #-} module Data.Repa.Flow.Auto.Debug ( -- * More more, more' diff --git a/Data/Repa/Flow/Auto/Format.hs b/Data/Repa/Flow/Auto/Format.hs new file mode 100644 --- /dev/null +++ b/Data/Repa/Flow/Auto/Format.hs @@ -0,0 +1,138 @@ + +module Data.Repa.Flow.Auto.Format + ( -- * Pre-defined data formats + module Data.Repa.Convert.Formats + + -- * Packing + + -- ** Standard + , packFormat_i, flatPackFormat_i + + -- ** With newlines + , packFormatLn_i, flatPackFormatLn_i + + -- ** Default ASCII format + , packAsciiLn_i, flatPackAsciiLn_i + + -- ** Keys and values + , keyPackAsciiLn_i) +where +import Data.Word +import Data.Char +import Data.Repa.Convert.Formats +import Data.Repa.Flow.Auto as F +import Data.Repa.Array as A +import Data.Repa.Array.Auto.Format as A +import qualified Data.Repa.Convert as C +import qualified Data.Repa.Flow.Generic as G +#include "repa-flow.h" + + +--------------------------------------------------------------------------------------------------- +-- | Pack elements into the given storage formats. +packFormat_i + :: (C.Packable format, Elem (Value format)) + => format -- ^ Desination format for data. + -> Sources (Value format) -- ^ Sources of values to be packed. + -> IO (Sources (Array Word8)) -- ^ Packed data. + +packFormat_i format ss + = let + packElem x + = let Just arr = A.packFormat format x + in arr + {-# INLINE packElem #-} + + in G.map_i (A.map packElem) ss +{-# INLINE_FLOW packFormat_i #-} + + +-- | Like `packFormat_i`, but return sources of flat bytes. +flatPackFormat_i + :: (C.Packable format, Elem (Value format)) + => format -> Sources (Value format) -> IO (Sources Word8) +flatPackFormat_i format ss + = concat_i =<< packFormat_i format ss +{-# INLINE_FLOW flatPackFormat_i #-} + + +--------------------------------------------------------------------------------------------------- +-- | Like `packFormat_i`, but also append a newline character +-- after every packed element. +packFormatLn_i + :: (C.Packable format, Elem (Value format)) + => format -- ^ Destination format for data. + -> Sources (Value format) -- ^ Sources of values to be packed. + -> IO (Sources (Array Word8)) -- ^ Packed data. + +packFormatLn_i format ss + = F.map_i (\arr -> A.concat $ A.fromList [arr, nl]) -- TODO: avoid copy + =<< packFormat_i format ss + where !nl = A.fromList [fromIntegral $ ord '\n'] +{-# INLINE_FLOW packFormatLn_i #-} + + +-- | Like `packFormatLn_i`, but return sources of flat bytes. +flatPackFormatLn_i + :: (C.Packable format, Elem (Value format)) + => format -> Sources (Value format) -> IO (Sources Word8) +flatPackFormatLn_i format ss + = concat_i =<< packFormatLn_i format ss +{-# INLINE_FLOW flatPackFormatLn_i #-} + + +--------------------------------------------------------------------------------------------------- +-- | Like `packFormatLn_i`, +-- but use a default, human-readable format to encode the values. +packAsciiLn_i + :: forall a + . ( C.FormatAscii a, a ~ Value (C.FormatAscii' a) + , Elem a + , Packable (C.FormatAscii' a)) + => Sources a -- ^ Sources of values to be packed. + -> IO (Sources (Array Word8)) -- ^ Packed data. + +packAsciiLn_i ss + = F.map_i (\arr -> A.concat $ A.fromList [arr, nl]) -- TODO: avoid copy + =<< packFormat_i (C.formatAscii proxy) ss + where !nl = A.fromList [fromIntegral $ ord '\n'] + proxy = (error "repa-flow: packAscii_i proxy" :: a) +{-# INLINE_FLOW packAsciiLn_i #-} + + +-- | Like `packAsciiLn_i`, but return sources of flat bytes. +flatPackAsciiLn_i + :: forall a + . ( C.FormatAscii a, a ~ Value (C.FormatAscii' a) + , Elem a + , Packable (C.FormatAscii' a)) + => Sources a -> IO (Sources Word8) +flatPackAsciiLn_i ss + = concat_i =<< packAsciiLn_i ss +{-# INLINE_FLOW flatPackAsciiLn_i #-} + + +--------------------------------------------------------------------------------------------------- +-- | Like `packFormatLn_i`, +-- but use a default, human-readable format to encode the values. +keyPackAsciiLn_i + :: forall a k + . ( C.FormatAscii a, a ~ Value (C.FormatAscii' a) + , Elem a + , Packable (C.FormatAscii' a) + , Elem k, Build k) + => Sources (k, a) -- ^ Sources of values to be packed. + -> IO (Sources (k, Array Word8)) -- ^ Packed data. + +keyPackAsciiLn_i ss + = let + proxy = (error "repa-flow: sndPackAsciiLn_i proxy" :: a) + !nl = A.fromList [fromIntegral $ ord '\n'] + packElem (k, x) + = let Just arr = A.packFormat (C.formatAscii proxy) x + in (k, A.concat $ A.fromList [arr, nl]) -- TODO: avoid copy + {-# INLINE packElem #-} + + in G.map_i (A.map packElem) ss +{-# INLINE_FLOW keyPackAsciiLn_i #-} + diff --git a/Data/Repa/Flow/Auto/IO.hs b/Data/Repa/Flow/Auto/IO.hs --- a/Data/Repa/Flow/Auto/IO.hs +++ b/Data/Repa/Flow/Auto/IO.hs @@ -18,26 +18,20 @@ , sourceRecords , sourceTSV , sourceCSV - , sourcePacked + , sourceFormatLn -- * Sinking , sinkBytes , sinkLines , sinkChars - , sinkPacked - - -- * Table IO - , toTable - , fromTable + , sinkFormatLn ) where import Data.Repa.Flow.Auto import Data.Repa.Flow.IO.Bucket import Data.Repa.Array.Material as A -import Data.Repa.Array.Auto.Unpack as A +import Data.Repa.Array.Auto.Format as A import Data.Repa.Array.Generic as A -import System.Directory -import System.FilePath import System.IO import Data.Word import qualified Data.Repa.Flow.Generic as G @@ -135,49 +129,20 @@ {-# INLINE sourceTSV #-} --- | Read packed binary data from some buckets and unpack the values --- to some `Sources`. --- --- The following uses the @colors.bin@ file produced by the `sinkPacked` example: --- --- @ --- > import Data.Repa.Flow as F --- > import Data.Repa.Convert.Format as F --- > :{ --- do let format = FixString ASCII 10 :*: Float64be :*: Int16be --- ss <- fromFiles' [\"colors.bin\"] $ sourcePacked format (error \"convert failed\") --- toList1 0 ss --- :} --- --- [\"red\" :*: (5.3 :*: 100), \"green\" :*: (2.8 :*: 93), \"blue\" :*: (0.99 :*: 42)] --- @ --- -sourcePacked - :: (Packable format, Target A (Value format)) - => format -- ^ Binary format for each value. - -> IO () -- ^ Action when a value cannot be converted. - -> Array B Bucket -- ^ Input buckets. +-- | Read the lines of a text file, +-- converting each line to values with the given format. +sourceFormatLn + :: ( Unpackable format + , Target A (Value format)) + => Integer -- ^ Chunk length. + -> IO () -- ^ Action when a line is too long. + -> IO (Array A Word8 -> IO ()) -- ^ Action if we can't convert a value. + -> format -- ^ Format of each line. + -> Array B Bucket -- ^ Source buckets. -> IO (Sources (Value format)) -sourcePacked format aFail bs - = return $ G.Sources (A.length bs) pull_sourcePacked - where - pull_sourcePacked i eat eject - = do let b = A.index bs i - op <- bIsOpen b - if not op - then eject - else do - eof <- bAtEnd b - if eof - then eject - else do - !chunk <- bGetArray b defaultChunkSize - case A.unpackForeign format (A.convert A chunk) of - Nothing -> aFail - Just vals -> eat vals - {-# INLINE pull_sourcePacked #-} -{-# INLINE_FLOW sourcePacked #-} +sourceFormatLn = G.sourceLinesFormat +{-# INLINE sourceFormatLn #-} --------------------------------------------------------------------------------------------------- @@ -201,8 +166,8 @@ {-# INLINE sinkLines #-} --- | Create sinks that convert values to a packed binary format and writes --- them to some buckets. +-- | Create sinks that convert values to some format and writes +-- them to buckets. -- -- @ -- > import Data.Repa.Flow as F @@ -215,36 +180,39 @@ -- , \"blue\" :*: 0.99 :*: 42 ] -- -- ss <- F.fromList 1 vals --- out <- toFiles' [\"colors.bin\"] $ sinkPacked format (error \"convert failed\") +-- out <- toFiles' [\"colors.bin\"] +-- $ sinkFormatLn format (error \"convert failed\") -- drainS ss out -- :} -- @ -- -sinkPacked - :: (Packable format, Bulk A (Value format)) +sinkFormatLn + :: (Packable format, Bulk A (Value format), Show format) => format -- ^ Binary format for each value. -> IO () -- ^ Action when a value cannot be serialized. -> Array B Bucket -- ^ Output buckets. -> IO (Sinks (Value format)) -sinkPacked format aFail bs - = return $ G.Sinks (A.length bs) push_sinkPacked eject_sinkPacked +sinkFormatLn format aFail bs + = return $ G.Sinks (A.length bs) + push_sinkFormatLn eject_sinkFormatLn where - push_sinkPacked i !chunk - = case A.packForeign format chunk of + push_sinkFormatLn i !chunk + = case A.packsFormatLn format chunk of Nothing -> aFail Just buf -> bPutArray (bs `index` i) (A.convert F buf) - {-# INLINE push_sinkPacked #-} + {-# INLINE push_sinkFormatLn #-} - eject_sinkPacked i + eject_sinkFormatLn i = bClose (bs `index` i) - {-# INLINE eject_sinkPacked #-} - -{-# INLINE_FLOW sinkPacked #-} + {-# INLINE eject_sinkFormatLn #-} +{-# INLINE_FLOW sinkFormatLn #-} --------------------------------------------------------------------------------------------------- --- | Create sinks that write values from some binary Repa table. +{- +-- | Create sinks that write values from some binary Repa table, +-- where all the values have a fixed length. toTable :: (Packable format, Bulk A (Value format)) => FilePath -- ^ Directory holding table. -> Int -- ^ Number of buckets to use. @@ -274,12 +242,13 @@ bs <- mapM newBucket names -- Create a sink bundle for the buckets. - kk <- sinkPacked format aFail (A.fromList B bs) + kk <- sinkFormatLn format aFail (A.fromList B bs) return $ Just kk {-# INLINE_FLOW toTable #-} --- | Create sources that read values from some binary Repa table. +-- | Create sources that read values from some binary Repa table, +-- where all the values have a fixed length. fromTable :: (Packable format, Target A (Value format)) => FilePath -- ^ Directory holding table. @@ -308,7 +277,7 @@ bs <- mapM newBucket fsRel -- Create a source bundle for the buckets. - ss <- sourcePacked format aFail (A.fromList B bs) + ss <- sourceFixedFormat format aFail (A.fromList B bs) return $ Just ss {-# INLINE_FLOW fromTable #-} - +-} diff --git a/Data/Repa/Flow/Auto/Select.hs b/Data/Repa/Flow/Auto/Select.hs new file mode 100644 --- /dev/null +++ b/Data/Repa/Flow/Auto/Select.hs @@ -0,0 +1,92 @@ + +module Data.Repa.Flow.Auto.Select + ( select_i, select_o + , discard_i, discard_o + , mask_i, mask_o) +where +import Data.Repa.Array as A +import Data.Repa.Flow.Auto.Base as F +import Data.Repa.Scalar.Singleton.Nat as D +import Data.Repa.Scalar.Product as D +import qualified Data.Repa.Flow.Generic as G +#include "repa-array.h" + + +--------------------------------------------------------------------------------------------------- +-- | Select a single column from a flow of rows of fields. +select_i + :: ( Select n (Array fs) + , Select' n (Array fs) ~ Array (Select' n fs)) + => Nat n -- ^ Index of column to keep. + -> Sources fs -- ^ Sources of complete rows. + -> IO (Sources (Select' n fs)) -- ^ Sources of selected column. + +select_i n ss + = G.map_i (select n) ss +{-# INLINE_FLOW select_i #-} + + +-- | Select a single column from a flow of fields. +select_o + :: ( Select n (Array fs) + , Select' n (Array fs) ~ Array (Select' n fs)) + => Nat n -- ^ Index of column to keep. + -> Sinks (Select' n fs) -- ^ Sinks for selected column. + -> IO (Sinks fs) -- ^ Sinks for complete rows. + +select_o n ss + = G.map_o (select n) ss +{-# INLINE_FLOW select_o #-} + + +--------------------------------------------------------------------------------------------------- +-- | Discard a single column from a flow of fields. +discard_i + :: ( Discard n (Array fs) + , Discard' n (Array fs) ~ Array (Discard' n fs)) + => Nat n -- ^ Index of column to discard. + -> Sources fs -- ^ Sources of complete rows. + -> IO (Sources (Discard' n fs)) -- ^ Sources of partial rows. + +discard_i n ss + = G.map_i (discard n) ss +{-# INLINE_FLOW discard_i #-} + + +-- | Discard a single column from a flow of fields. +discard_o + :: ( Discard n (Array fs) + , Discard' n (Array fs) ~ Array (Discard' n fs)) + => Nat n -- ^ Index of column to discard. + -> Sinks (Discard' n fs) -- ^ Sinks for partial rows. + -> IO (Sinks fs) -- ^ Sinks for complete rows. + +discard_o n ss + = G.map_o (discard n) ss +{-# INLINE_FLOW discard_o #-} + + +--------------------------------------------------------------------------------------------------- +-- | Mask columns from a flow of fields. +mask_i :: ( Mask ms (Array fs) + , Mask' ms (Array fs) ~ Array (Mask' ms fs)) + => ms -- ^ Column mask. + -> Sources fs -- ^ Sources of complete rows. + -> IO (Sources (Mask' ms fs)) -- ^ Sources of masked rows. + +mask_i ms ss + = G.map_i (mask ms) ss +{-# INLINE_FLOW mask_i #-} + + +-- | Mask columns from a flow of fields. +mask_o :: ( Mask ms (Array fs) + , Mask' ms (Array fs) ~ Array (Mask' ms fs)) + => ms -- ^ Column mask. + -> Sinks (Mask' ms fs) -- ^ Sources of complete rows. + -> IO (Sinks fs) -- ^ Sources of masked rows. + +mask_o ms ss + = G.map_o (mask ms) ss +{-# INLINE_FLOW mask_o #-} + diff --git a/Data/Repa/Flow/Auto/ZipWith.hs b/Data/Repa/Flow/Auto/ZipWith.hs new file mode 100644 --- /dev/null +++ b/Data/Repa/Flow/Auto/ZipWith.hs @@ -0,0 +1,120 @@ + +module Data.Repa.Flow.Auto.ZipWith + ( zipWith3_i + , zipWith4_i + , zipWith5_i + , zipWith6_i + , zipWith7_i) +where +import Data.Repa.Flow.Auto.Base +import Data.Repa.Array.Auto +import Data.Repa.Array.Material.Auto (Name(..)) +import qualified Data.Repa.Flow.Chunked as C hiding (next) +#include "repa-flow.h" + + +-- | Combine corresponding elements of three sources with the given function. +--- +-- TODO: do this chunkwise instead. +-- The core code has probably had an unswitch explosion. +-- +zipWith3_i + :: ( Flow a, Flow b, Flow c + , Build b, Build c, Build d) + => (a -> b -> c -> d) + -> Sources a -> Sources b -> Sources c + -> IO (Sources d) +zipWith3_i f sa sb sc + = C.szipWith_ii A (\_ a (b, c) -> f a b c) sa + =<< C.szipWith_ii A (\_ b c -> (b, c)) sb sc +{-# INLINE zipWith3_i #-} + + +-- | Combine corresponding elements of four sources with the given function. +--- +-- TODO: do this chunkwise instead. +-- The core code has probably had an unswitch explosion. +-- +zipWith4_i + :: ( Flow a, Flow b, Flow c, Flow d + , Build a, Build b, Build c, Build d, Build e) + => (a -> b -> c -> d -> e) + -> Sources a -> Sources b -> Sources c -> Sources d + -> IO (Sources e) +zipWith4_i f sa sb sc sd + = do sab <- C.szipWith_ii A (\_ a b -> (a, b)) sa sb + scd <- C.szipWith_ii A (\_ c d -> (c, d)) sc sd + result <- C.szipWith_ii A (\_ (a, b) (c, d) -> f a b c d) sab scd + return result +{-# NOINLINE zipWith4_i #-} +-- NOINLINE due to unswitch explosion. + + +-- | Combine corresponding elements of five sources with the given function. +--- +-- TODO: do this chunkwise instead. +-- The core code has probably had an unswitch explosion. +-- +zipWith5_i + :: ( Flow a, Flow b, Flow c, Flow d, Flow e + , Build a, Build b, Build c, Build d, Build e + , Build f) + => (a -> b -> c -> d -> e -> f) + -> Sources a -> Sources b -> Sources c -> Sources d -> Sources e + -> IO (Sources f) +zipWith5_i f sa sb sc sd se + = do sab <- C.szipWith_ii A (\_ a b -> (a, b)) sa sb + scd <- C.szipWith_ii A (\_ c d -> (c, d)) sc sd + result <- zipWith3_i (\(a, b) (c, d) e -> f a b c d e) + sab scd se + return result +{-# NOINLINE zipWith5_i #-} +-- NOINLINE due to unswitch explosion. + + +-- | Combine corresponding elements of six sources with the given function. +--- +-- TODO: do this chunkwise instead. +-- The core code has probably had an unswitch explosion. +-- +zipWith6_i + :: ( Flow a, Flow b, Flow c, Flow d, Flow e, Flow f + , Build a, Build b, Build c, Build d, Build e, Build f + , Build g) + => (a -> b -> c -> d -> e -> f -> g) + -> Sources a -> Sources b -> Sources c -> Sources d -> Sources e -> Sources f + -> IO (Sources g) +zipWith6_i ff sa sb sc sd se sf + = do sab <- C.szipWith_ii A (\_ a b -> (a, b)) sa sb + scd <- C.szipWith_ii A (\_ c d -> (c, d)) sc sd + result <- zipWith4_i (\(a, b) (c, d) e f -> ff a b c d e f) + sab scd se sf + return result +{-# NOINLINE zipWith6_i #-} +-- NOINLINE due to unswitch explosion. + + +-- | Combine corresponding elements of seven sources with the given function. +--- +-- TODO: do this chunkwise instead. +-- The core code has probably had an unswitch explosion. +-- +zipWith7_i + :: ( Flow a, Flow b, Flow c, Flow d, Flow e, Flow f, Flow g + , Build a, Build b, Build c, Build d, Build e, Build f, Build g + , Build h) + => (a -> b -> c -> d -> e -> f -> g -> h) + -> Sources a -> Sources b -> Sources c -> Sources d -> Sources e -> Sources f -> Sources g + -> IO (Sources h) + +zipWith7_i ff sa sb sc sd se sf sg + = do sab <- C.szipWith_ii A (\_ a b -> (a, b)) sa sb + scd <- C.szipWith_ii A (\_ c d -> (c, d)) sc sd + result <- zipWith5_i (\(a, b) (c, d) e f g -> ff a b c d e f g) + sab scd se sf sg + return result +{-# NOINLINE zipWith7_i #-} +-- NOINLINE due to unswitch explosion. + + + diff --git a/Data/Repa/Flow/Chunked.hs b/Data/Repa/Flow/Chunked.hs --- a/Data/Repa/Flow/Chunked.hs +++ b/Data/Repa/Flow/Chunked.hs @@ -7,6 +7,7 @@ -- * Evaluation , drainS + , consumeS -- * Conversion , fromList @@ -18,6 +19,9 @@ , finalize_i, finalize_o -- * Flow Operators + -- ** Replicate + , replicates_i + -- ** Mapping -- | If you want to work on a chunk at a time then use -- `Data.Repa.Flow.Generic.map_i` and @@ -25,6 +29,13 @@ , smap_i, smap_o , szipWith_ii + -- ** Processing + , process_i + + -- ** Unfolding + , unfolds_i + , StepUnfold (..) + -- ** Splitting , head_i @@ -40,22 +51,43 @@ , trigger_o -- ** Ignorance - , discard_o - , ignore_o) + , ignore_o + , abandon_o) where import Data.Repa.Flow.Chunked.Base import Data.Repa.Flow.Chunked.Map import Data.Repa.Flow.Chunked.Fold import Data.Repa.Flow.Chunked.Folds +import Data.Repa.Flow.Chunked.Generic import Data.Repa.Flow.Chunked.Groups -import Data.Repa.Flow.Chunked.Operator +import Data.Repa.Flow.Chunked.Process +import Data.Repa.Flow.Chunked.Replicate import Data.Repa.Flow.States +import qualified Data.Repa.Array.Generic as A import qualified Data.Repa.Flow.Generic as G #include "repa-flow.h" -- | Pull all available values from the sources and push them to the sinks. drainS :: (Next i, Monad m) - => Sources i m r a -> Sinks i m r a -> m () + => Sources i m r a -> Sinks i m r a -> m () drainS = G.drainS {-# INLINE drainS #-} + + +-- | Pull all available values from the sources and pass them to the given action. +consumeS :: (Next i, Monad m, A.Bulk r a) + => Sources i m r a -> (i -> a -> m ()) -> m () + +consumeS ss eatElem + = G.consumeS eatChunk ss + where + eatChunk ix arr + = mapM_ (eatElem ix) $ A.toList arr + {-# INLINE eatChunk #-} +{-# INLINE consumeS #-} + + + + + diff --git a/Data/Repa/Flow/Chunked/Base.hs b/Data/Repa/Flow/Chunked/Base.hs --- a/Data/Repa/Flow/Chunked/Base.hs +++ b/Data/Repa/Flow/Chunked/Base.hs @@ -16,7 +16,7 @@ import Control.Monad import qualified Data.Repa.Flow.Generic as G import qualified Data.Sequence as Q -import qualified Data.Foldable as Q +import qualified Data.Foldable as D import Prelude as P #include "repa-flow.h" @@ -127,13 +127,13 @@ -- As we've pulled whole chunks from the input stream, -- we now prepend the remaining ones back on. - let start = Q.length has - Q.length rest + let start = Q.length has - Q.length rest let stash = case mFinal of Nothing -> [] Just c -> [A.window start (Q.length rest) c] s2' <- G.prependOn_i (\i' -> i' == i) stash s2 - return (Q.toList here, s2') + return (D.toList here, s2') {-# INLINE_FLOW head_i #-} diff --git a/Data/Repa/Flow/Chunked/Fold.hs b/Data/Repa/Flow/Chunked/Fold.hs --- a/Data/Repa/Flow/Chunked/Fold.hs +++ b/Data/Repa/Flow/Chunked/Fold.hs @@ -6,7 +6,6 @@ import Data.Repa.Flow.Chunked.Base import Data.Repa.Flow.States as S import qualified Data.Repa.Flow.Generic as G -import qualified Data.Repa.Array.Generic.Target as A import qualified Data.Repa.Array.Generic.Index as A import qualified Data.Repa.Array.Generic as A #include "repa-flow.h" diff --git a/Data/Repa/Flow/Chunked/Folds.hs b/Data/Repa/Flow/Chunked/Folds.hs --- a/Data/Repa/Flow/Chunked/Folds.hs +++ b/Data/Repa/Flow/Chunked/Folds.hs @@ -5,10 +5,8 @@ where import Data.Repa.Flow.Chunked.Base import Data.Repa.Flow.States -import Data.Repa.Fusion.Unpack -import Data.Repa.Option +import Data.Repa.Scalar.Option import Data.Repa.Array.Generic.Index as A -import Data.Repa.Array.Generic.Target as A import Data.Repa.Array.Generic as A hiding (FoldsDict) import Data.Repa.Array.Meta.Window as A import Data.Repa.Array.Meta.Tuple as A @@ -17,7 +15,7 @@ -- | Dictionaries needed to perform a segmented fold. -type FoldsDict i m lSeg tSeg lElt tElt lGrp tGrp lRes tRes n a b +type FoldsDict i m lSeg lElt lGrp lRes n a b = ( States i m , Windowable lSeg (n, Int), Windowable lElt a , BulkI lSeg (n, Int) @@ -26,14 +24,12 @@ , BulkI lRes b , TargetI lElt a , TargetI lGrp n - , TargetI lRes b - , Unpack (Buffer lGrp n) tGrp - , Unpack (Buffer lRes b) tRes) + , TargetI lRes b) -- Folds ---------------------------------------------------------------------- -- | Segmented fold over vectors of segment lengths and input values. -folds_i :: FoldsDict i m lSeg tSeg lElt tElt lGrp tGrp lRes tRes n a b +folds_i :: FoldsDict i m lSeg lElt lGrp lRes n a b => Name lGrp -- ^ Layout for group names. -> Name lRes -- ^ Layout for fold results. -> (a -> b -> b) -- ^ Worker function. diff --git a/Data/Repa/Flow/Chunked/Generic.hs b/Data/Repa/Flow/Chunked/Generic.hs new file mode 100644 --- /dev/null +++ b/Data/Repa/Flow/Chunked/Generic.hs @@ -0,0 +1,73 @@ + +-- | Operators for chunked flows. +-- +-- Functions in this module are re-exports of the ones from +-- "Data.Repa.Flow.Generic", but using the `Sources` and `Sinks` +-- type synonyms for chunked flows. +-- +module Data.Repa.Flow.Chunked.Generic + ( -- * Watching + watch_i, watch_o + , trigger_o + + -- * Ignorance + , ignore_o + , abandon_o) +where +import Data.Repa.Flow.Chunked.Base +import Data.Repa.Array.Generic as A +import qualified Data.Repa.Flow.Generic as G +#include "repa-flow.h" + + +-- Watch ---------------------------------------------------------------------- +-- | Hook a monadic function to some sources, which will be passed every +-- chunk that is pulled from the result. +watch_i :: Monad m + => (i -> Array l a -> m ()) + -> Sources i m l a -> m (Sources i m l a) +watch_i = G.watch_i +{-# INLINE watch_i #-} + + +-- | Hook a monadic function to some sinks, which will be passed every +-- chunk that is pushed to the result. +watch_o :: Monad m + => (i -> Array l a -> m ()) + -> Sinks i m l a -> m (Sinks i m l a) + +watch_o = G.watch_o +{-# INLINE watch_o #-} + + +-- | Like `watch_o` but discard the incoming chunks after they are passed +-- to the function. +trigger_o :: Monad m + => i -> (i -> Array l a -> m ()) -> m (Sinks i m l a) +trigger_o = G.trigger_o +{-# INLINE trigger_o #-} + + +-- Ignorance ------------------------------------------------------------------ +-- | A sink that ignores all incoming data. +-- +-- * The sinks is strict in the *chunks*, so they are demanded before being +-- discarded. Haskell debugging thunks attached to the chunks will be +-- demanded, but thunks attached to elements may not be -- depending on +-- whether the chunk representation is strict in the elements. +-- +ignore_o :: Monad m => i -> m (Sinks i m l a) +ignore_o = G.ignore_o +{-# INLINE ignore_o #-} + + +-- | Yield a bundle of sinks of the given arity that drops all data on the +-- floor. +-- +-- This sink is non-strict in the chunks. +-- Haskell tracing thunks attached to the chunks will *not* be demanded. +-- +abandon_o :: Monad m => i -> m (Sinks i m l a) +abandon_o = G.abandon_o +{-# INLINE abandon_o #-} + diff --git a/Data/Repa/Flow/Chunked/Groups.hs b/Data/Repa/Flow/Chunked/Groups.hs --- a/Data/Repa/Flow/Chunked/Groups.hs +++ b/Data/Repa/Flow/Chunked/Groups.hs @@ -5,22 +5,18 @@ where import Data.Repa.Flow.Chunked.Base import Data.Repa.Flow.States -import Data.Repa.Fusion.Unpack import Data.Repa.Array.Meta.Tuple as A import Data.Repa.Array.Generic.Index as A -import Data.Repa.Array.Generic.Target as A import Data.Repa.Array.Generic as A hiding (GroupsDict) import qualified Data.Repa.Flow.Generic as G #include "repa-flow.h" -- | Dictionaries needed to perform a grouping. -type GroupsDict i m lVal lGrp tGrp lLen tLen a +type GroupsDict i m lVal lGrp lLen a = ( Flow i m lVal a, A.Index lVal ~ Int , TargetI lGrp a - , TargetI lLen Int - , Unpack (Buffer lGrp a) tGrp - , Unpack (Buffer lLen Int) tLen) + , TargetI lLen Int) -- Grouping ------------------------------------------------------------------- @@ -33,7 +29,7 @@ -- @ -- groupsBy_i - :: GroupsDict i m lVal lGrp tGrp lLen tLen a + :: GroupsDict i m lVal lGrp lLen a => Name lGrp -- ^ Layout for group names. -> Name lLen -- ^ Layout for group lengths. -> (a -> a -> Bool) -- ^ Whether successive elements should be grouped. diff --git a/Data/Repa/Flow/Chunked/Operator.hs b/Data/Repa/Flow/Chunked/Operator.hs deleted file mode 100644 --- a/Data/Repa/Flow/Chunked/Operator.hs +++ /dev/null @@ -1,73 +0,0 @@ - --- | Operators for chunked flows. --- --- Most functions in this module are re-exports of the ones from --- "Data.Repa.Flow.Generic.IO", but using the `Sources` and `Sinks` --- type synonyms for chunked flows. --- -module Data.Repa.Flow.Chunked.Operator - ( -- * Watching - watch_i, watch_o - , trigger_o - - -- * Ignorance - , discard_o - , ignore_o) -where -import Data.Repa.Flow.Chunked.Base -import Data.Repa.Array.Generic as A -import qualified Data.Repa.Flow.Generic as G -#include "repa-flow.h" - - --- Watch ---------------------------------------------------------------------- --- | Hook a monadic function to some sources, which will be passed every --- chunk that is pulled from the result. -watch_i :: Monad m - => (i -> Array l a -> m ()) - -> Sources i m l a -> m (Sources i m l a) -watch_i = G.watch_i -{-# INLINE watch_i #-} - - --- | Hook a monadic function to some sinks, which will be passed every --- chunk that is pushed to the result. -watch_o :: Monad m - => (i -> Array l a -> m ()) - -> Sinks i m l a -> m (Sinks i m l a) - -watch_o = G.watch_o -{-# INLINE watch_o #-} - - --- | Like `watch_o` but discard the incoming chunks after they are passed --- to the function. -trigger_o :: Monad m - => i -> (i -> Array l a -> m ()) -> m (Sinks i m l a) -trigger_o = G.trigger_o -{-# INLINE trigger_o #-} - - --- Ignorance ------------------------------------------------------------------ --- | A sink that ignores all incoming data. --- --- This sink is non-strict in the chunks. --- Haskell tracing thunks attached to the chunks will *not* be demanded. --- -ignore_o :: Monad m => i -> m (Sinks i m l a) -ignore_o = G.ignore_o -{-# INLINE ignore_o #-} - - --- | Yield a bundle of sinks of the given arity that drops all data on the --- floor. --- --- * The sinks is strict in the *chunks*, so they are demanded before being --- discarded. Haskell debugging thunks attached to the chunks will be --- demanded, but thunks attached to elements may not be -- depending on --- whether the chunk representation is strict in the elements. --- -discard_o :: Monad m => i -> m (Sinks i m l a) -discard_o = G.discard_o -{-# INLINE discard_o #-} - diff --git a/Data/Repa/Flow/Chunked/Process.hs b/Data/Repa/Flow/Chunked/Process.hs new file mode 100644 --- /dev/null +++ b/Data/Repa/Flow/Chunked/Process.hs @@ -0,0 +1,81 @@ + +module Data.Repa.Flow.Chunked.Process + ( process_i + , unfolds_i, A.StepUnfold(..)) +where +import Data.Repa.Flow.Chunked.Base +import qualified Data.Repa.Array.Generic as A +import qualified Data.Repa.Array.Generic.Index as A +import qualified Data.Repa.Flow.Generic.Base as G +#include "repa-flow.h" + + +-- | Apply a generic stream process to all the streams in a bundle of sources. +process_i + :: ( G.States i m + , A.BulkI lSrc a + , A.Bulk lDst b, A.Bulk lDst (A.Array lDst b) + , A.TargetI lDst b, A.TargetI lDst (A.Array lDst b)) + => (s -> a -> (s, A.Array lDst b)) -- ^ Worker function. + -> s -- ^ Initial state. + -> Sources i m lSrc a -- ^ Input sources + -> m (Sources i m lDst b) -- ^ Result sources. + +process_i f z (G.Sources n pullA) + = do + refs <- G.newRefs n z + + let pull_process i eatB ejectB + = do s1 <- G.readRefs refs i + pullA i (eatA_process s1) ejectA_process + + where eatA_process s1 xA + = case A.process A.name f s1 xA of + (s1', arrB) + -> do G.writeRefs refs i s1' + eatB arrB + {-# INLINE eatA_process #-} + + ejectA_process + = ejectB + {-# INLINE ejectA_process #-} + {-# INLINE pull_process #-} + + return $ G.Sources n pull_process +{-# INLINE_FLOW process_i #-} + + +-- | Apply a generic stream process to all the streams in a bundle of sources. +unfolds_i + :: ( G.States i m + , A.BulkI lSrc a + , A.Bulk lDst b + , A.TargetI lDst b) + => (a -> s -> A.StepUnfold s b) -- ^ Worker function. + -> s -- ^ Initial state. + -> Sources i m lSrc a -- ^ Input sources + -> m (Sources i m lDst b) -- ^ Result sources. + +unfolds_i f z (G.Sources n pullA) + = do + refs <- G.newRefs n z + + let pull_unfolds i eatB ejectB + = do s1 <- G.readRefs refs i + pullA i (eatA_unfolds s1) ejectA_unfolds + + where eatA_unfolds s1 xA + = case A.unfolds A.name f s1 xA of + (s1', arrB) + -> do G.writeRefs refs i s1' + eatB arrB + {-# INLINE eatA_unfolds #-} + + ejectA_unfolds + = ejectB + {-# INLINE ejectA_unfolds #-} + {-# INLINE pull_unfolds #-} + + return $ G.Sources n pull_unfolds +{-# INLINE_FLOW unfolds_i #-} + diff --git a/Data/Repa/Flow/Chunked/Replicate.hs b/Data/Repa/Flow/Chunked/Replicate.hs new file mode 100644 --- /dev/null +++ b/Data/Repa/Flow/Chunked/Replicate.hs @@ -0,0 +1,19 @@ + +module Data.Repa.Flow.Chunked.Replicate + ( replicates_i ) +where +import Data.Repa.Flow.Chunked.Base +import qualified Data.Repa.Flow.Generic as G +import qualified Data.Repa.Array.Generic as A + + +replicates_i + :: ( Flow i m lSrc (Int, a) + , A.TargetI lDst a) + => A.Name lDst + -> Sources i m lSrc (Int, a) + -> m (Sources i m lDst a) + +replicates_i nDst ss + = G.smap_i (\_ arr -> A.replicates nDst arr) ss +{-# INLINE replicates_i #-} diff --git a/Data/Repa/Flow/Generic.hs b/Data/Repa/Flow/Generic.hs --- a/Data/Repa/Flow/Generic.hs +++ b/Data/Repa/Flow/Generic.hs @@ -18,6 +18,7 @@ -- * Evaluation , drainS , drainP + , consumeS -- * Conversion , fromList @@ -52,6 +53,11 @@ , smap_i, smap_o , szipWith_ii, szipWith_io, szipWith_oi + -- ** Processing + , compact_i + , scan_i + , indexed_i + -- ** Connecting , dup_oo, dup_io, dup_oi , connect_i @@ -80,8 +86,8 @@ , rcapture_o -- ** Ignorance - , discard_o , ignore_o + , abandon_o -- ** Tracing , trace_o @@ -101,7 +107,7 @@ , dshuffleBy_o -- ** Chunking - , chunk_i + , chunkOn_i , unchunk_i) where import Data.Repa.Flow.States @@ -109,6 +115,7 @@ import Data.Repa.Flow.Generic.Connect import Data.Repa.Flow.Generic.List import Data.Repa.Flow.Generic.Map +import Data.Repa.Flow.Generic.Process import Data.Repa.Flow.Generic.Operator import Data.Repa.Flow.Generic.Eval import Data.Repa.Flow.Generic.Array.Distribute diff --git a/Data/Repa/Flow/Generic/Array/Chunk.hs b/Data/Repa/Flow/Generic/Array/Chunk.hs --- a/Data/Repa/Flow/Generic/Array/Chunk.hs +++ b/Data/Repa/Flow/Generic/Array/Chunk.hs @@ -1,27 +1,29 @@ module Data.Repa.Flow.Generic.Array.Chunk - (chunk_i) + (chunkOn_i) where import Data.Repa.Flow.Generic.Base import Data.Repa.Array.Generic as A -import Data.Repa.Array.Generic.Target as A import Data.Repa.Array.Generic.Index as A +import Data.Repa.Array.Generic.Target as A #include "repa-flow.h" --- | Take elements from a flow and pack them into chunks of the given --- maximum length. -chunk_i :: (Target lDst a, Index lDst ~ Int, States i IO) +-- | Take elements from a flow and pack them into chunks. +-- The chunks are limited to the given maximum length. +-- A predicate can also be supplied to detect the last element in a chunk. +chunkOn_i + :: (States i IO, TargetI lDst a) => Name lDst -- ^ Layout for result chunks. -> Int -- ^ Maximum chunk length. + -> (a -> Bool) -- ^ Detect the last element in a chunk. -> Sources i IO a -- ^ Element sources. -> IO (Sources i IO (Array lDst a)) -- ^ Chunk sources. -chunk_i nDst !maxLen (Sources n pullX) +chunkOn_i nDst !maxLen isEnd (Sources n pullX) = do - -- Refs for signalling how many elements we managed to read for - -- each chunk. - final <- newRefs n Nothing + -- Refs for signalling how many elements we managed to read for each chunk. + final <- newRefs n Nothing let pull_chunk i eat eject = do @@ -39,7 +41,12 @@ -- Write the next element to the chunk. eat_chunk x = do unsafeWriteBuffer chunk ix x - loop_chunk (ix + 1) + + -- Check if this element value signals the + -- end of a chunk. + if isEnd x + then writeRefs final i (Just $ ix + 1) + else loop_chunk (ix + 1) -- There are no more elements available from the soruce. eject_chunk @@ -70,5 +77,5 @@ {-# INLINE pull_chunk #-} return $ Sources n pull_chunk -{-# INLINE_FLOW chunk_i #-} +{-# INLINE_FLOW chunkOn_i #-} diff --git a/Data/Repa/Flow/Generic/Eval.hs b/Data/Repa/Flow/Generic/Eval.hs --- a/Data/Repa/Flow/Generic/Eval.hs +++ b/Data/Repa/Flow/Generic/Eval.hs @@ -1,7 +1,8 @@ module Data.Repa.Flow.Generic.Eval ( drainS - , drainP) + , drainP + , consumeS) where import Data.Repa.Flow.Generic.Base import Data.Repa.Eval.Gang as Eval @@ -70,3 +71,30 @@ {-# INLINE eject_drain #-} {-# INLINE drainMe #-} {-# INLINE_FLOW drainP #-} + + +-- | Pull all available values from the sources and pass them to the +-- given action. +consumeS :: (Next i, Monad m) + => (i -> a -> m ()) + -> Sources i m a + -> m () +consumeS eat (Sources nSources ipull) + = loop_consume first + where + loop_consume !ix + = ipull ix eat_consume eject_consume + where + eat_consume v + = do eat ix v + loop_consume ix + {-# INLINE eat_consume #-} + + eject_consume + = do case next ix nSources of + Nothing -> return () + Just ix' -> loop_consume ix' + {-# INLINE eject_consume #-} + {-# INLINE loop_consume #-} +{-# INLINE_FLOW consumeS #-} + diff --git a/Data/Repa/Flow/Generic/IO.hs b/Data/Repa/Flow/Generic/IO.hs --- a/Data/Repa/Flow/Generic/IO.hs +++ b/Data/Repa/Flow/Generic/IO.hs @@ -8,7 +8,10 @@ , sourceChars , sourceChunks , sourceRecords + , sourceLinesFormat + , sourceLinesFormatFromLazyByteString + -- * Sinking , sinkBytes , sinkChars @@ -24,3 +27,4 @@ import Data.Repa.Flow.IO.Bucket import Data.Repa.Flow.Generic.IO.Base as F import Data.Repa.Flow.Generic.IO.XSV as F +import Data.Repa.Flow.Generic.IO.Lines as F diff --git a/Data/Repa/Flow/Generic/IO/Base.hs b/Data/Repa/Flow/Generic/IO/Base.hs --- a/Data/Repa/Flow/Generic/IO/Base.hs +++ b/Data/Repa/Flow/Generic/IO/Base.hs @@ -21,7 +21,6 @@ import Data.Repa.Flow.Generic.IO.Sieve as F import Data.Repa.Flow.Generic.Map as F import Data.Repa.Flow.Generic.Base as F -import Data.Repa.Fusion.Unpack as F import Data.Repa.Array.Material as A import Data.Repa.Array.Meta.Delayed as A import Data.Repa.Array.Meta.Window as A @@ -163,7 +162,7 @@ sinkLines :: ( Bulk l Bucket , BulkI l1 (Array l2 Char) - , BulkI l2 Char, Unpack (Array l2 Char) t2) + , BulkI l2 Char) => Name l1 -- ^ Layout of chunks of lines. -> Name l2 -- ^ Layout of lines. -> Array l Bucket -- ^ Buckets. diff --git a/Data/Repa/Flow/Generic/IO/Lines.hs b/Data/Repa/Flow/Generic/IO/Lines.hs new file mode 100644 --- /dev/null +++ b/Data/Repa/Flow/Generic/IO/Lines.hs @@ -0,0 +1,126 @@ + +module Data.Repa.Flow.Generic.IO.Lines + ( sourceLinesFormat + , sourceLinesFormatFromLazyByteString) +where +import Data.Repa.Flow.Generic.IO.Base as F +import Data.Repa.Flow.Generic.Map as F +import Data.Repa.Flow.Generic.Base as F + +import Data.Repa.Array.Generic as A +import Data.Repa.Array.Material as A +import qualified Data.Repa.Array.Auto.Format as A + +import Data.Repa.Convert.Format as C + +import qualified Data.ByteString.Lazy as BSL +import Data.Char +import Data.Word +#include "repa-flow.h" + + +-- | Read lines from a named text file, +-- in a chunk-wise manner, +-- converting each line to values with the given format. +sourceLinesFormat + :: forall format + . (Unpackable format, Target A (Value format)) + => Integer -- ^ Chunk length. + -> IO () -- ^ Action if we find a line longer than the chunk length. + -> IO (Array A Word8 -> IO ()) -- ^ Action if we can't convert a row. + -> format -- ^ Format of each line. + -> Array B Bucket + -> IO (Sources Int IO (Array A (Value format))) + +sourceLinesFormat nChunk aFailLong _aFailConvert format bs + = do + -- Rows are separated by new lines. + let !nl = fromIntegral $ ord '\n' + let !nr = fromIntegral $ ord '\r' + + -- Stream chunks of data from the input file, + -- where the chunks end cleanly and line boundaries. + -- Filter out any stray CR characters allong the way. + sChunk <- sourceChunks nChunk (== nl) aFailLong bs + + sRows8 :: Sources Int IO (Array N (Array F Word8)) + <- map_i ( A.trimEnds (== nl) + . A.segmentOn (== nl) + . A.filter F (/= nr)) + sChunk + + -- Convert each value using the given format. + let unpackRow :: Array A Word8 -> Value format + unpackRow arr + = case A.unpackFormat format arr of + Nothing -> error ("no convert " ++ show arr) + -- TODO: impl proper pull function + -- so we can call aFailConvert if needed. + -- We shouldn't be throwing errors this deep in the library. + Just v -> v + {-# INLINE unpackRow #-} + + F.map_i (A.mapS A (unpackRow . A.convert A)) sRows8 +{-# INLINE sourceLinesFormat #-} + + +-- | Read lines from a lazy byte string, +-- in a chunk-wise manner, +-- converting each line to values with the given format. +sourceLinesFormatFromLazyByteString + :: (Unpackable format, Target A (Value format)) + => Int -- ^ Number of streams in the result bundle. + -> IO (Array A Word -> IO ()) -- ^ Action if we can't convert a row. + -> format -- ^ Format of each line. + -> BSL.ByteString -- ^ Lazy byte string. + -> Int -- ^ Skip this many header lines at the start. + -> IO (Sources Int IO (Array A (Value format))) + +sourceLinesFormatFromLazyByteString n _aFailConvert format bs0 nSkip + = do + -- Rows are separated by new lines. + let !nl = fromIntegral $ ord '\n' + let !nr = fromIntegral $ ord '\r' + + -- Give a copy of the bytestring to each stream. + refsBS <- newRefs n bs0 + refsSkip <- newRefs n nSkip + + let unpackRow arr + = case A.unpackFormat format arr of + Nothing -> error ("no convert " + ++ (show $ map (chr . fromIntegral) + $ A.toList arr)) + -- TODO: imlp proper pull function + -- so that we can call aFailConvert if needed. + -- We shouldn't be throwing errors this deep in the library. + Just v -> v + {-# INLINE unpackRow #-} + + let pull_fromString i eat eject + = do bs <- readRefs refsBS i + skip <- readRefs refsSkip i + if BSL.null bs + then eject + else do let (bsLine, bsRest) + = BSL.break (== nl) bs + + writeRefs refsBS i + $ BSL.dropWhile (== nl) bsRest + + if (skip >= 0) + then do writeRefs refsSkip i (skip - 1) + pull_fromString i eat eject + + else eat $ A.singleton A.A + $ unpackRow + $ A.convert A.A + $ A.fromByteString + $ BSL.toStrict + $ BSL.filter (/= nr) bsLine + {-# INLINE pull_fromString #-} + + return $ Sources n pull_fromString +{-# INLINE_FLOW sourceLinesFormatFromLazyByteString #-} + + diff --git a/Data/Repa/Flow/Generic/IO/Sieve.hs b/Data/Repa/Flow/Generic/IO/Sieve.hs --- a/Data/Repa/Flow/Generic/IO/Sieve.hs +++ b/Data/Repa/Flow/Generic/IO/Sieve.hs @@ -4,47 +4,131 @@ where import Data.Repa.Flow.Generic.Base import Data.Repa.Array.Material as A -import Data.Repa.Array.Generic.Convert as A +import Data.Repa.Array.Generic as A import Data.Repa.Array.Auto.IO as A +import qualified Data.HashTable.IO as Hash +import qualified System.Mem as System import System.IO import Data.Word +import Data.IORef + +import qualified Data.Vector.Mutable as M +import qualified Data.Vector as V + #include "repa-flow.h" -- | Create an output sieve that writes data to an indeterminate number of -- output files. Each new element is appended to its associated file. -- --- * TODO: --- This function keeps a maximum of 8 files open at once, closing --- and re-opening them in a least-recently-used order. --- Due to this behaviour it's fine to create thousands of separate --- output files without risking overflowing the process limit on --- the maximum number of useable file handles. +--- +-- We don't want to open and close a file every time we receieve data. +-- To avoid this, we instead batch data in memory for each file until +-- we have enough to warrant performing the IO operation. -- -sieve_o :: (a -> Maybe (FilePath, Array F Word8)) +sieve_o :: Int -- ^ Max payload size of in-memory data. + -> Int -- ^ Max number of in-memory chunks. + -> (a -> Maybe (FilePath, Array F Word8)) -- ^ Produce the desired file path and output -- record for this element, or `Nothing` if -- it should be discarded. -> IO (Sinks () IO a) -sieve_o diag +sieve_o sizeLimit chunksLimit diag = do + -- Store an array of chunks for each file. + -- We use a mutable vector of chunks, and store the number of used + -- slots in that vector separately. + (ht :: Hash.CuckooHashTable FilePath (Int, M.IOVector (Array F Word8))) + <- Hash.newSized 1024 - let push_sieve _ e + !refSize <- newIORef 0 + !refChunks <- newIORef 0 + + -- Flush the chunks for a single file to disk. + let flush_path (path, (n, mvec)) + = do + !vec <- V.unsafeFreeze mvec + + !h <- openBinaryFile path AppendMode + + -- Write out chunks for this file. + V.mapM_ (hPutArray h . convert A) + $ V.slice 0 n vec + + hClose h + + -- Delete the entry from the hash table. + -- This allows the space for the mutable vector of chunks to be reclaimed. + Hash.delete ht path + + + -- Flush all the chunks we have stored. + let flush_all + = do Hash.mapM_ flush_path ht + + -- Remember that we've accumulated this chunk into memory. + -- When we end up with too much data then we flush the whole lot + -- to the file system. + let acc_size !len + = do !sizeCurrent <- readIORef refSize + !chunksCurrent <- readIORef refChunks + + if (sizeCurrent + len) > sizeLimit + || (chunksCurrent + 1) > chunksLimit + then do + flush_all + writeIORef refSize 0 + writeIORef refChunks 0 + + else do + let !sizeCurrent' = sizeCurrent + len + let !chunksCurrent' = chunksCurrent + 1 + writeIORef refSize sizeCurrent' + writeIORef refChunks chunksCurrent' + + -- Accept a single incoming element. + let push_sieve _ !e + = case diag e of + + -- The provided diag function told us to drop this + -- element on the floor. Nothing -> return () + -- Accumulate a new chunk. Just (path, arr) - -> do -- TODO: repeatededly opening and closing the file - -- will be very slow. - h <- openBinaryFile path AppendMode - hPutArray h (convert arr) - hClose h + -> do + -- See if we already have a buffer for this file. + !mElem <- Hash.lookup ht path + case mElem of - -- TODO: ignore any more incoming data after being ejected. + -- We haven't seen chunks for this file before, + -- so create a new vector to hold them. + Nothing + -> do !mvec <- M.new 256 + M.write mvec 0 arr + Hash.insert ht path (1, mvec) + acc_size (A.length arr) + + -- We already have a chunk vector for this file. + Just (n, mvec) + -> do + -- If the chunk vector has no space the expand it. + !mvec' <- if n >= M.length mvec + then M.grow mvec (M.length mvec) + else return mvec + + M.write mvec' n arr + let !n' = n + 1 + Hash.insert ht path (n', mvec') + acc_size (A.length arr) + let eject_sieve _ - = return () + = do flush_all + System.performMajorGC return $ Sinks () push_sieve eject_sieve {-# INLINE sieve_o #-} + diff --git a/Data/Repa/Flow/Generic/Operator.hs b/Data/Repa/Flow/Generic/Operator.hs --- a/Data/Repa/Flow/Generic/Operator.hs +++ b/Data/Repa/Flow/Generic/Operator.hs @@ -34,8 +34,8 @@ , rcapture_o -- * Ignorance - , discard_o , ignore_o + , abandon_o -- * Tracing , trace_o) @@ -352,7 +352,7 @@ = atomicModifyIORef ref (\old -> ((i, x) : old, ())) {-# INLINE capture_eat #-} - k0 <- discard_o n + k0 <- ignore_o n k1 <- watch_o capture_eat k0 x <- use k1 @@ -367,43 +367,43 @@ trigger_o :: Monad m => i -> (i -> a -> m ()) -> m (Sinks i m a) trigger_o i f - = discard_o i >>= watch_o f + = ignore_o i >>= watch_o f {-# INLINE trigger_o #-} -- Ignorance------------------------------------------------------------------- --- | A sink that drops all data on the floor. +-- | A sink that ignores all incoming data. -- --- This sink is strict in the elements, so they are demanded before being --- discarded. Haskell debugging thunks attached to the elements will be --- demanded. +-- * This sink is strict in the elements, so they are demanded before being +-- discarded. +-- * Haskell debugging thunks attached to the elements will be +-- demanded. -- -discard_o :: Monad m +ignore_o :: Monad m => i -> m (Sinks i m a) -discard_o n - = return $ Sinks n push_discard eject_discard +ignore_o n + = return $ Sinks n push_ignore eject_ignore where - -- IMPORTANT: push_discard should be strict in the element so that + -- IMPORTANT: push_ignore should be strict in the element so that -- and Haskell tracing thunks attached to it are evaluated. - -- We *discard* the elements, but don't completely ignore them. - push_discard !_ !_ = return () - eject_discard !_ = return () -{-# INLINE_FLOW discard_o #-} + push_ignore !_ !_ = return () + eject_ignore !_ = return () +{-# INLINE_FLOW ignore_o #-} --- | A sink that ignores all incoming data. +-- | A sink that drops all data on the floor. -- --- This sink is non-strict in the elements. --- Haskell tracing thunks attached to the elements will *not* be demanded. +-- * This sink is non-strict in the elements. +-- * Haskell tracing thunks attached to the elements will *not* be demanded. -- -ignore_o :: Monad m +abandon_o :: Monad m => i -> m (Sinks i m a) -ignore_o n - = return $ Sinks n push_ignore eject_ignore +abandon_o n + = return $ Sinks n push_abandon eject_abandon where - push_ignore _ _ = return () - eject_ignore _ = return () -{-# INLINE_FLOW ignore_o #-} + push_abandon _ _ = return () + eject_abandon _ = return () +{-# INLINE_FLOW abandon_o #-} -- Trace ---------------------------------------------------------------------- @@ -424,6 +424,4 @@ (return ()) {-# NOINLINE trace_o #-} - - diff --git a/Data/Repa/Flow/Generic/Process.hs b/Data/Repa/Flow/Generic/Process.hs new file mode 100644 --- /dev/null +++ b/Data/Repa/Flow/Generic/Process.hs @@ -0,0 +1,87 @@ + +module Data.Repa.Flow.Generic.Process + ( -- * Processing + compact_i + + -- * Scanning + , scan_i + + -- * Indexed streams + , indexed_i) +where +import Data.Repa.Flow.Generic.Base + + +-- | Combination of 'fold' and 'filter'. +-- +-- We walk over the stream start-to-end, maintaining an accumulator. +-- At each point we can chose to emit an element, or not. +-- +compact_i + :: (Monad m, States i m) + => (s -> a -> (s, Maybe b)) + -> s + -> Sources i m a + -> m (Sources i m b) + +compact_i f s0 (Sources n pullA) + = do + refs <- newRefs n s0 + + let pull_process i eatB ejectB + = do s1 <- readRefs refs i + pullA i (eatA_process s1) ejectA_process + + where eatA_process s1 xA + = do case f s1 xA of + (s2', Nothing) + -> do writeRefs refs i s2' + pull_process i eatB ejectB + + (s2', Just xB) + -> do writeRefs refs i s2' + eatB xB + {-# INLINE eatA_process #-} + + ejectA_process = ejectB + {-# INLINE ejectA_process #-} + {-# INLINE pull_process #-} + + return $ Sources n pull_process +{-# INLINE compact_i #-} + + +-- | Start-to-end scan over each stream in a bundle. +scan_i + :: (Monad m, States i m) + => (s -> a -> s) + -> s + -> Sources i m a + -> m (Sources i m s) + +scan_i f s0 ss + = compact_i work_scan s0 ss + where + work_scan s x + = let s' = f s x + in (s', Just s') + {-# INLINE work_scan #-} +{-# INLINE scan_i #-} + + +-- | For each stream in a bundle of sources, +-- associated the element with their corresponding position in the stream. +-- +indexed_i + :: (Monad m, States i m) + => Sources i m a + -> m (Sources i m (Int, a)) + +indexed_i ss + = compact_i work_indexed 0 ss + where + work_indexed s x + = (s + 1, Just (s, x)) + {-# INLINE work_indexed #-} +{-# INLINE indexed_i #-} + diff --git a/Data/Repa/Flow/IO/Binary.hs b/Data/Repa/Flow/IO/Binary.hs deleted file mode 100644 --- a/Data/Repa/Flow/IO/Binary.hs +++ /dev/null @@ -1,38 +0,0 @@ - -module Data.Repa.Flow.IO.Binary - ( sourceBinary ) -where -import Data.Repa.Array.Material as A -import Data.Repa.Array.Generic as A -import Data.Repa.Flow.IO.Storable as F -import Data.Repa.Flow.IO.Bucket as F -import Data.Repa.Flow.Generic as G -#include "repa-flow.h" - - --- Move to Data.Repa.Flow.Binary -sourceBinary - :: F.Storable a - => Spec a -- ^ Specification of elements. - -> Integer -- ^ Number of elements per chunk. - -> Array B Bucket -- ^ Buckets of table. - -> IO (G.Sources Int IO (Array (Rep a) a)) - -sourceBinary spec lenElems bs - = return $ G.Sources (A.length bs) pull_sourceBinary - where - pull_sourceBinary i eat eject - = do let b = A.index bs i - op <- bIsOpen b - if not op - then eject - else do - eof <- bAtEnd b - if eof - then eject - else do - Just !chunk <- getArray spec lenElems b - eat chunk - {-# INLINE pull_sourceBinary #-} -{-# INLINE_FLOW sourceBinary #-} - diff --git a/Data/Repa/Flow/IO/Bucket.hs b/Data/Repa/Flow/IO/Bucket.hs --- a/Data/Repa/Flow/IO/Bucket.hs +++ b/Data/Repa/Flow/IO/Bucket.hs @@ -13,7 +13,7 @@ -- * Writing , toFiles, toFiles' , toDir - , toDirs' + , toDirs -- * Bucket IO , bClose @@ -86,12 +86,13 @@ -- | Wrap an existing file handle as a bucket. +-- +-- The starting position is set to 0. hBucket :: Handle -> IO Bucket hBucket h - = do pos <- hTell h - return $ Bucket + = return $ Bucket { bucketFilePath = Nothing - , bucketStartPos = pos + , bucketStartPos = 0 , bucketLength = Nothing , bucketHandle = h } {-# NOINLINE hBucket #-} @@ -99,14 +100,24 @@ -- From Files ----------------------------------------------------------------- -- | Open some files as buckets and use them as `Sources`. -fromFiles +fromFiles + :: [FilePath] + -> (Array B Bucket -> IO b) + -> IO b +fromFiles files use + = fromFiles' (A.fromList B files) use +{-# INLINE fromFiles #-} + + +-- | Like `fromFiles'`, but take a list of file paths. +fromFiles' :: (Bulk l FilePath, Target l Bucket) => Array l FilePath -- ^ Files to open. -> (Array l Bucket -> IO b) -- ^ Consumer. -> IO b -fromFiles paths use +fromFiles' paths use = do -- Open all the files, ending up with a list of buckets. bs <- mapM (flip openBucket ReadMode) $ A.toList paths @@ -116,19 +127,10 @@ let Just bsArr = A.fromListInto (A.layout paths) bs use bsArr -{-# NOINLINE fromFiles #-} - - --- | Like `fromFiles`, but take a list of file paths. -fromFiles' - :: [FilePath] - -> (Array B Bucket -> IO b) - -> IO b -fromFiles' files use - = fromFiles (A.fromList B files) use -{-# INLINE fromFiles' #-} +{-# NOINLINE fromFiles' #-} +-- From Dirs ------------------------------------------------------------------ -- | Open all the files in a directory as separate buckets. -- -- This operation may fail with the same exceptions as `getDirectoryContents`. @@ -142,7 +144,7 @@ let fsRel = P.map (dir ) $ P.filter (\f -> f /= "." && f /= "..") fs - fromFiles' fsRel use + fromFiles fsRel use {-# INLINE fromDir #-} @@ -265,14 +267,23 @@ -- Writing -------------------------------------------------------------------- -- | Open some files for writing as individual buckets and pass -- them to the given consumer. --- -toFiles :: (Bulk l FilePath, Target l Bucket) - => Array l FilePath -- ^ File paths. - -> (Array l Bucket -> IO b) - -- ^ Consumer. +toFiles :: [FilePath] -- ^ File paths. + -> (Array B Bucket -> IO b) -- ^ Worker writes data to buckets. -> IO b toFiles paths use + = toFiles' (A.fromList B paths) use +{-# INLINE toFiles #-} + + +-- | Like `toFiles`, but take an array of `FilePath`s. +toFiles' :: (Bulk l FilePath, Target l Bucket) + => Array l FilePath -- ^ File paths. + -> (Array l Bucket -> IO b) -- ^ Worker writes data to buckets. + -- ^ Consumer. + -> IO b + +toFiles' paths use = do -- Open all the files, ending up with a list of buckets. bs <- mapM (flip openBucket WriteMode) $ A.toList paths @@ -281,22 +292,12 @@ let Just bsArr = A.fromListInto (A.layout paths) bs use bsArr -{-# NOINLINE toFiles #-} +{-# NOINLINE toFiles' #-} --- -- TODO: Attached finalizers to the sinks so that file assocated with -- each stream is closed when that stream is ejected. --- | Like `toFiles`, but take a list of file paths. -toFiles' :: [FilePath] - -> (Array B Bucket -> IO b) - -> IO b - -toFiles' paths use - = toFiles (A.fromList B paths) use -{-# INLINE toFiles' #-} - - -- | Create a new directory of the given name, containing the given number -- of buckets. -- @@ -342,13 +343,13 @@ -- For each directory @somedir@ the files are named -- @somedir/000000@, @somedir/000001@, @somedir/000002@ and so on. -- -toDirs' :: Int -- ^ Number of buckets to create per directory. +toDirs :: Int -- ^ Number of buckets to create per directory. -> [FilePath] -- ^ Paths to directories. -> (Array (E B DIM2) Bucket -> IO b) -- ^ Consumer. -> IO b -toDirs' nBucketsPerDir paths use +toDirs nBucketsPerDir paths use | nBucketsPerDir <= 0 = do let Just bsArr = A.fromListInto (A.matrix B 0 0) [] @@ -384,7 +385,7 @@ bs use bsArr -{-# NOINLINE toDirs' #-} +{-# NOINLINE toDirs #-} diff --git a/Data/Repa/Flow/IO/Storable.hs b/Data/Repa/Flow/IO/Storable.hs deleted file mode 100644 --- a/Data/Repa/Flow/IO/Storable.hs +++ /dev/null @@ -1,135 +0,0 @@ - -module Data.Repa.Flow.IO.Storable - ( Storable (..) - , Spec (..)) -where -import Data.Repa.Array.Meta as A -import Data.Repa.Array.Generic as A -import Data.Repa.Array.Material as A -import Data.Repa.Array.Material.Foreign as AF -import Data.Repa.Array.Material.Strided as AS -import Data.Repa.Flow.IO.Bucket as F -import qualified Foreign.Storable as S -import Data.Int -#include "repa-flow.h" - - ---------------------------------------------------------------------------------------------------- --- | Class of element types that we can load and store to the file system. --- --- TODO: change to Persistable. --- -class Bulk (Rep a) a - => Storable a where - - -- | Specification of how the elements are arranged in the file. - -- - -- For atomic elements the specification is the storage type. - -- - -- For fixed-length arrays of elements, the specification contains the - -- element storage type as well as the array length. - -- - data Spec a - - -- | Representation tag used for an array of these elements. - -- - -- For atomic elements this will be `F`, for foreign arrays that - -- do not require extra copying after the data is read. - -- - -- For tuples of elements, this will be a tuple of strided arrays, - -- so the elements can also be used without copying. - -- - type Rep a - - -- | Get the size of a single element, in bytes. - sizeElem :: Spec a -> Int - - -- | Read an array of the given length from a bucket. - -- If the bucket does not contain a whole number of elements remaining - -- then `Nothing`. - getArray - :: Spec a -- ^ Element specification. - -> Integer -- ^ Number of elements to read. - -> Bucket -- ^ Bucket to read from. - -> IO (Maybe (Array (Rep a) a)) - - -- | Write an array to a bucket. -{- - putArray - :: Spec a -- ^ Element specification. - -> Bucket -- ^ Bucket to write to. - -> Array (Rep a) a -- ^ Array to write. - -> IO () --} - ---------------------------------------------------------------------------------------------------- --- | A native 32-bit integer. -instance Storable Int32 where - data Spec Int32 = SInt32 - type Rep Int32 = F - - sizeElem SInt32 = 4 - {-# INLINE_FLOW sizeElem #-} - - getArray SInt32 lenElems bucket - = do let bytesElem = sizeElem SInt32 - arr8 <- bGetArray bucket (lenElems * fromIntegral bytesElem) - let (startBytes, lenBytes, fptrBuf) - = AF.toForeignPtr arr8 - let lenElems' = lenBytes `div` sizeElem SInt32 - - if (startBytes /= 0) - || (lenBytes `mod` bytesElem /= 0) - then return Nothing - else return $ Just - $ AF.unsafeCast - $ AF.fromForeignPtr lenElems' fptrBuf - - - ---------------------------------------------------------------------------------------------------- --- | Two elements stored consecutively. -instance - ( Storable a, Storable b - , S.Storable a, S.Storable b) - => Storable (a, b) where - - data Spec (a, b) = S2 (Spec a) (Spec b) - type Rep (a, b) = T2 S S - - sizeElem (S2 sA sB) - = sizeElem sA + sizeElem sB - {-# INLINE_FLOW sizeElem #-} - - getArray (S2 sA sB) lenElems bucket - = do let bytesA = sizeElem sA - let bytesB = sizeElem sB - let bytesTuple = bytesA + bytesB - - -- Read an array containing raw bytes. - let lenBytes = lenElems * fromIntegral bytesTuple - arr8 <- bGetArray bucket lenBytes - - let lenBytes' = A.length arr8 - let lenElems' = lenBytes' `div` bytesTuple - - -- Check that we have received a whole number of elements. - if lenBytes' `mod` bytesTuple /= 0 - then return Nothing - else do - let (startBuf, _lenBuf, fptrBuf) - = AF.toForeignPtr arr8 - - let arrA = AS.unsafeCast - $ AS.fromForeignPtr - startBuf - bytesTuple (fromIntegral lenElems') fptrBuf - - let arrB = AS.unsafeCast - $ AS.fromForeignPtr - (startBuf + bytesA) - bytesTuple (fromIntegral lenElems') fptrBuf - - return $ Just $ T2Array arrA arrB - - diff --git a/Data/Repa/Flow/Simple.hs b/Data/Repa/Flow/Simple.hs --- a/Data/Repa/Flow/Simple.hs +++ b/Data/Repa/Flow/Simple.hs @@ -48,8 +48,8 @@ , trigger_o -- ** Ignorance - , discard_o , ignore_o + , abandon_o -- * Flow IO -- ** Sourcing diff --git a/Data/Repa/Flow/Simple/Operator.hs b/Data/Repa/Flow/Simple/Operator.hs --- a/Data/Repa/Flow/Simple/Operator.hs +++ b/Data/Repa/Flow/Simple/Operator.hs @@ -31,8 +31,8 @@ , trigger_o -- * Ignorance - , discard_o - , ignore_o) + , ignore_o + , abandon_o) where import Data.Repa.Flow.Simple.Base import Data.Repa.Flow.States (States (..)) @@ -204,22 +204,22 @@ -- Ignorance ------------------------------------------------------------------ --- | A sink that drops all data on the floor. +-- | A sink that ignores all incoming elements. -- -- This sink is strict in the elements, so they are demanded before being -- discarded. Haskell debugging thunks attached to the elements will be demanded. -discard_o :: Monad m +ignore_o :: Monad m => m (Sink m a) -discard_o = G.discard_o () -{-# INLINE discard_o #-} +ignore_o = G.ignore_o () +{-# INLINE ignore_o #-} --- | A sink that ignores all incoming elements. +-- | A sink that drops all data on the floor. -- -- This sink is non-strict in the elements. -- Haskell tracing thinks attached to the elements will *not* be demanded. -ignore_o :: Monad m +abandon_o :: Monad m => m (Sink m a) -ignore_o = G.ignore_o () -{-# INLINE ignore_o #-} +abandon_o = G.abandon_o () +{-# INLINE abandon_o #-} diff --git a/repa-flow.cabal b/repa-flow.cabal --- a/repa-flow.cabal +++ b/repa-flow.cabal @@ -1,5 +1,5 @@ Name: repa-flow -Version: 4.1.0.1 +Version: 4.2.2.1 License: BSD3 License-file: LICENSE Author: The Repa Development Team @@ -15,23 +15,28 @@ Library build-Depends: - base == 4.7.*, + base == 4.8.*, directory == 1.2.*, - filepath == 1.3.*, + filepath == 1.4.*, vector == 0.10.*, bytestring == 0.10.*, - primitive == 0.5.4.*, + primitive == 0.6.*, containers == 0.5.*, + hashtables == 1.2.1.*, text == 1.2.*, repa-eval == 4.0.0.*, - repa-stream == 4.1.0.*, - repa-array == 4.1.0.* + repa-stream == 4.2.2.*, + repa-scalar == 4.2.2.*, + repa-convert == 4.2.2.*, + repa-array == 4.2.2.* exposed-modules: Data.Repa.Flow.Auto Data.Repa.Flow.Auto.Debug Data.Repa.Flow.Auto.IO + Data.Repa.Flow.Auto.Format Data.Repa.Flow.Auto.SizedIO + Data.Repa.Flow.Auto.ZipWith Data.Repa.Flow.Chunked Data.Repa.Flow.Chunked.IO @@ -40,9 +45,7 @@ Data.Repa.Flow.Generic.Debug Data.Repa.Flow.Generic.IO - Data.Repa.Flow.IO.Binary Data.Repa.Flow.IO.Bucket - Data.Repa.Flow.IO.Storable Data.Repa.Flow.Simple @@ -51,18 +54,24 @@ Data.Repa.Flow other-modules: + Data.Repa.Flow.Auto.Base + Data.Repa.Flow.Auto.Select + Data.Repa.Flow.Chunked.Base Data.Repa.Flow.Chunked.Map Data.Repa.Flow.Chunked.Fold Data.Repa.Flow.Chunked.Folds Data.Repa.Flow.Chunked.Groups - Data.Repa.Flow.Chunked.Operator + Data.Repa.Flow.Chunked.Generic + Data.Repa.Flow.Chunked.Process + Data.Repa.Flow.Chunked.Replicate Data.Repa.Flow.Generic.Base Data.Repa.Flow.Generic.Connect Data.Repa.Flow.Generic.List Data.Repa.Flow.Generic.Map Data.Repa.Flow.Generic.Operator + Data.Repa.Flow.Generic.Process Data.Repa.Flow.Generic.Eval Data.Repa.Flow.Generic.Array.Distribute Data.Repa.Flow.Generic.Array.Shuffle @@ -71,6 +80,7 @@ Data.Repa.Flow.Generic.IO.Base Data.Repa.Flow.Generic.IO.Sieve Data.Repa.Flow.Generic.IO.XSV + Data.Repa.Flow.Generic.IO.Lines Data.Repa.Flow.Simple.Base Data.Repa.Flow.Simple.List @@ -91,20 +101,19 @@ extensions: CPP - BangPatterns - NoMonomorphismRestriction - RankNTypes MagicHash + RankNTypes + TypeFamilies + BangPatterns + PatternGuards + ConstraintKinds + ParallelListComp FlexibleContexts FlexibleInstances - PatternGuards - TypeFamilies - MultiParamTypeClasses + StandaloneDeriving ScopedTypeVariables + MultiParamTypeClasses FunctionalDependencies - ConstraintKinds ForeignFunctionInterface - StandaloneDeriving - ParallelListComp + NoMonomorphismRestriction -