git.iko.soy/distributed-process
on 590ab79 help sign in
660d554
swap
1 ahead, 469 behind
2 files changed+34−6
since splitsnapshots
unifiedsplit
src/Control/Distributed/Process.hs+9−6 src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs+25
src/Control/Distributed/Process.hs+9−6
@@ -1,3 +1,5 @@
1+{-# LANGUAGE ScopedTypeVariables #-}
2+
13{- | [Cloud Haskell]
24
35This is an implementation of Cloud Haskell, as described in
@@ -179,6 +181,7 @@ import Control.Distributed.Process.Internal.Closure.BuiltIn
179181 , splitCP
180182 , cpLink
181183 , cpSend
184+ , cpSendChan
182185 , cpNewChan
183186 , cpDelay
184187 )
@@ -421,14 +424,14 @@ spawnChannel :: forall a. Typeable a => Static (SerializableDict a)
421424 -> Closure (ReceivePort a -> Process ())
422425 -> Process (SendPort a)
423426spawnChannel dict nid proc = do
424− us <- getSelfPid
425− _ <- spawn nid (go us)
426− expect
427+ (sp, rp) <- newChan
428+ _ <- spawn nid (go sp)
429+ r <- receiveChan rp
427430 where
428− go :: ProcessId -> Closure (Process ())
429− go pid = cpNewChan dict
431+ go :: SendPort (SendPort a) -> Closure (Process ())
432+ go chan = cpNewChan dict
430433 `bindCP`
431− (cpSend (sdictSendPort dict) pid `splitCP` proc)
434+ (cpSendChan (sdictSendPort dict) chan `splitCP` proc)
432435 `bindCP`
433436 (idCP `closureCompose` staticClosure sndStatic)
434437
src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs+25
@@ -19,6 +19,7 @@ module Control.Distributed.Process.Internal.Closure.BuiltIn
1919 , cpLink
2020 , cpUnlink
2121 , cpSend
22+ , cpSendChan
2223 , cpExpect
2324 , cpNewChan
2425 -- * Support for some CH operations
@@ -59,6 +60,7 @@ import Control.Distributed.Process.Internal.Primitives
5960 , send
6061 , expect
6162 , newChan
63+ , sendChan
6264 , monitor
6365 , unmonitor
6466 , match
@@ -83,6 +85,7 @@ remoteTable =
8385 . registerStatic "$link" (toDynamic link)
8486 . registerStatic "$unlink" (toDynamic unlink)
8587 . registerStatic "$sendDict" (toDynamic (sendDict :: SerializableDict ANY -> ProcessId -> ANY -> Process ()))
88+ . registerStatic "$sendChanDict" (toDynamic (sendChanDict :: SerializableDict ANY -> (SendPort ANY) -> ANY -> Process ()))
8689 . registerStatic "$expectDict" (toDynamic (expectDict :: SerializableDict ANY -> Process ANY))
8790 . registerStatic "$newChanDict" (toDynamic (newChanDict :: SerializableDict ANY -> Process (SendPort ANY, ReceivePort ANY)))
8891 . registerStatic "$cpSplit" (toDynamic (cpSplit :: (ANY1 -> Process ANY3) -> (ANY2 -> Process ANY4) -> (ANY1, ANY2) -> Process (ANY3, ANY4)))
@@ -98,6 +101,12 @@ remoteTable =
98101 sendDict :: forall a. SerializableDict a -> ProcessId -> a -> Process ()
99102 sendDict SerializableDict = send
100103
104+ sendChanDict :: forall a . SerializableDict a
105+ -> SendPort a
106+ -> a
107+ -> Process ()
108+ sendChanDict SerializableDict = sendChan
109+
101110 expectDict :: forall a. SerializableDict a -> Process a
102111 expectDict SerializableDict = expect
103112
@@ -201,6 +210,9 @@ bindCP x f = bindProcessStatic `closureApplyStatic` x `closureApply` f
201210decodeProcessIdStatic :: Static (ByteString -> ProcessId)
202211decodeProcessIdStatic = staticLabel "$decodeProcessId"
203212
213+decodeSendPortStatic :: forall a . Typeable a => Static (ByteString -> SendPort a)
214+decodeSendPortStatic = staticLabel "$sdictSendPort_"
215+
204216-- | 'CP' version of 'link'
205217cpLink :: ProcessId -> Closure (Process ())
206218cpLink = closure (linkStatic `staticCompose` decodeProcessIdStatic) . encode
@@ -229,6 +241,19 @@ cpSend dict pid = closure decoder (encode pid)
229241 => Static (SerializableDict a -> ProcessId -> a -> Process ())
230242 sendDictStatic = staticLabel "$sendDict"
231243
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+
232257-- | 'CP' version of 'expect'
233258cpExpect :: Typeable a => Static (SerializableDict a) -> Closure (Process a)
234259cpExpect dict = staticClosure (expectDictStatic `staticApply` dict)