Here's a much simpler implementation for that sort of pattern, using channels to fan out work to threads. I added a dependency on Criterion because getCPUTime is basically useless for this kind of measurement on Mac OS X since it doesn't include the time that the process spent waiting on IO: {-# Language OverloadedStrings #-} import System.CPUTime import Network.Socket hiding(recv) import Network.Socket.ByteString import Control.Exception (handle, IOException) import System.Environment import Control.Concurrent import Control.Monad import Data.Either (partitionEithers) import Data.List (intercalate) import Criterion.Measurement (time, time_, secs) main :: IO () main = do (host, port, conc, reqs) <- fmap parse getArgs putStrLn $ "Connecting to " ++ host ++ " " ++ port (servAddr:_) <- getAddrInfo Nothing (Just host) (Just port) (diff, results) <- time $ process servAddr conc reqs let (errs, succs) = partitionEithers results numSuccs = length succs numErrs = length errs succTime = sum succs succAvg = succTime / fromIntegral numSuccs putStrLn $ unwords [show numSuccs, "successes,", show numErrs, "errors in", secs diff] when (numSuccs > 0) $ do putStrLn $ "min/max/avg request time: " ++ intercalate " / " (map secs [minimum succs, maximum succs, succAvg]) putStrLn $ show (round (fromIntegral reqs / diff) :: Int) ++ " r/s" parse :: [String] -> (String, String, Int, Int) parse [h,p,conc,reqs] = (h, p, read conc, read reqs) parse _ = error "usage client host port concurrency requests" process :: AddrInfo -> Int -> Int -> IO [Either IOException Double] process servAddr conc reqs = do reqChan <- newChan ackChan <- newChan let processThread = forever $ do _ <- readChan reqChan handle (return . Left) (fmap Right socketAction) >>= writeChan ackChan socketAction = time_ $ do sock <- socket (addrFamily servAddr) Stream defaultProtocol connect sock (addrAddress servAddr) sendAll sock "GET /\r\n\r\n" void $ recv sock 1024 close sock replicateM_ reqs (writeChan reqChan ()) replicateM_ (min conc reqs) (forkIO processThread) replicateM reqs (readChan ackChan) On Tue, Jan 7, 2014 at 10:55 AM, Branimir Maksimovic < branimir.maksimovic@gmail.com> wrote:
I have test network client, something like apache bench tool. It uses mvars to synchronize and everything is ok when compiled without -threaded. real 0m2.995s user 0m0.601s sys 0m2.391s
With -threaded compile option result is following: real 0m18.196s user 0m2.054s sys 0m3.313s
Seems that program is sleeping most of the time for some reason. I can't explain behavior as it seems that program is ok. It starts `concurrency` threads which wait on mvar to process next task.
Program follows:
{-# Language OverloadedStrings #-} import System.CPUTime import System.IO --import System.IO.Error import Network.Socket hiding(recv) import Network.Socket.ByteString import System.Environment import Control.Concurrent import Control.Exception
main = do n <- getArgs let (host,port,conc,reqs) = parse n putStrLn $ "Connecting to " ++ host ++ " " ++ port s <- getAddrInfo Nothing (Just host) (Just port) let servAddr = head s begin <- getCPUTime process servAddr conc reqs end <- getCPUTime let diff = (fromIntegral (end - begin))/(10^12) :: Double putStrLn $ show (round (fromIntegral reqs / diff)) ++ " r/s"
parse [h,p,conc,reqs] = (h,p,read conc::Int,read reqs::Int) parse _ = error "usage client host port concurrency requests"
process servAddr conc reqs = do let niter = if reqs >= conc then conc else reqs putStrLn $ "loop " ++ show niter mvars <- initThreads niter [] putStrLn $ "Initialized " ++ show niter let loop n (m:mvs) f | n>0 = do flag <- isEmptyMVar m if f > length mvars then putStrLn "busy" else return () if flag || f > length mvars then do putMVar m () loop (n-1) mvs 0 else loop n mvs (f+1) | otherwise = return () loop n [] f = if n>0 then loop n mvars f else return () putStrLn $ "length " ++ show (length mvars) loop (reqs-niter) mvars 0 where initThreads niter vars | niter > 0 = do mvar <- newMVar () forkIO $ process mvar initThreads (niter-1) (mvar:vars) | otherwise = return vars process mvar = do sock <- socket (addrFamily servAddr) Stream defaultProtocol connect sock (addrAddress servAddr) sendAll sock "Hello World!\n" buf <- recv sock 1024 close sock takeMVar mvar process mvar
_______________________________________________ Haskell-Cafe mailing list Haskell-Cafe@haskell.org http://www.haskell.org/mailman/listinfo/haskell-cafe