0328ade · distributed-process

git.iko.soy/distributed-process

on master help sign in

Add TH support for polymorphic closures

0328adeEdsko de Vries14 years ago

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

compare with master

unifiedsplit

@@ -91,12 +91,7 @@ module Control.Distributed.Process

91 ) where

93import Prelude hiding (catch)

94−import Data.Typeable (Typeable, typeOf)

95−import Control.Applicative ((<$>))

96−import Control.Exception (throw)

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

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

98−import Control.Distributed.Process.Internal.MessageT (getLocalNode)

99−import Control.Distributed.Process.Internal.Dynamic (fromDyn, dynTypeRep)

100import Control.Distributed.Process.Internal.Types

95import Control.Distributed.Process.Internal.Types

101 ( RemoteTable

96 ( RemoteTable

102 , NodeId(..)

97 , NodeId(..)

@@ -118,8 +113,6 @@ import Control.Distributed.Process.Internal.Types

118 , SendPort(..)

113 , SendPort(..)

119 , ReceivePort(..)

114 , ReceivePort(..)

120 , SerializableDict(..)

115 , SerializableDict(..)

121− , procMsg

122− , LocalNode(..)

123 , SendPortId(..)

116 , SendPortId(..)

124 , WhereIsReply(..)

117 , WhereIsReply(..)

125 )

118 )

@@ -130,7 +123,10 @@ import Control.Distributed.Process.Internal.Closure.BuiltIn

130 , expectClosure

123 , expectClosure

131 , serializableDictUnit

124 , serializableDictUnit

132 )

125 )

133−import Control.Distributed.Process.Internal.Closure.Combinators (cpSeq, cpBind)

126+import Control.Distributed.Process.Internal.Closure.Combinators

127+ ( cpSeq

128+ , cpBind

129+ )

134import Control.Distributed.Process.Internal.Primitives

130import Control.Distributed.Process.Internal.Primitives

135 ( -- Basic messaging

131 ( -- Basic messaging

136 send

132 send

@@ -176,12 +172,13 @@ import Control.Distributed.Process.Internal.Primitives

176 , whereisRemote

172 , whereisRemote

177 , whereisRemoteAsync

173 , whereisRemoteAsync

178 , nsendRemote

174 , nsendRemote

175+ -- Closures

176+ , unClosure

179 -- Auxiliary API

177 -- Auxiliary API

180 , catch

178 , catch

181 , expectTimeout

179 , expectTimeout

182 , spawnAsync

180 , spawnAsync

183 )

181 )

184−import Control.Distributed.Process.Internal.Closure (resolveClosure)

185

182

186-- INTERNAL NOTES

183-- INTERNAL NOTES

187--

184--

@@ -320,14 +317,16 @@ spawnSupervised nid proc = do

320 ref <- monitor them

317 ref <- monitor them

321 return (them, ref)

318 return (them, ref)

322

319

323−-- | Deserialize a closure

320+{-

324−unClosure :: forall a. Typeable a => Closure a -> Process a

321+spawnChannel :: forall a. Typeable a

325−unClosure (Closure (Static label) env) = do

322+ => NodeId

326− rtable <- remoteTable <$> procMsg getLocalNode

323+ -> Closure (ReceivePort a -> Process ())

327− case resolveClosure rtable label env of

324+ -> Process (SendPort a)

328− Nothing -> throw . userError $ "Unregistered closure " ++ show label

325+spawnChannel nid proc = do

329− Just dyn -> return $ fromDyn dyn (throw (typeError dyn))

326+ us <- getSelfPid

327+ undefined

330 where

328 where

331− typeError dyn = userError $ "lookupStatic type error: "

329+ rproc :: ProcessId -> Closure (Process ((), ReceivePort a))

332− ++ "cannot match " ++ show (dynTypeRep dyn)

330+ rproc pid = newChanClosure

333− ++ " against " ++ show (typeOf (undefined :: a))

331+ `cpBind` cpFirst (sendClosure undefined pid)

332+-}

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

@@ -1,18 +1,16 @@

1{-# LANGUAGE MagicHash #-}

2module Control.Distributed.Process.Internal.Closure

3 ( -- * Runtime support

4− initRemoteTable

5− , resolveClosure

4+ resolveClosure

6 ) where

5 ) where

8−import qualified Data.Map as Map (empty)

9import Data.Accessor ((^.))

7import Data.Accessor ((^.))

10import Data.ByteString.Lazy (ByteString)

8import Data.ByteString.Lazy (ByteString)

11import Data.Binary (decode)

9import Data.Binary (decode)

12import Data.Typeable (TypeRep)

10import Data.Typeable (TypeRep)

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

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

14import Control.Distributed.Process.Internal.Types

12import Control.Distributed.Process.Internal.Types

15− ( RemoteTable(RemoteTable)

13+ ( RemoteTable

16 , remoteTableLabel

14 , remoteTableLabel

17 , remoteTableDict

15 , remoteTableDict

18 , StaticLabel(..)

16 , StaticLabel(..)

@@ -30,17 +28,24 @@ import Control.Distributed.Process.Internal.Dynamic

30 , unsafeCoerce#

28 , unsafeCoerce#

31 )

29 )

32import Control.Distributed.Process.Internal.TypeRep () -- Binary instances

30import Control.Distributed.Process.Internal.TypeRep () -- Binary instances

31+

32+{-

33import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn

34 (remoteTable)

35+import qualified Control.Distributed.Process.Internal.Closure.Combinators as Combinators

36+ (remoteTable)

37+

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

39+initRemoteTable :: RemoteTable

40+initRemoteTable = BuiltIn.remoteTable

41+ . Combinators.remoteTable

42+ $ RemoteTable Map.empty Map.empty

43+-}

36--------------------------------------------------------------------------------

45--------------------------------------------------------------------------------

37-- Runtime support for closures --

46-- Runtime support for closures --

38--------------------------------------------------------------------------------

47--------------------------------------------------------------------------------

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

41−initRemoteTable :: RemoteTable

42−initRemoteTable = BuiltIn.remoteTable (RemoteTable Map.empty Map.empty)

43−

44resolveClosure :: RemoteTable -> StaticLabel -> ByteString -> Maybe Dynamic

49resolveClosure :: RemoteTable -> StaticLabel -> ByteString -> Maybe Dynamic

45-- Built-in closures

50-- Built-in closures

46resolveClosure rtable ClosureSend env = do

51resolveClosure rtable ClosureSend env = do

@@ -64,54 +69,6 @@ resolveClosure rtable ClosureApply env = do

64 f `dynApply` x

69 f `dynApply` x

65 where

70 where

66 (labelf, envf, labelx, envx) = decode env

71 (labelf, envf, labelx, envx) = decode env

67−resolveClosure _rtable ClosureConst env =

68− return $ Dynamic (decode env) (unsafeCoerce# const)

69−resolveClosure _rtable ClosureUnit _env =

70− return $ toDyn ()

71−-- Arrow combinators

72−resolveClosure _rtable CpId env =

73− return $ Dynamic (decode env) (unsafeCoerce# cpId)

74− where

75− cpId :: forall a. a -> Process a

76− cpId = return

77−resolveClosure _rtable CpComp env =

78− return $ Dynamic (decode env) (unsafeCoerce# cpComp)

79− where

80− cpComp :: forall a b c. (a -> Process b) -> (b -> Process c) -> a -> Process c

81− cpComp p q a = p a >>= q

82−resolveClosure _rtable CpFirst env =

83− return $ Dynamic (decode env) (unsafeCoerce# cpFirst)

84− where

85− cpFirst :: forall a b c. (a -> Process b) -> (a, c) -> Process (b, c)

86− cpFirst p (a, c) = do b <- p a ; return (b, c)

87−resolveClosure _rtable CpSwap env =

88− return $ Dynamic (decode env) (unsafeCoerce# cpSwap)

89− where

90− cpSwap :: forall a b. (a, b) -> Process (b, a)

91− cpSwap (a, b) = return (b, a)

92−resolveClosure _rtable CpCopy env =

93− return $ Dynamic (decode env) (unsafeCoerce# cpCopy)

94− where

95− cpCopy :: forall a. a -> Process (a, a)

96− cpCopy a = return (a, a)

97−resolveClosure _rtable CpLeft env =

98− return $ Dynamic (decode env) (unsafeCoerce# cpLeft)

99− where

100− cpLeft :: forall a b c. (a -> Process b) -> Either a c -> Process (Either b c)

101− cpLeft p (Left a) = do b <- p a ; return (Left b)

102− cpLeft _ (Right c) = return (Right c)

103−resolveClosure _rtable CpMirror env =

104− return $ Dynamic (decode env) (unsafeCoerce# cpMirror)

105− where

106− cpMirror :: forall a b. Either a b -> Process (Either b a)

107− cpMirror (Left a) = return (Right a)

108− cpMirror (Right b) = return (Left b)

109−resolveClosure _rtable CpUntag env =

110− return $ Dynamic (decode env) (unsafeCoerce# cpUntag)

111− where

112− cpUntag :: forall a. Either a a -> Process a

113− cpUntag (Left a) = return a

114− cpUntag (Right a) = return a

115resolveClosure rtable CpApply env =

72resolveClosure rtable CpApply env =

116 return $ Dynamic typApply (unsafeCoerce# cpApply)

73 return $ Dynamic typApply (unsafeCoerce# cpApply)

117 where

74 where

@@ -131,3 +88,8 @@ resolveClosure rtable CpApply env =

131resolveClosure rtable (UserStatic label) env = do

88resolveClosure rtable (UserStatic label) env = do

132 val <- rtable ^. remoteTableLabel label

89 val <- rtable ^. remoteTableLabel label

133 dynApply val (toDyn env)

90 dynApply val (toDyn env)

91+

92+-- User-defined polymorphic closures

93+resolveClosure rtable (PolyStatic label) env = do

94+ Dynamic _ val <- rtable ^. remoteTableLabel label

95+ return (Dynamic (decode env) val)

distributed-process/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs+1−1

@@ -13,7 +13,7 @@ module Control.Distributed.Process.Internal.Closure.BuiltIn

13 ) where

15import Data.Binary (encode)

16−import Data.Typeable (typeOf)

16+import Data.Typeable (Typeable, typeOf)

17import Control.Distributed.Process.Internal.Primitives (link, unlink)

18import Control.Distributed.Process.Internal.Types

19 ( ProcessId

distributed-process/src/Control/Distributed/Process/Internal/Closure/Combinators.hs+74−51

@@ -1,6 +1,7 @@

1module Control.Distributed.Process.Internal.Closure.Combinators

2+ ( remoteTable

2− ( -- * Generic combinators

3+ -- * Generic combinators

3− closureApply

4+ , closureApply

4 , closureConst

5 , closureConst

5 , closureUnit

6 , closureUnit

6 -- * Arrow combinators for processes

7 -- * Arrow combinators for processes

@@ -28,16 +29,75 @@ module Control.Distributed.Process.Internal.Closure.Combinators

28 ) where

29 ) where

30import Prelude hiding (lookup)

31import Prelude hiding (lookup)

31−import qualified Data.ByteString.Lazy as BS (empty)

32import Data.Binary (encode)

33import Data.Typeable (typeOf, Typeable)

34+import Data.Tuple (swap)

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

36+import Control.Monad ((>=>))

34import Control.Distributed.Process.Internal.Types

37import Control.Distributed.Process.Internal.Types

35 ( Closure(..)

38 ( Closure(..)

36 , Static(..)

39 , Static(..)

37 , StaticLabel(..)

40 , StaticLabel(..)

38 , Process

41 , Process

42+ , RemoteTable

39 )

43 )

40import Control.Distributed.Process.Internal.TypeRep () -- Binary instances

44import Control.Distributed.Process.Internal.TypeRep () -- Binary instances

45+import Control.Distributed.Process.Internal.Closure.TH (remotable, mkClosure)

46+import Control.Distributed.Process.Internal.Primitives (unClosure)

47+

48+--------------------------------------------------------------------------------

49+-- TH stuff --

50+--------------------------------------------------------------------------------

51+

52+idUnit :: () -> ()

53+idUnit = id

54+

55+returnProcess :: a -> Process a

56+returnProcess = return

57+

58+kleisliComposeProcess :: (a -> Process b) -> (b -> Process c) -> a -> Process c

59+kleisliComposeProcess = (>=>)

60+

61+firstProcess :: (a -> Process b) -> (a, c) -> Process (b, c)

62+firstProcess f (a, c) = f a >>= \b -> return (b, c)

63+

64+swapProcess :: (a, b) -> Process (b, a)

65+swapProcess = return . swap

66+

67+copyProcess :: a -> Process (a, a)

68+copyProcess x = return (x, x)

69+

70+leftProcess :: (a -> Process b) -> Either a c -> Process (Either b c)

71+leftProcess f (Left a) = Left <$> f a

72+leftProcess _ (Right b) = Right <$> return b

73+

74+mirrorProcess :: Either a b -> Process (Either b a)

75+mirrorProcess (Left x) = Right <$> return x

76+mirrorProcess (Right y) = Left <$> return y

77+

78+untagProcess :: Either a a -> Process a

79+untagProcess (Left x) = return x

80+untagProcess (Right y) = return y

81+

82+applyClosureProcess :: (Typeable a, Typeable b) => (Closure (a -> Process b), a) -> Process b

83+applyClosureProcess (procClosure, a) = do

84+ proc <- unClosure procClosure

85+ proc a

86+

87+remotable [ 'const

88+ , 'idUnit

89+ , 'returnProcess

90+ , 'kleisliComposeProcess

91+ , 'firstProcess

92+ , 'swapProcess

93+ , 'copyProcess

94+ , 'leftProcess

95+ , 'mirrorProcess

96+ , 'untagProcess

97+ ]

98+

99+remoteTable :: RemoteTable -> RemoteTable

100+remoteTable = __remoteTable

101

42--------------------------------------------------------------------------------

102--------------------------------------------------------------------------------

43-- Generic closure combinators --

103-- Generic closure combinators --

@@ -47,15 +107,11 @@ closureApply :: Closure (a -> b) -> Closure a -> Closure b

47closureApply (Closure (Static labelf) envf) (Closure (Static labelx) envx) =

107closureApply (Closure (Static labelf) envf) (Closure (Static labelx) envx) =

48 Closure (Static ClosureApply) $ encode (labelf, envf, labelx, envx)

108 Closure (Static ClosureApply) $ encode (labelf, envf, labelx, envx)

109

50−closureConst :: forall a b. (Typeable a, Typeable b)

110+closureConst :: (Typeable a, Typeable b) => Closure (a -> b -> a)

51− => Closure (a -> b -> a)

111+closureConst = $(mkClosure 'const)

52−closureConst = Closure (Static ClosureConst) (encode $ typeOf aux)

53− where

54− aux :: a -> b -> a

55− aux = undefined

112

57closureUnit :: Closure ()

113closureUnit :: Closure ()

58−closureUnit = Closure (Static ClosureUnit) BS.empty

114+closureUnit = $(mkClosure 'idUnit) ()

115

60--------------------------------------------------------------------------------

116--------------------------------------------------------------------------------

61-- Arrow combinators for processes --

117-- Arrow combinators for processes --

@@ -73,37 +129,19 @@ cpElim = flip closureApply closureUnit

129

74cpId :: forall a. Typeable a

130cpId :: forall a. Typeable a

75 => CP a a

131 => CP a a

76−cpId = Closure (Static CpId) (encode $ typeOf aux)

132+cpId = $(mkClosure 'returnProcess)

77− where

78− aux :: a -> Process a

79− aux = undefined

133

81cpComp :: forall a b c. (Typeable a, Typeable b, Typeable c)

134cpComp :: forall a b c. (Typeable a, Typeable b, Typeable c)

82 => CP a b -> CP b c -> CP a c

135 => CP a b -> CP b c -> CP a c

83−cpComp f g = comp `closureApply` f `closureApply` g

136+cpComp f g = $(mkClosure 'kleisliComposeProcess) `closureApply` f `closureApply` g

84− where

85− comp :: Closure ((a -> Process b) -> (b -> Process c) -> a -> Process c)

86− comp = Closure (Static CpComp) (encode $ typeOf aux)

87−

88− aux :: (a -> Process b) -> (b -> Process c) -> a -> Process c

89− aux = undefined

137

91cpFirst :: forall a b c. (Typeable a, Typeable b, Typeable c)

138cpFirst :: forall a b c. (Typeable a, Typeable b, Typeable c)

92 => CP a b -> CP (a, c) (b, c)

139 => CP a b -> CP (a, c) (b, c)

93−cpFirst = closureApply first

140+cpFirst = closureApply $(mkClosure 'firstProcess)

94− where

95− first :: Closure ((a -> Process b) -> (a, c) -> Process (b, c))

96− first = Closure (Static CpFirst) (encode $ typeOf aux)

97−

98− aux :: (a -> Process b) -> (a, c) -> Process (b, c)

99− aux = undefined

100

141

101cpSwap :: forall a b. (Typeable a, Typeable b)

142cpSwap :: forall a b. (Typeable a, Typeable b)

102 => CP (a, b) (b, a)

143 => CP (a, b) (b, a)

103−cpSwap = Closure (Static CpSwap) (encode $ typeOf aux)

144+cpSwap = $(mkClosure 'swapProcess)

104− where

105− aux :: (a, b) -> Process (b, a)

106− aux = undefined

107

145

108cpSecond :: (Typeable a, Typeable b, Typeable c)

146cpSecond :: (Typeable a, Typeable b, Typeable c)

109 => CP a b -> CP (c, a) (c, b)

147 => CP a b -> CP (c, a) (c, b)

@@ -115,10 +153,7 @@ cpPair f g = cpFirst f `cpComp` cpSecond g

115

153

116cpCopy :: forall a. Typeable a

154cpCopy :: forall a. Typeable a

117 => CP a (a, a)

155 => CP a (a, a)

118−cpCopy = Closure (Static CpCopy) (encode $ typeOf aux)

156+cpCopy = $(mkClosure 'copyProcess)

119− where

120− aux :: a -> Process (a, a)

121− aux = undefined

122

157

123cpFanOut :: (Typeable a, Typeable b, Typeable c)

158cpFanOut :: (Typeable a, Typeable b, Typeable c)

124 => CP a b -> CP a c -> CP a (b, c)

159 => CP a b -> CP a c -> CP a (b, c)

@@ -126,20 +161,11 @@ cpFanOut f g = cpCopy `cpComp` (f `cpPair` g)

126

161

127cpLeft :: forall a b c. (Typeable a, Typeable b, Typeable c)

162cpLeft :: forall a b c. (Typeable a, Typeable b, Typeable c)

128 => CP a b -> CP (Either a c) (Either b c)

163 => CP a b -> CP (Either a c) (Either b c)

129−cpLeft = closureApply left

164+cpLeft = closureApply $(mkClosure 'leftProcess)

130− where

131− left :: Closure ((a -> Process b) -> Either a c -> Process (Either b c))

132− left = Closure (Static CpLeft) (encode $ typeOf aux)

133−

134− aux :: (a -> Process b) -> Either a c -> Process (Either b c)

135− aux = undefined

136

165

137cpMirror :: forall a b. (Typeable a, Typeable b)

166cpMirror :: forall a b. (Typeable a, Typeable b)

138 => CP (Either a b) (Either b a)

167 => CP (Either a b) (Either b a)

139−cpMirror = Closure (Static CpMirror) (encode $ typeOf aux)

168+cpMirror = $(mkClosure 'mirrorProcess)

140− where

141− aux :: Either a b -> Process (Either b a)

142− aux = undefined

143

169

144cpRight :: forall a b c. (Typeable a, Typeable b, Typeable c)

170cpRight :: forall a b c. (Typeable a, Typeable b, Typeable c)

145 => CP a b -> CP (Either c a) (Either c b)

171 => CP a b -> CP (Either c a) (Either c b)

@@ -151,10 +177,7 @@ cpEither f g = cpLeft f `cpComp` cpRight g

151

177

152cpUntag :: forall a. Typeable a

178cpUntag :: forall a. Typeable a

153 => CP (Either a a) a

179 => CP (Either a a) a

154−cpUntag = Closure (Static CpUntag) (encode $ typeOf aux)

180+cpUntag = $(mkClosure 'untagProcess)

155− where

156− aux :: Either a a -> Process a

157− aux = undefined

158

181

159cpFanIn :: (Typeable a, Typeable b, Typeable c)

182cpFanIn :: (Typeable a, Typeable b, Typeable c)

160 => CP a c -> CP b c -> CP (Either a b) c

183 => CP a c -> CP b c -> CP (Either a b) c

distributed-process/src/Control/Distributed/Process/Internal/Closure/TH.hs+30−3

@@ -1,6 +1,7 @@

1-- | Template Haskell support

2--

3-- (In a separate file for convenience)

4+{-# LANGUAGE MagicHash #-}

4module Control.Distributed.Process.Internal.Closure.TH

5module Control.Distributed.Process.Internal.Closure.TH

5 ( -- * User-level API

6 ( -- * User-level API

6 remotable

7 remotable

@@ -24,8 +25,10 @@ import Language.Haskell.TH

24 -- Algebraic data types

25 -- Algebraic data types

25 , Dec

26 , Dec

26 , Exp

27 , Exp

27− , Type(AppT, ArrowT)

28+ , Type(AppT, ArrowT, ForallT, VarT)

28 , Info(VarI)

29 , Info(VarI)

30+ , TyVarBndr(PlainTV, KindedTV)

31+ , Pred(ClassP)

29 -- Lifted constructors

32 -- Lifted constructors

30 -- .. Literals

33 -- .. Literals

31 , stringL

34 , stringL

@@ -51,7 +54,11 @@ import Control.Distributed.Process.Internal.Types

51 , SerializableDict(..)

54 , SerializableDict(..)

52 , RuntimeSerializableSupport(..)

55 , RuntimeSerializableSupport(..)

53 )

56 )

54−import Control.Distributed.Process.Internal.Dynamic (Dynamic, toDyn)

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

58+ ( Dynamic(..)

59+ , toDyn

60+ , unsafeCoerce#

61+ )

55import Control.Distributed.Process.Internal.Primitives (send, expect)

62import Control.Distributed.Process.Internal.Primitives (send, expect)

57--------------------------------------------------------------------------------

64--------------------------------------------------------------------------------

@@ -104,6 +111,9 @@ generateDefs n = do

104 return (closure, insert)

111 return (closure, insert)

105 Just (origName, sdict `AppT` a) | sdict == serializableDict ->

112 Just (origName, sdict `AppT` a) | sdict == serializableDict ->

106 return ([], [| registerSerializableDict $(varE n) |])

113 return ([], [| registerSerializableDict $(varE n) |])

114+ Just (origName, ForallT vars [] tp) -> do

115+ (closure, label) <- generatePolyClosure origName vars tp

116+ return (closure, [| registerLabel $(stringE label) (Dynamic (error "Polymorphic closure") (unsafeCoerce# $(varE origName))) |])

107 _ ->

117 _ ->

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

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

109

119

@@ -117,7 +127,24 @@ generateClosure n arg res = do

117 return (closure, label)

127 return (closure, label)

118 where

128 where

119 label :: String

129 label :: String

120− label = show $ n

130+ label = show n

131+

132+-- | Generate a polymorphic closure

133+generatePolyClosure :: Name -> [TyVarBndr] -> Type -> Q ([Dec], String)

134+generatePolyClosure n xs typ = do

135+ closureTyp <- [t| Closure |]

136+ closure <- sequence

137+ [ sigD (closureName n) (return (ForallT xs (map typeable xs) (closureTyp `AppT` typ)) )

138+ , sfnD (closureName n) [| Closure (Static (PolyStatic ($(stringE label)))) (encode (typeOf (undefined :: $(return typ)))) |]

139+ ]

140+ return (closure, label)

141+ where

142+ label :: String

143+ label = show n

144+

145+ typeable :: TyVarBndr -> Pred

146+ typeable (PlainTV v) = ClassP (mkName "Typeable") [VarT v]

147+ typeable (KindedTV v _) = ClassP (mkName "Typeable") [VarT v]

121

148

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

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

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

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

distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+22−1

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

41 , whereisRemote

42 , whereisRemoteAsync

43 , nsendRemote

44+ -- * Closures

45+ , unClosure

44 -- * Auxiliary API

46 -- * Auxiliary API

45 , catch

47 , catch

46 , expectTimeout

48 , expectTimeout

@@ -55,7 +57,7 @@ module Control.Distributed.Process.Internal.Primitives

56import Prelude hiding (catch)

58import Prelude hiding (catch)

57import Data.Binary (decode)

59import Data.Binary (decode)

58−import Data.Typeable (Typeable)

60+import Data.Typeable (Typeable, typeOf)

59import Data.Time.Clock (getCurrentTime)

61import Data.Time.Clock (getCurrentTime)

60import Data.Time.Format (formatTime)

62import Data.Time.Format (formatTime)

61import System.Locale (defaultTimeLocale)

63import System.Locale (defaultTimeLocale)

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

108 , DidUnlinkPort(..)

110 , DidUnlinkPort(..)

109 , WhereIsReply(..)

111 , WhereIsReply(..)

110 , createMessage

112 , createMessage

113+ , Static(..)

111 )

114 )

112import Control.Distributed.Process.Internal.MessageT

115import Control.Distributed.Process.Internal.MessageT

113 ( sendMessage

116 ( sendMessage

@@ -115,6 +118,8 @@ import Control.Distributed.Process.Internal.MessageT

115 , getLocalNode

118 , getLocalNode

116 )

119 )

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

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

121+import Control.Distributed.Process.Internal.Closure (resolveClosure)

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

118

123

119--------------------------------------------------------------------------------

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

120-- Basic messaging --

125-- Basic messaging --

@@ -445,6 +450,22 @@ nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()

445nsendRemote nid label msg =

450nsendRemote nid label msg =

446 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))

451 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))

447

452

453+--------------------------------------------------------------------------------

454+-- Closures --

455+--------------------------------------------------------------------------------

456+

457+-- | Deserialize a closure

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

459+unClosure (Closure (Static label) env) = do

460+ rtable <- remoteTable <$> procMsg getLocalNode

461+ case resolveClosure rtable label env of

462+ Nothing -> throw . userError $ "Unregistered closure " ++ show label

463+ Just dyn -> return $ fromDyn dyn (throw (typeError dyn))

464+ where

465+ typeError dyn = userError $ "lookupStatic type error: "

466+ ++ "cannot match " ++ show (dynTypeRep dyn)

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

468+

448--------------------------------------------------------------------------------

469--------------------------------------------------------------------------------

449-- Auxiliary functions --

470-- Auxiliary functions --

450--------------------------------------------------------------------------------

471--------------------------------------------------------------------------------

distributed-process/src/Control/Distributed/Process/Internal/Types.hs+14−41

@@ -253,6 +253,7 @@ data ReceivePort a =

253 | ReceivePortBiased [ReceivePort a]

254 -- | A round-robin combination of receive ports

255 | ReceivePortRR (TVar [ReceivePort a])

256+ deriving Typeable

256

257

257--------------------------------------------------------------------------------

258--------------------------------------------------------------------------------

258-- Closures --

259-- Closures --

@@ -260,23 +261,13 @@ data ReceivePort a =

260

261

261data StaticLabel =

262data StaticLabel =

262 UserStatic String

263 UserStatic String

264+ | PolyStatic String

263 -- Built-in closures

265 -- Built-in closures

264 | ClosureReturn

266 | ClosureReturn

265 | ClosureSend

267 | ClosureSend

266 | ClosureExpect

268 | ClosureExpect

267 -- Generic closure combinators

269 -- Generic closure combinators

268 | ClosureApply

270 | ClosureApply

269− | ClosureConst

270− | ClosureUnit

271− -- Arrow combinators for processes

272− | CpId

273− | CpComp

274− | CpFirst

275− | CpSwap

276− | CpCopy

277− | CpLeft

278− | CpMirror

279− | CpUntag

280 | CpApply

271 | CpApply

281 deriving (Typeable, Show)

272 deriving (Typeable, Show)

282

273

@@ -555,40 +546,22 @@ instance Binary Identifier where

555

546

556instance Binary StaticLabel where

547instance Binary StaticLabel where

557 put (UserStatic string) = putWord8 0 >> put string

548 put (UserStatic string) = putWord8 0 >> put string

549+ put (PolyStatic string) = putWord8 1 >> put string

558− put ClosureReturn = putWord8 1

550+ put ClosureReturn = putWord8 2

559− put ClosureSend = putWord8 2

560− put ClosureExpect = putWord8 3

551+ put ClosureSend = putWord8 3

561− put ClosureApply = putWord8 4

552+ put ClosureExpect = putWord8 4

562− put ClosureConst = putWord8 5

553+ put ClosureApply = putWord8 6

563− put ClosureUnit = putWord8 6

564− put CpId = putWord8 7

565− put CpComp = putWord8 8

566− put CpFirst = putWord8 9

567− put CpSwap = putWord8 10

568− put CpCopy = putWord8 11

569− put CpLeft = putWord8 12

570− put CpMirror = putWord8 13

571− put CpUntag = putWord8 14

572− put CpApply = putWord8 15

554+ put CpApply = putWord8 17

573 get = do

555 get = do

574 header <- getWord8

556 header <- getWord8

575 case header of

557 case header of

576 0 -> UserStatic <$> get

558 0 -> UserStatic <$> get

577− 1 -> return ClosureReturn

559+ 1 -> PolyStatic <$> get

578− 2 -> return ClosureSend

560+ 2 -> return ClosureReturn

579− 3 -> return ClosureExpect

561+ 3 -> return ClosureSend

562+ 4 -> return ClosureExpect

580− 4 -> return ClosureApply

563+ 6 -> return ClosureApply

581− 5 -> return ClosureConst

582− 6 -> return ClosureUnit

583− 7 -> return CpId

584− 8 -> return CpComp

585− 9 -> return CpFirst

586− 10 -> return CpSwap

587− 11 -> return CpCopy

588− 12 -> return CpLeft

589− 13 -> return CpMirror

590− 14 -> return CpUntag

591− 15 -> return CpApply

564+ 17 -> return CpApply

592 _ -> fail "StaticLabel.get: invalid"

565 _ -> fail "StaticLabel.get: invalid"

593

566

594instance Binary WhereIsReply where

567instance Binary WhereIsReply where

distributed-process/src/Control/Distributed/Process/Node.hs+12−2

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

105 , typedChannelWithId

106 , WhereIsReply(..)

107 , messageToPayload

108+ , RemoteTable(..)

108 )

109 )

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

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

110import Control.Distributed.Process.Internal.MessageT

111import Control.Distributed.Process.Internal.MessageT

@@ -118,16 +119,25 @@ import Control.Distributed.Process.Internal.MessageT

118 )

119 )

119import Control.Distributed.Process.Internal.Dynamic (fromDynamic)

120import Control.Distributed.Process.Internal.Dynamic (fromDynamic)

120import Control.Distributed.Process.Internal.Closure

121import Control.Distributed.Process.Internal.Closure

121− ( initRemoteTable

122+ (

122− , resolveClosure

123+ resolveClosure

123 )

124 )

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

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

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

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

127+import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn

128+ (remoteTable)

129+import qualified Control.Distributed.Process.Internal.Closure.Combinators as Combinators

130+ (remoteTable)

126

131

127--------------------------------------------------------------------------------

132--------------------------------------------------------------------------------

128-- Initialization --

133-- Initialization --

129--------------------------------------------------------------------------------

134--------------------------------------------------------------------------------

130

135

136+initRemoteTable :: RemoteTable

137+initRemoteTable = BuiltIn.remoteTable

138+ . Combinators.remoteTable

139+ $ RemoteTable Map.empty Map.empty

140+

131-- | Initialize a new local node.

141-- | Initialize a new local node.

132--

142--

133-- Note that proper Cloud Haskell initialization and configuration is still

143-- Note that proper Cloud Haskell initialization and configuration is still

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

@@ -1,6 +1,7 @@

1module Main where

3import qualified Data.ByteString.Lazy as BSL (empty)

4+import Data.Typeable (Typeable)

4import Control.Monad (join, replicateM)

5import Control.Monad (join, replicateM)

5import Control.Exception (IOException, throw)

6import Control.Exception (IOException, throw)

6import Control.Concurrent (forkIO, threadDelay)

7import Control.Concurrent (forkIO, threadDelay)

@@ -44,7 +45,21 @@ returnForTestApply = id

44wait :: Int -> Process ()

45wait :: Int -> Process ()

45wait = liftIO . threadDelay

46wait = liftIO . threadDelay

47−$(remotable ['addInt, 'putInt, 'sendInt, 'sendPid, 'factorial, 'factorialOf, 'returnForTestApply, 'wait, 'serializableDictInt])

48+first :: (a, b) -> a

49+first (x, _) = x

50+

51+$(remotable

52+ [ 'addInt

53+ , 'putInt

54+ , 'sendInt

55+ , 'sendPid

56+ , 'factorial

57+ , 'factorialOf

58+ , 'returnForTestApply

59+ , 'wait

60+ , 'serializableDictInt

61+ , 'first

62+ ])

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

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

50testSendPureClosure transport rtable = do

65testSendPureClosure transport rtable = do