distributed-process 0.2.0.1 → 0.2.1

raw patch · 16 files changed

+1159/−732 lines, 16 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• distributed-process.cabal +5/−4
• src/Control/Distributed/Process.hs +42/−32
• src/Control/Distributed/Process/Closure.hs +205/−95
• src/Control/Distributed/Process/Internal/Closure.hs +0/−133
• src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs +0/−77
• src/Control/Distributed/Process/Internal/Closure/CP.hs +336/−0
• src/Control/Distributed/Process/Internal/Closure/Combinators.hs +0/−182
• src/Control/Distributed/Process/Internal/Closure/MkClosure.hs +25/−0
• src/Control/Distributed/Process/Internal/Closure/Resolution.hs +34/−0
• src/Control/Distributed/Process/Internal/Closure/Static.hs +190/−0
• src/Control/Distributed/Process/Internal/Closure/TH.hs +104/−63
• src/Control/Distributed/Process/Internal/Dynamic.hs +56/−1
• src/Control/Distributed/Process/Internal/Primitives.hs +22/−1
• src/Control/Distributed/Process/Internal/Types.hs +42/−82
• src/Control/Distributed/Process/Node.hs +9/−4
• tests/TestClosure.hs +89/−58

@@ -1,5 +1,5 @@ Name:          distributed-process -Version:       0.2.0.1+Version:       0.2.1 Cabal-Version: >=1.8 Build-Type:    Simple License:       BSD3 @@ -51,13 +51,14 @@                      Control.Distributed.Process.Internal.Primitives,                      Control.Distributed.Process.Internal.CQueue,                      Control.Distributed.Process.Internal.Dynamic,-                     Control.Distributed.Process.Internal.Closure,                      Control.Distributed.Process.Internal.TypeRep,                      Control.Distributed.Process.Internal.MessageT,                      Control.Distributed.Process.Internal.Types,-                     Control.Distributed.Process.Internal.Closure.BuiltIn,-                     Control.Distributed.Process.Internal.Closure.Combinators,+                     Control.Distributed.Process.Internal.Closure.Static,+                     Control.Distributed.Process.Internal.Closure.MkClosure,+                     Control.Distributed.Process.Internal.Closure.CP,                      Control.Distributed.Process.Internal.Closure.TH,+                     Control.Distributed.Process.Internal.Closure.Resolution,                      Control.Distributed.Process.Internal.Node   Extensions:        RankNTypes,                      ScopedTypeVariables,

@@ -87,16 +87,13 @@   , spawnSupervised   , spawnLink   , spawnMonitor+  , spawnChannel   , DidSpawn(..)   ) where  import Prelude hiding (catch)-import Data.Typeable (Typeable, typeOf)-import Control.Applicative ((<$>))-import Control.Exception (throw)+import Data.Typeable (Typeable) import Control.Monad.IO.Class (liftIO)-import Control.Distributed.Process.Internal.MessageT (getLocalNode)-import Control.Distributed.Process.Internal.Dynamic (fromDyn, dynTypeRep) import Control.Distributed.Process.Internal.Types    ( RemoteTable   , NodeId(..)@@ -118,19 +115,24 @@   , SendPort(..)   , ReceivePort(..)   , SerializableDict(..)-  , procMsg-  , LocalNode(..)   , SendPortId(..)   , WhereIsReply(..)   )-import Control.Distributed.Process.Internal.Closure.BuiltIn -  ( linkClosure-  , unlinkClosure-  , sendClosure-  , expectClosure-  , serializableDictUnit+import Control.Distributed.Process.Internal.Closure.CP+  ( cpSeq+  , cpBind+  , cpSend +  , cpExpect+  , cpLink+  , cpUnlink+  , cpNewChan+  , cpCancelL+  , cpSplit   )-import Control.Distributed.Process.Internal.Closure.Combinators (cpSeq, cpBind)+import Control.Distributed.Process.Internal.Closure.Static +  ( sdictUnit+  , sdictSendPort+  ) import Control.Distributed.Process.Internal.Primitives   ( -- Basic messaging     send @@ -176,12 +178,14 @@   , whereisRemote   , whereisRemoteAsync   , nsendRemote+    -- Closures+  , unClosure     -- Auxiliary API   , catch   , expectTimeout   , spawnAsync   )-import Control.Distributed.Process.Internal.Closure (resolveClosure)+import Control.Distributed.Process.Serializable (Serializable)  -- INTERNAL NOTES -- @@ -256,9 +260,9 @@   -- we call linkNode here, and unlinkNode after, then we might remove a link   -- that was already set up   mRef <- monitorNode nid-  sRef <- spawnAsync nid $ linkClosure us -                   `cpSeq` expectClosure serializableDictUnit-                   `cpSeq` unlinkClosure us+  sRef <- spawnAsync nid $ cpLink us +                   `cpSeq` cpExpect sdictUnit +                   `cpSeq` cpUnlink us                    `cpSeq` proc   mPid <- receiveWait      [ matchIf (\(DidSpawn ref _) -> ref == sRef)@@ -293,10 +297,10 @@ --  -- We monitor the remote process; if it dies before it can send a reply, we die -- too-call :: SerializableDict a -> NodeId -> Closure (Process a) -> Process a-call sdict@SerializableDict nid proc = do +call :: Serializable a => Static (SerializableDict a) -> NodeId -> Closure (Process a) -> Process a+call dict nid proc = do    us <- getSelfPid-  (_, mRef) <- spawnMonitor nid (proc `cpBind` sendClosure sdict us)+  (_, mRef) <- spawnMonitor nid (proc `cpBind` cpSend dict us)   -- We are guaranteed to receive the reply before the monitor notification   -- (if a reply is sent at all)   -- NOTE: This might not be true if we switch to unreliable delivery.@@ -316,18 +320,24 @@                 -> Process (ProcessId, MonitorRef) spawnSupervised nid proc = do   us   <- getSelfPid-  them <- spawn nid (linkClosure us `cpSeq` proc) +  them <- spawn nid (cpLink us `cpSeq` proc)    ref  <- monitor them   return (them, ref) --- | Deserialize a closure-unClosure :: forall a. Typeable a => Closure a -> Process a-unClosure (Closure (Static label) env) = do-    rtable <- remoteTable <$> procMsg getLocalNode -    case resolveClosure rtable label env of-      Nothing  -> throw . userError $ "Unregistered closure " ++ show label-      Just dyn -> return $ fromDyn dyn (throw (typeError dyn))+-- | Spawn a new process, supplying it with a new 'ReceivePort' and return+-- the corresponding 'SendPort'.+spawnChannel :: forall a. Typeable a => Static (SerializableDict a)+             -> NodeId+             -> Closure (ReceivePort a -> Process ()) +             -> Process (SendPort a)+spawnChannel dict nid proc = do+    us <- getSelfPid+    spawn nid (go us) +    expect   where-    typeError dyn = userError $ "lookupStatic type error: " -                 ++ "cannot match " ++ show (dynTypeRep dyn) -                 ++ " against " ++ show (typeOf (undefined :: a))+    go :: ProcessId -> Closure (Process ())+    go pid = cpNewChan dict +           `cpBind` +             (cpSend (sdictSendPort dict) pid `cpSplit` proc)+           `cpBind`+             cpCancelL

@@ -1,47 +1,26 @@--- | Implementation of 'Closure' that works around the absence of 'static'.------ [Built-in closures]------ We offer a number of standard commonly useful closures.------ [Closure combinators]------ Closures combinators allow to create closures from other closures. For--- example, 'spawnSupervised' is defined as follows:------ > spawnSupervised :: NodeId --- >                 -> Closure (Process ()) --- >                 -> Process (ProcessId, MonitorRef)--- > spawnSupervised nid proc = do--- >   us   <- getSelfPid--- >   them <- spawn nid (linkClosure us `cpSeq` proc) --- >   ref  <- monitor them--- >   return (them, ref)------ [User-defined closures]+-- | Static values and Closures ----- Suppose we have a monomorphic function+-- [Static values] ----- > addInt :: Int -> Int -> Int--- > addInt x y = x + y+-- /Towards Haskell in the Cloud/ (Epstein et al., Haskell Symposium 2011) +-- proposes a new type construct called 'static' that characterizes values that+-- are known statically. There is no support for 'static' in ghc yet, however,+-- so we emulate it using Template Haskell. Given a top-level definition ----- Then the Template Haskell splice+-- > f :: forall a1 .. an. T+-- > f = ... ----- > remotable ['addInt]+-- you can use a Template Haskell splice to create a static version of 'f': -- --- creates a function ------ > $(mkClosure 'addInt) :: Int -> Closure (Int -> Int)+-- > $(mkStatic 'f) :: forall a1 .. an. Static T -- --- which can be used to partially apply 'addInt' and turn it into a 'Closure',--- which can be sent across the network. Closures can be deserialized with ------ > unClosure :: Typeable a => Closure a -> Process a+-- Every module that you write that contains calls to 'mkStatic' needs to+-- have a call to 'remotable': ----- In general, given a monomorphic function @f :: a -> b@ the corresponding --- function @$(mkClosure 'f)@ will have type @a -> Closure b@.+-- > remotable [ 'f, 'g, ... ] ----- The call to 'remotable' will also generate a function+-- where you must pass every function (or other value) that you pass as an +-- argument to 'mkStatic'. The call to 'remotable' will create a definition -- -- > __remoteTable :: RemoteTable -> RemoteTable --@@ -49,97 +28,228 @@ -- Cloud Haskell. You should have (at most) one call to 'remotable' per module, -- and compose all created functions when initializing Cloud Haskell: ----- > let rtable = M1.__remoteTable+-- > let rtable :: RemoteTable +-- >     rtable = M1.__remoteTable -- >            . M2.__remoteTable -- >            . ... -- >            . Mn.__remoteTable -- >            $ initRemoteTable  ----- See Section 6, /Faking It/, of /Towards Haskell in the Cloud/ for more info. +-- [Composing static values] ----- [Serializable Dictionaries]+-- We generalize the notion of 'static' as described in the paper, and also+-- provide ----- Some functions (such as 'sendClosure' or 'returnClosure') require an--- explicit (reified) serializable dictionary. To create such a dictionary do+-- > staticApply :: Static (a -> b) -> Static a -> Static b ----- > serializableDictInt :: SerializableDict Int--- > serializableDictInt = SerializableDict +-- This makes it possible to define a rich set of combinators on 'static'+-- values, a number of which are provided in this module.+--+-- [Closures]+--+-- Suppose you have a process+--+-- > factorial :: Int -> Process Int+--+-- Then you can use the supplied Template Haskell function 'mkClosure' to define+--+-- > factorialClosure :: Int -> Closure (Process Int)+-- > factorialClosure = $(mkClosure 'factorial)+--+-- You can then pass 'factorialClosure n' to 'spawn', for example, to have a+-- remote node compute a factorial number.+--+-- In general, if you have a /monomorphic/ function+--+-- > f :: T1 -> T2 -- --- and then pass @'serializableDictInt@ to 'remotable'. This will fail if the--- type is not serializable.+-- then+--+-- > $(mkClosure 'f) :: T1 -> Closure T2+--+-- provided that 'T1' is serializable (*).+--+-- [Creating closures manually]+--+-- You don't /need/ to use 'mkClosure', however.  Closures are defined exactly+-- as described in /Towards Haskell in the Cloud/:+-- +-- > data Closure a = Closure (Static (ByteString -> a)) ByteString+--+-- The splice @$(mkClosure 'factorial)@ 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) +-- >             `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+--+-- > $(functionDict 'f) :: Static (SerializableDict T1)+--+-- This is the dictionary that 'mkClosure' uses.+--+-- [Combinators on Closures]+--+-- Support for 'staticApply' (described above) also means that we can define+-- combinators on Closures, and we provide a number of them in this module,+-- the most important of which is 'cpBind'. Have a look at the implementation+-- of 'Control.Distributed.Process.call' for an example use.+--+-- [Notes]+--+-- (*) If 'T1' is not serializable you will get a type error in the generated+--     code. Unfortunately, the Template Haskell infrastructure cannot check+--     a priori if 'T1' is serializable or not due to a bug in the Template+--     Haskell libraries (<http://hackage.haskell.org/trac/ghc/ticket/7066>)+--+-- (**) Even though 'staticDecode' is passed an explicit serialization +--      dictionary, we still need the 'Typeable' constraint because +--      'Static' is not the /true/ static. If it was, we could 'unstatic'+--      the dictionary and pattern match on it to bring the 'Typeable'+--      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     remotable+  , mkStatic   , mkClosure+  , functionSDict+    -- * Primitive operations on static values+  , staticApply+  , staticDuplicate+    -- * Static functionals+  , staticConst+  , staticFlip+  , staticFst+  , staticSnd+  , staticCompose+  , staticFirst+  , staticSecond+  , staticSplit+    -- * Static constants+  , staticUnit+    -- * Creating closures+  , staticDecode+  , staticClosure+  , toClosure+    -- * Serialization dictionaries (and their static versions)   , SerializableDict(..)-    -- * Built-in closures-  , linkClosure-  , unlinkClosure-  , sendClosure-  , returnClosure-  , expectClosure-    -- * Generic closure combinators-  , closureApply-  , closureConst-  , closureUnit-    -- * Arrow combinators for processes+  , sdictUnit+  , sdictProcessId+  , sdictSendPort+    -- * Definition of CP and the generalized arrow combinators   , CP   , cpIntro   , cpElim   , cpId   , cpComp   , cpFirst-  , cpSwap   , cpSecond-  , cpPair-  , cpCopy-  , cpFanOut-  , cpLeft-  , cpMirror-  , cpRight-  , cpEither-  , cpUntag-  , cpFanIn-  , cpApply-    -- * Derived combinators for processes+  , cpSplit+  , cpCancelL+  , cpCancelR+    -- * Closure versions of CH primitives+  , cpLink+  , cpUnlink+  , cpSend+  , cpExpect+  , cpNewChan+    -- * @Closure (Process a)@ as a not-quite-monad+  , cpReturn    , cpBind   , cpSeq   ) where  -import Control.Distributed.Process.Internal.Types (SerializableDict(..))-import Control.Distributed.Process.Internal.Closure.TH (remotable, mkClosure)-import Control.Distributed.Process.Internal.Closure.BuiltIn -  ( linkClosure-  , unlinkClosure-  , sendClosure-  , returnClosure-  , expectClosure+import Control.Distributed.Process.Internal.Types +  ( SerializableDict(..)+  , staticApply+  , staticDuplicate   )-import Control.Distributed.Process.Internal.Closure.Combinators -  ( -- Generic combinators-    closureApply-  , closureConst-  , closureUnit-    -- Arrow combinators for processes-  , CP+import Control.Distributed.Process.Internal.Closure.TH +  ( remotable+  , mkStatic+  , functionSDict+  )+import Control.Distributed.Process.Internal.Closure.Static+  ( -- Static functionals+    staticConst+  , staticFlip+  , staticFst+  , staticSnd+  , staticCompose+  , staticFirst+  , staticSecond+  , staticSplit+    -- Static constants+  , staticUnit+    -- Creating closures+  , staticDecode+  , staticClosure+  , toClosure+    -- Serialization dictionaries (and their static versions)+  , sdictUnit+  , sdictProcessId+  , sdictSendPort+  )+import Control.Distributed.Process.Internal.Closure.MkClosure (mkClosure)+import Control.Distributed.Process.Internal.Closure.CP+  ( -- Definition of CP and the generalized arrow combinators+    CP   , cpIntro   , cpElim   , cpId   , cpComp   , cpFirst-  , cpSwap   , cpSecond-  , cpPair-  , cpCopy-  , cpFanOut-  , cpLeft-  , cpMirror-  , cpRight-  , cpEither-  , cpUntag-  , cpFanIn-  , cpApply-    -- Derived process operators+  , cpSplit+  , cpCancelL+  , cpCancelR+    -- Closure versions of CH primitives+  , cpLink+  , cpUnlink+  , cpSend+  , cpExpect+  , cpNewChan+    -- @Closure (Process a)@ as a not-quite-monad+  , cpReturn    , cpBind   , cpSeq   )

@@ -1,133 +0,0 @@-{-# LANGUAGE MagicHash #-}-module Control.Distributed.Process.Internal.Closure -  ( -- * Runtime support-    initRemoteTable-  , resolveClosure-  ) where--import qualified Data.Map as Map (empty)-import Data.Accessor ((^.))-import Data.ByteString.Lazy (ByteString)-import Data.Binary (decode)-import Data.Typeable (TypeRep)-import Control.Applicative ((<$>))-import Control.Distributed.Process.Internal.Types-  ( RemoteTable(RemoteTable)-  , remoteTableLabel-  , remoteTableDict-  , StaticLabel(..)-  , ProcessId-  , Process-  , Closure(Closure)-  , Static(Static)-  , RuntimeSerializableSupport(..)-  )-import Control.Distributed.Process.Internal.Dynamic-  ( Dynamic(Dynamic)-  , toDyn-  , dynApp-  , dynApply-  , unsafeCoerce#-  )-import Control.Distributed.Process.Internal.TypeRep () -- Binary instances  -import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn-  (remoteTable)------------------------------------------------------------------------------------- Runtime support for closures                                               --------------------------------------------------------------------------------------- | Initial (empty) remote-call meta data-initRemoteTable :: RemoteTable-initRemoteTable = BuiltIn.remoteTable (RemoteTable Map.empty Map.empty)--resolveClosure :: RemoteTable -> StaticLabel -> ByteString -> Maybe Dynamic--- Built-in closures-resolveClosure rtable ClosureSend env = do-    rss <- rtable ^. remoteTableDict typ -    rssSend rss `dynApply` toDyn pid -  where-    (typ, pid) = decode env :: (TypeRep, ProcessId)-resolveClosure rtable ClosureReturn env = do-    rss <- rtable ^. remoteTableDict typ -    rssReturn rss `dynApply` toDyn arg -  where-    (typ, arg) = decode env :: (TypeRep, ByteString)-resolveClosure rtable ClosureExpect env = -    rssExpect <$> rtable ^. remoteTableDict typ-  where-    typ = decode env :: TypeRep--- Generic closure combinators-resolveClosure rtable ClosureApply env = do -    f <- resolveClosure rtable labelf envf-    x <- resolveClosure rtable labelx envx-    f `dynApply` x-  where-    (labelf, envf, labelx, envx) = decode env -resolveClosure _rtable ClosureConst env = -  return $ Dynamic (decode env) (unsafeCoerce# const)-resolveClosure _rtable ClosureUnit _env =-  return $ toDyn ()--- Arrow combinators-resolveClosure _rtable CpId env =-    return $ Dynamic (decode env) (unsafeCoerce# cpId)-  where-    cpId :: forall a. a -> Process a-    cpId = return-resolveClosure _rtable CpComp  env =-    return $ Dynamic (decode env) (unsafeCoerce# cpComp)-  where-    cpComp :: forall a b c. (a -> Process b) -> (b -> Process c) -> a -> Process c-    cpComp p q a = p a >>= q -resolveClosure _rtable CpFirst env =-    return $ Dynamic (decode env) (unsafeCoerce# cpFirst)-  where-    cpFirst :: forall a b c. (a -> Process b) -> (a, c) -> Process (b, c)-    cpFirst p (a, c) = do b <- p a ; return (b, c) -resolveClosure _rtable CpSwap env =-    return $ Dynamic (decode env) (unsafeCoerce# cpSwap) -  where-    cpSwap :: forall a b. (a, b) -> Process (b, a)-    cpSwap (a, b) = return (b, a) -resolveClosure _rtable CpCopy env =-    return $ Dynamic (decode env) (unsafeCoerce# cpCopy)-  where-    cpCopy :: forall a. a -> Process (a, a)-    cpCopy a = return (a, a) -resolveClosure _rtable CpLeft env =-    return $ Dynamic (decode env) (unsafeCoerce# cpLeft)-  where-    cpLeft :: forall a b c. (a -> Process b) -> Either a c -> Process (Either b c)-    cpLeft p (Left a)  = do b <- p a ; return (Left b) -    cpLeft _ (Right c) = return (Right c) -resolveClosure _rtable CpMirror env =-    return $ Dynamic (decode env) (unsafeCoerce# cpMirror)-  where-    cpMirror :: forall a b. Either a b -> Process (Either b a)-    cpMirror (Left a)  = return (Right a) -    cpMirror (Right b) = return (Left b)-resolveClosure _rtable CpUntag env =-    return $ Dynamic (decode env) (unsafeCoerce# cpUntag)-  where-    cpUntag :: forall a. Either a a -> Process a-    cpUntag (Left a)  = return a-    cpUntag (Right a) = return a-resolveClosure rtable CpApply env =-    return $ Dynamic typApply (unsafeCoerce# cpApply)-  where-    cpApply :: forall a b. (Closure (a -> Process b), a) -> Process b -    cpApply (Closure (Static flabel) fenv, x) = do-      let Just f = resolveClosure rtable flabel fenv-          Dynamic typResult val = f `dynApp` Dynamic typA (unsafeCoerce# x) -      if typResult == typProcB -        then unsafeCoerce# val-        else error $ "Type error in cpApply: "-                  ++ "mismatch between " ++ show typResult -                  ++ " and " ++ show typProcB--    (typApply, typA, typProcB) = decode env---- User defined closures-resolveClosure rtable (UserStatic label) env = do-  val <- rtable ^. remoteTableLabel label -  dynApply val (toDyn env)

@@ -1,77 +0,0 @@-{-# LANGUAGE TemplateHaskell #-}-module Control.Distributed.Process.Internal.Closure.BuiltIn -  ( -- * Runtime support for the builtin closures-    remoteTable-    -- * Serialization dictionaries-  , serializableDictUnit-    -- * Closures-  , linkClosure-  , unlinkClosure-  , sendClosure-  , returnClosure-  , expectClosure-  ) where--import Data.Binary (encode)-import Data.Typeable (typeOf)-import Control.Distributed.Process.Internal.Primitives (link, unlink)-import Control.Distributed.Process.Internal.Types -  ( ProcessId-  , Closure-  , Process-  , RemoteTable-  , SerializableDict(..)-  , Closure(..)-  , Static(..)-  , StaticLabel(..)-  )-import Control.Distributed.Process.Internal.Closure.TH (remotable, mkClosure)-import Control.Distributed.Process.Internal.TypeRep () -- Binary instances--serializableDictUnit :: SerializableDict ()-serializableDictUnit = SerializableDict--remotable ['link, 'unlink, 'serializableDictUnit] --remoteTable :: RemoteTable -> RemoteTable-remoteTable = __remoteTable------------------------------------------------------------------------------------- TH generated closures                                                      --------------------------------------------------------------------------------------- | Closure version of 'link'-linkClosure :: ProcessId -> Closure (Process ())-linkClosure = $(mkClosure 'link)---- | Closure version of 'unlink'-unlinkClosure :: ProcessId -> Closure (Process ())-unlinkClosure = $(mkClosure 'unlink)------------------------------------------------------------------------------------- Polymorphic closures                                                       -----                                                                            ----- TODO: These functions take a SerializableDict as argument. When we get     ----- proper support for static, ideally this argument disappears completely;    ----- but if not, it should turn into a static (SerializableDict a). We don't    ----- require them to be "static" here because we don't have a pure 'unstatic'   ----- function, and hence have no way of turning a static (SerializableDict a)   ----- into an actual SerialziableDict a (we need that in order to pattern match  ----- on the dictionary and bring the type class dictionary into scope, so that  ----- we can call 'encode' (for instance in 'returnClosure').                    --------------------------------------------------------------------------------------- | Closure version of 'send'-sendClosure :: forall a. SerializableDict a -> ProcessId -> Closure (a -> Process ())-sendClosure SerializableDict pid =-  Closure (Static ClosureSend) (encode (typeOf (undefined :: a), pid)) ---- | Return any value-returnClosure :: forall a. SerializableDict a -> a -> Closure (Process a)-returnClosure SerializableDict val =-  Closure (Static ClosureReturn) (encode (typeOf (undefined :: a), encode val))---- | Closure version of 'expect'-expectClosure :: forall a. SerializableDict a -> Closure (Process a)-expectClosure SerializableDict =-  Closure (Static ClosureExpect) (encode (typeOf (undefined :: a)))

@@ -0,0 +1,336 @@+-- | Combinator for process closures +{-# LANGUAGE MagicHash #-}+module Control.Distributed.Process.Internal.Closure.CP+  ( -- * Definition of CP and the generalized arrow combinators+    CP+  , cpIntro+  , cpElim+  , cpId+  , cpComp+  , cpFirst+  , cpSecond+  , cpSplit+  , cpCancelL+  , cpCancelR+    -- * Closure versions of CH primitives+  , cpLink+  , cpUnlink+  , cpSend+  , cpExpect+  , cpNewChan+    -- * @Closure (Process a)@ as a not-quite-monad+  , cpReturn +  , cpBind+  , cpSeq+    -- * Runtime support+  , __remoteTable+  ) where++import Data.Binary (encode)+import Data.ByteString.Lazy (ByteString)+import Data.Typeable (Typeable, typeOf, typeRepTyCon, TyCon)+import Data.Typeable.Internal (TypeRep(..))+import Control.Applicative ((<$>))+import Control.Monad ((>=>))+import Control.Distributed.Process.Serializable (Serializable)+import Control.Distributed.Process.Internal.Types+  ( Closure(Closure)+  , SerializableDict(SerializableDict)+  , Static(Static)+  , Process+  , staticApply+  , staticDuplicate+  , staticTypeOf+  , typeOfStaticLabel+  , ProcessId+  , LocalNode(remoteTable)+  , procMsg+  , SendPort+  , ReceivePort+  )+import Control.Distributed.Process.Internal.Primitives +  ( link+  , unlink+  , send+  , expect+  , 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+  , staticDecode+  , staticClosure+  , staticSplit+  , staticConst+  , staticUnit+  , staticFlip+  , staticFst+  , staticSnd+  , sdictProcessId+  )+import Control.Distributed.Process.Internal.Closure.MkClosure (mkClosure)+import Control.Distributed.Process.Internal.Dynamic +  ( Dynamic(Dynamic)+  , unsafeCoerce#+  , dynTypeRep+  , dynKleisli+  )++--------------------------------------------------------------------------------+-- Setup: A number of functions that we will pass to 'remotable'              --+--------------------------------------------------------------------------------++---- Type specializations of monadic operations on Processes -------------------++returnProcess :: a -> Process a+returnProcess = return++bindProcess :: Process a -> (a -> Process b) -> Process b+bindProcess = (>>=) ++kleisliProcess :: (a -> Process b) -> (b -> Process c) -> a -> Process c+kleisliProcess = (>=>)++parJoinProcess :: Process (a -> Process b) -> a -> Process b+parJoinProcess proc a = proc >>= ($ a)++---- Variations on standard or CH functions with an explicit dictionary arg ----++sendDict :: SerializableDict a -> ProcessId -> a -> Process ()+sendDict SerializableDict = send++expectDict :: SerializableDict a -> Process a+expectDict SerializableDict = expect++newChanDict :: SerializableDict a -> Process (SendPort a, ReceivePort a)+newChanDict SerializableDict = newChan++---- Serialization dictionaries ------------------------------------------------++-- | Specialized serialization dictionary required in 'cpBind'+sdictComp :: SerializableDict (ByteString, ByteString)+sdictComp = SerializableDict++---- Some specialised processes necessary to implement the combinators ---------++-- | Resolve a closure+unClosure :: Static a -> ByteString -> Process Dynamic+unClosure (Static label) env = do+  rtable <- remoteTable <$> procMsg getLocalNode +  case resolveClosure rtable label env of+    Nothing  -> fail "Derived.unClosure: resolveClosure failed"+    Just dyn -> return dyn++-- | Work around a bug in Typeable+-- (http://hackage.haskell.org/trac/ghc/ticket/5692)+compareWithoutFingerprint :: TypeRep -> TypeRep -> Bool+compareWithoutFingerprint (TypeRep _ con ts) (TypeRep _ con' ts') +  = con == con' && all (uncurry compareWithoutFingerprint) (zip ts ts')++-- | Remove a 'Dynamic' constructor, provided that the recorded type matches the+-- type of the first static argument (the value of that argument is not used)+unDynamic :: Static a -> Process Dynamic -> Process a+unDynamic (Static label) pdyn = do+  Dynamic typ val <- pdyn+  if compareWithoutFingerprint typ (typeOfStaticLabel label) -- typ == typeOfStaticLabel label +    then return (unsafeCoerce# val)+    else fail $ "unDynamic: cannot match " +             ++ show typ+             ++ " against expected type " +             ++ show (typeOfStaticLabel label)++-- | Dynamic kleisli composition+--+-- The first argument stops remotable from trying to generate a SerializableDict+-- for (Process Dynamic, Process Dynamic)+kleisliCompDyn :: () -> (Process Dynamic, Process Dynamic) -> Process Dynamic+kleisliCompDyn () (pf, pg) = do+    f <- pf -- a -> Process b+    g <- pg -- b -> Process c+    case dynKleisli tyConProcess kleisliProcess f g of+      Just dyn -> return dyn+      Nothing  -> fail $ "kleisliCompDyn: could not compose "+                      ++ show (dynTypeRep f)+                      ++ " with "+                      ++ show (dynTypeRep g)+  where+    tyConProcess :: TyCon+    tyConProcess = typeRepTyCon (typeOf (undefined :: Process ()))++cpFirstAux :: (a -> Process b) -> (a, c) -> Process (b, c)+cpFirstAux f (a, c) = f a >>= \b -> return (b, c) ++cpSecondAux :: (a -> Process b) -> (c, a) -> Process (c, b)+cpSecondAux f (c, a) = f a >>= \b -> return (c, b) ++---- Finally, the call to remotable --------------------------------------------++remotable [ -- Monadic operations+            'returnProcess+          , 'bindProcess+          , 'parJoinProcess+            -- CH primitives +          , 'link+          , 'unlink+            -- Explicit dictionaries+          , 'sendDict+          , 'expectDict+          , 'newChanDict+            -- Serialization dictionaries+          , 'sdictComp+            -- Specialized processes+          , 'unClosure+          , 'unDynamic+          , 'kleisliCompDyn+          , 'cpFirstAux+          , 'cpSecondAux+          ]++--------------------------------------------------------------------------------+-- Some derived static functions                                              --+--------------------------------------------------------------------------------++staticUndynamic :: Typeable a => Static (Process Dynamic -> Process a)+staticUndynamic = +  $(mkStatic 'unDynamic) `staticApply` staticDuplicate (staticTypeOf (undefined :: a))++staticKleisliCompDyn :: Static ((Process Dynamic, Process Dynamic) -> Process Dynamic)+staticKleisliCompDyn = $(mkStatic 'kleisliCompDyn) `staticApply` staticUnit ++--------------------------------------------------------------------------------+-- Definition of CP and the generalized arrow combinators                     --+--------------------------------------------------------------------------------++-- | 'CP a b' represents the closure of a process parameterized by 'a' and+-- returning 'b'. 'CP a b' forms a (restricted) generalized arrow+-- (<http://www.cs.berkeley.edu/~megacz/garrows/>)+type CP a b = Closure (a -> Process b)++-- | 'CP' introduction form +cpIntro :: forall a b. (Typeable a, Typeable b)+        => Closure (Process b) -> Closure (a -> Process b)+cpIntro (Closure static env) = Closure decoder env +  where+    decoder :: Static (ByteString -> a -> Process b)+    decoder = staticConst `staticCompose` static++-- | 'CP' elimination form+cpElim :: forall a. Typeable a => CP () a -> Closure (Process a)+cpElim (Closure static env) = Closure decoder env+  where+    decoder :: Static (ByteString -> Process a)+    decoder = staticFlip static `staticApply` staticUnit ++-- | Identity ('Closure' version of 'return')+cpId :: Typeable a => CP a a+cpId = staticClosure $(mkStatic 'returnProcess)++-- | Left-to-right composition ('Closure' version of '>=>')+cpComp :: forall a b c. (Typeable a, Typeable b, Typeable c)+       => CP a b -> CP b c -> CP a c+cpComp (Closure fstatic fenv) (Closure gstatic genv) =+    Closure decoder (encode (fenv, genv))+  where+    decoder :: Static (ByteString -> a -> Process c)+    decoder = $(mkStatic 'parJoinProcess)+            `staticCompose`+              staticUndynamic+            `staticCompose`+              staticKleisliCompDyn+            `staticCompose`+              (staticUnclosure fstatic `staticSplit` staticUnclosure gstatic)+            `staticCompose`+              staticDecode $(mkStatic 'sdictComp)    ++-- | First+cpFirst :: forall a b c. (Typeable a, Typeable b, Typeable c)+        => CP a b -> CP (a, c) (b, c)+cpFirst (Closure static env) = Closure decoder env+  where+    decoder :: Static (ByteString -> (a, c) -> Process (b, c))+    decoder = $(mkStatic 'cpFirstAux) `staticCompose` static++-- | Second +cpSecond :: forall a b c. (Typeable a, Typeable b, Typeable c)+        => CP a b -> CP (c, a) (c, b)+cpSecond (Closure static env) = Closure decoder env+  where+    decoder :: Static (ByteString -> (c, a) -> Process (c, b))+    decoder = $(mkStatic 'cpSecondAux) `staticCompose` static++-- | Split (Like 'Control.Arrow.***')+cpSplit :: (Typeable a, Typeable b, Typeable c, Typeable d)+        => CP a c -> CP b d -> CP (a, b) (c, d)+cpSplit f g = cpFirst f `cpComp` cpSecond g++-- | Left cancellation+cpCancelL :: Typeable a => CP ((), a) a+-- Closure (((), a) -> Process a)+cpCancelL = staticClosure ($(mkStatic 'returnProcess) `staticCompose` staticSnd) ++-- | Right cancellation+cpCancelR :: Typeable a => CP (a, ()) a+cpCancelR = staticClosure ($(mkStatic 'returnProcess) `staticCompose` staticFst)++--------------------------------------------------------------------------------+-- Closure versions of CH primitives                                          --+--------------------------------------------------------------------------------++-- | Closure version of 'link'+cpLink :: ProcessId -> Closure (Process ())+cpLink = $(mkClosure 'link)++-- | Closure version of 'unlink'+cpUnlink :: ProcessId -> Closure (Process ())+cpUnlink = $(mkClosure 'unlink)++-- | Closure version of 'send'+cpSend :: forall a. Typeable a +       => Static (SerializableDict a) -> ProcessId -> Closure (a -> Process ())+cpSend dict pid = Closure decoder (encode pid)+  where+    decoder :: Static (ByteString -> a -> Process ())+    decoder = ($(mkStatic 'sendDict) `staticApply` dict)+            `staticCompose` +              staticDecode sdictProcessId ++-- | Closure version of 'expect'+cpExpect :: Typeable a => Static (SerializableDict a) -> Closure (Process a)+cpExpect dict = staticClosure ($(mkStatic 'expectDict) `staticApply` dict)++-- | Closure version of 'newChan'+cpNewChan :: Typeable a +          => Static (SerializableDict a) +          -> Closure (Process (SendPort a, ReceivePort a))+cpNewChan dict = staticClosure ($(mkStatic 'newChanDict) `staticApply` dict)++--------------------------------------------------------------------------------+-- (Closure . Process) as a not-quite-monad                                   --+--------------------------------------------------------------------------------++-- | Not-quite-monadic 'return'+cpReturn :: forall a. Serializable a +         => Static (SerializableDict a) -> a -> Closure (Process a)+cpReturn dict x = Closure decoder (encode x)+  where+    decoder :: Static (ByteString -> Process a)+    decoder = $(mkStatic 'returnProcess) +            `staticCompose`+              staticDecode dict+++staticUnclosure :: Typeable a +                => Static a -> Static (ByteString -> Process Dynamic)+staticUnclosure s = +  $(mkStatic 'unClosure) `staticApply` staticDuplicate s ++-- | Not-quite-monadic bind ('>>=')+cpBind :: forall a b. (Typeable a, Typeable b)+       => Closure (Process a) -> Closure (a -> Process b) -> Closure (Process b)+cpBind x f = cpElim (cpIntro x `cpComp` f)++-- | Monadic sequencing ('>>')+cpSeq :: Closure (Process ()) -> Closure (Process ()) -> Closure (Process ())+cpSeq p q = p `cpBind` cpIntro q

@@ -1,182 +0,0 @@-module Control.Distributed.Process.Internal.Closure.Combinators -  ( -- * Generic combinators-    closureApply-  , closureConst-  , closureUnit-    -- * Arrow combinators for processes-  , CP-  , cpIntro-  , cpElim-  , cpId-  , cpComp-  , cpFirst-  , cpSwap-  , cpSecond-  , cpPair-  , cpCopy-  , cpFanOut-  , cpLeft-  , cpMirror-  , cpRight-  , cpEither-  , cpUntag-  , cpFanIn-  , cpApply-    -- * Derived process operators-  , cpBind-  , cpSeq-  ) where--import Prelude hiding (lookup)-import qualified Data.ByteString.Lazy as BS (empty)-import Data.Binary (encode)-import Data.Typeable (typeOf, Typeable)-import Control.Distributed.Process.Internal.Types-  ( Closure(..)-  , Static(..)-  , StaticLabel(..)-  , Process-  )-import Control.Distributed.Process.Internal.TypeRep () -- Binary instances------------------------------------------------------------------------------------- Generic closure combinators                                                -- -----------------------------------------------------------------------------------closureApply :: Closure (a -> b) -> Closure a -> Closure b-closureApply (Closure (Static labelf) envf) (Closure (Static labelx) envx) = -  Closure (Static ClosureApply) $ encode (labelf, envf, labelx, envx)--closureConst :: forall a b. (Typeable a, Typeable b) -          => Closure (a -> b -> a)-closureConst = Closure (Static ClosureConst) (encode $ typeOf aux)-  where-    aux :: a -> b -> a-    aux = undefined--closureUnit :: Closure ()-closureUnit = Closure (Static ClosureUnit) BS.empty------------------------------------------------------------------------------------- Arrow combinators for processes                                            -- -----------------------------------------------------------------------------------type CP a b = Closure (a -> Process b)--cpIntro :: (Typeable a, Typeable b)-        => Closure (Process b) -> CP a b -cpIntro = closureApply closureConst --cpElim :: Typeable a -       => CP () a -> Closure (Process a)-cpElim = flip closureApply closureUnit --cpId :: forall a. Typeable a -     => CP a a -cpId = Closure (Static CpId) (encode $ typeOf aux)-  where-    aux :: a -> Process a-    aux = undefined--cpComp :: forall a b c. (Typeable a, Typeable b, Typeable c) -       => CP a b -> CP b c -> CP a c-cpComp f g = comp `closureApply` f `closureApply` g -  where-    comp :: Closure ((a -> Process b) -> (b -> Process c) -> a -> Process c)-    comp = Closure (Static CpComp) (encode $ typeOf aux)-    -    aux :: (a -> Process b) -> (b -> Process c) -> a -> Process c-    aux = undefined--cpFirst :: forall a b c. (Typeable a, Typeable b, Typeable c)-        => CP a b -> CP (a, c) (b, c)-cpFirst = closureApply first-  where-    first :: Closure ((a -> Process b) -> (a, c) -> Process (b, c))-    first = Closure (Static CpFirst) (encode $ typeOf aux)--    aux :: (a -> Process b) -> (a, c) -> Process (b, c)-    aux = undefined--cpSwap :: forall a b. (Typeable a, Typeable b)-       => CP (a, b) (b, a)-cpSwap = Closure (Static CpSwap) (encode $ typeOf aux)-  where-    aux :: (a, b) -> Process (b, a)-    aux = undefined--cpSecond :: (Typeable a, Typeable b, Typeable c)-         => CP a b -> CP (c, a) (c, b)-cpSecond f = cpSwap `cpComp` cpFirst f `cpComp` cpSwap--cpPair :: (Typeable a, Typeable a', Typeable b, Typeable b')-        => CP a b -> CP a' b' -> CP (a, a') (b, b')-cpPair f g = cpFirst f `cpComp` cpSecond g--cpCopy :: forall a. Typeable a -       => CP a (a, a)-cpCopy = Closure (Static CpCopy) (encode $ typeOf aux)-  where-    aux :: a -> Process (a, a)-    aux = undefined--cpFanOut :: (Typeable a, Typeable b, Typeable c)-         => CP a b -> CP a c -> CP a (b, c)-cpFanOut f g = cpCopy `cpComp` (f `cpPair` g)         --cpLeft :: forall a b c. (Typeable a, Typeable b, Typeable c)-       => CP a b -> CP (Either a c) (Either b c)-cpLeft = closureApply left-  where-    left :: Closure ((a -> Process b) -> Either a c -> Process (Either b c)) -    left = Closure (Static CpLeft) (encode $ typeOf aux)--    aux :: (a -> Process b) -> Either a c -> Process (Either b c)-    aux = undefined--cpMirror :: forall a b. (Typeable a, Typeable b)-         => CP (Either a b) (Either b a)-cpMirror = Closure (Static CpMirror) (encode $ typeOf aux)-  where-    aux :: Either a b -> Process (Either b a)-    aux = undefined--cpRight :: forall a b c. (Typeable a, Typeable b, Typeable c)-        => CP a b -> CP (Either c a) (Either c b)-cpRight f = cpMirror `cpComp` cpLeft f `cpComp` cpMirror --cpEither :: (Typeable a, Typeable a', Typeable b, Typeable b')-         => CP a b -> CP a' b' -> CP (Either a a') (Either b b')-cpEither f g = cpLeft f `cpComp` cpRight g--cpUntag :: forall a. Typeable a-        => CP (Either a a) a-cpUntag = Closure (Static CpUntag) (encode $ typeOf aux)-  where-    aux :: Either a a -> Process a-    aux = undefined--cpFanIn :: (Typeable a, Typeable b, Typeable c) -        => CP a c -> CP b c -> CP (Either a b) c-cpFanIn f g = (f `cpEither` g) `cpComp` cpUntag --cpApply :: forall a b. (Typeable a, Typeable b)-        => CP (CP a b, a) b-cpApply = Closure (Static CpApply) $ encode ( typeOf aux-                                            , typeOf (undefined :: a) -                                            , typeOf (undefined :: Process b)-                                            )-  where-    aux :: (Closure (a -> Process b), a) -> Process b-    aux = undefined------------------------------------------------------------------------------------- Some derived operators for processes                                       -- -----------------------------------------------------------------------------------cpBind :: (Typeable a, Typeable b) -       => Closure (Process a) -> Closure (a -> Process b) -> Closure (Process b)-cpBind x f = cpElim $ cpIntro x `cpComp` f--cpSeq :: Closure (Process ()) -> Closure (Process ()) -> Closure (Process ())-cpSeq p q = p `cpBind` cpIntro q

@@ -0,0 +1,25 @@+{-# LANGUAGE TemplateHaskell #-}+module Control.Distributed.Process.Internal.Closure.MkClosure (mkClosure) where++import Data.Binary (encode)+import Language.Haskell.TH (Q, Exp, Name)+import Control.Distributed.Process.Internal.Types (Closure(Closure))+import Control.Distributed.Process.Internal.Closure.TH +  ( mkStatic+  , functionSDict+  )+import Control.Distributed.Process.Internal.Closure.Static +  ( staticCompose+  , staticDecode+  )++-- | Create a closure+--+-- If @f : T1 -> T2@ is a /monomorphic/ function +-- then @$(mkClosure 'f) :: T1 -> Closure T2@.+-- Be sure to pass 'f' to 'remotable'. +mkClosure :: Name -> Q Exp+mkClosure n = +  [|   Closure ($(mkStatic n) `staticCompose` staticDecode $(functionSDict n)) +     . encode+  |]

@@ -0,0 +1,34 @@+{-# LANGUAGE MagicHash #-}+module Control.Distributed.Process.Internal.Closure.Resolution (resolveClosure) where++import Data.Accessor ((^.))+import Data.ByteString.Lazy (ByteString)+import Control.Distributed.Process.Internal.Types+  ( RemoteTable+  , remoteTableLabel+  , Static(Static)+  , StaticLabel(..)+  )+import Control.Distributed.Process.Internal.Dynamic+  ( Dynamic(Dynamic)+  , toDyn+  , dynApply+  , unsafeCoerce#+  )+import Control.Distributed.Process.Internal.TypeRep () -- Binary instances  ++resolveStatic :: RemoteTable -> StaticLabel -> Maybe Dynamic+resolveStatic rtable (StaticLabel string typ) = do+  Dynamic _ val <- rtable ^. remoteTableLabel string+  return (Dynamic typ val)+resolveStatic rtable (StaticApply static1 static2) = do+  f <- resolveStatic rtable static1+  x <- resolveStatic rtable static2+  f `dynApply` x+resolveStatic _rtable (StaticDuplicate static typ) = +  return $ Dynamic typ (unsafeCoerce# (Static static))++resolveClosure :: RemoteTable -> StaticLabel -> ByteString -> Maybe Dynamic+resolveClosure rtable static env = do+  decoder <- resolveStatic rtable static+  decoder `dynApply` toDyn env

@@ -0,0 +1,190 @@+-- | Combinators on static values +{-# LANGUAGE MagicHash #-}+module Control.Distributed.Process.Internal.Closure.Static+  ( -- * Static functionals+    staticConst+  , staticFlip+  , staticFst+  , staticSnd+  , staticCompose+  , staticFirst+  , staticSecond+  , staticSplit+    -- * Static constants+  , staticUnit+    -- * Creating closures+  , staticDecode+  , staticClosure+  , toClosure+    -- * Serialization dictionaries (and their static versions)+  , sdictUnit+  , sdictProcessId+  , sdictSendPort+    -- * Runtime support+  , __remoteTable+  ) where++import Data.Binary (encode, decode)+import Data.ByteString.Lazy (ByteString, empty)+import Data.Typeable (Typeable)+import Control.Distributed.Process.Serializable (Serializable)+import Control.Distributed.Process.Internal.Types+  ( Closure(Closure)+  , SerializableDict(SerializableDict)+  , Static+  , staticApply+  , ProcessId+  , SendPort+  )+import Control.Distributed.Process.Internal.Closure.TH (remotable, mkStatic)+import qualified Control.Arrow as Arrow (first, second, (***))++--------------------------------------------------------------------------------+-- Setup: A number of functions that we will pass to 'remotable'              --+--------------------------------------------------------------------------------++---- Functionals ---------------------------------------------------------------++compose :: (b -> c) -> (a -> b) -> a -> c+compose = (.)++first :: (a -> b) -> (a, c) -> (b, c)+first = Arrow.first ++second :: (a -> b) -> (c, a) -> (c, b)+second = Arrow.second++split :: (a -> b) -> (a' -> b') -> (a, a') -> (b, b')+split = (Arrow.***)++---- Constants -----------------------------------------------------------------++unit :: ()+unit = ()++---- Variations on standard or CH functions with an explicit dictionary arg ----++decodeDict :: SerializableDict a -> ByteString -> a+decodeDict SerializableDict = decode++---- Serialization dictionaries ------------------------------------------------++sdictUnit_ :: SerializableDict ()+sdictUnit_ = SerializableDict++sdictProcessId_ :: SerializableDict ProcessId+sdictProcessId_ = SerializableDict++sdictSendPort_ :: SerializableDict a -> SerializableDict (SendPort a)+sdictSendPort_ SerializableDict = SerializableDict++---- Finally, the call to remotable --------------------------------------------++remotable [ -- Functionals (predefined)+            'const+          , 'flip+          , 'fst+          , 'snd+            -- Functionals (defined above)+          , 'compose+          , 'first+          , 'second+          , 'split+            -- Constants+          , 'unit+            -- Explicit dictionaries+          , 'decodeDict+            -- Serialization dictionaries+          , 'sdictUnit_+          , 'sdictProcessId_+          , 'sdictSendPort_+          ]++--------------------------------------------------------------------------------+-- Static versions of the functionals                                         -- +-- (We give these explicit names because they are useful outside this module) --+--------------------------------------------------------------------------------++-- | Static version of 'const'+staticConst :: (Typeable a, Typeable b) => Static (a -> b -> a)+staticConst = $(mkStatic 'const)++-- | Static version of 'flip'+staticFlip :: (Typeable a, Typeable b, Typeable c) +           => Static (a -> b -> c) -> Static (b -> a -> c)+staticFlip f = $(mkStatic 'flip) `staticApply` f           ++-- | Static version of 'fst'+staticFst :: (Typeable a, Typeable b)+          => Static ((a, b) -> a)+staticFst = $(mkStatic 'fst)++-- | Static version of 'snd'+staticSnd :: (Typeable a, Typeable b)+          => Static ((a, b) -> b)+staticSnd = $(mkStatic 'snd)++-- | Static version of ('Prelude..')+staticCompose :: (Typeable a, Typeable b, Typeable c) +              => Static (b -> c) -> Static (a -> b) -> Static (a -> c)+staticCompose f x = $(mkStatic 'compose) `staticApply` f `staticApply` x ++-- | Static version of 'Control.Arrow.first'+staticFirst :: (Typeable a, Typeable b, Typeable c)+            => Static ((a -> b) -> (a, c) -> (b, c))+staticFirst = $(mkStatic 'first)++-- | Static version of 'Control.Arrow.second'+staticSecond :: (Typeable a, Typeable b, Typeable c)+             => Static ((a -> b) -> (c, a) -> (c, b))+staticSecond = $(mkStatic 'second)++-- | Static version of ('Control.Arrow.***')+staticSplit :: (Typeable a, Typeable b, Typeable c, Typeable d) +            => Static (a -> c) -> Static (b -> d) -> Static ((a, b) -> (c, d))+staticSplit f g = $(mkStatic 'split) `staticApply` f `staticApply` g ++--------------------------------------------------------------------------------+-- Constants                                                                  --+--------------------------------------------------------------------------------++-- | Static version of '()'+staticUnit :: Static ()+staticUnit = $(mkStatic 'unit)++--------------------------------------------------------------------------------+-- Dictionaries                                                               --+--------------------------------------------------------------------------------++-- | Serialization dictionary for '()' +sdictUnit :: Static (SerializableDict ())+sdictUnit = $(mkStatic 'sdictUnit_)++-- | Serialization dictionary for 'ProcessId' +sdictProcessId :: Static (SerializableDict ProcessId)+sdictProcessId = $(mkStatic 'sdictProcessId_)++-- | Serialization dictionary for 'SendPort'+sdictSendPort :: Typeable a +              => Static (SerializableDict a) -> Static (SerializableDict (SendPort a))+sdictSendPort = staticApply $(mkStatic 'sdictSendPort_) ++--------------------------------------------------------------------------------+-- Creating closures                                                          --+--------------------------------------------------------------------------------++-- | Static decoder, given a static serialization dictionary.+--+-- See module documentation of "Control.Distributed.Process.Closure" for an+-- example.+staticDecode :: Typeable a => Static (SerializableDict a) -> Static (ByteString -> a)+staticDecode dict = $(mkStatic 'decodeDict) `staticApply` dict ++-- | Convert a static value into a closure.+staticClosure :: forall a. Typeable a => Static a -> Closure a+staticClosure static = Closure (staticConst `staticApply` static) empty++-- | Convert a serializable value into a closure.+toClosure :: forall a. Serializable a +          => Static (SerializableDict a) -> a -> Closure a+toClosure dict x = Closure (staticDecode dict) (encode x) 

@@ -1,15 +1,15 @@ -- | Template Haskell support -- -- (In a separate file for convenience)+{-# LANGUAGE MagicHash #-} module Control.Distributed.Process.Internal.Closure.TH    ( -- * User-level API     remotable-  , mkClosure+  , mkStatic+  , functionSDict   ) where  import Prelude hiding (lookup)-import Data.ByteString.Lazy (ByteString)-import Data.Binary (encode, decode) import Data.Accessor ((^=)) import Data.Typeable (typeOf) import Control.Applicative ((<$>))@@ -24,8 +24,10 @@     -- Algebraic data types   , Dec   , Exp-  , Type(AppT, ArrowT)+  , Type(AppT, ForallT, VarT, ArrowT)   , Info(VarI)+  , TyVarBndr(PlainTV, KindedTV)+  , Pred(ClassP)     -- Lifted constructors     -- .. Literals   , stringL@@ -41,18 +43,16 @@   ) import Control.Distributed.Process.Internal.Types   ( RemoteTable-  , Closure(..)-  , Static(..)-  , StaticLabel(..)-  , Process-  , ProcessId+  , Static(Static)+  , StaticLabel(StaticLabel)   , remoteTableLabel-  , remoteTableDict-  , SerializableDict(..)-  , RuntimeSerializableSupport(..)+  , SerializableDict(SerializableDict)   )-import Control.Distributed.Process.Internal.Dynamic (Dynamic, toDyn)-import Control.Distributed.Process.Internal.Primitives (send, expect)+import Control.Distributed.Process.Internal.Dynamic +  ( Dynamic(..)+  , unsafeCoerce#+  , toDyn+  )  -------------------------------------------------------------------------------- -- User-level API                                                             --@@ -63,15 +63,29 @@ remotable :: [Name] -> Q [Dec]  remotable ns = do   (closures, inserts) <- unzip <$> mapM generateDefs ns-  rtable <- createMetaData inserts +  rtable <- createMetaData (concat inserts)   return $ concat closures ++ rtable  +-- | Construct a static value.+--+-- If @f : forall a1 .. an. T@ +-- then @$(mkStatic 'f) :: forall a1 .. an. Static T@. +-- Be sure to pass 'f' to 'remotable'. +mkStatic :: Name -> Q Exp+mkStatic = varE . staticName++-- | Serialization dictionary for a function argument (see module header)+functionSDict :: Name -> Q Exp+functionSDict = varE . sdictName++{- -- | Create a closure -- --- If @f :: a -> b@ then @mkClosure :: a -> Closure b@. Make sure to pass 'f'--- as an argument to 'remotable' too.+-- See module documentation header for "Control.Distributed.Process.Closure"+-- for a detailed explanation and examples. mkClosure :: Name -> Q Exp mkClosure = varE . closureName +-}  -------------------------------------------------------------------------------- -- Internal (Template Haskell)                                                --@@ -84,60 +98,88 @@       __remoteTable = $(compose is)     |] --- | Generate the necessary definitions for one function ------ Given an (f :: a -> b) in module M, create: ---  1. f__closure :: a -> Closure b,---  2. registerLabel "M.f" (toDyn ((f . enc) :: ByteString -> b))--- --- Moreover, if b is of the form Process c, then additionally create---  3. registerSender (Process c) (send :: ProcessId -> c -> Process ())-generateDefs :: Name -> Q ([Dec], Q Exp)+generateDefs :: Name -> Q ([Dec], [Q Exp]) generateDefs n = do-  serializableDict <- [t| SerializableDict |]   mType <- getType n   case mType of-    Just (origName, ArrowT `AppT` arg `AppT` res) -> do-      (closure, label) <- generateClosure origName (return arg) (return res)-      let decoder = generateDecoder origName (return res)-          insert  = [| registerLabel $(stringE label) (toDyn $decoder) |]-      return (closure, insert)-    Just (origName, sdict `AppT` a) | sdict == serializableDict -> -      return ([], [| registerSerializableDict $(varE n) |])  +    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 |] ]+               )++      (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+             )     _ -> -      fail $ "remotable: " ++ show n ++ " is not a function"-   --- | Generate the closure creator (see 'generateDefs')-generateClosure :: Name -> Q Type -> Q Type -> Q ([Dec], String)-generateClosure n arg res = do-    closure <- sequence -      [ sigD (closureName n) [t| $arg -> Closure $res |]-      , sfnD (closureName n) [| Closure (Static (UserStatic ($(stringE label)))) . encode |]  +      fail $ "remotable: " ++ show n ++ " not found"++registerStatic :: String -> Dynamic -> RemoteTable -> RemoteTable+registerStatic label dyn = remoteTableLabel label ^= Just dyn ++-- | Generate a static value +generateStatic :: Name -> [TyVarBndr] -> Type -> Q [Dec]+generateStatic n xs typ = do+    staticTyp <- [t| Static |]+    sequence+      [ sigD (staticName n) $ +          return (ForallT xs +                  (map typeable xs) +                  (staticTyp `AppT` typ)+          )+      , sfnD (staticName n) +          [| Static $ StaticLabel +               $(stringE (show n)) +               (typeOf (undefined :: $(return typ)))+           |]       ]-    return (closure, label)   where-    label :: String -    label = show $ n---- | Generate the decoder (see 'generateDefs')-generateDecoder :: Name -> Q Type -> Q Exp -generateDecoder n res = [| $(varE n) . decode :: ByteString -> $res |]+    typeable :: TyVarBndr -> Pred+    typeable (PlainTV v)    = ClassP (mkName "Typeable") [VarT v] +    typeable (KindedTV v _) = ClassP (mkName "Typeable") [VarT v] --- | The name for the function that generates the closure-closureName :: Name -> Name-closureName n = mkName $ nameBase n ++ "__closure"+-- | Generate a function serialization dictionary+generateSDict :: Name -> Type -> Q [Dec]+generateSDict n typ = do+    sequence+      [ sigD (sdictName n) $ [t| Static (SerializableDict $(return typ)) |]+      , sfnD (sdictName n) +         [| Static $ StaticLabel +              $(stringE (show (sdictName n))) +              (typeOf (undefined :: SerializableDict $(return typ)))+          |]+      ] -registerLabel :: String -> Dynamic -> RemoteTable -> RemoteTable-registerLabel label dyn = remoteTableLabel label ^= Just dyn +staticName :: Name -> Name+staticName n = mkName $ nameBase n ++ "__static" -registerSerializableDict :: forall a. SerializableDict a -> RemoteTable -> RemoteTable-registerSerializableDict SerializableDict = -  let rss = RuntimeSerializableSupport {-                rssSend   = toDyn (send :: ProcessId -> a -> Process ()) -              , rssReturn = toDyn (return . decode :: ByteString -> Process a)  -              , rssExpect = toDyn (expect :: Process a)-              }-  in remoteTableDict (typeOf (undefined :: a)) ^= Just rss +sdictName :: Name -> Name+sdictName n = mkName $ nameBase n ++ "__sdict"  -------------------------------------------------------------------------------- -- Generic Template Haskell auxiliary functions                               --@@ -164,4 +206,3 @@ -- | Variation on 'funD' which takes a single expression to define the function sfnD :: Name -> Q Exp -> Q Dec sfnD n e = funD n [clause [] (normalB e) []] -

@@ -13,14 +13,27 @@   , dynTypeRep   , dynApply   , dynApp+  , dynBind+  , dynBind'+  , dynKleisli   , GHC.unsafeCoerce#   ) where -import Data.Typeable (Typeable, TypeRep, typeOf, funResultTy)+import Data.Typeable +  ( Typeable+  , TypeRep+  , typeOf+  , funResultTy+  , TyCon+  , splitTyConApp+  , mkFunTy+  )+import Data.Typeable.Internal (funTc) import qualified GHC.Prim as GHC (Any, unsafeCoerce#) import Data.Maybe (fromMaybe)  data Dynamic = Dynamic TypeRep GHC.Any+  deriving Typeable  toDyn :: Typeable a => a -> Dynamic toDyn x = Dynamic (typeOf x) (GHC.unsafeCoerce# x)@@ -55,3 +68,45 @@     typeError  = "Type error in dynamic application.\n"                ++ "Can't apply function " ++ show f               ++ " to argument " ++ show x++dynBind :: TyCon -> (forall a b. m a -> (a -> m b) -> m b) -> Dynamic -> Dynamic -> Maybe Dynamic+dynBind m bind (Dynamic t1 x) (Dynamic t2 f) = +  case splitTyConApp t1 of+    (m', [a]) | m' == m ->+      case funResultTy t2 a of+        Just mb -> Just (Dynamic mb (GHC.unsafeCoerce# bind x f))+        _  -> Nothing+    _ -> Nothing ++dynBind' :: TyCon -> (forall a b. m a -> (a -> m b) -> m b) -> Dynamic -> Dynamic -> Dynamic+dynBind' m bind x f = fromMaybe (error typeError) (dynBind m bind x f)+  where+    typeError = "Type error in dynamic bind.\nCan't bind " ++ show x ++ " to " ++ show f+++++-- | Dynamically typed Kleisli composition+dynKleisli :: TyCon  +           -> (forall a b c. (a -> m b) -> (b -> m c) -> a -> m c) +           -> Dynamic +           -> Dynamic +           -> Maybe Dynamic+dynKleisli m comp (Dynamic tf f) (Dynamic tg g) = +  case splitTyConApp tf of+    (arr1, [a, mb]) | arr1 == funTc -> +      case splitTyConApp tg of+        (arr2, [b, mc]) | arr2 == funTc ->+          case splitTyConApp mb of+            (m', [b']) | m' == m && b' == b ->+              case splitTyConApp mc of+                (m'', [_c]) | m'' == m ->+                  Just (Dynamic (mkFunTy a mc) (GHC.unsafeCoerce# comp f g))  +                _ -> +                  Nothing+            _ ->+              Nothing+        _ ->+          Nothing+    _ ->+      Nothing

@@ -41,6 +41,8 @@   , whereisRemote   , whereisRemoteAsync   , nsendRemote+    -- * Closures+  , unClosure     -- * Auxiliary API   , catch   , expectTimeout@@ -55,7 +57,7 @@  import Prelude hiding (catch) import Data.Binary (decode)-import Data.Typeable (Typeable)+import Data.Typeable (Typeable, typeOf) import Data.Time.Clock (getCurrentTime) import Data.Time.Format (formatTime) import System.Locale (defaultTimeLocale)@@ -108,6 +110,7 @@   , DidUnlinkPort(..)   , WhereIsReply(..)   , createMessage+  , Static(..)   ) import Control.Distributed.Process.Internal.MessageT    ( sendMessage@@ -115,6 +118,8 @@   , getLocalNode   )   import Control.Distributed.Process.Internal.Node (runLocalProcess)+import Control.Distributed.Process.Internal.Closure.Resolution (resolveClosure)+import Control.Distributed.Process.Internal.Dynamic (fromDyn, dynTypeRep)  -------------------------------------------------------------------------------- -- Basic messaging                                                            --@@ -444,6 +449,22 @@ nsendRemote :: Serializable a => NodeId -> String -> a -> Process () nsendRemote nid label msg =    sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))++--------------------------------------------------------------------------------+-- Closures                                                                   --+--------------------------------------------------------------------------------++-- | Deserialize a closure+unClosure :: forall a. Typeable a => Closure a -> Process a+unClosure (Closure (Static label) env) = do+    rtable <- remoteTable <$> procMsg getLocalNode +    case resolveClosure rtable label env of+      Nothing  -> throw . userError $ "Unregistered closure " ++ show label+      Just dyn -> return $ fromDyn dyn (throw (typeError dyn))+  where+    typeError dyn = userError $ "lookupStatic type error: " +                 ++ "cannot match " ++ show (dynTypeRep dyn) +                 ++ " against " ++ show (typeOf (undefined :: a))  -------------------------------------------------------------------------------- -- Auxiliary functions                                                        --

@@ -27,10 +27,13 @@     -- * Closures   , StaticLabel(..)   , Static(..) +  , staticApply+  , staticDuplicate+  , staticTypeOf+  , typeOfStaticLabel   , Closure(..)   , RemoteTable(..)   , SerializableDict(..)-  , RuntimeSerializableSupport(..)     -- * Messages    , Message(..)   , createMessage@@ -69,14 +72,13 @@   , typedChannels   , typedChannelWithId   , remoteTableLabels-  , remoteTableDicts   , remoteTableLabel-  , remoteTableDict   ) where  import Data.Map (Map) import Data.Int (Int32)-import Data.Typeable (Typeable, TypeRep)+import Data.Maybe (fromJust)+import Data.Typeable (Typeable, TypeRep, typeOf, funResultTy) import Data.Binary (Binary(put, get), putWord8, getWord8, encode) import qualified Data.ByteString as BSS (ByteString, concat) import qualified Data.ByteString.Lazy as BSL @@ -110,6 +112,7 @@   ) import Control.Distributed.Process.Internal.CQueue (CQueue) import Control.Distributed.Process.Internal.Dynamic (Dynamic) +import Control.Distributed.Process.Internal.TypeRep () -- Binary instances  -------------------------------------------------------------------------------- -- Node and process identifiers                                               --@@ -253,49 +256,49 @@   | ReceivePortBiased [ReceivePort a]      -- | A round-robin combination of receive ports   | ReceivePortRR (TVar [ReceivePort a]) +  deriving Typeable  -------------------------------------------------------------------------------- -- Closures                                                                   -- -------------------------------------------------------------------------------- -data StaticLabel = -    UserStatic String-  -- Built-in closures -  | ClosureReturn-  | ClosureSend -  | ClosureExpect-  -- Generic closure combinators-  | ClosureApply-  | ClosureConst-  | ClosureUnit-  -- Arrow combinators for processes-  | CpId-  | CpComp -  | CpFirst-  | CpSwap-  | CpCopy-  | CpLeft-  | CpMirror-  | CpUntag-  | CpApply+data StaticLabel =+    StaticLabel String TypeRep+  | StaticApply StaticLabel StaticLabel+  | StaticDuplicate StaticLabel TypeRep   deriving (Typeable, Show) --- | A static value is one that is bound at top-level.+-- | A static value is top-level bound or the application of two static values newtype Static a = Static StaticLabel    deriving (Typeable, Show) +-- | Apply two static values+staticApply :: Static (a -> b) -> Static a -> Static b+staticApply (Static f) (Static x) = Static (StaticApply f x)++-- | Co-monadic 'duplicate' for static values+staticDuplicate :: forall a. Typeable a => Static a -> Static (Static a)+staticDuplicate (Static x) = +  Static (StaticDuplicate x (typeOf (undefined :: Static a)))++staticTypeOf :: forall a. Typeable a => a -> Static a +staticTypeOf _ = Static (StaticLabel "undefined" (typeOf (undefined :: a)))++typeOfStaticLabel :: StaticLabel -> TypeRep+typeOfStaticLabel (StaticLabel _ typ) +  = typ +typeOfStaticLabel (StaticApply f x) +  = fromJust $ funResultTy (typeOfStaticLabel f) (typeOfStaticLabel x)+typeOfStaticLabel (StaticDuplicate _ typ)+  = typ+ -- | A closure is a static value and an encoded environment data Closure a = Closure (Static (BSL.ByteString -> a)) BSL.ByteString   deriving (Typeable, Show) --- | Used to fake 'static' (see paper)+-- | Runtime dictionary for 'unstatic' lookups  data RemoteTable = RemoteTable {-    -- | If the user creates a closure of type @a -> Closure b@ from a function-    -- @f : a -> b@, then '_remoteTableLabels' should have an entry for "f"-    -- of type @ByteString -> b@ (basically, @f . encode@)-    _remoteTableLabels  :: Map String Dynamic -    -- | Runtime counterpart to SerializableDict -  , _remoteTableDicts   :: Map TypeRep RuntimeSerializableSupport +    _remoteTableLabels :: Map String Dynamic    }  -- | Reification of 'Serializable' (see "Control.Distributed.Process.Closure")@@ -303,17 +306,6 @@     SerializableDict :: Serializable a => SerializableDict a   deriving (Typeable) --- | Runtime support for implementing "polymorphic" functions with a --- Serializable qualifier (sendClosure, returnClosure, ..). ------ We don't attempt to keep this minimal, but instead just add functions as--- convenient. This will be replaced anyway once 'static' has been implemented.-data RuntimeSerializableSupport = RuntimeSerializableSupport {-   rssSend   :: Dynamic- , rssReturn :: Dynamic- , rssExpect :: Dynamic- }- -------------------------------------------------------------------------------- -- Messages                                                                   -- --------------------------------------------------------------------------------@@ -554,42 +546,16 @@       _ -> fail "Identifier.get: invalid"  instance Binary StaticLabel where-  put (UserStatic string) = putWord8 0 >> put string-  put ClosureReturn = putWord8 1-  put ClosureSend   = putWord8 2-  put ClosureExpect = putWord8 3-  put ClosureApply  = putWord8 4-  put ClosureConst  = putWord8 5-  put ClosureUnit   = putWord8 6-  put CpId          = putWord8 7-  put CpComp        = putWord8 8-  put CpFirst       = putWord8 9-  put CpSwap        = putWord8 10-  put CpCopy        = putWord8 11 -  put CpLeft        = putWord8 12-  put CpMirror      = putWord8 13-  put CpUntag       = putWord8 14-  put CpApply       = putWord8 15+  put (StaticLabel string typ)    = putWord8 0 >> put string >> put typ +  put (StaticApply label1 label2) = putWord8 1 >> put label1 >> put label2+  put (StaticDuplicate label typ) = putWord8 2 >> put label >> put typ   get = do     header <- getWord8     case header of-      0  -> UserStatic <$> get-      1  -> return ClosureReturn-      2  -> return ClosureSend -      3  -> return ClosureExpect -      4  -> return ClosureApply-      5  -> return ClosureConst-      6  -> return ClosureUnit-      7  -> return CpId-      8  -> return CpComp -      9  -> return CpFirst-      10 -> return CpSwap-      11 -> return CpCopy-      12 -> return CpLeft-      13 -> return CpMirror-      14 -> return CpUntag-      15 -> return CpApply-      _  -> fail "StaticLabel.get: invalid"+      0 -> StaticLabel <$> get <*> get+      1 -> StaticApply <$> get <*> get+      2 -> StaticDuplicate <$> get <*> get+      _ -> fail "StaticLabel.get: invalid"   instance Binary WhereIsReply where   put (WhereIsReply label mPid) = put label >> put mPid@@ -629,14 +595,8 @@ remoteTableLabels :: Accessor RemoteTable (Map String Dynamic) remoteTableLabels = accessor _remoteTableLabels (\ls tbl -> tbl { _remoteTableLabels = ls }) -remoteTableDicts :: Accessor RemoteTable (Map TypeRep RuntimeSerializableSupport)-remoteTableDicts = accessor _remoteTableDicts (\ds tbl -> tbl { _remoteTableDicts = ds })- remoteTableLabel :: String -> Accessor RemoteTable (Maybe Dynamic) remoteTableLabel label = remoteTableLabels >>> DAC.mapMaybe label--remoteTableDict :: TypeRep -> Accessor RemoteTable (Maybe RuntimeSerializableSupport)-remoteTableDict label = remoteTableDicts >>> DAC.mapMaybe label  -------------------------------------------------------------------------------- -- MessageT monad                                                             --

@@ -105,6 +105,7 @@   , typedChannelWithId   , WhereIsReply(..)   , messageToPayload+  , RemoteTable(..)   ) import Control.Distributed.Process.Serializable (Serializable) import Control.Distributed.Process.Internal.MessageT @@ -117,16 +118,20 @@   , createMessage   ) import Control.Distributed.Process.Internal.Dynamic (fromDynamic)-import Control.Distributed.Process.Internal.Closure -  ( initRemoteTable-  , resolveClosure-  )+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)  -------------------------------------------------------------------------------- -- Initialization                                                             -- --------------------------------------------------------------------------------++initRemoteTable :: RemoteTable+initRemoteTable = Static.__remoteTable+                . CP.__remoteTable +                $ RemoteTable Map.empty  -- | Initialize a new local node.  -- 

@@ -1,7 +1,8 @@ module Main where -import qualified Data.ByteString.Lazy as BSL (empty)-import Control.Monad (join, replicateM)+import Data.ByteString.Lazy (empty)+import Data.Typeable (Typeable, typeOf)+import Control.Monad (join, replicateM, forever) import Control.Exception (IOException, throw) import Control.Concurrent (forkIO, threadDelay) import Control.Concurrent.MVar (MVar, newEmptyMVar, readMVar, takeMVar, putMVar)@@ -14,38 +15,82 @@ import Control.Distributed.Process.Internal.Types (Closure(..), Static(..), StaticLabel(..)) import TestAuxiliary -serializableDictInt :: SerializableDict Int-serializableDictInt = SerializableDict+sdictInt :: SerializableDict Int+sdictInt = SerializableDict +factorial :: Int -> Process Int+factorial 0 = return 1+factorial n = (n *) <$> factorial (n - 1) + addInt :: Int -> Int -> Int addInt x y = x + y  putInt :: Int -> MVar Int -> IO () putInt = flip putMVar -sendInt :: ProcessId -> Int -> Process ()-sendInt = send- sendPid :: ProcessId -> Process () sendPid toPid = do   fromPid <- getSelfPid   send toPid fromPid -factorial :: Int -> Process Int-factorial 0 = return 1-factorial n = (n *) <$> factorial (n - 1) +wait :: Int -> Process ()+wait = liftIO . threadDelay -factorialOf :: () -> Int -> Process Int-factorialOf () = factorial+-- | First argument indicates empty closure environment+typedPingServer :: () -> ReceivePort (SendPort ()) -> Process ()+typedPingServer () rport = forever $ do+  sport <- receiveChan rport+  sendChan sport () -returnForTestApply :: (Closure (Int -> Process Int), Int) -> (Closure (Int -> Process Int), Int)-returnForTestApply = id +remotable [ 'factorial+          , 'addInt+          , 'putInt+          , 'sendPid+          , 'sdictInt+          , 'wait+          , 'typedPingServer+          ] -wait :: Int -> Process ()-wait = liftIO . threadDelay+factorialClosure :: Int -> Closure (Process Int)+factorialClosure = $(mkClosure 'factorial)  -$(remotable ['addInt, 'putInt, 'sendInt, 'sendPid, 'factorial, 'factorialOf, 'returnForTestApply, 'wait, 'serializableDictInt])+addIntClosure :: Int -> Closure (Int -> Int)+addIntClosure = $(mkClosure 'addInt)  +putIntClosure :: Int -> Closure (MVar Int -> IO ())+putIntClosure = $(mkClosure 'putInt) ++sendPidClosure :: ProcessId -> Closure (Process ())+sendPidClosure = $(mkClosure 'sendPid) ++sendFac :: Int -> ProcessId -> Closure (Process ())+sendFac n pid = factorialClosure n `cpBind` cpSend $(mkStatic 'sdictInt) pid ++factorialOf :: Closure (Int -> Process Int)+factorialOf = staticClosure $(mkStatic 'factorial)++factorial' :: Int -> Closure (Process Int)+factorial' n = cpReturn $(mkStatic 'sdictInt) n `cpBind` factorialOf ++waitClosure :: Int -> Closure (Process ())+waitClosure = $(mkClosure 'wait) ++testUnclosure :: Transport -> RemoteTable -> IO ()+testUnclosure transport rtable = do+  node <- newLocalNode transport rtable+  runProcess node $ do+    120 <- join . unClosure $ factorialClosure 5+    return ()++testBind :: Transport -> RemoteTable -> IO ()+testBind transport rtable = do+  node <- newLocalNode transport rtable+  runProcess node $ do+    us <- getSelfPid+    join . unClosure $ sendFac 6 us +    (720 :: Int) <- expect +    return ()+ testSendPureClosure :: Transport -> RemoteTable -> IO () testSendPureClosure transport rtable = do   serverAddr <- newEmptyMVar@@ -56,14 +101,14 @@     addr <- forkProcess node $ do       cl <- expect       fn <- unClosure cl :: Process (Int -> Int)-      11 <- return $ fn 6+      13 <- return $ fn 6       liftIO $ putMVar serverDone ()      putMVar serverAddr addr     forkIO $ do     node <- newLocalNode transport rtable      theirAddr <- readMVar serverAddr-    runProcess node $ send theirAddr ($(mkClosure 'addInt) 5) +    runProcess node $ send theirAddr (addIntClosure 7)    takeMVar serverDone @@ -87,7 +132,7 @@   forkIO $ do     node <- newLocalNode transport rtable      theirAddr <- readMVar serverAddr-    runProcess node $ send theirAddr ($(mkClosure 'putInt) 5) +    runProcess node $ send theirAddr (putIntClosure 5)     takeMVar serverDone @@ -109,7 +154,7 @@     theirAddr <- readMVar serverAddr     runProcess node $ do       pid <- getSelfPid-      send theirAddr ($(mkClosure 'sendInt) pid) +      send theirAddr (cpSend $(mkStatic 'sdictInt) pid)        5 <- expect :: Process Int       liftIO $ putMVar clientDone () @@ -129,7 +174,7 @@     nid <- readMVar serverNodeAddr     runProcess node $ do       pid   <- getSelfPid-      pid'  <- spawn nid ($(mkClosure 'sendPid) pid)+      pid'  <- spawn nid (sendPidClosure pid)       pid'' <- expect       True <- return $ pid' == pid''       liftIO $ putMVar clientDone ()@@ -149,26 +194,11 @@     node <- newLocalNode transport rtable     nid <- readMVar serverNodeAddr     runProcess node $ do-      (120 :: Int) <- call serializableDictInt nid ($(mkClosure 'factorial) 5)+      (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorialClosure 5)       liftIO $ putMVar clientDone ()    takeMVar clientDone -sendFac :: Int -> ProcessId -> Closure (Process ())-sendFac n pid = $(mkClosure 'factorial) n `cpBind` $(mkClosure 'sendInt) pid--factorial' :: Int -> Closure (Process Int)-factorial' n = returnClosure serializableDictInt n `cpBind` $(mkClosure 'factorialOf) ()--testBind :: Transport -> RemoteTable -> IO ()-testBind transport rtable = do-  node <- newLocalNode transport rtable-  runProcess node $ do-    us <- getSelfPid-    join . unClosure $ sendFac 6 us -    (720 :: Int) <- expect -    return ()- testCallBind :: Transport -> RemoteTable -> IO () testCallBind transport rtable = do   serverNodeAddr <- newEmptyMVar@@ -182,7 +212,7 @@     node <- newLocalNode transport rtable     nid <- readMVar serverNodeAddr     runProcess node $ do-      (120 :: Int) <- call serializableDictInt nid (factorial' 5)+      (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorial' 5)       liftIO $ putMVar clientDone ()    takeMVar clientDone@@ -197,19 +227,6 @@     720 :: Int <- expect     return () -testApply :: Transport -> RemoteTable -> IO ()-testApply transport rtable = do-    node <- newLocalNode transport rtable-    runProcess node $ do-      (120 :: Int) <- join (unClosure bar)-      return ()-  where-    bar :: Closure (Process Int)-    bar = closureApply cpApply foo--    foo :: Closure (Closure (Int -> Process Int), Int)-    foo = $(mkClosure 'returnForTestApply) ($(mkClosure 'factorialOf) (), 5) - -- Test 'spawnSupervised' -- -- Set up a supervisor, spawn a child, then have a third process monitor the@@ -225,7 +242,7 @@     forkProcess node1 $ do       us <- getSelfPid       liftIO $ putMVar superPid us-      (child, _ref) <- spawnSupervised (localNodeId node2) ($(mkClosure 'wait) 1000000) +      (child, _ref) <- spawnSupervised (localNodeId node2) (waitClosure 1000000)        liftIO $ do         putMVar childPid child         threadDelay 500000 -- Give the child a chance to link to us@@ -249,7 +266,7 @@ testSpawnInvalid transport rtable = do   node <- newLocalNode transport rtable   runProcess node $ do-    (pid, ref) <- spawnMonitor (localNodeId node) (Closure (Static (UserStatic "ThisDoesNotExist")) BSL.empty)+    (pid, ref) <- spawnMonitor (localNodeId node) (Closure (Static (StaticLabel "ThisDoesNotExist" (typeOf ()))) empty)     ProcessMonitorNotification ref' pid' _reason <- expect      -- Depending on the exact interleaving, reason might be NoProc or the exception thrown by the absence of the static closure     True <- return $ ref' == ref && pid == pid'  @@ -261,26 +278,40 @@   runProcess node $ do     nodeId <- getSelfNode     us     <- getSelfPid-    them   <- spawn nodeId $ expectClosure serializableDictInt `cpBind` $(mkClosure 'sendInt) us+    them   <- spawn nodeId $ cpExpect $(mkStatic 'sdictInt) `cpBind` cpSend $(mkStatic 'sdictInt) us     send them (1234 :: Int)     (1234 :: Int) <- expect     return () +testSpawnChannel :: Transport -> RemoteTable -> IO ()+testSpawnChannel transport rtable = do+  [node1, node2] <- replicateM 2 $ newLocalNode transport rtable++  runProcess node1 $ do+    pingServer <- spawnChannel +                    (sdictSendPort sdictUnit)+                    (localNodeId node2)  +                    ($(mkClosure 'typedPingServer) ())+    (sendReply, receiveReply) <- newChan+    sendChan pingServer sendReply+    receiveChan receiveReply+ main :: IO () main = do   Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters   let rtable = __remoteTable initRemoteTable    runTests -    [ ("SendPureClosure", testSendPureClosure transport rtable)+    [ ("Unclosure",       testUnclosure       transport rtable)+    , ("Bind",            testBind            transport rtable)+    , ("SendPureClosure", testSendPureClosure transport rtable)     , ("SendIOClosure",   testSendIOClosure   transport rtable)     , ("SendProcClosure", testSendProcClosure transport rtable)     , ("Spawn",           testSpawn           transport rtable)     , ("Call",            testCall            transport rtable)-    , ("Bind",            testBind            transport rtable)     , ("CallBind",        testCallBind        transport rtable)     , ("Seq",             testSeq             transport rtable)-    , ("Apply",           testApply           transport rtable)     , ("SpawnSupervised", testSpawnSupervised transport rtable)     , ("SpawnInvalid",    testSpawnInvalid    transport rtable)     , ("ClosureExpect",   testClosureExpect   transport rtable)+    , ("SpawnChannel",    testSpawnChannel    transport rtable)     ]