git.iko.soy/distributed-process
on master help sign in
Start tidying up the code
03fd8b4Edsko de Vries
distributed-process/src/Control/Distributed/Process.hs+2−2
compare with master
unifiedsplit
@@ -91,7 +91,6 @@ import Control.Distributed.Process.Internal.Types
91 , LocalNode(..)
91 , LocalNode(..)
92 , LocalProcess(..)
92 , LocalProcess(..)
93 , Process(..)
93 , Process(..)
94− , runLocalProcess
95 , Closure(..)
94 , Closure(..)
96 , StaticLabel(Call)
95 , StaticLabel(Call)
97 , Static(..)
96 , Static(..)
@@ -118,7 +117,6 @@ import Control.Distributed.Process.Internal.Types
118 , TypedChannel(..)
117 , TypedChannel(..)
119 , ChannelId(..)
118 , ChannelId(..)
120 , Identifier(..)
119 , Identifier(..)
121− , resolveClosure
122 )
120 )
123import Control.Distributed.Process.Internal.MessageT
121import Control.Distributed.Process.Internal.MessageT
124 ( sendMessage
122 ( sendMessage
@@ -126,6 +124,8 @@ import Control.Distributed.Process.Internal.MessageT
126 , getLocalNode
124 , getLocalNode
127 )
125 )
128import Control.Distributed.Process.Internal.Dynamic (fromDyn, dynTypeRep)
126import Control.Distributed.Process.Internal.Dynamic (fromDyn, dynTypeRep)
127+import Control.Distributed.Process.Internal.Closure (resolveClosure)
128+import Control.Distributed.Process.Internal.Node (runLocalProcess)
129
129
130-- INTERNAL NOTES
130-- INTERNAL NOTES
131--
131--
distributed-process/src/Control/Distributed/Process/Closure.hs+7−161
@@ -38,7 +38,6 @@
38-- > $ initRemoteTable
38-- > $ initRemoteTable
39--
39--
40-- See Section 6, /Faking It/, of /Towards Haskell in the Cloud/ for more info.
40-- See Section 6, /Faking It/, of /Towards Haskell in the Cloud/ for more info.
41−{-# LANGUAGE TemplateHaskell #-}
42module Control.Distributed.Process.Closure
41module Control.Distributed.Process.Closure
43 ( -- * Primitive closures
42 ( -- * Primitive closures
44 remotable
43 remotable
@@ -72,65 +71,16 @@ module Control.Distributed.Process.Closure
72 ) where
71 ) where
73
72
74import Prelude hiding (lookup)
73import Prelude hiding (lookup)
75−import Data.ByteString.Lazy (ByteString, empty)
74+import qualified Data.ByteString.Lazy as BS (empty)
76−import Data.Binary (encode, decode)
75+import Data.Binary (encode)
77import Data.Typeable (typeOf, Typeable)
76import 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− )
105import Control.Distributed.Process.Internal.Types
77import Control.Distributed.Process.Internal.Types
106− ( RemoteTable
107− , Closure(..)
78+ ( Closure(..)
108 , Static(..)
79 , Static(..)
109 , StaticLabel(..)
80 , StaticLabel(..)
110 , Process
81 , Process
111− , ProcessId
112− , CallReply(..)
113− , registerLabel
114− , registerSender
115 )
82 )
116−import Control.Distributed.Process (send)
83+import Control.Distributed.Process.Internal.Closure.TH (remotable, mkClosure)
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
134
84
135--------------------------------------------------------------------------------
85--------------------------------------------------------------------------------
136-- Generic closure combinators --
86-- Generic closure combinators --
@@ -148,7 +98,7 @@ closureConst = Closure (Static ClosureConst) (encode $ typeOf aux)
148 aux = undefined
98 aux = undefined
149
99
150closureUnit :: Closure ()
100closureUnit :: Closure ()
151−closureUnit = Closure (Static ClosureUnit) empty
101+closureUnit = Closure (Static ClosureUnit) BS.empty
152
102
153--------------------------------------------------------------------------------
103--------------------------------------------------------------------------------
154-- Arrow combinators for processes --
104-- Arrow combinators for processes --
@@ -175,10 +125,10 @@ cpComp :: forall a b c. (Typeable a, Typeable b, Typeable c)
175 => CP a b -> CP b c -> CP a c
125 => CP a b -> CP b c -> CP a c
176cpComp f g = comp `closureApply` f `closureApply` g
126cpComp f g = comp `closureApply` f `closureApply` g
177 where
127 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)
179 comp = Closure (Static CpComp) (encode $ typeOf aux)
129 comp = Closure (Static CpComp) (encode $ typeOf aux)
180
130
181− aux :: (a -> Process b) -> (b -> Process c) -> (a -> Process c)
131+ aux :: (a -> Process b) -> (b -> Process c) -> a -> Process c
182 aux = undefined
132 aux = undefined
183
133
184cpFirst :: forall a b c. (Typeable a, Typeable b, Typeable c)
134cpFirst :: forall a b c. (Typeable a, Typeable b, Typeable c)
@@ -273,107 +223,3 @@ cpBind x f = cpElim $ cpIntro x `cpComp` f
273
223
274cpSeq :: Closure (Process ()) -> Closure (Process ()) -> Closure (Process ())
224cpSeq :: Closure (Process ()) -> Closure (Process ()) -> Closure (Process ())
275cpSeq p q = p `cpBind` cpIntro q
225cpSeq 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 @@
4-- monad).
4-- monad).
5module Control.Distributed.Process.Internal.MessageT
5module Control.Distributed.Process.Internal.MessageT
6 ( MessageT
6 ( MessageT
7+ , runMessageT
8+ , getLocalNode
7 , sendPayload
9 , sendPayload
8 , sendBinary
10 , sendBinary
9 , sendMessage
11 , sendMessage
10− , getLocalNode
12+ , payloadToMessage
13+ , createMessage
11 ) where
14 ) where
12
15
13import Data.Binary (Binary, encode)
16import Data.Binary (Binary, encode)
14−import Data.ByteString (ByteString)
17+import qualified Data.ByteString as BSS (ByteString, concat)
15−import qualified Data.ByteString.Lazy as BSL (toChunks)
18+import qualified Data.ByteString.Lazy as BSL (toChunks, fromChunks, splitAt)
16import Data.Map (Map)
19import Data.Map (Map)
20+import qualified Data.Map as Map (empty)
17import Data.Accessor (Accessor, accessor, (^=), (^.))
21import Data.Accessor (Accessor, accessor, (^=), (^.))
18import qualified Data.Accessor.Container as DAC (mapMaybe)
22import qualified Data.Accessor.Container as DAC (mapMaybe)
19import Control.Category ((>>>))
23import Control.Category ((>>>))
20import Control.Monad (unless, liftM)
24import Control.Monad (unless, liftM)
21−import Control.Monad.State (gets, modify)
25+import Control.Monad.State (gets, modify, evalStateT)
22import Control.Monad.IO.Class (MonadIO, liftIO)
26import Control.Monad.IO.Class (MonadIO, liftIO)
23import Control.Concurrent.Chan (writeChan)
27import Control.Concurrent.Chan (writeChan)
24import Control.Distributed.Process.Internal.Types
28import Control.Distributed.Process.Internal.Types
@@ -30,12 +34,17 @@ import Control.Distributed.Process.Internal.Types
30 , NCMsg(NCMsg, ctrlMsgSender, ctrlMsgSignal)
34 , NCMsg(NCMsg, ctrlMsgSender, ctrlMsgSignal)
31 , DiedReason(DiedDisconnect)
35 , DiedReason(DiedDisconnect)
32 , ProcessSignal(Died)
36 , ProcessSignal(Died)
37+ , Message(..)
33 , MessageT(..)
38 , MessageT(..)
34 , MessageState(..)
39 , MessageState(..)
35− , createMessage
36− , messageToPayload
37 )
40 )
38−import Control.Distributed.Process.Serializable (Serializable)
41+import Control.Distributed.Process.Serializable
42+ ( Serializable
43+ , fingerprint
44+ , encodeFingerprint
45+ , decodeFingerprint
46+ , sizeOfFingerprint
47+ )
39import qualified Network.Transport as NT
48import qualified Network.Transport as NT
40 ( EndPoint
49 ( EndPoint
41 , Connection
50 , Connection
@@ -45,10 +54,18 @@ import qualified Network.Transport as NT
45 , defaultConnectHints
54 , defaultConnectHints
46 )
55 )
47
56
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+
48getLocalNode :: Monad m => MessageT m LocalNode
65getLocalNode :: Monad m => MessageT m LocalNode
49getLocalNode = gets messageLocalNode
66getLocalNode = gets messageLocalNode
50
67
51−sendPayload :: MonadIO m => Identifier -> [ByteString] -> MessageT m ()
68+sendPayload :: MonadIO m => Identifier -> [BSS.ByteString] -> MessageT m ()
52sendPayload them payload = do
69sendPayload them payload = do
53 mConn <- connTo them
70 mConn <- connTo them
54 didSend <- case mConn of
71 didSend <- case mConn of
@@ -71,6 +88,36 @@ sendBinary them = sendPayload them . BSL.toChunks . encode
71sendMessage :: (MonadIO m, Serializable a) => Identifier -> a -> MessageT m ()
88sendMessage :: (MonadIO m, Serializable a) => Identifier -> a -> MessageT m ()
72sendMessage them = sendPayload them . messageToPayload . createMessage
89sendMessage them = sendPayload them . messageToPayload . createMessage
73
90
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+
74setupConnTo :: MonadIO m => Identifier -> MessageT m (Maybe NT.Connection)
121setupConnTo :: MonadIO m => Identifier -> MessageT m (Maybe NT.Connection)
75setupConnTo them = do
122setupConnTo them = do
76 endPoint <- localEndPoint `liftM` getLocalNode
123 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
13 , LocalProcess(..)
13 , LocalProcess(..)
14 , LocalProcessState(..)
14 , LocalProcessState(..)
15 , Process(..)
15 , Process(..)
16− , runLocalProcess
17 -- * Typed channels
16 -- * Typed channels
18 , LocalChannelId
17 , LocalChannelId
19 , ChannelId(..)
18 , ChannelId(..)
@@ -25,11 +24,7 @@ module Control.Distributed.Process.Internal.Types
25 , Static(..)
24 , Static(..)
26 , Closure(..)
25 , Closure(..)
27 , CallReply(..)
26 , CallReply(..)
28− , RemoteTable
27+ , RemoteTable(..)
29− , initRemoteTable
30− , registerLabel
31− , registerSender
32− , resolveClosure
33 -- * Messages
28 -- * Messages
34 , Message(..)
29 , Message(..)
35 -- * Node controller user-visible data types
30 -- * Node controller user-visible data types
@@ -44,10 +39,9 @@ module Control.Distributed.Process.Internal.Types
44 -- * Node controller internal data types
39 -- * Node controller internal data types
45 , NCMsg(..)
40 , NCMsg(..)
46 , ProcessSignal(..)
41 , ProcessSignal(..)
47− -- * Serialization/deserialization
42+ -- * MessageT monad
48− , createMessage
43+ , MessageT(..)
49− , messageToPayload
44+ , MessageState(..)
50− , payloadToMessage
51 -- * Accessors
45 -- * Accessors
52 , localProcesses
46 , localProcesses
53 , localPidCounter
47 , localPidCounter
@@ -58,27 +52,20 @@ module Control.Distributed.Process.Internal.Types
58 , channelCounter
52 , channelCounter
59 , typedChannels
53 , typedChannels
60 , typedChannelWithId
54 , typedChannelWithId
61− -- * MessageT monad
55+ , remoteTableLabels
62− , MessageT(..)
56+ , remoteTableSenders
63− , MessageState(..)
57+ , remoteTableLabel
64− , runMessageT
58+ , remoteTableSender
65 ) where
59 ) where
66
60
67import Data.Map (Map)
61import Data.Map (Map)
68−import qualified Data.Map as Map (empty)
69import Data.Int (Int32)
62import Data.Int (Int32)
70−import Data.Typeable (Typeable, TypeRep, TyCon, typeRepTyCon, typeOf)
63+import Data.Typeable (Typeable, TypeRep)
71−import Data.Binary (Binary, encode, decode, put, get, putWord8, getWord8)
64+import Data.Binary (Binary(put, get), putWord8, getWord8)
72import Data.ByteString.Lazy (ByteString)
65import Data.ByteString.Lazy (ByteString)
73−import Data.Accessor (Accessor, accessor, (^=), (^.))
66+import Data.Accessor (Accessor, accessor)
74import qualified Data.Accessor.Container as DAC (mapMaybe)
67import qualified Data.Accessor.Container as DAC (mapMaybe)
75−import qualified Data.ByteString.Lazy as BSL
68+import qualified Data.ByteString.Lazy as BSL (ByteString)
76− ( ByteString
77− , toChunks
78− , fromChunks
79− , splitAt
80− )
81−import qualified Data.ByteString as BSS (ByteString, concat)
82import Control.Category ((>>>))
69import Control.Category ((>>>))
83import Control.Exception (Exception)
70import Control.Exception (Exception)
84import Control.Concurrent (ThreadId)
71import Control.Concurrent (ThreadId)
@@ -87,28 +74,12 @@ import Control.Concurrent.Chan (Chan)
87import Control.Concurrent.STM (TChan, TVar)
74import Control.Concurrent.STM (TChan, TVar)
88import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)
75import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)
89import Control.Applicative (Applicative, (<$>), (<*>))
76import Control.Applicative (Applicative, (<$>), (<*>))
90−import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
77+import Control.Monad.Reader (MonadReader(..), ReaderT)
91import Control.Monad.IO.Class (MonadIO)
78import Control.Monad.IO.Class (MonadIO)
92−import Control.Monad.State (MonadState, StateT, evalStateT)
79+import Control.Monad.State (MonadState, StateT)
93−import Control.Distributed.Process.Serializable
80+import Control.Distributed.Process.Serializable (Fingerprint, Serializable)
94− ( Fingerprint
95− , Serializable
96− , encodeFingerprint
97− , decodeFingerprint
98− , fingerprint
99− , sizeOfFingerprint
100− )
101import Control.Distributed.Process.Internal.CQueue (CQueue)
81import Control.Distributed.Process.Internal.CQueue (CQueue)
102−import Control.Distributed.Process.Internal.Dynamic
82+import Control.Distributed.Process.Internal.Dynamic (Dynamic)
103− ( Dynamic(..)
104− , toDyn
105− , dynTypeRep
106− , dynApp
107− , dynBind
108− , dynApply
109− , unsafeCoerce#
110− )
111−import Control.Distributed.Process.Internal.TypeRep ()
112
83
113--------------------------------------------------------------------------------
84--------------------------------------------------------------------------------
114-- Node and process identifiers --
85-- Node and process identifiers --
@@ -192,10 +163,6 @@ newtype Process a = Process {
192 }
163 }
193 deriving (Functor, Monad, MonadIO, MonadReader LocalProcess, Typeable, Applicative)
164 deriving (Functor, Monad, MonadIO, MonadReader LocalProcess, Typeable, Applicative)
194
165
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−
199--------------------------------------------------------------------------------
166--------------------------------------------------------------------------------
200-- Typed channels --
167-- Typed channels --
201--------------------------------------------------------------------------------
168--------------------------------------------------------------------------------
@@ -264,105 +231,6 @@ data RemoteTable = RemoteTable {
264 , _remoteTableSenders :: Map TypeRep Dynamic
231 , _remoteTableSenders :: Map TypeRep Dynamic
265 }
232 }
266
233
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−
366--------------------------------------------------------------------------------
234--------------------------------------------------------------------------------
367-- Messages --
235-- Messages --
368--------------------------------------------------------------------------------
236--------------------------------------------------------------------------------
@@ -448,26 +316,6 @@ data ProcessSignal =
448 | Spawn (Closure (Process ())) SpawnRef
316 | Spawn (Closure (Process ())) SpawnRef
449 deriving Show
317 deriving Show
450
318
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−
471--------------------------------------------------------------------------------
319--------------------------------------------------------------------------------
472-- Binary instances --
320-- Binary instances --
473--------------------------------------------------------------------------------
321--------------------------------------------------------------------------------
@@ -640,14 +488,3 @@ data MessageState = MessageState {
640 messageLocalNode :: LocalNode
488 messageLocalNode :: LocalNode
641 , _messageConnections :: Map Identifier NT.Connection
489 , _messageConnections :: Map Identifier NT.Connection
642 }
490 }
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
66 , LocalProcess(..)
66 , LocalProcess(..)
67 , LocalProcessState(..)
67 , LocalProcessState(..)
68 , Process(..)
68 , Process(..)
69− , runLocalProcess
70 , DiedReason(..)
69 , DiedReason(..)
71 , NCMsg(..)
70 , NCMsg(..)
72 , ProcessSignal(..)
71 , ProcessSignal(..)
73− , payloadToMessage
74 , localPidCounter
72 , localPidCounter
75 , localPidUnique
73 , localPidUnique
76 , localProcessWithId
74 , localProcessWithId
77− , initRemoteTable
78 , MonitorRef(..)
75 , MonitorRef(..)
79 , MonitorNotification(..)
76 , MonitorNotification(..)
80 , LinkException(..)
77 , LinkException(..)
@@ -84,22 +81,27 @@ import Control.Distributed.Process.Internal.Types
84 , DidSpawn(..)
81 , DidSpawn(..)
85 , Closure(..)
82 , Closure(..)
86 , Static(..)
83 , Static(..)
87− , createMessage
88 , MessageT
84 , MessageT
89− , runMessageT
90 , TypedChannel(..)
85 , TypedChannel(..)
91 , Identifier(..)
86 , Identifier(..)
92 , ChannelId(..)
87 , ChannelId(..)
93 , typedChannelWithId
88 , typedChannelWithId
94− , resolveClosure
95 )
89 )
96import Control.Distributed.Process.Serializable (Serializable)
90import Control.Distributed.Process.Serializable (Serializable)
97import Control.Distributed.Process.Internal.MessageT
91import Control.Distributed.Process.Internal.MessageT
98 ( sendBinary
92 ( sendBinary
99 , sendMessage
93 , sendMessage
100 , getLocalNode
94 , getLocalNode
95+ , runMessageT
96+ , payloadToMessage
97+ , createMessage
101 )
98 )
102import Control.Distributed.Process.Internal.Dynamic (fromDynamic)
99import 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)
103
105
104--------------------------------------------------------------------------------
106--------------------------------------------------------------------------------
105-- Initialization --
107-- Initialization --