git.iko.soy/distributed-process
on 87bb129 help sign in
distributed-process/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs
320 lines 13 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 , sdictStatic
12 , sdictClosure
13 -- * Some static values
14 , sndStatic
15 -- * The CP type and associated combinators
16 , CP
17 , idCP
18 , splitCP
19 , returnCP
20 , bindCP
21 , seqCP
22 -- * CP versions of Cloud Haskell primitives
23 , decodeProcessIdStatic
24 , cpLink
25 , cpUnlink
26 , cpRelay
27 , cpSend
28 , cpExpect
29 , cpNewChan
30 -- * Support for some CH operations
31 , cpDelayed
32 , cpEnableTraceRemote
33 ) where
35import Data.ByteString.Lazy (ByteString)
36import Data.Binary (decode, encode)
37import Data.Rank1Typeable (Typeable, ANY, ANY1, ANY2, ANY3, ANY4)
38import Data.Rank1Dynamic (toDynamic)
39import Control.Distributed.Static
40 ( RemoteTable
41 , registerStatic
42 , Static
43 , staticLabel
44 , staticApply
45 , Closure
46 , closure
47 , closureApplyStatic
48 , closureApply
49 , staticCompose
50 , staticClosure
51 )
52import Control.Distributed.Process.Serializable
53 ( SerializableDict(..)
54 , Serializable
55 , TypeableDict(..)
56 )
57import Control.Distributed.Process.Internal.Types
58 ( Process
59 , ProcessId
60 , SendPort
61 , ReceivePort
62 , ProcessMonitorNotification(ProcessMonitorNotification)
63 )
64import Control.Distributed.Process.Internal.Primitives
65 ( link
66 , unlink
67 , relay
68 , send
69 , expect
70 , newChan
71 , monitor
72 , unmonitor
73 , match
74 , matchIf
75 , receiveWait
76 )
78--------------------------------------------------------------------------------
79-- Remote table --
80--------------------------------------------------------------------------------
82remoteTable :: RemoteTable -> RemoteTable
83remoteTable =
84 registerStatic "$decodeDict" (toDynamic (decodeDict :: SerializableDict ANY -> ByteString -> ANY))
85 . registerStatic "$sdictUnit" (toDynamic (SerializableDict :: SerializableDict ()))
86 . registerStatic "$sdictProcessId" (toDynamic (SerializableDict :: SerializableDict ProcessId))
87 . registerStatic "$sdictSendPort_" (toDynamic (sdictSendPort_ :: SerializableDict ANY -> SerializableDict (SendPort ANY)))
88 . registerStatic "$sdictStatic" (toDynamic (sdictStatic_ :: TypeableDict ANY -> SerializableDict (Static ANY)))
89 . registerStatic "$sdictClosure" (toDynamic (sdictClosure_ :: TypeableDict ANY -> SerializableDict (Closure ANY)))
90 . registerStatic "$returnProcess" (toDynamic (return :: ANY -> Process ANY))
91 . registerStatic "$seqProcess" (toDynamic ((>>) :: Process ANY1 -> Process ANY2 -> Process ANY2))
92 . registerStatic "$bindProcess" (toDynamic ((>>=) :: Process ANY1 -> (ANY1 -> Process ANY2) -> Process ANY2))
93 . registerStatic "$decodeProcessId" (toDynamic (decode :: ByteString -> ProcessId))
94 . registerStatic "$link" (toDynamic link)
95 . registerStatic "$unlink" (toDynamic unlink)
96 . registerStatic "$relay" (toDynamic relay)
97 . registerStatic "$sendDict" (toDynamic (sendDict :: SerializableDict ANY -> ProcessId -> ANY -> Process ()))
98 . registerStatic "$expectDict" (toDynamic (expectDict :: SerializableDict ANY -> Process ANY))
99 . registerStatic "$newChanDict" (toDynamic (newChanDict :: SerializableDict ANY -> Process (SendPort ANY, ReceivePort ANY)))
100 . registerStatic "$cpSplit" (toDynamic (cpSplit :: (ANY1 -> Process ANY3) -> (ANY2 -> Process ANY4) -> (ANY1, ANY2) -> Process (ANY3, ANY4)))
101 . registerStatic "$snd" (toDynamic (snd :: (ANY1, ANY2) -> ANY2))
102 . registerStatic "$delay" (toDynamic delay)
103 where
104 decodeDict :: forall a. SerializableDict a -> ByteString -> a
105 decodeDict SerializableDict = decode
107 sdictSendPort_ :: forall a. SerializableDict a -> SerializableDict (SendPort a)
108 sdictSendPort_ SerializableDict = SerializableDict
110 sdictStatic_ :: forall a. TypeableDict a -> SerializableDict (Static a)
111 sdictStatic_ TypeableDict = SerializableDict
113 sdictClosure_ :: forall a. TypeableDict a -> SerializableDict (Closure a)
114 sdictClosure_ TypeableDict = SerializableDict
116 sendDict :: forall a. SerializableDict a -> ProcessId -> a -> Process ()
117 sendDict SerializableDict = send
119 expectDict :: forall a. SerializableDict a -> Process a
120 expectDict SerializableDict = expect
122 newChanDict :: forall a. SerializableDict a -> Process (SendPort a, ReceivePort a)
123 newChanDict SerializableDict = newChan
125 cpSplit :: forall a b c d. (a -> Process c) -> (b -> Process d) -> (a, b) -> Process (c, d)
126 cpSplit f g (a, b) = do
127 c <- f a
128 d <- g b
129 return (c, d)
131--------------------------------------------------------------------------------
132-- Static dictionaries and associated operations --
133--------------------------------------------------------------------------------
135-- | Static decoder, given a static serialization dictionary.
136--
137-- See module documentation of "Control.Distributed.Process.Closure" for an
138-- example.
139staticDecode :: Typeable a => Static (SerializableDict a) -> Static (ByteString -> a)
140staticDecode dict = decodeDictStatic `staticApply` dict
141 where
142 decodeDictStatic :: Typeable a => Static (SerializableDict a -> ByteString -> a)
143 decodeDictStatic = staticLabel "$decodeDict"
145-- | Serialization dictionary for '()'
146sdictUnit :: Static (SerializableDict ())
147sdictUnit = staticLabel "$sdictUnit"
149-- | Serialization dictionary for 'ProcessId'
150sdictProcessId :: Static (SerializableDict ProcessId)
151sdictProcessId = staticLabel "$sdictProcessId"
153-- | Serialization dictionary for 'SendPort'
154sdictSendPort :: Typeable a
155 => Static (SerializableDict a) -> Static (SerializableDict (SendPort a))
156sdictSendPort = staticApply (staticLabel "$sdictSendPort_")
158-- | Serialization dictionary for 'Static'.
159sdictStatic :: Typeable a => Static (TypeableDict a) -> Static (SerializableDict (Static a))
160sdictStatic = staticApply (staticLabel "$sdictStatic")
162-- | Serialization dictionary for 'Closure'.
163sdictClosure :: Typeable a => Static (TypeableDict a) -> Static (SerializableDict (Closure a))
164sdictClosure = staticApply (staticLabel "$sdictClosure")
166--------------------------------------------------------------------------------
167-- Static values --
168--------------------------------------------------------------------------------
170sndStatic :: Static ((a, b) -> b)
171sndStatic = staticLabel "$snd"
173--------------------------------------------------------------------------------
174-- The CP type and associated combinators --
175--------------------------------------------------------------------------------
177-- | @CP a b@ is a process with input of type @a@ and output of type @b@
178type CP a b = Closure (a -> Process b)
180returnProcessStatic :: Typeable a => Static (a -> Process a)
181returnProcessStatic = staticLabel "$returnProcess"
183-- | 'CP' version of 'Control.Category.id'
184idCP :: Typeable a => CP a a
185idCP = staticClosure returnProcessStatic
187-- | 'CP' version of ('Control.Arrow.***')
188splitCP :: (Typeable a, Typeable b, Typeable c, Typeable d)
189 => CP a c -> CP b d -> CP (a, b) (c, d)
190splitCP p q = cpSplitStatic `closureApplyStatic` p `closureApply` q
191 where
192 cpSplitStatic :: Static ((a -> Process c) -> (b -> Process d) -> (a, b) -> Process (c, d))
193 cpSplitStatic = staticLabel "$cpSplit"
195-- | 'CP' version of 'Control.Monad.return'
196returnCP :: forall a. Serializable a
197 => Static (SerializableDict a) -> a -> Closure (Process a)
198returnCP dict x = closure decoder (encode x)
199 where
200 decoder :: Static (ByteString -> Process a)
201 decoder = returnProcessStatic
202 `staticCompose`
203 staticDecode dict
205-- | 'CP' version of ('Control.Monad.>>')
206seqCP :: (Typeable a, Typeable b)
207 => Closure (Process a) -> Closure (Process b) -> Closure (Process b)
208seqCP p q = seqProcessStatic `closureApplyStatic` p `closureApply` q
209 where
210 seqProcessStatic :: (Typeable a, Typeable b)
211 => Static (Process a -> Process b -> Process b)
212 seqProcessStatic = staticLabel "$seqProcess"
214-- | (Not quite the) 'CP' version of ('Control.Monad.>>=')
215bindCP :: forall a b. (Typeable a, Typeable b)
216 => Closure (Process a) -> CP a b -> Closure (Process b)
217bindCP x f = bindProcessStatic `closureApplyStatic` x `closureApply` f
218 where
219 bindProcessStatic :: (Typeable a, Typeable b)
220 => Static (Process a -> (a -> Process b) -> Process b)
221 bindProcessStatic = staticLabel "$bindProcess"
223--------------------------------------------------------------------------------
224-- CP versions of Cloud Haskell primitives --
225--------------------------------------------------------------------------------
227decodeProcessIdStatic :: Static (ByteString -> ProcessId)
228decodeProcessIdStatic = staticLabel "$decodeProcessId"
230-- | 'CP' version of 'link'
231cpLink :: ProcessId -> Closure (Process ())
232cpLink = closure (linkStatic `staticCompose` decodeProcessIdStatic) . encode
233 where
234 linkStatic :: Static (ProcessId -> Process ())
235 linkStatic = staticLabel "$link"
237-- | 'CP' version of 'unlink'
238cpUnlink :: ProcessId -> Closure (Process ())
239cpUnlink = closure (unlinkStatic `staticCompose` decodeProcessIdStatic) . encode
240 where
241 unlinkStatic :: Static (ProcessId -> Process ())
242 unlinkStatic = staticLabel "$unlink"
244-- | 'CP' version of 'send'
245cpSend :: forall a. Typeable a
246 => Static (SerializableDict a) -> ProcessId -> CP a ()
247cpSend dict pid = closure decoder (encode pid)
248 where
249 decoder :: Static (ByteString -> a -> Process ())
250 decoder = (sendDictStatic `staticApply` dict)
251 `staticCompose`
252 decodeProcessIdStatic
254 sendDictStatic :: Typeable a
255 => Static (SerializableDict a -> ProcessId -> a -> Process ())
256 sendDictStatic = staticLabel "$sendDict"
258-- | 'CP' version of 'expect'
259cpExpect :: Typeable a => Static (SerializableDict a) -> Closure (Process a)
260cpExpect dict = staticClosure (expectDictStatic `staticApply` dict)
261 where
262 expectDictStatic :: Typeable a => Static (SerializableDict a -> Process a)
263 expectDictStatic = staticLabel "$expectDict"
265-- | 'CP' version of 'newChan'
266cpNewChan :: Typeable a
267 => Static (SerializableDict a)
268 -> Closure (Process (SendPort a, ReceivePort a))
269cpNewChan dict = staticClosure (newChanDictStatic `staticApply` dict)
270 where
271 newChanDictStatic :: Typeable a
272 => Static (SerializableDict a -> Process (SendPort a, ReceivePort a))
273 newChanDictStatic = staticLabel "$newChanDict"
275-- | 'CP' version of 'relay'
276cpRelay :: ProcessId -> Closure (Process ())
277cpRelay = closure (relayStatic `staticCompose` decodeProcessIdStatic) . encode
278 where
279 relayStatic :: Static (ProcessId -> Process ())
280 relayStatic = staticLabel "$relay"
282-- TODO: move cpEnableTraceRemote into Trace/Primitives.hs
284cpEnableTraceRemote :: ProcessId -> Closure (Process ())
285cpEnableTraceRemote =
286 closure (enableTraceStatic `staticCompose` decodeProcessIdStatic) . encode
287 where
288 enableTraceStatic :: Static (ProcessId -> Process ())
289 enableTraceStatic = staticLabel "$enableTraceRemote"
291--------------------------------------------------------------------------------
292-- Support for spawn --
293--------------------------------------------------------------------------------
295-- | @delay them p@ is a process that waits for a signal (a message of type @()@)
296-- from 'them' (origin is not verified) before proceeding as @p@. In order to
297-- avoid waiting forever, @delay them p@ monitors 'them'. If it receives a
298-- monitor message instead, it proceeds as @p@ too.
299delay :: ProcessId -> Process () -> Process ()
300delay them p = do
301 ref <- monitor them
302 let sameRef (ProcessMonitorNotification ref' _ _) = ref == ref'
303 receiveWait [
304 match $ \() -> unmonitor ref
305 , matchIf sameRef $ \_ -> return ()
306 ]
307 p
309-- | 'CP' version of 'delay'
310cpDelayed :: ProcessId -> Closure (Process ()) -> Closure (Process ())
311cpDelayed = closureApply . cpDelay'
312 where
313 cpDelay' :: ProcessId -> Closure (Process () -> Process ())
314 cpDelay' pid = closure decoder (encode pid)
316 decoder :: Static (ByteString -> Process () -> Process ())
317 decoder = delayStatic `staticCompose` decodeProcessIdStatic
319 delayStatic :: Static (ProcessId -> Process () -> Process ())
320 delayStatic = staticLabel "$delay"