d9fcd8d · distributed-process

git.iko.soy/distributed-process

on master help sign in

Merge branch 'local-send' into development

d9fcd8dTim Watson14 years ago

Makefile+20−10

compare with master

unifiedsplit

@@ -2,26 +2,36 @@

3GHC ?= $(shell which ghc)

4CABAL ?= $(shell which cabal)

5+CABAL_DEV ?= $(shell which cabal-fav)

6BASE_GIT := git://github.com/haskell-distributed

7BASE_GIT := git://github.com/haskell-distributed

7REPOS=$(shell cat REPOS | sed '/^$$/d')

8REPOS=$(shell cat REPOS | sed '/^$$/d')

9.PHONY: all

10.PHONY: all

10−all: $(REPOS)

11+all: dev-install

12−$(REPOS):

13+.PHONY: dev-install

13− git clone $(BASE_GIT)/$@.git

14+ifneq (,$(CABAL_DEV))

15+dev-install:

16+ $(CABAL_DEV) install

17+else

18+dev-install:

19+ $(error install cabal-dev to proceed)

20+endif

15−.PHONY: install

22+.PHONY: ci

23+ci: travis-install travis-test

24+

25+.PHONY: travis-install

16−install: $(REPOS)

26+travis-install: $(REPOS)

17 $(CABAL) install --with-ghc=$(GHC) $(REPOS) --force-reinstalls

27 $(CABAL) install --with-ghc=$(GHC) $(REPOS) --force-reinstalls

18 $(CABAL) install

28 $(CABAL) install

20−.PHONY: ci

30+.PHONY: travis-test

21−ci: install test

22−

23−.PHONY: test

24−test:

31+travis-test:

25 $(CABAL) configure --enable-tests

32 $(CABAL) configure --enable-tests

26 $(CABAL) build

33 $(CABAL) build

27 $(CABAL) test --show-details=always

34 $(CABAL) test --show-details=always

35+

36+$(REPOS):

37+ git clone $(BASE_GIT)/$@.git

src/Control/Distributed/Process/Internal/Primitives.hs+99−46

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

163 , ProcessInfo(..)

164 , ProcessInfoNone(..)

165 , createMessage

166+ , createUnencodedMessage

166 , runLocalProcess

167 , runLocalProcess

167 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

168 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

168 , LocalProcessState

169 , LocalProcessState

@@ -181,6 +182,7 @@ import Control.Distributed.Process.Internal.WeakTQueue

181 , readTQueue

182 , readTQueue

182 , mkWeakTQueue

183 , mkWeakTQueue

183 )

184 )

185+import Unsafe.Coerce

184

186

185--------------------------------------------------------------------------------

187--------------------------------------------------------------------------------

186-- Basic messaging --

188-- Basic messaging --

@@ -192,11 +194,15 @@ send :: Serializable a => ProcessId -> a -> Process ()

192-- modify serializable to allow for stateful (IO) deserialization

194-- modify serializable to allow for stateful (IO) deserialization

193send them msg = do

195send them msg = do

194 proc <- ask

196 proc <- ask

197+ let node = localNodeId (processNode proc)

198+ destNode = (processNodeId them) in do

199+ case destNode == node of

200+ True -> sendLocal them msg

195− liftIO $ sendMessage (processNode proc)

201+ False -> liftIO $ sendMessage (processNode proc)

196− (ProcessIdentifier (processId proc))

202+ (ProcessIdentifier (processId proc))

197− (ProcessIdentifier them)

203+ (ProcessIdentifier them)

198− NoImplicitReconnect

204+ NoImplicitReconnect

199− msg

205+ msg

200

206

201-- | Wait for a message of a specific type

207-- | Wait for a message of a specific type

202expect :: forall a. Serializable a => Process a

208expect :: forall a. Serializable a => Process a

@@ -234,11 +240,16 @@ newChan = do

234sendChan :: Serializable a => SendPort a -> a -> Process ()

240sendChan :: Serializable a => SendPort a -> a -> Process ()

235sendChan (SendPort cid) msg = do

241sendChan (SendPort cid) msg = do

236 proc <- ask

242 proc <- ask

243+ let node = localNodeId (processNode proc)

244+ destNode = processNodeId (sendPortProcessId cid) in do

245+ case destNode == node of

246+ True -> sendChanLocal cid msg

247+ False -> do

237− liftIO $ sendBinary (processNode proc)

248+ liftIO $ sendBinary (processNode proc)

238− (ProcessIdentifier (processId proc))

249+ (ProcessIdentifier (processId proc))

239− (SendPortIdentifier cid)

250+ (SendPortIdentifier cid)

240− NoImplicitReconnect

251+ NoImplicitReconnect

241− msg

252+ msg

242

253

243-- | Wait for a message on a typed channel

254-- | Wait for a message on a typed channel

244receiveChan :: Serializable a => ReceivePort a -> Process a

255receiveChan :: Serializable a => ReceivePort a -> Process a

@@ -315,15 +326,22 @@ match = matchIf (const True)

315-- | Match against any message of the right type that satisfies a predicate

326-- | Match against any message of the right type that satisfies a predicate

316matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b

327matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b

317matchIf c p = Match $ MatchMsg $ \msg ->

328matchIf c p = Match $ MatchMsg $ \msg ->

318− case messageFingerprint msg == fingerprint (undefined :: a) of

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

319− True | c decoded -> Just (p decoded)

330+ False -> Nothing

331+ True -> case msg of

332+ (UnencodedMessage _ m) ->

333+ let m' = unsafeCoerce m :: a in

334+ case (c m') of

335+ True -> Just (p m')

336+ False -> Nothing

337+ (EncodedMessage _ _) ->

338+ if (c decoded) then Just (p decoded) else Nothing

320− where

339+ where

321− decoded :: a

340+ decoded :: a

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

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

323− -- bytestrings when the process calling 'matchIf' doesn't process

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

324− -- the values immediately

343+ -- the values immediately

325− !decoded = decode (messageEncoding msg)

344+ !decoded = decode (messageEncoding msg)

326− _ -> Nothing

327

345

328-- | Match against any message, regardless of the underlying (contained) type

346-- | Match against any message, regardless of the underlying (contained) type

329matchMessage :: (Message -> Process Message) -> Match Message

347matchMessage :: (Message -> Process Message) -> Match Message

@@ -340,11 +358,16 @@ matchMessageIf c p = Match $ MatchMsg $ \msg ->

340forward :: Message -> ProcessId -> Process ()

358forward :: Message -> ProcessId -> Process ()

341forward msg them = do

359forward msg them = do

342 proc <- ask

360 proc <- ask

361+ let node = localNodeId (processNode proc)

362+ destNode = (processNodeId them) in do

363+ case destNode == node of

364+ True -> sendCtrlMsg Nothing (LocalSend them msg)

343− liftIO $ sendPayload (processNode proc)

365+ False -> liftIO $ sendPayload (processNode proc)

344− (ProcessIdentifier (processId proc))

366+ (ProcessIdentifier (processId proc))

345− (ProcessIdentifier them)

367+ (ProcessIdentifier them)

346− NoImplicitReconnect

368+ NoImplicitReconnect

347− (messageToPayload msg)

369+ (messageToPayload msg)

370+

348

371

349-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are

372-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are

350-- 'Serializable' - like the datum they contain - but remember that deserializing

373-- 'Serializable' - like the datum they contain - but remember that deserializing

@@ -357,11 +380,11 @@ forward msg them = do

357-- send self (wrapMessage "blah")

380-- send self (wrapMessage "blah")

358-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)

381-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)

359-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)

382-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)

360−-- "blah" <- unwrapMessage m :: Process (Maybe String)

383+-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)

361-- @

384-- @

362--

385--

363wrapMessage :: Serializable a => a -> Message

386wrapMessage :: Serializable a => a -> Message

364−wrapMessage = createMessage

387+wrapMessage = createMessage

365

388

366-- | Attempt to unwrap a raw 'Message'.

389-- | Attempt to unwrap a raw 'Message'.

367-- If the type of the decoded message payload matches the expected type, the

390-- If the type of the decoded message payload matches the expected type, the

@@ -377,13 +400,16 @@ wrapMessage = createMessage

377unwrapMessage :: forall a. Serializable a => Message -> Process (Maybe a)

400unwrapMessage :: forall a. Serializable a => Message -> Process (Maybe a)

378unwrapMessage msg =

401unwrapMessage msg =

379 case messageFingerprint msg == fingerprint (undefined :: a) of

402 case messageFingerprint msg == fingerprint (undefined :: a) of

403+ False -> return Nothing

404+ True -> case msg of

405+ (UnencodedMessage _ m) ->

406+ let m' = unsafeCoerce m :: a

380− True -> return (Just (decoded))

407+ in return (Just m')

408+ (EncodedMessage _ _) ->

409+ return (Just (decoded))

381− where

410+ where

382− decoded :: a

411+ decoded :: a -- note [decoding]

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

384− -- bytestrings when the calling process doesn't evaluate immediately

385− !decoded = decode (messageEncoding msg)

412+ !decoded = decode (messageEncoding msg)

386− _ -> return Nothing

387

413

388-- | Attempt to handle a raw 'Message'.

414-- | Attempt to handle a raw 'Message'.

389-- If the type of the message matches the type of the first argument to

415-- If the type of the message matches the type of the first argument to

@@ -393,23 +419,28 @@ unwrapMessage msg =

393-- evaluation proceeds, the resulting value with be wrapped with @Just@.

419-- evaluation proceeds, the resulting value with be wrapped with @Just@.

394--

420--

395-- Intended for use in `catchesExit` and `matchAny` primitives.

421-- Intended for use in `catchesExit` and `matchAny` primitives.

396−--

422+--

397handleMessage :: forall a b. (Serializable a)

423handleMessage :: forall a b. (Serializable a)

398 => Message -> (a -> Process b) -> Process (Maybe b)

424 => Message -> (a -> Process b) -> Process (Maybe b)

399handleMessage msg proc = do

425handleMessage msg proc = do

400− case messageFingerprint msg == fingerprint (undefined :: a) of

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

427+ False -> return Nothing

428+ True -> case msg of

429+ (UnencodedMessage _ m) ->

430+ let m' = unsafeCoerce m :: a -- note [decoding]

431+ in do { r <- proc m'; return (Just r) }

432+ (EncodedMessage _ _) ->

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

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

402− where

434+ where

403− decoded :: a

435+ decoded :: a -- note [decoding]

404− !decoded = decode (messageEncoding msg)

436+ !decoded = decode (messageEncoding msg)

405− _ -> return Nothing

406

437

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

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

408-- message from the process mailbox. To handle arbitrary /raw/ messages once

439-- message from the process mailbox. To handle arbitrary /raw/ messages once

409-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.

440-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.

410--

441--

411matchAny :: forall b. (Message -> Process b) -> Match b

442matchAny :: forall b. (Message -> Process b) -> Match b

412−matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)

443+matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)

413

444

414-- | Match against an arbitrary message. Intended for use with 'handleMessage'

445-- | Match against an arbitrary message. Intended for use with 'handleMessage'

415-- and 'unwrapMessage', this function /only/ removes a message from the process

446-- and 'unwrapMessage', this function /only/ removes a message from the process

@@ -425,15 +456,26 @@ matchAnyIf :: forall a b. (Serializable a)

425 -> (Message -> Process b)

456 -> (Message -> Process b)

426 -> Match b

457 -> Match b

427matchAnyIf c p = Match $ MatchMsg $ \msg ->

458matchAnyIf c p = Match $ MatchMsg $ \msg ->

428− case messageFingerprint msg == fingerprint (undefined :: a) of

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

429− True | c decoded -> Just (p msg)

460+ True | check -> Just (p msg)

430 where

461 where

462+ check :: Bool

463+ !check =

464+ case msg of

465+ (EncodedMessage _ _) -> c decoded

466+ (UnencodedMessage _ m') -> c (unsafeCoerce m')

467+

431− decoded :: a

468+ decoded :: a -- note [decoding]

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

433− -- bytestrings when the calling process doesn't evaluate immediately

434 !decoded = decode (messageEncoding msg)

469 !decoded = decode (messageEncoding msg)

435 _ -> Nothing

470 _ -> Nothing

436

471

472+{- note [decoding]

473+For an EncodedMessage, we need to ensure the value is fully decoded so that

474+we don't hang to bytestrings if the calling process doesn't evaluate

475+immediately. For UnencodedMessage we know (because the fingerprint comparison

476+succeeds) that unsafeCoerce will not fail.

477+-}

478+

437-- | Remove any message from the queue

479-- | Remove any message from the queue

438matchUnknown :: Process b -> Match b

480matchUnknown :: Process b -> Match b

439matchUnknown p = Match $ MatchMsg (const (Just p))

481matchUnknown p = Match $ MatchMsg (const (Just p))

@@ -993,6 +1035,17 @@ trace s = do

993-- Auxiliary functions --

1035-- Auxiliary functions --

994--------------------------------------------------------------------------------

1036--------------------------------------------------------------------------------

995

1037

1038+sendLocal :: (Serializable a) => ProcessId -> a -> Process ()

1039+sendLocal pid msg =

1040+ sendCtrlMsg Nothing $ LocalSend pid (createUnencodedMessage msg)

1041+

1042+sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()

1043+sendChanLocal spId msg =

1044+ -- we *must* fully serialize/encode the message here, because

1045+ -- attempting to use `unsafeCoerce' in the node controller

1046+ -- won't work since we know nothing about the required type

1047+ sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)

1048+

996getMonitorRefFor :: Identifier -> Process MonitorRef

1049getMonitorRefFor :: Identifier -> Process MonitorRef

997getMonitorRefFor ident = do

1050getMonitorRefFor ident = do

998 proc <- ask

1051 proc <- ask

src/Control/Distributed/Process/Internal/Types.hs+38−16

@@ -30,6 +30,7 @@ module Control.Distributed.Process.Internal.Types

30 -- * Messages

31 , Message(..)

32 , createMessage

33+ , createUnencodedMessage

33 , messageToPayload

34 , messageToPayload

34 , payloadToMessage

35 , payloadToMessage

35 -- * Node controller user-visible data types

36 -- * Node controller user-visible data types

@@ -83,6 +84,7 @@ import qualified Data.ByteString.Lazy as BSL

83 , toChunks

84 , toChunks

84 , splitAt

85 , splitAt

85 , fromChunks

86 , fromChunks

87+ , length

86 )

88 )

87import qualified Data.ByteString.Lazy.Internal as BSL (ByteString(..))

89import qualified Data.ByteString.Lazy.Internal as BSL (ByteString(..))

88import Data.Accessor (Accessor, accessor)

90import Data.Accessor (Accessor, accessor)

@@ -303,26 +305,40 @@ data ReceivePort a =

303--------------------------------------------------------------------------------

305--------------------------------------------------------------------------------

304

306

305-- | Messages consist of their typeRep fingerprint and their encoding

307-- | Messages consist of their typeRep fingerprint and their encoding

306−data Message = Message

308+data Message =

309+ EncodedMessage

307 { messageFingerprint :: !Fingerprint

310 { messageFingerprint :: !Fingerprint

308 , messageEncoding :: !BSL.ByteString

311 , messageEncoding :: !BSL.ByteString

312+ } |

313+ forall a . Serializable a =>

314+ UnencodedMessage

315+ {

316+ messageFingerprint :: !Fingerprint

317+ , messagePayload :: !a

309 }

318 }

310 deriving (Typeable)

319 deriving (Typeable)

311

320

312instance Show Message where

321instance Show Message where

313− show (Message fp enc) = show enc ++ " :: " ++ showFingerprint fp []

322+ show (EncodedMessage fp enc) = show enc ++ " :: " ++ showFingerprint fp []

323+ show (UnencodedMessage fp _) = "[unencoded message] :: " ++ (showFingerprint fp [])

314

324

315-- | Turn any serialiable term into a message

325-- | Turn any serialiable term into a message

316createMessage :: Serializable a => a -> Message

326createMessage :: Serializable a => a -> Message

317−createMessage a = Message (fingerprint a) (encode a)

327+createMessage a = EncodedMessage (fingerprint a) (encode a)

328+

329+-- | Turn any serializable term into an unencoded/local message

330+createUnencodedMessage :: Serializable a => a -> Message

331+createUnencodedMessage a =

332+ let encoded = encode a in BSL.length encoded `seq` UnencodedMessage (fingerprint a) a

318

333

319-- | Serialize a message

334-- | Serialize a message

320messageToPayload :: Message -> [BSS.ByteString]

335messageToPayload :: Message -> [BSS.ByteString]

321−messageToPayload (Message fp enc) = encodeFingerprint fp : BSL.toChunks enc

336+messageToPayload (EncodedMessage fp enc) = encodeFingerprint fp : BSL.toChunks enc

337+messageToPayload (UnencodedMessage fp m) = messageToPayload ((EncodedMessage fp (encode m)))

322

338

323-- | Deserialize a message

339-- | Deserialize a message

324payloadToMessage :: [BSS.ByteString] -> Message

340payloadToMessage :: [BSS.ByteString] -> Message

325−payloadToMessage payload = Message fp (copy msg)

341+payloadToMessage payload = EncodedMessage fp (copy msg)

326 where

342 where

327 encFp :: BSL.ByteString

343 encFp :: BSL.ByteString

328 msg :: BSL.ByteString

344 msg :: BSL.ByteString

@@ -481,6 +497,8 @@ data ProcessSignal =

481 | WhereIs !String

497 | WhereIs !String

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

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

483 | NamedSend !String !Message

499 | NamedSend !String !Message

500+ | LocalSend !ProcessId !Message

501+ | LocalPortSend !SendPortId !Message

484 | Kill !ProcessId !String

502 | Kill !ProcessId !String

485 | Exit !ProcessId !Message

503 | Exit !ProcessId !Message

486 | GetInfo !ProcessId

504 | GetInfo !ProcessId

@@ -523,18 +541,20 @@ instance Binary MonitorRef where

523 get = MonitorRef <$> get <*> get

541 get = MonitorRef <$> get <*> get

524

542

525instance Binary ProcessSignal where

543instance Binary ProcessSignal where

526− put (Link pid) = putWord8 0 >> put pid

544+ put (Link pid) = putWord8 0 >> put pid

527− put (Unlink pid) = putWord8 1 >> put pid

545+ put (Unlink pid) = putWord8 1 >> put pid

528− put (Monitor ref) = putWord8 2 >> put ref

546+ put (Monitor ref) = putWord8 2 >> put ref

529− put (Unmonitor ref) = putWord8 3 >> put ref

547+ put (Unmonitor ref) = putWord8 3 >> put ref

530− put (Died who reason) = putWord8 4 >> put who >> put reason

548+ put (Died who reason) = putWord8 4 >> put who >> put reason

531− put (Spawn proc ref) = putWord8 5 >> put proc >> put ref

549+ put (Spawn proc ref) = putWord8 5 >> put proc >> put ref

532− put (WhereIs label) = putWord8 6 >> put label

550+ put (WhereIs label) = putWord8 6 >> put label

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

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

534− put (NamedSend label msg) = putWord8 8 >> put label >> put (messageToPayload msg)

552+ put (NamedSend label msg) = putWord8 8 >> put label >> put (messageToPayload msg)

535− put (Kill pid reason) = putWord8 9 >> put pid >> put reason

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

536− put (Exit pid reason) = putWord8 10 >> put pid >> put (messageToPayload reason)

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

555+ put (LocalSend pid msg) = putWord8 11 >> put pid >> put (messageToPayload msg)

556+ put (LocalPortSend sid msg) = putWord8 12 >> put sid >> put (messageToPayload msg)

537− put (GetInfo about) = putWord8 30 >> put about

557+ put (GetInfo about) = putWord8 30 >> put about

538 get = do

558 get = do

539 header <- getWord8

559 header <- getWord8

540 case header of

560 case header of

@@ -549,6 +569,8 @@ instance Binary ProcessSignal where

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

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

550 9 -> Kill <$> get <*> get

570 9 -> Kill <$> get <*> get

551 10 -> Exit <$> get <*> (payloadToMessage <$> get)

571 10 -> Exit <$> get <*> (payloadToMessage <$> get)

572+ 11 -> LocalSend <$> get <*> (payloadToMessage <$> get)

573+ 12 -> LocalPortSend <$> get <*> (payloadToMessage <$> get)

552 30 -> GetInfo <$> get

574 30 -> GetInfo <$> get

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

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

554

576

src/Control/Distributed/Process/Node.hs+52−16

@@ -126,7 +126,7 @@ import Control.Distributed.Process.Internal.Types

126 , DidUnlinkPort(..)

127 , SpawnRef

128 , DidSpawn(..)

129− , Message

129+ , Message(..)

130 , TypedChannel(..)

131 , Identifier(..)

132 , nodeOf

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

138 , WhereIsReply(..)

139 , messageToPayload

140 , payloadToMessage

141− , createMessage

141+ , createUnencodedMessage

142 , runLocalProcess

143 , firstNonReservedProcessId

144 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

@@ -166,12 +166,14 @@ import Control.Distributed.Process.Internal.Primitives

166 )

167import Control.Distributed.Process.Internal.Types (SendPort)

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

169−import Control.Distributed.Process.Internal.WeakTQueue (writeTQueue)

169+import Control.Distributed.Process.Internal.WeakTQueue (TQueue, writeTQueue)

170import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC

171 ( mapMaybe

172 , mapDefault

173 )

174

175+import Unsafe.Coerce

176+

175--------------------------------------------------------------------------------

177--------------------------------------------------------------------------------

176-- Initialization --

178-- Initialization --

177--------------------------------------------------------------------------------

179--------------------------------------------------------------------------------

@@ -577,6 +579,10 @@ nodeController = do

577 ncEffectWhereIs from label

579 ncEffectWhereIs from label

578 NCMsg from (NamedSend label msg') ->

580 NCMsg from (NamedSend label msg') ->

579 ncEffectNamedSend from label msg'

581 ncEffectNamedSend from label msg'

582+ NCMsg _ (LocalSend to msg') ->

583+ ncEffectLocalSend to msg'

584+ NCMsg _ (LocalPortSend to msg') ->

585+ ncEffectLocalPortSend to msg'

580 NCMsg (ProcessIdentifier from) (Kill to reason) ->

586 NCMsg (ProcessIdentifier from) (Kill to reason) ->

581 ncEffectKill from to reason

587 ncEffectKill from to reason

582 NCMsg (ProcessIdentifier from) (Exit to reason) ->

588 NCMsg (ProcessIdentifier from) (Exit to reason) ->

@@ -789,6 +795,34 @@ ncEffectNamedSend from label msg = do

789 NoImplicitReconnect

795 NoImplicitReconnect

790 (messageToPayload msg)

796 (messageToPayload msg)

791

797

798+-- [Issue #DP-20]

799+ncEffectLocalSend :: ProcessId -> Message -> NC ()

800+ncEffectLocalSend = postMessage

801+

802+-- [Issue #DP-20]

803+ncEffectLocalPortSend :: SendPortId -> Message -> NC ()

804+ncEffectLocalPortSend from msg =

805+ let pid = sendPortProcessId from

806+ cid = sendPortLocalId from

807+ in do

808+ withLocalProc pid $ \proc -> do

809+ mChan <- withMVar (processState proc) $ return . (^. typedChannelWithId cid)

810+ case mChan of

811+ -- in the unlikely event we know nothing about this channel id,

812+ -- there's little to be done - perhaps some logging/tracing though...

813+ Nothing -> return ()

814+ Just (TypedChannel chan') -> do

815+ -- If ch is Nothing, the process has given up the read end of

816+ -- the channel and we simply ignore the incoming message - this

817+ ch <- deRefWeak chan'

818+ forM_ ch $ \chan -> deliverChan msg chan

819+ where deliverChan :: forall a . Message -> TQueue a -> IO ()

820+ deliverChan (UnencodedMessage _ raw) chan' =

821+ atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)

822+ deliverChan (EncodedMessage _ _) _ =

823+ -- this will not happen unless someone screws with Primitives.hs

824+ error "invalid local channel delivery"

825+

792-- [Issue #69]

826-- [Issue #69]

793ncEffectKill :: ProcessId -> ProcessId -> String -> NC ()

827ncEffectKill :: ProcessId -> ProcessId -> String -> NC ()

794ncEffectKill from to reason = do

828ncEffectKill from to reason = do

@@ -887,21 +921,23 @@ notifyDied dest src reason mRef = do

887

921

888-- | [Unified: Table 8]

922-- | [Unified: Table 8]

889destNid :: ProcessSignal -> Maybe NodeId

923destNid :: ProcessSignal -> Maybe NodeId

890−destNid (Link ident) = Just $ nodeOf ident

924+destNid (Link ident) = Just $ nodeOf ident

891−destNid (Unlink ident) = Just $ nodeOf ident

925+destNid (Unlink ident) = Just $ nodeOf ident

892−destNid (Monitor ref) = Just $ nodeOf (monitorRefIdent ref)

926+destNid (Monitor ref) = Just $ nodeOf (monitorRefIdent ref)

893−destNid (Unmonitor ref) = Just $ nodeOf (monitorRefIdent ref)

927+destNid (Unmonitor ref) = Just $ nodeOf (monitorRefIdent ref)

894−destNid (Spawn _ _) = Nothing

928+destNid (Spawn _ _) = Nothing

895−destNid (Register _ _ _ _) = Nothing

929+destNid (Register _ _ _ _) = Nothing

896−destNid (WhereIs _) = Nothing

930+destNid (WhereIs _) = Nothing

897−destNid (NamedSend _ _) = Nothing

931+destNid (NamedSend _ _) = Nothing

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

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

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

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

900-- care of notifying remote nodes

934-- care of notifying remote nodes

901−destNid (Died _ _) = Nothing

935+destNid (Died _ _) = Nothing

902−destNid (Kill pid _) = Just $ processNodeId pid

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

903−destNid (Exit pid _) = Just $ processNodeId pid

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

904−destNid (GetInfo pid) = Just $ processNodeId pid

938+destNid (GetInfo pid) = Just $ processNodeId pid

939+destNid (LocalSend pid _) = Just $ processNodeId pid

940+destNid (LocalPortSend cid _) = Just $ processNodeId (sendPortProcessId cid)

905

941

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

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

907isLocal :: LocalNode -> Identifier -> Bool

943isLocal :: LocalNode -> Identifier -> Bool

@@ -938,7 +974,7 @@ isValidLocalIdentifier ident = do

938--------------------------------------------------------------------------------

974--------------------------------------------------------------------------------

939

975

940postAsMessage :: Serializable a => ProcessId -> a -> NC ()

976postAsMessage :: Serializable a => ProcessId -> a -> NC ()

941−postAsMessage pid = postMessage pid . createMessage

977+postAsMessage pid = postMessage pid . createUnencodedMessage

942

978

943postMessage :: ProcessId -> Message -> NC ()

979postMessage :: ProcessId -> Message -> NC ()

944postMessage pid msg = do

980postMessage pid msg = do

tests/TestCH.hs−1

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

32 , LocalNode(localEndPoint)

33 , ProcessExitException(..)

34 , nullProcessId

35− , Message

36 )

35 )

37import Control.Distributed.Process.Node

36import Control.Distributed.Process.Node

38import Control.Distributed.Process.Serializable (Serializable)

37import Control.Distributed.Process.Serializable (Serializable)