diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright (c) 2013, Brandon Simmons + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above + copyright notice, this list of conditions and the following + disclaimer in the documentation and/or other materials provided + with the distribution. + + * Neither the name of Brandon Simmons nor the names of other + contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/benchmarks/multi.hs b/benchmarks/multi.hs new file mode 100644 --- /dev/null +++ b/benchmarks/multi.hs @@ -0,0 +1,254 @@ +{-# LANGUAGE BangPatterns #-} +module Main where + +import qualified Control.Concurrent.Chan.Unagi as U +import qualified Control.Concurrent.Chan.Unagi.Unboxed as UU +#ifdef COMPARE_BENCHMARKS +import Control.Concurrent.Chan +import Control.Concurrent.STM +--import qualified Data.Concurrent.Queue.MichaelScott as MS +#endif + +import Control.Concurrent.MVar +import Control.Concurrent.Async +import Control.Monad +import Criterion.Main +import GHC.Conc + + +main :: IO () +main = do +-- save some time and don't let other chans choke: +#ifdef COMPARE_BENCHMARKS + let n = 100000 +#else + let n = 100000 -- TODO 1mil ? +#endif + + procs <- getNumCapabilities + unless (even procs) $ + error "Please run with +RTS -NX, where X is an even number" + let procs_div2 = procs `div` 2 + if procs_div2 >= 0 then return () + else error "Run with RTS +N2 or more" + + putStrLn $ "Running with capabilities: "++(show procs) + + (fill_empty_fastUI, fill_empty_fastUO) <- U.newChan + (fill_empty_fastUUI, fill_empty_fastUUO) <- UU.newChan -- TODO WHY IS THIS COMPILING BELOW??? +#ifdef COMPARE_BENCHMARKS + fill_empty_chan <- newChan + fill_empty_tqueue <- newTQueueIO + --fill_empty_tbqueue <- newTBQueueIO maxBound + --fill_empty_lockfree <- MS.newQ +#endif + + defaultMain $ + [ bgroup ("Operations on "++(show n)++" messages") $ + [ bgroup "unagi-chan Unagi" $ + -- 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 + -- 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 + -- 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 + 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 + 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 + + , 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 + -- TODO using Ints here instead of (); change others so we can properly compare? + , bench ("async "++(show procs)++" writers") $ 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 + 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 + + , bench "async Int writer, main thread read and sum" $ nfIO $ asyncSumIntUnagiUnboxed n + ] +#ifdef COMPARE_BENCHMARKS + , bgroup "Chan" $ + [ bench "async 1 writer 1 readers" $ asyncReadsWritesChan 1 1 n + , bench "oversubscribing: async 100 writers 100 readers" $ asyncReadsWritesChan 100 100 n + , bench ("async "++(show procs)++" writers") $ do + dones <- replicateM procs newEmptyMVar ; starts <- replicateM procs newEmptyMVar + mapM_ (\(start1,done1)-> forkIO $ takeMVar start1 >> replicateM_ (n `div` procs) (writeChan fill_empty_chan ()) >> 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 + dones <- replicateM procs newEmptyMVar ; starts <- replicateM procs newEmptyMVar + mapM_ (\(start1,done1)-> forkIO $ takeMVar start1 >> replicateM_ (n `div` procs) (readChan fill_empty_chan) >> putMVar done1 ()) $ zip starts dones + mapM_ (\v-> putMVar v ()) starts ; mapM_ (\v-> takeMVar v) dones + ] + , bgroup "TQueue" $ + [ bench "async 1 writers 1 readers" $ asyncReadsWritesTQueue 1 1 n + , bench "oversubscribing: async 100 writers 100 readers" $ asyncReadsWritesTQueue 100 100 n + -- This measures writer/writer contention: + , bench ("async "++(show procs)++" writers") $ do + dones <- replicateM procs newEmptyMVar ; starts <- replicateM procs newEmptyMVar + mapM_ (\(start1,done1)-> forkIO $ takeMVar start1 >> replicateM_ (n `div` procs) (atomically $ writeTQueue fill_empty_tqueue ()) >> 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 + dones <- replicateM procs newEmptyMVar ; starts <- replicateM procs newEmptyMVar + mapM_ (\(start1,done1)-> forkIO $ takeMVar start1 >> replicateM_ (n `div` procs) (atomically $ readTQueue fill_empty_tqueue) >> putMVar done1 ()) $ zip starts dones + mapM_ (\v-> putMVar v ()) starts ; mapM_ (\v-> takeMVar v) dones + ] + {- + , bgroup "TBQueue" $ + [ bench "async 1 writers 1 readers" $ asyncReadsWritesTBQueue 1 1 n + , bench "oversubscribing: async 100 writers 100 readers" $ asyncReadsWritesTBQueue 100 100 n + -- This measures writer/writer contention: + , bench ("async "++(show procs)++" writers") $ do + dones <- replicateM procs newEmptyMVar ; starts <- replicateM procs newEmptyMVar + mapM_ (\(start1,done1)-> forkIO $ takeMVar start1 >> replicateM_ (n `div` procs) (atomically $ writeTBQueue fill_empty_tbqueue ()) >> 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 + dones <- replicateM procs newEmptyMVar ; starts <- replicateM procs newEmptyMVar + mapM_ (\(start1,done1)-> forkIO $ takeMVar start1 >> replicateM_ (n `div` procs) (atomically $ readTBQueue fill_empty_tbqueue) >> putMVar done1 ()) $ zip starts dones + mapM_ (\v-> putMVar v ()) starts ; mapM_ (\v-> takeMVar v) dones + ] + -- michael-scott queue implementation, using atomic-primops + , bgroup "lockfree-queue" $ + [ bench "async 1 writer 1 readers" $ asyncReadsWritesLockfree 1 1 n + , bench "oversubscribing: async 100 writers 100 readers" $ asyncReadsWritesLockfree 100 100 n + , bench ("async "++(show procs)++" writers") $ do + dones <- replicateM procs newEmptyMVar ; starts <- replicateM procs newEmptyMVar + mapM_ (\(start1,done1)-> forkIO $ takeMVar start1 >> replicateM_ (n `div` procs) (MS.pushL fill_empty_lockfree ()) >> putMVar done1 ()) $ zip starts dones + mapM_ (\v-> putMVar v ()) starts ; mapM_ (\v-> takeMVar v) dones + , bench ("async "++(show procs)++" readers") $ do + dones <- replicateM procs newEmptyMVar ; starts <- replicateM procs newEmptyMVar + mapM_ (\(start1,done1)-> forkIO $ takeMVar start1 >> replicateM_ (n `div` procs) (msreadR fill_empty_lockfree) >> putMVar done1 ()) $ zip starts dones + mapM_ (\v-> putMVar v ()) starts ; mapM_ (\v-> takeMVar v) dones + ] + -} +#endif + ] +#ifdef COMPARE_BENCHMARKS + -- This is our set of benchmarks we use to create the graph we'll use in + -- the haddocks to demo performance + , bgroup ("Demo with messages x"++show n) $ + let runs = [1..procs_div2] + benchRun c = bench ("with "++(show c)++ "readers and "++(show c)++" writers") + in [ bgroup "Unagi " $ + map (\c-> benchRun c $ asyncReadsWritesUnagi c c n) runs + , bgroup "Unagi.Unboxed" $ + map (\c-> benchRun c $ asyncReadsWritesUnagiUnboxed c c n) runs + , bgroup "TQueue " $ + map (\c-> benchRun c $ asyncReadsWritesTQueue c c n) runs + , bgroup "Chan " $ + map (\c-> benchRun c $ asyncReadsWritesChan c c n) runs + ] +#endif + ] + + +-- TODO maybe factor out reads/writes/news, and hope they get inlined + +asyncReadsWritesUnagi :: Int -> Int -> Int -> IO () +asyncReadsWritesUnagi writers readers n = do + let nNice = n - rem n (lcm writers readers) + (i,o) <- U.newChan + rcvrs <- replicateM readers $ async $ replicateM_ (nNice `quot` readers) $ U.readChan o + _ <- replicateM writers $ async $ replicateM_ (nNice `quot` writers) $ U.writeChan i () + mapM_ wait rcvrs + +-- A slightly more realistic benchmark, lets us see effects of unboxed strict +-- in value, and inlining effects w/ partially applied writeChan +asyncSumIntUnagi :: Int -> IO Int +asyncSumIntUnagi n = do + (i,o) <- U.newChan + let readerSum 0 !tot = return tot + readerSum !n' !tot = U.readChan o >>= readerSum (n'-1) . (tot+) + _ <- async $ mapM_ (U.writeChan i) [1..n] -- NOTE: partially-applied writeChan + readerSum n 0 + + + +-- Unboxed Unagi: +-- NOTE: using Int here instead of (). TODO change others so we can properly compare? +asyncReadsWritesUnagiUnboxed :: Int -> Int -> Int -> IO () +asyncReadsWritesUnagiUnboxed writers readers n = do + let nNice = n - rem n (lcm writers readers) + (i,o) <- UU.newChan + rcvrs <- replicateM readers $ async $ replicateM_ (nNice `quot` readers) $ UU.readChan o + _ <- replicateM writers $ async $ replicateM_ (nNice `quot` writers) $ UU.writeChan i (0::Int) + mapM_ wait rcvrs + +asyncSumIntUnagiUnboxed :: Int -> IO Int +asyncSumIntUnagiUnboxed n = do + (i,o) <- UU.newChan + let readerSum 0 !tot = return tot + readerSum !n' !tot = UU.readChan o >>= readerSum (n'-1) . (tot+) + _ <- async $ mapM_ (UU.writeChan i) [1..n] -- NOTE: partially-applied writeChan + readerSum n 0 + + + +#ifdef COMPARE_BENCHMARKS + +asyncReadsWritesChan :: Int -> Int -> Int -> IO () +asyncReadsWritesChan writers readers n = do + let nNice = n - rem n (lcm writers readers) + c <- newChan + rcvrs <- replicateM readers $ async $ replicateM_ (nNice `quot` readers) $ readChan c + _ <- replicateM writers $ async $ replicateM_ (nNice `quot` writers) $ writeChan c () + mapM_ wait rcvrs + + +asyncReadsWritesTQueue :: Int -> Int -> Int -> IO () +asyncReadsWritesTQueue writers readers n = do + let nNice = n - rem n (lcm writers readers) + c <- newTQueueIO + rcvrs <- replicateM readers $ async $ replicateM_ (nNice `quot` readers) $ atomically $ readTQueue c + _ <- replicateM writers $ async $ replicateM_ (nNice `quot` writers) $ atomically $ writeTQueue c () + mapM_ wait rcvrs + + +{- +-- lockfree-queue +asyncReadsWritesLockfree :: Int -> Int -> Int -> IO () +asyncReadsWritesLockfree writers readers n = do + let nNice = n - rem n (lcm writers readers) + c <- MS.newQ + rcvrs <- replicateM readers $ async $ replicateM_ (nNice `quot` readers) $ msreadR c + _ <- replicateM writers $ async $ replicateM_ (nNice `quot` writers) $ MS.pushL c () + mapM_ wait rcvrs + +-- a busy-blocking read: +msreadR :: MS.LinkedQueue a -> IO a +msreadR q = MS.tryPopR q >>= maybe (msreadR q) return + +-- TBQueue +asyncReadsWritesTBQueue :: Int -> Int -> Int -> IO () +asyncReadsWritesTBQueue writers readers n = do + let nNice = n - rem n (lcm writers readers) + c <- newTBQueueIO 4096 + rcvrs <- replicateM readers $ async $ replicateM_ (nNice `quot` readers) $ atomically $ readTBQueue c + _ <- replicateM writers $ async $ replicateM_ (nNice `quot` writers) $ atomically $ writeTBQueue c () + mapM_ wait rcvrs +-} + +#endif diff --git a/benchmarks/single.hs b/benchmarks/single.hs new file mode 100644 --- /dev/null +++ b/benchmarks/single.hs @@ -0,0 +1,177 @@ +module Main + where + + +import qualified Control.Concurrent.Chan.Unagi as U +import qualified Control.Concurrent.Chan.Unagi.Unboxed as UU +#ifdef COMPARE_BENCHMARKS +import Control.Concurrent.Chan +import Control.Concurrent.STM +--import qualified Data.Concurrent.Queue.MichaelScott as MS +#endif + +import Control.Monad +import Criterion.Main + + +main :: IO () +main = do +-- save some time and don't let other chans choke: +#ifdef COMPARE_BENCHMARKS + let n = 100000 +#else + let n = 1000000 +#endif + + (fastEmptyUI,fastEmptyUO) <- U.newChan + (fastEmptyUUI,fastEmptyUUO) <- UU.newChan +#ifdef COMPARE_BENCHMARKS + chanEmpty <- newChan + tqueueEmpty <- newTQueueIO + --tbqueueEmpty <- newTBQueueIO 2 + --lockfreeQEmpty <- MS.newQ +#endif + + defaultMain $ + -- 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? +#ifdef COMPARE_BENCHMARKS + , bench "Chan" (writeChan chanEmpty () >> readChan chanEmpty) + , bench "TQueue" (atomically (writeTQueue tqueueEmpty () >> readTQueue tqueueEmpty)) + {- + -- TODO when comparing our bounded queues: + , bench "TBQueue" (atomically (writeTBQueue tbqueueEmpty () >> readTBQueue tbqueueEmpty)) + -- TODO when works with 7.8 + , bench "lockfree-queue" (MS.pushL lockfreeQEmpty () >> msreadR lockfreeQEmpty) + -} +#endif + ] + , 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 +#ifdef COMPARE_BENCHMARKS + , bench "Chan" $ runtestChan1 n + , bench "TQueue" $ runtestTQueue1 n + -- , bench "TBQueue" $ runtestTBQueue1 n + -- , bench "lockfree-queue" $ runtestLockfreeQueue1 n +#endif + ] + , bgroup "repeated write some, read some" $ + [ bench "unagi-chan Unagi" $ runtestSplitChanU2 n + , bench "unagi-chan Unagi.Unboxed" $ runtestSplitChanUU2 n +#ifdef COMPARE_BENCHMARKS + , bench "Chan" $ runtestChan2 n + , bench "TQueue" $ runtestTQueue2 n + -- , bench "TBQueue" $ runtestTBQueue2 n + -- , bench "lockfree-queue" $ runtestLockfreeQueue2 n +#endif + ] + ] + ] + +-- unagi-chan Unagi -- +runtestSplitChanU1, runtestSplitChanU2 :: Int -> IO () +runtestSplitChanU1 n = do + (i,o) <- U.newChan + replicateM_ n $ U.writeChan i () + replicateM_ n $ U.readChan o + +runtestSplitChanU2 n = do + (i,o) <- U.newChan + let n1000 = n `quot` 1000 + replicateM_ 1000 $ do + replicateM_ n1000 $ U.writeChan i () + replicateM_ n1000 $ U.readChan o + + +-- unagi-chan Unagi Unboxed -- +-- TODO comparing () to Int. Change everywhere? +runtestSplitChanUU1, runtestSplitChanUU2 :: Int -> IO () +runtestSplitChanUU1 n = do + (i,o) <- UU.newChan + replicateM_ n $ UU.writeChan i (0::Int) + replicateM_ n $ UU.readChan o + +runtestSplitChanUU2 n = do + (i,o) <- UU.newChan + let n1000 = n `quot` 1000 + replicateM_ 1000 $ do + replicateM_ n1000 $ UU.writeChan i (0::Int) + replicateM_ n1000 $ UU.readChan o + + + + +#ifdef COMPARE_BENCHMARKS +-- ---------- +-- Chan +runtestChan1, runtestChan2 :: Int -> IO () +runtestChan1 n = do + c <- newChan + replicateM_ n $ writeChan c () + replicateM_ n $ readChan c + +runtestChan2 n = do + c <- newChan + let n1000 = n `quot` 1000 + replicateM_ 1000 $ do + replicateM_ n1000 $ writeChan c () + replicateM_ n1000 $ readChan c + + +-- ---------- +-- TQueue + +runtestTQueue1, runtestTQueue2 :: Int -> IO () +runtestTQueue1 n = do + c <- newTQueueIO + replicateM_ n $ atomically $ writeTQueue c () + replicateM_ n $ atomically $ readTQueue c + +runtestTQueue2 n = do + c <- newTQueueIO + let n1000 = n `quot` 1000 + replicateM_ 1000 $ do + replicateM_ n1000 $ atomically $ writeTQueue c () + replicateM_ n1000 $ atomically $ readTQueue c + +{- +-- ---------- +-- TBQueue +runtestTBQueue1, runtestTBQueue2 :: Int -> IO () +runtestTBQueue1 n = do + c <- newTBQueueIO n -- The original benchmark must have blocked indefinitely here, no? + replicateM_ n $ atomically $ writeTBQueue c () + replicateM_ n $ atomically $ readTBQueue c + +runtestTBQueue2 n = do + c <- newTBQueueIO 4096 + let n1000 = n `quot` 1000 + replicateM_ 1000 $ do + replicateM_ n1000 $ atomically $ writeTBQueue c () + replicateM_ n1000 $ atomically $ readTBQueue c + +-- ---------- +-- from "lockfree-queue" +runtestLockfreeQueue1, runtestLockfreeQueue2 :: Int -> IO () +runtestLockfreeQueue1 n = do + c <- MS.newQ + replicateM_ n $ MS.pushL c () + replicateM_ n $ msreadR c + +runtestLockfreeQueue2 n = do + c <- MS.newQ + let n1000 = n `quot` 1000 + replicateM_ 1000 $ do + replicateM_ n1000 $ MS.pushL c () + replicateM_ n1000 $ msreadR c + +-- a busy-blocking read: +msreadR :: MS.LinkedQueue a -> IO a +msreadR q = MS.tryPopR q >>= maybe (msreadR q) return +-} +#endif diff --git a/core-example/Main.hs b/core-example/Main.hs new file mode 100644 --- /dev/null +++ b/core-example/Main.hs @@ -0,0 +1,156 @@ +module Main (main) where + +import Control.Monad +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 as C +import qualified Control.Concurrent.STM.TQueue as S +import Control.Concurrent.STM + +import Debug.Trace + +-- This is a copy of the "async 100 writers 100 readers" from chan-benchmarks + {- +main = do + (nm:other) <- getArgs + let n = 1000000 + (r,w) = case other of + [rS,wS] -> (read rS, read wS) + _ -> (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 + -} +-} +{- +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 + let n1000 = n `quot` 1000 + replicateM_ 1000 $ do + replicateM_ n1000 $ U.writeChan i () + replicateM_ n1000 $ U.readChan o + +runF :: Int -> IO () +runF n = do + (i,o) <- F.newChan + let n1000 = n `quot` 1000 + replicateM_ 1000 $ do + replicateM_ n1000 $ F.writeChan i () + replicateM_ n1000 $ F.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) + perReader = nNice `quot` readers + perWriter = (nNice `quot` writers) + vs <- replicateM readers newEmptyMVar + (i,o) <- U.newChan + let doRead = replicateM_ perReader $ theRead + theRead = U.readChan o + doWrite = replicateM_ perWriter $ theWrite + theWrite = U.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 + +-- ------------------------------------------------ +-- FOR COMPARISON: + +runS :: Int -> Int -> Int -> IO () +runS writers readers n = do + let nNice = n - rem n (lcm writers readers) + perReader = nNice `quot` readers + perWriter = (nNice `quot` writers) + vs <- replicateM readers newEmptyMVar + tq <- S.newTQueueIO + let doRead = replicateM_ perReader $ theRead + theRead = (atomically . S.readTQueue) tq + doWrite = replicateM_ perWriter $ theWrite + theWrite = atomically $ S.writeTQueue tq (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 + +runC :: Int -> Int -> Int -> IO () +runC writers readers n = do + let nNice = n - rem n (lcm writers readers) + perReader = nNice `quot` readers + perWriter = (nNice `quot` writers) + vs <- replicateM readers newEmptyMVar + c <- C.newChan + let doRead = replicateM_ perReader $ theRead + theRead = C.readChan c + doWrite = replicateM_ perWriter $ theWrite + theWrite = C.writeChan c (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 +-} diff --git a/src/Control/Concurrent/Chan/Unagi.hs b/src/Control/Concurrent/Chan/Unagi.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Chan/Unagi.hs @@ -0,0 +1,47 @@ +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. + -} + -- * Creating channels + newChan + , InChan(), OutChan() + -- * Channel operations + -- ** Reading + , readChan + , readChanOnException + , getChanContents + -- ** Writing + , writeChan + , writeList2Chan + -- ** Broadcasting + , dupChan + ) where +-- TODO additonal functions: +-- - write functions optimized for single-writer +-- - faster write/read-many that increments counter by N + +import Control.Concurrent.Chan.Unagi.Internal +-- For 'writeList2Chan', as in vanilla Chan +import System.IO.Unsafe ( unsafeInterleaveIO ) + + +-- | Create a new channel, returning its write and read ends. +newChan :: IO (InChan a, OutChan a) +newChan = newChanStarting (maxBound - 10) + -- lets us test counter overflow in tests and normal course of operation + +-- | Return a lazy list representing the contents of the supplied OutChan, much +-- like System.IO.hGetContents. +getChanContents :: OutChan a -> IO [a] +getChanContents ch = unsafeInterleaveIO (do + x <- readChan ch + xs <- getChanContents ch + return (x:xs) + ) + +-- | Write an entire list of items to a chan type. Writes here from multiple +-- threads may be interleaved, and infinite lists are supported. +writeList2Chan :: InChan a -> [a] -> IO () +{-# INLINABLE writeList2Chan #-} +writeList2Chan ch = sequence_ . map (writeChan ch) diff --git a/src/Control/Concurrent/Chan/Unagi/Internal.hs b/src/Control/Concurrent/Chan/Unagi/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Chan/Unagi/Internal.hs @@ -0,0 +1,341 @@ +{-# LANGUAGE BangPatterns , DeriveDataTypeable, CPP #-} +module Control.Concurrent.Chan.Unagi.Internal +#ifdef NOT_x86 + {-# WARNING "This library is unlikely to perform well on architectures without a fetch-and-add instruction" #-} +#endif + (sEGMENT_LENGTH + , InChan(..), OutChan(..), ChanEnd(..), StreamSegment, Cell(..), Stream(..) + , NextSegment(..), StreamHead(..) + , newChanStarting, writeChan, readChan, readChanOnException + , dupChan + ) + where + +-- Internals exposed for testing. + +import Control.Concurrent.MVar +import Data.IORef +import Control.Exception +import Control.Monad.Primitive(RealWorld) +import Data.Atomics.Counter.Fat +import Data.Atomics +import qualified Data.Primitive as P +import Control.Monad +import Control.Applicative +import Data.Bits +import Data.Typeable(Typeable) +import GHC.Exts(inline) + + + +-- 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 + +data InChan a = InChan !(Ticket (Cell a)) !(ChanEnd a) + deriving Typeable +newtype OutChan a = OutChan (ChanEnd a) + deriving Typeable + +instance Eq (InChan a) where + (InChan _ (ChanEnd _ _ headA)) == (InChan _ (ChanEnd _ _ headB)) + = headA == headB +instance Eq (OutChan a) where + (OutChan (ChanEnd _ _ headA)) == (OutChan (ChanEnd _ _ headB)) + = headA == headB + +-- TODO POTENTIAL CPP FLAGS (or functions) +-- - Strict element (or lazy? maybe also expose a writeChan' when relevant?) +-- - sEGMENT_LENGTH +-- - reads that clear the element immediately (or export as a special function?) + +-- InChan & OutChan are mostly identical, sharing a stream, but with +-- independent counters +data ChanEnd a = + -- an efficient producer of segments of length sEGMENT_LENGTH: + ChanEnd !(SegSource a) + -- Both Chan ends must start with the same counter value. + !AtomicCounter + -- the stream head; this must never point to a segment whose offset + -- is greater than the counter value + !(IORef (StreamHead a)) + deriving Typeable + +data StreamHead a = StreamHead !Int !(Stream a) + +--TODO later see if we get a benefit from the small array primops in 7.10, +-- which omit card-marking overhead and might have faster clone. +type StreamSegment a = P.MutableArray RealWorld (Cell a) + +-- TRANSITIONS and POSSIBLE VALUES: +-- During Read: +-- Empty -> Blocking +-- Written +-- Blocking (only when dupChan used) +-- During Write: +-- Empty -> Written +-- Blocking +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 +-- required. + +data Stream a = + Stream !(StreamSegment a) + -- The next segment in the stream; new segments are allocated and + -- put here as we go, with threads cooperating to allocate new + -- segments: + !(IORef (NextSegment a)) + +data NextSegment a = NoSegment | Next !(Stream a) + +-- we expose `startingCellOffset` for debugging correct behavior with overflow: +newChanStarting :: Int -> IO (InChan a, OutChan a) +{-# INLINE newChanStarting #-} +newChanStarting !startingCellOffset = do + segSource <- newSegmentSource + firstSeg <- segSource + -- collect a ticket to save for writer CAS + savedEmptyTkt <- readArrayElem firstSeg 0 + stream <- Stream firstSeg <$> newIORef NoSegment + let end = ChanEnd segSource + <$> newCounter (startingCellOffset - 1) + <*> newIORef (StreamHead startingCellOffset stream) + liftA2 (,) (InChan savedEmptyTkt <$> end) (OutChan <$> end) + +-- | Duplicate a chan: the returned @OutChan@ begins empty, but data written to +-- the argument @InChan@ from then on will be available from both the original +-- @OutChan@ and the one returned here, creating a kind of broadcast channel. +dupChan :: InChan a -> IO (OutChan a) +{-# INLINE dupChan #-} +dupChan (InChan _ (ChanEnd segSource counter streamHead)) = do + hLoc <- readIORef streamHead + loadLoadBarrier -- NOTE [1] + wCount <- readCounter counter + + counter' <- newCounter wCount + streamHead' <- newIORef hLoc + return $ OutChan (ChanEnd segSource counter' streamHead') + -- [1] We must read the streamHead before inspecting the counter; otherwise, + -- as writers write, the stream head pointer may advance past the cell + -- indicated by wCount. + +-- | Write a value to the channel. +writeChan :: InChan a -> a -> IO () +{-# INLINE writeChan #-} +writeChan (InChan savedEmptyTkt ce@(ChanEnd segSource _ _)) = \a-> mask_ $ do + (segIx, (Stream seg next)) <- moveToNextCell ce + (success,nonEmptyTkt) <- casArrayElem seg segIx savedEmptyTkt (Written a) + -- try to pre-allocate next segment; NOTE [1] + when (segIx == 0) $ void $ + waitingAdvanceStream next segSource 0 + when (not success) $ + case peekTicket nonEmptyTkt of + Blocking v -> putMVar v a + Empty -> error "Stored Empty Ticket went stale!" -- NOTE [2] + Written _ -> error "Nearly Impossible! Expected Blocking" + -- [1] the writer which arrives first to the first cell of a new segment is + -- tasked (somewhat arbitrarily) with trying to pre-allocate the *next* + -- segment hopefully ahead of any readers or writers who might need it. This + -- will race with any reader *or* writer that tries to read the next segment + -- and finds it's empty (see `waitingAdvanceStream`); when this wins + -- (hopefully the vast majority of the time) we avoid a throughput hit. + -- + -- [2] this assumes that the compiler is statically-allocating Empty, sharing + -- the constructor among all uses, and that it never moves it between + -- checking the pointer stored in the array and checking the pointer in the + -- cached Any Empty value. If this is incorrect then the Ticket approach to + -- CAS is equally incorrect (though maybe less likely to fail). + + +-- We would like our queue to behave like Chan in that an async exception +-- raised in a reader known to be blocked or about to block on an empty queue +-- never results in a lost message; this matches our simple intuition about the +-- mechanics of a queue: if you're last in line for a cupcake and have to leave +-- you wouldn't expect the cake that would have gone to you to disappear. +-- +-- But there are other systems that a busy cake shop might use that you could +-- easily imagine resulting in a lost cake, and that's a different question +-- from whether cakes are given to well-behaved customers in the order they +-- came out of the oven, or whether a customer leaving at the wrong moment +-- might cause the cake shop to burn down... +readChanOnExceptionUnmasked :: (IO a -> IO a) -> OutChan a -> IO a +{-# INLINE readChanOnExceptionUnmasked #-} +readChanOnExceptionUnmasked h = \(OutChan ce)-> do + (segIx, (Stream seg _)) <- moveToNextCell ce + cellTkt <- readArrayElem seg segIx + case peekTicket cellTkt of + Written a -> return a + Empty -> do + v <- newEmptyMVar + (success,elseWrittenCell) <- casArrayElem seg segIx cellTkt (Blocking v) + if success + then readBlocking v + else case peekTicket elseWrittenCell of + -- In the meantime a writer has written. Good! + Written a -> return a + -- ...or a dupChan reader initiated blocking: + Blocking v2 -> readBlocking v2 + _ -> error "Impossible! Expecting Written or Blocking" + Blocking v -> readBlocking v + -- N.B. must use `readMVar` here to support `dupChan`: + where readBlocking v = inline h $ readMVar v + + +-- | Read an element from the chan, blocking if the chan is empty. +-- +-- /Note re. exceptions/: When an async exception is raised during a @readChan@ +-- the message that the read would have returned is likely to be lost, even when +-- the read is known to be blocked on an empty queue. If you need to handle +-- this scenario, you can use 'readChanOnException'. +readChan :: OutChan a -> IO a +{-# INLINE readChan #-} +readChan = readChanOnExceptionUnmasked id + +-- | Like 'readChan' but allows recovery of the queue element which would have +-- been read, in the case that an async exception is raised during the read. To +-- be precise exceptions are raised, and the handler run, only when +-- @readChanOnException@ is blocking. +-- +-- The second argument is a handler that takes a blocking IO action returning +-- the element, and performs some recovery action. When the handler is called, +-- the passed @IO a@ is the only way to access the element. +readChanOnException :: OutChan a -> (IO a -> IO ()) -> IO a +{-# INLINE readChanOnException #-} +readChanOnException c h = mask_ $ + readChanOnExceptionUnmasked (\io-> io `onException` (h io)) c + +-- increments counter, finds stream segment of corresponding cell (updating the +-- stream head pointer as needed), and returns the stream segment and relative +-- index of our cell. +moveToNextCell :: ChanEnd a -> IO (Int, Stream a) +{-# INLINE moveToNextCell #-} +moveToNextCell (ChanEnd segSource counter streamHead) = do + (StreamHead offset0 str0) <- readIORef streamHead + -- NOTE [3] +#ifdef NOT_x86 + -- fetch-and-add is a full barrier on x86; otherwise we need to make sure + -- the read above occurrs before our fetch-and-add: + loadLoadBarrier +#endif + ix <- incrCounter 1 counter + let (segsAway, segIx) = assert ((ix - offset0) >= 0) $ + divMod_sEGMENT_LENGTH $! (ix - offset0) + -- (ix - offset0) `quotRem` sEGMENT_LENGTH + {-# INLINE go #-} + go 0 str = return str + go !n (Stream _ next) = + waitingAdvanceStream next segSource (nEW_SEGMENT_WAIT*segIx) -- NOTE [1] + >>= go (n-1) + str <- go segsAway str0 + when (segsAway > 0) $ do + let !offsetN = + offset0 + (segsAway `unsafeShiftL` lOG_SEGMENT_LENGTH) --(segsAway*sEGMENT_LENGTH) + writeIORef streamHead $ StreamHead offsetN str -- NOTE [2] + return (segIx,str) + -- [1] All readers or writers needing to work with a not-yet-created segment + -- race to create it, but those past index 0 have progressively long waits; 20 + -- is chosen as 20 readIORefs should be more than enough time for writer/reader + -- 0 to add the new segment (if it's not descheduled). + -- + -- [2] advancing the stream head pointer on segIx == sEGMENT_LENGTH - 1 would + -- be more correct, but this is simpler here. This may move the head pointer + -- BACKWARDS if the thread was descheduled, but that's not a correctness + -- issue. + -- + -- [3] There is a theoretical race condition here: thread reads head and is + -- descheduled, meanwhile other readers/writers increment counter one full + -- lap; when we increment we think we've found our cell in what is actually a + -- very old segment. However in this scenario all addressable memory will + -- have been consumed just by the array pointers whivh haven't been able to + -- be GC'd. So I don't think this is something to worry about. + + +-- thread-safely try to fill `nextSegRef` at the next offset with a new +-- segment, waiting some number of iterations (for other threads to handle it). +-- Returns nextSegRef's StreamSegment. +waitingAdvanceStream :: IORef (NextSegment a) -> SegSource a + -> Int -> IO (Stream a) +waitingAdvanceStream nextSegRef segSource = go where + go !wait = assert (wait >= 0) $ do + tk <- readForCAS nextSegRef + case peekTicket tk of + NoSegment + | wait > 0 -> go (wait - 1) + -- Create a potential next segment and try to insert it: + | otherwise -> do + potentialStrNext <- Stream <$> segSource + <*> newIORef NoSegment + (_,tkDone) <- casIORef nextSegRef tk (Next potentialStrNext) + -- If that failed another thread succeeded (no false negatives) + case peekTicket tkDone of + Next strNext -> return strNext + _ -> error "Impossible! This should only have been Next segment" + Next strNext -> return strNext + + +-- copying a template array with cloneMutableArray is much faster than creating +-- a new one; in fact it seems we need this in order to scale, since as cores +-- increase we don't have enough "runway" and can't allocate fast enough: +type SegSource a = IO (StreamSegment a) + +newSegmentSource :: IO (SegSource a) +newSegmentSource = do + arr <- P.newArray sEGMENT_LENGTH Empty + return (P.cloneMutableArray arr 0 sEGMENT_LENGTH) + +-- ---------- +-- CELLS AND GC: +-- +-- Each cell in a segment is assigned at most one reader and one writer +-- +-- When all readers disappear and writers continue we'll have at most one +-- segment-worth of garbage that can't be collected at a time; when writers +-- advance the head segment pointer, the previous may be GC'd. +-- +-- 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 + -} diff --git a/src/Control/Concurrent/Chan/Unagi/Unboxed.hs b/src/Control/Concurrent/Chan/Unagi/Unboxed.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Chan/Unagi/Unboxed.hs @@ -0,0 +1,52 @@ +module Control.Concurrent.Chan.Unagi.Unboxed ( + -- * Creating channels + newChan + , InChan(), OutChan() + -- * Channel operations + -- ** Reading + , readChan + , readChanOnException + , getChanContents + -- ** Writing + , writeChan + , writeList2Chan + -- ** Broadcasting + , dupChan + ) where + +-- Forked from src/Control/Concurrent/Chan/Unagi/Internal.hs at 443465 +-- +-- TODO additonal functions: +-- - write functions optimized for single-writer +-- - faster write/read-many that increments counter by N +-- - this could be used (or forked) to implement an efficient MPSC concurrent +-- ByteString or Text queue (where writes could be variable-sized chunks +-- and we incrCounter accordingly) without too much trouble. Useful? +-- - likewise a SPMC concurrent bytestring consumer? +-- - ...or interop with 'vector' lib + +import Control.Concurrent.Chan.Unagi.Unboxed.Internal +-- For 'writeList2Chan', as in vanilla Chan +import System.IO.Unsafe ( unsafeInterleaveIO ) +import Data.Primitive(Prim) + + +-- | Create a new channel, returning its write and read ends. +newChan :: Prim a=> IO (InChan a, OutChan a) +newChan = newChanStarting (maxBound - 10) + -- lets us test counter overflow in tests and normal course of operation + +-- | Return a lazy list representing the contents of the supplied OutChan, much +-- like System.IO.hGetContents. +getChanContents :: Prim a=> OutChan a -> IO [a] +getChanContents ch = unsafeInterleaveIO (do + x <- readChan ch + xs <- getChanContents ch + return (x:xs) + ) + +-- | Write an entire list of items to a chan type. Writes here from multiple +-- threads may be interleaved, and infinite lists are supported. +writeList2Chan :: Prim a=> InChan a -> [a] -> IO () +{-# INLINABLE writeList2Chan #-} +writeList2Chan ch = sequence_ . map (writeChan ch) diff --git a/src/Control/Concurrent/Chan/Unagi/Unboxed/Internal.hs b/src/Control/Concurrent/Chan/Unagi/Unboxed/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Chan/Unagi/Unboxed/Internal.hs @@ -0,0 +1,337 @@ +{-# LANGUAGE BangPatterns , DeriveDataTypeable, CPP , ScopedTypeVariables #-} +module Control.Concurrent.Chan.Unagi.Unboxed.Internal +#ifdef NOT_x86 + {-# WARNING "This library is unlikely to perform well on architectures without a fetch-and-add instruction" #-} +#endif + (sEGMENT_LENGTH + , InChan(..), OutChan(..), ChanEnd(..), Cell, Stream(..), ElementArray(..), SignalIntArray + , readElementArray, writeElementArray + , NextSegment(..), StreamHead(..) + , newChanStarting, writeChan, readChan, readChanOnException + , dupChan + ) + where + +-- Forked from src/Control/Concurrent/Chan/Unagi/Internal.hs at 443465. See +-- that implementation for additional details and notes which we omit here. +-- +-- Internals exposed for testing. +-- +-- TODO +-- - Look at how ByteString is implemented; maybe that approach with +-- ForeignPtr is better in some ways, or perhaps we can use their Internals? +-- - we can make IndexedMVar () and always write to ByteString +-- - Also 'vector' lib + + +import Data.IORef +import Control.Exception +import Control.Monad.Primitive(RealWorld) +import Data.Atomics.Counter.Fat +import Data.Atomics +import qualified Data.Primitive as P +import Control.Monad +import Control.Applicative +import Data.Bits +import Data.Typeable(Typeable) +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 + +newtype InChan a = InChan (ChanEnd a) + deriving Typeable +newtype OutChan a = OutChan (ChanEnd a) + deriving Typeable + +instance Eq (InChan a) where + (InChan (ChanEnd _ headA)) == (InChan (ChanEnd _ headB)) + = headA == headB +instance Eq (OutChan a) where + (OutChan (ChanEnd _ headA)) == (OutChan (ChanEnd _ headB)) + = headA == headB + + +-- InChan & OutChan are mostly identical, sharing a stream, but with +-- independent counters +data ChanEnd a = + -- Both Chan ends must start with the same counter value. + ChanEnd !AtomicCounter + -- the stream head; this must never point to a segment whose offset + -- is greater than the counter value + !(IORef (StreamHead a)) + deriving Typeable + +data StreamHead a = StreamHead !Int !(Stream a) + + +-- The array we actually store our Prim elements in +newtype ElementArray a = ElementArray (P.MutableByteArray RealWorld) +-- TODO +-- - we could easily use 'vector' to support a wider array of primitive +-- elements here. +-- - and what about Storable? +-- see http://stackoverflow.com/q/4908880/176841 +-- - else test combining signal and element arrays into a single one that +-- places signal cell next to element cell, and use Addr to access? +-- (see also TODOs under Stream) + +readElementArray :: (P.Prim a)=> ElementArray a -> Int -> IO a +{-# INLINE readElementArray #-} +readElementArray (ElementArray arr) i = P.readByteArray arr i + +writeElementArray :: (P.Prim a)=> ElementArray a -> Int -> a -> IO () +{-# INLINE writeElementArray #-} +writeElementArray (ElementArray arr) i a = P.writeByteArray arr i a + +-- We CAS on this, using Ints to signal (see below) +type SignalIntArray = P.MutableByteArray RealWorld + +-- TRANSITIONS and POSSIBLE VALUES: +-- During Read: +-- Empty -> Blocking +-- Written +-- Blocking (only when dupChan used) +-- During Write: +-- Empty -> Written +-- Blocking +{- +data Cell a = Empty -- 0 + | Written -- 1 + | Blocking -- 2 +-} +type Cell = Int +cellEmpty, cellWritten, cellBlocking :: Cell +cellEmpty = 0 +cellWritten = 1 +cellBlocking = 2 + + +segSource :: forall a. (P.Prim a)=> IO (SignalIntArray, ElementArray a) --ScopedTypeVariables +{-# INLINE segSource #-} +segSource = do + -- A largish pinned array seems like it would be the best choice here. + sigArr <- P.newAlignedPinnedByteArray + (P.sizeOf cellEmpty `unsafeShiftL` lOG_SEGMENT_LENGTH) -- times sEGMENT_LENGTH + (P.alignment cellEmpty) + eArr <- P.newAlignedPinnedByteArray + (P.sizeOf (undefined :: a) `unsafeShiftL` lOG_SEGMENT_LENGTH) + (P.alignment (undefined :: a)) + P.setByteArray sigArr 0 sEGMENT_LENGTH cellEmpty + 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) + -- For coordinating blocking between reader/writer; NOTE [1] + !(IndexedMVar a) + -- The next segment in the stream; NOTE [2] + !(IORef (NextSegment a)) + -- [1] An important property: we can switch out this implementation as long + -- as it utilizes a fresh MVar for each reader/writer pair. + -- + -- [2] new segments are allocated and put here as we go, with threads + -- cooperating to allocate new segments: +-- TODO +-- - we could replace Stream with a single funky MutableByteArray, even +-- replacing the IORef with a stored Addr to the next segment, which is +-- initialized to maxBound (an impossible value hopefully?) indicating +-- NoSegment +-- - except for our MVarIndexed in current implementation + +data NextSegment a = NoSegment | Next !(Stream a) + +-- we expose `startingCellOffset` for debugging correct behavior with overflow: +newChanStarting :: (P.Prim a)=> Int -> IO (InChan a, OutChan a) +{-# INLINE newChanStarting #-} +newChanStarting !startingCellOffset = do + stream <- uncurry Stream <$> segSource <*> newIndexedMVar <*> newIORef NoSegment + let end = ChanEnd + <$> newCounter (startingCellOffset - 1) + <*> newIORef (StreamHead startingCellOffset stream) + liftA2 (,) (InChan <$> end) (OutChan <$> end) + + +-- | Duplicate a chan: the returned @OutChan@ begins empty, but data written to +-- the argument @InChan@ from then on will be available from both the original +-- @OutChan@ and the one returned here, creating a kind of broadcast channel. +dupChan :: InChan a -> IO (OutChan a) +{-# INLINE dupChan #-} +dupChan (InChan (ChanEnd counter streamHead)) = do + hLoc <- readIORef streamHead + loadLoadBarrier -- NOTE [1] + wCount <- readCounter counter + OutChan <$> (ChanEnd <$> newCounter wCount <*> newIORef hLoc) + -- [1] We must read the streamHead before inspecting the counter; otherwise, + -- as writers write, the stream head pointer may advance past the cell + -- indicated by wCount and the first cell become unreachable. + +-- | Write a value to the channel. +writeChan :: (P.Prim a)=> InChan a -> a -> IO () +{-# INLINE writeChan #-} +writeChan (InChan ce) = \a-> mask_ $ do + (segIx, (Stream sigArr eArr mvarIndexed next)) <- moveToNextCell ce + -- NOTE!: must write element before signaling with CAS: + writeElementArray eArr segIx a +#ifdef NOT_x86 + -- TODO Should we include this for correctness sake? Will GHC ever move a write ahead of a CAS? + -- CAS provides a full barrier on x86; otherwise we need to make sure the + -- read above occurs before our fetch-and-add: + writeBarrier +#endif + actuallyWas <- casByteArrayInt sigArr segIx cellEmpty cellWritten + -- try to pre-allocate next segment; NOTE [1] + when (segIx == 0) $ void $ + waitingAdvanceStream next 0 + case actuallyWas of + 0 {- Empty -} -> return () + 2 {- Blocking -} -> putMVarIx mvarIndexed segIx a + 1 {- Written -} -> error "Nearly Impossible! Expected Blocking" + _ -> error "Invalid signal seen in writeChan!" + -- [1] the writer which arrives first to the first cell of a new segment is + -- tasked (somewhat arbitrarily) with trying to pre-allocate the *next* + -- segment hopefully ahead of any readers or writers who might need it. This + -- will race with any reader *or* writer that tries to read the next segment + -- and finds it's empty (see `waitingAdvanceStream`); when this wins + -- (hopefully the vast majority of the time) we avoid a throughput hit. + + +readChanOnExceptionUnmasked :: (P.Prim a)=> (IO a -> IO a) -> OutChan a -> IO a +{-# INLINE readChanOnExceptionUnmasked #-} +readChanOnExceptionUnmasked h = \(OutChan ce)-> do + (segIx, (Stream sigArr eArr mvarIndexed _)) <- moveToNextCell ce + -- NOTE!: must read signal before reading element. No barrier necessary. + let readBlocking = inline h $ readMVarIx mvarIndexed segIx -- NOTE [1] + -- optimistically try read w/out CAS + sig <- P.readByteArray sigArr segIx + case (sig :: Int) of + 1 {- Written -} -> readElementArray eArr segIx + 2 {- Blocking -} -> readBlocking + _ -> do + actuallyWas <- casByteArrayInt sigArr segIx cellEmpty cellBlocking + case actuallyWas of + -- succeeded writing Empty; proceed with blocking + 0 {- Empty -} -> readBlocking + -- else in the meantime, writer wrote + 1 {- Written -} -> readElementArray eArr segIx + -- else in the meantime a dupChan reader read, blocking + 2 {- Blocking -} -> readBlocking + _ -> error "Invalid signal seen in readChanOnExceptionUnmasked!" + -- [1] we must use `readMVarIx` here to support `dupChan`. It's also + -- important that the behavior of readMVarIx be identical to a readMVar on + -- the same MVar. + + +-- | Read an element from the chan, blocking if the chan is empty. +-- +-- /Note re. exceptions/: When an async exception is raised during a @readChan@ +-- the message that the read would have returned is likely to be lost, even when +-- the read is known to be blocked on an empty queue. If you need to handle +-- this scenario, you can use 'readChanOnException'. +readChan :: (P.Prim a)=> OutChan a -> IO a +{-# INLINE readChan #-} +readChan = readChanOnExceptionUnmasked id + +-- | Like 'readChan' but allows recovery of the queue element which would have +-- been read, in the case that an async exception is raised during the read. To +-- be precise exceptions are raised, and the handler run, only when +-- @readChanOnException@ is blocking. +-- +-- The second argument is a handler that takes a blocking IO action returning +-- the element, and performs some recovery action. When the handler is called, +-- the passed @IO a@ is the only way to access the element. +readChanOnException :: (P.Prim a)=> OutChan a -> (IO a -> IO ()) -> IO a +{-# INLINE readChanOnException #-} +readChanOnException c h = mask_ $ + readChanOnExceptionUnmasked (\io-> io `onException` (h io)) c + +-- increments counter, finds stream segment of corresponding cell (updating the +-- stream head pointer as needed), and returns the stream segment and relative +-- index of our cell. +moveToNextCell :: (P.Prim a)=> ChanEnd a -> IO (Int, Stream a) +{-# INLINE moveToNextCell #-} +moveToNextCell (ChanEnd counter streamHead) = do + (StreamHead offset0 str0) <- readIORef streamHead + -- NOTE [3] +#ifdef NOT_x86 + -- fetch-and-add is a full barrier on x86; otherwise we need to make sure + -- the read above occurrs before our fetch-and-add: + loadLoadBarrier +#endif + ix <- incrCounter 1 counter + let (segsAway, segIx) = assert ((ix - offset0) >= 0) $ + divMod_sEGMENT_LENGTH $! (ix - offset0) + -- (ix - offset0) `quotRem` sEGMENT_LENGTH + {-# INLINE go #-} + go 0 str = return str + go !n (Stream _ _ _ next) = + waitingAdvanceStream next (nEW_SEGMENT_WAIT*segIx) -- NOTE [1] + >>= go (n-1) + str <- go segsAway str0 + when (segsAway > 0) $ do + let !offsetN = + offset0 + (segsAway `unsafeShiftL` lOG_SEGMENT_LENGTH) --(segsAway*sEGMENT_LENGTH) + writeIORef streamHead $ StreamHead offsetN str -- NOTE [2] + return (segIx,str) + -- [1] All readers or writers needing to work with a not-yet-created segment + -- race to create it, but those past index 0 have progressively long waits; 20 + -- is chosen as 20 readIORefs should be more than enough time for writer/reader + -- 0 to add the new segment (if it's not descheduled). + -- + -- [2] advancing the stream head pointer on segIx == sEGMENT_LENGTH - 1 would + -- be more correct, but this is simpler here. This may move the head pointer + -- BACKWARDS if the thread was descheduled, but that's not a correctness + -- issue. + -- + -- [3] There is a theoretical race condition here: thread reads head and is + -- descheduled, meanwhile other readers/writers increment counter one full + -- lap; when we increment we think we've found our cell in what is actually a + -- very old segment. However in this scenario all addressable memory will + -- have been consumed just by the array pointers whivh haven't been able to + -- be GC'd. So I don't think this is something to worry about. + + +-- thread-safely try to fill `nextSegRef` at the next offset with a new +-- segment, waiting some number of iterations (for other threads to handle it). +-- Returns nextSegRef's StreamSegment. +waitingAdvanceStream :: (P.Prim a)=> IORef (NextSegment a) -> Int -> IO (Stream a) +waitingAdvanceStream nextSegRef = go where + go !wait = assert (wait >= 0) $ do + tk <- readForCAS nextSegRef + case peekTicket tk of + NoSegment + | wait > 0 -> go (wait - 1) + -- Create a potential next segment and try to insert it: + | otherwise -> do + potentialStrNext <- uncurry Stream + <$> segSource + <*> newIndexedMVar + <*> newIORef NoSegment + (_,tkDone) <- casIORef nextSegRef tk (Next potentialStrNext) + -- If that failed another thread succeeded (no false negatives) + case peekTicket tkDone of + 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) diff --git a/src/Data/Atomics/Counter/Fat.hs b/src/Data/Atomics/Counter/Fat.hs new file mode 100644 --- /dev/null +++ b/src/Data/Atomics/Counter/Fat.hs @@ -0,0 +1,39 @@ +module Data.Atomics.Counter.Fat ( + AtomicCounter() + , newCounter + , incrCounter + , readCounter + ) where + +-- An atomic counter padded with 64-bytes (an x86 cache line) on either side to +-- try to avoid false sharing. + +import Data.Primitive.MachDeps(sIZEOF_INT) +import Control.Monad.Primitive(RealWorld) +import Data.Primitive.ByteArray +import Data.Atomics(fetchAddByteArrayInt) + +newtype AtomicCounter = AtomicCounter (MutableByteArray RealWorld) + +sIZEOF_CACHELINE , cACHELINE_PADDED_INT_IX :: Int +sIZEOF_CACHELINE = 64 +cACHELINE_PADDED_INT_IX = (sIZEOF_CACHELINE `quot` 2) `quot` sIZEOF_INT + +newCounter :: Int -> IO AtomicCounter +{-# INLINE newCounter #-} +newCounter n = do + arr <- newAlignedPinnedByteArray + sIZEOF_CACHELINE + sIZEOF_CACHELINE + writeByteArray arr cACHELINE_PADDED_INT_IX n + return (AtomicCounter arr) + +incrCounter :: Int -> AtomicCounter -> IO Int +{-# INLINE incrCounter #-} +incrCounter incr (AtomicCounter arr) = + fetchAddByteArrayInt arr cACHELINE_PADDED_INT_IX incr + +readCounter :: AtomicCounter -> IO Int +{-# INLINE readCounter #-} +readCounter (AtomicCounter arr) = + readByteArray arr cACHELINE_PADDED_INT_IX diff --git a/src/Utilities.hs b/src/Utilities.hs new file mode 100644 --- /dev/null +++ b/src/Utilities.hs @@ -0,0 +1,78 @@ +{-# LANGUAGE BangPatterns #-} +module Utilities ( + -- * Utility Chans + -- ** Indexed MVars + IndexedMVar() + , newIndexedMVar, putMVarIx, readMVarIx + -- ** Other stuff + , nextHighestPowerOfTwo + ) where + +import Control.Concurrent.MVar +import Control.Exception +import Control.Applicative +import Data.Bits +import Data.Word +import Data.Atomics +import Data.IORef + +-- For now: a reverse-ordered assoc list; an IntMap might be better +newtype IndexedMVar a = IndexedMVar (IORef [(Int, MVar a)]) + +newIndexedMVar :: IO (IndexedMVar a) +newIndexedMVar = IndexedMVar <$> newIORef [] + + + +-- these really suck; sorry. + +readMVarIx :: IndexedMVar a -> Int -> IO a +{-# INLINE readMVarIx #-} +readMVarIx mvIx i = do + readMVar =<< getMVarIx mvIx i + +putMVarIx :: IndexedMVar a -> Int -> a -> IO () +{-# INLINE putMVarIx #-} +putMVarIx mvIx i a = do + flip putMVar a =<< getMVarIx mvIx i + +getMVarIx :: IndexedMVar a -> Int -> IO (MVar a) +{-# INLINE getMVarIx #-} +getMVarIx (IndexedMVar v) i = do + -- We're right to optimistically create this for readMVarIx, but throw this + -- away for most putMVarIx (from writers), probably. + mv <- newEmptyMVar + tk0 <- readForCAS v + let go tk = do + let !xs = peekTicket tk + case findInsert i mv xs of + Left alreadyPresentMVar -> return alreadyPresentMVar + Right xs' -> do + (success,newTk) <- casIORef v tk xs' + if success + then return mv + else go newTk + go tk0 + +-- Reverse-sorted: +findInsert :: Int -> mvar -> [(Int,mvar)] -> Either mvar [(Int,mvar)] +{-# INLINE findInsert #-} +findInsert i mv = ins where + ins [] = Right [(i,mv)] + ins xss@((i',x):xs) = + case compare i i' of + GT -> Right $ (i,mv):xss + EQ -> Left x + LT -> fmap ((i',x):) $ ins xs + + +-- Not particularly fast; if needs moar fast see +-- http://graphics.stanford.edu/~seander/bithacks.html#RoundUpPowerOf2 +-- +nextHighestPowerOfTwo :: Int -> Int +nextHighestPowerOfTwo 0 = 1 +nextHighestPowerOfTwo n = + let !nhp2 = 2 ^ (ceiling (logBase 2 $ fromIntegral $ abs n :: Float) :: Int) + -- ensure return value is actually a positive power of 2: + in assert (nhp2 > 0 && popCount (fromIntegral nhp2 :: Word) == 1) + nhp2 diff --git a/tests/Main.hs b/tests/Main.hs new file mode 100644 --- /dev/null +++ b/tests/Main.hs @@ -0,0 +1,52 @@ +module Main + where + +import System.IO +import Control.Concurrent +import Control.Exception + +-- implementation-agnostic tests: +import Deadlocks +import Smoke +import DupChan + +-- implementation-specific tests: +import Unagi +import UnagiUnboxed + +-- Other +import Atomics +import IndexedMVar + +main :: IO () +main = do + assertionsWorking <- try $ assert False $ return () + case assertionsWorking of + Left (AssertionFailed _) -> putStrLn "Assertions: On" + _ -> error "Assertions aren't working" + + procs <- getNumCapabilities + if procs < 2 + then error "Tests are only effective if more than 1 core is available" + else return () + hSetBuffering stdout NoBuffering + + -- ----------------------------------- + + -- test important properties of our atomic-primops: + atomicsMain + + indexedMVarMain + + -- do things catch fire? + smokeMain + + -- dupChan tests + dupChanMain + + -- check for deadlocks: + deadlocksMain + + -- unagi-specific tests + unagiMain + unagiUnboxedMain diff --git a/unagi-chan.cabal b/unagi-chan.cabal new file mode 100644 --- /dev/null +++ b/unagi-chan.cabal @@ -0,0 +1,165 @@ +name: unagi-chan +version: 0.1.0.0 + +synopsis: Fast and scalable concurrent queues for x86, with a Chan-like API + +description: + This library provides implementations of concurrent FIFO queues (for both + general boxed and primitive unboxed values) that are fast, perform well + under contention, and offer a Chan-like interface. The library may be of + 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. + . + <> + . + And here is a view on just the unagi implementations. + . + <> + . + +license: BSD3 +license-file: LICENSE +author: Brandon Simmons +maintainer: brandon.m.simmons@gmail.com +category: Concurrency +build-type: Simple +cabal-version: >=1.10 +-- currently uploaded to imgur; move to this eventually +--extra-doc-files: images/*.png +--cabal-version: >=1.18 + + +flag dev + default: False + manual: True + +source-repository head + type: git + location: https://github.com/jberryman/unagi-chan.git + branch: master + +library + hs-source-dirs: src + exposed-modules: Control.Concurrent.Chan.Unagi + , Control.Concurrent.Chan.Unagi.Unboxed + + other-modules: Control.Concurrent.Chan.Unagi.Internal + , Control.Concurrent.Chan.Unagi.Unboxed.Internal + , Utilities + , Data.Atomics.Counter.Fat + + ghc-options: -Wall -funbox-strict-fields + build-depends: base < 5 + , atomic-primops==0.6.0.5 + , primitive>=0.5.3 + default-language: Haskell2010 + + -- We'll need some additional barriers for correctness: + if !arch(i386) && !arch(x86_64) + cpp-options: -DNOT_x86 + + +-- Potential implementations roadmap (or we might just stick with this design +-- for this package): +-- +-- - fixed size MutableArray of purely-functional dequeues ("Tako") (fetch-and-add, then CAS) +-- - variant replacing CAS with blocking turn-taking, also play with leap-frogging cache-lines +-- - variant in STM (how to safely do the initial incrCounter at most once though?) +-- would also let us separate read and write buckets. +-- - bounded Tako variant +-- - maybe implement "Fast Concurrent Queues for x86 Processors" by Morrison & Afek (non-blocking, probably more clever) +-- - Also looks like a similar (but lockfree, as above) counter-based queue has been developed by FB: +-- https://github.com/facebook/folly/blob/master/folly/MPMCQueue.h +-- - boxed Unagi variant avoiding CAS with read simply spinning a few times and then calling yield, or something else + + +-- Please just build tests and run: +-- $ time ./dist/build/test/test +-- Doing `cabal test` takes forever for some reason. +test-suite test + type: exitcode-stdio-1.0 + ghc-options: -Wall -funbox-strict-fields + ghc-options: -O2 -rtsopts -threaded -with-rtsopts=-N + ghc-options: -fno-ignore-asserts + -- I guess we need to put 'src' here to get access to Internal modules + hs-source-dirs: tests, src + main-is: Main.hs + build-depends: base + , primitive>=0.5.3 + , atomic-primops==0.6.0.5 + , containers + default-language: Haskell2010 + +-- compare benchmarks with Chan, TQueue, and (eventually) lockfree-queue? +flag compare-benchmarks + default: False + manual: True + +benchmark single + type: exitcode-stdio-1.0 + ghc-options: -Wall -O2 -threaded -funbox-strict-fields -fforce-recomp -rtsopts + hs-source-dirs: benchmarks + default-language: Haskell2010 + default-extensions: CPP + build-depends: base + , unagi-chan + , criterion + if flag(compare-benchmarks) + cpp-options: -DCOMPARE_BENCHMARKS + build-depends: stm + -- , lockfree-queue + + main-is: single.hs + ghc-options: -with-rtsopts=-N1 + +-- To run comparison benchmark used in graph above, run: +-- $ cabal configure --enable-benchmarks -fcompare-benchmarks +-- $ cabal bench multi --benchmark-option=-omulti3.html --benchmark-option='Demo' +benchmark multi + type: exitcode-stdio-1.0 + ghc-options: -Wall -O2 -threaded -funbox-strict-fields -fforce-recomp -rtsopts + hs-source-dirs: benchmarks + default-language: Haskell2010 + default-extensions: CPP + build-depends: base + , unagi-chan + , criterion + if flag(compare-benchmarks) + cpp-options: -DCOMPARE_BENCHMARKS + build-depends: stm + -- , lockfree-queue + + main-is: multi.hs + ghc-options: -with-rtsopts=-N + build-depends: async + + +-- 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 + if !flag(dev) + buildable: False + + ghc-options: -ddump-to-file -ddump-simpl -dsuppress-module-prefixes -dsuppress-uniques -ddump-core-stats -ddump-inlinings + ghc-options: -O2 -rtsopts + + -- Either do threaded for eventlogging and simple timing... + --ghc-options: -threaded -with-rtsopts=-N2 + --ghc-options: -eventlog + -- ...or do non-threaded runtime + ghc-prof-options: -fprof-auto + --Relevant profiling RTS settings: -xt + -- TODO also check out +RTS -A10m, and look at output of -sstderr + + hs-source-dirs: core-example + main-is: Main.hs + build-depends: base + , stm + , unagi-chan + default-language: Haskell2010 +