0c7663e · distributed-process

git.iko.soy/distributed-process

on master help sign in

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

0c7663eEdsko de Vries14 years ago

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

compare with master

unifiedsplit

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

99 -- * Local versions of 'spawn'

100 , spawnLocal

101 , spawnChannelLocal

102+ -- * Reconnecting

103+ , reconnect

104+ , reconnectNode

105+ , reconnectPort

102 ) where

106 ) where

103

107

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

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

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

208 -- Auxiliary API

212 -- Auxiliary API

209 , expectTimeout

213 , expectTimeout

210 , spawnAsync

214 , spawnAsync

215+ -- Reconnecting

216+ , reconnect

217+ , reconnectNode

218+ , reconnectPort

211 )

219 )

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

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

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

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

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

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

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

336call dict nid proc = do

344call dict nid proc = do

337 us <- getSelfPid

345 us <- getSelfPid

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

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

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

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

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

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

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

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

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

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

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

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

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

353 ]

361 ]

362+ -- Clean up connection to pid

363+ reconnect pid

354 return a

364 return a

355 Left err ->

365 Left err ->

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

366 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

3 sendPayload

4 , sendBinary

5 , sendMessage

6+ , reconnect

6 ) where

7 ) where

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

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

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

89 Nothing -> setupConnBetween node from to

90 Nothing -> setupConnBetween node from to

90 where

91 where

91 nodeState = localState node

92 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

59 , unlinkPort

60 , monitorNode

61 , monitorPort

62+ -- * Reconnecting

63+ , reconnect

64+ , reconnectNode

65+ , reconnectPort

62 ) where

66 ) where

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

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

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

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

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

507 Right x -> return x

511 Right x -> return x

508

512

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+

509--------------------------------------------------------------------------------

551--------------------------------------------------------------------------------

510-- Auxiliary functions --

552-- Auxiliary functions --

511--------------------------------------------------------------------------------

553--------------------------------------------------------------------------------

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

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

384 | WhereIs String

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

386 | NamedSend String Message

387+ | Reconnect Identifier

387 deriving Show

388 deriving Show

388

389

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

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

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

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

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

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

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

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

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

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

431 get = do

433 get = do

432 header <- getWord8

434 header <- getWord8

433 case header of

435 case header of

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

440 6 -> WhereIs <$> get

442 6 -> WhereIs <$> get

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

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

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

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

445+ 9 -> Reconnect <$> get

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

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

444

447

445instance Binary DiedReason where

448instance Binary DiedReason where

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

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

119 ( sendBinary

120 , sendMessage

121 , sendPayload

122+ , reconnect

122 )

123 )

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

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

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

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

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

424 ncEffectWhereIs from label

425 ncEffectWhereIs from label

425 NCMsg from (NamedSend label msg') ->

426 NCMsg from (NamedSend label msg') ->

426 ncEffectNamedSend from label msg'

427 ncEffectNamedSend from label msg'

428+ NCMsg from (Reconnect to) ->

429+ ncEffectReconnect from to

427 unexpected ->

430 unexpected ->

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

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

429

432

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

547 (ProcessIdentifier pid)

550 (ProcessIdentifier pid)

548 (messageToPayload msg)

551 (messageToPayload msg)

549

552

553+-- Reconnecting

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

555+ncEffectReconnect from to = do

556+ node <- ask

557+ liftIO $ reconnect node from to

558+

550--------------------------------------------------------------------------------

559--------------------------------------------------------------------------------

551-- Auxiliary --

560-- Auxiliary --

552--------------------------------------------------------------------------------

561--------------------------------------------------------------------------------

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

592destNid (Register _ _) = Nothing

601destNid (Register _ _) = Nothing

593destNid (WhereIs _) = Nothing

602destNid (WhereIs _) = Nothing

594destNid (NamedSend _ _) = Nothing

603destNid (NamedSend _ _) = Nothing

604+destNid (Reconnect _) = Nothing

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

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

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

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

597-- care of notifying remote nodes

607-- care of notifying remote nodes