d31ba89 · distributed-process

git.iko.soy/distributed-process

rebased changes against latest from upstream

commitd31ba898e0340250c6243fa43114b72f030f3ae1 authorRodLogic <dev@rodlogic.net> dateMon 3 Dec 2012, 10:06 (14 years ago) parent883830f

5 files changed+169−2

compare with master

unifiedsplit

distributed-process/src/Control/Distributed/Process.hs+7−1 distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+64 distributed-process/src/Control/Distributed/Process/Internal/Types.hs+6 distributed-process/src/Control/Distributed/Process/Node.hs+24−1 distributed-process/tests/TestCH.hs+68

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

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

4343 , spawn

4444 , call

4545 , terminate

46+ , kill

47+ , exit

48+ , catchExit

4649 , ProcessTerminationException(..)

4750 , ProcessRegistrationException(..)

4851 , SpawnRef

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

184187 -- Process management

185188 , terminate

186189 , ProcessTerminationException(..)

190+ , exit

191+ , catchExit

192+ , kill

187193 , getSelfPid

188194 , getSelfNode

189195 -- Monitoring and linking

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

414420 node <- processNode <$> ask

415421 liftIO $ do

416422 mvar <- newEmptyMVar

417− forkProcess node $ do

423+ _ <- forkProcess node $ do

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

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

420426 (sport, rport) <- newChan

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

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

2424 -- * Process management

2525 , terminate

2626 , ProcessTerminationException(..)

27+ , kill

28+ , ProcessKillException(..)

29+ , exit

30+ , catchExit

31+ , ProcessExitException(..)

2732 , getSelfPid

2833 , getSelfNode

2934 -- * Monitoring and linking

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

328333terminate :: Process a

329334terminate = liftIO $ throwIO ProcessTerminationException

330335

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+

331395-- | Our own process ID

332396getSelfPid :: Process ProcessId

333397getSelfPid = processId <$> ask

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

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

442442 | WhereIs !String

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

444444 | NamedSend !String !Message

445+ | Kill !ProcessId !String

446+ | Exit !ProcessId !Message

445447 deriving Show

446448

447449--------------------------------------------------------------------------------

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

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

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

488490 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)

489493 get = do

490494 header <- getWord8

491495 case header of

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

498502 6 -> WhereIs <$> get

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

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

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

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

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

502508

503509instance Binary DiedReason where

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

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

138138 , impliesDeathOf

139139 )

140140import Control.Distributed.Process.Internal.Primitives

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

141+ ( register

142+ , finally

143+ , receiveWait

144+ , match

145+ , sendChan

146+ , ProcessKillException(..)

147+ , ProcessExitException(..)

148+ )

142149import Control.Distributed.Process.Internal.Types

143150 (SendPort)

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

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

506513 ncEffectWhereIs from label

507514 NCMsg from (NamedSend label msg') ->

508515 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

509520 unexpected ->

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

511522

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

712723 NoImplicitReconnect

713724 (messageToPayload msg)

714725

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+

715736--------------------------------------------------------------------------------

716737-- Auxiliary --

717738--------------------------------------------------------------------------------

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

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

762783-- care of notifying remote nodes

763784destNid (Died _ _) = Nothing

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

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

764787

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

766789isLocal :: LocalNode -> Identifier -> Bool

distributed-process/tests/TestCH.hs+68

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

838838

839839 takeMVar done

840840

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+

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

842906tests (transport, transportInternals) = [

843907 testGroup "Basic features" [

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

855919 , testCase "MatchAny" (testMatchAny transport)

856920 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout transport)

857921 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures transport)

922+ , testCase "KillLocal" (testKillLocal transport)

923+ , testCase "KillRemote" (testKillRemote transport)

924+ , testCase "ExitLocal" (testExitLocal transport)

925+ , testCase "ExitRemote" (testExitRemote transport)

858926 ]

859927 , testGroup "Monitoring and Linking" [

860928 -- Monitoring processes