diff --git a/core-program.cabal b/core-program.cabal --- a/core-program.cabal +++ b/core-program.cabal @@ -1,11 +1,11 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.34.4. +-- This file has been generated from package.yaml by hpack version 0.35.0. -- -- see: https://github.com/sol/hpack name: core-program -version: 0.5.2.0 +version: 0.6.0.1 synopsis: Opinionated Haskell Interoperability description: A library to help build command-line programs, both tools and longer-running daemons. @@ -27,7 +27,7 @@ license-file: LICENSE build-type: Simple tested-with: - GHC == 8.10.7, GHC == 9.2.2 + GHC == 8.10.7, GHC == 9.2.4 source-repository head type: git @@ -55,10 +55,9 @@ lib ghc-options: -Wall -Wwarn -fwarn-tabs build-depends: - async - , base >=4.11 && <5 + base >=4.11 && <5 , bytestring - , core-data >=0.3.4 + , core-data >=0.3.8 , core-text >=0.3.8 , directory , exceptions diff --git a/lib/Core/Program/Context.hs b/lib/Core/Program/Context.hs --- a/lib/Core/Program/Context.hs +++ b/lib/Core/Program/Context.hs @@ -37,8 +37,10 @@ subProgram, ) where +import Control.Concurrent (ThreadId) import Control.Concurrent.MVar (MVar, newEmptyMVar, newMVar, putMVar, readMVar) import Control.Concurrent.STM.TQueue (TQueue, newTQueueIO) +import Control.Concurrent.STM.TVar (TVar, newTVarIO) import Control.Exception.Safe qualified as Safe (throw) import Control.Monad.Catch (MonadCatch, MonadMask, MonadThrow (throwM)) import Control.Monad.Reader.Class (MonadReader (..)) @@ -175,6 +177,7 @@ , outputChannelFrom :: TQueue (Maybe Rope) -- communication channels , telemetryChannelFrom :: TQueue (Maybe Datum) -- machinery for telemetry , telemetryForwarderFrom :: Maybe Forwarder + , currentScopeFrom :: TVar (Set ThreadId) , currentDatumFrom :: MVar Datum , applicationDataFrom :: MVar τ } @@ -356,11 +359,12 @@ out <- newTQueueIO tel <- newTQueueIO - v <- newMVar (emptyDatum) + scope <- newTVarIO emptySet + v <- newMVar emptyDatum u <- newMVar t - return - $! Context + return $! + Context { programNameFrom = n , terminalWidthFrom = columns , terminalColouredFrom = coloured @@ -374,6 +378,7 @@ , outputChannelFrom = out , telemetryChannelFrom = tel , telemetryForwarderFrom = Nothing + , currentScopeFrom = scope , currentDatumFrom = v , applicationDataFrom = u } diff --git a/lib/Core/Program/Execute.hs b/lib/Core/Program/Execute.hs --- a/lib/Core/Program/Execute.hs +++ b/lib/Core/Program/Execute.hs @@ -104,26 +104,22 @@ lookupEnvironmentValue, ) where -import Control.Concurrent (threadDelay) -import Control.Concurrent.Async ( - ExceptionInLinkedThread (..), - ) -import Control.Concurrent.Async qualified as Async ( - async, - cancel, - race, - race_, - wait, +import Control.Concurrent ( + forkFinally, + forkIO, + killThread, + myThreadId, + threadDelay, ) import Control.Concurrent.MVar ( MVar, modifyMVar_, + newEmptyMVar, putMVar, readMVar, - ) -import Control.Concurrent.STM ( - atomically, + tryPutMVar, ) +import Control.Concurrent.STM (atomically) import Control.Concurrent.STM.TQueue ( TQueue, readTQueue, @@ -131,17 +127,20 @@ unGetTQueue, writeTQueue, ) +import Control.Concurrent.STM.TVar ( + readTVarIO, + ) import Control.Exception qualified as Base (throwIO) import Control.Exception.Safe qualified as Safe ( catch, - catchesAsync, throw, ) import Control.Monad ( + forM_, + forever, void, when, ) -import Control.Monad.Catch (Handler (..)) import Control.Monad.Reader.Class (MonadReader (ask)) import Core.Data.Clock import Core.Data.Structures @@ -171,51 +170,9 @@ findExecutable, ) import System.Exit (ExitCode (..)) -import System.Posix.Process qualified as Posix (exitImmediately) import System.Process.Typed (nullStream, proc, readProcess, setStdin) import Prelude hiding (log) --- --- If an exception escapes, we'll catch it here. The displayException value --- for some exceptions is really quit unhelpful, so we pattern match the --- wrapping gumpf away for cases as we encounter them. The final entry is the --- catch-all. --- --- Note this is called via Safe.catchesAsync because we want to be able to --- strip out ExceptionInLinkedThread (which is asynchronous and otherwise --- reasonably special) from the final output message. --- -escapeHandlers :: Context c -> [Handler IO ExitCode] -escapeHandlers context = - [ Handler (\(code :: ExitCode) -> pure code) - , Handler (\(ExceptionInLinkedThread _ e) -> bail e) - , Handler (\(e :: SomeException) -> bail e) - ] - where - bail :: Exception e => e -> IO ExitCode - bail e = - let text = intoRope (displayException e) - in do - subProgram context $ do - setVerbosityLevel Debug - critical text - pure (ExitFailure 127) - --- --- If an exception occurs in one of the output handlers, its failure causes --- a subsequent race condition when the program tries to clean up and drain --- the queues. So we use `exitImmediately` (which we normally avoid, as it --- unhelpfully destroys the parent process if you're in ghci) because we --- really need the process to go down and we're in an inconsistent state --- where debug or console output is no longer possible. --- -collapseHandler :: String -> SomeException -> IO () -collapseHandler problem e = do - putStr "error: " - putStrLn problem - print e - Posix.exitImmediately (ExitFailure 99) - {- | Trap any exceptions coming out of the given Program action, and discard them. The one and only time you want this is inside an endless loop: @@ -288,65 +245,81 @@ forwarder = telemetryForwarderFrom context -- set up signal handlers - _ <- - Async.async $ do - setupSignalHandlers quit level + _ <- forkIO $ do + setupSignalHandlers quit level -- set up standard output - o <- - Async.async $ do - processStandardOutput out + vo <- newEmptyMVar + _ <- + forkFinally + (processStandardOutput out) + (\_ -> putMVar vo ()) -- set up debug logger - l <- - Async.async $ do - processTelemetryMessages forwarder level out tel + vl <- newEmptyMVar + _ <- + forkFinally + (processTelemetryMessages forwarder level out tel) + (\_ -> putMVar vl ()) -- run actual program, ensuring to grab any otherwise uncaught exceptions. - code <- - Safe.catchesAsync + t1 <- forkIO $ do + Safe.catch ( do - result <- - Async.race - ( do - code <- readMVar quit - pure code - ) - ( do - -- execute actual "main" - _ <- subProgram context program - pure () - ) - - case result of - Left code' -> pure code' - Right () -> pure ExitSuccess + -- + -- execute actual "main". Note that we're not passing the + -- Scope into the program's Context; it stays the default + -- Nothing because the outer Scope is none of the + -- program's business and we absolutely don't want an + -- awaitAll to sit there and block on our machinery + -- threads. + -- + -- We use tryPutMVar here (rather than putMVar) because we + -- might already be on the way out and need to not block. + -- + _ <- subProgram context program + _ <- tryPutMVar quit ExitSuccess + pure () ) - (escapeHandlers context) + ( \(e :: SomeException) -> do + let text = intoRope (displayException e) + subProgram context $ do + setVerbosityLevel Debug + critical text + _ <- tryPutMVar quit (ExitFailure 127) + pure () + ) - -- instruct handlers to finish, and wait for the message queues to drain. - -- Allow 0.1 seconds, then timeout, in case something has gone wrong and - -- queues don't empty. - Async.race_ - ( do - atomically $ do - writeTQueue tel Nothing + -- wait for indication to terminate + code <- readMVar quit - Async.wait l + -- kill main thread + killThread t1 - atomically $ do - writeTQueue out Nothing + -- instruct handlers to finish, and wait for the message queues to + -- drain. Allow 10 seconds, then timeout, in case something has gone + -- wrong and queues don't empty. - Async.wait o - ) - ( do - threadDelay 10000000 + _ <- forkIO $ do + threadDelay 10000000 + putStrLn "error: Timeout" + Safe.throw (ExitFailure 99) - Async.cancel l - Async.cancel o - putStrLn "error: Timeout" - ) + _ <- forkIO $ do + let scope = currentScopeFrom context + pointers <- readTVarIO scope + forM_ pointers killThread + atomically $ do + writeTQueue tel Nothing + + readMVar vl + + atomically $ do + writeTQueue out Nothing + + readMVar vo + hFlush stdout -- exiting this way avoids "Exception: ExitSuccess" noise in GHCi @@ -356,9 +329,7 @@ processStandardOutput :: TQueue (Maybe Rope) -> IO () processStandardOutput out = - Safe.catch - (loop) - (collapseHandler "output processing collapsed") + loop where loop :: IO () loop = do @@ -370,6 +341,7 @@ Just text -> do hWrite stdout text B.hPut stdout (C.singleton '\n') + hFlush stdout loop -- @@ -393,9 +365,7 @@ Just _ -> do ignoreForever queue processTelemetryMessages (Just processor) v out tel = do - Safe.catch - (loopForever action v out tel) - (collapseHandler "telemetry processing collapsed") + loopForever action v out tel where action = telemetryHandlerFrom processor @@ -487,24 +457,31 @@ writeTQueue out (Just message) {- | -Safely exit the program with the supplied exit code. Current output and -debug queues will be flushed, and then the process will terminate. +Safely exit the program with the supplied exit code. Current output and debug +queues will be flushed, and then the process will terminate. This function +does not return. -} --- putting to the quit MVar initiates the cleanup and exit sequence, --- but throwing the exception also aborts execution and starts unwinding --- back up the stack. +-- putting to the quit MVar initiates the cleanup and exit sequence, but +-- throwing the asynchronous exception to self also aborts execution and +-- starts unwinding back up the stack. +-- +-- forever is used here to get an IO α as the return type. terminate :: Int -> Program τ α -terminate code = +terminate code = do + context <- ask + let quit = exitSemaphoreFrom context + let exit = case code of 0 -> ExitSuccess _ -> ExitFailure code - in do - context <- ask - let quit = exitSemaphoreFrom context - liftIO $ do - putMVar quit exit - Safe.throw exit + + liftIO $ do + putMVar quit exit + self <- myThreadId + killThread self + forever $ do + threadDelay maxBound -- undocumented getVerbosityLevel :: Program τ Verbosity diff --git a/lib/Core/Program/Threads.hs b/lib/Core/Program/Threads.hs --- a/lib/Core/Program/Threads.hs +++ b/lib/Core/Program/Threads.hs @@ -25,13 +25,13 @@ -} module Core.Program.Threads ( -- * Concurrency + createScope, forkThread, forkThread_, waitThread, waitThread_, waitThread', waitThreads', - linkThread, cancelThread, -- * Helper functions @@ -45,27 +45,18 @@ unThread, ) where -import Control.Concurrent.Async (Async, AsyncCancelled) -import Control.Concurrent.Async qualified as Async ( - async, - cancel, - concurrently, - concurrently_, - link, - race, - race_, - wait, - waitCatch, - ) -import Control.Concurrent.MVar ( - newMVar, - readMVar, - ) -import Control.Exception.Safe qualified as Safe (catch, catchAsync, throw) +import Control.Concurrent (ThreadId, forkIO, killThread) +import Control.Concurrent.MVar (MVar, newEmptyMVar, newMVar, putMVar, readMVar) +import Control.Concurrent.STM (atomically) +import Control.Concurrent.STM.TVar (modifyTVar', newTVarIO, readTVarIO) +import Control.Exception.Safe qualified as Safe (catch, finally, onException, throw) import Control.Monad ( + forM, + forM_, void, ) import Control.Monad.Reader.Class (MonadReader (ask)) +import Core.Data.Structures import Core.Program.Context import Core.Program.Logging import Core.System.Base @@ -74,31 +65,71 @@ {- | A thread for concurrent computation. -(this wraps __async__'s 'Async') +(this wraps __base__'s 'Control.Concurrent.ThreadId' along with a holder for +the result of the thread) + +@since 0.6.0 -} -newtype Thread α = Thread (Async α) +data Thread α = Thread + { threadPointerOf :: ThreadId + , threadOutcomeOf :: MVar (Either SomeException α) + } -unThread :: Thread α -> Async α -unThread (Thread a) = a +unThread :: Thread α -> ThreadId +unThread = threadPointerOf {- | -Fork a thread. The child thread will run in the same @Context@ as the calling -@Program@, including sharing the user-defined application state value. +Create a scope to enclose any subsequently spawned threads as a single group. +Ordinarily threads launched in Haskell are completely indepedent. Creating a +scope allows you to operate on a set of threads as a single group with +bi-directional exception passing. This is the basis of an approach called +/structured concurrency/. +When the execution flow exits the scope, any threads that were spawned within +it that are still running will be killed. + +If any of the child threads within the scope throws an exception, the other +remaining threads will be killed and then the original exception will be +propegated to this parent thread and re-thrown. + +@since 0.6.0 +-} +createScope :: Program τ α -> Program τ α +createScope program = do + context <- ask + + liftIO $ do + scope <- newTVarIO emptySet + + let context' = + context + { currentScopeFrom = scope + } + + Safe.finally + ( do + subProgram context' program + ) + ( do + pointers <- readTVarIO scope + forM_ pointers killThread + ) + +{- | +Fork a thread. The child thread will run in the same 'Context' as the calling +'Program', including sharing the user-defined application state value. + +If you want to find out what the result of a thread was use 'waitThread' on +the 'Thread' object returned from this function. If you don't need the +result, use 'forkThread_' instead. + Threads that are launched off as children are on their own! If the code in the child thread throws an exception that is /not/ caught within that thread, the exception will kill the thread. Threads dying without telling anyone is a bit of an anti-pattern, so this library logs a warning-level log message if this happens. -If you additionally want the exception to propagate back to the parent thread -(say, for example, you want your whole program to die if any of its worker -threads fail), then call 'linkThread' after forking. If you want the other -direction, that is, if you want the forked thread to be cancelled when its -parent is cancelled, then you need to be waiting on it using 'waitThread'. - -(this wraps __async__\'s 'Control.Concurrent.Async.async' which in turn wraps -__base__'s 'Control.Concurrent.forkIO') +(this wraps __base__'s 'Control.Concurrent.forkIO') @since 0.2.7 -} @@ -107,6 +138,7 @@ context <- ask let i = startTimeFrom context let v = currentDatumFrom context + let scope = currentScopeFrom context liftIO $ do -- if someone calls resetTimer in the thread it should just be that @@ -132,21 +164,33 @@ -- fork, and run nested program - a <- Async.async $ do + outcome <- newEmptyMVar + + pointer <- forkIO $ do Safe.catch - (subProgram context' program) - ( \(e :: SomeException) -> + ( do + actual <- subProgram context' program + putMVar outcome (Right actual) + ) + ( \(e :: SomeException) -> do let text = intoRope (displayException e) - in do - subProgram context' $ do - warn "Uncaught exception in thread" - debug "e" text - Safe.throw e + subProgram context' $ do + internal "Uncaught exception ending thread" + internal ("e = " <> text) + putMVar outcome (Left e) ) - return (Thread a) + atomically $ do + modifyTVar' scope (\pointers -> insertElement pointer pointers) -{-| + return + ( Thread + { threadPointerOf = pointer + , threadOutcomeOf = outcome + } + ) + +{- | Fork a thread with 'forkThread' but do not wait for a result. This is on the assumption that the sub program will either be a side-effect and over quickly, or long-running daemon thread (presumably containing a 'Control.Monad.forever' @@ -166,24 +210,20 @@ If the current thread making this call is cancelled (as a result of being on the losing side of 'concurrentThreads' or 'raceThreads' for example, or due to -an explicit call to 'cancelThread'), then the thread you are waiting on will -be cancelled. This is necessary to ensure that child threads are not leaked if -you nest `forkThread`s. - -(this wraps __async__\'s 'Control.Concurrent.Async.wait', taking care to -ensure the behaviour described above) +the current scope exiting), then the thread you are waiting on will be +cancelled too. This is necessary to ensure that child threads are not leaked +if you nest `forkThread`s. @since 0.2.7 -} waitThread :: Thread α -> Program τ α -waitThread (Thread a) = liftIO $ do - Safe.catchAsync - (Async.wait a) - ( \(e :: AsyncCancelled) -> do - Async.cancel a - Safe.throw e - ) +waitThread thread = do + result <- waitThread' thread + case result of + Left problem -> Safe.throw problem + Right actual -> pure actual + {- | Wait for the completion of a thread, discarding its result. This is particularly useful at the end of a do-block if you're waiting on a worker @@ -209,7 +249,7 @@ @since 0.2.7 -} waitThread_ :: Thread α -> Program τ () -waitThread_ = void . waitThread +waitThread_ thread = void (waitThread thread) {- | Wait for a thread to complete, returning the result if the computation was @@ -217,25 +257,32 @@ This basically is convenience for calling `waitThread` and putting `catch` around it, but as with all the other @wait*@ functions this ensures that if -the thread waiting is cancelled the cancellation is propagated to the thread +the thread waiting is killed the cancellation is propagated to the thread being watched as well. -(this wraps __async__\'s 'Control.Concurrent.Async.waitCatch') - @since 0.4.5 -} waitThread' :: Thread α -> Program τ (Either SomeException α) -waitThread' (Thread a) = liftIO $ do - Safe.catchAsync - ( do - result <- Async.waitCatch a - pure result - ) - ( \(e :: AsyncCancelled) -> do - Async.cancel a - Safe.throw e - ) +waitThread' thread = do + context <- ask + let scope = currentScopeFrom context + let outcome = threadOutcomeOf thread + let pointer = threadPointerOf thread + liftIO $ do + Safe.onException + ( do + result <- readMVar outcome -- blocks! + atomically $ do + modifyTVar' scope (\pointers -> removeElement pointer pointers) + pure result + ) + ( do + killThread pointer + atomically $ do + modifyTVar' scope (\pointers -> removeElement pointer pointers) + ) + {- | Wait for many threads to complete. This function is intended for the scenario where you fire off a number of worker threads with `forkThread` but rather @@ -271,54 +318,53 @@ to cancel so as to avoid those threads being leaked and continuing to run as zombies. This function takes care of that. -(this extends __async__\'s 'Control.Concurrent.Async.waitCatch' to work -across a list of Threads, taking care to ensure the cancellation behaviour -described throughout this module) +(this extends 'waitThread'' to work across a list of Threads, taking care to +ensure the cancellation behaviour described throughout this module) @since 0.4.5 -} waitThreads' :: [Thread α] -> Program τ [Either SomeException α] -waitThreads' ts = liftIO $ do - let as = fmap unThread ts - Safe.catchAsync - ( do - results <- mapM Async.waitCatch as - pure results - ) - ( \(e :: AsyncCancelled) -> do - mapM_ Async.cancel as - Safe.throw e - ) +waitThreads' threads = do + context <- ask + liftIO $ do + Safe.onException + ( do + subProgram context $ do + forM threads waitThread' + ) + ( do + -- + -- This is here because if this thread is cancelled it will + -- only be _one_ of the waitThread above that receives the + -- exception. All the other child threads need to be killed + -- too. + -- -{- | -Ordinarily if an exception is thrown in a forked thread that exception is -silently swollowed. If you instead need the exception to propegate back to the -parent thread, you can \"link\" the two together using this function. + let scope = currentScopeFrom context -(this wraps __async__\'s 'Control.Concurrent.Async.link') + forM_ threads $ \thread -> do + let pointer = threadPointerOf thread + killThread pointer -@since 0.4.2 --} -linkThread :: Thread α -> Program τ () -linkThread (Thread a) = do - liftIO $ do - Async.link a + atomically $ do + modifyTVar' scope (\pointers -> removeElement pointer pointers) + ) {- | Cancel a thread. -(this wraps __async__\'s 'Control.Concurrent.Async.cancel'. The underlying -mechanism used is to throw the 'AsyncCancelled' to the other thread. That -exception is asynchronous, so will not be trapped by a +(this wraps __base__\'s 'Control.Concurrent.killThread'. The underlying +mechanism used is to throw the 'GHC.Conc.ThreadKilled' exception to the other +thread. That exception is asynchronous, so will not be trapped by a 'Core.Program.Exceptions.catch' block and will indeed cause the thread receiving the exception to come to an end) @since 0.4.5 -} cancelThread :: Thread α -> Program τ () -cancelThread (Thread a) = do +cancelThread thread = do liftIO $ do - Async.cancel a + killThread (threadPointerOf thread) {- | Fork two threads and wait for both to finish. The return value is the pair of @@ -337,17 +383,16 @@ For a variant that ingores the return values and just waits for both see 'concurrentThreads_' below. -(this wraps __async__\'s 'Control.Concurrent.Async.concurrently') - @since 0.4.0 -} concurrentThreads :: Program τ α -> Program τ β -> Program τ (α, β) concurrentThreads one two = do - context <- ask - liftIO $ do - Async.concurrently - (subProgram context one) - (subProgram context two) + createScope $ do + a1 <- forkThread one + a2 <- forkThread two + result1 <- waitThread a1 + result2 <- waitThread a2 + pure (result1, result2) {- | Fork two threads and wait for both to finish. @@ -356,17 +401,10 @@ if either sub-program fails with an exception the other program which is still running will be cancelled and the original exception is then re-thrown. -(this wraps __async__\'s 'Control.Concurrent.Async.concurrently_') - @since 0.4.0 -} concurrentThreads_ :: Program τ α -> Program τ β -> Program τ () -concurrentThreads_ one two = do - context <- ask - liftIO $ do - Async.concurrently_ - (subProgram context one) - (subProgram context two) +concurrentThreads_ one two = void (concurrentThreads one two) {- | Fork two threads and race them against each other. This blocks until one or @@ -387,18 +425,27 @@ For a variant that ingores the return value and just races the threads see 'raceThreads_' below. -(this wraps __async__\'s 'Control.Concurrent.Async.race') - @since 0.4.0 -} raceThreads :: Program τ α -> Program τ β -> Program τ (Either α β) raceThreads one two = do - context <- ask - liftIO $ do - Async.race - (subProgram context one) - (subProgram context two) + createScope $ do + outcome <- liftIO $ do + newEmptyMVar + _ <- forkThread $ do + !result1 <- one + liftIO $ do + putMVar outcome (Left result1) + + _ <- forkThread $ do + !result2 <- two + liftIO $ do + putMVar outcome (Right result2) + + liftIO $ do + readMVar outcome + {- | Fork two threads and race them against each other. When one action completes the other will be cancelled with an exception. This is useful for enforcing @@ -413,14 +460,7 @@ ) @ -(this wraps __async__\'s 'Control.Concurrent.Async.race_') - @since 0.4.0 -} raceThreads_ :: Program τ α -> Program τ β -> Program τ () -raceThreads_ one two = do - context <- ask - liftIO $ do - Async.race_ - (subProgram context one) - (subProgram context two) +raceThreads_ one two = void (raceThreads one two)