distributed-process 0.2.1.3 → 0.2.1.4

raw patch · 11 files changed

+415/−355 lines, 11 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• distributed-process.cabal +1/−2
• src/Control/Distributed/Process.hs +6/−5
• src/Control/Distributed/Process/Closure.hs +125/−58
• src/Control/Distributed/Process/Internal/Closure/CP.hs +3/−3
• src/Control/Distributed/Process/Internal/Closure/TH.hs +71/−50
• src/Control/Distributed/Process/Internal/MessageT.hs +0/−130
• src/Control/Distributed/Process/Internal/Node.hs +84/−9
• src/Control/Distributed/Process/Internal/Primitives.hs +25/−21
• src/Control/Distributed/Process/Internal/Types.hs +21/−30
• src/Control/Distributed/Process/Node.hs +64/−47
• tests/TestClosure.hs +15/−0

@@ -1,5 +1,5 @@ Name:          distributed-process -Version:       0.2.1.3+Version:       0.2.1.4 Cabal-Version: >=1.8 Build-Type:    Simple License:       BSD3 @@ -52,7 +52,6 @@                      Control.Distributed.Process.Internal.CQueue,                      Control.Distributed.Process.Internal.Dynamic,                      Control.Distributed.Process.Internal.TypeRep,-                     Control.Distributed.Process.Internal.MessageT,                      Control.Distributed.Process.Internal.Types,                      Control.Distributed.Process.Internal.Closure.Static,                      Control.Distributed.Process.Internal.Closure.MkClosure,

@@ -89,7 +89,7 @@   , spawnMonitor   , spawnChannel   , DidSpawn(..)-    -- * Local versions of spawn+    -- * Local versions of 'spawn'    , spawnLocal   , spawnChannelLocal   ) where@@ -101,6 +101,8 @@  import Data.Typeable (Typeable) import Control.Monad.IO.Class (liftIO)+import Control.Applicative ((<$>))+import Control.Monad.Reader (ask) import Control.Distributed.Process.Internal.Types    ( RemoteTable   , NodeId(..)@@ -124,7 +126,7 @@   , SerializableDict(..)   , SendPortId(..)   , WhereIsReply(..)-  , procMsg+  , LocalProcess(processNode)   ) import Control.Distributed.Process.Internal.Closure.CP   ( cpSeq@@ -194,7 +196,6 @@   , spawnAsync   ) import Control.Distributed.Process.Serializable (Serializable)-import Control.Distributed.Process.Internal.MessageT (getLocalNode) import Control.Distributed.Process.Node (forkProcess)  -- INTERNAL NOTES@@ -369,7 +370,7 @@ -- | Spawn a process on the local node spawnLocal :: Process () -> Process ProcessId spawnLocal proc = do-  node <- procMsg $ getLocalNode+  node <- processNode <$> ask    liftIO $ forkProcess node proc  -- | Create a new typed channel, spawn a process on the local node, passing it@@ -378,7 +379,7 @@                   => (ReceivePort a -> Process ())                    -> Process (SendPort a) spawnChannelLocal proc = do -  node <- procMsg $ getLocalNode+  node <- processNode <$> ask    (sport, rport) <- newChan   _ <- liftIO $ forkProcess node (proc rport)   return sport

@@ -12,7 +12,7 @@ -- -- you can use a Template Haskell splice to create a static version of 'f': -- --- > $(mkStatic 'f) :: forall a1 .. an. Static T+-- > $(mkStatic 'f) :: forall a1 .. an. (Typeable a1, .., Typeable an) => Static T --  -- Every module that you write that contains calls to 'mkStatic' needs to -- have a call to 'remotable':@@ -35,29 +35,90 @@ -- >            . Mn.__remoteTable -- >            $ initRemoteTable  --+-- [Dealing with type class qualifiers]+-- +-- Although 'mkStatic' supports polymorphic types, it does not support+-- qualified types. For instance, you cannot call 'mkStatic' on+--+-- > decode :: Serializable a => ByteString -> a+--+-- Instead, you will need to reify the type class dictionary. Cloud Haskell+-- comes with a reified version of 'Serializable':+--+-- > data SerializableDict a where+-- >   SerializableDict :: Serializable a => SerializableDict a+--+-- Using the reified dictionary you can define +-- +-- > decodeDict :: SerializableDict a -> ByteString -> a+-- > decodeDict SerializableDict = decode+--+-- where 'decodeDict' is a normal (unqualified) polymorphic value and hence+-- can be passed as an argument to remotable:+--+-- > $(mkStatic 'decodeDict) :: Typeable a => Static (SerializableDict a -> ByteString -> a)+-- -- [Composing static values] ----- We generalize the notion of 'static' as described in the paper, and also--- provide+-- The version of 'static' provided by this implementation of Cloud Haskell is+-- strictly more expressive than the one proposed in the paper, and additionally+-- supports  -- -- > staticApply :: Static (a -> b) -> Static a -> Static b ----- This makes it possible to define a rich set of combinators on 'static'--- values, a number of which are provided in this module.+-- This is extremely useful. For example, Cloud Haskell comes with+-- 'staticDecode' defined as  --+-- > staticDecode :: Typeable a => Static (SerializableDict a) -> Static (ByteString -> a)+-- > staticDecode dict = $(mkStatic 'decodeDict) `staticApply` dict +--+-- 'staticDecode' is used when defining closures (see below), and makes+-- essential use of 'staticApply'. +--+-- Support for 'staticApply' also makes it possible to define a rich set of+-- combinators on 'static' values, a number of which are provided in this+-- module. +--+-- [Static serialization dictionaries]+--+-- Many Cloud Haskell primitives (like 'staticDecode', above) require static+-- serialization dictionaries. In principle these dictionaries require nothing+-- special; for instance, given some serializable type 'T' you can define +--+-- > sdictT :: SerializableDict T+-- > sdictT = SerializableDict+--+-- and then have+-- +-- > $(mkStatic 'sdictT) :: Static (SerializableDict T)+--+-- However, since these dictionaries are so frequently required Cloud Haskell+-- provides special support for them.  When you call 'remotable' on a+-- /monomorphic/ function @f :: T1 -> T2@+-- +-- > remotable ['f]+--+-- then a serialization dictionary is automatically created for you, which you+-- can access with+--+-- > $(functionSDict 'f) :: Static (SerializableDict T1)+--+-- In addition, if @f :: T1 -> Process T2@, then a second dictionary is created+--+-- > $(functionTDict 'f) :: Static (SerializableDict T2)+-- -- [Closures] -- -- Suppose you have a process ----- > factorial :: Int -> Process Int+-- > isPrime :: Integer -> Process Bool  ----- Then you can use the supplied Template Haskell function 'mkClosure' to define+-- Then  ----- > factorialClosure :: Int -> Closure (Process Int)--- > factorialClosure = $(mkClosure 'factorial)+-- > $(mkClosure 'isPrime) :: Integer -> Closure (Process Bool) ----- You can then pass 'factorialClosure n' to 'spawn', for example, to have a--- remote node compute a factorial number.+-- which you can then 'call', for example, to have a remote node check if +-- a number is prime. -- -- In general, if you have a /monomorphic/ function --@@ -76,55 +137,17 @@ --  -- > data Closure a = Closure (Static (ByteString -> a)) ByteString ----- The splice @$(mkClosure 'factorial)@ above expands to (prettified a bit): +-- The splice @$(mkClosure 'isPrime)@ above expands to (prettified a bit):  -- --- > factorialClosure :: Int -> Closure (Process Int)--- > factorialClosure n = Closure decoder (encode n)--- >   where--- >     decoder :: Static (ByteString -> Process Int)--- >     decoder = $(mkStatic 'factorial) +-- > let decoder :: Static (ByteString -> Process Bool) +-- >     decoder = $(mkStatic 'isPrime)  -- >             `staticCompose`  --- >               staticDecode $(functionSDict 'factorial)------ 'mkStatic' we have already seen:------ > $(mkStatic 'factorial) :: Static (Int -> Process Int)------ 'staticCompose' is function composition on static functions. 'staticDecode'--- has type (**)------ > staticDecode :: Typeable a --- >              => Static (SerializableDict a) -> Static (ByteString -> a)------ and gives you a static decoder, given a static Serializable dictionary.--- 'SerializableDict' is a reified type class dictionary, and defined simply as--- --- > data SerializableDict a where--- >   SerializableDict :: Serializable a => SerializableDict a------ That means that for any serialziable type 'T', you can define------ > sdictForMyType :: SerializableDict T--- > sdictForMyType = SerializableDict------ and then use------ > $(mkStatic 'sdictForMyType) :: Static (SerializableDict T)------ to obtain a static serializable dictionary for 'T' (make sure to pass--- 'sdictForMyType' to 'remotable'). --- --- However, since these serialization dictionaries are so frequently required,--- when you call 'remotable' on a monomorphic function @f : T1 -> T2@--- --- > remotable ['f]------ then a serialization dictionary is automatically created for you, which you--- can access with------ > $(functionSDict 'f) :: Static (SerializableDict T1)+-- >               staticDecode $(functionSDict 'isPrime)+-- > in Closure decoder (encode n) ----- This is the dictionary that 'mkClosure' uses.+-- where 'staticCompose' is composition of static functions. Note that+-- 'mkClosure' makes use of the static serialization dictionary +-- ('functionSDict') created by 'remotable'. -- -- [Combinators on Closures] --@@ -133,6 +156,47 @@ -- the most important of which is 'cpBind'. Have a look at the implementation -- of 'Control.Distributed.Process.call' for an example use. --+-- [Example]+--+-- Here is a small self-contained example that uses closures and serialization+-- dictionaries. It makes use of the Control.Distributed.Process.SimpleLocalnet+-- Cloud Haskell backend.+--+-- > {-# LANGUAGE TemplateHaskell #-}+-- > import System.Environment (getArgs)+-- > import Control.Distributed.Process+-- > import Control.Distributed.Process.Closure+-- > import Control.Distributed.Process.Backend.SimpleLocalnet+-- > import Control.Distributed.Process.Node (initRemoteTable)+-- > +-- > isPrime :: Integer -> Process Bool+-- > isPrime n = return . (n `elem`) . takeWhile (<= n) . sieve $ [2..]+-- >   where+-- >     sieve :: [Integer] -> [Integer]+-- >     sieve (p : xs) = p : sieve [x | x <- xs, x `mod` p > 0]+-- > +-- > remotable ['isPrime]+-- > +-- > master :: [NodeId] -> Process ()+-- > master [] = liftIO $ putStrLn "no slaves"+-- > master (slave:_) = do+-- >   isPrime79 <- call $(functionTDict 'isPrime) slave ($(mkClosure 'isPrime) (79 :: Integer))+-- >   liftIO $ print isPrime79 +-- > +-- > main :: IO ()+-- > main = do+-- >   args <- getArgs+-- >   case args of+-- >     ["master", host, port] -> do+-- >       backend <- initializeBackend host port rtable +-- >       startMaster backend master +-- >     ["slave", host, port] -> do+-- >       backend <- initializeBackend host port rtable +-- >       startSlave backend+-- >   where+-- >     rtable :: RemoteTable+-- >     rtable = __remoteTable initRemoteTable +-- -- [Notes] -- -- (*) If 'T1' is not serializable you will get a type error in the generated@@ -147,11 +211,13 @@ --      instance into scope, but unless proper 'static' support is added to --      ghc we need both the type class argument and the explicit dictionary.  module Control.Distributed.Process.Closure -  ( -- * User-defined closures+  ( -- * Creating static values     remotable   , mkStatic+    -- * Template-Haskell support for creating closures   , mkClosure   , functionSDict+  , functionTDict     -- * Primitive operations on static values   , staticApply   , staticDuplicate@@ -207,6 +273,7 @@   ( remotable   , mkStatic   , functionSDict+  , functionTDict   ) import Control.Distributed.Process.Internal.Closure.Static   ( -- Static functionals

@@ -31,6 +31,7 @@ import Data.Typeable (Typeable, typeOf, typeRepTyCon, TyCon) import Control.Applicative ((<$>)) import Control.Monad ((>=>))+import Control.Monad.Reader (ask) import Control.Distributed.Process.Serializable (Serializable) import Control.Distributed.Process.Internal.Types   ( Closure(Closure)@@ -43,7 +44,7 @@   , typeOfStaticLabel   , ProcessId   , LocalNode(remoteTable)-  , procMsg+  , LocalProcess(processNode)   , SendPort   , ReceivePort   )@@ -55,7 +56,6 @@   , newChan   ) import Control.Distributed.Process.Internal.Closure.TH (remotable, mkStatic)-import Control.Distributed.Process.Internal.MessageT (getLocalNode) import Control.Distributed.Process.Internal.Closure.Resolution (resolveClosure) import Control.Distributed.Process.Internal.Closure.Static    ( staticCompose@@ -118,7 +118,7 @@ -- | Resolve a closure unClosure :: Static a -> ByteString -> Process Dynamic unClosure (Static label) env = do-  rtable <- remoteTable <$> procMsg getLocalNode +  rtable <- remoteTable . processNode <$> ask    case resolveClosure rtable label env of     Nothing  -> fail "Derived.unClosure: resolveClosure failed"     Just dyn -> return dyn

@@ -7,6 +7,7 @@     remotable   , mkStatic   , functionSDict+  , functionTDict   ) where  import Prelude hiding (lookup)@@ -47,6 +48,7 @@   , StaticLabel(StaticLabel)   , remoteTableLabel   , SerializableDict(SerializableDict)+  , Process   ) import Control.Distributed.Process.Internal.Dynamic    ( Dynamic(..)@@ -74,18 +76,19 @@ mkStatic :: Name -> Q Exp mkStatic = varE . staticName --- | Serialization dictionary for a function argument (see module header)+-- | If @f : T1 -> T2@ is a monomorphic function +-- then @$(functionSDict 'f) :: Static (SerializableDict T1)@.+-- +-- Be sure to pass 'f' to 'remotable'. functionSDict :: Name -> Q Exp functionSDict = varE . sdictName -{---- | Create a closure--- --- See module documentation header for "Control.Distributed.Process.Closure"--- for a detailed explanation and examples.-mkClosure :: Name -> Q Exp-mkClosure = varE . closureName --}+-- | If @f : T1 -> Process T2@ is a monomorphic function+-- then @$(functionTDict 'f) :: Static (SerializableDict T2)@.+--+-- Be sure to pass 'f' to 'remotable'.+functionTDict :: Name -> Q Exp+functionTDict = varE . tdictName  -------------------------------------------------------------------------------- -- Internal (Template Haskell)                                                --@@ -100,44 +103,59 @@  generateDefs :: Name -> Q ([Dec], [Q Exp]) generateDefs n = do-  mType <- getType n-  case mType of-    Just (origName, typ) -> do-      let (typVars, typ') = case typ of ForallT vars [] mono -> (vars, mono)-                                        _                    -> ([], typ)-      (static, register) <- do-        static <- generateStatic origName typVars typ' -        let dyn = case typVars of -                    [] -> [| toDyn $(varE origName) |]-                    _  -> [| Dynamic (error "Polymorphic value") -                                     (unsafeCoerce# $(varE origName)) -                           |]-        return ( static-               , [ [| registerStatic $(stringE (show origName)) $dyn |] ]+    proc <- [t| Process |]+    mType <- getType n+    case mType of+      Just (origName, typ) -> do+        let (typVars, typ') = case typ of ForallT vars [] mono -> (vars, mono)+                                          _                    -> ([], typ)++        -- The main "static" entry                                  +        (static, register) <- makeStatic origName typVars typ' +         +        -- If n :: T1 -> T2, static serializable dictionary for T1 +        -- TODO: we should check if arg is an instance of Serializable, but we cannot+        -- http://hackage.haskell.org/trac/ghc/ticket/7066+        (sdict, registerSDict) <- case (typVars, typ') of+          ([], ArrowT `AppT` arg `AppT` _res) -> +            makeDict (sdictName origName) arg+          _ -> +            return ([], [])+        +        -- If n :: T1 -> Process T2, static serializable dictionary for T2+        -- TODO: check if T2 is serializable (same as above)+        (tdict, registerTDict) <- case (typVars, typ') of+          ([], ArrowT `AppT` _arg `AppT` (proc' `AppT` res)) | proc' == proc -> +            makeDict (tdictName origName) res +          _ ->+            return ([], [])+        +        return ( concat [static, sdict, tdict]+               , concat [register, registerSDict, registerTDict]                )+      _ -> +        fail $ "remotable: " ++ show n ++ " not found"+  where+    makeStatic :: Name -> [TyVarBndr] -> Type -> Q ([Dec], [Q Exp])+    makeStatic origName typVars typ = do +      static <- generateStatic origName typVars typ+      let dyn = case typVars of +                  [] -> [| toDyn $(varE origName) |]+                  _  -> [| Dynamic (error "Polymorphic value") +                                   (unsafeCoerce# $(varE origName)) +                         |]+      return ( static+             , [ [| registerStatic $(stringE (show origName)) $dyn |] ]+             ) -      (sdict, registerSDict) <- case (typVars, typ') of-        ([], ArrowT `AppT` arg `AppT` _res) -> do-          -- TODO: we should check if arg is an instance of Serializable, but we cannot-          -- http://hackage.haskell.org/trac/ghc/ticket/7066-          sdict <- generateSDict origName arg-          let dyn' = [| toDyn (SerializableDict -                                 :: SerializableDict $(return arg)-                              ) -                      |]-          return ( sdict-                 , [ [| registerStatic $(stringE (show (sdictName origName))) -                                       $dyn' -                      |] ]-                 )-        _ ->-          return ([], [])-      -      return ( static ++ sdict-             , register ++ registerSDict+    makeDict :: Name -> Type -> Q ([Dec], [Q Exp]) +    makeDict dictName typ = do+      sdict <- generateDict dictName typ +      let dyn = [| toDyn (SerializableDict :: SerializableDict $(return typ)) |]+      return ( sdict+             , [ [| registerStatic $(stringE (show dictName)) $dyn |] ]               )-    _ -> -      fail $ "remotable: " ++ show n ++ " not found"+        registerStatic :: String -> Dynamic -> RemoteTable -> RemoteTable registerStatic label dyn = remoteTableLabel label ^= Just dyn @@ -163,14 +181,14 @@     typeable (PlainTV v)    = ClassP (mkName "Typeable") [VarT v]      typeable (KindedTV v _) = ClassP (mkName "Typeable") [VarT v] --- | Generate a function serialization dictionary-generateSDict :: Name -> Type -> Q [Dec]-generateSDict n typ = do+-- | Generate a serialization dictionary with name 'n' for type 'typ' +generateDict :: Name -> Type -> Q [Dec]+generateDict n typ = do     sequence-      [ sigD (sdictName n) $ [t| Static (SerializableDict $(return typ)) |]-      , sfnD (sdictName n) +      [ sigD n $ [t| Static (SerializableDict $(return typ)) |]+      , sfnD n           [| Static $ StaticLabel -              $(stringE (show (sdictName n))) +              $(stringE (show n))                (typeOf (undefined :: SerializableDict $(return typ)))           |]       ]@@ -180,6 +198,9 @@  sdictName :: Name -> Name sdictName n = mkName $ nameBase n ++ "__sdict"++tdictName :: Name -> Name+tdictName n = mkName $ nameBase n ++ "__tdict"  -------------------------------------------------------------------------------- -- Generic Template Haskell auxiliary functions                               --

@@ -1,130 +0,0 @@--- | Add message sending capability to a monad--- --- NOTE: Not thread-safe (you should not do concurrent sends within the same--- monad).-module Control.Distributed.Process.Internal.MessageT -  ( MessageT-  , runMessageT-  , getLocalNode-  , sendPayload-  , sendBinary-  , sendMessage-  , payloadToMessage-  , createMessage-  ) where--import Data.Binary (Binary, encode)-import qualified Data.ByteString as BSS (ByteString)-import qualified Data.ByteString.Lazy as BSL (toChunks)-import Data.Map (Map)-import qualified Data.Map as Map (empty)-import Data.Accessor (Accessor, accessor, (^=), (^.))-import qualified Data.Accessor.Container as DAC (mapMaybe)-import Control.Category ((>>>))-import Control.Monad (unless, liftM)-import Control.Monad.State (gets, modify, evalStateT)-import Control.Monad.IO.Class (MonadIO, liftIO)-import Control.Concurrent.Chan (writeChan)-import Control.Distributed.Process.Internal.Types-  ( Identifier(..)-  , nodeOf-  , NodeId(nodeAddress) -  , LocalNode(localCtrlChan, localEndPoint)-  , NCMsg(NCMsg, ctrlMsgSender, ctrlMsgSignal)-  , DiedReason(DiedDisconnect)-  , ProcessSignal(Died)-  , MessageT(..)-  , MessageState(..)-  , createMessage-  , messageToPayload-  , payloadToMessage-  )-import Control.Distributed.Process.Serializable (Serializable)  -import qualified Network.Transport as NT -  ( EndPoint -  , Connection-  , connect-  , send-  , Reliability(ReliableOrdered)-  , defaultConnectHints-  )------------------------------------------------------------------------------------- API                                                                        -------------------------------------------------------------------------------------runMessageT :: Monad m => LocalNode -> MessageT m a -> m a-runMessageT localNode m = -  evalStateT (unMessageT m) $ initMessageState localNode --getLocalNode :: Monad m => MessageT m LocalNode-getLocalNode = gets messageLocalNode--sendPayload :: MonadIO m => Identifier -> [BSS.ByteString] -> MessageT m ()-sendPayload them payload = do-  mConn <- connTo them-  didSend <- case mConn of-    Just conn -> do-      didSend <- liftIO $ NT.send conn payload-      case didSend of-        Left _  -> return False-        Right _ -> return True -    Nothing -> return False-  unless didSend $ do-    node <- getLocalNode-    liftIO . writeChan (localCtrlChan node) $ NCMsg-      { ctrlMsgSender = them-      , ctrlMsgSignal = Died them DiedDisconnect-      }--sendBinary :: (MonadIO m, Binary a) => Identifier -> a -> MessageT m ()-sendBinary them = sendPayload them . BSL.toChunks . encode--sendMessage :: (MonadIO m, Serializable a) => Identifier -> a -> MessageT m ()-sendMessage them = sendPayload them . messageToPayload . createMessage------------------------------------------------------------------------------------- Internal                                                                   -------------------------------------------------------------------------------------initMessageState :: LocalNode -> MessageState-initMessageState localNode = MessageState {-     messageLocalNode   = localNode -  , _messageConnections = Map.empty-  }--setupConnTo :: MonadIO m => Identifier -> MessageT m (Maybe NT.Connection)-setupConnTo them = do-    endPoint <- localEndPoint `liftM` getLocalNode -    mConn    <- liftIO $ NT.connect endPoint -                                    (nodeAddress . nodeOf $ them) -                                    NT.ReliableOrdered -                                    NT.defaultConnectHints-    case mConn of -      Right conn -> do-        didSend <- liftIO $ NT.send conn (BSL.toChunks . encode $ them)-        case didSend of-          Left _ ->-            return Nothing-          Right () -> do-            modify $ messageConnectionTo them ^= Just conn-            return $ Just conn-      Left _ ->-        return Nothing--connTo :: MonadIO m => Identifier -> MessageT m (Maybe NT.Connection)-connTo them = do-  mConn <- gets (^. messageConnectionTo them)-  case mConn of-    Just conn -> return $ Just conn-    Nothing   -> setupConnTo them------------------------------------------------------------------------------------- Accessors                                                                  -------------------------------------------------------------------------------------messageConnections :: Accessor MessageState (Map Identifier NT.Connection)-messageConnections = accessor _messageConnections (\conns st -> st { _messageConnections = conns })--messageConnectionTo :: Identifier -> Accessor MessageState (Maybe NT.Connection)-messageConnectionTo them = messageConnections >>> DAC.mapMaybe them

@@ -1,16 +1,91 @@ module Control.Distributed.Process.Internal.Node -  ( runLocalProcess+  ( -- * Message sending+    sendPayload+  , sendBinary+  , sendMessage   ) where -import Control.Monad.Reader (runReaderT)+import Data.Accessor ((^.), (^=))+import Data.Binary (Binary, encode)+import qualified Data.ByteString.Lazy as BSL (toChunks)+import qualified Data.ByteString as BSS (ByteString)+import Control.Concurrent.MVar (withMVar, modifyMVar_)+import Control.Concurrent.Chan (writeChan)+import Control.Monad (unless)+import qualified Network.Transport as NT +  ( Connection+  , send+  , defaultConnectHints+  , connect+  , Reliability(ReliableOrdered)+  ) import Control.Distributed.Process.Internal.Types -  ( LocalNode-  , LocalProcess-  , Process(unProcess)+  ( LocalNode(localState, localEndPoint, localCtrlChan)+  , Identifier+  , localConnectionBetween+  , nodeAddress+  , nodeOf+  , messageToPayload+  , createMessage+  , NCMsg(..)+  , ProcessSignal(Died)+  , DiedReason(DiedDisconnect)   )-import Control.Distributed.Process.Internal.MessageT (runMessageT)  +import Control.Distributed.Process.Serializable (Serializable) --- | Deconstructor for 'Process' (not exported to the public API) -runLocalProcess :: LocalNode -> Process a -> LocalProcess -> IO a-runLocalProcess node proc = runMessageT node . runReaderT (unProcess proc) +--------------------------------------------------------------------------------+-- Message sending                                                            -- +-------------------------------------------------------------------------------- +sendPayload :: LocalNode -> Identifier -> Identifier -> [BSS.ByteString] -> IO ()+sendPayload node from to payload = do+  mConn <- connBetween node from to+  didSend <- case mConn of+    Just conn -> do+      didSend <- NT.send conn payload+      case didSend of+        Left _  -> return False+        Right _ -> return True +    Nothing -> return False+  unless didSend $+    writeChan (localCtrlChan node) NCMsg+      { ctrlMsgSender = to +      , ctrlMsgSignal = Died to DiedDisconnect+      }++sendBinary :: Binary a => LocalNode -> Identifier -> Identifier -> a -> IO ()+sendBinary node from to = sendPayload node from to . BSL.toChunks . encode++sendMessage :: Serializable a => LocalNode -> Identifier -> Identifier -> a -> IO ()+sendMessage node from to = sendPayload node from to . messageToPayload . createMessage++setupConnBetween :: LocalNode -> Identifier -> Identifier -> IO (Maybe NT.Connection)+setupConnBetween node from to = do+    mConn    <- NT.connect endPoint +                           (nodeAddress . nodeOf $ to) +                           NT.ReliableOrdered +                           NT.defaultConnectHints+    case mConn of +      Right conn -> do+        didSend <- NT.send conn (BSL.toChunks . encode $ to)+        case didSend of+          Left _ ->+            return Nothing+          Right () -> do+            modifyMVar_ nodeState $ return . +              (localConnectionBetween from to ^= Just conn)+            return $ Just conn+      Left _ ->+        return Nothing+  where+    endPoint  = localEndPoint node+    nodeState = localState node++connBetween :: LocalNode -> Identifier -> Identifier -> IO (Maybe NT.Connection)+connBetween node from to = do+    mConn <- withMVar nodeState $ return . (^. localConnectionBetween from to)+    case mConn of+      Just conn -> return $ Just conn+      Nothing   -> setupConnBetween node from to +  where+    nodeState = localState node

@@ -68,7 +68,7 @@ import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Applicative ((<$>)) import Control.Exception (Exception, throw)-import qualified Control.Exception as Exception (catch)+import qualified Control.Exception as Ex (catch) import Control.Concurrent.MVar (modifyMVar) import Control.Concurrent.Chan (writeChan) import Control.Concurrent.STM @@ -106,7 +106,6 @@   , TypedChannel(..)   , SendPortId(..)   , Identifier(..)-  , procMsg   , DidUnmonitor(..)   , DidUnlinkProcess(..)   , DidUnlinkNode(..)@@ -114,13 +113,9 @@   , WhereIsReply(..)   , createMessage   , Static(..)+  , runLocalProcess   )-import Control.Distributed.Process.Internal.MessageT -  ( sendMessage-  , sendBinary-  , getLocalNode-  )  -import Control.Distributed.Process.Internal.Node (runLocalProcess)+import Control.Distributed.Process.Internal.Node (sendMessage, sendBinary)  import Control.Distributed.Process.Internal.Closure.Resolution (resolveClosure) import Control.Distributed.Process.Internal.Dynamic (fromDyn, dynTypeRep) @@ -132,7 +127,12 @@ send :: Serializable a => ProcessId -> a -> Process () -- This requires a lookup on every send. If we want to avoid that we need to -- modify serializable to allow for stateful (IO) deserialization-send them msg = procMsg $ sendMessage (ProcessIdentifier them) msg +send them msg = do+  proc <- ask +  liftIO $ sendMessage (processNode proc) +                       (ProcessIdentifier (processId proc)) +                       (ProcessIdentifier them)+                       msg  -- | Wait for a message of a specific type expect :: forall a. Serializable a => Process a@@ -163,7 +163,12 @@  -- | Send a message on a typed channel sendChan :: Serializable a => SendPort a -> a -> Process ()-sendChan (SendPort cid) msg = procMsg $ sendBinary (SendPortIdentifier cid) msg +sendChan (SendPort cid) msg = do+  proc <- ask+  liftIO $ sendBinary (processNode proc)+                      (ProcessIdentifier (processId proc))+                      (SendPortIdentifier cid) +                      msg   -- | Wait for a message on a typed channel receiveChan :: Serializable a => ReceivePort a -> Process a@@ -261,7 +266,7 @@  -- | Get the node ID of our local node getSelfNode :: Process NodeId-getSelfNode = localNodeId <$> procMsg getLocalNode+getSelfNode = localNodeId . processNode <$> ask   -------------------------------------------------------------------------------- -- Monitoring and linking                                                     --@@ -318,11 +323,8 @@ -- | Catch exceptions within a process catch :: Exception e => Process a -> (e -> Process a) -> Process a catch p h = do-  node  <- procMsg getLocalNode   lproc <- ask-  let run :: Process a -> IO a-      run proc = runLocalProcess node proc lproc -  liftIO $ Exception.catch (run p) (run . h) +  liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)   -- | Like 'expect' but with a timeout expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)@@ -460,7 +462,7 @@ -- | Deserialize a closure unClosure :: forall a. Typeable a => Closure a -> Process a unClosure (Closure (Static label) env) = do-    rtable <- remoteTable <$> procMsg getLocalNode +    rtable <- remoteTable . processNode <$> ask      case resolveClosure rtable label env of       Nothing  -> throw . userError $ "Unregistered closure " ++ show label       Just dyn -> return $ fromDyn dyn (throw (typeError dyn))@@ -507,14 +509,16 @@             -> ProcessSignal -- ^ Message to send              -> Process () sendCtrlMsg mNid signal = do            -  us <- getSelfPid-  let msg = NCMsg { ctrlMsgSender = ProcessIdentifier us+  proc <- ask+  let msg = NCMsg { ctrlMsgSender = ProcessIdentifier (processId proc)                    , ctrlMsgSignal = signal                   }   case mNid of     Nothing -> do-      ctrlChan <- localCtrlChan <$> procMsg getLocalNode +      ctrlChan <- localCtrlChan . processNode <$> ask        liftIO $ writeChan ctrlChan msg      Just nid ->-      procMsg $ sendBinary (NodeIdentifier nid) msg-+      liftIO $ sendBinary (processNode proc)+                          (ProcessIdentifier (processId proc))+                          (NodeIdentifier nid)+                          msg

@@ -17,7 +17,7 @@   , LocalProcess(..)   , LocalProcessState(..)   , Process(..)-  , procMsg+  , runLocalProcess     -- * Typed channels   , LocalSendPortId    , SendPortId(..)@@ -58,14 +58,13 @@     -- * Node controller internal data types    , NCMsg(..)   , ProcessSignal(..)-    -- * MessageT monad-  , MessageT(..)-  , MessageState(..)     -- * Accessors   , localProcesses   , localPidCounter   , localPidUnique+  , localConnections   , localProcessWithId+  , localConnectionBetween   , monitorCounter   , spawnCounter   , channelCounter@@ -97,10 +96,8 @@ import Control.Concurrent.STM (TChan, TVar) import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection) import Control.Applicative (Applicative, (<$>), (<*>))-import Control.Monad.Reader (MonadReader(..), ReaderT)+import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT) import Control.Monad.IO.Class (MonadIO)-import Control.Monad.State (MonadState, StateT)-import qualified Control.Monad.Trans.Class as Trans (lift) import Control.Distributed.Process.Serializable    ( Fingerprint   , Serializable@@ -131,10 +128,7 @@   { lpidUnique  :: Int32   , lpidCounter :: Int32   }-  deriving (Eq, Ord, Typeable)--instance Show LocalProcessId where-  show = show . lpidCounter +  deriving (Eq, Ord, Typeable, Show)  -- | Process identifier data ProcessId = ProcessId @@ -187,9 +181,10 @@  -- | Local node state data LocalNodeState = LocalNodeState -  { _localProcesses  :: Map LocalProcessId LocalProcess-  , _localPidCounter :: Int32-  , _localPidUnique  :: Int32+  { _localProcesses   :: Map LocalProcessId LocalProcess+  , _localPidCounter  :: Int32+  , _localPidUnique   :: Int32+  , _localConnections :: Map (Identifier, Identifier) NT.Connection   }  -- | Processes running on our local node@@ -198,8 +193,13 @@   , processId     :: ProcessId   , processState  :: MVar LocalProcessState   , processThread :: ThreadId+  , processNode   :: LocalNode   } +-- | Deconstructor for 'Process' (not exported to the public API) +runLocalProcess :: LocalProcess -> Process a -> IO a+runLocalProcess lproc proc = runReaderT (unProcess proc) lproc+ -- | Local process state data LocalProcessState = LocalProcessState   { _monitorCounter :: Int32@@ -210,13 +210,10 @@  -- | The Cloud Haskell 'Process' type newtype Process a = Process { -    unProcess :: ReaderT LocalProcess (MessageT IO) a +    unProcess :: ReaderT LocalProcess IO a    }   deriving (Functor, Monad, MonadIO, MonadReader LocalProcess, Typeable, Applicative) -procMsg :: MessageT IO a -> Process a-procMsg = Process . Trans.lift - -------------------------------------------------------------------------------- -- Typed channels                                                             -- --------------------------------------------------------------------------------@@ -574,9 +571,15 @@ localPidUnique :: Accessor LocalNodeState Int32 localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq }) +localConnections :: Accessor LocalNodeState (Map (Identifier, Identifier) NT.Connection)+localConnections = accessor _localConnections (\conns st -> st { _localConnections = conns })+ localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess) localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid +localConnectionBetween :: Identifier -> Identifier -> Accessor LocalNodeState (Maybe NT.Connection)+localConnectionBetween from to = localConnections >>> DAC.mapMaybe (from, to)+ monitorCounter :: Accessor LocalProcessState Int32 monitorCounter = accessor _monitorCounter (\cnt st -> st { _monitorCounter = cnt }) @@ -597,15 +600,3 @@  remoteTableLabel :: String -> Accessor RemoteTable (Maybe Dynamic) remoteTableLabel label = remoteTableLabels >>> DAC.mapMaybe label------------------------------------------------------------------------------------- MessageT monad                                                             -------------------------------------------------------------------------------------newtype MessageT m a = MessageT { unMessageT :: StateT MessageState m a }-  deriving (Functor, Monad, MonadIO, MonadState MessageState, Applicative)--data MessageState = MessageState { -     messageLocalNode   :: LocalNode-  , _messageConnections :: Map Identifier NT.Connection -  }

@@ -38,7 +38,7 @@ import Control.Monad (void, when, forever) import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Monad.State (MonadState, StateT, evalStateT, gets, modify)-import qualified Control.Monad.Trans.Class as Trans (lift)+import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask) import Control.Exception (throwIO, SomeException, Exception, throwTo) import qualified Control.Exception as Exception (catch) import Control.Concurrent (forkIO)@@ -101,7 +101,6 @@   , Closure(..)   , Static(..)   , Message-  , MessageT   , TypedChannel(..)   , Identifier(..)   , nodeOf@@ -110,20 +109,18 @@   , WhereIsReply(..)   , messageToPayload   , RemoteTable(..)+  , payloadToMessage+  , createMessage+  , runLocalProcess   ) import Control.Distributed.Process.Serializable (Serializable)-import Control.Distributed.Process.Internal.MessageT +import Control.Distributed.Process.Internal.Dynamic (fromDynamic)+import Control.Distributed.Process.Internal.Closure.Resolution (resolveClosure) +import Control.Distributed.Process.Internal.Node    ( sendBinary   , sendMessage   , sendPayload-  , getLocalNode-  , runMessageT-  , payloadToMessage-  , createMessage   )-import Control.Distributed.Process.Internal.Dynamic (fromDynamic)-import Control.Distributed.Process.Internal.Closure.Resolution (resolveClosure) -import Control.Distributed.Process.Internal.Node (runLocalProcess) import Control.Distributed.Process.Internal.Primitives (expect, register) import qualified Control.Distributed.Process.Internal.Closure.Static as Static (__remoteTable) import qualified Control.Distributed.Process.Internal.Closure.CP as CP (__remoteTable)@@ -156,9 +153,10 @@ createBareLocalNode endPoint rtable = do   unq <- randomIO   state <- newMVar LocalNodeState -    { _localProcesses      = Map.empty-    , _localPidCounter     = 0 -    , _localPidUnique      = unq +    { _localProcesses   = Map.empty+    , _localPidCounter  = 0 +    , _localPidUnique   = unq +    , _localConnections = Map.empty     }   ctrlChan <- newChan   let node = LocalNode { localNodeId   = NodeId $ NT.address endPoint@@ -214,10 +212,11 @@                              , processId     = pid                              , processState  = pst                               , processThread = tid+                             , processNode   = node                              }     tid' <- forkIO $ do       reason <- Exception.catch -        (runLocalProcess node proc lproc >> return DiedNormal)+        (runLocalProcess lproc proc >> return DiedNormal)         (return . DiedException . (show :: SomeException -> String))       -- [Unified: Table 4, rules termination and exiting]       modifyMVar_ (localState node) $ @@ -287,7 +286,7 @@                          (Map.insert cid proc procs)                          chans                          ctrls-                    Nothing ->+                    Nothing ->                        -- Request for an unknown process.                        --                       -- TODO: We should treat this as a fatal error on the@@ -367,8 +366,8 @@ --------------------------------------------------------------------------------  runNodeController :: LocalNode -> IO ()-runNodeController node =-  runMessageT node (evalStateT (unNC nodeController) initNCState)+runNodeController =+  runReaderT (evalStateT (unNC nodeController) initNCState)  -------------------------------------------------------------------------------- -- Internal data types                                                        --@@ -383,11 +382,8 @@   , _registry :: Map String ProcessId   } -newtype NC a = NC { unNC :: StateT NCState (MessageT IO) a }-  deriving (Functor, Monad, MonadIO, MonadState NCState)--ncMsg :: MessageT IO a -> NC a-ncMsg = NC . Trans.lift+newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }+  deriving (Functor, Monad, MonadIO, MonadState NCState, MonadReader LocalNode)  initNCState :: NCState initNCState = NCState { _links    = Map.empty@@ -402,14 +398,17 @@ -- [Unified: Table 7] nodeController :: NC () nodeController = do-  node <- ncMsg getLocalNode +  node <- ask    forever $ do     msg  <- liftIO $ readChan (localCtrlChan node)      -- [Unified: Table 7, rule nc_forward]      case destNid (ctrlMsgSignal msg) of       Just nid' | nid' /= localNodeId node -> -        ncMsg $ sendBinary (NodeIdentifier nid') msg+        liftIO $ sendBinary node+                            (ctrlMsgSender msg)+                            (NodeIdentifier nid') +                            msg       _ ->          return () @@ -441,7 +440,7 @@                 -> Maybe MonitorRef -- ^ 'Nothing' to link                 -> NC () ncEffectMonitor from them mRef = do-  node <- ncMsg getLocalNode +  node <- ask    shouldLink <-      if not (isLocal node them)        then return True@@ -454,15 +453,21 @@     (False, True) -> -- [Unified: second rule]       notifyDied from them DiedUnknownId mRef      (False, False) -> -- [Unified: third rule]-      ncMsg $ sendBinary (NodeIdentifier $ processNodeId from) NCMsg -        { ctrlMsgSender = NodeIdentifier (localNodeId node)-        , ctrlMsgSignal = Died them DiedUnknownId-        }+      -- TODO: this is the right sender according to the Unified semantics,+      -- but perhaps having 'them' as the sender would make more sense+      -- (see also: notifyDied)+      liftIO $ sendBinary node+                          (NodeIdentifier $ localNodeId node)+                          (NodeIdentifier $ processNodeId from)+        NCMsg  +          { ctrlMsgSender = NodeIdentifier (localNodeId node)+          , ctrlMsgSignal = Died them DiedUnknownId+          }  -- [Unified: Table 11] ncEffectUnlink :: ProcessId -> Identifier -> NC () ncEffectUnlink from them = do-  node <- ncMsg getLocalNode +  node <- ask    when (isLocal node (ProcessIdentifier from)) $      case them of       ProcessIdentifier pid -> @@ -476,7 +481,7 @@ -- [Unified: Table 11] ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC () ncEffectUnmonitor from ref = do-  node <- ncMsg getLocalNode +  node <- ask    when (isLocal node (ProcessIdentifier from)) $      postAsMessage from $ DidUnmonitor ref   modify $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)@@ -484,7 +489,7 @@ -- [Unified: Table 12] ncEffectDied :: Identifier -> DiedReason -> NC () ncEffectDied ident reason = do-  node <- ncMsg getLocalNode+  node <- ask   (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))   (affectedMons,  unaffectedMons)  <- gets (splitNotif ident . (^. monitors)) @@ -509,9 +514,12 @@   -- If the closure does not exist, we spawn a process that throws an exception   -- This allows the remote node to find out what's happening   let proc = fromMaybe (fail $ "Error: unknown closure " ++ show cProc) mProc-  node <- ncMsg getLocalNode+  node <- ask   pid' <- liftIO $ forkProcess node proc-  ncMsg $ sendMessage (ProcessIdentifier pid) (DidSpawn ref pid') +  liftIO $ sendMessage node+                       (NodeIdentifier (localNodeId node))+                       (ProcessIdentifier pid) +                       (DidSpawn ref pid')   -- Unified semantics does not explicitly describe how to implement 'register', -- but mentions it's "very similar to nsend" (Table 14)@@ -525,18 +533,24 @@ -- Unified semantics does not explicitly describe 'whereis' ncEffectWhereIs :: ProcessId -> String -> NC () ncEffectWhereIs from label = do+  node <- ask   mPid <- gets (^. registryFor label)-  ncMsg $ sendMessage (ProcessIdentifier from) (WhereIsReply label mPid)+  liftIO $ sendMessage node+                       (NodeIdentifier (localNodeId node))+                       (ProcessIdentifier from) +                       (WhereIsReply label mPid)  -- [Unified: Table 14] ncEffectNamedSend :: Identifier -> String -> Message -> NC ()-ncEffectNamedSend _from label msg = do+ncEffectNamedSend from label msg = do   mPid <- gets (^. registryFor label)   -- If mPid is Nothing, we just ignore the named send (as per Table 14)-  -- TODO: messages don't carry a "from", but if they do, we should set it-  -- to be the 'from' of the original named send, not this node controller+  node <- ask   forM_ mPid $ \pid -> -    ncMsg $ sendPayload (ProcessIdentifier pid) (messageToPayload msg) +    liftIO $ sendPayload node+                         from+                         (ProcessIdentifier pid) +                         (messageToPayload msg)   -------------------------------------------------------------------------------- -- Auxiliary                                                                  --@@ -548,7 +562,7 @@            -> Maybe MonitorRef  -- ^ 'Nothing' for linking            -> NC () notifyDied dest src reason mRef = do-  node <- ncMsg getLocalNode +  node <- ask    case (isLocal node (ProcessIdentifier dest), mRef, src) of     (True, Just ref, ProcessIdentifier pid) ->       postAsMessage dest $ ProcessMonitorNotification ref pid reason @@ -565,10 +579,13 @@     (False, _, _) ->       -- TODO: why the change in sender? How does that affect 'reconnect' semantics?       -- (see [Unified: Table 10]-      ncMsg $ sendBinary (NodeIdentifier $ processNodeId dest) NCMsg-        { ctrlMsgSender = NodeIdentifier (localNodeId node) -        , ctrlMsgSignal = Died src reason-        }+      liftIO $ sendBinary node+                          (NodeIdentifier $ localNodeId node)+                          (NodeIdentifier $ processNodeId dest)+        NCMsg+          { ctrlMsgSender = NodeIdentifier (localNodeId node) +          , ctrlMsgSignal = Died src reason+          }        -- | [Unified: Table 8] destNid :: ProcessSignal -> Maybe NodeId@@ -592,13 +609,13 @@ -- | Lookup a local closure  unClosure :: Typeable a => Closure a -> NC (Maybe a) unClosure (Closure (Static label) env) = do-  rtable <- remoteTable <$> ncMsg getLocalNode+  rtable <- remoteTable <$> ask   return (resolveClosure rtable label env >>= fromDynamic)  -- | Check if an identifier refers to a valid local object isValidLocalIdentifier :: Identifier -> NC Bool isValidLocalIdentifier ident = do-  node <- ncMsg getLocalNode+  node <- ask   liftIO . withMVar (localState node) $ \nSt ->      case ident of       NodeIdentifier nid ->@@ -631,7 +648,7 @@  withLocalProc :: ProcessId -> (LocalProcess -> IO ()) -> NC ()  withLocalProc pid p = do-  node <- ncMsg getLocalNode +  node <- ask    liftIO $ do      -- By [Unified: table 6, rule missing_process] messages to dead processes     -- can silently be dropped

@@ -36,6 +36,12 @@ wait :: Int -> Process () wait = liftIO . threadDelay +isPrime :: Integer -> Process Bool+isPrime n = return . (n `elem`) . takeWhile (<= n) . sieve $ [2..]+  where+    sieve :: [Integer] -> [Integer]+    sieve (p : xs) = p : sieve [x | x <- xs, x `mod` p > 0]+ -- | First argument indicates empty closure environment typedPingServer :: () -> ReceivePort (SendPort ()) -> Process () typedPingServer () rport = forever $ do@@ -49,6 +55,7 @@           , 'sdictInt           , 'wait           , 'typedPingServer+          , 'isPrime           ]  factorialClosure :: Int -> Closure (Process Int)@@ -296,6 +303,13 @@     sendChan pingServer sendReply     receiveChan receiveReply +testTDict :: Transport -> RemoteTable -> IO ()+testTDict transport rtable = do+  [node1, node2] <- replicateM 2 $ newLocalNode transport rtable+  runProcess node1 $ do+    True <- call $(functionTDict 'isPrime) (localNodeId node2) ($(mkClosure 'isPrime) (79 :: Integer))+    return ()+ main :: IO () main = do   Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters@@ -314,4 +328,5 @@     , ("SpawnInvalid",    testSpawnInvalid    transport rtable)     , ("ClosureExpect",   testClosureExpect   transport rtable)     , ("SpawnChannel",    testSpawnChannel    transport rtable)+    , ("TDict",           testTDict           transport rtable)     ]