shared-resource-cache 0.2.0.3 → 0.2.0.4
raw patch · 3 files changed
+220/−50 lines, 3 filesdep +asyncdep +hspec
Dependencies added: async, hspec
Files
- shared-resource-cache.cabal +4/−2
- src/Data/SharedResourceCache/Internal/ExpiringSharedResourceCache.hs +36/−47
- test/Spec.hs +180/−1
@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack name: shared-resource-cache-version: 0.2.0.3+version: 0.2.0.4 synopsis: A thread-safe cache for sharing resources across threads with automatic expiry description: A cache designed for guaranteeing that different threads can share the same reference to a resource. For example, it allows threads to communicate via a shared [TChan broadcast channel](https://hackage.haskell.org/package/stm-2.5.3.1/docs/Control-Concurrent-STM-TChan.html).@@ -63,9 +63,11 @@ test ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wmissing-export-lists -Wmissing-home-modules -Wpartial-fields -Wredundant-constraints -threaded -rtsopts -with-rtsopts=-N build-depends:- base >=4.7 && <5+ async+ , base >=4.7 && <5 , focus ==1.0.* , hashable >=1.3 && <2+ , hspec , list-t ==1.0.* , resourcet ==1.3.* , shared-resource-cache
@@ -1,7 +1,7 @@ {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE LambdaCase #-} module Data.SharedResourceCache.Internal.ExpiringSharedResourceCache (CacheEntry(..), SharedResourceCache(..), CacheExpiryConfig, loadCacheableResource, handleSharerLeave, handleSharerLeaveSTM, handleSharerJoin) where- import Data.SharedResourceCache.Internal.CacheItem (CacheItem (CacheItem), decreaseSharersByOne, increaseSharersByOne)+ import Data.SharedResourceCache.Internal.CacheItem (CacheItem (CacheItem, cacheItem), decreaseSharersByOne, increaseSharersByOne) import Control.Concurrent ( MVar, ThreadId, putMVar, readMVar ) import qualified StmContainers.Map as M import Data.Hashable (Hashable)@@ -18,8 +18,16 @@ import Control.Concurrent.MVar (newEmptyMVar) import Control.Concurrent.STM.TVar (newTVar) import Data.Either (isLeft)- import Focus (Focus (Focus), Change (Leave, Remove))+ import Focus (Focus (Focus), Change (Leave, Remove, Set)) + data CacheDecision value+ -- The game is already in the cache+ = AlreadyPresent (CacheItem value)+ -- We're awaiting another thread loading the game+ | Await (MVar ())+ -- We have claimed ownership over loading the game+ | Claimed (MVar ()) + -- | A cache of resources that can be shared between multiple threads (such as a TChan broadcast channel.) -- data SharedResourceCache key value err = SharedResourceCache {@@ -51,62 +59,43 @@ loadCacheableResource :: Hashable key => SharedResourceCache key value err -> key -> IO (Either err (CacheItem value)) loadCacheableResource resourceCache@(SharedResourceCache cache _ loadResourceOp _ _ _) resourceId = do- existingItem <- atomically $ do- item <- M.lookup resourceId cache- case item of- Nothing -> pure Nothing- Just result@(LoadingEntry loadingMVar) -> pure (Just result)- Just result@(LoadedEntry resource) -> do- handleSharerJoin resourceCache resource resourceId- pure (Just result)+ signal <- newEmptyMVar+ existingItem <- atomically (M.focus (claimOwnershipOverLoad signal) resourceId cache) case existingItem of- Just (LoadedEntry item) -> pure $ Right item- Just (LoadingEntry loadedSignalMVar) -> do+ AlreadyPresent item -> atomically $ pure (Right item)+ Await loadedSignalMVar -> do -- Wait the for the other thread that is already loading the resource to signal that it has loaded the item -- into the cache then recursively start again readMVar loadedSignalMVar loadCacheableResource resourceCache resourceId- Nothing -> loadFreshlyIntoCache resourceCache resourceId-- loadFreshlyIntoCache :: forall key value err. Hashable key => SharedResourceCache key value err -> key -> IO (Either err (CacheItem value))- loadFreshlyIntoCache resourceCache@(SharedResourceCache cache _ loadResourceOp _ _ _) resourceId = do- maybeSemaphore <- takeOwnershipOfLoad resourceCache resourceId-- case maybeSemaphore of- -- Already loaded or loading in another thread, recursively start again- Nothing -> loadCacheableResource resourceCache resourceId-- -- We've claimed ownership of loading the item into the cache- Just semaphore -> do- -- We use uninterruptibleMask so that an async exception can't stop the sempahore remaining in the cache blocking any readers of the resource- -- from getting the resource. We're in the context of resourcet's 'allocate' so we're otherwise 'masked' apart from on blocking operations- result <- loadIntoCache resourceCache resourceId semaphore- `onException` uninterruptibleMask_ (adjustCacheEntryOnLoadError resourceCache resourceId >> signalCacheLoaded semaphore)-- uninterruptibleMask_ $ do- when (isLeft result) (adjustCacheEntryOnLoadError resourceCache resourceId)- signalCacheLoaded semaphore+ -- Load the item into the cache and then signal any other thread waiting for the resource that it is available+ Claimed claimMvar -> loadFreshlyIntoCache claimMvar resourceCache resourceId+ where+ claimOwnershipOverLoad signal = + Focus ( pure (Claimed signal, Set (LoadingEntry signal)))+ (\case+ LoadingEntry signal -> pure (Await signal, Leave)+ LoadedEntry entry -> do + handleSharerJoin resourceCache entry resourceId + pure (AlreadyPresent entry, Leave)+ ) - pure result+ loadFreshlyIntoCache :: forall key value err. Hashable key => MVar () -> SharedResourceCache key value err -> key -> IO (Either err (CacheItem value))+ loadFreshlyIntoCache semaphore resourceCache resourceId = do+ -- We use uninterruptibleMask so that an async exception can't stop the sempahore remaining in the cache blocking any readers of the resource+ -- from getting the resource. We're in the context of resourcet's 'allocate' so we're otherwise 'masked' apart from on blocking operations+ result <- loadIntoCache resourceCache resourceId semaphore+ `onException` uninterruptibleMask_ (adjustCacheEntryOnLoadError resourceCache resourceId >> signalCacheLoaded semaphore) + uninterruptibleMask_ $ do+ when (isLeft result) (adjustCacheEntryOnLoadError resourceCache resourceId)+ signalCacheLoaded semaphore + pure result where signalCacheLoaded semaphore = putMVar semaphore () - -- Returns a Just if we took ownership and Nothing if a thread is already loading the item or it has already been fully loaded- takeOwnershipOfLoad :: SharedResourceCache key value err -> key -> IO (Maybe (MVar ()))- takeOwnershipOfLoad resourceCache@(SharedResourceCache cache _ loadResourceOp _ _ _) resourceId = do- signal <- newEmptyMVar- atomically $ do- cachedItem <- M.lookup resourceId cache- case cachedItem of- Nothing -> do- let entry = LoadingEntry signal- M.insert entry resourceId cache- pure (Just signal)- _ -> pure Nothing- loadIntoCache :: SharedResourceCache key value err -> key -> MVar () -> IO (Either err (CacheItem value)) loadIntoCache resourceCache@(SharedResourceCache cache _ loadResourceOp _ _ _) resourceId _ = do resourceLoadResult <- loadResourceOp resourceId@@ -127,7 +116,7 @@ adjustCacheEntryOnLoadError (SharedResourceCache cache _ _ _ _ _) resourceId = atomically $ M.focus removeIfLoading resourceId cache - -- | Remove the cache entry if it's still in the loading state, so that another thread can claim+ -- | Remove the cache entry if it's still in the loading stitemate, so that another thread can claim -- the semaphore to try loading it. It shouldn't be possible for a loaded entry to enter the -- cache and _then_ an exception to occur. removeIfLoading :: Focus (CacheEntry value) STM ()
test/Spec.hs view
@@ -1,2 +1,181 @@+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE LambdaCase #-}+module Main (main) where++import Test.Hspec+import Data.SharedResourceCache+import Control.Monad.Trans.Resource (runResourceT, release)+import Control.Monad.IO.Class (liftIO)+import Control.Concurrent (threadDelay, newEmptyMVar, putMVar, readMVar)+import Control.Concurrent.Async (async, wait)+import Control.Monad.STM (atomically)+import Data.IORef (newIORef, readIORef, modifyIORef', atomicModifyIORef')+import Data.Time.Clock (getCurrentTime)++type Cache = SharedResourceCache String String String++shortExpiry :: CacheExpiryConfig+shortExpiry = CacheExpiryConfig 1 1++longExpiry :: CacheExpiryConfig+longExpiry = CacheExpiryConfig 100 100++makeTestCache :: Maybe (String -> IO ()) -> CacheExpiryConfig -> IO Cache+makeTestCache = makeGlobalSharedResourceCache (\key -> pure (Right ("value-" ++ key)))+ main :: IO ()-main = putStrLn "Test suite not yet implemented"+main = hspec $ do+ describe "SharedResourceCache" $ do++ it "loads a resource into the cache" $ do+ cache <- makeTestCache Nothing longExpiry++ result <- runResourceT $ do+ (_, val) <- getCacheableResource cache "key1"+ pure val++ result `shouldBe` Right "value-key1"++ it "returns the same resource instance for concurrent sharers" $ do+ loadCount <- newIORef (0 :: Int)+ cache :: Cache <- makeGlobalSharedResourceCache+ (\key -> do+ modifyIORef' loadCount (+ 1)+ pure (Right ("value-" ++ key)))+ Nothing+ longExpiry++ runResourceT $ do+ (_, val1) <- getCacheableResource cache "key1"+ liftIO $ do+ val2 <- runResourceT $ do+ (_, v) <- getCacheableResource cache "key1"+ pure v+ val1 `shouldBe` Right "value-key1"+ val2 `shouldBe` Right "value-key1"++ count <- readIORef loadCount+ count `shouldBe` 1++ it "propagates load errors without caching them" $ do+ callCount <- newIORef (0 :: Int)+ cache :: Cache <- makeGlobalSharedResourceCache+ (\_ -> do+ modifyIORef' callCount (+ 1)+ pure (Left "load failed"))+ Nothing+ longExpiry++ result1 <- runResourceT $ do+ (_, val) <- getCacheableResource cache "key1"+ pure val+ result1 `shouldBe` Left "load failed"++ result2 <- runResourceT $ do+ (_, val) <- getCacheableResource cache "key1"+ pure val+ result2 `shouldBe` Left "load failed"++ count <- readIORef callCount+ count `shouldBe` 2++ it "only loads once when multiple threads request the same uncached resource" $ do+ loadCount <- newIORef (0 :: Int)+ gate <- newEmptyMVar++ cache :: Cache <- makeGlobalSharedResourceCache+ (\key -> do+ atomicModifyIORef' loadCount (\n -> (n + 1, ()))+ readMVar gate+ pure (Right ("value-" ++ key)))+ Nothing+ longExpiry++ let numThreads = 5+ results <- mapM (\_ -> async $ runResourceT $ do+ (_, val) <- getCacheableResource cache "key1"+ pure val+ ) [1 :: Int ..numThreads]++ -- Give threads time to all reach the cache+ threadDelay 100000+ putMVar gate ()++ vals <- mapM wait results+ mapM_ (`shouldBe` Right "value-key1") vals++ count <- readIORef loadCount+ count `shouldBe` 1++ it "calls onRemoval when an item is evicted" $ do+ removedItems <- newIORef ([] :: [String])+ cache <- makeTestCache+ (Just (\val -> modifyIORef' removedItems (val :)))+ shortExpiry++ runResourceT $ do+ (_, _) <- getCacheableResource cache "key1"+ pure ()++ -- Wait for the sweep to evict it+ threadDelay 2500000++ removed <- readIORef removedItems+ removed `shouldBe` ["value-key1"]++ it "does not evict items that still have sharers" $ do+ removedItems <- newIORef ([] :: [String])+ cache <- makeTestCache+ (Just (\val -> modifyIORef' removedItems (val :)))+ shortExpiry++ runResourceT $ do+ (_, _) <- getCacheableResource cache "key1"+ -- Wait for a sweep cycle while still holding the resource+ liftIO $ threadDelay 2500000+ removed <- liftIO $ readIORef removedItems+ liftIO $ removed `shouldBe` []++ it "peek returns Nothing for uncached items" $ do+ cache <- makeTestCache Nothing longExpiry++ result <- runResourceT $ do+ (_, val) <- peekCacheableResource cache "key1"+ pure val++ result `shouldBe` Nothing++ it "peek returns the item if it is cached" $ do+ cache <- makeTestCache Nothing longExpiry++ runResourceT $ do+ (_, _) <- getCacheableResource cache "key1"+ (_, peeked) <- peekCacheableResource cache "key1"+ liftIO $ peeked `shouldBe` Just "value-key1"++ it "withCacheableResource provides the resource to the action" $ do+ cache <- makeTestCache Nothing longExpiry++ ref <- newIORef ("" :: String)+ withCacheableResource cache "key1" $ \case+ Right val -> modifyIORef' ref (const val)+ Left _ -> pure ()++ result <- readIORef ref+ result `shouldBe` "value-key1"++ it "withPeekCacheableResource returns Nothing for uncached items" $ do+ cache <- makeTestCache Nothing longExpiry++ now <- getCurrentTime+ result <- atomically $ withPeekCacheableResource cache "key1" pure now+ result `shouldBe` (Nothing :: Maybe String)++ it "releasing a resource early allows re-loading" $ do+ cache <- makeTestCache Nothing longExpiry++ runResourceT $ do+ (releaseKey, _) <- getCacheableResource cache "key1"+ release releaseKey+ (_, val2) <- getCacheableResource cache "key1"+ liftIO $ val2 `shouldBe` Right "value-key1"