a91fb7e · distributed-process

git.iko.soy/distributed-process

Change representation of ReceiveChan

This makes it easier to add additional functionality.

commita91fb7eb955fce406eed02fcc5e70798cdacc4f5 authorEdsko de Vries <edsko@edsko.net> dateTue 6 Nov 2012, 11:29 (14 years ago) parent4189949

4 files changed+39−31

compare with master

unifiedsplit

distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+17−21 distributed-process/src/Control/Distributed/Process/Internal/Types.hs+6−1 distributed-process/tests/TestCH.hs+15−8 distributed-process/tests/TestClosure.hs+1−1

distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+17−21

@@ -89,6 +89,7 @@ import Control.Distributed.Process.Internal.StrictMVar

8989import Control.Concurrent.Chan (writeChan)

9090import Control.Concurrent.STM

9191 ( STM

92+ , TVar

9293 , atomically

9394 , orElse

9495 , newTVar

@@ -181,7 +182,7 @@ newChan = do

181182 let sport = SendPort cid

182183 chan <- liftIO newTQueueIO

183184 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid

184− let rport = ReceivePortSingle chan

185+ let rport = ReceivePort $ readTQueue chan

185186 let tch = TypedChannel chan'

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

187188 . (typedChannelWithId lcid ^= Just tch)

@@ -215,35 +216,30 @@ receiveChanTimeout 0 ch = liftIO . atomically $

215216receiveChanTimeout n ch = liftIO . timeout n . atomically $

216217 receiveSTM ch

217218

218−-- | Receive a message from a typed channel as an STM transaction.

219−--

220−-- The transaction retries when no message is available.

221−receiveSTM :: ReceivePort a -> STM a

222−receiveSTM (ReceivePortSingle c) =

223− readTQueue c

224−receiveSTM (ReceivePortBiased ps) =

225− foldr1 orElse (map receiveSTM ps)

226−receiveSTM (ReceivePortRR psVar) = do

227− ps <- readTVar psVar

228− a <- foldr1 orElse (map receiveSTM ps)

229− writeTVar psVar (rotate ps)

230− return a

231− where

232− rotate :: [a] -> [a]

233− rotate [] = []

234− rotate (x:xs) = xs ++ [x]

235−

236219-- | Merge a list of typed channels.

237220--

238221-- The result port is left-biased: if there are messages available on more

239222-- than one port, the first available message is returned.

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

241−mergePortsBiased = return . ReceivePortBiased

224+mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM

242225

243226-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than

244227-- left-biased)

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

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

229+mergePortsRR = \ps -> do

230+ psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)

231+ return $ ReceivePort (rr psVar)

232+ where

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

234+ rotate [] = []

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

236+

237+ rr :: TVar [STM a] -> STM a

238+ rr psVar = do

239+ ps <- readTVar psVar

240+ a <- foldr1 orElse ps

241+ writeTVar psVar (rotate ps)

242+ return a

247243

248244--------------------------------------------------------------------------------

249245-- Advanced messaging --

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

@@ -82,7 +82,7 @@ import Control.Category ((>>>))

8282import Control.Exception (Exception)

8383import Control.Concurrent (ThreadId)

8484import Control.Concurrent.Chan (Chan)

85−import Control.Concurrent.STM (TVar)

85+import Control.Concurrent.STM (STM, TVar)

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

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

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

@@ -262,6 +262,10 @@ newtype SendPort a = SendPort {

262262 deriving (Typeable, Binary, Show, Eq, Ord)

263263

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

265+newtype ReceivePort a = ReceivePort { receiveSTM :: STM a }

266+ deriving Typeable

267+

268+{-

265269data ReceivePort a =

266270 -- | A single receive port

267271 ReceivePortSingle (TQueue a)

@@ -270,6 +274,7 @@ data ReceivePort a =

270274 -- | A round-robin combination of receive ports

271275 | ReceivePortRR (TVar [ReceivePort a])

272276 deriving Typeable

277+-}

273278

274279--------------------------------------------------------------------------------

275280-- Messages --

distributed-process/tests/TestCH.hs+15−8

@@ -443,29 +443,34 @@ testMergeChannels transport = do

443443 where

444444 -- Single layer of merging

445445 testFlat :: LocalNode -> Bool -> String -> IO ()

446− testFlat localNode biased expected =

447− runProcess localNode $ do

446+ testFlat localNode biased expected = do

447+ done <- newEmptyMVar

448+ forkProcess localNode $ do

448449 rs <- mapM charChannel "abc"

449450 m <- mergePorts biased rs

450451 xs <- replicateM 9 $ receiveChan m

451452 True <- return $ xs == expected

452− return ()

453+ liftIO $ putMVar done ()

454+ takeMVar done

453455

454456 -- Two layers of merging

455457 testNested :: LocalNode -> Bool -> Bool -> String -> IO ()

456− testNested localNode biasedInner biasedOuter expected =

457− runProcess localNode $ do

458+ testNested localNode biasedInner biasedOuter expected = do

459+ done <- newEmptyMVar

460+ forkProcess localNode $ do

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

459462 ms <- mapM (mergePorts biasedInner) rss

460463 m <- mergePorts biasedOuter ms

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

462465 True <- return $ xs == expected

463− return ()

466+ liftIO $ putMVar done ()

467+ takeMVar done

464468

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

466470 testBlocked :: LocalNode -> Bool -> IO ()

467471 testBlocked localNode biased = do

468472 vs <- replicateM 3 newEmptyMVar

473+ done <- newEmptyMVar

469474

470475 forkProcess localNode $ do

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

@@ -496,13 +501,15 @@ testMergeChannels transport = do

496501 , (sa, 'a')

497502 ]

498503

499− runProcess localNode $ do

504+ forkProcess localNode $ do

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

501506 liftIO $ mapM_ (uncurry putMVar) $ zip vs ss

502507 m <- mergePorts biased rs

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

504509 True <- return $ xs == "abcacbbacbcacabcba"

505− return ()

510+ liftIO $ putMVar done ()

511+

512+ takeMVar done

506513

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

508515 mergePorts True = mergePortsBiased

distributed-process/tests/TestClosure.hs+1−1

@@ -31,7 +31,7 @@ import Control.Distributed.Process.Internal.Types (NodeId(nodeAddress))

3131import Control.Distributed.Static (staticLabel, staticClosure)

3232

3333import Test.HUnit (Assertion)

34−import Test.Framework (Test, defaultMain, testGroup)

34+import Test.Framework (Test, defaultMain)

3535import Test.Framework.Providers.HUnit (testCase)

3636

3737--------------------------------------------------------------------------------