packages feed

hdph (empty) → 0.0.1

raw patch · 27 files changed

+5546/−0 lines, 27 filesdep +basedep +bytestringdep +cerealsetup-changed

Dependencies added: base, bytestring, cereal, containers, deepseq, hdph-closure, mtl, network, network-info, network-multicast, network-transport, network-transport-tcp, random, template-haskell, time

Files

+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Patrick Maier, Rob Stewart, 2012++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the names of the copyright holders nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,206 @@+Haskell Distributed Parallel Haskell+====================================++**Haskell distributed parallel Haskell (HdpH)** is a Haskell DSL for+parallel computation on distributed-memory architectures. HdpH is+implemented entirely in Haskell but does make use of a few GHC extensions,+most notably TemplateHaskell.++HdpH is described in some detail in the paper [Implementing a High-level Distributed-Memory Parallel Haskell in Haskell](http://www.macs.hw.ac.uk/~pm175/papers/Maier_Trinder_IFL2011_XT.pdf).+The paper on [The Design of SymGridPar2](http://www.macs.hw.ac.uk/~pm175/papers/Maier_Stewart_Trinder_SAC2013.pdf) paints the bigger picture of what HdpH was designed for and where future developments will lead.++This release is considered alpha stage.+++Building HdpH+-------------++Should be straightforward from the cabalised package `hdph`.+Note that `hdph` depends on `hdph-closure`, an independent package that+factors out the closure representation of HdpH.+++Running HdpH+------------++### Launch via `ssh`++HdpH comes with a few sample applications. The simplest one,+a distributed _Hello World_, is good for testing whether HdpH works.+For example, the following Bourne shell command will run _Hello World_+on three nodes, distributed over two hosts, `bwlf01` and `bwlf02`.++    $> for host in bwlf01 bwlf02 bwlf01; do ssh $host hdph/dist/build/hello/hello -numProcs=3 & done++This will launch three instances of the _Hello World_ binary `hello` (in+directory `hdph/dist/build/hello` relative to the user's home directory),+two on `bwlf01` and one on `bwlf02`.+Note that the obligatory option `-numProcs` must match the number of instances;+HdpH will likely hang otherwise.++Assuming that `ssh` has been set up to enable password-less logins, the output+should be something like this:++    Master Node:137.195.143.101:19754:0 wants to know: Who is here?+    Hello from Node:137.195.143.101:19754:0+    Hello from Node:137.195.143.102:38939:0+    Hello from Node:137.195.143.101:30062:0+++### Launch via `mpiexec`++Shell scripts and `ssh` are quite a cumbersome way of launching HdpH.+More convenient are dedicated parallel job launchers such as `mpiexec`,+which comes with any recent MPI distribution.+Even though this version of HdpH does not use MPI, it can still be launched+by `mpiexec`.+An MPICH installation, for example, might launch the _Hello World_ application+with the following command:++    $> mpiexec -hosts bwlf01,bwlf02,bwlf01 ./hello -numProcs=3++This will launch the _Hello World_ binary `hello` (in the current working+directory) 3 times, twice on `bwlf01` and once on `bwlf02`.+The expected output should look somewhat like this:++    Hello from Node:137.195.143.102:12701:0+    Hello from Node:137.195.143.101:15353:0+    Master Node:137.195.143.101:14247:0 wants to know: Who is here?+    Hello from Node:137.195.143.101:14247:0+++### Parallel launch via `mpiexec`++To test parallelism, there is a sample application computing the `n`th number+of the _Fibonacci_ sequence using a naive parallel divide-and-conquer algorithm.+For example, the following command will compute the 45th Fibonacci number in+parallel on 3 nodes, switching to sequential execution for subproblems below+a threshold of 30; a wall clock runtime of about 16 seconds is reported here.++    $> mpiexec -hosts bwlf01,bwlf02,bwlf03 ./fib -numProcs=3 v2 45 30+    {v2, seqThreshold=30, parThreshold=30} fib 45 = 1836311903 {runtime=16.157814s}++When passed the switch `-d1` HdpH will produce some summary output per+node relating to parallelism, eg. number of sparks generated per node,+maximum number of sparks residing on a node, number of times a node+requested work, and number of times work was scheduled in response.+An example run with `-d1` might look like this:++    $> mpiexec -hosts bwlf01,bwlf02,bwlf03 ./fib -numProcs=3 -d1 v2 45 30+    137.195.143.101:9759:0 #SPARK=1017   max_SPARK=154   max_THREAD=[3,1]+    137.195.143.101:9759:0 #FISH_sent=1   #SCHED_rcvd=0+    {v2, seqThreshold=30, parThreshold=30} fib 45 = 1836311903 {runtime=16.449183s}+    137.195.143.102:12551:0 #SPARK=308   max_SPARK=4   max_THREAD=[1,1]+    137.195.143.102:12551:0 #FISH_sent=215   #SCHED_rcvd=214+    137.195.143.103:27402:0 #SPARK=271   max_SPARK=4   max_THREAD=[0,1]+    137.195.143.103:27402:0 #FISH_sent=225   #SCHED_rcvd=224++This shows that `bwlf01` (IP address `137.195.143.101`) was the master node,+generating 1017 sparks in total, requesting work once (probably at the end+of the computation) and not receiving any.+The other nodes did generate some sparks themselves (308 and 271, respectively)+but they also both received a substantial number of sparks (214 and 224,+respectively).+Most importantly, each node requested work exactly once more often than+receiving it, which means no node was ever idling, except probably at the+very end of the computation.+++### Parallel launch via `mpiexec` on homogeneous multicores++So far HdpH ran in a single thread on each node.+To make use of multicores `mpiexec` can place several single-threaded nodes+on the same host, eg. the following examples will launch two and four nodes per+host, respectively; note that the number following the `-n` switch of `mpiexec`+must match the number following the `-numProcs` switch of `fib`.++    $> mpiexec -hosts bwlf01,bwlf02,bwlf03 -n 6 ./fib -numProcs=6 v2 45 30+    {v2, seqThreshold=30, parThreshold=30} fib 45 = 1836311903 {runtime=8.026417s}++    $> mpiexec -hosts bwlf01,bwlf02,bwlf03 -n 12 ./fib -numProcs=12 v2 45 30+    {v2, seqThreshold=30, parThreshold=30} fib 45 = 1836311903 {runtime=4.845796s}++The other possibility is to run HdpH itself in a multi-threaded mode.+The following two examples run HdpH on two and four threads per node,+respectively, expecting that the OS will bind each thread to a core.++    $> mpiexec -hosts bwlf01,bwlf02,bwlf03 ./fib -numProcs=3 -scheds=2 v2 45 30 +RTS -N2+    {v2, seqThreshold=30, parThreshold=30} fib 45 = 1836311903 {runtime=10.648305s}++    $> mpiexec -hosts bwlf01,bwlf02,bwlf03 ./fib -numProcs=3 -scheds=4 v2 45 30 +RTS -N4+    {v2, seqThreshold=30, parThreshold=30} fib 45 = 1836311903 {runtime=6.507969s}++Note that the GHC RTS switch `-N` determines the number of threads (or+HECs in GHC terminology) whereas the `fib` switch `-scheds` determines+how many of these threads run the HdpH scheduler loop. Whether there should+be as many schedulers as threads or less depends on the GHC version and+probably on the OS. Using one scheduler less than the number of threads+has been observed to reduce variability, and sometimes even bring+performance gains:++    $>  mpiexec -hosts bwlf01,bwlf02,bwlf03 ./fib -numProcs=3 -scheds=3 v2 45 30 +RTS -N4+    {v2, seqThreshold=30, parThreshold=30} fib 45 = 1836311903 {runtime=5.831932s}+++### Parallel launch via `mpiexec` on heterogeneous multicores++So far all hosts were assumed to have the same number of cores.+However, `mpiexec` can be used to launch on heterogeneous clusters.+For instance, the following call will launch HdpH with two threads each on+`bwlf01` and `bwlf02`, and with four threads on `bwlf03`; please see the man+page of `mpiexec` for an explanation of the command line format.++    $> mpiexec -hosts bwlf01,bwlf02,bwlf03 -n 2 ./fib -numProcs=3 -scheds=2 -d1 v2 45 30 +RTS -N2 : -n 1 ./fib -numProcs=3 -scheds=4 -d1 v2 45 30 +RTS -N4+    137.195.143.101:21394:0 #SPARK=1114   max_SPARK=183   max_THREAD=[3,1,1]+    137.195.143.101:21394:0 #FISH_sent=3   #SCHED_rcvd=1+    137.195.143.103:22027:0 #SPARK=261   max_SPARK=2   max_THREAD=[0,1,1,1,1]+    137.195.143.103:22027:0 #FISH_sent=209   #SCHED_rcvd=208+    137.195.143.102:22374:0 #SPARK=221   max_SPARK=3   max_THREAD=[1,1,1]+    137.195.143.102:22374:0 #FISH_sent=173   #SCHED_rcvd=172+    {v2, seqThreshold=30, parThreshold=30} fib 45 = 1836311903 {runtime=10.502205s}++The switch `-d1` reveals that `bwlf03` (IP address `137.195.143.103`) did+indeed run 4 schedulers because its `max_THREAD` list was length 5, and+the length of the `max_THREAD` list is always 1 plus the number of schedulers.+++### Launch caveats++At startup HdpH nodes open random ports and rely on UDP multicasts to discover+each other. This results in a number of limitations:+++ UDP multicasts must be routed between all nodes (which may imply that all+  hosts must reside in the same subnet).+++ The obligatory `-numProcs` switch, which tells an application how many+  nodes to expect, must match exactly the number of nodes launched.+++ Node discovery must be completed within a certain time frame (typically 10+  seconds).+++ No second HdpH application may be launched during the node discovery phase.++Any deviation from the above limitations is likely to cause HdpH to hang+forever.+++Related Projects+----------------++* [Cloud Haskell](http://haskell-distributed.github.com)++* [Par Monad and Friends](https://github.com/simonmar/monad-par)+++References+----------++1.  Patrick Maier, Phil Trinder.+    [Implementing a High-level Distributed-Memory Parallel Haskell in Haskell](http://www.macs.hw.ac.uk/~pm175/papers/Maier_Trinder_IFL2011_XT.pdf).+    Proc. 2011 Symposium on Implementation and Application of Functional Languages (IFL 2011), Springer LNCS 7257, pp. 35-50.++2.  Patrick Maier, Rob Stewart, Phil Trinder.+    [Reliable Scalable Symbolic Computation: The Design of SymGridPar2](http://www.macs.hw.ac.uk/~pm175/papers/Maier_Stewart_Trinder_SAC2013.pdf).+    Proc. 28th ACM Symposium On Applied Computing (SAC 2013), pp. 1677-1684.++3.  [HdpH development repository](https://github.com/PatrickMaier/HdpH) on github.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ TODO.md view
@@ -0,0 +1,19 @@+Haskell Distributed Parallel Haskell+====================================++To Do List+----------++* Clean up build.++    * Add `-Wall` switch to cabal file.+    * Suppress spurious warnings about orphan instances.+    * Remove unnecessary monad stack in the implementation of HdpH RTS.++* The work stealing protocol of HdpH is asynchronous and message based.+  TCP isn't a perfect fit; a message based transport protocol (like UDP)+  might be better, at least for small messages (ie. message size < MTU).++* Currenly, the `runParIO` function initiates startup and shutdown of the+  distributed HdpH VM. It might be beneficial to separate VM startup/shutdown+  from any actual parallel computation.
+ hdph.cabal view
@@ -0,0 +1,134 @@+name:          hdph+version:       0.0.1+synopsis:      Haskell distributed parallel Haskell+description:   Haskell distributed parallel Haskell (HdpH) is a Haskell DSL+               for distributed-memory parallelism, implemented entirely in+               Haskell (as supported by GHC).+homepage:      https://github.com/PatrickMaier/HdpH+license:       BSD3+license-file:  LICENSE+author:        Patrick Maier <C.Patrick.Maier@gmail.com>,+               Rob Stewart <robstewart57@gmail.com>+maintainer:    Patrick Maier <C.Patrick.Maier@gmail.com>+stability:     experimental+category:      Control, Parallelism, Distributed Computing, Monads+tested-with:   GHC == 7.4.1 || == 7.6.2+build-type:    Simple+cabal-version: >= 1.8++Library+  exposed-modules: Control.Parallel.HdpH,+                   Control.Parallel.HdpH.Conf,+                   Control.Parallel.HdpH.Strategies+  other-modules:   Control.Parallel.HdpH.Internal.Comm,+                   Control.Parallel.HdpH.Internal.Data.Deque,+                   Control.Parallel.HdpH.Internal.Data.Sem,+                   Control.Parallel.HdpH.Internal.GRef,+                   Control.Parallel.HdpH.Internal.IVar,+                   Control.Parallel.HdpH.Internal.Location,+                   Control.Parallel.HdpH.Internal.Misc,+                   Control.Parallel.HdpH.Internal.Scheduler,+                   Control.Parallel.HdpH.Internal.Sparkpool,+                   Control.Parallel.HdpH.Internal.State.GRef,+                   Control.Parallel.HdpH.Internal.State.Location,+                   Control.Parallel.HdpH.Internal.Threadpool,+                   Control.Parallel.HdpH.Internal.Type.GRef,+                   Control.Parallel.HdpH.Internal.Type.Location,+                   Control.Parallel.HdpH.Internal.Type.Par+  build-depends:   template-haskell,+                   base >= 4 && < 5,+                   cereal >= 0.3.3 && < 0.4,+                   bytestring == 0.10.*,+                   containers >= 0.1 && < 0.6,+                   deepseq >= 1.1 && < 2,+                   mtl >= 2 && < 3,+                   network == 2.4.*,+                   network-info == 0.2.*,+                   network-multicast >= 0.0.7 && < 0.1,+                   network-transport == 0.3.*,+                   network-transport-tcp == 0.3.*,+                   random >= 1 && < 2,+                   time >= 1.2 && < 2,+                   hdph-closure == 0.0.1+  hs-source-dirs:  src+  ghc-options:     ++Executable hello+  main-is:         Test/HdpH/hello.hs+  build-depends:   template-haskell,+                   base >= 4 && < 5,+                   cereal >= 0.3.3 && < 0.4,+                   bytestring == 0.10.*,+                   containers >= 0.1 && < 0.6,+                   deepseq >= 1.1 && < 2,+                   mtl >= 2 && < 3,+                   network == 2.4.*,+                   network-info == 0.2.*,+                   network-multicast >= 0.0.7 && < 0.1,+                   network-transport == 0.3.*,+                   network-transport-tcp == 0.3.*,+                   random >= 1 && < 2,+                   time >= 1.2 && < 2,+                   hdph-closure == 0.0.1+  hs-source-dirs:  src+  ghc-options:     -threaded++Executable fib+  main-is:         Test/HdpH/fib.hs+  build-depends:   template-haskell,+                   base >= 4 && < 5,+                   cereal >= 0.3.3 && < 0.4,+                   bytestring == 0.10.*,+                   containers >= 0.1 && < 0.6,+                   deepseq >= 1.1 && < 2,+                   mtl >= 2 && < 3,+                   network == 2.4.*,+                   network-info == 0.2.*,+                   network-multicast >= 0.0.7 && < 0.1,+                   network-transport == 0.3.*,+                   network-transport-tcp == 0.3.*,+                   random >= 1 && < 2,+                   time >= 1.2 && < 2,+                   hdph-closure == 0.0.1+  hs-source-dirs:  src+  ghc-options:     -threaded++Executable sumeuler+  main-is:         Test/HdpH/sumeuler.hs+  build-depends:   template-haskell,+                   base >= 4 && < 5,+                   cereal >= 0.3.3 && < 0.4,+                   bytestring == 0.10.*,+                   containers >= 0.1 && < 0.6,+                   deepseq >= 1.1 && < 2,+                   mtl >= 2 && < 3,+                   network == 2.4.*,+                   network-info == 0.2.*,+                   network-multicast >= 0.0.7 && < 0.1,+                   network-transport == 0.3.*,+                   network-transport-tcp == 0.3.*,+                   random >= 1 && < 2,+                   time >= 1.2 && < 2,+                   hdph-closure == 0.0.1+  hs-source-dirs:  src+  ghc-options:     -threaded++Executable nbody+  main-is:         Test/HdpH/nbody.hs+  build-depends:   template-haskell,+                   base >= 4 && < 5,+                   cereal >= 0.3.3 && < 0.4,+                   bytestring == 0.10.*,+                   containers >= 0.1 && < 0.6,+                   deepseq >= 1.1 && < 2,+                   mtl >= 2 && < 3,+                   network == 2.4.*,+                   network-info == 0.2.*,+                   network-multicast >= 0.0.7 && < 0.1,+                   network-transport == 0.3.*,+                   network-transport-tcp == 0.3.*,+                   random >= 1 && < 2,+                   time >= 1.2 && < 2,+                   hdph-closure == 0.0.1+  hs-source-dirs:  src+  ghc-options:     -threaded
+ src/Control/Parallel/HdpH.hs view
@@ -0,0 +1,345 @@+-- HdpH programming interface+--+-- Author: Patrick Maier, Rob Stewart+-----------------------------------------------------------------------------++{-# LANGUAGE GeneralizedNewtypeDeriving #-}  -- for 'GIVar' and 'NodeId'+{-# LANGUAGE TemplateHaskell #-}             -- for 'mkClosure', etc.++module Control.Parallel.HdpH+  ( -- $Intro++    -- * Par monad+    -- $Par_monad+    Par,        -- kind * -> *; instances: Functor, Monad+    runParIO_,     -- :: RTSConf -> Par () -> IO ()+    runParIO,      -- :: RTSConf -> Par a -> IO (Maybe a)++    -- * Operations in the Par monad+    -- $Par_ops+    done,       -- :: Par a+    myNode,     -- :: Par NodeId+    allNodes,   -- :: Par [NodeId]+    io,         -- :: IO a -> Par a+    eval,       -- :: a -> Par a+    force,      -- :: (NFData a) => a -> Par a+    fork,       -- :: Par () -> Par ()+    spark,      -- :: Closure (Par ()) -> Par ()+    pushTo,     -- :: Closure (Par ()) -> NodeId -> Par ()+    new,        -- :: Par (IVar a)+    put,        -- :: IVar a -> a -> Par ()+    get,        -- :: IVar a -> Par a+    tryGet,     -- :: IVar a -> Par (Maybe a)+    probe,      -- :: IVar a -> Par Bool+    glob,       -- :: IVar (Closure a) -> Par (GIVar (Closure a))+    rput,       -- :: GIVar (Closure a) -> Closure a -> Par ()++    -- * Locations+    NodeId,++    -- * Local and global IVars+    IVar,+    GIVar,+    at,++    -- * Explicit Closures+    module Control.Parallel.HdpH.Closure,++    -- * Runtime system configuration+    module Control.Parallel.HdpH.Conf,++    -- * This module's Static declaration+    declareStatic+  ) where++import Prelude hiding (error)+import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)+import Control.DeepSeq (NFData, deepseq)+import Control.Monad (when)+import Data.Functor ((<$>))+import Data.IORef (newIORef, readIORef, writeIORef)+import Data.Monoid (mconcat)+import Data.Serialize (Serialize)++import Control.Parallel.HdpH.Conf                            -- re-export whole module+import Control.Parallel.HdpH.Closure hiding (declareStatic)  -- re-export almost whole module+import qualified Control.Parallel.HdpH.Closure as Closure (declareStatic)+import qualified Control.Parallel.HdpH.Internal.Comm as Comm+       (myNode, allNodes, isMain, shutdown)+import qualified Control.Parallel.HdpH.Internal.IVar as IVar (IVar, GIVar)+import Control.Parallel.HdpH.Internal.IVar+       (hostGIVar, newIVar, putIVar, getIVar, pollIVar, probeIVar,+        globIVar, putGIVar)+import qualified Control.Parallel.HdpH.Internal.Location as Location+       (NodeId, dbgStaticTab)+import Control.Parallel.HdpH.Internal.Scheduler+       (RTS, liftThreadM, liftSparkM, liftCommM, liftIO,+        schedulerID, mkThread, execThread, sendPUSH)+import qualified Control.Parallel.HdpH.Internal.Scheduler as Scheduler (run_)+import Control.Parallel.HdpH.Internal.Sparkpool (putSpark)+import Control.Parallel.HdpH.Internal.Threadpool (putThread, putThreads)+import Control.Parallel.HdpH.Internal.Type.Par (ParM(Par), Thread(Atom))+++-----------------------------------------------------------------------------+-- Static declaration++-- | Static declaration of Static deserialisers used in explicit Closures+-- created or imported by this module.+-- This Static declaration must be imported by every main module using HdpH.+-- The imported Static declaration must be combined with the main module's own+-- Static declaration and registered; failure to do so may abort the program+-- at runtime.+declareStatic :: StaticDecl+declareStatic = mconcat+  [Closure.declareStatic,+   declare $(static 'rput_abs)]+++-----------------------------------------------------------------------------+-- $Intro+-- HdpH (/Haskell distributed parallel Haskell/) is a Haskell DSL for shared-+-- and distributed-memory parallelism, implemented entirely in Haskell+-- (as supported by the GHC). HdpH is described in the following paper:+--+-- P. Maier, P. W. Trinder.+-- /Implementing a High-level Distributed-Memory Parallel Haskell in Haskell/.+-- IFL 2011.+--+-- HdpH executes programs written in a monadic embedded DSL for shared-+-- and distributed-memory parallelism.  +-- HdpH operates a distributed runtime system, scheduling tasks+-- either explicitly (controled by the DSL) or implicitly (by work stealing).+-- The runtime system distinguishes between nodes and schedulers.+-- A /node/ is an OS process running HdpH (that is, a GHC-compiled executable+-- built on top of the HdpH library), whereas a /scheduler/ is a Haskell IO+-- thread executing HdpH expressions (that is, 'Par' monad computation).+-- As a rule of thumb, a node should correspond to a machine in a network,+-- and a scheduler should correspond to a core in a machine.+--+-- The semantics of HdpH was developed with fault tolerance in mind (though+-- this version of HdpH is not yet fault tolerant). In particular, HdpH+-- allows the replication computations, and the racing of computations+-- against each other. The price to pay for these features is that HdpH+-- cannot enforce determinism.+++-----------------------------------------------------------------------------+-- abstract Locations++-- | A 'NodeId' identifies a node (that is, an OS process running HdpH).+-- A 'NodeId' should be thought of as an abstract identifier which+-- instantiates the classes 'Eq', 'Ord', 'Show', 'NFData' and 'Serialize'.+newtype NodeId = NodeId Location.NodeId+                 deriving (Eq, Ord, NFData, Serialize)++-- Show instance (mainly for debugging)+instance Show NodeId where+  showsPrec _ (NodeId n) = showString "Node:". shows n+++-----------------------------------------------------------------------------+-- abstract IVars and GIVars++-- | An IVar is a write-once one place buffer.+-- IVars are abstract; they can be accessed and manipulated only by+-- the operations 'put', 'get', 'tryGet', 'probe' and 'glob'.+newtype IVar a = IVar (IVar.IVar RTS a)+++-- | A GIVar (short for /global/ IVar) is a globally unique handle referring+-- to an IVar.+-- Unlike IVars, GIVars can be compared and serialised.+-- They can also be written to remotely by the operation 'rput'.+newtype GIVar a = GIVar (IVar.GIVar RTS a)+                  deriving (Eq, Ord, NFData, Serialize)++-- Show instance (mainly for debugging)+instance Show (GIVar a) where+  showsPrec _ (GIVar gv) = showString "GIVar:" . shows gv+++-- | Returns the node hosting the IVar referred to by the given GIVar.+-- This function being pure implies that IVars cannot migrate between nodes.+at :: GIVar a -> NodeId+at (GIVar gv) = NodeId $ hostGIVar gv+++-----------------------------------------------------------------------------+-- abstract runtime system (don't export)++-- | Eliminate the 'RTS' monad down to 'IO' by running the given 'action';+-- aspects of the RTS's behaviour are controlled by the respective parameters+-- in the 'conf' argument.+runRTS_ :: RTSConf -> RTS () -> IO ()+runRTS_ = Scheduler.run_++-- | Return True iff this node is the root node.+isMainRTS :: RTS Bool+isMainRTS = liftCommM Comm.isMain++-- | Initiate RTS shutdown.+shutdownRTS :: RTS ()+shutdownRTS = liftCommM Comm.shutdown++-- Print global Static table to stdout, one entry a line.+printStaticTable :: RTS ()+printStaticTable =+  liftIO $ mapM_ putStrLn $+    "Static Table:" : map ("  " ++) showStaticTable+++-----------------------------------------------------------------------------+-- $Par_monad+-- 'Par' is the monad for parallel computations in HdpH.+-- It is a continuation monad, similar to the one described in paper+-- /A monad for deterministic parallelism/+-- by S. Marlow, R. Newton, S. Peyton Jones (Haskell 2011).++-- | 'Par' is type constructor of kind @*->*@ and an instance of classes+-- 'Functor' and 'Monad'.+-- 'Par' is defined in terms of a parametric continuation monad 'ParM'+-- by plugging in 'RTS', the state monad of the runtime system.+-- Since neither 'ParM' nor 'RTS' are exported, 'Par' can be considered+-- abstract.+type Par = ParM RTS+-- A newtype would be nicer than a type synonym but the resulting+-- wrapping and unwrapping destroys readability (if it is at all possible,+-- eg. inside Closures).+++-- | Eliminate the 'Par' monad by converting the given 'Par' action 'p'+-- into an 'RTS' action (to be executed on any one node of the distributed+-- runtime system).+runPar :: Par a -> RTS a+runPar p = do -- create an empty MVar expecting the result of action 'p'+              res <- liftIO $ newEmptyMVar++              -- fork 'p', combined with a write to above MVar;+              -- note that the starter thread (ie the 'fork') runs outwith+              -- any scheduler (and terminates quickly); the forked action+              -- (ie. 'p >>= ...') runs in a scheduler, however.+              execThread $ mkThread $ fork (p >>= io . putMVar res)++              -- block waiting for result+              liftIO $ takeMVar res+++-- | Eliminates the 'Par' monad by executing the given parallel computation 'p',+-- including setting up and initialising a distributed runtime system+-- according to the configuration parameter 'conf'.+-- This function lives in the IO monad because 'p' may be impure,+-- for instance, 'p' may exhibit non-determinism.+-- Caveat: Though the computation 'p' will only be started on a single root+-- node, 'runParIO_' must be executed on every node of the distributed runtime+-- system du to the SPMD nature of HdpH.+-- Note that the configuration parameter 'conf' applies to all nodes uniformly;+-- at present there is no support for heterogeneous configurations.+runParIO_ :: RTSConf -> Par () -> IO ()+runParIO_ conf p =+  runRTS_ conf $ do isMain <- isMainRTS+                    when isMain $ do+                      -- print Static table+                      when (Location.dbgStaticTab <= debugLvl conf)+                        printStaticTable+                      runPar p+                      shutdownRTS+++-- | Convenience: variant of 'runParIO_' which does return a result.+-- Caveat: The result is only returned on the root node; all other nodes+-- return 'Nothing'.+runParIO :: RTSConf -> Par a -> IO (Maybe a)+runParIO conf p = do res <- newIORef Nothing+                     runParIO_ conf (p >>= io . writeIORef res . Just)+                     readIORef res+++-----------------------------------------------------------------------------+-- $Par_ops+-- These operations form the HdpH DSL, a low-level API of for parallel+-- programming across shared- and distributed-memory architectures.+-- For a more high-level API see module "Control.Parallel.HdpH.Strategies".++-- | Terminates the current thread.+done :: Par a+done = Par $ \ _c -> Atom (return Nothing)++-- lifting RTS into the Par monad (really a monadic map); don't export+atom :: RTS a -> Par a+atom m = Par $ \ c -> Atom (return . Just . c =<< m)++-- | Returns the node this operation is currently executed on.+myNode :: Par NodeId+myNode = NodeId <$> (atom $ liftCommM $ Comm.myNode)++-- | Returns a list of all nodes currently forming the distributed+-- runtime system.+allNodes :: Par [NodeId]+allNodes = map NodeId <$> (atom $ liftCommM $ Comm.allNodes)++-- | Lifts an IO action into the Par monad.+io :: IO a -> Par a+io = atom . liftIO++-- | Evaluates its argument to weak head normal form.+eval :: a -> Par a+eval x = atom (x `seq` return x)++-- | Evaluates its argument to normal form (as defined by 'NFData' instance).+force :: (NFData a) => a -> Par a+force x = atom (x `deepseq` return x)++-- | Creates a new thread, to be executed on the current node.+fork :: Par () -> Par ()+fork = atom . liftThreadM . putThread . mkThread++-- | Creates a spark, to be available for work stealing.+-- The spark may be converted into a thread and executed locally, or it may+-- be stolen by another node and executed there.+spark :: Closure (Par ()) -> Par ()+spark clo = atom (schedulerID >>= \ i -> liftSparkM $ putSpark i clo)++-- | Pushes a computation to the given node, where it is eagerly converted+-- into a thread and executed.+pushTo :: Closure (Par ()) -> NodeId -> Par ()+pushTo clo (NodeId n) = atom $ sendPUSH clo n++-- | Creates a new empty IVar.+new :: Par (IVar a)+new = IVar <$> atom (liftIO $ newIVar)++-- | Writes to given IVar (without forcing the value written).+put :: IVar a -> a -> Par ()+put (IVar v) a = atom $ liftIO (putIVar v a) >>=+                        liftThreadM . putThreads++-- | Reads from given IVar; blocks if the IVar is empty.+get :: IVar a -> Par a+get (IVar v) = Par $ \ c -> Atom $ liftIO (getIVar v c) >>=+                                   maybe (return Nothing) (return . Just . c)++-- | Reads from given IVar; does not block but returns 'Nothing' if IVar empty.+tryGet :: IVar a -> Par (Maybe a)+tryGet (IVar v) = atom $ liftIO (pollIVar v)++-- | Tests whether given IVar is empty or full; does not block.+probe :: IVar a -> Par Bool+probe (IVar v) = atom $ liftIO (probeIVar v)++-- | Globalises given IVar, returning a globally unique handle;+-- this operation is restricted to IVars of 'Closure' type.+glob :: IVar (Closure a) -> Par (GIVar (Closure a))+glob (IVar v) = GIVar <$> atom (schedulerID >>= \ i -> liftIO $ globIVar i v)++-- | Writes to (possibly remote) IVar denoted by given global handle;+-- this operation is restricted to write valueso of 'Closure' type.+rput :: GIVar (Closure a) -> Closure a -> Par ()+rput gv clo = pushTo $(mkClosure [| rput_abs (gv, clo) |]) (at gv)++-- write to locally hosted global IVar; don't export+{-# INLINE rput_abs #-}+rput_abs :: (GIVar (Closure a), Closure a) -> Par ()+rput_abs (GIVar gv, clo) = atom $ schedulerID >>= \ i ->+                                  liftIO (putGIVar i gv clo) >>=+                                  liftThreadM . putThreads
+ src/Control/Parallel/HdpH/Conf.hs view
@@ -0,0 +1,95 @@+-- HpdH runtime configuration parameters+--+-- Author: Patrick Maier+-----------------------------------------------------------------------------++module Control.Parallel.HdpH.Conf+  ( -- * HdpH runtime system configuration parameters+    RTSConf(..),+    defaultRTSConf  -- :: RTSConf+  ) where++import Prelude++import Control.Parallel.HdpH.Internal.Location (dbgNone)+++-----------------------------------------------------------------------------+-- Runtime configuration parameters (for RTS monad stack)++-- | 'RTSConf' is a record data type collecting a number of parameter+-- governing the behaviour of the HdpH runtime system.+data RTSConf =+  RTSConf {+    debugLvl :: Int,+        -- ^ Debug level, a number defined in module+        -- "Control.Parallel.HdpH.Internal.Location".+        -- Default is 0 (corresponding to no debug output).++    scheds :: Int,+        -- ^ Number of concurrent schedulers per node. Must be positive and +        -- should be @<=@ to the number of HECs (as set by GHC RTS option +        -- @-N@). Default is 1.++    wakeupDly :: Int,+        -- ^ Interval in microseconds to wake up sleeping schedulers+        -- (which is necessary to recover from a race condition between+        -- concurrent schedulers). Must be positive. +        -- Default is 1000 (corresponding to 1 millisecond).++    maxHops :: Int,+        -- ^ Number of hops a FISH message may travel before being considered+        -- failed. Must be non-negative. Default is 7.++    maxFish :: Int,+        -- ^ Low sparkpool watermark for fishing. RTS will send FISH message +        -- unless size of spark pool is greater than 'maxFish' (or unless +        -- a FISH is outstanding). Must be non-negative;+        -- should be @<@ 'minSched'. Default is 1.++    minSched :: Int,+        -- ^ Low sparkpool watermark for scheduling. RTS will respond to FISH +        -- messages by SCHEDULEing sparks unless size of spark pool is less+        -- than 'minSched'. Must be non-negative; should be @>@ 'maxFish'.+        -- Default is 2.++    minFishDly :: Int,+        -- ^ After a failed FISH, minimal delay in microseconds before+        -- sending another FISH message; the actual delay is chosen randomly+        -- between 'minFishDly' and 'maxFishDly'. Must be non-negative; should+        -- be @<=@ 'maxFishDly'.+        -- Default is 10000 (corresponding to 10 milliseconds).++    maxFishDly :: Int,+        -- ^ After a failed FISH, maximal delay in microseconds before+        -- sending another FISH message; the actual delay is chosen randomly+        -- between 'minFishDly' and 'maxFishDly'. Must be non-negative; should+        -- be @>=@ 'minFishDly'.+        -- Default is 1000000 (corresponding to 1 second).++    numProcs   :: Int,+        -- ^ Number of nodes constituting the distributed runtime system.+        -- Must be positive. Default is 1.++    networkInterface :: String+        -- ^ Network interface, required to autodetect a node's+        -- IP address. The string must be one of the interface names +        -- returned by the POSIX command @ifconfig@. +        -- Default is @eth0@ (corresponding to the first Ethernet interface).+    }+++-- | Default runtime system configuration parameters.+defaultRTSConf :: RTSConf+defaultRTSConf =+  RTSConf {+    debugLvl   = dbgNone,  -- no debug information+    scheds     = 1,        -- only 1 scheduler by default+    wakeupDly  = 1000,     -- wake up one sleeping scheduler every millisecond+    maxHops    = 7,        -- no more than 7 hops per FISH+    maxFish    = 1,        -- send FISH when <= 1 spark in pool+    minSched   = 2,        -- reply with SCHEDULE when >= 2 sparks in pool+    minFishDly = 10000,    -- delay at least 10 milliseconds after failed FISH+    maxFishDly = 1000000,  -- delay up to 1 second after failed FISH+    numProcs   = 1,        -- only 1 node by default+    networkInterface = "eth0" }  -- first Ethernet adapter default inferface
+ src/Control/Parallel/HdpH/Internal/Comm.hs view
@@ -0,0 +1,634 @@+-- Node to node communication (via TCP)+--+-- Uses the transport layer abstraction for distributed Haskell communication+-- Hackage: http://hackage.haskell.org/package/distributed-process+-- GitHub:  https://github.com/haskell-distributed/distributed-process+--+-- Author: Rob Stewart, Patrick Maier+-----------------------------------------------------------------------------++{-# LANGUAGE CPP #-}+{-# LANGUAGE ScopedTypeVariables #-}         -- for 'IOException'++module Control.Parallel.HdpH.Internal.Comm+  ( -- * CommM monad+    CommM,           -- synonym: Control.Monad.Reader.ReaderT <State> IO+    run_,            -- :: RTSConf -> CommM () -> IO ()+    liftIO,          -- :: IO a -> CommM a++    -- * information about the virtual machine+    nodes,           -- :: CommM Int+    allNodes,        -- :: CommM [NodeId]+    myNode,          -- :: CommM NodeId+    isMain,          -- :: CommM Bool++    -- * sending and receiving messages+    Message,         -- synomyn: MPI.Msg (= Data.ByteString.Lazy.ByteString)+    send,            -- :: NodeId -> Message -> CommM ()+    receive,         -- :: CommM Message+    shutdown,        -- :: CommM ()+    waitShutdown     -- :: CommM ()+  ) where++import Prelude hiding (error)+import qualified Prelude (error)+import Control.DeepSeq (NFData(rnf),force)+import Control.Exception (throw)+import Control.Monad (unless,void,when,forever)+import Control.Monad.Reader (ReaderT, runReaderT, ask)+import Control.Monad.Trans (lift)+import Data.Functor ((<$>))+import Data.IORef (writeIORef,atomicModifyIORef)+import qualified Data.Serialize (Serialize, put, get)+import Data.Word (Word8)++import Control.Parallel.HdpH.Internal.Misc (encodeLazy, decodeLazy)+import Control.Parallel.HdpH.Conf +       (RTSConf(debugLvl),numProcs, networkInterface)+import Control.Parallel.HdpH.Internal.Location +       (NodeId, MyNodeException(NodeIdUnset), error, dbgNone)+import Control.Parallel.HdpH.Internal.State.Location (myNodeRef, debugRef)+import Control.Concurrent+import System.IO (hPutStrLn,stderr) -- Used by HDPH_DEBUG+import System.Exit (ExitCode(..),exitWith)+import System.Timeout (timeout)+++import qualified Data.ByteString.Lazy as Lazy (ByteString,toChunks,fromChunks)+import qualified Network.Transport as NT+import qualified Network.Transport.TCP as TCP+import Data.IORef (IORef, newIORef, readIORef)+import System.IO.Unsafe (unsafePerformIO)+import Data.Maybe+import qualified Data.Map as Map+import Data.List ((\\))+import Control.Exception (SomeException,try)++import Network.Multicast+import Network.Socket.ByteString (sendTo, recvFrom)+import Network.Socket (Socket)+import Network.Info+import Data.List (sort)+import System.Random (randomRs, newStdGen)++-----------------------------------------------------------------------------+-- state representation++data State =+  State { s_conf     :: RTSConf,  -- config data+          s_nodes    :: Int,      -- # nodes of virtual machine+          s_allNodes :: [NodeId], -- all nodes of virtual machine+          s_myNode   :: NodeId,   -- currently executing node+          s_isMain   :: Bool,     -- True iff currently executing is main node+          s_msgQ     :: MessageQ, -- queue holding received payload messages+          s_shutdown :: MVar () }  -- shutdown signal++-- concurrent message queue (storing payload messages)+type MessageQ = Chan Message++-- a message is an MPI message (here: a lazy byte string)+type Message = Lazy.ByteString++-----------------------------------------------------------------------------+-- CommM monad++-- CommM is a reader monad on top of the IO monad; mutable parts of the state +-- (namely the message queue) are implemented via mutable references.+type CommM = ReaderT State IO++-- lifting lower layers+liftIO :: IO a -> CommM a+liftIO = lift+++-----------------------------------------------------------------------------+-- access to individual bits of state++-- Number of nodes in the virtual machine.+nodes :: CommM Int+nodes = s_nodes <$> ask++-- List of all nodes in the virtual machine; head should be main node.+allNodes :: CommM [NodeId]+allNodes = s_allNodes <$> ask++-- The currently executing node.+myNode :: CommM NodeId+myNode = s_myNode <$> ask++-- True iff the currently executing node is the main node.+isMain :: CommM Bool+isMain = s_isMain <$> ask++-- internal use only: queue of received payload messages+msgQ :: CommM MessageQ+msgQ = s_msgQ <$> ask++-- |Internal use only: debug level+debug :: CommM Int+debug = debugLvl <$> s_conf <$> ask++-- |Block until receiving a shutdown signal.+waitShutdown :: CommM ()+waitShutdown = do+  mvar <- s_shutdown <$> ask+  liftIO $ takeMVar mvar+  shutdownTransport++-- |Called from 'waitShutdown', this closes+--  all connections, the local endpoint, and+--  then the transport layer, in that order+shutdownTransport :: CommM ()+shutdownTransport = do+{-+  -- Kill all connections+  liftIO $ killConnections <$> connectionLookup+  -- Kill my endpoint+  liftIO $ NT.closeEndPoint <$> myEndPoint+  -- Kill the transport layer+  trans <- liftIO lclTransport+  liftIO $ NT.closeTransport trans+-}+  liftIO shutdownTransportIO++-- |Used in an unclean shutdown when+--  the connection with the master node+--  has been unexpectedly closed.+shutdownTransportIO :: IO ()+shutdownTransportIO = do+  -- Kill all connections+--  remoteConnections <- connectionLookup+--  killConnections remoteConnections+  killConnections =<< connectionLookup+  -- Kill my endpoint+--  myEP <- myEndPoint+--  NT.closeEndPoint myEP+  NT.closeEndPoint =<< myEndPoint+  -- Kill the transport layer+--  trans <- lclTransport+--  NT.closeTransport trans+  NT.closeTransport =<< lclTransport++-- |Used during shutdown, to close+--  connections with all processes safely+killConnections :: Map.Map NodeId NT.Connection -> IO ()+killConnections remoteConnections = do+  let nodes = Map.keys remoteConnections+  mapM_ killConn nodes+  where+   killConn node = do+    let remoteConnection = fromJust $ Map.lookup node remoteConnections+    NT.close remoteConnection++-- |Initiate a shutdown from master process+shutdown :: CommM ()+shutdown = do+  targets <- allNodes+  -- broadcast Shutdown message to all nodes but sender+  liftIO $ broadcastMsg targets Shutdown++-- |Connection with the master process has been closed+--  unexpectedly, close transport layer+uncleanShutdown :: IO ()+uncleanShutdown = do+#ifdef HDPH_DEBUG+  dbg "Shutting down as main process died"+#endif+  -- The main process has terminated. Let's clean up.+  shutdownTransportIO+  exitWith (ExitFailure 9) -- force process termination++-----------------------------------------------------------------------------+-- running the CommM monad++-- Run the given 'CommM' action in the IO monad with the given config data+-- (which determines the debug level, see module HdpH.Internal.Location).+-- The 'action' must call 'waitShutdown' before it terminates on all nodes,+-- and at least one node must call 'shutdown'.+run_ :: RTSConf -> CommM () -> IO ()+run_ conf action = do++#ifdef HDPH_DEBUG+  dbg "run_.1"+#endif++  -- check debug level+  let debugLevel = debugLvl conf+  unless (debugLevel >= dbgNone) $+    Prelude.error "HdpH.Internal.Comm_MPI.run_: debug level < none"++  -- set debug level in HdpH.Internal.State.Location+  writeIORef debugRef debugLevel++#ifdef HDPH_DEBUG+  dbg "run_.2"+#endif+++#ifdef HDPH_DEBUG+  dbg "run_.3"+#endif++  myIP <- discoverMyIP conf -- uses network-info for local IP address identification++  -- Networking step 1: Create a transport+  transport <- tryCreateTransport myIP conf+  atomicModifyIORef lclTransportRef (\r -> (transport,r))++  -- Networking step 2: Create a local endpoint; write to myEndPointRef IORef+  Right myEP <- NT.newEndPoint transport+  let me = NT.address myEP+  -- Sets 'myEndPointRef'. Receive events from endpoint in transport layer+  atomicModifyIORef myEndPointRef (\r -> (myEP,r))+  (allNodes,main) <- nodeInfo conf+  let iAmMain = me == main++  -- set node ID in HdpH.Internal.State.Location+  atomicModifyIORef myNodeRef (\r -> (me,r))++#ifdef HDPH_DEBUG+  dbg "run_.4"+#endif++#ifdef HDPH_DEBUG+  dbg "run_.5"+#endif++  -- create initial state+  q <- newChan+  startBarrier <- newEmptyMVar -- (not used in TCP backend)+  stopBarrier  <- newEmptyMVar++#ifdef HDPH_DEBUG+  dbg "run_.6"+#endif++#ifdef HDPH_DEBUG+  dbg $ "run_.7 receiveServerTid = "+#endif++  if iAmMain+         then do+#ifdef HDPH_DEBUG+                 dbg "run_.7.root"+#endif+                 -- Networking step 4: Create NodeId's for all nodes+                 -- including endpoint address info. Also, create+                 -- connections between local endpoint and endpoint addresses+                 -- of all nodes. Write connections map to connectionlookupRef+                 -- and all NodeId's to allNodesRef+                 nodeConnections <- remoteEndPointAddrMap allNodes+                 atomicModifyIORef connectionLookupRef (\r -> (nodeConnections,r))+                 +                 recvAllReady (length allNodes - 1 ) -- blocking+                 broadcastMsg (allNodes \\ [me]) Booted+                 atomicModifyIORef mainEndpointAddrRef (const (myEP, ()))++         else do+#ifdef HDPH_DEBUG+                 dbg "run_.7.other"+#endif+                 -- See networking step 4, above.+                 let mainEP = main+                 nodeConnections <- remoteEndPointAddrMap allNodes+                 atomicModifyIORef connectionLookupRef (const (nodeConnections, ()))+                 atomicModifyIORef mainEndpointAddrRef (const (mainEP, ()))++                 -- Tells master that this node is ready+                 broadcastMsg [main] Ready+                 +                 -- waits for `Booted', means that main is connected to all nodes+                 waitForBootstrapConfirmation++  let s0 = State { s_conf     = conf,+                   s_nodes    = length allNodes,+                   s_allNodes = allNodes,+                   s_myNode   = me,+                   s_isMain   = iAmMain,+                   s_msgQ     = q,+                   s_shutdown = stopBarrier }++#ifdef HDPH_DEBUG+  dbg "run_.8"+#endif+  forkIO $ receiveServer q startBarrier stopBarrier+#ifdef HDPH_DEBUG+  dbg "run_.8b"+#endif+  -- run monad+  runReaderT action s0++#ifdef HDPH_DEBUG+  dbg "run_.9"+#endif++  -- reset HdpH.Internal.State.Location+  atomicModifyIORef myNodeRef (\r -> (throw NodeIdUnset,r))+  writeIORef debugRef dbgNone+  +#ifdef HDPH_DEBUG+  dbg "run_.10"+#endif++-----------------------------------------------------------------------------+-- internal messages++data Msg = Startup          -- startup completed message (main -> other)+         | Shutdown         -- shutdown system message (main -> other)+         | Booted+         | Ready+         | Payload Message  -- non-system message; arg (payload) to be queued+           deriving (Eq, Ord, Show)  -- Show inst only for debugging+++instance NFData Msg where+  rnf Startup         = ()+  rnf (Booted)        = ()+  rnf (Ready)         = ()+  rnf (Shutdown)      = ()+  rnf (Payload work)     = rnf work+++instance Data.Serialize.Serialize Msg where+  put Startup         = Data.Serialize.put (0 :: Word8)+  put (Booted)        = Data.Serialize.put (1 :: Word8)+  put (Ready)         = Data.Serialize.put (2 :: Word8)+  put (Shutdown)      = Data.Serialize.put (3 :: Word8)+  put (Payload work)  = Data.Serialize.put (4 :: Word8) >>+                        Data.Serialize.put work++  get = do tag <- Data.Serialize.get+           case tag :: Word8 of+             0 -> do return $ Startup+             1 -> do return $ Booted+             2 -> do return $ Ready+             3 -> do return $ Shutdown+             4 -> do work  <- Data.Serialize.get+                     return $ Payload work+++-----------------------------------------------------------------------------+-- sending messages (incl system messages)++-- Send a payload message.+send :: NodeId -> Message -> CommM ()+send dest message = lift $ send_ dest message++send_ :: NodeId -> Message -> IO ()+send_ dest message = do+  result <- try $ do+   remoteConnections <- connectionLookup+   let conn = fromJust $ Map.lookup dest remoteConnections+   -- Actual type of 'send' is Either (NT.FailedWith NT.SendErrorCode) ()+   NT.send conn (Lazy.toChunks (encodeLazy (Payload message)))+  case result of+   Left (e::SomeException) -> void (print e)+   Right _ -> return ()+++-- This needs to be separate because in `send_' (Payload _)+-- constructor is automatically added.+broadcastMsg :: [NodeId] -> Msg -> IO ()+broadcastMsg dests msg =+  mapM_ broadcastMsg' dests+   where+    serialized_msg = encodeLazy msg+    broadcastMsg' dest = do+     result <- try $ do+      remoteConnections <- connectionLookup+      let conn = fromJust $ Map.lookup dest remoteConnections+      -- Actual type of 'send' is Either (NT.FailedWith NT.SendErrorCode) ()+      _ <- NT.send conn (Lazy.toChunks serialized_msg)+      return ()+     case result of+      Left (e::SomeException) -> void (print e)+      Right _ -> return ()+++-----------------------------------------------------------------------------+-- receiving messages (incl system messages and message queue server)++-- Block to receive a message;+-- the sender must be encoded into the message, otherwise it is unknown.+receive :: CommM Message+receive = do q <- msgQ+             liftIO $ readChan q+++recv :: IO Msg+recv = do+     ep <- myEndPoint+     event <- NT.receive ep+     case event of+       NT.Received _ msg ->  return ((force . decodeLazy . Lazy.fromChunks) msg)+       NT.ErrorEvent (NT.TransportError e _) ->+         case e of+           NT.EventConnectionLost ep -> do+               mainEP <- mainEndpointAddr+               -- Let's check if the main node has died.+               -- If it has, we should give up.+               if mainEP == ep then do+                 -- Main process has terminated prematurely. Fatal.+                 uncleanShutdown+                 return Shutdown+                else do+ {- Enabled in ft-scheduler branch+                 -- Send a message to scheduler+                 remoteConnections <- connectionLookup+                 let x = Map.filterWithKey (\node _ -> ep == node) remoteConnections+                     deadNode = head $ Map.keys x -- should only be one+                     msg = Payload $ encodeLazy (Payload.DEADNODE deadNode)+                 return msg+-}+                 recv+           _ -> recv  -- links to "case e of"+       _ -> do        -- links to "case event of"+          -- ignore remaining NT.Event constructors for now+          -- i.e. [ConnectionClosed,ConnectionOpened,ReceivedMulticast,EndPointClosed]+          recv++-- Non-terminating computation, to be run in a separate thread.+-- Continually receives message, which it puts into the given+-- message queue or handles immediately (in the case of system messages).+-- * 'Shutdown' unblocks the shutdown barrier (thus terminating all actions).+receiveServer :: MessageQ -> MVar () -> MVar () -> IO ()+receiveServer q startBarrier stopBarrier = do+  hdl <- recv+  handleMsg hdl  -- NOTE: Changed from previous 'forkIO $ handleMsg hdl'+    -- Rationale: 'handleMsg' should be sufficiently lazy to run sequentially',+    -- plus it poses less danger of corruption on shutdown+  where +    handleMsg hdl =+      -- receive message and dispatch on constructor+      case hdl of+        -- 'Startup' not used in TCP backend            +        Startup ->        receiveServer q startBarrier stopBarrier+        Shutdown ->        -- lift shutdown barrier+                          putMVar stopBarrier ()+        Payload message -> -- queue the payload+                          do  writeChan q message+                              receiveServer q startBarrier stopBarrier+        _ -> error $ "Unexpected message in `receiveServer' " ++ show hdl+++-----------------------------------------------------------------------------+-- debugging++#ifdef HDPH_DEBUG+dbg :: String -> IO ()+dbg s = do+  hPutStrLn stderr $ ": HdpH.Internal.Comm_TCP." ++ s+#endif++-- |Used to setup and store a Map of NodeId -> NT.Connection+--  And also, creates a list of [NodeId] that is written+--  to the allNodesRef IORef+remoteEndPointAddrMap :: [NodeId] -> IO (Map.Map NodeId NT.Connection)+remoteEndPointAddrMap nodes = do+  mvar <- newMVar Map.empty -- to store connections+  mapM_ (connectToAllNodes mvar) nodes+  takeMVar mvar++------+-- Start up utilities++-- |Before the 'receiveServer' function is forked,+--  each node must receive a Booted payload from the master process.+--  This indicates that all nodes have sent an ALIVE payload to the master process.+waitForBootstrapConfirmation ::  IO ()+waitForBootstrapConfirmation = do+  msg <- recv+  case msg of+   Booted -> return ()+   _ -> waitForBootstrapConfirmation -- Hangover from UDP broadcasts, still wait for `Booted'++-- |Writing into an MVar the connection that has been+-- made with the remote node, to be written into+-- the connectionLookupRef IORef+connectToAllNodes :: MVar (Map.Map NodeId NT.Connection) -> NodeId -> IO ()+connectToAllNodes mvar remoteNode = do+   myEP <- myEndPoint+   x <- NT.connect myEP remoteNode NT.ReliableOrdered NT.defaultConnectHints+   case x of+    (Right newConnection) ->+      modifyMVar_ mvar $ \m ->+       return $ Map.insert remoteNode newConnection m+    (Left _) ->  connectToAllNodes mvar remoteNode -- keep retrying++-- |'action' is only executed once the master node+--  has receve NT.ConnectionOpened from all other nodes+recvAllReady :: Int -> IO ()+recvAllReady i =+  when (i > 0) $ do+   msg <- recv+   case msg of+     Ready -> recvAllReady (i-1)+     _ -> putStrLn $ "unexpected msg in recvAllReady: " ++ show msg+++---------------------------+-- Transport layer creation++tryCreateTransport :: IPv4 -> RTSConf -> IO NT.Transport+tryCreateTransport myIP conf =+  createTrans myIP (numProcs conf) 0++createTrans :: IPv4 -> Int -> Int -> IO NT.Transport+createTrans myIP tasks attempts = do+  rndsock <- genRandomSocket+  t <- TCP.createTransport (show myIP) (show rndsock) TCP.defaultTCPParameters+  case t of+   Right transport -> return transport+   Left e -> do+    let attempts' = attempts+1+    if attempts' == tasks then error ("Error creating transport: " ++ show e)+     else do +      createTrans myIP tasks attempts'++-----------------------+-- UDP based Node discovery++nodeInfo :: RTSConf ->  IO (SlaveNodes, MainNode)+nodeInfo conf = do+  _ <- forkIO $ broadcastTimeout 10000000+  all <- findSlaves (numProcs conf)+  let mainNode = head (sort all) -- election protocol+  return (all,mainNode)+   where+    broadcastTimeout i = do+     -- broadcast endpoint address via UDP for 10 seconds+     _ <- timeout i broadcastMyNode+     return ()++discoverMyIP :: RTSConf -> IO IPv4+discoverMyIP conf = do+  ns <- getNetworkInterfaces+  return $ myIP ns (networkInterface conf)++myIP :: [NetworkInterface] -> String -> IPv4+myIP interfaces interfaceName =+  let eth = filter (\x -> name x == interfaceName) interfaces+  in ipv4 $ head eth++type MainNode = NodeId+type SlaveNodes = [NodeId]++broadcastMyNode :: IO ()+broadcastMyNode = do+  myEP <- myEndPoint+  forever $ do+   (sock, addr) <- multicastSender "224.0.0.99" 9999+   sendTo sock (NT.endPointAddressToByteString (NT.address myEP)) addr+   threadDelay 100000++findSlaves :: Int -> IO SlaveNodes+findSlaves numNodesExpected = do+  sock <- multicastReceiver "224.0.0.99" 9999+  listenForNodes sock [] numNodesExpected+  +listenForNodes :: Socket -> SlaveNodes -> Int -> IO SlaveNodes+listenForNodes sock ns expected = do+    (msg, _) <- recvFrom sock 1024+    let remoteEndPointAddr = NT.EndPointAddress msg+    let n = if remoteEndPointAddr `elem` ns then [] else [remoteEndPointAddr]+        ns' = n ++ ns+    if length ns' == expected then return ns'+     else listenForNodes sock ns' expected++genRandomSocket :: IO Int+genRandomSocket = do+  gen <- newStdGen+  return $ head (randomRs (8000,40000) gen)++------+-- Enpoint and connection lookup IORefs++myEndPoint :: IO NT.EndPoint+myEndPoint = readIORef myEndPointRef++myEndPointRef :: IORef NT.EndPoint+myEndPointRef = unsafePerformIO $ newIORef $ throw NodeIdUnset+{-# NOINLINE myEndPointRef #-}    -- required to protect unsafePerformIO hack++connectionLookup :: IO (Map.Map NodeId NT.Connection)+connectionLookup = readIORef connectionLookupRef++connectionLookupRef :: forall k a. IORef (Map.Map k a)+connectionLookupRef = unsafePerformIO $ newIORef Map.empty+{-# NOINLINE connectionLookupRef #-}    -- required to protect unsafePerformIO hack++-- Used to watch when main node has failed.+-- If main node fails, shutdown transport layer and terminate.+mainEndpointAddr :: forall a. IO a+mainEndpointAddr = readIORef mainEndpointAddrRef++mainEndpointAddrRef :: forall a. IORef a+mainEndpointAddrRef = unsafePerformIO $ newIORef $ throw NodeIdUnset+{-# NOINLINE mainEndpointAddrRef #-}    -- required to protect unsafePerformIO hack++lclTransport :: IO NT.Transport+lclTransport = readIORef lclTransportRef++lclTransportRef :: IORef NT.Transport+lclTransportRef = unsafePerformIO $ newIORef $ throw NodeIdUnset+{-# NOINLINE lclTransportRef #-}    -- required to protect unsafePerformIO hack
+ src/Control/Parallel/HdpH/Internal/Data/Deque.hs view
@@ -0,0 +1,141 @@+-- Doubly-ended queue (deque) with size and max size+--+-- Author: Patrick Maier+-------------------------------------------------------------------------------++module Control.Parallel.HdpH.Internal.Data.Deque+  ( -- * functional deque+    Deque,        -- no instances+    empty,        -- :: Deque a+    fromList,     -- :: [a] -> Deque a+    pushFront,    -- :: Deque a -> a -> Deque a+    pushBack,     -- :: Deque a -> a -> Deque a+    popFront,     -- :: Deque a -> (Maybe a, Deque a)+    popBack,      -- :: Deque a -> (Maybe a, Deque a)+    first,        -- :: Deque a -> Maybe a+    last,         -- :: Deque a -> Maybe a+    null,         -- :: Deque a -> Bool+    length,       -- :: Deque a -> Int+    maxLength,    -- :: Deque a -> Int++    -- * stateful, concurrently accessible deque+    DequeIO,      -- no instances+    emptyIO,      -- :: IO (DequeIO a)+    fromListIO,   -- :: [a] -> IO (DequeIO a)+    pushFrontIO,  -- :: DequeIO a -> a -> IO ()+    pushBackIO,   -- :: DequeIO a -> a -> IO ()+    popFrontIO,   -- :: DequeIO a -> IO (Maybe a)+    popBackIO,    -- :: DequeIO a -> IO (Maybe a)+    firstIO,      -- :: DequeIO a -> IO (Maybe a)+    lastIO,       -- :: DequeIO a -> IO (Maybe a)+    nullIO,       -- :: DequeIO a -> IO Bool+    lengthIO,     -- :: DequeIO a -> IO Int+    maxLengthIO   -- :: DequeIO a -> IO Int+  ) where++import Prelude hiding (error, last, length, null)+import Data.Functor ((<$>))+import Data.IORef (IORef, newIORef, readIORef, atomicModifyIORef)+import qualified Data.List as List (length)+import Data.Sequence (Seq, (|>), (<|), ViewR((:>)), viewr, ViewL((:<)), viewl)+import qualified Data.Sequence as Seq (empty, null, length, fromList)+++-----------------------------------------------------------------------------+-- functional deque with size and max size (with amortised O(1) operations)++data Deque a = Deque { q  :: Seq a,  -- sequence of elements+                       mx :: !Int }  -- maximal length of above sequence++empty :: Deque a+empty = Deque { q = Seq.empty, mx = 0 }++fromList :: [a] -> Deque a+fromList xs = Deque { q = Seq.fromList xs, mx = List.length xs }++pushFront :: Deque a -> a -> Deque a+pushFront dq x = dq { q = x <| q dq, mx = max (length dq + 1) (maxLength dq) }++pushBack :: Deque a -> a -> Deque a+pushBack dq x = dq { q = q dq |> x, mx = max (length dq + 1) (maxLength dq) }++popFront :: Deque a -> (Maybe a, Deque a)+popFront dq = case viewl (q dq) of+                hd :< rest -> (Just hd, dq { q = rest })+                _          -> (Nothing, dq)++popBack :: Deque a -> (Maybe a, Deque a)+popBack dq = case viewr (q dq) of+                rest :> tl -> (Just tl, dq { q = rest })+                _          -> (Nothing, dq)++first :: Deque a -> Maybe a+first dq = case viewl (q dq) of+             x :< _ -> Just x+             _      -> Nothing++last :: Deque a -> Maybe a+last dq = case viewr (q dq) of+            _ :> x -> Just x+            _      -> Nothing++null :: Deque a -> Bool+null = Seq.null . q++length :: Deque a -> Int+length = Seq.length . q++maxLength :: Deque a -> Int+maxLength = mx+++-----------------------------------------------------------------------------+-- concurrently accessible deque (in the IO monad) with size and max size;+-- concurrent access is via a global lock on the deque.++newtype DequeIO a = DequeIO (IORef (Deque a))++emptyIO :: IO (DequeIO a)+emptyIO = DequeIO <$> newIORef empty++fromListIO :: [a] -> IO (DequeIO a)+fromListIO xs = DequeIO <$> newIORef (fromList xs)++pushFrontIO :: DequeIO a -> a -> IO ()+pushFrontIO (DequeIO dqRef) x =+  atomicModifyIORef dqRef $ \ dq -> (pushFront dq x, ())++pushBackIO :: DequeIO a -> a -> IO ()+pushBackIO (DequeIO dqRef) x =+  atomicModifyIORef dqRef $ \ dq -> (pushBack dq x, ())++popFrontIO :: DequeIO a -> IO (Maybe a)+popFrontIO (DequeIO dqRef) =+  atomicModifyIORef dqRef $ swap . popFront++popBackIO :: DequeIO a -> IO (Maybe a)+popBackIO (DequeIO dqRef) =+  atomicModifyIORef dqRef $ swap . popBack++firstIO :: DequeIO a -> IO (Maybe a)+firstIO (DequeIO dqRef) = first <$> readIORef dqRef++lastIO :: DequeIO a -> IO (Maybe a)+lastIO (DequeIO dqRef) = last <$> readIORef dqRef++nullIO :: DequeIO a -> IO Bool+nullIO (DequeIO dqRef) = null <$> readIORef dqRef++lengthIO :: DequeIO a -> IO Int+lengthIO (DequeIO dqRef) = length <$> readIORef dqRef++maxLengthIO :: DequeIO a -> IO Int+maxLengthIO (DequeIO dqRef) = maxLength <$> readIORef dqRef+++-----------------------------------------------------------------------------+-- auxiliary functions++-- NOTE: Should be exported by Data.Tuple.+swap :: (a,b) -> (b,a)+swap (a,b) = (b,a)
+ src/Control/Parallel/HdpH/Internal/Data/Sem.hs view
@@ -0,0 +1,70 @@+-- Simple racey semaphores without quantity and without starvation prevention+--+-- Author: Patrick Maier+-------------------------------------------------------------------------------++module Control.Parallel.HdpH.Internal.Data.Sem +  ( -- * semaphore type+    Sem,     -- no instances++    -- * basic operations+    new,     -- :: IO Sem    +    wait,    -- :: Sem -> IO ()+    signal,  -- :: Sem -> IO ()++    -- * convenience+    signalPeriodically  -- :: Sem -> Int -> IO ()+  ) where++import Prelude hiding (error)+import Control.Concurrent (threadDelay)+import Control.Concurrent.MVar (MVar, newEmptyMVar, takeMVar, putMVar)+import Control.Monad (join)+import Data.Functor ((<$>))+import Data.IORef (IORef, newIORef, readIORef, atomicModifyIORef)+++-------------------------------------------------------------------------------+-- A simple semaphore just maintains a list of threads being blocked on it,+-- and provides operations for waiting and signalling (blocking and wakeing+-- up one thread per operation, respectively).++newtype Sem = Sem (IORef [MVar ()])+++-- Creates a new semaphore.+new :: IO Sem+new = Sem <$> newIORef []+++-- Blocks calling thread, waiting for a 'signal' on the given semaphore.+-- Note that there can be races between 'wait' and 'signal' in that 'signal'+-- can find no blocked threads (and thus do nothing) just before another+-- thread is about to 'wait'. This effect is due to the semaphore not+-- actually guarding a quantity.+wait :: Sem -> IO ()+wait (Sem sem) = do+  b <- newEmptyMVar+  atomicModifyIORef sem $ \ blocked -> (b:blocked, ())+  takeMVar b+++-- Wakes up one blocked thread (actually, the last thread to block) if there+-- are any blocked threads.+signal :: Sem -> IO ()+signal (Sem sem) =+  join $+    atomicModifyIORef sem $ \ blocked ->+      case blocked of+        []        -> ([],      return ())+        b:blocked -> (blocked, putMVar b ())+++-- Nonterminating action periodically, every 'interval' microseconds,+-- signalling the given semaphore. This is one way to deal with the races+-- that may arise between 'wait' and 'signal'.+signalPeriodically :: Sem -> Int -> IO ()+signalPeriodically sem interval = do+  threadDelay interval+  signal sem+  signalPeriodically sem interval
+ src/Control/Parallel/HdpH/Internal/GRef.hs view
@@ -0,0 +1,231 @@+-- Global references+--+-- Author: Patrick Maier+-----------------------------------------------------------------------------++{-# LANGUAGE StandaloneDeriving #-}++module Control.Parallel.HdpH.Internal.GRef+  ( -- * global references+    GRef,       -- instances: Eq, Ord, Show, NFData, Serialize+    at,         -- :: GRef a -> NodeId++    -- * predicates on global references+    isLocal,    -- :: GRef a -> IO Bool+    isLive,     -- :: GRef a -> IO Bool++    -- * updating the registry+    globalise,  -- :: a -> IO (GRef a)+    free,       -- :: GRef a -> IO ()+    freeNow,    -- :: GRef a -> IO ()++    -- * dereferencing a global reference+    withGRef    -- :: GRef a -> (a -> IO b) -> IO b -> IO b+  ) where++import Prelude hiding (error)+import Control.Concurrent (forkIO)+import Control.DeepSeq (NFData(rnf))+import Control.Monad (unless)+import Data.Functor ((<$>))+import Data.IORef (readIORef, atomicModifyIORef)+import qualified Data.Map as Map (insert, delete, member, lookup)+import Data.Serialize (Serialize)+import qualified Data.Serialize (put, get)+import Unsafe.Coerce (unsafeCoerce)++import Control.Parallel.HdpH.Internal.Location+       (NodeId, myNode, error, debug, dbgGRef)+import Control.Parallel.HdpH.Internal.Misc (AnyType(Any))+import Control.Parallel.HdpH.Internal.Type.GRef+       (GRef(GRef), at, slot, GRefReg, lastSlot, table)+import Control.Parallel.HdpH.Internal.State.GRef (regRef)+++-----------------------------------------------------------------------------+-- Key facts about global references+--+-- * A global reference is a globally unique handle naming a Haskell value;+--   the type of the value is reflected in a phantom type argument to the+--   type of global reference, similar to the type of stable names.+--+-- * The link between a global reference and the value it names is established+--   by a registry mapping references to values. The registry mapping a+--   global reference resides on the node hosting its value. All operations+--   involving the reference must be executed on the hosting node; the only+--   exception is the function 'at', projecting a global reference to its+--   hosting node.+--+-- * The life time of a global reference is not linked to the life time of +--   the named value, and vice versa. One consequence is that global+--   references can never be re-used, unlike stable names.+--+-- * For now, global references must be freed explicitly (from the map+--   on the hosting node). This could (and should) be changed by using+--   weak pointers and finalizers.+++-----------------------------------------------------------------------------+-- global references (abstract outwith this module)+-- NOTE: Global references are hyperstrict.++-- Constructs a 'GRef' value of a given node ID and slot (on the given node);+-- ensures the resulting 'GRef' value is hyperstrict;+-- this constructor is not to be exported.+mkGRef :: NodeId -> Integer -> GRef a+mkGRef node i = rnf node `seq` rnf i `seq` GRef { slot = i, at = node }++instance Eq (GRef a) where+  ref1 == ref2 = slot ref1 == slot ref2 && at ref1 == at ref2++instance Ord (GRef a) where+  compare ref1 ref2 = case compare (slot ref1) (slot ref2) of+                        LT -> LT+                        GT -> GT+                        EQ -> compare (at ref1) (at ref2)+++-- Show instance (mainly for debugging)+instance Show (GRef a) where+  showsPrec _ ref =+    showString "GRef:" . shows (at ref) . showString "." . shows (slot ref)++instance NFData (GRef a)  -- default instance suffices (due to hyperstrictness)++instance Serialize (GRef a) where+  put ref = Data.Serialize.put (at ref) >>+            Data.Serialize.put (slot ref)+  get = do node <- Data.Serialize.get+           i <- Data.Serialize.get+           return $ mkGRef node i  -- 'mkGRef' ensures result is hyperstrict+++-----------------------------------------------------------------------------+-- predicates on global references++-- Monadic projection; True iff the current node hosts the object refered+-- to by the given global 'ref'.+isLocal :: GRef a -> IO Bool+isLocal ref = (at ref ==) <$> myNode+++-- Checks if a locally hosted global 'ref' is live.+-- Aborts with an error if 'ref' is a not hosted locally.+isLive :: GRef a -> IO Bool+isLive ref = do+  refIsLocal <- isLocal ref+  unless refIsLocal $+    error $ "HdpH.Internal.GRef.isLive: " ++ show ref ++ " not local"+  reg <- readIORef regRef+  return $ Map.member (slot ref) (table reg)+++-----------------------------------------------------------------------------+-- updating the registry++-- Registers its argument as a global object (hosted on the current node),+-- returning a fresh global reference. May block when attempting to access+-- the registry.+globalise :: a -> IO (GRef a)+globalise x = do+  node <- myNode+  ref <- atomicModifyIORef regRef (createEntry (Any x) node)+  debug dbgGRef $ "GRef.globalise " ++ show ref+  return ref+++-- Asynchronously frees a locally hosted global 'ref'; no-op if 'ref' is dead.+-- Aborts with an error if 'ref' is a not hosted locally.+free :: GRef a -> IO ()+free ref = do+  refIsLocal <- isLocal ref+  unless refIsLocal $+    error $ "HdpH.Internal.GRef.free: " ++ show ref ++ " not local"+  forkIO $ do debug dbgGRef $ "GRef.free " ++ show ref+              atomicModifyIORef regRef (deleteEntry $ slot ref)+  return ()+++-- Frees a locally hosted global 'ref'; no-op if 'ref' is dead.+-- Aborts with an error if 'ref' is a not hosted locally.+freeNow :: GRef a -> IO ()+freeNow ref = do+  refIsLocal <- isLocal ref+  unless refIsLocal $+    error $ "HdpH.Internal.GRef.freeNow: " ++ show ref ++ " not local"+  debug dbgGRef $ "GRef.freeNow " ++ show ref+  atomicModifyIORef regRef (deleteEntry $ slot ref)+++-- Create new entry in 'reg' (hosted on 'node') mapping to 'val'; not exported+createEntry :: AnyType -> NodeId -> GRefReg -> (GRefReg, GRef a)+createEntry val node reg =+  ref `seq` (reg', ref) where+    newSlot = lastSlot reg + 1+    ref = mkGRef node newSlot  -- 'seq' above forces hyperstrict 'ref' to NF+    reg' = reg { lastSlot = newSlot,+                 table    = Map.insert newSlot val (table reg) }+++-- Delete entry 'slot' from 'reg'; not exported+deleteEntry :: Integer -> GRefReg -> (GRefReg, ())+deleteEntry slot reg =+  (reg { table = Map.delete slot (table reg) }, ())+++-----------------------------------------------------------------------------+-- Dereferencing global refs++-- Attempts to dereference a locally hosted global 'ref' and apply 'action'+-- to the refered-to object; executes 'dead' if that is not possible (ie.+-- 'dead' acts as an exception handler) because the global 'ref' is dead.+-- Aborts with an error if 'ref' is a not hosted locally.+withGRef :: GRef a -> (a -> IO b) -> IO b -> IO b+withGRef ref action dead = do+  refIsLocal <- isLocal ref+  unless refIsLocal $+    error $ "HdpH.Internal.GRef.withGRef: " ++ show ref ++ " not local"+  reg <- readIORef regRef+  case Map.lookup (slot ref) (table reg) of+    Nothing      -> do debug dbgGRef $ "GRef.withGRef " ++ show ref ++ " dead"+                       dead+    Just (Any x) -> do action (unsafeCoerce x)+                    -- see below for an argument why unsafeCoerce is safe here+++-------------------------------------------------------------------------------+-- Notes on the design of the registry+--+-- * A global reference is represented as a pair consisting of the ID+--   of the hosting node together with its 'slot' in the registry on+--   that node. The slot is an unbounded integer so that there is an+--   infinite supply of slots. (Slots can't be re-used as there is no+--   global garbage collection of global references.)+--+-- * The registry maps slots to values, which are essentially untyped+--   (the type information being swallowed by an existential wrapper).+--   However, the value type information is not lost as it can be+--   recovered from the phantom type argument of its global reference.+--   In fact, the function 'withGRef' super-imposes a reference's phantom+--   type on to its value via 'unsafeCoerce'. The reasons why this is safe+--   are laid out below.+++-- Why 'unsafeCoerce' in safe in 'withGRef':+--+-- * Global references can only be created by the function 'globalise'.+--   Whenever this function generates a global reference 'ref' of type+--   'GRef t' it guarantees that 'ref' is globally fresh, ie. its+--   representation does not exist any where else in the system, nor has+--   it ever existed in the past. (Note that freshness relies on the+--   assumption that node IDs themselves are fresh, which is relevant+--   in case nodes can leave and join dynmically.)+--+-- * A consequence of global freshness is that there is a functional relation+--   from representations to phantom types of global references. For all+--   global references 'ref1 :: GRef t1' and 'ref2 :: GRef t2',+--   'at ref1 == at ref2 && slot ref1 == slot ref2' implies the identity+--   of the phantom types t1 and t2.+--+-- * Thus, we can safely super-impose (using 'unsafeCoerce') the phantom type+--   of a global reference on to its value.
+ src/Control/Parallel/HdpH/Internal/IVar.hs view
@@ -0,0 +1,157 @@+-- Local and global IVars+--+-- Author: Patrick Maier+-----------------------------------------------------------------------------++module Control.Parallel.HdpH.Internal.IVar+  ( -- * local IVar type+    IVar,       -- synonym: IVar m a = IORef <IVarContent m a>++    -- * operations on local IVars+    newIVar,    -- :: IO (IVar m a)+    putIVar,    -- :: IVar m a -> a -> IO [Thread m]+    getIVar,    -- :: IVar m a -> (a -> Thread m) -> IO (Maybe a)+    pollIVar,   -- :: IVar m a -> IO (Maybe a)+    probeIVar,  -- :: IVar m a -> IO Bool++    -- * global IVar type+    GIVar,      -- synonym: GIVar m a = GRef (IVar m a)++    -- * operations on global IVars+    globIVar,   -- :: Int -> IVar m a -> IO (GIVar m a)+    hostGIVar,  -- :: GIVar m a -> NodeId+    putGIVar    -- :: Int -> GIVar m a -> a -> IO [Thread m]+  ) where++import Prelude hiding (error)+import Data.Functor ((<$>))+import Data.IORef (IORef, newIORef, readIORef, atomicModifyIORef)+import Data.Maybe (isJust)++import Control.Parallel.HdpH.Internal.Location +       (NodeId, debug, dbgGIVar, dbgIVar)+import Control.Parallel.HdpH.Internal.GRef (GRef, at, globalise, free, withGRef)+import Control.Parallel.HdpH.Internal.Type.Par (Thread)+++-----------------------------------------------------------------------------+-- type of local IVars++-- An IVar is a mutable reference to either a value or a list of blocked+-- continuations (waiting for a value);+-- the parameter 'm' abstracts a monad (cf. module HdpH.Internal.Type.Par).+type IVar m a = IORef (IVarContent m a)++data IVarContent m a = Full a+                     | Blocked [a -> Thread m]+++-----------------------------------------------------------------------------+-- operations on local IVars, borrowing from+--    [1] Marlow et al. "A monad for deterministic parallelism". Haskell 2011.++-- Create a new, empty IVar.+newIVar :: IO (IVar m a)+newIVar = newIORef (Blocked [])+++-- Write 'x' to the IVar 'v' and return the list of blocked threads.+-- Unlike [1], multiple writes fail silently (ie. they do not change+-- the value stored, and return an empty list of threads).+putIVar :: IVar m a -> a -> IO [Thread m]+putIVar v x = do+  e <- readIORef v+  case e of+    Full _    -> do debug dbgIVar $ "Put to full IVar"+                    return []+    Blocked _ -> do maybe_ts <- atomicModifyIORef v fill_and_unblock+                    case maybe_ts of+                      Nothing -> do debug dbgIVar $ "Put to full IVar"+                                    return []+                      Just ts -> do debug dbgIVar $+                                      "Put to empty IVar; unblocking " +++                                      show (length ts) ++ " threads"+                                    return ts+      where+     -- fill_and_unblock :: IVarContent m a ->+     --                       (IVarContent m a, Maybe [Thread m])+        fill_and_unblock e =+          case e of+            Full _     -> (e,       Nothing)+            Blocked cs -> (Full x,  Just $ map ($ x) cs)+++-- Read from the given IVar 'v' and return the value if it is full.+-- Otherwise add the given continuation 'c' to the list of blocked+-- continuations and return nothing.+getIVar :: IVar m a -> (a -> Thread m) -> IO (Maybe a)+getIVar v c = do+  e <- readIORef v+  case e of+    Full x    -> do return (Just x)+    Blocked _ -> do maybe_x <- atomicModifyIORef v get_or_block+                    case maybe_x of+                      Just _  -> do return maybe_x+                      Nothing -> do debug dbgIVar $ "Blocking on IVar"+                                    return maybe_x+      where+     -- get_or_block :: IVarContent m a -> (IVarContent m a, Maybe a)+        get_or_block e =+          case e of+            Full x     -> (e,              Just x)+            Blocked cs -> (Blocked (c:cs), Nothing)+++-- Poll the given IVar 'v' and return its value if full, Nothing otherwise.+-- Does not block.+pollIVar :: IVar m a -> IO (Maybe a)+pollIVar v = do+  e <- readIORef v+  case e of+    Full x    -> return (Just x)+    Blocked _ -> return Nothing+++-- Probe whether the given IVar is full, returning True if it is.+-- Does not block.+probeIVar :: IVar m a -> IO Bool+probeIVar v = isJust <$> pollIVar v+++-----------------------------------------------------------------------------+-- type of global IVars; instances mostly inherited from global references++-- A global IVar is a global reference to an IVar; 'm' abstracts a monad.+-- NOTE: The HdpH interface will restrict the type parameter 'a' to +--       'Closure b' for some type 'b', but but the type constructor 'GIVar' +--       does not enforce this restriction.+type GIVar m a = GRef (IVar m a)+++-----------------------------------------------------------------------------+-- operations on global IVars++-- Returns node hosting given global IVar.+hostGIVar :: GIVar m a -> NodeId+hostGIVar = at+++-- Globalise the given IVar;+-- the scheduler ID argument may be used for logging.+globIVar :: Int -> IVar m a -> IO (GIVar m a)+globIVar schedID v = do+  gv <- globalise v+  debug dbgGIVar $ "New global IVar " ++ show gv+  return gv+++-- Write 'x' to the locally hosted global IVar 'gv', free 'gv' and return +-- the list of blocked threads. Like putIVar, multiple writes fail silently+-- (as do writes to a dead global IVar);+-- the scheduler ID argument may be used for logging.+putGIVar :: Int -> GIVar m a -> a -> IO [Thread m]+putGIVar schedID gv x = do+  debug dbgGIVar $ "Put to global IVar " ++ show gv+  ts <- withGRef gv (\ v -> putIVar v x) (return [])+  free gv    -- free 'gv' (eventually)+  return ts
+ src/Control/Parallel/HdpH/Internal/Location.hs view
@@ -0,0 +1,112 @@+-- Locations+--   includes API for error and debug messages+--+-- Author: Rob Stewart, Patrick Maier+-----------------------------------------------------------------------------++module Control.Parallel.HdpH.Internal.Location+  ( -- * node IDs (and their constitutent parts)+    NodeId,           -- instances: Eq, Ord, Show, NFData, Serialize++    -- * reading all node IDs and this node's own node ID+    allNodes,             -- :: IO [NodeId]+    myNode,               -- :: IO NodeId+    myNode',              -- :: IO (Maybe NodeId)+    MyNodeException(..),  -- instances: Exception, Show, Typeable++    -- * error messages tagged by emitting node+    error,  -- :: String -> a++    -- * debug messages tagged by emitting node+    debug,  -- :: Int -> String -> IO ()++    -- * debug levels+    dbgNone,       -- :: Int+    dbgStats,      -- :: Int+    dbgStaticTab,  -- :: Int+    dbgSpark,      -- :: Int+    dbgMsgSend,    -- :: Int+    dbgMsgRcvd,    -- :: Int+    dbgGIVar,      -- :: Int+    dbgIVar,       -- :: Int+    dbgGRef,       -- :: Int+    dbgFailure     -- :: Int+  ) where++import Prelude hiding (catch, error)+import qualified Prelude (error)+import Control.DeepSeq (NFData)+import Control.Exception (catch, evaluate)+import Control.Monad (when)+import Data.Functor ((<$>))+import Data.IORef (readIORef)+import Data.Serialize (Serialize)+import System.IO (stderr, hPutStrLn)+import System.IO.Unsafe (unsafePerformIO)++import Control.Parallel.HdpH.Internal.State.Location+       (myNodeRef, allNodesRef, debugRef)+import Control.Parallel.HdpH.Internal.Type.Location+       (NodeId, MyNodeException(NodeIdUnset))+++-----------------------------------------------------------------------------+-- reading this node's own node ID++-- Return this node's node ID;+-- raises 'NodeIdUnset :: MyNodeException' if node ID has not yet been set+-- (by module HdpH.Internal.Comm).+myNode :: IO NodeId+myNode = readIORef myNodeRef+++-- Return 'Just' this node's node ID, or 'Nothing' if ID has not yet been set.+myNode' :: IO (Maybe NodeId)+myNode' =+  catch (Just <$> (evaluate =<< myNode))+        (const $ return Nothing :: MyNodeException -> IO (Maybe NodeId))+++-- Return list of all nodes (with main node being head of the list),+-- provided the list has been initialised (by module HdpH.Internal.Comm);+-- otherwise returns the empty list.+allNodes :: IO [NodeId]+allNodes = readIORef allNodesRef+++-----------------------------------------------------------------------------+-- error messages tagged by emitting node++-- Abort with error 'message'.+error :: String -> a+error message = case unsafePerformIO myNode' of+                  Just node -> Prelude.error (show node ++ " " ++ message)+                  Nothing   -> Prelude.error message+++-----------------------------------------------------------------------------+-- debug messages tagged by emitting node++-- Output a debug 'message' to 'stderr' if the given 'level' is less than+-- or equal to the system level; 'level' should be positive.+debug :: Int -> String -> IO ()+debug level message = do+  sysLevel <- readIORef debugRef+  when (level <= sysLevel) $ do+    maybe_this <- myNode'+    case maybe_this of+      Just this -> hPutStrLn stderr $ show this ++ " " ++ message+      Nothing   -> hPutStrLn stderr $ "<unknown> " ++ message+++-- debug levels+dbgNone      = 0 :: Int  -- no debug output+dbgStats     = 1 :: Int  -- print final stats+dbgStaticTab = 2 :: Int  -- on main node, print Static table+dbgSpark     = 3 :: Int  -- spark created or converted+dbgMsgSend   = 4 :: Int  -- message to be sent+dbgMsgRcvd   = 5 :: Int  -- message being handled+dbgGIVar     = 6 :: Int  -- op on a GIVar (globalising or writing to)+dbgIVar      = 7 :: Int  -- blocking/unblocking on IVar (only log event type)+dbgGRef      = 8 :: Int  -- registry update (globalise or free)+dbgFailure   = 9 :: Int  -- e.g. Node failure
+ src/Control/Parallel/HdpH/Internal/Misc.hs view
@@ -0,0 +1,254 @@+-- Misc auxiliary types and functions that should probably be in other modules.+--+-- Author: Patrick Maier+-------------------------------------------------------------------------------++{-# LANGUAGE GADTs #-}               -- for existential types+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE FlexibleContexts #-}    -- for Forkable instances+{-# LANGUAGE FlexibleInstances #-}++module Control.Parallel.HdpH.Internal.Misc+  ( -- * existential wrapper type+    AnyType(..),  -- no instances++    -- * monads supporting forking of threads+    Forkable(     -- context: (Monad m) => Forkable m+      fork,       -- ::        m () -> m Control.Concurrent.ThreadId+      forkOn      -- :: Int -> m () -> m Control.Concurrent.ThreadId+    ),++    -- * continuation monad with stricter bind+    Cont(..),     -- instances: Functor, Monad++    -- * rotate a list (to the left)+    rotate,       --  :: Int -> [a] -> [a]++    -- * decode ByteStrings (without error reporting)+    decode,       -- :: Serialize a => Strict.ByteString -> a+    decodeLazy,   -- :: Serialize a => Lazy.ByteString -> a++    -- * encode ByteStrings (companions to the decoders above)+    encode,       -- :: Serialize a => a -> Strict.ByteString+    encodeLazy,   -- :: Serialize a => a -> Lazy.ByteString++    -- * encode/decode lists of bytes+    encodeBytes,  -- :: Serialize a => a -> [Word8]+    decodeBytes,  -- :: Serialize a => [Word8] -> a++    -- * destructors of Either values+    fromLeft,     -- :: Either a b -> a+    fromRight,    -- :: Either a b -> b++    -- * splitting a list+    splitAtFirst, -- :: (a -> Bool) -> [a] -> Maybe ([a], a, [a])++    -- * To remove an Eq element from a list+    rmElems, -- :: Eq a => a -> [a] -> [a]++    -- * action servers+    Action,       -- synonym: IO ()+    ActionServer, -- abstract, no instances+    newServer,    -- :: IO ActionServer+    killServer,   -- :: ActionServer -> IO ()+    reqAction,    -- :: ActionServer -> Action -> IO ()++    -- * timing IO actions+    timeIO        -- :: IO a -> IO (a, NominalDiffTime)+  ) where++import Prelude hiding (error)+import Control.Concurrent (ThreadId, forkIO, killThread)+import Control.Concurrent.Chan (Chan, newChan, writeChan, readChan)+import Control.DeepSeq (NFData(rnf))+import Control.Monad (join)+import Control.Monad.Reader (ReaderT, runReaderT, ask)+import Control.Monad.Trans (lift)+import qualified Data.ByteString +       as Strict (ByteString, foldl', unpack)+import qualified Data.ByteString.Lazy+       as Lazy (ByteString, foldl', pack, unpack)+import Data.Serialize (Serialize)+import qualified Data.Serialize (encode, decode, encodeLazy, decodeLazy)+import Data.Time.Clock (NominalDiffTime, diffUTCTime, getCurrentTime)+import Data.Word (Word8)+import qualified GHC.Conc (forkOn)  -- GHC specific!++import Control.Parallel.HdpH.Internal.Location (error)+++-------------------------------------------------------------------------------+-- NFData instances for strict and lazy bytestrings+-- by strictly folding rnf for Word8++-- THIS IS in the `bytestring' package from >= 0.10.0.0+-- This instance should be part of module 'Data.ByteString.Lazy'.+{-+instance NFData Lazy.ByteString where+  rnf = Lazy.foldl' (\ _ -> rnf) ()+-}++-------------------------------------------------------------------------------+-- Functionality missing in Data.List++rotate :: Int -> [a] -> [a]+rotate _ [] = []+rotate n xs = zipWith const (drop n $ cycle xs) xs+++-------------------------------------------------------------------------------+-- Functionality missing in Data.Serialize++encode :: Serialize a => a -> Strict.ByteString+encode = Data.Serialize.encode++decode :: Serialize a => Strict.ByteString -> a+decode bs =+  case Data.Serialize.decode bs of+    Right x  -> x+    Left msg -> error $ "HdpH.Internal.Misc.decode " +++                         showPrefix 10 bs ++ ": " ++ msg++encodeLazy :: Serialize a => a -> Lazy.ByteString+encodeLazy = Data.Serialize.encodeLazy++decodeLazy :: Serialize a => Lazy.ByteString -> a+decodeLazy bs =+  case Data.Serialize.decodeLazy bs of+    Right x  -> x+    Left msg -> error $ "HdpH.Internal.Misc.decodeLazy " +++                        showPrefixLazy 10 bs ++ ": " ++ msg++decodeBytes :: Serialize a => [Word8] -> a+decodeBytes = decodeLazy . Lazy.pack++encodeBytes :: Serialize a => a -> [Word8]+encodeBytes = Lazy.unpack . Data.Serialize.encodeLazy+++showPrefix :: Int -> Strict.ByteString -> String+showPrefix n bs = showListUpto n (Strict.unpack bs) ""++showPrefixLazy :: Int -> Lazy.ByteString -> String+showPrefixLazy n bs = showListUpto n (Lazy.unpack bs) ""++showListUpto :: (Show a) => Int -> [a] -> String -> String+showListUpto n []     = showString "[]"+showListUpto n (x:xs) = showString "[" . shows x . go (n - 1) xs+  where+    go _ [] = showString "]"+    go n (x:xs) | n > 0     = showString "," . shows x . go (n - 1) xs+                | otherwise = showString ",...]"+++-------------------------------------------------------------------------------+-- Existential type (serves as wrapper for values in heterogenous Maps)++data AnyType :: * where+  Any :: a -> AnyType+++-------------------------------------------------------------------------------+-- Split a list at the first occurence of the given predicate;+-- the witness to the splitting occurence is stored as the middle element.++splitAtFirst :: (a -> Bool) -> [a] -> Maybe ([a], a, [a])+splitAtFirst p xs = let (left, rest) = break p xs in+                    case rest of+                      []           -> Nothing+                      middle:right -> Just (left, middle, right)+++-------------------------------------------------------------------------------+-- Destructors for Either values++fromLeft :: Either a b -> a+fromLeft (Left x) = x+fromLeft _        = error "HdpH.Internal.Misc.fromLeft: wrong constructor"++fromRight :: Either a b -> b+fromRight (Right y) = y+fromRight _         = error "HdpH.Internal.Misc.fromRight: wrong constructor"+++-------------------------------------------------------------------------------+-- Remove an Eq element from a list++rmElems' :: Eq a => a -> [a] -> [a]+rmElems' deleted xs = [ x | x <- xs, x /= deleted ]+rmElems :: Eq a => [a] -> [a] -> [a]+rmElems [] xs = xs+rmElems [y] xs = rmElems' y xs+rmElems (y:ys) xs = rmElems ys (rmElems' y xs)++-----------------------------------------------------------------------------+-- Forkable class and instances; adapted from Control.Concurrent.MState++class (Monad m) => Forkable m where+  fork   ::        m () -> m ThreadId+  forkOn :: Int -> m () -> m ThreadId++instance Forkable IO where+  fork   = forkIO+  forkOn = GHC.Conc.forkOn+  -- NOTE: 'forkOn' may cause massive variations in performance.++instance (Forkable m) => Forkable (ReaderT i m) where+  fork       action = do state <- ask+                         lift $ fork $ runReaderT action state+  forkOn cpu action = do state <- ask+                         lift $ forkOn cpu $ runReaderT action state+++-----------------------------------------------------------------------------+-- Continuation monad with stricter bind; adapted from Control.Monad.Cont++newtype Cont r a = Cont { runCont :: (a -> r) -> r }++instance Functor (Cont r) where+    fmap f m = Cont $ \c -> runCont m (c . f)++-- The Monad instance is where we differ from Control.Monad.Cont,+-- the difference being the use of strict application ($!).+instance Monad (Cont r) where+    return a = Cont $ \ c -> c $! a+    m >>= k  = Cont $ \ c -> runCont m $ \ a -> runCont (k $! a) c+++-----------------------------------------------------------------------------+-- Action server++-- Actions are computations of type 'IO ()', and an action server is nothing+-- but a thread receiving actions over a channel and executing them one after+-- the other. Note that actions may block for a long time (eg. delay for +-- several seconds), in which case the server itself is blocked (which is+-- intended).++type Action = IO ()+data ActionServer = ActionServer (Chan Action) ThreadId++newServer :: IO ActionServer+newServer = do trigger <- newChan+               tid <- forkIO $ server trigger+               return (ActionServer trigger tid)++killServer :: ActionServer -> IO ()+killServer (ActionServer _ tid) = killThread tid++reqAction :: ActionServer -> Action -> IO ()+reqAction (ActionServer trigger _) = writeChan trigger++server :: Chan Action -> IO ()+server trigger = do join (readChan trigger)+                    server trigger+++-----------------------------------------------------------------------------+-- Timing an IO action++timeIO :: IO a -> IO (a, NominalDiffTime)+timeIO action = do t0 <- getCurrentTime+                   x <- action+                   t1 <- getCurrentTime+                   return (x, diffUTCTime t1 t0)+
+ src/Control/Parallel/HdpH/Internal/Scheduler.hs view
@@ -0,0 +1,276 @@+-- Work stealing scheduler and thread pools+--+-- Author: Patrick Maier+-----------------------------------------------------------------------------++{-# LANGUAGE GeneralizedNewtypeDeriving #-}  -- req'd for type 'RTS'+{-# LANGUAGE ScopedTypeVariables #-}         -- req'd for type annotations++module Control.Parallel.HdpH.Internal.Scheduler+  ( -- * abstract run-time system monad+    RTS,          -- instances: Monad, Functor+    run_,         -- :: RTSConf -> RTS () -> IO ()+    liftThreadM,  -- :: ThreadM RTS a -> RTS a+    liftSparkM,   -- :: SparkM RTS a -> RTS a+    liftCommM,    -- :: CommM a -> RTS a+    liftIO,       -- :: IO a -> RTS a++    -- * scheduler ID+    schedulerID,  -- :: RTS Int++    -- * converting and executing threads+    mkThread,     -- :: ParM RTS a -> Thread RTS+    execThread,   -- :: Thread RTS -> RTS ()++    -- * pushing sparks+    sendPUSH      -- :: Spark RTS -> NodeId -> RTS ()+  ) where++import Prelude hiding (error)+import Control.Concurrent (ThreadId, forkIO, killThread)+import Control.Monad (unless, replicateM)+import Data.Functor ((<$>))++import Control.Parallel.HdpH.Closure (unClosure)+import Control.Parallel.HdpH.Conf (RTSConf(scheds, wakeupDly))+import Control.Parallel.HdpH.Internal.Comm (CommM)+import qualified Control.Parallel.HdpH.Internal.Comm as Comm+       (myNode, send, receive, run_, waitShutdown)+import qualified Control.Parallel.HdpH.Internal.Data.Deque as Deque (emptyIO)+import qualified Control.Parallel.HdpH.Internal.Data.Sem as Sem+       (new, signalPeriodically)+import Control.Parallel.HdpH.Internal.Location+       (NodeId, dbgNone, dbgStats, dbgMsgSend, dbgMsgRcvd, error)+import qualified Control.Parallel.HdpH.Internal.Location as Location (debug)+import Control.Parallel.HdpH.Internal.Misc+       (encodeLazy, decodeLazy, ActionServer, newServer, killServer)+import Control.Parallel.HdpH.Internal.Sparkpool+       (SparkM, blockSched, getSpark, Msg(PUSH), dispatch, readPoolSize,+        readFishSentCtr, readSparkRcvdCtr, readSparkGenCtr, readMaxSparkCtr)+import qualified Control.Parallel.HdpH.Internal.Sparkpool as Sparkpool (run)+import Control.Parallel.HdpH.Internal.Threadpool+       (ThreadM, poolID, forkThreadM, stealThread, readMaxThreadCtrs)+import qualified Control.Parallel.HdpH.Internal.Threadpool as Threadpool+       (run, liftSparkM, liftCommM, liftIO)+import Control.Parallel.HdpH.Internal.Type.Par+       (ParM, unPar, Thread(Atom), Spark)+++-----------------------------------------------------------------------------+-- RTS monad++-- The RTS monad hides monad stack (IO, CommM, SparkM, ThreadM) as abstract.+newtype RTS a = RTS { unRTS :: ThreadM RTS a }+                deriving (Functor, Monad)+++-- Fork a new thread to execute the given 'RTS' action; the integer 'n'+-- dictates how much to rotate the thread pools (so as to avoid contention+-- due to concurrent access).+forkRTS :: Int -> RTS () -> RTS ThreadId+forkRTS n = liftThreadM . forkThreadM n . unRTS+++-- Eliminate the whole RTS monad stack down the IO monad by running the given+-- RTS action 'main'; aspects of the RTS's behaviour are controlled by+-- the respective parameters in the given RTSConf.+-- NOTE: This function start various threads (for executing schedulers, +--       a message handler, and various timeouts). On normal termination,+--       all these threads are killed. However, there is no cleanup in the +--       event of aborting execution due to an exception. The functions+--       for doing so (see Control.Execption) all live in the IO monad.+--       Maybe they could be lifted to the RTS monad by using the monad-peel+--       package.+run_ :: RTSConf -> RTS () -> IO ()+run_ conf main = do+  let n = scheds conf+  unless (n > 0) $+    error "HdpH.Internal.Scheduler.run_: no schedulers"++  -- allocate n+1 empty thread pools (numbered from 0 to n)+  pools <- mapM (\ k -> do { pool <- Deque.emptyIO; return (k,pool) }) [0 .. n]++  -- fork nowork server (for clearing the "FISH outstanding" flag on NOWORK)+  noWorkServer <- newServer++  -- create semaphore for idle schedulers+  idleSem <- Sem.new++  -- fork wakeup server (periodically waking up racey sleeping scheds)+  wakeupServerTid <- forkIO $ Sem.signalPeriodically idleSem (wakeupDly conf)++  -- start the RTS+  Comm.run_ conf $+    Sparkpool.run conf noWorkServer idleSem $+      Threadpool.run pools $+        unRTS $+          rts n noWorkServer wakeupServerTid++    where+      -- RTS action+      rts :: Int -> ActionServer -> ThreadId -> RTS ()+      rts scheds noWorkServer wakeupServerTid = do++        -- fork message handler (accessing thread pool 0)+        handlerTid <- forkRTS 0 handler++        -- fork schedulers (each accessing thread pool k, 1 <= k <= scheds)+        schedulerTids <- mapM (\ k -> forkRTS k scheduler) [1 .. scheds]++        -- run main RTS action+        main++        -- block waiting for shutdown barrier+        liftCommM $ Comm.waitShutdown++        -- print stats+        printFinalStats++        -- kill nowork server+        liftIO $ killServer noWorkServer++        -- kill wakeup server+        liftIO $ killThread wakeupServerTid++        -- kill message handler+        liftIO $ killThread handlerTid++        -- kill schedulers+        liftIO $ mapM_ killThread schedulerTids+++-- lifting lower layers+liftThreadM :: ThreadM RTS a -> RTS a+liftThreadM = RTS++liftSparkM :: SparkM RTS a -> RTS a+liftSparkM = liftThreadM . Threadpool.liftSparkM++liftCommM :: CommM a -> RTS a+liftCommM = liftThreadM . Threadpool.liftCommM++liftIO :: IO a -> RTS a+liftIO = liftThreadM . Threadpool.liftIO+++-- Return scheduler ID, that is ID of scheduler's own thread pool.+schedulerID :: RTS Int+schedulerID = liftThreadM poolID+++-----------------------------------------------------------------------------+-- cooperative scheduling++-- Execute the given thread until it blocks or terminates.+execThread :: Thread RTS -> RTS ()+execThread (Atom m) = m >>= maybe (return ()) execThread+++-- Try to get a thread from a thread pool or the spark pool and execute it+-- until it blocks or terminates, whence repeat forever; if there is no+-- thread to execute then block the scheduler (ie. its underlying IO thread).+scheduler :: RTS ()+scheduler = getThread >>= scheduleThread+++-- Execute given thread until it blocks or terminates, whence call 'scheduler'.+scheduleThread :: Thread RTS -> RTS ()+scheduleThread (Atom m) = m >>= maybe scheduler scheduleThread+++-- Try to steal a thread from any thread pool (with own pool preferred);+-- if there is none, try to convert a spark from the spark pool;+-- if there is none too, block the scheduler such that the 'getThread'+-- action will be repeated on wake up.+-- NOTE: Sleeping schedulers should be woken up+--       * after new threads have been added to a thread pool,+--       * after new sparks have been added to the spark pool, and+--       * once the delay after a NOWORK message has expired.+getThread :: RTS (Thread RTS)+getThread = do+  schedID <- schedulerID+  maybe_thread <- liftThreadM stealThread+  case maybe_thread of+    Just thread -> return thread+    Nothing     -> do+      maybe_spark <- liftSparkM $ getSpark schedID+      case maybe_spark of+        Just spark -> return $ mkThread $ unClosure spark+        Nothing    -> liftSparkM blockSched >> getThread+++-- Converts 'Par' computations into threads.+mkThread :: ParM RTS a -> Thread RTS+mkThread p = unPar p $ \ _c -> Atom (return Nothing)+++-----------------------------------------------------------------------------+-- pushed sparks++-- Send a 'spark' via PUSH message to the given 'target' unless 'target'+-- is the current node (in which case 'spark' is executed immediately).+sendPUSH :: Spark RTS -> NodeId -> RTS ()+sendPUSH spark target = do+  here <- liftCommM Comm.myNode+  if target == here+    then do+      -- short cut PUSH msg locally+      execSpark spark+    else do+      -- construct and send PUSH message+      let msg = PUSH spark :: Msg RTS+      debug dbgMsgSend $+        show msg ++ " ->> " ++ show target+      liftCommM $ Comm.send target $ encodeLazy msg+++-- Handle a PUSH message by converting the spark into a thread and+-- executing it immediately.+handlePUSH :: Msg RTS -> RTS ()+handlePUSH (PUSH spark) = execSpark spark+++-- Execute a spark (by converting it to a thread and executing).+execSpark :: Spark RTS -> RTS ()+execSpark spark = execThread $ mkThread $ unClosure spark+++-----------------------------------------------------------------------------+-- message handler; only PUSH messages are actually handled here in this+-- module, other messages are relegated to module Sparkpool.++-- Message handler, running continously (in its own thread) receiving+-- and handling messages (some of which may unblock threads or create sparks)+-- as they arrive.+handler :: RTS ()+handler = do+  msg <- decodeLazy <$> liftCommM Comm.receive+  sparks <- liftSparkM readPoolSize+  debug dbgMsgRcvd $+    ">> " ++ show msg ++ " #sparks=" ++ show sparks+  case msg of+    PUSH _ -> handlePUSH msg+    _      -> liftSparkM $ dispatch msg+  handler+++-----------------------------------------------------------------------------+-- auxiliary stuff++-- Print stats (#sparks, threads, FISH, ...) at appropriate debug level.+-- TODO: Log time elapsed since RTS is up+printFinalStats :: RTS ()+printFinalStats = do+  fishes <- liftSparkM $ readFishSentCtr+  scheds <- liftSparkM $ readSparkRcvdCtr+  sparks <- liftSparkM $ readSparkGenCtr+  max_sparks  <- liftSparkM $ readMaxSparkCtr+  maxs_threads <- liftThreadM $ readMaxThreadCtrs+  debug dbgStats $ "#SPARK=" ++ show sparks ++ "   " +++                   "max_SPARK=" ++ show max_sparks ++ "   " +++                   "max_THREAD=" ++ show maxs_threads+  debug dbgStats $ "#FISH_sent=" ++ show fishes ++ "   " +++                   "#SCHED_rcvd=" ++ show scheds++debug :: Int -> String -> RTS ()+debug level message = liftIO $ Location.debug level message
+ src/Control/Parallel/HdpH/Internal/Sparkpool.hs view
@@ -0,0 +1,484 @@+-- Spark pool and fishing+--+-- Author: Patrick Maier+-----------------------------------------------------------------------------++{-# LANGUAGE ScopedTypeVariables #-}  -- req'd for type annotations++module Control.Parallel.HdpH.Internal.Sparkpool+  ( -- * spark pool monad+    SparkM,      -- synonym: SparkM m = ReaderT <State m> CommM+    run,         -- :: RTSConf -> ActionServer -> Sem -> SparkM m a -> CommM a+    liftCommM,   -- :: Comm a -> SparkM m a+    liftIO,      -- :: IO a -> SparkM m a++    -- * blocking and unblocking idle schedulers+    blockSched,  -- :: SparkM m ()+    wakeupSched, -- :: Int -> SparkM m ()++    -- * local (ie. scheduler) access to spark pool+    getSpark,    -- :: Int -> SparkM m (Maybe (Spark m))+    putSpark,    -- :: Int -> Spark m -> SparkM m ()++    -- * messages+    Msg(..),         -- instances: Show, NFData, Serialize++    -- * handle messages related to fishing+    dispatch,        -- :: Msg m -> SparkM m ()+    handleFISH,      -- :: Msg m -> SparkM m ()+    handleSCHEDULE,  -- :: Msg m -> SparkM m ()+    handleNOWORK,    -- :: Msg m -> SparkM m ()++    -- * access to stats data+    readPoolSize,      -- :: SparkM m Int+    readFishSentCtr,   -- :: SparkM m Int+    readSparkRcvdCtr,  -- :: SparkM m Int+    readMaxSparkCtr,   -- :: SparkM m Int+    readSparkGenCtr,   -- :: SparkM m Int+    readSparkConvCtr   -- :: SparkM m Int+  ) where++import Prelude hiding (error)+import Control.Concurrent (threadDelay)+import Control.DeepSeq (NFData, rnf)+import Control.Monad (unless, when, replicateM_)+import Control.Monad.Reader (ReaderT, runReaderT, ask)+import Control.Monad.Trans (lift)+import Data.Functor ((<$>))+import Data.IORef (IORef, newIORef, readIORef, writeIORef, atomicModifyIORef)+import Data.Serialize (Serialize)+import qualified Data.Serialize (put, get)+import Data.Set (Set)+import qualified Data.Set as Set (size, fromList, singleton, notMember)+import Data.Word (Word8)+import System.Random (randomRIO)++import Control.Parallel.HdpH.Conf+       (RTSConf(maxHops, maxFish, minSched, minFishDly, maxFishDly))+import Control.Parallel.HdpH.Internal.Comm (CommM)+import qualified Control.Parallel.HdpH.Internal.Comm as Comm+       (liftIO, send, nodes, myNode, allNodes)+import Control.Parallel.HdpH.Internal.Data.Deque+       (DequeIO, emptyIO, pushFrontIO, pushBackIO, popFrontIO, popBackIO,+        lengthIO, maxLengthIO)+import Control.Parallel.HdpH.Internal.Data.Sem (Sem)+import qualified Control.Parallel.HdpH.Internal.Data.Sem as Sem (wait, signal)+import Control.Parallel.HdpH.Internal.Location+       (NodeId, dbgMsgSend, dbgSpark, error)+import qualified Control.Parallel.HdpH.Internal.Location as Location (debug)+import Control.Parallel.HdpH.Internal.Misc (encodeLazy, ActionServer, reqAction)+import Control.Parallel.HdpH.Internal.Type.Par (Spark)+++-----------------------------------------------------------------------------+-- SparkM monad++-- 'SparkM m' is a reader monad sitting on top of the 'CommM' monad;+-- the parameter 'm' abstracts a monad (cf. module HdpH.Internal.Type.Par).+type SparkM m = ReaderT (State m) CommM+++-- spark pool state (mutable bits held in IORefs and the like)+data State m =+  State {+    s_conf       :: RTSConf,               -- config data+    s_pool       :: DequeIO (Spark m),     -- actual spark pool+    s_sparkOrig  :: IORef (Maybe NodeId),  -- origin of most recent spark recvd+    s_fishing    :: IORef Bool,            -- True iff FISH outstanding+    s_noWork     :: ActionServer,          -- for clearing "FISH outstndg" flag+    s_idleScheds :: Sem,                   -- semaphore for idle schedulers+    s_fishSent   :: IORef Int,             -- #FISH sent+    s_sparkRcvd  :: IORef Int,             -- #sparks received+    s_sparkGen   :: IORef Int,             -- #sparks generated+    s_sparkConv  :: IORef Int }            -- #sparks converted+++-- Eliminates the 'SparkM' layer by executing the given 'SparkM' action on+-- an empty spark pool; expects a config data, an action server (for+-- clearing "FISH outstanding" flag) and a semaphore (for idle schedulers).+run :: RTSConf -> ActionServer -> Sem -> SparkM m a -> CommM a+run conf noWorkServer idleSem action = do+  -- set up spark pool state+  pool      <- Comm.liftIO $ emptyIO+  sparkOrig <- Comm.liftIO $ newIORef Nothing+  fishing   <- Comm.liftIO $ newIORef False+  fishSent  <- Comm.liftIO $ newIORef 0+  sparkRcvd <- Comm.liftIO $ newIORef 0+  sparkGen  <- Comm.liftIO $ newIORef 0+  sparkConv <- Comm.liftIO $ newIORef 0+  let s0 = State { s_conf       = conf,+                   s_pool       = pool,+                   s_sparkOrig  = sparkOrig,+                   s_fishing    = fishing,+                   s_noWork     = noWorkServer,+                   s_idleScheds = idleSem,+                   s_fishSent   = fishSent,+                   s_sparkRcvd  = sparkRcvd,+                   s_sparkGen   = sparkGen,+                   s_sparkConv  = sparkConv }+  -- run monad+  runReaderT action s0+++-- Lifting lower layers.+liftCommM :: CommM a -> SparkM m a+liftCommM = lift++liftIO :: IO a -> SparkM m a+liftIO = liftCommM . Comm.liftIO+++-----------------------------------------------------------------------------+-- access to state++getPool :: SparkM m (DequeIO (Spark m))+getPool = s_pool <$> ask++readPoolSize :: SparkM m Int+readPoolSize = getPool >>= liftIO . lengthIO++getSparkOrigHist :: SparkM m (IORef (Maybe NodeId))+getSparkOrigHist = s_sparkOrig <$> ask++readSparkOrigHist :: SparkM m (Maybe NodeId)+readSparkOrigHist = getSparkOrigHist >>= liftIO . readIORef++updateSparkOrigHist :: NodeId -> SparkM m ()+updateSparkOrigHist mostRecentOrigin = do+  sparkOrigHistRef <- getSparkOrigHist +  liftIO $ writeIORef sparkOrigHistRef (Just mostRecentOrigin)++getFishingFlag :: SparkM m (IORef Bool)+getFishingFlag = s_fishing <$> ask++getNoWorkServer :: SparkM m ActionServer+getNoWorkServer = s_noWork <$> ask+  +getIdleSchedsSem :: SparkM m Sem+getIdleSchedsSem = s_idleScheds <$> ask++getFishSentCtr :: SparkM m (IORef Int)+getFishSentCtr = s_fishSent <$> ask++readFishSentCtr :: SparkM m Int+readFishSentCtr = getFishSentCtr >>= readCtr++getSparkRcvdCtr :: SparkM m (IORef Int)+getSparkRcvdCtr = s_sparkRcvd <$> ask++readSparkRcvdCtr :: SparkM m Int+readSparkRcvdCtr = getSparkRcvdCtr >>= readCtr++getSparkGenCtr :: SparkM m (IORef Int)+getSparkGenCtr = s_sparkGen <$> ask++readSparkGenCtr :: SparkM m Int+readSparkGenCtr = getSparkGenCtr >>= readCtr++getSparkConvCtr :: SparkM m (IORef Int)+getSparkConvCtr = s_sparkConv <$> ask++readSparkConvCtr :: SparkM m Int+readSparkConvCtr = getSparkConvCtr >>= readCtr++readMaxSparkCtr :: SparkM m Int+readMaxSparkCtr = getPool >>= liftIO . maxLengthIO++getMaxHops :: SparkM m Int+getMaxHops = maxHops <$> s_conf <$> ask++getMaxFish :: SparkM m Int+getMaxFish = maxFish <$> s_conf <$> ask++getMinSched :: SparkM m Int+getMinSched = minSched <$> s_conf <$> ask++getMinFishDly :: SparkM m Int+getMinFishDly = minFishDly <$> s_conf <$> ask++getMaxFishDly :: SparkM m Int+getMaxFishDly = maxFishDly <$> s_conf <$> ask+++-----------------------------------------------------------------------------+-- blocking and unblocking idle schedulers++-- Put executing scheduler to sleep.+blockSched :: SparkM m ()+blockSched = getIdleSchedsSem >>= liftIO . Sem.wait+++-- Wake up 'n' sleeping schedulers.+wakeupSched :: Int -> SparkM m ()+wakeupSched n = getIdleSchedsSem >>= liftIO . replicateM_ n . Sem.signal+++-----------------------------------------------------------------------------+-- local access to spark pool++-- Get a spark from the front of the spark pool, if there is any;+-- possibly send a FISH message and update stats (ie. count sparks converted);+-- the scheduler ID argument may be used for logging.+getSpark :: Int -> SparkM m (Maybe (Spark m))+getSpark schedID = do+  pool <- getPool+  maybe_spark <- liftIO $ popFrontIO pool+  sendFISH+  case maybe_spark of+    Just _  -> do getSparkConvCtr >>= incCtr+                  sparks <- liftIO $ lengthIO pool+                  debug dbgSpark $+                    "#sparks=" ++ show sparks ++ " (spark converted)"+                  return maybe_spark+    Nothing -> do return maybe_spark+++-- Put a new spark at the back of the spark pool, wake up 1 sleeping scheduler,+-- and update stats (ie. count sparks generated locally);+-- the scheduler ID argument may be used for logging.+putSpark :: Int -> Spark m -> SparkM m ()+putSpark schedID spark = do+  pool <- getPool+  liftIO $ pushBackIO pool spark+  wakeupSched 1+  getSparkGenCtr >>= incCtr+  sparks <- liftIO $ lengthIO pool+  debug dbgSpark $+    "#sparks=" ++ show sparks ++ " (spark created)"+++-----------------------------------------------------------------------------+-- HdpH messages (peer to peer)++-- 4 different types of messages dealing with fishing and pushing sparks;+-- the parameter 's' abstracts the type of sparks+data Msg m = FISH        -- looking for work+               !NodeId     -- fishing node+               !NodeId     -- primary target (eg. where last spark came from)+               !Int        -- #hops FISH message may yet travel+           | NOWORK      -- reply to FISH sender (when there is no work)+           | SCHEDULE    -- reply to FISH sender (when there is work)+               (Spark m)   -- spark+               !NodeId     -- sender+           | PUSH        -- eagerly pushing work+               (Spark m)   -- spark+++-- Show instance (mainly for debugging)+instance Show (Msg m) where+  showsPrec _ (FISH fisher target hops) = showString "FISH(" . shows fisher .+                                          showString "," . shows target .+                                          showString "," . shows hops .+                                          showString ")"+  showsPrec _ (NOWORK)                  = showString "NOWORK"+  showsPrec _ (SCHEDULE _spark sender)  = showString "SCHEDULE(_," .+                                          shows sender . showString ")"+  showsPrec _ (PUSH _spark)             = showString "PUSH(_)"+++instance NFData (Msg m) where+  rnf (FISH fisher target hops) = rnf fisher `seq` rnf target `seq` rnf hops+  rnf (NOWORK)                  = ()+  rnf (SCHEDULE spark sender)   = rnf spark `seq` rnf sender+  rnf (PUSH spark)              = rnf spark+++instance Serialize (Msg m) where+  put (FISH fisher target hops) = Data.Serialize.put (0 :: Word8) >>+                                  Data.Serialize.put fisher >>+                                  Data.Serialize.put target >>+                                  Data.Serialize.put hops+  put (NOWORK)                  = Data.Serialize.put (1 :: Word8)+  put (SCHEDULE spark sender)   = Data.Serialize.put (2 :: Word8) >>+                                  Data.Serialize.put spark >>+                                  Data.Serialize.put sender+  put (PUSH spark)              = Data.Serialize.put (3 :: Word8) >>+                                  Data.Serialize.put spark++  get = do tag <- Data.Serialize.get+           case tag :: Word8 of+             0 -> do fisher <- Data.Serialize.get+                     target <- Data.Serialize.get+                     hops   <- Data.Serialize.get+                     return $ FISH fisher target hops+             1 -> do return $ NOWORK+             2 -> do spark  <- Data.Serialize.get+                     sender <- Data.Serialize.get+                     return $ SCHEDULE spark sender+             3 -> do spark  <- Data.Serialize.get+                     return $ PUSH spark+++-----------------------------------------------------------------------------+-- fishing and the like++-- Send a FISH message, but only if there is no FISH outstanding and the+-- number of sparks in the pool is less or equal to the 'maxFish' parameter;+-- the target is the sender of the most recent SCHEDULE message, if a+-- SCHEDULE message has yet been received, otherwise the target is random.+sendFISH :: SparkM m ()+sendFISH = do+  pool <- getPool+  fishingFlag <- getFishingFlag+  isFishing <- readFlag fishingFlag+  unless isFishing $ do+    -- no FISH currently outstanding+    nodes <- liftCommM $ Comm.nodes+    maxFish <- getMaxFish+    sparks <- liftIO $ lengthIO pool+    when (nodes > 1 && sparks <= maxFish) $ do+      -- there are other nodes and the pool has too few sparks;+      -- set flag indicating that we are going to send a FISH+      ok <- setFlag fishingFlag+      when ok $ do+        -- flag was clear before: go ahead sending FISH;+        -- construct message+        fisher <- liftCommM $ Comm.myNode+        hops <- getMaxHops+        -- target is node where most recent spark came from (if such exists)+        maybe_target <- readSparkOrigHist+        target <- case maybe_target of+                    Just node -> return node+                    Nothing   -> do allNodes <- liftCommM $ Comm.allNodes+                                    let avoidNodes = Set.singleton fisher+                                    -- select random target (other than fisher)+                                    randomOtherElem avoidNodes allNodes nodes+        let msg = FISH fisher target hops :: Msg m+        -- send message+        debug dbgMsgSend $+          show msg ++ " ->> " ++ show target+        liftCommM $ Comm.send target $ encodeLazy msg+        -- update stats+        getFishSentCtr >>= incCtr+++-- Dispatch FISH, SCHEDULE and NOWORK messages to their respective handlers.+dispatch :: Msg m -> SparkM m ()+dispatch msg@(FISH _ _ _)   = handleFISH msg+dispatch msg@(SCHEDULE _ _) = handleSCHEDULE msg+dispatch msg@(NOWORK)       = handleNOWORK msg+dispatch msg = error $ "HdpH.Internal.Sparkpool.dispatch: " +++                       show msg ++ " unexpected"+++-- Handle a FISH message; replies+-- * with SCHEDULE if pool has enough sparks, or else+-- * with NOWORK if FISH has travelled far enough, or else+-- * forwards FISH to a random node (other than fisher or target).+handleFISH :: forall m . Msg m -> SparkM m ()+handleFISH msg@(FISH fisher target hops) = do+  here <- liftCommM $ Comm.myNode+  sparks <- readPoolSize+  minSched <- getMinSched+  -- send SCHEDULE if pool has enough sparks+  done <- if sparks < minSched+             then do return False+             else do+               pool <- getPool+               maybe_spark <- liftIO $ popBackIO pool+               case maybe_spark of+                 Just spark -> do let msg = SCHEDULE spark here :: Msg m+                                  debug dbgMsgSend $+                                    show msg ++ " ->> " ++ show fisher+                                  liftCommM $ Comm.send fisher $ encodeLazy msg+                                  return True+                 Nothing    -> do return False+  unless done $ do+    -- no SCHEDULE sent; check whether to forward FISH+    nodes <- liftCommM $ Comm.nodes+    let avoidNodes = Set.fromList [fisher, target, here]+    if hops > 0 && nodes > Set.size avoidNodes+       then do  -- fwd FISH to random node (other than those in avoidNodes)+         allNodes <- liftCommM $ Comm.allNodes+         node <- randomOtherElem avoidNodes allNodes nodes+         let msg = FISH fisher target (hops - 1) :: Msg m+         debug dbgMsgSend $+           show msg ++ " ->> " ++ show node+         liftCommM $ Comm.send node $ encodeLazy msg+       else do  -- notify fisher that there is no work+         let msg = NOWORK :: Msg m+         debug dbgMsgSend $+           show msg ++ " ->> " ++ show fisher+         liftCommM $ Comm.send fisher $ encodeLazy msg+++-- Handle a SCHEDULE message;+-- * puts the spark at the front of the spark pool,+-- * records spark sender and updates stats, and+-- * clears the "FISH outstanding" flag.+handleSCHEDULE :: Msg m -> SparkM m ()+handleSCHEDULE msg@(SCHEDULE spark sender) = do+  -- put spark into pool+  pool <- getPool+  liftIO $ pushFrontIO pool spark+  -- record sender of spark+  updateSparkOrigHist sender+  -- update stats+  getSparkRcvdCtr >>= incCtr+  -- clear FISHING flag+  getFishingFlag >>= clearFlag+  return ()+++-- Handle a NOWORK message; +-- asynchronously, after a random delay, clear the "FISH outstanding" flag +-- and wake one scheduler (if some are sleeping) to resume fishing.+-- Rationale for random delay: to prevent FISH flooding when there is+--   (almost) no work.+handleNOWORK :: Msg m -> SparkM m ()+handleNOWORK msg@(NOWORK) = do+  fishingFlag   <- getFishingFlag+  noWorkServer  <- getNoWorkServer+  idleSchedsSem <- getIdleSchedsSem+  minDelay      <- getMinFishDly+  maxDelay      <- getMaxFishDly+  -- compose delay and clear flag action+  let action = do -- random delay+                  delay <- randomRIO (minDelay, max minDelay maxDelay)+                  threadDelay delay+                  -- clear fishing flag+                  atomicModifyIORef fishingFlag $ const (False, ())+                  -- wakeup 1 sleeping scheduler (to fish again)+                  Sem.signal idleSchedsSem+  -- post action request to server+  liftIO $ reqAction noWorkServer action+++-----------------------------------------------------------------------------+-- auxiliary stuff++readFlag :: IORef Bool -> SparkM m Bool+readFlag = liftIO . readIORef++-- Sets given 'flag'; returns True iff 'flag' did actually change.+setFlag :: IORef Bool -> SparkM m Bool+setFlag flag = liftIO $ atomicModifyIORef flag $ \ v -> (True, not v)++-- Clears given 'flag'; returns True iff 'flag' did actually change.+clearFlag :: IORef Bool -> SparkM m Bool+clearFlag flag = liftIO $ atomicModifyIORef flag $ \ v -> (False, v)+++readCtr :: IORef Int -> SparkM m Int+readCtr = liftIO . readIORef++incCtr :: IORef Int -> SparkM m ()+incCtr ctr = liftIO $ atomicModifyIORef ctr $ \ v ->+                        let v' = v + 1 in v' `seq` (v', ())+++-- 'randomOtherElem avoid xs n' returns a random element of the list 'xs'+-- different from any of the elements in the set 'avoid'.+-- Requirements: 'n <= length xs' and 'xs' contains no duplicates.+randomOtherElem :: (Ord a) => Set a -> [a] -> Int -> SparkM m a+randomOtherElem avoid xs n = do+  let candidates = filter (`Set.notMember` avoid) xs+  -- length candidates == length xs - Set.size avoid >= n - Set.size avoid+  i <- liftIO $ randomRIO (0, n - Set.size avoid - 1)+  -- 0 <= i <= n - Set.size avoid - 1 < length candidates+  return (candidates !! i)+++-- debugging+debug :: Int -> String -> SparkM m ()+debug level message = liftIO $ Location.debug level message
+ src/Control/Parallel/HdpH/Internal/State/GRef.hs view
@@ -0,0 +1,28 @@+-- Global references; state+--+-- Author: Patrick Maier+-----------------------------------------------------------------------------++{-# OPTIONS_GHC -fno-cse #-}  -- to protect unsafePerformIO hack++module Control.Parallel.HdpH.Internal.State.GRef+  ( -- * reference to this node's registry+    regRef  -- :: IORef GRefReg+  ) where++import Prelude+import Data.IORef (IORef, newIORef)+import qualified Data.Map as Map (empty)+import System.IO.Unsafe (unsafePerformIO)++import Control.Parallel.HdpH.Internal.Type.GRef+       (GRefReg(GRefReg), lastSlot, table)+++-----------------------------------------------------------------------------+-- reference to this node's registry, initially empty++regRef :: IORef GRefReg+regRef = unsafePerformIO $+           newIORef $ GRefReg { lastSlot = 0, table = Map.empty }+{-# NOINLINE regRef #-}   -- required to protect unsafePerformIO hack
+ src/Control/Parallel/HdpH/Internal/State/Location.hs view
@@ -0,0 +1,54 @@+-- Locations; state+--+-- Author: Rob Stewart, Patrick Maier+-----------------------------------------------------------------------------++{-# OPTIONS_GHC -fno-cse #-}  -- to protect unsafePerformIO hack++module Control.Parallel.HdpH.Internal.State.Location+  ( -- * reference to this node's ID+    myNodeRef,    -- :: IORef NodeId++    -- * reference to IDs of all nodes+    allNodesRef,  -- :: IORef [NodeId]++    -- * reference to system debug level+    debugRef      -- :: IORef Int+  ) where++import Prelude+import Control.Exception (throw)+import Data.IORef (IORef, newIORef)+import System.IO.Unsafe (unsafePerformIO)++import Control.Parallel.HdpH.Internal.Type.Location+       (NodeId, MyNodeException(NodeIdUnset))+++-----------------------------------------------------------------------------+-- reference to ID of this node;+-- will be set in module HdpH.Internal.Comm;+-- if unset, access will raise a 'MyNodeException'++myNodeRef :: IORef NodeId+myNodeRef = unsafePerformIO $ newIORef $ throw NodeIdUnset+{-# NOINLINE myNodeRef #-}    -- required to protect unsafePerformIO hack+++-----------------------------------------------------------------------------+-- reference to list of IDs of all nodes;+-- will be set (to non-empty list) in module HdpH_IO++allNodesRef :: IORef [NodeId]+allNodesRef = unsafePerformIO $ newIORef $ []+{-# NOINLINE allNodesRef #-}  -- required to protect unsafePerformIO hack+++-----------------------------------------------------------------------------+-- reference to system debug level;+-- may be set in module HdpH.Internal.Comm;+-- referenced value must be non-negative (0 means no debug output)++debugRef :: IORef Int+debugRef  = unsafePerformIO $ newIORef $ 0+{-# NOINLINE debugRef #-}     -- required to protect unsafePerformIO hack
+ src/Control/Parallel/HdpH/Internal/Threadpool.hs view
@@ -0,0 +1,149 @@+-- Thread pool and work stealing+--+-- Author: Patrick Maier+-----------------------------------------------------------------------------++module Control.Parallel.HdpH.Internal.Threadpool+  ( -- * thread pool monad+    ThreadM,      -- synonym: ThreadM m = ReaderT <State m> (SparkM m)+    run,          -- :: [DequeIO (Thread m)] -> ThreadM m a -> SparkM m a+    forkThreadM,  -- :: Int -> ThreadM m () ->+                  --      ThreadM m Control.Concurrent.ThreadId+    liftSparkM,   -- :: SparkM m a -> ThreadM m a+    liftCommM,    -- :: CommM a -> ThreadM m a+    liftIO,       -- :: IO a -> ThreadM m a++    -- * thread pool ID (of scheduler's own pool)+    poolID,       -- :: ThreadM m Int++    -- * putting threads into the scheduler's own pool+    putThread,    -- :: Thread m -> ThreadM m ()+    putThreads,   -- :: [Thread m] -> ThreadM m ()++    -- * stealing threads (from scheduler's own pool, or from other pools)+    stealThread,  -- :: ThreadM m (Maybe (Thread m))++    -- * statistics+    readMaxThreadCtrs  -- :: ThreadM m [Int]+  ) where++import Prelude hiding (error)+import Control.Concurrent (ThreadId)+import Control.Monad.Reader (ReaderT, runReaderT, ask)+import Control.Monad.Trans (lift)++import Control.Parallel.HdpH.Internal.Comm (CommM)+import Control.Parallel.HdpH.Internal.Data.Deque+       (DequeIO, pushFrontIO, popFrontIO, popBackIO, maxLengthIO)+import Control.Parallel.HdpH.Internal.Location (error)+import Control.Parallel.HdpH.Internal.Misc (Forkable, fork, rotate)+import Control.Parallel.HdpH.Internal.Sparkpool (SparkM, wakeupSched)+import qualified Control.Parallel.HdpH.Internal.Sparkpool as Sparkpool+       (liftCommM, liftIO)+import Control.Parallel.HdpH.Internal.Type.Par (Thread)+++-----------------------------------------------------------------------------+-- thread pool monad++-- 'ThreadM' is a reader monad sitting on top of the 'SparkM' monad;+-- the parameter 'm' abstracts a monad (cf. module HdpH.Internal.Type.Par).+type ThreadM m = ReaderT (State m) (SparkM m)+++-- thread pool state (mutable bits held in DequeIO)+type State m = [(Int, DequeIO (Thread m))]  -- list of actual thread pools,+                                            -- each with identifying Int++-- Eliminates the 'ThreadM' layer by executing the given 'action' (typically+-- a scheduler loop) on the given non-empty list of thread 'pools' (the first+-- of which is the scheduler's own pool).+-- NOTE: An empty list of pools is admitted but then 'action' must not call+--      'putThread', 'putThreads', 'stealThread' or 'readMaxThreadCtrs'.+run :: [(Int, DequeIO (Thread m))] -> ThreadM m a -> SparkM m a+run pools action = runReaderT action pools+++-- Execute the given 'ThreadM' action in a new thread, sharing the same+-- thread pools (but rotated by 'n' pools).+forkThreadM :: Int -> ThreadM m () -> ThreadM m ThreadId+forkThreadM n action = do+  pools <- getPools+  lift $ fork $ run (rotate n pools) action+++-- Lifting lower layers.+liftSparkM :: SparkM m a -> ThreadM m a+liftSparkM = lift++liftCommM :: CommM a -> ThreadM m a+liftCommM = liftSparkM . Sparkpool.liftCommM++liftIO :: IO a -> ThreadM m a+liftIO = liftSparkM . Sparkpool.liftIO+++-----------------------------------------------------------------------------+-- access to state++getPools :: ThreadM m [(Int, DequeIO (Thread m))]+getPools = do pools <- ask+              case pools of+                [] -> error "HdpH.Internal.Threadpool.getPools: no pools"+                _  -> return pools+++-----------------------------------------------------------------------------+-- access to thread pool++-- Return thread pool ID, that is ID of scheduler's own pool.+poolID :: ThreadM m Int+poolID = do+  my_pool:_ <- getPools+  return $ fst my_pool+++-- Read the max size of each thread pool.+readMaxThreadCtrs :: ThreadM m [Int]+readMaxThreadCtrs = getPools >>= liftIO . mapM (maxLengthIO . snd)+++-- Steal a thread from any thread pool, with own pool as highest priority;+-- threads from own pool are always taken from the front; threads from other+-- pools are stolen from the back of those pools.+-- Rationale: Preserve locality as much as possible for own threads; try+-- not to disturb locality for threads stolen from others.+stealThread :: ThreadM m (Maybe (Thread m))+stealThread = do+  my_pool:other_pools <- getPools+  maybe_thread <- liftIO $ popFrontIO $ snd my_pool+  case maybe_thread of+    Just _  -> return maybe_thread+    Nothing -> steal other_pools+      where+        steal :: [(Int, DequeIO (Thread m))] -> ThreadM m (Maybe (Thread m))+        steal []           = return Nothing+        steal (pool:pools) = do+          maybe_thread <- liftIO $ popBackIO $ snd pool+          case maybe_thread of+            Just _  -> return maybe_thread+            Nothing -> steal pools+++-- Put the given thread at the front of the executing scheduler's own pool;+-- wake up 1 sleeping scheduler (if there is any).+putThread :: Thread m -> ThreadM m ()+putThread thread = do+  my_pool:_ <- getPools+  liftIO $ pushFrontIO (snd my_pool) thread+  liftSparkM $ wakeupSched 1+++-- Put the given threads at the front of the executing scheduler's own pool;+-- the last thread in the list will end up at the front of the pool;+-- wake up as many sleeping schedulers as threads added.+putThreads :: [Thread m] -> ThreadM m ()+putThreads threads = do+  all_pools@(my_pool:_) <- getPools+  liftIO $ mapM_ (pushFrontIO $ snd my_pool) threads+  liftSparkM $ wakeupSched (min (length all_pools) (length threads))
+ src/Control/Parallel/HdpH/Internal/Type/GRef.hs view
@@ -0,0 +1,39 @@+-- Global references; types+--+-- Author: Patrick Maier+-----------------------------------------------------------------------------++module Control.Parallel.HdpH.Internal.Type.GRef+  ( -- * global references+    GRef(..),++    -- * registry for global references+    GRefReg(..)+  ) where++import Prelude+import Data.Map (Map)++import Control.Parallel.HdpH.Internal.Location (NodeId)+import Control.Parallel.HdpH.Internal.Misc (AnyType)+++-----------------------------------------------------------------------------+-- global references++-- Global references, comprising a locally unique slot on the hosting node+-- and the hosting node's ID. Note that the slot, represented by a positive+-- integer, must be unique over the life time of the hosting node (ie. it+-- can not be reused). Note also that the type constructor 'GRef' takes+-- a phantom type parameter, tracking the type of the referenced object.+data GRef a = GRef { slot :: !Integer,+                     at   :: !NodeId }+++-----------------------------------------------------------------------------+-- registry for global references++-- Registry, comprising of the most recently allocated slot and a table+-- mapping slots to objects (wrapped in an existential type).+data GRefReg = GRefReg { lastSlot :: !Integer,+                         table    :: Map Integer AnyType }
+ src/Control/Parallel/HdpH/Internal/Type/Location.hs view
@@ -0,0 +1,47 @@+-- Locations; types; wrapper module+--+-- Author: Rob Stewart, Patrick Maier+-----------------------------------------------------------------------------++{-# LANGUAGE DeriveDataTypeable #-}  -- for defining exceptions+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}++module Control.Parallel.HdpH.Internal.Type.Location+  ( -- * node IDs (and their constitutent parts)+    NodeId,++    -- * node ID exception+    MyNodeException(..)  -- instances: Exception, Show, Typeable+  ) where++import Prelude+import Control.DeepSeq (NFData)+import Control.Exception (Exception)+import Data.Serialize (Serialize)+import Data.Typeable (Typeable)+import Network.Transport (EndPointAddress(..))+++-----------------------------------------------------------------------------+-- node IDs (should be abstract and hyperstrict outwith this module)+-- HACK: identify node ID with end point++-- | A 'NodeId' identifies a node (that is, an OS process running HdpH).+-- A 'NodeId' should be thought of as an abstract identifier (though it is+-- not currently abstract) which instantiates the classes 'Eq', 'Ord',+-- 'Show', 'NFData' and 'Serialize'.+type NodeId = EndPointAddress++deriving instance NFData NodeId+deriving instance Serialize NodeId+++-----------------------------------------------------------------------------+-- exception raised when ID of this node is not set++data MyNodeException = NodeIdUnset+                       deriving (Show, Typeable)++instance Exception MyNodeException
+ src/Control/Parallel/HdpH/Internal/Type/Par.hs view
@@ -0,0 +1,44 @@+-- Par monad and thread representation; types+--+-- Author: Patrick Maier+-----------------------------------------------------------------------------++module Control.Parallel.HdpH.Internal.Type.Par+  ( -- * Par monad, threads and sparks+    ParM(..),+    Thread(..),+    Spark       -- synonym: Spark m = Closure (ParM m ())+  ) where++import Prelude++import Control.Parallel.HdpH.Closure (Closure)+++-----------------------------------------------------------------------------+-- Par monad, based on ideas from+--   [1] Claessen "A Poor Man's Concurrency Monad", JFP 9(3), 1999.+--   [2] Marlow et al. "A monad for deterministic parallelism". Haskell 2011.++-- 'ParM m' is a continuation monad, specialised to the return type 'Thread m';+-- 'm' abstracts a monad encapsulating the underlying state.+newtype ParM m a = Par { unPar :: (a -> Thread m) -> Thread m }++instance Functor (ParM m) where+    fmap f p = Par $ \ c -> unPar p (c . f)++-- The Monad instance is where we differ from Control.Monad.Cont,+-- the difference being the use of strict application ($!).+instance Monad (ParM m) where+    return a = Par $ \ c -> c $! a+    p >>= k  = Par $ \ c -> unPar p $ \ a -> unPar (k $! a) c+++-- A thread is determined by its actions, as described in this data type.+-- In [2] this type is called 'Trace'.+newtype Thread m = Atom (m (Maybe (Thread m)))  -- atomic action (in monad 'm')+                                                -- result is next action, maybe+++-- A spark is a 'Par' comp returning '()', wrapped into an explicit closure.+type Spark m = Closure (ParM m ())
+ src/Control/Parallel/HdpH/Strategies.hs view
@@ -0,0 +1,754 @@+-- Strategies (and skeletons) in the Par monad+--+-- Author: Patrick Maier+-----------------------------------------------------------------------------++{-# LANGUAGE ScopedTypeVariables #-}  -- for type annotations in Static decl+{-# LANGUAGE FlexibleInstances #-}    -- req'd for some 'ToClosure' instances+{-# LANGUAGE TemplateHaskell #-}      -- req'd for 'mkClosure', etc++module Control.Parallel.HdpH.Strategies+  ( -- * Strategy type+    Strategy,+    using,+    +    -- * Basic sequential strategies+    r0,+    rseq,+    rdeepseq,++    -- * Fully forcing Closure strategy+    forceC,+    forceCC,+    ForceCC(+      locForceCC+    ),+    StaticForceCC,+    staticForceCC,++    -- * Proto-strategies for generating parallelism+    ProtoStrategy,+    sparkClosure,+    pushClosure,++    -- * Strategies for lists+    evalList,+    evalClosureListClosure,+    parClosureList,+    pushClosureList,+    pushRandClosureList,++    -- ** Clustering strategies+    parClosureListClusterBy,+    parClosureListChunked,+    parClosureListSliced,++    -- * Task farm skeletons+    -- | Task farm skeletons are parallel maps, applying a function to a list+    -- in parallel. For technical reasons, the function to be applied must+    -- wrapped in a Closure (ie. a function Closure).++    -- ** Lazy task placement+    parMap,+    parMapNF,+    parMapChunked,+    parMapChunkedNF,+    parMapSliced,+    parMapSlicedNF,++    parClosureMapM,+    parMapM,+    parMapM_,++    -- ** Round-robin task placement+    pushMap,+    pushMapNF,++    pushClosureMapM,+    pushMapM,+    pushMapM_,++    -- ** Random task placement+    pushRandClosureMapM,+    pushRandMapM,+    pushRandMapM_,++    -- * Divide and conquer skeletons+    divideAndConquer,+    parDivideAndConquer,+    pushDivideAndConquer,++    -- * This module's Static declaration+    declareStatic    -- :: StaticDecl+  ) where++import Prelude+import Control.DeepSeq (NFData, deepseq)+import Control.Monad (zipWithM, zipWithM_)+import Data.Functor ((<$>))+import Data.List (transpose)+import Data.Monoid (mconcat)+import System.Random (randomRIO)++import Control.Parallel.HdpH +       (Par, io, fork, pushTo, spark, new, get, glob, rput,+        NodeId, IVar, GIVar,+        Env, LocT, here,+        Closure, unClosure, mkClosure, mkClosureLoc, apC, compC,+        ToClosure(locToClosure), toClosure, forceClosure,+        StaticToClosure, staticToClosure,+        Static, static, static_, staticLoc_,+        StaticDecl, declare)+import qualified Control.Parallel.HdpH as HdpH (declareStatic)+++-----------------------------------------------------------------------------+-- Static declaration++-- 'ToClosure' instance required for 'evalClosureListClosure'+instance ToClosure [Closure a] where locToClosure = $(here)++instance ForceCC (Closure a) where locForceCC = $(here)++declareStatic :: StaticDecl+declareStatic =+  mconcat+    [HdpH.declareStatic,  -- 'Static' decl of imported modules+     declare (staticToClosure :: forall a . StaticToClosure [Closure a]),+     declare (staticForceCC :: forall a . StaticForceCC (Closure a)),+     declare $(static 'sparkClosure_abs),+     declare $(static 'pushClosure_abs),+     declare $(static_ 'evalClosureListClosure),+     declare $(static 'parClosureMapM_abs),+     declare $(static 'parMapM_abs),+     declare $(static_ 'constReturnUnit),+     declare $(static 'parDivideAndConquer_abs),+     declare $(static 'pushDivideAndConquer_abs)]+++-----------------------------------------------------------------------------+-- Strategy type++-- | A @'Strategy'@ for type @a@ is a (semantic) identity in the @'Par'@ monad.+-- For an elaboration of this concept (in the context of the @Eval@ monad)+-- see the paper:+--   Marlow et al.+--   /Seq no more: Better Strategies for parallel Haskell./+--   Haskell 2010.+type Strategy a = a -> Par a++-- | Strategy application is actual application (in the @'Par'@ monad).+using :: a -> Strategy a -> Par a+using = flip ($)+++-----------------------------------------------------------------------------+-- Basic sequential strategies (polymorphic);+-- these are exactly as in the "Seq no more" paper.++-- | /Do Nothing/ strategy.+r0 :: Strategy a+r0 = return++-- | /Evaluate head-strict/ strategy; probably not very useful in HdpH.+rseq :: Strategy a+rseq x = x `seq` return x -- Order of eval irrelevant due to 2nd arg converging++-- | /Evaluate fully/ strategy.+rdeepseq :: (NFData a) => Strategy a+rdeepseq x = x `deepseq` return x  -- Order of eval irrelevant (2nd arg conv)+++-----------------------------------------------------------------------------+-- fully forcing strategy for Closures++-- | @forceC@ is the fully forcing @'Closure'@ strategy, ie. it fully normalises+-- the thunk inside an explicit @'Closure'@.+-- Importantly, @forceC@ alters the serialisable @'Closure'@ represention+-- so that serialisation will not force the @'Closure'@ again.+forceC :: (NFData a, ToClosure a) => Strategy (Closure a)+forceC clo = return $! forceClosure clo++-- Note that 'forceC clo' does not have the same effect as+-- * 'rdeepseq clo' (because 'forceC' changes the closure representation), or+-- * 'rdeepseq $ toClosure $ unClosure clo' (because 'forceC' does not force+--   the serialised environment of its result), or+-- * 'rdeepseq clo >> return (toClosure (unClosure clo))' (because this does+--   hang on to the old serialisable environment whereas 'forceC' replaces+--   the old enviroment with a new one).+--+-- Note that it does not make sense to construct a variant of 'forceC' that+-- would evaluate the thunk inside a Closure head-strict only. The reason is+-- that serialising such a Closure would turn it into a fully forced one.+++-----------------------------------------------------------------------------+-- fully forcing Closure strategy wrapped into a Closure+--+-- To enable passing strategy @'forceC'@ around in distributed contexts, it+-- has to be wrapped into a @'Closure'@. That is, this module should export+--+-- > forceCC :: (NFData a, ToClosure a) => Closure (Strategy (Closure a))+--+-- The tutorial in module 'Control.Parallel.HdpH.Closure' details how to cope+-- with the type class constraint by introducing a new class.++-- | @forceCC@ is a @'Closure'@ wrapping the fully forcing Closure strategy+-- @'forceC'@; see the tutorial in module 'Control.Parallel.HdpH.Closure' for+-- details on the implementation of @forceCC@.+forceCC :: (ForceCC a) => Closure (Strategy (Closure a))+forceCC = $(mkClosureLoc [| forceC |]) locForceCC++-- | Indexing class, recording which types support @'forceCC'@; see the+-- tutorial in module 'Control.Parallel.HdpH.Closure' for a more thorough+-- explanation.+class (NFData a, ToClosure a) => ForceCC a where+  -- | Only method of class @ForceCC@, recording the source location+  -- where an instance of @ForceCC@ is declared.+  locForceCC :: LocT (Strategy (Closure a))+                -- The phantom type argument of 'LocT' is the type of the thunk+                -- that is quoted and passed to 'mkClosureLoc' above.++-- | Type synonym for declaring the @'Static'@ deserialisers required by+-- @'ForceCC'@ instances; see the tutorial in module+-- 'Control.Parallel.HdpH.Closure' for a more thorough explanation.+type StaticForceCC a = Static (Env -> Strategy (Closure a))++-- | @'Static'@ deserialiser required by a 'ForceCC' instance; see the tutorial+-- in module 'Control.Parallel.HdpH.Closure' for a more thorough explanation.+staticForceCC :: (ForceCC a) => StaticForceCC a+staticForceCC = $(staticLoc_ 'forceC) locForceCC+++-----------------------------------------------------------------------------+-- proto-strategies for generating parallelism++-- | A @'ProtoStrategy'@ is almost a @'Strategy'@.+-- More precisely, a @'ProtoStrategy'@ for type @a@ is a /delayed/ (semantic)+-- identity function in the @'Par'@ monad, ie. it returns an @'IVar'@ (rather+-- than a term) of type @a@.+type ProtoStrategy a = a -> Par (IVar a)+++-- | @sparkClosure clo_strat@ is a @'ProtoStrategy'@ that sparks a @'Closure'@;+-- evaluation of the sparked @'Closure'@ is governed by the strategy+-- @'unClosure' clo_strat@.+sparkClosure :: Closure (Strategy (Closure a)) ->+                  ProtoStrategy (Closure a)+sparkClosure clo_strat clo = do+  v <- new+  gv <- glob v+  spark $(mkClosure [| sparkClosure_abs (clo, clo_strat, gv) |])+  return v++sparkClosure_abs :: (Closure a,+                     Closure (Strategy (Closure a)),+                     GIVar (Closure a))+                 -> Par ()+sparkClosure_abs (clo, clo_strat, gv) =+  (clo `using` unClosure clo_strat) >>= rput gv+++-- | @pushClosure clo_strat n@ is a @'ProtoStrategy'@ that pushes a @'Closure'@+-- to be executed in a new thread on node @n@;+-- evaluation of the pushed @'Closure'@ is governed by the strategy+-- @'unClosure' clo_strat@.+pushClosure :: Closure (Strategy (Closure a)) -> NodeId ->+                 ProtoStrategy (Closure a)+pushClosure clo_strat node clo = do+  v <- new+  gv <- glob v+  pushTo $(mkClosure [| pushClosure_abs (clo, clo_strat, gv) |]) node+  return v++pushClosure_abs :: (Closure a,+                    Closure (Strategy (Closure a)),+                    GIVar (Closure a))+                -> Par ()+pushClosure_abs (clo, clo_strat, gv) =+  fork $ (clo `using` unClosure clo_strat) >>= rput gv+++------------------------------------------------------------------------------+-- strategies for lists++-- 'evalList' is a (type-restricted) monadic map; should be suitably+-- generalisable for all data structures that support mapping over+-- | Evaluate each element of a list according to the given strategy.+evalList :: Strategy a -> Strategy [a]+evalList _strat []     = return []+evalList  strat (x:xs) = do x' <- strat x+                            xs' <- evalList strat xs+                            return (x':xs')+++-- | Specialisation of @'evalList'@ to a list of Closures (wrapped in a+-- Closure). Useful for building clustering strategies.+evalClosureListClosure :: Strategy (Closure a) -> Strategy (Closure [Closure a])+evalClosureListClosure strat clo =+  toClosure <$> (unClosure clo `using` evalList strat)+++-- | Evaluate each element of a list of Closures in parallel according to+-- the given strategy (wrapped in a Closure). Work is distributed by+-- lazy work stealing.+parClosureList :: Closure (Strategy (Closure a)) -> Strategy [Closure a]+parClosureList clo_strat xs = mapM (sparkClosure clo_strat) xs >>=+                              mapM get+++-- | Evaluate each element of a list of Closures in parallel according to+-- the given strategy (wrapped in a Closure). Work is pushed round-robin+-- to the given list of nodes.+pushClosureList :: Closure (Strategy (Closure a))+                -> [NodeId]+                -> Strategy [Closure a]+pushClosureList clo_strat nodes xs =+  zipWithM (pushClosure clo_strat) (cycle nodes) xs >>=+  mapM get+++-- | Evaluate each element of a list of Closures in parallel according to+-- the given strategy (wrapped in a Closure). Work is pushed randomly+-- to the given list of nodes.+pushRandClosureList :: Closure (Strategy (Closure a))+                    -> [NodeId]+                    -> Strategy [Closure a]+pushRandClosureList clo_strat nodes xs =+  mapM (\ x -> do { node <- rand; pushClosure clo_strat node x}) xs >>=+  mapM get+    where+      rand :: Par NodeId+      rand = (nodes !!) <$> io (randomRIO (0, length nodes - 1))+++------------------------------------------------------------------------------+-- clustering strategies++-- generic clustering strategy combinator+evalClusterBy :: (a -> b) -> (b -> a) -> Strategy b -> Strategy a+evalClusterBy cluster uncluster strat x =+  uncluster <$> (cluster x `using` strat)+++-- | @parClosureListClusterBy cluster uncluster@ is a generic parallel+-- clustering strategy combinator for lists of Closures, evaluating+-- clusters generated by @cluster@ in parallel.+-- Clusters are distributed by lazy work stealing.+-- The function @uncluster@ must be a /left inverse/ of @cluster@,+-- that is @uncluster . cluster@ must be the identity.+parClosureListClusterBy :: ([Closure a] -> [[Closure a]])+                        -> ([[Closure a]] -> [Closure a])+                        -> Closure (Strategy (Closure a))+                        -> Strategy [Closure a]+parClosureListClusterBy cluster uncluster clo_strat =+  evalClusterBy cluster' uncluster' strat'+    where cluster'   = map toClosure . cluster+          uncluster' = uncluster . map unClosure+       -- strat' :: Strategy [Closure [Closure a]]+          strat' = parClosureList clo_strat''+       -- clo_strat'' :: Closure (Strategy (Closure [Closure a]))+          clo_strat'' =+            $(mkClosure [| evalClosureListClosure |]) `apC` clo_strat+++-- | @parClosureListChunked n@ evaluates chunks of size @n@ of a list of+-- Closures in parallel according to the given strategy (wrapped in a Closure).+-- Chunks are distributed by lazy work stealing.+-- For instance, dividing the list @[c1,c2,c3,c4,c5]@ into chunks of size 3+-- results in the following list of chunks @[[c1,c2,c3], [c4,c5]]@.+parClosureListChunked :: Int+                      -> Closure (Strategy (Closure a))+                      -> Strategy [Closure a]+parClosureListChunked n = parClosureListClusterBy (chunk n) unchunk+++-- | @parClosureListSliced n@ evaluates @n@ slices of a list of Closures in+-- parallel according to the given strategy (wrapped in a Closure).+-- Slices are distributed by lazy work stealing.+-- For instance, dividing the list @[c1,c2,c3,c4,c5]@ into 3 slices+-- results in the following list of slices @[[c1,c4], [c2,c5], [c3]]@.+parClosureListSliced :: Int+                     -> Closure (Strategy (Closure a))+                     -> Strategy [Closure a]+parClosureListSliced n = parClosureListClusterBy (slice n) unslice+++-- clustering functions: chunking and slicing+chunk :: Int -> [a] -> [[a]]+chunk n | n <= 0    = chunk 1+        | otherwise = go+            where+              go [] = []+              go xs = ys : go zs where (ys,zs) = splitAt n xs++unchunk :: [[a]] -> [a]+unchunk = concat++slice :: Int -> [a] -> [[a]]+slice n = transpose . chunk n++unslice :: [[a]] -> [a]+unslice = concat . transpose+++------------------------------------------------------------------------------+-- skeletons++-- | Task farm, evaluates tasks (function Closure applied to an element+-- of the input list) in parallel and according to the given strategy (wrapped+-- in a Closure).+-- Note that @parMap@ should only be used if the terms in the input list are+-- already in normal form, as they may be forced sequentially otherwise.+parMap :: (ToClosure a)+       => Closure (Strategy (Closure b))+       -> Closure (a -> b)+       -> [a]+       -> Par [b]+parMap clo_strat clo_f xs =+  do clo_ys <- map f clo_xs `using` parClosureList clo_strat+     return $ map unClosure clo_ys+       where f = apC clo_f+             clo_xs = map toClosure xs++-- | Specialisation of @'parMap'@ to the fully forcing Closure strategy.+-- That is, @parMapNF@ forces every element of the output list to normalform.+parMapNF :: (ToClosure a, ForceCC b)+         => Closure (a -> b)+         -> [a]+         -> Par [b]+parMapNF = parMap forceCC+++-- | Chunking task farm, divides the input list into chunks of given size+-- and evaluates tasks (function Closure mapped on a chunk of the input list) +-- in parallel and according to the given strategy (wrapped in a Closure).+-- @parMapChunked@ should only be used if the terms in the input list+-- are already in normal form.+parMapChunked :: (ToClosure a)+              => Int+              -> Closure (Strategy (Closure b))+              -> Closure (a -> b)+              -> [a]+              -> Par [b]+parMapChunked n clo_strat clo_f xs =+  do clo_ys <- map f clo_xs `using` parClosureListChunked n clo_strat+     return $ map unClosure clo_ys+       where f = apC clo_f+             clo_xs = map toClosure xs++-- | Specialisation of @'parMapChunked'@ to the fully forcing Closure strategy.+parMapChunkedNF :: (ToClosure a, ForceCC b)+                => Int+                -> Closure (a -> b)+                -> [a]+                -> Par [b]+parMapChunkedNF n = parMapChunked n forceCC+++-- | Slicing task farm, divides the input list into given number of slices+-- and evaluates tasks (function Closure mapped on a slice of the input list) +-- in parallel and according to the given strategy (wrapped in a Closure).+-- @parMapSliced@ should only be used if the terms in the input list+-- are already in normal form.+parMapSliced :: (ToClosure a)+             => Int+             -> Closure (Strategy (Closure b))+             -> Closure (a -> b)+             -> [a]+             -> Par [b]+parMapSliced n clo_strat clo_f xs =+  do clo_ys <- map f clo_xs `using` parClosureListSliced n clo_strat+     return $ map unClosure clo_ys+       where f = apC clo_f+             clo_xs = map toClosure xs++-- | Specialisation of @'parMapSliced'@ to the fully forcing Closure strategy.+parMapSlicedNF :: (ToClosure a, ForceCC b)+               => Int+               -> Closure (a -> b)+               -> [a]+               -> Par [b]+parMapSlicedNF n = parMapSliced n forceCC+++-- | Monadic task farm for Closures, evaluates tasks (@'Par'@-monadic function+-- Closure applied to a Closure of the input list) in parallel.+-- Note the absence of a strategy argument; strategies aren't needed because+-- they can be baked into the monadic function Closure.+parClosureMapM :: Closure (Closure a -> Par (Closure b))+               -> [Closure a]+               -> Par [Closure b]+parClosureMapM clo_f clo_xs =+  do vs <- mapM spawn clo_xs+     mapM get vs+       where+         spawn clo_x = do+           v <- new+           gv <- glob v+           spark $(mkClosure [| parClosureMapM_abs (clo_f, clo_x, gv) |])+           return v++parClosureMapM_abs :: (Closure (Closure a -> Par (Closure b)),+                       Closure a,+                       GIVar (Closure b))+                   -> Par ()+parClosureMapM_abs (clo_f, clo_x, gv) = unClosure clo_f clo_x >>= rput gv+++-- | Monadic task farm, evaluates tasks (@'Par'@-monadic function Closure+-- applied to an element of the input list) in parallel.+-- Note the absence of a strategy argument; strategies aren't needed because+-- they can be baked into the monadic function Closure.+-- @parMap@ should only be used if the terms in the input list are already+-- in normal form, as they may be forced sequentially otherwise.+parMapM :: (ToClosure a)+        => Closure (a -> Par (Closure b))+        -> [a]+        -> Par [b]+parMapM clo_f xs =+  do vs <- mapM spawn xs+     mapM (\ v -> unClosure <$> get v) vs+       where+         spawn x = do+           let clo_x = toClosure x+           v <- new+           gv <- glob v+           spark $(mkClosure [| parMapM_abs (clo_f, clo_x, gv) |])+           return v++parMapM_abs :: (Closure (a -> Par (Closure b)), +                Closure a, +                GIVar (Closure b)) +            -> Par ()+parMapM_abs (clo_f, clo_x, gv) = unClosure (clo_f `apC` clo_x) >>= rput gv+++-- | Specialisation of @'parMapM'@, not returning any result.+parMapM_ :: (ToClosure a)+         => Closure (a -> Par b)+         -> [a]+         -> Par ()+parMapM_ clo_f xs = mapM_ (spark . apC (termParC `compC` clo_f) . toClosure) xs+-- Note that applying the @'termParC'@ transformation is necessary because+-- @'spark'@ only accepts Closures of type @Par ()@.++-- terminal arrow in the Par monad, wrapped in a Closure+termParC :: Closure (a -> Par ())+termParC = $(mkClosure [| constReturnUnit |])++{-# INLINE constReturnUnit #-}+constReturnUnit :: a -> Par ()+constReturnUnit = const (return ())+++-- | Task farm like @'parMap'@ but pushes tasks in a round-robin fashion+-- to the given list of nodes.+pushMap :: (ToClosure a)+        => Closure (Strategy (Closure b))+        -> [NodeId]+        -> Closure (a -> b)+        -> [a]+        -> Par [b]+pushMap clo_strat nodes clo_f xs =+  do clo_ys <- map f clo_xs `using` pushClosureList clo_strat nodes+     return $ map unClosure clo_ys+       where f = apC clo_f+             clo_xs = map toClosure xs++-- | Task farm like @'parMapNF'@ but pushes tasks in a round-robin fashion+-- to the given list of nodes.+pushMapNF :: (ToClosure a, ForceCC b)+          => [NodeId]+          -> Closure (a -> b)+          -> [a]+          -> Par [b]+pushMapNF = pushMap forceCC+++-- | Monadic task farm for Closures like @'parClosureMapM'@ but pushes tasks+-- in a round-robin fashion to the given list of nodes.+pushClosureMapM :: [NodeId]+                -> Closure (Closure a -> Par (Closure b))+                -> [Closure a]+                -> Par [Closure b]+pushClosureMapM nodes clo_f clo_xs =+  do vs <- zipWithM spawn (cycle nodes) clo_xs+     mapM get vs+       where+         spawn node clo_x = do+           v <- new+           gv <- glob v+           pushTo $(mkClosure [| parClosureMapM_abs (clo_f, clo_x, gv) |]) node+           return v+++-- | Monadic task farm like @'parMapM'@ but pushes tasks+-- in a round-robin fashion to the given list of nodes.+pushMapM :: (ToClosure a)+         => [NodeId]+         -> Closure (a -> Par (Closure b))+         -> [a]+         -> Par [b]+pushMapM nodes clo_f xs =+  do vs <- zipWithM spawn (cycle nodes) xs+     mapM (\ v -> unClosure <$> get v) vs+       where+         spawn node x = do+           let clo_x = toClosure x+           v <- new+           gv <- glob v+           pushTo $(mkClosure [| parMapM_abs (clo_f, clo_x, gv) |]) node+           return v+++-- | Monadic task farm like @'parMapM_'@ but pushes tasks+-- in a round-robin fashion to the given list of nodes.+pushMapM_ :: (ToClosure a)+          => [NodeId]+          -> Closure (a -> Par b)+          -> [a]+          -> Par ()+pushMapM_ nodes clo_f xs =+  zipWithM_+    (\ node x -> pushTo (compC termParC clo_f `apC` toClosure x) node)+    (cycle nodes)+    xs+++-- | Monadic task farm for Closures like @'parClosureMapM'@+-- but pushes to random nodes on the given list.+pushRandClosureMapM :: [NodeId]+                    -> Closure (Closure a -> Par (Closure b))+                    -> [Closure a]+                    -> Par [Closure b]+pushRandClosureMapM nodes clo_f clo_xs =+  do vs <- mapM spawn clo_xs+     mapM get vs+       where+         rand = (nodes !!) <$> io (randomRIO (0, length nodes - 1))+         spawn clo_x = do+           v <- new+           gv <- glob v+           node <- rand+           pushTo $(mkClosure [| parClosureMapM_abs (clo_f, clo_x, gv) |]) node+           return v+++-- | Monadic task farm like @'parMapM'@+-- but pushes to random nodes on the given list.+pushRandMapM :: (ToClosure a)+             => [NodeId]+             -> Closure (a -> Par (Closure b))+             -> [a]+             -> Par [b]+pushRandMapM nodes clo_f xs =+  do vs <- mapM spawn xs+     mapM (\ v -> unClosure <$> get v) vs+       where+         rand = (nodes !!) <$> io (randomRIO (0, length nodes - 1))+         spawn x = do+           let clo_x = toClosure x+           v <- new+           gv <- glob v+           node <- rand+           pushTo $(mkClosure [| parMapM_abs (clo_f, clo_x, gv) |]) node+           return v+++-- | Monadic task farm like @'parMapM_'@+-- but pushes to random nodes on the given list.+pushRandMapM_ :: (ToClosure a)+              => [NodeId]+              -> Closure (a -> Par b)+              -> [a]+              -> Par ()+pushRandMapM_ nodes clo_f xs =+  mapM_ spawn xs+    where+      rand = (nodes !!) <$> io (randomRIO (0, length nodes - 1))+      spawn x = do+        node <- rand+        pushTo (compC termParC clo_f `apC` toClosure x) node+++-- | Sequential divide-and-conquer skeleton.+-- @didvideAndConquer trivial decompose combine f x@ repeatedly decomposes+-- the problem @x@ until trivial, applies @f@ to the trivial sub-problems+-- and combines the solutions.+divideAndConquer :: (a -> Bool)      -- isTrivial+                 -> (a -> [a])       -- decomposeProblem+                 -> (a -> [b] -> b)  -- combineSolutions+                 -> (a -> b)         -- trivialAlgorithm+                 -> a                -- problem+                 -> b+divideAndConquer trivial decompose combine f x+  | trivial x = f x+  | otherwise = combine x $ map solveRec (decompose x)+      where+        solveRec = divideAndConquer trivial decompose combine f+++-- | Parallel divide-and-conquer skeleton with lazy work distribution.+-- @parDivideAndConquer trivial_clo decompose_clo combine_clo f_clo x@ follows+-- the divide-and-conquer pattern of @'divideAndConquer'@ except that, for+-- technical reasons, all arguments are Closures.+parDivideAndConquer :: Closure (Closure a -> Bool)+                    -> Closure (Closure a -> [Closure a])+                    -> Closure (Closure a -> [Closure b] -> Closure b)+                    -> Closure (Closure a -> Par (Closure b))+                    -> Closure a+                    -> Par (Closure b)+parDivideAndConquer trivial_clo decompose_clo combine_clo f_clo x+  | trivial x = f x+  | otherwise = combine x <$> parClosureMapM solveRec_clo (decompose x)+      where+        trivial   = unClosure trivial_clo+        decompose = unClosure decompose_clo+        combine   = unClosure combine_clo+        f         = unClosure f_clo+        solveRec_clo =+          $(mkClosure [| parDivideAndConquer_abs+                           (trivial_clo, decompose_clo, combine_clo, f_clo) |])++parDivideAndConquer_abs :: (Closure (Closure a -> Bool),+                            Closure (Closure a -> [Closure a]),+                            Closure (Closure a -> [Closure b] -> Closure b),+                            Closure (Closure a -> Par (Closure b)))+                        -> Closure a -> Par (Closure b)+parDivideAndConquer_abs (trivial_clo, decompose_clo, combine_clo, f_clo) =+  parDivideAndConquer trivial_clo decompose_clo combine_clo f_clo+++-- | Parallel divide-and-conquer skeleton with eager random work distribution,+-- pushing work to the given list of nodes.+-- @pushDivideAndConquer nodes trivial_clo decompose_clo combine_clo f_clo x@+-- follows the divide-and-conquer pattern of @'divideAndConquer'@ except that,+-- for technical reasons, all arguments are Closures.+pushDivideAndConquer :: [NodeId]+                     -> Closure (Closure a -> Bool)+                     -> Closure (Closure a -> [Closure a])+                     -> Closure (Closure a -> [Closure b] -> Closure b)+                     -> Closure (Closure a -> Par (Closure b))+                     -> Closure a+                     -> Par (Closure b)+pushDivideAndConquer ns trivial_clo decompose_clo combine_clo f_clo x+  | trivial x = f x+  | otherwise = combine x <$> pushRandClosureMapM ns solveRec_clo (decompose x)+      where+        trivial   = unClosure trivial_clo+        decompose = unClosure decompose_clo+        combine   = unClosure combine_clo+        f         = unClosure f_clo+        solveRec_clo =+          $(mkClosure [| pushDivideAndConquer_abs+                           (ns,trivial_clo,decompose_clo,combine_clo,f_clo) |])++pushDivideAndConquer_abs :: ([NodeId],+                             Closure (Closure a -> Bool),+                             Closure (Closure a -> [Closure a]),+                             Closure (Closure a -> [Closure b] -> Closure b),+                             Closure (Closure a -> Par (Closure b)))+                         -> Closure a -> Par (Closure b)+pushDivideAndConquer_abs (ns, trivial_clo, decompose_clo, combine_clo, f_clo) =+  pushDivideAndConquer ns trivial_clo decompose_clo combine_clo f_clo
+ src/Test/HdpH/fib.hs view
@@ -0,0 +1,280 @@+-- Fibonacci numbers in HdpH+--+-- Author: Patrick Maier+-----------------------------------------------------------------------------++{-# LANGUAGE TemplateHaskell #-}  -- req'd for mkClosure, etc++module Main where++import Prelude+import Control.Exception (evaluate)+import Control.Monad (when)+import Data.Functor ((<$>))+import Data.List (elemIndex, stripPrefix)+import Data.Maybe (fromJust)+import Data.Monoid (mconcat)+import Data.Time.Clock (NominalDiffTime, diffUTCTime, getCurrentTime)+import System.Environment (getArgs)+import System.IO (stdout, stderr, hSetBuffering, BufferMode(..))+import System.Random (mkStdGen, setStdGen)++import Control.Parallel.HdpH+       (RTSConf(..), defaultRTSConf,+        Par, runParIO,+        allNodes, force, fork, spark, new, get, put, glob, rput,+        GIVar, NodeId,+        Closure, unClosure, mkClosure,+        toClosure, ToClosure(locToClosure),+        static, static_, StaticToClosure, staticToClosure,+        StaticDecl, declare, register, here)+import qualified Control.Parallel.HdpH as HdpH (declareStatic)+import Control.Parallel.HdpH.Strategies +       (parDivideAndConquer, pushDivideAndConquer)+import qualified Control.Parallel.HdpH.Strategies as Strategies (declareStatic)+++-----------------------------------------------------------------------------+-- 'Static' declaration++instance ToClosure Int where locToClosure = $(here)+instance ToClosure Integer where locToClosure = $(here)++declareStatic :: StaticDecl+declareStatic =+  mconcat+    [HdpH.declareStatic,         -- declare Static deserialisers+     Strategies.declareStatic,   -- from imported modules+     declare (staticToClosure :: StaticToClosure Int),+     declare (staticToClosure :: StaticToClosure Integer),+     declare $(static 'dist_fib_abs),+     declare $(static 'dnc_trivial_abs),+     declare $(static_ 'dnc_decompose),+     declare $(static_ 'dnc_combine),+     declare $(static_ 'dnc_f)]+++-----------------------------------------------------------------------------+-- sequential Fibonacci++fib :: Int -> Integer+fib n | n <= 1    = 1+      | otherwise = fib (n-1) + fib (n-2)+++-----------------------------------------------------------------------------+-- parallel Fibonacci; shared memory++par_fib :: Int -> Int -> Par Integer+par_fib seqThreshold n+  | n <= k    = force $ fib n+  | otherwise = do v <- new+                   let job = par_fib seqThreshold (n - 1) >>=+                             force >>=+                             put v+                   fork job+                   y <- par_fib seqThreshold (n - 2)+                   x <- get v+                   force $ x + y+  where k = max 1 seqThreshold+++-----------------------------------------------------------------------------+-- parallel Fibonacci; distributed memory++dist_fib :: Int -> Int -> Int -> Par Integer+dist_fib seqThreshold parThreshold n+  | n <= k    = force $ fib n+  | n <= l    = par_fib seqThreshold n+  | otherwise = do+      v <- new+      gv <- glob v+      spark $(mkClosure [| dist_fib_abs (seqThreshold, parThreshold, n, gv) |])+      y <- dist_fib seqThreshold parThreshold (n - 2)+      clo_x <- get v+      force $ unClosure clo_x + y+  where k = max 1 seqThreshold+        l = parThreshold++dist_fib_abs :: (Int, Int, Int, GIVar (Closure Integer)) -> Par ()+dist_fib_abs (seqThreshold, parThreshold, n, gv) =+  dist_fib seqThreshold parThreshold (n - 1) >>=+  force >>=+  rput gv . toClosure+++-----------------------------------------------------------------------------+-- parallel Fibonacci; distributed memory; using sparking d-n-c skeleton++spark_skel_fib :: Int -> Int -> Par Integer+spark_skel_fib seqThreshold n = unClosure <$> skel (toClosure n)+  where +    skel = parDivideAndConquer+             $(mkClosure [| dnc_trivial_abs (seqThreshold) |])+             $(mkClosure [| dnc_decompose |])+             $(mkClosure [| dnc_combine |])+             $(mkClosure [| dnc_f |])++dnc_trivial_abs :: (Int) -> (Closure Int -> Bool)+dnc_trivial_abs (seqThreshold) =+  \ clo_n -> unClosure clo_n <= max 1 seqThreshold++dnc_decompose =+  \ clo_n -> let n = unClosure clo_n in [toClosure (n-1), toClosure (n-2)]++dnc_combine =+  \ _ clos -> toClosure $ sum $ map unClosure clos++dnc_f =+  \ clo_n -> toClosure <$> (force $ fib $ unClosure clo_n)+++-----------------------------------------------------------------------------+-- parallel Fibonacci; distributed memory; using pushing d-n-c skeleton++push_skel_fib :: [NodeId] -> Int -> Int -> Par Integer+push_skel_fib nodes seqThreshold n = unClosure <$> skel (toClosure n)+  where +    skel = pushDivideAndConquer+             nodes+             $(mkClosure [| dnc_trivial_abs (seqThreshold) |])+             $(mkClosure [| dnc_decompose |])+             $(mkClosure [| dnc_combine |])+             $(mkClosure [| dnc_f |])+++-----------------------------------------------------------------------------+-- initialisation, argument processing and 'main'++-- time an IO action+timeIO :: IO a -> IO (a, NominalDiffTime)+timeIO action = do t0 <- getCurrentTime+                   x <- action+                   t1 <- getCurrentTime+                   return (x, diffUTCTime t1 t0)+++-- initialize random number generator+initrand :: Int -> IO ()+initrand seed = do+  when (seed /= 0) $ do+    setStdGen (mkStdGen seed)+++-- parse runtime system config options (+ seed for random number generator)+parseOpts :: [String] -> (RTSConf, Int, [String])+parseOpts args = go (defaultRTSConf, 0, args) where+  go :: (RTSConf, Int, [String]) -> (RTSConf, Int, [String])+  go (conf, seed, [])   = (conf, seed, [])+  go (conf, seed, s:ss) =+   case stripPrefix "-rand=" s of+   Just s  -> go (conf, read s, ss)+   Nothing ->+    case stripPrefix "-d" s of+    Just s  -> go (conf { debugLvl = read s }, seed, ss)+    Nothing ->+     case stripPrefix "-scheds=" s of+     Just s  -> go (conf { scheds = read s }, seed, ss)+     Nothing ->+      case stripPrefix "-wakeup=" s of+      Just s  -> go (conf { wakeupDly = read s }, seed, ss)+      Nothing ->+       case stripPrefix "-hops=" s of+       Just s  -> go (conf { maxHops = read s }, seed, ss)+       Nothing ->+        case stripPrefix "-maxFish=" s of+        Just s  -> go (conf { maxFish = read s }, seed, ss)+        Nothing ->+         case stripPrefix "-minSched=" s of+         Just s  -> go (conf { minSched = read s }, seed, ss)+         Nothing ->+          case stripPrefix "-minNoWork=" s of+          Just s  -> go (conf { minFishDly = read s }, seed, ss)+          Nothing ->+           case stripPrefix "-numProcs=" s of+           Just s  -> go (conf { numProcs = read s }, seed, ss)+           Nothing ->+            case stripPrefix "-maxNoWork=" s of+            Just s  -> go (conf { maxFishDly = read s }, seed, ss)+            Nothing ->+             (conf, seed, s:ss)+++-- parse (optional) arguments in this order: +-- * version to run+-- * argument to Fibonacci function+-- * threshold below which to execute sequentially+-- * threshold below which to use shared-memory parallelism+parseArgs :: [String] -> (Int, Int, Int, Int)+parseArgs []     = (defVers, defN, defSeqThreshold, defParThreshold)+parseArgs (s:ss) =+  let go :: Int -> [String] -> (Int, Int, Int, Int)+      go v []           = (v, defN,    defSeqThreshold, defParThreshold)+      go v [s1]         = (v, read s1, defSeqThreshold, defParThreshold)+      go v [s1,s2]      = (v, read s1, read s2,         read s2)+      go v (s1:s2:s3:_) = (v, read s1, read s2,         read s3)+  in case stripPrefix "v" s of+       Just s' -> go (read s') ss+       Nothing -> go defVers (s:ss)++-- defaults for optional arguments+defVers         =  2 :: Int  -- version+defN            = 40 :: Int  -- Fibonacci argument+defParThreshold = 30 :: Int  -- shared-memory threshold+defSeqThreshold = 30 :: Int  -- sequential threshold+++main :: IO ()+main = do+  hSetBuffering stdout LineBuffering+  hSetBuffering stderr LineBuffering+  register declareStatic+  opts_args <- getArgs+  let (conf, seed, args) = parseOpts opts_args+  let (version, n, seqThreshold, parThreshold) = parseArgs args+  initrand seed+  case version of+      0 -> do (x, t) <- timeIO $ evaluate+                          (fib n)+              putStrLn $+                "{v0} fib " ++ show n ++ " = " ++ show x +++                " {runtime=" ++ show t ++ "}"+      1 -> do (output, t) <- timeIO $ evaluate =<< runParIO conf+                               (par_fib seqThreshold n)+              case output of+                Just x  -> putStrLn $+                             "{v1, " ++ +                             "seqThreshold=" ++ show seqThreshold ++ "} " +++                             "fib " ++ show n ++ " = " ++ show x +++                             " {runtime=" ++ show t ++ "}"+                Nothing -> return ()+      2 -> do (output, t) <- timeIO $ evaluate =<< runParIO conf+                               (dist_fib seqThreshold parThreshold n)+              case output of+                Just x  -> putStrLn $+                             "{v2, " +++                             "seqThreshold=" ++ show seqThreshold ++ ", " +++                             "parThreshold=" ++ show parThreshold ++ "} " +++                             "fib " ++ show n ++ " = " ++ show x +++                             " {runtime=" ++ show t ++ "}"+                Nothing -> return ()+      3 -> do (output, t) <- timeIO $ evaluate =<< runParIO conf+                               (spark_skel_fib seqThreshold n)+              case output of+                Just x  -> putStrLn $+                             "{v3, " +++                             "seqThreshold=" ++ show seqThreshold ++ "} " +++                             "fib " ++ show n ++ " = " ++ show x +++                             " {runtime=" ++ show t ++ "}"+                Nothing -> return ()+      4 -> do (output, t) <- timeIO $ evaluate =<< runParIO conf+                               (allNodes >>= \ nodes ->+                                push_skel_fib nodes seqThreshold n)+              case output of+                Just x  -> putStrLn $+                             "{v4, " +++                             "seqThreshold=" ++ show seqThreshold ++ "} " +++                             "fib " ++ show n ++ " = " ++ show x +++                             " {runtime=" ++ show t ++ "}"+                Nothing -> return ()+      _ -> return ()
+ src/Test/HdpH/hello.hs view
@@ -0,0 +1,111 @@+-- Hello World in HdpH+--+-- Author: Patrick Maier+-----------------------------------------------------------------------------++{-# LANGUAGE TemplateHaskell #-}++module Main where++import Prelude+import Data.List (stripPrefix)+import Data.Monoid (mconcat)+import System.Environment (getArgs)+import System.IO (stdout, stderr, hSetBuffering, BufferMode(..))+import System.Mem (performGC)++import Control.Parallel.HdpH+       (RTSConf(..), defaultRTSConf,+        Par, runParIO_,+        myNode, allNodes, io, pushTo, new, get, glob, rput,+        NodeId, IVar, GIVar,+        Closure, mkClosure,+        toClosure, ToClosure(locToClosure),+        static, StaticToClosure, staticToClosure,+        StaticDecl, declare, register, here)+import qualified Control.Parallel.HdpH as HdpH (declareStatic)+++-----------------------------------------------------------------------------+-- Static declaration++instance ToClosure () where locToClosure = $(here)++declareStatic :: StaticDecl+declareStatic = mconcat [HdpH.declareStatic,+                         declare (staticToClosure :: StaticToClosure ()),+                         declare $(static 'hello_abs)]+++-----------------------------------------------------------------------------+-- Hello World code++hello_world :: Par ()+hello_world = do+  master <- myNode+  io $ putStrLn $ "Master " ++ show master ++ " wants to know: Who is here?"+  world <- allNodes+  vs <- mapM push_hello world+  mapM_ get vs+    where+      push_hello :: NodeId -> Par (IVar (Closure ()))+      push_hello node = do+        v <- new+        done <- glob v+        pushTo $(mkClosure [| hello_abs done |]) node+        return v++hello_abs :: GIVar (Closure ()) -> Par ()+hello_abs done = do+  here <- myNode+  io $ putStrLn $ "Hello from " ++ show here+  rput done $ toClosure ()+++-----------------------------------------------------------------------------+-- initialisation, argument processing and 'main'++-- parse runtime system config options+parseOpts :: [String] -> (RTSConf, [String])+parseOpts args = go (defaultRTSConf, args) where+  go :: (RTSConf, [String]) -> (RTSConf, [String])+  go (conf, [])   = (conf, [])+  go (conf, s:ss) =+   case stripPrefix "-d" s of+   Just s  -> go (conf { debugLvl = read s }, ss)+   Nothing ->+    case stripPrefix "-scheds=" s of+    Just s  -> go (conf { scheds = read s }, ss)+    Nothing ->+     case stripPrefix "-wakeup=" s of+     Just s  -> go (conf { wakeupDly = read s }, ss)+     Nothing ->+      case stripPrefix "-hops=" s of+      Just s  -> go (conf { maxHops = read s }, ss)+      Nothing ->+       case stripPrefix "-maxFish=" s of+       Just s  -> go (conf { maxFish = read s }, ss)+       Nothing ->+        case stripPrefix "-minSched=" s of+        Just s  -> go (conf { minSched = read s }, ss)+        Nothing ->+         case stripPrefix "-minNoWork=" s of+         Just s  -> go (conf { minFishDly = read s }, ss)+         Nothing ->+          case stripPrefix "-maxNoWork=" s of+          Just s  -> go (conf { maxFishDly = read s }, ss)+          Nothing ->+           case stripPrefix "-numProcs=" s of+           Just s  -> go (conf { numProcs = read s }, ss)+           Nothing ->+            (conf, s:ss)+++main :: IO ()+main = do+  hSetBuffering stdout LineBuffering+  hSetBuffering stderr LineBuffering+  register declareStatic+  opts_args <- getArgs+  let (conf, _args) = parseOpts opts_args+  runParIO_ conf hello_world
+ src/Test/HdpH/nbody.hs view
@@ -0,0 +1,497 @@+-- N-body simulation; simple parallelisation of all pairs algorithm+--+-- Author: Patrick Maier+-----------------------------------------------------------------------------++{-# LANGUAGE FlexibleInstances #-}  -- req'd for some ToClosure instances+{-# LANGUAGE TemplateHaskell #-}    -- req'd for mkClosure, etc++module Main where++import Prelude+import Control.Applicative ((<$>), (<*>))+import Control.Exception (evaluate)+import Control.Monad (replicateM, when, (>=>))+import Control.DeepSeq (NFData(..), deepseq)+import Data.List (elemIndex, stripPrefix)+import Data.List (delete, foldl', tails, transpose)+import Data.Maybe (fromJust)+import Data.Monoid (mconcat)+import Data.Serialize (Serialize)+import qualified Data.Serialize (put, get)+import Data.Time.Clock (NominalDiffTime, diffUTCTime, getCurrentTime)+import System.Environment (getArgs)+import System.IO (stdout, stderr, hSetBuffering, BufferMode(..))+import System.Random (mkStdGen, setStdGen, randomIO)++import Control.Parallel.HdpH +       (RTSConf(..), defaultRTSConf,+        Par, runParIO,+        NodeId,+        myNode, allNodes, force, fork, new, get, put,+        mkClosure,+        toClosure, ToClosure(locToClosure),+        static, StaticToClosure, staticToClosure,+        StaticDecl, declare, register, here)+import qualified Control.Parallel.HdpH as HdpH (declareStatic)+import Control.Parallel.HdpH.Strategies +       (ForceCC(locForceCC), parMapNF, pushMapNF,+        StaticForceCC, staticForceCC)+import qualified Control.Parallel.HdpH.Strategies as Strategies (declareStatic)+++-----------------------------------------------------------------------------+-- Static declaration++instance ToClosure [Body] where locToClosure = $(here)+instance ToClosure [Vector3] where locToClosure = $(here)+instance ForceCC [Vector3] where locForceCC = $(here)++declareStatic :: StaticDecl+declareStatic =+  mconcat+    [HdpH.declareStatic,+     Strategies.declareStatic,+     declare (staticToClosure :: StaticToClosure [Body]),+     declare (staticToClosure :: StaticToClosure [Vector3]),+     declare (staticForceCC :: StaticForceCC [Vector3]),+     declare $(static 'part_dvs)]+++-----------------------------------------------------------------------------+-- 3D vectors and scalars++type Scalar = Double++data Vector3 = V3 {-# UNPACK #-} !Double+                  {-# UNPACK #-} !Double+                  {-# UNPACK #-} !Double+             deriving (Eq, Show)++instance NFData Vector3 where+  rnf (V3 x y z) = rnf x `seq` rnf y `seq` rnf z++instance Serialize Vector3 where+  put (V3 x y z) = Data.Serialize.put x >>+                   Data.Serialize.put y >>+                   Data.Serialize.put z+  get = do x <- Data.Serialize.get+           y <- Data.Serialize.get+           z <- Data.Serialize.get+           return $ V3 x y z+++randomVector3 :: IO Vector3+randomVector3 = do x <- randomIO+                   y <- randomIO+                   z <- randomIO+                   return $ V3 x y z+++{-# INLINE vzero #-}+vzero :: Vector3+vzero = V3 0.0 0.0 0.0+++-- vector sum and difference+{-# INLINE (.+.) #-}+{-# INLINE (.-.) #-}+(.+.), (.-.) :: Vector3 -> Vector3 -> Vector3+(V3 x1 x2 x3) .+. (V3 y1 y2 y3) = V3 (x1 + y1) (x2 + y2) (x3 + y3)+(V3 x1 x2 x3) .-. (V3 y1 y2 y3) = V3 (x1 - y1) (x2 - y2) (x3 - y3)++{-# INLINE vsum #-}+{-# INLINE vsum1 #-}+vsum, vsum1 :: [Vector3] -> Vector3+vsum vs = foldl' (.+.) vzero vs+vsum1 (v:vs) = foldl' (.+.) v vs+++-- scalar multiplication+{-# INLINE (*.) #-}+(*.) :: Scalar -> Vector3 -> Vector3+k *. (V3 x1 x2 x3) = V3 (k * x1) (k * x2) (k * x3)+++-- inner product+{-# INLINE (.*.) #-}+(.*.) :: Vector3 -> Vector3 -> Scalar+(V3 x1 x2 x3) .*. (V3 y1 y2 y3) = (x1 * y1) + (x2 * y2) + (x3 * y3)+++-----------------------------------------------------------------------------+-- N-body data representation++-- a body consists of position and mass (and position is key)+data Body = Body {+              pos  :: {-# UNPACK #-} !Vector3,  -- position [m]+              mass :: {-# UNPACK #-} !Scalar }  -- mass [kg]+            deriving (Show)++instance Eq Body where+  body1 == body2 = pos body1 == pos body2++instance NFData Body where+  rnf body = rnf (pos body) `seq` rnf (mass body)++instance Serialize Body where+  put body = Data.Serialize.put (pos body) >>+             Data.Serialize.put (mass body)+  get = do x <- Data.Serialize.get+           m <- Data.Serialize.get+           return $ Body { pos = x, mass = m }+++-- velocity [m/s]+type Vel = Vector3++-- differential velocity (aka acceleration) [m/s]+type DeltaVel = Vector3+++-- The N-body problem is represented by two vectors (lists of the same length)+-- of bodies and velocities+type NBody = [Body]+type NVel  = [Vel]+type NBodyConf = (NBody, NVel)++-- Differential velocities for a N-body problems (again a vector)+type NDeltaVel = [DeltaVel]+++-- Generate a random N-body instance (fully forced)+randomNBodyConf :: Int -> IO NBodyConf+randomNBodyConf n = do+  bodies <- replicateM n (Body <$> randomVector3 <*> randomIO)+  vs     <- replicateM n randomVector3+  return $ forceNF $ (bodies, vs)+++-----------------------------------------------------------------------------+-- energy innate to a N-body configuration++-- Newton's gravitational constant +g :: Scalar+g = 6.674/100000000000  -- 6.674 * 10^(-11) N m^2 / kg^2+++energy :: NBodyConf -> Scalar+energy (bodies, vs) =+  sum (zipWith eKin bodies vs) + sum (map ePots (tails bodies))+    where+      ePots []            = 0+      ePots (body:bodies) = sum (map (ePot body) bodies)++eKin :: Body -> Vel -> Scalar+eKin body v = 0.5 * mass body * (v .*. v)++ePot :: Body -> Body -> Scalar+ePot body1 body2 = g * mass body1 * mass body2 / dist+                     where+                       dist = sqrt (d .*. d)+                       d = pos body1 .-. pos body2+++-----------------------------------------------------------------------------+-- sequential N-body computation++-- time step ("delta t") of 1 ms+dt :: Scalar+dt = 0.001+++-- softening factor (avoiding distances becoming too small)+eps :: Scalar+eps = 0.01+++-- advance all bodies by 1 time step; input and output are chunks of the+-- given N-body problem (not lists of independent N-body problems);+-- outermost 'map' may be parallelised+advance :: [NBodyConf] -> [NBodyConf]+advance confs = forceNF confs'+  where+    bss = map fst confs+    confs' = map (\ (bodies, vs) ->+                    unzip $ zipWith3 update bodies vs $ dvs bodies bss) confs++-- update velocity and position of bodies+update :: Body -> Vel -> DeltaVel -> (Body, Vel)+update body v dv = (body', v')+                     where+                       body' = body { pos = pos body .+. (dt *. v') }+                       v' = v .+. dv+++-- differential velocities of bodies due to gravitational influence by sources;+-- the length of the output vector matches the length of the first argument;+-- innermost 'map' may be parallelised+dvs :: NBody -> [[Body]] -> NDeltaVel+dvs bodies source_chunks =+  map vsum $ transpose $ map (part_dvs bodies) source_chunks++-- differential velocities of bodies due to gravitational influence by a+-- chunk of sources; length of output vector matches length of first argument;+-- outermost 'map' may be parallelised+part_dvs :: NBody -> [Body] -> NDeltaVel+part_dvs bodies source_chunk =+  map (\ body -> vsum (map (dv body) source_chunk)) bodies++-- differential velocity of body due to gravitational influence by source+dv :: Body -> Body -> DeltaVel+dv body source+  | body == source = vzero+  | otherwise      = (dt * g * mass source / (distSquared * dist)) *. d+                       where+                         d = pos body .-. pos source+                         distSquared = d .*. d + eps+                         dist = sqrt distSquared+++-----------------------------------------------------------------------------+-- parallel N-body computation, shared memory++-- advance all bodies by 1 time step; same API as 'advance' (except for+-- the 'Par') but uses coarse-grain data parallelism+par_advance :: [NBodyConf] -> Par [NBodyConf]+par_advance confs =+  forkMapNF compute_chunk confs+    where+      bss = map fst confs+      compute_chunk (bodies, vs) =+        return $ unzip $ zipWith3 update bodies vs $ dvs bodies bss+++-----------------------------------------------------------------------------+-- parallel N-body computation, distributed memory (sparking)++-- advance all bodies by 1 time step; same API as 'advance' (except for+-- the 'Par') but uses dist-mem data parallelism (parMap)+dist_advance :: [NBodyConf] -> Par [NBodyConf]+dist_advance confs =+  forkMapNF compute_chunk confs+    where+      bss = map fst confs+      compute_chunk (bodies, vs) =+        unzip <$> zipWith3 update bodies vs <$> dist_dvs bodies bss++-- differential velocities of bodies due to gravitational influence by sources;+-- same API as 'dvs' (except for the 'Par') but uses dist-mem parMap+dist_dvs :: NBody -> [[Body]] -> Par NDeltaVel+dist_dvs bodies source_chunks =+  map vsum <$> transpose <$>+    parMapNF $(mkClosure [| part_dvs bodies |]) source_chunks+++-----------------------------------------------------------------------------+-- parallel N-body computation, distributed memory (pushing)++-- advance all bodies by 1 time step; same API as 'advance' (except for+-- the 'Par') but uses dist-mem data parallelism (pushMap)+dist_advance_push :: [NBodyConf] -> Par [NBodyConf]+dist_advance_push confs = do+  nodes <- allNodes+  me <- myNode+  let other_nodes = case delete me nodes of { [] -> nodes; others -> others }+  forkMapNF (compute_chunk $ other_nodes) confs+    where+      bss = map fst confs+      compute_chunk nodes (bodies, vs) =+        unzip <$> zipWith3 update bodies vs <$> dist_dvs_push nodes bodies bss++-- differential velocities of bodies due to gravitational influence by sources;+-- same API as 'dvs' (except for the 'Par') but uses dist-mem pushMap+dist_dvs_push :: [NodeId] -> NBody -> [[Body]] -> Par NDeltaVel+dist_dvs_push nodes bodies source_chunks =+  map vsum <$> transpose <$>+    pushMapNF nodes $(mkClosure [| part_dvs bodies |]) source_chunks+++-----------------------------------------------------------------------------+-- shared memory task farm++forkMapNF :: (NFData b) => (a -> Par b) -> [a] -> Par [b]+forkMapNF f = mapM spawn >=> mapM get+                where+                  spawn x = do v <- new+                               fork (f x >>= force >>= put v)+                               return v+++-----------------------------------------------------------------------------+-- chunking up lists; inverse of 'chunk n' is 'concat'++chunk :: Int -> [a] -> [[a]]+chunk n [] = []+chunk n xs = ys : chunk n zs where (ys,zs) = splitAt n xs+++-----------------------------------------------------------------------------+-- initialisation, argument processing and 'main'++-- initialize random number generator+initrand :: Int -> IO ()+initrand seed = do+  when (seed /= 0) $ do+    setStdGen (mkStdGen seed)+++-- parse runtime system config options (+ seed for random number generator)+parseOpts :: [String] -> (RTSConf, Int, [String])+parseOpts args = go (defaultRTSConf, 0, args) where+  go :: (RTSConf, Int, [String]) -> (RTSConf, Int, [String])+  go (conf, seed, [])   = (conf, seed, [])+  go (conf, seed, s:ss) =+   case stripPrefix "-rand=" s of+   Just s  -> go (conf, read s, ss)+   Nothing ->+    case stripPrefix "-d" s of+    Just s  -> go (conf { debugLvl = read s }, seed, ss)+    Nothing ->+     case stripPrefix "-scheds=" s of+     Just s  -> go (conf { scheds = read s }, seed, ss)+     Nothing ->+      case stripPrefix "-wakeup=" s of+      Just s  -> go (conf { wakeupDly = read s }, seed, ss)+      Nothing ->+       case stripPrefix "-hops=" s of+       Just s  -> go (conf { maxHops = read s }, seed, ss)+       Nothing ->+        case stripPrefix "-maxFish=" s of+        Just s  -> go (conf { maxFish = read s }, seed, ss)+        Nothing ->+         case stripPrefix "-minSched=" s of+         Just s  -> go (conf { minSched = read s }, seed, ss)+         Nothing ->+          case stripPrefix "-minNoWork=" s of+          Just s  -> go (conf { minFishDly = read s }, seed, ss)+          Nothing ->+           case stripPrefix "-numProcs=" s of+           Just s  -> go (conf { numProcs = read s }, seed, ss)+           Nothing ->+            case stripPrefix "-maxNoWork=" s of+            Just s  -> go (conf { maxFishDly = read s }, seed, ss)+            Nothing ->+             (conf, seed, s:ss)+++-- parse (optional) arguments in this order: +-- * version to run+-- * #bodies (randomly generated)+-- * #simulation steps+-- * size of chunks (evaluated sequentially, default = #bodies)+parseArgs :: [String] -> (Int, Int, Int, Int)+parseArgs []     = (defVers, defBodies, defSteps, defBodies)+parseArgs (s:ss) =+  let go :: Int -> [String] -> (Int, Int, Int, Int)+      go v []           = (v, defBodies, defSteps, defBodies)+      go v [s1]         = (v, read s1,   defSteps, read s1)+      go v [s1,s2]      = (v, read s1,   read s2,  read s1)+      go v (s1:s2:s3:_) = (v, read s1,   read s2,  read s3)+  in case stripPrefix "v" s of+       Just s' -> go (read s') ss+       Nothing -> go defVers (s:ss)+++defVers   =    0 :: Int+defBodies = 1024 :: Int+defSteps  =   20 :: Int+++main :: IO ()+main = do+  hSetBuffering stdout LineBuffering+  hSetBuffering stderr LineBuffering+  register declareStatic+  opts_args <- getArgs+  let (conf, seed, args) = parseOpts opts_args+  let (version, n, steps, chunk_size) = parseArgs args+  initrand seed+  -- on every node: generate random N-body instance, compute initial energy+  putStrLn $ "Generating random " ++ show n ++ "-body instance"+  conf0 <- randomNBodyConf n+  putStrLn $ "Computing initial energy E0"+  energy0 <- evaluate (energy conf0)+  putStrLn $ "E0 = " ++ show energy0 ++ "J"+  -- chunking up the initial config (lazily)+  let (bs0, vs0) = conf0+  let confs0 = zip (chunk chunk_size bs0) (chunk chunk_size vs0)+  case version of+      0 -> do (confs, t) <- timeIO $ evaluate+                              (iterate' steps advance confs0)+              let (bss, vss) = unzip confs+              let conf = (concat bss, concat vss)+              putStrLn $ "Computing energy drift"+              deltaE <- evaluate (energy conf - energy0)+              putStrLn $+                "{v0} nbody " ++ show n ++ " " ++ show steps +++                " --> deltaE = " ++ show deltaE ++ "J {runtime=" +++                show t ++ "}"+      1 -> do (output, t) <- timeIO $ evaluate =<< runParIO conf+                               (iterateM' steps par_advance confs0)+              case output of+                Just confs -> do let (bss, vss) = unzip confs+                                 let conf = (concat bss, concat vss)+                                 putStrLn $ "Computing energy drift"+                                 deltaE <- evaluate (energy conf - energy0)+                                 putStrLn $+                                   "{v1 chunksize=" ++ show chunk_size +++                                   "} nbody " ++ show n ++ " " ++ show steps +++                                   " --> deltaE = " ++ show deltaE +++                                   "J {runtime=" ++ show t ++ "}"+                Nothing     -> return ()+      2 -> do (output, t) <- timeIO $ evaluate =<< runParIO conf+                               (iterateM' steps dist_advance confs0)+              case output of+                Just confs -> do let (bss, vss) = unzip confs+                                 let conf = (concat bss, concat vss)+                                 putStrLn $ "Computing energy drift"+                                 deltaE <- evaluate (energy conf - energy0)+                                 putStrLn $+                                   "{v2 chunksize=" ++ show chunk_size +++                                   "} nbody " ++ show n ++ " " ++ show steps +++                                   " --> deltaE = " ++ show deltaE +++                                   "J {runtime=" ++ show t ++ "}"+                Nothing     -> return ()+      3 -> do (output, t) <- timeIO $ evaluate =<< runParIO conf+                               (iterateM' steps dist_advance_push confs0)+              case output of+                Just confs -> do let (bss, vss) = unzip confs+                                 let conf = (concat bss, concat vss)+                                 putStrLn $ "Computing energy drift"+                                 deltaE <- evaluate (energy conf - energy0)+                                 putStrLn $+                                   "{v3 chunksize=" ++ show chunk_size +++                                   "} nbody " ++ show n ++ " " ++ show steps +++                                   " --> deltaE = " ++ show deltaE +++                                   "J {runtime=" ++ show t ++ "}"+                Nothing     -> return ()+      _ -> return ()+++-----------------------------------------------------------------------------+-- auxiliary functions++-- time an IO action+timeIO :: IO a -> IO (a, NominalDiffTime)+timeIO action = do t0 <- getCurrentTime+                   x <- action+                   t1 <- getCurrentTime+                   return (x, diffUTCTime t1 t0)+++-- strict iteration+iterate' :: Int -> (a -> a) -> a -> a+iterate' n f x | n <= 0    = x+               | otherwise = let fx = f x in iterate' (n-1) f $! fx+++-- strict iteration in a monad+iterateM' :: (Monad m) => Int -> (a -> m a) -> a -> m a+iterateM' n f x | n <= 0    = return x+                | otherwise = do { fx <- f x; iterateM' (n-1) f $! fx }+++-- "deep" evaluation+forceNF :: (NFData a) => a -> a+forceNF x = x `deepseq` x
+ src/Test/HdpH/sumeuler.hs view
@@ -0,0 +1,353 @@+-- Sum of totients in HdpH+--+-- Author: Patrick Maier+-----------------------------------------------------------------------------++{-# LANGUAGE FlexibleInstances #-}  -- req'd for some ToClosure instances+{-# LANGUAGE TemplateHaskell #-}    -- req'd for mkClosure, etc++module Main where++import Prelude+import Control.Exception (evaluate)+import Control.Monad (when, (<=<))+import Data.List (elemIndex, stripPrefix)+import Data.Functor ((<$>))+import Data.List (transpose)+import Data.Maybe (fromJust)+import Data.Monoid (mconcat)+import Data.Time.Clock (NominalDiffTime, diffUTCTime, getCurrentTime)+import System.Environment (getArgs)+import System.IO (stdout, stderr, hSetBuffering, BufferMode(..))+import System.Random (mkStdGen, setStdGen)++import Control.Parallel.HdpH +       (RTSConf(..), defaultRTSConf,+        Par, runParIO,+        force, fork, spark, new, get, put, glob, rput,+        IVar, GIVar,+        Closure, unClosure, mkClosure,+        toClosure, ToClosure(locToClosure),+        static, static_, StaticToClosure, staticToClosure,+        StaticDecl, declare, register, here)+import qualified Control.Parallel.HdpH as HdpH (declareStatic)+import Control.Parallel.HdpH.Strategies +       (Strategy, ForceCC(locForceCC),+        parMapNF, parMapChunkedNF, parMapSlicedNF,+        StaticForceCC, staticForceCC)+import qualified Control.Parallel.HdpH.Strategies as Strategies (declareStatic)+++-----------------------------------------------------------------------------+-- Static declaration++instance ToClosure Int where locToClosure = $(here)+instance ToClosure [Int] where locToClosure = $(here)+instance ToClosure Integer where locToClosure = $(here)+instance ForceCC Integer where locForceCC = $(here)++declareStatic :: StaticDecl+declareStatic =+  mconcat+    [HdpH.declareStatic,         -- declare Static deserialisers+     Strategies.declareStatic,   -- from imported modules+     declare (staticToClosure :: StaticToClosure Int),+     declare (staticToClosure :: StaticToClosure [Int]),+     declare (staticToClosure :: StaticToClosure Integer),+     declare (staticForceCC :: StaticForceCC Integer),+     declare $(static 'spark_sum_euler_abs),+     declare $(static_ 'sum_totient),+     declare $(static_ 'totient)]+++-----------------------------------------------------------------------------+-- Euler's totient function (for positive integers)++totient :: Int -> Integer+totient n = toInteger $ length $ filter (\ k -> gcd n k == 1) [1 .. n]+++-----------------------------------------------------------------------------+-- sequential sum of totients++sum_totient :: [Int] -> Integer+sum_totient = sum . map totient+++-----------------------------------------------------------------------------+-- parallel sum of totients; shared memory++par_sum_totient_chunked :: Int -> Int -> Int -> Par Integer+par_sum_totient_chunked lower upper chunksize =+  sum <$> (mapM get =<< (mapM fork_sum_euler $ chunked_list))+    where+      chunked_list = chunk chunksize [upper, upper - 1 .. lower] :: [[Int]]+++par_sum_totient_sliced :: Int -> Int -> Int -> Par Integer+par_sum_totient_sliced lower upper slices =+  sum <$> (mapM get =<< (mapM fork_sum_euler $ sliced_list))+    where+      sliced_list = slice slices [upper, upper - 1 .. lower] :: [[Int]]+++fork_sum_euler :: [Int] -> Par (IVar Integer)+fork_sum_euler xs = do v <- new+                       fork $ force (sum_totient xs) >>= put v+                       return v+++-----------------------------------------------------------------------------+-- parallel sum of totients; distributed memory++dist_sum_totient_chunked :: Int -> Int -> Int -> Par Integer+dist_sum_totient_chunked lower upper chunksize = do+  sum <$> (mapM get_and_unClosure =<< (mapM spark_sum_euler $ chunked_list))+    where+      chunked_list = chunk chunksize [upper, upper - 1 .. lower] :: [[Int]]+++dist_sum_totient_sliced :: Int -> Int -> Int -> Par Integer+dist_sum_totient_sliced lower upper slices = do+  sum <$> (mapM get_and_unClosure =<< (mapM spark_sum_euler $ sliced_list))+    where+      sliced_list = slice slices [upper, upper - 1 .. lower] :: [[Int]]+++spark_sum_euler :: [Int] -> Par (IVar (Closure Integer))+spark_sum_euler xs = do +  v <- new+  gv <- glob v+  spark $(mkClosure [| spark_sum_euler_abs (xs, gv) |])+  return v++spark_sum_euler_abs :: ([Int], GIVar (Closure Integer)) -> Par ()+spark_sum_euler_abs (xs, gv) =+  force (sum_totient xs) >>= rput gv . toClosure++get_and_unClosure :: IVar (Closure a) -> Par a+get_and_unClosure = return . unClosure <=< get++-----------------------------------------------------------------------------+-- parallel sum of totients; distributed memory (using plain task farm)++farm_sum_totient_chunked :: Int -> Int -> Int -> Par Integer+farm_sum_totient_chunked lower upper chunksize =+  sum <$> parMapNF $(mkClosure [| sum_totient |]) chunked_list+    where+      chunked_list = chunk chunksize [upper, upper - 1 .. lower] :: [[Int]]+++farm_sum_totient_sliced :: Int -> Int -> Int -> Par Integer+farm_sum_totient_sliced lower upper slices =+  sum <$> parMapNF $(mkClosure [| sum_totient |]) sliced_list+    where+      sliced_list = slice slices [upper, upper - 1 .. lower] :: [[Int]]+++-----------------------------------------------------------------------------+-- parallel sum of totients; distributed memory (chunking/slicing task farms)++chunkfarm_sum_totient :: Int -> Int -> Int -> Par Integer+chunkfarm_sum_totient lower upper chunksize =+  sum <$> parMapChunkedNF chunksize $(mkClosure [| totient |]) list+    where+      list = [upper, upper - 1 .. lower] :: [Int]+++slicefarm_sum_totient :: Int -> Int -> Int -> Par Integer+slicefarm_sum_totient lower upper slices =+  sum <$> parMapSlicedNF slices $(mkClosure [| totient |]) list+    where+      list = [upper, upper - 1 .. lower] :: [Int]+++-----------------------------------------------------------------------------+-- chunking up lists; inverse of 'chunk n' is 'concat'++chunk :: Int -> [a] -> [[a]]+chunk n [] = []+chunk n xs = ys : chunk n zs where (ys,zs) = splitAt n xs+++-----------------------------------------------------------------------------+-- slicing lists; inverse of 'slice n' is 'unslice'++slice :: Int -> [a] -> [[a]]+slice n = transpose . chunk n++unslice :: [[a]] -> [a]+unslice = concat . transpose+++-----------------------------------------------------------------------------+-- initialisation, argument processing and 'main'++-- time an IO action+timeIO :: IO a -> IO (a, NominalDiffTime)+timeIO action = do t0 <- getCurrentTime+                   x <- action+                   t1 <- getCurrentTime+                   return (x, diffUTCTime t1 t0)+++-- initialize random number generator+initrand :: Int -> IO ()+initrand seed = do+  when (seed /= 0) $ do+    setStdGen (mkStdGen seed)+++-- parse runtime system config options (+ seed for random number generator)+parseOpts :: [String] -> (RTSConf, Int, [String])+parseOpts args = go (defaultRTSConf, 0, args) where+  go :: (RTSConf, Int, [String]) -> (RTSConf, Int, [String])+  go (conf, seed, [])   = (conf, seed, [])+  go (conf, seed, s:ss) =+   case stripPrefix "-rand=" s of+   Just s  -> go (conf, read s, ss)+   Nothing ->+    case stripPrefix "-d" s of+    Just s  -> go (conf { debugLvl = read s }, seed, ss)+    Nothing ->+     case stripPrefix "-scheds=" s of+     Just s  -> go (conf { scheds = read s }, seed, ss)+     Nothing ->+      case stripPrefix "-wakeup=" s of+      Just s  -> go (conf { wakeupDly = read s }, seed, ss)+      Nothing ->+       case stripPrefix "-hops=" s of+       Just s  -> go (conf { maxHops = read s }, seed, ss)+       Nothing ->+        case stripPrefix "-maxFish=" s of+        Just s  -> go (conf { maxFish = read s }, seed, ss)+        Nothing ->+         case stripPrefix "-minSched=" s of+         Just s  -> go (conf { minSched = read s }, seed, ss)+         Nothing ->+          case stripPrefix "-minNoWork=" s of+          Just s  -> go (conf { minFishDly = read s }, seed, ss)+          Nothing ->+           case stripPrefix "-numProcs=" s of+           Just s  -> go (conf { numProcs = read s }, seed, ss)+           Nothing ->+            case stripPrefix "-maxNoWork=" s of+            Just s  -> go (conf { maxFishDly = read s }, seed, ss)+            Nothing ->+             (conf, seed, s:ss)+++-- parse (optional) arguments in this order: +-- * version to run+-- * lower bound for Euler's totient function+-- * upper bound for Euler's totient function+-- * size of chunks (evaluated sequentially)+parseArgs :: [String] -> (Int, Int, Int, Int)+parseArgs []     = (defVers, defLower, defUpper, defChunk)+parseArgs (s:ss) =+  let go :: Int -> [String] -> (Int, Int, Int, Int)+      go v []           = (v, defLower, defUpper, defChunk)+      go v [s1]         = (v, defLower, read s1,  defChunk)+      go v [s1,s2]      = (v, read s1,  read s2,  defChunk)+      go v (s1:s2:s3:_) = (v, read s1,  read s2,  read s3)+  in case stripPrefix "v" s of+       Just s' -> go (read s') ss+       Nothing -> go defVers (s:ss)+++defVers  =     7 :: Int+defLower =     1 :: Int+defUpper = 20000 :: Int+defChunk =   100 :: Int+++main :: IO ()+main = do+  hSetBuffering stdout LineBuffering+  hSetBuffering stderr LineBuffering+  register declareStatic+  opts_args <- getArgs+  let (conf, seed, args) = parseOpts opts_args+  let (version, lower, upper, gran_arg) = parseArgs args+  initrand seed+  case version of+      0 -> do (x, t) <- timeIO $ evaluate+                          (sum_totient [upper, upper - 1 .. lower])+              putStrLn $+                "{v0} sum $ map totient [" ++ show lower ++ ".." +++                show upper ++ "] = " ++ show x +++                " {runtime=" ++ show t ++ "}"+      1 -> do (output, t) <- timeIO $ evaluate =<< runParIO conf+                               (par_sum_totient_chunked lower upper gran_arg)+              case output of+                Just x  -> putStrLn $+                             "{v1, chunksize=" ++ show gran_arg ++ "} " +++                             "sum $ map totient [" ++ show lower ++ ".." +++                             show upper ++ "] = " ++ show x +++                             " {runtime=" ++ show t ++ "}"+                Nothing -> return ()+      2 -> do (output, t) <- timeIO $ evaluate =<< runParIO conf+                               (dist_sum_totient_chunked lower upper gran_arg)+              case output of+                Just x  -> putStrLn $+                             "{v2, chunksize=" ++ show gran_arg ++ "} " +++                             "sum $ map totient [" ++ show lower ++ ".." +++                             show upper ++ "] = " ++ show x +++                             " {runtime=" ++ show t ++ "}"+                Nothing -> return ()+      3 -> do (output, t) <- timeIO $ evaluate =<< runParIO conf+                               (farm_sum_totient_chunked lower upper gran_arg)+              case output of+                Just x  -> putStrLn $+                             "{v3, chunksize=" ++ show gran_arg ++ "} " +++                             "sum $ map totient [" ++ show lower ++ ".." +++                             show upper ++ "] = " ++ show x +++                             " {runtime=" ++ show t ++ "}"+                Nothing -> return ()+      4 -> do (output, t) <- timeIO $ evaluate =<< runParIO conf+                               (chunkfarm_sum_totient lower upper gran_arg)+              case output of+                Just x  -> putStrLn $+                             "{v4, chunksize=" ++ show gran_arg ++ "} " +++                             "sum $ map totient [" ++ show lower ++ ".." +++                             show upper ++ "] = " ++ show x +++                             " {runtime=" ++ show t ++ "}"+                Nothing -> return ()+      5 -> do (output, t) <- timeIO $ evaluate =<< runParIO conf+                               (par_sum_totient_sliced lower upper gran_arg)+              case output of+                Just x  -> putStrLn $+                             "{v5, slices=" ++ show gran_arg ++ "} " +++                             "sum $ map totient [" ++ show lower ++ ".." +++                             show upper ++ "] = " ++ show x +++                             " {runtime=" ++ show t ++ "}"+                Nothing -> return ()+      6 -> do (output, t) <- timeIO $ evaluate =<< runParIO conf+                               (dist_sum_totient_sliced lower upper gran_arg)+              case output of+                Just x  -> putStrLn $+                             "{v6, slices=" ++ show gran_arg ++ "} " +++                             "sum $ map totient [" ++ show lower ++ ".." +++                             show upper ++ "] = " ++ show x +++                             " {runtime=" ++ show t ++ "}"+                Nothing -> return ()+      7 -> do (output, t) <- timeIO $ evaluate =<< runParIO conf+                               (farm_sum_totient_sliced lower upper gran_arg)+              case output of+                Just x  -> putStrLn $+                             "{v7, slices=" ++ show gran_arg ++ "} " +++                             "sum $ map totient [" ++ show lower ++ ".." +++                             show upper ++ "] = " ++ show x +++                             " {runtime=" ++ show t ++ "}"+                Nothing -> return ()+      8 -> do (output, t) <- timeIO $ evaluate =<< runParIO conf+                               (slicefarm_sum_totient lower upper gran_arg)+              case output of+                Just x  -> putStrLn $+                             "{v8, slices=" ++ show gran_arg ++ "} " +++                             "sum $ map totient [" ++ show lower ++ ".." +++                             show upper ++ "] = " ++ show x +++                             " {runtime=" ++ show t ++ "}"+                Nothing -> return ()++      _ -> return ()