Skip to content

Commit 82da337

Browse files
soulomoonmichaelpj
soulomoon
and
michaelpj
authored
Unify critical session running in hls (#4256)
* add thread to do shake restart * run session loader in thread --------- Co-authored-by: Michael Peyton Jones <[email protected]>
1 parent 7563439 commit 82da337

File tree

8 files changed

+159
-78
lines changed

8 files changed

+159
-78
lines changed

‎ghcide/ghcide.cabal

+1
Original file line numberDiff line numberDiff line change
@@ -148,6 +148,7 @@ library
148148
Development.IDE.Core.Shake
149149
Development.IDE.Core.Tracing
150150
Development.IDE.Core.UseStale
151+
Development.IDE.Core.WorkerThread
151152
Development.IDE.GHC.Compat
152153
Development.IDE.GHC.Compat.Core
153154
Development.IDE.GHC.Compat.CmdLine

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

+22-29
Original file line numberDiff line numberDiff line change
@@ -7,21 +7,19 @@ The logic for setting up a ghcide session by tapping into hie-bios.
77
module Development.IDE.Session
88
(SessionLoadingOptions(..)
99
,CacheDirs(..)
10-
,loadSession
1110
,loadSessionWithOptions
1211
,setInitialDynFlags
1312
,getHieDbLoc
14-
,runWithDb
1513
,retryOnSqliteBusy
1614
,retryOnException
1715
,Log(..)
16+
,runWithDb
1817
) where
1918

2019
-- Unfortunately, we cannot use loadSession with ghc-lib since hie-bios uses
2120
-- the real GHC library and the types are incompatible. Furthermore, when
2221
-- building with ghc-lib we need to make this Haskell agnostic, so no hie-bios!
2322

24-
import Control.Concurrent.Async
2523
import Control.Concurrent.Strict
2624
import Control.Exception.Safe as Safe
2725
import Control.Monad
@@ -100,14 +98,19 @@ import Control.Concurrent.STM.TQueue
10098
import Control.DeepSeq
10199
import Control.Exception (evaluate)
102100
import Control.Monad.IO.Unlift (MonadUnliftIO)
101+
import Control.Monad.Trans.Cont (ContT (ContT, runContT))
103102
import Data.Foldable (for_)
104103
import Data.HashMap.Strict (HashMap)
105104
import Data.HashSet (HashSet)
106105
import qualified Data.HashSet as Set
107106
import Database.SQLite.Simple
108107
import Development.IDE.Core.Tracing (withTrace)
108+
import Development.IDE.Core.WorkerThread (awaitRunInThread,
109+
withWorkerQueue)
109110
import Development.IDE.Session.Diagnostics (renderCradleError)
110-
import Development.IDE.Types.Shake (WithHieDb, toNoFileKey)
111+
import Development.IDE.Types.Shake (WithHieDb,
112+
WithHieDbShield (..),
113+
toNoFileKey)
111114
import HieDb.Create
112115
import HieDb.Types
113116
import HieDb.Utils
@@ -375,8 +378,10 @@ makeWithHieDbRetryable recorder rng hieDb f =
375378
-- writing. Actions are picked off one by one from the `HieWriterChan` and executed in serial
376379
-- by a worker thread using a dedicated database connection.
377380
-- This is done in order to serialize writes to the database, or else SQLite becomes unhappy
378-
runWithDb :: Recorder (WithPriority Log) -> FilePath -> (WithHieDb -> IndexQueue -> IO ()) -> IO ()
379-
runWithDb recorder fp k = do
381+
--
382+
-- Also see Note [Serializing runs in separate thread]
383+
runWithDb :: Recorder (WithPriority Log) -> FilePath -> ContT () IO (WithHieDbShield, IndexQueue)
384+
runWithDb recorder fp = ContT $ \k -> do
380385
-- use non-deterministic seed because maybe multiple HLS start at same time
381386
-- and send bursts of requests
382387
rng <- Random.newStdGen
@@ -394,18 +399,15 @@ runWithDb recorder fp k = do
394399
withWriteDbRetryable = makeWithHieDbRetryable recorder rng writedb
395400
withWriteDbRetryable initConn
396401

397-
chan <- newTQueueIO
398402

399-
withAsync (writerThread withWriteDbRetryable chan) $ \_ -> do
400-
withHieDb fp (\readDb -> k (makeWithHieDbRetryable recorder rng readDb) chan)
403+
-- Clear the index of any files that might have been deleted since the last run
404+
_ <- withWriteDbRetryable deleteMissingRealFiles
405+
_ <- withWriteDbRetryable garbageCollectTypeNames
406+
407+
runContT (withWorkerQueue (writer withWriteDbRetryable)) $ \chan ->
408+
withHieDb fp (\readDb -> k (WithHieDbShield $ makeWithHieDbRetryable recorder rng readDb, chan))
401409
where
402-
writerThread :: WithHieDb -> IndexQueue -> IO ()
403-
writerThread withHieDbRetryable chan = do
404-
-- Clear the index of any files that might have been deleted since the last run
405-
_ <- withHieDbRetryable deleteMissingRealFiles
406-
_ <- withHieDbRetryable garbageCollectTypeNames
407-
forever $ do
408-
l <- atomically $ readTQueue chan
410+
writer withHieDbRetryable l = do
409411
-- TODO: probably should let exceptions be caught/logged/handled by top level handler
410412
l withHieDbRetryable
411413
`Safe.catch` \e@SQLError{} -> do
@@ -435,11 +437,9 @@ getHieDbLoc dir = do
435437
-- This is the key function which implements multi-component support. All
436438
-- components mapping to the same hie.yaml file are mapped to the same
437439
-- HscEnv which is updated as new components are discovered.
438-
loadSession :: Recorder (WithPriority Log) -> FilePath -> IO (Action IdeGhcSession)
439-
loadSession recorder = loadSessionWithOptions recorder def
440440

441-
loadSessionWithOptions :: Recorder (WithPriority Log) -> SessionLoadingOptions -> FilePath -> IO (Action IdeGhcSession)
442-
loadSessionWithOptions recorder SessionLoadingOptions{..} rootDir = do
441+
loadSessionWithOptions :: Recorder (WithPriority Log) -> SessionLoadingOptions -> FilePath -> TQueue (IO ()) -> IO (Action IdeGhcSession)
442+
loadSessionWithOptions recorder SessionLoadingOptions{..} rootDir que = do
443443
let toAbsolutePath = toAbsolute rootDir -- see Note [Root Directory]
444444
cradle_files <- newIORef []
445445
-- Mapping from hie.yaml file to HscEnv, one per hie.yaml file
@@ -464,9 +464,6 @@ loadSessionWithOptions recorder SessionLoadingOptions{..} rootDir = do
464464
let res' = toAbsolutePath <$> res
465465
return $ normalise <$> res'
466466

467-
dummyAs <- async $ return (error "Uninitialised")
468-
runningCradle <- newVar dummyAs :: IO (Var (Async (IdeResult HscEnvEq,[FilePath])))
469-
470467
return $ do
471468
clientConfig <- getClientConfigAction
472469
extras@ShakeExtras{restartShakeSession, ideNc, knownTargetsVar, lspEnv
@@ -739,12 +736,8 @@ loadSessionWithOptions recorder SessionLoadingOptions{..} rootDir = do
739736
return (([renderPackageSetupException file e], Nothing), maybe [] pure hieYaml)
740737

741738
returnWithVersion $ \file -> do
742-
opts <- join $ mask_ $ modifyVar runningCradle $ \as -> do
743-
-- If the cradle is not finished, then wait for it to finish.
744-
void $ wait as
745-
asyncRes <- async $ getOptions file
746-
return (asyncRes, wait asyncRes)
747-
pure opts
739+
-- see Note [Serializing runs in separate thread]
740+
awaitRunInThread que $ getOptions file
748741

749742
-- | Run the specific cradle on a specific FilePath via hie-bios.
750743
-- This then builds dependencies or whatever based on the cradle, gets the

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

+2-1
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,7 @@ instance Pretty Log where
5353
LogOfInterest msg -> pretty msg
5454
LogFileExists msg -> pretty msg
5555

56+
5657
------------------------------------------------------------
5758
-- Exposed API
5859

@@ -65,7 +66,7 @@ initialise :: Recorder (WithPriority Log)
6566
-> Debouncer LSP.NormalizedUri
6667
-> IdeOptions
6768
-> WithHieDb
68-
-> IndexQueue
69+
-> ThreadQueue
6970
-> Monitoring
7071
-> FilePath -- ^ Root directory see Note [Root Directory]
7172
-> IO IdeState

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

+41-23
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,7 @@ module Development.IDE.Core.Shake(
7373
garbageCollectDirtyKeysOlderThan,
7474
Log(..),
7575
VFSModified(..), getClientConfigAction,
76+
ThreadQueue(..)
7677
) where
7778

7879
import Control.Concurrent.Async
@@ -123,6 +124,7 @@ import Development.IDE.Core.PositionMapping
123124
import Development.IDE.Core.ProgressReporting
124125
import Development.IDE.Core.RuleTypes
125126
import Development.IDE.Core.Tracing
127+
import Development.IDE.Core.WorkerThread
126128
import Development.IDE.GHC.Compat (NameCache,
127129
initNameCache,
128130
knownKeyNames)
@@ -262,6 +264,12 @@ data HieDbWriter
262264
-- with (currently) retry functionality
263265
type IndexQueue = TQueue (((HieDb -> IO ()) -> IO ()) -> IO ())
264266

267+
data ThreadQueue = ThreadQueue {
268+
tIndexQueue :: IndexQueue
269+
, tRestartQueue :: TQueue (IO ())
270+
, tLoaderQueue :: TQueue (IO ())
271+
}
272+
265273
-- Note [Semantic Tokens Cache Location]
266274
-- ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
267275
-- storing semantic tokens cache for each file in shakeExtras might
@@ -334,6 +342,10 @@ data ShakeExtras = ShakeExtras
334342
-- ^ Default HLS config, only relevant if the client does not provide any Config
335343
, dirtyKeys :: TVar KeySet
336344
-- ^ Set of dirty rule keys since the last Shake run
345+
, restartQueue :: TQueue (IO ())
346+
-- ^ Queue of restart actions to be run.
347+
, loaderQueue :: TQueue (IO ())
348+
-- ^ Queue of loader actions to be run.
337349
}
338350

339351
type WithProgressFunc = forall a.
@@ -648,7 +660,7 @@ shakeOpen :: Recorder (WithPriority Log)
648660
-> IdeReportProgress
649661
-> IdeTesting
650662
-> WithHieDb
651-
-> IndexQueue
663+
-> ThreadQueue
652664
-> ShakeOptions
653665
-> Monitoring
654666
-> Rules ()
@@ -658,8 +670,12 @@ shakeOpen :: Recorder (WithPriority Log)
658670
-> IO IdeState
659671
shakeOpen recorder lspEnv defaultConfig idePlugins debouncer
660672
shakeProfileDir (IdeReportProgress reportProgress)
661-
ideTesting@(IdeTesting testing)
662-
withHieDb indexQueue opts monitoring rules rootDir = mdo
673+
ideTesting
674+
withHieDb threadQueue opts monitoring rules rootDir = mdo
675+
-- see Note [Serializing runs in separate thread]
676+
let indexQueue = tIndexQueue threadQueue
677+
restartQueue = tRestartQueue threadQueue
678+
loaderQueue = tLoaderQueue threadQueue
663679

664680
#if MIN_VERSION_ghc(9,3,0)
665681
ideNc <- initNameCache 'r' knownKeyNames
@@ -784,31 +800,33 @@ delayedAction a = do
784800
extras <- ask
785801
liftIO $ shakeEnqueue extras a
786802

803+
787804
-- | Restart the current 'ShakeSession' with the given system actions.
788805
-- Any actions running in the current session will be aborted,
789806
-- but actions added via 'shakeEnqueue' will be requeued.
790807
shakeRestart :: Recorder (WithPriority Log) -> IdeState -> VFSModified -> String -> [DelayedAction ()] -> IO [Key] -> IO ()
791808
shakeRestart recorder IdeState{..} vfs reason acts ioActionBetweenShakeSession =
792-
withMVar'
793-
shakeSession
794-
(\runner -> do
795-
(stopTime,()) <- duration $ logErrorAfter 10 $ cancelShakeSession runner
796-
keys <- ioActionBetweenShakeSession
797-
-- it is every important to update the dirty keys after we enter the critical section
798-
-- see Note [Housekeeping rule cache and dirty key outside of hls-graph]
799-
atomically $ modifyTVar' (dirtyKeys shakeExtras) $ \x -> foldl' (flip insertKeySet) x keys
800-
res <- shakeDatabaseProfile shakeDb
801-
backlog <- readTVarIO $ dirtyKeys shakeExtras
802-
queue <- atomicallyNamed "actionQueue - peek" $ peekInProgress $ actionQueue shakeExtras
803-
804-
-- this log is required by tests
805-
logWith recorder Debug $ LogBuildSessionRestart reason queue backlog stopTime res
806-
)
807-
-- It is crucial to be masked here, otherwise we can get killed
808-
-- between spawning the new thread and updating shakeSession.
809-
-- See https://github.com/haskell/ghcide/issues/79
810-
(\() -> do
811-
(,()) <$> newSession recorder shakeExtras vfs shakeDb acts reason)
809+
void $ awaitRunInThread (restartQueue shakeExtras) $ do
810+
withMVar'
811+
shakeSession
812+
(\runner -> do
813+
(stopTime,()) <- duration $ logErrorAfter 10 $ cancelShakeSession runner
814+
keys <- ioActionBetweenShakeSession
815+
-- it is every important to update the dirty keys after we enter the critical section
816+
-- see Note [Housekeeping rule cache and dirty key outside of hls-graph]
817+
atomically $ modifyTVar' (dirtyKeys shakeExtras) $ \x -> foldl' (flip insertKeySet) x keys
818+
res <- shakeDatabaseProfile shakeDb
819+
backlog <- readTVarIO $ dirtyKeys shakeExtras
820+
queue <- atomicallyNamed "actionQueue - peek" $ peekInProgress $ actionQueue shakeExtras
821+
822+
-- this log is required by tests
823+
logWith recorder Debug $ LogBuildSessionRestart reason queue backlog stopTime res
824+
)
825+
-- It is crucial to be masked here, otherwise we can get killed
826+
-- between spawning the new thread and updating shakeSession.
827+
-- See https://github.com/haskell/ghcide/issues/79
828+
(\() -> do
829+
(,()) <$> newSession recorder shakeExtras vfs shakeDb acts reason)
812830
where
813831
logErrorAfter :: Seconds -> IO () -> IO ()
814832
logErrorAfter seconds action = flip withAsync (const action) $ do

‎ghcide/src/Development/IDE/Core/WorkerThread.hs

+54
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
{-
2+
Module : Development.IDE.Core.WorkerThread
3+
Author : @soulomoon
4+
SPDX-License-Identifier: Apache-2.0
5+
6+
Description : This module provides an API for managing worker threads in the IDE.
7+
see Note [Serializing runs in separate thread]
8+
-}
9+
module Development.IDE.Core.WorkerThread
10+
(withWorkerQueue, awaitRunInThread)
11+
where
12+
13+
import Control.Concurrent.Async (withAsync)
14+
import Control.Concurrent.STM
15+
import Control.Concurrent.Strict (newBarrier, signalBarrier,
16+
waitBarrier)
17+
import Control.Monad (forever)
18+
import Control.Monad.Cont (ContT (ContT))
19+
20+
{-
21+
Note [Serializing runs in separate thread]
22+
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
23+
We often want to take long-running actions using some resource that cannot be shared.
24+
In this instance it is useful to have a queue of jobs to run using the resource.
25+
Like the db writes, session loading in session loader, shake session restarts.
26+
27+
Originally we used various ways to implement this, but it was hard to maintain and error prone.
28+
Moreover, we can not stop these threads uniformly when we are shutting down the server.
29+
-}
30+
31+
-- | 'withWorkerQueue' creates a new 'TQueue', and launches a worker
32+
-- thread which polls the queue for requests and runs the given worker
33+
-- function on them.
34+
withWorkerQueue :: (t -> IO a) -> ContT () IO (TQueue t)
35+
withWorkerQueue workerAction = ContT $ \mainAction -> do
36+
q <- newTQueueIO
37+
withAsync (writerThread q) $ \_ -> mainAction q
38+
where
39+
writerThread q =
40+
forever $ do
41+
l <- atomically $ readTQueue q
42+
workerAction l
43+
44+
-- | 'awaitRunInThread' queues up an 'IO' action to be run by a worker thread,
45+
-- and then blocks until the result is computed.
46+
awaitRunInThread :: TQueue (IO ()) -> IO result -> IO result
47+
awaitRunInThread q act = do
48+
-- Take an action from TQueue, run it and
49+
-- use barrier to wait for the result
50+
barrier <- newBarrier
51+
atomically $ writeTQueue q $ do
52+
res <- act
53+
signalBarrier barrier res
54+
waitBarrier barrier

0 commit comments

Comments
 (0)