unagi-chan 0.1.0.2 → 0.1.1.0
raw patch · 8 files changed
+100/−146 lines, 8 filesdep ~atomic-primopsPVP ok
version bump matches the API change (PVP)
Dependency ranges changed: atomic-primops
API changes (from Hackage documentation)
Files
- benchmarks/multi.hs +8/−8
- benchmarks/single.hs +6/−6
- core-example/Main.hs +10/−55
- src/Control/Concurrent/Chan/Unagi.hs +1/−1
- src/Control/Concurrent/Chan/Unagi/Constants.hs +48/−0
- src/Control/Concurrent/Chan/Unagi/Internal.hs +4/−44
- src/Control/Concurrent/Chan/Unagi/Unboxed/Internal.hs +4/−22
- unagi-chan.cabal +19/−10
benchmarks/multi.hs view
@@ -49,23 +49,23 @@ -- this gives us a measure of effects of contention between -- readers and writers when compared with single-threaded -- version:- [ bench "async 1 writers 1 readers" $ asyncReadsWritesUnagi 1 1 n+ [ bench "async 1 writers 1 readers" $ nfIO $ asyncReadsWritesUnagi 1 1 n -- This is measuring the effects of bottlenecks caused by -- descheduling, context-switching overhead (forced by -- fairness properties in the case of MVar), as well as -- all of the above; this is probably less than -- informative. Try threadscope on a standalone test:- , bench "oversubscribing: async 100 writers 100 readers" $ asyncReadsWritesUnagi 100 100 n+ , bench "oversubscribing: async 100 writers 100 readers" $ nfIO $ asyncReadsWritesUnagi 100 100 n -- NOTE: this is a bit hackish, filling in one test and -- reading in the other; make sure memory usage isn't -- influencing mean: -- This measures writer/writer contention:- , bench ("async "++(show procs)++" writers") $ do+ , bench ("async "++(show procs)++" writers") $ nfIO $ do dones <- replicateM procs newEmptyMVar ; starts <- replicateM procs newEmptyMVar mapM_ (\(start1,done1)-> forkIO $ takeMVar start1 >> replicateM_ (n `div` procs) (U.writeChan fill_empty_fastUI ()) >> putMVar done1 ()) $ zip starts dones mapM_ (\v-> putMVar v ()) starts ; mapM_ (\v-> takeMVar v) dones -- This measures reader/reader contention:- , bench ("async "++(show procs)++" readers") $ do+ , bench ("async "++(show procs)++" readers") $ nfIO $ do dones <- replicateM procs newEmptyMVar ; starts <- replicateM procs newEmptyMVar mapM_ (\(start1,done1)-> forkIO $ takeMVar start1 >> replicateM_ (n `div` procs) (U.readChan fill_empty_fastUO) >> putMVar done1 ()) $ zip starts dones mapM_ (\v-> putMVar v ()) starts ; mapM_ (\v-> takeMVar v) dones@@ -73,14 +73,14 @@ , bench "async Int writer, main thread read and sum" $ nfIO $ asyncSumIntUnagi n ] , bgroup "unagi-chan Unagi.Unboxed" $- [ bench "async 1 writers 1 readers" $ asyncReadsWritesUnagiUnboxed 1 1 n- , bench "oversubscribing: async 100 writers 100 readers" $ asyncReadsWritesUnagiUnboxed 100 100 n+ [ bench "async 1 writers 1 readers" $ nfIO $ asyncReadsWritesUnagiUnboxed 1 1 n+ , bench "oversubscribing: async 100 writers 100 readers" $ nfIO $ asyncReadsWritesUnagiUnboxed 100 100 n -- TODO using Ints here instead of (); change others so we can properly compare?- , bench ("async "++(show procs)++" writers") $ do+ , bench ("async "++(show procs)++" writers") $ nfIO $ do dones <- replicateM procs newEmptyMVar ; starts <- replicateM procs newEmptyMVar mapM_ (\(start1,done1)-> forkIO $ takeMVar start1 >> replicateM_ (n `div` procs) (UU.writeChan fill_empty_fastUUI (0::Int)) >> putMVar done1 ()) $ zip starts dones mapM_ (\v-> putMVar v ()) starts ; mapM_ (\v-> takeMVar v) dones- , bench ("async "++(show procs)++" readers") $ do+ , bench ("async "++(show procs)++" readers") $ nfIO $ do dones <- replicateM procs newEmptyMVar ; starts <- replicateM procs newEmptyMVar mapM_ (\(start1,done1)-> forkIO $ takeMVar start1 >> replicateM_ (n `div` procs) (UU.readChan fill_empty_fastUUO) >> putMVar done1 ()) $ zip starts dones mapM_ (\v-> putMVar v ()) starts ; mapM_ (\v-> takeMVar v) dones
benchmarks/single.hs view
@@ -36,8 +36,8 @@ -- Very artificial; just adding up the costs of the takes/puts/reads -- involved in getting a single message in and out [ bgroup "Latency micro-benchmark" $- [ bench "unagi-chan Unagi" (U.writeChan fastEmptyUI () >> U.readChan fastEmptyUO)- , bench "unagi-chan Unagi.Unboxed" (UU.writeChan fastEmptyUUI (0::Int) >> UU.readChan fastEmptyUUO) -- TODO comparing Int writing to (). Change?+ [ bench "unagi-chan Unagi" $ nfIO (U.writeChan fastEmptyUI () >> U.readChan fastEmptyUO)+ , bench "unagi-chan Unagi.Unboxed" $ nfIO (UU.writeChan fastEmptyUUI (0::Int) >> UU.readChan fastEmptyUUO) -- TODO comparing Int writing to (). Change? #ifdef COMPARE_BENCHMARKS , bench "Chan" (writeChan chanEmpty () >> readChan chanEmpty) , bench "TQueue" (atomically (writeTQueue tqueueEmpty () >> readTQueue tqueueEmpty))@@ -51,8 +51,8 @@ ] , bgroup ("Throughput with "++show n++" messages") $ [ bgroup "sequential write all then read all" $- [ bench "unagi-chan Unagi" $ runtestSplitChanU1 n- , bench "unagi-chan Unagi.Unboxed" $ runtestSplitChanUU1 n+ [ bench "unagi-chan Unagi" $ nfIO $ runtestSplitChanU1 n+ , bench "unagi-chan Unagi.Unboxed" $ nfIO $ runtestSplitChanUU1 n #ifdef COMPARE_BENCHMARKS , bench "Chan" $ runtestChan1 n , bench "TQueue" $ runtestTQueue1 n@@ -61,8 +61,8 @@ #endif ] , bgroup "repeated write some, read some" $ - [ bench "unagi-chan Unagi" $ runtestSplitChanU2 n- , bench "unagi-chan Unagi.Unboxed" $ runtestSplitChanUU2 n+ [ bench "unagi-chan Unagi" $ nfIO $ runtestSplitChanU2 n+ , bench "unagi-chan Unagi.Unboxed" $ nfIO $ runtestSplitChanUU2 n #ifdef COMPARE_BENCHMARKS , bench "Chan" $ runtestChan2 n , bench "TQueue" $ runtestTQueue2 n
core-example/Main.hs view
@@ -4,9 +4,8 @@ import System.Environment import Control.Concurrent.MVar import Control.Concurrent-import qualified Control.Concurrent.Chan.Split as F import qualified Control.Concurrent.Chan.Unagi as U-+import qualified Control.Concurrent.Chan.Unagi.Unboxed as UU import qualified Control.Concurrent.Chan as C import qualified Control.Concurrent.STM.TQueue as S import Control.Concurrent.STM@@ -23,39 +22,16 @@ _ -> (100,100) putStrLn $ "Running with "++show r++" readers, and "++ show w++" writers." case nm of- "fast" -> runF w r n "unagi" -> runU w r n "chan" -> runC w r n "stm" -> runS w r n -}-main = do- [nm,n] <- getArgs- case nm of- "fast" -> runF (read n)- "unagi" -> runU (read n) -{---- NOTE compare memory usage to Chan; very nice! (TODO without profiling enabled, looking at +RTS -s) main = do- (i,o) <- U.newChan- let procs = 2- n = 100000 * 100- replicateM_ n (U.writeChan i ())- {-- dones <- replicateM procs newEmptyMVar ; starts <- replicateM procs newEmptyMVar- mapM_ (\(start1,done1)-> forkIO $ takeMVar start1 >> replicateM_ (n `div` procs) (U.writeChan i ()) >> putMVar done1 ()) $ zip starts dones- mapM_ (\v-> putMVar v ()) starts ; mapM_ (\v-> takeMVar v) dones - -}--}+ [n] <- getArgs+ -- runU (read n)+ runUU (read n) {--Notes:- stm:- throws OOM on 100x100- fast: - memory profiles are all over the map between runs- best profile was when with Running with 1 readers, and 100 writers !!--}- runU :: Int -> IO () runU n = do (i,o) <- U.newChan@@ -63,37 +39,16 @@ replicateM_ 1000 $ do replicateM_ n1000 $ U.writeChan i () replicateM_ n1000 $ U.readChan o--runF :: Int -> IO ()-runF n = do- (i,o) <- F.newChan+ -}+runUU :: Int -> IO ()+runUU n = do+ (i,o) <- UU.newChan let n1000 = n `quot` 1000 replicateM_ 1000 $ do- replicateM_ n1000 $ F.writeChan i ()- replicateM_ n1000 $ F.readChan o+ replicateM_ n1000 $ UU.writeChan i (0::Int)+ replicateM_ n1000 $ UU.readChan o {---- TODO fix this up with CPP for cleaner core-runF :: Int -> Int -> Int -> IO ()-runF writers readers n = do- let nNice = n - rem n (lcm writers readers)- perReader = nNice `quot` readers- perWriter = (nNice `quot` writers)- vs <- replicateM readers newEmptyMVar- (i,o) <- F.newChan- let doRead = replicateM_ perReader $ theRead- theRead = F.readChan o- doWrite = replicateM_ perWriter $ theWrite- theWrite = F.writeChan i (1 :: Int)- mapM_ (\v-> forkIO (traceEventIO "READER START" >> doRead >> putMVar v ())) vs-- wWaits <- replicateM writers newEmptyMVar- mapM_ (\v-> forkIO $ (takeMVar v >> traceEventIO "WRITER START" >> doWrite)) wWaits- mapM_ (\v-> putMVar v ()) wWaits-- mapM_ takeMVar vs -- await readers-- runU :: Int -> Int -> Int -> IO () runU writers readers n = do let nNice = n - rem n (lcm writers readers)
src/Control/Concurrent/Chan/Unagi.hs view
@@ -1,7 +1,7 @@ module Control.Concurrent.Chan.Unagi ( {- | General-purpose concurrent FIFO queue. If you are trying to send messages of a primitive unboxed type, you may wish to use "Control.Concurrent.Chan.Unagi.Unboxed"- which should be slightly faster and perform better when a queue frows very large.+ which should be slightly faster and perform better when a queue grows very large. -} -- * Creating channels newChan
+ src/Control/Concurrent/Chan/Unagi/Constants.hs view
@@ -0,0 +1,48 @@+module Control.Concurrent.Chan.Unagi.Constants + where++-- Constants for boxed and unboxed unagi.++import Data.Bits+import Control.Exception(assert)++divMod_sEGMENT_LENGTH :: Int -> (Int,Int)+{-# INLINE divMod_sEGMENT_LENGTH #-}+divMod_sEGMENT_LENGTH n = let d = n `unsafeShiftR` lOG_SEGMENT_LENGTH+ m = n .&. sEGMENT_LENGTH_MN_1+ in d `seq` m `seq` (d,m)++-- Constant for now: back-of-envelope considerations:+-- - making most of constant factor for cloning array of *any* size+-- - make most of overheads of moving to the next segment, etc.+-- - provide enough runway for creating next segment when 32 simultaneous writers +-- - the larger this the larger one-time cost for the lucky writer+-- - as arrays collect in heap, performance might suffer, so bigger arrays+-- give us a constant factor edge there. see:+-- http://stackoverflow.com/q/23462004/176841+--+sEGMENT_LENGTH :: Int+{-# INLINE sEGMENT_LENGTH #-}+sEGMENT_LENGTH = 1024 -- NOTE: THIS MUST REMAIN A POWER OF 2!++-- Number of reads on which to spin for new segment creation.+-- Back-of-envelope (time_to_create_new_segment / time_for_read_IOref) + margin.+-- See usage site.+--+-- NOTE: this was calculated for boxed Unagi, but it probably doesn't make a+-- measurable difference that we use it for Unagi.Unboxed too.+nEW_SEGMENT_WAIT :: Int+nEW_SEGMENT_WAIT = round (((14.6::Float) + 0.3*fromIntegral sEGMENT_LENGTH) / 3.7) + 10++-- TODO move these into a Constants INLINABLE file, +-- use in Unboxed as well+-- verify by running a benchmark on consts3++lOG_SEGMENT_LENGTH :: Int+lOG_SEGMENT_LENGTH = + let x = 10 -- ...pre-computed from...+ in assert (x == (round $ logBase (2::Float) $ fromIntegral sEGMENT_LENGTH))+ x++sEGMENT_LENGTH_MN_1 :: Int+sEGMENT_LENGTH_MN_1 = sEGMENT_LENGTH - 1
src/Control/Concurrent/Chan/Unagi/Internal.hs view
@@ -26,16 +26,14 @@ import Data.Typeable(Typeable) import GHC.Exts(inline) +import Control.Concurrent.Chan.Unagi.Constants --- Number of reads on which to spin for new segment creation.--- Back-of-envelope (time_to_create_new_segment / time_for_read_IOref) + margin.--- See usage site.-nEW_SEGMENT_WAIT :: Int-nEW_SEGMENT_WAIT = round (((14.6::Float) + 0.3*fromIntegral sEGMENT_LENGTH) / 3.7) + 10-+-- | The write end of a channel created with 'newChan'. data InChan a = InChan !(Ticket (Cell a)) !(ChanEnd a) deriving Typeable++-- | The read end of a channel created with 'newChan'. newtype OutChan a = OutChan (ChanEnd a) deriving Typeable @@ -80,19 +78,6 @@ data Cell a = Empty | Written a | Blocking !(MVar a) --- Constant for now: back-of-envelope considerations:--- - making most of constant factor for cloning array of *any* size--- - make most of overheads of moving to the next segment, etc.--- - provide enough runway for creating next segment when 32 simultaneous writers --- - the larger this the larger one-time cost for the lucky writer--- - as arrays collect in heap, performance might suffer, so bigger arrays--- give us a constant factor edge there. see:--- http://stackoverflow.com/q/23462004/176841----sEGMENT_LENGTH :: Int-{-# INLINE sEGMENT_LENGTH #-}-sEGMENT_LENGTH = 1024 -- NOTE: THIS MUST REMAIN A POWER OF 2!- -- NOTE In general we'll have two segments allocated at any given time in -- addition to the segment template, so in the worst case, when the program -- exits we will have allocated ~ 3 segments extra memory than was actually@@ -314,28 +299,3 @@ -- Readers blocked indefinitely should eventually raise a -- BlockedIndefinitelyOnMVar. -- ---------------lOG_SEGMENT_LENGTH, sEGMENT_LENGTH_MN_1 :: Int-lOG_SEGMENT_LENGTH = round $ logBase (2::Float) $ fromIntegral sEGMENT_LENGTH -- or bit shifts in loop-sEGMENT_LENGTH_MN_1 = sEGMENT_LENGTH - 1--divMod_sEGMENT_LENGTH :: Int -> (Int,Int)-{-# INLINE divMod_sEGMENT_LENGTH #-}-divMod_sEGMENT_LENGTH n = let d = n `unsafeShiftR` lOG_SEGMENT_LENGTH- m = n .&. sEGMENT_LENGTH_MN_1- in d `seq` m `seq` (d,m)--{- TESTS SKETCH- - validate with some benchmarks- - look over implementation for other assertions / micro-tests- - Make sure we never get False returned on casIORef where we no no conflicts, i.e. no false negatives- - also include arbitrary delays between readForCAS and the CAS- - (Not a test, but...) add a branch with a whole load of event logging that we can analyze (maybe in an automated test!)- - perhaps run num_threads readers and writers, plus some threads that can inspect the queues- for bad descheduling conditions we want to avoid.- - use quickcheck to generate 'new' chans that represent possible conditions and test those with write/read (or with our regular test suite?)- - we also want to test counter roll-over- -}
src/Control/Concurrent/Chan/Unagi/Unboxed/Internal.hs view
@@ -37,15 +37,13 @@ import GHC.Exts(inline) import Utilities ---- Number of reads on which to spin for new segment creation.--- Back-of-envelope (time_to_create_new_segment / time_for_read_IOref) + margin.--- See usage site.-nEW_SEGMENT_WAIT :: Int-nEW_SEGMENT_WAIT = round (((14.6::Float) + 0.3*fromIntegral sEGMENT_LENGTH) / 3.7) + 10+import Control.Concurrent.Chan.Unagi.Constants +-- | The write end of a channel created with 'newChan'. newtype InChan a = InChan (ChanEnd a) deriving Typeable++-- | The read end of a channel created with 'newChan'. newtype OutChan a = OutChan (ChanEnd a) deriving Typeable @@ -126,11 +124,6 @@ return (sigArr, ElementArray eArr) -sEGMENT_LENGTH :: Int-{-# INLINE sEGMENT_LENGTH #-}-sEGMENT_LENGTH = 1024 -- NOTE: THIS MUST REMAIN A POWER OF 2!-- data Stream a = Stream !SignalIntArray !(ElementArray a)@@ -324,14 +317,3 @@ Next strNext -> return strNext _ -> error "Impossible! This should only have been Next segment" Next strNext -> return strNext---lOG_SEGMENT_LENGTH, sEGMENT_LENGTH_MN_1 :: Int-lOG_SEGMENT_LENGTH = round $ logBase (2::Float) $ fromIntegral sEGMENT_LENGTH -- or bit shifts in loop-sEGMENT_LENGTH_MN_1 = sEGMENT_LENGTH - 1--divMod_sEGMENT_LENGTH :: Int -> (Int,Int)-{-# INLINE divMod_sEGMENT_LENGTH #-}-divMod_sEGMENT_LENGTH n = let d = n `unsafeShiftR` lOG_SEGMENT_LENGTH- m = n .&. sEGMENT_LENGTH_MN_1- in d `seq` m `seq` (d,m)
unagi-chan.cabal view
@@ -1,5 +1,5 @@ name: unagi-chan-version: 0.1.0.2+version: 0.1.1.0 synopsis: Fast and scalable concurrent queues for x86, with a Chan-like API @@ -10,10 +10,10 @@ limited usefulness outside of x86 architectures where the fetch-and-add instruction is not available. .- Here is an example benchmark measuring the time taken to write and then- read 100,000 messages, with work divided amongst increasing number of- readers and writers (time in ms), comparing against the top-performing- queues in the standard libraries.+ Here is an example benchmark measuring the time taken to concurrently write+ and read 100,000 messages, with work divided amongst increasing number of+ readers and writers, comparing against the top-performing queues in the+ standard libraries. Scale is milliseconds. . <<http://i.imgur.com/safKkCP.png>> .@@ -45,12 +45,15 @@ other-modules: Control.Concurrent.Chan.Unagi.Internal , Control.Concurrent.Chan.Unagi.Unboxed.Internal+ , Control.Concurrent.Chan.Unagi.Constants , Utilities , Data.Atomics.Counter.Fat ghc-options: -Wall -funbox-strict-fields build-depends: base < 5- , atomic-primops==0.6.0.5+ -- be conservative about atomic-primops, for now; really+ -- we're fine with any version that passes our tests:+ , atomic-primops >= 0.6.0.5 && <= 0.6.0.6 , primitive>=0.5.3 default-language: Haskell2010 @@ -58,7 +61,13 @@ if !arch(i386) && !arch(x86_64) cpp-options: -DNOT_x86 -+-- TODO+-- - Do a benchmark of multiple queues running in parallel, to see if we are+-- affected by global allocator issues with pinned memory:+-- http://thread.gmane.org/gmane.comp.lang.haskell.parallel/218+-- - On next benchmarks run, cut out "Demo with messages..with" and make unagi +-- view overlayed with drop shadow+-- -- Potential implementations roadmap (or we might just stick with this design -- for this package): --@@ -96,7 +105,7 @@ , UnagiUnboxed build-depends: base , primitive>=0.5.3- , atomic-primops==0.6.0.5+ , atomic-primops >= 0.6.0.5 && <= 0.6.0.6 , containers default-language: Haskell2010 @@ -150,7 +159,7 @@ -- for profiling, checking out core, etc executable dev-example- -- for n in `find dist/build/core-example/core-example-tmp -name '*dump-simpl'`; do cp $n "core-example/$(basename $n).$(git rev-parse --abbrev-ref HEAD)"; done+ -- for n in `find dist/build/dev-example/dev-example-tmp -name '*dump-simpl'`; do cp $n "core-example/$(basename $n).$(git rev-parse --abbrev-ref HEAD)"; done if !flag(dev) buildable: False else@@ -166,7 +175,7 @@ --ghc-options: -threaded -with-rtsopts=-N2 --ghc-options: -eventlog -- ...or do non-threaded runtime- ghc-prof-options: -fprof-auto+ --ghc-prof-options: -fprof-auto --Relevant profiling RTS settings: -xt -- TODO also check out +RTS -A10m, and look at output of -sstderr