repa-flow-4.2.3.1: Data-parallel data flows.

Data.Repa.Flow.Chunked

Synopsis

# Documentation

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

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

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

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) Source #

Shorthand for common type classes.

# Evaluation

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

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 () Source #

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) Source #

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) Source #

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] Source #

Drain a single source into a list of elements.

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

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) Source #

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) Source #

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) Source #

## 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) Source #

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) Source #

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) Source #

Combine the elements of two flows with the given function.

## Processing

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

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) MethodsshowsPrec :: 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) Source #

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

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) Source #

Dictionaries needed to perform a grouping.

## Folding

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.

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.

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) Source #

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) Source #

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) Source #

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) Source #

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) Source #

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) Source #

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.