git.iko.soy/distributed-process
on master help sign in
Define arrow operators for closures
commit3c1dda40d06c1ccb1a87c56998149d28de94c5c3 authorEdsko de Vries <edsko@edsko.net> dateWed 27 Jun 2012, 12:39 (14 years ago) parentec1ebfa
4 files changed+281−47
compare with master
unifiedsplit
distributed-process/src/Control/Distributed/Process.hs+2−1 distributed-process/src/Control/Distributed/Process/Closure.hs+158−30 distributed-process/src/Control/Distributed/Process/Internal/Types.hs+118−13 distributed-process/tests/TestClosure.hs+3−3
distributed-process/src/Control/Distributed/Process.hs+2−1
@@ -93,6 +93,7 @@ import Control.Distributed.Process.Internal.Types
9393 , Process(..)
9494 , runLocalProcess
9595 , Closure(..)
96+ , StaticLabel(Call)
9697 , Static(..)
9798 , Message(..)
9899 , MonitorRef(..)
@@ -316,7 +317,7 @@ spawn nid proc = do
316317call :: Serializable a => NodeId -> Closure (Process a) -> Process a
317318call nid (Closure (Static label) env) = do
318319 us <- getSelfPid
319− spawn nid (Closure (Static "$call") (encode (label, env, us)))
320+ spawn nid (Closure (Static Call) (encode (label, env, us)))
320321 CallReply a <- expect
321322 return a
322323
distributed-process/src/Control/Distributed/Process/Closure.hs+158−30
@@ -43,17 +43,36 @@ module Control.Distributed.Process.Closure
4343 ( -- * Primitive closures
4444 remotable
4545 , mkClosure
46− -- * Closure combinators
47− , applyClosure
48− , liftedBind
49− , liftedConst
50− , bindCP
51− , weakenCP
52− , seqCP
46+ -- * Generic closure combinators
47+ , closureApply
48+ , closureConst
49+ , closureUnit
50+ -- * Arrow combinators for processes
51+ , CP
52+ , cpIntro
53+ , cpElim
54+ , cpId
55+ , cpComp
56+ , cpFirst
57+ , cpSwap
58+ , cpSecond
59+ , cpPair
60+ , cpCopy
61+ , cpFanOut
62+ , cpLeft
63+ , cpMirror
64+ , cpRight
65+ , cpEither
66+ , cpUntag
67+ , cpFanIn
68+ , cpApply
69+ -- * Derived combinators for processes
70+ , cpBind
71+ , cpSeq
5372 ) where
5473
5574import Prelude hiding (lookup)
56−import Data.ByteString.Lazy (ByteString)
75+import Data.ByteString.Lazy (ByteString, empty)
5776import Data.Binary (encode, decode)
5877import Data.Typeable (typeOf, Typeable)
5978import Control.Applicative ((<$>))
@@ -87,6 +106,7 @@ import Control.Distributed.Process.Internal.Types
87106 ( RemoteTable
88107 , Closure(..)
89108 , Static(..)
109+ , StaticLabel(..)
90110 , Process
91111 , ProcessId
92112 , CallReply(..)
@@ -113,35 +133,143 @@ mkClosure :: Name -> Q Exp
113133mkClosure = varE . closureName
114134
115135--------------------------------------------------------------------------------
116−-- Closure combinators --
136+-- Generic closure combinators --
117137--------------------------------------------------------------------------------
118138
119−applyClosure :: Closure (a -> b) -> Closure a -> Closure b
120−applyClosure (Closure (Static labelf) envf) (Closure (Static labelx) envx) =
121− Closure (Static "$applyClosure") $ encode (labelf, envf, labelx, envx)
139+closureApply :: Closure (a -> b) -> Closure a -> Closure b
140+closureApply (Closure (Static labelf) envf) (Closure (Static labelx) envx) =
141+ Closure (Static ClosureApply) $ encode (labelf, envf, labelx, envx)
122142
123−liftedBind :: forall a b. (Typeable a, Typeable b)
124− => Closure (Process a -> (a -> Process b) -> Process b)
125−liftedBind = Closure (Static "$liftedBind") (encode typ)
126− where
127− typ = typeOf (undefined :: Process a -> (a -> Process b) -> Process b)
143+closureConst :: forall a b. (Typeable a, Typeable b)
144+ => Closure (a -> b -> a)
145+closureConst = Closure (Static ClosureConst) (encode $ typeOf aux)
146+ where
147+ aux :: a -> b -> a
148+ aux = undefined
149+
150+closureUnit :: Closure ()
151+closureUnit = Closure (Static ClosureUnit) empty
152+
153+--------------------------------------------------------------------------------
154+-- Arrow combinators for processes --
155+--------------------------------------------------------------------------------
128156
129−liftedConst :: forall a b. (Typeable a, Typeable b)
130− => Closure (a -> b -> a)
131−liftedConst = Closure (Static "$liftedConst") (encode typ)
157+type CP a b = Closure (a -> Process b)
158+
159+cpIntro :: (Typeable a, Typeable b)
160+ => Closure (Process b) -> CP a b
161+cpIntro = closureApply closureConst
162+
163+cpElim :: Typeable a
164+ => CP () a -> Closure (Process a)
165+cpElim = flip closureApply closureUnit
166+
167+cpId :: forall a. Typeable a
168+ => CP a a
169+cpId = Closure (Static CpId) (encode $ typeOf aux)
132170 where
133− typ = typeOf (undefined :: a -> b -> a)
171+ aux :: a -> Process a
172+ aux = undefined
134173
135−bindCP :: (Typeable a, Typeable b)
136− => Closure (Process a) -> Closure (a -> Process b) -> Closure (Process b)
137−bindCP x f = liftedBind `applyClosure` x `applyClosure` f
174+cpComp :: forall a b c. (Typeable a, Typeable b, Typeable c)
175+ => CP a b -> CP b c -> CP a c
176+cpComp f g = comp `closureApply` f `closureApply` g
177+ where
178+ comp :: Closure ((a -> Process b) -> (b -> Process c) -> (a -> Process c))
179+ comp = Closure (Static CpComp) (encode $ typeOf aux)
180+
181+ aux :: (a -> Process b) -> (b -> Process c) -> (a -> Process c)
182+ aux = undefined
183+
184+cpFirst :: forall a b c. (Typeable a, Typeable b, Typeable c)
185+ => CP a b -> CP (a, c) (b, c)
186+cpFirst = closureApply first
187+ where
188+ first :: Closure ((a -> Process b) -> (a, c) -> Process (b, c))
189+ first = Closure (Static CpFirst) (encode $ typeOf aux)
190+
191+ aux :: (a -> Process b) -> (a, c) -> Process (b, c)
192+ aux = undefined
193+
194+cpSwap :: forall a b. (Typeable a, Typeable b)
195+ => CP (a, b) (b, a)
196+cpSwap = Closure (Static CpSwap) (encode $ typeOf aux)
197+ where
198+ aux :: (a, b) -> Process (b, a)
199+ aux = undefined
200+
201+cpSecond :: (Typeable a, Typeable b, Typeable c)
202+ => CP a b -> CP (c, a) (c, b)
203+cpSecond f = cpSwap `cpComp` cpFirst f `cpComp` cpSwap
204+
205+cpPair :: (Typeable a, Typeable a', Typeable b, Typeable b')
206+ => CP a b -> CP a' b' -> CP (a, a') (b, b')
207+cpPair f g = cpFirst f `cpComp` cpSecond g
208+
209+cpCopy :: forall a. Typeable a
210+ => CP a (a, a)
211+cpCopy = Closure (Static CpCopy) (encode $ typeOf aux)
212+ where
213+ aux :: a -> Process (a, a)
214+ aux = undefined
215+
216+cpFanOut :: (Typeable a, Typeable b, Typeable c)
217+ => CP a b -> CP a c -> CP a (b, c)
218+cpFanOut f g = cpCopy `cpComp` (f `cpPair` g)
219+
220+cpLeft :: forall a b c. (Typeable a, Typeable b, Typeable c)
221+ => CP a b -> CP (Either a c) (Either b c)
222+cpLeft = closureApply left
223+ where
224+ left :: Closure ((a -> Process b) -> Either a c -> Process (Either b c))
225+ left = Closure (Static CpLeft) (encode $ typeOf aux)
226+
227+ aux :: (a -> Process b) -> Either a c -> Process (Either b c)
228+ aux = undefined
229+
230+cpMirror :: forall a b. (Typeable a, Typeable b)
231+ => CP (Either a b) (Either b a)
232+cpMirror = Closure (Static CpMirror) (encode $ typeOf aux)
233+ where
234+ aux :: Either a b -> Process (Either b a)
235+ aux = undefined
236+
237+cpRight :: forall a b c. (Typeable a, Typeable b, Typeable c)
238+ => CP a b -> CP (Either c a) (Either c b)
239+cpRight f = cpMirror `cpComp` cpLeft f `cpComp` cpMirror
240+
241+cpEither :: (Typeable a, Typeable a', Typeable b, Typeable b')
242+ => CP a b -> CP a' b' -> CP (Either a a') (Either b b')
243+cpEither f g = cpLeft f `cpComp` cpRight g
138244
139−weakenCP :: (Typeable a, Typeable b)
140− => Closure (Process b) -> Closure (a -> Process b)
141−weakenCP = applyClosure liftedConst
245+cpUntag :: forall a. Typeable a
246+ => CP (Either a a) a
247+cpUntag = Closure (Static CpUntag) (encode $ typeOf aux)
248+ where
249+ aux :: Either a a -> Process a
250+ aux = undefined
251+
252+cpFanIn :: (Typeable a, Typeable b, Typeable c)
253+ => CP a c -> CP b c -> CP (Either a b) c
254+cpFanIn f g = (f `cpEither` g) `cpComp` cpUntag
255+
256+cpApply :: forall a b. (Typeable a, Typeable b)
257+ => CP (CP a b, a) b
258+cpApply = Closure (Static CpApply) (encode $ typeOf aux)
259+ where
260+ aux :: (Closure (a -> Process b), a) -> Process b
261+ aux = undefined
262+
263+--------------------------------------------------------------------------------
264+-- Some derived operators for processes --
265+--------------------------------------------------------------------------------
266+
267+cpBind :: (Typeable a, Typeable b)
268+ => Closure (Process a) -> Closure (a -> Process b) -> Closure (Process b)
269+cpBind x f = cpElim $ cpIntro x `cpComp` f
142270
143−seqCP :: Closure (Process ()) -> Closure (Process ()) -> Closure (Process ())
144−seqCP p q = bindCP p (weakenCP q)
271+cpSeq :: Closure (Process ()) -> Closure (Process ()) -> Closure (Process ())
272+cpSeq p q = p `cpBind` cpIntro q
145273
146274--------------------------------------------------------------------------------
147275-- Internal (Template Haskell) --
@@ -190,7 +318,7 @@ generateClosure :: Name -> Q Type -> Q Type -> Q ([Dec], String)
190318generateClosure n arg res = do
191319 closure <- sequence
192320 [ sigD (closureName n) [t| $arg -> Closure $res |]
193− , sfnD (closureName n) [| Closure (Static $(stringE label)) . encode |]
321+ , sfnD (closureName n) [| Closure (Static (UserStatic ($(stringE label)))) . encode |]
194322 ]
195323 return (closure, label)
196324 where
distributed-process/src/Control/Distributed/Process/Internal/Types.hs+118−13
@@ -21,6 +21,7 @@ module Control.Distributed.Process.Internal.Types
2121 , SendPort(..)
2222 , ReceivePort(..)
2323 -- * Closures
24+ , StaticLabel(..)
2425 , Static(..)
2526 , Closure(..)
2627 , CallReply(..)
@@ -225,9 +226,28 @@ data ReceivePort a =
225226-- Closures --
226227--------------------------------------------------------------------------------
227228
229+data StaticLabel =
230+ UserStatic String
231+ -- Special support for 'call'
232+ | Call
233+ -- Generic closure combinators
234+ | ClosureApply
235+ | ClosureConst
236+ | ClosureUnit
237+ -- Arrow combinators for processes
238+ | CpId
239+ | CpComp
240+ | CpFirst
241+ | CpSwap
242+ | CpCopy
243+ | CpLeft
244+ | CpMirror
245+ | CpUntag
246+ | CpApply
247+ deriving (Typeable, Show)
248+
228249-- | A static value is one that is bound at top-level.
229−-- We represent it simply by a string
230−newtype Static a = Static String
250+newtype Static a = Static StaticLabel
231251 deriving (Typeable, Show)
232252
233253-- | A closure is a static value and an encoded environment
@@ -254,34 +274,83 @@ registerLabel label dyn = remoteTableLabel label ^= Just dyn
254274registerSender :: TypeRep -> Dynamic -> RemoteTable -> RemoteTable
255275registerSender typ dyn = remoteTableSender typ ^= Just dyn
256276
257−resolveClosure :: RemoteTable -> String -> ByteString -> Maybe Dynamic
258−resolveClosure rtable "$call" env = do
277+resolveClosure :: RemoteTable -> StaticLabel -> ByteString -> Maybe Dynamic
278+-- Special support for call
279+resolveClosure rtable Call env = do
259280 proc <- resolveClosure rtable label env'
260281 sender <- rtable ^. remoteTableSender (dynTypeRep proc)
261282 proc `bind` (sender `dynApp` toDyn (pid :: ProcessId))
262283 where
263284 (label, env', pid) = decode env
264− bind = dynBind tyConProcess bindProcess
265−resolveClosure rtable "$applyClosure" env = do
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
266290 f <- resolveClosure rtable labelf envf
267291 x <- resolveClosure rtable labelx envx
268292 f `dynApply` x
269293 where
270294 (labelf, envf, labelx, envx) = decode env
271−resolveClosure _rtable "$liftedBind" env =
272− return $ Dynamic (decode env) (unsafeCoerce# bindProcess)
273−resolveClosure _rtable "$liftedConst" env =
295+resolveClosure _rtable ClosureConst env =
274296 return $ Dynamic (decode env) (unsafeCoerce# const)
275−resolveClosure rtable label env = do
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 = undefined
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 = undefined
320+resolveClosure _rtable CpCopy env =
321+ return $ Dynamic (decode env) (unsafeCoerce# cpCopy)
322+ where
323+ cpCopy :: forall a. a -> Process (a, a)
324+ cpCopy = undefined
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 = undefined
330+resolveClosure _rtable CpMirror env =
331+ return $ Dynamic (decode env) (unsafeCoerce# cpMirror)
332+ where
333+ cpMirror :: forall a b. Either a b -> Process (Either b a)
334+ cpMirror = undefined
335+resolveClosure _rtable CpUntag env =
336+ return $ Dynamic (decode env) (unsafeCoerce# cpUntag)
337+ where
338+ cpUntag :: forall a. Either a a -> Process a
339+ cpUntag = undefined
340+resolveClosure _rtable CpApply env =
341+ return $ Dynamic (decode env) (unsafeCoerce# cpApply)
342+ where
343+ cpApply :: forall a b. Closure (Process (a -> Process b), a) -> Process b
344+ cpApply = undefined
345+
346+-- User defined closures
347+resolveClosure rtable (UserStatic label) env = do
276348 val <- rtable ^. remoteTableLabel label
277349 dynApply val (toDyn env)
278350
279351tyConProcess :: TyCon
280352tyConProcess = typeRepTyCon (typeOf (undefined :: Process ()))
281353
282−bindProcess :: Process a -> (a -> Process b) -> Process b
283−bindProcess = (>>=)
284−
285354--------------------------------------------------------------------------------
286355-- Messages --
287356--------------------------------------------------------------------------------
@@ -469,6 +538,42 @@ instance Binary Identifier where
469538 2 -> ChannelIdentifier <$> get
470539 _ -> fail "Identifier.get: invalid"
471540
541+instance Binary StaticLabel where
542+ put (UserStatic string) = putWord8 0 >> put string
543+ put Call = putWord8 1
544+ put ClosureApply = putWord8 2
545+ put ClosureConst = putWord8 3
546+ put ClosureUnit = putWord8 4
547+ put CpId = putWord8 5
548+ put CpComp = putWord8 6
549+ put CpFirst = putWord8 7
550+ put CpSwap = putWord8 8
551+ put CpCopy = putWord8 9
552+ put CpLeft = putWord8 10
553+ put CpMirror = putWord8 11
554+ put CpUntag = putWord8 12
555+ put CpApply = putWord8 13
556+ get = do
557+ header <- getWord8
558+ case header of
559+ 0 -> UserStatic <$> get
560+ 1 -> return Call
561+ 2 -> return ClosureApply
562+ 3 -> return ClosureConst
563+ 4 -> return ClosureUnit
564+ 5 -> return CpId
565+ 6 -> return CpComp
566+ 7 -> return CpFirst
567+ 8 -> return CpSwap
568+ 9 -> return CpCopy
569+ 10 -> return CpLeft
570+ 11 -> return CpMirror
571+ 12 -> return CpUntag
572+ 13 -> return CpApply
573+ _ -> fail "StaticLabel.get: invalid"
574+
575+
576+
472577--------------------------------------------------------------------------------
473578-- Accessors --
474579--------------------------------------------------------------------------------
distributed-process/tests/TestClosure.hs+3−3
@@ -147,10 +147,10 @@ testCall transport rtable = do
147147 takeMVar clientDone
148148
149149sendFac :: Int -> ProcessId -> Closure (Process ())
150−sendFac n pid = $(mkClosure 'factorial) n `bindCP` $(mkClosure 'sendInt) pid
150+sendFac n pid = $(mkClosure 'factorial) n `cpBind` $(mkClosure 'sendInt) pid
151151
152152factorial' :: Int -> Closure (Process Int)
153−factorial' n = $(mkClosure 'returnInt) n `bindCP` $(mkClosure 'factorialOf) ()
153+factorial' n = $(mkClosure 'returnInt) n `cpBind` $(mkClosure 'factorialOf) ()
154154
155155testBind :: Transport -> RemoteTable -> IO ()
156156testBind transport rtable = do
@@ -184,7 +184,7 @@ testSeq transport rtable = do
184184 node <- newLocalNode transport rtable
185185 runProcess node $ do
186186 us <- getSelfPid
187− join . unClosure $ sendFac 5 us `seqCP` sendFac 6 us
187+ join . unClosure $ sendFac 5 us `cpSeq` sendFac 6 us
188188 120 :: Int <- expect
189189 720 :: Int <- expect
190190 return ()