shared-resource-cache 0.1.0.3 → 0.2.0.0
raw patch · 6 files changed
+76/−57 lines, 6 filesdep +hashable
Dependencies added: hashable
Files
- CHANGELOG.md +11/−0
- shared-resource-cache.cabal +3/−1
- src/Data/SharedResourceCache.hs +30/−24
- src/Data/SharedResourceCache/Internal/Broom.hs +9/−9
- src/Data/SharedResourceCache/Internal/ExpiringSharedResourceCache.hs +22/−22
- src/Data/SharedResourceCache/Internal/Model.hs +1/−1
CHANGELOG.md view
@@ -8,6 +8,17 @@ ## Unreleased +## 0.2.0.0 - 2026-03-29++### Changed++- `SharedResourceCache` is now parameterised over the key type (`key`) rather than+ being fixed to `Text`, allowing any `Hashable` key type to be used.+- Type parameters of `SharedResourceCache` reordered and renamed: was+ `SharedResourceCache err a`, now `SharedResourceCache key value err`.+- All public functions that accept or return a resource ID now use the generic `key`+ type instead of `Text`.+ ## 0.1.0.3 - 2026-03-23 ### Fixed
@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack name: shared-resource-cache-version: 0.1.0.3+version: 0.2.0.0 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+ , hashable >=1.3 && <2 , list-t ==1.0.* , resourcet ==1.3.* , stm >=2.4 && <2.6@@ -62,6 +63,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+ , hashable >=1.3 && <2 , list-t ==1.0.* , resourcet ==1.3.* , shared-resource-cache
@@ -1,6 +1,6 @@ {-# LANGUAGE ScopedTypeVariables #-} --- | A cache that holds items (resources) while there are multiple threads using the +-- | A cache that holds items (resources) while there are multiple threads using the -- resource or until the item is eligible for removal with respect to the given cache -- expiry configuration. --@@ -20,7 +20,7 @@ import Control.Concurrent ( forkIO, killThread ) import Control.Concurrent.STM.TVar () import Control.Monad.STM ( STM, atomically )-import Data.Text ( Text )+import Data.Hashable (Hashable) import Data.Time ( UTCTime ) import qualified StmContainers.Map as M import Prelude@@ -32,13 +32,14 @@ import Data.SharedResourceCache.Internal.Model (CacheExpiryConfig(..)) import Control.Exception (uninterruptibleMask_) --- | Constructs a resource cache that is expected to be used for the lifetime of the program. Internally, it forks a thread to +-- | Constructs a resource cache that is expected to be used for the lifetime of the program. Internally, it forks a thread to -- manage periodically removing cache entries that have expired (as per the cache expiry configuration.)-makeGlobalSharedResourceCache - :: (Text -> IO (Either err a)) -- The action to load the given resource by ID- -> Maybe (a -> IO ()) -- An action to be executed when the item is removed from the cache- -> CacheExpiryConfig -- The configuration for when the cache item should be marked as expired and be eligible for removal from the cache- -> IO (SharedResourceCache err a)+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+ -> Maybe (value -> IO ()) -- ^ An action to be executed when the item is removed from the cache+ -> 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) makeGlobalSharedResourceCache loadResourceOp onRemoval cacheExpiryConfig@(CacheExpiryConfig sweepIntervalSeconds _) = do resourceCache <- M.newIO cleanUpMap <- M.newIO@@ -47,21 +48,21 @@ -- | The same as 'makeGlobalSharedResourceCache' but intended for cases where the cache doesn't live for the lifetime of the program. -- When the resource is freed (via MonadResource) the forked thread for cleaning the resource cache is killed.-makeSharedResourceCache :: (MonadResource m) => (Text -> IO (Either err a)) -> Maybe (a -> IO ()) -> CacheExpiryConfig -> m (ReleaseKey, SharedResourceCache err a)+makeSharedResourceCache :: forall key value err m. (MonadResource m, Hashable key) => (key -> IO (Either err value)) -> Maybe (value -> IO ()) -> CacheExpiryConfig -> m (ReleaseKey, SharedResourceCache key value err) makeSharedResourceCache loadResourceOp onRemoval cacheExpiryConfig = do allocate (allocateResource loadResourceOp onRemoval) deallocateResource where- allocateResource :: (Text -> IO (Either err a)) -> Maybe (a -> IO ()) -> IO (SharedResourceCache err a)+ allocateResource :: (key -> IO (Either err value)) -> Maybe (value -> IO ()) -> IO (SharedResourceCache key value err) allocateResource loadOp removal = makeGlobalSharedResourceCache loadOp removal cacheExpiryConfig - deallocateResource :: SharedResourceCache err a -> IO ()+ deallocateResource :: SharedResourceCache key value err -> IO () deallocateResource cache = do let threadId = cacheCleanupThreadId cache uninterruptibleMask_ (killThread threadId) -- | Executes the given action using the resource with the given ID. This function is a wrapper around the 'getCacheableResource' function, -- where the resource is freed at the end of the supplied action-withCacheableResource :: SharedResourceCache err a -> Text -> (Either err a -> IO ()) -> IO ()+withCacheableResource :: Hashable key => SharedResourceCache key value err -> key -> (Either err value -> IO ()) -> IO () withCacheableResource cache resourceId op = runResourceT $ do (_, cachedResource) <- getCacheableResource cache resourceId@@ -71,26 +72,31 @@ -- when the cache was constructed. -- -- If the item is already loaded into the cache, it is returned. If it is not yet loaded, the first thread requesting this resource to--- acquire a MVar will take ownership of loading it and signal to any other threads waiting for the resource that it has been loaded.--- If the IO action throws an error or results a 'Left', the MVar is removed from the cache and another waiting thread (or a future call if there are--- no current calls) can pick up the MVar to load the resource+-- insert an MVar into the cache will take ownership of loading it and signal to any other threads waiting for the resource that it has been loaded+-- (or an error occurred.) -- +-- If the IO action throws an error or results a 'Left', the MVar is removed from the cache and another waiting thread (or a future call if there are+-- no current calls) can insert a new MVar to try to load the resource+-- -- This action is executed in the 'MonadResource' monad typeclass -- context so that when the resource is freed, the cache can mark the cache item as having one less sharer. -- -- When there are no sharers remaining, the cache item is scheduled for eviction according to the configuration that was used at cache -- construction time ---getCacheableResource :: (MonadResource m) => SharedResourceCache err a -> Text -> m (ReleaseKey, Either err a)+getCacheableResource :: forall key value err m. (MonadResource m, Hashable key) =>+ SharedResourceCache key value err -> -- ^ The shared resource cache+ key -> -- ^ The key of the resource to be loaded+ m (ReleaseKey, Either err value) -- ^ The result of loading the resource and a release key to release the 'sharer' of the cache item early by calling 'Control.Monad.Trans.Resource.release' (this should be used with care to avoid an orphaned reference to a cache item that is no longer visible to other sharers.) getCacheableResource resourceCache resourceId = do (releaseKey, cacheEntry) <- allocate (allocateResource resourceCache resourceId) (deAllocateResource resourceCache) return (releaseKey, cacheItem <$> cacheEntry) where - allocateResource :: SharedResourceCache err a -> Text -> IO (Either err (CacheItem a))+ allocateResource :: SharedResourceCache key value err -> key -> IO (Either err (CacheItem value)) allocateResource = loadCacheableResource - deAllocateResource :: SharedResourceCache err a -> Either err (CacheItem a) -> IO ()+ deAllocateResource :: SharedResourceCache key value err -> Either err (CacheItem value) -> IO () deAllocateResource resourceCache resource = do case resource of Left _ -> pure ()@@ -100,21 +106,21 @@ -- -- If there are no other sharers of the resource once the resource action has been run to completion -- or the 'release key' is used to free it early then the item is scheduled for eviction from the cache.-peekCacheableResource :: MonadResource m => SharedResourceCache err a -> Text -> m (ReleaseKey, Maybe a)+peekCacheableResource :: forall key value err m. (MonadResource m, Hashable key) => SharedResourceCache key value err -> key -> m (ReleaseKey, Maybe value) peekCacheableResource resourceCache resourceId = do (releaseKey, cacheEntry) <- allocate (allocateResource resourceCache resourceId) (deAllocateResource resourceCache) return (releaseKey, cacheItem <$> cacheEntry) where- allocateResource :: SharedResourceCache err a -> Text -> IO (Maybe (CacheItem a))+ allocateResource :: SharedResourceCache key value err -> key -> IO (Maybe (CacheItem value)) allocateResource resourceCache resourceId = atomically $ do item <- M.lookup resourceId (cache resourceCache) case item of Just (LoadedEntry cachedItem) -> do handlerSharerJoin resourceCache cachedItem resourceId pure (Just cachedItem)- _ -> pure Nothing + _ -> pure Nothing - deAllocateResource :: SharedResourceCache err a -> Maybe (CacheItem a) -> IO ()+ deAllocateResource :: SharedResourceCache key value err -> Maybe (CacheItem value) -> IO () deAllocateResource resourceCache resource = case resource of Nothing -> pure ()@@ -126,10 +132,10 @@ -- If there are no other sharers of the resource once the action is complete then the item is scheduled -- for eviction ----- The 'now' parameter is used to schedule the item for eviction from the cache according to the +-- The 'now' parameter is used to schedule the item for eviction from the cache according to the -- expiry config and should be the current time. It is required to be passed in as getting the current time -- is an IO action that cannot be performed in an STM transaction.-withPeekCacheableResource :: SharedResourceCache err a -> Text -> (Maybe a -> STM b) -> UTCTime -> STM b+withPeekCacheableResource :: Hashable key => SharedResourceCache key value err -> key -> (Maybe value -> STM b) -> UTCTime -> STM b withPeekCacheableResource resourceCache resourceId action now = do resource <- M.lookup resourceId (cache resourceCache) case resource of
@@ -1,6 +1,7 @@+{-# LANGUAGE ScopedTypeVariables #-} module Data.SharedResourceCache.Internal.Broom (startBroomLoop, removeIfStale, scheduleCacheCleanup, removeScheduledCleanup) where import qualified StmContainers.Map as M- import Data.Text (Text)+ import Data.Hashable (Hashable) import Data.Time.Clock (NominalDiffTime, UTCTime) import Control.Concurrent.STM (STM) import Control.Monad (forever, when)@@ -10,8 +11,8 @@ import ListT (traverse_) import Data.Time (addUTCTime, getCurrentTime) import Data.SharedResourceCache.Internal.Model (CacheEntry (..), CacheExpiryConfig (..))- - startBroomLoop :: M.Map Text (CacheEntry err a) -> M.Map Text UTCTime -> Maybe (a -> IO ()) -> Int -> IO ()++ 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 -- We use 'listTNonAtomic since we do the removal itself atomically (followed by an IO action) and@@ -19,7 +20,7 @@ traverse_ (removeIfStale now resourceCache cleanUpMap onRemove) (M.listTNonAtomic cleanUpMap) threadDelay (sweepIntervalSeconds * 1000000) - removeIfStale :: UTCTime -> M.Map Text (CacheEntry err a) -> M.Map Text UTCTime -> Maybe (a -> IO ()) -> (Text, UTCTime) -> IO ()+ removeIfStale :: forall key err value. Hashable key => UTCTime -> M.Map key (CacheEntry value) -> M.Map key UTCTime -> Maybe (value -> IO ()) -> (key, UTCTime) -> IO () removeIfStale now cache cleanUpMap onRemove (resourceId, cacheExpiryTime) = when (now >= cacheExpiryTime) $ do removed <- atomically $ removeIfNoSharers cache cleanUpMap resourceId@@ -27,7 +28,7 @@ (Just removedItem, Just onRemovalFunction) -> onRemovalFunction removedItem _ -> return () where- removeIfNoSharers :: M.Map Text (CacheEntry err a) -> M.Map Text UTCTime -> Text -> STM (Maybe a)+ 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@@ -43,14 +44,13 @@ return Nothing _ -> pure Nothing - removeFromCache :: M.Map Text (CacheEntry err a) -> Text -> STM ()+ removeFromCache :: M.Map key (CacheEntry value) -> key -> STM () removeFromCache resourceCache resourceId = M.delete resourceId resourceCache - scheduleCacheCleanup :: M.Map Text UTCTime -> CacheExpiryConfig -> UTCTime -> Text -> STM ()+ scheduleCacheCleanup :: Hashable key => M.Map key UTCTime -> CacheExpiryConfig -> UTCTime -> key -> STM () scheduleCacheCleanup cleanUpMap (CacheExpiryConfig _ itemEligibleForRemovalAfterUnusedSeconds) now resourceId = do let eligibleForRemovalSeconds = fromIntegral itemEligibleForRemovalAfterUnusedSeconds :: NominalDiffTime M.insert (addUTCTime eligibleForRemovalSeconds now) resourceId cleanUpMap - removeScheduledCleanup :: M.Map Text UTCTime -> Text -> STM ()+ removeScheduledCleanup :: Hashable key => M.Map key UTCTime -> key -> STM () removeScheduledCleanup cleanUpMap resourceId = M.delete resourceId cleanUpMap-
@@ -3,13 +3,13 @@ import Data.SharedResourceCache.Internal.CacheItem (CacheItem (CacheItem), decreaseSharersByOne, increaseSharersByOne) import Control.Concurrent ( MVar, ThreadId, putMVar, readMVar ) import qualified StmContainers.Map as M- import Data.Text (Text)+ import Data.Hashable (Hashable) import Data.Time (UTCTime) import Control.Concurrent.STM (STM) import Control.Exception ( mask, uninterruptibleMask_, onException ) import Control.Monad.STM (atomically)- + import Control.Monad (void, when) import Data.SharedResourceCache.Internal.Broom (scheduleCacheCleanup, removeScheduledCleanup) import Data.SharedResourceCache.Internal.Model (CacheExpiryConfig, CacheEntry (..))@@ -17,37 +17,37 @@ import Control.Concurrent.MVar (newEmptyMVar) import Control.Concurrent.STM.TVar (newTVar) import Data.Either (isLeft)- + -- | A cache of resources that can be shared between multiple threads (such as a TChan broadcast channel.) --- data SharedResourceCache err a = SharedResourceCache {- cache :: M.Map Text (CacheEntry err a),- cleanUpMap :: M.Map Text UTCTime,- loadResourceOp :: Text -> IO (Either err a),- onRemoval :: Maybe (a -> IO ()),+ data SharedResourceCache key value err = SharedResourceCache {+ cache :: M.Map key (CacheEntry value),+ cleanUpMap :: M.Map key UTCTime,+ loadResourceOp :: key -> IO (Either err value),+ onRemoval :: Maybe (value -> IO ()), cacheCleanupThreadId :: ThreadId, cacheExpiryConfig :: CacheExpiryConfig } - handlerSharerJoin :: SharedResourceCache err a -> CacheItem a -> Text -> STM ()+ handlerSharerJoin :: Hashable key => SharedResourceCache key value err -> CacheItem value -> key -> STM () handlerSharerJoin cache cacheItem resourceId = do void $ increaseSharersByOne cacheItem removeScheduledCleanup (cleanUpMap cache) resourceId - handleSharerLeave :: SharedResourceCache err a -> CacheItem a -> Text -> IO ()- handleSharerLeave cache cacheItem resourceId = + handleSharerLeave :: Hashable key => SharedResourceCache key value err -> CacheItem value -> key -> IO ()+ handleSharerLeave cache cacheItem resourceId = -- We wrap this in an uninterruptibleMask_ so that we don't end up with items in the cache with phantom -- connections if the thread is killed during a blocking operation uninterruptibleMask_ $ do now <- getCurrentTime atomically (handleSharerLeaveSTM cache cacheItem resourceId now) - handleSharerLeaveSTM :: SharedResourceCache err a -> CacheItem a -> Text -> UTCTime -> STM ()+ handleSharerLeaveSTM :: Hashable key => SharedResourceCache key value err -> CacheItem value -> key -> UTCTime -> STM () handleSharerLeaveSTM (SharedResourceCache cache cleanUpMap _ _ _ config) cacheItem resourceId now = do newSharerCount <- decreaseSharersByOne cacheItem when (newSharerCount == 0) $ void (scheduleCacheCleanup cleanUpMap config now resourceId) - loadCacheableResource :: SharedResourceCache err a -> Text -> IO (Either err (CacheItem a))+ 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@@ -67,7 +67,7 @@ loadCacheableResource resourceCache resourceId Nothing -> loadFreshlyIntoCache resourceCache resourceId - loadFreshlyIntoCache :: SharedResourceCache err a -> Text -> IO (Either err (CacheItem a))+ 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 @@ -85,15 +85,15 @@ 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 err a -> Text -> IO (Maybe (MVar ()))+ takeOwnershipOfLoad :: SharedResourceCache key value err -> key -> IO (Maybe (MVar ())) takeOwnershipOfLoad resourceCache@(SharedResourceCache cache _ loadResourceOp _ _ _) resourceId = do signal <- newEmptyMVar atomically $ do@@ -105,14 +105,14 @@ pure (Just signal) _ -> pure Nothing - loadIntoCache :: SharedResourceCache err a -> Text -> MVar () -> IO (Either err (CacheItem a))+ loadIntoCache :: SharedResourceCache key value err -> key -> MVar () -> IO (Either err (CacheItem value)) loadIntoCache resourceCache@(SharedResourceCache cache _ loadResourceOp _ _ _) resourceId _ = do resourceLoadResult <- loadResourceOp resourceId case resourceLoadResult of Right resource -> putIntoCache resourceCache resourceId resource Left err -> pure (Left err) - putIntoCache :: SharedResourceCache err a -> Text -> a -> IO (Either err (CacheItem a))+ putIntoCache :: SharedResourceCache key value err -> key -> value -> IO (Either err (CacheItem value)) putIntoCache resourceCache@(SharedResourceCache cache _ loadResourceOp _ _ _) resourceId resource = do atomically $ do connections <- newTVar 0@@ -121,13 +121,13 @@ M.insert (LoadedEntry entry) resourceId cache pure (Right entry) - adjustCacheEntryOnLoadError :: SharedResourceCache err a -> Text -> IO ()- adjustCacheEntryOnLoadError resourceCache@(SharedResourceCache cache _ _ _ _ _) resourceId = do + adjustCacheEntryOnLoadError :: SharedResourceCache key value err -> key -> IO ()+ adjustCacheEntryOnLoadError resourceCache@(SharedResourceCache cache _ _ _ _ _) resourceId = do 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 + -- 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
@@ -11,4 +11,4 @@ -- ^ How long after a cache entry has had no sharers that it is eligible for removal from the cache by the sweeper background thread } - data CacheEntry err a = LoadedEntry (CacheItem a) | LoadingEntry (MVar ())+ data CacheEntry a = LoadedEntry (CacheItem a) | LoadingEntry (MVar ())