c5fff42 · distributed-process

git.iko.soy/distributed-process

Implement merging typed channels

commitc5fff428981ebc8952c9425a601ace45fb50ede9 authorEdsko de Vries <edsko@edsko.net> dateThu 21 Jun 2012, 15:27 (14 years ago) parentfad9b46

4 files changed+149−22

compare with master

unifiedsplit

distributed-process/src/Control/Distributed/Process.hs+38−11 distributed-process/src/Control/Distributed/Process/Internal/Types.hs+17−7 distributed-process/src/Control/Distributed/Process/Node.hs+3−2 distributed-process/tests/TestCH.hs+91−2

distributed-process/src/Control/Distributed/Process.hs+38−11

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

2121 , newChan

2222 , sendChan

2323 , receiveChan

24+ , mergePortsBiased

25+ , mergePortsRR

2426 -- * Advanced messaging

2527 , Match

2628 , receiveWait

@@ -68,11 +70,17 @@ import Control.Exception (Exception, throw)

6870import qualified Control.Exception as Exception (catch)

6971import Control.Concurrent.MVar (modifyMVar)

7072import Control.Concurrent.Chan (writeChan)

71−import Control.Distributed.Process.Internal.CQueue

72− ( newCQueue

73− , dequeue

74− , BlockSpec(..)

73+import Control.Concurrent.STM

74+ ( STM

75+ , atomically

76+ , orElse

77+ , newTChan

78+ , readTChan

79+ , newTVar

80+ , readTVar

81+ , writeTVar

7582 )

83+import Control.Distributed.Process.Internal.CQueue (dequeue, BlockSpec(..))

7684import Control.Distributed.Process.Serializable (Serializable, fingerprint)

7785import Data.Accessor ((^.), (^:), (^=))

7886import Control.Distributed.Process.Internal.Types

@@ -191,19 +199,19 @@ expect = receiveWait [match return]

191199-- Channels --

192200--------------------------------------------------------------------------------

193201

194−-- | Create a new typed channel (WORK IN PROGRESS)

202+-- | Create a new typed channel

195203newChan :: Serializable a => Process (SendPort a, ReceivePort a)

196204newChan = do

197205 proc <- ask

198206 liftIO . modifyMVar (processState proc) $ \st -> do

199− queue <- liftIO newCQueue

207+ chan <- liftIO . atomically $ newTChan

200208 let lcid = st ^. channelCounter

201209 cid = ChannelId { channelProcessId = processId proc

202210 , channelLocalId = lcid

203211 }

204212 sport = SendPort cid

205− rport = ReceivePort queue

206− tch = TypedChannel queue

213+ rport = ReceivePortSingle chan

214+ tch = TypedChannel chan

207215 return ( (channelCounter ^: (+ 1))

208216 . (typedChannelWithId lcid ^= Just tch)

209217 $ st

@@ -214,9 +222,28 @@ sendChan :: Serializable a => SendPort a -> a -> Process ()

214222sendChan (SendPort them) msg = lift $ sendBinary (ChannelIdentifier them) msg

215223

216224receiveChan :: Serializable a => ReceivePort a -> Process a

217−receiveChan (ReceivePort queue) = do

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

219− return msg

225+receiveChan = liftIO . atomically . receiveSTM

226+ where

227+ receiveSTM :: ReceivePort a -> STM a

228+ receiveSTM (ReceivePortSingle c) =

229+ readTChan c

230+ receiveSTM (ReceivePortBiased ps) =

231+ foldr1 orElse (map receiveSTM ps)

232+ receiveSTM (ReceivePortRR psVar) = do

233+ ps <- readTVar psVar

234+ a <- foldr1 orElse (map receiveSTM ps)

235+ writeTVar psVar (rotate ps)

236+ return a

237+

238+ rotate :: [a] -> [a]

239+ rotate [] = []

240+ rotate (x:xs) = xs ++ [x]

241+

242+mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)

243+mergePortsBiased = return . ReceivePortBiased

244+

245+mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)

246+mergePortsRR ps = liftIO . atomically $ ReceivePortRR <$> newTVar ps

220247

221248--------------------------------------------------------------------------------

222249-- Advanced messaging --

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

@@ -59,11 +59,6 @@ module Control.Distributed.Process.Internal.Types

5959 , runMessageT

6060 ) where

6161

62−import Control.Concurrent (ThreadId)

63−import Control.Concurrent.MVar (MVar)

64−import Control.Concurrent.Chan (Chan)

65−import Control.Category ((>>>))

66−import Control.Exception (Exception)

6762import Data.Map (Map)

6863import qualified Data.Map as Map (empty)

6964import Data.Int (Int32)

@@ -75,6 +70,12 @@ import Data.Accessor (Accessor, accessor)

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

7671import qualified Data.ByteString.Lazy as BSL (ByteString, toChunks, fromChunks, splitAt)

7772import qualified Data.ByteString as BSS (ByteString, concat)

73+import Control.Category ((>>>))

74+import Control.Exception (Exception)

75+import Control.Concurrent (ThreadId)

76+import Control.Concurrent.MVar (MVar)

77+import Control.Concurrent.Chan (Chan)

78+import Control.Concurrent.STM (TChan, TVar)

7879import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)

7980import Control.Applicative ((<$>), (<*>))

8081import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)

@@ -198,10 +199,19 @@ data ChannelId = ChannelId {

198199 }

199200 deriving (Show, Eq, Ord)

200201

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

202+data TypedChannel = forall a. Serializable a => TypedChannel (TChan a)

202203

204+-- | The send send of a typed channel (serializable)

203205newtype SendPort a = SendPort ChannelId deriving (Typeable, Binary)

204−newtype ReceivePort a = ReceivePort (CQueue a)

206+

207+-- | The receive end of a typed channel (not serializable)

208+data ReceivePort a =

209+ -- | A single receive port

210+ ReceivePortSingle (TChan a)

211+ -- | A left-biased combination of receive ports

212+ | ReceivePortBiased [ReceivePort a]

213+ -- | A round-robin combination of receive ports

214+ | ReceivePortRR (TVar [ReceivePort a])

205215

206216--------------------------------------------------------------------------------

207217-- Closures --

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

@@ -40,6 +40,7 @@ import Control.Concurrent.MVar

4040 , takeMVar

4141 )

4242import Control.Concurrent.Chan (newChan, writeChan, readChan)

43+import Control.Concurrent.STM (atomically, writeTChan)

4344import Control.Distributed.Process.Internal.CQueue (enqueue, newCQueue)

4445import qualified Network.Transport as NT

4546 ( Transport

@@ -208,8 +209,8 @@ handleIncomingMessages node = go [] Map.empty Map.empty Set.empty

208209 let msg = payloadToMessage payload

209210 enqueue (processQueue proc) msg

210211 go uninitConns procs chans ctrls

211− (_, Just (TypedChannel queue), _, _) -> do

212− enqueue queue . decode . BSL.fromChunks $ payload

212+ (_, Just (TypedChannel chan), _, _) -> do

213+ atomically $ writeTChan chan . decode . BSL.fromChunks $ payload

213214 go uninitConns procs chans ctrls

214215 (_, _, True, _) -> do

215216 writeChan ctrlChan (decode . BSL.fromChunks $ payload)

distributed-process/tests/TestCH.hs+91−2

@@ -12,7 +12,7 @@ import Control.Concurrent.MVar

1212 , takeMVar

1313 , readMVar

1414 )

15−import Control.Monad (replicateM_)

15+import Control.Monad (replicateM_, replicateM)

1616import Control.Monad.IO.Class (liftIO)

1717import Control.Exception (throwIO)

1818import Control.Applicative ((<$>), (<*>))

@@ -21,6 +21,7 @@ import Network.Transport.TCP (createTransport, defaultTCPParameters)

2121import Control.Distributed.Process

2222import Control.Distributed.Process.Internal.Types (LocalNode(localEndPoint))

2323import Control.Distributed.Process.Node

24+import Control.Distributed.Process.Serializable (Serializable)

2425import TestAuxiliary

2526

2627newtype Ping = Ping ProcessId

@@ -395,6 +396,93 @@ testTypedChannels transport = do

395396

396397 takeMVar clientDone

397398

399+-- | Test merging receive ports

400+testMergeChannels :: NT.Transport -> IO ()

401+testMergeChannels transport = do

402+ localNode <- newLocalNode transport initRemoteTable

403+ testFlat localNode True "aaabbbccc"

404+ testFlat localNode False "abcabcabc"

405+ testNested localNode True True "aaabbbcccdddeeefffggghhhiii"

406+ testNested localNode True False "adgadgadgbehbehbehcficficfi"

407+ testNested localNode False True "abcabcabcdefdefdefghighighi"

408+ testNested localNode False False "adgbehcfiadgbehcfiadgbehcfi"

409+ testBlocked localNode True

410+ testBlocked localNode False

411+ where

412+ -- Single layer of merging

413+ testFlat :: LocalNode -> Bool -> String -> IO ()

414+ testFlat localNode biased expected =

415+ runProcess localNode $ do

416+ rs <- mapM charChannel "abc"

417+ m <- mergePorts biased rs

418+ xs <- replicateM 9 $ receiveChan m

419+ True <- return $ xs == expected

420+ return ()

421+

422+ -- Two layers of merging

423+ testNested :: LocalNode -> Bool -> Bool -> String -> IO ()

424+ testNested localNode biasedInner biasedOuter expected =

425+ runProcess localNode $ do

426+ rss <- mapM (mapM charChannel) ["abc", "def", "ghi"]

427+ ms <- mapM (mergePorts biasedInner) rss

428+ m <- mergePorts biasedOuter ms

429+ xs <- replicateM (9 * 3) $ receiveChan m

430+ True <- return $ xs == expected

431+ return ()

432+

433+ -- Test that if no messages are (immediately) available, the scheduler makes no difference

434+ testBlocked :: LocalNode -> Bool -> IO ()

435+ testBlocked localNode biased = do

436+ vs <- replicateM 3 newEmptyMVar

437+

438+ forkProcess localNode $ do

439+ [sa, sb, sc] <- liftIO $ mapM readMVar vs

440+ mapM_ ((>> liftIO (threadDelay 10000)) . uncurry sendChan)

441+ [ -- a, b, c

442+ (sa, 'a')

443+ , (sb, 'b')

444+ , (sc, 'c')

445+ -- a, c, b

446+ , (sa, 'a')

447+ , (sc, 'c')

448+ , (sb, 'b')

449+ -- b, a, c

450+ , (sb, 'b')

451+ , (sa, 'a')

452+ , (sc, 'c')

453+ -- b, c, a

454+ , (sb, 'b')

455+ , (sc, 'c')

456+ , (sa, 'a')

457+ -- c, a, b

458+ , (sc, 'c')

459+ , (sa, 'a')

460+ , (sb, 'b')

461+ -- c, b, a

462+ , (sc, 'c')

463+ , (sb, 'b')

464+ , (sa, 'a')

465+ ]

466+

467+ runProcess localNode $ do

468+ (ss, rs) <- unzip <$> replicateM 3 newChan

469+ liftIO $ mapM_ (uncurry putMVar) $ zip vs ss

470+ m <- mergePorts biased rs

471+ xs <- replicateM (6 * 3) $ receiveChan m

472+ True <- return $ xs == "abcacbbacbcacabcba"

473+ return ()

474+

475+ mergePorts :: Serializable a => Bool -> [ReceivePort a] -> Process (ReceivePort a)

476+ mergePorts True = mergePortsBiased

477+ mergePorts False = mergePortsRR

478+

479+ charChannel :: Char -> Process (ReceivePort Char)

480+ charChannel c = do

481+ (sport, rport) <- newChan

482+ replicateM_ 3 $ sendChan sport c

483+ liftIO $ threadDelay 10000 -- Make sure messages have been sent

484+ return rport

485+

398486main :: IO ()

399487main = do

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

@@ -403,8 +491,9 @@ main = do

403491 , ("Math", testMath transport)

404492 , ("Timeout", testTimeout transport)

405493 , ("Timeout0", testTimeout0 transport)

406− , ("SendToTerminated", testSendToTerminated transport)

494+ , ("SendToTerminated", testSendToTerminated transport)

407495 , ("TypedChannnels", testTypedChannels transport)

496+ , ("MergeChannels", testMergeChannels transport)

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

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

410499 -- happen before the unmonitor or unlink