distributed-process 0.3.0 → 0.3.1

raw patch · 11 files changed

+342/−193 lines, 11 filesdep +distributed-processdep ~distributed-staticdep ~ghc-primdep ~template-haskellPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: distributed-process

Dependency ranges changed: distributed-static, ghc-prim, template-haskell

API changes (from Hackage documentation)

Files

• distributed-process.cabal +30/−56
• src/Control/Distributed/Process.hs +28/−9
• src/Control/Distributed/Process/Closure.hs +20/−15
• src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs +6/−5
• src/Control/Distributed/Process/Internal/Closure/TH.hs +2/−2
• src/Control/Distributed/Process/Internal/Node.hs +15/−3
• src/Control/Distributed/Process/Internal/Primitives.hs +44/−2
• src/Control/Distributed/Process/Internal/StrictMVar.hs +51/−0
• src/Control/Distributed/Process/Internal/Types.hs +22/−19
• src/Control/Distributed/Process/Node.hs +122/−81
• tests/TestClosure.hs +2/−1

@@ -1,5 +1,5 @@ Name:          distributed-process -Version:       0.3.0+Version:       0.3.1 Cabal-Version: >=1.8 Build-Type:    Simple License:       BSD3 @@ -29,6 +29,10 @@   Location: https://github.com/haskell-distributed/distributed-process   SubDir:   distributed-process +flag th+  description: Build with Template Haskell support+  default: True+ Library   Build-Depends:     base >= 4.4 && < 5,                      binary >= 0.5 && < 0.6,@@ -41,10 +45,9 @@                      containers >= 0.4 && < 0.6,                      old-locale >= 1.0 && < 1.1,                      time >= 1.2 && < 1.5,-                     template-haskell >= 2.6 && < 2.8,                      random >= 1.0 && < 1.1,-                     ghc-prim >= 0.2 && < 0.3,-                     distributed-static >= 0.1 && < 0.2,+                     ghc-prim >= 0.2 && < 0.4,+                     distributed-static >= 0.2 && < 0.3,                      rank1dynamic >= 0.1 && < 0.2,                      syb >= 0.3 && < 0.4   Exposed-modules:   Control.Distributed.Process,@@ -54,9 +57,9 @@                      Control.Distributed.Process.Internal.Primitives,                      Control.Distributed.Process.Internal.CQueue,                      Control.Distributed.Process.Internal.Types,-                     Control.Distributed.Process.Internal.Closure.TH,                      Control.Distributed.Process.Internal.Closure.BuiltIn,                      Control.Distributed.Process.Internal.Node+                     Control.Distributed.Process.Internal.StrictMVar   Extensions:        RankNTypes,                      ScopedTypeVariables,                      FlexibleInstances,@@ -65,75 +68,46 @@                      GADTs,                      GeneralizedNewtypeDeriving,                      DeriveDataTypeable,-                     CPP+                     CPP,+                     BangPatterns   ghc-options:       -Wall   HS-Source-Dirs:    src+  if flag(th)+     Build-Depends:   template-haskell >= 2.6 && < 2.9+     Exposed-modules: Control.Distributed.Process.Internal.Closure.TH+     CPP-Options:     -DTemplateHaskellSupport  Test-Suite TestCH   Type:              exitcode-stdio-1.0   Main-Is:           TestCH.hs   Build-Depends:     base >= 4.4 && < 5,-                     binary >= 0.5 && < 0.6,-                     network-transport >= 0.2 && < 0.3,-                     network-transport-tcp >= 0.2 && < 0.3,-                     stm >= 2.3 && < 2.5,-                     transformers >= 0.2 && < 0.4,-                     mtl >= 2.0 && < 2.2,-                     data-accessor >= 0.2 && < 0.3,-                     bytestring >= 0.9 && < 0.11,-                     containers >= 0.4 && < 0.6,-                     old-locale >= 1.0 && < 1.1,-                     time >= 1.2 && < 1.5,-                     template-haskell >= 2.6 && < 2.8,                      random >= 1.0 && < 1.1,-                     ghc-prim >= 0.2 && < 0.3,                      ansi-terminal >= 0.5 && < 0.6,-                     distributed-static >= 0.1 && < 0.2,-                     rank1dynamic >= 0.1 && < 0.2-  Extensions:        RankNTypes,+                     distributed-process >= 0.3 && < 0.4,+                     network-transport >= 0.2 && < 0.3,+                     network-transport-tcp >= 0.2 && < 0.3,+                     binary >= 0.5 && < 0.6+  Other-modules:     TestAuxiliary+  Extensions:        CPP,                      ScopedTypeVariables,-                     FlexibleInstances,-                     UndecidableInstances,-                     ExistentialQuantification,-                     GADTs,-                     GeneralizedNewtypeDeriving,                      DeriveDataTypeable,-                     CPP+                     GeneralizedNewtypeDeriving   ghc-options:       -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind -  HS-Source-Dirs:    tests src+  HS-Source-Dirs:    tests   Test-Suite TestClosure   Type:              exitcode-stdio-1.0   Main-Is:           TestClosure.hs   Build-Depends:     base >= 4.4 && < 5,-                     binary >= 0.5 && < 0.6,-                     network-transport >= 0.2 && < 0.3,-                     network-transport-tcp >= 0.2 && < 0.3,-                     stm >= 2.3 && < 2.5,-                     transformers >= 0.2 && < 0.4,-                     mtl >= 2.0 && < 2.2,-                     data-accessor >= 0.2 && < 0.3,-                     bytestring >= 0.9 && < 0.11,-                     containers >= 0.4 && < 0.6,-                     old-locale >= 1.0 && < 1.1,-                     time >= 1.2 && < 1.5,-                     template-haskell >= 2.6 && < 2.8,                      random >= 1.0 && < 1.1,-                     ghc-prim >= 0.2 && < 0.3,                      ansi-terminal >= 0.5 && < 0.6,-                     distributed-static >= 0.1 && < 0.2,-                     rank1dynamic >= 0.1 && < 0.2,-                     syb >= 0.3 && < 0.4+                     distributed-static >= 0.2 && < 0.3,+                     distributed-process >= 0.3 && < 0.4,+                     network-transport >= 0.2 && < 0.3,+                     network-transport-tcp >= 0.2 && < 0.3,+                     bytestring >= 0.9 && < 0.11   Other-modules:     TestAuxiliary-  Extensions:        RankNTypes,-                     ScopedTypeVariables,-                     FlexibleInstances,-                     UndecidableInstances,-                     ExistentialQuantification,-                     GADTs,-                     GeneralizedNewtypeDeriving,-                     DeriveDataTypeable,-                     TemplateHaskell,-                     CPP+  Extensions:        CPP, +                     ScopedTypeVariables   ghc-options:       -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind -  HS-Source-Dirs:    tests src+  HS-Source-Dirs:    tests 

@@ -63,7 +63,8 @@   , PortMonitorNotification(..)   , DiedReason(..)     -- * Closures-  , Closure(..)+  , Closure+  , closure   , Static   , unClosure   , RemoteTable@@ -98,6 +99,10 @@     -- * Local versions of 'spawn'    , spawnLocal   , spawnChannelLocal+    -- * Reconnecting+  , reconnect+  , reconnectNode+  , reconnectPort   ) where  #if ! MIN_VERSION_base(4,6,0)@@ -109,7 +114,8 @@ import Control.Applicative ((<$>)) import Control.Monad.Reader (ask) import Control.Distributed.Static -  ( Closure(..)+  ( Closure+  , closure   , Static   , RemoteTable   )@@ -206,6 +212,10 @@     -- Auxiliary API   , expectTimeout   , spawnAsync+    -- Reconnecting+  , reconnect+  , reconnectNode+  , reconnectPort   ) import Control.Distributed.Process.Serializable (Serializable) import Control.Distributed.Process.Node (forkProcess)@@ -294,14 +304,14 @@                    `seqCP` proc   mPid <- receiveWait      [ matchIf (\(DidSpawn ref _) -> ref == sRef)-              (\(DidSpawn _ pid) -> return $ Just pid)+              (\(DidSpawn _ pid) -> return $ Right pid)     , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)-              (\_ -> return Nothing)+              (\(NodeMonitorNotification _ _ err) -> return $ Left err)     ]   unmonitor mRef   case mPid of-    Nothing  -> fail "spawn: remote node failed"-    Just pid -> send pid () >> return pid+    Left err  -> fail $ "spawn: remote node failed: " ++ show err+    Right pid -> send pid () >> return pid  -- | Spawn a process and link to it --@@ -333,7 +343,7 @@ call :: Serializable a => Static (SerializableDict a) -> NodeId -> Closure (Process a) -> Process a call dict nid proc = do    us <- getSelfPid-  (_, mRef) <- spawnMonitor nid (proc `bindCP` cpSend dict us)+  (pid, mRef) <- spawnMonitor nid (proc `bindCP` cpSend dict us)   -- We are guaranteed to receive the reply before the monitor notification   -- (if a reply is sent at all)   -- NOTE: This might not be true if we switch to unreliable delivery.@@ -343,8 +353,17 @@               (\(ProcessMonitorNotification _ _ reason) -> return (Left reason))     ]   case mResult of-    Right a  -> unmonitor mRef >> return a-    Left err -> fail $ "call: remote process died: " ++ show err +    Right a  -> do+      -- Wait for the monitor message so that we the mailbox doesn't grow +      receiveWait +        [ matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)+                  (\(ProcessMonitorNotification {}) -> return ())+        ]+      -- Clean up connection to pid+      reconnect pid+      return a+    Left err -> +      fail $ "call: remote process died: " ++ show err   -- | Spawn a child process, have the child link to the parent and the parent -- monitor the child

@@ -1,3 +1,4 @@+{-# LANGUAGE CPP #-} -- | /Towards Haskell in the Cloud/ (Epstein et al., Haskell Symposium 2011) -- proposes a new type construct called 'static' that characterizes values that -- are known statically. Cloud Haskell uses the@@ -156,14 +157,8 @@ --      instance into scope, but unless proper 'static' support is added to --      ghc we need both the type class argument and the explicit dictionary.  module Control.Distributed.Process.Closure -  ( -- * Template Haskell support for creating static values and closures -    remotable-  , mkStatic-  , mkClosure-  , functionSDict-  , functionTDict-    -- * Serialization dictionaries (and their static versions)-  , SerializableDict(..)+  ( -- * Serialization dictionaries (and their static versions)+    SerializableDict(..)   , staticDecode   , sdictUnit   , sdictProcessId@@ -181,16 +176,17 @@   , cpSend   , cpExpect   , cpNewChan-  ) where --import Control.Distributed.Process.Serializable (SerializableDict(..))-import Control.Distributed.Process.Internal.Closure.TH -  ( remotable+#ifdef TemplateHaskellSupport +    -- * Template Haskell support for creating static values and closures +  , remotable   , mkStatic+  , mkClosure   , functionSDict   , functionTDict-  , mkClosure-  )+#endif+  ) where ++import Control.Distributed.Process.Serializable (SerializableDict(..)) import Control.Distributed.Process.Internal.Closure.BuiltIn   ( -- Static dictionaries and associated operations     staticDecode@@ -211,3 +207,12 @@   , cpExpect   , cpNewChan   )+#ifdef TemplateHaskellSupport +import Control.Distributed.Process.Internal.Closure.TH +  ( remotable+  , mkStatic+  , functionSDict+  , functionTDict+  , mkClosure+  )+#endif

@@ -33,7 +33,8 @@   , Static   , staticLabel   , staticApply-  , Closure(Closure)+  , Closure+  , closure   , closureApplyStatic   , closureApply   , staticCompose@@ -159,7 +160,7 @@ -- | 'CP' version of 'Control.Monad.return' returnCP :: forall a. Serializable a           => Static (SerializableDict a) -> a -> Closure (Process a)-returnCP dict x = Closure decoder (encode x)+returnCP dict x = closure decoder (encode x)   where     decoder :: Static (ByteString -> Process a)     decoder = returnProcessStatic@@ -193,14 +194,14 @@  -- | 'CP' version of 'link' cpLink :: ProcessId -> Closure (Process ())-cpLink = Closure (linkStatic `staticCompose` decodeProcessIdStatic) . encode +cpLink = closure (linkStatic `staticCompose` decodeProcessIdStatic) . encode    where     linkStatic :: Static (ProcessId -> Process ())     linkStatic = staticLabel "$link"  -- | 'CP' version of 'unlink' cpUnlink :: ProcessId -> Closure (Process ())-cpUnlink = Closure (unlinkStatic `staticCompose` decodeProcessIdStatic) . encode+cpUnlink = closure (unlinkStatic `staticCompose` decodeProcessIdStatic) . encode   where     unlinkStatic :: Static (ProcessId -> Process ())     unlinkStatic = staticLabel "$unlink"@@ -208,7 +209,7 @@ -- | 'CP' version of 'send' cpSend :: forall a. Typeable a         => Static (SerializableDict a) -> ProcessId -> CP a () -cpSend dict pid = Closure decoder (encode pid)+cpSend dict pid = closure decoder (encode pid)   where     decoder :: Static (ByteString -> a -> Process ())     decoder = (sendDictStatic `staticApply` dict)

@@ -52,7 +52,7 @@   , registerStatic   , Static   , staticLabel-  , Closure(Closure)+  , closure   , staticCompose   ) import Control.Distributed.Process.Internal.Types (Process)@@ -97,7 +97,7 @@  mkClosure :: Name -> Q Exp mkClosure n = -  [|   Closure ($(mkStatic n) `staticCompose` staticDecode $(functionSDict n)) +  [|   closure ($(mkStatic n) `staticCompose` staticDecode $(functionSDict n))       . encode   |] 

@@ -3,13 +3,14 @@     sendPayload   , sendBinary   , sendMessage+  , reconnect   ) where  import Data.Accessor ((^.), (^=)) import Data.Binary (Binary, encode) import qualified Data.ByteString.Lazy as BSL (toChunks) import qualified Data.ByteString as BSS (ByteString)-import Control.Concurrent.MVar (withMVar, modifyMVar_)+import Control.Distributed.Process.Internal.StrictMVar (withMVar, modifyMVar_) import Control.Concurrent.Chan (writeChan) import Control.Monad (unless) import qualified Network.Transport as NT @@ -18,6 +19,7 @@   , defaultConnectHints   , connect   , Reliability(ReliableOrdered)+  , close   ) import Control.Distributed.Process.Internal.Types    ( LocalNode(localState, localEndPoint, localCtrlChan)@@ -44,8 +46,8 @@     Just conn -> do       didSend <- NT.send conn payload       case didSend of-        Left _  -> return False-        Right _ -> return True +        Left _err -> return False +        Right ()  -> return True      Nothing -> return False   unless didSend $     writeChan (localCtrlChan node) NCMsg@@ -89,3 +91,13 @@       Nothing   -> setupConnBetween node from to    where     nodeState = localState node++reconnect :: LocalNode -> Identifier -> Identifier -> IO ()+reconnect node from to =+  modifyMVar_ (localState node) $ \st -> +    case st ^. localConnectionBetween from to of+      Nothing -> +        return st+      Just conn -> do+        NT.close conn +        return (localConnectionBetween from to ^= Nothing $ st)

@@ -59,6 +59,10 @@   , unlinkPort   , monitorNode   , monitorPort+    -- * Reconnecting+  , reconnect+  , reconnectNode+  , reconnectPort   ) where  #if ! MIN_VERSION_base(4,6,0)@@ -74,7 +78,7 @@ import Control.Applicative ((<$>)) import Control.Exception (Exception, throwIO, SomeException) import qualified Control.Exception as Ex (catch, mask)-import Control.Concurrent.MVar (modifyMVar)+import Control.Distributed.Process.Internal.StrictMVar (modifyMVar) import Control.Concurrent.Chan (writeChan) import Control.Concurrent.STM    ( STM@@ -346,7 +350,7 @@  -- | Lift 'Control.Exception.bracket' bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c-bracket before after thing = do+bracket before after thing =   mask $ \restore -> do     a <- before     r <- restore (thing a) `onException` after a@@ -505,6 +509,44 @@   case Static.unclosure rtable closure of     Left err -> fail $ "Could not resolve closure: " ++ err      Right x  -> return x++--------------------------------------------------------------------------------+-- Reconnecting                                                               --+--------------------------------------------------------------------------------++-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered+-- message passing. However, when network connections get disrupted this+-- illusion cannot always be maintained. Once a network connection breaks (even+-- temporarily) no further communication on that connection will be possible.+-- For example, if process A sends a message to process B, and A is then +-- notified (by monitor notification) that it got disconnected from B, A will+-- not be able to send any further messages to B, /unless/ A explicitly +-- indicates that it is acceptable to attempt to reconnect to B using the+-- Cloud Haskell 'reconnect' primitive. +--+-- Importantly, when A calls 'reconnect' it acknowledges that some messages to+-- B might have been lost. For instance, if A sends messages m1 and m2 to B,+-- then receives a monitor notification that its connection to B has been lost,+-- calls 'reconnect' and then sends m3, it is possible that B will receive m1+-- and m3 but not m2.+--+-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay+-- to attempt to reconnect on the next send/. In particular, if no further+-- communication attempts are made to B then A can use reconnect to clean up+-- its connection to B.+reconnect :: ProcessId -> Process ()+reconnect = +  sendCtrlMsg Nothing . Reconnect . ProcessIdentifier ++-- | Reconnect to a node. See 'reconnect' for more information.+reconnectNode :: NodeId -> Process ()+reconnectNode = +  sendCtrlMsg Nothing . Reconnect . NodeIdentifier ++-- | Reconnect to a sendport. See 'reconnect' for more information.+reconnectPort :: SendPort a -> Process ()+reconnectPort = +  sendCtrlMsg Nothing . Reconnect . SendPortIdentifier . sendPortId  -------------------------------------------------------------------------------- -- Auxiliary functions                                                        --

@@ -0,0 +1,51 @@+-- | Like Control.Concurrent.MVar.Strict but reduce to HNF, not NF+module Control.Distributed.Process.Internal.StrictMVar +  ( StrictMVar+  , newEmptyMVar+  , newMVar+  , takeMVar+  , putMVar+  , withMVar+  , modifyMVar_+  , modifyMVar+  ) where++import Control.Applicative ((<$>))+import Control.Monad ((>=>))+import Control.Exception (evaluate)+import qualified Control.Concurrent.MVar as MVar +  ( MVar+  , newEmptyMVar+  , newMVar+  , takeMVar+  , putMVar+  , withMVar+  , modifyMVar_+  , modifyMVar+  )++newtype StrictMVar a = StrictMVar (MVar.MVar a)++newEmptyMVar :: IO (StrictMVar a)+newEmptyMVar = StrictMVar <$> MVar.newEmptyMVar++newMVar :: a -> IO (StrictMVar a)+newMVar x = evaluate x >> StrictMVar <$> MVar.newMVar x++takeMVar :: StrictMVar a -> IO a+takeMVar (StrictMVar v) = MVar.takeMVar v++putMVar :: StrictMVar a -> a -> IO ()+putMVar (StrictMVar v) x = evaluate x >> MVar.putMVar v x++withMVar :: StrictMVar a -> (a -> IO b) -> IO b+withMVar (StrictMVar v) = MVar.withMVar v++modifyMVar_ :: StrictMVar a -> (a -> IO a) -> IO ()+modifyMVar_ (StrictMVar v) f = MVar.modifyMVar_ v (f >=> evaluate) ++modifyMVar :: StrictMVar a -> (a -> IO (a, b)) -> IO b+modifyMVar (StrictMVar v) f = MVar.modifyMVar v (f >=> evaluateFst)+  where+    evaluateFst :: (a, b) -> IO (a, b)+    evaluateFst (x, y) = evaluate x >> return (x, y)

@@ -77,7 +77,6 @@ import Control.Category ((>>>)) import Control.Exception (Exception) import Control.Concurrent (ThreadId)-import Control.Concurrent.MVar (MVar) import Control.Concurrent.Chan (Chan) import Control.Concurrent.STM (TChan, TVar) import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)@@ -94,6 +93,7 @@   , showFingerprint   ) import Control.Distributed.Process.Internal.CQueue (CQueue)+import Control.Distributed.Process.Internal.StrictMVar (StrictMVar) import Control.Distributed.Static (RemoteTable, Closure)  -- import Control.Distributed.Process.Internal.Dynamic (Dynamic) @@ -113,17 +113,17 @@ -- | A local process ID consists of a seed which distinguishes processes from -- different instances of the same local node and a counter data LocalProcessId = LocalProcessId -  { lpidUnique  :: Int32-  , lpidCounter :: Int32+  { lpidUnique  :: {-# UNPACK #-} !Int32+  , lpidCounter :: {-# UNPACK #-} !Int32   }   deriving (Eq, Ord, Typeable, Show)  -- | Process identifier data ProcessId = ProcessId    { -- | The ID of the node the process is running on-    processNodeId  :: NodeId+    processNodeId  :: !NodeId     -- | Node-local identifier for the process-  , processLocalId :: LocalProcessId +  , processLocalId :: {-# UNPACK #-} !LocalProcessId    }   deriving (Eq, Ord, Typeable) @@ -159,7 +159,7 @@     -- | The network endpoint associated with this node    , localEndPoint :: NT.EndPoint      -- | Local node state -  , localState :: MVar LocalNodeState+  , localState :: StrictMVar LocalNodeState     -- | Channel for the node controller   , localCtrlChan :: Chan NCMsg     -- | Runtime lookup table for supporting closures@@ -169,17 +169,17 @@  -- | Local node state data LocalNodeState = LocalNodeState -  { _localProcesses   :: Map LocalProcessId LocalProcess-  , _localPidCounter  :: Int32-  , _localPidUnique   :: Int32-  , _localConnections :: Map (Identifier, Identifier) NT.Connection+  { _localProcesses   :: !(Map LocalProcessId LocalProcess)+  , _localPidCounter  :: !Int32+  , _localPidUnique   :: !Int32+  , _localConnections :: !(Map (Identifier, Identifier) NT.Connection)   }  -- | Processes running on our local node data LocalProcess = LocalProcess    { processQueue  :: CQueue Message    , processId     :: ProcessId-  , processState  :: MVar LocalProcessState+  , processState  :: StrictMVar LocalProcessState   , processThread :: ThreadId   , processNode   :: LocalNode   }@@ -190,10 +190,10 @@  -- | Local process state data LocalProcessState = LocalProcessState-  { _monitorCounter :: Int32-  , _spawnCounter   :: Int32-  , _channelCounter :: Int32-  , _typedChannels  :: Map LocalSendPortId TypedChannel +  { _monitorCounter :: !Int32+  , _spawnCounter   :: !Int32+  , _channelCounter :: !Int32+  , _typedChannels  :: !(Map LocalSendPortId TypedChannel)   }  -- | The Cloud Haskell 'Process' type@@ -214,9 +214,9 @@ -- to create a SendPort. data SendPortId = SendPortId {     -- | The ID of the process that will receive messages sent on this port-    sendPortProcessId :: ProcessId+    sendPortProcessId :: {-# UNPACK #-} !ProcessId     -- | Process-local ID of the channel-  , sendPortLocalId   :: LocalSendPortId+  , sendPortLocalId   :: {-# UNPACK #-} !LocalSendPortId   }   deriving (Eq, Ord) @@ -279,9 +279,9 @@ -- | MonitorRef is opaque for regular Cloud Haskell processes  data MonitorRef = MonitorRef    { -- | ID of the entity to be monitored-    monitorRefIdent   :: Identifier+    monitorRefIdent   :: !Identifier     -- | Unique to distinguish multiple monitor requests by the same process-  , monitorRefCounter :: Int32+  , monitorRefCounter :: !Int32   }   deriving (Eq, Ord, Show) @@ -384,6 +384,7 @@   | WhereIs String   | Register String (Maybe ProcessId) -- Nothing to unregister   | NamedSend String Message+  | Reconnect Identifier   deriving Show  --------------------------------------------------------------------------------@@ -428,6 +429,7 @@   put (WhereIs label)       = putWord8 6 >> put label   put (Register label pid)  = putWord8 7 >> put label >> put pid   put (NamedSend label msg) = putWord8 8 >> put label >> put (messageToPayload msg) +  put (Reconnect dest)      = putWord8 9 >> put dest   get = do     header <- getWord8     case header of@@ -440,6 +442,7 @@       6 -> WhereIs <$> get       7 -> Register <$> get <*> get       8 -> NamedSend <$> get <*> (payloadToMessage <$> get)+      9 -> Reconnect <$> get       _ -> fail "ProcessSignal.get: invalid"  instance Binary DiedReason where

@@ -25,23 +25,24 @@   , toList   , partitionWithKey   , filterWithKey+  , elems   )-import qualified Data.List as List (delete, (\\)) import Data.Set (Set)-import qualified Data.Set as Set (empty, insert, delete, member, (\\), fromList)+import qualified Data.Set as Set (empty, insert, delete, member, filter) import Data.Foldable (forM_) import Data.Maybe (isJust) import Data.Typeable (Typeable) import Control.Category ((>>>)) import Control.Applicative ((<$>))-import Control.Monad (void, when, forever)+import Control.Monad (void, when) import Control.Monad.IO.Class (MonadIO, liftIO)-import Control.Monad.State (MonadState, StateT, evalStateT, gets, modify)+import Control.Monad.State.Strict (MonadState, StateT, evalStateT, gets)+import qualified Control.Monad.State.Strict as StateT (get, put) import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask) import Control.Exception (throwIO, SomeException, Exception, throwTo) import qualified Control.Exception as Exception (catch) import Control.Concurrent (forkIO)-import Control.Concurrent.MVar +import Control.Distributed.Process.Internal.StrictMVar   ( newMVar    , withMVar   , modifyMVar@@ -64,12 +65,17 @@   , address   , closeEndPoint   , ConnectionId+  , Connection+  , close   ) import Data.Accessor (Accessor, accessor, (^.), (^=), (^:)) import qualified Data.Accessor.Container as DAC (mapDefault, mapMaybe) import System.Random (randomIO) import Control.Distributed.Static (RemoteTable, Closure)-import qualified Control.Distributed.Static as Static (unclosure)+import qualified Control.Distributed.Static as Static +  ( unclosure+  , initRemoteTable+  ) import Control.Distributed.Process.Internal.Types    ( NodeId(..)   , LocalProcessId(..)@@ -85,6 +91,7 @@   , localPidCounter   , localPidUnique   , localProcessWithId+  , localConnections   , MonitorRef(..)   , ProcessMonitorNotification(..)   , NodeMonitorNotification(..)@@ -115,10 +122,10 @@   ( sendBinary   , sendMessage   , sendPayload+  , reconnect   ) import Control.Distributed.Process.Internal.Primitives (expect, register, finally) import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)-import qualified Control.Distributed.Static as Static (initRemoteTable)  -------------------------------------------------------------------------------- -- Initialization                                                             --@@ -159,11 +166,16 @@   void . forkIO $ handleIncomingMessages node   return node +-- Like 'Control.Monad.forever' but sans space leak+{-# INLINE forever' #-}+forever' :: Monad m => m a -> m b+forever' a = let a' = a >> a' in a'+ -- | Start and register the service processes on a node  -- (for now, this is only the logger) startServiceProcesses :: LocalNode -> IO () startServiceProcesses node = do-  logger <- forkProcess node . forever $ do+  logger <- forkProcess node . forever' $ do     (time, pid, string) <- expect :: Process (String, ProcessId, String)     liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string    runProcess node $ register "logger" logger@@ -184,75 +196,86 @@  -- | Spawn a new process on a local node forkProcess :: LocalNode -> Process () -> IO ProcessId-forkProcess node proc = modifyMVar (localState node) $ \st -> do-  let lpid  = LocalProcessId { lpidCounter = st ^. localPidCounter-                             , lpidUnique  = st ^. localPidUnique-                             }-  let pid   = ProcessId { processNodeId  = localNodeId node -                        , processLocalId = lpid-                        }-  pst <- newMVar LocalProcessState { _monitorCounter = 0-                                   , _spawnCounter   = 0-                                   , _channelCounter = 0-                                   , _typedChannels  = Map.empty-                                   }-  queue <- newCQueue-  (_, lproc) <- fixIO $ \ ~(tid, _) -> do-    let lproc = LocalProcess { processQueue  = queue-                             , processId     = pid-                             , processState  = pst -                             , processThread = tid-                             , processNode   = node-                             }-    tid' <- forkIO $ do-      reason <- Exception.catch -        (runLocalProcess lproc proc >> return DiedNormal)-        (return . DiedException . (show :: SomeException -> String))-      -- [Unified: Table 4, rules termination and exiting]-      modifyMVar_ (localState node) $ -        return . (localProcessWithId lpid ^= Nothing)-      writeChan (localCtrlChan node) NCMsg -        { ctrlMsgSender = ProcessIdentifier pid -        , ctrlMsgSignal = Died (ProcessIdentifier pid) reason -        }-    return (tid', lproc)+forkProcess node proc = modifyMVar (localState node) startProcess+  where+    startProcess :: LocalNodeState -> IO (LocalNodeState, ProcessId)+    startProcess st = do+      let lpid  = LocalProcessId { lpidCounter = st ^. localPidCounter+                                 , lpidUnique  = st ^. localPidUnique+                                 }+      let pid   = ProcessId { processNodeId  = localNodeId node +                            , processLocalId = lpid+                            }+      pst <- newMVar LocalProcessState { _monitorCounter = 0+                                       , _spawnCounter   = 0+                                       , _channelCounter = 0+                                       , _typedChannels  = Map.empty+                                       }+      queue <- newCQueue+      (_, lproc) <- fixIO $ \ ~(tid, _) -> do+        let lproc = LocalProcess { processQueue  = queue+                                 , processId     = pid+                                 , processState  = pst +                                 , processThread = tid+                                 , processNode   = node+                                 }+        tid' <- forkIO $ do+          reason <- Exception.catch +            (runLocalProcess lproc proc >> return DiedNormal)+            (return . DiedException . (show :: SomeException -> String))+          -- [Unified: Table 4, rules termination and exiting]+          modifyMVar_ (localState node) (cleanupProcess pid)+          writeChan (localCtrlChan node) NCMsg +            { ctrlMsgSender = ProcessIdentifier pid +            , ctrlMsgSignal = Died (ProcessIdentifier pid) reason +            }+        return (tid', lproc) -  if lpidCounter lpid == maxBound-    then do-      newUnique <- randomIO-      return ( (localProcessWithId lpid ^= Just lproc)-             . (localPidCounter ^= 0)-             . (localPidUnique ^= newUnique)-             $ st-             , pid-             )-    else-      return ( (localProcessWithId lpid ^= Just lproc)-             . (localPidCounter ^: (+ 1))+      if lpidCounter lpid == maxBound+        then do+          newUnique <- randomIO+          return ( (localProcessWithId lpid ^= Just lproc)+                 . (localPidCounter ^= 0)+                 . (localPidUnique ^= newUnique)+                 $ st+                 , pid+                 )+        else+          return ( (localProcessWithId lpid ^= Just lproc)+                 . (localPidCounter ^: (+ 1))+                 $ st+                 , pid +                 )++    cleanupProcess :: ProcessId -> LocalNodeState -> IO LocalNodeState+    cleanupProcess pid st = do+      let pid' = ProcessIdentifier pid +      let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (st ^. localConnections)+      mapM_ NT.close (Map.elems affected)+      return $ (localProcessWithId (processLocalId pid) ^= Nothing)+             . (localConnections ^= unaffected)              $ st-             , pid -             )  handleIncomingMessages :: LocalNode -> IO ()-handleIncomingMessages node = go [] Map.empty Map.empty Set.empty+handleIncomingMessages node = go Set.empty Map.empty Map.empty Set.empty   where-    go :: [NT.ConnectionId] -- ^ Connections whose purpose we don't yet know +    go :: Set NT.ConnectionId -- ^ Connections whose purpose we don't yet know         -> Map NT.ConnectionId LocalProcess -- ^ Connections to local processes        -> Map NT.ConnectionId TypedChannel -- ^ Connections to typed channels        -> Set NT.ConnectionId              -- ^ Connections to our controller        -> IO () -    go uninitConns procs chans ctrls = do+    go !uninitConns !procs !chans !ctrls = do       event <- NT.receive endpoint       case event of         NT.ConnectionOpened cid _rel _theirAddr ->           -- TODO: Check if _rel is ReliableOrdered, and if not, treat as           -- (**) below.-          go (cid : uninitConns) procs chans ctrls +          go (Set.insert cid uninitConns) procs chans ctrls          NT.Received cid payload ->            case ( Map.lookup cid procs                 , Map.lookup cid chans                , cid `Set.member` ctrls-               , cid `elem` uninitConns+               , cid `Set.member` uninitConns                ) of             (Just proc, _, _, _) -> do               let msg = payloadToMessage payload@@ -272,7 +295,7 @@                   mProc <- withMVar state $ return . (^. localProcessWithId lpid)                    case mProc of                     Just proc -> -                      go (List.delete cid uninitConns) +                      go (Set.delete cid uninitConns)                           (Map.insert cid proc procs)                          chans                          ctrls@@ -284,7 +307,7 @@                       -- remote node as having died, and we should close                       -- incoming connections (this requires a Transport layer                       -- extension). (**)-                      go (List.delete cid uninitConns) procs chans ctrls+                      go (Set.delete cid uninitConns) procs chans ctrls                 SendPortIdentifier chId -> do                   let lcid = sendPortLocalId chId                       lpid = processLocalId (sendPortProcessId chId)@@ -294,20 +317,20 @@                       mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)                       case mChannel of                         Just channel ->-                          go (List.delete cid uninitConns)+                          go (Set.delete cid uninitConns)                              procs                              (Map.insert cid channel chans)                              ctrls                         Nothing ->                           -- Unknown typed channel                           -- TODO (**) above-                          go (List.delete cid uninitConns) procs chans ctrls+                          go (Set.delete cid uninitConns) procs chans ctrls                     Nothing ->                       -- Unknown process                       -- TODO (**) above-                      go (List.delete cid uninitConns) procs chans ctrls+                      go (Set.delete cid uninitConns) procs chans ctrls                 NodeIdentifier _ ->-                  go (List.delete cid uninitConns)+                  go (Set.delete cid uninitConns)                      procs                      chans                      (Set.insert cid ctrls)@@ -316,7 +339,7 @@               -- TODO (**) above                go uninitConns procs chans ctrls         NT.ConnectionClosed cid -> -          go (List.delete cid uninitConns) +          go (Set.delete cid uninitConns)               (Map.delete cid procs)              (Map.delete cid chans)              (Set.delete cid ctrls)@@ -327,10 +350,11 @@             { ctrlMsgSender = nid             , ctrlMsgSignal = Died nid DiedDisconnect             }-          go (uninitConns List.\\ cids)-             (Map.filterWithKey (\k _ -> k `notElem` cids) procs)-             (Map.filterWithKey (\k _ -> k `notElem` cids) chans)-             (ctrls Set.\\ Set.fromList cids)+          let notRemoved k = k `notElem` cids+          go (Set.filter notRemoved uninitConns)+             (Map.filterWithKey (const . notRemoved) procs)+             (Map.filterWithKey (const . notRemoved) chans)+             (Set.filter notRemoved ctrls)         NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost Nothing _) _) ->           -- TODO: We should treat an asymetrical connection loss (incoming           -- connection broken, but outgoing connection still potentially ok)@@ -365,11 +389,11 @@  data NCState = NCState    {  -- Mapping from remote processes to linked local processes -    _links    :: Map Identifier (Set ProcessId) +    _links    :: !(Map Identifier (Set ProcessId))      -- Mapping from remote processes to monitoring local processes-  , _monitors :: Map Identifier (Set (ProcessId, MonitorRef))+  , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef)))      -- Process registry-  , _registry :: Map String ProcessId+  , _registry :: !(Map String ProcessId)   }  newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }@@ -389,7 +413,7 @@ nodeController :: NC () nodeController = do   node <- ask -  forever $ do+  forever' $ do     msg  <- liftIO $ readChan (localCtrlChan node)      -- [Unified: Table 7, rule nc_forward] @@ -421,6 +445,8 @@         ncEffectWhereIs from label       NCMsg from (NamedSend label msg') ->         ncEffectNamedSend from label msg'+      NCMsg from (Reconnect to) ->+        ncEffectReconnect from to       unexpected ->         error $ "nodeController: unexpected message " ++ show unexpected @@ -438,8 +464,8 @@   case (shouldLink, isLocal node (ProcessIdentifier from)) of     (True, _) ->  -- [Unified: first rule]       case mRef of-        Just ref -> modify $ monitorsFor them ^: Set.insert (from, ref)-        Nothing  -> modify $ linksFor them ^: Set.insert from +        Just ref -> modify' $ monitorsFor them ^: Set.insert (from, ref)+        Nothing  -> modify' $ linksFor them ^: Set.insert from      (False, True) -> -- [Unified: second rule]       notifyDied from them DiedUnknownId mRef      (False, False) -> -- [Unified: third rule]@@ -466,7 +492,7 @@         postAsMessage from $ DidUnlinkNode nid       SendPortIdentifier cid ->          postAsMessage from $ DidUnlinkPort cid -  modify $ linksFor them ^: Set.delete from+  modify' $ linksFor them ^: Set.delete from  -- [Unified: Table 11] ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()@@ -474,7 +500,7 @@   node <- ask    when (isLocal node (ProcessIdentifier from)) $      postAsMessage from $ DidUnmonitor ref-  modify $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)+  modify' $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)  -- [Unified: Table 12] ncEffectDied :: Identifier -> DiedReason -> NC ()@@ -495,7 +521,7 @@       when (localOnly <= isLocal node (ProcessIdentifier us)) $         notifyDied us them reason (Just ref) -  modify $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)+  modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)  -- [Unified: Table 13] ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()@@ -517,7 +543,7 @@ -- but mentions it's "very similar to nsend" (Table 14) ncEffectRegister :: String -> Maybe ProcessId -> NC () ncEffectRegister label mPid = -  modify $ registryFor label ^= mPid+  modify' $ registryFor label ^= mPid   -- An acknowledgement is not necessary. If we want a synchronous register,   -- it suffices to send a whereis requiry immediately after the register   -- (that may not suffice if we do decide for unreliable messaging instead)@@ -544,6 +570,12 @@                          (ProcessIdentifier pid)                           (messageToPayload msg)  +-- Reconnecting+ncEffectReconnect :: Identifier -> Identifier -> NC ()+ncEffectReconnect from to = do+  node <- ask+  liftIO $ reconnect node from to+ -------------------------------------------------------------------------------- -- Auxiliary                                                                  -- --------------------------------------------------------------------------------@@ -589,6 +621,7 @@ destNid (Register _ _)  = Nothing destNid (WhereIs _)     = Nothing destNid (NamedSend _ _) = Nothing+destNid (Reconnect _)   = Nothing -- We don't need to forward 'Died' signals; if monitoring/linking is setup, -- then when a local process dies the monitoring/linking machinery will take -- care of notifying remote nodes@@ -691,7 +724,7 @@ splitNotif :: Identifier            -> Map Identifier a            -> (Map Identifier a, Map Identifier a)-splitNotif ident = Map.partitionWithKey (const . impliesDeathOf ident)+splitNotif ident = Map.partitionWithKey (\k !v -> impliesDeathOf ident k)   -- | Does the death of one entity (node, project, channel) imply the death -- of another?@@ -712,3 +745,11 @@   cid' == cid _ `impliesDeathOf` _ =   False++--------------------------------------------------------------------------------+-- Strict evaluation of the state                                             --+--------------------------------------------------------------------------------++-- | Modify and evaluate the state+modify' :: MonadState s m => (s -> s) -> m ()+modify' f = StateT.get >>= \s -> StateT.put $! f s

@@ -1,3 +1,4 @@+{-# LANGUAGE TemplateHaskell #-} module Main where  import Data.ByteString.Lazy (empty)@@ -290,7 +291,7 @@   node <- newLocalNode transport rtable   done <- newEmptyMVar   forkProcess node $ do-    (pid, ref) <- spawnMonitor (localNodeId node) (Closure (staticLabel "ThisDoesNotExist") empty)+    (pid, ref) <- spawnMonitor (localNodeId node) (closure (staticLabel "ThisDoesNotExist") empty)     ProcessMonitorNotification ref' pid' _reason <- expect      -- Depending on the exact interleaving, reason might be NoProc or the exception thrown by the absence of the static closure     True <- return $ ref' == ref && pid == pid'