@@ -1,10 +1,9 @@
11module Main where
22
33import Prelude hiding (catch)
4−import Data.Binary (Binary)
4+import Data.Binary (Binary(..))
55import Data.Typeable (Typeable)
66import Data.Foldable (forM_)
7−import qualified Data.Map as Map (empty)
87import Control.Concurrent (forkIO, threadDelay)
98import Control.Concurrent.MVar ( MVar
109 , newEmptyMVar
@@ -15,7 +14,7 @@ import Control.Concurrent.MVar ( MVar
1514import Control.Monad (replicateM_)
1615import Control.Monad.IO.Class (liftIO)
1716import Control.Exception (throwIO)
18−import Control.Applicative ((<$>))
17+import Control.Applicative ((<$>), (<*>))
1918import qualified Network.Transport as NT (Transport, closeEndPoint)
2019import Network.Transport.TCP (createTransport, defaultTCPParameters)
2120import Control.Distributed.Process
@@ -44,13 +43,13 @@ testPing transport = do
4443
4544 -- Server
4645 forkIO $ do
47− localNode <- newLocalNode transport Map.empty
46+ localNode <- newLocalNode transport initRemoteCallMetaData
4847 addr <- forkProcess localNode ping
4948 putMVar serverAddr addr
5049
5150 -- Client
5251 forkIO $ do
53− localNode <- newLocalNode transport Map.empty
52+ localNode <- newLocalNode transport initRemoteCallMetaData
5453 pingServer <- readMVar serverAddr
5554
5655 let numPings = 10000
@@ -118,13 +117,13 @@ testMonitorUnreachable transport mOrL un = do
118117 done <- newEmptyMVar
119118
120119 forkIO $ do
121− localNode <- newLocalNode transport Map.empty
120+ localNode <- newLocalNode transport initRemoteCallMetaData
122121 addr <- forkProcess localNode . liftIO $ threadDelay 1000000
123122 closeLocalNode localNode
124123 putMVar deadProcess addr
125124
126125 forkIO $ do
127− localNode <- newLocalNode transport Map.empty
126+ localNode <- newLocalNode transport initRemoteCallMetaData
128127 theirAddr <- readMVar deadProcess
129128 runProcess localNode $
130129 monitorTestProcess theirAddr mOrL un DiedDisconnect Nothing done
@@ -139,13 +138,13 @@ testMonitorNormalTermination transport mOrL un = do
139138 done <- newEmptyMVar
140139
141140 forkIO $ do
142− localNode <- newLocalNode transport Map.empty
141+ localNode <- newLocalNode transport initRemoteCallMetaData
143142 addr <- forkProcess localNode $
144143 liftIO $ readMVar monitorSetup
145144 putMVar monitoredProcess addr
146145
147146 forkIO $ do
148− localNode <- newLocalNode transport Map.empty
147+ localNode <- newLocalNode transport initRemoteCallMetaData
149148 theirAddr <- readMVar monitoredProcess
150149 runProcess localNode $
151150 monitorTestProcess theirAddr mOrL un DiedNormal (Just monitorSetup) done
@@ -162,14 +161,14 @@ testMonitorAbnormalTermination transport mOrL un = do
162161 let err = userError "Abnormal termination"
163162
164163 forkIO $ do
165− localNode <- newLocalNode transport Map.empty
164+ localNode <- newLocalNode transport initRemoteCallMetaData
166165 addr <- forkProcess localNode . liftIO $ do
167166 readMVar monitorSetup
168167 throwIO err
169168 putMVar monitoredProcess addr
170169
171170 forkIO $ do
172− localNode <- newLocalNode transport Map.empty
171+ localNode <- newLocalNode transport initRemoteCallMetaData
173172 theirAddr <- readMVar monitoredProcess
174173 runProcess localNode $
175174 monitorTestProcess theirAddr mOrL un (DiedException (show err)) (Just monitorSetup) done
@@ -181,7 +180,7 @@ testMonitorLocalDeadProcess :: NT.Transport -> Bool -> Bool -> IO ()
181180testMonitorLocalDeadProcess transport mOrL un = do
182181 processDead <- newEmptyMVar
183182 processAddr <- newEmptyMVar
184− localNode <- newLocalNode transport Map.empty
183+ localNode <- newLocalNode transport initRemoteCallMetaData
185184 done <- newEmptyMVar
186185
187186 forkIO $ do
@@ -204,12 +203,12 @@ testMonitorRemoteDeadProcess transport mOrL un = do
204203 done <- newEmptyMVar
205204
206205 forkIO $ do
207− localNode <- newLocalNode transport Map.empty
206+ localNode <- newLocalNode transport initRemoteCallMetaData
208207 addr <- forkProcess localNode . liftIO $ putMVar processDead ()
209208 putMVar processAddr addr
210209
211210 forkIO $ do
212− localNode <- newLocalNode transport Map.empty
211+ localNode <- newLocalNode transport initRemoteCallMetaData
213212 theirAddr <- readMVar processAddr
214213 readMVar processDead
215214 runProcess localNode $ do
@@ -225,128 +224,83 @@ testMonitorDisconnect transport mOrL un = do
225224 done <- newEmptyMVar
226225
227226 forkIO $ do
228− localNode <- newLocalNode transport Map.empty
227+ localNode <- newLocalNode transport initRemoteCallMetaData
229228 addr <- forkProcess localNode . liftIO $ threadDelay 1000000
230229 putMVar processAddr addr
231230 readMVar monitorSetup
232231 NT.closeEndPoint (localEndPoint localNode)
233232
234233 forkIO $ do
235− localNode <- newLocalNode transport Map.empty
234+ localNode <- newLocalNode transport initRemoteCallMetaData
236235 theirAddr <- readMVar processAddr
237236 runProcess localNode $ do
238237 monitorTestProcess theirAddr mOrL un DiedDisconnect (Just monitorSetup) done
239238
240239 takeMVar done
241240
241+data Add = Add ProcessId Double Double deriving (Typeable)
242+data Divide = Divide ProcessId Double Double deriving (Typeable)
243+data DivByZero = DivByZero deriving (Typeable)
244+
245+instance Binary Add where
246+ put (Add pid x y) = put pid >> put x >> put y
247+ get = Add <$> get <*> get <*> get
248+
249+instance Binary Divide where
250+ put (Divide pid x y) = put pid >> put x >> put y
251+ get = Divide <$> get <*> get <*> get
252+
253+instance Binary DivByZero where
254+ put DivByZero = return ()
255+ get = return DivByZero
256+
257+math :: Process ()
258+math = do
259+ receiveWait
260+ [ match (\(Add pid x y) -> send pid (x + y))
261+ , matchIf (\(Divide _ _ y) -> y /= 0)
262+ (\(Divide pid x y) -> send pid (x / y))
263+ , match (\(Divide pid _ _) -> send pid DivByZero)
264+ ]
265+ math
242266
243−{-
244−-- Like 'testMonitor1', but throw an exception instead
245−testMonitor2 :: NT.Transport -> IO ()
246−testMonitor2 transport = do
247− deadProcess <- newEmptyMVar
248− done <- newEmptyMVar
249−
250− forkIO $ do
251− localNode <- newLocalNode transport Map.empty
252− addr <- forkProcess localNode $ return ()
253− closeLocalNode localNode
254− putMVar deadProcess addr
255−
256− forkIO $ do
257− localNode <- newLocalNode transport Map.empty
258− theirAddr <- readMVar deadProcess
259− runProcess localNode $ do
260− monitor theirAddr MaLink
261− pcatch (send theirAddr "Hi") $ \(ProcessMonitorException pid SrNoPing) -> do
262− True <- return $ pid == theirAddr
263− return ()
264− putMVar done ()
265−
266− takeMVar done
267−
268−-- The first send succeeds, connection is set up, but then the second send
269−-- fails.
270−--
271−-- TODO: should we specify that we receive exactly one notification per process
272−-- failure?
273−testMonitor3 :: NT.Transport -> IO ()
274−testMonitor3 transport = do
275− firstSend <- newEmptyMVar
276− serverAddr <- newEmptyMVar
277− serverDead <- newEmptyMVar
278− done <- newEmptyMVar
267+-- | Test the math server (i.e., receiveWait)
268+testMath :: NT.Transport -> IO ()
269+testMath transport = do
270+ serverAddr <- newEmptyMVar
271+ clientDone <- newEmptyMVar
279272
273+ -- Server
280274 forkIO $ do
281− localNode <- newLocalNode transport Map.empty
282− -- TODO: what happens when processes terminate?
283− addr <- forkProcess localNode $ return ()
275+ localNode <- newLocalNode transport initRemoteCallMetaData
276+ addr <- forkProcess localNode math
284277 putMVar serverAddr addr
285− readMVar firstSend
286− closeLocalNode localNode
287− threadDelay 10000 -- Give the TCP layer a chance to actually close the socket
288− putMVar serverDead ()
289278
279+ -- Client
290280 forkIO $ do
291− localNode <- newLocalNode transport Map.empty
292− theirAddr <- readMVar serverAddr
293− runProcess localNode $ do
294− monitor theirAddr MaMonitor
295− send theirAddr "Hi"
296− liftIO $ putMVar firstSend () >> readMVar serverDead
297− send theirAddr "Ho"
298− ProcessMonitorException pid SrNoPing <- expect
299− True <- return $ pid == theirAddr
300− -- We should not receive a second exception
301− send theirAddr "Hey"
302− Nothing <- expectTimeout 1000000 :: Process (Maybe ProcessMonitorException)
303− return ()
304− putMVar done ()
305−
306− takeMVar done
307−
308−
309−-- Like testMonitor3, except without the second send (so we must detect the
310−-- failure elsewhere)
311−testMonitor4 :: NT.Transport -> IO ()
312−testMonitor4 transport = do
313− firstSend <- newEmptyMVar
314− serverAddr <- newEmptyMVar
315− serverDead <- newEmptyMVar
316− done <- newEmptyMVar
281+ localNode <- newLocalNode transport initRemoteCallMetaData
282+ mathServer <- readMVar serverAddr
317283
318− forkIO $ do
319− localNode <- newLocalNode transport Map.empty
320− -- TODO: what happens when processes terminate?
321− addr <- forkProcess localNode $ return ()
322− putMVar serverAddr addr
323− readMVar firstSend
324− closeLocalNode localNode
325− threadDelay 10000 -- Give the TCP layer a chance to actually close the socket
326− putMVar serverDead ()
284+ runProcess localNode $ do
285+ pid <- getSelfPid
286+ send mathServer (Add pid 1 2)
287+ 3 <- expect :: Process Double
288+ send mathServer (Divide pid 8 2)
289+ 4 <- expect :: Process Double
290+ send mathServer (Divide pid 8 0)
291+ DivByZero <- expect
292+ liftIO $ putMVar clientDone ()
327293
328− forkIO $ do
329− localNode <- newLocalNode transport Map.empty
330− theirAddr <- readMVar serverAddr
331− runProcess localNode $ do
332− monitor theirAddr MaMonitor
333− send theirAddr "Hi"
334− liftIO $ putMVar firstSend () >> readMVar serverDead
335− ProcessMonitorException pid SrNoPing <- expect
336− True <- return $ pid == theirAddr
337− return ()
338− putMVar done ()
339−
340− takeMVar done
294+ takeMVar clientDone
341295
342−-- TODO: test/specify normal process termination
343−-}
296+-- TODO: test timeout
344297
345298main :: IO ()
346299main = do
347300 Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters
348301 runTests
349− [ ("Ping", testPing transport)
302+ [ ("Ping", testPing transport)
303+ , ("Math", testMath transport)
350304 -- The "missing" combinations in the list below don't make much sense, as
351305 -- we cannot guarantee that the monitor reply or link exception will not
352306 -- happen before the unmonitor or unlink
@@ -368,9 +322,4 @@ main = do
368322 , ("UnlinkNormalTermination", testMonitorNormalTermination transport False True)
369323 , ("UnlinkAbnormalTermination", testMonitorAbnormalTermination transport False True)
370324 , ("UnlinkDisconnect", testMonitorDisconnect transport False True)
371− {-
372− , ("Monitor2", testMonitor2 transport)
373− , ("Monitor3", testMonitor3 transport)
374− , ("Monitor4", testMonitor4 transport)
375− -}
376325 ]