5679927 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Partial fix for race condition in spawn

5679927Edsko de Vries14 years ago

distributed-process/src/Control/Distributed/Process.hs+10−17

compare with master

unifiedsplit

@@ -146,18 +146,16 @@ import Control.Distributed.Process.Internal.Types

146 )

147import Control.Distributed.Process.Serializable (Serializable, SerializableDict)

148import Control.Distributed.Process.Internal.Closure.BuiltIn

149− ( sdictUnit

150− , sdictSendPort

149+ ( sdictSendPort

151 , sndStatic

150 , sndStatic

152 , idCP

151 , idCP

153 , seqCP

152 , seqCP

154 , bindCP

153 , bindCP

155 , splitCP

154 , splitCP

156 , cpLink

155 , cpLink

157− , cpUnlink

158− , cpExpect

159 , cpSend

156 , cpSend

160 , cpNewChan

157 , cpNewChan

158+ , cpDelay

161 )

159 )

162import Control.Distributed.Process.Internal.Primitives

160import Control.Distributed.Process.Internal.Primitives

163 ( -- Basic messaging

161 ( -- Basic messaging

@@ -298,20 +296,15 @@ spawn :: NodeId -> Closure (Process ()) -> Process ProcessId

298spawn nid proc = do

296spawn nid proc = do

299 us <- getSelfPid

297 us <- getSelfPid

300 mRef <- monitorNode nid

298 mRef <- monitorNode nid

301− sRef <- spawnAsync nid $ cpLink us

299+ sRef <- spawnAsync nid (cpDelay us proc)

302− `seqCP` cpExpect sdictUnit

303− `seqCP` cpUnlink us

304− `seqCP` proc

305− mPid <- receiveWait

300+ receiveWait [

306− [ matchIf (\(DidSpawn ref _) -> ref == sRef)

301+ matchIf (\(DidSpawn ref _) -> ref == sRef) $ \(DidSpawn _ pid) -> do

307− (\(DidSpawn _ pid) -> return $ Right pid)

302+ unmonitor mRef

303+ send pid ()

304+ return pid

308− , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)

305+ , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef) $ \_ ->

309− (\(NodeMonitorNotification _ _ err) -> return $ Left err)

306+ return (nullProcessId nid)

310 ]

307 ]

311− unmonitor mRef

312− case mPid of

313− Left _err -> return (nullProcessId nid)

314− Right pid -> send pid () >> return pid

315

308

316-- | Spawn a process and link to it

309-- | Spawn a process and link to it

317--

310--

distributed-process/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs+39

@@ -21,6 +21,8 @@ module Control.Distributed.Process.Internal.Closure.BuiltIn

21 , cpSend

22 , cpExpect

23 , cpNewChan

24+ -- * Support for some CH operations

25+ , cpDelay

24 ) where

26 ) where

26import Data.ByteString.Lazy (ByteString)

28import Data.ByteString.Lazy (ByteString)

@@ -49,6 +51,7 @@ import Control.Distributed.Process.Internal.Types

49 , ProcessId

51 , ProcessId

50 , SendPort

52 , SendPort

51 , ReceivePort

53 , ReceivePort

54+ , ProcessMonitorNotification(ProcessMonitorNotification)

52 )

55 )

53import Control.Distributed.Process.Internal.Primitives

56import Control.Distributed.Process.Internal.Primitives

54 ( link

57 ( link

@@ -56,6 +59,11 @@ import Control.Distributed.Process.Internal.Primitives

56 , send

59 , send

57 , expect

60 , expect

58 , newChan

61 , newChan

62+ , monitor

63+ , unmonitor

64+ , match

65+ , matchIf

66+ , receiveWait

59 )

67 )

61--------------------------------------------------------------------------------

69--------------------------------------------------------------------------------

@@ -79,6 +87,7 @@ remoteTable =

79 . registerStatic "$newChanDict" (toDynamic (newChanDict :: SerializableDict ANY -> Process (SendPort ANY, ReceivePort ANY)))

87 . registerStatic "$newChanDict" (toDynamic (newChanDict :: SerializableDict ANY -> Process (SendPort ANY, ReceivePort ANY)))

80 . registerStatic "$cpSplit" (toDynamic (cpSplit :: (ANY1 -> Process ANY3) -> (ANY2 -> Process ANY4) -> (ANY1, ANY2) -> Process (ANY3, ANY4)))

88 . registerStatic "$cpSplit" (toDynamic (cpSplit :: (ANY1 -> Process ANY3) -> (ANY2 -> Process ANY4) -> (ANY1, ANY2) -> Process (ANY3, ANY4)))

81 . registerStatic "$snd" (toDynamic (snd :: (ANY1, ANY2) -> ANY2))

89 . registerStatic "$snd" (toDynamic (snd :: (ANY1, ANY2) -> ANY2))

90+ . registerStatic "$delay" (toDynamic delay)

82 where

91 where

83 decodeDict :: forall a. SerializableDict a -> ByteString -> a

92 decodeDict :: forall a. SerializableDict a -> ByteString -> a

84 decodeDict SerializableDict = decode

93 decodeDict SerializableDict = decode

@@ -236,3 +245,33 @@ cpNewChan dict = staticClosure (newChanDictStatic `staticApply` dict)

236 newChanDictStatic :: Typeable a

245 newChanDictStatic :: Typeable a

237 => Static (SerializableDict a -> Process (SendPort a, ReceivePort a))

246 => Static (SerializableDict a -> Process (SendPort a, ReceivePort a))

238 newChanDictStatic = staticLabel "$newChanDict"

247 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

4import Data.ByteString.Lazy (empty)

5import Data.Typeable (Typeable)

6−import Control.Monad (join, replicateM, forever)

6+import Control.Monad (join, replicateM, forever, replicateM_, void)

7import Control.Exception (IOException, throw)

8import Control.Concurrent (forkIO, threadDelay)

9import Control.Concurrent.MVar

@@ -424,6 +424,22 @@ testSpawnReconnect transport rtable transportInternals = do

424

425 takeMVar done

426

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+

427main :: IO ()

443main :: IO ()

428main = do

444main = do

429 Right (transport, transportInternals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters

445 Right (transport, transportInternals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters

@@ -444,5 +460,6 @@ main = do

444 , ("SpawnChannel", testSpawnChannel transport rtable)

460 , ("SpawnChannel", testSpawnChannel transport rtable)

445 , ("TDict", testTDict transport rtable)

461 , ("TDict", testTDict transport rtable)

446 , ("Fib", testFib transport rtable)

462 , ("Fib", testFib transport rtable)

463+ , ("SpawnTerminate", testSpawnTerminate transport rtable)

447 , ("SpawnReconnect", testSpawnReconnect transport rtable transportInternals)

464 , ("SpawnReconnect", testSpawnReconnect transport rtable transportInternals)

448 ]

465 ]