distributed-process 0.4.0.1 → 0.4.0.2
raw patch · 5 files changed
+85/−24 lines, 5 filesPVP: minor bump suggested
API additions: PVP suggests at least a minor version bump
API changes (from Hackage documentation)
+ Control.Distributed.Process.Internal.Closure.BuiltIn: cpDelay :: ProcessId -> Closure (Process ()) -> Closure (Process ())
Files
- distributed-process.cabal +2/−2
- src/Control/Distributed/Process.hs +10/−17
- src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs +39/−0
- src/Control/Distributed/Process/Internal/Closure/TH.hs +16/−4
- tests/TestClosure.hs +18/−1
distributed-process.cabal view
@@ -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
src/Control/Distributed/Process.hs view
@@ -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 --
src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs view
@@ -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"
src/Control/Distributed/Process/Internal/Closure/TH.hs view
@@ -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)
tests/TestClosure.hs view
@@ -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) ]