d31ba89 · distributed-process

git.iko.soy/distributed-process

on master help sign in

rebased changes against latest from upstream

d31ba89RodLogic14 years ago

distributed-process/src/Control/Distributed/Process.hs+7−1

compare with master

unifiedsplit

@@ -43,6 +43,9 @@ module Control.Distributed.Process

43 , spawn

44 , call

45 , terminate

46+ , kill

47+ , exit

48+ , catchExit

46 , ProcessTerminationException(..)

49 , ProcessTerminationException(..)

47 , ProcessRegistrationException(..)

50 , ProcessRegistrationException(..)

48 , SpawnRef

51 , SpawnRef

@@ -184,6 +187,9 @@ import Control.Distributed.Process.Internal.Primitives

184 -- Process management

187 -- Process management

185 , terminate

188 , terminate

186 , ProcessTerminationException(..)

189 , ProcessTerminationException(..)

190+ , exit

191+ , catchExit

192+ , kill

187 , getSelfPid

193 , getSelfPid

188 , getSelfNode

194 , getSelfNode

189 -- Monitoring and linking

195 -- Monitoring and linking

@@ -414,7 +420,7 @@ spawnChannelLocal proc = do

414 node <- processNode <$> ask

420 node <- processNode <$> ask

415 liftIO $ do

421 liftIO $ do

416 mvar <- newEmptyMVar

422 mvar <- newEmptyMVar

417− forkProcess node $ do

423+ _ <- forkProcess node $ do

418 -- It is important that we allocate the new channel in the new process,

424 -- It is important that we allocate the new channel in the new process,

419 -- because otherwise it will be associated with the wrong process ID

425 -- because otherwise it will be associated with the wrong process ID

420 (sport, rport) <- newChan

426 (sport, rport) <- newChan

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

@@ -24,6 +24,11 @@ module Control.Distributed.Process.Internal.Primitives

24 -- * Process management

25 , terminate

26 , ProcessTerminationException(..)

27+ , kill

28+ , ProcessKillException(..)

29+ , exit

30+ , catchExit

31+ , ProcessExitException(..)

27 , getSelfPid

32 , getSelfPid

28 , getSelfNode

33 , getSelfNode

29 -- * Monitoring and linking

34 -- * Monitoring and linking

@@ -328,6 +333,65 @@ instance Exception ProcessTerminationException

328terminate :: Process a

333terminate :: Process a

329terminate = liftIO $ throwIO ProcessTerminationException

334terminate = liftIO $ throwIO ProcessTerminationException

330

335

336+

337+-- | Forceful request to kill a process

338+kill :: ProcessId -> String -> Process ()

339+kill them reason = do

340+ mynid <- getSelfNode

341+ let nid = processNodeId them

342+ if mynid == nid

343+ then sendCtrlMsg Nothing (Kill them reason)

344+ else sendCtrlMsg (Just nid) (Kill them reason)

345+

346+-- | Thrown by 'kill'

347+newtype ProcessKillException =

348+ ProcessKillException String

349+ deriving (Typeable)

350+

351+instance Exception ProcessKillException

352+instance Show ProcessKillException where

353+ show (ProcessKillException reason) = reason

354+

355+

356+-- | Graceful request to exit a process

357+exit :: Serializable a => ProcessId -> a -> Process ()

358+exit them reason = do

359+ mynid <- getSelfNode

360+ if mynid == nid

361+ then sendCtrlMsg Nothing msg

362+ else sendCtrlMsg (Just nid) msg

363+ where

364+ nid = processNodeId them

365+ msg = Exit them (createMessage reason)

366+

367+

368+-- | Internal exception thrown indirectly by 'exit'

369+data ProcessExitException =

370+ ProcessExitException !ProcessId !Message

371+ deriving Typeable

372+

373+instance Exception ProcessExitException

374+instance Show ProcessExitException where

375+ show (ProcessExitException _ reason) = show reason

376+

377+-- | Catches ProcessExitException

378+catchExit :: forall a b . (Show a, Serializable a) => Process b -> (ProcessId -> a -> Process ()) -> Process b

379+catchExit act exitHandler = catch act handleExit

380+ where

381+ handleExit ex@(ProcessExitException from msg) =

382+ if messageFingerprint msg == fingerprint (undefined :: a)

383+ then do

384+ let reason = decoded

385+ exitHandler from reason

386+ liftIO . throwIO $ ProcessKillException (show reason)

387+ else liftIO $ throwIO ex

388+ where

389+ decoded :: a

390+ -- Make sure the value is fully decoded so that we don't hang to

391+ -- bytestrings when the process calling 'matchIf' doesn't process

392+ -- the values immediately

393+ !decoded = decode (messageEncoding msg)

394+

331-- | Our own process ID

395-- | Our own process ID

332getSelfPid :: Process ProcessId

396getSelfPid :: Process ProcessId

333getSelfPid = processId <$> ask

397getSelfPid = processId <$> ask

distributed-process/src/Control/Distributed/Process/Internal/Types.hs+6

@@ -442,6 +442,8 @@ data ProcessSignal =

442 | WhereIs !String

443 | Register !String !NodeId !(Maybe ProcessId) !Bool -- Use 'Nothing' to unregister, use True to force reregister

444 | NamedSend !String !Message

445+ | Kill !ProcessId !String

446+ | Exit !ProcessId !Message

445 deriving Show

447 deriving Show

446

448

447--------------------------------------------------------------------------------

449--------------------------------------------------------------------------------

@@ -486,6 +488,8 @@ instance Binary ProcessSignal where

486 put (WhereIs label) = putWord8 6 >> put label

488 put (WhereIs label) = putWord8 6 >> put label

487 put (Register label nid pid force) = putWord8 7 >> put label >> put nid >> put pid >> put force

489 put (Register label nid pid force) = putWord8 7 >> put label >> put nid >> put pid >> put force

488 put (NamedSend label msg) = putWord8 8 >> put label >> put (messageToPayload msg)

490 put (NamedSend label msg) = putWord8 8 >> put label >> put (messageToPayload msg)

491+ put (Kill pid reason) = putWord8 9 >> put pid >> put reason

492+ put (Exit pid reason) = putWord8 10 >> put pid >> put (messageToPayload reason)

489 get = do

493 get = do

490 header <- getWord8

494 header <- getWord8

491 case header of

495 case header of

@@ -498,6 +502,8 @@ instance Binary ProcessSignal where

498 6 -> WhereIs <$> get

502 6 -> WhereIs <$> get

499 7 -> Register <$> get <*> get <*> get <*> get

503 7 -> Register <$> get <*> get <*> get <*> get

500 8 -> NamedSend <$> get <*> (payloadToMessage <$> get)

504 8 -> NamedSend <$> get <*> (payloadToMessage <$> get)

505+ 9 -> Kill <$> get <*> get

506+ 10 -> Exit <$> get <*> (payloadToMessage <$> get)

501 _ -> fail "ProcessSignal.get: invalid"

507 _ -> fail "ProcessSignal.get: invalid"

502

508

503instance Binary DiedReason where

509instance Binary DiedReason where

distributed-process/src/Control/Distributed/Process/Node.hs+24−1

@@ -138,7 +138,14 @@ import Control.Distributed.Process.Internal.Messaging

138 , impliesDeathOf

139 )

140import Control.Distributed.Process.Internal.Primitives

141+ ( register

142+ , finally

141− (register, finally, receiveWait, match, sendChan)

143+ , receiveWait

144+ , match

145+ , sendChan

146+ , ProcessKillException(..)

147+ , ProcessExitException(..)

148+ )

142import Control.Distributed.Process.Internal.Types

149import Control.Distributed.Process.Internal.Types

143 (SendPort)

150 (SendPort)

144import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)

151import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)

@@ -506,6 +513,10 @@ nodeController = do

506 ncEffectWhereIs from label

513 ncEffectWhereIs from label

507 NCMsg from (NamedSend label msg') ->

514 NCMsg from (NamedSend label msg') ->

508 ncEffectNamedSend from label msg'

515 ncEffectNamedSend from label msg'

516+ NCMsg (ProcessIdentifier from) (Kill to reason) ->

517+ ncEffectKill from to reason

518+ NCMsg (ProcessIdentifier from) (Exit to reason) ->

519+ ncEffectExit from to reason

509 unexpected ->

520 unexpected ->

510 error $ "nodeController: unexpected message " ++ show unexpected

521 error $ "nodeController: unexpected message " ++ show unexpected

511

522

@@ -712,6 +723,16 @@ ncEffectNamedSend from label msg = do

712 NoImplicitReconnect

723 NoImplicitReconnect

713 (messageToPayload msg)

724 (messageToPayload msg)

714

725

726+-- [Issue #69]

727+ncEffectKill :: ProcessId -> ProcessId -> String -> NC ()

728+ncEffectKill _ to reason =

729+ throwException to $ ProcessKillException reason

730+

731+-- [Issue #69]

732+ncEffectExit :: ProcessId -> ProcessId -> Message -> NC ()

733+ncEffectExit from to reason =

734+ throwException to $ ProcessExitException from reason

735+

715--------------------------------------------------------------------------------

736--------------------------------------------------------------------------------

716-- Auxiliary --

737-- Auxiliary --

717--------------------------------------------------------------------------------

738--------------------------------------------------------------------------------

@@ -761,6 +782,8 @@ destNid (NamedSend _ _) = Nothing

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

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

762-- care of notifying remote nodes

783-- care of notifying remote nodes

763destNid (Died _ _) = Nothing

784destNid (Died _ _) = Nothing

785+destNid (Kill pid _) = Just $ processNodeId pid

786+destNid (Exit pid _) = Just $ processNodeId pid

764

787

765-- | Check if a process is local to our own node

788-- | Check if a process is local to our own node

766isLocal :: LocalNode -> Identifier -> Bool

789isLocal :: LocalNode -> Identifier -> Bool

distributed-process/tests/TestCH.hs+68

@@ -838,6 +838,70 @@ testReceiveChanFeatures transport = do

838

839 takeMVar done

840

841+testKillLocal :: NT.Transport -> Assertion

842+testKillLocal transport = do

843+ localNode <- newLocalNode transport initRemoteTable

844+

845+ pid <- forkProcess localNode $ do

846+ liftIO $ threadDelay 100000

847+

848+ runProcess localNode $ do

849+ ref <- monitor pid

850+ kill pid "TestKill"

851+ ProcessMonitorNotification ref' pid' (DiedException ex) <- expect

852+ True <- return $ ref == ref' && pid == pid' && ex == "TestKill"

853+ return ()

854+

855+testKillRemote :: NT.Transport -> Assertion

856+testKillRemote transport = do

857+ node1 <- newLocalNode transport initRemoteTable

858+ node2 <- newLocalNode transport initRemoteTable

859+

860+ pid <- forkProcess node1 $ do

861+ liftIO $ threadDelay 100000

862+

863+ runProcess node2 $ do

864+ ref <- monitor pid

865+ kill pid "TestKill"

866+ ProcessMonitorNotification ref' pid' (DiedException ex) <- expect

867+ True <- return $ ref == ref' && pid == pid' && ex == "TestKill"

868+ return ()

869+

870+testExitLocal :: NT.Transport -> Assertion

871+testExitLocal transport = do

872+ localNode <- newLocalNode transport initRemoteTable

873+

874+ pid <- forkProcess localNode $ do

875+ (liftIO $ threadDelay 100000)

876+ `catchExit` \from reason -> do

877+ say $ "Got " ++ reason ++ " from " ++ show from

878+ return ()

879+

880+ runProcess localNode $ do

881+ ref <- monitor pid

882+ exit pid "TestExit"

883+ ProcessMonitorNotification ref' pid' (DiedException ex) <- expect

884+ True <- return $ ref == ref' && pid == pid' && ex == "TestExit"

885+ return ()

886+

887+testExitRemote :: NT.Transport -> Assertion

888+testExitRemote transport = do

889+ node1 <- newLocalNode transport initRemoteTable

890+ node2 <- newLocalNode transport initRemoteTable

891+

892+ pid <- forkProcess node1 $ do

893+ (liftIO $ threadDelay 100000)

894+ `catchExit` \from reason -> do

895+ say $ "Got " ++ reason ++ " from " ++ show from

896+ return ()

897+

898+ runProcess node2 $ do

899+ ref <- monitor pid

900+ exit pid "TestExit"

901+ ProcessMonitorNotification ref' pid' (DiedException ex) <- expect

902+ True <- return $ ref == ref' && pid == pid' && ex == "TestExit"

903+ return ()

904+

841tests :: (NT.Transport, TransportInternals) -> [Test]

905tests :: (NT.Transport, TransportInternals) -> [Test]

842tests (transport, transportInternals) = [

906tests (transport, transportInternals) = [

843 testGroup "Basic features" [

907 testGroup "Basic features" [

@@ -855,6 +919,10 @@ tests (transport, transportInternals) = [

855 , testCase "MatchAny" (testMatchAny transport)

919 , testCase "MatchAny" (testMatchAny transport)

856 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout transport)

920 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout transport)

857 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures transport)

921 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures transport)

922+ , testCase "KillLocal" (testKillLocal transport)

923+ , testCase "KillRemote" (testKillRemote transport)

924+ , testCase "ExitLocal" (testExitLocal transport)

925+ , testCase "ExitRemote" (testExitRemote transport)

858 ]

926 ]

859 , testGroup "Monitoring and Linking" [

927 , testGroup "Monitoring and Linking" [

860 -- Monitoring processes

928 -- Monitoring processes