9bbd523 · distributed-process

git.iko.soy/distributed-process

Add support for 'call'

commit9bbd52351031dfc134147335fbbc57190b53b4b1 authorEdsko de Vries <edsko@edsko.net> dateFri 22 Jun 2012, 12:39 (14 years ago) parentc5fff42

4 files changed+94−23

compare with master

unifiedsplit

distributed-process/src/Control/Distributed/Process.hs+10−1 distributed-process/src/Control/Distributed/Process/Closure.hs+51−18 distributed-process/src/Control/Distributed/Process/Internal/Types.hs+8−3 distributed-process/tests/TestClosure.hs+25−1

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

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

3232 , matchUnknown

3333 -- * Process management

3434 , spawn

35+ , call

3536 , SpawnRef

3637 , getSelfPid

3738 , getSelfNode

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

5859 ) where

5960

6061import Prelude hiding (catch)

61−import Data.Binary (decode)

62+import Data.Binary (decode, encode)

6263import Data.Typeable (Typeable)

6364import Data.Dynamic (fromDyn)

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

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

108109 , spawnCounter

109110 , MessageT

110111 , Closure(..)

112+ , CallReply(..)

111113 , SendPort(..)

112114 , ReceivePort(..)

113115 , channelCounter

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

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

301303 ]

302304

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+

303312-- | Our own process ID

304313getSelfPid :: Process ProcessId

305314getSelfPid = processId <$> ask

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

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

4747import qualified Data.Map as Map (insert)

4848import Data.Dynamic (toDyn)

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

50+import Control.Monad (join)

5051import Language.Haskell.TH

5152 ( -- Q monad and operations

5253 Q

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

7778 ( RemoteTable

7879 , Closure(..)

7980 , Static(..)

81+ , Process

82+ , ProcessId

83+ , CallReply(..)

8084 )

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

8186

8287--------------------------------------------------------------------------------

8388-- Top-level API --

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

8792-- of functions

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

8994remotable ns = do

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

91− rtable <- createMetaData inserts

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

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

96+ rtable <- createMetaData (concat inserts)

97+ return $ concat closures ++ rtable

9398

9499-- | Create a closure

95100mkClosure :: Name -> Q Exp

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

110115--

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

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

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

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

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

116−process n = do

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')

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

124+generateDefs n = do

117125 mType <- getType n

118126 case mType of

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

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

121− decoder <- generateDecoder origName (return res)

122− let decoderE = varE (decoderName n)

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

124− return (closure, decoder, insert)

129+ let decoder = generateDecoder origName (return res)

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 $

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

142+

143+ return (closure, insert : insert')

125144 _ ->

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

127146

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

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

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

130149generateClosure n arg res = do

131150 closure <- sequence

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

137156 label :: String

138157 label = show $ n

139158

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

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

142−generateDecoder n res = do

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

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

145− ]

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+

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

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

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

170+

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 ()

178+ |]

146179

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

148181closureName :: Name -> Name

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

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

2525 -- * Closures

2626 , Static(..)

2727 , Closure(..)

28+ , CallReply(..)

2829 -- * Messages

2930 , Message(..)

3031 -- * Node controller user-visible data types

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

7778import Control.Concurrent.Chan (Chan)

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

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

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

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

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

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

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

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

181182newtype Process a = Process {

182183 unProcess :: ReaderT LocalProcess (MessageT IO) a

183184 }

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

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

185186

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

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

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

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

227228 deriving (Typeable, Show)

228229

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

231+newtype CallReply a = CallReply a

232+ deriving (Typeable, Binary)

233+

229234--------------------------------------------------------------------------------

230235-- Messages --

231236--------------------------------------------------------------------------------

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

449454--------------------------------------------------------------------------------

450455

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

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

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

453458

454459data MessageState = MessageState {

455460 messageLocalNode :: LocalNode

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

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

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

44import Control.Concurrent (forkIO)

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

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

67import Network.Transport (Transport)

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

89import Control.Distributed.Process

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

2425 fromPid <- getSelfPid

2526 send toPid fromPid

2627

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

28+factorial :: Int -> Process Int

29+factorial 0 = return 1

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

31+

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

2833

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

3035testSendPureClosure transport rtable = do

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

116121

117122 takeMVar clientDone

118123

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+

119142main :: IO ()

120143main = do

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

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

125148 , ("SendIOClosure", testSendIOClosure transport rtable)

126149 , ("SendProcClosure", testSendProcClosure transport rtable)

127150 , ("Spawn", testSpawn transport rtable)

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

128152 ]