301d2cd · distributed-process

git.iko.soy/distributed-process

on master help sign in

Make 'reconnect' synchronous. Add implicit recon.

301d2cdEdsko de Vries14 years ago

distributed-process/distributed-process.cabal+2−1

compare with master

unifiedsplit

@@ -86,7 +86,8 @@ Test-Suite TestCH

86 distributed-process >= 0.3 && < 0.4,

87 network-transport >= 0.2 && < 0.3,

88 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

90 Other-modules: TestAuxiliary

91 Other-modules: TestAuxiliary

91 Extensions: CPP,

92 Extensions: CPP,

92 ScopedTypeVariables,

93 ScopedTypeVariables,

distributed-process/src/Control/Distributed/Process/Internal/Messaging.hs+37−12

@@ -1,9 +1,10 @@

1module Control.Distributed.Process.Internal.Messaging

2 ( -- * Message sending

3+ ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

3− sendPayload

4+ , sendPayload

4 , sendBinary

5 , sendBinary

5 , sendMessage

6 , sendMessage

6− , reconnect

7+ , disconnect

7 ) where

8 ) where

9import Data.Accessor ((^.), (^=))

10import Data.Accessor ((^.), (^=))

@@ -12,7 +13,7 @@ import qualified Data.ByteString.Lazy as BSL (toChunks)

12import qualified Data.ByteString as BSS (ByteString)

13import qualified Data.ByteString as BSS (ByteString)

13import Control.Distributed.Process.Internal.StrictMVar (withMVar, modifyMVar_)

14import Control.Distributed.Process.Internal.StrictMVar (withMVar, modifyMVar_)

14import Control.Concurrent.Chan (writeChan)

15import Control.Concurrent.Chan (writeChan)

15−import Control.Monad (unless)

16+import Control.Monad (unless, when)

16import qualified Network.Transport as NT

17import qualified Network.Transport as NT

17 ( Connection

18 ( Connection

18 , send

19 , send

@@ -39,8 +40,16 @@ import Control.Distributed.Process.Serializable (Serializable)

39-- Message sending --

40-- Message sending --

40--------------------------------------------------------------------------------

41--------------------------------------------------------------------------------

43+data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect

44+ deriving (Eq, Show)

45+

42−sendPayload :: LocalNode -> Identifier -> Identifier -> [BSS.ByteString] -> IO ()

46+sendPayload :: LocalNode

47+ -> Identifier

48+ -> Identifier

49+ -> ImplicitReconnect

50+ -> [BSS.ByteString]

51+ -> IO ()

43−sendPayload node from to payload = do

52+sendPayload node from to implicitReconnect payload = do

44 mConn <- connBetween node from to

53 mConn <- connBetween node from to

45 didSend <- case mConn of

54 didSend <- case mConn of

46 Just conn -> do

55 Just conn -> do

@@ -49,17 +58,33 @@ sendPayload node from to payload = do

49 Left _err -> return False

58 Left _err -> return False

50 Right () -> return True

59 Right () -> return True

51 Nothing -> return False

60 Nothing -> return False

52− unless didSend $

61+ unless didSend $ do

53 writeChan (localCtrlChan node) NCMsg

62 writeChan (localCtrlChan node) NCMsg

54 { ctrlMsgSender = to

63 { ctrlMsgSender = to

55 , ctrlMsgSignal = Died to DiedDisconnect

64 , ctrlMsgSignal = Died to DiedDisconnect

56 }

65 }

66+ when (implicitReconnect == WithImplicitReconnect) $

67+ disconnect node from to

58−sendBinary :: Binary a => LocalNode -> Identifier -> Identifier -> a -> IO ()

69+sendBinary :: Binary a

70+ => LocalNode

71+ -> Identifier

72+ -> Identifier

73+ -> ImplicitReconnect

74+ -> a

75+ -> IO ()

76+sendBinary node from to implicitReconnect

59−sendBinary node from to = sendPayload node from to . BSL.toChunks . encode

77+ = sendPayload node from to implicitReconnect . BSL.toChunks . encode

61−sendMessage :: Serializable a => LocalNode -> Identifier -> Identifier -> a -> IO ()

79+sendMessage :: Serializable a

80+ => LocalNode

81+ -> Identifier

82+ -> Identifier

83+ -> ImplicitReconnect

84+ -> a

85+ -> IO ()

86+sendMessage node from to implicitReconnect =

62−sendMessage node from to = sendPayload node from to . messageToPayload . createMessage

87+ sendPayload node from to implicitReconnect . messageToPayload . createMessage

64setupConnBetween :: LocalNode -> Identifier -> Identifier -> IO (Maybe NT.Connection)

89setupConnBetween :: LocalNode -> Identifier -> Identifier -> IO (Maybe NT.Connection)

65setupConnBetween node from to = do

90setupConnBetween node from to = do

@@ -92,8 +117,8 @@ connBetween node from to = do

92 where

117 where

93 nodeState = localState node

118 nodeState = localState node

119

95−reconnect :: LocalNode -> Identifier -> Identifier -> IO ()

120+disconnect :: LocalNode -> Identifier -> Identifier -> IO ()

96−reconnect node from to =

121+disconnect node from to =

97 modifyMVar_ (localState node) $ \st ->

122 modifyMVar_ (localState node) $ \st ->

98 case st ^. localConnectionBetween from to of

123 case st ^. localConnectionBetween from to of

99 Nothing ->

124 Nothing ->

distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+17−5

@@ -123,7 +123,12 @@ import Control.Distributed.Process.Internal.Types

123 , createMessage

124 , runLocalProcess

125 )

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+ )

127

132

128--------------------------------------------------------------------------------

133--------------------------------------------------------------------------------

129-- Basic messaging --

134-- Basic messaging --

@@ -138,6 +143,7 @@ send them msg = do

138 liftIO $ sendMessage (processNode proc)

143 liftIO $ sendMessage (processNode proc)

139 (ProcessIdentifier (processId proc))

144 (ProcessIdentifier (processId proc))

140 (ProcessIdentifier them)

145 (ProcessIdentifier them)

146+ NoImplicitReconnect

141 msg

147 msg

142

148

143-- | Wait for a message of a specific type

149-- | Wait for a message of a specific type

@@ -174,6 +180,7 @@ sendChan (SendPort cid) msg = do

174 liftIO $ sendBinary (processNode proc)

180 liftIO $ sendBinary (processNode proc)

175 (ProcessIdentifier (processId proc))

181 (ProcessIdentifier (processId proc))

176 (SendPortIdentifier cid)

182 (SendPortIdentifier cid)

183+ NoImplicitReconnect

177 msg

184 msg

178

185

179-- | Wait for a message on a typed channel

186-- | Wait for a message on a typed channel

@@ -534,13 +541,17 @@ unClosure closure = do

534-- communication attempts are made to B then A can use reconnect to clean up

541-- communication attempts are made to B then A can use reconnect to clean up

535-- its connection to B.

542-- its connection to B.

536reconnect :: ProcessId -> Process ()

543reconnect :: ProcessId -> Process ()

537−reconnect =

544+reconnect them = do

538− sendCtrlMsg Nothing . Reconnect . ProcessIdentifier

545+ us <- getSelfPid

546+ node <- processNode <$> ask

547+ liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)

539

548

540-- | Reconnect to a sendport. See 'reconnect' for more information.

549-- | Reconnect to a sendport. See 'reconnect' for more information.

541reconnectPort :: SendPort a -> Process ()

550reconnectPort :: SendPort a -> Process ()

542−reconnectPort =

551+reconnectPort them = do

552+ us <- getSelfPid

553+ node <- processNode <$> ask

543− sendCtrlMsg Nothing . Reconnect . SendPortIdentifier . sendPortId

554+ liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))

544

555

545--------------------------------------------------------------------------------

556--------------------------------------------------------------------------------

546-- Auxiliary functions --

557-- Auxiliary functions --

@@ -592,4 +603,5 @@ sendCtrlMsg mNid signal = do

592 liftIO $ sendBinary (processNode proc)

603 liftIO $ sendBinary (processNode proc)

593 (ProcessIdentifier (processId proc))

604 (ProcessIdentifier (processId proc))

594 (NodeIdentifier nid)

605 (NodeIdentifier nid)

606+ WithImplicitReconnect

595 msg

607 msg

distributed-process/src/Control/Distributed/Process/Internal/Types.hs−3

@@ -384,7 +384,6 @@ data ProcessSignal =

384 | WhereIs String

385 | Register String (Maybe ProcessId) -- Nothing to unregister

386 | NamedSend String Message

387− | Reconnect Identifier

388 deriving Show

387 deriving Show

389

388

390--------------------------------------------------------------------------------

389--------------------------------------------------------------------------------

@@ -429,7 +428,6 @@ instance Binary ProcessSignal where

429 put (WhereIs label) = putWord8 6 >> put label

428 put (WhereIs label) = putWord8 6 >> put label

430 put (Register label pid) = putWord8 7 >> put label >> put pid

429 put (Register label pid) = putWord8 7 >> put label >> put pid

431 put (NamedSend label msg) = putWord8 8 >> put label >> put (messageToPayload msg)

430 put (NamedSend label msg) = putWord8 8 >> put label >> put (messageToPayload msg)

432− put (Reconnect dest) = putWord8 9 >> put dest

433 get = do

431 get = do

434 header <- getWord8

432 header <- getWord8

435 case header of

433 case header of

@@ -442,7 +440,6 @@ instance Binary ProcessSignal where

442 6 -> WhereIs <$> get

440 6 -> WhereIs <$> get

443 7 -> Register <$> get <*> get

441 7 -> Register <$> get <*> get

444 8 -> NamedSend <$> get <*> (payloadToMessage <$> get)

442 8 -> NamedSend <$> get <*> (payloadToMessage <$> get)

445− 9 -> Reconnect <$> get

446 _ -> fail "ProcessSignal.get: invalid"

443 _ -> fail "ProcessSignal.get: invalid"

447

444

448instance Binary DiedReason where

445instance Binary DiedReason where

distributed-process/src/Control/Distributed/Process/Node.hs+11−13

@@ -122,7 +122,7 @@ import Control.Distributed.Process.Internal.Messaging

122 ( sendBinary

123 , sendMessage

124 , sendPayload

125− , reconnect

125+ , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

126 )

127import Control.Distributed.Process.Internal.Primitives (expect, register, finally)

128import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)

@@ -422,6 +422,7 @@ nodeController = do

422 liftIO $ sendBinary node

423 (ctrlMsgSender msg)

424 (NodeIdentifier nid')

425+ WithImplicitReconnect

425 msg

426 msg

426 _ ->

427 _ ->

427 return ()

428 return ()

@@ -445,8 +446,6 @@ nodeController = do

445 ncEffectWhereIs from label

446 ncEffectWhereIs from label

446 NCMsg from (NamedSend label msg') ->

447 NCMsg from (NamedSend label msg') ->

447 ncEffectNamedSend from label msg'

448 ncEffectNamedSend from label msg'

448− NCMsg from (Reconnect to) ->

449− ncEffectReconnect from to

450 unexpected ->

449 unexpected ->

451 error $ "nodeController: unexpected message " ++ show unexpected

450 error $ "nodeController: unexpected message " ++ show unexpected

452

451

@@ -475,6 +474,7 @@ ncEffectMonitor from them mRef = do

475 liftIO $ sendBinary node

474 liftIO $ sendBinary node

476 (NodeIdentifier $ localNodeId node)

475 (NodeIdentifier $ localNodeId node)

477 (NodeIdentifier $ processNodeId from)

476 (NodeIdentifier $ processNodeId from)

477+ WithImplicitReconnect

478 NCMsg

479 { ctrlMsgSender = NodeIdentifier (localNodeId node)

480 , ctrlMsgSignal = Died them DiedUnknownId

@@ -537,6 +537,7 @@ ncEffectSpawn pid cProc ref = do

537 liftIO $ sendMessage node

538 (NodeIdentifier (localNodeId node))

539 (ProcessIdentifier pid)

540+ WithImplicitReconnect

540 (DidSpawn ref pid')

541 (DidSpawn ref pid')

541

542

542-- Unified semantics does not explicitly describe how to implement 'register',

543-- Unified semantics does not explicitly describe how to implement 'register',

@@ -556,6 +557,7 @@ ncEffectWhereIs from label = do

556 liftIO $ sendMessage node

557 liftIO $ sendMessage node

557 (NodeIdentifier (localNodeId node))

558 (NodeIdentifier (localNodeId node))

558 (ProcessIdentifier from)

559 (ProcessIdentifier from)

560+ WithImplicitReconnect

559 (WhereIsReply label mPid)

561 (WhereIsReply label mPid)

560

562

561-- [Unified: Table 14]

563-- [Unified: Table 14]

@@ -568,14 +570,9 @@ ncEffectNamedSend from label msg = do

568 liftIO $ sendPayload node

570 liftIO $ sendPayload node

569 from

571 from

570 (ProcessIdentifier pid)

572 (ProcessIdentifier pid)

573+ NoImplicitReconnect

571 (messageToPayload msg)

574 (messageToPayload msg)

572

575

573−-- Reconnecting

574−ncEffectReconnect :: Identifier -> Identifier -> NC ()

575−ncEffectReconnect from to = do

576− node <- ask

577− liftIO $ reconnect node from to

578−

579--------------------------------------------------------------------------------

576--------------------------------------------------------------------------------

580-- Auxiliary --

577-- Auxiliary --

581--------------------------------------------------------------------------------

578--------------------------------------------------------------------------------

@@ -601,11 +598,11 @@ notifyDied dest src reason mRef = do

601 (True, Nothing, SendPortIdentifier pid) ->

598 (True, Nothing, SendPortIdentifier pid) ->

602 throwException dest $ PortLinkException pid reason

599 throwException dest $ PortLinkException pid reason

603 (False, _, _) ->

600 (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]

606 liftIO $ sendBinary node

602 liftIO $ sendBinary node

607 (NodeIdentifier $ localNodeId node)

603 (NodeIdentifier $ localNodeId node)

608 (NodeIdentifier $ processNodeId dest)

604 (NodeIdentifier $ processNodeId dest)

605+ WithImplicitReconnect

609 NCMsg

606 NCMsg

610 { ctrlMsgSender = NodeIdentifier (localNodeId node)

607 { ctrlMsgSender = NodeIdentifier (localNodeId node)

611 , ctrlMsgSignal = Died src reason

608 , ctrlMsgSignal = Died src reason

@@ -621,7 +618,6 @@ destNid (Spawn _ _) = Nothing

621destNid (Register _ _) = Nothing

618destNid (Register _ _) = Nothing

622destNid (WhereIs _) = Nothing

619destNid (WhereIs _) = Nothing

623destNid (NamedSend _ _) = Nothing

620destNid (NamedSend _ _) = Nothing

624−destNid (Reconnect _) = Nothing

625-- We don't need to forward 'Died' signals; if monitoring/linking is setup,

621-- We don't need to forward 'Died' signals; if monitoring/linking is setup,

626-- then when a local process dies the monitoring/linking machinery will take

622-- then when a local process dies the monitoring/linking machinery will take

627-- care of notifying remote nodes

623-- care of notifying remote nodes

@@ -721,10 +717,12 @@ registryFor ident = registry >>> DAC.mapMaybe ident

721--

717--

722-- * the notifications for that process specifically and

718-- * the notifications for that process specifically and

723-- * the notifications for typed channels to that process.

719-- * the notifications for typed channels to that process.

720+--

721+-- See https://github.com/haskell/containers/issues/14 for the bang on _v.

724splitNotif :: Identifier

722splitNotif :: Identifier

725 -> Map Identifier a

723 -> Map Identifier a

726 -> (Map Identifier a, Map Identifier a)

724 -> (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)

728

726

729-- | Does the death of one entity (node, project, channel) imply the death

727-- | Does the death of one entity (node, project, channel) imply the death

730-- of another?

728-- of another?

distributed-process/tests/TestCH.hs+79−3

@@ -19,9 +19,17 @@ import Control.Monad (replicateM_, replicateM)

19import Control.Exception (throwIO)

20import Control.Applicative ((<$>), (<*>))

21import qualified Network.Transport as NT (Transport, closeEndPoint)

22+import Network.Socket (sClose)

23+import Network.Transport.TCP

24+ ( createTransportExposeInternals

25+ , TransportInternals(socketBetween)

22−import Network.Transport.TCP (createTransport, defaultTCPParameters)

26+ , defaultTCPParameters

27+ )

23import Control.Distributed.Process

28import 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+ )

25import Control.Distributed.Process.Node

33import Control.Distributed.Process.Node

26import Control.Distributed.Process.Serializable (Serializable)

34import Control.Distributed.Process.Serializable (Serializable)

27import TestAuxiliary

35import TestAuxiliary

@@ -578,9 +586,75 @@ testSpawnLocal transport = do

578 send pid sport

586 send pid sport

579 expect

587 expect

580

588

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+

581main :: IO ()

655main :: IO ()

582main = do

656main = do

583− Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters

657+ Right (transport, transportInternals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters

584 runTests

658 runTests

585 [ ("Ping", testPing transport)

659 [ ("Ping", testPing transport)

586 , ("Math", testMath transport)

660 , ("Math", testMath transport)

@@ -619,4 +693,6 @@ main = do

619 -- Monitoring nodes and channels

693 -- Monitoring nodes and channels

620 , ("MonitorNode", testMonitorNode transport)

694 , ("MonitorNode", testMonitorNode transport)

621 , ("MonitorChannel", testMonitorChannel transport)

695 , ("MonitorChannel", testMonitorChannel transport)

696+ -- Reconnect

697+ , ("Reconnect", testReconnect transport transportInternals)

622 ]

698 ]