Merge pull request haskell-distributed#263 from haskell-distributed/fd/more-unsafe-prims

facundominguez · facundominguez · commit 6deb13d07b57 · 2015-12-02T13:39:24.000-03:00
Unsafe primitives for usend and nsendRemote.
diff --git a/src/Control/Distributed/Process.hs b/src/Control/Distributed/Process.hs
@@ -40,8 +40,10 @@ module Control.Distributed.Process
   , mergePortsRR
     -- * Unsafe messaging variants
   , unsafeSend
+  , unsafeUSend
   , unsafeSendChan
   , unsafeNSend
+  , unsafeNSendRemote
   , unsafeWrapMessage
     -- * Advanced messaging
   , Match
@@ -212,8 +214,10 @@ import Control.Distributed.Process.Internal.Primitives
   , mergePortsBiased
   , mergePortsRR
   , unsafeSend
+  , unsafeUSend
   , unsafeSendChan
   , unsafeNSend
+  , unsafeNSendRemote
     -- Advanced messaging
   , Match
   , receiveWait
diff --git a/src/Control/Distributed/Process/Internal/Primitives.hs b/src/Control/Distributed/Process/Internal/Primitives.hs
@@ -23,8 +23,10 @@ module Control.Distributed.Process.Internal.Primitives
   , mergePortsRR
     -- * Unsafe messaging variants
   , unsafeSend
+  , unsafeUSend
   , unsafeSendChan
   , unsafeNSend
+  , unsafeNSendRemote
     -- * Advanced messaging
   , Match
   , receiveWait
@@ -279,6 +281,13 @@ usend them msg = do
       else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)
                                                      (createMessage msg)
 
+-- | /Unsafe/ variant of 'usend'. This function makes /no/ attempt to serialize
+-- the message when the destination process resides on the same local
+-- node. Therefore, a local receiver would need to be prepared to cope with any
+-- errors resulting from evaluation of the message.
+unsafeUSend :: Serializable a => ProcessId -> a -> Process ()
+unsafeUSend = Unsafe.usend
+
 -- | Wait for a message of a specific type
 expect :: forall a. Serializable a => Process a
 expect = receiveWait [match return]
@@ -1176,6 +1185,13 @@ nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
 nsendRemote nid label msg =
   sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
 
+-- | Named send to a process in a remote registry (asynchronous)
+-- This function makes /no/ attempt to serialize and (in the case when the
+-- destination process resides on the same local node) therefore ensure that
+-- the payload is fully evaluated before it is delivered.
+unsafeNSendRemote :: Serializable a => NodeId -> String -> a -> Process ()
+unsafeNSendRemote = Unsafe.nsendRemote
+
 --------------------------------------------------------------------------------
 -- Closures                                                                   --
 --------------------------------------------------------------------------------
diff --git a/src/Control/Distributed/Process/UnsafePrimitives.hs b/src/Control/Distributed/Process/UnsafePrimitives.hs
@@ -45,6 +45,8 @@ module Control.Distributed.Process.UnsafePrimitives
     send
   , sendChan
   , nsend
+  , nsendRemote
+  , usend
   , wrapMessage
   ) where
 
@@ -56,6 +58,7 @@ import Control.Distributed.Process.Internal.Messaging
 
 import Control.Distributed.Process.Internal.Types
   ( ProcessId(..)
+  , NodeId(..)
   , LocalNode(..)
   , LocalProcess(..)
   , Process(..)
@@ -65,6 +68,7 @@ import Control.Distributed.Process.Internal.Types
   , ImplicitReconnect(..)
   , SendPortId(..)
   , Message
+  , createMessage
   , sendPortProcessId
   , unsafeCreateUnencodedMessage
   )
@@ -78,6 +82,14 @@ nsend :: Serializable a => String -> a -> Process ()
 nsend label msg =
   sendCtrlMsg Nothing (NamedSend label (unsafeCreateUnencodedMessage msg))
 
+-- | Named send to a process in a remote registry (asynchronous)
+nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
+nsendRemote nid label msg = do
+  proc <- ask
+  if localNodeId (processNode proc) == nid
+    then nsend label msg
+    else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
+
 -- | Send a message
 send :: Serializable a => ProcessId -> a -> Process ()
 send them msg = do
@@ -96,6 +108,25 @@ send them msg = do
     unsafeSendLocal pid msg' =
       sendCtrlMsg Nothing $ LocalSend pid (unsafeCreateUnencodedMessage msg')
 
+-- | Send a message unreliably.
+--
+-- Unlike 'send', this function is insensitive to 'reconnect'. It will
+-- try to send the message regardless of the history of connection failures
+-- between the nodes.
+--
+-- Message passing with 'usend' is ordered for a given sender and receiver
+-- if the messages arrive at all.
+--
+usend :: Serializable a => ProcessId -> a -> Process ()
+usend them msg = do
+    proc <- ask
+    let there = processNodeId them
+    if localNodeId (processNode proc) == there
+      then sendCtrlMsg Nothing $
+             LocalSend them (unsafeCreateUnencodedMessage msg)
+      else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)
+                                                     (createMessage msg)
+
 -- | Send a message on a typed channel
 sendChan :: Serializable a => SendPort a -> a -> Process ()
 sendChan (SendPort cid) msg = do
