Use stm-stats to reduce contention in hls-graph

ghcide/src/Development/IDE/Main.hs

@@ -11,6 +11,7 @@ module Development.IDE.Main
 import           Control.Concurrent.Extra              (newLock, readVar,
                                                         withLock,
                                                         withNumCapabilities)
+import           Control.Concurrent.STM.Stats          (dumpSTMStats)
 import           Control.Exception.Safe                (Exception (displayException),
                                                         catchAny)
 import           Control.Monad.Extra                   (concatMapM, unless,
@@ -308,6 +309,7 @@ defaultMain Arguments{..} = do
                     vfs
                     hiedb
                     hieChan
+            dumpSTMStats
         Check argFiles -> do
           dir <- IO.getCurrentDirectory
           dbLoc <- getHieDbLoc dir

hls-graph/hls-graph.cabal

@@ -31,12 +31,18 @@ flag embed-files
   manual: True
   description: Embed data files into the shake library
 
+flag stm-stats
+  default: False
+  manual: True
+  description: Collect STM transaction stats
+
 source-repository head
   type:     git
   location: https://github.com/haskell/haskell-language-server
 
 library
   exposed-modules:
+    Control.Concurrent.STM.Stats
     Development.IDE.Graph
     Development.IDE.Graph.Classes
     Development.IDE.Graph.Database
@@ -82,6 +88,9 @@ library
         build-depends:
             file-embed >= 0.0.11,
             template-haskell
+  if flag(stm-stats)
+        cpp-options: -DSTM_STATS
+
 
   ghc-options:
     -Wall -Wredundant-constraints -Wno-name-shadowing

hls-graph/src/Control/Concurrent/STM/Stats.hs

@@ -0,0 +1,184 @@
+{-# LANGUAGE CPP                 #-}
+{-# LANGUAGE RecordWildCards     #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+module Control.Concurrent.STM.Stats
+    ( atomicallyNamed
+    , atomically
+    , getSTMStats
+    , dumpSTMStats
+    , module Control.Concurrent.STM
+    ) where
+
+import           Control.Concurrent.STM hiding (atomically)
+import qualified Control.Concurrent.STM as STM
+import           Data.Map               (Map)
+#ifdef STM_STATS
+import           Control.Exception      (BlockedIndefinitelyOnSTM, Exception,
+                                         catch, throwIO)
+import           Control.Monad
+import           Data.IORef
+import qualified Data.Map.Strict        as M
+import           Data.Time              (getCurrentTime)
+import           Data.Typeable          (Typeable)
+import           GHC.Conc               (unsafeIOToSTM)
+import           System.IO
+import           System.IO.Unsafe
+import           Text.Printf
+#endif
+
+atomicallyNamed :: String -> STM a -> IO a
+atomically :: STM a -> IO a
+dumpSTMStats :: IO ()
+getSTMStats :: IO (Map String (Int,Int))
+
+#ifndef STM_STATS
+
+getSTMStats = pure mempty
+atomicallyNamed _ = atomically
+dumpSTMStats = pure ()
+atomically = STM.atomically
+
+#else
+-- adapted from the STM.Stats package
+
+atomicallyNamed = trackNamedSTM
+atomically = trackSTM
+
+-- | Global state, seems to be unavoidable here.
+globalRetryCountMap :: IORef (Map String (Int,Int))
+globalRetryCountMap = unsafePerformIO (newIORef M.empty)
+{-# NOINLINE globalRetryCountMap #-}
+
+
+-- | For the most general transaction tracking function, 'trackSTMConf', all
+-- settings can be configured using a 'TrackSTMConf' value.
+data TrackSTMConf = TrackSTMConf
+    { tryThreshold      :: Maybe Int
+        -- ^ If the number of retries of one transaction run reaches this
+        -- count, a warning is issued at runtime. If set to @Nothing@, disables the warnings completely.
+    , globalTheshold    :: Maybe Int
+        -- ^ If the total number of retries of one named transaction reaches
+        -- this count, a warning is issued. If set to @Nothing@, disables the
+        -- warnings completely.
+    , extendException   :: Bool
+        -- ^ If this is set, a 'BlockedIndefinitelyOnSTM' exception is replaced
+        -- by a 'BlockedIndefinitelyOnNamedSTM' exception, carrying the name of
+        -- the exception.
+    , warnFunction      :: String -> IO ()
+        -- ^ Function to call when a warning is to be emitted.
+    , warnInSTMFunction :: String -> IO ()
+        -- ^ Function to call when a warning is to be emitted during an STM
+        -- transaction. This is possibly dangerous, see the documentation to
+        -- 'unsafeIOToSTM', but can be useful to detect transactions that keep
+        -- retrying forever.
+    }
+
+-- | The default settings are:
+--
+-- > defaultTrackSTMConf = TrackSTMConf
+-- >    { tryThreshold =      Just 10
+-- >    , globalTheshold =    Just 3000
+-- >    , exception =         True
+-- >    , warnFunction =      hPutStrLn stderr
+-- >    , warnInSTMFunction = \_ -> return ()
+-- >    }
+defaultTrackSTMConf :: TrackSTMConf
+defaultTrackSTMConf = TrackSTMConf
+    { tryThreshold = Just 10
+    , globalTheshold = Just 3000
+    , extendException = True
+    , warnFunction = hPutStrLn stderr
+    , warnInSTMFunction = \_ -> return ()
+    }
+
+-- | A drop-in replacement for 'atomically'. The statistics will list this, and
+-- all other unnamed transactions, as \"@_anonymous_@\" and
+-- 'BlockedIndefinitelyOnSTM' exceptions will not be replaced.
+-- See below for variants that give more control over the statistics and
+-- generated warnings.
+trackSTM :: STM a -> IO a
+trackSTM = trackSTMConf defaultTrackSTMConf { extendException = False } "_anonymous_"
+
+-- | Run 'atomically' and collect the retry statistics under the given name and using the default configuration, 'defaultTrackSTMConf'.
+trackNamedSTM :: String -> STM a -> IO a
+trackNamedSTM = trackSTMConf defaultTrackSTMConf
+
+-- | Run 'atomically' and collect the retry statistics under the given name,
+-- while issuing warnings when the configured thresholds are exceeded.
+trackSTMConf :: TrackSTMConf -> String -> STM a -> IO a
+trackSTMConf (TrackSTMConf {..}) name txm = do
+    counter <- newIORef 0
+    let wrappedTx =
+            do  unsafeIOToSTM $ do
+                    i <- atomicModifyIORef' counter incCounter
+                    when (warnPred i) $
+                        warnInSTMFunction $ msgPrefix ++ " reached try count of " ++ show i
+                txm
+    res <- if extendException
+          then STM.atomically wrappedTx
+              `catch` (\(_::BlockedIndefinitelyOnSTM) ->
+                       throwIO (BlockedIndefinitelyOnNamedSTM name))
+          else STM.atomically wrappedTx
+    i <- readIORef counter
+    doMB tryThreshold $ \threshold ->
+       when (i > threshold) $
+            warnFunction $ msgPrefix ++ " finished after " ++ show (i-1) ++ " retries"
+    incGlobalRetryCount (i - 1)
+    return res
+  where
+    doMB Nothing _  = return ()
+    doMB (Just x) m = m x
+    incCounter i = let j = i + 1 in (j, j)
+    warnPred j = case tryThreshold of
+        Nothing -> False
+        Just n  -> j >= 2*n && (j >= 4 * n || j `mod` (2 * n) == 0)
+    msgPrefix = "STM transaction " ++ name
+    incGlobalRetryCount i = do
+        (k,k') <- atomicModifyIORef' globalRetryCountMap $ \m ->
+                let (oldVal, m') = M.insertLookupWithKey
+                                    (\_ (a1,b1) (a2,b2) -> ((,) $! a1+a2) $! b1+b2)
+                                    name
+                                    (1,i)
+                                    m
+                in (m', let j = maybe 0 snd oldVal in (j,j+i))
+        doMB globalTheshold $ \globalRetryThreshold ->
+            when (k `div` globalRetryThreshold /= k' `div` globalRetryThreshold) $
+                warnFunction $ msgPrefix ++ " reached global retry count of " ++ show k'
+
+-- | If 'extendException' is set (which is the case with 'trackNamedSTM'), an
+-- occurrence of 'BlockedIndefinitelyOnSTM' is replaced by
+-- 'BlockedIndefinitelyOnNamedSTM', carrying the name of the transaction and
+-- thus giving more helpful error messages.
+newtype BlockedIndefinitelyOnNamedSTM = BlockedIndefinitelyOnNamedSTM String
+    deriving (Typeable)
+
+instance Show BlockedIndefinitelyOnNamedSTM where
+    showsPrec _ (BlockedIndefinitelyOnNamedSTM name) =
+        showString $ "thread blocked indefinitely in STM transaction" ++ name
+
+instance Exception BlockedIndefinitelyOnNamedSTM
+
+
+
+-- | Fetches the current transaction statistics data.
+--
+-- The map maps transaction names to counts of transaction commits and
+-- transaction retries.
+getSTMStats = readIORef globalRetryCountMap
+
+-- | Dumps the current transaction statistics data to 'System.IO.stderr'.
+dumpSTMStats = do
+    stats <- getSTMStats
+    time <- show <$> getCurrentTime
+    hPutStrLn stderr $ "STM transaction statistics (" ++ time ++ "):"
+    sequence_ $
+        hPrintf stderr "%-22s %10s %10s %10s\n" "Transaction" "Commits" "Retries" "Ratio" :
+        [ hPrintf stderr "%-22s %10d %10d %10.2f\n" name commits retries ratio
+        | (name,(commits,retries)) <- M.toList stats
+        , commits > 0 -- safeguard
+        , let ratio = fromIntegral retries / fromIntegral commits :: Double
+        ]
+
+
+#endif
+

hls-graph/src/Development/IDE/Graph/Database.hs

@@ -12,8 +12,7 @@ module Development.IDE.Graph.Database(
     shakeGetDirtySet,
     shakeGetCleanKeys
     ,shakeGetBuildEdges) where
-import           Control.Concurrent.STM                  (atomically,
-                                                          readTVarIO)
+import           Control.Concurrent.STM.Stats            (readTVarIO)
 import           Data.Dynamic
 import           Data.Maybe
 import           Development.IDE.Graph.Classes           ()
@@ -57,14 +56,15 @@ shakeGetBuildStep (ShakeDatabase _ _ db) = do
 unvoid :: Functor m => m () -> m a
 unvoid = fmap undefined
 
+-- | Assumes that the database is not running a build
 shakeRunDatabaseForKeys
     :: Maybe [Key]
       -- ^ Set of keys changed since last run. 'Nothing' means everything has changed
     -> ShakeDatabase
     -> [Action a]
     -> IO ([a], [IO ()])
 shakeRunDatabaseForKeys keysChanged (ShakeDatabase lenAs1 as1 db) as2 = do
-    atomically $ incDatabase db keysChanged
+    incDatabase db keysChanged
     as <- fmap (drop lenAs1) $ runActions db $ map unvoid as1 ++ as2
     return (as, [])
 

hls-graph/src/Development/IDE/Graph/Internal/Database.hs

@@ -16,7 +16,8 @@ module Development.IDE.Graph.Internal.Database (newDatabase, incDatabase, build,
 
 import           Control.Concurrent.Async
 import           Control.Concurrent.Extra
-import           Control.Concurrent.STM               (STM, atomically,
+import           Control.Concurrent.STM.Stats         (STM, atomically,
+                                                       atomicallyNamed,
                                                        modifyTVar', newTVarIO,
                                                        readTVarIO)
 import           Control.Exception
@@ -49,20 +50,24 @@ newDatabase databaseExtra databaseRules = do
     databaseValues <- atomically SMap.new
     pure Database{..}
 
--- | Increment the step and mark dirty
-incDatabase :: Database -> Maybe [Key] -> STM ()
+-- | Increment the step and mark dirty.
+--   Assumes that the database is not running a build
+incDatabase :: Database -> Maybe [Key] -> IO ()
 -- only some keys are dirty
 incDatabase db (Just kk) = do
-    modifyTVar'  (databaseStep db) $ \(Step i) -> Step $ i + 1
+    atomicallyNamed "incDatabase" $ modifyTVar'  (databaseStep db) $ \(Step i) -> Step $ i + 1
     transitiveDirtyKeys <- transitiveDirtySet db kk
     for_ transitiveDirtyKeys $ \k ->
-        SMap.focus updateDirty k (databaseValues db)
+        -- Updating all the keys atomically is not necessary
+        -- since we assume that no build is mutating the db.
+        -- Therefore run one transaction per key to minimise contention.
+        atomicallyNamed "incDatabase" $ SMap.focus updateDirty k (databaseValues db)
 
 -- all keys are dirty
 incDatabase db Nothing = do
-    modifyTVar'  (databaseStep db) $ \(Step i) -> Step $ i + 1
+    atomically $ modifyTVar'  (databaseStep db) $ \(Step i) -> Step $ i + 1
     let list = SMap.listT (databaseValues db)
-    flip ListT.traverse_ list $ \(k,_) -> do
+    atomicallyNamed "incDatabase - all " $ flip ListT.traverse_ list $ \(k,_) ->
         SMap.focus updateDirty k (databaseValues db)
 
 updateDirty :: Monad m => Focus.Focus KeyDetails m ()
@@ -93,7 +98,10 @@ builder db@Database{..} keys = withRunInIO $ \(RunInIO run) -> do
     -- Things that I need to force before my results are ready
     toForce <- liftIO $ newTVarIO []
     current <- liftIO $ readTVarIO databaseStep
-    results <- liftIO $ atomically $ for keys $ \id -> do
+    results <- liftIO $ for keys $ \id ->
+        -- Updating the status of all the dependencies atomically is not necessary.
+        -- Therefore, run one transaction per dep. to avoid contention
+        atomicallyNamed "builder" $ do
             -- Spawn the id if needed
             status <- SMap.lookup id databaseValues
             val <- case viewDirty current $ maybe (Dirty Nothing) keyStatus status of
@@ -165,7 +173,7 @@ compute db@Database{..} key mode result = do
                     (getResultDepsDefault [] previousDeps)
                     (HSet.fromList deps)
         _ -> pure ()
-    atomically $ SMap.focus (updateStatus $ Clean res) key databaseValues
+    atomicallyNamed "compute" $ SMap.focus (updateStatus $ Clean res) key databaseValues
     pure res
 
 updateStatus :: Monad m => Status -> Focus.Focus KeyDetails m ()
@@ -214,7 +222,7 @@ updateReverseDeps
     -> [Key] -- ^ Previous direct dependencies of Id
     -> HashSet Key -- ^ Current direct dependencies of Id
     -> IO ()
-updateReverseDeps myId db prev new = uninterruptibleMask_ $ atomically $ do
+updateReverseDeps myId db prev new = uninterruptibleMask_ $ do
     forM_ prev $ \d ->
         unless (d `HSet.member` new) $
             doOne (HSet.delete myId) d
@@ -223,20 +231,23 @@ updateReverseDeps myId db prev new = uninterruptibleMask_ $ atomically $ do
     where
         alterRDeps f =
             Focus.adjust (onKeyReverseDeps f)
-        doOne f id =
+        -- updating all the reverse deps atomically is not needed.
+        -- Therefore, run individual transactions for each update
+        -- in order to avoid contention
+        doOne f id = atomicallyNamed "updateReverseDeps" $
             SMap.focus (alterRDeps f) id (databaseValues db)
 
 getReverseDependencies :: Database -> Key -> STM (Maybe (HashSet Key))
 getReverseDependencies db = (fmap.fmap) keyReverseDeps  . flip SMap.lookup (databaseValues db)
 
-transitiveDirtySet :: Foldable t => Database -> t Key -> STM (HashSet Key)
+transitiveDirtySet :: Foldable t => Database -> t Key -> IO (HashSet Key)
 transitiveDirtySet database = flip State.execStateT HSet.empty . traverse_ loop
   where
     loop x = do
         seen <- State.get
         if x `HSet.member` seen then pure () else do
             State.put (HSet.insert x seen)
-            next <- lift $ getReverseDependencies database x
+            next <- lift $ atomically $ getReverseDependencies database x
             traverse_ loop (maybe mempty HSet.toList next)
 
 -- | IO extended to track created asyncs to clean them up when the thread is killed,

hls-graph/src/Development/IDE/Graph/Internal/Profile.hs

@@ -7,7 +7,7 @@
 
 module Development.IDE.Graph.Internal.Profile (writeProfile) where
 
-import           Control.Concurrent.STM                  (readTVarIO)
+import           Control.Concurrent.STM.Stats            (readTVarIO)
 import           Data.Bifunctor
 import qualified Data.ByteString.Lazy.Char8              as LBS
 import           Data.Char

hls-graph/src/Development/IDE/Graph/Internal/Types.hs

@@ -14,7 +14,7 @@ module Development.IDE.Graph.Internal.Types where
 import           Control.Applicative
 import           Control.Monad.Catch
 -- Needed in GHC 8.6.5
-import           Control.Concurrent.STM        (TVar, atomically)
+import           Control.Concurrent.STM.Stats  (TVar, atomically)
 import           Control.Monad.Fail
 import           Control.Monad.IO.Class
 import           Control.Monad.Trans.Reader