git.iko.soy/distributed-process
on master help sign in
Implement receiveChanTimeout
8b95ef1Edsko de Vries
distributed-process/src/Control/Distributed/Process.hs+3−1
compare with master
unifiedsplit
@@ -19,6 +19,7 @@ module Control.Distributed.Process
19 -- * Basic messaging
19 -- * Basic messaging
20 , send
20 , send
21 , expect
21 , expect
22+ , expectTimeout
22 -- * Channels
23 -- * Channels
23 , ReceivePort
24 , ReceivePort
24 , SendPort
25 , SendPort
@@ -26,6 +27,7 @@ module Control.Distributed.Process
26 , newChan
27 , newChan
27 , sendChan
28 , sendChan
28 , receiveChan
29 , receiveChan
30+ , receiveChanTimeout
29 , mergePortsBiased
31 , mergePortsBiased
30 , mergePortsRR
32 , mergePortsRR
31 -- * Advanced messaging
33 -- * Advanced messaging
@@ -91,7 +93,6 @@ module Control.Distributed.Process
91 , bracket_
93 , bracket_
92 , finally
94 , finally
93 -- * Auxiliary API
95 -- * Auxiliary API
94− , expectTimeout
95 , spawnAsync
96 , spawnAsync
96 , spawnSupervised
97 , spawnSupervised
97 , spawnLink
98 , spawnLink
@@ -215,6 +216,7 @@ import Control.Distributed.Process.Internal.Primitives
215 , finally
216 , finally
216 -- Auxiliary API
217 -- Auxiliary API
217 , expectTimeout
218 , expectTimeout
219+ , receiveChanTimeout
218 , spawnAsync
220 , spawnAsync
219 -- Reconnecting
221 -- Reconnecting
220 , reconnect
222 , reconnect
distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+26−13
@@ -54,6 +54,7 @@ module Control.Distributed.Process.Internal.Primitives
54 , finally
54 , finally
55 -- * Auxiliary API
55 -- * Auxiliary API
56 , expectTimeout
56 , expectTimeout
57+ , receiveChanTimeout
57 , spawnAsync
58 , spawnAsync
58 , linkNode
59 , linkNode
59 , linkPort
60 , linkPort
@@ -74,6 +75,7 @@ import Data.Binary (decode)
74import Data.Time.Clock (getCurrentTime)
75import Data.Time.Clock (getCurrentTime)
75import Data.Time.Format (formatTime)
76import Data.Time.Format (formatTime)
76import System.Locale (defaultTimeLocale)
77import System.Locale (defaultTimeLocale)
78+import System.Timeout (timeout)
77import Control.Monad.Reader (ask)
79import Control.Monad.Reader (ask)
78import Control.Monad.IO.Class (MonadIO, liftIO)
80import Control.Monad.IO.Class (MonadIO, liftIO)
79import Control.Applicative ((<$>))
81import Control.Applicative ((<$>))
@@ -188,7 +190,7 @@ newChan = do
188 )
190 )
189 where
191 where
190 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
192 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
191− finalizer processState lcid = modifyMVar_ processState $
193+ finalizer st lcid = modifyMVar_ st $
192 return . (typedChannelWithId lcid ^= Nothing)
194 return . (typedChannelWithId lcid ^= Nothing)
193
195
194-- | Send a message on a typed channel
196-- | Send a message on a typed channel
@@ -204,18 +206,29 @@ sendChan (SendPort cid) msg = do
204-- | Wait for a message on a typed channel
206-- | Wait for a message on a typed channel
205receiveChan :: Serializable a => ReceivePort a -> Process a
207receiveChan :: Serializable a => ReceivePort a -> Process a
206receiveChan = liftIO . atomically . receiveSTM
208receiveChan = liftIO . atomically . receiveSTM
207− where
208− receiveSTM :: ReceivePort a -> STM a
209− receiveSTM (ReceivePortSingle c) =
210− readTQueue c
211− receiveSTM (ReceivePortBiased ps) =
212− foldr1 orElse (map receiveSTM ps)
213− receiveSTM (ReceivePortRR psVar) = do
214− ps <- readTVar psVar
215− a <- foldr1 orElse (map receiveSTM ps)
216− writeTVar psVar (rotate ps)
217− return a
218
209
210+-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
211+-- non-blocking check for a message.
212+receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
213+receiveChanTimeout 0 ch = liftIO . atomically $
214+ (Just <$> receiveSTM ch) `orElse` return Nothing
215+receiveChanTimeout n ch = liftIO . timeout n . atomically $
216+ receiveSTM ch
217+
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
219 rotate :: [a] -> [a]
232 rotate :: [a] -> [a]
220 rotate [] = []
233 rotate [] = []
221 rotate (x:xs) = xs ++ [x]
234 rotate (x:xs) = xs ++ [x]
@@ -459,7 +472,7 @@ finally a sequel = bracket_ (return ()) sequel a
459
472
460-- | Like 'expect' but with a timeout
473-- | Like 'expect' but with a timeout
461expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
474expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
462−expectTimeout timeout = receiveTimeout timeout [match return]
475+expectTimeout n = receiveTimeout n [match return]
463
476
464-- | Asynchronous version of 'spawn'
477-- | Asynchronous version of 'spawn'
465--
478--
distributed-process/tests/TestCH.hs+43
@@ -695,6 +695,47 @@ testMatchAny transport = do
695
695
696 takeMVar clientDone
696 takeMVar clientDone
697
697
698+-- Test 'receiveChanTimeout'
699+testReceiveChanTimeout :: NT.Transport -> IO ()
700+testReceiveChanTimeout transport = do
701+ done <- newEmptyMVar
702+ sendPort <- newEmptyMVar
703+
704+ forkIO $ do
705+ localNode <- newLocalNode transport initRemoteTable
706+ runProcess localNode $ do
707+ -- Create a typed channel
708+ (sp, rp) <- newChan :: Process (SendPort Bool, ReceivePort Bool)
709+ liftIO $ putMVar sendPort sp
710+
711+ -- Wait for a message with a delay. No message arrives, we should get Nothing after 1 second
712+ Nothing <- receiveChanTimeout 1000000 rp
713+
714+ -- Wait for a message with a delay again. Now a message arrives after 0.5 seconds
715+ Just True <- receiveChanTimeout 1000000 rp
716+
717+ -- Wait for a message with zero timeout: non-blocking check. No message is available, we get Nothing
718+ Nothing <- receiveChanTimeout 0 rp
719+
720+ -- Again, but now there is a message available
721+ liftIO $ threadDelay 1000000
722+ Just False <- receiveChanTimeout 0 rp
723+
724+ liftIO $ putMVar done ()
725+
726+ forkIO $ do
727+ localNode <- newLocalNode transport initRemoteTable
728+ runProcess localNode $ do
729+ sp <- liftIO $ readMVar sendPort
730+
731+ liftIO $ threadDelay 1500000
732+ sendChan sp True
733+
734+ liftIO $ threadDelay 500000
735+ sendChan sp False
736+
737+ takeMVar done
738+
698main :: IO ()
739main :: IO ()
699main = do
740main = do
700 Right (transport, transportInternals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters
741 Right (transport, transportInternals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters
@@ -739,4 +780,6 @@ main = do
739 , ("MonitorChannel", testMonitorChannel transport)
780 , ("MonitorChannel", testMonitorChannel transport)
740 -- Reconnect
781 -- Reconnect
741 , ("Reconnect", testReconnect transport transportInternals)
782 , ("Reconnect", testReconnect transport transportInternals)
783+ -- ReceiveChanTimeout
784+ , ("ReceiveChanTimeout", testReceiveChanTimeout transport)
742 ]
785 ]