diff --git a/CHANGELOG.markdown b/CHANGELOG.markdown new file mode 100644 --- /dev/null +++ b/CHANGELOG.markdown @@ -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 diff --git a/benchmarks/multi.hs b/benchmarks/multi.hs --- a/benchmarks/multi.hs +++ b/benchmarks/multi.hs @@ -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 diff --git a/benchmarks/single.hs b/benchmarks/single.hs --- a/benchmarks/single.hs +++ b/benchmarks/single.hs @@ -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 diff --git a/core-example/Main.hs b/core-example/Main.hs --- a/core-example/Main.hs +++ b/core-example/Main.hs @@ -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 () diff --git a/src/Control/Concurrent/Chan/Unagi.hs b/src/Control/Concurrent/Chan/Unagi.hs --- a/src/Control/Concurrent/Chan/Unagi.hs +++ b/src/Control/Concurrent/Chan/Unagi.hs @@ -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 diff --git a/src/Control/Concurrent/Chan/Unagi/Bounded.hs b/src/Control/Concurrent/Chan/Unagi/Bounded.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Chan/Unagi/Bounded.hs @@ -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) diff --git a/src/Control/Concurrent/Chan/Unagi/Bounded/Internal.hs b/src/Control/Concurrent/Chan/Unagi/Bounded/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Chan/Unagi/Bounded/Internal.hs @@ -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 + diff --git a/src/Control/Concurrent/Chan/Unagi/Constants.hs b/src/Control/Concurrent/Chan/Unagi/Constants.hs --- a/src/Control/Concurrent/Chan/Unagi/Constants.hs +++ b/src/Control/Concurrent/Chan/Unagi/Constants.hs @@ -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... diff --git a/src/Control/Concurrent/Chan/Unagi/Internal.hs b/src/Control/Concurrent/Chan/Unagi/Internal.hs --- a/src/Control/Concurrent/Chan/Unagi/Internal.hs +++ b/src/Control/Concurrent/Chan/Unagi/Internal.hs @@ -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) -- ---------- diff --git a/src/Utilities.hs b/src/Utilities.hs --- a/src/Utilities.hs +++ b/src/Utilities.hs @@ -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) diff --git a/tests/Chan003.hs b/tests/Chan003.hs deleted file mode 100644 --- a/tests/Chan003.hs +++ /dev/null @@ -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!" diff --git a/tests/Deadlocks.hs b/tests/Deadlocks.hs --- a/tests/Deadlocks.hs +++ b/tests/Deadlocks.hs @@ -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 + diff --git a/tests/DupChan.hs b/tests/DupChan.hs --- a/tests/DupChan.hs +++ b/tests/DupChan.hs @@ -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 () diff --git a/tests/Implementations.hs b/tests/Implementations.hs --- a/tests/Implementations.hs +++ b/tests/Implementations.hs @@ -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) diff --git a/tests/Main.hs b/tests/Main.hs --- a/tests/Main.hs +++ b/tests/Main.hs @@ -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!" diff --git a/tests/Smoke.hs b/tests/Smoke.hs --- a/tests/Smoke.hs +++ b/tests/Smoke.hs @@ -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: diff --git a/tests/Unagi.hs b/tests/Unagi.hs --- a/tests/Unagi.hs +++ b/tests/Unagi.hs @@ -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: diff --git a/tests/UnagiUnboxed.hs b/tests/UnagiUnboxed.hs --- a/tests/UnagiUnboxed.hs +++ b/tests/UnagiUnboxed.hs @@ -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: diff --git a/unagi-chan.cabal b/unagi-chan.cabal --- a/unagi-chan.cabal +++ b/unagi-chan.cabal @@ -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. - . - <> - . - 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. . - <> + <> . 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