Data.Array.Parallel.Stream

Stream functions not implemented in Data.Vector

Flat stream operators

indexedS :: Stream a -> Stream (Int, a)

Tag each element of an stream with its index in that stream.

 indexed [42,93,13]
  = [(0,42), (1,93), (2,13)]

replicateEachS :: Int -> Stream (Int, a) -> Stream a

Given a stream of pairs containing a count an an element, replicate element the number of times given by the count.

The first parameter sets the size hint of the resulting stream.

 replicateEach 10 [(2,10), (5,20), (3,30)]
   = [10,10,20,20,20,20,20,30,30,30]

replicateEachRS :: Int -> Stream a -> Stream a

Repeat each element in the stream the given number of times.

 replicateEach 2 [10,20,30]
  = [10,10,20,20,30,30]

interleaveS :: Stream a -> Stream a -> Stream a

Interleave the elements of two streams. We alternate between the first and second streams, stopping when we can't find a matching element.

 interleave [2,3,4] [10,20,30] = [2,10,3,20,4,30]
 interleave [2,3]   [10,20,30] = [2,10,3,20]
 interleave [2,3,4] [10,20]    = [2,10,3,20,4]

combine2ByTagS :: Stream Tag -> Stream a -> Stream a -> Stream a

Combine two streams, using a tag stream to tell us which of the data streams to take the next element from.

If there are insufficient elements in the data strams for the provided tag stream then error.

 combine2ByTag [0,1,1,0,0,1] [1,2,3] [4,5,6]
  = [1,4,5,2,3,6]

enumFromToEachS :: Int -> Stream (Int, Int) -> Stream Int

Create a stream of integer ranges. The pairs in the input stream give the first and last value of each range.

The first parameter gives the size hint for the resulting stream.

 enumFromToEach 11 [(2,5), (10,16), (20,22)]
  = [2,3,4,5,10,11,12,13,14,15,16,20,21,22]

enumFromStepLenEachS :: Int -> Stream (Int, Int, Int) -> Stream Int

Create a stream of integer ranges. The triples in the input stream give the first value, increment, length of each range.

The first parameter gives the size hint for the resulting stream.

 enumFromStepLenEach [(1,1,5), (10,2,4), (20,3,5)]
  = [1,2,3,4,5,10,12,14,16,20,23,26,29,32]

Segmented stream operators

foldSS

Arguments

:: (a -> b -> a)

function to perform the fold

-> a

initial element of each fold

-> Stream Int

stream of segment lengths

-> Stream b

stream of input data

-> Stream a

stream of fold results

Segmented Stream fold. Take segments from the given stream and fold each using the supplied function and initial element.

 foldSS (+) 0 [2, 3, 2] [10, 20, 30, 40, 50, 60, 70]
  = [30,120,130]

fold1SS :: (a -> a -> a) -> Stream Int -> Stream a -> Stream a

Like foldSS, but use the first member of each chunk as the initial element for the fold.

combineSS

Arguments

:: Stream Bool

tag values

-> Stream Int

segment lengths for first data stream

-> Stream a

first data stream

-> Stream Int

segment lengths for second data stream

-> Stream a

second data stream

-> Stream a

Segmented Stream combine. Like combine2ByTagS, except that the tags select entire segments of each data stream, instead of selecting one element at a time.

 combineSS [True, True, False, True, False, False]
           [2,1,3] [10,20,30,40,50,60]
           [1,2,3] [11,22,33,44,55,66]
  = [10,20,30,11,40,50,60,22,33,44,55,66]

This says take two elements from the first stream, then another one element from the first stream, then one element from the second stream, then three elements from the first stream...

appendSS

Arguments

:: Stream Int

segment lengths for first data stream

-> Stream a

first data stream

-> Stream Int

segment lengths for second data stream

-> Stream a

second data stream

-> Stream a

Segmented Strem append. Append corresponding segments from each stream.

 appendSS [2, 1, 3] [10, 20, 30, 40, 50, 60]
          [1, 3, 2] [11, 22, 33, 44, 55, 66]
  = [10,20,11,30,22,33,44,40,50,60,55,66]

foldValuesR

Arguments

:: (a -> b -> a)

function to perform the fold

-> a

initial element for fold

-> Int

length of each segment

-> Stream b

data stream

-> Stream a

Segmented Stream fold, with a fixed segment length.

Like foldSS but use a fixed length for each segment.

indicesSS :: Int -> Int -> Stream Int -> Stream Int

Segmented Stream indices.

 indicesSS 15 4 [3, 5, 7]
  = [4,5,6,0,1,2,3,4,0,1,2,3,4,5,6]

Note that we can set the starting value of the first segment independently via the second argument of indicesSS. We use this when distributing arrays across worker threads, as a thread's chunk may not start exactly at a segment boundary, so the index of a thread's first data element may not be zero.