git.iko.soy/distributed-process
on master help sign in
First implementation of typed channels
commitfad9b46dbf13636dccd11906a7985209d5bc7b25 authorEdsko de Vries <edsko@edsko.net> dateWed 20 Jun 2012, 17:53 (14 years ago) parent4d0f927
5 files changed+63−12
compare with master
unifiedsplit
distributed-process/src/Control/Distributed/Process.hs+10 distributed-process/src/Control/Distributed/Process/Internal/CQueue.hs+1−1 distributed-process/src/Control/Distributed/Process/Internal/Types.hs+1−1 distributed-process/src/Control/Distributed/Process/Node.hs+20−6 distributed-process/tests/TestCH.hs+31−4
distributed-process/src/Control/Distributed/Process.hs+10
@@ -19,6 +19,8 @@ module Control.Distributed.Process
1919 , ReceivePort
2020 , SendPort
2121 , newChan
22+ , sendChan
23+ , receiveChan
2224 -- * Advanced messaging
2325 , Match
2426 , receiveWait
@@ -208,6 +210,14 @@ newChan = do
208210 , (sport, rport)
209211 )
210212
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+
211221--------------------------------------------------------------------------------
212222-- Advanced messaging --
213223--------------------------------------------------------------------------------
distributed-process/src/Control/Distributed/Process/Internal/CQueue.hs+1−1
@@ -51,7 +51,7 @@ dequeue (CQueue arrived incoming) blockSpec matches =
5151 go = mask $ \restore -> do
5252 arr <- takeMVar arrived
5353 -- 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)
5555 (arr', mb) <- onException (restore (checkArrived [] arr))
5656 (putMVar arrived arr)
5757 case (mb, blockSpec) of
distributed-process/src/Control/Distributed/Process/Internal/Types.hs+1−1
@@ -200,7 +200,7 @@ data ChannelId = ChannelId {
200200
201201data TypedChannel = forall a. Serializable a => TypedChannel (CQueue a)
202202
203−newtype SendPort a = SendPort ChannelId
203+newtype SendPort a = SendPort ChannelId deriving (Typeable, Binary)
204204newtype ReceivePort a = ReceivePort (CQueue a)
205205
206206--------------------------------------------------------------------------------
distributed-process/src/Control/Distributed/Process/Node.hs+20−6
@@ -89,6 +89,8 @@ import Control.Distributed.Process.Internal.Types
8989 , runMessageT
9090 , TypedChannel(..)
9191 , Identifier(..)
92+ , ChannelId(..)
93+ , typedChannelWithId
9294 )
9395import Control.Distributed.Process.Serializable (Serializable)
9496import Control.Distributed.Process.Internal.MessageT
@@ -229,12 +231,24 @@ handleIncomingMessages node = go [] Map.empty Map.empty Set.empty
229231 -- we cannot! Network.Transport does not provide this
230232 -- functionality. Not sure what the right approach is.
231233 -- For now, we just drop the incoming messages
232− go (List.delete cid uninitConns)
233− procs
234− chans
235− ctrls
236− ChannelIdentifier _chId ->
237− fail "Connecting to typed channels not yet implemented"
234+ go (List.delete cid uninitConns) procs chans ctrls
235+ ChannelIdentifier chId -> do
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
238252 NodeIdentifier _ ->
239253 go (List.delete cid uninitConns)
240254 procs
distributed-process/tests/TestCH.hs+31−4
@@ -369,15 +369,42 @@ testTimeout0 transport = do
369369
370370 takeMVar clientDone
371371
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+
372398main :: IO ()
373399main = do
374400 Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters
375401 runTests
376− [ ("Ping", testPing transport)
377− , ("Math", testMath transport)
378− , ("Timeout", testTimeout transport)
379− , ("Timeout0", testTimeout0 transport)
402+ [ ("Ping", testPing transport)
403+ , ("Math", testMath transport)
404+ , ("Timeout", testTimeout transport)
405+ , ("Timeout0", testTimeout0 transport)
380406 , ("SendToTerminated", testSendToTerminated transport)
407+ , ("TypedChannnels", testTypedChannels transport)
381408 -- The "missing" combinations in the list below don't make much sense, as
382409 -- we cannot guarantee that the monitor reply or link exception will not
383410 -- happen before the unmonitor or unlink