fad9b46 · distributed-process

distributed-process/src/Control/Distributed/Process.hs+10

compare with master

unifiedsplit

@@ -19,6 +19,8 @@ module Control.Distributed.Process

19 , ReceivePort

20 , SendPort

21 , newChan

22+ , sendChan

23+ , receiveChan

22 -- * Advanced messaging

24 -- * Advanced messaging

23 , Match

25 , Match

24 , receiveWait

26 , receiveWait

@@ -208,6 +210,14 @@ newChan = do

208 , (sport, rport)

210 , (sport, rport)

209 )

211 )

210

212

213+sendChan :: Serializable a => SendPort a -> a -> Process ()

214+sendChan (SendPort them) msg = lift $ sendBinary (ChannelIdentifier them) msg

215+

216+receiveChan :: Serializable a => ReceivePort a -> Process a

217+receiveChan (ReceivePort queue) = do

218+ Just msg <- liftIO $ dequeue queue Blocking [Just]

219+ return msg

220+

211--------------------------------------------------------------------------------

221--------------------------------------------------------------------------------

212-- Advanced messaging --

222-- Advanced messaging --

213--------------------------------------------------------------------------------

223--------------------------------------------------------------------------------

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

@@ -51,7 +51,7 @@ dequeue (CQueue arrived incoming) blockSpec matches =

51 go = mask $ \restore -> do

52 arr <- takeMVar arrived

53 -- We first check the arrived messages. If we timeout during this search,

54− -- we just put the MVar back (we haven't touched the Chan yet)

54+ -- we just put the MVar back (since we haven't touched the Chan yet)

55 (arr', mb) <- onException (restore (checkArrived [] arr))

56 (putMVar arrived arr)

57 case (mb, blockSpec) of

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

@@ -200,7 +200,7 @@ data ChannelId = ChannelId {

200

201data TypedChannel = forall a. Serializable a => TypedChannel (CQueue a)

202

203−newtype SendPort a = SendPort ChannelId

203+newtype SendPort a = SendPort ChannelId deriving (Typeable, Binary)

204newtype ReceivePort a = ReceivePort (CQueue a)

205

206--------------------------------------------------------------------------------

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

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

89 , runMessageT

90 , TypedChannel(..)

91 , Identifier(..)

92+ , ChannelId(..)

93+ , typedChannelWithId

92 )

94 )

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

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

94import Control.Distributed.Process.Internal.MessageT

96import Control.Distributed.Process.Internal.MessageT

@@ -229,12 +231,24 @@ handleIncomingMessages node = go [] Map.empty Map.empty Set.empty

229 -- we cannot! Network.Transport does not provide this

231 -- we cannot! Network.Transport does not provide this

230 -- functionality. Not sure what the right approach is.

232 -- functionality. Not sure what the right approach is.

231 -- For now, we just drop the incoming messages

233 -- For now, we just drop the incoming messages

232− go (List.delete cid uninitConns)

234+ go (List.delete cid uninitConns) procs chans ctrls

233− procs

234− chans

235− ctrls

236− ChannelIdentifier _chId ->

235+ ChannelIdentifier chId -> do

237− fail "Connecting to typed channels not yet implemented"

236+ let lcid = channelLocalId chId

237+ lpid = processLocalId (channelProcessId chId)

238+ mProc <- withMVar state $ return . (^. localProcessWithId lpid)

239+ case mProc of

240+ Just proc -> do

241+ mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)

242+ case mChannel of

243+ Just channel ->

244+ go (List.delete cid uninitConns)

245+ procs

246+ (Map.insert cid channel chans)

247+ ctrls

248+ Nothing ->

249+ go (List.delete cid uninitConns) procs chans ctrls

250+ Nothing ->

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

238 NodeIdentifier _ ->

252 NodeIdentifier _ ->

239 go (List.delete cid uninitConns)

253 go (List.delete cid uninitConns)

240 procs

254 procs

distributed-process/tests/TestCH.hs+31−4

@@ -369,15 +369,42 @@ testTimeout0 transport = do

369

370 takeMVar clientDone

371

372+-- | Test typed channels

373+testTypedChannels :: NT.Transport -> IO ()

374+testTypedChannels transport = do

375+ serverChannel <- newEmptyMVar :: IO (MVar (SendPort (SendPort Bool, Int)))

376+ clientDone <- newEmptyMVar

377+

378+ forkIO $ do

379+ localNode <- newLocalNode transport initRemoteTable

380+ forkProcess localNode $ do

381+ (serverSendPort, rport) <- newChan

382+ liftIO $ putMVar serverChannel serverSendPort

383+ (clientSendPort, i) <- receiveChan rport

384+ sendChan clientSendPort (even i)

385+ return ()

386+

387+ forkIO $ do

388+ localNode <- newLocalNode transport initRemoteTable

389+ serverSendPort <- readMVar serverChannel

390+ runProcess localNode $ do

391+ (clientSendPort, rport) <- newChan

392+ sendChan serverSendPort (clientSendPort, 5)

393+ False <- receiveChan rport

394+ liftIO $ putMVar clientDone ()

395+

396+ takeMVar clientDone

397+

372main :: IO ()

398main :: IO ()

373main = do

399main = do

374 Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters

400 Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters

375 runTests

401 runTests

376− [ ("Ping", testPing transport)

402+ [ ("Ping", testPing transport)

377− , ("Math", testMath transport)

403+ , ("Math", testMath transport)

378− , ("Timeout", testTimeout transport)

404+ , ("Timeout", testTimeout transport)

379− , ("Timeout0", testTimeout0 transport)

405+ , ("Timeout0", testTimeout0 transport)

380 , ("SendToTerminated", testSendToTerminated transport)

406 , ("SendToTerminated", testSendToTerminated transport)

407+ , ("TypedChannnels", testTypedChannels transport)

381 -- The "missing" combinations in the list below don't make much sense, as

408 -- The "missing" combinations in the list below don't make much sense, as

382 -- we cannot guarantee that the monitor reply or link exception will not

409 -- we cannot guarantee that the monitor reply or link exception will not

383 -- happen before the unmonitor or unlink

410 -- happen before the unmonitor or unlink