{-# LANGUAGE BlockArguments #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE CPP #-}
{-# LANGUAGE RecordWildCards #-}
{-# LANGUAGE TupleSections #-}
{-# LANGUAGE NumericUnderscores #-}
module GHC.Driver.MakeSem
(
#if !(defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH))
runJSemAbstractSem,
#endif
AbstractSem(..)
, withAbstractSem
)
where
#if defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH)
import System.Semaphore
( AbstractSem(..)
, withAbstractSem
)
#else
import GHC.Prelude
import GHC.Conc
import GHC.Data.OrdList
import GHC.IO.Exception
import GHC.Utils.Outputable
import GHC.Utils.Panic
import GHC.Utils.Json
import System.Semaphore
( AbstractSem(..)
, ClientSemaphore
, SemaphoreIdentifier
, SemaphoreToken
, openSemaphore
, releaseSemaphoreToken
, waitOnSemaphore
, withAbstractSem
)
import Control.Monad
import qualified Control.Monad.Catch as MC
import Control.Concurrent.MVar
import Control.Concurrent.STM
import Data.Foldable
import Data.Functor
import GHC.Stack
import Debug.Trace
data Jobserver
= Jobserver
{ Jobserver -> ClientSemaphore
jSemaphore :: !ClientSemaphore
, Jobserver -> TVar JobResources
jobs :: !(TVar JobResources)
, Jobserver -> TVar (Maybe (ThreadId, TMVar (Maybe SomeException)))
activeChild :: !(TVar (Maybe (ThreadId, TMVar (Maybe MC.SomeException))))
}
data JobserverOptions
= JobserverOptions
{ JobserverOptions -> Int
releaseDebounce :: !Int
, JobserverOptions -> Int
setNumCapsDebounce :: !Int
}
defaultJobserverOptions :: JobserverOptions
defaultJobserverOptions :: JobserverOptions
defaultJobserverOptions =
JobserverOptions
{ releaseDebounce :: Int
releaseDebounce = Int
1000
, setNumCapsDebounce :: Int
setNumCapsDebounce = Int
1000
}
data JobResources
= Jobs
{ JobResources -> Int
tokensOwned :: !Int
, JobResources -> Int
tokensFree :: !Int
, JobResources -> OrdList (TMVar ())
jobsWaiting :: !(OrdList (TMVar ()))
, JobResources -> [SemaphoreToken]
heldTokens :: [SemaphoreToken]
}
instance Outputable JobResources where
ppr :: JobResources -> SDoc
ppr Jobs{Int
[SemaphoreToken]
OrdList (TMVar ())
tokensOwned :: JobResources -> Int
tokensFree :: JobResources -> Int
jobsWaiting :: JobResources -> OrdList (TMVar ())
heldTokens :: JobResources -> [SemaphoreToken]
tokensOwned :: Int
tokensFree :: Int
jobsWaiting :: OrdList (TMVar ())
heldTokens :: [SemaphoreToken]
..}
= String -> SDoc
forall doc. IsLine doc => String -> doc
text String
"JobResources" SDoc -> SDoc -> SDoc
forall doc. IsLine doc => doc -> doc -> doc
<+>
( SDoc -> SDoc
forall doc. IsLine doc => doc -> doc
braces (SDoc -> SDoc) -> SDoc -> SDoc
forall a b. (a -> b) -> a -> b
$ [SDoc] -> SDoc
forall doc. IsLine doc => [doc] -> doc
hsep
[ String -> SDoc
forall doc. IsLine doc => String -> doc
text String
"owned=" SDoc -> SDoc -> SDoc
forall doc. IsLine doc => doc -> doc -> doc
<> Int -> SDoc
forall a. Outputable a => a -> SDoc
ppr Int
tokensOwned
, String -> SDoc
forall doc. IsLine doc => String -> doc
text String
"free=" SDoc -> SDoc -> SDoc
forall doc. IsLine doc => doc -> doc -> doc
<> Int -> SDoc
forall a. Outputable a => a -> SDoc
ppr Int
tokensFree
, String -> SDoc
forall doc. IsLine doc => String -> doc
text String
"num_waiting=" SDoc -> SDoc -> SDoc
forall doc. IsLine doc => doc -> doc -> doc
<> Int -> SDoc
forall a. Outputable a => a -> SDoc
ppr (OrdList (TMVar ()) -> Int
forall a. OrdList a -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length OrdList (TMVar ())
jobsWaiting)
] )
addToken :: SemaphoreToken -> JobResources -> JobResources
addToken :: SemaphoreToken -> JobResources -> JobResources
addToken SemaphoreToken
tok jobs :: JobResources
jobs@( Jobs { tokensOwned :: JobResources -> Int
tokensOwned = Int
owned, tokensFree :: JobResources -> Int
tokensFree = Int
free, heldTokens :: JobResources -> [SemaphoreToken]
heldTokens = [SemaphoreToken]
toks })
= JobResources
jobs { tokensOwned = owned + 1, tokensFree = free + 1, heldTokens = tok : toks }
addFreeToken :: JobResources -> JobResources
addFreeToken :: JobResources -> JobResources
addFreeToken jobs :: JobResources
jobs@( Jobs { tokensFree :: JobResources -> Int
tokensFree = Int
free })
= Bool -> SDoc -> JobResources -> JobResources
forall a. HasCallStack => Bool -> SDoc -> a -> a
assertPpr (JobResources -> Int
tokensOwned JobResources
jobs Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
free)
(String -> SDoc
forall doc. IsLine doc => String -> doc
text String
"addFreeToken:" SDoc -> SDoc -> SDoc
forall doc. IsLine doc => doc -> doc -> doc
<+> Int -> SDoc
forall a. Outputable a => a -> SDoc
ppr (JobResources -> Int
tokensOwned JobResources
jobs) SDoc -> SDoc -> SDoc
forall doc. IsLine doc => doc -> doc -> doc
<+> Int -> SDoc
forall a. Outputable a => a -> SDoc
ppr Int
free)
(JobResources -> JobResources) -> JobResources -> JobResources
forall a b. (a -> b) -> a -> b
$ JobResources
jobs { tokensFree = free + 1 }
removeFreeToken :: JobResources -> JobResources
removeFreeToken :: JobResources -> JobResources
removeFreeToken jobs :: JobResources
jobs@( Jobs { tokensFree :: JobResources -> Int
tokensFree = Int
free })
= Bool -> SDoc -> JobResources -> JobResources
forall a. HasCallStack => Bool -> SDoc -> a -> a
assertPpr (Int
free Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
0)
(String -> SDoc
forall doc. IsLine doc => String -> doc
text String
"removeFreeToken:" SDoc -> SDoc -> SDoc
forall doc. IsLine doc => doc -> doc -> doc
<+> Int -> SDoc
forall a. Outputable a => a -> SDoc
ppr Int
free)
(JobResources -> JobResources) -> JobResources -> JobResources
forall a b. (a -> b) -> a -> b
$ JobResources
jobs { tokensFree = free - 1 }
removeOwnedToken :: JobResources -> (SemaphoreToken, JobResources)
removeOwnedToken :: JobResources -> (SemaphoreToken, JobResources)
removeOwnedToken jobs :: JobResources
jobs@( Jobs { tokensOwned :: JobResources -> Int
tokensOwned = Int
owned, heldTokens :: JobResources -> [SemaphoreToken]
heldTokens = [SemaphoreToken]
toks })
= Bool
-> SDoc
-> (SemaphoreToken, JobResources)
-> (SemaphoreToken, JobResources)
forall a. HasCallStack => Bool -> SDoc -> a -> a
assertPpr (Int
owned Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
1)
(String -> SDoc
forall doc. IsLine doc => String -> doc
text String
"removeOwnedToken:" SDoc -> SDoc -> SDoc
forall doc. IsLine doc => doc -> doc -> doc
<+> Int -> SDoc
forall a. Outputable a => a -> SDoc
ppr Int
owned)
((SemaphoreToken, JobResources) -> (SemaphoreToken, JobResources))
-> (SemaphoreToken, JobResources) -> (SemaphoreToken, JobResources)
forall a b. (a -> b) -> a -> b
$ case [SemaphoreToken]
toks of
(SemaphoreToken
t:[SemaphoreToken]
rest) -> (SemaphoreToken
t, JobResources
jobs { tokensOwned = owned - 1, heldTokens = rest })
[] -> String -> (SemaphoreToken, JobResources)
forall a. HasCallStack => String -> a
panic String
"removeOwnedToken: no held tokens"
addJob :: TMVar () -> JobResources -> JobResources
addJob :: TMVar () -> JobResources -> JobResources
addJob TMVar ()
job jobs :: JobResources
jobs@( Jobs { jobsWaiting :: JobResources -> OrdList (TMVar ())
jobsWaiting = OrdList (TMVar ())
wait })
= JobResources
jobs { jobsWaiting = wait `SnocOL` job }
data JobserverState
= JobserverState
{ JobserverState -> JobserverAction
jobserverAction :: !JobserverAction
, JobserverState -> TVar Bool
canChangeNumCaps :: !(TVar Bool)
, JobserverState -> TVar Bool
canReleaseToken :: !(TVar Bool)
}
data JobserverAction
= Idle
| Acquiring
{ JobserverAction -> ThreadId
activeThreadId :: ThreadId
, JobserverAction -> TMVar (Maybe SomeException)
threadFinished :: TMVar (Maybe MC.SomeException) }
activeThread_maybe :: JobserverAction -> Maybe (TMVar (Maybe MC.SomeException))
activeThread_maybe :: JobserverAction -> Maybe (TMVar (Maybe SomeException))
activeThread_maybe JobserverAction
Idle = Maybe (TMVar (Maybe SomeException))
forall a. Maybe a
Nothing
activeThread_maybe (Acquiring { threadFinished :: JobserverAction -> TMVar (Maybe SomeException)
threadFinished = TMVar (Maybe SomeException)
tmvar }) = TMVar (Maybe SomeException) -> Maybe (TMVar (Maybe SomeException))
forall a. a -> Maybe a
Just TMVar (Maybe SomeException)
tmvar
guardAcquire :: JobResources -> Bool
guardAcquire :: JobResources -> Bool
guardAcquire ( Jobs { Int
tokensFree :: JobResources -> Int
tokensFree :: Int
tokensFree, OrdList (TMVar ())
jobsWaiting :: JobResources -> OrdList (TMVar ())
jobsWaiting :: OrdList (TMVar ())
jobsWaiting } )
= Int
tokensFree Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
0 Bool -> Bool -> Bool
&& Bool -> Bool
not (OrdList (TMVar ()) -> Bool
forall a. OrdList a -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null OrdList (TMVar ())
jobsWaiting)
guardRelease :: JobResources -> Bool
guardRelease :: JobResources -> Bool
guardRelease ( Jobs { Int
tokensFree :: JobResources -> Int
tokensFree :: Int
tokensFree, Int
tokensOwned :: JobResources -> Int
tokensOwned :: Int
tokensOwned, OrdList (TMVar ())
jobsWaiting :: JobResources -> OrdList (TMVar ())
jobsWaiting :: OrdList (TMVar ())
jobsWaiting } )
= OrdList (TMVar ()) -> Bool
forall a. OrdList a -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null OrdList (TMVar ())
jobsWaiting Bool -> Bool -> Bool
&& Int
tokensFree Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
0 Bool -> Bool -> Bool
&& Int
tokensOwned Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
1
acquireJob :: TVar JobResources -> IO ()
acquireJob :: TVar JobResources -> IO ()
acquireJob TVar JobResources
jobs_tvar = do
(job_tmvar, _jobs0) <- String
-> STM ((TMVar (), JobResources), Maybe JobResources)
-> IO (TMVar (), JobResources)
forall a. String -> STM (a, Maybe JobResources) -> IO a
tracedAtomically String
"acquire" (STM ((TMVar (), JobResources), Maybe JobResources)
-> IO (TMVar (), JobResources))
-> STM ((TMVar (), JobResources), Maybe JobResources)
-> IO (TMVar (), JobResources)
forall a b. (a -> b) -> a -> b
$
TVar JobResources
-> (JobResources -> STM ((TMVar (), JobResources), JobResources))
-> STM ((TMVar (), JobResources), Maybe JobResources)
forall a.
HasCallStack =>
TVar JobResources
-> (JobResources -> STM (a, JobResources))
-> STM (a, Maybe JobResources)
modifyJobResources TVar JobResources
jobs_tvar \ JobResources
jobs -> do
job_tmvar <- STM (TMVar ())
forall a. STM (TMVar a)
newEmptyTMVar
return ((job_tmvar, jobs), addJob job_tmvar jobs)
atomically $ takeTMVar job_tmvar
releaseJob :: TVar JobResources -> IO ()
releaseJob :: TVar JobResources -> IO ()
releaseJob TVar JobResources
jobs_tvar = do
String -> STM ((), Maybe JobResources) -> IO ()
forall a. String -> STM (a, Maybe JobResources) -> IO a
tracedAtomically String
"release" do
TVar JobResources
-> (JobResources -> STM ((), JobResources))
-> STM ((), Maybe JobResources)
forall a.
HasCallStack =>
TVar JobResources
-> (JobResources -> STM (a, JobResources))
-> STM (a, Maybe JobResources)
modifyJobResources TVar JobResources
jobs_tvar \ JobResources
jobs -> do
Bool -> SDoc -> STM ()
forall (m :: * -> *).
(HasCallStack, Applicative m) =>
Bool -> SDoc -> m ()
massertPpr (JobResources -> Int
tokensFree JobResources
jobs Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
< JobResources -> Int
tokensOwned JobResources
jobs)
(String -> SDoc
forall doc. IsLine doc => String -> doc
text String
"releaseJob: more free jobs than owned jobs!")
((), JobResources) -> STM ((), JobResources)
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return ((), JobResources -> JobResources
addFreeToken JobResources
jobs)
drainActiveChild :: Jobserver -> IO ()
drainActiveChild :: Jobserver -> IO ()
drainActiveChild (Jobserver { activeChild :: Jobserver -> TVar (Maybe (ThreadId, TMVar (Maybe SomeException)))
activeChild = TVar (Maybe (ThreadId, TMVar (Maybe SomeException)))
active_tvar }) = do
mb <- TVar (Maybe (ThreadId, TMVar (Maybe SomeException)))
-> IO (Maybe (ThreadId, TMVar (Maybe SomeException)))
forall a. TVar a -> IO a
readTVarIO TVar (Maybe (ThreadId, TMVar (Maybe SomeException)))
active_tvar
for_ mb $ \(ThreadId
tid, TMVar (Maybe SomeException)
tmv) -> do
ThreadId -> IO ()
killThread ThreadId
tid
IO (Maybe SomeException) -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO (Maybe SomeException) -> IO ())
-> IO (Maybe SomeException) -> IO ()
forall a b. (a -> b) -> a -> b
$ STM (Maybe SomeException) -> IO (Maybe SomeException)
forall a. STM a -> IO a
atomically (TMVar (Maybe SomeException) -> STM (Maybe SomeException)
forall a. TMVar a -> STM a
takeTMVar TMVar (Maybe SomeException)
tmv)
STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar (Maybe (ThreadId, TMVar (Maybe SomeException)))
-> Maybe (ThreadId, TMVar (Maybe SomeException)) -> STM ()
forall a. TVar a -> a -> STM ()
writeTVar TVar (Maybe (ThreadId, TMVar (Maybe SomeException)))
active_tvar Maybe (ThreadId, TMVar (Maybe SomeException))
forall a. Maybe a
Nothing
releaseAllHeld :: Jobserver -> IO ()
releaseAllHeld :: Jobserver -> IO ()
releaseAllHeld (Jobserver { jobs :: Jobserver -> TVar JobResources
jobs = TVar JobResources
jobs_tvar }) = do
Jobs { heldTokens = toks } <- TVar JobResources -> IO JobResources
forall a. TVar a -> IO a
readTVarIO TVar JobResources
jobs_tvar
forM_ toks $ \SemaphoreToken
t ->
IO (Either SomeException ()) -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO (Either SomeException ()) -> IO ())
-> IO (Either SomeException ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) e a.
(HasCallStack, MonadCatch m, Exception e) =>
m a -> m (Either e a)
MC.try @_ @MC.SomeException (HasCallStack => SemaphoreToken -> IO ()
SemaphoreToken -> IO ()
releaseSemaphoreToken SemaphoreToken
t)
dispatchTokens :: JobResources -> STM JobResources
dispatchTokens :: JobResources -> STM JobResources
dispatchTokens jobs :: JobResources
jobs@( Jobs { tokensFree :: JobResources -> Int
tokensFree = Int
toks_free, jobsWaiting :: JobResources -> OrdList (TMVar ())
jobsWaiting = OrdList (TMVar ())
wait } )
| Int
toks_free Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
0
, TMVar ()
next `ConsOL` OrdList (TMVar ())
rest <- OrdList (TMVar ())
wait
= do
TMVar () -> () -> STM ()
forall a. TMVar a -> a -> STM ()
putTMVar TMVar ()
next ()
let jobs' :: JobResources
jobs' = JobResources
jobs { tokensFree = toks_free - 1, jobsWaiting = rest }
JobResources -> STM JobResources
dispatchTokens JobResources
jobs'
| Bool
otherwise
= JobResources -> STM JobResources
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return JobResources
jobs
modifyJobResources :: HasCallStack => TVar JobResources
-> (JobResources -> STM (a, JobResources))
-> STM (a, Maybe JobResources)
modifyJobResources :: forall a.
HasCallStack =>
TVar JobResources
-> (JobResources -> STM (a, JobResources))
-> STM (a, Maybe JobResources)
modifyJobResources TVar JobResources
jobs_tvar JobResources -> STM (a, JobResources)
action = do
old_jobs <- TVar JobResources -> STM JobResources
forall a. TVar a -> STM a
readTVar TVar JobResources
jobs_tvar
(a, jobs) <- action old_jobs
massertPpr (null (jobsWaiting jobs) || tokensFree jobs >= tokensFree old_jobs) $
vcat [ text "modiyJobResources: pending jobs but fewer free tokens" ]
dispatched_jobs <- dispatchTokens jobs
writeTVar jobs_tvar dispatched_jobs
return (a, Just dispatched_jobs)
tracedAtomically_ :: String -> STM (Maybe JobResources) -> IO ()
tracedAtomically_ :: String -> STM (Maybe JobResources) -> IO ()
tracedAtomically_ String
s STM (Maybe JobResources)
act = String -> STM ((), Maybe JobResources) -> IO ()
forall a. String -> STM (a, Maybe JobResources) -> IO a
tracedAtomically String
s (((),) (Maybe JobResources -> ((), Maybe JobResources))
-> STM (Maybe JobResources) -> STM ((), Maybe JobResources)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> STM (Maybe JobResources)
act)
tracedAtomically :: String -> STM (a, Maybe JobResources) -> IO a
tracedAtomically :: forall a. String -> STM (a, Maybe JobResources) -> IO a
tracedAtomically String
origin STM (a, Maybe JobResources)
act = do
(a, mjr) <- STM (a, Maybe JobResources) -> IO (a, Maybe JobResources)
forall a. STM a -> IO a
atomically STM (a, Maybe JobResources)
act
forM_ mjr $ \ JobResources
jr -> do
String -> IO ()
traceEventIO (String
"jsem:" String -> String -> String
forall a. [a] -> [a] -> [a]
++ String -> JobResources -> String
renderJobResources String
origin JobResources
jr)
return a
renderJobResources :: String -> JobResources -> String
renderJobResources :: String -> JobResources -> String
renderJobResources String
origin (Jobs Int
own Int
free OrdList (TMVar ())
pending [SemaphoreToken]
_heldToks) = SDoc -> String
showSDocUnsafe (SDoc -> String) -> SDoc -> String
forall a b. (a -> b) -> a -> b
$ JsonDoc -> SDoc
renderJSON (JsonDoc -> SDoc) -> JsonDoc -> SDoc
forall a b. (a -> b) -> a -> b
$
[(String, JsonDoc)] -> JsonDoc
JSObject [ (String
"name", String -> JsonDoc
JSString String
origin)
, (String
"owned", Int -> JsonDoc
JSInt Int
own)
, (String
"free", Int -> JsonDoc
JSInt Int
free)
, (String
"pending", Int -> JsonDoc
JSInt (OrdList (TMVar ()) -> Int
forall a. OrdList a -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length OrdList (TMVar ())
pending) )
]
acquireThread :: Jobserver -> IO JobserverAction
acquireThread :: Jobserver -> IO JobserverAction
acquireThread (Jobserver { jSemaphore :: Jobserver -> ClientSemaphore
jSemaphore = ClientSemaphore
sem, jobs :: Jobserver -> TVar JobResources
jobs = TVar JobResources
jobs_tvar, activeChild :: Jobserver -> TVar (Maybe (ThreadId, TMVar (Maybe SomeException)))
activeChild = TVar (Maybe (ThreadId, TMVar (Maybe SomeException)))
active_tvar }) = do
threadFinished_tmvar <- IO (TMVar (Maybe SomeException))
forall a. IO (TMVar a)
newEmptyTMVarIO
tid <- MC.mask_ $ do
tid <- forkIO $ do
wait_res <- MC.try @_ @MC.SomeException $ waitOnSemaphore sem
tracedAtomically_ "acquire_thread" do
(r, jb) <- case wait_res of
Left (SomeException
e :: MC.SomeException) -> do
(Maybe SomeException, Maybe JobResources)
-> STM (Maybe SomeException, Maybe JobResources)
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return ((Maybe SomeException, Maybe JobResources)
-> STM (Maybe SomeException, Maybe JobResources))
-> (Maybe SomeException, Maybe JobResources)
-> STM (Maybe SomeException, Maybe JobResources)
forall a b. (a -> b) -> a -> b
$ (SomeException -> Maybe SomeException
forall a. a -> Maybe a
Just SomeException
e, Maybe JobResources
forall a. Maybe a
Nothing)
Right SemaphoreToken
tok -> do
TVar JobResources
-> (JobResources -> STM (Maybe SomeException, JobResources))
-> STM (Maybe SomeException, Maybe JobResources)
forall a.
HasCallStack =>
TVar JobResources
-> (JobResources -> STM (a, JobResources))
-> STM (a, Maybe JobResources)
modifyJobResources TVar JobResources
jobs_tvar \ JobResources
jobs ->
(Maybe SomeException, JobResources)
-> STM (Maybe SomeException, JobResources)
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return (Maybe SomeException
forall a. Maybe a
Nothing, SemaphoreToken -> JobResources -> JobResources
addToken SemaphoreToken
tok JobResources
jobs)
putTMVar threadFinished_tmvar r
return jb
atomically $ writeTVar active_tvar (Just (tid, threadFinished_tmvar))
return tid
labelThread tid "acquire_thread"
return $ Acquiring { activeThreadId = tid
, threadFinished = threadFinished_tmvar }
releaseThread :: Jobserver -> IO JobserverAction
releaseThread :: Jobserver -> IO JobserverAction
releaseThread (Jobserver { jobs :: Jobserver -> TVar JobResources
jobs = TVar JobResources
jobs_tvar }) = do
threadFinished_tmvar <- IO (TMVar (Maybe SomeException))
forall a. IO (TMVar a)
newEmptyTMVarIO
MC.mask_ do
mb_tok
<- tracedAtomically "pre_release" $
modifyJobResources jobs_tvar \ JobResources
jobs ->
if JobResources -> Bool
guardRelease JobResources
jobs
then let (SemaphoreToken
tok, JobResources
jobs') = JobResources -> (SemaphoreToken, JobResources)
removeOwnedToken (JobResources -> (SemaphoreToken, JobResources))
-> JobResources -> (SemaphoreToken, JobResources)
forall a b. (a -> b) -> a -> b
$ JobResources -> JobResources
removeFreeToken JobResources
jobs
in (Maybe SemaphoreToken, JobResources)
-> STM (Maybe SemaphoreToken, JobResources)
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return (SemaphoreToken -> Maybe SemaphoreToken
forall a. a -> Maybe a
Just SemaphoreToken
tok, JobResources
jobs')
else (Maybe SemaphoreToken, JobResources)
-> STM (Maybe SemaphoreToken, JobResources)
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return (Maybe SemaphoreToken
forall a. Maybe a
Nothing, JobResources
jobs)
case mb_tok of
Maybe SemaphoreToken
Nothing ->
JobserverAction -> IO JobserverAction
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return JobserverAction
Idle
Just SemaphoreToken
tok -> do
tid <- IO () -> IO ThreadId
forkIO (IO () -> IO ThreadId) -> IO () -> IO ThreadId
forall a b. (a -> b) -> a -> b
$ do
x <- forall (m :: * -> *) e a.
(HasCallStack, MonadCatch m, Exception e) =>
m a -> m (Either e a)
MC.try @_ @MC.SomeException (IO () -> IO (Either SomeException ()))
-> IO () -> IO (Either SomeException ())
forall a b. (a -> b) -> a -> b
$ HasCallStack => SemaphoreToken -> IO ()
SemaphoreToken -> IO ()
releaseSemaphoreToken SemaphoreToken
tok
tracedAtomically_ "post-release" $ do
(r, jobs) <- case x of
Left (SomeException
e :: MC.SomeException) -> do
TVar JobResources
-> (JobResources -> STM (Maybe SomeException, JobResources))
-> STM (Maybe SomeException, Maybe JobResources)
forall a.
HasCallStack =>
TVar JobResources
-> (JobResources -> STM (a, JobResources))
-> STM (a, Maybe JobResources)
modifyJobResources TVar JobResources
jobs_tvar \ JobResources
jobs ->
(Maybe SomeException, JobResources)
-> STM (Maybe SomeException, JobResources)
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return (SomeException -> Maybe SomeException
forall a. a -> Maybe a
Just SomeException
e, SemaphoreToken -> JobResources -> JobResources
addToken SemaphoreToken
tok JobResources
jobs)
Right ()
_ -> do
(Maybe SomeException, Maybe JobResources)
-> STM (Maybe SomeException, Maybe JobResources)
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return (Maybe SomeException
forall a. Maybe a
Nothing, Maybe JobResources
forall a. Maybe a
Nothing)
putTMVar threadFinished_tmvar r
return jobs
labelThread tid "release_thread"
return Idle
tryAcquire :: JobserverOptions
-> Jobserver
-> JobserverState
-> STM (IO JobserverState)
tryAcquire :: JobserverOptions
-> Jobserver -> JobserverState -> STM (IO JobserverState)
tryAcquire JobserverOptions
opts js :: Jobserver
js@( Jobserver { jobs :: Jobserver -> TVar JobResources
jobs = TVar JobResources
jobs_tvar })
st :: JobserverState
st@( JobserverState { jobserverAction :: JobserverState -> JobserverAction
jobserverAction = JobserverAction
Idle } )
= do
jobs <- TVar JobResources -> STM JobResources
forall a. TVar a -> STM a
readTVar TVar JobResources
jobs_tvar
guard $ guardAcquire jobs
return do
action <- acquireThread js
can_release_tvar <- registerDelay $ (releaseDebounce opts * 1000)
return $ st { jobserverAction = action
, canReleaseToken = can_release_tvar }
tryAcquire JobserverOptions
_ Jobserver
_ JobserverState
_ = STM (IO JobserverState)
forall a. STM a
retry
tryRelease :: Jobserver
-> JobserverState
-> STM (IO JobserverState)
tryRelease :: Jobserver -> JobserverState -> STM (IO JobserverState)
tryRelease sjs :: Jobserver
sjs@( Jobserver { jobs :: Jobserver -> TVar JobResources
jobs = TVar JobResources
jobs_tvar } )
st :: JobserverState
st@( JobserverState
{ jobserverAction :: JobserverState -> JobserverAction
jobserverAction = JobserverAction
Idle
, canReleaseToken :: JobserverState -> TVar Bool
canReleaseToken = TVar Bool
can_release_tvar } )
= do
jobs <- TVar JobResources -> STM JobResources
forall a. TVar a -> STM a
readTVar TVar JobResources
jobs_tvar
guard $ guardRelease jobs
can_release <- readTVar can_release_tvar
guard can_release
return do
action <- releaseThread sjs
return $ st { jobserverAction = action }
tryRelease Jobserver
_ JobserverState
_ = STM (IO JobserverState)
forall a. STM a
retry
tryNoticeIdle :: JobserverOptions
-> Jobserver
-> JobserverState
-> STM (IO JobserverState)
tryNoticeIdle :: JobserverOptions
-> Jobserver -> JobserverState -> STM (IO JobserverState)
tryNoticeIdle JobserverOptions
opts (Jobserver { jobs :: Jobserver -> TVar JobResources
jobs = TVar JobResources
jobs_tvar, activeChild :: Jobserver -> TVar (Maybe (ThreadId, TMVar (Maybe SomeException)))
activeChild = TVar (Maybe (ThreadId, TMVar (Maybe SomeException)))
active_tvar }) JobserverState
jobserver_state
| Just TMVar (Maybe SomeException)
threadFinished_tmvar <- JobserverAction -> Maybe (TMVar (Maybe SomeException))
activeThread_maybe (JobserverAction -> Maybe (TMVar (Maybe SomeException)))
-> JobserverAction -> Maybe (TMVar (Maybe SomeException))
forall a b. (a -> b) -> a -> b
$ JobserverState -> JobserverAction
jobserverAction JobserverState
jobserver_state
= TVar Bool -> TMVar (Maybe SomeException) -> STM (IO JobserverState)
sync_num_caps (JobserverState -> TVar Bool
canChangeNumCaps JobserverState
jobserver_state) TMVar (Maybe SomeException)
threadFinished_tmvar
| Bool
otherwise
= STM (IO JobserverState)
forall a. STM a
retry
where
sync_num_caps :: TVar Bool
-> TMVar (Maybe MC.SomeException)
-> STM (IO JobserverState)
sync_num_caps :: TVar Bool -> TMVar (Maybe SomeException) -> STM (IO JobserverState)
sync_num_caps TVar Bool
can_change_numcaps_tvar TMVar (Maybe SomeException)
threadFinished_tmvar = do
mb_ex <- TMVar (Maybe SomeException) -> STM (Maybe SomeException)
forall a. TMVar a -> STM a
takeTMVar TMVar (Maybe SomeException)
threadFinished_tmvar
for_ mb_ex MC.throwM
writeTVar active_tvar Nothing
Jobs { tokensOwned } <- readTVar jobs_tvar
can_change_numcaps <- readTVar can_change_numcaps_tvar
guard can_change_numcaps
return do
x <- getNumCapabilities
can_change_numcaps_tvar_2 <-
if x == tokensOwned
then return can_change_numcaps_tvar
else do
setNumCapabilities tokensOwned
registerDelay $ (setNumCapsDebounce opts * 1000)
return $
jobserver_state
{ jobserverAction = Idle
, canChangeNumCaps = can_change_numcaps_tvar_2 }
tryStopThread :: TVar JobResources
-> JobserverState
-> STM (IO JobserverState)
tryStopThread :: TVar JobResources -> JobserverState -> STM (IO JobserverState)
tryStopThread TVar JobResources
jobs_tvar JobserverState
jsj = do
case JobserverState -> JobserverAction
jobserverAction JobserverState
jsj of
Acquiring { activeThreadId :: JobserverAction -> ThreadId
activeThreadId = ThreadId
tid } -> do
jobs <- TVar JobResources -> STM JobResources
forall a. TVar a -> STM a
readTVar TVar JobResources
jobs_tvar
guard $ null (jobsWaiting jobs)
return do
killThread tid
return $ jsj { jobserverAction = Idle }
JobserverAction
_ -> STM (IO JobserverState)
forall a. STM a
retry
jobserverLoop :: JobserverOptions -> Jobserver -> IO ()
jobserverLoop :: JobserverOptions -> Jobserver -> IO ()
jobserverLoop JobserverOptions
opts sjs :: Jobserver
sjs@(Jobserver { jobs :: Jobserver -> TVar JobResources
jobs = TVar JobResources
jobs_tvar })
= do
true_tvar <- Bool -> IO (TVar Bool)
forall a. a -> IO (TVar a)
newTVarIO Bool
True
let init_state :: JobserverState
init_state =
JobserverState
{ jobserverAction :: JobserverAction
jobserverAction = JobserverAction
Idle
, canChangeNumCaps :: TVar Bool
canChangeNumCaps = TVar Bool
true_tvar
, canReleaseToken :: TVar Bool
canReleaseToken = TVar Bool
true_tvar }
loop init_state
where
loop :: JobserverState -> IO ()
loop JobserverState
s = do
action <- STM (IO JobserverState) -> IO (IO JobserverState)
forall a. STM a -> IO a
atomically (STM (IO JobserverState) -> IO (IO JobserverState))
-> STM (IO JobserverState) -> IO (IO JobserverState)
forall a b. (a -> b) -> a -> b
$ [STM (IO JobserverState)] -> STM (IO JobserverState)
forall (t :: * -> *) (f :: * -> *) a.
(Foldable t, Alternative f) =>
t (f a) -> f a
asum ([STM (IO JobserverState)] -> STM (IO JobserverState))
-> [STM (IO JobserverState)] -> STM (IO JobserverState)
forall a b. (a -> b) -> a -> b
$ (\JobserverState -> STM (IO JobserverState)
x -> JobserverState -> STM (IO JobserverState)
x JobserverState
s) ((JobserverState -> STM (IO JobserverState))
-> STM (IO JobserverState))
-> [JobserverState -> STM (IO JobserverState)]
-> [STM (IO JobserverState)]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$>
[ Jobserver -> JobserverState -> STM (IO JobserverState)
tryRelease Jobserver
sjs
, JobserverOptions
-> Jobserver -> JobserverState -> STM (IO JobserverState)
tryAcquire JobserverOptions
opts Jobserver
sjs
, JobserverOptions
-> Jobserver -> JobserverState -> STM (IO JobserverState)
tryNoticeIdle JobserverOptions
opts Jobserver
sjs
, TVar JobResources -> JobserverState -> STM (IO JobserverState)
tryStopThread TVar JobResources
jobs_tvar
]
s <- action
loop s
makeJobserver :: SemaphoreIdentifier -> IO (AbstractSem, IO ())
makeJobserver :: String -> IO (AbstractSem, IO ())
makeJobserver String
sem_ident = do
semaphore <- String -> IO (Either SemaphoreError ClientSemaphore)
openSemaphore String
sem_ident IO (Either SemaphoreError ClientSemaphore)
-> (Either SemaphoreError ClientSemaphore -> IO ClientSemaphore)
-> IO ClientSemaphore
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= (SemaphoreError -> IO ClientSemaphore)
-> (ClientSemaphore -> IO ClientSemaphore)
-> Either SemaphoreError ClientSemaphore
-> IO ClientSemaphore
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either SemaphoreError -> IO ClientSemaphore
forall e a. (HasCallStack, Exception e) => e -> IO a
forall (m :: * -> *) e a.
(MonadThrow m, HasCallStack, Exception e) =>
e -> m a
MC.throwM ClientSemaphore -> IO ClientSemaphore
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
let
init_jobs =
Jobs { tokensOwned :: Int
tokensOwned = Int
1
, tokensFree :: Int
tokensFree = Int
1
, jobsWaiting :: OrdList (TMVar ())
jobsWaiting = OrdList (TMVar ())
forall a. OrdList a
NilOL
, heldTokens :: [SemaphoreToken]
heldTokens = []
}
jobs_tvar <- newTVarIO init_jobs
active_tvar <- newTVarIO Nothing
let
opts = JobserverOptions
defaultJobserverOptions
sjs = Jobserver { jSemaphore :: ClientSemaphore
jSemaphore = ClientSemaphore
semaphore
, jobs :: TVar JobResources
jobs = TVar JobResources
jobs_tvar
, activeChild :: TVar (Maybe (ThreadId, TMVar (Maybe SomeException)))
activeChild = TVar (Maybe (ThreadId, TMVar (Maybe SomeException)))
active_tvar }
loop_finished_mvar <- newEmptyMVar
loop_tid <- forkIOWithUnmask \ forall a. IO a -> IO a
unmask -> do
r <- IO () -> IO (Either SomeException ())
forall e a. Exception e => IO a -> IO (Either e a)
try (IO () -> IO (Either SomeException ()))
-> IO () -> IO (Either SomeException ())
forall a b. (a -> b) -> a -> b
$ IO () -> IO ()
forall a. IO a -> IO a
unmask (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ JobserverOptions -> Jobserver -> IO ()
jobserverLoop JobserverOptions
opts Jobserver
sjs
drainActiveChild sjs
releaseAllHeld sjs
putMVar loop_finished_mvar $
case r of
Left SomeException
e
| Just AsyncException
ThreadKilled <- SomeException -> Maybe AsyncException
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
e
-> Maybe SomeException
forall a. Maybe a
Nothing
| Bool
otherwise
-> SomeException -> Maybe SomeException
forall a. a -> Maybe a
Just SomeException
e
Right () -> Maybe SomeException
forall a. Maybe a
Nothing
labelThread loop_tid "job_server"
let
acquireSem = TVar JobResources -> IO ()
acquireJob TVar JobResources
jobs_tvar
releaseSem = TVar JobResources -> IO ()
releaseJob TVar JobResources
jobs_tvar
cleanupSem = do
ThreadId -> IO ()
killThread ThreadId
loop_tid
mb_ex <- MVar (Maybe SomeException) -> IO (Maybe SomeException)
forall a. MVar a -> IO a
takeMVar MVar (Maybe SomeException)
loop_finished_mvar
for_ mb_ex MC.throwM
return (AbstractSem{..}, cleanupSem)
runJSemAbstractSem :: SemaphoreIdentifier
-> (AbstractSem -> IO a)
-> IO a
runJSemAbstractSem :: forall a. String -> (AbstractSem -> IO a) -> IO a
runJSemAbstractSem String
sem_ident AbstractSem -> IO a
action = ((forall a. IO a -> IO a) -> IO a) -> IO a
forall b.
HasCallStack =>
((forall a. IO a -> IO a) -> IO b) -> IO b
forall (m :: * -> *) b.
(MonadMask m, HasCallStack) =>
((forall a. m a -> m a) -> m b) -> m b
MC.mask \ forall a. IO a -> IO a
unmask -> do
(abs, cleanup) <- String -> IO (AbstractSem, IO ())
makeJobserver String
sem_ident
r <- try $ unmask $ action abs
case r of
Left (SomeException
e1 :: MC.SomeException) -> do
(_ :: Either MC.SomeException ()) <- IO () -> IO (Either SomeException ())
forall (m :: * -> *) e a.
(HasCallStack, MonadCatch m, Exception e) =>
m a -> m (Either e a)
MC.try IO ()
cleanup
MC.throwM e1
Right a
x -> IO ()
cleanup IO () -> a -> IO a
forall (f :: * -> *) a b. Functor f => f a -> b -> f b
$> a
x
#endif