d126187 · distributed-process

git.iko.soy/distributed-process

Implement unreliable send (usend).

Unlike 'send', the function 'usend' is insensitive to 'reconnect'. It will try to send the message regardless of the history of connection failures between the nodes. It saves the tediousness of dealing with 'monitor' and 'reconnect' calls when reliability is not required.

commitd126187b5cdbed976bf6d389d8e07a515dd81277 authorFacundo Domínguez <facundo.dominguez@tweag.io> dateWed 14 Jan 2015, 17:22 (12 years ago) parent87309ac

4 files changed+23

compare with master

unifiedsplit

src/Control/Distributed/Process.hs+2 src/Control/Distributed/Process/Internal/Primitives.hs+15 src/Control/Distributed/Process/Internal/Types.hs+3 src/Control/Distributed/Process/Node.hs+3

src/Control/Distributed/Process.hs+2

@@ -25,6 +25,7 @@ module Control.Distributed.Process

2525 , liftIO -- Reexported for convenience

2626 -- * Basic messaging

2727 , send

28+ , usend

2829 , expect

2930 , expectTimeout

3031 -- * Channels

@@ -195,6 +196,7 @@ import Control.Distributed.Process.Serializable (Serializable)

195196import Control.Distributed.Process.Internal.Primitives

196197 ( -- Basic messaging

197198 send

199+ , usend

198200 , expect

199201 -- Channels

200202 , newChan

src/Control/Distributed/Process/Internal/Primitives.hs+15

@@ -12,6 +12,7 @@

1212module Control.Distributed.Process.Internal.Primitives

1313 ( -- * Basic messaging

1414 send

15+ , usend

1516 , expect

1617 -- * Channels

1718 , newChan

@@ -254,6 +255,20 @@ send them msg = do

254255unsafeSend :: Serializable a => ProcessId -> a -> Process ()

255256unsafeSend = Unsafe.send

256257

258+-- | Send a message without any reliability or ordering guarantees.

259+--

260+-- Unlike 'send', this function is insensitive to 'reconnect'. It will

261+-- try to send the message regardless of the history of connection failures

262+-- between the nodes.

263+usend :: Serializable a => ProcessId -> a -> Process ()

264+usend them msg = do

265+ here <- getSelfNode

266+ let there = processNodeId them

267+ if here == there

268+ then sendLocal them msg

269+ else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)

270+ (createMessage msg)

271+

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

258273expect :: forall a. Serializable a => Process a

259274expect = receiveWait [match return]

src/Control/Distributed/Process/Internal/Types.hs+3

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

571571 | WhereIs !String

572572 | Register !String !NodeId !(Maybe ProcessId) !Bool -- Use 'Nothing' to unregister, use True to force reregister

573573 | NamedSend !String !Message

574+ | UnreliableSend !LocalProcessId !Message

574575 | LocalSend !ProcessId !Message

575576 | LocalPortSend !SendPortId !Message

576577 | Kill !ProcessId !String

@@ -626,6 +627,7 @@ instance Binary ProcessSignal where

626627 put (Exit pid reason) = putWord8 10 >> put pid >> put (messageToPayload reason)

627628 put (LocalSend to' msg) = putWord8 11 >> put to' >> put (messageToPayload msg)

628629 put (LocalPortSend sid msg) = putWord8 12 >> put sid >> put (messageToPayload msg)

630+ put (UnreliableSend lpid msg) = putWord8 13 >> put lpid >> put (messageToPayload msg)

629631 put (GetInfo about) = putWord8 30 >> put about

630632 put (SigShutdown) = putWord8 31

631633 put (GetNodeStats nid) = putWord8 32 >> put nid

@@ -645,6 +647,7 @@ instance Binary ProcessSignal where

645647 10 -> Exit <$> get <*> (payloadToMessage <$> get)

646648 11 -> LocalSend <$> get <*> (payloadToMessage <$> get)

647649 12 -> LocalPortSend <$> get <*> (payloadToMessage <$> get)

650+ 13 -> UnreliableSend <$> get <*> (payloadToMessage <$> get)

648651 30 -> GetInfo <$> get

649652 31 -> return SigShutdown

650653 32 -> GetNodeStats <$> get

src/Control/Distributed/Process/Node.hs+3

@@ -688,6 +688,8 @@ nodeController = do

688688 ncEffectWhereIs from label

689689 NCMsg _ (NamedSend label msg') ->

690690 ncEffectNamedSend label msg'

691+ NCMsg _ (UnreliableSend lpid msg') ->

692+ ncEffectLocalSend node (ProcessId (localNodeId node) lpid) msg'

691693 NCMsg _ (LocalSend to msg') ->

692694 ncEffectLocalSend node to msg'

693695 NCMsg _ (LocalPortSend to msg') ->

@@ -1062,6 +1064,7 @@ destNid (Spawn _ _) = Nothing

10621064destNid (Register _ _ _ _) = Nothing

10631065destNid (WhereIs _) = Nothing

10641066destNid (NamedSend _ _) = Nothing

1067+destNid (UnreliableSend _ _) = Nothing

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

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

10671070-- care of notifying remote nodes