unagi-chan 0.1.1.0 → 0.2.0.0
raw patch · 19 files changed
+743/−148 lines, 19 filesdep ~basePVP ok
version bump matches the API change (PVP)
Dependency ranges changed: base
API changes (from Hackage documentation)
+ Control.Concurrent.Chan.Unagi.Bounded: data InChan a
+ Control.Concurrent.Chan.Unagi.Bounded: data OutChan a
+ Control.Concurrent.Chan.Unagi.Bounded: dupChan :: InChan a -> IO (OutChan a)
+ Control.Concurrent.Chan.Unagi.Bounded: getChanContents :: OutChan a -> IO [a]
+ Control.Concurrent.Chan.Unagi.Bounded: newChan :: Int -> IO (InChan a, OutChan a)
+ Control.Concurrent.Chan.Unagi.Bounded: readChan :: OutChan a -> IO a
+ Control.Concurrent.Chan.Unagi.Bounded: readChanOnException :: OutChan a -> (IO a -> IO ()) -> IO a
+ Control.Concurrent.Chan.Unagi.Bounded: tryWriteChan :: InChan a -> a -> IO Bool
+ Control.Concurrent.Chan.Unagi.Bounded: writeChan :: InChan a -> a -> IO ()
+ Control.Concurrent.Chan.Unagi.Bounded: writeList2Chan :: InChan a -> [a] -> IO ()
Files
- CHANGELOG.markdown +9/−0
- benchmarks/multi.hs +32/−75
- benchmarks/single.hs +27/−6
- core-example/Main.hs +13/−1
- src/Control/Concurrent/Chan/Unagi.hs +4/−2
- src/Control/Concurrent/Chan/Unagi/Bounded.hs +59/−0
- src/Control/Concurrent/Chan/Unagi/Bounded/Internal.hs +473/−0
- src/Control/Concurrent/Chan/Unagi/Constants.hs +0/−4
- src/Control/Concurrent/Chan/Unagi/Internal.hs +3/−1
- src/Utilities.hs +9/−5
- tests/Chan003.hs +0/−25
- tests/Deadlocks.hs +45/−4
- tests/DupChan.hs +13/−0
- tests/Implementations.hs +3/−0
- tests/Main.hs +6/−2
- tests/Smoke.hs +34/−9
- tests/Unagi.hs +3/−2
- tests/UnagiUnboxed.hs +2/−2
- unagi-chan.cabal +8/−10
+ CHANGELOG.markdown view
@@ -0,0 +1,9 @@+### 0.1.1.0++- support new criterion and GHC 7.8.3+- small performance improvement to boxed unagi++### 0.2++- implement a bounded variant (See issue #1)+- address issue with stale tickets when running in GHCi
benchmarks/multi.hs view
@@ -3,13 +3,13 @@ import qualified Control.Concurrent.Chan.Unagi as U import qualified Control.Concurrent.Chan.Unagi.Unboxed as UU+import qualified Control.Concurrent.Chan.Unagi.Bounded as UB #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@@ -34,15 +34,6 @@ 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" $@@ -56,92 +47,37 @@ -- all of the above; this is probably less than -- informative. Try threadscope on a standalone test: , bench "oversubscribing: async 100 writers 100 readers" $ nfIO $ asyncReadsWritesUnagi 100 100 n- -- NOTE: this is a bit hackish, filling in one test and- -- reading in the other; make sure memory usage isn't- -- influencing mean:- -- This measures writer/writer contention:- , bench ("async "++(show procs)++" writers") $ nfIO $ do- dones <- replicateM procs newEmptyMVar ; starts <- replicateM procs newEmptyMVar- mapM_ (\(start1,done1)-> forkIO $ takeMVar start1 >> replicateM_ (n `div` procs) (U.writeChan fill_empty_fastUI ()) >> putMVar done1 ()) $ zip starts dones- mapM_ (\v-> putMVar v ()) starts ; mapM_ (\v-> takeMVar v) dones- -- This measures reader/reader contention:- , bench ("async "++(show procs)++" readers") $ nfIO $ do- dones <- replicateM procs newEmptyMVar ; starts <- replicateM procs newEmptyMVar- mapM_ (\(start1,done1)-> forkIO $ takeMVar start1 >> replicateM_ (n `div` procs) (U.readChan fill_empty_fastUO) >> putMVar done1 ()) $ zip starts dones- mapM_ (\v-> putMVar v ()) starts ; mapM_ (\v-> takeMVar v) dones- , bench "async Int writer, main thread read and sum" $ nfIO $ asyncSumIntUnagi n ] , bgroup "unagi-chan Unagi.Unboxed" $ [ bench "async 1 writers 1 readers" $ nfIO $ asyncReadsWritesUnagiUnboxed 1 1 n , bench "oversubscribing: async 100 writers 100 readers" $ nfIO $ asyncReadsWritesUnagiUnboxed 100 100 n- -- TODO using Ints here instead of (); change others so we can properly compare?- , bench ("async "++(show procs)++" writers") $ nfIO $ do- dones <- replicateM procs newEmptyMVar ; starts <- replicateM procs newEmptyMVar- mapM_ (\(start1,done1)-> forkIO $ takeMVar start1 >> replicateM_ (n `div` procs) (UU.writeChan fill_empty_fastUUI (0::Int)) >> putMVar done1 ()) $ zip starts dones- mapM_ (\v-> putMVar v ()) starts ; mapM_ (\v-> takeMVar v) dones- , bench ("async "++(show procs)++" readers") $ nfIO $ do- dones <- replicateM procs newEmptyMVar ; starts <- replicateM procs newEmptyMVar- mapM_ (\(start1,done1)-> forkIO $ takeMVar start1 >> replicateM_ (n `div` procs) (UU.readChan fill_empty_fastUUO) >> putMVar done1 ()) $ zip starts dones- mapM_ (\v-> putMVar v ()) starts ; mapM_ (\v-> takeMVar v) dones- , bench "async Int writer, main thread read and sum" $ nfIO $ asyncSumIntUnagiUnboxed n ]+ , bgroup "unagi-chan Unagi.Bounded" $+ [ bench "async 1 writers 1 readers" $ nfIO $ asyncReadsWritesUnagiBounded 4096 1 1 n -- TODO with different bounds, here and below+ , bench "oversubscribing: async 100 writers 100 readers" $ nfIO $ asyncReadsWritesUnagiBounded 4096 100 100 n+ , bench "async Int writer, main thread read and sum" $ nfIO $ asyncSumIntUnagiBounded 4096 n -- TODO with different bounds+ ] #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+ [ bench "async 1 writer 1 readers" $ nfIO $ asyncReadsWritesChan 1 1 n+ , bench "oversubscribing: async 100 writers 100 readers" $ nfIO $ asyncReadsWritesChan 100 100 n ] , 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+ [ bench "async 1 writers 1 readers" $ nfIO $ asyncReadsWritesTQueue 1 1 n+ , bench "oversubscribing: async 100 writers 100 readers" $ nfIO $ asyncReadsWritesTQueue 100 100 n ] {- , 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@@ -151,11 +87,13 @@ -- 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")+ benchRun c = bench ("with "++(show c)++ " readers and "++(show c)++" writers") . nfIO 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 "Unagi.Bounded (4096)" $+ map (\c-> benchRun c $ asyncReadsWritesUnagiBounded 4096 c c n) runs -- TODO with different bounds. , bgroup "TQueue " $ map (\c-> benchRun c $ asyncReadsWritesTQueue c c n) runs , bgroup "Chan " $@@ -203,6 +141,25 @@ 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+++-- Bounded Unagi:+-- NOTE: using Int here instead of (). TODO change others so we can properly compare?+asyncReadsWritesUnagiBounded :: Int -> Int -> Int -> Int -> IO ()+asyncReadsWritesUnagiBounded bnds writers readers n = do+ let nNice = n - rem n (lcm writers readers)+ (i,o) <- UB.newChan bnds+ rcvrs <- replicateM readers $ async $ replicateM_ (nNice `quot` readers) $ UB.readChan o+ _ <- replicateM writers $ async $ replicateM_ (nNice `quot` writers) $ UB.writeChan i (0::Int)+ mapM_ wait rcvrs++asyncSumIntUnagiBounded :: Int -> Int -> IO Int+asyncSumIntUnagiBounded bnds n = do+ (i,o) <- UB.newChan bnds+ let readerSum 0 !tot = return tot+ readerSum !n' !tot = UB.readChan o >>= readerSum (n'-1) . (tot+)+ _ <- async $ mapM_ (UB.writeChan i) [1..n] -- NOTE: partially-applied writeChan readerSum n 0
benchmarks/single.hs view
@@ -4,6 +4,7 @@ import qualified Control.Concurrent.Chan.Unagi as U import qualified Control.Concurrent.Chan.Unagi.Unboxed as UU+import qualified Control.Concurrent.Chan.Unagi.Bounded as UB #ifdef COMPARE_BENCHMARKS import Control.Concurrent.Chan import Control.Concurrent.STM@@ -25,6 +26,7 @@ (fastEmptyUI,fastEmptyUO) <- U.newChan (fastEmptyUUI,fastEmptyUUO) <- UU.newChan+ (fastEmptyUBI,fastEmptyUBO) <- UB.newChan 1024 -- only needs to be 1, but do apples-to-apples by matching sEGMENT_SIZE of other implementations #ifdef COMPARE_BENCHMARKS chanEmpty <- newChan tqueueEmpty <- newTQueueIO@@ -38,9 +40,11 @@ [ bgroup "Latency micro-benchmark" $ [ bench "unagi-chan Unagi" $ nfIO (U.writeChan fastEmptyUI () >> U.readChan fastEmptyUO) , bench "unagi-chan Unagi.Unboxed" $ nfIO (UU.writeChan fastEmptyUUI (0::Int) >> UU.readChan fastEmptyUUO) -- TODO comparing Int writing to (). Change?+ , bench "unagi-chan Unagi.Bounded 1024" $ nfIO (UB.writeChan fastEmptyUBI (0::Int) >> UB.readChan fastEmptyUBO) -- TODO comparing Int writing to (). Change?+ , bench "unagi-chan Unagi.Bounded 1024 with tryWriteChan" $ nfIO (UB.tryWriteChan fastEmptyUBI (0::Int) >> UB.readChan fastEmptyUBO) -- TODO comparing Int writing to (). Change? #ifdef COMPARE_BENCHMARKS- , bench "Chan" (writeChan chanEmpty () >> readChan chanEmpty)- , bench "TQueue" (atomically (writeTQueue tqueueEmpty () >> readTQueue tqueueEmpty))+ , bench "Chan" $ nfIO $ (writeChan chanEmpty () >> readChan chanEmpty)+ , bench "TQueue" $ nfIO $ (atomically (writeTQueue tqueueEmpty () >> readTQueue tqueueEmpty)) {- -- TODO when comparing our bounded queues: , bench "TBQueue" (atomically (writeTBQueue tbqueueEmpty () >> readTBQueue tbqueueEmpty))@@ -53,9 +57,10 @@ [ bgroup "sequential write all then read all" $ [ bench "unagi-chan Unagi" $ nfIO $ runtestSplitChanU1 n , bench "unagi-chan Unagi.Unboxed" $ nfIO $ runtestSplitChanUU1 n+ , bench "unagi-chan Unagi.Bounded" $ nfIO $ runtestSplitChanUB1 n #ifdef COMPARE_BENCHMARKS- , bench "Chan" $ runtestChan1 n- , bench "TQueue" $ runtestTQueue1 n+ , bench "Chan" $ nfIO $ runtestChan1 n+ , bench "TQueue" $ nfIO $ runtestTQueue1 n -- , bench "TBQueue" $ runtestTBQueue1 n -- , bench "lockfree-queue" $ runtestLockfreeQueue1 n #endif@@ -63,9 +68,10 @@ , bgroup "repeated write some, read some" $ [ bench "unagi-chan Unagi" $ nfIO $ runtestSplitChanU2 n , bench "unagi-chan Unagi.Unboxed" $ nfIO $ runtestSplitChanUU2 n+ , bench "unagi-chan Unagi.Bounded" $ nfIO $ runtestSplitChanUB2 n #ifdef COMPARE_BENCHMARKS- , bench "Chan" $ runtestChan2 n- , bench "TQueue" $ runtestTQueue2 n+ , bench "Chan" $ nfIO $ runtestChan2 n+ , bench "TQueue" $ nfIO $ runtestTQueue2 n -- , bench "TBQueue" $ runtestTBQueue2 n -- , bench "lockfree-queue" $ runtestLockfreeQueue2 n #endif@@ -104,6 +110,21 @@ replicateM_ n1000 $ UU.readChan o +-- unagi-chan Unagi Bounded --+-- NOTE: the first does no testing of the bounds checking overhead, while the+-- second does only a little. The multi.hs tests are a better place to look.+runtestSplitChanUB1, runtestSplitChanUB2 :: Int -> IO ()+runtestSplitChanUB1 n = do+ (i,o) <- UB.newChan n+ replicateM_ n $ UB.writeChan i ()+ replicateM_ n $ UB.readChan o++runtestSplitChanUB2 n = do+ let n1000 = n `quot` 1000+ (i,o) <- UB.newChan n1000+ replicateM_ 1000 $ do+ replicateM_ n1000 $ UB.writeChan i ()+ replicateM_ n1000 $ UB.readChan o #ifdef COMPARE_BENCHMARKS
core-example/Main.hs view
@@ -6,6 +6,7 @@ import Control.Concurrent import qualified Control.Concurrent.Chan.Unagi as U import qualified Control.Concurrent.Chan.Unagi.Unboxed as UU+import qualified Control.Concurrent.Chan.Unagi.Bounded as UB import qualified Control.Concurrent.Chan as C import qualified Control.Concurrent.STM.TQueue as S import Control.Concurrent.STM@@ -30,7 +31,8 @@ main = do [n] <- getArgs -- runU (read n)- runUU (read n)+ -- runUU (read n)+ runUB (read n) {- runU :: Int -> IO () runU n = do@@ -40,6 +42,7 @@ replicateM_ n1000 $ U.writeChan i () replicateM_ n1000 $ U.readChan o -}+{- runUU :: Int -> IO () runUU n = do (i,o) <- UU.newChan@@ -47,6 +50,15 @@ replicateM_ 1000 $ do replicateM_ n1000 $ UU.writeChan i (0::Int) replicateM_ n1000 $ UU.readChan o+ -}++runUB :: Int -> IO ()+runUB n = do+ let n1000 = n `quot` 1000+ (i,o) <- UB.newChan n1000+ replicateM_ 1000 $ do+ replicateM_ n1000 $ UB.writeChan i (0::Int)+ replicateM_ n1000 $ UB.readChan o {- runU :: Int -> Int -> Int -> IO ()
src/Control/Concurrent/Chan/Unagi.hs view
@@ -1,7 +1,9 @@ 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 grows very large.+ 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 grows very large. See also the bounded variant+ at "Control.Concurrent.Chan.Unagi.Bounded". -} -- * Creating channels newChan
+ src/Control/Concurrent/Chan/Unagi/Bounded.hs view
@@ -0,0 +1,59 @@+{-# LANGUAGE CPP #-}+module Control.Concurrent.Chan.Unagi.Bounded (+#ifdef NOT_x86+ {-# WARNING "This library is unlikely to perform well on architectures without a fetch-and-add instruction" #-}+#endif+#if __GLASGOW_HASKELL__ < 708+ {-# WARNING "Waking up blocked writers may be slower than desired in GHC<7.8 which makes readMVar non-blocking on full MVars. Considering upgrading." #-}+#endif+{- | A queue with bounded size, which supports a 'writeChan' which blocks when+ the number of messages grows larger than desired. The bounds are+ maintained loosely between @n@ and @n*2@; see the caveats and descriptions+ of semantics in 'readChan' and 'writeChan' for details.+ -}+ -- * Creating channels+ newChan+ , InChan(), OutChan()+ -- * Channel operations+ -- ** Reading+ , readChan+ , readChanOnException+ , getChanContents+ -- ** Writing+ , writeChan+ , tryWriteChan+ , writeList2Chan+ -- ** Broadcasting+ , dupChan+ ) where++-- forked from src/Control/Concurrent/Chan/Unagi.hs 43706b2++import Control.Concurrent.Chan.Unagi.Bounded.Internal+-- For 'writeList2Chan', as in vanilla Chan+import System.IO.Unsafe ( unsafeInterleaveIO ) +++-- | Create a new channel of the passed size, returning its write and read ends.+--+-- The passed integer bounds will be rounded up to the next highest power of+-- two, @n@. The queue may grow up to size @2*n@ (see 'writeChan' for details),+-- and the resulting chan pair requires O(n) space.+newChan :: Int -> IO (InChan a, OutChan a)+newChan size = newChanStarting (maxBound - 10) size+ -- 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/Bounded/Internal.hs view
@@ -0,0 +1,473 @@+{-# LANGUAGE BangPatterns , DeriveDataTypeable, CPP #-}+module Control.Concurrent.Chan.Unagi.Bounded.Internal+ ( InChan(..), OutChan(..), ChanEnd(..), StreamSegment, Cell(..), Stream(..)+ , writerCheckin, unblockWriters, tryWriterCheckin, WriterCheckpoint(..)+ , NextSegment(..), StreamHead(..)+ , newChanStarting, writeChan, readChan, readChanOnException+ , tryWriteChan+ , dupChan+ )+ where++-- NOTE: forked from src/Control/Concurrent/Chan/Unagi/Internal.hs 43706b2+-- some commentary not specific to this Bounded variant has been removed.+-- See the Unagi source for that.++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.Maybe(fromMaybe)+import Data.Typeable(Typeable)+import GHC.Exts(inline)++import Utilities(nextHighestPowerOfTwo)+++-- | The write end of a channel created with 'newChan'.+data InChan a = InChan (IO Int) -- readCounterReader, for tryWriteChan+ !(Ticket (Cell a)) + !(ChanEnd a)+ deriving Typeable++-- | The read end of a channel created with 'newChan'.+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++data ChanEnd a = + -- For efficient div and mod:+ ChanEnd !Int -- logBase 2 BOUNDS+ !Int -- BOUNDS - 1+ -- an efficient producer of segments of length BOUNDS:+ !(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)) -- NOTE [1]+ deriving Typeable+ -- [1] For the writers' ChanEnd: the segment that appears in the StreamHead is+ -- implicitly unlocked for writers (the segment size being equal to the chan+ -- bounds). See 'writeChan' for notes on how we make sure to keep this+ -- invariant.++data StreamHead a = StreamHead !Int !(Stream a)++-- This is always of length BOUNDS+type StreamSegment a = P.MutableArray RealWorld (Cell a)++data Cell a = Empty | Written a | Blocking !(MVar a)++data Stream a = + Stream !(StreamSegment a)+ -- The next segment in the stream; new segments are allocated and+ -- put here by the reader of index-0 of the previous segment. That+ -- reader (and the next one or two) also do a tryPutMVar to the MVar+ -- below, to indicate that any blocked writers may proceed.+ !(IORef (Maybe (NextSegment a)))+ -- writers that find Nothing above, must check in with a possibly+ -- blocking readMVar here before proceeding (the slow path):+++-- Next segment, installed by either a reader or a writer:+data NextSegment a = NextByWriter (Stream a) -- the next stream segment+ !WriterCheckpoint -- blocking for this segment+ -- a reader-installed one is implicitly unlocked for writers+ -- so needs no checkpoint:+ | NextByReader (Stream a)++-- helper accessors TODO consider making records+getNextRef :: NextSegment a -> IORef (Maybe (NextSegment a))+getNextRef x = (\(Stream _ nextSegRef)-> nextSegRef) $ getStr x++getStr :: NextSegment a -> Stream a+getStr (NextByReader str) = str+getStr (NextByWriter str _) = str++asReader, asWriter :: Bool+asReader = True+asWriter = False+++-- WRITER BLOCKING SCHEME OVERVIEW+-- -------------------------------+-- We use segments of size equal to the requested bounds. When a reader reads+-- index 0 of a segment it tries to pre-allocate the next segment, marking it+-- installed by reader (NextByReader), which indicates to writers who read it+-- that they may write and return without blocking (and this makes the queue+-- loosely bounded between size n and n*2).+--+-- Whenever a reader encounters a segment (in waitingAdvanceStream) installed+-- by a writer it unblocks writers and rewrites the NextBy* constructor to+-- NextByReader, replacing the installed stream segment.+--+-- Writers first make their write available (by writing to the segment) before+-- doing any blocking. This is more efficient, lets us handle async exceptions+-- in a principled way without changing the semantics. This also means that in+-- some cases a writer will install the next segment, marked installed by+-- writer, insicating that writers must checkin and block; readers will mark+-- these segments as installed by reader to avoid unnecessary overhead when the+-- segment becomes unlocked as described in the paragraph above.+--+-- The writer StreamHead is only ever updated by a writer that sees that a+-- segment is unlocked for writing (either because the writer has returned from+-- blocking on that segment, or because it sees that it was installed by a+-- reader); in this way a writer knows if its segment is at the StreamHead that+-- it is free to write and return without blocking (writerCheckin).+++newChanStarting :: Int -> Int -> IO (InChan a, OutChan a)+{-# INLINE newChanStarting #-}+newChanStarting !startingCellOffset !sizeDirty = do+ let !size = nextHighestPowerOfTwo sizeDirty+ !logBounds = round $ logBase (2::Float) $ fromIntegral size+ !boundsMn1 = size - 1++ segSource <- newSegmentSource size+ firstSeg <- segSource+ -- collect a ticket to save for writer CAS+ savedEmptyTkt <- readArrayElem firstSeg 0+ stream <- Stream firstSeg <$> newIORef Nothing+ let end = ChanEnd logBounds boundsMn1 segSource + <$> newCounter (startingCellOffset - 1)+ <*> newIORef (StreamHead startingCellOffset stream)+ endR@(ChanEnd _ _ _ counterR _) <- end+ endW <- end+ assert (size > 0 && (boundsMn1 + 1) == 2 ^ logBounds) $+ return ( InChan (readCounter counterR) savedEmptyTkt endW+ , OutChan endR )++-- | 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.+--+-- Writers will be blocked only when the fastest reader falls behind the+-- bounds; slower readers of duplicated 'OutChan' may fall arbitrarily behind.+dupChan :: InChan a -> IO (OutChan a)+{-# INLINE dupChan #-}+dupChan (InChan _ _ (ChanEnd logBounds boundsMn1 segSource counter streamHead)) = do+ hLoc <- readIORef streamHead+ loadLoadBarrier -- NOTE [1]+ wCount <- readCounter counter+ + counter' <- newCounter wCount + streamHead' <- newIORef hLoc+ return $ OutChan $ ChanEnd logBounds boundsMn1 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. If the chan is full this will block.+--+-- To be precise this /may/ block when the number of elements in the queue +-- @>= size@, and will certainly block when @>= size*2@, where @size@ is the+-- argument passed to 'newChan', rounded up to the next highest power of two.+--+-- /Note re. exceptions/: In the case that an async exception is raised +-- while blocking here, the write will succeed. When not blocking, exceptions+-- are masked. Thus writes always succeed once 'writeChan' is entered.+writeChan :: InChan a -> a -> IO ()+{-# INLINE writeChan #-}+writeChan c = \a-> writeChanWithBlocking True c a++writeChanWithBlocking :: Bool -> InChan a -> a -> IO ()+{-# INLINE writeChanWithBlocking #-}+writeChanWithBlocking canBlock (InChan _ savedEmptyTkt ce) a = mask_ $ do + (segIx, nextSeg, updateStreamHeadIfNecessary) <- moveToNextCell asWriter ce+ let (seg, maybeCheckpt) = case nextSeg of+ NextByWriter (Stream s _) checkpt -> (s, Just checkpt)+ -- if installed by reader, no need to check in:+ NextByReader (Stream s _) -> (s, Nothing)++ (success,nonEmptyTkt) <- casArrayElem seg segIx savedEmptyTkt (Written a)+ if success+ -- NOTE: We must only block AFTER writing to be async exception-safe.+ then maybe updateStreamHeadIfNecessary -- NOTE [2]+ ( \checkpt-> do+ unlocked <- if canBlock + then True <$ writerCheckin checkpt+ else tryWriterCheckin checkpt+ when unlocked $+ updateStreamHeadIfNecessary ) -- NOTE [1/2]+ maybeCheckpt+ + -- If CAS failed then a reader beat us, so we know we're not out of+ -- bounds and don't need to writerCheckin+ else case peekTicket nonEmptyTkt of+ Blocking v -> do putMVar v a+ updateStreamHeadIfNecessary -- NOTE [1] + Empty -> error "Stored Empty Ticket went stale!"+ Written _ -> error "Nearly Impossible! Expected Blocking"+ -- [1] At this point we know that 'seg' is unlocked for writers because a+ -- reader unblocked us, so it's safe to update the StreamHead with this+ -- segment (if we moved to a new segment). This way we maintain the invariant+ -- that the StreamHead segment is always known "unlocked" to writers.+ --+ -- [2] Similarly when in tryWriteChan we only update the stream head when+ -- we see that it was installed by reader, or we see that it was unlocked,+ -- but for the latter we check without blocking.++++-- | Try to write a value to the channel, aborting if the write is likely to+-- exceed the bounds, returning a @Bool@ indicating whether the write was+-- successful.+--+-- This function never blocks, but may occasionally write successfully to a+-- queue that is already "full". Unlike 'writeChan' this function treats the+-- requested bounds (raised to nearest power of two) strictly, rather than+-- using the @n .. n*2@ range. The more concurrent writes and reads that are+-- happening, the more inaccurate the estimate of the chan's size is likely to+-- be.+tryWriteChan :: InChan a -> a -> IO Bool+{-# INLINE tryWriteChan #-}+tryWriteChan c@(InChan readCounterReader _ (ChanEnd _ boundsMn1 _ counter _)) = \a-> do+ -- Similar caveats w/r/t counter overflow correctness as elsewhere apply+ -- here: where this would lap and give incorrect results we have already+ -- died with OOM:+ ixR <- readCounterReader+ ixW <- readCounter counter+ if ixW - ixR > boundsMn1 + then return False+ else writeChanWithBlocking False c a >> return True+++readChanOnExceptionUnmasked :: (IO a -> IO a) -> OutChan a -> IO a+{-# INLINE readChanOnExceptionUnmasked #-}+readChanOnExceptionUnmasked h = \(OutChan ce@(ChanEnd _ _ segSource _ _))-> do+ (segIx, nextSeg, updateStreamHeadIfNecessary) <- moveToNextCell asReader ce+ let (seg,next) = case nextSeg of+ NextByReader (Stream s n) -> (s,n)+ _ -> error "moveToNextCell returned a non-reader-installed next segment to readChanOnExceptionUnmasked"+ -- try to pre-allocate next segment:+ when (segIx == 0) $ void $+ waitingAdvanceStream asReader next segSource 0++ updateStreamHeadIfNecessary++ 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 :: Bool -> ChanEnd a -> IO (Int, NextSegment a, IO ())+{-# INLINE moveToNextCell #-}+moveToNextCell isReader (ChanEnd logBounds boundsMn1 segSource counter streamHead) = do+ (StreamHead offset0 str0) <- readIORef streamHead+#ifdef NOT_x86 + -- fetch-and-add is a full barrier on x86+ loadLoadBarrier+#endif+ ix <- incrCounter 1 counter+ let !relIx = ix - offset0+ !segsAway = relIx `unsafeShiftR` logBounds -- `div` bounds+ !segIx = relIx .&. boundsMn1 -- `mod` bounds+ ~nEW_SEGMENT_WAIT = (boundsMn1 `div` 12) + 25 + + go 0 nextSeg = return nextSeg+ go !n nextSeg =+ waitingAdvanceStream isReader (getNextRef nextSeg) segSource (nEW_SEGMENT_WAIT*segIx) -- NOTE [1]+ >>= go (n-1)+ + nextSeg <- assert (relIx >= 0) $+ -- go segsAway $ NextByReader str0 -- NOTE [2]+ -- NOTE: this is redundant, since `go` doesn't want to get+ -- inlined/unrolled+ if segsAway == 0 + then return $ NextByReader str0 + else go segsAway $ NextByReader str0 -- NOTE [2]++ -- writers and readers must perform this continuation at different points:+ let updateStreamHeadIfNecessary = + when (segsAway > 0) $ do+ let !offsetN = --(segsAway * bounds)+ offset0 + (segsAway `unsafeShiftL` logBounds) + writeIORef streamHead $ StreamHead offsetN $ getStr nextSeg++ return (segIx, nextSeg, updateStreamHeadIfNecessary)+ -- [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.+ -- The constant here is an approximation of the way we calculate it in+ -- Control.Concurrent.Chan.Unagi.Constants.nEW_SEGMENT_WAIT+ --+ -- [2] We start the loop with 'NextByReader' effectively meaning that the head+ -- segment was installed by a reader, or really just indicating that the+ -- writer has no need to check-in for blocking. This is always the case for+ -- the head stream; see `writeChan` NOTE 1.++++-- TODO play with inlining and look at core; we'd like the conditionals to disappear+-- INVARIANTS: +-- - if isReader, after returning, the nextSegRef will be marked NextByReader+-- - the 'nextSegRef' is only ever modified from Nothing -> Just (NextBy*)+waitingAdvanceStream :: Bool -> IORef (Maybe (NextSegment a)) -> SegSource a + -> Int -> IO (NextSegment a)+waitingAdvanceStream isReader nextSegRef segSource = go where+ cas tk = casIORef nextSegRef tk . Just++ -- extract the installed Just NextSegment from the result of the cas+ peekInstalled (_, nextSegTk) =+ fromMaybe (error "Impossible! This should only have been a Just NextBy* segment") $+ peekTicket nextSegTk++ readerUnblockAndReturn nextSeg = assert isReader $ case nextSeg of+ -- if a writer won, try to set as NextByReader so that every writer+ -- to this seg doesn't have to check in, and unblockWriters+ NextByWriter strAlreadyInstalled checkpt -> do+ unblockWriters checkpt -- idempotent+ let nextSeg' = NextByReader strAlreadyInstalled+ writeIORef nextSegRef $ Just nextSeg'+ return nextSeg'++ nextByReader -> return nextByReader++ go wait = assert (wait >= 0) $ do+ tk <- readForCAS nextSegRef+ case peekTicket tk of+ -- Rare, slow path: In readers, we outran reader 0 of the previous+ -- segment (or it was descheduled) who was tasked with setting this up+ -- In writers, there are number writer threads > bounds, or reader 0+ -- of previous segment was slow or descheduled.+ Nothing + | wait > 0 -> go (wait - 1)+ -- Create a potential next segment and try to insert it:+ | otherwise -> do + potentialStrNext <- Stream <$> segSource <*> newIORef Nothing+ if isReader+ then do+ -- This may fail because of either a competing reader or+ -- writer which certainly modified this to a Just value+ installed <- cas tk $ NextByReader potentialStrNext+#ifdef NOT_x86 + -- ensure strNext is in place before unblocking writers,+ -- where CAS is not a full barrier:+ writeBarrier+#endif+ readerUnblockAndReturn $ peekInstalled installed+ else do+ potentialCheckpt <- WriterCheckpoint <$> newEmptyMVar+ -- This may fail because of either a competing reader or+ -- writer which certainly modified this to a Just value+ peekInstalled <$> (cas tk $ + NextByWriter potentialStrNext potentialCheckpt)+ + -- Fast path: Another reader or writer has already advanced the+ -- stream. Most likely reader 0 of the last segment.+ Just nextSeg + | isReader -> readerUnblockAndReturn nextSeg+ | otherwise -> return nextSeg++type SegSource a = IO (StreamSegment a)++newSegmentSource :: Int -> IO (SegSource a)+newSegmentSource size = do+ -- NOTE: evaluate Empty seems to be required here in order to not raise+ -- "Stored Empty Ticket went stale!" exception when in GHCi.+ arr <- evaluate Empty >>= P.newArray size+ return (P.cloneMutableArray arr 0 size)+++-- This begins empty, but several readers will `put` without coordination, to+-- ensure it's filled. Meanwhile writers are blocked on a `readMVar` (see+-- writerCheckin) waiting to proceed. +newtype WriterCheckpoint = WriterCheckpoint (MVar ())++-- idempotent+unblockWriters :: WriterCheckpoint -> IO ()+unblockWriters (WriterCheckpoint v) =+ void $ tryPutMVar v ()++-- A writer knows that it doesn't need to call this when:+-- - its segment is in the StreamHead, or...+-- - its segment was reached by a NextByReader+writerCheckin :: WriterCheckpoint -> IO ()+writerCheckin (WriterCheckpoint v) = do+-- On GHC > 7.8 we have an atomic `readMVar`. On earlier GHC readMVar is+-- take+put, creating a race condition; in this case we use take+tryPut+-- ensuring the MVar stays full even if a reader's tryPut slips an () in:+#if __GLASGOW_HASKELL__ < 708+ takeMVar v >>= void . tryPutMVar v+#else+ void $ readMVar v+#endif+ -- make sure we can see the reader's segment creation once we unblock...+ loadLoadBarrier+ -- ... and proceed to readIORef the segment++-- returns immediately indicating whether the checkpt is currently unblocked.+tryWriterCheckin :: WriterCheckpoint -> IO Bool+tryWriterCheckin (WriterCheckpoint v) = do+-- On GHC > 7.8 we have an atomic `tryReadMVar`. On earlier GHC readMVar is+-- take+put, creating a race condition; in this case we use take+tryPut+-- ensuring the MVar stays full even if a reader's tryPut slips an () in:+ unblocked <- +#if __GLASGOW_HASKELL__ < 708+ tryTakeMVar v >>= maybe (return False) ((True <$) . tryPutMVar v)+#else+ tryTakeMVar v >>= maybe (return False) ((True <$) . tryPutMVar v)+ -- This is what we really want, unfortunately (and unfortunately for the+ -- hours of my life I'll never get back) this is buggy in GHC < 7.8.3:+ -- https://ghc.haskell.org/trac/ghc/ticket/9148+ --isJust <$> tryReadMVar v+#endif+ -- make sure we can see the reader's segment creation once we unblock...+ loadLoadBarrier+ return unblocked+ -- ... and proceed to readIORef the segment+
src/Control/Concurrent/Chan/Unagi/Constants.hs view
@@ -34,10 +34,6 @@ nEW_SEGMENT_WAIT :: Int nEW_SEGMENT_WAIT = round (((14.6::Float) + 0.3*fromIntegral sEGMENT_LENGTH) / 3.7) + 10 --- TODO move these into a Constants INLINABLE file, --- use in Unboxed as well--- verify by running a benchmark on consts3- lOG_SEGMENT_LENGTH :: Int lOG_SEGMENT_LENGTH = let x = 10 -- ...pre-computed from...
src/Control/Concurrent/Chan/Unagi/Internal.hs view
@@ -284,7 +284,9 @@ newSegmentSource :: IO (SegSource a) newSegmentSource = do- arr <- P.newArray sEGMENT_LENGTH Empty+ -- NOTE: evaluate Empty seems to be required here in order to not raise+ -- "Stored Empty Ticket went stale!" exception when in GHCi.+ arr <- evaluate Empty >>= P.newArray sEGMENT_LENGTH return (P.cloneMutableArray arr 0 sEGMENT_LENGTH) -- ----------
src/Utilities.hs view
@@ -71,8 +71,12 @@ -- 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+nextHighestPowerOfTwo n + | n > maxPowerOfTwo = error $ "The next power of two greater than "++(show n)++" exceeds the highest value representable by Int."+ | otherwise = + 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++ where maxPowerOfTwo = (floor $ sqrt $ (fromIntegral (maxBound :: Int)::Float)) ^ (2::Int)
− tests/Chan003.hs
@@ -1,25 +0,0 @@-module Chan003 (checkDeadlocksWriter) where--import Control.Concurrent-import qualified Control.Concurrent.Chan.Unagi as U-import Control.Exception-import Control.Monad---- OBSOLETE FOR NOW; we have more clever and careful deadlock tests in--- Deadlocks---- test for deadlocks from async exceptions raised in writer-checkDeadlocksWriter :: Int -> IO ()-checkDeadlocksWriter n = void $- replicateM_ n $ do- (i,o) <- U.newChan- wStart <- newEmptyMVar- wid <- forkIO (putMVar wStart () >> ( forever $ U.writeChan i (0::Int)) )- -- wait for writer to start- takeMVar wStart >> threadDelay 1- throwTo wid ThreadKilled- -- did killing the writer damage queue for writes or reads?- U.writeChan i (1::Int)- z <- U.readChan o- unless (z == 0) $- error "Writer never got a chance to write!"
tests/Deadlocks.hs view
@@ -6,12 +6,13 @@ import Control.Monad import Implementations+import qualified Control.Concurrent.Chan.Unagi.Bounded as UB deadlocksMain :: IO () deadlocksMain = do- let tries = 50000- + let tries = 10000+ putStrLn "===================" putStrLn "Testing Unagi:" -- ------@@ -34,6 +35,18 @@ checkDeadlocksWriter unboxedUnagiImpl tries putStrLn "OK" + putStrLn "==================="+ putStrLn "Testing Unagi.Bounded:"+ -- ------+ putStr $ " Checking for deadlocks from killed reader, x"++show tries++"... "+ -- bounds must be > 10000 here (note actual bounds rounded up to power of 2):+ checkDeadlocksReader (unagiBoundedImpl 50000) tries+ putStrLn "OK"+ -- ------+ putStr $ " Checking for deadlocks from killed writer, x"++show tries++"... "+ -- fragile bounds must be large enought to never be reached here:+ checkDeadlocksWriterBounded tries+ putStrLn "OK" -- -- Chan002.hs -- --@@ -49,7 +62,7 @@ procs <- getNumCapabilities let run _ 0 = putStrLn "" run retries n = do- when (retries > (times `div` 5)) $+ when (retries > (times `div` 3)) $ error "This test is taking too long. Please retry, and if still failing send the log to me" (i,o) <- newChan -- if we don't have at least three cores, then we need to write enough messages in first, before killing reader.@@ -59,7 +72,7 @@ takeMVar wStart >> threadDelay 1 -- wait until we're writing return $ Just wid - else do replicateM_ 10000 $ writeChan i (0::Int)+ else do replicateM_ 15000 $ writeChan i (0::Int) return Nothing rStart <- newEmptyMVar rid <- forkIO $ (putMVar rStart () >> (forever $ void $ readChan o))@@ -101,3 +114,31 @@ z <- readChan o unless (z == 0) $ error "Writer never got a chance to write!"++-- A bit ugly, but we need this slight variant for Bounded variant:+checkDeadlocksWriterBounded :: Int -> IO ()+checkDeadlocksWriterBounded cnt = go 0 cnt where+ go lates n + | lates > (cnt `div` 4) = error "This is taking too long; we probably need a bigger bounds, sorry." + | otherwise = + when (n > 0) $ do+ (i,o) <- UB.newChan (2^(14::Int))+ wStart <- newEmptyMVar+ wid <- forkIO (putMVar wStart () >> ( forever $ UB.writeChan i (0::Int)) )+ -- wait for writer to start+ takeMVar wStart >> threadDelay 1+ throwTo wid ThreadKilled+ -- did killing the writer damage queue for writes or reads?+ success <- UB.tryWriteChan i (1::Int)+ if success+ then do+ z <- UB.readChan o+ if (z /= 0)+ -- Writer never got a chance to write, retry:+ then go (lates+1) n+ -- OK:+ else go lates (n-1)++ -- throwTo probably didn't catch writeChan while running, retry:+ else go (lates+1) n+
tests/DupChan.hs view
@@ -33,6 +33,19 @@ replicateM_ 1000 $ dupChanTest2 unboxedUnagiImpl 10000 putStrLn "OK" + putStrLn "==================="+ putStrLn "Test dupChan Unagi.Bounded"+ -- NOTE: n must be <= bounds in dupChanTest1:+ forM_ [(4096,4096),(65536,50000),(4,2)] $ \(bounds, n)-> do+ -- ------+ putStr $ " Reader/Reader with bounds "++(show bounds)++"... "+ replicateM_ 1000 $ dupChanTest1 (unagiBoundedImpl bounds) n+ putStrLn "OK"+ forM_ [2, 1024, 65536] $ \bounds-> do+ putStr $ " Writer/dupChan+Reader with bounds "++(show bounds)++"... "+ replicateM_ 1000 $ dupChanTest2 (unagiBoundedImpl bounds) 10000+ putStrLn "OK"+ -- Check output where dupChan at known point in input stream, with two -- concurrent readers. dupChanTest1 :: Implementation inc outc Int -> Int -> IO ()
tests/Implementations.hs view
@@ -2,6 +2,7 @@ import qualified Control.Concurrent.Chan.Unagi as U import qualified Control.Concurrent.Chan.Unagi.Unboxed as UU+import qualified Control.Concurrent.Chan.Unagi.Bounded as UB import qualified Data.Primitive as P type Implementation inc outc a = (IO (inc a, outc a), inc a -> a -> IO (), outc a -> IO a, inc a -> IO (outc a))@@ -12,3 +13,5 @@ unboxedUnagiImpl :: (P.Prim a)=> Implementation UU.InChan UU.OutChan a unboxedUnagiImpl = (UU.newChan, UU.writeChan, UU.readChan, UU.dupChan) +unagiBoundedImpl :: Int -> Implementation UB.InChan UB.OutChan a+unagiBoundedImpl n = (UB.newChan n, UB.writeChan, UB.readChan, UB.dupChan)
tests/Main.hs view
@@ -13,6 +13,7 @@ -- implementation-specific tests: import Unagi import UnagiUnboxed+import UnagiBounded -- Other import Atomics@@ -20,6 +21,7 @@ main :: IO () main = do + -- Make sure testing environment is sane: assertionsWorking <- try $ assert False $ return () case assertionsWorking of Left (AssertionFailed _) -> putStrLn "Assertions: On"@@ -32,7 +34,6 @@ hSetBuffering stdout NoBuffering -- ------------------------------------ -- test important properties of our atomic-primops: atomicsMain @@ -47,6 +48,9 @@ -- check for deadlocks: deadlocksMain - -- unagi-specific tests+ -- implementation-specific tests unagiMain unagiUnboxedMain+ unagiBoundedMain++ putStrLn "ALL DONE!"
tests/Smoke.hs view
@@ -2,14 +2,25 @@ module Smoke (smokeMain) where import Control.Monad-import Control.Concurrent(forkIO)+import Control.Concurrent(forkIO,threadDelay) import qualified Control.Concurrent.Chan as C import Data.List+import Control.Exception+import qualified Control.Exception as E import Implementations +-- TODO This is real lame, probably just use async+lgErrs :: Bool -> String -> IO () -> IO ()+lgErrs expectingBlock nm = E.handle $ \e-> + let lg = putStrLn $ "!!! EXCEPTION IN "++nm++": "++(show e) + in case E.fromException e of+ Just BlockedIndefinitelyOnMVar -> when (not expectingBlock) lg+ Nothing -> lg+ + smokeMain :: IO ()-smokeMain = do+smokeMain = (do putStrLn "===================" putStrLn "Testing Unagi:" -- ------@@ -19,6 +30,7 @@ -- ------ testContention unagiImpl 2 2 1000000 + putStrLn "===================" putStrLn "Testing Unagi.Unboxed:" -- ------@@ -29,10 +41,23 @@ testContention unboxedUnagiImpl 2 2 1000000 + forM_ [1, 2, 4, 1024] $ \bounds-> do+ putStrLn "==================="+ putStrLn $ "Testing Unagi.Bounded with bounds "++(show bounds)+ -- ------+ putStr " FIFO smoke test... "+ fifoSmoke (unagiBoundedImpl bounds) 100000+ putStrLn "OK"+ -- ------+ testContention (unagiBoundedImpl bounds) 2 2 1000000++ ) `onException` (threadDelay 1000000) -- wait for lgErrs+ fifoSmoke :: Implementation inc outc Int -> Int -> IO () fifoSmoke (newChan,writeChan,readChan,_) n = do (i,o) <- newChan- mapM_ (writeChan i) [1..n]+ -- we need to fork this for Unagi.Bounded:+ void $ forkIO $ lgErrs False "fifoSmoke writeChan " $ mapM_ (writeChan i) [1..n] nsOut <- replicateM n $ readChan o unless (nsOut == [1..n]) $ error "Cough!"@@ -47,19 +72,19 @@ (i,o) <- newChan -- some will get blocked indefinitely:- void $ replicateM readers $ forkIO $ forever $+ void $ replicateM readers $ forkIO $ lgErrs True "testContention readChan o"$ forever $ readChan o >>= C.writeChan out - putStrLn $ "Sending "++(show $ length $ concat groups)++" messages, with "++(show readers)++" readers and "++(show writers)++" writers."- mapM_ (forkIO . mapM_ (writeChan i)) groups+ putStr $ " Sending "++(show $ length $ concat groups)++" messages, with "++(show readers)++" readers and "++(show writers)++" writers.... "+ mapM_ (forkIO . lgErrs False "testContention writeChan i " . mapM_ (writeChan i)) groups ns <- replicateM nNice (C.readChan out) isEmpty <- C.isEmptyChan out if sort ns == [1..nNice] && isEmpty then let d = interleaving ns- in if d < 0.75- then putStrLn $ "Not enough interleaving of threads: "++(show $ d)++". Please try again or report a bug"- else putStrLn $ "Success, with interleaving pct of "++(show $ d)++" (closer to 1 means we have higher confidence in the test)."+ in if d < 0.7 -- arbitrary+ then putStrLn $ "OK, BUT WARNING: low interleaving of threads: "++(show $ d)+ else putStrLn $ "OK" --, with interleaving pct of "++(show $ d)++" (closer to 1 means we have higher confidence in the test)." else error "What we put in isn't what we got out :(" -- --------- Helpers:
tests/Unagi.hs view
@@ -31,11 +31,12 @@ mapM_ correctInitialWrites [ (maxBound - UI.sEGMENT_LENGTH), (maxBound - UI.sEGMENT_LENGTH) - 1, maxBound, minBound, 0] putStrLn "OK" -- ------- let tries = 50000+ let tries = 10000 putStrLn $ "Checking for deadlocks from killed Unagi reader in a fancy way, x"++show tries checkDeadlocksReaderUnagi tries +-- TODO CONSIDER ADDING newChanStarting (or raplcing newChan) TO IMPLEMENTATIONS, AND CONSOLIDATE THESE IN Smoke.hs smoke :: Int -> IO () smoke n = smoke1 n >> smoke2 n @@ -135,7 +136,7 @@ checkDeadlocksReaderUnagi times = do let run 0 normalRetries numRace = putStrLn $ "Lates: "++(show normalRetries)++", Races: "++(show numRace) run n normalRetries numRace- | (normalRetries + numRace) > (times `div` 5) = error "This test is taking too long. Please retry, and if still failing send the log to me"+ | (normalRetries + numRace) > (times `div` 3) = error "This test is taking too long. Please retry, and if still failing send the log to me" | otherwise = do -- we'll kill the reader with our special exception half the time, -- expecting that we never get our race condition on those runs:
tests/UnagiUnboxed.hs view
@@ -34,7 +34,7 @@ mapM_ correctInitialWrites [ (maxBound - UI.sEGMENT_LENGTH), (maxBound - UI.sEGMENT_LENGTH) - 1, maxBound, minBound, 0] putStrLn "OK" -- ------- let tries = 50000+ let tries = 10000 putStrLn $ "Checking for deadlocks from killed Unagi reader in a fancy way, x"++show tries checkDeadlocksReaderUnagi tries @@ -137,7 +137,7 @@ checkDeadlocksReaderUnagi times = do let run 0 normalRetries numRace = putStrLn $ "Lates: "++(show normalRetries)++", Races: "++(show numRace) run n normalRetries numRace- | (normalRetries + numRace) > (times `div` 5) = error "This test is taking too long. Please retry, and if still failing send the log to me"+ | (normalRetries + numRace) > (times `div` 3) = error "This test is taking too long. Please retry, and if still failing send the log to me" | otherwise = do -- we'll kill the reader with our special exception half the time, -- expecting that we never get our race condition on those runs:
unagi-chan.cabal view
@@ -1,5 +1,5 @@ name: unagi-chan-version: 0.1.1.0+version: 0.2.0.0 synopsis: Fast and scalable concurrent queues for x86, with a Chan-like API @@ -13,13 +13,10 @@ Here is an example benchmark measuring the time taken to concurrently write and read 100,000 messages, with work divided amongst increasing number of readers and writers, comparing against the top-performing queues in the- standard libraries. Scale is milliseconds.- .- <<http://i.imgur.com/safKkCP.png>>- .- And here is a view on just the unagi implementations.+ standard libraries, with an inset graph showing a zoomed-in view on the+ implementations here. .- <<http://i.imgur.com/K6s2pXj.png>>+ <<http://i.imgur.com/J5rLUFn.png>> . license: BSD3@@ -32,6 +29,7 @@ -- currently uploaded to imgur; move to this eventually --extra-doc-files: images/*.png --cabal-version: >=1.18+extra-source-files: CHANGELOG.markdown source-repository head type: git@@ -42,9 +40,11 @@ hs-source-dirs: src exposed-modules: Control.Concurrent.Chan.Unagi , Control.Concurrent.Chan.Unagi.Unboxed+ , Control.Concurrent.Chan.Unagi.Bounded other-modules: Control.Concurrent.Chan.Unagi.Internal , Control.Concurrent.Chan.Unagi.Unboxed.Internal+ , Control.Concurrent.Chan.Unagi.Bounded.Internal , Control.Concurrent.Chan.Unagi.Constants , Utilities , Data.Atomics.Counter.Fat@@ -60,13 +60,12 @@ -- We'll need some additional barriers for correctness: if !arch(i386) && !arch(x86_64) cpp-options: -DNOT_x86+ -- TODO -- - Do a benchmark of multiple queues running in parallel, to see if we are -- affected by global allocator issues with pinned memory: -- http://thread.gmane.org/gmane.comp.lang.haskell.parallel/218--- - On next benchmarks run, cut out "Demo with messages..with" and make unagi --- view overlayed with drop shadow -- -- Potential implementations roadmap (or we might just stick with this design -- for this package):@@ -95,7 +94,6 @@ main-is: Main.hs other-modules: Atomics- , Chan003 , Deadlocks , DupChan , Implementations