From 06e3b8fbce85784bb4f95a9676a7371ae0288306 Mon Sep 17 00:00:00 2001 From: Patrick Date: Sun, 9 Jun 2024 03:32:54 +0800 Subject: [PATCH 1/9] add workerQueue --- .../session-loader/Development/IDE/Session.hs | 6 +-- ghcide/src/Development/IDE/Core/Compile.hs | 3 +- ghcide/src/Development/IDE/Core/FileStore.hs | 4 +- ghcide/src/Development/IDE/Core/Shake.hs | 10 ++-- .../src/Development/IDE/Core/WorkerThread.hs | 52 +++++++++++++++---- .../src/Development/IDE/LSP/LanguageServer.hs | 5 +- 6 files changed, 58 insertions(+), 22 deletions(-) diff --git a/ghcide/session-loader/Development/IDE/Session.hs b/ghcide/session-loader/Development/IDE/Session.hs index aaa74bcc8c..23f01f5d4c 100644 --- a/ghcide/session-loader/Development/IDE/Session.hs +++ b/ghcide/session-loader/Development/IDE/Session.hs @@ -94,7 +94,6 @@ import Data.Void import Control.Concurrent.STM.Stats (atomically, modifyTVar', readTVar, writeTVar) -import Control.Concurrent.STM.TQueue import Control.DeepSeq import Control.Exception (evaluate) import Control.Monad.IO.Unlift (MonadUnliftIO) @@ -105,7 +104,8 @@ import Data.HashSet (HashSet) import qualified Data.HashSet as Set import Database.SQLite.Simple import Development.IDE.Core.Tracing (withTrace) -import Development.IDE.Core.WorkerThread (awaitRunInThread, +import Development.IDE.Core.WorkerThread (WorkerQueue, + awaitRunInThread, withWorkerQueue) import Development.IDE.Session.Diagnostics (renderCradleError) import Development.IDE.Types.Shake (WithHieDb, @@ -438,7 +438,7 @@ getHieDbLoc dir = do -- components mapping to the same hie.yaml file are mapped to the same -- HscEnv which is updated as new components are discovered. -loadSessionWithOptions :: Recorder (WithPriority Log) -> SessionLoadingOptions -> FilePath -> TQueue (IO ()) -> IO (Action IdeGhcSession) +loadSessionWithOptions :: Recorder (WithPriority Log) -> SessionLoadingOptions -> FilePath -> WorkerQueue (IO ()) -> IO (Action IdeGhcSession) loadSessionWithOptions recorder SessionLoadingOptions{..} rootDir que = do let toAbsolutePath = toAbsolute rootDir -- see Note [Root Directory] cradle_files <- newIORef [] diff --git a/ghcide/src/Development/IDE/Core/Compile.hs b/ghcide/src/Development/IDE/Core/Compile.hs index f295e568c6..3cd9a244eb 100644 --- a/ghcide/src/Development/IDE/Core/Compile.hs +++ b/ghcide/src/Development/IDE/Core/Compile.hs @@ -129,6 +129,7 @@ import GHC.Driver.Config.CoreToStg.Prep #if MIN_VERSION_ghc(9,7,0) import Data.Foldable (toList) import GHC.Unit.Module.Warnings +import Development.IDE.Core.WorkerThread (writeWorkerQueue) #else import Development.IDE.Core.FileStore (shareFilePath) #endif @@ -899,7 +900,7 @@ indexHieFile se mod_summary srcPath !hash hf = do -- hiedb doesn't use the Haskell src, so we clear it to avoid unnecessarily keeping it around let !hf' = hf{hie_hs_src = mempty} modifyTVar' indexPending $ HashMap.insert srcPath hash - writeTQueue indexQueue $ \withHieDb -> do + writeWorkerQueue indexQueue $ \withHieDb -> do -- We are now in the worker thread -- Check if a newer index of this file has been scheduled, and if so skip this one newerScheduled <- atomically $ do diff --git a/ghcide/src/Development/IDE/Core/FileStore.hs b/ghcide/src/Development/IDE/Core/FileStore.hs index 6c0cb875b0..90dc69670b 100644 --- a/ghcide/src/Development/IDE/Core/FileStore.hs +++ b/ghcide/src/Development/IDE/Core/FileStore.hs @@ -22,7 +22,6 @@ module Development.IDE.Core.FileStore( ) where import Control.Concurrent.STM.Stats (STM, atomically) -import Control.Concurrent.STM.TQueue (writeTQueue) import Control.Exception import Control.Monad.Extra import Control.Monad.IO.Class @@ -40,6 +39,7 @@ import Development.IDE.Core.IdeConfiguration (isWorkspaceFile) import Development.IDE.Core.RuleTypes import Development.IDE.Core.Shake hiding (Log) import qualified Development.IDE.Core.Shake as Shake +import Development.IDE.Core.WorkerThread (writeWorkerQueue) import Development.IDE.GHC.Orphans () import Development.IDE.Graph import Development.IDE.Import.DependencyInformation @@ -247,7 +247,7 @@ typecheckParentsAction recorder nfp = do setSomethingModified :: VFSModified -> IdeState -> String -> IO [Key] -> IO () setSomethingModified vfs state reason actionBetweenSession = do -- Update database to remove any files that might have been renamed/deleted - atomically $ writeTQueue (indexQueue $ hiedbWriter $ shakeExtras state) (\withHieDb -> withHieDb deleteMissingRealFiles) + atomically $ writeWorkerQueue (indexQueue $ hiedbWriter $ shakeExtras state) (\withHieDb -> withHieDb deleteMissingRealFiles) void $ restartShakeSession (shakeExtras state) vfs reason [] actionBetweenSession registerFileWatches :: [String] -> LSP.LspT Config IO Bool diff --git a/ghcide/src/Development/IDE/Core/Shake.hs b/ghcide/src/Development/IDE/Core/Shake.hs index d426ba34f8..a439feb5f4 100644 --- a/ghcide/src/Development/IDE/Core/Shake.hs +++ b/ghcide/src/Development/IDE/Core/Shake.hs @@ -262,12 +262,12 @@ data HieDbWriter -- | Actions to queue up on the index worker thread -- The inner `(HieDb -> IO ()) -> IO ()` wraps `HieDb -> IO ()` -- with (currently) retry functionality -type IndexQueue = TQueue (((HieDb -> IO ()) -> IO ()) -> IO ()) +type IndexQueue = WorkerQueue (((HieDb -> IO ()) -> IO ()) -> IO ()) data ThreadQueue = ThreadQueue { tIndexQueue :: IndexQueue - , tRestartQueue :: TQueue (IO ()) - , tLoaderQueue :: TQueue (IO ()) + , tRestartQueue :: WorkerQueue (IO ()) + , tLoaderQueue :: WorkerQueue (IO ()) } -- Note [Semantic Tokens Cache Location] @@ -342,9 +342,9 @@ data ShakeExtras = ShakeExtras -- ^ 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 ()) + , restartQueue :: WorkerQueue (IO ()) -- ^ Queue of restart actions to be run. - , loaderQueue :: TQueue (IO ()) + , loaderQueue :: WorkerQueue (IO ()) -- ^ Queue of loader actions to be run. } diff --git a/ghcide/src/Development/IDE/Core/WorkerThread.hs b/ghcide/src/Development/IDE/Core/WorkerThread.hs index a38da77f38..87377e6d32 100644 --- a/ghcide/src/Development/IDE/Core/WorkerThread.hs +++ b/ghcide/src/Development/IDE/Core/WorkerThread.hs @@ -7,15 +7,17 @@ Description : This module provides an API for managing worker threads in the IDE see Note [Serializing runs in separate thread] -} module Development.IDE.Core.WorkerThread - (withWorkerQueue, awaitRunInThread) + (withWorkerQueue, awaitRunInThread, withWorkerQueueOfOne, WorkerQueue, writeWorkerQueue) where import Control.Concurrent.Async (withAsync) import Control.Concurrent.STM import Control.Concurrent.Strict (newBarrier, signalBarrier, waitBarrier) +import Control.Exception (finally) import Control.Monad (forever) import Control.Monad.Cont (ContT (ContT)) +import Control.Monad.IO.Class (liftIO) {- 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 Moreover, we can not stop these threads uniformly when we are shutting down the server. -} --- | 'withWorkerQueue' creates a new 'TQueue', and launches a worker +data WorkerQueue a = WorkerQueueOfOne (TMVar a) | WorkerQueueOfMany (TQueue a) + +writeWorkerQueue :: WorkerQueue a -> a -> STM () +writeWorkerQueue (WorkerQueueOfOne tvar) action = putTMVar tvar action +writeWorkerQueue (WorkerQueueOfMany tqueue) action = writeTQueue tqueue action + +newWorkerQueue :: STM (WorkerQueue a) +newWorkerQueue = WorkerQueueOfMany <$> newTQueue + +newWorkerQueueOfOne :: STM (WorkerQueue a) +newWorkerQueueOfOne = WorkerQueueOfOne <$> newEmptyTMVar + + +-- | 'withWorkerQueue' creates a new 'WorkerQueue', and launches a worker -- thread which polls the queue for requests and runs the given worker -- function on them. -withWorkerQueue :: (t -> IO a) -> ContT () IO (TQueue t) -withWorkerQueue workerAction = ContT $ \mainAction -> do - q <- newTQueueIO +withWorkerQueue :: (t -> IO a) -> ContT () IO (WorkerQueue t) +withWorkerQueue workerAction = do + q <- liftIO $ atomically newWorkerQueue + runWorkerQueue q workerAction + +-- | 'withWorkerQueueOfOne' creates a new 'WorkerQueue' that only allows one action to be queued at a time. +-- and one action can only be queued after the previous action has been done. +-- this is useful when we want to cancel the action waiting in the queue, if it's thread is cancelled. +-- e.g. session loading in session loader. When a shake session is restarted, we want to cancel the previous pending session loading. +withWorkerQueueOfOne :: (t -> IO a) -> ContT () IO (WorkerQueue t) +withWorkerQueueOfOne workerAction = do + q <- liftIO $ atomically newWorkerQueueOfOne + runWorkerQueue q workerAction + +runWorkerQueue :: WorkerQueue t -> (t -> IO a) -> ContT () IO (WorkerQueue t) +runWorkerQueue q workerAction = ContT $ \mainAction -> do withAsync (writerThread q) $ \_ -> mainAction q where writerThread q = forever $ do - l <- atomically $ readTQueue q - workerAction l + case q of + -- only remove the action from the queue after it has been run if it is a one-shot queue + WorkerQueueOfOne tvar -> do + l <- atomically $ readTMVar tvar + workerAction l `finally` atomically (takeTMVar tvar) + WorkerQueueOfMany q -> do + l <- atomically $ readTQueue q + workerAction l -- | 'awaitRunInThread' queues up an 'IO' action to be run by a worker thread, -- and then blocks until the result is computed. -awaitRunInThread :: TQueue (IO ()) -> IO result -> IO result +awaitRunInThread :: WorkerQueue (IO ()) -> IO result -> IO result awaitRunInThread q act = do -- Take an action from TQueue, run it and -- use barrier to wait for the result barrier <- newBarrier - atomically $ writeTQueue q $ do + atomically $ writeWorkerQueue q $ do res <- act signalBarrier barrier res waitBarrier barrier diff --git a/ghcide/src/Development/IDE/LSP/LanguageServer.hs b/ghcide/src/Development/IDE/LSP/LanguageServer.hs index cf7845ce08..6743b9932f 100644 --- a/ghcide/src/Development/IDE/LSP/LanguageServer.hs +++ b/ghcide/src/Development/IDE/LSP/LanguageServer.hs @@ -39,7 +39,8 @@ import Control.Monad.Trans.Cont (evalContT) import Development.IDE.Core.IdeConfiguration import Development.IDE.Core.Shake hiding (Log) import Development.IDE.Core.Tracing -import Development.IDE.Core.WorkerThread (withWorkerQueue) +import Development.IDE.Core.WorkerThread (withWorkerQueue, + withWorkerQueueOfOne) import qualified Development.IDE.Session as Session import Development.IDE.Types.Shake (WithHieDb, WithHieDbShield (..)) @@ -261,7 +262,7 @@ handleInit recorder defaultRoot getHieDbLoc getIdeState lifetime exitClientMsg c runWithWorkerThreads :: Recorder (WithPriority Session.Log) -> FilePath -> (WithHieDb -> ThreadQueue -> IO ()) -> IO () runWithWorkerThreads recorder dbLoc f = evalContT $ do sessionRestartTQueue <- withWorkerQueue id - sessionLoaderTQueue <- withWorkerQueue id + sessionLoaderTQueue <- withWorkerQueueOfOne id (WithHieDbShield hiedb, threadQueue) <- runWithDb recorder dbLoc liftIO $ f hiedb (ThreadQueue threadQueue sessionRestartTQueue sessionLoaderTQueue) From ca84bb5f96e975bbf56f4403c754a66fea9ca33a Mon Sep 17 00:00:00 2001 From: Patrick Date: Sun, 9 Jun 2024 03:42:12 +0800 Subject: [PATCH 2/9] Fix build --- ghcide/src/Development/IDE/Core/Compile.hs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ghcide/src/Development/IDE/Core/Compile.hs b/ghcide/src/Development/IDE/Core/Compile.hs index 3cd9a244eb..29a0ffea08 100644 --- a/ghcide/src/Development/IDE/Core/Compile.hs +++ b/ghcide/src/Development/IDE/Core/Compile.hs @@ -108,6 +108,7 @@ import GHC.Tc.Gen.Splice import GHC.Types.ForeignStubs import GHC.Types.HpcInfo import GHC.Types.TypeEnv +import Development.IDE.Core.WorkerThread (writeWorkerQueue) -- See Note [Guidelines For Using CPP In GHCIDE Import Statements] @@ -129,7 +130,6 @@ import GHC.Driver.Config.CoreToStg.Prep #if MIN_VERSION_ghc(9,7,0) import Data.Foldable (toList) import GHC.Unit.Module.Warnings -import Development.IDE.Core.WorkerThread (writeWorkerQueue) #else import Development.IDE.Core.FileStore (shareFilePath) #endif From b839ba7f274ece2c8d9fb8d220cd2e54d2c20f03 Mon Sep 17 00:00:00 2001 From: Patrick Date: Sun, 9 Jun 2024 03:43:19 +0800 Subject: [PATCH 3/9] fix import order --- ghcide/src/Development/IDE/Core/Compile.hs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ghcide/src/Development/IDE/Core/Compile.hs b/ghcide/src/Development/IDE/Core/Compile.hs index 29a0ffea08..cb83705a70 100644 --- a/ghcide/src/Development/IDE/Core/Compile.hs +++ b/ghcide/src/Development/IDE/Core/Compile.hs @@ -74,6 +74,7 @@ import Development.IDE.Core.Preprocessor import Development.IDE.Core.RuleTypes import Development.IDE.Core.Shake import Development.IDE.Core.Tracing (withTrace) +import Development.IDE.Core.WorkerThread (writeWorkerQueue) import Development.IDE.GHC.Compat hiding (loadInterface, parseHeader, parseModule, tcRnModule, writeHieFile) @@ -108,7 +109,6 @@ import GHC.Tc.Gen.Splice import GHC.Types.ForeignStubs import GHC.Types.HpcInfo import GHC.Types.TypeEnv -import Development.IDE.Core.WorkerThread (writeWorkerQueue) -- See Note [Guidelines For Using CPP In GHCIDE Import Statements] From 69043359ce8c65fd32a16c63ebab83141d6f54a0 Mon Sep 17 00:00:00 2001 From: Patrick Date: Sun, 9 Jun 2024 18:56:00 +0800 Subject: [PATCH 4/9] stylish --- ghcide/session-loader/Development/IDE/Session.hs | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/ghcide/session-loader/Development/IDE/Session.hs b/ghcide/session-loader/Development/IDE/Session.hs index 0667c75b98..a6e2a08a14 100644 --- a/ghcide/session-loader/Development/IDE/Session.hs +++ b/ghcide/session-loader/Development/IDE/Session.hs @@ -101,14 +101,14 @@ import Data.HashMap.Strict (HashMap) import Data.HashSet (HashSet) import qualified Data.HashSet as Set import Database.SQLite.Simple -import Development.IDE.Core.Tracing (withTrace) -import Development.IDE.Core.WorkerThread (WorkerQueue, - awaitRunInThread, - withWorkerQueue) -import Development.IDE.Session.Diagnostics (renderCradleError) -import Development.IDE.Types.Shake (WithHieDb, - WithHieDbShield (..), - toNoFileKey) +import Development.IDE.Core.Tracing (withTrace) +import Development.IDE.Core.WorkerThread (WorkerQueue, + awaitRunInThread, + withWorkerQueue) +import Development.IDE.Session.Diagnostics (renderCradleError) +import Development.IDE.Types.Shake (WithHieDb, + WithHieDbShield (..), + toNoFileKey) import HieDb.Create import HieDb.Types import HieDb.Utils From 7c47722c6d41225b644c47bc40ce2612c7a13e5e Mon Sep 17 00:00:00 2001 From: Patrick Date: Mon, 10 Jun 2024 18:11:32 +0800 Subject: [PATCH 5/9] add `waitUntilWorkerQueueEmpty` --- .../src/Development/IDE/Core/WorkerThread.hs | 26 ++++++++++++++----- 1 file changed, 20 insertions(+), 6 deletions(-) diff --git a/ghcide/src/Development/IDE/Core/WorkerThread.hs b/ghcide/src/Development/IDE/Core/WorkerThread.hs index 87377e6d32..3410353e47 100644 --- a/ghcide/src/Development/IDE/Core/WorkerThread.hs +++ b/ghcide/src/Development/IDE/Core/WorkerThread.hs @@ -7,7 +7,12 @@ Description : This module provides an API for managing worker threads in the IDE see Note [Serializing runs in separate thread] -} module Development.IDE.Core.WorkerThread - (withWorkerQueue, awaitRunInThread, withWorkerQueueOfOne, WorkerQueue, writeWorkerQueue) + (withWorkerQueue + , awaitRunInThread + , withWorkerQueueOfOne + , WorkerQueue + , writeWorkerQueue + , waitUntilWorkerQueueEmpty) where import Control.Concurrent.Async (withAsync) @@ -15,7 +20,7 @@ import Control.Concurrent.STM import Control.Concurrent.Strict (newBarrier, signalBarrier, waitBarrier) import Control.Exception (finally) -import Control.Monad (forever) +import Control.Monad (forever, unless) import Control.Monad.Cont (ContT (ContT)) import Control.Monad.IO.Class (liftIO) @@ -53,7 +58,7 @@ withWorkerQueue workerAction = do -- | 'withWorkerQueueOfOne' creates a new 'WorkerQueue' that only allows one action to be queued at a time. -- and one action can only be queued after the previous action has been done. --- this is useful when we want to cancel the action waiting in the queue, if it's thread is cancelled. +-- this is useful when we want to cancel the action waiting to be enqueue if it's thread is cancelled. -- e.g. session loading in session loader. When a shake session is restarted, we want to cancel the previous pending session loading. withWorkerQueueOfOne :: (t -> IO a) -> ContT () IO (WorkerQueue t) withWorkerQueueOfOne workerAction = do @@ -67,13 +72,22 @@ runWorkerQueue q workerAction = ContT $ \mainAction -> do writerThread q = forever $ do case q of - -- only remove the action from the queue after it has been run if it is a one-shot queue + -- only remove the action from the queue after it has done WorkerQueueOfOne tvar -> do l <- atomically $ readTMVar tvar workerAction l `finally` atomically (takeTMVar tvar) WorkerQueueOfMany q -> do - l <- atomically $ readTQueue q - workerAction l + l <- atomically $ peekTQueue q + workerAction l `finally` atomically (readTQueue q) + +-- | waitUntilWorkerQueueEmpty blocks until the worker queue is empty. +waitUntilWorkerQueueEmpty :: WorkerQueue a -> IO () +waitUntilWorkerQueueEmpty (WorkerQueueOfOne tvar) = atomically $ do + isEmpty <- isEmptyTMVar tvar + unless isEmpty retry +waitUntilWorkerQueueEmpty (WorkerQueueOfMany queue) = atomically $ do + isEmpty <- isEmptyTQueue queue + unless isEmpty retry -- | 'awaitRunInThread' queues up an 'IO' action to be run by a worker thread, -- and then blocks until the result is computed. From c7eeffbbe358a0f62a6055fdf7890b1593d65a86 Mon Sep 17 00:00:00 2001 From: Patrick Date: Mon, 10 Jun 2024 18:20:21 +0800 Subject: [PATCH 6/9] add `peekWorkerQueue` and `readWorkerQueue` --- .../src/Development/IDE/Core/WorkerThread.hs | 20 ++++++++++--------- 1 file changed, 11 insertions(+), 9 deletions(-) diff --git a/ghcide/src/Development/IDE/Core/WorkerThread.hs b/ghcide/src/Development/IDE/Core/WorkerThread.hs index 3410353e47..6cc9062488 100644 --- a/ghcide/src/Development/IDE/Core/WorkerThread.hs +++ b/ghcide/src/Development/IDE/Core/WorkerThread.hs @@ -37,6 +37,14 @@ Moreover, we can not stop these threads uniformly when we are shutting down the data WorkerQueue a = WorkerQueueOfOne (TMVar a) | WorkerQueueOfMany (TQueue a) +peekWorkerQueue :: WorkerQueue a -> STM a +peekWorkerQueue (WorkerQueueOfOne tvar) = readTMVar tvar +peekWorkerQueue (WorkerQueueOfMany tqueue) = peekTQueue tqueue + +readWorkerQueue :: WorkerQueue a -> STM a +readWorkerQueue (WorkerQueueOfOne tvar) = takeTMVar tvar +readWorkerQueue (WorkerQueueOfMany tqueue) = readTQueue tqueue + writeWorkerQueue :: WorkerQueue a -> a -> STM () writeWorkerQueue (WorkerQueueOfOne tvar) action = putTMVar tvar action writeWorkerQueue (WorkerQueueOfMany tqueue) action = writeTQueue tqueue action @@ -47,7 +55,6 @@ newWorkerQueue = WorkerQueueOfMany <$> newTQueue newWorkerQueueOfOne :: STM (WorkerQueue a) newWorkerQueueOfOne = WorkerQueueOfOne <$> newEmptyTMVar - -- | 'withWorkerQueue' creates a new 'WorkerQueue', and launches a worker -- thread which polls the queue for requests and runs the given worker -- function on them. @@ -71,14 +78,9 @@ runWorkerQueue q workerAction = ContT $ \mainAction -> do where writerThread q = forever $ do - case q of - -- only remove the action from the queue after it has done - WorkerQueueOfOne tvar -> do - l <- atomically $ readTMVar tvar - workerAction l `finally` atomically (takeTMVar tvar) - WorkerQueueOfMany q -> do - l <- atomically $ peekTQueue q - workerAction l `finally` atomically (readTQueue q) + -- peek the action from the queue, run it and then remove it from the queue + l <- atomically $ peekWorkerQueue q + workerAction l `finally` atomically (readWorkerQueue q) -- | waitUntilWorkerQueueEmpty blocks until the worker queue is empty. waitUntilWorkerQueueEmpty :: WorkerQueue a -> IO () From ed5949c92259a151b0ad7ceabb6b88d17a21eba5 Mon Sep 17 00:00:00 2001 From: Patrick Date: Mon, 10 Jun 2024 18:25:04 +0800 Subject: [PATCH 7/9] fix comment --- ghcide/src/Development/IDE/Core/WorkerThread.hs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/ghcide/src/Development/IDE/Core/WorkerThread.hs b/ghcide/src/Development/IDE/Core/WorkerThread.hs index 6cc9062488..644ae60f98 100644 --- a/ghcide/src/Development/IDE/Core/WorkerThread.hs +++ b/ghcide/src/Development/IDE/Core/WorkerThread.hs @@ -66,7 +66,9 @@ withWorkerQueue workerAction = do -- | 'withWorkerQueueOfOne' creates a new 'WorkerQueue' that only allows one action to be queued at a time. -- and one action can only be queued after the previous action has been done. -- this is useful when we want to cancel the action waiting to be enqueue if it's thread is cancelled. --- e.g. session loading in session loader. When a shake session is restarted, we want to cancel the previous pending session loading. +-- e.g. session loading in session loader. When a shake session is restarted +-- , we want to cancel the previous pending session loading. +-- since the hls-graph can handle the retrying of the session loading. withWorkerQueueOfOne :: (t -> IO a) -> ContT () IO (WorkerQueue t) withWorkerQueueOfOne workerAction = do q <- liftIO $ atomically newWorkerQueueOfOne From 3f8c4db2db5e7b7846b0d4dfd8875187dc1ebf70 Mon Sep 17 00:00:00 2001 From: Patrick Date: Mon, 10 Jun 2024 18:25:42 +0800 Subject: [PATCH 8/9] rename --- ghcide/src/Development/IDE/Core/WorkerThread.hs | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/ghcide/src/Development/IDE/Core/WorkerThread.hs b/ghcide/src/Development/IDE/Core/WorkerThread.hs index 644ae60f98..8c3358705f 100644 --- a/ghcide/src/Development/IDE/Core/WorkerThread.hs +++ b/ghcide/src/Development/IDE/Core/WorkerThread.hs @@ -38,16 +38,16 @@ Moreover, we can not stop these threads uniformly when we are shutting down the data WorkerQueue a = WorkerQueueOfOne (TMVar a) | WorkerQueueOfMany (TQueue a) peekWorkerQueue :: WorkerQueue a -> STM a -peekWorkerQueue (WorkerQueueOfOne tvar) = readTMVar tvar -peekWorkerQueue (WorkerQueueOfMany tqueue) = peekTQueue tqueue +peekWorkerQueue (WorkerQueueOfOne tVar) = readTMVar tVar +peekWorkerQueue (WorkerQueueOfMany tQueue) = peekTQueue tQueue readWorkerQueue :: WorkerQueue a -> STM a -readWorkerQueue (WorkerQueueOfOne tvar) = takeTMVar tvar -readWorkerQueue (WorkerQueueOfMany tqueue) = readTQueue tqueue +readWorkerQueue (WorkerQueueOfOne tVar) = takeTMVar tVar +readWorkerQueue (WorkerQueueOfMany tQueue) = readTQueue tQueue writeWorkerQueue :: WorkerQueue a -> a -> STM () -writeWorkerQueue (WorkerQueueOfOne tvar) action = putTMVar tvar action -writeWorkerQueue (WorkerQueueOfMany tqueue) action = writeTQueue tqueue action +writeWorkerQueue (WorkerQueueOfOne tVar) action = putTMVar tVar action +writeWorkerQueue (WorkerQueueOfMany tQueue) action = writeTQueue tQueue action newWorkerQueue :: STM (WorkerQueue a) newWorkerQueue = WorkerQueueOfMany <$> newTQueue @@ -86,8 +86,8 @@ runWorkerQueue q workerAction = ContT $ \mainAction -> do -- | waitUntilWorkerQueueEmpty blocks until the worker queue is empty. waitUntilWorkerQueueEmpty :: WorkerQueue a -> IO () -waitUntilWorkerQueueEmpty (WorkerQueueOfOne tvar) = atomically $ do - isEmpty <- isEmptyTMVar tvar +waitUntilWorkerQueueEmpty (WorkerQueueOfOne tVar) = atomically $ do + isEmpty <- isEmptyTMVar tVar unless isEmpty retry waitUntilWorkerQueueEmpty (WorkerQueueOfMany queue) = atomically $ do isEmpty <- isEmptyTQueue queue From 66e8075f5830ea7e42ba5f5021ba20a93f9f8a2b Mon Sep 17 00:00:00 2001 From: Patrick Date: Mon, 10 Jun 2024 18:35:39 +0800 Subject: [PATCH 9/9] refactor --- .../src/Development/IDE/Core/WorkerThread.hs | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/ghcide/src/Development/IDE/Core/WorkerThread.hs b/ghcide/src/Development/IDE/Core/WorkerThread.hs index 8c3358705f..dd10c5f7e6 100644 --- a/ghcide/src/Development/IDE/Core/WorkerThread.hs +++ b/ghcide/src/Development/IDE/Core/WorkerThread.hs @@ -37,10 +37,12 @@ Moreover, we can not stop these threads uniformly when we are shutting down the data WorkerQueue a = WorkerQueueOfOne (TMVar a) | WorkerQueueOfMany (TQueue a) +-- | peekWorkerQueue returns the next action in the queue without removing it. peekWorkerQueue :: WorkerQueue a -> STM a peekWorkerQueue (WorkerQueueOfOne tVar) = readTMVar tVar peekWorkerQueue (WorkerQueueOfMany tQueue) = peekTQueue tQueue +-- | readWorkerQueue returns the next action in the queue and removes it. readWorkerQueue :: WorkerQueue a -> STM a readWorkerQueue (WorkerQueueOfOne tVar) = takeTMVar tVar readWorkerQueue (WorkerQueueOfMany tQueue) = readTQueue tQueue @@ -49,6 +51,15 @@ writeWorkerQueue :: WorkerQueue a -> a -> STM () writeWorkerQueue (WorkerQueueOfOne tVar) action = putTMVar tVar action writeWorkerQueue (WorkerQueueOfMany tQueue) action = writeTQueue tQueue action +-- | waitUntilWorkerQueueEmpty blocks until the worker queue is empty. +waitUntilWorkerQueueEmpty :: WorkerQueue a -> STM () +waitUntilWorkerQueueEmpty (WorkerQueueOfOne tVar) = do + isEmpty <- isEmptyTMVar tVar + unless isEmpty retry +waitUntilWorkerQueueEmpty (WorkerQueueOfMany queue) = do + isEmpty <- isEmptyTQueue queue + unless isEmpty retry + newWorkerQueue :: STM (WorkerQueue a) newWorkerQueue = WorkerQueueOfMany <$> newTQueue @@ -84,14 +95,6 @@ runWorkerQueue q workerAction = ContT $ \mainAction -> do l <- atomically $ peekWorkerQueue q workerAction l `finally` atomically (readWorkerQueue q) --- | waitUntilWorkerQueueEmpty blocks until the worker queue is empty. -waitUntilWorkerQueueEmpty :: WorkerQueue a -> IO () -waitUntilWorkerQueueEmpty (WorkerQueueOfOne tVar) = atomically $ do - isEmpty <- isEmptyTMVar tVar - unless isEmpty retry -waitUntilWorkerQueueEmpty (WorkerQueueOfMany queue) = atomically $ do - isEmpty <- isEmptyTQueue queue - unless isEmpty retry -- | 'awaitRunInThread' queues up an 'IO' action to be run by a worker thread, -- and then blocks until the result is computed.