diff --git a/CHANGELOG.markdown b/CHANGELOG.markdown --- a/CHANGELOG.markdown +++ b/CHANGELOG.markdown @@ -12,3 +12,13 @@ - conditionally use tryReadMVar (as before) when GHC >= 7.8.3 - set proper CPP flags when running tests + +### 0.3.0.0 + +- fixed build on GHC 7.6 (thanks @Noeda) +- `Unagi.Unboxed` is now polymorphic in a new `UnagiPrim` class, which permits an optimization; defined instances are the same +- add new NoBlocking variants with reads that don't block, omiting some overhead + - these have a new `Stream` interface for reads with even lower overhead +- revisited memory barriers in light of https://github.com/rrnewton/haskell-lockfree/issues/39, and document them better +- Added `tryReadChan` functions to all variants +- get rid of upper bounds on `atomic-primops` diff --git a/benchmarks/multi.hs b/benchmarks/multi.hs --- a/benchmarks/multi.hs +++ b/benchmarks/multi.hs @@ -4,6 +4,8 @@ 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.Unagi.NoBlocking as UN +import qualified Control.Concurrent.Chan.Unagi.NoBlocking as UNU #ifdef COMPARE_BENCHMARKS import Control.Concurrent.Chan import Control.Concurrent.STM @@ -59,6 +61,26 @@ , 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 ] + , bgroup "unagi-chan Unagi.NoBlocking" $ + [ bench "async 1 writers 1 readers" $ nfIO $ asyncReadsWritesUnagiNoBlocking 1 1 n + , bench "oversubscribing: async 100 writers 100 readers" $ nfIO $ asyncReadsWritesUnagiNoBlocking 100 100 n + , bench "async Int writer, main thread read and sum" $ nfIO $ asyncSumIntUnagiNoBlocking n + ] + , bgroup "unagi-chan Unagi.NoBlocking Stream" $ + [ bench "async 1 writers 1 readers" $ nfIO $ asyncReadsWritesUnagiNoBlockingStream 1 1 n + , bench "oversubscribing: async 100 writers 100 readers" $ nfIO $ asyncReadsWritesUnagiNoBlockingStream 100 100 n + , bench "async Int writer, main thread read and sum" $ nfIO $ asyncSumIntUnagiNoBlockingStream n + ] + , bgroup "unagi-chan Unagi.NoBlocking.Unboxed" $ + [ bench "async 1 writers 1 readers" $ nfIO $ asyncReadsWritesUnagiNoBlockingUnboxed 1 1 n + , bench "oversubscribing: async 100 writers 100 readers" $ nfIO $ asyncReadsWritesUnagiNoBlockingUnboxed 100 100 n + , bench "async Int writer, main thread read and sum" $ nfIO $ asyncSumIntUnagiNoBlockingUnboxed n + ] + , bgroup "unagi-chan Unagi.NoBlocking.Unboxed Stream" $ + [ bench "async 1 writers 1 readers" $ nfIO $ asyncReadsWritesUnagiNoBlockingUnboxedStream 1 1 n + , bench "oversubscribing: async 100 writers 100 readers" $ nfIO $ asyncReadsWritesUnagiNoBlockingUnboxedStream 100 100 n + , bench "async Int writer, main thread read and sum" $ nfIO $ asyncSumIntUnagiNoBlockingUnboxedStream n + ] #ifdef COMPARE_BENCHMARKS , bgroup "Chan" $ [ bench "async 1 writer 1 readers" $ nfIO $ asyncReadsWritesChan 1 1 n @@ -94,6 +116,14 @@ 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 "Unagi.NoBlocking" $ + map (\c-> benchRun c $ asyncReadsWritesUnagiNoBlocking c c n) runs + , bgroup "Unagi.NoBlocking Stream" $ + map (\c-> benchRun c $ asyncReadsWritesUnagiNoBlockingStream c c n) runs + , bgroup "Unagi.NoBlocking.Unboxed" $ + map (\c-> benchRun c $ asyncReadsWritesUnagiNoBlockingUnboxed c c n) runs + , bgroup "Unagi.NoBlocking.Unboxed Stream" $ + map (\c-> benchRun c $ asyncReadsWritesUnagiNoBlockingUnboxedStream c c n) runs , bgroup "TQueue " $ map (\c-> benchRun c $ asyncReadsWritesTQueue c c n) runs , bgroup "Chan " $ @@ -123,8 +153,134 @@ _ <- async $ mapM_ (U.writeChan i) [1..n] -- NOTE: partially-applied writeChan readerSum n 0 +-- ------------------------- +-- NoBlocking variant: +asyncReadsWritesUnagiNoBlocking :: Int -> Int -> Int -> IO () +asyncReadsWritesUnagiNoBlocking writers readers n = do + -- A fairly reasonable heuristic: yield if we're oversubscribed, else do threadDelay: + procs <- getNumCapabilities + let pause = if (readers+writers) > procs then yield else threadDelay 1 + let nNice = n - rem n (lcm writers readers) + (i,o) <- UN.newChan + rcvrs <- replicateM readers $ async $ + replicateM_ (nNice `quot` readers) $ + UN.readChan pause o + _ <- replicateM writers $ async $ replicateM_ (nNice `quot` writers) $ UN.writeChan i () + mapM_ wait rcvrs +-- A slightly more realistic benchmark, lets us see effects of unboxed strict +-- in value, and inlining effects w/ partially applied writeChan +asyncSumIntUnagiNoBlocking :: Int -> IO Int +asyncSumIntUnagiNoBlocking n = do + (i,o) <- UN.newChan + let readerSum 0 !tot = return tot + readerSum !n' !tot = UN.readChan (threadDelay 1) o + >>= readerSum (n'-1) . (tot+) + _ <- async $ mapM_ (UN.writeChan i) [1..n] -- NOTE: partially-applied writeChan + readerSum n 0 + +-- Unagi.NoBlocking Stream interface: +asyncReadsWritesUnagiNoBlockingStream :: Int -> Int -> Int -> IO () +asyncReadsWritesUnagiNoBlockingStream writers readers n = do + -- A fairly reasonable heuristic: yield if we're oversubscribed, else do threadDelay: + procs <- getNumCapabilities + let pause = if (readers+writers) > procs then yield else threadDelay 1 + + let nNice = n - rem n (lcm writers readers) + (i,o) <- UN.newChan + strms <- UN.streamChan readers o + let doReads x str = when (x > 0) $ do + cns <- UN.tryReadNext str + case cns of + UN.Pending -> pause >> doReads x str + UN.Next _ str' -> doReads (x-1) str' + rcvrs <- mapM (async . doReads (nNice `quot` readers)) strms + _ <- replicateM writers $ async $ replicateM_ (nNice `quot` writers) $ UN.writeChan i () + mapM_ wait rcvrs + +-- A slightly more realistic benchmark, lets us see effects of unboxed strict +-- in value, and inlining effects w/ partially applied writeChan +asyncSumIntUnagiNoBlockingStream :: Int -> IO Int +asyncSumIntUnagiNoBlockingStream n = do + (i,o) <- UN.newChan + [ str0 ] <- UN.streamChan 1 o + let readerSum 0 !tot _ = return tot + readerSum !n' !tot str = do + cns <- UN.tryReadNext str + case cns of + UN.Pending -> threadDelay 1 >> readerSum n' tot str + UN.Next val str' -> readerSum (n'-1) (tot+val) str' + _ <- async $ mapM_ (UN.writeChan i) [1..n] -- NOTE: partially-applied writeChan + readerSum n 0 str0 + + + + +-- ------------------------- +-- NoBlocking.Unboxed variant: +asyncReadsWritesUnagiNoBlockingUnboxed :: Int -> Int -> Int -> IO () +asyncReadsWritesUnagiNoBlockingUnboxed writers readers n = do + -- A fairly reasonable heuristic: yield if we're oversubscribed, else do threadDelay: + procs <- getNumCapabilities + let pause = if (readers+writers) > procs then yield else threadDelay 1 + + let nNice = n - rem n (lcm writers readers) + (i,o) <- UNU.newChan + rcvrs <- replicateM readers $ async $ + replicateM_ (nNice `quot` readers) $ + UNU.readChan pause o + _ <- replicateM writers $ async $ replicateM_ (nNice `quot` writers) $ UNU.writeChan i () + mapM_ wait rcvrs + +-- A slightly more realistic benchmark, lets us see effects of unboxed strict +-- in value, and inlining effects w/ partially applied writeChan +asyncSumIntUnagiNoBlockingUnboxed :: Int -> IO Int +asyncSumIntUnagiNoBlockingUnboxed n = do + (i,o) <- UNU.newChan + let readerSum 0 !tot = return tot + readerSum !n' !tot = UNU.readChan (threadDelay 1) o + >>= readerSum (n'-1) . (tot+) + _ <- async $ mapM_ (UNU.writeChan i) [1..n] -- NOTE: partially-applied writeChan + readerSum n 0 + +-- Unagi.NoBlocking.Unboxed Stream interface: +asyncReadsWritesUnagiNoBlockingUnboxedStream :: Int -> Int -> Int -> IO () +asyncReadsWritesUnagiNoBlockingUnboxedStream writers readers n = do + -- A fairly reasonable heuristic: yield if we're oversubscribed, else do threadDelay: + procs <- getNumCapabilities + let pause = if (readers+writers) > procs then yield else threadDelay 1 + + let nNice = n - rem n (lcm writers readers) + (i,o) <- UNU.newChan + strms <- UNU.streamChan readers o + let doReads x str = when (x > 0) $ do + cns <- UNU.tryReadNext str + case cns of + UNU.Pending -> pause >> doReads x str + UNU.Next _ str' -> doReads (x-1) str' + rcvrs <- mapM (async . doReads (nNice `quot` readers)) strms + _ <- replicateM writers $ async $ replicateM_ (nNice `quot` writers) $ UNU.writeChan i () + mapM_ wait rcvrs + +-- A slightly more realistic benchmark, lets us see effects of unboxed strict +-- in value, and inlining effects w/ partially applied writeChan +asyncSumIntUnagiNoBlockingUnboxedStream :: Int -> IO Int +asyncSumIntUnagiNoBlockingUnboxedStream n = do + (i,o) <- UNU.newChan + [ str0 ] <- UNU.streamChan 1 o + let readerSum 0 !tot _ = return tot + readerSum !n' !tot str = do + cns <- UNU.tryReadNext str + case cns of + UNU.Pending -> threadDelay 1 >> readerSum n' tot str + UNU.Next val str' -> readerSum (n'-1) (tot+val) str' + _ <- async $ mapM_ (UNU.writeChan i) [1..n] -- NOTE: partially-applied writeChan + readerSum n 0 str0 + + + +-- ------------------------- -- Unboxed Unagi: -- NOTE: using Int here instead of (). TODO change others so we can properly compare? asyncReadsWritesUnagiUnboxed :: Int -> Int -> Int -> IO () @@ -144,6 +300,7 @@ 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 () diff --git a/benchmarks/single.hs b/benchmarks/single.hs --- a/benchmarks/single.hs +++ b/benchmarks/single.hs @@ -5,6 +5,8 @@ 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.Unagi.NoBlocking as UN +import qualified Control.Concurrent.Chan.Unagi.NoBlocking.Unboxed as UNU #ifdef COMPARE_BENCHMARKS import Control.Concurrent.Chan import Control.Concurrent.STM @@ -27,6 +29,8 @@ (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 + (fastEmptyUNI,fastEmptyUNO) <- UN.newChan + (fastEmptyUNUI,fastEmptyUNUO) <- UNU.newChan #ifdef COMPARE_BENCHMARKS chanEmpty <- newChan tqueueEmpty <- newTQueueIO @@ -42,6 +46,8 @@ , 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? + , bench "unagi-chan Unagi.NoBlocking" $ nfIO (UN.writeChan fastEmptyUNI (0::Int) >> tryReadChanErrUN fastEmptyUNO) -- TODO comparing Int writing to (). Change? + , bench "unagi-chan Unagi.NoBlocking.Unboxed" $ nfIO (UNU.writeChan fastEmptyUNUI (0::Int) >> tryReadChanErrUNU fastEmptyUNUO) -- TODO comparing Int writing to (). Change? #ifdef COMPARE_BENCHMARKS , bench "Chan" $ nfIO $ (writeChan chanEmpty () >> readChan chanEmpty) , bench "TQueue" $ nfIO $ (atomically (writeTQueue tqueueEmpty () >> readTQueue tqueueEmpty)) @@ -58,6 +64,10 @@ [ bench "unagi-chan Unagi" $ nfIO $ runtestSplitChanU1 n , bench "unagi-chan Unagi.Unboxed" $ nfIO $ runtestSplitChanUU1 n , bench "unagi-chan Unagi.Bounded" $ nfIO $ runtestSplitChanUB1 n + , bench "unagi-chan Unagi.NoBlocking" $ nfIO $ runtestSplitChanUN1 n + , bench "unagi-chan Unagi.NoBlocking Stream" $ nfIO $ runtestSplitChanUNStream1 n + , bench "unagi-chan Unagi.NoBlocking.Unboxed" $ nfIO $ runtestSplitChanUNU1 n + , bench "unagi-chan Unagi.NoBlocking.Unboxed Stream" $ nfIO $ runtestSplitChanUNUStream1 n #ifdef COMPARE_BENCHMARKS , bench "Chan" $ nfIO $ runtestChan1 n , bench "TQueue" $ nfIO $ runtestTQueue1 n @@ -69,6 +79,10 @@ [ bench "unagi-chan Unagi" $ nfIO $ runtestSplitChanU2 n , bench "unagi-chan Unagi.Unboxed" $ nfIO $ runtestSplitChanUU2 n , bench "unagi-chan Unagi.Bounded" $ nfIO $ runtestSplitChanUB2 n + , bench "unagi-chan Unagi.NoBlocking" $ nfIO $ runtestSplitChanUN2 n + , bench "unagi-chan Unagi.NoBlocking Stream" $ nfIO $ runtestSplitChanUNStream2 n + , bench "unagi-chan Unagi.NoBlocking.Unboxed" $ nfIO $ runtestSplitChanUNU2 n + , bench "unagi-chan Unagi.NoBlocking.Unboxed Stream" $ nfIO $ runtestSplitChanUNUStream2 n #ifdef COMPARE_BENCHMARKS , bench "Chan" $ nfIO $ runtestChan2 n , bench "TQueue" $ nfIO $ runtestTQueue2 n @@ -79,6 +93,21 @@ ] ] + +-- Helper for when we know a read should succeed immediately: +tryReadChanErrUN :: UN.OutChan a -> IO a +{-# INLINE tryReadChanErrUN #-} +tryReadChanErrUN oc = UN.tryReadChan oc + >>= UN.tryRead + >>= maybe (error "A read we expected to succeed failed!") return +tryReadChanErrUNU :: UNU.UnagiPrim a=> UNU.OutChan a -> IO a +{-# INLINE tryReadChanErrUNU #-} +tryReadChanErrUNU oc = UNU.tryReadChan oc + >>= UNU.tryRead + >>= maybe (error "A read we expected to succeed failed!") return + + + -- unagi-chan Unagi -- runtestSplitChanU1, runtestSplitChanU2 :: Int -> IO () runtestSplitChanU1 n = do @@ -92,6 +121,94 @@ replicateM_ 1000 $ do replicateM_ n1000 $ U.writeChan i () replicateM_ n1000 $ U.readChan o + +-- unagi-chan Unagi.NoBlocking -- +runtestSplitChanUN1, runtestSplitChanUN2 :: Int -> IO () +runtestSplitChanUN1 n = do + (i,o) <- UN.newChan + replicateM_ n $ UN.writeChan i () + replicateM_ n $ tryReadChanErrUN o + +runtestSplitChanUN2 n = do + (i,o) <- UN.newChan + let n1000 = n `quot` 1000 + replicateM_ 1000 $ do + replicateM_ n1000 $ UN.writeChan i () + replicateM_ n1000 $ tryReadChanErrUN o + +-- unagi-chan Unagi.NoBlocking Stream -- +runtestSplitChanUNStream1, runtestSplitChanUNStream2 :: Int -> IO () +runtestSplitChanUNStream1 n = do + (i,o) <- UN.newChan + [ oStream ] <- UN.streamChan 1 o + replicateM_ n $ UN.writeChan i () + -- consume until we hit empty: + let eat str = do + x <- UN.tryReadNext str + case x of + UN.Pending -> return () + UN.Next _ str' -> eat str' + eat oStream + +runtestSplitChanUNStream2 n = do + (i,o) <- UN.newChan + [ oStream ] <- UN.streamChan 1 o + let n1000 = n `quot` 1000 + let eat str = do + x <- UN.tryReadNext str + case x of + UN.Pending -> return str + UN.Next _ str' -> eat str' + writeAndEat iter str = unless (iter <=0) $ do + replicateM_ n1000 $ UN.writeChan i () + eat str >>= writeAndEat (iter-1) + + writeAndEat (1000::Int) oStream + + +-- unagi-chan Unagi.NoBlocking.Unboxed -- +runtestSplitChanUNU1, runtestSplitChanUNU2 :: Int -> IO () +runtestSplitChanUNU1 n = do + (i,o) <- UNU.newChan + replicateM_ n $ UNU.writeChan i (0::Int) + replicateM_ n $ tryReadChanErrUNU o + +runtestSplitChanUNU2 n = do + (i,o) <- UNU.newChan + let n1000 = n `quot` 1000 + replicateM_ 1000 $ do + replicateM_ n1000 $ UNU.writeChan i (0::Int) + replicateM_ n1000 $ tryReadChanErrUNU o + +-- unagi-chan Unagi.NoBlocking Stream -- +runtestSplitChanUNUStream1, runtestSplitChanUNUStream2 :: Int -> IO () +runtestSplitChanUNUStream1 n = do + (i,o) <- UNU.newChan + [ oStream ] <- UNU.streamChan 1 o + replicateM_ n $ UNU.writeChan i (0::Int) + -- consume until we hit empty: + let eat str = do + x <- UNU.tryReadNext str + case x of + UNU.Pending -> return () + UNU.Next _ str' -> eat str' + eat oStream + +runtestSplitChanUNUStream2 n = do + (i,o) <- UNU.newChan + [ oStream ] <- UNU.streamChan 1 o + let n1000 = n `quot` 1000 + let eat str = do + x <- UNU.tryReadNext str + case x of + UNU.Pending -> return str + UNU.Next _ str' -> eat str' + writeAndEat iter str = unless (iter <=0) $ do + replicateM_ n1000 $ UNU.writeChan i (0::Int) + eat str >>= writeAndEat (iter-1) + + writeAndEat (1000::Int) oStream + -- unagi-chan Unagi Unboxed -- diff --git a/core-example/Main.hs b/core-example/Main.hs --- a/core-example/Main.hs +++ b/core-example/Main.hs @@ -7,6 +7,9 @@ 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.Unagi.NoBlocking as UN +import qualified Control.Concurrent.Chan.Unagi.NoBlocking.Unboxed as UNU + import qualified Control.Concurrent.Chan as C import qualified Control.Concurrent.STM.TQueue as S import Control.Concurrent.STM @@ -30,10 +33,12 @@ main = do [n] <- getArgs - -- runU (read n) + runU (read n) -- runUU (read n) - runUB (read n) -{- + -- runUB (read n) + -- runUN (read n) + -- runUNStream (read n) + -- runUNUStream (read n) runU :: Int -> IO () runU n = do (i,o) <- U.newChan @@ -41,7 +46,6 @@ replicateM_ 1000 $ do replicateM_ n1000 $ U.writeChan i () replicateM_ n1000 $ U.readChan o - -} {- runUU :: Int -> IO () runUU n = do @@ -50,7 +54,6 @@ replicateM_ 1000 $ do replicateM_ n1000 $ UU.writeChan i (0::Int) replicateM_ n1000 $ UU.readChan o - -} runUB :: Int -> IO () runUB n = do @@ -59,6 +62,63 @@ replicateM_ 1000 $ do replicateM_ n1000 $ UB.writeChan i (0::Int) replicateM_ n1000 $ UB.readChan o + + +tryReadChanErrUN :: UN.OutChan a -> IO a +{-# INLINE tryReadChanErrUN #-} +tryReadChanErrUN oc = UN.tryReadChan oc + >>= UN.tryRead + >>= maybe (error "A read we expected to succeed failed!") return + +runUN n = do + (i,o) <- UN.newChan + let n1000 = n `quot` 1000 + replicateM_ 1000 $ do + replicateM_ n1000 $ UN.writeChan i () + replicateM_ n1000 $ tryReadChanErrUN o + +runUNStream n = do + (i,o) <- UN.newChan + [ oStream ] <- UN.streamChan 1 o + let n1000 = n `quot` 1000 + let eat str = do + x <- UN.tryReadNext str + case x of + UN.Pending -> return str + UN.Next _ str' -> eat str' + writeAndEat iter str = unless (iter <=0) $ do + replicateM_ n1000 $ UN.writeChan i () + eat str >>= writeAndEat (iter-1) + + writeAndEat (1000::Int) oStream + -} +tryReadChanErrUNU :: UNU.UnagiPrim a=> UNU.OutChan a -> IO a +{-# INLINE tryReadChanErrUNU #-} +tryReadChanErrUNU oc = UNU.tryReadChan oc + >>= UNU.tryRead + >>= maybe (error "A read we expected to succeed failed!") return + +runUNU n = do + (i,o) <- UNU.newChan + let n1000 = n `quot` 1000 + replicateM_ 1000 $ do + replicateM_ n1000 $ UNU.writeChan i (0::Int) + replicateM_ n1000 $ tryReadChanErrUNU o + +runUNUStream n = do + (i,o) <- UNU.newChan + [ oStream ] <- UNU.streamChan 1 o + let n1000 = n `quot` 1000 + let eat str = do + x <- UNU.tryReadNext str + case x of + UNU.Pending -> return str + UNU.Next _ str' -> eat str' + writeAndEat iter str = unless (iter <=0) $ do + replicateM_ n1000 $ UNU.writeChan i (0::Int) + eat str >>= writeAndEat (iter-1) + + writeAndEat (1000::Int) oStream {- 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 @@ -2,8 +2,10 @@ {- | 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. See also the bounded variant - at "Control.Concurrent.Chan.Unagi.Bounded". + perform better when a queue grows very large. If you need a bounded queue, + see "Control.Concurrent.Chan.Unagi.Bounded". And if your application doesn't + require blocking reads, or is single-producer or single-consumer, then + "Control.Concurrent.Chan.Unagi.NoBlocking" will offer lowest latency. -} -- * Creating channels newChan @@ -12,6 +14,8 @@ -- ** Reading , readChan , readChanOnException + , tryReadChan + , Element(..) , getChanContents -- ** Writing , writeChan @@ -24,6 +28,7 @@ -- - faster write/read-many that increments counter by N import Control.Concurrent.Chan.Unagi.Internal +import Control.Concurrent.Chan.Unagi.NoBlocking.Types -- For 'writeList2Chan', as in vanilla Chan import System.IO.Unsafe ( unsafeInterleaveIO ) diff --git a/src/Control/Concurrent/Chan/Unagi/Bounded.hs b/src/Control/Concurrent/Chan/Unagi/Bounded.hs --- a/src/Control/Concurrent/Chan/Unagi/Bounded.hs +++ b/src/Control/Concurrent/Chan/Unagi/Bounded.hs @@ -1,11 +1,12 @@ {-# LANGUAGE CPP #-} -module Control.Concurrent.Chan.Unagi.Bounded ( +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." #-} + {-# WARNING "Waking up blocked writers may be slower than desired in GHC<7.8 which makes readMVar non-blocking on full MVars. Nextidering 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 @@ -18,6 +19,8 @@ -- ** Reading , readChan , readChanOnException + , tryReadChan + , Element(..) , getChanContents -- ** Writing , writeChan @@ -30,6 +33,7 @@ -- forked from src/Control/Concurrent/Chan/Unagi.hs 43706b2 import Control.Concurrent.Chan.Unagi.Bounded.Internal +import Control.Concurrent.Chan.Unagi.NoBlocking.Types -- For 'writeList2Chan', as in vanilla Chan import System.IO.Unsafe ( unsafeInterleaveIO ) diff --git a/src/Control/Concurrent/Chan/Unagi/Bounded/Internal.hs b/src/Control/Concurrent/Chan/Unagi/Bounded/Internal.hs --- a/src/Control/Concurrent/Chan/Unagi/Bounded/Internal.hs +++ b/src/Control/Concurrent/Chan/Unagi/Bounded/Internal.hs @@ -4,7 +4,7 @@ , writerCheckin, unblockWriters, tryWriterCheckin, WriterCheckpoint(..) , NextSegment(..), StreamHead(..) , newChanStarting, writeChan, readChan, readChanOnException - , tryWriteChan + , tryWriteChan, tryReadChan , dupChan ) where @@ -28,6 +28,7 @@ import GHC.Exts(inline) import Utilities(nextHighestPowerOfTwo) +import qualified Control.Concurrent.Chan.Unagi.NoBlocking.Types as UT -- | The write end of a channel created with 'newChan'. @@ -160,15 +161,12 @@ {-# INLINE dupChan #-} dupChan (InChan _ _ (ChanEnd logBounds boundsMn1 segSource counter streamHead)) = do hLoc <- readIORef streamHead - loadLoadBarrier -- NOTE [1] + loadLoadBarrier 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. @@ -178,8 +176,9 @@ -- 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. +-- while blocking here, the write will nonetheless 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 @@ -197,11 +196,11 @@ 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 $ + (\checkpt-> do + segUnlocked <- if canBlock + then True <$ writerCheckin checkpt + else tryWriterCheckin checkpt + when segUnlocked $ updateStreamHeadIfNecessary ) -- NOTE [1/2] maybeCheckpt @@ -248,16 +247,8 @@ 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 +readChanOnExceptionUnmasked h = \oc-> do + (seg,segIx) <- startReadChan oc cellTkt <- readArrayElem seg segIx case peekTicket cellTkt of @@ -277,7 +268,48 @@ -- N.B. must use `readMVar` here to support `dupChan`: where readBlocking v = inline h $ readMVar v +-- factored out for `tryReadChan` below: +startReadChan :: OutChan a -> IO (StreamSegment a, Int) +{-# INLINE startReadChan #-} +startReadChan (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 + return (seg,segIx) + + +-- TODO we might want also a blocking `IO a` returned here, or use an opaque +-- Element type supporting blocking, since otherwise calling `tryReadChan` we +-- give up the ability to block on that element. Please open an issue if you +-- need this in the meantime. And also handling of lost elements on async +-- exceptions. And also isActive... + +-- | Returns immediately with an @'UT.Element' a@ future, which returns one +-- unique element when it becomes available via 'UT.tryRead'. +-- +-- /Note re. exceptions/: When an async exception is raised during a @tryReadChan@ +-- the message that the read would have returned is likely to be lost, just as +-- it would be when raised directly after this function returns. +tryReadChan :: OutChan a -> IO (UT.Element a) +{-# INLINE tryReadChan #-} +tryReadChan oc = do -- no mask necessary + (seg,segIx) <- startReadChan oc + + return $ UT.Element $ do + cell <- P.readArray seg segIx + case cell of + Written a -> return $ Just a + Empty -> return Nothing + Blocking v -> tryReadMVar 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@ @@ -309,10 +341,6 @@ {-# 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 @@ -380,10 +408,11 @@ 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. + -- 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: @@ -394,11 +423,8 @@ -- 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 + -- The segment we're reading from (or any *behind* the one + -- we're reading from) is always unblocked for writers: readerUnblockAndReturn $ peekInstalled installed else do potentialCheckpt <- WriterCheckpoint <$> newEmptyMVar @@ -446,26 +472,17 @@ #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 +tryWriterCheckin (WriterCheckpoint v) = -- 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. -- HOWEVER, tryReadMVar is also buggy in GHC < 7.8.3 -- https://ghc.haskell.org/trac/ghc/ticket/9148 - unblocked <- #ifdef TRYREADMVAR - isJust <$> tryReadMVar v + isJust <$> tryReadMVar v #else - tryTakeMVar v >>= maybe (return False) ((True <$) . tryPutMVar v) + tryTakeMVar v >>= maybe (return False) ((True <$) . tryPutMVar 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 @@ -12,7 +12,7 @@ m = n .&. sEGMENT_LENGTH_MN_1 in d `seq` m `seq` (d,m) --- Constant for now: back-of-envelope considerations: +-- Nexttant for now: back-of-envelope considerations: -- - making most of constant factor for cloning array of *any* size -- - make most of overheads of moving to the next segment, etc. -- - provide enough runway for creating next segment when 32 simultaneous writers 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 @@ -7,7 +7,9 @@ , InChan(..), OutChan(..), ChanEnd(..), StreamSegment, Cell(..), Stream(..) , NextSegment(..), StreamHead(..) , newChanStarting, writeChan, readChan, readChanOnException - , dupChan + , dupChan, tryReadChan + -- For Unagi.NoBlocking: + , moveToNextCell, waitingAdvanceStream, newSegmentSource ) where @@ -27,45 +29,56 @@ import GHC.Exts(inline) import Control.Concurrent.Chan.Unagi.Constants +import qualified Control.Concurrent.Chan.Unagi.NoBlocking.Types as UT +import Utilities(touchIORef) -- | The write end of a channel created with 'newChan'. -data InChan a = InChan !(Ticket (Cell a)) !(ChanEnd a) - deriving Typeable - --- | The read end of a channel created with 'newChan'. -newtype OutChan a = OutChan (ChanEnd a) +data InChan a = InChan !(Ticket (Cell a)) !(ChanEnd (Cell 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 +-- | The read end of a channel created with 'newChan'. +newtype OutChan a = OutChan (ChanEnd (Cell a)) + deriving (Eq,Typeable) + + -- TODO POTENTIAL CPP FLAGS (or functions) -- - Strict element (or lazy? maybe also expose a writeChan' when relevant?) -- - sEGMENT_LENGTH -- - reads that clear the element immediately (or export as a special function?) -- InChan & OutChan are mostly identical, sharing a stream, but with --- independent counters -data ChanEnd a = +-- independent counters. +-- +-- NOTE: we parameterize this, and its child types, by `cell_a` (instantiated +-- to `Cell a` in this module) instead of `a` so that we can use +-- `moveToNextCell`, `waitingAdvanceStream`, and `newSegmentSource` and all +-- the types below in Unagi.NoBlocking, which uses a different type `Cell a`; +-- Sorry! +data ChanEnd cell_a = -- an efficient producer of segments of length sEGMENT_LENGTH: - ChanEnd !(SegSource a) + ChanEnd !(SegSource cell_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)) + !(IORef (StreamHead cell_a)) deriving Typeable -data StreamHead a = StreamHead !Int !(Stream a) +instance Eq (ChanEnd a) where + (ChanEnd _ _ headA) == (ChanEnd _ _ headB) + = headA == headB + +data StreamHead cell_a = StreamHead !Int !(Stream cell_a) + --TODO later see if we get a benefit from the small array primops in 7.10, -- which omit card-marking overhead and might have faster clone. -type StreamSegment a = P.MutableArray RealWorld (Cell a) +type StreamSegment cell_a = P.MutableArray RealWorld cell_a -- TRANSITIONS and POSSIBLE VALUES: -- During Read: @@ -83,20 +96,20 @@ -- exits we will have allocated ~ 3 segments extra memory than was actually -- required. -data Stream a = - Stream !(StreamSegment a) +data Stream cell_a = + Stream !(StreamSegment cell_a) -- The next segment in the stream; new segments are allocated and -- put here as we go, with threads cooperating to allocate new -- segments: - !(IORef (NextSegment a)) + !(IORef (NextSegment cell_a)) -data NextSegment a = NoSegment | Next !(Stream a) +data NextSegment cell_a = NoSegment | Next !(Stream cell_a) -- we expose `startingCellOffset` for debugging correct behavior with overflow: newChanStarting :: Int -> IO (InChan a, OutChan a) {-# INLINE newChanStarting #-} newChanStarting !startingCellOffset = do - segSource <- newSegmentSource + segSource <- newSegmentSource Empty firstSeg <- segSource -- collect a ticket to save for writer CAS savedEmptyTkt <- readArrayElem firstSeg 0 @@ -121,13 +134,15 @@ return $ OutChan (ChanEnd segSource counter' streamHead') -- [1] We must read the streamHead before inspecting the counter; otherwise, -- as writers write, the stream head pointer may advance past the cell - -- indicated by wCount. + -- indicated by wCount. For the corresponding store-store barrier see [*] in + -- moveToNextCell -- | Write a value to the channel. writeChan :: InChan a -> a -> IO () {-# INLINE writeChan #-} writeChan (InChan savedEmptyTkt ce@(ChanEnd segSource _ _)) = \a-> mask_ $ do - (segIx, (Stream seg next)) <- moveToNextCell ce + (segIx, (Stream seg next), maybeUpdateStreamHead) <- moveToNextCell ce + maybeUpdateStreamHead (success,nonEmptyTkt) <- casArrayElem seg segIx savedEmptyTkt (Written a) -- try to pre-allocate next segment; NOTE [1] when (segIx == 0) $ void $ @@ -165,7 +180,8 @@ readChanOnExceptionUnmasked :: (IO a -> IO a) -> OutChan a -> IO a {-# INLINE readChanOnExceptionUnmasked #-} readChanOnExceptionUnmasked h = \(OutChan ce)-> do - (segIx, (Stream seg _)) <- moveToNextCell ce + (segIx, (Stream seg _), maybeUpdateStreamHead) <- moveToNextCell ce + maybeUpdateStreamHead cellTkt <- readArrayElem seg segIx case peekTicket cellTkt of Written a -> return a @@ -185,6 +201,33 @@ where readBlocking v = inline h $ readMVar v +-- TODO we might want also a blocking `IO a` returned here, or use an opaque +-- Element type supporting blocking, since otherwise calling `tryReadChan` we +-- give up the ability to block on that element. Please open an issue if you +-- need this in the meantime. And also handling of lost elements on async +-- exceptions. And also isActive... + +-- | Returns immediately with an @'UT.Element' a@ future, which returns one +-- unique element when it becomes available via 'UT.tryRead'. If you're using +-- this function exclusively you might find the implementation in +-- "Control.Concurrent.Chan.Unagi.NoBlocking" is faster. +-- +-- /Note re. exceptions/: When an async exception is raised during a @tryReadChan@ +-- the message that the read would have returned is likely to be lost, just as +-- it would be when raised directly after this function returns. +tryReadChan :: OutChan a -> IO (UT.Element a) +{-# INLINE tryReadChan #-} +tryReadChan (OutChan ce) = do -- no masking needed + (segIx, (Stream seg _), maybeUpdateStreamHead) <- moveToNextCell ce + maybeUpdateStreamHead + return $ UT.Element $ do + cell <- P.readArray seg segIx + case cell of + Written a -> return $ Just a + Empty -> return Nothing + Blocking v -> tryReadMVar 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@ @@ -208,20 +251,19 @@ readChanOnException c h = mask_ $ readChanOnExceptionUnmasked (\io-> io `onException` (h io)) c + +------------ NOTE: ALL CODE BELOW IS RE-USED IN Unagi.NoBlocking -------------- + + -- increments counter, finds stream segment of corresponding cell (updating the -- stream head pointer as needed), and returns the stream segment and relative -- index of our cell. -moveToNextCell :: ChanEnd a -> IO (Int, Stream a) +moveToNextCell :: ChanEnd cell_a -> IO (Int, Stream cell_a, IO ()) {-# INLINE moveToNextCell #-} moveToNextCell (ChanEnd segSource counter streamHead) = do (StreamHead offset0 str0) <- readIORef streamHead - -- NOTE [3] -#ifdef NOT_x86 - -- fetch-and-add is a full barrier on x86; otherwise we need to make sure - -- the read above occurrs before our fetch-and-add: - loadLoadBarrier -#endif - ix <- incrCounter 1 counter + -- NOTE [3/4] + ix <- incrCounter 1 counter -- [*] let (segsAway, segIx) = assert ((ix - offset0) >= 0) $ divMod_sEGMENT_LENGTH $! (ix - offset0) -- (ix - offset0) `quotRem` sEGMENT_LENGTH @@ -231,11 +273,14 @@ waitingAdvanceStream next segSource (nEW_SEGMENT_WAIT*segIx) -- NOTE [1] >>= go (n-1) str <- go segsAway str0 - when (segsAway > 0) $ do - let !offsetN = - offset0 + (segsAway `unsafeShiftL` lOG_SEGMENT_LENGTH) --(segsAway*sEGMENT_LENGTH) - writeIORef streamHead $ StreamHead offsetN str -- NOTE [2] - return (segIx,str) + -- In Unagi.NoBlocking we need to control when this is run (see also [5]): + let !maybeUpdateStreamHead = do + when (segsAway > 0) $ do + let !offsetN = + offset0 + (segsAway `unsafeShiftL` lOG_SEGMENT_LENGTH) --(segsAway*sEGMENT_LENGTH) + writeIORef streamHead $ StreamHead offsetN str + touchIORef streamHead -- NOTE [5] + return (segIx,str, maybeUpdateStreamHead) -- [1] All readers or writers needing to work with a not-yet-created segment -- race to create it, but those past index 0 have progressively long waits; 20 -- is chosen as 20 readIORefs should be more than enough time for writer/reader @@ -250,15 +295,25 @@ -- descheduled, meanwhile other readers/writers increment counter one full -- lap; when we increment we think we've found our cell in what is actually a -- very old segment. However in this scenario all addressable memory will - -- have been consumed just by the array pointers whivh haven't been able to + -- have been consumed just by the array pointers which haven't been able to -- be GC'd. So I don't think this is something to worry about. + -- + -- [4] We must ensure the read above doesn't move ahead of our incrCounter + -- below. But fetchAddByteArrayInt is meant to be a full barrier (for + -- compiler and processor) across architectures, so no explicit barrier is + -- needed here. + -- + -- [[5]] FOR Unagi.NoBlocking: This helps ensure that our (possibly last) use + -- of streamHead occurs after our (possibly last) write, for correctness of + -- 'isActive'. See NOTE 1 of 'Unagi.NoBlocking.writeChan' -- thread-safely try to fill `nextSegRef` at the next offset with a new -- segment, waiting some number of iterations (for other threads to handle it). -- Returns nextSegRef's StreamSegment. -waitingAdvanceStream :: IORef (NextSegment a) -> SegSource a - -> Int -> IO (Stream a) +waitingAdvanceStream :: IORef (NextSegment cell_a) -> SegSource cell_a + -> Int -> IO (Stream cell_a) +{-# NOINLINE waitingAdvanceStream #-} waitingAdvanceStream nextSegRef segSource = go where go !wait = assert (wait >= 0) $ do tk <- readForCAS nextSegRef @@ -280,13 +335,13 @@ -- copying a template array with cloneMutableArray is much faster than creating -- a new one; in fact it seems we need this in order to scale, since as cores -- increase we don't have enough "runway" and can't allocate fast enough: -type SegSource a = IO (StreamSegment a) +type SegSource cell_a = IO (StreamSegment cell_a) -newSegmentSource :: IO (SegSource a) -newSegmentSource = do +newSegmentSource :: cell_a -> IO (SegSource cell_a) +newSegmentSource cell_empty = 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 sEGMENT_LENGTH + arr <- evaluate cell_empty >>= P.newArray sEGMENT_LENGTH return (P.cloneMutableArray arr 0 sEGMENT_LENGTH) -- ---------- diff --git a/src/Control/Concurrent/Chan/Unagi/NoBlocking.hs b/src/Control/Concurrent/Chan/Unagi/NoBlocking.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Chan/Unagi/NoBlocking.hs @@ -0,0 +1,45 @@ +module Control.Concurrent.Chan.Unagi.NoBlocking ( +{- | General-purpose concurrent FIFO queue without blocking reads, and with + optimized variants for single-threaded producers and/or consumers. This + variant, and even more so the SP/SC variants, offer the lowest latency of + all of the implementations in this library. + -} + -- * Creating channels + newChan + , InChan(), OutChan() + -- * Channel operations + -- ** Reading + , tryReadChan + , readChan + , Element(..) + -- *** Utilities + , isActive + -- ** Writing + , writeChan + , writeList2Chan + -- ** Broadcasting + , dupChan + -- ** Streaming + , Stream(..), Next(..) + , streamChan + ) where + +-- Forked from src/Control/Concurrent/Chan/Unagi.hs at 065cd68010 + +-- TODO additonal functions: +-- - faster write/read-many that increments counter by N + +import Control.Concurrent.Chan.Unagi.NoBlocking.Internal hiding (Stream) +import Control.Concurrent.Chan.Unagi.NoBlocking.Types + +-- | Create a new channel, returning its write and read ends. +newChan :: IO (InChan a, OutChan a) +newChan = newChanStarting (maxBound - 10) + -- lets us test counter overflow in tests and normal course of operation + + +-- | 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/NoBlocking/Internal.hs b/src/Control/Concurrent/Chan/Unagi/NoBlocking/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Chan/Unagi/NoBlocking/Internal.hs @@ -0,0 +1,254 @@ +{-# LANGUAGE BangPatterns , DeriveDataTypeable, CPP #-} +module Control.Concurrent.Chan.Unagi.NoBlocking.Internal +#ifdef NOT_x86 + {-# WARNING "This library is unlikely to perform well on architectures without a fetch-and-add instruction" #-} +#endif + (sEGMENT_LENGTH + , InChan(..), OutChan(..), ChanEnd(..), StreamSegment, Cell, Stream(..) + , NextSegment(..), StreamHead(..) + , newChanStarting, writeChan, tryReadChan, readChan, UT.Element(..) + , dupChan + , streamChan + , isActive + ) + where + +-- Forked from src/Control/Concurrent/Chan/Unagi/Internal.hs at 065cd68010 +-- +-- Some detailed NOTEs present in Control.Concurrent.Chan.Unagi have been +-- removed here although they still pertain. If you intend to work on this +-- module, please be sure you're familiar with those concerns. +-- +-- The implementation here is Control.Concurrent.Chan.Unagi with the blocking +-- read mechanics removed, the required CAS rendevouz replaced with +-- writeArray/readArray, and MPSC/SPMC/SPSC variants that eliminate streamHead +-- updates and atomic operations on any 'S' sides. + +import Data.IORef +import Control.Exception +import Data.Atomics.Counter.Fat +import Data.Atomics +import qualified Data.Primitive as P +import Control.Monad +import Control.Applicative +import Data.Bits +import Data.Typeable(Typeable) + +import Control.Concurrent.Chan.Unagi.Internal( + newSegmentSource, moveToNextCell, waitingAdvanceStream, + ChanEnd(..), StreamHead(..), StreamSegment, Stream(..), NextSegment(..)) +import Control.Concurrent.Chan.Unagi.Constants +import qualified Control.Concurrent.Chan.Unagi.NoBlocking.Types as UT + + +-- | The write end of a channel created with 'newChan'. +data InChan a = InChan !(IORef Bool) -- Used for creating an OutChan in dupChan + !(ChanEnd (Cell a)) + deriving (Typeable,Eq) + +-- | The read end of a channel created with 'newChan'. +data OutChan a = OutChan !(IORef Bool) -- Is corresponding InChan still alive? + !(ChanEnd (Cell a)) + deriving (Typeable,Eq) + + +-- TRANSITIONS and POSSIBLE VALUES: +-- During Read: +-- Nothing +-- Just a +-- During Write: +-- Nothing -> Just a +type Cell a = Maybe a + + +-- we expose `startingCellOffset` for debugging correct behavior with overflow: +newChanStarting :: Int -> IO (InChan a, OutChan a) +{-# INLINE newChanStarting #-} +newChanStarting !startingCellOffset = do + segSource <- newSegmentSource Nothing + stream <- Stream <$> segSource + <*> newIORef NoSegment + let end = ChanEnd segSource + <$> newCounter (startingCellOffset - 1) + <*> newIORef (StreamHead startingCellOffset stream) + inEnd@(ChanEnd _ _ inHeadRef) <- end + finalizee <- newIORef True + void $ mkWeakIORef inHeadRef $ do -- NOTE [1] + -- make sure the array writes of any final writeChans occur before the + -- following writeIORef. See isActive [*]: + writeBarrier + writeIORef finalizee False + (,) (InChan finalizee inEnd) <$> (OutChan finalizee <$> end) + -- [1] We no longer get blocked indefinitely exception in readers when all + -- writers disappear, so we use finalizers. See also NOTE 1 in 'writeChan' and + -- implementation of 'isActive' below. + +-- | An action that returns @False@ sometime after the chan no longer has any +-- writers. +-- +-- After @False@ is returned, any 'UT.tryRead' which returns @Nothing@ can +-- be considered to be dead. Likewise for 'UT.tryReadNext'. Note that in the +-- blocking implementations a @BlockedIndefinitelyOnMVar@ exception is raised, +-- so this function is unnecessary. +isActive :: OutChan a -> IO Bool +isActive (OutChan finalizee _) = do + b <- readIORef finalizee + -- make sure that any tryRead that follows is not moved ahead. See + -- newChanStarting [*]: + loadLoadBarrier + return b + +-- | 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. +-- +-- See also 'streamChan' for a faster alternative that might be appropriate. +dupChan :: InChan a -> IO (OutChan a) +{-# INLINE dupChan #-} +dupChan (InChan finalizee (ChanEnd segSource counter streamHead)) = do + hLoc <- readIORef streamHead + loadLoadBarrier + wCount <- readCounter counter + counter' <- newCounter wCount + streamHead' <- newIORef hLoc + return $ OutChan finalizee $ ChanEnd segSource counter' streamHead' + + +-- | Write a value to the channel. +writeChan :: InChan a -> a -> IO () +{-# INLINE writeChan #-} +writeChan (InChan _ ce@(ChanEnd segSource _ _)) = \a-> mask_ $ do + (segIx, (Stream seg next), maybeUpdateStreamHead) <- moveToNextCell ce + P.writeArray seg segIx (Just a) + maybeUpdateStreamHead -- NOTE [1] + -- try to pre-allocate next segment: + when (segIx == 0) $ void $ + waitingAdvanceStream next segSource 0 + -- [1] We return the maybeUpdateStreamHead action from moveToNextCell rather + -- than running it before returning, because we must ensure that the + -- streamHead IORef is not GC'd (and its finalizer run) before the last + -- element is written; else the user has no way of being sure that it has read + -- the last element. See 'newChanStarting' and 'isActive'. + + + + + -- NOTE: this might be better named "claimElement" or something, but we'll + -- keep the name since it's the closest equivalent to a real "tryReadChan" + -- we can get in this design: + +-- | Returns immediately with an @'UT.Element' a@ future, which returns one +-- unique element when it becomes available via 'UT.tryRead'. +-- +-- /Note re. exceptions/: When an async exception is raised during a @tryReadChan@ +-- the message that the read would have returned is likely to be lost, just as +-- it would be when raised directly after this function returns. +tryReadChan :: OutChan a -> IO (UT.Element a) +{-# INLINE tryReadChan #-} +tryReadChan (OutChan _ ce) = do -- NOTE [1] + (segIx, (Stream seg _), maybeUpdateStreamHead) <- moveToNextCell ce + maybeUpdateStreamHead + return $ UT.Element $ P.readArray seg segIx + -- [1] We don't need to mask exceptions here. We say that exceptions raised in + -- tryReadChan are linearizable as occuring just before we are to return with our + -- element. Note that the two effects in moveToNextCell are to increment the + -- counter (this is the point after which we lose the read), and set up any + -- future segments required (all atomic operations). + + +-- | @readChan io c@ returns the next element from @c@, calling 'tryReadChan' +-- and looping on the 'UT.Element' returned, and calling @io@ at each iteration +-- when the element is not yet available. It throws 'BlockedIndefinitelyOnMVar' +-- when 'isActive' determines that a value will never be returned. +-- +-- When used like @readChan 'yield'@ or @readChan ('threadDelay' 10)@ this is +-- the semantic equivalent to the blocking @readChan@ in the other +-- implementations. +readChan :: IO () -> OutChan a -> IO a +{-# INLINE readChan #-} +readChan io oc = tryReadChan oc >>= \el-> + let peekMaybe f = UT.tryRead el >>= maybe f return + go = peekMaybe checkAndGo + checkAndGo = do + b <- isActive oc + if b then io >> go + -- Do a necessary final check of the element: + else peekMaybe $ throwIO BlockedIndefinitelyOnMVar + in go + + +-- TODO a write-side equivalent: +-- - can be made streaming agnostic? +-- - NOTE: if we're only streaming in and out, then using multiple chans is +-- possible (e.g. 3:6 is equivalent to 3 sets of 1:2 streaming chans) +-- +-- TODO MAYBE: overload `streamChan` for Streams too. +-- +-- TODO MAYBE mechanism for keeping stream consumers from drifting too far apart + + +-- | Produce the specified number of interleaved \"streams\" from a chan. +-- Nextuming a 'UI.Stream' is much faster than calling 'tryReadChan', and +-- might be useful when an MPSC queue is needed, or when multiple consumers +-- should be load-balanced in a round-robin fashion. +-- +-- Usage example: +-- +-- > do mapM_ ('writeChan' i) [1..9] +-- > [str1, str2, str2] <- 'streamChan' 3 o +-- > forkIO $ printStream str1 -- prints: 1,4,7 +-- > forkIO $ printStream str2 -- prints: 2,5,8 +-- > forkIO $ printStream str3 -- prints: 3,6,9 +-- > where +-- > printStream str = do +-- > h <- 'tryReadNext' str +-- > case h of +-- > 'Next' a str' -> print a >> printStream str' +-- > -- We know that all values were already written, so a Pending tells +-- > -- us we can exit; in other cases we might call 'yield' and then +-- > -- retry that same @'tryReadNext' str@: +-- > 'Pending' -> return () +-- +-- Be aware: if one stream consumer falls behind another (e.g. because it is +-- slower) the number of elements in the queue which can't be GC'd will grow. +-- You may want to do some coordination of 'UT.Stream' consumers to prevent +-- this. +streamChan :: Int -> OutChan a -> IO [UT.Stream a] +{-# INLINE streamChan #-} +streamChan period (OutChan _ (ChanEnd segSource counter streamHead)) = do + when (period < 1) $ error "Argument to streamChan must be > 0" + + (StreamHead offsetInitial strInitial) <- readIORef streamHead + -- Make sure the read above occurs before our readCounter: + loadLoadBarrier + -- Linearizable as the first unread element; N.B. (+1): + !ix0 <- (+1) <$> readCounter counter + + -- Adapted from moveToNextCell, given a stream segment location `str0` and + -- its offset, `offset0`, this navigates to the UT.Stream segment holding `ix` + -- and begins recursing in our UT.Stream wrappers + let stream !offset0 str0 !ix = UT.Stream $ do + -- Find our stream segment and relative index: + let (segsAway, segIx) = assert ((ix - offset0) >= 0) $ + divMod_sEGMENT_LENGTH $! (ix - offset0) + -- (ix - offset0) `quotRem` sEGMENT_LENGTH + {-# INLINE go #-} + go 0 str = return str + go !n (Stream _ next) = + waitingAdvanceStream next segSource (nEW_SEGMENT_WAIT*segIx) + >>= go (n-1) + -- the stream segment holding `ix`, and its calculated offset: + str@(Stream seg _) <- go segsAway str0 + let !strOffset = offset0+(segsAway `unsafeShiftL` lOG_SEGMENT_LENGTH) + -- (segsAway * sEGMENT_LENGTH) + mbA <- P.readArray seg segIx + case mbA of + Nothing -> return UT.Pending + -- Navigate to next cell and return this cell's value + -- along with the wrapped action to read from the next + -- cell and possibly recurse. + Just a -> return $ UT.Next a $ stream strOffset str (ix+period) + + return $ map (stream offsetInitial strInitial) $ + -- [ix0..(ix0+period-1)] -- WRONG (hint: overflow)! + take period $ iterate (+1) ix0 diff --git a/src/Control/Concurrent/Chan/Unagi/NoBlocking/Types.hs b/src/Control/Concurrent/Chan/Unagi/NoBlocking/Types.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Chan/Unagi/NoBlocking/Types.hs @@ -0,0 +1,62 @@ +module Control.Concurrent.Chan.Unagi.NoBlocking.Types where + +import Control.Applicative +import Control.Monad.Fix +import Control.Monad +import Data.Maybe + +-- Mostly here to avoid unfortunate name clash with our internal Stream type + + +-- | An infinite stream of elements. 'tryReadNext' can be called any number of +-- times from multiple threads, and returns a value which moves monotonically +-- from 'Pending' to 'Next' if and when a head element becomes available. +-- @isActive@ can be used to determine if the stream has expired. +newtype Stream a = Stream { tryReadNext :: IO (Next a) } + +data Next a = Next a (Stream a) -- ^ The next head element along with the tail @Stream@. + | Pending -- ^ The next element is not yet in the queue; you can retry 'tryReadNext' until a @Next@ is returned. + + +-- | An @IO@ action that returns a particular enqueued element when and if it +-- becomes available. +-- +-- Each @Element@ corresponds to a particular enqueued element, i.e. a returned +-- @Element@ always offers the only means to access one particular enqueued +-- item. The value returned by @tryRead@ moves monotonically from @Nothing@ +-- to @Just a@ when and if an element becomes available, and is idempotent at +-- that point. +newtype Element a = Element { tryRead :: IO (Maybe a) } + +-- Instances cribbed from MaybeT, from transformers v0.4.2.0 +instance Functor Element where + fmap f = Element . fmap (fmap f) . tryRead + +instance Applicative Element where + pure = return + (<*>) = ap + +instance Alternative Element where + empty = mzero + (<|>) = mplus + +instance Monad Element where + fail _ = Element (return Nothing) + return = Element . return . return + x >>= f = Element $ do + v <- tryRead x + case v of + Nothing -> return Nothing + Just y -> tryRead (f y) + +instance MonadPlus Element where + mzero = Element (return Nothing) + mplus x y = Element $ do + v <- tryRead x + case v of + Nothing -> tryRead y + Just _ -> return v + +instance MonadFix Element where + mfix f = Element (mfix (tryRead . f . fromMaybe bomb)) + where bomb = error "mfix (Element): inner computation returned Nothing" diff --git a/src/Control/Concurrent/Chan/Unagi/NoBlocking/Unboxed.hs b/src/Control/Concurrent/Chan/Unagi/NoBlocking/Unboxed.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Chan/Unagi/NoBlocking/Unboxed.hs @@ -0,0 +1,47 @@ +module Control.Concurrent.Chan.Unagi.NoBlocking.Unboxed ( +{- | General-purpose concurrent FIFO queue without blocking reads, and with + optimized variants for single-threaded producers and/or consumers. This + variant, and even more so the SP/SC variants, offer the lowest latency of + all of the implementations in this library. + -} + -- * Creating channels + newChan + , InChan(), OutChan() + , UnagiPrim(..) + -- * Channel operations + -- ** Reading + , tryReadChan + , readChan + , Element(..) + -- *** Utilities + , isActive + -- ** Writing + , writeChan + , writeList2Chan + -- ** Broadcasting + , dupChan + -- ** Streaming + , Stream(..), Next(..) + , streamChan + ) where + +-- Forked from src/Control/Concurrent/Chan/Unagi/NoBlocking.hs at 9e2306330e + +-- TODO additonal functions: +-- - faster write/read-many that increments counter by N + +import Control.Concurrent.Chan.Unagi.NoBlocking.Unboxed.Internal hiding (Stream) +import Control.Concurrent.Chan.Unagi.NoBlocking.Types +import Control.Concurrent.Chan.Unagi.Unboxed(UnagiPrim(..)) + +-- | Create a new channel, returning its write and read ends. +newChan :: UnagiPrim a=> IO (InChan a, OutChan a) +newChan = newChanStarting (maxBound - 10) + -- lets us test counter overflow in tests and normal course of operation + + +-- | Write an entire list of items to a chan type. Writes here from multiple +-- threads may be interleaved, and infinite lists are supported. +writeList2Chan :: UnagiPrim a=> InChan a -> [a] -> IO () +{-# INLINABLE writeList2Chan #-} +writeList2Chan ch = sequence_ . map (writeChan ch) diff --git a/src/Control/Concurrent/Chan/Unagi/NoBlocking/Unboxed/Internal.hs b/src/Control/Concurrent/Chan/Unagi/NoBlocking/Unboxed/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Chan/Unagi/NoBlocking/Unboxed/Internal.hs @@ -0,0 +1,297 @@ +{-# LANGUAGE BangPatterns , DeriveDataTypeable, CPP #-} +module Control.Concurrent.Chan.Unagi.NoBlocking.Unboxed.Internal +#ifdef NOT_x86 + {-# WARNING "This library is unlikely to perform well on architectures without a fetch-and-add instruction" #-} +#endif + (sEGMENT_LENGTH + , InChan(..), OutChan(..), ChanEnd(..), Cell, Stream(..) + , NextSegment(..), StreamHead(..) + , newChanStarting, writeChan, tryReadChan, readChan, UT.Element(..) + , dupChan + , streamChan + , isActive + ) + where + +-- Forked from src/Control/Concurrent/Chan/Unagi/NoBlocking/Internal.hs at +-- 9e2306330e with some code copied and modified from Unagi.Unboxed. +-- +-- The main motivation for this variant is that it lets us take full advantage +-- of the atomicUnicorn trick, so in both read and write we need only use +-- sigArr when the value to be written == atomicUnicorn. +-- +-- Some detailed NOTEs present in Control.Concurrent.Chan.Unagi.Unboxed have +-- been removed here although they still pertain. If you intend to work on this +-- module, please be sure you're familiar with those concerns. + +import Data.IORef +import Control.Exception +import Data.Atomics.Counter.Fat +import Data.Atomics +import qualified Data.Primitive as P +import Control.Monad +import Control.Applicative +import Data.Bits +import Data.Typeable(Typeable) +import Data.Maybe + +import Control.Concurrent.Chan.Unagi.Constants + +-- We can re-use much of the Unagi.Unboxed implementation here, and some of +-- Unagi.NoBlocking (at least our types, which is important): +import Control.Concurrent.Chan.Unagi.Unboxed.Internal( + ChanEnd(..), StreamHead(..), Cell, Stream(..) + , NextSegment(..), moveToNextCell, waitingAdvanceStream, segSource + , cellEmpty, readElementArray, writeElementArray + , SignalIntArray, ElementArray, UnagiPrim(..)) +import qualified Control.Concurrent.Chan.Unagi.NoBlocking.Types as UT + + +-- | The write end of a channel created with 'newChan'. +data InChan a = InChan !(IORef Bool) -- Used for creating an OutChan in dupChan + !(ChanEnd a) + deriving (Typeable) + +-- | The read end of a channel created with 'newChan'. +data OutChan a = OutChan !(IORef Bool) -- Is corresponding InChan still alive? + !(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 + + +newChanStarting :: (UnagiPrim a)=> Int -> IO (InChan a, OutChan a) +{-# INLINE newChanStarting #-} +newChanStarting !startingCellOffset = do + let undefinedNewIndexedMVar = return $ -- NOTE [1] + error "Unagi.NoBlocking.Unboxed tried to use initial fake IndexedMVar" + stream <- uncurry Stream <$> segSource + <*> undefinedNewIndexedMVar + <*> newIORef NoSegment + let end = ChanEnd + <$> newCounter (startingCellOffset - 1) + <*> newIORef (StreamHead startingCellOffset stream) + inEnd@(ChanEnd _ inHeadRef) <- end + finalizee <- newIORef True + void $ mkWeakIORef inHeadRef $ do + writeBarrier + writeIORef finalizee False + (,) (InChan finalizee inEnd) <$> (OutChan finalizee <$> end) + -- [1] We reuse most of Unagi.Unboxed's internals here, but unfortunately + -- that implementation uses a Stream type with an IndexedMVar to coordinate + -- blocking reads. Rather than do a lot of refactoring of Unagi.Unboxed, for + -- now we just fake it here. Unagi.Unboxed.waitingAdvanceStream will actually + -- create new IndexedMVars for each segment, but we hope at worst that they + -- will be GC'd immediately even when many segments-worth of elements are in + -- the queue; the main concern is not to accumulate lots of mutable boxed + -- objects. TODO better later, maybe. + + +-- | An action that returns @False@ sometime after the chan no longer has any +-- writers. +-- +-- After @False@ is returned, any 'UT.tryRead' which returns @Nothing@ can +-- be considered to be dead. Likewise for 'UT.tryReadNext'. Note that in the +-- blocking implementations a @BlockedIndefinitelyOnMVar@ exception is raised, +-- so this function is unnecessary. +isActive :: OutChan a -> IO Bool +isActive (OutChan finalizee _) = do + b <- readIORef finalizee + -- make sure that a tryRead that follows is not moved ahead: + loadLoadBarrier + return b + + +-- | 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. +-- +-- See also 'streamChan' for a faster alternative that might be appropriate. +dupChan :: InChan a -> IO (OutChan a) +{-# INLINE dupChan #-} +dupChan (InChan finalizee (ChanEnd counter streamHead)) = do + hLoc <- readIORef streamHead + loadLoadBarrier + wCount <- readCounter counter + counter' <- newCounter wCount + streamHead' <- newIORef hLoc + return $ OutChan finalizee $ ChanEnd counter' streamHead' + + +-- READING AND WRITING +-- +-- We re-use the internals of Unagi.Unboxed, but use them a bit differently; +-- in particular where Unagi.Unboxed uses its SignalIntArray to indicate the +-- status of the corresponding ElementArray cell, we use it only to +-- disambiguate an unwritten cell from a written cell of a "magic" value, +-- which we'll describe below. +-- +-- When we're reading and writing values that can be written atomically (see +-- atomicUnicorn), and when that particular value is not equal to that magic +-- value we get a fast write path: simply write to the eArr. Likewise when a +-- reader reads from eArr and sees something /= atomicUnicorn, it can simply +-- return with it. In all other cases readers and writers must check in at the +-- sigArr, as in Unagi.Unboxed. + +nonMagicCellWritten :: Int +nonMagicCellWritten = 1 +-- and also: `cellEmpty` (imported) + + + +-- | Write a value to the channel. +writeChan :: UnagiPrim a=> InChan a -> a -> IO () +{-# INLINE writeChan #-} +writeChan (InChan _ ce) = \a-> mask_ $ do + (segIx, (Stream sigArr eArr _ next), maybeUpdateStreamHead) <- moveToNextCell ce + -- NOTE!: must write element both before updating stream head (see + -- NoBlocking), and before signaling with CAS (if applicable): + writeElementArray eArr segIx a + + let magic = atomicUnicorn + when (isNothing magic || Just a == magic) $ do + -- in which case a reader can't tell we've written just from a (possibly + -- non-atomic) read from eArr: + writeBarrier -- NOTE [1] + P.writeByteArray sigArr segIx nonMagicCellWritten + + maybeUpdateStreamHead -- NOTE [2] + -- try to pre-allocate next segment: + when (segIx == 0) $ void $ + waitingAdvanceStream next 0 + -- [1] we need a write barrier here to make sure GHC maintains our ordering + -- such that the element is written before we signal its availability with + -- the write to sigArr that follows. See [2] in readChanOnExceptionUnmasked. + -- + -- [2] Our final use of the head reference. We must make sure this IORef is + -- not GC'd (and its finalizer run) until after our writes to the arrays + -- above. See definition of maybeUpdateStreamHead. + + +-- | Returns immediately with an @'UT.Element' a@ future, which returns one +-- unique element when it becomes available via 'UT.tryRead'. +-- +-- /Note re. exceptions/: When an async exception is raised during a @tryReadChan@ +-- the message that the read would have returned is likely to be lost, just as +-- it would be when raised directly after this function returns. +tryReadChan :: UnagiPrim a=> OutChan a -> IO (UT.Element a) +{-# INLINE tryReadChan #-} +tryReadChan (OutChan _ ce) = do -- see NoBlocking re. not masking + (segIx, (Stream sigArr eArr _ _), maybeUpdateStreamHead) <- moveToNextCell ce + maybeUpdateStreamHead + return $ UT.Element $ + tryReadChanInternals segIx sigArr eArr + +tryReadChanInternals :: UnagiPrim a=> Int -> SignalIntArray -> ElementArray a -> IO (Maybe a) +{-# INLINE tryReadChanInternals #-} +tryReadChanInternals segIx sigArr eArr = do + let readElem = readElementArray eArr segIx + slowRead = do + sig <- P.readByteArray sigArr segIx + if sig == nonMagicCellWritten + then do + loadLoadBarrier -- see [1] in writeChan + Just <$> readElem + else assert (sig == cellEmpty) $ + return Nothing + -- If we know writes of this type are atomic, we can determine if the + -- element has been written, and possibly return it without checking + -- sigArr. + case atomicUnicorn of + Just magic -> do + el <- readElem + if (el /= magic) + -- Then we know `el` was atomically written: + then return $ Just el + else slowRead + Nothing -> slowRead + + +-- | @readChan io c@ returns the next element from @c@, calling 'tryReadChan' +-- and looping on the 'UT.Element' returned, and calling @io@ at each iteration +-- when the element is not yet available. It throws 'BlockedIndefinitelyOnMVar' +-- when 'isActive' determines that a value will never be returned. +-- +-- When used like @readChan 'yield'@ or @readChan ('threadDelay' 10)@ this is +-- the semantic equivalent to the blocking @readChan@ in the other +-- implementations. +readChan :: UnagiPrim a=> IO () -> OutChan a -> IO a +{-# INLINE readChan #-} +readChan io oc = tryReadChan oc >>= \el-> + let peekMaybe f = UT.tryRead el >>= maybe f return + go = peekMaybe checkAndGo + checkAndGo = do + b <- isActive oc + if b then io >> go + -- Do a necessary final check of the element: + else peekMaybe $ throwIO BlockedIndefinitelyOnMVar + in go + + +-- | Produce the specified number of interleaved \"streams\" from a chan. +-- Nextuming a 'UI.Stream' is much faster than calling 'tryReadChan', and +-- might be useful when an MPSC queue is needed, or when multiple consumers +-- should be load-balanced in a round-robin fashion. +-- +-- Usage example: +-- +-- > do mapM_ ('writeChan' i) [1..9] +-- > [str1, str2, str2] <- 'streamChan' 3 o +-- > forkIO $ printStream str1 -- prints: 1,4,7 +-- > forkIO $ printStream str2 -- prints: 2,5,8 +-- > forkIO $ printStream str3 -- prints: 3,6,9 +-- > where +-- > printStream str = do +-- > h <- 'tryReadNext' str +-- > case h of +-- > 'Next' a str' -> print a >> printStream str' +-- > -- We know that all values were already written, so a Pending tells +-- > -- us we can exit; in other cases we might call 'yield' and then +-- > -- retry that same @'tryReadNext' str@: +-- > 'Pending' -> return () +-- +-- Be aware: if one stream consumer falls behind another (e.g. because it is +-- slower) the number of elements in the queue which can't be GC'd will grow. +-- You may want to do some coordination of 'UT.Stream' consumers to prevent +-- this. +streamChan :: UnagiPrim a=> Int -> OutChan a -> IO [UT.Stream a] +{-# INLINE streamChan #-} +streamChan period (OutChan _ (ChanEnd counter streamHead)) = do + when (period < 1) $ error "Argument to streamChan must be > 0" + + (StreamHead offsetInitial strInitial) <- readIORef streamHead + -- Make sure the read above occurs before our readCounter: + loadLoadBarrier + -- Linearizable as the first unread element; N.B. (+1): + !ix0 <- (+1) <$> readCounter counter + + -- Adapted from moveToNextCell, given a stream segment location `str0` and + -- its offset, `offset0`, this navigates to the UT.Stream segment holding `ix` + -- and begins recursing in our UT.Stream wrappers + let stream !offset0 str0 !ix = UT.Stream $ do + -- Find our stream segment and relative index: + let (segsAway, segIx) = assert ((ix - offset0) >= 0) $ + divMod_sEGMENT_LENGTH $! (ix - offset0) + -- (ix - offset0) `quotRem` sEGMENT_LENGTH + {-# INLINE go #-} + go 0 str = return str + go !n (Stream _ _ _ next) = + waitingAdvanceStream next (nEW_SEGMENT_WAIT*segIx) + >>= go (n-1) + -- the stream segment holding `ix`, and its calculated offset: + str@(Stream sigArr eArr _ _) <- go segsAway str0 + let !strOffset = offset0+(segsAway `unsafeShiftL` lOG_SEGMENT_LENGTH) + -- (segsAway * sEGMENT_LENGTH) + mbEl <- tryReadChanInternals segIx sigArr eArr + return $ case mbEl of + Nothing -> UT.Pending + Just el -> UT.Next el $ stream strOffset str (ix+period) + + return $ map (stream offsetInitial strInitial) $ + -- [ix0..(ix0+period-1)] -- WRONG (hint: overflow)! + take period $ iterate (+1) ix0 diff --git a/src/Control/Concurrent/Chan/Unagi/Unboxed.hs b/src/Control/Concurrent/Chan/Unagi/Unboxed.hs --- a/src/Control/Concurrent/Chan/Unagi/Unboxed.hs +++ b/src/Control/Concurrent/Chan/Unagi/Unboxed.hs @@ -2,10 +2,13 @@ -- * Creating channels newChan , InChan(), OutChan() + , UnagiPrim(..) -- * Channel operations -- ** Reading , readChan , readChanOnException + , tryReadChan + , Element(..) , getChanContents -- ** Writing , writeChan @@ -26,19 +29,19 @@ -- - ...or interop with 'vector' lib import Control.Concurrent.Chan.Unagi.Unboxed.Internal +import Control.Concurrent.Chan.Unagi.NoBlocking.Types -- For 'writeList2Chan', as in vanilla Chan import System.IO.Unsafe ( unsafeInterleaveIO ) -import Data.Primitive(Prim) -- | Create a new channel, returning its write and read ends. -newChan :: Prim a=> IO (InChan a, OutChan a) +newChan :: UnagiPrim a=> IO (InChan a, OutChan a) newChan = newChanStarting (maxBound - 10) -- lets us test counter overflow in tests and normal course of operation -- | Return a lazy list representing the contents of the supplied OutChan, much -- like System.IO.hGetContents. -getChanContents :: Prim a=> OutChan a -> IO [a] +getChanContents :: UnagiPrim a=> OutChan a -> IO [a] getChanContents ch = unsafeInterleaveIO (do x <- readChan ch xs <- getChanContents ch @@ -47,6 +50,6 @@ -- | Write an entire list of items to a chan type. Writes here from multiple -- threads may be interleaved, and infinite lists are supported. -writeList2Chan :: Prim a=> InChan a -> [a] -> IO () +writeList2Chan :: UnagiPrim a=> InChan a -> [a] -> IO () {-# INLINABLE writeList2Chan #-} writeList2Chan ch = sequence_ . map (writeChan ch) diff --git a/src/Control/Concurrent/Chan/Unagi/Unboxed/Internal.hs b/src/Control/Concurrent/Chan/Unagi/Unboxed/Internal.hs --- a/src/Control/Concurrent/Chan/Unagi/Unboxed/Internal.hs +++ b/src/Control/Concurrent/Chan/Unagi/Unboxed/Internal.hs @@ -4,24 +4,59 @@ {-# WARNING "This library is unlikely to perform well on architectures without a fetch-and-add instruction" #-} #endif (sEGMENT_LENGTH + , UnagiPrim(..) , InChan(..), OutChan(..), ChanEnd(..), Cell, Stream(..), ElementArray(..), SignalIntArray , readElementArray, writeElementArray - , NextSegment(..), StreamHead(..) + , NextSegment(..), StreamHead(..), segSource , newChanStarting, writeChan, readChan, readChanOnException - , dupChan + , dupChan, tryReadChan + -- for NoBlocking.Unboxed + , moveToNextCell, waitingAdvanceStream, cellEmpty ) where -- Forked from src/Control/Concurrent/Chan/Unagi/Internal.hs at 443465. See -- that implementation for additional details and notes which we omit here. -- --- Internals exposed for testing. +-- Internals exposed for testing and for re-use in Unagi.NoBlocking.Unboxed -- --- TODO --- - Look at how ByteString is implemented; maybe that approach with --- ForeignPtr is better in some ways, or perhaps we can use their Internals? --- - we can make IndexedMVar () and always write to ByteString --- - Also 'vector' lib +-- TODO integration w/ ByteString +-- - we'd need to make IndexedMVar () and always write to ByteString +-- - we'd need to switch to Storable probably. +-- - exporting slices of elements as ByteString +-- - lazy bytestring would be easiest because of segment boundaries +-- - might be tricky to do blocking checking efficiently +-- - writing bytearrays +-- - fast with memcpy +-- +-- TODO MAYBE another variation: +-- Either with a single reader, or a counter that tracks readers as they +-- exit a segment (so that we know when can be manually 'free'd), allowing +-- use of unmanaged memory, and: +-- - creatable with calloc +-- - replace IORef with an unboxed 1-element array holding Addr of MutableByteArray? or something... +-- - nullPtr can be used to get us references of (Maybe a) +-- > IORef (StreamHead ) +-- > Int + (Stream ) +-- > arr + arr + IndexedMvar + Maybe Stream +-- - We would need to move IndexedMvar into streamhead +-- - No CAS for Ptr/ForeignPtr but we can probably extract the mutablebytearray for CAS +-- Data.Primitive.ByteArray.mutableByteArrayContents ~> Addr +-- , and ForeignPtr holds an Addr# + MutableByteArray internally... +-- , use GHC.ForeignPtr and wrap MutableByteArray in PlainPtr and off to races +-- - we can re-use read segments as soon as they pass. +-- +-- TODO GHC 7.10 and/or someday: +-- - use segment length of e.g. 1022 to account for MutableByteArray +-- header, then align to cache line (note: we don't really need to use +-- div/mod here; just subtraction) This could be done in all +-- implementations. (boxed arrays are: 3 + n/128 + n words?? Who knows...) +-- - use a smaller sigArr of 1024 bytes (just makes segSource a little cheaper) +-- - here use a clever fetchAndAdd to distinguish 4 different cells (+0001, vs +0100, etc) +-- - the NoBlocking can read/write to individual bytes +-- - calloc for mutableByteArray, when/if available +-- - non-temporal writes that bypass the cache? See: http://lwn.net/Articles/255364/ +-- - SIMD stuff for batch writing, or zeroing, etc. etc import Data.IORef @@ -33,11 +68,15 @@ import Control.Monad import Control.Applicative import Data.Bits -import Data.Typeable(Typeable) import GHC.Exts(inline) -import Utilities +-- For instances: +import Data.Typeable(Typeable) +import Data.Int(Int8,Int16,Int32,Int64) +import Data.Word(Word,Word8,Word16,Word32,Word64) +import Utilities import Control.Concurrent.Chan.Unagi.Constants +import qualified Control.Concurrent.Chan.Unagi.NoBlocking.Types as UT -- | The write end of a channel created with 'newChan'. newtype InChan a = InChan (ChanEnd a) @@ -58,8 +97,8 @@ -- InChan & OutChan are mostly identical, sharing a stream, but with -- independent counters data ChanEnd a = - -- Both Chan ends must start with the same counter value. - ChanEnd !AtomicCounter + ChanEnd -- 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)) @@ -70,14 +109,6 @@ -- The array we actually store our Prim elements in newtype ElementArray a = ElementArray (P.MutableByteArray RealWorld) --- TODO --- - we could easily use 'vector' to support a wider array of primitive --- elements here. --- - and what about Storable? --- see http://stackoverflow.com/q/4908880/176841 --- - else test combining signal and element arrays into a single one that --- places signal cell next to element cell, and use Addr to access? --- (see also TODOs under Stream) readElementArray :: (P.Prim a)=> ElementArray a -> Int -> IO a {-# INLINE readElementArray #-} @@ -110,25 +141,93 @@ cellBlocking = 2 -segSource :: forall a. (P.Prim a)=> IO (SignalIntArray, ElementArray a) --ScopedTypeVariables +-- NOTE: attempts to make allocation and initialization faster via copying, or +-- other tricks failed; although a calloc was about 2x faster (but that was for +-- unmanaged memory) +segSource :: forall a. (UnagiPrim a)=> IO (SignalIntArray, ElementArray a) --ScopedTypeVariables {-# INLINE segSource #-} segSource = do -- A largish pinned array seems like it would be the best choice here. sigArr <- P.newAlignedPinnedByteArray (P.sizeOf cellEmpty `unsafeShiftL` lOG_SEGMENT_LENGTH) -- times sEGMENT_LENGTH (P.alignment cellEmpty) + -- NOTE: we need these to be aligned to (some multiple of) Word boundaries + -- for magic trick to be correct, and for assumptions about atomicity of + -- loads/stores to hold! eArr <- P.newAlignedPinnedByteArray - (P.sizeOf (undefined :: a) `unsafeShiftL` lOG_SEGMENT_LENGTH) + (P.sizeOf (undefined :: a) `unsafeShiftL` lOG_SEGMENT_LENGTH) (P.alignment (undefined :: a)) P.setByteArray sigArr 0 sEGMENT_LENGTH cellEmpty + -- If no atomicUnicorn then we always check in at sigArr, so no need to + -- initialize eArr: + maybe (return ()) + (P.setByteArray eArr 0 sEGMENT_LENGTH) (atomicUnicorn :: Maybe a) return (sigArr, ElementArray eArr) + -- NOTE: We always CAS this into place which provides write barrier, such + -- that arrays are fully initialized before they can be read. No + -- corresponding barrier is needed in waitingAdvanceStream. +-- | Our class of types supporting primitive array operations. Instance method +-- definitions are architecture-dependent. +class (P.Prim a, Eq a)=> UnagiPrim a where + -- | When the read and write operations of the underlying @Prim@ instances + -- on aligned memory are atomic, this may be set to @Just x@ where @x@ is + -- some rare (i.e. unlikely to occur frequently in your data) magic value; + -- this might help speed up some @UnagiPrim@ operations. + -- + -- Where those 'Prim' instance operations are not atomic, this *must* be + -- set to @Nothing@. + atomicUnicorn :: Maybe a + atomicUnicorn = Nothing + + +-- These ought all to be atomic for 32-bit or 64-bit systems: +instance UnagiPrim Char where + atomicUnicorn = Just '\1010101' +instance UnagiPrim Float where + atomicUnicorn = Just 0xDADADA +instance UnagiPrim Int where + atomicUnicorn = Just 0xDADADA +instance UnagiPrim Int8 where + atomicUnicorn = Just 113 +instance UnagiPrim Int16 where + atomicUnicorn = Just 0xDAD +instance UnagiPrim Int32 where + atomicUnicorn = Just 0xDADADA +instance UnagiPrim Word where + atomicUnicorn = Just 0xDADADA +instance UnagiPrim Word8 where + atomicUnicorn = Just 0xDA +instance UnagiPrim Word16 where + atomicUnicorn = Just 0xDADA +instance UnagiPrim Word32 where + atomicUnicorn = Just 0xDADADADA +instance UnagiPrim P.Addr where + atomicUnicorn = Just P.nullAddr +-- These should conservatively be expected to be atomic only on 64-bit +-- machines: +instance UnagiPrim Int64 where +#ifdef IS_64_BIT + atomicUnicorn = Just 0xDADADADADADA +#endif +instance UnagiPrim Word64 where +#ifdef IS_64_BIT + atomicUnicorn = Just 0xDADADADADADA +#endif +instance UnagiPrim Double where +#ifdef IS_64_BIT + atomicUnicorn = Just 0xDADADADADADA +#endif + +-- NOTE: we tried combining the SignalIntArray and ElementArray into a single +-- bytearray in the unagi-unboxed-combined-bytearray branch but saw no +-- significant improvement. data Stream a = Stream !SignalIntArray !(ElementArray a) -- For coordinating blocking between reader/writer; NOTE [1] - !(IndexedMVar a) + (IndexedMVar a) -- N.B. must remain non-strict for NoBlocking.Unboxed -- The next segment in the stream; NOTE [2] !(IORef (NextSegment a)) -- [1] An important property: we can switch out this implementation as long @@ -136,20 +235,15 @@ -- -- [2] new segments are allocated and put here as we go, with threads -- cooperating to allocate new segments: --- TODO --- - we could replace Stream with a single funky MutableByteArray, even --- replacing the IORef with a stored Addr to the next segment, which is --- initialized to maxBound (an impossible value hopefully?) indicating --- NoSegment --- - except for our MVarIndexed in current implementation data NextSegment a = NoSegment | Next !(Stream a) -- we expose `startingCellOffset` for debugging correct behavior with overflow: -newChanStarting :: (P.Prim a)=> Int -> IO (InChan a, OutChan a) +newChanStarting :: UnagiPrim a=> Int -> IO (InChan a, OutChan a) {-# INLINE newChanStarting #-} newChanStarting !startingCellOffset = do - stream <- uncurry Stream <$> segSource <*> newIndexedMVar <*> newIORef NoSegment + (sigArr0,eArr0) <- segSource + stream <- Stream sigArr0 eArr0 <$> newIndexedMVar <*> newIORef NoSegment let end = ChanEnd <$> newCounter (startingCellOffset - 1) <*> newIORef (StreamHead startingCellOffset stream) @@ -163,76 +257,120 @@ {-# INLINE dupChan #-} dupChan (InChan (ChanEnd counter streamHead)) = do hLoc <- readIORef streamHead - loadLoadBarrier -- NOTE [1] + loadLoadBarrier wCount <- readCounter counter OutChan <$> (ChanEnd <$> newCounter wCount <*> newIORef hLoc) - -- [1] We must read the streamHead before inspecting the counter; otherwise, - -- as writers write, the stream head pointer may advance past the cell - -- indicated by wCount and the first cell become unreachable. + -- | Write a value to the channel. -writeChan :: (P.Prim a)=> InChan a -> a -> IO () +writeChan :: UnagiPrim a=> InChan a -> a -> IO () {-# INLINE writeChan #-} writeChan (InChan ce) = \a-> mask_ $ do - (segIx, (Stream sigArr eArr mvarIndexed next)) <- moveToNextCell ce + (segIx, (Stream sigArr eArr mvarIndexed next), maybeUpdateStreamHead) <- moveToNextCell ce + maybeUpdateStreamHead -- NOTE!: must write element before signaling with CAS: writeElementArray eArr segIx a -#ifdef NOT_x86 - -- TODO Should we include this for correctness sake? Will GHC ever move a write ahead of a CAS? - -- CAS provides a full barrier on x86; otherwise we need to make sure the - -- read above occurs before our fetch-and-add: - writeBarrier -#endif - actuallyWas <- casByteArrayInt sigArr segIx cellEmpty cellWritten - -- try to pre-allocate next segment; NOTE [1] + actuallyWas <- casByteArrayInt sigArr segIx cellEmpty cellWritten -- NOTE[1] + -- try to pre-allocate next segment: when (segIx == 0) $ void $ waitingAdvanceStream next 0 case actuallyWas of + -- CAS SUCCEEDED: -- 0 {- Empty -} -> return () + -- CAS FAILED: -- 2 {- Blocking -} -> putMVarIx mvarIndexed segIx a + 1 {- Written -} -> error "Nearly Impossible! Expected Blocking" _ -> error "Invalid signal seen in writeChan!" - -- [1] the writer which arrives first to the first cell of a new segment is - -- tasked (somewhat arbitrarily) with trying to pre-allocate the *next* - -- segment hopefully ahead of any readers or writers who might need it. This - -- will race with any reader *or* writer that tries to read the next segment - -- and finds it's empty (see `waitingAdvanceStream`); when this wins - -- (hopefully the vast majority of the time) we avoid a throughput hit. + -- [1] casByteArrayInt provides the write barrier we need here to make sure + -- GHC maintains our ordering such that the element is written before we + -- signal its availability with the CAS to sigArr that follows. See [2] in + -- readChanOnExceptionUnmasked. -readChanOnExceptionUnmasked :: (P.Prim a)=> (IO a -> IO a) -> OutChan a -> IO a +readChanOnExceptionUnmasked :: UnagiPrim a=> (IO a -> IO a) -> OutChan a -> IO a {-# INLINE readChanOnExceptionUnmasked #-} readChanOnExceptionUnmasked h = \(OutChan ce)-> do - (segIx, (Stream sigArr eArr mvarIndexed _)) <- moveToNextCell ce - -- NOTE!: must read signal before reading element. No barrier necessary. - let readBlocking = inline h $ readMVarIx mvarIndexed segIx -- NOTE [1] - -- optimistically try read w/out CAS - sig <- P.readByteArray sigArr segIx - case (sig :: Int) of - 1 {- Written -} -> readElementArray eArr segIx - 2 {- Blocking -} -> readBlocking - _ -> do - actuallyWas <- casByteArrayInt sigArr segIx cellEmpty cellBlocking - case actuallyWas of - -- succeeded writing Empty; proceed with blocking - 0 {- Empty -} -> readBlocking - -- else in the meantime, writer wrote - 1 {- Written -} -> readElementArray eArr segIx - -- else in the meantime a dupChan reader read, blocking - 2 {- Blocking -} -> readBlocking - _ -> error "Invalid signal seen in readChanOnExceptionUnmasked!" + (segIx, (Stream sigArr eArr mvarIndexed _), maybeUpdateStreamHead) <- moveToNextCell ce + maybeUpdateStreamHead + let readBlocking = inline h $ readMVarIx mvarIndexed segIx -- NOTE [1] + readElem = readElementArray eArr segIx + slowRead = do + -- Assume probably blocking (Note: casByteArrayInt is a full barrier) + actuallyWas <- casByteArrayInt sigArr segIx cellEmpty cellBlocking -- NOTE [2] + case actuallyWas of + -- succeeded writing Empty; proceed with blocking + 0 {- Empty -} -> readBlocking + -- else in the meantime, writer wrote + 1 {- Written -} -> readElem + -- else in the meantime a dupChan reader read, blocking + 2 {- Blocking -} -> readBlocking + _ -> error "Invalid signal seen in readChanOnExceptionUnmasked!" + -- If we know writes of this element are atomic, we can determine if the + -- element has been written, and possibly return it without consulting + -- sigArr. + case atomicUnicorn of + Just magic -> do + el <- readElem + if (el /= magic) + -- We know `el` was atomically written: + then return el + else slowRead + Nothing -> slowRead -- [1] we must use `readMVarIx` here to support `dupChan`. It's also -- important that the behavior of readMVarIx be identical to a readMVar on -- the same MVar. + -- + -- [2] casByteArrayInt provides the loadLoadBarrier we need here. See [1] in + -- writeChan. + +-- TODO we might want also a blocking `IO a` returned here, or use an opaque +-- Element type supporting blocking, since otherwise calling `tryReadChan` we +-- give up the ability to block on that element. Please open an issue if you +-- need this in the meantime. And also handling of lost elements on async +-- exceptions. And also isActive... + +-- | Returns immediately with an @'UT.Element' a@ future, which returns one +-- unique element when it becomes available via 'UT.tryRead'. If you're using +-- this function exclusively you might find the implementation in +-- "Control.Concurrent.Chan.Unagi.NoBlocking.Unboxed" is faster. +-- +-- /Note re. exceptions/: When an async exception is raised during a @tryReadChan@ +-- the message that the read would have returned is likely to be lost, just as +-- it would be when raised directly after this function returns. +tryReadChan :: UnagiPrim a=> OutChan a -> IO (UT.Element a) +{-# INLINE tryReadChan #-} +tryReadChan (OutChan ce) = do -- no masking needed +-- NOTE: implementation adapted from readChanOnExceptionUnmasked: + (segIx, (Stream sigArr eArr mvarIndexed _), maybeUpdateStreamHead) <- moveToNextCell ce + maybeUpdateStreamHead + let readElem = readElementArray eArr segIx + slowRead = do + sig <- P.readByteArray sigArr segIx + case (sig :: Int) of + 0 {- Empty -} -> return Nothing + 1 {- Written -} -> loadLoadBarrier >> Just <$> readElem + 2 {- Blocking -} -> tryReadMVarIx mvarIndexed segIx + _ -> error "Invalid signal seen in tryReadChan!" + return $ UT.Element $ + case atomicUnicorn of + Just magic -> do + el <- readElem + if (el /= magic) + then return $ Just el + else slowRead + Nothing -> slowRead + + -- | Read an element from the chan, blocking if the chan is empty. -- -- /Note re. exceptions/: When an async exception is raised during a @readChan@ -- the message that the read would have returned is likely to be lost, even when -- the read is known to be blocked on an empty queue. If you need to handle -- this scenario, you can use 'readChanOnException'. -readChan :: (P.Prim a)=> OutChan a -> IO a +readChan :: UnagiPrim a=> OutChan a -> IO a {-# INLINE readChan #-} readChan = readChanOnExceptionUnmasked id @@ -244,24 +382,20 @@ -- The second argument is a handler that takes a blocking IO action returning -- the element, and performs some recovery action. When the handler is called, -- the passed @IO a@ is the only way to access the element. -readChanOnException :: (P.Prim a)=> OutChan a -> (IO a -> IO ()) -> IO a +readChanOnException :: UnagiPrim a=> OutChan a -> (IO a -> IO ()) -> IO a {-# INLINE readChanOnException #-} readChanOnException c h = mask_ $ readChanOnExceptionUnmasked (\io-> io `onException` (h io)) c + + -- increments counter, finds stream segment of corresponding cell (updating the -- stream head pointer as needed), and returns the stream segment and relative -- index of our cell. -moveToNextCell :: (P.Prim a)=> ChanEnd a -> IO (Int, Stream a) +moveToNextCell :: UnagiPrim a=> ChanEnd a -> IO (Int, Stream a, IO ()) {-# INLINE moveToNextCell #-} moveToNextCell (ChanEnd counter streamHead) = do (StreamHead offset0 str0) <- readIORef streamHead - -- NOTE [3] -#ifdef NOT_x86 - -- fetch-and-add is a full barrier on x86; otherwise we need to make sure - -- the read above occurrs before our fetch-and-add: - loadLoadBarrier -#endif ix <- incrCounter 1 counter let (segsAway, segIx) = assert ((ix - offset0) >= 0) $ divMod_sEGMENT_LENGTH $! (ix - offset0) @@ -269,36 +403,28 @@ {-# INLINE go #-} go 0 str = return str go !n (Stream _ _ _ next) = - waitingAdvanceStream next (nEW_SEGMENT_WAIT*segIx) -- NOTE [1] + waitingAdvanceStream next (nEW_SEGMENT_WAIT*segIx) >>= go (n-1) str <- go segsAway str0 - when (segsAway > 0) $ do - let !offsetN = - offset0 + (segsAway `unsafeShiftL` lOG_SEGMENT_LENGTH) --(segsAway*sEGMENT_LENGTH) - writeIORef streamHead $ StreamHead offsetN str -- NOTE [2] - return (segIx,str) - -- [1] All readers or writers needing to work with a not-yet-created segment - -- race to create it, but those past index 0 have progressively long waits; 20 - -- is chosen as 20 readIORefs should be more than enough time for writer/reader - -- 0 to add the new segment (if it's not descheduled). - -- - -- [2] advancing the stream head pointer on segIx == sEGMENT_LENGTH - 1 would - -- be more correct, but this is simpler here. This may move the head pointer - -- BACKWARDS if the thread was descheduled, but that's not a correctness - -- issue. - -- - -- [3] There is a theoretical race condition here: thread reads head and is - -- descheduled, meanwhile other readers/writers increment counter one full - -- lap; when we increment we think we've found our cell in what is actually a - -- very old segment. However in this scenario all addressable memory will - -- have been consumed just by the array pointers whivh haven't been able to - -- be GC'd. So I don't think this is something to worry about. + -- We need to return this continuation here for NoBlocking.Unboxed, which + -- needs to perform this action at different points in the reader and + -- writer. + let !maybeUpdateStreamHead = do + when (segsAway > 0) $ do + let !offsetN = + offset0 + (segsAway `unsafeShiftL` lOG_SEGMENT_LENGTH) --(segsAway*sEGMENT_LENGTH) + writeIORef streamHead $ StreamHead offsetN str + touchIORef streamHead -- NOTE [1] + return (segIx,str, maybeUpdateStreamHead) + -- [1] For NoBlocking.Unboxed: this helps ensure that streamHead is not GC'd + -- until `maybeUpdateStreamHead` is run in calling function. For correctness + -- of `isActive`. -- thread-safely try to fill `nextSegRef` at the next offset with a new -- segment, waiting some number of iterations (for other threads to handle it). -- Returns nextSegRef's StreamSegment. -waitingAdvanceStream :: (P.Prim a)=> IORef (NextSegment a) -> Int -> IO (Stream a) +waitingAdvanceStream :: (UnagiPrim a)=> IORef (NextSegment a) -> Int -> IO (Stream a) waitingAdvanceStream nextSegRef = go where go !wait = assert (wait >= 0) $ do tk <- readForCAS nextSegRef diff --git a/src/Data/Atomics/Counter/Fat.hs b/src/Data/Atomics/Counter/Fat.hs --- a/src/Data/Atomics/Counter/Fat.hs +++ b/src/Data/Atomics/Counter/Fat.hs @@ -12,12 +12,13 @@ import Control.Monad.Primitive(RealWorld) import Data.Primitive.ByteArray import Data.Atomics(fetchAddByteArrayInt) +import Control.Exception(assert) newtype AtomicCounter = AtomicCounter (MutableByteArray RealWorld) -sIZEOF_CACHELINE , cACHELINE_PADDED_INT_IX :: Int +sIZEOF_CACHELINE :: Int +{-# INLINE sIZEOF_CACHELINE #-} sIZEOF_CACHELINE = 64 -cACHELINE_PADDED_INT_IX = (sIZEOF_CACHELINE `quot` 2) `quot` sIZEOF_INT newCounter :: Int -> IO AtomicCounter {-# INLINE newCounter #-} @@ -25,15 +26,17 @@ arr <- newAlignedPinnedByteArray sIZEOF_CACHELINE sIZEOF_CACHELINE - writeByteArray arr cACHELINE_PADDED_INT_IX n - return (AtomicCounter arr) + writeByteArray arr 0 n + -- out of principle: + assert (sIZEOF_INT < sIZEOF_CACHELINE) $ + return (AtomicCounter arr) incrCounter :: Int -> AtomicCounter -> IO Int {-# INLINE incrCounter #-} incrCounter incr (AtomicCounter arr) = - fetchAddByteArrayInt arr cACHELINE_PADDED_INT_IX incr + fetchAddByteArrayInt arr 0 incr readCounter :: AtomicCounter -> IO Int {-# INLINE readCounter #-} readCounter (AtomicCounter arr) = - readByteArray arr cACHELINE_PADDED_INT_IX + readByteArray arr 0 diff --git a/src/Utilities.hs b/src/Utilities.hs --- a/src/Utilities.hs +++ b/src/Utilities.hs @@ -1,11 +1,12 @@ -{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE BangPatterns , MagicHash , UnboxedTuples #-} module Utilities ( -- * Utility Chans -- ** Indexed MVars IndexedMVar() - , newIndexedMVar, putMVarIx, readMVarIx - -- ** Other stuff + , newIndexedMVar, putMVarIx, readMVarIx, tryReadMVarIx + -- * Other stuff , nextHighestPowerOfTwo + , touchIORef ) where import Control.Concurrent.MVar @@ -15,6 +16,10 @@ import Data.Word import Data.Atomics import Data.IORef +import GHC.Prim(touch#) +import GHC.IORef(IORef(..)) +import GHC.STRef(STRef(..)) +import GHC.Base(IO(..)) -- For now: a reverse-ordered assoc list; an IntMap might be better newtype IndexedMVar a = IndexedMVar (IORef [(Int, MVar a)]) @@ -31,11 +36,17 @@ readMVarIx mvIx i = do readMVar =<< getMVarIx mvIx i +tryReadMVarIx :: IndexedMVar a -> Int -> IO (Maybe a) +{-# INLINE tryReadMVarIx #-} +tryReadMVarIx mvIx i = do + tryReadMVar =<< getMVarIx mvIx i + putMVarIx :: IndexedMVar a -> Int -> a -> IO () {-# INLINE putMVarIx #-} putMVarIx mvIx i a = do flip putMVar a =<< getMVarIx mvIx i +-- NOTE: this uses atomic actions to stay async exception safe: getMVarIx :: IndexedMVar a -> Int -> IO (MVar a) {-# INLINE getMVarIx #-} getMVarIx (IndexedMVar v) i = do @@ -80,3 +91,11 @@ nhp2 where maxPowerOfTwo = (floor $ sqrt $ (fromIntegral (maxBound :: Int)::Float)) ^ (2::Int) + +-- I'm not sure what happens if we try to use touch from +-- Control.Monad.Primitive on our boxed IORef (if it gets unboxed), so we do +-- this: +touchIORef :: IORef a -> IO () +touchIORef (IORef (STRef v)) = IO $ \s -> + case touch# v s of + s' -> (# s', () #) diff --git a/tests/Atomics.hs b/tests/Atomics.hs --- a/tests/Atomics.hs +++ b/tests/Atomics.hs @@ -26,7 +26,7 @@ putStrLn "OK" -- ------ putStr " CAS... " - testConsistentSuccessFailure + testNextistentSuccessFailure putStrLn "OK" -- catch real stupid bugs before machine gets hot: @@ -93,8 +93,8 @@ -- Test these assumptions: -- 1) If a CAS fails in thread 1 then another CAS (in thread 2, say) succeeded; i.e. no false negatives -- 2) In the case that thread 1's CAS failed, the ticket returned with (False,tk) will contain that newly-written value from thread 2 -testConsistentSuccessFailure :: IO () -testConsistentSuccessFailure = do +testNextistentSuccessFailure :: IO () +testNextistentSuccessFailure = do var <- newIORef "0" sem <- newIORef (0::Int) @@ -128,7 +128,7 @@ else do print res1 print res2 error "FAILURE!" - examine _ = error "Fix testConsistentSuccessFailure" + examine _ = error "Fix testNextistentSuccessFailure" forkSync :: IORef Int -> Int -> IO () -> IO ThreadId diff --git a/tests/Deadlocks.hs b/tests/Deadlocks.hs --- a/tests/Deadlocks.hs +++ b/tests/Deadlocks.hs @@ -23,8 +23,45 @@ putStr $ " Checking for deadlocks from killed writer, x"++show tries++"... " checkDeadlocksWriter unagiImpl tries putStrLn "OK" + + putStrLn "===================" + putStrLn "Testing Unagi (with tryReadChan):" + -- ------ + putStr $ " Checking for deadlocks from killed reader, x"++show tries++"... " + checkDeadlocksReader unagiTryReadImpl tries + putStrLn "OK" + -- No real need to checkDeadlocksWriter for tryReadChan. + + putStrLn "===================" + putStrLn "Testing Unagi.NoBlocking:" + -- ------ + putStr $ " Checking for deadlocks from killed reader, x"++show tries++"... " + checkDeadlocksReader unagiNoBlockingImpl tries + putStrLn "OK" + -- ------ + putStr $ " Checking for deadlocks from killed writer, x"++show tries++"... " + checkDeadlocksWriter unagiNoBlockingImpl tries + putStrLn "OK" putStrLn "===================" + putStrLn "Testing Unagi.NoBlocking.Unboxed:" + -- ------ + putStr $ " Checking for deadlocks from killed reader, x"++show tries++"... " + checkDeadlocksReader unagiNoBlockingUnboxedImpl tries + putStrLn "OK" + -- ------ + putStr $ " Checking for deadlocks from killed writer, x"++show tries++"... " + checkDeadlocksWriter unagiNoBlockingUnboxedImpl tries + putStrLn "OK" + + putStrLn "===================" + putStrLn "Testing Unagi.Unboxed (with tryReadChan):" + -- ------ + putStr $ " Checking for deadlocks from killed reader, x"++show tries++"... " + checkDeadlocksReader unboxedUnagiTryReadImpl tries + putStrLn "OK" + + putStrLn "===================" putStrLn "Testing Unagi.Unboxed:" -- ------ putStr $ " Checking for deadlocks from killed reader, x"++show tries++"... " @@ -41,6 +78,11 @@ 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 reader (tryReadChan), x"++show tries++"... " + -- bounds must be > 10000 here (note actual bounds rounded up to power of 2): + checkDeadlocksReader (unagiBoundedTryReadImpl 50000) tries putStrLn "OK" -- ------ putStr $ " Checking for deadlocks from killed writer, x"++show tries++"... " diff --git a/tests/DupChan.hs b/tests/DupChan.hs --- a/tests/DupChan.hs +++ b/tests/DupChan.hs @@ -22,7 +22,41 @@ putStr " Writer/dupChan+Reader... " replicateM_ 1000 $ dupChanTest2 unagiImpl 10000 putStrLn "OK" + putStrLn "===================" + putStrLn "Test dupChan Unagi (with tryReadChan):" + -- ------ + putStr " Reader/Reader... " + replicateM_ 1000 $ dupChanTest1 unagiTryReadImpl 50000 + putStrLn "OK" + -- ------ + putStr " Writer/dupChan+Reader... " + replicateM_ 1000 $ dupChanTest2 unagiTryReadImpl 10000 + putStrLn "OK" + + putStrLn "===================" + putStrLn "Test dupChan Unagi.NoBlocking:" + -- ------ + putStr " Reader/Reader... " + replicateM_ 1000 $ dupChanTest1 unagiNoBlockingImpl 50000 + putStrLn "OK" + -- ------ + putStr " Writer/dupChan+Reader... " + replicateM_ 1000 $ dupChanTest2 unagiNoBlockingImpl 10000 + putStrLn "OK" + + putStrLn "===================" + putStrLn "Test dupChan Unagi.NoBlocking.Unboxed:" + -- ------ + putStr " Reader/Reader... " + replicateM_ 1000 $ dupChanTest1 unagiNoBlockingUnboxedImpl 50000 + putStrLn "OK" + -- ------ + putStr " Writer/dupChan+Reader... " + replicateM_ 1000 $ dupChanTest2 unagiNoBlockingUnboxedImpl 10000 + putStrLn "OK" + + putStrLn "===================" putStrLn "Test dupChan Unagi.Unboxed:" -- ------ putStr " Reader/Reader... " @@ -34,16 +68,29 @@ putStrLn "OK" putStrLn "===================" - putStrLn "Test dupChan Unagi.Bounded" + putStrLn "Test dupChan Unagi.Unboxed (with tryReadChan):" + -- ------ + putStr " Reader/Reader... " + replicateM_ 1000 $ dupChanTest1 unboxedUnagiTryReadImpl 50000 + putStrLn "OK" + -- ------ + putStr " Writer/dupChan+Reader... " + replicateM_ 1000 $ dupChanTest2 unboxedUnagiTryReadImpl 10000 + putStrLn "OK" + + putStrLn "===================" + putStrLn "Test dupChan Unagi.Bounded (and with tryRead)" -- 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 + replicateM_ 100 $ dupChanTest1 (unagiBoundedImpl bounds) n + replicateM_ 100 $ dupChanTest1 (unagiBoundedTryReadImpl bounds) n putStrLn "OK" forM_ [2, 1024, 65536] $ \bounds-> do putStr $ " Writer/dupChan+Reader with bounds "++(show bounds)++"... " - replicateM_ 1000 $ dupChanTest2 (unagiBoundedImpl bounds) 10000 + replicateM_ 100 $ dupChanTest2 (unagiBoundedImpl bounds) 10000 + replicateM_ 100 $ dupChanTest2 (unagiBoundedTryReadImpl bounds) 10000 putStrLn "OK" -- Check output where dupChan at known point in input stream, with two diff --git a/tests/Implementations.hs b/tests/Implementations.hs --- a/tests/Implementations.hs +++ b/tests/Implementations.hs @@ -3,15 +3,47 @@ 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 +import qualified Control.Concurrent.Chan.Unagi.NoBlocking as UN +import qualified Control.Concurrent.Chan.Unagi.NoBlocking.Unboxed as UNU +import Control.Concurrent(yield, threadDelay) type Implementation inc outc a = (IO (inc a, outc a), inc a -> a -> IO (), outc a -> IO a, inc a -> IO (outc a)) -unagiImpl :: Implementation U.InChan U.OutChan a +unagiImpl , unagiTryReadImpl :: Implementation U.InChan U.OutChan a unagiImpl = (U.newChan, U.writeChan, U.readChan, U.dupChan) +unagiTryReadImpl = (U.newChan, U.writeChan, u_trying_readChan, U.dupChan) -unboxedUnagiImpl :: (P.Prim a)=> Implementation UU.InChan UU.OutChan a +unboxedUnagiImpl , unboxedUnagiTryReadImpl:: (UU.UnagiPrim a)=> Implementation UU.InChan UU.OutChan a unboxedUnagiImpl = (UU.newChan, UU.writeChan, UU.readChan, UU.dupChan) +unboxedUnagiTryReadImpl = (UU.newChan, UU.writeChan, uu_trying_readChan, UU.dupChan) -unagiBoundedImpl :: Int -> Implementation UB.InChan UB.OutChan a +unagiBoundedImpl , unagiBoundedTryReadImpl:: Int -> Implementation UB.InChan UB.OutChan a unagiBoundedImpl n = (UB.newChan n, UB.writeChan, UB.readChan, UB.dupChan) +unagiBoundedTryReadImpl n = (UB.newChan n, UB.writeChan, ub_trying_readChan, UB.dupChan) + +-- We use our yield "blocking" readChan here, and below: +unagiNoBlockingImpl :: Implementation UN.InChan UN.OutChan a +unagiNoBlockingImpl = (UN.newChan, UN.writeChan, UN.readChan yield, UN.dupChan) + +unagiNoBlockingUnboxedImpl :: (UU.UnagiPrim a)=> Implementation UNU.InChan UNU.OutChan a +unagiNoBlockingUnboxedImpl = (UNU.newChan, UNU.writeChan, UNU.readChan yield, UNU.dupChan) + +-- These have same semantics as corresponding `readChan`, so this is an easy +-- way to do smoke tests of `tryReadChan`: +uu_trying_readChan :: (UU.UnagiPrim a)=> UU.OutChan a -> IO a +uu_trying_readChan oc = do + e <- UU.tryReadChan oc + let go = UU.tryRead e >>= maybe (threadDelay 1000 >> go) return + go + +u_trying_readChan :: U.OutChan a -> IO a +u_trying_readChan oc = do + e <- U.tryReadChan oc + let go = U.tryRead e >>= maybe (threadDelay 1000 >> go) return + go + +ub_trying_readChan :: UB.OutChan a -> IO a +ub_trying_readChan oc = do + e <- UB.tryReadChan oc + let go = UB.tryRead e >>= maybe (threadDelay 1000 >> go) return + go diff --git a/tests/Main.hs b/tests/Main.hs --- a/tests/Main.hs +++ b/tests/Main.hs @@ -14,6 +14,8 @@ import Unagi import UnagiUnboxed import UnagiBounded +import UnagiNoBlocking +import UnagiNoBlockingUnboxed -- Other import Atomics @@ -52,5 +54,7 @@ unagiMain unagiUnboxedMain unagiBoundedMain + unagiNoBlockingMain + unagiNoBlockingUnboxedMain - putStrLn "ALL DONE!" + putStrLn "ALL TESTS PASSED!" diff --git a/tests/Smoke.hs b/tests/Smoke.hs --- a/tests/Smoke.hs +++ b/tests/Smoke.hs @@ -2,7 +2,7 @@ module Smoke (smokeMain) where import Control.Monad -import Control.Concurrent(forkIO,threadDelay) +import Control.Concurrent(forkIO,threadDelay,myThreadId,ThreadId) import qualified Control.Concurrent.Chan as C import Data.List import Control.Exception @@ -11,12 +11,19 @@ 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 +-- Rethrow +forkCatching :: Bool -> String -> IO () -> IO ThreadId +forkCatching expectingBlock nm io = do + mainTid <- myThreadId + let lg e = do putStrLn $ "!!! EXCEPTION IN "++nm++": "++(show e) + throwTo mainTid e + forkIO $ io `E.catches` [ + E.Handler (\e -> when (not expectingBlock) $ lg (e :: BlockedIndefinitelyOnMVar)) + , E.Handler (\e -> case (e :: AsyncException) of + ThreadKilled -> return () + _ -> lg e ) + ] + smokeMain :: IO () @@ -25,43 +32,95 @@ putStrLn "Testing Unagi:" -- ------ putStr " FIFO smoke test... " - fifoSmoke unagiImpl 100000 + fifoSmoke unagiImpl 1000000 putStrLn "OK" -- ------ testContention unagiImpl 2 2 1000000 + putStrLn "===================" + putStrLn "Testing Unagi (with tryReadChan):" + -- ------ + putStr " FIFO smoke test... " + fifoSmoke unagiTryReadImpl 1000000 + putStrLn "OK" + -- ------ + testContention unagiTryReadImpl 2 2 1000000 + putStrLn "===================" + putStrLn "Testing Unagi.NoBlocking:" + -- ------ + putStr " FIFO smoke test... " + fifoSmoke unagiNoBlockingImpl 1000000 + putStrLn "OK" + -- ------ + testContention unagiNoBlockingImpl 2 2 1000000 + + + putStrLn "===================" + putStrLn "Testing Unagi.NoBlocking.Unboxed:" + -- ------ + putStr " FIFO smoke test... " + fifoSmoke unagiNoBlockingUnboxedImpl 1000000 + putStrLn "OK" + -- ------ + testContention unagiNoBlockingUnboxedImpl 2 2 1000000 + + + putStrLn "===================" putStrLn "Testing Unagi.Unboxed:" -- ------ putStr " FIFO smoke test... " - fifoSmoke unboxedUnagiImpl 100000 + fifoSmoke unboxedUnagiImpl 1000000 putStrLn "OK" -- ------ testContention unboxedUnagiImpl 2 2 1000000 + putStrLn "===================" + putStrLn "Testing Unagi.Unboxed (with tryReadChan):" + -- ------ + putStr " FIFO smoke test... " + fifoSmoke unboxedUnagiTryReadImpl 1000000 + putStrLn "OK" + -- ------ + testContention unboxedUnagiTryReadImpl 2 2 1000000 + forM_ [1, 2, 4, 1024] $ \bounds-> do putStrLn "===================" - putStrLn $ "Testing Unagi.Bounded with bounds "++(show bounds) + putStrLn $ "Testing Unagi.Bounded (and with tryReadChan) with bounds "++(show bounds) -- ------ putStr " FIFO smoke test... " - fifoSmoke (unagiBoundedImpl bounds) 100000 + fifoSmoke (unagiBoundedImpl bounds) 1000000 + -- because this is slow: + when (bounds > 100) $ fifoSmoke (unagiBoundedTryReadImpl bounds) 1000000 putStrLn "OK" -- ------ testContention (unagiBoundedImpl bounds) 2 2 1000000 + -- because this is slow: + when (bounds > 100) $ testContention (unagiBoundedTryReadImpl bounds) 2 2 1000000 - ) `onException` (threadDelay 1000000) -- wait for lgErrs + ) `onException` (threadDelay 1000000) -- wait for forkCatching logging +-- Run two concurrent writer threads, making sure their respective sets of +-- writes arrived in order: fifoSmoke :: Implementation inc outc Int -> Int -> IO () fifoSmoke (newChan,writeChan,readChan,_) n = do (i,o) <- newChan - -- 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!" + let forkWriter p = void $ forkCatching False "fifoSmoke writeChan" $ mapM_ (writeChan i) p + forkWriter [1..n] + forkWriter [negate n .. -1] + -- Give a chance for writers to work on both cores, but we need the main + -- thread to run concurrently for bounded chans: + threadDelay 100000 + nsOut <- replicateM (n*2) $ readChan o + let (nsPos,nsNeg) = partition (>0) nsOut + unless (nsPos == [1..n] && nsNeg == [negate n .. -1]) $ + error $ "Cough!!"++(show nsOut) + +-- Break up a set of unique messages running them through multiple writers to +-- multiple readers (all concurrently), making sure they all came out the same testContention :: Implementation inc outc Int -> Int -> Int -> Int -> IO () testContention (newChan,writeChan,readChan,_) writers readers n = do let nNice = n - rem n (lcm writers readers) @@ -71,12 +130,13 @@ out <- C.newChan (i,o) <- newChan - -- some will get blocked indefinitely: - void $ replicateM readers $ forkIO $ lgErrs True "testContention readChan o"$ forever $ - readChan o >>= C.writeChan out + -- Real `readChan`s will get BlockedIndefinitelyOnMVar here, when o is dead, + -- but we need to kill them explicitly for our *TryReadImpl: + rIds <- replicateM readers $ forkCatching True "testContention readChan o"$ forever $ + readChan o >>= C.writeChan out 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 + mapM_ (forkCatching False "testContention writeChan i " . mapM_ (writeChan i)) groups ns <- replicateM nNice (C.readChan out) isEmpty <- C.isEmptyChan out @@ -86,6 +146,9 @@ 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 :(" + mapM_ (`throwTo` ThreadKilled) rIds + + -- --------- Helpers: diff --git a/tests/UnagiUnboxed.hs b/tests/UnagiUnboxed.hs --- a/tests/UnagiUnboxed.hs +++ b/tests/UnagiUnboxed.hs @@ -1,3 +1,4 @@ +{-# LANGUAGE RankNTypes , ScopedTypeVariables , BangPatterns #-} module UnagiUnboxed (unagiUnboxedMain) where -- Unagi-chan-specific tests @@ -10,6 +11,11 @@ import qualified Data.Primitive as P import Data.IORef +import Data.Int(Int8,Int16,Int32,Int64) +import Data.Word(Word,Word8,Word16,Word32,Word64) +import Data.Maybe +import Data.Typeable + import Control.Concurrent(forkIO,threadDelay) import Control.Concurrent.MVar import Control.Exception @@ -20,12 +26,19 @@ putStrLn "===================" putStrLn "Testing Unagi.Unboxed details:" -- ------ - -- ------ putStr "Smoke test at different starting offsets, spanning overflow... " mapM_ smoke $ [ (maxBound - UI.sEGMENT_LENGTH - 1) .. maxBound] ++ [minBound .. (minBound + UI.sEGMENT_LENGTH + 1)] putStrLn "OK" -- ------ + putStr "segSource sanity... " + applyToAllPrim segSourceMagicSanity + putStrLn "OK" + -- ------ + putStr "Atomicity of atomic unicorns... " + applyToAllPrim atomicUnicornAtomicicity + putStrLn "OK" + -- ------ putStr "Correct first write... " mapM_ correctFirstWrite [ (maxBound - 7), maxBound, minBound, 0] putStrLn "OK" @@ -40,8 +53,41 @@ smoke :: Int -> IO () -smoke n = smoke1 n >> smoke2 n +smoke n = do + smoke1 n + smoke2 n + -- test each of UnagiPrim + applyToAllPrim smokeManyElement + -- test for atomicUnicorns + applyToAllPrim smokeManyUnicorn +-- test a function against each Prim type elements not equal to atomicUnicorn +applyToAllPrim :: (forall e. (Num e, Typeable e, Show e, UnagiPrim e)=> e -> IO ()) -> IO () +applyToAllPrim f = do + f ('c' :: Char) + f (3.14159 :: Float) + f (-1.000000000000001 :: Double) + f (maxBound :: Int) + f (minBound :: Int8) + f (maxBound :: Int16) + f (minBound :: Int32) + f (maxBound :: Int64) + f (maxBound :: Word) + f (maxBound :: Word8) + f (minBound :: Word16) + f (minBound :: Word32) + f (minBound :: Word64) + f (P.nullAddr `P.plusAddr` 1024 :: P.Addr) + +-- TODO Maybe refactor & get rid of these +instance Show P.Addr where + show _ = "" + +instance Num P.Addr where + +instance Num Char where + + -- www.../rrr... spanning overflow smoke1 :: Int -> IO () smoke1 n = do @@ -49,9 +95,8 @@ let inp = [0 .. (UI.sEGMENT_LENGTH * 3)] mapM_ (writeChan i) inp outp <- getChanContents o - if and (zipWith (==) inp outp) - then return () - else error $ "Smoke test failed with starting offset of: "++(show n) + unless (and (zipWith (==) inp outp)) $ + error $ "Smoke test failed with starting offset of: "++(show n) -- w/r/w/r... spanning overflow smoke2 :: Int -> IO () @@ -62,9 +107,73 @@ where check i o x = do writeChan i x x' <- readChan o - if x == x' - then return () - else error $ "Smoke test failed with starting offset of: "++(show n)++"at write: "++(show x) + unless (x == x') $ + error $ "Smoke test failed with starting offset of: "++(show n)++"at write: "++(show x) + +-- for smoke checking size, and alignment of different Prim types, and allowing +-- testing of writing atomicUnicorn +smokeManyElement :: (Num e, Typeable e, Show e, UnagiPrim e)=> e -> IO () +smokeManyElement e = do + (i,o) <- newChan + let n = UI.sEGMENT_LENGTH*2 + 1 + replicateM_ n (writeChan i e) + outp <- getChanContents o + unless (all (== e) $ take n outp) $ + error $ "smokeManyElement failed with type "++(show $ typeOf e)++": "++(show e)++" /= "++(show outp) + +-- smokeManyElement for atomicUnicorn values +smokeManyUnicorn :: forall e. (Num e, Typeable e, Show e, UnagiPrim e)=> e -> IO () +smokeManyUnicorn _ = + maybe (return ()) smokeManyElement (atomicUnicorn :: Maybe e) + +-- check our segSource is doing what we expect with magic values: +segSourceMagicSanity :: forall e. (Num e, Typeable e, Show e, UnagiPrim e)=> e -> IO () +segSourceMagicSanity _ = + case atomicUnicorn :: Maybe e of + Nothing -> return () + Just e -> do + (_,eArr) <- UI.segSource :: IO (UI.SignalIntArray, UI.ElementArray e) + forM_ [0.. UI.sEGMENT_LENGTH-1] $ \i-> do + e' <- UI.readElementArray eArr i + unless (e == e') $ + error $ "in segSource, with type "++(show $ typeOf e)++", "++(show e)++" /= "++(show e') + +-- ------------- + +-- Make sure we get no tearing of adjacent word-size or smaller (as determined +-- by atomicUnicorn instantiation) values, making sure we cross a cache-line. +atomicUnicornAtomicicity :: forall e. (Num e, Typeable e, Show e, UnagiPrim e)=> e -> IO () +atomicUnicornAtomicicity _e = + when (isJust (atomicUnicorn :: Maybe e)) $ do + (_,eArr) <- UI.segSource :: IO (UI.SignalIntArray, UI.ElementArray e) + let iters = (64 `quot` P.sizeOf _e) + 1 + when (iters >= UI.sEGMENT_LENGTH) $ + error "Our sEGMENT_LENGTH is smaller than expected; please fix test" + -- just skip Addr for now TODO: + unless ( isJust (cast _e :: Maybe P.Addr) + || isJust (cast _e :: Maybe Char)) $ + forM_ [0.. iters] $ \i0 -> do + let i1 = i0+1 + rd = UI.readElementArray eArr + first0 <- rd i0 + first1 <- rd i1 + v0 <- newEmptyMVar + v1 <- newEmptyMVar + let incr f v i = go + where go _ 0 = putMVar v () + go expected n = do + val <- rd i + unless (val == expected) $ + error $ "atomicUnicornAtomicicity with type "++(show $ typeOf val)++" "++(show val)++" /= "++(show expected) + let !next = f val + UI.writeElementArray eArr i next + go next (n-1) + let counts = 1000000 :: Int + _ <- forkIO $ incr (+1) v0 i0 first0 counts + _ <- forkIO $ incr (subtract 1) v1 i1 first1 counts + -- BlockedIndefinitelyOnMVar means a problem TODO make better + takeMVar v0 >> takeMVar v1 + correctFirstWrite :: Int -> IO () correctFirstWrite n = do diff --git a/unagi-chan.cabal b/unagi-chan.cabal --- a/unagi-chan.cabal +++ b/unagi-chan.cabal @@ -1,7 +1,7 @@ name: unagi-chan -version: 0.2.0.1 +version: 0.3.0.0 -synopsis: Fast and scalable concurrent queues for x86, with a Chan-like API +synopsis: Fast concurrent queues with a Chan-like API, and more description: This library provides implementations of concurrent FIFO queues (for both @@ -10,10 +10,34 @@ limited usefulness outside of x86 architectures where the fetch-and-add instruction is not available. . + We export several variations of our design; some support additional + functionality while others try for lower latency by removing features or + making them more restrictive (e.g. in the @Unboxed@ variants). + . + - @Unagi@: a general-purpose near drop-in replacement for @Chan@. + . + - @Unagi.Unboxed@: like @Unagi@ but specialized for primitive types; this + may perform better if a queue grows very large. + . + - @Unagi.Bounded@: a bounded variant with blocking and non-blocking writes, + and other functionality where a notion of the queue's capacity is + required. + . + - @Unagi.NoBlocking@: lowest latency implementations for when blocking + reads aren't required. + . + - @Unagi.NoBlocking.Unboxed@: like @Unagi.NoBlocking@ but for primitive + types. + . + Some of these may be deprecated in the future if they are found to provide + little performance benefit, or no unique features; you should benchmark and + experiment with them for your use cases, and please submit pull requests + for additions to the benchmark suite that reflect what you find. + . 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, with an inset graph showing a zoomed-in view on the + standard libraries. The inset graph shows a zoomed-in view on the implementations here. . <> @@ -41,47 +65,57 @@ exposed-modules: Control.Concurrent.Chan.Unagi , Control.Concurrent.Chan.Unagi.Unboxed , Control.Concurrent.Chan.Unagi.Bounded + , Control.Concurrent.Chan.Unagi.NoBlocking + , Control.Concurrent.Chan.Unagi.NoBlocking.Unboxed other-modules: Control.Concurrent.Chan.Unagi.Internal , Control.Concurrent.Chan.Unagi.Unboxed.Internal , Control.Concurrent.Chan.Unagi.Bounded.Internal + , Control.Concurrent.Chan.Unagi.NoBlocking.Internal + , Control.Concurrent.Chan.Unagi.NoBlocking.Types + , Control.Concurrent.Chan.Unagi.NoBlocking.Unboxed.Internal , Control.Concurrent.Chan.Unagi.Constants , Utilities , Data.Atomics.Counter.Fat ghc-options: -Wall -funbox-strict-fields build-depends: base < 5 - -- be conservative about atomic-primops, for now; really - -- we're fine with any version that passes our tests: - , atomic-primops >= 0.6.0.5 && <= 0.6.0.6 + -- Hopefully if atomic-primops breaks in any subtle ways + -- our tests will be sufficient to notice: + , atomic-primops >= 0.6.0.5 , primitive>=0.5.3 + , ghc-prim default-language: Haskell2010 - -- We'll need some additional barriers for correctness: if !arch(i386) && !arch(x86_64) cpp-options: -DNOT_x86 + -- TODO: more complete list of 64-bit archs: + if arch(x86_64) + cpp-options: -DIS_64_BIT -- tryReadMVar is only available and non-broken on ghc >= 7.8.3 if impl(ghc >= 7.8.3) cpp-options: -DTRYREADMVAR -- TODO +-- For v0,4: +-- - More benchmarks, and test code we can analyze with ghc-events-analyze. +-- - Explore faster single-threaded write (see #11) +-- - Explore Stream interface for variants other than NoBlocking (see #11) +-- - Experiments w/ new GHC 7.10 stuff, and at least make sure buildable +-- ------- +-- - For GHC 7.10+ +-- - look at small arrays (w/out card-marking) +-- - re-benchmark array creation and adjust next segment wait -- - 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 -- --- Potential implementations roadmap (or we might just stick with this design --- for this package): +-- Possibly-similar prior work to look at: -- --- - fixed size MutableArray of purely-functional dequeues ("Tako") (fetch-and-add, then CAS) --- - variant replacing CAS with blocking turn-taking, also play with leap-frogging cache-lines --- - variant in STM (how to safely do the initial incrCounter at most once though?) --- would also let us separate read and write buckets. --- - bounded Tako variant -- - maybe implement "Fast Concurrent Queues for x86 Processors" by Morrison & Afek (non-blocking, probably more clever) -- - Also looks like a similar (but lockfree, as above) counter-based queue has been developed by FB: -- https://github.com/facebook/folly/blob/master/folly/MPMCQueue.h --- - boxed Unagi variant avoiding CAS with read simply spinning a few times and then calling yield, or something else -- Please just build tests and run: @@ -92,6 +126,9 @@ ghc-options: -Wall -funbox-strict-fields ghc-options: -O2 -rtsopts -threaded -with-rtsopts=-N ghc-options: -fno-ignore-asserts + -- for some hacks for Addr: + ghc-options: -fno-warn-orphans + ghc-options: -fno-warn-missing-methods -- I guess we need to put 'src' here to get access to Internal modules hs-source-dirs: tests, src main-is: Main.hs @@ -106,13 +143,16 @@ , UnagiUnboxed build-depends: base , primitive>=0.5.3 - , atomic-primops >= 0.6.0.5 && <= 0.6.0.6 + , atomic-primops >= 0.6.0.5 , containers + , ghc-prim default-language: Haskell2010 -- These have to be copied from 'library' section too! if !arch(i386) && !arch(x86_64) cpp-options: -DNOT_x86 + if arch(x86_64) + cpp-options: -DIS_64_BIT if impl(ghc >= 7.8.3) cpp-options: -DTRYREADMVAR @@ -183,7 +223,7 @@ --ghc-options: -threaded -with-rtsopts=-N2 --ghc-options: -eventlog -- ...or do non-threaded runtime - --ghc-prof-options: -fprof-auto + ghc-prof-options: -fprof-auto --Relevant profiling RTS settings: -xt -- TODO also check out +RTS -A10m, and look at output of -sstderr