concurrent-resource-map (empty) → 0.1.0.0

raw patch · 7 files changed

+373/−0 lines, 7 filesdep +basedep +concurrent-resource-mapdep +containerssetup-changed

Dependencies added: base, concurrent-resource-map, containers, random, stm

Files

• ChangeLog.md +3/−0
• LICENSE +30/−0
• README.md +16/−0
• Setup.hs +2/−0
• concurrent-resource-map.cabal +57/−0
• src/Data/ConcurrentResourceMap.hs +178/−0
• test/Spec.hs +87/−0

@@ -0,0 +1,3 @@+# Changelog for concurrent-resource-map++## Unreleased changes

@@ -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.

@@ -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.

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

@@ -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 <https://github.com/Fuuzetsu/concurrent-resource-map#readme>+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

@@ -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

@@ -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