685b8ec · distributed-process

git.iko.soy/distributed-process

on master help sign in

Cleanup, TODOs, start work on the backend

685b8ecEdsko de Vries14 years ago

distributed-process/distributed-process.cabal+3

compare with master

unifiedsplit

@@ -21,11 +21,14 @@ Library

21 template-haskell,

22 stm,

23 ghc-prim,

24+ network,

25+ network-multicast,

24 network-transport >= 0.1

26 network-transport >= 0.1

25 Exposed-modules: Control.Distributed.Process,

27 Exposed-modules: Control.Distributed.Process,

26 Control.Distributed.Process.Serializable,

28 Control.Distributed.Process.Serializable,

27 Control.Distributed.Process.Closure,

29 Control.Distributed.Process.Closure,

28 Control.Distributed.Process.Node

30 Control.Distributed.Process.Node

31+ Control.Distributed.Process.Backend.Local

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

32 Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell, ExistentialQuantification, RankNTypes, GADTs

30 ghc-options: -Wall

33 ghc-options: -Wall

31 HS-Source-Dirs: src

34 HS-Source-Dirs: src

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

@@ -199,10 +199,12 @@ import Control.Distributed.Process.Internal.Closure (resolveClosure)

199--------------------------------------------------------------------------------

200

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

202+--

203+-- We link to the remote process, so that if it dies, we die too.

202call :: SerializableDict a -> NodeId -> Closure (Process a) -> Process a

204call :: SerializableDict a -> NodeId -> Closure (Process a) -> Process a

203call sdict@SerializableDict nid proc = do

205call sdict@SerializableDict nid proc = do

204 us <- getSelfPid

206 us <- getSelfPid

205− spawn nid (proc `cpBind` sendClosure sdict us)

207+ spawnLink nid (proc `cpBind` sendClosure sdict us)

206 expect

208 expect

207

209

208-- | Spawn a child process, have the child link to the parent and the parent

210-- | Spawn a child process, have the child link to the parent and the parent

distributed-process/src/Control/Distributed/Process/Backend/Local.hsnew file+80

@@ -0,0 +1,80 @@

1+-- | Simple backend based on the TCP transport which offers node discovery

2+-- based on UDP multicast

3+module Control.Distributed.Process.Backend.Local

4+ ( LocalBackend(..)

5+ , initializeBackend

6+ ) where

8+import Data.Binary (encode)

9+import qualified Data.ByteString as BSS (concat, append)

10+import qualified Data.ByteString.Lazy as BSL (toChunks, length)

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

12+import Control.Exception (throw)

13+import Control.Monad (forM_)

14+import Control.Concurrent (forkIO)

15+import Control.Concurrent.MVar (MVar, newMVar, readMVar)

16+import Control.Distributed.Process.Internal.Types

17+ ( LocalNode

18+ , RemoteTable

19+ , NodeId

20+ )

21+import qualified Control.Distributed.Process.Node as Node

22+ ( newLocalNode

23+ , localNodeId

24+ )

25+import qualified Network.Transport.TCP as NT

26+ ( createTransport

27+ , defaultTCPParameters

28+ )

29+import Network.Transport.Internal (encodeInt32, decodeInt32)

30+import qualified Network.Transport as NT ()

31+import qualified Network.Socket as N (HostName, ServiceName, Socket, SockAddr)

32+import qualified Network.Socket.ByteString as NBS (recvFrom, sendTo)

33+import qualified Network.Multicast as NM (multicastSender, multicastReceiver)

34+

35+data LocalBackend = LocalBackend {

36+ newLocalNode :: IO LocalNode

37+ , findPeers :: IO [NodeId]

38+ }

39+

40+data LocalBackendState = LocalBackendState {

41+ localNodes :: [LocalNode]

42+ }

43+

44+-- | Initialize the backend

45+initializeBackend :: N.HostName -> N.ServiceName -> RemoteTable -> IO LocalBackend

46+initializeBackend host port rtable = do

47+ mTransport <- NT.createTransport host port NT.defaultTCPParameters

48+ mcastSender <- NM.multicastSender "224.0.0.99" 9999

49+ mcastReceiver <- NM.multicastReceiver "224.0.0.99" 9999

50+ backendState <- newMVar LocalBackendState { localNodes = [] }

51+ forkIO $ peerDiscoveryDaemon backendState mcastReceiver mcastSender

52+ case mTransport of

53+ Left err -> throw err

54+ Right transport -> return LocalBackend {

55+ newLocalNode = Node.newLocalNode transport rtable

56+ , findPeers = apiFindPeers mcastSender

57+ }

58+

59+-- | Peer discovery

60+apiFindPeers :: (N.Socket, N.SockAddr) -> IO [NodeId]

61+apiFindPeers (sock, addr) = do

62+ NBS.sendTo sock (encodeInt32 (0xCDAECDAE :: Int)) addr

63+ return []

64+

65+-- | Respond to peer discovery requests sent by other nodes

66+peerDiscoveryDaemon :: MVar LocalBackendState

67+ -> N.Socket

68+ -> (N.Socket, N.SockAddr)

69+ -> IO ()

70+peerDiscoveryDaemon backendState readSock (writeSock, addr) = do

71+ (msgLen, _remoteAddr) <- NBS.recvFrom readSock 4

72+

73+ case decodeInt32 msgLen of

74+ (0xCDAECDAE :: Int) -> do

75+ nodes <- localNodes <$> readMVar backendState

76+ forM_ nodes $ \node -> do

77+ let nodeAddr = encode (Node.localNodeId node)

78+ msg = encodeInt32 (BSL.length nodeAddr) `BSS.append`

79+ (BSS.concat . BSL.toChunks $ nodeAddr)

80+ NBS.sendTo writeSock msg addr

distributed-process/src/Control/Distributed/Process/Internal/Process/Primitives.hs+2

@@ -247,6 +247,8 @@ getSelfNode = localNodeId <$> procMsg getLocalNode

247-- | Link to a remote process (asynchronous)

248--

249-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').

250+-- Linking makes no distinction between normal and abnormal termination of

251+-- the remote process.

250link :: ProcessId -> Process ()

252link :: ProcessId -> Process ()

251link = sendCtrlMsg Nothing . Link

253link = sendCtrlMsg Nothing . Link

252

254

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

@@ -197,7 +197,7 @@ forkProcess node proc = modifyMVar (localState node) $ \st -> do

197 $ st

198 , pid

199 )

200−

200+

201handleIncomingMessages :: LocalNode -> IO ()

202handleIncomingMessages node = go [] Map.empty Map.empty Set.empty

203 where

@@ -210,6 +210,8 @@ handleIncomingMessages node = go [] Map.empty Map.empty Set.empty

210 event <- NT.receive endpoint

211 case event of

212 NT.ConnectionOpened cid _rel _theirAddr ->

213+ -- TODO: Check if _rel is ReliableOrdered, and if not, treat as

214+ -- (**) below.

213 go (cid : uninitConns) procs chans ctrls

215 go (cid : uninitConns) procs chans ctrls

214 NT.Received cid payload ->

216 NT.Received cid payload ->

215 case ( Map.lookup cid procs

217 case ( Map.lookup cid procs

@@ -240,10 +242,12 @@ handleIncomingMessages node = go [] Map.empty Map.empty Set.empty

240 ctrls

242 ctrls

241 Nothing ->

243 Nothing ->

242 -- Request for an unknown process.

244 -- Request for an unknown process.

243− -- TODO: We should close the incoming connection here, but

245+ --

244− -- we cannot! Network.Transport does not provide this

246+ -- TODO: We should treat this as a fatal error on the

245− -- functionality. Not sure what the right approach is.

247+ -- part of the remote node. That is, we should report the

246− -- For now, we just drop the incoming messages

248+ -- remote node as having died, and we should close

249+ -- incoming connections (this requires a Transport layer

250+ -- extension). (**)

247 go (List.delete cid uninitConns) procs chans ctrls

251 go (List.delete cid uninitConns) procs chans ctrls

248 ChannelIdentifier chId -> do

252 ChannelIdentifier chId -> do

249 let lcid = channelLocalId chId

253 let lcid = channelLocalId chId

@@ -259,8 +263,12 @@ handleIncomingMessages node = go [] Map.empty Map.empty Set.empty

259 (Map.insert cid channel chans)

263 (Map.insert cid channel chans)

260 ctrls

264 ctrls

261 Nothing ->

265 Nothing ->

266+ -- Unknown typed channel

267+ -- TODO (**) above

262 go (List.delete cid uninitConns) procs chans ctrls

268 go (List.delete cid uninitConns) procs chans ctrls

263 Nothing ->

269 Nothing ->

270+ -- Unknown process

271+ -- TODO (**) above

264 go (List.delete cid uninitConns) procs chans ctrls

272 go (List.delete cid uninitConns) procs chans ctrls

265 NodeIdentifier _ ->

273 NodeIdentifier _ ->

266 go (List.delete cid uninitConns)

274 go (List.delete cid uninitConns)

@@ -268,7 +276,8 @@ handleIncomingMessages node = go [] Map.empty Map.empty Set.empty

268 chans

276 chans

269 (Set.insert cid ctrls)

277 (Set.insert cid ctrls)

270 _ ->

278 _ ->

271− -- Unexpected message. We just drop it.

279+ -- Unexpected message

280+ -- TODO (**) above

272 go uninitConns procs chans ctrls

281 go uninitConns procs chans ctrls

273 NT.ConnectionClosed cid ->

282 NT.ConnectionClosed cid ->

274 go (List.delete cid uninitConns)

283 go (List.delete cid uninitConns)

@@ -283,9 +292,9 @@ handleIncomingMessages node = go [] Map.empty Map.empty Set.empty

283 , ctrlMsgSignal = Died nid DiedDisconnect

292 , ctrlMsgSignal = Died nid DiedDisconnect

284 }

293 }

285 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost Nothing _) _) ->

294 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost Nothing _) _) ->

286− -- We cannot really deal with the case that some (but not all)

295+ -- TODO: We should treat an asymetrical connection loss (incoming

287− -- incoming connections from a remote node have failed (what if this

296+ -- connection broken, but outgoing connection still potentially ok)

288− -- incoming connection corresponds to a typed channel, for instance?)

297+ -- as a fatal error on the part of the remote node (like (**), above)

289 fail "handleIncomingMessages: TODO"

298 fail "handleIncomingMessages: TODO"

290 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->

299 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->

291 fail $ "Cloud Haskell fatal error: end point failed: " ++ str

300 fail $ "Cloud Haskell fatal error: end point failed: " ++ str

distributed-process/tests/TestLocalBackend.hsnew file+27

@@ -0,0 +1,27 @@

1+import System.Environment (getArgs, getProgName)

2+import Control.Distributed.Process.Node (initRemoteTable)

3+import Control.Distributed.Process.Backend.Local

4+import Control.Concurrent (threadDelay)

6+server :: LocalBackend -> IO ()

7+server backend = do

8+ nodes <- findPeers backend

9+ print nodes

10+

11+client :: LocalBackend -> IO ()

12+client _ = threadDelay 30000000

13+

14+main :: IO ()

15+main = do

16+ prog <- getProgName

17+ args <- getArgs

18+

19+ case args of

20+ ["server", host, port] -> do

21+ backend <- initializeBackend host port initRemoteTable

22+ server backend

23+ ["client", host, port] -> do

24+ backend <- initializeBackend host port initRemoteTable

25+ client backend

26+ _ ->

27+ putStrLn $ "usage: " ++ prog ++ " (server | client) host port"