6a84b59 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Merge branch 'debug-trace'

6a84b59Tim Watson14 years ago

distributed-process/distributed-process.cabal+2−1

compare with master

unifiedsplit

@@ -61,6 +61,7 @@ Library

61 Control.Distributed.Process.Internal.Primitives,

62 Control.Distributed.Process.Internal.CQueue,

63 Control.Distributed.Process.Internal.Types,

64+ Control.Distributed.Process.Internal.Trace,

64 Control.Distributed.Process.Internal.Closure.BuiltIn,

65 Control.Distributed.Process.Internal.Closure.BuiltIn,

65 Control.Distributed.Process.Internal.Messaging,

66 Control.Distributed.Process.Internal.Messaging,

66 Control.Distributed.Process.Internal.StrictList,

67 Control.Distributed.Process.Internal.StrictList,

@@ -145,7 +146,7 @@ Test-Suite TestStats

145 ScopedTypeVariables,

146 ScopedTypeVariables,

146 DeriveDataTypeable,

147 DeriveDataTypeable,

147 GeneralizedNewtypeDeriving

148 GeneralizedNewtypeDeriving

148− ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

149+ ghc-options: -Wall -debug -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

149 HS-Source-Dirs: tests

150 HS-Source-Dirs: tests

150

151

152

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

@@ -83,6 +83,8 @@ module Control.Distributed.Process.Internal.Primitives

83 -- * Reconnecting

84 , reconnect

85 , reconnectPort

86+ -- * Tracing/Debugging

87+ , trace

86 ) where

88 ) where

88#if ! MIN_VERSION_base(4,6,0)

90#if ! MIN_VERSION_base(4,6,0)

@@ -168,6 +170,7 @@ import Control.Distributed.Process.Internal.Messaging

168 , sendPayload

170 , sendPayload

169 , disconnect

171 , disconnect

170 )

172 )

173+import qualified Control.Distributed.Process.Internal.Trace as Trace

171import Control.Distributed.Process.Internal.WeakTQueue

174import Control.Distributed.Process.Internal.WeakTQueue

172 ( newTQueueIO

175 ( newTQueueIO

173 , readTQueue

176 , readTQueue

@@ -892,6 +895,22 @@ reconnectPort them = do

892 node <- processNode <$> ask

895 node <- processNode <$> ask

893 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))

896 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))

894

897

898+--------------------------------------------------------------------------------

899+-- Debugging/Tracing --

900+--------------------------------------------------------------------------------

901+

902+-- | Send a message to the internal (system) trace facility. If tracing is

903+-- enabled, this will create a custom trace event in the event log. Specifically,

904+-- the environment variable @DISTRIBUTED_PROCESS_TRACE_FILE@ is set to a valid

905+-- path, then trace output will be written to that file. Otherwise, if the GHC

906+-- eventlog is enabled (i.e., you've compiled with @-eventlog@ set) and the

907+-- relevant @+RTS@ options given, then the message will be sent to the event log.

908+-- If neither facility is in use then trace messages will be silently dropped.

909+trace :: String -> Process ()

910+trace s = do

911+ node <- processNode <$> ask

912+ liftIO $ Trace.trace (localTracer node) s

913+

895--------------------------------------------------------------------------------

914--------------------------------------------------------------------------------

896-- Auxiliary functions --

915-- Auxiliary functions --

897--------------------------------------------------------------------------------

916--------------------------------------------------------------------------------

distributed-process/src/Control/Distributed/Process/Internal/Trace.hsnew file+122

@@ -0,0 +1,122 @@

1+-- | Simple (internal) system logging/tracing support.

2+module Control.Distributed.Process.Internal.Trace

3+ ( Tracer

4+ , TraceArg(..)

5+ , trace

6+ , traceFormat

7+ , startTracing

8+ , stopTracer

9+ ) where

10+

11+import Control.Concurrent (forkIO)

12+import Control.Concurrent.Chan (writeChan)

13+import Control.Concurrent.STM

14+ ( TQueue

15+ , newTQueueIO

16+ , readTQueue

17+ , writeTQueue

18+ , atomically

19+ )

20+import Control.Distributed.Process.Internal.Types

21+ ( Tracer(..)

22+ , LocalNode(..)

23+ , NCMsg(..)

24+ , Identifier(ProcessIdentifier)

25+ , ProcessSignal(NamedSend)

26+ , forever'

27+ , nullProcessId

28+ , createMessage

29+ )

30+import Control.Exception

31+ ( catch

32+ , throwTo

33+ , SomeException

34+ , AsyncException(ThreadKilled)

35+ )

36+import Data.List (intersperse)

37+import Data.Time.Clock (getCurrentTime)

38+import Data.Time.Format (formatTime)

39+import Debug.Trace (traceEventIO)

40+

41+import Prelude hiding (catch)

42+

43+import System.Environment (getEnv)

44+import System.IO

45+ ( Handle

46+ , IOMode(AppendMode)

47+ , BufferMode(..)

48+ , openFile

49+ , hClose

50+ , hPutStrLn

51+ , hSetBuffering

52+ )

53+import System.Locale (defaultTimeLocale)

54+

55+data TraceArg =

56+ TraceStr String

57+ | forall a. (Show a) => Trace a

58+

59+startTracing :: LocalNode -> IO LocalNode

60+startTracing node = do

61+ tracer <- defaultTracer node

62+ return node { localTracer = tracer }

63+

64+defaultTracer :: LocalNode -> IO Tracer

65+defaultTracer node = do

66+ catch (getEnv "DISTRIBUTED_PROCESS_TRACE_FILE" >>= logfileTracer)

67+ (\(_ :: IOError) -> defaultTracerAux node)

68+

69+defaultTracerAux :: LocalNode -> IO Tracer

70+defaultTracerAux node = do

71+ catch (getEnv "DISTRIBUTED_PROCESS_TRACE_CONSOLE" >> procTracer node)

72+ (\(_ :: IOError) -> return (EventLogTracer traceEventIO))

73+ where procTracer :: LocalNode -> IO Tracer

74+ procTracer n = return $ (LocalNodeTracer n)

75+

76+logfileTracer :: FilePath -> IO Tracer

77+logfileTracer p = do

78+ q <- newTQueueIO

79+ h <- openFile p AppendMode

80+ hSetBuffering h LineBuffering

81+ tid <- forkIO $ logger h q `catch` (\(_ :: SomeException) ->

82+ hClose h >> return ())

83+ return $ LogFileTracer tid q h

84+ where logger :: Handle -> TQueue String -> IO ()

85+ logger h q' = forever' $ do

86+ msg <- atomically $ readTQueue q'

87+ now <- getCurrentTime

88+ hPutStrLn h $ msg ++ (formatTime defaultTimeLocale " - %c" now)

89+

90+-- TODO: compatibility layer for GHC/base versions (e.g., where's killThread?)

91+

92+stopTracer :: Tracer -> IO () -- overzealous but harmless duplication of hClose

93+stopTracer (LogFileTracer tid _ h) = throwTo tid ThreadKilled >> hClose h

94+stopTracer _ = return ()

95+

96+trace :: Tracer -> String -> IO ()

97+trace (LogFileTracer _ q _) msg = atomically $ writeTQueue q msg

98+trace (LocalNodeTracer n) msg = sendTraceMsg n msg

99+trace (EventLogTracer t) msg = t msg

100+trace InactiveTracer _ = return ()

101+

102+traceFormat :: Tracer

103+ -> String

104+ -> [TraceArg]

105+ -> IO ()

106+traceFormat t d ls =

107+ trace t $ concat (intersperse d (map toS ls))

108+ where toS :: TraceArg -> String

109+ toS (TraceStr s) = s

110+ toS (Trace a) = show a

111+

112+sendTraceMsg :: LocalNode -> String -> IO ()

113+sendTraceMsg node string = do

114+ now <- getCurrentTime

115+ msg <- return $ (formatTime defaultTimeLocale "%c" now, string)

116+ emptyPid <- return $ (nullProcessId (localNodeId node))

117+ traceMsg <- return $ NCMsg {

118+ ctrlMsgSender = ProcessIdentifier (emptyPid)

119+ , ctrlMsgSignal = (NamedSend "logger" (createMessage msg))

120+ }

121+ writeChan (localCtrlChan node) traceMsg

122+

distributed-process/src/Control/Distributed/Process/Internal/Types.hs+52−28

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

14 , nullProcessId

15 -- * Local nodes and processes

16 , LocalNode(..)

17+ , Tracer(..)

17 , LocalNodeState(..)

18 , LocalNodeState(..)

18 , LocalProcess(..)

19 , LocalProcess(..)

19 , LocalProcessState(..)

20 , LocalProcessState(..)

@@ -67,6 +68,8 @@ module Control.Distributed.Process.Internal.Types

67 , channelCounter

68 , channelCounter

68 , typedChannels

69 , typedChannels

69 , typedChannelWithId

70 , typedChannelWithId

71+ -- * Utilities

72+ , forever'

70 ) where

73 ) where

72import System.Mem.Weak (Weak)

75import System.Mem.Weak (Weak)

@@ -88,6 +91,7 @@ import Control.Exception (Exception)

88import Control.Concurrent (ThreadId)

91import Control.Concurrent (ThreadId)

89import Control.Concurrent.Chan (Chan)

92import Control.Concurrent.Chan (Chan)

90import Control.Concurrent.STM (STM)

93import Control.Concurrent.STM (STM)

94+import qualified Control.Concurrent.STM as STM (TQueue)

91import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)

95import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)

92import Control.Applicative (Applicative, Alternative, (<$>), (<*>))

96import Control.Applicative (Applicative, Alternative, (<$>), (<*>))

93import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)

97import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)

@@ -106,6 +110,7 @@ import Control.Distributed.Process.Internal.StrictMVar (StrictMVar)

106import Control.Distributed.Process.Internal.WeakTQueue (TQueue)

110import Control.Distributed.Process.Internal.WeakTQueue (TQueue)

107import Control.Distributed.Static (RemoteTable, Closure)

111import Control.Distributed.Static (RemoteTable, Closure)

108import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC (mapMaybe)

112import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC (mapMaybe)

113+import System.IO (Handle)

109

114

110--------------------------------------------------------------------------------

115--------------------------------------------------------------------------------

111-- Node and process identifiers --

116-- Node and process identifiers --

@@ -175,19 +180,28 @@ nullProcessId nid =

175-- Local nodes and processes --

180-- Local nodes and processes --

176--------------------------------------------------------------------------------

181--------------------------------------------------------------------------------

177

182

183+-- | Required for system tracing in the node controller

184+data Tracer =

185+ LogFileTracer !ThreadId !(STM.TQueue String) !Handle

186+ | EventLogTracer !(String -> IO ())

187+ | LocalNodeTracer !LocalNode

188+ | InactiveTracer -- NB: never used, this is required to initialize LocalNode

189+

178-- | Local nodes

190-- | Local nodes

179data LocalNode = LocalNode

191data LocalNode = LocalNode

180 { -- | 'NodeId' of the node

192 { -- | 'NodeId' of the node

181− localNodeId :: !NodeId

193+ localNodeId :: !NodeId

182 -- | The network endpoint associated with this node

194 -- | The network endpoint associated with this node

183− , localEndPoint :: !NT.EndPoint

195+ , localEndPoint :: !NT.EndPoint

184 -- | Local node state

196 -- | Local node state

185− , localState :: !(StrictMVar LocalNodeState)

197+ , localState :: !(StrictMVar LocalNodeState)

186 -- | Channel for the node controller

198 -- | Channel for the node controller

187− , localCtrlChan :: !(Chan NCMsg)

199+ , localCtrlChan :: !(Chan NCMsg)

200+ -- | Current active system debug/trace log

201+ , localTracer :: !Tracer

188 -- | Runtime lookup table for supporting closures

202 -- | Runtime lookup table for supporting closures

189 -- TODO: this should be part of the CH state, not the local endpoint state

203 -- TODO: this should be part of the CH state, not the local endpoint state

190− , remoteTable :: !RemoteTable

204+ , remoteTable :: !RemoteTable

191 }

205 }

192

206

193data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect

207data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect

@@ -432,7 +446,7 @@ data WhereIsReply = WhereIsReply String (Maybe ProcessId)

432data RegisterReply = RegisterReply String Bool

446data RegisterReply = RegisterReply String Bool

433 deriving (Show, Typeable)

447 deriving (Show, Typeable)

434

448

435−-- | Provide information about a running process

449+-- | Provide information about a running process

436data ProcessInfo = ProcessInfo {

450data ProcessInfo = ProcessInfo {

437 infoNode :: NodeId

451 infoNode :: NodeId

438 , infoRegisteredNames :: [String]

452 , infoRegisteredNames :: [String]

@@ -441,21 +455,9 @@ data ProcessInfo = ProcessInfo {

441 , infoLinks :: [ProcessId]

455 , infoLinks :: [ProcessId]

442 } deriving (Show, Eq, Typeable)

456 } deriving (Show, Eq, Typeable)

443

457

444−instance Binary ProcessInfo where

445− get = ProcessInfo <$> get <*> get <*> get <*> get <*> get

446− put pInfo = put (infoNode pInfo)

447− >> put (infoRegisteredNames pInfo)

448− >> put (infoMessageQueueLength pInfo)

449− >> put (infoMonitors pInfo)

450− >> put (infoLinks pInfo)

451−

452data ProcessInfoNone = ProcessInfoNone DiedReason

458data ProcessInfoNone = ProcessInfoNone DiedReason

453 deriving (Show, Typeable)

459 deriving (Show, Typeable)

454

460

455−instance Binary ProcessInfoNone where

456− get = ProcessInfoNone <$> get

457− put (ProcessInfoNone r) = put r

458−

459--------------------------------------------------------------------------------

461--------------------------------------------------------------------------------

460-- Node controller internal data types --

462-- Node controller internal data types --

461--------------------------------------------------------------------------------

463--------------------------------------------------------------------------------

@@ -531,16 +533,16 @@ instance Binary ProcessSignal where

531 get = do

533 get = do

532 header <- getWord8

534 header <- getWord8

533 case header of

535 case header of

534− 0 -> Link <$> get

536+ 0 -> Link <$> get

535− 1 -> Unlink <$> get

537+ 1 -> Unlink <$> get

536− 2 -> Monitor <$> get

538+ 2 -> Monitor <$> get

537− 3 -> Unmonitor <$> get

539+ 3 -> Unmonitor <$> get

538− 4 -> Died <$> get <*> get

540+ 4 -> Died <$> get <*> get

539− 5 -> Spawn <$> get <*> get

541+ 5 -> Spawn <$> get <*> get

540− 6 -> WhereIs <$> get

542+ 6 -> WhereIs <$> get

541− 7 -> Register <$> get <*> get <*> get <*> get

543+ 7 -> Register <$> get <*> get <*> get <*> get

542− 8 -> NamedSend <$> get <*> (payloadToMessage <$> get)

544+ 8 -> NamedSend <$> get <*> (payloadToMessage <$> get)

543− 9 -> Kill <$> get <*> get

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

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

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

545 30 -> GetInfo <$> get

547 30 -> GetInfo <$> get

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

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

@@ -589,6 +591,18 @@ instance Binary RegisterReply where

589 put (RegisterReply label ok) = put label >> put ok

591 put (RegisterReply label ok) = put label >> put ok

590 get = RegisterReply <$> get <*> get

592 get = RegisterReply <$> get <*> get

591

593

594+instance Binary ProcessInfo where

595+ get = ProcessInfo <$> get <*> get <*> get <*> get <*> get

596+ put pInfo = put (infoNode pInfo)

597+ >> put (infoRegisteredNames pInfo)

598+ >> put (infoMessageQueueLength pInfo)

599+ >> put (infoMonitors pInfo)

600+ >> put (infoLinks pInfo)

601+

602+instance Binary ProcessInfoNone where

603+ get = ProcessInfoNone <$> get

604+ put (ProcessInfoNone r) = put r

605+

592--------------------------------------------------------------------------------

606--------------------------------------------------------------------------------

593-- Accessors --

607-- Accessors --

594--------------------------------------------------------------------------------

608--------------------------------------------------------------------------------

@@ -625,3 +639,13 @@ typedChannels = accessor _typedChannels (\cs st -> st { _typedChannels = cs })

625

639

626typedChannelWithId :: LocalSendPortId -> Accessor LocalProcessState (Maybe TypedChannel)

640typedChannelWithId :: LocalSendPortId -> Accessor LocalProcessState (Maybe TypedChannel)

627typedChannelWithId cid = typedChannels >>> DAC.mapMaybe cid

641typedChannelWithId cid = typedChannels >>> DAC.mapMaybe cid

642+

643+--------------------------------------------------------------------------------

644+-- Utilities --

645+--------------------------------------------------------------------------------

646+

647+-- Like 'Control.Monad.forever' but sans space leak

648+{-# INLINE forever' #-}

649+forever' :: Monad m => m a -> m b

650+forever' a = let a' = a >> a' in a'

651+

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

@@ -62,7 +62,9 @@ import Control.Distributed.Process.Internal.StrictMVar

62 , takeMVar

63 )

64import Control.Concurrent.Chan (newChan, writeChan, readChan)

65−import Control.Concurrent.STM (atomically)

65+import Control.Concurrent.STM

66+ ( atomically

67+ )

66import Control.Distributed.Process.Internal.CQueue

68import Control.Distributed.Process.Internal.CQueue

67 ( CQueue

69 ( CQueue

68 , enqueue

70 , enqueue

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

97 , LocalProcessId(..)

99 , LocalProcessId(..)

98 , ProcessId(..)

100 , ProcessId(..)

99 , LocalNode(..)

101 , LocalNode(..)

102+ , Tracer(InactiveTracer)

100 , LocalNodeState(..)

103 , LocalNodeState(..)

101 , LocalProcess(..)

104 , LocalProcess(..)

102 , LocalProcessState(..)

105 , LocalProcessState(..)

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

108 , localPidUnique

111 , localPidUnique

109 , localProcessWithId

112 , localProcessWithId

110 , localConnections

113 , localConnections

114+ , forever'

111 , MonitorRef(..)

115 , MonitorRef(..)

112 , ProcessMonitorNotification(..)

116 , ProcessMonitorNotification(..)

113 , NodeMonitorNotification(..)

117 , NodeMonitorNotification(..)

@@ -139,6 +143,12 @@ import Control.Distributed.Process.Internal.Types

139 , firstNonReservedProcessId

143 , firstNonReservedProcessId

140 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

144 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

141 )

145 )

146+import Control.Distributed.Process.Internal.Trace

147+ ( TraceArg(..)

148+ , traceFormat

149+ , startTracing

150+ , stopTracer

151+ )

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

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

143import Control.Distributed.Process.Internal.Messaging

153import Control.Distributed.Process.Internal.Messaging

144 ( sendBinary

154 ( sendBinary

@@ -195,16 +205,13 @@ createBareLocalNode endPoint rtable = do

195 , localEndPoint = endPoint

205 , localEndPoint = endPoint

196 , localState = state

206 , localState = state

197 , localCtrlChan = ctrlChan

207 , localCtrlChan = ctrlChan

208+ , localTracer = InactiveTracer

198 , remoteTable = rtable

209 , remoteTable = rtable

199 }

210 }

211+ tracedNode <- startTracing node

200− void . forkIO $ runNodeController node

212+ void . forkIO $ runNodeController tracedNode

201− void . forkIO $ handleIncomingMessages node

213+ void . forkIO $ handleIncomingMessages tracedNode

202− return node

214+ return tracedNode

203−

204−-- Like 'Control.Monad.forever' but sans space leak

205−{-# INLINE forever' #-}

206−forever' :: Monad m => m a -> m b

207−forever' a = let a' = a >> a' in a'

208

215

209-- | Start and register the service processes on a node

216-- | Start and register the service processes on a node

210-- (for now, this is only the logger)

217-- (for now, this is only the logger)

@@ -218,6 +225,10 @@ startServiceProcesses node = do

218 [ match $ \((time, pid, string) ::(String, ProcessId, String)) -> do

225 [ match $ \((time, pid, string) ::(String, ProcessId, String)) -> do

219 liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string

226 liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string

220 loop

227 loop

228+ , match $ \((time, string) :: (String, String)) -> do

229+ -- this is a 'trace' message from the local node tracer

230+ liftIO . hPutStrLn stderr $ time ++ " [trace] " ++ string

231+ loop

221 , match $ \(ch :: SendPort ()) -> -- a shutdown request

232 , match $ \(ch :: SendPort ()) -> -- a shutdown request

222 sendChan ch ()

233 sendChan ch ()

223 ]

234 ]

@@ -267,6 +278,7 @@ forkProcess node proc = modifyMVar (localState node) startProcess

267 reason <- Exception.catch

278 reason <- Exception.catch

268 (runLocalProcess lproc proc >> return DiedNormal)

279 (runLocalProcess lproc proc >> return DiedNormal)

269 (return . DiedException . (show :: SomeException -> String))

280 (return . DiedException . (show :: SomeException -> String))

281+

270 -- [Unified: Table 4, rules termination and exiting]

282 -- [Unified: Table 4, rules termination and exiting]

271 modifyMVar_ (localState node) (cleanupProcess pid)

283 modifyMVar_ (localState node) (cleanupProcess pid)

272 writeChan (localCtrlChan node) NCMsg

284 writeChan (localCtrlChan node) NCMsg

@@ -423,22 +435,27 @@ handleIncomingMessages node = go initConnectionState

423 $ st

435 $ st

424 )

436 )

425 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->

437 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->

426− fail $ "Cloud Haskell fatal error: end point failed: " ++ str

438+ fatal $ "Cloud Haskell fatal error: end point failed: " ++ str

427 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->

439 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->

428− fail $ "Cloud Haskell fatal error: transport failed: " ++ str

440+ fatal $ "Cloud Haskell fatal error: transport failed: " ++ str

429 NT.EndPointClosed ->

441 NT.EndPointClosed ->

430− return ()

442+ stopTracer (localTracer node) >> return ()

431 NT.ReceivedMulticast _ _ ->

443 NT.ReceivedMulticast _ _ ->

432 -- If we received a multicast message, something went horribly wrong

444 -- If we received a multicast message, something went horribly wrong

433 -- and we just give up

445 -- and we just give up

434− fail "Cloud Haskell fatal error: received unexpected multicast"

446+ fatal "Cloud Haskell fatal error: received unexpected multicast"

447+

448+ fatal :: String -> IO ()

449+ fatal msg = stopTracer (localTracer node) >> fail msg

435

450

436 invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()

451 invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()

437− invalidRequest cid st =

452+ invalidRequest cid st = do

438 -- TODO: We should treat this as a fatal error on the part of the remote

453 -- TODO: We should treat this as a fatal error on the part of the remote

439 -- node. That is, we should report the remote node as having died, and we

454 -- node. That is, we should report the remote node as having died, and we

440 -- should close incoming connections (this requires a Transport layer

455 -- should close incoming connections (this requires a Transport layer

441 -- extension).

456 -- extension).

457+ traceEventFmtIO node "" [(TraceStr "[network] invalid request: "),

458+ (Trace cid)]

442 go ( incomingAt cid ^= Nothing

459 go ( incomingAt cid ^= Nothing

443 $ st

460 $ st

444 )

461 )

@@ -490,6 +507,35 @@ instance Show ProcessKillException where

490 show (ProcessKillException pid reason) =

507 show (ProcessKillException pid reason) =

491 "killed-by=" ++ show pid ++ ",reason=" ++ reason

508 "killed-by=" ++ show pid ++ ",reason=" ++ reason

492

509

510+--------------------------------------------------------------------------------

511+-- Tracing/Debugging --

512+--------------------------------------------------------------------------------

513+

514+-- [Issue #104]

515+

516+traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC ()

517+traceNotifyDied node ident reason =

518+ case reason of

519+ DiedNormal -> return ()

520+ _ -> traceNcEventFmt node " "

521+ [(TraceStr "[node-controller]"),

522+ (Trace ident),

523+ (Trace reason)]

524+

525+traceNcEventFmt :: LocalNode -> String -> [TraceArg] -> NC ()

526+traceNcEventFmt node fmt args =

527+ liftIO $ traceEventFmtIO node fmt args

528+

529+traceEventFmtIO :: LocalNode

530+ -> String

531+ -> [TraceArg]

532+ -> IO ()

533+traceEventFmtIO node fmt args =

534+ withLocalTracer node $ \t -> traceFormat t fmt args

535+

536+withLocalTracer :: LocalNode -> (Tracer -> IO ()) -> IO ()

537+withLocalTracer node act = act (localTracer node)

538+

493--------------------------------------------------------------------------------

539--------------------------------------------------------------------------------

494-- Core functionality --

540-- Core functionality --

495--------------------------------------------------------------------------------

541--------------------------------------------------------------------------------

@@ -597,6 +643,7 @@ ncEffectUnmonitor from ref = do

597ncEffectDied :: Identifier -> DiedReason -> NC ()

643ncEffectDied :: Identifier -> DiedReason -> NC ()

598ncEffectDied ident reason = do

644ncEffectDied ident reason = do

599 node <- ask

645 node <- ask

646+ traceNotifyDied node ident reason

600 (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))

647 (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))

601 (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors))

648 (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors))

602

649

@@ -639,7 +686,6 @@ ncEffectDied ident reason = do

639 , ctrlMsgSignal = Died ident reason

686 , ctrlMsgSignal = Died ident reason

640 }

687 }

641

688

642−

643-- [Unified: Table 13]

689-- [Unified: Table 13]

644ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()

690ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()

645ncEffectSpawn pid cProc ref = do

691ncEffectSpawn pid cProc ref = do

@@ -841,18 +887,18 @@ notifyDied dest src reason mRef = do

841

887

842-- | [Unified: Table 8]

888-- | [Unified: Table 8]

843destNid :: ProcessSignal -> Maybe NodeId

889destNid :: ProcessSignal -> Maybe NodeId

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

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

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

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

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

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

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

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

848−destNid (Spawn _ _) = Nothing

894+destNid (Spawn _ _) = Nothing

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

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

850−destNid (WhereIs _) = Nothing

896+destNid (WhereIs _) = Nothing

851−destNid (NamedSend _ _) = Nothing

897+destNid (NamedSend _ _) = Nothing

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

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

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

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

854-- care of notifying remote nodes

900-- care of notifying remote nodes

855−destNid (Died _ _) = Nothing

901+destNid (Died _ _) = Nothing

856destNid (Kill pid _) = Just $ processNodeId pid

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

857destNid (Exit pid _) = Just $ processNodeId pid

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

858destNid (GetInfo pid) = Just $ processNodeId pid

904destNid (GetInfo pid) = Just $ processNodeId pid

distributed-process/tests/TestStats.hs+12−12

@@ -2,13 +2,12 @@

2{-# OPTIONS_GHC -fno-warn-orphans #-}

3module Main where

5−import Prelude hiding (catch, log)

5+import Control.Concurrent.MVar

6−import Test.Framework

6+ ( MVar

7− ( Test

7+ , newEmptyMVar

8− , defaultMain

8+ , putMVar

9− , testGroup

9+ , takeMVar

10 )

11−import Network.Transport.TCP

12import Control.Distributed.Process

11import Control.Distributed.Process

13import Control.Distributed.Process.Node

12import Control.Distributed.Process.Node

14 ( forkProcess

13 ( forkProcess

@@ -16,14 +15,15 @@ import Control.Distributed.Process.Node

16 , initRemoteTable

15 , initRemoteTable

17 , closeLocalNode

16 , closeLocalNode

18 , LocalNode)

17 , LocalNode)

19−import Control.Concurrent.MVar

20− ( MVar

21− , newEmptyMVar

22− , putMVar

23− , takeMVar

24− )

25import Data.Binary()

18import Data.Binary()

26import Data.Typeable()

19import Data.Typeable()

20+import Network.Transport.TCP

21+import Prelude hiding (catch, log)

22+import Test.Framework

23+ ( Test

24+ , defaultMain

25+ , testGroup

26+ )

27import Test.HUnit (Assertion)

28import Test.HUnit.Base (assertBool)

29import Test.Framework.Providers.HUnit (testCase)