2c89e89 · distributed-process

git.iko.soy/distributed-process

on master help sign in

More consistent naming

2c89e89Edsko de Vries14 years ago

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

compare with master

unifiedsplit

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

56 , expect

57 , send

58 , getSelfPid

59+ , monitor

59 , monitorProcess

60 , monitorProcess

60 , linkProcess

61 , linkProcess

61 -- * Monitoring

62 -- * Monitoring

@@ -127,56 +128,58 @@ import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))

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

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

128import System.Random (randomIO)

129import System.Random (randomIO)

129

130

131+-- | A local process ID consists of a seed which distinguishes processes from

132+-- different instances of the same local node and a counter

130data LocalProcessId = LocalProcessId

133data LocalProcessId = LocalProcessId

131− { pidUnique :: Int32

134+ { lpidUnique :: Int32

132− , pidCounter :: Int32

135+ , lpidCounter :: Int32

133 }

136 }

134 deriving (Eq, Ord, Typeable, Show)

137 deriving (Eq, Ord, Typeable, Show)

135

138

136−instance Binary LocalProcessId where

139+-- | A process ID combines a local process with with an endpoint address

137− put lpid = put (pidUnique lpid) >> put (pidCounter lpid)

140+-- (in other words, we identify nodes and endpoints)

138− get = liftM2 LocalProcessId get get

139−

140data ProcessId = ProcessId

141data ProcessId = ProcessId

141 { processAddress :: NT.EndPointAddress

142 { processAddress :: NT.EndPointAddress

142 , processLocalId :: LocalProcessId

143 , processLocalId :: LocalProcessId

143 }

144 }

144 deriving (Eq, Ord, Typeable, Show)

145 deriving (Eq, Ord, Typeable, Show)

145

146

146−instance Binary ProcessId where

147+-- | Messages consist of their typeRep fingerprint and their encoding

147− put pid = put (processAddress pid) >> put (processLocalId pid)

148− get = liftM2 ProcessId get get

149−

150data Message = Message

148data Message = Message

151 { messageFingerprint :: Fingerprint

149 { messageFingerprint :: Fingerprint

152 , messageEncoding :: BSL.ByteString

150 , messageEncoding :: BSL.ByteString

153 }

151 }

154

152

153+-- | Local nodes

155data LocalNode = LocalNode

154data LocalNode = LocalNode

156 { localEndPoint :: NT.EndPoint

155 { localEndPoint :: NT.EndPoint

157 , localState :: MVar LocalNodeState

156 , localState :: MVar LocalNodeState

158 }

157 }

159

158

159+-- | Local node state

160data LocalNodeState = LocalNodeState

161 { _localConnections :: Map ProcessId NT.Connection

162− , _localProcesses :: Map LocalProcessId ProcessState

162+ , _localProcesses :: Map LocalProcessId LocalProcess

163 , _localPidCounter :: Int32

164 , _localPidUnique :: Int32

165 }

166

167−data ProcessState = ProcessState

167+-- | Processes running on our local node

168+data LocalProcess = LocalProcess

168 { processQueue :: CQueue Message

169 { processQueue :: CQueue Message

169 , processNode :: LocalNode

170 , processNode :: LocalNode

170 , processId :: ProcessId

171 , processId :: ProcessId

171 }

172 }

172

173

174+-- The Cloud Haskell 'Process' type

173−newtype Process a = Process { unProcess :: ReaderT ProcessState IO a }

175+newtype Process a = Process { unProcess :: ReaderT LocalProcess IO a }

174− deriving (Functor, Monad, MonadIO, MonadReader ProcessState)

176+ deriving (Functor, Monad, MonadIO, MonadReader LocalProcess)

175

177

176--------------------------------------------------------------------------------

178--------------------------------------------------------------------------------

177-- Basic Cloud Haskell API --

179-- Basic Cloud Haskell API --

178--------------------------------------------------------------------------------

180--------------------------------------------------------------------------------

179

181

182+-- | Wait for a message of a specific type

180expect :: forall a. Serializable a => Process a

183expect :: forall a. Serializable a => Process a

181expect = do

184expect = do

182 queue <- processQueue <$> ask

185 queue <- processQueue <$> ask

@@ -184,6 +187,7 @@ expect = do

184 msg <- liftIO $ dequeueMatching queue ((== fp) . messageFingerprint)

187 msg <- liftIO $ dequeueMatching queue ((== fp) . messageFingerprint)

185 return (decode . messageEncoding $ msg)

188 return (decode . messageEncoding $ msg)

186

189

190+-- | Send a message

187send :: Serializable a => ProcessId -> a -> Process ()

191send :: Serializable a => ProcessId -> a -> Process ()

188send pid msg = do

192send pid msg = do

189 -- This requires a lookup on every send. If we want to avoid that we need to

193 -- This requires a lookup on every send. If we want to avoid that we need to

@@ -195,12 +199,19 @@ send pid msg = do

195 : BSL.toChunks (encode msg)

199 : BSL.toChunks (encode msg)

196 )

200 )

197

201

202+-- | Our own process ID

198getSelfPid :: Process ProcessId

203getSelfPid :: Process ProcessId

199getSelfPid = processId <$> ask

204getSelfPid = processId <$> ask

200

205

206+-- | Variation on 'monitorProcess' where the monitor is the current process

207+monitor :: ProcessId -> MonitorAction -> Process ()

208+monitor _them = undefined

209+

210+-- | Have one process monitor another

201monitorProcess :: ProcessId -> ProcessId -> MonitorAction -> Process ()

211monitorProcess :: ProcessId -> ProcessId -> MonitorAction -> Process ()

202monitorProcess = undefined

212monitorProcess = undefined

203

213

214+-- | Link failure in two processes

204linkProcess :: ProcessId -> Process ()

215linkProcess :: ProcessId -> Process ()

205linkProcess = undefined

216linkProcess = undefined

206

217

@@ -325,13 +336,13 @@ forkProcess :: LocalNode -> Process () -> IO ProcessId

325forkProcess node proc = do

336forkProcess node proc = do

326 queue <- newCQueue

337 queue <- newCQueue

327 state <- modifyMVar (localState node) $ \st -> do

338 state <- modifyMVar (localState node) $ \st -> do

328− let lpid = LocalProcessId { pidCounter = st ^. localPidCounter

339+ let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter

329− , pidUnique = st ^. localPidUnique

340+ , lpidUnique = st ^. localPidUnique

330 }

341 }

331 let pid = ProcessId { processAddress = NT.address (localEndPoint node)

342 let pid = ProcessId { processAddress = NT.address (localEndPoint node)

332 , processLocalId = lpid

343 , processLocalId = lpid

333 }

344 }

334− let state = ProcessState { processQueue = queue

345+ let state = LocalProcess { processQueue = queue

335 , processNode = node

346 , processNode = node

336 , processId = pid

347 , processId = pid

337 }

348 }

@@ -444,16 +455,28 @@ payloadToMessage payload = Message fp msg

444 fp = decodeFingerprint . BSS.concat . BSL.toChunks $ encFp

455 fp = decodeFingerprint . BSS.concat . BSL.toChunks $ encFp

445

456

446lpidToPayload :: LocalProcessId -> [BSS.ByteString]

457lpidToPayload :: LocalProcessId -> [BSS.ByteString]

447−lpidToPayload lpid = [ NTI.encodeInt32 (pidCounter lpid)

458+lpidToPayload lpid = [ NTI.encodeInt32 (lpidCounter lpid)

448− , NTI.encodeInt32 (pidUnique lpid)

459+ , NTI.encodeInt32 (lpidUnique lpid)

449 ]

460 ]

450

461

451payloadToLpid :: [BSS.ByteString] -> LocalProcessId

462payloadToLpid :: [BSS.ByteString] -> LocalProcessId

452payloadToLpid bss = let (bs1, bs2) = BSS.splitAt 4 . BSS.concat $ bss

463payloadToLpid bss = let (bs1, bs2) = BSS.splitAt 4 . BSS.concat $ bss

453− in LocalProcessId { pidCounter = NTI.decodeInt32 bs1

464+ in LocalProcessId { lpidCounter = NTI.decodeInt32 bs1

454− , pidUnique = NTI.decodeInt32 bs2

465+ , lpidUnique = NTI.decodeInt32 bs2

455 }

466 }

456

467

468+--------------------------------------------------------------------------------

469+-- Binary instances --

470+--------------------------------------------------------------------------------

471+

472+instance Binary LocalProcessId where

473+ put lpid = put (lpidUnique lpid) >> put (lpidCounter lpid)

474+ get = liftM2 LocalProcessId get get

475+

476+instance Binary ProcessId where

477+ put pid = put (processAddress pid) >> put (processLocalId pid)

478+ get = liftM2 ProcessId get get

479+

457--------------------------------------------------------------------------------

480--------------------------------------------------------------------------------

458-- Accessors --

481-- Accessors --

459--------------------------------------------------------------------------------

482--------------------------------------------------------------------------------

@@ -461,7 +484,7 @@ payloadToLpid bss = let (bs1, bs2) = BSS.splitAt 4 . BSS.concat $ bss

461localConnections :: Accessor LocalNodeState (Map ProcessId NT.Connection)

484localConnections :: Accessor LocalNodeState (Map ProcessId NT.Connection)

462localConnections = accessor _localConnections (\conns st -> st { _localConnections = conns })

485localConnections = accessor _localConnections (\conns st -> st { _localConnections = conns })

463

486

464−localProcesses :: Accessor LocalNodeState (Map LocalProcessId ProcessState)

487+localProcesses :: Accessor LocalNodeState (Map LocalProcessId LocalProcess)

465localProcesses = accessor _localProcesses (\procs st -> st { _localProcesses = procs })

488localProcesses = accessor _localProcesses (\procs st -> st { _localProcesses = procs })

466

489

467localPidCounter :: Accessor LocalNodeState Int32

490localPidCounter :: Accessor LocalNodeState Int32

@@ -473,5 +496,5 @@ localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq

473localConnectionTo :: ProcessId -> Accessor LocalNodeState (Maybe NT.Connection)

496localConnectionTo :: ProcessId -> Accessor LocalNodeState (Maybe NT.Connection)

474localConnectionTo pid = localConnections >>> DAC.mapMaybe pid

497localConnectionTo pid = localConnections >>> DAC.mapMaybe pid

475

498

476−localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe ProcessState)

499+localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess)

477localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid

500localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid