async-pool 0.9.1 → 0.9.2

raw patch · 5 files changed

+43/−15 lines, 5 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• Control/Concurrent/Async/Pool.hs +1/−1
• Control/Concurrent/Async/Pool/Async.hs +25/−4
• Control/Concurrent/Async/Pool/Internal.hs +12/−9
• async-pool.cabal +1/−1
• test/main.hs +4/−0

@@ -99,7 +99,7 @@     withAsyncOnWithUnmask,      -- ** Quering 'Async's-    wait, poll, waitCatch, cancel, cancelWith,+    wait, poll, waitCatch, cancel, cancelWith, cancelAll,      -- ** STM operations     waitSTM, pollSTM, waitCatchSTM,

@@ -107,7 +107,7 @@  -- | A 'Handle' is a unique identifier for a task submitted to a 'Pool'. type Handle    = Node-data State     = Ready | Starting | forall a. Started ThreadId (TMVar a)+data State     = Ready | Starting | Started ThreadId SomeTMVar data Status    = Pending | Completed deriving (Eq, Show) type TaskGraph = Gr (TVar State) Status @@ -163,14 +163,20 @@     forM_ (labNodes g) $ \(_h, st) -> do         x <- readTVar st         case x of-            Started _tid v -> waitTMVar v+            Started _tid v -> waitSomeTMVar v             _ -> retry +data SomeTMVar where+    SomeTMVar :: forall a. TMVar a -> SomeTMVar++waitSomeTMVar :: SomeTMVar -> STM ()+waitSomeTMVar (SomeTMVar mv) = waitTMVar mv+ data TaskGroup = TaskGroup     { pool    :: Pool     , avail   :: TVar Int       -- ^ The number of available execution slots in the pool.-    , pending :: forall a. TVar (IntMap (IO ThreadId, TMVar a))+    , pending :: TVar (IntMap (IO ThreadId, SomeTMVar))       -- ^ Nodes in the task graph that are waiting to start.     } @@ -243,7 +249,7 @@             doFork $ try (restore (action `finally` cleanup h))                 >>= atomically . putTMVar var -    modifyTVar (pending p) (IntMap.insert h (start, var))+    modifyTVar (pending p) (IntMap.insert h (start, SomeTMVar var))     tv <- newTVar Ready     modifyTVar (tasks (pool p)) (insNode (h, tv)) @@ -253,6 +259,13 @@         modifyTVar (avail p) succ         cleanupTask (pool p) h +-- | Like 'asyncUsing' but waits until there are free slots in the TaskGroup+asyncUsingLazy :: TaskGroup -> (IO () -> IO ThreadId) -> IO a -> STM (Async a)+asyncUsingLazy p doFork action = do+    availSlots <- readTVar (avail p)+    check (availSlots > 0)+    asyncUsing p doFork action+ -- | Return the next available thread identifier from the pool.  These are --   monotonically increasing integers. nextIdent :: Pool -> STM Int@@ -710,10 +723,18 @@ -- exception handler. {-# INLINE rawForkIO #-} rawForkIO :: IO () -> IO ThreadId+#if MIN_VERSION_base(4,17,0)+rawForkIO (IO action) = IO $ \ s ->+#else rawForkIO action = IO $ \ s ->+#endif    case (fork# action s) of (# s1, tid #) -> (# s1, ThreadId tid #)  {-# INLINE rawForkOn #-} rawForkOn :: Int -> IO () -> IO ThreadId+#if MIN_VERSION_base(4,17,0)+rawForkOn (I# cpu) (IO action) = IO $ \ s ->+#else rawForkOn (I# cpu) action = IO $ \ s ->+#endif    case (forkOn# cpu action s) of (# s1, tid #) -> (# s1, ThreadId tid #)

@@ -10,7 +10,7 @@ import qualified Control.Concurrent.Async as Async (withAsync) import           Control.Concurrent.Async.Pool.Async import           Control.Concurrent.STM-import           Control.Exception (SomeException, throwIO, finally)+import           Control.Exception (SomeException, throwIO, finally, bracket_) import           Control.Monad hiding (forM, forM_) import           Control.Monad.Base import           Control.Monad.IO.Class (MonadIO(..))@@ -23,11 +23,10 @@ import           Data.Monoid (Monoid(mempty), (<>)) import           Data.Traversable (Traversable(sequenceA), forM) import           Prelude hiding (mapM_, mapM, foldr, all, any, concatMap, foldl1)-import           Unsafe.Coerce  -- | Return a list of actions ready for execution, by checking the graph to --   ensure all dependencies have completed.-getReadyNodes :: TaskGroup -> TaskGraph -> STM (IntMap (IO ThreadId, TMVar a))+getReadyNodes :: TaskGroup -> TaskGraph -> STM (IntMap (IO ThreadId, SomeTMVar)) getReadyNodes p g = do     availSlots <- readTVar (avail p)     check (availSlots > 0)@@ -50,7 +49,7 @@  -- | Return a list of tasks ready to execute, and their related state --   variables from the dependency graph.-getReadyTasks :: TaskGroup -> STM [(TVar State, (IO ThreadId, TMVar a))]+getReadyTasks :: TaskGroup -> STM [(TVar State, (IO ThreadId, SomeTMVar))] getReadyTasks p = do     g <- readTVar (tasks (pool p))     map (first (getTaskVar g)) . IntMap.toList <$> getReadyNodes p g@@ -76,8 +75,7 @@ createTaskGroup p cnt = do     c <- newTVarIO cnt     m <- newTVarIO IntMap.empty-    -- Prior to GHC 8, this call to unsafeCoerce was not necessary.-    return $ TaskGroup p c (unsafeCoerce m)+    return $ TaskGroup p c m  -- | Execute tasks in a given task group.  The number of slots determines how --   many threads may execute concurrently.@@ -101,7 +99,9 @@ withTaskGroupIn p n f = createTaskGroup p n >>= \g ->     Async.withAsync (runTaskGroup g) $ const $ f g `finally` cancelAll g --- | Create both a pool, and a task group with a given number of execution slots.+-- | Create both a pool, and a task group with a given number of execution slots,+--   but with a bounded lifetime. Once the given function exits, all tasks (that +--   are still running) in the TaskGroup will be cancelled. withTaskGroup :: Int -> (TaskGroup -> IO b) -> IO b withTaskGroup n f = createPool >>= \p -> withTaskGroupIn p n f @@ -158,6 +158,9 @@ asyncAfter :: TaskGroup -> Async b -> IO a -> IO (Async a) asyncAfter p parent = asyncAfterAll p [taskHandle parent] +extraWorkerWhileBlocked :: TaskGroup -> IO a -> IO a+extraWorkerWhileBlocked p = bracket_ (atomically $ modifyTVar' (avail p) (+ 1)) (atomically $ modifyTVar' (avail p) ((-) 1))+ -- | Helper function used by several of the variants of 'mapTasks' below. mapTasksWorker :: Traversable t                => TaskGroup@@ -166,8 +169,8 @@                -> (Async a -> IO b)                -> IO (t c) mapTasksWorker p fs f g = do-    hs <- forM fs $ atomically . asyncUsing p rawForkIO-    f $ forM hs g+    hs <- forM fs $ atomically . asyncUsingLazy p rawForkIO+    extraWorkerWhileBlocked p $ f $ forM hs g  -- | Execute a group of tasks within the given task group, returning the --   results in order.  The order of execution is random, but the results are

@@ -1,5 +1,5 @@ Name:           async-pool-Version:        0.9.1+Version:        0.9.2 Synopsis:       A modified version of async that supports worker groups and many-to-many task dependencies License-file:   LICENSE License:        MIT

@@ -191,3 +191,7 @@           end <- getCurrentTime           let diff = diffUTCTime end start           diff < 1.2 `shouldBe` True++      it "nested mapTasks work" $ withTaskGroup 1 $ \p -> do+          mapTasks p ([mapTasks p [pure ()]])+          True `shouldBe` True