diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 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.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 <>. @@ -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 diff --git a/src/Data/SharedResourceCache.hs b/src/Data/SharedResourceCache.hs --- a/src/Data/SharedResourceCache.hs +++ b/src/Data/SharedResourceCache.hs @@ -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 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,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 - 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 @@ -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 diff --git a/src/Data/SharedResourceCache/Internal/Model.hs b/src/Data/SharedResourceCache/Internal/Model.hs --- a/src/Data/SharedResourceCache/Internal/Model.hs +++ b/src/Data/SharedResourceCache/Internal/Model.hs @@ -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 ())