unagi-chan 0.3.0.1 → 0.3.0.2
raw patch · 10 files changed
+1079/−20 lines, 10 filesdep ~atomic-primopsPVP ok
version bump matches the API change (PVP)
Dependency ranges changed: atomic-primops
API changes (from Hackage documentation)
Files
- CHANGELOG.markdown +6/−0
- benchmarks/multi.hs +1/−9
- src/Control/Concurrent/Chan/Unagi.hs +1/−1
- src/Control/Concurrent/Chan/Unagi/Bounded.hs +1/−1
- src/Control/Concurrent/Chan/Unagi/Unboxed.hs +1/−1
- src/Control/Concurrent/Chan/Unagi/Unboxed/Internal.hs +3/−3
- tests/UnagiBounded.hs +360/−0
- tests/UnagiNoBlocking.hs +347/−0
- tests/UnagiNoBlockingUnboxed.hs +350/−0
- unagi-chan.cabal +9/−5
CHANGELOG.markdown view
@@ -27,3 +27,9 @@ - fix upper bounds on atomic-primops again (made as revision to cabal metadata for 0.3.0.0 - fix some docs++### 0.3.0.2++- re-bump atomic-primops version; should now support 7.10+- fix missing other-modules for test suite+- fix getChanContents for GHC 7.10 (see GHC Trac #9965)
benchmarks/multi.hs view
@@ -36,16 +36,8 @@ putStrLn $ "Running with capabilities: "++(show procs) -- let writersContending = do- writersIndependent = do- defaultMain $- [ bgroup "Experiments" $- [ bench "Writers contending" $ nfIO $ writersContending- , bench "Writers independent" $ nfIO $ writersIndependent- ]- , bgroup ("Operations on "++(show n)++" messages") $+ [ bgroup ("Operations on "++(show n)++" messages") $ [ bgroup "unagi-chan Unagi" $ -- this gives us a measure of effects of contention between -- readers and writers when compared with single-threaded
src/Control/Concurrent/Chan/Unagi.hs view
@@ -42,7 +42,7 @@ -- like System.IO.hGetContents. getChanContents :: OutChan a -> IO [a] getChanContents ch = unsafeInterleaveIO (do- x <- readChan ch+ x <- unsafeInterleaveIO $ readChan ch xs <- getChanContents ch return (x:xs) )
src/Control/Concurrent/Chan/Unagi/Bounded.hs view
@@ -51,7 +51,7 @@ -- like System.IO.hGetContents. getChanContents :: OutChan a -> IO [a] getChanContents ch = unsafeInterleaveIO (do- x <- readChan ch+ x <- unsafeInterleaveIO $ readChan ch xs <- getChanContents ch return (x:xs) )
src/Control/Concurrent/Chan/Unagi/Unboxed.hs view
@@ -43,7 +43,7 @@ -- like System.IO.hGetContents. getChanContents :: UnagiPrim a=> OutChan a -> IO [a] getChanContents ch = unsafeInterleaveIO (do- x <- readChan ch+ x <- unsafeInterleaveIO $ readChan ch xs <- getChanContents ch return (x:xs) )
src/Control/Concurrent/Chan/Unagi/Unboxed/Internal.hs view
@@ -183,19 +183,19 @@ -- These ought all to be atomic for 32-bit or 64-bit systems:-instance UnagiPrim Char where+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+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+instance UnagiPrim Word where atomicUnicorn = Just 0xDADADA instance UnagiPrim Word8 where atomicUnicorn = Just 0xDA
+ tests/UnagiBounded.hs view
@@ -0,0 +1,360 @@+module UnagiBounded(unagiBoundedMain) where++import Control.Concurrent.Chan.Unagi.Bounded+import qualified Control.Concurrent.Chan.Unagi.Bounded.Internal as UI+import Control.Monad+import qualified Data.Primitive as P+import Data.IORef++import Control.Concurrent(forkIO,threadDelay)+import Control.Concurrent.MVar+import Control.Exception+import Data.Atomics.Counter.Fat+import Control.Applicative++import Data.Maybe(isNothing)++unagiBoundedMain :: IO ()+unagiBoundedMain = do+ putStrLn "==================="+ putStrLn "Testing Unagi.Bounded details:"+ -- ------+ putStr "Smoke test at different starting offsets, spanning overflow... "+ forM_ [1,2,4,1024, 3,5,513] $ \segLen-> do+ -- ------+ mapM_ (overflowSanity segLen) $ + [ (maxBound - segLen - 1) .. maxBound] + ++ [minBound .. (minBound + segLen + 1)]+ -- ------+ newChanSanity segLen+ putStrLn "OK"+ -- ------+ putStr "Correct first write and read... "+ mapM_ correctFirstWriteRead [ maxBound - 5, maxBound - 4, maxBound - 3, maxBound, minBound, 0]+ putStrLn "OK"+ -- ------+ let tries = 10000+ putStrLn $ "Checking for deadlocks from killed Unagi reader in a fancy way, x"++show tries+ checkDeadlocksReaderUnagiBounded tries+ -- ------+ putStr "Test bounds blocking... "+ forM_ [(1::Int),2,4,8] $ \n->+ testBoundsBlocking (2^n)+ putStrLn "OK"+ -- ------+ putStr "Test behavior of tryWriteChan... "+ forM_ [(1::Int),2,4,8] $ \n-> do+ tryWriteChanSmoke (2^n)+ tryWriteChanConcurrent (2^n)+ putStrLn "OK"+ -- ------+ putStrLn "Testing Unagi.Bounded components:"+ -- ------+ putStr " Checkpoint tests... "+ checkpointTest1 100000000+ checkpointTest2 100000000+ putStrLn "OK"+++-- TODO NOTE WHEN WE FACTOR THIS OUT OF UNAGI*-SPECIFIC TESTS, make function closed over sEGMENT_LENGTH as we do here.+-- NOTE: FORMERLY smoke2+--+-- w/r/w/r... spanning overflow+overflowSanity :: Int -> Int -> IO ()+overflowSanity segLen n = do+ (i,o) <- UI.newChanStarting n segLen+ let inp = [0 .. (segLen * 3)]+ mapM_ (check i o) inp+ where check i o x = do+ writeChan i x+ x' <- readChan o+ unless (x == x') $+ error $ "Smoke test failed with starting offset of: "++(show n)++"at write: "++(show x)++-- exercise power-of-two rounding, and make sure array sizes and recorded+-- bounds match up+newChanSanity :: Int -> IO ()+newChanSanity bnds = do+ (UI.InChan _ _ inCE, UI.OutChan outCE) <- newChan bnds+ check inCE+ check outCE+ checkSame inCE outCE+ where checkSame (UI.ChanEnd x y _ _ _) (UI.ChanEnd x' y' _ _ _) =+ unless (x == x' && y == y') $+ error $ "newChanSanity: "++(show (x,x',y,y'))++ check (UI.ChanEnd logBounds boundsMn1 _segSource _ _) = do+ let storedBoundsSane = (2^logBounds) == (boundsMn1 + 1)+ lengthEqualsBounds <- return True -- TODO get and check length from segSource arr, possible?+ unless (storedBoundsSane && lengthEqualsBounds) $+ error $ "newChanSanity: PLORT!"+++-- Basic first write sanity checking and writer unblocking mechanics+correctFirstWriteRead :: Int -> IO ()+correctFirstWriteRead n = do+ let size = 4+ (i, o@(UI.OutChan (UI.ChanEnd _logBounds _boundsMn1 _segSource _cntr strHeadRef))) <- UI.newChanStarting n size++ writeChan i ()++ (UI.StreamHead offset0 (UI.Stream arr next)) <- readIORef strHeadRef+ cell <- P.readArray arr 0+ case cell of+ UI.Written () -> return ()+ _ -> error "Expected a Write at index 0"+ unless (n == offset0)$+ error $ "offset0 /= "++(show n)++", instead == "++(show offset0)++ -- The next segment should not be set up yet:+ noSegment <- readIORef next+ unless (isNothing noSegment) $+ error "Next segment should not be set up yet!"++ -- First read should set up next segment for unblocked writers.+ () <- readChan o `onException` putStrLn "Read of first and only value failed!"+ nextSeg <- readIORef next+ case nextSeg of+ Nothing -> error "Next should have been set up after first read!"+ Just (UI.NextByReader (UI.Stream _arr' next')) -> do+ noSegment' <- readIORef next'+ unless (isNothing noSegment') $+ error "Next of next segment should not be set up yet!"+ _ -> error "Next marked installed by writer!"++++-- test for deadlocks caused by async exceptions in reader.+checkDeadlocksReaderUnagiBounded :: Int -> IO ()+checkDeadlocksReaderUnagiBounded times = do+ let run 0 normalRetries numRace = putStrLn $ "Lates: "++(show normalRetries)++", Races: "++(show numRace)+ run n normalRetries numRace+ | (normalRetries + numRace) > (times `div` 3) = error "This test is taking too long. Please retry, and if still failing send the log to me"+ | otherwise = do+ -- we'll kill the reader with our special exception half the time,+ -- expecting that we never get our race condition on those runs:+ let usingReadChanOnException = even n++ let numPreloaded = 10000+ (i,o) <- UI.newChanStarting 0 $ numPreloaded+2+ -- preload a chan with 0s+ replicateM_ numPreloaded $ writeChan i (0::Int)++ rStart <- newEmptyMVar+ saved <- newEmptyMVar -- for holding potential saved (IO a) from blocked readChanOnException+ rid <- forkIO $ (\rd-> putMVar rStart () >> (forever $ void $ rd o)) $+ if usingReadChanOnException + then flip readChanOnException ( putMVar saved )+ else readChan+ takeMVar rStart >> threadDelay 1+ throwTo rid ThreadKilled++ -- did killing reader damage queue for reads or writes?+ writeChan i 1 `onException` ( putStrLn "Exception from first writeChan!")+ writeChan i 2 `onException` ( putStrLn "Exception from second writeChan!")+ finalRead <- readChan o `onException` ( putStrLn "Exception from final readChan!")+ + oCnt <- readCounter $ (\(UI.OutChan(UI.ChanEnd _ _ _ cntr _))-> cntr) o+ iCnt <- readCounter $ (\(UI.InChan _ _ (UI.ChanEnd _ _ _ cntr _))-> cntr) i+ unless (iCnt == numPreloaded + 1) $ + error "The InChan counter doesn't match what we'd expect from numPreloaded!"++ case finalRead of+ 0 -> if oCnt <= 0 -- (technically, -1 means hasn't started yet)+ -- reader didn't have a chance to get started+ then putStr "0" >> run n (normalRetries+1) numRace+ -- normal run; we tested that killing a reader didn't+ -- break chan for other readers/writers:+ else putStr "." >> run (n-1) normalRetries numRace+ --+ -- Rare. Reader was killed after reading all pre-loaded messages+ -- but before starting what would be the blocking read:+ 1 | oCnt == numPreloaded -> putStr "X" >> run n normalRetries (numRace + 1)+ | otherwise -> error $ "Having read final 1, "+++ "Expecting a counter value of "++(show numPreloaded)+++ " but got: "++(show oCnt)+ 2 -> do when usingReadChanOnException $ do+ shouldBe1 <- join $ takeMVar saved+ unless (shouldBe1 == 1) $+ error "The handler for our readChanOnException should only have returned 1"+ unless (oCnt == numPreloaded + 1) $+ error $ "Having read final 2, "+++ "Expecting a counter value of "++(show $ numPreloaded+1)+++ " but got: "++(show oCnt)+ putStr "+" >> run n (normalRetries + 1) numRace++ _ -> error "Fix your #$%@ test!"++ run times 0 0+ putStrLn ""++-- idempotent & non-conflicting puts/takes+checkpointTest1 :: Int -> IO ()+checkpointTest1 n = do+ x <- newEmptyMVar+ checkpt <- UI.WriterCheckpoint <$> newEmptyMVar+ -- check writerCheckin and tryWriterCheckin concurrently against idempotent unblockWriters+ void $ forkIO ((replicateM_ n $ (UI.writerCheckin checkpt >> UI.tryWriterCheckin checkpt)) >> putMVar x ())+ threadDelay 10000+ replicateM_ n $ UI.unblockWriters checkpt+ takeMVar x `onException` putStrLn "checkpointTest1: Forked writerCheckin deadlocked!"++-- No deadlocks in read:+checkpointTest2 :: Int -> IO ()+checkpointTest2 n = do+ x <- newEmptyMVar+ checkpt <- UI.WriterCheckpoint <$> newEmptyMVar+ -- check writerCheckin and tryWriterCheckin concurrently against writerCheckin+ void $ forkIO ((replicateM_ n $ (UI.writerCheckin checkpt >> UI.tryWriterCheckin checkpt)) >> putMVar x ())+ threadDelay 10000+ UI.unblockWriters checkpt+ replicateM_ n $ UI.writerCheckin checkpt+ takeMVar x `onException` putStrLn "checkpointTest2: Forked writerCheckin deadlocked!"+++-- Test that our bounding works as expected+testBoundsBlocking :: Int -> IO ()+testBoundsBlocking bnds = do+ v <- newEmptyMVar+ (inC,outC) <- newChan bnds+ -- make sure none of these block.+ replicateM_ bnds $ writeChan inC ()+ -- but this blocks+ void $ forkIO $ writeChan inC () >> putMVar v ()+ threadDelay 100000+ noth <- tryTakeMVar v+ unless (noth == Nothing) $+ error "bnds+1 writeChan should have blocked!"+ -- ...until:+ readChan outC+ takeMVar v `onException` putStrLn "Read should have unblocked bnds+1 writer"+ -- Now we fork one which work together to fill remaining bnds-1 slots, without blocking:+ (replicateM_ (bnds-1) $ writeChan inC ())+ `onException` putStrLn "Writes should not have blocked in second segment"+ -- now we're at 2x bounds.+ -- This next, again, should block:+ v2 <- newEmptyMVar + void $ forkIO $ writeChan inC () >> putMVar v2 ()+ threadDelay 100000+ noth2 <- tryTakeMVar v2+ unless (noth2 == Nothing) $+ error "bnds*2+1 writeChan should have blocked!"+ -- and unblock as soon as (but no sooner than after bnds reads):+ replicateM_ (bnds-1) $ readChan outC+ threadDelay 100000+ noth3 <- tryTakeMVar v2+ unless (noth3 == Nothing) $+ error "bnds*2+1 writeChan should still have been blocked!"+ -- now this should unblock:+ readChan outC+ takeMVar v2 `onException` putStrLn "Read should have unblocked bnds*2+1 writer"+++{- OLD AND NO-LONGER RELEVANT+-- A fork of the above, for tryWriteChan+-- TODO these coule be more thoughtful/thorough+testTryWriteChan :: Int -> IO ()+testTryWriteChan bnds = do+ (inC,outC) <- newChan bnds+ -- make sure none of these block.+ trues <- replicateM bnds $ tryWriteChan inC ()+ unless (and trues) $+ error "Some of the first writes failed!"+ -- but this ought to fail:+ success1 <- tryWriteChan inC () `onException` putStrLn "Our tryTakeMVar seems to have blocked instead of returning False!"+ when success1 $+ error "bnds+1 tryWriteChan should have failed!"+ -- ...until we read:+ readChan outC+ -- then this should succeed:+ success2 <- tryWriteChan inC () `onException` putStrLn "Our tryTakeMVar number 2 seems to have blocked instead of returning true!"+ unless success2 $+ error "bnds+1 tryWriteChan should have succeeded this time!"+ --+ -- Now we fork one which work together to fill remaining bnds-1 slots, without blocking:+ successes2 <- replicateM (bnds-1) $ tryWriteChan inC ()+ `onException` putStrLn "tryWrites should not have blocked in second segment"+ unless (and successes2) $+ error "failures seen in succcesses2!"+ -- now we're at 2x bounds.+ -- This next, again, should block:+ success3 <- tryWriteChan inC () `onException` putStrLn "success3: Our tryTakeMVar seems to have blocked instead of returning False!"+ when success3 $+ error "bnds*2+1 tryWriteChan should have failed!"+ -- and unblock as soon as (but no sooner than after bnds reads):+ replicateM_ (bnds-1) $ readChan outC+ success4 <- tryWriteChan inC () `onException` putStrLn "success4: Our tryTakeMVar seems to have blocked instead of returning False!"+ when success4 $+ error "bnds*2+1 writeChan should still have failed!"+ -- now this should unblock:+ readChan outC+ success5 <- tryWriteChan inC () `onException` putStrLn "success5: Our tryTakeMVar seems to have blocked instead of returning False!"+ unless success5 $+ error "success5: should have succeeded!"++ -- Now fork a couple writers and make sure none block:+ vs <- replicateM 2 newEmptyMVar + forM_ vs $ \v-> forkIO $ do+ replicateM_ (5*bnds) $+ tryWriteChan inC ()+ putMVar v ()+ forM_ vs $ \v-> + takeMVar v `onException` putStrLn "A tryWriteChan seems to have blocked!"+-}+tryWriteChanSmoke :: Int -> IO ()+tryWriteChanSmoke bnds = do+ (inC,outC) <- newChan bnds+ -- make sure none of these block.+ trues <- replicateM bnds $ tryWriteChan inC ()+ unless (and trues) $+ error "Some of the first writes failed!"+ -- but this ought to fail:+ success1 <- tryWriteChan inC () `onException` putStrLn "Our tryTakeMVar seems to have blocked instead of returning False!"+ when success1 $+ error "bnds+1 tryWriteChan should have failed!"+ -- ...until we read:+ readChan outC+ -- then this should succeed:+ success2 <- tryWriteChan inC () `onException` putStrLn "Our tryTakeMVar number 2 seems to have blocked instead of returning true!"+ unless success2 $+ error "bnds+1 tryWriteChan should have succeeded this time!"+ -- And the next should fail again.+ success3 <- tryWriteChan inC () `onException` putStrLn "Our tryTakeMVar number 3 seems to have blocked instead of returning true!"+ when success3 $+ error "bnds+2 tryWriteChan should have failed!"+ readChan outC+ success4 <- tryWriteChan inC () `onException` putStrLn "Our tryTakeMVar number 4 seems to have blocked instead of returning true!"+ unless success4 $+ error "bnds+3 tryWriteChan should have succeeded this time!"++-- Now do some concurrency tests, forking readers and writers that retry until+-- they can fill their quota+tryWriteChanConcurrent :: Int -> IO ()+tryWriteChanConcurrent bnds = do+ (inC,outC) <- newChan bnds+ let n = 1000000+ writer :: Int -> Int -> MVar Int -> IO ()+ writer cnt failed v+ | cnt > 0 = do+ success <- tryWriteChan inC ()+ if success + then writer (cnt-1) failed v+ else writer cnt (failed+1) v -- or yield here+ | otherwise = putMVar v failed+ v1 <- newEmptyMVar+ -- Test with truly concurrent reader and writer:+ void $ forkIO $ writer n 0 v1+ replicateM_ n $ readChan outC+ _failed0 <- takeMVar v1+ -- print _failed0++ -- And now with hopefully some concurrent writers:+ v2 <- newEmptyMVar+ void $ forkIO $ writer n 0 v1+ void $ forkIO $ writer n 0 v2+ void $ forkIO $ replicateM_ (n*2) $ readChan outC+ _failed1 <- takeMVar v1+ _failed2 <- takeMVar v2+ -- print _failed1+ -- print _failed2+ return ()
+ tests/UnagiNoBlocking.hs view
@@ -0,0 +1,347 @@+module UnagiNoBlocking (unagiNoBlockingMain) where++-- Unagi-chan-specific tests++import Control.Concurrent.Chan.Unagi.NoBlocking+import qualified Control.Concurrent.Chan.Unagi.NoBlocking.Internal as UI+import Control.Monad+import qualified Data.Primitive as P+import Data.IORef+import System.Mem(performGC)+import Data.List(sort)++import Control.Concurrent(forkIO,yield,threadDelay)+import Control.Concurrent.MVar+import Control.Exception+import Data.Atomics.Counter.Fat++unagiNoBlockingMain :: IO ()+unagiNoBlockingMain = do+ putStrLn "==================="+ putStrLn "Testing Unagi.NoBlocking 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 "Correct first write... "+ mapM_ correctFirstWrite [ (maxBound - 7), maxBound, minBound, 0]+ putStrLn "OK"+ -- ------+ putStr "Correct initial writes... "+ mapM_ correctInitialWrites [ (maxBound - UI.sEGMENT_LENGTH), (maxBound - UI.sEGMENT_LENGTH) - 1, maxBound, minBound, 0]+ putStrLn "OK"+ -- ------+ putStr "Checking isActive... "+ replicateM_ 10 isActiveTest+ replicateM_ 10 readChanYieldTest+ putStrLn "OK"+ -- ------+ putStr "Checking streamChan... "+ streamChanSmoke+ replicateM_ 3 $ streamChanConcurrentStreamerReader 10000000+ replicateM_ 3 $ streamChanConcurrentStreamerWriter 10000000+ putStrLn "OK"+ -- ------+ let tries = 10000+ putStrLn $ "Checking for deadlocks from killed Unagi reader in a fancy way, x"++show tries+ checkDeadlocksReaderUnagi tries+++-- Helper for when we know a read should succeed immediately:+tryReadChanErr :: OutChan a -> IO a+tryReadChanErr oc = tryReadChan oc + >>= tryRead + >>= maybe (error "A read we expected to succeed failed!") return++-- TODO CONSIDER ADDING newChanStarting (or raplcing newChan) TO IMPLEMENTATIONS, AND CONSOLIDATE THESE IN Smoke.hs+smoke :: Int -> IO ()+smoke n = smoke1 n >> smoke2 n++-- www.../rrr... spanning overflow+smoke1 :: Int -> IO ()+smoke1 n = do+ (i,o) <- UI.newChanStarting n+ let inp = [0 .. (UI.sEGMENT_LENGTH * 3)]+ mapM_ (writeChan i) inp+ forM_ inp $ \inx-> do+ outx <- tryReadChanErr o+ unless (inx == outx) $+ error $ "Smoke test failed with starting offset of: "++(show n)++-- w/r/w/r... spanning overflow+smoke2 :: Int -> IO ()+smoke2 n = do+ (i,o) <- UI.newChanStarting n+ let inp = [0 .. (UI.sEGMENT_LENGTH * 3)]+ mapM_ (check i o) inp+ where check i o x = do+ writeChan i x+ x' <- tryReadChanErr o+ if x == x'+ then return ()+ else error $ "Smoke test failed with starting offset of: "++(show n)++"at write: "++(show x)++correctFirstWrite :: Int -> IO ()+correctFirstWrite n = do+ (i,oc@(UI.OutChan _ (UI.ChanEnd _ _ arrRef))) <- UI.newChanStarting n+ actv <- isActive oc+ unless actv $+ error "not isActive before first and only write"+ writeChan i ()+ (UI.StreamHead _ (UI.Stream arr _)) <- readIORef arrRef+ cell <- P.readArray arr 0+ case cell of+ Just () -> return ()+ _ -> error "Expected a Write at index 0"++-- check writes by doing a segment+1-worth of reads by hand+-- also check that segments pre-allocated as expected+correctInitialWrites :: Int -> IO ()+correctInitialWrites startN = do+ (i,(UI.OutChan _ (UI.ChanEnd _ _ arrRef))) <- UI.newChanStarting startN+ let writes = [0..UI.sEGMENT_LENGTH]+ mapM_ (writeChan i) writes+ (UI.StreamHead _ (UI.Stream arr next)) <- readIORef arrRef+ -- check all of first segment:+ forM_ (init writes) $ \ix-> do+ cell <- P.readArray arr ix+ case cell of+ Just n+ | n == ix -> return ()+ | otherwise -> error $ "Expected a Write at index "++(show ix)++" of same value but got "++(show n)+ _ -> error $ "Expected a Write at index "++(show ix)+ -- check last write:+ lastSeg <- readIORef next+ case lastSeg of+ (UI.Next (UI.Stream arr2 next2)) -> do+ cell <- P.readArray arr2 0+ case cell of+ Just n + | n == last writes -> return ()+ | otherwise -> error $ "Expected last write at index "++(show $ last writes)++" of same value but got "++(show n)+ _ -> error "Expected a last Write"+ -- check pre-allocation:+ n2 <- readIORef next2+ case n2 of + (UI.Next (UI.Stream _ next3)) -> do+ n3 <- readIORef next3+ case n3 of+ UI.NoSegment -> return ()+ _ -> error "Too many segments pre-allocated!"+ _ -> error "Next segment was not pre-allocated!"+ _ -> error "No last segment present!"++++{- NOTE: we would like this to pass (see trac #9030) but are happy to note that+ - it fails which somewhat validates the test below++testBlockedRecovery = do+ (i,o) <- newChan+ v <- newEmptyMVar+ rid <- forkIO (putMVar v () >> readChan o)+ takeMVar v+ threadDelay 1000+ throwTo rid ThreadKilled+ -- we race the exception-handler in `readChan` here...+ writeChan i ()+ -- In a buggy implementation, this would consistently win failing by losing+ -- the message and raising BlockedIndefinitely here:+ readChan o+-}+++-- test for deadlocks caused by async exceptions in reader.+checkDeadlocksReaderUnagi :: Int -> IO ()+checkDeadlocksReaderUnagi times = do+ let run 0 normalRetries numRace = putStrLn $ "Lates: "++(show normalRetries)++", Races: "++(show numRace)+ run n normalRetries numRace+ | (normalRetries + numRace) > (times `div` 3) = error "This test is taking too long. Please retry, and if still failing send the log to me"+ | otherwise = do+ (i,o) <- UI.newChanStarting 0+ -- preload a chan with 0s+ let numPreloaded = 10000+ replicateM_ numPreloaded $ writeChan i (0::Int)++ rStart <- newEmptyMVar+ rid <- forkIO $ (putMVar rStart () >> (forever $ void $ readChan yield o))+ takeMVar rStart >> threadDelay 1+ throwTo rid ThreadKilled++ -- did killing reader damage queue for reads or writes?+ writeChan i 1 `onException` ( putStrLn "Exception from first writeChan!")+ writeChan i 2 `onException` ( putStrLn "Exception from second writeChan!")+ finalRead <- tryReadChanErr o `onException` ( putStrLn "Exception from final tryReadChan!")+ + oCnt <- readCounter $ (\(UI.OutChan _ (UI.ChanEnd _ cntr _))-> cntr) o+ iCnt <- readCounter $ (\(UI.InChan _ (UI.ChanEnd _ cntr _))-> cntr) i+ unless (iCnt == numPreloaded + 1) $ + error "The InChan counter doesn't match what we'd expect from numPreloaded!"++ case finalRead of+ 0 -> if oCnt <= 0 -- (technically, -1 means hasn't started yet)+ -- reader didn't have a chance to get started+ then putStr "0" >> run n (normalRetries+1) numRace+ -- normal run; we tested that killing a reader didn't+ -- break chan for other readers/writers:+ else putStr "." >> run (n-1) normalRetries numRace+ --+ -- Rare. Reader was killed after reading all pre-loaded messages+ -- but before starting what would be the blocking read:+ 1 | oCnt == numPreloaded -> putStr "X" >> run n normalRetries (numRace + 1)+ | otherwise -> error $ "Having read final 1, "+++ "Expecting a counter value of "++(show numPreloaded)+++ " but got: "++(show oCnt)+ 2 -> do unless (oCnt == numPreloaded + 1) $+ error $ "Having read final 2, "+++ "Expecting a counter value of "++(show $ numPreloaded+1)+++ " but got: "++(show oCnt)+ putStr "+" >> run n (normalRetries + 1) numRace++ _ -> error "Fix your #$%@ test!"++ run times 0 0+ putStrLn ""++-- do a series of writes, forcing GC making sure remains true, then do the last+-- write and loop on forcing GC until we see False.+isActiveTest :: IO ()+isActiveTest = do+ let n = 100000+ k = n `div` 100+ let testWrites (inc,outc) = replicateM_ 100 $ do+ replicateM_ k $ writeChan inc ()+ performGC+ actv <- isActive outc+ unless actv $+ error "isActive returned False before last write!"++ lastWriteWait iters (inc,outc) = writeChan inc () >> go iters where+ go i | i < (0::Int) = error "Timed out waiting for isActive to return False. Anomaly or possible bug."+ | otherwise = do+ actv <- isActive outc+ when actv $ performGC >> go (i-1)+ -- first with newChan:+ c1 <- newChan+ testWrites c1+ lastWriteWait 1000 c1+ -- then with a duplicated channel:+ (inc2,_) <- newChan+ outc2 <- dupChan inc2+ testWrites (inc2,outc2)+ lastWriteWait 1000 (inc2,outc2)++-- Concurrently write [1..100000], while reading 100001 and prepending in a+-- IORef. Then a handler catches and puts () in an MVar which we wait on.+-- Then check that the elements were correct.+readChanYieldTest :: IO ()+readChanYieldTest = do+ let n = 100000 :: Int+ (inc,outc) <- newChan+ saving <- newIORef []+ exceptionRaised <- newIORef False+ goAhead <- newEmptyMVar++ let handling io = Control.Exception.catch io $ \BlockedIndefinitelyOnMVar ->+ writeIORef exceptionRaised True++ void $ forkIO $ replicateM_ (n+1) $ handling $ do+ x <- readChan yield outc+ modifyIORef' saving (x:)+ when (x == n) $ -- about to do final deadlocking loop:+ putMVar goAhead ()+ void $ forkIO $ forM_ [1..n] $ writeChan inc++ takeMVar goAhead+ out <- readIORef saving+ unless (out == [n,n-1..1]) $+ error "readChanYieldTest reads incorrect!"++ performGC+ threadDelay 100000+ raised <- readIORef exceptionRaised+ unless raised $+ error "Handler doesn't seem to have run in readChanYieldTest. Either a testing fluke or a bug."+++-- Smoke tests for different strides of interleaved streams, at different+-- offsets, with concurrent stream readers.+streamChanSmoke :: IO ()+streamChanSmoke =+ -- A few odd starting offsets, spanning Int/Counter overflow:+ forM_ [0, maxBound - UI.sEGMENT_LENGTH, maxBound - UI.sEGMENT_LENGTH + 1, maxBound, minBound] $ \startingOffset ->+ -- And few odd numbers of streams, where we especially want to exercise+ -- skips of entire segments: + forM_ [1,17, UI.sEGMENT_LENGTH, UI.sEGMENT_LENGTH+1, UI.sEGMENT_LENGTH*3+1] $ \numStreams-> do+ (i,o) <- UI.newChanStarting startingOffset+ let payload = 100000 :: Int+ -- and these are what we'll expect to see returned:+ let payloadPartsRev = map reverse $ [ [s,(s+numStreams).. payload ] | s<-[1..numStreams]]+ forM_ [1..payload] (writeChan i)+ strms <- streamChan numStreams o+ strmsReadOut <- replicateM numStreams newEmptyMVar+ unless (length strms == numStreams) $ + error $ "numStreams /= length strms: "+ ++(show numStreams)++" vs "++(show $ length strms)+ ++" at offset: "++(show startingOffset)+ forM_ (zip strms strmsReadOut) (forkIO . consumeUntilEmpty [])+ parts <- forM (zip strmsReadOut payloadPartsRev) $ \(v,expectedStack)-> do+ stack <- takeMVar v+ unless (stack == expectedStack) $ error $ "Incorrect stream reads: "++(show stack)+ return stack+ unless ((sort $ concat parts) == [1..payload]) $+ error $ "Somehow read parts weren't what we expected: "++(show parts)+ where consumeUntilEmpty stack (strm,v) = do+ h <- tryReadNext strm+ case h of+ (Next x xs) -> consumeUntilEmpty (x:stack) (xs,v)+ Pending -> putMVar v stack -- Done++-- Simple writer/streamer concurrency test+streamChanConcurrentStreamerWriter :: Int -> IO ()+streamChanConcurrentStreamerWriter n = do+ (i,o) <- newChan+ [strm] <- streamChan 1 o+ v <- newEmptyMVar+ let streamReader s stack itr failCnt + | failCnt > 4 = putMVar v $ Left "failCnt exceeded; possibly bug, but probably anomaly"+ | itr > n = putMVar v $ Right stack+ | otherwise = do+ xs <- tryReadNext s+ case xs of+ Pending -> threadDelay 1000 >> streamReader s stack itr (failCnt+1)+ Next x xs' -> streamReader xs' (x:stack) (itr+1) 0+ void $ forkIO $ streamReader strm [] (1::Int) (0::Int)+ void $ forkIO $ mapM_ (writeChan i) [1..n]+ strmOut <- either error return =<< takeMVar v+ unless (strmOut == [n,n-1..1]) $ + error $ "Stream reads were incorrect: "++(show strmOut)++-- Simple reader/streamer concurrency test+streamChanConcurrentStreamerReader :: Int -> IO ()+streamChanConcurrentStreamerReader n = do+ (i,o) <- newChan+ [strm] <- streamChan 1 o+ mapM_ (writeChan i) [1..n]+ vStream <- newEmptyMVar+ vOutchan <- newEmptyMVar+ let streamReader s stack = do+ xs <- tryReadNext s+ case xs of+ Pending -> putMVar vStream stack+ Next x xs' -> streamReader xs' (x:stack)++ outchanReader stack = do+ tryReadChan o >>= tryRead >>= maybe (putMVar vOutchan stack) (outchanReader . (:stack))+ void $ forkIO $ streamReader strm []+ void $ forkIO $ outchanReader []+ strmOut <- takeMVar vStream `onException` putStr " :in takeMVar vStream: "+ rdOut <- takeMVar vOutchan `onException` putStr " :in takeMVar vOutchan: "+ let correctOut = [n,n-1..1]+ unless (strmOut == correctOut) $ + error $ "Stream reads were incorrect: "++(show strmOut)+ unless (rdOut == correctOut) $ + error $ "OutChan reads were incorrect: "++(show rdOut)+
+ tests/UnagiNoBlockingUnboxed.hs view
@@ -0,0 +1,350 @@+module UnagiNoBlockingUnboxed (unagiNoBlockingUnboxedMain) where++import Control.Concurrent.Chan.Unagi.NoBlocking.Unboxed+import qualified Control.Concurrent.Chan.Unagi.NoBlocking.Unboxed.Internal as UI++import Control.Monad+import Data.IORef+import System.Mem(performGC)+import Data.List(sort)++import Control.Concurrent(forkIO,yield,threadDelay)+import Control.Concurrent.MVar+import Control.Exception+import Data.Atomics.Counter.Fat+import Data.Int(Int8)++-- copied from UnagiNoBlocking.hs at ddba5eb+--+-- Differences, in case we want TODO refactoring w/ a CPP macro to combine:+-- - type sigs: UnagiPrim constraints+-- - ChanEnd internals+-- - Omit correctInitialWrites/correctFirstWrite+-- - use of () as payload in isActiveTest (this can be changed to Int in+-- UnagiNoBlocking and elsewhere)+-- - added magicSmokeChan and magicSmokeStream++-- NOTE: in Unboxed we do much more thorough smoke tests for all existing+-- UnagiPrim. Since this implementation re-uses that machinery (e.g. there is+-- no 'sizeOf' or 'read/writeByteArray' of eArr in NoBlocking.Unboxed.Internal)+-- I don't think it's necessary to do that here.+--+-- I've also removed 'correctFirstWrite' and 'correctInitialWrites' which I+-- think are probably also unnecessary.+unagiNoBlockingUnboxedMain :: IO ()+unagiNoBlockingUnboxedMain = do+ putStrLn "==================="+ putStrLn "Testing Unagi.NoBlocking.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)]+ magicSmokeChan (atomicUnicorn::Maybe Int)+ magicSmokeChan (atomicUnicorn::Maybe Int8)+ magicSmokeStream (atomicUnicorn :: Maybe Int)+ magicSmokeStream (atomicUnicorn :: Maybe Int8)+ putStrLn "OK"+ -- ------+ putStr "Checking isActive... "+ replicateM_ 10 isActiveTest+ replicateM_ 10 readChanYieldTest+ putStrLn "OK"+ -- ------+ putStr "Checking streamChan... "+ streamChanSmoke+ replicateM_ 3 $ streamChanConcurrentStreamerReader 10000000+ replicateM_ 3 $ streamChanConcurrentStreamerWriter 10000000+ putStrLn "OK"+ -- ------+ let tries = 10000+ putStrLn $ "Checking for deadlocks from killed Unagi reader in a fancy way, x"++show tries+ checkDeadlocksReaderUnagi tries+++-- Helper for when we know a read should succeed immediately:+tryReadChanErr :: UnagiPrim a=> OutChan a -> IO a+tryReadChanErr oc = tryReadChan oc + >>= tryRead + >>= maybe (error "A read we expected to succeed failed!") return++smoke :: Int -> IO ()+smoke n = smoke1 n >> smoke2 n++-- www.../rrr... spanning overflow+smoke1 :: Int -> IO ()+smoke1 n = do+ (i,o) <- UI.newChanStarting n+ let inp = [0 .. (UI.sEGMENT_LENGTH * 3)]+ mapM_ (writeChan i) inp+ forM_ inp $ \inx-> do+ outx <- tryReadChanErr o+ unless (inx == outx) $+ error $ "Smoke test failed with starting offset of: "++(show n)++-- w/r/w/r... spanning overflow+smoke2 :: Int -> IO ()+smoke2 n = do+ (i,o) <- UI.newChanStarting n+ let inp = [0 .. (UI.sEGMENT_LENGTH * 3)]+ mapM_ (check i o) inp+ where check i o x = do+ writeChan i x+ x' <- tryReadChanErr o+ if x == x'+ then return ()+ else error $ "Smoke test failed with starting offset of: "++(show n)++"at write: "++(show x)++-- just validate that we can read and write the magic value correctly:+magicSmokeStream , magicSmokeChan :: UnagiPrim a=> Maybe a-> IO ()++magicSmokeChan Nothing = error "Pass us a type w/ a real atomicUnicorn plz"+magicSmokeChan (Just magic) = do+ unless (Just magic == atomicUnicorn) $ error "Pass us atomicUnicorn plz"+ (i,o) <- newChan+ el <- tryReadChan o + tryRead el+ >>= maybe (return ()) (const $ error "magicSmokeChan: should be empty!")+ writeChan i magic+ magic' <- tryRead el+ >>= maybe (error "magicSmokeChan: empty!") return+ unless (magic == magic') $ + error "magicSmokeChan: we didn't read back atomicUnicorn!"++magicSmokeStream Nothing = error "Pass us a type w/ a real atomicUnicorn plz"+magicSmokeStream (Just magic) = do+ unless (Just magic == atomicUnicorn) $ error "Pass us atomicUnicorn plz"+ (i,o) <- newChan+ strms <- streamChan 2 o+ case strms of+ [strm1,strm2] -> do+ writeChan i magic+ h <- tryReadNext strm1+ case h of+ Pending -> error "magicSmokeStream: strm1 pending!"+ (Next x1 xs1) -> do+ unless (x1 == magic) $ error "magicSmokeStream: x1 /= magic"+ h2 <- tryReadNext strm2+ case h2 of+ (Next _ _) -> error "magicSmokeStream: h2 /= Pending!"+ _ -> do writeChan i magic+ h2' <- tryReadNext strm2+ case h2' of+ Pending -> error "h2' == Pending!"+ Next x2 _ -> do + unless (x2 == magic) $ error "x2 /= magic"+ h' <- tryReadNext xs1+ case h' of+ (Next _ _) -> error "magicSmokeStream: h' /= Pending!"+ Pending -> return ()+ _ -> error "streamChan broken"++{- NOTE: we would like this to pass (see trac #9030) but are happy to note that+ - it fails which somewhat validates the test below++testBlockedRecovery = do+ (i,o) <- newChan+ v <- newEmptyMVar+ rid <- forkIO (putMVar v () >> readChan o)+ takeMVar v+ threadDelay 1000+ throwTo rid ThreadKilled+ -- we race the exception-handler in `readChan` here...+ writeChan i ()+ -- In a buggy implementation, this would consistently win failing by losing+ -- the message and raising BlockedIndefinitely here:+ readChan o+-}+++-- test for deadlocks caused by async exceptions in reader.+checkDeadlocksReaderUnagi :: Int -> IO ()+checkDeadlocksReaderUnagi times = do+ let run 0 normalRetries numRace = putStrLn $ "Lates: "++(show normalRetries)++", Races: "++(show numRace)+ run n normalRetries numRace+ | (normalRetries + numRace) > (times `div` 3) = error "This test is taking too long. Please retry, and if still failing send the log to me"+ | otherwise = do+ (i,o) <- UI.newChanStarting 0+ -- preload a chan with 0s+ let numPreloaded = 10000+ replicateM_ numPreloaded $ writeChan i (0::Int)++ rStart <- newEmptyMVar+ rid <- forkIO $ (putMVar rStart () >> (forever $ void $ readChan yield o))+ takeMVar rStart >> threadDelay 1+ throwTo rid ThreadKilled++ -- did killing reader damage queue for reads or writes?+ writeChan i 1 `onException` ( putStrLn "Exception from first writeChan!")+ writeChan i 2 `onException` ( putStrLn "Exception from second writeChan!")+ finalRead <- tryReadChanErr o `onException` ( putStrLn "Exception from final tryReadChan!")+ + oCnt <- readCounter $ (\(UI.OutChan _ (UI.ChanEnd cntr _))-> cntr) o+ iCnt <- readCounter $ (\(UI.InChan _ (UI.ChanEnd cntr _))-> cntr) i+ unless (iCnt == numPreloaded + 1) $ + error "The InChan counter doesn't match what we'd expect from numPreloaded!"++ case finalRead of+ 0 -> if oCnt <= 0 -- (technically, -1 means hasn't started yet)+ -- reader didn't have a chance to get started+ then putStr "0" >> run n (normalRetries+1) numRace+ -- normal run; we tested that killing a reader didn't+ -- break chan for other readers/writers:+ else putStr "." >> run (n-1) normalRetries numRace+ --+ -- Rare. Reader was killed after reading all pre-loaded messages+ -- but before starting what would be the blocking read:+ 1 | oCnt == numPreloaded -> putStr "X" >> run n normalRetries (numRace + 1)+ | otherwise -> error $ "Having read final 1, "+++ "Expecting a counter value of "++(show numPreloaded)+++ " but got: "++(show oCnt)+ 2 -> do unless (oCnt == numPreloaded + 1) $+ error $ "Having read final 2, "+++ "Expecting a counter value of "++(show $ numPreloaded+1)+++ " but got: "++(show oCnt)+ putStr "+" >> run n (normalRetries + 1) numRace++ _ -> error "Fix your #$%@ test!"++ run times 0 0+ putStrLn ""++-- do a series of writes, forcing GC making sure remains true, then do the last+-- write and loop on forcing GC until we see False.+isActiveTest :: IO ()+isActiveTest = do+ let n = 100000+ k = n `div` 100+ let testWrites (inc,outc) = replicateM_ 100 $ do+ replicateM_ k $ writeChan inc (0::Int)+ performGC+ actv <- isActive outc+ unless actv $+ error "isActive returned False before last write!"++ lastWriteWait iters (inc,outc) = writeChan inc (0::Int) >> go iters where+ go i | i < (0::Int) = error "Timed out waiting for isActive to return False. Anomaly or possible bug."+ | otherwise = do+ actv <- isActive outc+ when actv $ performGC >> go (i-1)+ -- first with newChan:+ c1 <- newChan+ testWrites c1+ lastWriteWait 1000 c1+ -- then with a duplicated channel:+ (inc2,_) <- newChan+ outc2 <- dupChan inc2+ testWrites (inc2,outc2)+ lastWriteWait 1000 (inc2,outc2)++-- Concurrently write [1..100000], while reading 100001 and prepending in a+-- IORef. Then a handler catches and puts () in an MVar which we wait on.+-- Then check that the elements were correct.+readChanYieldTest :: IO ()+readChanYieldTest = do+ let n = 100000 :: Int+ (inc,outc) <- newChan+ saving <- newIORef []+ exceptionRaised <- newIORef False+ goAhead <- newEmptyMVar++ let handling io = Control.Exception.catch io $ \BlockedIndefinitelyOnMVar ->+ writeIORef exceptionRaised True++ void $ forkIO $ replicateM_ (n+1) $ handling $ do+ x <- readChan yield outc+ modifyIORef' saving (x:)+ when (x == n) $ -- about to do final deadlocking loop:+ putMVar goAhead ()+ void $ forkIO $ forM_ [1..n] $ writeChan inc++ takeMVar goAhead+ out <- readIORef saving+ unless (out == [n,n-1..1]) $+ error "readChanYieldTest reads incorrect!"++ performGC+ threadDelay 100000+ raised <- readIORef exceptionRaised+ unless raised $+ error "Handler doesn't seem to have run in readChanYieldTest. Either a testing fluke or a bug."+++-- Smoke tests for different strides of interleaved streams, at different+-- offsets, with concurrent stream readers.+streamChanSmoke :: IO ()+streamChanSmoke =+ -- A few odd starting offsets, spanning Int/Counter overflow:+ forM_ [0, maxBound - UI.sEGMENT_LENGTH, maxBound - UI.sEGMENT_LENGTH + 1, maxBound, minBound] $ \startingOffset ->+ -- And few odd numbers of streams, where we especially want to exercise+ -- skips of entire segments: + forM_ [1,17, UI.sEGMENT_LENGTH, UI.sEGMENT_LENGTH+1, UI.sEGMENT_LENGTH*3+1] $ \numStreams-> do+ (i,o) <- UI.newChanStarting startingOffset+ let payload = 100000 :: Int+ -- and these are what we'll expect to see returned:+ let payloadPartsRev = map reverse $ [ [s,(s+numStreams).. payload ] | s<-[1..numStreams]]+ forM_ [1..payload] (writeChan i)+ strms <- streamChan numStreams o+ strmsReadOut <- replicateM numStreams newEmptyMVar+ unless (length strms == numStreams) $ + error $ "numStreams /= length strms: "+ ++(show numStreams)++" vs "++(show $ length strms)+ ++" at offset: "++(show startingOffset)+ forM_ (zip strms strmsReadOut) (forkIO . consumeUntilEmpty [])+ parts <- forM (zip strmsReadOut payloadPartsRev) $ \(v,expectedStack)-> do+ stack <- takeMVar v+ unless (stack == expectedStack) $ error $ "Incorrect stream reads: "++(show stack)+ return stack+ unless ((sort $ concat parts) == [1..payload]) $+ error $ "Somehow read parts weren't what we expected: "++(show parts)+ where consumeUntilEmpty stack (strm,v) = do+ h <- tryReadNext strm+ case h of+ (Next x xs) -> consumeUntilEmpty (x:stack) (xs,v)+ Pending -> putMVar v stack -- Done++-- Simple writer/streamer concurrency test+streamChanConcurrentStreamerWriter :: Int -> IO ()+streamChanConcurrentStreamerWriter n = do+ (i,o) <- newChan+ [strm] <- streamChan 1 o+ v <- newEmptyMVar+ let streamReader s stack itr failCnt + | failCnt > 4 = putMVar v $ Left "failCnt exceeded; possibly bug, but probably anomaly"+ | itr > n = putMVar v $ Right stack+ | otherwise = do+ xs <- tryReadNext s+ case xs of+ Pending -> threadDelay 1000 >> streamReader s stack itr (failCnt+1)+ Next x xs' -> streamReader xs' (x:stack) (itr+1) 0+ void $ forkIO $ streamReader strm [] (1::Int) (0::Int)+ void $ forkIO $ mapM_ (writeChan i) [1..n]+ strmOut <- either error return =<< takeMVar v+ unless (strmOut == [n,n-1..1]) $ + error $ "Stream reads were incorrect: "++(show strmOut)++-- Simple reader/streamer concurrency test+streamChanConcurrentStreamerReader :: Int -> IO ()+streamChanConcurrentStreamerReader n = do+ (i,o) <- newChan+ [strm] <- streamChan 1 o+ mapM_ (writeChan i) [1..n]+ vStream <- newEmptyMVar+ vOutchan <- newEmptyMVar+ let streamReader s stack = do+ xs <- tryReadNext s+ case xs of+ Pending -> putMVar vStream stack+ Next x xs' -> streamReader xs' (x:stack)++ outchanReader stack = do+ tryReadChan o >>= tryRead >>= maybe (putMVar vOutchan stack) (outchanReader . (:stack))+ void $ forkIO $ streamReader strm []+ void $ forkIO $ outchanReader []+ strmOut <- takeMVar vStream `onException` putStr " :in takeMVar vStream: "+ rdOut <- takeMVar vOutchan `onException` putStr " :in takeMVar vOutchan: "+ let correctOut = [n,n-1..1]+ unless (strmOut == correctOut) $ + error $ "Stream reads were incorrect: "++(show strmOut)+ unless (rdOut == correctOut) $ + error $ "OutChan reads were incorrect: "++(show rdOut)+
unagi-chan.cabal view
@@ -1,5 +1,5 @@ name: unagi-chan-version: 0.3.0.1+version: 0.3.0.2 synopsis: Fast concurrent queues with a Chan-like API, and more @@ -80,8 +80,7 @@ ghc-options: -Wall -funbox-strict-fields build-depends: base < 5- -- 0.6.1.1 is broken on GHC 7.10- , atomic-primops >= 0.6.0.5 && <= 0.6.1+ , atomic-primops >= 0.6.0.5 && <= 0.7 , primitive>=0.5.3 , ghc-prim default-language: Haskell2010@@ -123,7 +122,9 @@ test-suite test type: exitcode-stdio-1.0 ghc-options: -Wall -funbox-strict-fields- ghc-options: -O2 -rtsopts -threaded -with-rtsopts=-N+ ghc-options: -O2 -rtsopts -threaded + -- NOTE: configure --enable-profiling overrides this:+ ghc-options: -with-rtsopts=-N ghc-options: -fno-ignore-asserts -- for some hacks for Addr: ghc-options: -fno-warn-orphans @@ -140,9 +141,12 @@ , Smoke , Unagi , UnagiUnboxed+ , UnagiBounded+ , UnagiNoBlocking+ , UnagiNoBlockingUnboxed build-depends: base , primitive>=0.5.3- , atomic-primops >= 0.6.0.5 && <= 0.6.1+ , atomic-primops >= 0.6.0.5 && <= 0.7 , containers , ghc-prim default-language: Haskell2010