packages feed

remote (empty) → 0.1

raw patch · 33 files changed

+6572/−0 lines, 33 filesdep +basedep +binarydep +bytestringsetup-changed

Dependencies added: base, binary, bytestring, containers, directory, filepath, mtl, network, pureMD5, stm, syb, template-haskell, time, utf8-string

Files

+ LICENSE view
@@ -0,0 +1,31 @@+Copyright Jeff Epstein 2011++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 name of the author 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,175 @@+Cloud Haskell+=============++This is **Cloud Haskell**. It's a Haskell framework for distributed applications. Basically, it's a tool for writing applications that coordinate their work on a cluster of commodity computers or virtual machines. This is useful for providing highly reliable, redundant, long-running services, as well as for building compute-intensive applications that can benefit from lots of hardware.++Cloud Haskell has two interfaces:++* an interface based on message-passing between distributed processes. Think of it as Erlang (or MPI) in Haskell. We call this part the _process layer_.+* a fault-tolerant data-centric interface. We call this part the _task layer_. This layer makes it even easier to build distributed applications; the framework automatically takes care of moving your data around and recovering from hardware failure. This layer can be compared to Google's MapReduce but is in fact more flexible.++This file contains a (slightly out-of-date) introduction to the process layer. There are example programs included in the distribution.++We suggest perusing the up-to-date Haddock [documentation](http://www.cl.cam.ac.uk/~jee36/remote/) and the [paper](http://www.cl.cam.ac.uk/~jee36/remote.pdf). My [thesis](http://www.cl.cam.ac.uk/~jee36/thesis.pdf) additionally documents the task layer and some internals.+++Installation+------------+Assuming you have the Haskell Platform installed, just run these commands from the Cloud Haskell directory:++```+cabal configure+cabal install+```++You can then compile your own Cloud Haskell applications, or try out the programs in the examples directory.++Process layer: an introduction+------------------------------++Many programming languages expose concurrent programming as a shared memory model, wherein multiple, concurrently executing programs, or threads, can examine and manipulate variables common to them all. Coordination between threads is achieved with locks, mutexes, and other synchronization mechanisms. In Haskell, these facilities are available as `MVar`s.++In contrast, languages like Erlang eschew shared data and require that concurrent threads communicate only by message-passing. The key insight of Erlang and languages like it is that reasoning about concurrency is much easier without shared memory. Under a message-passing scheme, a thread provides a  recipient, given as a thread identifier, and a unit of data; that data will be transferred to the recipient's address space and placed in a queue, where it can be retrieved by the recipient. Because data is never shared implicitly, this is a particularly good model for distributed systems.++This framework presents a combined approach to distributed framework: while it provides an Erlang-style message-passing system, it lets the programmer use existing concurrency paradigms from Haskell.++Nodes and processes+-------------------++Location is represented by a _node_. Usually, a node corresponds to an instance of the Haskell runtime system; that is, each independently executed Haskell program exists in its own node. Multiple nodes may run concurrently on a single physical host system, but the intention is that nodes run on separate hosts, to take advantage of more hardware.++The basic unit of concurrency is the _process_ (as distinct from an OS process). A process is a concurrent calculation that can  participate in messaging. There is little overhead involved in starting and executing processes, so programmers can start as many as they need. ++Code that runs in a process is in the `ProcessM` monad.++Process management+------------------++Processes are created with the `spawn` function. Its type signatures will help explain its operation:++```haskell+spawn :: NodeId -> Closure (ProcessM ()) -> ProcessM ProcessId+```++`spawn` takes a `NodeId`, indicating where to run the process, and a `Closure`, indicating which function will start the process. This lets the programmer start arbitrary functions on other nodes, which may be running on other hosts. Actual code is not transmitted to the other node; instead, a function identifier is sent. This works on the assumption that all connected nodes are running identical copies of the compiled Haskell binary (unlike Erlang, which allows new code to be sent to remote nodes at runtime).++We encode the function identifier used to start remote processes in a `Closure`. Closures for remotely-callable functions are automatically generated, and are named after the original function with a `__closure` suffix. Therefore, if I have a function like this:++```haskell+greet :: String -> ProcessM ()+greet name = say ("Hello, " ++ name)+```++I can run it on some node (and get its PID) like this:++```haskell+pid <- spawn someNode (greet__closure "John Baptist")+```++The `greet__closure` symbol here identifies a _closure generator_ and is automatically created by the framework from user-defined functions; see the examples or documentation for more details.++You can send messages to a process given its PID. The `send` function corresponds to Erlang's ! operator.++```haskell+send :: (Serializable a) => ProcessId -> a -> ProcessM ()+```++Given a `ProcessId` and a chunk of serializable data (implementing the `Data.Binary.Binary` type class), we can send a message to the given process. The message will be transmitted across the network if necessary and placed in the process's message queue. Note that `send` will accept any type of data, as long as it implements Binary.++A process can receive messages by calling `expect`:++```haskell+expect :: (Serializable a) => ProcessM a+```++Note that `expect` is also polymorphic; the type of message to receive is usually inferred by the compiler. If a message of that type is in the queue, it will be returned. If multiple messages of that type are in the queue, they will be returned in FIFO order. If there are no messages of that type in the queue, the function will block until such a message arrives.++Channels+--------++A _channel_ provides an alternative to message transmission with `send` and `expect`. While `send` and `expect` allow transmission of messages of any type, channels require messages to be of uniform type. Channels work like a distributed equivalent of Haskell's `Control.Concurrent.Chan`. Unlike regular channels, distributed channels have distinct ends: a single receiving port and at least one sending port. Create a channel with a call to `newChannel`:++```haskell+newChannel :: (Serializable a) => ProcessM (SendPort a, ReceivePort a)+```++The resulting `SendPort` can be used with the `sendChannel` function to insert messages into the channel. The `ReceivePort` can be used to receive messages with 'receiveChannel'. The `SendChannel` can be serialized and sent as part of messages to other processes, which can then write to it; the `ReceiveChannel`, though, cannot be serialized, although it can be accessed from multiple threads on the same node.++Setup and walkthrough+---------------------++Here we'll provide a basic example of how to get started with your first project. ++We'll be running a program that will estimate pi, making use of available computing resources potentially on remote systems. There will be an arbitrary number of nodes, one of which will be designated the master, and the remaining nodes will be slaves. The slaves will estimate pi in such a way that their results can be combined by the master, and an approximation will be output. The more nodes, and the longer they run, the more precise the output.++In more detail: the master will assign each slave a region of the Halton sequence, and the slaves will use elements of the sequence to estimate the ratio of points in a unit square that fall within a unit circle, and that the master will sum these ratios. ++Here's the procedure, step by step.++1. Compile Pi6.hs. If you have the framework installed correctly, it should be sufficient to run:++        ghc --make Pi6++2. Select the machines you want to run the program on, and select one of them to be the master. All hosts must be connected on a local area network. For the purposes of this explanation, we'll assume that you will run your master node on a machine named `masterhost` and you will run two slave nodes each on machines named `slavehost1` and `slavehost2`.++3. Copy the compiled executable Pi6 to some location on each of the three hosts.++4. For each node, we need to create a configuration file. This is plain text file, usually named `config` and usually placed in the same directory with the executable. There are many possible settings that can be given in the configuration file, but only a few are necessary for this example; the rest have sensible defaults. On `masterhost`, create a file named `config` with the following content:++        cfgRole MASTER+        cfgHostName masterhost+        cfgKnownHosts masterhost slavehost1 slavehost2++    On `slavehost1`, create a file named `config` with the following content: ++        cfgRole WORKER+        cfgHostName slavehost1+        cfgKnownHosts masterhost slavehost1 slavehost2++    On `slavehost2`, create a file named `config` with the following content: ++        cfgRole WORKER+        cfgHostName slavehost2+        cfgKnownHosts masterhost slavehost1 slavehost2++A brief discussion of these settings and what they mean:++The `cfgRole` setting determines the node's initial behavior. This is a string which is used to differentiate the two kinds of nodes in this example. More complex distributed systems might have more different kinds of roles. In this case, WORKER nodes do nothing on startup, but just wait from a command from a master, whereas MASTER nodes seek out worker nodes and issue them commands.++The `cfgHostName` setting indicates to each node the name of the host it's running on.++The `cfgKnownHosts` setting provides a list of hosts that form part of this distributed execution. This is necessary so that the master node can find its subservient slave nodes.++Taken together, these three settings tell each node (a) its own name, (b) the names of other nodes and (c) their behavioral relationship.++5. Now, run the Pi6 program twice in each of the worker nodes. There should now be four worker nodes awaiting instructions.++6. To start the execution, run Pi6 on the master node. You should see output like this:++        2011-02-10 11:14:38.373856 UTC 0 pid://masterhost:48079/6/    SAY Starting...+        2011-02-10 11:14:38.374345 UTC 0 pid://masterhost:48079/6/    SAY Telling slave nid://slavehost1:33716/ to look at range 0..1000000+        2011-02-10 11:14:38.376479 UTC 0 pid://masterhost:48079/6/    SAY Telling slave nid://slavehost1:45343/ to look at range 1000000..2000000+        2011-02-10 11:14:38.382236 UTC 0 pid://masterhost:48079/6/    SAY Telling slave nid://slavehost2:51739/ to look at range 2000000..3000000+        2011-02-10 11:14:38.384613 UTC 0 pid://masterhost:48079/6/    SAY Telling slave nid://slavehost2:44756/ to look at range 3000000..4000000+        2011-02-10 11:14:56.720435 UTC 0 pid://masterhost:48079/6/    SAY Done: 31416061416061416061416061++Let's talk about what's going on here.++This output is generated by the framework's logging facility. Each line of output has the following fields, left-to-right: the date and time that the log entry was generated; the importance of the message (in this case 0); the process ID of the generating process; the subsystem or component that generated this message (in this case, SAY indicates that these messages were output by a call to the `say` function); and the body of the message. From these messages, we can see that the master node discovered four nodes running on two remote hosts; for each of them, the master emits a "Telling slave..." message. Note that although we had to specify the host names where the nodes were running in the config file, the master found all nodes running on each of those hosts. The log output also tells us which range of indices of the Halton sequence were assigned to each node. Each slave, having performed its calculation, sends its results back to the master, and when the master has received responses from all slaves, it prints out its estimate of pi and ends. The slave nodes continue running, waiting for another request. At this point, we could run the master again, or we can terminate the slaves manually with Ctrl-C or the kill command.++Contributors+------------++I am grateful for the contributions of the following people to this project:+ +* Alan Mycroft+* Andrew P. Black+* Conrad Parker+* Dylan Lukes+* John Hughes +* John Launchbury +* Koen Claessen +* Simon Peyton-Jones+* Thomas van Noort +* Warren Harris+
+ Remote.hs view
@@ -0,0 +1,78 @@+-- | Cloud Haskell (previously Remote Haskell) is a distributed computing+-- framework for Haskell. We can describe its interface+-- as roughly two levels: the /process layer/, consisting of+-- processes, messages, and fault monitoring; and the+-- /task layer/, consisting of tasks, promises, and fault recovery.+-- This summary module provides the most common interface+-- functions for both layers, although advanced users might want to import names+-- from the other constituent modules, as well.++module Remote ( -- * The process layer+                remoteInit,++                ProcessM, NodeId, ProcessId, MatchM,+                getSelfPid, getSelfNode,++                send,sendQuiet,++                spawn, spawnLocal, spawnAnd,+                spawnLink,+                callRemote, callRemotePure, callRemoteIO,+                AmSpawnOptions(..), defaultSpawnOptions,+                terminate,++                expect, receive, receiveWait, receiveTimeout,+                match, matchIf, matchUnknown, matchUnknownThrow, matchProcessDown,++                logS, say, LogSphere, LogTarget(..), LogFilter(..), LogConfig(..), LogLevel(..),+                setLogConfig, setNodeLogConfig, getLogConfig, defaultLogConfig, getCfgArgs,++                UnknownMessageException(..), ServiceException(..), +                TransmitException(..), TransmitStatus(..),++                nameSet, nameQuery, nameQueryOrStart,++                linkProcess, monitorProcess, unmonitorProcess,+                withMonitor, MonitorAction(..),+                ProcessMonitorException(..),++                getPeers, findPeerByRole, PeerInfo,++                remotable, RemoteCallMetaData, Lookup,++                Closure, makeClosure, invokeClosure,+                Payload, genericPut, genericGet, Serializable,++                -- * Channels++                SendPort, ReceivePort,+                newChannel, sendChannel, receiveChannel,+                +                CombinedChannelAction, combinedChannelAction,+                combinePortsBiased, combinePortsRR, mergePortsBiased, mergePortsRR,+                terminateChannel,++                -- * The task layer++                TaskM, runTask, Promise,+                newPromise, newPromiseHere, newPromiseAtRole, newPromiseNear,+                toPromise, toPromiseNear, toPromiseImm,+                readPromise,+                tlogS, tsay,++                TaskException(..),++                MapReduce(..), mapReduce,+                chunkify, shuffle,++              ) where++import Remote.Init+import Remote.Encoding+import Remote.Process+import Remote.Channel+import Remote.Call+import Remote.Task +import Remote.Peer+import Remote.Reg+import Remote.Closure
+ Remote/Call.hs view
@@ -0,0 +1,298 @@+{-# LANGUAGE TemplateHaskell #-}++-- | Provides Template Haskell-based tools+-- and syntactic sugar for dealing with closures+module Remote.Call (+         remotable,+         mkClosure,+         mkClosureRec,+        ) where++import Language.Haskell.TH+import Remote.Encoding (Payload,serialDecode,serialEncode,serialEncodePure)+import Control.Monad.Trans (liftIO)+import Control.Monad (liftM)+import Data.Maybe (isJust)+import Remote.Closure (Closure(..))+import Remote.Process (ProcessM)+import Remote.Reg (putReg,RemoteCallMetaData)+import Remote.Task (TaskM,serialEncodeA,serialDecodeA)++----------------------------------------------+-- * Compile-time metadata+----------------------------------------------++-- | A compile-time macro to expand a function name to its corresponding+-- closure name (if such a closure exists), suitable for use with+-- 'spawn', 'callRemote', etc+-- In general, using the syntax @$(mkClosure foo)@ is the same+-- as addressing the closure generator by name, that is,+-- @foo__closure@. In some cases you may need to use+-- 'mkClosureRec' instead.+mkClosure :: Name -> Q Exp+mkClosure n = do info <- reify n+                 case info of+                    VarI iname _ _ _ -> +                        do let newn = mkName $ show iname ++ "__closure"+                           newinfo <- reify newn+                           case newinfo of+                              VarI newiname _ _ _ -> varE newiname+                              _ -> error $ "Unexpected type of closure symbol for "++show n+                    _ -> error $ "No closure corresponding to "++show n++-- | A variant of 'mkClosure' suitable for expanding closures+-- of functions declared in the same module, including that+-- of the function it's used in. The Rec stands for recursive.+-- If you get the @Something is not in scope at a reify@ message+-- when using mkClosure, try using this function instead.+-- Using this function also turns off the static+-- checks used by mkClosure, and therefore you are responsible+-- for making sure that you use 'remotable' with each function+-- that may be an argument of mkClosureRec+mkClosureRec :: Name -> Q Exp+mkClosureRec name =+ do e <- makeEnv+    inf <- reify name+    case inf of+       VarI aname atype _ _ -> +          case nameModule aname of+             Just a -> case a == loc_module (eLoc e) of+                           False -> error "Can't use mkClosureRec across modules: use mkClosure instead"+                           True -> do (aat,aae) <- closureInfo e aname atype+                                      sigE (return aae) (return aat)+             _ -> error "mkClosureRec can't figure out module of symbol"+       _ -> error "mkClosureRec applied to something weird"+++closureInfo :: Env -> Name -> Type -> Q (Type,Exp)+closureInfo e named typed = +  do v <- theval+     return (thetype,v)+   where+     implFqn = loc_module (eLoc e) ++ "." ++ nameBase named ++ "__0__impl"+     (params, returns) = getReturns typed 0+     wrapit x = case isArrowType e x of+                    False -> AppT (eClosureT e) x+                    True -> wrapMonad e (eClosureT e) x+     thetype = putParams (params ++ [wrapit (putParams returns)])+     theval = lamE (map varP paramnames) (appE (appE [e|Closure|] (litE (stringL implFqn))) (appE [e|serialEncodePure|] (tupE (map varE paramnames))))+     paramnames = map (\x -> mkName $ 'a' : show x) [1..(length params)]++closureDecs :: Env -> Name -> Type -> Q [Dec]+closureDecs e n t =+  do (nt,ne) <- closureInfo e n t+     sequence [sigD closureName (return nt),+               funD closureName [clause [] (normalB $ return ne) []]]+  where closureName = mkName $ nameBase n ++ "__closure"++     +data Env = Env+  { eProcessM :: Type+  , eIO :: Type+  , eTaskM :: Type+  , ePayload :: Type+  , eLoc :: Loc+  , eLiftIO :: Exp+  , eReturn :: Exp+  , eClosure :: Exp+  , eClosureT :: Type+  }++makeEnv :: Q Env+makeEnv = +  do eProcessM <- [t| ProcessM |]+     eIO <- [t| IO |]+     eTaskM <- [t| TaskM |]+     eLoc <- location+     ePayload <- [t| Payload |]+     eLiftIO <- [e|liftIO|]+     eReturn <- [e|return|]+     eClosure <- [e|Closure|]+     eClosureT <- [t|Closure|]+     return Env {+                  eProcessM=eProcessM,+                  eIO = eIO,+                  eTaskM = eTaskM,+                  eLoc = eLoc,+                  ePayload=ePayload,+                  eLiftIO=eLiftIO,+                  eReturn=eReturn,+                  eClosure=eClosure,+                  eClosureT=eClosureT+                }++isMonad :: Env -> Type -> Bool+isMonad e t+  = t == eProcessM e+    || t == eIO e+    || t == eTaskM e++monadOf :: Env -> Type -> Maybe Type+monadOf e (AppT m _) |  isMonad e m = Just m+monadOf e _ = Nothing++restOf :: Env -> Type -> Type+restOf e (AppT m r ) | isMonad e m = r+restOf e r = r++wrapMonad :: Env -> Type -> Type -> Type+wrapMonad e monad val =+  case monadOf e val of+    Just t | t == monad -> val+    Just n -> AppT monad (restOf e val)+    Nothing -> AppT monad val++getReturns :: Type -> Int -> ([Type],[Type])+getReturns t shift = splitAt ((length arglist - 1) - shift) arglist+  where arglist = getParams t++countReturns :: Type -> Int+countReturns t = length $ getParams t++applyArgs :: Exp -> [Exp] -> Exp+applyArgs f [] = f+applyArgs f (l:r) = applyArgs (AppE f l) r++isArrowType :: Env -> Type -> Bool+isArrowType _ (AppT (AppT ArrowT _) _) = True +isArrowType e t | (isJust $ monadOf e t) && isArrowType e (restOf e t) = True+isArrowType _ _ = False++generateDecl :: Env -> Name -> Type -> Int -> Q [Dec]+generateDecl e name t shift =+   let+     implName = mkName (nameBase name ++ "__" ++ show shift ++ "__impl") +     implPlName = mkName (nameBase name ++ "__" ++ show shift ++ "__implPl")+     (params,returns) = getReturns t shift+     topmonad = case monadOf e $ last returns of+                 Just p | p == (eTaskM e) -> eTaskM e+                 _ -> eProcessM e+     lifter :: Exp -> ExpQ+     lifter x = case monadOf e $ putParams returns of+                 Just p | p == topmonad -> return x+                 Just p | p == eIO e -> return $ AppE (eLiftIO e) x+                 _ -> return $ AppE (eReturn e) x+     serialEncoder x = case topmonad of+                 p | p == eTaskM e -> appE [e|serialEncodeA|] x+                 _ -> appE [e|liftIO|] (appE [e|serialEncode|] x)+     serialDecoder x = case topmonad of+                 p | p == eTaskM e -> appE [e|serialDecodeA|] x+                 _ -> appE [e|liftIO|] (appE [e|serialDecode|] x)+     paramnames = map (\x -> 'a' : show x) [1..(length params)]+     paramnamesP = (map (varP . mkName) paramnames)+     paramnamesE = (map (VarE . mkName) paramnames)++     just a = conP (mkName "Prelude.Just") [a]++     impldec = sigD implName (appT (appT arrowT (return (ePayload e))) (return $ wrapMonad e topmonad $ putParams returns))+     impldef = funD implName [clause [varP (mkName "a")]+                 (normalB (doE [bindS (varP (mkName "res")) ((serialDecoder (varE (mkName "a")))),+                                noBindS (caseE (varE (mkName "res"))+                                        [match (just (tupP paramnamesP)) (normalB (lifter (applyArgs (VarE name) paramnamesE))) [],+                                         match wildP (normalB (appE [e|error|] (litE (stringL ("Bad decoding in closure splice of "++nameBase name))))) []])+                                      ]))+                      []]+     implPldec = sigD implPlName (return $ putParams $ [ePayload e,wrapMonad e topmonad (ePayload e)] )+     implPldef = funD implPlName [clause [varP (mkName "a")]+                                         (normalB (doE [bindS (varP (mkName "res")) ( (appE (varE implName) (varE (mkName "a")))),+                                                       noBindS ((serialEncoder (varE (mkName "res")))) ] )) [] ] +     base1 = [impldec,impldef]+     base2 = if isArrowType e $ putParams returns +                then []+                else [implPldec,implPldef]+  in do cld <- closureDecs e name t+        sequence $ base1++base2++(map return cld)+++generateDecls :: Env -> Name -> Q [Dec]+generateDecls e name = +   do tr <- getType name+      case tr of+        Nothing -> error "remotable applied to bad name"+        Just (fname,ftype) ->+-- Change the following line to: [0..countReturns ftype - 1]+-- to automatically enable partial closure generators+            liftM concat $ mapM (generateDecl e fname ftype) [0]++generateMetaData :: Env -> [Dec] -> Q [Dec]+generateMetaData e decls = sequence [sig,dec]+  where regDecls [] = []+        regDecls (first:rest) =+           case first of+              SigD named _ -> named : (regDecls rest)+              _ -> regDecls rest+        registryName = (mkName "__remoteCallMetaData")+        paramName = mkName "x"+        sig = sigD registryName [t| RemoteCallMetaData |]+        dec = funD registryName [clause [varP paramName] (normalB (toChain (regDecls decls))) []]+        fqn n = (maybe (loc_module (eLoc e)++".") ((flip (++))".") (nameModule n)) ++ nameBase n+        app2E op l r = appE (appE op l) r+        toChain [] = varE paramName+        toChain [h] = appE (app2E [e|putReg|] (varE h) (litE $ stringL (fqn h))) (varE paramName)+        toChain (h:t) = appE (app2E [e|putReg|] (varE h) (litE $ stringL (fqn h))) (toChain t)           ++-- | A compile-time macro to provide easy invocation of closures.+-- To use this, follow the following steps:+--+-- 1. First, enable Template Haskell in the module:+--+--   > {-# LANGUAGE TemplateHaskell #-}+--   > module Main where+--   > import Remote.Call (remotable)+--   >    ...+--+-- 2. Define your functions normally. Restrictions: function's type signature must be explicitly declared; no polymorphism; all parameters must implement Serializable; return value must be pure, or in one of the 'ProcessM', 'TaskM', or 'IO' monads; probably other restrictions as well.+--+--   > greet :: String -> ProcessM ()+--   > greet name = say ("Hello, "++name)+--   > badFib :: Integer -> Integer+--   > badFib 0 = 1+--   > badFib 1 = 1+--   > badFib n = badFib (n-1) + badFib (n-2)+--+-- 3. Use the 'remotable' function to automagically generate stubs and closure generators for your functions:+--+--   > $( remotable ['greet, 'badFib] )+--+--   'remotable' may be used only once per module.+--+-- 4. When you call 'remoteInit' (usually the first thing in your program), +-- be sure to give it the automagically generated function lookup tables+-- from all modules that use 'remotable':+--+--   > main = remoteInit (Just "config") [Main.__remoteCallMetaData, OtherModule.__remoteCallMetaData] initialProcess+--+-- 5. Now you can invoke your functions remotely. When a function expects a closure, give it the name+-- of the generated closure, rather than the name of the original function. If the function takes parameters,+-- so will the closure. To start the @greet@ function on @someNode@:+--+--   > spawn someNode (greet__closure "John Baptist")+--+-- Note that we say @greet__closure@ rather than just @greet@. If you prefer, you can use 'mkClosure' instead, i.e. @$(mkClosure 'greet)@, which will expand to @greet__closure@. To calculate a Fibonacci number remotely:+--+-- > val <- callRemotePure someNode (badFib__closure 5)+remotable :: [Name] -> Q [Dec]+remotable names =+   do env <- makeEnv+      newDecls <- liftM concat $ mapM (generateDecls env) names+      lookup <- generateMetaData env newDecls+      return $ newDecls ++ lookup++getType name = +  do info <- reify name+     case info of +       VarI iname itype _ _ -> return $ Just (iname,itype)+       _ -> return Nothing++putParams :: [Type] -> Type+putParams (afst:lst:[]) = AppT (AppT ArrowT afst) lst+putParams (afst:[]) = afst+putParams (afst:lst) = AppT (AppT ArrowT afst) (putParams lst)+putParams [] = error "Unexpected parameter type in remotable processing"++getParams :: Type -> [Type]+getParams typ = case typ of+                            AppT (AppT ArrowT b) c -> b : getParams c+                            b -> [b]+                                +
+ Remote/Channel.hs view
@@ -0,0 +1,161 @@+{-# LANGUAGE ExistentialQuantification,DeriveDataTypeable #-}++-- | This module provides typed channels, an alternative+-- approach to interprocess messaging. Typed channels+-- can be used in combination with or instead of the+-- the untyped channels available in the "Remote.Process"+-- module via 'send'.+module Remote.Channel (+                       -- * Basic typed channels+                       SendPort,ReceivePort,newChannel,sendChannel,receiveChannel,++                       -- * Combined typed channels+                       CombinedChannelAction,combinedChannelAction,+                       combinePortsBiased,combinePortsRR,mergePortsBiased,mergePortsRR,++                       -- * Terminate a channel+                       terminateChannel) where++import Remote.Process (ProcessM,send,getMessageType,getMessagePayload,setDaemonic,getProcess,prNodeRef,getNewMessageLocal,localFromPid,isPidLocal,TransmitException(..),TransmitStatus(..),spawnLocalAnd,ProcessId,Node,UnknownMessageException(..))+import Remote.Encoding (Serializable)++import Data.List (foldl')+import Data.Binary (Binary,get,put)+import Data.Typeable (Typeable)+import Control.Exception (throw)+import Control.Monad (when)+import Control.Monad.Trans (liftIO)+import Control.Concurrent.MVar (MVar,newEmptyMVar,takeMVar,readMVar,putMVar)+import Control.Concurrent.STM (STM,atomically,retry,orElse)+import Control.Concurrent.STM.TVar (TVar,newTVarIO,readTVar,writeTVar)++----------------------------------------------+-- * Channels+----------------------------------------------++-- | A channel is a unidirectional communication pipeline+-- with two ends: a sending port, and a receiving port.+-- This is the sending port. A process holding this+-- value can insert messages into the channel. SendPorts+-- themselves can also be sent to other processes.+-- The other side of the channel is the 'ReceivePort'.+newtype SendPort a = SendPort ProcessId deriving (Typeable)++-- | A process holding a ReceivePort can extract messages+-- from the channel, which we inserted by+-- the holder(s) of the corresponding 'SendPort'.+-- Critically, ReceivePorts, unlike SendPorts, are not serializable.+-- This means that you can only receive messages through a channel+-- on the node on which the channel was created.+data ReceivePort a = ReceivePortSimple ProcessId (MVar ())+                      | ReceivePortBiased [Node -> STM a]+                      | ReceivePortRR (TVar [Node -> STM a])++instance Binary (SendPort a) where+   put (SendPort pid) = put pid+   get = get >>= return . SendPort++-- | Create a new channel, and returns both the 'SendPort'+-- and 'ReceivePort' thereof.+newChannel :: (Serializable a) => ProcessM (SendPort a, ReceivePort a)+newChannel = do mv <- liftIO $ newEmptyMVar+                pid <- spawnLocalAnd (body mv) setDaemonic+                return (SendPort pid,+                       ReceivePortSimple pid mv)+     where body mv = liftIO (takeMVar mv)++-- | Inserts a new value into the channel.+sendChannel :: (Serializable a) => SendPort a -> a -> ProcessM ()+sendChannel (SendPort pid) a = send pid a++-- | Extract a value from the channel, in FIFO order.+receiveChannel :: (Serializable a) => ReceivePort a -> ProcessM a+receiveChannel rc = do p <- getProcess+                       channelCheckPids [rc]+                       node <- liftIO $ readMVar (prNodeRef p)+                       liftIO $ atomically $ receiveChannelImpl node rc++receiveChannelImpl :: (Serializable a) => Node -> ReceivePort a -> STM a+receiveChannelImpl node rc =+                       case rc of+                         ReceivePortBiased l -> foldl' orElse retry (map (\x -> x node) l)+                         ReceivePortRR mv -> do tv <- readTVar mv+                                                writeTVar mv (rotate tv)+                                                foldl' orElse retry (map (\x -> x node) tv)+                         ReceivePortSimple _ _ -> receiveChannelSimple node rc+      where rotate [] = []+            rotate (h:t) = t ++ [h]++data CombinedChannelAction b = forall a. (Serializable a) => CombinedChannelAction (ReceivePort a) (a -> b)++-- | Specifies a port and an adapter for combining ports via 'combinePortsBiased' and+-- 'combinePortsRR'.+combinedChannelAction :: (Serializable a) => ReceivePort a -> (a -> b) -> CombinedChannelAction b+combinedChannelAction = CombinedChannelAction++-- | This function lets us respond to messages on multiple channels+-- by combining several 'ReceivePort's into one. The resulting port+-- is the sum of the input ports, and will extract messages from all+-- of them in FIFO order. The input ports are specified by +-- 'combinedChannelAction', which also gives a converter function.+-- After combining the underlying receive ports can still+-- be used independently, as well.+-- We provide two ways to combine ports, which differ bias+-- they demonstrate in returning messages when more than one+-- underlying channel is nonempty. combinePortsBiased will+-- check ports in the order given by its argument, and so+-- if the first channel always was a message waiting, it will.+-- starve the other channels. The alternative is 'combinePortsRR'.+combinePortsBiased :: Serializable b => [CombinedChannelAction b] -> ProcessM (ReceivePort b)+combinePortsBiased chns = do mapM_ (\(CombinedChannelAction chn _ ) -> channelCheckPids [chn]) chns+                             return $ ReceivePortBiased [(\node -> receiveChannelImpl node chn >>= return . fun) | (CombinedChannelAction chn fun) <- chns]++-- | See 'combinePortsBiased'. This function differs from that one+-- in that the order that the underlying ports are checked is rotated+-- with each invocation, guaranteeing that, given enough invocations,+-- every channel will have a chance to contribute a message.+combinePortsRR :: Serializable b => [CombinedChannelAction b] -> ProcessM (ReceivePort b)+combinePortsRR chns = do mapM_ (\(CombinedChannelAction chn _ ) -> channelCheckPids [chn]) chns+                         tv <- liftIO $ newTVarIO [(\node -> receiveChannelImpl node chn >>= return . fun) | (CombinedChannelAction chn fun) <- chns]+                         return $ ReceivePortRR tv++-- | Similar to 'combinePortsBiased', with the difference that the+-- the underlying ports must be of the same type, and you don't+-- have the opportunity to provide an adapter function.+mergePortsBiased :: (Serializable a) => [ReceivePort a] -> ProcessM (ReceivePort a)+mergePortsBiased chns = do channelCheckPids chns+                           return $ ReceivePortBiased [(\node -> receiveChannelImpl node chn) | chn <- chns]++-- | Similar to 'combinePortsRR', with the difference that the+-- the underlying ports must be of the same type, and you don't+-- have the opportunity to provide an adapter function.+mergePortsRR :: (Serializable a) => [ReceivePort a] -> ProcessM (ReceivePort a)+mergePortsRR chns = do channelCheckPids chns+                       tv <- liftIO $ newTVarIO [(\node -> receiveChannelImpl node chn) | chn <- chns]+                       return $ ReceivePortRR tv++channelCheckPids :: (Serializable a) => [ReceivePort a] -> ProcessM ()+channelCheckPids chns = mapM_ checkPid chns+         where checkPid (ReceivePortSimple pid _) = do islocal <- isPidLocal pid+                                                       when (not islocal)+                                                          (throw $ TransmitException QteUnknownPid)+               checkPid _ = return ()++receiveChannelSimple :: (Serializable a) => Node -> ReceivePort a -> STM a+receiveChannelSimple node (ReceivePortSimple chpid _) = +             do mmsg <- getNewMessageLocal (node) (localFromPid chpid)+                case mmsg of+                   Nothing -> badPid+                   Just msg -> case getMessagePayload msg of+                                    Nothing -> throw $ UnknownMessageException (getMessageType msg)+                                    Just q -> return q+   where badPid = throw $ TransmitException QteUnknownPid++-- | Terminate a channel. After calling this function, 'receiveChannel'+-- on that port (or on any combined port based on it) will either +-- fail or block indefinitely, and 'sendChannel' on the corresponding+-- 'SendPort' will fail. Any unread messages remaining in the channel+-- will be lost.+terminateChannel :: (Serializable a) => ReceivePort a -> ProcessM ()+terminateChannel (ReceivePortSimple _ term) = liftIO $ putMVar (term) ()+terminateChannel _ = throw $ TransmitException QteUnknownPid
+ Remote/Closure.hs view
@@ -0,0 +1,37 @@+{-# LANGUAGE DeriveDataTypeable #-}++-- | A simple type to represent a closure, that is, a function+-- and its environment. The current implementation represents+-- functions as strings, but this could be theoretically+-- swapped out for the \"static\" mechanism described in the+-- paper.+module Remote.Closure (+                       Closure(..)+                       ) where++import Data.Binary (Binary,get,put)+import Data.Typeable (Typeable)+import Remote.Encoding (Payload)++-- | A data type representing a closure, that is, a function with its environment.+--   In spirit, this is actually:+--    +-- >   data Closure a where+-- >     Closure :: Serializable v => Static (v -> a) -> v -> Closure a     +--+--   where the Static type wraps a function with no non-static free variables.+--   We simulate this behavior by identifying top-level functions as strings.+--   See the paper for clarification.+data Closure a = Closure String Payload+     deriving (Typeable)++instance Show (Closure a) where+     show a = case a of+                (Closure fn _pl) -> show fn++instance Binary (Closure a) where+     get = do s <- get+              v <- get+              return $ Closure s v +     put (Closure s v) = put s >> put v+
+ Remote/Encoding.hs view
@@ -0,0 +1,196 @@+{-# LANGUAGE DeriveDataTypeable,CPP,FlexibleInstances,UndecidableInstances #-}++-- | This module provides the 'Serializable' type class and+-- functions to convert to and from 'Payload's. It's implemented+-- in terms of Haskell's "Data.Binary". The message sending+-- and receiving functionality in "Remote.Process" depends on this.+module Remote.Encoding (+          Serializable,+          serialEncode,+          serialEncodePure,+          serialDecode,+          serialDecodePure,+          dynamicDecodePure,+          dynamicEncodePure,+          Payload,+          DynamicPayload,+          PayloadLength,+          hPutPayload,+          hGetPayload,+          payloadLength,+          getPayloadType,+          getDynamicPayloadType,+          getPayloadContent,+          genericPut,+          genericGet) where++import Prelude hiding (id)+import qualified Prelude as Prelude++import Data.Binary (Binary,encode,decode,Put,Get,put,get,putWord8,getWord8)+import Control.Monad (liftM)+import Data.ByteString.Lazy (ByteString)+import qualified Data.ByteString.Lazy as B (hPut,hGet,length)+import Control.Exception (try,evaluate,ErrorCall)+import Data.Int (Int64)+import System.IO (Handle)+import Data.Typeable (typeOf,typeOf,Typeable)+import Data.Dynamic (Dynamic,toDyn,fromDynamic,dynTypeRep)+import Data.Generics (Data,gfoldl,gunfold, toConstr,constrRep,ConstrRep(..),repConstr,extQ,extR,dataTypeOf)++-- | Data that can be sent as a message must implement+-- this class. The class has no functions of its own,+-- but instead simply requires that the type implement+-- both 'Typeable' and 'Binary'. Typeable can usually+-- be derived automatically. Binary requires the put and get+-- functions, which can be easily implemented by hand,+-- or you can use the 'genericGet' and 'genericPut' flavors,+-- which will work automatically for types implementing+-- 'Data'.+class (Binary a,Typeable a) => Serializable a+instance (Binary a,Typeable a) => Serializable a++data Payload = Payload+                { +                  payloadType :: !ByteString,+                  payloadContent :: !ByteString+                } deriving (Typeable)+data DynamicPayload = DynamicPayload+                {+                  dynamicPayloadContent :: Dynamic+                }+type PayloadLength = Int64++instance Binary Payload where+  put pl = put (payloadType pl) >> put (payloadContent pl)+  get = get >>= \a -> get >>= \b -> return $ Payload {payloadType = a,payloadContent=b}++payloadLength :: Payload -> PayloadLength+payloadLength (Payload t c) = B.length t + B.length c++getPayloadContent :: Payload -> ByteString+getPayloadContent = payloadContent++getPayloadType :: Payload -> String+getPayloadType pl = decode $ payloadType pl++hPutPayload :: Handle -> Payload -> IO ()+hPutPayload h (Payload t c) = B.hPut h (encode (B.length t :: PayloadLength)) >>  +                       B.hPut h t >>+                       B.hPut h (encode (B.length c :: PayloadLength)) >>+                       B.hPut h c++hGetPayload :: Handle -> IO Payload+hGetPayload h = do tl <- B.hGet h (fromIntegral baseLen)+                   t <- B.hGet h (fromIntegral (decode tl :: PayloadLength))+                   cl <- B.hGet h (fromIntegral baseLen)+                   c <- B.hGet h (fromIntegral (decode cl :: PayloadLength))+                   return $ Payload {payloadType = t,payloadContent = c}+    where baseLen = B.length (encode (0::PayloadLength))++serialEncodePure :: (Serializable a) => a -> Payload+serialEncodePure a = let encoding = encode a+                      in encoding `seq` Payload {payloadType = encode $ show $ typeOf a,+                                                 payloadContent = encoding}++dynamicEncodePure :: (Serializable a) => a -> DynamicPayload+dynamicEncodePure a = DynamicPayload {dynamicPayloadContent = toDyn a}++dynamicDecodePure :: (Serializable a) => DynamicPayload -> Maybe a+dynamicDecodePure a = fromDynamic (dynamicPayloadContent a)++getDynamicPayloadType :: DynamicPayload -> String+getDynamicPayloadType a = show (dynTypeRep (dynamicPayloadContent a))++-- TODO I suspect that we will get better performance for big messages if let this be lazy+-- see also serialDecode+serialEncode :: (Serializable a) => a -> IO Payload+serialEncode a = do encoded <- evaluate $ encode a -- this evaluate is actually necessary, it turns out; it might be better to just use strict ByteStrings+                    return $ Payload {payloadType = encode $ show $ typeOf a,+                                        payloadContent = encoded}+++serialDecodePure :: (Serializable a) => Payload -> Maybe a+serialDecodePure a = (\id -> +                      let pc = payloadContent a+                      in+                        pc `seq`+                        if (decode $! payloadType a) == +                              show (typeOf $ id undefined)+                          then Just (id $! decode pc)+                          else Nothing ) Prelude.id+++serialDecode :: (Serializable a) => Payload -> IO (Maybe a)+serialDecode a = (\id ->+                      if (decode $ payloadType a) == +                            show (typeOf $ id undefined)+                         then do+                                 res <- try (evaluate $ decode (payloadContent a)) +                                    :: (Serializable a) => IO (Either ErrorCall a)+                                 case res of+                                  Left _ -> return $ Nothing+                                  Right v -> return $ Just $ id v+                         else return Nothing ) Prelude.id+++-- | Data types that can be used in messaging must+-- be serializable, which means that they must implement+-- the 'get' and 'put' methods from 'Binary'. If you+-- are too lazy to write these functions yourself,+-- you can delegate responsibility to this function.+-- It's usually sufficient to do something like this:+--+-- > import Data.Data (Data)+-- > import Data.Typeable (Typeable)+-- > import Data.Binary (Binary, get, put)+-- > data MyType = MkMyType Foobar Int [(String, Waddle Baz)]+-- >             | MkSpatula+-- >                  deriving (Data, Typeable)+-- > instance Binary MyType where+-- >    put = genericPut+-- >    get = genericGet+genericPut :: (Data a) => a ->  Put+genericPut = generic `extQ` genericString+   where generic what = fst $ gfoldl +            (\(before, a_to_b) a -> (before >> genericPut a, a_to_b a))+            (\x -> (serializeConstr (constrRep (toConstr what)), x))+            what+         genericString :: String -> Put+         genericString = put.encode++-- | This is the counterpart 'genericPut'+genericGet :: Data a => Get a+genericGet = generic `extR` genericString+   where generic = (\id -> liftM id $ deserializeConstr $ \constr_rep ->+                   gunfold (\n -> do n' <- n+                                     g' <- genericGet+                                     return $ n' g')+                           (return)+                           (repConstr (dataTypeOf (id undefined)) constr_rep)) Prelude.id+         genericString :: Get String+         genericString = do q <- get+                            return $ decode q++serializeConstr :: ConstrRep -> Put+serializeConstr (AlgConstr ix)   = putWord8 1 >> put ix+serializeConstr (IntConstr i)    = putWord8 2 >> put i+serializeConstr (FloatConstr r)  = putWord8 3 >> put r+#if __GLASGOW_HASKELL__ >= 611+serializeConstr (CharConstr c)   = putWord8 4 >> put c+#else+serializeConstr (StringConstr c)   = putWord8 4 >> put (head c)+#endif++deserializeConstr :: (ConstrRep -> Get a) -> Get a+deserializeConstr k = +      do constr_ix <- getWord8+         case constr_ix of+          1 -> get >>= \ix -> k (AlgConstr ix)+          2 -> get >>= \i -> k (IntConstr i)+          3 -> get >>= \r -> k (FloatConstr r)+#if __GLASGOW_HASKELL__ >= 611+          4 -> get >>= \c -> k (CharConstr c)+#else+          4 -> get >>= \c -> k (StringConstr (c:[]))+#endif
+ Remote/Init.hs view
@@ -0,0 +1,72 @@+-- | Exposes a high-level interface for starting a node of a distributed+-- program, taking into account a local configuration file, command+-- line arguments, and commonly-used system processes.+module Remote.Init (remoteInit) where++import qualified Prelude as Prelude+import Prelude hiding (lookup)++import Remote.Peer (startDiscoveryService)+import Remote.Task (__remoteCallMetaData)+import Remote.Process (startProcessRegistryService,suppressTransmitException,localRegistryRegisterNode,localRegistryHello,localRegistryUnregisterNode,+                       startProcessMonitorService,startNodeMonitorService,startLoggingService,startSpawnerService,ProcessM,readConfig,initNode,startLocalRegistry,+                       forkAndListenAndDeliver,waitForThreads,roleDispatch,Node,runLocalProcess,performFinalization,startFinalizerService)+import Remote.Reg (registerCalls,RemoteCallMetaData)++import System.Environment (getEnvironment)+import Control.Concurrent (threadDelay)+import Control.Monad.Trans (liftIO)+import Control.Exception (finally)+import Control.Concurrent.MVar (MVar,takeMVar,putMVar,newEmptyMVar)++startServices :: ProcessM ()+startServices = +           do+              startProcessRegistryService+              startNodeMonitorService+              startProcessMonitorService+              startLoggingService+              startDiscoveryService+              startSpawnerService+              startFinalizerService (suppressTransmitException localRegistryUnregisterNode >> return ())++dispatchServices :: MVar Node -> IO ()+dispatchServices node = do mv <- newEmptyMVar+                           _ <- runLocalProcess node (startServices >> liftIO (putMVar mv ()))+                           takeMVar mv++-- | This is the usual way create a single node of distributed program.+-- The intent is that 'remoteInit' be called in your program's 'Main.main'+-- function. A typical call takes this form:+--+-- > main = remoteInit (Just "config") [Main.__remoteCallMetaData] initialProcess+--+-- This will:+--+-- 1. Read the configuration file @config@ in the current directory or, if specified, from the file whose path is given by the environment variable @RH_CONFIG@. If the given file does not exist or is invalid, an exception will be thrown.+--+-- 2. Use the configuration given in the file as well as on the command-line to create a new node. The usual system processes will be started, including logging, discovery, and spawning.+--+-- 3. Compile-time metadata, generated by 'Remote.Call.remotable', will used for invoking closures. Metadata from each module must be explicitly mentioned.+--+-- 4. The function initialProcess will be called, given as a parameter a string indicating the value of the cfgRole setting of this node. initialProcess is provided by the user and provides an entrypoint for controlling node behavior on startup.+remoteInit :: Maybe FilePath -> [RemoteCallMetaData] -> (String -> ProcessM ()) -> IO ()+remoteInit defaultConfig metadata f = +       let+          defaultMetaData = [Remote.Task.__remoteCallMetaData]+          lookup = registerCalls (defaultMetaData ++ metadata)+       in+       do+          configFileName <- getConfigFileName+          cfg <- readConfig True configFileName+              -- TODO sanity-check cfg+          node <- initNode cfg lookup+          _ <- startLocalRegistry cfg False -- potentially fails silently+          forkAndListenAndDeliver node cfg+          dispatchServices node+          (roleDispatch node userFunction >> waitForThreads node) `finally` (performFinalization node)+          threadDelay 500000 -- TODO make configurable, or something+   where  getConfigFileName = do env <- getEnvironment+                                 return $ maybe defaultConfig Just (Prelude.lookup "RH_CONFIG" env)+          userFunction s = localRegistryHello >> localRegistryRegisterNode >> f s+
+ Remote/Peer.hs view
@@ -0,0 +1,160 @@+{-# LANGUAGE  DeriveDataTypeable #-}++-- | Exposes mechanisms for a program built on the "Remote.Process"+-- framework to discover nodes on the current network. Programs+-- can perform node discovery manually, or they can use "Remote.Task",+-- which does it automatically.+module Remote.Peer (PeerInfo,startDiscoveryService,getPeers,getPeersStatic,getPeersDynamic,findPeerByRole) where++import Prelude hiding (all, pi)++import Network.Socket (defaultHints,sendTo,recv,sClose,Socket,getAddrInfo,AddrInfoFlag(..),setSocketOption,addrFlags,addrSocketType,addrFamily,SocketType(..),Family(..),addrProtocol,SocketOption(..),AddrInfo,bindSocket,addrAddress,SockAddr(..),socket)+import Network.BSD (getProtocolNumber)+import Control.Concurrent.MVar (takeMVar, newMVar, modifyMVar_)+import Remote.Process (PeerInfo,pingNode,makeNodeFromHost,spawnLocalAnd,setDaemonic,TransmitStatus(..),TransmitException(..),PayloadDisposition(..),ptimeout,getSelfNode,sendSimple,cfgRole,cfgKnownHosts,cfgPeerDiscoveryPort,match,receiveWait,getSelfPid,getConfig,NodeId,PortId,ProcessM,ptry,localRegistryQueryNodes)+import Control.Monad.Trans (liftIO)+import Data.Typeable (Typeable)+import Data.Maybe (catMaybes)+import Data.Binary (Binary,get,put)+import Control.Exception (try,bracket,ErrorCall(..),throw)+import Data.List (nub)+import Control.Monad (filterM)+import qualified Data.Traversable as Traversable (mapM) +import qualified Data.Map as Map (unionsWith,insertWith,empty,lookup)++data DiscoveryInfo = DiscoveryInfo+     {+       discNodeId :: NodeId,+       discRole :: String+     } deriving (Typeable,Eq)++instance Binary DiscoveryInfo where+   put (DiscoveryInfo nid role) = put nid >> put role+   get = get >>= \nid -> get >>= \role -> return $ DiscoveryInfo nid role++getUdpSocket :: PortId -> IO (Socket,AddrInfo) -- mostly copied from Network.Socket+getUdpSocket port = do+    proto <- getProtocolNumber "udp"+    let hints = defaultHints { addrFlags = [AI_PASSIVE,AI_ADDRCONFIG]+                             , addrSocketType = Datagram+                             , addrFamily = AF_INET -- only INET supports broadcast+                             , addrProtocol = proto }+    addrs <- getAddrInfo (Just hints) Nothing (Just (show port))+    let addr = head addrs+    s <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)+    return (s,addr)++maxPacket :: Int+maxPacket = 1024++listenUdp :: PortId -> IO String+listenUdp port = +    bracket+        (getUdpSocket port)+	(\(s,_) -> sClose s)+	(\(sock,addr) -> do+	    setSocketOption sock ReuseAddr 1+	    bindSocket sock (addrAddress addr)+	    msg <- recv sock maxPacket+            return msg+	)++sendBroadcast :: PortId -> String -> IO ()+sendBroadcast port str +    | length str > maxPacket = throw $ TransmitException $ QteOther $ "sendBroadcast: Specified packet is too big for UDP broadcast, having a length of " ++ (show $ length str)+    | otherwise = bracket+        (getUdpSocket port >>= return . fst)+	(sClose)+	(\sock -> do+            setSocketOption sock Broadcast 1+	    _res <- sendTo sock str (SockAddrInet (toEnum port) (-1))+            return ()+	)++-- | Returns information about all nodes on the current network+-- that this node knows about. This function combines dynamic+-- and static mechanisms. See documentation on 'getPeersStatic' +-- and 'getPeersDynamic' for more info. This function depends+-- on the configuration values @cfgKnownHosts@ and @cfgPeerDiscoveryPort@.+getPeers :: ProcessM PeerInfo+getPeers = do a <- getPeersStatic+              b <- getPeersDynamic 500000+              verifyPeerInfo $ Map.unionsWith (\x y -> nub $ x ++ y) [a,b]++verifyPeerInfo :: PeerInfo -> ProcessM PeerInfo+verifyPeerInfo pi = Traversable.mapM verify1 pi+        where verify1 = filterM pingNode -- TODO ping should require a response++-- | Returns a PeerInfo, containing a list of known nodes ordered by role.+-- This information is acquired by querying the local node registry on+-- each of the hosts in the cfgKnownHosts entry in this node's config.+-- Hostnames that don't respond are assumed to be down and nodes running+-- on them won't be included in the results.+getPeersStatic :: ProcessM PeerInfo+getPeersStatic = do cfg <- getConfig+                    let peers = cfgKnownHosts cfg+                    peerinfos <- mapM (localRegistryQueryNodes . hostToNodeId) peers+                    return $ Map.unionsWith (\a b -> nub $ a ++ b) (catMaybes peerinfos)+     where hostToNodeId host = makeNodeFromHost host 0++-- | Returns a PeerInfo, containing a list of known nodes ordered by role.+-- This information is acquired by sending out a UDP broadcast on the+-- local network; active nodes running the discovery service +-- should respond with their information.+-- If nodes are running outside of the local network, or if UDP broadcasts+-- are disabled by firewall configuration, this won't return useful+-- information; in that case, use getPeersStatic.+-- This function takes a parameter indicating how long in microseconds +-- to wait for hosts to respond. A number like 50000 is usually good enough,+-- unless your network is highly congested or with high latency.+getPeersDynamic :: Int -> ProcessM PeerInfo+getPeersDynamic t = +   do pid <- getSelfPid+      cfg <- getConfig+      case (cfgPeerDiscoveryPort cfg) of+          0 -> return Map.empty+          port -> do  -- TODO should send broacast multiple times in case of packet loss+                      _ <- liftIO $ try $ sendBroadcast port (show pid) :: ProcessM (Either IOError ())+                      responses <- liftIO $ newMVar []+                      _ <- ptimeout t (receiveInfo responses)+                      res <- liftIO $ takeMVar responses+                      let all = map (\di -> (discRole di,[discNodeId di])) (nub res)+                      return $ foldl (\a (k,v) -> Map.insertWith (++) k v a ) Map.empty all+     where receiveInfo responses = let matchInfo = match (\x -> liftIO $ modifyMVar_ responses (\m -> return (x:m))) in+                         receiveWait [matchInfo] >> receiveInfo responses++-- | Given a PeerInfo returned by getPeersDynamic or getPeersStatic,+-- give a list of nodes registered as a particular role. If no nodes of+-- that role are found, the empty list is returned.+findPeerByRole :: PeerInfo -> String -> [NodeId]+findPeerByRole disc role = maybe [] id (Map.lookup role disc)++{- UNUSED+findRoles :: PeerInfo -> [String]+findRoles disc = Map.keys disc+-}++waitForDiscovery :: Int -> ProcessM Bool+waitForDiscovery delay +              | delay <= 0 = doit +              | otherwise = ptimeout delay doit >>= (return . maybe False id)+           where doit =+                   do cfg <- getConfig+                      msg <- liftIO $ listenUdp (cfgPeerDiscoveryPort cfg)+                      nodeid <- getSelfNode+                      res <- ptry $ sendSimple (read msg) (DiscoveryInfo {discNodeId=nodeid,discRole=cfgRole cfg}) PldUser+                                 :: ProcessM (Either ErrorCall TransmitStatus)+                      case res of+                         Right QteOK -> return True+                         _ -> return False++-- | Starts the discovery process, allowing this node to respond to+-- queries from getPeersDynamic. You don't want to call this yourself,+-- as it's called for you in 'Remote.Init.remoteInit'+startDiscoveryService :: ProcessM ()+startDiscoveryService = +   do cfg <- getConfig+      if cfgPeerDiscoveryPort cfg /= 0+         then spawnLocalAnd service setDaemonic >> return ()+         else return ()+  where service = waitForDiscovery 0 >> service
+ Remote/Process.hs view
@@ -0,0 +1,2891 @@+{-# LANGUAGE DeriveDataTypeable #-}++-- | This module is the core of Cloud Haskell. It provides +-- processes, messages, monitoring, and configuration.+module Remote.Process  (+                       -- * The Process monad+                       ProcessM,+                       NodeId,ProcessId,+                       PeerInfo,+                       nullPid,getSelfPid,getSelfNode,isPidLocal,++                       -- * Message receiving+                       expect,+                       MatchM,receive,receiveWait,receiveTimeout,+                       match,matchIf,matchCond,matchUnknown,matchUnknownThrow,matchProcessDown,++                       -- * Message sending+                       send,sendQuiet,++                       -- * Logging functions+                       logS,say,+                       LogSphere,LogLevel(..),LogTarget(..),LogFilter(..),LogConfig(..),+                       setLogConfig,getLogConfig,setNodeLogConfig,setRemoteNodeLogConfig,defaultLogConfig,++                       -- * Exception handling+                       ptry,ptimeout,pbracket,pfinally,+                       UnknownMessageException(..),ServiceException(..),+                       TransmitException(..),TransmitStatus(..),++                       -- * Process naming+                       nameSet, nameQuery, nameQueryOrStart,++                       -- * Process spawning and monitoring+                       spawnLocal,spawnLocalAnd,forkProcess,spawn,spawnAnd,spawnLink,unpause,+                       AmSpawnOptions(..),defaultSpawnOptions,MonitorAction(..),SignalReason(..),+                       ProcessMonitorException(..),linkProcess,monitorProcess,unmonitorProcess,withMonitor,pingNode,+                       callRemote,callRemotePure,callRemoteIO,+                       terminate,++                       -- * Config file+                       readConfig,emptyConfig,Config(..),getConfig, getCfgArgs,++                       -- * Initialization+                       initNode,roleDispatch,setDaemonic,+                       waitForThreads,performFinalization,forkAndListenAndDeliver,runLocalProcess,++                       -- * Closures+                       makeClosure,invokeClosure,evaluateClosure,+ +                       -- * Debugging aids+                       getQueueLength,nodeFromPid,localFromPid,hostFromNid,++                       -- * Various internals, not for general use+                       PortId,LocalProcessId,+                       localRegistryHello,localRegistryRegisterNode, localRegistryQueryNodes,localRegistryUnregisterNode,+                       sendSimple,makeNodeFromHost,getNewMessageLocal, getProcess,Message,Process,prNodeRef,+                       roundtripResponse,roundtripResponseAsync,roundtripQuery,roundtripQueryMulti,+                       makePayloadClosure,getLookup,diffTime,roundtripQueryImpl,roundtripQueryUnsafe,+                       PayloadDisposition(..),suppressTransmitException,Node,getMessagePayload,getMessageType,++                       -- * System service processes, not for general use+                       startSpawnerService, startLoggingService, startProcessMonitorService, startLocalRegistry, startFinalizerService,+                       startNodeMonitorService, startProcessRegistryService, standaloneLocalRegistry+                       )+                       where++import qualified Prelude as Prelude+import Prelude hiding (catch, id, init, last, lookup, pi)++import Control.Concurrent (forkIO,ThreadId,threadDelay)+import Control.Concurrent.MVar (MVar,newMVar, newEmptyMVar,takeMVar,putMVar,modifyMVar,modifyMVar_,readMVar)+import Control.Exception (ErrorCall(..),throwTo,bracket,try,Exception,throw,evaluate,finally,SomeException,catch)+import Control.Monad (foldM,when,liftM,forever)+import Control.Monad.Trans (MonadIO,liftIO)+import Data.Binary (Binary,put,get,putWord8,getWord8)+import Data.Char (isSpace,isDigit)+import Data.List (isSuffixOf,foldl', isPrefixOf)+import Data.Maybe (catMaybes,isNothing)+import Data.Typeable (Typeable)+import Data.Data (Data)+import Data.Unique (newUnique,hashUnique)+import System.IO (Handle,hClose,hSetBuffering,hGetChar,hPutChar,BufferMode(..),hFlush)+import System.IO.Error (isEOFError,isDoesNotExistError,isUserError)+import Network.BSD (getHostName)+import Network (HostName,PortID(..),listenOn,accept,sClose,connectTo)+import Network.Socket (setSocketOption,SocketOption(..),socketPort,aNY_PORT )+import qualified Data.Map as Map (Map,keys,fromList,unionWith,elems,singleton,member,update,empty,adjust,alter,insert,delete,lookup,toList,size,insertWith')+import Remote.Reg (getEntryByIdent,Lookup,empty)+import Remote.Encoding (serialEncode,serialDecode,serialEncodePure,serialDecodePure,dynamicEncodePure,dynamicDecodePure,DynamicPayload,Payload,Serializable,hPutPayload,hGetPayload,getPayloadType,getDynamicPayloadType)+import System.Environment (getArgs)+import qualified System.Timeout (timeout)+import Data.Time (toModifiedJulianDay,Day(..),picosecondsToDiffTime,getCurrentTime,diffUTCTime,UTCTime(..),utcToLocalZonedTime)+import Remote.Closure (Closure (..))+import Control.Concurrent.STM (STM,atomically,retry,orElse)+import Control.Concurrent.STM.TChan (TChan,isEmptyTChan,readTChan,newTChanIO,writeTChan)+import Control.Concurrent.Chan (newChan,readChan,writeChan)+import Control.Concurrent.STM.TVar (TVar,newTVarIO,readTVar,writeTVar)+import Control.Concurrent.QSem (QSem,newQSem,waitQSem,signalQSem)+import Data.IORef (IORef,newIORef,readIORef,writeIORef)+import Data.Fixed (Pico)++----------------------------------------------+-- * Process monad+----------------------------------------------++type PortId = Int+type LocalProcessId = Int++-- | The Config structure encapsulates the user-settable configuration options for each node.+-- This settings are usually read in from a configuration file or from the executable's+-- command line; in either case, see 'Remote.Init.remoteInit' and 'readConfig'+data Config = Config +         {+             cfgRole :: !String, -- ^ The user-assigned role of this node determines what its initial behavior is and how it presents itself to its peers. Default to NODE+             cfgHostName :: !HostName, -- ^ The hostname, used as a basis for creating the name of the node. If unspecified, the OS will be queried. Since the hostname is part of the nodename, the computer must be accessible to other nodes using this name.+             cfgListenPort :: !PortId, -- ^ The TCP port on which to listen to for new connections. If unassigned or 0, the OS will assign a free port.+             cfgLocalRegistryListenPort :: !PortId, -- ^ The TCP port on which to communicate with the local node registry, or to start the local node registry if it isn't already running. This defaults to 38813 and shouldn't be changed unless you have prohibitive firewall rules+             cfgPeerDiscoveryPort :: !PortId, -- ^ The UDP port on which local peer discovery broadcasts are sent. Defaults to 38813, and only matters if you rely on dynamic peer discovery+             cfgNetworkMagic :: !String, -- ^ The unique identifying string for this network or application. Must not contain spaces. The uniqueness of this string ensures that multiple applications running on the same physical network won't accidentally communicate with each other. All nodes of your application should have the same network magic. Defaults to MAGIC+             cfgKnownHosts :: ![String], -- ^ A list of hosts where nodes may be running. When 'Remote.Peer.getPeers' or 'Remote.Peer.getPeerStatic' is called, each host on this list will be queried for its nodes. Only matters if you rely on static peer discovery.+             cfgRoundtripTimeout :: !Int, -- ^ Microseconds to wait for a response from a system service on a remote node. If your network has high latency or congestion, you may need to increase this to avoid incorrect reports of node inaccessibility. 0 to wait indefinitely (not recommended).+             cfgMaxOutgoing :: !Int, -- ^ A limit on the number of simultaneous outgoing connections per node+             cfgPromiseFlushDelay :: !Int, -- ^ Time in microseconds before an in-memory promise is flushed to disk. 0 to disable disk flush entirely.+             cfgPromisePrefix :: !String, -- ^ Prepended to the filename of flushed promises.+             cfgArgs :: [String] -- ^ Command-line arguments that are not part of the node configuration are placed here and can be examined by your application+--             logConfig :: LogConfig+         } deriving (Show)++type ProcessTable = Map.Map LocalProcessId ProcessTableEntry++type AdminProcessTable = Map.Map ServiceId LocalProcessId++data Node = Node+       {+--           ndNodeId :: NodeId+           ndProcessTable :: ProcessTable,+           ndAdminProcessTable :: AdminProcessTable,+--           connectionTable :: Map.Map HostName Handle -- outgoing connections only+--           ndHostEntries :: [HostEntry],+           ndHostName :: HostName,+           ndListenPort :: PortId,+           ndConfig :: Config,+           ndLookup :: Lookup,+           ndLogConfig :: LogConfig,+           ndNodeFinalizer :: MVar (),+           ndNodeFinalized :: MVar (),+           ndOutgoing :: QSem,+           ndNextProcessId :: LocalProcessId+           --conncetion table -- each one is MVared+           --  also contains time info about last contact+       }++-- | Identifies a node somewhere on the network. These+-- can be queried from 'getPeers'. See also 'getSelfNode'+data NodeId = NodeId !HostName !PortId deriving (Typeable,Eq,Ord,Data)++-- | Identifies a process somewhere on the network. These+-- are produced by the 'spawn' family of functions and+-- consumed by 'send'. When a process ends, its process ID+-- ceases to be valid. See also 'getSelfPid'+data ProcessId = ProcessId !NodeId !LocalProcessId deriving (Typeable,Eq,Ord,Data)++instance Binary NodeId where+   put (NodeId h p) = put h >> put p+   get = get >>= \h -> +         get >>= \p -> +         return (NodeId h p)++instance Show NodeId where+   show (NodeId hostname portid) = concat ["nid://",hostname,":",show portid,"/"]++instance Read NodeId where+   readsPrec _ s = if isPrefixOf "nid://" s+                       then let a = drop 6 s+                                hostname = takeWhile ((/=) ':') a+                                b = drop (length hostname+1) a+                                port= takeWhile ((/=) '/') b+                                c = drop (length port+1) b+                                result = NodeId hostname (read port) in+                                if (not.null) hostname && (not.null) port+                                   then [(result,c)]+                                   else error "Bad parse looking for NodeId"+                       else error "Bad parse looking for NodeId"++instance Binary ProcessId where+   put (ProcessId n p) = put n >> put p+   get = get >>= \n -> +         get >>= \p -> +         return (ProcessId n p)++instance Show ProcessId where+   show (ProcessId (NodeId hostname portid) localprocessid) = concat ["pid://",hostname,":",show portid,"/",show localprocessid,"/"]++instance Read ProcessId where+   readsPrec _ s = if isPrefixOf "pid://" s+                       then let a = drop 6 s+                                hostname = takeWhile ((/=) ':') a+                                b = drop (length hostname+1) a+                                port= takeWhile ((/=) '/') b+                                c = drop (length port+1) b+                                pid = takeWhile ((/=) '/') c+                                d = drop (length pid+1) c+                                result = ProcessId (NodeId hostname (read port)) (read pid) in+                                if (not.null) hostname && (not.null) pid && (not.null) port+                                   then [(result,d)]+                                   else error "Bad parse looking for ProcessId"+                       else error "Bad parse looking for ProcessId"++data PayloadDisposition = PldUser |+                          PldAdmin+                          deriving (Typeable,Read,Show,Eq)++data Message = +   EncodedMessage { msgEDisposition :: PayloadDisposition, msgEHeader :: Maybe Payload, msgEPayload :: !Payload }+ | DynamicMessage { msgDDisposition :: PayloadDisposition, msgDHeader :: Maybe DynamicPayload, msgDPayload :: DynamicPayload }++makeMessage :: (Serializable msg, Serializable hdr) => Bool -> PayloadDisposition -> msg -> Maybe hdr -> Message+makeMessage True pld msg hdr = +   DynamicMessage {msgDDisposition=pld,msgDHeader=maybe Nothing (Just . dynamicEncodePure) hdr, msgDPayload=dynamicEncodePure msg}+makeMessage False pld msg hdr = +   EncodedMessage {msgEDisposition=pld,msgEHeader=maybe Nothing (Just . serialEncodePure) hdr,msgEPayload=serialEncodePure msg}++getMessageType :: Message -> String+getMessageType (EncodedMessage _ _ a) = getPayloadType a+getMessageType (DynamicMessage _ _ a) = getDynamicPayloadType a++getMessagePayload :: (Serializable a) => Message -> Maybe a+getMessagePayload (EncodedMessage _ _ a) = serialDecodePure a+getMessagePayload (DynamicMessage _ _ a) = dynamicDecodePure a++{- UNUSED+messageHasHeader :: Message -> Bool+messageHasHeader (EncodedMessage _ (Just _) _) = True+messageHasHeader (DynamicMessage _ (Just _) _) = True+messageHasHeader _ = False+-}++getMessageHeader :: (Serializable a) => Message -> Maybe a+getMessageHeader (EncodedMessage _ a _) = maybe Nothing serialDecodePure a+getMessageHeader (DynamicMessage _ a _) = maybe Nothing dynamicDecodePure a++getMessageDisposition :: Message -> PayloadDisposition+getMessageDisposition (EncodedMessage a _ _) = a+getMessageDisposition (DynamicMessage a _ _) = a++data ProcessTableEntry = ProcessTableEntry+  {+      pteThread :: ThreadId,+      pteChannel :: TChan Message,+      pteDeathT :: TVar Bool,+      pteDeath :: MVar (),+      pteDaemonic :: Bool+  }++data ProcessState = ProcessState+   {+       prQueue :: Queue Message+   }++data Process = Process+   {+       prPid :: ProcessId,+       prSelf :: LocalProcessId,+       prNodeRef :: MVar Node,+       prThread :: ThreadId,+       prChannel :: TChan Message,+       prState :: TVar ProcessState,+       prPool :: IORef (Map.Map NodeId Handle),+       prLogConfig :: Maybe LogConfig+   } +                           +-- | The monad ProcessM is the core of the process layer. Functions+-- in the ProcessM monad may participate in messaging and create+-- additional concurrent processes. You can create+-- a ProcessM context from an 'IO' context with the 'remoteInit' function.+data ProcessM a = ProcessM {runProcessM :: Process -> IO (Process,a)} deriving Typeable++instance Monad ProcessM where+    m >>= k = ProcessM $ (\p -> (runProcessM m) p >>= (\(news,newa) -> runProcessM (k newa) news))+    return x = ProcessM $ \s -> return (s,x)++instance Functor ProcessM where+    fmap f v = ProcessM $ (\p -> (runProcessM v) p >>= (\x -> return $ fmap f x))++instance MonadIO ProcessM where+    liftIO arg = ProcessM $ \pr -> (arg >>= (\x -> return (pr,x)))++getProcess :: ProcessM (Process)+getProcess = ProcessM $ \x -> return (x,x)++-- | Returns command-line arguments provided to the+-- executable, excluding any command line arguments+-- that were processed by the framework.+getCfgArgs :: ProcessM [String]+getCfgArgs = do cfg <- getConfig+                return $ cfgArgs cfg++getConfig :: ProcessM (Config)+getConfig = do p <- getProcess+               node <- liftIO $ readMVar (prNodeRef p)+               return $ ndConfig node++getConfigI :: MVar Node -> IO Config+getConfigI mnode = do node <- readMVar mnode+                      return $ ndConfig node++getLookup :: ProcessM (Lookup)+getLookup = do p <- getProcess+               node <- liftIO $ readMVar (prNodeRef p)+               return $ ndLookup node++putProcess :: Process -> ProcessM ()+putProcess p = ProcessM $ \_ -> return (p,())++----------------------------------------------+-- * Message pattern matching+----------------------------------------------++-- | This monad provides the state and structure for+-- matching received messages from the incoming message queue.+-- It's the interface between the 'receive' family of functions,+-- and the 'match' family, which together can express which+-- messages can be accepted.+data MatchM q a = MatchM { runMatchM :: MatchBlock -> STM ((MatchBlock,Maybe (ProcessM q)),a) }++instance Monad (MatchM q) where+    m >>= k = MatchM $ \mbi -> do+                (mb,a) <- runMatchM m mbi+                (mb',a2) <- runMatchM (k a) (fst mb)+                return (mb',a2)+                 +    return x = MatchM $ \mb -> return $ ((mb,Nothing),x)+--    fail _ = MatchM $ \_ -> return (False,Nothing)++returnHalt :: a -> ProcessM q -> MatchM q a+returnHalt x invoker = MatchM $ \mb -> return $ ((mb,Just (invoker)),x)++liftSTM :: STM a -> MatchM q a+liftSTM arg = MatchM $ \mb -> do a <- arg +                                 return ((mb,Nothing),a)+++data MatchBlock = MatchBlock+     {+        mbMessage :: Message+     }++getMatch :: MatchM q MatchBlock+getMatch = MatchM $ \x -> return ((x,Nothing), x)+++getNewMessage :: Process -> STM Message+getNewMessage p = readTChan $ prChannel p++getNewMessageLocal :: Node -> LocalProcessId -> STM (Maybe Message)+getNewMessageLocal node lpid = do mpte <- getProcessTableEntry (node) lpid+                                  case mpte of+                                     Just pte -> do +                                                    msg <- readTChan (pteChannel pte)+                                                    return $ Just msg+                                     Nothing -> return Nothing+                      +getQueueLength :: ProcessM Int+getQueueLength = +  do p <- getProcess+     liftIO $ atomically $ +          do q <- getCurrentMessages p+             return $ length q++getCurrentMessages :: Process -> STM [Message]+getCurrentMessages p = do+                         msgs <- cleanChannel (prChannel p) []+                         ps <- readTVar (prState p)+                         let q = (prQueue ps) +                         let newq = queueInsertMulti q msgs+                         writeTVar (prState p) ps {prQueue = newq}+                         return $ queueToList newq+     where cleanChannel c m = do isEmpty <- isEmptyTChan c+                                 if isEmpty +                                    then return m+                                    else do item <- readTChan c+                                            cleanChannel c (item:m)+                        +matchMessage :: [MatchM q ()] -> Message -> STM (Maybe (ProcessM q))+matchMessage matchers msg = do (_mb,r) <- (foldl orElse (retry) (map executor matchers)) `orElse` (return (theMatchBlock,Nothing))+                               return r+   where executor x = do +                         (ok@(_mb,matchfound),_) <- runMatchM x theMatchBlock+                         case matchfound of+                            Nothing -> retry+                            _ -> return ok+         theMatchBlock = MatchBlock {mbMessage = msg}++matchMessages :: [MatchM q ()] -> [(Message,STM ())] -> STM (Maybe (ProcessM q))+matchMessages matchers msgs = (foldl orElse (retry) (map executor msgs)) `orElse` (return Nothing)+   where executor (msg,acceptor) = do+                                      res <- matchMessage matchers msg+                                      case res of+                                         Nothing -> retry+                                         Just pmq -> acceptor >> return (Just pmq)++-- | Examines the message queue of the current process, matching each message against each of the+-- provided message pattern clauses (typically provided by a function from the 'match' family). If+-- a message matches, the corresponding handler is invoked and its result is returned. If no+-- message matches, Nothing is returned.+receive :: [MatchM q ()] -> ProcessM (Maybe q)+receive m = do p <- getProcess+               res <- convertErrorCall $ liftIO $ atomically $ +                          do+                             msgs <- getCurrentMessages p+                             matchMessages m (messageHandlerGenerator (prState p) msgs)+               case res of+                  Nothing -> return Nothing+                  Just n -> do q <- n+                               return $ Just q++-- | A simple way to receive messages. +-- This will return the first message received+-- of the specified type; if no such message+-- is available, the function will block.+-- Unlike the 'receive' family of functions,+-- this function does not allow the notion+-- of choice in message extraction.+expect :: (Serializable a) => ProcessM a+expect = receiveWait [match return]++-- | Examines the message queue of the current process, matching each message against each of the+-- provided message pattern clauses (typically provided by a function from the 'match' family). If+-- a message matches, the corresponding handler is invoked and its result is returned. If no+-- message matches, the function blocks until a matching message is received.+receiveWait :: [MatchM q ()] -> ProcessM q+receiveWait m = do f <- receiveWaitImpl m+                   f+++-- | Examines the message queue of the current process, matching each message against each of the+-- provided message pattern clauses (typically provided by a function from the 'match' family). If+-- a message matches, the corresponding handler is invoked and its result is returned. If no+-- message matches, the function blocks until a matching message is received, or until the+-- specified time in microseconds has elapsed, at which point it will return Nothing.+-- If the specified time is 0, this function is equivalent to 'receive'.+receiveTimeout :: Int -> [MatchM q ()] -> ProcessM (Maybe q)+receiveTimeout 0 m = receive m+receiveTimeout to m | to > 0 = +          do res <- ptimeout to $ receiveWaitImpl m+             case res of+               Nothing -> return Nothing+               Just f -> do q <- f+                            return $ Just q++messageHandlerGenerator :: TVar ProcessState -> [Message] -> [(Message, STM ())]+messageHandlerGenerator prSt msgs = +     map (\(m,q) -> (m,do ps <- readTVar prSt+                          writeTVar prSt ps {prQueue = queueFromList q})) (exclusionList msgs)+++receiveWaitImpl :: [MatchM q ()] -> ProcessM (ProcessM q)+receiveWaitImpl m = +            do p <- getProcess+               v <- attempt1 p+               attempt2 p v+      where +            attempt2 p v = +                case v of+                  Just n -> return n+                  Nothing -> do ret <- convertErrorCall $ liftIO $ atomically $ +                                         do +                                            msg <- getNewMessage p+                                            ps <- readTVar (prState p)+                                            let oldq = prQueue ps+                                            writeTVar (prState p) ps {prQueue = queueInsert oldq msg}+                                            matchMessages m [(msg,writeTVar (prState p) ps)]+                                attempt2 p ret+            attempt1 p = convertErrorCall $ liftIO $ atomically $ +                          do+                             msgs <- getCurrentMessages p+                             matchMessages m (messageHandlerGenerator (prState p) msgs)++convertErrorCall :: ProcessM a -> ProcessM a+convertErrorCall f =+   do a <- ptry ff+      case a of+        Right c -> return c+        Left b -> throw $ TransmitException $ QteOther $ show (b::ErrorCall)+  where ff = do q <- f+                q `seq` return q++{- UNUSED+matchDebug :: (Message -> ProcessM q) -> MatchM q ()+matchDebug f = do mb <- getMatch+                  returnHalt () (f (mbMessage mb))+-}++-- | A catch-all variant of 'match' that invokes user-provided code and+-- will extact any message from the queue. This is useful for matching+-- against messages that are not recognized. Since message matching patterns+-- are evaluated in order, this function, if used, should be the last element+-- in the list of matchers given to 'receiveWait' and similar functions.+matchUnknown :: ProcessM q -> MatchM q ()+matchUnknown body = returnHalt () body++-- | A variant of 'matchUnknown' that throws a 'UnknownMessageException'+-- if the process receives a message that isn't extracted by another message matcher.+-- Equivalent to:+--+-- > matchUnknown (throw (UnknownMessageException "..."))+matchUnknownThrow :: MatchM q ()+matchUnknownThrow = do mb <- getMatch+                       returnHalt () (throw $ UnknownMessageException (getMessageType (mbMessage mb)))++-- | Used to specify a message pattern in 'receiveWait' and related functions.+-- Only messages containing data of type /a/, where /a/ is the argument to the user-provided+-- function in the first parameter of 'match', will be removed from the queue, at which point+-- the user-provided function will be invoked.+match :: (Serializable a) => (a -> ProcessM q) -> MatchM q ()+match = matchCoreHeaderless  (const True)++-- | Similar to 'match', but allows for additional criteria to be checked prior to message acceptance.+-- Here, the first user-provided function operates as a filter, and the message will be accepted+-- only if it returns True. Once it's been accepted, the second user-defined function is invoked,+-- as in 'match'+matchIf :: (Serializable a) => (a -> Bool) -> (a -> ProcessM q) -> MatchM q ()+matchIf = matchCoreHeaderless ++matchCond :: (Serializable a) => (a -> Maybe (ProcessM q)) -> MatchM q ()+matchCond f = +   matchIf (not . isNothing . f) run+  where run a = case f a of+                   Nothing -> throw $ TransmitException $ QteOther $ "Indecesive predicate in matchCond"+                   Just q -> q ++matchCoreHeaderless :: (Serializable a) => (a -> Bool) -> (a -> ProcessM q) -> MatchM q ()+matchCoreHeaderless f g = matchCore (\(a,b) -> b==(Nothing::Maybe ()) && f a)+                                            (\(a,_) -> g a)++matchCore :: (Serializable a,Serializable b) => ((a,Maybe b) -> Bool) -> ((a,Maybe b) -> ProcessM q) -> MatchM q ()+matchCore cond body = +        do mb <- getMatch+           doit mb+ where +  doit mb = +        let    +           decodified = getMessagePayload (mbMessage mb)+           decodifiedh = getMessageHeader (mbMessage mb)+        in -- decodified `seq` decodifiedh `seq`          TODO is this good or bad?+            case decodified of+              Just x -> if cond (x,decodifiedh) +                           then returnHalt () (body (x,decodifiedh))+                           else liftSTM retry+              Nothing -> liftSTM retry++++----------------------------------------------+-- * Exceptions and return values+----------------------------------------------++-- | Thrown in response to a bad configuration+-- file or command line option, most likely+-- in 'Config'+data ConfigException = ConfigException String deriving (Show,Typeable)+instance Exception ConfigException++-- | Thrown by various network-related functions when+-- communication with a host has failed+data TransmitException = TransmitException TransmitStatus deriving (Show,Typeable)+instance Exception TransmitException++-- | Thrown by 'matchUnknownThrow' in response to a message+-- of a wrong type being received by a process+data UnknownMessageException = UnknownMessageException String deriving (Show,Typeable)+instance Exception UnknownMessageException++-- | Thrown by "Remote.Process" system services in response+-- to some problem+data ServiceException = ServiceException String deriving (Show,Typeable)+instance Exception ServiceException++-- | Used internally by 'terminate' to mark the orderly termination+-- of a process+data ProcessTerminationException = ProcessTerminationException deriving (Show,Typeable)+instance Exception ProcessTerminationException++data TransmitStatus = QteOK+                    | QteUnknownPid+                    | QteBadFormat+                    | QteOther String+                    | QtePleaseSendBody+                    | QteBadNetworkMagic+                    | QteNetworkError String+                    | QteEncodingError String+                    | QteDispositionFailed+                    | QteLoggingError+                    | QteConnectionTimeout+                    | QteUnknownCommand +                    | QteThrottle Int deriving (Show,Read,Typeable)++instance Binary TransmitStatus where+   put QteOK = putWord8 0+   put QteUnknownPid = putWord8 1+   put QteBadFormat = putWord8 2+   put (QteOther s) = putWord8 3 >> put s+   put QtePleaseSendBody = putWord8 4+   put QteBadNetworkMagic = putWord8 5+   put (QteNetworkError s) = putWord8 6 >> put s+   put (QteEncodingError s) = putWord8 7 >> put s+   put QteDispositionFailed = putWord8 8+   put QteLoggingError = putWord8 9+   put QteConnectionTimeout = putWord8 10+   put QteUnknownCommand = putWord8 11+   put (QteThrottle s) = putWord8 12+   get = do ch <- getWord8+            case ch of+              0 -> return QteOK+              1 -> return QteUnknownPid+              2 -> return QteBadFormat+              3 -> get >>= return . QteOther+              4 -> return QtePleaseSendBody+              5 -> return QteBadNetworkMagic+              6 -> get >>= return . QteNetworkError+              7 -> get >>= return . QteEncodingError+              8 -> return QteDispositionFailed+              9 -> return QteLoggingError+              10 -> return QteConnectionTimeout+              11 -> return QteUnknownCommand+              12 -> get >>= return . QteThrottle+++----------------------------------------------+-- * Node and process spawning+----------------------------------------------++-- | Creates a new 'Node' object, given the specified configuration (usually created by 'readConfig') and+-- function metadata table (usually create by 'Remote.Call.registerCalls'). You probably want to use+-- 'Remote.Init.remoteInit' instead of this lower-level function.+initNode :: Config -> Lookup -> IO (MVar Node)+initNode cfg lookup = +               do defaultHostName <- getHostName+                  let newHostName =+                       if null (cfgHostName cfg)+                          then defaultHostName+                          else cfgHostName cfg -- TODO it would be nice to check that is name actually makes sense, e.g. try to contact ourselves+                  theNewHostName <- evaluate newHostName+                  finalizer <- newEmptyMVar+                  finalized <- newEmptyMVar+                  outgoing <- newQSem (cfgMaxOutgoing cfg)+                  mvar <- newMVar Node {ndHostName=theNewHostName, +                                        ndProcessTable=Map.empty,+                                        ndAdminProcessTable=Map.empty,+                                        ndConfig=cfg,+                                        ndLookup=lookup,+                                        ndListenPort=0,+                                        ndNextProcessId=0,+                                        ndNodeFinalizer=finalizer,+                                        ndNodeFinalized=finalized,+                                        ndLogConfig=defaultLogConfig,+                                        ndOutgoing=outgoing}+                  return mvar++-- | Given a Node (created by 'initNode'), start execution of user-provided code+-- by invoking the given function with the node's 'cfgRole' string.+roleDispatch :: MVar Node -> (String -> ProcessM ()) -> IO ()+roleDispatch mnode func = do cfg <- getConfigI mnode+                             runLocalProcess mnode (func (cfgRole cfg)) >> return ()++-- | Start executing a process on the current node. This is a variation of 'spawnLocal'+-- which accepts two blocks of user-defined code. The first block+-- is the main body of the code to run concurrently. The second block is a "prefix"+-- which is run in the new process, prior to the main body, but its completion+-- is guaranteed before spawnAnd returns. Thus, the prefix code is useful for+-- initializing the new process synchronously.+spawnLocalAnd :: ProcessM () -> ProcessM () -> ProcessM ProcessId+spawnLocalAnd fun prefix = +                   do p <- getProcess+                      v <- liftIO $ newEmptyMVar+                      pid <- liftIO $ runLocalProcess (prNodeRef p) (myFun v)+                      liftIO $ takeMVar v+                      return pid+   where myFun mv = (prefix `pfinally` liftIO (putMVar mv ())) >> fun++-- | A synonym for 'spawnLocal'+forkProcess :: ProcessM () -> ProcessM ProcessId+forkProcess = spawnLocal++-- | Create a parallel process sharing the same message queue and PID.+-- Not safe for export, as doing any message receive operation could+-- result in a munged message queue. +forkProcessWeak :: ProcessM () -> ProcessM ()+forkProcessWeak f = do p <- getProcess+                       _res <- liftIO $ forkIO (runProcessM f p >> return ())+                       return ()++-- | Create a new process on the current node. Returns the new process's identifier.+-- Unlike 'spawn', this function does not need a 'Closure' or a 'NodeId'. +spawnLocal :: ProcessM () -> ProcessM ProcessId+spawnLocal fun = do p <- getProcess+                    liftIO $ runLocalProcess (prNodeRef p) fun++runLocalProcess :: MVar Node -> ProcessM () -> IO ProcessId+runLocalProcess node fun = +         do +             localprocessid <- modifyMVar node (\thenode -> return $ (thenode {ndNextProcessId=1+ndNextProcessId thenode},+                                                                               ndNextProcessId thenode))+             channel <- newTChanIO+             passProcess <- newEmptyMVar+             okay <- newEmptyMVar+             thread <- forkIO (runner okay node passProcess)+             thenodeid <- getNodeId node+             let pid = thenodeid `seq` ProcessId thenodeid localprocessid+             state <- newTVarIO $ ProcessState {prQueue = queueMake}+             pool <- newIORef Map.empty+             putMVar passProcess (mkProcess channel node localprocessid thread pid state pool)+             takeMVar okay+             return pid+         where+          notifyProcessDown p r = do nid <- getNodeId node+                                     let pp = adminGetPid nid ServiceProcessMonitor+                                     let msg = GlProcessDown (prPid p) r+                                     try $ sendBasic node pp (msg) (Nothing::Maybe ()) PldAdmin Nothing :: IO (Either SomeException TransmitStatus)--ignore result ok+          notifyProcessUp _p = return ()+          closePool p = do c <- readIORef (prPool p)+                           mapM hClose (Map.elems c)+          exceptionHandler e p = let shown = show e in+                notifyProcessDown (p) (SrException shown) >>+                 (try (logI node  (prPid p) "SYS" LoCritical (concat ["Process got unhandled exception ",shown]))::IO(Either SomeException ())) >> return () --ignore error+          exceptionCatcher p fun = +                 do notifyProcessUp (p)+                    res <- try (fun `catch` (\ProcessTerminationException -> return ()))+                    case res of+                      Left e -> exceptionHandler (e::SomeException) p+                      Right a -> notifyProcessDown (p) SrNormal >> closePool p >> return a+          runner okay node passProcess = do+                  p <- takeMVar passProcess+                  let init = do death <- newEmptyMVar+                                death2 <- newTVarIO False+                                let pte = (mkProcessTableEntry (prChannel p) (prThread p) death death2 )+                                insertProcessTableEntry node (prSelf p) pte+                                return pte+                  let action = runProcessM fun p+                  bracket (init)+                          (\pte -> atomically (writeTVar (pteDeathT pte) True) >> putMVar (pteDeath pte) ())+                          (\_ -> putMVar okay () >> +                                   (exceptionCatcher p (action>>=return . snd)) >> return ())+          insertProcessTableEntry node processid entry =+               modifyMVar_ node (\n -> +                    return $ n {ndProcessTable = Map.insert processid entry (ndProcessTable n)} )+          mkProcessTableEntry channel thread death death2 = +             ProcessTableEntry+                {+                  pteChannel = channel,+                  pteThread = thread,+                  pteDeath = death,+                  pteDeathT = death2,+                  pteDaemonic = False+                }+          mkProcess channel noderef localprocessid thread pid state pool =+             Process+                {+                  prPid = pid,+                  prSelf = localprocessid,+                  prChannel = channel,+                  prNodeRef = noderef,+                  prThread = thread,+                  prState = state,+                  prPool = pool,+                  prLogConfig = Nothing+                }+++----------------------------------------------+-- * Roundtrip conversations+----------------------------------------------++-- TODO this needs withMonitor a safe variant+-- To make this work with a ptimeout but still be able to return martial results,+-- they need to be stored in an MVar. Also, like roundtipQuery, this should have+-- two variants: a flavor that establishes monitors, and a timeout-based flavor.+-- This also needs an ASYNC variant, that will send data simultaneously, from multiple subprocesses+-- and finally, it needs a POLY variant, of type Pld -> [(ProcessId,a)] -> ProcessM [Either TransmitStatus b]+roundtripQueryMulti :: (Serializable a,Serializable b) => PayloadDisposition -> [ProcessId] -> a -> ProcessM [Either TransmitStatus b]+roundtripQueryMulti pld pids dat = -- TODO timeout+                      let +                          convidsM = mapM (\_ -> liftIO $ newConversationId) pids+                       in do convids <- convidsM+                             sender <- getSelfPid+                             let+                                convopids = zip convids pids+                                sending (convid,pid) = +                                   let hdr = RoundtripHeader {msgheaderConversationId = convid,+                                                              msgheaderSender = sender,+                                                              msgheaderDestination = pid}+                                    in do res <- sendTry pid dat (Just hdr) pld+                                          case res of+                                               QteOK -> return (convid,Nothing)+                                               n -> return (convid,Just (Left n))+                             res <- mapM sending convopids+                             let receiving c = if (any isNothing (Map.elems c)) +                                                   then do newmap <- receiveWait [matcher c]+                                                           receiving newmap+                                                   else return c+                                 matcher c = matchCore (\(_,h) -> case h of+                                                                       Just a -> (msgheaderConversationId a,msgheaderSender a) `elem` convopids+                                                                       Nothing -> False)+                                                         (\(b,Just h) -> +                                                               let newmap = Map.adjust (\_ -> (Just (Right b))) (msgheaderConversationId h) c+                                                                in return (newmap) )+                             m <- receiving (Map.fromList res)+                             return $ catMaybes (Map.elems m)++generalPid :: ProcessId -> ProcessId+generalPid (ProcessId n _p) = ProcessId n (-1)++roundtripQuery :: (Serializable a, Serializable b) => PayloadDisposition -> ProcessId -> a -> ProcessM (Either TransmitStatus b)+roundtripQuery pld pid dat =+    do res <- ptry $ withMonitor apid $ roundtripQueryImpl 0 pld pid dat Prelude.id []+       case res of+         Left (ServiceException s) -> return $ Left $ QteOther s+         Right (Left a) -> return (Left a)+         Right (Right a) -> return (Right a)+  where apid = case pld of+                   PldAdmin -> generalPid pid+                   _ -> pid++roundtripQueryLocal :: (Serializable a, Serializable b) => PayloadDisposition -> ProcessId -> a -> ProcessM (Either TransmitStatus b)+roundtripQueryLocal pld pid dat = roundtripQueryImpl 0 pld pid dat Prelude.id []++roundtripQueryUnsafe :: (Serializable a, Serializable b) => PayloadDisposition -> ProcessId -> a -> ProcessM (Either TransmitStatus b)+roundtripQueryUnsafe pld pid dat = +                       do cfg <- getConfig+                          roundtripQueryImpl (cfgRoundtripTimeout cfg) pld pid dat Prelude.id []++roundtripQueryImpl :: (Serializable a, Serializable b) => Int -> PayloadDisposition -> ProcessId -> a -> (b -> c) -> [MatchM (Either TransmitStatus c) ()] -> ProcessM (Either TransmitStatus c)+roundtripQueryImpl time pld pid dat converter additional =+    do mmatcher <- roundtripQueryImplSub pld pid dat (\_ y -> (return . Right . converter) y)+       case mmatcher of+            Left err -> return $ Left err+            Right matcher ->+                     receiver $ [ matcher (),+                                  matchProcessDown pid ((return . Left . QteNetworkError) "Remote partner unavailable"),+                                  matchProcessDown (generalPid pid) ((return . Left . QteNetworkError) "Remote partner unavailable")]+                                   ++ additional+   where +         receiver matchers = +                     case time of+                        0 -> do receiveWait matchers+                        n -> do res <- receiveTimeout n matchers+                                case res of+                                   Nothing -> return (Left QteConnectionTimeout)+                                   Just a -> return a++roundtripQueryImplSub :: (Serializable a, Serializable b) => PayloadDisposition -> ProcessId -> a -> (c -> b -> ProcessM q) -> ProcessM (Either TransmitStatus (c -> MatchM q ()))+roundtripQueryImplSub pld pid dat act =+    do convId <- liftIO $ newConversationId+       sender <- getSelfPid+       res <- mysend pid dat (Just RoundtripHeader {msgheaderConversationId = convId,msgheaderSender = sender,msgheaderDestination = pid})+       case res of+            QteOK -> return $ Right $ \c -> (matchCore (\(_,h) -> +              case h of+                 Just a -> msgheaderConversationId a == convId && msgheaderSender a == pid+                 Nothing -> False) (\(x,_) -> do vv <- act c x+                                                 return $ vv))+            err -> return (Left err)+   where +         mysend p d mh = sendTry p d mh pld++roundtripResponse :: (Serializable a, Serializable b) => (a -> ProcessM (b,q)) -> MatchM q ()+roundtripResponse f = roundtripResponseAsync myf False+     where myf inp verf = do (resp,ret) <- f inp+                             _ <- verf resp+                             return ret++roundtripResponseAsync :: (Serializable a, Serializable b) => (a -> (b -> ProcessM ()) -> ProcessM q) -> Bool -> MatchM q ()+roundtripResponseAsync f throwing =+       matchCore (\(_,h)->conditional h) transformer+   where conditional :: Maybe RoundtripHeader -> Bool+         conditional a = case a of+                           Just _ -> True+                           Nothing -> False+         transformer (m,Just h) = +                            let sender b =+                                 do res <- sendTry (msgheaderSender h) b (Just RoundtripHeader {msgheaderSender=msgheaderDestination h,msgheaderDestination = msgheaderSender h,msgheaderConversationId=msgheaderConversationId h}) PldUser+                                    case res of+                                      QteOK -> return ()+                                      _ -> if throwing+                                              then throw $ TransmitException res+                                              else logS "SYS" LoImportant ("roundtripResponse couldn't send response to "++show (msgheaderSender h)++" because "++show res)++                             in f m sender++data RoundtripHeader = RoundtripHeader+    {+       msgheaderConversationId :: Int,+       msgheaderSender :: ProcessId,+       msgheaderDestination :: ProcessId+    } deriving (Typeable)+instance Binary RoundtripHeader where+   put (RoundtripHeader a b c) = put a >> put b >> put c+   get = get >>= \a -> get >>= \b -> get >>= \c -> return $ RoundtripHeader a b c+                                 +----------------------------------------------+-- * Message sending+----------------------------------------------++-- | Sends a message to the given process. If the+-- process isn't running or can't be accessed,+-- this function will throw a 'TransmitException'.+-- The message must implement the 'Serializable' interface.+send :: (Serializable a) => ProcessId -> a -> ProcessM ()+send pid msg = sendSimple pid msg PldUser >>= +                                           (\x -> case x of+                                                    QteOK -> return ()+                                                    _ -> throw $ TransmitException x+                                          )++-- | Like 'send', but in case of error returns a value rather than throw+-- an exception.+sendQuiet :: (Serializable a) => ProcessId -> a -> ProcessM TransmitStatus+sendQuiet p m = sendSimple p m PldUser++sendSimple :: (Serializable a) => ProcessId -> a -> PayloadDisposition -> ProcessM TransmitStatus+sendSimple pid dat pld = sendTry pid dat (Nothing :: Maybe ()) pld++sendTry :: (Serializable a,Serializable b) => ProcessId -> a -> Maybe b -> PayloadDisposition -> ProcessM TransmitStatus+sendTry pid msg msghdr pld = getProcess >>= (\_p -> +       let+          timeoutFilter a =+             do cfg <- getConfig+-- TODO This is problematic. We should give up on transmitting after a certain period+-- of time if the remote host is just not responsive, but delays in message delivery,+-- even to legitimate peers, are unpredictable. Timeouts are currently disabled+-- but should be enabled if I can think of a good way to set them. This would+-- also be a good place to put automatic retries or to limit the number+-- of outgoing connections per node. //!!+                q <- case 0 of -- case cfgRoundtripTimeout cfg of +                   0 -> a+                   n -> do ff <- ptimeout n $ a+                           case ff of+                              Nothing -> return QteConnectionTimeout+                              Just r -> return r+                case q of+                   QteThrottle n -> liftIO (threadDelay n) >> timeoutFilter a+                   n -> return n+          tryAction = ptry action >>=+                    (\x -> case x of+                              Left l -> return $ QteEncodingError $ show (l::ErrorCall) -- catch errors from encoding+                              Right r -> return r)+               where action = +                      do p <- getProcess+                         timeoutFilter $ liftIO $ sendBasic (prNodeRef p) pid msg msghdr pld (Just $ prPool p)++       in tryAction )      ++sendBasic :: (Serializable a,Serializable b) => MVar Node -> ProcessId -> a -> Maybe b -> PayloadDisposition -> Maybe (IORef (Map.Map NodeId Handle)) -> IO TransmitStatus+sendBasic mnode pid msg msghdr pld pool = do+              nid <- getNodeId mnode+              let islocal = nodeFromPid pid == nid++              -- TODO It's important that the semantics of messaging are preserved+              -- regardles of the location of particular processes. Messages are+              -- fully evaluated when sent to a remote node as part of the serialization+              -- but they don't need to be when sending to a local process; the message+              -- is just plopped into a TChan, basically. Unfortunately, this means+              -- that exceptions associated with the serialization of data will be+              -- handled by the receiving process, rather than the sender, which+              -- it may not be prepared for. An easy way to crash a system process,+              -- then, is to send it a message contained undefined or divide by zero.+              -- To prevent this, we now serialize all messages, even those bound for+              -- local destinations.+              -- FORMERLY: let themsg = makeMessage islocal pld msg msghdr+              let themsg = makeMessage False pld msg msghdr+              (if islocal then sendRawLocal else sendRawRemote) mnode pid nid themsg pool++sendRawLocal :: MVar Node -> ProcessId -> NodeId -> Message -> Maybe (IORef (Map.Map NodeId Handle)) -> IO TransmitStatus+sendRawLocal noderef thepid _nodeid msg _+     | thepid == nullPid = return QteUnknownPid+     | otherwise = do cfg <- getConfigI noderef+                      messageHandler cfg noderef (getMessageDisposition msg) msg (cfgNetworkMagic cfg) (localFromPid thepid)++sendRawRemote :: MVar Node -> ProcessId -> NodeId -> Message -> Maybe (IORef (Map.Map NodeId Handle)) -> IO TransmitStatus+sendRawRemote noderef thepid nodeid msg (Just pool) =+    do apool <- readIORef pool+       ppool <- if Map.size apool > 10 -- trim pool+                   then mapM hClose (Map.elems apool) >> return Map.empty+                   else return apool+       let finded = (Map.lookup thenode ppool) +       (ret,h) <- sendRawRemoteImpl noderef thepid nodeid msg finded+       case ret of+         QteOK -> cleanup h ppool+         _ -> case finded of+                Nothing -> cleanup h ppool+                _ -> do (_ret2,newh) <- sendRawRemoteImpl noderef thepid nodeid msg Nothing+                        cleanup newh ppool+       return ret+  where+      thenode = (nodeFromPid thepid)+      cleanup h ppool =        +       case h of+         Just hh -> writeIORef pool (Map.insert thenode hh ppool)+         Nothing -> writeIORef pool (Map.delete thenode ppool)+sendRawRemote noderef thepid nodeid msg Nothing = +        liftM fst $ sendRawRemoteImpl noderef thepid nodeid msg Nothing++                  +sendRawRemoteImpl :: MVar Node -> ProcessId -> NodeId -> Message -> Maybe Handle -> IO (TransmitStatus,Maybe Handle)+sendRawRemoteImpl noderef thepid@(ProcessId (NodeId hostname portid) localpid) nodeid msg bigh+     | thepid == nullPid = return (QteUnknownPid,Nothing)+     | otherwise = +         do node <- readMVar noderef+            res <- withSem (ndOutgoing node) (try setup)+            case res of+               Right n -> return n+               Left l | isEOFError l -> return (QteNetworkError (show l),Nothing)+                      | isUserError l -> return (QteBadFormat,Nothing)+                      | isDoesNotExistError l -> return (QteNetworkError (show l),Nothing)+                      | otherwise -> return (QteOther $ show l,Nothing)+    where setup = bracket +                 (acquireConnection)+                 (\_ -> return ())+                 (sender)+          acquireConnection = +               case bigh of+                   Nothing ->+                      do h <- connectTo hostname (PortNumber $ toEnum portid)+                         hSetBuffering h (BlockBuffering Nothing)+                         return h+                   Just h -> return h+          sender h = do cfg <- getConfigI noderef+                        ret <- writeMessage h (cfgNetworkMagic cfg,localpid,nodeid,msg)+                        return (ret,Just h)+    +writeMessage :: Handle -> (String,LocalProcessId,NodeId,Message)-> IO TransmitStatus+writeMessage h (magic,dest,nodeid,(EncodedMessage msgDisp msgHdr msgMsg)) = +         do hPutStrZ h $ unwords ["Rmt!!",magic,show dest,show msgDisp,show fmt,show nodeid]+            hFlush h+            response <- hGetLineZ h+            resp <- readIO response :: IO TransmitStatus+            case resp of+               QtePleaseSendBody -> do maybe (return ()) (hPutPayload h) (msgHdr)+                                       hPutPayload h msgMsg+                                       hFlush h+                                       response2 <- hGetLineZ h+                                       resp2 <- readIO response2 :: IO TransmitStatus+                                       return resp2+               QteOK -> return QteBadFormat+               n -> return n                +         where fmt = case msgHdr of+                        Nothing -> (0::Int)+                        Just _ -> 2+writeMessage _ _ = throw $ ServiceException "writeMessage went down wrong pipe"++----------------------------------------------+-- * Message delivery+----------------------------------------------++-- | Starts a message-receive loop on the given node. You probably don't want to call this function yourself.+forkAndListenAndDeliver :: MVar Node -> Config -> IO ()+forkAndListenAndDeliver node cfg = do coord <- newEmptyMVar+                                      _tid <- forkIO $ listenAndDeliver node cfg (coord)+                                      result <- takeMVar coord+                                      maybe (return ()) throw result++writeResult :: Handle -> TransmitStatus -> IO ()+writeResult h er = hPutStrZ h (show er) >> hFlush h++readMessage :: Handle -> IO (String,LocalProcessId,NodeId,Message)+readMessage h = +              do line <- hGetLineZ h+                 case words line of+                  ["Rmt!!",magic,destp,disp,format,nodeid] ->+                     do adestp <- readIO destp :: IO LocalProcessId +                        adisp <- readIO disp :: IO PayloadDisposition+                        aformat <- readIO format :: IO Int+                        anodeid <- readIO nodeid :: IO NodeId++                        writeResult h QtePleaseSendBody+                        hFlush h+                        header <- case aformat of+                                     2 -> do hdr <- hGetPayload h+                                             return $ Just hdr+                                     _ -> return Nothing+                        body <- hGetPayload h+                        return (magic,adestp,anodeid,EncodedMessage { msgEHeader = header,+                                           msgEDisposition = adisp,+                                           msgEPayload = body+                                           })+                  _ -> throw $ userError "Bad message format"++getProcessTableEntry :: Node -> LocalProcessId -> STM (Maybe ProcessTableEntry)+getProcessTableEntry tbl adestp = let res = Map.lookup adestp (ndProcessTable tbl) in+                                  case res of+                                     Just x -> do isdead <- readTVar (pteDeathT x)+                                                  if isdead+                                                     then return Nothing+                                                     else return (Just x)+                                     Nothing -> return Nothing++deliver :: LocalProcessId -> MVar Node -> Message -> IO (Maybe ProcessTableEntry)+deliver adestp tbl msg = +         do+            node <- readMVar tbl+            atomically $ do+                            pte <- getProcessTableEntry node adestp ++                            maybe (return Nothing) (\x -> writeTChan (pteChannel x) msg >> return (Just x)) pte+                               +listenAndDeliver :: MVar Node -> Config -> MVar (Maybe IOError) -> IO ()+listenAndDeliver node cfg coord = +-- this listenon should be replaced with a lower-level listen call that binds+-- to the interface corresponding to the name specified in cfgNodeName+  do res <- try $ bracket (setupSocket) (sClose) (\s -> finishStartup >> sockBody s)+     case res of+       Left e -> putMVar coord (Just e)+       Right _ -> return ()+   where  +         finishStartup = putMVar coord Nothing+         setupSocket =            +               do sock <- listenOn whichPort+                  setSocketOption sock KeepAlive 1+                  realPort <- socketPort sock+                  modifyMVar_ node (\a -> return $ a {ndListenPort=fromEnum realPort}) +                  return sock+         whichPort = if cfgListenPort cfg /= 0+                        then PortNumber $ toEnum $ cfgListenPort cfg+                        else PortNumber aNY_PORT+         handleCommSafe h = +            (try $ handleComm h :: IO (Either IOError ())) >> return ()+{- UNUSED+         logNetworkError :: IOError -> IO ()+         logNetworkError n = return ()+         writeResultTry h q =+            do res <- try (writeResult h q)+               case res of+                  Left n -> logNetworkError n+                  Right q -> return ()+-}+         handleComm h = +            do (magic,adestp,_nodeid,msg) <- readMessage h+               res <- messageHandler cfg node (getMessageDisposition msg) msg magic adestp+               writeResult h res+               case res of+                 QteOK -> handleComm h+                 _ -> return ()+         sockBody s =+              do hchan <- newChan+                 _tid <- forkIO $ forever $ do h <- readChan hchan+                                               hSetBuffering h (BlockBuffering Nothing)+                                               forkIO $ (handleCommSafe h `finally` hClose h)+                 forever $ do (newh,_,_) <- accept s+                              writeChan hchan newh+++messageHandler :: Config -> MVar Node -> PayloadDisposition -> Message -> String -> LocalProcessId -> IO TransmitStatus+messageHandler cfg node pld = case pld of +                       PldAdmin -> dispositionAdminHandler+                       PldUser -> dispositionUserHandler+   where+         validMagic magic = magic == cfgNetworkMagic cfg         -- same network+                         || magic == localRegistryMagicMagic     -- message from local magic-neutral registry+                         || cfgRole cfg == localRegistryMagicRole-- message to local magic-neutral registry+         dispositionAdminHandler msg magic adestp +            | validMagic magic = +                 do +                    realPid <- adminLookupN (toEnum adestp) node+                    case realPid of+                         Left er -> return er+                         Right p -> do res <- deliver p node msg+                                       case res of+                                         Just _ -> return QteOK+                                         Nothing -> return QteUnknownPid+            | otherwise = return QteBadNetworkMagic++         dispositionUserHandler msg magic adestp+            | validMagic magic = +                     do +                        res <- deliver adestp node msg+                        case res of+                           Nothing -> return QteUnknownPid+                           Just _ -> return QteOK+            | otherwise = return QteBadNetworkMagic++-- | Blocks until all non-daemonic processes of the given+-- node have ended. Usually called on the main thread of a program.+waitForThreads :: MVar Node -> IO ()+waitForThreads mnode = do node <- takeMVar mnode+                          waitFor (Map.toList (ndProcessTable node)) node+  where waitFor lst node= case lst of+                             [] -> putMVar mnode node+                             (pn,pte):rest -> if (pteDaemonic pte)+                                                 then waitFor rest node+                                                 else do putMVar mnode node +                                                         -- this take and modify should be atomic to ensure no one squeezes in a message to a zombie process+                                                         takeMVar  (pteDeath pte)+                                                         modifyMVar_ mnode (\x -> return $ x {ndProcessTable=Map.delete pn (ndProcessTable x)})+                                                         waitForThreads mnode++----------------------------------------------+-- * Miscellaneous utilities+----------------------------------------------++paddedString :: Int -> String -> String+paddedString i s = s ++ (replicate (i-length s) ' ')++newConversationId :: IO Int+newConversationId = do d <- newUnique+                       return $ hashUnique d++-- TODO this is a huge performance bottleneck. Why? How to fix?+exclusionList :: [a] -> [(a,[a])]+exclusionList [] = []+exclusionList (a:rest) = each [] a rest+  where+      each before it [] = [(it,before)]+      each before it following@(next:after) = (it,before++following):(each (before++[it]) next after)++{-+dumpMessageQueue :: ProcessM [String]+dumpMessageQueue = do p <- getProcess+                      msgs <- cleanChannel+                      ps <- liftIO $ modifyIORef (prState p) (\x -> x {prQueue = queueInsertMulti  (prQueue x) msgs})+                      buf <- liftIO $ readIORef (prState p)+                      return $ map msgPayload (queueToList $ prQueue buf)+                      ++printDumpMessageQueue :: ProcessM ()+printDumpMessageQueue = do liftIO $ putStrLn "----BEGINDUMP------"+                           dump <- dumpMessageQueue+                           liftIO $ mapM putStrLn dump+                           liftIO $ putStrLn "----ENDNDUMP-------"+-}++{- UNUSED+duration :: Int -> ProcessM a -> ProcessM (Int,a)+duration t a = +           do time1 <- liftIO $ getCurrentTime+              result <- a+              time2 <- liftIO $ getCurrentTime+              return (t - diffTime time2 time1,result)+-}++diffTime :: UTCTime -> UTCTime -> Int+diffTime time2 time1 =+   picosecondsToMicroseconds (fromEnum (diffUTCTime time2 time1))+   where picosecondsToMicroseconds a = a `div` 1000000+++hGetLineZ :: Handle -> IO String+hGetLineZ h = loop [] >>= return . reverse+      where loop l = do c <- hGetChar h+                        case c of+                             '\0' -> return l+                             _ -> loop (c:l)++hPutStrZ :: Handle -> String -> IO ()+hPutStrZ h [] = hPutChar h '\0'+hPutStrZ h (c:l) = hPutChar h c >> hPutStrZ h l++buildPid :: ProcessM ()+buildPid = do  p <- getProcess+               node <- liftIO $ readMVar (prNodeRef p)+               let pid=ProcessId (NodeId (ndHostName node) (ndListenPort node))+                                 (prSelf p)+               putProcess $ p {prPid = pid}++nullPid :: ProcessId+nullPid = ProcessId (NodeId "0.0.0.0" 0) 0++-- | Returns the node ID of the node that the current process is running on.+getSelfNode :: ProcessM NodeId+getSelfNode = do (ProcessId n _p) <- getSelfPid+                 return n++getNodeId :: MVar Node -> IO NodeId+getNodeId mnode = do node <- readMVar mnode+                     return (NodeId (ndHostName node) (ndListenPort node))++-- | Returns the process ID of the current process.+getSelfPid :: ProcessM ProcessId+getSelfPid = do p <- getProcess+                case prPid p of+                  (ProcessId (NodeId _ 0) _) -> (buildPid >> getProcess >>= return.prPid)+                  _ -> return $ prPid p++nodeFromPid :: ProcessId -> NodeId+nodeFromPid (ProcessId nid _) = nid++makeNodeFromHost :: String -> PortId -> NodeId+makeNodeFromHost = NodeId++localFromPid :: ProcessId -> LocalProcessId+localFromPid (ProcessId _ lid) = lid++hostFromNid :: NodeId -> HostName+hostFromNid (NodeId hn _p) = hn++buildPidFromNodeId :: NodeId -> LocalProcessId -> ProcessId+buildPidFromNodeId n lp = ProcessId n lp++{- UNUSED+localServiceToPid :: LocalProcessId -> ProcessM ProcessId+localServiceToPid sid = do (ProcessId nid _lid) <- getSelfPid+                           return $ ProcessId nid sid+-}++-- | Returns true if the given process ID is associated with the current node.+-- Does not examine if the process is currently running.+isPidLocal :: ProcessId -> ProcessM Bool+isPidLocal pid = do mine <- getSelfPid+                    return (nodeFromPid mine == nodeFromPid pid)+++----------------------------------------------+-- * Exception handling+----------------------------------------------++suppressTransmitException :: ProcessM a -> ProcessM (Maybe a)+suppressTransmitException a = +     do res <- ptry a+        case res of+          Left (TransmitException _) -> return Nothing+          Right r -> return $ Just r++-- | A 'ProcessM'-flavoured variant of 'Control.Exception.try'+ptry :: (Exception e) => ProcessM a -> ProcessM (Either e a)+ptry f = do p <- getProcess+            res <- liftIO $ try (runProcessM f p)+            case res of+              Left e -> return $ Left e+              Right (newp,newanswer) -> ProcessM (\_ -> return (newp,Right newanswer))++{- UNUSED+-- | A 'ProcessM'-flavoured variant of 'Control.Exception.catch'+pcatch :: Exception e => ProcessM a -> (e -> ProcessM a) -> ProcessM a+pcatch code handler = do p <- getProcess+                         liftIO $ catch (liftM snd $ runProcessM code p) (\e -> liftM snd $ runProcessM (handler e) p)+                         +-}++-- | A 'ProcessM'-flavoured variant of 'System.Timeout.timeout'+ptimeout :: Int -> ProcessM a -> ProcessM (Maybe a)+ptimeout t f = do p <- getProcess+                  res <- liftIO $ System.Timeout.timeout t (runProcessM f p)+                  case res of+                    Nothing -> return Nothing+                    Just (newp,newanswer) -> ProcessM (\_ -> return (newp,Just newanswer))++-- | A 'ProcessM'-flavoured variant of 'Control.Exception.bracket'+pbracket :: (ProcessM a) -> (a -> ProcessM b) -> (a -> ProcessM c) -> ProcessM c+pbracket before after fun = +       do p <- getProcess+          (newp2,newanswer2) <- liftIO $ bracket +                         (runProcessM before p) +                         (\(newp,newanswer) -> runProcessM (after newanswer) newp) +                         (\(newp,newanswer) -> runProcessM (fun newanswer) newp)+          ProcessM (\_ -> return (newp2, newanswer2))   ++-- | A 'ProcessM'-flavoured variant of 'Control.Exception.finally'+pfinally :: ProcessM a -> ProcessM b -> ProcessM a+pfinally fun after = pbracket (return ()) (\_ -> after) (const fun)+++----------------------------------------------+-- * Configuration file+----------------------------------------------+  +emptyConfig :: Config+emptyConfig = Config {+                  cfgRole = "NODE",+                  cfgHostName = "",+                  cfgListenPort = 0,+                  cfgPeerDiscoveryPort = 38813,+                  cfgLocalRegistryListenPort = 38813,+                  cfgNetworkMagic = "MAGIC",+                  cfgKnownHosts = [],+                  cfgRoundtripTimeout = 10000000,+                  cfgMaxOutgoing = 50,+                  cfgPromiseFlushDelay = 5000000,+                  cfgPromisePrefix = "rpromise-",+                  cfgArgs = []+                  }++-- | Reads in configuration data from external sources, specifically from the command line arguments+-- and a configuration file. +-- The first parameter to this function determines whether command-line arguments are consulted.+-- If the second parameter is not 'Nothing' then it should be the name of the configuration file;+-- an exception will be thrown if the specified file does not exist.+-- Usually, this function shouldn't be called directly, but rather from 'Remote.Init.remoteInit',+-- which also takes into account environment variables.+-- Options set by command-line parameters have the highest precedence,+-- followed by options read from a configuration file; if a configuration option is not explicitly+-- specified anywhere, a reasonable default is used. The configuration file has a format, wherein+-- one configuration option is specified on each line; the first token on each line is the name+-- of the configuration option, followed by whitespace, followed by its value. Lines beginning with #+-- are comments. Thus:+--+-- > # This is a sample configuration file+-- > cfgHostName host3+-- > cfgKnownHosts host1 host2 host3 host4+--+-- Options may be specified on the command line similarly. Note that command-line arguments containing spaces must be quoted.+--+-- > ./MyProgram -cfgHostName=host3 -cfgKnownHosts='host1 host2 host3 host4'+readConfig :: Bool -> Maybe FilePath -> IO Config+readConfig useargs fp = do a <- safety (processConfigFile emptyConfig) ("while reading config file ")+                           b <- safety (processArgs a) "while parsing command line "+                           return b+     where  processConfigFile from = maybe (return from) (readConfigFile from) fp+            processArgs from = if useargs+                                  then readConfigArgs from+                                  else return from+            safety o s = do res <- try o :: IO (Either SomeException Config)+                            either (\e -> throw $ ConfigException $ s ++ show e) (return) res+            readConfigFile from afile = do contents <- readFile afile+                                           evaluate $ processConfig (lines contents) from+            readConfigArgs from = do args <- getArgs+                                     c <- evaluate $ processConfig (fst $ collectArgs args) from+                                     return $ c {cfgArgs = reverse $ snd $ collectArgs args}+            collectArgs the = foldl findArg ([],[]) the+            findArg (last,n) ('-':this) | isPrefixOf "cfg" this = ((map (\x -> if x=='=' then ' ' else x) this):last,n)+            findArg (last,n) a          = (last,a:n)++processConfig :: [String] -> Config -> Config+processConfig rawLines from = foldl processLine from rawLines+ where+  processLine cfg line = case words line of+                            ('#':_):_ -> cfg+                            [] -> cfg+                            [option,val] -> updateCfg cfg option val+                            option:rest | (not . null) rest -> updateCfg cfg option (unwords rest)+                            _ -> error $ "Bad configuration syntax: " ++ line+  updateCfg cfg "cfgRole" role = cfg {cfgRole=clean role}+  updateCfg cfg "cfgHostName" hn = cfg {cfgHostName=clean hn}+  updateCfg cfg "cfgListenPort" p = cfg {cfgListenPort=(read.isInt) p}+  updateCfg cfg "cfgPeerDiscoveryPort" p = cfg {cfgPeerDiscoveryPort=(read.isInt) p}+  updateCfg cfg "cfgRoundtripTimeout" p = cfg {cfgRoundtripTimeout=(read.isInt) p}+  updateCfg cfg "cfgPromiseFlushDelay" p = cfg {cfgPromiseFlushDelay=(read.isInt) p}+  updateCfg cfg "cfgMaxOutgoing" p = cfg {cfgMaxOutgoing=(read.isInt) p}+  updateCfg cfg "cfgPromisePrefix" p = cfg {cfgPromisePrefix=p}+  updateCfg cfg "cfgLocalRegistryListenPort" p = cfg {cfgLocalRegistryListenPort=(read.isInt) p}+  updateCfg cfg "cfgKnownHosts" m = cfg {cfgKnownHosts=words m}+  updateCfg cfg "cfgNetworkMagic" m = cfg {cfgNetworkMagic=clean m}+  updateCfg _   opt _ = error ("Unknown configuration option: "++opt)+  isInt s | all isDigit s = s+  isInt s = error ("Not a good number: "++s)+{- UNUSED+  nonempty s | (not.null) s = s+  nonempty b = error ("Unexpected empty item: " ++ b)+-}+  clean = filter (not.isSpace)++----------------------------------------------+-- * Logging+----------------------------------------------++-- | Specifies the importance of a particular log entry.+-- Can also be used to filter log output.+data LogLevel = LoSay -- ^ Non-suppressible application-level emission+              | LoFatal+              | LoCritical +              | LoImportant+              | LoStandard -- ^ The default log level+              | LoInformation+              | LoTrivial +                deriving (Eq,Ord,Enum,Show)++instance Binary LogLevel where+   put n = put $ fromEnum n+   get = get >>= return . toEnum++-- | Specifies the subsystem or region that is responsible for+-- generating a given log entry. This is useful in conjunction+-- with 'LogFilter' to limit displayed log output to the+-- particular area of your program that you are currently debugging.+-- The SYS, TSK, and SAY spheres are used by the framework+-- for messages relating to the Process layer, the Task layer,+-- and the 'say' function.+-- The remainder of values are free for use at the application level.+type LogSphere = String++-- | A preference as to what is done with log messages+data LogTarget = LtStdout -- ^ Messages will be output to the console; the default+               | LtForward NodeId  -- ^ Log messages will be forwarded to the given node; please don't set up a loop+               | LtFile FilePath -- ^ Log messages will be appended to the given file+               | LtForwarded  -- ^ Special value -- don't set this in your LogConfig!+               deriving (Typeable)++instance Binary LogTarget where+     put LtStdout = putWord8 0+     put (LtForward nid) = putWord8 1 >> put nid+     put (LtFile fp) = putWord8 2 >> put fp+     put LtForwarded = putWord8 3+     get = do a <- getWord8+              case a of+                0 -> return LtStdout+                1 -> get >>= return . LtForward+                2 -> get >>= return . LtFile+                3 -> return LtForwarded++-- | Specifies which log messages will be output. +-- All log messages of importance below the current+-- log level or not among the criterea given here+-- will be suppressed. This type lets you limit+-- displayed log messages to certain components.+data LogFilter = LfAll+               | LfOnly [LogSphere]+               | LfExclude [LogSphere] deriving (Typeable)++instance Binary LogFilter where+      put LfAll = putWord8 0+      put (LfOnly ls) = putWord8 1 >> put ls+      put (LfExclude le) = putWord8 2 >> put le+      get = do a <- getWord8+               case a of+                  0 -> return LfAll+                  1 -> get >>= return . LfOnly+                  2 -> get >>= return . LfExclude++-- | Expresses a current configuration of the logging+-- subsystem, which determines which log messages to +-- be output and where to send them when they are.+-- Both processes and nodes have log configurations,+-- set with 'setLogConfig' and 'setNodeLogConfig'+-- respectively. The node log configuration is+-- used for all processes that have not explicitly+-- set their log configuration. Otherwise, the+-- process log configuration takes priority.+data LogConfig = LogConfig+     {+       logLevel :: LogLevel, -- ^ The lowest message priority that will be displayed+       logTarget :: LogTarget, -- ^ Where to send messages+       logFilter :: LogFilter -- ^ Other filtering+     } deriving (Typeable)++instance Binary LogConfig where+   put (LogConfig ll lt lf) = put ll >> put lt >> put lf+   get = do ll <- get+            lt <- get+            lf <- get+            return $ LogConfig ll lt lf++data LogMessage = LogMessage UTCTime LogLevel LogSphere String ProcessId LogTarget+                | LogUpdateConfig LogConfig deriving (Typeable)++-- This correct instance of UTCTime+-- works on 32-bit systems and is+-- courtesy of Warren Harris+instance Binary UTCTime where+  put t = do+    let d = toModifiedJulianDay (utctDay t)+        s = realToFrac $ utctDayTime t :: Pico+        ps = truncate $ s * 1e12 :: Integer+    put d >> put ps+  get = do+    d <- get+    ps <- get+    return $ UTCTime (ModifiedJulianDay d) (picosecondsToDiffTime ps)++instance Binary LogMessage where+  put (LogMessage utc ll ls s pid target) = putWord8 0 >> put utc >> put ll >> put ls >> put s >> put pid >> put target+  put (LogUpdateConfig lc) = putWord8 1 >> put lc+  get = do a <- getWord8+           case a of+              0 -> do utc <- get+                      ll <- get+                      ls <- get+                      s <- get+                      pid <- get+                      target <- get+                      return $ LogMessage utc ll ls s pid target+              1 -> do lc <- get+                      return $ LogUpdateConfig lc++instance Show LogMessage where+  show (LogMessage utc ll ls s pid _) = concat [paddedString 30 (show utc)," ",(show $ fromEnum ll)," ",paddedString 28 (show pid)," ",ls," ",s]+  show (LogUpdateConfig _) = "LogUpdateConfig"++showLogMessage :: LogMessage -> IO String+showLogMessage (LogMessage utc ll ls s pid _) = +   do gmt <- utcToLocalZonedTime utc+      return $ concat [paddedString 30 (show gmt)," ",+                       (show $ fromEnum ll)," ",+                       paddedString 28 (show pid)," ",ls," ",s]++-- | The default log configuration represents+-- a starting point for setting your own+-- configuration. It is:+--+-- > logLevel = LoStandard+-- > logTarget = LtStdout+-- > logFilter = LfAll+defaultLogConfig :: LogConfig+defaultLogConfig = LogConfig+                   {+                      logLevel = LoStandard,+                      logTarget = LtStdout,+                      logFilter = LfAll+                   }++logApplyFilter :: LogConfig -> LogSphere -> LogLevel -> Bool+logApplyFilter cfg sph lev = filterSphere (logFilter cfg) +                          && filterLevel (logLevel cfg)+    where filterSphere LfAll = True+          filterSphere (LfOnly lst) = elem sph lst+          filterSphere (LfExclude lst) = not $ elem sph lst+          filterLevel ll = lev <= ll++-- | Uses the logging facility to produce non-filterable, programmatic output. Shouldn't be used+-- for informational logging, but rather for application-level output.+say :: String -> ProcessM ()+say v = v `seq` logS "SAY" LoSay v++-- | Gets the currently active log configuration+-- for the current process; if the current process+-- doesn't have a log configuration set, the process's+-- log configuration will be returned+getLogConfig :: ProcessM LogConfig+getLogConfig = do p <- getProcess+                  case (prLogConfig p) of+                    Just lc -> return lc+                    Nothing -> do node <- liftIO $ readMVar (prNodeRef p)+                                  return (ndLogConfig node)++-- | Set the process's log configuration. This overrides+-- any node-level log configuration+setLogConfig :: LogConfig -> ProcessM ()+setLogConfig lc = do p <- getProcess+                     putProcess (p {prLogConfig = Just lc})++-- | Sets the node's log configuration+setNodeLogConfig :: LogConfig -> ProcessM ()+setNodeLogConfig lc = do p <- getProcess+                         liftIO $ modifyMVar_ (prNodeRef p) (\x -> return $ x {ndLogConfig = lc})++-- | Sets the log configuration of a remote node.+-- May throw TransmitException+setRemoteNodeLogConfig :: NodeId -> LogConfig -> ProcessM ()+setRemoteNodeLogConfig nid lc = do res <- sendSimple (adminGetPid nid ServiceLog) (LogUpdateConfig lc) PldAdmin+                                   case res of +                                     QteOK -> return ()+                                     _n -> throw $ TransmitException $ QteLoggingError++logI :: MVar Node -> ProcessId -> LogSphere -> LogLevel -> String -> IO ()+logI mnode pid sph ll txt = do node <- readMVar mnode+                               when (filtered (ndLogConfig node))+                                 (sendMsg (ndLogConfig node))+         where filtered cfg = logApplyFilter cfg sph ll+               makeMsg cfg = do time <- getCurrentTime+                                return $ LogMessage time ll sph txt pid (logTarget cfg)+               sendMsg cfg = do msg <- makeMsg cfg+                                res <- let svc = (adminGetPid nid ServiceLog)+                                           nid = nodeFromPid pid +                                       in sendBasic mnode svc msg (Nothing::Maybe()) PldAdmin Nothing+                                case res of+                                    _ -> return () -- ignore error -- what can I do?+                                +                                  +-- | Generates a log entry, using the process's current logging configuration.+--+-- * 'LogSphere' indicates the subsystem generating this message. SYS in the case of componentes of the framework.+--+-- * 'LogLevel' indicates the importance of the message.+--+-- * The third parameter is the log message.+--+-- Both of the first two parameters may be used to filter log output.+logS :: LogSphere -> LogLevel -> String -> ProcessM ()+logS sph ll txt = do lc <- txt `seq` getLogConfig+                     when (filtered lc)+                         (sendMsg lc)+         where filtered cfg = logApplyFilter cfg sph ll+               makeMsg cfg = do time <- liftIO $ getCurrentTime+                                pid <- getSelfPid+                                return $ LogMessage time ll sph txt pid (logTarget cfg)+               sendMsg cfg = +                 do msg <- makeMsg cfg+                    nid <- getSelfNode+                    res <- let svc = (adminGetPid nid ServiceLog)+                               in sendSimple svc msg PldAdmin+                    case res of+                      QteOK -> return ()+                      _n -> throw $ TransmitException $ QteLoggingError++startLoggingService :: ProcessM ()+startLoggingService = serviceThread ServiceLog logger+   where logger = receiveWait [matchLogMessage,matchUnknownThrow] >> logger+         matchLogMessage = match (\msg ->+           do mylc <- getLogConfig+              case msg of+                (LogMessage _ _ _ _ _ LtForwarded) -> processMessage (logTarget mylc) msg+                (LogMessage _ _ _ _ _ _) -> processMessage (targetPreference msg) msg+                (LogUpdateConfig lc) -> setNodeLogConfig lc)+         targetPreference (LogMessage _ _ _ _ _ a) = a+         forwardify (LogMessage a b c d e _) = LogMessage a b c d e LtForwarded+         processMessage whereto txt =+          do smsg <- liftIO $ showLogMessage txt+             case whereto of+              LtStdout -> liftIO $ putStrLn smsg+              LtFile fp -> (ptry (liftIO (appendFile fp (smsg ++ "\n"))) :: ProcessM (Either IOError ()) ) >> return () -- ignore error - what can we do?+              LtForward nid -> do self <- getSelfNode+                                  when (self /= nid) +                                    (sendSimple (adminGetPid nid ServiceLog) (forwardify txt) PldAdmin >> return ()) -- ignore error -- what can we do?+              _n -> throw $ ConfigException $ "Invalid message forwarded setting"+++----------------------------------------------+-- * Node monitoring+----------------------------------------------++data NodeMonitorCommand = +           NodeMonitorStart NodeId+         | NodeMonitorPing+           deriving (Typeable)+instance Binary NodeMonitorCommand where +  put (NodeMonitorStart nid) = putWord8 0 >> put nid+  put (NodeMonitorPing) = putWord8 1+  get = do a <- getWord8+           case a of+              0 -> do b <- get+                      return $ NodeMonitorStart b+              1 -> return NodeMonitorPing++data NodeMonitorSignal = +         NodeMonitorNodeFailure NodeId deriving (Typeable)+instance Binary NodeMonitorSignal where +   put (NodeMonitorNodeFailure nid) = put nid+   get = do a <- get+            return $ NodeMonitorNodeFailure a++data NodeMonitorInformation = +           NodeMonitorNodeDown NodeId+           deriving (Typeable)+instance Binary NodeMonitorInformation where +   put (NodeMonitorNodeDown nid) = put nid+   get = do a <- get+            return $ NodeMonitorNodeDown a++monitorNode :: NodeId -> ProcessM ()+monitorNode nid = do +                     mynid <- getSelfNode+                     res <- roundtripQueryLocal PldAdmin (adminGetPid mynid ServiceNodeMonitor) (NodeMonitorStart nid)+                     case res of+                       Right () -> return ()+                       _ -> (throw $ ServiceException $ "while contacting node monitor "++show res)++-- | Sends a small message to the specified node to determine if it's alive.+-- If the node cannot be reached or does not respond within a time frame, the function+-- will return False.+pingNode :: NodeId -> ProcessM Bool+pingNode nid = do res <- roundtripQueryUnsafe PldAdmin (adminGetPid nid ServiceNodeMonitor) (NodeMonitorPing)   +                  case res of+                       Right () -> return True+                       _ -> return False++-- TODO this can be re-engineered to avoid setting up and tearing down TCP connections and threads+-- at every ping. Instead open up a TCP connection to the pingee and require it to send us a hello+-- every N seconds or die. Also, this has a bug in that the number of failures is not reset after a successful ping+startNodeMonitorService :: ProcessM ()+startNodeMonitorService = serviceThread ServiceNodeMonitor (service Map.empty)+  where+    service state = +      let matchCommand cmd = case cmd of+                               NodeMonitorPing -> return ((),state)+                               NodeMonitorStart nid -> do res <- addmonitor nid+                                                          return ((),res)+          matchSignal cmd = case cmd of+                               NodeMonitorNodeFailure nid -> do res <- handlefailure nid+                                                                return (res)+          listenaction nid mainpid =+             let onfailure = sendSimple mainpid (NodeMonitorNodeFailure nid)  PldUser >> return ()+                 loop = do res <- pingNode nid+                           case res of+                              False -> onfailure+                              True -> liftIO (threadDelay interpingtimeout) >> loop+             in+               do loop +          failurelimit = 3 -- TODO make configurable+          interpingtimeout = 5000000+          retrytimeout =  1000000+          reportfailure nid = do mynid <- getSelfNode+                                 sendSimple (adminGetPid mynid ServiceProcessMonitor) (GlNodeDown nid) PldAdmin+          handlefailure nid = case Map.lookup nid state of+                                    Just c -> if c >= failurelimit+                                                then do _ <- reportfailure nid+                                                        return (Map.delete nid state)+                                                else do mypid <- getSelfPid+                                                        _ <- spawnLocalAnd (liftIO (threadDelay retrytimeout) >> listenaction nid mypid) setDaemonic+                                                        return (Map.adjust succ nid state)+                                    Nothing -> return state+          addmonitor nid = case Map.member nid state of+                                  True -> return state+                                  False -> do mynid <- getSelfNode+                                              mypid <- getSelfPid+                                              if mynid==nid+                                                 then return state+                                                 else do _ <- spawnLocalAnd (listenaction nid mypid) setDaemonic+                                                         return $ Map.insert nid (0) state+                                                 +       in receiveWait [roundtripResponse matchCommand,+                       match matchSignal,+                       matchUnknownThrow] >>= service+        ++----------------------------------------------+-- * Service helpers+----------------------------------------------++data ServiceId =  +                 ServiceLog +               | ServiceSpawner +               | ServiceNodeRegistry +               | ServiceNodeMonitor +               | ServiceProcessMonitor +               | ServiceProcessRegistry+                 deriving (Ord,Eq,Enum,Show)++adminGetPid :: NodeId -> ServiceId -> ProcessId+adminGetPid nid sid = ProcessId nid (fromEnum sid)++adminDeregister :: ServiceId -> ProcessM ()+adminDeregister val = do p <- getProcess+                         pid <- getSelfPid+                         liftIO $ modifyMVar_ (prNodeRef p) (fun $ localFromPid pid)+            where fun pid node = return $ node {ndAdminProcessTable = Map.delete val (ndAdminProcessTable node)}++adminRegister :: ServiceId -> ProcessM ()+adminRegister val =  do p <- getProcess+                        pid <- getSelfPid+                        liftIO $ modifyMVar_ (prNodeRef p) (\node ->+                         if Map.member val (ndAdminProcessTable node) +                           then throw $ ServiceException $ "Duplicate administrative registration at index " ++ show val    +                           else (fun (localFromPid pid) node))+            where fun pid node = return $ node {ndAdminProcessTable = Map.insert val pid (ndAdminProcessTable node)}++{- UNUSED+adminLookup :: ServiceId -> ProcessM LocalProcessId+adminLookup val = do p <- getProcess+                     node <- liftIO $ readMVar (prNodeRef p)+                     case Map.lookup val (ndAdminProcessTable node) of+                        Nothing -> throw $ ServiceException $ "Request for unknown administrative service " ++ show val+                        Just x -> return x+-}++adminLookupN :: ServiceId -> MVar Node -> IO (Either TransmitStatus LocalProcessId)+adminLookupN val mnode = +                  do node <- readMVar mnode+                     case Map.lookup val (ndAdminProcessTable node) of+                        Nothing -> return $ Left QteUnknownPid+                        Just x -> return $ Right x++startFinalizerService :: ProcessM () -> ProcessM ()+startFinalizerService todo = spawnLocalAnd body prefix >> return ()+                 where prefix = setDaemonic+                       body = +                             do p <- getProcess+                                node <- liftIO $ readMVar (prNodeRef p)+                                liftIO $ takeMVar (ndNodeFinalizer node)+                                todo `pfinally` (liftIO $ putMVar (ndNodeFinalized node) ())+++performFinalization :: MVar Node -> IO ()+performFinalization mnode = do node <- readMVar mnode+                               putMVar (ndNodeFinalizer node) ()+                               takeMVar (ndNodeFinalized node)                              ++setDaemonic :: ProcessM ()+setDaemonic = do p <- getProcess+                 pid <- getSelfPid+                 liftIO $ modifyMVar_ (prNodeRef p) +                    (\node -> return $ node {ndProcessTable=Map.adjust (\pte -> pte {pteDaemonic=True}) (localFromPid pid) (ndProcessTable node)})++serviceThread :: ServiceId -> ProcessM () -> ProcessM ()+serviceThread v f = spawnLocalAnd (pbracket (return ())+                             (\_ -> adminDeregister v >> logError)+                             (\_ -> f)) (adminRegister v >> setDaemonic)+                        >> (return()) +          where logError = logS "SYS" LoFatal $ "System process "++show v++" has terminated" -- TODO maybe restart?+++----------------------------------------------+-- * Process registry service +----------------------------------------------+++data ProcessRegistryState = ProcessRegistryState (Map.Map String ProcessId) (Map.Map ProcessId String)++data ProcessRegistryCommand = ProcessRegistryQuery String (Maybe (Closure (ProcessM ())))+                            | ProcessRegistrySet String ProcessId deriving (Typeable)+instance Binary ProcessRegistryCommand where+  put (ProcessRegistryQuery a b) = putWord8 0 >> put a >> put b+  put (ProcessRegistrySet a b) = putWord8 1 >> put a >> put b+  get = do cmd <- getWord8+           case cmd of+             0 -> do a <- get+                     b <- get+                     return $ ProcessRegistryQuery a b+             1 -> do a <- get+                     b <- get+                     return $ ProcessRegistrySet a b+data ProcessRegistryAnswer = ProcessRegistryResponse (Maybe ProcessId)+                           | ProcessRegistryError String deriving (Typeable)+instance Binary ProcessRegistryAnswer where+  put (ProcessRegistryResponse a) = putWord8 0 >> put a+  put (ProcessRegistryError s) = putWord8 1 >> put s+  get = do a <- getWord8+           case a of+             0 -> get >>= (return . ProcessRegistryResponse)+             1 -> get >>= (return . ProcessRegistryError)++startProcessRegistryService :: ProcessM ()+startProcessRegistryService = serviceThread ServiceProcessRegistry (service initialState)+  where+    initialState = ProcessRegistryState Map.empty Map.empty+    service state@(ProcessRegistryState nameToPid pidToName) = +      let+        downs (ProcessMonitorException pid _why) =+          case Map.lookup pid pidToName of+            Just name ->+              let newPidToName = Map.delete pid pidToName+                  newNameToPid = Map.delete name nameToPid+               in return (ProcessRegistryState newNameToPid newPidToName)+            Nothing -> return state+        cmds cmd = +          case cmd of+            ProcessRegistrySet name pid ->+              case (Map.lookup pid pidToName, Map.lookup name nameToPid) of+                (Nothing,Nothing) -> +                   let newNameToPid = Map.insert name pid nameToPid+                       newPidToName = Map.insert pid name pidToName+                    in do islocal <- isPidLocal pid+                          case islocal of+                             True -> +                               do mypid <- getSelfPid+                                  ok <- monitorProcessQuiet mypid pid MaMonitor+                                  case ok of+                                    True -> return (ProcessRegistryResponse Nothing,ProcessRegistryState newNameToPid newPidToName)+                                    False -> return (ProcessRegistryError $ "Couldn't establish monitoring of task in naming "++name,state)+                             False -> return (ProcessRegistryError $ "Refuse to register nonlocal process" ++ show pid,state)+                (Nothing,_) -> return (ProcessRegistryError $ "The name "++name++" has already been registered",state)+                (_,_) -> return (ProcessRegistryError $ "The process "++show pid++" has already been registered",state)+            ProcessRegistryQuery name mClo ->+              case Map.lookup name nameToPid of+                Just pid -> return (ProcessRegistryResponse (Just pid),state)+                Nothing -> +                  case mClo of+                    Nothing -> return (ProcessRegistryResponse Nothing,state)+                    Just clo -> do mynid <- getSelfNode+                                   mypid <- getSelfPid+                                   pid <- spawnAnd mynid clo defaultSpawnOptions {amsoMonitor=Just (mypid,MaMonitor)}+                                   let newNameToPid = Map.insert name pid nameToPid+                                       newPidToName = Map.insert pid name pidToName+                                   return (ProcessRegistryResponse (Just pid),ProcessRegistryState newNameToPid newPidToName)+       in receiveWait [roundtripResponse cmds,match downs] >>= service+++-- TODO nameQueryOrWait :: NodeId -> String -> ProcessM ProcessId++-- | Similar to 'nameQuery' but if the named process doesn't exist,+-- it will be started from the given closure. If the process is+-- already running, the closure will be ignored.+nameQueryOrStart :: NodeId -> String -> Closure (ProcessM ()) -> ProcessM ProcessId+nameQueryOrStart nid name clo =+  let servicepid = adminGetPid nid ServiceProcessRegistry+      msg = ProcessRegistryQuery name (Just clo)+   in do res <- roundtripQueryUnsafe PldAdmin servicepid msg+         case res of+            Right (ProcessRegistryResponse (Just answer)) -> return answer+            Right (ProcessRegistryError s) -> throw $ ServiceException s+            _ -> throw $ ServiceException $ "Crazy talk from process registry"++-- | Query the PID of a named process on a particular node.+-- If no process of that name exists, or if that+-- process has ended, this function returns Nothing.+nameQuery :: NodeId -> String -> ProcessM (Maybe ProcessId)+nameQuery nid name =+  let servicepid = adminGetPid nid ServiceProcessRegistry+      msg = ProcessRegistryQuery name Nothing+   in do res <- roundtripQueryUnsafe PldAdmin servicepid msg+         case res of+            Right (ProcessRegistryResponse answer) -> return answer+            Right (ProcessRegistryError s) -> throw $ ServiceException s+            _ -> throw $ ServiceException $ "Crazy talk from process registry"++-- | Assigns a name to the current process. The name is local to the+-- node. On each node, each process may have only one name, and each+-- name may be given to only one node. If this function is called+-- more than once by the same process, or called more than once+-- with the name on a single node, it will throw a 'ServiceException'.+-- The PID of a named process can be queried later with 'nameQuery'. When the+-- named process ends, its name will again become available.+-- One reason to use named processes is to create node-local state.+-- This example lets each node have its own favorite color, which can+-- be changed and queried.+--+-- > nodeFavoriteColor :: ProcessM ()+-- > nodeFavoriteColor =+-- >  do nameSet "favorite_color"+-- >     loop Blue+-- >  where loop color =+-- >      receiveWait+-- >         [ match (\newcolor -> return newcolor),+-- >           match (\pid -> send pid color >> return color)+-- >         ] >>= loop+-- >+-- > setFavoriteColor :: NodeId -> Color -> ProcessM ()+-- > setFavoriteColor nid color =+-- >  do (Just pid) <- nameQuery nid "favorite_color"+-- >     send pid color+-- >+-- > getFavoriteColor :: NodeId -> ProcessM Color+-- > getFavoriteColor nid =+-- >  do (Just pid) <- nameQuery nid "favorite_color"+-- >     mypid <- getSelfPid+-- >     send pid mypid+-- >     expect+nameSet :: String -> ProcessM ()+nameSet name = +  do mynid <- getSelfNode+     mypid <- getSelfPid+     let servicepid = adminGetPid mynid ServiceProcessRegistry+         msg = ProcessRegistrySet name mypid+     res <- roundtripQueryLocal PldAdmin servicepid msg+     case res of+        Right (ProcessRegistryResponse Nothing) -> return ()+        Right (ProcessRegistryError s) -> throw $ ServiceException s+        _ -> throw $ ServiceException $ "Crazy talk from process registry"+++++----------------------------------------------+-- * Global and spawn service +----------------------------------------------++data GlSignal = GlProcessDown ProcessId SignalReason+              | GlNodeDown NodeId deriving (Typeable,Show)+instance Binary GlSignal where+   put (GlProcessDown a b) = putWord8 0 >> put a >> put b+   put (GlNodeDown a) = putWord8 1 >> put a+   get = do ch <- getWord8+            case ch of+              0 -> do a <- get+                      b <- get+                      return $ GlProcessDown a b+              1 -> do a <- get+                      return $ GlNodeDown a++data GlSynchronous = GlRequestMonitoring ProcessId ProcessId MonitorAction+                   | GlRequestMoniteeing ProcessId NodeId +                   | GlRequestUnmonitoring ProcessId ProcessId MonitorAction deriving (Typeable)+instance Binary GlSynchronous where+     put (GlRequestMonitoring a b c) = putWord8 0 >> put a >> put b >> put c+     put (GlRequestMoniteeing a b) = putWord8 1 >> put a >> put b+     put (GlRequestUnmonitoring a b c) = putWord8 2 >> put a >> put b >> put c+     get = do ch <- getWord8+              case ch of+                 0 -> do a <- get+                         b <- get+                         c <- get+                         return $  GlRequestMonitoring a b c+                 1 -> do a <- get+                         b <- get+                         return $ GlRequestMoniteeing a b+                 2 -> do a <- get+                         b <- get+                         c <- get+                         return $  GlRequestUnmonitoring a b c++data GlCommand = GlMonitor ProcessId ProcessId MonitorAction+               | GlUnmonitor ProcessId ProcessId MonitorAction deriving (Typeable)+instance Binary GlCommand where+   put (GlMonitor a b c) = putWord8 0 >> put a >> put b >> put c+   put (GlUnmonitor a b c) = putWord8 1 >> put a >> put b >> put c+   get = do ch <- getWord8+            case ch of+               0 -> do a <- get+                       b <- get+                       c <- get+                       return $ GlMonitor a b c+               1 -> do a <- get+                       b <- get+                       c <- get+                       return $ GlUnmonitor a b c++-- | The different kinds of monitoring available between processes.+data MonitorAction = MaMonitor -- ^ MaMonitor means that the monitor process will be sent a ProcessDownException message when the monitee terminates for any reason.+                   | MaLink -- ^ MaLink means that the monitor process will receive an asynchronous exception of type ProcessDownException when the monitee terminates for any reason++                   | MaLinkError -- ^ MaLinkError means that the monitor process will receive an asynchronous exception of type ProcessDownException when the monitee terminates abnormally++                   deriving (Typeable,Show,Ord,Eq)+instance Binary MonitorAction where +  put MaMonitor = putWord8 0+  put MaLink = putWord8 1+  put MaLinkError = putWord8 2+  get = getWord8 >>= \x ->+        case x of+          0 -> return MaMonitor+          1 -> return MaLink+          2 -> return MaLinkError++-- | Part of the notification system of process monitoring, indicating why the monitor is being notified.+data SignalReason = SrNormal  -- ^ the monitee terminated normally+                  | SrException String -- ^ the monitee terminated with an uncaught exception, which is given as a string+                  | SrNoPing -- ^ the monitee is believed to have ended or be inaccessible, as the node on which its running is not responding to pings. This may indicate a network bisection or that the remote node has crashed.+                  | SrInvalid -- ^ SrInvalid: the monitee was not running at the time of the attempt to establish monitoring+                    deriving (Typeable,Show)+instance Binary SignalReason where +     put SrNormal = putWord8 0+     put (SrException s) = putWord8 1 >> put s+     put SrNoPing = putWord8 2+     put SrInvalid = putWord8 3+     get = do a <- getWord8+              case a of+                 0 -> return SrNormal+                 1 -> get >>= return . SrException+                 2 -> return SrNoPing+                 3 -> return SrInvalid++type GlLinks = Map.Map ProcessId +                  (Map.Map (LocalProcessId,MonitorAction) (Int),+                   Map.Map LocalProcessId (Int),+                   Map.Map NodeId ())++data GlobalData = GlobalData+     {+        glLinks :: GlLinks,+        glNextId :: Integer,+        glSyncs :: Map.Map Integer (GlobalData -> MatchM GlobalData ())+     }++gdCombineEntry :: (Map.Map (LocalProcessId,MonitorAction) (Int),+                   Map.Map LocalProcessId (Int),+                   Map.Map NodeId ()) -> (Map.Map (LocalProcessId,MonitorAction) (Int),+                   Map.Map LocalProcessId (Int),+                   Map.Map NodeId ()) -> (Map.Map (LocalProcessId,MonitorAction) (Int),+                   Map.Map LocalProcessId (Int),+                   Map.Map NodeId ())+gdCombineEntry (newmonitors,newmonitees,newnodes) (oldmonitors,oldmonitees,oldnodes) = +    let finalnodes = Map.unionWith const newnodes oldnodes+        finalmonitors = Map.unionWith (+) newmonitors oldmonitors+        finalmonitees = Map.unionWith (+) newmonitees oldmonitees+     in (finalmonitors,finalmonitees,finalnodes)++gdAddMonitor :: GlLinks -> ProcessId -> MonitorAction -> LocalProcessId -> GlLinks+gdAddMonitor gl pid ma lpid = +   Map.insertWith' gdCombineEntry pid (Map.singleton (lpid,ma) 1, Map.empty,Map.empty) gl++gdDelMonitor :: GlLinks -> ProcessId -> MonitorAction -> LocalProcessId -> GlLinks+gdDelMonitor gl pid ma lpid = Map.adjust editentry pid gl+  where editentry (mons,mots,ns) = (fixmons mons,mots,ns)+        fixmons m = Map.update (\a -> if pred a == 0+                                         then Nothing+                                         else Just (pred a)) (lpid,ma) m++gdAddMonitee :: GlLinks -> ProcessId -> LocalProcessId -> GlLinks+gdAddMonitee gl pid lpid = +   Map.insertWith' gdCombineEntry pid (Map.empty,Map.singleton lpid 1,Map.empty) gl++gdDelMonitee :: GlLinks -> ProcessId -> LocalProcessId -> GlLinks+gdDelMonitee gl pid lpid = Map.adjust editentry pid gl+  where editentry (mons,mots,ns) = (mons,fixmons mots,ns)+        fixmons m = Map.update (\a -> if pred a == 0+                                         then Nothing+                                         else Just (pred a)) lpid m++glExpungeProcess :: GlLinks -> ProcessId -> NodeId -> GlLinks+glExpungeProcess gl pid myself = +                    let mine n = buildPidFromNodeId myself n+                     in case Map.lookup pid gl of+                             Nothing -> gl+                             Just (mons,mots,_ns) -> +                                  let s1 = Map.delete pid gl+                                      s2 = foldl' (\g (lp,_)-> Map.delete (mine lp) g) s1 (Map.keys mons)+                                      s3 = foldl' (\g lp -> Map.delete (mine lp) g) s2 (Map.keys mots)+                                   in s3++gdAddNode :: GlLinks -> ProcessId -> NodeId -> GlLinks+gdAddNode gl pid nid = Map.insertWith' gdCombineEntry pid (Map.empty,Map.empty,Map.singleton nid ()) gl++-- | The main form of notification to a monitoring process that a monitored process has terminated.+-- This data structure can be delivered to the monitor either as a message (if the monitor is+-- of type 'MaMonitor') or as an asynchronous exception (if the monitor is of type 'MaLink' or 'MaLinkError').+-- It contains the PID of the monitored process and the reason for its nofication.+data ProcessMonitorException = ProcessMonitorException ProcessId SignalReason deriving (Typeable)++instance Binary ProcessMonitorException where+  put (ProcessMonitorException pid sr) = put pid >> put sr+  get = do pid <- get+           sr <- get+           return $ ProcessMonitorException pid sr++instance Exception ProcessMonitorException++instance Show ProcessMonitorException where+  show (ProcessMonitorException pid why) = "ProcessMonitorException: " ++ show pid ++ " has terminated because "++show why++-- | Establishes bidirectional abnormal termination monitoring between the current+-- process and another. Monitoring established with linkProcess+-- is bidirectional and signals only in the event of abnormal termination.+-- In other words, @linkProcess a@ is equivalent to:+--+-- > monitorProcess mypid a MaLinkError+-- > monitorProcess a mypid MaLinkError+linkProcess :: ProcessId -> ProcessM ()+linkProcess p = do mypid <- getSelfPid +                   mynid <- getSelfNode+                   let servicepid = (adminGetPid mynid ServiceProcessMonitor) +                       msg1 = GlMonitor mypid p MaLinkError+                       msg2 = GlMonitor p mypid MaLinkError+                   res1 <- roundtripQueryLocal PldAdmin servicepid msg1+                   case res1 of +                      Right QteOK -> do res2 <- roundtripQueryLocal PldAdmin servicepid msg2+                                        case res2 of +                                           Right QteOK -> return ()+                                           Right err -> herr err+                                           Left err -> herr err+                      Right err -> herr err+                      Left err -> herr err+        where herr err = throw $ ServiceException $ "Error when linking process " ++ show p ++ ": " ++ show err++-- | A specialized version of 'match' (for use with 'receive', 'receiveWait' and friends) for catching process down+-- messages. This way processes can avoid waiting forever for a response from another process that has crashed.+-- Intended to be used within a 'withMonitor' block, e.g.:+--+-- > withMonitor apid $+-- >   do send apid QueryMsg+-- >      receiveWait +-- >      [+-- >        match (\AnswerMsg -> return "ok"),+-- >        matchProcessDown apid (return "aborted")   +-- >      ]+matchProcessDown :: ProcessId -> ProcessM q -> MatchM q ()+matchProcessDown pid f = matchIf (\(ProcessMonitorException p _) -> p==pid) (const f)++-- | Establishes temporary monitoring of another process. The process to be monitored is given in the+-- first parameter, and the code to run in the second. If the given process goes down while the code+-- in the second parameter is running, a process down message will be sent to the current process,+-- which can be handled by 'matchProcessDown'. +withMonitor :: ProcessId -> ProcessM a -> ProcessM a+withMonitor pid f = withMonitoring pid MaMonitor f++withMonitoring :: ProcessId -> MonitorAction -> ProcessM a -> ProcessM a+withMonitoring pid how f =  +                        do mypid <- getSelfPid+                           monitorProcess mypid pid how -- TODO if this throws a ServiceException, translate that into a trigger+                           a <- f `pfinally` safety (unmonitorProcess mypid pid how)+                           return a+              where safety n = ptry n :: ProcessM (Either ServiceException ()) +++-- | Establishes unidirectional processing of another process. The format is:+--+-- > monitorProcess monitor monitee action+--+-- Here,+--+-- * monitor is the process that will be notified if the monitee goes down+--+-- * monitee is the process that will be monitored+--+-- * action determines how the monitor will be notified+--+-- Monitoring will remain in place until one of the processes ends or until+-- 'unmonitorProcess' is called. Calls to 'monitorProcess' are cumulative,+-- such that calling 'monitorProcess' 3 three times on the same pair of processes+-- will ensure that monitoring will stay in place until 'unmonitorProcess' is called+-- three times on the same pair of processes.+-- If the monitee is not currently running, the monitor will be signalled immediately.+-- See also 'MonitorAction'.+monitorProcess :: ProcessId -> ProcessId -> MonitorAction -> ProcessM ()+monitorProcess monitor monitee how = monitorProcessImpl GlMonitor monitor monitee how True >> return ()++-- | Removes monitoring established by 'monitorProcess'. Note that the type of+-- monitoring, given in the third parameter, must match in order for monitoring+-- to be removed. If monitoring has not already been established between these+-- two processes, this function takes not action.+unmonitorProcess :: ProcessId -> ProcessId -> MonitorAction -> ProcessM ()+unmonitorProcess monitor monitee how = monitorProcessImpl GlUnmonitor monitor monitee how True >> return ()++monitorProcessQuiet :: ProcessId -> ProcessId -> MonitorAction -> ProcessM Bool+monitorProcessQuiet monitor monitee how = +  do res <- monitorProcessImpl GlMonitor monitor monitee how False+     case res of+       QteOK -> return True+       _ -> return False++monitorProcessImpl :: (ProcessId -> ProcessId -> MonitorAction -> GlCommand) -> ProcessId -> ProcessId -> MonitorAction -> Bool -> ProcessM TransmitStatus+monitorProcessImpl msgtype monitor monitee how throwit = +     let msg = msgtype monitor monitee how+      in do let servicepid = (adminGetPid (nodeFromPid monitee) ServiceProcessMonitor) +            res <- roundtripQueryUnsafe PldAdmin servicepid msg -- TODO: Prefer to send this message to the local service+            case res of+              Right QteOK -> return QteOK+              Right err -> herr err+              Left err -> herr err+  where herr err = +            case throwit of+              True -> throw $ ServiceException $ "Error when monitoring process " ++ show monitee ++ ": " ++ show err+              False -> return err++sendInterrupt :: (Exception e) => ProcessId -> e -> ProcessM Bool+sendInterrupt pid e = do islocal <- isPidLocal pid+                         case islocal of+                            False -> return False+                            True -> do p <- getProcess+                                       node <- liftIO $ readMVar (prNodeRef p)+                                       res <- liftIO $ atomically $ getProcessTableEntry node (localFromPid pid)+                                       case res of +                                          Just pte -> do liftIO $ throwTo (pteThread pte) e+                                                         return True+                                          Nothing -> return False++triggerMonitor :: ProcessId -> ProcessId -> MonitorAction -> SignalReason -> ProcessM ()+triggerMonitor towho aboutwho how why = +                 let +                      msg = ProcessMonitorException aboutwho why+                  in case how of+                       MaMonitor -> sendSimple towho msg PldUser  >> return ()+                       MaLink -> sendInterrupt towho msg  >> return ()+                       MaLinkError -> case why of+                                        SrNormal -> return ()+                                        _ -> sendInterrupt towho msg  >> return ()++startProcessMonitorService :: ProcessM ()+startProcessMonitorService = serviceThread ServiceProcessMonitor (service emptyGlobal)+  where +    emptyGlobal = GlobalData {glLinks=Map.empty, glNextId=0, glSyncs = Map.empty}+    getGlobalFor pid = adminGetPid (nodeFromPid pid) ServiceProcessMonitor+    checkliveness pid = do islocal <- isPidLocal pid+                           case islocal of+                             True -> isProcessUp (localFromPid pid)+                             False -> return True+    checklivenessandtrigger monitor monitee action = +                                            do a <- checkliveness monitee+                                               case a of+                                                  True -> return True+                                                  False -> do trigger monitor monitee action SrInvalid+                                                              return False+    trigger = triggerMonitor+{- UNUSED+    forward destinationnode msg = sendSimple (getGlobalFor destinationnode) msg PldAdmin+-}+    isProcessUp lpid = do p <- getProcess+                          node <- liftIO $ readMVar (prNodeRef p)+                          res <- liftIO $ atomically $ getProcessTableEntry node lpid+                          case res of+                            Nothing -> return (lpid<0)+                            Just _ -> return True+    removeLocalMonitee gl monitor monitee _action = +         gl {glLinks = gdDelMonitee (glLinks gl) monitor (localFromPid monitee) }+    removeLocalMonitor gl monitor monitee action =+         gl {glLinks = gdDelMonitor (glLinks gl) monitee action (localFromPid monitor) }+    addLocalMonitee gl monitor monitee _action =+         gl {glLinks = gdAddMonitee (glLinks gl) monitor (localFromPid monitee) }+    addLocalMonitor gl monitor monitee action = +         gl {glLinks = gdAddMonitor (glLinks gl) monitee action (localFromPid monitor) }+    addLocalNode gl monitor monitee _action =+         gl {glLinks = gdAddNode (glLinks gl) monitee (nodeFromPid monitor)}+    broadcast nids msg = mapM_ (\p -> forkProcessWeak $ ((ptimeout 5000000 $ sendSimple (adminGetPid p ServiceProcessMonitor) msg PldAdmin) >> return ())) nids+    handleProcessDown :: GlLinks -> ProcessId -> SignalReason -> ProcessM GlLinks+    handleProcessDown global pid why = +                     do islocal <- isPidLocal pid+                        mynid <- getSelfNode+                        case Map.lookup pid global of+                           Nothing -> return global+                           Just (monitors,_monitee,nodes) ->+                               do mapM_ (\(tellwho,how) -> trigger (buildPidFromNodeId mynid tellwho) pid how why) (Map.keys monitors)+                                  when (islocal)+                                    (broadcast (Map.keys nodes) (GlProcessDown pid why))+                                  mynid <- getSelfNode+                                  return $ glExpungeProcess global pid mynid+    service global = +         let+             additional = map (\receiver -> receiver global) (Map.elems (glSyncs global))+             matchPatterns = [match matchSignals,+                          roundtripResponseAsync matchCommands False,+                          roundtripResponse (matchSyncs global)] ++ additional+             matchSignals cmd = +                case cmd of+                  GlProcessDown pid why -> +                     do res <- handleProcessDown (glLinks global) pid why+                        return $ global { glLinks = res}+                  GlNodeDown nid -> let gl = glLinks global+                                        aslist = Map.keys gl+                                        pids = filter (\n -> nodeFromPid n == nid) aslist+                                     in do res <- foldM (\g p -> handleProcessDown g p SrNoPing) gl pids+                                           return $ global {glLinks = res}+             matchSyncs global cmd =+                case cmd of+                  GlRequestMonitoring monitee monitor action ->+                       let s1 = addLocalMonitor global monitor monitee action+                        in monitorNode (nodeFromPid monitee)+                             >> return (QteOK,s1)+                  GlRequestMoniteeing pid towho -> +                       do live <- checkliveness pid+                          case live of+                              True -> let s1 = global {glLinks = gdAddNode (glLinks global) pid towho}+                                       in return (QteOK,s1)+                              False -> return (QteUnknownPid,global)+                  GlRequestUnmonitoring monitee monitor action ->+                       let s1 = removeLocalMonitor global monitor monitee action+                        in return (QteOK,s1)+             matchCommands cmd ans = +                case cmd of+                  GlMonitor monitor monitee action -> +                    do ismoniteelocal <- isPidLocal monitee+                       ismonitorlocal <- isPidLocal monitor+                       case (ismoniteelocal,ismonitorlocal) of+                         (True,True) -> do live <- checklivenessandtrigger monitor monitee action+                                           case live of+                                              True -> let s1 = addLocalMonitee global monitor monitee action+                                                          s2 = addLocalMonitor s1 monitor monitee action+                                                       in ans QteOK >> return s2+                                              False -> ans QteOK >> return global+                         (True,False) -> do live <- checklivenessandtrigger monitor monitee action+                                            case live of+                                              False -> ans QteOK >> return global+                                              True -> let msg = GlRequestMonitoring monitee monitor action+                                                          receiver myId myGlobal myMsg =+                                                             let newGlobal = myGlobal {glSyncs = Map.delete myId (glSyncs myGlobal)}+                                                              in case myMsg of+                                                                   QteOK -> let s1 = addLocalNode newGlobal monitor monitee action+                                                                             in do _ <- ans QteOK+                                                                                   return s1+                                                                   err -> ans err >> return newGlobal+                                                       in do mmatch <- roundtripQueryImplSub PldAdmin (getGlobalFor monitor) msg (receiver (glNextId global))+                                                             case mmatch of+                                                                Left err -> ans err >> return global+                                                                Right mymatch -> return global {glNextId=glNextId global+1,+                                                                                                glSyncs=Map.insert (glNextId global) (mymatch) (glSyncs global)}+                         (False,True) -> let msg = GlRequestMoniteeing monitee (nodeFromPid monitor)+                                             receiver myId myGlobal myMsg =+                                               let newGlobal = myGlobal {glSyncs = Map.delete myId (glSyncs myGlobal)}+                                                in case myMsg of+                                                     QteOK -> let s1 = addLocalMonitor newGlobal monitor monitee action +                                                               in do monitorNode (nodeFromPid monitee)+                                                                     _ <- ans QteOK+                                                                     return s1 +                                                     QteUnknownPid -> do trigger monitor monitee action SrInvalid+                                                                         _ <- ans QteOK+                                                                         return newGlobal+                                                     err -> do _ <- ans err+                                                               return newGlobal+                                          in do mmatch <- roundtripQueryImplSub PldAdmin (getGlobalFor monitee) msg (receiver (glNextId global))+                                                case mmatch of+                                                   Left err -> ans err >> return global+                                                   Right mymatch -> return global {glNextId=glNextId global+1,+                                                                                   glSyncs=Map.insert (glNextId global) (mymatch) (glSyncs global)}+                         (False,False) -> do _ <- ans (QteOther "Requesting monitoring by third party node")+                                             return global+                  GlUnmonitor monitor monitee action -> +                    do ismoniteelocal <- isPidLocal monitee+                       ismonitorlocal <- isPidLocal monitor+                       case (ismoniteelocal,ismonitorlocal) of+                         (True,True) -> let s1 = removeLocalMonitee global monitor monitee action +                                            s2 = removeLocalMonitor s1 monitor monitee action+                                         in ans QteOK >> return s2+                         (True,False) -> let msg = GlRequestUnmonitoring monitee monitor action+                                             receiver myId myGlobal myMsg =+                                                let newGlobal = myGlobal {glSyncs=Map.delete myId (glSyncs myGlobal)}+                                                 in ans myMsg >> return newGlobal++                                          in do sync <- roundtripQueryImplSub PldAdmin (getGlobalFor monitee) msg (receiver (glNextId global))+                                                case sync of+                                                   Left err -> ans err >> return global+                                                   Right mymatch -> return global {glNextId=glNextId global+1,+                                                                                   glSyncs=Map.insert (glNextId global) (mymatch) (glSyncs global)} +                         (False,True) -> let s1 = removeLocalMonitor global monitor monitee action+                                          in ans QteOK >> return s1+                         (False,False) -> ans (QteOther "Requesting unmonitoring by third party node") >> return global+          in receiveWait matchPatterns >>= service++data AmSpawn = AmSpawn (Closure (ProcessM ())) AmSpawnOptions deriving (Typeable)+instance Binary AmSpawn where +    put (AmSpawn c o) =put c >> put o+    get=get >>= \c -> get >>= \o -> return $ AmSpawn c o+data AmCall = AmCall ProcessId (Closure Payload) deriving (Typeable)+instance Binary AmCall where+    put (AmCall pid clo) = put pid >> put clo+    get = do c <- get+             d <- get+             return $ AmCall c d++data AmSpawnOptions = AmSpawnOptions +        { +          amsoPaused :: Bool,+          amsoLink :: Maybe ProcessId,+          amsoMonitor :: Maybe (ProcessId,MonitorAction),+          amsoName :: Maybe String+        } deriving (Typeable) +instance Binary AmSpawnOptions where+    put (AmSpawnOptions a b c d) = put a >> put b >> put c >> put d+    get = do a <- get+             b <- get+             c <- get+             d <- get+             return $ AmSpawnOptions a b c d++defaultSpawnOptions :: AmSpawnOptions+defaultSpawnOptions = AmSpawnOptions {amsoPaused=False, amsoLink=Nothing, amsoMonitor=Nothing, amsoName=Nothing}++data AmSpawnUnpause = AmSpawnUnpause deriving (Typeable)+instance Binary AmSpawnUnpause where+   put AmSpawnUnpause = return ()+   get = return AmSpawnUnpause++-- TODO. Of course, this would need to be in a different module.+-- spawnWithChannel :: NodeId -> Closure (ReceivePort a -> ProcessM ()) -> ProcessM SendPort+-- spawnWithChannel nid clo = send ...++-- | Start a process running the code, given as a closure, on the specified node.+-- If successful, returns the process ID of the new process. If unsuccessful,+-- throw a 'TransmitException'. +spawn :: NodeId -> Closure (ProcessM ()) -> ProcessM ProcessId+spawn node clo = spawnAnd node clo defaultSpawnOptions++-- | Ends the current process in an orderly manner.+terminate :: ProcessM a+terminate = throw ProcessTerminationException++-- | If a remote process has been started in a paused state with 'spawnAnd' ,+-- it will be running but inactive until unpaused. Use this function to unpause+-- such a function. It has no effect on processes that are not paused or that+-- have already been unpaused.+unpause :: ProcessId -> ProcessM ()+unpause pid = send pid AmSpawnUnpause++-- | A variant of 'spawn' that starts the remote process with+-- bidirectoinal monitoring, as in 'linkProcess'+spawnLink :: NodeId -> Closure (ProcessM ()) -> ProcessM ProcessId+spawnLink node clo = do mypid <- getSelfPid+                        spawnAnd node clo defaultSpawnOptions {amsoLink=Just mypid}++-- | A variant of 'spawn' that allows greater control over how the remote process is started.+spawnAnd :: NodeId -> Closure (ProcessM ()) -> AmSpawnOptions -> ProcessM ProcessId+spawnAnd node clo opt = +    do res <- roundtripQueryUnsafe PldAdmin (adminGetPid node ServiceSpawner) (AmSpawn clo opt) +       case res of+         Left e -> throw $ TransmitException e+         Right pid -> return pid	++-- | Invokes a function on a remote node. The function must be+-- given by a closure. This function will block until the called+-- function completes or the connection is broken.+callRemote :: (Serializable a) => NodeId -> Closure (ProcessM a) -> ProcessM a+callRemote node clo = callRemoteImpl node clo++callRemoteIO :: (Serializable a) => NodeId -> Closure (IO a) -> ProcessM a+callRemoteIO node clo = callRemoteImpl node clo++callRemotePure :: (Serializable a) => NodeId -> Closure a -> ProcessM a+callRemotePure node clo = callRemoteImpl node clo++callRemoteImpl :: (Serializable a) => NodeId -> Closure b -> ProcessM a+callRemoteImpl node clo = +    let newclo = makePayloadClosure clo+     in case newclo of+          Nothing -> throw $ TransmitException QteUnknownCommand+          Just plclo ->+               do mypid <- getSelfPid+                  res <- roundtripQuery PldAdmin (adminGetPid node ServiceSpawner) (AmCall mypid plclo)+                  case res of+                     Right (Just mval) -> +                       do val <- liftIO $ serialDecode mval+                          case val of+                             Just a -> return a+                             _ -> throw $ TransmitException QteUnknownCommand+                     Left e -> throw (TransmitException e)+                     _ -> throw $ TransmitException QteUnknownCommand+++startSpawnerService :: ProcessM ()+startSpawnerService = serviceThread ServiceSpawner spawner+   where spawner = receiveWait [matchSpawnRequest,matchCallRequest,matchUnknownThrow] >> spawner+         exceptFilt :: SomeException -> q -> q+         exceptFilt _ q = q+         callWorker c responder = do a <- ptry $ invokeClosure c+                                     case a of+                                        Left q -> exceptFilt q (responder Nothing)+                                        Right Nothing -> responder Nothing+                                        Right (Just pl) -> responder (Just pl)+         spawnWorker c = do a <- invokeClosure c+                            case a of+                                Nothing -> (logS "SYS" LoCritical $ "Failed to invoke closure "++(show c)) --TODO it would be nice if this error could be propagated to the caller of spawn, at the very least it should throw an exception so a linked process will be notified +                                Just q -> q+         matchCallRequest = roundtripResponseAsync +               (\cmd sender -> case cmd of+                    AmCall _pid clo -> spawnLocal (callWorker clo sender) >> return ()) False+         matchSpawnRequest = roundtripResponse +               (\cmd -> case cmd of+                    AmSpawn c opt -> +                      let +                        namePostlude = case amsoName opt of+                                         Nothing -> return ()+                                         Just name -> nameSet name+                        pausePrelude = case amsoPaused opt of+                                            False -> return ()+                                            True -> receiveWait [match (\AmSpawnUnpause -> return ())]+                        linkPostlude = case amsoLink opt of+                                            Nothing -> return ()+                                            Just pid -> linkProcess pid+                        monitorPostlude = case amsoMonitor opt of+                                            Nothing -> return ()+                                            Just (pid,ma) -> do mypid <- getSelfPid+                                                                _ <- monitorProcessQuiet pid mypid ma+                                                                return ()+                      in do newpid <- spawnLocalAnd (pausePrelude >> spawnWorker c) (namePostlude >> linkPostlude >> monitorPostlude)+                            return (newpid,()))+++----------------------------------------------+-- * Local node registry+----------------------------------------------++{- UNUSED+localRegistryMagicProcess :: LocalProcessId+localRegistryMagicProcess = 38813+-}++localRegistryMagicMagic :: String+localRegistryMagicMagic = "__LocalRegistry"++localRegistryMagicRole :: String+localRegistryMagicRole = "__LocalRegistry"++type LocalProcessName = String++-- | Created by 'Remote.Peer.getPeers', this maps+-- each role to a list of nodes that have that role.+-- It can be examined directly or queried with+-- 'findPeerByRole'.+type PeerInfo = Map.Map String [NodeId]++data LocalNodeData = LocalNodeData {ldmRoles :: PeerInfo} ++{- UNUSED+type RegistryData = Map.Map String LocalNodeData+-}++data LocalProcessMessage =+        LocalNodeRegister String String NodeId+      | LocalNodeUnregister String String NodeId+      | LocalNodeQuery String+      | LocalNodeAnswer PeerInfo+      | LocalNodeResponseOK+      | LocalNodeResponseError String+      | LocalNodeHello+        deriving (Typeable)++instance Binary LocalProcessMessage where+   put (LocalNodeRegister m r n) = putWord8 0 >> put m >> put r >> put n+   put (LocalNodeUnregister m r n) = putWord8 1 >> put m >> put r >> put n+   put (LocalNodeQuery r) = putWord8 3 >> put r+   put (LocalNodeAnswer pi) = putWord8 4 >> put pi+   put (LocalNodeResponseOK) = putWord8 5+   put (LocalNodeResponseError s) = putWord8 6 >> put s+   put (LocalNodeHello) = putWord8 7+   get = do g <- getWord8+            case g of+              0 -> get >>= \m -> get >>= \r -> get >>= \n -> return (LocalNodeRegister m r n)+              1 -> get >>= \m -> get >>= \r -> get >>= \n -> return (LocalNodeUnregister m r n)+              3 -> get >>= return . LocalNodeQuery+              4 -> get >>= return . LocalNodeAnswer+              5 -> return LocalNodeResponseOK+              6 -> get >>= return . LocalNodeResponseError+              7 -> return LocalNodeHello++remoteRegistryPid :: NodeId -> ProcessM ProcessId+remoteRegistryPid nid =+     do let (NodeId hostname _) = nid+        cfg <- getConfig+        return $ adminGetPid (NodeId hostname (cfgLocalRegistryListenPort cfg)) ServiceNodeRegistry++localRegistryPid :: ProcessM ProcessId+localRegistryPid =+     do (NodeId hostname _) <- getSelfNode+        cfg <- getConfig+        return $ adminGetPid (NodeId hostname (cfgLocalRegistryListenPort cfg)) ServiceNodeRegistry++-- | Every host on which a node is running also needs a node registry,+-- which arbitrates those nodes can responds to peer queries. If+-- no registry is running, one will be automatically started+-- when the framework is started, but the registry can be started+-- independently, also. This function does that.+standaloneLocalRegistry :: String -> IO ()+standaloneLocalRegistry cfgn = do cfg <- readConfig True (Just cfgn)+                                  res <- startLocalRegistry cfg True+                                  liftIO $ putStrLn $ "Terminating standalone local registry: " ++ show res++-- | Contacts the local node registry and attempts to register current node. +-- You probably don't want to call this function yourself, as it's done for you in 'Remote.Init.remoteInit'+localRegistryRegisterNode :: ProcessM ()+localRegistryRegisterNode = localRegistryRegisterNodeImpl LocalNodeRegister++-- | Contacts the local node registry and attempts to unregister current node. +-- You probably don't want to call this function yourself, as it's done for you in 'Remote.Init.remoteInit'+localRegistryUnregisterNode :: ProcessM ()+localRegistryUnregisterNode = localRegistryRegisterNodeImpl LocalNodeUnregister++localRegistryRegisterNodeImpl :: (String -> String -> NodeId -> LocalProcessMessage) -> ProcessM ()+localRegistryRegisterNodeImpl cons = +    do cfg <- getConfig+       lrpid <- localRegistryPid+       nid <- getSelfNode+       let regMsg = cons (cfgNetworkMagic cfg) (cfgRole cfg) nid+       res <- roundtripQueryUnsafe PldAdmin lrpid regMsg+       case res of+          Left ts -> throw $ TransmitException ts+          Right LocalNodeResponseOK -> return ()+          Right (LocalNodeResponseError s) -> throw $ TransmitException $ QteOther s++-- | Contacts the local node registry and attempts to verify that it is alive.+-- If the local node registry cannot be contacted, an exception will be thrown.+localRegistryHello :: ProcessM ()+localRegistryHello = do lrpid <- localRegistryPid+                        res <- roundtripQueryUnsafe PldAdmin lrpid LocalNodeHello +                        case res of+                           (Right LocalNodeHello) -> return ()+                           (Left n) -> throw $ ConfigException $ "Can't talk to local node registry: " ++ show n+                           _ -> throw $ ConfigException $ "No response from local node registry"++localRegistryQueryNodes :: NodeId -> ProcessM (Maybe PeerInfo)+localRegistryQueryNodes nid = +    do cfg <- getConfig+       lrpid <- remoteRegistryPid nid+       let regMsg = LocalNodeQuery (cfgNetworkMagic cfg)+       res <- roundtripQueryUnsafe PldAdmin lrpid regMsg+       case res of+         Left _ts -> return Nothing+         Right (LocalNodeAnswer pi) -> return $ Just pi++-- TODO since local registries are potentially sticky, there is good reason+-- to ensure that they don't get corrupted; there should be some+-- kind of security precaution here, e.g. making sure that registrations+-- only come from local nodes+startLocalRegistry :: Config -> Bool -> IO TransmitStatus+startLocalRegistry cfg waitforever = startit+ where+  regConfig = cfg {cfgListenPort = cfgLocalRegistryListenPort cfg, cfgNetworkMagic=localRegistryMagicMagic, cfgRole = localRegistryMagicRole}+  handler tbl = receiveWait [roundtripResponse (registryCommand tbl)] >>= handler+  emptyNodeData = LocalNodeData {ldmRoles = Map.empty}+  lookup tbl magic role = case Map.lookup magic tbl of+                                 Nothing -> Nothing+                                 Just ldm -> Map.lookup role (ldmRoles ldm)+  remove tbl magic role nid =+             let+              roler Nothing = Nothing+              roler (Just lst) = Just $ filter ((/=)nid) lst+              removeFrom ldm = ldm {ldmRoles = Map.alter (roler) role (ldmRoles ldm)}+              remover Nothing = Nothing+              remover (Just ldm) = Just (removeFrom ldm)+             in Map.alter remover magic tbl+  insert tbl magic role nid = +             let+              roler Nothing = Just [nid]+              roler (Just lst) = if elem nid lst+                                    then (Just lst)+                                    else (Just (nid:lst))+              insertTo ldm = ldm {ldmRoles = Map.alter (roler) role (ldmRoles ldm)}+                                +              inserter Nothing = Just (insertTo emptyNodeData)+              inserter (Just ldm) = Just (insertTo ldm)+             in Map.alter inserter magic tbl+  registryCommand tbl (LocalNodeQuery magic) = +             case Map.lookup magic tbl of+                Nothing -> return (LocalNodeAnswer Map.empty,tbl)+                Just pi -> return (LocalNodeAnswer (ldmRoles pi),tbl)+  registryCommand tbl (LocalNodeUnregister magic role nid) =     -- check that given entry exists?+             return (LocalNodeResponseOK,remove tbl magic role nid)+  registryCommand tbl (LocalNodeRegister magic role nid) =+             case lookup tbl magic role of+                Nothing -> return (LocalNodeResponseOK,insert tbl magic role nid)+                Just nids -> if elem nid nids +                                then return (LocalNodeResponseError ("Multiple node registration for " ++ show nid ++ " as " ++ role),tbl)+                                else return (LocalNodeResponseOK,insert tbl magic role nid)+  registryCommand tbl (LocalNodeHello) = return (LocalNodeHello,tbl)+  registryCommand tbl _ = return (LocalNodeResponseError "Unknown command",tbl)+  startit = do node <- initNode regConfig Remote.Reg.empty+               res <- try $ forkAndListenAndDeliver node regConfig :: IO (Either SomeException ())+               case res of+                  Left e -> return $ QteNetworkError $ show e+                  Right () -> runLocalProcess node (startLoggingService >> +                                                    adminRegister ServiceNodeRegistry >> +                                                    handler Map.empty) >>+                              if waitforever +                                 then waitForThreads node >>+                                      (return $ QteOther "Local registry process terminated")+                                 else return QteOK+++----------------------------------------------+-- * Closures+----------------------------------------------++-- TODO makeClosure should verify that the given fun argument actually exists+makeClosure :: (Typeable a,Serializable v) => String -> v -> ProcessM (Closure a)+makeClosure fun env = do +                         enc <- liftIO $ serialEncode env +                         return $ Closure fun enc++makePayloadClosure :: Closure a -> Maybe (Closure Payload)+makePayloadClosure (Closure name arg) = +                case isSuffixOf "__impl" name of+                  False -> Nothing+                  True -> Just $ Closure (name++"Pl") arg++evaluateClosure :: (Typeable b) => Closure a -> ProcessM (Maybe (Payload -> b))+evaluateClosure (Closure name _) =+        do node <- getLookup+           return $ getEntryByIdent node name++invokeClosure :: (Typeable a) => Closure a -> ProcessM (Maybe a)+invokeClosure (Closure name arg) = +           (\_id ->+                do node <- getLookup+                   res <- sequence [pureFun node,ioFun node,procFun node]+                   case catMaybes res of+                      (a:_) -> return $ Just a+                      _ -> return Nothing ) Prelude.id+   where pureFun node = case getEntryByIdent node name of+                          Nothing -> return Nothing+                          Just x -> return $ Just $ (x arg)+         ioFun node =   case getEntryByIdent node name of+                          Nothing -> return Nothing+                          Just x -> liftIO (x arg) >>= (return.Just)+         procFun node = case getEntryByIdent node name of+                          Nothing -> return Nothing+                          Just x -> (x arg) >>= (return.Just)+++----------------------------------------------+-- * Simple functional queue+----------------------------------------------++data Queue a = Queue [a] [a]++queueMake :: Queue a+queueMake = Queue [] []++{- UNUSED+queueEmpty :: Queue a -> Bool+queueEmpty (Queue [] []) = True+queueEmpty _ = False+-}++queueInsert :: Queue a -> a -> Queue a+queueInsert (Queue incoming outgoing) a = Queue (a:incoming) outgoing++{- UNUSED+queueInsertAndLimit :: Queue a -> Int -> a -> Queue a+queueInsertAndLimit q limit a= +       let s1 = queueInsert q a+           s2 = if queueLength s1 > limit+                   then let (_,f) = queueRemove s1+                         in f+                   else s1+        in s2+-}++queueInsertMulti :: Queue a -> [a] -> Queue a+queueInsertMulti (Queue incoming outgoing) a = Queue (a++incoming) outgoing++{- UNUSED+queueRemove :: Queue a -> (Maybe a,Queue a)+queueRemove (Queue incoming (a:outgoing)) = (Just a,Queue incoming outgoing)+queueRemove (Queue l@(_:_) []) = queueRemove $ Queue [] (reverse l)+queueRemove q@(Queue [] []) = (Nothing,q)+-}++queueToList :: Queue a -> [a]+queueToList (Queue incoming outgoing) = outgoing ++ reverse incoming+queueFromList :: [a] -> Queue a+queueFromList l = Queue [] l+++{- UNUSED+queueLength :: Queue a -> Int+queueLength (Queue incoming outgoing) = length incoming + length outgoing -- should probably just store in the length in the structure++queueEach :: Queue a -> [(a,Queue a)]+queueEach q = case queueToList q of+                     [] -> []+                     a:rest -> each [] a rest+        where each before it []                     = [(it,queueFromList before)]+              each before it following@(next:after) = (it,queueFromList (before++following)):(each (before++[it]) next after)+-}++withSem :: QSem -> IO a -> IO a+withSem sem f = action `finally` signalQSem sem+   where action = waitQSem sem >> f++
+ Remote/Reg.hs view
@@ -0,0 +1,67 @@+-- | Runtime metadata functions, part of the +-- RPC mechanism+module Remote.Reg (+         registerCalls,+         Lookup,+         Identifier,+         putReg,+         getEntryByIdent,+         empty,+         RemoteCallMetaData+           ) where++import Data.Dynamic (Dynamic,toDyn,fromDynamic)+import Data.Typeable (Typeable)+import qualified Data.Map as Map (insert,lookup,Map,empty)++----------------------------------------------+-- * Runtime metadata+------------------------------++-- | Data of this type is generated at compile-time+-- by 'remotable' and can be used with 'registerCalls'+-- and 'remoteInit' to create a metadata lookup table, 'Lookup'.+-- The name '__remoteCallMetaData' will be present+-- in any module that uses 'remotable'.+type RemoteCallMetaData = Lookup -> Lookup+++type Identifier = String++data Entry = Entry {+               entryName :: Identifier,+               entryFunRef :: Dynamic+             }++-- | Creates a metadata lookup table based on compile-time metadata.+-- You probably don't want to call this function yourself, but instead+-- use 'Remote.Init.remoteInit'.+registerCalls :: [RemoteCallMetaData] -> Lookup+registerCalls [] = empty+registerCalls (h:rest) = let registered = registerCalls rest+                          in h registered++makeEntry :: (Typeable a) => Identifier -> a -> Entry+makeEntry ident funref = Entry {entryName=ident, entryFunRef=toDyn funref}++type IdentMap = Map.Map Identifier Entry+data Lookup = Lookup { identMap :: IdentMap }++putReg :: (Typeable a) => a -> Identifier -> Lookup -> Lookup+putReg a i l = putEntry l a i++putEntry :: (Typeable a) => Lookup -> a -> Identifier -> Lookup+putEntry amap value name = +                             Lookup {+                                identMap = Map.insert name entry (identMap amap)+                             }+  where+       entry = makeEntry name value+++getEntryByIdent :: (Typeable a) => Lookup -> Identifier -> Maybe a+getEntryByIdent amap ident = (Map.lookup ident (identMap amap)) >>= (\x -> fromDynamic (entryFunRef x))++empty :: Lookup+empty = Lookup {identMap = Map.empty}+
+ Remote/Task.hs view
@@ -0,0 +1,966 @@+{-# LANGUAGE DeriveDataTypeable #-}++-- | This module provides data dependency resolution and+-- fault tolerance via /promises/ (known elsewhere as /futures/).+-- It's implemented in terms of the "Remote.Process" module.+module Remote.Task (+                   -- * Tasks and promises+                   TaskM, Promise, PromiseList(..),+                   runTask,+                   newPromise, newPromiseAt, newPromiseNear, newPromiseHere, newPromiseAtRole,+                   toPromise, toPromiseAt, toPromiseNear, toPromiseImm,+                   readPromise, ++                   -- * MapReduce+                   MapReduce(..),+                   mapReduce,++                   -- * Useful auxilliaries+                   chunkify,+                   shuffle,+                   tsay,+                   tlogS,+                   Locality(..),+                   TaskException(..),++                   -- * Internals, not for general use+                   __remoteCallMetaData, serialEncodeA, serialDecodeA+                   ) where++import Remote.Reg (putReg,getEntryByIdent,RemoteCallMetaData)+import Remote.Encoding (serialEncodePure,hGetPayload,hPutPayload,Payload,getPayloadContent,Serializable,serialDecode,serialEncode)+import Remote.Process (roundtripQuery, ServiceException(..), TransmitStatus(..),diffTime,getConfig,Config(..),matchProcessDown,terminate,nullPid,monitorProcess,TransmitException(..),MonitorAction(..),ptry,LogConfig(..),getLogConfig,setNodeLogConfig,nodeFromPid,LogLevel(..),LogTarget(..),logS,getLookup,say,LogSphere,NodeId,ProcessM,ProcessId,PayloadDisposition(..),getSelfPid,getSelfNode,matchUnknownThrow,receiveWait,receiveTimeout,roundtripResponse,roundtripResponseAsync,roundtripQueryImpl,match,makePayloadClosure,spawn,spawnLocal,spawnLocalAnd,setDaemonic,send,makeClosure)+import Remote.Closure (Closure(..))+import Remote.Peer (getPeers)++import Data.Dynamic (Dynamic, toDyn, fromDynamic)+import System.IO (withFile,IOMode(..))+import System.Directory (renameFile)+import Data.Binary (Binary,get,put,putWord8,getWord8)+import Control.Exception (SomeException,Exception,throw)+import Data.Typeable (Typeable)+import Control.Monad (liftM,when)+import Control.Monad.Trans (liftIO)+import Control.Concurrent.MVar (MVar,modifyMVar,modifyMVar_,newMVar,newEmptyMVar,takeMVar,putMVar,readMVar,withMVar)+import qualified Data.Map as Map (Map,insert,lookup,empty,insertWith',toList)+import Data.List ((\\),union,nub,groupBy,sortBy,delete)+import Data.Time (UTCTime,getCurrentTime)++-- imports required for hashClosure; is there a lighter-weight of doing this?+import Data.Digest.Pure.MD5 (md5)+import Data.ByteString.Lazy.UTF8 (fromString)+import qualified Data.ByteString.Lazy as B (concat)++----------------------------------------------+-- * Promises and tasks+----------------------------------------------+++type PromiseId = Integer++type Hash = String++data PromiseList a = PlChunk a (Promise (PromiseList a))+                   | PlNil deriving Typeable++instance (Serializable a) => Binary (PromiseList a) where+   put (PlChunk a p) = putWord8 0 >> put a >> put p+   put PlNil = putWord8 1+   get = do w <- getWord8+            case w of+               0 -> do a <- get+                       p <- get+                       return $ PlChunk a p+               1 -> return PlNil++-- | The basic data type for expressing data dependence+-- in the 'TaskM' monad. A Promise represents a value that+-- may or may not have been computed yet; thus, it's like+-- a distributed thunk (in the sense of a non-strict unit+-- of evaluation). These are created by 'newPromise' and friends,+-- and the underlying value can be gotten with 'readPromise'.+data Promise a = PromiseBasic { _psRedeemer :: ProcessId, _psId :: PromiseId } +               | PromiseImmediate a deriving Typeable+-- psRedeemer should maybe be wrapped in an IORef so that it can be updated in case of node failure++instance (Serializable a) => Binary (Promise a) where+   put (PromiseBasic a b) = putWord8 0 >> put a >> put b+   put (PromiseImmediate a) = putWord8 1 >> put a+   get = do a <- getWord8+            case a of+              0 -> do b <- get+                      c <- get+                      return $ PromiseBasic b c+              1 -> do b <- get+                      return $ PromiseImmediate b++-- | Stores the data produced by a promise, in one of its+-- various forms. If it's currently in memory, we keep it+-- as a payload, to be decoded by its ultimate user (who+-- of course has the static type information), the time+-- it was last touched (so we know when to flush it),+-- and perhaps also a decoded version, so that it doesn't+-- need to be decoded repeatedly: this makes this go a lot+-- faster. If it's been flushed to disk, we keep track of+-- where, and if the promise didn't complete, but threw+-- an exception during its execution, we mark there here+-- as well: the exception will be propagated to+-- dependents.+data PromiseStorage = PromiseInMemory PromiseData UTCTime (Maybe Dynamic)+                    | PromiseOnDisk FilePath+                    | PromiseException String++type PromiseData = Payload+{- UNUSED+type TimeStamp = UTCTime+-}++-- | Keeps track of what we know about currently running promises.+-- The closure and locality and provided by the initial call to+-- newPromise, the nodeboss is where it is currently running.+-- We need this info to deal with complaints.+data PromiseRecord = PromiseRecord ProcessId (Closure PromiseData) Locality++data MasterState = MasterState+     {+         -- | Promise IDs are allocated serially from here+         msNextId :: PromiseId,+ +         -- | All currently known nodes, with the role, node ID, and node boss. Updated asychronously by prober thread+         msNodes :: MVar [(String,NodeId,ProcessId)],++         -- | Given a nodeboss, which promises belong to it. Not sure what this is good for+         msAllocation :: Map.Map ProcessId [PromiseId],++         -- | Given a promise, what do we know about it. Include its nodeboss, its closure, and its locality preference+         msPromises :: Map.Map PromiseId PromiseRecord,++         -- | The locality preference of new worker tasks, if not specified otherwise+         msDefaultLocality :: Locality+     }++data MmNewPromise = MmNewPromise (Closure Payload) Locality Queueing deriving (Typeable)+instance Binary MmNewPromise where +  get = do a <- get+           l <- get+           q <- get+           return $ MmNewPromise a l q+  put (MmNewPromise a l q) = put a >> put l >> put q++data MmNewPromiseResponse = MmNewPromiseResponse ProcessId PromiseId +                          | MmNewPromiseResponseFail deriving (Typeable)+instance Binary MmNewPromiseResponse where +   put (MmNewPromiseResponse a b) =+           do putWord8 0+              put a+              put b+   put MmNewPromiseResponseFail = putWord8 1+   get = do a <- getWord8+            case a of+              0 -> do b <- get +                      c <- get+                      return $ MmNewPromiseResponse b c+              1 -> return MmNewPromiseResponseFail++data MmStatus = MmStatus deriving Typeable+instance Binary MmStatus where+  get = return MmStatus+  put MmStatus = put ()++data MmStatusResponse = MmStatusResponse [NodeId] (Map.Map ProcessId [PromiseId]) deriving Typeable+instance Binary MmStatusResponse where+  get = do a <- get+           b <- get+           return $ MmStatusResponse a b+  put (MmStatusResponse a b) = put a >> put b++data MmComplain = MmComplain ProcessId PromiseId deriving (Typeable)+instance Binary MmComplain where +   put (MmComplain a b) = put a >> put b+   get = do a <- get+            b <- get+            return $ MmComplain a b++data MmComplainResponse = MmComplainResponse ProcessId deriving (Typeable)+instance Binary MmComplainResponse where+   put (MmComplainResponse a) = put a+   get = do a <- get+            return $ MmComplainResponse a++data TmNewPeer = TmNewPeer NodeId deriving (Typeable)+instance Binary TmNewPeer where +   get = do a <- get+            return $ TmNewPeer a+   put (TmNewPeer nid) = put nid++data NmStart = NmStart PromiseId (Closure Payload) Queueing deriving (Typeable)+instance Binary NmStart where +  get = do a <- get+           b <- get+           c <- get+           return $ NmStart a b c+  put (NmStart a b c) = put a >> put b >> put c++data NmStartResponse = NmStartResponse Bool deriving (Typeable)+instance Binary NmStartResponse where +  get = do a <- get+           return $ NmStartResponse a+  put (NmStartResponse a) = put a++data NmRedeem = NmRedeem PromiseId deriving (Typeable)+instance Binary NmRedeem where +   get = do a <- get+            return $ NmRedeem a+   put (NmRedeem prid) = put prid++data NmRedeemResponse = NmRedeemResponse Payload+                      | NmRedeemResponseUnknown+                      | NmRedeemResponseException deriving (Typeable)+instance Binary NmRedeemResponse where +   get = do a <- getWord8+            case a of+              0 -> do b <- get+                      return $ NmRedeemResponse b+              1 -> return NmRedeemResponseUnknown+              2 -> return NmRedeemResponseException+   put (NmRedeemResponse a) = putWord8 0 >> put a+   put (NmRedeemResponseUnknown) = putWord8 1+   put (NmRedeemResponseException) = putWord8 2++data TaskException = TaskException String deriving (Show,Typeable)+instance Exception TaskException++-- | (Currently ignored.)+data Queueing = QuNone+              | QuExclusive+              | QuSmall deriving (Typeable,Ord,Eq)++defaultQueueing :: Queueing+defaultQueueing = QuNone++instance Binary Queueing where+   put QuNone = putWord8 0+   put QuExclusive = putWord8 1+   put QuSmall = putWord8 2+   get = do a <- getWord8+            case a of+               0 -> return QuNone+               1 -> return QuExclusive+               2 -> return QuSmall++-- | A specification of preference+-- of where a promise should be allocated,+-- among the nodes visible to the master.+data Locality = LcUnrestricted -- ^ The promise can be placed anywhere.+              | LcDefault -- ^ The default preference is applied, which is for nodes having a role of NODE of WORKER+              | LcByRole [String] -- ^ Nodes having the given roles will be preferred+              | LcByNode [NodeId] -- ^ The given nodes will be preferred++instance Binary Locality where+   put LcUnrestricted = putWord8 0+   put LcDefault = putWord8 1+   put (LcByRole a) = putWord8 2 >> put a+   put (LcByNode a) = putWord8 3 >> put a+   get = do a <- getWord8+            case a of+               0 -> return LcUnrestricted+               1 -> return LcDefault+               2 -> do r <- get+                       return $ LcByRole r+               3 -> do r <- get+                       return $ LcByNode r++defaultLocality :: Locality+defaultLocality = LcByRole ["WORKER","NODE"]++taskError :: String -> a+taskError s = throw $ TaskException s++serialEncodeA :: (Serializable a) => a -> TaskM Payload+serialEncodeA = liftTask . liftIO . serialEncode++serialDecodeA :: (Serializable a) => Payload -> TaskM (Maybe a)+serialDecodeA = liftTask . liftIO . serialDecode++monitorTask :: ProcessId -> ProcessId -> ProcessM TransmitStatus+monitorTask monitor monitee+   = do res <- ptry $ monitorProcess monitor monitee MaMonitor +        case res of+           Right _ -> return QteOK+           Left (ServiceException e) -> return $ QteOther e++roundtripImpl :: (Serializable a, Serializable b) => ProcessId -> a -> ProcessM (Either TransmitStatus b)+roundtripImpl pid dat = roundtripQueryImpl 0 PldUser pid dat id []++roundtrip :: (Serializable a, Serializable b) => ProcessId -> a -> TaskM (Either TransmitStatus b)+roundtrip apid dat = +   TaskM $ \ts ->+     case Map.lookup apid (tsMonitoring ts) of+       Nothing -> do mypid <- getSelfPid+                     res0 <- monitorTask mypid apid+                     case res0 of+                        QteOK -> +                          do res <- roundtripImpl apid dat+                             return (ts {tsMonitoring=Map.insert apid () (tsMonitoring ts)},res)+                        _ -> return (ts,Left res0)+       Just _ -> do res <- roundtripImpl apid dat+                    return (ts,res)++-- roundtrip a b = liftTask $ roundtripQueryUnsafe PldUser a b++spawnDaemonic :: ProcessM () -> ProcessM ProcessId+spawnDaemonic p = spawnLocalAnd p setDaemonic++runWorkerNode :: ProcessId -> NodeId -> ProcessM ProcessId+runWorkerNode masterpid nid = +     do clo <- makeClosure "Remote.Task.runWorkerNode__impl" (masterpid) :: ProcessM (Closure (ProcessM ()))+        spawn nid clo ++runWorkerNode__impl :: Payload -> ProcessM ()+runWorkerNode__impl pl = +   do setDaemonic -- maybe it's good to have the node manager be daemonic, but prolly not. If so, the MASTERPID must be terminated when user-provided MASTERPROC ends+      mpid <- liftIO $ serialDecode pl+      case mpid of+         Just masterpid -> handler masterpid+         Nothing -> error "Failure to extract in rwn__impl"+ where handler masterpid = startNodeManager masterpid++passthrough__implPl :: Payload -> TaskM Payload+passthrough__implPl pl = return pl++passthrough__closure :: (Serializable a) => a -> Closure (TaskM a)+passthrough__closure a = Closure "Remote.Task.passthrough__impl" (serialEncodePure a)++__remoteCallMetaData :: RemoteCallMetaData+__remoteCallMetaData x = putReg runWorkerNode__impl "Remote.Task.runWorkerNode__impl" +                        (putReg passthrough__implPl "Remote.Task.passthrough__implPl" x)++updatePromiseInMemory :: PromiseStorage -> IO PromiseStorage+updatePromiseInMemory (PromiseInMemory p _ d) = do utc <- getCurrentTime+                                                   return $ PromiseInMemory p utc d+updatePromiseInMemory other = return other++makePromiseInMemory :: PromiseData -> Maybe Dynamic -> IO PromiseStorage+makePromiseInMemory p dyn = do utc <- getCurrentTime+                               return $ PromiseInMemory p utc dyn++forwardLogs :: Maybe ProcessId -> ProcessM ()+forwardLogs masterpid = +        do lc <- getLogConfig+           selfnid <- getSelfNode+           let newlc = lc {logTarget = case masterpid of+                                             Just mp+                                                | nodeFromPid mp /= selfnid -> LtForward $ nodeFromPid mp+                                             _ -> LtStdout}     +             in setNodeLogConfig newlc++hashClosure :: Closure a -> Hash+hashClosure (Closure s pl) = show $ md5 $ B.concat [fromString s, getPayloadContent pl]++undiskify :: FilePath -> MVar PromiseStorage -> ProcessM (Maybe PromiseData)+undiskify fpIn mps =+      do wrap $ liftIO $ modifyMVar mps (\val ->+            case val of+              PromiseOnDisk fp -> +                do pl <- withFile fp ReadMode hGetPayload+                   inmem <- makePromiseInMemory pl Nothing+                   return (inmem,Just pl)+              PromiseInMemory payload _ _ -> return (val,Just payload)+              _ -> return (val,Nothing))+  where wrap a = do res <- ptry a+                    case res of+                      Left e -> do logS "TSK" LoCritical $ "Error reading promise from file "++fpIn++": "++show (e::IOError)+                                   return Nothing+                      Right r -> return r++diskify :: FilePath -> MVar PromiseStorage -> Bool -> ProcessM ()+diskify fp mps reallywrite =+      do cfg <- getConfig+         when (cfgPromiseFlushDelay cfg > 0)+            (handler (cfgPromiseFlushDelay cfg))+    where+         handler delay =+           do _ <- receiveTimeout delay []+              again <- wrap $ liftIO $ modifyMVar mps (\val ->+                   case val of+                       PromiseInMemory payload utc _ ->+                           do now <- getCurrentTime+                              if diffTime now utc > delay+                                 then do when reallywrite $+                                             do liftIO $ withFile tmp WriteMode (\h -> hPutPayload h payload)+                                                renameFile tmp fp+                                         return (PromiseOnDisk fp,False)+                                 else return (val,True)+                       _      -> return (val,False))+              when again+                 (diskify fp mps reallywrite)+         tmp = fp ++ ".tmp"+         wrap a = do res <- ptry a+                     case res of+                         Left z -> do logS "TSK" LoImportant $ "Error writing promise to disk on file "++fp++": "++show (z::IOError)+                                      return False+                         Right v -> return v++startNodeWorker :: ProcessId -> NodeBossState -> +                   MVar PromiseStorage -> Closure Payload -> ProcessM ()+startNodeWorker masterpid nbs mps clo@(Closure cloname cloarg) = +  do self <- getSelfPid+     _ <- spawnLocalAnd (starter self) (prefix self)+     return ()+  where +        prefix nodeboss =+          do self <- getSelfPid+             monitorProcess self nodeboss MaLink+             setDaemonic +        starter nodeboss = -- TODO try to do an undiskify here, if the promise is left over from a previous, failed run+            let initialState = TaskState {tsMaster=masterpid,tsNodeBoss=Just nodeboss, +                                          tsPromiseCache=nsPromiseCache nbs, tsRedeemerForwarding=nsRedeemerForwarding nbs,+                                          tsMonitoring=Map.empty}+                tasker = do tbl <- liftTask $ getLookup+                            case getEntryByIdent tbl cloname of+                              Just funval -> +                                          do val <- funval cloarg+                                             p <- liftTaskIO $ makePromiseInMemory val Nothing+                                             liftTaskIO $ putMVar mps p+                                             cfg <- liftTask $ getConfig+                                             let cachefile = cfgPromisePrefix cfg++hashClosure clo+                                             liftTask $ diskify cachefile mps True+                              Nothing -> taskError $ "Failed looking up "++cloname++" in closure table"+             in do res <- ptry $ runTaskM tasker initialState :: ProcessM (Either SomeException (TaskState,()))+                   case res of+                     Left ex -> liftIO (putMVar mps (PromiseException (show ex))) >> throw ex+                     Right _ -> return ()++data NodeBossState = +        NodeBossState+        {+           nsPromiseCache :: MVar (Map.Map PromiseId (MVar PromiseStorage)),+           nsRedeemerForwarding :: MVar (Map.Map PromiseId ProcessId)+        }++startNodeManager :: ProcessId -> ProcessM ()+startNodeManager masterpid = +  let+      handler :: NodeBossState -> ProcessM a+      handler state = +        let promisecache = nsPromiseCache state+            nmStart = roundtripResponse (\(NmStart promise clo _queueing) -> +               do promisestore <- liftIO $ newEmptyMVar+                  ret <- liftIO $ modifyMVar promisecache +                    (\pc ->               let newpc = Map.insert promise promisestore pc +                                           in return (newpc,True))+                  when (ret)+                       (startNodeWorker masterpid state promisestore clo)+                  return (NmStartResponse ret,state)) +            nmTermination = matchProcessDown masterpid $+                     do forwardLogs Nothing+                        logS "TSK" LoInformation $ "Terminating nodeboss after my master "++show masterpid++" is gone"+                        terminate+            nmRedeem = roundtripResponseAsync (\(NmRedeem promise) ans -> +                     let answerer = do pc <- liftIO $ readMVar promisecache+                                       case Map.lookup promise pc of+                                         Nothing -> ans NmRedeemResponseUnknown+                                         Just v -> do rv <- liftIO $ readMVar v -- possibly long wait+                                                      case rv of+                                                         PromiseInMemory rrv _ _ -> +                                                                do liftIO $ modifyMVar_ v (\_ -> updatePromiseInMemory rv)+                                                                   ans (NmRedeemResponse rrv)+                                                         PromiseOnDisk fp -> do mpd <- undiskify fp v+                                                                                case mpd of+                                                                                   Nothing -> +                                                                                     ans (NmRedeemResponseUnknown)+                                                                                   Just a ->+                                                                                     ans (NmRedeemResponse a)+                                                                                diskify fp v False+                                                         PromiseException _ -> ans NmRedeemResponseException+                      in do _ <- spawnLocal answerer+                            return state) False+         in receiveWait [nmStart, nmRedeem, nmTermination, matchUnknownThrow] >>= handler+   in do forwardLogs $ Just masterpid+         mypid <- getSelfPid+         monitorProcess mypid masterpid MaMonitor+         logS "TSK" LoInformation $ "Starting a nodeboss owned by " ++ show masterpid+         pc <- liftIO $ newMVar Map.empty+         pf <- liftIO $ newMVar Map.empty+         let initState = NodeBossState {nsPromiseCache=pc,nsRedeemerForwarding=pf}+         handler initState++-- | Starts a new context for executing a 'TaskM' environment.+-- The node on which this function is run becomes a new master+-- in a Task application; as a result, the application should+-- only call this function once. The master will attempt to+-- control all nodes that it can find; if you are going to be+-- running more than one CH application on a single network,+-- be sure to give each application a different network+-- magic (via cfgNetworkMagic). The master TaskM environment+-- created by this function can then spawn other threads,+-- locally or remotely, using 'newPromise' and friends.++runTask :: TaskM a -> ProcessM a+runTask = startMaster++startMaster :: TaskM a -> ProcessM a+startMaster proc = +    do mvmaster <- liftIO $ newEmptyMVar+       mvdone <- liftIO $ newEmptyMVar+       master <- runMaster (masterproc mvdone mvmaster)+       liftIO $ putMVar mvmaster master+       liftIO $ takeMVar mvdone+   where masterproc mvdone mvmaster nodeboss = +           do master <- liftIO $ takeMVar mvmaster+              pc <- liftIO $ newMVar Map.empty+              pf <- liftIO $ newMVar Map.empty+              let initialState = TaskState {tsMaster=master,tsNodeBoss=Just nodeboss,+                                            tsPromiseCache=pc, tsRedeemerForwarding=pf,+                                            tsMonitoring=Map.empty}+              res <- liftM snd $ runTaskM proc initialState+              liftIO $ putMVar mvdone res++{- UNUSED+type LocationSelector = MasterState -> ProcessM (NodeId,ProcessId)+-}++selectLocation :: MasterState -> Locality -> ProcessM (Maybe (String,NodeId,ProcessId))+selectLocation ms locality =+   let nodes = msNodes ms+    in liftIO $ modifyMVar nodes+          (\n -> case n of+                    [] -> return (n,Nothing)+                    _ -> let dflt = (rotate n,Just $ head n)+                             filterify f = case filter f n of+                                             [] -> return dflt+                                             (a:_) -> return ((delete a n) ++ [a],Just a)+                          in  case cond locality of+                                LcUnrestricted -> return dflt+                                LcDefault -> return dflt+                                LcByRole l -> filterify (\(r,_,_) -> r `elem` l)+                                LcByNode l -> filterify (\(_,r,_) -> r `elem` l))+      where rotate [] = []+            rotate (h:t) = t ++ [h]+            cond l = case l of+                       LcDefault -> msDefaultLocality ms+                       _ -> l++countLocations :: MasterState -> ProcessM Int+countLocations ms = liftIO $ withMVar (msNodes ms) (\a -> return $ length a)++findPeers :: ProcessM [(String,NodeId)]+findPeers = liftM (concat . (map (\(role,v) -> [ (role,x) | x <- v] )) . Map.toList) getPeers++sendSilent :: (Serializable a) => ProcessId -> a -> ProcessM ()+sendSilent pid a = do res <- ptry $ send pid a+                      case res of+                        Left (TransmitException _) -> return ()+                        Right _ -> return ()++{- UNUSED+getStatus :: TaskM ()+getStatus = +   do master <- getMaster+      res <- roundtrip master MmStatus+      case res of+        Left _ -> return ()+        Right (MmStatusResponse nodes promises) ->+              let verboseNodes = intercalate ", " (map show nodes)+                  verbosePromises = intercalate "\n" $ map (\(nb,l) -> (show nb)++" -- "++intercalate "," (map show l)) (Map.toList promises) +               in tsay $ "\nKnown nodes: " ++ verboseNodes ++ "\n\nNodebosses: " ++ verbosePromises+-}++runMaster :: (ProcessId -> ProcessM ()) -> ProcessM ProcessId+runMaster masterproc = +  let+     probeOnce nodes seen masterpid =+        do recentlist <- findPeers -- TODO if a node fails to response to a probe even once, it's gone forever; be more flexible+           let newseen = seen `union` recentlist+           let topidlist = recentlist \\ seen+           let cleanOut n = filter (\(_,nid,_) -> nid `elem` (map snd recentlist)) n+           newlypidded <- mapM (\(role,nid) -> +                             do pid <- runWorkerNode masterpid nid+                                return (role,nid,pid)) topidlist           +           (_newlist,totalseen) <- liftIO $ modifyMVar nodes (\oldlist ->+                        return ((cleanOut oldlist) ++ newlypidded,(recentlist,newseen)))+           let newlyadded = totalseen \\ seen+           mapM_ (\nid -> sendSilent masterpid (TmNewPeer nid)) (map snd newlyadded)+           return totalseen+     proberDelay = 10000000 -- how often do we check the network to see what nodes are available?+     prober nodes seen masterpid =+        do totalseen <- probeOnce nodes seen masterpid+           _ <- receiveTimeout proberDelay [matchUnknownThrow]+           prober nodes totalseen masterpid+     master state = +        let +            tryAlloc clo promiseid locality queueing =+              do ns <- selectLocation state locality+                 case ns of+                    Nothing -> do logS "TSK" LoCritical "Attempt to allocate a task, but no nodes found"+                                  return Nothing+                    Just (_,nid,nodeboss) -> +                         do res <- roundtripQuery PldUser nodeboss (NmStart promiseid clo queueing) -- roundtripQuery monitors and then unmonitors, which generates a lot of traffic; we probably don't need to do this+                            case res of+                              Left e -> +                                 do logS "TSK" LoImportant $ "Failed attempt to start "++show clo++" on " ++show nid ++": "++show e+                                    return Nothing+                              Right (NmStartResponse True) -> return $ Just nodeboss+                              _ -> do logS "TSK" LoImportant $ "Failed attempt to start "++show clo++" on " ++show nid+                                      return Nothing+            basicAllocate clo promiseid locality queueing =+               do count <- countLocations state+                  res1 <- tryAlloc clo promiseid locality queueing+                  case res1 of +                     Just _ -> return res1+                     Nothing ->  -- TODO we should try all matching locations before moving on to Unrestricted+                       do res <- stubborn count $ tryAlloc clo promiseid LcUnrestricted queueing+                          case res of+                             Nothing -> do logS "TSK" LoCritical $ "Terminally failed to start "++show clo+                                           return res+                             _ -> return res+            statusMsg = roundtripResponse+              (\x -> case x of+                        MmStatus -> +                         do thenodes <- liftIO $ readMVar $ msNodes state+                            let knownNodes = map (\(_,n,_) -> n) thenodes+                                proctree = msAllocation state+                            return (MmStatusResponse knownNodes proctree,state))+            complainMsg = roundtripResponse+              (\x -> case x of+                       MmComplain procid promid -> +                         case Map.lookup promid (msPromises state) of+                            Nothing -> return (MmComplainResponse nullPid,state) -- failure+                            Just (PromiseRecord curprocid curclo curlocality) +                               | curprocid /= procid -> return (MmComplainResponse curprocid,state)+                               | otherwise -> +                                   do res <- basicAllocate curclo promid curlocality defaultQueueing+                                      case res of+                                        Nothing -> return (MmComplainResponse nullPid,state) -- failure+                                        Just newprocid ->+                                           let newpromises = Map.insert promid (PromiseRecord newprocid curclo curlocality) (msPromises state)+                                            in return (MmComplainResponse newprocid,state {msPromises=newpromises})) +            promiseMsg = roundtripResponse +              (\x -> case x of+                       MmNewPromise clo locality queueing -> +                          do +                             let promiseid = msNextId state+                             res <- basicAllocate clo promiseid locality queueing+                             case res of+                                Just nodeboss -> +                                        let newstate = state {msAllocation=newAllocation,msPromises=newPromises,msNextId=promiseid+1}+                                            newAllocation = Map.insertWith' (\a b -> nub $ a++b) nodeboss [promiseid] (msAllocation state)+                                            newPromises = Map.insert promiseid (PromiseRecord nodeboss clo locality) (msPromises state)+                                         in return (MmNewPromiseResponse nodeboss promiseid,newstate)+                                Nothing -> +                                         return (MmNewPromiseResponseFail,state))+            simpleMsg = match +              (\x -> case x of+                       TmNewPeer nid -> do logS "TSK" LoInformation $ "Found new peer " ++show nid+                                           return state)+         in receiveWait [simpleMsg, promiseMsg, complainMsg,statusMsg] >>= master -- TODO matchUnknownThrow+   in do nodes <- liftIO $ newMVar []+         selfnode <- getSelfNode+         selfpid <- getSelfPid+         let initState = MasterState {msNextId=0, msAllocation=Map.empty, msPromises=Map.empty, msNodes=nodes, msDefaultLocality = defaultLocality}+         masterpid <- spawnDaemonic (master initState)+         seennodes <- probeOnce nodes [] masterpid+         let getByNid _ [] = Nothing+             getByNid nid ((_,n,nodeboss):xs) = if nid==n then Just nodeboss else getByNid nid xs+         res <- liftIO $ withMVar nodes (\n -> return $ getByNid selfnode n)+         _ <- case res of+             Nothing -> taskError "Can't find self: make sure cfgKnownHosts includes the master"+             Just x -> spawnLocalAnd (masterproc x) (do myself <- getSelfPid+                                                        monitorProcess selfpid myself MaLinkError)+         _ <- spawnDaemonic (prober nodes seennodes masterpid)+         return masterpid++stubborn :: (Monad m) => Int -> m (Maybe a) -> m (Maybe a)+stubborn 0 a = a+stubborn n a | n>0+             = do r <- a+                  case r of+                     Just _ -> return r+                     Nothing -> stubborn (n-1) a++-- TODO: setDefaultLocality :: Locality -> TaskM ()++-- | Like 'newPromise', but creates a promise whose+-- values is already known. In other words, it puts+-- a given, already-calculated value in a promise.+-- Conceptually (but not syntactically, due to closures),+-- you can consider it like this:+--+-- > toPromise a = newPromise (return a)+toPromise :: (Serializable a) => a -> TaskM (Promise a)+toPromise = toPromiseAt LcDefault++-- | A variant of 'toPromise' that lets the user+-- express a locality preference, i.e. some information+-- about which node will become the owner of the+-- new promise. These preferences will not necessarily+-- be respected.+toPromiseAt :: (Serializable a) => Locality -> a -> TaskM (Promise a)+toPromiseAt locality a = newPromiseAt locality (passthrough__closure a)++-- | Similar to 'toPromiseAt' and 'newPromiseNear'+toPromiseNear :: (Serializable a,Serializable b) => Promise b -> a -> TaskM (Promise a)+toPromiseNear (PromiseImmediate _) = toPromise+-- TODO should I consult tsRedeemerForwarding here?+toPromiseNear (PromiseBasic prhost _prid) = toPromiseAt (LcByNode [nodeFromPid prhost])++-- | Creates an /immediate promise/, which is to say, a promise+-- in name only. Unlike a regular promise (created by 'toPromise'), +-- this kind of promise contains the value directly. The +-- advantage is that promise redemption is very fast, requiring+-- no network communication. The downside is that it the+-- underlying data will be copied along with the promise.+-- Useful only for small data.+toPromiseImm :: (Serializable a) => a -> TaskM (Promise a)+toPromiseImm = return . PromiseImmediate++-- | Given a function (expressed here as a closure, see "Remote.Call")+-- that computes a value, returns a token identifying that value.+-- This token, a 'Promise' can be moved about even if the+-- value hasn't been computed yet. The computing function +-- will be started somewhere among the nodes visible to the+-- current master, preferring those nodes that correspond+-- to the 'defaultLocality'. Afterwards, attempts to+-- redeem the promise with 'readPromise' will contact the node+-- where the function is executing.+newPromise :: (Serializable a) => Closure (TaskM a) -> TaskM (Promise a)+newPromise = newPromiseAt LcDefault++-- | A variant of 'newPromise' that prefers to start+-- the computing function on the same node as the caller.+-- Useful if you plan to use the resulting value+-- locally.+newPromiseHere :: (Serializable a) => Closure (TaskM a) -> TaskM (Promise a)+newPromiseHere clo =+     do mynode <- liftTask $ getSelfNode+        newPromiseAt (LcByNode [mynode]) clo++-- | A variant of 'newPromise' that prefers to start+-- the computing function on the same node where some+-- other promise lives. The other promise is not+-- evaluated.+newPromiseNear :: (Serializable a, Serializable b) => Promise b -> Closure (TaskM a) -> TaskM (Promise a)+newPromiseNear (PromiseImmediate _) = newPromise+newPromiseNear (PromiseBasic prhost _prid) = newPromiseAt (LcByNode [nodeFromPid prhost]) ++-- | A variant of 'newPromise' that prefers to start+-- the computing functions on some set of nodes that+-- have a given role (assigned by the cfgRole configuration+-- option).+newPromiseAtRole :: (Serializable a) => String -> Closure (TaskM a) -> TaskM (Promise a)+newPromiseAtRole role clo = newPromiseAt (LcByRole [role]) clo++-- | A variant of 'newPromise' that lets the user+-- specify a 'Locality'. The other flavors of newPromise,+-- such as 'newPromiseAtRole', 'newPromiseNear', and+-- 'newPromiseHere' at just shorthand for a call to this function.+newPromiseAt :: (Serializable a) => Locality -> Closure (TaskM a) -> TaskM (Promise a)+newPromiseAt locality clo = +   let realclo = makePayloadClosure clo+    in case realclo of+          Just plclo -> do master <- getMaster+                           res <- roundtrip master (MmNewPromise plclo locality defaultQueueing)+                           case res of+                              Right (MmNewPromiseResponse pid prid) -> return $ PromiseBasic pid prid+                              Right (MmNewPromiseResponseFail) ->+                                          taskError $ "Spawning of closure "++show clo++" by newPromise failed"+                              Left tms -> taskError $ "Spawning of closure "++show clo++" by newPromise resulted in "++show tms+          Nothing -> taskError $ "The specified closure, "++show clo++", can't produce payloads"++-- | Given a promise, gets the value that is being+-- calculated. If the calculation has finished,+-- the owning node will be contacted and the data+-- moved to the current node. If the calculation+-- has not finished, this function will block+-- until it has. If the calculation failed+-- by throwing an exception (e.g. divide by zero),+-- then this function will throw an excption as well+-- (a 'TaskException'). If the node owning the+-- promise is not accessible, the calculation+-- will be restarted.+readPromise :: (Serializable a) => Promise a -> TaskM a+readPromise (PromiseImmediate a) = return a+readPromise thepromise@(PromiseBasic prhost prid) = +       do mp <- lookupCachedPromise prid+          case mp of+            Nothing -> do fprhost <- liftM (maybe prhost id) $ lookupForwardedRedeemer prid+                          res <- roundtrip fprhost (NmRedeem prid)+                          case res of+                             Left e -> do tlogS "TSK" LoInformation $ "Complaining about promise " ++ show prid ++" on " ++show fprhost++" because of "++show e+                                          complain fprhost prid+                             Right NmRedeemResponseUnknown -> +                                       do tlogS "TSK" LoInformation $ "Complaining about promise " ++ show prid ++" on " ++show fprhost++" because allegedly unknown"+                                          complain fprhost prid+                             Right (NmRedeemResponse thedata) -> +                                                  do extracted <- extractFromPayload thedata+                                                     promiseinmem <- liftTaskIO $ makePromiseInMemory thedata (Just $ toDyn extracted)+                                                     putPromiseInCache prid promiseinmem+                                                     return extracted+                             Right NmRedeemResponseException ->+                                  taskError "Failed promise redemption" -- don't redeem, this is a terminal failure+            Just mv -> do val <- liftTaskIO $ readMVar mv -- possible long wait here+                          case val of -- TODO this read/write MVars should be combined!+                             PromiseInMemory v _utc thedyn -> +                                     case thedyn of+                                       Just thedynvalue -> +                                         case fromDynamic thedynvalue of+                                           Nothing -> do liftTask $ logS "TSK" LoStandard "Insufficiently dynamic promise cache"+                                                         extractFromPayload v+                                           Just realval -> do updated <- liftTaskIO $ makePromiseInMemory v thedyn+                                                              putPromiseInCache prid updated+                                                              return realval+                                       Nothing -> do extracted <- extractFromPayload v+                                                     updated <- liftTaskIO $ makePromiseInMemory v (Just $ toDyn extracted)+                                                     putPromiseInCache prid updated+                                                     return extracted+                             PromiseException _ -> taskError $ "Redemption of promise failed"+                             PromiseOnDisk fp -> do mpd <- liftTask $ undiskify fp mv+                                                    _ <- liftTask $ spawnLocal $ diskify fp mv False+                                                    case mpd of+                                                       Just dat -> extractFromPayload dat+                                                       _ -> taskError "Promise extraction from disk failed"+     where extractFromPayload v = do out <- liftTaskIO $ serialDecode v+                                     case out of+                                       Just r -> return r+                                       Nothing -> taskError "Unexpected payload type"+           complain fprhost prid =+                     do master <- getMaster+                        response <- roundtrip master (MmComplain fprhost prid)+                        case response of +                          Left a -> taskError $ "Couldn't file complaint with master about " ++ show fprhost ++ " because " ++ show a+                          Right (MmComplainResponse newhost) +                            | newhost == nullPid -> taskError $ "Couldn't file complaint with master about " ++ show fprhost+                            | otherwise -> do setForwardedRedeemer prid newhost+                                              readPromise thepromise++data TaskState = TaskState+      {+         tsMaster :: ProcessId,+         tsNodeBoss :: Maybe ProcessId,+         tsPromiseCache :: MVar (Map.Map PromiseId (MVar PromiseStorage)),+         tsRedeemerForwarding :: MVar (Map.Map PromiseId ProcessId),+         tsMonitoring :: Map.Map ProcessId ()+      }++data TaskM a = TaskM { runTaskM :: TaskState -> ProcessM (TaskState, a) } deriving (Typeable)++instance Monad TaskM where+   m >>= k = TaskM $ \ts -> do+                (ts',a) <- runTaskM m ts+                (ts'',a') <- runTaskM (k a) (ts')+                return (ts'',a')              +   return x = TaskM $ \ts -> return $ (ts,x)++lookupForwardedRedeemer :: PromiseId -> TaskM (Maybe ProcessId)+lookupForwardedRedeemer q = +   TaskM $ \ts ->+     liftIO $ withMVar (tsRedeemerForwarding ts) $ (\fwd ->+       let lo = Map.lookup q fwd +        in return (ts,lo))++setForwardedRedeemer :: PromiseId -> ProcessId -> TaskM ()+setForwardedRedeemer from to =+   TaskM $ \ts -> liftIO $ modifyMVar (tsRedeemerForwarding ts)  (\fwd ->+     let newmap = Map.insert from to fwd +      in return ( newmap,(ts,())  ) )++lookupCachedPromise :: PromiseId -> TaskM (Maybe (MVar PromiseStorage))+lookupCachedPromise prid = TaskM $ \ts -> +               do mv <- liftIO $ withMVar (tsPromiseCache ts)+                      (\pc -> return $ Map.lookup prid pc)+                  return (ts,mv)++putPromiseInCache :: PromiseId -> PromiseStorage -> TaskM ()+putPromiseInCache prid ps = TaskM $ \ts ->+        do liftIO $ modifyMVar_ (tsPromiseCache ts)+                (\pc -> do mv <- newMVar ps+                           return $ Map.insert prid mv pc)+           return (ts,())++getMaster :: TaskM ProcessId+getMaster = TaskM $ \ts -> return (ts,tsMaster ts)++liftTask :: ProcessM a -> TaskM a+liftTask a = TaskM $ \ts -> a >>= (\x -> return (ts,x))++liftTaskIO :: IO a -> TaskM a+liftTaskIO = liftTask . liftIO++-- | A Task-monadic version of 'Remote.Process.say'.+-- Puts text messages in the log.+tsay :: String -> TaskM ()+tsay a = liftTask $ say a++-- | Writes various kinds of messages to the+-- "Remote.Process" log.+tlogS :: LogSphere -> LogLevel -> String -> TaskM ()+tlogS a b c = liftTask $ logS a b c++----------------------------------------------+-- * MapReduce+----------------------------------------------++-- | A data structure that stores the important+-- user-provided functions that are the namesakes+-- of the MapReduce algorithm.+-- The number of mapper processes can be controlled+-- by the user by controlling the length of the string+-- returned by mtChunkify. The number of reducer+-- promises is controlled by the number of values+-- values returned by shuffler.+-- The user must provide their own mapper and reducer.+-- For many cases, the default chunkifier ('chunkify')+-- and shuffler ('shuffle') are adequate.+data MapReduce rawinput input middle1 middle2 result +    = MapReduce+      {+        mtMapper :: input -> Closure (TaskM [middle1]),+        mtReducer :: middle2 -> Closure (TaskM result),+        mtChunkify :: rawinput -> [input],+        mtShuffle :: [middle1] -> [middle2]+      }++-- | A convenient way to provide the 'mtShuffle' function+-- as part of 'mapReduce'. +shuffle :: Ord a => [(a,b)] -> [(a,[b])]+shuffle q = +    let semi = groupBy (\(a,_) (b,_) -> a==b) (sortBy (\(a,_) (b,_) -> compare a b) q)+     in map (\x -> (fst $ head x,map snd x)) semi ++-- | A convenient way to provide the 'mtChunkify' function+-- as part of 'mapReduce'. +chunkify :: Int -> [a] -> [[a]] +chunkify numChunks l +  | numChunks <= 0 = taskError "Can't chunkify into less than one chunk"+  | otherwise = splitSize (ceiling ((fromIntegral (length l) / fromIntegral numChunks) :: Double)) l+   where+      splitSize _ [] = []+      splitSize i v = let (first,second) = splitAt i v +                       in first : splitSize i second++-- | The MapReduce algorithm, implemented in a very+-- simple form on top of the Task layer. Its+-- use depends on four user-determined data types:+--+-- * input -- The data type provided as the input to the algorithm as a whole and given to the mapper.+--+-- * middle1 -- The output of the mapper. This may include some /key/ which is used by the shuffler to allocate data to reducers.+-- If you use the default shuffler, 'shuffle', this type must have the form @Ord a => (a,b)@.+--+-- * middle2 -- The output of the shuffler. The default shuffler emits a type in the form @Ord => (a,[b])@. Each middle2 output +-- by shuffler is given to a separate reducer.+--+-- * result -- The output of the reducer, upon being given a bunch of middles.+mapReduce :: (Serializable i,Serializable k,Serializable m,Serializable r) =>+             MapReduce ri i k m r -> ri -> TaskM [r]+mapReduce mr inputs =+    let chunks = (mtChunkify mr) inputs +     in do +           pmapResult <- mapM (\chunk -> +                 newPromise ((mtMapper mr) chunk) ) chunks+           mapResult <- mapM readPromise pmapResult+           let shuffled = (mtShuffle mr) (concat mapResult)+           pres <- mapM (\mid2 -> +                  newPromise ((mtReducer mr) mid2)) shuffled+           mapM readPromise pres+
+ Setup.hs view
@@ -0,0 +1,6 @@+module Main (main) where++import Distribution.Simple (defaultMain)++main :: IO ()+main = defaultMain
+ examples/kmeans/KMeans.hs view
@@ -0,0 +1,150 @@+{-# LANGUAGE TemplateHaskell,DeriveDataTypeable,BangPatterns #-}+module Main where++import Remote+import Remote.Process (roundtripResponse,setRemoteNodeLogConfig,getConfig,PayloadDisposition(..),roundtripQuery,roundtripQueryMulti)+import KMeansCommon++import Control.Exception (try,SomeException,evaluate)+import Control.Monad (liftM)+import Control.Monad.Trans (liftIO)+import System.Random (randomR,getStdRandom)+import Data.Typeable (Typeable)+import Data.Data (Data)+import Control.Exception (IOException)+import Data.Binary (Binary,get,put,encode,decode)+import Data.Maybe (fromJust)+import Data.List (minimumBy,sortBy)+import Data.Time+import Data.Either (rights)+import qualified Data.ByteString.Lazy as B+import qualified Data.Map as Map+import System.IO+import Debug.Trace++split :: Int -> [a] -> [[a]] +split numChunks l = splitSize (ceiling $ fromIntegral (length l) / fromIntegral numChunks) l+   where+      splitSize i v = let (first,second) = splitAt i v +                       in first : splitSize i second++broadcast :: (Serializable a) => [ProcessId] -> a -> ProcessM ()+broadcast pids dat = mapM_ (\pid -> send pid dat) pids++multiSpawn :: [NodeId] -> Closure (ProcessM ()) -> ProcessM [ProcessId]+multiSpawn nodes f = mapM (\node -> spawnLink node f) nodes+   where s n = do mypid <- getSelfNode+                  setRemoteNodeLogConfig n (LogConfig LoTrivial (LtForward mypid) LfAll)+                  spawnLink n f++mapperProcess :: ProcessM ()+mapperProcess = +        let mapProcess :: (Maybe [Vector],Maybe [ProcessId],Map.Map Int (Int,Vector)) -> ProcessM ()+            mapProcess (mvecs,mreducers,mresult) = +                         receiveWait+                         [+                            match (\vec -> do vecs<-liftIO $ readf vec+                                              say $ "Mapper read data file" +                                              return (Just vecs,mreducers,mresult)),+                            match (\reducers -> return (mvecs,Just reducers,mresult)),+                            roundtripResponse (\() -> return (mresult,(mvecs,mreducers,mresult))),+                            roundtripResponse+                                  (\clusters -> let tbl = analyze (fromJust mvecs) clustersandcenters Map.empty+                                                    clustersandcenters = map (\x -> (x,clusterCenter x)) clusters+                                                    reducers = fromJust mreducers+                                                    target clust = reducers !! (clust `mod` length reducers)+                                                    sendout (clustid,(count,sum)) =  send (target clustid) Cluster {clId = clustid,clCount=count, clSum=sum}+                                                 in do say $ "calculating: "++show (length reducers)++" reducers"+                                                       mapM_ sendout (Map.toList tbl)+                                                       return ((),(mvecs,mreducers,tbl))),+                            matchUnknownThrow+                         ] >>= mapProcess+            getit :: Handle -> IO [Vector]+            getit h = do l <- liftM lines $ hGetContents h+                         return (map read l) -- evaluate or return?+            readf fn = do h <- openFile fn ReadMode +                          getit h+            condtrace cond s val = if cond+                                      then trace s val+                                      else val+            analyze :: [Vector] -> [(Cluster,Vector)] -> Map.Map Int (Int,Vector) -> Map.Map Int (Int,Vector) +            analyze [] _ ht = ht+            analyze (v:vectors) clusters ht =+                   let theclust = assignToCluster clusters v+                       newh = ht `seq` theclust `seq` Map.insertWith' (\(a,v1) (b,v2) -> let av = addVector v1 v2 in av `seq` (a+b,av) ) theclust (1,v) ht+-- condtrace (blarg `mod` 1000 == 0) (show blarg) $ +                    in newh `seq` analyze vectors clusters newh +            assignToCluster :: [(Cluster,Vector)] -> Vector -> Int+            assignToCluster clusters vector = +                   let distances = map (\(x,center) -> (clId x,sqDistance center vector)) clusters+                    in fst $ minimumBy (\(_,a) (_,b) -> compare a b) distances+            doit = mapProcess (Nothing,Nothing,Map.empty)+         in doit >> return ()++reducerProcess :: ProcessM ()+reducerProcess = let reduceProcess :: ([Cluster],[Cluster]) -> ProcessM ()+                     reduceProcess (oldclusters,clusters) = +                          receiveWait [+                                       roundtripResponse (\() -> return (clusters,(clusters,[]))),+                                       match (\x -> return (oldclusters,combineClusters clusters x)),+                                       matchUnknownThrow] >>= reduceProcess+                     combineClusters :: [Cluster] -> Cluster -> [Cluster]+                     combineClusters [] a = [a]+                     combineClusters (fstclst:rest) clust | clId fstclst == clId clust = (Cluster {clId = clId fstclst,+                                                                                                   clCount = clCount fstclst + clCount clust,+                                                                                                   clSum = addVector (clSum fstclst) (clSum clust)}):rest+                     combineClusters (fstclst:res) clust = fstclst:(combineClusters res clust)+                  in reduceProcess ([],[]) >> return ()+++$( remotable ['mapperProcess, 'reducerProcess] )+++initialProcess "MASTER" = +  do peers <- getPeers+--     say $ "Got peers: " ++ show peers+     cfg <- getConfig+     let mappers = findPeerByRole peers "MAPPER"+     let reducers = findPeerByRole peers "REDUCER"+     let numreducers = length reducers+     let nummappers = length mappers+     say $ "Got " ++ show nummappers ++ " mappers and " ++ show numreducers ++ " reducers"+     clusters <- liftIO $ getClusters "kmeans-clusters"+     say $ "Got "++show (length clusters)++" clusters"+     mypid <- getSelfPid+     +     mapperPids <- multiSpawn mappers mapperProcess__closure++     reducerPids <- multiSpawn  reducers reducerProcess__closure+     broadcast mapperPids reducerPids+     mapM_ (\(pid,chunk) -> send pid chunk) (zip (mapperPids) (repeat "kmeans-points"))++     say "Starting iteration"+     starttime <- liftIO $ getCurrentTime+     let loop howmany clusters = do+           liftIO $ putStrLn $ show howmany+           roundtripQueryMulti PldUser mapperPids clusters :: ProcessM [Either TransmitStatus ()]+           res <- roundtripQueryMulti PldUser reducerPids () :: ProcessM [Either TransmitStatus [Cluster]]+           let newclusters = rights res+           let newclusters2 = (sortBy (\a b -> compare (clId a) (clId b)) (concat newclusters))+           if newclusters2 == clusters || howmany >= 4+              then do+                     donetime <- liftIO $ getCurrentTime+                     say $ "Converged in " ++ show howmany ++ " iterations and " ++ (show $ diffUTCTime donetime starttime)+                     pointmaps <- mapM (\pid -> do (Right m) <- roundtripQuery PldUser pid ()+                                                   return (m::Map.Map Int (Int,Vector))) mapperPids+                     let pointmap = map (\x -> sum $ map fst (Map.elems x)) pointmaps+                     say $ "Total points: " ++ (show $ sum pointmap)+--                     liftIO $ writeFile "kmeans-converged" $ readableShow (Map.toList pointmap)+                     --respoints <- roundtripQueryAsync PldUser mapperPids () :: ProcessM [Either TransmitStatus (Map.Map Int [Vector])]+                     +                     --liftIO $ B.writeFile "kmeans-converged" $ encode $ Map.toList $ Map.unionsWith (++) (rights respoints)+              else+                  loop (howmany+1) newclusters2+     loop 0 clusters++initialProcess "MAPPER" = receiveWait []+initialProcess "REDUCER" = receiveWait []+initialProcess _ = error "Role must be MAPPER or REDUCER or MASTER"++main = remoteInit (Just "config") [Main.__remoteCallMetaData] initialProcess
+ examples/kmeans/KMeans3.hs view
@@ -0,0 +1,67 @@+{-# LANGUAGE TemplateHaskell #-}+module Main where++import Remote++import Control.Monad.Trans+import Control.Monad+import Data.List (minimumBy)++import Debug.Trace++import KMeansCommon++type Line = String+type Word = String++mrMapper :: (Promise [Promise Vector], [Cluster]) -> TaskM [(ClusterId, Promise Vector)]+mrMapper (ppoints,clusters) =+   do points <- readPromise ppoints+      tsay $ "mapping "++show (length points)++" points and "++show (length clusters)++" clusters"+      mapM (assign (map (\c -> (clId c,clusterCenter c)) clusters)) points+  where assign clusters point =+           let distances point = map (\(clid,center) -> (clid,sqDistance center point)) clusters+               assignment point = fst $ minimumBy (\(_,a) (_,b) -> compare a b) (distances point)+            in do vp <- readPromise point+                  vp `seq` return (assignment vp,point)+++mrReducer :: (ClusterId,[Promise Vector]) -> TaskM Cluster+mrReducer (cid,l) = +   do tsay $ "reducing cluster id " ++ show cid ++ " with " ++ show (length l) ++" points"+      let emptyCluster = makeCluster cid []+       in foldM (\c pv -> do v <- readPromise pv+                             c `seq` return $ addToCluster c v) emptyCluster l++$( remotable ['mrMapper,  'mrReducer] )++again :: Int -> (b -> TaskM b) -> b -> TaskM b+again 0 f i = tsay "last iteration" >> f i+again n f i = do tsay (show n++" iterations remaining")+                 q <- f i+                 again (n-1) f q++initialProcess "MASTER" = +                   do setNodeLogConfig defaultLogConfig {logLevel = LoInformation} +                      clusters <- liftIO $ getClusters "kmeans-clusters"+                      points <- liftIO $ getPoints2 "kmeans-points" +                      say $ "starting master"+                      ans <- runTask $+                        do+                           vpoints <- mapM toPromise points+                           ppoints <- mapM toPromise (chunkify 5 vpoints)+                           let myMapReduce = +                                MapReduce +                                {+                                 mtMapper = mrMapper__closure,+                                 mtReducer = mrReducer__closure,+                                 mtChunkify = \clusts -> [(ps,clusts) | ps <- ppoints],+                                 mtShuffle = shuffle+                                }+                           again 4 (mapReduce myMapReduce) clusters+                      say $ "done" -- show ans+initialProcess _ =  setNodeLogConfig defaultLogConfig {logLevel = LoInformation} +                      >> say "starting worker" >> receiveWait [] ++main = remoteInit (Just "config") [Main.__remoteCallMetaData] initialProcess+
+ examples/kmeans/KMeansCommon.hs view
@@ -0,0 +1,86 @@+{-# LANGUAGE DeriveDataTypeable #-}+module KMeansCommon where++import Data.List (foldl')+import Data.Typeable (Typeable)+import Data.Data (Data)+import Data.Binary+import Data.Array.Unboxed+import qualified Data.ByteString.Lazy as B+import Debug.Trace++-- Change this and recompile to change the cardinality of the data set!+vectorSize :: Int+vectorSize = 100++type ClusterId = Int+data Vector = Vector !(UArray Int Double) deriving (Typeable,Eq)+instance Binary Vector where put (Vector a) = put a+                             get = do a<-get+                                      return $ Vector a++instance Read Vector where+  readsPrec i a = let f = readsPrec i a+              in [(Vector $! listArray (0,vectorSize-1) (fst $ head f),snd $ head f)]++instance Show Vector where+   showsPrec _ (Vector a) = showList $ elems a+  ++data Cluster = Cluster+               {+                  clId :: !ClusterId,+                  clCount :: !Int,+                  clSum :: !Vector+               } deriving (Show,Read,Typeable,Eq)+instance Binary Cluster where put (Cluster a b c) = put a>>put b>>put c+                              get = do a<-get+                                       b<-get+                                       c<-get+                                       return $ Cluster a b c++clusterCenter :: Cluster -> Vector+clusterCenter cl = let (Vector arr) = clSum cl+                       count = fromIntegral $ clCount cl+                       newelems = case count of+                                     0 -> replicate (vectorSize) 0+                                     _ -> map (\d -> d / count) (elems arr)+                    in Vector $ listArray (bounds arr) newelems++sqDistance :: Vector -> Vector -> Double+sqDistance (Vector a1) (Vector a2) = sum $ map (\(a,b) -> let dif = a-b in dif*dif) (zip (elems a1) (elems a2))++makeCluster :: ClusterId -> [Vector] -> Cluster+makeCluster clid vecs = Cluster {clId = clid, clCount = length vecs, clSum = vecsum}+   where vecsum = Vector sumArray+         sumArray = listArray (0,vectorSize-1) [ sum $ map ((flip(!))i) (map unVector vecs) | i<-[0..vectorSize-1] ]+         unVector (Vector x) = x+++addCluster :: Cluster -> Cluster -> Cluster+addCluster (Cluster aclid acount asum) (Cluster bclid bcount bsum) = Cluster aclid (acount+bcount) (addVector asum bsum)++addToCluster :: Cluster -> Vector -> Cluster+addToCluster (Cluster aclid acount asum) v = Cluster aclid (acount+1) (addVector asum v)++addVector :: Vector -> Vector -> Vector+addVector (Vector a ) (Vector b) = Vector $! listArray (bounds a) (map (\(a,b) -> a+b) (zip (elems a) (elems b)))++zeroVector :: Vector -> Vector+zeroVector (Vector a) = Vector $! listArray (bounds a) (repeat 0)++getPoints :: FilePath -> IO [Vector]+getPoints fp = do c <- readFile fp+                  return $ read c++getPoints2 :: FilePath -> IO [Vector]+getPoints2 fp = do c <- readFile fp+                   return $ glom (lines c)+     where glom [] = []+           glom (line:rest) = read line : glom rest++getClusters :: FilePath -> IO [Cluster]+getClusters fp = do c <- readFile fp+                    return $ read c++
+ examples/kmeans/MakeData.hs view
@@ -0,0 +1,35 @@+module Main where++import System.Random (randomR,getStdRandom)+import System.Environment+import KMeansCommon+import Data.Binary+import System.IO+import Data.Array.Unboxed+import qualified Data.ByteString.Lazy as B+++-- vectorsPerFile = 80000 -- should be 80000+numClusters = 10+vectorDimensions = KMeansCommon.vectorSize+minValue = -1.0+maxValue = 1.0++val = getStdRandom (randomR (minValue,maxValue))+vector = do vals <- mapM (const val) [1..vectorDimensions]+            return $ Vector $! listArray (0,vectorDimensions-1) vals+file vectorsPerFile = mapM (const vector) [1..vectorsPerFile]++clusters = mapM (\x -> do v <- vector+                          return $ Cluster {clId = x,clCount=1,clSum=v}) [1..numClusters]+            ++makeBig :: Int -> IO ()+makeBig i = do c <- clusters+               withFile "kmeans-points" WriteMode (\h -> mapM (\_ -> do a <- vector ; hPutStrLn h (show a)) [1..i] )  +               writeFile "kmeans-clusters" $ show c++main = do a <- getArgs+          case a of+            ["big",a] -> makeBig (read a)+            _ -> putStrLn "Syntax:\n\tMakeData big 8\n\t\nOutput is in kmeans-points and kmeans-clusters"
+ examples/kmeans/awsgo view
@@ -0,0 +1,286 @@+#!/usr/bin/env python++import sys+import time+import os+import math+import optparse++import boto+import boto.ec2+import subprocess++packages=""++script_header="""#!/bin/bash+append()+{+   file="$1"+   shift+   $@ | tee -a "$file"+}+suappend()+{+   file="$1"+   shift+   $@ | sudo tee -a "$file"+}+try()+{+   log="$HOME/$1"+   shift+   msg="$1"+   shift+   (echo "------ `date`: $@") >> $log+   ($@ 2>&1) | tee -a "$log" | tail -1 > tmp-log+   ret="${PIPESTATUS[0]}"+   if [ 0 != "$ret" ]+   then+     echo 1>&2 "Failed during $msg: `cat tmp-log`"+     rm tmp-log+     return 1+   fi+   rm tmp-log+   return 0+}+try1()+{+   if ! try "$@"+   then +     echo 1>&2 "Retrying..."+     sleep 10+     if ! try "$@"+     then    +        echo 1>&2 "Retrying again..."+        sleep 20+        if ! try "$@"+        then+           echo 1>&2 "Giving up."+           exit 1+        fi+     fi +   fi+   return 0+}+"""+exec_args=["-cfgRoundtripTimeout=20000000","+RTS","-K20000000"]+local_files = ["KMeans.hs","MakeData.hs","KMeansCommon.hs"]+node_config_data = {"apts":"ghc cabal-install git-core libghc6-binary-dev libghc6-mtl-dev libghc6-network-dev libghc6-stm-dev"}+node_config_script = script_header+"""+cd+try cfg.log "update apt repository" sudo apt-get -q -y update+try cfg.log "install apt packages" sudo apt-get -q -y install %(apts)s+try1 cfg.log "cabalupdate" cabal update+try1 cfg.log "cabalify" cabal install pureMD5 utf8-string directory crypto-api tagged data-default cereal semigroups+try cfg.log "retrieve git repository" git clone git://github.com/jepst/CloudHaskell.git+cd CloudHaskell+try cfg.log "configure local repository" cabal configure+try1 cfg.log "build local repository" cabal build+try1 cfg.log "install local repoistory" cabal install+cd+try cfg.log "compile executable" ghc -O3 --make KMeans.hs+""" % node_config_data++def writefile(fname,content):+    with open(fname,"w") as f:+        f.write(content)++def err(s):+    print s+    sys.exit(1)++def waitforallup(instances):+    while True:+        pending = False+        running = False+        for instance in instances:+            if instance.state == u'pending':+                pending = True+                instance.update()+            elif instance.state == u'running':+                running = True+            else:+                print("Encountered unexpected instance state: "+instance.state) # and terminate remaining instances?+        if pending:+            time.sleep(3)+        else:+            break ++ssh_opts=["-o","BatchMode=yes","-o","ServerAliveInterval=240","-o","StrictHostKeyChecking=no"]++def allinstances_xfer(keyfile,username,instances,thefile):+    processes = []+    for i in instances:+       cmd=["scp","-r","-q","-i",keyfile]+ssh_opts + thefile + [username+"@"+i.public_dns_name+":"]+       while True:+          subproc=subprocess.Popen(cmd,stderr=subprocess.STDOUT)+          result = subproc.wait()+          if result == 0:+             break+          else:+             print "Problem transfering",thefile,"to",i.public_dns_name," and retrying"+             time.sleep(60)++def allinstances_recv(keyfile,username,instances,thefile):+    processes = []+    for i in instances:+       cmd=["scp","-r","-q","-i",keyfile]+ssh_opts+ [ username+"@"+i.public_dns_name+":"+thefile,thefile]+       subproc=subprocess.Popen(cmd,stderr=subprocess.STDOUT)+       processes.append(subproc)+    for (i,p) in zip(instances,processes):+        result = p.wait()+        if result != 0:+            err("Error with remote receiving on instance "+i.id+" at "+i.public_dns_name)++def allinstances_runscript(keyfile,username,instances,scriptmaker):+    processes = []+    counter=0+    for i in instances:+       text = scriptmaker(i)+       fname="tmp-"+str(counter) # use random nm here+       writefile(fname,text)+       cmd1=["scp","-r","-q","-i",keyfile]+ssh_opts+[ fname ,username+"@"+i.public_dns_name+":"]+       cmd2=["ssh","-q","-i",keyfile]+ssh_opts+["-l",username,"-o","StrictHostKeyChecking=no", i.public_dns_name,"bash",fname]+       cmd = ["bash","-c",(" ".join(cmd1))+" ; "+(" ".join(cmd2))]+       subproc=subprocess.Popen(cmd,stderr=subprocess.STDOUT)+       processes.append(subproc)+       counter=counter+1+       time.sleep(5)+    for (i,p) in zip(instances,processes):+        result = p.wait()+        if result != 0:+            err("Error during remote script execution on instance "+i.id+" at "+i.public_dns_name)+    for i in range(counter):+        os.remove("tmp-"+str(i))+++def allinstances_exec(keyfile,username,instances,thecmd):+    processes = []+    for i in instances:+       cmd=["ssh","-i",keyfile,"-l",username]+ssh_opts+[ i.public_dns_name]+thecmd+#       print " ".join(cmd)+       subproc=subprocess.Popen(cmd,stderr=subprocess.STDOUT)+       processes.append(subproc)+    for (i,p) in zip(instances,processes):+        result = p.wait()+        if result != 0:+            err("Error "+str(result)+" with remote process on instance "+i.id+" at "+i.public_dns_name)++def allinstances_spawn(keyfile,username,instances,thecmd):+    processes = []+    for i in instances:+       cmd=["ssh","-n","-q","-i",keyfile,"-l",username]+ssh_opts+[ i.public_dns_name]+thecmd+       subproc=subprocess.Popen(cmd,stderr=subprocess.STDOUT)+       processes.append(subproc)+       time.sleep(5)++def hostscript(instance):+    hostsfile = "try cfg.log 'setup hostname' suappend /etc/hosts echo 127.0.1.1 `hostname`"+    return hostsfile++def main():+    parser = optparse.OptionParser()+    parser.add_option("-q", "--datamultiple",help="Data multiplier (default 1)",dest="data_multiple",default=1,type="int")+    parser.add_option("-r", "--region", help="Region (default eu-west-1)", dest="region", default="eu-west-1")+    parser.add_option("-i", "--image", help="Virtual machine image (default ami-311f2b45)",dest="ami",default="ami-311f2b45")+# other useful ubuntu instances at+# and here http://uec-images.ubuntu.com/releases/10.04/release/+# and http://aws.amazon.com/ec2/instance-types/+    parser.add_option("-m", "--nmapers", help="Number of mapper nodes (default 10)",dest="mapper_nodes",type="int",default=10)+    parser.add_option("-n", "--nreducers", help="Number of reducer nodes (default 3)",dest="reducer_nodes",type="int",default=3)+    parser.add_option("-p", "--nodesperhost",help="Number of nodes to start on each host (default 4)",dest="nodes_per_host",type="int",default=4)+    parser.add_option("-k", "--keypair",help="Keypair for starting EC2 instances (default aws) -- will look for corresponding file in ~/.ssh/",dest="keypair",default="aws")+    parser.add_option("-s", "--securitygroup",help="Name of security group for launching instances (default default)",dest="securitygroup",default="default")+    parser.add_option("-t","--instancetype",help="Name of instance type for launching instances (default m1.small)",dest="instancetype",default="m1.small")+    parser.add_option("-l","--login-user",help="Username for connecting to instances (default ubuntu)",dest="username",default="ubuntu")+    (options, args) = parser.parse_args()++    if os.sep in options.keypair:+        s = os.path.split(options.keypair)[-1]+        keypair = s.split(".pem")[0]+        keypairfile = os.path.expanduser(options.keypair)+        if not os.path.exists(keypairfile):+            err("Specified keypair file "+keypairfile+" doesn't exist")+    else:+        keypairbase = options.keypair+        attempts = [os.path.join("~",".ssh,",keypairbase),+                    os.path.join("~",".ssh,",keypairbase+".pem"),+                    os.path.join("~",".ec2",keypairbase),+                    os.path.join("~",".ec2",keypairbase+".pem") ]+        for attempt in attempts:+            path = os.path.expanduser(attempt)+            if os.path.exists(path):+                keypair = options.keypair+                keypairfile = path+                break+        else:+            err("Could not find a keyfile matching \""+keypairbase+"\" in ~/.ssh or ~/.ec2") ++    try:+        for r in boto.ec2.regions():+            if r.name == options.region:+                region = r+                break+        else:+            err("Region %s not found." % options.region)+    except AttributeError:+       err("Can't connect to AWS. Set AWS_SECRET_ACCESS_KEY and AWS_ACCESS_KEY_ID environment variables to the values visible under your Security Credentials screen.")+    +    ec2 = boto.connect_ec2(region=region)++    num_nodes = 1 + options.mapper_nodes + options.reducer_nodes+    numhosts = int (math.ceil (float(num_nodes) / options.nodes_per_host))++    print "Starting "+str(numhosts)+" instances of image "+options.ami+"."+    image = ec2.get_all_images(image_ids=[options.ami])[0]+    reservation = image.run(numhosts,numhosts,key_name=keypair,instance_type=options.instancetype,security_groups=[options.securitygroup])+    waitforallup(reservation.instances)+    time.sleep(30)++    print "Configuring instances."+    allinstances_xfer(keypairfile,options.username,reservation.instances,local_files)+    allinstances_runscript(keypairfile,options.username,reservation.instances,lambda a: node_config_script + hostscript(a) + "\n")++    config_file="cfgKnownHosts "+(" ".join([i.private_dns_name for i in reservation.instances]))+    writefile("config",config_file)+    allinstances_xfer(keypairfile,options.username,reservation.instances,["config"])++    print "Transfering data."+    mappers=[]+    reducers=[]+    master=[]+    insts = reservation.instances * options.nodes_per_host+    master.append(insts.pop(0))+    for i in range(options.mapper_nodes):+        mappers.append(insts.pop(0))+    for i in range(options.reducer_nodes):+        reducers.append(insts.pop(0))+    allinstances_xfer(keypairfile,options.username,reservation.instances,["kmeans-points","kmeans-clusters"])++    print "Starting worker nodes."+    allinstances_spawn(keypairfile,options.username,mappers,["./KMeans","-cfgRole=MAPPER"]+exec_args)+    allinstances_spawn(keypairfile,options.username,reducers,["./KMeans","-cfgRole=REDUCER"]+exec_args)+    time.sleep(30)++    print "Starting master node on "+master[0].public_dns_name+"."+    starttime=time.time()+    allinstances_exec(keypairfile,options.username,master,["./KMeans","-cfgRole=MASTER"]+exec_args)+    endtime=time.time()+#    print "Completed computation in "+str(endtime-starttime)+" seconds."++#    print "Retrieving results."+#    allinstances_recv(keypairfile,options.username,master,"kmeans-converged")++    print "Shutting down instances."+    for i in reservation.instances:+        i.stop()++def safe_main():+    try:+        main()+    except boto.exception.EC2ResponseError  as a:+        err("Error from EC2 service: "+a.error_message+": "+a.reason)++if __name__ == "__main__":+    safe_main()+
+ examples/kmeans/awskill view
@@ -0,0 +1,30 @@+#!/usr/bin/env python++import optparse++import boto+import boto.ec2++def main():+    parser = optparse.OptionParser()+    parser.add_option("-r", "--region", help="Region (default eu-west-1)", dest="region", default="eu-west-1")+    (options, args) = parser.parse_args()++    for r in boto.ec2.regions():+        if r.name == options.region:+            region = r+            break+        else:+            err("Region %s not found." % options.region)++    ec2 = boto.connect_ec2(region=region)++    reservation = ec2.get_all_instances()+    for i in reservation:+        for n in i.instances:+            print "Stopping ",n.public_dns_name+"."+            n.stop()++if __name__ == "__main__":+    main()+
+ examples/kmeans/awslist view
@@ -0,0 +1,29 @@+#!/usr/bin/env python++import optparse++import boto+import boto.ec2++def main():+    parser = optparse.OptionParser()+    parser.add_option("-r", "--region", help="Region (default eu-west-1)", dest="region", default="eu-west-1")+    (options, args) = parser.parse_args()++    for r in boto.ec2.regions():+        if r.name == options.region:+            region = r+            break+        else:+            err("Region %s not found." % options.region)++    ec2 = boto.connect_ec2(region=region)++    reservation = ec2.get_all_instances()+    for i in reservation:+        for n in i.instances:+            print "Found ",n.public_dns_name+"."++if __name__ == "__main__":+    main()+
+ examples/kmeans/kmeans view
@@ -0,0 +1,29 @@+#!/bin/bash++nmappers=2+nreducers=2++pids=""+host=`hostname`++args="-cfgRoundtripTimeout=2000000000 -cfgKnownHosts=$host +RTS -K20000000"++# ghc --make KMeans || exit 1++for i in $( seq 1 ${nmappers} )+do+  ./KMeans -cfgRole=MAPPER  $args &+  pid=$!+  pids="$pids $pid"+done++for i in $( seq 1 ${nreducers} )+do+  ./KMeans -cfgRole=REDUCER  $args &+  pid=$!+  pids="$pids $pid"+done++./KMeans -cfgRole=MASTER $args++kill $pids
+ examples/pi/Pi6.hs view
@@ -0,0 +1,45 @@+{-# LANGUAGE TemplateHaskell #-}+module Main where++import Remote+import PiCommon++worker :: Int -> Int -> ProcessId -> ProcessM ()+worker count offset master = +   let (numin,numout) = countPairs offset count +    in do send master (numin,numout)+          logS "PI" LoInformation ("Finished mapper from offset "++show offset)++$( remotable ['worker] )++initialProcess :: String -> ProcessM ()+initialProcess "WORKER" =+  receiveWait []++initialProcess "MASTER" = +  do { peers <- getPeers+     ; mypid <- getSelfPid+     ; let { workers = findPeerByRole peers "WORKER"+     ;       interval = 1000000+     ;       numberedworkers = (zip [0,interval..] workers) }+     ; mapM_ (\ (offset,nid) -> spawn nid (worker__closure (interval-1) offset mypid)) numberedworkers+     ; (x,y) <- receiveLoop (0,0) (length workers)+     ; let est = estimatePi (fromIntegral x) (fromIntegral y)+        in say ("Done: " ++ longdiv (fst est) (snd est) 20) }+  where +    estimatePi ni no | ni + no == 0 = (0,0)+                     | otherwise = (4 * ni , ni+no)+    receiveLoop a 0 = return a+    receiveLoop (numIn,numOut) n = +      let +        resultMatch = match (\ (x,y) -> return (x::Int,y::Int))+      in do { (newin,newout) <- receiveWait [resultMatch]+            ; let { x = numIn + newin+            ;       y = numOut + newout }+            ; receiveLoop (x,y) (n-1) }++initialProcess _ = error "Role must be WORKER or MASTER"++main = remoteInit (Just "config") [Main.__remoteCallMetaData] initialProcess++
+ examples/pi/Pi7.hs view
@@ -0,0 +1,40 @@+{-# LANGUAGE TemplateHaskell #-}+module Main where++import Remote+import PiCommon+import Data.List (foldl')++worker :: Int -> Int -> TaskM (Int,Int)+worker count offset = +   let (numin,numout) = countPairs offset count +    in do tlogS "PI" LoInformation ("Finished mapper from offset "++show offset)+          return (numin,numout)++$( remotable ['worker] )++initialProcess :: String -> ProcessM ()+initialProcess "WORKER" =+  receiveWait []++initialProcess "MASTER" = +  let+        interval = 1000000+        numworkers = 5+        numberedworkers = take numworkers [0,interval..]+        clos = map (worker__closure interval) numberedworkers+  in runTask $ +        do proms <- mapM newPromise clos+           res <- mapM readPromise proms+           let (sumx,sumy) = foldl' (\(a,b) (c,d) -> (a+c,b+d)) (0,0) res+           let (num,den) = estimatePi (fromIntegral sumx) (fromIntegral sumy)+           tsay ("Done: " ++ longdiv (num) (den) 20)+  where +    estimatePi ni no | ni + no == 0 = (0,0)+                     | otherwise = (4 * ni , ni+no)++initialProcess _ = error "Role must be WORKER or MASTER"++main = remoteInit (Just "config") [Main.__remoteCallMetaData] initialProcess++
+ examples/pi/PiCommon.hs view
@@ -0,0 +1,72 @@+module PiCommon where++import Data.List+import Data.Array++type Number = Double++data Seq = Seq {k::Int, x::Number, base::Int, q::Array Int Number, tobreak::Bool}++haltonSeq :: Int -> Int -> [Number]+haltonSeq offset thebase = let+                    digits = 64::Int+                    seqSetup :: Seq -> Int -> Number -> (Seq, Number)+                    seqSetup s j _ = +                                     let dj =  (k s `mod` base s)+                                         news = s {k = (k s - dj) `div` fromIntegral (base s), x = x s + (fromIntegral dj * ((q s) ! (j+1)))}     +                                      in+                                        (news,fromIntegral dj)+                    seqContinue :: Seq -> Int -> Number -> (Seq, Number)+                    seqContinue s j dj = +                                     if tobreak s+                                        then (s,dj)+                                        else +                                               let newdj = dj+1+                                                   newx = x s + (q s) ! (j+1)+                                               in+                                               if newdj < fromIntegral (base s)+                                                  then (s {x=newx,tobreak=True},newdj)+                                                  else (s {x = newx - if j==0 then 1 else (q s) ! j},0)++                    initialState base = let q = array (0,digits*2) [(i,v) | i <- [0..digits*2], let v = if i == 0 then 1 else ((q ! ((i)-1))/fromIntegral base)]+                                        in Seq {k=fromIntegral offset,x=0,tobreak=False,base=base,q=q}+                    theseq base = let+                        first :: (Int,[Number],Seq)+                        first = foldl' (\(n,li,s) _ -> let (news,r) = seqSetup s n 0+                                                       in (n+1,r:li,news)) (0,[],initialState base) [0..digits]+                        second :: [Number] -> Seq -> (Int,[Number],Seq)+                        second d s = foldl' (\(n,li,s) dj -> let (news,r) = seqContinue s n dj+                                                             in (n+1,r:li,news)) (0,[],s {tobreak=False}) d+                        in let (_,firstd,firsts) = first+                               therest1 :: [([Number],Seq)]+                               therest1 = iterate (\(d,s) -> let (_,newd,news) = second (reverse d) s in (newd,news)) (firstd,firsts)+                               therest :: [Number]+                               therest = map (\(_,s) -> x s) therest1+                           in therest+                        in (theseq thebase)++haltonPairs :: Int -> [(Number,Number)]+haltonPairs offset = +  zip (haltonSeq offset 2) (haltonSeq offset 3)++countPairs :: Int -> Int -> (Int,Int)+countPairs offset count = +  let range = take count (haltonPairs offset)+      numout = length (filter outCircle range)+   in (count-numout,numout)+  where+     outCircle (x,y) = +          let fx=x-0.5 +              fy=y-0.5+           in fx*fx + fy*fy > 0.25++longdiv :: Integer -> Integer -> Integer -> String+longdiv _ 0 _ = "<inf>"+longdiv numer denom places = +  let attempt = numer `div` denom +   in if places==0+         then "" +         else shows attempt (longdiv2 (numer - attempt*denom) denom (places -1))+   where longdiv2 numer denom places | numer `rem` denom == 0 = "0"+                                     | otherwise = longdiv (numer * 10) denom places+
+ examples/tests/Test-Call.hs view
@@ -0,0 +1,59 @@+{-# LANGUAGE TemplateHaskell,DeriveDataTypeable #-}+module Main where	++import Remote++import Control.Exception (throw)+import Data.Generics (Data)+import Data.Maybe (fromJust)+import Data.Typeable (Typeable)+import Control.Monad (when,forever)+import Control.Monad.Trans (liftIO)+import Data.Char (isUpper)+import Control.Concurrent (threadDelay)+import Data.Binary+import Control.Concurrent.MVar++sayHi :: ProcessId -> ProcessM ()+sayHi s = do liftIO $ threadDelay 500000+             say $ "Hi there, " ++ show s ++ "!"++add :: Int -> Int -> Int+add a b = a + b++badFac :: Integer -> Integer+badFac 0 = 1+badFac 1 = 1+badFac n = badFac (n-1) + badFac (n-2)++remotable ['sayHi, 'add, 'badFac]+++while :: (Monad m) => m Bool -> m ()+while a = do f <- a+             when (f)+                (while a >> return ())+             return ()++initialProcess "MASTER" = do+              mypid <- getSelfPid+              mynode <- getSelfNode+              peers <- getPeers++              let slaves = findPeerByRole peers "SLAVE"+              case slaves of+                (somenode:_) -> +                    do say $ "Running badFac on " ++show somenode+                       res <- callRemotePure somenode (badFac__closure 40)+                       say $ "Got result: " ++ show res+                _ -> say "Couldn't find a SLAVE node to run program on; start a SLAVE, then the MASTER"+++              return ()+initialProcess "SLAVE" = receiveWait []+initialProcess _ = liftIO $ putStrLn "Please use parameter -cfgRole=MASTER or -cfgRole=SLAVE"++testSend = remoteInit (Just "config") [Main.__remoteCallMetaData] initialProcess++main = testSend+
+ examples/tests/Test-Channel-Merge.hs view
@@ -0,0 +1,39 @@+{-# LANGUAGE TemplateHaskell,DeriveDataTypeable #-}+module Main where	++-- This example demonstrates the difference between+-- the biased and round-robin methods for merging+-- channels. Run this program twice, once with+-- the parameter "biased" and once with "rr";+-- the order that the messages will be received+-- in will change, even though the order that they+-- are sent in does not.++import Remote++channelCombiner args = case args of+                         ["biased"] ->  combinePortsBiased+                         ["rr"] -> combinePortsRR+                         _ -> error "Please specify 'biased' or 'rr' on the command line"++initialProcess _ = do+              mypid <- getSelfPid+              args <- getCfgArgs++              (sendchan,recvchan) <- newChannel+              (sendchan2,recvchan2) <- newChannel++              spawnLocal $ mapM_ (sendChannel sendchan) [1..(26::Int)]+              spawnLocal $ mapM_ (sendChannel sendchan2) ['A'..'Z']++              merged <- (channelCombiner args) [combinedChannelAction recvchan show,combinedChannelAction recvchan2 show]+              let go = do item <- receiveChannel merged+                          say $ "Got: " ++ show item+                          go+              go++main = remoteInit (Just "config") [] initialProcess++++
+ examples/tests/Test-Channel.hs view
@@ -0,0 +1,48 @@+{-# LANGUAGE TemplateHaskell,DeriveDataTypeable #-}+module Main where++-- Simple example of putting data in a channel+-- and then taking it out.	++import Remote++import Data.Binary (Binary,get,put)+import Data.Char (isUpper)+import Data.Generics (Data)+import Data.Typeable (Typeable)+import Prelude hiding (catch)+import Data.Typeable (typeOf)+import Control.Monad.Trans+import Control.Exception+import Control.Monad+import Data.Maybe (fromJust)+import Control.Concurrent++initialProcess "NODE" = do++              (sendchan,recvchan) <- newChannel++              a <- spawnLocal $ do+                              sendChannel sendchan "hi"+                              sendChannel sendchan "lumpy"+                              liftIO $ threadDelay 1000000+                              sendChannel sendchan "spatula"+                              sendChannel sendchan "noodle"+                              mapM_ (sendChannel sendchan) (map show [1..1000])+              liftIO $ threadDelay 500000+              receiveChannel recvchan >>= liftIO . print+              receiveChannel recvchan >>= liftIO . print+              receiveChannel recvchan >>= liftIO . print+              receiveChannel recvchan >>= liftIO . print+              receiveChannel recvchan >>= liftIO . print+              receiveChannel recvchan >>= liftIO . print+              receiveChannel recvchan >>= liftIO . print+              receiveChannel recvchan >>= liftIO . print+              receiveChannel recvchan >>= liftIO . print+              receiveChannel recvchan >>= liftIO . print++++main = remoteInit (Just "config") [] initialProcess++
+ examples/tests/Test-Closure.hs view
@@ -0,0 +1,111 @@+{-# LANGUAGE TemplateHaskell #-}+module Main where	++-- This file contains some examples of closures, which are the+-- way to express a function invocation in Cloud Haskell. You+-- need to use closures to run code on a remote system, which is,+-- after all, the whole point. See the documentation on the function+-- 'remotable' for more details. Understanding this example is+-- essential to effectively using Cloud Haskell.++import Remote+import Remote.Call (mkClosure)++import Data.List (sort)++-- Step #1: Define the functions that we would+-- like to call remotely. These are just regular+-- functions.+           +sayHi :: String -> ProcessM ()+sayHi s = say $ "Greetings, " ++ s++sayHiPM :: String -> ProcessM String+sayHiPM s = say ("Hello, "++s) >> +               return (sort s)++sayHiIO :: String -> IO String+sayHiIO s = putStrLn ("Hello, " ++ s) >>+                return (reverse s)++sayHiPure :: String -> String+sayHiPure s = "Hello, " ++ s++simpleSum :: Int -> Int -> Int+simpleSum a b = a + b + 1++-- You can also manually call a closure using invokeClosure.+-- This is what functions like spawn and callRemote do+-- internally.+runAnother :: Closure Int -> ProcessM String+runAnother c = do mi <- invokeClosure c+                  case mi of+                    Just i -> return $ concat $ replicate i "Starscream"+                    Nothing -> return "No good"++-- This a partial closure: some parameters are provided by the caller,+-- and some are provided after the closure is invoked. You have to+-- write the function in a funny way to make this work automatically,+-- but otherwise it's pretty straightforward.+funnyHi :: String -> ProcessM (Int -> ProcessM ())+funnyHi s = return $ \i -> say ("Hello, " ++ (concat $ replicate i (reverse s)))++-- Step #2: Automagically generate closures+-- for these functions using remotable. For each+-- given function n, remotable will create a+-- closure for that function named n__closure.+-- You can then use that closure with spawn, remoteCall,+-- and invokeClosure. See examples below.++remotable ['sayHi, 'sayHiIO,'sayHiPure, 'sayHiPM, 'funnyHi, 'runAnother, 'simpleSum]++initialProcess _ = do+              mynid <- getSelfNode++              -- spawn and callRemote (and their variants) run+              -- a function on a given node. We indicate which+              -- node by giving a node ID (to keep it simple+              -- we do everything on one node in this+              -- example), and we indicate which function to run+              -- by providing its closure.++              -- A simple spawn. Does not block, and the result+              -- we get back is the PID of the new process.+              p <- spawn mynid (sayHi__closure "Zoltan")+              say $ "Got result " ++ show p++              -- callRemote is like a synchronous version of spawn.+              -- It will block until the function ends, and returns+              -- its result.+              v <- callRemote mynid ( sayHiPM__closure "Jaroslav")+              say $ "Got result " ++ v++              -- We need a different function to call closures in the+              -- IO monad. Also, instead of using the "something__closure"+              -- syntax, you can call the Template Haskell mkClosure+              -- function, which expands to the same thing.+              w <- callRemoteIO mynid ( $(mkClosure 'sayHiIO) "Noodle")+              say $ "Got result " ++ show w++              -- Yet another version of callRemote for nonmonadic functions.+              q <- callRemotePure mynid (sayHiPure__closure "Spatula")+              say $ "Got result " ++ show q++              -- We can even give closures to closures. They can in turn run+              -- them indirectly (with spawn or callRemote) or directly+              -- (with invokeClosure).+              x <- callRemote mynid (runAnother__closure (simpleSum__closure 1 1))+              say $ "Got result " ++ x++              -- This function takes some parameters after closure invocation.+              -- So the value we get back from invokeClosure is actually+              -- a partially evaluated function.+              mfunnyFun <- invokeClosure (funnyHi__closure "Antwerp")+              case mfunnyFun of+                Just funnyFun -> funnyFun 3+                Nothing -> say "No good"++              return ()++main = remoteInit (Just "config") [Main.__remoteCallMetaData] initialProcess+
+ examples/tests/Test-MapReduce.hs view
@@ -0,0 +1,47 @@+{-# LANGUAGE TemplateHaskell #-}+module Main where++-- A simple word frequency counter using the task layer's mapreduce.++import Remote++import Control.Monad.Trans+import Control.Monad++import Debug.Trace++type Line = String+type Word = String++mrMapper :: [Line] -> TaskM [(Word, Int)]+mrMapper lines = +      return (concatMap (\line -> map (\w -> (w,1)) (words line)) lines)++mrReducer :: (Word,[Int]) -> TaskM (Word,Int)+mrReducer (w,p) = +    return (w,sum p)++$( remotable ['mrMapper,  'mrReducer] )++myMapReduce = MapReduce +               {+                 mtMapper = mrMapper__closure,+                 mtReducer = mrReducer__closure,+                 mtChunkify = chunkify 5,+                 mtShuffle = shuffle+               }++initialProcess "MASTER" = +                   do args <- getCfgArgs+                      case args of+                        [filename] -> +                          do file <- liftIO $ readFile filename+                             ans <- runTask $+                                do mapReduce myMapReduce (lines file)+                             say $ show ans+                        _ -> say "When starting MASTER, please also provide a filename on the command line"+initialProcess "WORKER" = receiveWait [] +initialProcess _ = say "You need to start this program as either a MASTER or a WORKER. Set the appropiate value of cfgRole on the command line or in the config file."++main = remoteInit (Just "config") [Main.__remoteCallMetaData] initialProcess+
+ examples/tests/Test-Message.hs view
@@ -0,0 +1,62 @@+{-# LANGUAGE DeriveDataTypeable,TemplateHaskell #-}+module Main where++-- A convoluted example showing some different ways+-- messages can be passed between processes.++import Remote++import Data.Typeable (Typeable)+import Data.Data (Data)+import Data.Binary (Binary,get,put)+import System.Random (randomR,getStdRandom)+import Control.Concurrent (threadDelay)+import Control.Monad (when)+import Control.Monad.Trans (liftIO)++data Chunk = Chunk Int deriving (Typeable,Data)++instance Binary Chunk where+    get = genericGet+    put = genericPut++slaveWorker :: ProcessM ()+slaveWorker = +   receiveWait [match(\sndport -> mapM (sendrand sndport) [0..50] >> sendChannel sndport (Chunk 0))]+  where sendrand sndport _ = +          do newval <- liftIO $ getStdRandom (randomR (1,100))+             sendChannel sndport (Chunk newval)++$( remotable ['slaveWorker] )++testNode :: ProcessId -> ProcessM ()+testNode pid =+  let getvals mainpid rcv =+        do (Chunk val) <- receiveChannel rcv+           send mainpid val+           when (val /= 0)+              (getvals mainpid rcv)+   in+    do (sendport, receiveport) <- newChannel +       self <- getSelfPid+       send pid sendport+       spawnLocal $ getvals self receiveport+       return ()+                      ++initialProcess "MASTER" =+     do peers <- getPeers+        let slaves = findPeerByRole peers "SLAVE"++        slavepids <- mapM (\slave -> spawn slave slaveWorker__closure) slaves+        mapM_ testNode slavepids++        let getsome = +             do res <- receiveWait [match(\i -> say ("Got " ++ show (i::Int)) >> return (i/=0))]+                when res getsome+        mapM_ (const getsome) slaves++initialProcess "SLAVE" = receiveWait []++main = remoteInit (Just "config") [Main.__remoteCallMetaData] initialProcess+
+ examples/tests/Test-Task.hs view
@@ -0,0 +1,55 @@+{-# LANGUAGE TemplateHaskell #-}+module Main where++-- A demonstration of the data dependency resolution +-- of the Task layer. Each newPromise spawns a task+-- that calculates a value. Calls to adder and diver+-- take integers, but calls to merger take Promises,+-- which are an expression of the value computed+-- or yet-to-be-computed by another task. The+-- call to readPromise will retrieve that value+-- or wait until it's available.++-- You can run this program on one node, or+-- on several; the task layer will automatically+-- allocate tasks to whatever nodes it can talk to.+-- Unlike the Process layer, the programmer doesn't+-- need to specify the location.++import Remote++adder :: Int -> Int -> TaskM Int+adder a b = do return $ a + b++diver :: Int -> Int -> TaskM Int+diver a b = do return $ a `div` b++merger :: Promise Int -> Promise Int -> TaskM Int+merger a b = do a1 <- readPromise a+                b1 <- readPromise b+                return $ a1+b1+++$( remotable ['diver,'adder, 'merger] )++initialProcess "MASTER" = +                   do ans <- runTask $+                          do +                             a <- newPromise (diver__closure 300 9)+                             b <- newPromise (diver__closure 300 5)+                             c <- newPromise (diver__closure 300 1)+                             d <- newPromise (adder__closure 300 9)+                             e <- newPromise (adder__closure 300 5)+                             f <- newPromise (adder__closure 300 1)+                             g <- newPromise (merger__closure a b)+                             h <- newPromise (merger__closure g c)+                             i <- newPromise (merger__closure c d)+                             j <- newPromise (merger__closure e f)+                             k <- newPromise (merger__closure i j)+                             l <- newPromise (merger__closure k g)+                             readPromise l+                      say $ show ans+initialProcess "WORKER" = receiveWait [] ++main = remoteInit (Just "config") [Main.__remoteCallMetaData] initialProcess+
+ examples/tests/config view
@@ -0,0 +1,103 @@+# config+# Cloud Haskell config file.+# Most Cloud Haskell applications will look for this file+# in the current directory, although it's ultimately up+# to the programmer which filename to use or if to+# use a config file at all. You may override+# the default location with the RH_CONFIG+# environment variable.+# All the options given in this file may+# also be given on the command line of a+# Cloud Haskell application. Options on the+# command line override options in a config+# file. For example, to set the role of an+# application, use the following command+# line:+#      ./MyApp -cfgRole=MASTER+#+#   This is a reasonable default config file.++# The role of a node determines its initial behavior.+# Typical roles including MASTER or WORKER. Typically,+# when starting an application, this value must be+# specified, either here or on the command line.+# Default: NODE+cfgRole NODE++# If specified, this option will override the+# default assignment of host name, which forms+# port of all node IDs and process IDs originating+# on this node.+### cfgHostName somehostname++# If nonzero, will force the node to listen for+# new incoming connections on a specific port.+# Otherwise, the port is assigned by the OS.+# Default: 0+cfgListenPort 0++# This is the UDP port on which local network+# broadcasts are sent to discover peers.+# If 0, no dynamic peer discovery is performed.+# This value must be the same for all nodes+# that want to participate in dynamic discovery+# with each other.+# Default: 38813+cfgPeerDiscoveryPort 38813++# A unique token that nodes exchange when they+# communicate. The intent is to prevent different+# applications on the same network from mixing+# up their nodes. This value must be the same+# for all nodes that want to communicate with each+# other.+# Default: MAGIC+cfgNetworkMagic MAGIC++# Specifies the port where the node registration+# server will bind. This value must be the same+# for all nodes that want to communicate with each+# other.+# Default: 38813+cfgLocalRegistryListenPort 38813++# A list of hosts where nodes will be searched for.+# All hosts that you wish to communicate with should+# be included in this list. By host name, we mean+# the DNS name or IP address of a computer accessible+# over the network interface where a node registry+# may be running. Hostnames are separate by spaces.+### cfgKnownHosts somehostname anotherhostname thirdhostname++# The maximum delay, in microseconds, to tolerate+# delays from service services. In an extremely congested+# network atmosphere, you may want to increase this.+# If 0, applications waiting for dead services+# may wait indefinitely.+# Default: 10000000+cfgRoundtripTimeout 10000000++# The maximum number of concurrent outgoing TCP+# connections (for sending messages) from this node.+# If outgoing message density is higher than this+# number, some processes will block.+# Default: 50+cfgMaxOutgoing 50++# Relates to the task layer only. The amount of+# time, in microseconds, to wait for an unused+# promise to flush to a disk file. Flushing is good,+# because it frees up memory for other data, but+# if this value is too low, heavy disk access+# may slow down your program. If 0, promises+# are never flush to disk.+# Default: 5000000+cfgPromiseFlushDelay 5000000++# Relates to the task layer only. The prefix+# of the files to be written when they+# are flushed. May contain a full path.+# Default: rpromise-+cfgPromisePrefix rpromise-++
+ remote.cabal view
@@ -0,0 +1,41 @@+Name:                remote+Version:             0.1+Cabal-Version:       >=1.8+Description:         Fault-tolerant distributed computing framework+synopsis:            Cloud Haskell+License:             BSD3+License-file:        LICENSE+Extra-Source-Files:  README.md+Author:              Jeff Epstein <jee36@cam.ac.uk>+Maintainer:          Jeff Epstein <jee36@cam.ac.uk>+Build-Type:          Simple+tested-with:         GHC ==6.12.1+Category:            Distributed Computing++extra-source-files:+    examples/kmeans/KMeans.hs+    examples/kmeans/KMeans3.hs+    examples/kmeans/KMeansCommon.hs+    examples/kmeans/MakeData.hs+    examples/kmeans/awsgo+    examples/kmeans/awskill+    examples/kmeans/awslist+    examples/kmeans/kmeans+    examples/pi/Pi6.hs+    examples/pi/Pi7.hs+    examples/pi/PiCommon.hs+    examples/tests/Test-Call.hs+    examples/tests/Test-Channel-Merge.hs+    examples/tests/Test-Channel.hs+    examples/tests/Test-Closure.hs+    examples/tests/Test-MapReduce.hs+    examples/tests/Test-Message.hs+    examples/tests/Test-Task.hs+    examples/tests/config++library+  Build-Depends:       base >= 4 && < 5, time, filepath, containers, network, syb, mtl, binary, bytestring, template-haskell, stm, pureMD5, utf8-string, directory+  ghc-options:         -Wall+  Extensions:          TemplateHaskell, FlexibleInstances, UndecidableInstances, CPP, ExistentialQuantification, DeriveDataTypeable+  Exposed-Modules:     Remote.Process, Remote.Encoding, Remote.Call, Remote.Reg, Remote.Peer, Remote.Init, Remote.Closure, Remote.Channel, Remote.Task, Remote+