core-program 0.5.2.0 → 0.6.0.1

raw patch · 4 files changed

+279/−258 lines, 4 filesdep −asyncdep ~core-dataPVP ok

version bump matches the API change (PVP)

Dependencies removed: async

Dependency ranges changed: core-data

API changes (from Hackage documentation)

Files

• core-program.cabal +5/−6
• lib/Core/Program/Context.hs +8/−3
• lib/Core/Program/Execute.hs +97/−120
• lib/Core/Program/Threads.hs +169/−129

@@ -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

@@ -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             }

@@ -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

@@ -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)