diff --git a/ChangeLog.md b/ChangeLog.md new file mode 100644 --- /dev/null +++ b/ChangeLog.md @@ -0,0 +1,3 @@ +# Changelog for concurrent-resource-map + +## Unreleased changes diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright Mateusz Kowalczyk (c) 2020 + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above + copyright notice, this list of conditions and the following + disclaimer in the documentation and/or other materials provided + with the distribution. + + * Neither the name of Mateusz Kowalczyk nor the names of other + contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,16 @@ +# concurrent-resource-map + +User-counted resource map with automatic resource collection, aimed to +be used in concurrent setting. + +Consider having some sort of resource that you need properly +initialised and cleaned up but only once and only for as long as there +are interested users (threads). Further, instead of only a single +resource, you want a collection of such resources, keyed on some +value. + +This package implements a simple bracket-based scheme that manipulates +a resource map by counting number of initalisations and clean-ups: if +it detects that the cleanup is from the last user, it removes the +resource from the map all together. See the code/hackage for +documentation. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/concurrent-resource-map.cabal b/concurrent-resource-map.cabal new file mode 100644 --- /dev/null +++ b/concurrent-resource-map.cabal @@ -0,0 +1,57 @@ +cabal-version: 1.12 + +-- This file has been generated from package.yaml by hpack version 0.33.0. +-- +-- see: https://github.com/sol/hpack +-- +-- hash: 88961aac04b5bb12c97a5cb17d6ffa38769426d569c941425645c1799284d0f3 + +name: concurrent-resource-map +version: 0.1.0.0 +synopsis: Concurrent resource map +description: Please see the README on GitHub at +category: Data +homepage: https://github.com/Fuuzetsu/concurrent-resource-map#readme +bug-reports: https://github.com/Fuuzetsu/concurrent-resource-map/issues +author: Mateusz Kowalczyk +maintainer: fuuzetsu@fuuzetsu.co.uk +copyright: 2020 Mateusz Kowalczyk +license: BSD3 +license-file: LICENSE +tested-with: GHC ==8.10.1 || ==8.8.3 || ==8.6.3 || ==8.4.4 || ==8.2.2 || ==8.0.2 || ==7.10.3 +build-type: Simple +extra-source-files: + README.md + ChangeLog.md + +source-repository head + type: git + location: https://github.com/Fuuzetsu/concurrent-resource-map + +library + exposed-modules: + Data.ConcurrentResourceMap + other-modules: + Paths_concurrent_resource_map + hs-source-dirs: + src + ghc-options: -O2 + build-depends: + base + , containers + default-language: Haskell2010 + +test-suite concurrent-resource-map-test + type: exitcode-stdio-1.0 + main-is: Spec.hs + other-modules: + Paths_concurrent_resource_map + hs-source-dirs: + test + ghc-options: -threaded -rtsopts -with-rtsopts=-N + build-depends: + base >=4.8 && <5 + , concurrent-resource-map + , random + , stm + default-language: Haskell2010 diff --git a/src/Data/ConcurrentResourceMap.hs b/src/Data/ConcurrentResourceMap.hs new file mode 100644 --- /dev/null +++ b/src/Data/ConcurrentResourceMap.hs @@ -0,0 +1,178 @@ +module Data.ConcurrentResourceMap + ( ConcurrentResourceMap + , newResourceMap + , withSharedResource + ) where + +import Control.Exception +import qualified Control.Concurrent.MVar as MVar +import Control.Concurrent.MVar (MVar) +import qualified Data.Map.Strict as Map +import Data.Map.Strict (Map) + +data Resource r = Uninitialised | Initialised !r + +-- | Some resource with a count of the users (threads) using it. +-- +-- Internal invariant: if users = 0 then resource = Uninitialised +data CountedResource r = CountedResource + { users :: !Int + , resource :: !(Resource r) + } + +-- | A map of shared resources @r@ keyed by @k@. +newtype ConcurrentResourceMap k r = C + (MVar (Map k (MVar (CountedResource r)))) + +-- | Create an empty resource map. +newResourceMap :: Ord k => IO (ConcurrentResourceMap k r) +newResourceMap = fmap C $ MVar.newMVar Map.empty + +-- | Use a resource that can be accessed concurrently via multiple +-- threads but is only initialised and destroyed on as-needed basis. +-- If number of users falls to 0, the resource is destroyed. If a new +-- user joins and resource is not available, it's created. +-- +-- Calls to 'withSharedResource' can even be nested if you need access +-- to resources with different keys in the same map. Calling +-- 'withSharedResource' in a nested matter on same resource key should +-- have no real adverse effects either. +withSharedResource + :: Ord k + => ConcurrentResourceMap k r + -- ^ Resource map. Create with 'newResourceMap'. + -> k + -- ^ Key for the resource. This allows you to have many of the same + -- type of resource but separated: for example, one group of + -- threads could be holding onto a logging handle to syslog while + -- another could be holding a handle to a file. + -> IO r + -- ^ Initialise resource. Only ran if the resource is not yet + -- initialised. Does not run in masked context so if you need to + -- stop async exceptions, you should use 'mask' yourself. If the + -- action fails (throws an exception), the user fails and we enter + -- cleanup. + -> (r -> IO ()) + -- ^ Destroy the resource if it was initialised. Ran by last alive + -- user when it's exiting. Unlike initialisation, this _is_ ran in + -- masked context. If this action fails (by throwing an exception + -- itself), the resource will be assumed to be uninitialised and the + -- exception will be re-thrown. + -- + -- Therefore, if your cleanup can fail in a way that you have to + -- know about/recover from, you should catch exceptions coming out + -- out 'withSharedResource'. As you get reference to the resource + -- in the @act@, you're able to store it/monitor it yourself and + -- decide to take any appropriate actions in the future such as + -- blocking other threads from running initialisation again until + -- you've cleaned up the resource yourself. + -> (r -> IO a) + -- ^ Run an actionwith the initialised resource. Note that the + -- availability of this resource only ensures that the user-given + -- initialisers/destructors have been ran appropriate number of + -- times: it of course makes no guarantees as to what the resource + -- represents. For example, if it's a 'System.Process.ProcessHandle' + -- or a database connection, there's no guarantee that the process + -- is alive or that the database connection is still available. For + -- resources that can dynamically fail, you should implement some + -- sort of monitoring yourself. + -> IO a +withSharedResource vm k initResource destroyResource act = bracket + (addUser vm k) + (removeUser vm k destroyResource) + -- Don't leak the internal MVar to the user! This ensures that we + -- can safely remove it from the resource map when we exit through + -- 'removeUser'. + actWithResource + where + actWithResource rvar = do + r <- MVar.modifyMVar rvar $ \cr -> case cr of + CountedResource { resource = Uninitialised } -> do + r <- initResource + return (cr { resource = Initialised r }, r) + CountedResource { resource = Initialised r } -> + return (cr, r) + act r + +-- | Adds a user at given key. If it's the first user, creates the +-- underlying map. +-- +-- Should be used as initialising action in 'bracket' along with +-- 'removeUser'. +addUser + :: Ord k + => ConcurrentResourceMap k r -> k -> IO (MVar (CountedResource r)) +addUser (C vm) k = MVar.modifyMVar vm $ \m -> case Map.lookup k m of + -- We're the first user of this resource, make the counted + -- resource. + Nothing -> do + v <- MVar.newMVar CountedResource { users = 1, resource = Uninitialised } + return (Map.insert k v m, v) + -- Other users already exist, increase the count only. + Just vc -> do + MVar.modifyMVar_ vc $ \cr -> + return cr { users = users cr + 1 } + return (m, vc) + +-- | Remove user for the given key. If it's the last user, removes the +-- counted resource from the map completely. +-- +-- Should be used as cleanup action in 'bracket' along with 'addUser'. +removeUser + :: Ord k + => ConcurrentResourceMap k r + -> k + -- ^ The resource from inside the map. + -> (r -> IO ()) + -- ^ Destroy resource. + -> MVar (CountedResource r) + -- ^ Internal ref + -> IO () +removeUser (C vm) k destroyResource vc = do + cr <- MVar.takeMVar vc + let newCount = users cr - 1 + cr' = cr { users = newCount } + case cr' of + -- We're the last ones around, uninitialise. + CountedResource { users = 0, resource = Initialised r } -> do + let uninitialise = MVar.putMVar vc cr' { resource = Uninitialised } + -- Destroy the resource if we can. If we fail, uninitialise it + -- anyway and re-throw the exception. + destroy = do + destroyResource r `onException` uninitialise + uninitialise + -- We were the last ones around and whether we managed to + -- destroy the resource or not, we want to remove the internal + -- MVar from the resource map if we're still the last ones. + destroy `finally` cleanFromResourceMap + + -- Resource is uninitialised or there are some other users + -- around, simply replace the content with updated user counter. + _ -> MVar.putMVar vc cr' + where + cleanFromResourceMap = MVar.modifyMVar_ vm $ \m -> case Map.lookup k m of + -- The resource is not even in the map anymore. This could + -- happen if since we decreased the count, a new user came in, + -- increased the count, finished and cleaned up before we did. + -- Seems unlikely but not impossible: in this case, there's + -- nothing left for us to do. + Nothing -> return m + Just rvar -> do + m'cr <- MVar.tryTakeMVar rvar + case m'cr of + -- We took the lock and see that there are still no + -- remaining users. We must be the last ones and must be the + -- only user holding a useful reference to the MVar as it + -- can only be created and passed from addUser and it's + -- internal API. Remove it from the map. + Just CountedResource { users = 0 } -> return (Map.delete k m) + -- We were able to take the MVar but there are some other + -- users around again: we want to keep the original map. Put + -- the value back and keep original the original mapping. + Just cr -> do + MVar.putMVar rvar cr + return m + -- We were either unable to take the MVar which means + -- someone else has started to use it and so, we shouldn't + -- delete it. Leave it as-is in the map. + Nothing -> return m diff --git a/test/Spec.hs b/test/Spec.hs new file mode 100644 --- /dev/null +++ b/test/Spec.hs @@ -0,0 +1,87 @@ +module Main (main) where + +import Control.Concurrent +import Control.Concurrent.STM +import Control.Exception +import Control.Monad +import Data.ConcurrentResourceMap +import Data.IORef +import Data.Typeable +import System.Random + +-- | A 'resource' that asserts a number of concurrent initialisations. +-- If it's ever not 0 or 1, something went wrong and we want to fail. +newtype SingleInitResource = SIR (IORef Int) + +newSIR :: IO SingleInitResource +newSIR = fmap SIR $ newIORef 0 + +initSIR :: SingleInitResource -> IO () +initSIR (SIR r) = atomicModifyIORef' r $ \x -> case x of + 0 -> (1, ()) + n -> error $ "Tried to init SIR when it was non-one, it's: " ++ show n + +destroySIR :: SingleInitResource -> IO () +destroySIR (SIR r) = atomicModifyIORef' r $ \x -> case x of + 1 -> (0, ()) + n -> error $ "Tried to destroy SIR when it was non-one, it's: " ++ show n + +checkInitialised :: SingleInitResource -> IO () +checkInitialised (SIR r) = readIORef r >>= \x -> case x of + 1 -> return () + n -> error $ "Checking initialised SIR found non-one value, it's: " ++ show n + +checkDestroyed :: SingleInitResource -> IO () +checkDestroyed (SIR r) = readIORef r >>= \x -> case x of + 0 -> return () + n -> error $ "Checking destroyed SIR found non-zero value, it's: " ++ show n + +data IntentionalFailure = IntentionalFailure + deriving (Typeable, Show) +instance Exception IntentionalFailure + +main :: IO () +main = do + m <- newResourceMap + leftThreads <- newTVarIO numThreads + sirs <- fmap (zip [0 :: Int .. ]) $ replicateM 10 newSIR + + replicateM_ numThreads $ forkIO $ do + let act [] = return () + act ((k, sir):ss) = withSharedResource m k + (fragileInit sir >> return sir) + (\r -> smallDelay >> destroySIR r) + -- Introduce small delay to allow for resource contention + -- instead of most threads executing instantly creating a fairly + -- sequenential test scenario. + -- + -- Further, this tests that we can nest withSharedResource. + (\r -> smallDelay >> checkInitialised r >> act ss) + -- Always modify thread exit count when main action finishes. + + actIncr = act sirs `finally` atomically (modifyTVar' leftThreads pred) + -- Silence intentional failures, no need to pollute test + -- output. + silenceIntentional IntentionalFailure = return () + actIncr `catch` silenceIntentional + + -- Wait for everything to finish. + atomically $ readTVar leftThreads >>= check . (== 0) + forM_ sirs $ \(_, sir) -> checkDestroyed sir + where + fragileInit :: SingleInitResource -> IO () + fragileInit sir = do + smallDelay + -- 25% of the time, throw an exception from initialiser + shouldFail <- fmap (== 4) $ randomRIO (1, 4 :: Int) + if shouldFail + then throwIO IntentionalFailure + else initSIR sir + + smallDelay :: IO () + smallDelay = + -- Random delay up to 100 µs + randomRIO (0, 100) >>= threadDelay + + numThreads :: Int + numThreads = 50000