diff --git a/Data/Repa/Flow.hs b/Data/Repa/Flow.hs --- a/Data/Repa/Flow.hs +++ b/Data/Repa/Flow.hs @@ -8,9 +8,11 @@ -- be read in chunks, using the default chunk size of 64kBytes. -- -- @ --- > import Data.Repa.Flow --- > import Data.Repa.Flow.Default.Debug --- > ws <- fromFiles [\"\/usr\/share\/dict\/words\", \"\/usr\/share\/dict\/cracklib-small\"] sourceLines +-- > import Data.Repa.Array as A +-- > import Data.Repa.Flow as F +-- > import Data.Repa.Flow.Auto.Debug as F +-- +-- > 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. @@ -51,17 +53,11 @@ -- -- @ -- > import Data.Char --- > up <- map_i B (mapS U toUpper) ws +-- > up <- F.map_i (A.map toUpper) ws -- > more 0 up -- Just [\"UTOPIAN\",\"UTOPIAN'S\",\"UTOPIANS\",\"UTOPIAS\",\"UTRECHT\" ...] -- @ -- --- The `B` and `U` are `Layout` names that indicate how the chunks for the --- result streams should be arranged in memory. In this case the chunks --- are `B`-oxed arrays of `U`-nboxed arrays of characters. Other useful --- layouts are `F` which stores data in foreign memory, and `N` for nested --- arrays. --- -- Flows are data-parallel, which means operators like `map_i` apply to all -- streams in the bundle. The second stream has been converted to upper-case -- as well: @@ -75,7 +71,7 @@ -- so open a file for each stream: -- -- @ --- > out <- toFiles ["out1.txt", "out2.txt"] $ sinkLines B U +-- > out <- toFiles' ["out1.txt", "out2.txt"] sinkLines -- @ -- -- Note that the @ws@ and @up@ we used before were bundles of stream @@ -88,7 +84,7 @@ -- we can `drainS` all of the data from the former into the latter. -- -- @ --- > drainS up out +-- > F.drainS up out -- @ -- -- At this point we can run an external shell command to check the output. @@ -122,12 +118,6 @@ , sourcesArity , sinksArity - -- * States and Arrays - , module Data.Repa.Flow.States - , module Data.Repa.Eval.Array - , module Data.Repa.Array - , module Data.Repa.Array.Material - -- * Evaluation , drainS , drainP @@ -169,7 +159,10 @@ , GroupsDict -- ** Folding + -- *** Complete , foldlS, foldlAllS + + -- *** Segmented , folds_i, FoldsDict , foldGroupsBy_i, FoldGroupsDict @@ -186,22 +179,23 @@ , sourceLines , sourceChars , sourceBytes + , sourcePacked -- ** Sinking , sinkChars , sinkLines - , sinkBytes) + , sinkBytes + , sinkPacked + ) where -import Data.Repa.Flow.Default -import Data.Repa.Flow.Default.Debug -import Data.Repa.Flow.Default.IO +import Data.Repa.Flow.Auto +import Data.Repa.Flow.Auto.Debug +import Data.Repa.Flow.Auto.IO import Data.Repa.Flow.IO.Bucket import Data.Repa.Flow.States -import Data.Repa.Eval.Array - -import Data.Repa.Array - hiding (fromList, Index, GroupsDict, FoldsDict) +import Data.Repa.Array.Generic + hiding (fromList, GroupsDict, FoldsDict) import Data.Repa.Array.Material hiding (fromLists) diff --git a/Data/Repa/Flow/Auto.hs b/Data/Repa/Flow/Auto.hs new file mode 100644 --- /dev/null +++ b/Data/Repa/Flow/Auto.hs @@ -0,0 +1,499 @@ + +-- | This module defines the default specialisation of flows that +-- appears in "Data.Repa.Flow". Each stream in the bundle is indexed +-- by a single integer, and stream state is stored using the IO monad. +-- +module Data.Repa.Flow.Auto + ( -- * Flow types + Sources + , Sinks + , Flow + , sourcesArity + , sinksArity + + -- * Conversion + , fromList, fromLists + , toList1, toLists1 + + -- * Evaluation + , drainS + , drainP + + -- * Flow Operators + -- ** 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 + + -- ** Connecting + , dup_oo + , dup_io + , dup_oi + , connect_i + + -- ** Watching + , watch_i, watch_o + , trigger_o + + -- ** Ignorance + , discard_o + , ignore_o + + -- ** Splitting + , head_i + + -- ** Grouping + , groups_i + , groupsBy_i + , GroupsDict + + -- ** Folding + -- *** Complete + , foldlS, foldlAllS + + -- *** Segmented + , folds_i, FoldsDict + , foldGroupsBy_i, FoldGroupsDict + + -- * Finalizers + , finalize_i, finalize_o + ) +where +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) +import Control.Monad +#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. +-- +-- * If the `Sources` and `Sinks` have different numbers of streams then +-- we only evaluate the common subset. +-- +drainS :: Sources a -> Sinks a -> IO () +drainS = G.drainS +{-# INLINE drainS #-} + + +-- | Pull all available values from the sources and push them to the sinks, +-- in parallel. We fork a thread for each of the streams and evaluate +-- them all in parallel. +-- +-- * If the `Sources` and `Sinks` have different numbers of streams then +-- we only evaluate the common subset. +-- +drainP :: Sources a -> Sinks a -> IO () +drainP = G.drainP +{-# 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. +-- +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 #-} + + +-- Finalizers ----------------------------------------------------------------- +-- | Attach a finalizer to some sources. +-- +-- * For a given source, the finalizer will be called the first time a +-- consumer of that source tries to pull an element when no more +-- are available. +-- +-- * The finalizer is given the index of the source that ended. +-- +-- * The finalizer will be run after any finalizers already attached +-- to the source. +-- +--- +-- TODO: make the finalizer run just the first time. +-- +finalize_i + :: (Int -> IO ()) + -> Sources a -> IO (Sources a) +finalize_i f s + = G.finalize_i f s +{-# INLINE finalize_i #-} + + +-- | Attach a finalizer to some sinks. +-- +-- * For a given sink, the finalizer will be called the first time +-- that sink is ejected. +-- +-- * The finalizer is given the index of the sink that was ejected. +-- +-- * The finalizer will be run after any finalizers already attached +-- to the sink. +-- +--- +-- TODO: make the finalizer run just the first time. +-- +finalize_o + :: (Int -> IO ()) + -> Sinks a -> IO (Sinks a) +finalize_o f k + = G.finalize_o f k +{-# INLINE finalize_o #-} + + +-- Mapping -------------------------------------------------------------------- +-- | Apply a function to all elements pulled from some sources. +map_i :: (Flow a, Build b bt) + => (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) + => (a -> b) -> Sinks b -> IO (Sinks a) +map_o f s = C.smap_o (\_ x -> f x) s +{-# INLINE map_o #-} + + +-- Connecting ----------------------------------------------------------------- +-- | Send the same data to two consumers. +-- +-- Given two argument sinks, yield a result sink. +-- Pushing to the result sink causes the same element to be pushed to both +-- argument sinks. +dup_oo :: Sinks a -> Sinks a -> IO (Sinks a) +dup_oo = G.dup_oo +{-# INLINE dup_oo #-} + + +-- | Send the same data to two consumers. +-- +-- Given an argument source and argument sink, yield a result source. +-- Pulling an element from the result source pulls from the argument source, +-- and pushes that element to the sink, as well as returning it via the +-- result source. +-- +dup_io :: Sources a -> Sinks a -> IO (Sources a) +dup_io = G.dup_io +{-# INLINE dup_io #-} + + +-- | Send the same data to two consumers. +-- +-- Like `dup_io` but with the arguments flipped. +-- +dup_oi :: Sinks a -> Sources a -> IO (Sources a) +dup_oi = G.dup_oi +{-# INLINE dup_oi #-} + + +-- | Connect an argument source to two result sources. +-- +-- Pulling from either result source pulls from the argument source. +-- Each result source only gets the elements pulled at the time, +-- so if one side pulls all the elements the other side won't get any. +-- +connect_i :: Sources a -> IO (Sources a, Sources a) +connect_i = G.connect_i +{-# INLINE connect_i #-} + + +-- Watching ------------------------------------------------------------------- +-- | Hook a worker function to some sources, which will be passed every +-- chunk that is pulled from each source. +-- +-- * The worker is also passed the source index of the chunk that was pulled. +-- +watch_i :: (Int -> A.Array A a -> IO ()) + -> Sources a -> IO (Sources a) +watch_i f s = G.watch_i f s +{-# INLINE watch_i #-} + + +-- | Hook a worker function to some sinks, which will be passed every +-- chunk that is pushed to each sink. +-- +-- * The worker is also passed the source index of the chunk that was pushed. +-- +watch_o :: (Int -> A.Array A a -> IO ()) + -> Sinks a -> IO (Sinks a) +watch_o f k = G.watch_o f k +{-# INLINE watch_o #-} + + +-- | Create a bundle of sinks of the given arity that pass incoming chunks +-- to a worker function. +-- +-- * This is like `watch_o`, except that the incoming chunks are discarded +-- after they are passed to the worker function +-- +trigger_o :: Int -> (Int -> A.Array A a -> IO ()) + -> IO (Sinks a) +trigger_o arity f + = G.trigger_o arity f +{-# INLINE trigger_o #-} + + +-- Ignorance ------------------------------------------------------------------ +-- | Create a bundle of sinks of the given arity that drop 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 :: Int -> IO (Sinks a) +discard_o = G.discard_o +{-# INLINE discard_o #-} + + +-- | Create a bundle of sinks of the given arity that drop all data on the +-- floor. +-- +-- * As opposed to `discard_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 #-} + + +-- 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 +-- remaining elements. +-- +-- * We pull /whole chunks/ from the source stream until we have +-- at least the desired number of elements. The leftover elements +-- in the final chunk are visible in the result `Sources`. +-- +head_i :: Flow a + => Int -> Int -> Sources a -> IO (Maybe ([a], Sources a)) +head_i ix len s + | ix >= G.sourcesArity s = return Nothing + | otherwise + = liftM Just $ C.head_i len s ix +{-# INLINE head_i #-} + + + +-- Grouping ------------------------------------------------------------------- +-- | Scan through some sources to find runs of matching elements, +-- and count the lengths of those runs. +-- +-- @ +-- > F.toList1 0 =<< F.groups_i =<< F.fromList 1 "waabbbblle" +-- [(\'w\',1),(\'a\',2),(\'b\',4),(\'l\',2),(\'e\',1)] +-- @ +-- +groups_i + :: (GroupsDict a u1 u2, Eq a) + => Sources a -- ^ Input elements. + -> IO (Sources (a, Int)) + -- ^ Starting element and length of groups. +groups_i s + = groupsBy_i (==) s +{-# INLINE groups_i #-} + + +-- | Like `groupsBy`, but take a function to determine whether two consecutive +-- values should be in the same group. +groupsBy_i + :: GroupsDict a u1 u2 + => (a -> a -> Bool) -- ^ Fn to check if consecutive elements + -- are in the same group. + -> Sources a -- ^ Input elements. + -> IO (Sources (a, Int)) + -- ^ Starting element and length of groups. +groupsBy_i f s + = G.map_i (A.convert A) + =<< C.groupsBy_i A A f s +{-# INLINE groupsBy_i #-} + + +-- | Dictionaries needed to perform a grouping. +type GroupsDict a u1 u2 + = C.GroupsDict Int IO A A u1 A u2 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) + => (a -> b -> a) -- ^ Combining funtion. + -> a -- ^ Starting value. + -> Sources b -- ^ Input elements to fold. + -> IO (A.Array A a) + +foldlS f z ss + = C.foldlS A f z ss +{-# INLINE foldlS #-} + + +-- | Fold all the elements of each stream in a bundle, one stream after the +-- other, returning an array of fold results. +foldlAllS + :: (Flow b) + => (a -> b -> a) -- ^ Combining funtion. + -> a -- ^ Starting value. + -> Sources b -- ^ Input elements to fold. + -> IO a + +foldlAllS f z ss + = C.foldlAllS f z ss +{-# INLINE foldlAllS #-} + + +-- | Given streams of lengths and values, perform a segmented fold where +-- fold segments of values of the corresponding lengths are folded +-- together. +-- +-- @ +-- > sSegs <- F.fromList 1 [(\'a\', 1), (\'b\', 2), (\'c\', 4), (\'d\', 0), (\'e\', 1), (\'f\', 5 :: Int)] +-- > sVals <- F.fromList 1 [10, 20, 30, 40, 50, 60, 70, 80, 90 :: Int] +-- +-- > F.toList1 0 =<< F.folds_i (+) 0 sSegs sVals +-- [(\'a\',10),(\'b\',50),(\'c\',220),(\'d\',0),(\'e\',80)] +-- @ +-- +-- If not enough input elements are available to fold a complete segment +-- then no output is produced for that segment. However, trailing zero +-- length segments still produce the initial value for the fold. +-- +-- @ +-- > sSegs <- F.fromList 1 [(\'a\', 1), (\'b\', 2), (\'c\', 0), (\'d\', 0), (\'e\', 0 :: Int)] +-- > sVals <- F.fromList 1 [10, 20, 30 :: Int] +-- +-- > F.toList1 0 =<< F.folds_i (*) 1 sSegs sVals +-- [(\'a\',10),(\'b\',600),(\'c\',1),(\'d\',1),(\'e\',1)] +-- @ +-- +folds_i :: FoldsDict n a b u1 u2 u3 u4 + => (a -> b -> b) -- ^ Worker function. + -> b -- ^ Initial state when folding each segment. + -> Sources (n, Int) -- ^ Segment lengths. + -> Sources a -- ^ Input elements to fold. + -> IO (Sources (n, b)) -- ^ Result elements. + +folds_i f z sLen sVal + = G.map_i (A.convert A) + =<< C.folds_i A A f z sLen sVal +{-# 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 + + +-- | Combination of `groupsBy_i` and `folds_i`. We determine the the segment +-- lengths while performing the folds. +-- +-- Note that a SQL-like groupby aggregations can be performed using this +-- function, provided the data is pre-sorted on the group key. For example, +-- we can take the average of some groups of values: +-- +-- @ +-- > sKeys <- F.fromList 1 "waaaabllle" +-- > sVals <- F.fromList 1 [10, 20, 30, 40, 50, 60, 70, 80, 90, 100 :: Double] +-- +-- > sResult \<- F.map_i (\\(key, (acc, n)) -\> (key, acc / n)) +-- =\<\< F.foldGroupsBy_i (==) (\\x (acc, n) -> (acc + x, n + 1)) (0, 0) sKeys sVals +-- +-- > F.toList1 0 sResult +-- [10.0,35.0,60.0,80.0,100.0] +-- @ +-- +foldGroupsBy_i + :: (FoldGroupsDict n a b u1 u2) + => (n -> n -> Bool) -- ^ Fn to check if consecutive elements + -- are in the same group. + -> (a -> b -> b) -- ^ Worker function for the fold. + -> b -- ^ Initial when folding each segment. + -> Sources n -- ^ Names that determine groups. + -> Sources a -- ^ Values to fold. + -> IO (Sources (n, b)) + +foldGroupsBy_i pGroup f z sNames sVals + = do segLens <- groupsBy_i pGroup sNames + folds_i f z segLens sVals +{-# INLINE foldGroupsBy_i #-} + + +type FoldGroupsDict n a b u1 u2 + = ( 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) diff --git a/Data/Repa/Flow/Auto/Debug.hs b/Data/Repa/Flow/Auto/Debug.hs new file mode 100644 --- /dev/null +++ b/Data/Repa/Flow/Auto/Debug.hs @@ -0,0 +1,90 @@ +{-# LANGUAGE OverlappingInstances, TypeSynonymInstances, FlexibleInstances #-} +module Data.Repa.Flow.Auto.Debug + ( -- * More + more, more' + + -- * More (tabular) + , moret, moret' + + -- * More (raw) + , morer, morer' + + -- * Nicer + , Nicer (..) + , Presentable (..)) +where +import Data.Repa.Nice.Present +import Data.Repa.Nice.Tabulate +import Data.Repa.Nice +import Data.Repa.Flow.Auto +import Control.Monad +import Data.List as L +import Data.Text as T +import Prelude as P +#include "repa-flow.h" + + +------------------------------------------------------------------------------- +-- | Given a source index and a length, pull enough chunks from the source +-- to build a list of the requested length, and discard the remaining +-- elements in the final chunk. +-- +-- * This function is intended for interactive debugging. +-- If you want to retain the rest of the final chunk then use `head_i`. +-- +more :: (Flow a, Nicer a) + => Int -- ^ Index of source in bundle. + -> Sources a -- ^ Bundle of sources. + -> IO (Maybe [Nice a]) +more i ss = more' i 20 ss +{-# INLINE more #-} + + +-- | Like `more` but also specify now many elements you want. +more' :: (Flow a, Nicer a) + => Int -> Int -> Sources a -> IO (Maybe [Nice a]) +more' ix len s + = liftM (liftM (L.map nice . fst)) $ head_i ix len s +{-# INLINE_FLOW more' #-} + + +------------------------------------------------------------------------------- +-- | Like `more`, but print results in a tabular form to the console. +moret :: (Flow a, Nicer [a], Presentable (Nice [a])) + => Int -- ^ Index of source in bundle. + -> Sources a -- ^ Bundle of sources. + -> IO () + +moret i ss = moret' i 20 ss +{-# INLINE moret #-} + + +-- | Like `more'`, but print results in tabular form to the console. +moret' :: ( Flow a, Nicer [a], Presentable (Nice [a])) + => Int -> Int -> Sources a -> IO () + +moret' ix len s + = do Just (vals, _) <- head_i ix len s + putStrLn $ T.unpack $ tabulate $ nice vals +{-# INLINE_FLOW moret' #-} + + +------------------------------------------------------------------------------- +-- | Like `more`, but show elements in their raw format. +morer :: Flow a + => Int -- ^ Index of source in bundle. + -> Sources a -- ^ Bundle of sources. + -> IO (Maybe [a]) + +morer i ss = morer' i 20 ss +{-# INLINE morer #-} + + +-- | Like `more'`, but show elements in their raw format. +morer' :: Flow a + => Int -> Int -> Sources a -> IO (Maybe [a]) +morer' ix len s + = liftM (liftM fst) $ head_i ix len s +{-# INLINE_FLOW morer' #-} + + diff --git a/Data/Repa/Flow/Auto/IO.hs b/Data/Repa/Flow/Auto/IO.hs new file mode 100644 --- /dev/null +++ b/Data/Repa/Flow/Auto/IO.hs @@ -0,0 +1,314 @@ + +-- | Read and write files. +-- +-- The functions in this module are wrappers for the ones in +-- "Data.Repa.Flow.Default.SizedIO" that use a default chunk size of +-- 64kBytes and just call `error` if the source file appears corruped. +-- +module Data.Repa.Flow.Auto.IO + ( defaultChunkSize + + -- * Buckets + , module Data.Repa.Flow.IO.Bucket + + -- * Sourcing + , sourceBytes + , sourceChars + , sourceLines + , sourceRecords + , sourceTSV + , sourceCSV + , sourcePacked + + -- * Sinking + , sinkBytes + , sinkLines + , sinkChars + , sinkPacked + + -- * Table IO + , toTable + , fromTable + ) +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.Generic as A +import System.Directory +import System.FilePath +import System.IO +import Data.Word +import qualified Data.Repa.Flow.Generic as G +import qualified Data.Repa.Flow.Generic.IO as G +import qualified Data.Repa.Flow.Auto.SizedIO as F +import Prelude as P +#include "repa-flow.h" + + +-- | The default chunk size of 64kBytes. +defaultChunkSize :: Integer +defaultChunkSize = 64 * 1024 + + +--------------------------------------------------------------------------------------------------- +-- | Read data from some files, using the given chunk length. +sourceBytes :: Array B Bucket -> IO (Sources Word8) +sourceBytes = F.sourceBytes defaultChunkSize +{-# INLINE sourceBytes #-} + + +-- | Read 8-bit ASCII characters from some files, using the given chunk length. +sourceChars :: Array B Bucket -> IO (Sources Char) +sourceChars = F.sourceChars defaultChunkSize +{-# INLINE sourceChars #-} + + +-- | Read complete records of data form a file, into chunks of the given length. +-- We read as many complete records as will fit into each chunk. +-- +-- The records are separated by a special terminating character, which the +-- given predicate detects. After reading a chunk of data we seek the file to +-- just after the last complete record that was read, so we can continue to +-- read more complete records next time. +-- +-- If we cannot fit at least one complete record in the chunk then perform +-- the given failure action. Limiting the chunk length guards against the +-- case where a large input file is malformed, as we won't try to read the +-- whole file into memory. +-- +-- +-- * Data is read into foreign memory without copying it through the GHC heap. +-- * The provided file handle must support seeking, else you'll get an +-- exception. +-- * Each file is closed the first time the consumer tries to pull a +-- record from the associated stream when no more are available. +-- +sourceRecords + :: (Word8 -> Bool) -- ^ Detect the end of a record. + -> Array B Bucket -- ^ Input Buckets. + -> IO (Sources (Array A Word8)) +sourceRecords pSep + = F.sourceRecords defaultChunkSize pSep + $ error $ "Record exceeds chunk size of " + ++ show defaultChunkSize ++ "bytes." +{-# INLINE sourceRecords #-} + + +-- | Read complete lines of data from a text file, using the given chunk length. +-- We read as many complete lines as will fit into each chunk. +-- +-- * The trailing new-line characters are discarded. +-- * Data is read into foreign memory without copying it through the GHC heap. +-- * The provided file handle must support seeking, else you'll get an +-- exception. +-- * Each file is closed the first time the consumer tries to pull a line +-- from the associated stream when no more are available. +-- +sourceLines + :: Array B Bucket -> IO (Sources (Array A Char)) +sourceLines + = F.sourceLines defaultChunkSize + $ error $ "Line exceeds chunk size of " + ++ show defaultChunkSize ++ "bytes." +{-# INLINE sourceLines #-} + + +-- | Read a file containing Comma-Separated-Values. +sourceCSV :: Array B Bucket + -> IO (Sources (Array A (Array A Char))) +sourceCSV + = F.sourceCSV defaultChunkSize + $ error $ "Line exceeds chunk size of " + ++ show defaultChunkSize ++ "bytes." +{-# INLINE sourceCSV #-} + + +-- | Read a file containing Tab-Separated-Values. +sourceTSV :: Array B Bucket + -> IO (Sources (Array A (Array A Char))) +sourceTSV + = F.sourceTSV defaultChunkSize + $ error $ "Line exceeds chunk size of " + ++ show defaultChunkSize ++ "bytes." +{-# 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. + -> 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 #-} + + +--------------------------------------------------------------------------------------------------- +-- | Write 8-bit bytes to some files. +sinkBytes :: Array B Bucket -> IO (Sinks Word8) +sinkBytes bs + = G.map_o (A.convert F) + =<< G.sinkBytes bs +{-# INLINE sinkBytes #-} + + +-- | Write 8-bit ASCII characters to some files. +sinkChars :: Array B Bucket -> IO (Sinks Char) +sinkChars = G.sinkChars +{-# INLINE sinkChars #-} + + +-- | Write vectors of text lines to the given files handles. +sinkLines :: Array B Bucket -> IO (Sinks (Array A Char)) +sinkLines = G.sinkLines A A +{-# INLINE sinkLines #-} + + +-- | Create sinks that convert values to a packed binary format and writes +-- them to some buckets. +-- +-- @ +-- > import Data.Repa.Flow as F +-- > import Data.Repa.Convert.Format as F +-- > :{ +-- do let format = FixString ASCII 10 :*: Float64be :*: Int16be +-- let vals = listFormat format +-- [ \"red\" :*: 5.3 :*: 100 +-- , \"green\" :*: 2.8 :*: 93 +-- , \"blue\" :*: 0.99 :*: 42 ] +-- +-- ss <- F.fromList 1 vals +-- out <- toFiles' [\"colors.bin\"] $ sinkPacked format (error \"convert failed\") +-- drainS ss out +-- :} +-- @ +-- +sinkPacked + :: (Packable format, Bulk A (Value 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 + where + push_sinkPacked i !chunk + = case A.packForeign format chunk of + Nothing -> aFail + Just buf -> bPutArray (bs `index` i) (A.convert F buf) + {-# INLINE push_sinkPacked #-} + + eject_sinkPacked i + = bClose (bs `index` i) + {-# INLINE eject_sinkPacked #-} + +{-# INLINE_FLOW sinkPacked #-} + + +--------------------------------------------------------------------------------------------------- +-- | Create sinks that write values from some binary Repa table. +toTable :: (Packable format, Bulk A (Value format)) + => FilePath -- ^ Directory holding table. + -> Int -- ^ Number of buckets to use. + -> format -- ^ Format of data. + -> IO () -- ^ Action when a value cannot be serialised. + -> IO (Maybe (Sinks (Value format))) + +toTable path nBuckets format aFail + | nBuckets <= 0 + = return $ Nothing + + | otherwise + = do + createDirectory path + + -- Create all the bucket files. + let makeName i = path ((replicate (6 - (P.length $ show i)) '0') ++ show i) + let names = [makeName i | i <- [0 .. nBuckets - 1]] + let newBucket file + = do h <- openBinaryFile file WriteMode + return $ Bucket + { bucketFilePath = Just file + , bucketStartPos = 0 + , bucketLength = Nothing + , bucketHandle = h } + + bs <- mapM newBucket names + + -- Create a sink bundle for the buckets. + kk <- sinkPacked format aFail (A.fromList B bs) + return $ Just kk +{-# INLINE_FLOW toTable #-} + + +-- | Create sources that read values from some binary Repa table. +fromTable + :: (Packable format, Target A (Value format)) + => FilePath -- ^ Directory holding table. + -> format -- ^ Format of data. + -> IO () -- ^ Action when a value cannot be deserialised. + -> IO (Maybe (Sources (Value format))) + +fromTable path format aFail + = do + -- All the files in the table directory. + fs <- getDirectoryContents path + + -- Filter out special file names and make them relative to the dir stem. + let fsRel + = P.map (path ) + $ P.filter (\f -> f /= "." && f /= "..") fs + + let newBucket file + = do h <- openBinaryFile file ReadMode + return $ Bucket + { bucketFilePath = Just file + , bucketStartPos = 0 + , bucketLength = Nothing + , bucketHandle = h } + + bs <- mapM newBucket fsRel + + -- Create a source bundle for the buckets. + ss <- sourcePacked format aFail (A.fromList B bs) + return $ Just ss +{-# INLINE_FLOW fromTable #-} + diff --git a/Data/Repa/Flow/Auto/SizedIO.hs b/Data/Repa/Flow/Auto/SizedIO.hs new file mode 100644 --- /dev/null +++ b/Data/Repa/Flow/Auto/SizedIO.hs @@ -0,0 +1,114 @@ + +-- | Read and write files. +module Data.Repa.Flow.Auto.SizedIO + ( -- * Buckets + module Data.Repa.Flow.IO.Bucket + + -- * Sourcing + , sourceBytes + , sourceChars + , sourceLines + , sourceRecords + , sourceTSV + , sourceCSV + ) +where +import Data.Repa.Flow.Auto +import Data.Repa.Flow.IO.Bucket +import Data.Repa.Array.Generic as A +import Data.Repa.Array.Material as A +import Data.Repa.Array.Meta.Delayed as A +import qualified Data.Repa.Flow.Generic as G +import qualified Data.Repa.Flow.Generic.IO as G +import Data.Word +import Data.Char +#include "repa-flow.h" + + +-- Sourcing --------------------------------------------------------------------------------------- +-- | Like `F.sourceBytes`, but with the default chunk size. +sourceBytes + :: Integer -> Array B Bucket -> IO (Sources Word8) +sourceBytes i bs + = G.map_i (A.convert A) + =<< G.sourceBytes i bs +{-# INLINE sourceBytes #-} + + +-- | Like `F.sourceChars`, but with the default chunk size. +sourceChars + :: Integer -> Array B Bucket -> IO (Sources Char) +sourceChars i bs + = G.map_i (A.convert A) + =<< G.sourceChars i bs +{-# INLINE sourceChars #-} + + +-- | Like `F.sourceLines`, but with the default chunk size and error action. +sourceLines + :: Integer -- ^ Size of chunk to read in bytes. + -> IO () -- ^ Action to perform if we can't get a + -- whole record. + -> Array B Bucket -- ^ Buckets. + -> IO (Sources (Array A Char)) + +sourceLines nChunk fails bs + = G.map_i ((A.convert A). chopChunk) + =<< G.sourceRecords nChunk isNewLine fails bs + where + isNewLine :: Word8 -> Bool + isNewLine x = x == nl + {-# INLINE isNewLine #-} + + chopChunk chunk + = A.mapElems (A.computeS A . A.map (chr . fromIntegral)) + $ A.trimEnds (== nl) chunk + {-# INLINE chopChunk #-} + + nl :: Word8 + !nl = fromIntegral $ ord '\n' +{-# NOINLINE sourceLines #-} + + +-- | Like `F.sourceRecords`, but with the default chunk size and error action. +sourceRecords + :: Integer -- ^ Size of chunk to read in bytes. + -> (Word8 -> Bool) -- ^ Detect the end of a record. + -> IO () -- ^ Action to perform if we can't get a + -- whole record. + -> Array B Bucket -- ^ File handles. + -> IO (Sources (Array A Word8)) + +sourceRecords i pSep aFail bs + = G.map_i (A.convert A) + =<< G.sourceRecords i pSep aFail bs +{-# INLINE sourceRecords #-} + + +-- | Read a file containing Comma-Separated-Values. +sourceCSV + :: Integer -- ^ Chunk length. + -> IO () -- ^ Action to perform if we find line longer + -- than the chunk length. + -> Array B Bucket -- ^ Buckets + -> IO (Sources (Array A (Array A Char))) + +sourceCSV i aFail bs + = G.map_i (A.convert A) + =<< G.sourceCSV i aFail bs +{-# INLINE sourceCSV #-} + + +-- | Read a file containing Tab-Separated-Values. +sourceTSV + :: Integer -- ^ Chunk length. + -> IO () -- ^ Action to perform if we find line longer + -- than the chunk length. + -> Array B Bucket -- ^ Buckets + -> IO (Sources (Array A (Array A Char))) + +sourceTSV i aFail bs + = G.map_i (A.convert A) + =<< G.sourceTSV i aFail bs +{-# INLINE sourceTSV #-} + 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 @@ -9,13 +9,14 @@ , head_i , finalize_i, finalize_o) where +import Data.Repa.Flow.States as F +import Data.Repa.Array.Generic as A +import Data.Repa.Array.Generic.Index as A +import Data.Repa.Array.Meta.Window as A +import Control.Monad +import qualified Data.Repa.Flow.Generic as G import qualified Data.Sequence as Q import qualified Data.Foldable as Q -import Data.Repa.Flow.States -import Data.Repa.Array as A -import Data.Repa.Eval.Array as A -import qualified Data.Repa.Flow.Generic as G -import Control.Monad import Prelude as P #include "repa-flow.h" 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 @@ -5,8 +5,10 @@ where import Data.Repa.Flow.Chunked.Base import Data.Repa.Flow.States as S -import qualified Data.Repa.Array as A 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 @@ -7,9 +7,12 @@ import Data.Repa.Flow.States import Data.Repa.Fusion.Unpack import Data.Repa.Option -import Data.Repa.Array as A hiding (FoldsDict) -import Data.Repa.Eval.Array as A -import qualified Data.Repa.Flow.Generic as G +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 +import qualified Data.Repa.Flow.Generic as G #include "repa-flow.h" @@ -24,8 +27,8 @@ , TargetI lElt a , TargetI lGrp n , TargetI lRes b - , Unpack (IOBuffer lGrp n) tGrp - , Unpack (IOBuffer lRes b) tRes) + , Unpack (Buffer lGrp n) tGrp + , Unpack (Buffer lRes b) tRes) -- Folds ---------------------------------------------------------------------- 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 @@ -6,9 +6,11 @@ import Data.Repa.Flow.Chunked.Base import Data.Repa.Flow.States import Data.Repa.Fusion.Unpack -import Data.Repa.Array as A hiding (GroupsDict) -import Data.Repa.Eval.Array as A -import qualified Data.Repa.Flow.Generic as G +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" @@ -17,8 +19,8 @@ = ( Flow i m lVal a, A.Index lVal ~ Int , TargetI lGrp a , TargetI lLen Int - , Unpack (IOBuffer lGrp a) tGrp - , Unpack (IOBuffer lLen Int) tLen) + , Unpack (Buffer lGrp a) tGrp + , Unpack (Buffer lLen Int) tLen) -- Grouping ------------------------------------------------------------------- diff --git a/Data/Repa/Flow/Chunked/IO.hs b/Data/Repa/Flow/Chunked/IO.hs --- a/Data/Repa/Flow/Chunked/IO.hs +++ b/Data/Repa/Flow/Chunked/IO.hs @@ -17,7 +17,7 @@ where import Data.Repa.Flow.IO.Bucket import Data.Repa.Flow.Chunked.Base -import Data.Repa.Array as A +import Data.Repa.Array.Generic as A import Data.Repa.Array.Material as A import qualified Data.Repa.Flow.Generic.IO as G import Data.Word diff --git a/Data/Repa/Flow/Chunked/Map.hs b/Data/Repa/Flow/Chunked/Map.hs --- a/Data/Repa/Flow/Chunked/Map.hs +++ b/Data/Repa/Flow/Chunked/Map.hs @@ -5,8 +5,10 @@ where import Data.Repa.Flow.Chunked.Base import Data.Repa.Flow.States -import Data.Repa.Array as A -import Data.Repa.Eval.Array as A +import Data.Repa.Array.Generic as A +import Data.Repa.Array.Generic.Index as A +import Data.Repa.Array.Meta.Window as A +import Data.Repa.Array.Meta.Delayed as A import qualified Data.Repa.Flow.Generic as G #include "repa-flow.h" diff --git a/Data/Repa/Flow/Chunked/Operator.hs b/Data/Repa/Flow/Chunked/Operator.hs --- a/Data/Repa/Flow/Chunked/Operator.hs +++ b/Data/Repa/Flow/Chunked/Operator.hs @@ -15,7 +15,7 @@ , ignore_o) where import Data.Repa.Flow.Chunked.Base -import Data.Repa.Array as A +import Data.Repa.Array.Generic as A import qualified Data.Repa.Flow.Generic as G #include "repa-flow.h" diff --git a/Data/Repa/Flow/Default.hs b/Data/Repa/Flow/Default.hs deleted file mode 100644 --- a/Data/Repa/Flow/Default.hs +++ /dev/null @@ -1,516 +0,0 @@ - --- | This module defines the default specialisation of flows that --- appears in "Data.Repa.Flow". Each stream in the bundle is indexed --- by a single integer, and stream state is stored using the IO monad. --- -module Data.Repa.Flow.Default - ( -- * Flow types - Sources - , Sinks - , Flow - , sourcesArity - , sinksArity - - -- * States and Arrays - , module Data.Repa.Flow.States - , module Data.Repa.Eval.Array - , module Data.Repa.Array - , module Data.Repa.Array.Material - - -- * Evaluation - , drainS - , drainP - - -- * Conversion - , fromList, fromLists - , toList1, toLists1 - - -- * Finalizers - , finalize_i, finalize_o - - -- * Flow Operators - -- ** 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 - - -- ** Connecting - , dup_oo - , dup_io - , dup_oi - , connect_i - - -- ** Watching - , watch_i, watch_o - , trigger_o - - -- ** Ignorance - , discard_o - , ignore_o - - -- ** Splitting - , head_i - - -- ** Grouping - , groups_i - , groupsBy_i - , GroupsDict - - -- ** Folding - , foldlS, foldlAllS - , folds_i, FoldsDict - , foldGroupsBy_i, FoldGroupsDict) -where -import Data.Repa.Flow.States -import Data.Repa.Eval.Array -import Data.Repa.Eval.Array as A - -import Data.Repa.Array - hiding (FoldsDict, GroupsDict, Index, fromList) - -import Data.Repa.Array as A - hiding (FoldsDict, GroupsDict, fromList) - -import Data.Repa.Array.Material hiding (fromLists) -import Data.Repa.Fusion.Unpack as A -import qualified Data.Repa.Flow.Chunked as C hiding (next) -import qualified Data.Repa.Flow.Generic as G hiding (next) -import Control.Monad -#include "repa-flow.h" - - --- | A bundle of stream sources, where the elements of the stream --- are chunked into arrays. --- --- The chunks have some `Layout` @l@ and contain elements of type @a@. --- See "Data.Repa.Array" for the available layouts. -type Sources l a = C.Sources Int IO l a - - --- | A bundle of stream sinks, where the elements of the stream --- are chunked into arrays. --- -type Sinks l a = C.Sinks Int IO l a - - --- | Yield the number of streams in the bundle. -sourcesArity :: Sources l a -> Int -sourcesArity = G.sourcesArity - - --- | Yield the number of streams in the bundle. -sinksArity :: Sinks l a -> Int -sinksArity = G.sinksArity - - --- | Shorthand for common type classes. -type Flow l a = C.Flow Int IO l 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. --- --- * If the `Sources` and `Sinks` have different numbers of streams then --- we only evaluate the common subset. --- -drainS :: Sources l a -> Sinks l a -> IO () -drainS = G.drainS -{-# INLINE drainS #-} - - --- | Pull all available values from the sources and push them to the sinks, --- in parallel. We fork a thread for each of the streams and evaluate --- them all in parallel. --- --- * If the `Sources` and `Sinks` have different numbers of streams then --- we only evaluate the common subset. --- -drainP :: Sources l a -> Sinks l a -> IO () -drainP = G.drainP -{-# 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. --- -fromList :: A.TargetI l a - => Name l -> Int -> [a] -> IO (Sources l a) -fromList l xs = C.fromList l 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 :: A.TargetI l a - => Name l -> Int -> [[a]] -> IO (Sources l a) -fromLists nDst xss = C.fromLists nDst xss -{-# INLINE fromLists #-} - - --- | Drain a single source from a bundle into a list of elements. -toList1 :: A.BulkI l a - => Int -> Sources l 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 :: A.BulkI l a - => Int -> Sources l a -> IO [[a]] -toLists1 ix s - | ix >= G.sourcesArity s = return [] - | otherwise = C.toLists1 ix s -{-# INLINE toLists1 #-} - - --- Finalizers ----------------------------------------------------------------- --- | Attach a finalizer to some sources. --- --- * For a given source, the finalizer will be called the first time a --- consumer of that source tries to pull an element when no more --- are available. --- --- * The finalizer is given the index of the source that ended. --- --- * The finalizer will be run after any finalizers already attached --- to the source. --- ---- --- TODO: make the finalizer run just the first time. --- -finalize_i - :: (Int -> IO ()) - -> Sources l a -> IO (Sources l a) -finalize_i f s - = G.finalize_i f s -{-# INLINE finalize_i #-} - - --- | Attach a finalizer to some sinks. --- --- * For a given sink, the finalizer will be called the first time --- that sink is ejected. --- --- * The finalizer is given the index of the sink that was ejected. --- --- * The finalizer will be run after any finalizers already attached --- to the sink. --- ---- --- TODO: make the finalizer run just the first time. --- -finalize_o - :: (Int -> IO ()) - -> Sinks l a -> IO (Sinks l a) -finalize_o f k - = G.finalize_o f k -{-# INLINE finalize_o #-} - - --- Mapping -------------------------------------------------------------------- --- | Apply a function to all elements pulled from some sources. -map_i :: (Flow l1 a, A.TargetI l2 b) - => Name l2 -> (a -> b) -> Sources l1 a -> IO (Sources l2 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 l1 a, A.TargetI l2 b) - => Name l1 -> (a -> b) -> Sinks l2 b -> IO (Sinks l1 a) -map_o _ f s = C.smap_o (\_ x -> f x) s -{-# INLINE map_o #-} - - --- Connecting ----------------------------------------------------------------- --- | Send the same data to two consumers. --- --- Given two argument sinks, yield a result sink. --- Pushing to the result sink causes the same element to be pushed to both --- argument sinks. -dup_oo :: Sinks l a -> Sinks l a -> IO (Sinks l a) -dup_oo = G.dup_oo -{-# INLINE dup_oo #-} - - --- | Send the same data to two consumers. --- --- Given an argument source and argument sink, yield a result source. --- Pulling an element from the result source pulls from the argument source, --- and pushes that element to the sink, as well as returning it via the --- result source. --- -dup_io :: Sources l a -> Sinks l a -> IO (Sources l a) -dup_io = G.dup_io -{-# INLINE dup_io #-} - - --- | Send the same data to two consumers. --- --- Like `dup_io` but with the arguments flipped. --- -dup_oi :: Sinks l a -> Sources l a -> IO (Sources l a) -dup_oi = G.dup_oi -{-# INLINE dup_oi #-} - - --- | Connect an argument source to two result sources. --- --- Pulling from either result source pulls from the argument source. --- Each result source only gets the elements pulled at the time, --- so if one side pulls all the elements the other side won't get any. --- -connect_i :: Sources l a -> IO (Sources l a, Sources l a) -connect_i = G.connect_i -{-# INLINE connect_i #-} - - --- Watching ------------------------------------------------------------------- --- | Hook a worker function to some sources, which will be passed every --- chunk that is pulled from each source. --- --- * The worker is also passed the source index of the chunk that was pulled. --- -watch_i :: (Int -> Array l a -> IO ()) - -> Sources l a -> IO (Sources l a) -watch_i f s = G.watch_i f s -{-# INLINE watch_i #-} - - --- | Hook a worker function to some sinks, which will be passed every --- chunk that is pushed to each sink. --- --- * The worker is also passed the source index of the chunk that was pushed. --- -watch_o :: (Int -> Array l a -> IO ()) - -> Sinks l a -> IO (Sinks l a) -watch_o f k = G.watch_o f k -{-# INLINE watch_o #-} - - --- | Create a bundle of sinks of the given arity that pass incoming chunks --- to a worker function. --- --- * This is like `watch_o`, except that the incoming chunks are discarded --- after they are passed to the worker function --- -trigger_o :: Int -> (Int -> Array l a -> IO ()) - -> IO (Sinks l a) -trigger_o arity f - = G.trigger_o arity f -{-# INLINE trigger_o #-} - - --- Ignorance ------------------------------------------------------------------ --- | Create a bundle of sinks of the given arity that drop 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 :: Int -> IO (Sinks l a) -discard_o = G.discard_o -{-# INLINE discard_o #-} - - --- | Create a bundle of sinks of the given arity that drop all data on the --- floor. --- --- * As opposed to `discard_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 l a) -ignore_o = G.ignore_o -{-# INLINE ignore_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 --- remaining elements. --- --- * We pull /whole chunks/ from the source stream until we have --- at least the desired number of elements. The leftover elements --- in the final chunk are visible in the result `Sources`. --- -head_i :: (A.Windowable l a, A.Index l ~ Int) - => Int -> Int -> Sources l a -> IO (Maybe ([a], Sources l a)) -head_i ix len s - | ix >= G.sourcesArity s = return Nothing - | otherwise - = liftM Just $ C.head_i len s ix -{-# INLINE head_i #-} - - --- Grouping ------------------------------------------------------------------- --- | Scan through some sources to find runs of matching elements, --- and count the lengths of those runs. --- --- @ --- > import Data.Repa.Flow --- > toList1 0 =<< groups_i U U =<< fromList U 1 "waabbbblle" --- Just [(\'w\',1),(\'a\',2),(\'b\',4),(\'l\',2),(\'e\',1)] --- @ --- -groups_i - :: (GroupsDict lVal lGrp tGrp lLen tLen a, Eq a) - => Name lGrp -- ^ Layout of result groups. - -> Name lLen -- ^ Layout of result lengths. - -> Sources lVal a -- ^ Input elements. - -> IO (Sources (T2 lGrp lLen) (a, Int)) - -- ^ Starting element and length of groups. -groups_i nGrp nLen s - = groupsBy_i nGrp nLen (==) s -{-# INLINE groups_i #-} - - --- | Like `groupsBy`, but take a function to determine whether two consecutive --- values should be in the same group. -groupsBy_i - :: GroupsDict lVal lGrp tGrp lLen tLen a - => Name lGrp -- ^ Layout of result groups. - -> Name lLen -- ^ Layout of result lengths. - -> (a -> a -> Bool) -- ^ Fn to check if consecutive elements - -- are in the same group. - -> Sources lVal a -- ^ Input elements. - -> IO (Sources (T2 lGrp lLen) (a, Int)) - -- ^ Starting element and length of groups. -groupsBy_i nGrp nLen f s - = C.groupsBy_i nGrp nLen f s -{-# INLINE groupsBy_i #-} - - --- | Dictionaries needed to perform a grouping. -type GroupsDict lVal lGrp tGrp lLen tLen a - = C.GroupsDict Int IO lVal lGrp tGrp lLen tLen a - - --- Folding -------------------------------------------------------------------- --- | Fold all the elements of each stream in a bundle, one stream after the --- other, returning an array of fold results. -foldlS - :: ( A.Target lDst a, A.Index lDst ~ Int - , A.BulkI lSrc b) - => A.Name lDst -- ^ Layout for result. - -> (a -> b -> a) -- ^ Combining funtion. - -> a -- ^ Starting value. - -> Sources lSrc b -- ^ Input elements to fold. - -> IO (A.Array lDst a) - -foldlS n f z ss - = C.foldlS n f z ss -{-# INLINE foldlS #-} - - --- | Fold all the elements of each stream in a bundle, one stream after the --- other, returning an array of fold results. -foldlAllS - :: A.BulkI lSrc b - => (a -> b -> a) -- ^ Combining funtion. - -> a -- ^ Starting value. - -> Sources lSrc b -- ^ Input elements to fold. - -> IO a - -foldlAllS f z ss - = C.foldlAllS f z ss -{-# INLINE foldlAllS #-} - - --- | Given streams of lengths and values, perform a segmented fold where --- fold segments of values of the corresponding lengths are folded --- together. --- --- @ --- > import Data.Repa.Flow --- > sSegs <- fromList U 1 [(\'a\', 1), (\'b\', 2), (\'c\', 4), (\'d\', 0), (\'e\', 1), (\'f\', 5 :: Int)] --- > sVals <- fromList U 1 [10, 20, 30, 40, 50, 60, 70, 80, 90 :: Int] --- > toList1 0 =<< folds_i U U (+) 0 sSegs sVals --- Just [(\'a\',10),(\'b\',50),(\'c\',220),(\'d\',0),(\'e\',80)] --- @ --- --- If not enough input elements are available to fold a complete segment --- then no output is produced for that segment. However, trailing zero --- length segments still produce the initial value for the fold. --- --- @ --- > import Data.Repa.Flow --- > sSegs <- fromList U 1 [(\'a\', 1), (\'b\', 2), (\'c\', 0), (\'d\', 0), (\'e\', 0 :: Int)] --- > sVals <- fromList U 1 [10, 20, 30 :: Int] --- > toList1 0 =<< folds_i U U (*) 1 sSegs sVals --- Just [(\'a\',10),(\'b\',600),(\'c\',1),(\'d\',1),(\'e\',1)] --- @ --- -folds_i :: (FoldsDict lSeg tSeg lElt tElt lGrp tGrp lRes tRes n a b) - => Name lGrp -- ^ Layout for group names. - -> Name lRes -- ^ Layout for fold results. - -> (a -> b -> b) -- ^ Worker function. - -> b -- ^ Initial state when folding each segment. - -> Sources lSeg (n, Int) -- ^ Segment lengths. - -> Sources lElt a -- ^ Input elements to fold. - -> IO (Sources (T2 lGrp lRes) (n, b)) -- ^ Result elements. - -folds_i nGrp nRes f z sLen sVal - = C.folds_i nGrp nRes f z sLen sVal -{-# INLINE folds_i #-} - --- | Dictionaries needed to perform a segmented fold. -type FoldsDict lSeg tSeg lElt tElt lGrp tGrp lRes tRes n a b - = C.FoldsDict Int IO lSeg tSeg lElt tElt lGrp tGrp lRes tRes n a b - - --- | Combination of `groupsBy_i` and `folds_i`. We determine the the segment --- lengths while performing the folds. --- --- Note that a SQL-like groupby aggregations can be performed using this --- function, provided the data is pre-sorted on the group key. For example, --- we can take the average of some groups of values: --- --- @ --- > import Data.Repa.Flow --- > sKeys <- fromList U 1 "waaaabllle" --- > sVals <- fromList U 1 [10, 20, 30, 40, 50, 60, 70, 80, 90, 100 :: Double] --- --- > sResult \<- map_i U (\\(key, (acc, n)) -\> (key, acc / n)) --- =\<\< foldGroupsBy_i U U (==) (\\x (acc, n) -> (acc + x, n + 1)) (0, 0) sKeys sVals --- --- > toList1 0 sResult --- Just [10.0,35.0,60.0,80.0,100.0] --- @ --- -foldGroupsBy_i - :: ( FoldGroupsDict lSeg tSeg lVal tVal lGrp tGrp lRes tRes n a b) - => Name lGrp -- ^ Layout for group names. - -> Name lRes -- ^ Layout for fold results. - -> (n -> n -> Bool) -- ^ Fn to check if consecutive elements - -- are in the same group. - -> (a -> b -> b) -- ^ Worker function for the fold. - -> b -- ^ Initial when folding each segment. - -> Sources lSeg n -- ^ Names that determine groups. - -> Sources lVal a -- ^ Values to fold. - -> IO (Sources (T2 lGrp lRes) (n, b)) - -foldGroupsBy_i nGrp nRes pGroup f z sNames sVals - = do segLens <- groupsBy_i nGrp U pGroup sNames - folds_i nGrp nRes f z segLens sVals -{-# INLINE foldGroupsBy_i #-} - - -type FoldGroupsDict lSeg tSeg lElt tElt lGrp tGrp lRes tRes n a b - = ( A.BulkI lSeg n - , A.Material lElt a, A.Index lElt ~ Int - , A.Material lGrp n, A.Index lGrp ~ Int - , A.Material lRes b, A.Index lRes ~ Int - , Unpack (IOBuffer lGrp n) tGrp - , Unpack (IOBuffer lRes b) tRes) diff --git a/Data/Repa/Flow/Default/Debug.hs b/Data/Repa/Flow/Default/Debug.hs deleted file mode 100644 --- a/Data/Repa/Flow/Default/Debug.hs +++ /dev/null @@ -1,94 +0,0 @@ -{-# LANGUAGE OverlappingInstances, TypeSynonymInstances, FlexibleInstances #-} -module Data.Repa.Flow.Default.Debug - (-- * More - more, more' - - -- * More (tabular) - , moret, moret' - - -- * More (raw) - , morer, morer' - - -- * Nicer - , Nicer (..) - , Presentable (..)) -where -import Data.Repa.Nice.Present -import Data.Repa.Nice.Tabulate -import Data.Repa.Nice -import Data.Repa.Flow.Default hiding (next) -import qualified Data.Repa.Array as A -import Control.Monad -import Data.List as L -import Data.Text as T -import Prelude as P -#include "repa-flow.h" - - -------------------------------------------------------------------------------- --- | Given a source index and a length, pull enough chunks from the source --- to build a list of the requested length, and discard the remaining --- elements in the final chunk. --- --- * This function is intended for interactive debugging. --- If you want to retain the rest of the final chunk then use `head_i`. --- -more :: (Windowable l a, A.Index l ~ Int, Nicer a) - => Int -- ^ Index of source in bundle. - -> Sources l a -- ^ Bundle of sources. - -> IO (Maybe [Nice a]) -more i ss = more' i 20 ss -{-# INLINE more #-} - - --- | Like `more` but also specify now many elements you want. -more' :: (Windowable l a, A.Index l ~ Int, Nicer a) - => Int -> Int -> Sources l a -> IO (Maybe [Nice a]) -more' ix len s - = liftM (liftM (L.map nice . fst)) $ head_i ix len s -{-# INLINE_FLOW more' #-} - - -------------------------------------------------------------------------------- --- | Like `more`, but print results in a tabular form to the console. -moret :: ( A.Windowable l a, A.Index l ~ Int - , Nicer [a], Presentable (Nice [a])) - => Int -- ^ Index of source in bundle. - -> Sources l a -- ^ Bundle of sources. - -> IO () - -moret i ss = moret' i 20 ss -{-# INLINE moret #-} - - --- | Like `more'`, but print results in tabular form to the console. -moret' :: ( A.Windowable l a, A.Index l ~ Int - , Nicer [a], Presentable (Nice [a])) - => Int -> Int -> Sources l a -> IO () - -moret' ix len s - = do Just (vals, _) <- head_i ix len s - putStrLn $ T.unpack $ tabulate $ nice vals -{-# INLINE_FLOW moret' #-} - - -------------------------------------------------------------------------------- --- | Like `more`, but show elements in their raw format. -morer :: (A.Windowable l a, A.Index l ~ Int) - => Int -- ^ Index of source in bundle. - -> Sources l a -- ^ Bundle of sources. - -> IO (Maybe [a]) - -morer i ss = morer' i 20 ss -{-# INLINE morer #-} - - --- | Like `more'`, but show elements in their raw format. -morer' :: (A.Windowable l a, A.Index l ~ Int) - => Int -> Int -> Sources l a -> IO (Maybe [a]) -morer' ix len s - = liftM (liftM fst) $ head_i ix len s -{-# INLINE_FLOW morer' #-} - - - diff --git a/Data/Repa/Flow/Default/IO.hs b/Data/Repa/Flow/Default/IO.hs deleted file mode 100644 --- a/Data/Repa/Flow/Default/IO.hs +++ /dev/null @@ -1,159 +0,0 @@ - --- | Read and write files. --- --- The functions in this module are wrappers for the ones in --- "Data.Repa.Flow.Default.SizedIO" that use a default chunk size of --- 64kBytes and just call `error` if the source file appears corruped. -module Data.Repa.Flow.Default.IO - ( defaultChunkSize - - -- * Buckets - , module Data.Repa.Flow.IO.Bucket - - -- * Sourcing - , sourceCSV - , sourceTSV - , sourceRecords - , sourceLines - , sourceChars - , sourceBytes - - -- * Sinking - , sinkChars - , sinkLines - , sinkBytes) -where -import Data.Repa.Flow.Default -import Data.Repa.Flow.IO.Bucket -import Data.Repa.Fusion.Unpack -import Data.Word -import qualified Data.Repa.Flow.Default.SizedIO as F -#include "repa-flow.h" - - --- | The default chunk size of 64kBytes. -defaultChunkSize :: Integer -defaultChunkSize = 64 * 1024 - - --- | Read a file containing Comma-Separated-Values. -sourceCSV - :: BulkI l Bucket - => Array l Bucket -> IO (Sources N (Array N (Array F Char))) -sourceCSV - = F.sourceCSV defaultChunkSize - $ error $ "Line exceeds chunk size of " - ++ show defaultChunkSize ++ "bytes." -{-# INLINE sourceCSV #-} - - --- | Read a file containing Tab-Separated-Values. -sourceTSV - :: BulkI l Bucket - => Array l Bucket -> IO (Sources N (Array N (Array F Char))) -sourceTSV - = F.sourceTSV defaultChunkSize - $ error $ "Line exceeds chunk size of " - ++ show defaultChunkSize ++ "bytes." -{-# INLINE sourceTSV #-} - - --- | Read complete records of data form a file, into chunks of the given length. --- We read as many complete records as will fit into each chunk. --- --- The records are separated by a special terminating character, which the --- given predicate detects. After reading a chunk of data we seek the file to --- just after the last complete record that was read, so we can continue to --- read more complete records next time. --- --- If we cannot fit at least one complete record in the chunk then perform --- the given failure action. Limiting the chunk length guards against the --- case where a large input file is malformed, as we won't try to read the --- whole file into memory. --- --- --- * Data is read into foreign memory without copying it through the GHC heap. --- * The provided file handle must support seeking, else you'll get an --- exception. --- * Each file is closed the first time the consumer tries to pull a --- record from the associated stream when no more are available. --- -sourceRecords - :: BulkI l Bucket - => (Word8 -> Bool) -- ^ Detect the end of a record. - -> Array l Bucket -- ^ Buckets. - -> IO (Sources N (Array F Word8)) -sourceRecords pSep - = F.sourceRecords defaultChunkSize pSep - $ error $ "Record exceeds chunk size of " - ++ show defaultChunkSize ++ "bytes." -{-# INLINE sourceRecords #-} - - --- | Read complete lines of data from a text file, using the given chunk length. --- We read as many complete lines as will fit into each chunk. --- --- * The trailing new-line characters are discarded. --- * Data is read into foreign memory without copying it through the GHC heap. --- * The provided file handle must support seeking, else you'll get an --- exception. --- * Each file is closed the first time the consumer tries to pull a line --- from the associated stream when no more are available. --- -sourceLines - :: BulkI l Bucket - => Array l Bucket -> IO (Sources N (Array F Char)) -sourceLines - = F.sourceLines defaultChunkSize - $ error $ "Line exceeds chunk size of " - ++ show defaultChunkSize ++ "bytes." -{-# INLINE sourceLines #-} - - --- | Read 8-bit ASCII characters from some files, using the given chunk length. -sourceChars - :: BulkI l Bucket - => Array l Bucket -> IO (Sources F Char) -sourceChars = F.sourceChars defaultChunkSize -{-# INLINE sourceChars #-} - - --- | Read data from some files, using the given chunk length. -sourceBytes - :: BulkI l Bucket - => Array l Bucket -> IO (Sources F Word8) -sourceBytes = F.sourceBytes defaultChunkSize -{-# INLINE sourceBytes #-} - - --- | Write vectors of text lines to the given files handles. --- --- * Data is copied into a new buffer to insert newlines before being --- written out. --- -sinkLines - :: ( BulkI l Bucket - , BulkI l1 (Array l2 Char) - , BulkI l2 Char, Unpack (Array l2 Char) t2) - => Name l1 -- ^ Layout of chunks. - -> Name l2 -- ^ Layout of lines in chunks. - -> Array l Bucket -- ^ Buckets - -> IO (Sinks l1 (Array l2 Char)) -sinkLines = F.sinkLines -{-# INLINE sinkLines #-} - - --- | Write 8-bit ASCII characters to some files. -sinkChars - :: BulkI l Bucket - => Array l Bucket -> IO (Sinks F Char) -sinkChars = F.sinkChars -{-# INLINE sinkChars #-} - - --- | Write bytes to some file. -sinkBytes - :: BulkI l Bucket - => Array l Bucket -> IO (Sinks F Word8) -sinkBytes = F.sinkBytes -{-# INLINE sinkBytes #-} diff --git a/Data/Repa/Flow/Default/IO/CSV.hs b/Data/Repa/Flow/Default/IO/CSV.hs deleted file mode 100644 --- a/Data/Repa/Flow/Default/IO/CSV.hs +++ /dev/null @@ -1,52 +0,0 @@ - -module Data.Repa.Flow.Default.IO.CSV - (sourceCSV) -where -import Data.Repa.Flow.Default -import Data.Repa.Flow.IO.Bucket -import Data.Repa.Array as A -import Data.Repa.Array.Material as A -import Data.Char -import qualified Data.Repa.Flow.Generic.IO as G -import qualified Data.Repa.Flow.Generic as G -#include "repa-flow.h" - - --- | Read a file containing Tab-Separated-Values. --- --- TODO: handle escaped commas. --- TODO: check CSV file standard. --- -sourceCSV - :: BulkI l Bucket - => Integer -- Chunk length. - -> IO () -- Action to perform if we find line longer - -- than the chunk length. - -> Array l Bucket -- File paths. - -> IO (Sources N (Array N (Array F Char))) - -sourceCSV nChunk aFail bs - = do - -- Rows are separated by new lines, - -- fields are separated by commas. - let !nl = fromIntegral $ ord '\n' - let !nr = fromIntegral $ ord '\r' - let !nt = fromIntegral $ ord ',' - - -- Stream chunks of data from the input file, where the chunks end - -- cleanly at line boundaries. - sChunk <- G.sourceChunks nChunk (== nl) aFail bs - sRows8 <- G.map_i (A.diceSep nt nl . A.filter U (/= nr)) sChunk - - -- Convert element data from Word8 to Char. - -- Chars take 4 bytes each, but are standard Haskell and pretty - -- print properly. We've done the dicing on the smaller Word8 - -- version, and now map across the elements vector in the array - -- to do the conversion. - sRows <- G.map_i - (A.mapElems (A.mapElems - (A.computeS F . A.map (chr . fromIntegral)))) - sRows8 - - return sRows -{-# INLINE sourceCSV #-} diff --git a/Data/Repa/Flow/Default/IO/TSV.hs b/Data/Repa/Flow/Default/IO/TSV.hs deleted file mode 100644 --- a/Data/Repa/Flow/Default/IO/TSV.hs +++ /dev/null @@ -1,48 +0,0 @@ - -module Data.Repa.Flow.Default.IO.TSV - (sourceTSV) -where -import Data.Repa.Flow.Default -import Data.Repa.Flow.IO.Bucket -import Data.Repa.Array as A -import Data.Repa.Array.Material as A -import Data.Char -import qualified Data.Repa.Flow.Generic.IO as G -import qualified Data.Repa.Flow.Generic as G -#include "repa-flow.h" - - --- | Read a file containing Tab-Separated-Values. -sourceTSV - :: BulkI l Bucket - => Integer -- Chunk length. - -> IO () -- Action to perform if we find line longer - -- than the chunk length. - -> Array l Bucket -- File paths. - -> IO (Sources N (Array N (Array F Char))) - -sourceTSV nChunk aFail bs - = do - -- Rows are separated by new lines, - -- fields are separated by tabs. - let !nl = fromIntegral $ ord '\n' - let !nr = fromIntegral $ ord '\r' - let !nt = fromIntegral $ ord '\t' - - -- Stream chunks of data from the input file, where the chunks end - -- cleanly at line boundaries. - sChunk <- G.sourceChunks nChunk (== nl) aFail bs - sRows8 <- G.map_i (A.diceSep nt nl . A.filter U (/= nr)) sChunk - - -- Convert element data from Word8 to Char. - -- Chars take 4 bytes each, but are standard Haskell and pretty - -- print properly. We've done the dicing on the smaller Word8 - -- version, and now map across the elements vector in the array - -- to do the conversion. - sRows <- G.map_i - (A.mapElems (A.mapElems - (A.computeS F . A.map (chr . fromIntegral)))) - sRows8 - - return sRows -{-# INLINE sourceTSV #-} diff --git a/Data/Repa/Flow/Default/SizedIO.hs b/Data/Repa/Flow/Default/SizedIO.hs deleted file mode 100644 --- a/Data/Repa/Flow/Default/SizedIO.hs +++ /dev/null @@ -1,120 +0,0 @@ - --- | Read and write files. -module Data.Repa.Flow.Default.SizedIO - ( module Data.Repa.Flow.IO.Bucket - - -- * Sourcing - , sourceBytes - , sourceChars - , sourceLines - , sourceRecords - , sourceTSV - , sourceCSV - - -- * Sinking - , sinkBytes - , sinkChars - , sinkLines) -where -import Data.Repa.Flow.Default -import Data.Repa.Flow.IO.Bucket -import Data.Repa.Flow.Default.IO.TSV as F -import Data.Repa.Flow.Default.IO.CSV as F -import Data.Repa.Eval.Array as A -import Data.Repa.Array.Material as A -import Data.Repa.Fusion.Unpack as F -import Data.Repa.Array as A -import qualified Data.Repa.Flow.Generic as G -import qualified Data.Repa.Flow.Generic.IO as G -import Data.Word -import Data.Char -#include "repa-flow.h" - - --- Sourcing --------------------------------------------------------------------------------------- --- | Like `F.sourceBytes`, but with the default chunk size. -sourceBytes - :: BulkI l Bucket - => Integer -> Array l Bucket -> IO (Sources F Word8) -sourceBytes i bs = G.sourceBytes i bs -{-# INLINE sourceBytes #-} - - --- | Like `F.sourceChars`, but with the default chunk size. -sourceChars - :: BulkI l Bucket - => Integer -> Array l Bucket -> IO (Sources F Char) -sourceChars i bs = G.sourceChars i bs -{-# INLINE sourceChars #-} - - --- | Like `F.sourceLines`, but with the default chunk size and error action. -sourceLines - :: BulkI l Bucket - => Integer -- ^ Size of chunk to read in bytes. - -> IO () -- ^ Action to perform if we can't get a - -- whole record. - -> Array l Bucket -- ^ Buckets. - -> IO (Sources N (Array F Char)) -sourceLines nChunk fails bs - = G.map_i chopChunk - =<< G.sourceRecords nChunk isNewLine fails bs - where - isNewLine :: Word8 -> Bool - isNewLine x = x == nl - {-# INLINE isNewLine #-} - - chopChunk chunk - = A.mapElems (A.computeS name . A.map (chr . fromIntegral)) - $ A.trimEnds (== nl) chunk - {-# INLINE chopChunk #-} - - nl :: Word8 - !nl = fromIntegral $ ord '\n' -{-# NOINLINE sourceLines #-} - - --- | Like `F.sourceRecords`, but with the default chunk size and error action. -sourceRecords - :: BulkI l Bucket - => Integer -- ^ Size of chunk to read in bytes. - -> (Word8 -> Bool) -- ^ Detect the end of a record. - -> IO () -- ^ Action to perform if we can't get a - -- whole record. - -> Array l Bucket -- ^ File handles. - -> IO (Sources N (Array F Word8)) -sourceRecords i pSep aFail bs - = G.sourceRecords i pSep aFail bs -{-# INLINE sourceRecords #-} - - --- Sinking ---------------------------------------------------------------------------------------- --- | An alias for `F.sinkBytes`. -sinkBytes - :: BulkI l Bucket - => Array l Bucket -> IO (Sinks F Word8) -sinkBytes bs = G.sinkBytes bs -{-# INLINE sinkBytes #-} - - --- | An alias for `F.sinkChars`. -sinkChars - :: BulkI l Bucket - => Array l Bucket -> IO (Sinks F Char) -sinkChars bs = G.sinkChars bs -{-# INLINE sinkChars #-} - - --- | An alias for `F.sinkLines`. -sinkLines - :: ( BulkI l Bucket - , BulkI l1 (Array l2 Char) - , BulkI l2 Char, Unpack (Array l2 Char) t2) - => Name l1 -- ^ Layout for chunks of lines. - -> Name l2 -- ^ Layout for lines. - -> Array l Bucket -- ^ Buckets - -> IO (Sinks l1 (Array l2 Char)) -sinkLines n1 n2 bs - = G.sinkLines n1 n2 bs -{-# INLINE sinkLines #-} - 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 @@ -3,8 +3,9 @@ (chunk_i) where import Data.Repa.Flow.Generic.Base -import Data.Repa.Eval.Array as A -import Data.Repa.Array as A +import Data.Repa.Array.Generic as A +import Data.Repa.Array.Generic.Target as A +import Data.Repa.Array.Generic.Index as A #include "repa-flow.h" diff --git a/Data/Repa/Flow/Generic/Array/Distribute.hs b/Data/Repa/Flow/Generic/Array/Distribute.hs --- a/Data/Repa/Flow/Generic/Array/Distribute.hs +++ b/Data/Repa/Flow/Generic/Array/Distribute.hs @@ -9,7 +9,8 @@ , ddistribute2_o) where import Data.Repa.Flow.Generic.Base -import Data.Repa.Array +import Data.Repa.Array.Generic +import Data.Repa.Array.Generic.Index import Prelude hiding (length) #include "repa-flow.h" diff --git a/Data/Repa/Flow/Generic/Array/Shuffle.hs b/Data/Repa/Flow/Generic/Array/Shuffle.hs --- a/Data/Repa/Flow/Generic/Array/Shuffle.hs +++ b/Data/Repa/Flow/Generic/Array/Shuffle.hs @@ -7,7 +7,10 @@ import Data.Repa.Flow.Generic.Base as F import Data.Repa.Flow.Generic.Map as F import Data.Repa.Flow.Generic.Operator as F -import Data.Repa.Array as A +import Data.Repa.Array.Material as A +import Data.Repa.Array.Generic as A +import Data.Repa.Array.Generic.Index as A +import Data.Repa.Array.Meta as A import Data.Repa.Eval.Elt import Control.Monad #include "repa-flow.h" diff --git a/Data/Repa/Flow/Generic/Array/Unchunk.hs b/Data/Repa/Flow/Generic/Array/Unchunk.hs --- a/Data/Repa/Flow/Generic/Array/Unchunk.hs +++ b/Data/Repa/Flow/Generic/Array/Unchunk.hs @@ -3,7 +3,7 @@ (unchunk_i) where import Data.Repa.Flow.Generic.Base -import Data.Repa.Array as A +import Data.Repa.Array.Generic as A #include "repa-flow.h" diff --git a/Data/Repa/Flow/Generic/Base.hs b/Data/Repa/Flow/Generic/Base.hs --- a/Data/Repa/Flow/Generic/Base.hs +++ b/Data/Repa/Flow/Generic/Base.hs @@ -11,7 +11,7 @@ , finalize_o) where import Data.Repa.Flow.States -import Data.Repa.Array as A +import Data.Repa.Array.Generic.Index as A import Control.Monad #include "repa-flow.h" 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 @@ -15,195 +15,12 @@ , sinkLines -- * Sieving - , sieve_o) + , sieve_o + + -- * Tables + , sourceCSV + , sourceTSV) where import Data.Repa.Flow.IO.Bucket -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 as A -import Data.Char -import System.IO -import Data.Word -import Prelude as P -#include "repa-flow.h" - - --- Sourcing --------------------------------------------------------------------------------------- --- | Read complete records of data form a bucket, into chunks of the given length. --- We read as many complete records as will fit into each chunk. --- --- The records are separated by a special terminating character, which the --- given predicate detects. After reading a chunk of data we seek the bucket to --- just after the last complete record that was read, so we can continue to --- read more complete records next time. --- --- If we cannot fit at least one complete record in the chunk then perform --- the given failure action. Limiting the chunk length guards against the --- case where a large input file is malformed, as we won't try to read the --- whole file into memory. --- --- * Data is read into foreign memory without copying it through the GHC heap. --- * The provided file handle must support seeking, else you'll get an exception. --- -sourceRecords - :: BulkI l Bucket - => Integer -- ^ Chunk length in bytes. - -> (Word8 -> Bool) -- ^ Detect the end of a record. - -> IO () -- ^ Action to perform if we can't get a whole record. - -> Array l Bucket -- ^ Source buckets. - -> IO (Sources Int IO (Array N (Array F Word8))) - -sourceRecords len pSep aFail hs - = smap_i (\_ !c -> A.segmentOn pSep c) - =<< sourceChunks len pSep aFail hs -{-# INLINE sourceRecords #-} - - --- | Like `sourceRecords`, but produce all records in a single vector. -sourceChunks - :: BulkI l Bucket - => Integer -- ^ Chunk length in bytes. - -> (Word8 -> Bool) -- ^ Detect the end of a record. - -> IO () -- ^ Action to perform if we can't get a whole record. - -> Array l Bucket -- ^ Source buckets. - -> IO (Sources (Index l) IO (Array F Word8)) - -sourceChunks len pSep aFail bs - = return $ Sources (A.extent $ A.layout bs) pull_sourceChunks - where - pull_sourceChunks i eat eject - = let b = bs `index` i - in bAtEnd b >>= \eof -> - if eof - -- We're at the end of the file. - then eject - - else do - -- Read a new chunk from the file. - arr <- bGetArray b len - - -- Find the end of the last record in the file. - let !mLenSlack = findIndex pSep (A.reverse arr) - - case mLenSlack of - -- If we couldn't find the end of record then apply the failure action. - Nothing -> aFail - - -- Work out how long the record is. - Just lenSlack - -> do let !lenArr = A.length arr - let !ixSplit = lenArr - lenSlack - - -- Seek the file to just after the last complete record. - bSeek b RelativeSeek (fromIntegral $ negate lenSlack) - - -- Eat complete records at the start of the chunk. - eat $ window 0 ixSplit arr - {-# INLINE pull_sourceChunks #-} -{-# INLINE_FLOW sourceChunks #-} - - --- | Read 8-byte ASCII characters from some files, using the given chunk length. --- --- * Data is read into foreign memory without copying it through the GHC heap. --- * All chunks have the same size, except possibly the last one. ----- -sourceChars - :: Bulk l Bucket - => Integer -- ^ Chunk length in bytes. - -> Array l Bucket -- ^ Buckets. - -> IO (Sources (Index l) IO (Array F Char)) -sourceChars len bs - = smap_i (\_ !c -> A.computeS F $ A.map (chr . fromIntegral) c) - =<< sourceBytes len bs -{-# INLINE sourceChars #-} - - --- | Read data from some files, using the given chunk length. --- --- * Data is read into foreign memory without copying it through the GHC heap. --- * All chunks have the same size, except possibly the last one. --- -sourceBytes - :: Bulk l Bucket - => Integer -- ^ Chunk length in bytes. - -> Array l Bucket -- ^ Buckets. - -> IO (Sources (Index l) IO (Array F Word8)) -sourceBytes len bs - = return $ Sources (A.extent $ A.layout bs) pull_sourceBytes - where - pull_sourceBytes 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 len - eat chunk - {-# INLINE pull_sourceBytes #-} -{-# INLINE_FLOW sourceBytes #-} - - --- Sinking ---------------------------------------------------------------------------------------- --- | Write vectors of text lines to the given files handles. --- --- * Data is copied into a new buffer to insert newlines before being --- written out. --- -sinkLines - :: ( Bulk l Bucket - , BulkI l1 (Array l2 Char) - , BulkI l2 Char, Unpack (Array l2 Char) t2) - => Name l1 -- ^ Layout of chunks of lines. - -> Name l2 -- ^ Layout of lines. - -> Array l Bucket -- ^ Buckets. - -> IO (Sinks (Index l) IO (Array l1 (Array l2 Char))) -sinkLines _ _ !bs - = smap_o (\_ !c -> computeS F $ A.map (fromIntegral . ord) $ concatWith F fl c) - =<< sinkBytes bs - where !fl = A.fromList F ['\n'] -{-# INLINE sinkLines #-} - - --- | Write chunks of 8-byte ASCII characters to the given file handles. --- --- * Data is copied into a foreign buffer to truncate the characters --- to 8-bits each before being written out. --- -sinkChars - :: (Bulk l Bucket, BulkI r Char) - => Array l Bucket -- ^ Buckets. - -> IO (Sinks (Index l) IO (Array r Char)) -sinkChars !bs - = smap_o (\_ !c -> computeS F $ A.map (fromIntegral . ord) c) - =<< sinkBytes bs -{-# INLINE sinkChars #-} - - --- | Write chunks of bytes to the given file handles. --- --- * Data is written out directly from the provided buffer. --- -sinkBytes - :: Bulk l Bucket - => Array l Bucket -- ^ Buckets. - -> IO (Sinks (Index l) IO (Array F Word8)) -sinkBytes !bs - = do let push_sinkBytes i !chunk - = bPutArray (bs `index` i) chunk - {-# NOINLINE push_sinkBytes #-} - - let eject_sinkBytes i - = bClose (bs `index` i) - {-# INLINE eject_sinkBytes #-} - - return $ Sinks (A.extent $ A.layout bs) push_sinkBytes eject_sinkBytes -{-# INLINE_FLOW sinkBytes #-} - +import Data.Repa.Flow.Generic.IO.Base as F +import Data.Repa.Flow.Generic.IO.XSV as F diff --git a/Data/Repa/Flow/Generic/IO/Base.hs b/Data/Repa/Flow/Generic/IO/Base.hs new file mode 100644 --- /dev/null +++ b/Data/Repa/Flow/Generic/IO/Base.hs @@ -0,0 +1,212 @@ + +module Data.Repa.Flow.Generic.IO.Base + ( -- * Buckets + module Data.Repa.Flow.IO.Bucket + + -- * Sourcing + , sourceBytes + , sourceChars + , sourceChunks + , sourceRecords + + -- * Sinking + , sinkBytes + , sinkChars + , sinkLines + + -- * Sieving + , sieve_o) +where +import Data.Repa.Flow.IO.Bucket +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 +import Data.Repa.Array.Generic as A +import Data.Repa.Array.Generic.Index as A +import Data.Char +import System.IO +import Data.Word +import Prelude as P +#include "repa-flow.h" + + +-- Sourcing --------------------------------------------------------------------------------------- +-- | Read complete records of data form a bucket, into chunks of the given length. +-- We read as many complete records as will fit into each chunk. +-- +-- The records are separated by a special terminating character, which the +-- given predicate detects. After reading a chunk of data we seek the bucket to +-- just after the last complete record that was read, so we can continue to +-- read more complete records next time. +-- +-- If we cannot fit at least one complete record in the chunk then perform +-- the given failure action. Limiting the chunk length guards against the +-- case where a large input file is malformed, as we won't try to read the +-- whole file into memory. +-- +-- * Data is read into foreign memory without copying it through the GHC heap. +-- * The provided file handle must support seeking, else you'll get an exception. +-- +sourceRecords + :: BulkI l Bucket + => Integer -- ^ Chunk length in bytes. + -> (Word8 -> Bool) -- ^ Detect the end of a record. + -> IO () -- ^ Action to perform if we can't get a whole record. + -> Array l Bucket -- ^ Source buckets. + -> IO (Sources Int IO (Array N (Array F Word8))) + +sourceRecords len pSep aFail hs + = smap_i (\_ !c -> A.segmentOn pSep c) + =<< sourceChunks len pSep aFail hs +{-# INLINE sourceRecords #-} + + +-- | Like `sourceRecords`, but produce all records in a single vector. +sourceChunks + :: BulkI l Bucket + => Integer -- ^ Chunk length in bytes. + -> (Word8 -> Bool) -- ^ Detect the end of a record. + -> IO () -- ^ Action to perform if we can't get a whole record. + -> Array l Bucket -- ^ Source buckets. + -> IO (Sources (Index l) IO (Array F Word8)) + +sourceChunks len pSep aFail bs + = return $ Sources (A.extent $ A.layout bs) pull_sourceChunks + where + pull_sourceChunks i eat eject + = let b = bs `index` i + in bAtEnd b >>= \eof -> + if eof + -- We're at the end of the file. + then eject + + else do + -- Read a new chunk from the file. + arr <- bGetArray b len + + -- Find the end of the last record in the file. + let !mLenSlack = findIndex pSep (A.reverse arr) + + case mLenSlack of + -- If we couldn't find the end of record then apply the failure action. + Nothing -> aFail + + -- Work out how long the record is. + Just lenSlack + -> do let !lenArr = A.length arr + let !ixSplit = lenArr - lenSlack + + -- Seek the file to just after the last complete record. + bSeek b RelativeSeek (fromIntegral $ negate lenSlack) + + -- Eat complete records at the start of the chunk. + eat $ window 0 ixSplit arr + {-# INLINE pull_sourceChunks #-} +{-# INLINE_FLOW sourceChunks #-} + + +-- | Read 8-byte ASCII characters from some files, using the given chunk length. +-- +-- * Data is read into foreign memory without copying it through the GHC heap. +-- * All chunks have the same size, except possibly the last one. +---- +sourceChars + :: Bulk l Bucket + => Integer -- ^ Chunk length in bytes. + -> Array l Bucket -- ^ Buckets. + -> IO (Sources (Index l) IO (Array F Char)) +sourceChars len bs + = smap_i (\_ !c -> A.computeS F $ A.map (chr . fromIntegral) c) + =<< sourceBytes len bs +{-# INLINE sourceChars #-} + + +-- | Read data from some files, using the given chunk length. +-- +-- * Data is read into foreign memory without copying it through the GHC heap. +-- * All chunks have the same size, except possibly the last one. +-- +sourceBytes + :: Bulk l Bucket + => Integer -- ^ Chunk length in bytes. + -> Array l Bucket -- ^ Buckets. + -> IO (Sources (Index l) IO (Array F Word8)) +sourceBytes len bs + = return $ Sources (A.extent $ A.layout bs) pull_sourceBytes + where + pull_sourceBytes 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 len + eat chunk + {-# INLINE pull_sourceBytes #-} +{-# INLINE_FLOW sourceBytes #-} + + +-- Sinking ---------------------------------------------------------------------------------------- +-- | Write vectors of text lines to the given files handles. +-- +-- * Data is copied into a new buffer to insert newlines before being +-- written out. +-- +sinkLines + :: ( Bulk l Bucket + , BulkI l1 (Array l2 Char) + , BulkI l2 Char, Unpack (Array l2 Char) t2) + => Name l1 -- ^ Layout of chunks of lines. + -> Name l2 -- ^ Layout of lines. + -> Array l Bucket -- ^ Buckets. + -> IO (Sinks (Index l) IO (Array l1 (Array l2 Char))) +sinkLines _ _ !bs + = smap_o (\_ !c -> computeS F $ A.map (fromIntegral . ord) $ concatWith F fl c) + =<< sinkBytes bs + where !fl = A.fromList F ['\n'] +{-# INLINE sinkLines #-} + + +-- | Write chunks of 8-byte ASCII characters to the given file handles. +-- +-- * Data is copied into a foreign buffer to truncate the characters +-- to 8-bits each before being written out. +-- +sinkChars + :: (Bulk l Bucket, BulkI r Char) + => Array l Bucket -- ^ Buckets. + -> IO (Sinks (Index l) IO (Array r Char)) +sinkChars !bs + = smap_o (\_ !c -> computeS F $ A.map (fromIntegral . ord) c) + =<< sinkBytes bs +{-# INLINE sinkChars #-} + + +-- | Write chunks of bytes to the given file handles. +-- +-- * Data is written out directly from the provided buffer. +-- +sinkBytes + :: Bulk l Bucket + => Array l Bucket -- ^ Buckets. + -> IO (Sinks (Index l) IO (Array F Word8)) +sinkBytes !bs + = do let push_sinkBytes i !chunk + = bPutArray (bs `index` i) chunk + {-# NOINLINE push_sinkBytes #-} + + let eject_sinkBytes i + = bClose (bs `index` i) + {-# INLINE eject_sinkBytes #-} + + return $ Sinks (A.extent $ A.layout bs) push_sinkBytes eject_sinkBytes +{-# INLINE_FLOW sinkBytes #-} + 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 @@ -3,9 +3,9 @@ (sieve_o) where import Data.Repa.Flow.Generic.Base -import Data.Repa.Array as A import Data.Repa.Array.Material as A -import Data.Repa.IO.Array as A +import Data.Repa.Array.Generic.Convert as A +import Data.Repa.Array.Auto.IO as A import System.IO import Data.Word #include "repa-flow.h" @@ -39,7 +39,7 @@ -> do -- TODO: repeatededly opening and closing the file -- will be very slow. h <- openBinaryFile path AppendMode - hPutArray h arr + hPutArray h (convert arr) hClose h -- TODO: ignore any more incoming data after being ejected. diff --git a/Data/Repa/Flow/Generic/IO/XSV.hs b/Data/Repa/Flow/Generic/IO/XSV.hs new file mode 100644 --- /dev/null +++ b/Data/Repa/Flow/Generic/IO/XSV.hs @@ -0,0 +1,90 @@ + +module Data.Repa.Flow.Generic.IO.XSV + ( sourceCSV + , sourceTSV) +where +import Data.Repa.Flow.Generic.Base +import Data.Repa.Flow.Generic.Map +import Data.Repa.Flow.Generic.IO.Base +import Data.Repa.Array.Meta.Delayed as A +import Data.Repa.Array.Generic as A +import Data.Repa.Array.Material as A +import Data.Char +#include "repa-flow.h" + + +------------------------------------------------------------------------------- +-- | Read a file containing Comma-Separated-Values. +-- +-- TODO: handle escaped commas. +-- TODO: check CSV file standard. +-- +sourceCSV + :: BulkI l Bucket + => Integer -- Chunk length. + -> IO () -- Action to perform if we find line longer + -- than the chunk length. + -> Array l Bucket -- File paths. + -> IO (Sources Int IO (Array N (Array N (Array F Char)))) + +sourceCSV nChunk aFail bs + = do + -- Rows are separated by new lines, + -- fields are separated by commas. + let !nl = fromIntegral $ ord '\n' + let !nr = fromIntegral $ ord '\r' + let !nt = fromIntegral $ ord ',' + + -- Stream chunks of data from the input file, where the chunks end + -- cleanly at line boundaries. + sChunk <- sourceChunks nChunk (== nl) aFail bs + sRows8 <- map_i (A.diceSep nt nl . A.filter U (/= nr)) sChunk + + -- Convert element data from Word8 to Char. + -- Chars take 4 bytes each, but are standard Haskell and pretty + -- print properly. We've done the dicing on the smaller Word8 + -- version, and now map across the elements vector in the array + -- to do the conversion. + sRows <- map_i (A.mapElems (A.mapElems + (A.computeS F . A.map (chr . fromIntegral)))) + sRows8 + + return sRows +{-# INLINE sourceCSV #-} + + +------------------------------------------------------------------------------- +-- | Read a file containing Tab-Separated-Values. +sourceTSV + :: BulkI l Bucket + => Integer -- Chunk length. + -> IO () -- Action to perform if we find line longer + -- than the chunk length. + -> Array l Bucket -- File paths. + -> IO (Sources Int IO (Array N (Array N (Array F Char)))) + +sourceTSV nChunk aFail bs + = do + -- Rows are separated by new lines, + -- fields are separated by tabs. + let !nl = fromIntegral $ ord '\n' + let !nr = fromIntegral $ ord '\r' + let !nt = fromIntegral $ ord '\t' + + -- Stream chunks of data from the input file, where the chunks end + -- cleanly at line boundaries. + sChunk <- sourceChunks nChunk (== nl) aFail bs + sRows8 <- map_i (A.diceSep nt nl . A.filter U (/= nr)) sChunk + + -- Convert element data from Word8 to Char. + -- Chars take 4 bytes each, but are standard Haskell and pretty + -- print properly. We've done the dicing on the smaller Word8 + -- version, and now map across the elements vector in the array + -- to do the conversion. + sRows <- map_i (A.mapElems (A.mapElems + (A.computeS F . A.map (chr . fromIntegral)))) + sRows8 + + return sRows +{-# INLINE sourceTSV #-} + 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 @@ -44,7 +44,8 @@ import Data.Repa.Flow.Generic.List import Data.Repa.Flow.Generic.Connect import Data.Repa.Flow.Generic.Base -import Data.Repa.Array as A +import Data.Repa.Array.Generic.Index as A +import Data.Repa.Array.Generic as A import Data.IORef import Control.Monad import Debug.Trace diff --git a/Data/Repa/Flow/IO/Binary.hs b/Data/Repa/Flow/IO/Binary.hs new file mode 100644 --- /dev/null +++ b/Data/Repa/Flow/IO/Binary.hs @@ -0,0 +1,38 @@ + +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 @@ -1,7 +1,6 @@ module Data.Repa.Flow.IO.Bucket - ( Bucket - , bucketLength + ( Bucket (..) , openBucket , hBucket @@ -24,18 +23,19 @@ , bGetArray , bPutArray) where -import Data.Repa.Array as A -import Data.Repa.Array.Material as A -import Data.Repa.Array.RowWise as A -import Data.Repa.IO.Array as A -import qualified Foreign.Storable as Foreign -import qualified Foreign.Marshal.Alloc as Foreign +import Data.Repa.Array.Material as A +import Data.Repa.Array.Generic as A +import Data.Repa.Array.Meta.Dense as A +import Data.Repa.Array.Meta.RowWise as A +import Data.Repa.Array.Auto.IO as A +import qualified Foreign.Storable as Foreign +import qualified Foreign.Marshal.Alloc as Foreign import Control.Monad import Data.Word import System.IO import System.FilePath import System.Directory -import Prelude as P +import Prelude as P -- | A bucket represents portion of a whole data-set on disk, @@ -302,6 +302,7 @@ -- -- If the directory is named @somedir@ then the files are named -- @somedir/000000@, @somedir/000001@, @somedir/000002@ and so on. +-- toDir :: Int -- ^ Number of buckets to create. -> FilePath -- ^ Path to directory. -> (Array B Bucket -> IO b) -- ^ Consumer. @@ -491,7 +492,8 @@ -> let lenRemain = lenMax - posBucket in min lenWanted lenRemain - hGetArray (bucketHandle bucket) + liftM (convert F) + $ hGetArray (bucketHandle bucket) $ fromIntegral len {-# NOINLINE bGetArray #-} @@ -499,6 +501,6 @@ -- | Put some data in a bucket. bPutArray :: Bucket -> Array F Word8 -> IO () bPutArray bucket arr - = hPutArray (bucketHandle bucket) arr + = hPutArray (bucketHandle bucket) (convert A arr) {-# NOINLINE bPutArray #-} diff --git a/Data/Repa/Flow/IO/Storable.hs b/Data/Repa/Flow/IO/Storable.hs new file mode 100644 --- /dev/null +++ b/Data/Repa/Flow/IO/Storable.hs @@ -0,0 +1,135 @@ + +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/IO.hs b/Data/Repa/Flow/Simple/IO.hs --- a/Data/Repa/Flow/Simple/IO.hs +++ b/Data/Repa/Flow/Simple/IO.hs @@ -8,9 +8,9 @@ where import Data.Repa.Flow.IO.Bucket import Data.Repa.Flow.Simple.Base -import Data.Word -import Data.Repa.Array as A +import Data.Repa.Array.Generic as A import Data.Repa.Array.Material as A +import Data.Word import qualified Data.Repa.Flow.Generic.IO as G #include "repa-flow.h" 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.0.0.2 +Version: 4.1.0.1 License: BSD3 License-file: LICENSE Author: The Repa Development Team @@ -23,24 +23,26 @@ primitive == 0.5.4.*, containers == 0.5.*, text == 1.2.*, - repa-stream == 4.0.0.*, repa-eval == 4.0.0.*, - repa-array == 4.0.0.* + repa-stream == 4.1.0.*, + repa-array == 4.1.0.* exposed-modules: + Data.Repa.Flow.Auto + Data.Repa.Flow.Auto.Debug + Data.Repa.Flow.Auto.IO + Data.Repa.Flow.Auto.SizedIO + Data.Repa.Flow.Chunked Data.Repa.Flow.Chunked.IO - Data.Repa.Flow.Default - Data.Repa.Flow.Default.Debug - Data.Repa.Flow.Default.IO - Data.Repa.Flow.Default.SizedIO - Data.Repa.Flow.Generic 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 @@ -56,9 +58,6 @@ Data.Repa.Flow.Chunked.Groups Data.Repa.Flow.Chunked.Operator - Data.Repa.Flow.Default.IO.TSV - Data.Repa.Flow.Default.IO.CSV - Data.Repa.Flow.Generic.Base Data.Repa.Flow.Generic.Connect Data.Repa.Flow.Generic.List @@ -69,7 +68,9 @@ Data.Repa.Flow.Generic.Array.Shuffle Data.Repa.Flow.Generic.Array.Chunk Data.Repa.Flow.Generic.Array.Unchunk + Data.Repa.Flow.Generic.IO.Base Data.Repa.Flow.Generic.IO.Sieve + Data.Repa.Flow.Generic.IO.XSV Data.Repa.Flow.Simple.Base Data.Repa.Flow.Simple.List