git.iko.soy/distributed-process
on master help sign in
Make 'reconnect' synchronous. Add implicit recon.
commit301d2cd3e6fe6a4239a6bf5f6b7a869835dd6627 authorEdsko de Vries <edsko@edsko.net> dateThu 23 Aug 2012, 12:53 (14 years ago) parent38bee39
6 files changed+146−37
compare with master
unifiedsplit
distributed-process/distributed-process.cabal+2−1 distributed-process/src/Control/Distributed/Process/Internal/Messaging.hs+37−12 distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+17−5 distributed-process/src/Control/Distributed/Process/Internal/Types.hs−3 distributed-process/src/Control/Distributed/Process/Node.hs+11−13 distributed-process/tests/TestCH.hs+79−3
distributed-process/distributed-process.cabal+2−1
@@ -86,7 +86,8 @@ Test-Suite TestCH
8686 distributed-process >= 0.3 && < 0.4,
8787 network-transport >= 0.2 && < 0.3,
8888 network-transport-tcp >= 0.2 && < 0.3,
89− binary >= 0.5 && < 0.6
89+ binary >= 0.5 && < 0.6,
90+ network >= 2.3 && < 2.4
9091 Other-modules: TestAuxiliary
9192 Extensions: CPP,
9293 ScopedTypeVariables,
distributed-process/src/Control/Distributed/Process/Internal/Messaging.hs+37−12
@@ -1,9 +1,10 @@
11module Control.Distributed.Process.Internal.Messaging
22 ( -- * Message sending
3− sendPayload
3+ ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)
4+ , sendPayload
45 , sendBinary
56 , sendMessage
6− , reconnect
7+ , disconnect
78 ) where
89
910import Data.Accessor ((^.), (^=))
@@ -12,7 +13,7 @@ import qualified Data.ByteString.Lazy as BSL (toChunks)
1213import qualified Data.ByteString as BSS (ByteString)
1314import Control.Distributed.Process.Internal.StrictMVar (withMVar, modifyMVar_)
1415import Control.Concurrent.Chan (writeChan)
15−import Control.Monad (unless)
16+import Control.Monad (unless, when)
1617import qualified Network.Transport as NT
1718 ( Connection
1819 , send
@@ -39,8 +40,16 @@ import Control.Distributed.Process.Serializable (Serializable)
3940-- Message sending --
4041--------------------------------------------------------------------------------
4142
42−sendPayload :: LocalNode -> Identifier -> Identifier -> [BSS.ByteString] -> IO ()
43−sendPayload node from to payload = do
43+data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect
44+ deriving (Eq, Show)
45+
46+sendPayload :: LocalNode
47+ -> Identifier
48+ -> Identifier
49+ -> ImplicitReconnect
50+ -> [BSS.ByteString]
51+ -> IO ()
52+sendPayload node from to implicitReconnect payload = do
4453 mConn <- connBetween node from to
4554 didSend <- case mConn of
4655 Just conn -> do
@@ -49,17 +58,33 @@ sendPayload node from to payload = do
4958 Left _err -> return False
5059 Right () -> return True
5160 Nothing -> return False
52− unless didSend $
61+ unless didSend $ do
5362 writeChan (localCtrlChan node) NCMsg
5463 { ctrlMsgSender = to
5564 , ctrlMsgSignal = Died to DiedDisconnect
5665 }
66+ when (implicitReconnect == WithImplicitReconnect) $
67+ disconnect node from to
5768
58−sendBinary :: Binary a => LocalNode -> Identifier -> Identifier -> a -> IO ()
59−sendBinary node from to = sendPayload node from to . BSL.toChunks . encode
69+sendBinary :: Binary a
70+ => LocalNode
71+ -> Identifier
72+ -> Identifier
73+ -> ImplicitReconnect
74+ -> a
75+ -> IO ()
76+sendBinary node from to implicitReconnect
77+ = sendPayload node from to implicitReconnect . BSL.toChunks . encode
6078
61−sendMessage :: Serializable a => LocalNode -> Identifier -> Identifier -> a -> IO ()
62−sendMessage node from to = sendPayload node from to . messageToPayload . createMessage
79+sendMessage :: Serializable a
80+ => LocalNode
81+ -> Identifier
82+ -> Identifier
83+ -> ImplicitReconnect
84+ -> a
85+ -> IO ()
86+sendMessage node from to implicitReconnect =
87+ sendPayload node from to implicitReconnect . messageToPayload . createMessage
6388
6489setupConnBetween :: LocalNode -> Identifier -> Identifier -> IO (Maybe NT.Connection)
6590setupConnBetween node from to = do
@@ -92,8 +117,8 @@ connBetween node from to = do
92117 where
93118 nodeState = localState node
94119
95−reconnect :: LocalNode -> Identifier -> Identifier -> IO ()
96−reconnect node from to =
120+disconnect :: LocalNode -> Identifier -> Identifier -> IO ()
121+disconnect node from to =
97122 modifyMVar_ (localState node) $ \st ->
98123 case st ^. localConnectionBetween from to of
99124 Nothing ->
distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+17−5
@@ -123,7 +123,12 @@ import Control.Distributed.Process.Internal.Types
123123 , createMessage
124124 , runLocalProcess
125125 )
126−import Control.Distributed.Process.Internal.Messaging (sendMessage, sendBinary)
126+import Control.Distributed.Process.Internal.Messaging
127+ ( sendMessage
128+ , sendBinary
129+ , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)
130+ , disconnect
131+ )
127132
128133--------------------------------------------------------------------------------
129134-- Basic messaging --
@@ -138,6 +143,7 @@ send them msg = do
138143 liftIO $ sendMessage (processNode proc)
139144 (ProcessIdentifier (processId proc))
140145 (ProcessIdentifier them)
146+ NoImplicitReconnect
141147 msg
142148
143149-- | Wait for a message of a specific type
@@ -174,6 +180,7 @@ sendChan (SendPort cid) msg = do
174180 liftIO $ sendBinary (processNode proc)
175181 (ProcessIdentifier (processId proc))
176182 (SendPortIdentifier cid)
183+ NoImplicitReconnect
177184 msg
178185
179186-- | Wait for a message on a typed channel
@@ -534,13 +541,17 @@ unClosure closure = do
534541-- communication attempts are made to B then A can use reconnect to clean up
535542-- its connection to B.
536543reconnect :: ProcessId -> Process ()
537−reconnect =
538− sendCtrlMsg Nothing . Reconnect . ProcessIdentifier
544+reconnect them = do
545+ us <- getSelfPid
546+ node <- processNode <$> ask
547+ liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
539548
540549-- | Reconnect to a sendport. See 'reconnect' for more information.
541550reconnectPort :: SendPort a -> Process ()
542−reconnectPort =
543− sendCtrlMsg Nothing . Reconnect . SendPortIdentifier . sendPortId
551+reconnectPort them = do
552+ us <- getSelfPid
553+ node <- processNode <$> ask
554+ liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
544555
545556--------------------------------------------------------------------------------
546557-- Auxiliary functions --
@@ -592,4 +603,5 @@ sendCtrlMsg mNid signal = do
592603 liftIO $ sendBinary (processNode proc)
593604 (ProcessIdentifier (processId proc))
594605 (NodeIdentifier nid)
606+ WithImplicitReconnect
595607 msg
distributed-process/src/Control/Distributed/Process/Internal/Types.hs−3
@@ -384,7 +384,6 @@ data ProcessSignal =
384384 | WhereIs String
385385 | Register String (Maybe ProcessId) -- Nothing to unregister
386386 | NamedSend String Message
387− | Reconnect Identifier
388387 deriving Show
389388
390389--------------------------------------------------------------------------------
@@ -429,7 +428,6 @@ instance Binary ProcessSignal where
429428 put (WhereIs label) = putWord8 6 >> put label
430429 put (Register label pid) = putWord8 7 >> put label >> put pid
431430 put (NamedSend label msg) = putWord8 8 >> put label >> put (messageToPayload msg)
432− put (Reconnect dest) = putWord8 9 >> put dest
433431 get = do
434432 header <- getWord8
435433 case header of
@@ -442,7 +440,6 @@ instance Binary ProcessSignal where
442440 6 -> WhereIs <$> get
443441 7 -> Register <$> get <*> get
444442 8 -> NamedSend <$> get <*> (payloadToMessage <$> get)
445− 9 -> Reconnect <$> get
446443 _ -> fail "ProcessSignal.get: invalid"
447444
448445instance Binary DiedReason where
distributed-process/src/Control/Distributed/Process/Node.hs+11−13
@@ -122,7 +122,7 @@ import Control.Distributed.Process.Internal.Messaging
122122 ( sendBinary
123123 , sendMessage
124124 , sendPayload
125− , reconnect
125+ , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)
126126 )
127127import Control.Distributed.Process.Internal.Primitives (expect, register, finally)
128128import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)
@@ -422,6 +422,7 @@ nodeController = do
422422 liftIO $ sendBinary node
423423 (ctrlMsgSender msg)
424424 (NodeIdentifier nid')
425+ WithImplicitReconnect
425426 msg
426427 _ ->
427428 return ()
@@ -445,8 +446,6 @@ nodeController = do
445446 ncEffectWhereIs from label
446447 NCMsg from (NamedSend label msg') ->
447448 ncEffectNamedSend from label msg'
448− NCMsg from (Reconnect to) ->
449− ncEffectReconnect from to
450449 unexpected ->
451450 error $ "nodeController: unexpected message " ++ show unexpected
452451
@@ -475,6 +474,7 @@ ncEffectMonitor from them mRef = do
475474 liftIO $ sendBinary node
476475 (NodeIdentifier $ localNodeId node)
477476 (NodeIdentifier $ processNodeId from)
477+ WithImplicitReconnect
478478 NCMsg
479479 { ctrlMsgSender = NodeIdentifier (localNodeId node)
480480 , ctrlMsgSignal = Died them DiedUnknownId
@@ -537,6 +537,7 @@ ncEffectSpawn pid cProc ref = do
537537 liftIO $ sendMessage node
538538 (NodeIdentifier (localNodeId node))
539539 (ProcessIdentifier pid)
540+ WithImplicitReconnect
540541 (DidSpawn ref pid')
541542
542543-- Unified semantics does not explicitly describe how to implement 'register',
@@ -556,6 +557,7 @@ ncEffectWhereIs from label = do
556557 liftIO $ sendMessage node
557558 (NodeIdentifier (localNodeId node))
558559 (ProcessIdentifier from)
560+ WithImplicitReconnect
559561 (WhereIsReply label mPid)
560562
561563-- [Unified: Table 14]
@@ -568,14 +570,9 @@ ncEffectNamedSend from label msg = do
568570 liftIO $ sendPayload node
569571 from
570572 (ProcessIdentifier pid)
573+ NoImplicitReconnect
571574 (messageToPayload msg)
572575
573−-- Reconnecting
574−ncEffectReconnect :: Identifier -> Identifier -> NC ()
575−ncEffectReconnect from to = do
576− node <- ask
577− liftIO $ reconnect node from to
578−
579576--------------------------------------------------------------------------------
580577-- Auxiliary --
581578--------------------------------------------------------------------------------
@@ -601,11 +598,11 @@ notifyDied dest src reason mRef = do
601598 (True, Nothing, SendPortIdentifier pid) ->
602599 throwException dest $ PortLinkException pid reason
603600 (False, _, _) ->
604− -- TODO: why the change in sender? How does that affect 'reconnect' semantics?
605− -- (see [Unified: Table 10]
601+ -- The change in sender comes from [Unified: Table 10]
606602 liftIO $ sendBinary node
607603 (NodeIdentifier $ localNodeId node)
608604 (NodeIdentifier $ processNodeId dest)
605+ WithImplicitReconnect
609606 NCMsg
610607 { ctrlMsgSender = NodeIdentifier (localNodeId node)
611608 , ctrlMsgSignal = Died src reason
@@ -621,7 +618,6 @@ destNid (Spawn _ _) = Nothing
621618destNid (Register _ _) = Nothing
622619destNid (WhereIs _) = Nothing
623620destNid (NamedSend _ _) = Nothing
624−destNid (Reconnect _) = Nothing
625621-- We don't need to forward 'Died' signals; if monitoring/linking is setup,
626622-- then when a local process dies the monitoring/linking machinery will take
627623-- care of notifying remote nodes
@@ -721,10 +717,12 @@ registryFor ident = registry >>> DAC.mapMaybe ident
721717--
722718-- * the notifications for that process specifically and
723719-- * the notifications for typed channels to that process.
720+--
721+-- See https://github.com/haskell/containers/issues/14 for the bang on _v.
724722splitNotif :: Identifier
725723 -> Map Identifier a
726724 -> (Map Identifier a, Map Identifier a)
727−splitNotif ident = Map.partitionWithKey (\k !v -> impliesDeathOf ident k)
725+splitNotif ident = Map.partitionWithKey (\k !_v -> impliesDeathOf ident k)
728726
729727-- | Does the death of one entity (node, project, channel) imply the death
730728-- of another?
distributed-process/tests/TestCH.hs+79−3
@@ -19,9 +19,17 @@ import Control.Monad (replicateM_, replicateM)
1919import Control.Exception (throwIO)
2020import Control.Applicative ((<$>), (<*>))
2121import qualified Network.Transport as NT (Transport, closeEndPoint)
22−import Network.Transport.TCP (createTransport, defaultTCPParameters)
22+import Network.Socket (sClose)
23+import Network.Transport.TCP
24+ ( createTransportExposeInternals
25+ , TransportInternals(socketBetween)
26+ , defaultTCPParameters
27+ )
2328import Control.Distributed.Process
24−import Control.Distributed.Process.Internal.Types (LocalNode(localEndPoint))
29+import Control.Distributed.Process.Internal.Types
30+ ( NodeId(nodeAddress)
31+ , LocalNode(localEndPoint)
32+ )
2533import Control.Distributed.Process.Node
2634import Control.Distributed.Process.Serializable (Serializable)
2735import TestAuxiliary
@@ -578,9 +586,75 @@ testSpawnLocal transport = do
578586 send pid sport
579587 expect
580588
589+testReconnect :: NT.Transport -> TransportInternals -> IO ()
590+testReconnect transport transportInternals = do
591+ [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable
592+ let nid1 = localNodeId node1
593+ nid2 = localNodeId node2
594+ processA <- newEmptyMVar
595+ [sendTestOk, registerTestOk] <- replicateM 2 newEmptyMVar
596+
597+ forkProcess node1 $ do
598+ us <- getSelfPid
599+ liftIO $ putMVar processA us
600+ msg1 <- expect
601+ msg2 <- expect
602+ True <- return $ msg1 == "message 1" && msg2 == "message 3"
603+ liftIO $ putMVar sendTestOk ()
604+
605+ forkProcess node2 $ do
606+ {-
607+ - Make sure there is no implicit reconnect on normal message sending
608+ -}
609+
610+ them <- liftIO $ readMVar processA
611+ send them "message 1" >> liftIO (threadDelay 100000)
612+
613+ -- Simulate network failure
614+ liftIO $ do
615+ sock <- socketBetween transportInternals (nodeAddress nid1) (nodeAddress nid2)
616+ sClose sock
617+
618+ -- Should not arrive
619+ send them "message 2"
620+
621+ -- Should arrive
622+ reconnect them
623+ send them "message 3"
624+
625+ liftIO $ takeMVar sendTestOk
626+
627+ {-
628+ - Test that there *is* implicit reconnect on node controller messages
629+ -}
630+
631+ us <- getSelfPid
632+ registerRemote nid1 "a" us >> liftIO (threadDelay 100000) -- registerRemote is asynchronous
633+
634+ -- Simulate network failure
635+ liftIO $ do
636+ sock <- socketBetween transportInternals (nodeAddress nid1) (nodeAddress nid2)
637+ sClose sock
638+ threadDelay 10000
639+
640+ -- Should not happen
641+ registerRemote nid1 "b" us
642+
643+ -- Should happen
644+ registerRemote nid1 "c" us
645+
646+ -- Check
647+ Just _ <- whereisRemote nid1 "a"
648+ Nothing <- whereisRemote nid1 "b"
649+ Just _ <- whereisRemote nid1 "c"
650+
651+ liftIO $ putMVar registerTestOk ()
652+
653+ takeMVar registerTestOk
654+
581655main :: IO ()
582656main = do
583− Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters
657+ Right (transport, transportInternals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters
584658 runTests
585659 [ ("Ping", testPing transport)
586660 , ("Math", testMath transport)
@@ -619,4 +693,6 @@ main = do
619693 -- Monitoring nodes and channels
620694 , ("MonitorNode", testMonitorNode transport)
621695 , ("MonitorChannel", testMonitorChannel transport)
696+ -- Reconnect
697+ , ("Reconnect", testReconnect transport transportInternals)
622698 ]