git.iko.soy/distributed-process
on master help sign in
Subtle modification to implicit reconnect
b1d2a76Edsko de Vries
distributed-process/distributed-process.cabal+2−1
compare with master
unifiedsplit
@@ -106,7 +106,8 @@ Test-Suite TestClosure
106 distributed-process >= 0.3 && < 0.4,
106 distributed-process >= 0.3 && < 0.4,
107 network-transport >= 0.2 && < 0.3,
107 network-transport >= 0.2 && < 0.3,
108 network-transport-tcp >= 0.2 && < 0.3,
108 network-transport-tcp >= 0.2 && < 0.3,
109− bytestring >= 0.9 && < 0.11
109+ bytestring >= 0.9 && < 0.11,
110+ network >= 2.3 && < 2.4
110 Other-modules: TestAuxiliary
111 Other-modules: TestAuxiliary
111 Extensions: CPP,
112 Extensions: CPP,
112 ScopedTypeVariables
113 ScopedTypeVariables
distributed-process/src/Control/Distributed/Process.hs+2−5
@@ -138,6 +138,7 @@ import Control.Distributed.Process.Internal.Types
138 , SendPortId(..)
138 , SendPortId(..)
139 , WhereIsReply(..)
139 , WhereIsReply(..)
140 , LocalProcess(processNode)
140 , LocalProcess(processNode)
141+ , nullProcessId
141 )
142 )
142import Control.Distributed.Process.Serializable (Serializable, SerializableDict)
143import Control.Distributed.Process.Serializable (Serializable, SerializableDict)
143import Control.Distributed.Process.Internal.Closure.BuiltIn
144import Control.Distributed.Process.Internal.Closure.BuiltIn
@@ -289,10 +290,6 @@ import Control.Distributed.Process.Node (forkProcess)
289spawn :: NodeId -> Closure (Process ()) -> Process ProcessId
290spawn :: NodeId -> Closure (Process ()) -> Process ProcessId
290spawn nid proc = do
291spawn nid proc = do
291 us <- getSelfPid
292 us <- getSelfPid
292− -- Since we throw an exception when the remote node dies, we could use
293− -- linkNode instead. However, we don't have a way of "scoped" linking so if
294− -- we call linkNode here, and unlinkNode after, then we might remove a link
295− -- that was already set up
296 mRef <- monitorNode nid
293 mRef <- monitorNode nid
297 sRef <- spawnAsync nid $ cpLink us
294 sRef <- spawnAsync nid $ cpLink us
298 `seqCP` cpExpect sdictUnit
295 `seqCP` cpExpect sdictUnit
@@ -306,7 +303,7 @@ spawn nid proc = do
306 ]
303 ]
307 unmonitor mRef
304 unmonitor mRef
308 case mPid of
305 case mPid of
309− Left err -> fail $ "spawn: remote node failed: " ++ show err
306+ Left _err -> return (nullProcessId nid)
310 Right pid -> send pid () >> return pid
307 Right pid -> send pid () >> return pid
311
308
312-- | Spawn a process and link to it
309-- | Spawn a process and link to it
distributed-process/src/Control/Distributed/Process/Internal/Messaging.hs+4−1
@@ -63,8 +63,11 @@ sendPayload node from to implicitReconnect payload = do
63 { ctrlMsgSender = to
63 { ctrlMsgSender = to
64 , ctrlMsgSignal = Died to DiedDisconnect
64 , ctrlMsgSignal = Died to DiedDisconnect
65 }
65 }
66− when (implicitReconnect == WithImplicitReconnect) $
66+ when (implicitReconnect == WithImplicitReconnect) $ do
67 disconnect node from to
67 disconnect node from to
68+ -- Try once more
69+ -- (TODO: not quite sure this is the Right Thing)
70+ sendPayload node from to NoImplicitReconnect payload
68
71
69sendBinary :: Binary a
72sendBinary :: Binary a
70 => LocalNode
73 => LocalNode
distributed-process/src/Control/Distributed/Process/Node.hs+1
@@ -257,6 +257,7 @@ forkProcess node proc = modifyMVar (localState node) startProcess
257 . (localConnections ^= unaffected)
257 . (localConnections ^= unaffected)
258 $ st
258 $ st
259
259
260+-- TODO: invariant: if P reconnects to Q, Q must consider its connection to be broken at that point
260handleIncomingMessages :: LocalNode -> IO ()
261handleIncomingMessages :: LocalNode -> IO ()
261handleIncomingMessages node = go Set.empty Map.empty Map.empty Set.empty
262handleIncomingMessages node = go Set.empty Map.empty Map.empty Set.empty
262 where
263 where
distributed-process/tests/TestCH.hs+2−2
@@ -644,7 +644,7 @@ testReconnect transport transportInternals = do
644 sClose sock
644 sClose sock
645 threadDelay 10000
645 threadDelay 10000
646
646
647− -- Should not happen
647+ -- This will happen due to implicit reconnect
648 registerRemote nid1 "b" us
648 registerRemote nid1 "b" us
649
649
650 -- Should happen
650 -- Should happen
@@ -652,7 +652,7 @@ testReconnect transport transportInternals = do
652
652
653 -- Check
653 -- Check
654 Just _ <- whereisRemote nid1 "a"
654 Just _ <- whereisRemote nid1 "a"
655− Nothing <- whereisRemote nid1 "b"
655+ Just _ <- whereisRemote nid1 "b"
656 Just _ <- whereisRemote nid1 "c"
656 Just _ <- whereisRemote nid1 "c"
657
657
658 liftIO $ putMVar registerTestOk ()
658 liftIO $ putMVar registerTestOk ()
distributed-process/tests/TestClosure.hs+59−3
@@ -6,13 +6,27 @@ import Data.Typeable (Typeable)
6import Control.Monad (join, replicateM, forever)
6import Control.Monad (join, replicateM, forever)
7import Control.Exception (IOException, throw)
7import Control.Exception (IOException, throw)
8import Control.Concurrent (forkIO, threadDelay)
8import Control.Concurrent (forkIO, threadDelay)
9−import Control.Concurrent.MVar (MVar, newEmptyMVar, readMVar, takeMVar, putMVar)
9+import Control.Concurrent.MVar
10+ ( MVar
11+ , newEmptyMVar
12+ , readMVar
13+ , takeMVar
14+ , putMVar
15+ , modifyMVar_
16+ , newMVar
17+ )
10import Control.Applicative ((<$>))
18import Control.Applicative ((<$>))
11import Network.Transport (Transport)
19import Network.Transport (Transport)
12−import Network.Transport.TCP (createTransport, defaultTCPParameters)
20+import Network.Transport.TCP
21+ ( createTransportExposeInternals
22+ , defaultTCPParameters
23+ , TransportInternals(socketBetween)
24+ )
25+import Network.Socket (sClose)
13import Control.Distributed.Process
26import Control.Distributed.Process
14import Control.Distributed.Process.Closure
27import Control.Distributed.Process.Closure
15import Control.Distributed.Process.Node
28import Control.Distributed.Process.Node
29+import Control.Distributed.Process.Internal.Types (NodeId(nodeAddress))
16import Control.Distributed.Static (staticLabel, staticClosure)
30import Control.Distributed.Static (staticLabel, staticClosure)
17import TestAuxiliary
31import TestAuxiliary
18
32
@@ -53,6 +67,9 @@ typedPingServer () rport = forever $ do
53 sport <- receiveChan rport
67 sport <- receiveChan rport
54 sendChan sport ()
68 sendChan sport ()
55
69
70+signal :: ProcessId -> Process ()
71+signal pid = send pid ()
72+
56remotable [ 'factorial
73remotable [ 'factorial
57 , 'addInt
74 , 'addInt
58 , 'putInt
75 , 'putInt
@@ -62,6 +79,7 @@ remotable [ 'factorial
62 , 'typedPingServer
79 , 'typedPingServer
63 , 'isPrime
80 , 'isPrime
64 , 'quintuple
81 , 'quintuple
82+ , 'signal
65 ]
83 ]
66
84
67-- Just try creating a static polymorphic value
85-- Just try creating a static polymorphic value
@@ -337,9 +355,46 @@ testTDict transport rtable = do
337 liftIO $ putMVar done ()
355 liftIO $ putMVar done ()
338 takeMVar done
356 takeMVar done
339
357
358+
359+testSpawnReconnect :: Transport -> RemoteTable -> TransportInternals -> IO ()
360+testSpawnReconnect transport rtable transportInternals = do
361+ [node1, node2] <- replicateM 2 $ newLocalNode transport rtable
362+ let nid1 = localNodeId node1
363+ nid2 = localNodeId node2
364+ done <- newEmptyMVar
365+ iv <- newMVar (0 :: Int)
366+
367+ incr <- forkProcess node1 $ forever $ do
368+ () <- expect
369+ liftIO $ modifyMVar_ iv (return . (+ 1))
370+
371+ forkProcess node2 $ do
372+ _pid1 <- spawn nid1 ($(mkClosure 'signal) incr)
373+
374+ -- Simulate network failure
375+ liftIO $ do
376+ sock <- socketBetween transportInternals (nodeAddress nid2) (nodeAddress nid1)
377+ sClose sock
378+
379+ -- Implicit reconnect: spawn happens; but since we also receive the
380+ -- monitor signal, the process might not actually be started
381+ _pid2 <- spawn nid1 ($(mkClosure 'signal) incr)
382+
383+ -- As does this one
384+ _pid3 <- spawn nid1 ($(mkClosure 'signal) incr)
385+
386+ liftIO $ threadDelay 10000
387+
388+ count <- liftIO $ takeMVar iv
389+ True <- return $ count == 2 || count == 3 -- Depending on whether we get the monitor notification first or not
390+
391+ liftIO $ putMVar done ()
392+
393+ takeMVar done
394+
340main :: IO ()
395main :: IO ()
341main = do
396main = do
342− Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters
397+ Right (transport, transportInternals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters
343 let rtable = __remoteTable initRemoteTable
398 let rtable = __remoteTable initRemoteTable
344 runTests
399 runTests
345 [ ("Unclosure", testUnclosure transport rtable)
400 [ ("Unclosure", testUnclosure transport rtable)
@@ -356,4 +411,5 @@ main = do
356 , ("ClosureExpect", testClosureExpect transport rtable)
411 , ("ClosureExpect", testClosureExpect transport rtable)
357 , ("SpawnChannel", testSpawnChannel transport rtable)
412 , ("SpawnChannel", testSpawnChannel transport rtable)
358 , ("TDict", testTDict transport rtable)
413 , ("TDict", testTDict transport rtable)
414+ , ("SpawnReconnect", testSpawnReconnect transport rtable transportInternals)
359 ]
415 ]