unagi-chan 0.1.0.1 → 0.1.0.2
raw patch · 10 files changed
+893/−1 lines, 10 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
Files
- tests/Atomics.hs +160/−0
- tests/Chan003.hs +25/−0
- tests/Deadlocks.hs +103/−0
- tests/DupChan.hs +81/−0
- tests/Implementations.hs +14/−0
- tests/IndexedMVar.hs +35/−0
- tests/Smoke.hs +72/−0
- tests/Unagi.hs +195/−0
- tests/UnagiUnboxed.hs +197/−0
- unagi-chan.cabal +11/−1
+ tests/Atomics.hs view
@@ -0,0 +1,160 @@+module Atomics (atomicsMain) where++import Control.Concurrent+import Data.Atomics.Counter.Fat+import Data.Atomics+import Data.IORef+import Control.Monad+import qualified Data.Set as Set+import Data.List+import Data.Bits++atomicsMain :: IO ()+atomicsMain = do+ putStrLn "Testing atomic-primops:"+ -- ------+ putStr " counter smoke test... "+ counterSane+ putStrLn "OK"+ -- ------+ putStr " counter overflow... "+ testCounterOverflow+ putStrLn "OK"+ -- ------+ putStr " counter is atomic... "+ counterTest+ putStrLn "OK"+ -- ------+ putStr " CAS... "+ testConsistentSuccessFailure+ putStrLn "OK"++-- catch real stupid bugs before machine gets hot:+counterSane :: IO ()+counterSane = do+ cntr <- newCounter 1337+ n <- readCounter cntr+ unless (n == 1337) $ error "newCounter magnificently broken"+ n2 <- incrCounter 1 cntr+ n2' <- readCounter cntr+ unless (n2 == 1338 && n2' == 1338) $ error "incrCounter magnificently broken"++cHUNK_SIZE, maxInt, minInt :: Int+cHUNK_SIZE = 32 -- MUST REMAIN POWER OF TWO for now+-- make sure incrCounter doesn't silently change to Integer or something+maxInt = maxBound+minInt = minBound++-- Test some properties of our counter we'd like to assume:+testCounterOverflow :: IO ()+testCounterOverflow = do+ let x `ourMod` y = + -- also do a little sanity checking for the bitwise mod over Int+ -- that we use heavily, and expect in some cases to roll over w/out+ -- breaks:+ let xmy = x .&. (y-1)+ in if xmy == x `mod` y+ then xmy+ else error "Our bitwise mod isn't working the way we expect in testCounterOverflow"+ cntr <- newCounter (maxInt - (cHUNK_SIZE `div` 2)) + spanningCntr <- replicateM cHUNK_SIZE (incrCounter 1 cntr)+ -- make sure our test is working+ if all (>0) spanningCntr || all (<0) spanningCntr+ then error "Sequence meant to span maxBound of counter not actually spanning"+ else return ()++ let l = map (`ourMod` cHUNK_SIZE) spanningCntr+ l' = (dropWhile (/= 0) l) ++ (takeWhile (/= 0) l)++ -- (1) test that we overflow the counter without any breaks and our mod function is working properly:+ unless (l' == [0..(cHUNK_SIZE - 1)]) $ + error $ "Uh Oh: "++(show l')++ -- (2) test that Ints and counter overflow in exactly the same way+ let spanningInts = take cHUNK_SIZE $ iterate (+1) (maxInt - (cHUNK_SIZE `div` 2) + 1) + unless (spanningInts == spanningCntr) $ do+ error $ "Ints overflow differently than counter: "+++ "\nInt: "++(show spanningInts)+++ "\nCounter: "++(show spanningCntr)++ -- (We don't use this property)+ cntr2 <- newCounter maxInt+ mbnd <- incrCounter 1 cntr2+ unless (mbnd == minInt) $ + error $ "Incrementing counter at maxbound didn't yield minBound"++ -- (3) test subtraction across boundary: count - newFirstIndex, for window spanning boundary.+ cntr3 <- newCounter (maxInt - 1)+ let ls = take 30 $ iterate (+1) $ maxInt - 10+ cs <- mapM (\x-> fmap (subtract x) $ incrCounter 1 cntr3) ls+ unless (cs == replicate 30 10) $ + error $ "Derp. We don't know how subtraction works: "++(show cs)++-- 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+ var <- newIORef "0"++ sem <- newIORef (0::Int)+ outs <- replicateM 2 newEmptyMVar ++ _ <- forkSync sem 2 $ test "a" var (outs!!0)+ _ <- forkSync sem 2 $ test "b" var (outs!!1)++ mapM takeMVar outs >>= examine+ -- w/r/t (2) above: we only try to find an element read along with False+ -- which wasn't sent by another thread, which isn't ideal+ where attempts = 100000+ test tag var out = do+ + res <- forM [(1::Int)..attempts] $ \x-> do+ let str = (tag++(show x))+ tk <- readForCAS var+ (b,tk') <- casIORef var tk str+ return (if b then str else peekTicket tk' , b)+ putMVar out res++ examine [res1, res2] = do+ -- any failures in either should be marked as successes in the other+ let (successes1,failures1) = (\(x,y)-> (Set.fromList $ map fst x, map fst y)) $ partition snd res1+ (successes2,failures2) = (\(x,y)-> (Set.fromList $ map fst x, map fst y)) $ partition snd res2+ ok1 = all (flip Set.member successes2) failures1+ ok2 = all (flip Set.member successes1) failures2+ if ok1 && ok2+ then when (length failures1 < (attempts `div` 6) || length failures2 < (attempts `div` 6)) $+ error "There was not enough contention to trust test. Please retry."+ else do print res1+ print res2+ error "FAILURE!"+ examine _ = error "Fix testConsistentSuccessFailure"++ +forkSync :: IORef Int -> Int -> IO () -> IO ThreadId+forkSync sem target io = + forkIO $ (busyWait >> io)+ where busyWait =+ atomicModifyIORef' sem (\n-> (n+1,())) >> wait+ wait = do+ n <- readIORef sem+ unless (n == target) wait+ ++counterTest :: IO ()+counterTest = do+ let n = 10000000+ nOut <- testAtomicCounter n+ when (nOut /= n) $+ error $ "Counter broken: expecting "++(show n)++" got "++(show nOut)++testAtomicCounter :: Int -> IO Int+testAtomicCounter n = do+ procs <- getNumCapabilities++ counter <- newCounter (0::Int)+ dones <- replicateM procs newEmptyMVar ; starts <- replicateM procs newEmptyMVar+ mapM_ (\(start1,done1)-> forkIO $ takeMVar start1 >> replicateM_ (n `div` procs) (incrCounter 1 counter) >> putMVar done1 ()) $ zip starts dones+ mapM_ (\v-> putMVar v ()) starts ; mapM_ (\v-> takeMVar v) dones+ + readCounter counter
+ tests/Chan003.hs view
@@ -0,0 +1,25 @@+module Chan003 (checkDeadlocksWriter) where++import Control.Concurrent+import qualified Control.Concurrent.Chan.Unagi as U+import Control.Exception+import Control.Monad++-- OBSOLETE FOR NOW; we have more clever and careful deadlock tests in+-- Deadlocks++-- test for deadlocks from async exceptions raised in writer+checkDeadlocksWriter :: Int -> IO ()+checkDeadlocksWriter n = void $+ replicateM_ n $ do+ (i,o) <- U.newChan+ wStart <- newEmptyMVar+ wid <- forkIO (putMVar wStart () >> ( forever $ U.writeChan i (0::Int)) )+ -- wait for writer to start+ takeMVar wStart >> threadDelay 1+ throwTo wid ThreadKilled+ -- did killing the writer damage queue for writes or reads?+ U.writeChan i (1::Int)+ z <- U.readChan o+ unless (z == 0) $+ error "Writer never got a chance to write!"
+ tests/Deadlocks.hs view
@@ -0,0 +1,103 @@+module Deadlocks (deadlocksMain) where++import Control.Concurrent.MVar+import Control.Concurrent(getNumCapabilities,threadDelay,forkIO)+import Control.Exception+import Control.Monad++import Implementations+++deadlocksMain :: IO ()+deadlocksMain = do+ let tries = 50000+ + putStrLn "==================="+ putStrLn "Testing Unagi:"+ -- ------+ putStr $ " Checking for deadlocks from killed reader, x"++show tries++"... "+ checkDeadlocksReader unagiImpl tries+ putStrLn "OK"+ -- ------+ putStr $ " Checking for deadlocks from killed writer, x"++show tries++"... "+ checkDeadlocksWriter unagiImpl tries+ putStrLn "OK"+ + putStrLn "==================="+ putStrLn "Testing Unagi.Unboxed:"+ -- ------+ putStr $ " Checking for deadlocks from killed reader, x"++show tries++"... "+ checkDeadlocksReader unboxedUnagiImpl tries+ putStrLn "OK"+ -- ------+ putStr $ " Checking for deadlocks from killed writer, x"++show tries++"... "+ checkDeadlocksWriter unboxedUnagiImpl tries+ putStrLn "OK"++++-- -- Chan002.hs -- --++++-- test for deadlocks caused by async exceptions in reader.+checkDeadlocksReader :: Implementation inc outc Int -> Int -> IO ()+checkDeadlocksReader (newChan,writeChan,readChan,_) times = do+ -- this might become an argument, indicating whether a killed reader might+ -- result in one missing element (currently all do)+ let mightDropOne = True+ procs <- getNumCapabilities+ let run _ 0 = putStrLn ""+ run retries n = do+ when (retries > (times `div` 5)) $+ error "This test is taking too long. Please retry, and if still failing send the log to me"+ (i,o) <- newChan+ -- if we don't have at least three cores, then we need to write enough messages in first, before killing reader.+ maybeWid <- if procs > 4 -- NOTE 4 might mean only two real cores, so be conservative here.+ then do wStart <- newEmptyMVar+ wid <- forkIO $ (putMVar wStart () >> (forever $ writeChan i (0::Int)))+ takeMVar wStart >> threadDelay 1 -- wait until we're writing+ return $ Just wid+ + else do replicateM_ 10000 $ writeChan i (0::Int)+ return Nothing+ rStart <- newEmptyMVar+ rid <- forkIO $ (putMVar rStart () >> (forever $ void $ readChan o))+ takeMVar rStart >> threadDelay 1+ throwTo rid ThreadKilled+ -- did killing reader damage queue for reads or writes?+ writeChan i 1 `onException` ( putStrLn "Exception from writeChan 1")+ when mightDropOne $+ writeChan i 2 `onException` ( putStrLn "Exception from last writeChan 2")+ z <- readChan o `onException` ( putStrLn "Exception from last readChan")+ -- clean up:+ case maybeWid of + Just wid -> throwTo wid ThreadKilled ; _ -> return ()+ case z of+ 0 -> putStr "." >> run retries (n-1)+ -- reader probably killed while blocked or before writer even wrote anything+ 1 -> putStr "+" >> run (retries+1) n + 2 | not mightDropOne -> error "Fix tests; 2 when not mightDropOne"+ | otherwise -> putStr "*" >> run (retries+1) n+ _ -> error "Fix checkDeadlocksReader test; not 0, 1, or 2"+ run 0 times+++-- -- Chan003.hs -- --+++-- test for deadlocks from async exceptions raised in writer+checkDeadlocksWriter :: Implementation inc outc Int -> Int -> IO ()+checkDeadlocksWriter (newChan,writeChan,readChan,_) n = void $+ replicateM_ n $ do+ (i,o) <- newChan+ wStart <- newEmptyMVar+ wid <- forkIO (putMVar wStart () >> ( forever $ writeChan i (0::Int)) )+ -- wait for writer to start+ takeMVar wStart >> threadDelay 1+ throwTo wid ThreadKilled+ -- did killing the writer damage queue for writes or reads?+ writeChan i (1::Int)+ z <- readChan o+ unless (z == 0) $+ error "Writer never got a chance to write!"
+ tests/DupChan.hs view
@@ -0,0 +1,81 @@+module DupChan (dupChanMain) where++-- implementation-agnostic tests of `dupChan`++import Control.Concurrent.MVar+import Control.Concurrent(forkIO,throwTo)+import Control.Exception(AsyncException(ThreadKilled))+import Control.Monad++import Implementations+++dupChanMain :: IO ()+dupChanMain = do+ putStrLn "==================="+ putStrLn "Test dupChan Unagi:"+ -- ------+ putStr " Reader/Reader... "+ replicateM_ 1000 $ dupChanTest1 unagiImpl 50000+ putStrLn "OK"+ -- ------+ putStr " Writer/dupChan+Reader... "+ replicateM_ 1000 $ dupChanTest2 unagiImpl 10000+ putStrLn "OK"+ putStrLn "==================="+ putStrLn "Test dupChan Unagi.Unboxed:"+ -- ------+ putStr " Reader/Reader... "+ replicateM_ 1000 $ dupChanTest1 unboxedUnagiImpl 50000+ putStrLn "OK"+ -- ------+ putStr " Writer/dupChan+Reader... "+ replicateM_ 1000 $ dupChanTest2 unboxedUnagiImpl 10000+ putStrLn "OK"++-- Check output where dupChan at known point in input stream, with two+-- concurrent readers.+dupChanTest1 :: Implementation inc outc Int -> Int -> IO ()+dupChanTest1 (newChan,writeChan,readChan,dupChan) n = do+ let s1 = [1.. ndiv2]+ s2 = [(ndiv2+1)..n]+ ndiv2 = n `div` 2+ (i,o) <- newChan+ mapM_ (writeChan i) s1+ oDup <- dupChan i+ mapM_ (writeChan i) s2++ out <- newEmptyMVar+ outDup <- newEmptyMVar++ _ <- forkIO $ (replicateM n (readChan o) >>= putMVar out)+ _ <- forkIO $ (replicateM ndiv2 (readChan oDup) >>= putMVar outDup)++ x <- takeMVar out+ y <- takeMVar outDup++ unless (x == s1++s2) $+ error ""+ unless (y == s2) $ + error $ "dupChan returned unexpected results: "++(show y)++-- Check concurrent writes with dupChan + reads, check all reads are some+-- contiguous part of the input stream.+dupChanTest2 :: Implementation inc outc Int -> Int -> IO ()+dupChanTest2 (newChan,writeChan,readChan,dupChan) n = do+ (i,o) <- newChan+ out <- newEmptyMVar+ writer <- forkIO $ mapM_ (writeChan i) [(0::Int)..]+ + s1 <- replicateM n (readChan o)+ _ <- forkIO (dupChan i >>= replicateM n . readChan >>= putMVar out)+ s2 <- replicateM n (readChan o)+ s3 <- takeMVar out++ throwTo writer ThreadKilled ++ unless (all increm [s1,s2,s3]) $ do+ print [s1,s2,s3]+ error "All read streams should be incrementing by one, without breaks"+ where increm [] = error "Fix dupChanTest2"+ increm xss@(x:_) = xss == take n [x..]
+ tests/Implementations.hs view
@@ -0,0 +1,14 @@+module Implementations where++import qualified Control.Concurrent.Chan.Unagi as U+import qualified Control.Concurrent.Chan.Unagi.Unboxed as UU+import qualified Data.Primitive as P++type Implementation inc outc a = (IO (inc a, outc a), inc a -> a -> IO (), outc a -> IO a, inc a -> IO (outc a))++unagiImpl :: Implementation U.InChan U.OutChan a+unagiImpl = (U.newChan, U.writeChan, U.readChan, U.dupChan)++unboxedUnagiImpl :: (P.Prim a)=> Implementation UU.InChan UU.OutChan a+unboxedUnagiImpl = (UU.newChan, UU.writeChan, UU.readChan, UU.dupChan)+
+ tests/IndexedMVar.hs view
@@ -0,0 +1,35 @@+module IndexedMVar (indexedMVarMain) where++import Utilities+import Control.Monad+import Control.Concurrent+import Control.Exception(onException)+import Data.IORef++indexedMVarMain :: IO ()+indexedMVarMain = do+ putStr "Test indexedMVar... "+ replicateM_ 1000 $ + indexedMVarTest 128+ putStrLn "OK"++indexedMVarTest :: Int -> IO ()+indexedMVarTest n = do+ let xs = [0..n]+ mvIx <- newIndexedMVar+ start <- newIORef (0::Int)+ _ <- forkIO $ do+ writeIORef start 1+ mapM_ (\x-> putMVarIx mvIx x x `onException` (putStr "In writer: ")) xs+ + busyWait start+ mapM_ (\x-> (readMVarIx mvIx x `onException` (putStr "In reader: ")) >>= assertEq x) xs+ where assertEq x x' = when (x /= x') $ error $ (show x)++"/="++(show x')++-- we should really abstract this out and use elsewhere+busyWait :: IORef Int -> IO ()+busyWait ref = busy where+ busy = do+ -- N.B. w/ readIORef we loop forever:+ n <- atomicModifyIORef ref (\x-> (x,x)) + unless (n == 1) busy
+ tests/Smoke.hs view
@@ -0,0 +1,72 @@+{-# LANGUAGE BangPatterns #-}+module Smoke (smokeMain) where++import Control.Monad+import Control.Concurrent(forkIO)+import qualified Control.Concurrent.Chan as C+import Data.List++import Implementations++smokeMain :: IO ()+smokeMain = do+ putStrLn "==================="+ putStrLn "Testing Unagi:"+ -- ------+ putStr " FIFO smoke test... "+ fifoSmoke unagiImpl 100000+ putStrLn "OK"+ -- ------+ testContention unagiImpl 2 2 1000000++ putStrLn "==================="+ putStrLn "Testing Unagi.Unboxed:"+ -- ------+ putStr " FIFO smoke test... "+ fifoSmoke unboxedUnagiImpl 100000+ putStrLn "OK"+ -- ------+ testContention unboxedUnagiImpl 2 2 1000000+++fifoSmoke :: Implementation inc outc Int -> Int -> IO ()+fifoSmoke (newChan,writeChan,readChan,_) n = do+ (i,o) <- newChan+ mapM_ (writeChan i) [1..n]+ nsOut <- replicateM n $ readChan o+ unless (nsOut == [1..n]) $+ error "Cough!"++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)+ -- e.g. [[1,2,3,4,5],[6,7,8,9,10]] for 2 2 10+ groups = map (\i-> [i.. i - 1 + nNice `quot` writers]) $ [1, (nNice `quot` writers + 1).. nNice]+ -- force list; don't change --+ out <- C.newChan++ (i,o) <- newChan+ -- some will get blocked indefinitely:+ void $ replicateM readers $ forkIO $ forever $+ readChan o >>= C.writeChan out+ + putStrLn $ "Sending "++(show $ length $ concat groups)++" messages, with "++(show readers)++" readers and "++(show writers)++" writers."+ mapM_ (forkIO . mapM_ (writeChan i)) groups++ ns <- replicateM nNice (C.readChan out)+ isEmpty <- C.isEmptyChan out+ if sort ns == [1..nNice] && isEmpty+ then let d = interleaving ns+ in if d < 0.75+ then putStrLn $ "Not enough interleaving of threads: "++(show $ d)++". Please try again or report a bug"+ else putStrLn $ "Success, with interleaving pct of "++(show $ d)++" (closer to 1 means we have higher confidence in the test)."+ else error "What we put in isn't what we got out :("++-- --------- Helpers:++-- approx measure of interleaving (and hence contention) in test+interleaving :: (Num a, Eq a) => [a] -> Float+interleaving [] = 0+interleaving (x:xs) = (snd $ foldl' countNonIncr (x,0) xs) / l+ where l = fromIntegral $ length xs+ countNonIncr (x0,!cnt) x1 = (x1, if x1 == x0+1 then cnt else cnt+1)
+ tests/Unagi.hs view
@@ -0,0 +1,195 @@+module Unagi (unagiMain) where++-- Unagi-chan-specific tests++import Control.Concurrent.Chan.Unagi+import qualified Control.Concurrent.Chan.Unagi.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++unagiMain :: IO ()+unagiMain = do+ putStrLn "==================="+ putStrLn "Testing Unagi 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"+ -- ------+ let tries = 50000+ putStrLn $ "Checking for deadlocks from killed Unagi reader in a fancy way, x"++show tries+ checkDeadlocksReaderUnagi tries+++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+ outp <- getChanContents o+ if and (zipWith (==) inp outp)+ then return ()+ else 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' <- readChan 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,UI.OutChan (UI.ChanEnd _ _ arrRef)) <- UI.newChanStarting n+ writeChan i ()+ (UI.StreamHead _ (UI.Stream arr _)) <- readIORef arrRef+ cell <- P.readArray arr 0+ case cell of+ UI.Written () -> 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+ UI.Written 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+ UI.Written 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` 5) = 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++ (i,o) <- UI.newChanStarting 0+ -- preload a chan with 0s+ let numPreloaded = 10000+ 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 ""
+ tests/UnagiUnboxed.hs view
@@ -0,0 +1,197 @@+module UnagiUnboxed (unagiUnboxedMain) where++-- Unagi-chan-specific tests+--+-- Forked from tests/Unagi.hs at 337600f++import Control.Concurrent.Chan.Unagi.Unboxed+import qualified Control.Concurrent.Chan.Unagi.Unboxed.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++unagiUnboxedMain :: IO ()+unagiUnboxedMain = do+ 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 "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"+ -- ------+ let tries = 50000+ putStrLn $ "Checking for deadlocks from killed Unagi reader in a fancy way, x"++show tries+ checkDeadlocksReaderUnagi tries+++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+ outp <- getChanContents o+ if and (zipWith (==) inp outp)+ then return ()+ else 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' <- readChan 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,UI.OutChan (UI.ChanEnd _ arrRef)) <- UI.newChanStarting n+ writeChan i (7::Int)+ (UI.StreamHead _ (UI.Stream sigArr eArr _ _)) <- readIORef arrRef+ cell <- UI.readElementArray eArr 0 :: IO Int+ case cell of+ 7 -> return ()+ _ -> error "Expected a write at index 0 of 7"+ sigCell <- P.readByteArray sigArr 0 :: IO Int+ unless (sigCell == 1) $ -- i.e. cellWritten+ error "Expected cellWritten at sigArr!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 sigArr eArr _ next)) <- readIORef arrRef+ -- check all of first segment:+ forM_ (init writes) $ \ix-> do+ cell <- UI.readElementArray eArr ix :: IO Int+ sigCell <- P.readByteArray sigArr ix :: IO Int+ unless (cell == ix && sigCell == 1) $+ error $ "Expected a write at index "++(show ix)++" of same value but got "++(show cell)++" with signal "++(show sigCell)+ -- check last write:+ lastSeg <- readIORef next+ case lastSeg of+ (UI.Next (UI.Stream sigArr2 eArr2 _ next2)) -> do+ cell <- UI.readElementArray eArr2 0 :: IO Int+ sigCell <- P.readByteArray sigArr2 0 :: IO Int+ unless (cell == last writes && sigCell == 1) $+ error $ "Expected last write at index "++(show $ last writes)++" of same value but got "++(show cell)+ -- 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` 5) = 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++ (i,o) <- UI.newChanStarting 0+ -- preload a chan with 0s+ let numPreloaded = 10000+ 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 ""
unagi-chan.cabal view
@@ -1,5 +1,5 @@ name: unagi-chan-version: 0.1.0.1+version: 0.1.0.2 synopsis: Fast and scalable concurrent queues for x86, with a Chan-like API @@ -84,6 +84,16 @@ -- I guess we need to put 'src' here to get access to Internal modules hs-source-dirs: tests, src main-is: Main.hs+ other-modules:+ Atomics+ , Chan003+ , Deadlocks+ , DupChan+ , Implementations+ , IndexedMVar+ , Smoke+ , Unagi+ , UnagiUnboxed build-depends: base , primitive>=0.5.3 , atomic-primops==0.6.0.5