Reset changes disabling the optimiser and avoid GHC INLINE bug

Author: hyperthunk

distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs

@@ -44,7 +44,7 @@ import Control.Distributed.Process.Internal.Types
 import Control.Distributed.Process.Node
 import Control.Distributed.Process.Debug
 import Control.Distributed.Process.Management.Internal.Types
-import Control.Distributed.Process.Tests.Internal.Utils (shouldBe)
+import Control.Distributed.Process.Tests.Internal.Utils (shouldBe, pause)
 import Control.Distributed.Process.Serializable (Serializable)
 
 import Test.HUnit (Assertion, assertBool, assertFailure)
@@ -87,10 +87,9 @@ verifyClient :: String -> MVar Bool -> IO ()
 verifyClient s b = takeMVar b >>= assertBool s
 
 expectPing :: MVar Bool ->  Process ()
-expectPing mv = expect >>= liftIO . putMVar mv . checkPing
+expectPing mv = expect  >>= liftIO . putMVar mv . checkPing
   where
     checkPing (Ping _) = True
-    checkPing _        = False
 
 -- | Quick and dirty synchronous version of whereisRemoteAsync
 whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)
@@ -240,10 +239,10 @@ testPing TestTransport{..} = do
       pid <- getSelfPid
       replicateM_ numPings $ do
         send pingServer (Pong pid)
-        p <- expect
+        p <- expectTimeout 3000000
         case p of
-          Ping _ -> return ()
-          _      -> die "Unexpected message"
+          Just (Ping _) -> return ()
+          Nothing       -> die "Failed to receive Ping"
 
     putMVar clientDone ()
 
@@ -560,7 +559,7 @@ testMergeChannels TestTransport{..} = do
         ss <- liftIO $ mapM readMVar vs
         case ss of
           [sa, sb, sc] ->
-            mapM_ ((>> liftIO (threadDelay 10000)) . uncurry sendChan)
+            mapM_ ((>> pause 10000) . uncurry sendChan)
               [ -- a, b, c
                 (sa, 'a')
               , (sb, 'b')

src/Control/Distributed/Process/Internal/Primitives.hs

@@ -335,13 +335,16 @@ newChan = do
 sendChan :: Serializable a => SendPort a -> a -> Process ()
 sendChan (SendPort cid) msg = do
   proc <- ask
-  let node     = localNodeId (processNode proc)
-      destNode = processNodeId (sendPortProcessId cid) in do
-  case destNode == node of
+  let node = processNode proc
+      pid  = processId proc
+      us   = localNodeId node
+      them = processNodeId (sendPortProcessId cid) in do
+  liftIO $ traceEvent (localEventBus node) (MxSentToPort pid cid $ wrapMessage msg)
+  case them == us of
     True  -> sendChanLocal cid msg
     False -> do
-      liftIO $ sendBinary (processNode proc)
-                          (ProcessIdentifier (processId proc))
+      liftIO $ sendBinary node
+                          (ProcessIdentifier pid)
                           (SendPortIdentifier cid)
                           NoImplicitReconnect
                           msg

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

@@ -21,6 +21,7 @@ import Control.Concurrent.STM
 import Control.Distributed.Process.Internal.Types
   ( Process
   , ProcessId
+  , SendPortId
   , Message
   , DiedReason
   , NodeId
@@ -59,8 +60,12 @@ data MxEvent =
     -- ^ fired whenever a message is sent from a local process
   | MxSentToName       String       ProcessId Message
     -- ^ fired whenever a named send occurs
+  | MxSentToPort       ProcessId    SendPortId Message
+    -- ^ fired whenever a sendChan occurs
   | MxReceived         ProcessId    Message
     -- ^ fired whenever a message is received by a local process
+  | MxReceivedPort     SendPortId   Message
+    -- ^ fired whenever a message is received via a typed channel
   | MxConnected        ConnectionId EndPointAddress
     -- ^ fired when a network-transport connection is first established
   | MxDisconnected     ConnectionId EndPointAddress

src/Control/Distributed/Process/Node.hs

@@ -492,7 +492,7 @@ type IncomingConnection = (NT.EndPointAddress, IncomingTarget)
 data IncomingTarget =
     Uninit
   | ToProc ProcessId (Weak (CQueue Message))
-  | ToChan TypedChannel
+  | ToChan SendPortId TypedChannel
   | ToNode
 
 data ConnectionState = ConnectionState {
@@ -547,13 +547,17 @@ handleIncomingMessages node = go initConnectionState
                 enqueue queue msg -- 'enqueue' is strict
                 trace node (MxReceived pid msg)
               go st
-            Just (_, ToChan (TypedChannel chan')) -> do
+            Just (_, ToChan chId (TypedChannel chan')) -> do
               mChan <- deRefWeak chan'
               -- If mChan is Nothing, the process has given up the read end of
               -- the channel and we simply ignore the incoming message
-              forM_ mChan $ \chan -> atomically $
-                -- We make sure the message is fully decoded when it is enqueued
-                writeTQueue chan $! decode (BSL.fromChunks payload)
+              forM_ mChan $ \chan -> do
+                msg' <- atomically $ do
+                  msg <- return $! decode (BSL.fromChunks payload)
+                  -- We make sure the message is fully decoded when it is enqueued
+                  writeTQueue chan msg
+                  return msg
+                trace node $ MxReceivedPort chId $ unsafeCreateUnencodedMessage msg'
               go st
             Just (_, ToNode) -> do
               let ctrlMsg = decode . BSL.fromChunks $ payload
@@ -580,7 +584,7 @@ handleIncomingMessages node = go initConnectionState
                       mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
                       case mChannel of
                         Just channel ->
-                          go (incomingAt cid ^= Just (src, ToChan channel) $ st)
+                          go (incomingAt cid ^= Just (src, ToChan chId channel) $ st)
                         Nothing ->
                           invalidRequest cid st $
                             "incoming attempt to connect to unknown channel of"
@@ -1050,11 +1054,12 @@ ncEffectLocalPortSend from msg = do
         -- If ch is Nothing, the process has given up the read end of
         -- the channel and we simply ignore the incoming message - this
         ch <- deRefWeak chan'
-        forM_ ch $ \chan -> deliverChan msg chan
-  where deliverChan :: forall a . Message -> TQueue a -> IO ()
-        deliverChan (UnencodedMessage _ raw) chan' =
+        forM_ ch $ \chan -> deliverChan node from msg chan
+  where deliverChan :: forall a . LocalNode -> SendPortId -> Message -> TQueue a -> IO ()
+        deliverChan n p (UnencodedMessage _ raw) chan' = do
             atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)
-        deliverChan (EncodedMessage   _ _) _ =
+            trace n (MxReceivedPort p $ unsafeCreateUnencodedMessage raw)
+        deliverChan _ _ (EncodedMessage   _ _) _ =
             -- this will not happen unless someone screws with Primitives.hs
             error "invalid local channel delivery"
 

src/Control/Distributed/Process/UnsafePrimitives.hs

@@ -178,20 +178,22 @@ usend them msg = do
 sendChan :: Serializable a => SendPort a -> a -> Process ()
 sendChan (SendPort cid) msg = do
   proc <- ask
-  let node     = localNodeId (processNode proc)
-      destNode = processNodeId (sendPortProcessId cid) in do
-  case destNode == node of
-    True  -> unsafeSendChanLocal cid msg
-    False ->
-      liftIO $ sendBinary (processNode proc)
-                          (ProcessIdentifier (processId proc))
-                          (SendPortIdentifier cid)
-                          NoImplicitReconnect
-                          msg
+  let node = processNode proc
+      pid  = processId proc
+      us   = localNodeId node
+      them = processNodeId (sendPortProcessId cid)
+      msg' = wrapMessage msg in do
+  liftIO $ traceEvent (localEventBus node) (MxSentToPort pid cid msg')
+  if them == us
+    then unsafeSendChanLocal cid msg' -- NB: we wrap to P.Message !!!
+    else liftIO $ sendBinary node
+                             (ProcessIdentifier pid)
+                             (SendPortIdentifier cid)
+                             NoImplicitReconnect
+                             msg
   where
-    unsafeSendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
-    unsafeSendChanLocal spId msg' =
-      sendCtrlMsg Nothing $ LocalPortSend spId (wrapMessage msg')
+    unsafeSendChanLocal :: SendPortId -> Message -> Process ()
+    unsafeSendChanLocal p m = sendCtrlMsg Nothing $ LocalPortSend p m
 
 -- | Create an unencoded @Message@ for any @Serializable@ type.
 wrapMessage :: Serializable a => a -> Message

stack-ghc-7.10.3.yaml

@@ -14,7 +14,7 @@ extra-deps:
 - network-transport-0.5.2
 - network-transport-inmemory-0.5.2
 - rematch-0.2.0.0
-- exceptions-0.8.2.1 # test dependency
+- exceptions-0.8.2.1
 
 flags:
   distributed-process-tests: