3c1dda4 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Define arrow operators for closures

3c1dda4Edsko de Vries14 years ago

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

compare with master

unifiedsplit

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

93 , Process(..)

94 , runLocalProcess

95 , Closure(..)

96+ , StaticLabel(Call)

96 , Static(..)

97 , Static(..)

97 , Message(..)

98 , Message(..)

98 , MonitorRef(..)

99 , MonitorRef(..)

@@ -316,7 +317,7 @@ spawn nid proc = do

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

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

317call nid (Closure (Static label) env) = do

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

318 us <- getSelfPid

319 us <- getSelfPid

319− spawn nid (Closure (Static "$call") (encode (label, env, us)))

320+ spawn nid (Closure (Static Call) (encode (label, env, us)))

320 CallReply a <- expect

321 CallReply a <- expect

321 return a

322 return a

322

323

distributed-process/src/Control/Distributed/Process/Closure.hs+158−30

@@ -43,17 +43,36 @@ module Control.Distributed.Process.Closure

43 ( -- * Primitive closures

44 remotable

45 , mkClosure

46+ -- * Generic closure combinators

47+ , closureApply

48+ , closureConst

49+ , closureUnit

46− -- * Closure combinators

50+ -- * Arrow combinators for processes

47− , applyClosure

51+ , CP

48− , liftedBind

52+ , cpIntro

49− , liftedConst

53+ , cpElim

50− , bindCP

54+ , cpId

51− , weakenCP

55+ , cpComp

52− , seqCP

56+ , cpFirst

57+ , cpSwap

58+ , cpSecond

59+ , cpPair

60+ , cpCopy

61+ , cpFanOut

62+ , cpLeft

63+ , cpMirror

64+ , cpRight

65+ , cpEither

66+ , cpUntag

67+ , cpFanIn

68+ , cpApply

69+ -- * Derived combinators for processes

70+ , cpBind

71+ , cpSeq

53 ) where

72 ) where

55import Prelude hiding (lookup)

74import Prelude hiding (lookup)

56−import Data.ByteString.Lazy (ByteString)

75+import Data.ByteString.Lazy (ByteString, empty)

57import Data.Binary (encode, decode)

76import Data.Binary (encode, decode)

58import Data.Typeable (typeOf, Typeable)

77import Data.Typeable (typeOf, Typeable)

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

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

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

87 ( RemoteTable

106 ( RemoteTable

88 , Closure(..)

107 , Closure(..)

89 , Static(..)

108 , Static(..)

109+ , StaticLabel(..)

90 , Process

110 , Process

91 , ProcessId

111 , ProcessId

92 , CallReply(..)

112 , CallReply(..)

@@ -113,35 +133,143 @@ mkClosure :: Name -> Q Exp

113mkClosure = varE . closureName

133mkClosure = varE . closureName

114

134

115--------------------------------------------------------------------------------

135--------------------------------------------------------------------------------

116−-- Closure combinators --

136+-- Generic closure combinators --

117--------------------------------------------------------------------------------

137--------------------------------------------------------------------------------

118

138

119−applyClosure :: Closure (a -> b) -> Closure a -> Closure b

139+closureApply :: Closure (a -> b) -> Closure a -> Closure b

120−applyClosure (Closure (Static labelf) envf) (Closure (Static labelx) envx) =

140+closureApply (Closure (Static labelf) envf) (Closure (Static labelx) envx) =

121− Closure (Static "$applyClosure") $ encode (labelf, envf, labelx, envx)

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

122

142

123−liftedBind :: forall a b. (Typeable a, Typeable b)

143+closureConst :: forall a b. (Typeable a, Typeable b)

124− => Closure (Process a -> (a -> Process b) -> Process b)

144+ => Closure (a -> b -> a)

125−liftedBind = Closure (Static "$liftedBind") (encode typ)

145+closureConst = Closure (Static ClosureConst) (encode $ typeOf aux)

126− where

146+ where

127− typ = typeOf (undefined :: Process a -> (a -> Process b) -> Process b)

147+ aux :: a -> b -> a

148+ aux = undefined

149+

150+closureUnit :: Closure ()

151+closureUnit = Closure (Static ClosureUnit) empty

152+

153+--------------------------------------------------------------------------------

154+-- Arrow combinators for processes --

155+--------------------------------------------------------------------------------

128

156

157+type CP a b = Closure (a -> Process b)

158+

159+cpIntro :: (Typeable a, Typeable b)

160+ => Closure (Process b) -> CP a b

161+cpIntro = closureApply closureConst

162+

163+cpElim :: Typeable a

164+ => CP () a -> Closure (Process a)

165+cpElim = flip closureApply closureUnit

166+

129−liftedConst :: forall a b. (Typeable a, Typeable b)

167+cpId :: forall a. Typeable a

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

168+ => CP a a

131−liftedConst = Closure (Static "$liftedConst") (encode typ)

169+cpId = Closure (Static CpId) (encode $ typeOf aux)

132 where

170 where

171+ aux :: a -> Process a

133− typ = typeOf (undefined :: a -> b -> a)

172+ aux = undefined

134

173

174+cpComp :: forall a b c. (Typeable a, Typeable b, Typeable c)

175+ => CP a b -> CP b c -> CP a c

176+cpComp f g = comp `closureApply` f `closureApply` g

177+ where

178+ comp :: Closure ((a -> Process b) -> (b -> Process c) -> (a -> Process c))

179+ comp = Closure (Static CpComp) (encode $ typeOf aux)

180+

181+ aux :: (a -> Process b) -> (b -> Process c) -> (a -> Process c)

182+ aux = undefined

183+

184+cpFirst :: forall a b c. (Typeable a, Typeable b, Typeable c)

185+ => CP a b -> CP (a, c) (b, c)

186+cpFirst = closureApply first

187+ where

188+ first :: Closure ((a -> Process b) -> (a, c) -> Process (b, c))

189+ first = Closure (Static CpFirst) (encode $ typeOf aux)

190+

191+ aux :: (a -> Process b) -> (a, c) -> Process (b, c)

192+ aux = undefined

193+

194+cpSwap :: forall a b. (Typeable a, Typeable b)

195+ => CP (a, b) (b, a)

196+cpSwap = Closure (Static CpSwap) (encode $ typeOf aux)

197+ where

198+ aux :: (a, b) -> Process (b, a)

199+ aux = undefined

200+

201+cpSecond :: (Typeable a, Typeable b, Typeable c)

202+ => CP a b -> CP (c, a) (c, b)

203+cpSecond f = cpSwap `cpComp` cpFirst f `cpComp` cpSwap

204+

205+cpPair :: (Typeable a, Typeable a', Typeable b, Typeable b')

206+ => CP a b -> CP a' b' -> CP (a, a') (b, b')

207+cpPair f g = cpFirst f `cpComp` cpSecond g

208+

209+cpCopy :: forall a. Typeable a

210+ => CP a (a, a)

211+cpCopy = Closure (Static CpCopy) (encode $ typeOf aux)

212+ where

213+ aux :: a -> Process (a, a)

214+ aux = undefined

215+

216+cpFanOut :: (Typeable a, Typeable b, Typeable c)

217+ => CP a b -> CP a c -> CP a (b, c)

218+cpFanOut f g = cpCopy `cpComp` (f `cpPair` g)

219+

220+cpLeft :: forall a b c. (Typeable a, Typeable b, Typeable c)

221+ => CP a b -> CP (Either a c) (Either b c)

222+cpLeft = closureApply left

223+ where

224+ left :: Closure ((a -> Process b) -> Either a c -> Process (Either b c))

225+ left = Closure (Static CpLeft) (encode $ typeOf aux)

226+

227+ aux :: (a -> Process b) -> Either a c -> Process (Either b c)

228+ aux = undefined

229+

135−bindCP :: (Typeable a, Typeable b)

230+cpMirror :: forall a b. (Typeable a, Typeable b)

136− => Closure (Process a) -> Closure (a -> Process b) -> Closure (Process b)

231+ => CP (Either a b) (Either b a)

137−bindCP x f = liftedBind `applyClosure` x `applyClosure` f

232+cpMirror = Closure (Static CpMirror) (encode $ typeOf aux)

233+ where

234+ aux :: Either a b -> Process (Either b a)

235+ aux = undefined

236+

237+cpRight :: forall a b c. (Typeable a, Typeable b, Typeable c)

238+ => CP a b -> CP (Either c a) (Either c b)

239+cpRight f = cpMirror `cpComp` cpLeft f `cpComp` cpMirror

240+

241+cpEither :: (Typeable a, Typeable a', Typeable b, Typeable b')

242+ => CP a b -> CP a' b' -> CP (Either a a') (Either b b')

243+cpEither f g = cpLeft f `cpComp` cpRight g

138

244

245+cpUntag :: forall a. Typeable a

246+ => CP (Either a a) a

247+cpUntag = Closure (Static CpUntag) (encode $ typeOf aux)

248+ where

249+ aux :: Either a a -> Process a

250+ aux = undefined

251+

139−weakenCP :: (Typeable a, Typeable b)

252+cpFanIn :: (Typeable a, Typeable b, Typeable c)

253+ => CP a c -> CP b c -> CP (Either a b) c

254+cpFanIn f g = (f `cpEither` g) `cpComp` cpUntag

255+

256+cpApply :: forall a b. (Typeable a, Typeable b)

257+ => CP (CP a b, a) b

258+cpApply = Closure (Static CpApply) (encode $ typeOf aux)

259+ where

140− => Closure (Process b) -> Closure (a -> Process b)

260+ aux :: (Closure (a -> Process b), a) -> Process b

141−weakenCP = applyClosure liftedConst

261+ aux = undefined

262+

263+--------------------------------------------------------------------------------

264+-- Some derived operators for processes --

265+--------------------------------------------------------------------------------

266+

267+cpBind :: (Typeable a, Typeable b)

268+ => Closure (Process a) -> Closure (a -> Process b) -> Closure (Process b)

269+cpBind x f = cpElim $ cpIntro x `cpComp` f

142

270

143−seqCP :: Closure (Process ()) -> Closure (Process ()) -> Closure (Process ())

271+cpSeq :: Closure (Process ()) -> Closure (Process ()) -> Closure (Process ())

144−seqCP p q = bindCP p (weakenCP q)

272+cpSeq p q = p `cpBind` cpIntro q

145

273

146--------------------------------------------------------------------------------

274--------------------------------------------------------------------------------

147-- Internal (Template Haskell) --

275-- Internal (Template Haskell) --

@@ -190,7 +318,7 @@ generateClosure :: Name -> Q Type -> Q Type -> Q ([Dec], String)

190generateClosure n arg res = do

318generateClosure n arg res = do

191 closure <- sequence

319 closure <- sequence

192 [ sigD (closureName n) [t| $arg -> Closure $res |]

320 [ sigD (closureName n) [t| $arg -> Closure $res |]

193− , sfnD (closureName n) [| Closure (Static $(stringE label)) . encode |]

321+ , sfnD (closureName n) [| Closure (Static (UserStatic ($(stringE label)))) . encode |]

194 ]

322 ]

195 return (closure, label)

323 return (closure, label)

196 where

324 where

distributed-process/src/Control/Distributed/Process/Internal/Types.hs+118−13

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

21 , SendPort(..)

22 , ReceivePort(..)

23 -- * Closures

24+ , StaticLabel(..)

24 , Static(..)

25 , Static(..)

25 , Closure(..)

26 , Closure(..)

26 , CallReply(..)

27 , CallReply(..)

@@ -225,9 +226,28 @@ data ReceivePort a =

225-- Closures --

226-- Closures --

226--------------------------------------------------------------------------------

227--------------------------------------------------------------------------------

227

228

229+data StaticLabel =

230+ UserStatic String

231+ -- Special support for 'call'

232+ | Call

233+ -- Generic closure combinators

234+ | ClosureApply

235+ | ClosureConst

236+ | ClosureUnit

237+ -- Arrow combinators for processes

238+ | CpId

239+ | CpComp

240+ | CpFirst

241+ | CpSwap

242+ | CpCopy

243+ | CpLeft

244+ | CpMirror

245+ | CpUntag

246+ | CpApply

247+ deriving (Typeable, Show)

248+

228-- | A static value is one that is bound at top-level.

249-- | A static value is one that is bound at top-level.

229−-- We represent it simply by a string

230−newtype Static a = Static String

250+newtype Static a = Static StaticLabel

231 deriving (Typeable, Show)

251 deriving (Typeable, Show)

232

252

233-- | A closure is a static value and an encoded environment

253-- | A closure is a static value and an encoded environment

@@ -254,34 +274,83 @@ registerLabel label dyn = remoteTableLabel label ^= Just dyn

254registerSender :: TypeRep -> Dynamic -> RemoteTable -> RemoteTable

274registerSender :: TypeRep -> Dynamic -> RemoteTable -> RemoteTable

255registerSender typ dyn = remoteTableSender typ ^= Just dyn

275registerSender typ dyn = remoteTableSender typ ^= Just dyn

256

276

257−resolveClosure :: RemoteTable -> String -> ByteString -> Maybe Dynamic

277+resolveClosure :: RemoteTable -> StaticLabel -> ByteString -> Maybe Dynamic

258−resolveClosure rtable "$call" env = do

278+-- Special support for call

279+resolveClosure rtable Call env = do

259 proc <- resolveClosure rtable label env'

280 proc <- resolveClosure rtable label env'

260 sender <- rtable ^. remoteTableSender (dynTypeRep proc)

281 sender <- rtable ^. remoteTableSender (dynTypeRep proc)

261 proc `bind` (sender `dynApp` toDyn (pid :: ProcessId))

282 proc `bind` (sender `dynApp` toDyn (pid :: ProcessId))

262 where

283 where

263 (label, env', pid) = decode env

284 (label, env', pid) = decode env

264− bind = dynBind tyConProcess bindProcess

285+ bind = dynBind tyConProcess bindProcess

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

287+ bindProcess = (>>=)

288+-- Generic closure combinators

265−resolveClosure rtable "$applyClosure" env = do

289+resolveClosure rtable ClosureApply env = do

266 f <- resolveClosure rtable labelf envf

290 f <- resolveClosure rtable labelf envf

267 x <- resolveClosure rtable labelx envx

291 x <- resolveClosure rtable labelx envx

268 f `dynApply` x

292 f `dynApply` x

269 where

293 where

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

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

271−resolveClosure _rtable "$liftedBind" env =

295+resolveClosure _rtable ClosureConst env =

272− return $ Dynamic (decode env) (unsafeCoerce# bindProcess)

273−resolveClosure _rtable "$liftedConst" env =

274 return $ Dynamic (decode env) (unsafeCoerce# const)

296 return $ Dynamic (decode env) (unsafeCoerce# const)

297+resolveClosure _rtable ClosureUnit _env =

298+ return $ toDyn ()

299+-- Arrow combinators

300+resolveClosure _rtable CpId env =

301+ return $ Dynamic (decode env) (unsafeCoerce# cpId)

302+ where

303+ cpId :: forall a. a -> Process a

304+ cpId = return

305+resolveClosure _rtable CpComp env =

306+ return $ Dynamic (decode env) (unsafeCoerce# cpComp)

307+ where

308+ cpComp :: forall a b c. (a -> Process b) -> (b -> Process c) -> a -> Process c

309+ cpComp p q a = p a >>= q

310+resolveClosure _rtable CpFirst env =

311+ return $ Dynamic (decode env) (unsafeCoerce# cpFirst)

312+ where

313+ cpFirst :: forall a b c. (a -> Process b) -> (a, c) -> Process (b, c)

314+ cpFirst = undefined

315+resolveClosure _rtable CpSwap env =

316+ return $ Dynamic (decode env) (unsafeCoerce# cpSwap)

317+ where

318+ cpSwap :: forall a b. (a, b) -> Process (b, a)

319+ cpSwap = undefined

320+resolveClosure _rtable CpCopy env =

321+ return $ Dynamic (decode env) (unsafeCoerce# cpCopy)

322+ where

323+ cpCopy :: forall a. a -> Process (a, a)

324+ cpCopy = undefined

325+resolveClosure _rtable CpLeft env =

326+ return $ Dynamic (decode env) (unsafeCoerce# cpLeft)

327+ where

328+ cpLeft :: forall a b c. (a -> Process b) -> Either a c -> Process (Either b c)

329+ cpLeft = undefined

330+resolveClosure _rtable CpMirror env =

331+ return $ Dynamic (decode env) (unsafeCoerce# cpMirror)

332+ where

333+ cpMirror :: forall a b. Either a b -> Process (Either b a)

334+ cpMirror = undefined

335+resolveClosure _rtable CpUntag env =

336+ return $ Dynamic (decode env) (unsafeCoerce# cpUntag)

337+ where

338+ cpUntag :: forall a. Either a a -> Process a

339+ cpUntag = undefined

340+resolveClosure _rtable CpApply env =

341+ return $ Dynamic (decode env) (unsafeCoerce# cpApply)

342+ where

343+ cpApply :: forall a b. Closure (Process (a -> Process b), a) -> Process b

344+ cpApply = undefined

345+

346+-- User defined closures

275−resolveClosure rtable label env = do

347+resolveClosure rtable (UserStatic label) env = do

276 val <- rtable ^. remoteTableLabel label

348 val <- rtable ^. remoteTableLabel label

277 dynApply val (toDyn env)

349 dynApply val (toDyn env)

278

350

279tyConProcess :: TyCon

351tyConProcess :: TyCon

280tyConProcess = typeRepTyCon (typeOf (undefined :: Process ()))

352tyConProcess = typeRepTyCon (typeOf (undefined :: Process ()))

281

353

282−bindProcess :: Process a -> (a -> Process b) -> Process b

283−bindProcess = (>>=)

284−

285--------------------------------------------------------------------------------

354--------------------------------------------------------------------------------

286-- Messages --

355-- Messages --

287--------------------------------------------------------------------------------

356--------------------------------------------------------------------------------

@@ -469,6 +538,42 @@ instance Binary Identifier where

469 2 -> ChannelIdentifier <$> get

538 2 -> ChannelIdentifier <$> get

470 _ -> fail "Identifier.get: invalid"

539 _ -> fail "Identifier.get: invalid"

471

540

541+instance Binary StaticLabel where

542+ put (UserStatic string) = putWord8 0 >> put string

543+ put Call = putWord8 1

544+ put ClosureApply = putWord8 2

545+ put ClosureConst = putWord8 3

546+ put ClosureUnit = putWord8 4

547+ put CpId = putWord8 5

548+ put CpComp = putWord8 6

549+ put CpFirst = putWord8 7

550+ put CpSwap = putWord8 8

551+ put CpCopy = putWord8 9

552+ put CpLeft = putWord8 10

553+ put CpMirror = putWord8 11

554+ put CpUntag = putWord8 12

555+ put CpApply = putWord8 13

556+ get = do

557+ header <- getWord8

558+ case header of

559+ 0 -> UserStatic <$> get

560+ 1 -> return Call

561+ 2 -> return ClosureApply

562+ 3 -> return ClosureConst

563+ 4 -> return ClosureUnit

564+ 5 -> return CpId

565+ 6 -> return CpComp

566+ 7 -> return CpFirst

567+ 8 -> return CpSwap

568+ 9 -> return CpCopy

569+ 10 -> return CpLeft

570+ 11 -> return CpMirror

571+ 12 -> return CpUntag

572+ 13 -> return CpApply

573+ _ -> fail "StaticLabel.get: invalid"

574+

575+

576+

472--------------------------------------------------------------------------------

577--------------------------------------------------------------------------------

473-- Accessors --

578-- Accessors --

474--------------------------------------------------------------------------------

579--------------------------------------------------------------------------------

distributed-process/tests/TestClosure.hs+3−3

@@ -147,10 +147,10 @@ testCall transport rtable = do

147 takeMVar clientDone

148

149sendFac :: Int -> ProcessId -> Closure (Process ())

150−sendFac n pid = $(mkClosure 'factorial) n `bindCP` $(mkClosure 'sendInt) pid

150+sendFac n pid = $(mkClosure 'factorial) n `cpBind` $(mkClosure 'sendInt) pid

151

152factorial' :: Int -> Closure (Process Int)

153−factorial' n = $(mkClosure 'returnInt) n `bindCP` $(mkClosure 'factorialOf) ()

153+factorial' n = $(mkClosure 'returnInt) n `cpBind` $(mkClosure 'factorialOf) ()

154

155testBind :: Transport -> RemoteTable -> IO ()

156testBind transport rtable = do

@@ -184,7 +184,7 @@ testSeq transport rtable = do

184 node <- newLocalNode transport rtable

185 runProcess node $ do

186 us <- getSelfPid

187− join . unClosure $ sendFac 5 us `seqCP` sendFac 6 us

187+ join . unClosure $ sendFac 5 us `cpSeq` sendFac 6 us

188 120 :: Int <- expect

189 720 :: Int <- expect

190 return ()