packages feed

kazura-queue (empty) → 0.1.0.0

raw patch · 17 files changed

+2782/−0 lines, 17 filesdep +HUnitdep +QuickCheckdep +asyncsetup-changed

Dependencies added: HUnit, QuickCheck, async, atomic-primops, base, containers, criterion, deepseq, doctest, exceptions, free, hspec, hspec-expectations, kazura-queue, mtl, primitive, stm, transformers

Files

+ LICENSE view
@@ -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.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ bench/Main.hs view
@@ -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++
+ kazura-queue.cabal view
@@ -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
+ src/Control/Concurrent/KazuraQueue.hs view
@@ -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+
+ src/Control/Concurrent/WVar.hs view
@@ -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+
+ test/KazuraQueueConcurrentSpec.hs view
@@ -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+
+ test/KazuraQueueSpec.hs view
@@ -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+
+ test/Spec.hs view
@@ -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+
+ test/Test/Concurrent.hs view
@@ -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++
+ test/Test/Expectations.hs view
@@ -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+
+ test/Test/KazuraQueue.hs view
@@ -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)]+
+ test/Test/Util.hs view
@@ -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"+
+ test/Test/WVar.hs view
@@ -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)
+ test/WVarConcurrentSpec.hs view
@@ -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+++
+ test/WVarSpec.hs view
@@ -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+
+ test/doctest.hs view
@@ -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"+    ]