Skip to content

Commit 06e3b8f

Browse files
soulomoonsoulomoon
committed
add workerQueue
1 parent 82da337 commit 06e3b8f

File tree

6 files changed

+58
-22
lines changed

6 files changed

+58
-22
lines changed

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -94,7 +94,6 @@ import Data.Void
9494

9595
import Control.Concurrent.STM.Stats (atomically, modifyTVar',
9696
readTVar, writeTVar)
97-
import Control.Concurrent.STM.TQueue
9897
import Control.DeepSeq
9998
import Control.Exception (evaluate)
10099
import Control.Monad.IO.Unlift (MonadUnliftIO)
@@ -105,7 +104,8 @@ import Data.HashSet (HashSet)
105104
import qualified Data.HashSet as Set
106105
import Database.SQLite.Simple
107106
import Development.IDE.Core.Tracing (withTrace)
108-
import Development.IDE.Core.WorkerThread (awaitRunInThread,
107+
import Development.IDE.Core.WorkerThread (WorkerQueue,
108+
awaitRunInThread,
109109
withWorkerQueue)
110110
import Development.IDE.Session.Diagnostics (renderCradleError)
111111
import Development.IDE.Types.Shake (WithHieDb,
@@ -438,7 +438,7 @@ getHieDbLoc dir = do
438438
-- components mapping to the same hie.yaml file are mapped to the same
439439
-- HscEnv which is updated as new components are discovered.
440440

441-
loadSessionWithOptions :: Recorder (WithPriority Log) -> SessionLoadingOptions -> FilePath -> TQueue (IO ()) -> IO (Action IdeGhcSession)
441+
loadSessionWithOptions :: Recorder (WithPriority Log) -> SessionLoadingOptions -> FilePath -> WorkerQueue (IO ()) -> IO (Action IdeGhcSession)
442442
loadSessionWithOptions recorder SessionLoadingOptions{..} rootDir que = do
443443
let toAbsolutePath = toAbsolute rootDir -- see Note [Root Directory]
444444
cradle_files <- newIORef []

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -129,6 +129,7 @@ import GHC.Driver.Config.CoreToStg.Prep
129129
#if MIN_VERSION_ghc(9,7,0)
130130
import Data.Foldable (toList)
131131
import GHC.Unit.Module.Warnings
132+
import Development.IDE.Core.WorkerThread (writeWorkerQueue)
132133
#else
133134
import Development.IDE.Core.FileStore (shareFilePath)
134135
#endif
@@ -899,7 +900,7 @@ indexHieFile se mod_summary srcPath !hash hf = do
899900
-- hiedb doesn't use the Haskell src, so we clear it to avoid unnecessarily keeping it around
900901
let !hf' = hf{hie_hs_src = mempty}
901902
modifyTVar' indexPending $ HashMap.insert srcPath hash
902-
writeTQueue indexQueue $ \withHieDb -> do
903+
writeWorkerQueue indexQueue $ \withHieDb -> do
903904
-- We are now in the worker thread
904905
-- Check if a newer index of this file has been scheduled, and if so skip this one
905906
newerScheduled <- atomically $ do

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@ module Development.IDE.Core.FileStore(
2222
) where
2323

2424
import Control.Concurrent.STM.Stats (STM, atomically)
25-
import Control.Concurrent.STM.TQueue (writeTQueue)
2625
import Control.Exception
2726
import Control.Monad.Extra
2827
import Control.Monad.IO.Class
@@ -40,6 +39,7 @@ import Development.IDE.Core.IdeConfiguration (isWorkspaceFile)
4039
import Development.IDE.Core.RuleTypes
4140
import Development.IDE.Core.Shake hiding (Log)
4241
import qualified Development.IDE.Core.Shake as Shake
42+
import Development.IDE.Core.WorkerThread (writeWorkerQueue)
4343
import Development.IDE.GHC.Orphans ()
4444
import Development.IDE.Graph
4545
import Development.IDE.Import.DependencyInformation
@@ -247,7 +247,7 @@ typecheckParentsAction recorder nfp = do
247247
setSomethingModified :: VFSModified -> IdeState -> String -> IO [Key] -> IO ()
248248
setSomethingModified vfs state reason actionBetweenSession = do
249249
-- Update database to remove any files that might have been renamed/deleted
250-
atomically $ writeTQueue (indexQueue $ hiedbWriter $ shakeExtras state) (\withHieDb -> withHieDb deleteMissingRealFiles)
250+
atomically $ writeWorkerQueue (indexQueue $ hiedbWriter $ shakeExtras state) (\withHieDb -> withHieDb deleteMissingRealFiles)
251251
void $ restartShakeSession (shakeExtras state) vfs reason [] actionBetweenSession
252252

253253
registerFileWatches :: [String] -> LSP.LspT Config IO Bool

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

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -262,12 +262,12 @@ data HieDbWriter
262262
-- | Actions to queue up on the index worker thread
263263
-- The inner `(HieDb -> IO ()) -> IO ()` wraps `HieDb -> IO ()`
264264
-- with (currently) retry functionality
265-
type IndexQueue = TQueue (((HieDb -> IO ()) -> IO ()) -> IO ())
265+
type IndexQueue = WorkerQueue (((HieDb -> IO ()) -> IO ()) -> IO ())
266266

267267
data ThreadQueue = ThreadQueue {
268268
tIndexQueue :: IndexQueue
269-
, tRestartQueue :: TQueue (IO ())
270-
, tLoaderQueue :: TQueue (IO ())
269+
, tRestartQueue :: WorkerQueue (IO ())
270+
, tLoaderQueue :: WorkerQueue (IO ())
271271
}
272272

273273
-- Note [Semantic Tokens Cache Location]
@@ -342,9 +342,9 @@ data ShakeExtras = ShakeExtras
342342
-- ^ Default HLS config, only relevant if the client does not provide any Config
343343
, dirtyKeys :: TVar KeySet
344344
-- ^ Set of dirty rule keys since the last Shake run
345-
, restartQueue :: TQueue (IO ())
345+
, restartQueue :: WorkerQueue (IO ())
346346
-- ^ Queue of restart actions to be run.
347-
, loaderQueue :: TQueue (IO ())
347+
, loaderQueue :: WorkerQueue (IO ())
348348
-- ^ Queue of loader actions to be run.
349349
}
350350

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

Lines changed: 43 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -7,15 +7,17 @@ Description : This module provides an API for managing worker threads in the IDE
77
see Note [Serializing runs in separate thread]
88
-}
99
module Development.IDE.Core.WorkerThread
10-
(withWorkerQueue, awaitRunInThread)
10+
(withWorkerQueue, awaitRunInThread, withWorkerQueueOfOne, WorkerQueue, writeWorkerQueue)
1111
where
1212

1313
import Control.Concurrent.Async (withAsync)
1414
import Control.Concurrent.STM
1515
import Control.Concurrent.Strict (newBarrier, signalBarrier,
1616
waitBarrier)
17+
import Control.Exception (finally)
1718
import Control.Monad (forever)
1819
import Control.Monad.Cont (ContT (ContT))
20+
import Control.Monad.IO.Class (liftIO)
1921

2022
{-
2123
Note [Serializing runs in separate thread]
@@ -28,27 +30,59 @@ Originally we used various ways to implement this, but it was hard to maintain a
2830
Moreover, we can not stop these threads uniformly when we are shutting down the server.
2931
-}
3032

31-
-- | 'withWorkerQueue' creates a new 'TQueue', and launches a worker
33+
data WorkerQueue a = WorkerQueueOfOne (TMVar a) | WorkerQueueOfMany (TQueue a)
34+
35+
writeWorkerQueue :: WorkerQueue a -> a -> STM ()
36+
writeWorkerQueue (WorkerQueueOfOne tvar) action = putTMVar tvar action
37+
writeWorkerQueue (WorkerQueueOfMany tqueue) action = writeTQueue tqueue action
38+
39+
newWorkerQueue :: STM (WorkerQueue a)
40+
newWorkerQueue = WorkerQueueOfMany <$> newTQueue
41+
42+
newWorkerQueueOfOne :: STM (WorkerQueue a)
43+
newWorkerQueueOfOne = WorkerQueueOfOne <$> newEmptyTMVar
44+
45+
46+
-- | 'withWorkerQueue' creates a new 'WorkerQueue', and launches a worker
3247
-- thread which polls the queue for requests and runs the given worker
3348
-- function on them.
34-
withWorkerQueue :: (t -> IO a) -> ContT () IO (TQueue t)
35-
withWorkerQueue workerAction = ContT $ \mainAction -> do
36-
q <- newTQueueIO
49+
withWorkerQueue :: (t -> IO a) -> ContT () IO (WorkerQueue t)
50+
withWorkerQueue workerAction = do
51+
q <- liftIO $ atomically newWorkerQueue
52+
runWorkerQueue q workerAction
53+
54+
-- | 'withWorkerQueueOfOne' creates a new 'WorkerQueue' that only allows one action to be queued at a time.
55+
-- and one action can only be queued after the previous action has been done.
56+
-- this is useful when we want to cancel the action waiting in the queue, if it's thread is cancelled.
57+
-- e.g. session loading in session loader. When a shake session is restarted, we want to cancel the previous pending session loading.
58+
withWorkerQueueOfOne :: (t -> IO a) -> ContT () IO (WorkerQueue t)
59+
withWorkerQueueOfOne workerAction = do
60+
q <- liftIO $ atomically newWorkerQueueOfOne
61+
runWorkerQueue q workerAction
62+
63+
runWorkerQueue :: WorkerQueue t -> (t -> IO a) -> ContT () IO (WorkerQueue t)
64+
runWorkerQueue q workerAction = ContT $ \mainAction -> do
3765
withAsync (writerThread q) $ \_ -> mainAction q
3866
where
3967
writerThread q =
4068
forever $ do
41-
l <- atomically $ readTQueue q
42-
workerAction l
69+
case q of
70+
-- only remove the action from the queue after it has been run if it is a one-shot queue
71+
WorkerQueueOfOne tvar -> do
72+
l <- atomically $ readTMVar tvar
73+
workerAction l `finally` atomically (takeTMVar tvar)
74+
WorkerQueueOfMany q -> do
75+
l <- atomically $ readTQueue q
76+
workerAction l
4377

4478
-- | 'awaitRunInThread' queues up an 'IO' action to be run by a worker thread,
4579
-- and then blocks until the result is computed.
46-
awaitRunInThread :: TQueue (IO ()) -> IO result -> IO result
80+
awaitRunInThread :: WorkerQueue (IO ()) -> IO result -> IO result
4781
awaitRunInThread q act = do
4882
-- Take an action from TQueue, run it and
4983
-- use barrier to wait for the result
5084
barrier <- newBarrier
51-
atomically $ writeTQueue q $ do
85+
atomically $ writeWorkerQueue q $ do
5286
res <- act
5387
signalBarrier barrier res
5488
waitBarrier barrier

‎ghcide/src/Development/IDE/LSP/LanguageServer.hs‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,8 @@ import Control.Monad.Trans.Cont (evalContT)
3939
import Development.IDE.Core.IdeConfiguration
4040
import Development.IDE.Core.Shake hiding (Log)
4141
import Development.IDE.Core.Tracing
42-
import Development.IDE.Core.WorkerThread (withWorkerQueue)
42+
import Development.IDE.Core.WorkerThread (withWorkerQueue,
43+
withWorkerQueueOfOne)
4344
import qualified Development.IDE.Session as Session
4445
import Development.IDE.Types.Shake (WithHieDb,
4546
WithHieDbShield (..))
@@ -261,7 +262,7 @@ handleInit recorder defaultRoot getHieDbLoc getIdeState lifetime exitClientMsg c
261262
runWithWorkerThreads :: Recorder (WithPriority Session.Log) -> FilePath -> (WithHieDb -> ThreadQueue -> IO ()) -> IO ()
262263
runWithWorkerThreads recorder dbLoc f = evalContT $ do
263264
sessionRestartTQueue <- withWorkerQueue id
264-
sessionLoaderTQueue <- withWorkerQueue id
265+
sessionLoaderTQueue <- withWorkerQueueOfOne id
265266
(WithHieDbShield hiedb, threadQueue) <- runWithDb recorder dbLoc
266267
liftIO $ f hiedb (ThreadQueue threadQueue sessionRestartTQueue sessionLoaderTQueue)
267268

0 commit comments

Comments
 (0)