diff --git a/distributed-process.cabal b/distributed-process.cabal --- a/distributed-process.cabal +++ b/distributed-process.cabal @@ -1,12 +1,12 @@ Name: distributed-process -Version: 0.4.0.1 +Version: 0.4.0.2 Cabal-Version: >=1.8 Build-Type: Simple License: BSD3 License-File: LICENSE Copyright: Well-Typed LLP Author: Duncan Coutts, Nicolas Wu, Edsko de Vries -Maintainer: edsko@well-typed.com, dcoutts@well-typed.com +Maintainer: edsko@well-typed.com, duncan@well-typed.com Stability: experimental Homepage: http://github.com/haskell-distributed/distributed-process Bug-Reports: mailto:edsko@well-typed.com diff --git a/src/Control/Distributed/Process.hs b/src/Control/Distributed/Process.hs --- a/src/Control/Distributed/Process.hs +++ b/src/Control/Distributed/Process.hs @@ -146,18 +146,16 @@ ) import Control.Distributed.Process.Serializable (Serializable, SerializableDict) import Control.Distributed.Process.Internal.Closure.BuiltIn - ( sdictUnit - , sdictSendPort + ( sdictSendPort , sndStatic , idCP , seqCP , bindCP , splitCP , cpLink - , cpUnlink - , cpExpect , cpSend , cpNewChan + , cpDelay ) import Control.Distributed.Process.Internal.Primitives ( -- Basic messaging @@ -298,20 +296,15 @@ spawn nid proc = do us <- getSelfPid mRef <- monitorNode nid - sRef <- spawnAsync nid $ cpLink us - `seqCP` cpExpect sdictUnit - `seqCP` cpUnlink us - `seqCP` proc - mPid <- receiveWait - [ matchIf (\(DidSpawn ref _) -> ref == sRef) - (\(DidSpawn _ pid) -> return $ Right pid) - , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef) - (\(NodeMonitorNotification _ _ err) -> return $ Left err) + sRef <- spawnAsync nid (cpDelay us proc) + receiveWait [ + matchIf (\(DidSpawn ref _) -> ref == sRef) $ \(DidSpawn _ pid) -> do + unmonitor mRef + send pid () + return pid + , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef) $ \_ -> + return (nullProcessId nid) ] - unmonitor mRef - case mPid of - Left _err -> return (nullProcessId nid) - Right pid -> send pid () >> return pid -- | Spawn a process and link to it -- diff --git a/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs b/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs --- a/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs +++ b/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs @@ -21,6 +21,8 @@ , cpSend , cpExpect , cpNewChan + -- * Support for some CH operations + , cpDelay ) where import Data.ByteString.Lazy (ByteString) @@ -49,6 +51,7 @@ , ProcessId , SendPort , ReceivePort + , ProcessMonitorNotification(ProcessMonitorNotification) ) import Control.Distributed.Process.Internal.Primitives ( link @@ -56,6 +59,11 @@ , send , expect , newChan + , monitor + , unmonitor + , match + , matchIf + , receiveWait ) -------------------------------------------------------------------------------- @@ -79,6 +87,7 @@ . registerStatic "$newChanDict" (toDynamic (newChanDict :: SerializableDict ANY -> Process (SendPort ANY, ReceivePort ANY))) . registerStatic "$cpSplit" (toDynamic (cpSplit :: (ANY1 -> Process ANY3) -> (ANY2 -> Process ANY4) -> (ANY1, ANY2) -> Process (ANY3, ANY4))) . registerStatic "$snd" (toDynamic (snd :: (ANY1, ANY2) -> ANY2)) + . registerStatic "$delay" (toDynamic delay) where decodeDict :: forall a. SerializableDict a -> ByteString -> a decodeDict SerializableDict = decode @@ -236,3 +245,33 @@ newChanDictStatic :: Typeable a => Static (SerializableDict a -> Process (SendPort a, ReceivePort a)) newChanDictStatic = staticLabel "$newChanDict" + +-------------------------------------------------------------------------------- +-- Support for spawn -- +-------------------------------------------------------------------------------- + +-- | @delay them p@ is a process that waits for a signal (a message of type @()@) +-- from 'them' (origin is not verified) before proceeding as @p@. In order to +-- avoid waiting forever, @delay them p@ monitors 'them'. If it receives a +-- monitor message instead it simply terminates. +delay :: ProcessId -> Process () -> Process () +delay them p = do + ref <- monitor them + let sameRef (ProcessMonitorNotification ref' _ _) = ref == ref' + receiveWait [ + match $ \() -> unmonitor ref >> p + , matchIf sameRef $ \_ -> return () + ] + +-- | 'CP' version of 'delay' +cpDelay :: ProcessId -> Closure (Process ()) -> Closure (Process ()) +cpDelay = closureApply . cpDelay' + where + cpDelay' :: ProcessId -> Closure (Process () -> Process ()) + cpDelay' pid = closure decoder (encode pid) + + decoder :: Static (ByteString -> Process () -> Process ()) + decoder = delayStatic `staticCompose` decodeProcessIdStatic + + delayStatic :: Static (ProcessId -> Process () -> Process ()) + delayStatic = staticLabel "$delay" diff --git a/src/Control/Distributed/Process/Internal/Closure/TH.hs b/src/Control/Distributed/Process/Internal/Closure/TH.hs --- a/src/Control/Distributed/Process/Internal/Closure/TH.hs +++ b/src/Control/Distributed/Process/Internal/Closure/TH.hs @@ -16,6 +16,8 @@ ( -- Q monad and operations Q , reify + , Loc(loc_module) + , location -- Names , Name , mkName @@ -192,7 +194,7 @@ [] -> [| toDynamic $(varE origName) |] _ -> [| toDynamic ($(varE origName) :: $(monomorphize typVars typ)) |] return ( static - , [ [| registerStatic $(stringE (show origName)) $dyn |] ] + , [ [| registerStatic $(showFQN origName) $dyn |] ] ) makeDict :: Name -> Type -> Q ([Dec], [Q Exp]) @@ -200,7 +202,7 @@ sdict <- generateDict dictName typ let dyn = [| toDynamic (SerializableDict :: SerializableDict $(return typ)) |] return ( sdict - , [ [| registerStatic $(stringE (show dictName)) $dyn |] ] + , [ [| registerStatic $(showFQN dictName) $dyn |] ] ) -- | Turn a polymorphic type into a monomorphic type using ANY and co @@ -238,7 +240,7 @@ (map typeable xs) (staticTyp `AppT` typ) ) - , sfnD (staticName n) [| staticLabel $(stringE (show n)) |] + , sfnD (staticName n) [| staticLabel $(showFQN n) |] ] where typeable :: TyVarBndr -> Pred @@ -249,7 +251,7 @@ generateDict n typ = do sequence [ sigD n $ [t| Static (SerializableDict $(return typ)) |] - , sfnD n [| staticLabel $(stringE (show n)) |] + , sfnD n [| staticLabel $(showFQN n) |] ] staticName :: Name -> Name @@ -291,3 +293,13 @@ tyVarBndrName :: TyVarBndr -> Name tyVarBndrName (PlainTV n) = n tyVarBndrName (KindedTV n _) = n + +-- | Fully qualified name; that is, the name and the _current_ module +-- +-- We ignore the module part of the Name argument (which may or may not exist) +-- because we construct various names (`staticName`, `sdictName`, `tdictName`) +-- and those names certainly won't have Module components. +showFQN :: Name -> Q Exp +showFQN n = do + loc <- location + stringE (loc_module loc ++ "." ++ nameBase n) diff --git a/tests/TestClosure.hs b/tests/TestClosure.hs --- a/tests/TestClosure.hs +++ b/tests/TestClosure.hs @@ -3,7 +3,7 @@ import Data.ByteString.Lazy (empty) import Data.Typeable (Typeable) -import Control.Monad (join, replicateM, forever) +import Control.Monad (join, replicateM, forever, replicateM_, void) import Control.Exception (IOException, throw) import Control.Concurrent (forkIO, threadDelay) import Control.Concurrent.MVar @@ -424,6 +424,22 @@ takeMVar done +-- | 'spawn' used to ave a race condition which would be triggered if the +-- spawning process terminates immediately after spawning +testSpawnTerminate :: Transport -> RemoteTable -> IO () +testSpawnTerminate transport rtable = do + slave <- newLocalNode transport rtable + master <- newLocalNode transport rtable + masterDone <- newEmptyMVar + + runProcess master $ do + us <- getSelfPid + replicateM_ 1000 . spawnLocal . void . spawn (localNodeId slave) $ $(mkClosure 'signal) us + replicateM_ 1000 $ (expect :: Process ()) + liftIO $ putMVar masterDone () + + takeMVar masterDone + main :: IO () main = do Right (transport, transportInternals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters @@ -444,5 +460,6 @@ , ("SpawnChannel", testSpawnChannel transport rtable) , ("TDict", testTDict transport rtable) , ("Fib", testFib transport rtable) + , ("SpawnTerminate", testSpawnTerminate transport rtable) , ("SpawnReconnect", testSpawnReconnect transport rtable transportInternals) ]