git.iko.soy/distributed-process
on master help sign in
Start tidying up the code

Although we'll still have a central Types file, all other functionality will be in separate files. This might require some recursive modules -- so be it. Also, we should make sure that the functions we export from public modules (i.e., those not in Internal/) are in fact designed to be in the public API. This is only the first step towards this goal. To be continued.

commit03fd8b4596d18a524bfc706c362fec601e61fc9a authorEdsko de Vries <edsko@edsko.net> dateWed 27 Jun 2012, 17:50 (14 years ago) parent9f46a9f
8 files changed+418−356
compare with master
unifiedsplit
distributed-process/src/Control/Distributed/Process.hs+2−2 distributed-process/src/Control/Distributed/Process/Closure.hs+7−161 distributed-process/src/Control/Distributed/Process/Internal/Closure.hs+138 distributed-process/src/Control/Distributed/Process/Internal/Closure/TH.hs+176 distributed-process/src/Control/Distributed/Process/Internal/MessageT.hs+55−8 distributed-process/src/Control/Distributed/Process/Internal/Node.hs+16 distributed-process/src/Control/Distributed/Process/Internal/Types.hs+16−179 distributed-process/src/Control/Distributed/Process/Node.hs+8−6
distributed-process/src/Control/Distributed/Process.hs+2−2
@@ -91,7 +91,6 @@ import Control.Distributed.Process.Internal.Types
9191 , LocalNode(..)
9292 , LocalProcess(..)
9393 , Process(..)
94− , runLocalProcess
9594 , Closure(..)
9695 , StaticLabel(Call)
9796 , Static(..)
@@ -118,7 +117,6 @@ import Control.Distributed.Process.Internal.Types
118117 , TypedChannel(..)
119118 , ChannelId(..)
120119 , Identifier(..)
121− , resolveClosure
122120 )
123121import Control.Distributed.Process.Internal.MessageT
124122 ( sendMessage
@@ -126,6 +124,8 @@ import Control.Distributed.Process.Internal.MessageT
126124 , getLocalNode
127125 )
128126import Control.Distributed.Process.Internal.Dynamic (fromDyn, dynTypeRep)
127+import Control.Distributed.Process.Internal.Closure (resolveClosure)
128+import Control.Distributed.Process.Internal.Node (runLocalProcess)
129129
130130-- INTERNAL NOTES
131131--
distributed-process/src/Control/Distributed/Process/Closure.hs+7−161
@@ -38,7 +38,6 @@
3838-- > $ initRemoteTable
3939--
4040-- See Section 6, /Faking It/, of /Towards Haskell in the Cloud/ for more info.
41−{-# LANGUAGE TemplateHaskell #-}
4241module Control.Distributed.Process.Closure
4342 ( -- * Primitive closures
4443 remotable
@@ -72,65 +71,16 @@ module Control.Distributed.Process.Closure
7271 ) where
7372
7473import Prelude hiding (lookup)
75−import Data.ByteString.Lazy (ByteString, empty)
76−import Data.Binary (encode, decode)
74+import qualified Data.ByteString.Lazy as BS (empty)
75+import Data.Binary (encode)
7776import Data.Typeable (typeOf, Typeable)
78−import Control.Applicative ((<$>))
79−import Language.Haskell.TH
80− ( -- Q monad and operations
81− Q
82− , reify
83− -- Names
84− , Name
85− , mkName
86− , nameBase
87− -- Algebraic data types
88− , Dec
89− , Exp
90− , Type(AppT, ArrowT)
91− , Info(VarI)
92− -- Lifted constructors
93− -- .. Literals
94− , stringL
95− -- .. Patterns
96− , normalB
97− , clause
98− -- .. Expressions
99− , varE
100− , litE
101− -- .. Top-level declarations
102− , funD
103− , sigD
104− )
10577import Control.Distributed.Process.Internal.Types
106− ( RemoteTable
107− , Closure(..)
78+ ( Closure(..)
10879 , Static(..)
10980 , StaticLabel(..)
11081 , Process
111− , ProcessId
112− , CallReply(..)
113− , registerLabel
114− , registerSender
11582 )
116−import Control.Distributed.Process (send)
117−import Control.Distributed.Process.Internal.Dynamic (toDyn)
118−
119−--------------------------------------------------------------------------------
120−-- Top-level API --
121−--------------------------------------------------------------------------------
122−
123−-- | Create the closure, decoder, and metadata definitions for the given list
124−-- of functions
125−remotable :: [Name] -> Q [Dec]
126−remotable ns = do
127− (closures, inserts) <- unzip <$> mapM generateDefs ns
128− rtable <- createMetaData (concat inserts)
129− return $ concat closures ++ rtable
130−
131−-- | Create a closure
132−mkClosure :: Name -> Q Exp
133−mkClosure = varE . closureName
83+import Control.Distributed.Process.Internal.Closure.TH (remotable, mkClosure)
13484
13585--------------------------------------------------------------------------------
13686-- Generic closure combinators --
@@ -148,7 +98,7 @@ closureConst = Closure (Static ClosureConst) (encode $ typeOf aux)
14898 aux = undefined
14999
150100closureUnit :: Closure ()
151−closureUnit = Closure (Static ClosureUnit) empty
101+closureUnit = Closure (Static ClosureUnit) BS.empty
152102
153103--------------------------------------------------------------------------------
154104-- Arrow combinators for processes --
@@ -175,10 +125,10 @@ cpComp :: forall a b c. (Typeable a, Typeable b, Typeable c)
175125 => CP a b -> CP b c -> CP a c
176126cpComp f g = comp `closureApply` f `closureApply` g
177127 where
178− comp :: Closure ((a -> Process b) -> (b -> Process c) -> (a -> Process c))
128+ comp :: Closure ((a -> Process b) -> (b -> Process c) -> a -> Process c)
179129 comp = Closure (Static CpComp) (encode $ typeOf aux)
180130
181− aux :: (a -> Process b) -> (b -> Process c) -> (a -> Process c)
131+ aux :: (a -> Process b) -> (b -> Process c) -> a -> Process c
182132 aux = undefined
183133
184134cpFirst :: forall a b c. (Typeable a, Typeable b, Typeable c)
@@ -273,107 +223,3 @@ cpBind x f = cpElim $ cpIntro x `cpComp` f
273223
274224cpSeq :: Closure (Process ()) -> Closure (Process ()) -> Closure (Process ())
275225cpSeq p q = p `cpBind` cpIntro q
276−
277−--------------------------------------------------------------------------------
278−-- Internal (Template Haskell) --
279−--------------------------------------------------------------------------------
280−
281−-- | Generate the code to add the metadata to the CH runtime
282−createMetaData :: [Q Exp] -> Q [Dec]
283−createMetaData is =
284− [d| __remoteTable :: RemoteTable -> RemoteTable ;
285− __remoteTable = $(compose is)
286− |]
287−
288−-- | Generate the necessary definitions for one function
289−--
290−-- Given an (f :: a -> b) in module M, create:
291−-- 1. f__closure :: a -> Closure b,
292−-- 2. registerLabel "M.f" (toDyn ((f . enc) :: ByteString -> b))
293−--
294−-- Moreover, if b is of the form Process c, then additionally create
295−-- 3. registerSender (Process c) (send :: ProcessId -> c -> Process ())
296−generateDefs :: Name -> Q ([Dec], [Q Exp])
297−generateDefs n = do
298− mType <- getType n
299− case mType of
300− Just (origName, ArrowT `AppT` arg `AppT` res) -> do
301− (closure, label) <- generateClosure origName (return arg) (return res)
302− let decoder = generateDecoder origName (return res)
303− insert = [| registerLabel $(stringE label) (toDyn $decoder) |]
304−
305− -- Generate special closure to support 'call'
306− mResult <- processResult res
307− insert' <- case mResult of
308− Nothing ->
309− return []
310− Just result -> do
311− let sender = generateSender result
312− return . return $
313− [| registerSender $(typeToTypeRep res) (toDyn $sender) |]
314−
315− return (closure, insert : insert')
316− _ ->
317− fail $ "remotable: " ++ show n ++ " is not a function"
318−
319−-- | Generate the closure creator (see 'generateDefs')
320−generateClosure :: Name -> Q Type -> Q Type -> Q ([Dec], String)
321−generateClosure n arg res = do
322− closure <- sequence
323− [ sigD (closureName n) [t| $arg -> Closure $res |]
324− , sfnD (closureName n) [| Closure (Static (UserStatic ($(stringE label)))) . encode |]
325− ]
326− return (closure, label)
327− where
328− label :: String
329− label = show $ n
330−
331−-- | If 't' is of the form 'Process b', return 'Just b'
332−processResult :: Type -> Q (Maybe (Q Type))
333−processResult t = do
334− process <- [t| Process |]
335− case t of
336− p `AppT` a | p == process -> return . Just . return $ a
337− _ -> return Nothing
338−
339−-- | Generate the decoder (see 'generateDefs')
340−generateDecoder :: Name -> Q Type -> Q Exp
341−generateDecoder n res = [| $(varE n) . decode :: ByteString -> $res |]
342−
343−-- | Generate the sender. This is necessary to support 'call'
344−generateSender :: Q Type -> Q Exp
345−generateSender tp =
346− [| (\pid -> send pid . CallReply) :: ProcessId -> $tp -> Process () |]
347−
348−-- | The name for the function that generates the closure
349−closureName :: Name -> Name
350−closureName n = mkName $ nameBase n ++ "__closure"
351−
352−--------------------------------------------------------------------------------
353−-- Generic Template Haskell auxiliary functions --
354−--------------------------------------------------------------------------------
355−
356−typeToTypeRep :: Type -> Q Exp
357−typeToTypeRep tp = [| typeOf (undefined :: $(return tp)) |]
358−
359−-- | Compose a set of expressions
360−compose :: [Q Exp] -> Q Exp
361−compose [] = [| id |]
362−compose [e] = e
363−compose (e:es) = [| $e . $(compose es) |]
364−
365−-- | Literal string as an expression
366−stringE :: String -> Q Exp
367−stringE = litE . stringL
368−
369−-- | Look up the "original name" (module:name) and type of a top-level function
370−getType :: Name -> Q (Maybe (Name, Type))
371−getType name = do
372− info <- reify name
373− case info of
374− VarI origName typ _ _ -> return $ Just (origName, typ)
375− _ -> return Nothing
376−
377−-- | Variation on 'funD' which takes a single expression to define the function
378−sfnD :: Name -> Q Exp -> Q Dec
379−sfnD n e = funD n [clause [] (normalB e) []]
distributed-process/src/Control/Distributed/Process/Internal/Closure.hsnew file+138
@@ -0,0 +1,138 @@
1+{-# LANGUAGE MagicHash #-}
2+module Control.Distributed.Process.Internal.Closure
3+ ( -- * Runtime support
4+ initRemoteTable
5+ , registerLabel
6+ , registerSender
7+ , resolveClosure
8+ ) where
9+
10+import qualified Data.Map as Map (empty)
11+import Data.Accessor ((^=), (^.))
12+import Data.Typeable (TypeRep, TyCon, typeRepTyCon, typeOf)
13+import Data.ByteString.Lazy (ByteString)
14+import Data.Binary (decode)
15+import Control.Distributed.Process.Internal.Types
16+ ( RemoteTable(RemoteTable)
17+ , remoteTableLabel
18+ , remoteTableSender
19+ , StaticLabel(..)
20+ , ProcessId
21+ , Process
22+ , Closure(Closure)
23+ , Static(Static)
24+ )
25+import Control.Distributed.Process.Internal.Dynamic
26+ ( Dynamic(Dynamic)
27+ , dynTypeRep
28+ , toDyn
29+ , dynApp
30+ , dynBind
31+ , dynApply
32+ , unsafeCoerce#
33+ )
34+import Control.Distributed.Process.Internal.TypeRep () -- Binary instances
35+
36+--------------------------------------------------------------------------------
37+-- Runtime support for closures --
38+--------------------------------------------------------------------------------
39+
40+-- | Initial (empty) remote-call meta data
41+initRemoteTable :: RemoteTable
42+initRemoteTable = RemoteTable Map.empty Map.empty
43+
44+registerLabel :: String -> Dynamic -> RemoteTable -> RemoteTable
45+registerLabel label dyn = remoteTableLabel label ^= Just dyn
46+
47+registerSender :: TypeRep -> Dynamic -> RemoteTable -> RemoteTable
48+registerSender typ dyn = remoteTableSender typ ^= Just dyn
49+
50+resolveClosure :: RemoteTable -> StaticLabel -> ByteString -> Maybe Dynamic
51+-- Special support for call
52+resolveClosure rtable Call env = do
53+ proc <- resolveClosure rtable label env'
54+ sender <- rtable ^. remoteTableSender (dynTypeRep proc)
55+ proc `bind` (sender `dynApp` toDyn (pid :: ProcessId))
56+ where
57+ (label, env', pid) = decode env
58+ bind = dynBind tyConProcess bindProcess
59+ bindProcess :: Process a -> (a -> Process b) -> Process b
60+ bindProcess = (>>=)
61+-- Generic closure combinators
62+resolveClosure rtable ClosureApply env = do
63+ f <- resolveClosure rtable labelf envf
64+ x <- resolveClosure rtable labelx envx
65+ f `dynApply` x
66+ where
67+ (labelf, envf, labelx, envx) = decode env
68+resolveClosure _rtable ClosureConst env =
69+ return $ Dynamic (decode env) (unsafeCoerce# const)
70+resolveClosure _rtable ClosureUnit _env =
71+ return $ toDyn ()
72+-- Arrow combinators
73+resolveClosure _rtable CpId env =
74+ return $ Dynamic (decode env) (unsafeCoerce# cpId)
75+ where
76+ cpId :: forall a. a -> Process a
77+ cpId = return
78+resolveClosure _rtable CpComp env =
79+ return $ Dynamic (decode env) (unsafeCoerce# cpComp)
80+ where
81+ cpComp :: forall a b c. (a -> Process b) -> (b -> Process c) -> a -> Process c
82+ cpComp p q a = p a >>= q
83+resolveClosure _rtable CpFirst env =
84+ return $ Dynamic (decode env) (unsafeCoerce# cpFirst)
85+ where
86+ cpFirst :: forall a b c. (a -> Process b) -> (a, c) -> Process (b, c)
87+ cpFirst p (a, c) = do b <- p a ; return (b, c)
88+resolveClosure _rtable CpSwap env =
89+ return $ Dynamic (decode env) (unsafeCoerce# cpSwap)
90+ where
91+ cpSwap :: forall a b. (a, b) -> Process (b, a)
92+ cpSwap (a, b) = return (b, a)
93+resolveClosure _rtable CpCopy env =
94+ return $ Dynamic (decode env) (unsafeCoerce# cpCopy)
95+ where
96+ cpCopy :: forall a. a -> Process (a, a)
97+ cpCopy a = return (a, a)
98+resolveClosure _rtable CpLeft env =
99+ return $ Dynamic (decode env) (unsafeCoerce# cpLeft)
100+ where
101+ cpLeft :: forall a b c. (a -> Process b) -> Either a c -> Process (Either b c)
102+ cpLeft p (Left a) = do b <- p a ; return (Left b)
103+ cpLeft _ (Right c) = return (Right c)
104+resolveClosure _rtable CpMirror env =
105+ return $ Dynamic (decode env) (unsafeCoerce# cpMirror)
106+ where
107+ cpMirror :: forall a b. Either a b -> Process (Either b a)
108+ cpMirror (Left a) = return (Right a)
109+ cpMirror (Right b) = return (Left b)
110+resolveClosure _rtable CpUntag env =
111+ return $ Dynamic (decode env) (unsafeCoerce# cpUntag)
112+ where
113+ cpUntag :: forall a. Either a a -> Process a
114+ cpUntag (Left a) = return a
115+ cpUntag (Right a) = return a
116+resolveClosure rtable CpApply env =
117+ return $ Dynamic typApply (unsafeCoerce# cpApply)
118+ where
119+ cpApply :: forall a b. (Closure (a -> Process b), a) -> Process b
120+ cpApply (Closure (Static flabel) fenv, x) = do
121+ let Just f = resolveClosure rtable flabel fenv
122+ Dynamic typResult val = f `dynApp` Dynamic typA (unsafeCoerce# x)
123+ if typResult == typProcB
124+ then unsafeCoerce# val
125+ else error $ "Type error in cpApply: "
126+ ++ "mismatch between " ++ show typResult
127+ ++ " and " ++ show typProcB
128+
129+ (typApply, typA, typProcB) = decode env
130+
131+-- User defined closures
132+resolveClosure rtable (UserStatic label) env = do
133+ val <- rtable ^. remoteTableLabel label
134+ dynApply val (toDyn env)
135+
136+tyConProcess :: TyCon
137+tyConProcess = typeRepTyCon (typeOf (undefined :: Process ()))
138+
distributed-process/src/Control/Distributed/Process/Internal/Closure/TH.hsnew file+176
@@ -0,0 +1,176 @@
1+-- | Template Haskell support
2+--
3+-- (In a separate file for convenience)
4+module Control.Distributed.Process.Internal.Closure.TH
5+ ( -- * User-level API
6+ remotable
7+ , mkClosure
8+ ) where
9+
10+import Prelude hiding (lookup)
11+import Data.ByteString.Lazy (ByteString)
12+import Data.Binary (encode, decode)
13+import Data.Typeable (typeOf, Typeable)
14+import Control.Applicative ((<$>))
15+import Control.Distributed.Process.Internal.Types
16+ ( RemoteTable
17+ , Closure(..)
18+ , Static(..)
19+ , StaticLabel(..)
20+ , Process
21+ , ProcessId
22+ , CallReply(..)
23+ )
24+import Language.Haskell.TH
25+ ( -- Q monad and operations
26+ Q
27+ , reify
28+ -- Names
29+ , Name
30+ , mkName
31+ , nameBase
32+ -- Algebraic data types
33+ , Dec
34+ , Exp
35+ , Type(AppT, ArrowT)
36+ , Info(VarI)
37+ -- Lifted constructors
38+ -- .. Literals
39+ , stringL
40+ -- .. Patterns
41+ , normalB
42+ , clause
43+ -- .. Expressions
44+ , varE
45+ , litE
46+ -- .. Top-level declarations
47+ , funD
48+ , sigD
49+ )
50+import Control.Distributed.Process (send)
51+import Control.Distributed.Process.Internal.Dynamic (toDyn)
52+import Control.Distributed.Process.Internal.Closure
53+ ( registerLabel
54+ , registerSender
55+ )
56+
57+--------------------------------------------------------------------------------
58+-- User-level API --
59+--------------------------------------------------------------------------------
60+
61+-- | Create the closure, decoder, and metadata definitions for the given list
62+-- of functions
63+remotable :: [Name] -> Q [Dec]
64+remotable ns = do
65+ (closures, inserts) <- unzip <$> mapM generateDefs ns
66+ rtable <- createMetaData (concat inserts)
67+ return $ concat closures ++ rtable
68+
69+-- | Create a closure
70+mkClosure :: Name -> Q Exp
71+mkClosure = varE . closureName
72+
73+--------------------------------------------------------------------------------
74+-- Internal (Template Haskell) --
75+--------------------------------------------------------------------------------
76+
77+-- | Generate the code to add the metadata to the CH runtime
78+createMetaData :: [Q Exp] -> Q [Dec]
79+createMetaData is =
80+ [d| __remoteTable :: RemoteTable -> RemoteTable ;
81+ __remoteTable = $(compose is)
82+ |]
83+
84+-- | Generate the necessary definitions for one function
85+--
86+-- Given an (f :: a -> b) in module M, create:
87+-- 1. f__closure :: a -> Closure b,
88+-- 2. registerLabel "M.f" (toDyn ((f . enc) :: ByteString -> b))
89+--
90+-- Moreover, if b is of the form Process c, then additionally create
91+-- 3. registerSender (Process c) (send :: ProcessId -> c -> Process ())
92+generateDefs :: Name -> Q ([Dec], [Q Exp])
93+generateDefs n = do
94+ mType <- getType n
95+ case mType of
96+ Just (origName, ArrowT `AppT` arg `AppT` res) -> do
97+ (closure, label) <- generateClosure origName (return arg) (return res)
98+ let decoder = generateDecoder origName (return res)
99+ insert = [| registerLabel $(stringE label) (toDyn $decoder) |]
100+
101+ -- Generate special closure to support 'call'
102+ mResult <- processResult res
103+ insert' <- case mResult of
104+ Nothing ->
105+ return []
106+ Just result -> do
107+ let sender = generateSender result
108+ return . return $
109+ [| registerSender $(typeToTypeRep res) (toDyn $sender) |]
110+
111+ return (closure, insert : insert')
112+ _ ->
113+ fail $ "remotable: " ++ show n ++ " is not a function"
114+
115+-- | Generate the closure creator (see 'generateDefs')
116+generateClosure :: Name -> Q Type -> Q Type -> Q ([Dec], String)
117+generateClosure n arg res = do
118+ closure <- sequence
119+ [ sigD (closureName n) [t| $arg -> Closure $res |]
120+ , sfnD (closureName n) [| Closure (Static (UserStatic ($(stringE label)))) . encode |]
121+ ]
122+ return (closure, label)
123+ where
124+ label :: String
125+ label = show $ n
126+
127+-- | If 't' is of the form 'Process b', return 'Just b'
128+processResult :: Type -> Q (Maybe (Q Type))
129+processResult t = do
130+ process <- [t| Process |]
131+ case t of
132+ p `AppT` a | p == process -> return . Just . return $ a
133+ _ -> return Nothing
134+
135+-- | Generate the decoder (see 'generateDefs')
136+generateDecoder :: Name -> Q Type -> Q Exp
137+generateDecoder n res = [| $(varE n) . decode :: ByteString -> $res |]
138+
139+-- | Generate the sender. This is necessary to support 'call'
140+generateSender :: Q Type -> Q Exp
141+generateSender tp =
142+ [| (\pid -> send pid . CallReply) :: ProcessId -> $tp -> Process () |]
143+
144+-- | The name for the function that generates the closure
145+closureName :: Name -> Name
146+closureName n = mkName $ nameBase n ++ "__closure"
147+
148+--------------------------------------------------------------------------------
149+-- Generic Template Haskell auxiliary functions --
150+--------------------------------------------------------------------------------
151+
152+typeToTypeRep :: Type -> Q Exp
153+typeToTypeRep tp = [| typeOf (undefined :: $(return tp)) |]
154+
155+-- | Compose a set of expressions
156+compose :: [Q Exp] -> Q Exp
157+compose [] = [| id |]
158+compose [e] = e
159+compose (e:es) = [| $e . $(compose es) |]
160+
161+-- | Literal string as an expression
162+stringE :: String -> Q Exp
163+stringE = litE . stringL
164+
165+-- | Look up the "original name" (module:name) and type of a top-level function
166+getType :: Name -> Q (Maybe (Name, Type))
167+getType name = do
168+ info <- reify name
169+ case info of
170+ VarI origName typ _ _ -> return $ Just (origName, typ)
171+ _ -> return Nothing
172+
173+-- | Variation on 'funD' which takes a single expression to define the function
174+sfnD :: Name -> Q Exp -> Q Dec
175+sfnD n e = funD n [clause [] (normalB e) []]
176+
distributed-process/src/Control/Distributed/Process/Internal/MessageT.hs+55−8
@@ -4,21 +4,25 @@
44-- monad).
55module Control.Distributed.Process.Internal.MessageT
66 ( MessageT
7+ , runMessageT
8+ , getLocalNode
79 , sendPayload
810 , sendBinary
911 , sendMessage
10− , getLocalNode
12+ , payloadToMessage
13+ , createMessage
1114 ) where
1215
1316import Data.Binary (Binary, encode)
14−import Data.ByteString (ByteString)
15−import qualified Data.ByteString.Lazy as BSL (toChunks)
17+import qualified Data.ByteString as BSS (ByteString, concat)
18+import qualified Data.ByteString.Lazy as BSL (toChunks, fromChunks, splitAt)
1619import Data.Map (Map)
20+import qualified Data.Map as Map (empty)
1721import Data.Accessor (Accessor, accessor, (^=), (^.))
1822import qualified Data.Accessor.Container as DAC (mapMaybe)
1923import Control.Category ((>>>))
2024import Control.Monad (unless, liftM)
21−import Control.Monad.State (gets, modify)
25+import Control.Monad.State (gets, modify, evalStateT)
2226import Control.Monad.IO.Class (MonadIO, liftIO)
2327import Control.Concurrent.Chan (writeChan)
2428import Control.Distributed.Process.Internal.Types
@@ -30,12 +34,17 @@ import Control.Distributed.Process.Internal.Types
3034 , NCMsg(NCMsg, ctrlMsgSender, ctrlMsgSignal)
3135 , DiedReason(DiedDisconnect)
3236 , ProcessSignal(Died)
37+ , Message(..)
3338 , MessageT(..)
3439 , MessageState(..)
35− , createMessage
36− , messageToPayload
3740 )
38−import Control.Distributed.Process.Serializable (Serializable)
41+import Control.Distributed.Process.Serializable
42+ ( Serializable
43+ , fingerprint
44+ , encodeFingerprint
45+ , decodeFingerprint
46+ , sizeOfFingerprint
47+ )
3948import qualified Network.Transport as NT
4049 ( EndPoint
4150 , Connection
@@ -45,10 +54,18 @@ import qualified Network.Transport as NT
4554 , defaultConnectHints
4655 )
4756
57+--------------------------------------------------------------------------------
58+-- API --
59+--------------------------------------------------------------------------------
60+
61+runMessageT :: Monad m => LocalNode -> MessageT m a -> m a
62+runMessageT localNode m =
63+ evalStateT (unMessageT m) $ initMessageState localNode
64+
4865getLocalNode :: Monad m => MessageT m LocalNode
4966getLocalNode = gets messageLocalNode
5067
51−sendPayload :: MonadIO m => Identifier -> [ByteString] -> MessageT m ()
68+sendPayload :: MonadIO m => Identifier -> [BSS.ByteString] -> MessageT m ()
5269sendPayload them payload = do
5370 mConn <- connTo them
5471 didSend <- case mConn of
@@ -71,6 +88,36 @@ sendBinary them = sendPayload them . BSL.toChunks . encode
7188sendMessage :: (MonadIO m, Serializable a) => Identifier -> a -> MessageT m ()
7289sendMessage them = sendPayload them . messageToPayload . createMessage
7390
91+--------------------------------------------------------------------------------
92+-- Serialization/deserialization --
93+--------------------------------------------------------------------------------
94+
95+-- | Turn any serialiable term into a message
96+createMessage :: Serializable a => a -> Message
97+createMessage a = Message (fingerprint a) (encode a)
98+
99+-- | Serialize a message
100+messageToPayload :: Message -> [BSS.ByteString]
101+messageToPayload (Message fp enc) = encodeFingerprint fp : BSL.toChunks enc
102+
103+-- | Deserialize a message
104+payloadToMessage :: [BSS.ByteString] -> Message
105+payloadToMessage payload = Message fp msg
106+ where
107+ (encFp, msg) = BSL.splitAt (fromIntegral sizeOfFingerprint)
108+ $ BSL.fromChunks payload
109+ fp = decodeFingerprint . BSS.concat . BSL.toChunks $ encFp
110+
111+--------------------------------------------------------------------------------
112+-- Internal --
113+--------------------------------------------------------------------------------
114+
115+initMessageState :: LocalNode -> MessageState
116+initMessageState localNode = MessageState {
117+ messageLocalNode = localNode
118+ , _messageConnections = Map.empty
119+ }
120+
74121setupConnTo :: MonadIO m => Identifier -> MessageT m (Maybe NT.Connection)
75122setupConnTo them = do
76123 endPoint <- localEndPoint `liftM` getLocalNode
distributed-process/src/Control/Distributed/Process/Internal/Node.hsnew file+16
@@ -0,0 +1,16 @@
1+module Control.Distributed.Process.Internal.Node
2+ ( runLocalProcess
3+ ) where
4+
5+import Control.Monad.Reader (runReaderT)
6+import Control.Distributed.Process.Internal.Types
7+ ( LocalNode
8+ , LocalProcess
9+ , Process(unProcess)
10+ )
11+import Control.Distributed.Process.Internal.MessageT (runMessageT)
12+
13+-- | Deconstructor for 'Process' (not exported to the public API)
14+runLocalProcess :: LocalNode -> Process a -> LocalProcess -> IO a
15+runLocalProcess node proc = runMessageT node . runReaderT (unProcess proc)
16+
distributed-process/src/Control/Distributed/Process/Internal/Types.hs+16−179
@@ -13,7 +13,6 @@ module Control.Distributed.Process.Internal.Types
1313 , LocalProcess(..)
1414 , LocalProcessState(..)
1515 , Process(..)
16− , runLocalProcess
1716 -- * Typed channels
1817 , LocalChannelId
1918 , ChannelId(..)
@@ -25,11 +24,7 @@ module Control.Distributed.Process.Internal.Types
2524 , Static(..)
2625 , Closure(..)
2726 , CallReply(..)
28− , RemoteTable
29− , initRemoteTable
30− , registerLabel
31− , registerSender
32− , resolveClosure
27+ , RemoteTable(..)
3328 -- * Messages
3429 , Message(..)
3530 -- * Node controller user-visible data types
@@ -44,10 +39,9 @@ module Control.Distributed.Process.Internal.Types
4439 -- * Node controller internal data types
4540 , NCMsg(..)
4641 , ProcessSignal(..)
47− -- * Serialization/deserialization
48− , createMessage
49− , messageToPayload
50− , payloadToMessage
42+ -- * MessageT monad
43+ , MessageT(..)
44+ , MessageState(..)
5145 -- * Accessors
5246 , localProcesses
5347 , localPidCounter
@@ -58,27 +52,20 @@ module Control.Distributed.Process.Internal.Types
5852 , channelCounter
5953 , typedChannels
6054 , typedChannelWithId
61− -- * MessageT monad
62− , MessageT(..)
63− , MessageState(..)
64− , runMessageT
55+ , remoteTableLabels
56+ , remoteTableSenders
57+ , remoteTableLabel
58+ , remoteTableSender
6559 ) where
6660
6761import Data.Map (Map)
68−import qualified Data.Map as Map (empty)
6962import Data.Int (Int32)
70−import Data.Typeable (Typeable, TypeRep, TyCon, typeRepTyCon, typeOf)
71−import Data.Binary (Binary, encode, decode, put, get, putWord8, getWord8)
63+import Data.Typeable (Typeable, TypeRep)
64+import Data.Binary (Binary(put, get), putWord8, getWord8)
7265import Data.ByteString.Lazy (ByteString)
73−import Data.Accessor (Accessor, accessor, (^=), (^.))
66+import Data.Accessor (Accessor, accessor)
7467import qualified Data.Accessor.Container as DAC (mapMaybe)
75−import qualified Data.ByteString.Lazy as BSL
76− ( ByteString
77− , toChunks
78− , fromChunks
79− , splitAt
80− )
81−import qualified Data.ByteString as BSS (ByteString, concat)
68+import qualified Data.ByteString.Lazy as BSL (ByteString)
8269import Control.Category ((>>>))
8370import Control.Exception (Exception)
8471import Control.Concurrent (ThreadId)
@@ -87,28 +74,12 @@ import Control.Concurrent.Chan (Chan)
8774import Control.Concurrent.STM (TChan, TVar)
8875import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)
8976import Control.Applicative (Applicative, (<$>), (<*>))
90−import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
77+import Control.Monad.Reader (MonadReader(..), ReaderT)
9178import Control.Monad.IO.Class (MonadIO)
92−import Control.Monad.State (MonadState, StateT, evalStateT)
93−import Control.Distributed.Process.Serializable
94− ( Fingerprint
95− , Serializable
96− , encodeFingerprint
97− , decodeFingerprint
98− , fingerprint
99− , sizeOfFingerprint
100− )
79+import Control.Monad.State (MonadState, StateT)
80+import Control.Distributed.Process.Serializable (Fingerprint, Serializable)
10181import Control.Distributed.Process.Internal.CQueue (CQueue)
102−import Control.Distributed.Process.Internal.Dynamic
103− ( Dynamic(..)
104− , toDyn
105− , dynTypeRep
106− , dynApp
107− , dynBind
108− , dynApply
109− , unsafeCoerce#
110− )
111−import Control.Distributed.Process.Internal.TypeRep ()
82+import Control.Distributed.Process.Internal.Dynamic (Dynamic)
11283
11384--------------------------------------------------------------------------------
11485-- Node and process identifiers --
@@ -192,10 +163,6 @@ newtype Process a = Process {
192163 }
193164 deriving (Functor, Monad, MonadIO, MonadReader LocalProcess, Typeable, Applicative)
194165
195−-- | Deconstructor for 'Process' (not exported to the public API)
196−runLocalProcess :: LocalNode -> Process a -> LocalProcess -> IO a
197−runLocalProcess node proc = runMessageT node . runReaderT (unProcess proc)
198−
199166--------------------------------------------------------------------------------
200167-- Typed channels --
201168--------------------------------------------------------------------------------
@@ -264,105 +231,6 @@ data RemoteTable = RemoteTable {
264231 , _remoteTableSenders :: Map TypeRep Dynamic
265232 }
266233
267−-- | Initial (empty) remote-call meta data
268−initRemoteTable :: RemoteTable
269−initRemoteTable = RemoteTable Map.empty Map.empty
270−
271−registerLabel :: String -> Dynamic -> RemoteTable -> RemoteTable
272−registerLabel label dyn = remoteTableLabel label ^= Just dyn
273−
274−registerSender :: TypeRep -> Dynamic -> RemoteTable -> RemoteTable
275−registerSender typ dyn = remoteTableSender typ ^= Just dyn
276−
277−resolveClosure :: RemoteTable -> StaticLabel -> ByteString -> Maybe Dynamic
278−-- Special support for call
279−resolveClosure rtable Call env = do
280− proc <- resolveClosure rtable label env'
281− sender <- rtable ^. remoteTableSender (dynTypeRep proc)
282− proc `bind` (sender `dynApp` toDyn (pid :: ProcessId))
283− where
284− (label, env', pid) = decode env
285− bind = dynBind tyConProcess bindProcess
286− bindProcess :: Process a -> (a -> Process b) -> Process b
287− bindProcess = (>>=)
288−-- Generic closure combinators
289−resolveClosure rtable ClosureApply env = do
290− f <- resolveClosure rtable labelf envf
291− x <- resolveClosure rtable labelx envx
292− f `dynApply` x
293− where
294− (labelf, envf, labelx, envx) = decode env
295−resolveClosure _rtable ClosureConst env =
296− return $ Dynamic (decode env) (unsafeCoerce# const)
297−resolveClosure _rtable ClosureUnit _env =
298− return $ toDyn ()
299−-- Arrow combinators
300−resolveClosure _rtable CpId env =
301− return $ Dynamic (decode env) (unsafeCoerce# cpId)
302− where
303− cpId :: forall a. a -> Process a
304− cpId = return
305−resolveClosure _rtable CpComp env =
306− return $ Dynamic (decode env) (unsafeCoerce# cpComp)
307− where
308− cpComp :: forall a b c. (a -> Process b) -> (b -> Process c) -> a -> Process c
309− cpComp p q a = p a >>= q
310−resolveClosure _rtable CpFirst env =
311− return $ Dynamic (decode env) (unsafeCoerce# cpFirst)
312− where
313− cpFirst :: forall a b c. (a -> Process b) -> (a, c) -> Process (b, c)
314− cpFirst p (a, c) = do b <- p a ; return (b, c)
315−resolveClosure _rtable CpSwap env =
316− return $ Dynamic (decode env) (unsafeCoerce# cpSwap)
317− where
318− cpSwap :: forall a b. (a, b) -> Process (b, a)
319− cpSwap (a, b) = return (b, a)
320−resolveClosure _rtable CpCopy env =
321− return $ Dynamic (decode env) (unsafeCoerce# cpCopy)
322− where
323− cpCopy :: forall a. a -> Process (a, a)
324− cpCopy a = return (a, a)
325−resolveClosure _rtable CpLeft env =
326− return $ Dynamic (decode env) (unsafeCoerce# cpLeft)
327− where
328− cpLeft :: forall a b c. (a -> Process b) -> Either a c -> Process (Either b c)
329− cpLeft p (Left a) = do b <- p a ; return (Left b)
330− cpLeft _ (Right c) = return (Right c)
331−resolveClosure _rtable CpMirror env =
332− return $ Dynamic (decode env) (unsafeCoerce# cpMirror)
333− where
334− cpMirror :: forall a b. Either a b -> Process (Either b a)
335− cpMirror (Left a) = return (Right a)
336− cpMirror (Right b) = return (Left b)
337−resolveClosure _rtable CpUntag env =
338− return $ Dynamic (decode env) (unsafeCoerce# cpUntag)
339− where
340− cpUntag :: forall a. Either a a -> Process a
341− cpUntag (Left a) = return a
342− cpUntag (Right a) = return a
343−resolveClosure rtable CpApply env =
344− return $ Dynamic typApply (unsafeCoerce# cpApply)
345− where
346− cpApply :: forall a b. (Closure (a -> Process b), a) -> Process b
347− cpApply (Closure (Static flabel) fenv, x) = do
348− let Just f = resolveClosure rtable flabel fenv
349− Dynamic typResult val = f `dynApp` Dynamic typA (unsafeCoerce# x)
350− if typResult == typProcB
351− then unsafeCoerce# val
352− else error $ "Type error in cpApply: "
353− ++ "mismatch between " ++ show typResult
354− ++ " and " ++ show typProcB
355−
356− (typApply, typA, typProcB) = decode env
357−
358−-- User defined closures
359−resolveClosure rtable (UserStatic label) env = do
360− val <- rtable ^. remoteTableLabel label
361− dynApply val (toDyn env)
362−
363−tyConProcess :: TyCon
364−tyConProcess = typeRepTyCon (typeOf (undefined :: Process ()))
365−
366234--------------------------------------------------------------------------------
367235-- Messages --
368236--------------------------------------------------------------------------------
@@ -448,26 +316,6 @@ data ProcessSignal =
448316 | Spawn (Closure (Process ())) SpawnRef
449317 deriving Show
450318
451−--------------------------------------------------------------------------------
452−-- Serialization/deserialization --
453−--------------------------------------------------------------------------------
454−
455−-- | Turn any serialiable term into a message
456−createMessage :: Serializable a => a -> Message
457−createMessage a = Message (fingerprint a) (encode a)
458−
459−-- | Serialize a message
460−messageToPayload :: Message -> [BSS.ByteString]
461−messageToPayload (Message fp enc) = encodeFingerprint fp : BSL.toChunks enc
462−
463−-- | Deserialize a message
464−payloadToMessage :: [BSS.ByteString] -> Message
465−payloadToMessage payload = Message fp msg
466− where
467− (encFp, msg) = BSL.splitAt (fromIntegral sizeOfFingerprint)
468− $ BSL.fromChunks payload
469− fp = decodeFingerprint . BSS.concat . BSL.toChunks $ encFp
470−
471319--------------------------------------------------------------------------------
472320-- Binary instances --
473321--------------------------------------------------------------------------------
@@ -640,14 +488,3 @@ data MessageState = MessageState {
640488 messageLocalNode :: LocalNode
641489 , _messageConnections :: Map Identifier NT.Connection
642490 }
643−
644−runMessageT :: Monad m => LocalNode -> MessageT m a -> m a
645−runMessageT localNode m =
646− evalStateT (unMessageT m) $ initMessageState localNode
647−
648−initMessageState :: LocalNode -> MessageState
649−initMessageState localNode = MessageState {
650− messageLocalNode = localNode
651− , _messageConnections = Map.empty
652− }
653−
distributed-process/src/Control/Distributed/Process/Node.hs+8−6
@@ -66,15 +66,12 @@ import Control.Distributed.Process.Internal.Types
6666 , LocalProcess(..)
6767 , LocalProcessState(..)
6868 , Process(..)
69− , runLocalProcess
7069 , DiedReason(..)
7170 , NCMsg(..)
7271 , ProcessSignal(..)
73− , payloadToMessage
7472 , localPidCounter
7573 , localPidUnique
7674 , localProcessWithId
77− , initRemoteTable
7875 , MonitorRef(..)
7976 , MonitorNotification(..)
8077 , LinkException(..)
@@ -84,22 +81,27 @@ import Control.Distributed.Process.Internal.Types
8481 , DidSpawn(..)
8582 , Closure(..)
8683 , Static(..)
87− , createMessage
8884 , MessageT
89− , runMessageT
9085 , TypedChannel(..)
9186 , Identifier(..)
9287 , ChannelId(..)
9388 , typedChannelWithId
94− , resolveClosure
9589 )
9690import Control.Distributed.Process.Serializable (Serializable)
9791import Control.Distributed.Process.Internal.MessageT
9892 ( sendBinary
9993 , sendMessage
10094 , getLocalNode
95+ , runMessageT
96+ , payloadToMessage
97+ , createMessage
10198 )
10299import Control.Distributed.Process.Internal.Dynamic (fromDynamic)
100+import Control.Distributed.Process.Internal.Closure
101+ ( initRemoteTable
102+ , resolveClosure
103+ )
104+import Control.Distributed.Process.Internal.Node (runLocalProcess)
103105
104106--------------------------------------------------------------------------------
105107-- Initialization --