Data.Repa.Flow.Chunked

Documentation

module Data.Repa.Flow.States

type Sources i m l e = Sources i m (Array l e)

A bundle of sources, where the elements are chunked into arrays.

type Sinks i m l e = Sinks i m (Array l e)

A bundle of sinks, where the elements are chunked into arrays.

type Flow i m l a = (Ord i, Monad m, BulkI l a, States i m)

Shorthand for common type classes.

Evaluation

drainS :: (Next i, Monad m) => Sources i m r a -> Sinks i m r a -> m ()

Pull all available values from the sources and push them to the sinks.

consumeS :: (Next i, Monad m, Bulk r a) => Sources i m r a -> (i -> a -> m ()) -> m ()

Pull all available values from the sources and pass them to the given action.

Conversion

fromList :: (States i m, TargetI l a) => Name l -> i -> [a] -> m (Sources i m l a)

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.

fromLists :: (States i m, TargetI l a) => Name l -> i -> [[a]] -> m (Sources i m l a)

Like fromLists but take a list of lists, where each of the inner lists is packed into a single chunk.

toList1 :: (States i m, BulkI l a) => i -> Sources i m l a -> m [a]

Drain a single source into a list of elements.

toLists1 :: (States i m, BulkI l a) => i -> Sources i m l a -> m [[a]]

Drain a single source into a list of chunks.

Finalizers

finalize_i :: States i m => (i -> m ()) -> Sources i m l a -> m (Sources i m l a)

Attach a finalizer to a bundle of sources.

For each stream in the bundle, the finalizer will be called the first time a consumer of that stream tries to pull an element when no more are available.

The provided finalizer will be run after any finalizers already attached to the source.

finalize_o :: States i m => (i -> m ()) -> Sinks i m l a -> m (Sinks i m l a)

Attach a finalizer to a bundle of sinks.

The finalizer will be called the first time the stream is ejected.

The provided finalizer will be run after any finalizers already attached to the sink.

Flow Operators

Replicate

replicates_i :: (Flow i m lSrc (Int, a), TargetI lDst a) => Name lDst -> Sources i m lSrc (Int, a) -> m (Sources i m lDst a)

Mapping

If you want to work on a chunk at a time then use map_i and map_o from Data.Repa.Flow.Generic.

smap_i :: (Flow i m l1 a, TargetI l2 b) => (i -> a -> b) -> Sources i m l1 a -> m (Sources i m l2 b)

Map a function over elements pulled from a source.

smap_o :: (Flow i m l1 a, TargetI l2 b) => (i -> a -> b) -> Sinks i m l2 b -> m (Sinks i m l1 a)

Map a function over elements pushed into a sink.

szipWith_ii :: (States i m, BulkI lSrc1 a, BulkI lSrc2 b, TargetI lDst c, Windowable lSrc1 a, Windowable lSrc2 b) => Name lDst -> (i -> a -> b -> c) -> Sources i m lSrc1 a -> Sources i m lSrc2 b -> m (Sources i m lDst c)

Combine the elements of two flows with the given function.

Processing

process_i

Arguments

:: (States i m, BulkI lSrc a, Bulk lDst b, Bulk lDst (Array lDst b), TargetI lDst b, TargetI lDst (Array lDst b))

=> (s -> a -> (s, Array lDst b))

Worker function.

-> s

Initial state.

-> Sources i m lSrc a

Input sources

-> m (Sources i m lDst b)

Result sources.

Apply a generic stream process to all the streams in a bundle of sources.

Unfolding

unfolds_i

Arguments

:: (States i m, BulkI lSrc a, Bulk lDst b, TargetI lDst b)

=> (a -> s -> StepUnfold s b)

Worker function.

-> s

Initial state.

-> Sources i m lSrc a

Input sources

-> m (Sources i m lDst b)

Result sources.

Apply a generic stream process to all the streams in a bundle of sources.

data StepUnfold s a :: * -> * -> *

Constructors

StepUnfoldGive a s

StepUnfoldNext a s

StepUnfoldBump s

StepUnfoldFinish s

Instances

(Show s, Show a) => Show (StepUnfold s a)

Methods

showsPrec :: Int -> StepUnfold s a -> ShowS

show :: StepUnfold s a -> String

showList :: [StepUnfold s a] -> ShowS

Splitting

head_i :: (States i m, Windowable l a, Index l ~ Int) => Int -> Sources i m l a -> i -> m ([a], Sources i m l a)

Split the given number of elements from the head of a source, retrurning those elements in a list, and yielding a new source for the rest.

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

Grouping

groupsBy_i

Arguments

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

-> Sources i m lVal a

Source values.

-> m (Sources i m (T2 lGrp lLen) (a, Int))

From a stream of values which has consecutive runs of idential values, produce a stream of the lengths of these runs.

  groupsBy (==) [4, 4, 4, 3, 3, 1, 1, 1, 4] 
   => [3, 2, 3, 1]

type GroupsDict i m lVal lGrp lLen a = (Flow i m lVal a, Index lVal ~ Int, TargetI lGrp a, TargetI lLen Int)

Dictionaries needed to perform a grouping.

Folding

foldlS

Arguments

:: (States Int m, Target lDst a, Index lDst ~ Int, BulkI lSrc b)

=> Name lDst

Destination layout.

-> (a -> b -> a)

Combining function.

-> a

Starting value for fold.

-> Sources Int m lSrc b

Input elements to fold.

-> m (Array lDst a)

Fold all elements of all streams in a bundle individually, returning an array of per-stream results.

foldlAllS

Arguments

:: (States () m, BulkI lSrc b)

=> (a -> b -> a)

Combining function.

-> a

Starting value for fold.

-> Sources Int m lSrc b

Input elements to fold.

-> m a

Fold all elements of all streams in a bundle together, one stream after the other, returning the single final value.

folds_i

Arguments

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

-> b

Initial state when folding each segment.

-> Sources i m lSeg (n, Int)

Segment lengths.

-> Sources i m lElt a

Input elements to fold.

-> m (Sources i m (T2 lGrp lRes) (n, b))

Result elements.

Segmented fold over vectors of segment lengths and input values.

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), BulkI lElt a, BulkI lGrp n, BulkI lRes b, TargetI lElt a, TargetI lGrp n, TargetI lRes b)

Dictionaries needed to perform a segmented fold.

Watching

watch_i :: Monad m => (i -> Array l a -> m ()) -> Sources i m l a -> m (Sources i m l a)

Hook a monadic function to some sources, which will be passed every chunk that is pulled from the result.

watch_o :: Monad m => (i -> Array l a -> m ()) -> Sinks i m l a -> m (Sinks i m l a)

Hook a monadic function to some sinks, which will be passed every chunk that is pushed to the result.

trigger_o :: Monad m => i -> (i -> Array l a -> m ()) -> m (Sinks i m l a)

Like watch_o but discard the incoming chunks after they are passed to the function.

Ignorance

ignore_o :: Monad m => i -> m (Sinks i m l a)

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.

abandon_o :: Monad m => i -> m (Sinks i m l a)

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.