0c7663e · distributed-process

git.iko.soy/distributed-process

Implement reconnect. Use it to clean up in 'call'

commit0c7663e3b83042e9362b740a7286a565106f0205 authorEdsko de Vries <edsko@edsko.net> dateWed 15 Aug 2012, 15:56 (14 years ago) parent765585c

5 files changed+72−1

compare with master

unifiedsplit

distributed-process/src/Control/Distributed/Process.hs+11−1 distributed-process/src/Control/Distributed/Process/Internal/Node.hs+6 distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+42 distributed-process/src/Control/Distributed/Process/Internal/Types.hs+3 distributed-process/src/Control/Distributed/Process/Node.hs+10

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

@@ -99,6 +99,10 @@ module Control.Distributed.Process

9999 -- * Local versions of 'spawn'

100100 , spawnLocal

101101 , spawnChannelLocal

102+ -- * Reconnecting

103+ , reconnect

104+ , reconnectNode

105+ , reconnectPort

102106 ) where

103107

104108#if ! MIN_VERSION_base(4,6,0)

@@ -208,6 +212,10 @@ import Control.Distributed.Process.Internal.Primitives

208212 -- Auxiliary API

209213 , expectTimeout

210214 , spawnAsync

215+ -- Reconnecting

216+ , reconnect

217+ , reconnectNode

218+ , reconnectPort

211219 )

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

213221import Control.Distributed.Process.Node (forkProcess)

@@ -335,7 +343,7 @@ spawnMonitor nid proc = do

335343call :: Serializable a => Static (SerializableDict a) -> NodeId -> Closure (Process a) -> Process a

336344call dict nid proc = do

337345 us <- getSelfPid

338− (_, mRef) <- spawnMonitor nid (proc `bindCP` cpSend dict us)

346+ (pid, mRef) <- spawnMonitor nid (proc `bindCP` cpSend dict us)

339347 -- We are guaranteed to receive the reply before the monitor notification

340348 -- (if a reply is sent at all)

341349 -- NOTE: This might not be true if we switch to unreliable delivery.

@@ -351,6 +359,8 @@ call dict nid proc = do

351359 [ matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)

352360 (\(ProcessMonitorNotification {}) -> return ())

353361 ]

362+ -- Clean up connection to pid

363+ reconnect pid

354364 return a

355365 Left err ->

356366 fail $ "call: remote process died: " ++ show err

distributed-process/src/Control/Distributed/Process/Internal/Node.hs+6

@@ -3,6 +3,7 @@ module Control.Distributed.Process.Internal.Node

33 sendPayload

44 , sendBinary

55 , sendMessage

6+ , reconnect

67 ) where

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

@@ -89,3 +90,8 @@ connBetween node from to = do

8990 Nothing -> setupConnBetween node from to

9091 where

9192 nodeState = localState node

93+

94+reconnect :: LocalNode -> Identifier -> Identifier -> IO ()

95+reconnect node from to =

96+ modifyMVar_ (localState node) $ return .

97+ (localConnectionBetween from to ^= Nothing)

distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+42

@@ -59,6 +59,10 @@ module Control.Distributed.Process.Internal.Primitives

5959 , unlinkPort

6060 , monitorNode

6161 , monitorPort

62+ -- * Reconnecting

63+ , reconnect

64+ , reconnectNode

65+ , reconnectPort

6266 ) where

6367

6468#if ! MIN_VERSION_base(4,6,0)

@@ -506,6 +510,44 @@ unClosure closure = do

506510 Left err -> fail $ "Could not resolve closure: " ++ err

507511 Right x -> return x

508512

513+--------------------------------------------------------------------------------

514+-- Reconnecting --

515+--------------------------------------------------------------------------------

516+

517+-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered

518+-- message passing. However, when network connections get disrupted this

519+-- illusion cannot always be maintained. Once a network connection breaks (even

520+-- temporarily) no further communication on that connection will be possible.

521+-- For example, if process A sends a message to process B, and A is then

522+-- notified (by monitor notification) that it got disconnected from B, A will

523+-- not be able to send any further messages to B, /unless/ A explicitly

524+-- indicates that it is acceptable to attempt to reconnect to B using the

525+-- Cloud Haskell 'reconnect' primitive.

526+--

527+-- Importantly, when A calls 'reconnect' it acknowledges that some messages to

528+-- B might have been lost. For instance, if A sends messages m1 and m2 to B,

529+-- then receives a monitor notification that its connection to B has been lost,

530+-- calls 'reconnect' and then sends m3, it is possible that B will receive m1

531+-- and m3 but not m2.

532+--

533+-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay

534+-- to attempt to reconnect on the next send/. In particular, if no further

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

536+-- its connection to B.

537+reconnect :: ProcessId -> Process ()

538+reconnect =

539+ sendCtrlMsg Nothing . Reconnect . ProcessIdentifier

540+

541+-- | Reconnect to a node. See 'reconnect' for more information.

542+reconnectNode :: NodeId -> Process ()

543+reconnectNode =

544+ sendCtrlMsg Nothing . Reconnect . NodeIdentifier

545+

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

547+reconnectPort :: SendPort a -> Process ()

548+reconnectPort =

549+ sendCtrlMsg Nothing . Reconnect . SendPortIdentifier . sendPortId

550+

509551--------------------------------------------------------------------------------

510552-- Auxiliary functions --

511553--------------------------------------------------------------------------------

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

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

384384 | WhereIs String

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

386386 | NamedSend String Message

387+ | Reconnect Identifier

387388 deriving Show

388389

389390--------------------------------------------------------------------------------

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

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

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

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

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

431433 get = do

432434 header <- getWord8

433435 case header of

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

440442 6 -> WhereIs <$> get

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

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

445+ 9 -> Reconnect <$> get

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

444447

445448instance Binary DiedReason where

distributed-process/src/Control/Distributed/Process/Node.hs+10

@@ -119,6 +119,7 @@ import Control.Distributed.Process.Internal.Node

119119 ( sendBinary

120120 , sendMessage

121121 , sendPayload

122+ , reconnect

122123 )

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

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

@@ -424,6 +425,8 @@ nodeController = do

424425 ncEffectWhereIs from label

425426 NCMsg from (NamedSend label msg') ->

426427 ncEffectNamedSend from label msg'

428+ NCMsg from (Reconnect to) ->

429+ ncEffectReconnect from to

427430 unexpected ->

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

429432

@@ -547,6 +550,12 @@ ncEffectNamedSend from label msg = do

547550 (ProcessIdentifier pid)

548551 (messageToPayload msg)

549552

553+-- Reconnecting

554+ncEffectReconnect :: Identifier -> Identifier -> NC ()

555+ncEffectReconnect from to = do

556+ node <- ask

557+ liftIO $ reconnect node from to

558+

550559--------------------------------------------------------------------------------

551560-- Auxiliary --

552561--------------------------------------------------------------------------------

@@ -592,6 +601,7 @@ destNid (Spawn _ _) = Nothing

592601destNid (Register _ _) = Nothing

593602destNid (WhereIs _) = Nothing

594603destNid (NamedSend _ _) = Nothing

604+destNid (Reconnect _) = Nothing

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

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

597607-- care of notifying remote nodes