git.iko.soy/distributed-process
on master help sign in
Subtle modification to implicit reconnect
commitb1d2a76adfdfd8b6771aba1e67e1fc36c984ad37 authorEdsko de Vries <edsko@edsko.net> dateFri 24 Aug 2012, 12:10 (14 years ago) parent63bfd19
6 files changed+70−12
compare with master
unifiedsplit
distributed-process/distributed-process.cabal+2−1 distributed-process/src/Control/Distributed/Process.hs+2−5 distributed-process/src/Control/Distributed/Process/Internal/Messaging.hs+4−1 distributed-process/src/Control/Distributed/Process/Node.hs+1 distributed-process/tests/TestCH.hs+2−2 distributed-process/tests/TestClosure.hs+59−3
distributed-process/distributed-process.cabal+2−1
@@ -106,7 +106,8 @@ Test-Suite TestClosure
106106 distributed-process >= 0.3 && < 0.4,
107107 network-transport >= 0.2 && < 0.3,
108108 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
110111 Other-modules: TestAuxiliary
111112 Extensions: CPP,
112113 ScopedTypeVariables
distributed-process/src/Control/Distributed/Process.hs+2−5
@@ -138,6 +138,7 @@ import Control.Distributed.Process.Internal.Types
138138 , SendPortId(..)
139139 , WhereIsReply(..)
140140 , LocalProcess(processNode)
141+ , nullProcessId
141142 )
142143import Control.Distributed.Process.Serializable (Serializable, SerializableDict)
143144import Control.Distributed.Process.Internal.Closure.BuiltIn
@@ -289,10 +290,6 @@ import Control.Distributed.Process.Node (forkProcess)
289290spawn :: NodeId -> Closure (Process ()) -> Process ProcessId
290291spawn nid proc = do
291292 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
296293 mRef <- monitorNode nid
297294 sRef <- spawnAsync nid $ cpLink us
298295 `seqCP` cpExpect sdictUnit
@@ -306,7 +303,7 @@ spawn nid proc = do
306303 ]
307304 unmonitor mRef
308305 case mPid of
309− Left err -> fail $ "spawn: remote node failed: " ++ show err
306+ Left _err -> return (nullProcessId nid)
310307 Right pid -> send pid () >> return pid
311308
312309-- | 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
6363 { ctrlMsgSender = to
6464 , ctrlMsgSignal = Died to DiedDisconnect
6565 }
66− when (implicitReconnect == WithImplicitReconnect) $
66+ when (implicitReconnect == WithImplicitReconnect) $ do
6767 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
6871
6972sendBinary :: Binary a
7073 => LocalNode
distributed-process/src/Control/Distributed/Process/Node.hs+1
@@ -257,6 +257,7 @@ forkProcess node proc = modifyMVar (localState node) startProcess
257257 . (localConnections ^= unaffected)
258258 $ st
259259
260+-- TODO: invariant: if P reconnects to Q, Q must consider its connection to be broken at that point
260261handleIncomingMessages :: LocalNode -> IO ()
261262handleIncomingMessages node = go Set.empty Map.empty Map.empty Set.empty
262263 where
distributed-process/tests/TestCH.hs+2−2
@@ -644,7 +644,7 @@ testReconnect transport transportInternals = do
644644 sClose sock
645645 threadDelay 10000
646646
647− -- Should not happen
647+ -- This will happen due to implicit reconnect
648648 registerRemote nid1 "b" us
649649
650650 -- Should happen
@@ -652,7 +652,7 @@ testReconnect transport transportInternals = do
652652
653653 -- Check
654654 Just _ <- whereisRemote nid1 "a"
655− Nothing <- whereisRemote nid1 "b"
655+ Just _ <- whereisRemote nid1 "b"
656656 Just _ <- whereisRemote nid1 "c"
657657
658658 liftIO $ putMVar registerTestOk ()
distributed-process/tests/TestClosure.hs+59−3
@@ -6,13 +6,27 @@ import Data.Typeable (Typeable)
66import Control.Monad (join, replicateM, forever)
77import Control.Exception (IOException, throw)
88import 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+ )
1018import Control.Applicative ((<$>))
1119import 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)
1326import Control.Distributed.Process
1427import Control.Distributed.Process.Closure
1528import Control.Distributed.Process.Node
29+import Control.Distributed.Process.Internal.Types (NodeId(nodeAddress))
1630import Control.Distributed.Static (staticLabel, staticClosure)
1731import TestAuxiliary
1832
@@ -53,6 +67,9 @@ typedPingServer () rport = forever $ do
5367 sport <- receiveChan rport
5468 sendChan sport ()
5569
70+signal :: ProcessId -> Process ()
71+signal pid = send pid ()
72+
5673remotable [ 'factorial
5774 , 'addInt
5875 , 'putInt
@@ -62,6 +79,7 @@ remotable [ 'factorial
6279 , 'typedPingServer
6380 , 'isPrime
6481 , 'quintuple
82+ , 'signal
6583 ]
6684
6785-- Just try creating a static polymorphic value
@@ -337,9 +355,46 @@ testTDict transport rtable = do
337355 liftIO $ putMVar done ()
338356 takeMVar done
339357
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+
340395main :: IO ()
341396main = do
342− Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters
397+ Right (transport, transportInternals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters
343398 let rtable = __remoteTable initRemoteTable
344399 runTests
345400 [ ("Unclosure", testUnclosure transport rtable)
@@ -356,4 +411,5 @@ main = do
356411 , ("ClosureExpect", testClosureExpect transport rtable)
357412 , ("SpawnChannel", testSpawnChannel transport rtable)
358413 , ("TDict", testTDict transport rtable)
414+ , ("SpawnReconnect", testSpawnReconnect transport rtable transportInternals)
359415 ]