Unify critical session running in hls

ghcide/ghcide.cabal

@@ -148,6 +148,7 @@ library
     Development.IDE.Core.Shake
     Development.IDE.Core.Tracing
     Development.IDE.Core.UseStale
+    Development.IDE.Core.Thread
     Development.IDE.GHC.Compat
     Development.IDE.GHC.Compat.Core
     Development.IDE.GHC.Compat.CmdLine

ghcide/session-loader/Development/IDE/Session.hs

@@ -7,21 +7,19 @@
 module Development.IDE.Session
   (SessionLoadingOptions(..)
   ,CacheDirs(..)
-  ,loadSession
   ,loadSessionWithOptions
   ,setInitialDynFlags
   ,getHieDbLoc
-  ,runWithDb
   ,retryOnSqliteBusy
   ,retryOnException
   ,Log(..)
+  ,dbThread
   ) where
 
 -- Unfortunately, we cannot use loadSession with ghc-lib since hie-bios uses
 -- the real GHC library and the types are incompatible. Furthermore, when
 -- building with ghc-lib we need to make this Haskell agnostic, so no hie-bios!
 
-import           Control.Concurrent.Async
 import           Control.Concurrent.Strict
 import           Control.Exception.Safe               as Safe
 import           Control.Monad
@@ -100,14 +98,19 @@
 import           Control.DeepSeq
 import           Control.Exception                    (evaluate)
 import           Control.Monad.IO.Unlift              (MonadUnliftIO)
+import           Control.Monad.Trans.Cont             (ContT (ContT), evalContT)
 import           Data.Foldable                        (for_)
 import           Data.HashMap.Strict                  (HashMap)
 import           Data.HashSet                         (HashSet)
 import qualified Data.HashSet                         as Set
 import           Database.SQLite.Simple
+import           Development.IDE.Core.Thread          (ThreadRun (..),
+                                                       blockRunInThread)
 import           Development.IDE.Core.Tracing         (withTrace)
 import           Development.IDE.Session.Diagnostics  (renderCradleError)
-import           Development.IDE.Types.Shake          (WithHieDb, toNoFileKey)
+import           Development.IDE.Types.Shake          (WithHieDb,
+                                                       WithHieDbShield (..),
+                                                       toNoFileKey)
 import           HieDb.Create
 import           HieDb.Types
 import           HieDb.Utils
@@ -375,44 +378,35 @@
 -- writing. Actions are picked off one by one from the `HieWriterChan` and executed in serial
 -- by a worker thread using a dedicated database connection.
 -- This is done in order to serialize writes to the database, or else SQLite becomes unhappy
-runWithDb :: Recorder (WithPriority Log) -> FilePath -> (WithHieDb -> IndexQueue -> IO ()) -> IO ()
-runWithDb recorder fp k = do
-  -- use non-deterministic seed because maybe multiple HLS start at same time
-  -- and send bursts of requests
-  rng <- Random.newStdGen
-  -- Delete the database if it has an incompatible schema version
-  retryOnSqliteBusy
-    recorder
-    rng
-    (withHieDb fp (const $ pure ()) `Safe.catch` \IncompatibleSchemaVersion{} -> removeFile fp)
-
-  withHieDb fp $ \writedb -> do
-    -- the type signature is necessary to avoid concretizing the tyvar
-    -- e.g. `withWriteDbRetryable initConn` without type signature will
-    -- instantiate tyvar `a` to `()`
-    let withWriteDbRetryable :: WithHieDb
-        withWriteDbRetryable = makeWithHieDbRetryable recorder rng writedb
-    withWriteDbRetryable initConn
-
-    chan <- newTQueueIO
-
-    withAsync (writerThread withWriteDbRetryable chan) $ \_ -> do
-      withHieDb fp (\readDb -> k (makeWithHieDbRetryable recorder rng readDb) chan)
-  where
-    writerThread :: WithHieDb -> IndexQueue -> IO ()
-    writerThread withHieDbRetryable chan = do
-      -- Clear the index of any files that might have been deleted since the last run
-      _ <- withHieDbRetryable deleteMissingRealFiles
-      _ <- withHieDbRetryable garbageCollectTypeNames
-      forever $ do
-        l <- atomically $ readTQueue chan
-        -- TODO: probably should let exceptions be caught/logged/handled by top level handler
-        l withHieDbRetryable
+--
+-- see Note [Serializing runs in separate thread]
+dbThread ::
+        ThreadRun
+            (Recorder (WithPriority Log), FilePath)
+            WithHieDbShield -- ^ writer resource
+            WithHieDbShield -- ^ reader resource
+            (((HieDb -> IO a) -> IO a) -> IO ())
+dbThread = ThreadRun {
+    tWorker = \(recorder, _fp) (WithHieDbShield withWriter) l ->  l withWriter
           `Safe.catch` \e@SQLError{} -> do
             logWith recorder Error $ LogHieDbWriterThreadSQLiteError e
           `Safe.catchAny` \f -> do
             logWith recorder Error $ LogHieDbWriterThreadException f
-
+    ,
+    tRunWithResource = \(recorder, fp) f -> do
+        rng <- Random.newStdGen
+        retryOnSqliteBusy
+            recorder
+            rng
+            (withHieDb fp (const $ pure ()) `Safe.catch` \IncompatibleSchemaVersion{} -> removeFile fp)
+        evalContT $ do
+            writeDb <- ContT $ withHieDb fp
+            readDb <- ContT $ withHieDb fp
+            let withWriteDbRetryable :: WithHieDb
+                withWriteDbRetryable = makeWithHieDbRetryable recorder rng writeDb
+            liftIO $ withWriteDbRetryable initConn
+            liftIO $ f (WithHieDbShield withWriteDbRetryable) (WithHieDbShield (makeWithHieDbRetryable recorder rng readDb))
+}
 
 getHieDbLoc :: FilePath -> IO FilePath
 getHieDbLoc dir = do
@@ -435,11 +429,9 @@
 -- This is the key function which implements multi-component support. All
 -- components mapping to the same hie.yaml file are mapped to the same
 -- HscEnv which is updated as new components are discovered.
-loadSession :: Recorder (WithPriority Log) -> FilePath -> IO (Action IdeGhcSession)
-loadSession recorder = loadSessionWithOptions recorder def
 
-loadSessionWithOptions :: Recorder (WithPriority Log) -> SessionLoadingOptions -> FilePath -> IO (Action IdeGhcSession)
-loadSessionWithOptions recorder SessionLoadingOptions{..} rootDir = do
+loadSessionWithOptions :: Recorder (WithPriority Log) -> SessionLoadingOptions -> FilePath -> TQueue (IO ()) -> IO (Action IdeGhcSession)
+loadSessionWithOptions recorder SessionLoadingOptions{..} rootDir que = do
   let toAbsolutePath = toAbsolute rootDir -- see Note [Root Directory]
   cradle_files <- newIORef []
   -- Mapping from hie.yaml file to HscEnv, one per hie.yaml file
@@ -464,9 +456,6 @@
       let res' = toAbsolutePath <$> res
       return $ normalise <$> res'
 
-  dummyAs <- async $ return (error "Uninitialised")
-  runningCradle <- newVar dummyAs :: IO (Var (Async (IdeResult HscEnvEq,[FilePath])))
-
   return $ do
     clientConfig <- getClientConfigAction
     extras@ShakeExtras{restartShakeSession, ideNc, knownTargetsVar, lspEnv
@@ -665,7 +654,7 @@
                  InstallationMismatch{..} ->
                      return (([renderPackageSetupException cfp GhcVersionMismatch{..}], Nothing),[])
                  InstallationChecked _compileTime _ghcLibCheck -> do
                    atomicModifyIORef' cradle_files (\xs -> (cfp:xs,()))
                    session (hieYaml, toNormalizedFilePath' cfp, opts, libDir)
              -- Failure case, either a cradle error or the none cradle
              Left err -> do
@@ -739,12 +728,8 @@
                 return (([renderPackageSetupException file e], Nothing), maybe [] pure hieYaml)
 
     returnWithVersion $ \file -> do
-      opts <- join $ mask_ $ modifyVar runningCradle $ \as -> do
-        -- If the cradle is not finished, then wait for it to finish.
-        void $ wait as
-        asyncRes <- async $ getOptions file
-        return (asyncRes, wait asyncRes)
-      pure opts
+      -- see Note [Serializing runs in separate thread]
+      blockRunInThread que $ getOptions file
 
 -- | Run the specific cradle on a specific FilePath via hie-bios.
 -- This then builds dependencies or whatever based on the cradle, gets the

ghcide/src/Development/IDE/Core/Service.hs

@@ -31,6 +31,7 @@ import           Ide.Plugin.Config
 import qualified Language.LSP.Protocol.Types      as LSP
 import qualified Language.LSP.Server              as LSP
 
+import           Control.Concurrent.STM           (TQueue)
 import           Control.Monad
 import qualified Development.IDE.Core.FileExists  as FileExists
 import qualified Development.IDE.Core.OfInterest  as OfInterest
@@ -53,6 +54,7 @@ instance Pretty Log where
     LogOfInterest msg -> pretty msg
     LogFileExists msg -> pretty msg
 
+
 ------------------------------------------------------------
 -- Exposed API
 
@@ -65,7 +67,7 @@ initialise :: Recorder (WithPriority Log)
            -> Debouncer LSP.NormalizedUri
            -> IdeOptions
            -> WithHieDb
-           -> IndexQueue
+           -> ThreadQueue
            -> Monitoring
            -> FilePath -- ^ Root directory see Note [Root Directory]
            -> IO IdeState

ghcide/src/Development/IDE/Core/Shake.hs

@@ -73,6 +73,7 @@
     garbageCollectDirtyKeysOlderThan,
     Log(..),
     VFSModified(..), getClientConfigAction,
+    ThreadQueue(..)
     ) where
 
 import           Control.Concurrent.Async
@@ -122,8 +123,9 @@
 import           Development.IDE.Core.PositionMapping
 import           Development.IDE.Core.ProgressReporting
 import           Development.IDE.Core.RuleTypes
+import           Development.IDE.Core.Thread
 import           Development.IDE.Core.Tracing
 import           Development.IDE.GHC.Compat             (NameCache,
                                                          initNameCache,
                                                          knownKeyNames)
 import           Development.IDE.GHC.Orphans            ()
@@ -262,6 +264,12 @@
 -- with (currently) retry functionality
 type IndexQueue = TQueue (((HieDb -> IO ()) -> IO ()) -> IO ())
 
+data ThreadQueue = ThreadQueue {
+    tIndexQueue     :: IndexQueue
+    , tRestartQueue :: TQueue (IO ())
+    , tLoaderQueue  :: TQueue (IO ())
+}
+
 -- Note [Semantic Tokens Cache Location]
 -- ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 -- storing semantic tokens cache for each file in shakeExtras might
@@ -334,6 +342,10 @@
       -- ^ Default HLS config, only relevant if the client does not provide any Config
     , dirtyKeys :: TVar KeySet
       -- ^ Set of dirty rule keys since the last Shake run
+    , restartQueue :: TQueue (IO ())
+      -- ^ Queue of restart actions to be run.
+    , loaderQueue :: TQueue (IO ())
+      -- ^ Queue of loader actions to be run.
     }
 
 type WithProgressFunc = forall a.
@@ -648,7 +660,7 @@
           -> IdeReportProgress
           -> IdeTesting
           -> WithHieDb
-          -> IndexQueue
+          -> ThreadQueue
           -> ShakeOptions
           -> Monitoring
           -> Rules ()
@@ -658,8 +670,12 @@
           -> IO IdeState
 shakeOpen recorder lspEnv defaultConfig idePlugins debouncer
   shakeProfileDir (IdeReportProgress reportProgress)
-  ideTesting@(IdeTesting testing)
-  withHieDb indexQueue opts monitoring rules rootDir = mdo
+  ideTesting
+  withHieDb threadQueue opts monitoring rules rootDir = mdo
+    -- see Note [Serializing runs in separate thread]
+    let indexQueue = tIndexQueue threadQueue
+        restartQueue = tRestartQueue threadQueue
+        loaderQueue = tLoaderQueue threadQueue
 
 #if MIN_VERSION_ghc(9,3,0)
     ideNc <- initNameCache 'r' knownKeyNames
@@ -784,31 +800,33 @@
   extras <- ask
   liftIO $ shakeEnqueue extras a
 
+
 -- | Restart the current 'ShakeSession' with the given system actions.
 --   Any actions running in the current session will be aborted,
 --   but actions added via 'shakeEnqueue' will be requeued.
 shakeRestart :: Recorder (WithPriority Log) -> IdeState -> VFSModified -> String -> [DelayedAction ()] -> IO [Key] -> IO ()
 shakeRestart recorder IdeState{..} vfs reason acts ioActionBetweenShakeSession =
-    withMVar'
-        shakeSession
-        (\runner -> do
-              (stopTime,()) <- duration $ logErrorAfter 10 $ cancelShakeSession runner
-              keys <- ioActionBetweenShakeSession
-              -- it is every important to update the dirty keys after we enter the critical section
-              -- see Note [Housekeeping rule cache and dirty key outside of hls-graph]
-              atomically $ modifyTVar' (dirtyKeys shakeExtras) $ \x -> foldl' (flip insertKeySet) x keys
-              res <- shakeDatabaseProfile shakeDb
-              backlog <- readTVarIO $ dirtyKeys shakeExtras
-              queue <- atomicallyNamed "actionQueue - peek" $ peekInProgress $ actionQueue shakeExtras
-
-              -- this log is required by tests
-              logWith recorder Debug $ LogBuildSessionRestart reason queue backlog stopTime res
-        )
-        -- It is crucial to be masked here, otherwise we can get killed
-        -- between spawning the new thread and updating shakeSession.
-        -- See https://github.com/haskell/ghcide/issues/79
-        (\() -> do
-          (,()) <$> newSession recorder shakeExtras vfs shakeDb acts reason)
+    void $ blockRunInThread (restartQueue shakeExtras) $ do
+        withMVar'
+            shakeSession
+            (\runner -> do
+                (stopTime,()) <- duration $ logErrorAfter 10 $ cancelShakeSession runner
+                keys <- ioActionBetweenShakeSession
+                -- it is every important to update the dirty keys after we enter the critical section
+                -- see Note [Housekeeping rule cache and dirty key outside of hls-graph]
+                atomically $ modifyTVar' (dirtyKeys shakeExtras) $ \x -> foldl' (flip insertKeySet) x keys
+                res <- shakeDatabaseProfile shakeDb
+                backlog <- readTVarIO $ dirtyKeys shakeExtras
+                queue <- atomicallyNamed "actionQueue - peek" $ peekInProgress $ actionQueue shakeExtras
+
+                -- this log is required by tests
+                logWith recorder Debug $ LogBuildSessionRestart reason queue backlog stopTime res
+            )
+            -- It is crucial to be masked here, otherwise we can get killed
+            -- between spawning the new thread and updating shakeSession.
+            -- See https://github.com/haskell/ghcide/issues/79
+            (\() -> do
+            (,()) <$> newSession recorder shakeExtras vfs shakeDb acts reason)
     where
         logErrorAfter :: Seconds -> IO () -> IO ()
         logErrorAfter seconds action = flip withAsync (const action) $ do

ghcide/src/Development/IDE/Core/Thread.hs

@@ -0,0 +1,67 @@
+module Development.IDE.Core.Thread
+    ( ThreadRun(..), runWithThread, blockRunInThread)
+ where
+
+import           Control.Concurrent.Async
+import           Control.Concurrent.STM
+import           Control.Concurrent.Strict (newBarrier, signalBarrier,
+                                            waitBarrier)
+import           Control.Monad             (forever)
+import           Control.Monad.Cont        (ContT (ContT))
+
+-- Note [Serializing runs in separate thread]
+-- ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+-- In a lof cases we want to have a separate thread that will serialize the runs of the actions.
+-- Like the db writes, session loading in session loader, shake session restarts.
+--
+-- Originally we used various ways to implement this, but it was hard to maintain and error prone.
+-- Moreover, we can not stop these threads uniformly when we are shutting down the server.
+--
+-- `Development.IDE.Core.Thread` module provides a declarative api to implement this easily.
+-- In `ThreadRun` data type:
+-- * `tRunWithResource`: is used to create the resources needed to perform the long running action.
+-- * `tWorker`: is the action we want to run in separate thread serially.
+--
+-- runWithThread will create a worker thread to run along with the main thread.
+-- runWithThread provides `resource` created by `tRunWithResource` and a `TQueue` to send the actions to run.
+-- The worker thread will serialize the runs of the actions from the TQueue.
+
+
+data ThreadRun input workerResource resource arg = ThreadRun {
+        tRunWithResource ::
+            input -- ^ input of running
+            -> (workerResource -> resource -> IO ()) -- ^ the long running action need to run with resource
+            -> IO (),
+        tWorker -- ^ A single action we want to run in separate thread serially
+            :: input -- ^ input of running
+            -> workerResource -- ^ writer resource
+            -> arg -- ^ argument to run
+            -> IO ()
+}
+
+-- | runWithThread
+-- Run a long running action with a additional running thread
+-- The additional thread will serialize runs of the actions from the TQueue.
+-- Return ContT to run the action
+runWithThread :: ThreadRun input workerResource resource arg -> input -> ContT () IO (resource, TQueue arg)
+runWithThread ThreadRun{..} ip = ContT $ \f -> do
+    tRunWithResource ip $ \w r -> do
+        q <- newTQueueIO
+        withAsync (writerThread w q) $ \_ -> f (r, q)
+    where
+        writerThread r q =
+            forever $ do
+                l <- atomically $ readTQueue q
+                tWorker ip r l
+
+
+-- | blockRunInThread run and wait for the result
+-- Take an action from TQueue, run it and
+-- use barrier to wait for the result
+blockRunInThread :: TQueue (IO ()) -> IO result -> IO result
+blockRunInThread q act = do
+    barrier <- newBarrier
+    atomically $ writeTQueue q $ do
+        res <- act
+        signalBarrier barrier res
+    waitBarrier barrier