git.iko.soy/distributed-process
on master help sign in
spike adding cpSendChan
590ab79Tim Watson
src/Control/Distributed/Process.hs+9−6
compare with master
unifiedsplit
@@ -1,3 +1,5 @@
1+{-# LANGUAGE ScopedTypeVariables #-}
2+
1{- | [Cloud Haskell]
3{- | [Cloud Haskell]
2
4
3This is an implementation of Cloud Haskell, as described in
5This is an implementation of Cloud Haskell, as described in
@@ -179,6 +181,7 @@ import Control.Distributed.Process.Internal.Closure.BuiltIn
179 , splitCP
181 , splitCP
180 , cpLink
182 , cpLink
181 , cpSend
183 , cpSend
184+ , cpSendChan
182 , cpNewChan
185 , cpNewChan
183 , cpDelay
186 , cpDelay
184 )
187 )
@@ -421,14 +424,14 @@ spawnChannel :: forall a. Typeable a => Static (SerializableDict a)
421 -> Closure (ReceivePort a -> Process ())
424 -> Closure (ReceivePort a -> Process ())
422 -> Process (SendPort a)
425 -> Process (SendPort a)
423spawnChannel dict nid proc = do
426spawnChannel dict nid proc = do
424− us <- getSelfPid
427+ (sp, rp) <- newChan
425− _ <- spawn nid (go us)
428+ _ <- spawn nid (go sp)
426− expect
429+ r <- receiveChan rp
427 where
430 where
428− go :: ProcessId -> Closure (Process ())
431+ go :: SendPort (SendPort a) -> Closure (Process ())
429− go pid = cpNewChan dict
432+ go chan = cpNewChan dict
430 `bindCP`
433 `bindCP`
431− (cpSend (sdictSendPort dict) pid `splitCP` proc)
434+ (cpSendChan (sdictSendPort dict) chan `splitCP` proc)
432 `bindCP`
435 `bindCP`
433 (idCP `closureCompose` staticClosure sndStatic)
436 (idCP `closureCompose` staticClosure sndStatic)
434
437
src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs+25
@@ -19,6 +19,7 @@ module Control.Distributed.Process.Internal.Closure.BuiltIn
19 , cpLink
19 , cpLink
20 , cpUnlink
20 , cpUnlink
21 , cpSend
21 , cpSend
22+ , cpSendChan
22 , cpExpect
23 , cpExpect
23 , cpNewChan
24 , cpNewChan
24 -- * Support for some CH operations
25 -- * Support for some CH operations
@@ -59,6 +60,7 @@ import Control.Distributed.Process.Internal.Primitives
59 , send
60 , send
60 , expect
61 , expect
61 , newChan
62 , newChan
63+ , sendChan
62 , monitor
64 , monitor
63 , unmonitor
65 , unmonitor
64 , match
66 , match
@@ -83,6 +85,7 @@ remoteTable =
83 . registerStatic "$link" (toDynamic link)
85 . registerStatic "$link" (toDynamic link)
84 . registerStatic "$unlink" (toDynamic unlink)
86 . registerStatic "$unlink" (toDynamic unlink)
85 . registerStatic "$sendDict" (toDynamic (sendDict :: SerializableDict ANY -> ProcessId -> ANY -> Process ()))
87 . registerStatic "$sendDict" (toDynamic (sendDict :: SerializableDict ANY -> ProcessId -> ANY -> Process ()))
88+ . registerStatic "$sendChanDict" (toDynamic (sendChanDict :: SerializableDict ANY -> (SendPort ANY) -> ANY -> Process ()))
86 . registerStatic "$expectDict" (toDynamic (expectDict :: SerializableDict ANY -> Process ANY))
89 . registerStatic "$expectDict" (toDynamic (expectDict :: SerializableDict ANY -> Process ANY))
87 . registerStatic "$newChanDict" (toDynamic (newChanDict :: SerializableDict ANY -> Process (SendPort ANY, ReceivePort ANY)))
90 . registerStatic "$newChanDict" (toDynamic (newChanDict :: SerializableDict ANY -> Process (SendPort ANY, ReceivePort ANY)))
88 . registerStatic "$cpSplit" (toDynamic (cpSplit :: (ANY1 -> Process ANY3) -> (ANY2 -> Process ANY4) -> (ANY1, ANY2) -> Process (ANY3, ANY4)))
91 . registerStatic "$cpSplit" (toDynamic (cpSplit :: (ANY1 -> Process ANY3) -> (ANY2 -> Process ANY4) -> (ANY1, ANY2) -> Process (ANY3, ANY4)))
@@ -98,6 +101,12 @@ remoteTable =
98 sendDict :: forall a. SerializableDict a -> ProcessId -> a -> Process ()
101 sendDict :: forall a. SerializableDict a -> ProcessId -> a -> Process ()
99 sendDict SerializableDict = send
102 sendDict SerializableDict = send
100
103
104+ sendChanDict :: forall a . SerializableDict a
105+ -> SendPort a
106+ -> a
107+ -> Process ()
108+ sendChanDict SerializableDict = sendChan
109+
101 expectDict :: forall a. SerializableDict a -> Process a
110 expectDict :: forall a. SerializableDict a -> Process a
102 expectDict SerializableDict = expect
111 expectDict SerializableDict = expect
103
112
@@ -201,6 +210,9 @@ bindCP x f = bindProcessStatic `closureApplyStatic` x `closureApply` f
201decodeProcessIdStatic :: Static (ByteString -> ProcessId)
210decodeProcessIdStatic :: Static (ByteString -> ProcessId)
202decodeProcessIdStatic = staticLabel "$decodeProcessId"
211decodeProcessIdStatic = staticLabel "$decodeProcessId"
203
212
213+decodeSendPortStatic :: forall a . Typeable a => Static (ByteString -> SendPort a)
214+decodeSendPortStatic = staticLabel "$sdictSendPort_"
215+
204-- | 'CP' version of 'link'
216-- | 'CP' version of 'link'
205cpLink :: ProcessId -> Closure (Process ())
217cpLink :: ProcessId -> Closure (Process ())
206cpLink = closure (linkStatic `staticCompose` decodeProcessIdStatic) . encode
218cpLink = closure (linkStatic `staticCompose` decodeProcessIdStatic) . encode
@@ -229,6 +241,19 @@ cpSend dict pid = closure decoder (encode pid)
229 => Static (SerializableDict a -> ProcessId -> a -> Process ())
241 => Static (SerializableDict a -> ProcessId -> a -> Process ())
230 sendDictStatic = staticLabel "$sendDict"
242 sendDictStatic = staticLabel "$sendDict"
231
243
244+cpSendChan :: forall a. Typeable a
245+ => Static (SerializableDict a) -> SendPort a -> CP a ()
246+cpSendChan dict sp = closure decoder (encode sp)
247+ where
248+ decoder :: Static (ByteString -> a -> Process ())
249+ decoder = (sendChanDictStatic `staticApply` dict)
250+ `staticCompose`
251+ decodeSendPortStatic
252+
253+ sendChanDictStatic :: Typeable a
254+ => Static (SerializableDict a -> SendPort a -> a -> Process ())
255+ sendChanDictStatic = staticLabel "$sendChanDict"
256+
232-- | 'CP' version of 'expect'
257-- | 'CP' version of 'expect'
233cpExpect :: Typeable a => Static (SerializableDict a) -> Closure (Process a)
258cpExpect :: Typeable a => Static (SerializableDict a) -> Closure (Process a)
234cpExpect dict = staticClosure (expectDictStatic `staticApply` dict)
259cpExpect dict = staticClosure (expectDictStatic `staticApply` dict)