e0dec7d · distributed-process

git.iko.soy/distributed-process

More compositional way of supporting 'call'

Rather than statically determine which closures we might potentially 'call', instead statically generate a mapping from types of the form 'Process a' to specialized processes (send :: ProcessId -> a -> Process ()) and then do a runtime lookup when we execute a call. This might seem like a step backwards (before we didn't need to do a runtime lookup for which sender to use), but it opens the doors for providing Closure combinators.

commite0dec7d17742d8be9923bcf667cc87e78ef7d781 authorEdsko de Vries <edsko@edsko.net> dateTue 26 Jun 2012, 11:52 (14 years ago) parente3c0bbe

6 files changed+156−95

compare with master

unifiedsplit

distributed-process/distributed-process.cabal+3−3 distributed-process/src/Control/Distributed/Process.hs+15−15 distributed-process/src/Control/Distributed/Process/Closure.hs+19−25 distributed-process/src/Control/Distributed/Process/Internal/Dynamic.hs+26−19 distributed-process/src/Control/Distributed/Process/Internal/Types.hs+82−27 distributed-process/src/Control/Distributed/Process/Node.hs+11−6

distributed-process/distributed-process.cabal+3−3

@@ -26,7 +26,7 @@ Library

2626 Control.Distributed.Process.Serializable,

2727 Control.Distributed.Process.Closure,

2828 Control.Distributed.Process.Node

29− Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell, ExistentialQuantification

29+ Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell, ExistentialQuantification, RankNTypes

3030 ghc-options: -Wall

3131 HS-Source-Dirs: src

3232

@@ -46,7 +46,7 @@ Test-Suite TestCH

4646 stm,

4747 ghc-prim,

4848 network-transport >= 0.1

49− Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell, ExistentialQuantification

49+ Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell, ExistentialQuantification, RankNTypes

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

5151 HS-Source-Dirs: tests src

5252

@@ -67,6 +67,6 @@ Test-Suite TestClosure

6767 template-haskell,

6868 stm,

6969 network-transport >= 0.1

70− Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell, ExistentialQuantification

70+ Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell, ExistentialQuantification, RankNTypes

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

7272 HS-Source-Dirs: tests src

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

@@ -61,9 +61,9 @@ module Control.Distributed.Process

6161 ) where

6262

6363import Prelude hiding (catch)

64+import Data.Maybe (fromJust)

6465import Data.Binary (decode, encode)

65−import Data.Typeable (Typeable)

66−import qualified Data.Map as Map ((!))

66+import Data.Typeable (Typeable, typeOf)

6767import Control.Monad.Reader (ask)

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

6969import qualified Control.Monad.Trans.Class as Trans (lift)

@@ -118,13 +118,14 @@ import Control.Distributed.Process.Internal.Types

118118 , TypedChannel(..)

119119 , ChannelId(..)

120120 , Identifier(..)

121+ , resolveClosure

121122 )

122123import Control.Distributed.Process.Internal.MessageT

123124 ( sendMessage

124125 , sendBinary

125126 , getLocalNode

126127 )

127−import Control.Distributed.Process.Internal.Dynamic (fromDyn)

128+import Control.Distributed.Process.Internal.Dynamic (fromDyn, dynTypeRep)

128129

129130-- INTERNAL NOTES

130131--

@@ -314,9 +315,9 @@ spawn nid proc = do

314315

315316-- | Run a process remotely and wait for it to reply

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

317−call nid proc@(Closure (Static label) _) = do

318+call nid (Closure (Static label) env) = do

318319 us <- getSelfPid

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

320+ spawn nid (Closure (Static "$call") (encode (label, env, us)))

320321 CallReply a <- expect

321322 return a

322323

@@ -366,10 +367,16 @@ unmonitor = sendCtrlMsg Nothing . Unmonitor

366367--------------------------------------------------------------------------------

367368

368369-- | Deserialize a closure

369−unClosure :: Typeable a => Closure a -> Process a

370+unClosure :: forall a. Typeable a => Closure a -> Process a

370371unClosure (Closure (Static label) env) = do

371− dec <- lookupStatic label

372− return (dec env)

372+ rtable <- remoteTable <$> lift getLocalNode

373+ let dyn = fromJust (resolveClosure rtable label)

374+ dec = fromDyn dyn (throw (typeError dyn))

375+ return (dec env)

376+ where

377+ typeError proc = userError $ "lookupStatic type error: "

378+ ++ "cannot match " ++ show (dynTypeRep proc)

379+ ++ "against " ++ show (typeOf (undefined :: a))

373380

374381--------------------------------------------------------------------------------

375382-- Auxiliary API --

@@ -435,12 +442,5 @@ sendCtrlMsg mNid signal = do

435442 Just nid ->

436443 lift $ sendBinary (NodeIdentifier nid) msg

437444

438−lookupStatic :: Typeable a => String -> Process a

439−lookupStatic label = do

440− rtable <- remoteTable <$> lift getLocalNode

441− return $ fromDyn (rtable Map.! label) (throw typeError)

442− where

443− typeError = userError "lookupStatic type error"

444−

445445lift :: MessageT IO a -> Process a

446446lift = Process . Trans.lift

distributed-process/src/Control/Distributed/Process/Closure.hs+19−25

@@ -44,9 +44,8 @@ module Control.Distributed.Process.Closure (remotable, mkClosure) where

4444import Prelude hiding (lookup)

4545import Data.ByteString.Lazy (ByteString)

4646import Data.Binary (encode, decode)

47−import qualified Data.Map as Map (insert)

47+import Data.Typeable (typeOf)

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

49−import Control.Monad (join)

5049import Language.Haskell.TH

5150 ( -- Q monad and operations

5251 Q

@@ -80,8 +79,10 @@ import Control.Distributed.Process.Internal.Types

8079 , Process

8180 , ProcessId

8281 , CallReply(..)

82+ , registerLabel

83+ , registerSender

8384 )

84−import Control.Distributed.Process (send, unClosure)

85+import Control.Distributed.Process (send)

8586import Control.Distributed.Process.Internal.Dynamic (toDyn)

8687

8788--------------------------------------------------------------------------------

@@ -115,11 +116,10 @@ createMetaData is =

115116--

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

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

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

119+-- 2. registerLabel "M.f" (toDyn ((f . enc) :: ByteString -> b))

119120--

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

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

122−-- (see 'generateCallClosure')

122+-- 3. registerSender (Process c) (send :: ProcessId -> c -> Process ())

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

124124generateDefs n = do

125125 mType <- getType n

@@ -127,18 +127,17 @@ generateDefs n = do

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

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

129129 let decoder = generateDecoder origName (return res)

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

131−

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

131+

132132 -- Generate special closure to support 'call'

133133 mResult <- processResult res

134134 insert' <- case mResult of

135135 Nothing ->

136136 return []

137137 Just result -> do

138− let encoder = generateCallClosure origName (return result)

139− label' = label ++ "__call"

138+ let sender = generateSender result

140139 return . return $

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

140+ [| registerSender $(typeToTypeRep res) (toDyn $sender) |]

142141

143142 return (closure, insert : insert')

144143 _ ->

@@ -157,38 +156,33 @@ generateClosure n arg res = do

157156 label = show $ n

158157

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

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

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

161160processResult t = do

162161 process <- [t| Process |]

163162 case t of

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

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

165164 _ -> return Nothing

166165

167166-- | Generate the decoder (see 'generateDefs')

168167generateDecoder :: Name -> Q Type -> Q Exp

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

170169

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− |]

170+-- | Generate the sender. This is necessary to support 'call'

171+generateSender :: Q Type -> Q Exp

172+generateSender tp =

173+ [| (\pid -> send pid . CallReply) :: ProcessId -> $tp -> Process () |]

179174

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

181176closureName :: Name -> Name

182177closureName n = mkName $ nameBase n ++ "__closure"

183178

184−-- | The name for the decoder

185−decoderName :: Name -> Name

186−decoderName n = mkName $ nameBase n ++ "__dec"

187−

188179--------------------------------------------------------------------------------

189180-- Generic Template Haskell auxiliary functions --

190181--------------------------------------------------------------------------------

191182

183+typeToTypeRep :: Type -> Q Exp

184+typeToTypeRep tp = [| typeOf (undefined :: $(return tp)) |]

185+

192186-- | Compose a set of expressions

193187compose :: [Q Exp] -> Q Exp

194188compose [] = [| id |]

distributed-process/src/Control/Distributed/Process/Internal/Dynamic.hs+26−19

@@ -10,8 +10,11 @@ module Control.Distributed.Process.Internal.Dynamic

1010 , toDyn

1111 , fromDyn

1212 , fromDynamic

13− , dynBindIO

14− , dynBindIO'

13+ , dynTypeRep

14+ , dynBind

15+ , dynBind'

16+ , dynApply

17+ , dynApp

1518 ) where

1619

1720import Data.Typeable

@@ -19,7 +22,6 @@ import Data.Typeable

1922 , TypeRep

2023 , typeOf

2124 , TyCon

22− , typeRepTyCon

2325 , splitTyConApp

2426 , funResultTy

2527 )

@@ -49,24 +51,29 @@ dynTypeRep (Dynamic t _) = t

4951instance Show Dynamic where

5052 show = show . dynTypeRep

5153

52−tyConIO :: TyCon

53−tyConIO = typeRepTyCon (typeOf (undefined :: IO ()))

54−

55−bindIO :: IO a -> (a -> IO b) -> IO b

56−bindIO = (>>=)

57−

58−dynBindIO :: Dynamic -> Dynamic -> Maybe Dynamic

59−-- IO a -> (a -> IO b) -> IO b

60−dynBindIO (Dynamic t1 x) (Dynamic t2 f) =

54+dynBind :: TyCon -> (forall a b. m a -> (a -> m b) -> m b) -> Dynamic -> Dynamic -> Maybe Dynamic

55+dynBind m bind (Dynamic t1 x) (Dynamic t2 f) =

6156 case splitTyConApp t1 of

62− (io, [a]) | io == tyConIO ->

57+ (m', [a]) | m' == m ->

6358 case funResultTy t2 a of

64− Just ioB -> Just (Dynamic ioB (GHC.unsafeCoerce# bindIO x f))

59+ Just mb -> Just (Dynamic mb (GHC.unsafeCoerce# bind x f))

6560 _ -> Nothing

6661 _ -> Nothing

67−

68−dynBindIO' :: Dynamic -> Dynamic -> Dynamic

69−dynBindIO' x f = fromMaybe (error typeError) (dynBindIO x f)

62+

63+dynBind' :: TyCon -> (forall a b. m a -> (a -> m b) -> m b) -> Dynamic -> Dynamic -> Dynamic

64+dynBind' m bind x f = fromMaybe (error typeError) (dynBind m bind x f)

65+ where

66+ typeError = "Type error in dynamic bind.\nCan't bind " ++ show x ++ " to " ++ show f

67+

68+dynApply :: Dynamic -> Dynamic -> Maybe Dynamic

69+dynApply (Dynamic t1 f) (Dynamic t2 x) =

70+ case funResultTy t1 t2 of

71+ Just t3 -> Just (Dynamic t3 (GHC.unsafeCoerce# f x))

72+ Nothing -> Nothing

73+

74+dynApp :: Dynamic -> Dynamic -> Dynamic

75+dynApp f x = fromMaybe (error typeError) (dynApply f x)

7076 where

71− typeError = "Type error in dynamic bind.\n"

72− ++ "Can't bind " ++ show x ++ " to " ++ show f

77+ typeError = "Type error in dynamic application.\n"

78+ ++ "Can't apply function " ++ show f

79+ ++ " to argument " ++ show x

distributed-process/src/Control/Distributed/Process/Internal/Types.hs+82−27

@@ -1,10 +1,7 @@

11-- | We collect all types used internally in a single module to avoid using

22-- mutually recursive modules

33module Control.Distributed.Process.Internal.Types

4− ( -- * Global CH state

5− RemoteTable

6− , initRemoteTable

7− , -- * Node and process identifiers

4+ ( -- * Node and process identifiers

85 NodeId(..)

96 , LocalProcessId(..)

107 , ProcessId(..)

@@ -26,6 +23,11 @@ module Control.Distributed.Process.Internal.Types

2623 , Static(..)

2724 , Closure(..)

2825 , CallReply(..)

26+ , RemoteTable

27+ , initRemoteTable

28+ , registerLabel

29+ , registerSender

30+ , resolveClosure

2931 -- * Messages

3032 , Message(..)

3133 -- * Node controller user-visible data types

@@ -63,12 +65,17 @@ module Control.Distributed.Process.Internal.Types

6365import Data.Map (Map)

6466import qualified Data.Map as Map (empty)

6567import Data.Int (Int32)

66−import Data.Typeable (Typeable)

67−import Data.Binary (Binary, encode, put, get, putWord8, getWord8)

68+import Data.Typeable (Typeable, TypeRep, TyCon, typeRepTyCon, typeOf)

69+import Data.Binary (Binary, encode, decode, put, get, putWord8, getWord8)

6870import Data.ByteString.Lazy (ByteString)

69−import Data.Accessor (Accessor, accessor)

71+import Data.Accessor (Accessor, accessor, (^=), (^.))

7072import qualified Data.Accessor.Container as DAC (mapMaybe)

71−import qualified Data.ByteString.Lazy as BSL (ByteString, toChunks, fromChunks, splitAt)

73+import qualified Data.ByteString.Lazy as BSL

74+ ( ByteString

75+ , toChunks

76+ , fromChunks

77+ , splitAt

78+ )

7279import qualified Data.ByteString as BSS (ByteString, concat)

7380import Control.Category ((>>>))

7481import Control.Exception (Exception)

@@ -81,26 +88,23 @@ import Control.Applicative (Applicative, (<$>), (<*>))

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

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

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

84−import Control.Distributed.Process.Serializable ( Fingerprint

85− , Serializable

86− , encodeFingerprint

87− , decodeFingerprint

88− , fingerprint

89− , sizeOfFingerprint

90− )

91+import Control.Distributed.Process.Serializable

92+ ( Fingerprint

93+ , Serializable

94+ , encodeFingerprint

95+ , decodeFingerprint

96+ , fingerprint

97+ , sizeOfFingerprint

98+ )

9199import Control.Distributed.Process.Internal.CQueue (CQueue)

92−import Control.Distributed.Process.Internal.Dynamic (Dynamic)

93−

94−--------------------------------------------------------------------------------

95−-- Global state --

96−--------------------------------------------------------------------------------

97−

98−-- | Used to fake 'static' (see paper)

99−type RemoteTable = Map String Dynamic

100−

101−-- | Initial (empty) remote-call meta data

102−initRemoteTable :: RemoteTable

103−initRemoteTable = Map.empty

100+import Control.Distributed.Process.Internal.Dynamic

101+ ( Dynamic

102+ , toDyn

103+ , fromDynamic

104+ , dynTypeRep

105+ , dynApp

106+ , dynBind

107+ )

104108

105109--------------------------------------------------------------------------------

106110-- Node and process identifiers --

@@ -231,6 +235,45 @@ data Closure a = Closure (Static (ByteString -> a)) ByteString

231235newtype CallReply a = CallReply a

232236 deriving (Typeable, Binary)

233237

238+-- | Used to fake 'static' (see paper)

239+data RemoteTable = RemoteTable {

240+ _remoteTableLabels :: Map String Dynamic

241+ , _remoteTableSenders :: Map TypeRep Dynamic

242+ }

243+

244+-- | Initial (empty) remote-call meta data

245+initRemoteTable :: RemoteTable

246+initRemoteTable = RemoteTable Map.empty Map.empty

247+

248+registerLabel :: String -> Dynamic -> RemoteTable -> RemoteTable

249+registerLabel label dyn = remoteTableLabel label ^= Just dyn

250+

251+registerSender :: TypeRep -> Dynamic -> RemoteTable -> RemoteTable

252+registerSender typ dyn = remoteTableSender typ ^= Just dyn

253+

254+resolveClosure :: RemoteTable -> String -> Maybe Dynamic

255+resolveClosure rtable "$call" = Just (toDyn aux)

256+ where

257+ -- TODO: error handling

258+ aux :: ByteString -> Process ()

259+ aux env = do

260+ let (label, env', pid) = decode env :: (String, ByteString, ProcessId)

261+ Just proc = resolveClosure rtable label

262+ proc' = dynApp proc (toDyn env')

263+ Just enc = rtable ^. remoteTableSender (dynTypeRep proc')

264+ enc' = dynApp enc (toDyn pid)

265+ Just bound = dynBind tyConProcess bindProcess proc' enc'

266+ Just proc'' = fromDynamic bound

267+ proc''

268+

269+resolveClosure rtable label = rtable ^. remoteTableLabel label

270+

271+tyConProcess :: TyCon

272+tyConProcess = typeRepTyCon (typeOf (undefined :: Process ()))

273+

274+bindProcess :: Process a -> (a -> Process b) -> Process b

275+bindProcess = (>>=)

276+

234277--------------------------------------------------------------------------------

235278-- Messages --

236279--------------------------------------------------------------------------------

@@ -449,6 +492,18 @@ typedChannels = accessor _typedChannels (\cs st -> st { _typedChannels = cs })

449492typedChannelWithId :: LocalChannelId -> Accessor LocalProcessState (Maybe TypedChannel)

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

451494

495+remoteTableLabels :: Accessor RemoteTable (Map String Dynamic)

496+remoteTableLabels = accessor _remoteTableLabels (\ls tbl -> tbl { _remoteTableLabels = ls })

497+

498+remoteTableSenders :: Accessor RemoteTable (Map TypeRep Dynamic)

499+remoteTableSenders = accessor _remoteTableSenders (\es tbl -> tbl { _remoteTableSenders = es })

500+

501+remoteTableLabel :: String -> Accessor RemoteTable (Maybe Dynamic)

502+remoteTableLabel label = remoteTableLabels >>> DAC.mapMaybe label

503+

504+remoteTableSender :: TypeRep -> Accessor RemoteTable (Maybe Dynamic)

505+remoteTableSender typ = remoteTableSenders >>> DAC.mapMaybe typ

506+

452507--------------------------------------------------------------------------------

453508-- MessageT monad --

454509--------------------------------------------------------------------------------

distributed-process/src/Control/Distributed/Process/Node.hs+11−6

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

9191 , Identifier(..)

9292 , ChannelId(..)

9393 , typedChannelWithId

94+ , resolveClosure

9495 )

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

9697import Control.Distributed.Process.Internal.MessageT

@@ -429,11 +430,15 @@ ncEffectProcessDied pid reason = do

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

430431ncEffectSpawn pid cProc ref = do

431432 mProc <- unClosure cProc

432− -- TODO: what should we do when unClosure returns Nothing?

433− forM_ mProc $ \proc -> do

434− node <- lift getLocalNode

435− pid' <- liftIO $ forkProcess node proc

436− lift $ sendMessage (ProcessIdentifier pid) (DidSpawn ref pid')

433+ case mProc of

434+ Just proc -> do

435+ node <- lift getLocalNode

436+ pid' <- liftIO $ forkProcess node proc

437+ lift $ sendMessage (ProcessIdentifier pid) (DidSpawn ref pid')

438+ Nothing ->

439+ -- TODO: what should we do when unClosure returns Nothing?

440+ liftIO . putStrLn $ "Error: closure " ++ show cProc ++ " not found"

441+

437442

438443--------------------------------------------------------------------------------

439444-- Auxiliary --

@@ -488,7 +493,7 @@ isLocal nid pid = processNodeId pid == localNodeId nid

488493unClosure :: Typeable a => Closure a -> NC (Maybe a)

489494unClosure (Closure (Static label) env) = do

490495 rtable <- remoteTable <$> lift getLocalNode

491− case Map.lookup label rtable of

496+ case resolveClosure rtable label of

492497 Nothing -> return Nothing

493498 Just dyn -> case fromDynamic dyn of

494499 Nothing -> return Nothing