9bbd523 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Add support for 'call'

9bbd523Edsko de Vries14 years ago

distributed-process/src/Control/Distributed/Process.hs+10−1

compare with master

unifiedsplit

@@ -32,6 +32,7 @@ module Control.Distributed.Process

32 , matchUnknown

33 -- * Process management

34 , spawn

35+ , call

35 , SpawnRef

36 , SpawnRef

36 , getSelfPid

37 , getSelfPid

37 , getSelfNode

38 , getSelfNode

@@ -58,7 +59,7 @@ module Control.Distributed.Process

58 ) where

59 ) where

60import Prelude hiding (catch)

61import Prelude hiding (catch)

61−import Data.Binary (decode)

62+import Data.Binary (decode, encode)

62import Data.Typeable (Typeable)

63import Data.Typeable (Typeable)

63import Data.Dynamic (fromDyn)

64import Data.Dynamic (fromDyn)

64import qualified Data.Map as Map ((!))

65import qualified Data.Map as Map ((!))

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

108 , spawnCounter

109 , spawnCounter

109 , MessageT

110 , MessageT

110 , Closure(..)

111 , Closure(..)

112+ , CallReply(..)

111 , SendPort(..)

113 , SendPort(..)

112 , ReceivePort(..)

114 , ReceivePort(..)

113 , channelCounter

115 , channelCounter

@@ -300,6 +302,13 @@ spawn nid proc = do

300 (\(DidSpawn _ pid) -> return pid)

302 (\(DidSpawn _ pid) -> return pid)

301 ]

303 ]

302

304

305+call :: Serializable a => NodeId -> Closure (Process a) -> Process a

306+call nid proc@(Closure (Static label) _) = do

307+ us <- getSelfPid

308+ spawn nid (Closure (Static (label ++ "__call")) (encode (proc, us)))

309+ CallReply a <- expect

310+ return a

311+

303-- | Our own process ID

312-- | Our own process ID

304getSelfPid :: Process ProcessId

313getSelfPid :: Process ProcessId

305getSelfPid = processId <$> ask

314getSelfPid = processId <$> ask

distributed-process/src/Control/Distributed/Process/Closure.hs+51−18

@@ -47,6 +47,7 @@ import Data.Binary (encode, decode)

47import qualified Data.Map as Map (insert)

48import Data.Dynamic (toDyn)

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

50+import Control.Monad (join)

50import Language.Haskell.TH

51import Language.Haskell.TH

51 ( -- Q monad and operations

52 ( -- Q monad and operations

52 Q

53 Q

@@ -77,7 +78,11 @@ import Control.Distributed.Process.Internal.Types

77 ( RemoteTable

78 ( RemoteTable

78 , Closure(..)

79 , Closure(..)

79 , Static(..)

80 , Static(..)

81+ , Process

82+ , ProcessId

83+ , CallReply(..)

80 )

84 )

85+import Control.Distributed.Process (send, unClosure)

82--------------------------------------------------------------------------------

87--------------------------------------------------------------------------------

83-- Top-level API --

88-- Top-level API --

@@ -87,9 +92,9 @@ import Control.Distributed.Process.Internal.Types

87-- of functions

92-- of functions

88remotable :: [Name] -> Q [Dec]

93remotable :: [Name] -> Q [Dec]

89remotable ns = do

94remotable ns = do

90− (closures, decoders, inserts) <- unzip3 <$> mapM process ns

95+ (closures, inserts) <- unzip <$> mapM generateDefs ns

91− rtable <- createMetaData inserts

96+ rtable <- createMetaData (concat inserts)

92− return $ concat closures ++ concat decoders ++ rtable

97+ return $ concat closures ++ rtable

94-- | Create a closure

99-- | Create a closure

95mkClosure :: Name -> Q Exp

100mkClosure :: Name -> Q Exp

@@ -110,22 +115,36 @@ createMetaData is =

110--

115--

111-- Given an (f :: a -> b) in module M, create:

116-- Given an (f :: a -> b) in module M, create:

112-- 1. f__closure :: a -> Closure b,

117-- 1. f__closure :: a -> Closure b,

113−-- 2. f__dec :: ByteString -> b

114−-- 3. Map.insert "M.f" (toDyn f__dec)

118+-- 2. Map.insert "M.f" (toDyn ((f . enc) :: ByteString -> b))

119+--

120+-- Moreover, if b is of the form Process c, then additionally create

121+-- 3. Map.insert "M.f__call" (... :: ByteString -> Process ())

122+-- (see 'generateCallClosure')

115−process :: Name -> Q ([Dec], [Dec], Q Exp)

123+generateDefs :: Name -> Q ([Dec], [Q Exp])

116−process n = do

124+generateDefs n = do

117 mType <- getType n

125 mType <- getType n

118 case mType of

126 case mType of

119 Just (origName, ArrowT `AppT` arg `AppT` res) -> do

127 Just (origName, ArrowT `AppT` arg `AppT` res) -> do

120 (closure, label) <- generateClosure origName (return arg) (return res)

128 (closure, label) <- generateClosure origName (return arg) (return res)

121− decoder <- generateDecoder origName (return res)

129+ let decoder = generateDecoder origName (return res)

122− let decoderE = varE (decoderName n)

130+ insert = [| Map.insert $(stringE label) (toDyn $decoder) |]

131+

132+ -- Generate special closure to support 'call'

133+ mResult <- processResult res

134+ insert' <- case mResult of

135+ Nothing ->

136+ return []

137+ Just result -> do

138+ let encoder = generateCallClosure origName (return result)

139+ label' = label ++ "__call"

140+ return . return $

123− insert = [| Map.insert $(stringE label) (toDyn $decoderE) |]

141+ [| Map.insert $(stringE label') (toDyn $encoder) |]

142+

124− return (closure, decoder, insert)

143+ return (closure, insert : insert')

125 _ ->

144 _ ->

126 fail $ "remotable: " ++ show n ++ " is not a function"

145 fail $ "remotable: " ++ show n ++ " is not a function"

127

146

128−-- | Generate the closure creator (see 'process')

147+-- | Generate the closure creator (see 'generateDefs')

129generateClosure :: Name -> Q Type -> Q Type -> Q ([Dec], String)

148generateClosure :: Name -> Q Type -> Q Type -> Q ([Dec], String)

130generateClosure n arg res = do

149generateClosure n arg res = do

131 closure <- sequence

150 closure <- sequence

@@ -137,12 +156,26 @@ generateClosure n arg res = do

137 label :: String

156 label :: String

138 label = show $ n

157 label = show $ n

139

158

159+-- | If 't' is of the form 'Process b', return 'Just b'

160+processResult :: Type -> Q (Maybe Type)

161+processResult t = do

162+ process <- [t| Process |]

163+ case t of

164+ p `AppT` a | p == process -> return $ Just a

165+ _ -> return Nothing

166+

140−-- | Generate the decoder (see 'process')

167+-- | Generate the decoder (see 'generateDefs')

141−generateDecoder :: Name -> Q Type -> Q [Dec]

168+generateDecoder :: Name -> Q Type -> Q Exp

142−generateDecoder n res = do

169+generateDecoder n res = [| $(varE n) . decode :: ByteString -> $res |]

143− sequence [ sigD (decoderName n) [t| ByteString -> $res |]

170+

144− , sfnD (decoderName n) [| $(varE n) . decode |]

171+-- | Generate the call-closure. This is necessary to support 'call'

172+generateCallClosure :: Name -> Q Type -> Q Exp

173+generateCallClosure n t =

174+ [| (\env -> do

175+ let (cl, them) = decode env :: (Closure (Process $t), ProcessId)

176+ join (unClosure cl) >>= send them . CallReply

177+ ) :: ByteString -> Process ()

145− ]

178+ |]

146

179

147-- | The name for the function that generates the closure

180-- | The name for the function that generates the closure

148closureName :: Name -> Name

181closureName :: Name -> Name

distributed-process/src/Control/Distributed/Process/Internal/Types.hs+8−3

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

25 -- * Closures

26 , Static(..)

27 , Closure(..)

28+ , CallReply(..)

28 -- * Messages

29 -- * Messages

29 , Message(..)

30 , Message(..)

30 -- * Node controller user-visible data types

31 -- * Node controller user-visible data types

@@ -77,7 +78,7 @@ import Control.Concurrent.MVar (MVar)

77import Control.Concurrent.Chan (Chan)

78import Control.Concurrent.Chan (Chan)

78import Control.Concurrent.STM (TChan, TVar)

79import Control.Concurrent.STM (TChan, TVar)

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

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

80−import Control.Applicative ((<$>), (<*>))

81+import Control.Applicative (Applicative, (<$>), (<*>))

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

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

82import Control.Monad.IO.Class (MonadIO)

83import Control.Monad.IO.Class (MonadIO)

83import Control.Monad.State (MonadState, StateT, evalStateT)

84import Control.Monad.State (MonadState, StateT, evalStateT)

@@ -181,7 +182,7 @@ data LocalProcessState = LocalProcessState

181newtype Process a = Process {

182newtype Process a = Process {

182 unProcess :: ReaderT LocalProcess (MessageT IO) a

183 unProcess :: ReaderT LocalProcess (MessageT IO) a

183 }

184 }

184− deriving (Functor, Monad, MonadIO, MonadReader LocalProcess, Typeable)

185+ deriving (Functor, Monad, MonadIO, MonadReader LocalProcess, Typeable, Applicative)

185

186

186-- | Deconstructor for 'Process' (not exported to the public API)

187-- | Deconstructor for 'Process' (not exported to the public API)

187runLocalProcess :: LocalNode -> Process a -> LocalProcess -> IO a

188runLocalProcess :: LocalNode -> Process a -> LocalProcess -> IO a

@@ -226,6 +227,10 @@ newtype Static a = Static String

226data Closure a = Closure (Static (ByteString -> a)) ByteString

227data Closure a = Closure (Static (ByteString -> a)) ByteString

227 deriving (Typeable, Show)

228 deriving (Typeable, Show)

228

229

230+-- | Used for the implementation of 'call'

231+newtype CallReply a = CallReply a

232+ deriving (Typeable, Binary)

233+

229--------------------------------------------------------------------------------

234--------------------------------------------------------------------------------

230-- Messages --

235-- Messages --

231--------------------------------------------------------------------------------

236--------------------------------------------------------------------------------

@@ -449,7 +454,7 @@ typedChannelWithId cid = typedChannels >>> DAC.mapMaybe cid

449--------------------------------------------------------------------------------

454--------------------------------------------------------------------------------

450

455

451newtype MessageT m a = MessageT { unMessageT :: StateT MessageState m a }

456newtype MessageT m a = MessageT { unMessageT :: StateT MessageState m a }

452− deriving (Functor, Monad, MonadIO, MonadState MessageState)

457+ deriving (Functor, Monad, MonadIO, MonadState MessageState, Applicative)

453

458

454data MessageState = MessageState {

459data MessageState = MessageState {

455 messageLocalNode :: LocalNode

460 messageLocalNode :: LocalNode

distributed-process/tests/TestClosure.hs+25−1

@@ -3,6 +3,7 @@ module Main where

3import Control.Monad.IO.Class (liftIO)

4import Control.Concurrent (forkIO)

5import Control.Concurrent.MVar (MVar, newEmptyMVar, readMVar, takeMVar, putMVar)

6+import Control.Applicative ((<$>))

6import Network.Transport (Transport)

7import Network.Transport (Transport)

7import Network.Transport.TCP (createTransport, defaultTCPParameters)

8import Network.Transport.TCP (createTransport, defaultTCPParameters)

8import Control.Distributed.Process

9import Control.Distributed.Process

@@ -24,7 +25,11 @@ sendPid toPid = do

24 fromPid <- getSelfPid

25 fromPid <- getSelfPid

25 send toPid fromPid

26 send toPid fromPid

28+factorial :: Int -> Process Int

29+factorial 0 = return 1

30+factorial n = (n *) <$> factorial (n - 1)

31+

27−$(remotable ['addInt, 'putInt, 'sendInt, 'sendPid])

32+$(remotable ['addInt, 'putInt, 'sendInt, 'sendPid, 'factorial])

29testSendPureClosure :: Transport -> RemoteTable -> IO ()

34testSendPureClosure :: Transport -> RemoteTable -> IO ()

30testSendPureClosure transport rtable = do

35testSendPureClosure transport rtable = do

@@ -116,6 +121,24 @@ testSpawn transport rtable = do

116

121

117 takeMVar clientDone

122 takeMVar clientDone

118

123

124+testCall :: Transport -> RemoteTable -> IO ()

125+testCall transport rtable = do

126+ serverNodeAddr <- newEmptyMVar

127+ clientDone <- newEmptyMVar

128+

129+ forkIO $ do

130+ node <- newLocalNode transport rtable

131+ putMVar serverNodeAddr (localNodeId node)

132+

133+ forkIO $ do

134+ node <- newLocalNode transport rtable

135+ nid <- readMVar serverNodeAddr

136+ runProcess node $ do

137+ (120 :: Int) <- call nid ($(mkClosure 'factorial) 5)

138+ liftIO $ putMVar clientDone ()

139+

140+ takeMVar clientDone

141+

119main :: IO ()

142main :: IO ()

120main = do

143main = do

121 Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters

144 Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters

@@ -125,4 +148,5 @@ main = do

125 , ("SendIOClosure", testSendIOClosure transport rtable)

148 , ("SendIOClosure", testSendIOClosure transport rtable)

126 , ("SendProcClosure", testSendProcClosure transport rtable)

149 , ("SendProcClosure", testSendProcClosure transport rtable)

127 , ("Spawn", testSpawn transport rtable)

150 , ("Spawn", testSpawn transport rtable)

151+ , ("Call", testCall transport rtable)

128 ]

152 ]