repa-flow 4.0.0.2 → 4.1.0.1

raw patch · 33 files changed

+1592/−1259 lines, 33 filesdep ~repa-arraydep ~repa-streamPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: repa-array, repa-stream

API changes (from Hackage documentation)

Files

• Data/Repa/Flow.hs +20/−26
• Data/Repa/Flow/Auto.hs +499/−0
• Data/Repa/Flow/Auto/Debug.hs +90/−0
• Data/Repa/Flow/Auto/IO.hs +314/−0
• Data/Repa/Flow/Auto/SizedIO.hs +114/−0
• Data/Repa/Flow/Chunked/Base.hs +6/−5
• Data/Repa/Flow/Chunked/Fold.hs +3/−1
• Data/Repa/Flow/Chunked/Folds.hs +8/−5
• Data/Repa/Flow/Chunked/Groups.hs +7/−5
• Data/Repa/Flow/Chunked/IO.hs +1/−1
• Data/Repa/Flow/Chunked/Map.hs +4/−2
• Data/Repa/Flow/Chunked/Operator.hs +1/−1
• Data/Repa/Flow/Default.hs +0/−516
• Data/Repa/Flow/Default/Debug.hs +0/−94
• Data/Repa/Flow/Default/IO.hs +0/−159
• Data/Repa/Flow/Default/IO/CSV.hs +0/−52
• Data/Repa/Flow/Default/IO/TSV.hs +0/−48
• Data/Repa/Flow/Default/SizedIO.hs +0/−120
• Data/Repa/Flow/Generic/Array/Chunk.hs +3/−2
• Data/Repa/Flow/Generic/Array/Distribute.hs +2/−1
• Data/Repa/Flow/Generic/Array/Shuffle.hs +4/−1
• Data/Repa/Flow/Generic/Array/Unchunk.hs +1/−1
• Data/Repa/Flow/Generic/Base.hs +1/−1
• Data/Repa/Flow/Generic/IO.hs +7/−190
• Data/Repa/Flow/Generic/IO/Base.hs +212/−0
• Data/Repa/Flow/Generic/IO/Sieve.hs +3/−3
• Data/Repa/Flow/Generic/IO/XSV.hs +90/−0
• Data/Repa/Flow/Generic/Operator.hs +2/−1
• Data/Repa/Flow/IO/Binary.hs +38/−0
• Data/Repa/Flow/IO/Bucket.hs +13/−11
• Data/Repa/Flow/IO/Storable.hs +135/−0
• Data/Repa/Flow/Simple/IO.hs +2/−2
• repa-flow.cabal +12/−11

@@ -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)

@@ -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)

@@ -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' #-}++

@@ -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 #-}+

@@ -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 #-}+

@@ -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" 

@@ -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"  

@@ -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 ----------------------------------------------------------------------

@@ -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 -------------------------------------------------------------------

@@ -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

@@ -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" 

@@ -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" 

@@ -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)

@@ -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' #-}---

@@ -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 #-}

@@ -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 #-}

@@ -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 #-}

@@ -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 #-}-

@@ -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"  

@@ -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" 

@@ -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"

@@ -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"  

@@ -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" 

@@ -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

@@ -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 #-}+

@@ -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.

@@ -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 #-}+

@@ -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

@@ -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 #-}+

@@ -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 #-} 

@@ -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++

@@ -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" 

@@ -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