464c6ba · distributed-process

git.iko.soy/distributed-process

on master help sign in

Deal with spawning invalid closures

464c6baEdsko de Vries14 years ago

distributed-process/src/Control/Distributed/Process.hs+4

compare with master

unifiedsplit

@@ -59,6 +59,8 @@ module Control.Distributed.Process

59 , expectTimeout

60 , spawnAsync

61 , spawnSupervised

62+ , spawnLink

63+ , spawnMonitor

62 , DidSpawn(..)

64 , DidSpawn(..)

63 ) where

65 ) where

@@ -127,6 +129,8 @@ import Control.Distributed.Process.Internal.Process.Primitives

127 , catch

129 , catch

128 , expectTimeout

130 , expectTimeout

129 , spawnAsync

131 , spawnAsync

132+ , spawnLink

133+ , spawnMonitor

130 )

134 )

131import Control.Distributed.Process.Internal.Closure (resolveClosure)

135import Control.Distributed.Process.Internal.Closure (resolveClosure)

132

136

distributed-process/src/Control/Distributed/Process/Internal/Process/Primitives.hs+20

@@ -34,6 +34,8 @@ module Control.Distributed.Process.Internal.Process.Primitives

34 , catch

35 , expectTimeout

36 , spawnAsync

37+ , spawnLink

38+ , spawnMonitor

37 ) where

39 ) where

39import Prelude hiding (catch)

41import Prelude hiding (catch)

@@ -289,6 +291,24 @@ spawnAsync nid proc = do

289 sendCtrlMsg (Just nid) $ Spawn proc spawnRef

291 sendCtrlMsg (Just nid) $ Spawn proc spawnRef

290 return spawnRef

292 return spawnRef

291

293

294+-- | Spawn a process and link to it

295+--

296+-- Note that this is just the sequential composition of 'spawn' and 'link'.

297+-- (The "Unified" semantics that underlies Cloud Haskell does not even support

298+-- a synchronous link operation)

299+spawnLink :: NodeId -> Closure (Process ()) -> Process ProcessId

300+spawnLink nid proc = do

301+ pid <- spawn nid proc

302+ link pid

303+ return pid

304+

305+-- | Like 'spawnLink', but monitor the spawned process

306+spawnMonitor :: NodeId -> Closure (Process ()) -> Process (ProcessId, MonitorRef)

307+spawnMonitor nid proc = do

308+ pid <- spawn nid proc

309+ ref <- monitor pid

310+ return (pid, ref)

311+

292--------------------------------------------------------------------------------

312--------------------------------------------------------------------------------

293-- Auxiliary functions --

313-- Auxiliary functions --

294--------------------------------------------------------------------------------

314--------------------------------------------------------------------------------

distributed-process/src/Control/Distributed/Process/Node.hs+12−16

@@ -18,7 +18,7 @@ import qualified Data.List as List (delete)

18import Data.Set (Set)

19import qualified Data.Set as Set (empty, insert, delete, member)

20import Data.Foldable (forM_)

21−import Data.Maybe (isJust)

21+import Data.Maybe (isJust, fromMaybe)

22import Data.Typeable (Typeable)

23import Control.Category ((>>>))

24import Control.Applicative ((<$>))

@@ -450,15 +450,12 @@ ncEffectProcessDied pid reason = do

450ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()

451ncEffectSpawn pid cProc ref = do

452 mProc <- unClosure cProc

453− case mProc of

453+ -- If the closure does not exist, we spawn a process that throws an exception

454− Just proc -> do

454+ -- This allows the remote node to find out what's happening

455+ let proc = fromMaybe (fail $ "Error: unknown closure " ++ show cProc) mProc

455− node <- ncMsg getLocalNode

456+ node <- ncMsg getLocalNode

456− pid' <- liftIO $ forkProcess node proc

457+ pid' <- liftIO $ forkProcess node proc

457− ncMsg $ sendMessage (ProcessIdentifier pid) (DidSpawn ref pid')

458+ ncMsg $ sendMessage (ProcessIdentifier pid) (DidSpawn ref pid')

458− Nothing ->

459− -- TODO: what should we do when unClosure returns Nothing?

460− liftIO . putStrLn $ "Error: closure " ++ show cProc ++ " not found"

461−

462

459

463--------------------------------------------------------------------------------

460--------------------------------------------------------------------------------

464-- Auxiliary --

461-- Auxiliary --

@@ -494,12 +491,11 @@ destNid (Monitor pid _) =

494 Just . processNodeId $ pid

491 Just . processNodeId $ pid

495destNid (Unmonitor ref) =

492destNid (Unmonitor ref) =

496 Just . processNodeId . monitorRefPid $ ref

493 Just . processNodeId . monitorRefPid $ ref

497−destNid (Died (NodeIdentifier _) _) =

494+-- We don't need to forward 'Died' signals; if monitoring/linking is setup,

495+-- then when a local process dies the monitoring/linking machinery will take

496+-- care of notifying remote nodes

497+destNid (Died _ _) =

498− Nothing

498+ Nothing

499−destNid (Died (ProcessIdentifier _pid) _) =

500− fail "destNid: TODO"

501−destNid (Died (ChannelIdentifier _cid) _) =

502− fail "destNid: TODO"

503destNid (Spawn _ _) =

499destNid (Spawn _ _) =

504 Nothing

500 Nothing

505

501

distributed-process/tests/TestClosure.hs+13

@@ -1,5 +1,6 @@

1module Main where

3+import qualified Data.ByteString.Lazy as BSL (empty)

3import Control.Monad (join, replicateM)

4import Control.Monad (join, replicateM)

4import Control.Monad.IO.Class (liftIO)

5import Control.Monad.IO.Class (liftIO)

5import Control.Exception (IOException, throw)

6import Control.Exception (IOException, throw)

@@ -11,6 +12,7 @@ import Network.Transport.TCP (createTransport, defaultTCPParameters)

11import Control.Distributed.Process

12import Control.Distributed.Process

12import Control.Distributed.Process.Closure

13import Control.Distributed.Process.Closure

13import Control.Distributed.Process.Node

14import Control.Distributed.Process.Node

15+import Control.Distributed.Process.Internal.Types (Closure(..), Static(..), StaticLabel(..))

14import TestAuxiliary

16import TestAuxiliary

16serializableDictInt :: SerializableDict Int

18serializableDictInt :: SerializableDict Int

@@ -244,6 +246,16 @@ testSpawnSupervised transport rtable = do

244 supervisorDeath :: IOException

246 supervisorDeath :: IOException

245 supervisorDeath = userError "Supervisor died"

247 supervisorDeath = userError "Supervisor died"

246

248

249+testSpawnInvalid :: Transport -> RemoteTable -> IO ()

250+testSpawnInvalid transport rtable = do

251+ node <- newLocalNode transport rtable

252+ runProcess node $ do

253+ (pid, ref) <- spawnMonitor (localNodeId node) (Closure (Static (UserStatic "ThisDoesNotExist")) BSL.empty)

254+ MonitorNotification ref' pid' _reason <- expect

255+ -- Depending on the exact interleaving, reason might be NoProc or the exception thrown by the absence of the static closure

256+ True <- return $ ref' == ref && pid == pid'

257+ return ()

258+

247main :: IO ()

259main :: IO ()

248main = do

260main = do

249 Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters

261 Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters

@@ -259,4 +271,5 @@ main = do

259 , ("Seq", testSeq transport rtable)

271 , ("Seq", testSeq transport rtable)

260 , ("Apply", testApply transport rtable)

272 , ("Apply", testApply transport rtable)

261 , ("SpawnSupervised", testSpawnSupervised transport rtable)

273 , ("SpawnSupervised", testSpawnSupervised transport rtable)

274+ , ("SpawnInvalid", testSpawnInvalid transport rtable)

262 ]

275 ]