git.iko.soy/distributed-process
on master help sign in
Partial fix for race condition in spawn

This fix is "partial" because it relies on the strong ordering guarantees provided by the TCP transport; with other transports it may still break.

commit5679927aa9bccced3af8c479e089859fa589fcd3 authorEdsko de Vries <edsko@edsko.net> dateTue 23 Oct 2012, 11:21 (14 years ago) parent5c263cf
3 files changed+67−18
compare with master
unifiedsplit
distributed-process/src/Control/Distributed/Process.hs+10−17 distributed-process/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs+39 distributed-process/tests/TestClosure.hs+18−1
distributed-process/src/Control/Distributed/Process.hs+10−17
@@ -146,18 +146,16 @@ import Control.Distributed.Process.Internal.Types
146146 )
147147import Control.Distributed.Process.Serializable (Serializable, SerializableDict)
148148import Control.Distributed.Process.Internal.Closure.BuiltIn
149− ( sdictUnit
150− , sdictSendPort
149+ ( sdictSendPort
151150 , sndStatic
152151 , idCP
153152 , seqCP
154153 , bindCP
155154 , splitCP
156155 , cpLink
157− , cpUnlink
158− , cpExpect
159156 , cpSend
160157 , cpNewChan
158+ , cpDelay
161159 )
162160import Control.Distributed.Process.Internal.Primitives
163161 ( -- Basic messaging
@@ -298,20 +296,15 @@ spawn :: NodeId -> Closure (Process ()) -> Process ProcessId
298296spawn nid proc = do
299297 us <- getSelfPid
300298 mRef <- monitorNode nid
301− sRef <- spawnAsync nid $ cpLink us
302− `seqCP` cpExpect sdictUnit
303− `seqCP` cpUnlink us
304− `seqCP` proc
305− mPid <- receiveWait
306− [ matchIf (\(DidSpawn ref _) -> ref == sRef)
307− (\(DidSpawn _ pid) -> return $ Right pid)
308− , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)
309− (\(NodeMonitorNotification _ _ err) -> return $ Left err)
299+ sRef <- spawnAsync nid (cpDelay us proc)
300+ receiveWait [
301+ matchIf (\(DidSpawn ref _) -> ref == sRef) $ \(DidSpawn _ pid) -> do
302+ unmonitor mRef
303+ send pid ()
304+ return pid
305+ , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef) $ \_ ->
306+ return (nullProcessId nid)
310307 ]
311− unmonitor mRef
312− case mPid of
313− Left _err -> return (nullProcessId nid)
314− Right pid -> send pid () >> return pid
315308
316309-- | Spawn a process and link to it
317310--
distributed-process/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs+39
@@ -21,6 +21,8 @@ module Control.Distributed.Process.Internal.Closure.BuiltIn
2121 , cpSend
2222 , cpExpect
2323 , cpNewChan
24+ -- * Support for some CH operations
25+ , cpDelay
2426 ) where
2527
2628import Data.ByteString.Lazy (ByteString)
@@ -49,6 +51,7 @@ import Control.Distributed.Process.Internal.Types
4951 , ProcessId
5052 , SendPort
5153 , ReceivePort
54+ , ProcessMonitorNotification(ProcessMonitorNotification)
5255 )
5356import Control.Distributed.Process.Internal.Primitives
5457 ( link
@@ -56,6 +59,11 @@ import Control.Distributed.Process.Internal.Primitives
5659 , send
5760 , expect
5861 , newChan
62+ , monitor
63+ , unmonitor
64+ , match
65+ , matchIf
66+ , receiveWait
5967 )
6068
6169--------------------------------------------------------------------------------
@@ -79,6 +87,7 @@ remoteTable =
7987 . registerStatic "$newChanDict" (toDynamic (newChanDict :: SerializableDict ANY -> Process (SendPort ANY, ReceivePort ANY)))
8088 . registerStatic "$cpSplit" (toDynamic (cpSplit :: (ANY1 -> Process ANY3) -> (ANY2 -> Process ANY4) -> (ANY1, ANY2) -> Process (ANY3, ANY4)))
8189 . registerStatic "$snd" (toDynamic (snd :: (ANY1, ANY2) -> ANY2))
90+ . registerStatic "$delay" (toDynamic delay)
8291 where
8392 decodeDict :: forall a. SerializableDict a -> ByteString -> a
8493 decodeDict SerializableDict = decode
@@ -236,3 +245,33 @@ cpNewChan dict = staticClosure (newChanDictStatic `staticApply` dict)
236245 newChanDictStatic :: Typeable a
237246 => Static (SerializableDict a -> Process (SendPort a, ReceivePort a))
238247 newChanDictStatic = staticLabel "$newChanDict"
248+
249+--------------------------------------------------------------------------------
250+-- Support for spawn --
251+--------------------------------------------------------------------------------
252+
253+-- | @delay them p@ is a process that waits for a signal (a message of type @()@)
254+-- from 'them' (origin is not verified) before proceeding as @p@. In order to
255+-- avoid waiting forever, @delay them p@ monitors 'them'. If it receives a
256+-- monitor message instead it simply terminates.
257+delay :: ProcessId -> Process () -> Process ()
258+delay them p = do
259+ ref <- monitor them
260+ let sameRef (ProcessMonitorNotification ref' _ _) = ref == ref'
261+ receiveWait [
262+ match $ \() -> unmonitor ref >> p
263+ , matchIf sameRef $ \_ -> return ()
264+ ]
265+
266+-- | 'CP' version of 'delay'
267+cpDelay :: ProcessId -> Closure (Process ()) -> Closure (Process ())
268+cpDelay = closureApply . cpDelay'
269+ where
270+ cpDelay' :: ProcessId -> Closure (Process () -> Process ())
271+ cpDelay' pid = closure decoder (encode pid)
272+
273+ decoder :: Static (ByteString -> Process () -> Process ())
274+ decoder = delayStatic `staticCompose` decodeProcessIdStatic
275+
276+ delayStatic :: Static (ProcessId -> Process () -> Process ())
277+ delayStatic = staticLabel "$delay"
distributed-process/tests/TestClosure.hs+18−1
@@ -3,7 +3,7 @@ module Main where
33
44import Data.ByteString.Lazy (empty)
55import Data.Typeable (Typeable)
6−import Control.Monad (join, replicateM, forever)
6+import Control.Monad (join, replicateM, forever, replicateM_, void)
77import Control.Exception (IOException, throw)
88import Control.Concurrent (forkIO, threadDelay)
99import Control.Concurrent.MVar
@@ -424,6 +424,22 @@ testSpawnReconnect transport rtable transportInternals = do
424424
425425 takeMVar done
426426
427+-- | 'spawn' used to ave a race condition which would be triggered if the
428+-- spawning process terminates immediately after spawning
429+testSpawnTerminate :: Transport -> RemoteTable -> IO ()
430+testSpawnTerminate transport rtable = do
431+ slave <- newLocalNode transport rtable
432+ master <- newLocalNode transport rtable
433+ masterDone <- newEmptyMVar
434+
435+ runProcess master $ do
436+ us <- getSelfPid
437+ replicateM_ 1000 . spawnLocal . void . spawn (localNodeId slave) $ $(mkClosure 'signal) us
438+ replicateM_ 1000 $ (expect :: Process ())
439+ liftIO $ putMVar masterDone ()
440+
441+ takeMVar masterDone
442+
427443main :: IO ()
428444main = do
429445 Right (transport, transportInternals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters
@@ -444,5 +460,6 @@ main = do
444460 , ("SpawnChannel", testSpawnChannel transport rtable)
445461 , ("TDict", testTDict transport rtable)
446462 , ("Fib", testFib transport rtable)
463+ , ("SpawnTerminate", testSpawnTerminate transport rtable)
447464 , ("SpawnReconnect", testSpawnReconnect transport rtable transportInternals)
448465 ]