2e682fb · distributed-process

git.iko.soy/distributed-process

on master help sign in

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

2e682fbEdsko de Vries14 years ago

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

compare with master

unifiedsplit

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

98 -- * Local versions of 'spawn'

99 , spawnLocal

100 , spawnChannelLocal

101+ -- * Reconnecting

102+ , reconnect

103+ , reconnectNode

104+ , reconnectPort

101 ) where

105 ) where

102

106

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

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

@@ -206,6 +210,10 @@ import Control.Distributed.Process.Internal.Primitives

206 -- Auxiliary API

210 -- Auxiliary API

207 , expectTimeout

211 , expectTimeout

208 , spawnAsync

212 , spawnAsync

213+ -- Reconnecting

214+ , reconnect

215+ , reconnectNode

216+ , reconnectPort

209 )

217 )

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

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

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

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

@@ -333,7 +341,7 @@ spawnMonitor nid proc = do

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

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

334call dict nid proc = do

342call dict nid proc = do

335 us <- getSelfPid

343 us <- getSelfPid

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

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

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

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

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

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

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

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

@@ -349,6 +357,8 @@ call dict nid proc = do

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

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

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

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

351 ]

359 ]

360+ -- Clean up connection to pid

361+ reconnect pid

352 return a

362 return a

353 Left err ->

363 Left err ->

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

364 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 ((^.), (^=))

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

90 Nothing -> setupConnBetween node from to

91 Nothing -> setupConnBetween node from to

91 where

92 where

92 nodeState = localState node

93 nodeState = localState node

94+

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

96+reconnect node from to =

97+ modifyMVar_ (localState node) $ return .

98+ (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

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

385 | WhereIs String

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

387 | NamedSend String Message

388+ | Reconnect Identifier

388 deriving Show

389 deriving Show

389

390

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

391--------------------------------------------------------------------------------

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

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

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

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

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

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

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

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

432 get = do

434 get = do

433 header <- getWord8

435 header <- getWord8

434 case header of

436 case header of

@@ -441,6 +443,7 @@ instance Binary ProcessSignal where

441 6 -> WhereIs <$> get

443 6 -> WhereIs <$> get

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

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

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

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

446+ 9 -> Reconnect <$> get

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

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

445

448

446instance Binary DiedReason where

449instance Binary DiedReason where

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

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

121 ( sendBinary

122 , sendMessage

123 , sendPayload

124+ , reconnect

124 )

125 )

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

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

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

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

@@ -434,6 +435,8 @@ nodeController = do

434 ncEffectWhereIs from label

435 ncEffectWhereIs from label

435 NCMsg from (NamedSend label msg') ->

436 NCMsg from (NamedSend label msg') ->

436 ncEffectNamedSend from label msg'

437 ncEffectNamedSend from label msg'

438+ NCMsg from (Reconnect to) ->

439+ ncEffectReconnect from to

437 unexpected ->

440 unexpected ->

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

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

439

442

@@ -557,6 +560,12 @@ ncEffectNamedSend from label msg = do

557 (ProcessIdentifier pid)

560 (ProcessIdentifier pid)

558 (messageToPayload msg)

561 (messageToPayload msg)

559

562

563+-- Reconnecting

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

565+ncEffectReconnect from to = do

566+ node <- ask

567+ liftIO $ reconnect node from to

568+

560--------------------------------------------------------------------------------

569--------------------------------------------------------------------------------

561-- Auxiliary --

570-- Auxiliary --

562--------------------------------------------------------------------------------

571--------------------------------------------------------------------------------

@@ -602,6 +611,7 @@ destNid (Spawn _ _) = Nothing

602destNid (Register _ _) = Nothing

611destNid (Register _ _) = Nothing

603destNid (WhereIs _) = Nothing

612destNid (WhereIs _) = Nothing

604destNid (NamedSend _ _) = Nothing

613destNid (NamedSend _ _) = Nothing

614+destNid (Reconnect _) = Nothing

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

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

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

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

607-- care of notifying remote nodes

617-- care of notifying remote nodes