dph-prim-par 0.5.1.1 → 0.6.0.1

raw patch · 36 files changed

+3931/−1502 lines, 36 filesdep ~basedep ~dph-basedep ~dph-prim-interfacePVP ok

version bump matches the API change (PVP)

Dependency ranges changed: base, dph-base, dph-prim-interface, dph-prim-seq, old-time, vector

API changes (from Hackage documentation)

Files

• Data/Array/Parallel/Unlifted.hs +227/−151
• Data/Array/Parallel/Unlifted/Distributed.hs +50/−30
• Data/Array/Parallel/Unlifted/Distributed/Arrays.hs +180/−164
• Data/Array/Parallel/Unlifted/Distributed/Basics.hs +22/−20
• Data/Array/Parallel/Unlifted/Distributed/Combinators.hs +110/−74
• Data/Array/Parallel/Unlifted/Distributed/DistST.hs +24/−17
• Data/Array/Parallel/Unlifted/Distributed/Gang.hs +58/−51
• Data/Array/Parallel/Unlifted/Distributed/Scalars.hs +12/−12
• Data/Array/Parallel/Unlifted/Distributed/TheGang.hs +16/−14
• Data/Array/Parallel/Unlifted/Distributed/Types.hs +15/−489
• Data/Array/Parallel/Unlifted/Distributed/Types/Base.hs +108/−0
• Data/Array/Parallel/Unlifted/Distributed/Types/Maybe.hs +45/−0
• Data/Array/Parallel/Unlifted/Distributed/Types/Prim.hs +250/−0
• Data/Array/Parallel/Unlifted/Distributed/Types/Tuple.hs +145/−0
• Data/Array/Parallel/Unlifted/Distributed/Types/USSegd.hs +130/−0
• Data/Array/Parallel/Unlifted/Distributed/Types/USegd.hs +113/−0
• Data/Array/Parallel/Unlifted/Distributed/Types/UVSegd.hs +128/−0
• Data/Array/Parallel/Unlifted/Distributed/Types/Unit.hs +45/−0
• Data/Array/Parallel/Unlifted/Distributed/Types/Vector.hs +61/−0
• Data/Array/Parallel/Unlifted/Distributed/USSegd.hs +245/−0
• Data/Array/Parallel/Unlifted/Distributed/USegd.hs +349/−0
• Data/Array/Parallel/Unlifted/Parallel.hs +66/−30
• Data/Array/Parallel/Unlifted/Parallel/Basics.hs +53/−39
• Data/Array/Parallel/Unlifted/Parallel/Combinators.hs +50/−35
• Data/Array/Parallel/Unlifted/Parallel/Enum.hs +32/−31
• Data/Array/Parallel/Unlifted/Parallel/Extracts.hs +104/−0
• Data/Array/Parallel/Unlifted/Parallel/Permute.hs +13/−10
• Data/Array/Parallel/Unlifted/Parallel/Segmented.hs +76/−147
• Data/Array/Parallel/Unlifted/Parallel/Subarrays.hs +10/−7
• Data/Array/Parallel/Unlifted/Parallel/Sums.hs +38/−26
• Data/Array/Parallel/Unlifted/Parallel/Text.hs +4/−6
• Data/Array/Parallel/Unlifted/Parallel/UPSSegd.hs +318/−0
• Data/Array/Parallel/Unlifted/Parallel/UPSegd.hs +260/−40
• Data/Array/Parallel/Unlifted/Parallel/UPSel.hs +56/−91
• Data/Array/Parallel/Unlifted/Parallel/UPVSegd.hs +478/−0
• dph-prim-par.cabal +40/−18

@@ -1,100 +1,139 @@-{-# LANGUAGE PackageImports, CPP #-}---- | Primitive parallel combinators that work on flat, unlifted arrays.---   Some of them don't actually have parallel implementations, so we bail out---   to the regular sequential ones.+{-# LANGUAGE CPP, NoMonomorphismRestriction #-}+{-# OPTIONS -fno-warn-missing-signatures #-}+-- | Parallel implementation of the segmented array API defined in @dph-prim-interface@. -----   This set of combinators is used when the program is comiled with @-fdph-par@.---   When compiling with @-fdph-seq@, the ones in the @dph-prim-seq@ package are used---   instead. The @dph-prim-seq package@ exports the same names, but all combinators---   are implemented sequentially.+--   Some of them don't yet have parallel implementations, so we fall back+--   to the sequential ones from @dph-prim-seq@. -----   The API is defined in @DPH_Header.h@ and @DPH_Interface.h@ to ensure that both---   @dph-prim-par@ and @dph-prim-seq@ really do export the same symbols.-+--   /WARNING:/ Although this library is intended to be used as a target+--   for the DPH vectoriser, it is also fine to use it directly from non+--   DPH programs. However, this library does not support nested parallelism+--   by itself. If you try to run further parallel computations in the workers+--   passed to `map`, `zipWith`, `fold` etc, then they will just run+--   sequentially.+---+--   This API is used by the @dph-lifted-*@ libraries, and is defined in+--   @DPH_Header.h@ and @DPH_Interface.h@. We use header files to ensure+--   that this API is implemented identically by both the +--   @dph-prim-par@ and @dph-prim-seq@ packages.+-- #include "DPH_Header.h"- import Data.Array.Parallel.Unlifted.Parallel import Data.Array.Parallel.Base.TracePrim import Data.Array.Parallel.Unlifted.Distributed ( DT )-import qualified Data.Array.Parallel.Unlifted.Sequential.Vector         as Seq-import qualified Data.Array.Parallel.Unlifted.Sequential.Segmented      as Seq-import Data.Array.Parallel.Unlifted.Sequential.Vector (Unbox,Vector)-import Prelude (($!)) +import Data.Array.Parallel.Unlifted.Sequential.Vector           (Unbox,   Vector)+import Data.Array.Parallel.Unlifted.Vectors                     (Unboxes, Vectors)+import Data.Array.Parallel.Unlifted.Parallel.UPSel+import qualified Data.Array.Parallel.Unlifted.Parallel.UPSegd   as UPSegd+import qualified Data.Array.Parallel.Unlifted.Parallel.UPSSegd  as UPSSegd+import qualified Data.Array.Parallel.Unlifted.Parallel.UPVSegd  as UPVSegd+import qualified Data.Array.Parallel.Unlifted.Sequential.Vector as Seq+import qualified Data.Array.Parallel.Unlifted.Vectors           as US+import qualified Data.Array.Parallel.Unlifted.Sequential        as Seq+import Prelude (($!)) #include "DPH_Interface.h" -class (Unbox a, DT a) => Elt a+-- NOTE +-- See DPH_Interface.h for documentation. +--+-- The defs should appear in the same order as they are listed in DPH_Interface.h+--+-- Operations with at least O(n) time will print trace messages to console when+-- dph-base/D/A/P/Config.tracePrimEnabled is set to True.+-- +-- Basics ---------------------------------------------------------------------+class (Unbox a,   DT a) => Elt a++-- | Arrays are stored as unboxed vectors. +--   They have bulk-strict semantics, so demanding one element demands them all. type Array      = Vector-type Segd       = UPSegd-type Sel2       = UPSel2-type SelRep2    = UPSelRep2  --- Constant time operations ------------------------------------------------------   We don't bother tracing these.--length                  = Seq.length-empty                   = Seq.empty-zip                     = Seq.zip-unzip                   = Seq.unzip-fsts                    = Seq.fsts-snds                    = Seq.snds-(!:)                    = (Seq.!)--elementsSel2_0          = elementsUPSel2_0-elementsSel2_1          = elementsUPSel2_1-repSel2                 = repUPSel2--mkSelRep2               = mkUPSelRep2-indicesSelRep2          = indicesUPSelRep2-elementsSelRep2_0       = elementsUPSelRep2_0-elementsSelRep2_1       = elementsUPSelRep2_1--mkSegd                  = mkUPSegd-lengthSegd              = lengthUPSegd-lengthsSegd             = lengthsUPSegd-indicesSegd             = indicesUPSegd-elementsSegd            = elementsUPSegd+-- Constructors ---------------------------------------------------------------+empty   = Seq.empty +(+:+) arr1 arr2+        =  tracePrim (TraceAppend (Seq.length arr1 + Seq.length arr2))+        $! (Seq.++) arr1 arr2 ----------------------------------------------------------------------------------- These take least O(n) time in the length of the vector.---   NOTE: That actual tracing is only enabled when ---         dph-base/D/A/P/Config.tracePrimEnabled is set to True,---         otherwise tracePrim is a no-op.+append_s segd xd xs yd ys+ = let  arr     = appendSUP segd xd xs yd ys+   in   tracePrim (TraceAppend_s (Seq.length arr)) arr  replicate n val          =  tracePrim (TraceReplicate n)         $! replicateUP n val +replicate_s segd arr+        =  tracePrim (TraceReplicate_s (Seq.length arr))+        $! UPSegd.replicateWithP segd arr ++replicate_rs n arr+        =  tracePrim (TraceReplicate_rs n (Seq.length arr))+        $! replicateRSUP n arr+ repeat n _ arr         =  tracePrim (TraceRepeat n (Seq.length arr))         $! repeatUP n arr +indexed arr+        =  tracePrim (TraceIndexed (Seq.length arr))+        $! indexedUP arr +indices_s segd+ = let  arr     = UPSegd.indicesP segd+   in   tracePrim (TraceIndices_s (Seq.length arr)) arr++enumFromTo from to+ = let  arr     = enumFromToUP from to+   in   tracePrim (TraceEnumFromTo (Seq.length arr)) arr++enumFromThenTo from thn to+ = let  arr     = enumFromThenToUP from thn to+   in   tracePrim (TraceEnumFromThenTo (Seq.length arr)) arr+   +enumFromStepLen from step len+ = let  arr     = enumFromStepLenUP from step len+   in   tracePrim (TraceEnumFromStepLen (Seq.length arr)) arr++enumFromStepLenEach n starts steps lens+ = let  arr     = enumFromStepLenEachUP n starts steps lens+   in   tracePrim (TraceEnumFromStepLenEach (Seq.length arr)) arr+++-- Projections ----------------------------------------------------------------+length          = Seq.length+index           = Seq.index+indexs          = indexsFromVector+indexs_avs      = indexsFromVectorsUPVSegd+ extract arr i n         =  tracePrim (TraceExtract (Seq.length arr) i n)         $! Seq.extract arr i n +extracts_nss    = extractsFromNestedUPSSegd+extracts_ass    = extractsFromVectorsUPSSegd+extracts_avs    = extractsFromVectorsUPVSegd  drop n arr         =  tracePrim (TraceDrop n (Seq.length arr))         $! dropUP n arr  +-- Update ---------------------------------------------------------------------+update arrSrc arrNew+        =  tracePrim (TraceUpdate (Seq.length arrSrc) (Seq.length arrNew))+        $! updateUP arrSrc arrNew+++-- Permutation ---------------------------------------------------------------- permute arrSrc arrIxs         =  tracePrim (TracePermute (Seq.length arrSrc))         $! Seq.permute arrSrc arrIxs --bpermuteDft len f arrIxs-        =  tracePrim (TraceBPermuteDft len)-        $! Seq.bpermuteDft len f arrIxs-- bpermute arrSrc arrIxs         =  tracePrim (TraceBPermute (Seq.length arrSrc))         $! bpermuteUP arrSrc arrIxs@@ -104,105 +143,101 @@         =  tracePrim (TraceMBPermute (Seq.length arrSrc))         $! Seq.mbpermute f arrSrc streamIxs --update arrSrc arrNew-        =  tracePrim (TraceUpdate (Seq.length arrSrc) (Seq.length arrNew))-        $! updateUP arrSrc arrNew---(+:+) arr1 arr2-        =  tracePrim (TraceAppend (Seq.length arr1 + Seq.length arr2))-        $! (Seq.++) arr1 arr2---interleave arr1 arr2-        =  tracePrim (TraceInterleave (Seq.length arr1 + Seq.length arr2))-        $! interleaveUP arr1 arr2---pack arrSrc arrFlag-        =  tracePrim (TracePack (Seq.length arrSrc))-        $! packUP arrSrc arrFlag---combine arrSel arr1 arr2-        =  tracePrim (TraceCombine (Seq.length arrSel))-        $! combineUP arrSel arr1 arr2+bpermuteDft len f arrIxs+        =  tracePrim (TraceBPermuteDft len)+        $! Seq.bpermuteDft len f arrIxs  -combine2 arrTag sel arr1 arr2-        =  tracePrim (TraceCombine2 (Seq.length arrTag))-        $! combine2UP arrTag sel arr1 arr2+-- Zipping and Unzipping ------------------------------------------------------+zip     = Seq.zip+zip3    = Seq.zip3+unzip   = Seq.unzip+unzip3  = Seq.unzip3+fsts    = Seq.fsts+snds    = Seq.snds  +-- Map and ZipWith ------------------------------------------------------------ map f arr         =  tracePrim (TraceMap (Seq.length arr))         $! mapUP f arr --filter f src- = let  dst     = filterUP f src-   in   tracePrim (TraceFilter (Seq.length src) (Seq.length dst)) dst-- zipWith f arr1 arr2         =  tracePrim (TraceZipWith (Seq.length arr1) (Seq.length arr2))         $! zipWithUP f arr1 arr2  +-- Scans and Folds ------------------------------------------------------------+scan f x arr+        =  tracePrim (TraceScan (Seq.length arr))+        $! scanUP f x arr+ fold f x arr         =  tracePrim (TraceFold (Seq.length arr))         $! foldUP f x arr +fold_s f x segd arr+        =  tracePrim (TraceFold_s (Seq.length arr))+        $! UPSegd.foldWithP f x segd arr++fold_ss = UPSSegd.foldWithP++fold_r f z segSize arr+        =  tracePrim (TraceFold_r (Seq.length arr))+        $! Seq.foldlRU f z segSize arr          fold1 f arr         =  tracePrim (TraceFold1 (Seq.length arr))         $! Seq.fold1 f arr --and arr =  tracePrim (TraceAnd (Seq.length arr))-        $! andUP arr+fold1_s f segd arr+        =  tracePrim (TraceFold1_s (Seq.length arr))+        $! UPSegd.fold1WithP f segd arr +fold1_ss = UPSSegd.fold1WithP  sum arr =  tracePrim (TraceSum (Seq.length arr))         $! sumUP arr -                        -scan f x arr-        =  tracePrim (TraceScan (Seq.length arr))-        $! scanUP f x arr+sum_r x arr+        =  tracePrim (TraceSum_r (Seq.length arr))+        $! sumRUP x arr -        -indexed arr-        =  tracePrim (TraceIndexed (Seq.length arr))-        $! indexedUP arr+and arr =  tracePrim (TraceAnd (Seq.length arr))+        $! andUP arr  -enumFromTo from to- = let  arr     = enumFromToUP from to-   in   tracePrim (TraceEnumFromTo (Seq.length arr)) arr+-- Pack and Filter ------------------------------------------------------------+pack arrSrc arrFlag+        =  tracePrim (TracePack (Seq.length arrSrc))+        $! packUP arrSrc arrFlag -        -enumFromThenTo from thn to- = let  arr     = enumFromThenToUP from thn to-   in   tracePrim (TraceEnumFromThenTo (Seq.length arr)) arr+filter f src+ = let  dst     = filterUP f src+   in   tracePrim (TraceFilter (Seq.length src) (Seq.length dst)) dst -   -enumFromStepLen from step len- = let  arr     = enumFromStepLenUP from step len-   in   tracePrim (TraceEnumFromStepLen (Seq.length arr)) arr +-- Combine and Interleave -----------------------------------------------------+combine arrSel arr1 arr2+        =  tracePrim (TraceCombine (Seq.length arrSel))+        $! combineUP arrSel arr1 arr2 -enumFromStepLenEach n starts steps lens- = let  arr     = enumFromStepLenEachUP n starts steps lens-   in   tracePrim (TraceEnumFromStepLenEach (Seq.length arr)) arr+combine2 arrTag sel arr1 arr2+        =  tracePrim (TraceCombine2 (Seq.length arrTag))+        $! combine2UP arrTag sel arr1 arr2 +interleave arr1 arr2+        =  tracePrim (TraceInterleave (Seq.length arr1 + Seq.length arr2))+        $! interleaveUP arr1 arr2 ++-- Selectors ------------------------------------------------------------------+type Sel2               = UPSel2+ mkSel2 tag is n0 n1 rep         =  tracePrim (TraceMkSel2 (Seq.length is))         $! mkUPSel2 tag is n0 n1 rep - tagsSel2 sel  = let  tags    = tagsUPSel2 sel    in   tracePrim (TraceTagsSel2 (Seq.length tags)) tags@@ -213,54 +248,95 @@    in   tracePrim (TraceIndicesSel2 (Seq.length arr)) arr  -replicate_s segd arr-        =  tracePrim (TraceReplicate_s (Seq.length arr))-        $! replicateSUP segd arr+elementsSel2_0                  = elementsUPSel2_0+elementsSel2_1                  = elementsUPSel2_1+repSel2                         = repUPSel2 +type SelRep2                    = UPSelRep2+mkSelRep2                       = mkUPSelRep2 -replicate_rs n arr-        =  tracePrim (TraceReplicate_rs n (Seq.length arr))-        $! replicateRSUP n arr+indicesSelRep2                  = indicesUPSelRep2+elementsSelRep2_0               = elementsUPSelRep2_0+elementsSelRep2_1               = elementsUPSelRep2_1  -append_s segd xd xs yd ys- = let  arr     = appendSUP segd xd xs yd ys-   in   tracePrim (TraceAppend_s (Seq.length arr)) arr+-- Segment Descriptors --------------------------------------------------------+type Segd                       = UPSegd.UPSegd+mkSegd                          = UPSegd.mkUPSegd+validSegd                       = UPSegd.valid+emptySegd                       = UPSegd.empty+singletonSegd                   = UPSegd.singleton+lengthSegd                      = UPSegd.length+lengthsSegd                     = UPSegd.takeLengths+indicesSegd                     = UPSegd.takeIndices+elementsSegd                    = UPSegd.takeElements -        -fold_s f x segd arr-        =  tracePrim (TraceFold_s (Seq.length arr))-        $! foldSUP f x segd arr -        -fold1_s f segd arr-        =  tracePrim (TraceFold1_s (Seq.length arr))-        $! fold1SUP f segd arr+-- Scattered Segment Descriptors ----------------------------------------------+type SSegd                      = UPSSegd.UPSSegd+mkSSegd                         = UPSSegd.mkUPSSegd+promoteSegdToSSegd              = UPSSegd.fromUPSegd+validSSegd                      = UPSSegd.valid+emptySSegd                      = UPSSegd.empty+singletonSSegd                  = UPSSegd.singleton+isContiguousSSegd               = UPSSegd.isContiguous+lengthOfSSegd                   = UPSSegd.length+lengthsOfSSegd                  = UPSSegd.takeLengths+indicesOfSSegd                  = UPSSegd.takeIndices+startsOfSSegd                   = UPSSegd.takeStarts+sourcesOfSSegd                  = UPSSegd.takeSources+getSegOfSSegd                   = UPSSegd.getSeg+appendSSegd                     = UPSSegd.appendWith  -fold_r f z segSize arr-        =  tracePrim (TraceFold_r (Seq.length arr))-        $! Seq.foldlRU f z segSize arr+-- Virtual Segment Descriptors ------------------------------------------------+type VSegd                      = UPVSegd.UPVSegd+mkVSegd                         = UPVSegd.mkUPVSegd+validVSegd                      = UPVSegd.valid+emptyVSegd                      = UPVSegd.empty+singletonVSegd                  = UPVSegd.singleton+replicatedVSegd                 = UPVSegd.replicated+promoteSegdToVSegd              = UPVSegd.fromUPSegd+promoteSSegdToVSegd             = UPVSegd.fromUPSSegd+isManifestVSegd                 = UPVSegd.isManifest+isContiguousVSegd               = UPVSegd.isContiguous+lengthOfVSegd                   = UPVSegd.length+takeVSegidsOfVSegd              = UPVSegd.takeVSegids+takeVSegidsRedundantOfVSegd     = UPVSegd.takeVSegidsRedundant+takeSSegdOfVSegd                = UPVSegd.takeUPSSegd+takeSSegdRedundantOfVSegd       = UPVSegd.takeUPSSegdRedundant+takeLengthsOfVSegd              = UPVSegd.takeLengths+getSegOfVSegd                   = UPVSegd.getSeg+unsafeDemoteToSSegdOfVSegd      = UPVSegd.unsafeDemoteToUPSSegd+unsafeDemoteToSegdOfVSegd       = UPVSegd.unsafeDemoteToUPSegd+updateVSegsOfVSegd              = UPVSegd.updateVSegs+updateVSegsReachableOfVSegd     = UPVSegd.updateVSegsReachable+appendVSegd                     = UPVSegd.appendWith+combine2VSegd                   = UPVSegd.combine2  -sum_r x arr-        =  tracePrim (TraceSum_r (Seq.length arr))-        $! sumRUP x arr-+-- Irregular 2D arrays --------------------------------------------------------+class (Unboxes a, DT a) => Elts a -indices_s segd- = let  arr     = indicesSUP segd-   in   tracePrim (TraceIndices_s (Seq.length arr)) arr+type Arrays                     = Vectors+emptys                          = US.empty+lengths                         = US.length+singletons                      = US.singleton+unsafeIndexs                    = US.unsafeIndex+unsafeIndex2s                   = US.unsafeIndex2+appends                         = US.append+fromVectors                     = US.fromVector+toVectors                       = US.toVector  --- Random arrays -------------------------------------------randoms                 = Seq.random-randomRs                = Seq.randomR+-- Random arrays --------------------------------------------------------------+randoms                         = Seq.random+randomRs                        = Seq.randomR  --- IO -----------------------------------------------------+-- IO ------------------------------------------------------------------------- class Seq.UIO a => IOElt a-hPut                    = Seq.hPut-hGet                    = Seq.hGet-toList                  = Seq.toList-fromList                = Seq.fromList+hPut                            = Seq.hPut+hGet                            = Seq.hGet+toList                          = Seq.toList+fromList                        = Seq.fromList

@@ -1,47 +1,67 @@ -- | Distributed types and operations.-module Data.Array.Parallel.Unlifted.Distributed (-  -- * Gang operations-  Gang, forkGang, gangSize,+--+--   * This is an internal API and shouldn't need to be used directly.+--     Client programs should use "Data.Array.Parallel.Unlifted"+--+module Data.Array.Parallel.Unlifted.Distributed +        ( -- * Gang operations+          Gang, forkGang, gangSize -  -- * Gang hacks-  theGang,+          -- * Gang hacks+        , theGang -  -- * Distributed types and classes-  DT(..),+          -- * Distributed types and classes+        , DT(..) -  -- * Higher-order combinators-  mapD, zipWithD, foldD, scanD,+          -- * Higher-order combinators+        , mapD, zipWithD+        , foldD+        , scanD -  -- * Equality-  eqD, neqD,+          -- * Equality+        , eqD, neqD -  -- * Distributed scalars-  scalarD,-  andD, orD, sumD,+          -- * Distributed scalars+        , scalarD+        , andD, orD+        , sumD -  -- * Distributed pairs-  zipD, unzipD, fstD, sndD,+          -- * Distributed pairs+        , zipD, unzipD+        , fstD, sndD -  -- * Distributed arrays-  lengthD, splitLenD, splitLenIdxD,-  splitD, splitAsD, joinLengthD, joinD, splitJoinD, joinDM,-  splitSegdD, splitSegdD', splitSD,-  lengthUSegdD, lengthsUSegdD, indicesUSegdD, elementsUSegdD,-  Distribution, balanced, unbalanced,+          -- * Distributed arrays+        , lengthD+        , splitLenD+        , splitLenIdxD+        , splitD+        , splitAsD+        , joinLengthD+        , joinD+        , splitJoinD+        , joinDM+        , glueSegdD+        , carryD -  -- * Permutations-  permuteD, bpermuteD, atomicUpdateD,+        , Distribution+        , balanced+        , unbalanced -  -- * Debugging-  fromD, toD, debugD-) where+          -- * Permutations+        , permuteD, bpermuteD -import Data.Array.Parallel.Unlifted.Distributed.Gang (-  Gang, forkGang, gangSize)+          -- * Update+        , atomicUpdateD++          -- * Debugging+        , fromD, toD, debugD)+where import Data.Array.Parallel.Unlifted.Distributed.TheGang-import Data.Array.Parallel.Unlifted.Distributed.Types import Data.Array.Parallel.Unlifted.Distributed.Combinators import Data.Array.Parallel.Unlifted.Distributed.Scalars import Data.Array.Parallel.Unlifted.Distributed.Arrays+import Data.Array.Parallel.Unlifted.Distributed.USegd import Data.Array.Parallel.Unlifted.Distributed.Basics+import Data.Array.Parallel.Unlifted.Distributed.Types+import Data.Array.Parallel.Unlifted.Distributed.Gang (Gang, forkGang, gangSize) 

@@ -1,38 +1,40 @@+{-# OPTIONS -Wall -fno-warn-orphans -fno-warn-missing-signatures #-} {-# LANGUAGE EmptyDataDecls, ScopedTypeVariables #-} {-# LANGUAGE CPP #-} #include "fusion-phases.h"  -- | Operations on distributed arrays.-module Data.Array.Parallel.Unlifted.Distributed.Arrays (-  lengthD, splitLenD, splitLenIdxD,-  splitAsD, splitD, joinLengthD, joinD, splitJoinD, joinDM,-  splitSegdD, splitSegdD', splitSD,+module Data.Array.Parallel.Unlifted.Distributed.Arrays +        ( -- * Distribution phantom parameter+          Distribution, balanced, unbalanced -  permuteD, bpermuteD, atomicUpdateD,+         -- * Array Lengths+        , lengthD, splitLenD, splitLenIdxD -  Distribution, balanced, unbalanced-) where-import Data.Array.Parallel.Base ( ST, runST)-import Data.Array.Parallel.Unlifted.Sequential.Vector as Seq-import Data.Array.Parallel.Unlifted.Sequential.Segmented-import Data.Array.Parallel.Unlifted.Distributed.Gang (-  Gang, gangSize, seqGang)-import Data.Array.Parallel.Unlifted.Distributed.DistST (-  DistST, stToDistST, myIndex )-import Data.Array.Parallel.Unlifted.Distributed.Types (-  DT, Dist, mkDPrim, indexD, lengthD, newD, writeMD, zipD, unzipD, fstD, sndD,-  elementsUSegdD,-  checkGangD)-import Data.Array.Parallel.Unlifted.Distributed.Basics-import Data.Array.Parallel.Unlifted.Distributed.Combinators-import Data.Array.Parallel.Unlifted.Distributed.Scalars (-  sumD)+         -- * Splitting and joining+        , splitAsD, splitD, joinLengthD, joinD, splitJoinD, joinDM -import Data.Bits ( shiftR )-import Control.Monad ( when )+         -- * Permutations+        , permuteD, bpermuteD -import GHC.Base ( quotInt, remInt )+          -- * Update+        , atomicUpdateD +          -- * Carry+        , carryD)+ where+import Data.Array.Parallel.Base (ST, runST)+import Data.Array.Parallel.Unlifted.Distributed.Gang+import Data.Array.Parallel.Unlifted.Distributed.DistST+import Data.Array.Parallel.Unlifted.Distributed.Types+import Data.Array.Parallel.Unlifted.Distributed.Combinators+import Data.Array.Parallel.Unlifted.Distributed.Scalars+import Data.Array.Parallel.Unlifted.Sequential.Vector   (Vector, MVector, Unbox)+import qualified Data.Array.Parallel.Unlifted.Sequential.Vector as Seq+import GHC.Base      ( quotInt, remInt )+import Control.Monad++here :: String -> String here s = "Data.Array.Parallel.Unlifted.Distributed.Arrays." Prelude.++ s  @@ -43,35 +45,47 @@ data Distribution  balanced :: Distribution+balanced   = error $ here "balanced: touched" {-# NOINLINE balanced #-}-balanced = error $ here "balanced: touched" + unbalanced :: Distribution-{-# NOINLINE unbalanced #-} unbalanced = error $ here "unbalanced: touched"+{-# NOINLINE unbalanced #-}   -- Splitting and Joining array lengths ------------------------------------------- | Distribute an array length over a 'Gang'.+-- | O(threads).+--   Distribute an array length over a 'Gang'. --   Each thread holds the number of elements it's reponsible for.+--   If the array length doesn't split evenly among the threads then the first+--   threads get a few more elements.+--+--   @splitLenD theGangN4 511+--      = [128,128,128,127]@+--  splitLenD :: Gang -> Int -> Dist Int-{-# INLINE splitLenD #-} splitLenD g n = generateD_cheap g len   where     !p = gangSize g     !l = n `quotInt` p-    !m = n `remInt` p+    !m = n `remInt`  p      {-# INLINE [0] len #-}     len i | i < m     = l+1           | otherwise = l+{-# INLINE splitLenD #-}  --- | Distribute an array length over a 'Gang'.+-- | O(threads).+--   Distribute an array length over a 'Gang'. --   Each thread holds the number of elements it's responsible for,  --   and the index of the start of its chunk.-splitLenIdxD :: Gang -> Int -> Dist (Int,Int)-{-# INLINE splitLenIdxD #-}+--+--   @splitLenIdxD theGangN4 511 +--      = [(128,0),(128,128),(128,256),(127,384)]@+--+splitLenIdxD :: Gang -> Int -> Dist (Int, Int) splitLenIdxD g n = generateD_cheap g len_idx   where     !p = gangSize g@@ -81,38 +95,48 @@     {-# INLINE [0] len_idx #-}     len_idx i | i < m     = (l+1, i*(l+1))               | otherwise = (l,   i*l + m)+{-# INLINE splitLenIdxD #-}  --- | Get the overall length of a distributed array.---   We ask each thread for its chunk length, and sum them all up.+-- | O(threads).+--   Get the overall length of a distributed array.+--   This is implemented by reading the chunk length from each thread, +--   and summing them up. joinLengthD :: Unbox a => Gang -> Dist (Vector a) -> Int-{-# INLINE joinLengthD #-} joinLengthD g = sumD g . lengthD+{-# INLINE joinLengthD #-}                                                  -- Splitting and Joining arrays ----------------------------------------------- -- | Distribute an array over a 'Gang' such that each threads gets the given --   number of elements.+--+--   @splitAsD theGangN4 (splitLenD theGangN4 10) [1 2 3 4 5 6 7 8 9 0]+--      = [[1 2 3] [4 5 6] [7 8] [9 0]]@+--  splitAsD :: Unbox a => Gang -> Dist Int -> Vector a -> Dist (Vector a)-{-# INLINE_DIST splitAsD #-}-splitAsD g dlen !arr = zipWithD (seqGang g) (Seq.slice arr) is dlen+splitAsD g dlen !arr +  = zipWithD (seqGang g) (Seq.slice "splitAsD" arr) is dlen   where     is = fst $ scanD g (+) 0 dlen+{-# INLINE_DIST splitAsD #-}   -- | Distribute an array over a 'Gang'. -- --   NOTE: This is defined in terms of splitD_impl to avoid introducing loops --         through RULES. Without it, splitJoinD would be a loop breaker.+--  splitD :: Unbox a => Gang -> Distribution -> Vector a -> Dist (Vector a)-{-# INLINE_DIST splitD #-} splitD g _ arr = splitD_impl g arr+{-# INLINE_DIST splitD #-} + splitD_impl :: Unbox a => Gang -> Vector a -> Dist (Vector a)-{-# INLINE_DIST splitD_impl #-}-splitD_impl g !arr = generateD_cheap g (\i -> Seq.slice arr (idx i) (len i))+splitD_impl g !arr +  = generateD_cheap g (\i -> Seq.slice "splitD_impl" arr (idx i) (len i))   where-    n = Seq.length arr+    n  = Seq.length arr     !p = gangSize g     !l = n `quotInt` p     !m = n `remInt` p@@ -124,24 +148,31 @@     {-# INLINE [0] len #-}     len i | i < m     = l+1           | otherwise = l+{-# INLINE_DIST splitD_impl #-}   -- | Join a distributed array.+--   Join sums up the array lengths of each chunk, allocates a new result array, +--   and copies each chunk into the result. -- --   NOTE: This is defined in terms of joinD_impl to avoid introducing loops --         through RULES. Without it, splitJoinD would be a loop breaker.+-- joinD :: Unbox a => Gang -> Distribution -> Dist (Vector a) -> Vector a-{-# INLINE CONLIKE [1] joinD #-} joinD g _ darr  = joinD_impl g darr+{-# INLINE CONLIKE [1] joinD #-} + joinD_impl :: forall a. Unbox a => Gang -> Dist (Vector a) -> Vector a-{-# INLINE_DIST joinD_impl #-}-joinD_impl g !darr = checkGangD (here "joinD") g darr $-                     Seq.new n (\ma -> zipWithDST_ g (copy ma) di darr)+joinD_impl g !darr +  = checkGangD (here "joinD") g darr +  $ Seq.new n (\ma -> zipWithDST_ g (copy ma) di darr)   where-    (!di,!n) = scanD g (+) 0 $ lengthD darr+    (!di,!n)      = scanD g (+) 0 $ lengthD darr+     copy :: forall s. MVector s a -> Int -> Vector a -> DistST s ()-    copy ma i arr = stToDistST (Seq.copy (mslice i (Seq.length arr) ma) arr)+    copy ma i arr = stToDistST (Seq.copy (Seq.mslice i (Seq.length arr) ma) arr)+{-# INLINE_DIST joinD_impl #-}   -- | Split a vector over a gang, run a distributed computation, then@@ -152,23 +183,24 @@         -> (Dist (Vector a) -> Dist (Vector b))         -> Vector a         -> Vector b+splitJoinD g f !xs +  = joinD_impl g (f (splitD_impl g xs)) {-# INLINE_DIST splitJoinD #-}-splitJoinD g f !xs = joinD_impl g (f (splitD_impl g xs))    -- | Join a distributed array, yielding a mutable global array joinDM :: Unbox a => Gang -> Dist (Vector a) -> ST s (MVector s a)+joinDM g darr + = checkGangD (here "joinDM") g darr + $ do   marr <- Seq.newM n+        zipWithDST_ g (copy marr) di darr+        return marr+ where+        (!di,!n) = scanD g (+) 0 $ lengthD darr++        copy ma i arr = stToDistST (Seq.copy (Seq.mslice i (Seq.length arr) ma) arr) {-# INLINE joinDM #-}-joinDM g darr = checkGangD (here "joinDM") g darr $-                do-                  marr <- Seq.newM n-                  zipWithDST_ g (copy marr) di darr-                  return marr-  where-    (!di,!n) = scanD g (+) 0 $ lengthD darr-    ---    copy ma i arr = stToDistST (Seq.copy (mslice i (Seq.length arr) ma) arr)   {-# RULES@@ -211,20 +243,24 @@  -- Permutation ---------------------------------------------------------------- -- | Permute for distributed arrays.-permuteD :: forall a. Unbox a => Gang -> Dist (Vector a) -> Dist (Vector Int) -> Vector a-{-# INLINE_DIST permuteD #-}-permuteD g darr dis = Seq.new n (\ma -> zipWithDST_ g (permute ma) darr dis)+permuteD +        :: forall a. Unbox a +        => Gang -> Dist (Vector a) -> Dist (Vector Int) -> Vector a+permuteD g darr dis +  = Seq.new n (\ma -> zipWithDST_ g (permute ma) darr dis)   where     n = joinLengthD g darr+     permute :: forall s. MVector s a -> Vector a -> Vector Int -> DistST s ()     permute ma arr is = stToDistST (Seq.mpermute ma arr is)+{-# INLINE_DIST permuteD #-}   -- NOTE: The bang is necessary because the array must be fully evaluated -- before we pass it to the parallel computation. bpermuteD :: Unbox a => Gang -> Vector a -> Dist (Vector Int) -> Dist (Vector a)-{-# INLINE bpermuteD #-} bpermuteD g !as ds = mapD g (Seq.bpermute as) ds+{-# INLINE bpermuteD #-}   -- Update ---------------------------------------------------------------------@@ -233,119 +269,99 @@ -- error. atomicUpdateD :: forall a. Unbox a              => Gang -> Dist (Vector a) -> Dist (Vector (Int,a)) -> Vector a+atomicUpdateD g darr upd + = runST + $ do   marr <- joinDM g darr+        mapDST_ g (update marr) upd+        Seq.unsafeFreeze marr+ where+        update :: forall s. MVector s a -> Vector (Int,a) -> DistST s ()+        update marr arr = stToDistST (Seq.mupdate marr arr) {-# INLINE atomicUpdateD #-}-atomicUpdateD g darr upd = runST (-  do-    marr <- joinDM g darr-    mapDST_ g (update marr) upd-    Seq.unsafeFreeze marr-  )-  where-    update :: forall s. MVector s a -> Vector (Int,a) -> DistST s ()-    update marr arr = stToDistST (Seq.mupdate marr arr)  ---- Splitting and Joining segment descriptors ----------------------------------splitSegdD :: Gang -> USegd -> Dist USegd-{-# NOINLINE splitSegdD #-}-splitSegdD g !segd = mapD g lengthsToUSegd-                   $ splitAsD g d lens-  where-    !d = snd-       . mapAccumLD g chunk 0-       . splitLenD g-       $ elementsUSegd segd--    n    = lengthUSegd segd-    lens = lengthsUSegd segd--    chunk !i !k = let !j = go i k-                  in (j,j-i)--    go !i !k | i >= n    = i-             | m == 0    = go (i+1) k-             | k <= 0    = i-             | otherwise = go (i+1) (k-m)-      where-        m = lens ! i---search :: Int -> Vector Int -> Int-search !x ys = go 0 (Seq.length ys)-  where-    go i n | n <= 0        = i-           | (ys!mid) < x = go (mid+1) (n-half-1)-           | otherwise     = go i half-      where-        half = n `shiftR` 1-        mid  = i + half---chunk :: USegd -> Int -> Int -> Bool -> (# Vector Int, Int, Int #)-chunk !segd !di !dn is_last-  = (# lens', k-left_len, left_off #)-  where-    !lens' = runST (do-                      mlens' <- Seq.newM n'-                      when (left /= 0) $ Seq.write mlens' 0 left-                      Seq.copy (Seq.mdrop left_len mlens')-                               (Seq.slice lens k (k'-k))-                      when (right /= 0) $ Seq.write mlens' (n' - 1) right-                      Seq.unsafeFreeze mlens')--    lens = lengthsUSegd segd-    idxs = indicesUSegd segd-    n    = Seq.length lens--    k  = search di idxs-    k' | is_last   = n-       | otherwise = search (di+dn) idxs--    left  | k == n    = dn-          | otherwise = min ((idxs!k) - di) dn--    right | k' == k   = 0-          | otherwise = di + dn - (idxs ! (k'-1))--    left_len | left == 0   = 0-             | otherwise   = 1--    left_off | left == 0   = 0-             | otherwise   = di - idxs ! (k-1)+-- Carry ----------------------------------------------------------------------+-- | Selectively combine the last elements of some chunks with the+--   first elements of others.+--+--   NOTE: This runs sequentially and should only be used for testing purposes.+--+-- @+-- pprp $ splitD theGang unbalanced $ fromList [80, 10, 20, 40, 50, 10 :: Int]+-- DVector lengths: [2,2,1,1]+--         chunks:  [[80,10],[20,40],[50],[10]]+-- +--  pprp $ fst +--       $ carryD theGang (+) 0 +--          (mkDPrim $ fromList [True, False, True, False]) +--          (splitD theGang unbalanced $ fromList [80, 10, 20, 40, 50, 10 :: Int])+--+--  DVector lengths: [1,2,0,1]+--          chunks: [[80],[30,40],[],[60]]+-- @+--+carryD  :: forall a+        .  (Unbox a, DT a)+        => Gang +        -> (a -> a -> a) -> a+        -> Dist Bool+        -> Dist (Vector a)+        -> (Dist (Vector a), a) -    n' = left_len + (k'-k)+carryD gang f zero shouldCarry vec+ = runST + $ do   md      <- newMD gang+        acc     <- carryD' f zero shouldCarry vec md+        d       <- unsafeFreezeMD md+        return (d, acc)  -splitSegdD' :: Gang -> USegd -> Dist ((USegd,Int),Int)-{-# INLINE splitSegdD' #-}-splitSegdD' g !segd = imapD g mk-                         (splitLenIdxD g-                         (elementsUSegd segd))-  where-    !p = gangSize g--    mk i (dn,di) = case chunk segd di dn (i == p-1) of-                     (# lens, l, o #) -> ((lengthsToUSegd lens,l),o)+carryD' :: forall a s+        .  (Unbox a, DT a)+        => (a -> a -> a) -> a+        -> Dist Bool+        -> Dist (Vector a)+        -> MDist (Vector a) s+        -> ST s a +carryD' f zero shouldCarry vec md_+ = go md_ zero 0+ where go (md :: MDist (Vector a) s) prev ix+        | ix >= sizeD vec    = return prev+        | otherwise+        = do let chunk :: Vector a+                 !chunk      = indexD (here "carryD'") vec ix+             let !chunkLen   = Seq.length chunk -joinSegD :: Gang -> Dist USegd -> USegd-{-# INLINE_DIST joinSegD #-}-joinSegD g = lengthsToUSegd-           . joinD g unbalanced-           . mapD g lengthsUSegd+             -- Whether to carry the last value of this chunk into the next chunk+             let !carry      = indexD (here "carryD") shouldCarry ix +             -- The new length for this chunk+             let !chunkLen'  +                   | chunkLen == 0 = 0+                   | carry         = chunkLen - 1+                   | otherwise     = chunkLen -splitSD :: Unbox a => Gang -> Dist USegd -> Vector a -> Dist (Vector a)-{-# INLINE_DIST splitSD #-}-splitSD g dsegd xs = splitAsD g (elementsUSegdD dsegd) xs+             -- The new value of the accumulator+             let acc            = f prev (Seq.index (here "carryD'") chunk 0)+                +             -- Allocate a mutable vector to hold the new chunk and copy+             -- source elements into it.+             mchunk' <- Seq.newM chunkLen'+             Seq.copy mchunk' (Seq.slice (here "carryD'") chunk 0 chunkLen') -{-# RULES+             when (chunkLen' /= 0)+              $ Seq.write mchunk' 0 acc -"splitSD/splitJoinD" forall g d f xs.-  splitSD g d (splitJoinD g f xs) = f (splitSD g d xs)+             -- Store the new chunk in the gang+             chunk'  <- Seq.unsafeFreeze mchunk'+             writeMD md ix chunk' -"splitSD/Seq.zip" forall g d xs ys.-  splitSD g d (Seq.zip xs ys) = zipWithD g Seq.zip (splitSD g d xs)-                                             (splitSD g d ys)+             -- What value to carry into the next chunk+             let next+                  | chunkLen' == 0      = acc+                  | carry               = Seq.index (here "next") chunk (chunkLen - 1)+                  | otherwise           = zero+                               +             go md next (ix + 1) -  #-}

@@ -1,44 +1,46 @@+{-# OPTIONS -Wall -fno-warn-orphans #-}+ -- | Basic operations on distributed types.-module Data.Array.Parallel.Unlifted.Distributed.Basics (-  eqD, neqD, toD, fromD-) where-import Data.Array.Parallel.Unlifted.Distributed.Gang (-  Gang, gangSize)-import Data.Array.Parallel.Unlifted.Distributed.Types (-  DT, Dist, indexD, newD, writeMD,-  checkGangD)-import Data.Array.Parallel.Unlifted.Distributed.Combinators (-  zipWithD)-import Data.Array.Parallel.Unlifted.Distributed.Scalars (-  andD, orD)+module Data.Array.Parallel.Unlifted.Distributed.Basics +        (eqD, neqD, toD, fromD)+where+import Data.Array.Parallel.Unlifted.Distributed.Gang +import Data.Array.Parallel.Unlifted.Distributed.Types+import Data.Array.Parallel.Unlifted.Distributed.Combinators +import Data.Array.Parallel.Unlifted.Distributed.Scalars import Control.Monad ( zipWithM_ ) -+here :: String -> String here s = "Data.Array.Parallel.Unlifted.Distributed.Basics." ++ s   -- | Test whether to distributed values are equal.  --   This requires a 'Gang' and hence can't be defined in terms of 'Eq'. eqD :: (Eq a, DT a) => Gang -> Dist a -> Dist a -> Bool-eqD g dx dy = andD g (zipWithD g (==) dx dy)+eqD g dx dy +        = andD g (zipWithD g (==) dx dy)   -- | Test whether to distributed values are not equal. --   This requires a 'Gang' and hence can't be defined in terms of 'Eq'. neqD :: (Eq a, DT a) => Gang -> Dist a -> Dist a -> Bool-neqD g dx dy = orD g (zipWithD g (/=) dx dy)-+neqD g dx dy +        = orD g (zipWithD g (/=) dx dy)   -- | Generate a distributed value from the first @p@ elements of a list.---   /NOTE:/ For debugging only, don't use in production code.+-- +--   * For debugging only, don't use in production code. toD :: DT a => Gang -> [a] -> Dist a toD g xs = newD g (\md -> zipWithM_ (writeMD md) [0 .. gangSize g - 1] xs)   -- | Yield all elements of a distributed value.---   /NOTE:/ For debugging only, don't use in production code.+--+--   * For debugging only, don't use in production code. fromD :: DT a => Gang -> Dist a -> [a]-fromD g dt = checkGangD (here "fromDT") g dt $-             map (indexD dt) [0 .. gangSize g - 1]+fromD g dt +        = checkGangD (here "fromDT") g dt +        $ map   (indexD (here "fromD") dt) +                [0 .. gangSize g - 1] 

@@ -1,42 +1,49 @@+{-# OPTIONS -Wall -fno-warn-orphans -fno-warn-missing-signatures #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE CPP #-} #include "fusion-phases.h"  -- | Standard combinators for distributed types.-module Data.Array.Parallel.Unlifted.Distributed.Combinators (-  generateD, generateD_cheap,-  imapD, mapD, zipD, unzipD, fstD, sndD, zipWithD, izipWithD,-  foldD, scanD, mapAccumLD,+module Data.Array.Parallel.Unlifted.Distributed.Combinators +        ( generateD, generateD_cheap+        , imapD, mapD+        , zipD, unzipD+        , fstD, sndD+        , zipWithD, izipWithD+        , foldD+        , scanD+        , mapAccumLD -  -- * Monadic combinators-  mapDST_, mapDST, zipWithDST_, zipWithDST-) where+           -- * Monadic combinators+        , mapDST_, mapDST, zipWithDST_, zipWithDST)+where import Data.Array.Parallel.Base ( ST, runST)-import Data.Array.Parallel.Unlifted.Distributed.Gang (-  Gang, gangSize)-import Data.Array.Parallel.Unlifted.Distributed.Types (-  DT, Dist, MDist, indexD, zipD, unzipD, fstD, sndD, deepSeqD,-  newMD, writeMD, unsafeFreezeMD,-  checkGangD, measureD, debugD)+import Data.Array.Parallel.Unlifted.Distributed.Gang+import Data.Array.Parallel.Unlifted.Distributed.Types import Data.Array.Parallel.Unlifted.Distributed.DistST-import Debug.Trace   here s = "Data.Array.Parallel.Unlifted.Distributed.Combinators." ++ s ---- | Create a distributed value, given a function that makes the value in each thread.+-- | Create a distributed value, given a function to create the instance+--   for each thread. generateD :: DT a => Gang -> (Int -> a) -> Dist a-{-# NOINLINE generateD #-} generateD g f          = runDistST g (myIndex >>= return . f)+{-# NOINLINE generateD #-}  --- | Create a distributed value, but run it sequentially (I think?)+-- | Create a distributed value, but do it sequentially.+--  +--   This function is used when we want to operate on a distributed value, but+--   there isn't much data involved. For example, if we want to distribute +--   a single integer to each thread, then there's no need to fire up the +--   gang for this.+--    generateD_cheap :: DT a => Gang -> (Int -> a) -> Dist a-{-# NOINLINE generateD_cheap #-} generateD_cheap g f          = runDistST_seq g (myIndex >>= return . f)+{-# NOINLINE generateD_cheap #-}   -- Mapping --------------------------------------------------------------------@@ -45,26 +52,34 @@ --   As opposed to `imapD'` this version also deepSeqs each element before --   passing it to the function. imapD :: (DT a, DT b) => Gang -> (Int -> a -> b) -> Dist a -> Dist b-{-# INLINE [0] imapD #-} imapD g f d = imapD' g (\i x -> x `deepSeqD` f i x) d+{-# INLINE [0] imapD #-}   -- | Map a function across all elements of a distributed value. --   The worker function also gets the current thread index. imapD' :: (DT a, DT b) => Gang -> (Int -> a -> b) -> Dist a -> Dist b+imapD' g f !d +  = checkGangD (here "imapD") g d+  $ runDistST g +        (do i <- myIndex+            x <- myD d+            return (f i x)) {-# NOINLINE imapD' #-}-imapD' g f !d = checkGangD (here "imapD") g d-                (runDistST g (do-                                i <- myIndex-                                x <- myD d-                                return (f i x)))  --- | Map a function over a distributed value.+-- | Map a function to every instance of a distributed value.+--+--   This applies the function to every thread, but not every value held+--   by the thread. If you want that then use something like:+-- +--   @mapD theGang (V.map (+ 1)) :: Dist (Vector Int) -> Dist (Vector Int)@+-- mapD :: (DT a, DT b) => Gang -> (a -> b) -> Dist a -> Dist b-{-# INLINE mapD #-} mapD g = imapD g . const+{-# INLINE mapD #-} + {-# RULES  "imapD/generateD" forall gang f g.@@ -83,17 +98,18 @@ -- | Combine two distributed values with the given function. zipWithD :: (DT a, DT b, DT c)          => Gang -> (a -> b -> c) -> Dist a -> Dist b -> Dist c-{-# INLINE zipWithD #-} zipWithD g f dx dy = mapD g (uncurry f) (zipD dx dy)+{-# INLINE zipWithD #-}   -- | Combine two distributed values with the given function. --   The worker function also gets the index of the current thread. izipWithD :: (DT a, DT b, DT c)           => Gang -> (Int -> a -> b -> c) -> Dist a -> Dist b -> Dist c-{-# INLINE izipWithD #-} izipWithD g f dx dy = imapD g (\i -> uncurry (f i)) (zipD dx dy)+{-# INLINE izipWithD #-} + {-# RULES "zipD/imapD[1]" forall gang f xs ys.   zipD (imapD gang f xs) ys@@ -115,54 +131,66 @@   -- Folding ----------------------------------------------------------------------- | Fold a distributed value.+-- | Fold all the instances of a distributed value. foldD :: DT a => Gang -> (a -> a -> a) -> Dist a -> a-{-# NOINLINE foldD #-}-foldD g f !d = checkGangD ("here foldD") g d $-              fold 1 (d `indexD` 0)+foldD g f !d +  = checkGangD ("here foldD") g d +  $ fold 1 (indexD (here "foldD") d 0)   where     !n = gangSize g     --     fold i x | i == n    = x-             | otherwise = fold (i+1) (f x $ d `indexD` i)+             | otherwise = fold (i+1) (f x $ indexD (here "foldD") d i)+{-# NOINLINE foldD #-}  --- | Prefix sum of a distributed value.+-- | Prefix sum of the instances of a distributed value. scanD :: forall a. DT a => Gang -> (a -> a -> a) -> a -> Dist a -> (Dist a, a)-{-# NOINLINE scanD #-}-scanD g f z !d = checkGangD (here "scanD") g d $-                 runST (do-                   md <- newMD g-                   s  <- scan md 0 z-                   d' <- unsafeFreezeMD md-                   return (d',s))+scanD g f z !d+  = checkGangD (here "scanD") g d +  $ runST (do+          md <- newMD g+          s  <- scan md 0 z+          d' <- unsafeFreezeMD md+          return (d',s))   where     !n = gangSize g+         scan :: forall s. MDist a s -> Int -> a -> ST s a-    scan md i !x | i == n    = return x-                 | otherwise = do-                                 writeMD md i x-                                 scan md (i+1) (f x $ d `indexD` i)+    scan md i !x+        | i == n    = return x+        | otherwise+        = do    writeMD md i x+                scan md (i+1) (f x $ indexD (here "scanD") d i)+{-# NOINLINE scanD #-} + -- | Combination of map and fold.-mapAccumLD :: forall a b acc. (DT a, DT b)-           => Gang -> (acc -> a -> (acc,b))-                   -> acc -> Dist a -> (acc,Dist b)-{-# INLINE_DIST mapAccumLD #-}-mapAccumLD g f acc !d = checkGangD (here "mapAccumLD") g d $-                        runST (do-                          md   <- newMD g-                          acc' <- go md 0 acc-                          d'   <- unsafeFreezeMD md-                          return (acc',d'))+mapAccumLD +        :: forall a b acc. (DT a, DT b)+        => Gang+        -> (acc -> a      -> (acc, b))+        ->  acc -> Dist a -> (acc, Dist b)++mapAccumLD g f acc !d+  = checkGangD (here "mapAccumLD") g d +  $ runST (do+        md   <- newMD g+        acc' <- go md 0 acc+        d'   <- unsafeFreezeMD md+        return (acc',d'))   where     !n = gangSize g+         go :: MDist b s -> Int -> acc -> ST s acc-    go md i acc | i == n    = return acc-                | otherwise = case f acc (d `indexD` i) of-                                (acc',b) -> do-                                              writeMD md i b-                                              go md (i+1) acc'+    go md i acc'+        | i == n    = return acc'+        | otherwise+        = case f acc' (indexD (here "mapAccumLD") d i) of+                (acc'',b) -> do+                      writeMD md i b+                      go md (i+1) acc''+{-# INLINE_DIST mapAccumLD #-}                                   -- Versions that work on DistST -----------------------------------------------@@ -172,33 +200,41 @@ -- model andlead to a deadlock. Hence the bangs.  mapDST_ :: DT a => Gang -> (a -> DistST s ()) -> Dist a -> ST s ()+mapDST_ g p d + = mapDST_' g (\x -> x `deepSeqD` p x) d {-# INLINE mapDST_ #-}-mapDST_ g p d = mapDST_' g (\x -> x `deepSeqD` p x) d   mapDST_' :: DT a => Gang -> (a -> DistST s ()) -> Dist a -> ST s ()-mapDST_' g p !d = checkGangD (here "mapDST_") g d $-                  distST_ g (myD d >>= p)+mapDST_' g p !d + = checkGangD (here "mapDST_") g d + $ distST_ g (myD d >>= p)   mapDST :: (DT a, DT b) => Gang -> (a -> DistST s b) -> Dist a -> ST s (Dist b)-{-# INLINE mapDST #-} mapDST g p !d = mapDST' g (\x -> x `deepSeqD` p x) d+{-# INLINE mapDST #-} + mapDST' :: (DT a, DT b) => Gang -> (a -> DistST s b) -> Dist a -> ST s (Dist b)-mapDST' g p !d = checkGangD (here "mapDST_") g d $-                 distST g (myD d >>= p)+mapDST' g p !d + = checkGangD (here "mapDST_") g d + $ distST g (myD d >>= p)  -zipWithDST_ :: (DT a, DT b)-            => Gang -> (a -> b -> DistST s ()) -> Dist a -> Dist b -> ST s ()+zipWithDST_ +        :: (DT a, DT b)+        => Gang -> (a -> b -> DistST s ()) -> Dist a -> Dist b -> ST s ()+zipWithDST_ g p !dx !dy + = mapDST_ g (uncurry p) (zipD dx dy) {-# INLINE zipWithDST_ #-}-zipWithDST_ g p !dx !dy = mapDST_ g (uncurry p) (zipD dx dy)  -zipWithDST :: (DT a, DT b, DT c)-           => Gang-           -> (a -> b -> DistST s c) -> Dist a -> Dist b -> ST s (Dist c)+zipWithDST +        :: (DT a, DT b, DT c)+        => Gang+        -> (a -> b -> DistST s c) -> Dist a -> Dist b -> ST s (Dist c)+zipWithDST g p !dx !dy + = mapDST g (uncurry p) (zipD dx dy) {-# INLINE zipWithDST #-}-zipWithDST g p !dx !dy = mapDST g (uncurry p) (zipD dx dy) 

@@ -1,3 +1,4 @@+{-# OPTIONS -Wall -fno-warn-orphans -fno-warn-missing-signatures #-} {-# LANGUAGE ScopedTypeVariables #-} -- | Distributed ST computations. --@@ -7,15 +8,19 @@ --  thread. -- -- /TODO:/ Add facilities for implementing parallel scans etc.-module Data.Array.Parallel.Unlifted.Distributed.DistST (-  DistST, stToDistST, distST_, distST, runDistST, runDistST_seq, traceDistST,-  myIndex, myD, readMyMD, writeMyMD-) where-import Data.Array.Parallel.Base (-  ST, runST)+module Data.Array.Parallel.Unlifted.Distributed.DistST +        ( DistST+        , stToDistST+        , distST_, distST+        , runDistST, runDistST_seq+        , traceDistST+        , myIndex+        , myD+        , readMyMD, writeMyMD)+where+import Data.Array.Parallel.Base (ST, runST) import Data.Array.Parallel.Unlifted.Distributed.Gang-import Data.Array.Parallel.Unlifted.Distributed.Types (-  DT(..), Dist, MDist)+import Data.Array.Parallel.Unlifted.Distributed.Types (DT(..), Dist, MDist)  import Control.Monad (liftM) @@ -38,62 +43,62 @@  -- | Yields the index of the current thread within its gang. myIndex :: DistST s Int-{-# INLINE myIndex #-} myIndex = DistST return+{-# INLINE myIndex #-}   -- | Lifts an 'ST' computation into the 'DistST' monad. --   The lifted computation should be data parallel. stToDistST :: ST s a -> DistST s a+stToDistST p = DistST $ \_ -> p {-# INLINE stToDistST #-}-stToDistST p = DistST $ \i -> p   -- | Yields the 'Dist' element owned by the current thread. myD :: DT a => Dist a -> DistST s a+myD dt = liftM (indexD "myD" dt) myIndex {-# NOINLINE myD #-}-myD dt = liftM (indexD dt) myIndex   -- | Yields the 'MDist' element owned by the current thread. readMyMD :: DT a => MDist a s -> DistST s a-{-# NOINLINE readMyMD #-} readMyMD mdt   = do	i <- myIndex 	stToDistST $ readMD mdt i+{-# NOINLINE readMyMD #-}   -- | Writes the 'MDist' element owned by the current thread. writeMyMD :: DT a => MDist a s -> a -> DistST s ()-{-# NOINLINE writeMyMD #-} writeMyMD mdt x   = do	i <- myIndex 	stToDistST $ writeMD mdt i x+{-# NOINLINE writeMyMD #-}   -- | Execute a data-parallel computation on a 'Gang'. --   The same DistST comutation runs on each thread. distST_ :: Gang -> DistST s () -> ST s ()-{-# INLINE distST_ #-} distST_ g = gangST g . unDistST+{-# INLINE distST_ #-}   -- | Execute a data-parallel computation, yielding the distributed result. distST :: DT a => Gang -> DistST s a -> ST s (Dist a)-{-# INLINE distST #-} distST g p   = do	md <- newMD g         distST_ g $ writeMyMD md =<< p         unsafeFreezeMD md+{-# INLINE distST #-}   -- | Run a data-parallel computation, yielding the distributed result. runDistST :: DT a => Gang -> (forall s. DistST s a) -> Dist a-{-# NOINLINE runDistST #-} runDistST g p = runST (distST g p)+{-# NOINLINE runDistST #-} + runDistST_seq :: forall a. DT a => Gang -> (forall s. DistST s a) -> Dist a-{-# NOINLINE runDistST_seq #-} runDistST_seq g p = runST (   do      md <- newMD g@@ -106,6 +111,8 @@                             writeMD md i =<< unDistST p i                             go md (i+1)             | otherwise = return ()+{-# NOINLINE runDistST_seq #-}+  traceDistST :: String -> DistST s () traceDistST s = DistST $ \n -> traceGangST ("Worker " ++ show n ++ ": " ++ s)

@@ -1,32 +1,33 @@+{-# OPTIONS -Wall -fno-warn-orphans -fno-warn-missing-signatures #-} {-# LANGUAGE CPP #-} --- | Gang primitives.------ /TODO:/------ * Implement busy waiting.------ * Benchmark.------ * Generalise thread indices?-+-- If a work request is sent to the gang while another is already running+-- then just run it sequentially instead of dying. #define SEQ_IF_GANG_BUSY 1-#define TRACE_GANG 0 -module Data.Array.Parallel.Unlifted.Distributed.Gang (-  Gang, seqGang, forkGang, gangSize, gangIO, gangST, traceGang, traceGangST -) where+-- Trace all work requests sent to the gang.+#define TRACE_GANG 0 +-- | Gang primitives.+module Data.Array.Parallel.Unlifted.Distributed.Gang +        ( Gang+        , seqGang+        , forkGang+        , gangSize+        , gangIO, gangST+        , traceGang, traceGangST )+where import GHC.IO import GHC.ST-import GHC.Conc                  ( forkOn )-import GHC.Exts                  ( traceEvent )-+import Control.Concurrent        (forkOn) import Control.Concurrent.MVar-import Control.Exception         ( assert )-import Control.Monad             ( zipWithM, zipWithM_ )+import Control.Exception         (assert)+import Control.Monad +#if TRACE_GANG+import GHC.Exts                  (traceEvent) import System.Time ( ClockTime(..), getClockTime )+#endif   -- Requests and operations on them -------------------------------------------- -- | The 'Req' type encapsulates work requests for individual members of a gang. @@ -35,9 +36,9 @@ 	--   by writing to the MVar. 	= ReqDo	       (Int -> IO ()) (MVar ()) -	-- | Tell the worker that we're shutting the gang down. The worker should-	--   signal that it's received the equest down by writing to the MVar before-	--   returning to its caller (forkGang) 	+	-- | Tell the worker that we're shutting the gang down.+        --   The worker should signal that it's received the equest down by+        --   writing to the MVar before returning to its caller (forkGang) 	 	| ReqShutdown  (MVar ())  @@ -53,17 +54,16 @@ waitReq :: Req -> IO () waitReq req  = case req of-	ReqDo     fn varDone	-> takeMVar varDone+	ReqDo     _ varDone	-> takeMVar varDone 	ReqShutdown varDone	-> takeMVar varDone   -- Thread gangs and operations on them ---------------------------------------- -- | A 'Gang' is a group of threads which execute arbitrary work requests.---   To get the gang to do work, write Req-uest values---   to its MVars-data Gang = Gang !Int           -- Number of 'Gang' threads-                 [MVar Req]     -- One 'MVar' per thread-                 (MVar Bool)    -- Indicates whether the 'Gang' is busy+data Gang +        = Gang !Int           -- Number of 'Gang' threads+               [MVar Req]     -- One 'MVar' per thread+               (MVar Bool)    -- Indicates whether the 'Gang' is busy   instance Show Gang where@@ -91,7 +91,8 @@ 		start 	<- getGangTime 		action threadId 		end 	<- getGangTime-		traceGang $ "Worker " ++ show threadId ++ " end (" ++ diffTime start end ++ ")"+		traceGang $ "Worker " ++ show threadId +                          ++ " end (" ++ diffTime start end ++ ")" 		 		putMVar varDone () 		gangWorker threadId varReq@@ -102,18 +103,19 @@   -- | Finaliser for worker threads.---   We want to shutdown the corresponding thread when it's MVar becomes unreachable.---     Without this Repa programs can complain about "Blocked indefinitely on an MVar"---     because worker threads are still blocked on the request MVars when the program ends.---     Whether the finalizer is called or not is very racey. It happens about 1 in 10 runs---     when for the repa-edgedetect benchmark, and less often with the others.+--   We want to shutdown the corresponding thread when it's MVar becomes+--   unreachable. Without this the program can compilain about +--   "Blocked indefinitely on an MVar" because worker threads are still+--   blocked on the request MVars when the program ends. Whether this finalizer+--   is called or not is very racey. It can happen 1 in 10 times, or less often. --  --   We're relying on the comment in System.Mem.Weak that says---    "If there are no other threads to run, the runtime system will check for runnable---     finalizers before declaring the system to be deadlocked."+--   "If there are no other threads to run, the runtime system will check for+--    runnable finalizers before declaring the system to be deadlocked." -- ---   If we were creating and destroying the gang cleanly we wouldn't need this, but theGang ---     is created with a top-level unsafePerformIO. Hacks beget hacks beget hacks...+--   If we were creating and destroying the gang cleanly we wouldn't need this,+--   but theGang is created with a top-level unsafePerformIO.+--   Hacks beget hacks beget hacks... -- finaliseWorker :: MVar Req -> IO () finaliseWorker varReq@@ -131,7 +133,8 @@ 	-- Create the vars we'll use to issue work requests. 	mvs	<- sequence . replicate n $ newEmptyMVar 	-	-- Add finalisers so we can shut the workers down cleanly if they become unreachable.+	-- Add finalisers so we can shut the workers down cleanly if they+        -- become unreachable. 	mapM_ (\var -> addMVarFinalizer var (finaliseWorker var)) mvs  	-- Create all the worker threads@@ -144,22 +147,21 @@ 	return $ Gang n mvs busy  --- | The number of threads in the 'Gang'.+-- | O(1). Yield the number of threads in the 'Gang'. gangSize :: Gang -> Int gangSize (Gang n _ _) = n   -- | Issue work requests for the 'Gang' and wait until they have been executed.---   If the gang is already busy then just run the action in the---   requesting thread. ------   TODO: We might want to print a configurable warning that this is happening.---+--   If the gang is already busy then just run the action in the requesting+--   thread.  gangIO	:: Gang 	-> (Int -> IO ()) 	-> IO () -gangIO (Gang n [] busy)  p = mapM_ p [0 .. n-1]+gangIO (Gang n [] _)  p + = mapM_ p [0 .. n-1]+ #if SEQ_IF_GANG_BUSY gangIO (Gang n mvs busy) p   = do	traceGang   "gangIO: issuing work requests (SEQ_IF_GANG_BUSY)"@@ -170,17 +172,19 @@ 	 then mapM_ p [0 .. n-1] 	 else do 		parIO n mvs p-		swapMVar busy False+		_ <- swapMVar busy False 		return () #else gangIO (Gang n mvs busy) p = parIO n mvs p #endif + -- | Issue some requests to the worker threads and wait for them to complete. parIO 	:: Int			-- ^ Number of threads in the gang. 	-> [MVar Req]		-- ^ Request vars for worker threads.-	-> (Int -> IO ())	-- ^ Action to run in all the workers, it's given the ix of-				--   the particular worker thread it's running on.+	-> (Int -> IO ())	-- ^ Action to run in all the workers, it's+                                --   given the ix of the particular worker+                                ---  thread it's running on. 	-> IO ()  parIO n mvs p @@ -190,7 +194,7 @@ 	reqs	<- sequence . replicate n $ newReq p  	traceGang "parIO: issuing requests"-	zipWithM putMVar mvs reqs+	zipWithM_ putMVar mvs reqs  	traceGang "parIO: waiting for requests to complete" 	mapM_ waitReq reqs@@ -214,6 +218,7 @@ diffTime :: Integer -> Integer -> String diffTime x y = show (y-x) +-- | Emit a GHC event for debugging. traceGang :: String -> IO () traceGang s  = do	t <- getGangTime@@ -226,11 +231,13 @@ diffTime :: () -> () -> String diffTime _ _ = "" +-- | Emit a GHC event for debugging. traceGang :: String -> IO () traceGang _ = return ()- #endif ++-- | Emit a GHC event for debugging, in the `ST` monad. traceGangST :: String -> ST s () traceGangST s = unsafeIOToST (traceGang s) 

@@ -1,21 +1,21 @@--- | Distributed scalars.+{-# OPTIONS -Wall -fno-warn-orphans -fno-warn-missing-signatures #-}++-- | Operations on distributed scalars. --   With a distributed value like (Dist Int), each thread has its own integer,  --   which may or may not have the same values as the ones on other threads.-module Data.Array.Parallel.Unlifted.Distributed.Scalars (-  unitD, scalarD,-  orD, andD, sumD-) where--import Data.Array.Parallel.Unlifted.Distributed.Gang (-  Gang)-import Data.Array.Parallel.Unlifted.Distributed.Types (-  DT, Dist, unitD)-import Data.Array.Parallel.Unlifted.Distributed.Combinators (-  mapD, foldD)+module Data.Array.Parallel.Unlifted.Distributed.Scalars +        ( scalarD+        , orD, andD+        , sumD)+where+import Data.Array.Parallel.Unlifted.Distributed.Gang+import Data.Array.Parallel.Unlifted.Distributed.Types+import Data.Array.Parallel.Unlifted.Distributed.Combinators   -- | Distribute a scalar. --   Each thread gets its own copy of the same value.+--   Example:  scalarD theGangN4 10 = [10, 10, 10, 10]  scalarD :: DT a => Gang -> a -> Dist a scalarD g x = mapD g (const x) (unitD g) 

@@ -1,20 +1,22 @@---- DPH programs always used a single, shared gang of threads.--- The gang exists at top level, and is initialised unsafely.--- --- The Vectoriser guarantees that the gang is only used by a single--- computation at a time.----module Data.Array.Parallel.Unlifted.Distributed.TheGang (-  theGang-) where-+{-# OPTIONS -Wall -fno-warn-orphans -fno-warn-missing-signatures #-}+module Data.Array.Parallel.Unlifted.Distributed.TheGang +        (theGang)+where import Data.Array.Parallel.Unlifted.Distributed.Gang -+import Control.Concurrent (getNumCapabilities) import System.IO.Unsafe (unsafePerformIO)-import GHC.Conc (numCapabilities) +-- | DPH programs use this single, shared gang of threads.+--   The gang exists at top level, and is initialised at program start.+-- +--   The vectoriser guarantees that the gang is only used by a single+--   computation at a time. This is true because the program produced+--   by the vector only uses flat parallelism, so parallel computations+--   don't invoke further parallel computations. If the vectorised program+--   tries to use nested parallelism then there is a bug in the vectoriser,+--   and the code will run sequentially.+-- theGang :: Gang+theGang = unsafePerformIO (getNumCapabilities >>= forkGang) {-# NOINLINE theGang #-}-theGang = unsafePerformIO (forkGang numCapabilities) 

@@ -1,495 +1,21 @@-{-# OPTIONS -fno-warn-incomplete-patterns #-}+{-# OPTIONS -Wall -fno-warn-orphans -fno-warn-missing-signatures #-} {-# LANGUAGE CPP #-}- #include "fusion-phases.h"  -- | Distributed types.-module Data.Array.Parallel.Unlifted.Distributed.Types (-  -- * Distributed types-  DT, Dist, MDist, DPrim(..),--  -- * Operations on immutable distributed types-  indexD, unitD, zipD, unzipD, fstD, sndD, lengthD,-  newD,-  -- zipSD, unzipSD, fstSD, sndSD,-  deepSeqD,--  lengthUSegdD, lengthsUSegdD, indicesUSegdD, elementsUSegdD,--  -- * Operations on mutable distributed types-  newMD, readMD, writeMD, unsafeFreezeMD,--  -- * Assertions-  checkGangD, checkGangMD,--  -- * Debugging functions-  sizeD, sizeMD, measureD, debugD-) where--import Data.Array.Parallel.Unlifted.Distributed.Gang (-  Gang, gangSize )-import Data.Array.Parallel.Unlifted.Sequential.Vector ( Unbox, Vector )-import qualified Data.Array.Parallel.Unlifted.Sequential.Vector as V-import Data.Array.Parallel.Unlifted.Sequential.Segmented-import Data.Array.Parallel.Base--import qualified Data.Vector.Unboxed as V-import qualified Data.Vector.Unboxed.Mutable as MV-import qualified Data.Vector as BV-import qualified Data.Vector.Mutable as MBV--import Data.Word     (Word8)-import Control.Monad (liftM, liftM2, liftM3)--import Data.List ( intercalate )--infixl 9 `indexD`--here s = "Data.Array.Parallel.Unlifted.Distributed.Types." ++ s----- Distributed Types ------------------------------------------------------------- | Class of distributable types. Instances of 'DT' can be---   distributed across all workers of a 'Gang'. ---   All such types must be hyperstrict as we do not want to pass thunks---   into distributed computations.-class DT a where-  data Dist  a-  data MDist a :: * -> *--  -- | Extract a single element of an immutable distributed value.-  indexD         :: Dist a -> Int -> a--  -- | Create an unitialised distributed value for the given 'Gang'.-  --   The gang is used (only) to know how many elements are needed-  --   in the distributed value.-  newMD          :: Gang                  -> ST s (MDist a s)--  -- | Extract an element from a mutable distributed value.-  readMD         :: MDist a s -> Int      -> ST s a--  -- | Write an element of a mutable distributed value.-  writeMD        :: MDist a s -> Int -> a -> ST s ()--  -- | Unsafely freeze a mutable distributed value.-  unsafeFreezeMD :: MDist a s             -> ST s (Dist a)--  deepSeqD       :: a -> b -> b-  deepSeqD = seq---  -- Debugging -------------------------  -- | Number of elements in the distributed value.-  --   For debugging only, as we shouldn't depend on the size of the gang.-  sizeD :: Dist a -> Int--  -- | Number of elements in the mutable distributed value.-  --   For debugging only, as we shouldn't care about the actual number.-  sizeMD :: MDist a s -> Int--  -- | Show a distributed value.-  --   For debugging only.-  measureD :: a -> String-  measureD _ = "None"---- Show instance (for debugging only)-instance (Show a, DT a) => Show (Dist a) where-  show d = show (Prelude.map (indexD d) [0 .. sizeD d - 1])------ Checking ---------------------------------------------------------------------- | Check that the sizes of the 'Gang' and of the distributed value match.-checkGangD :: DT a => String -> Gang -> Dist a -> b -> b-checkGangD loc g d v = checkEq loc "Wrong gang" (gangSize g) (sizeD d) v----- | Check that the sizes of the 'Gang' and of the mutable distributed value match.-checkGangMD :: DT a => String -> Gang -> MDist a s -> b -> b-checkGangMD loc g d v = checkEq loc "Wrong gang" (gangSize g) (sizeMD d) v----- Operations -------------------------------------------------------------------- | Given a computation that can write its result to a mutable distributed value, ---   run the computation to generate an immutable distributed value.-newD :: DT a => Gang -> (forall s . MDist a s -> ST s ()) -> Dist a-newD g init =-  runST (do-           mdt <- newMD g-           init mdt-           unsafeFreezeMD mdt)---- | Show all members of a distributed value.-debugD :: DT a => Dist a -> String-debugD d = "["-         ++ intercalate "," [measureD (indexD d i) | i <- [0 .. sizeD d-1]]-         ++ "]"----- DPrim ------------------------------------------------------------------------- | For distributed primitive values, we can just store all the members in---   a vector. The vector has the same length as the number of threads in the gang.----class Unbox e => DPrim e where--  -- | Make an immutable distributed value.-  mkDPrim :: V.Vector e -> Dist  e--  -- | Unpack an immutable distributed value back into a vector.-  unDPrim :: Dist  e -> V.Vector e--  -- | Make a mutable distributed value.-  mkMDPrim :: MV.STVector s e -> MDist  e s--  -- | Unpack a mutable distributed value back into a vector.-  unMDPrim :: MDist  e s -> MV.STVector s e----- | Get the member corresponding to a thread index.-primIndexD :: DPrim a => Dist a -> Int -> a-{-# INLINE primIndexD #-}-primIndexD = (V.!) . unDPrim----- | Create a new distributed value, having as many members as threads---   in the given 'Gang'.-primNewMD :: DPrim a => Gang -> ST s (MDist a s)-{-# INLINE primNewMD #-}-primNewMD = liftM mkMDPrim . MV.new . gangSize----- | Read the member of a distributed value corresponding to the given thread index.-primReadMD :: DPrim a => MDist a s -> Int -> ST s a-{-# INLINE primReadMD #-}-primReadMD = MV.read . unMDPrim----- | Write the member of a distributed value corresponding to the given thread index.-primWriteMD :: DPrim a => MDist a s -> Int -> a -> ST s ()-{-# INLINE primWriteMD #-}-primWriteMD = MV.write . unMDPrim----- | Freeze a mutable distributed value to an immutable one.---   You promise not to update the mutable one any further.-primUnsafeFreezeMD :: DPrim a => MDist a s -> ST s (Dist a)-{-# INLINE primUnsafeFreezeMD #-}-primUnsafeFreezeMD = liftM mkDPrim . V.unsafeFreeze . unMDPrim----- | Get the size of a distributed value, that is, the number of threads---   in the gang that it was created for.-primSizeD :: DPrim a => Dist a -> Int-{-# INLINE primSizeD #-}-primSizeD = V.length . unDPrim----- | Get the size of a distributed mutable value, that is, the number of threads---   in the gang it was created for.-primSizeMD :: DPrim a => MDist a s -> Int-{-# INLINE primSizeMD #-}-primSizeMD = MV.length . unMDPrim----- Unit ------------------------------------------------------------------------instance DT () where-  data Dist ()    = DUnit  !Int-  data MDist () s = MDUnit !Int--  indexD  (DUnit n) i       = check (here "indexD[()]") n i $ ()-  newMD                     = return . MDUnit . gangSize-  readMD   (MDUnit n) i     = check (here "readMD[()]")  n i $-                               return ()-  writeMD  (MDUnit n) i ()  = check (here "writeMD[()]") n i $-                               return ()-  unsafeFreezeMD (MDUnit n) = return $ DUnit n--  sizeD  = error "dph-prim-par:sizeD[()] undefined"-  sizeMD = error "dph-prim-par:sizeMD[()] undefined"---- | Yield a distributed unit.-unitD :: Gang -> Dist ()-{-# INLINE_DIST unitD #-}-unitD = DUnit . gangSize----- Bool ------------------------------------------------------------------------instance DPrim Bool where-  mkDPrim           = DBool-  unDPrim (DBool a) = a--  mkMDPrim            = MDBool-  unMDPrim (MDBool a) = a--instance DT Bool where-  data Dist  Bool   = DBool  !(V.Vector    Bool)-  data MDist Bool s = MDBool !(MV.STVector s Bool)--  indexD         = primIndexD-  newMD          = primNewMD-  readMD         = primReadMD-  writeMD        = primWriteMD-  unsafeFreezeMD = primUnsafeFreezeMD-  sizeD          = primSizeD-  sizeMD         = primSizeMD----- Char ------------------------------------------------------------------------instance DPrim Char where-  mkDPrim           = DChar-  unDPrim (DChar a) = a--  mkMDPrim            = MDChar-  unMDPrim (MDChar a) = a--instance DT Char where-  data Dist  Char   = DChar  !(V.Vector    Char)-  data MDist Char s = MDChar !(MV.STVector s Char)--  indexD         = primIndexD-  newMD          = primNewMD-  readMD         = primReadMD-  writeMD        = primWriteMD-  unsafeFreezeMD = primUnsafeFreezeMD-  sizeD          = primSizeD-  sizeMD         = primSizeMD----- Int -------------------------------------------------------------------------instance DPrim Int where-  mkDPrim          = DInt-  unDPrim (DInt a) = a--  mkMDPrim            = MDInt-  unMDPrim (MDInt a) = a--instance DT Int where-  data Dist  Int   = DInt  !(V.Vector    Int)-  data MDist Int s = MDInt !(MV.STVector s Int)--  indexD         = primIndexD-  newMD          = primNewMD-  readMD         = primReadMD-  writeMD        = primWriteMD-  unsafeFreezeMD = primUnsafeFreezeMD-  sizeD          = primSizeD-  sizeMD         = primSizeMD--  measureD n = "Int " ++ show n----- Word8 -----------------------------------------------------------------------instance DPrim Word8 where-  mkDPrim            = DWord8-  unDPrim (DWord8 a) = a--  mkMDPrim             = MDWord8-  unMDPrim (MDWord8 a) = a--instance DT Word8 where-  data Dist  Word8   = DWord8  !(V.Vector    Word8)-  data MDist Word8 s = MDWord8 !(MV.STVector s Word8)--  indexD         = primIndexD-  newMD          = primNewMD-  readMD         = primReadMD-  writeMD        = primWriteMD-  unsafeFreezeMD = primUnsafeFreezeMD-  sizeD          = primSizeD-  sizeMD         = primSizeMD----- Float -----------------------------------------------------------------------instance DPrim Float where-  mkDPrim            = DFloat-  unDPrim (DFloat a) = a--  mkMDPrim             = MDFloat-  unMDPrim (MDFloat a) = a--instance DT Float where-  data Dist  Float   = DFloat  !(V.Vector    Float)-  data MDist Float s = MDFloat !(MV.STVector s Float)--  indexD         = primIndexD-  newMD          = primNewMD-  readMD         = primReadMD-  writeMD        = primWriteMD-  unsafeFreezeMD = primUnsafeFreezeMD-  sizeD          = primSizeD-  sizeMD         = primSizeMD----- Double ----------------------------------------------------------------------instance DPrim Double where-  mkDPrim             = DDouble-  unDPrim (DDouble a) = a--  mkMDPrim              = MDDouble-  unMDPrim (MDDouble a) = a--instance DT Double where-  data Dist  Double   = DDouble  !(V.Vector    Double)-  data MDist Double s = MDDouble !(MV.STVector s Double)--  indexD         = primIndexD-  newMD          = primNewMD-  readMD         = primReadMD-  writeMD        = primWriteMD-  unsafeFreezeMD = primUnsafeFreezeMD-  sizeD          = primSizeD-  sizeMD         = primSizeMD----- Pairs -----------------------------------------------------------------------instance (DT a, DT b) => DT (a,b) where-  data Dist  (a,b)   = DProd  !(Dist a)    !(Dist b)-  data MDist (a,b) s = MDProd !(MDist a s) !(MDist b s)--  indexD d i               = (fstD d `indexD` i,sndD d `indexD` i)-  newMD g                  = liftM2 MDProd (newMD g) (newMD g)-  readMD  (MDProd xs ys) i = liftM2 (,) (readMD xs i) (readMD ys i)-  writeMD (MDProd xs ys) i (x,y)-                            = writeMD xs i x >> writeMD ys i y-  unsafeFreezeMD (MDProd xs ys)-                            = liftM2 DProd (unsafeFreezeMD xs)-                                           (unsafeFreezeMD ys)--  {-# INLINE deepSeqD #-}-  deepSeqD (x,y) z = deepSeqD x (deepSeqD y z)--  sizeD  (DProd  x _) = sizeD  x-  sizeMD (MDProd x _) = sizeMD x--  measureD (x,y) = "Pair " ++ "(" ++ measureD x ++ ") (" ++  measureD y ++ ")"----- | Pairing of distributed values.--- /The two values must belong to the same/ 'Gang'.-zipD :: (DT a, DT b) => Dist a -> Dist b -> Dist (a,b)-{-# INLINE [0] zipD #-}-zipD !x !y = checkEq (here "zipDT") "Size mismatch" (sizeD x) (sizeD y) $-             DProd x y---- | Unpairing of distributed values.-unzipD :: (DT a, DT b) => Dist (a,b) -> (Dist a, Dist b)-{-# INLINE_DIST unzipD #-}-unzipD (DProd dx dy) = (dx,dy)---- | Extract the first elements of a distributed pair.-fstD :: (DT a, DT b) => Dist (a,b) -> Dist a-{-# INLINE_DIST fstD #-}-fstD = fst . unzipD---- | Extract the second elements of a distributed pair.-sndD :: (DT a, DT b) => Dist (a,b) -> Dist b-{-# INLINE_DIST sndD #-}-sndD = snd . unzipD----- Maybe -----------------------------------------------------------------------instance DT a => DT (Maybe a) where-  data Dist  (Maybe a)   = DMaybe  !(Dist  Bool)   !(Dist  a)-  data MDist (Maybe a) s = MDMaybe !(MDist Bool s) !(MDist a s)--  indexD (DMaybe bs as) i-    | bs `indexD` i       = Just $ as `indexD` i-    | otherwise           = Nothing-  newMD g = liftM2 MDMaybe (newMD g) (newMD g)-  readMD (MDMaybe bs as) i =-    do-      b <- readMD bs i-      if b then liftM Just $ readMD as i-           else return Nothing-  writeMD (MDMaybe bs as) i Nothing  = writeMD bs i False-  writeMD (MDMaybe bs as) i (Just x) = writeMD bs i True-                                     >> writeMD as i x-  unsafeFreezeMD (MDMaybe bs as) = liftM2 DMaybe (unsafeFreezeMD bs)-                                                 (unsafeFreezeMD as)--  {-# INLINE deepSeqD #-}-  deepSeqD Nothing  z = z-  deepSeqD (Just x) z = deepSeqD x z--  sizeD  (DMaybe  b _) = sizeD  b-  sizeMD (MDMaybe b _) = sizeMD b--  measureD Nothing = "Nothing"-  measureD (Just x) = "Just (" ++ measureD x ++ ")"----- Vector ----------------------------------------------------------------------instance Unbox a => DT (V.Vector a) where-  data Dist  (Vector a)   = DVector  !(Dist  Int)   !(BV.Vector      (Vector a))-  data MDist (Vector a) s = MDVector !(MDist Int s) !(MBV.STVector s (Vector a))--  indexD (DVector _ a) i = a BV.! i-  newMD g = liftM2 MDVector (newMD g) (MBV.replicate (gangSize g)-                                         (error "MDist (Vector a) - uninitalised"))-  readMD (MDVector _ marr) = MBV.read marr-  writeMD (MDVector mlen marr) i a =-    do-      writeMD mlen i (V.length a)-      MBV.write marr i $! a-  unsafeFreezeMD (MDVector len a) = liftM2 DVector (unsafeFreezeMD len)-                                               (BV.unsafeFreeze a)-  sizeD  (DVector  _ a) = BV.length  a-  sizeMD (MDVector _ a) = MBV.length a--  measureD xs = "Vector " ++ show (V.length xs)----- | Yield the distributed length of a distributed array.-lengthD :: Unbox a => Dist (Vector a) -> Dist Int-lengthD (DVector l _) = l----- USegd -----------------------------------------------------------------------instance DT USegd where-  data Dist  USegd   = DUSegd  !(Dist (Vector Int))-                               !(Dist (Vector Int))-                               !(Dist Int)-  data MDist USegd s = MDUSegd !(MDist (Vector Int) s)-                               !(MDist (Vector Int) s)-                               !(MDist Int        s)--  indexD (DUSegd lens idxs eles) i-          = mkUSegd (indexD lens i) (indexD idxs i) (indexD eles i)-  newMD g = liftM3 MDUSegd (newMD g) (newMD g) (newMD g)-  readMD (MDUSegd lens idxs eles) i-          = liftM3 mkUSegd (readMD lens i) (readMD idxs i) (readMD eles i)-  writeMD (MDUSegd lens idxs eles) i segd-          = do-              writeMD lens i (lengthsUSegd  segd)-              writeMD idxs i (indicesUSegd  segd)-              writeMD eles i (elementsUSegd segd)-  unsafeFreezeMD (MDUSegd lens idxs eles)-          = liftM3 DUSegd (unsafeFreezeMD lens)-                          (unsafeFreezeMD idxs)-                          (unsafeFreezeMD eles)--  deepSeqD segd z = deepSeqD (lengthsUSegd  segd)-                  $ deepSeqD (indicesUSegd  segd)-                  $ deepSeqD (elementsUSegd segd) z--  sizeD  (DUSegd  _ _ eles) = sizeD eles-  sizeMD (MDUSegd _ _ eles) = sizeMD eles--  measureD segd = "Segd " ++ show (lengthUSegd segd) ++ " " ++ show (elementsUSegd segd)--lengthUSegdD :: Dist USegd -> Dist Int-{-# INLINE_DIST lengthUSegdD #-}-lengthUSegdD (DUSegd lens _ _) = lengthD lens--lengthsUSegdD :: Dist USegd -> Dist (Vector Int)-{-# INLINE_DIST lengthsUSegdD #-}-lengthsUSegdD (DUSegd lens _ _ ) = lens--indicesUSegdD :: Dist USegd -> Dist (Vector Int)-{-# INLINE_DIST indicesUSegdD #-}-indicesUSegdD (DUSegd _ idxs _) = idxs+module Data.Array.Parallel.Unlifted.Distributed.Types +        ( module Data.Array.Parallel.Unlifted.Distributed.Types.Vector+        , module Data.Array.Parallel.Unlifted.Distributed.Types.Maybe+        , module Data.Array.Parallel.Unlifted.Distributed.Types.Tuple+        , module Data.Array.Parallel.Unlifted.Distributed.Types.Prim+        , module Data.Array.Parallel.Unlifted.Distributed.Types.Unit+        , module Data.Array.Parallel.Unlifted.Distributed.Types.Base)+where+import Data.Array.Parallel.Unlifted.Distributed.Types.Vector+import Data.Array.Parallel.Unlifted.Distributed.Types.Maybe+import Data.Array.Parallel.Unlifted.Distributed.Types.Tuple+import Data.Array.Parallel.Unlifted.Distributed.Types.Prim+import Data.Array.Parallel.Unlifted.Distributed.Types.Unit+import Data.Array.Parallel.Unlifted.Distributed.Types.Base -elementsUSegdD :: Dist USegd -> Dist Int-{-# INLINE_DIST elementsUSegdD #-}-elementsUSegdD (DUSegd _ _ dns) = dns 

@@ -0,0 +1,108 @@+{-# OPTIONS -Wall -fno-warn-orphans -fno-warn-missing-signatures #-}+module Data.Array.Parallel.Unlifted.Distributed.Types.Base +        ( -- * Distributable Types+          DT(..)+        +          -- * Checking+        , checkGangD+        , checkGangMD++          -- * General Operations+        , newD+        , debugD)+where+import Data.Array.Parallel.Unlifted.Distributed.Gang    (Gang, gangSize)+import Data.Array.Parallel.Base+import Data.List                                        (intercalate)++here :: String -> String+here s = "Data.Array.Parallel.Unlifted.Distributed.Types.Base." ++ s+++-- Distributed Types ----------------------------------------------------------+infixl 9 `indexD`++-- | Class of distributable types. Instances of 'DT' can be+--   distributed across all workers of a 'Gang'. +--   All such types must be hyperstrict as we do not want to pass thunks+--   into distributed computations.+class DT a where+  data Dist  a+  data MDist a :: * -> *++  -- | Extract a single element of an immutable distributed value.+  indexD         :: String -> Dist a -> Int -> a++  -- | Create an unitialised distributed value for the given 'Gang'.+  --   The gang is used (only) to know how many elements are needed+  --   in the distributed value.+  newMD          :: Gang                  -> ST s (MDist a s)++  -- | Extract an element from a mutable distributed value.+  readMD         :: MDist a s -> Int      -> ST s a++  -- | Write an element of a mutable distributed value.+  writeMD        :: MDist a s -> Int -> a -> ST s ()++  -- | Unsafely freeze a mutable distributed value.+  unsafeFreezeMD :: MDist a s             -> ST s (Dist a)++  -- | Ensure a distributed value is fully evaluated.+  deepSeqD       :: a -> b -> b+  deepSeqD = seq+++  -- Debugging ------------------------+  -- | Number of elements in the distributed value.+  -- +  --   * For debugging only, as code shouldn't be sensitive to the return value.+  sizeD :: Dist a -> Int++  -- | Number of elements in the mutable distributed value.+  --  +  --   * For debugging only, as code shouldn't be sensitive to the return value.+  sizeMD :: MDist a s -> Int++  -- | Show a distributed value.+  --+  --   * For debugging only.+  measureD :: a -> String+  measureD _ = "None"+++-- Show -----------------------------------------------------------------------+-- Show instance (for debugging only) --+instance (Show a, DT a) => Show (Dist a) where+  show d = show (Prelude.map (indexD (here "show") d) [0 .. sizeD d - 1])+++-- Checking -------------------------------------------------------------------+-- | Check that the sizes of the 'Gang' and of the distributed value match.+checkGangD :: DT a => String -> Gang -> Dist a -> b -> b+checkGangD loc g d v+        = checkEq loc "Wrong gang" (gangSize g) (sizeD d) v+++-- | Check that the sizes of the 'Gang' and of the mutable distributed value match.+checkGangMD :: DT a => String -> Gang -> MDist a s -> b -> b+checkGangMD loc g d v+        = checkEq loc "Wrong gang" (gangSize g) (sizeMD d) v+++-- Operations -----------------------------------------------------------------+-- | Given a computation that can write its result to a mutable distributed value, +--   run the computation to generate an immutable distributed value.+newD :: DT a => Gang -> (forall s . MDist a s -> ST s ()) -> Dist a+newD g mkInit =+  runST (do+           mdt <- newMD g+           mkInit mdt+           unsafeFreezeMD mdt)++-- | Show all members of a distributed value.+debugD :: DT a => Dist a -> String+debugD d = "["+         ++ intercalate "," [measureD (indexD (here "debugD") d i) +                            | i <- [0 .. sizeD d-1]]+         ++ "]"+

@@ -0,0 +1,45 @@+{-# OPTIONS -Wall -fno-warn-orphans -fno-warn-missing-signatures #-}++-- | Distribution of Maybes.+module Data.Array.Parallel.Unlifted.Distributed.Types.Maybe where+import Data.Array.Parallel.Unlifted.Distributed.Types.Prim      ()+import Data.Array.Parallel.Unlifted.Distributed.Types.Base+import Control.Monad++instance DT a => DT (Maybe a) where+  data Dist  (Maybe a)   = DMaybe  !(Dist  Bool)   !(Dist  a)+  data MDist (Maybe a) s = MDMaybe !(MDist Bool s) !(MDist a s)++  indexD str (DMaybe bs as) i+    |        indexD (str ++ "/indexD[Maybe]") bs i+    = Just $ indexD (str ++ "/indexD[Maybe]" ++ str) as i+    | otherwise           = Nothing++  newMD g+   = liftM2 MDMaybe (newMD g) (newMD g)++  readMD (MDMaybe bs as) i +   = do b <- readMD bs i+        if b then liftM Just $ readMD as i+             else return Nothing++  writeMD (MDMaybe bs _) i Nothing +   = writeMD bs i False++  writeMD (MDMaybe bs as) i (Just x)+   = do writeMD bs i True+        writeMD as i x++  unsafeFreezeMD (MDMaybe bs as)+   = liftM2 DMaybe (unsafeFreezeMD bs)+                   (unsafeFreezeMD as)++  {-# INLINE deepSeqD #-}+  deepSeqD Nothing  z   = z+  deepSeqD (Just x) z   = deepSeqD x z++  sizeD  (DMaybe  b _)  = sizeD  b+  sizeMD (MDMaybe b _)  = sizeMD b++  measureD Nothing      = "Nothing"+  measureD (Just x)     = "Just (" ++ measureD x ++ ")"

@@ -0,0 +1,250 @@+{-# OPTIONS -Wall -fno-warn-orphans -fno-warn-missing-signatures #-}++-- | Distribution of values of primitive types.+module Data.Array.Parallel.Unlifted.Distributed.Types.Prim +        ( DPrim (..)+        , DT    (..)+        , Dist  (..))+where+import Data.Array.Parallel.Unlifted.Distributed.Types.Base+import Data.Array.Parallel.Unlifted.Distributed.Gang+import Data.Array.Parallel.Unlifted.Sequential.Vector+import Data.Array.Parallel.Base+import Data.Array.Parallel.Pretty+import Data.Word+import Control.Monad+import qualified Data.Array.Parallel.Unlifted.Sequential.Vector as V+import qualified Data.Vector.Unboxed.Mutable                    as MV+import Prelude as P++-- DPrim ----------------------------------------------------------------------+-- | For distributed primitive values, we can just store all the members in+--   a vector. The vector has the same length as the number of threads in the gang.+--+class Unbox e => DPrim e where++  -- | Make an immutable distributed value.+  mkDPrim :: V.Vector e -> Dist  e++  -- | Unpack an immutable distributed value back into a vector.+  unDPrim :: Dist  e -> V.Vector e++  -- | Make a mutable distributed value.+  mkMDPrim :: MV.STVector s e -> MDist  e s++  -- | Unpack a mutable distributed value back into a vector.+  unMDPrim :: MDist  e s -> MV.STVector s e+++-- | Get the member corresponding to a thread index.+primIndexD :: DPrim a => String -> Dist a -> Int -> a+primIndexD str = (V.index (str P.++ "/primIndexD")) . unDPrim+{-# INLINE primIndexD #-}+++-- | Create a new distributed value, having as many members as threads+--   in the given 'Gang'.+primNewMD :: DPrim a => Gang -> ST s (MDist a s)+primNewMD = liftM mkMDPrim . MV.new . gangSize+{-# INLINE primNewMD #-}+++-- | Read the member of a distributed value corresponding to the given thread index.+primReadMD :: DPrim a => MDist a s -> Int -> ST s a+primReadMD = MV.read . unMDPrim+{-# INLINE primReadMD #-}+++-- | Write the member of a distributed value corresponding to the given thread index.+primWriteMD :: DPrim a => MDist a s -> Int -> a -> ST s ()+primWriteMD = MV.write . unMDPrim+{-# INLINE primWriteMD #-}+++-- | Freeze a mutable distributed value to an immutable one.+--   You promise not to update the mutable one any further.+primUnsafeFreezeMD :: DPrim a => MDist a s -> ST s (Dist a)+primUnsafeFreezeMD = liftM mkDPrim . V.unsafeFreeze . unMDPrim+{-# INLINE primUnsafeFreezeMD #-}+++-- | Get the size of a distributed value, that is, the number of threads+--   in the gang that it was created for.+primSizeD :: DPrim a => Dist a -> Int+primSizeD = V.length . unDPrim+{-# INLINE primSizeD #-}+++-- | Get the size of a distributed mutable value, that is, the number of threads+--   in the gang it was created for.+primSizeMD :: DPrim a => MDist a s -> Int+primSizeMD = MV.length . unMDPrim+{-# INLINE primSizeMD #-}+++-- Bool -----------------------------------------------------------------------+instance DPrim Bool where+  mkDPrim           = DBool+  unDPrim (DBool a) = a++  mkMDPrim            = MDBool+  unMDPrim (MDBool a) = a+++instance DT Bool where+  data Dist  Bool   = DBool  !(V.Vector    Bool)+  data MDist Bool s = MDBool !(MV.STVector s Bool)++  indexD         = primIndexD+  newMD          = primNewMD+  readMD         = primReadMD+  writeMD        = primWriteMD+  unsafeFreezeMD = primUnsafeFreezeMD+  sizeD          = primSizeD+  sizeMD         = primSizeMD+++-- Ordering -----------------------------------------------------------------------+instance DPrim Ordering where+  mkDPrim               = DOrdering+  unDPrim (DOrdering a) = a++  mkMDPrim                = MDOrdering+  unMDPrim (MDOrdering a) = a+++instance DT Ordering where+  data Dist  Ordering   = DOrdering  !(V.Vector    Ordering)+  data MDist Ordering s = MDOrdering !(MV.STVector s Ordering)++  indexD         = primIndexD+  newMD          = primNewMD+  readMD         = primReadMD+  writeMD        = primWriteMD+  unsafeFreezeMD = primUnsafeFreezeMD+  sizeD          = primSizeD+  sizeMD         = primSizeMD+++-- Integer -----------------------------------------------------------------------+-- FIXME: fake instances+instance DPrim Integer+instance DT Integer+++-- Char -----------------------------------------------------------------------+instance DPrim Char where+  mkDPrim           = DChar+  unDPrim (DChar a) = a++  mkMDPrim            = MDChar+  unMDPrim (MDChar a) = a+++instance DT Char where+  data Dist  Char   = DChar  !(V.Vector    Char)+  data MDist Char s = MDChar !(MV.STVector s Char)++  indexD         = primIndexD+  newMD          = primNewMD+  readMD         = primReadMD+  writeMD        = primWriteMD+  unsafeFreezeMD = primUnsafeFreezeMD+  sizeD          = primSizeD+  sizeMD         = primSizeMD+++-- Int ------------------------------------------------------------------------+instance DPrim Int where+  mkDPrim          = DInt+  unDPrim (DInt a) = a++  mkMDPrim            = MDInt+  unMDPrim (MDInt a) = a+++instance DT Int where+  data Dist  Int   = DInt  !(V.Vector    Int)+  data MDist Int s = MDInt !(MV.STVector s Int)++  indexD         = primIndexD+  newMD          = primNewMD+  readMD         = primReadMD+  writeMD        = primWriteMD+  unsafeFreezeMD = primUnsafeFreezeMD+  sizeD          = primSizeD+  sizeMD         = primSizeMD++  measureD n = "Int " P.++ show n++instance PprPhysical (Dist Int) where+ pprp (DInt xs)+  =  text "DInt" <+> text (show $ V.toList xs)+++-- Word8 ----------------------------------------------------------------------+instance DPrim Word8 where+  mkDPrim            = DWord8+  unDPrim (DWord8 a) = a++  mkMDPrim             = MDWord8+  unMDPrim (MDWord8 a) = a+++instance DT Word8 where+  data Dist  Word8   = DWord8  !(V.Vector    Word8)+  data MDist Word8 s = MDWord8 !(MV.STVector s Word8)++  indexD         = primIndexD+  newMD          = primNewMD+  readMD         = primReadMD+  writeMD        = primWriteMD+  unsafeFreezeMD = primUnsafeFreezeMD+  sizeD          = primSizeD+  sizeMD         = primSizeMD+++-- Float ----------------------------------------------------------------------+instance DPrim Float where+  mkDPrim            = DFloat+  unDPrim (DFloat a) = a++  mkMDPrim             = MDFloat+  unMDPrim (MDFloat a) = a+++instance DT Float where+  data Dist  Float   = DFloat  !(V.Vector    Float)+  data MDist Float s = MDFloat !(MV.STVector s Float)++  indexD         = primIndexD+  newMD          = primNewMD+  readMD         = primReadMD+  writeMD        = primWriteMD+  unsafeFreezeMD = primUnsafeFreezeMD+  sizeD          = primSizeD+  sizeMD         = primSizeMD+++-- Double ---------------------------------------------------------------------+instance DPrim Double where+  mkDPrim             = DDouble+  unDPrim (DDouble a) = a++  mkMDPrim              = MDDouble+  unMDPrim (MDDouble a) = a+++instance DT Double where+  data Dist  Double   = DDouble  !(V.Vector    Double)+  data MDist Double s = MDDouble !(MV.STVector s Double)++  indexD         = primIndexD+  newMD          = primNewMD+  readMD         = primReadMD+  writeMD        = primWriteMD+  unsafeFreezeMD = primUnsafeFreezeMD+  sizeD          = primSizeD+  sizeMD         = primSizeMD++

@@ -0,0 +1,145 @@+{-# OPTIONS -Wall -fno-warn-orphans -fno-warn-missing-signatures #-}+{-# LANGUAGE CPP #-}+#include "fusion-phases.h"++-- | Distribution of Tuples+module Data.Array.Parallel.Unlifted.Distributed.Types.Tuple +        ( -- * Pairs+          zipD, unzipD, fstD, sndD+        +           -- * Triples+        , zip3D, unzip3D)+where+import Data.Array.Parallel.Unlifted.Distributed.Types.Base+import Data.Array.Parallel.Base+import Data.Array.Parallel.Pretty+import Control.Monad++here :: String -> String+here s = "Data.Array.Parallel.Unlifted.Distributed.Types.Tuple." ++ s+++-- Pairs ----------------------------------------------------------------------+instance (DT a, DT b) => DT (a,b) where+  data Dist  (a,b)   = DProd  !(Dist a)    !(Dist b)+  data MDist (a,b) s = MDProd !(MDist a s) !(MDist b s)++  indexD str d i+   = ( indexD (str ++ "/indexD[Tuple2]") (fstD d) i+     , indexD (str ++ "/indexD[Tuple2]") (sndD d) i)++  newMD g+   = liftM2 MDProd (newMD g) (newMD g)++  readMD  (MDProd xs ys) i+   = liftM2 (,) (readMD xs i) (readMD ys i)++  writeMD (MDProd xs ys) i (x,y)+   = do writeMD xs i x+        writeMD ys i y++  unsafeFreezeMD (MDProd xs ys)+   = liftM2 DProd (unsafeFreezeMD xs)+                  (unsafeFreezeMD ys)++  {-# INLINE deepSeqD #-}+  deepSeqD (x, y) z +   = deepSeqD x (deepSeqD y z)++  sizeD  (DProd  x _) = sizeD  x+  sizeMD (MDProd x _) = sizeMD x++  measureD (x, y) +   = "Pair " ++ "(" ++ measureD x ++ ") (" ++  measureD y ++ ")"+++instance (PprPhysical (Dist a), PprPhysical (Dist b)) +        => PprPhysical (Dist (a, b)) where+ pprp (DProd xs ys)+  = text "DProd"+  $$ (nest 8 $ vcat+        [ pprp xs+        , pprp ys ])+++-- | Pairing of distributed values.+--   The two values must belong to the same 'Gang'.+zipD :: (DT a, DT b) => Dist a -> Dist b -> Dist (a,b)+zipD !x !y +        = checkEq (here "zipDT") "Size mismatch" (sizeD x) (sizeD y) +        $ DProd x y+{-# INLINE [0] zipD #-}+++-- | Unpairing of distributed values.+unzipD :: (DT a, DT b) => Dist (a,b) -> (Dist a, Dist b)+unzipD (DProd dx dy) = (dx,dy)+{-# INLINE_DIST unzipD #-}+++-- | Extract the first elements of a distributed pair.+fstD :: (DT a, DT b) => Dist (a,b) -> Dist a+fstD = fst . unzipD+{-# INLINE_DIST fstD #-}+++-- | Extract the second elements of a distributed pair.+sndD :: (DT a, DT b) => Dist (a,b) -> Dist b+sndD = snd . unzipD+{-# INLINE_DIST sndD #-}+++-- Triples --------------------------------------------------------------------+instance (DT a, DT b, DT c) => DT (a,b,c) where+  data Dist  (a,b,c)   = DProd3  !(Dist a)    !(Dist b)    !(Dist c)+  data MDist (a,b,c) s = MDProd3 !(MDist a s) !(MDist b s) !(MDist c s)++  indexD str (DProd3 xs ys zs) i+   = ( indexD (here $ "indexD[Tuple3]/" ++ str) xs i+     , indexD (here $ "indexD[Tuple3]/" ++ str) ys i+     , indexD (here $ "indexD[Tuple3]/" ++ str) zs i)++  newMD g+   = liftM3 MDProd3 (newMD g) (newMD g) (newMD g)++  readMD  (MDProd3 xs ys zs) i+   = liftM3 (,,) (readMD xs i) (readMD ys i) (readMD zs i)++  writeMD (MDProd3 xs ys zs) i (x,y,z)+   = do writeMD xs i x+        writeMD ys i y+        writeMD zs i z++  unsafeFreezeMD (MDProd3 xs ys zs)+   = liftM3 DProd3 (unsafeFreezeMD xs)+                   (unsafeFreezeMD ys)+                   (unsafeFreezeMD zs)++  {-# INLINE deepSeqD #-}+  deepSeqD (x,y,z) k +   = deepSeqD x (deepSeqD y (deepSeqD z k))++  sizeD  (DProd3  x _ _) = sizeD  x+  sizeMD (MDProd3 x _ _) = sizeMD x++  measureD (x,y,z)+   = "Triple " +        ++ "(" ++ measureD x ++ ") "+        ++ "(" ++ measureD y ++ ") "+        ++ "(" ++ measureD z ++ ")"+++-- | Pairing of distributed values.+-- /The two values must belong to the same/ 'Gang'.+zip3D   :: (DT a, DT b, DT c) => Dist a -> Dist b -> Dist c -> Dist (a,b,c)+zip3D !x !y !z+        = checkEq (here "zip3DT") "Size mismatch" (sizeD x) (sizeD y) +        $ checkEq (here "zip3DT") "Size mismatch" (sizeD x) (sizeD z) +        $ DProd3 x y z+{-# INLINE [0] zip3D #-}+++-- | Unpairing of distributed values.+unzip3D  :: (DT a, DT b, DT c) => Dist (a,b,c) -> (Dist a, Dist b, Dist c)+unzip3D (DProd3 dx dy dz) = (dx,dy,dz)+{-# INLINE_DIST unzip3D #-}

@@ -0,0 +1,130 @@+{-# OPTIONS -Wall -fno-warn-orphans -fno-warn-missing-signatures #-}+{-# LANGUAGE CPP #-}+#include "fusion-phases.h"++-- | Distribution of Segment Descriptors+module Data.Array.Parallel.Unlifted.Distributed.Types.USSegd +        ( lengthD+        , takeLengthsD+        , takeIndicesD+        , takeElementsD+        , takeStartsD+        , takeSourcesD+        , takeUSegdD)+where+import Data.Array.Parallel.Unlifted.Distributed.Types.Base+import Data.Array.Parallel.Unlifted.Sequential.USSegd                   (USSegd)+import Data.Array.Parallel.Unlifted.Sequential.USegd                    (USegd)+import Data.Array.Parallel.Unlifted.Sequential.Vector                   (Vector)+import Data.Array.Parallel.Pretty+import Control.Monad+import Prelude                                                          as P+import qualified Data.Array.Parallel.Unlifted.Distributed.Types.USegd   as DUSegd+import qualified Data.Array.Parallel.Unlifted.Distributed.Types.Vector  as DV+import qualified Data.Array.Parallel.Unlifted.Sequential.USSegd         as USSegd+++instance DT USSegd where+  data Dist USSegd   +        = DUSSegd  !(Dist (Vector Int))         -- segment starts+                   !(Dist (Vector Int))         -- segment sources+                   !(Dist USegd)                -- distributed usegd++  data MDist USSegd s +        = MDUSSegd !(MDist (Vector Int) s)      -- segment starts+                   !(MDist (Vector Int) s)      -- segment sources+                   !(MDist USegd        s)      -- distributed usegd++  indexD str (DUSSegd starts sources usegds) i+   = USSegd.mkUSSegd+        (indexD (str ++ "/indexD[USSegd]") starts i)+        (indexD (str ++ "/indexD[USSegd]") sources i)+        (indexD (str ++ "/indexD[USSegd]") usegds i)++  newMD g+   = liftM3 MDUSSegd (newMD g) (newMD g) (newMD g)++  readMD (MDUSSegd starts sources usegds) i+   = liftM3 USSegd.mkUSSegd (readMD starts i) (readMD sources i) (readMD usegds i)++  writeMD (MDUSSegd starts sources usegds) i ussegd+   = do writeMD starts  i (USSegd.takeStarts  ussegd)+        writeMD sources i (USSegd.takeSources ussegd)+        writeMD usegds  i (USSegd.takeUSegd   ussegd)++  unsafeFreezeMD (MDUSSegd starts sources usegds)+   = liftM3 DUSSegd (unsafeFreezeMD starts)+                    (unsafeFreezeMD sources)+                    (unsafeFreezeMD usegds)++  deepSeqD ussegd z+   = deepSeqD (USSegd.takeStarts  ussegd)+   $ deepSeqD (USSegd.takeSources ussegd)+   $ deepSeqD (USSegd.takeUSegd   ussegd) z++  sizeD  (DUSSegd  _ _ usegd) = sizeD usegd+  sizeMD (MDUSSegd _ _ usegd) = sizeMD usegd++  measureD ussegd +   = "USSegd "  P.++ show (USSegd.takeStarts    ussegd)+   P.++ " "     P.++ show (USSegd.takeSources   ussegd)+   P.++ " "     P.++ measureD (USSegd.takeUSegd ussegd)+++instance PprPhysical (Dist USSegd) where+ pprp (DUSSegd starts sources usegds)+  =  text "DUSSegd"+  $$ (nest 7 $ vcat+        [ text "starts:  " <+> pprp starts+        , text "sources: " <+> pprp sources+        , text "usegds:  " <+> pprp usegds])+++-- | O(1). Yield the overall number of segments.+lengthD :: Dist USSegd -> Dist Int+lengthD (DUSSegd starts _ _) +        = DV.lengthD starts+{-# INLINE_DIST lengthD #-}+++-- | O(1). Yield the lengths of the individual segments.+takeLengthsD :: Dist USSegd -> Dist (Vector Int)+takeLengthsD (DUSSegd _ _ usegds)+        = DUSegd.takeLengthsD usegds+{-# INLINE_DIST takeLengthsD #-}+++-- | O(1). Yield the segment indices.+takeIndicesD :: Dist USSegd -> Dist (Vector Int)+takeIndicesD (DUSSegd _ _ usegds)+        = DUSegd.takeIndicesD usegds+{-# INLINE_DIST takeIndicesD #-}+++-- | O(1). Yield the number of data elements.+takeElementsD :: Dist USSegd -> Dist Int+takeElementsD (DUSSegd _ _ usegds)+        = DUSegd.takeElementsD usegds+{-# INLINE_DIST takeElementsD #-}+++-- | O(1). Yield the starting indices.+takeStartsD :: Dist USSegd -> Dist (Vector Int)+takeStartsD (DUSSegd starts _ _)+        = starts+{-# INLINE_DIST takeStartsD #-}+        ++-- | O(1). Yield the source ids+takeSourcesD :: Dist USSegd -> Dist (Vector Int)+takeSourcesD (DUSSegd _ sources _)+        = sources+{-# INLINE_DIST takeSourcesD #-}+++-- | O(1). Yield the USegd+takeUSegdD :: Dist USSegd -> Dist USegd+takeUSegdD (DUSSegd _ _ usegd)+        = usegd+{-# INLINE_DIST takeUSegdD #-}+

@@ -0,0 +1,113 @@+{-# OPTIONS -Wall -fno-warn-orphans -fno-warn-missing-signatures #-}+{-# LANGUAGE CPP #-}+#include "fusion-phases.h"++-- | Distribution of Segment Descriptors+module Data.Array.Parallel.Unlifted.Distributed.Types.USegd +        ( mkDUSegd+        , lengthD+        , takeLengthsD+        , takeIndicesD+        , takeElementsD)+where+import Data.Array.Parallel.Unlifted.Distributed.Types.Base+import Data.Array.Parallel.Unlifted.Sequential.USegd                    (USegd)+import Data.Array.Parallel.Unlifted.Sequential.Vector                   (Vector)+import Data.Array.Parallel.Pretty+import Control.Monad+import qualified Data.Array.Parallel.Unlifted.Distributed.Types.Vector  as DV+import qualified Data.Array.Parallel.Unlifted.Sequential.USegd          as USegd+import Prelude                                                          as P+++instance DT USegd where+  data Dist  USegd   +        = DUSegd  !(Dist (Vector Int))          -- segment lengths+                  !(Dist (Vector Int))          -- segment indices+                  !(Dist Int)                   -- number of elements in this chunk++  data MDist USegd s +        = MDUSegd !(MDist (Vector Int) s)       -- segment lengths+                  !(MDist (Vector Int) s)       -- segment indices+                  !(MDist Int        s)         -- number of elements in this chunk++  indexD str (DUSegd lens idxs eles) i+   = USegd.mkUSegd+        (indexD (str ++ "/indexD[USegd]") lens i)+        (indexD (str ++ "/indexD[USegd]") idxs i)+        (indexD (str ++ "/indexD[USegd]") eles i)++  newMD g+   = liftM3 MDUSegd (newMD g) (newMD g) (newMD g)++  readMD (MDUSegd lens idxs eles) i+   = liftM3 USegd.mkUSegd (readMD lens i) (readMD idxs i) (readMD eles i)++  writeMD (MDUSegd lens idxs eles) i segd+   = do writeMD lens i (USegd.takeLengths  segd)+        writeMD idxs i (USegd.takeIndices  segd)+        writeMD eles i (USegd.takeElements segd)++  unsafeFreezeMD (MDUSegd lens idxs eles)+   = liftM3 DUSegd (unsafeFreezeMD lens)+                   (unsafeFreezeMD idxs)+                   (unsafeFreezeMD eles)++  deepSeqD segd z+   = deepSeqD (USegd.takeLengths  segd)+   $ deepSeqD (USegd.takeIndices  segd)+   $ deepSeqD (USegd.takeElements segd) z++  sizeD  (DUSegd  _ _ eles) = sizeD eles+  sizeMD (MDUSegd _ _ eles) = sizeMD eles++  measureD segd +   = "Segd " P.++ show (USegd.length segd)+   P.++ " "  P.++ show (USegd.takeElements segd)+++instance PprPhysical (Dist USegd) where+ pprp (DUSegd lens indices elements)+  =  text "DUSegd"+  $$ (nest 7 $ vcat+        [ text "lengths: " <+> pprp lens+        , text "indices: " <+> pprp indices+        , text "elements:" <+> pprp elements])+++-- | O(1). Construct a distributed segment descriptor+mkDUSegd +        :: Dist (Vector Int)    -- ^ segment lengths+        -> Dist (Vector Int)    -- ^ segment indices+        -> Dist Int             -- ^ number of elements in each chunk+        -> Dist USegd++mkDUSegd = DUSegd+++-- | O(1). Yield the overall number of segments.+lengthD :: Dist USegd -> Dist Int+lengthD (DUSegd lens _ _) +        = DV.lengthD lens+{-# INLINE_DIST lengthD #-}+++-- | O(1). Yield the lengths of the individual segments.+takeLengthsD :: Dist USegd -> Dist (Vector Int)+takeLengthsD (DUSegd lens _ _ )+        = lens+{-# INLINE_DIST takeLengthsD #-}+++-- | O(1). Yield the segment indices of a segment descriptor.+takeIndicesD :: Dist USegd -> Dist (Vector Int)+takeIndicesD (DUSegd _ idxs _)+        = idxs+{-# INLINE_DIST takeIndicesD #-}+++-- | O(1). Yield the number of data elements.+takeElementsD :: Dist USegd -> Dist Int+takeElementsD (DUSegd _ _ dns)+        = dns+{-# INLINE_DIST takeElementsD #-}

@@ -0,0 +1,128 @@+{-# OPTIONS -Wall -fno-warn-orphans -fno-warn-missing-signatures #-}+{-# LANGUAGE CPP #-}+#include "fusion-phases.h"++-- | Distribution of Virtual Segment Descriptors+module Data.Array.Parallel.Unlifted.Distributed.Types.UVSegd +        ( lengthD+        , takeLengthsD+        , takeIndicesD+        , takeElementsD+        , takeStartsD+        , takeSourcesD+        , takeVSegidsD+        , takeUSSegdD)+where+import Data.Array.Parallel.Unlifted.Distributed.Types.Base+import Data.Array.Parallel.Unlifted.Sequential.UVSegd                   (UVSegd)+import Data.Array.Parallel.Unlifted.Sequential.USSegd                   (USSegd)+import Data.Array.Parallel.Unlifted.Sequential.Vector+import Data.Array.Parallel.Pretty+import Control.Monad+import Prelude                                                          as P+import qualified Data.Array.Parallel.Unlifted.Sequential.UVSegd         as UVSegd+import qualified Data.Array.Parallel.Unlifted.Distributed.Types.USSegd  as DUSegd+++instance DT UVSegd where+  data Dist UVSegd   +        = DUVSegd  !(Dist (Vector Int))         -- vsegids+                   !(Dist USSegd)               -- distributed ussegd++  data MDist UVSegd s +        = MDUVSegd !(MDist (Vector Int) s)      -- vsegids+                   !(MDist USSegd       s)      -- distributed ussegd++  indexD str (DUVSegd vsegids ussegds) i+   = UVSegd.mkUVSegd+        (indexD (str P.++ "/indexD[UVSegd]") vsegids i)+        (indexD (str P.++ "/indexD[UVSegd]") ussegds i)++  newMD g+   = liftM2 MDUVSegd (newMD g) (newMD g)++  readMD (MDUVSegd vsegids ussegds) i+   = liftM2 UVSegd.mkUVSegd (readMD vsegids i) (readMD ussegds i)++  writeMD (MDUVSegd vsegids ussegds) i uvsegd+   = do writeMD vsegids  i (UVSegd.takeVSegids  uvsegd)+        writeMD ussegds  i (UVSegd.takeUSSegd   uvsegd)++  unsafeFreezeMD (MDUVSegd vsegids ussegds)+   = liftM2 DUVSegd (unsafeFreezeMD vsegids)+                    (unsafeFreezeMD ussegds)++  deepSeqD uvsegd z+   = deepSeqD (UVSegd.takeVSegids  uvsegd)+   $ deepSeqD (UVSegd.takeUSSegd   uvsegd) z++  sizeD  (DUVSegd  _ ussegd) = sizeD ussegd+  sizeMD (MDUVSegd _ ussegd) = sizeMD ussegd++  measureD uvsegd +   = "UVSegd " P.++ show (UVSegd.takeVSegids    uvsegd)+   P.++ " "    P.++ measureD (UVSegd.takeUSSegd uvsegd)+++instance PprPhysical (Dist UVSegd) where+ pprp (DUVSegd vsegids ussegds)+  =  text "DUVSegd"+  $$ (nest 7 $ vcat+        [ text "vsegids: " <+> pprp vsegids+        , text "ussegds: " <+> pprp ussegds])+++-- | O(1). Yield the overall number of segments.+lengthD :: Dist UVSegd -> Dist Int+lengthD (DUVSegd _ ussegd) +        = DUSegd.lengthD ussegd+{-# INLINE_DIST lengthD #-}+++-- | O(1). Yield the lengths of the individual segments.+takeLengthsD :: Dist UVSegd -> Dist (Vector Int)+takeLengthsD (DUVSegd _ ussegd)+        = DUSegd.takeLengthsD ussegd+{-# INLINE_DIST takeLengthsD #-}+++-- | O(1). Yield the segment indices.+takeIndicesD :: Dist UVSegd -> Dist (Vector Int)+takeIndicesD (DUVSegd _ ussegd)+        = DUSegd.takeIndicesD ussegd+{-# INLINE_DIST takeIndicesD #-}+++-- | O(1). Yield the number of data elements.+takeElementsD :: Dist UVSegd -> Dist Int+takeElementsD (DUVSegd _ ussegd)+        = DUSegd.takeElementsD ussegd+{-# INLINE_DIST takeElementsD #-}+++-- | O(1). Yield the starting indices.+takeStartsD :: Dist UVSegd -> Dist (Vector Int)+takeStartsD (DUVSegd _ ussegd)+        = DUSegd.takeStartsD ussegd+{-# INLINE_DIST takeStartsD #-}+        +        +-- | O(1). Yield the source ids+takeSourcesD :: Dist UVSegd -> Dist (Vector Int)+takeSourcesD (DUVSegd _ ussegd)+        = DUSegd.takeSourcesD ussegd+{-# INLINE_DIST takeSourcesD #-}+++-- | O(1). Yield the vsegids+takeVSegidsD :: Dist UVSegd -> Dist (Vector Int)+takeVSegidsD (DUVSegd vsegids _)+        = vsegids+{-# INLINE_DIST takeVSegidsD #-}+++-- | O(1). Yield the USSegd+takeUSSegdD :: Dist UVSegd -> Dist USSegd+takeUSSegdD (DUVSegd _ ussegd)+        = ussegd+{-# INLINE_DIST takeUSSegdD #-}

@@ -0,0 +1,45 @@+{-# OPTIONS -Wall -fno-warn-orphans -fno-warn-missing-signatures #-}+{-# LANGUAGE CPP #-}+#include "fusion-phases.h"++-- | Distribution of unit values.+module Data.Array.Parallel.Unlifted.Distributed.Types.Unit +        (unitD)+where+import Data.Array.Parallel.Unlifted.Distributed.Types.Base+import Data.Array.Parallel.Unlifted.Distributed.Gang+import Data.Array.Parallel.Base++here :: String -> String+here s = "Data.Array.Parallel.Unlifted.Distributed.Types.Unit." ++ s++instance DT () where+  data Dist ()    = DUnit  !Int+  data MDist () s = MDUnit !Int++  indexD str (DUnit n) i+   = check (str ++ "/indexD[Unit]") n i+   $  ()++  newMD+   = return . MDUnit . gangSize++  readMD   (MDUnit n) i+   = check (here "readMD")  n i+   $ return ()++  writeMD  (MDUnit n) i ()+   = check (here "writeMD") n i+   $ return ()++  unsafeFreezeMD (MDUnit n)+   = return $ DUnit n++  sizeD  = error $ here "sizeD  undefined"+  sizeMD = error $ here "sizeMD undefined"+++-- | Yield a distributed unit.+unitD :: Gang -> Dist ()+unitD = DUnit . gangSize+{-# INLINE_DIST unitD #-}

@@ -0,0 +1,61 @@+{-# OPTIONS -Wall -fno-warn-orphans -fno-warn-missing-signatures #-}++-- | Distribution of Vectors.+module Data.Array.Parallel.Unlifted.Distributed.Types.Vector+        (lengthD)+where+import qualified Data.Array.Parallel.Base               as B+import Data.Array.Parallel.Unlifted.Distributed.Types.Prim+import Data.Array.Parallel.Unlifted.Distributed.Gang+import Data.Array.Parallel.Pretty+import Data.Array.Parallel.Unlifted.Sequential.Vector   as V+import qualified Data.Vector                            as BV+import qualified Data.Vector.Mutable                    as MBV+import Prelude                                          as P+import Control.Monad+++here :: String -> String+here s = "Data.Array.Parallel.Unlifted.Distributed.Types.Vector." P.++ s+++instance Unbox a => DT (V.Vector a) where+  data Dist  (Vector a)   = DVector  !(Dist  Int)   !(BV.Vector      (Vector a))+  data MDist (Vector a) s = MDVector !(MDist Int s) !(MBV.STVector s (Vector a))++  indexD str (DVector _ a) i+   = B.check (here ("indexD[Vector]/" P.++ str)) (BV.length a) i $ a BV.! i++  newMD g+   = liftM2 MDVector+        (newMD g) +        (MBV.replicate (gangSize g) (error "MDist (Vector a) - uninitalised"))++  readMD (MDVector _ marr)+   = MBV.read marr++  writeMD (MDVector mlen marr) i a +   = do writeMD mlen i (V.length a)+        MBV.write marr i $! a++  unsafeFreezeMD (MDVector len a)+   = liftM2 DVector (unsafeFreezeMD len)+                    (BV.unsafeFreeze a)++  sizeD  (DVector  _ a) = BV.length  a+  sizeMD (MDVector _ a) = MBV.length a++  measureD xs           = "Vector " P.++ show (V.length xs)+++instance (Unbox a, Show a) => PprPhysical (Dist (V.Vector a)) where+ pprp (DVector (DInt lengths) chunks)+  = text "DVector"+  $$ (nest 8 $ vcat+        [ text "lengths:" <+> (text $ show $ V.toList lengths)+        , text "chunks: " <+> (text $ show $ BV.toList $ BV.map V.toList chunks) ])+++-- | Yield the distributed length of a distributed array.+lengthD :: Unbox a => Dist (Vector a) -> Dist Int+lengthD (DVector l _) = l

@@ -0,0 +1,245 @@+{-# OPTIONS -Wall -fno-warn-orphans -fno-warn-missing-signatures #-}+{-# LANGUAGE CPP #-}+#include "fusion-phases.h"++-- | Operations on Distributed Segment Descriptors+module Data.Array.Parallel.Unlifted.Distributed.USSegd +        (splitSSegdOnElemsD)+where+import Data.Array.Parallel.Unlifted.Distributed.Arrays+import Data.Array.Parallel.Unlifted.Distributed.Combinators+import Data.Array.Parallel.Unlifted.Distributed.Types+import Data.Array.Parallel.Unlifted.Distributed.Gang+import Data.Array.Parallel.Unlifted.Sequential.USSegd                   (USSegd)+import Data.Array.Parallel.Unlifted.Sequential.Vector                   (Vector)+import Data.Array.Parallel.Base+import Data.Bits                                                        (shiftR)+import Control.Monad                                                    (when)+import Data.Array.Parallel.Unlifted.Distributed.Types.USSegd            ()+import qualified Data.Array.Parallel.Unlifted.Sequential.USegd          as USegd+import qualified Data.Array.Parallel.Unlifted.Sequential.USSegd         as USSegd+import qualified Data.Array.Parallel.Unlifted.Sequential.Vector         as Seq++here :: String -> String+here s = "Data.Array.Parallel.Unlifted.Distributed.USSegd." ++ s++-------------------------------------------------------------------------------+-- | Split a segment descriptor across the gang, element wise.+--   We try to put the same number of elements on each thread, which means+--   that segments are sometimes split across threads.+--+--   Each thread gets a slice of segment descriptor, the segid of the first +--   slice, and the offset of the first slice in its segment.+--   +--   Example:+--    In this picture each X represents 5 elements, and we have 5 segements in total.+--+-- @   segs:    ----------------------- --- ------- --------------- -------------------+--    elems:  |X X X X X X X X X|X X X X X X X X X|X X X X X X X X X|X X X X X X X X X|+--            |     thread1     |     thread2     |     thread3     |     thread4     |+--    segid:  0                 0                 3                 4+--    offset: 0                 45                0                 5+--+--    pprp $ splitSegdOnElemsD theGang +--          $ lengthsToUSegd $ fromList [60, 10, 20, 40, 50 :: Int]+--+--     segd:    DUSegd lengths:  DVector lengths: [1,3,2,1]+--                                        chunks:  [[45],[15,10,20],[40,5],[45]]+--                     indices:  DVector lengths: [1,3,2,1]+--                                        chunks:  [[0], [0,15,25], [0,40],[0]]+--                    elements:  DInt [45,45,45,45]+--+--     segids: DInt [0,0,3,4]     (segment id of first slice on thread)+--    offsets: DInt [0,45,0,5]    (offset of that slice in its segment)+-- @+--+splitSSegdOnElemsD :: Gang -> USSegd -> Dist ((USSegd,Int),Int)+splitSSegdOnElemsD g !segd +  = {-# SCC "splitSSegdOnElemsD" #-}+    imapD g mk (splitLenIdxD g (USegd.takeElements $ USSegd.takeUSegd segd))+  where +        -- Number of threads in gang.+        !nThreads = gangSize g+++        -- Build a USSegd from just the lengths, starts and sources fields.+        --   The indices and elems fields of the contained USegd are +        --   generated from the lengths.+        buildUSSegd :: Vector Int -> Vector Int -> Vector Int -> USSegd+        buildUSSegd lengths starts sources+                = USSegd.mkUSSegd starts sources+                $ USegd.fromLengths lengths++        -- Determine what elements go on a thread+        mk :: Int                  -- Thread index.+           -> (Int, Int)           -- Number of elements on this thread,+                                   --   and starting offset into the flat array.+           -> ((USSegd, Int), Int) -- Segd for this thread, segid of first slice,+                                   --   and offset of first slice.++        mk i (nElems, ixStart) +         = case chunk segd ixStart nElems (i == nThreads - 1) of+            (# lengths, starts, sources, l, o #) +             -> ((buildUSSegd lengths starts sources, l), o)++{-# NOINLINE splitSSegdOnElemsD #-}+--  NOINLINE because it's complicated and won't fuse with anything.+--  This function has a large body of code and we don't want to blow up+--  the client modules by inlining it everywhere.+++-------------------------------------------------------------------------------+-- | Determine what elements go on a thread.+--   The 'chunk' refers to the a chunk of the flat array, and is defined+--   by a set of segment slices. +--+--   Example:+--    In this picture each X represents 5 elements, and we have 5 segements in total.+--+-- @  segs:    ----------------------- --- ------- --------------- -------------------+--    elems:  |X X X X X X X X X|X X X X X X X X X|X X X X X X X X X|X X X X X X X X X|+--            |     thread1     |     thread2     |     thread3     |     thread4     |+--    segid:  0                 0                 3                 4+--    offset: 0                 45                0                 5+--    k:               0                 1                 3                 5+--    k':              1                 3                 5                 5+--    left:            0                 15                0                 45+--    right:           45                20                5                 0+--    left_len:        0                 1                 0                 1+--    left_off:        0                 45                0                 5+--    n':              1                 3                 2                 1+-- @+chunk   :: USSegd          -- ^ Segment descriptor of entire array.+        -> Int            -- ^ Starting offset into the flat array for the first+                          --    slice on this thread.+        -> Int            -- ^ Number of elements in this thread.+        -> Bool           -- ^ Whether this is the last thread in the gang.+        -> (# Vector Int  --    Lengths of segment slices, +            , Vector Int  --    Starting index of data in its vector+            , Vector Int  --    Source id+            , Int         --    segid of first slice+            , Int #)      --    offset of first slice.++chunk !ussegd !nStart !nElems is_last+  = (# lengths', starts', sources', k-left_len, left_off #)+  where+    -- Lengths of all segments.+    -- eg: [60, 10, 20, 40, 50]+    lengths     = USSegd.takeLengths ussegd++    -- Indices indices of all segments.+    -- eg: [0, 60, 70, 90, 130]+    indices     = USSegd.takeIndices ussegd++    -- Starting indices for all segments.+    starts      = USSegd.takeStarts ussegd++    -- Source ids for all segments.+    sources     = USSegd.takeSources ussegd+    +    -- Total number of segments defined by segment descriptor.+    -- eg: 5+    n    = Seq.length lengths++    -- Segid of the first seg that starts after the left of this chunk.+    k    = search nStart indices++    -- Segid of the first seg that starts after the right of this chunk.+    k'       | is_last     = n+             | otherwise   = search (nStart + nElems) indices++    -- The length of the left-most slice of this chunk.+    left     | k == n      = nElems+             | otherwise   = min ((Seq.index (here "chunk") indices k) - nStart) nElems++    -- The length of the right-most slice of this chunk.+    length_right   +             | k' == k     = 0+             | otherwise   = nStart + nElems - (Seq.index (here "chunk") indices (k'-1))++    -- Whether the first element in this chunk is an internal element of+    -- of a segment. Alternatively, indicates that the first element of +    -- the chunk is not the first element of a segment.            +    left_len | left == 0   = 0+             | otherwise   = 1++    -- If the first element of the chunk starts within a segment, +    -- then gives the index within that segment, otherwise 0.+    left_off | left == 0   = 0+             | otherwise   = nStart - (Seq.index (here "chunk") indices (k-1))++    -- How many segments this chunk straddles.+    n' = left_len + (k'-k)++    -- Create the lengths for this chunk by first copying out the lengths+    -- from the original segment descriptor. If the slices on the left+    -- and right cover partial segments, then we update the corresponding+    -- lengths.+    (!lengths', !starts', !sources')+     = runST (do+            -- Create a new array big enough to hold all the lengths for this chunk.+            mlengths' <- Seq.newM n'+            msources' <- Seq.newM n'+            mstarts'  <- Seq.newM n'++            -- If the first element is inside a segment, +            --   then update the length to be the length of the slice.+            when (left /= 0) +             $ do Seq.write mlengths' 0 left+                  Seq.write mstarts'  0 (Seq.index (here "chunk") starts  (k - left_len) + left_off)+                  Seq.write msources' 0 (Seq.index (here "chunk") sources (k - left_len))++            -- Copy out array lengths for this chunk.+            Seq.copy (Seq.mdrop left_len mlengths') (Seq.slice (here "chunk") lengths k (k'-k))+            Seq.copy (Seq.mdrop left_len mstarts')  (Seq.slice (here "chunk")  starts k (k'-k))+            Seq.copy (Seq.mdrop left_len msources') (Seq.slice (here "chunk") sources k (k'-k))++            -- If the last element is inside a segment, +            --   then update the length to be the length of the slice.+            when (length_right /= 0)+             $ do Seq.write mlengths' (n' - 1) length_right++            clengths' <- Seq.unsafeFreeze mlengths'+            cstarts'  <- Seq.unsafeFreeze mstarts'+            csources' <- Seq.unsafeFreeze msources'+            return (clengths', cstarts', csources'))++{-      = trace +        (render $ vcat+                [ text "CHUNK"+                , pprp segd+                , text "nStart:  " <+> int nStart+                , text "nElems:  " <+> int nElems+                , text "k:       " <+> int k+                , text "k':      " <+> int k'+                , text "left:    " <+> int left+                , text "right:   " <+> int right+                , text "left_len:" <+> int left_len+                , text "left_off:" <+> int left_off+                , text "n':      " <+> int n'+                , text ""]) lens'+-}++{-# INLINE chunk #-}+--  INLINE even though it should be inlined into splitSSegdOnElemsD anyway+--  because that function contains the only use.+++-------------------------------------------------------------------------------+-- O(log n).+-- Given a monotonically increasing vector of `Int`s,+-- find the first element that is larger than the given value.+-- +-- eg  search 75 [0, 60, 70, 90, 130] = 90+--     search 43 [0, 60, 70, 90, 130] = 60+--+search :: Int -> Vector Int -> Int+search !x ys = go 0 (Seq.length ys)+  where+    go i n | n <= 0        = i+           | Seq.index (here "search") ys mid < x+           = go (mid + 1) (n - half - 1)+           | otherwise     = go i half+      where+        half = n `shiftR` 1+        mid  = i + half

@@ -0,0 +1,349 @@+{-# OPTIONS -Wall -fno-warn-orphans -fno-warn-missing-signatures #-}+{-# LANGUAGE CPP #-}+#include "fusion-phases.h"++-- | Operations on Distributed Segment Descriptors+module Data.Array.Parallel.Unlifted.Distributed.USegd +        ( splitSegdOnSegsD+        , splitSegdOnElemsD+        , splitSD+        , joinSegdD+        , glueSegdD)+where+import Data.Array.Parallel.Unlifted.Distributed.Arrays+import Data.Array.Parallel.Unlifted.Distributed.Combinators+import Data.Array.Parallel.Unlifted.Distributed.Types+import Data.Array.Parallel.Unlifted.Distributed.Gang+import Data.Array.Parallel.Unlifted.Sequential.USegd                    (USegd)+import Data.Array.Parallel.Unlifted.Sequential.Vector                   (Vector, Unbox)+import Data.Array.Parallel.Base+import Data.Bits                                                        (shiftR)+import Control.Monad                                                    (when)+import qualified Data.Array.Parallel.Unlifted.Distributed.Types.USegd   as DUSegd+import qualified Data.Array.Parallel.Unlifted.Sequential.USegd          as USegd+import qualified Data.Array.Parallel.Unlifted.Sequential.Vector         as Seq++here :: String -> String+here s = "Data.Array.Parallel.Unlifted.Distributed.USegd." ++ s++-------------------------------------------------------------------------------+-- | Split a segment descriptor across the gang, segment wise.+--   Whole segments are placed on each thread, and we try to balance out+--   the segments so each thread has the same number of array elements.+--+--   We don't split segments across threads, as this would limit our ability+--   to perform intra-thread fusion of lifted operations. The down side+--   of this is that if we have few segments with an un-even size distribution+--   then large segments can cause the gang to become unbalanced.+--+--   In the following example the segment with size 100 dominates and+--   unbalances the gang. There is no reason to put any segments on the+--   the last thread because we need to wait for the first to finish anyway.+--+--   @ > pprp $ splitSegdOnSegsD theGang+--            $ lengthsToUSegd $ fromList [100, 10, 20, 40, 50  :: Int]+-- +--     DUSegd lengths:   DVector lengths:  [ 1,    3,         1,  0]+--                                chunks:  [[100],[10,20,40],[50],[]]+-- +--            indices:   DVector lengths:  [1,3,1,0]+--                                chunks:  [[0],  [0,10,30], [0], []]+--+--            elements:  DInt [100,70,50,0]+--   @+--+--  NOTE: This splitSegdOnSegsD function isn't currently used.+--+splitSegdOnSegsD :: Gang -> USegd -> Dist USegd+splitSegdOnSegsD g !segd +  = mapD g USegd.fromLengths+  $ splitAsD g d lens+  where+    !d   = snd+         . mapAccumLD g chunks 0+         . splitLenD g+         $ USegd.takeElements segd++    n    = USegd.length segd+    lens = USegd.takeLengths segd++    chunks !i !k +      = let !j = go i k+        in  (j,j-i)++    go !i !k | i >= n    = i+             | m == 0    = go (i+1) k+             | k <= 0    = i+             | otherwise = go (i+1) (k-m)+      where+        m = Seq.index (here "splitSegdOnSegsD") lens i+{-# NOINLINE splitSegdOnSegsD #-}+++-------------------------------------------------------------------------------+-- | Split a segment descriptor across the gang, element wise.+--   We try to put the same number of elements on each thread, which means+--   that segments are sometimes split across threads.+--+--   Each thread gets a slice of segment descriptor, the segid of the first +--   slice, and the offset of the first slice in its segment.+--   +--   Example:+--    In this picture each X represents 5 elements, and we have 5 segements in total.+--+-- @  segs:    ----------------------- --- ------- --------------- -------------------+--    elems:  |X X X X X X X X X|X X X X X X X X X|X X X X X X X X X|X X X X X X X X X|+--            |     thread1     |     thread2     |     thread3     |     thread4     |+--    segid:  0                 0                 3                 4+--    offset: 0                 45                0                 5+--+--    pprp $ splitSegdOnElemsD theGang4+--          $ lengthsToUSegd $ fromList [60, 10, 20, 40, 50 :: Int]+--+--     segd:    DUSegd lengths:  DVector lengths: [1,3,2,1]+--                                        chunks:  [[45],[15,10,20],[40,5],[45]]+--                     indices:  DVector lengths: [1,3,2,1]+--                                        chunks:  [[0], [0,15,25], [0,40],[0]]+--                    elements:  DInt [45,45,45,45]+--+--     segids: DInt [0,0,3,4]     (segment id of first slice on thread)+--    offsets: DInt [0,45,0,5]    (offset of that slice in its segment)+-- @+--+splitSegdOnElemsD :: Gang -> USegd -> Dist ((USegd,Int),Int)+splitSegdOnElemsD g !segd +  = {-# SCC "splitSegdOnElemsD" #-} +    imapD g mk (splitLenIdxD g (USegd.takeElements segd))+  where +        -- Number of threads in gang.+        !nThreads = gangSize g++        -- Determine what elements go on a thread+        mk :: Int                  -- Thread index.+           -> (Int, Int)           -- Number of elements on this thread,+                                   --   and starting offset into the flat array.+           -> ((USegd, Int), Int)  -- Segd for this thread, segid of first slice,+                                   --   and offset of first slice.++        mk i (nElems, ixStart) +         = case getChunk segd ixStart nElems (i == nThreads - 1) of+            (# lens, l, o #) -> ((USegd.fromLengths lens, l), o)++{-# NOINLINE splitSegdOnElemsD #-}+--  NOINLINE because this function has a large body of code and we don't want+--  to blow up the client modules by inlining it everywhere.+++-------------------------------------------------------------------------------+-- | Determine what elements go on a thread.+--   The 'chunk' refers to the a chunk of the flat array, and is defined+--   by a set of segment slices. +--+--   Example:+--    In this picture each X represents 5 elements, and we have 5 segements in total.+--+-- @+--    segs:    ----------------------- --- ------- --------------- -------------------+--    elems:  |X X X X X X X X X|X X X X X X X X X|X X X X X X X X X|X X X X X X X X X|+--            |     thread1     |     thread2     |     thread3     |     thread4     |+--    segid:  0                 0                 3                 4+--    offset: 0                 45                0                 5+--    k:               0                 1                 3                 5+--    k':              1                 3                 5                 5+--    left:            0                 15                0                 45+--    right:           45                20                5                 0+--    left_len:        0                 1                 0                 1+--    left_off:        0                 45                0                 5+--    n':              1                 3                 2                 1+-- @+getChunk+        :: USegd          -- ^ Segment descriptor of entire array.+        -> Int            -- ^ Starting offset into the flat array for the first+                          --   slice on this thread.+        -> Int            -- ^ Number of elements in this thread.+        -> Bool           -- ^ Whether this is the last thread in the gang.+        -> (# Vector Int  --   Lengths of segment slices, +            , Int         --     segid of first slice,+            , Int #)      --     offset of first slice.++getChunk !segd !nStart !nElems is_last+  = (# lens'', k-left_len, left_off #)+  where+    -- Lengths of all segments.+    -- eg: [60, 10, 20, 40, 50]+    !lens = USegd.takeLengths segd++    -- Indices indices of all segments.+    -- eg: [0, 60, 70, 90, 130]+    !idxs = USegd.takeIndices segd+    +    -- Total number of segments defined by segment descriptor.+    -- eg: 5+    !n    = Seq.length lens++    -- Segid of the first seg that starts after the left of this chunk.+    !k    = search nStart idxs++    -- Segid of the first seg that starts after the right of this chunk.+    !k'       | is_last     = n+              | otherwise   = search (nStart + nElems) idxs++    -- The length of the left-most slice of this chunk.+    !left     | k == n      = nElems+              | otherwise   = min ((Seq.index (here "getChunk") idxs k) - nStart) nElems++    -- The length of the right-most slice of this chunk.+    !right    | k' == k     = 0+              | otherwise   = nStart + nElems - (Seq.index (here "getChunk") idxs (k'-1))++    -- Whether the first element in this chunk is an internal element of+    -- of a segment. Alternatively, indicates that the first element of +    -- the chunk is not the first element of a segment.            +    !left_len | left == 0   = 0+              | otherwise   = 1++    -- If the first element of the chunk starts within a segment, +    -- then gives the index within that segment, otherwise 0.+    !left_off | left == 0   = 0+              | otherwise   = nStart - (Seq.index (here "getChunk") idxs (k-1))++    -- How many segments this chunk straddles.+    !n' = left_len + (k'-k)++    -- Create the lengths for this chunk by first copying out the lengths+    -- from the original segment descriptor. If the slices on the left+    -- and right cover partial segments, then we update the corresponding+    -- lengths.+    !lens' +     = runST (do+            -- Create a new array big enough to hold all the lengths for this chunk.+            !mlens' <- Seq.newM n'++            -- If the first element is inside a segment, +            --   then update the length to be the length of the slice.+            when (left /= 0) +             $ Seq.write mlens' 0 left++            -- Copy out array lengths for this chunk.+            Seq.copy (Seq.mdrop left_len mlens')+                     (Seq.slice "getChunk" lens k (k'-k))++            -- If the last element is inside a segment, +            --   then update the length to be the length of the slice.+            when (right /= 0)+             $ Seq.write mlens' (n' - 1) right++            Seq.unsafeFreeze mlens')++    !lens'' = lens'+{-      = trace +        (render $ vcat+                [ text "CHUNK"+                , pprp segd+                , text "nStart:  " <+> int nStart+                , text "nElems:  " <+> int nElems+                , text "k:       " <+> int k+                , text "k':      " <+> int k'+                , text "left:    " <+> int left+                , text "right:   " <+> int right+                , text "left_len:" <+> int left_len+                , text "left_off:" <+> int left_off+                , text "n':      " <+> int n'+                , text ""]) lens'+-}++{-# INLINE getChunk #-}+--  INLINE even though it should be inlined into splitSSegdOnElemsD anyway+--  because that function contains the only use.+++-------------------------------------------------------------------------------+-- O(log n). Given a monotonically increasing vector of `Int`s,+-- find the first element that is larger than the given value.+-- +-- eg  search 75 [0, 60, 70, 90, 130] = 90+--     search 43 [0, 60, 70, 90, 130] = 60+--+search :: Int -> Vector Int -> Int+search !x ys = go 0 (Seq.length ys)+  where+    go i n | n <= 0        = i++           | Seq.index (here "search") ys mid < x  +           = go (mid + 1) (n - half - 1)++           | otherwise     = go i half+      where+        half = n `shiftR` 1+        mid  = i + half+++-------------------------------------------------------------------------------+-- | time O(segs)+--   Join a distributed segment descriptor into a global one.+--   This simply joins the distributed lengths and indices fields, but does+--   not reconstruct the original segment descriptor as it was before splitting.+-- +-- @ > pprp $ joinSegdD theGang4 +--         $ fstD $ fstD $ splitSegdOnElemsD theGang+--         $ lengthsToUSegd $ fromList [60, 10, 20, 40, 50]+-- +--   USegd lengths:  [45,15,10,20,40,5,45]+--         indices:  [0,45,60,70,90,130,135]+--         elements: 180+-- @+-- +-- TODO: sequential runtime is O(segs) due to application of lengthsToUSegd+-- +joinSegdD :: Gang -> Dist USegd -> USegd+joinSegdD gang+        = USegd.fromLengths+        . joinD gang unbalanced+        . mapD  gang USegd.takeLengths+{-# INLINE_DIST joinSegdD #-}+++-------------------------------------------------------------------------------+-- | Glue a distributed segment descriptor back into the original global one.+--   Prop:  glueSegdD gang $ splitSegdOnElems gang usegd = usegd+--+--   NOTE: This is runs sequentially and should only be used for testing purposes.+--+glueSegdD :: Gang -> Dist ((USegd, Int), Int)  -> Dist USegd+glueSegdD gang bundle+ = let  !usegd           = fstD $ fstD $ bundle+        !lengths         = DUSegd.takeLengthsD usegd+                +        !firstSegOffsets = sndD bundle++        -- | Whether the last segment in this chunk extends into the next chunk.+        segSplits :: Dist Bool+        !segSplits+         = generateD_cheap gang $ \ix +         -> if ix >= sizeD lengths - 1+             then False+             else indexD (here "glueSegdD") firstSegOffsets (ix + 1) /= 0++        !lengths'       = fst $ carryD gang (+)                  0 segSplits lengths+        !dusegd'        = mapD gang USegd.fromLengths lengths'++  in    dusegd'+{-# INLINE_DIST glueSegdD #-}+++-------------------------------------------------------------------------------+splitSD :: Unbox a => Gang -> Dist USegd -> Vector a -> Dist (Vector a)+splitSD g dsegd xs+        = splitAsD g (DUSegd.takeElementsD dsegd) xs+{-# INLINE_DIST splitSD #-}++{-# RULES++"splitSD/splitJoinD" forall g d f xs.+  splitSD g d (splitJoinD g f xs) = f (splitSD g d xs)++"splitSD/Seq.zip" forall g d xs ys.+  splitSD g d (Seq.zip xs ys) = zipWithD g Seq.zip (splitSD g d xs)+                                             (splitSD g d ys)++  #-}

@@ -1,39 +1,75 @@--- | Parallel operations on unlifted arrays-module Data.Array.Parallel.Unlifted.Parallel (-  UPSegd, UPSel2, UPSelRep2,--  bpermuteUP, updateUP,--  enumFromToUP, enumFromThenToUP, enumFromStepLenUP, enumFromStepLenEachUP,--  mapUP, filterUP, packUP, combineUP, combine2UP,-  zipWithUP, foldUP, scanUP,+{-# LANGUAGE CPP #-}+#include "fusion-phases.h" -  andUP, sumUP,+-- | Parallel operations on unlifted arrays+--+--   * This is an internal API and shouldn't need to be used directly.+--     Client programs should use "Data.Array.Parallel.Unlifted" -  tagsUPSel2, indicesUPSel2, elementsUPSel2_0, elementsUPSel2_1,-  selUPSel2, repUPSel2, mkUPSel2,-  mkUPSelRep2, indicesUPSelRep2, elementsUPSelRep2_0, elementsUPSelRep2_1,+--   NOTE: Each of the sections in the export list corresponds to one of the+--         Parallel modules, and the names are in the same order as in those+--         modules.+--+module Data.Array.Parallel.Unlifted.Parallel +        ( -- * Basics+          lengthUP+        , nullUP+        , emptyUP+        , indexedUP+        , replicateUP+        , repeatUP+        , interleaveUP+  +          -- * Combinators+        , mapUP+        , filterUP+        , packUP+        , combineUP,  combine2UP+        , zipWithUP+        , foldUP,     fold1UP+        , foldlUP,    foldl1UP+        , scanUP+  +          -- * Enum+        , enumFromToUP+        , enumFromThenToUP+        , enumFromStepLenUP+        , enumFromStepLenEachUP+  +          -- * Permute+        , bpermuteUP+        , updateUP -  lengthUPSegd, lengthsUPSegd, indicesUPSegd, elementsUPSegd,-  segdUPSegd, distUPSegd,-  lengthsToUPSegd, mkUPSegd,- -  replicateSUP, replicateRSUP, appendSUP, indicesSUP,-  foldSUP, foldRUP, fold1SUP, sumSUP, sumRUP,+          -- * Segmented+        , replicateRSUP+        , appendSUP+        , foldRUP+        , sumRUP -  indexedUP, replicateUP, repeatUP, interleaveUP,+          -- * Index and Extracts+        , indexsFromVector+        , indexsFromVectorsUPVSegd+        , extractsFromNestedUPSSegd+        , extractsFromVectorsUPSSegd+        , extractsFromVectorsUPVSegd -  dropUP-) where-import Data.Array.Parallel.Unlifted.Parallel.Permute-import Data.Array.Parallel.Unlifted.Parallel.Combinators+          -- * Subarrays+        , dropUP+  +          -- * Sums+        , andUP+        , orUP+        , allUP,     anyUP+        , sumUP,     productUP+        , maximumUP, maximumByUP+        , maximumIndexByUP)+where import Data.Array.Parallel.Unlifted.Parallel.Basics-import Data.Array.Parallel.Unlifted.Parallel.Sums+import Data.Array.Parallel.Unlifted.Parallel.Combinators import Data.Array.Parallel.Unlifted.Parallel.Enum+import Data.Array.Parallel.Unlifted.Parallel.Permute+import Data.Array.Parallel.Unlifted.Parallel.Extracts import Data.Array.Parallel.Unlifted.Parallel.Segmented+import Data.Array.Parallel.Unlifted.Parallel.Text       () import Data.Array.Parallel.Unlifted.Parallel.Subarrays-import Data.Array.Parallel.Unlifted.Parallel.UPSegd-import Data.Array.Parallel.Unlifted.Parallel.UPSel-import Data.Array.Parallel.Unlifted.Parallel.Text ()-+import Data.Array.Parallel.Unlifted.Parallel.Sums

@@ -2,66 +2,80 @@ #include "fusion-phases.h"  -- | Basic operations on parallel unlifted arrays.-module Data.Array.Parallel.Unlifted.Parallel.Basics (-  lengthUP, nullUP, indexedUP,-  replicateUP, repeatUP, interleaveUP-) where-+module Data.Array.Parallel.Unlifted.Parallel.Basics +        ( emptyUP+        , replicateUP+        , repeatUP+        , lengthUP+        , nullUP+        , interleaveUP+        , indexedUP)+where import Data.Array.Parallel.Unlifted.Sequential.Vector as Seq import Data.Array.Parallel.Unlifted.Distributed-import Data.Array.Parallel.Unlifted.Parallel.Combinators ( mapUP )-import Data.Array.Parallel.Unlifted.Parallel.Enum        ( enumFromToUP )-import Data.Array.Parallel.Unlifted.Parallel.Permute     ( bpermuteUP )+import Data.Array.Parallel.Unlifted.Parallel.Combinators (mapUP)+import Data.Array.Parallel.Unlifted.Parallel.Enum        (enumFromToUP)+import Data.Array.Parallel.Unlifted.Parallel.Permute     (bpermuteUP)+import GHC.Base                                           (remInt) -import GHC.Base ( remInt ) --- NOTE: some of the functions are exactly the same as the U version---- | Test whether the given array is empty-nullUP :: Unbox e => Vector e -> Bool-nullUP  = (== 0) . Seq.length---- | Yield an empty array+-- | O(1). Construct an empty array. emptyUP :: Unbox e => Vector e emptyUP = Seq.new 0 (const $ return ())--lengthUP :: Unbox e => Vector e -> Int-lengthUP = Seq.length+{-# INLINE_UP emptyUP #-}   -- | Yield an array where all elements contain the same value replicateUP :: Unbox e => Int -> e -> Vector e+replicateUP n !e +        = joinD theGang balanced+        . mapD theGang (\n' ->Seq.replicate n' e)+        $ splitLenD theGang n {-# INLINE_UP replicateUP #-}-replicateUP n !e = joinD theGang balanced-                 . mapD theGang (\n ->Seq.replicate n e)-                 $ splitLenD theGang n   -- | Repeat an array the given number of times. repeatUP :: Unbox e => Int -> Vector e -> Vector e-{-# INLINE_UP repeatUP #-}-repeatUP n es = seq m-              . bpermuteUP es-              . mapUP (\i -> i `remInt` m)-              $ enumFromToUP 0 (m*n-1)+repeatUP n es +        = seq m+        . bpermuteUP es+        . mapUP (\i -> i `remInt` m)+        $ enumFromToUP 0 (m*n-1)   where     m = Seq.length es+{-# INLINE_UP repeatUP #-} ++-- | O(1). Take the length of an array.+lengthUP :: Unbox e => Vector e -> Int+lengthUP = Seq.length+{-# INLINE_UP lengthUP #-}+++-- | O(1). Test whether the given array is empty+nullUP :: Unbox e => Vector e -> Bool+nullUP  = (== 0) . Seq.length+{-# INLINE_UP nullUP #-}++ -- | Interleave elements of two arrays interleaveUP :: Unbox e => Vector e -> Vector e -> Vector e+interleaveUP xs ys+        = joinD theGang unbalanced+        $ zipWithD theGang Seq.interleave+                (splitD theGang balanced xs)+                (splitD theGang balanced ys) {-# INLINE_UP interleaveUP #-}-interleaveUP xs ys = joinD theGang unbalanced-                     (zipWithD theGang Seq.interleave-                       (splitD theGang balanced xs)-                       (splitD theGang balanced ys))-   ++ -- | Associate each element of the array with its index indexedUP :: (DT e, Unbox e) => Vector e -> Vector (Int,e)-{-# INLINE_U indexedUP #-}-indexedUP = splitJoinD theGang indexedFn -  where+indexedUP + = splitJoinD theGang indexedFn + where     sizes  arr   = fst $ scanD theGang (+) 0 $ lengthD arr-    indexedFn = \arr -> (zipWithD theGang (\o -> Seq.map (\(x,y) -> (x + o, y))) (sizes arr) $ -                        mapD theGang Seq.indexed arr)--+    indexedFn    = \arr -> zipWithD theGang +                                (\o -> Seq.map (\(x,y) -> (x + o, y)))+                                (sizes arr) +                         $  mapD theGang Seq.indexed arr+{-# INLINE_UP indexedUP #-}

@@ -1,32 +1,39 @@ {-# LANGUAGE CPP #-} #include "fusion-phases.h" --- | Parallel combinators for unlifted arrays-module Data.Array.Parallel.Unlifted.Parallel.Combinators (-  mapUP, filterUP, packUP, combineUP, combine2UP,-  zipWithUP, foldUP, fold1UP, foldl1UP, scanUP-) where-+-- | Parallel combinators for unlifted arrays. +module Data.Array.Parallel.Unlifted.Parallel.Combinators +        ( mapUP+        , filterUP+        , packUP+        , combineUP, combine2UP+        , zipWithUP+        , foldUP, foldlUP, fold1UP, foldl1UP+        , scanUP)+where import Data.Array.Parallel.Base-import Data.Array.Parallel.Unlifted.Sequential.Vector as Seq import Data.Array.Parallel.Unlifted.Distributed import Data.Array.Parallel.Unlifted.Parallel.UPSel+import Data.Array.Parallel.Unlifted.Sequential.Vector as Seq +here :: String -> String+here s = "Data.Array.Parallel.Unlifted.Parallel.Combinators." Prelude.++ s --- | Apply a worker to all elements of a vector.++-- | Apply a worker to all elements of an array. mapUP :: (Unbox a, Unbox b) => (a -> b) -> Vector a -> Vector b-{-# INLINE mapUP #-} mapUP f xs          = splitJoinD theGang (mapD theGang (Seq.map f)) xs+{-# INLINE_UP mapUP #-}   -- | Keep elements that match the given predicate. filterUP :: Unbox a => (a -> Bool) -> Vector a -> Vector a-{-# INLINE filterUP #-} filterUP f         = joinD  theGang unbalanced         . mapD   theGang (Seq.filter f)         . splitD theGang unbalanced+{-# INLINE_UP filterUP #-}   -- | Take elements of an array where a flag value is true, and pack them into@@ -35,9 +42,9 @@ --   * The souce and flag arrays must have the same length, but this is not checked. -- packUP :: Unbox e => Vector e -> Vector Bool -> Vector e-{-# INLINE_UP packUP #-} packUP xs flags          = Seq.fsts . filterUP snd $ Seq.zip xs flags+{-# INLINE_UP packUP #-}   -- | Combine two vectors based on a selector. @@ -48,10 +55,13 @@ --     but this is not checked. --   combineUP :: Unbox a => Vector Bool -> Vector a -> Vector a -> Vector a-{-# INLINE combineUP #-} combineUP flags xs ys -        = combine2UP tags (mkUPSelRep2 tags) xs ys+        = checkEq (here "combineUP")+                ("tags length /= sum of args length")+                (Seq.length flags) (Seq.length xs + Seq.length ys)+        $ combine2UP tags (mkUPSelRep2 tags) xs ys         where tags = Seq.map (fromBool . not) flags+{-# INLINE combineUP #-}   -- | Combine two vectors based on a selector. @@ -59,28 +69,29 @@ --   * The data vectors must have enough elements to satisfy the selector, --     but this is not checked. -----   TODO: What is the difference between the Tag and the UPSelRep2?--- combine2UP :: Unbox a => Vector Tag -> UPSelRep2 -> Vector a -> Vector a -> Vector a-{-# INLINE_UP combine2UP #-} combine2UP tags rep !xs !ys -        = joinD    theGang balanced+        = checkEq (here "combine2UP")+                ("tags length /= sum of args length")+                (Seq.length tags) (Seq.length xs + Seq.length ys)+        $ joinD    theGang balanced         $ zipWithD theGang go rep         $ splitD   theGang balanced tags         where   go ((i,j), (m,n)) ts                   = Seq.combine2ByTag ts -                        (Seq.slice xs i m)-                        (Seq.slice ys j n)-    +                        (Seq.slice (here "combine2UP") xs i m)+                        (Seq.slice (here "combine2UP") ys j n)+{-# INLINE_UP combine2UP #-} --- | Combine two vectors into a third.++-- | Apply a worker function to correponding elements of two arrays. zipWithUP :: (Unbox a, Unbox b, Unbox c)            => (a -> b -> c) -> Vector a -> Vector b -> Vector c-{-# INLINE zipWithUP #-} zipWithUP f xs ys         = splitJoinD theGang                  (mapD theGang (Seq.map (uncurry f)))                  (Seq.zip xs ys)+{-# INLINE_UP zipWithUP #-}   -- | Undirected fold.@@ -88,6 +99,7 @@ --   standard fold function from the Haskell Prelude. -- --   * The worker function must be associative.+-- --   * The provided starting element must be neutral with respect to the worker. --     For example 0 is neutral wrt (+) and 1 is neutral wrt (*). --@@ -96,69 +108,72 @@ --   then we fold together all the results in the main thread. -- foldUP  :: (Unbox a, DT a) => (a -> a -> a) -> a -> Vector a -> a-{-# INLINE foldUP #-} foldUP f !z xs         = foldD theGang f                 (mapD   theGang (Seq.fold f z)                 (splitD theGang unbalanced xs))+{-# INLINE_UP foldUP #-}   -- | Left fold over an array.  -- --   * If the vector is empty then this returns the provided neural element.+-- --   * The worker function must be associative.+-- --   * The provided starting element must be neutral with respect to the worker, --     see `foldUP` for discussion. -- foldlUP :: (DT a, Unbox a) => (a -> a -> a) -> a -> Vector a -> a-{-# INLINE_UP foldlUP #-} foldlUP f z arr    | Seq.null arr = z   | otherwise    = foldl1UP f arr+{-# INLINE_UP foldlUP #-}   -- | Alias for `foldl1UP` fold1UP :: (DT a, Unbox a) => (a -> a -> a) -> Vector a -> a-{-# INLINE fold1UP #-} fold1UP = foldl1UP+{-# INLINE_UP fold1UP #-}   -- | Left fold over an array, using the first element of the vector as the --   neural element. -- --   * If the vector contains no elements then you'll get a bounds-check error.+-- --   * The worker function must be associative.+-- --   * The provided starting element must be neutral with respect to the worker, --     see `foldUP` for discussion. -----   TODO: The two type class constraints are in a different order. Does that matter?--- foldl1UP :: (DT a, Unbox a) => (a -> a -> a) -> Vector a -> a-{-# INLINE_U foldl1UP #-} foldl1UP f arr          = (maybe z (f z)-        . foldD  theGang combine+        . foldD  theGang combine'         . mapD   theGang (Seq.foldl1Maybe f)         . splitD theGang unbalanced) arr         where-                z = arr ! 0-                combine (Just x) (Just y) = Just (f x y)-                combine (Just x) Nothing  = Just x-                combine Nothing  (Just y) = Just y-                combine Nothing  Nothing  = Nothing+                z = Seq.index (here "fold1UP") arr 0+                combine' (Just x) (Just y) = Just (f x y)+                combine' (Just x) Nothing  = Just x+                combine' Nothing  (Just y) = Just y+                combine' Nothing  Nothing  = Nothing+{-# INLINE_UP foldl1UP #-}   -- | Prefix scan. Similar to fold, but produce an array of the intermediate states. -- --   * The worker function must be associative.+--  --   * The provided starting element must be neutral with respect to the worker, --     see `foldUP` for discussion. -- scanUP :: (DT a, Unbox a) => (a -> a -> a) -> a -> Vector a -> Vector a-{-# INLINE_UP scanUP #-} scanUP f z          = splitJoinD theGang go         where   go xs = let (ds,zs) = unzipD $ mapD theGang (Seq.scanRes f z) xs                             zs'     = fst (scanD theGang f z zs)                         in  zipWithD theGang (Seq.map . f) zs' ds+{-# INLINE_UP scanUP #-} 

@@ -2,59 +2,59 @@ #include "fusion-phases.h"  -- | Enum-related parallel operations on unlifted arrays-module Data.Array.Parallel.Unlifted.Parallel.Enum (-  enumFromToUP, enumFromThenToUP, enumFromStepLenUP, enumFromStepLenEachUP    -) where+module Data.Array.Parallel.Unlifted.Parallel.Enum +        ( enumFromToUP+        , enumFromThenToUP+        , enumFromStepLenUP+        , enumFromStepLenEachUP)+where import Data.Array.Parallel.Unlifted.Sequential.Vector as Seq-import Data.Array.Parallel.Unlifted.Distributed (-  mapD, scanD, zipD, splitLenIdxD, joinD, splitD, balanced, unbalanced,-  theGang)-import Data.Array.Parallel.Unlifted.Parallel.Combinators (-  mapUP)-import GHC.Base ( divInt )+import Data.Array.Parallel.Unlifted.Distributed+import Data.Array.Parallel.Unlifted.Parallel.Combinators (mapUP)+import GHC.Base                                          (divInt)   delay_inline :: a -> a-{-# INLINE [0] delay_inline #-} delay_inline x = x+{-# INLINE [0] delay_inline #-}   enumFromToUP :: (Unbox a, Enum a) => a -> a -> Vector a-{-# INLINE enumFromToUP #-}-enumFromToUP start end = mapUP toEnum (enumFromStepLenUP start' 1 len)-  where-    start' = fromEnum start-    end'   = fromEnum end-    len    = delay_inline max (end' - start' + 1) 0+enumFromToUP start end + = mapUP toEnum (enumFromStepLenUP start' 1 len)+ where  start' = fromEnum start+        end'   = fromEnum end+        len    = delay_inline max (end' - start' + 1) 0+{-# INLINE_UP enumFromToUP #-}   enumFromThenToUP :: (Unbox a, Enum a) => a -> a -> a -> Vector a-{-# INLINE enumFromThenToUP #-}-enumFromThenToUP start next end =-  mapUP toEnum (enumFromStepLenUP start' delta len)-  where-    start' = fromEnum start-    next'  = fromEnum next-    end'   = fromEnum end-    delta  = next' - start'-    -- distance between start' and end' expressed in deltas-    dist   = (end' - start' + delta) `divInt` delta-    len    = max dist 0+enumFromThenToUP start next end + = mapUP toEnum (enumFromStepLenUP start' delta len)+ where  start' = fromEnum start+        next'  = fromEnum next+        end'   = fromEnum end+        delta  = next' - start' +        -- distance between start' and end' expressed in deltas+        dist   = (end' - start' + delta) `divInt` delta+        len    = max dist 0+{-# INLINE_UP enumFromThenToUP #-} + enumFromStepLenUP :: Int -> Int -> Int -> Vector Int-{-# INLINE enumFromStepLenUP #-} enumFromStepLenUP start delta len =   joinD theGang balanced   (mapD theGang gen   (splitLenIdxD theGang len))   where     gen (n,i) = Seq.enumFromStepLen (i * delta + start) delta n+{-# INLINE_UP enumFromStepLenUP #-}  -enumFromStepLenEachUP :: Int -> Vector Int -> Vector Int -> Vector Int -> Vector Int-{-# INLINE enumFromStepLenEachUP #-}-enumFromStepLenEachUP n starts steps lens+enumFromStepLenEachUP +        :: Int -> Vector Int -> Vector Int -> Vector Int -> Vector Int+enumFromStepLenEachUP _n starts steps lens   = joinD theGang unbalanced   $ mapD theGang enum   $ splitD theGang unbalanced (Seq.zip (Seq.zip starts steps) lens)@@ -62,4 +62,5 @@     enum ps = let (qs, llens) = Seq.unzip ps                   (lstarts, lsteps) = Seq.unzip qs               in Seq.enumFromStepLenEach (Seq.sum llens) lstarts lsteps llens+{-# INLINE_UP enumFromStepLenEachUP #-} 

@@ -0,0 +1,104 @@+{-# LANGUAGE CPP #-}+#include "fusion-phases.h"++-- | Parallel combinators for segmented unboxed arrays+module Data.Array.Parallel.Unlifted.Parallel.Extracts +        ( -- * Scattered indexing+          indexsFromVector+        , indexsFromVectorsUPVSegd++          -- * Scattered extracts+        , extractsFromNestedUPSSegd+        , extractsFromVectorsUPSSegd+        , extractsFromVectorsUPVSegd)+where+import Data.Array.Parallel.Unlifted.Parallel.UPSSegd                    (UPSSegd)+import Data.Array.Parallel.Unlifted.Parallel.UPVSegd                    (UPVSegd)+import Data.Array.Parallel.Unlifted.Sequential.Vector                   as Seq+import Data.Array.Parallel.Unlifted.Vectors                             (Vectors)+import qualified Data.Array.Parallel.Unlifted.Parallel.UPSSegd          as UPSSegd+import qualified Data.Array.Parallel.Unlifted.Parallel.UPVSegd          as UPVSegd+import qualified Data.Array.Parallel.Unlifted.Sequential.UVSegd         as UVSegd+import qualified Data.Array.Parallel.Unlifted.Vectors                   as US+import qualified Data.Array.Parallel.Unlifted.Stream                    as US+import qualified Data.Array.Parallel.Unlifted.Sequential                as Seq+import qualified Data.Vector                                            as V+++-- Indexvs --------------------------------------------------------------------+-- | Lookup elements from a `Vector`.+--+--   TODO: make this parallel.+--+indexsFromVector+        :: Unbox a+        => Vector a -> Vector Int -> Vector a++indexsFromVector = Seq.indexsFromVector+++-- | Lookup elements from some `Vectors` through a `UPVSegd`.+--+--   TODO: make this parallel.+--+indexsFromVectorsUPVSegd +        :: (Unbox a, US.Unboxes a)+        => Vectors a -> UPVSegd -> Vector (Int, Int) -> Vector a++indexsFromVectorsUPVSegd vectors upvsegd vsrcixs+ = let  -- Because we're just doing indexing here, we don't need the culled+        -- vsegids or ussegd, and can just use the redundant version.+        !vsegids  = UPVSegd.takeVSegidsRedundant upvsegd+        !upssegd  = UPVSegd.takeUPSSegdRedundant upvsegd+        !ussegd   = UPSSegd.takeUSSegd upssegd+   in   Seq.unstream+         $ US.streamElemsFromVectors     vectors+         $ US.streamSrcIxsThroughUSSegd  ussegd+         $ US.streamSrcIxsThroughVSegids vsegids+         $ Seq.stream vsrcixs+{-# INLINE_U indexsFromVectorsUPVSegd #-}+++-- Extracts -------------------------------------------------------------------+-- | Copy segments from a nested vectors and concatenate them into a new array.+extractsFromNestedUPSSegd+        :: Unbox a+        => UPSSegd -> V.Vector (Vector a) -> Vector a++extractsFromNestedUPSSegd upssegd vectors+        = Seq.unstream +        $ US.streamSegsFromNestedUSSegd+                vectors+                (UPSSegd.takeUSSegd upssegd)+{-# INLINE_U extractsFromNestedUPSSegd #-}+++-- | TODO: make this parallel.+extractsFromVectorsUPSSegd+        :: (Unbox a, US.Unboxes a)+        => UPSSegd+        -> Vectors a+        -> Vector a++extractsFromVectorsUPSSegd upssegd vectors+        = Seq.extractsFromVectorsUSSegd+                (UPSSegd.takeUSSegd upssegd) +                vectors+{-# INLINE_UP extractsFromVectorsUPSSegd #-}+++-- | TODO: make this parallel.+extractsFromVectorsUPVSegd+        :: (Unbox a, US.Unboxes a)+        => UPVSegd+        -> Vectors a+        -> Vector a++extractsFromVectorsUPVSegd upvsegd vectors+        = Seq.unstream +        $ US.streamSegsFromVectorsUVSegd vectors+        $ UVSegd.mkUVSegd +                (UPVSegd.takeVSegidsRedundant upvsegd)+                (UPSSegd.takeUSSegd $ UPVSegd.takeUPSSegdRedundant upvsegd)+{-# INLINE_UP extractsFromVectorsUPVSegd #-}+

@@ -1,17 +1,13 @@-{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE CPP, ScopedTypeVariables #-}+#include "fusion-phases.h"  -- | Parallel permutations for unlifted arrays-module Data.Array.Parallel.Unlifted.Parallel.Permute (-  bpermuteUP, updateUP-) where+module Data.Array.Parallel.Unlifted.Parallel.Permute +        ( bpermuteUP, updateUP)+where import Data.Array.Parallel.Unlifted.Sequential.Vector as Seq import Data.Array.Parallel.Unlifted.Distributed --bpermuteUP :: Unbox a => Vector a -> Vector Int -> Vector a-{-# INLINE bpermuteUP #-}-bpermuteUP as is = splitJoinD theGang (bpermuteD theGang as) is- {-   We can't support this for arbitrary types. The problem is:   what happens if the second array maps multiple elements to the same position?@@ -29,8 +25,14 @@   atomic. Otherwise, we do a sequential update. -} +-- | Backwards permutation.+bpermuteUP :: Unbox a => Vector a -> Vector Int -> Vector a+bpermuteUP as is = splitJoinD theGang (bpermuteD theGang as) is+{-# INLINE_UP bpermuteUP #-}+++-- | Update elements in an array. updateUP :: forall a. Unbox a => Vector a -> Vector (Int,a) -> Vector a-{-# INLINE updateUP #-} updateUP as us   {- hasAtomicWriteMU (undefined :: a)    = atomicUpdateD theGang (splitD theGang unbalanced as)@@ -39,4 +41,5 @@    | otherwise   = Seq.update as us+{-# INLINE_UP updateUP #-} 

@@ -1,213 +1,142 @@ {-# LANGUAGE CPP #-}- #include "fusion-phases.h"  -- | Parallel combinators for segmented unboxed arrays-module Data.Array.Parallel.Unlifted.Parallel.Segmented (-  replicateSUP, replicateRSUP, appendSUP, indicesSUP,-  foldSUP, foldRUP, fold1SUP, sumSUP, sumRUP-) where-import Data.Array.Parallel.Unlifted.Sequential.Vector as Seq-import Data.Array.Parallel.Unlifted.Sequential.Segmented+module Data.Array.Parallel.Unlifted.Parallel.Segmented +        ( replicateRSUP+        , appendSUP+        , foldRUP+        , sumRUP)+where import Data.Array.Parallel.Unlifted.Distributed-import Data.Array.Parallel.Unlifted.Parallel.Combinators (-  mapUP, zipWithUP, packUP, combineUP)-import Data.Array.Parallel.Unlifted.Parallel.Sums (-  sumUP )-import Data.Array.Parallel.Unlifted.Parallel.Basics (-  replicateUP, repeatUP)-import Data.Array.Parallel.Unlifted.Parallel.Enum-import Data.Array.Parallel.Unlifted.Parallel.Permute ( bpermuteUP )-import Data.Array.Parallel.Unlifted.Parallel.UPSegd-import qualified Data.Vector.Fusion.Stream as S+import Data.Array.Parallel.Unlifted.Parallel.Basics+import Data.Array.Parallel.Unlifted.Parallel.UPSegd                     (UPSegd)+import Data.Array.Parallel.Unlifted.Sequential.USegd                    (USegd)+import Data.Array.Parallel.Unlifted.Sequential.Vector                   as Seq+import qualified Data.Array.Parallel.Unlifted.Parallel.UPSegd           as UPSegd+import qualified Data.Array.Parallel.Unlifted.Sequential                as Seq+import qualified Data.Array.Parallel.Unlifted.Sequential.USegd          as USegd import Data.Vector.Fusion.Stream.Monadic ( Stream(..), Step(..) ) import Data.Vector.Fusion.Stream.Size    ( Size(..) )-import Control.Monad.ST ( ST, runST )----- replicate ------------------------------------------------------------------+import qualified Data.Vector.Fusion.Stream                              as S --- | Segmented replication, using a segment descriptor.-replicateSUP :: Unbox a => UPSegd -> Vector a -> Vector a-{-# INLINE_UP replicateSUP #-}-replicateSUP segd !xs -  = joinD theGang balanced-  . mapD theGang rep-  $ distUPSegd segd-  where-    rep ((dsegd,di),_)-      = replicateSU dsegd (Seq.slice xs di (lengthUSegd dsegd))+here :: String -> String+here s = "Data.Array.Parallel.Unlifted.Parallel.Segmented." Prelude.++ s  +-- replicate ------------------------------------------------------------------ -- | Segmented replication. --   Each element in the vector is replicated the given number of times. --    --   @replicateRSUP 2 [1, 2, 3, 4, 5] = [1, 1, 2, 2, 3, 3, 4, 4, 5, 5]@ --+ --   TODO: make this efficient---  replicateRSUP :: Unbox a => Int -> Vector a -> Vector a-{-# INLINE_UP replicateRSUP #-} replicateRSUP n xs-        = replicateSUP (lengthsToUPSegd (replicateUP (Seq.length xs) n)) xs+        = UPSegd.replicateWithP (UPSegd.fromLengths (replicateUP (Seq.length xs) n)) xs+{-# INLINE_UP replicateRSUP #-}  --- append ---------------------------------------------------------------------+-- Append --------------------------------------------------------------------- -- | Segmented append. appendSUP         :: Unbox a-        => UPSegd       -- ^ segment descriptor of result array-        -> UPSegd       -- ^ segment descriptor of first array-        -> Vector a     -- ^ data of first array-        -> UPSegd       -- ^ segment descriptor of second array-        -> Vector a     -- ^ data of first array+        => UPSegd+        -> UPSegd -> Vector a+        -> UPSegd -> Vector a         -> Vector a -{-# INLINE_UP appendSUP #-} appendSUP segd !xd !xs !yd !ys   = joinD theGang balanced   . mapD  theGang append-  $ distUPSegd segd-  where append ((segd,seg_off),el_off)+  $ UPSegd.takeDistributed segd+  where append ((segd',seg_off),el_off)          = Seq.unstream-         $ appendSegS (segdUPSegd xd) xs-                      (segdUPSegd yd) ys-                      (elementsUSegd segd)+         $ appendSegS (UPSegd.takeUSegd xd) xs+                      (UPSegd.takeUSegd yd) ys+                      (USegd.takeElements segd')                       seg_off el_off+{-# INLINE_UP appendSUP #-}  +-- append --------------------------------------------------------------------- appendSegS         :: Unbox a      -        => USegd        -- ^ segment descriptor of first array-        -> Vector a     -- ^ data of first array-        -> USegd        -- ^ segment descriptor of second array-        -> Vector a     -- ^ data of second array-        -> Int          -- +        => USegd        -- ^ Segment descriptor of first array.+        -> Vector a     -- ^ Data of first array+        -> USegd        -- ^ Segment descriptor of second array.+        -> Vector a     -- ^ Data of second array.         -> Int         -> Int+        -> Int         -> S.Stream a -{-# INLINE_STREAM appendSegS #-} appendSegS !xd !xs !yd !ys !n seg_off el_off   = Stream next state (Exact n)   where-    !xlens = lengthsUSegd xd-    !ylens = lengthsUSegd yd+    !xlens = USegd.takeLengths xd+    !ylens = USegd.takeLengths yd +    {-# INLINE index1 #-}+    index1  = Seq.index (here "appendSegS")++    {-# INLINE index2 #-}+    index2  = Seq.index (here "appendSegS")+         state       | n == 0 = Nothing-      | el_off < xlens ! seg_off-      = let i = (indicesUSegd xd ! seg_off) + el_off-            j = indicesUSegd yd ! seg_off-            k = (lengthsUSegd xd ! seg_off) - el_off+      | el_off < xlens `index1` seg_off+      = let i = (USegd.takeIndices xd `index1` seg_off) + el_off+            j =  USegd.takeIndices yd `index1` seg_off+            k = (USegd.takeLengths xd `index1` seg_off) - el_off         in  Just (False, seg_off, i, j, k, n)        | otherwise       = let -- NOTE: *not* indicesUSegd xd ! (seg_off+1) since seg_off+1             -- might be out of bounds-            i       = (indicesUSegd xd ! seg_off) + (lengthsUSegd xd ! seg_off)-            el_off' = el_off - lengthsUSegd xd ! seg_off-            j       = (indicesUSegd yd ! seg_off) + el_off'-            k       = (lengthsUSegd yd ! seg_off) - el_off'+            i       = (USegd.takeIndices xd `index1` seg_off) + (USegd.takeLengths xd `index1` seg_off)+            el_off' = el_off - USegd.takeLengths xd `index1` seg_off+            j       = (USegd.takeIndices yd `index1` seg_off) + el_off'+            k       = (USegd.takeLengths yd `index1` seg_off) - el_off'         in  Just (True, seg_off, i, j, k, n)      {-# INLINE next #-}     next Nothing = return Done -    next (Just (False, seg, i, j, k, n))-      | n == 0    = return Done-      | k == 0    = return $ Skip (Just (True, seg, i, j, ylens ! seg, n))-      | otherwise = return $ Yield (xs!i) (Just (False, seg, i+1, j, k-1, n-1))+    next (Just (False, seg, i, j, k, n'))+      | n' == 0    = return Done+      | k  == 0    = return $ Skip (Just (True, seg, i, j, ylens `index1` seg, n'))+      | otherwise  = return $ Yield (xs `index2` i) (Just (False, seg, i+1, j, k-1, n'-1)) -    next (Just (True, seg, i, j, k, n))-      | n == 0    = return Done-      | k == 0+    next (Just (True, seg, i, j, k, n'))+      | n' == 0    = return Done+      | k  == 0       = let !seg' = seg+1-        in  return $ Skip (Just (False, seg', i, j, xlens ! seg', n))--      | otherwise = return $ Yield (ys!j) (Just (True, seg, i, j+1, k-1, n-1))----- fold ------------------------------------------------------------------------fixupFold :: Unbox a => (a -> a -> a) -> MVector s a-          -> Dist (Int,Vector a) -> ST s ()-{-# NOINLINE fixupFold #-}-fixupFold f !mrs !dcarry = go 1-  where-    !p = gangSize theGang+        in  return $ Skip (Just (False, seg', i, j, xlens `index1` seg', n')) -    go i | i >= p = return ()-         | Seq.null c = go (i+1)-         | otherwise   = do-                           x <- Seq.read mrs k-                           Seq.write mrs k (f x (c ! 0))-                           go (i+1)-      where-        (k,c) = indexD dcarry i+      | otherwise = return $ Yield (ys `index2` j) (Just (True, seg, i, j+1, k-1, n'-1))+{-# INLINE_STREAM appendSegS #-}  -folds :: Unbox a => (a -> a -> a)-              -> (USegd -> Vector a -> Vector a) -> UPSegd -> Vector a -> Vector a-{-# INLINE folds #-}-folds f g segd xs = dcarry `seq` drs `seq` runST (-  do-    mrs <- joinDM theGang drs-    fixupFold f mrs dcarry-    Seq.unsafeFreeze mrs)+-- foldR ----------------------------------------------------------------------+-- | Regular segmented fold.+foldRUP :: (Unbox a, Unbox b) => (b -> a -> b) -> b -> Int -> Vector a -> Vector b+foldRUP f z !segSize xs +  = joinD theGang unbalanced+          (mapD theGang +              (Seq.foldlRU f z segSize)+              (splitAsD theGang (mapD theGang (*segSize) dlen) xs))   where-    (dcarry,drs)-          = unzipD-          $ mapD theGang partial-          $ zipD (distUPSegd segd)-                 (splitD theGang balanced xs)--    partial (((segd,k),off), as)-      = let rs = g segd as-            {-# INLINE [0] n #-}-            n | off == 0  = 0-              | otherwise = 1-        in-        ((k, Seq.take n rs), Seq.drop n rs)---foldSUP :: Unbox a => (a -> a -> a) -> a -> UPSegd -> Vector a -> Vector a-{-# INLINE foldSUP #-}-foldSUP f !z = folds f (foldlSU f z)---fold1SUP :: Unbox a => (a -> a -> a) -> UPSegd -> Vector a -> Vector a-{-# INLINE fold1SUP #-}-fold1SUP f = folds f (fold1SU f)---sumSUP :: (Num e, Unbox e) => UPSegd -> Vector e -> Vector e-{-# INLINE sumSUP #-}-sumSUP = foldSUP (+) 0+    noOfSegs    = Seq.length xs `div` segSize+    dlen        = splitLenD theGang noOfSegs+{-# INLINE_UP foldRUP #-}  +-- sumR -----------------------------------------------------------------------+-- | Regular segmented sum. sumRUP :: (Num e, Unbox e) => Int -> Vector e -> Vector e-{-# INLINE sumRUP #-} sumRUP = foldRUP (+) 0---foldRUP :: (Unbox a, Unbox b) => (b -> a -> b) -> b -> Int -> Vector a -> Vector b-{-# INLINE foldRUP #-}-foldRUP f z !segSize xs = -   joinD theGang unbalanced-    (mapD theGang -      (foldlRU f z segSize)-      (splitAsD theGang (mapD theGang (*segSize) dlen) xs))-  where-    noOfSegs = Seq.length xs `div` segSize-    dlen = splitLenD theGang noOfSegs-+{-# INLINE_UP sumRUP #-} --- indices ---------------------------------------------------------------------indicesSUP :: UPSegd -> Vector Int-{-# INLINE_UP indicesSUP #-}-indicesSUP = joinD theGang balanced-           . mapD theGang indices-           . distUPSegd-  where-    indices ((segd,k),off) = indicesSU' off segd 

@@ -1,15 +1,18 @@ {-# LANGUAGE CPP #-}- #include "fusion-phases.h"  -- | Subarrays of flat unlifted arrays.-module Data.Array.Parallel.Unlifted.Parallel.Subarrays (-  dropUP-) where+module Data.Array.Parallel.Unlifted.Parallel.Subarrays +        (dropUP)+where import Data.Array.Parallel.Unlifted.Sequential.Vector as Seq-import Data.Array.Parallel.Unlifted.Distributed  +-- | Drop a the element at the provided index from a vector. dropUP :: Unbox e => Int -> Vector e -> Vector e-dropUP n xs = Seq.slice xs (min (max 0 n) (Seq.length xs)) (min (Seq.length xs) (Seq.length xs - n)) -{-# INLINE_U dropUP #-}+dropUP n xs +        = Seq.slice "dropUP"+                xs  +                (min (max 0 n)       (Seq.length xs))+                (min (Seq.length xs) (Seq.length xs - n)) +{-# INLINE_UP dropUP #-}

@@ -1,68 +1,80 @@--- | Sum-like parallel combinators for unlifted arrays-module Data.Array.Parallel.Unlifted.Parallel.Sums (-  andUP, orUP, sumUP-) where+{-# LANGUAGE CPP #-}+#include "fusion-phases.h" +-- | Sum-like parallel combinators for unlifted arrays+module Data.Array.Parallel.Unlifted.Parallel.Sums +        ( andUP, orUP+        , allUP, anyUP+        , sumUP, productUP+        , maximumUP, maximumByUP+        , maximumIndexByUP)+where import Data.Array.Parallel.Unlifted.Sequential.Vector as Seq import Data.Array.Parallel.Unlifted.Distributed-import Data.Array.Parallel.Unlifted.Parallel.Combinators (-  foldUP, foldl1UP, fold1UP, mapUP)-import Data.Array.Parallel.Unlifted.Parallel.Basics ( -  indexedUP)+import Data.Array.Parallel.Unlifted.Parallel.Combinators+import Data.Array.Parallel.Unlifted.Parallel.Basics (indexedUP) + -- | Compute the logical AND of all the elements in a array. andUP :: Vector Bool -> Bool-{-# INLINE andUP #-} andUP = foldUP (&&) True+{-# INLINE_UP andUP #-}   -- | Compute the logical OR of all the elements in a array. orUP :: Vector Bool -> Bool-{-# INLINE orUP #-} orUP = foldUP (||) False+{-# INLINE_UP orUP #-} + -- | Check whether all the elements in a array meet the given predicate. allUP :: Unbox e => (e -> Bool) -> Vector e -> Bool-{-# INLINE allUP #-} allUP p = andUP . mapUP p+{-# INLINE_UP allUP #-} + -- | Check whether any of the elements in a array meet the given predicate. anyUP :: Unbox e => (e -> Bool) -> Vector e -> Bool-{-# INLINE anyUP #-} anyUP p =  orUP . mapUP p+{-# INLINE_UP anyUP #-}   -- | Compute the sum all the elements of a array. sumUP :: (Unbox a, DT a, Num a) => Vector a -> a-{-# INLINE sumUP #-} sumUP = foldUP (+) 0+{-# INLINE_UP sumUP #-}   -- | Compute the product of all the elements of an array. productUP :: (DT e, Num e, Unbox e) => Vector e -> e-{-# INLINE productUP #-} productUP = foldUP (*) 1+{-# INLINE_UP productUP #-}   -- | Determine the maximum element in an array. maximumUP :: (DT e, Ord e, Unbox e) => Vector e -> e-{-# INLINE maximumUP #-} maximumUP = fold1UP max+{-# INLINE_UP maximumUP #-}  ---  |Determine the maximum element in an array under the given ordering+-- | Determine the maximum element in an array under the given ordering maximumByUP :: (DT e, Unbox e) => (e -> e -> Ordering) -> Vector e -> e-{-# INLINE maximumByUP #-}-maximumByUP = fold1UP . maxBy+maximumByUP +  = fold1UP . maxBy   where-    maxBy compare x y = case x `compare` y of-                          LT -> y-                          _  -> x+    maxBy compare' x y +        = case x `compare'` y of+           LT -> y+           _  -> x+{-# INLINE_UP maximumByUP #-} --- | Determine the index of the maximum element in an array under the given---   ordering-maximumIndexByUP :: (DT e, Unbox e) => (e -> e -> Ordering) -> Vector e -> Int-{-# INLINE maximumIndexByUP #-}-maximumIndexByUP cmp = fst . maximumByUP cmp' . indexedUP++-- | Determine the index of the maximum element in an array under the+--   given ordering+maximumIndexByUP +        :: (DT e, Unbox e) => (e -> e -> Ordering) -> Vector e -> Int+maximumIndexByUP cmp +  = fst . maximumByUP cmp' . indexedUP   where     cmp' (_,x) (_,y) = cmp x y+{-# INLINE_UP maximumIndexByUP #-}

@@ -1,11 +1,9 @@+{-# OPTIONS -fno-warn-orphans #-} -- | Read\/Show instances for segmented unlifted arrays. module Data.Array.Parallel.Unlifted.Parallel.Text () where--import Data.Array.Parallel.Base (-  Read(..), showsApp)-import Data.Array.Parallel.Unlifted.Parallel.UPSegd (-  UPSegd, lengthsUPSegd )+import Data.Array.Parallel.Base.Text (showsApp)+import Data.Array.Parallel.Unlifted.Parallel.UPSegd (UPSegd, takeLengths)  instance Show UPSegd where-  showsPrec k = showsApp k "toUPSegd" . lengthsUPSegd+  showsPrec k = showsApp k "toUPSegd" . takeLengths

@@ -0,0 +1,318 @@+{-# LANGUAGE CPP #-}+#include "fusion-phases.h"++-- | Parallel Scattered Segment descriptors.+--+--   See "Data.Array.Parallel.Unlifted" for how this works.+--+module Data.Array.Parallel.Unlifted.Parallel.UPSSegd +        ( -- * Types+          UPSSegd, valid++          -- * Constructors+        , mkUPSSegd, fromUSSegd, fromUPSegd+        , empty, singleton+  +          -- * Predicates+        , isContiguous++          -- * Projections+        , length+        , takeUSSegd+        , takeDistributed+        , takeLengths+        , takeIndices+        , takeElements+        , takeStarts+        , takeSources+        , getSeg+  +          -- * Append+        , appendWith  ++          -- * Segmented Folds+        , foldWithP+        , fold1WithP+        , sumWithP+        , foldSegsWithP)+where+import Data.Array.Parallel.Pretty                                 hiding (empty)+import Data.Array.Parallel.Unlifted.Distributed+import Data.Array.Parallel.Unlifted.Parallel.UPSegd               (UPSegd)+import Data.Array.Parallel.Unlifted.Sequential.USSegd             (USSegd)+import Data.Array.Parallel.Unlifted.Sequential.Vector             (Vector,  MVector, Unbox)+import Data.Array.Parallel.Unlifted.Vectors                       (Vectors, Unboxes)+import qualified Data.Array.Parallel.Unlifted.Parallel.UPSegd     as UPSegd+import qualified Data.Array.Parallel.Unlifted.Distributed.USSegd  as DUSSegd+import qualified Data.Array.Parallel.Unlifted.Sequential.USSegd   as USSegd+import qualified Data.Array.Parallel.Unlifted.Sequential.Vector   as US+import qualified Data.Array.Parallel.Unlifted.Sequential          as Seq+import Control.Monad.ST+import Prelude hiding (length)++here :: String -> String+here s = "Data.Array.Parallel.Unlifted.Parallel.UPSSegd." ++ s+++-- | Parallel Scattered Segment sescriptor+data UPSSegd +        = UPSSegd +        { upssegd_ussegd        :: !USSegd+          -- ^ Segment descriptor that describes the whole array.++        , upssegd_dssegd        :: Dist ((USSegd,Int),Int)+          -- ^ Segment descriptor for each chunk, +          --   along with segment id of first slice in the chunk,+          --   and the offset of that slice in its segment.+          --   See docs of `splitSegdOfElemsD` for an example.+        }+        deriving Show+++instance PprPhysical UPSSegd where+ pprp (UPSSegd ussegd dssegd)+  =  text "UPSSegd"+  $$ (nest 7 $ vcat+        [ text "ussegd:  " <+> pprp ussegd+        , text "dssegd:  " <+> pprp dssegd])+++-- | O(1).+--   Check the internal consistency of a scattered segment descriptor.+--- +--   * TODO: this doesn't do any checks yet+valid :: UPSSegd -> Bool+valid _ = True+{-# NOINLINE valid #-}+--  NOINLINE because it's only used during debugging anyway.+++-- Constructors ---------------------------------------------------------------+-- | Construct a new segment descriptor.+mkUPSSegd +        :: Vector Int   -- ^ Starting index of each segment in its flat array.+        -> Vector Int   -- ^ Source id of the flat array to tach each segment from.+        -> UPSegd       -- ^ Contiguous (unscattered) segment descriptor.+        -> UPSSegd++mkUPSSegd starts sources upsegd+        = fromUSSegd (USSegd.mkUSSegd starts sources (UPSegd.takeUSegd upsegd))+{-# INLINE_UP mkUPSSegd #-}+++-- | Promote a global `USSegd` to a parallel `UPSSegd` by distributing+--   it across the gang.+fromUSSegd :: USSegd -> UPSSegd+fromUSSegd ssegd +        = UPSSegd ssegd (DUSSegd.splitSSegdOnElemsD theGang ssegd)+{-# INLINE_UP fromUSSegd #-}+++-- | Promote a plain `UPSegd` to a `UPSSegd`, by assuming that all segments+--   come from a single flat array with source id 0.+---+--   * TODO:+--     This sequentially constructs the indices and source fields, and we+--     throw out the existing distributed `USegd`. We could probably keep+--     some of the existing fields and save reconstructing them.+--+fromUPSegd :: UPSegd -> UPSSegd+fromUPSegd upsegd+        = fromUSSegd $ USSegd.fromUSegd $ UPSegd.takeUSegd upsegd+{-# INLINE_UP fromUPSegd #-}+++-- | O(1). Yield an empty segment descriptor, with no elements or segments.+empty :: UPSSegd+empty   = fromUSSegd USSegd.empty+{-# INLINE_UP empty #-}+++-- | O(1).+--   Yield a singleton segment descriptor.+--   The single segment covers the given number of elements.+singleton :: Int -> UPSSegd+singleton n = fromUSSegd $ USSegd.singleton n+{-# INLINE_UP singleton #-}+++-- Predicates -----------------------------------------------------------------+-- INLINE trivial predicates as they'll expand to a simple calls.++-- | O(1). True when the starts are identical to the usegd indices field and+--   the sources are all 0's. +--+--   In this case all the data elements are in one contiguous flat+--   array, and consumers can avoid looking at the real starts and+--   sources fields.+--+isContiguous :: UPSSegd -> Bool+isContiguous    = USSegd.isContiguous . upssegd_ussegd+{-# INLINE isContiguous #-}+++-- Projections ----------------------------------------------------------------+-- INLINE trivial projections as they'll expand to a single record selector.++-- | O(1). Yield the overall number of segments.+length :: UPSSegd -> Int+length          = USSegd.length . upssegd_ussegd+{-# INLINE length #-}++-- | O(1). Yield the global `USegd` of a `UPSegd`+takeUSSegd :: UPSSegd -> USSegd+takeUSSegd      = upssegd_ussegd+{-# INLINE takeUSSegd #-}+++-- | O(1). Yield the distributed `USegd` of a `UPSegd`+takeDistributed :: UPSSegd -> Dist ((USSegd, Int), Int)+takeDistributed = upssegd_dssegd+{-# INLINE takeDistributed #-}+++-- | O(1). Yield the lengths of the individual segments.+takeLengths :: UPSSegd -> Vector Int+takeLengths     = USSegd.takeLengths . upssegd_ussegd+{-# INLINE takeLengths #-}+++-- | O(1). Yield the segment indices.+takeIndices :: UPSSegd -> Vector Int+takeIndices     = USSegd.takeIndices . upssegd_ussegd+{-# INLINE takeIndices #-}+++-- | O(1). Yield the total number of data elements.+--+--  @takeElements upssegd = sum (takeLengths upssegd)@+--+takeElements :: UPSSegd -> Int+takeElements    = USSegd.takeElements . upssegd_ussegd+{-# INLINE takeElements #-}+++-- | O(1). Yield the starting indices.+takeStarts :: UPSSegd -> Vector Int+takeStarts      = USSegd.takeStarts . upssegd_ussegd+{-# INLINE takeStarts #-}+++-- | O(1). Yield the source ids.+takeSources :: UPSSegd -> Vector Int+takeSources     = USSegd.takeSources . upssegd_ussegd +{-# INLINE takeSources #-}+++-- | O(1). Get the length, segment index, starting index, and source id of a segment.+getSeg :: UPSSegd -> Int -> (Int, Int, Int, Int)+getSeg upssegd ix+        = USSegd.getSeg (upssegd_ussegd upssegd) ix+{-# INLINE_UP getSeg #-}+++-- Append ---------------------------------------------------------------------+-- | O(n)+--   Produce a segment descriptor that describes the result of appending two+--   segmented arrays.+--+--   Appending two nested arrays is an index space transformation. Because+--   a `UPSSegd` can contain segments from multiple flat data arrays, we can+--   represent the result of the append without copying elements from the+--   underlying flat data arrays.+---+--   * TODO: This calls out to the sequential version.+--+appendWith+        :: UPSSegd              -- ^ Segment descriptor of first nested array.+        -> Int                  -- ^ Number of flat data arrays used to represent first nested array.+        -> UPSSegd              -- ^ Segment descriptor of second nested array. +        -> Int                  -- ^ Number of flat data arrays used to represent second nested array.+        -> UPSSegd+appendWith upssegd1 pdatas1+           upssegd2 pdatas2+ = fromUSSegd + $ USSegd.appendWith+        (upssegd_ussegd upssegd1) pdatas1+        (upssegd_ussegd upssegd2) pdatas2+{-# NOINLINE appendWith #-}+--  NOINLINE because we're not using it yet.+++-- Fold -----------------------------------------------------------------------+-- | Fold segments specified by a `UPSSegd`.+foldWithP :: (Unbox a, Unboxes a)+         => (a -> a -> a) -> a -> UPSSegd -> Vectors a -> Vector a+foldWithP f !z  = foldSegsWithP f (Seq.foldlSSU f z)+{-# INLINE_UP foldWithP #-}+++-- | Fold segments specified by a `UPSSegd`, with a non-empty vector.+fold1WithP :: (Unbox a, Unboxes a)+           => (a -> a -> a) -> UPSSegd -> Vectors a -> Vector a+fold1WithP f    = foldSegsWithP f (Seq.fold1SSU f)+{-# INLINE_UP fold1WithP #-}+++-- | Sum up segments specified by a `UPSSegd`.+sumWithP :: (Num a, Unbox a, Unboxes a)+        => UPSSegd -> Vectors a -> Vector a+sumWithP = foldWithP (+) 0+{-# INLINE_UP sumWithP #-}+++-- | Fold the segments specified by a `UPSSegd`.+--+--   Low level function takes a per-element worker and a per-segment worker.+--   It folds all the segments with the per-segment worker, then uses the+--   per-element worker to fixup the partial results when a segment +--   is split across multiple threads.+--   +foldSegsWithP+        :: (Unbox a, Unboxes a)+        => (a -> a -> a)+        -> (USSegd -> Vectors a -> Vector a)+        -> UPSSegd -> Vectors a -> Vector a++foldSegsWithP fElem fSeg segd xss + = dcarry `seq` drs `seq` +   runST (do+        mrs <- joinDM theGang drs+        fixupFold fElem mrs dcarry+        US.unsafeFreeze mrs)++ where  (dcarry,drs)+          = unzipD+          $ mapD theGang partial (takeDistributed segd)++        partial ((ssegd, k), off)+         = let rs = fSeg ssegd xss+               {-# INLINE [0] n #-}+               n | off == 0  = 0+                 | otherwise = 1++           in  ((k, US.take n rs), US.drop n rs)+{-# INLINE_UP foldSegsWithP #-}+++fixupFold+        :: Unbox a+        => (a -> a -> a)+        -> MVector s a+        -> Dist (Int,Vector a)+        -> ST s ()++fixupFold f !mrs !dcarry = go 1+  where+    !p = gangSize theGang++    go i | i >= p    = return ()+         | US.null c = go (i+1)+         | otherwise   +         = do   x <- US.read mrs k+                US.write mrs k (f x (US.index (here "fixupFold") c 0))+                go (i + 1)+      where+        (k,c) = indexD (here "fixupFold") dcarry i+{-# NOINLINE fixupFold #-}+

@@ -2,67 +2,287 @@ #include "fusion-phases.h"  -- | Parallel segment descriptors.-module Data.Array.Parallel.Unlifted.Parallel.UPSegd (+--+--   See "Data.Array.Parallel.Unlifted" for how this works.+--+module Data.Array.Parallel.Unlifted.Parallel.UPSegd +        ( -- * Types+          UPSegd(..)+        , valid -  -- * Types-  UPSegd,+          -- * Constructors+        , mkUPSegd, fromUSegd+        , empty, singleton, fromLengths -  -- * Operations on segment descriptors-  lengthUPSegd, lengthsUPSegd, indicesUPSegd, elementsUPSegd,-  segdUPSegd, distUPSegd,-  lengthsToUPSegd, mkUPSegd-) where+          -- * Projections+        , length+        , takeUSegd+        , takeDistributed+        , takeLengths+        , takeIndices+        , takeElements -import Data.Array.Parallel.Unlifted.Sequential.Vector as Seq-import Data.Array.Parallel.Unlifted.Sequential.Segmented.USegd+          -- * Indices+        , indicesP+  +          -- * Replicate+        , replicateWithP+    +          -- * Segmented Folds+        , foldWithP+        , fold1WithP+        , sumWithP+        , foldSegsWithP)+where import Data.Array.Parallel.Unlifted.Distributed+import Data.Array.Parallel.Unlifted.Sequential.USegd             (USegd)+import qualified Data.Array.Parallel.Unlifted.Distributed.USegd  as USegd+import qualified Data.Array.Parallel.Unlifted.Sequential         as Seq+import qualified Data.Array.Parallel.Unlifted.Sequential.Vector  as US+import qualified Data.Array.Parallel.Unlifted.Sequential.USegd   as USegd+import Data.Array.Parallel.Pretty                                hiding (empty)+import Data.Array.Parallel.Unlifted.Sequential.Vector  (Vector, MVector, Unbox)+import Control.Monad.ST+import Prelude  hiding (length) -data UPSegd = UPSegd { upsegd_usegd :: !USegd-                     , upsegd_dsegd :: Dist ((USegd,Int),Int)-                     }+here :: String -> String+here s = "Data.Array.Parallel.Unlifted.Parallel.UPSegd." ++ s  -lengthUPSegd :: UPSegd -> Int-{-# INLINE lengthUPSegd #-}-lengthUPSegd = lengthUSegd . upsegd_usegd+-- | A parallel segment descriptor holds a global (undistributed) segment+--   desciptor, as well as a distributed version. The distributed version+--   describes how to split work on the segmented array over the gang. +data UPSegd +        = UPSegd +        { upsegd_usegd :: !USegd+          -- ^ Segment descriptor that describes the whole array. +        , upsegd_dsegd :: Dist ((USegd,Int),Int)+          -- ^ Segment descriptor for each chunk, +          --   along with segment id of first slice in the chunk,+          --   and the offset of that slice in its segment.+          --   See docs of `splitSegdOfElemsD` for an example.+        } -lengthsUPSegd :: UPSegd -> Vector Int-{-# INLINE lengthsUPSegd #-}-lengthsUPSegd = lengthsUSegd . upsegd_usegd +-- Pretty ---------------------------------------------------------------------+instance PprPhysical UPSegd where+ pprp (UPSegd usegd dsegd)+  =  text "UPSegd"+  $$ (nest 7 $ vcat+        [ text "usegd:  "  <+> pprp usegd+        , text "dsegd:  "  <+> pprp dsegd]) -indicesUPSegd :: UPSegd -> Vector Int-{-# INLINE indicesUPSegd #-}-indicesUPSegd = indicesUSegd . upsegd_usegd +-- Valid ----------------------------------------------------------------------+-- | O(1).+--   Check the internal consistency of a parallel segment descriptor.+--- +--   * TODO: this doesn't do any checks yet+valid :: UPSegd -> Bool+valid _ = True+{-# NOINLINE valid #-}+--  NOINLINE because it's only used during debugging anyway. -elementsUPSegd :: UPSegd -> Int-{-# INLINE elementsUPSegd #-}-elementsUPSegd = elementsUSegd . upsegd_usegd +-- Constructors ---------------------------------------------------------------+-- | O(1). Construct a new parallel segment descriptor.+mkUPSegd +        :: Vector Int   -- ^ Length of each segment.+        -> Vector Int   -- ^ Starting index of each segment.+        -> Int          -- ^ Total number of elements in the flat array.+        -> UPSegd -segdUPSegd :: UPSegd -> USegd-{-# INLINE segdUPSegd #-}-segdUPSegd = upsegd_usegd+mkUPSegd lens idxs n+        = fromUSegd (USegd.mkUSegd lens idxs n)+{-# INLINE_UP mkUPSegd #-}  -distUPSegd :: UPSegd -> Dist ((USegd,Int),Int)-{-# INLINE distUPSegd #-}-distUPSegd = upsegd_dsegd+-- | Convert a global `USegd` to a parallel `UPSegd` by distributing +--   it across the gang.+fromUSegd :: USegd -> UPSegd+fromUSegd segd   = UPSegd segd (USegd.splitSegdOnElemsD theGang segd)+{-# INLINE_UP fromUSegd #-}  -lengthsToUPSegd :: Vector Int -> UPSegd-{-# INLINE lengthsToUPSegd #-}-lengthsToUPSegd = toUPSegd . lengthsToUSegd+-- | O(1). Construct an empty segment descriptor, with no elements or segments.+empty :: UPSegd+empty           = fromUSegd USegd.empty+{-# INLINE_UP empty #-}  -mkUPSegd :: Vector Int -> Vector Int -> Int -> UPSegd-{-# INLINE mkUPSegd #-}-mkUPSegd lens idxs n = toUPSegd (mkUSegd lens idxs n)+-- | O(1). Construct a singleton segment descriptor.+--   The single segment covers the given number of elements.+singleton :: Int -> UPSegd+singleton n     = fromUSegd $ USegd.singleton n+{-# INLINE_UP singleton #-}  -toUPSegd :: USegd -> UPSegd-{-# INLINE toUPSegd #-}-toUPSegd segd = UPSegd segd (splitSegdD' theGang segd)+-- | O(n). Convert an array of segment lengths into a parallel segment descriptor.+-- +--   The array contains the length of each segment, and we compute the +--   indices from that. Runtime is O(n) in the number of segments.+--+fromLengths :: Vector Int -> UPSegd+fromLengths     = fromUSegd . USegd.fromLengths+{-# INLINE_UP fromLengths #-} ++-- Projections ----------------------------------------------------------------+-- INLINE trivial projections as they'll expand to a single record selector.++-- | O(1). Yield the overall number of segments.+length :: UPSegd -> Int+length          = USegd.length . upsegd_usegd+{-# INLINE length #-}+++-- | O(1). Yield the global `USegd` of a `UPSegd`.+takeUSegd :: UPSegd -> USegd+takeUSegd       = upsegd_usegd+{-# INLINE takeUSegd #-}+++-- | O(1). Yield the distributed `USegd` of a `UPSegd`.+--   +--  We get a plain `USegd` for each chunk, the segment id of the first+--  slice in the chunk, and the starting offset of that slice in its segment.+-- +takeDistributed :: UPSegd -> Dist ((USegd,Int),Int)+takeDistributed = upsegd_dsegd+{-# INLINE takeDistributed #-}+++-- | O(1). Yield the lengths of the individual segments.+takeLengths :: UPSegd -> Vector Int+takeLengths     = USegd.takeLengths . upsegd_usegd+{-# INLINE takeLengths #-}+++-- | O(1). Yield the segment indices.+takeIndices :: UPSegd -> Vector Int+takeIndices     = USegd.takeIndices . upsegd_usegd+{-# INLINE takeIndices #-}+++-- | O(1). Yield the total number of array elements.+-- +--  @takeElements upsegd = sum (takeLengths upsegd)@+--+takeElements :: UPSegd -> Int+takeElements    = USegd.takeElements . upsegd_usegd+{-# INLINE takeElements #-}+++-- Indices --------------------------------------------------------------------+-- | O(n). Yield a vector containing indicies that give the position of each +--         member of the flat array in its corresponding segment.+--+--  @indicesP (fromLengths [5, 2, 3]) = [0,1,2,3,4,0,1,0,1,2]@+--+indicesP :: UPSegd -> Vector Int+indicesP+        = joinD theGang balanced+        . mapD  theGang indices+        . takeDistributed+  where+    indices ((segd,_k),off) = Seq.indicesSU' off segd+{-# NOINLINE indicesP #-}+--  NOINLINE because we're not using it yet.+++-- Replicate ------------------------------------------------------------------+-- | Copying segmented replication. Each element of the vector is physically +--   copied according to the length of each segment in the segment descriptor.+--+--   @replicateWith (fromLengths [3, 1, 2]) [5, 6, 7] = [5, 5, 5, 6, 7, 7]@+--+replicateWithP :: Unbox a => UPSegd -> Vector a -> Vector a+replicateWithP segd !xs +  = joinD theGang balanced+  . mapD  theGang rep+  $ takeDistributed segd+  where+    rep ((dsegd,di),_)+      = Seq.replicateSU dsegd +      $ US.slice (here "replicateWithP")+                xs di (USegd.length dsegd)+{-# INLINE_UP replicateWithP #-}+++-- Fold -----------------------------------------------------------------------+-- | Fold segments specified by a `UPSegd`.+foldWithP :: Unbox a+         => (a -> a -> a) -> a -> UPSegd -> Vector a -> Vector a+foldWithP f !z  = foldSegsWithP f (Seq.foldlSU f z)+{-# INLINE_UP foldWithP #-}+++-- | Fold segments specified by a `UPSegd`, with a non-empty vector.+fold1WithP :: Unbox a+         => (a -> a -> a) -> UPSegd -> Vector a -> Vector a+fold1WithP f    = foldSegsWithP f (Seq.fold1SU f)+{-# INLINE_UP fold1WithP #-}+++-- | Sum up segments specified by a `UPSegd`.+sumWithP :: (Num e, Unbox e) => UPSegd -> Vector e -> Vector e+sumWithP = foldWithP (+) 0+{-# INLINE_UP sumWithP #-}+++-- | Fold the segments specified by a `UPSegd`.+--+--   This low level function takes a per-element worker and a per-segment worker.+--   It folds all the segments with the per-segment worker, then uses the+--   per-element worker to fixup the partial results when a segment +--   is split across multiple threads.+--   +foldSegsWithP+        :: Unbox a+        => (a -> a -> a)+        -> (USegd -> Vector a -> Vector a)+        -> UPSegd -> Vector a -> Vector a++{-# INLINE_UP foldSegsWithP #-}+foldSegsWithP fElem fSeg segd xs + = dcarry `seq` drs `seq` +   runST (do+        mrs <- joinDM theGang drs+        fixupFold fElem mrs dcarry+        US.unsafeFreeze mrs)++ where  (dcarry,drs)+          = unzipD+          $ mapD theGang partial+          $ zipD (takeDistributed segd)+                 (splitD theGang balanced xs)++        partial (((segd', k), off), as)+         = let rs = fSeg segd' as+               {-# INLINE [0] n #-}+               n | off == 0  = 0+                 | otherwise = 1++           in  ((k, US.take n rs), US.drop n rs)+++fixupFold+        :: Unbox a+        => (a -> a -> a)+        -> MVector s a+        -> Dist (Int,Vector a)+        -> ST s ()+{-# NOINLINE fixupFold #-}+fixupFold f !mrs !dcarry = go 1+  where+    !p = gangSize theGang++    go i | i >= p = return ()+         | US.null c = go (i+1)+         | otherwise   = do+                           x <- US.read mrs k+                           US.write mrs k (f x (US.index (here "fixupFold") c 0))+                           go (i + 1)+      where+        (k,c) = indexD (here "fixupFold") dcarry i

@@ -2,16 +2,25 @@ #include "fusion-phases.h"  -- | Parallel selectors.-module Data.Array.Parallel.Unlifted.Parallel.UPSel (-  -- * Types-  UPSel2, UPSelRep2,+module Data.Array.Parallel.Unlifted.Parallel.UPSel +        ( -- * Types+          UPSel2+        , UPSelRep2 -  -- * Operations on segment descriptors-  tagsUPSel2, indicesUPSel2, elementsUPSel2_0, elementsUPSel2_1,-  selUPSel2, repUPSel2, mkUPSel2,-  mkUPSelRep2, indicesUPSelRep2, elementsUPSelRep2_0, elementsUPSelRep2_1,-) where-import Data.Array.Parallel.Unlifted.Sequential.Vector as Seq+          -- * Operations+        , tagsUPSel2+        , indicesUPSel2+        , elementsUPSel2_0+        , elementsUPSel2_1+        , selUPSel2+        , repUPSel2+        , mkUPSel2+        , mkUPSelRep2+        , indicesUPSelRep2+        , elementsUPSelRep2_0+        , elementsUPSelRep2_1)+where+import Data.Array.Parallel.Unlifted.Sequential.Vector   as US import Data.Array.Parallel.Unlifted.Sequential.USel import Data.Array.Parallel.Unlifted.Distributed import Data.Array.Parallel.Base (Tag, tagToInt)@@ -29,102 +38,54 @@         , upsel2_rep  :: UPSelRep2 }  --- | A `UPSelRep2` describes how to distribute the two data vectors---   corresponding to a `UPSel2` across several PEs. -----   Suppose we want to perform the following combine operation:------ @---    combine [0,0,1,1,0,1,0,0,1] [A0,A1,A2,A3,A4] [B0,B1,B2,B3] ---     = [A0,A1,B0,B1,A2,B2,A3,A4,B3]--- @------   The first array is the tags array, that says which of the data arrays to---   get each successive element from. As `combine` is difficult to compute---   in parallel, if we are going to perform several combines with the same---   tag array, we can precompute a selector that tells us where to get each---   element. The selector contains the original tags, as well as the source---   index telling us where to get each element for the result array.--- --- @---    [0,0,1,1,0,1,0,0,1]      -- tags    (which data vector to take the elem from)---    [0,1,0,1,2,2,3,4,3]      -- indices (where in the vector to take the elem from)--- @------  Suppose we want to distribute the combine operation across 3 PEs. It's---  easy to split the selector like so:------ @       ---     PE0                PE1               PE2---    [0,0,1]            [1,0,1]           [0,0,1]   -- tags---    [0,1,0]            [1,2,2]           [3,4,3]   -- indices--- @------  We now need to split the two data arrays. Each PE needs slices of the data---  arrays that correspond to the parts of the selector that were given to it.---  For the current example we get:------ @---    PE0                PE1               PE2---    [A0,A1]            [A2]              [A3,A4]---    [B0]               [B1,B2]           [B3]--- @------  The `UPSelRep2` contains the starting index and length of each of of these---  slices:------ @---         PE0                PE1               PE2---    ((0, 0), (2, 1))   ((2, 1), (1, 2))  ((3, 3), (2, 1))---    indices   lens      indices  lens    indices  lens--- @--- type UPSelRep2         = Dist ((Int,Int), (Int,Int)) +-- Projections ----------------------------------------------------------------+-- INLINE trivial projections as they'll expand to a single record selector.  -- | O(1). Get the tags of a selector. tagsUPSel2 :: UPSel2 -> Vector Tag+tagsUPSel2      = tagsUSel2 .  upsel2_usel {-# INLINE tagsUPSel2 #-}-tagsUPSel2 = tagsUSel2 .  upsel2_usel   -- | O(1). Get the indices of a selector. indicesUPSel2 :: UPSel2 -> Vector Int+indicesUPSel2   = indicesUSel2 . upsel2_usel {-# INLINE indicesUPSel2 #-}-indicesUPSel2 = indicesUSel2 . upsel2_usel  --- | O(1). TODO: What is this for?+-- | O(1). Get the number of elements that will be taken from the first array. elementsUPSel2_0 :: UPSel2 -> Int-{-# INLINE elementsUPSel2_0 #-} elementsUPSel2_0 = elementsUSel2_0 . upsel2_usel+{-# INLINE elementsUPSel2_0 #-}  --- | O(1). TODO: What is this for?+-- | O(1). Get the number of elements that will be taken from the second array. elementsUPSel2_1 :: UPSel2 -> Int-{-# INLINE elementsUPSel2_1 #-} elementsUPSel2_1 = elementsUSel2_1 . upsel2_usel+{-# INLINE elementsUPSel2_1 #-}  --- | O(1). TODO: What is this for?+-- | O(1). Take the sequential `USel2` from a `UPSel2`. selUPSel2 :: UPSel2 -> USel2+selUPSel2       = upsel2_usel {-# INLINE selUPSel2 #-}-selUPSel2 = upsel2_usel  --- | O(1). TODO: What is this for?+-- | O(1). Take the `UPSelRep2` from a `UPSel2`. repUPSel2 :: UPSel2 -> UPSelRep2+repUPSel2       = upsel2_rep {-# INLINE repUPSel2 #-}-repUPSel2 = upsel2_rep   -- Representation selectors ------------------------------------------------------- | Computes a `UPSelRep2` from an array of tags. This is used when parallelising---   a `combine` operation. See the docs for `UPSelRep2` for details.+-- | Computes a `UPSelRep2` from an array of tags.+--   This is used when parallelising a `combine` operation.+--   See the docs for `UPSelRep2` for details. mkUPSelRep2 :: Vector Tag -> UPSelRep2-{-# INLINE mkUPSelRep2 #-} mkUPSelRep2 tags = zipD idxs lens   where     lens = mapD   theGang count@@ -133,38 +94,42 @@     idxs = fst          $ scanD theGang add (0,0) lens -    count bs = let ones = Seq.sum (Seq.map tagToInt bs)-               in (Seq.length bs - ones,ones)+    count bs = let ones = US.sum (US.map tagToInt bs)+               in (US.length bs - ones,ones)      add (x1,y1) (x2,y2) = (x1+x2, y1+y2)+{-# INLINE_UP mkUPSelRep2 #-}   indicesUPSelRep2 :: Vector Tag -> UPSelRep2 -> Vector Int-{-# INLINE indicesUPSelRep2 #-}-indicesUPSelRep2 tags rep = joinD theGang balanced-                          $ zipWithD theGang indices-                                             (splitD theGang balanced tags)-                                              rep+indicesUPSelRep2 tags rep +        = joinD theGang balanced+        $ zipWithD theGang indices+             (splitD theGang balanced tags)+              rep   where-    indices tags ((i,j), (m,n))-      = Seq.combine2ByTag tags (Seq.enumFromStepLen i 1 m)-                               (Seq.enumFromStepLen j 1 n)+    indices tags' ((i,j), (m,n))+      = US.combine2ByTag tags' (US.enumFromStepLen i 1 m)+                               (US.enumFromStepLen j 1 n)+{-# INLINE_UP indicesUPSelRep2 #-}  --- | O(n).+-- | O(n). Count the number of elements to take from the first array. elementsUPSelRep2_0 :: Vector Tag -> UPSelRep2 -> Int-{-# INLINE elementsUPSelRep2_0 #-}-elementsUPSelRep2_0 _ = sumD theGang . fstD . sndD+elementsUPSelRep2_0 _+        = sumD theGang . fstD . sndD+{-# INLINE_UP elementsUPSelRep2_0 #-}  --- | O(n).+-- | O(n). Count the number of elements to take from the second array. elementsUPSelRep2_1 :: Vector Tag -> UPSelRep2 -> Int-{-# INLINE elementsUPSelRep2_1 #-}-elementsUPSelRep2_1 _ = sumD theGang . sndD . sndD+elementsUPSelRep2_1 _+        = sumD theGang . sndD . sndD+{-# INLINE_UP elementsUPSelRep2_1 #-}   -- | O(1). Construct a selector. Wrapper for `UPSel2`. mkUPSel2 :: Vector Tag -> Vector Int -> Int -> Int -> UPSelRep2 -> UPSel2-{-# INLINE mkUPSel2 #-}-mkUPSel2 tags is n0 n1 rep = UPSel2 (mkUSel2 tags is n0 n1) rep-+mkUPSel2 tags is n0 n1 rep+        = UPSel2 (mkUSel2 tags is n0 n1) rep+{-# INLINE_UP mkUPSel2 #-}

@@ -0,0 +1,478 @@+{-# LANGUAGE CPP #-}+#include "fusion-phases.h"+{-# OPTIONS -Wall -fno-warn-orphans -fno-warn-missing-signatures #-}++-- | Parallel virtual segment descriptors.+--+--   See "Data.Array.Parallel.Unlifted" for how this works.+--+module Data.Array.Parallel.Unlifted.Parallel.UPVSegd +        ( -- * Types+          UPVSegd++          -- * Consistency check+        , valid++          -- * Constructors+        , mkUPVSegd+        , fromUPSegd+        , fromUPSSegd+        , empty+        , singleton+        , replicated+        +        -- * Predicates+        , isManifest+        , isContiguous++        -- * Projections+        , length+        , takeVSegids, takeVSegidsRedundant+        , takeUPSSegd, takeUPSSegdRedundant+        , takeLengths+        , getSeg++        -- * Demotion+        , unsafeDemoteToUPSSegd+        , unsafeDemoteToUPSegd++        -- * Operators+        , updateVSegs+        , updateVSegsReachable++        , appendWith+        , combine2)+where+import Data.Array.Parallel.Unlifted.Parallel.Permute+import Data.Array.Parallel.Unlifted.Parallel.UPSel              (UPSel2)+import Data.Array.Parallel.Unlifted.Parallel.UPSSegd            (UPSSegd)+import Data.Array.Parallel.Unlifted.Parallel.UPSegd             (UPSegd)+import Data.Array.Parallel.Unlifted.Sequential.Vector           (Vector)+import Data.Array.Parallel.Pretty                               hiding (empty)+import Prelude                                                  hiding (length)+import qualified Data.Array.Parallel.Unlifted.Sequential.Vector as US+import qualified Data.Array.Parallel.Unlifted.Sequential.USSegd as USSegd+import qualified Data.Array.Parallel.Unlifted.Parallel.UPSel    as UPSel+import qualified Data.Array.Parallel.Unlifted.Parallel.UPSegd   as UPSegd+import qualified Data.Array.Parallel.Unlifted.Parallel.UPSSegd  as UPSSegd++here :: String -> String+here s = "Data.Array.Parallel.Unlifted.Parallel.UPVSegd." Prelude.++ s++-- UPVSegd ---------------------------------------------------------------------+-- | Parallel Virtual Segment descriptor.+--   +data UPVSegd +        = UPVSegd +        { upvsegd_manifest      :: !Bool+          -- ^ When the vsegids field holds a lazy (V.enumFromTo 0 (len - 1))+          --   then this field is True. This lets us perform some operations like+          --   demoteToUPSSegd without actually creating the vsegids field.+        +          -- | Virtual segment identifiers that indicate what physical segment+          --   to use for each virtual segment.+        , upvsegd_vsegids_redundant     :: Vector Int           -- LAZY FIELD +        , upvsegd_vsegids_culled        :: Vector Int           -- LAZY FIELD+        +          -- | Scattered segment descriptor that defines how physical segments+          --   are layed out in memory.+        , upvsegd_upssegd_redundant     :: UPSSegd              -- LAZY FIELD+        , upvsegd_upssegd_culled        :: UPSSegd              -- LAZY FIELD+        +        -- IMPORTANT:+        -- When vsegids are transformed due to a segmented replication operation, +        -- if some of the segment lengths were zero, then we will end up with +        -- physical segments that are unreachable from the vsegids.+        -- +        -- For some operations (like indexing) the fact that we have unreachable+        -- psegids doesn't matter, but for others (like segmented fold) it does.+        -- The problem is that we perform segmented fold by first folding all +        -- the physical segments, then replicating the results according to the +        -- vsegids. If no vsegids referenced a physical segment then we didn't +        -- need to fold it.+        -- +        -- When vsegids are updated the version that may have unreachable psegs+        -- is stored in the vsegids_redundant and upssegd_redundant. The _culled+        -- versions are then set to a SUSPENDED call to callOnVSegids. If no+        -- consumers every demand the culled version then we never need to compute+        -- it.+        -- +        -- The vsegids_redundant field must also be lazy (no bang) because when it+        -- has the value (V.enumFromTo 0 (len - 1)) we want to avoid building the+        -- enumeration unless it's strictly demanded.+        }+        deriving (Show)+++-- | Pretty print the physical representation of a `UVSegd`+instance PprPhysical UPVSegd where+ pprp (UPVSegd _ _ vsegids _ upssegd)+  = vcat+  [ text "UPVSegd" $$ (nest 7 $ text "vsegids: " <+> (text $ show $ US.toList vsegids))+  , pprp upssegd ]+++-- | O(1). Check the internal consistency of a virutal segmentation descriptor.+---+--   * TODO: this doesn't do any checks yet.+--+valid :: UPVSegd -> Bool+valid UPVSegd{} = True+{-# NOINLINE valid #-}+--  NOINLINE because it's only used during debugging anyway.+++-- Constructors ---------------------------------------------------------------+-- NOTE: these are NOINLINE for now just so it's easier to read the core.+--       we can INLINE them later.++-- | O(1). Construct a new virtual segment descriptor.+mkUPVSegd+        :: Vector Int   -- ^ Array saying which physical segment to use for+                        --   each virtual segment.+        -> UPSSegd      -- ^ Scattered segment descriptor defining the physical+                        --   segments.+        -> UPVSegd++mkUPVSegd vsegids ussegd+        = UPVSegd False vsegids vsegids ussegd ussegd+{-# INLINE_UP mkUPVSegd #-}+++-- | O(segs). Promote a `UPSSegd` to a `UPVSegd`.+--   The result contains one virtual segment for every physical segment+--   defined by the `UPSSegd`.+---+--   * TODO: make this parallel, use parallel version of enumFromTo.+--+fromUPSSegd :: UPSSegd -> UPVSegd+fromUPSSegd upssegd+ = let  vsegids = US.enumFromTo 0 (UPSSegd.length upssegd - 1)+   in   UPVSegd True vsegids vsegids upssegd upssegd+{-# INLINE_UP fromUPSSegd #-}+++-- | O(segs). Promote a `UPSegd` to a `UPVSegd`.+--   All segments are assumed to come from a flat array with sourceid 0.+--   The result contains one virtual segment for every physical segment+--   the provided `UPSegd`.+--+fromUPSegd :: UPSegd -> UPVSegd+fromUPSegd      = fromUPSSegd . UPSSegd.fromUPSegd+{-# INLINE_UP fromUPSegd #-}+++-- | O(1). Construct an empty segment descriptor, with no elements or segments.+empty :: UPVSegd+empty+ = let  vsegids = US.empty+        upssegd = UPSSegd.empty+   in   UPVSegd True vsegids vsegids upssegd upssegd+{-# INLINE_UP empty #-}+++-- | O(1). Construct a singleton segment descriptor.+--   The single segment covers the given number of elements in a flat array+--   with sourceid 0.+singleton :: Int -> UPVSegd+singleton n+ = let  vsegids = US.singleton 0+        upssegd = UPSSegd.singleton n+   in   UPVSegd True vsegids vsegids upssegd upssegd+{-# INLINE_UP singleton #-}+++-- | O(1). Construct a `UPVSegd` that describes an array created by replicating+--   a single segment several times.+---+--   NOTE: This is a helpful target for rewrite rules, because when we +--   see a 'replicated' we know that all segments in the virtual array+--   point to the same data.+replicated +        :: Int          -- ^ Length of segment.+        -> Int          -- ^ Number of times replicated.+        -> UPVSegd++replicated len reps+ = let  -- We have a single physical segment.+        ssegd   = UPSSegd.singleton len++        -- All virtual segments point to the same physical segment.+   in   mkUPVSegd (US.replicate reps 0) ssegd                           -- TODO: use parallel replicate+{-# INLINE_U replicated #-}+++-- Predicates -----------------------------------------------------------------+-- | O(1). Checks whether all the segments are manifest (unshared / non-virtual).+--   If this is the case, then the vsegids field will be [0..len-1]. +--+--   Consumers can check this field, avoid demanding the vsegids field.+--   This can avoid the need for it to be constructed in the first place, due to+--   lazy evaluation.+--+isManifest :: UPVSegd -> Bool+isManifest      = upvsegd_manifest+{-# INLINE isManifest #-}+++-- | O(1). True when the starts are identical to the usegd indices field and+--   the sources are all 0's. +--+--   In this case all the data elements are in one contiguous flat+--   array, and consumers can avoid looking at the real starts and+--   sources fields.+--+isContiguous    :: UPVSegd -> Bool+isContiguous    = UPSSegd.isContiguous . upvsegd_upssegd_culled+{-# INLINE isContiguous #-}+++-- Projections ----------------------------------------------------------------+-- INLINE trivial projections as they'll expand to a single record selector.++-- | O(1). Yield the overall number of segments.+length :: UPVSegd -> Int+length          = US.length . upvsegd_vsegids_redundant+{-# INLINE length #-}+++-- | O(1). Yield the virtual segment ids of `UPVSegd`.+takeVSegids :: UPVSegd -> Vector Int+takeVSegids     = upvsegd_vsegids_culled+{-# INLINE takeVSegids #-}+++-- | O(1). Take the vsegids of a `UPVSegd`, but don't require that every physical+--   segment is referenced by some virtual segment.+--+--   If you're just performing indexing and don't need the invariant that all+--   physical segments are reachable from some virtual segment, then use this+--   version as it's faster. This sidesteps the code that maintains the invariant.+--+--   The stated O(1) complexity assumes that the array has already been fully+--   evalauted. If this is not the case then we can avoid demanding the result+--   of a prior computation on the vsegids, thus reducing the cost attributed+--   to that prior computation.+takeVSegidsRedundant :: UPVSegd -> Vector Int+takeVSegidsRedundant = upvsegd_vsegids_redundant+{-# INLINE takeVSegidsRedundant #-}+++-- | O(1). Yield the `UPSSegd` of `UPVSegd`.+takeUPSSegd :: UPVSegd -> UPSSegd+takeUPSSegd     = upvsegd_upssegd_culled+{-# INLINE takeUPSSegd #-}+++-- | O(1). Take the `UPSSegd` of a `UPVSegd`, but don't require that every physical+--   segment is referenced by some virtual segment.+--+--   See the note in `takeVSegidsRedundant`.+takeUPSSegdRedundant :: UPVSegd -> UPSSegd+takeUPSSegdRedundant    = upvsegd_upssegd_redundant+{-# INLINE takeUPSSegdRedundant #-}+++-- | O(segs). Yield the lengths of the segments described by a `UPVSegd`.+---+--   * TODO: This is slow and sequential.+--+takeLengths :: UPVSegd -> Vector Int+takeLengths (UPVSegd manifest _ vsegids _ upssegd)+ | manifest     = UPSSegd.takeLengths upssegd+ | otherwise    + = let !lengths        = (UPSSegd.takeLengths upssegd)+   in  US.map (US.index (here "takeLengths") lengths) vsegids+{-# NOINLINE takeLengths #-}+--  NOINLINE because we don't want a case expression due to the test on the +--  manifest flag to appear in the core program.+++-- | O(1). Get the length, starting index, and source id of a segment.+---+--  NOTE: We don't return the segment index field from the `USSegd` as this refers+--        to the flat index relative to the `SSegd` array, rather than +--        relative to the UVSegd array. If we tried to promote the `USSegd` index+--        to a `UVSegd` index it could overflow.+--+getSeg :: UPVSegd -> Int -> (Int, Int, Int)+getSeg upvsegd ix+ = let  vsegids = upvsegd_vsegids_redundant upvsegd+        upssegd = upvsegd_upssegd_redundant upvsegd+        (len, _index, start, source)+                = UPSSegd.getSeg upssegd (US.index (here "getSeg") vsegids ix)+   in   (len, start, source)+{-# INLINE_UP getSeg #-}+++-- Demotion -------------------------------------------------------------------+-- | O(segs). Yield a `UPSSegd` that describes each segment of a `UPVSegd`+--   individually.+--+--   By doing this we lose information about which virtual segments+--   correspond to the same physical segments.+--+--   /WARNING/: Trying to take the `UPSegd` of a nested array that has been+--   constructed with replication can cause index space overflow. This is+--   because the virtual size of the corresponding flat data can be larger+--   than physical memory. If this happens then indices fields and +--   element count in the result will be invalid.+-- +unsafeDemoteToUPSSegd :: UPVSegd -> UPSSegd+unsafeDemoteToUPSSegd upvsegd+ | upvsegd_manifest upvsegd     = upvsegd_upssegd_culled upvsegd        -- TODO: take the redundant ones+ | otherwise+ = let  vsegids         = upvsegd_vsegids_culled upvsegd+        upssegd         = upvsegd_upssegd_culled upvsegd+        starts'         = bpermuteUP (UPSSegd.takeStarts  upssegd) vsegids+        sources'        = bpermuteUP (UPSSegd.takeSources upssegd) vsegids+        lengths'        = bpermuteUP (UPSSegd.takeLengths upssegd) vsegids+        upsegd'         = UPSegd.fromLengths lengths'+   in   UPSSegd.mkUPSSegd starts' sources' upsegd'+{-# NOINLINE unsafeDemoteToUPSSegd #-}+--  NOINLINE because it's complicated and won't fuse with anything.+--  In core we want to see when VSegds are being demoted.+++-- | O(segs). Yield a `UPSegd` that describes each segment of a `UPVSegd`+--   individually, assuming all segments have been concatenated to +--   remove scattering.+--+--   * See the warning in `unsafeDemoteToUPSSegd`.+---+--   * TODO: if the upvsegd is manifest and contiguous this can be O(1).+--+unsafeDemoteToUPSegd :: UPVSegd -> UPSegd+unsafeDemoteToUPSegd (UPVSegd _ _ vsegids _ upssegd)+        = {-# SCC "unsafeDemoteToUPSegd" #-}+          UPSegd.fromLengths+        $ bpermuteUP (UPSSegd.takeLengths upssegd) vsegids+{-# NOINLINE unsafeDemoteToUPSegd #-}+--  NOINLINE because it's complicated and won't fuse with anything.+--  In core we want to see when VSegds are being demoted.+++-- Operators ------------------------------------------------------------------+-- | Update the vsegids of a `UPVSegd`, and then cull the physical+--   segment descriptor so that all physical segments are reachable from+--   some virtual segment.+--+--   This function lets you perform filtering operations on the virtual segments,+--   while maintaining the invariant that all physical segments are referenced+--   by some virtual segment.+---+--   * TODO: make this parallel.+--     It runs the sequential 'cull' then reconstructs the UPSSegd.+-- +updateVSegs :: (Vector Int -> Vector Int) -> UPVSegd -> UPVSegd+updateVSegs fUpdate (UPVSegd _ vsegids _ upssegd _)+ = let  +        -- When we transform the vsegids, we don't know whether they all +        -- made it into the result. +        vsegids_redundant      = fUpdate vsegids+ +        -- Cull the psegs down to just those reachable from the vsegids, +        -- but do it lazilly so consumers can avoid demanding this +        -- culled version and save creating it.+        (  vsegids_culled+         , ussegd_culled)       = USSegd.cullOnVSegids vsegids_redundant+                                $ UPSSegd.takeUSSegd upssegd++        upssegd_culled          = UPSSegd.fromUSSegd ussegd_culled++   in   UPVSegd False+                vsegids_redundant vsegids_culled+                upssegd           upssegd_culled+{-# NOINLINE updateVSegs #-}+--  NOINLINE because we want to see this happening in core.+++-- | Update the vsegids  of `UPVSegd`, where the result is guaranteed to+--   cover all physical segments.+--+--   Using this version saves performing the 'cull' operation which +--   discards unreachable physical segments.+--+--   * The resulting vsegids must cover all physical segments.+--     If they do not then there will be physical segments that are not +--     reachable from some virtual segment, and subsequent operations+--     like segmented fold will have the wrong work complexity.+--+updateVSegsReachable :: (Vector Int -> Vector Int) -> UPVSegd -> UPVSegd+updateVSegsReachable fUpdate (UPVSegd _ _ vsegids _ upssegd)+ = let  vsegids' = fUpdate vsegids+   in   UPVSegd False vsegids' vsegids' upssegd upssegd+{-# NOINLINE updateVSegsReachable #-}+--  NOINLINE because we want to see this happening in core.+++-- Append ---------------------------------------------------------------------+-- | Produce a segment descriptor that describes the result of appending two arrays.+--- +--   * TODO: make this parallel.+--+appendWith+        :: UPVSegd      -- ^ Descriptor of first array.+        -> Int          -- ^ Number of flat physical arrays for first descriptor.+        -> UPVSegd      -- ^ Descriptor of second array.+        -> Int          -- ^ Number of flat physical arrays for second descriptor.+        -> UPVSegd++appendWith+        (UPVSegd _ _ vsegids1 _ upssegd1) pdatas1+        (UPVSegd _ _ vsegids2 _ upssegd2) pdatas2++ = let  -- vsegids releative to appended psegs+        vsegids1' = vsegids1+        vsegids2' = US.map (+ UPSSegd.length upssegd1) vsegids2+        +        -- append the vsegids+        vsegids'  = vsegids1' US.++ vsegids2'++        -- All data from the source arrays goes into the result+        upssegd'  = UPSSegd.appendWith+                                upssegd1 pdatas1+                                upssegd2 pdatas2+                                 +   in   UPVSegd False vsegids' vsegids' upssegd' upssegd'+{-# NOINLINE appendWith #-}+--  NOINLINE because it doesn't need to be specialised+--           and we're worried about code explosion.+++-- Combine --------------------------------------------------------------------+-- | Combine two virtual segment descriptors.+---+--   * TODO: make this parallel. +--+combine2+        :: UPSel2       -- ^ Selector for the combine operation.+        -> UPVSegd      -- ^ Descriptor of first array.+        -> Int          -- ^ Number of flat physical arrays for first descriptor.+        -> UPVSegd      -- ^ Descriptor of second array.+        -> Int          -- ^ Number of flat physical arrays for second descriptor.+        -> UPVSegd+        +combine2+        upsel2+        (UPVSegd _ _ vsegids1 _ upssegd1) pdatas1+        (UPVSegd _ _ vsegids2 _ upssegd2) pdatas2++ = let  -- vsegids relative to combined psegs+        vsegids1' = vsegids1+        vsegids2' = US.map (+ (US.length vsegids1)) vsegids2++        -- combine the vsegids+        vsegids'  = US.combine2ByTag (UPSel.tagsUPSel2 upsel2)+                                    vsegids1' vsegids2'++         -- All data from the source arrays goes into the result+        upssegd'  = UPSSegd.appendWith+                                upssegd1 pdatas1+                                upssegd2 pdatas2+                                  +   in   UPVSegd False vsegids' vsegids' upssegd' upssegd'+{-# NOINLINE combine2 #-}+--  NOINLINE because it doesn't need to be specialised+--           and we're worried about code explosion.+

@@ -1,12 +1,14 @@ Name:           dph-prim-par-Version:        0.5.1.1+Version:        0.6.0.1 License:        BSD3 License-File:   LICENSE Author:         The DPH Team Maintainer:     Ben Lippmeier <benl@cse.unsw.edu.au> Homepage:       http://www.haskell.org/haskellwiki/GHC/Data_Parallel_Haskell Category:       Data Structures-Synopsis:       Parallel Primitives for Data-Parallel Haskell.+Synopsis:       Data Parallel Haskell segmented arrays. (production version)+Description:    Production implementation of the flat parallel array API defined+                in dph-prim-interface.  Cabal-Version:  >= 1.6 Build-Type:     Simple@@ -14,39 +16,59 @@ Library   Exposed-Modules:         Data.Array.Parallel.Unlifted.Distributed+        Data.Array.Parallel.Unlifted.Distributed.Gang+        Data.Array.Parallel.Unlifted.Distributed.TheGang         Data.Array.Parallel.Unlifted.Parallel+        Data.Array.Parallel.Unlifted.Parallel.UPSegd+        Data.Array.Parallel.Unlifted.Parallel.UPSSegd+        Data.Array.Parallel.Unlifted.Parallel.UPVSegd+        Data.Array.Parallel.Unlifted.Parallel.UPSel         Data.Array.Parallel.Unlifted+   Other-Modules:-        Data.Array.Parallel.Unlifted.Distributed.Gang-        Data.Array.Parallel.Unlifted.Distributed.TheGang         Data.Array.Parallel.Unlifted.Distributed.DistST         Data.Array.Parallel.Unlifted.Distributed.Types         Data.Array.Parallel.Unlifted.Distributed.Combinators         Data.Array.Parallel.Unlifted.Distributed.Scalars         Data.Array.Parallel.Unlifted.Distributed.Arrays+        Data.Array.Parallel.Unlifted.Distributed.USegd+        Data.Array.Parallel.Unlifted.Distributed.USSegd         Data.Array.Parallel.Unlifted.Distributed.Basics-        Data.Array.Parallel.Unlifted.Parallel.Combinators-        Data.Array.Parallel.Unlifted.Parallel.Sums+        Data.Array.Parallel.Unlifted.Distributed.Types.USegd+        Data.Array.Parallel.Unlifted.Distributed.Types.USSegd+        Data.Array.Parallel.Unlifted.Distributed.Types.UVSegd+        Data.Array.Parallel.Unlifted.Distributed.Types.Vector+        Data.Array.Parallel.Unlifted.Distributed.Types.Maybe+        Data.Array.Parallel.Unlifted.Distributed.Types.Tuple+        Data.Array.Parallel.Unlifted.Distributed.Types.Prim+        Data.Array.Parallel.Unlifted.Distributed.Types.Unit+        Data.Array.Parallel.Unlifted.Distributed.Types.Base         Data.Array.Parallel.Unlifted.Parallel.Basics-        Data.Array.Parallel.Unlifted.Parallel.Permute+        Data.Array.Parallel.Unlifted.Parallel.Combinators         Data.Array.Parallel.Unlifted.Parallel.Enum+        Data.Array.Parallel.Unlifted.Parallel.Extracts+        Data.Array.Parallel.Unlifted.Parallel.Permute         Data.Array.Parallel.Unlifted.Parallel.Segmented         Data.Array.Parallel.Unlifted.Parallel.Subarrays-        Data.Array.Parallel.Unlifted.Parallel.UPSegd-        Data.Array.Parallel.Unlifted.Parallel.UPSel+        Data.Array.Parallel.Unlifted.Parallel.Sums         Data.Array.Parallel.Unlifted.Parallel.Text    Exposed: False    Extensions: TypeFamilies, GADTs, RankNTypes,-              BangPatterns, MagicHash, UnboxedTuples, TypeOperators-  GHC-Options: -Odph -funbox-strict-fields -fcpr-off+              BangPatterns, MagicHash, UnboxedTuples, TypeOperators,+              FlexibleInstances, FlexibleContexts ++  GHC-Options:+        -Odph -funbox-strict-fields+        -fcpr-off -Wall+   Build-Depends:  -        base     == 4.4.*,-        random   == 1.0.*,-        vector   == 0.7.*,-        old-time == 1.0.*,-        dph-base == 0.5.*,-        dph-prim-interface == 0.5.*,-        dph-prim-seq       == 0.5.*+        base               == 4.5.*,+        random             == 1.0.*,+        vector             == 0.9.*,+        old-time           == 1.1.*,+        dph-base           == 0.6.*,+        dph-prim-interface == 0.6.*,+        dph-prim-seq       == 0.6.*