repa-flow 4.1.0.1 → 4.2.2.1

raw patch · 32 files changed

+1377/−586 lines, 32 filesdep +hashtablesdep +repa-convertdep +repa-scalardep ~basedep ~filepathdep ~primitivePVP ok

version bump matches the API change (PVP)

Dependencies added: hashtables, repa-convert, repa-scalar

Dependency ranges changed: base, filepath, primitive, repa-array, repa-stream

API changes (from Hackage documentation)

Files

• Data/Repa/Flow.hs +29/−30
• Data/Repa/Flow/Auto.hs +103/−91
• Data/Repa/Flow/Auto/Base.hs +161/−0
• Data/Repa/Flow/Auto/Debug.hs +1/−1
• Data/Repa/Flow/Auto/Format.hs +138/−0
• Data/Repa/Flow/Auto/IO.hs +38/−69
• Data/Repa/Flow/Auto/Select.hs +92/−0
• Data/Repa/Flow/Auto/ZipWith.hs +120/−0
• Data/Repa/Flow/Chunked.hs +36/−4
• Data/Repa/Flow/Chunked/Base.hs +3/−3
• Data/Repa/Flow/Chunked/Fold.hs +0/−1
• Data/Repa/Flow/Chunked/Folds.hs +4/−8
• Data/Repa/Flow/Chunked/Generic.hs +73/−0
• Data/Repa/Flow/Chunked/Groups.hs +3/−7
• Data/Repa/Flow/Chunked/Operator.hs +0/−73
• Data/Repa/Flow/Chunked/Process.hs +81/−0
• Data/Repa/Flow/Chunked/Replicate.hs +19/−0
• Data/Repa/Flow/Generic.hs +9/−2
• Data/Repa/Flow/Generic/Array/Chunk.hs +18/−11
• Data/Repa/Flow/Generic/Eval.hs +29/−1
• Data/Repa/Flow/Generic/IO.hs +4/−0
• Data/Repa/Flow/Generic/IO/Base.hs +1/−2
• Data/Repa/Flow/Generic/IO/Lines.hs +126/−0
• Data/Repa/Flow/Generic/IO/Sieve.hs +101/−17
• Data/Repa/Flow/Generic/Operator.hs +24/−26
• Data/Repa/Flow/Generic/Process.hs +87/−0
• Data/Repa/Flow/IO/Binary.hs +0/−38
• Data/Repa/Flow/IO/Bucket.hs +38/−37
• Data/Repa/Flow/IO/Storable.hs +0/−135
• Data/Repa/Flow/Simple.hs +1/−1
• Data/Repa/Flow/Simple/Operator.hs +10/−10
• repa-flow.cabal +28/−19

@@ -10,9 +10,10 @@ -- @ -- > import Data.Repa.Array           as A -- > import Data.Repa.Flow            as F+-- > import Data.Repa.Flow.IO         as F -- > import Data.Repa.Flow.Auto.Debug as F ----- > ws <- fromFiles' [\"\/usr\/share\/dict\/words\", \"\/usr\/share\/dict\/cracklib-small\"] sourceLines+-- > ws <- fromFiles [\"\/usr\/share\/dict\/words\", \"\/usr\/share\/dict\/cracklib-small\"] sourceLines -- @ -- --   Show the first few elements of the first chunk of the first file.@@ -71,7 +72,7 @@ --   so open a file for each stream: -- -- @--- > out <- toFiles' ["out1.txt", "out2.txt"] sinkLines+-- > out <- toFiles ["out1.txt", "out2.txt"] sinkLines -- @ -- --   Note that the @ws@ and @up@ we used before were bundles of stream @@ -123,19 +124,31 @@         , drainP          -- * Conversion+        -- ** List conversion         , fromList,             fromLists         , toList1,              toLists1 +        -- ** Array conversion+        , fromArray,            fromArrays+        , toArray1,             toArrays1+         -- * Finalizers         , finalize_i,           finalize_o          -- * Flow Operators+        -- ** Replicating+        , replicates_i+         -- ** Mapping-          -- | If you want to work on a chunk at a time then use -          --   `Data.Repa.Flow.Generic.map_i` and-          --   `Data.Repa.Flow.Generic.map_o` from "Data.Repa.Flow.Generic".+        -- | If you want to work on a chunk at a time then use +        --   `Data.Repa.Flow.Generic.map_i` and+        --   `Data.Repa.Flow.Generic.map_o` from "Data.Repa.Flow.Generic".         , map_i,                map_o+        , zipWith_i +        -- | Higher arity zipWith functions.+        , module Data.Repa.Flow.Auto.ZipWith+         -- ** Connecting         , dup_oo         , dup_io@@ -147,12 +160,20 @@         , trigger_o          -- ** Ignorance-        , discard_o         , ignore_o+        , abandon_o          -- ** Splitting         , head_i +        -- ** Concatenation+        , concat_i++        -- ** Selecting+        , select_i,             select_o+        , discard_i,            discard_o+        , mask_i,               mask_o+         -- ** Grouping         , groups_i         , groupsBy_i@@ -164,33 +185,11 @@          -- *** Segmented         , folds_i,              FoldsDict-        , foldGroupsBy_i,       FoldGroupsDict--        -- * Flow I/O-        , defaultChunkSize--        -- ** Buckets-        , module Data.Repa.Flow.IO.Bucket--        -- ** Sourcing-        , sourceCSV-        , sourceTSV-        , sourceRecords-        , sourceLines-        , sourceChars-        , sourceBytes-        , sourcePacked--        -- ** Sinking-        , sinkChars-        , sinkLines-        , sinkBytes-        , sinkPacked-        )+        , foldGroupsBy_i,       FoldGroupsDict) where import Data.Repa.Flow.Auto import Data.Repa.Flow.Auto.Debug-import Data.Repa.Flow.Auto.IO+import Data.Repa.Flow.Auto.ZipWith import Data.Repa.Flow.IO.Bucket import Data.Repa.Flow.States 

@@ -12,20 +12,36 @@         , sinksArity          -- * Conversion+        -- ** List conversion         , fromList,             fromLists         , toList1,              toLists1 +        -- ** Array conversion+        , fromArray,            fromArrays+        , toArray1,             toArrays1+         -- * Evaluation         , drainS         , drainP+        , consumeS          -- * Flow Operators+        -- ** Replicating+        , replicates_i+         -- ** Mapping         -- | If you want to work on a chunk at a time then use          --   `Data.Repa.Flow.Generic.map_i` and         --   `Data.Repa.Flow.Generic.map_o` from "Data.Repa.Flow.Generic".         , map_i,                map_o+        , zipWith_i +        -- ** Processing+        , process_i++        -- | Higher arity zipWith functions.+        , module Data.Repa.Flow.Auto.ZipWith+         -- ** Connecting         , dup_oo         , dup_io@@ -37,12 +53,20 @@         , trigger_o          -- ** Ignorance-        , discard_o         , ignore_o+        , abandon_o          -- ** Splitting         , head_i +        -- ** Concatenation+        , concat_i++        -- ** Selecting+        , select_i,             select_o+        , discard_i,            discard_o+        , mask_i,               mask_o+         -- ** Grouping         , groups_i         , groupsBy_i@@ -60,14 +84,15 @@         , finalize_i,           finalize_o         ) where+import Data.Repa.Flow.Auto.ZipWith+import Data.Repa.Flow.Auto.Select+import Data.Repa.Flow.Auto.Base import Data.Repa.Array.Auto                             hiding (fromList, fromLists)  import Data.Repa.Array.Material.Auto                    (A(..), Name(..))-import Data.Repa.Fusion.Unpack                          as A import qualified Data.Repa.Array.Meta.Window            as A import qualified Data.Repa.Array.Material               as A-import qualified Data.Repa.Array.Generic.Target         as A import qualified Data.Repa.Array.Generic                as A import qualified Data.Repa.Flow.Chunked                 as C hiding (next) import qualified Data.Repa.Flow.Generic                 as G hiding (next)@@ -75,31 +100,6 @@ #include "repa-flow.h"  --- | A bundle of stream sources, where the elements of the stream---   are chunked into arrays.-type Sources a  = C.Sources Int IO A a----- | A bundle of stream sinks,   where the elements of the stream---   are chunked into arrays.----type Sinks a    = C.Sinks Int IO A a----- | Yield the number of streams in the bundle.-sourcesArity :: Sources a -> Int-sourcesArity = G.sourcesArity----- | Yield the number of streams in the bundle.-sinksArity :: Sinks a -> Int-sinksArity = G.sinksArity----- | Shorthand for common type classes.-type Flow a     = (C.Flow  Int IO A a, A.Windowable A a)-- -- Evaluation ----------------------------------------------------------------- -- | Pull all available values from the sources and push them to the sinks. --   Streams in the bundle are processed sequentially, from first to last.@@ -124,43 +124,13 @@ {-# INLINE drainP #-}  --- Conversion -------------------------------------------------------------------- | Given an arity and a list of elements, yield sources that each produce all---   the elements. ------   * All elements are stuffed into a single chunk, and each stream is given---     the same chunk.+-- | Pull all available values from the sources and pass them to the+--   given action. ---fromList :: Build a t-         => Int -> [a] -> IO (Sources a)-fromList xs = C.fromList A xs-{-# INLINE fromList #-}----- | Like `fromLists_i` but take a list of lists. Each each of the inner---   lists is packed into a single chunk.-fromLists :: Build a t-          => Int -> [[a]] -> IO (Sources a)-fromLists xss = C.fromLists A xss-{-# INLINE fromLists #-}----- | Drain a single source from a bundle into a list of elements.-toList1   :: Build a t-          => Int -> Sources a -> IO [a]-toList1 ix s  - | ix >= G.sourcesArity s = return []- | otherwise              = C.toList1 ix s -{-# INLINE toList1 #-}----- | Drain a single source from a bundle into a list of chunks.-toLists1  :: Build a t-          => Int -> Sources a -> IO [[a]]-toLists1 ix s- | ix >= G.sourcesArity s = return []- | otherwise              = C.toLists1 ix s -{-# INLINE toLists1 #-}+consumeS :: A.Bulk A a+         => Sources a -> (Int -> a -> IO ()) -> IO ()+consumeS = C.consumeS+{-# INLINE consumeS #-}   -- Finalizers -----------------------------------------------------------------@@ -207,21 +177,66 @@ {-# INLINE finalize_o #-}  +-- Replicating ----------------------------------------------------------------+-- | Segmented replicate.+replicates_i +        :: (Flow (Int, a), Build a)+        => Sources (Int, a)     -- ^ Source of segment lengths and values.+        -> IO (Sources a)       +replicates_i = C.replicates_i A+{-# INLINE replicates_i #-}++ -- Mapping -------------------------------------------------------------------- -- | Apply a function to all elements pulled from some sources.-map_i   :: (Flow a, Build b bt)+map_i   :: (Flow a, Build b)         => (a -> b) -> Sources a -> IO (Sources b) map_i f s = C.smap_i (\_ x -> f x) s {-# INLINE map_i #-}   -- | Apply a function to all elements pushed to some sinks.-map_o   :: (Flow a, Build b bt)+map_o   :: (Flow a, Build b)         => (a -> b) -> Sinks b   -> IO (Sinks a) map_o f s = C.smap_o (\_ x -> f x) s {-# INLINE map_o #-}  +-- | Combine corresponding elements of two sources with the given function.+zipWith_i +        :: (Flow a, Flow b, Build c)+        => (a -> b -> c) +        -> Sources a -> Sources b +        -> IO (Sources c)+zipWith_i f sa sb+        = C.szipWith_ii A (\_ a b -> f a b) sa sb+{-# INLINE zipWith_i #-}+++-- Processing -----------------------------------------------------------------+-- | Apply a generic stream process to a bundle of sources.+process_i+        :: ( Flow a, Flow b, Build b)+        => (s -> a -> (s, Array b))     -- ^ Worker function.+        -> s                            -- ^ Initial state.+        ->     Sources a                -- ^ Input sources.+        -> IO (Sources b)++process_i = C.process_i+{-# INLINE process_i #-}+++-- Concatenation --------------------------------------------------------------+-- | Concatenate a flow of arrays into a flow of the elements.+concat_i+        :: (Flow a, Build a)+        => Sources (Array a)+        -> IO (Sources a)+concat_i ss+        = G.map_i (A.concat A) ss+{-# INLINE concat_i #-}++ -- Connecting ----------------------------------------------------------------- -- | Send the same data to two consumers. --@@ -303,31 +318,30 @@  -- Ignorance ------------------------------------------------------------------ -- | Create a bundle of sinks of the given arity that drop all data on the---   floor.+--   floor.  -----   * The sinks is strict in the *chunks*, so they are demanded before being---     discarded.  --   * Haskell debugging thunks attached to the chunks will be --     demanded, but thunks attached to elements may not be -- depending on --     whether the chunk representation is strict in the elements. ---discard_o :: Int -> IO (Sinks a)-discard_o = G.discard_o-{-# INLINE discard_o #-}+ignore_o :: Int -> IO (Sinks a)+ignore_o  = G.ignore_o+{-# INLINE ignore_o #-}   -- | Create a bundle of sinks of the given arity that drop all data on the---   floor. +--   floor. -----   * As opposed to `discard_o` the sinks are non-strict in the chunks.+--   * As opposed to `ignore_o` the sinks are non-strict in the chunks. --   * Haskell debugging thunks attached to the chunks will *not* be  --     demanded. ---ignore_o :: Int -> IO (Sinks a)-ignore_o  = G.ignore_o-{-# INLINE ignore_o #-}+abandon_o :: Int -> IO (Sinks a)+abandon_o = G.abandon_o+{-# INLINE abandon_o #-}  + -- Splitting ------------------------------------------------------------------ -- | Given a source index and a length, split the a list of that --   length from the front of the source. Yields a new source for the@@ -357,7 +371,7 @@ -- @ -- groups_i-        :: (GroupsDict a u1 u2, Eq a)+        :: (GroupsDict a, Eq a)         => Sources a       -- ^ Input elements.         -> IO (Sources (a, Int))                                  -- ^ Starting element and length of groups.@@ -369,7 +383,7 @@ -- | Like `groupsBy`, but take a function to determine whether two consecutive --   values should be in the same group. groupsBy_i-        :: GroupsDict a u1 u2+        :: GroupsDict a         => (a -> a -> Bool)     -- ^ Fn to check if consecutive elements                                 --   are in the same group.         -> Sources a       -- ^ Input elements.@@ -382,15 +396,15 @@   -- | Dictionaries needed to perform a grouping.-type GroupsDict a u1 u2-        = C.GroupsDict Int IO A A u1 A u2 a+type GroupsDict a+        = C.GroupsDict Int IO A A A a   -- Folding -------------------------------------------------------------------- -- | Fold all the elements of each stream in a bundle, one stream after the --   other, returning an array of fold results. foldlS-        :: (Flow b, Build a at)+        :: (Flow b, Build a)         => (a -> b -> a)                -- ^ Combining funtion.         -> a                            -- ^ Starting value.         -> Sources b                    -- ^ Input elements to fold.@@ -404,7 +418,7 @@ -- | Fold all the elements of each stream in a bundle, one stream after the --   other, returning an array of fold results. foldlAllS-        :: (Flow b)+        :: Flow b         => (a -> b -> a)                -- ^ Combining funtion.         -> a                            -- ^ Starting value.         -> Sources b                    -- ^ Input elements to fold.@@ -439,7 +453,7 @@ -- [(\'a\',10),(\'b\',600),(\'c\',1),(\'d\',1),(\'e\',1)] -- @ ---folds_i :: FoldsDict n a b u1 u2 u3 u4+folds_i :: FoldsDict n a b         => (a -> b -> b)          -- ^ Worker function.         -> b                      -- ^ Initial state when folding each segment.         -> Sources (n, Int)       -- ^ Segment lengths.@@ -452,8 +466,8 @@ {-# INLINE folds_i #-}  -- | Dictionaries needed to perform a segmented fold.-type FoldsDict n a b u1 u2 u3 u4-        = C.FoldsDict Int IO A u1 A u2 A u3 A u4 n a b+type FoldsDict n a b+        = C.FoldsDict Int IO A A A A n a b   -- | Combination of `groupsBy_i` and `folds_i`. We determine the the segment@@ -475,7 +489,7 @@ -- @ -- foldGroupsBy_i-        :: (FoldGroupsDict n a b u1 u2)+        :: (FoldGroupsDict n a b)         => (n -> n -> Bool)     -- ^ Fn to check if consecutive elements                                 --   are in the same group.         -> (a -> b -> b)        -- ^ Worker function for the fold.@@ -490,10 +504,8 @@ {-# INLINE foldGroupsBy_i #-}   -type FoldGroupsDict  n a b u1 u2+type FoldGroupsDict  n a b       = ( A.BulkI    A n         , A.Material A a         , A.Material A n-        , A.Material A b-        , Unpack  (A.Buffer A n) u1-        , Unpack  (A.Buffer A b) u2)+        , A.Material A b)

@@ -0,0 +1,161 @@++module Data.Repa.Flow.Auto.Base+        ( Sources+        , Sinks+        , Flow+        , sourcesArity+        , sinksArity++        -- * Conversion+        -- ** List conversion+        , fromList,             fromLists+        , toList1,              toLists1++        -- ** Array conversion+        , fromArray,            fromArrays+        , toArray1,             toArrays1)+where+import Data.Repa.Array.Auto+        hiding (fromList, fromLists)+import Data.Repa.Array.Material.Auto                    (A(..), Name(..))+import qualified Data.Repa.Flow.Chunked                 as C+import qualified Data.Repa.Flow.Generic                 as G+import qualified Data.Repa.Array.Meta.Window            as A+import qualified Data.Repa.Array.Auto                   as A+#include "repa-flow.h"+++-- | A bundle of stream sources, where the elements of the stream+--   are chunked into arrays.+type Sources a  = C.Sources Int IO A a+++-- | A bundle of stream sinks,   where the elements of the stream+--   are chunked into arrays.+--+type Sinks a    = C.Sinks Int IO A a+++-- | Shorthand for common type classes.+type Flow a     = (C.Flow  Int IO A a, A.Windowable A a)+++-- | Yield the number of streams in the bundle.+sourcesArity :: Sources a -> Int+sourcesArity = G.sourcesArity+{-# INLINE_FLOW sourcesArity #-}+++-- | Yield the number of streams in the bundle.+sinksArity :: Sinks a -> Int+sinksArity = G.sinksArity+{-# INLINE_FLOW sinksArity #-}+++-- Conversion -----------------------------------------------------------------+-- | Given an arity and a list of elements,+--   yield sources that each produce all the elements. +--+--   * All elements are stuffed into a single chunk,+--     and each stream is given the same chunk.+--+fromList :: Build a+         => Int -> [a] -> IO (Sources a)+fromList xs = C.fromList A xs+{-# INLINE_FLOW fromList #-}+++-- | Like `fromList` but take a list of lists.+--   Each each of the inner lists is packed into a single chunk.+fromLists :: Build a+          => Int -> [[a]] -> IO (Sources a)+fromLists xss = C.fromLists A xss+{-# INLINE_FLOW fromLists #-}+++-- | Drain a single source from a bundle into a list of elements.+--+--   * If the index does not specify a valid stream then the result will+--     be empty.+--  +toList1   :: Build a+          => Int -> Sources a -> IO [a]+toList1 ix s  + | ix < 0 || ix >= G.sourcesArity s + = return []++ | otherwise+ = C.toList1 ix s +{-# INLINE_FLOW toList1 #-}+++-- | Drain a single source from a bundle into a list of chunks.+--+--   * If the index does not specify a valid stream then the result will +--     be empty.+--+toLists1  :: Build a+          => Int -> Sources a -> IO [[a]]+toLists1 ix s+ | ix < 0 || ix >= G.sourcesArity s + = return []++ | otherwise+ = C.toLists1 ix s +{-# INLINE_FLOW toLists1 #-}+++-------------------------------------------------------------------------------+-- | Given an arity and an array of elements,+--   yield sources that each produce all the elements.+--+--   * All elements are stuffed into a single chunk,+--     and each stream is given the same chunk.+--+fromArray :: Build a+          => Int -> Array a -> IO (Sources a)+fromArray n arr = G.fromList n [arr]+{-# INLINE_FLOW fromArray #-}+++-- | Like `fromArray` but take an array of arrays.+--   Each of the inner arrays is packed into a single chunk.+fromArrays :: (Elem a, Build a)+           => Int -> Array (Array a) -> IO (Sources a)+fromArrays n arrs = G.fromList n (A.toList arrs)+{-# INLINE_FLOW fromArrays #-}+++-- | Drain a single source from a bundle into an array of elements.+--+--   * If the index does not specify a valid stream then the result will+--     be empty.+--+toArray1   :: (Elem a, Build a)+           => Int -> Sources a -> IO (Array a)+toArray1 ix ss+ | ix < 0 || ix >= G.sourcesArity ss+ = return $ A.fromList []++ | otherwise+ = do   chunks  <- G.toList1 ix ss+        return  $  A.concat $ A.fromList chunks+{-# INLINE_FLOW toArray1 #-}+++-- | Drain a single source from a bundle into an array of elements.+--+--   * If the index does not specify a valid stream then the result will+--     be empty.+--+toArrays1  :: (Elem a, Build a)+           => Int -> Sources a -> IO (Array (Array a))+toArrays1 ix ss+ | ix < 0 || ix >= G.sourcesArity ss+ = return $ A.fromList []++ | otherwise+ = do   chunks  <- G.toList1 ix ss+        return  $ A.fromList chunks+{-# INLINE_FLOW toArrays1 #-}+

@@ -1,4 +1,4 @@-{-# LANGUAGE OverlappingInstances, TypeSynonymInstances, FlexibleInstances #-}+{-# LANGUAGE TypeSynonymInstances, FlexibleInstances #-} module Data.Repa.Flow.Auto.Debug         ( -- * More           more,         more'

@@ -0,0 +1,138 @@++module Data.Repa.Flow.Auto.Format+        ( -- * Pre-defined data formats+          module Data.Repa.Convert.Formats++          -- * Packing++          -- ** Standard+        , packFormat_i,         flatPackFormat_i+          +          -- ** With newlines+        , packFormatLn_i,       flatPackFormatLn_i++          -- ** Default ASCII format+        , packAsciiLn_i,        flatPackAsciiLn_i++          -- ** Keys and values+        , keyPackAsciiLn_i)+where+import Data.Word+import Data.Char+import Data.Repa.Convert.Formats+import Data.Repa.Flow.Auto                              as F+import Data.Repa.Array                                  as A+import Data.Repa.Array.Auto.Format                      as A+import qualified Data.Repa.Convert                      as C+import qualified Data.Repa.Flow.Generic                 as G+#include "repa-flow.h"+++---------------------------------------------------------------------------------------------------+-- | Pack elements into the given storage formats.+packFormat_i+        :: (C.Packable format, Elem (Value format))+        => format                       -- ^ Desination format for data.+        -> Sources (Value format)       -- ^ Sources of values to be packed.+        -> IO (Sources (Array Word8))   -- ^ Packed data.++packFormat_i format ss+ = let+        packElem x +         = let Just arr = A.packFormat format x+           in  arr+        {-# INLINE packElem #-}++   in   G.map_i (A.map packElem) ss+{-# INLINE_FLOW packFormat_i #-}+++-- | Like `packFormat_i`, but return sources of flat bytes.+flatPackFormat_i +        :: (C.Packable format, Elem (Value format))+        => format -> Sources (Value format) -> IO (Sources Word8)+flatPackFormat_i format ss +        = concat_i =<< packFormat_i format ss+{-# INLINE_FLOW flatPackFormat_i #-}+++---------------------------------------------------------------------------------------------------+-- | Like `packFormat_i`, but also append a newline character+--   after every packed element.+packFormatLn_i+        :: (C.Packable format, Elem (Value format))+        => format                       -- ^ Destination format for data.+        -> Sources (Value format)       -- ^ Sources of values to be packed.+        -> IO (Sources (Array Word8))   -- ^ Packed data.++packFormatLn_i format ss+        =   F.map_i (\arr -> A.concat $ A.fromList [arr, nl])   -- TODO: avoid copy+        =<< packFormat_i format ss+        where   !nl = A.fromList [fromIntegral $ ord '\n']+{-# INLINE_FLOW packFormatLn_i #-}+++-- | Like `packFormatLn_i`, but return sources of flat bytes.+flatPackFormatLn_i +        :: (C.Packable format, Elem (Value format))+        => format -> Sources (Value format) -> IO (Sources Word8)+flatPackFormatLn_i format ss+        = concat_i =<< packFormatLn_i format ss+{-# INLINE_FLOW flatPackFormatLn_i #-}+++---------------------------------------------------------------------------------------------------+-- | Like `packFormatLn_i`,+--   but use a default, human-readable format to encode the values.+packAsciiLn_i+        :: forall a +        . ( C.FormatAscii a, a ~ Value (C.FormatAscii' a)+          , Elem a+          , Packable (C.FormatAscii' a))+        => Sources a                    -- ^ Sources of values to be packed.+        -> IO (Sources (Array Word8))   -- ^ Packed data.++packAsciiLn_i ss+        =   F.map_i (\arr -> A.concat $ A.fromList [arr, nl])    -- TODO: avoid copy+        =<< packFormat_i (C.formatAscii proxy) ss+        where   !nl     = A.fromList [fromIntegral $ ord '\n']+                proxy   = (error "repa-flow: packAscii_i proxy" :: a)+{-# INLINE_FLOW packAsciiLn_i #-}+++-- | Like `packAsciiLn_i`, but return sources of flat bytes.+flatPackAsciiLn_i+        :: forall a +        . ( C.FormatAscii a, a ~ Value (C.FormatAscii' a)+          , Elem a+          , Packable (C.FormatAscii' a))+        => Sources a -> IO (Sources Word8)  +flatPackAsciiLn_i ss+        = concat_i =<< packAsciiLn_i ss+{-# INLINE_FLOW flatPackAsciiLn_i #-}+++---------------------------------------------------------------------------------------------------+-- | Like `packFormatLn_i`,+--   but use a default, human-readable format to encode the values.+keyPackAsciiLn_i+        :: forall a k+        . ( C.FormatAscii a, a ~ Value (C.FormatAscii' a)+          , Elem a+          , Packable (C.FormatAscii' a)+          , Elem k, Build k)+        => Sources (k, a)                       -- ^ Sources of values to be packed.+        -> IO (Sources (k, Array Word8))        -- ^ Packed data.++keyPackAsciiLn_i ss+ = let  +        proxy   = (error "repa-flow: sndPackAsciiLn_i proxy" :: a)+        !nl     = A.fromList [fromIntegral $ ord '\n']+        packElem (k, x)+         = let  Just arr = A.packFormat (C.formatAscii proxy) x+           in   (k, A.concat $ A.fromList [arr, nl])            -- TODO: avoid copy+        {-# INLINE packElem #-}++   in   G.map_i (A.map packElem) ss+{-# INLINE_FLOW keyPackAsciiLn_i #-}+

@@ -18,26 +18,20 @@         , sourceRecords         , sourceTSV         , sourceCSV-        , sourcePacked+        , sourceFormatLn            -- * Sinking         , sinkBytes         , sinkLines         , sinkChars-        , sinkPacked--          -- * Table IO-        , toTable-        , fromTable+        , sinkFormatLn         ) where import Data.Repa.Flow.Auto import Data.Repa.Flow.IO.Bucket import Data.Repa.Array.Material                 as A-import Data.Repa.Array.Auto.Unpack              as A+import Data.Repa.Array.Auto.Format              as A import Data.Repa.Array.Generic                  as A-import System.Directory-import System.FilePath import System.IO import Data.Word import qualified Data.Repa.Flow.Generic         as G@@ -135,49 +129,20 @@ {-# INLINE sourceTSV #-}  --- | Read packed binary data from some buckets and unpack the values---   to some `Sources`.------ The following uses the @colors.bin@ file produced by the `sinkPacked` example:------ @--- > import Data.Repa.Flow            as F--- > import Data.Repa.Convert.Format  as F--- > :{---   do let format = FixString ASCII 10 :*: Float64be :*: Int16be---      ss <- fromFiles' [\"colors.bin\"] $ sourcePacked format (error \"convert failed\")---      toList1 0 ss---   :}------ [\"red\" :*: (5.3 :*: 100), \"green\" :*: (2.8 :*: 93), \"blue\" :*: (0.99 :*: 42)]--- @----sourcePacked-        :: (Packable format, Target A (Value format))-        => format                       -- ^ Binary format for each value.-        -> IO ()                        -- ^ Action when a value cannot be converted.-        -> Array B Bucket               -- ^ Input buckets.+-- | Read the lines of a text file,+--   converting each line to values with the given format.+sourceFormatLn+        :: ( Unpackable format+           , Target A (Value format))+        => Integer                      -- ^ Chunk length.+        -> IO ()                        -- ^ Action when a line is too long.+        -> IO (Array A Word8 -> IO ())  -- ^ Action if we can't convert a value.+        -> format                       -- ^ Format of each line.+        -> Array B Bucket               -- ^ Source buckets.         -> IO (Sources (Value format)) -sourcePacked format aFail bs- = return $ G.Sources (A.length bs) pull_sourcePacked- where-        pull_sourcePacked i eat eject-         = do let b = A.index bs i-              op <- bIsOpen b-              if not op -               then eject-               else do-                eof <- bAtEnd b-                if eof-                 then eject-                 else do-                        !chunk  <- bGetArray b defaultChunkSize-                        case A.unpackForeign format (A.convert A chunk) of-                         Nothing        -> aFail-                         Just vals      -> eat vals-        {-# INLINE pull_sourcePacked #-}-{-# INLINE_FLOW sourcePacked #-}+sourceFormatLn = G.sourceLinesFormat+{-# INLINE sourceFormatLn #-}   ---------------------------------------------------------------------------------------------------@@ -201,8 +166,8 @@ {-# INLINE sinkLines #-}  --- | Create sinks that convert values to a packed binary format and writes---   them to some buckets.+-- | Create sinks that convert values to some format and writes+--   them to buckets. -- -- @ -- > import Data.Repa.Flow           as F@@ -215,36 +180,39 @@ --                    , \"blue\"  :*: 0.99   :*: 42 ] -- --      ss  <- F.fromList 1 vals---      out <- toFiles' [\"colors.bin\"] $ sinkPacked format (error \"convert failed\")+--      out <- toFiles' [\"colors.bin\"] +--          $  sinkFormatLn format (error \"convert failed\") --      drainS ss out --   :} -- @ ---sinkPacked -        :: (Packable format, Bulk A (Value format))+sinkFormatLn+        :: (Packable format, Bulk A (Value format), Show format)         => format                       -- ^ Binary format for each value.         -> IO ()                        -- ^ Action when a value cannot be serialized.         -> Array B Bucket               -- ^ Output buckets.         -> IO (Sinks (Value format)) -sinkPacked format aFail bs- = return $ G.Sinks (A.length bs) push_sinkPacked eject_sinkPacked+sinkFormatLn format aFail bs+ = return $ G.Sinks (A.length bs) +                    push_sinkFormatLn eject_sinkFormatLn  where  -        push_sinkPacked i !chunk-         = case A.packForeign format chunk of+        push_sinkFormatLn i !chunk+         = case A.packsFormatLn format chunk of                 Nothing   -> aFail                 Just buf  -> bPutArray (bs `index` i) (A.convert F buf)-        {-# INLINE push_sinkPacked #-}+        {-# INLINE push_sinkFormatLn #-} -        eject_sinkPacked i +        eject_sinkFormatLn i           = bClose (bs `index` i)-        {-# INLINE eject_sinkPacked #-}--{-# INLINE_FLOW sinkPacked #-}+        {-# INLINE eject_sinkFormatLn #-}+{-# INLINE_FLOW sinkFormatLn #-}   ------------------------------------------------------------------------------------------------------ | Create sinks that write values from some binary Repa table.+{-+-- | Create sinks that write values from some binary Repa table,+--   where all the values have a fixed length. toTable :: (Packable format, Bulk A (Value format))         => FilePath             -- ^ Directory holding table.         -> Int                  -- ^ Number of buckets to use.@@ -274,12 +242,13 @@         bs <- mapM newBucket names          -- Create a sink bundle for the buckets.-        kk <- sinkPacked format aFail (A.fromList B bs)+        kk <- sinkFormatLn format aFail (A.fromList B bs)         return $ Just kk {-# INLINE_FLOW toTable #-}  --- | Create sources that read values from some binary Repa table.+-- | Create sources that read values from some binary Repa table,+--   where all the values have a fixed length. fromTable         :: (Packable format, Target A (Value format))         => FilePath             -- ^ Directory holding table.@@ -308,7 +277,7 @@         bs <- mapM newBucket fsRel          -- Create a source bundle for the buckets.-        ss <- sourcePacked format aFail (A.fromList B bs)+        ss <- sourceFixedFormat format aFail (A.fromList B bs)         return $ Just ss {-# INLINE_FLOW fromTable #-}-+-}

@@ -0,0 +1,92 @@++module Data.Repa.Flow.Auto.Select+        ( select_i,     select_o+        , discard_i,    discard_o+        , mask_i,       mask_o)+where+import Data.Repa.Array                          as A+import Data.Repa.Flow.Auto.Base                 as F+import Data.Repa.Scalar.Singleton.Nat           as D+import Data.Repa.Scalar.Product                 as D+import qualified Data.Repa.Flow.Generic         as G+#include "repa-array.h"+++---------------------------------------------------------------------------------------------------+-- | Select a single column from a flow of rows of fields.+select_i +        :: ( Select  n (Array fs)+           , Select' n (Array fs) ~ Array (Select' n fs))+        => Nat n                        -- ^ Index of column to keep.+        -> Sources fs                   -- ^ Sources of complete rows.+        -> IO (Sources (Select' n fs))  -- ^ Sources of selected column.++select_i n ss+        = G.map_i (select n) ss+{-# INLINE_FLOW select_i #-}+++-- | Select a single column from a flow of fields.+select_o+        :: ( Select  n (Array fs)+           , Select' n (Array fs) ~ Array (Select' n fs))+        => Nat n                        -- ^ Index of column to keep.+        -> Sinks (Select' n fs)         -- ^ Sinks for selected column.+        -> IO (Sinks fs)                -- ^ Sinks for complete rows. ++select_o n ss+        = G.map_o (select n) ss+{-# INLINE_FLOW select_o #-}+        ++---------------------------------------------------------------------------------------------------+-- | Discard a single column from a flow of fields.+discard_i+        :: ( Discard  n (Array fs)+           , Discard' n (Array fs) ~ Array (Discard' n fs))+        => Nat n                        -- ^ Index of column to discard.+        -> Sources fs                   -- ^ Sources of complete rows.+        -> IO (Sources (Discard' n fs)) -- ^ Sources of partial  rows.++discard_i n ss+        = G.map_i (discard n) ss+{-# INLINE_FLOW discard_i #-}+++-- | Discard a single column from a flow of fields.+discard_o+        :: ( Discard  n (Array fs)+           , Discard' n (Array fs) ~ Array (Discard' n fs))+        => Nat n                        -- ^ Index of column to discard.+        -> Sinks (Discard' n fs)        -- ^ Sinks for partial rows.+        -> IO (Sinks fs)                -- ^ Sinks for complete rows.++discard_o n ss+        = G.map_o (discard n) ss+{-# INLINE_FLOW discard_o #-}+++---------------------------------------------------------------------------------------------------+-- | Mask columns from a flow of fields.+mask_i  :: ( Mask  ms (Array fs)+           , Mask' ms (Array fs) ~ Array (Mask' ms fs))+        => ms                           -- ^ Column mask.+        -> Sources fs                   -- ^ Sources of complete rows.+        -> IO (Sources (Mask' ms fs))   -- ^ Sources of masked rows.++mask_i ms ss+        = G.map_i (mask ms) ss+{-# INLINE_FLOW mask_i #-}+++-- | Mask columns from a flow of fields.+mask_o  :: ( Mask  ms (Array fs)+           , Mask' ms (Array fs) ~ Array (Mask' ms fs))+        => ms                           -- ^ Column mask.+        -> Sinks (Mask' ms fs)          -- ^ Sources of complete rows.+        -> IO (Sinks fs)                -- ^ Sources of masked rows.++mask_o ms ss+        = G.map_o (mask ms) ss+{-# INLINE_FLOW mask_o #-}+

@@ -0,0 +1,120 @@++module Data.Repa.Flow.Auto.ZipWith+        ( zipWith3_i+        , zipWith4_i+        , zipWith5_i+        , zipWith6_i+        , zipWith7_i)+where+import Data.Repa.Flow.Auto.Base+import Data.Repa.Array.Auto+import Data.Repa.Array.Material.Auto                    (Name(..))+import qualified Data.Repa.Flow.Chunked                 as C hiding (next)+#include "repa-flow.h"+++-- | Combine corresponding elements of three sources with the given function.+---+--   TODO: do this chunkwise instead. +--         The core code has probably had an unswitch explosion.+--+zipWith3_i +        :: ( Flow a, Flow b, Flow c+           , Build b, Build c, Build d)+        => (a -> b -> c -> d) +        -> Sources a -> Sources b -> Sources c+        -> IO (Sources d)+zipWith3_i f sa sb sc+        =   C.szipWith_ii A (\_ a (b, c) -> f a b c) sa +        =<< C.szipWith_ii A (\_ b c      -> (b, c))  sb sc+{-# INLINE zipWith3_i #-}+++-- | Combine corresponding elements of four sources with the given function.+---+--   TODO: do this chunkwise instead.+--         The core code has probably had an unswitch explosion.+--+zipWith4_i +        :: ( Flow  a, Flow  b, Flow  c, Flow  d+           , Build a, Build b, Build c, Build d, Build e)+        => (a -> b -> c -> d -> e) +        -> Sources a -> Sources b -> Sources c -> Sources d+        -> IO (Sources e)+zipWith4_i f sa sb sc sd+ = do   sab     <- C.szipWith_ii A (\_ a b -> (a, b)) sa sb+        scd     <- C.szipWith_ii A (\_ c d -> (c, d)) sc sd+        result  <- C.szipWith_ii A (\_ (a, b) (c, d) -> f a b c d) sab scd+        return result+{-# NOINLINE zipWith4_i #-}+--  NOINLINE due to unswitch explosion.+++-- | Combine corresponding elements of five sources with the given function.+---+--   TODO: do this chunkwise instead.+--         The core code has probably had an unswitch explosion.+--+zipWith5_i +        :: ( Flow  a, Flow  b, Flow  c, Flow  d, Flow  e+           , Build a, Build b, Build c, Build d, Build e+           , Build f)+        => (a -> b -> c -> d -> e -> f) +        -> Sources a -> Sources b -> Sources c -> Sources d -> Sources e+        -> IO (Sources f)+zipWith5_i f sa sb sc sd se+ = do   sab     <- C.szipWith_ii  A (\_ a b -> (a, b)) sa sb+        scd     <- C.szipWith_ii  A (\_ c d -> (c, d)) sc sd+        result  <- zipWith3_i       (\(a, b) (c, d) e -> f a b c d e) +                                    sab scd se+        return result+{-# NOINLINE zipWith5_i #-}+--  NOINLINE due to unswitch explosion.+++-- | Combine corresponding elements of six sources with the given function.+---+--   TODO: do this chunkwise instead.+--         The core code has probably had an unswitch explosion.+--+zipWith6_i +        :: ( Flow  a, Flow  b, Flow  c, Flow  d, Flow  e, Flow  f+           , Build a, Build b, Build c, Build d, Build e, Build f+           , Build g)+        => (a -> b -> c -> d -> e -> f -> g) +        -> Sources a -> Sources b -> Sources c -> Sources d -> Sources e -> Sources f+        -> IO (Sources g)+zipWith6_i ff sa sb sc sd se sf+ = do   sab     <- C.szipWith_ii  A (\_ a b -> (a, b)) sa sb+        scd     <- C.szipWith_ii  A (\_ c d -> (c, d)) sc sd+        result  <- zipWith4_i       (\(a, b) (c, d) e f -> ff a b c d e f) +                                    sab scd se sf+        return result+{-# NOINLINE zipWith6_i #-}+--  NOINLINE due to unswitch explosion.+++-- | Combine corresponding elements of seven sources with the given function.+---+--   TODO: do this chunkwise instead.+--         The core code has probably had an unswitch explosion.+--+zipWith7_i +        :: ( Flow  a, Flow  b, Flow  c, Flow  d, Flow  e, Flow  f, Flow  g+           , Build a, Build b, Build c, Build d, Build e, Build f, Build g+           , Build h)+        => (a -> b -> c -> d -> e -> f -> g -> h) +        -> Sources a -> Sources b -> Sources c -> Sources d -> Sources e -> Sources f -> Sources g+        -> IO (Sources h)++zipWith7_i ff sa sb sc sd se sf sg+ = do   sab     <- C.szipWith_ii  A (\_ a b -> (a, b)) sa sb+        scd     <- C.szipWith_ii  A (\_ c d -> (c, d)) sc sd+        result  <- zipWith5_i       (\(a, b) (c, d) e f g -> ff a b c d e f g) +                                    sab scd se sf sg+        return result+{-# NOINLINE zipWith7_i #-}+--  NOINLINE due to unswitch explosion.+++

@@ -7,6 +7,7 @@            -- * Evaluation         , drainS+        , consumeS            -- * Conversion         , fromList@@ -18,6 +19,9 @@         , finalize_i,   finalize_o            -- * Flow Operators+          -- ** Replicate+        , replicates_i+           -- ** Mapping           -- | If you want to work on a chunk at a time then use            --   `Data.Repa.Flow.Generic.map_i` and@@ -25,6 +29,13 @@         , smap_i,       smap_o         , szipWith_ii +          -- ** Processing+        , process_i++          -- ** Unfolding+        , unfolds_i+        , StepUnfold    (..)+           -- ** Splitting         , head_i @@ -40,22 +51,43 @@         , trigger_o            -- ** Ignorance-        , discard_o-        , ignore_o)+        , ignore_o+        , abandon_o) where import Data.Repa.Flow.Chunked.Base import Data.Repa.Flow.Chunked.Map import Data.Repa.Flow.Chunked.Fold import Data.Repa.Flow.Chunked.Folds+import Data.Repa.Flow.Chunked.Generic import Data.Repa.Flow.Chunked.Groups-import Data.Repa.Flow.Chunked.Operator+import Data.Repa.Flow.Chunked.Process+import Data.Repa.Flow.Chunked.Replicate import Data.Repa.Flow.States+import qualified Data.Repa.Array.Generic        as A import qualified Data.Repa.Flow.Generic         as G #include "repa-flow.h"   -- | Pull all available values from the sources and push them to the sinks. drainS   :: (Next i, Monad m)-        => Sources i m r a -> Sinks i m r a -> m ()+         => Sources i m r a -> Sinks i m r a -> m () drainS = G.drainS {-# INLINE drainS #-}+++-- | Pull all available values from the sources and pass them to the given action.+consumeS :: (Next i, Monad m, A.Bulk r a)+         => Sources i m r a -> (i -> a -> m ()) -> m ()++consumeS ss eatElem+ = G.consumeS eatChunk ss+ where  +        eatChunk ix arr+         = mapM_ (eatElem ix) $ A.toList arr+        {-# INLINE eatChunk #-}+{-# INLINE consumeS #-}+++++

@@ -16,7 +16,7 @@ import Control.Monad import qualified Data.Repa.Flow.Generic         as G import qualified Data.Sequence                  as Q-import qualified Data.Foldable                  as Q+import qualified Data.Foldable                  as D import Prelude                                  as P #include "repa-flow.h" @@ -127,13 +127,13 @@          -- As we've pulled whole chunks from the input stream,         -- we now prepend the remaining ones back on.-        let start  =  Q.length has - Q.length rest+        let start  = Q.length has - Q.length rest         let stash  = case mFinal of                         Nothing -> []                         Just c  -> [A.window start (Q.length rest) c]          s2'        <- G.prependOn_i (\i' -> i' == i) stash s2-        return  (Q.toList here, s2')+        return  (D.toList here, s2') {-# INLINE_FLOW head_i #-}  

@@ -6,7 +6,6 @@ import Data.Repa.Flow.Chunked.Base import Data.Repa.Flow.States                    as S import qualified Data.Repa.Flow.Generic         as G-import qualified Data.Repa.Array.Generic.Target as A import qualified Data.Repa.Array.Generic.Index  as A import qualified Data.Repa.Array.Generic        as A #include "repa-flow.h"

@@ -5,10 +5,8 @@ where import Data.Repa.Flow.Chunked.Base import Data.Repa.Flow.States-import Data.Repa.Fusion.Unpack-import Data.Repa.Option+import Data.Repa.Scalar.Option import Data.Repa.Array.Generic.Index            as A-import Data.Repa.Array.Generic.Target           as A import Data.Repa.Array.Generic                  as A hiding (FoldsDict) import Data.Repa.Array.Meta.Window              as A import Data.Repa.Array.Meta.Tuple               as A@@ -17,7 +15,7 @@   -- | Dictionaries needed to perform a segmented fold.-type FoldsDict i m lSeg tSeg lElt tElt lGrp tGrp lRes tRes n a b+type FoldsDict i m lSeg lElt lGrp lRes n a b         = ( States i m           , Windowable lSeg (n, Int), Windowable lElt a           , BulkI   lSeg (n, Int)@@ -26,14 +24,12 @@           , BulkI   lRes b           , TargetI lElt a           , TargetI lGrp n-          , TargetI lRes b-          , Unpack (Buffer lGrp n) tGrp-          , Unpack (Buffer lRes b) tRes)+          , TargetI lRes b)   -- Folds ---------------------------------------------------------------------- -- | Segmented fold over vectors of segment lengths and input values.-folds_i :: FoldsDict i m lSeg tSeg lElt tElt lGrp tGrp lRes tRes n a b+folds_i :: FoldsDict i m lSeg lElt lGrp lRes n a b         => Name lGrp                 -- ^ Layout for group names.         -> Name lRes                 -- ^ Layout for fold results.         -> (a -> b -> b)             -- ^ Worker function.

@@ -0,0 +1,73 @@++-- | Operators for chunked flows.+--+--   Functions in this module are re-exports of the ones from+--   "Data.Repa.Flow.Generic", but using the `Sources` and `Sinks`+--   type synonyms for chunked flows.+--+module Data.Repa.Flow.Chunked.Generic+        ( -- * Watching+          watch_i,      watch_o+        , trigger_o++          -- * Ignorance+        , ignore_o+        , abandon_o)+where+import Data.Repa.Flow.Chunked.Base+import Data.Repa.Array.Generic                  as A+import qualified Data.Repa.Flow.Generic         as G+#include "repa-flow.h"+++-- Watch ----------------------------------------------------------------------+-- | Hook a monadic function to some sources, which will be passed every+--   chunk that is pulled from the result.+watch_i :: Monad m+        => (i -> Array l a -> m ()) +        -> Sources i m l a  -> m (Sources i m l a)+watch_i = G.watch_i+{-# INLINE watch_i #-}+++-- | Hook a monadic function to some sinks, which will be passed every +--   chunk that is pushed to the result.+watch_o :: Monad m+        => (i -> Array l a -> m ())+        -> Sinks i m l a ->  m (Sinks i m l a)++watch_o = G.watch_o+{-# INLINE watch_o #-}+++-- | Like `watch_o` but discard the incoming chunks after they are passed+--   to the function.+trigger_o :: Monad m+          => i -> (i -> Array l a -> m ()) -> m (Sinks i m l a)+trigger_o = G.trigger_o+{-# INLINE trigger_o #-}+++-- Ignorance ------------------------------------------------------------------+-- | A sink that ignores all incoming data.+--+--   * The sinks is strict in the *chunks*, so they are demanded before being+--     discarded. Haskell debugging thunks attached to the chunks will be+--     demanded, but thunks attached to elements may not be -- depending on+--     whether the chunk representation is strict in the elements.+--+ignore_o :: Monad m => i -> m (Sinks i m l a)+ignore_o  = G.ignore_o+{-# INLINE ignore_o #-}+++-- | Yield a bundle of sinks of the given arity that drops all data on the+--   floor.+--+--   This sink is non-strict in the chunks. +--   Haskell tracing thunks attached to the chunks will *not* be demanded.+--+abandon_o :: Monad m => i -> m (Sinks i m l a)+abandon_o = G.abandon_o+{-# INLINE abandon_o #-}+

@@ -5,22 +5,18 @@ where import Data.Repa.Flow.Chunked.Base import Data.Repa.Flow.States-import Data.Repa.Fusion.Unpack import Data.Repa.Array.Meta.Tuple               as A import Data.Repa.Array.Generic.Index            as A-import Data.Repa.Array.Generic.Target           as A import Data.Repa.Array.Generic                  as A  hiding (GroupsDict) import qualified Data.Repa.Flow.Generic         as G #include "repa-flow.h"   -- | Dictionaries needed to perform a grouping.-type GroupsDict  i m lVal lGrp tGrp lLen tLen a+type GroupsDict  i m lVal lGrp lLen a         = ( Flow i m lVal a, A.Index lVal ~ Int           , TargetI  lGrp  a-          , TargetI  lLen Int-          , Unpack  (Buffer lGrp a)   tGrp-          , Unpack  (Buffer lLen Int) tLen)+          , TargetI  lLen Int)   -- Grouping -------------------------------------------------------------------@@ -33,7 +29,7 @@ -- @ --  groupsBy_i -        :: GroupsDict i m lVal lGrp tGrp lLen tLen a+        :: GroupsDict i m lVal lGrp lLen a         => Name lGrp             -- ^ Layout for group names.         -> Name lLen             -- ^ Layout for group lengths.         -> (a -> a -> Bool)      -- ^ Whether successive elements should be grouped.

@@ -1,73 +0,0 @@---- | Operators for chunked flows.------   Most functions in this module are re-exports of the ones from---   "Data.Repa.Flow.Generic.IO", but using the `Sources` and `Sinks`---   type synonyms for chunked flows.----module Data.Repa.Flow.Chunked.Operator-        ( -- * Watching-          watch_i,      watch_o-        , trigger_o--          -- * Ignorance-        , discard_o-        , ignore_o)-where-import Data.Repa.Flow.Chunked.Base-import Data.Repa.Array.Generic                  as A-import qualified Data.Repa.Flow.Generic         as G-#include "repa-flow.h"----- Watch ------------------------------------------------------------------------- | Hook a monadic function to some sources, which will be passed every---   chunk that is pulled from the result.-watch_i :: Monad m-        => (i -> Array l a -> m ()) -        -> Sources i m l a  -> m (Sources i m l a)-watch_i = G.watch_i-{-# INLINE watch_i #-}----- | Hook a monadic function to some sinks, which will be passed every ---   chunk that is pushed to the result.-watch_o :: Monad m-        => (i -> Array l a -> m ())-        -> Sinks i m l a ->  m (Sinks i m l a)--watch_o = G.watch_o-{-# INLINE watch_o #-}----- | Like `watch_o` but discard the incoming chunks after they are passed---   to the function.-trigger_o :: Monad m-          => i -> (i -> Array l a -> m ()) -> m (Sinks i m l a)-trigger_o = G.trigger_o-{-# INLINE trigger_o #-}----- Ignorance --------------------------------------------------------------------- | A sink that ignores all incoming data.------   This sink is non-strict in the chunks. ---   Haskell tracing thunks attached to the chunks will *not* be demanded.----ignore_o :: Monad m => i -> m (Sinks i m l a)-ignore_o  = G.ignore_o-{-# INLINE ignore_o #-}----- | Yield a bundle of sinks of the given arity that drops all data on the---   floor.------   * The sinks is strict in the *chunks*, so they are demanded before being---     discarded. Haskell debugging thunks attached to the chunks will be---     demanded, but thunks attached to elements may not be -- depending on---     whether the chunk representation is strict in the elements.----discard_o :: Monad m => i -> m (Sinks i m l a)-discard_o = G.discard_o-{-# INLINE discard_o #-}-

@@ -0,0 +1,81 @@++module Data.Repa.Flow.Chunked.Process+        ( process_i +        , unfolds_i,    A.StepUnfold(..))+where+import Data.Repa.Flow.Chunked.Base+import qualified Data.Repa.Array.Generic        as A+import qualified Data.Repa.Array.Generic.Index  as A+import qualified Data.Repa.Flow.Generic.Base    as G+#include "repa-flow.h"+++-- | Apply a generic stream process to all the streams in a bundle of sources.+process_i+        :: ( G.States i m+           , A.BulkI   lSrc a+           , A.Bulk    lDst b, A.Bulk    lDst (A.Array lDst b)+           , A.TargetI lDst b, A.TargetI lDst (A.Array lDst b))+        => (s -> a -> (s, A.Array lDst b))      -- ^ Worker function.+        -> s                                    -- ^ Initial state.+        ->    Sources i m lSrc a                -- ^ Input sources+        -> m (Sources i m lDst b)               -- ^ Result sources.++process_i f z (G.Sources n pullA)+ = do+        refs    <- G.newRefs n z++        let pull_process i eatB ejectB+             = do s1 <- G.readRefs refs i+                  pullA i (eatA_process s1) ejectA_process++             where eatA_process s1 xA+                    = case A.process A.name f s1 xA of+                        (s1', arrB) +                         -> do  G.writeRefs refs i s1'+                                eatB arrB+                   {-# INLINE eatA_process #-}++                   ejectA_process +                    = ejectB+                   {-# INLINE ejectA_process #-}+            {-# INLINE pull_process #-}++        return $ G.Sources n pull_process+{-# INLINE_FLOW process_i #-}+++-- | Apply a generic stream process to all the streams in a bundle of sources.+unfolds_i+        :: ( G.States i m+           , A.BulkI   lSrc a+           , A.Bulk    lDst b+           , A.TargetI lDst b)+        => (a -> s -> A.StepUnfold s b)         -- ^ Worker function.+        -> s                                    -- ^ Initial state.+        ->    Sources i m lSrc a                -- ^ Input sources+        -> m (Sources i m lDst b)               -- ^ Result sources.++unfolds_i f z (G.Sources n pullA)+ = do+        refs    <- G.newRefs n z++        let pull_unfolds i eatB ejectB+             = do s1 <- G.readRefs refs i+                  pullA i (eatA_unfolds s1) ejectA_unfolds++             where eatA_unfolds s1 xA+                    = case A.unfolds A.name f s1 xA of+                        (s1', arrB) +                         -> do  G.writeRefs refs i s1'+                                eatB arrB+                   {-# INLINE eatA_unfolds #-}++                   ejectA_unfolds+                    = ejectB+                   {-# INLINE ejectA_unfolds #-}+            {-# INLINE pull_unfolds #-}++        return $ G.Sources n pull_unfolds+{-# INLINE_FLOW unfolds_i #-}+

@@ -0,0 +1,19 @@++module Data.Repa.Flow.Chunked.Replicate+        ( replicates_i )+where+import Data.Repa.Flow.Chunked.Base+import qualified Data.Repa.Flow.Generic         as G+import qualified Data.Repa.Array.Generic        as A+++replicates_i +        :: ( Flow i m lSrc (Int, a)+           , A.TargetI lDst a)+        => A.Name lDst+        -> Sources    i m lSrc (Int, a)+        -> m (Sources i m lDst a)++replicates_i nDst ss+        = G.smap_i (\_ arr -> A.replicates nDst arr) ss+{-# INLINE replicates_i #-}

@@ -18,6 +18,7 @@           -- * Evaluation         , drainS         , drainP+        , consumeS            -- * Conversion         , fromList@@ -52,6 +53,11 @@         , smap_i,       smap_o         , szipWith_ii,  szipWith_io,    szipWith_oi +          -- ** Processing+        , compact_i+        , scan_i+        , indexed_i+           -- ** Connecting         , dup_oo,       dup_io,         dup_oi         , connect_i@@ -80,8 +86,8 @@         , rcapture_o            -- ** Ignorance-        , discard_o         , ignore_o+        , abandon_o            -- ** Tracing         , trace_o@@ -101,7 +107,7 @@         , dshuffleBy_o            -- ** Chunking-        , chunk_i+        , chunkOn_i         , unchunk_i) where import Data.Repa.Flow.States@@ -109,6 +115,7 @@ import Data.Repa.Flow.Generic.Connect import Data.Repa.Flow.Generic.List import Data.Repa.Flow.Generic.Map+import Data.Repa.Flow.Generic.Process import Data.Repa.Flow.Generic.Operator import Data.Repa.Flow.Generic.Eval import Data.Repa.Flow.Generic.Array.Distribute

@@ -1,27 +1,29 @@  module Data.Repa.Flow.Generic.Array.Chunk-        (chunk_i)+        (chunkOn_i) where import Data.Repa.Flow.Generic.Base import Data.Repa.Array.Generic                  as A-import Data.Repa.Array.Generic.Target           as A import Data.Repa.Array.Generic.Index            as A+import Data.Repa.Array.Generic.Target           as A #include "repa-flow.h"  --- | Take elements from a flow and pack them into chunks of the given---   maximum length.-chunk_i :: (Target lDst a, Index lDst ~ Int, States i IO)+-- | Take elements from a flow and pack them into chunks.+--   The chunks are limited to the given maximum length.+--   A predicate can also be supplied to detect the last element in a chunk.+chunkOn_i +        :: (States i IO, TargetI lDst a)         => Name lDst                            -- ^ Layout for result chunks.         -> Int                                  -- ^ Maximum chunk length.+        -> (a -> Bool)                          -- ^ Detect the last element in a chunk.         -> Sources i IO a                       -- ^ Element sources.         -> IO (Sources i IO (Array lDst a))     -- ^ Chunk sources. -chunk_i nDst !maxLen (Sources n pullX)+chunkOn_i nDst !maxLen isEnd (Sources n pullX)  = do-        -- Refs for signalling how many elements we managed to read for-        -- each chunk.-        final  <- newRefs n Nothing+        -- Refs for signalling how many elements we managed to read for each chunk.+        final   <- newRefs n Nothing          let pull_chunk i eat eject              = do @@ -39,7 +41,12 @@                                 -- Write the next element to the chunk.                                 eat_chunk x                                  = do   unsafeWriteBuffer chunk ix x-                                        loop_chunk (ix + 1)++                                        -- Check if this element value signals the +                                        -- end of a chunk.+                                        if isEnd x +                                         then writeRefs final i (Just $ ix + 1)+                                         else loop_chunk (ix + 1)                                          -- There are no more elements available from the soruce.                                 eject_chunk@@ -70,5 +77,5 @@             {-# INLINE pull_chunk #-}          return $ Sources n pull_chunk-{-# INLINE_FLOW chunk_i #-}+{-# INLINE_FLOW chunkOn_i #-} 

@@ -1,7 +1,8 @@  module Data.Repa.Flow.Generic.Eval         ( drainS-        , drainP)+        , drainP+        , consumeS) where import Data.Repa.Flow.Generic.Base import Data.Repa.Eval.Gang                      as Eval@@ -70,3 +71,30 @@                 {-# INLINE eject_drain #-}         {-# INLINE drainMe #-} {-# INLINE_FLOW drainP #-}+++-- | Pull all available values from the sources and pass them to the+--   given action.+consumeS :: (Next i, Monad m)+         => (i -> a -> m ())+         -> Sources i m a +         -> m ()+consumeS eat (Sources nSources ipull)+ = loop_consume first+ where  +        loop_consume !ix+         = ipull ix eat_consume eject_consume+         where  +                eat_consume v+                 = do eat ix v+                      loop_consume ix+                {-# INLINE eat_consume #-}++                eject_consume +                 = do case next ix nSources of+                        Nothing  -> return ()+                        Just ix' -> loop_consume ix'+                {-# INLINE eject_consume #-}+        {-# INLINE loop_consume #-}+{-# INLINE_FLOW consumeS #-}+

@@ -8,7 +8,10 @@         , sourceChars         , sourceChunks         , sourceRecords+        , sourceLinesFormat +        , sourceLinesFormatFromLazyByteString+           -- * Sinking         , sinkBytes         , sinkChars@@ -24,3 +27,4 @@ import Data.Repa.Flow.IO.Bucket import Data.Repa.Flow.Generic.IO.Base           as F import Data.Repa.Flow.Generic.IO.XSV            as F+import Data.Repa.Flow.Generic.IO.Lines          as F

@@ -21,7 +21,6 @@ import Data.Repa.Flow.Generic.IO.Sieve          as F import Data.Repa.Flow.Generic.Map               as F import Data.Repa.Flow.Generic.Base              as F-import Data.Repa.Fusion.Unpack                  as F import Data.Repa.Array.Material                 as A import Data.Repa.Array.Meta.Delayed             as A import Data.Repa.Array.Meta.Window              as A@@ -163,7 +162,7 @@ sinkLines          :: ( Bulk  l Bucket            , BulkI l1 (Array l2 Char)-           , BulkI l2 Char, Unpack (Array l2 Char) t2)+           , BulkI l2 Char)         => Name  l1             -- ^ Layout of chunks of lines.         -> Name  l2             -- ^ Layout of lines.         -> Array l Bucket       -- ^ Buckets.

@@ -0,0 +1,126 @@++module Data.Repa.Flow.Generic.IO.Lines+        ( sourceLinesFormat+        , sourceLinesFormatFromLazyByteString)+where+import Data.Repa.Flow.Generic.IO.Base           as F+import Data.Repa.Flow.Generic.Map               as F+import Data.Repa.Flow.Generic.Base              as F++import Data.Repa.Array.Generic                  as A+import Data.Repa.Array.Material                 as A+import qualified Data.Repa.Array.Auto.Format    as A++import Data.Repa.Convert.Format                 as C++import qualified Data.ByteString.Lazy           as BSL+import Data.Char+import Data.Word+#include "repa-flow.h"+++-- | Read lines from a named text file,+--   in a chunk-wise manner,+--   converting each line to values with the given format.+sourceLinesFormat+        :: forall format+        .  (Unpackable format, Target A (Value format))+        => Integer                      -- ^ Chunk length.+        -> IO ()                        -- ^ Action if we find a line longer than the chunk length.+        -> IO (Array A Word8 -> IO ())  -- ^ Action if we can't convert a row.+        -> format                       -- ^ Format of each line.+        -> Array B Bucket+        -> IO (Sources Int IO (Array A (Value format)))++sourceLinesFormat nChunk aFailLong _aFailConvert format bs+ = do+        -- Rows are separated by new lines.+        let !nl  = fromIntegral $ ord '\n'+        let !nr  = fromIntegral $ ord '\r'++        -- Stream chunks of data from the input file, +        -- where the chunks end cleanly and line boundaries.+        -- Filter out any stray CR characters allong the way.+        sChunk  <- sourceChunks nChunk (== nl) aFailLong bs++        sRows8  :: Sources Int IO (Array N (Array F Word8))+                <- map_i ( A.trimEnds  (== nl)+                         . A.segmentOn (== nl) +                         . A.filter F  (/= nr)) +                         sChunk++        -- Convert each value using the given format.+        let unpackRow :: Array A Word8 -> Value format+            unpackRow arr+             = case A.unpackFormat format arr of+                 Nothing -> error ("no convert " ++ show arr)+                        -- TODO: impl proper pull function+                        -- so we can call aFailConvert if needed.+                        -- We shouldn't be throwing errors this deep in the library.+                 Just v  -> v+            {-# INLINE unpackRow #-}++        F.map_i (A.mapS A (unpackRow . A.convert A)) sRows8+{-# INLINE sourceLinesFormat #-}+++-- | Read lines from a lazy byte string,+--   in a chunk-wise manner,+--   converting each line to values with the given format.+sourceLinesFormatFromLazyByteString+        :: (Unpackable format, Target A (Value format))+        => Int                          -- ^ Number of streams in the result bundle.+        -> IO (Array A Word -> IO ())   -- ^ Action if we can't convert a row.+        -> format                       -- ^ Format of each line.+        -> BSL.ByteString               -- ^ Lazy byte string.+        -> Int                          -- ^ Skip this many header lines at the start.+        -> IO (Sources Int IO (Array A (Value format)))++sourceLinesFormatFromLazyByteString n _aFailConvert format bs0 nSkip+ = do+        -- Rows are separated by new lines.+        let !nl  = fromIntegral $ ord '\n'+        let !nr  = fromIntegral $ ord '\r'++        -- Give a copy of the bytestring to each stream.+        refsBS          <- newRefs n bs0+        refsSkip        <- newRefs n nSkip++        let unpackRow arr+             = case A.unpackFormat format arr of+                 Nothing -> error ("no convert " +                                        ++ (show $ map (chr . fromIntegral) +                                                 $ A.toList arr))+                        -- TODO: imlp proper pull function+                        -- so that we can call aFailConvert if needed.+                        -- We shouldn't be throwing errors this deep in the library.+                 Just v  -> v+            {-# INLINE unpackRow #-}++        let pull_fromString i eat eject+             = do bs   <- readRefs refsBS   i+                  skip <- readRefs refsSkip i+                  if BSL.null bs+                   then eject+                   else do let (bsLine, bsRest) +                                = BSL.break (== nl) bs++                           writeRefs refsBS i +                            $ BSL.dropWhile (== nl) bsRest++                           if (skip >= 0)+                             then do writeRefs refsSkip i (skip - 1)+                                     pull_fromString i eat eject++                             else eat $ A.singleton A.A+                                      $ unpackRow+                                      $ A.convert A.A+                                      $ A.fromByteString +                                      $ BSL.toStrict +                                      $ BSL.filter (/= nr) bsLine+            {-# INLINE pull_fromString #-}++        return $ Sources n pull_fromString+{-# INLINE_FLOW sourceLinesFormatFromLazyByteString #-}++

@@ -4,47 +4,131 @@ where import Data.Repa.Flow.Generic.Base import Data.Repa.Array.Material         as A-import Data.Repa.Array.Generic.Convert  as A+import Data.Repa.Array.Generic          as A import Data.Repa.Array.Auto.IO          as A+import qualified Data.HashTable.IO      as Hash+import qualified System.Mem             as System import System.IO import Data.Word+import Data.IORef++import qualified Data.Vector.Mutable    as M+import qualified Data.Vector            as V+ #include "repa-flow.h"   -- | Create an output sieve that writes data to an indeterminate number of --   output files. Each new element is appended to its associated file. -----   * TODO: ---     This function keeps a maximum of 8 files open at once, closing---     and re-opening them in a least-recently-used order.---     Due to this behaviour it's fine to create thousands of separate---     output files without risking overflowing the process limit on ---     the maximum number of useable file handles.+---+--   We don't want to open and close a file every time we receieve data.+--   To avoid this, we instead batch data in memory for each file until+--   we have enough to warrant performing the IO operation. ---sieve_o :: (a -> Maybe (FilePath, Array F Word8))   +sieve_o :: Int                  -- ^ Max payload size of in-memory data.+        -> Int                  -- ^ Max number of in-memory chunks.+        -> (a -> Maybe (FilePath, Array F Word8))                                    -- ^ Produce the desired file path and output                                 --   record for this element, or `Nothing` if                                 --   it should be discarded.         -> IO (Sinks () IO a) -sieve_o diag+sieve_o sizeLimit chunksLimit diag  = do+        -- Store an array of chunks for each file.+        -- We use a mutable vector of chunks, and store the number of used +        -- slots in that vector separately.+        (ht :: Hash.CuckooHashTable FilePath (Int, M.IOVector (Array F Word8)))+         <- Hash.newSized 1024 -        let push_sieve _ e+        !refSize   <- newIORef 0+        !refChunks <- newIORef 0++        -- Flush the chunks for a single file to disk.+        let flush_path (path, (n, mvec))+             = do +                  !vec   <- V.unsafeFreeze mvec++                  !h     <- openBinaryFile path AppendMode ++                  -- Write out chunks for this file.+                  V.mapM_ (hPutArray h . convert A) +                        $ V.slice 0 n vec++                  hClose h++                  -- Delete the entry from the hash table.+                  -- This allows the space for the mutable vector of chunks to be reclaimed.+                  Hash.delete ht path+++        -- Flush all the chunks we have stored.+        let flush_all +             = do Hash.mapM_ flush_path ht++        -- Remember that we've accumulated this chunk into memory.+        -- When we end up with too much data then we flush the whole lot+        -- to the file system.+        let acc_size !len+             = do !sizeCurrent    <- readIORef refSize+                  !chunksCurrent  <- readIORef refChunks++                  if  (sizeCurrent   + len) > sizeLimit+                   || (chunksCurrent + 1)   > chunksLimit+                   then do+                        flush_all+                        writeIORef refSize   0+                        writeIORef refChunks 0++                   else do+                        let !sizeCurrent'   = sizeCurrent + len+                        let !chunksCurrent' = chunksCurrent + 1+                        writeIORef refSize   sizeCurrent'+                        writeIORef refChunks chunksCurrent'++        -- Accept a single incoming element.+        let push_sieve _ !e+              = case diag e of++                -- The provided diag function told us to drop this+                -- element on the floor.                 Nothing                   -> return () +                -- Accumulate a new chunk.                 Just (path, arr)-                 -> do  -- TODO: repeatededly opening and closing the file -                        --       will be very slow.-                        h       <- openBinaryFile path AppendMode-                        hPutArray h (convert arr)-                        hClose h+                 -> do  +                        -- See if we already have a buffer for this file.+                        !mElem   <- Hash.lookup ht path+                        case mElem of -        -- TODO: ignore any more incoming data after being ejected.+                         -- We haven't seen chunks for this file before, +                         -- so create a new vector to hold them.+                         Nothing+                          -> do !mvec    <- M.new 256+                                M.write mvec 0 arr+                                Hash.insert ht path (1, mvec)+                                acc_size (A.length arr)++                         -- We already have a chunk vector for this file.+                         Just (n, mvec)+                          -> do +                                -- If the chunk vector has no space the expand it.+                                !mvec'  <- if n >= M.length mvec+                                                then M.grow mvec (M.length mvec)+                                                else return mvec++                                M.write mvec' n arr+                                let !n' = n + 1+                                Hash.insert ht path (n', mvec')+                                acc_size (A.length arr)+         let eject_sieve _ -             = return ()+             = do flush_all+                  System.performMajorGC          return  $ Sinks () push_sieve eject_sieve {-# INLINE sieve_o #-}+

@@ -34,8 +34,8 @@         , rcapture_o            -- * Ignorance-        , discard_o         , ignore_o+        , abandon_o            -- * Tracing         , trace_o)@@ -352,7 +352,7 @@              = atomicModifyIORef ref (\old -> ((i, x) : old, ()))             {-# INLINE capture_eat #-} -        k0       <- discard_o n+        k0       <- ignore_o n         k1       <- watch_o   capture_eat k0          x        <- use k1@@ -367,43 +367,43 @@ trigger_o :: Monad m            => i -> (i -> a -> m ()) -> m (Sinks i m a) trigger_o i f- = discard_o i >>= watch_o f+ = ignore_o i >>= watch_o f {-# INLINE trigger_o #-}   -- Ignorance---------------------------------------------------------------------- | A sink that drops all data on the floor.+-- | A sink that ignores all incoming data. -----   This sink is strict in the elements, so they are demanded before being---   discarded. Haskell debugging thunks attached to the elements will be---   demanded.+--   * This sink is strict in the elements, so they are demanded before being+--     discarded. +--   * Haskell debugging thunks attached to the elements will be+--     demanded. ---discard_o :: Monad m +ignore_o  :: Monad m            => i -> m (Sinks i m a)-discard_o n- = return $ Sinks n push_discard eject_discard+ignore_o n+ = return $ Sinks n push_ignore eject_ignore  where  -        -- IMPORTANT: push_discard should be strict in the element so that+        -- IMPORTANT: push_ignore should be strict in the element so that         -- and Haskell tracing thunks attached to it are evaluated.-        -- We *discard* the elements, but don't completely ignore them.-        push_discard  !_ !_ = return ()-        eject_discard !_    = return ()-{-# INLINE_FLOW discard_o #-}+        push_ignore  !_ !_ = return ()+        eject_ignore !_    = return ()+{-# INLINE_FLOW ignore_o #-}  --- | A sink that ignores all incoming data.+-- | A sink that drops all data on the floor. -----   This sink is non-strict in the elements. ---   Haskell tracing thunks attached to the elements will *not* be demanded.+--   * This sink is non-strict in the elements. +--   * Haskell tracing thunks attached to the elements will *not* be demanded. ---ignore_o  :: Monad m +abandon_o :: Monad m            => i -> m (Sinks i m a)-ignore_o n- = return $ Sinks n push_ignore eject_ignore+abandon_o n+ = return $ Sinks n push_abandon eject_abandon  where-        push_ignore  _ _   = return ()-        eject_ignore _     = return ()-{-# INLINE_FLOW ignore_o #-}+        push_abandon  _ _   = return ()+        eject_abandon _     = return ()+{-# INLINE_FLOW abandon_o #-}   -- Trace ----------------------------------------------------------------------@@ -424,6 +424,4 @@                  (return ())  {-# NOINLINE trace_o #-}-- 

@@ -0,0 +1,87 @@++module Data.Repa.Flow.Generic.Process+        ( -- * Processing+          compact_i++          -- * Scanning+        , scan_i++          -- * Indexed streams+        , indexed_i)+where+import Data.Repa.Flow.Generic.Base+++-- | Combination of 'fold' and 'filter'.+--+--   We walk over the stream start-to-end, maintaining an accumulator.+--   At each point we can chose to emit an element, or not.+--+compact_i+        :: (Monad m, States i m)+        => (s -> a -> (s, Maybe b))+        -> s+        -> Sources i m a +        -> m (Sources i m b)++compact_i f s0 (Sources n pullA)+ = do+        refs    <- newRefs n s0++        let pull_process i eatB ejectB+             = do s1    <- readRefs refs i+                  pullA i (eatA_process s1) ejectA_process++             where eatA_process s1 xA+                    = do case f s1 xA of+                          (s2', Nothing) +                           -> do writeRefs refs i s2'+                                 pull_process i eatB ejectB++                          (s2', Just xB)+                           -> do writeRefs refs i s2'+                                 eatB xB+                   {-# INLINE eatA_process #-}++                   ejectA_process = ejectB+                   {-# INLINE ejectA_process #-}+            {-# INLINE pull_process #-}++        return $ Sources n pull_process+{-# INLINE compact_i #-}+++-- | Start-to-end scan over each stream in a bundle.+scan_i +        :: (Monad m, States i m)+        => (s -> a -> s)+        -> s+        -> Sources i m a+        -> m (Sources i m s)++scan_i f s0 ss+ = compact_i work_scan s0 ss+ where+        work_scan s x +         = let s' = f s x+           in  (s', Just s')+        {-# INLINE work_scan #-}+{-# INLINE scan_i #-}+++-- | For each stream in a bundle of sources, +--   associated the element with their corresponding position in the stream.+-- +indexed_i+        :: (Monad m, States i m)+        => Sources i m a+        -> m (Sources i m (Int, a))++indexed_i ss+ = compact_i work_indexed 0 ss+ where+        work_indexed s x+         = (s + 1, Just (s, x))+        {-# INLINE work_indexed #-}+{-# INLINE indexed_i #-}+

@@ -1,38 +0,0 @@--module Data.Repa.Flow.IO.Binary-        ( sourceBinary )-where-import Data.Repa.Array.Material         as A-import Data.Repa.Array.Generic          as A-import Data.Repa.Flow.IO.Storable       as F-import Data.Repa.Flow.IO.Bucket         as F-import Data.Repa.Flow.Generic           as G-#include "repa-flow.h"----- Move to Data.Repa.Flow.Binary-sourceBinary -        :: F.Storable a-        => Spec a                       -- ^ Specification of elements.-        -> Integer                      -- ^ Number of elements per chunk.-        -> Array B Bucket               -- ^ Buckets of table.-        -> IO (G.Sources Int IO (Array (Rep a) a))--sourceBinary spec lenElems bs- = return $ G.Sources (A.length bs) pull_sourceBinary- where-        pull_sourceBinary i eat eject-         = do   let b   = A.index bs i-                op      <- bIsOpen b-                if not op-                 then eject-                 else do-                  eof   <- bAtEnd b-                  if eof-                   then eject-                   else do-                        Just !chunk <- getArray spec lenElems b-                        eat chunk-        {-# INLINE pull_sourceBinary #-}-{-# INLINE_FLOW sourceBinary #-}-

@@ -13,7 +13,7 @@           -- * Writing         , toFiles,              toFiles'         , toDir-        ,                       toDirs'+        , toDirs            -- * Bucket IO         , bClose@@ -86,12 +86,13 @@   -- | Wrap an existing file handle as a bucket.+--+--   The starting position is set to 0. hBucket :: Handle -> IO Bucket hBucket h- = do   pos     <- hTell h-        return  $ Bucket+ =      return  $ Bucket                 { bucketFilePath        = Nothing-                , bucketStartPos        = pos+                , bucketStartPos        = 0                 , bucketLength          = Nothing                 , bucketHandle          = h } {-# NOINLINE hBucket #-}@@ -99,14 +100,24 @@  -- From Files ----------------------------------------------------------------- -- | Open some files as buckets and use them as `Sources`.-fromFiles +fromFiles+        :: [FilePath]+        -> (Array B Bucket -> IO b)+        -> IO b+fromFiles files use + = fromFiles' (A.fromList B files) use+{-# INLINE fromFiles #-}+++-- | Like `fromFiles'`, but take a list of file paths.+fromFiles'          ::  (Bulk l FilePath, Target l Bucket)         =>  Array l FilePath                    -- ^ Files to open.         -> (Array l Bucket -> IO b)                                                   -- ^ Consumer.         -> IO b -fromFiles paths use+fromFiles' paths use  = do            -- Open all the files, ending up with a list of buckets.         bs      <- mapM (flip openBucket ReadMode) $ A.toList paths@@ -116,19 +127,10 @@         let Just bsArr =  A.fromListInto (A.layout paths) bs          use bsArr-{-# NOINLINE fromFiles #-}----- | Like `fromFiles`, but take a list of file paths.-fromFiles'-        :: [FilePath]-        -> (Array B Bucket -> IO b)-        -> IO b-fromFiles' files use - = fromFiles (A.fromList B files) use-{-# INLINE fromFiles' #-}+{-# NOINLINE fromFiles' #-}  +-- From Dirs ------------------------------------------------------------------ -- | Open all the files in a directory as separate buckets. -- --   This operation may fail with the same exceptions as `getDirectoryContents`.@@ -142,7 +144,7 @@         let fsRel                        =  P.map (dir </>)                  $  P.filter (\f -> f /= "." && f /= "..") fs-        fromFiles' fsRel use+        fromFiles fsRel use {-# INLINE fromDir #-}  @@ -265,14 +267,23 @@ -- Writing -------------------------------------------------------------------- -- | Open some files for writing as individual buckets and pass --   them to the given consumer.----toFiles :: (Bulk l FilePath, Target l Bucket)-        =>  Array l FilePath            -- ^ File paths.-        -> (Array l Bucket -> IO b)-                                        -- ^ Consumer.+toFiles :: [FilePath]                   -- ^ File paths.+        -> (Array B Bucket -> IO b)     -- ^ Worker writes data to buckets.         -> IO b  toFiles paths use+        = toFiles' (A.fromList B paths) use+{-# INLINE toFiles #-}+++-- | Like `toFiles`, but take an array of `FilePath`s.+toFiles' :: (Bulk l FilePath, Target l Bucket)+         =>  Array l FilePath           -- ^ File paths.+         -> (Array l Bucket -> IO b)    -- ^ Worker writes data to buckets.+                                        -- ^ Consumer.+         -> IO b++toFiles' paths use  = do   -- Open all the files, ending up with a list of buckets.         bs             <- mapM (flip openBucket WriteMode) $ A.toList paths @@ -281,22 +292,12 @@         let Just bsArr =  A.fromListInto (A.layout paths) bs          use bsArr-{-# NOINLINE toFiles #-}+{-# NOINLINE toFiles' #-} --- --   TODO: Attached finalizers to the sinks so that file assocated with --   each stream is closed when that stream is ejected.  --- | Like `toFiles`, but take a list of file paths.-toFiles' :: [FilePath] -         -> (Array B Bucket -> IO b)-         -> IO b--toFiles' paths use-        = toFiles (A.fromList B paths) use-{-# INLINE toFiles' #-}-- -- | Create a new directory of the given name, containing the given number --   of buckets.  --@@ -342,13 +343,13 @@ --   For each directory @somedir@ the files are named --   @somedir/000000@, @somedir/000001@, @somedir/000002@ and so on. ---toDirs'  :: Int                  -- ^ Number of buckets to create per directory.+toDirs  :: Int                  -- ^ Number of buckets to create per directory.         -> [FilePath]           -- ^ Paths to directories.         -> (Array (E B DIM2) Bucket -> IO b)                                      -- ^ Consumer.         -> IO b -toDirs' nBucketsPerDir paths use+toDirs nBucketsPerDir paths use  | nBucketsPerDir <= 0  = do   let Just bsArr                  = A.fromListInto (A.matrix B 0 0) []@@ -384,7 +385,7 @@                         bs          use bsArr-{-# NOINLINE toDirs' #-}+{-# NOINLINE toDirs #-}   

@@ -1,135 +0,0 @@--module Data.Repa.Flow.IO.Storable-        ( Storable      (..)-        , Spec          (..))-where-import Data.Repa.Array.Meta                     as A-import Data.Repa.Array.Generic                  as A-import Data.Repa.Array.Material                 as A-import Data.Repa.Array.Material.Foreign         as AF-import Data.Repa.Array.Material.Strided         as AS-import Data.Repa.Flow.IO.Bucket                 as F-import qualified Foreign.Storable               as S-import Data.Int-#include "repa-flow.h"--------------------------------------------------------------------------------------------------------- | Class of element types that we can load and store to the file system.------   TODO: change to Persistable. --- -class Bulk (Rep a) a-   => Storable a where-- -- | Specification of how the elements are arranged in the file.- -- - --   For atomic elements the specification is the storage type.- --- --   For fixed-length arrays of elements, the specification contains the - --   element storage type as well as the array length.- --   - data Spec a   -- -- | Representation tag used for an array of these elements.- --- --   For atomic elements this will be `F`, for foreign arrays that - --   do not require extra copying after the data is read.- --- --   For tuples of elements, this will be a tuple of strided arrays,- --   so the elements can also be used without copying.- --- type Rep  a-- -- | Get the size of a single element, in bytes.- sizeElem :: Spec a -> Int-- -- | Read an array of the given length from a bucket.- --   If the bucket does not contain a whole number of elements remaining - --   then `Nothing`.- getArray-        :: Spec  a              -- ^ Element specification.-        -> Integer              -- ^ Number of elements to read.-        -> Bucket               -- ^ Bucket to read from.-        -> IO (Maybe (Array (Rep a) a))-- -- | Write an array to a bucket.-{-- putArray-        :: Spec  a              -- ^ Element specification.-        -> Bucket               -- ^ Bucket to write to.-        -> Array (Rep a) a      -- ^ Array to write.-        -> IO ()--}-------------------------------------------------------------------------------------------------------- | A native 32-bit integer.-instance Storable Int32  where- data Spec Int32  = SInt32- type Rep  Int32  = F-- sizeElem SInt32  = 4- {-# INLINE_FLOW sizeElem #-}-- getArray SInt32 lenElems bucket-  = do  let bytesElem   = sizeElem SInt32-        arr8            <- bGetArray bucket (lenElems * fromIntegral bytesElem)-        let (startBytes, lenBytes, fptrBuf)-                        = AF.toForeignPtr arr8-        let lenElems'   = lenBytes `div` sizeElem SInt32--        if  (startBytes /= 0)-         || (lenBytes   `mod` bytesElem /= 0)-         then return Nothing-         else return $ Just  -                $ AF.unsafeCast-                $ AF.fromForeignPtr lenElems' fptrBuf---------------------------------------------------------------------------------------------------------- | Two elements stored consecutively.-instance -        (   Storable a,   Storable b-        , S.Storable a, S.Storable b)-        => Storable (a, b) where-- data Spec (a, b) = S2 (Spec a) (Spec b)- type Rep  (a, b) = T2 S S-- sizeElem (S2 sA sB)-  = sizeElem sA + sizeElem sB- {-# INLINE_FLOW sizeElem #-}-- getArray (S2 sA sB) lenElems bucket-  = do  let bytesA      = sizeElem sA-        let bytesB      = sizeElem sB-        let bytesTuple  = bytesA + bytesB--        -- Read an array containing raw bytes.-        let lenBytes    = lenElems * fromIntegral bytesTuple-        arr8    <- bGetArray bucket lenBytes--        let lenBytes'   = A.length arr8-        let lenElems'   = lenBytes' `div` bytesTuple--        -- Check that we have received a whole number of elements.-        if lenBytes' `mod` bytesTuple /= 0-         then return Nothing-         else do-                let (startBuf, _lenBuf, fptrBuf) -                         = AF.toForeignPtr arr8--                let arrA = AS.unsafeCast-                         $ AS.fromForeignPtr -                                startBuf -                                bytesTuple (fromIntegral lenElems') fptrBuf--                let arrB = AS.unsafeCast-                         $ AS.fromForeignPtr -                                (startBuf + bytesA)-                                bytesTuple (fromIntegral lenElems') fptrBuf--                return $ Just $ T2Array arrA arrB--

@@ -48,8 +48,8 @@         , trigger_o            -- ** Ignorance-        , discard_o         , ignore_o+        , abandon_o            -- * Flow IO           -- ** Sourcing

@@ -31,8 +31,8 @@         , trigger_o            -- * Ignorance-        , discard_o-        , ignore_o)+        , ignore_o+        , abandon_o) where import Data.Repa.Flow.Simple.Base import Data.Repa.Flow.States                    (States (..))@@ -204,22 +204,22 @@   -- Ignorance --------------------------------------------------------------------- | A sink that drops all data on the floor.+-- | A sink that ignores all incoming elements. -- --   This sink is strict in the elements, so they are demanded before being --   discarded. Haskell debugging thunks attached to the elements will be demanded.-discard_o :: Monad m +ignore_o  :: Monad m            => m (Sink m a)-discard_o = G.discard_o ()-{-# INLINE discard_o #-}+ignore_o = G.ignore_o ()+{-# INLINE ignore_o #-}  --- | A sink that ignores all incoming elements.+-- | A sink that drops all data on the floor. -- --   This sink is non-strict in the elements.  --   Haskell tracing thinks attached to the elements will *not* be demanded.-ignore_o  :: Monad m +abandon_o :: Monad m            => m (Sink m a)-ignore_o = G.ignore_o ()-{-# INLINE ignore_o #-}+abandon_o = G.abandon_o ()+{-# INLINE abandon_o #-} 

@@ -1,5 +1,5 @@ Name:           repa-flow-Version:        4.1.0.1+Version:        4.2.2.1 License:        BSD3 License-file:   LICENSE Author:         The Repa Development Team@@ -15,23 +15,28 @@  Library   build-Depends: -        base                    == 4.7.*,+        base                    == 4.8.*,         directory               == 1.2.*,-        filepath                == 1.3.*,+        filepath                == 1.4.*,         vector                  == 0.10.*,         bytestring              == 0.10.*,-        primitive               == 0.5.4.*,+        primitive               == 0.6.*,         containers              == 0.5.*,+        hashtables              == 1.2.1.*,         text                    == 1.2.*,         repa-eval               == 4.0.0.*,-        repa-stream             == 4.1.0.*,-        repa-array              == 4.1.0.*+        repa-stream             == 4.2.2.*,+        repa-scalar             == 4.2.2.*,+        repa-convert            == 4.2.2.*,+        repa-array              == 4.2.2.*    exposed-modules:         Data.Repa.Flow.Auto         Data.Repa.Flow.Auto.Debug         Data.Repa.Flow.Auto.IO+        Data.Repa.Flow.Auto.Format                 Data.Repa.Flow.Auto.SizedIO+        Data.Repa.Flow.Auto.ZipWith          Data.Repa.Flow.Chunked         Data.Repa.Flow.Chunked.IO@@ -40,9 +45,7 @@         Data.Repa.Flow.Generic.Debug         Data.Repa.Flow.Generic.IO -        Data.Repa.Flow.IO.Binary         Data.Repa.Flow.IO.Bucket-        Data.Repa.Flow.IO.Storable          Data.Repa.Flow.Simple @@ -51,18 +54,24 @@         Data.Repa.Flow    other-modules:+        Data.Repa.Flow.Auto.Base+        Data.Repa.Flow.Auto.Select+         Data.Repa.Flow.Chunked.Base         Data.Repa.Flow.Chunked.Map         Data.Repa.Flow.Chunked.Fold         Data.Repa.Flow.Chunked.Folds         Data.Repa.Flow.Chunked.Groups-        Data.Repa.Flow.Chunked.Operator+        Data.Repa.Flow.Chunked.Generic+        Data.Repa.Flow.Chunked.Process+        Data.Repa.Flow.Chunked.Replicate          Data.Repa.Flow.Generic.Base         Data.Repa.Flow.Generic.Connect         Data.Repa.Flow.Generic.List         Data.Repa.Flow.Generic.Map         Data.Repa.Flow.Generic.Operator+        Data.Repa.Flow.Generic.Process         Data.Repa.Flow.Generic.Eval         Data.Repa.Flow.Generic.Array.Distribute         Data.Repa.Flow.Generic.Array.Shuffle@@ -71,6 +80,7 @@         Data.Repa.Flow.Generic.IO.Base         Data.Repa.Flow.Generic.IO.Sieve         Data.Repa.Flow.Generic.IO.XSV+        Data.Repa.Flow.Generic.IO.Lines          Data.Repa.Flow.Simple.Base         Data.Repa.Flow.Simple.List@@ -91,20 +101,19 @@    extensions:         CPP-        BangPatterns-        NoMonomorphismRestriction-        RankNTypes         MagicHash+        RankNTypes+        TypeFamilies+        BangPatterns+        PatternGuards+        ConstraintKinds+        ParallelListComp         FlexibleContexts         FlexibleInstances-        PatternGuards-        TypeFamilies-        MultiParamTypeClasses+        StandaloneDeriving         ScopedTypeVariables+        MultiParamTypeClasses         FunctionalDependencies-        ConstraintKinds         ForeignFunctionInterface-        StandaloneDeriving-        ParallelListComp+        NoMonomorphismRestriction -