diff --git a/shared-resource-cache.cabal b/shared-resource-cache.cabal --- a/shared-resource-cache.cabal +++ b/shared-resource-cache.cabal @@ -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 diff --git a/src/Data/SharedResourceCache/Internal/ExpiringSharedResourceCache.hs b/src/Data/SharedResourceCache/Internal/ExpiringSharedResourceCache.hs --- a/src/Data/SharedResourceCache/Internal/ExpiringSharedResourceCache.hs +++ b/src/Data/SharedResourceCache/Internal/ExpiringSharedResourceCache.hs @@ -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 () diff --git a/test/Spec.hs b/test/Spec.hs --- a/test/Spec.hs +++ b/test/Spec.hs @@ -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"