diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright Asakamirai (c) 2015 + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above + copyright notice, this list of conditions and the following + disclaimer in the documentation and/or other materials provided + with the distribution. + + * Neither the name of Asakamirai nor the names of other + contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/bench/Main.hs b/bench/Main.hs new file mode 100644 --- /dev/null +++ b/bench/Main.hs @@ -0,0 +1,161 @@ +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE Rank2Types #-} + +module Main where + +import qualified Control.Concurrent.KazuraQueue as KZR + +-- import qualified Control.Concurrent.Chan.Unagi as UNG + +import qualified Control.Concurrent as CC +import qualified Control.Concurrent.Chan as Chan +import qualified Control.Concurrent.MVar as MVar +import qualified Control.Concurrent.STM as STM +import qualified Control.Exception as E +import qualified Control.Monad as M + +import qualified Criterion.Main as CR +import qualified Criterion.Types as CR + +async :: IO a -> IO (MVar.MVar a, CC.ThreadId) +async act = do + mv <- MVar.newEmptyMVar + thid <- CC.forkIO $ do + M.void . M.forever . CC.threadDelay $ 1000000 * 1000000 + M.void $ act >>= MVar.tryPutMVar mv + M.void . CC.forkIO $ do + (act >>= MVar.putMVar mv) `E.finally` CC.killThread thid + return (mv, thid) + +wait :: (MVar.MVar a, CC.ThreadId) -> IO a +wait = MVar.readMVar . fst + +type IterationSize = Int +type QueueNum = Int +type WriteAction = Int -> IO () +type ReadAction = IO Int +type WriteThreadNum = Int +type ReadThreadNum = Int +type ItemNum = Int +type TestOne = WriteThreadNum -> ReadThreadNum -> IO () + +data QueueActions v = forall q . QueueActions + { newQueue :: IO q + , writeQueue :: q -> v -> IO () + , readQueue :: q -> IO v + } + +------------------------ + +testSpeed :: ItemNum -> QueueActions Int -> TestOne +testSpeed inum (QueueActions newAct writeAct readAct) wth rth = go + where + wnum = inum `div` wth + rnum = inum `div` rth + ws = [0..(wnum-1)] :: [Int] + asyncw = M.replicateM wth . async + asyncr = M.replicateM rth . async + go = do + q <- newAct + wws <- asyncw $ M.forM_ ws $ writeAct q + rws <- asyncr . M.replicateM_ rnum $ readAct q + M.forM_ wws wait + M.forM_ rws wait + + +testCost :: IterationSize -> QueueNum -> ItemNum -> QueueActions Int -> TestOne +testCost itersize qnum inum (QueueActions newAct writeAct readAct) wth rth = do + itrq <- STM.newTQueueIO + M.replicateM_ itersize . STM.atomically $ STM.writeTQueue itrq () + qws <- M.replicateM qnum . async $ go itrq + M.forM_ qws wait + where + wnum = inum `div` wth + rnum = inum `div` rth + ws = [0..(wnum-1)] :: [Int] + asyncw = M.replicateM wth . async + asyncr = M.replicateM rth . async + go itrq = do + ma <- STM.atomically $ STM.tryReadTQueue itrq + case ma of + Nothing -> return () + Just () -> do + q <- newAct + wws <- asyncw $ M.forM_ ws $ writeAct q + rws <- asyncr . M.replicateM_ rnum $ readAct q + M.forM_ wws wait + M.forM_ rws wait + go itrq + +testChan :: QueueActions a +testChan = QueueActions + { newQueue = Chan.newChan + , writeQueue = Chan.writeChan + , readQueue = Chan.readChan + } + +{- +testUChan :: QueueActions a +testUChan = QueueActions + { newQueue = UNG.newChan + , writeQueue = UNG.writeChan . fst + , readQueue = UNG.readChan . snd + } +--} + +testTChan :: QueueActions a +testTChan = QueueActions + { newQueue = STM.newTChanIO + , writeQueue = \ q -> STM.atomically . STM.writeTChan q + , readQueue = STM.atomically . STM.readTChan + } + +testTQueue :: QueueActions a +testTQueue = QueueActions + { newQueue = STM.newTQueueIO + , writeQueue = \ q -> STM.atomically . STM.writeTQueue q + , readQueue = STM.atomically . STM.readTQueue + } + +testKZRQueue :: QueueActions a +testKZRQueue = QueueActions + { newQueue = KZR.newQueue + , writeQueue = KZR.writeQueue + , readQueue = KZR.readQueue + } + +main :: IO () +main = do + CR.defaultMainWith configSpeed + [ + CR.bgroup "KazuraQueue" $ testcases $ testSpeed_ testKZRQueue + -- , CR.bgroup "Unagi" $ testcases $ testSpeed_ testUChan + , CR.bgroup "Chan" $ testcases $ testSpeed_ testChan + , CR.bgroup "TQueue" $ testcases $ testSpeed_ testTQueue + , CR.bgroup "TChan" $ testcases $ testSpeed_ testTChan + ] + CR.defaultMainWith configCost + [ + CR.bgroup "KazuraQueue" $ testcases $ testCost_ testKZRQueue + -- , CR.bgroup "Unagi" $ testcases $ testCost_ testUChan + , CR.bgroup "Chan" $ testcases $ testCost_ testChan + , CR.bgroup "TQueue" $ testcases $ testCost_ testTQueue + , CR.bgroup "TChan" $ testcases $ testCost_ testTChan + ] + where + configCost = CR.defaultConfig + { CR.reportFile = Just "report_cost.html" } + configSpeed = CR.defaultConfig + { CR.reportFile = Just "report_speed.html" } + testSpeed_ = testSpeed 90000 + testCost_ = testCost 20 10 45000 + testcases test = + [ testcase 1 1 test + , testcase 3 1 test + , testcase 1 3 test + ] + testcase wth rth test = + CR.bench (show wth ++ "." ++ show rth) $ + CR.nfIO $ test wth rth + + diff --git a/kazura-queue.cabal b/kazura-queue.cabal new file mode 100644 --- /dev/null +++ b/kazura-queue.cabal @@ -0,0 +1,100 @@ +name: kazura-queue +version: 0.1.0.0 +synopsis: Fast concurrent queues much inspired by unagi-chan +description: + "kazura-queue" provides an implementation of FIFO queue. + It is faster than Chan, TQueue or TChan by the benefit of fetch-and-add + instruction. + Main motivation of this package is to solve some difficulty of "unagi-chan". + - In "unagi-chan", the item in the queue/chan can be lost when async + exception is throwed to the read thread while waiting for read. + (Although it has handler to recover lost item, + it is difficult to keep FIFO in such case) + - In "unagi-chan", garbage items of the queue cannot be collected + immediately. + Since the buffer in the queue has the reference to the items until the + buffer is garbage-collected. + "kazura-queue" is slightly slower than "unagi-chan" instead of solving + these issues. + And "kazura-queue" lost broadcast function to improve the second issue. + It means that kazura-queue is not "Chan" but is just "Queue". +homepage: http://github.com/asakamirai/kazura-queue +license: BSD3 +license-file: LICENSE +author: Asakamirai +maintainer: asakamirai_hackage@towanowa.net +copyright: 2015 Asakamirai +category: Concurrency +build-type: Simple +cabal-version: >=1.10 + +source-repository head + type: git + location: https://github.com/asakamirai/kazura-queue + branch: master + +library + hs-source-dirs: src + exposed-modules: Control.Concurrent.WVar + Control.Concurrent.KazuraQueue + build-depends: base >= 4.7 && < 5 + , primitive >= 0.5.3 + , atomic-primops >= 0.8 + , async >= 2.0 + , containers >= 0.5 + ghc-options: -Wall -O2 + default-language: Haskell2010 + +benchmark kazura-queue-bench + type: exitcode-stdio-1.0 + hs-source-dirs: bench + main-is: Main.hs + ghc-options: -Wall -O2 -threaded -rtsopts -with-rtsopts=-N4 + build-depends: base + , kazura-queue + , async >= 2.0 + , containers >= 0.5 + , stm >= 2.4 + , criterion >= 1.1 + -- , unagi-chan >= 0.4.0.0 + default-language: Haskell2010 + +test-suite kazura-queue-doctest + type: exitcode-stdio-1.0 + hs-source-dirs: test + main-is: doctest.hs + build-depends: base + , kazura-queue + , doctest >= 0.10 + , QuickCheck >= 2.8 + ghc-options: -Wall -O2 -threaded -rtsopts -with-rtsopts=-N + default-language: Haskell2010 + +test-suite kazura-queue-test + type: exitcode-stdio-1.0 + hs-source-dirs: test + main-is: Spec.hs + other-modules: Test.Concurrent + , Test.Expectations + , Test.KazuraQueue + , Test.Util + , Test.WVar + , WVarSpec + , WVarConcurrentSpec + , KazuraQueueSpec + , KazuraQueueConcurrentSpec + build-depends: base + , kazura-queue + , HUnit >= 1.2 + , hspec >= 2.1 + , hspec-expectations >= 0.7 + , QuickCheck >= 2.8 + , deepseq >= 1.4 + , containers >= 0.5 + , mtl >= 2.2 + , transformers >= 0.4 + , free >= 4.12 + , exceptions >= 0.8 + , async >= 2.0 + ghc-options: -Wall -O2 -threaded -rtsopts -with-rtsopts=-N + default-language: Haskell2010 diff --git a/src/Control/Concurrent/KazuraQueue.hs b/src/Control/Concurrent/KazuraQueue.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/KazuraQueue.hs @@ -0,0 +1,397 @@ +{-# LANGUAGE TupleSections #-} +{-# LANGUAGE TypeFamilies #-} + +-- | KazuraQueue is the fast queue implementation inspired by unagi-chan. + +module Control.Concurrent.KazuraQueue + ( Queue + , newQueue + , readQueue + , readQueueWithoutMask + , tryReadQueue + , tryReadQueueWithoutMask + , writeQueue + , writeQueueWithoutMask + , lengthQueue + , lengthQueue' + ) where + +import Control.Concurrent.WVar (WCached, WTicket, WVar) +import qualified Control.Concurrent.WVar as WVar + +import qualified Control.Concurrent as CC +import Control.Concurrent.MVar (MVar) +import qualified Control.Concurrent.MVar as MVar +import qualified Control.Exception as E +import qualified Control.Monad as M +import Control.Monad.Primitive (RealWorld) + +import qualified Data.Atomics as Atm +import qualified Data.Atomics.Counter as Atm +import Data.Bits ((.&.)) +import qualified Data.Bits as Bits +import Data.IORef (IORef) +import qualified Data.IORef as Ref +import qualified Data.Primitive.Array as Arr + +-------------------------------- +-- constants and its utilities + +{-# INLINE bufferLength #-} +bufferLength :: Int +bufferLength = 64 + +{-# INLINE logBufferLength #-} +logBufferLength :: Int +logBufferLength = 6 + +{-# INLINE divModBufferLength #-} +divModBufferLength :: Int -> (Int,Int) +divModBufferLength n = d `seq` m `seq` (d,m) + where + d = n `Bits.unsafeShiftR` logBufferLength + m = n .&. (bufferLength - 1) + +-------------------------------- +-- Queue + +-- | Type of a Queue. /a/ is the type of an item in the Queue. +data Queue a = Queue + { queueWriteStream :: {-# UNPACK #-} !(IORef (Stream a)) + , queueWriteCounter :: {-# UNPACK #-} !Atm.AtomicCounter + , queueReadStream :: {-# UNPACK #-} !(IORef (Stream a)) + , queueReadState :: {-# UNPACK #-} !(WVar (ReadState a)) + , queueNoneTicket :: !(Atm.Ticket (Item a)) + } + +data ReadState a = ReadState + { rsCounter :: {-# UNPACK #-} !Atm.AtomicCounter + , rsLimit :: {-# UNPACK #-} !StreamIndex + } + +type Buffer a = Arr.MutableArray RealWorld (Item a) + +type BufferSource a = IO (Buffer a) + +data Item a = + Item a + | None + | Wait {-# UNPACK #-} !(MVar a) + | Done + +data Stream a = Stream + { streamBuffer :: {-# UNPACK #-} !(Buffer a) + , streamNext :: {-# UNPACK #-} !(IORef (NextStream a)) + , streamOffset :: {-# UNPACK #-} !StreamIndex + } + +data NextStream a = + NextStream {-# UNPACK #-} !(Stream a) + | NextSource !(BufferSource a) + +type StreamIndex = Int +type BufferIndex = Int + +------------------------------ + +newBufferSource :: IO (BufferSource a) +newBufferSource = do + arr <- Arr.newArray bufferLength None + return (Arr.cloneMutableArray arr 0 bufferLength) + +newReadState :: StreamIndex -> IO (WVar (ReadState a)) +newReadState strIdx = do + rcounter <- Atm.newCounter strIdx + WVar.newWVar ReadState + { rsCounter = rcounter + , rsLimit = strIdx + } + +-- | Create a new empty 'Queue'. +newQueue :: IO (Queue a) +newQueue = do + bufSrc <- newBufferSource + buf <- bufSrc + noneTicket <- Atm.readArrayElem buf 0 + next <- Ref.newIORef $ NextSource bufSrc + let stream = Stream buf next initialOffset + wstream <- Ref.newIORef stream + wcounter <- Atm.newCounter initialIndex + rstream <- Ref.newIORef stream + rsvar <- newReadState initialIndex + return Queue + { queueWriteStream = wstream + , queueWriteCounter = wcounter + , queueReadStream = rstream + , queueReadState = rsvar + , queueNoneTicket = noneTicket + } + where + -- for test of counter overflow + initialOffset = maxBound - 3 + initialIndex = initialOffset - 1 + +---------------------------------------------------------- + +{-# INLINE waitItem #-} +waitItem :: Buffer a -> BufferIndex -> IO () +waitItem buf bufIdx = do + ticket <- Atm.readArrayElem buf bufIdx + case Atm.peekTicket ticket of + None -> do + mv <- MVar.newEmptyMVar + (_ret, next) <- Atm.casArrayElem buf bufIdx ticket $! Wait mv + case Atm.peekTicket next of + None -> error "impossible case waitItem" + Wait mv' -> M.void $ MVar.readMVar mv' + _ -> return () + Wait mv -> M.void $ MVar.readMVar mv + _ -> return () + +{-# INLINE writeItem #-} +writeItem :: Buffer a -> BufferIndex -> Atm.Ticket (Item a) -> a -> IO () +writeItem buf bufIdx ticket a = do + (suc, next) <- Atm.casArrayElem buf bufIdx ticket (Item a) + M.unless suc $ case Atm.peekTicket next of + Wait mv -> do + Arr.writeArray buf bufIdx $ Item a + MVar.putMVar mv a + _ -> error "impossible case writeItem" + +---------------------------------------------------------- + +-- | Read an item from the 'Queue'. +{-# INLINE readQueue #-} +readQueue :: Queue a -> IO a +readQueue = E.mask_ . readQueueWithoutMask + +-- | Non-masked version of 'readQueue'. +-- It is not safe for asynchronous exception. +{-# INLINE readQueueWithoutMask #-} +readQueueWithoutMask :: Queue a -> IO a +readQueueWithoutMask queue@(Queue _ _ _ rsvar _) = + WVar.cacheWVar rsvar >>= readQueueRaw queue + +readQueueRaw :: Queue a -> WCached (ReadState a) -> IO a +readQueueRaw queue rswc0 = do + rstr0 <- Ref.readIORef rstrRef + strIdx <- Atm.incrCounter 1 rcounter + if rlimit0 - strIdx >= 0 + then readStream rstrRef rstr0 strIdx + else do + rswt1 <- extendReadStreamWithLock rstr0 rswc0 True True + let rswc1 = rswc0 { WVar.cachedTicket = rswt1 } + readQueueRaw queue rswc1 + where + rstrRef = queueReadStream queue + rswt0 = WVar.cachedTicket rswc0 + (ReadState rcounter rlimit0) = WVar.readWTicket rswt0 + +-- | Try to read an item from the 'Queue'. It never blocks. +-- +-- Note: It decreases "length" of 'Queue' even when it returns Nothing. +-- In such case, "length" will be lower than 0. +{-# INLINE tryReadQueue #-} +tryReadQueue :: Queue a -> IO (Maybe a) +tryReadQueue = E.mask_ . tryReadQueueWithoutMask + +-- | Non-masked version of 'tryReadQueue'. +-- It is not safe for asynchronous exception. +{-# INLINE tryReadQueueWithoutMask #-} +tryReadQueueWithoutMask :: Queue a -> IO (Maybe a) +tryReadQueueWithoutMask queue@(Queue _ _ _ rsvar _) = + WVar.cacheWVar rsvar >>= tryReadQueueRaw queue + +tryReadQueueRaw :: Queue a -> WCached (ReadState a) -> IO (Maybe a) +tryReadQueueRaw queue rswc0 = do + rstr0 <- Ref.readIORef rstrRef + strIdx <- Atm.incrCounter 1 rcounter + if rlimit0 - strIdx >= 0 + then Just <$> readStream rstrRef rstr0 strIdx + else do + rswt1 <- extendReadStreamWithLock rstr0 rswc0 False False + let rswc1 = rswc0 { WVar.cachedTicket = rswt1 } + (ReadState _ rlimit1) = WVar.readWTicket rswt1 + loop <- if rlimit1 /= rlimit0 + then return True + else do + wcount <- Atm.readCounter wcounter + if wcount - strIdx >= 0 + then CC.yield >> return True + else return False + if loop + then tryReadQueueRaw queue rswc1 + else return Nothing + where + rstrRef = queueReadStream queue + rswt0 = WVar.cachedTicket rswc0 + (ReadState rcounter rlimit0) = WVar.readWTicket rswt0 + wcounter = queueWriteCounter queue + +{-# INLINE readStream #-} +readStream :: IORef (Stream a) -> Stream a -> StreamIndex -> IO a +readStream rstrRef rstr0 strIdx = do + (bufIdx, rstr1) <- targetStream rstr0 strIdx + M.when (bufIdx == 0) $ Ref.writeIORef rstrRef rstr1 + let buf = streamBuffer rstr1 + item <- Arr.readArray buf bufIdx + Arr.writeArray buf bufIdx Done + case item of + Item a -> return a + _ -> error "impossible case readQueue" + +extendReadStreamWithLock :: + Stream a + -> WCached (ReadState a) + -> Bool + -> Bool + -> IO (WTicket (ReadState a)) +extendReadStreamWithLock rstr0 rswc0 waitLock waitWrite = do + (suc, rswt1) <- WVar.tryTakeWCached rswc0 + let rstate1 = WVar.readWTicket rswt1 + if suc + then do + rstate2 <- extendReadStream rstate1 rstr0 waitWrite + `E.onException` WVar.putWCached rswc0 rstate1 + WVar.putWCached rswc0 rstate2 + else do + let rswc1 = rswc0 { WVar.cachedTicket = rswt1 } + if waitLock + then WVar.readFreshWCached rswc1 + else do + rswc2 <- WVar.recacheWCached rswc1 + return $ WVar.cachedTicket rswc2 + +{-# INLINE extendReadStream #-} +extendReadStream :: ReadState a -> Stream a -> Bool -> IO (ReadState a) +extendReadStream rstate0 rstr0 waitWrite = do + (rlimitNext1, rstr1) <- searchStreamReadLimit rstr0 rlimitNext0 + if rlimitNext0 /= rlimitNext1 + then newRState rlimitNext1 + else if waitWrite + then do + let (Stream buf1 _ offset1) = rstr1 + bufIdx1 = rlimitNext1 - offset1 + waitItem buf1 bufIdx1 + (rlimitNext2, _) <- searchStreamReadLimit rstr1 rlimitNext1 + newRState rlimitNext2 + else return rstate0 + where + rlimit0 = rsLimit rstate0 + rlimitNext0 = rlimit0 + 1 + newRState rlimitNext = do + rcounter <- Atm.newCounter rlimit0 + return rstate0 + { rsCounter = rcounter + , rsLimit = rlimitNext - 1 + } + +-- | Write an item to the 'Queue'. +-- The item is evaluated (WHNF) before actual queueing. +writeQueue :: Queue a -> a -> IO () +writeQueue queue = E.mask_ . writeQueueRaw queue + +-- | Non-masked version of 'writeQueue'. +-- It is not safe for asynchronous exception. +{-# INLINE writeQueueRaw #-} +writeQueueWithoutMask :: Queue a -> a -> IO () +writeQueueWithoutMask = writeQueueRaw + +writeQueueRaw :: Queue a -> a -> IO () +writeQueueRaw (Queue wstrRef wcounter _ _ noneTicket) a = do + wstr0 <- Ref.readIORef wstrRef + strIdx <- Atm.incrCounter 1 wcounter + (bufIdx, wstr1) <- targetStream wstr0 strIdx + writeItem (streamBuffer wstr1) bufIdx noneTicket a + M.when (bufIdx == 0) $ Ref.writeIORef wstrRef wstr1 + +{-# INLINE targetStream #-} +targetStream :: Stream a -> StreamIndex -> IO (BufferIndex, Stream a) +targetStream str0@(Stream _ _ offset) strIdx = do + let (strNum, bufIdx) = divModBufferLength $ strIdx - offset + str1 <- getStream strNum bufIdx str0 + return (bufIdx, str1) + where + {-# INLINE getStream #-} + getStream 0 _ strA = return strA + getStream n bufIdx strA = do + strB <- waitNextStream strA bufIdx + getStream (n-1) bufIdx strB + +{-# NOINLINE waitNextStream #-} +waitNextStream :: Stream a -> Int -> IO (Stream a) +waitNextStream (Stream _ nextStrRef offset) = go + where + {-# INLINE go #-} + go wait = do + ticket <- Atm.readForCAS nextStrRef + case Atm.peekTicket ticket of + NextStream strNext -> return strNext + nextSrc@(NextSource bufSrc) + | wait > 0 -> do + CC.yield + go (wait - 1) + | otherwise -> do + newBuf <- bufSrc + newNext <- Ref.newIORef nextSrc + let nextStrCand = NextStream Stream + { streamBuffer = newBuf + , streamNext = newNext + , streamOffset = offset + bufferLength + } + (_, next) <- Atm.casIORef nextStrRef ticket nextStrCand + case Atm.peekTicket next of + NextStream nextStr -> return nextStr + NextSource _ -> go 1 + +-- | Search 'Stream' and return 'StreamIndex' and its 'Stream' +-- of the oldest unavailable Item. +{-# INLINE searchStreamReadLimit #-} +searchStreamReadLimit :: Stream a -> StreamIndex -> IO (StreamIndex, Stream a) +searchStreamReadLimit baseStr strIdx = + go (strIdx - streamOffset baseStr) baseStr + where + {-# INLINE go #-} + go bufIdx stream@(Stream buf _ offset) = do + ret <- searchBufferReadLimit buf bufIdx + case ret of + Just retBufIdx -> return (offset + retBufIdx, stream) + Nothing -> waitNextStream stream 0 >>= go 0 + +-- | Search 'Buffer' and return 'BufferIndex' +-- of the oldest unavailable Item. +-- If all Item in the Buffer is ready, return Nothing. +{-# INLINE searchBufferReadLimit #-} +searchBufferReadLimit :: Buffer a -> BufferIndex -> IO (Maybe BufferIndex) +searchBufferReadLimit buf = go + where + {-# INLINE go #-} + go bufIdx + | idxIsOutOfBuf = return Nothing + | otherwise = do + item <- Arr.readArray buf bufIdx + case item of + None -> return $ Just bufIdx + Wait _ -> return $ Just bufIdx + _ -> go $ bufIdx + 1 + where + idxIsOutOfBuf = bufIdx >= bufferLength + +-- | Get the length of the items in the 'Queue'. +-- +-- Caution: It returns the value which is lower than 0 +-- when the Queue is empty and some threads are waiting for new value. +lengthQueue :: Queue a -> IO Int +lengthQueue (Queue _ wcounter _ rsvar _) = do + rs <- WVar.readWVar rsvar + wcount <- Atm.readCounter wcounter + rcount <- Atm.readCounter $ rsCounter rs + return $ wcount - rcount + +-- | Non-minus version of 'lengthQueue'. +lengthQueue' :: Queue a -> IO Int +lengthQueue' queue = f <$> lengthQueue queue + where + f i | i > 0 = i + | otherwise = 0 + diff --git a/src/Control/Concurrent/WVar.hs b/src/Control/Concurrent/WVar.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/WVar.hs @@ -0,0 +1,278 @@ +{-# LANGUAGE BangPatterns #-} + +-- | WVar is waitable 'IORef'. +-- It is similar to 'MVar' but different at some points. +-- +-- * The latest (cached) value can be read +-- while someone is updating the value. +-- * Put operation can overwrite the value if the value is fresh +-- and cannot be blocked for waiting empty. +-- * WVar is strict. It means that the new value storing into the WVar +-- will be evaluated (WHNF) before actual storing. +-- +-- There are two states in the user viewpoint. +-- +-- [@Fresh@] The 'WVar' is not being updated. +-- This state corresponds to to full state of MVar. +-- [@Updating@] The 'WVar' is being updated by someone. +-- This state corresponds to to empty state of MVar. +-- However, cached previous value can be read while Updating. + +module Control.Concurrent.WVar + ( + -- * WVar + -- $wvar + WVar + , newWVar + , takeWVar + , tryTakeWVar + , putWVar + , readWVar + , readFreshWVar + , tryReadFreshWVar + -- * WCached + -- $wcached + , WCached(..) + , WTicket + , cacheWVar + , recacheWCached + , readWTicket + , takeWCached + , tryTakeWCached + , putWCached + , tryPutWCached + , readWCached + , readFreshWCached + , tryReadFreshWCached + ) + where + +import Control.Concurrent.MVar (MVar) +import qualified Control.Concurrent.MVar as MVar +import qualified Control.Monad as M +import qualified Data.Atomics as Atm +import Data.IORef (IORef) +import qualified Data.IORef as Ref + +------------------------------ +-- WVar + +-- | "a" is the type of data in the WVar. +newtype WVar a = WVar (IORef (WContent a)) + deriving Eq + +-- | Create a fresh 'WVar' that contains the supplied value. +{-# INLINE newWVar #-} +newWVar :: a -> IO (WVar a) +newWVar !a = WVar <$> Ref.newIORef WContent + { wvalue = a + , wstate = Fresh + } + +-- | Take the value of a 'WVar' like 'Control.Concurrent.MVar.takeMVar'. +-- It blocks when the 'WVar' is being updated. +{-# INLINE takeWVar #-} +takeWVar :: WVar a -> IO a +takeWVar wv = do + wc <- cacheWVar wv + wt1 <- takeWCached wc + return $ readWTicket wt1 + +-- | Non-blocking version of 'takeWVar'. +{-# INLINE tryTakeWVar #-} +tryTakeWVar :: WVar a -> IO (Bool, a) +tryTakeWVar wv = cacheWVar wv >>= go + where + go wc = do + (suc, wt1) <- tryTakeWCached wc + case (suc, wstate (Atm.peekTicket wt1)) of + (False, Fresh) -> go $ WCached wv wt1 + _ -> return (suc, readWTicket wt1) + +-- | Put the supplied value into a 'WVar'. +-- It performs simple "write" when the 'WVar' is Fresh. +-- When the supplied value is already evaluated, it never blocks. +{-# INLINE putWVar #-} +putWVar :: WVar a -> a -> IO () +putWVar wv a = do + wc <- cacheWVar wv + M.void $ putWCached wc a + +-- | Read the cached value of the 'WVar'. It never blocks. +{-# INLINE readWVar #-} +readWVar :: WVar a -> IO a +readWVar (WVar ref) = wvalue <$> Ref.readIORef ref + +-- | Read the fresh value of the 'WVar'. +-- It blocks and waits for a fresh value +-- when the 'WVar' is being updated by someone. +{-# INLINE readFreshWVar #-} +readFreshWVar :: WVar a -> IO a +readFreshWVar wv = do + wc <- cacheWVar wv + readWTicket <$> readFreshWCached wc + +-- | Non-blocking version of 'readFreshWVar' +{-# INLINE tryReadFreshWVar #-} +tryReadFreshWVar :: WVar a -> IO (Bool, a) +tryReadFreshWVar wv = do + wc <- cacheWVar wv + (suc, wt1) <- tryReadFreshWCached wc + return (suc, readWTicket wt1) + +------------------------------ +-- $wcached +-- Low level types and functions of 'WVar'. + +-- | WCached consists of WVar and its cached ticket. +data WCached a = WCached + { cachedVar :: {-# UNPACK #-} !(WVar a) + , cachedTicket :: WTicket a + } deriving Eq + +instance Show a => Show (WCached a) where + show (WCached _ wt) = "WCached " ++ show (readWTicket wt) + +type WTicket a = Atm.Ticket (WContent a) +data WContent a = WContent + { wvalue :: !a + , wstate :: !(WState a) + } + +instance Show a => Show (WContent a) where + show wcnt = concat [show (wstate wcnt), " ", show (wvalue wcnt)] + +-- | State of the 'WVar' +data WState a = + Fresh + -- ^ The value of the 'WVar' is fresh. Not updating now. + | Updating + -- ^ The value of the 'WVar' is updating. + -- No one wait for the fresh value. + | Waiting {-# UNPACK #-} !(MVar (WTicket a)) + -- ^ The value of WVar is updating + -- and the fresh value is waited for by someone. + +instance Show (WState a) where + show Fresh = "Fresh" + show Updating = "Updating" + show (Waiting _) = "Waiting" + +-- | Cache the current value of the 'WVar' and create 'WCached'. +{-# INLINE cacheWVar #-} +cacheWVar :: WVar a -> IO (WCached a) +cacheWVar wv@(WVar ref) = do + wt <- Atm.readForCAS ref + return WCached { cachedVar = wv, cachedTicket = wt } + +-- | Recache the 'WCached'. +-- +-- @recacheWCached = cacheWVar . cachedVar@ +recacheWCached :: WCached a -> IO (WCached a) +recacheWCached = cacheWVar . cachedVar + +-- | Read the value of the 'WTicket' +{-# INLINE readWTicket #-} +readWTicket :: WTicket a -> a +readWTicket = wvalue . Atm.peekTicket + +-- | Take the value of the 'WCached' like 'Control.Concurrent.MVar.takeMVar'. +-- It blocks when the 'WCached' is being updated. +{-# INLINE takeWCached #-} +takeWCached :: WCached a -> IO (WTicket a) +takeWCached wc@(WCached wv@(WVar ref) wt0) = do + let wcnt0 = Atm.peekTicket wt0 + (suc, wt2) <- case wstate wcnt0 of + Fresh -> do + (suc, wt1) <- Atm.casIORef ref wt0 wcnt0 { wstate = Updating } + if suc + then return (True, wt1) + else return (False, wt1) + Updating -> do + wt1 <- beginWaitWCached wc + return (False, wt1) -- mv will be read in next time + Waiting mv -> do + wt1 <- MVar.readMVar mv + return (False, wt1) + if suc + then return wt2 + else takeWCached $ WCached wv wt2 + +-- | Non-blocking version of 'takeWCached'. +{-# INLINE tryTakeWCached #-} +tryTakeWCached :: WCached a -> IO (Bool, WTicket a) +tryTakeWCached (WCached (WVar ref) wt0) = do + let wcnt0 = Atm.peekTicket wt0 + case wstate wcnt0 of + Fresh -> Atm.casIORef ref wt0 wcnt0 { wstate = Updating } + Updating -> return (False, wt0) + Waiting _ -> return (False, wt0) + +-- | Put the value to the 'WCached'. +-- It performs simple "write" when the 'WVar' is /Fresh/. +-- When the supplied value is already evaluated, it never blocks. +{-# INLINE putWCached #-} +putWCached :: WCached a -> a -> IO (WTicket a) +putWCached wc0 !a = do + (suc, wt1) <- tryPutWCached wc0 a + if suc + then return wt1 + else do + let wc1 = wc0 { cachedTicket = wt1 } + putWCached wc1 a + +-- | Put the value to a 'WCached'. +-- It performs simple "write" when the 'WVar' is /Fresh/. +-- It fails when the cache is obsoleted. +-- When the supplied value is already evaluated, it never blocks. +{-# INLINE tryPutWCached #-} +tryPutWCached :: WCached a -> a -> IO (Bool, WTicket a) +tryPutWCached (WCached (WVar ref) wt0) !a = do + let !wcnt1 = WContent { wvalue = a, wstate = Fresh } + (suc, wt1) <- Atm.casIORef ref wt0 wcnt1 + M.when suc $ case wstate $ Atm.peekTicket wt0 of + -- putMVar is never blocked + -- because it's done after casIORef had succeeded. + Waiting mv -> MVar.putMVar mv wt1 + _ -> return () + return (suc, wt1) + +-- | Read the cached value of the 'WCached'. It never blocks. +{-# INLINE readWCached #-} +readWCached :: WCached a -> a +readWCached (WCached _ wt0) = readWTicket wt0 + +-- | Read the /Fresh/ value of the 'WCached'. +-- It blocks and waits for a /Fresh/ value +-- when the 'WCached' is being updated by someone. +{-# INLINE readFreshWCached #-} +readFreshWCached :: WCached a -> IO (WTicket a) +readFreshWCached wc@(WCached wv wt0) = do + let wcnt0 = Atm.peekTicket wt0 + (suc, wt2) <- case wstate wcnt0 of + Fresh -> return (True, wt0) + Updating -> do + wt1 <- beginWaitWCached wc + return (False, wt1) + Waiting mv -> do + wt1 <- MVar.readMVar mv + return (True, wt1) + if suc + then return wt2 + else readFreshWCached $ WCached wv wt2 + +-- | Non-blocking version of 'readFreshWCached' +{-# INLINE tryReadFreshWCached #-} +tryReadFreshWCached :: WCached a -> IO (Bool, WTicket a) +tryReadFreshWCached (WCached _ wt0) = case wstate $ Atm.peekTicket wt0 of + Fresh -> return (True, wt0) + _ -> return (False, wt0) + +{-# INLINE beginWaitWCached #-} +beginWaitWCached :: WCached a -> IO (WTicket a) +beginWaitWCached (WCached (WVar ref) wt0) = do + mv <- MVar.newEmptyMVar + let wcnt0 = Atm.peekTicket wt0 + wcnt1 = wcnt0 { wstate = Waiting mv } + snd <$> Atm.casIORef ref wt0 wcnt1 + diff --git a/test/KazuraQueueConcurrentSpec.hs b/test/KazuraQueueConcurrentSpec.hs new file mode 100644 --- /dev/null +++ b/test/KazuraQueueConcurrentSpec.hs @@ -0,0 +1,303 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TupleSections #-} + +module KazuraQueueConcurrentSpec where + +import qualified Test.Concurrent as T +import qualified Test.Expectations as T +import qualified Test.KazuraQueue as T +import qualified Test.Util as T + +import qualified Test.Hspec as HS +import qualified Test.Hspec.QuickCheck as HS +import qualified Test.QuickCheck as Q + +import qualified Control.Concurrent as CC +import qualified Control.Concurrent.Async as AS +import qualified Control.Concurrent.KazuraQueue as KQ +import qualified Control.Exception as E +import qualified Control.Monad as M + +import qualified Data.Foldable as TF +import qualified Data.IORef as Ref +import qualified Data.List as L +import qualified Data.Map.Strict as Map +import Data.Monoid ((<>)) +import qualified Data.Set as Set +import qualified Data.Traversable as TF + +writeQueueSpec :: HS.Spec +writeQueueSpec = HS.describe "writeQueue" $ do + T.whenItemsInQueue (1,10) $ \ prepare -> do + HS.prop "write and read values concurrently" . prepare $ \ (q, pre) -> do + (val1 :: Int, val2, val3) <- Q.generate Q.arbitrary + T.mapConcurrently_ + [ KQ.writeQueue q val1 `T.shouldNotBlock` 500000 + , KQ.writeQueue q val2 `T.shouldNotBlock` 500000 + , KQ.writeQueue q val3 `T.shouldNotBlock` 500000 + ] + let len0 = length pre + q `T.queueLengthShouldBeIn` (len0, 3+len0) + ret1 <- M.replicateM len0 $ + KQ.readQueue q `T.shouldNotBlock` 500000 + ret2 <- M.replicateM 3 $ + KQ.readQueue q `T.shouldNotBlock` 500000 + let ret = ret1 ++ ret2 + T.oneOf + [ ret `T.shouldBe` (pre ++ [val1, val2, val3]) + , ret `T.shouldBe` (pre ++ [val1, val3, val2]) + , ret `T.shouldBe` (pre ++ [val2, val1, val3]) + , ret `T.shouldBe` (pre ++ [val2, val3, val1]) + , ret `T.shouldBe` (pre ++ [val3, val1, val2]) + , ret `T.shouldBe` (pre ++ [val3, val2, val1]) + ] + +readQueueSpec :: HS.Spec +readQueueSpec = HS.describe "readQueue" $ do + T.whenQueueIsEmpty $ \ prepare -> do + HS.prop "values are read in order (thread awakes out of order)" . prepare $ \ q -> do + waits0 <- M.replicateM 3 $ KQ.readQueue q `T.shouldBlock` 500000 + q `T.queueLengthShouldBeIn` (-3, 0) + (val1 :: Int, val2, val3) <- Q.generate Q.arbitrary + + KQ.writeQueue q val1 `T.shouldNotBlock` 500000 + (r1, waits1) <- waits0 `T.onlyOneShouldAwakeFinish` 500000 + q `T.queueLengthShouldBeIn` (-3, 0) + KQ.writeQueue q val2 `T.shouldNotBlock` 500000 + (r2, waits2) <- waits1 `T.onlyOneShouldAwakeFinish` 500000 + q `T.queueLengthShouldBeIn` (-3, 0) + KQ.writeQueue q val3 `T.shouldNotBlock` 500000 + (r3, _) <- waits2 `T.onlyOneShouldAwakeFinish` 500000 + q `T.queueLengthShouldBeIn` (-3, 0) + (r1, r2, r3) `T.shouldBe` (val1, val2, val3) + +tryReadQueueSpec :: HS.Spec +tryReadQueueSpec = HS.describe "tryReadQueue" $ do + T.whenItemsInQueue (1,10) $ \ prepare -> do + HS.prop "values are read in order" . prepare $ \ (q, pre :: [Int]) -> do + mrets <- M.replicateM 10 $ KQ.tryReadQueue q `T.shouldNotBlock` 500000 + q `T.queueLengthShouldBeIn` (-10, 0) + let nothings = L.replicate (10 - length pre) Nothing + expected = (Just <$> pre) <> nothings + mrets `T.shouldBe` expected + T.whenQueueIsEmpty $ \ prepare -> do + HS.prop "read value after writing" . prepare $ \ q -> do + (val1 :: Int, val2, val3, val4) <- Q.generate Q.arbitrary + KQ.writeQueue q val1 `T.shouldNotBlock` 500000 + mret1 <- KQ.tryReadQueue q `T.shouldNotBlock` 500000 + mret1 `T.shouldBe` Just val1 + KQ.writeQueue q val2 `T.shouldNotBlock` 500000 + mret2 <- KQ.tryReadQueue q `T.shouldNotBlock` 500000 + mret2 `T.shouldBe` Just val2 + KQ.writeQueue q val3 `T.shouldNotBlock` 500000 + mret3 <- KQ.tryReadQueue q `T.shouldNotBlock` 500000 + mret3 `T.shouldBe` Just val3 + KQ.writeQueue q val4 `T.shouldNotBlock` 500000 + mret4 <- KQ.tryReadQueue q `T.shouldNotBlock` 500000 + mret4 `T.shouldBe` Just val4 + +readWriteQueueSpec :: HS.Spec +readWriteQueueSpec = HS.describe "readWriteQueueSpec" $ do + T.whenQueueIsEmpty $ \ prepare -> do + HS.prop "read/write = 1/1" . prepare $ \ q -> do + test (1,10000) (1,10000) q + HS.prop "read/write = 1/10" . prepare $ \ q -> do + test (1,10000) (10,1000) q + HS.prop "read/write = 10/1" . prepare $ \ q -> do + test (10,1000) (1,10000) q + HS.prop "read/write = 10/10" . prepare $ \ q -> do + test (10,1000) (10,1000) q + where + test :: (Int,Int) -> (Int,Int) -> KQ.Queue (Int,Int) -> IO () + test readConfig writeConfig q = do + (results, writtens) <- + readConcurrent q readConfig + `T.concurrently` writeConcurrent q writeConfig + case checkEachResult results of + Right _ -> return () + Left str -> T.assertFailure str + let result = L.concat results + written = L.concat writtens + resultSet = Set.fromList result + writtenSet = Set.fromList written + length result `T.shouldBe` length written + (writtenSet `Set.difference` resultSet) `T.shouldBe` Set.empty + checkEachResult = TF.traverse checkEachItems + checkEachItems = L.foldl' checkItems $ Right Map.empty + checkItems (Right mp) (thnum, num) + | Map.lookup thnum mp < Just num = Right $ Map.insert thnum num mp + | Map.lookup thnum mp > Just num = Left "broken order" + | otherwise = Left "duplicated value" + checkItems err _ = err + readItems q size = M.replicateM size $ KQ.readQueue q + writeItems q items = do + TF.for_ items $ KQ.writeQueue q + return items + readConcurrent q (thsize, itemsize) = do + ass <- M.replicateM thsize . AS.async $ readItems q itemsize + TF.for ass AS.wait + writeConcurrent q (thsize, itemsize) = do + ass <- TF.for [1..thsize] $ \ thnum -> + AS.async . writeItems q $ fmap (thnum,) [1..itemsize] + TF.for ass AS.wait + +tryReadWriteQueueSpec :: HS.Spec +tryReadWriteQueueSpec = HS.describe "tryReadWriteQueueSpec" $ do + T.whenQueueIsEmpty $ \ prepare -> do + HS.prop "read/write = 1/1" . prepare $ \ q -> do + test (1,10000) (1,10000) q + HS.prop "read/write = 1/10" . prepare $ \ q -> do + test (1,10000) (10,1000) q + HS.prop "read/write = 10/1" . prepare $ \ q -> do + test (10,1000) (1,10000) q + HS.prop "read/write = 10/10" . prepare $ \ q -> do + test (10,1000) (10,1000) q + where + test :: (Int,Int) -> (Int,Int) -> KQ.Queue (Int,Int) -> IO () + test readConfig writeConfig q = do + (results, writtens) <- + readConcurrent q readConfig + `T.concurrently` writeConcurrent q writeConfig + case checkEachResult results of + Right _ -> return () + Left str -> T.assertFailure str + let result = L.concat results + written = L.concat writtens + resultSet = Set.fromList result + writtenSet = Set.fromList written + length result `T.shouldBe` length written + (writtenSet `Set.difference` resultSet) `T.shouldBe` Set.empty + checkEachResult = TF.traverse checkEachItems + checkEachItems = L.foldl' checkItems $ Right Map.empty + checkItems (Right mp) (thnum, num) + | Map.lookup thnum mp < Just num = Right $ Map.insert thnum num mp + | Map.lookup thnum mp > Just num = Left "broken order" + | otherwise = Left "duplicated value" + checkItems err _ = err + readItem q = do + mret <- KQ.tryReadQueue q + case mret of + Just ret -> return ret + Nothing -> do + CC.yield + readItem q + readItems q size = M.replicateM size $ readItem q + writeItems q items = do + TF.for_ items $ KQ.writeQueue q + return items + readConcurrent q (thsize, itemsize) = do + ass <- M.replicateM thsize . AS.async $ readItems q itemsize + TF.for ass AS.wait + writeConcurrent q (thsize, itemsize) = do + ass <- TF.for [1..thsize] $ \ thnum -> + AS.async . writeItems q $ fmap (thnum,) [1..itemsize] + TF.for ass AS.wait + +readQueueWithExceptionSpec :: HS.Spec +readQueueWithExceptionSpec = HS.describe "readQueueWithExceptionSpec" $ do + T.whenQueueIsEmpty $ \ prepare -> do + HS.prop "read/write = 1/1" . prepare $ \ q -> do + test (1,10000) (1,10000) q + HS.prop "read/write = 1/10" . prepare $ \ q -> do + test (1,10000) (10,1000) q + HS.prop "read/write = 10/1" . prepare $ \ q -> do + test (10,1000) (1,10000) q + HS.prop "read/write = 10/10" . prepare $ \ q -> do + test (10,1000) (10,1000) q + HS.prop "read/write ratio random 100000" . prepare $ \ q -> do + let genthnum = Q.arbitrary `Q.suchThat` (> 0) + `Q.suchThat` ((== 0).(100000 `mod`)) + rthnum <- Q.generate $ genthnum + wthnum <- Q.generate $ genthnum + let rnum = 100000 `div` rthnum + wnum = 100000 `div` wthnum + test (rthnum,rnum) (wthnum,wnum) q + where + test :: (Int,Int) -> (Int,Int) -> KQ.Queue (Int,Int) -> IO () + test readConfig writeConfig q = do + (results, writtens) <- readConcurrent q readConfig + `T.concurrently` writeConcurrent q writeConfig +-- putStrLn "-------------------" +-- print results +-- putStrLn "-------------------" + case checkEachResult results of + Right _ -> return () + Left str -> T.assertFailure str + let result = L.concat results + written = L.concat writtens + resultSet = Set.fromList result + writtenSet = Set.fromList written + length result `T.shouldBe` length written + (writtenSet `Set.difference` resultSet) `T.shouldBe` Set.empty + checkEachResult = TF.traverse checkEachItems + checkEachItems = L.foldl' checkItems $ Right Map.empty + checkItems (Right mp) (thnum, num) + | Map.lookup thnum mp < Just num = Right $ Map.insert thnum num mp + | Map.lookup thnum mp > Just num = Left "broken order" + | otherwise = Left "duplicated value" + checkItems err _ = err + readItem refItems refCount q = do + r <- KQ.readQueueWithoutMask q + Ref.modifyIORef refCount (1+) + Ref.modifyIORef refItems (r:) + tryReadItem refItems refCount q = do + mr <- KQ.tryReadQueueWithoutMask q + case mr of + Just r -> do + Ref.modifyIORef refCount (1+) + Ref.modifyIORef refItems (r:) + Nothing -> do + CC.yield + tryReadItem refItems refCount q + readItemOne refItems refCount q = E.mask_ $ do + select <- Q.generate Q.arbitrary + if select + then readItem refItems refCount q + else tryReadItem refItems refCount q + readItems refItems refCount q size restore !c = do + M.void . T.ignoreException . restore $ readItemOne refItems refCount q + count <- Ref.readIORef refCount + M.when (count < size && c < size * 100) $ + readItems refItems refCount q size restore $ c + 1 + readConcurrent q (thsize, itemsize) = do + ass <- E.mask $ \ restore -> + M.replicateM thsize . AS.async $ do + refItems <- Ref.newIORef [] + refCount <- Ref.newIORef 0 + readItems refItems refCount q itemsize restore (0 :: Int) + reverse <$> Ref.readIORef refItems + M.void . AS.async $ T.throwExceptionRandomly ass + TF.for ass AS.wait + writeItem refItems q = E.mask_ $ do + items <- Ref.readIORef refItems + case items of + [] -> return Nothing + v:next -> do + KQ.writeQueueWithoutMask q v + Ref.writeIORef refItems next + return $ Just v + writeItems refItems q restore !c = do + mmwritten <- T.ignoreException . restore $ writeItem refItems q + case mmwritten of + Just Nothing -> return () + _ -> writeItems refItems q restore $ c + 1 + writeConcurrent q (thsize, itemsize) = do + ass <- E.mask $ \ restore -> + TF.for [1..thsize] $ \ thnum -> AS.async $ do + let items = fmap (thnum,) [1..itemsize] :: [(Int, Int)] + refItems <- Ref.newIORef items + writeItems refItems q restore (0 :: Int) + return items + M.void . AS.async $ T.throwExceptionRandomly ass + TF.for ass AS.wait + +spec :: HS.Spec +spec = HS.describe "KazuraQueue concurrent specs" $ do + writeQueueSpec + readQueueSpec + tryReadQueueSpec + readWriteQueueSpec + tryReadWriteQueueSpec + readQueueWithExceptionSpec + diff --git a/test/KazuraQueueSpec.hs b/test/KazuraQueueSpec.hs new file mode 100644 --- /dev/null +++ b/test/KazuraQueueSpec.hs @@ -0,0 +1,114 @@ +{-# LANGUAGE ScopedTypeVariables #-} + +module KazuraQueueSpec where + +import qualified Test.Expectations as T +import qualified Test.KazuraQueue as T + +import qualified Test.Hspec as HS +import qualified Test.QuickCheck as Q + +import qualified Control.Concurrent.KazuraQueue as KQ +import qualified Control.Monad as M + +import qualified Data.IORef as Ref + +import qualified System.Mem.Weak as Weak + +writeQueueSpec :: HS.Spec +writeQueueSpec = HS.describe "writeQueue" $ do + T.whenQueueIsEmpty $ \ prepare -> do + HS.it "write the value without blocking" . prepare $ \ q -> do + v :: Int <- Q.generate Q.arbitrary + KQ.lengthQueue q `T.shouldReturn` 0 + KQ.writeQueue q v `T.shouldNotBlock` 500000 + q `T.queueLengthShouldBeIn` (0, 1) + T.whenItemsInQueue (1,10) $ \ prepare -> do + HS.it "write the value without blocking" . prepare $ \ (q, pre) -> do + let len0 = length pre + KQ.lengthQueue q `T.shouldReturn` len0 + v :: Int <- Q.generate Q.arbitrary + KQ.writeQueue q v `T.shouldNotBlock` 500000 + q `T.queueLengthShouldBeIn` (len0, len0 + 1) + +readQueueSpec :: HS.Spec +readQueueSpec = HS.describe "readQueue" $ do + T.whenQueueIsEmpty $ \ prepare -> do + HS.it "blocks until some one writes item" . prepare $ \ q -> do + wait <- KQ.readQueue q `T.shouldBlock` 500000 + q `T.queueLengthShouldBeIn` (-1, 0) + val :: Int <- Q.generate Q.arbitrary + KQ.writeQueue q val `T.shouldNotBlock` 500000 + r <- wait `T.shouldAwakeFinish` 500000 + r `T.shouldBe` val + q `T.queueLengthShouldBeIn` (-1, 0) + HS.it "block and awake out of order (values are in order)" . prepare $ \ q -> do + waits0 <- M.replicateM 2 $ KQ.readQueue q `T.shouldBlock` 500000 + q `T.queueLengthShouldBeIn` (-2, 0) + (val1 :: Int, val2) <- Q.generate Q.arbitrary + + KQ.writeQueue q val1 `T.shouldNotBlock` 500000 + (r1, waits1) <- waits0 `T.onlyOneShouldAwakeFinish` 500000 + q `T.queueLengthShouldBeIn` (-2, 0) + + KQ.writeQueue q val2 `T.shouldNotBlock` 500000 + (r2, _) <- waits1 `T.onlyOneShouldAwakeFinish` 500000 + q `T.queueLengthShouldBeIn` (-2, 0) + + (r1, r2) `T.shouldBe` (val1, val2) + T.whenQueueIsEmpty $ \ prepare -> do + HS.it "the item in a Queue is not evaluated by write/read" . prepare $ \ q -> do + KQ.writeQueue q ([1..] :: [Int]) `T.shouldNotBlock` 500000 + M.void $ KQ.readQueue q `T.shouldNotBlock` 500000 + T.whenQueueIsEmpty $ \ prepare -> do + HS.it "the item in a Queue can be garbage collected after read" . prepare $ \ q -> do + ref <- Ref.newIORef True + weak <- Weak.mkWeakPtr ref Nothing + KQ.writeQueue q ref `T.shouldNotBlock` 500000 + T.shouldNotBeGarbageCollected weak + M.void $ KQ.readQueue q `T.shouldNotBlock` 500000 + T.shouldBeGarbageCollected weak + T.whenItemsInQueue (1,10) $ \ prepare -> do + HS.it "read one value without blocking" . prepare $ \ (q, pre) -> do + r :: Int <- KQ.readQueue q `T.shouldNotBlock` 500000 + r `T.shouldBe` head pre + +tryReadQueueSpec :: HS.Spec +tryReadQueueSpec = HS.describe "tryReadQueue" $ do + T.whenQueueIsEmpty $ \ prepare -> do + HS.it "immediately returns without reading value" . prepare $ \ q -> do + mret1 <- KQ.tryReadQueue q `T.shouldNotBlock` 500000 + mret1 `T.shouldBe` Nothing + q `T.queueLengthShouldBeIn` (-1, 0) + wait <- KQ.readQueue q `T.shouldBlock` 500000 + q `T.queueLengthShouldBeIn` (-2, 0) + mret2 <- KQ.tryReadQueue q `T.shouldNotBlock` 500000 + mret2 `T.shouldBe` Nothing + q `T.queueLengthShouldBeIn` (-3, 0) + val :: Int <- Q.generate Q.arbitrary + KQ.writeQueue q val `T.shouldNotBlock` 500000 + r <- wait `T.shouldAwakeFinish` 500000 + r `T.shouldBe` val + q `T.queueLengthShouldBeIn` (-3, 0) + mret3 <- KQ.tryReadQueue q `T.shouldNotBlock` 500000 + mret3 `T.shouldBe` Nothing + q `T.queueLengthShouldBeIn` (-4, 0) + HS.it "read value after writing" . prepare $ \ q -> do + (val1 :: Int, val2) <- Q.generate Q.arbitrary + KQ.writeQueue q val1 `T.shouldNotBlock` 500000 + mret1 <- KQ.tryReadQueue q `T.shouldNotBlock` 500000 + mret1 `T.shouldBe` Just val1 + KQ.writeQueue q val2 `T.shouldNotBlock` 500000 + mret2 <- KQ.tryReadQueue q `T.shouldNotBlock` 500000 + mret2 `T.shouldBe` Just val2 + T.whenItemsInQueue (1,10) $ \ prepare -> do + HS.it "read one value without blocking" . prepare $ \ (q, pre) -> do + r :: Maybe Int <- KQ.tryReadQueue q `T.shouldNotBlock` 500000 + r `T.shouldBe` Just (head pre) + +spec :: HS.Spec +spec = HS.describe "KazuraQueue basic specs" $ do + writeQueueSpec + readQueueSpec + tryReadQueueSpec + diff --git a/test/Spec.hs b/test/Spec.hs new file mode 100644 --- /dev/null +++ b/test/Spec.hs @@ -0,0 +1,14 @@ + +import qualified KazuraQueueConcurrentSpec as KQCSpec +import qualified KazuraQueueSpec as KQSpec +import qualified Test.Hspec as HS +import qualified WVarConcurrentSpec as WVCSpec +import qualified WVarSpec as WVSpec + +main :: IO () +main = HS.hspec $ do + WVSpec.spec + WVCSpec.spec + KQSpec.spec + KQCSpec.spec + diff --git a/test/Test/Concurrent.hs b/test/Test/Concurrent.hs new file mode 100644 --- /dev/null +++ b/test/Test/Concurrent.hs @@ -0,0 +1,147 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE ScopedTypeVariables #-} + +module Test.Concurrent where + +import qualified Test.QuickCheck as Q + +import qualified Control.Concurrent as CC +import qualified Control.Concurrent.Async as AS +import qualified Control.Concurrent.MVar as MV +import qualified Control.Exception as E +import Control.Monad ((>=>)) +import qualified Control.Monad as M +import qualified GHC.Conc as CC + +import qualified Data.Maybe as MB +import qualified Data.Traversable as TF +import Data.Typeable (Typeable) + +class HasThread th where + threadId :: th -> IO CC.ThreadId + throwTo :: E.Exception e => th -> e -> IO () + throwTo th e = threadId th >>= flip E.throwTo e + +threadStatus :: HasThread th => th -> IO CC.ThreadStatus +threadStatus = threadId >=> CC.threadStatus + +instance HasThread CC.ThreadId where + threadId = return + +instance HasThread (AS.Async x) where + threadId = return . AS.asyncThreadId + +isFinish :: CC.ThreadStatus -> Bool +isFinish CC.ThreadFinished = True +isFinish CC.ThreadDied = True +isFinish _ = False + +isStop :: CC.ThreadStatus -> Bool +isStop CC.ThreadRunning = False +isStop _ = True + +withWaitStart :: (IO () -> IO x) -> IO x +withWaitStart actf = do + mv <- MV.newEmptyMVar + mdelay <- Q.generate $ arbitraryDelay 20000 + async <- AS.async . actf $ MV.readMVar mv + case mdelay of + Just delay -> CC.threadDelay delay + Nothing -> return () + CC.putMVar mv () + AS.wait async + +concurrently :: IO a -> IO b -> IO (a, b) +concurrently act1 act2 = do + mdelay1 <- Q.generate $ arbitraryDelay 20000 + mdelay2 <- Q.generate $ arbitraryDelay 20000 + withWaitStart $ \ wait -> + wrap wait mdelay1 act1 `AS.concurrently` wrap wait mdelay2 act2 + where + wrap :: IO () -> Maybe Int -> IO a -> IO a + wrap wait mdelay act = wait >> TF.for mdelay CC.threadDelay >> act + +mapConcurrently :: [IO a] -> IO [a] +mapConcurrently acts = do + let len = length acts + mds <- Q.generate . Q.vectorOf len $ fmap (`mod` 20000) <$> Q.arbitrary + withWaitStart $ \ wait -> do + AS.mapConcurrently id $ wrap wait <$> zip mds acts + where + wrap :: IO () -> (Maybe Int, IO a) -> IO a + wrap wait (mdelay, act) = wait >> TF.for mdelay CC.threadDelay >> act + +mapConcurrently_ :: [IO a] -> IO () +mapConcurrently_ = M.void . mapConcurrently + +waitStop :: HasThread th => th -> IO CC.ThreadStatus +waitStop th = snd . head <$> waitAny isStop [th] + +waitFinish :: HasThread th => th -> IO CC.ThreadStatus +waitFinish th = snd . head <$> waitFinishAny [th] + +waitFinishAny :: HasThread th => [th] -> IO [(Int, CC.ThreadStatus)] +waitFinishAny = waitAny isFinish + +waitAny :: HasThread th => + (CC.ThreadStatus -> Bool) -> [th] -> IO [(Int, CC.ThreadStatus)] +waitAny = waitAnyAtLeast 1 + +waitAnyAtLeast :: HasThread th => + Int -> (CC.ThreadStatus -> Bool) -> [th] -> IO [(Int, CC.ThreadStatus)] +waitAnyAtLeast num f ths = go + where + go = do + statuses <- M.sequence $ threadStatus <$> ths + let satisfied = filter (f . snd) $ zip [0..] statuses + if length satisfied >= num + then return satisfied + else CC.threadDelay 1 >> go + +data RandomException = RandomException Int String + deriving (Show, Typeable) +instance E.Exception RandomException + +ignoreException :: IO a -> IO (Maybe a) +ignoreException act = (Just <$> act) + `E.catch` \ (_err :: RandomException) -> do +-- E.uninterruptibleMask_ $ putStrLn $ "---- Exception throwed : " ++ show _err + return Nothing + +ignoreException_ :: IO a -> IO () +ignoreException_ = M.void . ignoreException + +runningThreadId :: HasThread th => th -> IO (Maybe CC.ThreadId) +runningThreadId th = do + status <- threadStatus th + if isFinish status + then return Nothing + else Just <$> threadId th + +throwExceptionRandomly :: HasThread th => [th] -> IO () +throwExceptionRandomly ths = go (1 :: Int) + where + getAlives = fmap MB.catMaybes . M.sequence $ runningThreadId <$> ths + go !c = do + mdelay <- Q.generate $ arbitraryDelay $ 20000 * c + case mdelay of + Just delay -> CC.threadDelay delay + Nothing -> return () + alives <- getAlives + if length alives == 0 + then return () + else do + alive <- Q.generate $ Q.elements alives + throwTo alive . RandomException c $ show mdelay ++ " : " ++ show (length alives) + go $ c+1 + +arbitraryDelay :: Int -> Q.Gen (Maybe Int) +arbitraryDelay limit = do + mbase <- Q.arbitrary + multi1 <- (+1) . abs <$> Q.arbitrary + multi2 <- (+1) . abs <$> Q.arbitrary + case mbase of + Just base -> return . Just . (`mod` limit) $ base * multi1 * multi2 + Nothing -> return Nothing + + diff --git a/test/Test/Expectations.hs b/test/Test/Expectations.hs new file mode 100644 --- /dev/null +++ b/test/Test/Expectations.hs @@ -0,0 +1,244 @@ +{-# LANGUAGE ScopedTypeVariables #-} + +module Test.Expectations + ( module Test.Expectations + , module EX + ) where + +import Test.Hspec.Expectations as EX hiding (shouldBe, shouldContain, + shouldNotBe, shouldReturn, shouldSatisfy, + shouldThrow) + +import qualified Test.Concurrent as T +import qualified Test.Util as T + +import qualified Test.HUnit as HU + +import qualified Control.Concurrent.Async as AS +import qualified Control.Exception as E +import qualified Control.Monad as M + +import qualified GHC.Conc as CC + +import qualified Data.List as L +import qualified Data.Maybe as MB +import Data.Monoid ((<>)) +import qualified Data.Typeable as TP + +import qualified System.Mem as Mem +import qualified System.Mem.Weak as Weak +import qualified System.Timeout as ST + +import Prelude hiding (and, fail, or) + +assertFailure :: String -> IO x +assertFailure desc = do + HU.assertFailure desc + error "dummy" + +failWithException :: (String -> String) -> IO x -> IO y +failWithException descf wait = proc `E.catch` returnError + where + returnError (E.SomeException err) = do + assertFailure . descf $ "but aborted with: " <> show err + proc = do + M.void wait + assertFailure $ descf "died with no exception (test maybe wrong)" + +-------------------- +-- assertions + +assertTrue :: Bool -> String -> IO () +assertTrue True _ = return () +assertTrue False desc = HU.assertFailure desc + +assertEqual :: (Show x, Eq x) => x -> x -> IO () +assertEqual expected actual = assertTrue (expected == actual) desc + where + desc = mkDesc 80 "expected" expectedStr <> "\n" <> + mkDesc 80 "actual " actualStr + expectedStr = show expected + actualStr = show actual + +assertNotEqual :: (Show x, Eq x) => x -> x -> IO () +assertNotEqual expected actual = assertTrue (expected /= actual) desc + where + desc = mkDesc 80 "expected not to be" expectedStr <> "\n" <> + mkDesc 80 "actual" actualStr + expectedStr = show expected + actualStr = show actual + +shouldBe :: (Show x, Eq x) => x -> x -> IO () +shouldBe = flip assertEqual + +shouldNotBe :: (Show x, Eq x) => x -> x -> IO () +shouldNotBe = flip assertNotEqual + +shouldSatisfy :: Show x => x -> (x -> Bool) -> IO () +shouldSatisfy x f = assertTrue (f x) desc + where + desc = "assertion failed on: " ++ show x + +shouldContain :: (Show x, Eq x) => [x] -> [x] -> IO () +shouldContain actual expected = assertTrue result $ desc + where + result = expected `L.isInfixOf` actual + desc = mkDesc 80 "expected to contain" expectedStr <> "\n" <> + mkDesc 80 "but actual" actualStr + expectedStr = show expected + actualStr = show actual + +shouldReturn :: (Show x, Eq x) => IO x -> x -> IO () +shouldReturn act expected = do + ex <- E.try act + case ex of + Right x -> x `shouldBe` expected + Left (E.SomeException err) -> HU.assertFailure $ desc err + where + desc err = mkDesc 80 "expected" expectedStr <> "\n" <> + mkDesc 80 "but exception throwed" (show err) + expectedStr = show expected + +shouldThrow :: (Show x, E.Exception e) => IO x -> Selector e -> IO () +shouldThrow act selector = do + ex <- E.try act + case ex of + Right x -> HU.assertFailure $ descX x + Left (err :: E.SomeException) -> case E.fromException err of + Just e | selector e -> return () + | otherwise -> HU.assertFailure $ descF err + Nothing -> HU.assertFailure $ descE err + where + descX x = expectedStr <> mkDesc 80 "but returned" (show x) + descF e = expectedStr <> mkDesc 80 "but selector failed for" (show e) + descE e = expectedStr <> mkDesc 80 "but exception throwed" (show e) + expectedStr = mkDesc 80 "expected to throw" exceptedType <> "\n" + exceptedType = (show . TP.typeOf . instanceOf) selector + instanceOf :: Selector a -> a + instanceOf _ = error "dummy data of shouldThrow" + +-- gc expectations + +shouldBeGarbageCollected :: Weak.Weak x -> IO () +shouldBeGarbageCollected weak = do + Mem.performGC + mref <- Weak.deRefWeak weak + case mref of + Just _ -> assertFailure "expected to be garbage collected but does not" + Nothing -> return () + +shouldNotBeGarbageCollected :: Weak.Weak x -> IO () +shouldNotBeGarbageCollected weak = do + Mem.performGC + mv <- Weak.deRefWeak weak + case mv of + Just _ -> return () + Nothing -> assertFailure "expected not to be garbage collected but garbage collected" + +-- concurrent expectations + +shouldBlock :: IO x -> Int -> IO (AS.Async x) +shouldBlock act time = do + async <- AS.async act + mstatus <- ST.timeout time $ T.waitStop async + case mstatus of + Just CC.ThreadFinished -> HU.assertFailure $ desc "but finished" + Just CC.ThreadDied -> failWithException desc $ AS.wait async + Just (CC.ThreadBlocked _) -> return () + _ -> HU.assertFailure $ desc "but still running" + return async + where + desc str = "expected to block in " <> show time <> " nanosec\n" <> str + +shouldStillBlock :: AS.Async x -> Int -> IO () +shouldStillBlock async time = M.void $ AS.wait async `shouldBlock` time + +shouldAwakeFinish :: AS.Async x -> Int -> IO x +shouldAwakeFinish async time = do + mstatus <- ST.timeout time $ T.waitFinish async + status <- MB.fromMaybe (T.threadStatus async) $ return <$> mstatus + let wait = AS.wait async + case status of + CC.ThreadFinished -> return () + CC.ThreadDied -> failWithException desc $ AS.wait async + CC.ThreadBlocked reason -> + HU.assertFailure . desc $ "but still blocked with " <> show reason + CC.ThreadRunning -> + HU.assertFailure . desc $ "but still running" + wait + where + desc str = "expected to awake and finish in " + <> show time <> " nanosec\n" <> str + +onlyOneShouldAwakeFinish :: [AS.Async x] -> Int -> IO (x, [AS.Async x]) +onlyOneShouldAwakeFinish asyncs0 time = do + mret <- ST.timeout time $ T.waitFinishAny asyncs0 + (idx, status) <- case mret of + Just [ret] -> return ret + Just _ -> assertFailure . desc $ "but not only one finished" + Nothing -> assertFailure . desc $ "but still running" + let (async, asyncs1) = T.pickUp idx asyncs0 + wait = AS.wait async + x <- case status of + CC.ThreadFinished -> wait + CC.ThreadDied -> failWithException desc wait + _ -> assertFailure . desc $ "unknown case" + return (x, asyncs1) + where + desc str = "expected to awake only one and finish in " + <> show time <> " nanosec\n" <> str + +shouldNotBlock :: IO x -> Int -> IO x +shouldNotBlock act time = do + async <- AS.async act + mstatus <- ST.timeout time $ T.waitStop async + let wait = AS.wait async + case mstatus of + Just CC.ThreadFinished -> return () + Just CC.ThreadDied -> failWithException desc wait + Just (CC.ThreadBlocked reason) -> do + HU.assertFailure . desc $ "but blocked with " <> show reason + _ -> HU.assertFailure $ desc "but still running" + wait + where + desc str = "expected not to block and finish in " <> show time <> + " nanosec\n" <> str + +shouldFinish :: IO x -> Int -> IO x +shouldFinish act time = do + async <- AS.async act + mstatus <- ST.timeout time $ T.waitFinish async + let wait = AS.wait async + case mstatus of + Just CC.ThreadFinished -> return () + Just CC.ThreadDied -> failWithException desc wait + Just (CC.ThreadBlocked reason) -> do + HU.assertFailure . desc $ "but blocked with " <> show reason + _ -> HU.assertFailure $ desc "but still running" + wait + where + desc str = "expected to finish in " <> show time <> + " nanosec\n" <> str + +--------------------------- +--- util + +mkDesc :: Int -> String -> String -> String +mkDesc len s1 s2 + | length s1 + length str > len = s1 <> ":\n\t" <> str + | otherwise = s1 <> ": " <> str + where + str = truncateString 512 s2 + +truncateString :: Int -> String -> String +truncateString len str + | null rest = res + | otherwise = res <> "..." + where + (res, rest) = L.splitAt len str + +and :: [a -> Bool] -> (a -> Bool) +and (c:cs) a | c a = and cs a + | otherwise = False +and [] _ = True + diff --git a/test/Test/KazuraQueue.hs b/test/Test/KazuraQueue.hs new file mode 100644 --- /dev/null +++ b/test/Test/KazuraQueue.hs @@ -0,0 +1,64 @@ + +module Test.KazuraQueue where + +import qualified Test.Expectations as T + +import qualified Test.Hspec as HS +import qualified Test.QuickCheck as Q + +import qualified Control.Concurrent.KazuraQueue as KQ + +import qualified Control.Concurrent as CC +import qualified Control.Concurrent.Async as AS +import qualified Control.Exception as E +import qualified Control.Monad as M + +import qualified Data.Foldable as TF + +import qualified System.Timeout as ST + +timeout :: IO r -> IO r +timeout act = do + mr <- ST.timeout (10 * 1000000) act + case mr of + Just r -> return r + Nothing -> T.assertFailure "timeout 10sec" + +whenQueueIsEmpty :: (((KQ.Queue x -> IO r) -> IO r) -> HS.Spec) -> HS.Spec +whenQueueIsEmpty f = HS.describe "when Queue is empty" $ f prepare + where + prepare :: (KQ.Queue x -> IO r) -> IO r + prepare iof = do + q <- KQ.newQueue + timeout . prependIndefiniteBlock q $ iof q + +whenItemsInQueue :: Q.Arbitrary x => + (Int, Int) -> ((((KQ.Queue x, [x]) -> IO r) -> IO r) -> HS.Spec) -> HS.Spec +whenItemsInQueue range f = HS.describe "when some items in Queue" $ f prepare + where + (minSize, maxSize) = range + len = maxSize - minSize + 1 + prepare :: Q.Arbitrary x => ((KQ.Queue x, [x]) -> IO r) -> IO r + prepare iof = do + num <- (+ minSize) . (`mod` len) . abs <$> Q.generate Q.arbitrary + vals <- M.replicateM num $ Q.generate Q.arbitrary + queue <- KQ.newQueue + TF.for_ vals $ KQ.writeQueue queue + timeout . prependIndefiniteBlock queue $ iof (queue, vals) + +prependIndefiniteBlock :: KQ.Queue x -> IO r -> IO r +prependIndefiniteBlock queue io = do + async <- AS.async $ do + CC.threadDelay $ 50 * 1000000 -- 50 sec + KQ.writeQueue queue $ error "indefinitly blocked" + io `E.finally` AS.cancel async + +--------------- expectations + +queueLengthShouldBeIn :: KQ.Queue x -> (Int, Int) -> IO () +queueLengthShouldBeIn q (minv, maxv) = do + len <- KQ.lengthQueue q `T.shouldNotBlock` 500000 + len `T.shouldSatisfy` T.and [(>= minv), (<= maxv)] + alen <- KQ.lengthQueue' q `T.shouldNotBlock` 500000 + alen `T.shouldSatisfy` T.and [(>= max minv 0), (<= max maxv 0)] + diff --git a/test/Test/Util.hs b/test/Test/Util.hs new file mode 100644 --- /dev/null +++ b/test/Test/Util.hs @@ -0,0 +1,38 @@ +{-# LANGUAGE TupleSections #-} + +module Test.Util where + +import qualified Test.QuickCheck as Q + +import qualified Control.Exception as E +import qualified Control.Monad as M + +import qualified Data.List as L + +orElse :: IO x -> IO x -> IO x +orElse io1 io2 = io1 `E.catch` \ (E.SomeException _) -> io2 + +oneOf :: [IO x] -> IO x +oneOf (io:[]) = io +oneOf (io:ios) = io `orElse` oneOf ios +oneOf [] = error "actions must include at least one element" + +oneOfWithIndex :: [IO x] -> IO (x, Int) +oneOfWithIndex = go 0 + where + go _ [] = error "actions must include at least one element" + go x (io:[]) = (, x) <$> io + go x (io:ios) = ((, x) <$> io) + `E.catch` \ (E.SomeException _) -> go (x+1) ios + +genSatisfy :: Q.Arbitrary a => Int -> (a -> Bool) -> IO [a] +genSatisfy num f = M.replicateM num . Q.generate $ Q.arbitrary `Q.suchThat` f + +pickUp :: Int -> [x] -> (x, [x]) +pickUp idx xs = (rx, hxs ++ drop 1 txs) + where + (hxs, txs) = L.splitAt idx xs + rx = case take 1 txs of + r:_ -> r + _ -> error "invalid index to pickUp" + diff --git a/test/Test/WVar.hs b/test/Test/WVar.hs new file mode 100644 --- /dev/null +++ b/test/Test/WVar.hs @@ -0,0 +1,62 @@ + +module Test.WVar where + +import qualified Test.Hspec as HS +import qualified Test.QuickCheck as Q + +import qualified Control.Concurrent.WVar as WV + +import qualified Control.Monad as M + +withLatestCache :: + (IO (v, v, WV.WVar v, WV.WCached v) -> r) -> IO (v, WV.WVar v) -> r +withLatestCache f prepare = f prepare' + where + prepare' = do + (val, wv) <- prepare + wc <- WV.cacheWVar wv + return (val, val, wv, wc) + +whenWVarIsFresh :: + (IO (Int, WV.WVar Int) -> HS.Spec) -> HS.Spec +whenWVarIsFresh f = HS.describe "when WVar is fresh" $ f prepare + where + prepare = do + val <- Q.generate Q.arbitrary + wv <- WV.newWVar val + return (val, wv) + +whenWVarIsUpdating :: Q.Arbitrary v => + (IO (v, WV.WVar v) -> HS.Spec) -> HS.Spec +whenWVarIsUpdating f = HS.describe "when WVar is updating" $ f prepare + where + prepare = do + val <- Q.generate Q.arbitrary + wv <- WV.newWVar val + M.void $ WV.takeWVar wv + return (val, wv) + +whenWVarIsFreshButCacheStaled :: (Eq v, Q.Arbitrary v) => + (IO (v, v, WV.WVar v, WV.WCached v) -> HS.Spec) -> HS.Spec +whenWVarIsFreshButCacheStaled f = + HS.describe "when WVar is fresh but cache staled" $ f prepare + where + prepare = do + val1 <- Q.generate Q.arbitrary + val2 <- Q.generate $ Q.arbitrary `Q.suchThat` (/= val1) + wv <- WV.newWVar val1 + wc <- WV.cacheWVar wv + WV.putWVar wv val2 + return (val1, val2, wv, wc) + +whenWVarIsUpdatingAndCacheStaled :: (Eq v, Q.Arbitrary v) => + (IO (v, v, WV.WVar v, WV.WCached v) -> HS.Spec) -> HS.Spec +whenWVarIsUpdatingAndCacheStaled f = + HS.describe "when WVar is updating and cache staled" $ f prepare + where + prepare = do + val <- Q.generate Q.arbitrary + wv <- WV.newWVar val + wc <- WV.cacheWVar wv + M.void $ WV.takeWVar wv + return (val, val, wv, wc) diff --git a/test/WVarConcurrentSpec.hs b/test/WVarConcurrentSpec.hs new file mode 100644 --- /dev/null +++ b/test/WVarConcurrentSpec.hs @@ -0,0 +1,545 @@ +{-# LANGUAGE ScopedTypeVariables #-} + +module WVarConcurrentSpec where + +import qualified Test.Concurrent as T +import qualified Test.Expectations as T +import qualified Test.Util as T +import qualified Test.WVar as T + +import qualified Test.Hspec as HS +import qualified Test.Hspec.QuickCheck as HS + +import qualified Control.Concurrent.WVar as WV +import qualified Control.Monad as M + +import qualified Data.List as L + +takeWVarSeqSpec :: HS.Spec +takeWVarSeqSpec = HS.describe "takeWVar" $ do + T.whenWVarIsFresh $ \ prepare -> do + HS.prop "takes the value before or after putWVar" $ do + (val1, wv) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWVar wv val2 `T.concurrently` do + ret <- WV.takeWVar wv `T.shouldNotBlock` 500000 + T.oneOf + [ ret `T.shouldBe` val1 + , ret `T.shouldBe` val2 + ] + HS.prop "take different value" $ do + (val1, wv) <- prepare + [val2, val3, val4] <- T.genSatisfy 3 (/= val1) + ret <- T.mapConcurrently + [ WV.takeWVar wv <* WV.putWVar wv val2 + , WV.takeWVar wv <* WV.putWVar wv val3 + , WV.takeWVar wv <* WV.putWVar wv val4 + ] + T.oneOf + [ ret `T.shouldBe` [val1,val2,val3] + , ret `T.shouldBe` [val1,val4,val2] + , ret `T.shouldBe` [val3,val1,val2] + , ret `T.shouldBe` [val4,val1,val3] + , ret `T.shouldBe` [val3,val4,val1] + , ret `T.shouldBe` [val4,val2,val1] + ] + T.whenWVarIsUpdating $ \ prepare -> do + HS.prop "takes the value after putWVar" $ do + (val1, wv) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWVar wv val2 `T.concurrently` do + ret <- WV.takeWVar wv + ret `T.shouldBe` val2 + HS.prop "take different value" $ do + (val1 :: Int, wv) <- prepare + [val2, val3, val4, val5] <- T.genSatisfy 4 (/= val1) + ret <- T.mapConcurrently + [ WV.putWVar wv val2 >> return val2 + , WV.takeWVar wv <* WV.putWVar wv val3 + , WV.takeWVar wv <* WV.putWVar wv val4 + , WV.takeWVar wv <* WV.putWVar wv val5 + ] + T.oneOf + [ ret `T.shouldBe` [val2,val2,val3,val4] + , ret `T.shouldBe` [val2,val2,val5,val3] + , ret `T.shouldBe` [val2,val4,val2,val3] + , ret `T.shouldBe` [val2,val5,val2,val4] + , ret `T.shouldBe` [val2,val4,val5,val2] + , ret `T.shouldBe` [val2,val5,val3,val2] + ] + +tryTakeWVarSeqSpec :: HS.Spec +tryTakeWVarSeqSpec = HS.describe "tryTakeWVar" $ do + T.whenWVarIsFresh $ \ prepare -> do + HS.prop "takes the value before or after putWVar" $ do + (val1 :: Int, wv) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWVar wv val2 `T.concurrently` do + ret <- WV.tryTakeWVar wv `T.shouldNotBlock` 500000 + T.oneOf + [ ret `T.shouldBe` (True, val1) + , ret `T.shouldBe` (True, val2) + ] + HS.prop "all read same value but only one succeeded" $ do + (val :: Int, wv) <- prepare + ret <- T.mapConcurrently + [ WV.tryTakeWVar wv + , WV.tryTakeWVar wv + , WV.tryTakeWVar wv + ] + T.oneOf + [ ret `T.shouldBe` [(True,val),(False,val),(False,val)] + , ret `T.shouldBe` [(False,val),(True,val),(False,val)] + , ret `T.shouldBe` [(False,val),(False,val),(True,val)] + ] + T.whenWVarIsUpdating $ \ prepare -> do + HS.prop "takes the value before(failure) or after(success) putWVar" $ do + (val1 :: Int, wv) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWVar wv val2 `T.concurrently` do + ret <- WV.tryTakeWVar wv `T.shouldNotBlock` 500000 + T.oneOf + [ ret `T.shouldBe` (False, val1) + , ret `T.shouldBe` (True, val2) + ] + HS.prop "all read same value and fail" $ do + (val :: Int, wv) <- prepare + ret <- T.mapConcurrently + [ WV.tryTakeWVar wv + , WV.tryTakeWVar wv + , WV.tryTakeWVar wv + ] + ret `T.shouldBe` [(False,val),(False,val),(False,val)] + +readFreshWVarSeqSpec :: HS.Spec +readFreshWVarSeqSpec = HS.describe "readFreshWVar" $ do + T.whenWVarIsFresh $ \ prepare -> do + HS.prop "reads the value before or after putWVar" $ do + (val1 :: Int, wv) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWVar wv val2 `T.concurrently` do + ret <- WV.readFreshWVar wv `T.shouldNotBlock` 500000 + T.oneOf + [ ret `T.shouldBe` val1 + , ret `T.shouldBe` val2 + ] + HS.prop "read same value" $ do + (val :: Int, wv) <- prepare + ret <- T.mapConcurrently + [ WV.readFreshWVar wv + , WV.readFreshWVar wv + , WV.readFreshWVar wv + ] + ret `T.shouldBe` [val,val,val] + T.whenWVarIsUpdating $ \ prepare -> do + HS.prop "reads the value after putWVar" $ do + (val1 :: Int, wv) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWVar wv val2 `T.concurrently` do + ret <- WV.readFreshWVar wv + ret `T.shouldBe` val2 + HS.prop "read same value" $ do + (val1 :: Int, wv) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + ret <- T.mapConcurrently + [ WV.putWVar wv val2 >> return val2 + , WV.readFreshWVar wv + , WV.readFreshWVar wv + , WV.readFreshWVar wv + ] + ret `T.shouldBe` [val2,val2,val2,val2] + +tryReadFreshWVarSeqSpec :: HS.Spec +tryReadFreshWVarSeqSpec = HS.describe "tryReadWVar" $ do + T.whenWVarIsFresh $ \ prepare -> do + HS.prop "reads the old value" $ do + (val1 :: Int, wv) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWVar wv val2 `T.concurrently` do + ret <- WV.tryReadFreshWVar wv `T.shouldNotBlock` 500000 + T.oneOf + [ ret `T.shouldBe` (True, val1) + , ret `T.shouldBe` (True, val2) + ] + HS.prop "all read same value and succeed" $ do + (val :: Int, wv) <- prepare + ret <- T.mapConcurrently + [ WV.tryReadFreshWVar wv + , WV.tryReadFreshWVar wv + , WV.tryReadFreshWVar wv + ] + ret `T.shouldBe` [(True,val),(True,val),(True,val)] + T.whenWVarIsUpdating $ \ prepare -> do + HS.prop "reads the value before(failure) or after(success) putWVar" $ do + (val1 :: Int, wv) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWVar wv val2 `T.concurrently` do + ret <- WV.tryReadFreshWVar wv `T.shouldNotBlock` 500000 + T.oneOf + [ ret `T.shouldBe` (False, val1) + , ret `T.shouldBe` (True, val2) + ] + HS.prop "all read same value and fail" $ do + (val :: Int, wv) <- prepare + ret <- T.mapConcurrently + [ WV.tryReadFreshWVar wv + , WV.tryReadFreshWVar wv + , WV.tryReadFreshWVar wv + ] + ret `T.shouldBe` [(False,val),(False,val),(False,val)] + +takeWCachedSeqSpec :: HS.Spec +takeWCachedSeqSpec = HS.describe "takeWCached" $ do + T.whenWVarIsFresh . T.withLatestCache $ \ prepare -> do + HS.prop "takes the value before or after putWCached" $ do + (_, val1 :: Int, _, wc) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWCached wc val2 `T.concurrently` do + wt <- WV.takeWCached wc `T.shouldNotBlock` 500000 + T.oneOf + [ WV.readWTicket wt `T.shouldBe` val1 + , WV.readWTicket wt `T.shouldBe` val2 + ] + HS.prop "take different value" $ do + (_, val1 :: Int, _, wc) <- prepare + [val2, val3, val4] <- T.genSatisfy 3 (/= val1) + ret <- fmap WV.readWTicket <$> T.mapConcurrently + [ WV.takeWCached wc <* WV.putWCached wc val2 + , WV.takeWCached wc <* WV.putWCached wc val3 + , WV.takeWCached wc <* WV.putWCached wc val4 + ] + T.oneOf + [ ret `T.shouldBe` [val1,val2,val3] + , ret `T.shouldBe` [val1,val4,val2] + , ret `T.shouldBe` [val3,val1,val2] + , ret `T.shouldBe` [val4,val1,val3] + , ret `T.shouldBe` [val3,val4,val1] + , ret `T.shouldBe` [val4,val2,val1] + ] + T.whenWVarIsUpdating . T.withLatestCache $ \ prepare -> do + HS.prop "takes the value after putWCached" $ do + (_, val1 :: Int, _, wc) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWCached wc val2 `T.concurrently` do + ret <- WV.takeWCached wc + WV.readWTicket ret `T.shouldBe` val2 + HS.prop "take different value" $ do + (_, val1 :: Int, _, wc) <- prepare + [val2, val3, val4, val5] <- T.genSatisfy 4 (/= val1) + ret <- T.mapConcurrently + [ WV.putWCached wc val2 >> return val2 + , WV.readWTicket <$> WV.takeWCached wc <* WV.putWCached wc val3 + , WV.readWTicket <$> WV.takeWCached wc <* WV.putWCached wc val4 + , WV.readWTicket <$> WV.takeWCached wc <* WV.putWCached wc val5 + ] + T.oneOf + [ ret `T.shouldBe` [val2,val2,val3,val4] + , ret `T.shouldBe` [val2,val2,val5,val3] + , ret `T.shouldBe` [val2,val4,val2,val3] + , ret `T.shouldBe` [val2,val5,val2,val4] + , ret `T.shouldBe` [val2,val4,val5,val2] + , ret `T.shouldBe` [val2,val5,val3,val2] + ] + T.whenWVarIsFreshButCacheStaled $ \ prepare -> do + HS.prop "takes the value before or after putWCached" $ do + (_, val1 :: Int, _, wc) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWCached wc val2 `T.concurrently` do + wt <- WV.takeWCached wc `T.shouldNotBlock` 500000 + T.oneOf + [ WV.readWTicket wt `T.shouldBe` val1 + , WV.readWTicket wt `T.shouldBe` val2 + ] + HS.prop "take different value" $ do + (_, val1 :: Int, _, wc) <- prepare + [val2, val3, val4] <- T.genSatisfy 3 (/= val1) + ret <- fmap WV.readWTicket <$> T.mapConcurrently + [ WV.takeWCached wc <* WV.putWCached wc val2 + , WV.takeWCached wc <* WV.putWCached wc val3 + , WV.takeWCached wc <* WV.putWCached wc val4 + ] + T.oneOf + [ ret `T.shouldBe` [val1,val2,val3] + , ret `T.shouldBe` [val1,val4,val2] + , ret `T.shouldBe` [val3,val1,val2] + , ret `T.shouldBe` [val4,val1,val3] + , ret `T.shouldBe` [val3,val4,val1] + , ret `T.shouldBe` [val4,val2,val1] + ] + T.whenWVarIsUpdatingAndCacheStaled $ \ prepare -> do + HS.prop "takes the value after putWCached" $ do + (_, val1 :: Int, _, wc) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWCached wc val2 `T.concurrently` do + ret <- WV.takeWCached wc + WV.readWTicket ret `T.shouldBe` val2 + HS.prop "take different value" $ do + (_, val1 :: Int, _, wc) <- prepare + [val2, val3, val4, val5] <- T.genSatisfy 4 (/= val1) + ret <- T.mapConcurrently + [ WV.putWCached wc val2 >> return val2 + , WV.readWTicket <$> WV.takeWCached wc <* WV.putWCached wc val3 + , WV.readWTicket <$> WV.takeWCached wc <* WV.putWCached wc val4 + , WV.readWTicket <$> WV.takeWCached wc <* WV.putWCached wc val5 + ] + T.oneOf + [ ret `T.shouldBe` [val2,val2,val3,val4] + , ret `T.shouldBe` [val2,val2,val5,val3] + , ret `T.shouldBe` [val2,val4,val2,val3] + , ret `T.shouldBe` [val2,val5,val2,val4] + , ret `T.shouldBe` [val2,val4,val5,val2] + , ret `T.shouldBe` [val2,val5,val3,val2] + ] + +tryTakeWCachedSeqSpec :: HS.Spec +tryTakeWCachedSeqSpec = HS.describe "tryTakeWCached" $ do + T.whenWVarIsFresh . T.withLatestCache $ \ prepare -> do + HS.prop "takes the value before(failure)/after(success) putWCached" $ do + (_, val1 :: Int, _, wc) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWCached wc val2 `T.concurrently` do + (ret, wt) <- WV.tryTakeWCached wc `T.shouldNotBlock` 500000 + T.oneOf + [ (ret, WV.readWTicket wt) `T.shouldBe` (True, val1) + , (ret, WV.readWTicket wt) `T.shouldBe` (False, val2) + ] + HS.prop "all read same value but only one succeeded" $ do + (_, val1 :: Int, _, wc) <- prepare + ret <- fmap (fmap WV.readWTicket) <$> T.mapConcurrently + [ WV.tryTakeWCached wc + , WV.tryTakeWCached wc + , WV.tryTakeWCached wc + ] + T.oneOf + [ ret `T.shouldBe` [(True,val1),(False,val1),(False,val1)] + , ret `T.shouldBe` [(False,val1),(True,val1),(False,val1)] + , ret `T.shouldBe` [(False,val1),(False,val1),(True,val1)] + ] + T.whenWVarIsUpdating . T.withLatestCache $ \ prepare -> do + HS.prop "takes the value before or after putWCached with failure" $ do + (_, val1 :: Int, _, wc) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWCached wc val2 `T.concurrently` do + (ret, wt) <- WV.tryTakeWCached wc `T.shouldNotBlock` 500000 + T.oneOf + [ (ret, WV.readWTicket wt) `T.shouldBe` (False, val1) + , (ret, WV.readWTicket wt) `T.shouldBe` (False, val2) + ] + HS.prop "all read same value and fail" $ do + (_, val1 :: Int, _, wc) <- prepare + ret <- fmap (fmap WV.readWTicket) <$> T.mapConcurrently + [ WV.tryTakeWCached wc + , WV.tryTakeWCached wc + , WV.tryTakeWCached wc + ] + ret `T.shouldBe` [(False,val1),(False,val1),(False,val1)] + T.whenWVarIsFreshButCacheStaled $ \ prepare -> do + HS.prop "takes the value before or after putWCached with failure" $ do + (_, val1 :: Int, _, wc) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWCached wc val2 `T.concurrently` do + (ret, wt) <- WV.tryTakeWCached wc `T.shouldNotBlock` 500000 + T.oneOf + [ (ret, WV.readWTicket wt) `T.shouldBe` (False, val1) + , (ret, WV.readWTicket wt) `T.shouldBe` (False, val2) + ] + HS.prop "all read same value and fail" $ do + (_, val1 :: Int, _, wc) <- prepare + ret <- fmap (fmap WV.readWTicket) <$> T.mapConcurrently + [ WV.tryTakeWCached wc + , WV.tryTakeWCached wc + , WV.tryTakeWCached wc + ] + ret `T.shouldBe` [(False,val1),(False,val1),(False,val1)] + T.whenWVarIsUpdatingAndCacheStaled $ \ prepare -> do + HS.prop "takes the value before or after putWCached with failure" $ do + (_, val1 :: Int, _, wc) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWCached wc val2 `T.concurrently` do + (ret, wt) <- WV.tryTakeWCached wc `T.shouldNotBlock` 500000 + T.oneOf + [ (ret, WV.readWTicket wt) `T.shouldBe` (False, val1) + , (ret, WV.readWTicket wt) `T.shouldBe` (False, val2) + ] + HS.prop "all read same value and fail" $ do + (_, val1 :: Int, _, wc) <- prepare + ret <- fmap (fmap WV.readWTicket) <$> T.mapConcurrently + [ WV.tryTakeWCached wc + , WV.tryTakeWCached wc + , WV.tryTakeWCached wc + ] + ret `T.shouldBe` [(False,val1),(False,val1),(False,val1)] + +readFreshWCachedSeqSpec :: HS.Spec +readFreshWCachedSeqSpec = HS.describe "readFreshWCached" $ do + T.whenWVarIsFresh . T.withLatestCache $ \ prepare -> do + HS.prop "reads the value before or after putWCached" $ do + (val1 :: Int, _, _, wc) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWCached wc val2 `T.concurrently` do + wt <- WV.readFreshWCached wc `T.shouldNotBlock` 500000 + WV.readWTicket wt `T.shouldBe` val1 + HS.prop "read same value" $ do + (val1 :: Int, _, _, wc) <- prepare + ret <- fmap WV.readWTicket <$> T.mapConcurrently + [ WV.readFreshWCached wc + , WV.readFreshWCached wc + , WV.readFreshWCached wc + ] + ret `T.shouldBe` [val1,val1,val1] + T.whenWVarIsUpdating . T.withLatestCache $ \ prepare -> do + HS.prop "reads the value after putWCached" $ do + (val1 :: Int, _, _, wc) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWCached wc val2 `T.concurrently` do + wt <- WV.readFreshWCached wc + WV.readWTicket wt `T.shouldBe` val2 + HS.prop "read same value" $ do + (val1 :: Int, _, _, wc) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + ret <- T.mapConcurrently + [ WV.putWCached wc val2 >> return val2 + , WV.readWTicket <$> WV.readFreshWCached wc + , WV.readWTicket <$> WV.readFreshWCached wc + , WV.readWTicket <$> WV.readFreshWCached wc + ] + ret `T.shouldBe` [val2,val2,val2,val2] + T.whenWVarIsFreshButCacheStaled $ \ prepare -> do + HS.prop "reads the value before or after putWCached" $ do + (val1 :: Int, _, _, wc) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWCached wc val2 `T.concurrently` do + wt <- WV.readFreshWCached wc `T.shouldNotBlock` 500000 + WV.readWTicket wt `T.shouldBe` val1 + HS.prop "read same value" $ do + (val1 :: Int, _, _, wc) <- prepare + ret <- fmap WV.readWTicket <$> T.mapConcurrently + [ WV.readFreshWCached wc + , WV.readFreshWCached wc + , WV.readFreshWCached wc + ] + ret `T.shouldBe` [val1,val1,val1] + T.whenWVarIsUpdatingAndCacheStaled $ \ prepare -> do + HS.prop "reads the value after putWCached" $ do + (val1 :: Int, _, _, wc) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWCached wc val2 `T.concurrently` do + ret <- WV.readWTicket <$> WV.readFreshWCached wc + ret `T.shouldBe` val1 + HS.prop "read same value" $ do + (val1 :: Int, _, _, wc) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + ret <- T.mapConcurrently + [ WV.putWCached wc val2 >> return val2 + , WV.readWTicket <$> WV.readFreshWCached wc + , WV.readWTicket <$> WV.readFreshWCached wc + , WV.readWTicket <$> WV.readFreshWCached wc + ] + ret `T.shouldBe` [val2,val1,val1,val1] + +tryReadFreshWCachedSeqSpec :: HS.Spec +tryReadFreshWCachedSeqSpec = HS.describe "tryReadWCached" $ do + T.whenWVarIsFresh . T.withLatestCache $ \ prepare -> do + HS.prop "reads the old value and succeed" $ do + (val1 :: Int, _, _, wc) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWCached wc val2 `T.concurrently` do + ret <- WV.tryReadFreshWCached wc `T.shouldNotBlock` 500000 + fmap WV.readWTicket ret `T.shouldBe` (True, val1) + HS.prop "all read same value and succeed" $ do + (val1 :: Int, _, _, wc) <- prepare + ret <- fmap (fmap WV.readWTicket) <$> T.mapConcurrently + [ WV.tryReadFreshWCached wc + , WV.tryReadFreshWCached wc + , WV.tryReadFreshWCached wc + ] + ret `T.shouldBe` [(True,val1),(True,val1),(True,val1)] + T.whenWVarIsUpdating . T.withLatestCache $ \ prepare -> do + HS.prop "reads the old value and fail" $ do + (val1 :: Int, _, _, wc) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWCached wc val2 `T.concurrently` do + ret <- WV.tryReadFreshWCached wc `T.shouldNotBlock` 500000 + fmap WV.readWTicket ret `T.shouldBe` (False, val1) + HS.prop "all read same value and fail" $ do + (val1 :: Int, _, _, wc) <- prepare + ret <- fmap (fmap WV.readWTicket) <$> T.mapConcurrently + [ WV.tryReadFreshWCached wc + , WV.tryReadFreshWCached wc + , WV.tryReadFreshWCached wc + ] + ret `T.shouldBe` [(False,val1),(False,val1),(False,val1)] + T.whenWVarIsFreshButCacheStaled $ \ prepare -> do + HS.prop "reads the old value and succeed" $ do + (val1 :: Int, _, _, wc) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWCached wc val2 `T.concurrently` do + ret <- WV.tryReadFreshWCached wc `T.shouldNotBlock` 500000 + fmap WV.readWTicket ret `T.shouldBe` (True, val1) + HS.prop "all read same value and succeed" $ do + (val1 :: Int, _, _, wc) <- prepare + ret <- fmap (fmap WV.readWTicket) <$> T.mapConcurrently + [ WV.tryReadFreshWCached wc + , WV.tryReadFreshWCached wc + , WV.tryReadFreshWCached wc + ] + ret `T.shouldBe` [(True,val1),(True,val1),(True,val1)] + T.whenWVarIsUpdating . T.withLatestCache $ \ prepare -> do + HS.prop "reads the old value and fail" $ do + (val1 :: Int, _, _, wc) <- prepare + [val2] <- T.genSatisfy 1 (/= val1) + M.void $ WV.putWCached wc val2 `T.concurrently` do + ret <- WV.tryReadFreshWCached wc `T.shouldNotBlock` 500000 + fmap WV.readWTicket ret `T.shouldBe` (False, val1) + HS.prop "all read same value and fail" $ do + (val1 :: Int, _, _, wc) <- prepare + ret <- fmap (fmap WV.readWTicket) <$> T.mapConcurrently + [ WV.tryReadFreshWCached wc + , WV.tryReadFreshWCached wc + , WV.tryReadFreshWCached wc + ] + ret `T.shouldBe` [(False,val1),(False,val1),(False,val1)] + +tryTakeAndPutCachedSeqSpec :: HS.Spec +tryTakeAndPutCachedSeqSpec = + HS.prop "tryTakeWCached and putWCached perform atomic modification" $ do + wv <- WV.newWVar (0 :: Int) + wc <- WV.cacheWVar wv + ret <- L.sort . L.concat <$> T.mapConcurrently (countConc10 wc) + ret `T.shouldBe` [1..1000] + where + countConc10 wc = L.replicate 10 $ count100 wc + count100 wc = M.replicateM 100 $ countOne wc + countOne wc = do + (suc, wt1) <- WV.tryTakeWCached wc + let val = WV.readWTicket wt1 + if suc + then do + wt2 <- WV.putWCached wc $ val + 1 + return $ WV.readWTicket wt2 + else do + wt2 <- WV.readFreshWCached wc { WV.cachedTicket = wt1 } + countOne wc { WV.cachedTicket = wt2 } + +wvarSpec :: HS.Spec +wvarSpec = do + takeWVarSeqSpec + tryTakeWVarSeqSpec + readFreshWVarSeqSpec + tryReadFreshWVarSeqSpec + +wcachedSpec :: HS.Spec +wcachedSpec = do + takeWCachedSeqSpec + tryTakeWCachedSeqSpec + readFreshWCachedSeqSpec + tryReadFreshWCachedSeqSpec + +spec :: HS.Spec +spec = HS.describe "WVar concurrent specs" $ do + HS.describe "WVar" wvarSpec + HS.describe "WCached" wcachedSpec + HS.describe "combination" tryTakeAndPutCachedSeqSpec + + + diff --git a/test/WVarSpec.hs b/test/WVarSpec.hs new file mode 100644 --- /dev/null +++ b/test/WVarSpec.hs @@ -0,0 +1,273 @@ +{-# LANGUAGE ScopedTypeVariables #-} + +module WVarSpec where + +import qualified Test.Expectations as T +import qualified Test.WVar as T + +import qualified Test.Hspec as HS +import qualified Test.QuickCheck as Q + +import qualified Control.Concurrent.WVar as WV +import qualified Control.Monad as M + +takeWVarSpec :: HS.Spec +takeWVarSpec = HS.describe "takeWVar" $ do + T.whenWVarIsFresh $ \ prepare -> do + HS.it "takes the value without blocking" $ do + (val, wv) <- prepare + r <- WV.takeWVar wv `T.shouldNotBlock` 500000 + r `T.shouldBe` val + T.whenWVarIsUpdating $ \ prepare -> do + HS.it "blocks until WVar becomes fresh" $ do + (val1 :: Int, wv) <- prepare + val2 <- Q.generate $ Q.arbitrary `Q.suchThat` (/= val1) + wait <- WV.takeWVar wv `T.shouldBlock` 500000 + M.void $ WV.putWVar wv val2 `T.shouldNotBlock` 500000 + r <- wait `T.shouldAwakeFinish` 500000 + r `T.shouldBe` val2 + +tryTakeWVarSpec :: HS.Spec +tryTakeWVarSpec = HS.describe "tryTakeWVar" $ do + T.whenWVarIsFresh $ \ prepare -> do + HS.it "takes the value without blocking" $ do + (val :: Int, wv) <- prepare + r <- WV.tryTakeWVar wv `T.shouldNotBlock` 500000 + r `T.shouldBe` (True, val) + T.whenWVarIsUpdating $ \ prepare -> do + HS.it "takes the latest value without blocking" $ do + (val :: Int, wv) <- prepare + r <- WV.tryTakeWVar wv `T.shouldNotBlock` 500000 + r `T.shouldBe` (False, val) + +putWVarSpec :: HS.Spec +putWVarSpec = do + T.whenWVarIsFresh $ \ prepare -> do + HS.it "writes the value without blocking" $ do + (val1 :: Int, wv) <- prepare + val2 <- Q.generate $ Q.arbitrary `Q.suchThat` (/= val1) + M.void $ WV.putWVar wv val2 `T.shouldNotBlock` 500000 + r <- WV.takeWVar wv `T.shouldNotBlock` 500000 + r `T.shouldBe` val2 + T.whenWVarIsUpdating $ \ prepare -> do + HS.it "writes the value without blocking" $ do + (val1 :: Int, wv) <- prepare + val2 <- Q.generate $ Q.arbitrary `Q.suchThat` (/= val1) + M.void $ WV.putWVar wv val2 `T.shouldNotBlock` 500000 + r <- WV.takeWVar wv `T.shouldNotBlock` 500000 + r `T.shouldBe` val2 + +readWVarSpec :: HS.Spec +readWVarSpec = HS.describe "readWVar" $ do + T.whenWVarIsFresh $ \ prepare -> do + HS.it "reads the value without blocking" $ do + (val :: Int, wv) <- prepare + r <- WV.readWVar wv `T.shouldNotBlock` 500000 + r `T.shouldBe` val + T.whenWVarIsUpdating $ \ prepare -> do + HS.it "reads the value without blocking" $ do + (val :: Int, wv) <- prepare + r <- WV.readWVar wv `T.shouldNotBlock` 500000 + r `T.shouldBe` val + +readFreshWVarSpec :: HS.Spec +readFreshWVarSpec = HS.describe "readFreshWVar" $ do + T.whenWVarIsFresh $ \ prepare -> do + HS.it "reads the value without blocking" $ do + (val :: Int, wv) <- prepare + r <- WV.readFreshWVar wv `T.shouldNotBlock` 500000 + r `T.shouldBe` val + T.whenWVarIsUpdating $ \ prepare -> do + HS.it "blocks until WVar becomes fresh" $ do + (val1 :: Int, wv) <- prepare + val2 <- Q.generate $ Q.arbitrary `Q.suchThat` (/= val1) + wait <- WV.readFreshWVar wv `T.shouldBlock` 500000 + M.void $ WV.putWVar wv val2 `T.shouldNotBlock` 500000 + r <- wait `T.shouldAwakeFinish` 500000 + r `T.shouldBe` val2 + +tryReadFreshWVarSpec :: HS.Spec +tryReadFreshWVarSpec = HS.describe "readFreshWVar" $ do + T.whenWVarIsFresh $ \ prepare -> do + HS.it "reads the value without blocking" $ do + (val :: Int, wv) <- prepare + r <- WV.tryReadFreshWVar wv `T.shouldNotBlock` 500000 + r `T.shouldBe` (True, val) + T.whenWVarIsUpdating $ \ prepare -> do + HS.it "reads the value without blocking" $ do + (val :: Int, wv) <- prepare + r <- WV.tryReadFreshWVar wv `T.shouldNotBlock` 500000 + r `T.shouldBe` (False, val) + +takeWCachedSpec :: HS.Spec +takeWCachedSpec = HS.describe "takeWCached" $ do + T.whenWVarIsFresh $ T.withLatestCache takeValueWithoutBlocking + T.whenWVarIsUpdating $ T.withLatestCache blocksUntilWVarBecomeFresh + T.whenWVarIsFreshButCacheStaled takeValueWithoutBlocking + T.whenWVarIsUpdatingAndCacheStaled blocksUntilWVarBecomeFresh + where + takeValueWithoutBlocking prepare = do + HS.it "takes the latest value without blocking" $ do + (_, val :: Int, _, wc) <- prepare + wt <- WV.takeWCached wc `T.shouldNotBlock` 500000 + WV.readWTicket wt `T.shouldBe` val + blocksUntilWVarBecomeFresh prepare = do + HS.it "blocks until WVar becomes fresh" $ do + (_, val1 :: Int, wv, wc) <- prepare + val2 <- Q.generate $ Q.arbitrary `Q.suchThat` (/= val1) + wait <- WV.takeWCached wc `T.shouldBlock` 500000 + M.void $ WV.putWVar wv val2 `T.shouldNotBlock` 500000 + wt <- wait `T.shouldAwakeFinish` 500000 + WV.readWTicket wt `T.shouldBe` val2 + +tryTakeWCachedSpec :: HS.Spec +tryTakeWCachedSpec = HS.describe "tryTakeWCached" $ do + T.whenWVarIsFresh $ \ prepare -> do + HS.it "takes the value without blocking" $ do + (val :: Int, wv) <- prepare + wc <- WV.cacheWVar wv + (ret, wt) <- WV.tryTakeWCached wc `T.shouldNotBlock` 500000 + ret `T.shouldBe` True + WV.readWTicket wt `T.shouldBe` val + T.whenWVarIsUpdating $ T.withLatestCache failToTakeButReadLatest + T.whenWVarIsFreshButCacheStaled failToTakeButReadLatest + T.whenWVarIsUpdatingAndCacheStaled failToTakeButReadLatest + where + failToTakeButReadLatest prepare = do + HS.it "fails but reads the latest value without blocking" $ do + (_, val :: Int, _, wc) <- prepare + (ret, wt) <- WV.tryTakeWCached wc `T.shouldNotBlock` 500000 + ret `T.shouldBe` False + WV.readWTicket wt `T.shouldBe` val + +putWCachedSpec :: HS.Spec +putWCachedSpec = HS.describe "putWCached" $ do + T.whenWVarIsFresh $ T.withLatestCache writeValueWithoutBlocking + T.whenWVarIsUpdating $ T.withLatestCache writeValueWithoutBlocking + T.whenWVarIsFreshButCacheStaled writeValueWithoutBlocking + T.whenWVarIsUpdatingAndCacheStaled writeValueWithoutBlocking + where + writeValueWithoutBlocking prepare = do + HS.it "writes the value without blocking" $ do + (_, val1 :: Int, _, wc) <- prepare + val2 <- Q.generate $ Q.arbitrary `Q.suchThat` (/= val1) + wt <- WV.putWCached wc val2 `T.shouldNotBlock` 500000 + WV.readWTicket wt `T.shouldBe` val2 + +tryPutWCachedSpec :: HS.Spec +tryPutWCachedSpec = HS.describe "tryPutWCached" $ do + T.whenWVarIsFresh $ T.withLatestCache writeValueWithoutBlocking + T.whenWVarIsUpdating $ T.withLatestCache writeValueWithoutBlocking + T.whenWVarIsFreshButCacheStaled failToWrite + T.whenWVarIsUpdatingAndCacheStaled failToWrite + where + writeValueWithoutBlocking prepare = do + HS.it "writes the value without blocking" $ do + (_, val1 :: Int, _, wc) <- prepare + val2 <- Q.generate $ Q.arbitrary `Q.suchThat` (/= val1) + (ret, wt) <- WV.tryPutWCached wc val2 `T.shouldNotBlock` 500000 + ret `T.shouldBe` True + WV.readWTicket wt `T.shouldBe` val2 + failToWrite prepare = do + HS.it "fails to write the value without blocking" $ do + (_, val1 :: Int, _, wc) <- prepare + val2 <- Q.generate $ Q.arbitrary `Q.suchThat` (/= val1) + (ret, wt) <- WV.tryPutWCached wc val2 `T.shouldNotBlock` 500000 + ret `T.shouldBe` False + WV.readWTicket wt `T.shouldBe` val1 + +readWCachedSpec :: HS.Spec +readWCachedSpec = HS.describe "readWCached" $ do + T.whenWVarIsFresh $ T.withLatestCache readLatestValue + T.whenWVarIsUpdating $ T.withLatestCache readLatestValue + T.whenWVarIsFreshButCacheStaled readOldValue + T.whenWVarIsUpdatingAndCacheStaled readOldValue + where + readLatestValue prepare = do + HS.it "reads the latest value in the ticket" $ do + (_, val :: Int, _, wc) <- prepare + WV.readWCached wc `T.shouldBe` val + readOldValue prepare = do + HS.it "reads the old value in the ticket" $ do + (val :: Int, _, _, wc) <- prepare + WV.readWCached wc `T.shouldBe` val + +readFreshWCachedSpec :: HS.Spec +readFreshWCachedSpec = HS.describe "readFreshWCached" $ do + T.whenWVarIsFresh $ \ prepare -> do + HS.it "reads the latest value without blocking" $ do + (val :: Int, wv) <- prepare + wc <- WV.cacheWVar wv + wt <- WV.readFreshWCached wc `T.shouldNotBlock` 500000 + WV.readWTicket wt `T.shouldBe` val + T.whenWVarIsUpdating $ \ prepare -> do + HS.it "blocks until WVar becomes fresh" $ do + (val1 :: Int, wv) <- prepare + val2 <- Q.generate $ Q.arbitrary `Q.suchThat` (/= val1) + wc <- WV.cacheWVar wv + wait <- WV.readFreshWCached wc `T.shouldBlock` 500000 + M.void $ WV.putWVar wv val2 `T.shouldNotBlock` 500000 + wt <- wait `T.shouldAwakeFinish` 500000 + WV.readWTicket wt `T.shouldBe` val2 + T.whenWVarIsFreshButCacheStaled readOldValue + T.whenWVarIsUpdatingAndCacheStaled readOldValue + where + readOldValue prepare = do + HS.it "reads the old value without blocking" $ do + (val :: Int, _, _, wc) <- prepare + wt <- WV.readFreshWCached wc `T.shouldNotBlock` 500000 + WV.readWTicket wt `T.shouldBe` val + +tryReadFreshWCachedSpec :: HS.Spec +tryReadFreshWCachedSpec = HS.describe "tryReadFreshWCached" $ do + T.whenWVarIsFresh $ \ prepare -> do + HS.it "reads the value without blocking" $ do + (val :: Int, wv) <- prepare + wc <- WV.cacheWVar wv + (ret, wt) <- WV.tryReadFreshWCached wc `T.shouldNotBlock` 500000 + ret `T.shouldBe` True + WV.readWTicket wt `T.shouldBe` val + T.whenWVarIsUpdating $ \ prepare -> do + HS.it "reads the value without blocking" $ do + (val :: Int, wv) <- prepare + wc <- WV.cacheWVar wv + (ret, wt) <- WV.tryReadFreshWCached wc `T.shouldNotBlock` 500000 + ret `T.shouldBe` False + WV.readWTicket wt `T.shouldBe` val + T.whenWVarIsFreshButCacheStaled $ \ prepare -> do + HS.it "reads the old value without blocking" $ do + (val :: Int, _, _, wc) <- prepare + (ret, wt) <- WV.tryReadFreshWCached wc `T.shouldNotBlock` 500000 + ret `T.shouldBe` True + WV.readWTicket wt `T.shouldBe` val + T.whenWVarIsUpdatingAndCacheStaled $ \ prepare -> do + HS.it "reads the old value without blocking" $ do + (val :: Int, _, _, wc) <- prepare + (ret, wt) <- WV.tryReadFreshWCached wc `T.shouldNotBlock` 500000 + ret `T.shouldBe` True + WV.readWTicket wt `T.shouldBe` val + +wvarSpec :: HS.Spec +wvarSpec = do + takeWVarSpec + tryTakeWVarSpec + putWVarSpec + readWVarSpec + readFreshWVarSpec + tryReadFreshWVarSpec + +wcachedSpec :: HS.Spec +wcachedSpec = do + takeWCachedSpec + tryTakeWCachedSpec + putWCachedSpec + tryPutWCachedSpec + readWCachedSpec + readFreshWCachedSpec + tryReadFreshWCachedSpec + +spec :: HS.Spec +spec = HS.describe "WVar basic specs" $ do + HS.describe "WVar" wvarSpec + HS.describe "WCached" wcachedSpec + diff --git a/test/doctest.hs b/test/doctest.hs new file mode 100644 --- /dev/null +++ b/test/doctest.hs @@ -0,0 +1,10 @@ +module Main where + +import Test.DocTest + +main :: IO () +main = doctest + [ "-isrc" + , "src/Control/Concurrent/WVar.hs" + , "src/Control/Concurrent/KazuraQueue.hs" + ]