next-ref (empty) → 0.1.0.0

raw patch · 5 files changed

+233/−0 lines, 5 filesdep +basedep +hspecdep +next-refsetup-changed

Dependencies added: base, hspec, next-ref, stm

Files

• LICENSE +30/−0
• Setup.hs +2/−0
• next-ref.cabal +33/−0
• src/Control/Concurrent/NextRef.hs +97/−0
• test/Spec.hs +71/−0

@@ -0,0 +1,30 @@+Copyright skedge.me (c) 2016++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 Author name here 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.

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

@@ -0,0 +1,33 @@+name: next-ref+version: 0.1.0.0+cabal-version: >=1.10+build-type: Simple+license: BSD3+license-file: LICENSE+copyright: 2016 skedge.me+maintainer: jonathangfischoff@gmail.com+synopsis: A concurrency primitive for a slow consumer.+description:+    A concurrency primitive for a slow consumer that can tolerate missing some updates.+category: Web+author: Jonathan Fischoff++library+    exposed-modules:+        Control.Concurrent.NextRef+    build-depends:+        base >=4.7 && <5,+        stm >=2.4.4.1 && <2.5+    default-language: Haskell2010+    hs-source-dirs: src++test-suite next-ref-test+    type: exitcode-stdio-1.0+    main-is: Spec.hs+    build-depends:+        base >=4.8.2.0 && <4.9,+        next-ref >=0.1.0.0 && <0.2,+        hspec >=2.2.3 && <2.3+    default-language: Haskell2010+    hs-source-dirs: test+    ghc-options: -threaded -rtsopts -with-rtsopts=-N

@@ -0,0 +1,97 @@+{-# LANGUAGE LambdaCase, RecordWildCards, BangPatterns #-}+{-| This package contains a concurrency primitive which can be used to limit an+    expensive consumer from running unnecessarily. Crucially the consumer must+    be able to tolerate missing some updates.++    'NextRef' provides non-blocking writes, blocking reads, and non-blocking +    reads.++    The blocking read interface ('takeNextRef') will not necessarily present +    all values. ++    Additionally the 'NextRef' can be 'closed'. This is useful to graceful+    shutdown the consumer when the producer closes the 'NextRef'+    +-}+module Control.Concurrent.NextRef +  ( NextRef +  , newNextRef+  , takeNextRef+  , readLast+  , writeNextRef+  , modifyNextRef +  , close+  , open+  , status+  , Status (..)+  ) where+import Control.Concurrent.STM+import Data.IORef++-- | Status is used to prevent future reads. When the status is 'Closed'+--   'takeNextRef' will always return 'Nothing'. When the status is +--   open it will return Just. This is based off of the design of 'TMQueue'+--   from the 'stm-chans' package+data Status = Open | Closed+  deriving (Show, Eq, Ord, Read, Enum, Bounded)++-- | A concurrency primitive for a slow consumer that can tolerate+--   missing some updates.+data NextRef a = NextRef  +  { nrAccum     :: IORef a+  , nrNextValue :: TMVar a+  , nrStatus    :: TVar  Status +  }++-- | Create a 'NextVar'+newNextRef :: a -> IO (NextRef a)+newNextRef x = NextRef <$> newIORef x <*> newTMVarIO x <*> newTVarIO Open++-- | Block until the next value is available. If the 'NextVar' is +--   closed it returns 'Nothing' immediantly. +takeNextRef :: NextRef a -> IO (Maybe a)+takeNextRef NextRef {..} = atomically $ readTVar nrStatus >>= \case+  Closed -> return Nothing+  Open   -> Just <$> takeTMVar nrNextValue++update :: NextRef a -> a -> IO ()+update NextRef {..} !new = atomically $ readTVar nrStatus >>= \case+  Closed -> return ()+  Open   -> do+    tryTakeTMVar nrNextValue+    putTMVar     nrNextValue new++-- | Read the most recent value. Non-blocking+readLast :: NextRef a -> IO a+readLast NextRef {..} = readIORef nrAccum++tupleResult :: (a, b) -> (a, (a, b))+tupleResult (x, y) = (x, (x, y))++-- | Write a new value. Never blocks.+writeNextRef :: NextRef a -> a -> IO ()+writeNextRef nv@(NextRef {..}) newValue = do +  writeIORef nrAccum newValue+  update nv newValue++-- | Apply a function to current value to produce the next value and return +--   a result. +modifyNextRef :: NextRef a -> (a -> (a, b)) -> IO b+modifyNextRef nv@(NextRef {..}) f = do+  (!newValue, !result) <- atomicModifyIORef' nrAccum $ tupleResult . f+  update nv newValue+  return result++-- | Modify the status of the 'NextRef' to 'Closed'. All future reads+--   using 'takeNextRef' will result a 'Nothing'. 'readLast' is unaffected.+close :: NextRef a -> IO ()+close NextRef {..} = atomically $ writeTVar nrStatus Closed++-- | Modify the status of the 'NextRef' to 'Closed'. All future reads+--   using 'takeNextRef' will return a 'Just'. 'readLast' is unaffected.+open :: NextRef a -> IO ()+open NextRef {..} = atomically $ writeTVar nrStatus Open++-- | Get the current status of the 'NextRef'+status :: NextRef a -> IO Status+status = atomically . readTVar . nrStatus

@@ -0,0 +1,71 @@+{-# LANGUAGE LambdaCase #-}+import Test.Hspec+import Control.Concurrent.NextRef+import GHC.Conc+import Control.Monad (void, replicateM_)++while :: IO Bool -> IO () -> IO ()+while test act = test >>= \case +  True  -> act >> while test act+  False -> return ()+  ++main :: IO ()+main = hspec $ describe "NextRef" $ do +  it "newNextRef/takeNext/takeNext blocks" $ do+    ref <- newNextRef ()+    takeNextRef ref+    threadId <- forkIO $ void $ takeNextRef ref+    +    while ((== ThreadRunning) `fmap` threadStatus threadId) $ threadDelay 100000+    +    stat <- threadStatus threadId +    stat `shouldBe` ThreadBlocked BlockedOnSTM+    +  it "writing never blocks" $ do+    ref <- newNextRef 1+    replicateM_ 10 $ writeNextRef ref 2+    True `shouldBe` True +  +  it "readLast does not block" $ do+    ref <- newNextRef ()+    takeNextRef ref+    readLast ref+    +    True `shouldBe` True +    +  it "takeNext/write/takeNext doesn't block, updates correctly" $ do+    ref <- newNextRef (1 :: Int)+    actual <- takeNextRef ref+    actual `shouldBe` Just 1+    +    writeNextRef ref 2+    +    actual1 <- takeNextRef ref+    actual1 `shouldBe` Just 2+    +  it "takeNext/modify/takeNext doesn't block, updates correctly" $ do+    ref <- newNextRef (1 :: Int)+    actual <- takeNextRef ref+    actual `shouldBe` Just 1+    +    modifyNextRef ref (\x -> (x+1, ()))+    +    actual1 <- takeNextRef ref+    actual1 `shouldBe` Just 2+    +  it "close/takeNext give nothing" $ do+    ref <- newNextRef ()+    close ref+    actual <- takeNextRef ref+    actual `shouldBe` Nothing+    +  it "close/open/takeNext gives value" $ do+    ref <- newNextRef ()+    close ref+    open  ref+    actual <- takeNextRef ref+    actual `shouldBe` Just ()+  +    +