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.1 +version: 0.2.0.2 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 <>. @@ -44,6 +44,7 @@ ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wmissing-export-lists -Wmissing-home-modules -Wpartial-fields -Wredundant-constraints build-depends: base >=4.7 && <5 + , focus ==1.0.* , hashable >=1.3 && <2 , list-t ==1.0.* , resourcet ==1.3.* @@ -63,6 +64,7 @@ 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 + , focus ==1.0.* , hashable >=1.3 && <2 , list-t ==1.0.* , resourcet ==1.3.* diff --git a/src/Data/SharedResourceCache.hs b/src/Data/SharedResourceCache.hs --- a/src/Data/SharedResourceCache.hs +++ b/src/Data/SharedResourceCache.hs @@ -27,7 +27,7 @@ import Control.Monad.Trans.Resource (MonadResource, ReleaseKey, runResourceT, allocate) import Control.Monad.IO.Class (liftIO) import Data.SharedResourceCache.Internal.CacheItem (CacheItem(..)) -import Data.SharedResourceCache.Internal.ExpiringSharedResourceCache (CacheEntry(..), SharedResourceCache (..), CacheExpiryConfig, loadCacheableResource, handleSharerLeave, handlerSharerJoin, handleSharerLeaveSTM) +import Data.SharedResourceCache.Internal.ExpiringSharedResourceCache (CacheEntry(..), SharedResourceCache (..), CacheExpiryConfig, loadCacheableResource, handleSharerLeave, handleSharerJoin, handleSharerLeaveSTM) import Data.SharedResourceCache.Internal.Broom (startBroomLoop) import Data.SharedResourceCache.Internal.Model (CacheExpiryConfig(..)) import Control.Exception (uninterruptibleMask_) @@ -36,7 +36,7 @@ -- manage periodically removing cache entries that have expired (as per the cache expiry configuration.) makeGlobalSharedResourceCache :: Hashable key - => (key -> IO (Either err value)) -- ^ The action to load the given resource by ID. If the 'Either' result is a 'Left' or the IO action throws, the result is not stored in the cache + => (key -> IO (Either err value)) -- ^ The action to load the given resource by ID. If the 'Either' result is a 'Left' or the IO action throws, the result is not stored in the cache. Wrapping the IO function in a timeout to avoid the resource becoming inaccessible to other threads if the operation hangs should be considered. -> Maybe (value -> IO ()) -- ^ An action to be executed when the item is removed from the cache. If this action throws an exception it is silently ignored. -> CacheExpiryConfig -- ^ The configuration for when the cache item should be marked as expired and be eligible for removal from the cache -> IO (SharedResourceCache key value err) @@ -116,7 +116,7 @@ item <- M.lookup resourceId (cache resourceCache) case item of Just (LoadedEntry cachedItem) -> do - handlerSharerJoin resourceCache cachedItem resourceId + handleSharerJoin resourceCache cachedItem resourceId pure (Just cachedItem) _ -> pure Nothing @@ -140,7 +140,7 @@ resource <- M.lookup resourceId (cache resourceCache) case resource of Just (LoadedEntry item) -> do - handlerSharerJoin resourceCache item resourceId + handleSharerJoin resourceCache item resourceId result <- action (Just (cacheItem item)) handleSharerLeaveSTM resourceCache item resourceId now pure result diff --git a/src/Data/SharedResourceCache/Internal/Broom.hs b/src/Data/SharedResourceCache/Internal/Broom.hs --- a/src/Data/SharedResourceCache/Internal/Broom.hs +++ b/src/Data/SharedResourceCache/Internal/Broom.hs @@ -1,4 +1,5 @@ {-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE LambdaCase #-} module Data.SharedResourceCache.Internal.Broom (startBroomLoop, removeIfStale, scheduleCacheCleanup, removeScheduledCleanup) where import qualified StmContainers.Map as M import Data.Hashable (Hashable) @@ -12,7 +13,8 @@ import ListT (traverse_) import Data.Time (addUTCTime, getCurrentTime) import Data.SharedResourceCache.Internal.Model (CacheEntry (..), CacheExpiryConfig (..)) - + import Focus (Focus (Focus), Change (Leave, Remove)) + startBroomLoop :: Hashable key => M.Map key (CacheEntry value) -> M.Map key UTCTime -> Maybe (value -> IO ()) -> Int -> IO () startBroomLoop resourceCache cleanUpMap onRemove sweepIntervalSeconds = forever $ do now <- getCurrentTime @@ -30,23 +32,21 @@ _ -> return () where removeIfNoSharers :: M.Map key (CacheEntry value) -> M.Map key UTCTime -> key -> STM (Maybe value) - removeIfNoSharers cache cleanupMap resourceId = do - resource <- M.lookup resourceId cache - case resource of - Just (LoadedEntry cached) -> do - sharers <- numberOfSharers cached + removeIfNoSharers cache cleanupMap resourceId = M.focus removeIfNoSharersStrategy resourceId cache + removeIfNoSharersStrategy :: Focus (CacheEntry value) STM (Maybe value) + removeIfNoSharersStrategy = Focus + (pure (Nothing, Leave)) + (\case + (LoadedEntry cached) -> do + sharers <- numberOfSharers cached if sharers == 0 then do - removeFromCache cache resourceId - removeScheduledCleanup cleanupMap resourceId - return (Just (cacheItem cached)) - else - return Nothing - _ -> pure Nothing - - removeFromCache :: M.Map key (CacheEntry value) -> key -> STM () - removeFromCache resourceCache resourceId = M.delete resourceId resourceCache + removeScheduledCleanup cleanUpMap resourceId + pure (Just (cacheItem cached), Remove) + else pure (Nothing, Leave) + _ -> pure (Nothing, Leave) + ) scheduleCacheCleanup :: Hashable key => M.Map key UTCTime -> CacheExpiryConfig -> UTCTime -> key -> STM () scheduleCacheCleanup cleanUpMap (CacheExpiryConfig _ itemEligibleForRemovalAfterUnusedSeconds) now resourceId = do 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,5 +1,6 @@ {-# LANGUAGE ScopedTypeVariables #-} -module Data.SharedResourceCache.Internal.ExpiringSharedResourceCache (CacheEntry(..), SharedResourceCache(..), CacheExpiryConfig, loadCacheableResource, handleSharerLeave, handleSharerLeaveSTM, handlerSharerJoin) where +{-# 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 Control.Concurrent ( MVar, ThreadId, putMVar, readMVar ) import qualified StmContainers.Map as M @@ -17,6 +18,7 @@ import Control.Concurrent.MVar (newEmptyMVar) import Control.Concurrent.STM.TVar (newTVar) import Data.Either (isLeft) + import Focus (Focus (Focus), Change (Leave, Remove)) -- | A cache of resources that can be shared between multiple threads (such as a TChan broadcast channel.) -- @@ -29,8 +31,8 @@ cacheExpiryConfig :: CacheExpiryConfig } - handlerSharerJoin :: Hashable key => SharedResourceCache key value err -> CacheItem value -> key -> STM () - handlerSharerJoin cache cacheItem resourceId = do + handleSharerJoin :: Hashable key => SharedResourceCache key value err -> CacheItem value -> key -> STM () + handleSharerJoin cache cacheItem resourceId = do void $ increaseSharersByOne cacheItem removeScheduledCleanup (cleanUpMap cache) resourceId @@ -55,7 +57,7 @@ Nothing -> pure Nothing Just result@(LoadingEntry loadingMVar) -> pure (Just result) Just result@(LoadedEntry resource) -> do - handlerSharerJoin resourceCache resource resourceId + handleSharerJoin resourceCache resource resourceId pure (Just result) case existingItem of @@ -117,18 +119,21 @@ atomically $ do connections <- newTVar 0 let entry = CacheItem resource connections - handlerSharerJoin resourceCache entry resourceId + handleSharerJoin resourceCache entry resourceId M.insert (LoadedEntry entry) resourceId cache pure (Right entry) adjustCacheEntryOnLoadError :: SharedResourceCache key value err -> key -> IO () - adjustCacheEntryOnLoadError resourceCache@(SharedResourceCache cache _ _ _ _ _) resourceId = do + adjustCacheEntryOnLoadError (SharedResourceCache cache _ _ _ _ _) resourceId = + atomically $ M.focus removeIfLoading resourceId cache - atomically $ do - result <- M.lookup resourceId cache - case result of - -- Error occurred before we were able to load the cache entry - delete it so that another thread can claim - -- the semaphore to try loading it - Just (LoadingEntry _) -> M.delete resourceId cache - -- It shouldn't be possible for a loaded entry to enter the cache and _then_ an exception to occur - _ -> pure () + -- | Remove the cache entry if it's still in the loading state, 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 () + removeIfLoading = Focus + (pure ((), Leave)) + (\case + LoadingEntry _ -> pure ((), Remove) + _ -> pure ((), Leave) + )