603646d · distributed-process

git.iko.soy/distributed-process

on master help sign in

Add support for logging

603646dEdsko de Vries14 years ago

distributed-process/distributed-process.cabal+7

compare with master

unifiedsplit

@@ -23,6 +23,8 @@ Library

23 ghc-prim,

24 network,

25 network-multicast,

26+ time,

27+ old-locale,

26 network-transport >= 0.1

28 network-transport >= 0.1

27 Exposed-modules: Control.Distributed.Process,

29 Exposed-modules: Control.Distributed.Process,

28 Control.Distributed.Process.Serializable,

30 Control.Distributed.Process.Serializable,

@@ -48,6 +50,8 @@ Test-Suite TestCH

48 template-haskell,

50 template-haskell,

49 stm,

51 stm,

50 ghc-prim,

52 ghc-prim,

53+ time,

54+ old-locale,

51 network-transport >= 0.1

55 network-transport >= 0.1

52 Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell, ExistentialQuantification, RankNTypes, GADTs, ImpredicativeTypes

56 Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell, ExistentialQuantification, RankNTypes, GADTs, ImpredicativeTypes

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

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

@@ -69,6 +73,8 @@ Test-Suite TestClosure

69 ansi-terminal,

73 ansi-terminal,

70 template-haskell,

74 template-haskell,

71 stm,

75 stm,

76+ time,

77+ old-locale,

72 network-transport >= 0.1

78 network-transport >= 0.1

73 Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell, ExistentialQuantification, RankNTypes, GADTs, ImpredicativeTypes

79 Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell, ExistentialQuantification, RankNTypes, GADTs, ImpredicativeTypes

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

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

@@ -89,6 +95,7 @@ Test-Suite TestLocalBackend

89 ansi-terminal,

95 ansi-terminal,

90 template-haskell,

96 template-haskell,

91 stm,

97 stm,

98+ old-locale,

92 network-transport >= 0.1

99 network-transport >= 0.1

93 Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell, ExistentialQuantification, RankNTypes, GADTs, ImpredicativeTypes

100 Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell, ExistentialQuantification, RankNTypes, GADTs, ImpredicativeTypes

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

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

distributed-process/src/Control/Distributed/Process.hs+4

@@ -64,6 +64,8 @@ module Control.Distributed.Process

64 , Static

65 , unClosure

66 , RemoteTable

67+ -- * Logging

68+ , say

67 -- * Registry

69 -- * Registry

68 , register

70 , register

69 , unregister

71 , unregister

@@ -155,6 +157,8 @@ import Control.Distributed.Process.Internal.Primitives

155 , monitorNode

157 , monitorNode

156 , monitorPort

158 , monitorPort

157 , unmonitor

159 , unmonitor

160+ -- Logging

161+ , say

158 -- Registry

162 -- Registry

159 , register

163 , register

160 , unregister

164 , unregister

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

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

29 , unlink

30 , monitor

31 , unmonitor

32+ -- * Logging

33+ , say

32 -- * Registry

34 -- * Registry

33 , register

35 , register

34 , unregister

36 , unregister

@@ -53,6 +55,9 @@ module Control.Distributed.Process.Internal.Primitives

53import Prelude hiding (catch)

55import Prelude hiding (catch)

54import Data.Binary (decode)

56import Data.Binary (decode)

55import Data.Typeable (Typeable,)

57import Data.Typeable (Typeable,)

58+import Data.Time.Clock (getCurrentTime)

59+import Data.Time.Format (formatTime)

60+import System.Locale (defaultTimeLocale)

56import Control.Monad.Reader (ask)

61import Control.Monad.Reader (ask)

57import Control.Monad.IO.Class (MonadIO, liftIO)

62import Control.Monad.IO.Class (MonadIO, liftIO)

58import Control.Applicative ((<$>))

63import Control.Applicative ((<$>))

@@ -362,6 +367,22 @@ unlinkPortAsync :: SendPort a -> Process ()

362unlinkPortAsync (SendPort cid) =

367unlinkPortAsync (SendPort cid) =

363 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid

368 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid

364

369

370+--------------------------------------------------------------------------------

371+-- Logging --

372+--------------------------------------------------------------------------------

373+

374+-- | Log a string

375+--

376+-- @say message@ sends a message (time, pid of the current process, message)

377+-- to the process registered as 'logger'. By default, this process simply

378+-- sends the string to 'stderr'. Individual Cloud Haskell backends might

379+-- replace this with a different logger process, however.

380+say :: String -> Process ()

381+say string = do

382+ now <- liftIO getCurrentTime

383+ us <- getSelfPid

384+ nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)

385+

365--------------------------------------------------------------------------------

386--------------------------------------------------------------------------------

366-- Registry --

387-- Registry --

367--------------------------------------------------------------------------------

388--------------------------------------------------------------------------------

distributed-process/src/Control/Distributed/Process/Node.hs+38−21

@@ -9,7 +9,7 @@ module Control.Distributed.Process.Node

9 ) where

11import Prelude hiding (catch)

12−import System.IO (fixIO)

12+import System.IO (fixIO, hPutStrLn, stderr)

13import qualified Data.ByteString.Lazy as BSL (fromChunks)

14import Data.Binary (decode)

15import Data.Map (Map)

@@ -120,6 +120,7 @@ import Control.Distributed.Process.Internal.Closure

120 , resolveClosure

121 )

122import Control.Distributed.Process.Internal.Node (runLocalProcess)

123+import Control.Distributed.Process.Internal.Primitives (expect, register)

123

124

124--------------------------------------------------------------------------------

125--------------------------------------------------------------------------------

125-- Initialization --

126-- Initialization --

@@ -131,26 +132,42 @@ import Control.Distributed.Process.Internal.Node (runLocalProcess)

131-- to do.

132-- to do.

132newLocalNode :: NT.Transport -> RemoteTable -> IO LocalNode

133newLocalNode :: NT.Transport -> RemoteTable -> IO LocalNode

133newLocalNode transport rtable = do

134newLocalNode transport rtable = do

134− mEndPoint <- NT.newEndPoint transport

135+ mEndPoint <- NT.newEndPoint transport

135− case mEndPoint of

136+ case mEndPoint of

136− Left ex -> throwIO ex

137+ Left ex -> throwIO ex

137− Right endPoint -> do

138+ Right endPoint -> do

139+ localNode <- createBareLocalNode endPoint rtable

140+ startServiceProcesses localNode

141+ return localNode

142+

143+-- | Create a new local node (without any service processes running)

144+createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode

145+createBareLocalNode endPoint rtable = do

138− unq <- randomIO

146+ unq <- randomIO

139− state <- newMVar LocalNodeState

147+ state <- newMVar LocalNodeState

140− { _localProcesses = Map.empty

148+ { _localProcesses = Map.empty

141− , _localPidCounter = 0

149+ , _localPidCounter = 0

142− , _localPidUnique = unq

150+ , _localPidUnique = unq

143− }

151+ }

144− ctrlChan <- newChan

152+ ctrlChan <- newChan

145− let node = LocalNode { localNodeId = NodeId $ NT.address endPoint

153+ let node = LocalNode { localNodeId = NodeId $ NT.address endPoint

146− , localEndPoint = endPoint

154+ , localEndPoint = endPoint

147− , localState = state

155+ , localState = state

148− , localCtrlChan = ctrlChan

156+ , localCtrlChan = ctrlChan

149− , remoteTable = rtable

157+ , remoteTable = rtable

150− }

158+ }

151− void . forkIO $ runNodeController node

159+ void . forkIO $ runNodeController node

152− void . forkIO $ handleIncomingMessages node

160+ void . forkIO $ handleIncomingMessages node

153− return node

161+ return node

162+

163+-- | Start and register the service processes on a node

164+-- (for now, this is only the logger)

165+startServiceProcesses :: LocalNode -> IO ()

166+startServiceProcesses node = do

167+ logger <- forkProcess node . forever $ do

168+ (time, pid, string) <- expect :: Process (String, ProcessId, String)

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

170+ runProcess node $ register "logger" logger

154

171

155-- | Force-close a local node

172-- | Force-close a local node

156--

173--

distributed-process/tests/TestLocalBackend.hs+17−3

@@ -1,18 +1,32 @@

1import System.Environment (getArgs, getProgName)

2−import Control.Distributed.Process.Node (initRemoteTable)

2+import Control.Distributed.Process (say)

3+import Control.Distributed.Process.Node

4+ ( initRemoteTable

5+ , forkProcess

6+ , runProcess

7+ )

3import Control.Distributed.Process.Backend.Local

8import Control.Distributed.Process.Backend.Local

4import Control.Concurrent (threadDelay)

9import Control.Concurrent (threadDelay)

5−import Control.Monad (replicateM)

10+import Control.Monad (forever)

11+import Control.Monad.IO.Class (liftIO)

7server :: LocalBackend -> IO ()

13server :: LocalBackend -> IO ()

8server backend = do

14server backend = do

9 node <- newLocalNode backend

15 node <- newLocalNode backend

16+ runProcess node $

17+ say "this is the server"

18+ {-

10 nodes <- findPeers backend 1000000

19 nodes <- findPeers backend 1000000

11 print nodes

20 print nodes

21+ -}

22+ threadDelay 30000000

13client :: LocalBackend -> IO ()

24client :: LocalBackend -> IO ()

14client backend = do

25client backend = do

15− [node1, node2] <- replicateM 2 $ newLocalNode backend

26+ node <- newLocalNode backend

27+ forkProcess node . forever $ do

28+ say "this is a client node1"

29+ liftIO $ threadDelay 5000000

16 threadDelay 30000000

30 threadDelay 30000000

18main :: IO ()

32main :: IO ()