e7cd226 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Merge branch 'semi-typed-recv'

e7cd226Tim Watson14 years ago

distributed-process/ChangeLog+4

compare with master

unifiedsplit

@@ -1,3 +1,7 @@

2+* Expose deferred type checking for AbstractMessage

3+* Provide improved exception handling for deferred type checked exit reasons

12012-11-22 Edsko de Vries <edsko@well-typed.com> 0.4.1

52012-11-22 Edsko de Vries <edsko@well-typed.com> 0.4.1

3* Make behaviour of 'register' more Erlang-like (register will now fail if the

7* Make behaviour of 'register' more Erlang-like (register will now fail if the

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

@@ -5,7 +5,7 @@

5-- Peyton Jones

6-- (<http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/>),

7-- although some of the details are different. The precise message passing

8−-- semantics are based on /A unified semantics for future Erlang/ by Hans

8+-- semantics are based on /A unified semantics for future Erlang/ by Hans

9-- Svensson, Lars-Åke Fredlund and Clara Benac Earle.

10module Control.Distributed.Process

11 ( -- * Basic types

@@ -39,6 +39,7 @@ module Control.Distributed.Process

39 , matchUnknown

40 , AbstractMessage(..)

41 , matchAny

42+ , matchAnyIf

42 , matchChan

43 , matchChan

43 -- * Process management

44 -- * Process management

44 , spawn

45 , spawn

@@ -48,6 +49,7 @@ module Control.Distributed.Process

48 , kill

49 , kill

49 , exit

50 , exit

50 , catchExit

51 , catchExit

52+ , catchesExit

51 , ProcessTerminationException(..)

53 , ProcessTerminationException(..)

52 , ProcessRegistrationException(..)

54 , ProcessRegistrationException(..)

53 , SpawnRef

55 , SpawnRef

@@ -192,6 +194,7 @@ import Control.Distributed.Process.Internal.Primitives

192 , matchUnknown

194 , matchUnknown

193 , AbstractMessage(..)

195 , AbstractMessage(..)

194 , matchAny

196 , matchAny

197+ , matchAnyIf

195 , matchChan

198 , matchChan

196 -- Process management

199 -- Process management

197 , terminate

200 , terminate

@@ -199,6 +202,7 @@ import Control.Distributed.Process.Internal.Primitives

199 , die

202 , die

200 , exit

203 , exit

201 , catchExit

204 , catchExit

205+ , catchesExit

202 , kill

206 , kill

203 , getSelfPid

207 , getSelfPid

204 , getSelfNode

208 , getSelfNode

@@ -361,7 +365,11 @@ spawnMonitor nid proc = do

361-- "Control.Distributed.Process.Closure".

365-- "Control.Distributed.Process.Closure".

362--

366--

363-- See also 'spawn'.

367-- See also 'spawn'.

368+call :: Serializable a

364−call :: Serializable a => Static (SerializableDict a) -> NodeId -> Closure (Process a) -> Process a

369+ => Static (SerializableDict a)

370+ -> NodeId

371+ -> Closure (Process a)

372+ -> Process a

365call dict nid proc = do

373call dict nid proc = do

366 us <- getSelfPid

374 us <- getSelfPid

367 (pid, mRef) <- spawnMonitor nid (proc `bindCP` cpSend dict us)

375 (pid, mRef) <- spawnMonitor nid (proc `bindCP` cpSend dict us)

distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+99−18

@@ -21,6 +21,7 @@ module Control.Distributed.Process.Internal.Primitives

21 , matchUnknown

22 , AbstractMessage(..)

23 , matchAny

24+ , matchAnyIf

24 , matchChan

25 , matchChan

25 -- * Process management

26 -- * Process management

26 , terminate

27 , terminate

@@ -29,6 +30,9 @@ module Control.Distributed.Process.Internal.Primitives

29 , kill

30 , kill

30 , exit

31 , exit

31 , catchExit

32 , catchExit

33+ , catchesExit

34+ -- keep the exception constructor hidden, so that handling exit

35+ -- reasons must take place via the 'catchExit' family of primitives

32 , ProcessExitException()

36 , ProcessExitException()

33 , getSelfPid

37 , getSelfPid

34 , getSelfNode

38 , getSelfNode

@@ -311,24 +315,68 @@ matchIf c p = Match $ MatchMsg $ \msg ->

311 !decoded = decode (messageEncoding msg)

315 !decoded = decode (messageEncoding msg)

312 _ -> Nothing

316 _ -> Nothing

313

317

318+-- | Represents a received message and provides two basic operations on it.

314data AbstractMessage = AbstractMessage {

319data AbstractMessage = AbstractMessage {

315− forward :: ProcessId -> Process ()

320+ forward :: ProcessId -> Process () -- ^ forward the message to @ProcessId@

321+ , maybeHandleMessage :: forall a b. (Serializable a)

322+ => (a -> Process b) -> Process (Maybe b) {- ^ Handle the message.

323+ If the type of the message matches the type of the first argument to

324+ the supplied expression, then the expression will be evaluated against

325+ it. If this runtime type checking fails, then @Nothing@ will be returned

326+ to indicate the fact. If the check succeeds and evaluation proceeds

327+ however, the resulting value with be wrapped with @Just@.

328+ -}

316 }

329 }

317

330

318−-- | Match against an arbitrary message

331+-- | Match against an arbitrary message. 'matchAny' removes the first available

332+-- message from the process mailbox, and via the 'AbstractMessage' type,

333+-- supports forwarding /or/ handling the message /if/ it is of the correct

334+-- type. If /not/ of the right type, then the 'AbstractMessage'

335+-- @maybeHandleMessage@ function will not evaluate the supplied expression,

336+-- /but/ the message will still have been removed from the process mailbox!

337+--

319matchAny :: forall b. (AbstractMessage -> Process b) -> Match b

338matchAny :: forall b. (AbstractMessage -> Process b) -> Match b

320−matchAny p = Match $ MatchMsg $ Just . p . abstract

339+matchAny p = Match $ MatchMsg $ Just . p . abstract

340+

341+-- | Match against an arbitrary message. 'matchAnyIf' will /only/ remove the

342+-- message from the process mailbox, /if/ the supplied condition matches. The

343+-- success (or failure) of runtime type checks in @maybeHandleMessage@ does not

344+-- count here, i.e., if the condition evaluates to @True@ then the message will

345+-- be removed from the process mailbox and decoded, but that does /not/

346+-- guarantee that an expression passed to @maybeHandleMessage@ will pass the

347+-- runtime type checks and therefore be evaluated. If the types do not match

348+-- up, then @maybeHandleMessage@ returns 'Nothing'.

349+matchAnyIf :: forall a b. (Serializable a)

350+ => (a -> Bool)

351+ -> (AbstractMessage -> Process b)

352+ -> Match b

353+matchAnyIf c p = Match $ MatchMsg $ \msg ->

354+ case messageFingerprint msg == fingerprint (undefined :: a) of

355+ True | c decoded -> Just (p (abstract msg))

321− where

356+ where

357+ decoded :: a

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

359+ -- bytestrings when the calling process doesn't evaluate immediately

360+ !decoded = decode (messageEncoding msg)

361+ _ -> Nothing

362+

322− abstract :: Message -> AbstractMessage

363+abstract :: Message -> AbstractMessage

323− abstract msg = AbstractMessage {

364+abstract msg = AbstractMessage {

324− forward = \them -> do

365+ forward = \them -> do

325− proc <- ask

366+ proc <- ask

326− liftIO $ sendPayload (processNode proc)

367+ liftIO $ sendPayload (processNode proc)

327− (ProcessIdentifier (processId proc))

368+ (ProcessIdentifier (processId proc))

328− (ProcessIdentifier them)

369+ (ProcessIdentifier them)

329− NoImplicitReconnect

370+ NoImplicitReconnect

330− (messageToPayload msg)

371+ (messageToPayload msg)

372+ , maybeHandleMessage = \(proc :: (a -> Process b)) -> do

373+ case messageFingerprint msg == fingerprint (undefined :: a) of

374+ True -> do { r <- proc (decoded :: a); return (Just r) }

375+ where

376+ decoded :: a

377+ !decoded = decode (messageEncoding msg)

378+ _ -> return Nothing

331− }

379+ }

332

380

333-- | Remove any message from the queue

381-- | Remove any message from the queue

334matchUnknown :: Process b -> Match b

382matchUnknown :: Process b -> Match b

@@ -349,13 +397,15 @@ terminate :: Process a

349terminate = liftIO $ throwIO ProcessTerminationException

397terminate = liftIO $ throwIO ProcessTerminationException

350

398

351-- [Issue #110]

399-- [Issue #110]

352−-- | Die immediately - throws a 'ProcessExitException' with the given @reason@

400+-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.

353die :: Serializable a => a -> Process b

401die :: Serializable a => a -> Process b

354die reason = do

402die reason = do

355 pid <- getSelfPid

403 pid <- getSelfPid

356 liftIO $ throwIO (ProcessExitException pid (createMessage reason))

404 liftIO $ throwIO (ProcessExitException pid (createMessage reason))

357

405

358−-- | Forceful request to kill a process

406+-- | Forceful request to kill a process. Where 'exit' provides an exception

407+-- that can be caught and handled, 'kill' throws an unexposed exception type

408+-- which cannot be handled explicitly (by type).

359kill :: ProcessId -> String -> Process ()

409kill :: ProcessId -> String -> Process ()

360-- NOTE: We send the message to our local node controller, which will then

410-- NOTE: We send the message to our local node controller, which will then

361-- forward it to a remote node controller (if applicable). Sending it directly

411-- forward it to a remote node controller (if applicable). Sending it directly

@@ -363,7 +413,9 @@ kill :: ProcessId -> String -> Process ()

363-- 'monitor' or 'link' request.

413-- 'monitor' or 'link' request.

364kill them reason = sendCtrlMsg Nothing (Kill them reason)

414kill them reason = sendCtrlMsg Nothing (Kill them reason)

365

415

366−-- | Graceful request to exit a process

416+-- | Graceful request to exit a process. Throws 'ProcessExitException' with the

417+-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this

418+-- manner can be handled using the 'catchExit' family of functions.

367exit :: Serializable a => ProcessId -> a -> Process ()

419exit :: Serializable a => ProcessId -> a -> Process ()

368-- NOTE: We send the message to our local node controller, which will then

420-- NOTE: We send the message to our local node controller, which will then

369-- forward it to a remote node controller (if applicable). Sending it directly

421-- forward it to a remote node controller (if applicable). Sending it directly

@@ -371,7 +423,13 @@ exit :: Serializable a => ProcessId -> a -> Process ()

371-- 'monitor' or 'link' request.

423-- 'monitor' or 'link' request.

372exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))

424exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))

373

425

374−-- | Catches ProcessExitException

426+-- | Catches 'ProcessExitException'. The handler will not be applied unless its

427+-- type matches the encoded data stored in the exception (see the /reason/

428+-- argument given to the 'exit' primitive). If the handler cannot be applied,

429+-- the exception will be re-thrown.

430+--

431+-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.

432+-- To handle multiple /reasons/ of differing types, see 'catchesExit'.

375catchExit :: forall a b . (Show a, Serializable a)

433catchExit :: forall a b . (Show a, Serializable a)

376 => Process b

434 => Process b

377 -> (ProcessId -> a -> Process b)

435 -> (ProcessId -> a -> Process b)

@@ -388,6 +446,29 @@ catchExit act exitHandler = catch act handleExit

388 -- bytestrings if the caller doesn't use the value immediately

446 -- bytestrings if the caller doesn't use the value immediately

389 !decoded = decode (messageEncoding msg)

447 !decoded = decode (messageEncoding msg)

390

448

449+-- | As 'Control.Exception.catches' but allows for multiple handlers. Because

450+-- 'ProcessExitException' stores the exit @reason@ as a typed, encoded message,

451+-- a handler must accept an input of the expected type. In order to handle

452+-- a list of potentially different handlers (and therefore input types), a

453+-- handler passed to 'catchesExit' must accept 'AbstractMessage' and return

454+-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).

455+--

456+-- See 'maybeHandleMessage' and 'AsbtractMessage' for more details.

457+catchesExit :: Process b

458+ -> [(ProcessId -> AbstractMessage -> (Process (Maybe b)))]

459+ -> Process b

460+catchesExit act handlers = catch act ((flip handleExit) handlers)

461+ where

462+ handleExit :: ProcessExitException

463+ -> [(ProcessId -> AbstractMessage -> Process (Maybe b))]

464+ -> Process b

465+ handleExit ex [] = liftIO $ throwIO ex

466+ handleExit ex@(ProcessExitException from msg) (h:hs) = do

467+ r <- h from (abstract msg)

468+ case r of

469+ Nothing -> handleExit ex hs

470+ Just p -> return p

471+

391-- | Our own process ID

472-- | Our own process ID

392getSelfPid :: Process ProcessId

473getSelfPid :: Process ProcessId

393getSelfPid = processId <$> ask

474getSelfPid = processId <$> ask

distributed-process/tests/TestCH.hs+135

@@ -757,6 +757,121 @@ testMatchAny transport = do

757

758 takeMVar clientDone

759

760+-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server

761+-- in between, however we block 'Divide' requests ....

762+testMatchAnyHandle :: NT.Transport -> Assertion

763+testMatchAnyHandle transport = do

764+ proxyAddr <- newEmptyMVar

765+ clientDone <- newEmptyMVar

766+

767+ -- Math server

768+ forkIO $ do

769+ localNode <- newLocalNode transport initRemoteTable

770+ mathServer <- forkProcess localNode math

771+ proxyServer <- forkProcess localNode $ forever $ do

772+ receiveWait [

773+ matchAny (maybeForward mathServer)

774+ ]

775+ putMVar proxyAddr proxyServer

776+

777+ -- Client

778+ forkIO $ do

779+ localNode <- newLocalNode transport initRemoteTable

780+ mathServer <- readMVar proxyAddr

781+

782+ runProcess localNode $ do

783+ pid <- getSelfPid

784+ send mathServer (Add pid 1 2)

785+ 3 <- expect :: Process Double

786+ send mathServer (Divide pid 8 2)

787+ Nothing <- (expectTimeout 100000) :: Process (Maybe Double)

788+ liftIO $ putMVar clientDone ()

789+

790+ takeMVar clientDone

791+ where maybeForward :: ProcessId -> AbstractMessage -> Process (Maybe ())

792+ maybeForward s msg =

793+ maybeHandleMessage msg (\m@(Add _ _ _) -> send s m)

794+

795+testMatchAnyNoHandle :: NT.Transport -> Assertion

796+testMatchAnyNoHandle transport = do

797+ addr <- newEmptyMVar

798+ clientDone <- newEmptyMVar

799+ serverDone <- newEmptyMVar

800+

801+ -- Math server

802+ forkIO $ do

803+ localNode <- newLocalNode transport initRemoteTable

804+ server <- forkProcess localNode $ forever $ do

805+ receiveWait [

806+ matchAnyIf

807+ -- the condition has type `Add -> Bool`

808+ (\(Add _ _ _) -> True)

809+ -- the match `AbstractMessage -> Process ()` will succeed!

810+ (\m -> do

811+ -- `String -> Process ()` does *not* match the input types however

812+ r <- (maybeHandleMessage m (\(_ :: String) -> die "NONSENSE" ))

813+ case r of

814+ Nothing -> return ()

815+ Just _ -> die "NONSENSE")

816+ ]

817+ -- we *must* have removed the message from our mailbox though!!!

818+ Nothing <- receiveTimeout 100000 [ match (\(Add _ _ _) -> return ()) ]

819+ liftIO $ putMVar serverDone ()

820+ putMVar addr server

821+

822+ -- Client

823+ forkIO $ do

824+ localNode <- newLocalNode transport initRemoteTable

825+ server <- readMVar addr

826+

827+ runProcess localNode $ do

828+ pid <- getSelfPid

829+ send server (Add pid 1 2)

830+ -- we only care about the client having sent a message, so we're done

831+ liftIO $ putMVar clientDone ()

832+

833+ takeMVar clientDone

834+ takeMVar serverDone

835+

836+-- | Test 'matchAnyIf'. We provide an /echo/ server, but it ignores requests

837+-- unless the text body @/= "bar"@ - this case should time out rather than

838+-- removing the message from the process mailbox.

839+testMatchAnyIf :: NT.Transport -> Assertion

840+testMatchAnyIf transport = do

841+ echoAddr <- newEmptyMVar

842+ clientDone <- newEmptyMVar

843+

844+ -- echo server

845+ forkIO $ do

846+ localNode <- newLocalNode transport initRemoteTable

847+ echoServer <- forkProcess localNode $ forever $ do

848+ receiveWait [

849+ matchAnyIf (\(_ :: ProcessId, (s :: String)) -> s /= "bar")

850+ handleMessage

851+ ]

852+ putMVar echoAddr echoServer

853+

854+ -- Client

855+ forkIO $ do

856+ localNode <- newLocalNode transport initRemoteTable

857+ server <- readMVar echoAddr

858+

859+ runProcess localNode $ do

860+ pid <- getSelfPid

861+ send server (pid, "foo")

862+ "foo" <- expect

863+ send server (pid, "baz")

864+ "baz" <- expect

865+ send server (pid, "bar")

866+ Nothing <- (expectTimeout 100000) :: Process (Maybe Double)

867+ liftIO $ putMVar clientDone ()

868+

869+ takeMVar clientDone

870+ where handleMessage :: AbstractMessage -> Process (Maybe ())

871+ handleMessage msg =

872+ maybeHandleMessage msg (\(pid :: ProcessId, (m :: String))

873+ -> do { send pid m; return () })

874+

760-- Test 'receiveChanTimeout'

875-- Test 'receiveChanTimeout'

761testReceiveChanTimeout :: NT.Transport -> Assertion

876testReceiveChanTimeout :: NT.Transport -> Assertion

762testReceiveChanTimeout transport = do

877testReceiveChanTimeout transport = do

@@ -894,6 +1009,22 @@ testKillRemote transport = do

894

1009

895 takeMVar done

1010 takeMVar done

896

1011

1012+testCatchesExit :: NT.Transport -> Assertion

1013+testCatchesExit transport = do

1014+ localNode <- newLocalNode transport initRemoteTable

1015+ done <- newEmptyMVar

1016+

1017+ _ <- forkProcess localNode $ do

1018+ (die ("foobar", 123 :: Int))

1019+ `catchesExit` [

1020+ (\_ m -> maybeHandleMessage m (\(_ :: String) -> return ()))

1021+ , (\_ m -> maybeHandleMessage m (\(_ :: Maybe Int) -> return ()))

1022+ , (\_ m -> maybeHandleMessage m (\(_ :: String, _ :: Int)

1023+ -> (liftIO $ putMVar done ()) >> return ()))

1024+ ]

1025+

1026+ takeMVar done

1027+

897testDie :: NT.Transport -> Assertion

1028testDie :: NT.Transport -> Assertion

898testDie transport = do

1029testDie transport = do

899 localNode <- newLocalNode transport initRemoteTable

1030 localNode <- newLocalNode transport initRemoteTable

@@ -987,12 +1118,16 @@ tests (transport, transportInternals) = [

987 , testCase "RemoteRegistry" (testRemoteRegistry transport)

1118 , testCase "RemoteRegistry" (testRemoteRegistry transport)

988 , testCase "SpawnLocal" (testSpawnLocal transport)

1119 , testCase "SpawnLocal" (testSpawnLocal transport)

989 , testCase "MatchAny" (testMatchAny transport)

1120 , testCase "MatchAny" (testMatchAny transport)

1121+ , testCase "MatchAnyHandle" (testMatchAnyHandle transport)

1122+ , testCase "MatchAnyNoHandle" (testMatchAnyNoHandle transport)

1123+ , testCase "MatchAnyIf" (testMatchAnyIf transport)

990 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout transport)

1124 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout transport)

991 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures transport)

1125 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures transport)

992 , testCase "KillLocal" (testKillLocal transport)

1126 , testCase "KillLocal" (testKillLocal transport)

993 , testCase "KillRemote" (testKillRemote transport)

1127 , testCase "KillRemote" (testKillRemote transport)

994 , testCase "Die" (testDie transport)

1128 , testCase "Die" (testDie transport)

995 , testCase "PrettyExit" (testPrettyExit transport)

1129 , testCase "PrettyExit" (testPrettyExit transport)

1130+ , testCase "CatchesExit" (testCatchesExit transport)

996 , testCase "ExitLocal" (testExitLocal transport)

1131 , testCase "ExitLocal" (testExitLocal transport)

997 , testCase "ExitRemote" (testExitRemote transport)

1132 , testCase "ExitRemote" (testExitRemote transport)

998 ]

1133 ]