fe428ee · distributed-process

src/Control/Distributed/Process.hs+2

compare with master

unifiedsplit

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

25 , liftIO -- Reexported for convenience

26 -- * Basic messaging

27 , send

28+ , usend

28 , expect

29 , expect

29 , expectTimeout

30 , expectTimeout

30 -- * Channels

31 -- * Channels

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

196import Control.Distributed.Process.Internal.Primitives

197import Control.Distributed.Process.Internal.Primitives

197 ( -- Basic messaging

198 ( -- Basic messaging

198 send

199 send

200+ , usend

199 , expect

201 , expect

200 -- Channels

202 -- Channels

201 , newChan

203 , newChan

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

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

12module Control.Distributed.Process.Internal.Primitives

13 ( -- * Basic messaging

14 send

15+ , usend

15 , expect

16 , expect

16 -- * Channels

17 -- * Channels

17 , newChan

18 , newChan

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

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

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

255unsafeSend = Unsafe.send

256unsafeSend = Unsafe.send

256

257

258+-- | Send a message unreliably.

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

264+-- Message passing with 'usend' is ordered for a given sender and receiver

265+-- if the messages arrive at all.

266+--

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

268+usend them msg = do

269+ here <- getSelfNode

270+ let there = processNodeId them

271+ if here == there

272+ then sendLocal them msg

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

274+ (createMessage msg)

275+

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

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

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

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

259expect = receiveWait [match return]

278expect = receiveWait [match return]

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

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

572 | WhereIs !String

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

574 | NamedSend !String !Message

575+ | UnreliableSend !LocalProcessId !Message

575 | LocalSend !ProcessId !Message

576 | LocalSend !ProcessId !Message

576 | LocalPortSend !SendPortId !Message

577 | LocalPortSend !SendPortId !Message

577 | Kill !ProcessId !String

578 | Kill !ProcessId !String

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

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

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

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

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

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

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

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

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

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

631 put (SigShutdown) = putWord8 31

633 put (SigShutdown) = putWord8 31

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

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

@@ -646,6 +648,7 @@ instance Binary ProcessSignal where

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

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

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

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

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

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

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

649 30 -> GetInfo <$> get

652 30 -> GetInfo <$> get

650 31 -> return SigShutdown

653 31 -> return SigShutdown

651 32 -> GetNodeStats <$> get

654 32 -> GetNodeStats <$> get

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

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

693 ncEffectWhereIs from label

694 NCMsg _ (NamedSend label msg') ->

695 ncEffectNamedSend label msg'

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

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

696 NCMsg _ (LocalSend to msg') ->

698 NCMsg _ (LocalSend to msg') ->

697 ncEffectLocalSend node to msg'

699 ncEffectLocalSend node to msg'

698 NCMsg _ (LocalPortSend to msg') ->

700 NCMsg _ (LocalPortSend to msg') ->

@@ -1067,6 +1069,7 @@ destNid (Spawn _ _) = Nothing

1067destNid (Register _ _ _ _) = Nothing

1069destNid (Register _ _ _ _) = Nothing

1068destNid (WhereIs _) = Nothing

1070destNid (WhereIs _) = Nothing

1069destNid (NamedSend _ _) = Nothing

1071destNid (NamedSend _ _) = Nothing

1072+destNid (UnreliableSend _ _) = Nothing

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

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

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

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

1072-- care of notifying remote nodes

1075-- care of notifying remote nodes