git.iko.soy/distributed-process
on 56158df help sign in
distributed-process/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs
300 lines 12 kB history raw
1{-# LANGUAGE ScopedTypeVariables #-}
2{-# LANGUAGE RankNTypes #-}
3module Control.Distributed.Process.Internal.Closure.BuiltIn
4 ( -- * Remote table
5 remoteTable
6 -- * Static dictionaries and associated operations
7 , staticDecode
8 , sdictUnit
9 , sdictProcessId
10 , sdictSendPort
11 -- * Some static values
12 , sndStatic
13 -- * The CP type and associated combinators
14 , CP
15 , idCP
16 , splitCP
17 , returnCP
18 , bindCP
19 , seqCP
20 -- * CP versions of Cloud Haskell primitives
21 , decodeProcessIdStatic
22 , cpLink
23 , cpUnlink
24 , cpRelay
25 , cpSend
26 , cpExpect
27 , cpNewChan
28 -- * Support for some CH operations
29 , cpDelay
30 , cpEnableTraceRemote
31 ) where
33import Data.ByteString.Lazy (ByteString)
34import Data.Binary (decode, encode)
35import Data.Rank1Typeable (Typeable, ANY, ANY1, ANY2, ANY3, ANY4)
36import Data.Rank1Dynamic (toDynamic)
37import Control.Distributed.Static
38 ( RemoteTable
39 , registerStatic
40 , Static
41 , staticLabel
42 , staticApply
43 , Closure
44 , closure
45 , closureApplyStatic
46 , closureApply
47 , staticCompose
48 , staticClosure
49 )
50import Control.Distributed.Process.Serializable
51 ( SerializableDict(..)
52 , Serializable
53 )
54import Control.Distributed.Process.Internal.Types
55 ( Process
56 , ProcessId
57 , SendPort
58 , ReceivePort
59 , ProcessMonitorNotification(ProcessMonitorNotification)
60 )
61import Control.Distributed.Process.Internal.Primitives
62 ( link
63 , unlink
64 , relay
65 , send
66 , expect
67 , newChan
68 , monitor
69 , unmonitor
70 , match
71 , matchIf
72 , receiveWait
73 )
75--------------------------------------------------------------------------------
76-- Remote table --
77--------------------------------------------------------------------------------
79remoteTable :: RemoteTable -> RemoteTable
80remoteTable =
81 registerStatic "$decodeDict" (toDynamic (decodeDict :: SerializableDict ANY -> ByteString -> ANY))
82 . registerStatic "$sdictUnit" (toDynamic (SerializableDict :: SerializableDict ()))
83 . registerStatic "$sdictProcessId" (toDynamic (SerializableDict :: SerializableDict ProcessId))
84 . registerStatic "$sdictSendPort_" (toDynamic (sdictSendPort_ :: SerializableDict ANY -> SerializableDict (SendPort ANY)))
85 . registerStatic "$returnProcess" (toDynamic (return :: ANY -> Process ANY))
86 . registerStatic "$seqProcess" (toDynamic ((>>) :: Process ANY1 -> Process ANY2 -> Process ANY2))
87 . registerStatic "$bindProcess" (toDynamic ((>>=) :: Process ANY1 -> (ANY1 -> Process ANY2) -> Process ANY2))
88 . registerStatic "$decodeProcessId" (toDynamic (decode :: ByteString -> ProcessId))
89 . registerStatic "$link" (toDynamic link)
90 . registerStatic "$unlink" (toDynamic unlink)
91 . registerStatic "$relay" (toDynamic relay)
92 . registerStatic "$sendDict" (toDynamic (sendDict :: SerializableDict ANY -> ProcessId -> ANY -> Process ()))
93 . registerStatic "$expectDict" (toDynamic (expectDict :: SerializableDict ANY -> Process ANY))
94 . registerStatic "$newChanDict" (toDynamic (newChanDict :: SerializableDict ANY -> Process (SendPort ANY, ReceivePort ANY)))
95 . registerStatic "$cpSplit" (toDynamic (cpSplit :: (ANY1 -> Process ANY3) -> (ANY2 -> Process ANY4) -> (ANY1, ANY2) -> Process (ANY3, ANY4)))
96 . registerStatic "$snd" (toDynamic (snd :: (ANY1, ANY2) -> ANY2))
97 . registerStatic "$delay" (toDynamic delay)
98 where
99 decodeDict :: forall a. SerializableDict a -> ByteString -> a
100 decodeDict SerializableDict = decode
102 sdictSendPort_ :: forall a. SerializableDict a -> SerializableDict (SendPort a)
103 sdictSendPort_ SerializableDict = SerializableDict
105 sendDict :: forall a. SerializableDict a -> ProcessId -> a -> Process ()
106 sendDict SerializableDict = send
108 expectDict :: forall a. SerializableDict a -> Process a
109 expectDict SerializableDict = expect
111 newChanDict :: forall a. SerializableDict a -> Process (SendPort a, ReceivePort a)
112 newChanDict SerializableDict = newChan
114 cpSplit :: forall a b c d. (a -> Process c) -> (b -> Process d) -> (a, b) -> Process (c, d)
115 cpSplit f g (a, b) = do
116 c <- f a
117 d <- g b
118 return (c, d)
120--------------------------------------------------------------------------------
121-- Static dictionaries and associated operations --
122--------------------------------------------------------------------------------
124-- | Static decoder, given a static serialization dictionary.
125--
126-- See module documentation of "Control.Distributed.Process.Closure" for an
127-- example.
128staticDecode :: Typeable a => Static (SerializableDict a) -> Static (ByteString -> a)
129staticDecode dict = decodeDictStatic `staticApply` dict
130 where
131 decodeDictStatic :: Typeable a => Static (SerializableDict a -> ByteString -> a)
132 decodeDictStatic = staticLabel "$decodeDict"
134-- | Serialization dictionary for '()'
135sdictUnit :: Static (SerializableDict ())
136sdictUnit = staticLabel "$sdictUnit"
138-- | Serialization dictionary for 'ProcessId'
139sdictProcessId :: Static (SerializableDict ProcessId)
140sdictProcessId = staticLabel "$sdictProcessId"
142-- | Serialization dictionary for 'SendPort'
143sdictSendPort :: Typeable a
144 => Static (SerializableDict a) -> Static (SerializableDict (SendPort a))
145sdictSendPort = staticApply (staticLabel "$sdictSendPort_")
147--------------------------------------------------------------------------------
148-- Static values --
149--------------------------------------------------------------------------------
151sndStatic :: Static ((a, b) -> b)
152sndStatic = staticLabel "$snd"
154--------------------------------------------------------------------------------
155-- The CP type and associated combinators --
156--------------------------------------------------------------------------------
158-- | @CP a b@ is a process with input of type @a@ and output of type @b@
159type CP a b = Closure (a -> Process b)
161returnProcessStatic :: Typeable a => Static (a -> Process a)
162returnProcessStatic = staticLabel "$returnProcess"
164-- | 'CP' version of 'Control.Category.id'
165idCP :: Typeable a => CP a a
166idCP = staticClosure returnProcessStatic
168-- | 'CP' version of ('Control.Arrow.***')
169splitCP :: (Typeable a, Typeable b, Typeable c, Typeable d)
170 => CP a c -> CP b d -> CP (a, b) (c, d)
171splitCP p q = cpSplitStatic `closureApplyStatic` p `closureApply` q
172 where
173 cpSplitStatic :: Static ((a -> Process c) -> (b -> Process d) -> (a, b) -> Process (c, d))
174 cpSplitStatic = staticLabel "$cpSplit"
176-- | 'CP' version of 'Control.Monad.return'
177returnCP :: forall a. Serializable a
178 => Static (SerializableDict a) -> a -> Closure (Process a)
179returnCP dict x = closure decoder (encode x)
180 where
181 decoder :: Static (ByteString -> Process a)
182 decoder = returnProcessStatic
183 `staticCompose`
184 staticDecode dict
186-- | 'CP' version of ('Control.Monad.>>')
187seqCP :: (Typeable a, Typeable b)
188 => Closure (Process a) -> Closure (Process b) -> Closure (Process b)
189seqCP p q = seqProcessStatic `closureApplyStatic` p `closureApply` q
190 where
191 seqProcessStatic :: (Typeable a, Typeable b)
192 => Static (Process a -> Process b -> Process b)
193 seqProcessStatic = staticLabel "$seqProcess"
195-- | (Not quite the) 'CP' version of ('Control.Monad.>>=')
196bindCP :: forall a b. (Typeable a, Typeable b)
197 => Closure (Process a) -> CP a b -> Closure (Process b)
198bindCP x f = bindProcessStatic `closureApplyStatic` x `closureApply` f
199 where
200 bindProcessStatic :: (Typeable a, Typeable b)
201 => Static (Process a -> (a -> Process b) -> Process b)
202 bindProcessStatic = staticLabel "$bindProcess"
204--------------------------------------------------------------------------------
205-- CP versions of Cloud Haskell primitives --
206--------------------------------------------------------------------------------
208decodeProcessIdStatic :: Static (ByteString -> ProcessId)
209decodeProcessIdStatic = staticLabel "$decodeProcessId"
211-- | 'CP' version of 'link'
212cpLink :: ProcessId -> Closure (Process ())
213cpLink = closure (linkStatic `staticCompose` decodeProcessIdStatic) . encode
214 where
215 linkStatic :: Static (ProcessId -> Process ())
216 linkStatic = staticLabel "$link"
218-- | 'CP' version of 'unlink'
219cpUnlink :: ProcessId -> Closure (Process ())
220cpUnlink = closure (unlinkStatic `staticCompose` decodeProcessIdStatic) . encode
221 where
222 unlinkStatic :: Static (ProcessId -> Process ())
223 unlinkStatic = staticLabel "$unlink"
225-- | 'CP' version of 'send'
226cpSend :: forall a. Typeable a
227 => Static (SerializableDict a) -> ProcessId -> CP a ()
228cpSend dict pid = closure decoder (encode pid)
229 where
230 decoder :: Static (ByteString -> a -> Process ())
231 decoder = (sendDictStatic `staticApply` dict)
232 `staticCompose`
233 decodeProcessIdStatic
235 sendDictStatic :: Typeable a
236 => Static (SerializableDict a -> ProcessId -> a -> Process ())
237 sendDictStatic = staticLabel "$sendDict"
239-- | 'CP' version of 'expect'
240cpExpect :: Typeable a => Static (SerializableDict a) -> Closure (Process a)
241cpExpect dict = staticClosure (expectDictStatic `staticApply` dict)
242 where
243 expectDictStatic :: Typeable a => Static (SerializableDict a -> Process a)
244 expectDictStatic = staticLabel "$expectDict"
246-- | 'CP' version of 'newChan'
247cpNewChan :: Typeable a
248 => Static (SerializableDict a)
249 -> Closure (Process (SendPort a, ReceivePort a))
250cpNewChan dict = staticClosure (newChanDictStatic `staticApply` dict)
251 where
252 newChanDictStatic :: Typeable a
253 => Static (SerializableDict a -> Process (SendPort a, ReceivePort a))
254 newChanDictStatic = staticLabel "$newChanDict"
256-- | 'CP' version of 'relay'
257cpRelay :: ProcessId -> Closure (Process ())
258cpRelay = closure (relayStatic `staticCompose` decodeProcessIdStatic) . encode
259 where
260 relayStatic :: Static (ProcessId -> Process ())
261 relayStatic = staticLabel "$relay"
263-- TODO: move cpEnableTraceRemote into Trace/Primitives.hs
265cpEnableTraceRemote :: ProcessId -> Closure (Process ())
266cpEnableTraceRemote =
267 closure (enableTraceStatic `staticCompose` decodeProcessIdStatic) . encode
268 where
269 enableTraceStatic :: Static (ProcessId -> Process ())
270 enableTraceStatic = staticLabel "$enableTraceRemote"
272--------------------------------------------------------------------------------
273-- Support for spawn --
274--------------------------------------------------------------------------------
276-- | @delay them p@ is a process that waits for a signal (a message of type @()@)
277-- from 'them' (origin is not verified) before proceeding as @p@. In order to
278-- avoid waiting forever, @delay them p@ monitors 'them'. If it receives a
279-- monitor message instead it simply terminates.
280delay :: ProcessId -> Process () -> Process ()
281delay them p = do
282 ref <- monitor them
283 let sameRef (ProcessMonitorNotification ref' _ _) = ref == ref'
284 receiveWait [
285 match $ \() -> unmonitor ref >> p
286 , matchIf sameRef $ \_ -> return ()
287 ]
289-- | 'CP' version of 'delay'
290cpDelay :: ProcessId -> Closure (Process ()) -> Closure (Process ())
291cpDelay = closureApply . cpDelay'
292 where
293 cpDelay' :: ProcessId -> Closure (Process () -> Process ())
294 cpDelay' pid = closure decoder (encode pid)
296 decoder :: Static (ByteString -> Process () -> Process ())
297 decoder = delayStatic `staticCompose` decodeProcessIdStatic
299 delayStatic :: Static (ProcessId -> Process () -> Process ())
300 delayStatic = staticLabel "$delay"