unagi-chan 0.2.0.1 → 0.3.0.0

raw patch · 26 files changed

+1891/−273 lines, 26 filesdep +ghc-primdep ~atomic-primopsdep ~primitivePVP ok

version bump matches the API change (PVP)

Dependencies added: ghc-prim

Dependency ranges changed: atomic-primops, primitive

API changes (from Hackage documentation)

Files

• CHANGELOG.markdown +10/−0
• benchmarks/multi.hs +157/−0
• benchmarks/single.hs +117/−0
• core-example/Main.hs +65/−5
• src/Control/Concurrent/Chan/Unagi.hs +7/−2
• src/Control/Concurrent/Chan/Unagi/Bounded.hs +6/−2
• src/Control/Concurrent/Chan/Unagi/Bounded/Internal.hs +64/−47
• src/Control/Concurrent/Chan/Unagi/Constants.hs +1/−1
• src/Control/Concurrent/Chan/Unagi/Internal.hs +98/−43
• src/Control/Concurrent/Chan/Unagi/NoBlocking.hs +45/−0
• src/Control/Concurrent/Chan/Unagi/NoBlocking/Internal.hs +254/−0
• src/Control/Concurrent/Chan/Unagi/NoBlocking/Types.hs +62/−0
• src/Control/Concurrent/Chan/Unagi/NoBlocking/Unboxed.hs +47/−0
• src/Control/Concurrent/Chan/Unagi/NoBlocking/Unboxed/Internal.hs +297/−0
• src/Control/Concurrent/Chan/Unagi/Unboxed.hs +7/−4
• src/Control/Concurrent/Chan/Unagi/Unboxed/Internal.hs +228/−102
• src/Data/Atomics/Counter/Fat.hs +9/−6
• src/Utilities.hs +22/−3
• tests/Atomics.hs +4/−4
• tests/Deadlocks.hs +42/−0
• tests/DupChan.hs +50/−3
• tests/Implementations.hs +36/−4
• tests/Main.hs +5/−1
• tests/Smoke.hs +84/−21
• tests/UnagiUnboxed.hs +117/−8
• unagi-chan.cabal +57/−17

@@ -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`

@@ -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 ()

@@ -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 --

@@ -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 ()

@@ -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 )  

@@ -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 )  

@@ -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-

@@ -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 

@@ -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)  -- ----------

@@ -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)

@@ -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

@@ -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"

@@ -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)

@@ -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

@@ -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)

@@ -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

@@ -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

@@ -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', () #)

@@ -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

@@ -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++"... "

@@ -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

@@ -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

@@ -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!"

@@ -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: 

@@ -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 _ = "<addr>"++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

@@ -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.     .     <<http://i.imgur.com/J5rLUFn.png>>@@ -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