shared-resource-cache 0.2.0.1 → 0.2.0.2
raw patch · 4 files changed
+41/−34 lines, 4 filesdep +focusPVP ok
version bump matches the API change (PVP)
Dependencies added: focus
API changes (from Hackage documentation)
Files
- shared-resource-cache.cabal +3/−1
- src/Data/SharedResourceCache.hs +4/−4
- src/Data/SharedResourceCache/Internal/Broom.hs +15/−15
- src/Data/SharedResourceCache/Internal/ExpiringSharedResourceCache.hs +19/−14
@@ -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 <<https://hackage.haskell.org/package/stm-2.5.3.1/docs/Control-Concurrent-STM-TChan.html TChan broadcast channel>>. @@ -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.*
@@ -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
@@ -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
@@ -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)+ )