distributed-process 0.3.1 → 0.4.0

raw patch · 20 files changed

+1306/−432 lines, 20 filesdep +networkdep ~network-transportdep ~network-transport-tcpnew-component:exe:distributed-process-channelsnew-component:exe:distributed-process-latencynew-component:exe:distributed-process-spawnsnew-component:exe:distributed-process-throughputPVP ok

version bump matches the API change (PVP)

Dependencies added: network

Dependency ranges changed: network-transport, network-transport-tcp

API changes (from Hackage documentation)

Files

• benchmarks/Channels.hs +41/−0
• benchmarks/Latency.hs +36/−0
• benchmarks/Spawns.hs +45/−0
• benchmarks/Throughput.hs +70/−0
• distributed-process.cabal +71/−10
• src/Control/Distributed/Process.hs +21/−15
• src/Control/Distributed/Process/Closure.hs +2/−0
• src/Control/Distributed/Process/Internal/CQueue.hs +40/−15
• src/Control/Distributed/Process/Internal/Closure/TH.hs +83/−47
• src/Control/Distributed/Process/Internal/Messaging.hs +169/−0
• src/Control/Distributed/Process/Internal/Node.hs +0/−103
• src/Control/Distributed/Process/Internal/Primitives.hs +122/−58
• src/Control/Distributed/Process/Internal/StrictContainerAccessors.hs +21/−0
• src/Control/Distributed/Process/Internal/StrictList.hs +25/−0
• src/Control/Distributed/Process/Internal/StrictMVar.hs +11/−1
• src/Control/Distributed/Process/Internal/Types.hs +83/−50
• src/Control/Distributed/Process/Internal/WeakTQueue.hs +101/−0
• src/Control/Distributed/Process/Node.hs +145/−122
• tests/TestCH.hs +127/−7
• tests/TestClosure.hs +93/−4

@@ -0,0 +1,41 @@+-- | Like Latency, but creating lots of channels+import System.Environment+import Control.Monad+import Control.Applicative+import Control.Distributed.Process+import Control.Distributed.Process.Node+import Network.Transport.TCP (createTransport, defaultTCPParameters)+import Data.Binary (encode, decode)+import qualified Data.ByteString.Lazy as BSL++pingServer :: Process ()+pingServer = forever $ do+  them <- expect+  sendChan them ()+  -- TODO: should this be automatic?+  reconnectPort them++pingClient :: Int -> ProcessId -> Process ()+pingClient n them = do+  replicateM_ n $ do+    (sc, rc) <- newChan :: Process (SendPort (), ReceivePort ())+    send them sc+    receiveChan rc+  liftIO . putStrLn $ "Did " ++ show n ++ " pings"++initialProcess :: String -> Process () +initialProcess "SERVER" = do+  us <- getSelfPid+  liftIO $ BSL.writeFile "pingServer.pid" (encode us)+  pingServer +initialProcess "CLIENT" = do+  n <- liftIO $ getLine+  them <- liftIO $ decode <$> BSL.readFile "pingServer.pid"+  pingClient (read n) them++main :: IO ()+main = do+  [role, host, port] <- getArgs+  Right transport <- createTransport host port defaultTCPParameters+  node <- newLocalNode transport initRemoteTable +  runProcess node $ initialProcess role 

@@ -0,0 +1,36 @@+import System.Environment+import Control.Monad+import Control.Applicative+import Control.Distributed.Process+import Control.Distributed.Process.Node+import Network.Transport.TCP (createTransport, defaultTCPParameters)+import Data.Binary (encode, decode)+import qualified Data.ByteString.Lazy as BSL++pingServer :: Process ()+pingServer = forever $ do+  them <- expect+  send them ()++pingClient :: Int -> ProcessId -> Process ()+pingClient n them = do+  us <- getSelfPid+  replicateM_ n $ send them us >> (expect :: Process ())+  liftIO . putStrLn $ "Did " ++ show n ++ " pings"++initialProcess :: String -> Process () +initialProcess "SERVER" = do+  us <- getSelfPid+  liftIO $ BSL.writeFile "pingServer.pid" (encode us)+  pingServer +initialProcess "CLIENT" = do+  n <- liftIO $ getLine+  them <- liftIO $ decode <$> BSL.readFile "pingServer.pid"+  pingClient (read n) them++main :: IO ()+main = do+  [role, host, port] <- getArgs+  Right transport <- createTransport host port defaultTCPParameters+  node <- newLocalNode transport initRemoteTable +  runProcess node $ initialProcess role 

@@ -0,0 +1,45 @@+-- | Like Throughput, but send every ping from a different process+-- (i.e., require a lightweight connection per ping)+import System.Environment+import Control.Monad+import Control.Applicative+import Control.Distributed.Process+import Control.Distributed.Process.Node+import Network.Transport.TCP (createTransport, defaultTCPParameters)+import Data.Binary (encode, decode)+import qualified Data.ByteString.Lazy as BSL++counter :: Process ()+counter = go 0+  where+    go :: Int -> Process () +    go !n = do+      b <- expect+      case b of+        Nothing   -> go (n + 1)+        Just them -> send them n >> go 0++count :: Int -> ProcessId -> Process ()+count n them = do+  us <- getSelfPid+  replicateM_ n . spawnLocal $ send them (Nothing :: Maybe ProcessId)+  send them (Just us)+  n' <- expect+  liftIO $ print (n == n')++initialProcess :: String -> Process () +initialProcess "SERVER" = do+  us <- getSelfPid+  liftIO $ BSL.writeFile "counter.pid" (encode us)+  counter+initialProcess "CLIENT" = do+  n <- liftIO $ getLine+  them <- liftIO $ decode <$> BSL.readFile "counter.pid"+  count (read n) them++main :: IO ()+main = do+  [role, host, port] <- getArgs+  Right transport <- createTransport host port defaultTCPParameters+  node <- newLocalNode transport initRemoteTable +  runProcess node $ initialProcess role 

@@ -0,0 +1,70 @@+{-# LANGUAGE DeriveDataTypeable #-}+import System.Environment+import Control.Monad+import Control.Applicative+import Control.Distributed.Process+import Control.Distributed.Process.Node+import Network.Transport.TCP (createTransport, defaultTCPParameters)+import Data.Binary+import qualified Data.ByteString.Lazy as BSL+import Data.Typeable++data SizedList a = SizedList { size :: Int , elems :: [a] }+  deriving (Typeable)++instance Binary a => Binary (SizedList a) where+  put (SizedList sz xs) = put sz >> mapM_ put xs+  get = do +    sz <- get +    xs <- getMany sz+    return (SizedList sz xs)++-- Copied from Data.Binary+getMany :: Binary a => Int -> Get [a]+getMany = go []+ where+    go xs 0 = return $! reverse xs+    go xs i = do x <- get+                 x `seq` go (x:xs) (i-1)+{-# INLINE getMany #-}++nats :: Int -> SizedList Int+nats = \n -> SizedList n (aux n)+  where+    aux 0 = []+    aux n = n : aux (n - 1)++counter :: Process ()+counter = go 0+  where+    go :: Int -> Process () +    go !n =+      receiveWait +        [ match $ \xs   -> go (n + size (xs :: SizedList Int))+        , match $ \them -> send them n >> go 0+        ]++count :: (Int, Int) -> ProcessId -> Process ()+count (packets, sz) them = do+  us <- getSelfPid+  replicateM_ packets $ send them (nats sz)+  send them us+  n' <- expect+  liftIO $ print (packets * sz, n' == packets * sz)++initialProcess :: String -> Process () +initialProcess "SERVER" = do+  us <- getSelfPid+  liftIO $ BSL.writeFile "counter.pid" (encode us)+  counter+initialProcess "CLIENT" = do+  n <- liftIO getLine+  them <- liftIO $ decode <$> BSL.readFile "counter.pid"+  count (read n) them++main :: IO ()+main = do+  [role, host, port] <- getArgs+  Right transport <- createTransport host port defaultTCPParameters+  node <- newLocalNode transport initRemoteTable +  runProcess node $ initialProcess role 

@@ -1,5 +1,5 @@ Name:          distributed-process -Version:       0.3.1+Version:       0.4.0 Cabal-Version: >=1.8 Build-Type:    Simple License:       BSD3 @@ -33,10 +33,14 @@   description: Build with Template Haskell support   default: True +flag benchmarks+  description: Build benchmarks+  default: False+ Library   Build-Depends:     base >= 4.4 && < 5,                      binary >= 0.5 && < 0.6,-                     network-transport >= 0.2 && < 0.3,+                     network-transport >= 0.3 && < 0.4,                      stm >= 2.3 && < 2.5,                      transformers >= 0.2 && < 0.4,                      mtl >= 2.0 && < 2.2,@@ -58,8 +62,11 @@                      Control.Distributed.Process.Internal.CQueue,                      Control.Distributed.Process.Internal.Types,                      Control.Distributed.Process.Internal.Closure.BuiltIn,-                     Control.Distributed.Process.Internal.Node-                     Control.Distributed.Process.Internal.StrictMVar+                     Control.Distributed.Process.Internal.Messaging,+                     Control.Distributed.Process.Internal.StrictList,+                     Control.Distributed.Process.Internal.StrictMVar,+                     Control.Distributed.Process.Internal.WeakTQueue+                     Control.Distributed.Process.Internal.StrictContainerAccessors   Extensions:        RankNTypes,                      ScopedTypeVariables,                      FlexibleInstances,@@ -84,9 +91,10 @@                      random >= 1.0 && < 1.1,                      ansi-terminal >= 0.5 && < 0.6,                      distributed-process >= 0.3 && < 0.4,-                     network-transport >= 0.2 && < 0.3,-                     network-transport-tcp >= 0.2 && < 0.3,-                     binary >= 0.5 && < 0.6+                     network-transport >= 0.3 && < 0.4,+                     network-transport-tcp >= 0.3 && < 0.4,+                     binary >= 0.5 && < 0.6,+                     network >= 2.3 && < 2.5   Other-modules:     TestAuxiliary   Extensions:        CPP,                      ScopedTypeVariables,@@ -103,11 +111,64 @@                      ansi-terminal >= 0.5 && < 0.6,                      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+                     network-transport >= 0.3 && < 0.4,+                     network-transport-tcp >= 0.3 && < 0.4,+                     bytestring >= 0.9 && < 0.11,+                     network >= 2.3 && < 2.5   Other-modules:     TestAuxiliary   Extensions:        CPP,                       ScopedTypeVariables   ghc-options:       -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind    HS-Source-Dirs:    tests ++Executable distributed-process-throughput +  if flag(benchmarks)+    Build-Depends:   base >= 4.4 && < 5,+                     distributed-process >= 0.3 && < 0.4,+                     network-transport-tcp >= 0.3 && < 0.4,+                     bytestring >= 0.9 && < 0.11,+                     binary >= 0.5 && < 0.6+  else+    buildable: False+  Main-Is:           benchmarks/Throughput.hs+  ghc-options:       -Wall+  Extensions:        BangPatterns++Executable distributed-process-latency+  if flag(benchmarks)+    Build-Depends:   base >= 4.4 && < 5,+                     distributed-process >= 0.3 && < 0.4,+                     network-transport-tcp >= 0.3 && < 0.4,+                     bytestring >= 0.9 && < 0.11,+                     binary >= 0.5 && < 0.6+  else+    buildable: False+  Main-Is:           benchmarks/Latency.hs+  ghc-options:       -Wall+  Extensions:        BangPatterns++Executable distributed-process-channels+  if flag(benchmarks)+    Build-Depends:   base >= 4.4 && < 5,+                     distributed-process >= 0.3 && < 0.4,+                     network-transport-tcp >= 0.3 && < 0.4,+                     bytestring >= 0.9 && < 0.11,+                     binary >= 0.5 && < 0.6+  else+    buildable: False+  Main-Is:           benchmarks/Channels.hs+  ghc-options:       -Wall+  Extensions:        BangPatterns++Executable distributed-process-spawns+  if flag(benchmarks)+    Build-Depends:   base >= 4.4 && < 5,+                     distributed-process >= 0.3 && < 0.4,+                     network-transport-tcp >= 0.3 && < 0.4,+                     bytestring >= 0.9 && < 0.11,+                     binary >= 0.5 && < 0.6+  else+    buildable: False+  Main-Is:           benchmarks/Spawns.hs+  ghc-options:       -Wall+  Extensions:        BangPatterns

@@ -35,6 +35,8 @@   , match   , matchIf   , matchUnknown+  , AbstractMessage(..) +  , matchAny     -- * Process management   , spawn   , call@@ -66,6 +68,7 @@   , Closure   , closure   , Static+  , unStatic   , unClosure   , RemoteTable     -- * Logging@@ -77,7 +80,6 @@   , nsend   , registerRemote   , unregisterRemote-  , whereisRemote   , whereisRemoteAsync   , nsendRemote   , WhereIsReply(..)@@ -101,7 +103,6 @@   , spawnChannelLocal     -- * Reconnecting   , reconnect-  , reconnectNode   , reconnectPort   ) where @@ -113,11 +114,14 @@ import Control.Monad.IO.Class (liftIO) import Control.Applicative ((<$>)) import Control.Monad.Reader (ask)+import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar) import Control.Distributed.Static    ( Closure   , closure   , Static   , RemoteTable+  , closureCompose+  , staticClosure   ) import Control.Distributed.Process.Internal.Types    ( NodeId(..)@@ -138,8 +142,9 @@   , SendPortId(..)   , WhereIsReply(..)   , LocalProcess(processNode)+  , nullProcessId   )-import Control.Distributed.Process.Serializable (SerializableDict)+import Control.Distributed.Process.Serializable (Serializable, SerializableDict) import Control.Distributed.Process.Internal.Closure.BuiltIn   ( sdictUnit    , sdictSendPort@@ -154,7 +159,6 @@   , cpSend   , cpNewChan   )-import Control.Distributed.Static (closureCompose, staticClosure) import Control.Distributed.Process.Internal.Primitives   ( -- Basic messaging     send @@ -172,6 +176,8 @@   , match   , matchIf   , matchUnknown+  , AbstractMessage(..) +  , matchAny     -- Process management   , terminate   , ProcessTerminationException(..)@@ -197,10 +203,10 @@   , nsend   , registerRemote   , unregisterRemote-  , whereisRemote   , whereisRemoteAsync   , nsendRemote     -- Closures+  , unStatic   , unClosure     -- Exception handling   , catch@@ -214,10 +220,8 @@   , spawnAsync     -- Reconnecting   , reconnect-  , reconnectNode   , reconnectPort   )-import Control.Distributed.Process.Serializable (Serializable) import Control.Distributed.Process.Node (forkProcess)  -- INTERNAL NOTES@@ -293,10 +297,6 @@ spawn :: NodeId -> Closure (Process ()) -> Process ProcessId spawn nid proc = do   us   <- getSelfPid-  -- Since we throw an exception when the remote node dies, we could use-  -- linkNode instead. However, we don't have a way of "scoped" linking so if-  -- we call linkNode here, and unlinkNode after, then we might remove a link-  -- that was already set up   mRef <- monitorNode nid   sRef <- spawnAsync nid $ cpLink us                     `seqCP` cpExpect sdictUnit @@ -310,7 +310,7 @@     ]   unmonitor mRef   case mPid of-    Left err  -> fail $ "spawn: remote node failed: " ++ show err+    Left _err -> return (nullProcessId nid)     Right pid -> send pid () >> return pid  -- | Spawn a process and link to it@@ -411,6 +411,12 @@                   -> Process (SendPort a) spawnChannelLocal proc = do    node <- processNode <$> ask -  (sport, rport) <- newChan-  _ <- liftIO $ forkProcess node (proc rport)-  return sport+  liftIO $ do+    mvar <- newEmptyMVar+    forkProcess node $ do+      -- It is important that we allocate the new channel in the new process,+      -- because otherwise it will be associated with the wrong process ID+      (sport, rport) <- newChan+      liftIO $ putMVar mvar sport+      proc rport+    takeMVar mvar 

@@ -179,6 +179,7 @@ #ifdef TemplateHaskellSupport      -- * Template Haskell support for creating static values and closures    , remotable+  , remotableDecl   , mkStatic   , mkClosure   , functionSDict@@ -210,6 +211,7 @@ #ifdef TemplateHaskellSupport  import Control.Distributed.Process.Internal.Closure.TH    ( remotable+  , remotableDecl   , mkStatic   , functionSDict   , functionTDict

@@ -1,13 +1,15 @@ -- | Concurrent queue for single reader, single writer+{-# LANGUAGE MagicHash, UnboxedTuples #-} module Control.Distributed.Process.Internal.CQueue    ( CQueue   , BlockSpec(..)   , newCQueue   , enqueue   , dequeue+  , mkWeakCQueue   ) where -import Control.Concurrent.MVar (MVar, newMVar, takeMVar, putMVar)+import Prelude hiding (length, reverse) import Control.Concurrent.STM    ( atomically   , TChan@@ -19,16 +21,34 @@ import Control.Applicative ((<$>), (<*>)) import Control.Exception (mask, onException) import System.Timeout (timeout)+import Control.Distributed.Process.Internal.StrictMVar +  ( StrictMVar(StrictMVar)+  , newMVar+  , takeMVar+  , putMVar+  )+import Control.Distributed.Process.Internal.StrictList+  ( StrictList(..)+  , reverse+  , reverse'+  )+import GHC.MVar (MVar(MVar))+import GHC.IO (IO(IO)) +import GHC.Prim (mkWeak#)+import GHC.Weak (Weak(Weak))  -- We use a TCHan rather than a Chan so that we have a non-blocking read-data CQueue a = CQueue (MVar [a]) -- Arrived-                       (TChan a)  -- Incoming+data CQueue a = CQueue (StrictMVar (StrictList a)) -- Arrived+                       (TChan a)                   -- Incoming  newCQueue :: IO (CQueue a)-newCQueue = CQueue <$> newMVar [] <*> atomically newTChan+newCQueue = CQueue <$> newMVar Nil <*> atomically newTChan +-- | Enqueue an element+--+-- Enqueue is strict. enqueue :: CQueue a -> a -> IO ()-enqueue (CQueue _arrived incoming) a = atomically $ writeTChan incoming a +enqueue (CQueue _arrived incoming) !a = atomically $ writeTChan incoming a   data BlockSpec =      NonBlocking@@ -51,7 +71,7 @@       arr <- takeMVar arrived        -- We first check the arrived messages. If we get interrupted during this       -- search, we just put the MVar back (we haven't read from the Chan yet)-      (arr', mb) <- onException (restore (checkArrived [] arr))+      (arr', mb) <- onException (restore (checkArrived Nil arr))                                 (putMVar arrived arr)        case (mb, blockSpec) of         (Just b, _) -> do @@ -65,15 +85,15 @@           timeout n $ checkBlocking arr'      -- We reverse the accumulator on return only if we find a match-    checkArrived :: [a] -> [a] -> IO ([a], Maybe b)-    checkArrived acc []     = return (acc, Nothing)-    checkArrived acc (x:xs) = +    checkArrived :: StrictList a -> StrictList a -> IO (StrictList a, Maybe b)+    checkArrived acc Nil = return (acc, Nothing)+    checkArrived acc (Cons x xs) =        case check x of-        Just y  -> return (reverse acc ++ xs, Just y)-        Nothing -> checkArrived (x:acc) xs+        Just y  -> return (reverse' acc xs, Just y)+        Nothing -> checkArrived (Cons x acc) xs      -- If we call checkBlocking there may or may not be a timeout-    checkBlocking :: [a] -> IO b+    checkBlocking :: StrictList a -> IO b     checkBlocking acc = do       -- readTChan is a blocking call, and hence is interruptable. If it is        -- interrupted, we put the value of the accumulator in 'arrived'  @@ -82,18 +102,18 @@       x <- onException (atomically $ readTChan incoming)                        (putMVar arrived $ reverse acc)       case check x of-        Nothing -> checkBlocking (x:acc)+        Nothing -> checkBlocking (Cons x acc)         Just y  -> putMVar arrived (reverse acc) >> return y       -- checkNonBlocking is only called if there is no timeout-    checkNonBlocking :: [a] -> IO (Maybe b)+    checkNonBlocking :: StrictList a -> IO (Maybe b)     checkNonBlocking acc = do       -- tryReadTChan is *not* interruptible       mx <- atomically $ tryReadTChan incoming       case mx of         Nothing -> putMVar arrived (reverse acc) >> return Nothing         Just x  -> case check x of-          Nothing -> checkNonBlocking (x:acc)+          Nothing -> checkNonBlocking (Cons x acc)           Just y  -> putMVar arrived (reverse acc) >> return (Just y)              check :: a -> Maybe b@@ -103,3 +123,8 @@     checkMatches []     _ = Nothing     checkMatches (m:ms) a = case m a of Nothing -> checkMatches ms a                                         Just b  -> Just b++-- | Weak reference to a CQueue+mkWeakCQueue :: CQueue a -> IO () -> IO (Weak (CQueue a))+mkWeakCQueue m@(CQueue (StrictMVar (MVar m#)) _) f = IO $ \s ->+  case mkWeak# m# m f s of (# s1, w #) -> (# s1, Weak w #)

@@ -3,6 +3,7 @@ module Control.Distributed.Process.Internal.Closure.TH    ( -- * User-level API     remotable+  , remotableDecl   , mkStatic   , functionSDict   , functionTDict@@ -20,7 +21,7 @@   , mkName   , nameBase     -- Algebraic data types-  , Dec+  , Dec(SigD)   , Exp   , Type(AppT, ForallT, VarT, ArrowT)   , Info(VarI)@@ -39,6 +40,7 @@   , funD   , sigD   )+import Data.Maybe (catMaybes)   import Data.Binary (encode) import Data.Generics (everywhereM, mkM, gmapM) import Data.Rank1Dynamic (toDynamic)@@ -69,10 +71,47 @@ -- of functions remotable :: [Name] -> Q [Dec]  remotable ns = do-  (closures, inserts) <- unzip <$> mapM generateDefs ns-  rtable <- createMetaData (concat inserts)-  return $ concat closures ++ rtable +    types <- mapM getType ns +    (closures, inserts) <- unzip <$> mapM generateDefs types +    rtable <- createMetaData (mkName "__remoteTable") (concat inserts)+    return $ concat closures ++ rtable  +-- | Like 'remotable', but parameterized by the declaration of a function+-- instead of the function name. So where for 'remotable' you'd do+--+-- > f :: T1 -> T2+-- > f = ...+-- >+-- > remotable ['f]+--+-- with 'remotableDecl' you would instead do+--+-- > remotableDecl [+-- >    [d| f :: T1 -> T2 ;+-- >        f = ...+-- >      |]+-- >  ]+--+-- 'remotableDecl' creates the function specified as well as the various+-- dictionaries and static versions that 'remotable' also creates.+-- 'remotableDecl' is sometimes necessary when you want to refer to, say,+-- @$(mkClosure 'f)@ within the definition of @f@ itself.+--+-- NOTE: 'remotableDecl' creates @__remoteTableDecl@ instead of @__remoteTable@+-- so that you can use both 'remotable' and 'remotableDecl' within the same+-- module.+remotableDecl :: [Q [Dec]] -> Q [Dec]+remotableDecl qDecs = do+    decs <- concat <$> sequence qDecs+    let types = catMaybes (map typeOf decs)+    (closures, inserts) <- unzip <$> mapM generateDefs types +    rtable <- createMetaData (mkName "__remoteTableDecl") (concat inserts)+    return $ decs ++ concat closures ++ rtable +  where+    typeOf :: Dec -> Maybe (Name, Type)+    typeOf (SigD name typ) = Just (name, typ)+    typeOf _               = Nothing+ -- | Construct a static value. -- -- If @f : forall a1 .. an. T@ @@ -95,6 +134,10 @@ functionTDict :: Name -> Q Exp functionTDict = varE . tdictName +-- | If @f : T1 -> T2@ then @$(mkClosure 'f) :: T1 -> Closure T2@. +--+-- TODO: The current version of mkClosure is too polymorphic +-- (@forall a. Binary a => a -> Closure T2). mkClosure :: Name -> Q Exp mkClosure n =    [|   closure ($(mkStatic n) `staticCompose` staticDecode $(functionSDict n)) @@ -106,49 +149,44 @@ --------------------------------------------------------------------------------  -- | Generate the code to add the metadata to the CH runtime-createMetaData :: [Q Exp] -> Q [Dec]-createMetaData is = -  [d| __remoteTable :: RemoteTable -> RemoteTable ;-      __remoteTable = $(compose is)-    |]+createMetaData :: Name -> [Q Exp] -> Q [Dec]+createMetaData name is = +  sequence [ sigD name [t| RemoteTable -> RemoteTable |] +           , sfnD name (compose is)+           ] -generateDefs :: Name -> Q ([Dec], [Q Exp])-generateDefs n = do+generateDefs :: (Name, Type) -> Q ([Dec], [Q Exp])+generateDefs (origName, fullType) = do     proc <- [t| Process |]-    mType <- getType n-    case mType of-      Just (origName, typ) -> do-        let (typVars, typ') = case typ of ForallT vars [] mono -> (vars, mono)-                                          _                    -> ([], typ)+    let (typVars, typ') = case fullType of ForallT vars [] mono -> (vars, mono)+                                           _                    -> ([], fullType) -        -- The main "static" entry                                  -        (static, register) <- makeStatic origName typVars typ' -         -        -- If n :: T1 -> T2, static serializable dictionary for T1 -        -- TODO: we should check if arg is an instance of Serializable, but we cannot-        -- http://hackage.haskell.org/trac/ghc/ticket/7066-        (sdict, registerSDict) <- case (typVars, typ') of-          ([], ArrowT `AppT` arg `AppT` _res) -> -            makeDict (sdictName origName) arg-          _ -> -            return ([], [])-        -        -- If n :: T1 -> Process T2, static serializable dictionary for T2-        -- TODO: check if T2 is serializable (same as above)-        (tdict, registerTDict) <- case (typVars, typ') of-          ([], ArrowT `AppT` _arg `AppT` (proc' `AppT` res)) | proc' == proc -> -            makeDict (tdictName origName) res -          _ ->-            return ([], [])-        -        return ( concat [static, sdict, tdict]-               , concat [register, registerSDict, registerTDict]-               )+    -- The main "static" entry                                  +    (static, register) <- makeStatic typVars typ' +     +    -- If n :: T1 -> T2, static serializable dictionary for T1 +    -- TODO: we should check if arg is an instance of Serializable, but we cannot+    -- http://hackage.haskell.org/trac/ghc/ticket/7066+    (sdict, registerSDict) <- case (typVars, typ') of+      ([], ArrowT `AppT` arg `AppT` _res) -> +        makeDict (sdictName origName) arg       _ -> -        fail $ "remotable: " ++ show n ++ " not found"+        return ([], [])+    +    -- If n :: T1 -> Process T2, static serializable dictionary for T2+    -- TODO: check if T2 is serializable (same as above)+    (tdict, registerTDict) <- case (typVars, typ') of+      ([], ArrowT `AppT` _arg `AppT` (proc' `AppT` res)) | proc' == proc -> +        makeDict (tdictName origName) res +      _ ->+        return ([], [])+    +    return ( concat [static, sdict, tdict]+           , concat [register, registerSDict, registerTDict]+           )   where-    makeStatic :: Name -> [TyVarBndr] -> Type -> Q ([Dec], [Q Exp])-    makeStatic origName typVars typ = do +    makeStatic :: [TyVarBndr] -> Type -> Q ([Dec], [Q Exp])+    makeStatic typVars typ = do        static <- generateStatic origName typVars typ       let dyn = case typVars of                    [] -> [| toDynamic $(varE origName) |]@@ -190,8 +228,6 @@         Just t  -> t     applySubst s t = gmapM (mkM (applySubst s)) t -    - -- | Generate a static value  generateStatic :: Name -> [TyVarBndr] -> Type -> Q [Dec] generateStatic n xs typ = do@@ -240,12 +276,12 @@ stringE = litE . stringL  -- | Look up the "original name" (module:name) and type of a top-level function-getType :: Name -> Q (Maybe (Name, Type))+getType :: Name -> Q (Name, Type) getType name = do    info <- reify name   case info of -    VarI origName typ _ _ -> return $ Just (origName, typ)-    _                     -> return Nothing+    VarI origName typ _ _ -> return (origName, typ)+    _                     -> fail $ show name ++ " not found"  -- | Variation on 'funD' which takes a single expression to define the function sfnD :: Name -> Q Exp -> Q Dec

@@ -0,0 +1,169 @@+module Control.Distributed.Process.Internal.Messaging+  ( sendPayload+  , sendBinary+  , sendMessage+  , disconnect +  , closeImplicitReconnections +  , impliesDeathOf+  ) where++import Data.Accessor ((^.), (^=))+import Data.Binary (Binary, encode)+import qualified Data.Map as Map (partitionWithKey, elems)+import qualified Data.ByteString.Lazy as BSL (toChunks)+import qualified Data.ByteString as BSS (ByteString)+import Control.Distributed.Process.Internal.StrictMVar (withMVar, modifyMVar_)+import Control.Concurrent.Chan (writeChan)+import Control.Monad (unless)+import qualified Network.Transport as NT +  ( Connection+  , send+  , defaultConnectHints+  , connect+  , Reliability(ReliableOrdered)+  , close+  )+import Control.Distributed.Process.Internal.Types +  ( LocalNode(localState, localEndPoint, localCtrlChan)+  , Identifier+  , localConnections+  , localConnectionBetween+  , nodeAddress+  , nodeOf+  , messageToPayload+  , createMessage+  , NCMsg(..)+  , ProcessSignal(Died)+  , DiedReason(DiedDisconnect)+  , ImplicitReconnect(WithImplicitReconnect)+  , NodeId+  , ProcessId(processNodeId)+  , SendPortId(sendPortProcessId)+  , Identifier(NodeIdentifier, ProcessIdentifier, SendPortIdentifier)+  )+import Control.Distributed.Process.Serializable (Serializable)++--------------------------------------------------------------------------------+-- Message sending                                                            -- +--------------------------------------------------------------------------------++sendPayload :: LocalNode +            -> Identifier+            -> Identifier+            -> ImplicitReconnect+            -> [BSS.ByteString] +            -> IO ()+sendPayload node from to implicitReconnect payload = do+  mConn <- connBetween node from to implicitReconnect+  didSend <- case mConn of+    Just conn -> do+      didSend <- NT.send conn payload+      case didSend of+        Left _err -> return False +        Right ()  -> return True +    Nothing -> return False+  unless didSend $ do+    writeChan (localCtrlChan node) NCMsg+      { ctrlMsgSender = to +      , ctrlMsgSignal = Died to DiedDisconnect+      }++sendBinary :: Binary a +           => LocalNode +           -> Identifier +           -> Identifier +           -> ImplicitReconnect+           -> a +           -> IO ()+sendBinary node from to implicitReconnect +  = sendPayload node from to implicitReconnect . BSL.toChunks . encode++sendMessage :: Serializable a +            => LocalNode +            -> Identifier +            -> Identifier +            -> ImplicitReconnect+            -> a +            -> IO ()+sendMessage node from to implicitReconnect = +  sendPayload node from to implicitReconnect . messageToPayload . createMessage++setupConnBetween :: LocalNode +                 -> Identifier +                 -> Identifier +                 -> ImplicitReconnect +                 -> IO (Maybe NT.Connection)+setupConnBetween node from to implicitReconnect = do+    mConn <- NT.connect endPoint +                        (nodeAddress . nodeOf $ to) +                        NT.ReliableOrdered +                        NT.defaultConnectHints+    case mConn of +      Right conn -> do+        didSend <- NT.send conn (BSL.toChunks . encode $ to)+        case didSend of+          Left _ ->+            return Nothing+          Right () -> do+            modifyMVar_ nodeState $ return . +              (localConnectionBetween from to ^= Just (conn, implicitReconnect))+            return $ Just conn+      Left _ ->+        return Nothing+  where+    endPoint  = localEndPoint node+    nodeState = localState node++connBetween :: LocalNode +            -> Identifier +            -> Identifier +            -> ImplicitReconnect+            -> IO (Maybe NT.Connection)+connBetween node from to implicitReconnect = do+    mConn <- withMVar nodeState $ return . (^. localConnectionBetween from to)+    case mConn of+      Just (conn, _) -> +        return $ Just conn+      Nothing -> +        setupConnBetween node from to implicitReconnect +  where+    nodeState = localState node++disconnect :: LocalNode -> Identifier -> Identifier -> IO ()+disconnect 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)++closeImplicitReconnections :: LocalNode -> Identifier -> IO ()+closeImplicitReconnections node to = +  modifyMVar_ (localState node) $ \st -> do+    let shouldClose (_, to') (_, WithImplicitReconnect) = to `impliesDeathOf` to'+        shouldClose _ _ = False+    let (affected, unaffected) = Map.partitionWithKey shouldClose (st ^. localConnections)+    mapM_ (NT.close . fst) (Map.elems affected)+    return (localConnections ^= unaffected $ st)++-- | @a `impliesDeathOf` b@ is true if the death of @a@ (for instance, a node)+-- implies the death of @b@ (for instance, a process on that node)+impliesDeathOf :: Identifier+               -> Identifier+               -> Bool+NodeIdentifier nid `impliesDeathOf` NodeIdentifier nid' = +  nid' == nid+NodeIdentifier nid `impliesDeathOf` ProcessIdentifier pid =+  processNodeId pid == nid+NodeIdentifier nid `impliesDeathOf` SendPortIdentifier cid =+  processNodeId (sendPortProcessId cid) == nid+ProcessIdentifier pid `impliesDeathOf` ProcessIdentifier pid' =+  pid' == pid+ProcessIdentifier pid `impliesDeathOf` SendPortIdentifier cid =+  sendPortProcessId cid == pid+SendPortIdentifier cid `impliesDeathOf` SendPortIdentifier cid' =+  cid' == cid+_ `impliesDeathOf` _ =+  False

@@ -1,103 +0,0 @@-module Control.Distributed.Process.Internal.Node -  ( -- * Message sending-    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.Distributed.Process.Internal.StrictMVar (withMVar, modifyMVar_)-import Control.Concurrent.Chan (writeChan)-import Control.Monad (unless)-import qualified Network.Transport as NT -  ( Connection-  , send-  , defaultConnectHints-  , connect-  , Reliability(ReliableOrdered)-  , close-  )-import Control.Distributed.Process.Internal.Types -  ( LocalNode(localState, localEndPoint, localCtrlChan)-  , Identifier-  , localConnectionBetween-  , nodeAddress-  , nodeOf-  , messageToPayload-  , createMessage-  , NCMsg(..)-  , ProcessSignal(Died)-  , DiedReason(DiedDisconnect)-  )-import Control.Distributed.Process.Serializable (Serializable)------------------------------------------------------------------------------------- Message sending                                                            -- -----------------------------------------------------------------------------------sendPayload :: LocalNode -> Identifier -> Identifier -> [BSS.ByteString] -> IO ()-sendPayload node from to payload = do-  mConn <- connBetween node from to-  didSend <- case mConn of-    Just conn -> do-      didSend <- NT.send conn payload-      case didSend of-        Left _err -> return False -        Right ()  -> return True -    Nothing -> return False-  unless didSend $-    writeChan (localCtrlChan node) NCMsg-      { ctrlMsgSender = to -      , ctrlMsgSignal = Died to DiedDisconnect-      }--sendBinary :: Binary a => LocalNode -> Identifier -> Identifier -> a -> IO ()-sendBinary node from to = sendPayload node from to . BSL.toChunks . encode--sendMessage :: Serializable a => LocalNode -> Identifier -> Identifier -> a -> IO ()-sendMessage node from to = sendPayload node from to . messageToPayload . createMessage--setupConnBetween :: LocalNode -> Identifier -> Identifier -> IO (Maybe NT.Connection)-setupConnBetween node from to = do-    mConn    <- NT.connect endPoint -                           (nodeAddress . nodeOf $ to) -                           NT.ReliableOrdered -                           NT.defaultConnectHints-    case mConn of -      Right conn -> do-        didSend <- NT.send conn (BSL.toChunks . encode $ to)-        case didSend of-          Left _ ->-            return Nothing-          Right () -> do-            modifyMVar_ nodeState $ return . -              (localConnectionBetween from to ^= Just conn)-            return $ Just conn-      Left _ ->-        return Nothing-  where-    endPoint  = localEndPoint node-    nodeState = localState node--connBetween :: LocalNode -> Identifier -> Identifier -> IO (Maybe NT.Connection)-connBetween node from to = do-    mConn <- withMVar nodeState $ return . (^. localConnectionBetween from to)-    case mConn of-      Just conn -> return $ Just conn-      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)

@@ -19,6 +19,8 @@   , match   , matchIf   , matchUnknown+  , AbstractMessage(..) +  , matchAny     -- * Process management   , terminate   , ProcessTerminationException(..)@@ -38,11 +40,11 @@   , nsend   , registerRemote   , unregisterRemote-  , whereisRemote   , whereisRemoteAsync   , nsendRemote     -- * Closures   , unClosure+  , unStatic     -- * Exception handling   , catch   , mask@@ -61,7 +63,6 @@   , monitorPort     -- * Reconnecting   , reconnect-  , reconnectNode   , reconnectPort   ) where @@ -78,14 +79,16 @@ import Control.Applicative ((<$>)) import Control.Exception (Exception, throwIO, SomeException) import qualified Control.Exception as Ex (catch, mask)-import Control.Distributed.Process.Internal.StrictMVar (modifyMVar)+import Control.Distributed.Process.Internal.StrictMVar +  ( StrictMVar+  , modifyMVar+  , modifyMVar_+  ) import Control.Concurrent.Chan (writeChan) import Control.Concurrent.STM    ( STM   , atomically   , orElse-  , newTChan-  , readTChan   , newTVar   , readTVar   , writeTVar@@ -93,9 +96,9 @@ import Control.Distributed.Process.Internal.CQueue (dequeue, BlockSpec(..)) import Control.Distributed.Process.Serializable (Serializable, fingerprint) import Data.Accessor ((^.), (^:), (^=))-import Control.Distributed.Static (Closure)+import Control.Distributed.Static (Closure, Static) import Data.Rank1Typeable (Typeable)-import qualified Control.Distributed.Static as Static (unclosure)+import qualified Control.Distributed.Static as Static (unstatic, unclosure) import Control.Distributed.Process.Internal.Types    ( NodeId(..)   , ProcessId(..)@@ -123,8 +126,22 @@   , WhereIsReply(..)   , createMessage   , runLocalProcess+  , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)+  , LocalProcessState+  , LocalSendPortId+  , messageToPayload   )-import Control.Distributed.Process.Internal.Node (sendMessage, sendBinary) +import Control.Distributed.Process.Internal.Messaging +  ( sendMessage+  , sendBinary+  , sendPayload+  , disconnect+  ) +import Control.Distributed.Process.Internal.WeakTQueue +  ( newTQueueIO+  , readTQueue+  , mkWeakTQueue+  )  -------------------------------------------------------------------------------- -- Basic messaging                                                            --@@ -139,6 +156,7 @@   liftIO $ sendMessage (processNode proc)                         (ProcessIdentifier (processId proc))                         (ProcessIdentifier them)+                       NoImplicitReconnect                        msg  -- | Wait for a message of a specific type@@ -152,21 +170,26 @@ -- | Create a new typed channel newChan :: Serializable a => Process (SendPort a, ReceivePort a) newChan = do-  proc <- ask -  liftIO . modifyMVar (processState proc) $ \st -> do-    chan <- liftIO . atomically $ newTChan-    let lcid  = st ^. channelCounter-        cid   = SendPortId { sendPortProcessId = processId proc-                           , sendPortLocalId   = lcid-                           }-        sport = SendPort cid -        rport = ReceivePortSingle chan-        tch   = TypedChannel chan -    return ( (channelCounter ^: (+ 1))-           . (typedChannelWithId lcid ^= Just tch)-           $ st-           , (sport, rport)-           )+    proc <- ask +    liftIO . modifyMVar (processState proc) $ \st -> do+      let lcid  = st ^. channelCounter+      let cid   = SendPortId { sendPortProcessId = processId proc+                             , sendPortLocalId   = lcid+                             }+      let sport = SendPort cid +      chan  <- liftIO newTQueueIO+      chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid+      let rport = ReceivePortSingle chan+      let tch   = TypedChannel chan' +      return ( (channelCounter ^: (+ 1))+             . (typedChannelWithId lcid ^= Just tch)+             $ st+             , (sport, rport)+             )+  where+    finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()+    finalizer processState lcid = modifyMVar_ processState $ +      return . (typedChannelWithId lcid ^= Nothing)  -- | Send a message on a typed channel sendChan :: Serializable a => SendPort a -> a -> Process ()@@ -175,6 +198,7 @@   liftIO $ sendBinary (processNode proc)                       (ProcessIdentifier (processId proc))                       (SendPortIdentifier cid) +                      NoImplicitReconnect                       msg   -- | Wait for a message on a typed channel@@ -183,7 +207,7 @@   where     receiveSTM :: ReceivePort a -> STM a     receiveSTM (ReceivePortSingle c) = -      readTChan c+      readTQueue c     receiveSTM (ReceivePortBiased ps) =       foldr1 orElse (map receiveSTM ps)     receiveSTM (ReceivePortRR psVar) = do@@ -243,12 +267,35 @@ -- | Match against any message of the right type that satisfies a predicate matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b matchIf c p = Match $ \msg -> -  let decoded :: a-      decoded = decode . messageEncoding $ msg in-  if messageFingerprint msg == fingerprint (undefined :: a) && c decoded-    then Just $ p decoded -    else Nothing+   case messageFingerprint msg == fingerprint (undefined :: a) of+     True | c decoded -> Just (p decoded)+       where+         decoded :: a+         -- Make sure the value is fully decoded so that we don't hang to +         -- bytestrings when the process calling 'matchIf' doesn't process+         -- the values immediately+         !decoded = decode (messageEncoding msg)+     _ -> Nothing +data AbstractMessage = AbstractMessage {+    forward :: ProcessId -> Process ()+  }++-- | Match against an arbitrary message+matchAny :: forall b. (AbstractMessage -> Process b) -> Match b+matchAny p = Match $ Just . p . abstract+  where +    abstract :: Message -> AbstractMessage+    abstract msg = AbstractMessage {+        forward = \them -> do+          proc <- ask+          liftIO $ sendPayload (processNode proc)+                               (ProcessIdentifier (processId proc))+                               (ProcessIdentifier them)+                               NoImplicitReconnect +                               (messageToPayload msg)+      }+ -- | Remove any message from the queue matchUnknown :: Process b -> Match b matchUnknown = Match . const . Just@@ -291,7 +338,12 @@ monitor :: ProcessId -> Process MonitorRef  monitor = monitor' . ProcessIdentifier  --- | Remove a link (synchronous)+-- | Remove a link +--+-- This is synchronous in the sense that once it returns you are guaranteed+-- that no exception will be raised if the remote process dies. However, it is+-- asynchronous in the sense that we do not wait for a response from the remote +-- node. unlink :: ProcessId -> Process () unlink pid = do   unlinkAsync pid@@ -299,7 +351,9 @@                         (\_ -> return ())                ] --- | Remove a node link (synchronous)+-- | Remove a node link +--+-- This has the same synchronous/asynchronous nature as 'unlink'.  unlinkNode :: NodeId -> Process () unlinkNode nid = do   unlinkNodeAsync nid@@ -307,7 +361,9 @@                         (\_ -> return ())               ] --- | Remove a channel (send port) link (synchronous)+-- | Remove a channel (send port) link+--+-- This has the same synchronous/asynchronous nature as 'unlink'.  unlinkPort :: SendPort a -> Process () unlinkPort sport = do   unlinkPortAsync sport@@ -315,7 +371,9 @@                         (\_ -> return ())               ] --- | Remove a monitor (synchronous)+-- | Remove a monitor +--+-- This has the same synchronous/asynchronous nature as 'unlink'.  unmonitor :: MonitorRef -> Process () unmonitor ref = do   unmonitorAsync ref@@ -382,12 +440,12 @@   sendCtrlMsg (Just nid) $ Spawn proc spawnRef   return spawnRef --- | Monitor a node+-- | Monitor a node (asynchronous) monitorNode :: NodeId -> Process MonitorRef monitorNode =    monitor' . NodeIdentifier --- | Monitor a typed channel+-- | Monitor a typed channel (asynchronous) monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef monitorPort (SendPort cid) =    monitor' (SendPortIdentifier cid) @@ -397,11 +455,11 @@ unmonitorAsync =    sendCtrlMsg Nothing . Unmonitor --- | Link to a node+-- | Link to a node (asynchronous) linkNode :: NodeId -> Process () linkNode = link' . NodeIdentifier  --- | Link to a channel (send port)+-- | Link to a channel (asynchronous) linkPort :: SendPort a -> Process () linkPort (SendPort cid) =    link' (SendPortIdentifier cid)@@ -465,7 +523,7 @@ unregisterRemote nid label =   sendCtrlMsg (Just nid) (Register label Nothing) --- | Query the local process registry (synchronous).+-- | Query the local process registry whereis :: String -> Process (Maybe ProcessId) whereis label = do   sendCtrlMsg Nothing (WhereIs label)@@ -473,17 +531,15 @@                         (\(WhereIsReply _ mPid) -> return mPid)               ] --- | Query a remote process registry (synchronous)-whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)-whereisRemote nid label = do-  whereisRemoteAsync nid label-  receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')-                        (\(WhereIsReply _ mPid) -> return mPid)-              ]- -- | Query a remote process registry (asynchronous) ----- Reply will come in the form of a 'WhereIsReply' message+-- Reply will come in the form of a 'WhereIsReply' message. +--+-- There is currently no synchronous version of 'whereisRemoteAsync': if+-- you implement one yourself, be sure to take into account that the remote+-- node might die or get disconnect before it can respond (i.e. you should+-- use 'monitorNode' and take appropriate action when you receive a +-- 'NodeMonitorNotification'). whereisRemoteAsync :: NodeId -> String -> Process () whereisRemoteAsync nid label =    sendCtrlMsg (Just nid) (WhereIs label)@@ -502,8 +558,16 @@ -- Closures                                                                   -- -------------------------------------------------------------------------------- --- | Deserialize a closure-unClosure :: forall a. Typeable a => Closure a -> Process a+-- | Resolve a static value+unStatic :: Typeable a => Static a -> Process a+unStatic static = do+  rtable <- remoteTable . processNode <$> ask+  case Static.unstatic rtable static of+    Left err -> fail $ "Could not resolve static value: " ++ err+    Right x  -> return x++-- | Resolve a closure+unClosure :: Typeable a => Closure a -> Process a unClosure closure = do   rtable <- remoteTable . processNode <$> ask    case Static.unclosure rtable closure of@@ -535,18 +599,17 @@ -- 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 them = do+  us <- getSelfPid+  node <- processNode <$> ask+  liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)  -- | Reconnect to a sendport. See 'reconnect' for more information. reconnectPort :: SendPort a -> Process ()-reconnectPort = -  sendCtrlMsg Nothing . Reconnect . SendPortIdentifier . sendPortId+reconnectPort them = do +  us <- getSelfPid+  node <- processNode <$> ask+  liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))  -------------------------------------------------------------------------------- -- Auxiliary functions                                                        --@@ -598,4 +661,5 @@       liftIO $ sendBinary (processNode proc)                           (ProcessIdentifier (processId proc))                           (NodeIdentifier nid)+                          WithImplicitReconnect                           msg

@@ -0,0 +1,21 @@+module Control.Distributed.Process.Internal.StrictContainerAccessors +  ( mapMaybe+  , mapDefault+  ) where++import Prelude hiding (map)+import Data.Accessor+import Data.Map (Map)+import qualified Data.Map as Map (lookup, insert, delete, findWithDefault)++mapMaybe :: Ord key => key -> Accessor (Map key elem) (Maybe elem)+mapMaybe key = accessor+  (Map.lookup key)+  (\mVal map -> case mVal of+      Nothing  -> Map.delete key map+      Just val -> val `seq` Map.insert key val map)++mapDefault :: Ord key => elem -> key -> Accessor (Map key elem) elem+mapDefault def key = accessor+  (Map.findWithDefault def key)+  (\val map -> val `seq` Map.insert key val map)

@@ -0,0 +1,25 @@+-- | Spine and element strict list+module Control.Distributed.Process.Internal.StrictList +  ( StrictList(Cons, Nil)+  , length +  , reverse+  , reverse'+  ) where++import Prelude hiding (length, reverse)++-- | Strict list+data StrictList a = Cons !a !(StrictList a) | Nil++length :: StrictList a -> Int+length Nil         = 0+length (Cons _ xs) = 1 + length xs++-- | Reverse a strict list+reverse :: StrictList a -> StrictList a +reverse xs = reverse' xs Nil++-- | @reverseStrict' xs ys@ is 'reverse xs ++ ys' if they were lists+reverse' :: StrictList a -> StrictList a -> StrictList a +reverse' Nil         ys = ys+reverse' (Cons x xs) ys = reverse' xs (Cons x ys)

@@ -1,6 +1,7 @@ -- | Like Control.Concurrent.MVar.Strict but reduce to HNF, not NF+{-# LANGUAGE MagicHash, UnboxedTuples #-} module Control.Distributed.Process.Internal.StrictMVar -  ( StrictMVar+  ( StrictMVar(StrictMVar)   , newEmptyMVar   , newMVar   , takeMVar@@ -8,6 +9,7 @@   , withMVar   , modifyMVar_   , modifyMVar+  , mkWeakMVar   ) where  import Control.Applicative ((<$>))@@ -23,6 +25,10 @@   , modifyMVar_   , modifyMVar   )+import GHC.MVar (MVar(MVar))+import GHC.IO (IO(IO)) +import GHC.Prim (mkWeak#)+import GHC.Weak (Weak(Weak))  newtype StrictMVar a = StrictMVar (MVar.MVar a) @@ -49,3 +55,7 @@   where     evaluateFst :: (a, b) -> IO (a, b)     evaluateFst (x, y) = evaluate x >> return (x, y)++mkWeakMVar :: StrictMVar a -> IO () -> IO (Weak (StrictMVar a))+mkWeakMVar m@(StrictMVar (MVar m#)) f = IO $ \s ->+  case mkWeak# m# m f s of (# s1, w #) -> (# s1, Weak w #)

@@ -10,6 +10,8 @@   , ProcessId(..)   , Identifier(..)   , nodeOf+  , firstNonReservedProcessId+  , nullProcessId     -- * Local nodes and processes   , LocalNode(..)   , LocalNodeState(..)@@ -17,6 +19,7 @@   , LocalProcessState(..)   , Process(..)   , runLocalProcess+  , ImplicitReconnect(..)     -- * Typed channels   , LocalSendPortId    , SendPortId(..)@@ -61,24 +64,25 @@   , typedChannelWithId   ) where +import System.Mem.Weak (Weak) import Data.Map (Map) import Data.Int (Int32) import Data.Typeable (Typeable) import Data.Binary (Binary(put, get), putWord8, getWord8, encode)-import qualified Data.ByteString as BSS (ByteString, concat)+import qualified Data.ByteString as BSS (ByteString, concat, copy) import qualified Data.ByteString.Lazy as BSL    ( ByteString   , toChunks   , splitAt   , fromChunks   )+import qualified Data.ByteString.Lazy.Internal as BSL (ByteString(..)) import Data.Accessor (Accessor, accessor)-import qualified Data.Accessor.Container as DAC (mapMaybe) import Control.Category ((>>>)) import Control.Exception (Exception) import Control.Concurrent (ThreadId) import Control.Concurrent.Chan (Chan)-import Control.Concurrent.STM (TChan, TVar)+import Control.Concurrent.STM (TVar) import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection) import Control.Applicative (Applicative, (<$>), (<*>)) import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)@@ -94,10 +98,9 @@   ) import Control.Distributed.Process.Internal.CQueue (CQueue) import Control.Distributed.Process.Internal.StrictMVar (StrictMVar)+import Control.Distributed.Process.Internal.WeakTQueue (TQueue) import Control.Distributed.Static (RemoteTable, Closure)---- import Control.Distributed.Process.Internal.Dynamic (Dynamic) --- import Control.Distributed.Process.Internal.TypeRep (compareTypeRep) -- and Binary instances+import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC (mapMaybe)  -------------------------------------------------------------------------------- -- Node and process identifiers                                               --@@ -105,7 +108,7 @@  -- | Node identifier  newtype NodeId = NodeId { nodeAddress :: NT.EndPointAddress }-  deriving (Eq, Ord, Binary)+  deriving (Eq, Ord, Binary, Typeable)  instance Show NodeId where   show (NodeId addr) = "nid://" ++ show addr @@ -133,9 +136,9 @@  -- | Union of all kinds of identifiers  data Identifier = -    NodeIdentifier NodeId-  | ProcessIdentifier ProcessId -  | SendPortIdentifier SendPortId+    NodeIdentifier !NodeId+  | ProcessIdentifier !ProcessId +  | SendPortIdentifier !SendPortId   deriving (Eq, Ord)  instance Show Identifier where@@ -149,39 +152,64 @@ nodeOf (SendPortIdentifier cid) = processNodeId (sendPortProcessId cid)  --------------------------------------------------------------------------------+-- Special PIDs                                                               --+--------------------------------------------------------------------------------++firstNonReservedProcessId :: Int32+firstNonReservedProcessId = 1++nullProcessId :: NodeId -> ProcessId+nullProcessId nid = +  ProcessId { processNodeId  = nid+            , processLocalId = LocalProcessId { lpidUnique  = 0+                                              , lpidCounter = 0+                                              }+            }++-------------------------------------------------------------------------------- -- Local nodes and processes                                                  -- --------------------------------------------------------------------------------  -- | Local nodes data LocalNode = LocalNode    { -- | 'NodeId' of the node-    localNodeId :: NodeId+    localNodeId :: !NodeId     -- | The network endpoint associated with this node -  , localEndPoint :: NT.EndPoint +  , localEndPoint :: !NT.EndPoint      -- | Local node state -  , localState :: StrictMVar LocalNodeState+  , localState :: !(StrictMVar LocalNodeState)     -- | Channel for the node controller-  , localCtrlChan :: Chan NCMsg+  , localCtrlChan :: !(Chan NCMsg)     -- | Runtime lookup table for supporting closures     -- TODO: this should be part of the CH state, not the local endpoint state-  , remoteTable :: RemoteTable +  , remoteTable :: !RemoteTable    } +data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect+  deriving (Eq, Show)+ -- | Local node state data LocalNodeState = LocalNodeState -  { _localProcesses   :: !(Map LocalProcessId LocalProcess)+  { -- | Processes running on this node+    _localProcesses   :: !(Map LocalProcessId LocalProcess)+    -- | Counter to assign PIDs    , _localPidCounter  :: !Int32+    -- | The 'unique' value used to create PIDs (so that processes on+    -- restarted nodes have new PIDs)   , _localPidUnique   :: !Int32-  , _localConnections :: !(Map (Identifier, Identifier) NT.Connection)+    -- | Outgoing connections+  , _localConnections :: !(Map (Identifier, Identifier) +                               (NT.Connection, ImplicitReconnect))   }  -- | Processes running on our local node data LocalProcess = LocalProcess -  { processQueue  :: CQueue Message -  , processId     :: ProcessId-  , processState  :: StrictMVar LocalProcessState-  , processThread :: ThreadId-  , processNode   :: LocalNode+  { processQueue  :: !(CQueue Message)+  , processWeakQ  :: !(Weak (CQueue Message))+  , processId     :: !ProcessId+  , processState  :: !(StrictMVar LocalProcessState)+  , processThread :: !ThreadId+  , processNode   :: !LocalNode   }  -- | Deconstructor for 'Process' (not exported to the public API) @@ -224,7 +252,7 @@   show (SendPortId (ProcessId (NodeId addr) (LocalProcessId _ plid)) clid)       = "cid://" ++ show addr ++ ":" ++ show plid ++ ":" ++ show clid -data TypedChannel = forall a. Serializable a => TypedChannel (TChan a)+data TypedChannel = forall a. Serializable a => TypedChannel (Weak (TQueue a))  -- | The send send of a typed channel (serializable) newtype SendPort a = SendPort { @@ -236,7 +264,7 @@ -- | The receive end of a typed channel (not serializable) data ReceivePort a =      -- | A single receive port-    ReceivePortSingle (TChan a)+    ReceivePortSingle (TQueue a)     -- | A left-biased combination of receive ports    | ReceivePortBiased [ReceivePort a]      -- | A round-robin combination of receive ports@@ -249,8 +277,8 @@  -- | Messages consist of their typeRep fingerprint and their encoding data Message = Message -  { messageFingerprint :: Fingerprint -  , messageEncoding    :: BSL.ByteString+  { messageFingerprint :: !Fingerprint +  , messageEncoding    :: !BSL.ByteString   }  instance Show Message where@@ -266,12 +294,20 @@  -- | Deserialize a message payloadToMessage :: [BSS.ByteString] -> Message-payloadToMessage payload = Message fp msg+payloadToMessage payload = Message fp (copy msg)    where+    encFp :: BSL.ByteString+    msg   :: BSL.ByteString     (encFp, msg) = BSL.splitAt (fromIntegral sizeOfFingerprint)                   $ BSL.fromChunks payload ++    fp :: Fingerprint     fp = decodeFingerprint . BSS.concat . BSL.toChunks $ encFp +    copy :: BSL.ByteString -> BSL.ByteString+    copy (BSL.Chunk bs BSL.Empty) = BSL.Chunk (BSS.copy bs) BSL.Empty+    copy bsl = BSL.fromChunks . return . BSS.concat . BSL.toChunks $ bsl + -------------------------------------------------------------------------------- -- Node controller user-visible data types                                    -- --------------------------------------------------------------------------------@@ -287,32 +323,32 @@  -- | Message sent by process monitors data ProcessMonitorNotification = -    ProcessMonitorNotification MonitorRef ProcessId DiedReason+    ProcessMonitorNotification !MonitorRef !ProcessId !DiedReason   deriving (Typeable, Show)  -- | Message sent by node monitors data NodeMonitorNotification = -    NodeMonitorNotification MonitorRef NodeId DiedReason+    NodeMonitorNotification !MonitorRef !NodeId !DiedReason   deriving (Typeable, Show)  -- | Message sent by channel (port) monitors data PortMonitorNotification = -    PortMonitorNotification MonitorRef SendPortId DiedReason+    PortMonitorNotification !MonitorRef !SendPortId !DiedReason   deriving (Typeable, Show)  -- | Exceptions thrown when a linked process dies data ProcessLinkException = -    ProcessLinkException ProcessId DiedReason+    ProcessLinkException !ProcessId !DiedReason   deriving (Typeable, Show)  -- | Exception thrown when a linked node dies data NodeLinkException = -    NodeLinkException NodeId DiedReason+    NodeLinkException !NodeId !DiedReason   deriving (Typeable, Show)  -- | Exception thrown when a linked channel (port) dies data PortLinkException = -    PortLinkException SendPortId DiedReason+    PortLinkException !SendPortId !DiedReason   deriving (Typeable, Show)  instance Exception ProcessLinkException@@ -325,7 +361,7 @@     DiedNormal     -- | The process exited with an exception     -- (provided as 'String' because 'Exception' does not implement 'Binary')-  | DiedException String+  | DiedException !String     -- | We got disconnected from the process node   | DiedDisconnect     -- | The process node died@@ -368,23 +404,22 @@  -- | Messages to the node controller data NCMsg = NCMsg -  { ctrlMsgSender :: Identifier -  , ctrlMsgSignal :: ProcessSignal+  { ctrlMsgSender :: !Identifier +  , ctrlMsgSignal :: !ProcessSignal   }   deriving Show  -- | Signals to the node controller (see 'NCMsg') data ProcessSignal =-    Link Identifier -  | Unlink Identifier -  | Monitor MonitorRef-  | Unmonitor MonitorRef-  | Died Identifier DiedReason-  | Spawn (Closure (Process ())) SpawnRef -  | WhereIs String-  | Register String (Maybe ProcessId) -- Nothing to unregister-  | NamedSend String Message-  | Reconnect Identifier+    Link !Identifier +  | Unlink !Identifier +  | Monitor !MonitorRef+  | Unmonitor !MonitorRef+  | Died Identifier !DiedReason+  | Spawn !(Closure (Process ())) !SpawnRef +  | WhereIs !String+  | Register !String !(Maybe ProcessId) -- Use 'Nothing' to unregister+  | NamedSend !String !Message   deriving Show  --------------------------------------------------------------------------------@@ -429,7 +464,6 @@   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@@ -442,7 +476,6 @@       6 -> WhereIs <$> get       7 -> Register <$> get <*> get       8 -> NamedSend <$> get <*> (payloadToMessage <$> get)-      9 -> Reconnect <$> get       _ -> fail "ProcessSignal.get: invalid"  instance Binary DiedReason where@@ -498,13 +531,13 @@ localPidUnique :: Accessor LocalNodeState Int32 localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq }) -localConnections :: Accessor LocalNodeState (Map (Identifier, Identifier) NT.Connection)+localConnections :: Accessor LocalNodeState (Map (Identifier, Identifier) (NT.Connection, ImplicitReconnect)) localConnections = accessor _localConnections (\conns st -> st { _localConnections = conns })  localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess) localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid -localConnectionBetween :: Identifier -> Identifier -> Accessor LocalNodeState (Maybe NT.Connection)+localConnectionBetween :: Identifier -> Identifier -> Accessor LocalNodeState (Maybe (NT.Connection, ImplicitReconnect)) localConnectionBetween from to = localConnections >>> DAC.mapMaybe (from, to)  monitorCounter :: Accessor LocalProcessState Int32

@@ -0,0 +1,101 @@+-- | Clone of Control.Concurrent.STM.TQueue with support for mkWeakTQueue+--+-- Not all functionality from the original module is available: unGetTQueue,+-- peekTQueue and tryPeekTQueue are missing. In order to implement these we'd+-- need to be able to touch# the write end of the queue inside unGetTQueue, but+-- that means we need a version of touch# that works within the STM monad.+{-# LANGUAGE MagicHash, UnboxedTuples #-}+module Control.Distributed.Process.Internal.WeakTQueue (+  -- * Original functionality+  TQueue,+  newTQueue,+  newTQueueIO,+  readTQueue,+  tryReadTQueue,+  writeTQueue,+  isEmptyTQueue,+  -- * New functionality+  mkWeakTQueue+  ) where++import Prelude hiding (read)+import GHC.Conc+import Data.Typeable (Typeable)+import GHC.IO (IO(IO)) +import GHC.Prim (mkWeak#)+import GHC.Weak (Weak(Weak))++--------------------------------------------------------------------------------+-- Original functionality                                                     --+--------------------------------------------------------------------------------++-- | 'TQueue' is an abstract type representing an unbounded FIFO channel.+data TQueue a = TQueue {-# UNPACK #-} !(TVar [a])+                       {-# UNPACK #-} !(TVar [a])+  deriving Typeable++instance Eq (TQueue a) where+  TQueue a _ == TQueue b _ = a == b++-- |Build and returns a new instance of 'TQueue'+newTQueue :: STM (TQueue a)+newTQueue = do+  read  <- newTVar []+  write <- newTVar []+  return (TQueue read write)++-- |@IO@ version of 'newTQueue'.  This is useful for creating top-level+-- 'TQueue's using 'System.IO.Unsafe.unsafePerformIO', because using+-- 'atomically' inside 'System.IO.Unsafe.unsafePerformIO' isn't+-- possible.+newTQueueIO :: IO (TQueue a)+newTQueueIO = do+  read  <- newTVarIO []+  write <- newTVarIO []+  return (TQueue read write)++-- |Write a value to a 'TQueue'.+writeTQueue :: TQueue a -> a -> STM ()+writeTQueue (TQueue _read write) a = do+  listend <- readTVar write+  writeTVar write (a:listend)++-- |Read the next value from the 'TQueue'.+readTQueue :: TQueue a -> STM a+readTQueue (TQueue read write) = do+  xs <- readTVar read+  case xs of+    (x:xs') -> do writeTVar read xs'+                  return x+    [] -> do ys <- readTVar write+             case ys of+               [] -> retry+               _  -> case reverse ys of+                       [] -> error "readTQueue"+                       (z:zs) -> do writeTVar write []+                                    writeTVar read zs+                                    return z++-- | A version of 'readTQueue' which does not retry. Instead it+-- returns @Nothing@ if no value is available.+tryReadTQueue :: TQueue a -> STM (Maybe a)+tryReadTQueue c = fmap Just (readTQueue c) `orElse` return Nothing++-- |Returns 'True' if the supplied 'TQueue' is empty.+isEmptyTQueue :: TQueue a -> STM Bool+isEmptyTQueue (TQueue read write) = do+  xs <- readTVar read+  case xs of+    (_:_) -> return False+    [] -> do ys <- readTVar write+             case ys of+               [] -> return True+               _  -> return False++--------------------------------------------------------------------------------+-- New functionality                                                          --+--------------------------------------------------------------------------------++mkWeakTQueue :: TQueue a -> IO () -> IO (Weak (TQueue a))+mkWeakTQueue q@(TQueue _read (TVar write#)) f = IO $ \s ->+  case mkWeak# write# q f s of (# s', w #) -> (# s', Weak w #) 

@@ -1,4 +1,7 @@ -- | Local nodes+--+-- TODO: Calls to 'sendBinary' and co by the node controller may stall the+-- node controller. module Control.Distributed.Process.Node    ( LocalNode   , newLocalNode@@ -14,21 +17,19 @@ #endif  import System.IO (fixIO, hPutStrLn, stderr)+import System.Mem.Weak (Weak, deRefWeak) import qualified Data.ByteString.Lazy as BSL (fromChunks) import Data.Binary (decode) import Data.Map (Map) import qualified Data.Map as Map    ( empty-  , lookup-  , insert-  , delete   , toList   , partitionWithKey-  , filterWithKey   , elems+  , filterWithKey   ) import Data.Set (Set)-import qualified Data.Set as Set (empty, insert, delete, member, filter)+import qualified Data.Set as Set (empty, insert, delete, member) import Data.Foldable (forM_) import Data.Maybe (isJust) import Data.Typeable (Typeable)@@ -52,8 +53,13 @@   , takeMVar   ) import Control.Concurrent.Chan (newChan, writeChan, readChan)-import Control.Concurrent.STM (atomically, writeTChan)-import Control.Distributed.Process.Internal.CQueue (enqueue, newCQueue)+import Control.Concurrent.STM (atomically)+import Control.Distributed.Process.Internal.CQueue +  ( CQueue+  , enqueue+  , newCQueue+  , mkWeakCQueue+  ) import qualified Network.Transport as NT    ( Transport   , EndPoint@@ -67,9 +73,10 @@   , ConnectionId   , Connection   , close+  , EndPointAddress+  , Reliability(ReliableOrdered)   ) 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 @@ -116,16 +123,24 @@   , payloadToMessage   , createMessage   , runLocalProcess+  , firstNonReservedProcessId+  , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)   ) import Control.Distributed.Process.Serializable (Serializable)-import Control.Distributed.Process.Internal.Node +import Control.Distributed.Process.Internal.Messaging   ( sendBinary   , sendMessage   , sendPayload-  , reconnect+  , closeImplicitReconnections +  , impliesDeathOf   ) import Control.Distributed.Process.Internal.Primitives (expect, register, finally) import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)+import Control.Distributed.Process.Internal.WeakTQueue (writeTQueue)+import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC+  ( mapMaybe+  , mapDefault+  )  -------------------------------------------------------------------------------- -- Initialization                                                             --@@ -151,7 +166,7 @@   unq <- randomIO   state <- newMVar LocalNodeState      { _localProcesses   = Map.empty-    , _localPidCounter  = 0 +    , _localPidCounter  = firstNonReservedProcessId      , _localPidUnique   = unq      , _localConnections = Map.empty     }@@ -212,8 +227,10 @@                                        , _typedChannels  = Map.empty                                        }       queue <- newCQueue+      weakQueue <- mkWeakCQueue queue (return ())       (_, lproc) <- fixIO $ \ ~(tid, _) -> do         let lproc = LocalProcess { processQueue  = queue+                                 , processWeakQ  = weakQueue                                  , processId     = pid                                  , processState  = pst                                   , processThread = tid@@ -235,7 +252,7 @@         then do           newUnique <- randomIO           return ( (localProcessWithId lpid ^= Just lproc)-                 . (localPidCounter ^= 0)+                 . (localPidCounter ^= firstNonReservedProcessId)                  . (localPidUnique ^= newUnique)                  $ st                  , pid@@ -251,63 +268,91 @@     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)+      mapM_ (NT.close . fst) (Map.elems affected)       return $ (localProcessWithId (processLocalId pid) ^= Nothing)              . (localConnections ^= unaffected)              $ st +--------------------------------------------------------------------------------+-- Handle incoming messages                                                   --+--------------------------------------------------------------------------------++type IncomingConnection = (NT.EndPointAddress, IncomingTarget)++data IncomingTarget =  +    Uninit+  | ToProc (Weak (CQueue Message))+  | ToChan TypedChannel+  | ToNode++data ConnectionState = ConnectionState {+    _incoming     :: !(Map NT.ConnectionId IncomingConnection)+  , _incomingFrom :: !(Map NT.EndPointAddress (Set NT.ConnectionId))+  }++initConnectionState :: ConnectionState+initConnectionState = ConnectionState {+    _incoming     = Map.empty+  , _incomingFrom = Map.empty+  }++incoming :: Accessor ConnectionState (Map NT.ConnectionId IncomingConnection)+incoming = accessor _incoming (\conns st -> st { _incoming = conns })++incomingAt :: NT.ConnectionId -> Accessor ConnectionState (Maybe IncomingConnection)+incomingAt cid = incoming >>> DAC.mapMaybe cid++incomingFrom :: NT.EndPointAddress -> Accessor ConnectionState (Set NT.ConnectionId) +incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr+  where+    aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })+ handleIncomingMessages :: LocalNode -> IO ()-handleIncomingMessages node = go Set.empty Map.empty Map.empty Set.empty+handleIncomingMessages node = go initConnectionState    where-    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 :: ConnectionState -> IO () +    go !st = 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 (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 `Set.member` uninitConns-               ) of-            (Just proc, _, _, _) -> do-              let msg = payloadToMessage payload-              enqueue (processQueue proc) msg-              go uninitConns procs chans ctrls -            (_, Just (TypedChannel chan), _, _) -> do-              atomically $ writeTChan chan . decode . BSL.fromChunks $ payload-              go uninitConns procs chans ctrls -            (_, _, True, _) -> do+        NT.ConnectionOpened cid rel theirAddr ->+          if rel == NT.ReliableOrdered +            then go ( (incomingAt cid ^= Just (theirAddr, Uninit))+                    . (incomingFrom theirAddr ^: Set.insert cid)+                    $ st+                    )+            else invalidRequest cid st+        NT.Received cid payload ->+          case st ^. incomingAt cid of+            Just (_, ToProc weakQueue) -> do+              mQueue <- deRefWeak weakQueue+              forM_ mQueue $ \queue -> do +                -- TODO: if we find that the queue is Nothing, should we remove+                -- it from the NC state? (and same for channels, below)+                let msg = payloadToMessage payload+                enqueue queue msg -- 'enqueue' is strict+              go st +            Just (_, ToChan (TypedChannel chan')) -> do+              mChan <- deRefWeak chan'+              -- If mChan is Nothing, the process has given up the read end of+              -- the channel and we simply ignore the incoming message+              forM_ mChan $ \chan -> atomically $  +                -- We make sure the message is fully decoded when it is enqueued+                writeTQueue chan $! decode (BSL.fromChunks payload)+              go st +            Just (_, ToNode) -> do               let ctrlMsg = decode . BSL.fromChunks $ payload-              writeChan ctrlChan ctrlMsg-              go uninitConns procs chans ctrls -            (_, _, _, True) -> +              writeChan ctrlChan $! ctrlMsg+              go st +            Just (src, Uninit) ->                case decode (BSL.fromChunks payload) of                 ProcessIdentifier pid -> do                   let lpid = processLocalId pid                   mProc <- withMVar state $ return . (^. localProcessWithId lpid)                    case mProc of                     Just proc -> -                      go (Set.delete cid uninitConns) -                         (Map.insert cid proc procs)-                         chans-                         ctrls+                      go (incomingAt cid ^= Just (src, ToProc (processWeakQ proc)) $ st)                      Nothing -> -                      -- Request for an unknown process. -                      ---                      -- TODO: We should treat this as a fatal error on the-                      -- part of the remote node. That is, we should report the-                      -- remote node as having died, and we should close-                      -- incoming connections (this requires a Transport layer-                      -- extension). (**)-                      go (Set.delete cid uninitConns) procs chans ctrls+                      invalidRequest cid st                 SendPortIdentifier chId -> do                   let lcid = sendPortLocalId chId                       lpid = processLocalId (sendPortProcessId chId)@@ -316,50 +361,40 @@                     Just proc -> do                       mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)                       case mChannel of-                        Just channel ->-                          go (Set.delete cid uninitConns)-                             procs-                             (Map.insert cid channel chans)-                             ctrls+                        Just channel -> +                          go (incomingAt cid ^= Just (src, ToChan channel) $ st)                         Nothing ->-                          -- Unknown typed channel-                          -- TODO (**) above-                          go (Set.delete cid uninitConns) procs chans ctrls+                          invalidRequest cid st                     Nothing ->-                      -- Unknown process-                      -- TODO (**) above-                      go (Set.delete cid uninitConns) procs chans ctrls-                NodeIdentifier _ ->-                  go (Set.delete cid uninitConns)-                     procs-                     chans-                     (Set.insert cid ctrls)-            _ ->-              -- Unexpected message-              -- TODO (**) above -              go uninitConns procs chans ctrls+                      invalidRequest cid st+                NodeIdentifier nid ->+                  if nid == localNodeId node+                    then go (incomingAt cid ^= Just (src, ToNode) $ st)+                    else invalidRequest cid st +            Nothing ->+              invalidRequest cid st         NT.ConnectionClosed cid -> -          go (Set.delete cid uninitConns) -             (Map.delete cid procs)-             (Map.delete cid chans)-             (Set.delete cid ctrls)-        NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost (Just theirAddr) cids) _) -> do +          case st ^. incomingAt cid of+            Nothing -> +              invalidRequest cid st+            Just (src, _) -> +              go ( (incomingAt cid ^= Nothing)+                 . (incomingFrom src ^: Set.delete cid)+                 $ st+                 )+        NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr) _) -> do            -- [Unified table 9, rule node_disconnect]           let nid = NodeIdentifier $ NodeId theirAddr           writeChan ctrlChan NCMsg              { ctrlMsgSender = nid             , ctrlMsgSignal = Died nid DiedDisconnect             }-          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)-          -- as a fatal error on the part of the remote node (like (**), above)-          fail "handleIncomingMessages: TODO"+          let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))+          closeImplicitReconnections node nid +          go ( (incomingFrom theirAddr ^= Set.empty)+             . (incoming ^: Map.filterWithKey (const . notLost))+             $ st +             )         NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->           fail $ "Cloud Haskell fatal error: end point failed: " ++ str          NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->@@ -370,7 +405,17 @@           -- If we received a multicast message, something went horribly wrong           -- and we just give up           fail "Cloud Haskell fatal error: received unexpected multicast"-    ++    invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()+    invalidRequest cid st = +      -- TODO: We should treat this as a fatal error on the part of the remote+      -- node. That is, we should report the remote node as having died, and we+      -- should close incoming connections (this requires a Transport layer+      -- extension).+      go ( incomingAt cid ^= Nothing+         $ st +         )+     state    = localState node     endpoint = localEndPoint node     ctrlChan = localCtrlChan node@@ -422,6 +467,7 @@         liftIO $ sendBinary node                             (ctrlMsgSender msg)                             (NodeIdentifier nid') +                            WithImplicitReconnect                             msg       _ ->          return ()@@ -445,8 +491,6 @@         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 @@ -475,6 +519,7 @@       liftIO $ sendBinary node                           (NodeIdentifier $ localNodeId node)                           (NodeIdentifier $ processNodeId from)+                          WithImplicitReconnect         NCMsg             { ctrlMsgSender = NodeIdentifier (localNodeId node)           , ctrlMsgSignal = Died them DiedUnknownId@@ -537,6 +582,7 @@   liftIO $ sendMessage node                        (NodeIdentifier (localNodeId node))                        (ProcessIdentifier pid) +                       WithImplicitReconnect                        (DidSpawn ref pid')   -- Unified semantics does not explicitly describe how to implement 'register',@@ -556,6 +602,7 @@   liftIO $ sendMessage node                        (NodeIdentifier (localNodeId node))                        (ProcessIdentifier from) +                       WithImplicitReconnect                        (WhereIsReply label mPid)  -- [Unified: Table 14]@@ -568,14 +615,9 @@     liftIO $ sendPayload node                          from                          (ProcessIdentifier pid) +                         NoImplicitReconnect                          (messageToPayload msg)  --- Reconnecting-ncEffectReconnect :: Identifier -> Identifier -> NC ()-ncEffectReconnect from to = do-  node <- ask-  liftIO $ reconnect node from to- -------------------------------------------------------------------------------- -- Auxiliary                                                                  -- --------------------------------------------------------------------------------@@ -601,11 +643,11 @@     (True, Nothing, SendPortIdentifier pid) ->       throwException dest $ PortLinkException pid reason      (False, _, _) ->-      -- TODO: why the change in sender? How does that affect 'reconnect' semantics?-      -- (see [Unified: Table 10]+      -- The change in sender comes from [Unified: Table 10]       liftIO $ sendBinary node                           (NodeIdentifier $ localNodeId node)                           (NodeIdentifier $ processNodeId dest)+                          WithImplicitReconnect         NCMsg           { ctrlMsgSender = NodeIdentifier (localNodeId node)            , ctrlMsgSignal = Died src reason@@ -621,7 +663,6 @@ 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@@ -721,33 +762,15 @@ -- -- * the notifications for that process specifically and  -- * the notifications for typed channels to that process.+--+-- See https://github.com/haskell/containers/issues/14 for the bang on _v. splitNotif :: Identifier            -> Map Identifier a            -> (Map Identifier a, Map Identifier a)-splitNotif ident = Map.partitionWithKey (\k !v -> impliesDeathOf ident k) ---- | Does the death of one entity (node, project, channel) imply the death--- of another?-impliesDeathOf :: Identifier -- ^ Who died -               -> Identifier -- ^ Who's being watched -               -> Bool       -- ^ Does this death implies the death of the watchee? -NodeIdentifier nid `impliesDeathOf` NodeIdentifier nid' = -  nid' == nid-NodeIdentifier nid `impliesDeathOf` ProcessIdentifier pid =-  processNodeId pid == nid-NodeIdentifier nid `impliesDeathOf` SendPortIdentifier cid =-  processNodeId (sendPortProcessId cid) == nid-ProcessIdentifier pid `impliesDeathOf` ProcessIdentifier pid' =-  pid' == pid-ProcessIdentifier pid `impliesDeathOf` SendPortIdentifier cid =-  sendPortProcessId cid == pid-SendPortIdentifier cid `impliesDeathOf` SendPortIdentifier cid' =-  cid' == cid-_ `impliesDeathOf` _ =-  False+splitNotif ident = Map.partitionWithKey (\k !_v -> ident `impliesDeathOf` k)   ----------------------------------------------------------------------------------- Strict evaluation of the state                                             --+-- Auxiliary                                                                  -- --------------------------------------------------------------------------------  -- | Modify and evaluate the state

@@ -15,13 +15,21 @@   , takeMVar   , readMVar   )-import Control.Monad (replicateM_, replicateM)+import Control.Monad (replicateM_, replicateM, forever) import Control.Exception (throwIO) import Control.Applicative ((<$>), (<*>)) import qualified Network.Transport as NT (Transport, closeEndPoint)-import Network.Transport.TCP (createTransport, defaultTCPParameters)+import Network.Socket (sClose)+import Network.Transport.TCP +  ( createTransportExposeInternals+  , TransportInternals(socketBetween)+  , defaultTCPParameters+  ) import Control.Distributed.Process-import Control.Distributed.Process.Internal.Types (LocalNode(localEndPoint))+import Control.Distributed.Process.Internal.Types +  ( NodeId(nodeAddress) +  , LocalNode(localEndPoint)+  ) import Control.Distributed.Process.Node import Control.Distributed.Process.Serializable (Serializable) import TestAuxiliary@@ -40,6 +48,13 @@   send partner (Ping self)   ping +-- | Quick and dirty synchronous version of whereisRemoteAsync+whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)+whereisRemote nid string = do+  whereisRemoteAsync nid string+  WhereIsReply _ mPid <- expect+  return mPid+ -- | Basic ping test testPing :: NT.Transport -> IO () testPing transport = do@@ -515,8 +530,10 @@     pid <- forkProcess node1 $ do       sport <- expect :: Process (SendPort ())       ref <- monitorPort sport-      PortMonitorNotification ref' port' DiedNormal <- expect-      return $ ref' == ref && port' == sendPortId sport +      PortMonitorNotification ref' port' reason <- expect+      -- reason might be DiedUnknownId if the receive port is GCed before the +      -- monitor is established (TODO: not sure that this is reasonable)+      return $ ref' == ref && port' == sendPortId sport && (reason == DiedNormal || reason == DiedUnknownId)       liftIO $ putMVar gotNotification ()      runProcess node2 $ do@@ -541,7 +558,7 @@     Ping pid' <- expect      True <- return $ pingServer == pid'     return ()-    + testRemoteRegistry :: NT.Transport -> IO () testRemoteRegistry transport = do   node1 <- newLocalNode transport initRemoteTable@@ -578,9 +595,109 @@     send pid sport     expect +testReconnect :: NT.Transport -> TransportInternals -> IO ()+testReconnect transport transportInternals = do+  [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable+  let nid1 = localNodeId node1+      nid2 = localNodeId node2+  processA <- newEmptyMVar+  [sendTestOk, registerTestOk] <- replicateM 2 newEmptyMVar++  forkProcess node1 $ do+    us <- getSelfPid+    liftIO $ putMVar processA us+    msg1 <- expect+    msg2 <- expect+    True <- return $ msg1 == "message 1" && msg2 == "message 3"+    liftIO $ putMVar sendTestOk ()++  forkProcess node2 $ do+    {-+     - Make sure there is no implicit reconnect on normal message sending+     -}++    them <- liftIO $ readMVar processA+    send them "message 1" >> liftIO (threadDelay 100000)++    -- Simulate network failure+    liftIO $ do +      sock <- socketBetween transportInternals (nodeAddress nid1) (nodeAddress nid2)+      sClose sock+      threadDelay 10000++    -- Should not arrive+    send them "message 2" ++    -- Should arrive+    reconnect them+    send them "message 3" ++    liftIO $ takeMVar sendTestOk++    {-+     - Test that there *is* implicit reconnect on node controller messages+     -}+    +    us <- getSelfPid+    registerRemote nid1 "a" us >> liftIO (threadDelay 100000) -- registerRemote is asynchronous++    -- Simulate network failure+    liftIO $ do +      sock <- socketBetween transportInternals (nodeAddress nid1) (nodeAddress nid2)+      sClose sock+      threadDelay 10000++    -- This will happen due to implicit reconnect +    registerRemote nid1 "b" us++    -- Should happen+    registerRemote nid1 "c" us++    -- Check+    Just _  <- whereisRemote nid1 "a" +    Just _  <- whereisRemote nid1 "b" +    Just _  <- whereisRemote nid1 "c" ++    liftIO $ putMVar registerTestOk ()++  takeMVar registerTestOk++-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server +-- in between+testMatchAny :: NT.Transport -> IO ()+testMatchAny transport = do+  proxyAddr <- newEmptyMVar +  clientDone <- newEmptyMVar++  -- Math server+  forkIO $ do+    localNode <- newLocalNode transport initRemoteTable +    mathServer <- forkProcess localNode math+    proxyServer <- forkProcess localNode $ forever $ do +      msg <- receiveWait [ matchAny return ]+      forward msg mathServer+    putMVar proxyAddr proxyServer ++  -- Client+  forkIO $ do+    localNode <- newLocalNode transport initRemoteTable+    mathServer <- readMVar proxyAddr++    runProcess localNode $ do+      pid <- getSelfPid+      send mathServer (Add pid 1 2)+      3 <- expect :: Process Double  +      send mathServer (Divide pid 8 2)+      4 <- expect :: Process Double+      send mathServer (Divide pid 8 0)+      DivByZero <- expect+      liftIO $ putMVar clientDone ()++  takeMVar clientDone+ main :: IO () main = do-  Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters+  Right (transport, transportInternals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters   runTests      [ ("Ping",             testPing             transport)     , ("Math",             testMath             transport) @@ -593,6 +710,7 @@     , ("Registry",         testRegistry         transport)     , ("RemoteRegistry",   testRemoteRegistry   transport)     , ("SpawnLocal",       testSpawnLocal       transport)+    , ("MatchAny",         testMatchAny         transport)       -- Monitoring processes       --       -- The "missing" combinations in the list below don't make much sense, as@@ -619,4 +737,6 @@       -- Monitoring nodes and channels     , ("MonitorNode",                  testMonitorNode                transport)     , ("MonitorChannel",               testMonitorChannel             transport)+      -- Reconnect+    , ("Reconnect",                    testReconnect                  transport transportInternals)     ]

@@ -6,13 +6,28 @@ import Control.Monad (join, replicateM, forever) import Control.Exception (IOException, throw) import Control.Concurrent (forkIO, threadDelay)-import Control.Concurrent.MVar (MVar, newEmptyMVar, readMVar, takeMVar, putMVar)+import Control.Concurrent.MVar +  ( MVar+  , newEmptyMVar+  , readMVar+  , takeMVar+  , putMVar+  , modifyMVar_+  , newMVar+  ) import Control.Applicative ((<$>))+import System.Random (randomIO) import Network.Transport (Transport)-import Network.Transport.TCP (createTransport, defaultTCPParameters)+import Network.Transport.TCP +  ( createTransportExposeInternals+  , defaultTCPParameters+  , TransportInternals(socketBetween)+  )+import Network.Socket (sClose)   import Control.Distributed.Process import Control.Distributed.Process.Closure import Control.Distributed.Process.Node+import Control.Distributed.Process.Internal.Types (NodeId(nodeAddress)) import Control.Distributed.Static (staticLabel, staticClosure) import TestAuxiliary @@ -53,6 +68,9 @@   sport <- receiveChan rport   sendChan sport () +signal :: ProcessId -> Process ()+signal pid = send pid ()+ remotable [ 'factorial           , 'addInt           , 'putInt@@ -62,8 +80,30 @@           , 'typedPingServer           , 'isPrime           , 'quintuple+          , 'signal           ] +randomElement :: [a] -> IO a+randomElement xs = do+  ix <- randomIO+  return (xs !! (ix `mod` length xs))++remotableDecl [ +    [d| dfib :: ([NodeId], SendPort Integer, Integer) -> Process () ;+        dfib (_, reply, 0) = sendChan reply 0+        dfib (_, reply, 1) = sendChan reply 1+        dfib (nids, reply, n) = do+          nid1 <- liftIO $ randomElement nids+          nid2 <- liftIO $ randomElement nids+          (sport, rport) <- newChan+          spawn nid1 $ $(mkClosure 'dfib) (nids, sport, n - 2)+          spawn nid2 $ $(mkClosure 'dfib) (nids, sport, n - 1)+          n1 <- receiveChan rport+          n2 <- receiveChan rport+          sendChan reply $ n1 + n2+      |]+  ]+ -- Just try creating a static polymorphic value staticQuintuple :: (Typeable a, Typeable b, Typeable c, Typeable d, Typeable e)                 => Static (a -> b -> c -> d -> e -> (a, b, c, d, e))@@ -337,10 +377,57 @@     liftIO $ putMVar done ()   takeMVar done +simulateNetworkFailure :: TransportInternals -> NodeId -> NodeId -> Process ()+simulateNetworkFailure transportInternals fr to = liftIO $ do+  threadDelay 10000+  sock <- socketBetween transportInternals (nodeAddress fr) (nodeAddress to) +  sClose sock+  threadDelay 10000++testFib :: Transport -> RemoteTable -> IO ()+testFib transport rtable = do+  nodes <- replicateM 4 $ newLocalNode transport rtable+  done <- newEmptyMVar++  forkProcess (head nodes) $ do+    (sport, rport) <- newChan+    spawnLocal $ dfib (map localNodeId nodes, sport, 10)+    55 <- receiveChan rport :: Process Integer+    liftIO $ putMVar done ()++  takeMVar done++testSpawnReconnect :: Transport -> RemoteTable -> TransportInternals -> IO ()+testSpawnReconnect transport rtable transportInternals = do+  [node1, node2] <- replicateM 2 $ newLocalNode transport rtable +  let nid1 = localNodeId node1+      nid2 = localNodeId node2+  done <- newEmptyMVar +  iv <- newMVar (0 :: Int) ++  incr <- forkProcess node1 $ forever $ do+    () <- expect+    liftIO $ modifyMVar_ iv (return . (+ 1))++  forkProcess node2 $ do+    _pid1 <- spawn nid1 ($(mkClosure 'signal) incr) +    simulateNetworkFailure transportInternals nid2 nid1+    _pid2 <- spawn nid1 ($(mkClosure 'signal) incr) +    _pid3 <- spawn nid1 ($(mkClosure 'signal) incr) ++    liftIO $ threadDelay 100000++    count <- liftIO $ takeMVar iv+    True <- return $ count == 2 || count == 3 -- It depends on which message we get first in 'spawn' ++    liftIO $ putMVar done ()++  takeMVar done+ main :: IO () main = do-  Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters-  let rtable = __remoteTable initRemoteTable +  Right (transport, transportInternals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters+  let rtable = __remoteTable . __remoteTableDecl $ initRemoteTable    runTests      [ ("Unclosure",       testUnclosure       transport rtable)     , ("Bind",            testBind            transport rtable)@@ -356,4 +443,6 @@     , ("ClosureExpect",   testClosureExpect   transport rtable)     , ("SpawnChannel",    testSpawnChannel    transport rtable)     , ("TDict",           testTDict           transport rtable)+    , ("Fib",             testFib             transport rtable)+    , ("SpawnReconnect",  testSpawnReconnect  transport rtable transportInternals)     ]