a91fb7e · distributed-process

git.iko.soy/distributed-process

on master help sign in

Change representation of ReceiveChan

a91fb7eEdsko de Vries14 years ago

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

compare with master

unifiedsplit

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

89import Control.Concurrent.Chan (writeChan)

90import Control.Concurrent.STM

91 ( STM

92+ , TVar

92 , atomically

93 , atomically

93 , orElse

94 , orElse

94 , newTVar

95 , newTVar

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

181 let sport = SendPort cid

182 let sport = SendPort cid

182 chan <- liftIO newTQueueIO

183 chan <- liftIO newTQueueIO

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

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

184− let rport = ReceivePortSingle chan

185+ let rport = ReceivePort $ readTQueue chan

185 let tch = TypedChannel chan'

186 let tch = TypedChannel chan'

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

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

187 . (typedChannelWithId lcid ^= Just tch)

188 . (typedChannelWithId lcid ^= Just tch)

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

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

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

216 receiveSTM ch

217 receiveSTM ch

217

218

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−

236-- | Merge a list of typed channels.

219-- | Merge a list of typed channels.

237--

220--

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

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

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

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

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

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

241−mergePortsBiased = return . ReceivePortBiased

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

242

225

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

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

244-- left-biased)

227-- left-biased)

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

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

229+mergePortsRR = \ps -> do

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

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

247

243

248--------------------------------------------------------------------------------

244--------------------------------------------------------------------------------

249-- Advanced messaging --

245-- Advanced messaging --

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

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

82import Control.Exception (Exception)

83import Control.Concurrent (ThreadId)

84import Control.Concurrent.Chan (Chan)

85−import Control.Concurrent.STM (TVar)

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

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

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

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

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

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

263

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

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

266+ deriving Typeable

267+

268+{-

265data ReceivePort a =

269data ReceivePort a =

266 -- | A single receive port

270 -- | A single receive port

267 ReceivePortSingle (TQueue a)

271 ReceivePortSingle (TQueue a)

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

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

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

271 | ReceivePortRR (TVar [ReceivePort a])

275 | ReceivePortRR (TVar [ReceivePort a])

272 deriving Typeable

276 deriving Typeable

277+-}

273

278

274--------------------------------------------------------------------------------

279--------------------------------------------------------------------------------

275-- Messages --

280-- Messages --

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

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

443 where

444 -- Single layer of merging

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

446− testFlat localNode biased expected =

446+ testFlat localNode biased expected = do

447+ done <- newEmptyMVar

447− runProcess localNode $ do

448+ forkProcess localNode $ do

448 rs <- mapM charChannel "abc"

449 rs <- mapM charChannel "abc"

449 m <- mergePorts biased rs

450 m <- mergePorts biased rs

450 xs <- replicateM 9 $ receiveChan m

451 xs <- replicateM 9 $ receiveChan m

451 True <- return $ xs == expected

452 True <- return $ xs == expected

452− return ()

453+ liftIO $ putMVar done ()

454+ takeMVar done

453

455

454 -- Two layers of merging

456 -- Two layers of merging

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

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

456− testNested localNode biasedInner biasedOuter expected =

458+ testNested localNode biasedInner biasedOuter expected = do

459+ done <- newEmptyMVar

457− runProcess localNode $ do

460+ forkProcess localNode $ do

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

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

459 ms <- mapM (mergePorts biasedInner) rss

462 ms <- mapM (mergePorts biasedInner) rss

460 m <- mergePorts biasedOuter ms

463 m <- mergePorts biasedOuter ms

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

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

462 True <- return $ xs == expected

465 True <- return $ xs == expected

463− return ()

466+ liftIO $ putMVar done ()

467+ takeMVar done

464

468

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

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

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

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

467 testBlocked localNode biased = do

471 testBlocked localNode biased = do

468 vs <- replicateM 3 newEmptyMVar

472 vs <- replicateM 3 newEmptyMVar

473+ done <- newEmptyMVar

469

474

470 forkProcess localNode $ do

475 forkProcess localNode $ do

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

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

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

496 , (sa, 'a')

501 , (sa, 'a')

497 ]

502 ]

498

503

499− runProcess localNode $ do

504+ forkProcess localNode $ do

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

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

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

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

502 m <- mergePorts biased rs

507 m <- mergePorts biased rs

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

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

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

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

505− return ()

510+ liftIO $ putMVar done ()

511+

512+ takeMVar done

506

513

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

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

508 mergePorts True = mergePortsBiased

515 mergePorts True = mergePortsBiased

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

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

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

33import Test.HUnit (Assertion)

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

34+import Test.Framework (Test, defaultMain)

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

37--------------------------------------------------------------------------------