5fc844f · distributed-process

git.iko.soy/distributed-process

on master help sign in

Reset changes disabling the optimiser and avoid GHC INLINE bug

5fc844fTim Watson8 years ago

distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs+6−7

compare with master

unifiedsplit

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

44import Control.Distributed.Process.Node

45import Control.Distributed.Process.Debug

46import Control.Distributed.Process.Management.Internal.Types

47−import Control.Distributed.Process.Tests.Internal.Utils (shouldBe)

47+import Control.Distributed.Process.Tests.Internal.Utils (shouldBe, pause)

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

50import Test.HUnit (Assertion, assertBool, assertFailure)

@@ -87,10 +87,9 @@ verifyClient :: String -> MVar Bool -> IO ()

87verifyClient s b = takeMVar b >>= assertBool s

89expectPing :: MVar Bool -> Process ()

90−expectPing mv = expect >>= liftIO . putMVar mv . checkPing

90+expectPing mv = expect >>= liftIO . putMVar mv . checkPing

91 where

92 checkPing (Ping _) = True

93− checkPing _ = False

95-- | Quick and dirty synchronous version of whereisRemoteAsync

94-- | Quick and dirty synchronous version of whereisRemoteAsync

96whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)

95whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)

@@ -240,10 +239,10 @@ testPing TestTransport{..} = do

240 pid <- getSelfPid

239 pid <- getSelfPid

241 replicateM_ numPings $ do

240 replicateM_ numPings $ do

242 send pingServer (Pong pid)

241 send pingServer (Pong pid)

243− p <- expect

242+ p <- expectTimeout 3000000

244 case p of

243 case p of

245− Ping _ -> return ()

244+ Just (Ping _) -> return ()

246− _ -> die "Unexpected message"

245+ Nothing -> die "Failed to receive Ping"

247

246

248 putMVar clientDone ()

247 putMVar clientDone ()

249

248

@@ -560,7 +559,7 @@ testMergeChannels TestTransport{..} = do

560 ss <- liftIO $ mapM readMVar vs

559 ss <- liftIO $ mapM readMVar vs

561 case ss of

560 case ss of

562 [sa, sb, sc] ->

561 [sa, sb, sc] ->

563− mapM_ ((>> liftIO (threadDelay 10000)) . uncurry sendChan)

562+ mapM_ ((>> pause 10000) . uncurry sendChan)

564 [ -- a, b, c

563 [ -- a, b, c

565 (sa, 'a')

564 (sa, 'a')

566 , (sb, 'b')

565 , (sb, 'b')

src/Control/Distributed/Process/Internal/Primitives.hs+8−5

@@ -335,13 +335,16 @@ newChan = do

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

336sendChan (SendPort cid) msg = do

337 proc <- ask

338− let node = localNodeId (processNode proc)

338+ let node = processNode proc

339+ pid = processId proc

340+ us = localNodeId node

339− destNode = processNodeId (sendPortProcessId cid) in do

341+ them = processNodeId (sendPortProcessId cid) in do

342+ liftIO $ traceEvent (localEventBus node) (MxSentToPort pid cid $ wrapMessage msg)

340− case destNode == node of

343+ case them == us of

341 True -> sendChanLocal cid msg

344 True -> sendChanLocal cid msg

342 False -> do

345 False -> do

343− liftIO $ sendBinary (processNode proc)

346+ liftIO $ sendBinary node

344− (ProcessIdentifier (processId proc))

347+ (ProcessIdentifier pid)

345 (SendPortIdentifier cid)

348 (SendPortIdentifier cid)

346 NoImplicitReconnect

349 NoImplicitReconnect

347 msg

350 msg

src/Control/Distributed/Process/Management/Internal/Types.hs+5

@@ -21,6 +21,7 @@ import Control.Concurrent.STM

21import Control.Distributed.Process.Internal.Types

22 ( Process

23 , ProcessId

24+ , SendPortId

24 , Message

25 , Message

25 , DiedReason

26 , DiedReason

26 , NodeId

27 , NodeId

@@ -59,8 +60,12 @@ data MxEvent =

59 -- ^ fired whenever a message is sent from a local process

60 -- ^ fired whenever a message is sent from a local process

60 | MxSentToName String ProcessId Message

61 | MxSentToName String ProcessId Message

61 -- ^ fired whenever a named send occurs

62 -- ^ fired whenever a named send occurs

63+ | MxSentToPort ProcessId SendPortId Message

64+ -- ^ fired whenever a sendChan occurs

62 | MxReceived ProcessId Message

65 | MxReceived ProcessId Message

63 -- ^ fired whenever a message is received by a local process

66 -- ^ fired whenever a message is received by a local process

67+ | MxReceivedPort SendPortId Message

68+ -- ^ fired whenever a message is received via a typed channel

64 | MxConnected ConnectionId EndPointAddress

69 | MxConnected ConnectionId EndPointAddress

65 -- ^ fired when a network-transport connection is first established

70 -- ^ fired when a network-transport connection is first established

66 | MxDisconnected ConnectionId EndPointAddress

71 | MxDisconnected ConnectionId EndPointAddress

src/Control/Distributed/Process/Node.hs+15−10

@@ -492,7 +492,7 @@ type IncomingConnection = (NT.EndPointAddress, IncomingTarget)

492data IncomingTarget =

493 Uninit

494 | ToProc ProcessId (Weak (CQueue Message))

495− | ToChan TypedChannel

495+ | ToChan SendPortId TypedChannel

496 | ToNode

497

498data ConnectionState = ConnectionState {

@@ -547,13 +547,17 @@ handleIncomingMessages node = go initConnectionState

547 enqueue queue msg -- 'enqueue' is strict

548 trace node (MxReceived pid msg)

549 go st

550− Just (_, ToChan (TypedChannel chan')) -> do

550+ Just (_, ToChan chId (TypedChannel chan')) -> do

551 mChan <- deRefWeak chan'

552 -- If mChan is Nothing, the process has given up the read end of

553 -- the channel and we simply ignore the incoming message

554− forM_ mChan $ \chan -> atomically $

554+ forM_ mChan $ \chan -> do

555+ msg' <- atomically $ do

556+ msg <- return $! decode (BSL.fromChunks payload)

555− -- We make sure the message is fully decoded when it is enqueued

557+ -- We make sure the message is fully decoded when it is enqueued

556− writeTQueue chan $! decode (BSL.fromChunks payload)

558+ writeTQueue chan msg

559+ return msg

560+ trace node $ MxReceivedPort chId $ unsafeCreateUnencodedMessage msg'

557 go st

561 go st

558 Just (_, ToNode) -> do

562 Just (_, ToNode) -> do

559 let ctrlMsg = decode . BSL.fromChunks $ payload

563 let ctrlMsg = decode . BSL.fromChunks $ payload

@@ -580,7 +584,7 @@ handleIncomingMessages node = go initConnectionState

580 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)

584 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)

581 case mChannel of

585 case mChannel of

582 Just channel ->

586 Just channel ->

583− go (incomingAt cid ^= Just (src, ToChan channel) $ st)

587+ go (incomingAt cid ^= Just (src, ToChan chId channel) $ st)

584 Nothing ->

588 Nothing ->

585 invalidRequest cid st $

589 invalidRequest cid st $

586 "incoming attempt to connect to unknown channel of"

590 "incoming attempt to connect to unknown channel of"

@@ -1050,11 +1054,12 @@ ncEffectLocalPortSend from msg = do

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

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

1051 -- the channel and we simply ignore the incoming message - this

1055 -- the channel and we simply ignore the incoming message - this

1052 ch <- deRefWeak chan'

1056 ch <- deRefWeak chan'

1053− forM_ ch $ \chan -> deliverChan msg chan

1057+ forM_ ch $ \chan -> deliverChan node from msg chan

1054− where deliverChan :: forall a . Message -> TQueue a -> IO ()

1058+ where deliverChan :: forall a . LocalNode -> SendPortId -> Message -> TQueue a -> IO ()

1055− deliverChan (UnencodedMessage _ raw) chan' =

1059+ deliverChan n p (UnencodedMessage _ raw) chan' = do

1056 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)

1060 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)

1061+ trace n (MxReceivedPort p $ unsafeCreateUnencodedMessage raw)

1057− deliverChan (EncodedMessage _ _) _ =

1062+ deliverChan _ _ (EncodedMessage _ _) _ =

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

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

1059 error "invalid local channel delivery"

1064 error "invalid local channel delivery"

1060

1065

src/Control/Distributed/Process/UnsafePrimitives.hs+15−13

@@ -178,20 +178,22 @@ usend them msg = do

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

179sendChan (SendPort cid) msg = do

180 proc <- ask

181+ let node = processNode proc

182+ pid = processId proc

181− let node = localNodeId (processNode proc)

183+ us = localNodeId node

182− destNode = processNodeId (sendPortProcessId cid) in do

184+ them = processNodeId (sendPortProcessId cid)

183− case destNode == node of

185+ msg' = wrapMessage msg in do

186+ liftIO $ traceEvent (localEventBus node) (MxSentToPort pid cid msg')

187+ if them == us

184− True -> unsafeSendChanLocal cid msg

188+ then unsafeSendChanLocal cid msg' -- NB: we wrap to P.Message !!!

185− False ->

186− liftIO $ sendBinary (processNode proc)

189+ else liftIO $ sendBinary node

187− (ProcessIdentifier (processId proc))

190+ (ProcessIdentifier pid)

188− (SendPortIdentifier cid)

191+ (SendPortIdentifier cid)

189− NoImplicitReconnect

192+ NoImplicitReconnect

190− msg

193+ msg

191 where

194 where

192− unsafeSendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()

195+ unsafeSendChanLocal :: SendPortId -> Message -> Process ()

193− unsafeSendChanLocal spId msg' =

194− sendCtrlMsg Nothing $ LocalPortSend spId (wrapMessage msg')

196+ unsafeSendChanLocal p m = sendCtrlMsg Nothing $ LocalPortSend p m

195

197

196-- | Create an unencoded @Message@ for any @Serializable@ type.

198-- | Create an unencoded @Message@ for any @Serializable@ type.

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

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

stack-ghc-7.10.3.yaml+1−1

@@ -14,7 +14,7 @@ extra-deps:

14- network-transport-0.5.2

15- network-transport-inmemory-0.5.2

16- rematch-0.2.0.0

17−- exceptions-0.8.2.1 # test dependency

17+- exceptions-0.8.2.1

19flags:

20 distributed-process-tests: