git.iko.soy/distributed-process
on master help sign in
Ping example now works
d2ec55dEdsko de Vries
distributed-process/src/Control/Distributed/Process.hs+1
compare with master
unifiedsplit
@@ -193,6 +193,7 @@ handleIncomingMessages node = go [] IntMap.empty
193 Just proc -> do
193 Just proc -> do
194 let msg = payloadToMessage payload
194 let msg = payloadToMessage payload
195 enqueue (processQueue proc) msg
195 enqueue (processQueue proc) msg
196+ go halfOpenConns openConns
196 Nothing -> if cid `elem` halfOpenConns
197 Nothing -> if cid `elem` halfOpenConns
197 then do
198 then do
198 let pid = NTI.decodeInt32 . BSS.concat $ payload
199 let pid = NTI.decodeInt32 . BSS.concat $ payload
distributed-process/src/Control/Distributed/Process/Internal/CQueue.hs+4−4
@@ -26,14 +26,14 @@ dequeueMatching (CQueue arrived incoming) matches =
26 modifyMVar arrived (checkArrived [])
26 modifyMVar arrived (checkArrived [])
27 where
27 where
28 checkArrived :: [a] -> [a] -> IO ([a], a)
28 checkArrived :: [a] -> [a] -> IO ([a], a)
29− checkArrived xs' [] = checkIncomming xs'
29+ checkArrived xs' [] = checkIncoming xs'
30 checkArrived xs' (x:xs)
30 checkArrived xs' (x:xs)
31 | matches x = return (reverse xs' ++ xs, x)
31 | matches x = return (reverse xs' ++ xs, x)
32 | otherwise = checkArrived (x:xs') xs
32 | otherwise = checkArrived (x:xs') xs
33
33
34− checkIncomming :: [a] -> IO ([a], a)
34+ checkIncoming :: [a] -> IO ([a], a)
35− checkIncomming xs' = do
35+ checkIncoming xs' = do
36 x <- readChan incoming
36 x <- readChan incoming
37 if matches x
37 if matches x
38 then return (reverse xs', x)
38 then return (reverse xs', x)
39− else checkIncomming (x:xs')
39+ else checkIncoming (x:xs')
distributed-process/tests/TestCH.hs+7−4
@@ -9,7 +9,6 @@ import Control.Concurrent.MVar ( newEmptyMVar
9 , readMVar
9 , readMVar
10 )
10 )
11import Control.Monad (replicateM_)
11import Control.Monad (replicateM_)
12−import Control.Monad.IO.Class (liftIO)
13import Control.Distributed.Process
12import Control.Distributed.Process
14import Network.Transport.TCP (createTransport, defaultTCPParameters)
13import Network.Transport.TCP (createTransport, defaultTCPParameters)
15
14
@@ -44,15 +43,19 @@ testPing = do
44 localNode <- newLocalNode transport
43 localNode <- newLocalNode transport
45 pingServer <- readMVar serverAddr
44 pingServer <- readMVar serverAddr
46
45
46+ let numPings = 10000
47+
47 runProcess localNode $ do
48 runProcess localNode $ do
48 pid <- getSelfPid
49 pid <- getSelfPid
49− replicateM_ 10 $ do
50+ replicateM_ numPings $ do
50 send pingServer (Pong pid)
51 send pingServer (Pong pid)
51− msg@(Ping _) <- expect
52+ Ping _ <- expect
52− liftIO $ print msg
53+ return ()
53
54
55+ putStrLn $ "Client did " ++ show numPings ++ " pings"
54 putMVar clientDone ()
56 putMVar clientDone ()
55
57
58+
56 takeMVar clientDone
59 takeMVar clientDone
57
60
58main :: IO ()
61main :: IO ()