packages feed

unagi-chan (empty) → 0.1.0.0

raw patch · 13 files changed

+1730/−0 lines, 13 filesdep +asyncdep +atomic-primopsdep +basesetup-changed

Dependencies added: async, atomic-primops, base, containers, criterion, primitive, stm, unagi-chan

Files

+ LICENSE view
@@ -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.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ benchmarks/multi.hs view
@@ -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
+ benchmarks/single.hs view
@@ -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
+ core-example/Main.hs view
@@ -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+-}
+ src/Control/Concurrent/Chan/Unagi.hs view
@@ -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)
+ src/Control/Concurrent/Chan/Unagi/Internal.hs view
@@ -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+ -}
+ src/Control/Concurrent/Chan/Unagi/Unboxed.hs view
@@ -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)
+ src/Control/Concurrent/Chan/Unagi/Unboxed/Internal.hs view
@@ -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)
+ src/Data/Atomics/Counter/Fat.hs view
@@ -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
+ src/Utilities.hs view
@@ -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
+ tests/Main.hs view
@@ -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
+ unagi-chan.cabal view
@@ -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.+    .+    <<http://i.imgur.com/safKkCP.png>>+    .+    And here is a view on just the unagi implementations.+    .+    <<http://i.imgur.com/K6s2pXj.png>>+    .+    +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+