core-program 0.6.8.0 → 0.6.9.2

raw patch · 5 files changed

+333/−2 lines, 5 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• core-program.cabal +3/−2
• lib/Core/Program.hs +2/−0
• lib/Core/Program/Execute.hs +39/−0
• lib/Core/Program/Threads.hs +34/−0
• lib/Core/Program/Workers.hs +255/−0

@@ -1,11 +1,11 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.35.1.+-- This file has been generated from package.yaml by hpack version 0.35.2. -- -- see: https://github.com/sol/hpack  name:           core-program-version:        0.6.8.0+version:        0.6.9.2 synopsis:       Opinionated Haskell Interoperability description:    A library to help build command-line programs, both tools and                 longer-running daemons.@@ -45,6 +45,7 @@       Core.Program.Notify       Core.Program.Threads       Core.Program.Unlift+      Core.Program.Workers       Core.System       Core.System.Base       Core.System.External

@@ -22,6 +22,7 @@       -- and more.       module Core.Program.Execute     , module Core.Program.Threads+    , module Core.Program.Workers     , module Core.Program.Unlift     , module Core.Program.Metadata     , module Core.Program.Exceptions@@ -55,3 +56,4 @@ import Core.Program.Notify import Core.Program.Threads import Core.Program.Unlift+import Core.Program.Workers

@@ -77,6 +77,7 @@     , getConsoleWidth     , getApplicationState     , setApplicationState+    , modifyApplicationState     , changeProgram        -- * Useful actions@@ -602,6 +603,44 @@     liftIO $ do         let v = applicationDataFrom context         modifyMVar_ v (\_ -> pure user)++{- |+Modify the user supplied top-level application state in a single atomic action+combining getting the value and replacing it. Following the pattern of other+@modify@ functions in the Haskell ecosystem, this takes a function which+allows you to take limited actions with the existing value, returning the new+value that should be stored.++@++    'modifyApplicationState'+        ( \settings{answer = a} ->+            'pure'+                (settings+                    { answer = a + 1+                    }+                )+        )+@++While the function you need to supply is in 'Program' @τ@ and so able to do+general work if necessary, some care should be taken to return from the action+as quickly as possible; this call will be blocking other consumers of the+top-level application state until it returns.++@since 0.6.9+-}+modifyApplicationState :: (τ -> Program τ τ) -> Program τ ()+modifyApplicationState program = do+    context <- ask+    liftIO $ do+        let v = applicationDataFrom context+        modifyMVar_+            v+            ( \user -> do+                user' <- subProgram context (program user)+                pure user'+            )  {- | Sometimes you need to change the type of the application state from what is

@@ -40,11 +40,13 @@     , concurrentThreads_     , raceThreads     , raceThreads_+    , timeoutThread        -- * Internals     , Thread     , unThread     , Terminator (..)+    , Timeout (..)     ) where  import Control.Concurrent (ThreadId, forkIO, killThread)@@ -60,6 +62,7 @@ import Control.Monad.Reader.Class (MonadReader (ask)) import Core.Data.Structures import Core.Program.Context+import Core.Program.Execute import Core.Program.Logging import Core.System.Base import Core.Text.Rope@@ -511,3 +514,34 @@ linkThread :: Thread α -> Program τ () linkThread _ = pure () {-# DEPRECATED linkThread "Exceptions are bidirectional so linkThread no longer needed" #-}++{-|+If a timeout is exceeded this exception will be thrown by 'timeoutThread'.++@since 0.6.9+-}+data Timeout = Timeout deriving (Show)++instance Exception Timeout++{- |+Run a program that needs to complete before the given number of seconds have+elapsed. This will return the result of the sub program or throw the 'Timeout'+exception if the limit is exceeded.++@since 0.6.9+-}+timeoutThread :: Rational -> Program τ α -> Program τ α+timeoutThread seconds program = do+    result <-+        raceThreads+            ( do+                sleepThread seconds+                pure Timeout+            )+            ( do+                program+            )+    case result of+        Left e -> throw e+        Right a -> pure a

@@ -0,0 +1,255 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE ImportQualifiedPost #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE StrictData #-}+{-# OPTIONS_GHC -fno-warn-orphans #-}+{-# OPTIONS_HADDOCK prune #-}++{- |+Utility functions for building programs which consume work off of a queue.++Frequently you need to receive items from an external system and perform work+on them. One way to structure such a program is to feed the items into a queue+and then consume those items one at a time. That, of course, is+slow—especially when then worker has to itself carry out computationally+intensive tasks or interact itself with external systems. So we want to have+multiple workers running, but only to an extent limited by the number of cores+available, the number of external connections allowed, or some other+constraint.++This library allows you to add items to a queue, then launch worker threads to+consume those items at up to a specified maximum amount of concurrency.+-}+module Core.Program.Workers+    ( -- * Work Queue+      newQueue+    , writeQueue+    , writeQueue'+    , finishQueue++      -- * Worker Threads+    , runWorkers_+    , mapWorkers++      -- * Internals+    , Queue+    , unQueue+    , getMachineSize+    ) where++import Control.Concurrent.STM (atomically)+import Control.Concurrent.STM.TQueue (TQueue, flushTQueue, newTQueueIO, readTQueue, unGetTQueue, writeTQueue)+import Control.Monad+    ( forM+    )+import Core.Program.Context+import Core.Program.Threads+import Core.System.Base+import GHC.Conc (getNumCapabilities)++{- |+Report back the number of processor cores that are available as Haskell+"capabilities" (this was set when you launched the program with+'Core.Program.Execute.execute'). This can best be used to set the number of+concurrent worker threads when running 'runWorkers_' or 'mapWorkers'.++@since 0.6.9+-}+getMachineSize :: Program τ Int+getMachineSize = liftIO $ do+    getNumCapabilities++{- |+A queue which has an end, someday.++(this is a thin wrapper over the __stm__ 'TQueue' type)++@since 0.6.9+-}+newtype Queue α = Queue (TQueue (Maybe α))++{- |+Initialize a new queue.++@since 0.6.9+-}+newQueue :: Program τ (Queue α)+newQueue = do+    queue <- liftIO $ do+        newTQueueIO+    pure (Queue queue)++{- |+Add an item to the queue.++@since 0.6.9+-}+writeQueue :: Queue α -> α -> Program τ ()+writeQueue (Queue queue) item = do+    liftIO $ do+        atomically $ do+            writeTQueue queue (Just item)++{- |+Add a list of items to the queue.++@since 0.6.9+-}+writeQueue' :: Foldable ω => Queue α -> ω α -> Program τ ()+writeQueue' (Queue queue) items = do+    liftIO $ do+        atomically $ do+            mapM_+                ( \item ->+                    writeTQueue queue (Just item)+                )+                items++{- |+Indicate that you are finished adding queue, thereby allowing the worker+threads consuming from the queue to complete and return.++Remember that you can call at any time, even before you have launched the+worker threads with 'runWorkers_'.++@since 0.6.9+-}+finishQueue :: Queue α -> Program τ ()+finishQueue (Queue queue) = do+    liftIO $ do+        atomically $ do+            writeTQueue queue Nothing++{- |+Access the underlying queue. We make use of the STM 'TQueue' type, so you'll+want the following imports:++@+import "Control.Concurrent.STM" ('atomically')+import "Control.Concurrent.STM.TQueue" ('TQueue', 'writeTQueue')+@++Having accessed the underlying queue you can write items, wrapped in 'Just', to+it directly:++@+    'liftIO' $ do+        'atomically' $ do+            'writeTQueue' queue ('Just' item)+@++A 'Nothing' written to the underlying queue will signal the worker threads+that the end of input has been reached and they can safely return.++@since 0.6.9+-}+unQueue :: Queue α -> TQueue (Maybe α)+unQueue (Queue queue) = queue++{- |+Run a pool of worker threads which consume items off the work queue.++Once you have an action that enqueues items with 'writeQueue' you can then+launch the worker threads:++@+    'runWorkers_' 16 worker queue+@++consuming 16 items at a time concurrently in this example.++It is assumed that the workers have a way of communicating their results+onwards, either because they are side-effecting in the real world themselves,+or because you have passed in some  'Control.Concurrent.MVar' or 'TQueue' to+collect the results.++@since 0.6.9+-}+runWorkers_ :: Int -> (α -> Program τ ()) -> Queue α -> Program τ ()+runWorkers_ n action (Queue queue) = do+    createScope $ do+        ts <- forM [1 .. n] $ \_ -> do+            forkThread $ do+                loop+        _ <- waitThreads' ts+        pure ()+  where+    loop = do+        possibleItem <- liftIO $ do+            atomically $ do+                readTQueue queue -- blocks+        case possibleItem of+            Nothing -> do+                --+                -- We put the Nothing back so that other workers can also shutdown.+                --+                liftIO $ do+                    atomically $ do+                        unGetTQueue queue Nothing+            Just item -> do+                --+                -- Do the work+                --+                action item+                loop++{- |+Map a pool of workers over a list concurrently.++Simply forking one Haskell thread for every item in a list is a suprisingly+reasonable choice in many circumstances given how good Haskell's concurrency+machinery is, and in this library can be achieved by 'Control.Monad.forM'ing+'forkThread' over a list of items. But if you need tighter control over the+amount of concurrency—as is often the case when doing something+computationally heavy or making requests of an external service with known+limitations—then you are better off using this convenience function.++(this was originally modelled on __async__\'s+'Control.Concurrent.Async.mapConcurrently'. That implementation has the+drawback that the number of threads created is set by the size of the+structure being traversed. Here we set the amount of concurrency explicitly.)++Be aware that the order of items in the output list is non-deterministic and+will depend on the order that the action function completes, not the order of+items in the input.++@since 0.6.9+-}+mapWorkers :: Int -> (α -> Program τ β) -> [α] -> Program τ [β]+mapWorkers n action items = do+    inputs <- newQueue++    outputs <- liftIO $ do+        newTQueueIO :: IO (TQueue β)++    --+    -- Load the input list into a queue followed by a terminator.+    --++    writeQueue' inputs items+    finishQueue inputs++    --+    -- Invoke the general concurrent workers tool above to process the queue.+    --++    runWorkers_+        n+        ( \item -> do+            result <- action item+            liftIO $ do+                atomically $ do+                    writeTQueue outputs result+        )+        inputs++    --+    -- Convert the results back to a list.+    --++    liftIO $ do+        atomically $ do+            flushTQueue outputs